fix(select,deps,delegation): follow-up fixes stranded past #129's merge

Cargo.toml

@@ -112,7 +112,19 @@ chrono-tz = { version = "0.10", optional = true }
 # `STANDARD_NO_PAD`, so `harness::multimodal::data_uri` picks by length instead
 # of trying both — trying both decoded a multi-MB image twice for every
 # unpadded payload.
-base64 = { version = "0.23", optional = true }
+#
+# Pinned to `0.22`, one minor behind what dependabot proposed (#125), on
+# purpose: `hyper-util` -> `hyper-rustls` -> `reqwest` already pulls `0.22.1`
+# into every consumer's graph (openhuman, tinyagents, tinycortex, tinymemory),
+# so a direct `0.23` requirement here resolves as a SECOND copy rather than
+# reusing that one — +1 package with no capability gained, which is exactly
+# what trips a consumer's kernel-dependency-floor ratchet (see
+# `tinyhumansai/openhuman`'s `scripts/kernel-floor.limits`). Unify with the
+# ecosystem's version instead of letting this crate's direct dependency get
+# ahead of it. Dependabot will likely re-propose `0.23` once `reqwest`'s own
+# tree moves past `0.22` — that will be fine to accept at that point, since it
+# will no longer create a duplicate.
+base64 = { version = "0.22", optional = true }
 # `flate2` decompresses `application/gzip` data URIs that carry an
 # `original_mime=` parameter — the shape a renderer emits when it compresses an
 # attachment before handing it over. Decompression is bounded by `Read::take`

src/graph/checkpoint/mod.rs

@@ -56,16 +56,45 @@ use crate::{Result, TinyAgentsError};
 /// prefix; `what` names the field being decoded (e.g. `"record"`,
 /// `"next_nodes"`) with no leading `"decode "` — this function supplies that
 /// wording once, alongside the tag.
