From 871fc95fa056aae1ec4dd932d7fa378f03eeef7a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 12:45:01 -0700 Subject: [PATCH 1/7] retry: opt-in backoff sleep for model retries RetryPolicy gains a backoff_sleep flag (off by default) and a sleep_backoff helper. The agent-loop retry loop and RetryMiddleware now sleep for the computed backoff when opted in via with_backoff_sleep, so transient provider/network failures are retried after a real, growing delay instead of back-to-back. Default stays sleep-free to keep tests deterministic. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/harness/agent_loop/mod.rs | 17 ++++++++++------- src/harness/middleware/library/mod.rs | 13 ++++++++----- src/harness/retry/mod.rs | 27 +++++++++++++++++++++++++++ src/harness/retry/types.rs | 10 ++++++++++ 4 files changed, 55 insertions(+), 12 deletions(-) diff --git a/src/harness/agent_loop/mod.rs b/src/harness/agent_loop/mod.rs index dfe0b06..4e0de75 100644 --- a/src/harness/agent_loop/mod.rs +++ b/src/harness/agent_loop/mod.rs @@ -50,10 +50,12 @@ //! # Backoff //! //! Retry backoff durations are *computed* via -//! [`crate::harness::retry::RetryPolicy::backoff_for_attempt`] but the loop does -//! **not** sleep: keeping the loop sleep-free makes tests fast and -//! deterministic. A real provider integration may choose to sleep for the -//! computed duration before retrying. +//! [`crate::harness::retry::RetryPolicy::backoff_for_attempt`]. Whether the loop +//! actually sleeps for that duration is opt-in: it is off by default (keeping +//! tests fast and deterministic) and enabled per policy via +//! [`crate::harness::retry::RetryPolicy::with_backoff_sleep`], so a real +//! provider integration retries after a genuine, growing delay while unit tests +//! stay sleep-free. mod types; @@ -811,9 +813,10 @@ impl AgentHarness { call_id: call_id.clone(), attempt, }); - // Compute (but do not sleep on) the backoff so the - // loop stays fast and deterministic in tests. - let _ = self.policy.retry.backoff_for_attempt(attempt); + // Sleep for the backoff only when the policy opts in + // (`with_backoff_sleep`); otherwise this is a no-op so + // the loop stays fast and deterministic in tests. + self.policy.retry.sleep_backoff(attempt).await; continue; } break Err(error); diff --git a/src/harness/middleware/library/mod.rs b/src/harness/middleware/library/mod.rs index d034a6c..14265a9 100644 --- a/src/harness/middleware/library/mod.rs +++ b/src/harness/middleware/library/mod.rs @@ -19,7 +19,8 @@ //! # Testability //! //! None of these middleware sleep on the wall clock in a way that tests cannot -//! control: [`RetryMiddleware`] computes but never sleeps on backoff, +//! control: [`RetryMiddleware`] sleeps on backoff only when its policy opts in +//! via [`RetryPolicy::with_backoff_sleep`] (off by default), //! [`TimeoutMiddleware`] is exercised under `tokio::time` paused-time tests, and //! [`RateLimitMiddleware`] takes an injectable clock and a configurable poll //! interval so its wait loop can be driven deterministically. @@ -66,8 +67,9 @@ impl RetryMiddleware { /// Returns the policy-derived backoff for the given retry `attempt`. /// - /// Exposed for callers that want to sleep before a retry; the middleware - /// itself never sleeps. + /// Exposed for callers that want to inspect the backoff. The middleware + /// itself sleeps between retries only when the policy opts in via + /// [`RetryPolicy::with_backoff_sleep`]; otherwise it retries back-to-back. pub fn backoff_for_attempt(&self, attempt: usize) -> Duration { self.policy.backoff_for_attempt(attempt) } @@ -95,8 +97,9 @@ impl ModelMiddleware for Retry attempt += 1; let call_id = CallId::new(format!("{}-model", ctx.run_id())); ctx.emit(AgentEvent::RetryScheduled { call_id, attempt }); - // Compute (but do not sleep on) the backoff. - let _ = self.policy.backoff_for_attempt(attempt); + // Sleep for the backoff only when the policy opts in + // (`with_backoff_sleep`); a no-op otherwise. + self.policy.sleep_backoff(attempt).await; continue; } return Err(error); diff --git a/src/harness/retry/mod.rs b/src/harness/retry/mod.rs index d775fdd..e8906d8 100644 --- a/src/harness/retry/mod.rs +++ b/src/harness/retry/mod.rs @@ -66,6 +66,33 @@ impl RetryPolicy { self } + /// Enables or disables actually sleeping for the computed backoff between + /// retries. + /// + /// Off by default so tests stay deterministic and fast. Enable it in + /// production so a transient failure is retried after a real, growing delay + /// rather than back-to-back. See [`RetryPolicy::backoff_sleep`]. + pub fn with_backoff_sleep(mut self, sleep: bool) -> Self { + self.backoff_sleep = sleep; + self + } + + /// Sleeps for this policy's backoff before the given retry `attempt`, but + /// only when [`RetryPolicy::backoff_sleep`] is enabled. + /// + /// A single, reusable helper so every retry loop that honors a + /// [`RetryPolicy`] gets identical, opt-in backoff behavior. A no-op (returns + /// immediately) when sleeping is disabled or the computed backoff is zero. + pub async fn sleep_backoff(&self, attempt: usize) { + if !self.backoff_sleep { + return; + } + let backoff = self.backoff_for_attempt(attempt); + if backoff > Duration::ZERO { + tokio::time::sleep(backoff).await; + } + } + /// Returns `true` when another attempt should be made. /// /// `attempt` is zero-indexed: `0` is the first attempt, `1` is the first diff --git a/src/harness/retry/types.rs b/src/harness/retry/types.rs index 33da805..b0e373b 100644 --- a/src/harness/retry/types.rs +++ b/src/harness/retry/types.rs @@ -38,6 +38,15 @@ pub struct RetryPolicy { /// When `true`, the caller should supply a `[0, 1)` random value to /// [`RetryPolicy::backoff_for_attempt_with`] to add jitter. pub jitter: bool, + /// When `true`, retry loops that honor this policy actually + /// [`tokio::time::sleep`] for the computed backoff between attempts. + /// + /// Defaults to `false` so the harness stays fast and deterministic under + /// test: the backoff is *computed* (for observability) but not slept on. + /// Production/streaming callers opt in via + /// [`RetryPolicy::with_backoff_sleep`] so a transient provider/network + /// failure is retried after a real, growing delay instead of back-to-back. + pub backoff_sleep: bool, } impl Default for RetryPolicy { @@ -48,6 +57,7 @@ impl Default for RetryPolicy { max_backoff_ms: 30_000, multiplier: 2.0, jitter: false, + backoff_sleep: false, } } } From 4a6e2bdc244443ed8e39166d103a9f02694dcfb8 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 12:53:17 -0700 Subject: [PATCH 2/7] graph: node-level retry + resumable failure checkpoints Add opt-in CompiledGraph::with_node_retry(RetryPolicy): each node handler is re-run from its start on a retryable (model/tool) error, up to the policy cap, emitting GraphEvent::NodeRetryScheduled and sleeping the opt-in backoff. Both the sequential and parallel runners drive handlers through the shared run_node_with_retry helper (which also applies the per-node timeout). When a handler fails after retries (or a non-retryable error hits), the runner now hands back partial progress instead of discarding it: completed branches' updates are folded into committed state and a resumable failure-boundary checkpoint is persisted (failed node scheduled as next_nodes, successful branches as completed_tasks, error + failed node in metadata). The Failed status carries that checkpoint id, so a crashed/exhausted run can be resumed or continued rather than lost. Non-checkpointed runs abort exactly as before. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/compiled/mod.rs | 252 +++++++++++++++++++++++++++++++++--- src/graph/compiled/types.rs | 9 ++ src/graph/stream/types.rs | 13 ++ 3 files changed, 253 insertions(+), 21 deletions(-) diff --git a/src/graph/compiled/mod.rs b/src/graph/compiled/mod.rs index f4d17d4..1993dfd 100644 --- a/src/graph/compiled/mod.rs +++ b/src/graph/compiled/mod.rs @@ -48,7 +48,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, SystemTime}; use crate::graph::builder::{ - Branch, BuilderNode, END, ForkId, NodeContext, NodeFuture, NodeMeta, START, + Branch, BuilderNode, END, ForkId, NodeContext, NodeFuture, NodeHandler, NodeMeta, START, }; use crate::graph::checkpoint::{ Checkpoint, CheckpointConfig, CheckpointTuple, Checkpointer, DurabilityMode, @@ -63,6 +63,7 @@ use crate::graph::stream::{GraphEvent, GraphEventSink}; use crate::harness::ids::{ CheckpointId, ExecutionStatus, GraphId, InterruptId, NodeId, RunId, ThreadId, }; +use crate::harness::retry::is_retryable; use crate::{Result, TinyAgentsError}; static SEQ: AtomicU64 = AtomicU64::new(0); @@ -110,6 +111,26 @@ struct StepRun { goto_map: HashMap>, /// The lowest-index branch interrupt, if any (its active-set index + value). interrupt: Option<(usize, Interrupt)>, + /// A node-handler failure that survived the node-retry policy, if any. When + /// set, `updates` still carries the updates of the branches that completed + /// *before* the failing branch, so the executor can fold that partial + /// progress into committed state and persist a resumable failure boundary. + failure: Option, +} + +/// A node-handler failure captured by a runner so the executor can persist a +/// resumable failure-boundary checkpoint instead of discarding partial progress. +struct StepFailure { + /// The node whose handler ultimately failed (after any retries). + failed_node: NodeId, + /// Nodes that completed successfully before the failure, in active-set order + /// (recorded as the checkpoint's completed tasks). + completed: Vec, + /// Nodes to re-run on resume: the failed node first, then the not-yet-folded + /// members of the step (recorded as the checkpoint's next nodes). + pending: Vec, + /// The escalated error. + error: TinyAgentsError, } /// One scheduled activation in the active set: the node plus an optional @@ -184,6 +205,7 @@ impl CompiledGraph { max_concurrency, node_timeout, durability: crate::graph::checkpoint::DurabilityMode::default(), + node_retry: None, } } @@ -232,6 +254,27 @@ impl CompiledGraph { self } + /// Sets the per-node [`RetryPolicy`] applied around every node handler. + /// + /// Opt-in network resilience for the graph: when a node handler fails with a + /// [retryable][crate::harness::retry::is_retryable] error (a model or tool + /// error — the transient class), the executor re-runs the node from its + /// start up to the policy's attempt cap, emitting a + /// [`GraphEvent::NodeRetryScheduled`](crate::graph::stream::GraphEvent::NodeRetryScheduled) + /// before each retry. Backoff between attempts is slept on only when the + /// policy opts in via + /// [`RetryPolicy::with_backoff_sleep`](crate::harness::retry::RetryPolicy::with_backoff_sleep). + /// + /// Non-retryable errors, and retryable errors once attempts are exhausted, + /// escalate — and, on a checkpointed thread, leave a resumable + /// failure-boundary checkpoint (see [`CompiledGraph::resume`]) so the run can + /// be restarted or continued rather than lost. Without a policy (the + /// default) the first node error aborts the run immediately. + pub fn with_node_retry(mut self, policy: crate::harness::retry::RetryPolicy) -> Self { + self.node_retry = Some(policy); + self + } + /// Sets the checkpoint namespace (used by subgraph wrappers). pub fn with_namespace(mut self, namespace: Vec) -> Self { self.namespace = namespace; @@ -776,7 +819,7 @@ where self.emit(GraphEvent::RunStarted { run_id: run_id.clone(), }); - self.fail_run(&run_id, &thread_id, started_at, steps, &err) + self.fail_run(&run_id, &thread_id, started_at, steps, &err, None) .await; return Err(err); } @@ -820,14 +863,14 @@ where .min(self.recursion_policy.max_total_steps); if steps >= step_limit { let err = TinyAgentsError::RecursionLimit(step_limit); - self.fail_run(&run_id, &thread_id, started_at, steps, &err) + self.fail_run(&run_id, &thread_id, started_at, steps, &err, None) .await; return Err(err); } // Node-loop recursion: enforce `max_visits_per_node` per activation. for activation in &active { if let Err(err) = recursion.record_node_visit(&mut node_visits, &activation.node) { - self.fail_run(&run_id, &thread_id, started_at, steps, &err) + self.fail_run(&run_id, &thread_id, started_at, steps, &err, None) .await; return Err(err); } @@ -871,10 +914,11 @@ where updates, goto_map, interrupt, + failure, } = match run_result { Ok(step_run) => step_run, Err(err) => { - self.fail_run(&run_id, &thread_id, started_at, steps, &err) + self.fail_run(&run_id, &thread_id, started_at, steps, &err, None) .await; return Err(err); } @@ -893,12 +937,53 @@ where let child_runs_meta = serde_json::to_value(&step_child_runs).unwrap_or(serde_json::Value::Null); + // Node-handler failure (survived any node-retry policy): the updates + // of the branches that completed before it are already folded into + // `state` above, so persist a resumable failure-boundary checkpoint + // scheduling the failed node (and the not-yet-run tail) for a later + // `resume`/`retry`, record a `Failed` status carrying the error and + // that checkpoint, and abort. Without a checkpointer/thread the + // checkpoint is a no-op and the run aborts exactly as before. + if let Some(fail) = failure { + let StepFailure { + failed_node, + completed, + pending, + error, + } = fail; + let checkpoint_id = self + .persist_failure_checkpoint( + &thread_id, + &run_id, + &state, + &pending, + &completed, + parent_checkpoint.clone(), + steps, + &failed_node, + &error, + &recursion_meta, + &child_runs_meta, + ) + .await?; + self.fail_run( + &run_id, + &thread_id, + started_at, + steps, + &error, + checkpoint_id, + ) + .await; + return Err(error); + } + // Interrupt: persist a checkpoint whose next nodes are the // not-yet-completed members of this step (interrupted node first), // then return control to the caller. if let Some((index, emitted)) = interrupt { if let Err(err) = self.require_interrupt_durability(&thread_id) { - self.fail_run(&run_id, &thread_id, started_at, steps, &err) + self.fail_run(&run_id, &thread_id, started_at, steps, &err, None) .await; return Err(err); } @@ -1052,6 +1137,11 @@ where /// Emits a [`GraphEvent::RunFailed`] and records a terminal `Failed` status /// for a run that aborted with `err`. + /// + /// `checkpoint_id` is the resumable failure-boundary checkpoint when the run + /// left one (a node-handler failure on a checkpointed thread), or `None` for + /// a structural/non-resumable abort. When present it is recorded on the + /// status so an observer can locate the checkpoint to `resume`/`retry` from. async fn fail_run( &self, run_id: &RunId, @@ -1059,6 +1149,7 @@ where started_at: SystemTime, steps: usize, err: &TinyAgentsError, + checkpoint_id: Option, ) { self.emit(GraphEvent::RunFailed { run_id: run_id.clone(), @@ -1069,9 +1160,66 @@ where status.current_step = steps; status.ended_at = Some(SystemTime::now()); status.error = Some(err.to_string()); + status.checkpoint_id = checkpoint_id; self.save_status(status).await; } + /// Persists a resumable failure-boundary checkpoint for a node-handler + /// failure that survived the node-retry policy. + /// + /// Mirrors the interrupt boundary: `next_nodes` schedules the failed node + /// (and any not-yet-run members of the step) so `resume`/`retry` re-runs + /// exactly what did not complete, while `completed_tasks` records the + /// branches that already succeeded (their updates are folded into `state` + /// before this is called). The rendered error and failed node id are stamped + /// into the checkpoint metadata for diagnosis. A no-op returning `None` when + /// no checkpointer/thread is configured — the run then aborts without a + /// resumable checkpoint, exactly as before this policy existed. + #[allow(clippy::too_many_arguments)] + async fn persist_failure_checkpoint( + &self, + thread_id: &Option, + run_id: &RunId, + state: &State, + next_nodes: &[NodeId], + completed_tasks: &[NodeId], + parent: Option, + step: usize, + failed_node: &NodeId, + error: &TinyAgentsError, + recursion: &serde_json::Value, + child_runs: &serde_json::Value, + ) -> Result> { + let (Some(checkpointer), Some(thread)) = (&self.checkpointer, thread_id) else { + return Ok(None); + }; + let checkpoint = Checkpoint { + thread_id: thread.to_string(), + checkpoint_id: next_id("ckpt"), + run_id: Some(run_id.to_string()), + parent_checkpoint_id: parent, + namespace: self.namespace.clone(), + state: state.clone(), + next_nodes: next_nodes.to_vec(), + completed_tasks: completed_tasks.to_vec(), + pending_writes: Vec::new(), + interrupts: Vec::new(), + metadata: serde_json::json!({ + "source": "loop", + "step": step, + "recursion": recursion, + "child_runs": child_runs, + "failed_node": failed_node.as_str(), + "error": error.to_string(), + }), + }; + let id = checkpointer.put(checkpoint).await?; + self.emit(GraphEvent::CheckpointSaved { + checkpoint_id: id.clone(), + }); + Ok(Some(id)) + } + /// Builds the per-task [`NodeContext`] for `node_id` at the given branch. /// /// `fork` carries the branch identity in a concurrent step (`None` in @@ -1123,6 +1271,51 @@ where } } + /// Runs one node handler under the graph's node-retry policy. + /// + /// Builds a fresh handler future (and re-clones the context) for each + /// attempt, so a retried node re-runs from its start — matching the durable + /// execution model, where a node is never suspended mid-flight. On a + /// [retryable][crate::harness::retry::is_retryable] error, when a + /// [`RetryPolicy`](crate::harness::retry::RetryPolicy) is configured and + /// permits another attempt, it emits + /// [`GraphEvent::NodeRetryScheduled`], sleeps the (opt-in) backoff, and + /// retries. Non-retryable errors, absence of a policy, or an exhausted + /// attempt budget return the error unchanged. The per-node timeout still + /// bounds every individual attempt via [`Self::run_node_future`]. + async fn run_node_with_retry( + &self, + node_id: &NodeId, + handler: &Arc>, + state: &State, + ctx: NodeContext, + step: usize, + ) -> Result> { + let mut attempt = 0usize; + loop { + let fut = handler(state.clone(), ctx.clone()); + match self.run_node_future(node_id, fut).await { + Ok(result) => return Ok(result), + Err(error) => { + let retry = self + .node_retry + .as_ref() + .filter(|policy| policy.should_retry(attempt) && is_retryable(&error)); + let Some(policy) = retry else { + return Err(error); + }; + attempt += 1; + self.emit(GraphEvent::NodeRetryScheduled { + node: node_id.clone(), + step, + attempt, + }); + policy.sleep_backoff(attempt).await; + } + } + } + } + /// Folds a single successful branch result into the step accumulators. /// /// Pushes the node to `visited`, records updates/goto, emits the matching @@ -1196,6 +1389,7 @@ where let mut updates: Vec = Vec::new(); let mut goto_map: HashMap> = HashMap::new(); let mut interrupt: Option<(usize, Interrupt)> = None; + let mut failure: Option = None; for (index, activation) in active.iter().enumerate() { let node_id = &activation.node; @@ -1226,7 +1420,7 @@ where child_runs, ); let result = match self - .run_node_future(node_id, (node.handler)(state.clone(), ctx)) + .run_node_with_retry(node_id, &node.handler, state, ctx, step) .await { Ok(result) => result, @@ -1236,7 +1430,16 @@ where step, error: error.to_string(), }); - return Err(error); + // Preserve the progress of the branches that already ran: + // record them as completed and schedule this node plus the + // not-yet-run tail for a resumable retry. + failure = Some(StepFailure { + failed_node: node_id.clone(), + completed: activation_nodes(&active[..index]), + pending: activation_nodes(&active[index..]), + error, + }); + break; } }; @@ -1258,6 +1461,7 @@ where updates, goto_map, interrupt, + failure, }) } @@ -1287,9 +1491,11 @@ where frames: &[RecursionFrame], child_runs: &ChildRunSink, ) -> Result> { - let timeout = self.node_timeout; // Build one forked context + future per branch. Node lookup and resume - // consumption happen up front so the futures borrow nothing local. + // consumption happen up front so the futures borrow nothing mutable; each + // branch drives its handler through the node-retry policy (which also + // applies the per-node timeout), so a transient failure in one branch is + // retried without disturbing its siblings. let mut futures = Vec::with_capacity(active.len()); for (index, activation) in active.iter().enumerate() { let node_id = &activation.node; @@ -1325,18 +1531,11 @@ where frames, child_runs, ); - let raw = (node.handler)(state.clone(), ctx); + let handler = node.handler.clone(); let owned_node = node_id.clone(); futures.push(async move { - match timeout { - Some(d) => match tokio::time::timeout(d, raw).await { - Ok(result) => result, - Err(_) => Err(TinyAgentsError::Timeout(format!( - "node `{owned_node}` exceeded its {d:?} timeout" - ))), - }, - None => raw.await, - } + self.run_node_with_retry(&owned_node, &handler, state, ctx, step) + .await }); } @@ -1361,6 +1560,7 @@ where let mut updates: Vec = Vec::new(); let mut goto_map: HashMap> = HashMap::new(); let mut interrupt: Option<(usize, Interrupt)> = None; + let mut failure: Option = None; for (index, (activation, result)) in active.iter().zip(results).enumerate() { let node_id = &activation.node; @@ -1372,7 +1572,16 @@ where step, error: error.to_string(), }); - return Err(error); + // The lowest-index failing branch is terminal: fold the + // lower-index successes (already applied above) and schedule + // this branch plus the rest for a resumable retry. + failure = Some(StepFailure { + failed_node: node_id.clone(), + completed: activation_nodes(&active[..index]), + pending: activation_nodes(&active[index..]), + error, + }); + break; } }; @@ -1394,6 +1603,7 @@ where updates, goto_map, interrupt, + failure, }) } diff --git a/src/graph/compiled/types.rs b/src/graph/compiled/types.rs index d6f36d4..bfb45c9 100644 --- a/src/graph/compiled/types.rs +++ b/src/graph/compiled/types.rs @@ -67,6 +67,14 @@ pub struct CompiledGraph { /// When checkpoints are persisted relative to execution (default /// [`DurabilityMode::Sync`]). pub(crate) durability: DurabilityMode, + /// Optional per-node retry policy applied around each node handler. When + /// set, a handler that fails with a + /// [retryable][crate::harness::retry::is_retryable] error is re-run from its + /// start (with opt-in backoff) up to the policy's attempt cap before the + /// failure escalates. `None` (default) preserves the original + /// abort-on-first-error behavior. Configured via + /// [`CompiledGraph::with_node_retry`](crate::graph::CompiledGraph::with_node_retry). + pub(crate) node_retry: Option, } impl std::fmt::Debug for CompiledGraph { @@ -108,6 +116,7 @@ impl Clone for CompiledGraph { max_concurrency: self.max_concurrency, node_timeout: self.node_timeout, durability: self.durability, + node_retry: self.node_retry.clone(), } } } diff --git a/src/graph/stream/types.rs b/src/graph/stream/types.rs index 1d7f16a..8836aee 100644 --- a/src/graph/stream/types.rs +++ b/src/graph/stream/types.rs @@ -84,6 +84,17 @@ pub enum GraphEvent { /// Rendered error. error: String, }, + /// A node handler failed with a retryable error and a retry was scheduled + /// under the graph's node-retry policy. Emitted before the (opt-in) backoff + /// wait and the re-run of the node from its start. + NodeRetryScheduled { + /// Node id. + node: NodeId, + /// Step number. + step: usize, + /// The 1-based retry attempt about to be made. + attempt: usize, + }, /// A node produced a state update applied at the boundary. StateUpdated { /// Node id. @@ -162,6 +173,7 @@ impl GraphEvent { GraphEvent::NodeStarted { .. } => "node.started", GraphEvent::NodeCompleted { .. } => "node.completed", GraphEvent::NodeFailed { .. } => "node.failed", + GraphEvent::NodeRetryScheduled { .. } => "node.retry_scheduled", GraphEvent::StateUpdated { .. } => "state.updated", GraphEvent::RouteSelected { .. } => "route.selected", GraphEvent::CheckpointSaved { .. } => "checkpoint.saved", @@ -185,6 +197,7 @@ impl GraphEvent { | GraphEvent::NodeStarted { step, .. } | GraphEvent::NodeCompleted { step, .. } | GraphEvent::NodeFailed { step, .. } + | GraphEvent::NodeRetryScheduled { step, .. } | GraphEvent::StateUpdated { step, .. } | GraphEvent::ContextForked { step, .. } => Some(*step), GraphEvent::RunCompleted { steps, .. } => Some(*steps), From 0a43d8a8abc0154cbfc50f5dd05c2cd62dd49541 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 12:58:22 -0700 Subject: [PATCH 3/7] graph: retry() helper + resilience tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add CompiledGraph::retry(thread) — shorthand for resume with an empty command — to re-run a failed run's failure-boundary checkpoint, and document the inspect/update_state/retry feedback loop for continuing on user input. Tests: node retry recovers a transient failure (one NodeRetryScheduled per blip); exhausted retries leave a resumable checkpoint that retry() completes; a failure with no retry policy is still resumable (resumable-abort default); edit-state-then-retry runs against edited state; parallel partial progress is preserved (successful branch folded, only failed branch rescheduled). Plus a paused-time test that backoff sleep fires only when opted in. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- src/graph/compiled/mod.rs | 21 +++++ src/graph/compiled/test.rs | 187 +++++++++++++++++++++++++++++++++++++ src/harness/retry/test.rs | 32 +++++++ 3 files changed, 240 insertions(+) diff --git a/src/graph/compiled/mod.rs b/src/graph/compiled/mod.rs index 1993dfd..58ec03f 100644 --- a/src/graph/compiled/mod.rs +++ b/src/graph/compiled/mod.rs @@ -425,6 +425,27 @@ where .await } + /// Retries a failed run from its latest (failure-boundary) checkpoint, + /// re-running the node that failed and the not-yet-run tail of that step. + /// + /// This is the resume counterpart for the *failure* path (as opposed to a + /// human interrupt): after a node handler aborts a checkpointed run — a + /// transient outage that outlived the node-retry policy, or a hard crash — + /// the run leaves a resumable checkpoint (see + /// [`CompiledGraph::with_node_retry`]). Calling `retry` re-runs exactly what + /// did not complete, carrying no resume value. It is shorthand for + /// [`CompiledGraph::resume`] with an empty [`Command`]. + /// + /// To continue on *user feedback* instead of a bare retry, first inspect the + /// committed state with + /// [`get_state`](CompiledGraph::get_state), edit it with + /// [`update_state`](CompiledGraph::update_state), then call `retry` (or + /// `resume`) — the edited state is what the re-run sees. + pub async fn retry(&self, thread_id: impl Into) -> Result> { + self.resume_from(thread_id, ResumeTarget::Latest, Command::new()) + .await + } + /// Resumes a run from a specific checkpoint (time-travel resume). /// /// [`ResumeTarget::Latest`] behaves exactly like [`CompiledGraph::resume`]; diff --git a/src/graph/compiled/test.rs b/src/graph/compiled/test.rs index 6356e6e..27a10c8 100644 --- a/src/graph/compiled/test.rs +++ b/src/graph/compiled/test.rs @@ -9,6 +9,7 @@ use crate::graph::command::{Command, Interrupt, NodeResult, Send}; use crate::graph::reducer::ClosureStateReducer; use crate::graph::stream::{CollectingSink, GraphEvent}; use crate::harness::ids::ExecutionStatus; +use crate::harness::retry::RetryPolicy; use serde_json::json; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; @@ -1316,3 +1317,189 @@ async fn node_timeout_fails_slow_handler() { let err = graph.run(0).await.unwrap_err(); assert!(matches!(err, TinyAgentsError::Timeout(_))); } + +// ── Network resilience: node retry + resumable failures ────────────────────── + +/// A single-node graph whose handler fails (with a retryable model error) the +/// first `fail_times` invocations, then succeeds with `+1`. The shared counter +/// lets a test observe how many attempts were made. +fn flaky_graph(fail_times: usize, attempts: Arc) -> CompiledGraph { + GraphBuilder::::overwrite() + .add_node("flaky", move |s, _c: NodeContext| { + let attempts = attempts.clone(); + async move { + let n = attempts.fetch_add(1, AtomicOrdering::SeqCst); + if n < fail_times { + Err(TinyAgentsError::Model(format!("transient blip {n}"))) + } else { + Ok(NodeResult::Update(s + 1)) + } + } + }) + .set_entry("flaky") + .set_finish("flaky") + .compile() + .unwrap() +} + +#[tokio::test] +async fn node_retry_recovers_transient_failure() { + let attempts = Arc::new(AtomicUsize::new(0)); + let sink = Arc::new(CollectingSink::new()); + // Fail twice, succeed on the third attempt; the policy allows 1 try + 3 + // retries, so recovery is within budget. + let graph = flaky_graph(2, attempts.clone()) + .with_node_retry(RetryPolicy::default().with_max_attempts(4)) + .with_event_sink(sink.clone()); + + let run = graph.run(10).await.unwrap(); + assert_eq!(run.state, 11); + assert_eq!(run.status.status, ExecutionStatus::Completed); + assert_eq!(attempts.load(AtomicOrdering::SeqCst), 3); + + let retries = sink + .events() + .into_iter() + .filter(|e| matches!(e, GraphEvent::NodeRetryScheduled { .. })) + .count(); + assert_eq!(retries, 2, "one retry per transient failure"); +} + +#[tokio::test] +async fn exhausted_retries_leave_a_resumable_failure_checkpoint() { + let cp = Arc::new(InMemoryCheckpointer::::new()); + let attempts = Arc::new(AtomicUsize::new(0)); + // Fail the first 3 invocations. With 1 try + 1 retry the first run exhausts + // its budget (2 attempts: n=0,1) and aborts, leaving a resumable checkpoint. + let graph = flaky_graph(3, attempts.clone()) + .with_node_retry(RetryPolicy::default().with_max_attempts(2)) + .with_checkpointer(cp.clone()); + + let err = graph.run_with_thread("net", 100).await.unwrap_err(); + assert!(matches!(err, TinyAgentsError::Model(_)), "got {err:?}"); + assert_eq!(attempts.load(AtomicOrdering::SeqCst), 2, "1 try + 1 retry"); + + // The failure boundary is durable: the checkpoint schedules the failed node + // for re-run at the first superstep. + let list = cp.list("net").await.unwrap(); + let last = list.last().expect("a failure checkpoint was persisted"); + assert_eq!(last.next_nodes, vec![NodeId::from("flaky")]); + let snapshot = graph.get_state("net", None).await.unwrap().unwrap(); + assert_eq!( + snapshot.metadata.step, 1, + "failure boundary is at the first superstep" + ); + + // Retry: attempt n=2 fails, retry n=3 (3<3 is false) succeeds — the run + // completes without losing the earlier progress. + let resumed = graph.retry("net").await.unwrap(); + assert_eq!( + resumed.state, 101, + "resume re-runs the failed node to success" + ); + assert_eq!(resumed.status.status, ExecutionStatus::Completed); + assert_eq!(attempts.load(AtomicOrdering::SeqCst), 4); +} + +#[tokio::test] +async fn node_failure_without_retry_policy_is_resumable() { + // No node-retry policy configured: the first failure aborts the run, but a + // checkpointed thread still leaves a resumable failure boundary (the + // "resumable abort" default) rather than losing the run. + let cp = Arc::new(InMemoryCheckpointer::::new()); + let attempts = Arc::new(AtomicUsize::new(0)); + let graph = flaky_graph(1, attempts.clone()).with_checkpointer(cp.clone()); + + let err = graph.run_with_thread("once", 7).await.unwrap_err(); + assert!(matches!(err, TinyAgentsError::Model(_)), "got {err:?}"); + assert_eq!( + attempts.load(AtomicOrdering::SeqCst), + 1, + "no retries attempted" + ); + + // Failed status carries the resumable checkpoint id. + let status = graph.get_state("once", None).await.unwrap().unwrap(); + assert_eq!(status.next_nodes, vec![NodeId::from("flaky")]); + + // The transient condition has cleared; retry completes the run. + let resumed = graph.retry("once").await.unwrap(); + assert_eq!(resumed.state, 8); + assert_eq!(resumed.status.status, ExecutionStatus::Completed); +} + +#[tokio::test] +async fn edit_state_then_retry_uses_the_edited_state() { + // User-feedback continuation: after a failure, the operator edits committed + // state via update_state, then retries; the re-run sees the edited value. + let cp = Arc::new(InMemoryCheckpointer::::new()); + let attempts = Arc::new(AtomicUsize::new(0)); + let graph = flaky_graph(1, attempts.clone()).with_checkpointer(cp.clone()); + + graph.run_with_thread("feedback", 0).await.unwrap_err(); + + // Operator bumps the committed state by +40, inheriting the failure + // boundary's pending nodes (`flaky`), then retries. The node adds +1 to + // whatever state it now sees. + graph.update_state("feedback", 40, None).await.unwrap(); + let resumed = graph.retry("feedback").await.unwrap(); + assert_eq!( + resumed.state, 41, + "retry runs against the edited state (40) + 1" + ); + assert_eq!(resumed.status.status, ExecutionStatus::Completed); +} + +#[tokio::test] +async fn parallel_partial_progress_is_preserved_on_failure() { + // A parallel step where one branch succeeds and a lower/higher-index branch + // fails: the successful branch's update is folded into committed state and + // the failure checkpoint schedules only the failed branch for re-run. + let cp = Arc::new(InMemoryCheckpointer::::new()); + let attempts = Arc::new(AtomicUsize::new(0)); + let graph = GraphBuilder::::new() + .with_parallel(true) + .set_reducer(ClosureStateReducer::new(|s: i32, u: i32| Ok(s + u))) + .add_node("seed", |_s, _c: NodeContext| async move { + Ok(NodeResult::Command(Command::goto(["ok", "flaky"]))) + }) + .add_node("ok", |_s, _c: NodeContext| async move { + Ok(NodeResult::Update(1)) + }) + .add_node("flaky", move |_s, _c: NodeContext| { + let attempts = attempts.clone(); + async move { + // Fail on the first invocation, succeed on resume. + if attempts.fetch_add(1, AtomicOrdering::SeqCst) == 0 { + Err(TinyAgentsError::Model("branch blip".into())) + } else { + Ok(NodeResult::Update(10)) + } + } + }) + .set_entry("seed") + .mark_command_routing("seed") + .set_finish("ok") + .set_finish("flaky") + .compile() + .unwrap() + .with_checkpointer(cp.clone()); + + // First run: seed fans out; the "ok" branch commits +1, "flaky" aborts. + graph.run_with_thread("fanout", 0).await.unwrap_err(); + let snapshot = graph.get_state("fanout", None).await.unwrap().unwrap(); + assert_eq!( + snapshot.values, 1, + "the successful branch's +1 is preserved" + ); + assert!( + snapshot.next_nodes.contains(&NodeId::from("flaky")), + "only the failed branch is scheduled for re-run: {:?}", + snapshot.next_nodes + ); + + // Resume: the flaky branch now succeeds (+10) without re-running "ok". + let resumed = graph.retry("fanout").await.unwrap(); + assert_eq!(resumed.state, 11, "1 (preserved) + 10 (re-run branch)"); + assert_eq!(resumed.status.status, ExecutionStatus::Completed); +} diff --git a/src/harness/retry/test.rs b/src/harness/retry/test.rs index 869ba45..c0fafcc 100644 --- a/src/harness/retry/test.rs +++ b/src/harness/retry/test.rs @@ -179,3 +179,35 @@ fn rate_limiter_refill_caps_at_capacity() { let later = start + Duration::from_secs(60); assert_eq!(limiter.available(later), 5); } + +#[test] +fn backoff_sleep_defaults_off_and_is_opt_in() { + // Default policy does not sleep, keeping retry loops deterministic in tests. + assert!(!RetryPolicy::default().backoff_sleep); + // The builder flips it on for production callers. + assert!( + RetryPolicy::default() + .with_backoff_sleep(true) + .backoff_sleep + ); +} + +#[tokio::test(start_paused = true)] +async fn sleep_backoff_waits_only_when_enabled() { + use tokio::time::Instant as TokioInstant; + + // Disabled (default): returns immediately with no virtual time elapsed. + let policy = RetryPolicy::default(); + let t0 = TokioInstant::now(); + policy.sleep_backoff(1).await; + assert_eq!(t0.elapsed(), Duration::ZERO); + + // Enabled: advances virtual time by the computed backoff (attempt 1 = + // initial_backoff_ms with the default multiplier applied at attempt^power). + let sleeping = RetryPolicy::default().with_backoff_sleep(true); + let expected = sleeping.backoff_for_attempt(1); + let t1 = TokioInstant::now(); + sleeping.sleep_backoff(1).await; + assert_eq!(t1.elapsed(), expected); + assert!(expected > Duration::ZERO); +} From 17cec870c7e3505ad82ec56a653800a33c8cec5f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 2 Jul 2026 13:04:53 -0700 Subject: [PATCH 4/7] examples+docs: resilient_graph + fault-tolerance docs Add examples/resilient_graph.rs demonstrating node-level retry absorbing a transient fetch blip and a resumable failure checkpoint that retry() restarts after an outage clears. Rewrite docs/modules/graph/fault-tolerance.md to document the implemented behavior, add a resilience section to the graph module docs and wiki, and list the example in README/Examples. Claude-Session: https://claude.ai/code/session_01NRL2bim9Gz3UN3SRtpeXEX --- README.md | 2 + docs/modules/graph/fault-tolerance.md | 118 ++++++++++++++-------- examples/resilient_graph.rs | 135 ++++++++++++++++++++++++++ src/graph/compiled/mod.rs | 30 +++++- 4 files changed, 243 insertions(+), 42 deletions(-) create mode 100644 examples/resilient_graph.rs diff --git a/README.md b/README.md index 13ee8c1..198fada 100644 --- a/README.md +++ b/README.md @@ -228,6 +228,8 @@ All live in [`examples/`](examples/): - **`basic_graph`** — a minimal typed state graph: `START`, nodes, edges, `END`. - **`complex_graph`** — conditional routing, fanout, and richer topology. - **`durable_graph`** — checkpoints, resume, and time-travel over supersteps. +- **`resilient_graph`** — node-level retry over transient failures, plus a + resumable failure checkpoint that `retry` restarts after an outage clears. - **`agent_loop_tools`** — the agent ↔ tool loop the harness runs. - **`orchestrator_subagents`** — **recursion in action:** an orchestrator agent that calls sub-agents as tools, with depth tracking and rolled-up usage. diff --git a/docs/modules/graph/fault-tolerance.md b/docs/modules/graph/fault-tolerance.md index 0fcca22..df16c3f 100644 --- a/docs/modules/graph/fault-tolerance.md +++ b/docs/modules/graph/fault-tolerance.md @@ -1,42 +1,80 @@ # Graph Error Handling And Fault Tolerance -Default behavior: - -- node error fails the run -- retry only if policy allows it -- timeout fails the node -- checkpoint remains at the last completed boundary -- pending writes from completed tasks remain available when the checkpointer - supports them - -Target behavior: - -- route node errors to a node-specific or default error handler -- expose handled errors in task events and checkpoint task metadata -- retry with backoff and jitter -- support optional nodes by explicit policy -- support cooperative drain/shutdown with a drain reason -- distinguish graph recursion errors from node failures - -Error types should distinguish: - -- missing start -- duplicate node -- missing node -- missing edge target -- duplicate route -- missing route -- invalid command target -- invalid parent graph command -- invalid send target -- recursion limit -- checkpoint required -- checkpoint missing -- interrupt resume mismatch -- reducer conflict -- invalid concurrent update -- node timeout -- node failure -- graph drained -- serialization failure -- checkpoint backend failure +The graph runtime survives transient network failures and makes a hard failure +restartable, using two opt-in mechanisms plus the existing checkpoint/resume +machinery. + +## Node retry (surviving network blips) + +`CompiledGraph::with_node_retry(RetryPolicy)` wraps every node handler. When a +handler fails with a [retryable] error — `TinyAgentsError::Model` or +`TinyAgentsError::Tool`, the transient class classified by +`harness::retry::is_retryable` — the executor re-runs the node **from its start** +up to the policy's `max_attempts`, emitting `GraphEvent::NodeRetryScheduled` +before each retry. Because a node is never suspended mid-flight, a retry rebuilds +the handler future and context from scratch, matching the durable-execution +model. + +Backoff between attempts is computed from the `RetryPolicy` (exponential, with +optional jitter) but only **slept on** when the policy opts in via +`RetryPolicy::with_backoff_sleep(true)`. The default is sleep-free so unit tests +stay deterministic; production callers enable it so retries wait a real, growing +delay. The same opt-in switch governs the harness model-call retry loop and +`RetryMiddleware`. + +Both the sequential and parallel runners drive handlers through one shared +`run_node_with_retry` helper, which also applies the per-node timeout — so a +retried attempt is still bounded by `node_timeout`. + +## Resumable failures (restarting / continuing after a hard failure) + +When a handler fails **beyond** the retry budget, or with a non-retryable error, +the run is not lost. On a checkpointed thread the executor: + +1. folds the updates of the branches that completed *before* the failing one + into committed state (partial parallel progress is preserved, not discarded); +2. persists a **failure-boundary checkpoint** whose `next_nodes` schedule the + failed node (and the not-yet-run tail of the step) for re-run, records the + successful nodes as `completed_tasks`, and stamps the rendered `error` and + `failed_node` into the checkpoint metadata; +3. records a `Failed` run status carrying that checkpoint id; +4. returns the error to the caller. + +Without a checkpointer/thread the run aborts immediately with the error, exactly +as before — the failure checkpoint is a no-op. + +### Restarting and continuing + +Because `resume`/`resume_from` replay from a checkpoint's `next_nodes`, a failed +run resumes exactly like an interrupted one: + +- `CompiledGraph::retry(thread)` — shorthand for `resume` with an empty command; + re-runs the failed node and the not-yet-run tail from the failure boundary. +- **Continue on user feedback** — inspect committed state with `get_state`, edit + it with `update_state` (attributed through the reducers), then `retry`/`resume`. + The re-run sees the edited state. + +See the `resilient_graph` example (`cargo run --example resilient_graph`) for +both mechanisms end to end. + +## Error taxonomy + +`TinyAgentsError` distinguishes structural/config errors (non-resumable) from +node failures (resumable on a checkpointed thread): + +- structural / not retried, not resumable: missing start, missing node, missing + edge target, missing route, invalid command/parent/send target, recursion + limit, node-visit limit, sub-agent depth, checkpoint required/missing, resume + mismatch, reducer conflict, invalid concurrent update, serialization failure, + checkpoint backend failure; +- transient / retryable and resumable: `Model`, `Tool` (and `Timeout` aborts the + node but leaves a resumable boundary like any other node failure). + +## Future work + +- Route node errors to a node-specific or default error-handler node. +- Cooperative drain/shutdown with a drain reason. +- Populate the checkpoint `pending_writes` list explicitly (today partial + progress is folded into committed state instead). + +[retryable]: ../../../src/harness/retry/mod.rs diff --git a/examples/resilient_graph.rs b/examples/resilient_graph.rs new file mode 100644 index 0000000..6fdb2bb --- /dev/null +++ b/examples/resilient_graph.rs @@ -0,0 +1,135 @@ +//! Surviving network blips and restarting a failed run. +//! +//! Two resilience primitives, on one durable graph: +//! +//! 1. **Node-level retry.** [`CompiledGraph::with_node_retry`] re-runs a node +//! from its start when its handler fails with a transient +//! ([retryable][tinyagents::harness::retry::is_retryable]) error — a model or +//! tool/network error. A `fetch` node here fails its first two attempts (a +//! simulated connectivity blip) and succeeds on the third, so the run +//! completes without any operator involvement. +//! +//! 2. **Resumable failure + restart.** A `commit` node fails *harder* than the +//! retry budget allows. On a checkpointed thread the run does not vanish: it +//! persists a resumable failure-boundary checkpoint (the failed node is +//! scheduled to re-run, earlier progress is preserved) and returns the error. +//! The "outage" then clears and [`CompiledGraph::retry`] restarts the run +//! from that checkpoint to completion. A run could equally be *continued on +//! user feedback* by editing state with `update_state` before `retry`. +//! +//! Run with: +//! +//! ```text +//! cargo run --example resilient_graph +//! ``` + +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use tinyagents::harness::ids::ExecutionStatus; +use tinyagents::harness::retry::RetryPolicy; +use tinyagents::{ + GraphBuilder, InMemoryCheckpointer, NodeContext, NodeResult, Result, TinyAgentsError, +}; + +/// A tiny pipeline state: how far we got, plus an audit log. +#[derive(Clone, Debug, Default)] +struct Pipeline { + fetched: bool, + committed: bool, + log: Vec, +} + +#[tokio::main] +async fn main() -> Result<()> { + let checkpointer = Arc::new(InMemoryCheckpointer::::new()); + + // A flaky `fetch`: fails (retryable model error) its first two attempts, + // then succeeds — a transient connectivity blip the node-retry policy + // absorbs on its own. + let fetch_attempts = Arc::new(AtomicUsize::new(0)); + // A `commit` that stays down until the "outage" clears between runs. The + // outer restart, not the inner retry policy, is what recovers it. + let outage = Arc::new(AtomicUsize::new(0)); + + let fa = fetch_attempts.clone(); + let og = outage.clone(); + let graph = GraphBuilder::::overwrite() + .add_node("fetch", move |mut state: Pipeline, _c: NodeContext| { + let fa = fa.clone(); + async move { + let n = fa.fetch_add(1, Ordering::SeqCst); + if n < 2 { + return Err(TinyAgentsError::Model(format!( + "connection reset (fetch attempt {})", + n + 1 + ))); + } + state.fetched = true; + state.log.push(format!("fetched on attempt {}", n + 1)); + Ok(NodeResult::Update(state)) + } + }) + .add_node("commit", move |mut state: Pipeline, _c: NodeContext| { + let og = og.clone(); + async move { + // Down while the outage flag is 0; healthy once it is raised. + if og.load(Ordering::SeqCst) == 0 { + return Err(TinyAgentsError::Tool( + "commit endpoint unreachable (outage)".into(), + )); + } + state.committed = true; + state.log.push("committed".into()); + Ok(NodeResult::Update(state)) + } + }) + .set_entry("fetch") + .add_edge("fetch", "commit") + .set_finish("commit") + .compile() + .unwrap() + // 1 try + 3 retries per node, absorbing the fetch blips. + .with_node_retry(RetryPolicy::default().with_max_attempts(4)) + .with_checkpointer(checkpointer.clone()); + + // ---- First run: fetch self-heals, but commit is fully down. ------------ + let thread = "orders-4711"; + let err = graph + .run_with_thread(thread, Pipeline::default()) + .await + .expect_err("commit is down, so the first run fails"); + println!("first run failed as expected: {err}"); + println!( + " fetch took {} attempts", + fetch_attempts.load(Ordering::SeqCst) + ); + + // The failure left a resumable checkpoint: `fetch`'s progress is committed + // and `commit` is scheduled to re-run. + let snapshot = graph + .get_state(thread, None) + .await? + .expect("a failure-boundary checkpoint exists"); + println!( + " durable state: fetched={}, committed={}", + snapshot.values.fetched, snapshot.values.committed + ); + println!(" will re-run on resume: {:?}", snapshot.next_nodes); + println!(" checkpoints on thread: {}", checkpointer.count(thread)); + + // ---- The outage clears; restart the run from the checkpoint. ----------- + outage.store(1, Ordering::SeqCst); + let resumed = graph.retry(thread).await?; + assert_eq!(resumed.status.status, ExecutionStatus::Completed); + println!("\nrestarted to completion:"); + println!( + " fetched={}, committed={}", + resumed.state.fetched, resumed.state.committed + ); + for line in &resumed.state.log { + println!(" - {line}"); + } + + Ok(()) +} diff --git a/src/graph/compiled/mod.rs b/src/graph/compiled/mod.rs index 58ec03f..01ecb92 100644 --- a/src/graph/compiled/mod.rs +++ b/src/graph/compiled/mod.rs @@ -32,11 +32,37 @@ //! the fan-in / join: lower-index branches' updates are applied first. //! - The *lowest-index* branch that errors or interrupts is the step's terminal //! outcome. Updates produced by lower-index successful branches are still -//! applied/persisted; an error aborts the run, an interrupt persists a -//! checkpoint whose pending nodes are that branch and every later active node. +//! applied/persisted; an error persists a resumable failure boundary (see +//! below) and aborts, an interrupt persists a checkpoint whose pending nodes +//! are that branch and every later active node. //! - Because branches run on cloned snapshots and never share mutable state, //! concurrency is data-race free; the reducer alone resolves conflicting //! writes (deterministically, by index). +//! +//! ## Network resilience and resumable failures +//! +//! Two opt-in mechanisms make a run durable under transient failure and +//! restartable after a hard one: +//! +//! - **Node retry.** With +//! [`CompiledGraph::with_node_retry`], a node whose handler fails with a +//! [retryable][crate::harness::retry::is_retryable] error (a model or tool +//! error — the transient class) is re-run from its start up to the policy's +//! attempt cap, emitting +//! [`GraphEvent::NodeRetryScheduled`](crate::graph::stream::GraphEvent::NodeRetryScheduled) +//! and sleeping the opt-in backoff between attempts. A single network blip is +//! absorbed without touching the run. +//! - **Resumable failure.** When a handler fails beyond the retry budget (or the +//! error is non-retryable), the executor does not discard the step. On a +//! checkpointed thread it folds the branches that already completed into +//! committed state and persists a failure-boundary checkpoint whose +//! `next_nodes` schedule the failed node (and the not-yet-run tail) for a +//! later [`CompiledGraph::resume`]/[`CompiledGraph::retry`], with the error +//! and failed node stamped into the checkpoint metadata. The run then reports +//! `Failed` (carrying that checkpoint id) and returns the error. A caller can +//! restart it as-is, or continue on operator feedback by editing state with +//! [`CompiledGraph::update_state`] before resuming. Without a checkpointer the +//! run aborts immediately, exactly as before. mod types; From 2cf67e73b7f822e07d8d18b95202f30e78223036 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 3 Jul 2026 11:45:20 -0700 Subject: [PATCH 5/7] feat(observability): opt-in model/tool payload capture for Langfuse Add PayloadCapture policy (default payload-free) that gates capturing model request/completion and tool args/result onto ModelCompleted/ToolCompleted events. The Langfuse harness exporter now populates generation/tool input/output from those fields and defaults trace metadata from run lineage. Fixes empty Input/Output panels and empty trace metadata (tinyhumansai/tinyagents#6). Claude-Session: https://claude.ai/code/session_019zY42oUyeXGxHGesQkTKWV --- src/harness/agent_loop/mod.rs | 26 ++++ src/harness/events/types.rs | 24 ++++ src/harness/observability/langfuse.rs | 124 +++++++++++++++++- src/harness/observability/test.rs | 4 + src/harness/runtime/types.rs | 53 ++++++++ src/harness/testkit/test.rs | 6 + src/harness/testkit/types.rs | 7 +- tests/e2e_orchestrator_subagents.rs | 2 + tests/e2e_registry_observability_contracts.rs | 6 + 9 files changed, 248 insertions(+), 4 deletions(-) diff --git a/src/harness/agent_loop/mod.rs b/src/harness/agent_loop/mod.rs index 4e0de75..3d2efcf 100644 --- a/src/harness/agent_loop/mod.rs +++ b/src/harness/agent_loop/mod.rs @@ -444,6 +444,14 @@ impl AgentHarness { model: binding.model, streaming, }; + // Snapshot the request messages for observability before `request` + // is moved into the model-wrap onion, gated by the capture policy so + // payload-free runs never serialize prompt text. + let captured_input = self + .policy + .capture + .model_io + .then(|| serde_json::to_value(&request.messages).unwrap_or(Value::Null)); let mut response = self .middleware .run_wrapped_model(ctx, state, request, &base) @@ -466,9 +474,16 @@ impl AgentHarness { let record = ctx.emit(AgentEvent::UsageRecorded { usage }); status.set_last_event(record.id); } + let captured_output = self + .policy + .capture + .model_io + .then(|| serde_json::to_value(&response.message).unwrap_or(Value::Null)); let record = ctx.emit(AgentEvent::ModelCompleted { call_id, usage: response.usage, + input: captured_input, + output: captured_output, }); status.set_last_event(record.id); @@ -628,6 +643,10 @@ impl AgentHarness { }); status.set_last_event(record.id); + // Snapshot the arguments for observability before `call` is + // moved into the tool-wrap onion, gated by the capture policy. + let captured_input = self.policy.capture.tool_io.then(|| call.arguments.clone()); + // The real tool call is the innermost base of the tool-wrap // onion (same before -> wrap -> after ordering as the model // path): lifecycle `before_tool` ran above, the wrap onion runs @@ -646,9 +665,16 @@ impl AgentHarness { run.tool_calls += 1; status.tool_calls = run.tool_calls; status.active_tool_calls.retain(|c| c != &tool_call_id); + let captured_output = self + .policy + .capture + .tool_io + .then(|| Value::String(result.content.clone())); let record = ctx.emit(AgentEvent::ToolCompleted { call_id: tool_call_id, tool_name, + input: captured_input, + output: captured_output, }); status.set_last_event(record.id); diff --git a/src/harness/events/types.rs b/src/harness/events/types.rs index e512411..be79030 100644 --- a/src/harness/events/types.rs +++ b/src/harness/events/types.rs @@ -75,6 +75,18 @@ pub enum AgentEvent { /// Token usage reported by the provider, when available. #[serde(default, skip_serializing_if = "Option::is_none")] usage: Option, + /// The request messages sent to the model, captured only when + /// [`PayloadCapture::model_io`][crate::harness::runtime::PayloadCapture::model_io] + /// is enabled. `None` in the default payload-free mode. Populated so an + /// exporter can render the prompt in a generation's Input panel. + #[serde(default, skip_serializing_if = "Option::is_none")] + input: Option, + /// The model completion (assistant message), captured only when + /// [`PayloadCapture::model_io`][crate::harness::runtime::PayloadCapture::model_io] + /// is enabled. `None` in the default payload-free mode. Populated so an + /// exporter can render the completion in a generation's Output panel. + #[serde(default, skip_serializing_if = "Option::is_none")] + output: Option, }, /// A tool-selection middleware filtered the model-visible tool set before a @@ -103,6 +115,18 @@ pub enum AgentEvent { call_id: CallId, /// Name of the tool that was invoked. tool_name: String, + /// The arguments the tool was invoked with, captured only when + /// [`PayloadCapture::tool_io`][crate::harness::runtime::PayloadCapture::tool_io] + /// is enabled. `None` in the default payload-free mode. Populated so an + /// exporter can render the call in a tool observation's Input panel. + #[serde(default, skip_serializing_if = "Option::is_none")] + input: Option, + /// The tool result content, captured only when + /// [`PayloadCapture::tool_io`][crate::harness::runtime::PayloadCapture::tool_io] + /// is enabled. `None` in the default payload-free mode. Populated so an + /// exporter can render the result in a tool observation's Output panel. + #[serde(default, skip_serializing_if = "Option::is_none")] + output: Option, }, /// The model called a tool that is not registered, and the run's diff --git a/src/harness/observability/langfuse.rs b/src/harness/observability/langfuse.rs index 60cf070..ba47698 100644 --- a/src/harness/observability/langfuse.rs +++ b/src/harness/observability/langfuse.rs @@ -156,6 +156,7 @@ impl LangfuseClient { .first() .map(|obs| iso_ms(obs.ts_ms)) .unwrap_or_else(|| iso_ms(now_ms())); + let metadata = trace_metadata(&trace, observations); let mut batch = Vec::with_capacity(observations.len() + 1); batch.push(json!({ @@ -172,7 +173,7 @@ impl LangfuseClient { "release": trace.release, "version": trace.version, "tags": if trace.tags.is_empty() { Value::Null } else { json!(trace.tags) }, - "metadata": if trace.metadata.is_null() { Value::Null } else { trace.metadata }, + "metadata": metadata, })), })); @@ -277,6 +278,33 @@ fn resolve_trace_id( }) } +/// Builds the trace-level metadata, defaulting useful run-lineage coordinates +/// (root/first run ids and thread, mirroring what the graph exporter folds in) +/// so a harness trace is correlatable even when the caller passes no metadata. +/// Any caller-supplied [`LangfuseTraceConfig::metadata`] keys are merged on top +/// and win on collision. Returns [`Value::Null`] only when there is nothing to +/// attach, so [`clean_nulls`] drops the field entirely. +fn trace_metadata(trace: &LangfuseTraceConfig, observations: &[AgentObservation]) -> Value { + let mut metadata = serde_json::Map::new(); + if let Some(first) = observations.first() { + metadata.insert("root_run_id".to_string(), json!(first.root_run_id.as_str())); + metadata.insert("run_id".to_string(), json!(first.run_id.as_str())); + if let Some(parent) = &first.parent_run_id { + metadata.insert("parent_run_id".to_string(), json!(parent.as_str())); + } + } + if let Value::Object(extra) = &trace.metadata { + for (k, v) in extra { + metadata.insert(k.clone(), v.clone()); + } + } + if metadata.is_empty() { + Value::Null + } else { + Value::Object(metadata) + } +} + fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value { let timestamp = iso_ms(obs.ts_ms); let metadata = json!({ @@ -287,7 +315,12 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value { "event": obs.event, }); match &obs.event { - AgentEvent::ModelCompleted { call_id, usage } => json!({ + AgentEvent::ModelCompleted { + call_id, + usage, + input, + output, + } => json!({ "id": obs.event_id.as_str(), "timestamp": timestamp, "type": "generation-create", @@ -298,10 +331,17 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value { "startTime": timestamp, "endTime": timestamp, "usage": usage.map(langfuse_usage), + "input": input, + "output": output, "metadata": metadata, })), }), - AgentEvent::ToolCompleted { call_id, tool_name } => json!({ + AgentEvent::ToolCompleted { + call_id, + tool_name, + input, + output, + } => json!({ "id": obs.event_id.as_str(), "timestamp": timestamp, "type": "tool-create", @@ -311,6 +351,8 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value { "name": tool_name, "startTime": timestamp, "endTime": timestamp, + "input": input, + "output": output, "metadata": metadata, })), }), @@ -469,6 +511,8 @@ mod tests { total_tokens: 7, ..Default::default() }), + input: None, + output: None, }, ), ], @@ -479,8 +523,82 @@ mod tests { assert_eq!(events[0]["type"], "trace-create"); assert_eq!(events[0]["body"]["id"], "root-1"); assert_eq!(events[0]["body"]["userId"], "user-1"); + // Trace metadata is defaulted from run lineage even without a caller value. + assert_eq!(events[0]["body"]["metadata"]["root_run_id"], "root-1"); + assert_eq!(events[0]["body"]["metadata"]["run_id"], "run-1"); assert_eq!(events[2]["type"], "generation-create"); assert_eq!(events[2]["body"]["id"], "model-call"); assert_eq!(events[2]["body"]["usage"]["input"], 3); + // Payload-free generation: no input/output body fields. + assert!(events[2]["body"].get("input").is_none()); + assert!(events[2]["body"].get("output").is_none()); + } + + #[test] + fn populates_generation_and_tool_io_when_captured() { + let client = + LangfuseClient::proxy("https://backend.test/telemetry/langfuse/ingestion", "t") + .unwrap(); + let batch = client + .build_ingestion_batch( + LangfuseTraceConfig::default(), + &[ + obs( + 0, + AgentEvent::ModelCompleted { + call_id: CallId::new("model-call"), + usage: None, + input: Some(json!([{ "role": "user", "content": "hi" }])), + output: Some(json!({ "content": "hello there" })), + }, + ), + obs( + 1, + AgentEvent::ToolCompleted { + call_id: CallId::new("tool-call"), + tool_name: "lookup".to_string(), + input: Some(json!({ "query": "weather" })), + output: Some(json!("sunny")), + }, + ), + ], + ) + .unwrap(); + + let events = batch["batch"].as_array().unwrap(); + assert_eq!(events[1]["type"], "generation-create"); + assert_eq!(events[1]["body"]["input"][0]["content"], "hi"); + assert_eq!(events[1]["body"]["output"]["content"], "hello there"); + assert_eq!(events[2]["type"], "tool-create"); + assert_eq!(events[2]["body"]["input"]["query"], "weather"); + assert_eq!(events[2]["body"]["output"], "sunny"); + } + + #[test] + fn merges_caller_trace_metadata_over_defaults() { + let client = + LangfuseClient::proxy("https://backend.test/telemetry/langfuse/ingestion", "t") + .unwrap(); + let batch = client + .build_ingestion_batch( + LangfuseTraceConfig { + metadata: json!({ "deployment": "prod", "root_run_id": "override" }), + ..Default::default() + }, + &[obs( + 0, + AgentEvent::RunStarted { + run_id: RunId::new("run-1"), + thread_id: None, + }, + )], + ) + .unwrap(); + + let meta = &batch["batch"].as_array().unwrap()[0]["body"]["metadata"]; + assert_eq!(meta["deployment"], "prod"); + // Caller keys win on collision with the defaulted lineage. + assert_eq!(meta["root_run_id"], "override"); + assert_eq!(meta["run_id"], "run-1"); } } diff --git a/src/harness/observability/test.rs b/src/harness/observability/test.rs index 1d44ed3..ae378cf 100644 --- a/src/harness/observability/test.rs +++ b/src/harness/observability/test.rs @@ -117,6 +117,8 @@ fn agent_latency_metrics_include_model_tool_and_run_elapsed() { AgentEvent::ModelCompleted { call_id: model_id.clone(), usage: None, + input: None, + output: None, }, ), obs( @@ -133,6 +135,8 @@ fn agent_latency_metrics_include_model_tool_and_run_elapsed() { AgentEvent::ToolCompleted { call_id: tool_id.clone(), tool_name: "lookup".to_string(), + input: None, + output: None, }, ), obs( diff --git a/src/harness/runtime/types.rs b/src/harness/runtime/types.rs index 55fb792..18c7893 100644 --- a/src/harness/runtime/types.rs +++ b/src/harness/runtime/types.rs @@ -73,6 +73,53 @@ pub enum UnknownToolPolicy { }, } +/// Controls whether the agent loop captures model and tool **payloads** +/// (prompt/completion text, tool arguments/results) onto the +/// [`AgentEvent::ModelCompleted`][crate::harness::events::AgentEvent::ModelCompleted] +/// and [`AgentEvent::ToolCompleted`][crate::harness::events::AgentEvent::ToolCompleted] +/// events it emits. +/// +/// The observability layer is **payload-free by default**: events carry only +/// ids, counters, and usage so privacy-sensitive deployments never journal or +/// export prompt text or tool I/O. Opt in per-family here to surface the request +/// messages + completion (`model_io`) and tool arguments + result (`tool_io`) so +/// downstream exporters — notably the Langfuse exporter — can populate the +/// Input/Output panels of a generation or tool observation. +/// +/// Captured payloads flow through the same event pipeline as everything else, so +/// a [`RedactingSink`][crate::harness::observability::RedactingSink] configured +/// with secret substrings still masks them before they reach a journal or +/// exporter. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct PayloadCapture { + /// Capture the request messages and the model completion onto every + /// [`AgentEvent::ModelCompleted`][crate::harness::events::AgentEvent::ModelCompleted]. + pub model_io: bool, + /// Capture the tool arguments and the tool result onto every + /// [`AgentEvent::ToolCompleted`][crate::harness::events::AgentEvent::ToolCompleted]. + pub tool_io: bool, +} + +impl PayloadCapture { + /// A capture policy with both model and tool I/O enabled. + /// + /// Prefer this only when the observability pipeline is trusted (or a + /// [`RedactingSink`][crate::harness::observability::RedactingSink] is in + /// place), since it journals and can export full prompt/completion text and + /// tool arguments/results. + pub const fn all() -> Self { + Self { + model_io: true, + tool_io: true, + } + } + + /// `true` when neither model nor tool payloads are captured (the default). + pub const fn is_disabled(&self) -> bool { + !self.model_io && !self.tool_io + } +} + #[derive(Clone, Debug, PartialEq)] pub struct RunPolicy { /// Hard run limits enforced fail-closed by the agent loop. @@ -85,6 +132,11 @@ pub struct RunPolicy { pub fallback: Option, /// Response format attached to every model request when set. pub default_response_format: Option, + /// Whether the loop captures model/tool payloads onto completion events. + /// + /// Defaults to [`PayloadCapture::default`] (payload-free), preserving the + /// privacy-preserving behavior where events carry only ids and usage. + pub capture: PayloadCapture, /// Default caching policy for the run. /// /// The loop consults [`CachePolicy::response_cache_enabled`] only when a @@ -102,6 +154,7 @@ impl Default for RunPolicy { retry: RetryPolicy::default(), fallback: None, default_response_format: None, + capture: PayloadCapture::default(), // Caching defaults ON, but is gated by an attached `ResponseCache`, // so a harness with no cache never caches regardless of this flag. cache: CachePolicy { diff --git a/src/harness/testkit/test.rs b/src/harness/testkit/test.rs index 6f561aa..2fadd59 100644 --- a/src/harness/testkit/test.rs +++ b/src/harness/testkit/test.rs @@ -327,6 +327,8 @@ fn make_trajectory() -> Vec { AgentEvent::ModelCompleted { call_id: CallId::new("c1"), usage: None, + input: None, + output: None, }, AgentEvent::ToolStarted { call_id: CallId::new("t1"), @@ -335,6 +337,8 @@ fn make_trajectory() -> Vec { AgentEvent::ToolCompleted { call_id: CallId::new("t1"), tool_name: "search".into(), + input: None, + output: None, }, AgentEvent::ModelStarted { call_id: CallId::new("c2"), @@ -343,6 +347,8 @@ fn make_trajectory() -> Vec { AgentEvent::ModelCompleted { call_id: CallId::new("c2"), usage: None, + input: None, + output: None, }, AgentEvent::RunCompleted { run_id: RunId::new("r1"), diff --git a/src/harness/testkit/types.rs b/src/harness/testkit/types.rs index 36ef3bf..35fab91 100644 --- a/src/harness/testkit/types.rs +++ b/src/harness/testkit/types.rs @@ -253,7 +253,12 @@ pub struct EventRecorder { /// let events = vec![ /// AgentEvent::RunStarted { run_id: RunId::new("r1"), thread_id: None }, /// AgentEvent::ModelStarted { call_id: CallId::new("c1"), model: "gpt".into() }, -/// AgentEvent::ModelCompleted { call_id: CallId::new("c1"), usage: None }, +/// AgentEvent::ModelCompleted { +/// call_id: CallId::new("c1"), +/// usage: None, +/// input: None, +/// output: None, +/// }, /// AgentEvent::RunCompleted { run_id: RunId::new("r1") }, /// ]; /// let traj = Trajectory::from_events(events); diff --git a/tests/e2e_orchestrator_subagents.rs b/tests/e2e_orchestrator_subagents.rs index c985b18..554dd0e 100644 --- a/tests/e2e_orchestrator_subagents.rs +++ b/tests/e2e_orchestrator_subagents.rs @@ -171,6 +171,8 @@ async fn orchestrator_resolves_and_runs_only_the_chosen_subagents() -> Result<() sink.emit(AgentEvent::ToolCompleted { call_id: CallId::new(call_id), tool_name: name.clone(), + input: None, + output: None, }); Ok::<(String, String), TinyAgentsError>((name, result.content)) } diff --git a/tests/e2e_registry_observability_contracts.rs b/tests/e2e_registry_observability_contracts.rs index f27d78d..d46731a 100644 --- a/tests/e2e_registry_observability_contracts.rs +++ b/tests/e2e_registry_observability_contracts.rs @@ -171,6 +171,8 @@ fn component_metadata_and_event_kinds_are_stable_serializable_contracts() { AgentEvent::ModelCompleted { call_id: CallId::new("model-1"), usage: None, + input: None, + output: None, }, AgentEvent::ToolStarted { call_id: CallId::new("tool-1"), @@ -179,6 +181,8 @@ fn component_metadata_and_event_kinds_are_stable_serializable_contracts() { AgentEvent::ToolCompleted { call_id: CallId::new("tool-1"), tool_name: "lookup".into(), + input: None, + output: None, }, AgentEvent::StateUpdate, AgentEvent::MiddlewareStarted { name: "mw".into() }, @@ -300,6 +304,8 @@ async fn event_sinks_journals_and_status_stores_preserve_run_lineage() { journal.append(AgentEvent::ToolCompleted { call_id: CallId::new("tool-1"), tool_name: "lookup".into(), + input: None, + output: None, }); assert_eq!(journal.len(), 2); assert_eq!(journal.replay_from(1)[0].event.kind(), "tool.completed"); From e7f696879c78bb4634a1910ae9a7853efca49b22 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 3 Jul 2026 11:46:51 -0700 Subject: [PATCH 6/7] test(observability): e2e assertion for captured Langfuse generation/tool I/O Drive a real harness run with PayloadCapture enabled, journal it, and assert the exported Langfuse batch populates generation Input/Output and tool-create Input/Output, plus defaulted trace metadata (tinyhumansai/tinyagents#6). Claude-Session: https://claude.ai/code/session_019zY42oUyeXGxHGesQkTKWV --- tests/e2e_observability.rs | 89 +++++++++++++++++++++++++++++++++++++- 1 file changed, 88 insertions(+), 1 deletion(-) diff --git a/tests/e2e_observability.rs b/tests/e2e_observability.rs index b310ec2..b822dfd 100644 --- a/tests/e2e_observability.rs +++ b/tests/e2e_observability.rs @@ -27,7 +27,7 @@ use tinyagents::harness::ids::{ExecutionStatus, RunId}; use tinyagents::harness::message::{AssistantMessage, ContentBlock, Message}; use tinyagents::harness::model::ModelResponse; use tinyagents::harness::providers::MockModel; -use tinyagents::harness::runtime::AgentHarness; +use tinyagents::harness::runtime::{AgentHarness, PayloadCapture, RunPolicy}; use tinyagents::harness::testkit::FakeTool; use tinyagents::harness::tool::ToolCall; use tinyagents::harness::usage::Usage; @@ -184,6 +184,93 @@ async fn harness_run_journals_redacted_replayable_events() { assert_eq!(tail.first().unwrap().offset, 2); } +#[tokio::test] +async fn harness_run_with_capture_exports_generation_and_tool_io() { + // With PayloadCapture enabled, a real harness run journals model/tool + // payloads on its completion events; feeding those durable observations + // through the Langfuse exporter populates the generation Input/Output and + // the tool-create Input/Output panels — the fix for tinyhumansai/tinyagents#6. + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness + .register_model( + "capture-model", + Arc::new(MockModel::with_responses(vec![ + tool_call_response("call-1", "lookup", json!({ "q": "weather" })), + text_response("all done"), + ])), + ) + .set_default_model("capture-model") + .register_tool(Arc::new(FakeTool::returning("lookup", "tool-output"))) + .with_policy(RunPolicy { + capture: PayloadCapture::all(), + ..Default::default() + }); + + let journal: Arc = Arc::new(InMemoryEventJournal::new()); + let run_id = RunId::new("run-capture"); + let ctx: RunContext<()> = RunContext::new(RunConfig::new(run_id.as_str()), ()); + ctx.events + .subscribe(Arc::new(JournalSink::new(journal.clone(), run_id.clone()))); + + harness + .invoke_in_context_with_status(&(), ctx, vec![Message::user("please look up")]) + .await + .expect("run succeeds"); + + let observations = journal + .read_from(run_id.as_str(), 0) + .await + .expect("replay journal"); + + // The captured payloads survive on the durable observations. + let model_completed = observations + .iter() + .find(|o| matches!(o.event, AgentEvent::ModelCompleted { .. })) + .expect("a model completion is journaled"); + match &model_completed.event { + AgentEvent::ModelCompleted { input, output, .. } => { + assert!(input.is_some(), "captured model input rides the event"); + assert!(output.is_some(), "captured model output rides the event"); + } + _ => unreachable!(), + } + + // Export through the harness Langfuse client and assert the body carries I/O. + let client = LangfuseClient::proxy("https://backend.test", "tok").expect("client"); + let batch = client + .build_ingestion_batch(LangfuseTraceConfig::default(), &observations) + .expect("build batch"); + let events = batch["batch"].as_array().expect("batch array"); + + // Trace metadata is defaulted from the run lineage even with no caller value. + assert_eq!(events[0]["type"], "trace-create"); + assert_eq!(events[0]["body"]["metadata"]["root_run_id"], "run-capture"); + + let generations: Vec<_> = events + .iter() + .filter(|e| e["type"] == "generation-create") + .collect(); + assert_eq!(generations.len(), 2, "one generation per model call"); + assert!( + generations + .iter() + .all(|g| !g["body"]["input"].is_null() && !g["body"]["output"].is_null()), + "every generation Input/Output panel is populated" + ); + // The final generation's completion carries the answer text. + assert_eq!( + generations[1]["body"]["output"]["content"][0]["text"], + "all done" + ); + + let tool = events + .iter() + .find(|e| e["type"] == "tool-create") + .expect("a tool observation"); + assert_eq!(tool["body"]["input"]["q"], "weather"); + assert_eq!(tool["body"]["output"], "tool-output"); +} + /// A two-node line graph over `i32` with overwrite semantics: `a -> b`. fn line_graph() -> tinyagents::CompiledGraph { GraphBuilder::::overwrite() From 0149256710564aa04cf1ee2ac5b25f2f7cd64693 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 3 Jul 2026 11:49:24 -0700 Subject: [PATCH 7/7] docs(observability): document opt-in payload capture and defaulted trace metadata Claude-Session: https://claude.ai/code/session_019zY42oUyeXGxHGesQkTKWV --- docs/modules/harness/observability.md | 29 +++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/docs/modules/harness/observability.md b/docs/modules/harness/observability.md index cc1abdb..fbd436c 100644 --- a/docs/modules/harness/observability.md +++ b/docs/modules/harness/observability.md @@ -320,6 +320,35 @@ Redaction policies should handle: Redacted events should preserve structural fields such as ids, event kinds, timings, and counters so traces remain useful. +## Payload Capture + +The event stream is **payload-free by default**: `AgentEvent::ModelCompleted` +and `AgentEvent::ToolCompleted` carry only ids and usage, never the prompt, +completion, tool arguments, or tool result. This keeps journals and exporters +safe for privacy-sensitive deployments without any configuration. + +`RunPolicy::capture` (a `PayloadCapture { model_io, tool_io }`) opts a run into +carrying those payloads onto the completion events: + +- `model_io` — attaches the request messages (`input`) and the model completion + (`output`) to every `ModelCompleted`. +- `tool_io` — attaches the tool arguments (`input`) and the tool result + (`output`) to every `ToolCompleted`. + +Capture is snapshotted inside the agent loop before the request/call value is +moved into the model or tool wrap onion, so it never perturbs execution. Because +captured payloads ride the ordinary event pipeline, a `RedactingSink` still +masks them before they reach a journal or exporter, and the durable journal +replays them verbatim. + +Downstream, the Langfuse exporter reads these fields to populate the +Input/Output panels of a generation (`generation-create`) and a tool +observation (`tool-create`). With capture disabled the fields are absent and the +observation renders with ids, timing, and usage only. The harness exporter also +defaults the trace-level metadata from the run lineage (root/first run ids and +parent), mirroring the graph exporter, so a trace is correlatable even when the +caller passes no metadata. + ## Graph Events State-graph event payloads should be rich enough for a UI to render run