fix: batch agent recorder import integrity state (#32)

Summary
- Add a flush hook to record adapters and let integrity wrapping choose immediate or on-flush state persistence.
- Keep live run integrity state crash-durable after each record, but persist import integrity state once after the replayable batch completes.
- Document the import/run durability distinction and add coverage for on-flush state persistence.

Testing
- node scripts/sync-platform-package-versions.js
- cd tools/agent-recorder && cargo fmt --check
- cd tools/agent-recorder && cargo test --quiet
- cd tools/agent-recorder && cargo test --examples --quiet

Notes/Risks
- Interrupted integrity-backed imports may leave partial output/state alignment that should be discarded or rerun from a clean/aligned backend and state file.
- This preserves per-record durability for run.

Author: thiennamdinh

docs/info/src/content/docs/tools.md

@@ -62,7 +62,7 @@ Tool calls and tool results are message annotations. Edges such as `follows-mess
 
 ### Integrity
 
-`agent-recorder` can add optional forward-integrity metadata to each record. For the 0.1.0 tool, treat this as initial tamper-evidence and forward-integrity plumbing for review and experimentation, not as a cryptographically audited production guarantee. The public metadata contains the algorithm, key id, absolute index, payload hash, and authenticator. Local key-evolution state stays private and is not stored in backend records.
+`agent-recorder` can add optional forward-integrity metadata to each record. For the 0.1.x tool, treat this as initial tamper-evidence and forward-integrity plumbing for review and experimentation, not as a cryptographically audited production guarantee. The public metadata contains the algorithm, key id, absolute index, payload hash, and authenticator. Local key-evolution state stays private and is not stored in backend records.
 
 The integrity layer authenticates each indexed record independently. Backend storage, including Sync Web, remains responsible for ordering and history. The recorder provides `read --integrity-key`, `verify`, `status`, and `rekey` commands for verification and emergency key cutover workflows.
 

tools/agent-recorder/Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "agent-recorder"
-version = "0.1.0"
+version = "0.1.1"
 edition = "2021"
 license = "MIT"
 publish = false

tools/agent-recorder/Cargo.toml

@@ -154,7 +154,7 @@ The stable public JSON uses kebab-case where field names are controlled by `agen
 
 ## Integrity
 
-Integrity is optional and backend-independent. In the 0.1.0 tool, treat this as initial tamper-evidence and forward-integrity plumbing for review and experimentation, not as a cryptographically audited production guarantee. Enable it on writes with `--integrity agent-recorder-integrity-v1`, an integrity state path, and an initial key.
+Integrity is optional and backend-independent. In the 0.1.x tool, treat this as initial tamper-evidence and forward-integrity plumbing for review and experimentation, not as a cryptographically audited production guarantee. Enable it on writes with `--integrity agent-recorder-integrity-v1`, an integrity state path, and an initial key.
 
 ```sh
 AGENT_RECORDER_INTEGRITY_KEY='example secret' \
@@ -197,6 +197,8 @@ Local integrity state stores private future keys and is not written into records
 
 At one-based event `i`, compute `h = v2(i)` and generate future edge keys for levels `0..h`, each targeting `i + 2^d`. At event `j`, consume pending incoming keys whose target is `j`; after the backend write succeeds, consumed keys are deleted and the local state advances. A verifier with the root key derives `K_i` in logarithmic time and verifies the selected indexed record independently. Cryptographic review is still recommended before relying on this for high-assurance audit workflows.
 
+`run` persists integrity state after each new record because it is live capture. `import` treats the input as a replayable batch and persists integrity state after the batch completes, avoiding one durable state-file rewrite per imported record. If an integrity-backed import is interrupted, discard the partial output or rerun from a clean/aligned backend and state file.
+
 Read with verification:
 
 ```sh
@@ -271,7 +273,7 @@ A real signed JSON-LD example is checked in at `examples/integrity-agent-record.
 
 The `Agent Recorder Binaries` GitHub Actions workflow builds downloadable `agent-recorder-*` binaries for Linux, Linux musl/Alpine, macOS, and Windows targets. Branch workflow artifacts can be downloaded for testing before a tagged release.
 
-Tagged releases use `agent-recorder-v*` tags, for example `agent-recorder-v0.1.0`. Ledger binary releases use separate `ledger-v*` tags so agent-recorder artifacts do not mix with ledger/journal release assets.
+Tagged releases use `agent-recorder-v*` tags, for example `agent-recorder-v0.1.1`. Ledger binary releases use separate `ledger-v*` tags so agent-recorder artifacts do not mix with ledger/journal release assets.
 
 ## Sync Web Backend
 

tools/agent-recorder/README.md

@@ -13,7 +13,7 @@ use serde::Deserialize;
 use crate::adapters::AdapterRegistry;
 use crate::integrity::{
     integrity_status, load_state, rekey_state, verify_indexed_records, IntegrityKey,
-    IntegrityRecordAdapter, VerificationStatus, ALGORITHM,
+    IntegrityRecordAdapter, IntegrityStatePersistence, VerificationStatus, ALGORITHM,
 };
 use crate::records::{
     sync_web::{
@@ -445,6 +445,7 @@ fn cmd_run(
         args.recorder_data.as_deref(),
         &args.sync_web,
         &args.integrity,
+        IntegrityStatePersistence::Immediate,
     )?;
     let roots = vec![PathBuf::from(args.agent_data)];
     let mut seen = HashSet::new();
@@ -499,10 +500,13 @@ fn cmd_import(
         args.recorder_data.as_deref(),
         &args.sync_web,
         &args.integrity,
+        IntegrityStatePersistence::OnFlush,
     )?;
     let roots = vec![PathBuf::from(args.agent_data)];
     let report = if let Some(sink) = sink.as_mut() {
-        import_records(adapter.as_ref(), &roots, sink.as_mut())?
+        let report = import_records(adapter.as_ref(), &roots, sink.as_mut())?;
+        sink.flush()?;
+        report
     } else {
         let stdout = io::stdout();
         let mut stdout = stdout.lock();
@@ -775,6 +779,7 @@ fn create_record_adapter(
     recorder_data: Option<&str>,
     sync_web: &SyncWebArgs,
     integrity: &IntegrityArgs,
+    integrity_persistence: IntegrityStatePersistence,
 ) -> Result<Option<Box<dyn RecordAdapter>>> {
     let Some(recorder) = recorder else {
         if integrity.algorithm.is_some() {
@@ -815,12 +820,15 @@ fn create_record_adapter(
         .ok_or_else(|| anyhow!("integrity logging requires --integrity-state PATH"))?;
     let init_key = resolve_optional_integrity_key(integrity)?;
     let reader = create_record_reader(registry, recorder, recorder_data, sync_web)?;
-    Ok(Some(Box::new(IntegrityRecordAdapter::create_checked(
-        sink,
-        state,
-        init_key,
-        Some(reader.as_ref()),
-    )?)))
+    Ok(Some(Box::new(
+        IntegrityRecordAdapter::create_checked_with_persistence(
+            sink,
+            state,
+            init_key,
+            Some(reader.as_ref()),
+            integrity_persistence,
+        )?,
+    )))
 }
 
 fn resolve_optional_integrity_key(args: &IntegrityArgs) -> Result<Option<IntegrityKey>> {

tools/agent-recorder/src/cli.rs

@@ -75,6 +75,14 @@ pub struct IntegrityRecordAdapter {
     inner: Box<dyn RecordAdapter>,
     state_path: PathBuf,
     state: IntegrityState,
+    persistence: IntegrityStatePersistence,
+    dirty: bool,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum IntegrityStatePersistence {
+    Immediate,
+    OnFlush,
 }
 
 /// Root verifier/signing secret for one integrity stream.
@@ -154,6 +162,22 @@ impl IntegrityRecordAdapter {
         state_path: impl AsRef<Path>,
         init_key: Option<IntegrityKey>,
         reader: Option<&dyn RecordReader>,
+    ) -> Result<Self> {
+        Self::create_checked_with_persistence(
+            inner,
+            state_path,
+            init_key,
+            reader,
+            IntegrityStatePersistence::Immediate,
+        )
+    }
+
+    pub fn create_checked_with_persistence(
+        inner: Box<dyn RecordAdapter>,
+        state_path: impl AsRef<Path>,
+        init_key: Option<IntegrityKey>,
+        reader: Option<&dyn RecordReader>,
+        persistence: IntegrityStatePersistence,
     ) -> Result<Self> {
         let state_path = state_path.as_ref().to_path_buf();
         let mut state = load_or_create_state(&state_path, init_key.as_ref())?;
@@ -166,6 +190,8 @@ impl IntegrityRecordAdapter {
             inner,
             state_path,
             state,
+            persistence,
+            dirty: false,
         })
     }
 }
@@ -184,7 +210,19 @@ impl RecordAdapter for IntegrityRecordAdapter {
         self.inner.log(&signed)?;
 
         advance_state(&mut self.state, index, generated)?;
-        store_state(&self.state_path, &self.state)?;
+        match self.persistence {
+            IntegrityStatePersistence::Immediate => store_state(&self.state_path, &self.state)?,
+            IntegrityStatePersistence::OnFlush => self.dirty = true,
+        }
+        Ok(())
+    }
+
+    fn flush(&mut self) -> Result<()> {
+        self.inner.flush()?;
+        if self.dirty {
+            store_state(&self.state_path, &self.state)?;
+            self.dirty = false;
+        }
         Ok(())
     }
 }

tools/agent-recorder/src/integrity/mod.rs

@@ -58,6 +58,10 @@ pub trait RecordAdapter {
     fn name(&self) -> &'static str;
     /// Append or otherwise persist one normalized record.
     fn log(&mut self, record: &GraphRecord) -> Result<()>;
+    /// Flush any buffered adapter state after a batch of records.
+    fn flush(&mut self) -> Result<()> {
+        Ok(())
+    }
 }
 
 /// Readable backend used by `read`, `verify`, `status`, and integrity recovery.

tools/agent-recorder/src/records.rs

@@ -46,6 +46,11 @@ impl RecordAdapter for JsonlRecordAdapter {
         self.writer.flush()?;
         Ok(())
     }
+
+    fn flush(&mut self) -> Result<()> {
+        self.writer.flush()?;
+        Ok(())
+    }
 }
 
 impl RecordReader for JsonlRecordReader {

tools/agent-recorder/src/records/jsonl.rs

@@ -7,7 +7,7 @@ use agent_recorder::{
     integrity::{
         integrity_status, rekey_state, verify_indexed_record, verify_indexed_records,
         verify_record, IntegrityAlignment, IntegrityKey, IntegrityRecordAdapter,
-        VerificationStatus,
+        IntegrityStatePersistence, VerificationStatus,
     },
     records,
     records::{RecordAdapter, RecordReader, RecordSelector},
@@ -526,6 +526,38 @@ fn integrity_wrapper_signs_jsonl_records_and_verifies_range() -> Result<()> {
     Ok(())
 }
 
+#[test]
+fn integrity_on_flush_persists_state_after_batch() -> Result<()> {
+    let state = std::env::temp_dir().join(format!(
+        "agent-recorder-integrity-on-flush-{}.json",
+        std::process::id()
+    ));
+    let _ = fs::remove_file(&state);
+    let key = IntegrityKey::from_secret("batch-secret");
+    let records = import_fixture("pi")?;
+
+    let mut adapter = IntegrityRecordAdapter::create_checked_with_persistence(
+        Box::new(MemoryRecordAdapter::default()),
+        &state,
+        Some(key),
+        None,
+        IntegrityStatePersistence::OnFlush,
+    )?;
+    adapter.log(&records[0])?;
+    adapter.log(&records[1])?;
+
+    let before_flush = agent_recorder::integrity::load_state(&state)?;
+    assert_eq!(before_flush.next_index, 0);
+
+    adapter.flush()?;
+    let after_flush = agent_recorder::integrity::load_state(&state)?;
+    assert_eq!(after_flush.next_index, 2);
+    assert!(after_flush.future_keys.iter().all(|key| key.target >= 2));
+
+    let _ = fs::remove_file(state);
+    Ok(())
+}
+
 #[test]
 fn integrity_status_reports_alignment() -> Result<()> {
     let (out, state, _key) = write_integrity_fixture("status")?;