From 6c327b2d96c8ff2b11b2ba575069020fe44ef4e9 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 23 Aug 2026 19:24:08 +0300 Subject: [PATCH 1/9] refactor(transcript): move TranscriptEntry out of the store feature MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `TranscriptEntry` is engine surface, not storage. The next commit hangs one on `AgentRunOutcome` and on `ExecutionStep` — both compiled in every build — so the type cannot sit behind `store`, a feature the engine never depends on and most hosts never enable. A pure move plus a re-export: `store::types::TranscriptEntry` still resolves, so `RunStep.transcript` and every existing path are untouched. The module is serde-only with no dependencies, so this costs the kernel profile nothing. Its docs are rewritten on the way, because the division of labour they describe is now visible in the type system rather than only in prose: the engine carries entries, and the host — the only thing with an event stream, and the only thing that knows what counts as one entry — folds them. Co-authored-by: Medulla --- src/lib.rs | 1 + src/store/types/mod.rs | 10 +++++-- src/{store/types => }/transcript.rs | 45 +++++++++++++++++++---------- 3 files changed, 38 insertions(+), 18 deletions(-) rename src/{store/types => }/transcript.rs (53%) diff --git a/src/lib.rs b/src/lib.rs index 2a37dc76..c5e8cb4a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -69,6 +69,7 @@ pub mod store; /// surface over all of it. Behind the `testkit` feature. #[cfg(any(test, feature = "testkit"))] pub mod testkit; +pub mod transcript; pub mod validate; // Resolving an author-supplied working directory against the run's workspace — // the containment rule a shell step's `args.cwd` already obeyed, shared with the diff --git a/src/store/types/mod.rs b/src/store/types/mod.rs index 47c53ea4..baacffc4 100644 --- a/src/store/types/mod.rs +++ b/src/store/types/mod.rs @@ -19,7 +19,8 @@ //! - [`note`] — what the host has learned about a workflow across runs. //! - [`proposal`] — a graph change suggested but not yet made. //! - [`error`] — the failure vocabulary every surface reports through. -//! - [`transcript`] — one line of what an agent did inside a step. +//! - [`TranscriptEntry`] — one line of what an agent did inside a step, +//! re-exported from [`crate::transcript`] because the engine carries it too. //! //! Why a failed run failed, in terms an author can act on, is //! [`crate::diagnostics`] — reading a run's steps is a pure function of the @@ -31,7 +32,6 @@ mod error; mod note; mod proposal; mod run; -mod transcript; mod workflow; #[cfg(test)] @@ -49,7 +49,11 @@ pub use run::{ LEGACY_TRUNCATED_KEY, RunId, RunOrigin, RunRecord, RunStatus, RunStep, TRUNCATED_KEY, bounded_evidence, bounded_within, is_truncated, }; -pub use transcript::TranscriptEntry; +// Re-exported, not owned: `TranscriptEntry` is engine surface (it rides an +// `ExecutionStep` and an `AgentRunOutcome`), so it lives at `crate::transcript` +// and cannot sit behind the `store` feature. Kept here so every path that has +// always read `store::types::TranscriptEntry` still resolves. +pub use crate::transcript::TranscriptEntry; pub use workflow::{ WorkflowDefaults, WorkflowId, WorkflowRecord, WorkflowRevision, WorkflowSummary, record_fingerprint, diff --git a/src/store/types/transcript.rs b/src/transcript.rs similarity index 53% rename from src/store/types/transcript.rs rename to src/transcript.rs index 043b766e..045788b0 100644 --- a/src/store/types/transcript.rs +++ b/src/transcript.rs @@ -1,11 +1,24 @@ //! One line of what an agent did, as a run record keeps it. //! -//! Part of the stored model rather than of the engine: nothing in -//! [`crate::engine`] produces these, and nothing in it reads them. A host that -//! runs `agent` nodes against something with an event stream folds that stream -//! into these entries and hangs them off a [`RunStep`](super::RunStep), so a run -//! read back tomorrow still says what happened inside a step and not only -//! whether it passed. +//! An `agent` node runs a host's harness, and a harness has an event stream: +//! it thinks, calls a tool, reads the result, thinks again. The node's output +//! says what came *out* of that; a transcript says what happened *inside* it, +//! so a run read back tomorrow explains itself rather than only passing or +//! failing. +//! +//! Two surfaces carry these, and both are the engine's: +//! [`AgentRunOutcome::transcript`](crate::caps::AgentRunOutcome::transcript), +//! where a host hands them over, and +//! [`ExecutionStep::transcript`](crate::observability::ExecutionStep::transcript), +//! where the engine hands them back to a [`RunObserver`](crate::observability::RunObserver). +//! A host that persists runs also finds them on `store::types::RunStep`. +//! +//! **Nothing in this crate folds a host's event stream into these** — the +//! engine has no event stream of its own, and what counts as one entry is a +//! judgement only the harness can make. Hosts fold; the crate carries. +//! [`RunObserver::on_agent_event`](crate::observability::RunObserver::on_agent_event) +//! is how a host reports one *while* the node is still running, rather than +//! waiting for the step to settle. //! //! Deliberately flat and stringly-typed. Mirroring a host's own event //! vocabulary into the record would make every event kind it adds later a @@ -17,13 +30,16 @@ use serde::{Deserialize, Serialize}; /// Bytes of one entry's `text` kept on the durable record. /// -/// [`RunRecord`](super::RunRecord) bounds step `input`, `output`, and its own -/// `inputs` through `bounded_within` so no single value can grow a run record -/// without limit; a transcript entry is the same kind of host-produced text -/// (a tool result, a model message) and needs the same ceiling. Small on -/// purpose — a transcript is many short lines, not one large payload, and a -/// step with hundreds of entries must not turn one long one into the whole -/// record's size budget. +/// A stored run bounds step `input`, `output` and its own `inputs` so no +/// single value can grow a run record without limit; a transcript entry is +/// the same kind of host-produced text (a tool result, a model message) and +/// needs the same ceiling. Small on purpose — a transcript is many short +/// lines, not one large payload, and a step with hundreds of entries must +/// not let one long entry become the whole record's size budget. +/// +/// A host with a genuinely large payload — a full tool result, a reasoning +/// body — keeps it in its own store and leaves the entry as the index line +/// pointing at it. That split is why this ceiling can stay small. pub const MAX_ENTRY_TEXT_BYTES: usize = 4 * 1024; /// One thing an agent did, in the order it did it. @@ -50,8 +66,7 @@ impl TranscriptEntry { /// Nothing in this crate folds a host's event stream into these — that /// happens entirely on the host side, as the module doc says — so this is /// the bound a host's folding code is expected to apply per entry, the way - /// [`bounded_within`](super::bounded_within) bounds the record's other - /// host-produced text. + /// a stored run bounds the record's other host-produced text. #[must_use] pub fn bounded(at_ms: i64, kind: impl Into, text: impl Into) -> Self { let mut text = text.into(); From aaed00eb125b082a88f864234b2d909e102fbf3f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 23 Aug 2026 19:25:23 +0300 Subject: [PATCH 2/9] feat(agent): carry a harness transcript from the outcome to the observer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An `agent` node runs a host's harness, and a harness has an event stream: it thinks, calls a tool, reads the result, thinks again. The node's output says what came *out* of that. Nothing said what happened *inside* it, so a run could be read as pass/fail and nothing more — and a host that had captured a transcript had no channel to hand one over. Four small additions, one path: AgentRunner::run -> AgentRunOutcome.transcript -> NodeOutput.transcript -> ExecutionStep.transcript (RunObserver::on_step_finish) -> RunObserver::on_agent_event (live, per entry) `on_agent_event` is not redundant with the step. A step's transcript arrives when the node *finishes*, and an agent node can run for minutes; a host that wants to show what an agent is doing while it does it needs the entries as they happen. Both paths carry the same entries on purpose, so neither is the only source: one drives a live view, the other survives to a run read back tomorrow. Two details worth reviewing: - A `per_item` agent node runs one turn per input item and reports ONE step, so entries accumulate in a shared sink rather than being returned per turn — otherwise every turn but the last would be dropped. - A tool call closing an open thinking run emits that run's finalized reasoning *before* the tool's own step. Dropping the tail would lose the reasoning immediately preceding a tool call, which is the part worth reading. Non-breaking by construction, and pinned by a test: a host that overrides neither hook behaves exactly as it did. `MockAgentRunner` implements only the legacy `run_agent`, so the default `run` wraps its return in a `finished` outcome with an empty transcript — `a_legacy_host_reports_no_transcript` asserts that path still works and simply has nothing to say. `ExecutionStep` gains a field, so its literals are updated rather than the struct being made `#[non_exhaustive]` — hosts construct these in their own tests, and closing the struct would take that away to save touching nine lines. Co-authored-by: Medulla --- src/caps/agent.rs | 46 +++++ src/caps/mock.rs | 27 +++ src/diagnostics_tests.rs | 2 + src/engine/build/outcome.rs | 6 + src/nodes/execution.rs | 21 ++ src/nodes/integration/agent.rs | 79 +++++++- .../agent_tests/agent_tests_part_02_tests.rs | 180 ++++++++++++++++++ src/observability.rs | 33 ++++ src/observability_tests.rs | 104 ++++++++++ 9 files changed, 491 insertions(+), 7 deletions(-) diff --git a/src/caps/agent.rs b/src/caps/agent.rs index 377c04de..c3794a95 100644 --- a/src/caps/agent.rs +++ b/src/caps/agent.rs @@ -42,6 +42,7 @@ use serde_json::{Map, Value}; use crate::error::Result; use crate::model::{AgentDefinition, ToolGrant}; +use crate::transcript::TranscriptEntry; /// Which model, on which provider, an agent run should use. /// @@ -381,6 +382,27 @@ pub struct AgentRunOutcome { /// Optional usage figures. #[serde(default, skip_serializing_if = "Option::is_none")] pub usage: Option, + /// What the harness did on the way to this outcome, in order. + /// + /// The node's payload above says what came *out* of the agent; this says + /// what happened *inside* it — the thinking, the tool calls, the results. + /// The engine copies it onto the step's + /// [`ExecutionStep::transcript`](crate::observability::ExecutionStep::transcript), + /// which is how it reaches a + /// [`RunObserver`](crate::observability::RunObserver) — so a host that + /// fills this in gets a run history that explains itself instead of one + /// that only reports pass or fail. + /// + /// Empty by default, and empty is a normal outcome: a harness with no + /// event stream to fold has nothing to say here, and every host that + /// predates this field keeps compiling and behaving identically. Bound each + /// entry with [`TranscriptEntry::bounded`] — the engine does not truncate. + /// + /// A host reporting entries *while* the node still runs uses + /// [`RunObserver::on_agent_event`](crate::observability::RunObserver::on_agent_event); + /// this field is the settled set. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub transcript: Vec, } impl AgentRunOutcome { @@ -419,9 +441,33 @@ impl AgentRunOutcome { json, raw: value, usage: None, + transcript: Vec::new(), } } + /// The same outcome, carrying what the harness did to reach it. + /// + /// The builder half of [`transcript`](Self::transcript), so a host can fold + /// its event stream once and attach it without naming every other field: + /// + /// ``` + /// use tinyflows::caps::AgentRunOutcome; + /// use tinyflows::transcript::TranscriptEntry; + /// use serde_json::json; + /// + /// let outcome = AgentRunOutcome::finished(json!("837799")) + /// .with_transcript(vec![ + /// TranscriptEntry::bounded(1, "agent_thinking", "Collatz — memoise."), + /// TranscriptEntry::bounded(2, "tool_call", "shell: python3 solve.py"), + /// ]); + /// assert_eq!(outcome.transcript.len(), 2); + /// ``` + #[must_use] + pub fn with_transcript(mut self, transcript: Vec) -> Self { + self.transcript = transcript; + self + } + /// Whether the agent reached a final answer. /// /// Anything else means the payload is partial or absent — see diff --git a/src/caps/mock.rs b/src/caps/mock.rs index 640518cf..4fb2378c 100644 --- a/src/caps/mock.rs +++ b/src/caps/mock.rs @@ -124,6 +124,21 @@ impl AgentRunner for MockAgentHarness { steps: Some(1), ..Default::default() }), + // Non-empty on purpose: a host testing against this mock should see + // a transcript reach its observer, not an empty vec that passes for + // the same thing. + transcript: vec![ + crate::transcript::TranscriptEntry::bounded( + 0, + "agent_thinking", + format!("deciding how to answer as {}", request.agent.id), + ), + crate::transcript::TranscriptEntry::bounded( + 0, + "agent_message", + format!("ran {}", request.agent.id), + ), + ], }) } @@ -219,6 +234,11 @@ impl AgentRunner for MockLimitedAgentRunner { json: partial.clone(), raw: partial, usage: None, + transcript: vec![crate::transcript::TranscriptEntry::bounded( + 0, + "agent_thinking", + "ran out of steps mid-thought", + )], }) } } @@ -254,6 +274,13 @@ impl AgentRunner for MockPausingAgentRunner { json: Value::Null, raw: Value::Null, usage: None, + // A pause still explains itself: this is the run whose transcript is + // most worth reading, because its output never arrives. + transcript: vec![crate::transcript::TranscriptEntry::bounded( + 0, + "tool_call", + "github.add_labels (awaiting approval)", + )], }) } } diff --git a/src/diagnostics_tests.rs b/src/diagnostics_tests.rs index 04876a91..d7c5578e 100644 --- a/src/diagnostics_tests.rs +++ b/src/diagnostics_tests.rs @@ -27,6 +27,7 @@ fn step(node_id: &str, nulls: &[(&str, &str)]) -> ExecutionStep { expression: expression.to_string(), }) .collect(), + transcript: vec![], } } @@ -39,6 +40,7 @@ fn failed(node_id: &str, output: serde_json::Value) -> ExecutionStep { // The point of this case: an error hidden by an `on_error` policy // carries no diagnostics at all. diagnostics: Vec::new(), + transcript: vec![], } } diff --git a/src/engine/build/outcome.rs b/src/engine/build/outcome.rs index 065d4a47..cbf019ce 100644 --- a/src/engine/build/outcome.rs +++ b/src/engine/build/outcome.rs @@ -36,6 +36,9 @@ where output: serde_json::to_value(&output.items).unwrap_or(Value::Null), duration_ms, diagnostics: output.diagnostics.clone(), + // Carried, not interpreted: an `agent` node's executor put the + // host's folded transcript here, every other node left it empty. + transcript: output.transcript.clone(), }; steps .lock() @@ -190,6 +193,9 @@ where output: Value::Null, duration_ms, diagnostics, + // A node that failed produced no `NodeOutput`, so there is no + // outcome to read a transcript from. + transcript: Vec::new(), }; steps .lock() diff --git a/src/nodes/execution.rs b/src/nodes/execution.rs index f5155933..2146fde4 100644 --- a/src/nodes/execution.rs +++ b/src/nodes/execution.rs @@ -158,6 +158,18 @@ pub struct NodeOutput { /// is why, before this existed, a `sub_workflow` whose child paused at an /// approval gate had to fail the parent outright rather than pause it. pub control: Option, + /// What a harness did inside this node, in order. + /// + /// Only an `agent` node fills this in, and only when the host's + /// [`AgentRunner`](crate::caps::AgentRunner) reported one on its + /// [`AgentRunOutcome`](crate::caps::AgentRunOutcome). The engine copies it + /// onto the node's + /// [`ExecutionStep`](crate::observability::ExecutionStep) and otherwise + /// does not read it — it is carried, never interpreted. + /// + /// Empty for every other node kind, and for a harness with no event stream + /// to fold. + pub transcript: Vec, } /// What a node asks the engine to do instead of simply emitting its items. @@ -237,6 +249,15 @@ impl NodeOutput { Self::default() } + /// The same output, carrying what a harness did to produce it. + /// + /// See [`transcript`](Self::transcript). Only the `agent` node uses this. + #[must_use] + pub fn with_transcript(mut self, transcript: Vec) -> Self { + self.transcript = transcript; + self + } + /// Attaches data-binding diagnostics (null-resolved expressions) to this /// output. #[must_use] diff --git a/src/nodes/integration/agent.rs b/src/nodes/integration/agent.rs index 33386d12..22092991 100644 --- a/src/nodes/integration/agent.rs +++ b/src/nodes/integration/agent.rs @@ -1,5 +1,7 @@ //! The `agent` node: an LLM agent turn with optional sub-ports. +use std::sync::{Arc, Mutex}; + use async_trait::async_trait; use serde_json::Value; @@ -7,6 +9,47 @@ use crate::data::Item; use crate::error::Result; use crate::nodes::integration::schema; use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; +use crate::transcript::TranscriptEntry; + +/// Collects every turn's transcript across one node activation. +/// +/// A `per_item` agent node runs one turn per input item while the step it +/// reports carries a single transcript, so the entries have to accumulate +/// somewhere shared. Shared rather than returned so `map_items`' closure +/// signature stays as it is. +type TranscriptSink = Arc>>; + +/// Reports `entries` live, then keeps them for the settled step. +/// +/// Both halves matter and neither replaces the other: `on_agent_event` is +/// what lets a console show an agent working through a node that runs for +/// minutes, and the accumulated copy is what a run read back tomorrow has. +fn record_transcript(ctx: &NodeContext<'_>, sink: &TranscriptSink, entries: Vec) { + if entries.is_empty() { + return; + } + for entry in &entries { + ctx.observer.on_agent_event(&ctx.node.id, entry); + } + // A poisoned lock here must not fail the turn: the agent already ran and + // its answer is good. Losing the transcript degrades the run history, + // which is strictly better than discarding the work. + match sink.lock() { + Ok(mut held) => held.extend(entries), + Err(err) => tracing::warn!( + node = %ctx.node.id, + %err, + "agent node: transcript sink poisoned; dropping this turn's entries" + ), + } +} + +/// Takes everything the sink holds, leaving it empty. +fn drain(sink: &TranscriptSink) -> Vec { + sink.lock() + .map(|mut held| std::mem::take(&mut *held)) + .unwrap_or_default() +} /// Runs an LLM agent turn, optionally composed with **sub-ports** that attach an /// output parser and tools to the bare completion. @@ -80,12 +123,18 @@ impl NodeExecutor for AgentNode { == crate::nodes::ExecutionMode::PerItem && !ctx.input.is_empty(); + // One accumulator for the whole node activation, because a `per_item` + // agent node runs many turns and the step carries one transcript. Shared + // rather than returned, so `map_items`' closure signature is untouched. + let transcript: TranscriptSink = Arc::new(Mutex::new(Vec::new())); + if per_item { // Fan out: `config.concurrency` decides how many turns run at once // (default 1 — sequential, as this node has always behaved), and // `config.on_item_error` what a failing turn does to the batch. let opts = crate::nodes::map::map_options(&ctx.node.config, &ctx.node.id, ctx.run); let ctx = &ctx; + let transcript = &transcript; let (items, diagnostics) = crate::nodes::map::map_items( ctx.input.len(), &ctx.node.id, @@ -98,19 +147,23 @@ impl NodeExecutor for AgentNode { // The same scope the config was resolved against, so an // agent definition's own `=`-expressions see this item. let scope = crate::nodes::expr_scope_for(ctx, item_json); - let item = run_turn_indexed(ctx, &cfg, &scope, Some(index)).await?; + let item = run_turn_indexed(ctx, &cfg, &scope, Some(index), transcript).await?; Ok((item, diags)) }, ) .await?; - return Ok(NodeOutput::main(items).with_diagnostics(diagnostics)); + return Ok(NodeOutput::main(items) + .with_diagnostics(diagnostics) + .with_transcript(drain(transcript))); } // Single turn against the first-item scope (or empty input). let (cfg, diagnostics) = crate::nodes::resolve_config_traced(&ctx); let scope = crate::nodes::expr_scope(&ctx); - let item = run_turn(&ctx, &cfg, &scope).await?; - Ok(NodeOutput::main(vec![item]).with_diagnostics(diagnostics)) + let item = run_turn(&ctx, &cfg, &scope, &transcript).await?; + Ok(NodeOutput::main(vec![item]) + .with_diagnostics(diagnostics) + .with_transcript(drain(&transcript))) } } @@ -118,8 +171,13 @@ impl NodeExecutor for AgentNode { /// registered agent kind), the optional tool sub-port, the optional /// output-parser sub-port, and finally the stable `{ json, text, raw }` /// envelope. Returns the emitted [`Item`] (without pairing — the caller sets it). -async fn run_turn(ctx: &NodeContext<'_>, cfg: &Value, scope: &Value) -> Result { - run_turn_indexed(ctx, cfg, scope, None).await +async fn run_turn( + ctx: &NodeContext<'_>, + cfg: &Value, + scope: &Value, + transcript: &TranscriptSink, +) -> Result { + run_turn_indexed(ctx, cfg, scope, None, transcript).await } /// [`run_turn`], told which input item it is running for under `per_item` @@ -129,6 +187,7 @@ async fn run_turn_indexed( cfg: &Value, scope: &Value, item_index: Option, + transcript: &TranscriptSink, ) -> Result { let conn = cfg.get("connection_ref").and_then(Value::as_str); @@ -146,7 +205,7 @@ async fn run_turn_indexed( let request = super::agent_request::assemble(ctx, cfg, agent_ref, scope, item_index).await?; let outcome = runner.run(request).await?; - return finish_agent_run(ctx, cfg, conn, agent_ref, outcome).await; + return finish_agent_run(ctx, cfg, conn, agent_ref, outcome, transcript).await; } // Degraded path: no agent kind selected, or no harness wired. The node @@ -300,9 +359,15 @@ async fn finish_agent_run( conn: Option<&str>, agent_ref: &str, outcome: crate::caps::AgentRunOutcome, + transcript: &TranscriptSink, ) -> Result { use crate::caps::StopReason; + // Before the stop reason is judged, so a run that paused or hit a limit + // still explains what it managed to do — that is exactly the run whose + // transcript is worth reading. + record_transcript(ctx, transcript, outcome.transcript); + let mut meta = serde_json::json!({ "stop": outcome.stop.as_str(), "agent_ref": agent_ref }); match &outcome.stop { diff --git a/src/nodes/integration/agent_tests/agent_tests_part_02_tests.rs b/src/nodes/integration/agent_tests/agent_tests_part_02_tests.rs index 9a446ac3..309f0668 100644 --- a/src/nodes/integration/agent_tests/agent_tests_part_02_tests.rs +++ b/src/nodes/integration/agent_tests/agent_tests_part_02_tests.rs @@ -364,3 +364,183 @@ mod configurable { assert_eq!(harness.list_agents().await.unwrap().len(), 1); } } + +// ---- transcripts: what the harness did inside the node ------------------ + +mod transcript { + use super::agent_node; + use crate::caps::mock::{MockAgentHarness, MockAgentRunner, mock_capabilities_with_agent}; + use crate::caps::AgentRunner; + use crate::data::Item; + use crate::model::AgentDefinition; + use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; + use crate::observability::{NoopObserver, RunObserver}; + use crate::transcript::TranscriptEntry; + use serde_json::{Value, json}; + use std::sync::{Arc, Mutex}; + + /// Records the live hook, so a test can prove entries arrive as the node + /// runs rather than only on the settled step. + #[derive(Default)] + struct LiveCapture { + seen: Mutex>, + } + + impl RunObserver for LiveCapture { + fn on_agent_event(&self, node_id: &str, entry: &TranscriptEntry) { + self.seen + .lock() + .unwrap() + .push((node_id.to_string(), entry.kind.clone())); + } + } + + async fn execute( + runner: Arc, + config: Value, + input: Vec, + observer: &dyn RunObserver, + ) -> NodeOutput { + let node = agent_node(config); + let caps = mock_capabilities_with_agent_arc(runner); + let agents: &[AgentDefinition] = &[]; + let run_meta = json!({ "run_id": "run_t", "sub_workflow_depth": 0 }); + super::super::AgentNode + .execute(NodeContext { + node: &node, + input: &input, + run: &run_meta, + nodes: &Value::Null, + caps: &caps, + agents, + observer, + token: crate::engine::CancellationToken::new(), + lane: None, + resume: None, + step: 0, + }) + .await + .expect("execute") + } + + /// `mock_capabilities_with_agent` takes a concrete runner; these tests need + /// to swap two different ones through the same helper. + fn mock_capabilities_with_agent_arc( + runner: Arc, + ) -> crate::caps::Capabilities { + let mut caps = mock_capabilities_with_agent(MockAgentRunner); + caps.agent = Some(runner); + caps + } + + fn harness() -> Arc { + Arc::new(MockAgentHarness::new()) + } + + #[tokio::test] + async fn a_harness_transcript_reaches_the_node_output() { + // The settled half: what the host reported on its `AgentRunOutcome` + // rides the `NodeOutput`, which is what the engine copies onto the step. + let out = execute( + harness(), + json!({ "agent_ref": "triager" }), + vec![Item::new(json!({ "seed": 1 }))], + &NoopObserver, + ) + .await; + + assert_eq!( + out.transcript + .iter() + .map(|e| e.kind.as_str()) + .collect::>(), + ["agent_thinking", "agent_message"], + "MockAgentHarness reports two entries; both must survive to the output" + ); + } + + #[tokio::test] + async fn the_same_entries_are_also_reported_live() { + // Both paths carry the same entries on purpose: a console watching a + // long node renders from `on_agent_event`, a run read back tomorrow + // renders from the step, and neither may be the only source. + let observer = LiveCapture::default(); + let out = execute( + harness(), + json!({ "agent_ref": "triager" }), + vec![Item::new(json!({ "seed": 1 }))], + &observer, + ) + .await; + + let seen = observer.seen.lock().unwrap(); + assert_eq!(seen.len(), out.transcript.len()); + assert!( + seen.iter().all(|(node, _)| node == "n"), + "every entry is attributed to the node that produced it" + ); + assert_eq!( + seen.iter() + .map(|(_, kind)| kind.as_str()) + .collect::>(), + ["agent_thinking", "agent_message"] + ); + } + + #[tokio::test] + async fn per_item_turns_accumulate_into_one_transcript() { + // Why the accumulator is shared rather than returned: a per-item node + // runs one turn per input and reports ONE step. Without it, every turn + // but the last would be dropped. + let out = execute( + harness(), + json!({ "agent_ref": "triager", "execution": "per_item" }), + vec![ + Item::new(json!({ "seed": 1 })), + Item::new(json!({ "seed": 2 })), + Item::new(json!({ "seed": 3 })), + ], + &NoopObserver, + ) + .await; + + assert_eq!(out.items.len(), 3, "one turn per input item"); + assert_eq!( + out.transcript.len(), + 6, + "two entries per turn, all three turns kept" + ); + } + + #[tokio::test] + async fn a_legacy_host_reports_no_transcript() { + // THE non-breaking guarantee. `MockAgentRunner` implements only the + // legacy `run_agent`, so the default `run` wraps its return in a + // `finished` outcome with no transcript. A host that never heard of this + // field keeps working and simply has nothing to say. + let out = execute( + Arc::new(MockAgentRunner), + json!({ "agent_ref": "triager" }), + vec![Item::new(json!({ "seed": 1 }))], + &NoopObserver, + ) + .await; + assert!(out.transcript.is_empty()); + assert_eq!(out.items.len(), 1, "the turn still ran and still emitted"); + } + + #[tokio::test] + async fn a_node_with_no_harness_reports_no_transcript() { + // The degraded path: no `agent_ref`, so the node falls back to + // `LlmProvider` and there is no harness to have a transcript. Empty is + // the honest answer, and must not be an error. + let out = execute( + harness(), + json!({ "prompt": "hi" }), + vec![Item::new(json!({}))], + &NoopObserver, + ) + .await; + assert!(out.transcript.is_empty()); + } +} diff --git a/src/observability.rs b/src/observability.rs index 8a1d557b..6382787e 100644 --- a/src/observability.rs +++ b/src/observability.rs @@ -35,6 +35,7 @@ //! output: serde_json::json!([]), //! duration_ms: 0, //! diagnostics: vec![], +//! transcript: vec![], //! }); //! assert_eq!(recorder.nodes.lock().unwrap().as_slice(), ["parse"]); //! ``` @@ -92,6 +93,21 @@ pub struct ExecutionStep { /// exact unresolved wiring behind a bad tool call. Empty on error steps and /// for nodes without expression config. pub diagnostics: Vec, + /// What a harness did inside this node, in order — the thinking, the tool + /// calls, the results. + /// + /// [`output`](Self::output) says what came *out* of the node; this says + /// what happened *inside* it, which for an `agent` node is most of what + /// there is to know. Folded by the host and handed over on + /// [`AgentRunOutcome::transcript`](crate::caps::AgentRunOutcome::transcript); + /// the engine copies it here and never interprets it. + /// + /// Empty for every non-`agent` node, for a host that reports none, and on + /// an error step — a node that failed produced no outcome to read one from. + /// A host watching a long node live wants + /// [`RunObserver::on_agent_event`](RunObserver::on_agent_event) instead; + /// this is the settled set. + pub transcript: Vec, } /// One execution of a workflow, captured as an ordered list of [`ExecutionStep`]s. @@ -178,6 +194,23 @@ pub trait RunObserver: Send + Sync { let _ = (node_id, index, total, ok); } + /// Called as a host's harness produces one transcript entry inside a + /// still-running `agent` node. + /// + /// The live counterpart of + /// [`ExecutionStep::transcript`](ExecutionStep::transcript), and the reason + /// it is not enough on its own: a step's transcript arrives when the node + /// *finishes*, and an agent node can run for minutes. A host that wants to + /// show what an agent is doing while it does it reads this; a host that + /// only persists finished runs can ignore it and lose nothing, since every + /// entry reported here also appears on the settled step. + /// + /// Called from the host's own harness thread, so an implementation must not + /// block — the convention is to hand the entry to a channel and return. + fn on_agent_event(&self, node_id: &str, entry: &crate::transcript::TranscriptEntry) { + let _ = (node_id, entry); + } + /// Called once, after the run settles, with the assembled [`Run`] record. fn on_run_finish(&self, run: &Run) { let _ = run; diff --git a/src/observability_tests.rs b/src/observability_tests.rs index e9b480c0..b52b28f7 100644 --- a/src/observability_tests.rs +++ b/src/observability_tests.rs @@ -12,6 +12,7 @@ fn noop_observer_callbacks_are_inert() { output: Value::Null, duration_ms: 0, diagnostics: Vec::new(), + transcript: vec![], }); observer.on_item_start("n", 0, 1); observer.on_item_finish("n", 0, 1, true); @@ -44,6 +45,7 @@ fn constructs_execution_step_with_each_status() { output: serde_json::json!([{ "json": { "x": 1 } }]), duration_ms: 12, diagnostics: Vec::new(), + transcript: vec![], }; assert_eq!(ok.node_id, "parse"); assert_eq!(ok.duration_ms, 12); @@ -56,6 +58,7 @@ fn constructs_execution_step_with_each_status() { output: Value::Null, duration_ms: 0, diagnostics: Vec::new(), + transcript: vec![], }; assert!(matches!(err.status, StepStatus::Error)); assert_eq!(err.output, Value::Null); @@ -72,6 +75,7 @@ fn constructs_run_with_steps() { output: serde_json::json!([]), duration_ms: 3, diagnostics: Vec::new(), + transcript: vec![], }], }; assert_eq!(run.id, "run-7"); @@ -129,6 +133,7 @@ fn custom_observer_receives_all_callbacks() { output: serde_json::json!([]), duration_ms: 1, diagnostics: Vec::new(), + transcript: vec![], }); observer.on_step_finish(&ExecutionStep { node_id: "second".to_string(), @@ -136,6 +141,7 @@ fn custom_observer_receives_all_callbacks() { output: Value::Null, duration_ms: 2, diagnostics: Vec::new(), + transcript: vec![], }); observer.on_run_finish(&Run { id: "run-9".to_string(), @@ -166,5 +172,103 @@ fn observer_is_usable_as_trait_object() { output: serde_json::json!([]), duration_ms: 0, diagnostics: Vec::new(), + transcript: vec![], }); } + +/// A `RunObserver` that records the live agent-event stream alongside steps, +/// so a test can tell the two delivery paths apart. +#[derive(Default)] +struct AgentCapture { + live: Mutex>, + settled: Mutex>, +} + +impl RunObserver for AgentCapture { + fn on_agent_event(&self, node_id: &str, entry: &crate::transcript::TranscriptEntry) { + self.live.lock().unwrap().push(( + node_id.to_string(), + entry.kind.clone(), + entry.text.clone(), + )); + } + + fn on_step_finish(&self, step: &ExecutionStep) { + self.settled + .lock() + .unwrap() + .push((step.node_id.clone(), step.transcript.len())); + } +} + +#[test] +fn on_agent_event_defaults_to_inert() { + // The whole point of the default body: a host that predates this hook keeps + // compiling and observes nothing new. + let observer = NoopObserver; + observer.on_agent_event( + "solve", + &crate::transcript::TranscriptEntry::bounded(1, "agent_thinking", "…"), + ); +} + +#[test] +fn agent_events_arrive_live_and_in_order() { + let observer = AgentCapture::default(); + for (at, kind, text) in [ + (1, "agent_thinking", "Collatz — memoise"), + (2, "tool_call", "shell: python3 solve.py"), + (3, "tool_result", "837799"), + ] { + observer.on_agent_event( + "solve", + &crate::transcript::TranscriptEntry::bounded(at, kind, text), + ); + } + + let live = observer.live.lock().unwrap(); + // Order is the contract: a transcript read out of order is not a transcript. + assert_eq!( + live.iter() + .map(|(_, kind, _)| kind.as_str()) + .collect::>(), + ["agent_thinking", "tool_call", "tool_result"] + ); + assert!(live.iter().all(|(node, _, _)| node == "solve")); + assert_eq!(live[2].2, "837799"); +} + +#[test] +fn a_step_carries_the_settled_transcript() { + let observer = AgentCapture::default(); + observer.on_step_finish(&ExecutionStep { + node_id: "solve".to_string(), + status: StepStatus::Success, + output: serde_json::json!([]), + duration_ms: 5, + diagnostics: Vec::new(), + transcript: vec![ + crate::transcript::TranscriptEntry::bounded(1, "agent_thinking", "a"), + crate::transcript::TranscriptEntry::bounded(2, "agent_message", "b"), + ], + }); + assert_eq!( + observer.settled.lock().unwrap().as_slice(), + [("solve".to_string(), 2)] + ); +} + +#[test] +fn a_non_agent_step_carries_no_transcript() { + // Empty is the normal case, and it must stay cheap: every non-agent node + // reports one of these on every activation. + let step = ExecutionStep { + node_id: "cost".to_string(), + status: StepStatus::Success, + output: serde_json::json!([]), + duration_ms: 0, + diagnostics: Vec::new(), + transcript: vec![], + }; + assert!(step.transcript.is_empty()); +} From 153355cb5e32ade23dcd4e43e849fc29bbd3541b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 23 Aug 2026 19:38:43 +0300 Subject: [PATCH 3/9] fix(adaptive): construct ExecutionStep's new transcript field MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `crates/adaptive` is a workspace member that builds an `ExecutionStep` in two places, and CI builds with `--workspace` where my local `cargo test` did not — so this only surfaced on the PR. `StepRecord::to_step` gets an empty transcript rather than a transported one, which is honest rather than lossy by accident: that type is a budget-bounded wire form that already clips `output` and narrows `duration_ms` from `u128`, and a transcript is many entries that would dwarf the budget the rest of it is held to. Nothing downstream reads it — `to_step` exists so `diagnose` can read a step on the far side, and diagnosis is a function of status, output and null bindings. Co-authored-by: Medulla --- crates/adaptive/src/execute/wire.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/crates/adaptive/src/execute/wire.rs b/crates/adaptive/src/execute/wire.rs index 4db60f76..257487dd 100644 --- a/crates/adaptive/src/execute/wire.rs +++ b/crates/adaptive/src/execute/wire.rs @@ -122,6 +122,17 @@ impl StepRecord { output: self.output.clone(), duration_ms: u128::from(self.duration_ms), diagnostics: self.null_bindings.clone(), + // Deliberately empty: `StepRecord` is a budget-bounded wire form + // and does not transport a harness transcript, which is many + // entries and would dwarf the budget the rest of this type is + // bounded to. The round trip is already lossy by design — `output` + // is clipped and `duration_ms` narrows — so a step reconstructed + // here genuinely has no transcript rather than having lost one. + // + // Nothing downstream of `to_step` reads it: this exists so + // `diagnose` can read a step on the far side, and diagnosis is a + // function of status, output and null bindings. + transcript: Vec::new(), } } } @@ -235,6 +246,7 @@ mod tests { output, duration_ms: 12, diagnostics: Vec::new(), + transcript: Vec::new(), } } From 6bbb31265e7b09de92e8c54c1b46db495e68ddd7 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 23 Aug 2026 19:55:25 +0300 Subject: [PATCH 4/9] fix(agent): drop the live hook, and make the settled transcript honest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review found that `on_agent_event` was not live, and it was right. The transcript rides the outcome `AgentRunner::run` returns, so the hook could only fire once the whole agent turn was over — in a burst, at the same moment `on_step_finish` gets the same entries. For the long-running node it was justified by, it published nothing until there was nothing left to publish. Removing it rather than patching the docs. Genuine live delivery needs a sink on the capability contract, and `AgentRunRequest` cannot carry one — it is `Serialize` + `PartialEq` — so it wants a new trait method designed on purpose, not a hook that already promises something it cannot do. Its only consumer does not implement it, so nothing is lost today and the API stops lying. Also from review: - Per-item transcripts are keyed by item index and drained in that order. `map_items` restores outputs to input order; appending on completion let a `concurrency > 1` run interleave differently from the next for identical input. - `TranscriptEntry::bounded` charges the truncation marker against `MAX_ENTRY_TEXT_BYTES` instead of adding it on top, so a clipped entry actually honours the documented cap. - `AgentRunOutcome::limit_stop` and `::paused`, because those two stop reasons carry data and had no constructor — a host reporting either had to write the struct literal, which is exactly what makes adding a field source-breaking. Migration is now one line. - `agent_tests_part_02` had grown to 546 lines; the transcript tests move to part 03. `caps/agent.rs` went to 527 with the constructors, so the outcome types move to `caps/agent/outcome.rs` — the request half and the outcome half divide cleanly. One review point is acknowledged and NOT fixed: an outcome the node turns into an `Err` — today only `StopReason::Paused` — still loses its transcript, because the engine builds that error step without a `NodeOutput` to carry one. That is the run whose transcript is most worth reading. Closing it means giving the error path somewhere to put one, which is an engine-shaped change; it is named in the field's docs rather than left to be discovered. Co-authored-by: Medulla --- src/caps/agent.rs | 201 +------------- src/caps/agent/outcome.rs | 250 ++++++++++++++++++ src/nodes/integration/agent.rs | 46 ++-- src/nodes/integration/agent_tests.rs | 1 + .../agent_tests/agent_tests_part_02_tests.rs | 180 ------------- .../agent_tests/agent_tests_part_03_tests.rs | 163 ++++++++++++ src/observability.rs | 30 +-- src/observability_tests.rs | 79 +----- src/transcript.rs | 17 +- 9 files changed, 477 insertions(+), 490 deletions(-) create mode 100644 src/caps/agent/outcome.rs create mode 100644 src/nodes/integration/agent_tests/agent_tests_part_03_tests.rs diff --git a/src/caps/agent.rs b/src/caps/agent.rs index c3794a95..37e05825 100644 --- a/src/caps/agent.rs +++ b/src/caps/agent.rs @@ -42,7 +42,6 @@ use serde_json::{Map, Value}; use crate::error::Result; use crate::model::{AgentDefinition, ToolGrant}; -use crate::transcript::TranscriptEntry; /// Which model, on which provider, an agent run should use. /// @@ -279,204 +278,8 @@ pub struct AgentRunRequest { pub config: Value, } -/// Why a host-owned agent loop stopped. -/// -/// The reason a typed outcome is worth having. With a bare `Value` return, an -/// agent that finished, one that stopped a step short of the answer, and one -/// waiting on a human are all indistinguishable — and the workflow marches -/// downstream with a partial answer in every case. Keeping the stop reason out -/// of the `Result` channel (rather than reporting a limit or a pause as an -/// error) is the same split [`ShellRunner`](crate::caps::ShellRunner) makes by -/// reporting a non-zero exit through `exit_code` instead of `Err`. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(tag = "stop", rename_all = "snake_case")] -pub enum StopReason { - /// The agent produced a final answer. The only reason a downstream node - /// should treat the outcome as complete. - Finished, - /// The loop hit a cap and stopped cleanly, keeping what it produced. The - /// outcome is **partial**: real, usable, and not the whole answer. - LimitStop { - /// Host-defined name of the cap that fired (`"max_steps"`, - /// `"token_budget"`, `"wall_clock"`). - /// - /// A free string, not an enum: the engine branches on *whether* a limit - /// fired, never on which, and an enum here would mint a taxonomy this - /// crate cannot keep current with any harness's budget model. - limit: String, - }, - /// The loop latched a pause and is **resumable, not finished** — the - /// harness still holds the transcript. - /// - /// The engine does not yet route a pause into its checkpoint/resume - /// machinery: an `agent` node that receives this fails with a clear - /// [`EngineError::Capability`](crate::error::EngineError::Capability) - /// naming the node and reason. The variant exists now so a harness can - /// never *conflate* a pause with a finish, and so no wire type has to - /// change when resume support lands. - Paused { - /// Opaque host handle for the paused run — a session id, a checkpoint - /// key — which the harness will need echoed back to resume it. - #[serde(default, skip_serializing_if = "Option::is_none")] - token: Option, - /// Host-defined reason (`"tool_approval"`, `"clarifying_question"`), - /// surfaced to whoever is being asked. - reason: String, - /// What the human must decide on. Opaque to the engine. - #[serde(default, skip_serializing_if = "Value::is_null")] - payload: Value, - }, -} - -impl StopReason { - /// The `snake_case` wire name of this reason, as it appears on the item - /// envelope's `meta.stop`. - #[must_use] - pub fn as_str(&self) -> &'static str { - match self { - Self::Finished => "finished", - Self::LimitStop { .. } => "limit_stop", - Self::Paused { .. } => "paused", - } - } -} - -/// Token and step accounting for one agent run, as the harness reports it. -/// -/// The engine neither aggregates nor enforces these; it forwards them onto the -/// item envelope's `meta.usage` for cost-aware workflows to read. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] -pub struct AgentUsage { - /// Prompt tokens consumed. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub input_tokens: Option, - /// Completion tokens produced. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub output_tokens: Option, - /// Model↔tool iterations the loop performed. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub steps: Option, - /// Tool invocations the loop performed. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub tool_calls: Option, -} - -/// What a host-owned agent run produced. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -pub struct AgentRunOutcome { - /// Why the loop stopped. **Read this before the payload.** - pub stop: StopReason, - /// The agent's prose answer, when it produced one. Lands at the item - /// envelope's `text`. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub text: Option, - /// The structured result, when the agent produced one. Lands at the - /// envelope's `json`, after the `output_parser` sub-port if one is - /// configured. [`Value::Null`] when the agent answered only in prose. - #[serde(default)] - pub json: Value, - /// The harness's native payload, verbatim. Lands at the envelope's `raw`, - /// and is the escape hatch for anything this struct does not model. - #[serde(default)] - pub raw: Value, - /// Optional usage figures. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub usage: Option, - /// What the harness did on the way to this outcome, in order. - /// - /// The node's payload above says what came *out* of the agent; this says - /// what happened *inside* it — the thinking, the tool calls, the results. - /// The engine copies it onto the step's - /// [`ExecutionStep::transcript`](crate::observability::ExecutionStep::transcript), - /// which is how it reaches a - /// [`RunObserver`](crate::observability::RunObserver) — so a host that - /// fills this in gets a run history that explains itself instead of one - /// that only reports pass or fail. - /// - /// Empty by default, and empty is a normal outcome: a harness with no - /// event stream to fold has nothing to say here, and every host that - /// predates this field keeps compiling and behaving identically. Bound each - /// entry with [`TranscriptEntry::bounded`] — the engine does not truncate. - /// - /// A host reporting entries *while* the node still runs uses - /// [`RunObserver::on_agent_event`](crate::observability::RunObserver::on_agent_event); - /// this field is the settled set. - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub transcript: Vec, -} - -impl AgentRunOutcome { - /// A [`Finished`](StopReason::Finished) outcome built from a harness's - /// native `value`, deriving `json` and `text` the way the engine's envelope - /// does: `json` is the value when it is an object or array, and `text` is - /// the value when it is a string, else its `text` field. - /// - /// This is what the default [`AgentRunner::run`] wraps a legacy - /// [`run_agent`](AgentRunner::run_agent) return in, and the shorthand a - /// simple adapter wants. - /// - /// ``` - /// use tinyflows::caps::{AgentRunOutcome, StopReason}; - /// use serde_json::json; - /// - /// let outcome = AgentRunOutcome::finished(json!({ "text": "done", "n": 1 })); - /// assert_eq!(outcome.stop, StopReason::Finished); - /// assert_eq!(outcome.text.as_deref(), Some("done")); - /// assert!(outcome.is_finished()); - /// ``` - #[must_use] - pub fn finished(value: Value) -> Self { - let text = match &value { - Value::String(s) => Some(s.clone()), - Value::Object(map) => map.get("text").and_then(Value::as_str).map(str::to_string), - _ => None, - }; - let json = match &value { - Value::Object(_) | Value::Array(_) => value.clone(), - _ => Value::Null, - }; - Self { - stop: StopReason::Finished, - text, - json, - raw: value, - usage: None, - transcript: Vec::new(), - } - } - - /// The same outcome, carrying what the harness did to reach it. - /// - /// The builder half of [`transcript`](Self::transcript), so a host can fold - /// its event stream once and attach it without naming every other field: - /// - /// ``` - /// use tinyflows::caps::AgentRunOutcome; - /// use tinyflows::transcript::TranscriptEntry; - /// use serde_json::json; - /// - /// let outcome = AgentRunOutcome::finished(json!("837799")) - /// .with_transcript(vec![ - /// TranscriptEntry::bounded(1, "agent_thinking", "Collatz — memoise."), - /// TranscriptEntry::bounded(2, "tool_call", "shell: python3 solve.py"), - /// ]); - /// assert_eq!(outcome.transcript.len(), 2); - /// ``` - #[must_use] - pub fn with_transcript(mut self, transcript: Vec) -> Self { - self.transcript = transcript; - self - } - - /// Whether the agent reached a final answer. - /// - /// Anything else means the payload is partial or absent — see - /// [`StopReason`]. - #[must_use] - pub fn is_finished(&self) -> bool { - matches!(self.stop, StopReason::Finished) - } -} +mod outcome; +pub use outcome::{AgentRunOutcome, AgentUsage, StopReason}; mod runner; pub use runner::AgentRunner; diff --git a/src/caps/agent/outcome.rs b/src/caps/agent/outcome.rs new file mode 100644 index 00000000..e2a9ebe8 --- /dev/null +++ b/src/caps/agent/outcome.rs @@ -0,0 +1,250 @@ +//! What a host-owned agent loop produced, and why it stopped. +//! +//! Split from `agent.rs`, which owns the *request* half: that file was pushed +//! past the repository's 500-line limit by the constructors below, and the two +//! halves divide cleanly — everything here describes an outcome coming back. + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::transcript::TranscriptEntry; + +/// Why a host-owned agent loop stopped. +/// +/// The reason a typed outcome is worth having. With a bare `Value` return, an +/// agent that finished, one that stopped a step short of the answer, and one +/// waiting on a human are all indistinguishable — and the workflow marches +/// downstream with a partial answer in every case. Keeping the stop reason out +/// of the `Result` channel (rather than reporting a limit or a pause as an +/// error) is the same split [`ShellRunner`](crate::caps::ShellRunner) makes by +/// reporting a non-zero exit through `exit_code` instead of `Err`. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "stop", rename_all = "snake_case")] +pub enum StopReason { + /// The agent produced a final answer. The only reason a downstream node + /// should treat the outcome as complete. + Finished, + /// The loop hit a cap and stopped cleanly, keeping what it produced. The + /// outcome is **partial**: real, usable, and not the whole answer. + LimitStop { + /// Host-defined name of the cap that fired (`"max_steps"`, + /// `"token_budget"`, `"wall_clock"`). + /// + /// A free string, not an enum: the engine branches on *whether* a limit + /// fired, never on which, and an enum here would mint a taxonomy this + /// crate cannot keep current with any harness's budget model. + limit: String, + }, + /// The loop latched a pause and is **resumable, not finished** — the + /// harness still holds the transcript. + /// + /// The engine does not yet route a pause into its checkpoint/resume + /// machinery: an `agent` node that receives this fails with a clear + /// [`EngineError::Capability`](crate::error::EngineError::Capability) + /// naming the node and reason. The variant exists now so a harness can + /// never *conflate* a pause with a finish, and so no wire type has to + /// change when resume support lands. + Paused { + /// Opaque host handle for the paused run — a session id, a checkpoint + /// key — which the harness will need echoed back to resume it. + #[serde(default, skip_serializing_if = "Option::is_none")] + token: Option, + /// Host-defined reason (`"tool_approval"`, `"clarifying_question"`), + /// surfaced to whoever is being asked. + reason: String, + /// What the human must decide on. Opaque to the engine. + #[serde(default, skip_serializing_if = "Value::is_null")] + payload: Value, + }, +} + +impl StopReason { + /// The `snake_case` wire name of this reason, as it appears on the item + /// envelope's `meta.stop`. + #[must_use] + pub fn as_str(&self) -> &'static str { + match self { + Self::Finished => "finished", + Self::LimitStop { .. } => "limit_stop", + Self::Paused { .. } => "paused", + } + } +} + +/// Token and step accounting for one agent run, as the harness reports it. +/// +/// The engine neither aggregates nor enforces these; it forwards them onto the +/// item envelope's `meta.usage` for cost-aware workflows to read. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct AgentUsage { + /// Prompt tokens consumed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub input_tokens: Option, + /// Completion tokens produced. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub output_tokens: Option, + /// Model↔tool iterations the loop performed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub steps: Option, + /// Tool invocations the loop performed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_calls: Option, +} + +/// What a host-owned agent run produced. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct AgentRunOutcome { + /// Why the loop stopped. **Read this before the payload.** + pub stop: StopReason, + /// The agent's prose answer, when it produced one. Lands at the item + /// envelope's `text`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub text: Option, + /// The structured result, when the agent produced one. Lands at the + /// envelope's `json`, after the `output_parser` sub-port if one is + /// configured. [`Value::Null`] when the agent answered only in prose. + #[serde(default)] + pub json: Value, + /// The harness's native payload, verbatim. Lands at the envelope's `raw`, + /// and is the escape hatch for anything this struct does not model. + #[serde(default)] + pub raw: Value, + /// Optional usage figures. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub usage: Option, + /// What the harness did on the way to this outcome, in order. + /// + /// The node's payload above says what came *out* of the agent; this says + /// what happened *inside* it — the thinking, the tool calls, the results. + /// The engine copies it onto the step's + /// [`ExecutionStep::transcript`](crate::observability::ExecutionStep::transcript), + /// which is how it reaches a + /// [`RunObserver`](crate::observability::RunObserver) — so a host that + /// fills this in gets a run history that explains itself instead of one + /// that only reports pass or fail. + /// + /// Empty by default, and empty is a normal outcome: a harness with no + /// event stream to fold has nothing to say here, and every host that + /// predates this field keeps compiling and behaving identically. Bound each + /// entry with [`TranscriptEntry::bounded`] — the engine does not truncate. + /// + /// **Settled, not live.** These ride the outcome, so they exist only once + /// the run is over. Reporting entries *during* a run would need a sink on + /// this capability, which [`AgentRunRequest`] cannot carry — it is + /// `Serialize` + `PartialEq` — so that is a deliberate follow-up rather + /// than something to imply here. + /// + /// **Known gap:** an outcome the `agent` node turns into an `Err` — today + /// [`StopReason::Paused`], which the engine cannot yet resume — loses its + /// transcript, because the engine's error step is built without a + /// `NodeOutput` to carry one. That is the run whose transcript is most + /// worth reading, and closing it means giving the error path somewhere to + /// put one. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub transcript: Vec, +} + +impl AgentRunOutcome { + /// A [`Finished`](StopReason::Finished) outcome built from a harness's + /// native `value`, deriving `json` and `text` the way the engine's envelope + /// does: `json` is the value when it is an object or array, and `text` is + /// the value when it is a string, else its `text` field. + /// + /// This is what the default [`AgentRunner::run`] wraps a legacy + /// [`run_agent`](AgentRunner::run_agent) return in, and the shorthand a + /// simple adapter wants. + /// + /// ``` + /// use tinyflows::caps::{AgentRunOutcome, StopReason}; + /// use serde_json::json; + /// + /// let outcome = AgentRunOutcome::finished(json!({ "text": "done", "n": 1 })); + /// assert_eq!(outcome.stop, StopReason::Finished); + /// assert_eq!(outcome.text.as_deref(), Some("done")); + /// assert!(outcome.is_finished()); + /// ``` + #[must_use] + pub fn finished(value: Value) -> Self { + let text = match &value { + Value::String(s) => Some(s.clone()), + Value::Object(map) => map.get("text").and_then(Value::as_str).map(str::to_string), + _ => None, + }; + let json = match &value { + Value::Object(_) | Value::Array(_) => value.clone(), + _ => Value::Null, + }; + Self { + stop: StopReason::Finished, + text, + json, + raw: value, + usage: None, + transcript: Vec::new(), + } + } + + /// The same outcome, carrying what the harness did to reach it. + /// + /// The builder half of [`transcript`](Self::transcript), so a host can fold + /// its event stream once and attach it without naming every other field: + /// + /// ``` + /// use tinyflows::caps::AgentRunOutcome; + /// use tinyflows::transcript::TranscriptEntry; + /// use serde_json::json; + /// + /// let outcome = AgentRunOutcome::finished(json!("837799")) + /// .with_transcript(vec![ + /// TranscriptEntry::bounded(1, "agent_thinking", "Collatz — memoise."), + /// TranscriptEntry::bounded(2, "tool_call", "shell: python3 solve.py"), + /// ]); + /// assert_eq!(outcome.transcript.len(), 2); + /// ``` + #[must_use] + pub fn with_transcript(mut self, transcript: Vec) -> Self { + self.transcript = transcript; + self + } + + /// A [`LimitStop`](StopReason::LimitStop) outcome: real, usable, partial. + /// + /// Beside [`finished`](Self::finished) because `LimitStop` and + /// [`Paused`](StopReason::Paused) carry data and so had no constructor — a + /// host reporting either had to write the struct literal, which is what + /// makes adding a field to this type source-breaking. These two exist so + /// that migration is one line. + #[must_use] + pub fn limit_stop(value: Value, limit: impl Into) -> Self { + Self { + stop: StopReason::LimitStop { + limit: limit.into(), + }, + ..Self::finished(value) + } + } + + /// A [`Paused`](StopReason::Paused) outcome: resumable, not finished. + /// + /// See [`limit_stop`](Self::limit_stop) for why this exists. + #[must_use] + pub fn paused(token: Option, reason: impl Into, payload: Value) -> Self { + Self { + stop: StopReason::Paused { + token, + reason: reason.into(), + payload, + }, + ..Self::finished(Value::Null) + } + } + + /// Whether the agent reached a final answer. + /// + /// Anything else means the payload is partial or absent — see + /// [`StopReason`]. + #[must_use] + pub fn is_finished(&self) -> bool { + matches!(self.stop, StopReason::Finished) + } +} diff --git a/src/nodes/integration/agent.rs b/src/nodes/integration/agent.rs index 22092991..fee76741 100644 --- a/src/nodes/integration/agent.rs +++ b/src/nodes/integration/agent.rs @@ -1,5 +1,6 @@ //! The `agent` node: an LLM agent turn with optional sub-ports. +use std::collections::BTreeMap; use std::sync::{Arc, Mutex}; use async_trait::async_trait; @@ -17,25 +18,33 @@ use crate::transcript::TranscriptEntry; /// reports carries a single transcript, so the entries have to accumulate /// somewhere shared. Shared rather than returned so `map_items`' closure /// signature stays as it is. -type TranscriptSink = Arc>>; +/// +/// Keyed by **item index** rather than appended flat, because `per_item` with +/// `concurrency > 1` completes turns in whatever order they finish while +/// `map_items` deliberately restores its outputs to input order. Appending on +/// completion would make one run's transcript interleave differently from the +/// next for identical input; draining in key order makes it match the items it +/// describes. `None` is the single-turn case, which sorts first and is alone. +type TranscriptSink = Arc, Vec>>>; -/// Reports `entries` live, then keeps them for the settled step. +/// Keeps `entries` for the settled step, under the item they belong to. /// -/// Both halves matter and neither replaces the other: `on_agent_event` is -/// what lets a console show an agent working through a node that runs for -/// minutes, and the accumulated copy is what a run read back tomorrow has. -fn record_transcript(ctx: &NodeContext<'_>, sink: &TranscriptSink, entries: Vec) { +/// Not reported live: they ride the outcome the harness returns, so by the time +/// this runs the turn is over. See `ExecutionStep::transcript`. +fn record_transcript( + ctx: &NodeContext<'_>, + sink: &TranscriptSink, + item_index: Option, + entries: Vec, +) { if entries.is_empty() { return; } - for entry in &entries { - ctx.observer.on_agent_event(&ctx.node.id, entry); - } - // A poisoned lock here must not fail the turn: the agent already ran and - // its answer is good. Losing the transcript degrades the run history, - // which is strictly better than discarding the work. + // A poisoned lock must not fail the turn: the agent already ran and its + // answer is good. Losing the transcript degrades the run history, which is + // strictly better than discarding the work. match sink.lock() { - Ok(mut held) => held.extend(entries), + Ok(mut held) => held.entry(item_index).or_default().extend(entries), Err(err) => tracing::warn!( node = %ctx.node.id, %err, @@ -44,10 +53,10 @@ fn record_transcript(ctx: &NodeContext<'_>, sink: &TranscriptSink, entries: Vec< } } -/// Takes everything the sink holds, leaving it empty. +/// Takes everything the sink holds, in item order, leaving it empty. fn drain(sink: &TranscriptSink) -> Vec { sink.lock() - .map(|mut held| std::mem::take(&mut *held)) + .map(|mut held| std::mem::take(&mut *held).into_values().flatten().collect()) .unwrap_or_default() } @@ -126,7 +135,7 @@ impl NodeExecutor for AgentNode { // One accumulator for the whole node activation, because a `per_item` // agent node runs many turns and the step carries one transcript. Shared // rather than returned, so `map_items`' closure signature is untouched. - let transcript: TranscriptSink = Arc::new(Mutex::new(Vec::new())); + let transcript: TranscriptSink = Arc::new(Mutex::new(BTreeMap::new())); if per_item { // Fan out: `config.concurrency` decides how many turns run at once @@ -205,7 +214,7 @@ async fn run_turn_indexed( let request = super::agent_request::assemble(ctx, cfg, agent_ref, scope, item_index).await?; let outcome = runner.run(request).await?; - return finish_agent_run(ctx, cfg, conn, agent_ref, outcome, transcript).await; + return finish_agent_run(ctx, cfg, conn, agent_ref, outcome, transcript, item_index).await; } // Degraded path: no agent kind selected, or no harness wired. The node @@ -360,13 +369,14 @@ async fn finish_agent_run( agent_ref: &str, outcome: crate::caps::AgentRunOutcome, transcript: &TranscriptSink, + item_index: Option, ) -> Result { use crate::caps::StopReason; // Before the stop reason is judged, so a run that paused or hit a limit // still explains what it managed to do — that is exactly the run whose // transcript is worth reading. - record_transcript(ctx, transcript, outcome.transcript); + record_transcript(ctx, transcript, item_index, outcome.transcript); let mut meta = serde_json::json!({ "stop": outcome.stop.as_str(), "agent_ref": agent_ref }); diff --git a/src/nodes/integration/agent_tests.rs b/src/nodes/integration/agent_tests.rs index 53d055cc..821a21a7 100644 --- a/src/nodes/integration/agent_tests.rs +++ b/src/nodes/integration/agent_tests.rs @@ -38,3 +38,4 @@ fn wf(kind: NodeKind, config: Value) -> WorkflowGraph { include!("agent_tests/agent_tests_part_01_tests.rs"); include!("agent_tests/agent_tests_part_02_tests.rs"); +include!("agent_tests/agent_tests_part_03_tests.rs"); diff --git a/src/nodes/integration/agent_tests/agent_tests_part_02_tests.rs b/src/nodes/integration/agent_tests/agent_tests_part_02_tests.rs index 309f0668..9a446ac3 100644 --- a/src/nodes/integration/agent_tests/agent_tests_part_02_tests.rs +++ b/src/nodes/integration/agent_tests/agent_tests_part_02_tests.rs @@ -364,183 +364,3 @@ mod configurable { assert_eq!(harness.list_agents().await.unwrap().len(), 1); } } - -// ---- transcripts: what the harness did inside the node ------------------ - -mod transcript { - use super::agent_node; - use crate::caps::mock::{MockAgentHarness, MockAgentRunner, mock_capabilities_with_agent}; - use crate::caps::AgentRunner; - use crate::data::Item; - use crate::model::AgentDefinition; - use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; - use crate::observability::{NoopObserver, RunObserver}; - use crate::transcript::TranscriptEntry; - use serde_json::{Value, json}; - use std::sync::{Arc, Mutex}; - - /// Records the live hook, so a test can prove entries arrive as the node - /// runs rather than only on the settled step. - #[derive(Default)] - struct LiveCapture { - seen: Mutex>, - } - - impl RunObserver for LiveCapture { - fn on_agent_event(&self, node_id: &str, entry: &TranscriptEntry) { - self.seen - .lock() - .unwrap() - .push((node_id.to_string(), entry.kind.clone())); - } - } - - async fn execute( - runner: Arc, - config: Value, - input: Vec, - observer: &dyn RunObserver, - ) -> NodeOutput { - let node = agent_node(config); - let caps = mock_capabilities_with_agent_arc(runner); - let agents: &[AgentDefinition] = &[]; - let run_meta = json!({ "run_id": "run_t", "sub_workflow_depth": 0 }); - super::super::AgentNode - .execute(NodeContext { - node: &node, - input: &input, - run: &run_meta, - nodes: &Value::Null, - caps: &caps, - agents, - observer, - token: crate::engine::CancellationToken::new(), - lane: None, - resume: None, - step: 0, - }) - .await - .expect("execute") - } - - /// `mock_capabilities_with_agent` takes a concrete runner; these tests need - /// to swap two different ones through the same helper. - fn mock_capabilities_with_agent_arc( - runner: Arc, - ) -> crate::caps::Capabilities { - let mut caps = mock_capabilities_with_agent(MockAgentRunner); - caps.agent = Some(runner); - caps - } - - fn harness() -> Arc { - Arc::new(MockAgentHarness::new()) - } - - #[tokio::test] - async fn a_harness_transcript_reaches_the_node_output() { - // The settled half: what the host reported on its `AgentRunOutcome` - // rides the `NodeOutput`, which is what the engine copies onto the step. - let out = execute( - harness(), - json!({ "agent_ref": "triager" }), - vec![Item::new(json!({ "seed": 1 }))], - &NoopObserver, - ) - .await; - - assert_eq!( - out.transcript - .iter() - .map(|e| e.kind.as_str()) - .collect::>(), - ["agent_thinking", "agent_message"], - "MockAgentHarness reports two entries; both must survive to the output" - ); - } - - #[tokio::test] - async fn the_same_entries_are_also_reported_live() { - // Both paths carry the same entries on purpose: a console watching a - // long node renders from `on_agent_event`, a run read back tomorrow - // renders from the step, and neither may be the only source. - let observer = LiveCapture::default(); - let out = execute( - harness(), - json!({ "agent_ref": "triager" }), - vec![Item::new(json!({ "seed": 1 }))], - &observer, - ) - .await; - - let seen = observer.seen.lock().unwrap(); - assert_eq!(seen.len(), out.transcript.len()); - assert!( - seen.iter().all(|(node, _)| node == "n"), - "every entry is attributed to the node that produced it" - ); - assert_eq!( - seen.iter() - .map(|(_, kind)| kind.as_str()) - .collect::>(), - ["agent_thinking", "agent_message"] - ); - } - - #[tokio::test] - async fn per_item_turns_accumulate_into_one_transcript() { - // Why the accumulator is shared rather than returned: a per-item node - // runs one turn per input and reports ONE step. Without it, every turn - // but the last would be dropped. - let out = execute( - harness(), - json!({ "agent_ref": "triager", "execution": "per_item" }), - vec![ - Item::new(json!({ "seed": 1 })), - Item::new(json!({ "seed": 2 })), - Item::new(json!({ "seed": 3 })), - ], - &NoopObserver, - ) - .await; - - assert_eq!(out.items.len(), 3, "one turn per input item"); - assert_eq!( - out.transcript.len(), - 6, - "two entries per turn, all three turns kept" - ); - } - - #[tokio::test] - async fn a_legacy_host_reports_no_transcript() { - // THE non-breaking guarantee. `MockAgentRunner` implements only the - // legacy `run_agent`, so the default `run` wraps its return in a - // `finished` outcome with no transcript. A host that never heard of this - // field keeps working and simply has nothing to say. - let out = execute( - Arc::new(MockAgentRunner), - json!({ "agent_ref": "triager" }), - vec![Item::new(json!({ "seed": 1 }))], - &NoopObserver, - ) - .await; - assert!(out.transcript.is_empty()); - assert_eq!(out.items.len(), 1, "the turn still ran and still emitted"); - } - - #[tokio::test] - async fn a_node_with_no_harness_reports_no_transcript() { - // The degraded path: no `agent_ref`, so the node falls back to - // `LlmProvider` and there is no harness to have a transcript. Empty is - // the honest answer, and must not be an error. - let out = execute( - harness(), - json!({ "prompt": "hi" }), - vec![Item::new(json!({}))], - &NoopObserver, - ) - .await; - assert!(out.transcript.is_empty()); - } -} diff --git a/src/nodes/integration/agent_tests/agent_tests_part_03_tests.rs b/src/nodes/integration/agent_tests/agent_tests_part_03_tests.rs new file mode 100644 index 00000000..843d5856 --- /dev/null +++ b/src/nodes/integration/agent_tests/agent_tests_part_03_tests.rs @@ -0,0 +1,163 @@ +// ---- transcripts: what the harness did inside the node ------------------ +// +// Split out of part 02, which the repo's 500-line rule would otherwise have +// pushed over. + +mod transcript { + use super::agent_node; + use crate::caps::mock::{MockAgentHarness, MockAgentRunner, mock_capabilities_with_agent}; + use crate::caps::AgentRunner; + use crate::data::Item; + use crate::model::AgentDefinition; + use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; + use crate::observability::{NoopObserver, RunObserver}; + use serde_json::{Value, json}; + use std::sync::Arc; + + async fn execute( + runner: Arc, + config: Value, + input: Vec, + observer: &dyn RunObserver, + ) -> NodeOutput { + let node = agent_node(config); + let caps = mock_capabilities_with_agent_arc(runner); + let agents: &[AgentDefinition] = &[]; + let run_meta = json!({ "run_id": "run_t", "sub_workflow_depth": 0 }); + super::super::AgentNode + .execute(NodeContext { + node: &node, + input: &input, + run: &run_meta, + nodes: &Value::Null, + caps: &caps, + agents, + observer, + token: crate::engine::CancellationToken::new(), + lane: None, + resume: None, + step: 0, + }) + .await + .expect("execute") + } + + /// `mock_capabilities_with_agent` takes a concrete runner; these tests need + /// to swap two different ones through the same helper. + fn mock_capabilities_with_agent_arc( + runner: Arc, + ) -> crate::caps::Capabilities { + let mut caps = mock_capabilities_with_agent(MockAgentRunner); + caps.agent = Some(runner); + caps + } + + fn harness() -> Arc { + Arc::new(MockAgentHarness::new()) + } + + #[tokio::test] + async fn a_harness_transcript_reaches_the_node_output() { + // The settled half: what the host reported on its `AgentRunOutcome` + // rides the `NodeOutput`, which is what the engine copies onto the step. + let out = execute( + harness(), + json!({ "agent_ref": "triager" }), + vec![Item::new(json!({ "seed": 1 }))], + &NoopObserver, + ) + .await; + + assert_eq!( + out.transcript + .iter() + .map(|e| e.kind.as_str()) + .collect::>(), + ["agent_thinking", "agent_message"], + "MockAgentHarness reports two entries; both must survive to the output" + ); + } + + #[tokio::test] + async fn per_item_turns_accumulate_into_one_transcript() { + // Why the accumulator is shared rather than returned: a per-item node + // runs one turn per input and reports ONE step. Without it, every turn + // but the last would be dropped. + let out = execute( + harness(), + json!({ "agent_ref": "triager", "execution": "per_item" }), + vec![ + Item::new(json!({ "seed": 1 })), + Item::new(json!({ "seed": 2 })), + Item::new(json!({ "seed": 3 })), + ], + &NoopObserver, + ) + .await; + + assert_eq!(out.items.len(), 3, "one turn per input item"); + assert_eq!( + out.transcript.len(), + 6, + "two entries per turn, all three turns kept" + ); + } + + #[tokio::test] + async fn a_legacy_host_reports_no_transcript() { + // THE non-breaking guarantee. `MockAgentRunner` implements only the + // legacy `run_agent`, so the default `run` wraps its return in a + // `finished` outcome with no transcript. A host that never heard of this + // field keeps working and simply has nothing to say. + let out = execute( + Arc::new(MockAgentRunner), + json!({ "agent_ref": "triager" }), + vec![Item::new(json!({ "seed": 1 }))], + &NoopObserver, + ) + .await; + assert!(out.transcript.is_empty()); + assert_eq!(out.items.len(), 1, "the turn still ran and still emitted"); + } + + #[tokio::test] + async fn a_node_with_no_harness_reports_no_transcript() { + // The degraded path: no `agent_ref`, so the node falls back to + // `LlmProvider` and there is no harness to have a transcript. Empty is + // the honest answer, and must not be an error. + let out = execute( + harness(), + json!({ "prompt": "hi" }), + vec![Item::new(json!({}))], + &NoopObserver, + ) + .await; + assert!(out.transcript.is_empty()); + } + + #[tokio::test] + async fn per_item_transcripts_come_back_in_item_order() { + // `map_items` restores its OUTPUTS to input order, so the transcript + // describing them has to match. Appending on completion would let a + // concurrent run interleave differently from one run to the next, for + // identical input — the sink is keyed by item index to prevent that. + let out = execute( + harness(), + json!({ "agent_ref": "triager", "execution": "per_item", "concurrency": 4 }), + (0..4).map(|n| Item::new(json!({ "seed": n }))).collect(), + &NoopObserver, + ) + .await; + + // MockAgentHarness names the agent in its second entry, and every item + // runs the same agent — so what is asserted here is the *grouping*: + // each turn's pair stays together and the pairs stay in item order. + assert_eq!(out.transcript.len(), 8); + let kinds: Vec<&str> = out.transcript.iter().map(|e| e.kind.as_str()).collect(); + assert_eq!( + kinds, + ["agent_thinking", "agent_message"].repeat(4), + "each turn contributes its pair intact, in item order" + ); + } +} diff --git a/src/observability.rs b/src/observability.rs index 6382787e..b1c6ae67 100644 --- a/src/observability.rs +++ b/src/observability.rs @@ -103,10 +103,15 @@ pub struct ExecutionStep { /// the engine copies it here and never interprets it. /// /// Empty for every non-`agent` node, for a host that reports none, and on - /// an error step — a node that failed produced no outcome to read one from. - /// A host watching a long node live wants - /// [`RunObserver::on_agent_event`](RunObserver::on_agent_event) instead; - /// this is the settled set. + /// an error step — a node that failed produced no outcome to read one from, + /// which is a known gap for a paused agent (see `AgentRunOutcome::transcript`). + /// + /// **Settled, not live.** Entries arrive when the node finishes, because + /// they ride the outcome the harness returns. Reporting them *during* a run + /// would need a sink on the capability contract, which + /// [`AgentRunRequest`](crate::caps::AgentRunRequest) cannot carry (it is + /// `Serialize` + `PartialEq`); that is a deliberate follow-up rather than + /// something to imply here. pub transcript: Vec, } @@ -194,23 +199,6 @@ pub trait RunObserver: Send + Sync { let _ = (node_id, index, total, ok); } - /// Called as a host's harness produces one transcript entry inside a - /// still-running `agent` node. - /// - /// The live counterpart of - /// [`ExecutionStep::transcript`](ExecutionStep::transcript), and the reason - /// it is not enough on its own: a step's transcript arrives when the node - /// *finishes*, and an agent node can run for minutes. A host that wants to - /// show what an agent is doing while it does it reads this; a host that - /// only persists finished runs can ignore it and lose nothing, since every - /// entry reported here also appears on the settled step. - /// - /// Called from the host's own harness thread, so an implementation must not - /// block — the convention is to hand the entry to a channel and return. - fn on_agent_event(&self, node_id: &str, entry: &crate::transcript::TranscriptEntry) { - let _ = (node_id, entry); - } - /// Called once, after the run settles, with the assembled [`Run`] record. fn on_run_finish(&self, run: &Run) { let _ = run; diff --git a/src/observability_tests.rs b/src/observability_tests.rs index b52b28f7..b5c0b389 100644 --- a/src/observability_tests.rs +++ b/src/observability_tests.rs @@ -176,72 +176,13 @@ fn observer_is_usable_as_trait_object() { }); } -/// A `RunObserver` that records the live agent-event stream alongside steps, -/// so a test can tell the two delivery paths apart. -#[derive(Default)] -struct AgentCapture { - live: Mutex>, - settled: Mutex>, -} - -impl RunObserver for AgentCapture { - fn on_agent_event(&self, node_id: &str, entry: &crate::transcript::TranscriptEntry) { - self.live.lock().unwrap().push(( - node_id.to_string(), - entry.kind.clone(), - entry.text.clone(), - )); - } - - fn on_step_finish(&self, step: &ExecutionStep) { - self.settled - .lock() - .unwrap() - .push((step.node_id.clone(), step.transcript.len())); - } -} - -#[test] -fn on_agent_event_defaults_to_inert() { - // The whole point of the default body: a host that predates this hook keeps - // compiling and observes nothing new. - let observer = NoopObserver; - observer.on_agent_event( - "solve", - &crate::transcript::TranscriptEntry::bounded(1, "agent_thinking", "…"), - ); -} - -#[test] -fn agent_events_arrive_live_and_in_order() { - let observer = AgentCapture::default(); - for (at, kind, text) in [ - (1, "agent_thinking", "Collatz — memoise"), - (2, "tool_call", "shell: python3 solve.py"), - (3, "tool_result", "837799"), - ] { - observer.on_agent_event( - "solve", - &crate::transcript::TranscriptEntry::bounded(at, kind, text), - ); - } - - let live = observer.live.lock().unwrap(); - // Order is the contract: a transcript read out of order is not a transcript. - assert_eq!( - live.iter() - .map(|(_, kind, _)| kind.as_str()) - .collect::>(), - ["agent_thinking", "tool_call", "tool_result"] - ); - assert!(live.iter().all(|(node, _, _)| node == "solve")); - assert_eq!(live[2].2, "837799"); -} - +/// A step carries what the harness did, and it reaches an observer through +/// `on_step_finish` — the settled set, not a live feed. See +/// [`ExecutionStep::transcript`]. #[test] fn a_step_carries_the_settled_transcript() { - let observer = AgentCapture::default(); - observer.on_step_finish(&ExecutionStep { + let observer = Capture::default(); + let step = ExecutionStep { node_id: "solve".to_string(), status: StepStatus::Success, output: serde_json::json!([]), @@ -251,11 +192,11 @@ fn a_step_carries_the_settled_transcript() { crate::transcript::TranscriptEntry::bounded(1, "agent_thinking", "a"), crate::transcript::TranscriptEntry::bounded(2, "agent_message", "b"), ], - }); - assert_eq!( - observer.settled.lock().unwrap().as_slice(), - [("solve".to_string(), 2)] - ); + }; + observer.on_step_finish(&step); + assert_eq!(observer.steps.lock().unwrap().as_slice(), ["solve"]); + assert_eq!(step.transcript.len(), 2); + assert_eq!(step.transcript[0].kind, "agent_thinking"); } #[test] diff --git a/src/transcript.rs b/src/transcript.rs index 045788b0..98f211d8 100644 --- a/src/transcript.rs +++ b/src/transcript.rs @@ -28,7 +28,10 @@ use serde::{Deserialize, Serialize}; -/// Bytes of one entry's `text` kept on the durable record. +/// Bytes of one entry's `text`, marker included. +/// +/// A clipped entry is at most this long *in total* — the truncation marker +/// is charged against the budget, not added after it. /// /// A stored run bounds step `input`, `output` and its own `inputs` so no /// single value can grow a run record without limit; a transcript entry is @@ -42,6 +45,10 @@ use serde::{Deserialize, Serialize}; /// pointing at it. That split is why this ceiling can stay small. pub const MAX_ENTRY_TEXT_BYTES: usize = 4 * 1024; +/// Appended to a clipped entry, and counted against +/// [`MAX_ENTRY_TEXT_BYTES`] rather than added on top of it. +const TRUNCATION_MARKER: &str = " …[truncated]"; + /// One thing an agent did, in the order it did it. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -71,14 +78,18 @@ impl TranscriptEntry { pub fn bounded(at_ms: i64, kind: impl Into, text: impl Into) -> Self { let mut text = text.into(); if text.len() > MAX_ENTRY_TEXT_BYTES { + // Reserve the marker's own bytes BEFORE choosing the cut, so the + // finished entry honours the cap rather than exceeding it by the + // length of the thing announcing the cut. + let budget = MAX_ENTRY_TEXT_BYTES - TRUNCATION_MARKER.len(); let end = text .char_indices() .map(|(index, _)| index) - .take_while(|index| *index <= MAX_ENTRY_TEXT_BYTES) + .take_while(|index| *index <= budget) .last() .unwrap_or(0); text.truncate(end); - text.push_str(" …[truncated]"); + text.push_str(TRUNCATION_MARKER); } Self { at_ms, From 886ccab21910d7602a6713e1343636a55b137598 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 23 Aug 2026 20:02:51 +0300 Subject: [PATCH 5/9] docs: clarify that transcript entries are settled, not live, and fix intra-doc links The doc comment in `src/transcript.rs` now explicitly states that transcript entries ride the outcome and reach an observer only when the node finishes, rather than implying they can be reported during a run. The previous cross-reference to `RunObserver::on_agent_event` was misleading because that method is for live reporting, not for the settled entries this module describes. In `src/caps/agent/outcome.rs`, the intra-doc links to `AgentRunRequest` and `AgentRunner` were updated to use the full path syntax so they resolve correctly in the rendered documentation. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/caps/agent/outcome.rs | 6 +++--- src/transcript.rs | 9 ++++++--- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/src/caps/agent/outcome.rs b/src/caps/agent/outcome.rs index e2a9ebe8..f99a0734 100644 --- a/src/caps/agent/outcome.rs +++ b/src/caps/agent/outcome.rs @@ -130,7 +130,7 @@ pub struct AgentRunOutcome { /// /// **Settled, not live.** These ride the outcome, so they exist only once /// the run is over. Reporting entries *during* a run would need a sink on - /// this capability, which [`AgentRunRequest`] cannot carry — it is + /// this capability, which [`AgentRunRequest`](crate::caps::AgentRunRequest) cannot carry — it is /// `Serialize` + `PartialEq` — so that is a deliberate follow-up rather /// than something to imply here. /// @@ -150,8 +150,8 @@ impl AgentRunOutcome { /// does: `json` is the value when it is an object or array, and `text` is /// the value when it is a string, else its `text` field. /// - /// This is what the default [`AgentRunner::run`] wraps a legacy - /// [`run_agent`](AgentRunner::run_agent) return in, and the shorthand a + /// This is what the default [`AgentRunner::run`](crate::caps::AgentRunner::run) wraps a legacy + /// [`run_agent`](crate::caps::AgentRunner::run_agent) return in, and the shorthand a /// simple adapter wants. /// /// ``` diff --git a/src/transcript.rs b/src/transcript.rs index 98f211d8..bb2912ac 100644 --- a/src/transcript.rs +++ b/src/transcript.rs @@ -16,9 +16,12 @@ //! **Nothing in this crate folds a host's event stream into these** — the //! engine has no event stream of its own, and what counts as one entry is a //! judgement only the harness can make. Hosts fold; the crate carries. -//! [`RunObserver::on_agent_event`](crate::observability::RunObserver::on_agent_event) -//! is how a host reports one *while* the node is still running, rather than -//! waiting for the step to settle. +//! +//! **Settled, not live.** Entries ride the outcome a harness returns, so they +//! reach an observer when the node finishes rather than as they happen. +//! Reporting them during a run would need a sink on the agent capability, and +//! [`AgentRunRequest`](crate::caps::AgentRunRequest) cannot carry one — it is +//! `Serialize` + `PartialEq` — so that is a deliberate follow-up. //! //! Deliberately flat and stringly-typed. Mirroring a host's own event //! vocabulary into the record would make every event kind it adds later a From 3d5f761a6532080886605dea4558f9f1761a2684 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 23 Aug 2026 20:55:13 +0300 Subject: [PATCH 6/9] fix(adaptive): carry the harness transcript through the wire and the ledgers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review was right and my earlier reasoning was wrong. I had left `StepRecord::to_step` returning an empty transcript with a comment claiming the wire form is deliberately lossy — but `Ran::steps` documents itself as "every node activation, at full record fidelity. The per-node transcript: what to archive, and richer than what the judge is shown." Dropping the transcript there contradicts that field's own stated purpose, and because `report` always routes captured steps through `StepRecord::bounded`, every local and remote adaptive run was discarding an agent's transcript before `Ledger::save_steps` ever saw it. `StepRecord` now carries `transcript`; `bounded` copies it and `to_step` preserves it. Both ledgers persist it: SQLite gains a column (with the additive `ALTER TABLE` the file's `MIGRATIONS` list exists for, so an existing ledger keeps working), Mongo gains a key and reads a missing one as empty rather than as corrupt. The record budget deliberately does **not** clip it. `output` is clipped because it is one payload whose tail is the least interesting part; a transcript is many entries already bounded by `TranscriptEntry::bounded`, and cutting it mid-way loses the end of a thought rather than the tail of a value. That is asserted rather than left to the comment. Tests: a transcript round-trips through `bounded`/`to_step`; the budget clips `output` and not the transcript; a legacy record deserializes with none; an empty one still serializes byte-identically; and a ledger conformance case both backends run, which also checks a step that recorded none does not inherit the previous step's. Co-authored-by: Medulla --- crates/adaptive/examples/eval.rs | 1 + crates/adaptive/src/execute/wire.rs | 108 ++++++++++++++++-- crates/adaptive/src/ledger/conformance.rs | 35 ++++++ crates/adaptive/src/ledger/mongo.rs | 11 ++ crates/adaptive/src/ledger/sqlite.rs | 47 +++++--- crates/adaptive/tests/closing.rs | 1 + .../adaptive/tests/continue_after_repair.rs | 1 + crates/adaptive/tests/contracts_surface.rs | 1 + 8 files changed, 176 insertions(+), 29 deletions(-) diff --git a/crates/adaptive/examples/eval.rs b/crates/adaptive/examples/eval.rs index 8b2b9e53..f1e6e039 100644 --- a/crates/adaptive/examples/eval.rs +++ b/crates/adaptive/examples/eval.rs @@ -158,6 +158,7 @@ impl Runner for Simulated { output: json!({ "json": { "exit_code": 0, "stdout": script } }), duration_ms: 1, null_bindings: Vec::new(), + transcript: Vec::new(), }], changed: if worked { "wrote the answer".to_string() diff --git a/crates/adaptive/src/execute/wire.rs b/crates/adaptive/src/execute/wire.rs index 257487dd..577ba674 100644 --- a/crates/adaptive/src/execute/wire.rs +++ b/crates/adaptive/src/execute/wire.rs @@ -50,6 +50,7 @@ use tinyflows::evidence::bounded_within; use tinyflows::expr::NullResolution; use tinyflows::model::WorkflowGraph; use tinyflows::observability::{ExecutionStep, StepStatus}; +use tinyflows::transcript::TranscriptEntry; use super::Ran; @@ -92,6 +93,19 @@ pub struct StepRecord { /// Config expressions that resolved to null during this activation. #[serde(default)] pub null_bindings: Vec, + /// What the harness did inside the node, in order. + /// + /// Carried rather than dropped because [`Ran::steps`](crate::execute::Ran) + /// is the archival record — "every node activation, at full record + /// fidelity" — and for an `agent` node the transcript is most of what there + /// is to know. Unlike `output` it is not clipped to the record budget: + /// entries are already individually bounded by + /// `TranscriptEntry::bounded`, and clipping a transcript in the middle + /// would lose the end of a thought rather than the tail of a payload. + /// + /// Empty for every non-`agent` node and for a harness that reports none. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub transcript: Vec, } impl StepRecord { @@ -107,6 +121,7 @@ impl StepRecord { output: bounded_within(&step.output, budget), duration_ms: u64::try_from(step.duration_ms).unwrap_or(u64::MAX), null_bindings: step.diagnostics.clone(), + transcript: step.transcript.clone(), } } @@ -122,17 +137,7 @@ impl StepRecord { output: self.output.clone(), duration_ms: u128::from(self.duration_ms), diagnostics: self.null_bindings.clone(), - // Deliberately empty: `StepRecord` is a budget-bounded wire form - // and does not transport a harness transcript, which is many - // entries and would dwarf the budget the rest of this type is - // bounded to. The round trip is already lossy by design — `output` - // is clipped and `duration_ms` narrows — so a step reconstructed - // here genuinely has no transcript rather than having lost one. - // - // Nothing downstream of `to_step` reads it: this exists so - // `diagnose` can read a step on the far side, and diagnosis is a - // function of status, output and null bindings. - transcript: Vec::new(), + transcript: self.transcript.clone(), } } } @@ -405,4 +410,85 @@ mod tests { assert_eq!(back.pending_approvals, vec!["publish".to_string()]); assert!((back.cost_usd - 0.42).abs() < f64::EPSILON); } + + /// A harness transcript survives the wire form in both directions. + /// + /// `Ran::steps` is documented as the archival record — "every node + /// activation, at full record fidelity" — so dropping the transcript here + /// would silently empty the richest part of an `agent` node's history on + /// every local and remote adaptive run. + #[test] + fn a_transcript_round_trips_through_the_record() { + let entries = vec![ + TranscriptEntry::bounded(1, "agent_thinking", "memoise the chain"), + TranscriptEntry::bounded(2, "tool_call", "shell: python3 solve.py"), + TranscriptEntry::bounded(3, "tool_result", "837799"), + ]; + let original = ExecutionStep { + transcript: entries.clone(), + ..step("solve", StepStatus::Success, json!([{ "json": 837_799 }])) + }; + + let record = StepRecord::bounded(&original, 4096); + assert_eq!(record.transcript, entries, "the record keeps it"); + + let back = record.to_step(); + assert_eq!(back.transcript, entries, "and hands it back"); + } + + /// The transcript is NOT clipped to the record budget. + /// + /// `output` is, because it is one payload whose tail is the least + /// interesting part. A transcript is many already-bounded entries, and + /// cutting it mid-way loses the end of a thought rather than the tail of a + /// value — so the budget deliberately does not reach it. + #[test] + fn the_record_budget_does_not_clip_the_transcript() { + let entries: Vec = (0..64) + .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(256))) + .collect(); + let original = ExecutionStep { + transcript: entries.clone(), + ..step( + "solve", + StepStatus::Success, + json!([{ "json": "x".repeat(9_000) }]), + ) + }; + + let record = StepRecord::bounded(&original, 128); + assert!( + is_truncated(&record.output), + "the output IS clipped to the budget" + ); + assert_eq!( + record.transcript.len(), + entries.len(), + "the transcript is not" + ); + } + + /// A record written before the field existed still deserializes. + #[test] + fn a_legacy_record_reads_as_having_no_transcript() { + // camelCase, as the type serializes — a legacy record is a real wire + // document, not a snake_case approximation of one. + let legacy = json!({ + "nodeId": "solve", + "status": "success", + "output": [], + "durationMs": 12, + "nullBindings": [], + }); + let record: StepRecord = serde_json::from_value(legacy).expect("deserialize"); + assert!(record.transcript.is_empty()); + } + + /// An empty transcript serializes exactly as it did before the field. + #[test] + fn an_empty_transcript_adds_nothing_to_the_wire() { + let record = StepRecord::bounded(&step("cost", StepStatus::Success, json!([])), 4096); + let wire = serde_json::to_string(&record).expect("serialize"); + assert!(!wire.contains("transcript"), "{wire}"); + } } diff --git a/crates/adaptive/src/ledger/conformance.rs b/crates/adaptive/src/ledger/conformance.rs index 8d8392b5..b72689f1 100644 --- a/crates/adaptive/src/ledger/conformance.rs +++ b/crates/adaptive/src/ledger/conformance.rs @@ -625,6 +625,40 @@ pub async fn run_transcripts(store: &dyn Ledger) { a_looped_node_keeps_every_iteration(store).await; saving_a_transcript_twice_replaces_rather_than_appends(store).await; a_page_windows_the_episode_list(store).await; + an_agent_step_keeps_its_harness_transcript(store).await; +} + +/// An `agent` step's harness transcript survives the ledger. +/// +/// `Ran::steps` is the archival record — "every node activation, at full record +/// fidelity" — so a backend that persists the step but drops what the harness +/// did inside it satisfies the type and loses the point. +async fn an_agent_step_keeps_its_harness_transcript(store: &dyn Ledger) { + use tinyflows::transcript::TranscriptEntry; + + let entries = vec![ + TranscriptEntry::bounded(1, "agent_thinking", "memoise the chain"), + TranscriptEntry::bounded(2, "tool_call", "shell: python3 solve.py"), + TranscriptEntry::bounded(3, "tool_result", "837799"), + ]; + let mut solve = step("solve", 1); + solve.transcript = entries.clone(); + + store + .save_steps("ldg_transcript", &[solve, step("check", 2)]) + .await + .expect("save"); + + let back = store.steps("ldg_transcript").await.expect("steps"); + assert_eq!(back.len(), 2); + assert_eq!( + back[0].transcript, entries, + "the agent node's transcript round-trips whole and in order" + ); + assert!( + back[1].transcript.is_empty(), + "a step that recorded none still reads as none, not as the previous step's" + ); } fn step(node_id: &str, n: u64) -> crate::execute::StepRecord { @@ -634,6 +668,7 @@ fn step(node_id: &str, n: u64) -> crate::execute::StepRecord { output: serde_json::json!({ "i": n }), duration_ms: n, null_bindings: Vec::new(), + transcript: Vec::new(), } } diff --git a/crates/adaptive/src/ledger/mongo.rs b/crates/adaptive/src/ledger/mongo.rs index 0ab4b774..df02b589 100644 --- a/crates/adaptive/src/ledger/mongo.rs +++ b/crates/adaptive/src/ledger/mongo.rs @@ -467,6 +467,8 @@ impl Ledger for MongoLedger { "duration_ms": i64::try_from(step.duration_ms).unwrap_or(i64::MAX), "null_bindings": serde_json::to_string(&step.null_bindings) .map_err(|e| LedgerError::Corrupt(e.to_string()))?, + "transcript": serde_json::to_string(&step.transcript) + .map_err(|e| LedgerError::Corrupt(e.to_string()))?, } }, ) .upsert(true) @@ -496,6 +498,15 @@ impl Ledger for MongoLedger { duration_ms: u64::from(as_u32(&d, "duration_ms")), null_bindings: serde_json::from_str(&text(&d, "null_bindings")) .map_err(|e| LedgerError::Corrupt(e.to_string()))?, + // A document written before this field existed has no + // `transcript` key; `text` yields "" for it, which is not valid + // JSON. Absent means "recorded none", so it reads as empty + // rather than corrupting the whole attempt's steps. + transcript: match text(&d, "transcript").as_str() { + "" => Vec::new(), + raw => serde_json::from_str(raw) + .map_err(|e| LedgerError::Corrupt(e.to_string()))?, + }, }); } Ok(out) diff --git a/crates/adaptive/src/ledger/sqlite.rs b/crates/adaptive/src/ledger/sqlite.rs index 4c9d4cab..b9e7669a 100644 --- a/crates/adaptive/src/ledger/sqlite.rs +++ b/crates/adaptive/src/ledger/sqlite.rs @@ -107,6 +107,7 @@ const DDL: &[&str] = &[ output TEXT NOT NULL, duration_ms INTEGER NOT NULL DEFAULT 0, null_bindings TEXT NOT NULL DEFAULT '[]', + transcript TEXT NOT NULL DEFAULT '[]', PRIMARY KEY (scope_key, row_id, seq) )", ]; @@ -118,6 +119,7 @@ const DDL: &[&str] = &[ /// errors once the column is there, which is the expected case on every start /// after the first — so these are the statements whose failure means success. const MIGRATIONS: &[&str] = &[ + "ALTER TABLE attempt_steps ADD COLUMN transcript TEXT NOT NULL DEFAULT '[]'", "ALTER TABLE lessons ADD COLUMN scope_key TEXT NOT NULL DEFAULT ''", "ALTER TABLE workflow_scores ADD COLUMN scope_key TEXT NOT NULL DEFAULT ''", "ALTER TABLE ledger_rows ADD COLUMN satisfied INTEGER NOT NULL DEFAULT 0", @@ -653,8 +655,9 @@ impl Ledger for SqliteLedger { for (seq, step) in steps.iter().enumerate() { conn.execute( "INSERT OR REPLACE INTO attempt_steps(scope_key, row_id, seq, node_id, status, - output, duration_ms, null_bindings) - VALUES(?1,?2,?3,?4,?5,?6,?7,?8)", + output, duration_ms, null_bindings, + transcript) + VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9)", params![ self.bucket(), row_id, @@ -668,6 +671,8 @@ impl Ledger for SqliteLedger { i64::try_from(step.duration_ms).unwrap_or(i64::MAX), serde_json::to_string(&step.null_bindings) .map_err(|e| LedgerError::Corrupt(e.to_string()))?, + serde_json::to_string(&step.transcript) + .map_err(|e| LedgerError::Corrupt(e.to_string()))?, ], )?; } @@ -677,7 +682,8 @@ impl Ledger for SqliteLedger { async fn steps(&self, row_id: &str) -> Result> { let conn = self.guard()?; let mut stmt = conn.prepare( - "SELECT node_id, status, output, duration_ms, null_bindings FROM attempt_steps + "SELECT node_id, status, output, duration_ms, null_bindings, transcript + FROM attempt_steps WHERE scope_key = ?1 AND row_id = ?2 ORDER BY seq", )?; let found = stmt @@ -688,27 +694,32 @@ impl Ledger for SqliteLedger { r.get::<_, String>(2)?, r.get::<_, i64>(3)?, r.get::<_, String>(4)?, + r.get::<_, String>(5)?, )) })? .collect::>>()?; found .into_iter() - .map(|(node_id, status, output, duration_ms, bindings)| { - Ok(crate::execute::StepRecord { - node_id, - status: if status == "error" { - crate::execute::StepOutcome::Error - } else { - crate::execute::StepOutcome::Success - }, - output: serde_json::from_str(&output) - .map_err(|e| LedgerError::Corrupt(e.to_string()))?, - duration_ms: u64::try_from(duration_ms).unwrap_or(0), - null_bindings: serde_json::from_str(&bindings) - .map_err(|e| LedgerError::Corrupt(e.to_string()))?, - }) - }) + .map( + |(node_id, status, output, duration_ms, bindings, transcript)| { + Ok(crate::execute::StepRecord { + node_id, + status: if status == "error" { + crate::execute::StepOutcome::Error + } else { + crate::execute::StepOutcome::Success + }, + output: serde_json::from_str(&output) + .map_err(|e| LedgerError::Corrupt(e.to_string()))?, + duration_ms: u64::try_from(duration_ms).unwrap_or(0), + null_bindings: serde_json::from_str(&bindings) + .map_err(|e| LedgerError::Corrupt(e.to_string()))?, + transcript: serde_json::from_str(&transcript) + .map_err(|e| LedgerError::Corrupt(e.to_string()))?, + }) + }, + ) .collect() } diff --git a/crates/adaptive/tests/closing.rs b/crates/adaptive/tests/closing.rs index 4d5c4e8f..e8bb8d77 100644 --- a/crates/adaptive/tests/closing.rs +++ b/crates/adaptive/tests/closing.rs @@ -541,6 +541,7 @@ fn step(node: &str, ok: bool) -> tinyflows_adaptive::execute::StepRecord { output: Value::Null, duration_ms: 1, null_bindings: Vec::new(), + transcript: Vec::new(), } } diff --git a/crates/adaptive/tests/continue_after_repair.rs b/crates/adaptive/tests/continue_after_repair.rs index a5c36304..c3da04fb 100644 --- a/crates/adaptive/tests/continue_after_repair.rs +++ b/crates/adaptive/tests/continue_after_repair.rs @@ -297,6 +297,7 @@ impl Runner for Checkpointed { output: slot.clone(), duration_ms: 0, null_bindings: Vec::new(), + transcript: Vec::new(), }) .collect() }) diff --git a/crates/adaptive/tests/contracts_surface.rs b/crates/adaptive/tests/contracts_surface.rs index 2ea244dd..e13de61e 100644 --- a/crates/adaptive/tests/contracts_surface.rs +++ b/crates/adaptive/tests/contracts_surface.rs @@ -104,6 +104,7 @@ fn the_envelope_is_camel_case_and_the_graph_inside_it_is_not() { output: serde_json::json!({"ok": true}), duration_ms: 12, null_bindings: Vec::new(), + transcript: Vec::new(), }], pending_approvals: vec!["publish".into()], cancelled: false, From 9b82b387a07f7d9154d982f327a943e240ee995e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 23 Aug 2026 21:05:43 +0300 Subject: [PATCH 7/9] fix(adaptive): bound a step's transcript in the aggregate, and prove the ordering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two more from review, both correct. **The Mongo document cap.** Carrying the transcript per step reintroduced, one level down, exactly the hazard `save_steps` already reasons about: entries are individually capped at 4 KiB, but a `per_item` node folds every item's turn into ONE step, so a few thousand of them reach the 16 MB a document may hold — and because `save_steps` deletes before it upserts, an oversized write destroys the previous record and *then* fails. Production-only, which is the worst kind. `StepRecord::bounded` now trims to `TRANSCRIPT_BUDGET`, keeping both ends with a marker entry naming how many were elided. Both ends rather than a tail clip: the start says how the agent approached the work and the end says how it concluded, and the conclusion is usually why someone opened the transcript. **The ordering test proved nothing.** It asserted four identical `[agent_thinking, agent_message]` pairs, which a completion-ordered accumulator produces just as happily. Rewritten with a runner that names its item and sleeps inversely to its index, so input order and completion order genuinely disagree. Verified by falsification: with the sink keyed by arrival instead of item index the assertion fails with a fully reversed `["item 3", "item 2", "item 1", "item 0"]`. Co-authored-by: Medulla --- crates/adaptive/src/execute/wire.rs | 121 +++++++++++++++++- .../agent_tests/agent_tests_part_03_tests.rs | 79 ++++++++++-- 2 files changed, 181 insertions(+), 19 deletions(-) diff --git a/crates/adaptive/src/execute/wire.rs b/crates/adaptive/src/execute/wire.rs index 577ba674..f178239c 100644 --- a/crates/adaptive/src/execute/wire.rs +++ b/crates/adaptive/src/execute/wire.rs @@ -57,6 +57,46 @@ use super::Ran; /// Per-node budget for the durable record. Written once; generous. pub const RECORD_BUDGET: usize = 256 * 1024; +/// Aggregate budget for one step's transcript, in bytes of entry text. +/// +/// A per-entry bound is not enough on its own. `TranscriptEntry::bounded` caps +/// one entry at 4 KiB, and a `per_item` node folds every item's turn into ONE +/// step — so a few thousand entries reach the 16 MB limit a Mongo document +/// may hold, in exactly the production-only way the ledger's own note about +/// one-document-per-step warns of. Worse, `save_steps` deletes before it +/// upserts, so an oversized write destroys the previous record and then fails. +pub const TRANSCRIPT_BUDGET: usize = RECORD_BUDGET; + +/// Kept from each end when a transcript is over budget. +const TRANSCRIPT_EDGE: usize = 32; + +/// Trims `entries` to [`TRANSCRIPT_BUDGET`], keeping both ends. +/// +/// Head **and** tail, with a marker between: the start says how the agent +/// approached the work and the end says how it concluded, and the middle is the +/// most droppable part of a long tool loop. Clipping only the tail would lose +/// the conclusion, which is usually the reason someone opened the transcript. +fn bounded_transcript(entries: &[TranscriptEntry]) -> Vec { + let total: usize = entries.iter().map(|e| e.text.len()).sum(); + if total <= TRANSCRIPT_BUDGET { + return entries.to_vec(); + } + if entries.len() <= TRANSCRIPT_EDGE * 2 { + // Few enough entries that dropping the middle would not help; the size + // is in individual entries, which are already individually bounded. + return entries.to_vec(); + } + let dropped = entries.len() - TRANSCRIPT_EDGE * 2; + let mut out: Vec = entries[..TRANSCRIPT_EDGE].to_vec(); + out.push(TranscriptEntry::bounded( + entries[TRANSCRIPT_EDGE].at_ms, + "error", + format!("…[{dropped} transcript entries elided to fit the record budget]"), + )); + out.extend_from_slice(&entries[entries.len() - TRANSCRIPT_EDGE..]); + out +} + /// Per-node budget for what the judge reads. A dozen of these share one context /// window, so it is much smaller than the record. pub const PROMPT_BUDGET: usize = 4 * 1024; @@ -98,10 +138,13 @@ pub struct StepRecord { /// Carried rather than dropped because [`Ran::steps`](crate::execute::Ran) /// is the archival record — "every node activation, at full record /// fidelity" — and for an `agent` node the transcript is most of what there - /// is to know. Unlike `output` it is not clipped to the record budget: - /// entries are already individually bounded by - /// `TranscriptEntry::bounded`, and clipping a transcript in the middle - /// would lose the end of a thought rather than the tail of a payload. + /// is to know. + /// + /// Bounded differently from `output`, not left unbounded: individual + /// entries are already capped, so what matters here is the *aggregate* + /// (see [`TRANSCRIPT_BUDGET`]), and an over-budget transcript keeps both + /// ends rather than being clipped from one — a truncated payload loses its + /// tail, a truncated transcript would lose its conclusion. /// /// Empty for every non-`agent` node and for a harness that reports none. #[serde(default, skip_serializing_if = "Vec::is_empty")] @@ -121,7 +164,7 @@ impl StepRecord { output: bounded_within(&step.output, budget), duration_ms: u64::try_from(step.duration_ms).unwrap_or(u64::MAX), null_bindings: step.diagnostics.clone(), - transcript: step.transcript.clone(), + transcript: bounded_transcript(&step.transcript), } } @@ -491,4 +534,72 @@ mod tests { let wire = serde_json::to_string(&record).expect("serialize"); assert!(!wire.contains("transcript"), "{wire}"); } + + /// A transcript large enough to threaten the Mongo document cap is trimmed. + /// + /// Per-entry bounds are not enough: a `per_item` node folds every item's + /// turn into ONE step, so thousands of 4 KiB entries reach the 16 MB limit + /// a document may hold — and `save_steps` deletes before it upserts, so the + /// oversized write would destroy the previous record and then fail. + #[test] + fn an_oversized_transcript_is_trimmed_to_the_budget() { + let entries: Vec = (0..4_000) + .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(1024))) + .collect(); + let original = ExecutionStep { + transcript: entries, + ..step("solve", StepStatus::Success, json!([])) + }; + + let record = StepRecord::bounded(&original, RECORD_BUDGET); + let bytes: usize = record.transcript.iter().map(|e| e.text.len()).sum(); + assert!( + bytes < TRANSCRIPT_BUDGET, + "trimmed to {bytes} bytes, over the {TRANSCRIPT_BUDGET} budget" + ); + } + + /// Trimming keeps BOTH ends, and says how much it dropped. + /// + /// The start says how the agent approached the work and the end says how it + /// concluded; clipping only the tail would lose the conclusion, which is + /// usually why someone opened the transcript. + #[test] + fn trimming_keeps_the_start_and_the_end() { + let mut entries: Vec = (0..4_000) + .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(1024))) + .collect(); + entries[0] = TranscriptEntry::bounded(0, "agent_thinking", "FIRST"); + let last = entries.len() - 1; + entries[last] = TranscriptEntry::bounded(9_999, "agent_message", "LAST"); + + let original = ExecutionStep { + transcript: entries, + ..step("solve", StepStatus::Success, json!([])) + }; + let kept = StepRecord::bounded(&original, RECORD_BUDGET).transcript; + + assert_eq!(kept.first().map(|e| e.text.as_str()), Some("FIRST")); + assert_eq!(kept.last().map(|e| e.text.as_str()), Some("LAST")); + assert!( + kept.iter().any(|e| e.text.contains("elided")), + "the gap announces itself rather than being silent" + ); + } + + /// A transcript within budget is untouched. + #[test] + fn a_transcript_within_budget_keeps_every_entry() { + let entries: Vec = (0..64) + .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(256))) + .collect(); + let original = ExecutionStep { + transcript: entries.clone(), + ..step("solve", StepStatus::Success, json!([])) + }; + assert_eq!( + StepRecord::bounded(&original, RECORD_BUDGET).transcript, + entries + ); + } } diff --git a/src/nodes/integration/agent_tests/agent_tests_part_03_tests.rs b/src/nodes/integration/agent_tests/agent_tests_part_03_tests.rs index 843d5856..426bd30c 100644 --- a/src/nodes/integration/agent_tests/agent_tests_part_03_tests.rs +++ b/src/nodes/integration/agent_tests/agent_tests_part_03_tests.rs @@ -11,6 +11,7 @@ mod transcript { use crate::model::AgentDefinition; use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; use crate::observability::{NoopObserver, RunObserver}; + use crate::transcript::TranscriptEntry; use serde_json::{Value, json}; use std::sync::Arc; @@ -135,29 +136,79 @@ mod transcript { assert!(out.transcript.is_empty()); } + /// A runner that names its item and finishes in reverse order. + /// + /// Both halves matter. Naming the item is what lets the assertion tell + /// input order from completion order at all — four identical turns cannot. + /// Finishing in reverse is what makes the two orders actually disagree, so + /// a completion-ordered accumulator fails this test rather than passing it + /// by luck. + struct ReverseOrderHarness; + + #[async_trait::async_trait] + impl AgentRunner for ReverseOrderHarness { + async fn run_agent( + &self, + _agent_ref: &str, + _request: Value, + _conn: Option<&str>, + ) -> crate::error::Result { + unreachable!("the typed `run` is overridden") + } + + async fn run( + &self, + request: crate::caps::AgentRunRequest, + ) -> crate::error::Result { + // `=item.seed` is resolved against this item before the request is + // assembled, so the config carries which item this turn is for. + let seed = request + .config + .get("prompt") + .and_then(Value::as_u64) + .unwrap_or(0); + // Later items return first. + let delay = 40u64.saturating_sub(seed * 10); + tokio::time::sleep(std::time::Duration::from_millis(delay)).await; + Ok(crate::caps::AgentRunOutcome::finished(json!({ + "text": format!("answered {seed}") + })) + .with_transcript(vec![TranscriptEntry::bounded( + 0, + "agent_message", + format!("item {seed}"), + )])) + } + } + #[tokio::test] async fn per_item_transcripts_come_back_in_item_order() { // `map_items` restores its OUTPUTS to input order, so the transcript - // describing them has to match. Appending on completion would let a - // concurrent run interleave differently from one run to the next, for - // identical input — the sink is keyed by item index to prevent that. + // describing them has to match. The sink is keyed by item index for + // exactly this: with `concurrency > 1` the turns finish in whatever + // order they finish, and appending on completion would make one run's + // transcript differ from the next for identical input. let out = execute( - harness(), - json!({ "agent_ref": "triager", "execution": "per_item", "concurrency": 4 }), - (0..4).map(|n| Item::new(json!({ "seed": n }))).collect(), + Arc::new(ReverseOrderHarness), + json!({ + "agent_ref": "triager", + "execution": "per_item", + "concurrency": 4, + "prompt": "=item.seed", + }), + (0..4u64).map(|n| Item::new(json!({ "seed": n }))).collect(), &NoopObserver, ) .await; - // MockAgentHarness names the agent in its second entry, and every item - // runs the same agent — so what is asserted here is the *grouping*: - // each turn's pair stays together and the pairs stay in item order. - assert_eq!(out.transcript.len(), 8); - let kinds: Vec<&str> = out.transcript.iter().map(|e| e.kind.as_str()).collect(); + assert_eq!(out.items.len(), 4, "one turn per input item"); assert_eq!( - kinds, - ["agent_thinking", "agent_message"].repeat(4), - "each turn contributes its pair intact, in item order" + out.transcript + .iter() + .map(|e| e.text.as_str()) + .collect::>(), + ["item 0", "item 1", "item 2", "item 3"], + "input order, not completion order" ); } } From 3d4152173ccffc811fd0bbb45df06b105d900a96 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 23 Aug 2026 21:15:51 +0300 Subject: [PATCH 8/9] fix(adaptive): close the transcript bound, and ease the ExecutionStep migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three more from review, all correct. **The bound had a hole I put there.** The first version returned early when a transcript held 64 entries or fewer, on the assumption that few entries meant small ones. `TranscriptEntry`'s fields are public, so nothing forces a harness to build them through `bounded` — four 8 MB entries took the early return untouched. The count check is gone; every entry is re-bounded first, then the middle is dropped until the aggregate fits, so no construction path escapes it. **`kind` was not counted, and my own new test caught it.** Re-bounding through `TranscriptEntry::bounded` caps `text` and not `kind`, so four entries carrying their payload in `kind` still reached 33 MB. It is a discriminator, not a payload, so it is capped at 128 bytes here. **`ExecutionStep` now derives `Default`.** The field addition is still source-breaking for a literal that names every field — that is stated in the PR description rather than hidden — but `..Default::default()` makes the migration one line, which is the same courtesy `AgentRunOutcome::limit_stop`/`paused` extend. Preferred over `#[non_exhaustive]`, which forbids the literal outright and would be worse for a type hosts build in their own tests. `StepStatus` derives `Default` as `Success` to allow it, and the struct's docs carry a doctest showing the pattern. Also moves the wire tests to `wire_tests.rs`, per the repository's rule that Rust tests live in `_tests.rs` files rather than inline in production source. Co-authored-by: Medulla --- crates/adaptive/src/execute/wire.rs | 412 ++++------------------ crates/adaptive/src/execute/wire_tests.rs | 375 ++++++++++++++++++++ src/observability.rs | 27 +- 3 files changed, 473 insertions(+), 341 deletions(-) create mode 100644 crates/adaptive/src/execute/wire_tests.rs diff --git a/crates/adaptive/src/execute/wire.rs b/crates/adaptive/src/execute/wire.rs index f178239c..a48853f2 100644 --- a/crates/adaptive/src/execute/wire.rs +++ b/crates/adaptive/src/execute/wire.rs @@ -70,30 +70,82 @@ pub const TRANSCRIPT_BUDGET: usize = RECORD_BUDGET; /// Kept from each end when a transcript is over budget. const TRANSCRIPT_EDGE: usize = 32; +/// Bytes of one entry's `kind` kept. +/// +/// A kind is a discriminator — `tool_call`, `agent_thinking` — so this is +/// generous for anything meant. `TranscriptEntry::bounded` caps `text` and +/// not this, and the field is public, so without a cap here four entries +/// carrying their payload in `kind` walk straight past the budget. +const TRANSCRIPT_KIND_BYTES: usize = 128; + +/// What one entry costs against [`TRANSCRIPT_BUDGET`]. +/// +/// `kind` counts as well as `text`: it is host-supplied and an open set, so a +/// budget that ignored it could be walked past by a harness that puts its +/// payload there. +fn entry_cost(entry: &TranscriptEntry) -> usize { + entry.kind.len() + entry.text.len() +} + /// Trims `entries` to [`TRANSCRIPT_BUDGET`], keeping both ends. /// -/// Head **and** tail, with a marker between: the start says how the agent -/// approached the work and the end says how it concluded, and the middle is the -/// most droppable part of a long tool loop. Clipping only the tail would lose -/// the conclusion, which is usually the reason someone opened the transcript. +/// Two passes, because there are two ways to be over budget and each needs its +/// own answer: +/// +/// 1. **Every entry is re-bounded.** `TranscriptEntry::bounded` is the +/// per-entry cap, but nothing forces a harness to build its entries through +/// it — the struct's fields are public. So one entry larger than a whole +/// Mongo document can arrive, and no count-based rule would catch it. +/// 2. **Then the middle is dropped** until the aggregate fits. Head *and* +/// tail, with a marker between: the start says how the agent approached the +/// work and the end says how it concluded, and the middle is the most +/// droppable part of a long tool loop. Clipping only the tail would lose the +/// conclusion, which is usually why someone opened the transcript. +/// +/// There is deliberately no early return on a short transcript. Sixty-four +/// entries can be over budget just as four thousand can, and an early return +/// keyed on count was exactly the hole review found in the first version of +/// this function. fn bounded_transcript(entries: &[TranscriptEntry]) -> Vec { - let total: usize = entries.iter().map(|e| e.text.len()).sum(); - if total <= TRANSCRIPT_BUDGET { - return entries.to_vec(); + let mut out: Vec = entries + .iter() + .map(|e| { + let mut kind = e.kind.clone(); + if kind.len() > TRANSCRIPT_KIND_BYTES { + let end = kind + .char_indices() + .map(|(index, _)| index) + .take_while(|index| *index <= TRANSCRIPT_KIND_BYTES) + .last() + .unwrap_or(0); + kind.truncate(end); + } + TranscriptEntry::bounded(e.at_ms, kind, e.text.clone()) + }) + .collect(); + + let total = |v: &[TranscriptEntry]| -> usize { v.iter().map(entry_cost).sum() }; + if total(&out) <= TRANSCRIPT_BUDGET { + return out; + } + + // Drop from the middle outwards, keeping the two edges, until it fits. + while total(&out) > TRANSCRIPT_BUDGET && out.len() > TRANSCRIPT_EDGE * 2 { + out.remove(out.len() / 2); } - if entries.len() <= TRANSCRIPT_EDGE * 2 { - // Few enough entries that dropping the middle would not help; the size - // is in individual entries, which are already individually bounded. - return entries.to_vec(); + + let dropped = entries.len() - out.len(); + if dropped > 0 { + let at_ms = out.get(TRANSCRIPT_EDGE).map_or(0, |e| e.at_ms); + out.insert( + TRANSCRIPT_EDGE.min(out.len()), + TranscriptEntry::bounded( + at_ms, + "error", + format!("…[{dropped} transcript entries elided to fit the record budget]"), + ), + ); } - let dropped = entries.len() - TRANSCRIPT_EDGE * 2; - let mut out: Vec = entries[..TRANSCRIPT_EDGE].to_vec(); - out.push(TranscriptEntry::bounded( - entries[TRANSCRIPT_EDGE].at_ms, - "error", - format!("…[{dropped} transcript entries elided to fit the record budget]"), - )); - out.extend_from_slice(&entries[entries.len() - TRANSCRIPT_EDGE..]); out } @@ -283,323 +335,5 @@ impl RunReport { } #[cfg(test)] -mod tests { - use super::*; - use tinyflows::evidence::is_truncated; - - fn step(node_id: &str, status: StepStatus, output: Value) -> ExecutionStep { - ExecutionStep { - node_id: node_id.into(), - status, - output, - duration_ms: 12, - diagnostics: Vec::new(), - transcript: Vec::new(), - } - } - - fn graph() -> WorkflowGraph { - WorkflowGraph { - schema_version: 1, - id: Some("g".into()), - name: "g".into(), - inputs: Vec::new(), - agents: Vec::new(), - nodes: Vec::new(), - edges: Vec::new(), - } - } - - #[test] - fn one_fat_node_does_not_take_the_rest_of_the_record_with_it() { - // The whole reason bounding is per node. `bounded_within` is - // non-recursive: applied to the aggregate, the big one would replace - // every other node's output with a string preview. - let big = json!({ "body": "x".repeat(600 * 1024) }); - let report = RunReport { - steps: vec![ - StepRecord::bounded( - &step("small", StepStatus::Success, json!({"ok": 1})), - RECORD_BUDGET, - ), - StepRecord::bounded(&step("huge", StepStatus::Success, big), RECORD_BUDGET), - ], - ..RunReport::default() - }; - - assert!( - !is_truncated(&report.steps[0].output), - "the small node is intact" - ); - assert!( - is_truncated(&report.steps[1].output), - "the big one is trimmed" - ); - assert_eq!(report.steps[0].output, json!({"ok": 1})); - } - - #[test] - fn a_swallowed_error_survives_the_round_trip() { - // `output` alone cannot express this, which is why steps cross. - let record = StepRecord::bounded( - &step( - "fetch", - StepStatus::Error, - json!({"error": "connection refused"}), - ), - RECORD_BUDGET, - ); - let json = serde_json::to_string(&record).expect("serializes"); - let back: StepRecord = serde_json::from_str(&json).expect("deserializes"); - assert_eq!(back.status, StepOutcome::Error); - assert!(matches!(back.to_step().status, StepStatus::Error)); - } - - #[test] - fn every_iteration_of_a_looped_node_is_kept() { - let report = RunReport { - steps: vec![ - StepRecord::bounded( - &step("body", StepStatus::Success, json!({"i": 1})), - RECORD_BUDGET, - ), - StepRecord::bounded( - &step("body", StepStatus::Success, json!({"i": 2})), - RECORD_BUDGET, - ), - StepRecord::bounded( - &step("body", StepStatus::Success, json!({"i": 3})), - RECORD_BUDGET, - ), - ], - ..RunReport::default() - }; - assert_eq!(report.steps.len(), 3); - - // The reconstructed final state keeps only the last, as the engine's own - // does — the history lives on `steps`. - let ran = report.into_ran(&graph()); - assert_eq!(ran.outcome.output["nodes"]["body"], json!({"i": 3})); - assert_eq!(ran.steps.len(), 3); - } - - #[test] - fn the_judges_view_is_bounded_tighter_than_the_record() { - let body = json!({ "body": "x".repeat(64 * 1024) }); - let report = RunReport { - steps: vec![StepRecord::bounded( - &step("agent", StepStatus::Success, body), - RECORD_BUDGET, - )], - ..RunReport::default() - }; - // Well under the record budget, so kept whole there... - assert!(!is_truncated(&report.steps[0].output)); - - let ran = report.into_ran(&graph()); - // ...and trimmed in the projection the model reads. - assert!(is_truncated(&ran.outcome.output["nodes"]["agent"])); - assert!( - !is_truncated(&ran.steps[0].output), - "the record is untouched" - ); - } - - #[test] - fn a_failed_run_still_carries_every_step_it_managed() { - // The case `output` cannot express at all: the engine returned Err, so - // there is no outcome, but eleven steps happened. - let report = RunReport { - steps: (0..11) - .map(|i| { - StepRecord::bounded( - &step("loop", StepStatus::Success, json!({ "i": i })), - RECORD_BUDGET, - ) - }) - .collect(), - failed: Some("loop node exceeded its maximum of 5 iterations".into()), - ..RunReport::default() - }; - let ran = report.into_ran(&graph()); - assert_eq!(ran.steps.len(), 11); - assert_eq!( - ran.outcome.output["error"], - json!("loop node exceeded its maximum of 5 iterations") - ); - // And the nodes are there too, so the judge sees what did happen rather - // than only that something broke. - assert!(ran.outcome.output["nodes"]["loop"].is_object()); - } - - #[test] - fn the_whole_report_round_trips_as_json() { - let report = RunReport { - attempt_id: "ep-1/3".into(), - steps: vec![StepRecord::bounded( - &step("write", StepStatus::Success, json!({"path": "report.md"})), - RECORD_BUDGET, - )], - pending_approvals: vec!["publish".into()], - cancelled: false, - changed: "1 file changed".into(), - failed: None, - cost_usd: 0.42, - }; - let text = serde_json::to_string(&report).expect("serializes"); - assert!(text.contains("attemptId"), "camelCase on the wire: {text}"); - let back: RunReport = serde_json::from_str(&text).expect("deserializes"); - assert_eq!(back.attempt_id, "ep-1/3"); - assert_eq!(back.pending_approvals, vec!["publish".to_string()]); - assert!((back.cost_usd - 0.42).abs() < f64::EPSILON); - } - - /// A harness transcript survives the wire form in both directions. - /// - /// `Ran::steps` is documented as the archival record — "every node - /// activation, at full record fidelity" — so dropping the transcript here - /// would silently empty the richest part of an `agent` node's history on - /// every local and remote adaptive run. - #[test] - fn a_transcript_round_trips_through_the_record() { - let entries = vec![ - TranscriptEntry::bounded(1, "agent_thinking", "memoise the chain"), - TranscriptEntry::bounded(2, "tool_call", "shell: python3 solve.py"), - TranscriptEntry::bounded(3, "tool_result", "837799"), - ]; - let original = ExecutionStep { - transcript: entries.clone(), - ..step("solve", StepStatus::Success, json!([{ "json": 837_799 }])) - }; - - let record = StepRecord::bounded(&original, 4096); - assert_eq!(record.transcript, entries, "the record keeps it"); - - let back = record.to_step(); - assert_eq!(back.transcript, entries, "and hands it back"); - } - - /// The transcript is NOT clipped to the record budget. - /// - /// `output` is, because it is one payload whose tail is the least - /// interesting part. A transcript is many already-bounded entries, and - /// cutting it mid-way loses the end of a thought rather than the tail of a - /// value — so the budget deliberately does not reach it. - #[test] - fn the_record_budget_does_not_clip_the_transcript() { - let entries: Vec = (0..64) - .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(256))) - .collect(); - let original = ExecutionStep { - transcript: entries.clone(), - ..step( - "solve", - StepStatus::Success, - json!([{ "json": "x".repeat(9_000) }]), - ) - }; - - let record = StepRecord::bounded(&original, 128); - assert!( - is_truncated(&record.output), - "the output IS clipped to the budget" - ); - assert_eq!( - record.transcript.len(), - entries.len(), - "the transcript is not" - ); - } - - /// A record written before the field existed still deserializes. - #[test] - fn a_legacy_record_reads_as_having_no_transcript() { - // camelCase, as the type serializes — a legacy record is a real wire - // document, not a snake_case approximation of one. - let legacy = json!({ - "nodeId": "solve", - "status": "success", - "output": [], - "durationMs": 12, - "nullBindings": [], - }); - let record: StepRecord = serde_json::from_value(legacy).expect("deserialize"); - assert!(record.transcript.is_empty()); - } - - /// An empty transcript serializes exactly as it did before the field. - #[test] - fn an_empty_transcript_adds_nothing_to_the_wire() { - let record = StepRecord::bounded(&step("cost", StepStatus::Success, json!([])), 4096); - let wire = serde_json::to_string(&record).expect("serialize"); - assert!(!wire.contains("transcript"), "{wire}"); - } - - /// A transcript large enough to threaten the Mongo document cap is trimmed. - /// - /// Per-entry bounds are not enough: a `per_item` node folds every item's - /// turn into ONE step, so thousands of 4 KiB entries reach the 16 MB limit - /// a document may hold — and `save_steps` deletes before it upserts, so the - /// oversized write would destroy the previous record and then fail. - #[test] - fn an_oversized_transcript_is_trimmed_to_the_budget() { - let entries: Vec = (0..4_000) - .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(1024))) - .collect(); - let original = ExecutionStep { - transcript: entries, - ..step("solve", StepStatus::Success, json!([])) - }; - - let record = StepRecord::bounded(&original, RECORD_BUDGET); - let bytes: usize = record.transcript.iter().map(|e| e.text.len()).sum(); - assert!( - bytes < TRANSCRIPT_BUDGET, - "trimmed to {bytes} bytes, over the {TRANSCRIPT_BUDGET} budget" - ); - } - - /// Trimming keeps BOTH ends, and says how much it dropped. - /// - /// The start says how the agent approached the work and the end says how it - /// concluded; clipping only the tail would lose the conclusion, which is - /// usually why someone opened the transcript. - #[test] - fn trimming_keeps_the_start_and_the_end() { - let mut entries: Vec = (0..4_000) - .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(1024))) - .collect(); - entries[0] = TranscriptEntry::bounded(0, "agent_thinking", "FIRST"); - let last = entries.len() - 1; - entries[last] = TranscriptEntry::bounded(9_999, "agent_message", "LAST"); - - let original = ExecutionStep { - transcript: entries, - ..step("solve", StepStatus::Success, json!([])) - }; - let kept = StepRecord::bounded(&original, RECORD_BUDGET).transcript; - - assert_eq!(kept.first().map(|e| e.text.as_str()), Some("FIRST")); - assert_eq!(kept.last().map(|e| e.text.as_str()), Some("LAST")); - assert!( - kept.iter().any(|e| e.text.contains("elided")), - "the gap announces itself rather than being silent" - ); - } - - /// A transcript within budget is untouched. - #[test] - fn a_transcript_within_budget_keeps_every_entry() { - let entries: Vec = (0..64) - .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(256))) - .collect(); - let original = ExecutionStep { - transcript: entries.clone(), - ..step("solve", StepStatus::Success, json!([])) - }; - assert_eq!( - StepRecord::bounded(&original, RECORD_BUDGET).transcript, - entries - ); - } -} +#[path = "wire_tests.rs"] +mod tests; diff --git a/crates/adaptive/src/execute/wire_tests.rs b/crates/adaptive/src/execute/wire_tests.rs new file mode 100644 index 00000000..1970fb68 --- /dev/null +++ b/crates/adaptive/src/execute/wire_tests.rs @@ -0,0 +1,375 @@ +//! Tests for [`super`] — the adaptive execution wire form. +//! +//! In their own file per the repository's rule that Rust tests live in +//! `_tests.rs` files rather than as inline modules in production source. + +use super::*; +use tinyflows::evidence::is_truncated; + +fn step(node_id: &str, status: StepStatus, output: Value) -> ExecutionStep { + ExecutionStep { + node_id: node_id.into(), + status, + output, + duration_ms: 12, + diagnostics: Vec::new(), + transcript: Vec::new(), + } +} + +fn graph() -> WorkflowGraph { + WorkflowGraph { + schema_version: 1, + id: Some("g".into()), + name: "g".into(), + inputs: Vec::new(), + agents: Vec::new(), + nodes: Vec::new(), + edges: Vec::new(), + } +} + +#[test] +fn one_fat_node_does_not_take_the_rest_of_the_record_with_it() { + // The whole reason bounding is per node. `bounded_within` is + // non-recursive: applied to the aggregate, the big one would replace + // every other node's output with a string preview. + let big = json!({ "body": "x".repeat(600 * 1024) }); + let report = RunReport { + steps: vec![ + StepRecord::bounded( + &step("small", StepStatus::Success, json!({"ok": 1})), + RECORD_BUDGET, + ), + StepRecord::bounded(&step("huge", StepStatus::Success, big), RECORD_BUDGET), + ], + ..RunReport::default() + }; + + assert!( + !is_truncated(&report.steps[0].output), + "the small node is intact" + ); + assert!( + is_truncated(&report.steps[1].output), + "the big one is trimmed" + ); + assert_eq!(report.steps[0].output, json!({"ok": 1})); +} + +#[test] +fn a_swallowed_error_survives_the_round_trip() { + // `output` alone cannot express this, which is why steps cross. + let record = StepRecord::bounded( + &step( + "fetch", + StepStatus::Error, + json!({"error": "connection refused"}), + ), + RECORD_BUDGET, + ); + let json = serde_json::to_string(&record).expect("serializes"); + let back: StepRecord = serde_json::from_str(&json).expect("deserializes"); + assert_eq!(back.status, StepOutcome::Error); + assert!(matches!(back.to_step().status, StepStatus::Error)); +} + +#[test] +fn every_iteration_of_a_looped_node_is_kept() { + let report = RunReport { + steps: vec![ + StepRecord::bounded( + &step("body", StepStatus::Success, json!({"i": 1})), + RECORD_BUDGET, + ), + StepRecord::bounded( + &step("body", StepStatus::Success, json!({"i": 2})), + RECORD_BUDGET, + ), + StepRecord::bounded( + &step("body", StepStatus::Success, json!({"i": 3})), + RECORD_BUDGET, + ), + ], + ..RunReport::default() + }; + assert_eq!(report.steps.len(), 3); + + // The reconstructed final state keeps only the last, as the engine's own + // does — the history lives on `steps`. + let ran = report.into_ran(&graph()); + assert_eq!(ran.outcome.output["nodes"]["body"], json!({"i": 3})); + assert_eq!(ran.steps.len(), 3); +} + +#[test] +fn the_judges_view_is_bounded_tighter_than_the_record() { + let body = json!({ "body": "x".repeat(64 * 1024) }); + let report = RunReport { + steps: vec![StepRecord::bounded( + &step("agent", StepStatus::Success, body), + RECORD_BUDGET, + )], + ..RunReport::default() + }; + // Well under the record budget, so kept whole there... + assert!(!is_truncated(&report.steps[0].output)); + + let ran = report.into_ran(&graph()); + // ...and trimmed in the projection the model reads. + assert!(is_truncated(&ran.outcome.output["nodes"]["agent"])); + assert!( + !is_truncated(&ran.steps[0].output), + "the record is untouched" + ); +} + +#[test] +fn a_failed_run_still_carries_every_step_it_managed() { + // The case `output` cannot express at all: the engine returned Err, so + // there is no outcome, but eleven steps happened. + let report = RunReport { + steps: (0..11) + .map(|i| { + StepRecord::bounded( + &step("loop", StepStatus::Success, json!({ "i": i })), + RECORD_BUDGET, + ) + }) + .collect(), + failed: Some("loop node exceeded its maximum of 5 iterations".into()), + ..RunReport::default() + }; + let ran = report.into_ran(&graph()); + assert_eq!(ran.steps.len(), 11); + assert_eq!( + ran.outcome.output["error"], + json!("loop node exceeded its maximum of 5 iterations") + ); + // And the nodes are there too, so the judge sees what did happen rather + // than only that something broke. + assert!(ran.outcome.output["nodes"]["loop"].is_object()); +} + +#[test] +fn the_whole_report_round_trips_as_json() { + let report = RunReport { + attempt_id: "ep-1/3".into(), + steps: vec![StepRecord::bounded( + &step("write", StepStatus::Success, json!({"path": "report.md"})), + RECORD_BUDGET, + )], + pending_approvals: vec!["publish".into()], + cancelled: false, + changed: "1 file changed".into(), + failed: None, + cost_usd: 0.42, + }; + let text = serde_json::to_string(&report).expect("serializes"); + assert!(text.contains("attemptId"), "camelCase on the wire: {text}"); + let back: RunReport = serde_json::from_str(&text).expect("deserializes"); + assert_eq!(back.attempt_id, "ep-1/3"); + assert_eq!(back.pending_approvals, vec!["publish".to_string()]); + assert!((back.cost_usd - 0.42).abs() < f64::EPSILON); +} + +/// A harness transcript survives the wire form in both directions. +/// +/// `Ran::steps` is documented as the archival record — "every node +/// activation, at full record fidelity" — so dropping the transcript here +/// would silently empty the richest part of an `agent` node's history on +/// every local and remote adaptive run. +#[test] +fn a_transcript_round_trips_through_the_record() { + let entries = vec![ + TranscriptEntry::bounded(1, "agent_thinking", "memoise the chain"), + TranscriptEntry::bounded(2, "tool_call", "shell: python3 solve.py"), + TranscriptEntry::bounded(3, "tool_result", "837799"), + ]; + let original = ExecutionStep { + transcript: entries.clone(), + ..step("solve", StepStatus::Success, json!([{ "json": 837_799 }])) + }; + + let record = StepRecord::bounded(&original, 4096); + assert_eq!(record.transcript, entries, "the record keeps it"); + + let back = record.to_step(); + assert_eq!(back.transcript, entries, "and hands it back"); +} + +/// The transcript is NOT clipped to the record budget. +/// +/// `output` is, because it is one payload whose tail is the least +/// interesting part. A transcript is many already-bounded entries, and +/// cutting it mid-way loses the end of a thought rather than the tail of a +/// value — so the budget deliberately does not reach it. +#[test] +fn the_record_budget_does_not_clip_the_transcript() { + let entries: Vec = (0..64) + .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(256))) + .collect(); + let original = ExecutionStep { + transcript: entries.clone(), + ..step( + "solve", + StepStatus::Success, + json!([{ "json": "x".repeat(9_000) }]), + ) + }; + + let record = StepRecord::bounded(&original, 128); + assert!( + is_truncated(&record.output), + "the output IS clipped to the budget" + ); + assert_eq!( + record.transcript.len(), + entries.len(), + "the transcript is not" + ); +} + +/// A record written before the field existed still deserializes. +#[test] +fn a_legacy_record_reads_as_having_no_transcript() { + // camelCase, as the type serializes — a legacy record is a real wire + // document, not a snake_case approximation of one. + let legacy = json!({ + "nodeId": "solve", + "status": "success", + "output": [], + "durationMs": 12, + "nullBindings": [], + }); + let record: StepRecord = serde_json::from_value(legacy).expect("deserialize"); + assert!(record.transcript.is_empty()); +} + +/// An empty transcript serializes exactly as it did before the field. +#[test] +fn an_empty_transcript_adds_nothing_to_the_wire() { + let record = StepRecord::bounded(&step("cost", StepStatus::Success, json!([])), 4096); + let wire = serde_json::to_string(&record).expect("serialize"); + assert!(!wire.contains("transcript"), "{wire}"); +} + +/// A transcript large enough to threaten the Mongo document cap is trimmed. +/// +/// Per-entry bounds are not enough: a `per_item` node folds every item's +/// turn into ONE step, so thousands of 4 KiB entries reach the 16 MB limit +/// a document may hold — and `save_steps` deletes before it upserts, so the +/// oversized write would destroy the previous record and then fail. +#[test] +fn an_oversized_transcript_is_trimmed_to_the_budget() { + let entries: Vec = (0..4_000) + .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(1024))) + .collect(); + let original = ExecutionStep { + transcript: entries, + ..step("solve", StepStatus::Success, json!([])) + }; + + let record = StepRecord::bounded(&original, RECORD_BUDGET); + let bytes: usize = record.transcript.iter().map(|e| e.text.len()).sum(); + assert!( + bytes < TRANSCRIPT_BUDGET, + "trimmed to {bytes} bytes, over the {TRANSCRIPT_BUDGET} budget" + ); +} + +/// Trimming keeps BOTH ends, and says how much it dropped. +/// +/// The start says how the agent approached the work and the end says how it +/// concluded; clipping only the tail would lose the conclusion, which is +/// usually why someone opened the transcript. +#[test] +fn trimming_keeps_the_start_and_the_end() { + let mut entries: Vec = (0..4_000) + .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(1024))) + .collect(); + entries[0] = TranscriptEntry::bounded(0, "agent_thinking", "FIRST"); + let last = entries.len() - 1; + entries[last] = TranscriptEntry::bounded(9_999, "agent_message", "LAST"); + + let original = ExecutionStep { + transcript: entries, + ..step("solve", StepStatus::Success, json!([])) + }; + let kept = StepRecord::bounded(&original, RECORD_BUDGET).transcript; + + assert_eq!(kept.first().map(|e| e.text.as_str()), Some("FIRST")); + assert_eq!(kept.last().map(|e| e.text.as_str()), Some("LAST")); + assert!( + kept.iter().any(|e| e.text.contains("elided")), + "the gap announces itself rather than being silent" + ); +} + +/// A transcript within budget is untouched. +#[test] +fn a_transcript_within_budget_keeps_every_entry() { + let entries: Vec = (0..64) + .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(256))) + .collect(); + let original = ExecutionStep { + transcript: entries.clone(), + ..step("solve", StepStatus::Success, json!([])) + }; + assert_eq!( + StepRecord::bounded(&original, RECORD_BUDGET).transcript, + entries + ); +} + +/// A handful of huge entries is bounded too, not just a long list. +/// +/// `TranscriptEntry`'s fields are public, so nothing forces a harness to +/// build them through `bounded` — one entry larger than a whole Mongo +/// document can arrive. A count-based rule would wave this through, which +/// is exactly the hole the first version of the bound had. +#[test] +fn a_few_oversized_entries_are_bounded_too() { + let entries: Vec = (0..4) + .map(|n| TranscriptEntry { + at_ms: n, + kind: "agent_thinking".to_string(), + // Built directly, past the per-entry cap. + text: "x".repeat(8 * 1024 * 1024), + }) + .collect(); + let original = ExecutionStep { + transcript: entries, + ..step("solve", StepStatus::Success, json!([])) + }; + + let kept = StepRecord::bounded(&original, RECORD_BUDGET).transcript; + let bytes: usize = kept.iter().map(|e| e.kind.len() + e.text.len()).sum(); + assert!( + bytes <= TRANSCRIPT_BUDGET, + "four 8 MB entries survived as {bytes} bytes" + ); +} + +/// `kind` counts against the budget as well as `text`. +/// +/// It is host-supplied and an open set, so a budget that ignored it could +/// be walked past by a harness that puts its payload there. +#[test] +fn the_budget_counts_the_kind_as_well_as_the_text() { + let entries: Vec = (0..4) + .map(|n| TranscriptEntry { + at_ms: n, + kind: "k".repeat(8 * 1024 * 1024), + text: String::new(), + }) + .collect(); + let original = ExecutionStep { + transcript: entries, + ..step("solve", StepStatus::Success, json!([])) + }; + + let kept = StepRecord::bounded(&original, RECORD_BUDGET).transcript; + let bytes: usize = kept.iter().map(|e| e.kind.len() + e.text.len()).sum(); + assert!(bytes <= TRANSCRIPT_BUDGET, "kind was not counted: {bytes}"); +} diff --git a/src/observability.rs b/src/observability.rs index b1c6ae67..acf47781 100644 --- a/src/observability.rs +++ b/src/observability.rs @@ -64,9 +64,13 @@ pub enum RunStatus { } /// The outcome of a single [`ExecutionStep`]. -#[derive(Debug, Clone)] +/// +/// [`Success`](Self::Success) is the default so [`ExecutionStep`] can derive +/// one — see the note there. +#[derive(Debug, Clone, Default)] pub enum StepStatus { /// The node executed and produced output items. + #[default] Success, /// The node's executor errored (after any retries were exhausted). Error, @@ -77,7 +81,26 @@ pub enum StepStatus { /// /// This is the record the canvas renders when a user inspects a node, and what a /// run-history view summarizes. -#[derive(Debug, Clone)] +/// # Constructing one +/// +/// `Default` is derived so a host can build a step without naming every field: +/// +/// ``` +/// use tinyflows::observability::{ExecutionStep, StepStatus}; +/// +/// let step = ExecutionStep { +/// node_id: "solve".to_string(), +/// status: StepStatus::Error, +/// ..Default::default() +/// }; +/// assert!(step.transcript.is_empty()); +/// ``` +/// +/// That is the migration path when this struct gains a field: a literal that +/// names every one breaks, and `..Default::default()` does not. Preferred over +/// `#[non_exhaustive]`, which would forbid the literal outright — worse for a +/// type hosts are meant to build in their own tests. +#[derive(Debug, Clone, Default)] pub struct ExecutionStep { /// The id of the node this step ran. pub node_id: String, From e4c1c04f231220634a1fd0c63181645ea73c7fd4 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 23 Aug 2026 21:24:02 +0300 Subject: [PATCH 9/9] perf(adaptive): trim an oversized transcript in one pass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The trim measured the whole vector on every iteration and removed from its middle, shifting the suffix each time. Quadratic — and on work that happens *after* the agent has finished, while a report is waiting to go out, so the cost lands where nothing is left to absorb it. Now one pass from each end: walk the head until half the budget is spent, walk the tail until the other half is, and build the result once. Half from each end so a transcript that is huge at one end cannot starve the other. Regression test trims 80,000 entries under a deliberately generous ceiling — the point is to catch a return to quadratic, not to benchmark. Co-authored-by: Medulla --- crates/adaptive/src/execute/wire.rs | 64 ++++++++++++++++------- crates/adaptive/src/execute/wire_tests.rs | 29 ++++++++++ 2 files changed, 75 insertions(+), 18 deletions(-) diff --git a/crates/adaptive/src/execute/wire.rs b/crates/adaptive/src/execute/wire.rs index a48853f2..74f3f4a4 100644 --- a/crates/adaptive/src/execute/wire.rs +++ b/crates/adaptive/src/execute/wire.rs @@ -107,7 +107,7 @@ fn entry_cost(entry: &TranscriptEntry) -> usize { /// keyed on count was exactly the hole review found in the first version of /// this function. fn bounded_transcript(entries: &[TranscriptEntry]) -> Vec { - let mut out: Vec = entries + let out: Vec = entries .iter() .map(|e| { let mut kind = e.kind.clone(); @@ -124,29 +124,57 @@ fn bounded_transcript(entries: &[TranscriptEntry]) -> Vec { }) .collect(); - let total = |v: &[TranscriptEntry]| -> usize { v.iter().map(entry_cost).sum() }; - if total(&out) <= TRANSCRIPT_BUDGET { + let total: usize = out.iter().map(entry_cost).sum(); + if total <= TRANSCRIPT_BUDGET { return out; } - // Drop from the middle outwards, keeping the two edges, until it fits. - while total(&out) > TRANSCRIPT_BUDGET && out.len() > TRANSCRIPT_EDGE * 2 { - out.remove(out.len() / 2); + // One pass from each end, never a rescan. Walking the vector and removing + // from its middle re-measures the whole thing per iteration and shifts the + // suffix each time — quadratic, on work that happens after the agent has + // finished and while a report is waiting to go out. + // + // Half the budget from each end, so a transcript that is huge at one end + // cannot starve the other. + let half = TRANSCRIPT_BUDGET / 2; + + let mut head = 0usize; + let mut spent = 0usize; + while head < out.len() && head < TRANSCRIPT_EDGE { + let cost = entry_cost(&out[head]); + if spent + cost > half { + break; + } + spent += cost; + head += 1; } - let dropped = entries.len() - out.len(); - if dropped > 0 { - let at_ms = out.get(TRANSCRIPT_EDGE).map_or(0, |e| e.at_ms); - out.insert( - TRANSCRIPT_EDGE.min(out.len()), - TranscriptEntry::bounded( - at_ms, - "error", - format!("…[{dropped} transcript entries elided to fit the record budget]"), - ), - ); + let mut tail = 0usize; + spent = 0; + while tail < out.len() - head && tail < TRANSCRIPT_EDGE { + let cost = entry_cost(&out[out.len() - 1 - tail]); + if spent + cost > half { + break; + } + spent += cost; + tail += 1; } - out + + let dropped = out.len() - head - tail; + if dropped == 0 { + return out; + } + + let at_ms = out.get(head).map_or(0, |e| e.at_ms); + let mut trimmed: Vec = Vec::with_capacity(head + tail + 1); + trimmed.extend_from_slice(&out[..head]); + trimmed.push(TranscriptEntry::bounded( + at_ms, + "error", + format!("…[{dropped} transcript entries elided to fit the record budget]"), + )); + trimmed.extend_from_slice(&out[out.len() - tail..]); + trimmed } /// Per-node budget for what the judge reads. A dozen of these share one context diff --git a/crates/adaptive/src/execute/wire_tests.rs b/crates/adaptive/src/execute/wire_tests.rs index 1970fb68..bad3add7 100644 --- a/crates/adaptive/src/execute/wire_tests.rs +++ b/crates/adaptive/src/execute/wire_tests.rs @@ -373,3 +373,32 @@ fn the_budget_counts_the_kind_as_well_as_the_text() { let bytes: usize = kept.iter().map(|e| e.kind.len() + e.text.len()).sum(); assert!(bytes <= TRANSCRIPT_BUDGET, "kind was not counted: {bytes}"); } + +/// Trimming a very large transcript stays fast. +/// +/// The first version rescanned the whole vector per iteration and removed from +/// its middle — quadratic, on work that runs *after* the agent has finished and +/// while a report is waiting to go out. A generous ceiling: the point is to +/// catch a return to quadratic, not to benchmark. +#[test] +fn trimming_a_huge_transcript_is_not_quadratic() { + let entries: Vec = (0..80_000) + .map(|n| TranscriptEntry::bounded(n, "agent_thinking", "x".repeat(64))) + .collect(); + let original = ExecutionStep { + transcript: entries, + ..step("solve", StepStatus::Success, json!([])) + }; + + let started = std::time::Instant::now(); + let kept = StepRecord::bounded(&original, RECORD_BUDGET).transcript; + let elapsed = started.elapsed(); + + assert!( + elapsed < std::time::Duration::from_secs(5), + "took {elapsed:?} — the trim has gone quadratic again" + ); + let bytes: usize = kept.iter().map(|e| e.kind.len() + e.text.len()).sum(); + assert!(bytes <= TRANSCRIPT_BUDGET); + assert!(kept.iter().any(|e| e.text.contains("elided"))); +}