mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-28 13:32:23 +00:00
fix(orchestration): app_state runtime-cache regression + orchestration review fixes (#4431)
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
sanil-23
co-authored by
Steven Enamakel
Steven Enamakel
parent
be386040c4
commit
59462e47f7
@@ -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(<TinyPlaceOrchestrationTab />);
|
||||
|
||||
|
||||
@@ -514,10 +514,7 @@ export default function TinyPlaceOrchestrationTab() {
|
||||
<p>
|
||||
{t('tinyplaceOrchestration.failedToLoad')}: {messagesState.message}
|
||||
</p>
|
||||
<Button
|
||||
variant="secondary"
|
||||
size="sm"
|
||||
onClick={() => selected && selectChat(selected.id)}>
|
||||
<Button variant="secondary" size="sm" onClick={() => void refresh()}>
|
||||
{t('common.retry')}
|
||||
</Button>
|
||||
</div>
|
||||
|
||||
@@ -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]);
|
||||
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -771,7 +771,19 @@ pub fn peek_cached_current_user_identity() -> Option<crate::openhuman::agent::pr
|
||||
/// Return the cached runtime snapshot when it is still within
|
||||
/// `RUNTIME_SNAPSHOT_TTL`, else `None`. Kept as a small helper so both the
|
||||
/// fast-path read and the post-lock double-check share identical freshness logic.
|
||||
/// A service-status mock is injected via `OPENHUMAN_SERVICE_MOCK` (test-only env
|
||||
/// hook that production `service` status already honors). While it is active the
|
||||
/// runtime snapshot must never be served from — or written to — the process-
|
||||
/// global cache: the mock's state changes between calls, so caching it would
|
||||
/// both mask the freshly-injected value and poison later (non-mocked) reads.
|
||||
fn service_status_mock_active() -> bool {
|
||||
std::env::var_os("OPENHUMAN_SERVICE_MOCK").is_some()
|
||||
}
|
||||
|
||||
fn fresh_cached_runtime_snapshot(config: &Config, req_id: u64) -> Option<RuntimeSnapshot> {
|
||||
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
|
||||
}
|
||||
|
||||
@@ -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<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();
|
||||
// `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::<OrchestrationState>::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<GraphTopology> {
|
||||
struct NoopRuntime;
|
||||
#[async_trait]
|
||||
impl OrchestrationRuntime for NoopRuntime {
|
||||
async fn frontend_instruct(&self, _s: &OrchestrationState) -> anyhow::Result<String> {
|
||||
Ok(String::new())
|
||||
}
|
||||
async fn frontend_compile(&self, _s: &OrchestrationState) -> anyhow::Result<String> {
|
||||
Ok(String::new())
|
||||
}
|
||||
async fn execute(&self, _s: &OrchestrationState) -> anyhow::Result<ExecuteOutcome> {
|
||||
Ok(ExecuteOutcome {
|
||||
reply: String::new(),
|
||||
trace: String::new(),
|
||||
})
|
||||
}
|
||||
async fn compress(&self, _s: &OrchestrationState) -> anyhow::Result<CompressedEntry> {
|
||||
Ok(CompressedEntry::default())
|
||||
}
|
||||
async fn world_diff(&self, _s: &OrchestrationState) -> anyhow::Result<WorldDiffEntry> {
|
||||
Ok(WorldDiffEntry::default())
|
||||
}
|
||||
async fn context_utilization(&self, _s: &OrchestrationState) -> anyhow::Result<f32> {
|
||||
Ok(0.0)
|
||||
}
|
||||
async fn evict(&self, _s: &OrchestrationState) -> anyhow::Result<EvictionOutcome> {
|
||||
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())
|
||||
}
|
||||
@@ -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<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();
|
||||
// `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::<OrchestrationState>::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<GraphTopology> {
|
||||
struct NoopRuntime;
|
||||
#[async_trait]
|
||||
impl OrchestrationRuntime for NoopRuntime {
|
||||
async fn frontend_instruct(&self, _s: &OrchestrationState) -> anyhow::Result<String> {
|
||||
Ok(String::new())
|
||||
}
|
||||
async fn frontend_compile(&self, _s: &OrchestrationState) -> anyhow::Result<String> {
|
||||
Ok(String::new())
|
||||
}
|
||||
async fn execute(&self, _s: &OrchestrationState) -> anyhow::Result<ExecuteOutcome> {
|
||||
Ok(ExecuteOutcome {
|
||||
reply: String::new(),
|
||||
trace: String::new(),
|
||||
})
|
||||
}
|
||||
async fn compress(&self, _s: &OrchestrationState) -> anyhow::Result<CompressedEntry> {
|
||||
Ok(CompressedEntry::default())
|
||||
}
|
||||
async fn world_diff(&self, _s: &OrchestrationState) -> anyhow::Result<WorldDiffEntry> {
|
||||
Ok(WorldDiffEntry::default())
|
||||
}
|
||||
async fn context_utilization(&self, _s: &OrchestrationState) -> anyhow::Result<f32> {
|
||||
Ok(0.0)
|
||||
}
|
||||
async fn evict(&self, _s: &OrchestrationState) -> anyhow::Result<EvictionOutcome> {
|
||||
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;
|
||||
|
||||
@@ -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<OrchestrationMessage>,
|
||||
) -> 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()
|
||||
}
|
||||
|
||||
@@ -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: `<counterpart>#<session>#<latest_seq>`.
|
||||
assert!(event_signature(&s).contains("cycle=@peer#h1#0"));
|
||||
assert!(event_signature(&s).contains("reply=no"));
|
||||
assert_eq!(world_mutation(&s), "(no reply)");
|
||||
|
||||
|
||||
@@ -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<dyn OrchestrationRuntime>,
|
||||
) -> 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"),
|
||||
|
||||
@@ -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<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()
|
||||
}
|
||||
pub use steering::{current_steering, with_steering, DEFAULT_STEERING};
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<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()
|
||||
}
|
||||
@@ -359,11 +359,16 @@ fn handle_status(_params: Map<String, Value>) -> 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<String> =
|
||||
conn.query_row("SELECT MAX(last_message_at) FROM sessions", [], |r| {
|
||||
r.get::<_, Option<String>>(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<String> = conn.query_row(
|
||||
"SELECT MAX(last_message_at) FROM sessions \
|
||||
WHERE session_id NOT IN ('master', 'subconscious')",
|
||||
[],
|
||||
|r| r.get::<_, Option<String>>(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))
|
||||
|
||||
@@ -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<T>(
|
||||
.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();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user