diff --git a/crates/session-store/src/fs_store.rs b/crates/session-store/src/fs_store.rs index 19a2c6d00..e5babb45c 100644 --- a/crates/session-store/src/fs_store.rs +++ b/crates/session-store/src/fs_store.rs @@ -211,7 +211,11 @@ impl FsStore { } } - fn parse_jsonl(content: &[u8]) -> Result, StoreError> { + fn parse_jsonl( + content: &[u8], + session_id: SessionId, + segment_id: SegmentId, + ) -> Result, StoreError> { let complete = Self::complete_jsonl_prefix(content); let content = std::str::from_utf8(complete).map_err(|error| StoreError::Corrupt { line: complete[..error.valid_up_to()] @@ -222,15 +226,19 @@ impl FsStore { message: error.to_string(), })?; let mut entries = Vec::new(); - for (i, line) in content.lines().enumerate() { - if line.trim().is_empty() { - continue; + let mut record_start = 0_u64; + for (index, record) in content.split_inclusive('\n').enumerate() { + let line = record.strip_suffix('\n').unwrap_or(record); + if !line.trim().is_empty() { + let mut entry: LogEntry = + serde_json::from_str(line).map_err(|error| StoreError::Corrupt { + line: index + 1, + message: error.to_string(), + })?; + entry.ensure_legacy_session_entry_id(session_id, segment_id, record_start); + entries.push(entry); } - let entry: T = serde_json::from_str(line).map_err(|e| StoreError::Corrupt { - line: i + 1, - message: e.to_string(), - })?; - entries.push(entry); + record_start = record_start.saturating_add(record.len() as u64); } Ok(entries) } @@ -294,7 +302,7 @@ impl Store for FsStore { return Err(StoreError::NotFound(segment_id)); } let content = fs::read(&path)?; - Self::parse_jsonl(&content) + Self::parse_jsonl(&content, session_id, segment_id) } fn list_sessions(&self) -> Result, StoreError> { diff --git a/crates/session-store/src/history.rs b/crates/session-store/src/history.rs index e960bfec5..4ed2bd2bb 100644 --- a/crates/session-store/src/history.rs +++ b/crates/session-store/src/history.rs @@ -4,7 +4,10 @@ use serde::{Deserialize, Serialize}; use crate::LoggedItem; -/// Stable logical identity of one model-visible history entry. +/// Stable logical identity of one durable public Session entry. +/// +/// Model-visible history entries and typed Run transitions share this identity +/// space so every projection can reconcile the same committed record. /// /// This value is generated at the trusted Worker session boundary and copied /// unchanged across fork, rewind, compaction retention, restore, and reboot. diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index fe3ab0e65..651e8d2c4 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -1041,15 +1041,16 @@ fn project_session_snapshot_for_segment( )), LogEntry::RunYielded { ts, + entry_id, reason, active_run_turn_count: count, } => { active_run_turn_count = Some(*count); - entries.push(legacy_entry( + entries.push(run_transition_entry( + entry_id.as_ref(), &session_key, segment_id.as_ref(), log_index, - 0, *ts, SessionSnapshotEntryData::RunYielded { reason: *reason, @@ -1059,15 +1060,16 @@ fn project_session_snapshot_for_segment( } LogEntry::RunResumed { ts, + entry_id, source, active_run_turn_count: count, } => { active_run_turn_count = Some(*count); - entries.push(legacy_entry( + entries.push(run_transition_entry( + entry_id.as_ref(), &session_key, segment_id.as_ref(), log_index, - 0, *ts, SessionSnapshotEntryData::RunResumed { source: *source, @@ -1075,29 +1077,30 @@ fn project_session_snapshot_for_segment( }, )); } - LogEntry::RunCancelled { ts } => { + LogEntry::RunCancelled { ts, entry_id } => { active_run_turn_count = None; - entries.push(legacy_entry( + entries.push(run_transition_entry( + entry_id.as_ref(), &session_key, segment_id.as_ref(), log_index, - 0, *ts, SessionSnapshotEntryData::RunCancelled, )); } LogEntry::RunErrored { ts, + entry_id, message, failure, .. } => { active_run_turn_count = None; - entries.push(legacy_entry( + entries.push(run_transition_entry( + entry_id.as_ref(), &session_key, segment_id.as_ref(), log_index, - 0, *ts, SessionSnapshotEntryData::RunError { message: message.clone(), @@ -1234,6 +1237,25 @@ fn derivation_ids(entry: &crate::LoggedSystemHistoryEntry) -> Vec { .unwrap_or_default() } +fn run_transition_entry( + persisted_entry_id: Option<&crate::LoggedSessionHistoryEntryId>, + session_key: &SessionId, + segment_id: Option<&SegmentId>, + log_index: usize, + timestamp: u64, + data: SessionSnapshotEntryData, +) -> SessionSnapshotEntry { + SessionSnapshotEntry { + entry_id: persisted_entry_id + .map(|entry_id| entry_id.0.clone()) + .unwrap_or_else(|| legacy_entry_id(session_key, segment_id, log_index, 0)), + timestamp, + provenance: SessionEntryProvenance::LegacyUnknown, + derived_from: Vec::new(), + data, + } +} + fn legacy_entry( session_key: &SessionId, segment_id: Option<&SegmentId>, @@ -2812,22 +2834,344 @@ mod tests { )); } + fn run_entry_ids(entries: &[SessionSnapshotEntry]) -> Vec { + entries + .iter() + .filter(|entry| { + matches!( + entry.data, + SessionSnapshotEntryData::RunYielded { .. } + | SessionSnapshotEntryData::RunResumed { .. } + | SessionSnapshotEntryData::RunCancelled + | SessionSnapshotEntryData::RunError { .. } + ) + }) + .map(|entry| entry.entry_id.clone()) + .collect() + } + + #[test] + fn saved_run_transition_ids_match_snapshot_retained_and_paged_history() { + let worker_name = "worker-run-identities"; + let session_id = crate::new_session_id(); + let segment_id = crate::new_segment_id(); + let initial = vec![ + segment_start(session_id, Vec::new(), None, None), + LogEntry::AnnotatedUserInput { + ts: 2, + segments: vec![Segment::text("compact")], + history: vec![user_message(1)], + extensions: Vec::new(), + }, + ]; + let (_root, aggregate_root) = persist_history_fixture( + worker_name, + session_id, + segment_id, + vec![(segment_id, initial)], + ); + let store = WorkerSessionStore::new(aggregate_root.join("session")).unwrap(); + let transitions = [ + LogEntry::RunYielded { + ts: 10, + entry_id: None, + reason: protocol::RunYieldReason::Compaction, + active_run_turn_count: 2, + }, + LogEntry::RunResumed { + ts: 10, + entry_id: None, + source: protocol::RunResumeSource::Compaction, + active_run_turn_count: 2, + }, + LogEntry::RunErrored { + ts: 10, + entry_id: None, + interrupted: false, + message: "mid-run compaction failed: unavailable".into(), + failure: Some(protocol::RunFailureKind::Compaction), + }, + LogEntry::RunErrored { + ts: 10, + entry_id: None, + interrupted: false, + message: "mid-run compaction failed: unavailable".into(), + failure: Some(protocol::RunFailureKind::Compaction), + }, + LogEntry::RunCancelled { + ts: 10, + entry_id: None, + }, + ]; + for entry in transitions { + crate::segment::append_entry(&store, session_id, segment_id, entry).unwrap(); + } + + let stored = store.read_all(session_id, segment_id).unwrap(); + let stored_ids = run_entry_ids( + &project_session_snapshot_for_segment(session_id, Some(segment_id), &stored).entries, + ); + assert_eq!(stored_ids.len(), 5); + assert_eq!(stored_ids.iter().collect::>().len(), 5); + assert!( + stored_ids + .iter() + .all(|entry_id| !entry_id.starts_with("l-run-")) + ); + + let retained = read_retained_session_snapshot( + &aggregate_root, + worker_name, + DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES, + ) + .unwrap(); + assert_eq!(run_entry_ids(&retained.snapshot.entries), stored_ids); + + let page = read_retained_session_history_page( + &aggregate_root, + worker_name, + None, + Some(1), + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!(page.turns.len(), 1); + assert_eq!(run_entry_ids(&page.turns[0].entries), stored_ids); + + let reconnect = read_retained_session_snapshot( + &aggregate_root, + worker_name, + DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES, + ) + .unwrap(); + assert_eq!(run_entry_ids(&reconnect.snapshot.entries), stored_ids); + } + + #[test] + fn run_transition_ids_survive_segment_lineage_and_history_page_boundaries() { + let worker_name = "worker-run-lineage-identities"; + let session_id = crate::new_session_id(); + let source_segment = crate::new_segment_id(); + let active_segment = crate::new_segment_id(); + let source = vec![ + segment_start(session_id, Vec::new(), None, None), + LogEntry::AnnotatedUserInput { + ts: 2, + segments: vec![Segment::text("source turn")], + history: vec![user_message(1)], + extensions: Vec::new(), + }, + ]; + let (_root, aggregate_root) = persist_history_fixture( + worker_name, + session_id, + source_segment, + vec![(source_segment, source)], + ); + let store = WorkerSessionStore::new(aggregate_root.join("session")).unwrap(); + crate::segment::append_entry( + &store, + session_id, + source_segment, + LogEntry::RunYielded { + ts: 3, + entry_id: None, + reason: protocol::RunYieldReason::Compaction, + active_run_turn_count: 1, + }, + ) + .unwrap(); + store + .append( + session_id, + source_segment, + &LogEntry::TurnEnd { + ts: 4, + turn_count: 1, + }, + ) + .unwrap(); + store + .create_segment( + session_id, + active_segment, + &[ + segment_start( + session_id, + vec![user_message(1)], + None, + Some(SegmentOrigin { + segment_id: source_segment, + at_turn_index: 1, + }), + ), + LogEntry::AnnotatedUserInput { + ts: 5, + segments: vec![Segment::text("active turn")], + history: vec![user_message(2)], + extensions: Vec::new(), + }, + ], + ) + .unwrap(); + for entry in [ + LogEntry::RunResumed { + ts: 6, + entry_id: None, + source: protocol::RunResumeSource::Compaction, + active_run_turn_count: 1, + }, + LogEntry::RunErrored { + ts: 7, + entry_id: None, + interrupted: false, + message: "mid-run compaction failed".into(), + failure: Some(protocol::RunFailureKind::Compaction), + }, + ] { + crate::segment::append_entry(&store, session_id, active_segment, entry).unwrap(); + } + WorkerAggregateStore::new(&aggregate_root, worker_name) + .unwrap() + .write(&crate::WorkerMetadata::new( + worker_name, + Some(crate::WorkerActiveSegmentRef::active_segment( + session_id, + active_segment, + )), + )) + .unwrap(); + + let active_ids = run_entry_ids( + &project_session_snapshot_for_segment( + session_id, + Some(active_segment), + &store.read_all(session_id, active_segment).unwrap(), + ) + .entries, + ); + let source_ids = run_entry_ids( + &project_session_snapshot_for_segment( + session_id, + Some(source_segment), + &store.read_all(session_id, source_segment).unwrap(), + ) + .entries, + ); + assert_eq!(active_ids.len(), 2); + assert_eq!(source_ids.len(), 1); + + let newest = read_retained_session_history_page( + &aggregate_root, + worker_name, + None, + Some(1), + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!(run_entry_ids(&newest.turns[0].entries), active_ids); + assert!(newest.has_more); + let older = read_retained_session_history_page( + &aggregate_root, + worker_name, + newest.next_cursor.as_deref(), + Some(1), + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!(run_entry_ids(&older.turns[0].entries), source_ids); + } + + #[test] + fn legacy_run_transition_ids_use_stable_distinct_record_positions() { + let worker_name = "worker-legacy-run-identities"; + let session_id = crate::new_session_id(); + let segment_id = crate::new_segment_id(); + let log = vec![ + segment_start(session_id, Vec::new(), None, None), + LogEntry::AnnotatedUserInput { + ts: 2, + segments: vec![Segment::text("legacy")], + history: vec![user_message(1)], + extensions: Vec::new(), + }, + LogEntry::RunErrored { + ts: 10, + entry_id: None, + interrupted: false, + message: "same legacy failure".into(), + failure: None, + }, + LogEntry::RunErrored { + ts: 10, + entry_id: None, + interrupted: false, + message: "same legacy failure".into(), + failure: None, + }, + ]; + let (_root, aggregate_root) = + persist_history_fixture(worker_name, session_id, segment_id, vec![(segment_id, log)]); + let log_path = aggregate_root + .join("session") + .join("segments") + .join(format!("{segment_id}.jsonl")); + let persisted_before = std::fs::read(&log_path).unwrap(); + assert!(!String::from_utf8_lossy(&persisted_before).contains("entry_id\":\"l-run-")); + + let retained = read_retained_session_snapshot( + &aggregate_root, + worker_name, + DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES, + ) + .unwrap(); + let snapshot_ids = run_entry_ids(&retained.snapshot.entries); + assert_eq!(snapshot_ids.len(), 2); + assert_ne!(snapshot_ids[0], snapshot_ids[1]); + assert!( + snapshot_ids + .iter() + .all(|entry_id| entry_id.starts_with("l-run-")) + ); + + let page = read_retained_session_history_page( + &aggregate_root, + worker_name, + None, + Some(1), + RetainedHistoryReadLimits::default(), + ) + .unwrap(); + assert_eq!(run_entry_ids(&page.turns[0].entries), snapshot_ids); + let reread = read_retained_session_snapshot( + &aggregate_root, + worker_name, + DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES, + ) + .unwrap(); + assert_eq!(run_entry_ids(&reread.snapshot.entries), snapshot_ids); + assert_eq!(std::fs::read(log_path).unwrap(), persisted_before); + } + #[test] fn run_transitions_project_with_timestamps_and_typed_causes() { let session_id = crate::new_session_id(); let log = vec![ LogEntry::RunYielded { ts: 10, + entry_id: None, reason: protocol::RunYieldReason::Compaction, active_run_turn_count: 2, }, LogEntry::RunResumed { ts: 20, + entry_id: None, source: protocol::RunResumeSource::Compaction, active_run_turn_count: 2, }, LogEntry::RunErrored { ts: 30, + entry_id: None, interrupted: false, message: "mid-run compaction failed: unavailable".into(), failure: Some(protocol::RunFailureKind::Compaction), @@ -2869,7 +3213,10 @@ mod tests { ts: 10, trigger: protocol::InvokeKind::UserSend, }, - LogEntry::RunCancelled { ts: 20 }, + LogEntry::RunCancelled { + ts: 20, + entry_id: None, + }, ], ); diff --git a/crates/session-store/src/segment.rs b/crates/session-store/src/segment.rs index 5275b2c6b..cbd1a12b4 100644 --- a/crates/session-store/src/segment.rs +++ b/crates/session-store/src/segment.rs @@ -347,6 +347,7 @@ pub fn save_run_yielded( segment_id, LogEntry::RunYielded { ts: segment_log::now_millis(), + entry_id: None, reason, active_run_turn_count, }, @@ -367,6 +368,7 @@ pub fn save_run_resumed( segment_id, LogEntry::RunResumed { ts: segment_log::now_millis(), + entry_id: None, source, active_run_turn_count, }, @@ -390,6 +392,7 @@ pub fn save_run_errored( segment_id, LogEntry::RunErrored { ts: segment_log::now_millis(), + entry_id: None, interrupted, message, failure: Some(RunFailureKind::Engine), @@ -568,7 +571,8 @@ pub fn append_entry( store: &impl Store, session_id: SessionId, segment_id: SegmentId, - entry: LogEntry, + mut entry: LogEntry, ) -> Result<(), StoreError> { + entry.ensure_session_entry_id(); store.append(session_id, segment_id, &entry) } diff --git a/crates/session-store/src/segment_log.rs b/crates/session-store/src/segment_log.rs index 736fa81ba..fb61009b0 100644 --- a/crates/session-store/src/segment_log.rs +++ b/crates/session-store/src/segment_log.rs @@ -140,6 +140,12 @@ pub enum LogEntry { /// preceding Invoke and a later RunResumed or terminal record. RunYielded { ts: u64, + /// Stable public identity allocated before the record is committed. + /// `None` is accepted only when reading logs written before this field + /// existed; read paths derive the same legacy identity from the durable + /// record position without rewriting the log. + #[serde(default, skip_serializing_if = "Option::is_none")] + entry_id: Option, reason: RunYieldReason, active_run_turn_count: usize, }, @@ -149,19 +155,27 @@ pub enum LogEntry { /// It deliberately does not assert that a subsequent model call occurred. RunResumed { ts: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + entry_id: Option, source: RunResumeSource, active_run_turn_count: usize, }, /// The logical Run ended because the user intentionally cancelled it. /// This is terminal, but is not an execution failure. - RunCancelled { ts: u64 }, + RunCancelled { + ts: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + entry_id: Option, + }, /// `run()` / `resume()` が `EngineError` で終了した。 /// `EngineError` は `Serialize` 不可なので `message` のみ lossy 保持する。 /// Audit-only metadata: replay は `interrupted` のみ反映する。 RunErrored { ts: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + entry_id: Option, interrupted: bool, message: String, /// Older records omit this field. New records distinguish failures in @@ -236,6 +250,48 @@ pub struct SegmentOrigin { pub at_turn_index: usize, } +impl LogEntry { + /// Allocate the durable public identity for a newly committed Run + /// transition. Calling this repeatedly is idempotent so the exact value + /// written to storage is also the value published to live subscribers. + pub fn ensure_session_entry_id(&mut self) { + let entry_id = match self { + Self::RunYielded { entry_id, .. } + | Self::RunResumed { entry_id, .. } + | Self::RunCancelled { entry_id, .. } + | Self::RunErrored { entry_id, .. } => entry_id, + _ => return, + }; + if entry_id.is_none() { + *entry_id = Some(crate::LoggedSessionHistoryEntryId::new()); + } + } + + /// Supply an in-memory compatibility identity for a legacy Run transition + /// that predates persisted entry IDs. The physical record position is + /// stable within the append-only Segment and distinguishes equal payloads. + /// This helper never writes the synthesized value back to the log. + pub(crate) fn ensure_legacy_session_entry_id( + &mut self, + session_id: crate::SessionId, + segment_id: crate::SegmentId, + record_start: u64, + ) { + let entry_id = match self { + Self::RunYielded { entry_id, .. } + | Self::RunResumed { entry_id, .. } + | Self::RunCancelled { entry_id, .. } + | Self::RunErrored { entry_id, .. } => entry_id, + _ => return, + }; + if entry_id.is_none() { + *entry_id = Some(crate::LoggedSessionHistoryEntryId(format!( + "l-run-{session_id}-{segment_id}-{record_start}" + ))); + } + } +} + /// State collected from log entries. #[derive(Debug, Clone)] pub struct RestoredState { @@ -878,6 +934,7 @@ mod tests { }, LogEntry::RunYielded { ts: 300, + entry_id: None, reason: RunYieldReason::Compaction, active_run_turn_count: 2, }, @@ -888,6 +945,7 @@ mod tests { }, LogEntry::RunResumed { ts: 500, + entry_id: None, source: RunResumeSource::Compaction, active_run_turn_count: 2, }, @@ -909,7 +967,10 @@ mod tests { ts: 200, turn_count: 1, }, - LogEntry::RunCancelled { ts: 300 }, + LogEntry::RunCancelled { + ts: 300, + entry_id: None, + }, ]); assert!(!state.last_run_interrupted); diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index 35218068c..a23f50bb4 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/crates/session-store/src/worker_session_store.rs @@ -61,6 +61,8 @@ pub(crate) struct RetainedLogRecord { pub(crate) struct RetainedSegmentReader { file: File, + session_id: SessionId, + segment_id: SegmentId, read_end: u64, buffer_start: u64, buffer: Vec, @@ -103,7 +105,12 @@ impl RetainedSegmentReader { message: "empty retained Segment log record".to_string(), }); } - let entry = serde_json::from_slice(&record)?; + let mut entry: LogEntry = serde_json::from_slice(&record)?; + entry.ensure_legacy_session_entry_id( + self.session_id, + self.segment_id, + record_start, + ); return Ok(( Some(RetainedLogRecord { start_offset: record_start, @@ -255,7 +262,10 @@ impl WorkerSessionStore { if byte_len > max_bytes { return Err(StoreError::ReadLimitExceeded); } - Ok((parse_jsonl(&bytes)?, byte_len)) + Ok(( + parse_session_log_jsonl(&bytes, session_id, segment_id)?, + byte_len, + )) } pub(crate) fn read_first_log_record_read_only_bounded( @@ -264,6 +274,7 @@ impl WorkerSessionStore { segment_id: SegmentId, max_bytes: u64, ) -> Result<(Option, u64, u64), StoreError> { + let record_start = 0; self.validate_retained_session(session_id)?; let file = open_retained_file(&self.log_path(segment_id)).map_err(|error| { if error.kind() == std::io::ErrorKind::NotFound { @@ -289,7 +300,8 @@ impl WorkerSessionStore { message: "empty retained Segment log record".to_string(), }); } - let entry = serde_json::from_slice(&record)?; + let mut entry: LogEntry = serde_json::from_slice(&record)?; + entry.ensure_legacy_session_entry_id(session_id, segment_id, record_start); Ok(( Some(RetainedLogRecord { start_offset: 0, @@ -308,6 +320,7 @@ impl WorkerSessionStore { start_offset: u64, max_bytes: u64, ) -> Result<(Option, u64), StoreError> { + let record_start = start_offset; self.validate_retained_session(session_id)?; let mut file = open_retained_file(&self.log_path(segment_id)).map_err(|error| { if error.kind() == std::io::ErrorKind::NotFound { @@ -334,7 +347,8 @@ impl WorkerSessionStore { message: "empty retained Segment log record".to_string(), }); } - let entry = serde_json::from_slice(&record)?; + let mut entry: LogEntry = serde_json::from_slice(&record)?; + entry.ensure_legacy_session_entry_id(session_id, segment_id, record_start); Ok(( Some(RetainedLogRecord { start_offset, @@ -353,6 +367,7 @@ impl WorkerSessionStore { end_offset: u64, max_bytes: u64, ) -> Result { + let record_start = start_offset; self.validate_retained_session(session_id)?; let record_len = end_offset @@ -387,7 +402,8 @@ impl WorkerSessionStore { }); } record.pop(); - let entry = serde_json::from_slice(&record)?; + let mut entry: LogEntry = serde_json::from_slice(&record)?; + entry.ensure_legacy_session_entry_id(session_id, segment_id, record_start); Ok(RetainedLogRecord { start_offset, end_offset, @@ -429,6 +445,8 @@ impl WorkerSessionStore { } Ok(RetainedSegmentReader { file, + session_id, + segment_id, read_end: before_offset, buffer_start: before_offset, buffer: Vec::new(), @@ -645,7 +663,7 @@ impl Store for WorkerSessionStore { if !path.exists() { return Err(StoreError::NotFound(segment_id)); } - parse_jsonl(&fs::read(path)?) + parse_session_log_jsonl(&fs::read(path)?, session_id, segment_id) } fn list_sessions(&self) -> Result, StoreError> { @@ -1154,6 +1172,38 @@ fn complete_jsonl_prefix(content: &[u8]) -> &[u8] { .unwrap_or(&[]) } +fn parse_session_log_jsonl( + content: &[u8], + session_id: SessionId, + segment_id: SegmentId, +) -> Result, StoreError> { + let complete = complete_jsonl_prefix(content); + let content = std::str::from_utf8(complete).map_err(|error| StoreError::Corrupt { + line: complete[..error.valid_up_to()] + .iter() + .filter(|byte| **byte == b'\n') + .count() + + 1, + message: error.to_string(), + })?; + let mut entries = Vec::new(); + let mut record_start = 0_u64; + for (index, record) in content.split_inclusive('\n').enumerate() { + let line = record.strip_suffix('\n').unwrap_or(record); + if !line.trim().is_empty() { + let mut entry: LogEntry = + serde_json::from_str(line).map_err(|error| StoreError::Corrupt { + line: index + 1, + message: error.to_string(), + })?; + entry.ensure_legacy_session_entry_id(session_id, segment_id, record_start); + entries.push(entry); + } + record_start = record_start.saturating_add(record.len() as u64); + } + Ok(entries) +} + fn parse_jsonl(content: &[u8]) -> Result, StoreError> { let complete = complete_jsonl_prefix(content); let content = std::str::from_utf8(complete).map_err(|error| StoreError::Corrupt { diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index c9c56e400..15aa8a62c 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -3800,6 +3800,7 @@ mod completion_flow_tests { let run_errored = session_store::LogEntry::RunErrored { ts: 3, + entry_id: None, interrupted: false, message: "provider unavailable".into(), failure: None, diff --git a/crates/worker/src/ipc/protocol_session.rs b/crates/worker/src/ipc/protocol_session.rs index 8b3af7dd8..cf83db223 100644 --- a/crates/worker/src/ipc/protocol_session.rs +++ b/crates/worker/src/ipc/protocol_session.rs @@ -117,11 +117,128 @@ pub async fn dispatch_worker_protocol_method( #[cfg(test)] mod tests { use super::*; + use session_store::Store; + + use crate::segment_log_sink::SegmentLogSink; + use crate::worker::{LogWriterHandle, SegmentState}; + + fn committed_entry(event: Event) -> protocol::SessionSnapshotEntry { + match event { + Event::SessionEntryCommitted { entry } => entry, + other => panic!("expected SessionEntryCommitted, got {other:?}"), + } + } + + #[test] + fn persisted_run_transitions_keep_identity_across_live_and_snapshot_projection() { + let temp = tempfile::tempdir().unwrap(); + let store = session_store::FsStore::new(temp.path()).unwrap(); + let session_id = session_store::new_session_id(); + let segment_id = session_store::new_segment_id(); + let start = LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id, + system_prompt: None, + config: agen::llm_client::RequestConfig::default(), + history: Vec::new(), + forked_from: None, + compacted_from: None, + }; + store + .create_segment(session_id, segment_id, std::slice::from_ref(&start)) + .unwrap(); + let sink = SegmentLogSink::with_initial(vec![start]); + let writer = LogWriterHandle { + store: store.clone(), + state: SegmentState::new(session_id, segment_id, 1), + sink: sink.clone(), + in_flight: None, + }; + let (_, mut receiver) = sink.subscribe_with_snapshot(); + let transitions = [ + LogEntry::RunYielded { + ts: 10, + entry_id: None, + reason: protocol::RunYieldReason::Compaction, + active_run_turn_count: 2, + }, + LogEntry::RunResumed { + ts: 10, + entry_id: None, + source: protocol::RunResumeSource::Compaction, + active_run_turn_count: 2, + }, + LogEntry::RunErrored { + ts: 10, + entry_id: None, + interrupted: false, + message: "same compaction failure".into(), + failure: Some(protocol::RunFailureKind::Compaction), + }, + LogEntry::RunErrored { + ts: 10, + entry_id: None, + interrupted: false, + message: "same compaction failure".into(), + failure: Some(protocol::RunFailureKind::Compaction), + }, + LogEntry::RunCancelled { + ts: 10, + entry_id: None, + }, + ]; + + let mut live_ids = Vec::new(); + for transition in transitions { + writer.append_entry(transition).unwrap(); + let committed = receiver.try_recv().expect("committed transition broadcast"); + live_ids.push( + committed_entry(live_log_entry_event(committed).expect("live projection")).entry_id, + ); + } + assert_eq!( + live_ids + .iter() + .collect::>() + .len(), + 5 + ); + + let (mirror, _) = sink.subscribe_with_snapshot(); + let snapshot = session_store::public_snapshot::project_current_session_snapshot(&mirror); + let snapshot_ids: Vec<_> = snapshot + .entries + .iter() + .filter(|entry| { + matches!( + entry.data, + protocol::SessionSnapshotEntryData::RunYielded { .. } + | protocol::SessionSnapshotEntryData::RunResumed { .. } + | protocol::SessionSnapshotEntryData::RunError { .. } + | protocol::SessionSnapshotEntryData::RunCancelled + ) + }) + .map(|entry| entry.entry_id.clone()) + .collect(); + assert_eq!(snapshot_ids, live_ids); + + let persisted = store.read_all(session_id, segment_id).unwrap(); + let persisted_snapshot = + session_store::public_snapshot::project_current_session_snapshot(&persisted); + let persisted_ids: Vec<_> = persisted_snapshot + .entries + .iter() + .filter(|entry| snapshot_ids.contains(&entry.entry_id)) + .map(|entry| entry.entry_id.clone()) + .collect(); + assert_eq!(persisted_ids, live_ids); + } #[test] fn durable_run_transition_maps_to_session_entry_committed() { let event = live_log_entry_event(LogEntry::RunYielded { ts: 42, + entry_id: Some(session_store::LoggedSessionHistoryEntryId::new()), reason: protocol::RunYieldReason::Compaction, active_run_turn_count: 3, }) diff --git a/crates/worker/src/segment_log_sink.rs b/crates/worker/src/segment_log_sink.rs index 75825e73f..55da19660 100644 --- a/crates/worker/src/segment_log_sink.rs +++ b/crates/worker/src/segment_log_sink.rs @@ -330,6 +330,7 @@ mod tests { // Durable logical-Run transitions share the committed-entry lane. sink.publish(LogEntry::RunResumed { ts: now_millis(), + entry_id: None, source: protocol::RunResumeSource::Compaction, active_run_turn_count: 1, }); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index b8ad139db..f96f9e7b6 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -1426,7 +1426,8 @@ where /// Append `entry` to the log: disk write → counter bump → in-memory /// mirror push → broadcast. The Store owns physical write ordering and /// partial-write recovery; publication happens only after it returns Ok. - pub fn append_entry(&self, entry: LogEntry) -> Result<(), StoreError> { + pub fn append_entry(&self, mut entry: LogEntry) -> Result<(), StoreError> { + entry.ensure_session_entry_id(); let _append_guard = self .state .append_lock @@ -3055,7 +3056,8 @@ impl Worker { /// Append `entry` to the session log AND publish it through the /// broadcast sink. The Store is the commit boundary: a failed write is /// never counted or published. - pub(crate) fn commit_entry(&self, entry: LogEntry) -> Result<(), StoreError> { + pub(crate) fn commit_entry(&self, mut entry: LogEntry) -> Result<(), StoreError> { + entry.ensure_session_entry_id(); let loc = self.segment_state.location(); self.store.append(loc.session_id, loc.segment_id, &entry)?; self.segment_state.increment_entries(); @@ -4707,6 +4709,7 @@ impl Worker { .expect("pause/yield resume must retain the logical-run turn budget"); self.commit_entry(LogEntry::RunResumed { ts: segment_log::now_millis(), + entry_id: None, source, active_run_turn_count, })?; @@ -5024,6 +5027,7 @@ impl Worker { // marked as already delivered to clients through the session-log sink. self.commit_entry(LogEntry::RunErrored { ts: segment_log::now_millis(), + entry_id: None, interrupted: false, message, failure: Some(RunFailureKind::Compaction), @@ -5435,6 +5439,7 @@ impl Worker { { self.commit_entry(LogEntry::RunErrored { ts: segment_log::now_millis(), + entry_id: None, interrupted: false, message: message.to_string(), failure: Some(RunFailureKind::Compaction), @@ -5458,6 +5463,7 @@ impl Worker { EngineRunExit::Yielded => { self.commit_entry(LogEntry::RunYielded { ts: segment_log::now_millis(), + entry_id: None, reason: RunYieldReason::Compaction, active_run_turn_count: active_run_turn_count .expect("yielded logical run must retain its turn budget"), @@ -5474,11 +5480,13 @@ impl Worker { EngineRunExit::Interrupted(RunInterruptionReason::Cancelled) => { self.commit_entry(LogEntry::RunCancelled { ts: segment_log::now_millis(), + entry_id: None, })?; } EngineRunExit::Interrupted(reason) => { self.commit_entry(LogEntry::RunErrored { ts: segment_log::now_millis(), + entry_id: None, interrupted, message: run_interruption_reason_message(reason), failure: Some(RunFailureKind::Engine), @@ -11442,6 +11450,7 @@ mod build_summary_prompt_tests { worker .commit_entry(LogEntry::RunErrored { ts: segment_log::now_millis(), + entry_id: None, interrupted: true, message: "provider failed".to_string(), failure: Some(RunFailureKind::Engine), @@ -11472,6 +11481,7 @@ mod build_summary_prompt_tests { worker .commit_entry(LogEntry::RunCancelled { ts: segment_log::now_millis(), + entry_id: None, }) .unwrap(); assert_eq!( @@ -11482,6 +11492,7 @@ mod build_summary_prompt_tests { worker .commit_entry(LogEntry::RunResumed { ts: segment_log::now_millis(), + entry_id: None, source: RunResumeSource::Compaction, active_run_turn_count: 1, })