diff --git a/docs/tinyagents-full-migration-plan/99-deletion-ledger.md b/docs/tinyagents-full-migration-plan/99-deletion-ledger.md index 80adf8b5a..e615cb1bc 100644 --- a/docs/tinyagents-full-migration-plan/99-deletion-ledger.md +++ b/docs/tinyagents-full-migration-plan/99-deletion-ledger.md @@ -18,6 +18,15 @@ they land. - [x] `context/pipeline.rs` (454) + `context/guard.rs` (236, keep stats structs) — 03.1 - [x] `context/tool_result_budget.rs` (172) — 03.1 - [x] `harness/payload_summarizer.rs` (490) — 01.4 +- [x] `tinyagents/middleware.rs::MicrocompactMiddleware` struct + impl (~46) — + W2-microcompact (2026-07-03): upstreamed into the vendored crate as + `tinyagents::harness::middleware::MicrocompactMiddleware` + (`tinyhumansai/tinyagents@feat/microcompact-middleware`, gitlink bumped); + OpenHuman now constructs the crate type with `CLEARED_PLACEHOLDER` and + events off, so behavior is byte-identical. The in-house struct + impl are + deleted; the retained OpenHuman tests assert parity against the crate type. + Was the C3-corrected/C5 "extract-then-delete" item (the local microcompact + was NOT 1.5.0-superseded; this PR did the extraction). ## Deletable after SDK-surface adoption @@ -84,6 +93,19 @@ they land. crate-internal `agent/harness/turn_subagent_usage.rs` (176) task-local — 06 (live until crate budget/run-tree accounting avoids duplicate `UsageRecorded` and covers parent-turn rollups) + W2-budget-dedupe (2026-07-03): dedupe guard landed — the event bridge now + records a model call's `UsageRecorded` exactly once, keyed on the run-scoped + iteration, so the observe-only crate `BudgetMiddleware`'s re-emit can't + double-count (`observability::OpenhumanEventBridge::record_usage`, `[budget]`). + Crate `BudgetMiddleware` installed OBSERVE-ONLY (empty `BudgetLimits`) at + `tinyagents/mod.rs`; local `CostBudgetMiddleware` demoted to a + divergence-logging shadow (`[budget_shadow]`, `after_agent`) but STILL + authoritative for enforcement. Flip criteria (must ALL hold before deleting + this row): (1) ≥ 500 parent+subagent turns with zero `[budget_shadow]` + divergence; (2) crate pricing table wired for money budgets; (3) run-tree + rollup via a shared `BudgetTracker` replacing the `turn_subagent_usage` + task-local. See the flip-criteria comment at the `tinyagents/mod.rs` + registration site. - [ ] `agent/dispatcher.rs` (609) + `harness/parse.rs` (833) legacy tool-call parsing — after XML/P-format transcripts read from the store and no live path parses provider text (04.2 + verify) @@ -94,6 +116,15 @@ they land. ## Deletable after session-store cutover (04.2 phase 4) +> **04.2 phase 2 landed (W2-shadow-reads, 2026-07-03):** a store-backed shadow +> reader runs beside the legacy transcript reader, normalizes both sides via +> `session_import/convert.rs`, and logs `[session_shadow_read]` divergence +> (compact, no-PII). Legacy stays authoritative; gated by +> `agent.session_shadow_reads` (default OFF) + `OPENHUMAN_SESSION_SHADOW_READS` +> kill switch. The rows below stay `[ ]` until the shadow logs are +> divergence-clean across the fixture matrix and reads are flipped to the store +> (phase 3), which is the precondition for these deletions. + - [ ] `session/transcript.rs` (1347) + tests (978) - [ ] `session/migration.rs` (373) + tests - [ ] `session/turn/session_io.rs` (391) diff --git a/docs/tinyagents-full-migration-plan/HANDOFF-2026-07-03.md b/docs/tinyagents-full-migration-plan/HANDOFF-2026-07-03.md index 934aab4a8..136df4bf2 100644 --- a/docs/tinyagents-full-migration-plan/HANDOFF-2026-07-03.md +++ b/docs/tinyagents-full-migration-plan/HANDOFF-2026-07-03.md @@ -102,50 +102,69 @@ All slice branches were created FROM `feat/tinyagents-c0-15-baseline` / `mod.rs` when merged together — merge C0 first, then slices one at a time, resolving against C0's tree. -## Wave 2 status (2026-07-03, late) +## Wave 1 (merged) All five wave-1 slice branches were MERGED into -`feat/tinyagents-c0-15-baseline` (light conflicts only; fmt commit on top; -50 targeted tests green post-merge) and pushed to PR **#4473**, which now -carries the entire wave-1 scope. +`feat/tinyagents-c0-15-baseline` and pushed to PR **#4473**, which landed the +entire wave-1 scope into `tinyhumansai/openhuman:main`. -Wave 2 was launched as a 4-agent workflow but **died on the monthly spend -limit** before any work landed. Its scoped slices (prompts preserved in the -session workflow script `tinyagents-continuation-wave2-*.js`, resumable via -`resumeFromRunId: wf_d6f5d26d-7e4`): +## Wave 2 status (2026-07-04) — DONE, on `feat/tinyagents-wave2` -1. `W2-shadow-reads` — 04.2 phase 2: store-backed shadow reader, compare with - legacy render, log divergence, legacy stays authoritative - (flag `agent.session_shadow_reads`, env kill switch). -2. `W2-budget-dedupe` — single-owner `UsageRecorded` recording (dedupe guard) - → install crate `BudgetMiddleware` observe-only; local `CostBudgetMiddleware` - demoted to divergence-logging shadow; flip criteria documented. -3. `W2-microcompact-upstream` — implement microcompact IN `vendor/tinyagents` - (branch `feat/microcompact-middleware`), push submodule upstream, then swap - OpenHuman to the crate version + delete local (gitlink bump ONLY if the - submodule push succeeded). -4. `W2-replay-rpc` — `openhuman.agent_run_events` (paged, `next_offset`), - `agent_run_status`, `agent_runs_active` controllers over the C4 journal/ - status seams, registry pattern. +All four wave-2 slices are implemented and integrated onto ONE branch +`feat/tinyagents-wave2` (off `upstream/main`), one focused commit each, and +opened as a single combined PR. Adapter-first throughout: legacy authoritative, +crate features shadow + log divergence, deletions gated on proven parity. + +1. `W2-microcompact-upstream` — **done.** The generic + `MicrocompactMiddleware` (caller-supplied placeholder, opt-in `Compressed` + event, idempotent tool-body clearing) was implemented IN the vendored crate + and pushed to `tinyhumansai/tinyagents@feat/microcompact-middleware` + (commit `ac73382`) BEFORE the gitlink bump. OpenHuman swapped to the crate + type (constructed with `CLEARED_PLACEHOLDER`, events off → byte-identical); + local struct+impl deleted, OpenHuman tests retargeted as the parity contract. + Crate tests: 5 green. Ledger row ticked. +2. `W2-shadow-reads` — **done.** 04.2 phase 2 store-backed shadow reader beside + the legacy transcript reader; `[session_shadow_read]` divergence logging; + flag `agent.session_shadow_reads` (default OFF) + `OPENHUMAN_SESSION_SHADOW_READS` + kill switch. Legacy authoritative. (Flip → reads-from-store is phase 3, the + ~9k-line deletion unlock.) +3. `W2-budget-dedupe` — **done.** Event bridge records each model call's usage + exactly once (dedupe guard keyed on the run-scoped iteration cursor); crate + `BudgetMiddleware` installed observe-only (empty `BudgetLimits`); local + `CostBudgetMiddleware` demoted to a `[budget_shadow]` divergence logger, still + authoritative for enforcement. Three flip criteria documented at the + `tinyagents/mod.rs` registration site + in the ledger. +4. `W2-replay-rpc` — **done.** Read-only `openhuman.agent_run_events` (paged, + `next_offset`, capped limit), `agent_run_status`, `agent_runs_active` + controllers over the C4 journal/status seams; direct `AgentObservation`/ + `HarnessRunStatus` serde projection (no PII); registered via + `src/core/all.rs` per the controller-migration checklist. (Resolves the + "replay RPC unexposed" C4 follow-up.) + +**Test note:** targeted crate tests (microcompact) ran green locally; the full +core-crate test binary would not link locally under this box's memory ceiling +(single-rustc codegen thrash), so `cargo check --lib --tests` is the local +compile gate and CI runs the actual targeted + full suites + coverage. ## Merge / PR state -- Wave-1 execution branches are LOCAL (not pushed) except as noted below. -- C0 PR: see PR link in the section below / `gh pr list --repo - tinyhumansai/openhuman --author @me`. -- Git etiquette (user rules): push to `origin` (senamakel fork), PR against - `upstream` (tinyhumansai) with `--head senamakel:`; explicit - `git add ` only; never commit on main. +- `feat/tinyagents-wave2` pushed to `origin` (senamakel fork); ONE combined PR + vs `upstream` (`--head senamakel:feat/tinyagents-wave2`): **PR #4483** + (`tinyhumansai/openhuman#4483`). +- Submodule branch `feat/microcompact-middleware` pushed to + `tinyhumansai/tinyagents`; gitlink bumped to `ac73382`. Fresh worktrees must + `git submodule update --init vendor/tinyagents`. +- Git etiquette (user rules): push to `origin`, PR against `upstream` with + `--head senamakel:`; explicit `git add `; never commit on main. -## Suggested next steps (wave 2) +## Suggested next steps (wave 3) -1. Merge C0 PR; rebase + push + PR the five slice branches (stack on C0). -2. 04.2 shadow reads → read cutover (biggest deletion unlock, ~9k lines gated - on it, incl. dispatcher/parse/pformat). -3. C5 upstream extractions INTO `vendor/tinyagents` (now editable in-tree): - microcompact (new — see corrections), multimodal resolver, dialect layer, - overflow-to-artifact, hooks traits. -4. C3 remainder: UsageRecorded de-dup → crate `BudgetMiddleware` → delete - local `CostBudgetMiddleware`; exposure-shadow parity audit → flip owner. -5. Flip C2 shadows to authoritative once divergence logs are clean; wire the - goals migration helper to boot. +1. Flip `session_shadow_reads` → reads-from-store once the fixture-matrix + divergence logs are clean (biggest deletion unlock, ~9k incl. + dispatcher/parse/pformat). +2. Flip crate `BudgetMiddleware` → enforcing owner once the 3 documented + criteria hold; delete local `CostBudgetMiddleware` + `turn_subagent_usage.rs`. +3. Open a tinyagents PR for `feat/microcompact-middleware` (currently a pushed + branch, not merged) so the gitlink can later track a tagged release. +4. Continue C5 upstream extractions (multimodal resolver, dialect layer, + overflow-to-artifact, hooks traits) + C2 shadow → authoritative flips. diff --git a/src/core/all.rs b/src/core/all.rs index ab5eece79..1ed2d06e6 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -131,6 +131,10 @@ fn build_registered_controllers() -> Vec { controllers.extend(crate::openhuman::webview_apis::all_webview_apis_registered_controllers()); // Agent definition and prompt inspection controllers.extend(crate::openhuman::agent::all_agent_registered_controllers()); + // Read-only agent run replay + status over the durable journal/status seams + // (agent_run_events / agent_run_status / agent_runs_active). + controllers + .extend(crate::openhuman::tinyagents::replay::all_agent_replay_registered_controllers()); // Persistent agent profiles (flavours): name, soul, memory sources, skills, MCP, connectors. controllers.extend(crate::openhuman::profiles::all_profiles_registered_controllers()); // User-facing agent registry: defaults, enablement, custom agents, tool policy. @@ -383,6 +387,8 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::mcp_registry::all_mcp_registry_controller_schemas()); schemas.extend(crate::openhuman::webview_apis::all_webview_apis_controller_schemas()); schemas.extend(crate::openhuman::agent::all_agent_controller_schemas()); + // Read-only agent run replay + status controllers (workstream 05.x). + schemas.extend(crate::openhuman::tinyagents::replay::all_agent_replay_controller_schemas()); schemas.extend(crate::openhuman::profiles::all_profiles_controller_schemas()); schemas.extend(crate::openhuman::agent_registry::all_agent_registry_controller_schemas()); schemas.extend(crate::openhuman::agent_experience::all_agent_experience_controller_schemas()); diff --git a/src/openhuman/agent/harness/session/turn/session_io.rs b/src/openhuman/agent/harness/session/turn/session_io.rs index 1bd40ef56..41acb8c38 100644 --- a/src/openhuman/agent/harness/session/turn/session_io.rs +++ b/src/openhuman/agent/harness/session/turn/session_io.rs @@ -34,6 +34,11 @@ impl Agent { } let loaded_count = session.messages.len(); log::info!("[transcript] loaded {} messages for resume", loaded_count); + // Best-effort store-backed shadow read (issue #4249, + // 04.2 phase 2). Observes + logs divergence only; the + // legacy transcript just loaded stays authoritative and + // is what feeds the resume below. Gated OFF by default. + self.maybe_shadow_read_session_store(&path, &session); let bounded = self.bound_cached_transcript_messages(session.messages); if bounded.len() < loaded_count { log::warn!( @@ -322,6 +327,59 @@ impl Agent { }); } + /// Store-backed **shadow read** of a just-loaded session transcript. + /// + /// Beside the legacy authoritative reader (`try_load_session_transcript`), + /// read the same session back from the TinyAgents journal store, normalize + /// both sides through the importer's `session_import::convert` machinery, + /// compare, and log any divergence (`[session_shadow_read]`, issue #4249, + /// 04.2 phase 2). Additive and gated on the default-**OFF** + /// `AgentConfig::session_shadow_reads` flag + /// (`OPENHUMAN_SESSION_SHADOW_READS` is a kill switch): when disabled this + /// is a cheap early return. + /// + /// The legacy transcript stays authoritative — this only observes. The + /// comparison runs on a spawned background task so it never slows the + /// authoritative read, and every store-read error is treated as "no shadow + /// available" (logged at debug), never propagated. + fn maybe_shadow_read_session_store( + &self, + path: &std::path::Path, + session: &transcript::SessionTranscript, + ) { + use crate::openhuman::session_import::live; + + // Config flag (default OFF) gates the shadow read; the env kill switch + // can still force it off. `self.config` is the effective per-agent config. + if !live::shadow_reads_enabled(self.config.session_shadow_reads) { + return; + } + + // Same session key the write side / importer use: the transcript stem. + let Some(stem) = path + .file_stem() + .and_then(|s| s.to_str()) + .map(str::to_string) + else { + log::debug!( + "[session_shadow_read] skipped: no file stem for {}", + path.display() + ); + return; + }; + + let workspace = self.workspace_dir.clone(); + let transcript = session.clone(); + log::debug!( + "[session_shadow_read] scheduled stem={stem} workspace={} legacy_messages={}", + workspace.display(), + transcript.messages.len() + ); + tokio::spawn(async move { + let _ = live::shadow_read_compare(&workspace, &stem, &transcript).await; + }); + } + // ───────────────────────────────────────────────────────────────── // Session-memory extraction. // ───────────────────────────────────────────────────────────────── diff --git a/src/openhuman/config/schema/agent.rs b/src/openhuman/config/schema/agent.rs index 8fdd4ffe4..6d163fa9e 100644 --- a/src/openhuman/config/schema/agent.rs +++ b/src/openhuman/config/schema/agent.rs @@ -259,12 +259,36 @@ pub struct AgentConfig { /// [`crate::openhuman::session_import::live::dual_write_enabled`]. #[serde(default = "default_session_dual_write")] pub session_dual_write: bool, + + /// Store-backed **shadow read** of a resumed session's messages: on the + /// legacy transcript read path (`session/turn/session_io.rs` → + /// `try_load_session_transcript`), also read the same session back from the + /// TinyAgents journal (`{workspace}/tinyagents_store/journal`), normalize + /// both sides through the importer's `session_import::convert` machinery, + /// compare, and log any divergence (`[session_shadow_read]`, issue #4249, + /// sessions 04.2 phase 2). + /// + /// Defaults **OFF** (unlike `session_dual_write`, which defaults ON): this + /// is an observation-only parity probe with no product effect. The legacy + /// JSONL read stays authoritative — the shadow read only observes and logs + /// on a background task; a store-read failure is treated as "no shadow + /// available" and never breaks or slows the authoritative read. The + /// `OPENHUMAN_SESSION_SHADOW_READS` env var is a pure **kill switch**: a + /// falsy value (`0`/`false`/`no`/`off`/`disable`) forces the shadow read + /// OFF regardless of config; it can never force it ON. See + /// [`crate::openhuman::session_import::live::shadow_reads_enabled`]. + #[serde(default = "default_session_shadow_reads")] + pub session_shadow_reads: bool, } fn default_session_dual_write() -> bool { true } +fn default_session_shadow_reads() -> bool { + false +} + fn default_tool_result_budget_bytes() -> usize { crate::openhuman::context::DEFAULT_TOOL_RESULT_BUDGET_BYTES } @@ -395,6 +419,7 @@ impl Default for AgentConfig { tool_result_budget_bytes: default_tool_result_budget_bytes(), agent_timeout_secs: default_agent_timeout_secs(), session_dual_write: default_session_dual_write(), + session_shadow_reads: default_session_shadow_reads(), } } } diff --git a/src/openhuman/session_import/live.rs b/src/openhuman/session_import/live.rs index 275f2cbf8..b35798e42 100644 --- a/src/openhuman/session_import/live.rs +++ b/src/openhuman/session_import/live.rs @@ -30,7 +30,7 @@ use super::convert::{ build_descriptor, effective_thread_id, journal_messages, sanitize_store_name, stream_name, }; use super::ops::{open_session_stores, SessionStores}; -use super::types::{DescriptorSource, NS_SESSIONS}; +use super::types::{DescriptorSource, JournalMessage, NS_SESSIONS}; /// Kill-switch env var for the live session-store dual-write. The config flag /// (`AgentConfig::session_dual_write`) defaults ON; setting this env var to a @@ -38,12 +38,18 @@ use super::types::{DescriptorSource, NS_SESSIONS}; /// [`dual_write_enabled`]. const DUAL_WRITE_ENV: &str = "OPENHUMAN_SESSION_DUAL_WRITE"; -/// Whether the `OPENHUMAN_SESSION_DUAL_WRITE` kill switch is engaged (set to a -/// falsey value). Unset — or any non-falsey value — leaves the mirror driven by -/// the config flag. Read live (not cached) so a config reload / env change is -/// honored on the next turn. -fn kill_switch_engaged() -> bool { - match std::env::var(DUAL_WRITE_ENV) { +/// Kill-switch env var for the store-backed session shadow read. The config +/// flag (`AgentConfig::session_shadow_reads`) defaults OFF; setting this env +/// var to a falsey value forces the shadow read OFF even when the flag is ON. +/// It can never force the shadow read ON. See [`shadow_reads_enabled`]. +const SHADOW_READ_ENV: &str = "OPENHUMAN_SESSION_SHADOW_READS"; + +/// Whether `var` is set to a case-insensitive falsey value +/// (`0`/`false`/`no`/`off`/`disable`/`disabled`). Unset — or any non-falsey +/// value — is not a kill. Read live (not cached) so a config reload / env +/// change is honored on the next turn/read. +fn env_kill_switch_engaged(var: &str) -> bool { + match std::env::var(var) { Ok(v) => matches!( v.trim().to_ascii_lowercase().as_str(), "0" | "false" | "no" | "off" | "disable" | "disabled" @@ -52,6 +58,14 @@ fn kill_switch_engaged() -> bool { } } +/// Whether the `OPENHUMAN_SESSION_DUAL_WRITE` kill switch is engaged (set to a +/// falsey value). Unset — or any non-falsey value — leaves the mirror driven by +/// the config flag. Read live (not cached) so a config reload / env change is +/// honored on the next turn. +fn kill_switch_engaged() -> bool { + env_kill_switch_engaged(DUAL_WRITE_ENV) +} + /// Store-registry name under which the session KV store is registered on each /// turn's `RunContext.stores` (issue #4249, 04.1). Slash-free so it round-trips /// the crate `FileStore` name sanitizer. This is a forward-looking, @@ -191,3 +205,147 @@ pub async fn write_live_turn( ); Ok(()) } + +// ───────────────────────────────────────────────────────────────────────────── +// Store-backed SHADOW READ (issue #4249, sessions 04.2 phase 2) +// +// Beside the legacy authoritative transcript reader +// (`session/turn/session_io.rs` → `try_load_session_transcript`), read the +// same session's messages back from the crate journal store, normalize both +// sides through the same `convert` machinery the dual-write uses, compare, and +// log divergence. Legacy stays authoritative: this observes + logs only and +// never affects, fails, or slows the authoritative read. +// ───────────────────────────────────────────────────────────────────────────── + +/// Whether the store-backed session **shadow read** is enabled for this read. +/// +/// `config_enabled` is the `AgentConfig::session_shadow_reads` flag, which +/// **defaults OFF** (unlike `session_dual_write`). The +/// `OPENHUMAN_SESSION_SHADOW_READS` env var is a pure kill switch: an explicit +/// falsey value (case-insensitive `0`/`false`/`no`/`off`/`disable`/`disabled`) +/// forces the shadow read OFF regardless of config; it can never force it ON. +/// Read live (never cached) so a config reload / env change is honored on the +/// next read. Mirrors the [`dual_write_enabled`] flag/env idiom exactly, only +/// with the default flipped and no default-on behavior. +pub fn shadow_reads_enabled(config_enabled: bool) -> bool { + let killed = env_kill_switch_engaged(SHADOW_READ_ENV); + let enabled = config_enabled && !killed; + log::debug!( + "[session_shadow_read] decision config_enabled={config_enabled} kill_switch={killed} enabled={enabled}" + ); + enabled +} + +/// Outcome of one shadow-read comparison. Returned for tests/observability; +/// the compact divergence summary is also logged (`[session_shadow_read]`). +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ShadowReadOutcome { + /// The store stream rendered exactly the legacy transcript's messages. + Match { messages: usize }, + /// The store stream diverged from the legacy render. Carries only compact + /// counts + the first differing index — never message bodies (PII). + Divergence { + legacy: usize, + shadow: usize, + first_diff: Option, + }, + /// No shadow available: the store read errored, or the stream is + /// empty/absent for a non-empty legacy transcript (e.g. dual-write was off + /// when this session was written). Treated as non-divergent — the legacy + /// read is authoritative regardless. + Unavailable, +} + +/// Read a session's messages back from the crate journal store +/// (`{workspace}/tinyagents_store/journal`, stream `session.{stem}.messages`) +/// as normalized [`JournalMessage`]s — the same shape the importer and live +/// dual-write write. A missing stream yields an empty vec (not an error). +async fn read_shadow_messages(workspace: &Path, session_key: &str) -> Result> { + let SessionStores { journal, .. } = open_session_stores(workspace); + let stream = stream_name(session_key); + let records = journal + .read_from(&stream, 0) + .await + .with_context(|| format!("shadow read of stream {stream}"))?; + let mut out = Vec::with_capacity(records.len()); + for (offset, value) in records { + let msg: JournalMessage = serde_json::from_value(value) + .with_context(|| format!("shadow record shape at offset {offset}"))?; + out.push(msg); + } + Ok(out) +} + +/// Shadow-read the given session back from the store and compare it against the +/// legacy transcript, logging divergence. Legacy stays authoritative — the +/// caller ignores the returned outcome for control flow (it exists for tests / +/// observability). Best-effort: any store-read error is logged at debug and +/// reported as [`ShadowReadOutcome::Unavailable`]; it never breaks or slows the +/// authoritative read. +/// +/// Both sides are normalized through the importer's `convert` machinery +/// ([`journal_messages`]) so live/legacy renders are directly comparable, then +/// compared by message count and normalized content. On mismatch a **compact** +/// summary (counts + first differing index) is warn-logged; message bodies are +/// never emitted (PII). +pub async fn shadow_read_compare( + workspace: &Path, + session_key: &str, + legacy: &SessionTranscript, +) -> ShadowReadOutcome { + let expected = journal_messages(legacy); + log::debug!( + "[session_shadow_read] enter stem={session_key} workspace={} legacy_messages={}", + workspace.display(), + expected.len() + ); + + let shadow = match read_shadow_messages(workspace, session_key).await { + Ok(v) => v, + Err(err) => { + log::debug!( + "[session_shadow_read] store read error stem={session_key}: {err:#} — no shadow available" + ); + return ShadowReadOutcome::Unavailable; + } + }; + + // Empty/absent store stream against a non-empty legacy transcript: the + // session simply was not mirrored (dual-write off when it was written). + // Treat as "no shadow" rather than a spurious divergence. + if shadow.is_empty() && !expected.is_empty() { + log::debug!( + "[session_shadow_read] no store stream stem={session_key} legacy_messages={} — no shadow available", + expected.len() + ); + return ShadowReadOutcome::Unavailable; + } + + if shadow == expected { + log::debug!( + "[session_shadow_read] parity OK stem={session_key} messages={}", + expected.len() + ); + return ShadowReadOutcome::Match { + messages: expected.len(), + }; + } + + // Divergence: first index where the two normalized renders differ, or the + // shorter length when one is a strict prefix of the other. Compact only. + let first_diff = expected + .iter() + .zip(shadow.iter()) + .position(|(a, b)| a != b) + .or_else(|| (expected.len() != shadow.len()).then(|| expected.len().min(shadow.len()))); + log::warn!( + "[session_shadow_read] DIVERGENCE stem={session_key} legacy_count={} shadow_count={} first_diff={first_diff:?}", + expected.len(), + shadow.len() + ); + ShadowReadOutcome::Divergence { + legacy: expected.len(), + shadow: shadow.len(), + first_diff, + } +} diff --git a/src/openhuman/session_import/live_tests.rs b/src/openhuman/session_import/live_tests.rs index a42b70d9d..5d9073dad 100644 --- a/src/openhuman/session_import/live_tests.rs +++ b/src/openhuman/session_import/live_tests.rs @@ -14,7 +14,10 @@ use tempfile::TempDir; use tinyagents::harness::store::{AppendStore, FileStore, JsonlAppendStore, Store}; use super::convert::{sanitize_store_name, stream_name}; -use super::live::{dual_write_enabled, write_live_turn}; +use super::live::{ + dual_write_enabled, shadow_read_compare, shadow_reads_enabled, write_live_turn, + ShadowReadOutcome, +}; use super::ops::store_root; use super::types::{JournalMessage, SessionDescriptor, NS_SESSIONS}; use crate::openhuman::agent::harness::session::transcript::{ @@ -204,3 +207,148 @@ fn config_flag_and_env_kill_switch() { None => std::env::remove_var(ENV), } } + +// ── Store-backed shadow read (issue #4249, 04.2 phase 2) ──────────────────── + +/// A session written by the dual-write and then read back through the shadow +/// reader must render byte-for-byte the messages the legacy JSONL reader +/// produces — no divergence. This is the read-path twin of +/// [`live_dual_write_matches_legacy_jsonl_render`]: it drives the *same* +/// writers, then compares via the actual `shadow_read_compare` path and asserts +/// a clean [`ShadowReadOutcome::Match`]. +#[tokio::test] +async fn shadow_read_roundtrip_matches_legacy() { + let ws = TempDir::new().expect("tempdir"); + let stem = "1719_orchestrator"; + let jsonl_path = ws.path().join("session_raw").join(format!("{stem}.jsonl")); + + let base_messages = vec![ChatMessage::user("hi"), ChatMessage::assistant("done")]; + let meta = meta("t-root"); + let usage = turn_usage(); + + // (1) Legacy authoritative write. (2) Live dual-write into the store. + write_transcript(&jsonl_path, &base_messages, &meta, Some(&usage)).expect("legacy write"); + let mut live_messages = base_messages.clone(); + let last_assistant = live_messages + .iter() + .rposition(|m| m.role == "assistant") + .expect("assistant message present"); + attach_turn_usage_metadata(&mut live_messages[last_assistant], &usage); + let store_transcript = SessionTranscript { + meta: meta.clone(), + messages: live_messages, + }; + write_live_turn(ws.path(), stem, &store_transcript) + .await + .expect("live dual-write"); + + // The legacy reader materializes this SessionTranscript for a resume; the + // shadow reader compares it against the store stream. + let legacy = read_transcript(&jsonl_path).expect("read legacy transcript"); + let outcome = shadow_read_compare(ws.path(), stem, &legacy).await; + assert_eq!( + outcome, + ShadowReadOutcome::Match { + messages: legacy.messages.len() + }, + "round-tripped shadow read must match the legacy render with no divergence" + ); +} + +/// When the store has no stream for the session (dual-write never ran), the +/// shadow read reports [`ShadowReadOutcome::Unavailable`] rather than a spurious +/// divergence, and a content mismatch is reported as +/// [`ShadowReadOutcome::Divergence`] with a compact first-diff index. +#[tokio::test] +async fn shadow_read_unavailable_and_divergence() { + let ws = TempDir::new().expect("tempdir"); + let stem = "1719_orchestrator"; + let meta = meta("t-root"); + + // No store write yet: empty/absent stream against a non-empty legacy + // transcript → Unavailable (no shadow), never a divergence. + let legacy = SessionTranscript { + meta: meta.clone(), + messages: vec![ChatMessage::user("hi"), ChatMessage::assistant("done")], + }; + assert_eq!( + shadow_read_compare(ws.path(), stem, &legacy).await, + ShadowReadOutcome::Unavailable, + "absent store stream must be reported as no shadow available" + ); + + // Now mirror the two-message transcript, then compare against a legacy + // transcript that has an extra trailing message: divergence at index 2. + write_live_turn(ws.path(), stem, &legacy) + .await + .expect("live dual-write"); + let diverging = SessionTranscript { + meta, + messages: vec![ + ChatMessage::user("hi"), + ChatMessage::assistant("done"), + ChatMessage::user("more"), + ], + }; + assert_eq!( + shadow_read_compare(ws.path(), stem, &diverging).await, + ShadowReadOutcome::Divergence { + legacy: 3, + shadow: 2, + first_diff: Some(2), + }, + "a trailing legacy-only message must be reported as a compact divergence" + ); +} + +/// The shadow read is driven by the `AgentConfig::session_shadow_reads` config +/// flag (default **OFF**) with the `OPENHUMAN_SESSION_SHADOW_READS` env var as a +/// pure kill switch (can only force OFF, never ON). This exercises the decision +/// matrix directly — the gate `maybe_shadow_read_session_store` early-returns +/// (never invoking the reader) whenever this returns `false`. Env mutation is +/// process-global, so all assertions live in one serial test and the var is +/// restored on exit. +#[test] +fn shadow_read_flag_and_env_kill_switch() { + const ENV: &str = "OPENHUMAN_SESSION_SHADOW_READS"; + let prior = std::env::var(ENV).ok(); + + // Config OFF (the default) disables regardless of env — reader not invoked. + std::env::remove_var(ENV); + assert!( + !shadow_reads_enabled(false), + "config off (default) disables the shadow read" + ); + + // Config ON enables when the env is unset. + assert!( + shadow_reads_enabled(true), + "config on + no env enables the shadow read" + ); + + // A falsey env value is the kill switch: forces OFF even with config ON. + for killed in ["0", "false", "no", "off", "disable", "disabled", "OFF"] { + std::env::set_var(ENV, killed); + assert!( + !shadow_reads_enabled(true), + "kill switch value {killed:?} must force off even with flag on" + ); + } + + // A non-falsey env value does not force on: config still governs, and it + // can never turn a default-off flag on. + std::env::set_var(ENV, "1"); + assert!( + shadow_reads_enabled(true), + "non-falsey env leaves config ON on" + ); + assert!( + !shadow_reads_enabled(false), + "non-falsey env cannot force a default-off flag on" + ); + + match prior { + Some(v) => std::env::set_var(ENV, v), + None => std::env::remove_var(ENV), + } +} diff --git a/src/openhuman/tinyagents/middleware.rs b/src/openhuman/tinyagents/middleware.rs index 23421c80f..e53e5c68d 100644 --- a/src/openhuman/tinyagents/middleware.rs +++ b/src/openhuman/tinyagents/middleware.rs @@ -9,7 +9,10 @@ //! //! - [`MicrocompactMiddleware`] (`before_model`) — clear the bodies of older //! tool-result messages (keeping the N most recent) so a long tool-heavy -//! thread stays cheap without dropping chat history. +//! thread stays cheap without dropping chat history. This is now the crate +//! [`tinyagents::harness::middleware::MicrocompactMiddleware`], constructed +//! with OpenHuman's [`CLEARED_PLACEHOLDER`] wording; the in-house copy was +//! upstreamed (see `99-deletion-ledger.md`). //! - [`ToolOutputMiddleware`] (`after_tool`) — apply the per-tool-result byte //! cap and (optionally) the semantic payload summarizer to each tool result //! as it returns, before it enters the transcript. @@ -28,8 +31,8 @@ use tinyagents::harness::context::RunContext; use tinyagents::harness::events::AgentEvent; use tinyagents::harness::message::{ContentBlock, Message as TaMessage}; use tinyagents::harness::middleware::{ - AgentRun, ContextualToolSelectionMiddleware, Middleware, MiddlewareToolOutcome, - ToolAllowlistMiddleware, ToolHandler, ToolMiddleware, + AgentRun, BudgetTracker, ContextualToolSelectionMiddleware, MicrocompactMiddleware, Middleware, + MiddlewareToolOutcome, ToolAllowlistMiddleware, ToolHandler, ToolMiddleware, }; use tinyagents::harness::model::{ModelRequest, PromptSegment, SegmentRole}; use tinyagents::harness::no_progress::{NoProgress, NoProgressTracker, ToolAttempt}; @@ -268,9 +271,14 @@ impl TurnContextMiddleware { })); } if self.microcompact_keep_recent > 0 { - harness.push_middleware(Arc::new(MicrocompactMiddleware { - keep_recent: self.microcompact_keep_recent, - })); + // Crate middleware (upstreamed from the in-house copy). Constructed + // with OpenHuman's model-facing placeholder so behavior is + // byte-identical to the deleted local version. Events stay off (the + // default) to preserve the prior silent-rewrite behavior. + harness.push_middleware(Arc::new(MicrocompactMiddleware::new( + self.microcompact_keep_recent, + CLEARED_PLACEHOLDER, + ))); } // Handoff runs BEFORE the tool-output budget so an oversized payload is // stashed + replaced with a short placeholder first; the byte cap would @@ -638,53 +646,6 @@ impl Middleware<()> for PromptCacheSegmentMiddleware { } } -/// `before_model`: clear the bodies of older tool-result messages, keeping the -/// `keep_recent` most recent verbatim. The graph analogue of -/// `context::microcompact` — bounds a tool-heavy thread's cost without dropping -/// any chat turns. Idempotent: an already-cleared body is left as the -/// placeholder. -struct MicrocompactMiddleware { - keep_recent: usize, -} - -#[async_trait] -impl Middleware<()> for MicrocompactMiddleware { - fn name(&self) -> &str { - "microcompact" - } - - async fn before_model( - &self, - _ctx: &mut RunContext<()>, - _state: &(), - request: &mut ModelRequest, - ) -> TaResult<()> { - let tool_idxs: Vec = request - .messages - .iter() - .enumerate() - .filter(|(_, m)| matches!(m, TaMessage::Tool(_))) - .map(|(i, _)| i) - .collect(); - if tool_idxs.len() <= self.keep_recent { - return Ok(()); - } - let cut = tool_idxs.len() - self.keep_recent; - for &i in &tool_idxs[..cut] { - // Skip messages already reduced to the placeholder; otherwise swap the - // body for it (idempotent, preserves the tool_call_id). - if request.messages[i].text() == CLEARED_PLACEHOLDER { - continue; - } - if let TaMessage::Tool(t) = &request.messages[i] { - let id = t.tool_call_id.clone(); - request.messages[i] = TaMessage::tool(id, CLEARED_PLACEHOLDER); - } - } - Ok(()) - } -} - /// `after_tool`: apply the semantic payload summarizer (when configured) and /// then the hard per-tool-result byte cap to each tool result's model-facing /// content, before it enters the transcript. The graph analogue of the byte cap @@ -1448,14 +1409,54 @@ impl Middleware<()> for MemoryProtocolMiddleware { /// model call spends (issue #4249, Phase 5). Reads the global /// [`CostTracker`](crate::openhuman::cost) and, when cost budgets are configured /// and already exceeded, fails the run before the provider call; a warning -/// threshold logs but proceeds. +/// threshold logs but proceeds. This enforcement path stays **authoritative**. /// /// Self-gating: a no-op unless a global tracker exists and `config.enabled` with /// a limit is set (`check_budget` returns `Allowed` otherwise). Complements the /// post-call `StopHookMiddleware` per-turn USD cap. Projecting the *next* call's /// cost pre-spend (vs the already-exceeded check here) needs an input-token /// estimate — a follow-up. -pub(crate) struct CostBudgetMiddleware; +/// +/// # Shadow role (W2-budget-dedupe) +/// +/// When built with [`with_shadow`](Self::with_shadow), this middleware is ALSO a +/// divergence-logging shadow over the observe-only crate +/// [`BudgetMiddleware`](tinyagents::harness::middleware::BudgetMiddleware). It +/// keeps enforcing exactly as before, but at `after_agent` it compares the +/// crate `BudgetMiddleware`'s shared [`BudgetTracker`] accumulation against the +/// authoritative runtime [`AgentRun::usage`] and logs `[budget_shadow]` parity +/// or divergence (compact numeric summary; no PII). Both accumulate the same +/// per-call `response.usage`, so token totals must match once the crate +/// middleware is on the path — this is the parity signal that must be clean +/// before enforcement can flip to the crate owner (see the flip-criteria comment +/// at the registration site in `tinyagents/mod.rs`). Cost is intentionally NOT +/// compared: the observe-only crate middleware has no pricing table, so its cost +/// stays zero while the local path prices via `cost::catalog` — cost parity is a +/// flip-criteria follow-up. +pub(crate) struct CostBudgetMiddleware { + /// Observe-only crate `BudgetMiddleware`'s shared tracker handle, for the + /// end-of-run `[budget_shadow]` comparison. `None` when the shadow is not + /// installed (isolated unit tests of the enforcement gate). + shadow_tracker: Option, +} + +impl CostBudgetMiddleware { + /// Enforcement-only gate with no shadow comparison (isolated unit tests). + pub(crate) fn new() -> Self { + Self { + shadow_tracker: None, + } + } + + /// Enforcement gate that ALSO compares its per-run token accounting against + /// the observe-only crate `BudgetMiddleware`'s shared `tracker` at end of run + /// and logs `[budget_shadow]` parity/divergence. + pub(crate) fn with_shadow(tracker: BudgetTracker) -> Self { + Self { + shadow_tracker: Some(tracker), + } + } +} #[async_trait] impl Middleware<()> for CostBudgetMiddleware { @@ -1504,6 +1505,56 @@ impl Middleware<()> for CostBudgetMiddleware { _ => Ok(()), } } + + /// Shadow parity check (W2-budget-dedupe). Enforcement already happened per + /// call in `before_model`; here we only observe. Compares the observe-only + /// crate `BudgetMiddleware`'s accumulated token spend against the runtime's + /// authoritative `AgentRun::usage` and logs `[budget_shadow]` divergence. + /// Never fails the run. + async fn after_agent( + &self, + _ctx: &mut RunContext<()>, + _state: &(), + run: &mut AgentRun, + ) -> TaResult<()> { + let Some(tracker) = &self.shadow_tracker else { + return Ok(()); + }; + let crate_usage = tracker.snapshot().usage; // UsageTotals (crate shadow) + let local = run.usage; // UsageTotals (runtime authoritative) + let l = &local.usage; + let c = &crate_usage.usage; + let diverged = l.input_tokens != c.input_tokens + || l.output_tokens != c.output_tokens + || l.cache_read_tokens != c.cache_read_tokens + || l.total_tokens != c.total_tokens + || local.calls != crate_usage.calls; + if diverged { + tracing::warn!( + local_calls = local.calls, + crate_calls = crate_usage.calls, + local_in = l.input_tokens, + crate_in = c.input_tokens, + local_out = l.output_tokens, + crate_out = c.output_tokens, + local_cached = l.cache_read_tokens, + crate_cached = c.cache_read_tokens, + local_total = l.total_tokens, + crate_total = c.total_tokens, + "[budget_shadow] divergence: crate BudgetMiddleware token accounting differs from authoritative AgentRun.usage" + ); + } else { + tracing::debug!( + calls = local.calls, + input = l.input_tokens, + output = l.output_tokens, + cached = l.cache_read_tokens, + total = l.total_tokens, + "[budget_shadow] parity: crate BudgetMiddleware token accounting matches AgentRun.usage" + ); + } + Ok(()) + } } /// `after_tool`: stop (or nudge) the run when tool calls keep failing with no @@ -1841,11 +1892,15 @@ mod tests { assert_eq!(msgs[0].text(), "only system"); } - // ── MicrocompactMiddleware ────────────────────────────────────────────── + // ── MicrocompactMiddleware (crate) ────────────────────────────────────── + // + // These assert the crate `MicrocompactMiddleware`, constructed with + // OpenHuman's `CLEARED_PLACEHOLDER`, reproduces the deleted in-house + // middleware byte-for-byte — the parity contract for the upstream swap. #[tokio::test] async fn microcompact_clears_older_tool_bodies_and_keeps_recent() { - let mw = MicrocompactMiddleware { keep_recent: 1 }; + let mw = MicrocompactMiddleware::new(1, CLEARED_PLACEHOLDER); let mut req = ModelRequest::new(vec![ TaMessage::system("sys"), TaMessage::user("hello"), @@ -1869,7 +1924,7 @@ mod tests { #[tokio::test] async fn microcompact_is_a_noop_when_within_keep_recent() { - let mw = MicrocompactMiddleware { keep_recent: 5 }; + let mw = MicrocompactMiddleware::new(5, CLEARED_PLACEHOLDER); let mut req = ModelRequest::new(vec![TaMessage::tool("t1", "A"), TaMessage::tool("t2", "B")]); mw.before_model(&mut ctx(), &(), &mut req).await.unwrap(); @@ -1879,7 +1934,7 @@ mod tests { #[tokio::test] async fn microcompact_is_idempotent() { - let mw = MicrocompactMiddleware { keep_recent: 1 }; + let mw = MicrocompactMiddleware::new(1, CLEARED_PLACEHOLDER); let mut req = ModelRequest::new(vec![ TaMessage::tool("t1", "FIRST"), TaMessage::tool("t2", "SECOND"), @@ -1999,11 +2054,46 @@ mod tests { async fn cost_budget_is_a_noop_without_a_global_tracker() { // No global CostTracker is installed in the unit-test process, so the // gate self-disables and the model call proceeds. - let mw = CostBudgetMiddleware; + let mw = CostBudgetMiddleware::new(); let mut req = ModelRequest::new(vec![TaMessage::user("hi")]); assert!(mw.before_model(&mut ctx(), &(), &mut req).await.is_ok()); } + // ── CostBudgetMiddleware shadow (W2-budget-dedupe) ────────────────────── + + /// The shadow comparison at `after_agent` logs parity when the crate + /// `BudgetMiddleware`'s tracker matches the runtime `AgentRun.usage`, and + /// never fails the run — in both the matching and diverging cases. It also + /// must be inert (no panic, `Ok`) when no shadow tracker is installed. + #[tokio::test] + async fn cost_budget_shadow_after_agent_never_fails_the_run() { + use tinyagents::harness::usage::Usage; + + // No shadow tracker: after_agent is a silent no-op. + let plain = CostBudgetMiddleware::new(); + let mut run = AgentRun::new(); + run.usage.record(Usage::new(100, 40)); + assert!(plain.after_agent(&mut ctx(), &(), &mut run).await.is_ok()); + + // Matching tracker (parity): the crate tracker accumulated the same + // single call's usage the runtime recorded into `run.usage`. + let tracker = BudgetTracker::new(); + tracker.record(Usage::new(100, 40), Default::default()); + let shadow = CostBudgetMiddleware::with_shadow(tracker.clone()); + let mut run = AgentRun::new(); + run.usage.record(Usage::new(100, 40)); + assert!(shadow.after_agent(&mut ctx(), &(), &mut run).await.is_ok()); + + // Diverging tracker (crate missed a call): still only logs, never fails. + let mut diverged_run = AgentRun::new(); + diverged_run.usage.record(Usage::new(100, 40)); + diverged_run.usage.record(Usage::new(10, 5)); + assert!(shadow + .after_agent(&mut ctx(), &(), &mut diverged_run) + .await + .is_ok()); + } + // ── RepeatedToolFailureMiddleware ─────────────────────────────────────── fn failing_result(name: &str, err: &str) -> TaToolResult { diff --git a/src/openhuman/tinyagents/mod.rs b/src/openhuman/tinyagents/mod.rs index 4d79cdd92..c2af63d36 100644 --- a/src/openhuman/tinyagents/mod.rs +++ b/src/openhuman/tinyagents/mod.rs @@ -29,6 +29,7 @@ pub(crate) mod observability; pub(crate) mod orchestration; pub(crate) mod payload_summarizer; mod policy_denial; +pub(crate) mod replay; pub(crate) mod retriever; mod routes; mod run_cancellation_context; @@ -46,8 +47,8 @@ use tinyagents::harness::context::{RunConfig, RunContext}; use tinyagents::harness::events::EventSink; use tinyagents::harness::message::Message as TaMessage; use tinyagents::harness::middleware::{ - ContextCompressionMiddleware, MessageTrimMiddleware, PromptCacheGuardMiddleware, - ToolPolicyMiddleware as TaToolPolicyMiddleware, + BudgetLimits, BudgetMiddleware, ContextCompressionMiddleware, MessageTrimMiddleware, + PromptCacheGuardMiddleware, ToolPolicyMiddleware as TaToolPolicyMiddleware, }; use tinyagents::harness::model::CapabilitySet; use tinyagents::harness::runtime::{AgentHarness, RunPolicy, UnknownToolPolicy}; @@ -1338,10 +1339,51 @@ fn assemble_turn_harness( let tool_policies = harness.tools().policies(); context_mw.install(&mut harness, tool_policies); - // Pre-call cost budget gate (issue #4249, Phase 5): fail before a model call - // when OpenHuman's daily/monthly cost budget is already exceeded. Self-gating - // — a no-op unless cost budgets are configured. - harness.push_middleware(Arc::new(middleware::CostBudgetMiddleware)); + // Observe-only crate `BudgetMiddleware` (W2-budget-dedupe / workstream 06). + // Installed with empty `BudgetLimits` so it NEVER enforces or halts: its + // `before_model` preflight has no configured limit to trip, and its + // `after_model` only folds each call's usage into its shared `BudgetTracker`. + // It also re-emits `AgentEvent::UsageRecorded` per call (on top of the + // runtime's own emit); the event bridge dedupes those by model-call iteration + // so the global cost tracker still records each call exactly once (see + // `observability::OpenhumanEventBridge::record_usage`). Enforcement STAYS with + // the local `CostBudgetMiddleware` below (authoritative: reads the global + // daily/monthly `CostTracker`). + // + // FLIP CRITERIA — what must hold before the crate `BudgetMiddleware` becomes + // the enforcing owner and the local `CostBudgetMiddleware` + the + // `agent/harness/turn_subagent_usage.rs` task-local are DELETED (deletion + // ledger row: "crate-internal CostBudgetMiddleware + turn_subagent_usage.rs + // task-local", `docs/tinyagents-full-migration-plan/99-deletion-ledger.md`): + // 1. ≥ 500 production turns across BOTH parent and sub-agent runs with + // ZERO `[budget_shadow]` divergence log lines — proving the crate + // tracker's per-run token accounting matches the authoritative runtime + // `AgentRun.usage` on every model call. + // 2. A pricing table wired via `BudgetMiddleware::with_pricing(..)` at + // parity with `cost::catalog::estimate_cost_usd`, so the crate can own + // MONEY (USD) budgets. Today the shadow compares TOKENS only (the + // observe-only crate middleware has no pricing, so its cost stays $0) + // and the local gate is the sole money-budget authority. + // 3. Run-tree rollup wired: the same shared `BudgetTracker` handed to every + // sub-agent harness so a parent budget halts a recursive run pre-spend — + // replacing the `turn_subagent_usage` parent-turn rollup (06-cost step 3 + // / 07.2 TaskStore rollup). + // Until all three hold, this middleware is observe-only and the local gate + // enforces. + let shadow_budget = Arc::new(BudgetMiddleware::new(BudgetLimits::default())); + let shadow_budget_tracker = shadow_budget.tracker(); + harness.push_middleware(shadow_budget); + + // Pre-call cost budget gate (issue #4249, Phase 5) — AUTHORITATIVE + // enforcement: fail before a model call when OpenHuman's daily/monthly cost + // budget is already exceeded. Self-gating — a no-op unless cost budgets are + // configured. Demoted to a divergence-logging shadow owner (W2-budget-dedupe): + // it keeps enforcing exactly as before, but ALSO compares its per-run token + // accounting against the observe-only crate `BudgetMiddleware` above at end of + // run and logs `[budget_shadow]` parity/divergence. + harness.push_middleware(Arc::new(middleware::CostBudgetMiddleware::with_shadow( + shadow_budget_tracker, + ))); // Autocompaction parity: when the provider's context window is known, install // the two-stage context-management step (issue #4249). diff --git a/src/openhuman/tinyagents/observability.rs b/src/openhuman/tinyagents/observability.rs index cb5590831..0fd16c1ce 100644 --- a/src/openhuman/tinyagents/observability.rs +++ b/src/openhuman/tinyagents/observability.rs @@ -142,6 +142,15 @@ pub(crate) struct OpenhumanEventBridge { /// Shared `call_id → (success, failure)` side-channel written by /// `ToolOutcomeCaptureMiddleware`; read when projecting `ToolCallCompleted`. failure_map: ToolFailureMap, + /// Model-call iterations whose `UsageRecorded` has already been folded into + /// the global cost tracker (W2-budget-dedupe). A single model call can now + /// surface **two** `UsageRecorded` events — one from the harness runtime + /// (`agent_loop`, always) and one from the observe-only crate + /// `BudgetMiddleware::after_model` — both carrying identical usage and both + /// delivered to this bridge. Keyed on the run-scoped model-call identity (the + /// iteration cursor, bumped once per `ModelStarted`) so a given call's usage + /// is recorded exactly once. See [`OpenhumanEventBridge::record_usage`]. + recorded_iterations: Mutex>, state: Mutex, } @@ -183,6 +192,7 @@ impl OpenhumanEventBridge { cursor, tool_names, failure_map, + recorded_iterations: Mutex::new(std::collections::HashSet::new()), state: Mutex::new(BridgeState::default()), }) } @@ -231,6 +241,32 @@ impl OpenhumanEventBridge { /// `TurnCostUpdated` so the UI footer stays live. fn record_usage(&self, usage: &Usage) { let iteration = self.iteration(); + // Dedupe guard (W2-budget-dedupe): record a given model call's usage into + // the global cost tracker **exactly once**. Installing the observe-only + // crate `BudgetMiddleware` makes each model call emit two `UsageRecorded` + // events (the runtime's own at `agent_loop` + the middleware's + // `after_model` re-emit), both reaching this listener with identical + // usage. The two events have *distinct* stable ids, so an event-id key + // would not collapse them — instead we key on the run-scoped model-call + // identity: the iteration cursor, bumped once per `ModelStarted`. This + // bridge instance is per-run (parent or child scope), so the set is + // naturally (run, turn)-scoped. First writer for an iteration records; + // any later `UsageRecorded` for the same iteration is a duplicate. + { + let mut seen = self + .recorded_iterations + .lock() + .unwrap_or_else(|p| p.into_inner()); + if !seen.insert(iteration) { + tracing::debug!( + iteration, + model = %self.model, + child = self.scope.is_some(), + "[budget] duplicate UsageRecorded for model call — skipping double record" + ); + return; + } + } // Provider-reported charged USD has no home in the crate `Usage` (all // token counts), so estimate this call's cost from catalogued per-MTok // rates. Fixes the long-standing $0 cost on the tinyagents path, where @@ -683,6 +719,64 @@ mod tests { assert_eq!((input, output), (100, 40)); } + /// W2-budget-dedupe: two `UsageRecorded` events for the *same* model call + /// (as happens once the observe-only crate `BudgetMiddleware` re-emits usage + /// its `after_model` folded, on top of the runtime's own emit) must be + /// recorded into the bridge accounting **exactly once**. Without the dedupe + /// guard the totals would double. + #[tokio::test] + async fn duplicate_usage_for_same_model_call_is_recorded_once() { + let (tx, mut rx) = tokio::sync::mpsc::channel(64); + let bridge = OpenhumanEventBridge::new(Some(tx), "mock-model", 10); + let sink = EventSink::new(); + sink.subscribe(bridge.clone()); + + // One model call → one `ModelStarted` (iteration cursor → 1). + sink.emit(AgentEvent::ModelStarted { + call_id: "c1".into(), + model: "mock-model".to_string(), + }); + // Same call surfaces usage twice (runtime emit + BudgetMiddleware re-emit). + sink.emit(AgentEvent::UsageRecorded { + usage: Usage::new(100, 40), + }); + sink.emit(AgentEvent::UsageRecorded { + usage: Usage::new(100, 40), + }); + + // Totals reflect a single record, not two. + let (input, output, _) = bridge.totals(); + assert_eq!( + (input, output), + (100, 40), + "the duplicate UsageRecorded for the same iteration must be skipped" + ); + + // Exactly one `TurnCostUpdated` footer emit for the call. + let mut cost_updates = 0; + while let Ok(p) = rx.try_recv() { + if matches!(p, AgentProgress::TurnCostUpdated { .. }) { + cost_updates += 1; + } + } + assert_eq!(cost_updates, 1, "footer must update once per model call"); + + // A genuinely new model call (iteration cursor → 2) records again. + sink.emit(AgentEvent::ModelStarted { + call_id: "c2".into(), + model: "mock-model".to_string(), + }); + sink.emit(AgentEvent::UsageRecorded { + usage: Usage::new(10, 5), + }); + let (input, output, _) = bridge.totals(); + assert_eq!( + (input, output), + (110, 45), + "a distinct model call (new iteration) must still record" + ); + } + // NOTE: the former `sentinel_tool_started_is_not_forwarded` test was removed // here. The #4249 migration (commit 60097ba8d, "use sdk unknown tool // recovery") deleted `UNKNOWN_TOOL_SENTINEL` + `UnknownToolRewriteMiddleware` diff --git a/src/openhuman/tinyagents/replay/mod.rs b/src/openhuman/tinyagents/replay/mod.rs new file mode 100644 index 000000000..a3e11efb7 --- /dev/null +++ b/src/openhuman/tinyagents/replay/mod.rs @@ -0,0 +1,14 @@ +//! Read-only agent run replay + status RPC surface (workstream 05.x). +//! +//! Thin controllers over the C4 durable journal/status seams in +//! [`crate::openhuman::tinyagents::journal`]. Everything here is a reader: +//! no mutation, no writes, no security/approval/sandbox bypass. See +//! [`schemas`] for the three `agent`-namespace controllers and [`ops`] for the +//! workspace-parameterized read logic. + +pub(crate) mod ops; +mod schemas; + +pub(crate) use schemas::{ + all_agent_replay_controller_schemas, all_agent_replay_registered_controllers, +}; diff --git a/src/openhuman/tinyagents/replay/ops.rs b/src/openhuman/tinyagents/replay/ops.rs new file mode 100644 index 000000000..028848157 --- /dev/null +++ b/src/openhuman/tinyagents/replay/ops.rs @@ -0,0 +1,303 @@ +//! Read-only business logic for the agent replay/status RPC surface +//! (workstream 05.x). Every function here is a *reader* over the C4 durable +//! journal + status seams built in +//! [`crate::openhuman::tinyagents::journal`] — it opens the same +//! `{workspace}/tinyagents_store/{kv,journal}` stores and never writes, mutates, +//! or bypasses any security/approval/sandbox gate. +//! +//! The controller layer ([`super::schemas`]) resolves the configured workspace +//! and delegates here; these functions take an explicit `workspace` path so they +//! are unit-testable against a temp store (mirroring the `journal.rs` tests). + +use std::path::Path; + +use tinyagents::harness::events::HarnessRunStatus; +use tinyagents::harness::ids::ExecutionStatus; +use tinyagents::harness::observability::{ + AgentObservation, HarnessEventJournal, HarnessStatusStore, StoreEventJournal, +}; + +use crate::openhuman::session_import::ops::open_session_stores; +use crate::openhuman::tinyagents::journal::FileStatusStore; + +/// Default page size for [`read_run_events_page`] when the caller omits `limit`. +pub(crate) const DEFAULT_EVENTS_LIMIT: u64 = 200; + +/// Hard cap on a single replay page so one RPC can never fan a whole run into a +/// single response. +pub(crate) const MAX_EVENTS_LIMIT: u64 = 1000; + +/// One page of a run's durable event stream. +/// +/// `events` are [`AgentObservation`]s (the crate's durable observability +/// envelope — `event_id` / `run_id` / lineage / `offset` / `ts_ms` / typed +/// `event`) in ascending `offset` order. `next_offset` is the `offset` to pass +/// back to fetch the following page, or `None` once the stream is drained. +pub(crate) struct RunEventsPage { + pub events: Vec, + pub next_offset: Option, +} + +/// Paged late-attach replay reader over the durable journal. +/// +/// Returns up to `limit` observations for `run_id` whose stream offset is +/// `>= offset`, in order, plus a `next_offset` cursor (`None` when the last page +/// drained the stream). Backed by the C4 journal seam +/// ([`StoreEventJournal::read_from`], the same reader +/// [`crate::openhuman::tinyagents::journal::read_run_events`] uses). Best-effort: +/// a missing store or unknown run yields an empty page, not an error. +pub(crate) async fn read_run_events_page( + workspace: &Path, + run_id: &str, + offset: u64, + limit: u64, +) -> anyhow::Result { + // Guard the page size: clamp a zero/absurd limit into [1, MAX]. + let effective_limit = limit.clamp(1, MAX_EVENTS_LIMIT); + log::debug!( + "[agent] replay read_run_events_page run_id={run_id} offset={offset} \ + limit={limit} effective_limit={effective_limit}" + ); + + let stores = open_session_stores(workspace); + let journal = StoreEventJournal::new(stores.journal); + // Read one extra record to detect whether a further page exists without a + // second store round-trip. + let mut events = journal.read_from(run_id, offset).await.map_err(|e| { + anyhow::anyhow!("[agent] replay read_run_events_page failed run_id={run_id}: {e}") + })?; + + let has_more = events.len() as u64 > effective_limit; + if has_more { + events.truncate(effective_limit as usize); + } + // Offsets are monotonic within a run, so the cursor is simply "one past the + // last returned offset". `None` when this page drained the stream. + let next_offset = if has_more { + events.last().map(|obs| obs.offset + 1) + } else { + None + }; + + log::debug!( + "[agent] replay read_run_events_page run_id={run_id} returned={} next_offset={:?}", + events.len(), + next_offset + ); + Ok(RunEventsPage { + events, + next_offset, + }) +} + +/// Latest durable [`HarnessRunStatus`] for `run_id`, or `None` when the run is +/// unknown. Backed by the C4 status seam +/// ([`crate::openhuman::tinyagents::journal::read_run_status`] / +/// [`FileStatusStore::get_status`]). +pub(crate) async fn read_run_status( + workspace: &Path, + run_id: &str, +) -> anyhow::Result> { + log::debug!("[agent] replay read_run_status run_id={run_id}"); + let stores = open_session_stores(workspace); + let store = FileStatusStore::new(stores.kv); + let status = store.get_status(run_id).await.map_err(|e| { + anyhow::anyhow!("[agent] replay read_run_status failed run_id={run_id}: {e}") + })?; + log::debug!( + "[agent] replay read_run_status run_id={run_id} found={}", + status.is_some() + ); + Ok(status) +} + +/// Is a run still live (i.e. eligible for the "active" listing)? +/// +/// Mirrors the liveness predicate the crate's status store uses for +/// `list_active` (Pending / Running / Interrupted). +fn is_active(status: &HarnessRunStatus) -> bool { + matches!( + status.status, + ExecutionStatus::Pending | ExecutionStatus::Running | ExecutionStatus::Interrupted + ) +} + +/// Active runs, optionally filtered by `thread_id` and/or `root_run_id`. +/// +/// Backed by the C4 status seam: +/// - no filter → [`FileStatusStore::list_active`] +/// - `thread_id` → [`FileStatusStore::list_by_thread`] +/// - `root_run_id` → [`FileStatusStore::list_by_root`] +/// +/// The thread/root store queries return *all* runs (active and terminal), so the +/// active-liveness predicate is always applied on top — this controller only +/// ever surfaces live runs. When both filters are supplied, the base query uses +/// `thread_id` and the result is further restricted to `root_run_id`. +pub(crate) async fn list_active_runs( + workspace: &Path, + thread_id: Option<&str>, + root_run_id: Option<&str>, +) -> anyhow::Result> { + log::debug!( + "[agent] replay list_active_runs thread_id={:?} root_run_id={:?}", + thread_id, + root_run_id + ); + let stores = open_session_stores(workspace); + let store = FileStatusStore::new(stores.kv); + + let base = match (thread_id, root_run_id) { + (Some(thread), _) => store.list_by_thread(thread).await, + (None, Some(root)) => store.list_by_root(root).await, + (None, None) => store.list_active().await, + } + .map_err(|e| anyhow::anyhow!("[agent] replay list_active_runs failed: {e}"))?; + + let mut runs: Vec = base.into_iter().filter(is_active).collect(); + // If a caller supplied BOTH a thread and a root, the thread query drove the + // base list; narrow it to the requested root as well. + if thread_id.is_some() { + if let Some(root) = root_run_id { + runs.retain(|s| s.root_run_id.as_str() == root); + } + } + + log::debug!("[agent] replay list_active_runs returned={}", runs.len()); + Ok(runs) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + + use tinyagents::harness::events::{AgentEvent, EventSink}; + use tinyagents::harness::ids::{ComponentId, HarnessPhase, ThreadId}; + use tinyagents::harness::observability::{FanOutSink, JournalSink, StoreEventJournal}; + + use crate::openhuman::tinyagents::journal::mint_run_id; + + /// Seed `count` durable events for a fresh run under `workspace`, returning + /// the run id. Mirrors the seam wiring in the `journal.rs` tests: a run + /// [`EventSink`] seeded with the run id (so persisted `event_id`s are the + /// stable `{run_id}-evt-{offset}`) fanning out into a [`StoreEventJournal`]. + async fn seed_run_events(workspace: &Path, count: usize) -> String { + let stores = open_session_stores(workspace); + let run_id = mint_run_id(); + let journal: Arc = + Arc::new(StoreEventJournal::new(stores.journal)); + let sink = EventSink::with_stream_id(run_id.as_str()); + let journal_sink = JournalSink::new(journal, run_id.clone()); + sink.subscribe(Arc::new(FanOutSink::new().with(Arc::new(journal_sink)))); + for i in 0..count { + sink.emit(AgentEvent::ToolStarted { + call_id: format!("c{i}").into(), + tool_name: format!("tool-{i}"), + }); + } + run_id.as_str().to_string() + } + + /// Paging boundary: a page smaller than the stream reports a `next_offset` + /// cursor; the final page drains to `None` and never over-reads. + #[tokio::test] + async fn read_run_events_page_pages_and_drains() { + let tmp = std::env::temp_dir().join(format!("oh-replay-page-{}", uuid::Uuid::new_v4())); + let run_id = seed_run_events(&tmp, 3).await; + + // First page (limit 2) returns offsets 0,1 with a cursor at offset 2. + let page1 = read_run_events_page(&tmp, &run_id, 0, 2).await.unwrap(); + assert_eq!(page1.events.len(), 2); + assert_eq!(page1.events[0].offset, 0); + assert_eq!(page1.events[1].offset, 1); + assert_eq!(page1.next_offset, Some(2), "more events remain"); + + // Second page resumes at the cursor and drains → next_offset None. + let page2 = read_run_events_page(&tmp, &run_id, page1.next_offset.unwrap(), 2) + .await + .unwrap(); + assert_eq!(page2.events.len(), 1); + assert_eq!(page2.events[0].offset, 2); + assert_eq!(page2.next_offset, None, "stream drained on the last page"); + + // A page exactly the size of the remaining stream still drains to None + // (no phantom extra page). + let exact = read_run_events_page(&tmp, &run_id, 0, 3).await.unwrap(); + assert_eq!(exact.events.len(), 3); + assert_eq!(exact.next_offset, None); + + // Unknown run → empty page, not an error. + let empty = read_run_events_page(&tmp, "run.does-not-exist", 0, 10) + .await + .unwrap(); + assert!(empty.events.is_empty()); + assert_eq!(empty.next_offset, None); + + let _ = std::fs::remove_dir_all(&tmp); + } + + /// Status reader returns `None` for a run that was never recorded. + #[tokio::test] + async fn read_run_status_none_for_unknown_run() { + let tmp = std::env::temp_dir().join(format!("oh-replay-status-{}", uuid::Uuid::new_v4())); + let missing = read_run_status(&tmp, "run.nope").await.unwrap(); + assert!(missing.is_none()); + let _ = std::fs::remove_dir_all(&tmp); + } + + /// Active listing surfaces a started run and filters by thread; a completed + /// run is excluded. + #[tokio::test] + async fn list_active_runs_returns_started_and_filters_by_thread() { + let tmp = std::env::temp_dir().join(format!("oh-replay-active-{}", uuid::Uuid::new_v4())); + let store = FileStatusStore::new(open_session_stores(&tmp).kv); + + // A running run on thread-A. + let run_a = mint_run_id(); + let mut status_a = + HarnessRunStatus::new(run_a.clone(), ComponentId::new("mock-model".to_string())) + .with_thread(ThreadId::new("thread-A")); + status_a.mark_running(HarnessPhase::Model); + store.put_status(status_a).await.unwrap(); + + // A completed run on thread-B (must NOT appear in the active listing). + let run_b = mint_run_id(); + let mut status_b = + HarnessRunStatus::new(run_b.clone(), ComponentId::new("mock-model".to_string())) + .with_thread(ThreadId::new("thread-B")); + status_b.mark_running(HarnessPhase::Model); + status_b.mark_completed(); + store.put_status(status_b).await.unwrap(); + + // No filter: only the running run. + let active = list_active_runs(&tmp, None, None).await.unwrap(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].run_id.as_str(), run_a.as_str()); + + // Filter by thread-A: the running run is returned. + let by_thread_a = list_active_runs(&tmp, Some("thread-A"), None) + .await + .unwrap(); + assert_eq!(by_thread_a.len(), 1); + assert_eq!(by_thread_a[0].run_id.as_str(), run_a.as_str()); + + // Filter by thread-B: the only run there is completed → excluded. + let by_thread_b = list_active_runs(&tmp, Some("thread-B"), None) + .await + .unwrap(); + assert!(by_thread_b.is_empty()); + + // Filter by an unknown thread: empty. + let by_thread_none = list_active_runs(&tmp, Some("nope"), None).await.unwrap(); + assert!(by_thread_none.is_empty()); + + // Filter by root_run_id (a top-level run's root equals its own id). + let by_root = list_active_runs(&tmp, None, Some(run_a.as_str())) + .await + .unwrap(); + assert_eq!(by_root.len(), 1); + assert_eq!(by_root[0].run_id.as_str(), run_a.as_str()); + + let _ = std::fs::remove_dir_all(&tmp); + } +} diff --git a/src/openhuman/tinyagents/replay/schemas.rs b/src/openhuman/tinyagents/replay/schemas.rs new file mode 100644 index 000000000..c7c599844 --- /dev/null +++ b/src/openhuman/tinyagents/replay/schemas.rs @@ -0,0 +1,348 @@ +//! Read-only JSON-RPC controllers for agent run replay + status +//! (workstream 05.x), over the C4 durable journal/status seams. +//! +//! Three controllers, all in the `agent` namespace, all **read-only** (no +//! mutation, no security/approval/sandbox bypass, no writes): +//! +//! - `openhuman.agent_run_events` — paged late-attach replay of a run's durable +//! event stream. Params: `run_id`, `offset` (default 0), `limit` (default +//! [`DEFAULT_EVENTS_LIMIT`], capped at [`MAX_EVENTS_LIMIT`]). Returns +//! `{ events: [...], next_offset: }` — `null` +//! once the stream is drained. +//! - `openhuman.agent_run_status` — latest [`HarnessRunStatus`] for `run_id`, +//! or `null` for an unknown run. +//! - `openhuman.agent_runs_active` — active runs, optionally filtered by +//! `thread_id` and/or `root_run_id`. Returns `{ runs: [...] }`. +//! +//! ## Serialization / DTO note +//! +//! Events are surfaced as the crate's [`AgentObservation`] serde shape and +//! statuses as the [`HarnessRunStatus`] serde shape, projected **directly** — +//! no bespoke DTO. This is deliberate: both types are exactly what the C4 layer +//! already persists as JSON in `{workspace}/tinyagents_store` (an +//! `AgentObservation` is the durable journal record; a `HarnessRunStatus` is the +//! durable status snapshot), so a direct projection is guaranteed round-trip +//! stable and cannot drift from what a replay reconstructs. `AgentObservation` +//! carries `{ event_id, run_id, parent_run_id?, root_run_id, offset, ts_ms, +//! event }`, where `event` is the internally-tagged (`"kind"`) `AgentEvent` +//! enum; `HarnessRunStatus` carries ids/lineage, `status`, `current_phase`, +//! call counters, usage/cost totals, and timestamps — never prompt text, tool +//! arguments, or provider payloads. + +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; + +use tinyagents::harness::events::HarnessRunStatus; +use tinyagents::harness::observability::AgentObservation; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; + +use super::ops::{ + list_active_runs, read_run_events_page, read_run_status, DEFAULT_EVENTS_LIMIT, MAX_EVENTS_LIMIT, +}; + +const NAMESPACE: &str = "agent"; + +// --------------------------------------------------------------------------- +// Params +// --------------------------------------------------------------------------- + +#[derive(Debug, Deserialize)] +struct RunEventsParams { + run_id: String, + #[serde(default)] + offset: u64, + #[serde(default)] + limit: Option, +} + +#[derive(Debug, Deserialize)] +struct RunStatusParams { + run_id: String, +} + +#[derive(Debug, Deserialize, Default)] +struct RunsActiveParams { + #[serde(default)] + thread_id: Option, + #[serde(default)] + root_run_id: Option, +} + +// --------------------------------------------------------------------------- +// Responses (direct projection of the crate serde shapes — see module docs) +// --------------------------------------------------------------------------- + +#[derive(Debug, Serialize)] +struct RunEventsResponse { + events: Vec, + /// Cursor to fetch the next page, or `null` once the stream is drained. + next_offset: Option, +} + +#[derive(Debug, Serialize)] +struct RunsActiveResponse { + runs: Vec, +} + +// --------------------------------------------------------------------------- +// Registration +// --------------------------------------------------------------------------- + +/// All read-only replay/status controller schemas (workstream 05.x). +pub(crate) fn all_agent_replay_controller_schemas() -> Vec { + vec![ + replay_schema("run_events"), + replay_schema("run_status"), + replay_schema("runs_active"), + ] +} + +/// All read-only replay/status registered controllers (workstream 05.x). +pub(crate) fn all_agent_replay_registered_controllers() -> Vec { + vec![ + RegisteredController { + schema: replay_schema("run_events"), + handler: handle_run_events, + }, + RegisteredController { + schema: replay_schema("run_status"), + handler: handle_run_status, + }, + RegisteredController { + schema: replay_schema("runs_active"), + handler: handle_runs_active, + }, + ] +} + +fn replay_schema(function: &str) -> ControllerSchema { + match function { + "run_events" => ControllerSchema { + namespace: NAMESPACE, + function: "run_events", + description: + "Read-only paged replay of a durable agent run's event stream (late-attach \ + reconnect/backfill). Returns AgentObservations at offset >= `offset` in order, \ + plus `next_offset` (null once drained). Never mutates state.", + inputs: vec![ + FieldSchema { + name: "run_id", + ty: TypeSchema::String, + comment: "Durable run id (as minted by the journal, e.g. `run.`).", + required: true, + }, + FieldSchema { + name: "offset", + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment: "Start stream offset (inclusive). Default 0 replays the whole run.", + required: false, + }, + FieldSchema { + name: "limit", + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment: "Max events in this page. Defaults to 200, capped at 1000.", + required: false, + }, + ], + outputs: vec![ + FieldSchema { + name: "events", + ty: TypeSchema::Array(Box::new(TypeSchema::Json)), + comment: "Ordered AgentObservations (event_id, run_id, lineage, offset, \ + ts_ms, typed `event`).", + required: true, + }, + FieldSchema { + name: "next_offset", + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment: "Cursor for the next page, or null when the stream is drained.", + required: false, + }, + ], + }, + "run_status" => ControllerSchema { + namespace: NAMESPACE, + function: "run_status", + description: + "Read-only latest durable status snapshot (HarnessRunStatus) for a run, or null \ + when the run is unknown. Counters/phase/usage/cost only — never prompts or \ + payloads.", + inputs: vec![FieldSchema { + name: "run_id", + ty: TypeSchema::String, + comment: "Durable run id to look up.", + required: true, + }], + outputs: vec![FieldSchema { + name: "status", + ty: TypeSchema::Option(Box::new(TypeSchema::Json)), + comment: "The HarnessRunStatus snapshot, or null for an unknown run.", + required: false, + }], + }, + "runs_active" => ControllerSchema { + namespace: NAMESPACE, + function: "runs_active", + description: + "Read-only listing of active (pending/running/interrupted) agent runs, optionally \ + filtered by `thread_id` and/or `root_run_id`. Never mutates state.", + inputs: vec![ + FieldSchema { + name: "thread_id", + ty: TypeSchema::Option(Box::new(TypeSchema::String)), + comment: "Restrict to runs on this conversation thread.", + required: false, + }, + FieldSchema { + name: "root_run_id", + ty: TypeSchema::Option(Box::new(TypeSchema::String)), + comment: "Restrict to descendants of this root run.", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "runs", + ty: TypeSchema::Array(Box::new(TypeSchema::Json)), + comment: "Active HarnessRunStatus snapshots matching the filter.", + required: true, + }], + }, + _ => ControllerSchema { + namespace: NAMESPACE, + function: "unknown", + description: "Unknown agent replay controller.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "error", + ty: TypeSchema::String, + comment: "Lookup error details.", + required: true, + }], + }, + } +} + +// --------------------------------------------------------------------------- +// Handlers +// --------------------------------------------------------------------------- + +/// Resolve the configured internal workspace whose `tinyagents_store/` holds the +/// journal + status stores. +async fn configured_workspace() -> Result { + let config = crate::openhuman::config::Config::load_or_init() + .await + .map_err(|e| format!("failed to load config: {e}"))?; + Ok(config.workspace_dir) +} + +fn handle_run_events(params: Map) -> ControllerFuture { + Box::pin(async move { + let payload: RunEventsParams = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("invalid params: {e}"))?; + let limit = payload + .limit + .unwrap_or(DEFAULT_EVENTS_LIMIT) + .min(MAX_EVENTS_LIMIT); + log::debug!( + "[rpc] openhuman.agent_run_events run_id={} offset={} limit={}", + payload.run_id, + payload.offset, + limit + ); + + let workspace = configured_workspace().await?; + let page = read_run_events_page(&workspace, &payload.run_id, payload.offset, limit) + .await + .map_err(|e| format!("read run events failed: {e:#}"))?; + log::debug!( + "[rpc] openhuman.agent_run_events run_id={} returned={} next_offset={:?}", + payload.run_id, + page.events.len(), + page.next_offset + ); + + let response = RunEventsResponse { + events: page.events, + next_offset: page.next_offset, + }; + serde_json::to_value(response).map_err(|e| format!("serialize response failed: {e}")) + }) +} + +fn handle_run_status(params: Map) -> ControllerFuture { + Box::pin(async move { + let payload: RunStatusParams = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("invalid params: {e}"))?; + log::debug!("[rpc] openhuman.agent_run_status run_id={}", payload.run_id); + + let workspace = configured_workspace().await?; + let status = read_run_status(&workspace, &payload.run_id) + .await + .map_err(|e| format!("read run status failed: {e:#}"))?; + log::debug!( + "[rpc] openhuman.agent_run_status run_id={} found={}", + payload.run_id, + status.is_some() + ); + + // Bare projection: object for a known run, `null` for an unknown one. + serde_json::to_value(status).map_err(|e| format!("serialize response failed: {e}")) + }) +} + +fn handle_runs_active(params: Map) -> ControllerFuture { + Box::pin(async move { + let payload: RunsActiveParams = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("invalid params: {e}"))?; + log::debug!( + "[rpc] openhuman.agent_runs_active thread_id={:?} root_run_id={:?}", + payload.thread_id, + payload.root_run_id + ); + + let workspace = configured_workspace().await?; + let runs = list_active_runs( + &workspace, + payload.thread_id.as_deref(), + payload.root_run_id.as_deref(), + ) + .await + .map_err(|e| format!("list active runs failed: {e:#}"))?; + log::debug!("[rpc] openhuman.agent_runs_active returned={}", runs.len()); + + serde_json::to_value(RunsActiveResponse { runs }) + .map_err(|e| format!("serialize response failed: {e}")) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn controller_inventory_is_stable() { + let schemas = all_agent_replay_controller_schemas(); + assert_eq!(schemas.len(), 3); + assert!(schemas.iter().all(|s| s.namespace == "agent")); + let functions: Vec<&str> = schemas.iter().map(|s| s.function).collect(); + assert!(functions.contains(&"run_events")); + assert!(functions.contains(&"run_status")); + assert!(functions.contains(&"runs_active")); + + let controllers = all_agent_replay_registered_controllers(); + assert_eq!(controllers.len(), 3); + // rpc method names follow openhuman._. + let methods: Vec = controllers.iter().map(|c| c.rpc_method_name()).collect(); + assert!(methods.contains(&"openhuman.agent_run_events".to_string())); + assert!(methods.contains(&"openhuman.agent_run_status".to_string())); + assert!(methods.contains(&"openhuman.agent_runs_active".to_string())); + } + + #[tokio::test] + async fn run_events_rejects_missing_run_id() { + let err = handle_run_events(Map::new()).await.unwrap_err(); + assert!(err.contains("invalid params"), "{err}"); + } +} diff --git a/src/openhuman/tinyagents/tests.rs b/src/openhuman/tinyagents/tests.rs index 376060455..6d7d458ce 100644 --- a/src/openhuman/tinyagents/tests.rs +++ b/src/openhuman/tinyagents/tests.rs @@ -508,11 +508,12 @@ fn adapter_inventory_registers_model_tools_and_middleware() { // Lifecycle middleware, in registration order: memory-protocol enforcement // (outermost), repeated-tool-failure breaker, shadow tool-exposure, // prompt-cache segment + guard, cache-align + tool-output - // (TurnContextMiddleware::defaults), cost budget, context compression + - // message trim (window known + autocompact on), SDK tool-policy projection, - // tool-outcome capture, arg recovery. + // (TurnContextMiddleware::defaults), observe-only crate BudgetMiddleware + // (W2-budget-dedupe), cost budget (local enforcement + budget_shadow), + // context compression + message trim (window known + autocompact on), SDK + // tool-policy projection, tool-outcome capture, arg recovery. let mw = assembled.harness.middleware(); - assert_eq!(mw.len(), 13, "lifecycle middleware inventory"); + assert_eq!(mw.len(), 14, "lifecycle middleware inventory"); // Around-tool wraps: approval/security + CLI/RPC-only scope gate (no // builder tool policy on this call). assert_eq!(mw.tool_middleware_len(), 2, "tool middleware inventory"); diff --git a/vendor/tinyagents b/vendor/tinyagents index a9500184b..ac7338241 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit a9500184b3d6e87e43019e757d4ca622a418b9d9 +Subproject commit ac7338241d50f2ecbfe52e518626b31d52c16ce6