diff --git a/src/openhuman/agent_orchestration/delegation.rs b/src/openhuman/agent_orchestration/delegation.rs index 0f05a14c2..33d1f620a 100644 --- a/src/openhuman/agent_orchestration/delegation.rs +++ b/src/openhuman/agent_orchestration/delegation.rs @@ -23,7 +23,8 @@ use crate::openhuman::agent::harness::subagent_runner::{run_subagent, SubagentRu use crate::openhuman::agent_orchestration::parent_context::build_root_parent; use crate::openhuman::config::Config; use crate::openhuman::tinyagents::delegation::{ - run_delegation, DelegationConfig, DelegationStage, DelegationStageOutput, DelegationState, + run_or_resume_delegation, DelegationConfig, DelegationStage, DelegationStageOutput, + DelegationState, }; use tinyagents::graph::checkpoint::Checkpointer; use tinyagents::graph::SqliteCheckpointer; @@ -107,6 +108,10 @@ pub(crate) async fn run_subagent_delegation( Ok(DelegationStageOutput { text: outcome.output, approved, + // Persist the exact prompt this stage sent so the + // execute step's `StepRecord` carries its provenance + // (read only for the execute stage; #3884). + prompt: Some(prompt), }) } Err(e) => Err(format!("delegation stage {stage:?} failed: {e}")), @@ -125,7 +130,12 @@ pub(crate) async fn run_subagent_delegation( // stays off here until a human-review delegation surface wires it. ..DelegationConfig::default() }; - run_delegation(delegation_config, run_stage).await + // Resume-aware entry (#3884): with today's fresh-per-run `thread_id` this + // is always a fresh run; reusing a stable `thread_id` resumes from the + // last checkpoint boundary instead of restarting. + run_or_resume_delegation(delegation_config, run_stage) + .await + .map(|outcome| outcome.state) }; if current_parent().is_some() { @@ -160,11 +170,7 @@ fn build_stage_prompt(stage: DelegationStage, task: &str, state: &DelegationStat ) } DelegationStage::Review => { - let result = state - .executions - .last() - .map(String::as_str) - .unwrap_or("(no execution produced)"); + let result = state.last_result().unwrap_or("(no execution produced)"); format!( "Review the result below against the task. If it fully and correctly \ accomplishes the task, reply with `APPROVE` on the first line. Otherwise \ diff --git a/src/openhuman/tinyagents/delegation.rs b/src/openhuman/tinyagents/delegation.rs index 9654e9327..67dc6688d 100644 --- a/src/openhuman/tinyagents/delegation.rs +++ b/src/openhuman/tinyagents/delegation.rs @@ -32,7 +32,7 @@ use std::sync::Arc; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; -use tinyagents::graph::checkpoint::Checkpointer; +use tinyagents::graph::checkpoint::{Checkpoint, Checkpointer}; use tinyagents::graph::export::GraphTopology; use tinyagents::graph::recursion::RecursionPolicy; use tinyagents::graph::ClosureStateReducer; @@ -61,26 +61,58 @@ pub(crate) struct DelegationStageOutput { /// Only meaningful for [`DelegationStage::Review`]: `true` approves the /// execution and ends the loop; `false` requests another revision. pub(crate) approved: bool, + /// The exact prompt handed to this stage's worker, when it surfaces one. + /// Persisted into [`StepRecord::prompt`] for per-step provenance (read only + /// for the execute stage; ignored elsewhere). `None` when the worker does not + /// surface a prompt — e.g. the deterministic test mock. + pub(crate) prompt: Option, } impl DelegationStageOutput { - /// A plain non-review stage output (the `approved` flag is unused). + /// A plain non-review stage output (the `approved` flag is unused and no + /// prompt is surfaced). pub(crate) fn done(text: impl Into) -> Self { Self { text: text.into(), approved: true, + prompt: None, } } } +/// Current on-disk schema version for a checkpointed [`DelegationState`]. Bumped +/// only on a breaking state-shape change — introduced with the +/// `executions: Vec` → `Vec` migration (issue #3884). +/// Pre-versioned records deserialize to `0` via `#[serde(default)]`, so a resume +/// can tell a stale checkpoint from a current one. +pub(crate) const CURRENT_SCHEMA_VERSION: u32 = 1; + +/// One completed execute-stage pass, recorded durably so a resumed run knows +/// exactly how far it got and can render/finalize per step rather than from a +/// flat text log. Replaces the former `executions: Vec` (issue #3884). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub(crate) struct StepRecord { + /// 0-based execute pass: `0` is the first execution, `n` the n-th revision. + pub(crate) index: usize, + /// The exact prompt handed to the execute sub-agent for this pass — per-step + /// provenance and the seam a later plan-edit slice (#3881) diffs against. + /// Empty when the worker did not surface one. + #[serde(default)] + pub(crate) prompt: String, + /// The sub-agent's result text — the value the former `Vec` entry held. + pub(crate) result: String, +} + /// Typed working state threaded through (and checkpointed across) the delegation /// graph. Serde-serializable so a [`Checkpointer`] can persist and restore it. #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub(crate) struct DelegationState { /// The plan produced by the `plan` stage. pub(crate) plan: Option, - /// One entry per execution pass (the first plus each revision). - pub(crate) executions: Vec, + /// One record per execution pass (the first plus each revision), typed so a + /// resumed run can render/finalize per step (widened from `Vec`, + /// issue #3884). + pub(crate) executions: Vec, /// One entry per review pass. pub(crate) reviews: Vec, /// Number of revisions the reviewer requested (loops back to `execute`). @@ -100,12 +132,47 @@ pub(crate) struct DelegationState { /// (deny semantics: block the action, finalize as denied). #[serde(default)] pub(crate) denied: bool, + /// On-disk schema version, stamped [`CURRENT_SCHEMA_VERSION`] on a fresh run + /// and defaulting to `0` for pre-versioned checkpoints. + /// [`run_or_resume_delegation`] expires any checkpoint whose version is below + /// `CURRENT_SCHEMA_VERSION` (and any that fails to deserialize) instead of + /// resuming or returning it — so a shape change that stays structurally + /// decodable is still not misread. + #[serde(default)] + pub(crate) schema_version: u32, +} + +impl DelegationState { + /// A fresh run's initial state, stamped with the current schema version so + /// its checkpoints are self-identifying. + fn new_run() -> Self { + Self { + schema_version: CURRENT_SCHEMA_VERSION, + ..Self::default() + } + } + + /// The latest execution result text, if any — the projection the review + /// prompt and the finalize summary read. + pub(crate) fn last_result(&self) -> Option<&str> { + self.executions.last().map(|r| r.result.as_str()) + } + + /// The execution result texts in order — the flat projection used for the + /// durable approval-interrupt payload, kept `Vec<&str>` so that wire shape + /// is unchanged from the pre-#3884 `Vec`. + pub(crate) fn executions_texts(&self) -> Vec<&str> { + self.executions.iter().map(|r| r.result.as_str()).collect() + } } /// Reducer updates emitted by the delegation nodes. enum DelegationUpdate { Plan(String), - Execution(String), + Execution { + prompt: String, + result: String, + }, Review { note: String, approved: bool, @@ -243,9 +310,10 @@ where "[delegation] running sub-agent delegation graph" ); + let initial = DelegationState::new_run(); let execution = match thread_id.clone() { - Some(tid) => graph.run_with_thread(tid, DelegationState::default()).await, - None => graph.run(DelegationState::default()).await, + Some(tid) => graph.run_with_thread(tid, initial).await, + None => graph.run(initial).await, } .map_err(|e| format!("delegation graph run failed: {e}"))?; @@ -271,6 +339,183 @@ pub(crate) async fn resume_delegation( decision: Value, run_stage: F, ) -> Result +where + F: Fn(DelegationStage, DelegationState) -> Fut + Clone + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + let approved = decision_is_approve(&decision); + tracing::info!( + approved, + "[interrupt] resuming durable delegation graph with approval decision" + ); + let mut command = Command::default(); + command.resume = Some(decision); + resume_graph(config, command, run_stage).await +} + +/// Run the delegation graph, resuming from the last checkpoint boundary when the +/// configured thread has a live, compatible, non-terminal checkpoint, else +/// starting fresh (issue #3884 — node-level checkpoint & resume). +/// +/// Classifies the thread's latest checkpoint and routes accordingly: +/// - **resumable** (a crash/failure left a mid-run boundary) → re-run only the +/// not-yet-completed nodes from that boundary via [`CompiledGraph::resume`] +/// with an empty command — never restarting from `plan`, and never re-running +/// an already-completed step (its [`StepRecord`] is restored from the state); +/// - **terminal** (already finalized/cancelled) → return the stored final state +/// without re-running (idempotent re-invocation of a stable thread); +/// - **absent** (no checkpoint) → a fresh durable run; +/// - **incompatible** (an undecodable record — e.g. a pre-#3884 `Vec` +/// `executions` shape) → log, best-effort prune, and start fresh. The decode +/// error is swallowed into a fresh-run decision, never propagated, never a panic. +/// +/// Callers that mint a unique `thread_id` per run (today's default) always take +/// the fresh path unchanged; the resume paths activate only when a caller reuses +/// a stable `thread_id`, so this is byte-compatible for existing callers while +/// wiring the resume seam that #3881 (plan-edit resume) builds on. +pub(crate) async fn run_or_resume_delegation( + config: DelegationConfig, + run_stage: F, +) -> Result +where + F: Fn(DelegationStage, DelegationState) -> Fut + Clone + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + // Without a checkpointer + thread there is nothing to resume from. + let (Some(cp), Some(tid)) = (config.checkpointer.clone(), config.thread_id.clone()) else { + return run_delegation_durable(config, run_stage).await; + }; + + match cp.get(tid.as_str(), None).await { + // A checkpoint written under an older state schema (e.g. a pre-#3884 + // record whose `executions` happened to be empty and so still decoded + // into `Vec`) is expired rather than resumed/returned — its + // semantics may not match the current graph. This is what makes + // `schema_version` an actual guard, not just documentation, and closes + // the empty-`executions` gap that a decode failure alone cannot catch. + Ok(Some(checkpoint)) if checkpoint.state.schema_version < CURRENT_SCHEMA_VERSION => { + tracing::warn!( + thread_id = %tid, + schema_version = checkpoint.state.schema_version, + current = CURRENT_SCHEMA_VERSION, + "[delegation] checkpoint predates the current state schema; pruning and starting fresh" + ); + prune_thread(cp.as_ref(), &tid).await; + run_delegation_durable(config, run_stage).await + } + Ok(Some(checkpoint)) if checkpoint_is_resumable(&checkpoint) => { + tracing::info!( + thread_id = %tid, + "[delegation] resuming durable delegation from its last checkpoint boundary" + ); + // A crash/failure resume carries no decision value — an empty command + // simply re-runs the pending node(s) (the crate's `retry` semantics). + resume_graph(config, Command::default(), run_stage).await + } + Ok(Some(checkpoint)) => { + // Terminal: return the finalized state without re-running. Defensive: + // if this checkpoint still carried an unconsumed interrupt, surface it + // instead of silently dropping it. The current routing never produces + // this (an interrupt boundary schedules its node and is classified + // resumable above), but a future schedule could, and a dropped pause + // would strand a run. + let pending = checkpoint.interrupts.first().map(|i| PendingApproval { + interrupt_id: i.id.clone(), + node: i.node.as_str().to_string(), + payload: i.payload.clone(), + thread_id: tid.clone(), + }); + if pending.is_some() { + tracing::warn!( + thread_id = %tid, + "[delegation] terminal-classified checkpoint carried a pending interrupt; surfacing it" + ); + } else { + tracing::info!( + thread_id = %tid, + "[delegation] thread already terminal; returning finalized state without re-running" + ); + } + Ok(DelegationOutcome { + state: checkpoint.state, + pending, + }) + } + Ok(None) => { + tracing::debug!( + thread_id = %tid, + "[delegation] no checkpoint for thread; starting a fresh durable run" + ); + run_delegation_durable(config, run_stage).await + } + // Only a *decode / shape-incompatibility* read error expires the + // checkpoint. An operational error (SQLite busy / I/O / poisoned lock) + // must NOT silently restart a valid resumable run — it is propagated so + // durable work is retried by the caller, not dropped. + Err(e) if is_incompatible_checkpoint_error(&e) => { + tracing::warn!( + thread_id = %tid, + error = %e, + "[delegation] undecodable/incompatible checkpoint; pruning and starting fresh" + ); + prune_thread(cp.as_ref(), &tid).await; + run_delegation_durable(config, run_stage).await + } + Err(e) => { + tracing::error!( + thread_id = %tid, + error = %e, + "[delegation] checkpoint read failed (operational); not restarting — propagating error" + ); + Err(format!( + "delegation checkpoint read failed for thread {tid}: {e}" + )) + } + } +} + +/// Best-effort prune of a dead/expired checkpoint thread so it is not re-probed +/// forever. Failure to prune is non-fatal (logged at debug). +async fn prune_thread(cp: &dyn Checkpointer, thread_id: &str) { + if let Err(e) = cp.delete_thread(thread_id).await { + tracing::debug!( + thread_id = %thread_id, + error = %e, + "[delegation] could not prune checkpoint thread (non-fatal)" + ); + } +} + +/// Whether a `Checkpointer::get` error is a decode / shape-incompatibility (safe +/// to expire the checkpoint) rather than an operational failure (SQLite busy / +/// I/O / poisoned lock — must not silently restart durable work). The vendored +/// `SqliteCheckpointer` reports both as `TinyAgentsError::Checkpoint(String)` but +/// tags decode failures with a `"decode …"` context (`sqlite.rs`: +/// `decode record` / `decode namespace` / `decode next_nodes`) — the only stable +/// discriminator it exposes. +fn is_incompatible_checkpoint_error(e: &tinyagents::TinyAgentsError) -> bool { + matches!(e, tinyagents::TinyAgentsError::Checkpoint(msg) if msg.contains("decode")) +} + +/// Whether a loaded checkpoint still has work to resume: a non-finalized, +/// non-cancelled run that still schedules a real (non-`END`) node. +fn checkpoint_is_resumable(checkpoint: &Checkpoint) -> bool { + if checkpoint.state.final_output.is_some() || checkpoint.state.cancelled { + return false; + } + checkpoint.next_nodes.iter().any(|n| n.as_str() != END) +} + +/// Rebuild the delegation graph (its node closures are not serializable — only +/// the typed state is checkpointed) with the same checkpointer + `thread_id`, and +/// re-enter it at the latest checkpoint's pending node(s) via +/// [`CompiledGraph::resume`]. `command` carries an approver's decision for the +/// human-approval interrupt path, or is empty for a plain crash/failure resume. +async fn resume_graph( + config: DelegationConfig, + command: Command, + run_stage: F, +) -> Result where F: Fn(DelegationStage, DelegationState) -> Fut + Clone + Send + Sync + 'static, Fut: Future> + Send + 'static, @@ -295,16 +540,6 @@ where ))) .with_checkpointer(cp); - let approved = decision_is_approve(&decision); - tracing::info!( - thread_id = %thread_id, - approved, - "[interrupt] resuming durable delegation graph with approval decision" - ); - - let mut command = Command::default(); - command.resume = Some(decision); - let execution = graph .resume(thread_id.clone(), command) .await @@ -385,7 +620,14 @@ where ClosureStateReducer::new(|mut s: DelegationState, u: DelegationUpdate| { match u { DelegationUpdate::Plan(p) => s.plan = Some(p), - DelegationUpdate::Execution(e) => s.executions.push(e), + DelegationUpdate::Execution { prompt, result } => { + let index = s.executions.len(); + s.executions.push(StepRecord { + index, + prompt, + result, + }); + } DelegationUpdate::Review { note, approved } => { s.reviews.push(note); s.approved = approved; @@ -454,7 +696,10 @@ where .map_err(to_node_err)?; Ok(NodeResult::Command( Command::default() - .with_update(DelegationUpdate::Execution(out.text)) + .with_update(DelegationUpdate::Execution { + prompt: out.prompt.unwrap_or_default(), + result: out.text, + }) .with_goto(["review"]), )) } @@ -527,7 +772,7 @@ where let payload = json!({ "kind": "delegation_review", "reviews": s.reviews, - "executions": s.executions, + "executions": s.executions_texts(), "revisions": s.revisions, }); tracing::info!( @@ -564,7 +809,7 @@ where let summary = s .executions .last() - .cloned() + .map(|r| r.result.clone()) .unwrap_or_else(|| "".to_string()); let final_text = if s.cancelled { format!("cancelled after {} execution(s)", s.executions.len()) @@ -672,6 +917,7 @@ mod tests { Ok(DelegationStageOutput { text: format!("review-{n}"), approved: n >= reject_first, + prompt: None, }) } } @@ -686,6 +932,8 @@ mod tests { .expect("runs"); assert_eq!(state.plan.as_deref(), Some("PLAN")); assert_eq!(state.executions.len(), 1, "one execution, no revision"); + assert_eq!(state.executions[0].index, 0); + assert_eq!(state.executions[0].result, "EXEC"); assert_eq!(state.reviews.len(), 1); assert_eq!(state.revisions, 0); assert!(state.approved); @@ -848,4 +1096,429 @@ mod tests { "at least one super-step boundary checkpoint persisted" ); } + + /// The boxed-future type every inline test runner returns. + type BoxedStageFut = + std::pin::Pin> + Send>>; + + #[tokio::test] + async fn execution_records_capture_index_and_prompt() { + // A worker that surfaces an execute prompt; assert the per-step record + // carries index + prompt + result (issue #3884). + let runner = move |stage: DelegationStage, _s: DelegationState| { + Box::pin(async move { + match stage { + DelegationStage::Plan => Ok(DelegationStageOutput::done("PLAN")), + DelegationStage::Execute => Ok(DelegationStageOutput { + text: "EXEC".to_string(), + approved: true, + prompt: Some("EXEC-PROMPT".to_string()), + }), + DelegationStage::Review => Ok(DelegationStageOutput { + text: "APPROVE".to_string(), + approved: true, + prompt: None, + }), + } + }) as BoxedStageFut + }; + let state = run_delegation(DelegationConfig::default(), runner) + .await + .expect("runs"); + assert_eq!(state.executions.len(), 1); + assert_eq!(state.executions[0].index, 0); + assert_eq!(state.executions[0].result, "EXEC"); + assert_eq!(state.executions[0].prompt, "EXEC-PROMPT"); + } + + #[test] + fn legacy_string_executions_do_not_deserialize_into_step_records() { + // A pre-#3884 checkpoint stored `executions` as a flat string array. It + // must fail to load into the new `Vec` — which is exactly + // what makes `run_or_resume_delegation` expire the stale checkpoint + // instead of misreading it. + let legacy = r#"{"plan":"P","executions":["raw a","raw b"],"reviews":[],"revisions":0,"approved":true,"final_output":null,"cancelled":false}"#; + assert!(serde_json::from_str::(legacy).is_err()); + } + + #[test] + fn schema_version_defaults_to_zero_and_step_records_round_trip() { + // A pre-versioned record (no `schema_version`) loads as version 0. + let unversioned = r#"{"plan":null,"executions":[],"reviews":[],"revisions":0,"approved":false,"final_output":null,"cancelled":false}"#; + let s: DelegationState = serde_json::from_str(unversioned).expect("loads"); + assert_eq!(s.schema_version, 0); + + // A fresh run stamps the current version and round-trips step records. + let mut fresh = DelegationState::new_run(); + assert_eq!(fresh.schema_version, CURRENT_SCHEMA_VERSION); + fresh.executions.push(StepRecord { + index: 0, + prompt: "P".to_string(), + result: "R".to_string(), + }); + let json = serde_json::to_string(&fresh).expect("serializes"); + let back: DelegationState = serde_json::from_str(&json).expect("round-trips"); + assert_eq!(back.schema_version, CURRENT_SCHEMA_VERSION); + assert_eq!(back.executions[0].result, "R"); + assert_eq!(back.executions[0].prompt, "P"); + } + + #[tokio::test] + async fn run_or_resume_starts_fresh_without_a_checkpoint() { + let dir = tempfile::tempdir().unwrap(); + let cp: Arc> = Arc::new( + tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path()), + ); + let config = DelegationConfig { + checkpointer: Some(cp), + thread_id: Some("fresh-1".to_string()), + ..DelegationConfig::default() + }; + let outcome = run_or_resume_delegation(config, flow_runner(0)) + .await + .expect("runs fresh"); + assert!(outcome.state.final_output.is_some()); + assert_eq!(outcome.state.executions.len(), 1); + assert_eq!(outcome.state.schema_version, CURRENT_SCHEMA_VERSION); + } + + #[tokio::test] + async fn run_or_resume_continues_from_last_boundary_after_a_crash() { + let dir = tempfile::tempdir().unwrap(); + let cp: Arc> = Arc::new( + tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path()), + ); + // Count stage invocations across BOTH the crashed run and the resume, to + // prove plan/execute are NOT re-run on resume. + let plan_runs = Arc::new(AtomicUsize::new(0)); + let exec_runs = Arc::new(AtomicUsize::new(0)); + let review_runs = Arc::new(AtomicUsize::new(0)); + let make_runner = |crash_first_review: bool| { + let plan_runs = plan_runs.clone(); + let exec_runs = exec_runs.clone(); + let review_runs = review_runs.clone(); + move |stage: DelegationStage, _s: DelegationState| { + let plan_runs = plan_runs.clone(); + let exec_runs = exec_runs.clone(); + let review_runs = review_runs.clone(); + Box::pin(async move { + match stage { + DelegationStage::Plan => { + plan_runs.fetch_add(1, Ordering::SeqCst); + Ok(DelegationStageOutput::done("PLAN")) + } + DelegationStage::Execute => { + exec_runs.fetch_add(1, Ordering::SeqCst); + Ok(DelegationStageOutput { + text: "EXEC".to_string(), + approved: true, + prompt: Some("EXEC-PROMPT".to_string()), + }) + } + DelegationStage::Review => { + let n = review_runs.fetch_add(1, Ordering::SeqCst); + if crash_first_review && n == 0 { + Err("simulated crash during review".to_string()) + } else { + Ok(DelegationStageOutput { + text: "APPROVE".to_string(), + approved: true, + prompt: None, + }) + } + } + } + }) as BoxedStageFut + } + }; + let make_config = || DelegationConfig { + checkpointer: Some(cp.clone()), + thread_id: Some("resume-1".to_string()), + ..DelegationConfig::default() + }; + + // Run 1: crashes during the first review. plan + execute completed and + // were checkpointed at their super-step boundaries; the run returns Err. + let crashed = run_or_resume_delegation(make_config(), make_runner(true)).await; + assert!(crashed.is_err(), "first run crashes at review"); + assert_eq!(plan_runs.load(Ordering::SeqCst), 1); + assert_eq!(exec_runs.load(Ordering::SeqCst), 1); + + // Run 2 (resume): a resumable checkpoint exists → re-enter at the pending + // node (review), NOT plan; plan/execute must not run again. + let resumed = run_or_resume_delegation(make_config(), make_runner(false)) + .await + .expect("resumes"); + assert!( + resumed.state.final_output.is_some(), + "resumed run finalizes" + ); + assert_eq!( + plan_runs.load(Ordering::SeqCst), + 1, + "plan not re-run on resume" + ); + assert_eq!( + exec_runs.load(Ordering::SeqCst), + 1, + "execute not re-run on resume" + ); + assert_eq!( + resumed.state.executions.len(), + 1, + "the pre-crash execution survived; no duplicate" + ); + assert_eq!(resumed.state.executions[0].result, "EXEC"); + } + + #[tokio::test] + async fn run_or_resume_is_idempotent_on_a_finalized_thread() { + let dir = tempfile::tempdir().unwrap(); + let cp: Arc> = Arc::new( + tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path()), + ); + let stage_runs = Arc::new(AtomicUsize::new(0)); + let make_runner = || { + let stage_runs = stage_runs.clone(); + move |stage: DelegationStage, _s: DelegationState| { + let stage_runs = stage_runs.clone(); + Box::pin(async move { + stage_runs.fetch_add(1, Ordering::SeqCst); + match stage { + DelegationStage::Review => Ok(DelegationStageOutput { + text: "APPROVE".to_string(), + approved: true, + prompt: None, + }), + _ => Ok(DelegationStageOutput::done("X")), + } + }) as BoxedStageFut + } + }; + let make_config = || DelegationConfig { + checkpointer: Some(cp.clone()), + thread_id: Some("done-1".to_string()), + ..DelegationConfig::default() + }; + + let first = run_or_resume_delegation(make_config(), make_runner()) + .await + .expect("runs"); + assert!(first.state.final_output.is_some()); + let after_first = stage_runs.load(Ordering::SeqCst); + assert!(after_first > 0, "the first run invoked stage workers"); + + // Re-invoke the SAME thread: a terminal checkpoint → return the finalized + // state without re-running any stage worker. + let second = run_or_resume_delegation(make_config(), make_runner()) + .await + .expect("idempotent"); + assert!(second.state.final_output.is_some()); + assert_eq!( + stage_runs.load(Ordering::SeqCst), + after_first, + "no stage worker re-ran on a finalized thread" + ); + } + + #[tokio::test] + async fn incompatible_checkpoint_expires_to_a_fresh_run() { + // A pre-#3884 checkpoint (executions as `Vec`) left in the store + // must not crash a resume: `run_or_resume_delegation` expires it and runs + // fresh, never panicking or surfacing the decode error. + #[derive(Clone, Debug, Serialize, Deserialize)] + struct LegacyState { + plan: Option, + executions: Vec, + reviews: Vec, + revisions: usize, + approved: bool, + final_output: Option, + cancelled: bool, + } + let dir = tempfile::tempdir().unwrap(); + // Seed the store as the OLD state type under the thread. + let legacy_cp: tinyagents::graph::checkpoint::FileCheckpointer = + tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path()); + let legacy = Checkpoint { + thread_id: "legacy-1".to_string(), + checkpoint_id: "cp-legacy".to_string(), + run_id: None, + parent_checkpoint_id: None, + namespace: vec![], + state: LegacyState { + plan: Some("old".to_string()), + executions: vec!["a".to_string(), "b".to_string()], + reviews: vec![], + revisions: 0, + approved: true, + final_output: None, + cancelled: false, + }, + next_nodes: vec![], + completed_tasks: vec![], + pending_writes: vec![], + interrupts: vec![], + pending_activations: None, + barrier_arrivals: vec![], + metadata: json!({}), + }; + legacy_cp.put(legacy).await.expect("seed legacy checkpoint"); + + // Reopen the SAME store as the current state type and resume: the + // undecodable record is expired and a fresh run completes. + let cp: Arc> = + Arc::new(tinyagents::graph::checkpoint::FileCheckpointer::< + DelegationState, + >::new(dir.path())); + let config = DelegationConfig { + checkpointer: Some(cp), + thread_id: Some("legacy-1".to_string()), + ..DelegationConfig::default() + }; + let outcome = run_or_resume_delegation(config, flow_runner(0)) + .await + .expect("expires stale checkpoint and runs fresh"); + assert!(outcome.state.final_output.is_some(), "fresh run completed"); + assert_eq!(outcome.state.executions.len(), 1); + assert_eq!(outcome.state.executions[0].result, "EXEC"); + } + + #[tokio::test] + async fn checkpoint_below_current_schema_version_expires_to_fresh_run() { + // A decodable but OLD-schema checkpoint (schema_version defaults to 0 — + // e.g. a pre-#3884 record whose executions happened to be empty) must be + // expired, not resumed: `schema_version` is a real guard, not just a doc. + let dir = tempfile::tempdir().unwrap(); + let seed: tinyagents::graph::checkpoint::FileCheckpointer = + tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path()); + let checkpoint = Checkpoint { + thread_id: "old-schema".to_string(), + checkpoint_id: "cp-old".to_string(), + run_id: None, + parent_checkpoint_id: None, + namespace: vec![], + state: DelegationState { + plan: Some("stale".to_string()), + ..Default::default() + }, + next_nodes: vec![], + completed_tasks: vec![], + pending_writes: vec![], + interrupts: vec![], + pending_activations: None, + barrier_arrivals: vec![], + metadata: json!({}), + }; + assert_eq!( + checkpoint.state.schema_version, 0, + "an un-stamped record is version 0" + ); + seed.put(checkpoint) + .await + .expect("seed old-schema checkpoint"); + + let cp: Arc> = + Arc::new(tinyagents::graph::checkpoint::FileCheckpointer::< + DelegationState, + >::new(dir.path())); + let config = DelegationConfig { + checkpointer: Some(cp), + thread_id: Some("old-schema".to_string()), + ..DelegationConfig::default() + }; + let outcome = run_or_resume_delegation(config, flow_runner(0)) + .await + .expect("expires + fresh"); + assert!(outcome.state.final_output.is_some(), "fresh run completed"); + assert_eq!( + outcome.state.schema_version, CURRENT_SCHEMA_VERSION, + "fresh run stamped the current version" + ); + assert_eq!( + outcome.state.plan.as_deref(), + Some("PLAN"), + "re-planned from scratch, not resumed with the stale plan" + ); + } + + #[test] + fn incompatible_checkpoint_error_matches_decode_not_operational() { + use tinyagents::TinyAgentsError; + // Decode / shape-incompatibility → safe to expire. + assert!(is_incompatible_checkpoint_error( + &TinyAgentsError::Checkpoint( + "sqlite checkpointer: decode record: invalid type: string".to_string() + ) + )); + assert!(is_incompatible_checkpoint_error( + &TinyAgentsError::Checkpoint("sqlite checkpointer: decode next_nodes: eof".to_string()) + )); + // Operational failures must NOT be treated as incompatible (they must + // propagate, not silently restart durable work). + assert!(!is_incompatible_checkpoint_error( + &TinyAgentsError::Checkpoint( + "sqlite checkpointer: query latest checkpoint: database is locked".to_string() + ) + )); + assert!(!is_incompatible_checkpoint_error( + &TinyAgentsError::Checkpoint( + "sqlite checkpointer: connection lock poisoned".to_string() + ) + )); + assert!(!is_incompatible_checkpoint_error(&TinyAgentsError::Resume( + "no checkpoint".to_string() + ))); + } + + #[tokio::test] + async fn terminal_checkpoint_with_a_pending_interrupt_surfaces_it() { + // A terminal-classified checkpoint that still carries an interrupt must + // surface it, not silently drop `pending`. (The live routing never + // produces this shape; the terminal branch is defensive.) + let dir = tempfile::tempdir().unwrap(); + let seed: tinyagents::graph::checkpoint::FileCheckpointer = + tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path()); + let mut state = DelegationState::new_run(); + state.final_output = Some("done".to_string()); + let checkpoint = Checkpoint { + thread_id: "terminal-interrupt".to_string(), + checkpoint_id: "cp-ti".to_string(), + run_id: None, + parent_checkpoint_id: None, + namespace: vec![], + state, + next_nodes: vec![], + completed_tasks: vec![], + pending_writes: vec![], + interrupts: vec![Interrupt::with_id( + "intr-1", + "approval", + json!({ "kind": "delegation_review" }), + )], + pending_activations: None, + barrier_arrivals: vec![], + metadata: json!({}), + }; + seed.put(checkpoint).await.expect("seed terminal+interrupt"); + + let cp: Arc> = + Arc::new(tinyagents::graph::checkpoint::FileCheckpointer::< + DelegationState, + >::new(dir.path())); + let config = DelegationConfig { + checkpointer: Some(cp), + thread_id: Some("terminal-interrupt".to_string()), + ..DelegationConfig::default() + }; + let outcome = run_or_resume_delegation(config, flow_runner(0)) + .await + .expect("terminal"); + let pending = outcome + .pending + .expect("the carried interrupt is surfaced, not dropped"); + assert_eq!(pending.node, "approval"); + assert_eq!(pending.interrupt_id, "intr-1"); + assert_eq!(outcome.state.final_output.as_deref(), Some("done")); + } }