From c6b9d3f0ee9167a8d927c7a0edfccd52ffb904fd Mon Sep 17 00:00:00 2001 From: VantaNode <208094903+VantaNode@users.noreply.github.com> Date: Wed, 5 Aug 2026 13:21:03 -0700 Subject: [PATCH 1/3] fix(cloud): incrementally replay imported histories --- .../src/sources/claude_code/history.rs | 103 ++- .../src/sources/claude_code/history_tests.rs | 17 + .../src/sources/codex/app/index.rs | 25 +- .../src/sources/codex/app/mod.rs | 5 +- .../src/sources/codex/app/transcript.rs | 138 ++++ .../src/sources/cursor_ide/history.rs | 18 + src-tauri/src/commands/handler_list.inc | 2 + src-tauri/src/orgtrack/history_commands.rs | 113 +++ .../imported/__tests__/sources.test.ts | 7 + .../externalHistory/imported/cloudReplay.ts | 25 + .../tauri/externalHistory/imported/index.ts | 26 + .../org2CloudSessionSync.shrink.test.ts | 9 +- .../Org2Cloud/org2CloudSessionSync.ts | 737 ++++++++++++++++-- .../Org2Cloud/org2CloudSessionSync.types.ts | 11 + src/features/Org2Cloud/org2CloudSyncAtoms.ts | 42 +- .../org2CloudSyncEngine.sessions.test.ts | 196 +++++ 16 files changed, 1389 insertions(+), 85 deletions(-) create mode 100644 src/api/tauri/externalHistory/imported/cloudReplay.ts 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 e379362ed..63aa56580 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 @@ -400,16 +400,33 @@ fn load_claude_turn_range( start_offset: u64, end_offset: u64, turn_id: &str, +) -> Result, String> { + load_claude_turn_range_with_sequence( + file, + session_id, + start_offset, + end_offset, + usize::try_from(start_offset).unwrap_or(usize::MAX), + Some(turn_id), + ) +} + +fn load_claude_turn_range_with_sequence( + file: &mut fs::File, + session_id: &str, + start_offset: u64, + end_offset: u64, + start_sequence: usize, + forced_first_user_id: Option<&str>, ) -> Result, String> { file.seek(SeekFrom::Start(start_offset)) .map_err(|err| format!("Failed to seek Claude history: {err}"))?; let take = file.take(end_offset.saturating_sub(start_offset)); - let start_sequence = usize::try_from(start_offset).unwrap_or(usize::MAX); load_claude_code_history_from_reader( session_id, BufReader::new(take), start_sequence, - Some(turn_id), + forced_first_user_id, ) } @@ -427,13 +444,13 @@ pub fn load_claude_code_initial_window_for_session( }); } - let file_len = fs::metadata(&path) + let file_len = fs::metadata(path.as_path()) .map_err(|err| format!("Failed to stat Claude history {}: {err}", path.display()))? .len(); let first_loaded_turn = indexed .len() .saturating_sub(recent_turn_count.max(1).min(indexed.len())); - let mut file = fs::File::open(&path) + let mut file = fs::File::open(path.as_path()) .map_err(|err| format!("Failed to open Claude history {}: {err}", path.display()))?; let mut chunks = Vec::with_capacity(indexed.len().saturating_mul(2)); for (index, turn) in indexed.iter().enumerate() { @@ -475,7 +492,7 @@ pub fn load_claude_code_turn_windows_for_session( let file_stem = claude_file_stem_from_session_id(session_id)?; let path = resolve_claude_session_path(conn, file_stem)?; let indexed = index_claude_user_turns(session_id, &path)?; - let file_len = fs::metadata(&path) + let file_len = fs::metadata(path.as_path()) .map_err(|err| format!("Failed to stat Claude history {}: {err}", path.display()))? .len(); let positions = indexed @@ -483,7 +500,7 @@ pub fn load_claude_code_turn_windows_for_session( .enumerate() .map(|(index, turn)| (turn.start_offset, index)) .collect::>(); - let mut file = fs::File::open(&path) + let mut file = fs::File::open(path.as_path()) .map_err(|err| format!("Failed to open Claude history {}: {err}", path.display()))?; turn_ids @@ -518,6 +535,68 @@ pub fn load_claude_code_turn_windows_for_session( .collect() } +pub fn load_claude_code_cloud_turn_windows_for_session( + conn: &Connection, + session_id: &str, + turn_ids: &[String], + start_sequence: usize, +) -> Result, String> { + let file_stem = claude_file_stem_from_session_id(session_id)?; + let path = resolve_claude_session_path(conn, file_stem)?; + load_claude_code_cloud_turn_windows_from_path(session_id, &path, turn_ids, start_sequence) +} + +fn load_claude_code_cloud_turn_windows_from_path( + session_id: &str, + path: &Path, + turn_ids: &[String], + start_sequence: usize, +) -> Result, String> { + let file_len = fs::metadata(path) + .map_err(|err| format!("Failed to stat Claude history {}: {err}", path.display()))? + .len(); + let offsets = turn_ids + .iter() + .map(|turn_id| { + claude_window_turn_offset(turn_id) + .ok_or_else(|| format!("Invalid Claude cloud turn id: {turn_id}")) + }) + .collect::, _>>()?; + if offsets + .windows(2) + .any(|pair| pair[0] >= pair[1] || pair[1] >= file_len) + || offsets.first().is_some_and(|offset| *offset >= file_len) + { + return Err("Claude cloud turn offsets are out of order or out of bounds".to_string()); + } + let mut file = fs::File::open(path) + .map_err(|err| format!("Failed to open Claude history {}: {err}", path.display()))?; + let mut next_sequence = start_sequence; + + turn_ids + .iter() + .enumerate() + .map(|(index, turn_id)| { + let offset = offsets[index]; + let end_offset = offsets.get(index + 1).copied().unwrap_or(file_len); + let chunks = load_claude_turn_range_with_sequence( + &mut file, + session_id, + offset, + end_offset, + next_sequence, + None, + )?; + next_sequence = next_sequence.saturating_add(chunks.len()); + Ok(imported_history::window::ImportedHistoryTurnWindow { + loaded_event_count: chunks.len(), + chunks, + turn_id: turn_id.clone(), + }) + }) + .collect() +} + pub fn load_claude_code_turn_index_for_session( conn: &Connection, session_id: &str, @@ -534,6 +613,18 @@ pub fn load_claude_code_turn_index_for_session( Ok(projected) } +pub fn load_claude_code_turn_ids_for_session( + conn: &Connection, + session_id: &str, +) -> Result, String> { + let file_stem = claude_file_stem_from_session_id(session_id)?; + let path = resolve_claude_session_path(conn, file_stem)?; + Ok(index_claude_user_turns(session_id, &path)? + .into_iter() + .map(|turn| turn.user_chunk.chunk_id) + .collect()) +} + /// Cheap freshness probe for one session's transcript: `(mtime_ms, size_bytes)`. /// Auto-refresh callers compare it against the previous probe and skip the /// full read/parse/merge pipeline when the source file has not changed — diff --git a/src-tauri/crates/orgtrack-core/src/sources/claude_code/history_tests.rs b/src-tauri/crates/orgtrack-core/src/sources/claude_code/history_tests.rs index 0f1143de2..d7316ced4 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/claude_code/history_tests.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/claude_code/history_tests.rs @@ -133,6 +133,23 @@ fn byte_index_discovers_rounds_without_parsing_tool_result_bodies() { assert!(!rendered.contains("third")); assert!(!rendered.contains(&large_output)); + let full = load_claude_code_history_from_path("claudecodeapp-window", &path) + .expect("load full transcript"); + let turn_ids = indexed + .iter() + .map(|turn| claude_window_turn_id(turn.start_offset)) + .collect::>(); + let cloud = + load_claude_code_cloud_turn_windows_from_path("claudecodeapp-window", &path, &turn_ids, 0) + .expect("load exact cloud turns") + .into_iter() + .flat_map(|window| window.chunks) + .collect::>(); + assert_eq!( + serde_json::to_value(cloud).expect("serialize cloud chunks"), + serde_json::to_value(full).expect("serialize full chunks") + ); + // Body-size surrogate: round 1 is followed by tool_use + tool_result + // text (3 lines); rounds 2 and 3 by one assistant line each. Placeholder // rounds surface these as bodyEventCount — without them the flat-view 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 d1283ad25..a728547e3 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 @@ -24,8 +24,9 @@ use super::meta::{ resume_codex_session_meta_with_title, session_meta_to_cache_input, CodexSessionMetaParse, }; use super::transcript::{ - load_codex_app_from_path, load_codex_app_initial_window_from_path, - load_codex_app_turn_from_path, CodexAppInitialWindow, CodexAppTurnWindow, + load_codex_app_cloud_turn_from_path, load_codex_app_from_path, + load_codex_app_initial_window_from_path, load_codex_app_turn_from_path, + load_codex_app_turn_ids_from_path, CodexAppInitialWindow, CodexAppTurnWindow, }; use super::{ CodexAppRecentPath, CodexAppSessionPage, CodexAppSourceMetadata, @@ -116,6 +117,26 @@ pub fn load_codex_app_turn_for_session( Ok(window) } +pub fn load_codex_app_turn_ids_for_session( + conn: &Connection, + session_id: &str, +) -> Result, String> { + let file_stem = codex_file_stem_from_session_id(session_id)?; + let path = resolve_codex_session_path(conn, file_stem)?; + load_codex_app_turn_ids_from_path(&path) +} + +pub fn load_codex_app_cloud_turn_for_session( + conn: &Connection, + session_id: &str, + turn_id: &str, + start_sequence: usize, +) -> Result, String> { + let file_stem = codex_file_stem_from_session_id(session_id)?; + let path = resolve_codex_session_path(conn, file_stem)?; + load_codex_app_cloud_turn_from_path(session_id, &path, turn_id, start_sequence) +} + #[derive(Debug, Clone)] struct CodexChildSessionLink { session_id: String, diff --git a/src-tauri/crates/orgtrack-core/src/sources/codex/app/mod.rs b/src-tauri/crates/orgtrack-core/src/sources/codex/app/mod.rs index a79f17f69..ba236e748 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/codex/app/mod.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/codex/app/mod.rs @@ -24,8 +24,9 @@ mod transcript; pub use index::{ codex_thread_id_from_file_stem, list_codex_app_recent_paths, list_codex_app_reconciliation_sessions, list_codex_app_sessions_paginated, - load_codex_app_for_session, load_codex_app_initial_window_for_session, - load_codex_app_turn_for_session, + load_codex_app_cloud_turn_for_session, load_codex_app_for_session, + load_codex_app_initial_window_for_session, load_codex_app_turn_for_session, + load_codex_app_turn_ids_for_session, }; pub use meta::{resolve_codex_transcript_for_thread_id_near_path, CodexTranscriptLocator}; pub(crate) use normalize::normalize_codex_tool_calls; diff --git a/src-tauri/crates/orgtrack-core/src/sources/codex/app/transcript.rs b/src-tauri/crates/orgtrack-core/src/sources/codex/app/transcript.rs index 14c17fb76..220309481 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/codex/app/transcript.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/codex/app/transcript.rs @@ -241,6 +241,7 @@ enum CodexTranscriptCollectionMode<'a> { Full, Initial { recent_turn_count: usize }, Turn { turn_id: &'a str }, + FirstTurn, } struct CompletedCodexTurn { @@ -322,6 +323,11 @@ impl<'a> CodexTranscriptCollector<'a> { } return; } + CodexTranscriptCollectionMode::FirstTurn => { + self.output.append(&mut self.current); + self.selected_turn_found = true; + return; + } CodexTranscriptCollectionMode::Initial { .. } => {} } @@ -497,6 +503,16 @@ pub fn load_codex_app_from_path( Ok(chunks) } +pub(crate) fn load_codex_app_turn_ids_from_path(path: &Path) -> Result, String> { + let signature = codex_transcript_file_signature(path)?; + let mut catalog = load_codex_turn_catalog(path, signature)?; + catalog.sort_unstable_by_key(|entry| entry.byte_offset); + Ok(catalog + .into_iter() + .map(|entry| codex_lazy_turn_id(entry.byte_offset)) + .collect()) +} + pub fn load_codex_app_initial_window_from_path( session_id: &str, path: &Path, @@ -604,6 +620,61 @@ pub fn load_codex_app_turn_from_path( }) } +pub(crate) fn load_codex_app_cloud_turn_from_path( + session_id: &str, + path: &Path, + turn_id: &str, + start_sequence: usize, +) -> Result, String> { + let Some(user_offset) = codex_lazy_turn_offset(turn_id) else { + return Ok(Vec::new()); + }; + let start_offset = codex_cloud_turn_start_offset(path, user_offset)?; + let (chunks, _, _) = load_codex_app_from_path_with_mode( + session_id, + path, + CodexTranscriptCollectionMode::FirstTurn, + start_offset, + start_sequence, + )?; + Ok(chunks) +} + +fn codex_cloud_turn_start_offset(path: &Path, user_offset: u64) -> Result { + if user_offset == 0 { + return Ok(0); + } + let read_start = user_offset.saturating_sub(CODEX_REVERSE_SCAN_MAX_LINE_BYTES as u64); + let read_len = usize::try_from(user_offset - read_start).unwrap_or_default(); + let mut file = fs::File::open(path) + .map_err(|err| format!("Failed to open Codex history {}: {err}", path.display()))?; + file.seek(SeekFrom::Start(read_start)).map_err(|err| { + format!( + "Failed to seek Codex history {} to {read_start}: {err}", + path.display() + ) + })?; + let mut prefix = vec![0u8; read_len]; + file.read_exact(&mut prefix) + .map_err(|err| format!("Failed to read Codex turn prefix: {err}"))?; + let mut line_end = prefix.len(); + while line_end > 0 && matches!(prefix[line_end - 1], b'\n' | b'\r') { + line_end -= 1; + } + let line_start = prefix[..line_end] + .iter() + .rposition(|byte| *byte == b'\n') + .map_or(0, |index| index + 1); + let Ok(previous) = serde_json::from_slice::(&prefix[line_start..line_end]) + else { + return Ok(user_offset); + }; + if previous.payload.get("type").and_then(Value::as_str) == Some("task_started") { + return Ok(read_start.saturating_add(line_start as u64)); + } + Ok(user_offset) +} + fn codex_lazy_turn_sequence(byte_offset: u64) -> usize { usize::try_from(byte_offset).unwrap_or(usize::MAX) } @@ -1972,6 +2043,73 @@ fn reasoning_text_from_payload(payload: &Value) -> Option { mod window_cache_tests { use super::*; + #[test] + fn cloud_turn_ids_are_source_offsets_in_transcript_order() { + let temp_dir = std::env::temp_dir().join(format!( + "orgii-codex-cloud-turn-ids-test-{}", + std::process::id() + )); + std::fs::create_dir_all(&temp_dir).expect("create temp dir"); + let path = temp_dir.join("rollout.jsonl"); + let first = r#"{"timestamp":"2026-08-05T10:00:00Z","payload":{"type":"user_message","message":"first"}}"#; + let assistant = r#"{"timestamp":"2026-08-05T10:00:01Z","payload":{"type":"assistant_message","message":"reply"}}"#; + let second = r#"{"timestamp":"2026-08-05T10:01:00Z","payload":{"type":"user_message","message":"second"}}"#; + std::fs::write(&path, format!("{first}\n{assistant}\n{second}\n")).expect("write fixture"); + + let ids = load_codex_app_turn_ids_from_path(&path).expect("load turn ids"); + let second_offset = first.len() + 1 + assistant.len() + 1; + assert_eq!( + ids, + vec![ + "codex-user-0".to_string(), + format!("codex-user-{second_offset}") + ] + ); + + std::fs::remove_file(&path).expect("remove fixture"); + std::fs::remove_dir(&temp_dir).expect("remove temp dir"); + } + + #[test] + fn cloud_turn_windows_preserve_full_sequence_ids() { + let temp_dir = std::env::temp_dir().join(format!( + "orgii-codex-cloud-turn-window-test-{}", + std::process::id() + )); + std::fs::create_dir_all(&temp_dir).expect("create temp dir"); + let path = temp_dir.join("rollout.jsonl"); + let content = r#"{"timestamp":"2026-08-05T10:00:00Z","payload":{"type":"task_started","turn_id":"provider-turn-1"}} +{"timestamp":"2026-08-05T10:00:01Z","payload":{"type":"user_message","message":"first"}} +{"timestamp":"2026-08-05T10:01:00Z","payload":{"type":"task_started","turn_id":"provider-turn-2"}} +{"timestamp":"2026-08-05T10:01:01Z","payload":{"type":"user_message","message":"second"}} +"#; + std::fs::write(&path, content).expect("write fixture"); + + let full = + load_codex_app_from_path("codexapp-cloud-window", &path).expect("load full transcript"); + let ids = load_codex_app_turn_ids_from_path(&path).expect("load turn ids"); + let mut cloud = Vec::new(); + let mut next_sequence = 0usize; + for turn_id in ids { + let chunks = load_codex_app_cloud_turn_from_path( + "codexapp-cloud-window", + &path, + &turn_id, + next_sequence, + ) + .expect("load cloud turn"); + next_sequence += chunks.len(); + cloud.extend(chunks); + } + assert_eq!( + serde_json::to_value(cloud).expect("serialize cloud chunks"), + serde_json::to_value(full).expect("serialize full chunks") + ); + + std::fs::remove_file(&path).expect("remove fixture"); + std::fs::remove_dir(&temp_dir).expect("remove temp dir"); + } + #[test] fn codex_turn_offset_cache_bounds_sessions_and_turns() { let signature = CodexTranscriptSignature { diff --git a/src-tauri/crates/orgtrack-core/src/sources/cursor_ide/history.rs b/src-tauri/crates/orgtrack-core/src/sources/cursor_ide/history.rs index 720361534..5b337c265 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/cursor_ide/history.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/cursor_ide/history.rs @@ -610,6 +610,24 @@ pub fn load_turn_window_for_session( }) } +pub fn load_turn_ids_for_session(session_id: &str) -> Result, String> { + let composer_id = strip_session_prefix(session_id); + let Some(cursor_conn) = open_cursor_db() else { + return Ok(Vec::new()); + }; + let composer = load_composer_for_order(&cursor_conn, composer_id)?; + let order = load_complete_bubble_order( + &cursor_conn, + composer_id, + &composer.full_conversation_headers_only, + )?; + Ok(order + .into_iter() + .filter(|header| header.bubble_type == CURSOR_BUBBLE_TYPE_USER) + .map(|header| header.bubble_id) + .collect()) +} + #[cfg(test)] #[path = "history_tests.rs"] mod tests; diff --git a/src-tauri/src/commands/handler_list.inc b/src-tauri/src/commands/handler_list.inc index e00bdc350..4a62b153c 100644 --- a/src-tauri/src/commands/handler_list.inc +++ b/src-tauri/src/commands/handler_list.inc @@ -1034,6 +1034,8 @@ orgtrack::history_commands::cursor_ide_full_refresh, orgtrack::history_commands::cursor_ide_turn_window, orgtrack::history_commands::imported_history_initial_window, orgtrack::history_commands::imported_history_turn_windows, +orgtrack::history_commands::imported_history_cloud_turn_ids, +orgtrack::history_commands::imported_history_cloud_turn_windows, orgtrack::history_commands::codex_app_chunks, orgtrack::history_commands::codex_app_initial_window, orgtrack::history_commands::codex_app_turn_window, diff --git a/src-tauri/src/orgtrack/history_commands.rs b/src-tauri/src/orgtrack/history_commands.rs index 1792cfd24..27d57cded 100644 --- a/src-tauri/src/orgtrack/history_commands.rs +++ b/src-tauri/src/orgtrack/history_commands.rs @@ -49,6 +49,7 @@ const IMPORTED_TURN_PROJECTION_CACHE_CAPACITY: usize = 8; const IMPORTED_TURN_PROJECTION_LIMIT_PER_SESSION: usize = 4_096; const CODEX_INITIAL_RECENT_TURN_COUNT: usize = 1; const IMPORTED_INITIAL_RECENT_TURN_COUNT: usize = 1; +const IMPORTED_CLOUD_TURN_WINDOW_LIMIT: usize = 50; /// Fidelity of a projection entering the cache. Window pre-warms are built /// without parsing every round body (empty `modified_files`, fabricated @@ -841,6 +842,118 @@ pub async fn imported_history_turn_windows( .map_err(|err| format!("Task join error: {err}"))? } +#[derive(serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ImportedHistoryCloudTurnWindow { + pub turn_id: String, + pub chunks: Vec, +} + +/// Ordered user-turn ids for providers whose source readers can seek to one +/// turn without materializing the complete transcript. This is intentionally +/// a capability-gated surface: callers must retain the authoritative full +/// loader as the fallback for unsupported or rewritten sources. +#[tauri::command] +pub async fn imported_history_cloud_turn_ids(session_id: String) -> Result, String> { + tokio::task::spawn_blocking(move || { + if session_id.starts_with(orgtrack_core::sources::claude_code::SESSION_PREFIX) { + let conn = open_cache_conn()?; + return claude_code_history::load_claude_code_turn_ids_for_session(&conn, &session_id); + } + if session_id.starts_with(orgtrack_core::sources::codex::SESSION_PREFIX) { + let conn = open_cache_conn()?; + return codex_app::load_codex_app_turn_ids_for_session(&conn, &session_id); + } + if session_id.starts_with(orgtrack_core::sources::cursor_ide::CURSORIDE_SESSION_PREFIX) { + return cursor_db_history::load_turn_ids_for_session(&session_id); + } + Err(format!( + "Session {session_id} does not support incremental cloud replay windows" + )) + }) + .await + .map_err(|err| format!("Task join error: {err}"))? +} + +/// Load exact user-bounded turns for incremental cloud replay preparation. +/// The limit bounds one IPC response; a larger delta safely falls back to the +/// existing full authoritative loader in the frontend. +#[tauri::command] +pub async fn imported_history_cloud_turn_windows( + session_id: String, + mut turn_ids: Vec, + start_sequence: usize, +) -> Result, String> { + if turn_ids.len() > IMPORTED_CLOUD_TURN_WINDOW_LIMIT { + return Err(format!( + "At most {IMPORTED_CLOUD_TURN_WINDOW_LIMIT} cloud replay turns can be loaded at once" + )); + } + if turn_ids.iter().any(|turn_id| turn_id.len() > 1_024) { + return Err("Imported history turn id is too long".to_string()); + } + let mut seen = HashSet::with_capacity(turn_ids.len()); + turn_ids.retain(|turn_id| seen.insert(turn_id.clone())); + if turn_ids.is_empty() { + return Ok(Vec::new()); + } + tokio::task::spawn_blocking(move || { + if session_id.starts_with(orgtrack_core::sources::claude_code::SESSION_PREFIX) { + let conn = open_cache_conn()?; + return claude_code_history::load_claude_code_cloud_turn_windows_for_session( + &conn, + &session_id, + &turn_ids, + start_sequence, + ) + .map(|windows| { + windows + .into_iter() + .map(|window| ImportedHistoryCloudTurnWindow { + turn_id: window.turn_id, + chunks: window.chunks, + }) + .collect() + }); + } + if session_id.starts_with(orgtrack_core::sources::codex::SESSION_PREFIX) { + let conn = open_cache_conn()?; + let mut next_sequence = start_sequence; + return turn_ids + .into_iter() + .map(|turn_id| { + let chunks = codex_app::load_codex_app_cloud_turn_for_session( + &conn, + &session_id, + &turn_id, + next_sequence, + )?; + next_sequence = next_sequence.saturating_add(chunks.len()); + Ok(ImportedHistoryCloudTurnWindow { turn_id, chunks }) + }) + .collect(); + } + if session_id.starts_with(orgtrack_core::sources::cursor_ide::CURSORIDE_SESSION_PREFIX) { + return turn_ids + .into_iter() + .map(|turn_id| { + let window = + cursor_db_history::load_turn_window_for_session(&session_id, &turn_id)?; + Ok(ImportedHistoryCloudTurnWindow { + turn_id, + chunks: window.chunks, + }) + }) + .collect(); + } + Err(format!( + "Session {session_id} does not support incremental cloud replay windows" + )) + }) + .await + .map_err(|err| format!("Task join error: {err}"))? +} + #[tauri::command] pub async fn codex_app_chunks( session_id: String, diff --git a/src/api/tauri/externalHistory/imported/__tests__/sources.test.ts b/src/api/tauri/externalHistory/imported/__tests__/sources.test.ts index a6f7253b2..7232b8143 100644 --- a/src/api/tauri/externalHistory/imported/__tests__/sources.test.ts +++ b/src/api/tauri/externalHistory/imported/__tests__/sources.test.ts @@ -139,6 +139,13 @@ describe("imported history source registry", () => { } }); + it("enables bounded cloud replay only for providers with exact turn seeks", () => { + const capable = IMPORTED_HISTORY_SOURCES.filter( + (source) => source.loadCloudTurnIds && source.loadCloudTurnWindows + ).map((source) => source.sourceId); + expect(capable).toEqual(["cursor_ide", "codex_app", "claude_code"]); + }); + it("resolves source metadata by session id prefix", () => { expect( getImportedHistorySourceBySessionId("codexapp-rollout-1")?.sourceId diff --git a/src/api/tauri/externalHistory/imported/cloudReplay.ts b/src/api/tauri/externalHistory/imported/cloudReplay.ts new file mode 100644 index 000000000..641dc64d7 --- /dev/null +++ b/src/api/tauri/externalHistory/imported/cloudReplay.ts @@ -0,0 +1,25 @@ +import { invoke } from "@tauri-apps/api/core"; + +import type { ActivityChunk } from "@src/types/session/session"; + +export interface ImportedHistoryCloudTurnWindow { + turnId: string; + chunks: ActivityChunk[]; +} + +export async function importedHistoryCloudTurnIds( + sessionId: string +): Promise { + return invoke("imported_history_cloud_turn_ids", { sessionId }); +} + +export async function importedHistoryCloudTurnWindows(args: { + sessionId: string; + turnIds: string[]; + startSequence: number; +}): Promise { + return invoke( + "imported_history_cloud_turn_windows", + args + ); +} diff --git a/src/api/tauri/externalHistory/imported/index.ts b/src/api/tauri/externalHistory/imported/index.ts index 042840989..defa53bd8 100644 --- a/src/api/tauri/externalHistory/imported/index.ts +++ b/src/api/tauri/externalHistory/imported/index.ts @@ -25,6 +25,11 @@ import { warpHistoryChunks } from "../sources/warp"; import { windsurfHistoryChunks } from "../sources/windsurf"; import { workBuddyHistoryChunks } from "../sources/workbuddy"; import { zcodeHistoryChunks } from "../sources/zcode"; +import { + type ImportedHistoryCloudTurnWindow, + importedHistoryCloudTurnIds, + importedHistoryCloudTurnWindows, +} from "./cloudReplay"; import { IMPORTED_HISTORY_SOURCE_DESCRIPTORS, type ImportedHistoryListCategory, @@ -40,6 +45,7 @@ export type { ImportedHistorySourceId, }; export { IMPORTED_HISTORY_SOURCE_DESCRIPTORS }; +export type { ImportedHistoryCloudTurnWindow }; export { importedHistoryInitialWindow, importedHistoryTurnWindows, @@ -55,6 +61,17 @@ export interface ImportedHistorySource extends ImportedHistorySourceDescriptor { loadPreviewChunks(sessionId: string): Promise; /** Complete source transcript used for cloud replay/fork publication. */ loadFullTranscriptChunks(sessionId: string): Promise; + /** + * Bounded turn-addressable read used by Cloud after an authoritative full + * anchor exists. Unsupported providers omit both methods and retain the + * complete-transcript fallback. + */ + loadCloudTurnIds?(sessionId: string): Promise; + loadCloudTurnWindows?( + sessionId: string, + turnIds: string[], + startSequence: number + ): Promise; /** * Optional freshness probe (one backend `stat`). When present, the replay * auto-refresh compares it against the previous tick and skips the full @@ -104,6 +121,9 @@ export const IMPORTED_HISTORY_SOURCES: readonly ImportedHistorySource[] = [ ).chunks; }, loadFullTranscriptChunks: cursorIdeChunks, + loadCloudTurnIds: importedHistoryCloudTurnIds, + loadCloudTurnWindows: (sessionId, turnIds, startSequence) => + importedHistoryCloudTurnWindows({ sessionId, turnIds, startSequence }), }, { ...descriptorFor("cursor_cli"), @@ -120,6 +140,9 @@ export const IMPORTED_HISTORY_SOURCES: readonly ImportedHistorySource[] = [ return (await codexAppInitialWindow(sessionId)).chunks; }, loadFullTranscriptChunks: codexAppChunks, + loadCloudTurnIds: importedHistoryCloudTurnIds, + loadCloudTurnWindows: (sessionId, turnIds, startSequence) => + importedHistoryCloudTurnWindows({ sessionId, turnIds, startSequence }), }, { ...descriptorFor("claude_code"), @@ -127,6 +150,9 @@ export const IMPORTED_HISTORY_SOURCES: readonly ImportedHistorySource[] = [ loadPreviewChunks: loadGenericPreviewChunks, loadFullTranscriptChunks: claudeCodeHistoryChunks, statTranscript: claudeCodeHistoryStat, + loadCloudTurnIds: importedHistoryCloudTurnIds, + loadCloudTurnWindows: (sessionId, turnIds, startSequence) => + importedHistoryCloudTurnWindows({ sessionId, turnIds, startSequence }), }, { ...descriptorFor("opencode"), diff --git a/src/features/Org2Cloud/org2CloudSessionSync.shrink.test.ts b/src/features/Org2Cloud/org2CloudSessionSync.shrink.test.ts index f3a869ad3..6a60675e8 100644 --- a/src/features/Org2Cloud/org2CloudSessionSync.shrink.test.ts +++ b/src/features/Org2Cloud/org2CloudSessionSync.shrink.test.ts @@ -56,7 +56,14 @@ async function pushPass( ): Promise { sync.beginPass(); sync.noteSessionEventActivity(SESSION.session_id); - vi.spyOn(sync, "loadPushEvents").mockResolvedValueOnce(events); + vi.spyOn( + sync as unknown as { + loadFullPushEvents: ( + sessionId: string + ) => Promise<{ events: SessionEvent[] }>; + }, + "loadFullPushEvents" + ).mockResolvedValueOnce({ events }); await sync.pushSession(AUTH, ORG_ID, SESSION, SCOPE_KEY, ACCESS); } diff --git a/src/features/Org2Cloud/org2CloudSessionSync.ts b/src/features/Org2Cloud/org2CloudSessionSync.ts index b07561e8a..28be9487f 100644 --- a/src/features/Org2Cloud/org2CloudSessionSync.ts +++ b/src/features/Org2Cloud/org2CloudSessionSync.ts @@ -36,7 +36,10 @@ import type { PreparedPushEvents, PreparedPushPlan, } from "./org2CloudSessionSync.types"; -import type { CollabSessionPushCursor } from "./org2CloudSyncAtoms"; +import type { + CollabSessionPushCursor, + ImportedReplayCheckpoint, +} from "./org2CloudSyncAtoms"; import { type CloudSessionTurnSummary, isOrg2SyncErrorCode, @@ -64,6 +67,22 @@ export const SESSION_SEGMENT_UPLOAD_BATCH_SIZE = 16; /** Hash only a bounded event window at once; large CLI histories can be GBs. */ const EVENT_HASH_CONCURRENCY = 16; +const IMPORTED_INCREMENTAL_TURN_LIMIT = 50; +const IMPORTED_INCREMENTAL_SEGMENT_LIMIT = 16; + +interface ImportedReplayAnchorDraft { + turnIds: string[]; + lastTurnStartEventIndex: number; + lastTurnStartChunkIndex: number; +} + +interface LoadedPushEvents { + events: SessionEvent[]; + anchorDraft?: ImportedReplayAnchorDraft; + precomputedEventHashes?: string[]; + precomputedLocalFrozenEventCount?: number; + baseChunkCount?: number; +} /** Per-session transient retry policy (org entitlement failures back off elsewhere). */ export const SESSION_PUSH_RETRY_BASE_MS = 60_000; @@ -111,6 +130,108 @@ async function hashEventsBounded(events: SessionEvent[]): Promise { } return hashes; } + +async function hashStringList(values: readonly string[]): Promise { + return sha256Hex(values.join("\n")); +} + +function trimMerkleFrontier( + frontier: Array +): Array { + const trimmed = [...frontier]; + while (trimmed.at(-1) == null) trimmed.pop(); + return trimmed; +} + +async function buildMerkleFrontier( + eventHashes: readonly string[] +): Promise> { + const frontier: Array = []; + let level = [...eventHashes]; + let height = 0; + while (level.length > 0) { + if (level.length % 2 === 1) { + frontier[height] = level[level.length - 1]; + level = level.slice(0, -1); + } + const parents: string[] = []; + for ( + let start = 0; + start < level.length; + start += EVENT_HASH_CONCURRENCY * 2 + ) { + const batch = level.slice(start, start + EVENT_HASH_CONCURRENCY * 2); + parents.push( + ...(await Promise.all( + Array.from({ length: batch.length / 2 }, (_, index) => + hashStringList([batch[index * 2], batch[index * 2 + 1]]) + ) + )) + ); + } + level = parents; + height += 1; + } + return trimMerkleFrontier(frontier); +} + +async function appendMerkleFrontier( + current: readonly (string | null)[], + currentCount: number, + eventHashes: readonly string[] +): Promise> { + const frontier = [...current]; + let count = currentCount; + for (const eventHash of eventHashes) { + let node = eventHash; + let height = 0; + while (Math.floor(count / 2 ** height) % 2 === 1) { + const left = frontier[height]; + if (!left) throw new Error("Invalid imported replay Merkle frontier"); + node = await hashStringList([left, node]); + frontier[height] = null; + height += 1; + } + frontier[height] = node; + count += 1; + } + return trimMerkleFrontier(frontier); +} + +async function merkleFrontierCommitment( + frontier: readonly (string | null)[], + eventCount: number +): Promise { + return sha256Hex( + stableStringify({ eventCount, frontier: trimMerkleFrontier([...frontier]) }) + ); +} + +function isValidMerkleFrontier( + frontier: readonly (string | null)[], + eventCount: number +): boolean { + if ( + !Number.isSafeInteger(eventCount) || + eventCount < 0 || + frontier.length > 54 + ) { + return false; + } + let remaining = eventCount; + for (const node of frontier) { + if (Boolean(node) !== (remaining % 2 === 1)) return false; + remaining = Math.floor(remaining / 2); + } + return remaining === 0; +} + +function lastUserChunkIndex(chunks: readonly ActivityChunk[]): number { + for (let index = chunks.length - 1; index >= 0; index -= 1) { + if (chunks[index].function === "user_message") return index; + } + return -1; +} /** * Owns one session's metadata/event push plane, including persisted cursors, * event-clean stamps, OCC re-anchors, and retract bookkeeping. @@ -319,64 +440,405 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { broadcastOrgControlChangedToPeers(orgId, "sessions"); } - /** Load the complete native or external-history transcript for upload. */ - async loadPushEvents(sessionId: string): Promise { + private async loadFullPushEvents( + sessionId: string + ): Promise { if (isImportedHistorySession(sessionId)) { const source = getImportedHistorySourceBySessionId(sessionId); - if (!source) return []; + if (!source) return { events: [] }; const chunks = await source.loadFullTranscriptChunks(sessionId); - if (!Array.isArray(chunks) || chunks.length === 0) return []; - return processChunksRust(chunks, sessionId); + if (!Array.isArray(chunks) || chunks.length === 0) { + return { events: [] }; + } + const events = await processChunksRust(chunks, sessionId); + if (!source.loadCloudTurnIds || !source.loadCloudTurnWindows) { + return { events }; + } + try { + // Source turn ids are provider-native seek cursors. They intentionally + // need not equal normalized event ids (Codex uses byte offsets here), + // so prove the final turn boundary by normalizing its exact window and + // matching it against the authoritative transcript suffix. + const turnIds = await source.loadCloudTurnIds(sessionId); + if ( + turnIds.some((turnId) => !turnId) || + new Set(turnIds).size !== turnIds.length + ) { + return { events }; + } + const lastTurnId = turnIds.at(-1); + if (lastTurnId) { + const lastTurnStartChunkIndex = lastUserChunkIndex(chunks); + if (lastTurnStartChunkIndex < 0) return { events }; + const windows = await source.loadCloudTurnWindows( + sessionId, + [lastTurnId], + lastTurnStartChunkIndex + ); + if ( + windows.length === 1 && + windows[0].turnId === lastTurnId && + windows[0].chunks.length > 0 + ) { + const lastTurnEvents = await processChunksRust( + windows[0].chunks, + sessionId + ); + const lastTurnStartEventIndex = + events.length - lastTurnEvents.length; + if ( + lastTurnEvents.length > 0 && + lastTurnStartEventIndex >= 0 && + stableStringify(events.slice(lastTurnStartEventIndex)) === + stableStringify(lastTurnEvents) + ) { + return { + events, + anchorDraft: { + turnIds, + lastTurnStartEventIndex, + lastTurnStartChunkIndex, + }, + }; + } + } + } + } catch (error) { + log.warn( + `could not establish incremental replay anchor for ${sessionId}; ` + + "using the authoritative full path", + error + ); + } + return { events }; } const persisted = await eventStoreProxy.getPersistedEvents(sessionId); - if (persisted.length > 0 || !isCliSession(sessionId)) return persisted; + if (persisted.length > 0 || !isCliSession(sessionId)) { + return { events: persisted }; + } // Live CLI sessions keep their transcript of record in the CLI's native // store (account-profile aware) and never write the events cache, so a // persisted read alone pushes a hollow session: metadata with no replay, // and the pass then stamps the event plane clean. Load the full native // transcript through the same command the session-resume path uses. const chunks = (await rpc.cli.chunks({ sessionId })) as ActivityChunk[]; - if (!Array.isArray(chunks) || chunks.length === 0) return []; - return processChunksRust(chunks, sessionId); + if (!Array.isArray(chunks) || chunks.length === 0) return { events: [] }; + return { events: await processChunksRust(chunks, sessionId) }; + } + + /** Authoritative complete loader retained for first anchor and recovery. */ + async loadPushEvents(sessionId: string): Promise { + return (await this.loadFullPushEvents(sessionId)).events; + } + + private async tryLoadIncrementalImportedPushEvents( + sessionId: string, + cursor: CollabSessionPushCursor + ): Promise<(LoadedPushEvents & { baseEventCount: number }) | null> { + const checkpoint = cursor.importedReplay; + if (!checkpoint || checkpoint.version !== 1) return null; + const source = getImportedHistorySourceBySessionId(sessionId); + if (!source?.loadCloudTurnIds || !source.loadCloudTurnWindows) return null; + if ( + !isValidMerkleFrontier( + checkpoint.frozenHashFrontier, + cursor.frozenEventCount + ) + ) { + return null; + } + if ( + (await merkleFrontierCommitment( + checkpoint.frozenHashFrontier, + cursor.frozenEventCount + )) !== cursor.frozenChainHash + ) { + return null; + } + + const turnIds = await source.loadCloudTurnIds(sessionId); + if ( + turnIds.some((turnId) => !turnId) || + new Set(turnIds).size !== turnIds.length + ) { + return null; + } + const reloadIndex = turnIds.indexOf(checkpoint.reloadTurnId); + if (reloadIndex < 0) return null; + if ( + (await hashStringList(turnIds.slice(0, reloadIndex))) !== + checkpoint.prefixTurnIdsHash + ) { + return null; + } + const reloadTurnIds = turnIds.slice(reloadIndex); + if ( + reloadTurnIds.length === 0 || + reloadTurnIds.length > IMPORTED_INCREMENTAL_TURN_LIMIT + ) { + return null; + } + const windows = await source.loadCloudTurnWindows( + sessionId, + reloadTurnIds, + checkpoint.retainedChunkCount + ); + if ( + windows.length !== reloadTurnIds.length || + windows.some( + (window, index) => + window.turnId !== reloadTurnIds[index] || window.chunks.length === 0 + ) + ) { + return null; + } + + const events: SessionEvent[] = []; + let lastTurnStartEventIndex = 0; + let precedingChunkCount = 0; + let lastTurnStartChunkIndex = 0; + for (let index = 0; index < windows.length; index += 1) { + if (index === windows.length - 1) { + lastTurnStartEventIndex = events.length; + lastTurnStartChunkIndex = precedingChunkCount; + } + events.push( + ...(await processChunksRust(windows[index].chunks, sessionId)) + ); + precedingChunkCount += windows[index].chunks.length; + } + const expectedBase = + cursor.frozenEventCount - checkpoint.frozenOverlapCount; + if ( + expectedBase < 0 || + checkpoint.retainedEventCount !== expectedBase || + checkpoint.frozenOverlapCount > events.length + ) { + return null; + } + const perEventHashes = await hashEventsBounded(events); + if ( + (await hashStringList( + perEventHashes.slice(0, checkpoint.frozenOverlapCount) + )) !== checkpoint.frozenOverlapHash + ) { + return null; + } + const totalEventCount = checkpoint.retainedEventCount + events.length; + if (totalEventCount < cursor.pushedCount) return null; + + const localFrozenEventCount = computeFrozenEventCount(events); + const priorFrozenInsideWindow = + cursor.frozenEventCount - checkpoint.retainedEventCount; + if (localFrozenEventCount < priorFrozenInsideWindow) return null; + const newFrozenEvents = events.slice( + priorFrozenInsideWindow, + localFrozenEventCount + ); + if ( + splitFrozenIntoSegments(newFrozenEvents, cursor.frozenSeq + 1).length > + IMPORTED_INCREMENTAL_SEGMENT_LIMIT + ) { + return null; + } + return { + baseEventCount: checkpoint.retainedEventCount, + baseChunkCount: checkpoint.retainedChunkCount, + events, + anchorDraft: { + turnIds, + lastTurnStartEventIndex, + lastTurnStartChunkIndex, + }, + precomputedEventHashes: perEventHashes, + precomputedLocalFrozenEventCount: localFrozenEventCount, + }; + } + + private async buildImportedReplayCheckpoint( + draft: ImportedReplayAnchorDraft | undefined, + baseEventCount: number, + baseChunkCount: number, + events: readonly SessionEvent[], + perEventHashes: readonly string[], + frozenEventCount: number, + frozenHashFrontier: Array | undefined + ): Promise { + if (!draft || draft.turnIds.length === 0 || !frozenHashFrontier) { + return undefined; + } + const retainedEventCount = baseEventCount + draft.lastTurnStartEventIndex; + if (frozenEventCount < retainedEventCount) return undefined; + const frozenOverlapCount = frozenEventCount - retainedEventCount; + if (draft.lastTurnStartEventIndex + frozenOverlapCount > events.length) { + return undefined; + } + return { + version: 1, + reloadTurnId: draft.turnIds[draft.turnIds.length - 1], + prefixTurnIdsHash: await hashStringList(draft.turnIds.slice(0, -1)), + retainedEventCount, + retainedChunkCount: baseChunkCount + draft.lastTurnStartChunkIndex, + frozenOverlapCount, + frozenOverlapHash: await hashStringList( + perEventHashes.slice( + draft.lastTurnStartEventIndex, + draft.lastTurnStartEventIndex + frozenOverlapCount + ) + ), + frozenHashFrontier, + }; + } + + private createPreparedPushEvents( + stampAtRead: number, + mode: "full" | "incremental", + baseEventCount: number, + loaded: LoadedPushEvents, + cursor?: CollabSessionPushCursor + ): PreparedPushEvents { + const { events, anchorDraft } = loaded; + let planPromise: Promise | null = null; + const plan = (): Promise => { + if (!planPromise) { + planPromise = (async () => { + const perEventHashes = + loaded.precomputedEventHashes ?? (await hashEventsBounded(events)); + const localFrozenEventCount = + loaded.precomputedLocalFrozenEventCount ?? + computeFrozenEventCount(events); + const frozenEventCount = baseEventCount + localFrozenEventCount; + const totalEventCount = baseEventCount + events.length; + const tailEvents = events.slice(localFrozenEventCount); + const tailHash = + tailEvents.length > 0 ? await computeSegmentHash(tailEvents) : null; + const usesIncrementalHash = + Boolean(anchorDraft) || + (mode === "incremental" && Boolean(cursor?.importedReplay)); + const frozenHashMode = usesIncrementalHash ? "merkle-v1" : "flat-v1"; + let frozenHashFrontier: Array | undefined; + let frozenChainHash: string; + if (mode === "incremental" && cursor) { + const priorFrozenInsideWindow = + cursor.frozenEventCount - baseEventCount; + const newFrozenHashes = perEventHashes.slice( + priorFrozenInsideWindow, + localFrozenEventCount + ); + const currentFrontier = cursor.importedReplay?.frozenHashFrontier; + if (!currentFrontier) { + throw new Error( + "Incremental imported replay lost its hash frontier" + ); + } + frozenHashFrontier = await appendMerkleFrontier( + currentFrontier, + cursor.frozenEventCount, + newFrozenHashes + ); + frozenChainHash = await merkleFrontierCommitment( + frozenHashFrontier, + frozenEventCount + ); + } else if (usesIncrementalHash) { + frozenHashFrontier = await buildMerkleFrontier( + perEventHashes.slice(0, localFrozenEventCount) + ); + frozenChainHash = await merkleFrontierCommitment( + frozenHashFrontier, + frozenEventCount + ); + } else { + frozenChainHash = await this.computeFrozenChainHash( + perEventHashes, + localFrozenEventCount + ); + } + return { + perEventHashes, + frozenHashMode, + totalEventCount, + frozenEventCount, + localFrozenEventCount, + tailEvents, + tailHash, + frozenChainHash, + importedReplay: await this.buildImportedReplayCheckpoint( + anchorDraft, + baseEventCount, + loaded.baseChunkCount ?? 0, + events, + perEventHashes, + frozenEventCount, + frozenHashFrontier + ), + }; + })(); + } + return planPromise; + }; + return { stampAtRead, mode, baseEventCount, events, plan }; + } + + private async computeFrozenHashAtCount( + perEventHashes: string[], + frozenEventCount: number, + mode: PreparedPushPlan["frozenHashMode"] + ): Promise { + if (mode === "flat-v1") { + return this.computeFrozenChainHash(perEventHashes, frozenEventCount); + } + const frontier = await buildMerkleFrontier( + perEventHashes.slice(0, frozenEventCount) + ); + return merkleFrontierCommitment(frontier, frozenEventCount); } private preparePushEventsForPass( - sessionId: string + sessionId: string, + cursor?: CollabSessionPushCursor, + forceFull = false ): Promise { - const cached = this.passPushPrepareCache.get(sessionId); + const cursorKey = + !forceFull && cursor?.importedReplay + ? stableStringify(cursor.importedReplay) + : "full"; + const prepareKey = `${sessionId}:${cursorKey}`; + const cached = this.passPushPrepareCache.get(prepareKey); if (cached) return cached; const prepared = (async (): Promise => { const stampAtRead = this.eventActivityStamps.get(sessionId) ?? 0; - const events = await this.loadPushEvents(sessionId); - let planPromise: Promise | null = null; - const plan = (): Promise => { - if (!planPromise) { - planPromise = (async () => { - const perEventHashes = await hashEventsBounded(events); - const frozenEventCount = computeFrozenEventCount(events); - const tailEvents = events.slice(frozenEventCount); - const tailHash = - tailEvents.length > 0 - ? await computeSegmentHash(tailEvents) - : null; - const frozenChainHash = await this.computeFrozenChainHash( - perEventHashes, - frozenEventCount + if (!forceFull && cursor && isImportedHistorySession(sessionId)) { + try { + const incremental = await this.tryLoadIncrementalImportedPushEvents( + sessionId, + cursor + ); + if (incremental) { + return this.createPreparedPushEvents( + stampAtRead, + "incremental", + incremental.baseEventCount, + incremental, + cursor ); - return { - perEventHashes, - frozenEventCount, - tailEvents, - tailHash, - frozenChainHash, - }; - })(); + } + } catch (error) { + log.warn( + `incremental replay preparation failed for ${sessionId}; ` + + "using the authoritative full path", + error + ); } - return planPromise; - }; - return { stampAtRead, events, plan }; + } + return this.createPreparedPushEvents( + stampAtRead, + "full", + 0, + await this.loadFullPushEvents(sessionId) + ); })(); - this.passPushPrepareCache.set(sessionId, prepared); + this.passPushPrepareCache.set(prepareKey, prepared); return prepared; } @@ -451,9 +913,9 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { ); return; } - const { stampAtRead, events, plan } = - await this.preparePushEventsForPass(sessionId); const cursor = this.getCursor(orgId, sessionId); + const prepared = await this.preparePushEventsForPass(sessionId, cursor); + const { stampAtRead, mode, baseEventCount, events } = prepared; if (!cursor && events.length === 0) { await this.upsertMetadataIfChanged( auth, @@ -467,8 +929,9 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { } const shrinkKey = `${orgId}:${sessionId}`; let confirmedShrink = false; - if (cursor && events.length < cursor.pushedCount) { - if (events.length === 0) { + const preparedPlan = await prepared.plan(); + if (cursor && preparedPlan.totalEventCount < cursor.pushedCount) { + if (preparedPlan.totalEventCount === 0) { // A hollow local read can NEVER authorize erasing the cloud copy. // An empty store (wiped cache, missing provider DB, rebuilding // import) reads zero on EVERY pass, so consecutive-pass @@ -486,18 +949,24 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { ); return; } - if (this.sessionShrinkCandidates.get(shrinkKey) === events.length) { + if ( + this.sessionShrinkCandidates.get(shrinkKey) === + preparedPlan.totalEventCount + ) { this.sessionShrinkCandidates.delete(shrinkKey); confirmedShrink = true; log.info( - `persisted read for ${sessionId} returned ${events.length} events ` + + `persisted read for ${sessionId} returned ${preparedPlan.totalEventCount} events ` + `on consecutive passes while the cloud cursor covers ` + `${cursor.pushedCount}; re-anchoring via epoch rewrite` ); } else { - this.sessionShrinkCandidates.set(shrinkKey, events.length); + this.sessionShrinkCandidates.set( + shrinkKey, + preparedPlan.totalEventCount + ); log.warn( - `persisted read for ${sessionId} returned ${events.length} events ` + + `persisted read for ${sessionId} returned ${preparedPlan.totalEventCount} events ` + `but the cloud cursor covers ${cursor.pushedCount}; skipping` ); return; @@ -508,22 +977,93 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { const { perEventHashes, + frozenHashMode, + totalEventCount, frozenEventCount, + localFrozenEventCount, tailEvents, tailHash, frozenChainHash, - } = await plan(); + importedReplay, + } = preparedPlan; + + if (cursor && mode === "incremental") { + const priorFrozenInsideWindow = cursor.frozenEventCount - baseEventCount; + const newFrozenEvents = events.slice( + priorFrozenInsideWindow, + localFrozenEventCount + ); + if ( + newFrozenEvents.length === 0 && + tailHash === cursor.tailHash && + totalEventCount === cursor.pushedCount + ) { + await this.upsertMetadataIfChanged( + auth, + orgId, + session, + scopeKey, + access + ); + if (importedReplay) { + this.setCursor({ + ...cursor, + frozenChainHash, + importedReplay, + }); + } + this.markEventPlaneClean(orgId, session, stampAtRead); + return; + } + await this.upsertMetadataIfChanged( + auth, + orgId, + session, + scopeKey, + access + ); + try { + await this.appendIncrementalSession( + auth, + orgId, + sessionId, + cursor, + newFrozenEvents, + preparedPlan + ); + } catch (error) { + if (!isOrg2SyncErrorCode(error, "ORG2_CONFLICT")) throw error; + const fullPrepared = await this.preparePushEventsForPass( + sessionId, + cursor, + true + ); + const fullPlan = await fullPrepared.plan(); + await this.rewriteSession(auth, orgId, session, scopeKey, access, { + events: fullPrepared.events, + ...fullPlan, + newEpoch: null, + }); + } + broadcastOrgControlChangedToPeers(orgId, "sessions"); + this.markEventPlaneClean(orgId, session, stampAtRead); + void this.publishTurnIndexBestEffort(auth, orgId, session, stampAtRead); + return; + } if (cursor) { let frozenIntact = !confirmedShrink && frozenEventCount >= cursor.frozenEventCount; + const cursorHashMode = cursor.importedReplay ? "merkle-v1" : "flat-v1"; + frozenIntact = frozenIntact && cursorHashMode === frozenHashMode; if (frozenIntact && cursor.frozenEventCount > 0) { const chainAtCursor = cursor.frozenEventCount === frozenEventCount ? frozenChainHash - : await this.computeFrozenChainHash( + : await this.computeFrozenHashAtCount( perEventHashes, - cursor.frozenEventCount + cursor.frozenEventCount, + frozenHashMode ); frozenIntact = chainAtCursor === cursor.frozenChainHash; } @@ -550,7 +1090,7 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { if ( newFrozenEvents.length === 0 && tailHash === cursor.tailHash && - events.length === cursor.pushedCount + totalEventCount === cursor.pushedCount ) { await this.upsertMetadataIfChanged( auth, @@ -583,10 +1123,14 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { { events, perEventHashes, + frozenHashMode, + totalEventCount, frozenEventCount, + localFrozenEventCount, frozenChainHash, tailEvents, tailHash, + importedReplay, } ); broadcastOrgControlChangedToPeers(orgId, "sessions"); @@ -603,10 +1147,14 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { await this.rewriteSession(auth, orgId, session, scopeKey, access, { events, perEventHashes, + frozenHashMode, + totalEventCount, frozenEventCount, + localFrozenEventCount, frozenChainHash, tailEvents, tailHash, + importedReplay, newEpoch: null, }); this.markEventPlaneClean(orgId, session, stampAtRead); @@ -623,10 +1171,14 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { await this.rewriteSession(auth, orgId, session, scopeKey, access, { events, perEventHashes, + frozenHashMode, + totalEventCount, frozenEventCount, + localFrozenEventCount, frozenChainHash, tailEvents, tailHash, + importedReplay, newEpoch: cursor.epoch + 1, }); this.markEventPlaneClean(orgId, session, stampAtRead); @@ -637,10 +1189,14 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { await this.rewriteSession(auth, orgId, session, scopeKey, access, { events, perEventHashes, + frozenHashMode, + totalEventCount, frozenEventCount, + localFrozenEventCount, frozenChainHash, tailEvents, tailHash, + importedReplay, newEpoch: 1, }); this.markEventPlaneClean(orgId, session, stampAtRead); @@ -710,6 +1266,49 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { } } + /** + * One bounded append for a validated imported-history suffix. Large deltas + * never enter this path: preparation falls back to the authoritative full + * planner before any network mutation. + */ + private async appendIncrementalSession( + auth: Org2CloudAuthState, + orgId: string, + sessionId: string, + cursor: CollabSessionPushCursor, + newFrozenEvents: SessionEvent[], + plan: PreparedPushPlan + ): Promise { + const frozenSegments = splitFrozenIntoSegments( + newFrozenEvents, + cursor.frozenSeq + 1 + ); + if (frozenSegments.length > IMPORTED_INCREMENTAL_SEGMENT_LIMIT) { + throw new Error("Incremental imported replay exceeded its segment bound"); + } + await this.client.appendSessionEvents(auth.accessToken, { + orgId, + sessionId, + expectedEpoch: cursor.epoch, + expectedFrozenSeq: cursor.frozenSeq, + expectedTailHash: cursor.tailHash, + newFrozenSegments: frozenSegments, + tail: plan.tailEvents.length > 0 ? plan.tailEvents : null, + totalCount: plan.totalEventCount, + }); + this.setCursor({ + orgId, + sessionId, + epoch: cursor.epoch, + frozenSeq: cursor.frozenSeq + frozenSegments.length, + pushedCount: plan.totalEventCount, + frozenEventCount: plan.frozenEventCount, + frozenChainHash: plan.frozenChainHash, + tailHash: plan.tailHash, + ...(plan.importedReplay ? { importedReplay: plan.importedReplay } : {}), + }); + } + /** * Extend an established epoch in statement-timeout-safe batches. Every * acknowledged batch advances the durable cursor, so a transport failure or @@ -722,14 +1321,7 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { sessionId: string, initialCursor: CollabSessionPushCursor, frozenSegments: ReturnType, - plan: { - events: SessionEvent[]; - perEventHashes: string[]; - frozenEventCount: number; - frozenChainHash: string; - tailEvents: SessionEvent[]; - tailHash: string | null; - } + plan: PreparedPushPlan & { events: SessionEvent[] } ): Promise { let cursor = initialCursor; // An empty frozen delta still needs one append to replace the mutable tail @@ -754,14 +1346,15 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { const nextChainHash = nextFrozenEventCount === plan.frozenEventCount ? plan.frozenChainHash - : await this.computeFrozenChainHash( + : await this.computeFrozenHashAtCount( plan.perEventHashes, - nextFrozenEventCount + nextFrozenEventCount, + plan.frozenHashMode ); const nextTail = finalBatch ? plan.tailEvents : []; const nextTailHash = finalBatch ? plan.tailHash : null; const nextPushedCount = finalBatch - ? plan.events.length + ? plan.totalEventCount : nextFrozenEventCount; await this.client.appendSessionEvents(auth.accessToken, { @@ -783,6 +1376,9 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { frozenEventCount: nextFrozenEventCount, frozenChainHash: nextChainHash, tailHash: nextTailHash, + ...(finalBatch && plan.importedReplay + ? { importedReplay: plan.importedReplay } + : {}), }; this.setCursor(cursor); } @@ -796,13 +1392,8 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { session: Session, scopeKey: string | null, access: CloudPushAccess, - plan: { + plan: PreparedPushPlan & { events: SessionEvent[]; - perEventHashes: string[]; - frozenEventCount: number; - frozenChainHash: string; - tailEvents: SessionEvent[]; - tailHash: string | null; newEpoch: number | null; } ): Promise { @@ -831,9 +1422,10 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { const initialChainHash = initialFrozenEventCount === plan.frozenEventCount ? plan.frozenChainHash - : await this.computeFrozenChainHash( + : await this.computeFrozenHashAtCount( plan.perEventHashes, - initialFrozenEventCount + initialFrozenEventCount, + plan.frozenHashMode ); await this.client.rewriteSessionEvents(auth.accessToken, { orgId, @@ -844,7 +1436,7 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { !progressive && plan.tailEvents.length > 0 ? plan.tailEvents : null, totalCount: progressive ? initialFrozenEventCount - : plan.events.length, + : plan.totalEventCount, }); const cursor: CollabSessionPushCursor = { orgId, @@ -853,10 +1445,13 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { frozenSeq: initialSegments.length, pushedCount: progressive ? initialFrozenEventCount - : plan.events.length, + : plan.totalEventCount, frozenEventCount: initialFrozenEventCount, frozenChainHash: initialChainHash, tailHash: progressive ? null : plan.tailHash, + ...(!progressive && plan.importedReplay + ? { importedReplay: plan.importedReplay } + : {}), }; this.setCursor(cursor); if (progressive) { diff --git a/src/features/Org2Cloud/org2CloudSessionSync.types.ts b/src/features/Org2Cloud/org2CloudSessionSync.types.ts index 2ae3bfb9e..52d5bfdfe 100644 --- a/src/features/Org2Cloud/org2CloudSessionSync.types.ts +++ b/src/features/Org2Cloud/org2CloudSessionSync.types.ts @@ -5,6 +5,7 @@ */ import type { SessionEvent } from "@src/engines/SessionCore/core/types"; +import type { ImportedReplayCheckpoint } from "./org2CloudSyncAtoms"; import * as org2CloudSyncClient from "./org2CloudSyncClient"; /** Client seam so tests inject fetch-free fakes. */ @@ -25,14 +26,24 @@ export type Org2CloudSyncClientDeps = Pick< export interface PreparedPushPlan { perEventHashes: string[]; + frozenHashMode: "flat-v1" | "merkle-v1"; + /** Absolute event count, including any validated omitted prefix. */ + totalEventCount: number; + /** Absolute frozen line, including any validated omitted prefix. */ frozenEventCount: number; + /** Frozen line within `PreparedPushEvents.events`. */ + localFrozenEventCount: number; tailEvents: SessionEvent[]; tailHash: string | null; frozenChainHash: string; + importedReplay?: ImportedReplayCheckpoint; } export interface PreparedPushEvents { stampAtRead: number; + mode: "full" | "incremental"; + /** Absolute count of validated events omitted from `events`. */ + baseEventCount: number; events: SessionEvent[]; plan(): Promise; } diff --git a/src/features/Org2Cloud/org2CloudSyncAtoms.ts b/src/features/Org2Cloud/org2CloudSyncAtoms.ts index 1710b3cc2..534b2ece1 100644 --- a/src/features/Org2Cloud/org2CloudSyncAtoms.ts +++ b/src/features/Org2Cloud/org2CloudSyncAtoms.ts @@ -30,8 +30,8 @@ function cloudStorageKey(name: string): string { /** * Owner-side segments push cursor, per (orgId, sessionId) — design §7.3. * The per-event hash vector itself is NOT persisted: `frozenChainHash` is a - * sha256 chain over the frozen region's per-event hashes, which detects - * frozen-region mutation with O(1) storage. Losing a cursor (reinstall, + * compact commitment over the frozen region's per-event hashes, which detects + * frozen-region mutation without retaining the transcript. Losing a cursor (reinstall, * cleared storage) is safe — the next push re-anchors through the server * OCC check (rewrite at server epoch + 1). Inherited verbatim from the * retired self-hosted engine (cloud-parity Phase E moved the type here). @@ -47,10 +47,34 @@ export interface CollabSessionPushCursor { pushedCount: number; /** Events covered by the frozen region (local frozen-line position). */ frozenEventCount: number; - /** sha256 over the concatenated per-event hashes of the frozen region. */ + /** Integrity commitment over the per-event hashes of the frozen region. */ frozenChainHash: string; /** segment_hash of the last pushed tail (null = tail was empty). */ tailHash: string | null; + /** + * Source-local checkpoint for bounded imported-history refreshes. It is + * optional so existing/native cursors retain their current wire behavior; + * losing or invalidating it only forces one authoritative full re-anchor. + */ + importedReplay?: ImportedReplayCheckpoint; +} + +export interface ImportedReplayCheckpoint { + version: 1; + /** Last user turn, reloaded because it may have been the mutable tail. */ + reloadTurnId: string; + /** Hash of every ordered turn id strictly before reloadTurnId. */ + prefixTurnIdsHash: string; + /** Absolute normalized-event count before reloadTurnId. */ + retainedEventCount: number; + /** Absolute provider chunk sequence before reloadTurnId. */ + retainedChunkCount: number; + /** Frozen events inside reloadTurnId covered by the current cloud cursor. */ + frozenOverlapCount: number; + /** Hash aggregate of those overlap events. */ + frozenOverlapHash: string; + /** Binary Merkle frontier for exactly `frozenEventCount` event hashes. */ + frozenHashFrontier: Array; } const RepoScopesSchema = z.record(z.string(), z.array(z.string())); @@ -82,6 +106,18 @@ const CloudPushCursorSchema = z.object({ frozenEventCount: z.number(), frozenChainHash: z.string(), tailHash: z.string().nullable(), + importedReplay: z + .object({ + version: z.literal(1), + reloadTurnId: z.string(), + prefixTurnIdsHash: z.string(), + retainedEventCount: z.number().int().nonnegative(), + retainedChunkCount: z.number().int().nonnegative(), + frozenOverlapCount: z.number().int().nonnegative(), + frozenOverlapHash: z.string(), + frozenHashFrontier: z.array(z.string().nullable()).max(54), + }) + .optional(), }) satisfies z.ZodType; const CloudPushCursorsSchema = z.record(z.string(), CloudPushCursorSchema); diff --git a/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts b/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts index 7f7012c49..dbc9f0bf2 100644 --- a/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts +++ b/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts @@ -1,5 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { ImportedHistorySource } from "@src/api/tauri/externalHistory"; import { rpc } from "@src/api/tauri/rpc"; import type { SessionEvent } from "@src/engines/SessionCore/core/types"; import type { Session } from "@src/store/session/sessionAtom/types"; @@ -858,6 +859,201 @@ describe("Org2CloudSyncEngine session publishing", () => { expect(loadFullTranscriptChunks).toHaveBeenCalledTimes(1); }); + it("replays only the mutable turn and appended turns after an imported full anchor", async () => { + const sessionId = "cursoride-incremental-thread-1"; + type CloudReplaySource = ImportedHistorySource & + Required< + Pick + >; + const source = getImportedHistorySourceBySessionId( + sessionId + ) as CloudReplaySource; + const turnChunks = { + "turn-a": [{ chunk_id: "raw-a", function: "user_message" }], + "turn-b": [{ chunk_id: "raw-b", function: "user_message" }], + "turn-c": [{ chunk_id: "raw-c", function: "user_message" }], + "turn-d": [{ chunk_id: "raw-d", function: "user_message" }], + } as const; + const turnEvents = { + "turn-a": [makeEvent("event-a-user"), makeEvent("event-a-result")], + "turn-b": [makeEvent("event-b-user"), makeEvent("event-b-result")], + "turn-c": [makeEvent("event-c-user"), makeEvent("event-c-result")], + "turn-d": [makeEvent("event-d-user"), makeEvent("event-d-result")], + } as const; + let authoritativeChunks: Array<{ + readonly chunk_id: string; + readonly function: string; + }> = [...turnChunks["turn-a"], ...turnChunks["turn-b"]]; + let authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ]; + const loadFullTranscriptChunks = vi + .spyOn(source, "loadFullTranscriptChunks") + .mockImplementation(async () => authoritativeChunks as never); + const loadCloudTurnIds = vi + .spyOn(source, "loadCloudTurnIds") + .mockResolvedValue(["turn-a", "turn-b"]); + const loadCloudTurnWindows = vi + .spyOn(source, "loadCloudTurnWindows") + .mockImplementation(async (_sessionId, turnIds) => + turnIds.map((turnId) => ({ + turnId, + chunks: turnChunks[turnId as keyof typeof turnChunks] as never, + })) + ); + processChunksRustMock.mockImplementation(async (chunks) => { + if (chunks === authoritativeChunks) return authoritativeEvents; + const turnId = Object.entries(turnChunks).find( + ([, candidate]) => candidate[0]?.chunk_id === chunks[0]?.chunk_id + )?.[0] as keyof typeof turnEvents | undefined; + return turnId ? [...turnEvents[turnId]] : []; + }); + store.set(sessionsAtom, [ + { ...SESSION, session_id: sessionId, orgId: "personal-org" }, + ]); + + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + + const anchored = store.get(org2CloudPushCursorsAtom)[`corg-1:${sessionId}`]; + expect(anchored.importedReplay).toMatchObject({ + reloadTurnId: "turn-b", + retainedEventCount: 2, + retainedChunkCount: 1, + frozenOverlapCount: 2, + }); + expect(anchored.importedReplay?.frozenHashFrontier.length).toBeGreaterThan( + 0 + ); + expect(client.rewriteSessionEvents).toHaveBeenCalledTimes(1); + + loadFullTranscriptChunks.mockClear(); + loadCloudTurnIds.mockResolvedValue(["turn-a", "turn-b", "turn-c"]); + client.appendSessionEvents.mockClear(); + store.set(sessionsAtom, [ + { + ...SESSION, + session_id: sessionId, + orgId: "personal-org", + updated_at: "2026-08-04T15:01:00.000Z", + }, + ]); + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + + expect(loadFullTranscriptChunks).not.toHaveBeenCalled(); + expect(loadCloudTurnWindows).toHaveBeenLastCalledWith( + sessionId, + ["turn-b", "turn-c"], + 1 + ); + expect(client.appendSessionEvents).toHaveBeenCalledTimes(1); + expect(client.appendSessionEvents.mock.calls[0][1]).toMatchObject({ + expectedEpoch: anchored.epoch, + expectedFrozenSeq: anchored.frozenSeq, + totalCount: 6, + }); + expect( + client.appendSessionEvents.mock.calls[0][1].newFrozenSegments.flatMap( + (segment) => segment.events + ) + ).toEqual(turnEvents["turn-c"]); + const advanced = store.get(org2CloudPushCursorsAtom)[`corg-1:${sessionId}`]; + expect(advanced).toMatchObject({ pushedCount: 6, frozenEventCount: 6 }); + expect(advanced.importedReplay).toMatchObject({ + reloadTurnId: "turn-c", + retainedEventCount: 4, + retainedChunkCount: 2, + frozenOverlapCount: 2, + }); + + // A changed source prefix invalidates the provider cursor even when its + // normalized event bytes happen to remain equal. Recovery is one full + // authoritative read, after which the new prefix is checkpointed. + authoritativeChunks = [ + ...turnChunks["turn-a"], + ...turnChunks["turn-b"], + ...turnChunks["turn-c"], + ]; + authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ...turnEvents["turn-c"], + ]; + loadFullTranscriptChunks.mockClear(); + loadCloudTurnIds.mockResolvedValue(["turn-x", "turn-b", "turn-c"]); + store.set(sessionsAtom, [ + { + ...SESSION, + session_id: sessionId, + orgId: "personal-org", + updated_at: "2026-08-04T15:02:00.000Z", + }, + ]); + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + expect(loadFullTranscriptChunks).toHaveBeenCalledTimes(1); + + // If another writer wins OCC after incremental preparation, recovery must + // discard the bounded suffix and re-read the full source before rewriting + // at the server's epoch. A suffix must never be used as a rewrite body. + authoritativeChunks = [ + ...turnChunks["turn-a"], + ...turnChunks["turn-b"], + ...turnChunks["turn-c"], + ...turnChunks["turn-d"], + ]; + authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ...turnEvents["turn-c"], + ...turnEvents["turn-d"], + ]; + loadFullTranscriptChunks.mockClear(); + loadCloudTurnIds.mockResolvedValue([ + "turn-x", + "turn-b", + "turn-c", + "turn-d", + ]); + client.appendSessionEvents.mockClear(); + client.rewriteSessionEvents.mockClear(); + client.appendSessionEvents.mockRejectedValueOnce(conflictError()); + client.getSessionEvents.mockResolvedValueOnce({ + epoch: 5, + frozenSeq: 9, + tailHash: "server-tail", + count: 9, + segments: [], + }); + store.set(sessionsAtom, [ + { + ...SESSION, + session_id: sessionId, + orgId: "personal-org", + updated_at: "2026-08-04T15:03:00.000Z", + }, + ]); + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + expect(client.appendSessionEvents).toHaveBeenCalledTimes(1); + expect(loadFullTranscriptChunks).toHaveBeenCalledTimes(1); + expect(client.rewriteSessionEvents).toHaveBeenCalledTimes(1); + expect(client.rewriteSessionEvents.mock.calls[0][1]).toMatchObject({ + newEpoch: 6, + totalCount: 8, + }); + + loadFullTranscriptChunks.mockRestore(); + loadCloudTurnIds.mockRestore(); + loadCloudTurnWindows.mockRestore(); + }); + it("publishes a roster-refreshed external replay in an inactive background org after one quiet timer", async () => { const sessionId = "codexapp-background-thread-1"; const source = getImportedHistorySourceBySessionId(sessionId); From 2b9a60818f89ffc0ec2a4564906115ed7455fa63 Mon Sep 17 00:00:00 2001 From: Neonforge <48338160+Neonforge98@users.noreply.github.com> Date: Wed, 5 Aug 2026 15:54:18 -0700 Subject: [PATCH 2/3] fix(cloud): harden incremental imported replay checkpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups for the bounded imported-history replay: - Fix an infinite synchronous loop in trimMerkleFrontier: a zero-frozen frontier ([].at(-1) == null with pop() a no-op) froze the renderer the first time a merkle plan carried no frozen events. - Validate the cursor's frozen chain in either hash mode instead of gating intactness on mode equality. Pre-checkpoint flat-v1 cursors now migrate to the merkle checkpoint through the ordinary delta append, and a transiently failed turn-id probe downgrades the same way — neither path re-uploads an intact history through an epoch rewrite anymore. - Adopt an upgraded checkpoint locally on the unchanged-session early-out so the next delta takes the bounded path without a network write. - Length-delimit hashStringList via stableStringify; provider-native turn ids are free-form strings and a newline join could collide across element boundaries. - Compute the shrink-dance observation count without forcing the plan, so a first shrink observation no longer hashes the full transcript just to skip the pass. - Error on unparseable Codex cloud turn ids like the Claude reader does instead of returning a silently empty window. - Extract the Merkle frontier helpers into org2CloudMerkleFrontier.ts with direct unit tests (build/append equivalence, JSON round-trip stability, frontier structure validation) plus engine tests for flat-cursor migration, transient probe failure, and checkpoint persistence. Pre-commit hook ran. Total eslint: 20, total circular: 0 --- .../src/sources/codex/app/transcript.rs | 18 +- src-tauri/src/orgtrack/history_commands.rs | 3 + .../Org2Cloud/org2CloudMerkleFrontier.test.ts | 98 +++++ .../Org2Cloud/org2CloudMerkleFrontier.ts | 145 ++++++ .../Org2Cloud/org2CloudSessionSync.ts | 189 +++----- src/features/Org2Cloud/org2CloudSyncAtoms.ts | 6 +- .../org2CloudSyncEngine.sessions.test.ts | 416 ++++++++++++++++++ 7 files changed, 754 insertions(+), 121 deletions(-) create mode 100644 src/features/Org2Cloud/org2CloudMerkleFrontier.test.ts create mode 100644 src/features/Org2Cloud/org2CloudMerkleFrontier.ts diff --git a/src-tauri/crates/orgtrack-core/src/sources/codex/app/transcript.rs b/src-tauri/crates/orgtrack-core/src/sources/codex/app/transcript.rs index 220309481..50e66edce 100644 --- a/src-tauri/crates/orgtrack-core/src/sources/codex/app/transcript.rs +++ b/src-tauri/crates/orgtrack-core/src/sources/codex/app/transcript.rs @@ -626,8 +626,12 @@ pub(crate) fn load_codex_app_cloud_turn_from_path( turn_id: &str, start_sequence: usize, ) -> Result, String> { + // Error like the Claude reader does: an unparseable id means the caller's + // checkpoint is stale or corrupt, and the frontend maps a reader error to + // the authoritative full path. A silent empty window would instead be + // indistinguishable from a legitimately empty turn. let Some(user_offset) = codex_lazy_turn_offset(turn_id) else { - return Ok(Vec::new()); + return Err(format!("Invalid Codex cloud turn id: {turn_id}")); }; let start_offset = codex_cloud_turn_start_offset(path, user_offset)?; let (chunks, _, _) = load_codex_app_from_path_with_mode( @@ -2110,6 +2114,18 @@ mod window_cache_tests { std::fs::remove_dir(&temp_dir).expect("remove temp dir"); } + #[test] + fn cloud_turn_rejects_an_unparseable_turn_id() { + let error = load_codex_app_cloud_turn_from_path( + "codexapp-cloud-window", + Path::new("unused.jsonl"), + "not-a-codex-turn-id", + 0, + ) + .expect_err("invalid id must error, not read as empty"); + assert!(error.contains("Invalid Codex cloud turn id")); + } + #[test] fn codex_turn_offset_cache_bounds_sessions_and_turns() { let signature = CodexTranscriptSignature { diff --git a/src-tauri/src/orgtrack/history_commands.rs b/src-tauri/src/orgtrack/history_commands.rs index 27d57cded..e877cc183 100644 --- a/src-tauri/src/orgtrack/history_commands.rs +++ b/src-tauri/src/orgtrack/history_commands.rs @@ -934,6 +934,9 @@ pub async fn imported_history_cloud_turn_windows( .collect(); } if session_id.starts_with(orgtrack_core::sources::cursor_ide::CURSORIDE_SESSION_PREFIX) { + // start_sequence is intentionally unused here: Cursor chunk ids + // come from stable bubble ids in the provider DB, not from a + // position-derived sequence, so windows are position-independent. return turn_ids .into_iter() .map(|turn_id| { diff --git a/src/features/Org2Cloud/org2CloudMerkleFrontier.test.ts b/src/features/Org2Cloud/org2CloudMerkleFrontier.test.ts new file mode 100644 index 000000000..ae83641e3 --- /dev/null +++ b/src/features/Org2Cloud/org2CloudMerkleFrontier.test.ts @@ -0,0 +1,98 @@ +import { describe, expect, it } from "vitest"; + +import { + appendMerkleFrontier, + buildMerkleFrontier, + hashStringList, + isValidMerkleFrontier, + merkleFrontierCommitment, +} from "./org2CloudMerkleFrontier"; + +function fakeHashes(count: number): string[] { + return Array.from({ length: count }, (_, index) => + `h${index}`.padEnd(64, "0") + ); +} + +describe("org2CloudMerkleFrontier", () => { + it("append after any split equals the batch build over the whole sequence", async () => { + // The incremental path extends the persisted frontier with only the new + // event hashes; correctness of every later commitment check rests on the + // two constructions agreeing at every count and split point. + for (const total of [0, 1, 2, 3, 4, 5, 8, 13, 16, 31, 32, 33, 64]) { + const hashes = fakeHashes(total); + const batch = await buildMerkleFrontier(hashes); + for (const split of new Set([ + 0, + 1, + Math.floor(total / 2), + Math.max(0, total - 1), + total, + ])) { + const appended = await appendMerkleFrontier( + await buildMerkleFrontier(hashes.slice(0, split)), + split, + hashes.slice(split) + ); + expect( + await merkleFrontierCommitment(appended, total), + `split ${split} of ${total}` + ).toBe(await merkleFrontierCommitment(batch, total)); + } + } + }); + + it("commitments survive a JSON persistence round trip of the frontier", async () => { + // buildMerkleFrontier leaves holes at even heights; storage persists + // them as null. Both forms must commit identically or every restart + // silently invalidates the checkpoint. + for (const total of [1, 4, 5, 21]) { + const frontier = await buildMerkleFrontier(fakeHashes(total)); + const reloaded = JSON.parse(JSON.stringify(frontier)) as Array< + string | null + >; + expect(await merkleFrontierCommitment(reloaded, total)).toBe( + await merkleFrontierCommitment(frontier, total) + ); + expect(isValidMerkleFrontier(reloaded, total)).toBe(true); + } + }); + + it("validates frontier structure against the count's binary digits", async () => { + const frontier = await buildMerkleFrontier(fakeHashes(5)); // bits 101 + expect(isValidMerkleFrontier(frontier, 5)).toBe(true); + expect(isValidMerkleFrontier(frontier, 4)).toBe(false); + expect(isValidMerkleFrontier(frontier, 6)).toBe(false); + expect(isValidMerkleFrontier(frontier, 7)).toBe(false); + expect(isValidMerkleFrontier([], 0)).toBe(true); + expect(isValidMerkleFrontier([], 1)).toBe(false); + expect(isValidMerkleFrontier(frontier, -5)).toBe(false); + expect(isValidMerkleFrontier(frontier, 5.5)).toBe(false); + expect(isValidMerkleFrontier(frontier, Number.MAX_SAFE_INTEGER + 1)).toBe( + false + ); + const oversized = Array.from({ length: 55 }, () => null); + expect(isValidMerkleFrontier(oversized, 0)).toBe(false); + }); + + it("append refuses a frontier that disagrees with its claimed count", async () => { + // Height 0 must hold a node when the count is odd; the checkpoint + // validator screens this, and append double-checks it defensively. + await expect( + appendMerkleFrontier([null], 1, fakeHashes(1)) + ).rejects.toThrow("Invalid imported replay Merkle frontier"); + }); + + it("hashes string lists without element-boundary collisions", async () => { + // Provider-native turn ids are free-form external strings; a plain + // separator join would let ["a\nb"] collide with ["a", "b"] inside + // prefixTurnIdsHash. + expect(await hashStringList(["a\nb"])).not.toBe( + await hashStringList(["a", "b"]) + ); + expect(await hashStringList([])).not.toBe(await hashStringList([""])); + expect(await hashStringList(["x", "y"])).toBe( + await hashStringList(["x", "y"]) + ); + }); +}); diff --git a/src/features/Org2Cloud/org2CloudMerkleFrontier.ts b/src/features/Org2Cloud/org2CloudMerkleFrontier.ts new file mode 100644 index 000000000..697872323 --- /dev/null +++ b/src/features/Org2Cloud/org2CloudMerkleFrontier.ts @@ -0,0 +1,145 @@ +/** + * Merkle-frontier commitments for the imported-history replay checkpoint + * (`ImportedReplayCheckpoint.frozenHashFrontier`). A frontier stores, for + * each set bit of the committed event count, the root of one perfect subtree + * over the frozen per-event hash sequence — O(log n) persisted state that + * supports O(log n) append while still committing to the entire prefix. + * + * Heights whose bit is 0 hold null. `buildMerkleFrontier` leaves them as + * array holes; every consumer normalizes holes through `[...frontier]` + * spreads (holes become explicit `undefined`, which `stableStringify` + * serializes as `null`), and JSON persistence stores them as `null` — the + * commitment is byte-identical in both forms, which the engine round-trip + * test pins down. + */ +import { + sha256Hex, + stableStringify, +} from "../TeamCollaboration/collabSyncUtils"; + +/** Hash only a bounded window at once; large CLI histories can be GBs. */ +export const EVENT_HASH_CONCURRENCY = 16; + +/** + * Ceiling on persisted frontier heights (2^54 events, far beyond any real + * transcript while staying inside safe-integer arithmetic). Mirrored by the + * `CloudPushCursorSchema` zod bound so a corrupt persisted cursor is + * rejected at load rather than trusted at validation time. + */ +export const MERKLE_FRONTIER_MAX_HEIGHT = 54; + +/** + * Order-preserving hash of a string list. The input is length-delimited via + * `stableStringify`, so values containing any separator (provider-native + * turn ids are free-form external strings) cannot collide across element + * boundaries the way a plain join could. + */ +export async function hashStringList( + values: readonly string[] +): Promise { + return sha256Hex(stableStringify(values)); +} + +function trimMerkleFrontier( + frontier: Array +): Array { + const trimmed = [...frontier]; + // The length guard is load-bearing: `[].at(-1) == null` holds and `pop()` + // leaves an empty array empty, so without it an all-null frontier — the + // legitimate commitment of zero frozen events — spins this synchronous + // loop forever and freezes the renderer. + while (trimmed.length > 0 && trimmed.at(-1) == null) trimmed.pop(); + return trimmed; +} + +/** Batch-build the frontier for a complete hash sequence. */ +export async function buildMerkleFrontier( + eventHashes: readonly string[] +): Promise> { + const frontier: Array = []; + let level = [...eventHashes]; + let height = 0; + while (level.length > 0) { + if (level.length % 2 === 1) { + frontier[height] = level[level.length - 1]; + level = level.slice(0, -1); + } + const parents: string[] = []; + for ( + let start = 0; + start < level.length; + start += EVENT_HASH_CONCURRENCY * 2 + ) { + const batch = level.slice(start, start + EVENT_HASH_CONCURRENCY * 2); + parents.push( + ...(await Promise.all( + Array.from({ length: batch.length / 2 }, (_, index) => + hashStringList([batch[index * 2], batch[index * 2 + 1]]) + ) + )) + ); + } + level = parents; + height += 1; + } + return trimMerkleFrontier(frontier); +} + +/** + * Extend a frontier that already commits to `currentCount` hashes with new + * hashes, exactly as binary-counter increments. Must agree with + * `buildMerkleFrontier` over the concatenated sequence — the property test + * in `org2CloudMerkleFrontier.test.ts` holds the two constructions equal. + */ +export async function appendMerkleFrontier( + current: readonly (string | null)[], + currentCount: number, + eventHashes: readonly string[] +): Promise> { + const frontier = [...current]; + let count = currentCount; + for (const eventHash of eventHashes) { + let node = eventHash; + let height = 0; + while (Math.floor(count / 2 ** height) % 2 === 1) { + const left = frontier[height]; + if (!left) throw new Error("Invalid imported replay Merkle frontier"); + node = await hashStringList([left, node]); + frontier[height] = null; + height += 1; + } + frontier[height] = node; + count += 1; + } + return trimMerkleFrontier(frontier); +} + +/** The persisted O(1) commitment: frontier plus its exact event count. */ +export async function merkleFrontierCommitment( + frontier: readonly (string | null)[], + eventCount: number +): Promise { + return sha256Hex( + stableStringify({ eventCount, frontier: trimMerkleFrontier([...frontier]) }) + ); +} + +/** Structural check: node presence must match the count's binary digits. */ +export function isValidMerkleFrontier( + frontier: readonly (string | null)[], + eventCount: number +): boolean { + if ( + !Number.isSafeInteger(eventCount) || + eventCount < 0 || + frontier.length > MERKLE_FRONTIER_MAX_HEIGHT + ) { + return false; + } + let remaining = eventCount; + for (const node of frontier) { + if (Boolean(node) !== (remaining % 2 === 1)) return false; + remaining = Math.floor(remaining / 2); + } + return remaining === 0; +} diff --git a/src/features/Org2Cloud/org2CloudSessionSync.ts b/src/features/Org2Cloud/org2CloudSessionSync.ts index 28be9487f..4dd364a9d 100644 --- a/src/features/Org2Cloud/org2CloudSessionSync.ts +++ b/src/features/Org2Cloud/org2CloudSessionSync.ts @@ -26,6 +26,14 @@ import { computeSegmentHash } from "../TeamCollaboration/sync/collabGzip"; import type { CloudPushAccess } from "./org2CloudAccessSettings"; import type { Org2CloudAuthState } from "./org2CloudAuthAtom"; import { broadcastOrgControlChangedToPeers } from "./org2CloudControlBus"; +import { + EVENT_HASH_CONCURRENCY, + appendMerkleFrontier, + buildMerkleFrontier, + hashStringList, + isValidMerkleFrontier, + merkleFrontierCommitment, +} from "./org2CloudMerkleFrontier"; import { buildCloudSessionMetadata, metadataPayloadForHash, @@ -65,8 +73,6 @@ const HEAD_READ_AFTER_SEQ = 2_147_483_647; */ export const SESSION_SEGMENT_UPLOAD_BATCH_SIZE = 16; -/** Hash only a bounded event window at once; large CLI histories can be GBs. */ -const EVENT_HASH_CONCURRENCY = 16; const IMPORTED_INCREMENTAL_TURN_LIMIT = 50; const IMPORTED_INCREMENTAL_SEGMENT_LIMIT = 16; @@ -131,101 +137,6 @@ async function hashEventsBounded(events: SessionEvent[]): Promise { return hashes; } -async function hashStringList(values: readonly string[]): Promise { - return sha256Hex(values.join("\n")); -} - -function trimMerkleFrontier( - frontier: Array -): Array { - const trimmed = [...frontier]; - while (trimmed.at(-1) == null) trimmed.pop(); - return trimmed; -} - -async function buildMerkleFrontier( - eventHashes: readonly string[] -): Promise> { - const frontier: Array = []; - let level = [...eventHashes]; - let height = 0; - while (level.length > 0) { - if (level.length % 2 === 1) { - frontier[height] = level[level.length - 1]; - level = level.slice(0, -1); - } - const parents: string[] = []; - for ( - let start = 0; - start < level.length; - start += EVENT_HASH_CONCURRENCY * 2 - ) { - const batch = level.slice(start, start + EVENT_HASH_CONCURRENCY * 2); - parents.push( - ...(await Promise.all( - Array.from({ length: batch.length / 2 }, (_, index) => - hashStringList([batch[index * 2], batch[index * 2 + 1]]) - ) - )) - ); - } - level = parents; - height += 1; - } - return trimMerkleFrontier(frontier); -} - -async function appendMerkleFrontier( - current: readonly (string | null)[], - currentCount: number, - eventHashes: readonly string[] -): Promise> { - const frontier = [...current]; - let count = currentCount; - for (const eventHash of eventHashes) { - let node = eventHash; - let height = 0; - while (Math.floor(count / 2 ** height) % 2 === 1) { - const left = frontier[height]; - if (!left) throw new Error("Invalid imported replay Merkle frontier"); - node = await hashStringList([left, node]); - frontier[height] = null; - height += 1; - } - frontier[height] = node; - count += 1; - } - return trimMerkleFrontier(frontier); -} - -async function merkleFrontierCommitment( - frontier: readonly (string | null)[], - eventCount: number -): Promise { - return sha256Hex( - stableStringify({ eventCount, frontier: trimMerkleFrontier([...frontier]) }) - ); -} - -function isValidMerkleFrontier( - frontier: readonly (string | null)[], - eventCount: number -): boolean { - if ( - !Number.isSafeInteger(eventCount) || - eventCount < 0 || - frontier.length > 54 - ) { - return false; - } - let remaining = eventCount; - for (const node of frontier) { - if (Boolean(node) !== (remaining % 2 === 1)) return false; - remaining = Math.floor(remaining / 2); - } - return remaining === 0; -} - function lastUserChunkIndex(chunks: readonly ActivityChunk[]): number { for (let index = chunks.length - 1; index >= 0; index -= 1) { if (chunks[index].function === "user_message") return index; @@ -794,6 +705,37 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { return merkleFrontierCommitment(frontier, frozenEventCount); } + /** + * True when a commitment over this pass's per-event hashes at the cursor's + * frozen line reproduces the cursor's stored chain hash in either hash + * mode. The cursor's likely mode is tried first; the second pass only runs + * across a flat↔merkle transition, which is rare and bounded to in-memory + * hashing of the already-loaded hash vector. + */ + private async frozenChainMatchesCursor( + cursor: CollabSessionPushCursor, + plan: PreparedPushPlan + ): Promise { + const preferred: PreparedPushPlan["frozenHashMode"] = cursor.importedReplay + ? "merkle-v1" + : "flat-v1"; + const other: PreparedPushPlan["frozenHashMode"] = + preferred === "merkle-v1" ? "flat-v1" : "merkle-v1"; + for (const mode of [preferred, other]) { + const chainAtCursor = + cursor.frozenEventCount === plan.frozenEventCount && + mode === plan.frozenHashMode + ? plan.frozenChainHash + : await this.computeFrozenHashAtCount( + plan.perEventHashes, + cursor.frozenEventCount, + mode + ); + if (chainAtCursor === cursor.frozenChainHash) return true; + } + return false; + } + private preparePushEventsForPass( sessionId: string, cursor?: CollabSessionPushCursor, @@ -929,9 +871,12 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { } const shrinkKey = `${orgId}:${sessionId}`; let confirmedShrink = false; - const preparedPlan = await prepared.plan(); - if (cursor && preparedPlan.totalEventCount < cursor.pushedCount) { - if (preparedPlan.totalEventCount === 0) { + // Equals the plan's totalEventCount without forcing the plan: the shrink + // dance below returns without pushing on its first observation, and + // hashing a GB-scale transcript just to skip would defeat this pass. + const observedTotalEventCount = baseEventCount + events.length; + if (cursor && observedTotalEventCount < cursor.pushedCount) { + if (observedTotalEventCount === 0) { // A hollow local read can NEVER authorize erasing the cloud copy. // An empty store (wiped cache, missing provider DB, rebuilding // import) reads zero on EVERY pass, so consecutive-pass @@ -950,23 +895,19 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { return; } if ( - this.sessionShrinkCandidates.get(shrinkKey) === - preparedPlan.totalEventCount + this.sessionShrinkCandidates.get(shrinkKey) === observedTotalEventCount ) { this.sessionShrinkCandidates.delete(shrinkKey); confirmedShrink = true; log.info( - `persisted read for ${sessionId} returned ${preparedPlan.totalEventCount} events ` + + `persisted read for ${sessionId} returned ${observedTotalEventCount} events ` + `on consecutive passes while the cloud cursor covers ` + `${cursor.pushedCount}; re-anchoring via epoch rewrite` ); } else { - this.sessionShrinkCandidates.set( - shrinkKey, - preparedPlan.totalEventCount - ); + this.sessionShrinkCandidates.set(shrinkKey, observedTotalEventCount); log.warn( - `persisted read for ${sessionId} returned ${preparedPlan.totalEventCount} events ` + + `persisted read for ${sessionId} returned ${observedTotalEventCount} events ` + `but the cloud cursor covers ${cursor.pushedCount}; skipping` ); return; @@ -975,6 +916,7 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { this.sessionShrinkCandidates.delete(shrinkKey); } + const preparedPlan = await prepared.plan(); const { perEventHashes, frozenHashMode, @@ -1054,18 +996,19 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { if (cursor) { let frozenIntact = !confirmedShrink && frozenEventCount >= cursor.frozenEventCount; - const cursorHashMode = cursor.importedReplay ? "merkle-v1" : "flat-v1"; - frozenIntact = frozenIntact && cursorHashMode === frozenHashMode; if (frozenIntact && cursor.frozenEventCount > 0) { - const chainAtCursor = - cursor.frozenEventCount === frozenEventCount - ? frozenChainHash - : await this.computeFrozenHashAtCount( - perEventHashes, - cursor.frozenEventCount, - frozenHashMode - ); - frozenIntact = chainAtCursor === cursor.frozenChainHash; + // The cursor's commitment may be in either hash mode: flat-v1 cursors + // predate the imported-replay checkpoint, a failed turn-id probe + // downgrades a checkpointed cursor, and an interrupted batch append + // persists a merkle commitment without its checkpoint. Both modes + // commit to the same per-event hashes, so intactness accepts a match + // in either — an intact history rides the delta append and adopts + // this plan's mode there; a mode change alone must never force the + // O(total) epoch rewrite. + frozenIntact = await this.frozenChainMatchesCursor( + cursor, + preparedPlan + ); } if (!frozenIntact) { @@ -1099,6 +1042,14 @@ export class Org2CloudSessionSync extends Org2CloudSessionSyncState { scopeKey, access ); + if (importedReplay && frozenChainHash !== cursor.frozenChainHash) { + // Same content in an upgraded hash mode: converge the local + // cursor (a checkpoint plus its merkle commitment) so the next + // delta takes the bounded path — no network write is needed. + // The downgrade direction deliberately keeps the cursor: a + // still-valid checkpoint must survive a transiently failed probe. + this.setCursor({ ...cursor, frozenChainHash, importedReplay }); + } this.markEventPlaneClean(orgId, session, stampAtRead); return; } diff --git a/src/features/Org2Cloud/org2CloudSyncAtoms.ts b/src/features/Org2Cloud/org2CloudSyncAtoms.ts index 534b2ece1..80d152cc0 100644 --- a/src/features/Org2Cloud/org2CloudSyncAtoms.ts +++ b/src/features/Org2Cloud/org2CloudSyncAtoms.ts @@ -23,6 +23,8 @@ import { z } from "zod/v4"; import { createZodJsonStorage } from "@src/util/core/storage/zodStorage"; +import { MERKLE_FRONTIER_MAX_HEIGHT } from "./org2CloudMerkleFrontier"; + function cloudStorageKey(name: string): string { return `orgii:org2-cloud-v1:${name}`; } @@ -115,7 +117,9 @@ const CloudPushCursorSchema = z.object({ retainedChunkCount: z.number().int().nonnegative(), frozenOverlapCount: z.number().int().nonnegative(), frozenOverlapHash: z.string(), - frozenHashFrontier: z.array(z.string().nullable()).max(54), + frozenHashFrontier: z + .array(z.string().nullable()) + .max(MERKLE_FRONTIER_MAX_HEIGHT), }) .optional(), }) satisfies z.ZodType; diff --git a/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts b/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts index dbc9f0bf2..f0475ace4 100644 --- a/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts +++ b/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts @@ -1054,6 +1054,422 @@ describe("Org2CloudSyncEngine session publishing", () => { loadCloudTurnWindows.mockRestore(); }); + it("upgrades a pre-checkpoint flat cursor with a delta append, never an epoch rewrite", async () => { + const sessionId = "cursoride-flat-migration-thread-1"; + type CloudReplaySource = ImportedHistorySource & + Required< + Pick + >; + const source = getImportedHistorySourceBySessionId( + sessionId + ) as CloudReplaySource; + const turnChunks = { + "turn-a": [{ chunk_id: "raw-a", function: "user_message" }], + "turn-b": [{ chunk_id: "raw-b", function: "user_message" }], + "turn-c": [{ chunk_id: "raw-c", function: "user_message" }], + "turn-d": [{ chunk_id: "raw-d", function: "user_message" }], + } as const; + const turnEvents = { + "turn-a": [makeEvent("event-a-user"), makeEvent("event-a-result")], + "turn-b": [makeEvent("event-b-user"), makeEvent("event-b-result")], + "turn-c": [makeEvent("event-c-user"), makeEvent("event-c-result")], + "turn-d": [makeEvent("event-d-user"), makeEvent("event-d-result")], + } as const; + let authoritativeChunks: Array<{ + readonly chunk_id: string; + readonly function: string; + }> = [...turnChunks["turn-a"], ...turnChunks["turn-b"]]; + let authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ]; + const loadFullTranscriptChunks = vi + .spyOn(source, "loadFullTranscriptChunks") + .mockImplementation(async () => authoritativeChunks as never); + // Duplicate ids disable the anchor probe, so the first push persists a + // pre-checkpoint flat-v1 cursor exactly like one written before this + // feature existed. + const loadCloudTurnIds = vi + .spyOn(source, "loadCloudTurnIds") + .mockResolvedValue(["turn-a", "turn-a"]); + const loadCloudTurnWindows = vi + .spyOn(source, "loadCloudTurnWindows") + .mockImplementation(async (_sessionId, turnIds) => + turnIds.map((turnId) => ({ + turnId, + chunks: turnChunks[turnId as keyof typeof turnChunks] as never, + })) + ); + processChunksRustMock.mockImplementation(async (chunks) => { + if (chunks === authoritativeChunks) return authoritativeEvents; + const turnId = Object.entries(turnChunks).find( + ([, candidate]) => candidate[0]?.chunk_id === chunks[0]?.chunk_id + )?.[0] as keyof typeof turnEvents | undefined; + return turnId ? [...turnEvents[turnId]] : []; + }); + store.set(sessionsAtom, [ + { ...SESSION, session_id: sessionId, orgId: "personal-org" }, + ]); + + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + + const flatCursor = store.get(org2CloudPushCursorsAtom)[ + `corg-1:${sessionId}` + ]; + expect(flatCursor).toMatchObject({ epoch: 1, pushedCount: 4 }); + expect(flatCursor.importedReplay).toBeUndefined(); + expect(client.rewriteSessionEvents).toHaveBeenCalledTimes(1); + + // The probe recovers and one turn is appended. Migration to the merkle + // checkpoint must ride the ordinary delta append: re-uploading the whole + // intact history would spend O(total) network on every legacy cursor. + authoritativeChunks = [ + ...turnChunks["turn-a"], + ...turnChunks["turn-b"], + ...turnChunks["turn-c"], + ]; + authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ...turnEvents["turn-c"], + ]; + loadCloudTurnIds.mockResolvedValue(["turn-a", "turn-b", "turn-c"]); + client.rewriteSessionEvents.mockClear(); + client.appendSessionEvents.mockClear(); + store.set(sessionsAtom, [ + { + ...SESSION, + session_id: sessionId, + orgId: "personal-org", + updated_at: "2026-08-04T15:01:00.000Z", + }, + ]); + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + + expect(client.rewriteSessionEvents).not.toHaveBeenCalled(); + expect(client.appendSessionEvents).toHaveBeenCalledTimes(1); + expect(client.appendSessionEvents.mock.calls[0][1]).toMatchObject({ + expectedEpoch: 1, + totalCount: 6, + }); + expect( + client.appendSessionEvents.mock.calls[0][1].newFrozenSegments.flatMap( + (segment) => segment.events + ) + ).toEqual(turnEvents["turn-c"]); + const upgraded = store.get(org2CloudPushCursorsAtom)[`corg-1:${sessionId}`]; + expect(upgraded).toMatchObject({ epoch: 1, pushedCount: 6 }); + expect(upgraded.importedReplay).toMatchObject({ reloadTurnId: "turn-c" }); + + // The upgraded checkpoint must actually enable the bounded path. + authoritativeChunks = [ + ...turnChunks["turn-a"], + ...turnChunks["turn-b"], + ...turnChunks["turn-c"], + ...turnChunks["turn-d"], + ]; + authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ...turnEvents["turn-c"], + ...turnEvents["turn-d"], + ]; + loadCloudTurnIds.mockResolvedValue([ + "turn-a", + "turn-b", + "turn-c", + "turn-d", + ]); + loadFullTranscriptChunks.mockClear(); + client.appendSessionEvents.mockClear(); + store.set(sessionsAtom, [ + { + ...SESSION, + session_id: sessionId, + orgId: "personal-org", + updated_at: "2026-08-04T15:02:00.000Z", + }, + ]); + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + + expect(loadFullTranscriptChunks).not.toHaveBeenCalled(); + expect(client.rewriteSessionEvents).not.toHaveBeenCalled(); + expect(client.appendSessionEvents).toHaveBeenCalledTimes(1); + expect(client.appendSessionEvents.mock.calls[0][1]).toMatchObject({ + totalCount: 8, + }); + + loadFullTranscriptChunks.mockRestore(); + loadCloudTurnIds.mockRestore(); + loadCloudTurnWindows.mockRestore(); + }); + + it("keeps an intact history on the append path when the turn-id probe fails transiently", async () => { + const sessionId = "cursoride-transient-probe-thread-1"; + type CloudReplaySource = ImportedHistorySource & + Required< + Pick + >; + const source = getImportedHistorySourceBySessionId( + sessionId + ) as CloudReplaySource; + const turnChunks = { + "turn-a": [{ chunk_id: "raw-a", function: "user_message" }], + "turn-b": [{ chunk_id: "raw-b", function: "user_message" }], + "turn-c": [{ chunk_id: "raw-c", function: "user_message" }], + "turn-d": [{ chunk_id: "raw-d", function: "user_message" }], + } as const; + const turnEvents = { + "turn-a": [makeEvent("event-a-user"), makeEvent("event-a-result")], + "turn-b": [makeEvent("event-b-user"), makeEvent("event-b-result")], + "turn-c": [makeEvent("event-c-user"), makeEvent("event-c-result")], + "turn-d": [makeEvent("event-d-user"), makeEvent("event-d-result")], + } as const; + let authoritativeChunks: Array<{ + readonly chunk_id: string; + readonly function: string; + }> = [...turnChunks["turn-a"], ...turnChunks["turn-b"]]; + let authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ]; + const loadFullTranscriptChunks = vi + .spyOn(source, "loadFullTranscriptChunks") + .mockImplementation(async () => authoritativeChunks as never); + const loadCloudTurnIds = vi + .spyOn(source, "loadCloudTurnIds") + .mockResolvedValue(["turn-a", "turn-b"]); + const loadCloudTurnWindows = vi + .spyOn(source, "loadCloudTurnWindows") + .mockImplementation(async (_sessionId, turnIds) => + turnIds.map((turnId) => ({ + turnId, + chunks: turnChunks[turnId as keyof typeof turnChunks] as never, + })) + ); + processChunksRustMock.mockImplementation(async (chunks) => { + if (chunks === authoritativeChunks) return authoritativeEvents; + const turnId = Object.entries(turnChunks).find( + ([, candidate]) => candidate[0]?.chunk_id === chunks[0]?.chunk_id + )?.[0] as keyof typeof turnEvents | undefined; + return turnId ? [...turnEvents[turnId]] : []; + }); + store.set(sessionsAtom, [ + { ...SESSION, session_id: sessionId, orgId: "personal-org" }, + ]); + + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + expect( + store.get(org2CloudPushCursorsAtom)[`corg-1:${sessionId}`].importedReplay + ).toBeDefined(); + expect(client.rewriteSessionEvents).toHaveBeenCalledTimes(1); + + // One turn is appended while the id probe fails for the whole pass (the + // incremental attempt and the full-path re-anchor both reject). A read + // hiccup is not evidence of history mutation: the pass must fall back to + // one full READ and a delta append, never a full re-UPLOAD. + authoritativeChunks = [ + ...turnChunks["turn-a"], + ...turnChunks["turn-b"], + ...turnChunks["turn-c"], + ]; + authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ...turnEvents["turn-c"], + ]; + loadCloudTurnIds + .mockRejectedValueOnce(new Error("transient: source db is locked")) + .mockRejectedValueOnce(new Error("transient: source db is locked")); + loadFullTranscriptChunks.mockClear(); + client.rewriteSessionEvents.mockClear(); + client.appendSessionEvents.mockClear(); + store.set(sessionsAtom, [ + { + ...SESSION, + session_id: sessionId, + orgId: "personal-org", + updated_at: "2026-08-04T15:01:00.000Z", + }, + ]); + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + + expect(loadFullTranscriptChunks).toHaveBeenCalledTimes(1); + expect(client.rewriteSessionEvents).not.toHaveBeenCalled(); + expect(client.appendSessionEvents).toHaveBeenCalledTimes(1); + expect( + client.appendSessionEvents.mock.calls[0][1].newFrozenSegments.flatMap( + (segment) => segment.events + ) + ).toEqual(turnEvents["turn-c"]); + // Without a probe there is no checkpoint to carry: the cursor downgrades + // to flat-v1 consistently instead of keeping a stale merkle checkpoint. + const downgraded = store.get(org2CloudPushCursorsAtom)[ + `corg-1:${sessionId}` + ]; + expect(downgraded).toMatchObject({ epoch: 1, pushedCount: 6 }); + expect(downgraded.importedReplay).toBeUndefined(); + + // Once the probe recovers, the next delta re-anchors the checkpoint — + // again via the ordinary append, with no rewrite anywhere in the cycle. + authoritativeChunks = [ + ...turnChunks["turn-a"], + ...turnChunks["turn-b"], + ...turnChunks["turn-c"], + ...turnChunks["turn-d"], + ]; + authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ...turnEvents["turn-c"], + ...turnEvents["turn-d"], + ]; + loadCloudTurnIds.mockResolvedValue([ + "turn-a", + "turn-b", + "turn-c", + "turn-d", + ]); + client.appendSessionEvents.mockClear(); + store.set(sessionsAtom, [ + { + ...SESSION, + session_id: sessionId, + orgId: "personal-org", + updated_at: "2026-08-04T15:02:00.000Z", + }, + ]); + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + + expect(client.rewriteSessionEvents).not.toHaveBeenCalled(); + expect(client.appendSessionEvents).toHaveBeenCalledTimes(1); + const reanchored = store.get(org2CloudPushCursorsAtom)[ + `corg-1:${sessionId}` + ]; + expect(reanchored).toMatchObject({ epoch: 1, pushedCount: 8 }); + expect(reanchored.importedReplay).toMatchObject({ reloadTurnId: "turn-d" }); + + loadFullTranscriptChunks.mockRestore(); + loadCloudTurnIds.mockRestore(); + loadCloudTurnWindows.mockRestore(); + }); + + it("survives a JSON persistence round trip of the imported replay checkpoint", async () => { + const sessionId = "cursoride-roundtrip-thread-1"; + type CloudReplaySource = ImportedHistorySource & + Required< + Pick + >; + const source = getImportedHistorySourceBySessionId( + sessionId + ) as CloudReplaySource; + const turnChunks = { + "turn-a": [{ chunk_id: "raw-a", function: "user_message" }], + "turn-b": [{ chunk_id: "raw-b", function: "user_message" }], + "turn-c": [{ chunk_id: "raw-c", function: "user_message" }], + } as const; + const turnEvents = { + "turn-a": [makeEvent("event-a-user"), makeEvent("event-a-result")], + "turn-b": [makeEvent("event-b-user"), makeEvent("event-b-result")], + "turn-c": [makeEvent("event-c-user"), makeEvent("event-c-result")], + } as const; + let authoritativeChunks: Array<{ + readonly chunk_id: string; + readonly function: string; + }> = [...turnChunks["turn-a"], ...turnChunks["turn-b"]]; + let authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ]; + const loadFullTranscriptChunks = vi + .spyOn(source, "loadFullTranscriptChunks") + .mockImplementation(async () => authoritativeChunks as never); + const loadCloudTurnIds = vi + .spyOn(source, "loadCloudTurnIds") + .mockResolvedValue(["turn-a", "turn-b"]); + const loadCloudTurnWindows = vi + .spyOn(source, "loadCloudTurnWindows") + .mockImplementation(async (_sessionId, turnIds) => + turnIds.map((turnId) => ({ + turnId, + chunks: turnChunks[turnId as keyof typeof turnChunks] as never, + })) + ); + processChunksRustMock.mockImplementation(async (chunks) => { + if (chunks === authoritativeChunks) return authoritativeEvents; + const turnId = Object.entries(turnChunks).find( + ([, candidate]) => candidate[0]?.chunk_id === chunks[0]?.chunk_id + )?.[0] as keyof typeof turnEvents | undefined; + return turnId ? [...turnEvents[turnId]] : []; + }); + store.set(sessionsAtom, [ + { ...SESSION, session_id: sessionId, orgId: "personal-org" }, + ]); + + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + expect( + store.get(org2CloudPushCursorsAtom)[`corg-1:${sessionId}`].importedReplay + ).toBeDefined(); + + // The merkle frontier is built with array holes at even heights; the + // storage layer persists them as JSON null. The commitment recomputed + // from the reloaded null form must still match the stored chain hash, + // otherwise every app restart silently loses the bounded path. + store.set( + org2CloudPushCursorsAtom, + JSON.parse(JSON.stringify(store.get(org2CloudPushCursorsAtom))) + ); + + authoritativeChunks = [ + ...turnChunks["turn-a"], + ...turnChunks["turn-b"], + ...turnChunks["turn-c"], + ]; + authoritativeEvents = [ + ...turnEvents["turn-a"], + ...turnEvents["turn-b"], + ...turnEvents["turn-c"], + ]; + loadCloudTurnIds.mockResolvedValue(["turn-a", "turn-b", "turn-c"]); + loadFullTranscriptChunks.mockClear(); + client.appendSessionEvents.mockClear(); + store.set(sessionsAtom, [ + { + ...SESSION, + session_id: sessionId, + orgId: "personal-org", + updated_at: "2026-08-04T15:01:00.000Z", + }, + ]); + await engine.runSyncPass(); + vi.setSystemTime(Date.now() + EXTERNAL_HISTORY_ACTIVITY_DEBOUNCE_MS + 1); + await engine.runSyncPass(); + + expect(loadFullTranscriptChunks).not.toHaveBeenCalled(); + expect(client.appendSessionEvents).toHaveBeenCalledTimes(1); + expect(client.appendSessionEvents.mock.calls[0][1]).toMatchObject({ + totalCount: 6, + }); + + loadFullTranscriptChunks.mockRestore(); + loadCloudTurnIds.mockRestore(); + loadCloudTurnWindows.mockRestore(); + }); + it("publishes a roster-refreshed external replay in an inactive background org after one quiet timer", async () => { const sessionId = "codexapp-background-thread-1"; const source = getImportedHistorySourceBySessionId(sessionId); From 88aadae8c7c430f5cda6768a2cd512049f28e102 Mon Sep 17 00:00:00 2001 From: Neonforge <48338160+Neonforge98@users.noreply.github.com> Date: Wed, 5 Aug 2026 20:35:46 -0700 Subject: [PATCH 3/3] fix(cloud): drop only malformed push cursors instead of the whole store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit createZodJsonStorage answers a failed whole-store parse with the initial value, so ONE malformed cursor entry (disk corruption, or a future checkpoint version rolled back onto this build) reset EVERY push cursor at load — and a full reset re-anchors every previously pushed session through an epoch rewrite on its next pass, an epoch-churn storm in the #608 shape whose individual rewrites all look legitimate. The imported replay checkpoint made this failure surface strictly larger (literal version, bounded frontier, integer constraints), so the cursors record now parses per entry and sheds only invalid ones: losing one cursor is the designed recovery — that session alone re-anchors through the server OCC check. Pre-commit hook ran. Total eslint: 18, total circular: 0 --- .../Org2Cloud/org2CloudSyncAtoms.test.ts | 65 +++++++++++++++++++ src/features/Org2Cloud/org2CloudSyncAtoms.ts | 29 ++++++++- 2 files changed, 93 insertions(+), 1 deletion(-) create mode 100644 src/features/Org2Cloud/org2CloudSyncAtoms.test.ts diff --git a/src/features/Org2Cloud/org2CloudSyncAtoms.test.ts b/src/features/Org2Cloud/org2CloudSyncAtoms.test.ts new file mode 100644 index 000000000..f3a0b79ab --- /dev/null +++ b/src/features/Org2Cloud/org2CloudSyncAtoms.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, it, vi } from "vitest"; + +import { CloudPushCursorsSchema } from "./org2CloudSyncAtoms"; + +const VALID_CURSOR = { + orgId: "org-1", + sessionId: "session-1", + epoch: 1, + frozenSeq: 2, + pushedCount: 6, + frozenEventCount: 6, + frozenChainHash: "hash", + tailHash: null, + importedReplay: { + version: 1, + reloadTurnId: "turn-b", + prefixTurnIdsHash: "prefix-hash", + retainedEventCount: 4, + retainedChunkCount: 4, + frozenOverlapCount: 2, + frozenOverlapHash: "overlap-hash", + frozenHashFrontier: [null, "node-a", "node-b"], + }, +}; + +describe("CloudPushCursorsSchema", () => { + it("drops only the malformed entries, never the whole store", () => { + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + // One healthy cursor beside every rollback/corruption shape the store + // can meet: a future checkpoint version, an oversized frontier, and a + // non-object entry. A whole-store reset here would re-anchor EVERY + // pushed session through an epoch rewrite; dropping one entry costs + // exactly one session's re-anchor. + const parsed = CloudPushCursorsSchema.parse({ + "org-1:session-1": VALID_CURSOR, + "org-1:future-version": { + ...VALID_CURSOR, + importedReplay: { ...VALID_CURSOR.importedReplay, version: 2 }, + }, + "org-1:oversized-frontier": { + ...VALID_CURSOR, + importedReplay: { + ...VALID_CURSOR.importedReplay, + frozenHashFrontier: Array.from({ length: 55 }, () => null), + }, + }, + "org-1:not-a-cursor": "garbage", + }); + expect(Object.keys(parsed)).toEqual(["org-1:session-1"]); + expect(parsed["org-1:session-1"]).toEqual(VALID_CURSOR); + expect(warn).toHaveBeenCalledTimes(3); + warn.mockRestore(); + }); + + it("keeps flat cursors without a checkpoint untouched", () => { + const { importedReplay: _checkpoint, ...flat } = VALID_CURSOR; + const parsed = CloudPushCursorsSchema.parse({ "org-1:flat": flat }); + expect(parsed["org-1:flat"]).toEqual(flat); + expect(parsed["org-1:flat"].importedReplay).toBeUndefined(); + }); + + it("parses an empty store to an empty record", () => { + expect(CloudPushCursorsSchema.parse({})).toEqual({}); + }); +}); diff --git a/src/features/Org2Cloud/org2CloudSyncAtoms.ts b/src/features/Org2Cloud/org2CloudSyncAtoms.ts index 80d152cc0..bb847f2b0 100644 --- a/src/features/Org2Cloud/org2CloudSyncAtoms.ts +++ b/src/features/Org2Cloud/org2CloudSyncAtoms.ts @@ -124,7 +124,34 @@ const CloudPushCursorSchema = z.object({ .optional(), }) satisfies z.ZodType; -const CloudPushCursorsSchema = z.record(z.string(), CloudPushCursorSchema); +/** + * Per-entry tolerant store parse. `createZodJsonStorage` answers a failed + * whole-store parse with the initial value — for this store that would reset + * EVERY push cursor, and a full reset re-anchors every previously pushed + * session through an epoch rewrite on its next pass (fleet-wide churn in the + * #608 shape). One malformed entry (disk corruption, or a future checkpoint + * version rolled back to this build) must instead cost exactly one cursor: + * losing one is the designed recovery — that session alone re-anchors + * through the server OCC check. + */ +export const CloudPushCursorsSchema = z + .record(z.string(), z.unknown()) + .transform((entries) => { + const cursors: Record = {}; + for (const [key, value] of Object.entries(entries)) { + const parsed = CloudPushCursorSchema.safeParse(value); + if (parsed.success) { + cursors[key] = parsed.data; + } else { + // Rate limiting is unnecessary: this runs once per storage load. + console.warn( + `[org2CloudSyncAtoms] dropped invalid push cursor "${key}"; ` + + "its session re-anchors on the next pass" + ); + } + } + return cursors; + }); /** Keyed by `${orgId}:${sessionId}` (cloud org ids, no collision risk). */ export const org2CloudPushCursorsAtom = atomWithStorage<