mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(agent): TinyAgents migration wave 2 — microcompact upstream, session shadow reads, budget dedupe, replay RPC (#4483)
This commit is contained in:
Steven Enamakel
@@ -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)
|
||||
|
||||
@@ -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:<branch>`; explicit
|
||||
`git add <paths>` 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:<branch>`; explicit `git add <paths>`; 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.
|
||||
|
||||
@@ -131,6 +131,10 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
|
||||
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<ControllerSchema> {
|
||||
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());
|
||||
|
||||
@@ -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.
|
||||
// ─────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -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(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<usize>,
|
||||
},
|
||||
/// 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<Vec<JournalMessage>> {
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<usize> = 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<BudgetTracker>,
|
||||
}
|
||||
|
||||
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 {
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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<std::collections::HashSet<u32>>,
|
||||
state: Mutex<BridgeState>,
|
||||
}
|
||||
|
||||
@@ -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`
|
||||
|
||||
@@ -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,
|
||||
};
|
||||
@@ -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<AgentObservation>,
|
||||
pub next_offset: Option<u64>,
|
||||
}
|
||||
|
||||
/// 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<RunEventsPage> {
|
||||
// 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<Option<HarnessRunStatus>> {
|
||||
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<Vec<HarnessRunStatus>> {
|
||||
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<HarnessRunStatus> = 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<dyn HarnessEventJournal> =
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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: [<AgentObservation>...], next_offset: <int|null> }` — `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: [<HarnessRunStatus>...] }`.
|
||||
//!
|
||||
//! ## 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<u64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct RunStatusParams {
|
||||
run_id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, Default)]
|
||||
struct RunsActiveParams {
|
||||
#[serde(default)]
|
||||
thread_id: Option<String>,
|
||||
#[serde(default)]
|
||||
root_run_id: Option<String>,
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Responses (direct projection of the crate serde shapes — see module docs)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
struct RunEventsResponse {
|
||||
events: Vec<AgentObservation>,
|
||||
/// Cursor to fetch the next page, or `null` once the stream is drained.
|
||||
next_offset: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
struct RunsActiveResponse {
|
||||
runs: Vec<HarnessRunStatus>,
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Registration
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// All read-only replay/status controller schemas (workstream 05.x).
|
||||
pub(crate) fn all_agent_replay_controller_schemas() -> Vec<ControllerSchema> {
|
||||
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<RegisteredController> {
|
||||
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.<hex>`).",
|
||||
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<std::path::PathBuf, String> {
|
||||
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<String, Value>) -> 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<String, Value>) -> 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<String, Value>) -> 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.<namespace>_<function>.
|
||||
let methods: Vec<String> = 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}");
|
||||
}
|
||||
}
|
||||
@@ -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");
|
||||
|
||||
Vendored
+1
-1
Submodule vendor/tinyagents updated: a9500184b3...ac7338241d
Reference in New Issue
Block a user