fix: 22 audit findings across graph, harness, providers, language, and repl/rlm

docs/modules/graph/checkpointing.md

@@ -177,6 +177,27 @@ The checkpoint core lives in `src/graph/checkpoint/`:
   `Async` degrades to `Sync`. `Exit` persists only the terminal checkpoint and
   any interrupt boundary (interrupts must persist so the run can resume). Set
   it with `CompiledGraph::with_durability(mode)`.
+
+Two guarantees make `Async` durability safe to reason about. **Background writes
+are ordered**: each spawned write awaits the previously spawned one before its
+own `put`, because every bundled backend defines a thread's "latest" checkpoint
+by insertion order — an unserialized writer could let boundary N land after
+boundary N+1 and make a concurrent reader (or a `retry` racing a straggler
+append) see a stale record as latest. A write whose predecessor failed is
+skipped entirely rather than appending a record whose `parent_checkpoint_id`
+points at something that never persisted; the predecessor's error is what the
+run reports. **Every exit drains**: not only the terminal, interrupt, and
+failure boundaries but also the mid-run aborts — recursion limit, run deadline,
+node-visit limit, reducer and routing errors — settle in-flight writes before
+returning `Err`. Dropping the tracker would detach those tasks, discarding their
+outcome and racing a caller that immediately calls `retry(thread_id)`.
+
+`prune` is namespace-aware: the recency window applies per namespace, so the
+lineages of embedded subgraphs (which share the parent's thread id under their
+own namespace, and are never referenced by a parent record's
+`parent_checkpoint_id`) are retained rather than deleted along with old parent
+records.
+
 - `CheckpointConfig { thread_id, checkpoint_id, namespace }` — checkpoint
   coordinates. `CheckpointConfig::latest(thread_id)` addresses the newest
   checkpoint at the root namespace.

src/graph/checkpoint/mod.rs

@@ -266,7 +266,13 @@ where
     ///
     /// Strategy (lineage- and delta-safe):
     ///
-    /// 1. Protect the most recent `keep_last` checkpoints (listing order).
+    /// 1. Protect the most recent `keep_last` checkpoints (listing order) *of
+    ///    every namespace present in the thread*. An embedded subgraph writes
+    ///    its checkpoints under the parent's thread id but its own namespace,
+    ///    and its lineage is disjoint from the parent's (no parent-namespace
+    ///    record ever references a child-namespace id), so a thread-wide
+    ///    recency window would delete the child lineage outright and leave the
+    ///    thread unresumable.
     /// 2. Walk the `parent_checkpoint_id` chain of every protected checkpoint
     ///    and protect every ancestor reached. This is what honors the
     ///    delta-channel warning: a kept checkpoint that only stores a delta (or
@@ -283,18 +289,26 @@ where
         if metas.is_empty() {
             return Ok(0);
         }
-        let keep_last = keep_last.max(1).min(metas.len());
+        let keep_last = keep_last.max(1);
 
         // Index by id so ancestor walks are O(depth).
         let mut parent_of: HashMap<&str, Option<&str>> = HashMap::new();
         for m in &metas {
             parent_of.insert(m.checkpoint_id.as_str(), m.parent_checkpoint_id.as_deref());
         }
+        // Group by namespace so each lineage (the root run and every embedded
+        // subgraph run sharing this thread) gets its own recency window.
+        let mut by_namespace: HashMap<&Vec<String>, Vec<&CheckpointMetadata>> = HashMap::new();
+        for m in &metas {
+            by_namespace.entry(&m.namespace).or_default().push(m);
+        }
 
         let mut protected: HashSet<String> = HashSet::new();
-        // Step 1: the recency window.
-        for m in metas.iter().rev().take(keep_last) {
-            protected.insert(m.checkpoint_id.clone());
+        // Step 1: the recency window, per namespace.
+        for group in by_namespace.values() {
+            for m in group.iter().rev().take(keep_last) {
+                protected.insert(m.checkpoint_id.clone());
+            }
         }
         // Step 2: expand to every ancestor of a protected checkpoint.
         let window: Vec<String> = protected.iter().cloned().collect();

src/graph/checkpoint/test.rs

@@ -345,6 +345,36 @@ async fn prune_zero_keeps_latest_and_its_chain() {
     assert_eq!(cp.count("t"), 2);
 }
 
+#[tokio::test]
+async fn prune_keeps_a_window_per_namespace() {
+    let cp = InMemoryCheckpointer::<i32>::new();
+    // An embedded subgraph writes under the parent's thread but its own
+    // namespace, interleaved with the parent's records. The two lineages are
+    // disjoint — no parent record ever references a child id — so a
+    // thread-wide recency window would delete the child lineage outright and
+    // leave the thread unresumable.
+    let sub = |id: &str, parent: Option<&str>, step: usize| {
+        let mut c = checkpoint("t", id, parent, step);
+        c.namespace = vec!["sub".to_string()];
+        c
+    };
+    cp.put(checkpoint("t", "p1", None, 1)).await.unwrap();
+    cp.put(sub("s1", None, 1)).await.unwrap();
+    cp.put(sub("s2", Some("s1"), 2)).await.unwrap();
+    cp.put(checkpoint("t", "p2", Some("p1"), 2)).await.unwrap();
+
+    // Keep the last 1 of each namespace plus its ancestors: {p2, p1} and
+    // {s2, s1} — nothing is deleted, and the subgraph stays resolvable.
+    let removed = cp.prune("t", 1).await.unwrap();
+    assert_eq!(removed, 0);
+    let child = cp
+        .get_scoped("t", None, &["sub".to_string()])
+        .await
+        .unwrap()
+        .expect("the subgraph namespace must stay resumable after prune");
+    assert_eq!(child.checkpoint_id, "s2");
+}
+
 // ---- File-backed checkpointer ---------------------------------------------
 
 mod file_backend {