fix: preserve invocation recovery across rewind and revalidate replay authority

crates/session-store/src/fs_store.rs

@@ -383,8 +383,11 @@ impl Store for FsStore {
             content.push_str(&serde_json::to_string(entry)?);
             content.push('\n');
         }
-        fs::write(&path, content.as_bytes())?;
-        Ok(())
+        let _guard = self
+            .append_lock
+            .lock()
+            .map_err(|_| std::io::Error::other("session store append lock was poisoned"))?;
+        crate::worker_session_store::atomic_write_bytes(&path, content.as_bytes())
     }
 
     fn exists(&self, session_id: SessionId, segment_id: SegmentId) -> Result<bool, StoreError> {

crates/session-store/src/store.rs

@@ -110,7 +110,14 @@ pub trait Store: Send + Sync {
     /// segment ID without its session ID (e.g. legacy `--session <UUID>`).
     fn lookup_session_of(&self, segment_id: SegmentId) -> Result<Option<SessionId>, StoreError>;
 
-    /// Create a new segment within `session_id`, with initial entries.
+    /// Create or atomically replace a segment within `session_id` with all entries.
+    ///
+    /// Readers and crash recovery must observe either the previous complete log
+    /// or the complete replacement, never a truncated prefix or partial record.
+    /// Failure before replacement leaves the old log intact; an error after the
+    /// atomic commit (for example a durability sync failure) may leave the complete
+    /// replacement. Filesystem implementations must sync replacement bytes before
+    /// rename and the containing directory afterward before reporting success.
     fn create_segment(
         &self,
         session_id: SessionId,

crates/session-store/src/worker_session_store.rs

@@ -1166,7 +1166,7 @@ fn atomic_write_json<T: Serialize>(path: &Path, value: &T) -> Result<(), StoreEr
     atomic_write_bytes(path, &bytes)
 }
 
-fn atomic_write_bytes(path: &Path, bytes: &[u8]) -> Result<(), StoreError> {
+pub(crate) fn atomic_write_bytes(path: &Path, bytes: &[u8]) -> Result<(), StoreError> {
     let parent = path
         .parent()
         .ok_or_else(|| std::io::Error::other("Worker Session path has no parent"))?;

crates/session-store/tests/fs_store_test.rs

@@ -115,6 +115,100 @@ fn create_segment_writes_all_entries() {
     assert_eq!(state.session_id, Some(sid));
 }
 
+#[test]
+fn create_segment_replacement_keeps_open_old_log_and_commits_complete_new_log() {
+    use std::io::Read;
+    let dir = tempfile::tempdir().unwrap();
+    let store = FsStore::new(dir.path()).unwrap();
+    let sid = new_session_id();
+    let segid = new_segment_id();
+    let old = vec![
+        nil_session_start(1, sid),
+        LogEntry::TurnEnd {
+            ts: 2,
+            turn_count: 1,
+        },
+    ];
+    store.create_segment(sid, segid, &old).unwrap();
+    let path = dir
+        .path()
+        .join(sid.to_string())
+        .join(format!("{segid}.jsonl"));
+    let old_bytes = std::fs::read(&path).unwrap();
+    let mut old_reader = std::fs::File::open(&path).unwrap();
+    let new = vec![
+        nil_session_start(1, sid),
+        LogEntry::Extension {
+            ts: 3,
+            domain: "recovery-checkpoint".into(),
+            payload: serde_json::json!({"receipt": "retained"}),
+        },
+    ];
+    store.create_segment(sid, segid, &new).unwrap();
+    let mut retained_inode = Vec::new();
+    old_reader.read_to_end(&mut retained_inode).unwrap();
+    // An in-place fs::write would truncate the inode held by this reader.
+    assert_eq!(retained_inode, old_bytes);
+    let reopened = FsStore::new(dir.path()).unwrap();
+    assert_eq!(
+        serde_json::to_value(reopened.read_all(sid, segid).unwrap()).unwrap(),
+        serde_json::to_value(&new).unwrap()
+    );
+    assert_eq!(reopened.read_entry_count(sid, segid).unwrap(), new.len());
+    assert_eq!(
+        std::fs::read_dir(path.parent().unwrap()).unwrap().count(),
+        1,
+        "successful replacement must not leave temporary files"
+    );
+}
+
+#[test]
+fn create_segment_atomic_replacement_readers_only_observe_complete_old_or_new() {
+    let dir = tempfile::tempdir().unwrap();
+    let store = FsStore::new(dir.path()).unwrap();
+    let sid = new_session_id();
+    let segid = new_segment_id();
+    let old = vec![
+        nil_session_start(1, sid),
+        LogEntry::Extension {
+            ts: 2,
+            domain: "old-evidence".into(),
+            payload: serde_json::json!({"result": "old"}),
+        },
+    ];
+    let new = vec![
+        nil_session_start(1, sid),
+        LogEntry::Extension {
+            ts: 3,
+            domain: "new-checkpoint".into(),
+            payload: serde_json::json!({"result": "new"}),
+        },
+    ];
+    store.create_segment(sid, segid, &old).unwrap();
+    let old_json = serde_json::to_value(&old).unwrap();
+    let new_json = serde_json::to_value(&new).unwrap();
+    let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
+    let reader_store = store.clone();
+    let reader_barrier = barrier.clone();
+    let reader = std::thread::spawn(move || {
+        reader_barrier.wait();
+        for _ in 0..1000 {
+            let observed =
+                serde_json::to_value(reader_store.read_all(sid, segid).unwrap()).unwrap();
+            assert!(
+                observed == old_json || observed == new_json,
+                "replacement must never expose a partial log: {observed}"
+            );
+        }
+    });
+    barrier.wait();
+    for _ in 0..50 {
+        store.create_segment(sid, segid, &new).unwrap();
+        store.create_segment(sid, segid, &old).unwrap();
+    }
+    reader.join().unwrap();
+}
+
 #[test]
 fn list_sessions_and_segments() {
     let dir = tempfile::tempdir().unwrap();