test(core): persisted-schema round-trip fuzz + reject future checkpoint versions (#31) (#49)

Adds core/tests/test_persisted_schema_roundtrip.rs — a dependency-free,
seeded round-trip fuzz over the persisted graph (LoopCheckpoint, RunRecord
incl. all three 3.3.0 AgentEvent variants, SubagentTaskSnapshot, TraceEvent,
VerificationReport, Message/ContentBlock, and the largest root SessionData),
asserting serialize→deserialize→re-serialize stability. Plus cross-version
coverage: backward-compat (pre-3.3.0 payloads missing schema_version /
identity fields / using the `todos` alias load with defaults), forward-compat
(unknown fields ignored — guards rolling cluster upgrades against a stray
deny_unknown_fields), the untagged ToolResultContentField and the persisted
untagged PermissionRule dual-form hazards.

Closes a real gap surfaced while writing it: LoopCheckpoint documented that
loads from a future schema version must be rejected, but no impl enforced it.
Adds LoopCheckpoint::ensure_loadable() and calls it in both the file and
memory store load paths, so resume_run surfaces the error and the live-run
sink starts fresh rather than misinterpreting an incompatible checkpoint.

Co-authored-by: Claude <claude@anthropic.com>

Author: ZhiXiao-Lin

core/src/loop_checkpoint.rs

@@ -69,6 +69,36 @@ pub struct LoopCheckpoint {
     pub checkpoint_ms: u64,
 }
 
+impl LoopCheckpoint {
+    /// Reject a checkpoint written by a *newer*, incompatible schema
+    /// version than this build understands — honoring the contract on
+    /// [`LOOP_CHECKPOINT_SCHEMA_VERSION`].
+    ///
+    /// Field *additions* are absorbed transparently by `#[serde(default)]`,
+    /// so an older checkpoint (lower `schema_version`, including a pre-v1
+    /// `0`) always remains loadable. A *future* version, however, may have
+    /// changed the meaning of existing fields or the tool-round boundary
+    /// semantics; resuming from one risks silent corruption (e.g.
+    /// re-running a non-idempotent tool on the wrong side of the boundary).
+    ///
+    /// [`SessionStore`](crate::store::SessionStore) impls call this right
+    /// after deserialization, so both `resume_run` (which surfaces the
+    /// error to the caller) and the live-run [`LoopCheckpointSink`] (which
+    /// logs and starts fresh) refuse to act on an unreadable checkpoint.
+    pub fn ensure_loadable(&self) -> anyhow::Result<()> {
+        if self.schema_version > LOOP_CHECKPOINT_SCHEMA_VERSION {
+            anyhow::bail!(
+                "loop checkpoint for run {} has schema version {} but this build supports at \
+                 most {}; refusing to resume from an incompatible future checkpoint",
+                self.run_id,
+                self.schema_version,
+                LOOP_CHECKPOINT_SCHEMA_VERSION
+            );
+        }
+        Ok(())
+    }
+}
+
 /// Receiver of per-tool-round checkpoints.
 ///
 /// The framework ships one adapter:

core/src/store/file_store.rs

@@ -466,8 +466,12 @@ impl SessionStore for FileSessionStore {
         let json = fs::read_to_string(&path)
             .await
             .with_context(|| format!("Failed to read loop checkpoint from {}", path.display()))?;
-        let checkpoint = serde_json::from_str(&json)
+        let checkpoint: LoopCheckpoint = serde_json::from_str(&json)
             .with_context(|| format!("Failed to parse loop checkpoint from {}", path.display()))?;
+        // Refuse to hand back a checkpoint written by a future, incompatible
+        // schema version — its field semantics may differ (see the contract
+        // on LoopCheckpoint::ensure_loadable).
+        checkpoint.ensure_loadable()?;
         Ok(Some(checkpoint))
     }
 

core/src/store/memory_store.rs

@@ -154,7 +154,15 @@ impl SessionStore for MemorySessionStore {
     }
 
     async fn load_loop_checkpoint(&self, run_id: &str) -> Result<Option<LoopCheckpoint>> {
-        Ok(self.loop_checkpoints.read().await.get(run_id).cloned())
+        match self.loop_checkpoints.read().await.get(run_id).cloned() {
+            // Enforce the same future-schema rejection as the file store so
+            // the contract holds uniformly across backends.
+            Some(cp) => {
+                cp.ensure_loadable()?;
+                Ok(Some(cp))
+            }
+            None => Ok(None),
+        }
     }
 
     async fn delete_loop_checkpoint(&self, run_id: &str) -> Result<()> {