From daf1f6ed5089acb8640dca536adc14755ee6e64d Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 20 Oct 2026 03:52:55 +0900 Subject: [PATCH] fix: route retained Session migration through Runtime staging --- crates/worker-runtime/src/fs_store.rs | 57 +- crates/worker-runtime/src/retention.rs | 112 ++-- .../retention/fixtures/system-input-v4.jsonl | 2 + .../src/retention/runtime_v5_tests.rs | 559 ++++++++++++++++++ 4 files changed, 680 insertions(+), 50 deletions(-) create mode 100644 crates/worker-runtime/src/retention/fixtures/system-input-v4.jsonl create mode 100644 crates/worker-runtime/src/retention/runtime_v5_tests.rs diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 5b9c35b85..de799b323 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -657,18 +657,32 @@ fn plan_runtime_store_migration( )); } if current_schema_version == SCHEMA_VERSION { + // Runtime and Session schemas evolve independently. Probe only through + // Session Store's non-mutating current-schema admission boundary; any + // conversion still runs on the existing owner-locked staging tree. + let migration_required = crate::retention::sessions_require_migration(root)?; + let excluded_ephemeral_paths = if migration_required { + runtime_tree_exclusions(root)? + } else { + Vec::new() + }; + let archived_orphan_worker_count = if migration_required { + orphaned_worker_directories(root)?.len() + } else { + 0 + }; let plan = FsRuntimeStoreMigrationPlan { current_schema_version, target_schema_version: SCHEMA_VERSION, - migration_required: false, + migration_required, worker_count: 0, - archived_orphan_worker_count: 0, + archived_orphan_worker_count, migrated_worker_aggregate_count: 0, migrated_diagnostic_worker_ref_count: 0, cleared_diagnostic_worker_ref_count: 0, mapping_digest: legacy_worker_identity_mapping_digest(&[]), mappings: Vec::new(), - excluded_ephemeral_paths: Vec::new(), + excluded_ephemeral_paths, }; return Ok((plan, Vec::new())); } @@ -1808,7 +1822,9 @@ fn migrate_runtime_store( Ok(plan) } -fn archive_orphaned_worker_directories(root: &Path) -> Result { +fn orphaned_worker_directories( + root: &Path, +) -> Result, RuntimeError> { let workers_dir = root.join(WORKERS_DIR); let mut orphaned = fs::read_dir(&workers_dir) .map_err(|source| runtime_io_error("read Worker records", &workers_dir, source))? @@ -1821,6 +1837,11 @@ fn archive_orphaned_worker_directories(root: &Path) -> Result>(); orphaned.sort_by(|(left, _), (right, _)| left.cmp(right)); + Ok(orphaned) +} + +fn archive_orphaned_worker_directories(root: &Path) -> Result { + let orphaned = orphaned_worker_directories(root)?; if orphaned.is_empty() { return Ok(0); } @@ -1855,6 +1876,21 @@ fn migrate_runtime_store_in_place( if !plan.migration_required { return Ok(plan); } + if plan.current_schema_version == SCHEMA_VERSION { + // Session-only migration: do not reserialize Runtime/Worker identity, + // Restore guards, recovery fences, or retained execution receipts. + // As in a Runtime schema migration, retain orphan trees outside the + // Worker catalog before validating the completed staging tree. + let archived = archive_orphaned_worker_directories(root)?; + if archived != plan.archived_orphan_worker_count { + return Err(runtime_store_corrupt( + root, + "orphan count changed during Session migration".into(), + )); + } + crate::retention::migrate_staged_sessions(root)?; + return Ok(plan); + } let runtime_path = root.join(RUNTIME_FILE); let bytes = fs::read(&runtime_path).map_err(|error| runtime_io_error("read", &runtime_path, error))?; @@ -1948,11 +1984,9 @@ fn migrate_runtime_store_in_place( ), )); } - // Schema 11 already migrated Sessions; the recovery-fence bump must not - // rewrite retained history or archive attestations. - if plan.current_schema_version != SOURCE_SCHEMA_VERSION { - crate::retention::migrate_staged_sessions(root)?; - } + // A Runtime schema bump says nothing about the retained Session schema. + // Session Store owns versioned conversion, including v4 -> generic v5. + crate::retention::migrate_staged_sessions(root)?; let (document, _) = migrate_runtime_document( document, plan.current_schema_version, @@ -3011,9 +3045,8 @@ mod tests { } else { drop(result.unwrap()); assert_eq!(stored_schema_version(&root), u64::from(SCHEMA_VERSION)); - let manifest: serde_json::Value = - read_json(&session.join("session.json"), "test").unwrap(); - assert_eq!(manifest["schema_version"], 4); + session_store::WorkerSessionStore::open_read_only(&session) + .expect("startup migration must admit the current Session schema"); 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); diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index 9777e7390..d488c9608 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -1204,9 +1204,13 @@ fn archive_storage( } } -/// 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> { +// Share discovery and erasure fencing between the read-only Runtime preflight +// and its private staging pass. Session Store remains the authority for schema +// admission and conversion; Runtime must not duplicate its version table. +fn for_each_live_session( + root: &Path, + mut visit: impl FnMut(&Path, bool) -> Result<(), RuntimeError>, +) -> Result<(), RuntimeError> { for directory in ["workers", "orphaned-workers", "archives/workers"] { let directory = root.join(directory); let entries = match fs::read_dir(&directory) { @@ -1253,44 +1257,72 @@ pub(crate) fn migrate_staged_sessions(root: &Path) -> Result<(), RuntimeError> { 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", - )?; - } + visit(&aggregate, directory == root.join("archives/workers"))?; } } Ok(()) } +/// Read-only admission probe: a rejected Session needs the existing staged +/// migration attempt (which will also reject corrupt/unsupported inputs). +/// Never create a writable Session store or rewrite archive attestations here. +pub(crate) fn sessions_require_migration(root: &Path) -> Result { + let mut required = false; + for_each_live_session(root, |aggregate, _| { + required |= + session_store::WorkerSessionStore::open_read_only(aggregate.join("session")).is_err(); + Ok(()) + })?; + Ok(required) +} + +/// 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_each_live_session(root, |aggregate, is_archive| { + let session = aggregate.join("session"); + let archive = if is_archive { + let manifest: WorkerSessionArchiveManifest = read_json( + &aggregate.join("manifest.json"), + "read archive before Session migration", + )?; + Some(validate_existing_archive(aggregate, &manifest, false)?) + } else { + None + }; + // Current Sessions need no writable admission, and their storage + // attestations must remain byte-for-byte unchanged. + if session_store::WorkerSessionStore::open_read_only(&session).is_ok() { + return Ok(()); + } + 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, @@ -1753,12 +1785,16 @@ fn temporary_path(path: &Path) -> PathBuf { path.with_file_name(format!(".{name}.tmp-{}-{sequence}", std::process::id())) } +#[cfg(all(test, feature = "fs-store"))] +#[path = "retention/runtime_v5_tests.rs"] +mod runtime_v5_tests; + #[cfg(test)] mod tests { use super::*; use std::sync::{Arc, Barrier}; - fn write_json(path: &Path, value: &impl Serialize) { + pub(super) fn write_json(path: &Path, value: &impl Serialize) { fs::create_dir_all(path.parent().unwrap()).unwrap(); fs::write(path, serde_json::to_vec_pretty(value).unwrap()).unwrap(); } @@ -1786,7 +1822,7 @@ mod tests { fs::write(worker.join("runs/attempt-a/worker.sock"), b"not retained").unwrap(); } - fn request( + pub(super) fn request( worker_id: WorkerId, disposition: SessionDisposition, ) -> WorkerRetentionExecutionRequest { diff --git a/crates/worker-runtime/src/retention/fixtures/system-input-v4.jsonl b/crates/worker-runtime/src/retention/fixtures/system-input-v4.jsonl new file mode 100644 index 000000000..00a1557f2 --- /dev/null +++ b/crates/worker-runtime/src/retention/fixtures/system-input-v4.jsonl @@ -0,0 +1,2 @@ +{"kind":"annotated_system_item","ts":101,"entry":{"item":{"kind":"journal_startup","body":"journal body\nunchanged","snapshot":{"entries":[{"text":"private snapshot"}]},"snapshot_sha256":"snapshot-digest","prompt_provenance":{"workspace_id":"workspace","source_digest":"source","projection_digest":"projection","logical_name":"journal_startup"}},"metadata":{"entry_id":"journal-id","origin":{"kind":"legacy_unknown"},"derivation":{"sources":["source-id"]}}},"extensions":[{"domain":"migration-fixture","payload":{"kept":true}}]} +{"kind":"annotated_system_item","ts":103,"entry":{"item":{"kind":"subject_behavior_refresh","body":"rendered subject body","behavior_md":"exact behavior","prompt_provenance":{"source_digest":"subject-source","projection_digest":"subject-projection","logical_name":"subject_behavior"}},"metadata":{"entry_id":"subject-id","origin":{"kind":"backend_instruction","operation_id":"op"}}}} diff --git a/crates/worker-runtime/src/retention/runtime_v5_tests.rs b/crates/worker-runtime/src/retention/runtime_v5_tests.rs new file mode 100644 index 000000000..8c0123606 --- /dev/null +++ b/crates/worker-runtime/src/retention/runtime_v5_tests.rs @@ -0,0 +1,559 @@ +//! Runtime integration, not a second implementation of Session migration. +//! The Feature records are frozen v4 wire copied from the Session Store fixture; +//! only the empty segment envelope is built with current APIs. +use super::tests::{request, write_json}; +use super::*; +use crate::fs_store::{FsRuntimeStore, FsRuntimeStoreOptions}; +use serde_json::{Value, json}; +use session_store::{LogEntry, SegmentId, SessionId, Store, WorkerSessionStore}; +use std::collections::BTreeMap; + +const V4_SYSTEM_INPUTS: &str = include_str!("fixtures/system-input-v4.jsonl"); + +fn write_v4_session(aggregate: &Path) -> (SessionId, SegmentId) { + let root = aggregate.join("session"); + let session = session_store::new_session_id(); + let segment = session_store::new_segment_id(); + let store = WorkerSessionStore::new(&root).unwrap(); + store + .create_segment( + session, + segment, + &[LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id: session, + system_prompt: None, + config: Default::default(), + history: vec![], + forked_from: None, + compacted_from: None, + }], + ) + .unwrap(); + let path = root.join("segments").join(format!("{segment}.jsonl")); + let mut bytes = fs::read(&path).unwrap(); + bytes.extend_from_slice(V4_SYSTEM_INPUTS.as_bytes()); + fs::write(path, bytes).unwrap(); + write_json( + &root.join("session.json"), + &json!({ + "schema_version": 4, "session_id": session, + }), + ); + (session, segment) +} + +fn stopped_worker(root: &Path, schema: u32) -> (WorkerId, SessionId, SegmentId) { + let worker = WorkerId::now_v7(); + let aggregate = root.join("workers").join(worker.to_string()); + // Serialize request defaults just as a production aggregate writer does. + let request: crate::catalog::CreateWorkerRequest = serde_json::from_value(json!({ + "worker_id": worker, + "workspace_api": {"workspace_id": "workspace-a", "base_url": "https://workspace.invalid"}, + "create_fingerprint": "runtime-v5-fixture", + "profile": {"kind": "builtin", "value": "builtin:coder"}, + "profile_source": {"kind": "workspace_config", "archive": { + "id": "archive-1", "digest": "sha256:archive", "size_bytes": 0, + "source_graph": {"source_count": 0, "total_source_bytes": 0, + "entrypoints": {}, "import_count": 0} + }} + })) + .unwrap(); + write_json( + &aggregate.join("worker.json"), + &json!({ + "schema_version": schema, + "worker_ref": {"worker_id": worker}, "worker_id": worker, + "profile": request.profile, + "profile_source": request.profile_source.reference(), + "display_name": null, "config_bundle": null, + "created_at_ms": 1234, "status": "stopped", "workspace_id": "workspace-a", + "last_finished_submission_request_id": "retained-submission", + "restore_guard": {"observation_token": "retained-token", + "active_request_id": null, "owners": {}}, + "execution_state": {"state": "available", "execution": { + "request": request, "binding": {}, "restore_intent": "explicit" + }} + }), + ); + let (session, segment) = write_v4_session(&aggregate); + let mut metadata = session_store::WorkerMetadata::new( + format!("worker-runtime-{worker}"), + Some(session_store::WorkerActiveSegmentRef::active_segment( + session, segment, + )), + ); + metadata.workspace_id = Some("workspace-a".into()); + write_json(&aggregate.join("metadata.json"), &metadata); + (worker, session, segment) +} + +fn runtime_fixture(root: &Path, schema: u32) -> (WorkerId, SessionId, SegmentId) { + fs::create_dir_all(root.join("workers")).unwrap(); + let fenced_worker = WorkerId::now_v7(); + write_json( + &root.join("runtime.json"), + &json!({ + "schema_version": schema, "display_name": "retained Runtime", + "backend": "fs_store", "status": "running", "next_diagnostic_id": 1, + "config_bundles": {}, "workspace_owners": {"workspace-a": "owner-a"}, + "diagnostics": [], "subject_releases": [], + "subject_create_recovery_fences": {fenced_worker.to_string(): { + "request": {"workspace_id": "workspace-a", "worker_id": fenced_worker, + "create_fingerprint": "retained-create", "operation_id": "retained-recovery"}, + "response": null + }} + }), + ); + stopped_worker(root, schema) +} + +// Include every file, not just schema markers: failures must preserve original +// history, metadata, archive receipts, checksums and retention evidence together. +fn tree_bytes(root: &Path) -> BTreeMap> { + fn walk(root: &Path, at: &Path, result: &mut BTreeMap>) { + for entry in fs::read_dir(at).unwrap() { + let entry = entry.unwrap(); + if entry.file_type().unwrap().is_dir() { + walk(root, &entry.path(), result); + } else { + result.insert( + entry.path().strip_prefix(root).unwrap().to_path_buf(), + fs::read(entry.path()).unwrap(), + ); + } + } + } + let mut result = BTreeMap::new(); + walk(root, root, &mut result); + result +} + +fn assert_generic(root: &Path, session: SessionId, segment: SegmentId) { + use session_store::{InputCompactionUse, InputEvidenceUse, InputObservationUse}; + let store = WorkerSessionStore::open_read_only(root).unwrap(); + let entries = store.read_all_read_only(session, segment).unwrap(); + assert_eq!(entries.len(), 3); + let originals: Vec = V4_SYSTEM_INPUTS + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + for (log, old) in entries[1..].iter().zip(originals) { + let LogEntry::AnnotatedSystemItem { entry, .. } = log else { + panic!("lost System record: {log:?}"); + }; + let current = serde_json::to_value(log).unwrap(); + assert_eq!(current["ts"], old["ts"]); + for field in ["entry_id", "origin", "derivation"] { + assert_eq!( + current["entry"]["metadata"][field], + old["entry"]["metadata"][field] + ); + } + assert_eq!(current["entry"]["item"]["kind"], "input"); + assert_eq!( + entry.item.history_text(), + old["entry"]["item"]["body"].as_str().unwrap() + ); + let input = entry.metadata.system_input.as_ref().unwrap(); + let usage = input.usage(); + assert_eq!(usage.recording_material, InputEvidenceUse::Exclude); + assert_eq!(usage.memory_evidence, InputEvidenceUse::Exclude); + assert_eq!(usage.observation, InputObservationUse::Hidden); + let kind = old["entry"]["item"]["kind"].as_str().unwrap(); + let schema = if kind == "journal_startup" { + assert_eq!(usage.compaction, InputCompactionUse::ReferenceInput); + assert_eq!(current["extensions"], old["extensions"]); + "journal:startup" + } else { + assert_eq!(usage.compaction, InputCompactionUse::Retain); + "subject:behavior" + }; + let payload = input.feature_payload().unwrap(); + assert_eq!(payload.schema(), schema); + assert_eq!(payload.version(), 1); + let mut expected_payload = old["entry"]["item"].clone(); + expected_payload.as_object_mut().unwrap().remove("kind"); + expected_payload.as_object_mut().unwrap().remove("body"); + assert_eq!( + payload.decode::(schema, 1).unwrap(), + expected_payload + ); + let public = serde_json::to_value(input.public_item().unwrap()).unwrap(); + assert_eq!(public["kind"], kind); + assert_eq!(public["body"], old["entry"]["item"]["body"]); + assert_eq!(public["presentation"], json!({"kind": "context"})); + } +} + +fn completed_v4_archive( + root: &Path, + schema: u32, +) -> ( + SessionId, + SegmentId, + WorkerRetentionExecutionRequest, + WorkerRetentionExecutionResult, +) { + let (worker, session, segment) = stopped_worker(root, schema); + let provider = FsWorkerRetentionProvider::new(root); + let request = request(worker, SessionDisposition::Archive); + // Frozen old-runtime receipt: ingestion and removal completed before the + // current strict decoder existed. Do not ask the v5 reader to mint it. + let manifest = + 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(manifest), + 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(); + (session, segment, request, completed) +} + +#[test] +fn runtime_v11_and_current_route_v4_stopped_orphan_archive_through_staging() { + for runtime_schema in [11, 12] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + let (worker, session, segment) = runtime_fixture(&root, runtime_schema); + let orphan = root + .join("orphaned-workers") + .join(WorkerId::now_v7().to_string()); + let (orphan_session, orphan_segment) = write_v4_session(&orphan); + // Include an unregistered tree still under workers, not just an + // already-archived orphan. Both Runtime source schemas must fence it + // out of the live catalog without discarding its Session. + let loose_id = WorkerId::now_v7(); + let loose = root.join("workers").join(loose_id.to_string()); + let (loose_session, loose_segment) = write_v4_session(&loose); + let retained_loose = root.join("orphaned-workers").join(loose_id.to_string()); + let (archived_session, archived_segment, request, completed) = + completed_v4_archive(&root, runtime_schema); + let provider = FsWorkerRetentionProvider::new(&root); + let archive = provider.archive_dir("archive-a").unwrap(); + // Also cover an archive which already has a local storage attestation + // from an earlier schema migration. Keep ingestion bytes immutable; + // harmless marker whitespace makes current storage differ from them. + if runtime_schema == 12 { + let marker = archive.join("session/session.json"); + let mut bytes = fs::read(&marker).unwrap(); + bytes.push(b'\n'); + fs::write(marker, bytes).unwrap(); + let ingestion = completed.archive.clone().unwrap(); + let (checksum_sha256, bytes, file_count) = checksum_files( + &collect_files(&archive.join("session"), "test").unwrap(), + "test", + ) + .unwrap(); + write_json( + &archive.join(ARCHIVE_STORAGE_FILE), + &MigratedArchiveStorage { + schema_version: 1, + ingestion, + checksum_sha256, + bytes, + file_count, + }, + ); + } + let storage_before = fs::read(archive.join(ARCHIVE_STORAGE_FILE)).ok(); + let manifest_before = fs::read(archive.join("manifest.json")).unwrap(); + let receipt_path = provider.operation_path(&request.operation_id).unwrap(); + let receipt_before = fs::read(&receipt_path).unwrap(); + let stopped = root.join("workers").join(worker.to_string()); + assert!(WorkerSessionStore::open_read_only(stopped.join("session")).is_err()); + // Current decode itself must not be made permissive to fix routing. + for line in V4_SYSTEM_INPUTS.lines() { + assert!(serde_json::from_str::(line).is_err()); + } + let before = tree_bytes(&root); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); + let plan = FsRuntimeStore::migration_plan(&options).unwrap(); + assert!(plan.migration_required); + assert_eq!(plan.archived_orphan_worker_count, 1); + assert_eq!(plan.current_schema_version, runtime_schema); + assert_eq!(tree_bytes(&root), before, "preflight mutated source"); + // Exercise normal startup, not just the private migration helper. + let opened = FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap(); + assert!( + matches!( + FsRuntimeStore::migrate(&options), + Err(RuntimeError::RuntimeStoreAlreadyOpen { .. }) + ), + "owner lock not fenced" + ); + drop(opened); + assert_generic(&stopped.join("session"), session, segment); + assert_generic(&orphan.join("session"), orphan_session, orphan_segment); + assert!(!loose.exists()); + assert_generic( + &retained_loose.join("session"), + loose_session, + loose_segment, + ); + assert_generic(&archive.join("session"), archived_session, archived_segment); + assert_eq!( + fs::read(archive.join("manifest.json")).unwrap(), + manifest_before + ); + assert_eq!(fs::read(&receipt_path).unwrap(), receipt_before); + let (manifest, archive_session) = provider.session_archive("archive-a").unwrap(); + assert_eq!(completed.archive.as_ref(), Some(&manifest)); + assert_eq!(provider.execute(&request).unwrap(), completed); + if let Some(storage_before) = storage_before { + assert_ne!( + fs::read(archive.join(ARCHIVE_STORAGE_FILE)).unwrap(), + storage_before + ); + } + let storage: MigratedArchiveStorage = + read_json(&archive.join(ARCHIVE_STORAGE_FILE), "test").unwrap(); + assert_eq!(storage.ingestion, manifest); + let (checksum, bytes, count) = + checksum_files(&collect_files(&archive_session, "test").unwrap(), "test").unwrap(); + assert_eq!( + (storage.checksum_sha256, storage.bytes, storage.file_count), + (checksum, bytes, count) + ); + if runtime_schema == 12 { + for name in [ + PathBuf::from("runtime.json"), + PathBuf::from(format!("workers/{worker}/worker.json")), + PathBuf::from(format!("workers/{worker}/metadata.json")), + ] { + assert_eq!(fs::read(root.join(&name)).unwrap(), before[&name]); + } + } + let once = tree_bytes(&root); + assert!( + !FsRuntimeStore::migrate(&options) + .unwrap() + .migration_required + ); + drop(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").unwrap()); + assert_eq!(tree_bytes(&root), once, "retry rewrote migrated evidence"); + assert!(!temp.path().join(".runtime.schema-v12-staging").exists()); + assert!(!temp.path().join(".runtime.pre-schema-v12-backup").exists()); + // session-storage, not the immutable receipt's old checksum, attests + // current bytes. Tampering with those bytes must remain detectable. + let path = archive_session + .join("segments") + .join(format!("{archived_segment}.jsonl")); + fs::write(path, b"tampered\n").unwrap(); + assert!(provider.session_archive("archive-a").is_err()); + } +} + +#[test] +fn runtime_v5_archive_failure_discards_staging_without_losing_originals_and_retry_converges() { + for runtime_schema in [11, 12] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + let (worker, session, segment) = runtime_fixture(&root, runtime_schema); + let (archived_session, archived_segment, request, completed) = + completed_v4_archive(&root, runtime_schema); + let provider = FsWorkerRetentionProvider::new(&root); + let archive = provider.archive_dir("archive-a").unwrap(); + let path = archive + .join("session/segments") + .join(format!("{archived_segment}.jsonl")); + let original = fs::read(&path).unwrap(); + fs::write(&path, b"corrupt archive\n").unwrap(); + let before = tree_bytes(&root); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); + assert!( + FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + // workers are visited before archives: earlier successful staged + // conversion must not leak into the original tree on this later error. + assert!(FsRuntimeStore::open_or_create(root.clone(), "runtime-a").is_err()); + assert_eq!(tree_bytes(&root), before); + assert!(!archive.join(ARCHIVE_STORAGE_FILE).exists()); + fs::write(&path, original).unwrap(); + FsRuntimeStore::migrate(&options).unwrap(); + assert_generic( + &root + .join("workers") + .join(worker.to_string()) + .join("session"), + session, + segment, + ); + assert_generic(&archive.join("session"), archived_session, archived_segment); + assert_eq!(provider.execute(&request).unwrap(), completed); + } +} + +#[test] +fn runtime_v5_typed_legacy_payload_failure_keeps_valid_archive_evidence_on_retries() { + for (from, to) in [ + ( + "\"snapshot_sha256\":\"snapshot-digest\"", + "\"snapshot_sha256\":42", + ), + ("\"behavior_md\":\"exact behavior\"", "\"behavior_md\":[]"), + ] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + runtime_fixture(&root, 12); + let (worker, _, segment) = stopped_worker(&root, 12); + let provider = FsWorkerRetentionProvider::new(&root); + let path = provider + .worker_dir(worker) + .join("session/segments") + .join(format!("{segment}.jsonl")); + let original = fs::read_to_string(&path).unwrap(); + assert!(original.contains(from)); + fs::write(&path, original.replace(from, to)).unwrap(); + let request = request(worker, SessionDisposition::Archive); + let manifest = + commit_session_archive(&provider, &request, &provider.worker_dir(worker)).unwrap(); + fs::remove_dir_all(provider.worker_dir(worker)).unwrap(); + let archive = provider.archive_dir("archive-a").unwrap(); + // This archive checksum is valid; failure must come from the legacy + // typed decoder, not from deliberately broken ingestion evidence. + validate_existing_archive(&archive, &manifest, false).unwrap(); + let before = tree_bytes(&root); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); + for _ in 0..2 { + let error = FsRuntimeStore::migrate(&options).unwrap_err(); + let RuntimeError::StoreCorrupt { + operation, path, .. + } = error + else { + panic!("unexpected failure: {error}"); + }; + assert_eq!(operation, "migrate Session schema"); + assert!(path.starts_with(temp.path().join(".runtime.schema-v12-staging"))); + assert_eq!(tree_bytes(&root), before); + validate_existing_archive(&archive, &manifest, false).unwrap(); + } + } +} + +#[test] +fn runtime_v5_probe_and_staging_skip_erased_originals() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + let (worker, live_session, live_segment) = runtime_fixture(&root, 12); + let erased = root + .join("orphaned-workers") + .join(WorkerId::now_v7().to_string()); + let (session, segment) = write_v4_session(&erased); + runtime_session_retention(&root) + .delete( + session, + session_store::retention::SessionDeletionIntent::ErasePermanently, + || Ok(()), + ) + .unwrap(); + fs::write( + erased + .join("session/segments") + .join(format!("{segment}.jsonl")), + b"fenced malformed JSON\n", + ) + .unwrap(); + let before = tree_bytes(&erased); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); + assert!( + FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + FsRuntimeStore::migrate(&options).unwrap(); + assert_generic( + &root + .join("workers") + .join(worker.to_string()) + .join("session"), + live_session, + live_segment, + ); + assert_eq!(tree_bytes(&erased), before); + assert!( + !FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + assert!(matches!( + runtime_session_retention(&root).ensure_live(session), + Err(session_store::StoreError::SessionErased(_)) + )); +} + +#[test] +fn runtime_v5_rejects_broken_or_future_source_without_rewriting_evidence() { + for corruption in ["broken-json", "future-schema"] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + let (worker, session, segment) = runtime_fixture(&root, 12); + let session_root = root + .join("workers") + .join(worker.to_string()) + .join("session"); + if corruption == "broken-json" { + fs::write( + session_root + .join("segments") + .join(format!("{segment}.jsonl")), + b"{not valid JSON\n", + ) + .unwrap(); + } else { + write_json( + &session_root.join("session.json"), + &json!({ + "schema_version": 999, "session_id": session, + }), + ); + } + let before = tree_bytes(&root); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); + assert!( + FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + assert_eq!(tree_bytes(&root), before); + assert!(FsRuntimeStore::migrate(&options).is_err()); + assert_eq!(tree_bytes(&root), before); + assert!(!temp.path().join(".runtime.schema-v12-staging").exists()); + assert!(!temp.path().join(".runtime.pre-schema-v12-backup").exists()); + } +} + +#[cfg(unix)] +#[test] +fn current_runtime_session_migration_still_rejects_live_worker_sockets() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + let (worker, _, _) = runtime_fixture(&root, 12); + let before = tree_bytes(&root); + let runs = root.join("workers").join(worker.to_string()).join("runs"); + fs::create_dir_all(runs.join("1")).unwrap(); + let listener = std::os::unix::net::UnixListener::bind(runs.join("1/worker.sock")).unwrap(); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); + let error = FsRuntimeStore::migrate(&options).unwrap_err(); + assert!(error.to_string().contains("active Worker socket")); + drop(listener); + fs::remove_dir_all(runs).unwrap(); + assert_eq!(tree_bytes(&root), before); + assert!(!temp.path().join(".runtime.schema-v12-staging").exists()); +}