From cea0225a6e2d9e23c59a7d4a453f99bc2e1eba97 Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 19 Oct 2026 13:52:43 +0900 Subject: [PATCH] fix: migrate legacy session history and retained archives --- crates/session-store/src/fs_store.rs | 19 + .../session-store/src/legacy_session_log.rs | 59 +++ .../src/legacy_session_log/tests.rs | 102 +++++ crates/session-store/src/public_index.rs | 6 +- crates/session-store/src/public_material.rs | 4 +- .../src/public_material/tests.rs | 139 ++++++ crates/session-store/src/public_snapshot.rs | 2 + crates/session-store/src/segment.rs | 1 + crates/session-store/src/segment_log.rs | 31 +- .../session-store/src/worker_session_store.rs | 88 +++- crates/session-store/tests/fs_store_test.rs | 6 +- crates/standalone/src/subjektiv_sessions.rs | 1 + crates/worker-runtime/src/fs_store.rs | 123 +++++- crates/worker-runtime/src/retention.rs | 405 +++++++++++++++++- crates/worker-runtime/src/runtime.rs | 12 +- crates/worker/src/session_history.rs | 40 +- crates/worker/src/worker.rs | 4 + crates/worker/tests/compact_events_test.rs | 19 +- .../workspace-schema-migrations.md | 8 + 19 files changed, 960 insertions(+), 109 deletions(-) create mode 100644 crates/session-store/src/legacy_session_log/tests.rs diff --git a/crates/session-store/src/fs_store.rs b/crates/session-store/src/fs_store.rs index cbde85674..b6cade6b5 100644 --- a/crates/session-store/src/fs_store.rs +++ b/crates/session-store/src/fs_store.rs @@ -68,6 +68,24 @@ impl FsStore { ) -> Result { let root = root.into(); fs::create_dir_all(&root)?; + for entry in fs::read_dir(&root)? { + let entry = entry?; + if entry.file_type()?.is_dir() + && let Some(session_id) = entry + .file_name() + .to_str() + .and_then(|name| name.parse().ok()) + { + let _guard = match retention.lock_live(session_id) { + Ok(guard) => guard, + // Interrupted erasure may leave bytes behind. Never read or + // upgrade them, and do not block unrelated live Sessions. + Err(StoreError::SessionErased(_)) => continue, + Err(error) => return Err(error), + }; + crate::worker_session_store::migrate_fs_session(&entry.path(), session_id)?; + } + } Ok(Self { root, append_lock: Arc::new(Mutex::new(())), @@ -434,6 +452,7 @@ impl Store for FsStore { let path = self.log_path(session_id, segment_id); if let Some(parent) = path.parent() { fs::create_dir_all(parent)?; + crate::worker_session_store::migrate_fs_session(parent, session_id)?; } let mut content = String::new(); for entry in entries { diff --git a/crates/session-store/src/legacy_session_log.rs b/crates/session-store/src/legacy_session_log.rs index bb9dd1dd5..5c465d0b9 100644 --- a/crates/session-store/src/legacy_session_log.rs +++ b/crates/session-store/src/legacy_session_log.rs @@ -8,6 +8,9 @@ use agen::llm_client::types::RequestConfig; use protocol::Segment; use serde::Deserialize; +#[cfg(test)] +mod tests; + use crate::{ LogEntry, LoggedHistoryEntry, LoggedItem, LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, LoggedSystemHistoryEntry, SegmentId, @@ -83,6 +86,7 @@ pub(crate) fn decode_entry( LegacySessionLogEntryV2::History(entry) => Entry::History(entry), LegacySessionLogEntryV2::Current(entry) => Entry::Current(entry), }, + 3 => Entry::Current(serde_json::from_str(line)?), _ => unreachable!("legacy decoder called for unsupported schema {schema_version}"), }; Ok(match entry { @@ -93,6 +97,61 @@ pub(crate) fn decode_entry( }) } +/// Schema 3 used DerivedSummary for Host compaction seeds. Only the trusted +/// creation boundary may identify one; arbitrary text and later seed items do not. +/// Include already-upgraded entries so an interrupted multi-file migration resumes. +pub(crate) fn compaction_summary_ids<'a>( + entries: impl Iterator, +) -> std::collections::HashSet { + entries + .filter_map(|entry| { + let LogEntry::AnnotatedSegmentStart { + compacted_from: Some(_), + forked_from: None, + history, + .. + } = entry + else { + return None; + }; + let first = history.first()?; + (matches!( + first.metadata.origin, + LoggedSessionHistoryOrigin::DerivedSummary + | LoggedSessionHistoryOrigin::CompactionSummary + ) && matches!( + first.item, + LoggedItem::Message { + role: crate::LoggedRole::System, + .. + } + ) && agen::Item::from(first.item.clone()) + .as_text() + .is_some_and(|text| text.starts_with("[Compacted context summary]\n\n"))) + .then(|| first.metadata.entry_id.clone()) + }) + .collect() +} + +pub(crate) fn migrate_compaction_origins( + entries: &mut [LogEntry], + summaries: &std::collections::HashSet, +) { + // Only inherited seed copies can share these identities. Appended input is + // not a construction boundary and must never be reclassified from its text. + for entry in entries { + if let LogEntry::AnnotatedSegmentStart { history, .. } = entry { + for entry in history { + if summaries.contains(&entry.metadata.entry_id) + && entry.metadata.origin == LoggedSessionHistoryOrigin::DerivedSummary + { + entry.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; + } + } + } + } +} + enum Entry { History(LegacyHistoryLogEntry), Current(LogEntry), diff --git a/crates/session-store/src/legacy_session_log/tests.rs b/crates/session-store/src/legacy_session_log/tests.rs new file mode 100644 index 000000000..3c72462c5 --- /dev/null +++ b/crates/session-store/src/legacy_session_log/tests.rs @@ -0,0 +1,102 @@ +use super::*; + +fn seed( + origin: LoggedSessionHistoryOrigin, + compacted: bool, + role: crate::LoggedRole, + text: &str, +) -> LogEntry { + LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id: SessionId::now_v7(), + system_prompt: None, + config: Default::default(), + forked_from: None, + compacted_from: compacted.then(|| SegmentOrigin { + segment_id: SegmentId::now_v7(), + at_turn_index: 0, + }), + history: vec![LoggedHistoryEntry { + item: LoggedItem::Message { + role, + content: vec![crate::LoggedContentPart::Text { text: text.into() }], + }, + metadata: LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId("summary".into()), + origin, + derivation: Some(crate::LoggedHistoryDerivation { + sources: vec![LoggedSessionHistoryEntryId("source".into())], + }), + }, + }], + } +} + +#[test] +fn only_trusted_compaction_seed_is_migrated_and_retries_update_inherited_copies() { + let text = "[Compacted context summary]\n\nsummary"; + for (compacted, role, text) in [ + (false, crate::LoggedRole::System, text), + (true, crate::LoggedRole::User, text), + ( + true, + crate::LoggedRole::System, + "[Session TaskStore snapshot] tasks", + ), + ] { + let entry = seed( + LoggedSessionHistoryOrigin::DerivedSummary, + compacted, + role, + text, + ); + assert!(compaction_summary_ids([&entry].into_iter()).is_empty()); + } + for origin in [ + LoggedSessionHistoryOrigin::DerivedSummary, + LoggedSessionHistoryOrigin::CompactionSummary, + ] { + let creation = seed(origin, true, crate::LoggedRole::System, text); + let inherited = seed( + LoggedSessionHistoryOrigin::DerivedSummary, + false, + crate::LoggedRole::System, + text, + ); + let LogEntry::AnnotatedSegmentStart { history, .. } = &inherited else { + panic!() + }; + let mut pasted = history[0].clone(); + pasted.metadata.entry_id = LoggedSessionHistoryEntryId("pasted".into()); + let appended = LogEntry::AnnotatedUserInput { + ts: 2, + history: vec![pasted.clone()], + segments: vec![], + extensions: vec![], + }; + let mut entries = vec![creation, inherited, appended]; + let ids = compaction_summary_ids(entries.iter()); + migrate_compaction_origins(&mut entries, &ids); + for entry in &entries[..2] { + let LogEntry::AnnotatedSegmentStart { history, .. } = entry else { + panic!() + }; + assert_eq!( + history[0].metadata.origin, + LoggedSessionHistoryOrigin::CompactionSummary + ); + assert_eq!(history[0].metadata.entry_id.0, "summary"); + assert_eq!( + history[0].metadata.derivation.as_ref().unwrap().sources[0].0, + "source" + ); + } + let LogEntry::AnnotatedUserInput { history, .. } = &entries[2] else { + panic!() + }; + assert_eq!(history, &[pasted]); + let once = serde_json::to_vec(&entries).unwrap(); + migrate_compaction_origins(&mut entries, &ids); + assert_eq!(serde_json::to_vec(&entries).unwrap(), once); + } +} diff --git a/crates/session-store/src/public_index.rs b/crates/session-store/src/public_index.rs index 1a8392c7e..24932e397 100644 --- a/crates/session-store/src/public_index.rs +++ b/crates/session-store/src/public_index.rs @@ -1602,6 +1602,7 @@ mod tests { segment_id(20), "completed", LogEntry::RunCompleted { + entry_id: None, ts: 4, interrupted: false, result: EngineResult::Finished, @@ -1640,7 +1641,10 @@ mod tests { ( segment_id(24), "abandoned", - LogEntry::PausedTurnAbandoned { ts: 4 }, + LogEntry::PausedTurnAbandoned { + entry_id: None, + ts: 4, + }, ), ]; for (segment_id, text, outcome) in outcomes { diff --git a/crates/session-store/src/public_material.rs b/crates/session-store/src/public_material.rs index bd4a6576c..6487bdade 100644 --- a/crates/session-store/src/public_material.rs +++ b/crates/session-store/src/public_material.rs @@ -814,7 +814,9 @@ fn boundary_reference( offset: u64, ) -> String { let id = match record { - LogEntry::RunYielded { entry_id, .. } + LogEntry::RunCompleted { entry_id, .. } + | LogEntry::PausedTurnAbandoned { entry_id, .. } + | LogEntry::RunYielded { entry_id, .. } | LogEntry::RunResumed { entry_id, .. } | LogEntry::RunCancelled { entry_id, .. } | LogEntry::RunErrored { entry_id, .. } => entry_id.as_ref(), diff --git a/crates/session-store/src/public_material/tests.rs b/crates/session-store/src/public_material/tests.rs index 234db56b1..b77642840 100644 --- a/crates/session-store/src/public_material/tests.rs +++ b/crates/session-store/src/public_material/tests.rs @@ -244,6 +244,7 @@ fn late_run_boundary_is_new_material_without_changing_previously_supplied_bodies ), ( LogEntry::RunCompleted { + entry_id: None, ts: 5, interrupted: false, result: agen::EngineResult::Finished, @@ -256,6 +257,7 @@ fn late_run_boundary_is_new_material_without_changing_previously_supplied_bodies ), ( LogEntry::RunCompleted { + entry_id: None, ts: 5, interrupted: true, result: agen::EngineResult::LimitReached, @@ -662,6 +664,143 @@ fn fork_clips_parent_suffix_and_maps_inherited_copy_to_original_segment() { ); } +#[test] +fn schema_v3_compaction_migrates_persistently_and_strict_reader_keeps_original_ids() { + for standalone in [false, true] { + let temp = tempfile::tempdir().unwrap(); + let session = crate::new_session_id(); + let root = crate::new_segment_id(); + let compact = crate::new_segment_id(); + let mut summary = message( + "summary", + LoggedRole::System, + "[Compacted context summary]\n\nold summary", + ); + summary.metadata.origin = LoggedSessionHistoryOrigin::DerivedSummary; + summary.metadata.derivation = Some(crate::LoggedHistoryDerivation { + sources: vec![LoggedSessionHistoryEntryId("u".into())], + }); + let mut start = seed(session, vec![summary.clone()]); + if let LogEntry::AnnotatedSegmentStart { compacted_from, .. } = &mut start { + *compacted_from = Some(SegmentOrigin { + segment_id: root, + at_turn_index: 0, + }); + } + let records = [ + start, + output("new", "new"), + cancelled(), + LogEntry::RunCompleted { + ts: 6, + entry_id: None, + interrupted: false, + result: agen::EngineResult::Finished, + active_run_turn_count: None, + }, + LogEntry::PausedTurnAbandoned { + ts: 7, + entry_id: None, + }, + ]; + let session_root = if standalone { + temp.path().join(session.to_string()) + } else { + temp.path().to_path_buf() + }; + let paths = if standalone { + session_root.clone() + } else { + session_root.join("segments") + }; + std::fs::create_dir_all(&paths).unwrap(); + for (id, entries) in [ + ( + root, + vec![seed( + session, + vec![message("u", LoggedRole::User, "request")], + )], + ), + (compact, records.to_vec()), + ] { + let bytes: String = entries + .iter() + .map(|entry| format!("{}\n", serde_json::to_string(entry).unwrap())) + .collect(); + std::fs::write(paths.join(format!("{id}.jsonl")), bytes).unwrap(); + } + if !standalone { + std::fs::write( + session_root.join("session.json"), + serde_json::to_vec( + &serde_json::json!({"schema_version": 3, "session_id": session}), + ) + .unwrap(), + ) + .unwrap(); + } + let open = || { + if standalone { + SessionPublicMaterialReader::open_fs( + temp.path(), + session, + compact, + Default::default(), + ) + } else { + SessionPublicMaterialReader::open_worker(&session_root, compact, Default::default()) + } + }; + assert!( + open().is_err(), + "strict reader must not reinterpret old provenance" + ); + let migrate = || { + if standalone { + FsStore::new(temp.path()).unwrap(); + } else { + WorkerSessionStore::new(&session_root).unwrap(); + } + }; + migrate(); + let reader = open().unwrap(); + assert_eq!(&ids(&reader)[..2], ["Eu", "Enew"]); + let path = paths.join(format!("{compact}.jsonl")); + let bytes = std::fs::read(&path).unwrap(); + let entries: Vec = String::from_utf8(bytes.clone()) + .unwrap() + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + let LogEntry::AnnotatedSegmentStart { history, .. } = &entries[0] else { + panic!() + }; + summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; + assert_eq!(history, &[summary]); + // Run-transition ids synthesized from byte offsets must be frozen before + // the changed summary record shifts every subsequent record offset. + let mut old_offset = 0; + for (index, original) in records.iter().enumerate() { + if index >= 2 { + let mut expected = original.clone(); + expected.ensure_legacy_session_entry_id(session, compact, old_offset); + assert_eq!( + serde_json::to_value(&entries[index]).unwrap(), + serde_json::to_value(expected).unwrap() + ); + assert_eq!( + ids(&reader)[index], + format!("El-run-{session}-{compact}-{old_offset}") + ); + } + old_offset += serde_json::to_vec(original).unwrap().len() as u64 + 1; + } + migrate(); + assert_eq!(std::fs::read(&path).unwrap(), bytes); + } +} + #[test] fn midturn_compaction_uses_exact_source_ids_and_preserves_discarded_originals() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index 5705b515e..824d7dffd 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -1705,6 +1705,7 @@ pub(crate) fn project_session_snapshot_for_segment( interrupted, result, active_run_turn_count: persisted_count, + .. } => { if *interrupted && matches!(result, agen::EngineResult::Yielded) { let count = (*persisted_count) @@ -2038,6 +2039,7 @@ mod tests { fn complete(ts: u64) -> LogEntry { LogEntry::RunCompleted { + entry_id: None, ts, interrupted: false, result: agen::EngineResult::Finished, diff --git a/crates/session-store/src/segment.rs b/crates/session-store/src/segment.rs index 3078c298f..368b4e5b2 100644 --- a/crates/session-store/src/segment.rs +++ b/crates/session-store/src/segment.rs @@ -329,6 +329,7 @@ pub fn save_run_completed( session_id, segment_id, LogEntry::RunCompleted { + entry_id: None, ts: segment_log::now_millis(), interrupted, result, diff --git a/crates/session-store/src/segment_log.rs b/crates/session-store/src/segment_log.rs index 543c3752b..97f347dad 100644 --- a/crates/session-store/src/segment_log.rs +++ b/crates/session-store/src/segment_log.rs @@ -122,6 +122,8 @@ pub enum LogEntry { /// Replay restores both interruption state and any resumable logical-run /// turn budget. RunCompleted { + #[serde(default, skip_serializing_if = "Option::is_none")] + entry_id: Option, ts: u64, interrupted: bool, result: EngineResult, @@ -194,7 +196,11 @@ pub enum LogEntry { /// A paused interrupted turn was explicitly abandoned without calling /// `run()` or `resume()` again. Replay clears the interrupted marker so /// the restored Worker is idle and future user input starts a normal new turn. - PausedTurnAbandoned { ts: u64 }, + PausedTurnAbandoned { + ts: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + entry_id: Option, + }, /// `RequestConfig` changed. ConfigChanged { ts: u64, config: RequestConfig }, @@ -253,7 +259,9 @@ impl LogEntry { /// 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::RunCompleted { entry_id, .. } + | Self::PausedTurnAbandoned { entry_id, .. } + | Self::RunYielded { entry_id, .. } | Self::RunResumed { entry_id, .. } | Self::RunCancelled { entry_id, .. } | Self::RunErrored { entry_id, .. } => entry_id, @@ -275,7 +283,9 @@ impl LogEntry { record_start: u64, ) { let entry_id = match self { - Self::RunYielded { entry_id, .. } + Self::RunCompleted { entry_id, .. } + | Self::PausedTurnAbandoned { entry_id, .. } + | Self::RunYielded { entry_id, .. } | Self::RunResumed { entry_id, .. } | Self::RunCancelled { entry_id, .. } | Self::RunErrored { entry_id, .. } => entry_id, @@ -586,6 +596,7 @@ mod tests { turn_count: 1, }, LogEntry::RunCompleted { + entry_id: None, ts: 3200, interrupted: false, result: EngineResult::Finished, @@ -893,12 +904,16 @@ mod tests { compacted_from: None, }, LogEntry::RunCompleted { + entry_id: None, ts: 100, interrupted: true, result: EngineResult::Paused, active_run_turn_count: Some(1), }, - LogEntry::PausedTurnAbandoned { ts: 200 }, + LogEntry::PausedTurnAbandoned { + entry_id: None, + ts: 200, + }, ]); assert!(!state.last_run_interrupted); assert_eq!(state.active_run_turn_count, None); @@ -1056,6 +1071,7 @@ mod tests { turn_count: 2, }, LogEntry::RunCompleted { + entry_id: None, ts: 300, interrupted: true, result: EngineResult::LimitReached, @@ -1069,13 +1085,16 @@ mod tests { #[test] fn paused_turn_abandoned_entry_round_trip_via_json() { - let entry = LogEntry::PausedTurnAbandoned { ts: 12345 }; + let entry = LogEntry::PausedTurnAbandoned { + entry_id: None, + ts: 12345, + }; let json = serde_json::to_string(&entry).unwrap(); let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); assert_eq!(parsed["kind"], "paused_turn_abandoned"); let decoded: LogEntry = serde_json::from_str(&json).unwrap(); match decoded { - LogEntry::PausedTurnAbandoned { ts } => assert_eq!(ts, 12345), + LogEntry::PausedTurnAbandoned { ts, .. } => assert_eq!(ts, 12345), other => panic!("expected PausedTurnAbandoned, got {other:?}"), } } diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index 354e75644..b19180b3f 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/crates/session-store/src/worker_session_store.rs @@ -36,7 +36,8 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::time::SystemTime; -const SESSION_SCHEMA_VERSION: u32 = 3; +const SESSION_SCHEMA_VERSION: u32 = 4; +const ANNOTATED_SESSION_SCHEMA_VERSION: u32 = 3; const PREVIOUS_SESSION_SCHEMA_VERSION: u32 = 2; const LEGACY_SESSION_SCHEMA_VERSION: u32 = 1; const SESSION_FILE: &str = "session.json"; @@ -212,9 +213,11 @@ impl WorkerSessionStore { SESSION_SCHEMA_VERSION => { validate_canonical_segment_logs(&root)?; } - PREVIOUS_SESSION_SCHEMA_VERSION | LEGACY_SESSION_SCHEMA_VERSION => { - migrate_segment_logs_to_v3( - &root, + LEGACY_SESSION_SCHEMA_VERSION + | PREVIOUS_SESSION_SCHEMA_VERSION + | ANNOTATED_SESSION_SCHEMA_VERSION => { + migrate_segment_logs_to_v4( + segment_log_paths(&root)?, manifest.session_id, manifest.schema_version, )?; @@ -1107,7 +1110,12 @@ fn open_retained_file(path: &Path) -> std::io::Result { } fn segment_log_paths(root: &Path) -> Result, StoreError> { - let segments = root.join(SEGMENTS_DIR); + segment_log_paths_in_directory(&root.join(SEGMENTS_DIR)) +} + +fn segment_log_paths_in_directory( + segments: &Path, +) -> Result, StoreError> { if !segments.exists() { return Ok(Vec::new()); } @@ -1150,21 +1158,22 @@ fn segment_log_paths(root: &Path) -> Result, StoreErro Ok(paths) } -fn migrate_segment_logs_to_v3( - root: &Path, +fn migrate_segment_logs_to_v4( + paths: Vec<(SegmentId, PathBuf)>, session_id: SessionId, source_schema_version: u32, ) -> Result<(), StoreError> { struct MigrationPlan { path: PathBuf, source: Vec, + entries: Vec, output: Vec, } // Phase 1 is strictly read-only. Every segment must parse and canonicalize // successfully before the first authoritative byte is replaced. let mut plans = Vec::new(); - for (segment_id, path) in segment_log_paths(root)? { + for (segment_id, path) in paths { let source = fs::read(&path)?; let canonical = parse_legacy_jsonl(source_schema_version, session_id, segment_id, &source) .map_err(|error| StoreError::Corrupt { @@ -1174,17 +1183,25 @@ fn migrate_segment_logs_to_v3( path.display() ), })?; - let mut output = Vec::new(); - for entry in canonical { - serde_json::to_writer(&mut output, &entry)?; - output.push(b'\n'); - } plans.push(MigrationPlan { path, source, - output, + entries: canonical, + output: Vec::new(), }); } + // Resolve trusted creation boundaries first, then update inherited copies by + // logical identity too (fork/rewind seeds retain the same entry id). + let summaries = crate::legacy_session_log::compaction_summary_ids( + plans.iter().flat_map(|plan| plan.entries.iter()), + ); + for plan in &mut plans { + crate::legacy_session_log::migrate_compaction_origins(&mut plan.entries, &summaries); + for entry in std::mem::take(&mut plan.entries) { + serde_json::to_writer(&mut plan.output, &entry)?; + plan.output.push(b'\n'); + } + } // Fence the complete preflight snapshot before starting phase 2. Session // open is the exclusive restore boundary; this additionally fails closed @@ -1207,6 +1224,30 @@ fn migrate_segment_logs_to_v3( Ok(()) } +/// Upgrade the standalone per-Session layout using the same persisted format +/// migration as Worker aggregates. Called only at writable Host store startup. +pub(crate) fn migrate_fs_session(root: &Path, session_id: SessionId) -> Result<(), StoreError> { + let marker = root.join("session-schema.json"); + let version: u32 = match fs::read(&marker) { + Ok(bytes) => serde_json::from_slice(&bytes)?, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + ANNOTATED_SESSION_SCHEMA_VERSION + } + Err(error) => return Err(error.into()), + }; + match version { + SESSION_SCHEMA_VERSION => Ok(()), + ANNOTATED_SESSION_SCHEMA_VERSION => { + migrate_segment_logs_to_v4(segment_log_paths_in_directory(root)?, session_id, version)?; + atomic_write_json(&marker, &SESSION_SCHEMA_VERSION) + } + _ => Err(StoreError::Corrupt { + line: 0, + message: format!("unsupported standalone Session schema {version}"), + }), + } +} + fn validate_canonical_segment_logs(root: &Path) -> Result<(), StoreError> { for (_, path) in segment_log_paths(root)? { let _: Vec = parse_jsonl(&fs::read(&path)?)?; @@ -1223,17 +1264,26 @@ fn parse_legacy_jsonl( let text = std::str::from_utf8(bytes).map_err(|error| { serde_json::Error::io(std::io::Error::new(std::io::ErrorKind::InvalidData, error)) })?; - text.lines() + let mut offset = 0_u64; + text.split_inclusive('\n') .enumerate() - .filter(|(_, line)| !line.trim().is_empty()) - .map(|(line_index, line)| { - crate::legacy_session_log::decode_entry( + .filter_map(|(line_index, line)| { + let start = offset; + offset += line.len() as u64; + (!line.trim().is_empty()).then_some((line_index, line, start)) + }) + .map(|(line_index, line, start)| { + let mut entry = crate::legacy_session_log::decode_entry( schema_version, line, session_id, segment_id, line_index, - ) + )?; + // Byte offsets change during migration. Persist identities previously + // synthesized from the OLD offsets before rewriting any record. + entry.ensure_legacy_session_entry_id(session_id, segment_id, start); + Ok(entry) }) .collect() } diff --git a/crates/session-store/tests/fs_store_test.rs b/crates/session-store/tests/fs_store_test.rs index db8ec035c..1ecb81d01 100644 --- a/crates/session-store/tests/fs_store_test.rs +++ b/crates/session-store/tests/fs_store_test.rs @@ -62,6 +62,7 @@ fn round_trip_write_and_read() { turn_count: 1, }, LogEntry::RunCompleted { + entry_id: None, ts: 3200, interrupted: false, result: EngineResult::Finished, @@ -157,9 +158,10 @@ fn create_segment_replacement_keeps_open_old_log_and_commits_complete_new_log() assert_eq!(reopened.read_entry_count(sid, segid).unwrap(), new.len()); assert_eq!( std::fs::read_dir(path.parent().unwrap()).unwrap().count(), - 1, - "successful replacement must not leave temporary files" + 2, + "successful replacement leaves only the segment and Session schema marker" ); + assert!(path.parent().unwrap().join("session-schema.json").is_file()); } #[test] diff --git a/crates/standalone/src/subjektiv_sessions.rs b/crates/standalone/src/subjektiv_sessions.rs index 92b0b5d87..36a75df06 100644 --- a/crates/standalone/src/subjektiv_sessions.rs +++ b/crates/standalone/src/subjektiv_sessions.rs @@ -1693,6 +1693,7 @@ allow = [] // A valid JSON record without its final newline is not committed. log.write_all( &serde_json::to_vec(&LogEntry::RunCompleted { + entry_id: None, ts: 4, interrupted: false, result: agen::EngineResult::Finished, diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 0b06a9cc4..b46888e91 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -39,7 +39,8 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicU8, AtomicU64, Ordering}; -const SCHEMA_VERSION: u32 = 10; +const SCHEMA_VERSION: u32 = 11; +const EXECUTION_SCHEMA_VERSION: u32 = 10; // Frozen source schemas: do not move their meaning when adding a migration. const AGGREGATE_SCHEMA_VERSION: u32 = 9; const PREVIOUS_SCHEMA_VERSION: u32 = 8; @@ -640,8 +641,8 @@ fn plan_runtime_store_migration( format!("Runtime store schema version {schema_version} is out of range"), ) })?; - let staging = migration_sibling(root, "schema-v10-staging")?; - let backup = migration_sibling(root, "pre-schema-v10-backup")?; + let staging = migration_sibling(root, "schema-v11-staging")?; + let backup = migration_sibling(root, "pre-schema-v11-backup")?; if staging.exists() || backup.exists() { return Err(runtime_store_corrupt( root, @@ -674,11 +675,12 @@ fn plan_runtime_store_migration( | LEGACY_SCHEMA_VERSION | PREVIOUS_SCHEMA_VERSION | AGGREGATE_SCHEMA_VERSION + | EXECUTION_SCHEMA_VERSION ) { return Err(runtime_store_corrupt( &runtime_path, format!( - "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, {AGGREGATE_SCHEMA_VERSION}, or {SCHEMA_VERSION}" + "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, {AGGREGATE_SCHEMA_VERSION}, {EXECUTION_SCHEMA_VERSION}, or {SCHEMA_VERSION}" ), )); } @@ -1008,6 +1010,23 @@ fn migrate_worker_record( mapping: Option<&LegacyWorkerIdentityMapping>, identity_path: &Path, ) -> Result { + if source_schema_version == EXECUTION_SCHEMA_VERSION { + if identity_document + .get("schema_version") + .and_then(serde_json::Value::as_u64) + != Some(u64::from(EXECUTION_SCHEMA_VERSION)) + { + return Err(runtime_store_corrupt( + identity_path, + "Worker schema does not match Runtime migration source".into(), + )); + } + identity_document["schema_version"] = serde_json::Value::from(SCHEMA_VERSION); + let aggregate: WorkerAggregateRecord = serde_json::from_value(identity_document) + .map_err(|error| runtime_store_corrupt(identity_path, error.to_string()))?; + aggregate.clone().validate(identity_path)?; + return Ok(aggregate); + } if source_schema_version == AGGREGATE_SCHEMA_VERSION { migrate_v9_restore_guard(&mut identity_document, identity_path)?; remove_legacy_job_input_counter(&mut identity_document); @@ -1499,14 +1518,19 @@ fn remove_runtime_migration_directory(path: &Path) -> Result<(), RuntimeError> { } fn recover_runtime_store_migration(root: &Path) -> Result<(), RuntimeError> { - for suffix in ["schema-v9-staging", "pre-schema-v9-backup"] { + for suffix in [ + "schema-v9-staging", + "pre-schema-v9-backup", + "schema-v10-staging", + "pre-schema-v10-backup", + ] { let earlier = migration_sibling(root, suffix)?; if earlier.exists() { - return Err(runtime_store_corrupt(root, "unfinished schema-v9 migration requires explicit recovery before schema-v10 startup".into())); + return Err(runtime_store_corrupt(root, "unfinished earlier Runtime migration requires explicit recovery before schema-v11 startup".into())); } } - let staging = migration_sibling(root, "schema-v10-staging")?; - let backup = migration_sibling(root, "pre-schema-v10-backup")?; + let staging = migration_sibling(root, "schema-v11-staging")?; + let backup = migration_sibling(root, "pre-schema-v11-backup")?; let root_exists = migration_directory_exists(root)?; let staging_exists = migration_directory_exists(&staging)?; let backup_exists = migration_directory_exists(&backup)?; @@ -1708,8 +1732,8 @@ fn migrate_runtime_store( if !plan.migration_required { return Ok(plan); } - let staging = migration_sibling(root, "schema-v10-staging")?; - let backup = migration_sibling(root, "pre-schema-v10-backup")?; + let staging = migration_sibling(root, "schema-v11-staging")?; + let backup = migration_sibling(root, "pre-schema-v11-backup")?; if staging.exists() || backup.exists() { return Err(runtime_store_corrupt( root, @@ -1903,6 +1927,7 @@ fn migrate_runtime_store_in_place( ), )); } + crate::retention::migrate_staged_sessions(root)?; let (document, _) = migrate_runtime_document( document, plan.current_schema_version, @@ -2683,6 +2708,70 @@ mod tests { document["schema_version"].as_u64().unwrap() } + #[test] + fn schema_v10_startup_migrates_stopped_sessions_atomically() { + for corrupt in [false, true] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + write_empty_schema_v8_store(&root); + let mut runtime: serde_json::Value = + read_json(&root.join(RUNTIME_FILE), "test").unwrap(); + runtime["schema_version"] = serde_json::json!(10); + atomic_write_json(&root.join(RUNTIME_FILE), &runtime, "test").unwrap(); + let worker_id = WorkerId::now_v7(); + let aggregate = root.join(WORKERS_DIR).join(worker_id.to_string()); + fs::create_dir_all(&aggregate).unwrap(); + let record = migrate_worker_record( + schema_v7_worker_document(worker_id), + 7, + None, + &aggregate.join(WORKER_FILE), + ) + .unwrap(); + let mut record = serde_json::to_value(record).unwrap(); + record["schema_version"] = serde_json::json!(10); + atomic_write_json(&aggregate.join(WORKER_FILE), &record, "test").unwrap(); + let session = aggregate.join("session"); + fs::create_dir_all(session.join("segments")).unwrap(); + atomic_write_json(&session.join("session.json"), &serde_json::json!({"schema_version":3,"session_id":session_store::new_session_id()}), "test").unwrap(); + if corrupt { + fs::write( + session + .join("segments") + .join(format!("{}.jsonl", session_store::new_segment_id())), + b"invalid\n", + ) + .unwrap(); + } + let before = fs::read(session.join("session.json")).unwrap(); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-test"); + assert!( + FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + assert_eq!(fs::read(session.join("session.json")).unwrap(), before); + let result = FsRuntimeStore::open_or_create(root.clone(), "runtime-test"); + if corrupt { + assert!(result.is_err()); + assert_eq!(stored_schema_version(&root), 10); + assert_eq!(fs::read(session.join("session.json")).unwrap(), before); + let record: serde_json::Value = + read_json(&aggregate.join(WORKER_FILE), "test").unwrap(); + assert_eq!(record["schema_version"], 10); + } else { + drop(result.unwrap()); + assert_eq!(stored_schema_version(&root), 11); + let manifest: serde_json::Value = + read_json(&session.join("session.json"), "test").unwrap(); + assert_eq!(manifest["schema_version"], 4); + let once = fs::read(session.join("session.json")).unwrap(); + drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap()); + assert_eq!(fs::read(session.join("session.json")).unwrap(), once); + } + } + } + #[test] fn auth_initialized_directory_without_snapshot_is_a_fresh_store() { let parent = tempfile::tempdir().unwrap(); @@ -3062,7 +3151,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); write_empty_schema_v8_store(&root); - let staging = migration_sibling(&root, "schema-v10-staging").unwrap(); + let staging = migration_sibling(&root, "schema-v11-staging").unwrap(); fs::create_dir_all(&staging).unwrap(); fs::write(staging.join("partial"), b"partial").unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -3077,7 +3166,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); write_empty_schema_v8_store(&root); - let backup = migration_sibling(&root, "pre-schema-v10-backup").unwrap(); + let backup = migration_sibling(&root, "pre-schema-v11-backup").unwrap(); fs::rename(&root, &backup).unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -3091,10 +3180,10 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); write_empty_schema_v8_store(&root); - let staging = migration_sibling(&root, "schema-v10-staging").unwrap(); + let staging = migration_sibling(&root, "schema-v11-staging").unwrap(); copy_runtime_tree(&root, &root, &staging).unwrap(); migrate_runtime_store_in_place(&staging, "test-runtime").unwrap(); - let backup = migration_sibling(&root, "pre-schema-v10-backup").unwrap(); + let backup = migration_sibling(&root, "pre-schema-v11-backup").unwrap(); fs::rename(&root, &backup).unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -3109,10 +3198,10 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); write_empty_schema_v8_store(&root); - let staging = migration_sibling(&root, "schema-v10-staging").unwrap(); + let staging = migration_sibling(&root, "schema-v11-staging").unwrap(); fs::create_dir_all(&staging).unwrap(); fs::write(staging.join(RUNTIME_FILE), b"{\"schema_version\":8}").unwrap(); - let backup = migration_sibling(&root, "pre-schema-v10-backup").unwrap(); + let backup = migration_sibling(&root, "pre-schema-v11-backup").unwrap(); fs::rename(&root, &backup).unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -3129,7 +3218,7 @@ mod tests { let root = temp.path().join("runtime-store"); write_empty_schema_v8_store(&root); migrate_runtime_store_in_place(&root, "test-runtime").unwrap(); - let backup = migration_sibling(&root, "pre-schema-v10-backup").unwrap(); + let backup = migration_sibling(&root, "pre-schema-v11-backup").unwrap(); write_empty_schema_v8_store(&backup); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index 4e6f19427..bc7994de9 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -640,7 +640,7 @@ impl FsWorkerRetentionProvider { ) -> Result<(WorkerSessionArchiveManifest, PathBuf), RuntimeError> { let (manifest, session_root) = self.session_archive_identity(archive_id)?; let archive_dir = self.archive_dir(archive_id)?; - let manifest = validate_existing_archive(&archive_dir, &manifest)?; + let manifest = validate_existing_archive(&archive_dir, &manifest, false)?; Ok((manifest, session_root)) } @@ -1108,7 +1108,7 @@ fn commit_session_archive( }; let archive_dir = provider.archive_dir(archive_id)?; if archive_dir.exists() { - return validate_existing_archive(&archive_dir, &manifest); + return validate_existing_archive(&archive_dir, &manifest, true); } let parent = archive_dir .parent() @@ -1150,7 +1150,7 @@ fn commit_session_archive( source, })?; sync_directory(parent, "archive Worker Session")?; - validate_existing_archive(&archive_dir, &manifest) + validate_existing_archive(&archive_dir, &manifest, true) })(); if result.is_err() { let _ = fs::remove_dir_all(&staging); @@ -1158,17 +1158,165 @@ fn commit_session_archive( result } +// The public archive manifest is an immutable ingestion receipt also retained by +// Backend. Schema migration replaces the payload, not that receipt. This local +// storage attestation binds the new bytes to the exact verified ingestion receipt. +// It is published only with the entire staged Runtime tree, never by a reader. +#[derive(Serialize, Deserialize)] +struct MigratedArchiveStorage { + schema_version: u32, + ingestion: WorkerSessionArchiveManifest, + checksum_sha256: String, + bytes: u64, + file_count: u64, +} + +const ARCHIVE_STORAGE_FILE: &str = "session-storage.json"; + +fn archive_storage( + archive_dir: &Path, + ingestion: &WorkerSessionArchiveManifest, +) -> Result, RuntimeError> { + let path = archive_dir.join(ARCHIVE_STORAGE_FILE); + match fs::read(&path) { + Ok(bytes) => { + let storage: MigratedArchiveStorage = + serde_json::from_slice(&bytes).map_err(|error| RuntimeError::StoreCorrupt { + operation: "read migrated archive storage", + path: path.clone(), + message: error.to_string(), + })?; + if storage.schema_version != 1 || &storage.ingestion != ingestion { + return Err(RuntimeError::StoreCorrupt { + operation: "read migrated archive storage", + path, + message: "archive storage attestation does not match ingestion receipt".into(), + }); + } + Ok(Some(storage)) + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(source) => Err(RuntimeError::StoreIo { + operation: "read migrated archive storage", + path, + source, + }), + } +} + +/// Called only on the private Runtime migration staging tree under its owner +/// lock. Failure discards the whole tree, including manifests and attestations. +pub(crate) fn migrate_staged_sessions(root: &Path) -> Result<(), RuntimeError> { + for directory in ["workers", "orphaned-workers", "archives/workers"] { + let directory = root.join(directory); + let entries = match fs::read_dir(&directory) { + Ok(entries) => entries, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(source) => { + return Err(RuntimeError::StoreIo { + operation: "enumerate Session migration", + path: directory, + source, + }); + } + }; + for entry in entries { + let entry = entry.map_err(|source| RuntimeError::StoreIo { + operation: "enumerate Session migration", + path: directory.clone(), + source, + })?; + let aggregate = entry.path(); + let session = aggregate.join("session"); + if !session + .join("session.json") + .try_exists() + .map_err(|source| RuntimeError::StoreIo { + operation: "inspect Session migration", + path: session.clone(), + source, + })? + { + continue; + } + let identity: CanonicalSessionManifest = read_json( + &session.join("session.json"), + "read Session migration identity", + )?; + let session_id = identity.session_id.parse().map_err(|_| { + RuntimeError::InvalidRequest("Session migration identity is invalid".into()) + })?; + match runtime_session_retention(root).ensure_live(session_id) { + Ok(()) => {} + // Erasure fences remain authoritative even if physical removal + // was interrupted. Do not inspect or migrate fenced originals. + Err(session_store::StoreError::SessionErased(_)) => continue, + Err(error) => return Err(retention_error(error)), + } + let archive = if directory == root.join("archives/workers") { + let manifest: WorkerSessionArchiveManifest = read_json( + &aggregate.join("manifest.json"), + "read archive before Session migration", + )?; + Some(validate_existing_archive(&aggregate, &manifest, false)?) + } else { + None + }; + session_store::WorkerSessionStore::new(&session).map_err(|error| { + RuntimeError::StoreCorrupt { + operation: "migrate Session schema", + path: session.clone(), + message: error.to_string(), + } + })?; + if let Some(ingestion) = archive { + let (checksum_sha256, bytes, file_count) = checksum_files( + &collect_files(&session, "migrate archive storage")?, + "migrate archive storage", + )?; + atomic_write_json( + &aggregate.join(ARCHIVE_STORAGE_FILE), + &MigratedArchiveStorage { + schema_version: 1, + ingestion, + checksum_sha256, + bytes, + file_count, + }, + "commit migrated archive storage", + )?; + } + } + } + Ok(()) +} + fn validate_existing_archive( archive_dir: &Path, expected: &WorkerSessionArchiveManifest, + require_current_source: bool, ) -> Result { let existing: WorkerSessionArchiveManifest = read_json( &archive_dir.join("manifest.json"), "verify Worker Session archive", )?; + let storage = archive_storage(archive_dir, &existing)?; let mut comparable_expected = expected.clone(); comparable_expected.archived_at_unix_seconds = existing.archived_at_unix_seconds; - if existing != comparable_expected { + let mut current = existing.clone(); + if let Some(storage) = &storage { + current.content_checksum_sha256 = storage.checksum_sha256.clone(); + current.content_bytes = storage.bytes; + current.content_file_count = storage.file_count; + } + // Reads use the original receipt; a retry with a migrated live source uses + // the current storage summary. All other identity/policy fields must agree. + let expected_manifest = if require_current_source { + ¤t + } else { + &existing + }; + if expected_manifest != &comparable_expected { return Err(RuntimeError::StoreCorrupt { operation: "verify Worker Session archive", path: archive_dir.join("manifest.json"), @@ -1180,9 +1328,9 @@ fn validate_existing_archive( "verify Worker Session archive", )?; let (checksum, bytes, count) = checksum_files(&files, "verify Worker Session archive")?; - if checksum != existing.content_checksum_sha256 - || bytes != existing.content_bytes - || count != existing.content_file_count + if checksum != current.content_checksum_sha256 + || bytes != current.content_bytes + || count != current.content_file_count { return Err(RuntimeError::StoreCorrupt { operation: "verify Worker Session archive", @@ -1660,6 +1808,249 @@ mod tests { } } + fn old_compacted_source(root: &Path, worker_id: WorkerId) -> session_store::SegmentId { + use session_store::{ + LogEntry, LoggedContentPart, LoggedHistoryEntry, LoggedItem, LoggedRole, + LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, + Store, + }; + let worker = root.join("workers").join(worker_id.to_string()); + write_json( + &worker.join("worker.json"), + &serde_json::json!({"workspace_id": "workspace-a"}), + ); + let store = session_store::WorkerSessionStore::new(worker.join("session")).unwrap(); + let session_id = session_store::new_session_id(); + let parent = session_store::new_segment_id(); + let compact = session_store::new_segment_id(); + let original = LoggedHistoryEntry { + item: LoggedItem::Message { + role: LoggedRole::User, + content: vec![LoggedContentPart::Text { + text: "request".into(), + }], + }, + metadata: LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId("original".into()), + origin: LoggedSessionHistoryOrigin::LocalHumanInput, + derivation: None, + }, + }; + let summary = LoggedHistoryEntry { + item: LoggedItem::Message { + role: LoggedRole::System, + content: vec![LoggedContentPart::Text { + text: "[Compacted context summary]\n\nsummary".into(), + }], + }, + metadata: LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId("summary".into()), + origin: LoggedSessionHistoryOrigin::DerivedSummary, + derivation: Some(session_store::LoggedHistoryDerivation { + sources: vec![original.metadata.entry_id.clone()], + }), + }, + }; + for (id, history, compacted_from) in [ + (parent, vec![original], None), + ( + compact, + vec![summary], + Some(session_store::SegmentOrigin { + segment_id: parent, + at_turn_index: 0, + }), + ), + ] { + store + .create_segment( + session_id, + id, + &[LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id, + system_prompt: None, + config: Default::default(), + history, + forked_from: None, + compacted_from, + }], + ) + .unwrap(); + } + let mut metadata = session_store::WorkerMetadata::new( + worker_id.to_string(), + Some(session_store::WorkerActiveSegmentRef::active_segment( + session_id, compact, + )), + ); + metadata.workspace_id = Some("workspace-a".into()); + write_json(&worker.join("metadata.json"), &metadata); + write_json( + &worker.join("session/session.json"), + &serde_json::json!({"schema_version": 3, "session_id": session_id}), + ); + compact + } + + #[test] + fn session_migration_repairs_failed_archive_and_preserves_ingestion_receipt_on_retry() { + let temp = tempfile::tempdir().unwrap(); + let worker = WorkerId::now_v7(); + let compact = old_compacted_source(temp.path(), worker); + let provider = FsWorkerRetentionProvider::new(temp.path()); + let request = request(worker, SessionDisposition::Archive); + assert!( + provider + .execute(&request) + .unwrap_err() + .to_string() + .contains("Archived canonical source is unreadable") + ); + let archive_dir = provider.archive_dir("archive-a").unwrap(); + let original_manifest = fs::read(archive_dir.join("manifest.json")).unwrap(); + // Production invokes this only in the private tree before atomic switch. + migrate_staged_sessions(temp.path()).unwrap(); + assert_eq!( + fs::read(archive_dir.join("manifest.json")).unwrap(), + original_manifest + ); + let (manifest, session_root) = provider.session_archive("archive-a").unwrap(); + let reader = session_store::public_material::SessionPublicMaterialReader::open_worker( + &session_root, + compact, + Default::default(), + ) + .unwrap(); + assert_eq!(reader.descriptors().len(), 1); + let once = fs::read(archive_dir.join(ARCHIVE_STORAGE_FILE)).unwrap(); + migrate_staged_sessions(temp.path()).unwrap(); + assert_eq!( + fs::read(archive_dir.join(ARCHIVE_STORAGE_FILE)).unwrap(), + once + ); + let result = provider.execute(&request).unwrap(); + assert!(result.source_removed); + assert_eq!(result.archive.as_ref(), Some(&manifest)); + assert_eq!(provider.execute(&request).unwrap(), result); + assert!(!provider.worker_dir(worker).exists()); + // Both historical receipt and rewritten payload remain verified. + fs::write( + session_root + .join("segments") + .join(format!("{compact}.jsonl")), + b"tampered", + ) + .unwrap(); + assert!(provider.session_archive("archive-a").is_err()); + } + + #[test] + fn session_migration_updates_published_archive_without_changing_completed_receipt() { + let temp = tempfile::tempdir().unwrap(); + let worker = WorkerId::now_v7(); + let compact = old_compacted_source(temp.path(), worker); + let provider = FsWorkerRetentionProvider::new(temp.path()); + let path = provider + .worker_dir(worker) + .join("session/segments") + .join(format!("{compact}.jsonl")); + let old = fs::read_to_string(&path).unwrap(); + fs::write(&path, old.replace("derived_summary", "compaction_summary")).unwrap(); + let request = request(worker, SessionDisposition::Archive); + // Frozen schema-v3 completed-operation fixture: the old Runtime had + // already accepted this payload and removed its live source. + let archive = + commit_session_archive(&provider, &request, &provider.worker_dir(worker)).unwrap(); + let completed = WorkerRetentionExecutionResult { + operation_id: request.operation_id.clone(), + input_fingerprint: request.input_fingerprint.clone(), + worker_id: worker, + session_disposition: SessionDisposition::Archive, + diagnostics_disposition: DiagnosticsDisposition::Purge, + archive: Some(archive), + source_removed: true, + diagnostics_retained: false, + }; + write_json( + &provider.operation_path(&request.operation_id).unwrap(), + &RetentionOperationReceipt { + schema_version: OPERATION_SCHEMA_VERSION, + request: request.clone(), + result: completed.clone(), + }, + ); + fs::remove_dir_all(provider.worker_dir(worker)).unwrap(); + let receipt = fs::read(provider.operation_path(&request.operation_id).unwrap()).unwrap(); + migrate_staged_sessions(temp.path()).unwrap(); + let (manifest, root) = provider.session_archive("archive-a").unwrap(); + assert_eq!(completed.archive.as_ref(), Some(&manifest)); + assert_eq!( + fs::read(provider.operation_path(&request.operation_id).unwrap()).unwrap(), + receipt + ); + assert_eq!(provider.execute(&request).unwrap(), completed); + assert!(session_store::WorkerSessionStore::open_read_only(&root).is_ok()); + let storage_path = provider + .archive_dir("archive-a") + .unwrap() + .join(ARCHIVE_STORAGE_FILE); + let mut storage: MigratedArchiveStorage = read_json(&storage_path, "test").unwrap(); + storage.ingestion.archive_id = "another-archive".into(); + write_json(&storage_path, &storage); + assert!(provider.session_archive("archive-a").is_err()); + } + + #[test] + fn session_migration_rejects_corrupt_archive_before_replacing_its_bytes() { + let temp = tempfile::tempdir().unwrap(); + let worker = WorkerId::now_v7(); + let compact = old_compacted_source(temp.path(), worker); + let provider = FsWorkerRetentionProvider::new(temp.path()); + let request = request(worker, SessionDisposition::Archive); + assert!(provider.execute(&request).is_err()); + let archive = provider.archive_dir("archive-a").unwrap(); + let path = archive + .join("session/segments") + .join(format!("{compact}.jsonl")); + fs::write(&path, b"corrupt\n").unwrap(); + assert!(migrate_staged_sessions(temp.path()).is_err()); + assert_eq!(fs::read(&path).unwrap(), b"corrupt\n"); + assert!(!archive.join(ARCHIVE_STORAGE_FILE).exists()); + } + + #[test] + fn session_migration_does_not_read_or_upgrade_erased_originals() { + let temp = tempfile::tempdir().unwrap(); + let worker = WorkerId::now_v7(); + let compact = old_compacted_source(temp.path(), worker); + let session = temp + .path() + .join("workers") + .join(worker.to_string()) + .join("session"); + let manifest = fs::read(session.join("session.json")).unwrap(); + let identity: CanonicalSessionManifest = serde_json::from_slice(&manifest).unwrap(); + let id = identity.session_id.parse().unwrap(); + let retention = runtime_session_retention(temp.path()); + retention + .delete( + id, + session_store::retention::SessionDeletionIntent::ErasePermanently, + || Ok(()), + ) + .unwrap(); + let path = session.join("segments").join(format!("{compact}.jsonl")); + fs::write(&path, b"unreadable fenced content\n").unwrap(); + migrate_staged_sessions(temp.path()).unwrap(); + assert_eq!(fs::read(session.join("session.json")).unwrap(), manifest); + assert_eq!(fs::read(path).unwrap(), b"unreadable fenced content\n"); + assert!(matches!( + retention.ensure_live(id), + Err(session_store::StoreError::SessionErased(_)) + )); + } + #[test] fn archive_is_verified_before_source_removal_and_retry_converges() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index b0c523cf9..d988a491d 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -13097,11 +13097,9 @@ mod tests { display_name: None, }) .unwrap_err(); - assert!( - error - .to_string() - .contains("unsupported Runtime store schema version 2; expected 6, 7, 8, 9, or 10") - ); + assert!(error.to_string().contains( + "unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, or 11" + )); let _ = std::fs::remove_dir_all(root); } @@ -13142,7 +13140,7 @@ mod tests { let worker_aggregate: serde_json::Value = serde_json::from_slice(&std::fs::read(worker_store_dir.join("worker.json")).unwrap()) .unwrap(); - assert_eq!(worker_aggregate["schema_version"], serde_json::json!(10)); + assert_eq!(worker_aggregate["schema_version"], serde_json::json!(11)); assert_eq!(worker_aggregate["status"], serde_json::json!("stopped")); assert_eq!( worker_aggregate["execution_state"]["state"], @@ -14166,7 +14164,7 @@ mod tests { ); let migrated_aggregate: serde_json::Value = serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); - assert_eq!(migrated_aggregate["schema_version"], serde_json::json!(10)); + assert_eq!(migrated_aggregate["schema_version"], serde_json::json!(11)); assert!(migrated_aggregate.get("request").is_none()); assert!(migrated_aggregate.get("execution").is_none()); assert_eq!( diff --git a/crates/worker/src/session_history.rs b/crates/worker/src/session_history.rs index d48927a32..4ba00e943 100644 --- a/crates/worker/src/session_history.rs +++ b/crates/worker/src/session_history.rs @@ -101,10 +101,8 @@ fn from_logged(entry: &LoggedHistoryEntry) -> HistoryEntry { + LogEntry::AnnotatedSegmentStart { history: seed, .. } => { history = seed.iter().map(from_logged).collect(); - // Older Host compactions always put this DerivedSummary system item - // first in the replacement seed. Only that trusted construction boundary - // may interpret the marker; never infer origin from appended/pasted text. - if compacted_from.is_some() - && let Some(first) = history.first_mut() - && first.annotation.origin == WorkerHistoryProvenance::DerivedSummary - && matches!( - &first.item, - Item::Message { - role: agen::Role::System, - .. - } - ) - && first - .item - .as_text() - .is_some_and(|text| text.starts_with(COMPACTION_SUMMARY_HEADER)) - { - first.annotation.origin = WorkerHistoryProvenance::CompactionSummary; - } } LogEntry::AnnotatedUserInput { history: input, .. } => { history.extend(input.iter().map(from_logged)) @@ -181,7 +155,7 @@ mod tests { use session_store::LogEntry; #[test] - fn restore_recognizes_only_legacy_host_compaction_seed_summary() { + fn restore_preserves_persisted_origin_without_legacy_reinterpretation() { let summary = history_entry( Item::system_message(format!("{COMPACTION_SUMMARY_HEADER}host summary")), WorkerHistoryProvenance::DerivedSummary, @@ -213,11 +187,7 @@ mod tests { .unwrap(); assert_eq!( restored[0].annotation.origin, - if compacted { - WorkerHistoryProvenance::CompactionSummary - } else { - WorkerHistoryProvenance::DerivedSummary - } + WorkerHistoryProvenance::DerivedSummary ); assert_eq!( restored[1].annotation.origin, diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 1420f3eb4..8c9a06246 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -5576,6 +5576,7 @@ impl Worker { self.apply_interrupt_prep()?; self.last_run_interrupted = false; self.commit_entry(LogEntry::PausedTurnAbandoned { + entry_id: None, ts: segment_log::now_millis(), })?; Ok(()) @@ -6636,6 +6637,7 @@ impl Worker { EngineRunExit::Yielded | EngineRunExit::Interrupted(_) => unreachable!(), }; self.commit_entry(LogEntry::RunCompleted { + entry_id: None, ts: segment_log::now_millis(), interrupted, result, @@ -6653,6 +6655,7 @@ impl Worker { } EngineRunExit::Interrupted(RunInterruptionReason::LimitReached) => { self.commit_entry(LogEntry::RunCompleted { + entry_id: None, ts: segment_log::now_millis(), interrupted: false, result: EngineResult::LimitReached, @@ -14512,6 +14515,7 @@ permission = "write" ); worker .commit_entry(LogEntry::RunCompleted { + entry_id: None, ts: segment_log::now_millis(), interrupted: false, result: EngineResult::Finished, diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index ad46564ab..2aeb8f521 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -2468,8 +2468,7 @@ async fn pasted_markers_and_other_derived_context_remain_ordinary_overview_mater } #[tokio::test] -async fn restored_legacy_summary_requires_compacted_lineage_first_position_origin_role_and_marker() -{ +async fn restore_does_not_reinterpret_legacy_summary_markers() { use session_store::LoggedSessionHistoryOrigin::{ CompactionSummary, DerivedSummary, LegacyUnknown, }; @@ -2525,13 +2524,9 @@ async fn restored_legacy_summary_requires_compacted_lineage_first_position_origi .iter() .find(|entry| entry.annotation.entry_id == candidate_id) .unwrap(); - assert_eq!( - restored.annotation.origin == CompactionSummary, - case == "trusted", - "{case}" - ); - // A restore-time upgrade is an in-memory interpretation of the trusted - // seed; it must not silently rewrite the historical durable record. + assert_ne!(restored.annotation.origin, CompactionSummary, "{case}"); + // Migration belongs to the persisted store, not this restore path. + // Even a legacy Host seed must not be silently reinterpreted here. assert!( replacement_seed(&fixture.worker) .iter() @@ -2541,11 +2536,7 @@ async fn restored_legacy_summary_requires_compacted_lineage_first_position_origi fixture.worker.compact(0).await.unwrap(); let input = first_compactor_input(&observed.requests()[1]); - if case == "trusted" { - assert_full_initial_summary(&input, &inherited); - } else { - assert!(!input.contains(INITIAL_SUMMARY_HEADING), "{case}"); - } + assert!(!input.contains(INITIAL_SUMMARY_HEADING), "{case}"); } } diff --git a/docs/development/workspace-schema-migrations.md b/docs/development/workspace-schema-migrations.md index 450757548..fa91cfbf3 100644 --- a/docs/development/workspace-schema-migrations.md +++ b/docs/development/workspace-schema-migrations.md @@ -7,3 +7,11 @@ schema変更では、既存の順序付きmigrationへ新しい処理を追加 検証には、最古の対応schemaからの更新、保存値と制約の維持、新規作成結果との一致、失敗時のrollbackを含める。途中まで完了した更新の再開や、元DBを変更しないdry-runを提供する場合は、その保証も実装経路で確認する。schema番号だけのassertionでは不足する。 古いbaselineの対応終了は、対象となる保存データを移行または明示的に破棄する判断を含め、独立にレビューする。残す最古のschema、移行・破棄の責任、影響するテストを明記する。所有storeのmigration経路と検証を共用し、起動・管理操作ごとに別のDDL判定を増やさない。 + +## Session履歴の移行 + +旧Host compactionのoriginはSession storeのversioned migrationで永続的に更新する。通常復元やpublic readerでは文字列markerによる読み替えを行わない。本文・logical entry ID・derivationを維持し、位置由来のRun boundary IDは書換え前のoffsetから保存する。StandaloneのSessionも同じ変換を使う。 + +Runtimeの移行はowner lock下のstaging treeで停止中Worker・orphan・既存Session archiveを更新し、成功したtreeだけを公開する。全treeのコピーに必要な空き容量が必要。失敗時は元treeを維持し、切替中の中断は既存のstaging/backup回復経路で扱う。 + +Archiveの`manifest.json`はBackend catalogと完了receiptにも保存された取り込み時の証拠なので変更しない。移行前のchecksumを検証したうえでpayloadを現行形式に置換し、元manifest全体と現行payloadのchecksum/bytes/file countを結ぶ`session-storage.json`を同じtree切替で公開する。旧payloadは読取り用に残さない。通常のarchive検証では現行payloadを検証し、APIとoperation replayには元の取り込みreceiptを返す。移行後の削除再試行では、現行原本とarchiveの現行storage summaryが一致することを要求する。