From c81658355fcf0ed5a4153a878d5180186a23f325 Mon Sep 17 00:00:00 2001 From: Alfredo Montesinos <22755327+ForkedInTime@users.noreply.github.com> Date: Mon, 3 Aug 2026 15:53:00 -0700 Subject: [PATCH] fix(session): a torn write no longer destroys the whole conversation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 (data integrity & persistence, 5 files, 1,779 LOC). One real data-loss path; the rest of the phase checked out clean and is recorded as cleared rather than changed. ## An interrupted append lost the entire session `Session::append` writes JSONL and never fsynced, so a turn was reported as saved while the bytes could still be in the page cache. `load_messages` then parsed with `.collect::>()`, meaning **any** malformed line failed the whole load. Together those are a total-loss path: crash or power-cut mid-append leaves a half-written final line, and the next `/resume` fails outright — losing the entire conversation, which is the one thing sessions exist to prevent. Two changes: - `append` now fsyncs, so a completed turn is durable rather than merely buffered. - `load_messages` tolerates a torn *final* line: it is dropped with a warning and the rest of the history loads. Losing one turn beats losing all of them. Corruption anywhere **other** than the last line is still a hard error, and deliberately so. Silently skipping a middle line could drop a `tool_use` while keeping its `tool_result`, which the API rejects outright — a subtly broken conversation is worse than a clear error, and the message now names the line. The parsing moved into `parse_message_lines` so this behaviour is testable against an explicit file rather than the global sessions directory. ## Session export was not atomic `export` used a direct `fs::write`, which truncates first — an interrupted export left the user with an empty file where their transcript was. Now uses the same `atomic_write` as the session files themselves. ## Checked and cleared, not changed - **Compaction cannot break tool_use/tool_result pairing.** `snip_compact` never removes blocks; it replaces `ToolResult` *content* with a placeholder and leaves structure intact. This was the sharpest risk on the phase plan — a mismatch is a hard API 400 that bricks a session — and the design is already correct. - **memory.rs durability is SQLite's.** Not a hand-rolled write path. - **memory.rs SQL is fully parameterised**, `LIMIT` included. No injection. - **`SessionMeta::save` and `overwrite` already use atomic_write** from an earlier audit; only `append` and `export` had been missed. QA (triple-checked): 606 tests, 0 failures. Clippy clean under the CI gate. Release 19.08 MB. Zero panics in production code across all five Phase 3 files. Phase 1 and 2 suites re-run green. The torn-line fix verified by restoring the intolerant `collect()`, which fails 2 tests. Co-Authored-By: Arch Linux --- src/session/mod.rs | 156 +++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 150 insertions(+), 6 deletions(-) diff --git a/src/session/mod.rs b/src/session/mod.rs index 58517fe..ad6b350 100644 --- a/src/session/mod.rs +++ b/src/session/mod.rs @@ -116,6 +116,11 @@ impl Session { file.write_all(line.as_bytes()).await?; file.write_all(b"\n").await?; } + // Durability. Without this the turn is reported as saved while the bytes + // may still be in the page cache, so a crash loses it — and can leave a + // half-written final line behind (see `load_messages`, which tolerates + // exactly that). + file.sync_all().await?; // Update preview from first user message if not yet set if self.meta.preview.is_empty() @@ -162,11 +167,7 @@ impl Session { Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), Err(e) => return Err(e.into()), }; - content - .lines() - .filter(|l| !l.trim().is_empty()) - .map(|l| serde_json::from_str::(l).map_err(anyhow::Error::from)) - .collect() + parse_message_lines(id, &content) } /// List all saved sessions, newest first. @@ -247,7 +248,10 @@ impl Session { } } - fs::write(dest, &out).await?; + // Same treatment as the session files themselves: a direct write + // truncates the destination first, so an interrupted export leaves the + // user with an empty or half-written file where their transcript was. + atomic_write(dest, out.as_bytes()).await?; Ok(dest.to_path_buf()) } @@ -407,6 +411,50 @@ fn human_session_name() -> String { .unwrap_or_else(|| "New session".to_string()) } + +/// Parse a session's JSONL body. +/// +/// Split out from `load_messages` so the torn-tail and mid-file-corruption +/// behaviour can be tested against an explicit file rather than the global +/// sessions directory. +fn parse_message_lines(id: &str, content: &str) -> Result> { + let lines: Vec<&str> = content.lines().filter(|l| !l.trim().is_empty()).collect(); + let total = lines.len(); + let mut out = Vec::with_capacity(total); + + for (i, line) in lines.into_iter().enumerate() { + match serde_json::from_str::(line) { + Ok(m) => out.push(m), + Err(e) => { + // A torn *final* line is the expected shape of a crash + // mid-append: nothing else references it, so dropping it + // recovers the whole session minus one turn. Previously any + // bad line failed the entire load via `collect()`, which + // turned a half-written last line into total loss of the + // conversation — the one thing sessions exist to prevent. + if i + 1 == total { + tracing::warn!( + "session {id}: discarding incomplete final line \ + (likely an interrupted write): {e}" + ); + break; + } + // Corruption anywhere else is not a torn write. Skipping it + // could drop a tool_use while keeping its tool_result, which + // the API rejects outright — a subtly broken conversation is + // worse than a clear error. + return Err(anyhow::anyhow!( + "session {id} is corrupt at line {} of {total}: {e}. \ + Refusing to load a partial history — later messages may \ + depend on it.", + i + 1 + )); + } + } + } + Ok(out) +} + /// Atomic file write: write to a sibling temp file, fsync, then rename over /// the target. Survives mid-write crashes — the target is either the old /// content or the new content, never a truncated splice. Falls back to a @@ -536,3 +584,99 @@ mod atomic_write_tests { assert_eq!(tokio::fs::read_to_string(&target).await.unwrap(), "x"); } } + +#[cfg(test)] +mod durability_tests { + use super::*; + use crate::api::types::{ContentBlock, Role}; + + fn msg(text: &str) -> Message { + Message { + role: Role::User, + content: vec![ContentBlock::Text { text: text.into() }], + } + } + + /// Write a JSONL file directly so these tests do not depend on the real + /// sessions directory or on `Session::new`'s side effects. + fn write_jsonl(dir: &std::path::Path, id: &str, msgs: &[Message], torn_tail: Option) { + let mut body = String::new(); + for (i, m) in msgs.iter().enumerate() { + let line = serde_json::to_string(m).unwrap(); + if let Some(keep) = torn_tail.filter(|_| i + 1 == msgs.len()) { + let keep = keep.min(line.len()); + body.push_str(&line[..keep]); // no trailing newline: a cut-off write + } else { + body.push_str(&line); + body.push('\n'); + } + } + std::fs::write(dir.join(format!("{id}.jsonl")), body).unwrap(); + } + + fn parse(dir: &std::path::Path, id: &str) -> Result> { + // Mirror of load_messages' parsing over an explicit path, so the test + // does not have to relocate the global sessions directory. + let content = std::fs::read_to_string(dir.join(format!("{id}.jsonl"))).unwrap_or_default(); + parse_message_lines(id, &content) + } + + /// The regression: a crash mid-append leaves the final line cut off. That + /// used to fail the entire load via `collect()`, losing the whole + /// conversation rather than one turn. + #[test] + fn torn_final_line_costs_one_turn_not_the_session() { + let d = tempfile::tempdir().unwrap(); + write_jsonl(d.path(), "s", &[msg("one"), msg("two"), msg("three")], Some(14)); + + let got = parse(d.path(), "s").expect("a torn tail must not fail the load"); + assert_eq!(got.len(), 2, "complete turns survive, the torn one is dropped"); + } + + /// Corruption that is not a torn tail must fail loudly: silently skipping a + /// middle line can drop a tool_use while keeping its tool_result, which the + /// API rejects outright. A subtly broken conversation is worse than an error. + #[test] + fn mid_file_corruption_is_reported_with_its_location() { + let d = tempfile::tempdir().unwrap(); + write_jsonl(d.path(), "s", &[msg("one"), msg("two"), msg("three")], None); + let p = d.path().join("s.jsonl"); + let mut lines: Vec = std::fs::read_to_string(&p) + .unwrap() + .lines() + .map(str::to_string) + .collect(); + lines[1] = "{ not json".into(); + std::fs::write(&p, lines.join("\n")).unwrap(); + + let err = parse(d.path(), "s").expect_err("must not silently skip"); + let m = err.to_string(); + assert!(m.contains("corrupt"), "{m}"); + assert!(m.contains("line 2"), "must locate the damage: {m}"); + } + + #[test] + fn intact_history_round_trips() { + let d = tempfile::tempdir().unwrap(); + write_jsonl(d.path(), "s", &[msg("a"), msg("b"), msg("c")], None); + assert_eq!(parse(d.path(), "s").unwrap().len(), 3); + } + + #[test] + fn empty_history_is_not_corruption() { + let d = tempfile::tempdir().unwrap(); + std::fs::write(d.path().join("s.jsonl"), "").unwrap(); + assert!(parse(d.path(), "s").unwrap().is_empty()); + } + + /// Blank lines are padding, not damage. + #[test] + fn blank_lines_are_ignored() { + let d = tempfile::tempdir().unwrap(); + write_jsonl(d.path(), "s", &[msg("a"), msg("b")], None); + let p = d.path().join("s.jsonl"); + let c = std::fs::read_to_string(&p).unwrap(); + std::fs::write(&p, c.replace('\n', "\n\n")).unwrap(); + assert_eq!(parse(d.path(), "s").unwrap().len(), 2); + } +}