Repository navigation
fix(orchestration): app_state runtime-cache regression + orchestration review fixes - #4431
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughUpdates orchestration socket handling, retry behavior, cycle derivation and steering timing, graph checkpoint storage, world diff uniqueness, status filtering, prompt catalog entries, and a documentation code fence. ChangesOrchestration runtime and UI fixes
Prompt catalog and docs
Estimated code review effort: 3 (Moderate) | ~25 minutes Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
coderabbitai
Bot
added
bug
rust-core
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: afd20c98dd
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| Ok(NodeResult::Command( | ||
| Command::default() | ||
| .with_update(OrchestrationUpdate::Pass1 { instructions }) | ||
| .with_goto(["execute"]), |
There was a problem hiding this comment.
When the front-end agent decides an inbound DM is simple and follows the prompt by calling reply_to_channel on pass 1, frontend_instruct still returns only a String; this branch unconditionally stores that string as agent_instructions and jumps to execute. As a result the advertised fast-path direct reply is never used for first-pass completions, and simple messages still invoke the reasoning core, which can delay or change the response instead of sending the channel response immediately.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Deferring this as a design decision rather than applying it here. The wake graph is deliberately two-pass: frontend_instruct (pass 1) only ever returns macro-instructions and always routes to execute; the terminate predicate is a channel_response produced on the pass-2 frontend_compile. Adding a pass-1 fast-path direct reply would change the OrchestrationRuntime trait contract (pass 1 would need to optionally emit a channel_response) and the router's goto logic — a spec-level change to the graph shape, not a localized fix. The current design still sends the reply; it just always routes through the reasoning core once. Flagging for @sanil-23 to decide whether the fast-path is in scope for this PR or a follow-up.
| // Deterministic cycle id: session + the latest seq in this window. A | ||
| // resumed run over the same window recomputes the same id, so the | ||
| // per-cycle store writes (compressed_history, world_diff) dedupe. | ||
| let latest_seq = messages.iter().map(|m| m.seq).max().unwrap_or(0); | ||
| let cycle_id = format!("{session_id}#{latest_seq}"); |
There was a problem hiding this comment.
When two linked agents report the same harness_session_id and line number, this unscoped cycle_id collides even though the store keys sessions by (agent_id, session_id). The collision causes compressed_history and world_diff rows keyed by cycle_id to be ignored or reused for the wrong peer, so one agent's wake cycle can suppress another agent's compressed history/world diff; the checkpoint thread id is also session-only, so the same scenario can mix graph checkpoints. Include the counterpart/agent id in the deterministic run key.
Useful? React with 👍 / 👎.
| useEffect(() => { | ||
| const socket = socketService.getSocket(); | ||
| if (!socket) return; |
There was a problem hiding this comment.
If the orchestration tab mounts while the core socket is still being created or during reconnect, getSocket() returns null here and the effect exits permanently, because no dependency changes when the socket later appears. The rest of socketService provides on/off wrappers that queue listeners until a socket exists; using the raw socket here means live orchestration:message refetches silently never attach in that startup/reconnect path.
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 10
🧹 Nitpick comments (3)
src/openhuman/orchestration/store.rs (1)
353-381: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate row-mapping logic instead of reusing
map_message_row.
list_recent_messages'squery_mapclosure re-implements the exact same field mapping asmap_message_row(Line 237), which was specifically factored out "so it isCopyand can be reused ... without a borrow-lifetime tangle." If a column is added/renamed later, only one of the two copies may get updated.♻️ Reuse map_message_row
let rows = stmt - .query_map(params![agent_id, session_id, limit], |row| { - let chat_kind: String = row.get(3)?; - Ok(OrchestrationMessage { - id: row.get(0)?, - agent_id: row.get(1)?, - session_id: row.get(2)?, - chat_kind: crate::openhuman::orchestration::types::ChatKind::from_str(&chat_kind), - role: row.get(4)?, - body: row.get(5)?, - timestamp: row.get(6)?, - seq: row.get(7)?, - }) - })? + .query_map(params![agent_id, session_id, limit], map_message_row)? .collect::<std::result::Result<Vec<_>, _>>()?;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/orchestration/store.rs` around lines 353 - 381, list_recent_messages currently duplicates the row-to-OrchestrationMessage mapping already handled by map_message_row, so update the query_map closure to reuse that helper instead of re-reading each column manually. Keep the existing ordering and reverse-into-chronological behavior, but route each row through map_message_row so future message field changes stay centralized.src/openhuman/orchestration/schemas.rs (1)
465-485: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTest doesn't exercise
handle_sessions_list.The test only calls
store::list_sessionsand constructspinned_placeholder(...)directly, asserting on values it just set itself. It never invokeshandle_sessions_list, so it doesn't actually verify the handler produces the two pinned placeholders when the store is empty.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/orchestration/schemas.rs` around lines 465 - 485, The test currently bypasses the real sessions handler and only checks store::list_sessions plus pinned_placeholder directly, so it does not verify the behavior under test. Update sessions_list_includes_pinned_windows_when_empty to invoke handle_sessions_list (using the same empty tempdir workspace/config setup) and assert the returned session list contains the two pinned placeholders, referencing handle_sessions_list, list_sessions, and pinned_placeholder to keep the test aligned with the actual handler behavior.src/openhuman/tinyagents/topology.rs (1)
45-66: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider adding a test for the new
orchestration:waketopology entry.
all_topologies_includes_the_member_graphonly checks the member graph. Given this PR's own CI break was caused by a missing registration slipping through untested, a parallel assertion here (orchestration:wakepresent +report.ok) would guard against future regressions in this specific list.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/tinyagents/topology.rs` around lines 45 - 66, Add a test covering the new `orchestration:wake` entry returned by `all_graph_topologies()` in `topology.rs`; the current `all_topologies_includes_the_member_graph` only verifies `agent_teams:member`. Update or add a test to assert that `orchestration:wake` is present in the list and that its `GraphTopologyReport` is successful (`ok`), so this registry-style function is protected against missing registrations.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@app/src/components/intelligence/TinyPlaceOrchestrationTab.tsx`:
- Around line 512-523: The messages error-state Retry button currently calls
selectChat(selected.id), but selectChat in TinyPlaceOrchestrationTab
early-returns for the already selected chat, so retry does nothing. Update the
Button onClick in the messagesState.status === 'error' branch to call the
existing refresh() path, or introduce a retryMessages/chatKey helper that
bypasses the selectChat guard and actually reloads the current chat’s messages.
In `@app/src/lib/orchestration/useOrchestrationChats.ts`:
- Around line 313-332: The orchestration socket listeners are only attached once
via useEffect and return early when socketService.getSocket() is null, so they
never re-subscribe when the socket appears. Update useOrchestrationChats to
register the orchestration:message and orchestration_message handlers through
socketService.on/off, or otherwise make the effect rerun when the socket becomes
available. Keep the existing handler logic and cleanup, but ensure the listeners
are queued/attached even if the socket is created after the effect runs.
In `@gitbooks/developing/architecture/orchestration.md`:
- Around line 16-31: The fenced architecture diagram in the orchestration
documentation uses a bare code block, which triggers markdownlint MD040; update
the opening fence for the diagram block to specify a language such as text so
the section stays lint-clean and highlighted. Use the fenced block around the
orchestration flow diagram as the target and keep the content unchanged.
In `@src/openhuman/orchestration/graph/mod.rs`:
- Around line 46-408: `mod.rs` currently contains substantial orchestration
logic instead of being export-focused. Move the `OrchestrationRuntime`
implementation support, `build_orchestration_graph`, `run_orchestration_graph`,
and related reducer/update wiring into a sibling module such as `graph/build.rs`
or `graph/runtime.rs`, then keep `mod.rs` limited to `pub mod` and `pub use`
re-exports. Make sure the public symbols still resolve from `graph/mod.rs` so
existing callers continue to use `build_orchestration_graph` and
`run_orchestration_graph` unchanged.
In `@src/openhuman/orchestration/ops.rs`:
- Around line 122-146: `seed_state` is advancing the global cycle counter even
when `invoke_with_runtime` later exits early for no new work, which can expire
steering directives too soon. Refactor the flow around `seed_state` and
`invoke_with_runtime` so the new-work check happens before calling
`store::bump_cycle_counter`, or split transcript loading from cycle start so the
counter is only bumped when a real reasoning cycle will run. Keep
`current_steering_directive` injection in `seed_state`, but gate the bump behind
the same condition that allows the graph to proceed.
In `@src/openhuman/orchestration/reasoning_agent/mod.rs`:
- Around line 15-36: Move the steering implementation out of the reasoning_agent
mod.rs export hub: the task-local ORCHESTRATION_STEERING, DEFAULT_STEERING, and
the with_steering/current_steering functions should live in a new
reasoning_agent/steering.rs module instead of mod.rs. Keep mod.rs focused on
module declarations and re-exports only, then update any imports/usages of
with_steering and current_steering to reference the new steering module while
preserving the existing behavior and public API.
In `@src/openhuman/orchestration/schemas.rs`:
- Around line 196-226: `handle_sessions_list` (and the related RPC handlers
noted in this schema module) contains business logic directly in `schemas.rs`
instead of delegating to `ops.rs`. Move the session aggregation and
pinned-placeholder logic out of `handle_sessions_list` into an `ops.rs`
function, then keep the handler focused on loading config, calling that
operation, and serializing the response; follow the same pattern for the other
`handle_*` functions in this file to preserve the controller/schema separation.
- Around line 362-367: The ingest-health query in the schema logic is counting
pinned “master”/“subconscious” sessions, so manual owner DMs can make ingestion
look healthy when it is stalled. Update the `MAX(last_message_at)` lookup in the
ingestion health path to exclude pinned windows and align
`ingest_last_message_at` with `store::ingest_cursor_lag` so both reflect only
real ingestion traffic; use the existing session-query logic around
`conn.query_row(...)` and `handle_send_master_message` as the key places to
adjust filtering.
- Around line 283-289: Wrap the awaited tiny.place send in a timeout so
`handle_tinyplace_signal_send_message` cannot block the RPC indefinitely if the
network stalls. Update the send path in `schemas.rs` where `send_params` is
built and the
`crate::openhuman::tinyplace::handle_tinyplace_signal_send_message(...).await`
call is made, adding a bounded timeout and returning a clear error if it
expires. Keep the existing error mapping behavior, but distinguish timeout
failures from normal send errors in the `map_err` handling.
In `@src/openhuman/orchestration/store.rs`:
- Around line 422-475: `append_world_diff` currently computes `next_seq` with
`SELECT MAX(seq) + 1` and then inserts outside a transaction, so concurrent
calls can race and assign duplicate seq values for the same agent/session. Fix
this by making the seq allocation and insert atomic inside `append_world_diff`
(or the `with_connection` call path) using a transaction, and/or enforcing a
`UNIQUE(agent_id, session_id, seq)` constraint with retry handling on conflict.
---
Nitpick comments:
In `@src/openhuman/orchestration/schemas.rs`:
- Around line 465-485: The test currently bypasses the real sessions handler and
only checks store::list_sessions plus pinned_placeholder directly, so it does
not verify the behavior under test. Update
sessions_list_includes_pinned_windows_when_empty to invoke handle_sessions_list
(using the same empty tempdir workspace/config setup) and assert the returned
session list contains the two pinned placeholders, referencing
handle_sessions_list, list_sessions, and pinned_placeholder to keep the test
aligned with the actual handler behavior.
In `@src/openhuman/orchestration/store.rs`:
- Around line 353-381: list_recent_messages currently duplicates the
row-to-OrchestrationMessage mapping already handled by map_message_row, so
update the query_map closure to reuse that helper instead of re-reading each
column manually. Keep the existing ordering and reverse-into-chronological
behavior, but route each row through map_message_row so future message field
changes stay centralized.
In `@src/openhuman/tinyagents/topology.rs`:
- Around line 45-66: Add a test covering the new `orchestration:wake` entry
returned by `all_graph_topologies()` in `topology.rs`; the current
`all_topologies_includes_the_member_graph` only verifies `agent_teams:member`.
Update or add a test to assert that `orchestration:wake` is present in the list
and that its `GraphTopologyReport` is successful (`ok`), so this registry-style
function is protected against missing registrations.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: c52d9415-1d7b-4aa0-84e3-6fbc75c3f359
📒 Files selected for processing (54)
app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsxapp/src/components/intelligence/TinyPlaceOrchestrationTab.tsxapp/src/lib/i18n/ar.tsapp/src/lib/i18n/bn.tsapp/src/lib/i18n/de.tsapp/src/lib/i18n/en.tsapp/src/lib/i18n/es.tsapp/src/lib/i18n/fr.tsapp/src/lib/i18n/hi.tsapp/src/lib/i18n/id.tsapp/src/lib/i18n/it.tsapp/src/lib/i18n/ko.tsapp/src/lib/i18n/pl.tsapp/src/lib/i18n/pt.tsapp/src/lib/i18n/ru.tsapp/src/lib/i18n/zh-CN.tsapp/src/lib/orchestration/orchestrationClient.tsapp/src/lib/orchestration/useOrchestrationChats.tsgitbooks/developing/architecture/orchestration.mdsrc/core/all.rssrc/core/jsonrpc.rssrc/core/socketio.rssrc/openhuman/about_app/catalog_data.rssrc/openhuman/agent_registry/agents/loader.rssrc/openhuman/config/schema/orchestration.rssrc/openhuman/mcp_server/resources.rssrc/openhuman/orchestration/bus.rssrc/openhuman/orchestration/frontend_agent/agent.tomlsrc/openhuman/orchestration/frontend_agent/graph.rssrc/openhuman/orchestration/frontend_agent/mod.rssrc/openhuman/orchestration/frontend_agent/prompt.mdsrc/openhuman/orchestration/frontend_agent/prompt.rssrc/openhuman/orchestration/graph/compress.rssrc/openhuman/orchestration/graph/mod.rssrc/openhuman/orchestration/graph/state.rssrc/openhuman/orchestration/graph/tests.rssrc/openhuman/orchestration/graph/world_diff.rssrc/openhuman/orchestration/mod.rssrc/openhuman/orchestration/ops.rssrc/openhuman/orchestration/reasoning_agent/agent.tomlsrc/openhuman/orchestration/reasoning_agent/graph.rssrc/openhuman/orchestration/reasoning_agent/mod.rssrc/openhuman/orchestration/reasoning_agent/prompt.mdsrc/openhuman/orchestration/reasoning_agent/prompt.rssrc/openhuman/orchestration/schemas.rssrc/openhuman/orchestration/steering.rssrc/openhuman/orchestration/store.rssrc/openhuman/orchestration/tools.rssrc/openhuman/orchestration/types.rssrc/openhuman/subconscious/engine.rssrc/openhuman/tinyagents/topology.rssrc/openhuman/tinyplace/mod.rssrc/openhuman/tools/mod.rssrc/openhuman/tools/ops.rs
| const LOG: &str = "orchestration"; | ||
|
|
||
| /// The reasoning core's output for one cycle. | ||
| pub struct ExecuteOutcome { | ||
| /// The answer the front end compiles into a channel reply on pass 2. | ||
| pub reply: String, | ||
| /// Raw execution trace (assistant text + tool/sub-agent activity) that the | ||
| /// `compress` node condenses 20:1. | ||
| pub trace: String, | ||
| } | ||
|
|
||
| /// Result of the `context_guard` eviction pass. | ||
| pub struct EvictionOutcome { | ||
| /// How many oldest compressed-history entries were pushed to memory + dropped. | ||
| pub evicted: usize, | ||
| /// Utilization (0.0–1.0) after eviction. | ||
| pub new_utilization: f32, | ||
| } | ||
|
|
||
| /// Every behaviour-bearing operation of the wake graph, injected as one trait so | ||
| /// the graph structure is hermetically testable with a single stub. | ||
| #[async_trait] | ||
| pub trait OrchestrationRuntime: Send + Sync { | ||
| /// Front-end pass 1 — raw traffic → macro-instructions for the reasoning core. | ||
| async fn frontend_instruct(&self, state: &OrchestrationState) -> anyhow::Result<String>; | ||
| /// Front-end pass 2 — reasoning reply → finished channel-response text. | ||
| async fn frontend_compile(&self, state: &OrchestrationState) -> anyhow::Result<String>; | ||
| /// Reasoning core — applies steering, spawns sub-agents, returns reply + trace. | ||
| async fn execute(&self, state: &OrchestrationState) -> anyhow::Result<ExecuteOutcome>; | ||
| /// 20:1-compress the cycle's execution trace and persist a store row. | ||
| async fn compress(&self, state: &OrchestrationState) -> anyhow::Result<CompressedEntry>; | ||
| /// Append one world-state-diff timeline entry (store-persisted, append-only). | ||
| async fn world_diff(&self, state: &OrchestrationState) -> anyhow::Result<WorldDiffEntry>; | ||
| /// Context-window utilization (0.0–1.0) for this cycle's accumulated state. | ||
| async fn context_utilization(&self, state: &OrchestrationState) -> anyhow::Result<f32>; | ||
| /// Evict the oldest compressed-history entries to memory RAG. | ||
| async fn evict(&self, state: &OrchestrationState) -> anyhow::Result<EvictionOutcome>; | ||
| /// Send the compiled `channel_response` back over the tiny.place DM. | ||
| async fn send_dm(&self, counterpart_agent_id: &str, body: &str) -> anyhow::Result<()>; | ||
| } | ||
|
|
||
| /// Reducer update emitted by an orchestration node. Exactly one per node result | ||
| /// (the crate applies a single `Update` per boundary). Public only because it | ||
| /// appears in the [`CompiledGraph`] type parameter [`build_orchestration_graph`] | ||
| /// returns; the variants are an internal reducer detail. | ||
| pub enum OrchestrationUpdate { | ||
| /// Front-end pass 1: store macro-instructions + advance the pass counter. | ||
| Pass1 { instructions: String }, | ||
| /// Front-end pass 2: store the finished channel response + advance the pass | ||
| /// counter. Its presence is the terminate predicate. | ||
| Pass2 { channel_response: String }, | ||
| /// Reasoning core produced a reply + trace. | ||
| Executed { reply: String, trace: String }, | ||
| /// Compression node appended a compressed-history entry. | ||
| PushCompressed(CompressedEntry), | ||
| /// World-diff node appended a timeline entry. | ||
| PushWorldDiff(WorldDiffEntry), | ||
| /// Context guard measured utilization (no eviction). | ||
| Context(f32), | ||
| /// Context guard evicted `count` oldest entries and reset utilization. | ||
| Evicted { count: usize, utilization: f32 }, | ||
| /// The outbound DM was dispatched — latch so it can never double-send. | ||
| DmSent, | ||
| /// No state change (structural nodes). | ||
| Noop, | ||
| } | ||
|
|
||
| /// Lift an injected node's `anyhow` error into the graph error type. | ||
| fn graph_err(e: anyhow::Error) -> tinyagents::TinyAgentsError { | ||
| tinyagents::TinyAgentsError::Graph(e.to_string()) | ||
| } | ||
|
|
||
| /// Build (but do not run) the orchestration wake graph. Shared by | ||
| /// [`run_orchestration_graph`] and [`orchestration_graph_topology`]. | ||
| /// | ||
| /// `max_supersteps` is the loop-continuity backstop; `evict_threshold` is the | ||
| /// context-guard eviction trigger (clamped 0.8–0.9 by the caller). | ||
| pub fn build_orchestration_graph( | ||
| runtime: Arc<dyn OrchestrationRuntime>, | ||
| max_supersteps: u32, | ||
| evict_threshold: f32, | ||
| ) -> anyhow::Result<CompiledGraph<OrchestrationState, OrchestrationUpdate>> { | ||
| let mut builder = GraphBuilder::<OrchestrationState, OrchestrationUpdate>::new().set_reducer( | ||
| ClosureStateReducer::new(|mut s: OrchestrationState, u: OrchestrationUpdate| { | ||
| match u { | ||
| OrchestrationUpdate::Pass1 { instructions } => { | ||
| s.agent_instructions = Some(instructions); | ||
| s.pass += 1; | ||
| } | ||
| OrchestrationUpdate::Pass2 { channel_response } => { | ||
| s.channel_response = Some(channel_response); | ||
| s.pass += 1; | ||
| } | ||
| OrchestrationUpdate::Executed { reply, trace } => { | ||
| s.agent_reply = Some(reply); | ||
| s.execution_trace = trace; | ||
| } | ||
| OrchestrationUpdate::PushCompressed(entry) => s.compressed_history.push(entry), | ||
| OrchestrationUpdate::PushWorldDiff(entry) => s.world_state_diff.entries.push(entry), | ||
| OrchestrationUpdate::Context(util) => s.context_utilization = util, | ||
| OrchestrationUpdate::Evicted { count, utilization } => { | ||
| let drop = count.min(s.compressed_history.len()); | ||
| s.compressed_history.drain(0..drop); | ||
| s.context_utilization = utilization; | ||
| } | ||
| OrchestrationUpdate::DmSent => s.dm_sent = true, | ||
| OrchestrationUpdate::Noop => {} | ||
| } | ||
| Ok(s) | ||
| }), | ||
| ); | ||
|
|
||
| // `normalize`: window already folded into state before the run (ops::seed_state). | ||
| builder = builder.add_node( | ||
| "normalize", | ||
| |s: OrchestrationState, _c: NodeContext| async move { | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, cycle_id = %s.cycle_id, | ||
| node = "normalize", messages = s.messages.len(), "[orchestration] node.enter", | ||
| ); | ||
| Ok(NodeResult::Update(OrchestrationUpdate::Noop)) | ||
| }, | ||
| ); | ||
|
|
||
| // `frontend`: the router. Two-pass, Quick LLM, conditional goto. | ||
| { | ||
| let runtime = runtime.clone(); | ||
| builder = builder.add_node("frontend", move |s: OrchestrationState, _c: NodeContext| { | ||
| let runtime = runtime.clone(); | ||
| async move { | ||
| let pass = s.pass + 1; | ||
|
|
||
| // Defensive terminate: a response already exists (re-entry / resume). | ||
| if s.channel_response.is_some() { | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, node = "frontend", pass, | ||
| route = "send_dm", reason = "channel_response_present", | ||
| "[orchestration] node.route", | ||
| ); | ||
| return Ok(NodeResult::Command( | ||
| Command::default().with_goto(["send_dm"]), | ||
| )); | ||
| } | ||
|
|
||
| // Loop-continuity backstop (spec §5): never cycle past the cap. | ||
| if pass > max_supersteps { | ||
| let body = s.agent_reply.clone().unwrap_or_else(|| "…".to_string()); | ||
| tracing::warn!( | ||
| target: LOG, session_id = %s.session_id, node = "frontend", pass, | ||
| route = "send_dm", reason = "max_supersteps_backstop", | ||
| "[orchestration] node.route", | ||
| ); | ||
| return Ok(NodeResult::Command( | ||
| Command::default() | ||
| .with_update(OrchestrationUpdate::Pass2 { | ||
| channel_response: body, | ||
| }) | ||
| .with_goto(["send_dm"]), | ||
| )); | ||
| } | ||
|
|
||
| // Pass 2: reasoning replied → compile the channel response. | ||
| if s.agent_reply.is_some() { | ||
| let body = runtime.frontend_compile(&s).await.map_err(graph_err)?; | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, node = "frontend", pass, | ||
| route = "send_dm", reason = "reply_ready", "[orchestration] node.route", | ||
| ); | ||
| return Ok(NodeResult::Command( | ||
| Command::default() | ||
| .with_update(OrchestrationUpdate::Pass2 { | ||
| channel_response: body, | ||
| }) | ||
| .with_goto(["send_dm"]), | ||
| )); | ||
| } | ||
|
|
||
| // Pass 1: raw traffic → macro-instructions, hand down to the core. | ||
| let instructions = runtime.frontend_instruct(&s).await.map_err(graph_err)?; | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, node = "frontend", pass, | ||
| route = "execute", reason = "first_pass", "[orchestration] node.route", | ||
| ); | ||
| Ok(NodeResult::Command( | ||
| Command::default() | ||
| .with_update(OrchestrationUpdate::Pass1 { instructions }) | ||
| .with_goto(["execute"]), | ||
| )) | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| // `execute`: reasoning core — applies steering, spawns sub-agents, sets reply. | ||
| { | ||
| let runtime = runtime.clone(); | ||
| builder = builder.add_node("execute", move |s: OrchestrationState, _c: NodeContext| { | ||
| let runtime = runtime.clone(); | ||
| async move { | ||
| let out = runtime.execute(&s).await.map_err(graph_err)?; | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, cycle_id = %s.cycle_id, | ||
| node = "execute", trace_len = out.trace.len(), "[orchestration] node.exit", | ||
| ); | ||
| Ok(NodeResult::Update(OrchestrationUpdate::Executed { | ||
| reply: out.reply, | ||
| trace: out.trace, | ||
| })) | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| // `compress`: 20:1-compress the cycle trace, persist a compressed-history row. | ||
| { | ||
| let runtime = runtime.clone(); | ||
| builder = builder.add_node("compress", move |s: OrchestrationState, _c: NodeContext| { | ||
| let runtime = runtime.clone(); | ||
| async move { | ||
| let entry = runtime.compress(&s).await.map_err(graph_err)?; | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, cycle_id = %s.cycle_id, | ||
| node = "compress", covered = entry.covered_messages, "[orchestration] node.exit", | ||
| ); | ||
| Ok(NodeResult::Update(OrchestrationUpdate::PushCompressed(entry))) | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| // `world_diff`: append one append-only timeline entry, persist a store row. | ||
| { | ||
| let runtime = runtime.clone(); | ||
| builder = builder.add_node( | ||
| "world_diff", | ||
| move |s: OrchestrationState, _c: NodeContext| { | ||
| let runtime = runtime.clone(); | ||
| async move { | ||
| let entry = runtime.world_diff(&s).await.map_err(graph_err)?; | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, cycle_id = %s.cycle_id, | ||
| node = "world_diff", seq = entry.seq, "[orchestration] node.exit", | ||
| ); | ||
| Ok(NodeResult::Update(OrchestrationUpdate::PushWorldDiff( | ||
| entry, | ||
| ))) | ||
| } | ||
| }, | ||
| ); | ||
| } | ||
|
|
||
| // `send_dm`: the outbound Signal reply. Sends at most once (dm_sent latch). | ||
| { | ||
| let runtime = runtime.clone(); | ||
| builder = builder.add_node("send_dm", move |s: OrchestrationState, _c: NodeContext| { | ||
| let runtime = runtime.clone(); | ||
| async move { | ||
| if s.dm_sent { | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, node = "send_dm", | ||
| reason = "already_sent", "[orchestration] node.skip", | ||
| ); | ||
| } else if let Some(body) = s.channel_response.as_deref() { | ||
| runtime | ||
| .send_dm(&s.counterpart_agent_id, body) | ||
| .await | ||
| .map_err(graph_err)?; | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, node = "send_dm", | ||
| counterpart = %s.counterpart_agent_id, "[orchestration] node.sent", | ||
| ); | ||
| } | ||
| Ok(NodeResult::Update(OrchestrationUpdate::DmSent)) | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| // `context_guard`: utilization + eviction. Runs after all mutations, before END. | ||
| { | ||
| let runtime = runtime.clone(); | ||
| builder = builder.add_node( | ||
| "context_guard", | ||
| move |s: OrchestrationState, _c: NodeContext| { | ||
| let runtime = runtime.clone(); | ||
| async move { | ||
| let util = runtime.context_utilization(&s).await.map_err(graph_err)?; | ||
| if util >= evict_threshold { | ||
| let ev = runtime.evict(&s).await.map_err(graph_err)?; | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, node = "context_guard", | ||
| utilization = util, evicted = ev.evicted, | ||
| new_utilization = ev.new_utilization, "[orchestration] node.evict", | ||
| ); | ||
| Ok(NodeResult::Update(OrchestrationUpdate::Evicted { | ||
| count: ev.evicted, | ||
| utilization: ev.new_utilization, | ||
| })) | ||
| } else { | ||
| tracing::debug!( | ||
| target: LOG, session_id = %s.session_id, node = "context_guard", | ||
| utilization = util, "[orchestration] node.exit", | ||
| ); | ||
| Ok(NodeResult::Update(OrchestrationUpdate::Context(util))) | ||
| } | ||
| } | ||
| }, | ||
| ); | ||
| } | ||
|
|
||
| let graph = builder | ||
| .add_node( | ||
| "done", | ||
| |_s: OrchestrationState, _c: NodeContext| async move { | ||
| Ok(NodeResult::Update(OrchestrationUpdate::Noop)) | ||
| }, | ||
| ) | ||
| .add_edge("normalize", "frontend") | ||
| .add_edge("execute", "compress") | ||
| .add_edge("compress", "world_diff") | ||
| .add_edge("world_diff", "frontend") | ||
| .add_edge("send_dm", "context_guard") | ||
| .add_edge("context_guard", "done") | ||
| .set_entry("normalize") | ||
| .mark_command_routing("frontend") | ||
| .set_finish("done") | ||
| .compile() | ||
| .map_err(|e| anyhow::anyhow!("orchestration graph compile failed: {e}"))?; | ||
| Ok(graph) | ||
| } | ||
|
|
||
| /// Drive one wake cycle for `state.session_id`, checkpointing every super-step | ||
| /// boundary under thread `orchestration:<session_id>`. Returns the terminal state. | ||
| pub async fn run_orchestration_graph( | ||
| config: Arc<Config>, | ||
| runtime: Arc<dyn OrchestrationRuntime>, | ||
| state: OrchestrationState, | ||
| ) -> anyhow::Result<OrchestrationState> { | ||
| let max = config.orchestration.max_supersteps; | ||
| let threshold = config.orchestration.effective_evict_threshold(); | ||
| let thread_id = format!("orchestration:{}", state.session_id); | ||
| let label = thread_id.clone(); | ||
| let checkpointer = Arc::new(SqlRunLedgerCheckpointer::<OrchestrationState>::new(config)); | ||
|
|
||
| tracing::debug!( | ||
| target: LOG, session_id = %state.session_id, %thread_id, | ||
| messages = state.messages.len(), "[orchestration] graph.run.enter", | ||
| ); | ||
|
|
||
| let graph = build_orchestration_graph(runtime, max, threshold)? | ||
| .with_checkpointer(checkpointer) | ||
| .with_event_sink(Arc::new(GraphTracingSink::new(label))); | ||
|
|
||
| let exec = graph | ||
| .run_with_thread(thread_id, state) | ||
| .await | ||
| .map_err(|e| anyhow::anyhow!("orchestration graph run failed: {e}"))?; | ||
|
|
||
| tracing::debug!( | ||
| target: LOG, session_id = %exec.state.session_id, steps = exec.steps, | ||
| dm_sent = exec.state.dm_sent, pass = exec.state.pass, | ||
| compressed = exec.state.compressed_history.len(), | ||
| diff_entries = exec.state.world_state_diff.entries.len(), | ||
| "[orchestration] graph.run.exit", | ||
| ); | ||
| Ok(exec.state) | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
Business logic in mod.rs violates the export-focused convention.
This mod.rs implements the OrchestrationRuntime trait, the reducer, the entire graph builder (build_orchestration_graph), and the run entrypoint (run_orchestration_graph) — substantial business logic rather than module wiring. Consider moving this into a sibling file (e.g. graph/build.rs or graph/runtime.rs), leaving mod.rs with only pub mod/pub use declarations, consistent with how compress.rs/state.rs/world_diff.rs are already split out as siblings.
As per coding guidelines: "src/openhuman/**/mod.rs: mod.rs files should be export-focused only: mod/pub mod, pub use, and controller schema wiring; no business logic."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/openhuman/orchestration/graph/mod.rs` around lines 46 - 408, `mod.rs`
currently contains substantial orchestration logic instead of being
export-focused. Move the `OrchestrationRuntime` implementation support,
`build_orchestration_graph`, `run_orchestration_graph`, and related
reducer/update wiring into a sibling module such as `graph/build.rs` or
`graph/runtime.rs`, then keep `mod.rs` limited to `pub mod` and `pub use`
re-exports. Make sure the public symbols still resolve from `graph/mod.rs` so
existing callers continue to use `build_orchestration_graph` and
`run_orchestration_graph` unchanged.
Source: Coding guidelines
| tokio::task_local! { | ||
| /// The active subconscious steering directive for the current wake cycle, | ||
| /// scoped by the `execute` node around the reasoning agent's turn. | ||
| static ORCHESTRATION_STEERING: String; | ||
| } | ||
|
|
||
| /// Default alignment directive used when no steering directive is active. | ||
| pub const DEFAULT_STEERING: &str = "No active steering directive. Stay aligned with the user's \ | ||
| stated goals and prior context; prefer correctness and safety over speed."; | ||
|
|
||
| /// Scope `steering` for the duration of `fut` (the reasoning agent's turn), so | ||
| /// the prompt builder reads it. Box-pins the inner future to keep the combined | ||
| /// task-local + agent-loop future heap-allocated (same rationale as | ||
| /// [`crate::openhuman::agent::turn_origin::with_origin`]). | ||
| pub async fn with_steering<F: Future>(steering: String, fut: F) -> F::Output { | ||
| ORCHESTRATION_STEERING.scope(steering, Box::pin(fut)).await | ||
| } | ||
|
|
||
| /// The current steering directive, or `None` when no cycle scoped one. | ||
| pub fn current_steering() -> Option<String> { | ||
| ORCHESTRATION_STEERING.try_with(|s| s.clone()).ok() | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Move steering logic out of mod.rs.
mod.rs should stay export-focused (mod/pub mod/pub use only); this file defines the task-local, DEFAULT_STEERING, and the with_steering/current_steering business logic directly.
As per coding guidelines, "src/openhuman/**/mod.rs: mod.rs files should be export-focused only: mod/pub mod, pub use, and controller schema wiring; no business logic."
♻️ Proposed structure
-pub mod graph;
-pub mod prompt;
-
-use std::future::Future;
-
-tokio::task_local! {
- static ORCHESTRATION_STEERING: String;
-}
-
-pub const DEFAULT_STEERING: &str = "...";
-
-pub async fn with_steering<F: Future>(steering: String, fut: F) -> F::Output {
- ORCHESTRATION_STEERING.scope(steering, Box::pin(fut)).await
-}
-
-pub fn current_steering() -> Option<String> {
- ORCHESTRATION_STEERING.try_with(|s| s.clone()).ok()
-}
+pub mod graph;
+pub mod prompt;
+mod steering;
+
+pub use steering::{current_steering, with_steering, DEFAULT_STEERING};Then move the task-local, constant, and both functions into reasoning_agent/steering.rs.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/openhuman/orchestration/reasoning_agent/mod.rs` around lines 15 - 36,
Move the steering implementation out of the reasoning_agent mod.rs export hub:
the task-local ORCHESTRATION_STEERING, DEFAULT_STEERING, and the
with_steering/current_steering functions should live in a new
reasoning_agent/steering.rs module instead of mod.rs. Keep mod.rs focused on
module declarations and re-exports only, then update any imports/usages of
with_steering and current_steering to reference the new steering module while
preserving the existing behavior and public API.
Source: Coding guidelines
afd20c9 to
d94bf3d
Compare
sanil-23
changed the title
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/openhuman/orchestration/graph/state.rs (1)
53-56: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winStale doc comment:
cycle_idfield doc omitscounterpart_agent_id.The field doc says the id is derived from
session_id+ latest seq, butseed()(Line 109) now also folds incounterpart_agent_idto scope the key per-peer. The inline comment insideseed()(Lines 102-108) is accurate; the struct-level doc should match to avoid confusing future readers about what scopescompressed_history/world_diffrows.📝 Suggested doc fix
/// Stable id for this wake cycle. Derived deterministically from - /// `session_id` + the latest message seq so a resumed run reuses it and the - /// compressed-history / world-diff store writes stay idempotent. + /// `counterpart_agent_id` + `session_id` + the latest message seq so a + /// resumed run reuses it and the compressed-history / world-diff store + /// writes stay idempotent and peer-scoped. pub cycle_id: String,🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/orchestration/graph/state.rs` around lines 53 - 56, The struct-level doc for `cycle_id` in `WakeCycleState` is stale because it only mentions `session_id` plus latest seq and omits `counterpart_agent_id`. Update the `cycle_id` field comment to match the current `seed()` behavior by stating that the id is derived from `session_id`, `counterpart_agent_id`, and the latest message seq so the per-peer scope is clear and consistent with the `seed()` logic.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@src/openhuman/orchestration/graph/state.rs`:
- Around line 53-56: The struct-level doc for `cycle_id` in `WakeCycleState` is
stale because it only mentions `session_id` plus latest seq and omits
`counterpart_agent_id`. Update the `cycle_id` field comment to match the current
`seed()` behavior by stating that the id is derived from `session_id`,
`counterpart_agent_id`, and the latest message seq so the per-peer scope is
clear and consistent with the `seed()` logic.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 5fdd95a3-9a1f-4f93-8cd5-4fb4c1b18296
📒 Files selected for processing (10)
app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsxapp/src/components/intelligence/TinyPlaceOrchestrationTab.tsxapp/src/lib/orchestration/useOrchestrationChats.tsgitbooks/developing/architecture/orchestration.mdsrc/openhuman/mcp_server/resources.rssrc/openhuman/orchestration/graph/state.rssrc/openhuman/orchestration/graph/world_diff.rssrc/openhuman/orchestration/ops.rssrc/openhuman/orchestration/schemas.rssrc/openhuman/orchestration/store.rs
✅ Files skipped from review due to trivial changes (1)
- gitbooks/developing/architecture/orchestration.md
🚧 Files skipped from review as they are similar to previous changes (5)
- app/src/components/intelligence/TinyPlaceOrchestrationTab.tsx
- app/src/lib/orchestration/useOrchestrationChats.ts
- src/openhuman/mcp_server/resources.rs
- app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsx
- src/openhuman/orchestration/schemas.rs
d94bf3d to
56f4457
Compare
coderabbitai
Bot
added
the
agent
|
@coderabbitai review |
✅ Action performedReview finished.
|
| -- writer hits a constraint error and its cycle retries (idempotent by | ||
| -- `cycle_id`) rather than silently corrupting the append-only sequence. | ||
| CREATE UNIQUE INDEX IF NOT EXISTS idx_world_diff_session | ||
| ON world_diff (agent_id, session_id, seq); |
There was a problem hiding this comment.
The UNIQUE-index change is a silent no-op on every existing database, so the duplicate-timeline race it claims to close stays open in production.
The schema is applied by conn.execute_batch(SCHEMA_DDL) on every connection open (line 109), with no DROP INDEX, PRAGMA user_version, or migration step anywhere. This keeps the same index name and only adds UNIQUE:
CREATE UNIQUE INDEX IF NOT EXISTS idx_world_diff_session ON world_diff (agent_id, session_id, seq);Per SQLite, CREATE ... IF NOT EXISTS against a name that already exists is a complete no-op — the UNIQUE qualifier is never reconsidered. Any DB created before this PR already has a non-unique idx_world_diff_session, so the upgrade leaves it non-unique. Verified directly: after applying the old then the new DDL, pragma_index_list('world_diff') still reports idx_world_diff_session | unique=0, and two rows with the same (agent_id, session_id, seq) but different cycle_id both insert:
idx_world_diff_session|0
rows after duplicate insert:|2
So the stated protection ("the second writer hits a constraint error and its cycle retries") holds only for freshly-created databases; every already-deployed core stays vulnerable to exactly the duplicate MAX(seq)+1 race this is meant to fix.
Needs a real migration: DROP INDEX IF EXISTS idx_world_diff_session; before recreating (after de-duping any existing rows), or use a new index name.
There was a problem hiding this comment.
Good catch — you're right, CREATE UNIQUE INDEX IF NOT EXISTS under the same name is a no-op on any DB that already has the non-unique index. Fixed in aedbf9f with a real migration in SCHEMA_DDL: DROP INDEX IF EXISTS idx_world_diff_session, de-dupe pre-existing race rows (keep earliest per key so the UNIQUE build can't fail), then CREATE UNIQUE INDEX ... idx_world_diff_session_uniq under a new name so it actually applies on already-created stores. All three steps are idempotent under execute_batch on every open. Added migrates_legacy_nonunique_world_diff_index_to_unique which builds a legacy DB (old non-unique index + two same-(agent,session,seq) rows) and asserts post-open the rows are de-duped and a fresh duplicate insert is rejected.
| // per-cycle store writes dedupe. | ||
| let latest_seq = messages.iter().map(|m| m.seq).max().unwrap_or(0); | ||
| let cycle_id = format!("{session_id}#{latest_seq}"); | ||
| let cycle_id = format!("{counterpart_agent_id}#{session_id}#{latest_seq}"); |
There was a problem hiding this comment.
Changing the cycle_id format breaks replay-dedup for any cycle checkpointed across the upgrade, re-appending a duplicate world-diff / compressed-history row.
cycle_id is the PRIMARY KEY of both world_diff and compressed_history, and idempotent replay works by recomputing the same id for a resumed window (append_world_diff short-circuits on SELECT seq ... WHERE cycle_id = ?). This changes the format from {session}#{seq} to {counterpart}#{session}#{seq}.
A cycle that crashed mid-graph before advancing its cursor (rows written under the old h1#0) will, on its post-upgrade retry, compute @peer#h1#0, miss the existing row, and append a second timeline entry (new seq) plus a duplicate compressed-history row. The window is narrow (only cycles in-flight exactly at the upgrade boundary) and — compounding the world_diff UNIQUE-index issue — the non-unique index won't catch the duplicate seq. Low severity, but a concrete correctness gap at the migration seam worth a note or a one-time cleanup.
There was a problem hiding this comment.
Agreed, and thanks — real but narrow (a single in-flight cycle exactly at the upgrade boundary). Documented in aedbf9f: the worst case is one extra world_diff/compressed_history entry, not corruption, and the feature is Beta-gated behind [orchestration], so I've left it as an explicit migration-seam note rather than a one-time cleanup. Happy to add a cursor-based cleanup if you'd prefer belt-and-suspenders.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/openhuman/orchestration/store.rs`:
- Around line 88-90: The de-dupe migration in the world_diff cleanup leaves
terminal_state stale for sessions where the deleted duplicate was the last
write. After the DELETE in the store.rs migration, recompute terminal_state for
each affected agent_id/session_id based on the remaining world_diff rows, and
update the associated migration/test coverage in the same flow. Extend the
existing migration test around append_world_diff and the duplicate-removal case
to use distinct mutations and assert that terminal_state no longer points at a
removed mutation.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 14e5894f-b922-405c-86cb-771b2dbd0245
📒 Files selected for processing (2)
src/openhuman/orchestration/graph/state.rssrc/openhuman/orchestration/store.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- src/openhuman/orchestration/graph/state.rs
009e312 to
628ea69
Compare
…ce mock tinyhumansai#4399 raised RUNTIME_SNAPSHOT_TTL 2s -> 10s to fix a rebuild-stampede perf bug, but app_state_snapshot_degrades_runtime_service_status_failures ages the cache out with a 2.1s sleep — fine for the old 2s TTL, now < the 10s one, so the test reads a stale cached snapshot from a prior test and never sees the injected 'forced status failure' (asserts service degrades to Unknown). Deterministic; only surfaced now because main (with tinyhumansai#4399) doesn't compile until tinyhumansai#4431 repairs the tinyhumansai#4399/tinyhumansai#4430 conflict, so this is the first branch to actually run it. Fix: when OPENHUMAN_SERVICE_MOCK is active (a test-only env hook production service status already honors), never serve from or write to the process-global runtime cache — the mock's state changes between calls, so caching masks the injected value and poisons later non-mocked reads. Full config_auth_app_state_connectivity_e2e binary: 50 passed.
…n review fixes Rebased onto current main (8aab529), which already carries the tinyhumansai#4399/tinyhumansai#4430 compile reconciliation (graph_fn Some, SqliteCheckpointer migration, MCP catalog) via tinyhumansai#4433 — so this is slimmed to the fixes main still lacks: - app_state: bypass the runtime-snapshot cache when OPENHUMAN_SERVICE_MOCK is active. tinyhumansai#4399 raised RUNTIME_SNAPSHOT_TTL 2s->10s, so the mock-service e2e (which ages the cache with a 2.1s sleep) reads a stale snapshot and never sees the injected failure. Bypassing read+write under the mock fixes it (full config_auth_app_state_connectivity_e2e binary: 50 passed). - orchestration/ops: split steering out of seed_state into apply_cycle_steering, invoked only after the has_new_work guard, so no-op wakes don't consume a cycle tick and prematurely expire steering (CodeRabbit). - orchestration/store: user_version-gated world-diff migration — drop legacy non-unique index, de-dupe race rows, unique index, reconcile terminal_state (M3gA-Mind + CodeRabbit) + test. - orchestration/graph/state: agent-scoped cycle_id (avoid cross-peer collision) + migration-seam note (Codex + M3gA-Mind). - orchestration/schemas: ingest-health MAX(last_message_at) excludes pinned master/subconscious windows (CodeRabbit). - frontend: orchestration socket listeners via socketService.on/off (queues through startup/reconnect) + messages Retry -> refresh() (CodeRabbit/Codex). - docs: orchestration.md fenced-block language (markdownlint).
4d81444 to
025afa4
Compare
sanil-23
changed the title
# Conflicts: # src/openhuman/app_state/ops.rs
…tion Move the wake-graph runtime seam, builder, run entrypoint and topology out of graph/mod.rs into a graph/build.rs sibling, and the steering task-local helpers out of reasoning_agent/mod.rs into reasoning_agent/steering.rs. Both mod.rs files are now export-only (mod/pub mod + pub use), matching the project's canonical module shape. Also refresh the cycle_id field doc to note the counterpart_agent_id scoping. Addresses @coderabbitai on graph/mod.rs:417 and reasoning_agent/mod.rs:36. Claude-Session: https://claude.ai/code/session_01JEcYM2Dt3P9v34VMXLxkMm
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 8ca278e91c
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| fn service_status_mock_active() -> bool { | ||
| std::env::var_os("OPENHUMAN_SERVICE_MOCK").is_some() |
There was a problem hiding this comment.
When OPENHUMAN_SERVICE_MOCK is set to a falsy value such as 0, false, or off, src/openhuman/service/mock.rs::is_enabled() treats the mock as disabled, but this helper still returns true just because the variable exists. In that environment service::status uses the real backend while every app-state runtime snapshot bypasses and skips RUNTIME_SNAPSHOT_CACHE, so calls rebuild the service/screen/local-AI snapshot instead of using the TTL cache; this should use the same truthy parsing as the service mock gate.
Useful? React with 👍 / 👎.
Rebased onto current main (
8aab529de), which already carries the #4399/#4430 compile reconciliation (graph_fnSome,SqliteCheckpointermigration, MCP catalog) via #4433. This PR is slimmed to the fixes main still lacks — no more redundant compile diffs.Fixes
RUNTIME_SNAPSHOT_TTL2s→10s; theapp_state_snapshot_degrades_runtime_service_status_failurese2e ages the cache with a 2.1s sleep, so it now reads a stale snapshot and never sees the injected failure. Fix: bypass the runtime-snapshot cache (read + write) whenOPENHUMAN_SERVICE_MOCKis active (a test-only env production already honors). Fullconfig_auth_app_state_connectivity_e2ebinary: 50 passed.seed_stateintoapply_cycle_steering, invoked only after thehas_new_workguard, so no-op wakes don't consume a cycle tick and prematurely expire steering. (CodeRabbit)user_version-gated world-diff migration — drop the legacy non-unique index, de-dupe race rows, create the unique index, reconcileterminal_state+ test. (M3gA-Mind + CodeRabbit)cycle_id(avoids cross-peercompressed_history/world_diffcollision) + migration-seam note. (Codex + M3gA-Mind)MAX(last_message_at)excludes pinned master/subconscious windows so manual owner DMs can't mask a stalled pipeline. (CodeRabbit)socketService.on/off(queues through startup/reconnect) + messages Retry →refresh(). (CodeRabbit/Codex)orchestration.mdfenced-block language (markdownlint MD040).Validation (local)
cargo clippy -p openhuman0 errors;orchestration49 tests +config_auth_app_state_connectivity_e2ebinary 50 tests pass;cargo fmt --checkclean; frontendvitest(tab) 8 +tscclean.Note
The app_state fix is the one item main urgently needs (its own Rust E2E/Coverage will hit the same regression). Substantive orchestration stages 3–8 are already on main (#4425 + #4430 + #4433 reconciliation); this lands the remaining review-quality fixes on top.