+///
+/// **Only `what == "record"` — the whole-`Checkpoint<State>` decode, where
+/// `State` is the one thing this module knows can legitimately evolve across
+/// a caller's schema versions — is eligible for `[schema]` at all.**
+/// `"namespace"`, `"next_nodes"`, `"header"` and `"write payload"` are
+/// checkpoint-internal engine metadata with no versioned shape of their own;
+/// a `Category::Data` failure decoding any of those is corruption by
+/// construction; classifying it as `[schema]` and pruning would have been
+/// simply wrong.
+///
+/// Even for `"record"`, `Category::Data` is a HEURISTIC, not a proof: this
+/// module is generic over `State` (see the module docs) and has no way to
+/// positively identify a specific caller's known-legacy shape from a byte
+/// stream alone — a decode failure produced by data corruption that happens
+/// to still be syntactically valid JSON (e.g. a `state.schema_version` field
+/// whose value was flipped from a number to a string) also classifies as
+/// `Category::Data`, indistinguishable at this layer from a genuine older
+/// schema. This is an accepted, bounded residual risk, not a claim of
+/// perfect precision: the worst case is an unnecessary fresh restart from
+/// `plan` for what was actually corruption, which is a strict improvement
+/// over the state before this classifier existed (100% of decode failures,
+/// corrupt or not, were pruned identically). Closing the remaining gap
+/// precisely would need the generic `Checkpointer` trait to expose the raw
+/// bytes on a failed decode so a caller could apply its OWN positive-shape
+/// check — a real API addition, tracked as follow-up rather than done here.
 pub(super) fn decode_json_err(
     backend: &str,
     what: &str,
     err: serde_json::Error,
 ) -> TinyAgentsError {
-    let tag = match err.classify() {
-        serde_json::error::Category::Data => "schema",
-        serde_json::error::Category::Syntax
-        | serde_json::error::Category::Eof
-        | serde_json::error::Category::Io => "corrupt",
+    let tag = if what == "record" {
+        match err.classify() {
+            serde_json::error::Category::Data => "schema",
+            serde_json::error::Category::Syntax
+            | serde_json::error::Category::Eof
+            | serde_json::error::Category::Io => "corrupt",
+        }
+    } else {
+        "corrupt"
     };
     TinyAgentsError::Checkpoint(format!("{backend}: decode [{tag}] {what}: {err}"))
 }

src/graph/dag/mod.rs

@@ -63,8 +63,12 @@ pub use types::{DagIssue, DagNode};
 /// Reports whether the dependency edges contain a cycle.
 ///
 /// Edges naming an id that is not a declared node are ignored (see the module
-/// docs). Duplicate ids are collapsed to one node, so they cannot produce a
-/// false positive.
+/// docs). Duplicate ids are collapsed to one node (first declaration wins),
+/// so they cannot produce a false positive — but a caller that wants to add
+/// edges to a node id that already exists must merge them into that one
+/// declaration; a second `DagNode` with the same id has its edges silently
+/// dropped instead of contributing to the cycle check (see the comment at
+/// the dedupe below).
 pub fn has_cycle(nodes: &[DagNode<'_>]) -> bool {
     let ids: HashSet<&str> = nodes.iter().map(|n| n.id).collect();
     let mut indegree: HashMap<&str, usize> = ids.iter().map(|&id| (id, 0)).collect();
@@ -75,6 +79,16 @@ pub fn has_cycle(nodes: &[DagNode<'_>]) -> bool {
     // two conflicting declarations of the same id (e.g. `a[]` and `a[b]`) get
     // merged into edges that no single declaration actually names, which can
     // fabricate a cycle that would not exist under either declaration alone.
+    //
+    // Consequence for callers: a caller that wants to add dependency edges to
+    // a node that already exists in `nodes` must merge those edges into that
+    // node's single `DagNode` declaration, not append a second `DagNode` with
+    // the same id and the new edges — the second declaration's edges are
+    // silently dropped here (first declaration wins), so a real cycle they
+    // would have introduced goes undetected. This is unreachable in a caller
+    // that only ever mints fresh, never-reused ids (e.g. a fresh UUID per
+    // node), but is a live footgun for any caller that revisits an existing
+    // id.
     let mut processed: HashSet<&str> = HashSet::with_capacity(ids.len());
     for node in nodes {
         if !processed.insert(node.id) {

src/graph/delegation/graph.rs

@@ -181,8 +181,25 @@ where
     // pause via `ApprovalRequestCard`), which parks a live in-memory chat turn
     // and is deliberately left untouched.
     if require_review_approval {
+        let cancel_approval = cancel.clone();
         builder = builder.add_node("approval", move |s: DelegationState, ctx: NodeContext| {
+            let cancel = cancel_approval.clone();
             async move {
+                // Cancellation must be honoured here exactly like every other
+                // node: without this check, a cancellation that arrives while
+                // a gated run is waiting for approval (or during the
+                // preceding review worker) is invisible to this boundary. A
+                // decisionless retry would then interrupt again indefinitely
+                // (nothing ever routes it to `finalize`), while an approving
+                // resume would finalize successfully despite the run having
+                // been cancelled.
+                if cancel.is_cancelled() {
+                    return Ok(NodeResult::Command(
+                        Command::default()
+                            .with_update(DelegationUpdate::Cancelled)
+                            .with_goto(["finalize"]),
+                    ));
+                }
                 match ctx.resume {
                     None => {
                         let payload = json!({

src/graph/delegation/run.rs

@@ -157,6 +157,48 @@ where
     F: Fn(DelegationStage, DelegationState) -> Fut + Clone + Send + Sync + 'static,
     Fut: Future<Output = Result<DelegationStageOutput, String>> + Send + 'static,
 {
+    // This is the public entry point a host calls directly with an approver's
+    // decision — it does NOT go through `run_or_resume_delegation`, so it must
+    // provide its own two protections that function otherwise only gets:
+    //
+    // 1. Per-thread serialization: two concurrent approval callbacks for the
+    //    SAME paused thread (a retried callback racing with a deny, say) must
+    //    not both load the same interrupt checkpoint and independently
+    //    finalize opposite decisions, appending competing histories. Uses the
+    //    SAME `thread_lock` map as `run_or_resume_delegation`, so the two
+    //    entry points serialize against each other too, not just themselves.
+    // 2. Schema-version validation: `CompiledGraph::resume` (inside
+    //    `resume_graph`) applies the decision to whatever checkpoint it
+    //    finds, with no schema check of its own — that check lives in
+    //    `run_or_resume_delegation`'s match arms, which this path bypasses.
+    //    During a rollback or mixed-version deployment, serde can decode a
+    //    checkpoint written by a newer binary by ignoring its added fields,
+    //    letting an older binary consume the approval and finalize the state
+    //    under outdated semantics. Reject rather than guess.
+    let thread_id = config
+        .thread_id
+        .clone()
+        .ok_or_else(|| "delegation resume requires a thread_id".to_string())?;
+    let cp = config
+        .checkpointer
+        .clone()
+        .ok_or_else(|| "delegation resume requires a checkpointer".to_string())?;
+
+    let lock = thread_lock(&thread_id);
+    let _guard = lock.lock().await;
+
+    if let Some(checkpoint) = cp
+        .get(thread_id.as_str(), None)
+        .await
+        .map_err(|e| format!("delegation checkpoint read failed for thread {thread_id}: {e}"))?
+        && checkpoint.state.schema_version != CURRENT_SCHEMA_VERSION
+    {
+        return Err(format!(
+            "delegation resume refused for thread {thread_id}: checkpoint schema_version {}              does not match this binary's schema_version {} — the approval decision was NOT              applied",
+            checkpoint.state.schema_version, CURRENT_SCHEMA_VERSION
+        ));
+    }
+
     let approved = decision_is_approve(&decision);
     tracing::info!(
         approved,