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 4db60f76..74f3f4a4 100644 --- a/crates/adaptive/src/execute/wire.rs +++ b/crates/adaptive/src/execute/wire.rs @@ -50,12 +50,133 @@ 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; /// 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; + +/// 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. +/// +/// 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 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: usize = out.iter().map(entry_cost).sum(); + if total <= TRANSCRIPT_BUDGET { + return out; + } + + // 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 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; + } + + 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 /// window, so it is much smaller than the record. pub const PROMPT_BUDGET: usize = 4 * 1024; @@ -92,6 +213,22 @@ 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. + /// + /// 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")] + pub transcript: Vec, } impl StepRecord { @@ -107,6 +244,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: bounded_transcript(&step.transcript), } } @@ -122,6 +260,7 @@ impl StepRecord { output: self.output.clone(), duration_ms: u128::from(self.duration_ms), diagnostics: self.null_bindings.clone(), + transcript: self.transcript.clone(), } } } @@ -224,173 +363,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(), - } - } - - 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); - } -} +#[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..bad3add7 --- /dev/null +++ b/crates/adaptive/src/execute/wire_tests.rs @@ -0,0 +1,404 @@ +//! 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}"); +} + +/// 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"))); +} 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, diff --git a/src/caps/agent.rs b/src/caps/agent.rs index 377c04de..37e05825 100644 --- a/src/caps/agent.rs +++ b/src/caps/agent.rs @@ -278,159 +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, -} - -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, - } - } - - /// 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..f99a0734 --- /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`](crate::caps::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`](crate::caps::AgentRunner::run) wraps a legacy + /// [`run_agent`](crate::caps::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/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/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/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..fee76741 100644 --- a/src/nodes/integration/agent.rs +++ b/src/nodes/integration/agent.rs @@ -1,5 +1,8 @@ //! 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; use serde_json::Value; @@ -7,6 +10,55 @@ 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. +/// +/// 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>>>; + +/// Keeps `entries` for the settled step, under the item they belong to. +/// +/// 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; + } + // 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.entry(item_index).or_default().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, in item order, leaving it empty. +fn drain(sink: &TranscriptSink) -> Vec { + sink.lock() + .map(|mut held| std::mem::take(&mut *held).into_values().flatten().collect()) + .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 +132,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(BTreeMap::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 +156,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 +180,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 +196,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 +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).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 @@ -300,9 +368,16 @@ async fn finish_agent_run( conn: Option<&str>, 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, item_index, 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.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_03_tests.rs b/src/nodes/integration/agent_tests/agent_tests_part_03_tests.rs new file mode 100644 index 00000000..426bd30c --- /dev/null +++ b/src/nodes/integration/agent_tests/agent_tests_part_03_tests.rs @@ -0,0 +1,214 @@ +// ---- 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 crate::transcript::TranscriptEntry; + 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()); + } + + /// 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. 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( + 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; + + assert_eq!(out.items.len(), 4, "one turn per input item"); + assert_eq!( + out.transcript + .iter() + .map(|e| e.text.as_str()) + .collect::>(), + ["item 0", "item 1", "item 2", "item 3"], + "input order, not completion order" + ); + } +} diff --git a/src/observability.rs b/src/observability.rs index 8a1d557b..acf47781 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"]); //! ``` @@ -63,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, @@ -76,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, @@ -92,6 +116,26 @@ 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, + /// 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, } /// One execution of a workflow, captured as an ordered list of [`ExecutionStep`]s. diff --git a/src/observability_tests.rs b/src/observability_tests.rs index e9b480c0..b5c0b389 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,44 @@ fn observer_is_usable_as_trait_object() { output: serde_json::json!([]), duration_ms: 0, diagnostics: Vec::new(), + transcript: vec![], }); } + +/// 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 = Capture::default(); + let step = 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"), + ], + }; + 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] +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()); +} 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/store/types/transcript.rs deleted file mode 100644 index 043b766e..00000000 --- a/src/store/types/transcript.rs +++ /dev/null @@ -1,74 +0,0 @@ -//! 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. -//! -//! Deliberately flat and stringly-typed. Mirroring a host's own event -//! vocabulary into the record would make every event kind it adds later a -//! breaking change to a file format that must stay readable by older builds. A -//! reader meeting an unfamiliar `kind` still has a timestamp and a line of text -//! to render. - -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. -pub const MAX_ENTRY_TEXT_BYTES: usize = 4 * 1024; - -/// One thing an agent did, in the order it did it. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct TranscriptEntry { - /// Epoch milliseconds, as the host stamped the event. - pub at_ms: i64, - /// The host event kind this was folded from — `agent_message`, `tool_call`, - /// `tool_result`, `agent_thinking`, `error`, and so on. - /// - /// Carried verbatim rather than mapped to a closed set, so a kind added to - /// a host's wire vocabulary later shows up here without a change to this - /// file. - pub kind: String, - /// The renderable line: the message text, the tool's one-line summary, the - /// error message. - pub text: String, -} - -impl TranscriptEntry { - /// Build an entry with `text` capped at [`MAX_ENTRY_TEXT_BYTES`]. - /// - /// 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. - #[must_use] - 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 { - let end = text - .char_indices() - .map(|(index, _)| index) - .take_while(|index| *index <= MAX_ENTRY_TEXT_BYTES) - .last() - .unwrap_or(0); - text.truncate(end); - text.push_str(" …[truncated]"); - } - Self { - at_ms, - kind: kind.into(), - text, - } - } -} diff --git a/src/transcript.rs b/src/transcript.rs new file mode 100644 index 00000000..bb2912ac --- /dev/null +++ b/src/transcript.rs @@ -0,0 +1,103 @@ +//! One line of what an agent did, as a run record keeps it. +//! +//! 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. +//! +//! **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 +//! breaking change to a file format that must stay readable by older builds. A +//! reader meeting an unfamiliar `kind` still has a timestamp and a line of text +//! to render. + +use serde::{Deserialize, Serialize}; + +/// 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 +/// 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; + +/// 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")] +pub struct TranscriptEntry { + /// Epoch milliseconds, as the host stamped the event. + pub at_ms: i64, + /// The host event kind this was folded from — `agent_message`, `tool_call`, + /// `tool_result`, `agent_thinking`, `error`, and so on. + /// + /// Carried verbatim rather than mapped to a closed set, so a kind added to + /// a host's wire vocabulary later shows up here without a change to this + /// file. + pub kind: String, + /// The renderable line: the message text, the tool's one-line summary, the + /// error message. + pub text: String, +} + +impl TranscriptEntry { + /// Build an entry with `text` capped at [`MAX_ENTRY_TEXT_BYTES`]. + /// + /// 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 + /// 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(); + 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 <= budget) + .last() + .unwrap_or(0); + text.truncate(end); + text.push_str(TRUNCATION_MARKER); + } + Self { + at_ms, + kind: kind.into(), + text, + } + } +}