diff --git a/app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsx b/app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsx index c0f298a2a..cd9a8f0c2 100644 --- a/app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsx +++ b/app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsx @@ -34,8 +34,7 @@ vi.mock('../../lib/orchestration/orchestrationClient', async importOriginal => { }); vi.mock('../../services/socketService', () => { - const socket = { on: vi.fn(), off: vi.fn() }; - return { socketService: { getSocket: vi.fn(() => socket) } }; + return { socketService: { on: vi.fn(), off: vi.fn(), getSocket: vi.fn(() => null) } }; }); vi.mock('../../lib/i18n/I18nContext', () => ({ useT: () => ({ t: (k: string) => k }) })); @@ -52,7 +51,7 @@ const pairingAcceptMock = vi.mocked(apiClient.orchestrationPairing.acceptRequest const pairingDeclineMock = vi.mocked(apiClient.orchestrationPairing.declineRequest); const pairingBlockMock = vi.mocked(apiClient.orchestrationPairing.blockRequest); -const getSocketMock = vi.mocked(socketService.getSocket); +const socketOnMock = vi.mocked(socketService.on); const PINNED_SESSIONS = [ { @@ -260,13 +259,11 @@ describe('TinyPlaceOrchestrationTab', () => { it('refetches when an orchestration:message socket event fires', async () => { let handler: ((payload: unknown) => void) | undefined; - const socket = { - on: vi.fn((event: string, cb: (payload: unknown) => void) => { - if (event === 'orchestration:message') handler = cb; - }), - off: vi.fn(), - }; - getSocketMock.mockReturnValue(socket as never); + // Listeners are registered through socketService.on (which queues until the + // socket exists), so capture the handler there rather than on a raw socket. + socketOnMock.mockImplementation(((event: string, cb: (payload: unknown) => void) => { + if (event === 'orchestration:message') handler = cb; + }) as never); render(); diff --git a/app/src/components/intelligence/TinyPlaceOrchestrationTab.tsx b/app/src/components/intelligence/TinyPlaceOrchestrationTab.tsx index 5f5f5633a..af6d94c13 100644 --- a/app/src/components/intelligence/TinyPlaceOrchestrationTab.tsx +++ b/app/src/components/intelligence/TinyPlaceOrchestrationTab.tsx @@ -514,10 +514,7 @@ export default function TinyPlaceOrchestrationTab() {

{t('tinyplaceOrchestration.failedToLoad')}: {messagesState.message}

- diff --git a/app/src/lib/orchestration/useOrchestrationChats.ts b/app/src/lib/orchestration/useOrchestrationChats.ts index d583a2f12..6383c299f 100644 --- a/app/src/lib/orchestration/useOrchestrationChats.ts +++ b/app/src/lib/orchestration/useOrchestrationChats.ts @@ -311,8 +311,6 @@ export function useOrchestrationChats(t: Translate): UseOrchestrationChatsResult // Live updates: refetch the affected chat + sessions list on new messages. useEffect(() => { - const socket = socketService.getSocket(); - if (!socket) return; const handler = (payload: unknown) => { const event = payload as OrchestrationMessageEvent | null; if (!event || typeof event !== 'object') return; @@ -323,11 +321,14 @@ export function useOrchestrationChats(t: Translate): UseOrchestrationChatsResult void loadMessages(affected); } }; - socket.on('orchestration:message', handler); - socket.on('orchestration_message', handler); + // Register through socketService (not the raw socket): it queues listeners + // until the socket exists, so live refetch still attaches when the tab mounts + // while the core socket is still being created or is reconnecting. + socketService.on('orchestration:message', handler); + socketService.on('orchestration_message', handler); return () => { - socket.off('orchestration:message', handler); - socket.off('orchestration_message', handler); + socketService.off('orchestration:message', handler); + socketService.off('orchestration_message', handler); }; }, [loadSessions, loadMessages, refreshStatus]); diff --git a/gitbooks/developing/architecture/orchestration.md b/gitbooks/developing/architecture/orchestration.md index d1a82c25c..5686187fc 100644 --- a/gitbooks/developing/architecture/orchestration.md +++ b/gitbooks/developing/architecture/orchestration.md @@ -13,7 +13,7 @@ the staged plan under [`docs/plans/subconscious-orchestration/`](../../../docs/p ## End-to-end flow -``` +```text Claude Code / Codex session └─ tinyplace harness wrapper — tails the session JSONL → SessionEnvelopeV1 └─ Signal E2E DM → owner agent's tiny.place inbox [tagged sessionId, source, role] diff --git a/src/openhuman/app_state/ops.rs b/src/openhuman/app_state/ops.rs index 4d95c6ade..d5412bc70 100644 --- a/src/openhuman/app_state/ops.rs +++ b/src/openhuman/app_state/ops.rs @@ -771,7 +771,19 @@ pub fn peek_cached_current_user_identity() -> Option bool { + std::env::var_os("OPENHUMAN_SERVICE_MOCK").is_some() +} + fn fresh_cached_runtime_snapshot(config: &Config, req_id: u64) -> Option { + if service_status_mock_active() { + return None; + } let cache = RUNTIME_SNAPSHOT_CACHE.lock(); let entry = cache.as_ref()?; // A snapshot built for a different config identity is a miss: rebuild against @@ -929,11 +941,15 @@ async fn build_runtime_snapshot(config: &Config, req_id: u64) -> RuntimeSnapshot service: service.0, }; - *RUNTIME_SNAPSHOT_CACHE.lock() = Some(CachedRuntimeSnapshot { - snapshot: snapshot.clone(), - fetched_at: Instant::now(), - config_key: config.workspace_dir.clone(), - }); + // Don't cache a snapshot built under an injected service mock (see + // `service_status_mock_active`) — it would poison later non-mocked reads. + if !service_status_mock_active() { + *RUNTIME_SNAPSHOT_CACHE.lock() = Some(CachedRuntimeSnapshot { + snapshot: snapshot.clone(), + fetched_at: Instant::now(), + config_key: config.workspace_dir.clone(), + }); + } snapshot } diff --git a/src/openhuman/orchestration/graph/build.rs b/src/openhuman/orchestration/graph/build.rs new file mode 100644 index 000000000..615773d85 --- /dev/null +++ b/src/openhuman/orchestration/graph/build.rs @@ -0,0 +1,435 @@ +//! Wake-graph construction and execution. +//! +//! Holds the injected [`OrchestrationRuntime`] seam, the reducer + node wiring +//! ([`build_orchestration_graph`]), the durable run entrypoint +//! ([`run_orchestration_graph`]), and the structure-only topology export +//! ([`orchestration_graph_topology`]). See the [module docs](super) for the +//! graph shape and routing rules. + +use std::sync::Arc; + +use async_trait::async_trait; +use tinyagents::graph::export::GraphTopology; +use tinyagents::graph::{ + ClosureStateReducer, Command, CompiledGraph, GraphBuilder, NodeContext, NodeResult, +}; + +use super::state::{CompressedEntry, OrchestrationState, WorldDiffEntry}; +use crate::openhuman::config::Config; +use crate::openhuman::tinyagents::observability::GraphTracingSink; +use tinyagents::graph::SqliteCheckpointer; + +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; + /// Front-end pass 2 — reasoning reply → finished channel-response text. + async fn frontend_compile(&self, state: &OrchestrationState) -> anyhow::Result; + /// Reasoning core — applies steering, spawns sub-agents, returns reply + trace. + async fn execute(&self, state: &OrchestrationState) -> anyhow::Result; + /// 20:1-compress the cycle's execution trace and persist a store row. + async fn compress(&self, state: &OrchestrationState) -> anyhow::Result; + /// Append one world-state-diff timeline entry (store-persisted, append-only). + async fn world_diff(&self, state: &OrchestrationState) -> anyhow::Result; + /// Context-window utilization (0.0–1.0) for this cycle's accumulated state. + async fn context_utilization(&self, state: &OrchestrationState) -> anyhow::Result; + /// Evict the oldest compressed-history entries to memory RAG. + async fn evict(&self, state: &OrchestrationState) -> anyhow::Result; + /// 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, + max_supersteps: u32, + evict_threshold: f32, +) -> anyhow::Result> { + let mut builder = GraphBuilder::::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:`. Returns the terminal state. +pub async fn run_orchestration_graph( + config: Arc, + runtime: Arc, + state: OrchestrationState, +) -> anyhow::Result { + 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(); + // `SqlRunLedgerCheckpointer` was retired in favor of the crate's own + // `SqliteCheckpointer` (see `agent_orchestration/delegation.rs`); mirrors + // that swap here with a dedicated `orchestration_graph_checkpoints.db`. + let checkpoint_db = config + .workspace_dir + .join("orchestration_graph_checkpoints.db"); + let checkpointer = Arc::new( + SqliteCheckpointer::::open(&checkpoint_db) + .map_err(|e| anyhow::anyhow!("open durable orchestration checkpoint store: {e}"))?, + ); + + 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) +} + +/// Structure-only [`GraphTopology`] of the wake graph for debug / inspection. +/// Built with a no-op runtime — exposes only node names, edges, and routing. +pub fn orchestration_graph_topology() -> anyhow::Result { + struct NoopRuntime; + #[async_trait] + impl OrchestrationRuntime for NoopRuntime { + async fn frontend_instruct(&self, _s: &OrchestrationState) -> anyhow::Result { + Ok(String::new()) + } + async fn frontend_compile(&self, _s: &OrchestrationState) -> anyhow::Result { + Ok(String::new()) + } + async fn execute(&self, _s: &OrchestrationState) -> anyhow::Result { + Ok(ExecuteOutcome { + reply: String::new(), + trace: String::new(), + }) + } + async fn compress(&self, _s: &OrchestrationState) -> anyhow::Result { + Ok(CompressedEntry::default()) + } + async fn world_diff(&self, _s: &OrchestrationState) -> anyhow::Result { + Ok(WorldDiffEntry::default()) + } + async fn context_utilization(&self, _s: &OrchestrationState) -> anyhow::Result { + Ok(0.0) + } + async fn evict(&self, _s: &OrchestrationState) -> anyhow::Result { + Ok(EvictionOutcome { + evicted: 0, + new_utilization: 0.0, + }) + } + async fn send_dm(&self, _c: &str, _b: &str) -> anyhow::Result<()> { + Ok(()) + } + } + + let graph = build_orchestration_graph(Arc::new(NoopRuntime), 12, 0.85)?; + Ok(graph.topology()) +} diff --git a/src/openhuman/orchestration/graph/mod.rs b/src/openhuman/orchestration/graph/mod.rs index dc03027b4..8735af4a0 100644 --- a/src/openhuman/orchestration/graph/mod.rs +++ b/src/openhuman/orchestration/graph/mod.rs @@ -25,438 +25,16 @@ //! ordering) are unit-testable with a single stub, while production wires the //! real agents / store / memory in [`super::ops`]. +pub mod build; pub mod compress; pub mod state; pub mod world_diff; -use std::sync::Arc; - -use async_trait::async_trait; -use tinyagents::graph::export::GraphTopology; -use tinyagents::graph::{ - ClosureStateReducer, Command, CompiledGraph, GraphBuilder, NodeContext, NodeResult, +pub use build::{ + build_orchestration_graph, orchestration_graph_topology, run_orchestration_graph, + EvictionOutcome, ExecuteOutcome, OrchestrationRuntime, OrchestrationUpdate, }; - -use crate::openhuman::config::Config; -use crate::openhuman::tinyagents::observability::GraphTracingSink; -use tinyagents::graph::SqliteCheckpointer; - pub use state::{CompressedEntry, OrchestrationState, WorldDiff, WorldDiffEntry}; -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; - /// Front-end pass 2 — reasoning reply → finished channel-response text. - async fn frontend_compile(&self, state: &OrchestrationState) -> anyhow::Result; - /// Reasoning core — applies steering, spawns sub-agents, returns reply + trace. - async fn execute(&self, state: &OrchestrationState) -> anyhow::Result; - /// 20:1-compress the cycle's execution trace and persist a store row. - async fn compress(&self, state: &OrchestrationState) -> anyhow::Result; - /// Append one world-state-diff timeline entry (store-persisted, append-only). - async fn world_diff(&self, state: &OrchestrationState) -> anyhow::Result; - /// Context-window utilization (0.0–1.0) for this cycle's accumulated state. - async fn context_utilization(&self, state: &OrchestrationState) -> anyhow::Result; - /// Evict the oldest compressed-history entries to memory RAG. - async fn evict(&self, state: &OrchestrationState) -> anyhow::Result; - /// 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, - max_supersteps: u32, - evict_threshold: f32, -) -> anyhow::Result> { - let mut builder = GraphBuilder::::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:`. Returns the terminal state. -pub async fn run_orchestration_graph( - config: Arc, - runtime: Arc, - state: OrchestrationState, -) -> anyhow::Result { - 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(); - // `SqlRunLedgerCheckpointer` was retired in favor of the crate's own - // `SqliteCheckpointer` (see `agent_orchestration/delegation.rs`); mirrors - // that swap here with a dedicated `orchestration_graph_checkpoints.db`. - let checkpoint_db = config - .workspace_dir - .join("orchestration_graph_checkpoints.db"); - let checkpointer = Arc::new( - SqliteCheckpointer::::open(&checkpoint_db) - .map_err(|e| anyhow::anyhow!("open durable orchestration checkpoint store: {e}"))?, - ); - - 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) -} - -/// Structure-only [`GraphTopology`] of the wake graph for debug / inspection. -/// Built with a no-op runtime — exposes only node names, edges, and routing. -pub fn orchestration_graph_topology() -> anyhow::Result { - struct NoopRuntime; - #[async_trait] - impl OrchestrationRuntime for NoopRuntime { - async fn frontend_instruct(&self, _s: &OrchestrationState) -> anyhow::Result { - Ok(String::new()) - } - async fn frontend_compile(&self, _s: &OrchestrationState) -> anyhow::Result { - Ok(String::new()) - } - async fn execute(&self, _s: &OrchestrationState) -> anyhow::Result { - Ok(ExecuteOutcome { - reply: String::new(), - trace: String::new(), - }) - } - async fn compress(&self, _s: &OrchestrationState) -> anyhow::Result { - Ok(CompressedEntry::default()) - } - async fn world_diff(&self, _s: &OrchestrationState) -> anyhow::Result { - Ok(WorldDiffEntry::default()) - } - async fn context_utilization(&self, _s: &OrchestrationState) -> anyhow::Result { - Ok(0.0) - } - async fn evict(&self, _s: &OrchestrationState) -> anyhow::Result { - Ok(EvictionOutcome { - evicted: 0, - new_utilization: 0.0, - }) - } - async fn send_dm(&self, _c: &str, _b: &str) -> anyhow::Result<()> { - Ok(()) - } - } - - let graph = build_orchestration_graph(Arc::new(NoopRuntime), 12, 0.85)?; - Ok(graph.topology()) -} - #[cfg(test)] mod tests; diff --git a/src/openhuman/orchestration/graph/state.rs b/src/openhuman/orchestration/graph/state.rs index 2806ef0ea..7b7aa63c6 100644 --- a/src/openhuman/orchestration/graph/state.rs +++ b/src/openhuman/orchestration/graph/state.rs @@ -51,8 +51,10 @@ pub struct OrchestrationState { /// Master window). pub session_id: String, /// 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. The agent id scopes the key so two linked peers + /// reporting the same `harness_session_id`/seq don't collide (see `seed`). pub cycle_id: String, /// The tiny.place `@handle` of the counterpart the reply DM goes back to. pub counterpart_agent_id: String, @@ -98,15 +100,28 @@ impl OrchestrationState { messages: Vec, ) -> Self { let session_id = session_id.into(); - // 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 counterpart_agent_id = counterpart_agent_id.into(); + // Deterministic cycle id: agent + session + the latest seq in this + // window. The agent id scopes the key so two linked peers reporting the + // same `harness_session_id`/seq don't collide on `compressed_history` / + // `world_diff` rows (the store keys sessions by `(agent_id, session_id)`). + // A resumed run over the same window recomputes the same id, so the + // per-cycle store writes dedupe. + // + // Migration seam: this format changed from `{session}#{seq}` to + // `{counterpart}#{session}#{seq}`. A cycle that was checkpointed under + // the old id and only retries *after* the upgrade recomputes a new id, + // so its already-written `world_diff` / `compressed_history` rows are not + // re-matched. The window is a single in-flight cycle exactly at the + // upgrade boundary (Beta feature, gated by `[orchestration]`); the worst + // case is one extra timeline entry, not corruption. Not worth a one-time + // cleanup here — noted so the boundary behaviour is explicit. 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}"); Self { session_id, cycle_id, - counterpart_agent_id: counterpart_agent_id.into(), + counterpart_agent_id, messages, ..Self::default() } diff --git a/src/openhuman/orchestration/graph/world_diff.rs b/src/openhuman/orchestration/graph/world_diff.rs index 23281e4a3..ec340852b 100644 --- a/src/openhuman/orchestration/graph/world_diff.rs +++ b/src/openhuman/orchestration/graph/world_diff.rs @@ -52,7 +52,8 @@ mod tests { #[test] fn signature_and_mutation_reflect_terminal_state() { let mut s = OrchestrationState::seed("h1", "@peer", Vec::new()); - assert!(event_signature(&s).contains("cycle=h1#0")); + // cycle id is agent-scoped: `##`. + assert!(event_signature(&s).contains("cycle=@peer#h1#0")); assert!(event_signature(&s).contains("reply=no")); assert_eq!(world_mutation(&s), "(no reply)"); diff --git a/src/openhuman/orchestration/ops.rs b/src/openhuman/orchestration/ops.rs index d39948cac..ae3b3a6f1 100644 --- a/src/openhuman/orchestration/ops.rs +++ b/src/openhuman/orchestration/ops.rs @@ -130,19 +130,30 @@ pub fn seed_state( if messages.is_empty() { return Ok(None); } - let mut state = + let state = OrchestrationState::seed(session_id.to_string(), agent_id.to_string(), messages); - // Out-of-band writer pattern (spec §6): bump the global reasoning-cycle - // counter and load the current (non-expired) subconscious steering - // directive into state at cycle start — the reasoning `execute` node then - // weaves it into its prompt. The subconscious never edges into the graph. + Ok(Some(state)) + }) + .map_err(|e| format!("seed_state: {e}")) +} + +/// Bump the global reasoning-cycle counter and load the current (non-expired) +/// subconscious steering directive into `state` — the reasoning `execute` node +/// then weaves it into its prompt (out-of-band writer pattern, spec §6; the +/// subconscious never edges into the graph). +/// +/// Called only *after* the idempotence check confirms the wake will proceed, so +/// no-op triggers (retries, duplicate ingest events, debounce edge cases) don't +/// consume a cycle tick and prematurely expire an active steering directive. +fn apply_cycle_steering(config: &Config, state: &mut OrchestrationState) -> Result<(), String> { + store::with_connection(&config.workspace_dir, |conn| { let cycle = store::bump_cycle_counter(conn)?; state.subconscious_steering = store::current_steering_directive(conn, cycle)? .map(|d| d.text) .filter(|t| !t.trim().is_empty()); - Ok(Some(state)) + Ok(()) }) - .map_err(|e| format!("seed_state: {e}")) + .map_err(|e| format!("apply_cycle_steering: {e}")) } /// The highest message seq currently persisted for the session. @@ -399,7 +410,7 @@ pub async fn invoke_with_runtime( session_id: &str, runtime: Arc, ) -> Result<(), String> { - let Some(state) = seed_state(config, agent_id, session_id)? else { + let Some(mut state) = seed_state(config, agent_id, session_id)? else { log::debug!(target: LOG, "[orchestration] wake.skip_empty session={session_id}"); return Ok(()); }; @@ -411,6 +422,10 @@ pub async fn invoke_with_runtime( ); return Ok(()); } + // Only now that the cycle is confirmed to proceed: advance the reasoning-cycle + // counter and inject the current steering directive. Keeping this after the + // idempotence guard prevents no-op wakes from expiring steering early. + apply_cycle_steering(config, &mut state)?; // Defer under a paused/throttled scheduler gate — the permit is held for the // whole cycle so background pressure backs off without dropping the message. @@ -1008,7 +1023,11 @@ mod tests { Ok(()) }) .unwrap(); - let state = seed_state(&config, "@peer", "h1").unwrap().expect("seeded"); + let mut state = seed_state(&config, "@peer", "h1").unwrap().expect("seeded"); + // Steering is injected once the wake is confirmed to proceed (mirrors + // `invoke_with_runtime` calling `apply_cycle_steering` after the + // idempotence guard), not by `seed_state` itself. + apply_cycle_steering(&config, &mut state).unwrap(); assert_eq!( state.subconscious_steering.as_deref(), Some("prioritize the billing migration"), diff --git a/src/openhuman/orchestration/reasoning_agent/mod.rs b/src/openhuman/orchestration/reasoning_agent/mod.rs index 397a37c85..d56f4e501 100644 --- a/src/openhuman/orchestration/reasoning_agent/mod.rs +++ b/src/openhuman/orchestration/reasoning_agent/mod.rs @@ -3,34 +3,12 @@ //! ([`crate::openhuman::agent_registry::agents::loader`]). //! //! The current subconscious steering directive for a cycle is carried into the -//! agent's system prompt via a task-local ([`ORCHESTRATION_STEERING`]) that the -//! `execute` node scopes around the turn (see [`with_steering`]); `prompt::build` -//! reads it (or falls back to [`DEFAULT_STEERING`]). +//! agent's system prompt via a task-local ([`steering::ORCHESTRATION_STEERING`]) +//! that the `execute` node scopes around the turn (see [`with_steering`]); +//! `prompt::build` reads it (or falls back to [`DEFAULT_STEERING`]). pub mod graph; pub mod prompt; +pub mod steering; -use std::future::Future; - -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(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 { - ORCHESTRATION_STEERING.try_with(|s| s.clone()).ok() -} +pub use steering::{current_steering, with_steering, DEFAULT_STEERING}; diff --git a/src/openhuman/orchestration/reasoning_agent/prompt.rs b/src/openhuman/orchestration/reasoning_agent/prompt.rs index 54100dbbf..6709657eb 100644 --- a/src/openhuman/orchestration/reasoning_agent/prompt.rs +++ b/src/openhuman/orchestration/reasoning_agent/prompt.rs @@ -1,8 +1,9 @@ //! System prompt builder for the `reasoning_agent` built-in. //! //! Assembled per cycle: the base archetype + the active subconscious steering -//! directive (from the [`super::ORCHESTRATION_STEERING`] task-local, or -//! [`super::DEFAULT_STEERING`] when none is set) + tool/safety/workspace context. +//! directive (from the [`super::steering::ORCHESTRATION_STEERING`] task-local, +//! or [`super::DEFAULT_STEERING`] when none is set) + tool/safety/workspace +//! context. use crate::openhuman::context::prompt::{ render_safety, render_tools, render_workspace, PromptContext, diff --git a/src/openhuman/orchestration/reasoning_agent/steering.rs b/src/openhuman/orchestration/reasoning_agent/steering.rs new file mode 100644 index 000000000..067fca34b --- /dev/null +++ b/src/openhuman/orchestration/reasoning_agent/steering.rs @@ -0,0 +1,31 @@ +//! Subconscious steering directive plumbing for the reasoning core. +//! +//! The current steering directive for a wake cycle is carried into the reasoning +//! agent's system prompt via a task-local ([`ORCHESTRATION_STEERING`]) that the +//! `execute` node scopes around the turn (see [`with_steering`]); `prompt::build` +//! reads it (or falls back to [`DEFAULT_STEERING`]). + +use std::future::Future; + +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(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 { + ORCHESTRATION_STEERING.try_with(|s| s.clone()).ok() +} diff --git a/src/openhuman/orchestration/schemas.rs b/src/openhuman/orchestration/schemas.rs index cb313fb99..99e01737a 100644 --- a/src/openhuman/orchestration/schemas.rs +++ b/src/openhuman/orchestration/schemas.rs @@ -359,11 +359,16 @@ fn handle_status(_params: Map) -> ControllerFuture { created_at: d.created_at, expires_after_cycles: d.expires_after_cycles, }); - // MAX() always returns exactly one row (NULL when empty). - let ingest_last: Option = - conn.query_row("SELECT MAX(last_message_at) FROM sessions", [], |r| { - r.get::<_, Option>(0) - })?; + // MAX() always returns exactly one row (NULL when empty). Exclude the + // pinned master/subconscious windows: they're bumped by manual owner + // DMs (`handle_send_master_message`) and steering writes, which would + // otherwise mask a stalled real ingestion pipeline with fresh traffic. + let ingest_last: Option = conn.query_row( + "SELECT MAX(last_message_at) FROM sessions \ + WHERE session_id NOT IN ('master', 'subconscious')", + [], + |r| r.get::<_, Option>(0), + )?; let lag = store::ingest_cursor_lag(conn)?; let last_error = store::kv_get(conn, "orchestration:last_error")?; Ok((steering, ingest_last, lag, last_error)) diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index 39eca3f6e..e836219fe 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -72,8 +72,12 @@ const SCHEMA_DDL: &str = " timestamp TEXT NOT NULL ); - CREATE INDEX IF NOT EXISTS idx_world_diff_session - ON world_diff (agent_id, session_id, seq); + -- The UNIQUE index on (agent_id, session_id, seq) — which makes a racing + -- `MAX(seq)+1` allocation impossible to persist twice — is created by the + -- one-time `migrate()` step (user_version-gated), not here: the initial + -- release shipped a NON-unique `idx_world_diff_session`, and + -- `CREATE UNIQUE INDEX IF NOT EXISTS` under that name is a no-op on stores + -- that already have it. See `migrate`. -- Stage 6: append-only subconscious steering directives. 'Current' directive -- is the latest row with superseded_by IS NULL that has not expired @@ -104,9 +108,45 @@ pub fn with_connection( .with_context(|| format!("open orchestration DB: {}", db_path.display()))?; conn.execute_batch(SCHEMA_DDL) .context("initialise orchestration schema")?; + migrate(&conn).context("migrate orchestration schema")?; f(&conn) } +/// One-time, `user_version`-gated migrations. Runs after the idempotent +/// `SCHEMA_DDL`; each version block executes exactly once per DB. +fn migrate(conn: &Connection) -> Result<()> { + let version: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?; + + // v1 — enforce uniqueness of the world-diff timeline position. + // The initial release created `idx_world_diff_session` NON-unique, so a + // race between concurrent `MAX(seq)+1` allocations could persist duplicate + // `(agent_id, session_id, seq)` rows. Drop the legacy index, de-dupe any + // pre-existing race rows (keep the earliest per key), create the unique + // index under a new name, then reconcile `terminal_state` so it can't point + // at a mutation from a deleted duplicate. Runs once (guarded), so it never + // rewrites `terminal_state` on steady-state opens. + if version < 1 { + conn.execute_batch( + "DROP INDEX IF EXISTS idx_world_diff_session; + DELETE FROM world_diff WHERE rowid NOT IN ( + SELECT MIN(rowid) FROM world_diff GROUP BY agent_id, session_id, seq + ); + CREATE UNIQUE INDEX IF NOT EXISTS idx_world_diff_session_uniq + ON world_diff (agent_id, session_id, seq); + INSERT OR REPLACE INTO kv (k, v) + SELECT 'terminal_state:' || wd.agent_id || ':' || wd.session_id, + wd.world_mutation + FROM world_diff wd + WHERE wd.seq = ( + SELECT MAX(seq) FROM world_diff w2 + WHERE w2.agent_id = wd.agent_id AND w2.session_id = wd.session_id + ); + PRAGMA user_version = 1;", + )?; + } + Ok(()) +} + /// True if a relay message id is already persisted. This guard MUST run before /// decryption so the non-idempotent Signal double-ratchet is never advanced /// twice for the same message. @@ -863,4 +903,79 @@ mod tests { }) .unwrap(); } + + #[test] + fn migrates_legacy_nonunique_world_diff_index_to_unique() { + let tmp = tempfile::tempdir().unwrap(); + let db_path = tmp.path().join("orchestration").join("orchestration.db"); + std::fs::create_dir_all(db_path.parent().unwrap()).unwrap(); + + // Simulate a pre-migration store: the world_diff table with the OLD + // NON-unique index, holding two rows that share (agent, session, seq) — + // the exact duplicate the concurrent `MAX(seq)+1` race could produce. + { + let conn = Connection::open(&db_path).unwrap(); + conn.execute_batch( + "CREATE TABLE world_diff ( + cycle_id TEXT PRIMARY KEY, seq INTEGER NOT NULL, session_id TEXT NOT NULL, + agent_id TEXT NOT NULL, event_signature TEXT NOT NULL, + world_mutation TEXT NOT NULL, delta TEXT NOT NULL, timestamp TEXT NOT NULL); + CREATE INDEX idx_world_diff_session ON world_diff (agent_id, session_id, seq);", + ) + .unwrap(); + // Distinct mutations so we can prove terminal_state is reconciled. + conn.execute( + "INSERT INTO world_diff VALUES ('c1', 1, 's', '@a', 'sig', 'mut_c1', 'd', 't')", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO world_diff VALUES ('c2', 1, 's', '@a', 'sig', 'mut_c2', 'd', 't')", + [], + ) + .unwrap(); + conn.execute("CREATE TABLE kv (k TEXT PRIMARY KEY, v TEXT NOT NULL)", []) + .unwrap(); + // Stale terminal_state pointing at the duplicate that will be deleted. + conn.execute( + "INSERT INTO kv VALUES ('terminal_state:@a:s', 'mut_c2')", + [], + ) + .unwrap(); + } + + // Opening through with_connection applies SCHEMA_DDL + the migration. + with_connection(tmp.path(), |conn| { + // Legacy duplicate race rows are de-duped to one per (agent, session, seq). + let n: i64 = conn.query_row( + "SELECT COUNT(*) FROM world_diff WHERE agent_id='@a' AND session_id='s' AND seq=1", + [], + |r| r.get(0), + )?; + assert_eq!(n, 1, "duplicate legacy rows de-duped"); + + // terminal_state is reconciled to the surviving row's mutation (not the deleted one). + let ts: String = + conn.query_row("SELECT v FROM kv WHERE k='terminal_state:@a:s'", [], |r| { + r.get(0) + })?; + assert_eq!( + ts, "mut_c1", + "terminal_state reconciled to the surviving row" + ); + + // The index is now UNIQUE — a fresh duplicate (agent, session, seq) is rejected. + let dup = conn.execute( + "INSERT INTO world_diff VALUES ('c3', 1, 's', '@a', 'sig', 'mut', 'd', 't')", + [], + ); + assert!(dup.is_err(), "unique index rejects a duplicate seq"); + + // Migration is one-shot: user_version was bumped. + let uv: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?; + assert_eq!(uv, 1, "migration marks user_version"); + Ok(()) + }) + .unwrap(); + } }