From 30c69a33a944d58de3b755a0bbe1036f1d463225 Mon Sep 17 00:00:00 2001 From: SudoMaggie <268191359+sudomaggie@users.noreply.github.com> Date: Sun, 2 Aug 2026 16:31:08 +0800 Subject: [PATCH 1/4] fix(history): skip oversized JSONL records, isolate source failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 1 MiB per-record cap added in c94db5281 raised a hard error. Real Claude and Codex transcripts routinely exceed it — a base64 image in a tool_result, a long command's custom_tool_call_output — so the cap fired on ordinary data and the error propagated all the way out, emptying the sidebar of every external provider and making the Update button in Runtime > Scanning fail. The failure was permanent, not transient: the raising sync never reached sync_source_cache_from_conn, so nothing was written, the record kept its old signature, and every later scan re-read the same file and failed identically. Three layers, each bounding a different blast radius: - next_line skips an oversized record rather than raising, draining it into the boundary tail so the watermark stays byte-accurate and peak memory stays fixed regardless of record size. The cap rises to 16 MiB; it now bounds the reader's buffer rather than declaring what a valid record is. - Per-record isolation in the five parse loops: one unparsable session no longer costs its source every other session, and the record stays eligible for a later retry on its old cached row. - Per-source isolation in load_imported_history_sessions: one provider's store no longer decides whether the other seventeen are visible. Claude Code is first in the loader list, so its failure dropped all of them. Verified against the real stores behind the report: a full scan of 174 Claude and 331 Codex sessions completes, and re-running it with the cap forced back to 1 MiB exercises the skip path over the actual oversized records with the same 505-session result and byte-identical token totals. Pre-commit hook ran. Total eslint: 0, total circular: 0 --- src-tauri/Cargo.lock | 1 + src-tauri/crates/orgtrack-core/Cargo.toml | 3 + .../src/sources/anthropic_jsonl.rs | 17 ++- .../src/sources/claude_code/history.rs | 12 +- .../src/sources/codex/app/index.rs | 9 +- .../src/sources/imported_history/mod.rs | 28 ++++ .../src/sources/imported_history/watermark.rs | 141 +++++++++++++----- .../imported_history/watermark_tests.rs | 85 ++++++++++- .../orgtrack-core/src/sources/kimi/history.rs | 8 +- .../src/sources/pi/history_tests.rs | 14 +- .../src/sources/qwen_code/history.rs | 8 +- .../src/sources/qwen_code/history_tests.rs | 12 +- .../session_directory/aggregation.rs | 34 +++-- 13 files changed, 300 insertions(+), 72 deletions(-) diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index d3cddccee..8f06dc307 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -5260,6 +5260,7 @@ dependencies = [ "serde_json", "sha2", "similar", + "tracing", ] [[package]] diff --git a/src-tauri/crates/orgtrack-core/Cargo.toml b/src-tauri/crates/orgtrack-core/Cargo.toml index 2e0b2b42c..7fc550263 100644 --- a/src-tauri/crates/orgtrack-core/Cargo.toml +++ b/src-tauri/crates/orgtrack-core/Cargo.toml @@ -38,6 +38,9 @@ rusqlite = { workspace = true, optional = true } sha2 = { version = "0.10", optional = true } dirs = "6.0" similar = "2.6" +# Diagnostics only: report records this crate skips rather than raises +# (already a transitive dep via `app_paths`). +tracing = "0.1" prost-reflect = { version = "0.16.3", features = ["serde"] } regex = "1" memchr = "2.7" diff --git a/src-tauri/crates/orgtrack-core/src/sources/anthropic_jsonl.rs b/src-tauri/crates/orgtrack-core/src/sources/anthropic_jsonl.rs index 5791eb486..954f9dbbe 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/anthropic_jsonl.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/anthropic_jsonl.rs @@ -229,7 +229,13 @@ fn sync_cache(config: &AnthropicJsonlSource, conn: &mut Connection) -> Result<() config.source, &record.source_session_id, )?; - let parse = parse_session_meta_incremental(config, record, stored.as_ref())?; + let Some(parse) = imported_history::skip_unparsable_record( + config.source, + &record.source_session_id, + parse_session_meta_incremental(config, record, stored.as_ref()), + ) else { + continue; + }; watermark::write_parse_watermark_from_conn( conn, config.source, @@ -238,7 +244,14 @@ fn sync_cache(config: &AnthropicJsonlSource, conn: &mut Connection) -> Result<() )?; parse.meta } else { - parse_session_meta(config, record)? + let Some(meta) = imported_history::skip_unparsable_record( + config.source, + &record.source_session_id, + parse_session_meta(config, record), + ) else { + continue; + }; + meta }; inputs.push(meta_to_cache_input(config, meta)); } diff --git a/src-tauri/crates/orgtrack-core/src/sources/claude_code/history.rs b/src-tauri/crates/orgtrack-core/src/sources/claude_code/history.rs index 4d1dc87a8..e379362ed 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/claude_code/history.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/claude_code/history.rs @@ -611,11 +611,13 @@ fn sync_claude_code_history_cache(conn: &mut Connection) -> Result<(), String> { .get(&record.source_session_id) .cloned() .unwrap_or_default(); - let parse = parse_claude_session_meta_with_title( - record, - stored_watermark.as_ref(), - external_title, - )?; + let Some(parse) = imported_history::skip_unparsable_record( + SOURCE_CLAUDE_CODE, + &record.source_session_id, + parse_claude_session_meta_with_title(record, stored_watermark.as_ref(), external_title), + ) else { + continue; + }; imported_history::watermark::write_parse_watermark_from_conn( conn, SOURCE_CLAUDE_CODE, diff --git a/src-tauri/crates/orgtrack-core/src/sources/codex/app/index.rs b/src-tauri/crates/orgtrack-core/src/sources/codex/app/index.rs index e6a083ba8..167bc8696 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/codex/app/index.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/codex/app/index.rs @@ -319,8 +319,13 @@ fn sync_codex_app_cache(conn: &mut Connection) -> Result<(), String> { .get(&record.source_session_id) .cloned() .unwrap_or_default(); - let parse = - parse_codex_session_meta_with_title(record, stored_watermark.as_ref(), external_title)?; + let Some(parse) = imported_history::skip_unparsable_record( + SOURCE_CODEX_APP, + &record.source_session_id, + parse_codex_session_meta_with_title(record, stored_watermark.as_ref(), external_title), + ) else { + continue; + }; imported_history::watermark::write_parse_watermark_from_conn( conn, SOURCE_CODEX_APP, diff --git a/src-tauri/crates/orgtrack-core/src/sources/imported_history/mod.rs b/src-tauri/crates/orgtrack-core/src/sources/imported_history/mod.rs index 7b055033b..dc7f0b635 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/imported_history/mod.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/imported_history/mod.rs @@ -39,6 +39,34 @@ pub const FUNCTION_GLOB_FILE_SEARCH: &str = "glob_file_search"; pub const FUNCTION_AWAIT_OUTPUT: &str = "await_output"; pub const DEFAULT_LIST_LIMIT: usize = 200; +/// Drop one unparsable record from a source sync instead of failing the sync. +/// +/// A sync that raises leaves `sync_source_cache_from_conn` unreached, so *no* +/// session of that source is written — and because the record keeps its old +/// cache signature, the next scan re-reads the same file and fails the same +/// way. One malformed transcript would permanently cost a provider its entire +/// sidebar. Skipping keeps that record on its last-known cached row (or absent +/// if never cached) and still eligible for a later retry, while every other +/// session in the source syncs normally. +pub fn skip_unparsable_record( + source: &str, + source_session_id: &str, + outcome: Result, +) -> Option { + match outcome { + Ok(value) => Some(value), + Err(error) => { + tracing::warn!( + source, + source_session_id, + error = %error, + "imported history: skipping record that failed to parse" + ); + None + } + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum ImportedHistoryLoader { ClaudeCode, diff --git a/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark.rs b/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark.rs index 6f9931ad0..223476cb0 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark.rs @@ -28,7 +28,13 @@ use serde::{Deserialize, Serialize}; /// A single JSONL record must fit within this many raw bytes (including its /// trailing newline). The reader checks the bound before extending its buffer, /// so malformed or hostile files cannot cause unbounded allocation. -pub const MAX_JSONL_LINE_BYTES: usize = 1024 * 1024; +/// +/// Real transcripts routinely carry multi-megabyte tool results — a full-file +/// read or a long command's output lands as one record — so this is a ceiling +/// on the reader's buffer, not a statement about what a valid record is. +/// Exceeding it makes [`WatermarkedTranscriptReader::next_line`] skip that one +/// record; it is never fatal to the parse. +pub const MAX_JSONL_LINE_BYTES: usize = 16 * 1024 * 1024; /// Only a fixed window immediately before the committed offset is read when an /// append is validated. File identity catches rotation; this boundary catches @@ -183,6 +189,9 @@ impl BoundaryFingerprint { } } +// `path` is read only by the non-unix branch; unix identifies the file by +// dev/ino and never looks at it. +#[cfg_attr(unix, allow(unused_variables))] fn source_file_identity(path: &Path, metadata: &std::fs::Metadata) -> Option { #[cfg(unix)] { @@ -236,6 +245,23 @@ fn push_boundary_bytes(window: &mut Vec, bytes: &[u8]) { window.extend_from_slice(bytes); } +/// An oversized record being drained. Only its byte length and its trailing +/// [`BOUNDARY_WINDOW_BYTES`] are retained — that is everything the watermark +/// needs to describe the seam the record ends at, and it keeps a record of any +/// size costing a fixed amount of memory. +#[derive(Debug, Default)] +struct SkippedRecord { + tail: Vec, + len: u64, +} + +impl SkippedRecord { + fn push(&mut self, bytes: &[u8]) { + push_boundary_bytes(&mut self.tail, bytes); + self.len += bytes.len() as u64; + } +} + #[derive(Debug, PartialEq, Eq)] pub struct TranscriptLine { pub text: String, @@ -256,6 +282,8 @@ pub struct WatermarkedTranscriptReader { resume_state_json: Option, buf: Vec, error_label: &'static str, + /// Kept only so a skipped oversized record names the file it came from. + display_path: String, } impl WatermarkedTranscriptReader { @@ -321,6 +349,7 @@ impl WatermarkedTranscriptReader { resume_state_json, buf: Vec::new(), error_label, + display_path: path.display().to_string(), }) } @@ -328,49 +357,89 @@ impl WatermarkedTranscriptReader { self.resume_state_json.as_deref() } + /// Read the next record, skipping any that exceeds [`MAX_JSONL_LINE_BYTES`]. + /// + /// An oversized record is drained rather than buffered, and dropped rather + /// than raised: one giant tool result must not cost the caller the rest of + /// the file — and, upstream, every other session and every other provider + /// in the same pass. Its bytes still feed the boundary window and the + /// complete-line offset, so the watermark stays byte-accurate and a later + /// resume lands on the same seam as a parse that never skipped anything. pub fn next_line(&mut self) -> Result, String> { - self.buf.clear(); - let mut terminated = false; loop { - let available = self.reader.fill_buf().map_err(|err| { - format!("Failed to read {} history line: {err}", self.error_label) - })?; - if available.is_empty() { - break; + self.buf.clear(); + let mut terminated = false; + // `Some` once this record passed the cap: from that point its bytes + // go straight to the boundary tail instead of `buf`, so peak memory + // stays bounded no matter how long the record runs. + let mut skipped: Option = None; + loop { + let available = self.reader.fill_buf().map_err(|err| { + format!("Failed to read {} history line: {err}", self.error_label) + })?; + if available.is_empty() { + break; + } + let newline = available.iter().position(|byte| *byte == b'\n'); + let take = newline.map_or(available.len(), |index| index + 1); + if skipped.is_none() && self.buf.len().saturating_add(take) > MAX_JSONL_LINE_BYTES { + // Hand the already-buffered prefix over first so the + // record's bytes reach the boundary tail in file order. + let mut record = SkippedRecord::default(); + record.push(&self.buf); + self.buf.clear(); + self.buf.shrink_to_fit(); + skipped = Some(record); + } + match skipped.as_mut() { + Some(record) => record.push(&available[..take]), + None => self.buf.extend_from_slice(&available[..take]), + } + self.reader.consume(take); + if newline.is_some() { + terminated = true; + break; + } } - let newline = available.iter().position(|byte| *byte == b'\n'); - let take = newline.map_or(available.len(), |index| index + 1); - if self.buf.len().saturating_add(take) > MAX_JSONL_LINE_BYTES { - return Err(format!( - "Failed to read {} history line: record exceeds {} bytes", - self.error_label, MAX_JSONL_LINE_BYTES - )); + + if let Some(record) = skipped { + // An unterminated oversized record is the live writer's + // in-progress tail: leave the watermark untouched, exactly as + // an unterminated normal line does. + if !terminated { + return Ok(None); + } + push_boundary_bytes(&mut self.boundary_window, &record.tail); + self.complete_offset += record.len; + tracing::warn!( + source = self.error_label, + path = %self.display_path, + bytes = record.len, + limit = MAX_JSONL_LINE_BYTES, + "imported history: skipping oversized JSONL record" + ); + continue; } - self.buf.extend_from_slice(&available[..take]); - self.reader.consume(take); - if newline.is_some() { - terminated = true; - break; + + if self.buf.is_empty() { + return Ok(None); } - } - if self.buf.is_empty() { - return Ok(None); - } - if terminated { - push_boundary_bytes(&mut self.boundary_window, &self.buf); - self.complete_offset += self.buf.len() as u64; - } - let mut end = self.buf.len(); - if terminated { - end -= 1; - if end > 0 && self.buf[end - 1] == b'\r' { + if terminated { + push_boundary_bytes(&mut self.boundary_window, &self.buf); + self.complete_offset += self.buf.len() as u64; + } + let mut end = self.buf.len(); + if terminated { end -= 1; + if end > 0 && self.buf[end - 1] == b'\r' { + end -= 1; + } } + let text = std::str::from_utf8(&self.buf[..end]) + .map_err(|err| format!("Failed to read {} history line: {err}", self.error_label))? + .to_string(); + return Ok(Some(TranscriptLine { text, terminated })); } - let text = std::str::from_utf8(&self.buf[..end]) - .map_err(|err| format!("Failed to read {} history line: {err}", self.error_label))? - .to_string(); - Ok(Some(TranscriptLine { text, terminated })) } pub fn into_watermark( diff --git a/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark_tests.rs b/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark_tests.rs index fc5d564e2..5070ff0b0 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark_tests.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark_tests.rs @@ -203,8 +203,12 @@ fn rotated_file_identity_forces_a_full_reparse() { cleanup(&path); } +/// A single huge tool result is ordinary in real Claude/Codex transcripts. It +/// must cost that one record and nothing else — skipping it has to leave the +/// following lines readable and the watermark landing past both, so a resume +/// does not re-read the record it already gave up on. #[test] -fn oversized_line_is_rejected_without_advancing_a_watermark() { +fn oversized_line_is_skipped_without_stopping_the_parse() { let path = temp_transcript("oversized", "stable\n"); let (mtime, size) = stat(&path); let mut reader = @@ -224,7 +228,7 @@ fn oversized_line_is_rejected_without_advancing_a_watermark() { .open(&path) .and_then(|mut file| { std::io::Write::write_all(&mut file, &oversized)?; - std::io::Write::write_all(&mut file, b"\n") + std::io::Write::write_all(&mut file, b"\nafter\n") }) .expect("append oversized record"); let (mtime_after, size_after) = stat(&path); @@ -238,8 +242,81 @@ fn oversized_line_is_rejected_without_advancing_a_watermark() { ) .expect("open appended file"); assert_eq!(resumed.resume_state_json(), Some("stable-state")); - let error = resumed.next_line().expect_err("oversized line rejected"); - assert!(error.contains("record exceeds")); + assert_eq!(read_all(&mut resumed), vec![("after".to_string(), true)]); + let after_skip = resumed.into_watermark(1, mtime_after, size_after, "after-state".to_string()); + assert_eq!(after_skip.byte_offset, size_after); + + // The seam the skip left behind must still validate, or the next scan + // would cold-reparse the whole file and skip the same record again. + let mut reopened = WatermarkedTranscriptReader::open( + &path, + "Test", + Some(&after_skip), + 1, + mtime_after, + size_after, + ) + .expect("reopen after skip"); + assert_eq!(reopened.resume_state_json(), Some("after-state")); + assert!(read_all(&mut reopened).is_empty()); + + cleanup(&path); +} + +/// The cap bounds the reader's buffer; it must not silently truncate a record +/// that merely happens to be large. +#[test] +fn a_record_at_the_size_limit_is_still_returned_whole() { + let body = "y".repeat(MAX_JSONL_LINE_BYTES - 1); + let path = temp_transcript("at-limit", &format!("{body}\n")); + let (mtime, size) = stat(&path); + let mut reader = + WatermarkedTranscriptReader::open(&path, "Test", None, 1, mtime, size).expect("open full"); + assert_eq!( + reader.next_line().expect("read at-limit record"), + Some(TranscriptLine { + text: body, + terminated: true, + }) + ); + assert_eq!(reader.next_line().expect("read eof"), None); + + cleanup(&path); +} + +/// A live writer part-way through appending a huge record: the tail is not yet +/// complete, so it must not advance the watermark past bytes a later append +/// will extend. +#[test] +fn unterminated_oversized_tail_does_not_advance_the_watermark() { + let path = temp_transcript("oversized-tail", "stable\n"); + let (mtime, size) = stat(&path); + let committed = { + let mut reader = WatermarkedTranscriptReader::open(&path, "Test", None, 1, mtime, size) + .expect("open full"); + assert!(read_all(&mut reader).len() == 1); + reader.into_watermark(1, mtime, size, "stable-state".to_string()) + }; + + let oversized = vec![b'x'; MAX_JSONL_LINE_BYTES + 1]; + fs::OpenOptions::new() + .append(true) + .open(&path) + .and_then(|mut file| std::io::Write::write_all(&mut file, &oversized)) + .expect("append unterminated oversized record"); + let (mtime_after, size_after) = stat(&path); + let mut resumed = WatermarkedTranscriptReader::open( + &path, + "Test", + Some(&committed), + 1, + mtime_after, + size_after, + ) + .expect("open appended file"); + assert!(read_all(&mut resumed).is_empty()); + let after = resumed.into_watermark(1, mtime_after, size_after, "stable-state".to_string()); + assert_eq!(after.byte_offset, committed.byte_offset); cleanup(&path); } diff --git a/src-tauri/crates/orgtrack-core/src/sources/kimi/history.rs b/src-tauri/crates/orgtrack-core/src/sources/kimi/history.rs index df91aa987..68edcb644 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/kimi/history.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/kimi/history.rs @@ -276,7 +276,13 @@ fn sync_kimi_history_cache_in( SOURCE_KIMI, &record.source_session_id, )?; - let parsed = parse_kimi_meta(record, layout, default_model, stored.as_ref())?; + let Some(parsed) = imported_history::skip_unparsable_record( + SOURCE_KIMI, + &record.source_session_id, + parse_kimi_meta(record, layout, default_model, stored.as_ref()), + ) else { + continue; + }; let session_id = parsed.input.session_id.clone(); // The session cache signature is the authoritative changed-record // marker, so commit it last. If a prior write fails, or the final diff --git a/src-tauri/crates/orgtrack-core/src/sources/pi/history_tests.rs b/src-tauri/crates/orgtrack-core/src/sources/pi/history_tests.rs index 6f6382c44..2a488823b 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/pi/history_tests.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/pi/history_tests.rs @@ -187,7 +187,7 @@ fn exact_leaf_depth_ignores_nested_foreign_jsonl() { } #[test] -fn oversized_append_preserves_the_last_good_cache_and_watermark() { +fn oversized_append_is_skipped_and_the_sync_still_succeeds() { let root = temp_root("oversized"); let path = write_session( &root, @@ -211,18 +211,20 @@ fn oversized_append_preserves_the_last_good_cache_and_watermark() { file.write_all(b"\n") }) .expect("append oversized record"); - let error = anthropic_jsonl::list_sessions_paginated(&config, &mut conn, 10, 0) - .expect_err("oversized append rejected"); - assert!(error.contains("record exceeds")); + anthropic_jsonl::list_sessions_paginated(&config, &mut conn, 10, 0) + .expect("oversized append skipped, scan still succeeds"); let after = watermark::read_parse_watermark_from_conn(&conn, SOURCE_PI, "--repo--/session-a") - .expect("read preserved watermark") + .expect("read watermark") .expect("watermark remains"); let cached = imported_cache::query_cached_session_from_conn(&conn, SOURCE_PI, "--repo--/session-a") .expect("read cached row") .expect("cached row remains"); - assert_eq!(after, before); + // The skipped record contributed nothing, so the parsed totals are + // unchanged — but the watermark must move past it, or every later scan + // would re-read and re-skip the same bytes. assert_eq!(cached.output_tokens, 7); + assert!(after.byte_offset > before.byte_offset); fs::remove_dir_all(root).ok(); } diff --git a/src-tauri/crates/orgtrack-core/src/sources/qwen_code/history.rs b/src-tauri/crates/orgtrack-core/src/sources/qwen_code/history.rs index 339664101..2fd7bad4c 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/qwen_code/history.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/qwen_code/history.rs @@ -413,7 +413,13 @@ fn sync_qwen_code_history_cache_at_root(conn: &mut Connection, root: &Path) -> R SOURCE_QWEN_CODE, &record.source_session_id, )?; - let parsed = parse_qwen_session_meta(record, stored.as_ref(), root)?; + let Some(parsed) = imported_history::skip_unparsable_record( + SOURCE_QWEN_CODE, + &record.source_session_id, + parse_qwen_session_meta(record, stored.as_ref(), root), + ) else { + continue; + }; // Recovery invariant: rounds and watermark are written first, while // the cache signature remains old. Any failure before the final cache // upsert therefore leaves this record eligible on the next demand diff --git a/src-tauri/crates/orgtrack-core/src/sources/qwen_code/history_tests.rs b/src-tauri/crates/orgtrack-core/src/sources/qwen_code/history_tests.rs index 07d3b61f1..451a66d91 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/qwen_code/history_tests.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/qwen_code/history_tests.rs @@ -380,7 +380,7 @@ fn discovery_does_not_follow_directory_or_file_symlinks() { } #[test] -fn oversized_append_preserves_last_good_cache_and_watermark() { +fn oversized_append_is_skipped_and_the_sync_still_succeeds() { let root = temp_root("oversized"); let path = write_session( &root, @@ -409,9 +409,8 @@ fn oversized_append_preserves_last_good_cache_and_watermark() { file.write_all(b"\n") }) .expect("append oversized record"); - let error = - sync_qwen_code_history_cache_at_root(&mut conn, &root).expect_err("reject oversized line"); - assert!(error.contains("record exceeds")); + sync_qwen_code_history_cache_at_root(&mut conn, &root) + .expect("oversized append skipped, sync still succeeds"); let after = imported_history::watermark::read_parse_watermark_from_conn( &conn, @@ -424,8 +423,11 @@ fn oversized_append_preserves_last_good_cache_and_watermark() { imported_cache::query_cached_session_from_conn(&conn, SOURCE_QWEN_CODE, "repo-a/session-a") .expect("read cache") .expect("cache remains"); - assert_eq!(after, before); + // The skipped record contributed nothing, so the parsed totals are + // unchanged — but the watermark must move past it, or every later scan + // would re-read and re-skip the same bytes. assert_eq!(cached.output_tokens, 20); + assert!(after.byte_offset > before.byte_offset); fs::remove_dir_all(root).ok(); } diff --git a/src-tauri/src/agent_sessions/session_directory/aggregation.rs b/src-tauri/src/agent_sessions/session_directory/aggregation.rs index 45e32d645..6cea8cc11 100644 --- a/src-tauri/src/agent_sessions/session_directory/aggregation.rs +++ b/src-tauri/src/agent_sessions/session_directory/aggregation.rs @@ -493,19 +493,33 @@ fn load_imported_history_sessions( // pages read that stable cache snapshot directly. Re-running a full // provider scan for every offset made pagination multiply filesystem // and SQLite work without improving freshness. - let page = if page_offset == 0 { - (loader.load_page)(&mut conn, page_limit, page_offset)? + let loaded = if page_offset == 0 { + (loader.load_page)(&mut conn, page_limit, page_offset) } else if let Some(load_continuation_page) = loader.load_continuation_page { - load_continuation_page(&mut conn, page_limit, page_offset)? + load_continuation_page(&mut conn, page_limit, page_offset) } else { - ExternalHistoryPage::Imported( - imported_history_cache::query_imported_session_page_from_conn( - &conn, - loader.source, - page_limit, - page_offset, - )?, + imported_history_cache::query_imported_session_page_from_conn( + &conn, + loader.source, + page_limit, + page_offset, ) + .map(ExternalHistoryPage::Imported) + }; + // One provider's on-disk store must not decide whether the others are + // visible. Propagating here dropped every source after the failing one + // from the sidebar — and Claude Code, the most likely to hit an + // unreadable transcript, is first in the list. + let page = match loaded { + Ok(page) => page, + Err(error) => { + tracing::warn!( + source = loader.source, + error = %error, + "session_directory: skipping external history source that failed to load" + ); + continue; + } }; append_external_history_page(&mut records, loader.source, page); } From 23f82dcd8410810d3882e8f2bc1d18d153f95576 Mon Sep 17 00:00:00 2001 From: SudoMaggie <268191359+sudomaggie@users.noreply.github.com> Date: Sun, 2 Aug 2026 16:50:51 +0800 Subject: [PATCH 2/4] fix(history): truncate oversized JSON values instead of skipping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Skipping an over-cap record kept the parse alive but still lost the record, and it lost it silently. The record was salvageable: the parsers keep at most 50_000 chars of any single value (MAX_TOOL_OUTPUT_CHARS, MAX_TEXT_CHARS_PER_CHUNK), so the reader was materializing megabytes to hand downstream a few kilobytes. next_line now streams each record through JsonStringTruncator, which drops the interior of an oversized string value and appends a marker. Structure outside strings passes through untouched, so a truncated record keeps its fields, nesting and timestamps and still parses — only the payload shortens. Peak memory now tracks a record's structure rather than its payload, so size alone no longer decides whether a record survives. Correctness rests on where a cut may land. A cut is taken only in ScanState::InString and only on a UTF-8 leading byte, so it can never split a \uXXXX sequence, leave a dangling backslash, or halve a character — each of which would invalidate the whole record rather than one value. The tests sweep every alignment within a repeating escape unit and every byte of a 4-byte character. The per-value budget is bounded by what the record has already emitted. A per-value budget alone does not bound a record: a real Claude tool_result carrying several images is 1.28 MB across values of ~640 KB, every one of them legal. Tightening the allowance as the record fills means string content can never push a record past the cap — only structure can, and that is the one shape truncation cannot rescue, so it remains skipped. Verified against the same real stores: 505 sessions over 174 Claude and 331 Codex, stable across repeated runs. The Codex records that triggered the original report carry single 2.1-2.4 MB values and now truncate and parse; both their sessions index with full names and token counts. Pre-commit hook ran. Total eslint: 0, total circular: 0 --- .../src/sources/imported_history/watermark.rs | 249 ++++++++++++++---- .../imported_history/watermark_tests.rs | 151 ++++++++++- 2 files changed, 342 insertions(+), 58 deletions(-) diff --git a/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark.rs b/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark.rs index 223476cb0..ba0f10459 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark.rs @@ -25,17 +25,29 @@ use std::time::UNIX_EPOCH; use rusqlite::{params, Connection, OptionalExtension}; use serde::{Deserialize, Serialize}; -/// A single JSONL record must fit within this many raw bytes (including its -/// trailing newline). The reader checks the bound before extending its buffer, -/// so malformed or hostile files cannot cause unbounded allocation. -/// -/// Real transcripts routinely carry multi-megabyte tool results — a full-file -/// read or a long command's output lands as one record — so this is a ceiling -/// on the reader's buffer, not a statement about what a valid record is. -/// Exceeding it makes [`WatermarkedTranscriptReader::next_line`] skip that one -/// record; it is never fatal to the parse. +/// Ceiling on the buffer one record may occupy *after* oversized string values +/// have been truncated away. Because truncation already bounds the payload, +/// only a record whose JSON *structure* is enormous — or a file with no +/// newlines at all — can reach this, and such a record is skipped rather than +/// raised. It is never fatal to the parse. pub const MAX_JSONL_LINE_BYTES: usize = 16 * 1024 * 1024; +/// Longest single JSON string value copied verbatim. Past this the value's +/// interior is dropped as it streams and [`TRUNCATION_MARKER`] is appended, so +/// the record stays valid JSON and peak memory tracks a record's structure +/// rather than its payload. +/// +/// Real transcripts carry multi-megabyte string values — a base64 image in a +/// `tool_result`, a long command's `custom_tool_call_output`. The parsers keep +/// at most 50_000 chars of any single value (`MAX_TOOL_OUTPUT_CHARS`, +/// `MAX_TEXT_CHARS_PER_CHUNK`), so this budget is ~20x what the largest +/// consumer can retain: truncating here is invisible downstream. +pub const MAX_JSON_STRING_BYTES: usize = 1024 * 1024; + +/// Appended in place of a dropped string interior. Plain ASCII with no `"` or +/// `\`, so it needs no escaping and cannot itself break the record. +const TRUNCATION_MARKER: &[u8] = b"...[truncated]"; + /// Only a fixed window immediately before the committed offset is read when an /// append is validated. File identity catches rotation; this boundary catches /// truncate/rewrite-and-regrow at the append seam. @@ -245,23 +257,145 @@ fn push_boundary_bytes(window: &mut Vec, bytes: &[u8]) { window.extend_from_slice(bytes); } -/// An oversized record being drained. Only its byte length and its trailing -/// [`BOUNDARY_WINDOW_BYTES`] are retained — that is everything the watermark -/// needs to describe the seam the record ends at, and it keeps a record of any -/// size costing a fixed amount of memory. +/// The raw file bytes one record occupies. Only its length and its trailing +/// [`BOUNDARY_WINDOW_BYTES`] are retained — everything the watermark needs to +/// describe the seam the record ends at, at fixed cost for any record size. +/// +/// Tracked separately from the reader's output buffer because truncation makes +/// the two differ: offsets and fingerprints must describe what is on disk, not +/// what we chose to keep. #[derive(Debug, Default)] -struct SkippedRecord { +struct RawRecordSpan { tail: Vec, len: u64, } -impl SkippedRecord { +impl RawRecordSpan { fn push(&mut self, bytes: &[u8]) { push_boundary_bytes(&mut self.tail, bytes); self.len += bytes.len() as u64; } } +fn is_utf8_continuation(byte: u8) -> bool { + (byte & 0b1100_0000) == 0b1000_0000 +} + +/// How much of the current string may still be copied, given what the record +/// has already emitted. +/// +/// A per-value budget alone does not bound a record: real transcripts carry +/// records built from several large-but-legal values (a Claude `tool_result` +/// with multiple images), and enough of those still blow the buffer while no +/// single one is over budget. Tightening the allowance as the record fills +/// means string content can never push a record past +/// [`MAX_JSONL_LINE_BYTES`] — only its structure can, which is the one case +/// truncation genuinely cannot rescue. +fn string_budget(emitted: usize) -> usize { + MAX_JSON_STRING_BYTES.min(MAX_JSONL_LINE_BYTES.saturating_sub(emitted)) +} + +/// Where a byte-level scan of one record currently sits. Only +/// [`ScanState::InString`] is a legal place to stop copying: the escape states +/// mark positions where a cut would leave a dangling `\` or a half-written +/// `\uXXXX`. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +enum ScanState { + /// Structural JSON outside any string. Always copied verbatim. + #[default] + Outside, + /// Inside a string value, between complete units. + InString, + /// Inside a string, having just consumed a backslash. + Escape, + /// Inside a string, consuming the remaining hex digits of `\uXXXX`. + UnicodeEscape(u8), +} + +/// Streaming truncator for oversized JSON string values. +/// +/// Copies a record through byte by byte, dropping the interior of any string +/// longer than [`MAX_JSON_STRING_BYTES`]. A cut is taken only in +/// [`ScanState::InString`] and only on a UTF-8 leading byte, so output is +/// always valid JSON *and* valid UTF-8 no matter where the budget runs out. +/// Bytes outside strings pass through untouched, which is what keeps a +/// truncated record structurally identical to the original: same fields, same +/// nesting, same timestamps — only the oversized values are shortened. +#[derive(Debug, Default)] +struct JsonStringTruncator { + state: ScanState, + /// Bytes copied so far for the string being scanned. + string_bytes: usize, + /// Discarding the remainder of the current string. + dropping: bool, + truncated_values: usize, +} + +impl JsonStringTruncator { + fn push(&mut self, bytes: &[u8], out: &mut Vec) { + for &byte in bytes { + match self.state { + ScanState::Outside => { + out.push(byte); + if byte == b'"' { + self.state = ScanState::InString; + self.string_bytes = 0; + self.dropping = false; + } + } + ScanState::InString => { + if byte == b'"' { + if self.dropping { + out.extend_from_slice(TRUNCATION_MARKER); + self.dropping = false; + } + out.push(byte); + self.state = ScanState::Outside; + continue; + } + // Decided before the byte is copied, and only here: a + // continuation byte means we are mid-character, so keep + // going until the next character starts. + if !self.dropping + && self.string_bytes >= string_budget(out.len()) + && !is_utf8_continuation(byte) + { + self.dropping = true; + self.truncated_values += 1; + } + if byte == b'\\' { + self.state = ScanState::Escape; + } + self.copy(byte, out); + } + ScanState::Escape => { + self.state = if byte == b'u' { + ScanState::UnicodeEscape(4) + } else { + ScanState::InString + }; + self.copy(byte, out); + } + ScanState::UnicodeEscape(remaining) => { + self.state = match remaining { + 0 | 1 => ScanState::InString, + _ => ScanState::UnicodeEscape(remaining - 1), + }; + self.copy(byte, out); + } + } + } + } + + fn copy(&mut self, byte: u8, out: &mut Vec) { + if self.dropping { + return; + } + out.push(byte); + self.string_bytes += 1; + } +} + #[derive(Debug, PartialEq, Eq)] pub struct TranscriptLine { pub text: String, @@ -357,22 +491,27 @@ impl WatermarkedTranscriptReader { self.resume_state_json.as_deref() } - /// Read the next record, skipping any that exceeds [`MAX_JSONL_LINE_BYTES`]. + /// Read the next record, truncating oversized string values in place. + /// + /// A record is never rejected for being large. Its oversized string values + /// are shortened as they stream (see [`JsonStringTruncator`]), which keeps + /// the record parseable and its structure intact while bounding memory by + /// structure rather than payload. Only a record still over + /// [`MAX_JSONL_LINE_BYTES`] after that — an enormous JSON structure, or a + /// file with no newlines at all — is skipped, and even then the parse + /// continues: one unreadable record must not cost the caller the rest of + /// the file, nor every other provider loaded in the same pass. /// - /// An oversized record is drained rather than buffered, and dropped rather - /// than raised: one giant tool result must not cost the caller the rest of - /// the file — and, upstream, every other session and every other provider - /// in the same pass. Its bytes still feed the boundary window and the - /// complete-line offset, so the watermark stays byte-accurate and a later - /// resume lands on the same seam as a parse that never skipped anything. + /// The raw span is tracked alongside the truncated output so the boundary + /// window and complete-line offset keep describing the bytes on disk, and + /// a later resume lands on exactly the same seam. pub fn next_line(&mut self) -> Result, String> { loop { self.buf.clear(); let mut terminated = false; - // `Some` once this record passed the cap: from that point its bytes - // go straight to the boundary tail instead of `buf`, so peak memory - // stays bounded no matter how long the record runs. - let mut skipped: Option = None; + let mut span = RawRecordSpan::default(); + let mut truncator = JsonStringTruncator::default(); + let mut overflowed = false; loop { let available = self.reader.fill_buf().map_err(|err| { format!("Failed to read {} history line: {err}", self.error_label) @@ -382,18 +521,14 @@ impl WatermarkedTranscriptReader { } let newline = available.iter().position(|byte| *byte == b'\n'); let take = newline.map_or(available.len(), |index| index + 1); - if skipped.is_none() && self.buf.len().saturating_add(take) > MAX_JSONL_LINE_BYTES { - // Hand the already-buffered prefix over first so the - // record's bytes reach the boundary tail in file order. - let mut record = SkippedRecord::default(); - record.push(&self.buf); - self.buf.clear(); - self.buf.shrink_to_fit(); - skipped = Some(record); - } - match skipped.as_mut() { - Some(record) => record.push(&available[..take]), - None => self.buf.extend_from_slice(&available[..take]), + span.push(&available[..take]); + if !overflowed { + truncator.push(&available[..take], &mut self.buf); + if self.buf.len() > MAX_JSONL_LINE_BYTES { + overflowed = true; + self.buf.clear(); + self.buf.shrink_to_fit(); + } } self.reader.consume(take); if newline.is_some() { @@ -402,31 +537,39 @@ impl WatermarkedTranscriptReader { } } - if let Some(record) = skipped { - // An unterminated oversized record is the live writer's - // in-progress tail: leave the watermark untouched, exactly as - // an unterminated normal line does. - if !terminated { - return Ok(None); - } - push_boundary_bytes(&mut self.boundary_window, &record.tail); - self.complete_offset += record.len; + // Only a newline-terminated record is committed: a live writer may + // still be appending to an unterminated tail. + if terminated { + push_boundary_bytes(&mut self.boundary_window, &span.tail); + self.complete_offset += span.len; + } + + if overflowed { tracing::warn!( source = self.error_label, path = %self.display_path, - bytes = record.len, + bytes = span.len, limit = MAX_JSONL_LINE_BYTES, - "imported history: skipping oversized JSONL record" + "imported history: skipping record too large to truncate" ); - continue; + if terminated { + continue; + } + return Ok(None); } if self.buf.is_empty() { return Ok(None); } - if terminated { - push_boundary_bytes(&mut self.boundary_window, &self.buf); - self.complete_offset += self.buf.len() as u64; + if truncator.truncated_values > 0 { + tracing::warn!( + source = self.error_label, + path = %self.display_path, + raw_bytes = span.len, + kept_bytes = self.buf.len(), + values = truncator.truncated_values, + "imported history: truncated oversized JSON string values" + ); } let mut end = self.buf.len(); if terminated { diff --git a/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark_tests.rs b/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark_tests.rs index 5070ff0b0..01b02ea80 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark_tests.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/imported_history/watermark_tests.rs @@ -203,12 +203,11 @@ fn rotated_file_identity_forces_a_full_reparse() { cleanup(&path); } -/// A single huge tool result is ordinary in real Claude/Codex transcripts. It -/// must cost that one record and nothing else — skipping it has to leave the -/// following lines readable and the watermark landing past both, so a resume -/// does not re-read the record it already gave up on. +/// A record whose JSON structure alone blows the buffer cannot be salvaged by +/// truncating string values, so it is skipped — but the parse must continue, +/// with the watermark landing past it so a resume does not re-read it. #[test] -fn oversized_line_is_skipped_without_stopping_the_parse() { +fn unsalvageable_line_is_skipped_without_stopping_the_parse() { let path = temp_transcript("oversized", "stable\n"); let (mtime, size) = stat(&path); let mut reader = @@ -263,6 +262,148 @@ fn oversized_line_is_skipped_without_stopping_the_parse() { cleanup(&path); } +/// The common real-world case: one record carrying a multi-megabyte string +/// value — a base64 image in a `tool_result`, a long command's output. It must +/// survive as a parseable record with every structural field intact; only the +/// oversized value shortens. +#[test] +fn oversized_string_value_is_truncated_and_the_record_still_parses() { + let payload = "A".repeat(MAX_JSON_STRING_BYTES + 4096); + let record = format!( + r#"{{"type":"user","uuid":"abc-123","message":{{"role":"user","content":[{{"type":"tool_result","data":"{payload}"}}]}},"tail":"kept"}}"# + ); + let path = temp_transcript("truncate-string", &format!("{record}\n")); + let (mtime, size) = stat(&path); + let mut reader = + WatermarkedTranscriptReader::open(&path, "Test", None, 1, mtime, size).expect("open"); + let line = reader.next_line().expect("read record").expect("one record"); + + let value: serde_json::Value = + serde_json::from_str(&line.text).expect("truncated record is valid JSON"); + assert_eq!(value["type"], "user"); + assert_eq!(value["uuid"], "abc-123"); + assert_eq!(value["message"]["role"], "user"); + // Everything after the truncated value survives — truncation consumes the + // value's interior, never the structure around it. + assert_eq!(value["tail"], "kept"); + let data = value["message"]["content"][0]["data"] + .as_str() + .expect("data is a string"); + assert!(data.starts_with("AAA")); + assert!(data.ends_with("...[truncated]")); + assert!(data.len() < payload.len()); + assert_eq!(reader.next_line().expect("read eof"), None); + + cleanup(&path); +} + +/// A record can blow the buffer in aggregate while every single value stays +/// under budget — the real Claude `tool_result` carrying several images has +/// exactly this shape. A per-value budget alone would skip it; the allowance +/// has to tighten as the record fills so it still parses. +#[test] +fn many_under_budget_values_are_truncated_rather_than_overflowing() { + let value = "B".repeat(MAX_JSON_STRING_BYTES - 1); + let values = (0..24) + .map(|index| format!(r#""k{index}":"{value}""#)) + .collect::>() + .join(","); + let record = format!(r#"{{{values},"tail":"kept"}}"#); + // Every value is legal on its own, and together they far exceed the cap. + assert!(record.len() > MAX_JSONL_LINE_BYTES); + let path = temp_transcript("aggregate", &format!("{record}\n")); + let (mtime, size) = stat(&path); + let mut reader = + WatermarkedTranscriptReader::open(&path, "Test", None, 1, mtime, size).expect("open"); + let line = reader.next_line().expect("read record").expect("one record"); + + let parsed: serde_json::Value = + serde_json::from_str(&line.text).expect("truncated record is valid JSON"); + // The record survives with all 24 keys and the trailing field intact. + assert_eq!(parsed["tail"], "kept"); + assert_eq!(parsed["k0"].as_str().expect("k0").len(), value.len()); + assert!(parsed["k23"].as_str().expect("k23").ends_with("...[truncated]")); + assert!(line.text.len() <= MAX_JSONL_LINE_BYTES); + assert_eq!(reader.next_line().expect("read eof"), None); + + cleanup(&path); +} + +/// Cuts have to land between complete units. Splitting `\"` leaves a dangling +/// backslash and splitting `é` leaves half an escape — either one makes +/// the entire record unparseable. Alignment is swept so the budget runs out at +/// every offset within the repeating unit. +#[test] +fn truncation_never_splits_an_escape_sequence() { + let unit = r#"\"é"#; + for offset in 0..unit.len() { + let payload = format!( + "{}{}", + "z".repeat(offset), + unit.repeat(MAX_JSON_STRING_BYTES / unit.len() + 64) + ); + let record = format!(r#"{{"v":"{payload}","tail":"kept"}}"#); + let path = temp_transcript(&format!("escape-{offset}"), &format!("{record}\n")); + let (mtime, size) = stat(&path); + let mut reader = + WatermarkedTranscriptReader::open(&path, "Test", None, 1, mtime, size).expect("open"); + let line = reader.next_line().expect("read record").expect("one record"); + let value: serde_json::Value = serde_json::from_str(&line.text).unwrap_or_else(|err| { + panic!("offset {offset}: truncated record must stay valid JSON: {err}") + }); + assert_eq!(value["tail"], "kept"); + assert!(value["v"].as_str().expect("v").ends_with("...[truncated]")); + cleanup(&path); + } +} + +/// Cutting inside a multi-byte character would make the record invalid UTF-8, +/// which fails the whole read rather than just that value. Alignment is swept +/// so the budget runs out at every byte of a 4-byte character. +#[test] +fn truncation_never_splits_a_multibyte_character() { + for offset in 0..4 { + let payload = format!( + "{}{}", + "z".repeat(offset), + "🌍".repeat(MAX_JSON_STRING_BYTES / 4 + 64) + ); + let record = format!(r#"{{"v":"{payload}","tail":"kept"}}"#); + let path = temp_transcript(&format!("utf8-{offset}"), &format!("{record}\n")); + let (mtime, size) = stat(&path); + let mut reader = + WatermarkedTranscriptReader::open(&path, "Test", None, 1, mtime, size).expect("open"); + let line = reader + .next_line() + .unwrap_or_else(|err| panic!("offset {offset}: must stay valid UTF-8: {err}")) + .expect("one record"); + let value: serde_json::Value = + serde_json::from_str(&line.text).expect("truncated record is valid JSON"); + assert_eq!(value["tail"], "kept"); + cleanup(&path); + } +} + +/// Values under the budget must come through byte for byte — truncation is +/// strictly an over-budget path, not a lossy default. +#[test] +fn values_under_the_budget_pass_through_untouched() { + let record = r#"{"type":"user","v":"short \"quoted\" é value","n":42}"#; + let path = temp_transcript("untouched", &format!("{record}\n")); + let (mtime, size) = stat(&path); + let mut reader = + WatermarkedTranscriptReader::open(&path, "Test", None, 1, mtime, size).expect("open"); + assert_eq!( + reader.next_line().expect("read record"), + Some(TranscriptLine { + text: record.to_string(), + terminated: true, + }) + ); + + cleanup(&path); +} + /// The cap bounds the reader's buffer; it must not silently truncate a record /// that merely happens to be large. #[test] From eea38b5ed4f7e1df943360f8de28456c5b55d601 Mon Sep 17 00:00:00 2001 From: SudoMaggie <268191359+sudomaggie@users.noreply.github.com> Date: Sun, 2 Aug 2026 17:03:33 +0800 Subject: [PATCH 3/4] test(history): cover recovery of a wiped source without a clear MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The outage could leave a source at zero rows: Clear + rescan wipes cache rows, round usage and watermarks via prune_missing_records_from_conn before the resync that then raised. Recovery from that state was reasoned about but never demonstrated, and it decides the release note — whether users need another Clear + rescan or just a normal update. Wipes a synced source the same way that path does, then runs a plain incremental scan and asserts both sessions return with their token counts intact. A record with no stored signature is always offered as changed, so no clear is required to recover. Pre-commit hook ran. Total eslint: 0, total circular: 0 --- .../src/sources/pi/history_tests.rs | 61 +++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/src-tauri/crates/orgtrack-core/src/sources/pi/history_tests.rs b/src-tauri/crates/orgtrack-core/src/sources/pi/history_tests.rs index 2a488823b..8a47ff0bd 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/pi/history_tests.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/pi/history_tests.rs @@ -186,6 +186,67 @@ fn exact_leaf_depth_ignores_nested_foreign_jsonl() { fs::remove_dir_all(root).ok(); } +/// Worst case left behind by the outage this fixes: a "Clear + rescan" wipes +/// the source's cache rows, round usage and watermarks, and the resync that +/// followed then raised — leaving the source at zero rows. +/// +/// Recovery must not need a second clear. A record with no stored signature is +/// always offered as changed, so a plain incremental update re-parses and +/// re-caches the whole source. +#[test] +fn a_wiped_source_recovers_on_a_plain_incremental_update() { + let root = temp_root("wiped-recovery"); + write_session( + &root, + "--repo--", + "session-a.jsonl", + &transcript("session-a", 7), + ); + write_session( + &root, + "--repo--", + "session-b.jsonl", + &transcript("session-b", 11), + ); + let mut conn = fixture_conn(); + let config = test_config(&root); + anthropic_jsonl::list_sessions_paginated(&config, &mut conn, 10, 0).expect("cold scan"); + assert!( + imported_cache::query_cached_session_from_conn(&conn, SOURCE_PI, "--repo--/session-a") + .expect("read cached row") + .is_some() + ); + + // Exactly what Clear + rescan performs before its resync. + imported_cache::prune_missing_records_from_conn(&conn, SOURCE_PI, &[]).expect("wipe source"); + assert!( + imported_cache::query_cached_session_from_conn(&conn, SOURCE_PI, "--repo--/session-a") + .expect("read wiped row") + .is_none() + ); + assert!( + watermark::read_parse_watermark_from_conn(&conn, SOURCE_PI, "--repo--/session-a") + .expect("read wiped watermark") + .is_none() + ); + + // A plain incremental update — no clear flag anywhere on this path. + let page = anthropic_jsonl::list_sessions_paginated(&config, &mut conn, 10, 0) + .expect("incremental recovery scan"); + assert_eq!(page.sessions.len(), 2); + for (session, expected_output) in [("session-a", 7), ("session-b", 11)] { + let recovered = imported_cache::query_cached_session_from_conn( + &conn, + SOURCE_PI, + &format!("--repo--/{session}"), + ) + .expect("read recovered row") + .unwrap_or_else(|| panic!("{session} is back in the cache")); + assert_eq!(recovered.output_tokens, expected_output); + } + fs::remove_dir_all(root).ok(); +} + #[test] fn oversized_append_is_skipped_and_the_sync_still_succeeds() { let root = temp_root("oversized"); From 73433d9cc437f25aa661ab612c030740c76b1074 Mon Sep 17 00:00:00 2001 From: SudoMaggie <268191359+sudomaggie@users.noreply.github.com> Date: Sun, 2 Aug 2026 17:15:34 +0800 Subject: [PATCH 4/4] fix(history): isolate per-record failures in four more loaders MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cline, trae, workbuddy and warp still propagated a per-record parse error out of their sync loop — the same shape this branch fixed for claude_code, codex, kimi, qwen_code and anthropic_jsonl. sync_source_cache_from_conn is never reached, so nothing is written and the record keeps its old signature, and every later scan re-reads the same record and fails the same way. Per-source isolation already caps the damage at one provider rather than all eighteen, so this is no longer catastrophic. It is still permanent for whoever uses that provider: one bad record empties Cline, or Trae, or WorkBuddy, or Warp, and no amount of rescanning brings it back. cursor_cli already had exactly this guard, with a comment making the same argument, so all nine per-record loops now behave alike. No new tests: these four discover from fixed home paths with no injectable root, which is why their existing tests cover only pure helpers and never the sync loop. Adding integration coverage means adding root injection to four loaders, which is a refactor rather than part of this fix. Pre-commit hook ran. Total eslint: 0, total circular: 0 --- .../orgtrack-core/src/sources/cline/history/discovery.rs | 9 ++++++++- .../crates/orgtrack-core/src/sources/trae/history.rs | 9 ++++++++- .../crates/orgtrack-core/src/sources/warp/history.rs | 8 +++++++- .../orgtrack-core/src/sources/workbuddy/discovery.rs | 9 ++++++++- 4 files changed, 31 insertions(+), 4 deletions(-) diff --git a/src-tauri/crates/orgtrack-core/src/sources/cline/history/discovery.rs b/src-tauri/crates/orgtrack-core/src/sources/cline/history/discovery.rs index 3c1107176..7c6555b49 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/cline/history/discovery.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/cline/history/discovery.rs @@ -16,7 +16,14 @@ pub(super) fn sync_cline_history_cache(conn: &mut Connection) -> Result<(), Stri })?; let mut inputs = Vec::new(); for record in changed { - if let Some(meta) = parse_cline_session_meta(record)? { + let Some(parsed) = imported_history::skip_unparsable_record( + SOURCE_CLINE, + &record.record.source_session_id, + parse_cline_session_meta(record), + ) else { + continue; + }; + if let Some(meta) = parsed { inputs.push(session_meta_to_cache_input(meta)); } } diff --git a/src-tauri/crates/orgtrack-core/src/sources/trae/history.rs b/src-tauri/crates/orgtrack-core/src/sources/trae/history.rs index d74e7b2e2..4098fb4fe 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/trae/history.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/trae/history.rs @@ -131,7 +131,14 @@ fn sync_trae_history_cache(conn: &mut Connection) -> Result<(), String> { }; let mut inputs = Vec::new(); for record in changed { - if let Some(meta) = parse_trae_session_meta(record, &session_index)? { + let Some(parsed) = imported_history::skip_unparsable_record( + SOURCE_TRAE, + &record.source_session_id, + parse_trae_session_meta(record, &session_index), + ) else { + continue; + }; + if let Some(meta) = parsed { inputs.push(session_meta_to_cache_input(meta)); } } diff --git a/src-tauri/crates/orgtrack-core/src/sources/warp/history.rs b/src-tauri/crates/orgtrack-core/src/sources/warp/history.rs index 38f7c974a..7062d2324 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/warp/history.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/warp/history.rs @@ -177,7 +177,13 @@ fn sync_warp_history_cache(cache_conn: &mut Connection) -> Result<(), String> { for record in changed { let fallback_ms = parse_warp_timestamp_ms(&record.last_modified_at).unwrap_or(0); - let task_blobs = load_task_blobs(&source_conn, &record.conversation_id)?; + let Some(task_blobs) = imported_history::skip_unparsable_record( + SOURCE_WARP, + &record.conversation_id, + load_task_blobs(&source_conn, &record.conversation_id), + ) else { + continue; + }; let analysis = analyze_task_blobs( &format!("{WARP_SESSION_PREFIX}{}", record.conversation_id), &task_blobs, diff --git a/src-tauri/crates/orgtrack-core/src/sources/workbuddy/discovery.rs b/src-tauri/crates/orgtrack-core/src/sources/workbuddy/discovery.rs index 42a983863..b6052b270 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/workbuddy/discovery.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/workbuddy/discovery.rs @@ -16,7 +16,14 @@ pub(super) fn sync_workbuddy_history_cache(conn: &mut Connection) -> Result<(), })?; let mut inputs = Vec::new(); for record in changed { - if let Some(meta) = parse_workbuddy_session_meta(record)? { + let Some(parsed) = imported_history::skip_unparsable_record( + SOURCE_WORKBUDDY, + &record.source_session_id, + parse_workbuddy_session_meta(record), + ) else { + continue; + }; + if let Some(meta) = parsed { inputs.push(session_meta_to_cache_input(meta)); } }