From b5a307811b609b0b532e01a75a3755e3ba726355 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 16 Oct 2026 04:58:53 +0900 Subject: [PATCH] fix: reconcile erased Archive removals and exercise live compaction recording Reconcile immutable Runtime removal input under a confirmed permanent erase fence. A confirmed source_removed receipt is audited without republishing erased archives; unknown receipt plus attested remaining aggregate permits an audited stale plan and proof-bound purge replacement. Unknown receipt and inventory remain unresolved. Distinguish domain tombstones from scoped Runtime physical erasure acknowledgements. Erase diagnostic originals and interrupted staging bound to the same archived Session/removal operation, validating Workspace/Worker/fingerprint and saved receipts without rewriting their historical retained flags. Finalize Backend removal only after physical acknowledgement, including previous Retain disposition. Connect public standalone protocol, real Worker request-threshold compaction twice within one turn, cancellation and real pre-AI input rollback, original canonical readers, existing Jobs and sequential Journal tools. Gate generation while compaction completes; verify all durable originals were served, summary/rolled-back input exclusion, exact final cursor, no extra input for successor drain, and one ongoing open entry. External model streams/usage are scripted, not semantic evaluation. Validation: Backend1112lib+11bin (existing ignore1), Runtime429lib+38bin, standalone72lib+27integration (existing model ignore1), runtime-api10, root check/doc/fmt/diff/naming, Server/Runtime builds, isolated worker_environment2 passed. Earlier broad suites remain recorded on the prior source. Fixtures and injected boundaries are not arbitrary crash/remote-transport-loss coverage; worker_environment only verifies startup/connection/cleanup. Acceptance clarification: T-734 assignment answer event22 explicitly defers real-model semantic condition6 to user use, excluding it from this completion scope. It remains unexecuted and unevaluated, not passed. Technical review findings still require a fresh independent verdict on this source. --- crates/runtime-api/src/lib.rs | 3 + .../standalone/src/subjektiv/journal/tests.rs | 13 + .../src/subjektiv/journal/tests/compaction.rs | 429 +++++++++++++++ .../tests/session_material_tests.rs | 102 +++- crates/worker-runtime/src/retention.rs | 92 +++- crates/workspace-server/src/retention.rs | 201 ++++++- crates/workspace-server/src/server.rs | 99 +++- .../src/server/subjektiv_journal.rs | 45 ++ .../server/tests/subjektiv_journal_tests.rs | 500 ++++++++++++++++++ crates/workspace-server/src/store.rs | 65 +++ 10 files changed, 1536 insertions(+), 13 deletions(-) create mode 100644 crates/standalone/src/subjektiv/journal/tests/compaction.rs diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index 6ae183d6a..d62dd7704 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -812,6 +812,9 @@ pub struct SessionMaterialEraseRequest { pub expected_session_id: String, } +/// Acknowledges the permanent fence and physical erasure of Session originals, +/// including archived diagnostics bound to the same removal operations. Historical +/// retention receipts remain immutable; their old retained flags are not current availability. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct SessionMaterialEraseResponse { pub session_id: String, diff --git a/crates/standalone/src/subjektiv/journal/tests.rs b/crates/standalone/src/subjektiv/journal/tests.rs index 5cdf832d3..6ddcaa6bc 100644 --- a/crates/standalone/src/subjektiv/journal/tests.rs +++ b/crates/standalone/src/subjektiv/journal/tests.rs @@ -7,6 +7,8 @@ use crate::store::{StaleLeasePolicy, StandaloneWorkerAllocation, StandaloneWorke use session_store::{Store, WorkerMetadataStore}; use std::sync::atomic::AtomicBool; +mod compaction; + struct Fixture { _temp: tempfile::TempDir, _allocation: Option, @@ -494,6 +496,7 @@ struct JournalScript { gate_first: bool, pause_after_record: bool, record_saved: Arc, + source_reads: Arc>>, } impl JournalScript { fn response( @@ -565,6 +568,16 @@ impl JournalScript { let id = format!("source-{index}-{offset}"); if let Some(page) = result(&id) { let page = &page["value"]; + let key = format!("{}:{id}", state["generation_id"].as_str().unwrap()); + let text = page["text"].as_str().unwrap().to_owned(); + if let Some(previous) = + self.source_reads.lock().unwrap().insert(key, text.clone()) + { + assert_eq!( + previous, text, + "frozen source read changed within a generation" + ); + } if let Some(next) = page["next_byte"].as_u64() { assert!(next > offset); offset = next; diff --git a/crates/standalone/src/subjektiv/journal/tests/compaction.rs b/crates/standalone/src/subjektiv/journal/tests/compaction.rs new file mode 100644 index 000000000..096a4d4c3 --- /dev/null +++ b/crates/standalone/src/subjektiv/journal/tests/compaction.rs @@ -0,0 +1,429 @@ +//! Public standalone protocol through real mid-turn compaction and Journal Jobs. +//! Only external model streams are scripted. No history, attribution, hook or Job +//! result is injected; cancellation and rollback use the controller's public commands. +//! Reported model usage and a low configured request threshold force two rewrites. +//! The test attests durable public originals, not unfinished display deltas or meaning. + +use super::*; +use agen::llm_client::{ + ClientError, LlmClient, Request, + event::{Event, ResponseStatus, StatusEvent, StopReason, UsageEvent}, + types::Item, +}; +use futures::{Stream, StreamExt}; +use std::pin::Pin; +use std::sync::atomic::AtomicUsize; + +type ResponseStream = Pin> + Send>>; +const SUMMARY: &str = "DERIVED_COMPACTION_SENTINEL_NOT_AN_ORIGINAL"; +const ROLLED_BACK: &str = "ROLLED_BACK_INPUT_MUST_NOT_REACH_JOURNAL"; +const PARTIAL: &str = "public partial answer before cancellation"; + +#[derive(Clone, Default)] +struct CompactingDialog { + shared: DialogAndJournal, + dialog_calls: Arc, + compactions: Arc, + rollback_entered: Arc, + partial_stream_paused: Arc, +} + +fn events(events: Vec) -> ResponseStream { + Box::pin(futures::stream::iter(events.into_iter().map(Ok))) +} +fn done() -> ResponseStream { + events(vec![ + Event::text_block_start(0), + Event::text_delta(0, "done"), + Event::text_block_stop(0, Some(StopReason::EndTurn)), + ]) +} + +#[async_trait::async_trait] +impl LlmClient for CompactingDialog { + async fn stream(&self, request: Request) -> Result { + if request + .tools + .iter() + .any(|tool| tool.name == "write_summary") + { + if request.items.iter().any(|item| matches!(item, Item::ToolResult { call_id, .. } if call_id == "compact-summary")) { + return Ok(done()); + } + let count = self.compactions.fetch_add(1, Ordering::SeqCst); + assert!(count < 2, "unexpected additional compaction"); + return Ok(events(vec![ + Event::tool_use_start(0, "compact-summary", "write_summary"), + Event::tool_input_delta(0, serde_json::json!({"text": SUMMARY}).to_string()), + Event::tool_use_stop(0), + Event::Status(StatusEvent { + status: ResponseStatus::Completed, + }), + ])); + } + if request.tools.iter().any(|tool| { + matches!( + tool.name.as_str(), + "ReadJournalState" | "SubmitMemorySurface" | "SubmitJobResult" + ) + }) { + return self.shared.stream(request).await; + } + match self.dialog_calls.fetch_add(1, Ordering::SeqCst) { + step @ (0 | 1) => { + assert!(request.tools.iter().any(|tool| tool.name == "TaskCreate")); + Ok(events(vec![ + Event::tool_use_start(0, format!("public-task-{step}"), "TaskCreate"), + Event::tool_input_delta( + 0, + serde_json::json!({ + "subject": format!("original activity {step}"), + "description": format!("original task result before compaction {step}") + }) + .to_string(), + ), + Event::tool_use_stop(0), + Event::Usage(UsageEvent { + input_tokens: Some(100_000), + output_tokens: Some(10), + total_tokens: Some(100_010), + ..Default::default() + }), + Event::Status(StatusEvent { + status: ResponseStatus::Completed, + }), + ])) + } + 2 => { + let paused = self.partial_stream_paused.clone(); + Ok(Box::pin( + events(vec![ + Event::text_block_start(0), + Event::text_delta(0, PARTIAL), + Event::text_block_stop(0, None), + ]) + .chain(futures::stream::once(async move { + paused.notify_one(); + std::future::pending::>().await + })), + )) + } + 3 => { + // No AI event: cancelling this request must roll back the submitted input. + self.rollback_entered.notify_one(); + std::future::pending::<()>().await; + unreachable!() + } + other => panic!("unexpected dialog request {other}"), + } + } + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } +} + +async fn cancel_and_wait( + client: &mut client::Client, + sequence: u64, +) { + client + .send(&protocol::Method::Cancel { + command: protocol::WorkerCommandEnvelope::new(sequence), + }) + .await + .unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(20), async { + loop { + match client.next_event().await.unwrap().unwrap() { + protocol::Event::RunEnd { result } => { + assert_ne!(result, protocol::RunResult::Finished); + return; + } + protocol::Event::Error { code, message } => panic!("{code:?}: {message}"), + _ => {} + } + } + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn public_mid_turn_compactions_preserve_originals_and_drain_after_cancel_and_rollback() { + tokio::time::timeout(std::time::Duration::from_secs(60), async { + let temp = tempfile::tempdir().unwrap(); + let state = temp.path().join("state"); + let catalog = StandaloneSubjects::open(&state).unwrap(); + let subject = catalog.create("compaction-journal", None).unwrap(); + let model = CompactingDialog { + shared: DialogAndJournal { + journal: JournalScript { + gate_first: true, + ..Default::default() + }, + ..Default::default() + }, + ..Default::default() + }; + let mut launch = crate::StandaloneLaunchConfig::new( + temp.path(), + &state, + manifest::ProfileSelector::source_named( + manifest::ProfileRegistrySource::Builtin, + "standalone-subjektiv", + ), + "public-compaction-journal", + ) + .with_subject(subject.id.clone()) + .resolve() + .unwrap(); + launch.profile.manifest.compaction = Some(manifest::CompactionConfig { + threshold: None, + request_threshold: Some(50_000), + retained_tokens: 0, + ..Default::default() + }); + let host = crate::StandaloneHost::start_with_model_client(launch, model.clone()) + .await + .unwrap(); + let session = host.record().active_session_id; + let workers = StandaloneWorkerStore::open_existing(&state).unwrap(); + let source_store = session_store::FsStore::new_with_retention( + workers.sessions_dir(host.record().worker_id), + crate::store::session_retention(&state), + ) + .unwrap(); + let mut client = host.connect(); + client + .send(&protocol::Method::submit_text( + protocol::new_submission_request_id(), + "original long turn survives compaction", + )) + .await + .unwrap(); + let mut completed_compactions = 0; + loop { + match client.next_event().await.unwrap().unwrap() { + protocol::Event::CompactionProgress { + compaction: Some(progress), + } if progress.phase == protocol::CompactionPhase::Committing => { + assert_eq!( + progress.trigger, + protocol::CompactionTrigger::RequestThreshold + ); + completed_compactions += 1; + } + protocol::Event::TextDelta { text } if text.contains(PARTIAL) => break, + protocol::Event::RunEnd { .. } => panic!("dialog ended before cancellation"), + protocol::Event::Error { code, message } => panic!("{code:?}: {message}"), + _ => {} + } + } + assert_eq!(model.compactions.load(Ordering::SeqCst), 2); + assert_eq!(completed_compactions, 2); + model.shared.journal.entered.notified().await; + let context = JournalHostContext::validated_session( + &catalog.store, + catalog + .store + .session_attribution(&session.to_string()) + .unwrap() + .unwrap(), + ) + .unwrap(); + let reader = JournalHostContext::validated_reader(&catalog.store, &subject.id).unwrap(); + let journal = catalog + .store + .list_journals(&reader, None, 1) + .unwrap() + .items + .pop() + .unwrap(); + let first = catalog + .store + .active_journal_generation(&context, &journal.id) + .unwrap() + .unwrap(); + assert!( + catalog + .store + .list_journal_entries(&reader, &journal.id, None, 10) + .unwrap() + .items + .is_empty(), + "compaction must finish before generation is released" + ); + model.partial_stream_paused.notified().await; + cancel_and_wait(&mut client, 1).await; + client + .send(&protocol::Method::submit_text( + protocol::new_submission_request_id(), + ROLLED_BACK, + )) + .await + .unwrap(); + model.rollback_entered.notified().await; + let before_rollback: Vec<_> = source_store + .list_segments(session) + .unwrap() + .iter() + .flat_map(|segment| source_store.read_all(session, *segment).unwrap()) + .collect(); + assert!( + serde_json::to_string(&before_rollback) + .unwrap() + .contains(ROLLED_BACK), + "the submitted input must reach the actual durable rollback boundary" + ); + cancel_and_wait(&mut client, 2).await; + assert_eq!( + catalog + .store + .active_journal_generation(&context, &journal.id) + .unwrap() + .unwrap(), + first, + "compaction/cancellation must not alter the running generation's frozen input" + ); + + assert!(matches!(source_store.delete_session(session), + Err(session_store::StoreError::SessionPinned(id)) if id == session)); + let segments = source_store.list_segments(session).unwrap(); + assert_eq!(segments.len(), 3); + let raw: Vec<_> = segments + .iter() + .flat_map(|segment| source_store.read_all(session, *segment).unwrap()) + .collect(); + assert_eq!( + raw.iter() + .filter(|entry| matches!( + entry, + session_store::LogEntry::AnnotatedSegmentStart { + compacted_from: Some(_), + .. + } + )) + .count(), + 2 + ); + assert!( + !serde_json::to_string(&raw).unwrap().contains(ROLLED_BACK), + "pre-AI cancellation must roll back the speculative input" + ); + assert!( + serde_json::to_string(&raw).unwrap().contains(SUMMARY), + "real rewrite did not persist its summary" + ); + let metadata = + session_store::FsWorkerStore::new(workers.worker_metadata_dir(host.record().worker_id)) + .unwrap() + .read_by_name(&host.record().storage_key) + .unwrap() + .unwrap(); + let originals = + session_store::public_material::SessionPublicMaterialReader::open_fs_with_retention( + &workers.sessions_dir(host.record().worker_id), + session, + metadata.active.unwrap().segment_id.unwrap(), + session_store::public_material::PublicMaterialLimits::default(), + crate::store::session_retention(&state), + ) + .unwrap(); + let final_original = originals.descriptors().last().unwrap(); + let expected_cursor = (final_original.canonical_order + 1, 0); + let canonical: Vec<_> = originals + .descriptors() + .map(|descriptor| { + let page = originals + .read_content(descriptor, 0, MAX_JOURNAL_READ_BYTES) + .unwrap(); + assert_eq!(page.next_offset, descriptor.total_bytes); + page.content + }) + .collect(); + assert!(!canonical.join("\n").contains(ROLLED_BACK)); + assert!(!canonical.join("\n").contains(SUMMARY)); + + // Resume the existing runner, then its one durable successor, with no more input. + let jobs = host.jobs(); + model.shared.journal.release.notify_one(); + jobs.wait(&first.job.job_id).await.unwrap(); + model.shared.journal.entered.notified().await; + let second = catalog + .store + .active_journal_generation(&context, &journal.id) + .unwrap() + .unwrap(); + assert_ne!(first.id, second.id); + model.shared.journal.release.notify_one(); + jobs.wait(&second.job.job_id).await.unwrap(); + assert!(!catalog.store.journal_retention_required(&context).unwrap()); + let (saved, open) = catalog + .store + .journal_resume_state(&context, &journal.id) + .unwrap(); + let open = open.unwrap(); + assert!(!open.ongoing.is_empty()); + let entries = catalog + .store + .list_journal_entries(&reader, &journal.id, None, 10) + .unwrap(); + assert_eq!( + entries.items.len(), + 1, + "compaction/turn boundaries must not fix or duplicate the open entry" + ); + assert_eq!(entries.items[0].id, open.id); + let supplied = model + .shared + .journal + .source_reads + .lock() + .unwrap() + .values() + .cloned() + .collect::>() + .join("\n"); + for original in [ + "original long turn survives compaction", + "original activity 0", + "original activity 1", + ] { + assert!( + supplied.contains(original), + "Journal did not receive {original}" + ); + } + for original in canonical { + assert!( + supplied.contains(&original), + "a durable public original was skipped" + ); + } + assert!( + !supplied.contains(SUMMARY), + "derived summary substituted for original input" + ); + assert!( + !supplied.contains(ROLLED_BACK), + "cancelled input was admitted after rollback" + ); + assert_eq!((saved.process_order, saved.process_byte), expected_cursor); + let position = catalog + .store + .read_journal_text( + &reader, + &journal.id, + None, + JournalTextField::CurrentPosition, + None, + 0, + MAX_JOURNAL_READ_BYTES, + ) + .unwrap(); + assert!(!position.text.is_empty()); + host.shutdown().await.unwrap(); + }) + .await + .expect("bounded public compaction/Journal integration"); +} diff --git a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs index a5d06c6ba..b7bb8513b 100644 --- a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs @@ -659,8 +659,38 @@ fn catalog_rejects_oversize_record_with_explicit_resource_limit() { fn archive_scope_and_legacy_active_pointer_are_checked_and_erase_retry_never_restores_originals() { let f = Fixture::new(); f.stop(); - let request = f.retention_request(SessionDisposition::Archive); - f.runtime.execute_worker_retention(&request).unwrap(); + let mut request = f.retention_request(SessionDisposition::Archive); + request.diagnostics_disposition = DiagnosticsDisposition::Retain; + let log = f + .root() + .parent() + .unwrap() + .join("runs/run-original/worker.out.log"); + std::fs::create_dir_all(log.parent().unwrap()).unwrap(); + std::fs::write(&log, "archived diagnostic original").unwrap(); + let original_receipt = f.runtime.execute_worker_retention(&request).unwrap(); + assert!(original_receipt.diagnostics_retained); + let diagnostics = f + .temp + .path() + .join("archives/diagnostics") + .join(&request.operation_id); + assert!( + diagnostics + .join("runs/run-original/worker.out.log") + .is_file() + ); + let staged_diagnostics = + diagnostics.with_file_name(format!(".{}.tmp-42-1", request.operation_id)); + std::fs::create_dir_all(&staged_diagnostics).unwrap(); + std::fs::write( + staged_diagnostics.join("partial-original.log"), + "incomplete copy", + ) + .unwrap(); + let unrelated = diagnostics.with_file_name("another-operation"); + std::fs::create_dir_all(&unrelated).unwrap(); + std::fs::write(unrelated.join("untouched.log"), "unrelated diagnostic").unwrap(); let source = Source::Archived { archive_id: "canonical-archive".into(), }; @@ -756,12 +786,76 @@ fn archive_scope_and_legacy_active_pointer_are_checked_and_erase_retry_never_res .pin_session_public_source_scoped(&f.scope, &source, &f.session.to_string(), pin) .is_err() ); - // An old removal receipt is evidence only and must not republish erased bytes. - f.runtime.execute_worker_retention(&request).unwrap(); + // An old receipt is historical evidence, not permission to republish erased bytes. + assert_eq!( + f.runtime.execute_worker_retention(&request).unwrap(), + original_receipt + ); + assert!(!diagnostics.exists()); + assert!(!staged_diagnostics.exists()); + assert!(unrelated.join("untouched.log").is_file()); assert!(!root.exists()); assert!(retention.pins(f.session).unwrap().is_empty()); } +#[test] +fn erasure_rejects_mismatched_archived_diagnostics_then_resumes_under_the_same_fence() { + let f = Fixture::new(); + f.stop(); + let mut request = f.retention_request(SessionDisposition::Archive); + request.diagnostics_disposition = DiagnosticsDisposition::Retain; + let receipt = f.runtime.execute_worker_retention(&request).unwrap(); + let source = Source::Archived { + archive_id: request.archive_id.clone().unwrap(), + }; + let manifest = f + .temp + .path() + .join("archives/diagnostics") + .join(&request.operation_id) + .join("manifest.json"); + let original: serde_json::Value = + serde_json::from_slice(&std::fs::read(&manifest).unwrap()).unwrap(); + for field in ["workspace_id", "input_fingerprint", "source_worker_id"] { + let mut changed = original.clone(); + changed[field] = serde_json::json!(uuid::Uuid::now_v7().to_string()); + std::fs::write(&manifest, serde_json::to_vec(&changed).unwrap()).unwrap(); + assert!( + f.runtime + .delete_session_public_source_scoped( + &f.scope, + &source, + &f.session.to_string(), + SessionDeletionIntent::ErasePermanently + ) + .is_err() + ); + assert!(manifest.exists()); + assert!(matches!( + runtime_session_retention(f.temp.path()).ensure_live(f.session), + Err(session_store::StoreError::SessionErased(_)) + )); + } + std::fs::write(&manifest, serde_json::to_vec(&original).unwrap()).unwrap(); + f.runtime + .delete_session_public_source_scoped( + &f.scope, + &source, + &f.session.to_string(), + SessionDeletionIntent::ErasePermanently, + ) + .unwrap(); + assert!(!manifest.parent().unwrap().exists()); + assert_eq!( + f.runtime.execute_worker_retention(&request).unwrap(), + receipt + ); + assert!( + receipt.diagnostics_retained, + "historical receipt must not be rewritten after erasure" + ); +} + #[test] fn material_request_bounds_return_explicit_unavailable_without_previews() { let f = Fixture::new(); diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index a9f5c1cce..4e6f19427 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -297,14 +297,15 @@ pub(crate) fn remove_archived_session_copies( } let session_root = entry.path().join("session"); let owner = match fs::read(entry.path().join("manifest.json")) { - Ok(bytes) => Some( - serde_json::from_slice::(&bytes)?.source_session_id, - ), + Ok(bytes) => Some(serde_json::from_slice::( + &bytes, + )?), Err(error) if error.kind() == std::io::ErrorKind::NotFound => None, Err(error) => return Err(error.into()), }; if let Some(owner) = owner { - if owner == session.to_string() { + if owner.source_session_id == session.to_string() { + erase_archived_diagnostics(runtime_root, &owner)?; match fs::remove_dir_all(&session_root) { Ok(()) => (), Err(error) if error.kind() == std::io::ErrorKind::NotFound => (), @@ -326,6 +327,89 @@ pub(crate) fn remove_archived_session_copies( Ok(()) } +/// Erasure includes diagnostic originals of the same archived removal operation. +/// The immutable removal receipt is historical evidence and is never rewritten. +/// Staging names are bound to that exact operation; unrelated archives are untouched. +fn erase_archived_diagnostics( + runtime_root: &Path, + archive: &WorkerSessionArchiveManifest, +) -> Result<(), session_store::StoreError> { + let invalid = || { + session_store::StoreError::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Archived diagnostics erasure identity mismatch", + )) + }; + if archive.schema_version != ARCHIVE_SCHEMA_VERSION { + return Err(invalid()); + } + validate_id("operation_id", &archive.operation_id).map_err(|_| invalid())?; + let receipt_path = runtime_root + .join("retention/operations") + .join(format!("{}.json", archive.operation_id)); + if let Some(receipt) = read_operation_receipt(&receipt_path).map_err(|_| invalid())? { + if receipt.request.workspace_id != archive.workspace_id + || receipt.request.source_runtime_id != archive.source_runtime_id + || receipt.request.worker_id != archive.source_worker_id + || receipt.request.operation_id != archive.operation_id + || receipt.request.input_fingerprint != archive.input_fingerprint + || receipt.request.archive_id.as_ref() != Some(&archive.archive_id) + || receipt.result.archive.as_ref() != Some(archive) + { + return Err(invalid()); + } + } + let parent = runtime_root.join("archives/diagnostics"); + let entries = match fs::read_dir(&parent) { + Ok(entries) => entries, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error.into()), + }; + let staging_prefix = format!(".{}.tmp-", archive.operation_id); + for entry in entries { + let entry = entry?; + let name = entry.file_name(); + let Some(name) = name.to_str() else { + continue; + }; + let staged = name.strip_prefix(&staging_prefix).is_some_and(|suffix| { + suffix.split_once('-').is_some_and(|(pid, sequence)| { + !pid.is_empty() + && !sequence.is_empty() + && pid.bytes().all(|b| b.is_ascii_digit()) + && sequence.bytes().all(|b| b.is_ascii_digit()) + }) + }); + if name != archive.operation_id && !staged { + continue; + } + if !entry.file_type()?.is_dir() { + return Err(invalid()); + } + let manifest = match fs::read(entry.path().join("manifest.json")) { + Ok(bytes) => Some(serde_json::from_slice::( + &bytes, + )?), + Err(error) if error.kind() == std::io::ErrorKind::NotFound && staged => None, + Err(error) => return Err(error.into()), + }; + if let Some(manifest) = manifest { + if manifest.schema_version != ARCHIVE_SCHEMA_VERSION + || manifest.workspace_id != archive.workspace_id + || manifest.source_runtime_id != archive.source_runtime_id + || manifest.source_worker_id != archive.source_worker_id + || manifest.operation_id != archive.operation_id + || manifest.input_fingerprint != archive.input_fingerprint + { + return Err(invalid()); + } + } + fs::remove_dir_all(entry.path())?; + } + sync_directory(&parent, "erase archived Worker diagnostics").map_err(|_| invalid())?; + Ok(()) +} + pub(crate) fn retention_error(error: session_store::StoreError) -> RuntimeError { RuntimeError::InvalidRequest(format!("Session retention rejected operation: {error}")) } diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index 655b75ca0..415f444e9 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -110,6 +110,50 @@ impl ErasedJournalSession { } } +/// Physical Runtime erasure acknowledgement, not merely a domain tombstone. +/// Only the scoped erase request and a matching Session response construct it. +/// The Runtime acknowledgement includes associated archived diagnostic originals; +/// historical retention receipts retain their original values. +#[derive(Clone, Debug)] +pub struct RuntimeJournalErasureAck { + domain: ErasedJournalSession, + source: runtime_api::SessionPublicSource, +} + +impl ErasedJournalSession { + pub(crate) fn erase_runtime_source( + self, + runtime: &crate::hosts::RuntimeRegistry, + source: runtime_api::SessionPublicSource, + ) -> crate::Result { + let response = runtime + .session_material_erase( + &self.worker.runtime_id, + runtime_api::SessionMaterialEraseRequest { + workspace_id: self.workspace_id.clone(), + source: source.clone(), + expected_session_id: self.session_id.clone(), + }, + ) + .map_err(|e| e.into_error())?; + if response.session_id != self.session_id { + return Err(StoreError::RepositoryConflict( + "Session erase receipt differs from the requested identity".into(), + )); + } + Ok(RuntimeJournalErasureAck { + domain: self, + source, + }) + } +} + +impl RuntimeJournalErasureAck { + pub(crate) fn domain_proof(&self) -> &ErasedJournalSession { + &self.domain + } +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum WorkerRemovalBlocker { @@ -490,12 +534,125 @@ impl SqliteWorkspaceStore { Ok(()) } + /// An unknown acceptance is not failure. Supersede only after physical erasure + /// fenced Archive effects and Runtime attested the same remaining aggregate. + pub(crate) fn supersede_erased_worker_removal( + &self, + prepared: &PreparedWorkerRemoval, + ack: &RuntimeJournalErasureAck, + inventory: &WorkerRetentionInventory, + ) -> Result<(), WorkerRetentionError> { + let proof = ack.domain_proof(); + if proof.workspace_id != prepared.plan.workspace_id + || proof.worker != prepared.plan.worker + || prepared.plan.session_disposition != SessionDisposition::Archive + || inventory.workspace_id != proof.workspace_id + || inventory.runtime_id != proof.worker.runtime_id + || inventory.worker_id.to_string() != proof.worker.worker_id + || inventory + .session_id + .as_ref() + .is_some_and(|id| id != &proof.session_id) + { + return Err(WorkerRetentionError::Invalid( + "Erasure reconciliation source mismatch".into(), + )); + } + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let plan = load_plan_op(&tx, &prepared.plan.operation_id)? + .ok_or_else(|| StoreError::InvalidInput("operation missing".into()))?; + if plan != prepared.plan + || !matches!( + plan.state, + WorkerRemovalPlanState::Planned + | WorkerRemovalPlanState::Executing + | WorkerRemovalPlanState::Failed + ) + { + return Err(StoreError::InvalidInput("active operation mismatch".into())); + } + insert_erasure_reconciliation_audit(&tx, &plan, ack, None)?; + mark_stale(&tx, &plan, "journal_session_erased_archive_fenced")?; + tx.commit()?; + Ok(()) + })?; + Ok(()) + } + + pub(crate) fn audit_erased_worker_removal_receipt( + &self, + prepared: &PreparedWorkerRemoval, + ack: &RuntimeJournalErasureAck, + result: &WorkerRetentionExecutionResult, + ) -> Result<(), WorkerRetentionError> { + let proof = ack.domain_proof(); + if proof.workspace_id != prepared.plan.workspace_id + || proof.worker != prepared.plan.worker + || result.operation_id != prepared.plan.operation_id + || result.input_fingerprint != prepared.plan.input_fingerprint + || result.worker_id.to_string() != proof.worker.worker_id + || !result.source_removed + || result.session_disposition != prepared.plan.session_disposition + || result.diagnostics_disposition != prepared.plan.diagnostics_disposition + || result.archive.as_ref().is_none_or(|a| { + a.source_session_id != proof.session_id + || Some(&a.archive_id) != prepared.plan.archive_id.as_ref() + || a.workspace_id != proof.workspace_id + || a.source_runtime_id != proof.worker.runtime_id + || a.source_worker_id.to_string() != proof.worker.worker_id + || a.operation_id != result.operation_id + || a.input_fingerprint != result.input_fingerprint + }) + { + return Err(WorkerRetentionError::Invalid( + "Erasure receipt mismatch".into(), + )); + } + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + insert_erasure_reconciliation_audit(&tx, &prepared.plan, ack, Some(result))?; + tx.commit()?; + Ok(()) + })?; + Ok(()) + } + pub fn commit_worker_removal( &self, workspace_id: &str, operation_id: &str, fp: &str, result: &WorkerRetentionExecutionResult, + ) -> Result { + self.commit_worker_removal_inner(workspace_id, operation_id, fp, result, None) + } + + /// Receipt-confirmed removal after erasure must not republish the erased Archive. + /// The original operation, fingerprint and Runtime result remain unchanged. + pub(crate) fn commit_erased_worker_removal( + &self, + workspace_id: &str, + prepared: &PreparedWorkerRemoval, + result: &WorkerRetentionExecutionResult, + ack: &RuntimeJournalErasureAck, + ) -> Result { + self.commit_worker_removal_inner( + workspace_id, + &prepared.plan.operation_id, + &prepared.plan.input_fingerprint, + result, + Some(ack), + ) + } + + fn commit_worker_removal_inner( + &self, + workspace_id: &str, + operation_id: &str, + fp: &str, + result: &WorkerRetentionExecutionResult, + erasure: Option<&RuntimeJournalErasureAck>, ) -> Result { self.with_conn_mut(|conn|{ let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; @@ -519,6 +676,13 @@ impl SqliteWorkspaceStore { return Err(StoreError::InvalidInput("Runtime retention result does not match removal plan".into())); } if !result.source_removed{return Err(StoreError::InvalidInput("Runtime source was not removed".into()));} + if let Some(ack) = erasure { + let proof = ack.domain_proof(); + if proof.workspace_id != workspace_id || proof.worker != plan.worker + || result.archive.as_ref().is_none_or(|a| a.source_session_id != proof.session_id) { + return Err(StoreError::InvalidInput("Erased removal receipt requires matching source and physical Runtime erasure acknowledgement".into())); + } + } let worker=load_worker(&tx,workspace_id,&plan.worker)?.ok_or_else(||StoreError::InvalidInput("Worker missing before commit".into()))?; if worker.updated_at!=plan.worker_updated_at{return Err(StoreError::InvalidInput(format!("stale:{}:Worker updated_at changed",plan.plan_id)));} if worker.retention_state=="pinned" { return Err(StoreError::InvalidInput(format!("stale:{}:hold added",plan.plan_id))); } @@ -527,6 +691,7 @@ impl SqliteWorkspaceStore { let now=Utc::now().to_rfc3339(); if let Some(a)=&result.archive{ if plan.archive_id.as_deref()!=Some(&a.archive_id)||a.workspace_id!=workspace_id||a.source_runtime_id!=plan.worker.runtime_id||a.source_worker_id.to_string()!=plan.worker.worker_id||a.policy_id!=plan.policy_id||a.operation_id!=operation_id||a.input_fingerprint!=fp{return Err(StoreError::InvalidInput("archive manifest mismatch".into()));} + if erasure.is_none() { let expires_at=match plan.archive_retention { ArchiveRetention::Forever=>None, ArchiveRetention::ForSeconds{seconds}=>{let seconds=i64::try_from(seconds).map_err(|_|StoreError::InvalidInput("archive retention deadline overflow".into()))?;Some((Utc::now()+chrono::Duration::seconds(seconds)).to_rfc3339())} }; tx.execute("INSERT OR IGNORE INTO worker_session_archives(archive_id,workspace_id,runtime_id,worker_id,session_id,checksum_sha256,content_bytes,policy_id,policy_digest,operation_id,committed_at,expires_at) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12)",params![a.archive_id,workspace_id,plan.worker.runtime_id,plan.worker.worker_id,a.source_session_id,a.content_checksum_sha256,a.content_bytes,plan.policy_id,plan.policy_digest,operation_id,now,expires_at])?; let observe_grants = { @@ -574,20 +739,21 @@ impl SqliteWorkspaceStore { )?; } } + } }else if plan.session_disposition==SessionDisposition::Archive{return Err(StoreError::InvalidInput("archive manifest missing".into()));} match plan.diagnostics_disposition { DiagnosticsDisposition::Purge if result.diagnostics_retained => return Err(StoreError::InvalidInput("Runtime retained diagnostics for purge disposition".into())), DiagnosticsDisposition::Retain if !result.diagnostics_retained => return Err(StoreError::InvalidInput("Runtime did not retain diagnostics".into())), - DiagnosticsDisposition::Retain => { + DiagnosticsDisposition::Retain if erasure.is_none() => { let seconds=plan.diagnostics_retention_seconds.ok_or_else(||StoreError::InvalidInput("diagnostics retention deadline missing".into()))?; let seconds=i64::try_from(seconds).map_err(|_|StoreError::InvalidInput("diagnostics retention deadline overflow".into()))?; let expires_at=(Utc::now()+chrono::Duration::seconds(seconds)).to_rfc3339(); tx.execute("INSERT OR IGNORE INTO worker_diagnostics_archives(operation_id,workspace_id,runtime_id,worker_id,policy_id,policy_digest,committed_at,expires_at) VALUES(?1,?2,?3,?4,?5,?6,?7,?8)",params![operation_id,workspace_id,plan.worker.runtime_id,plan.worker.worker_id,plan.policy_id,plan.policy_digest,now,expires_at])?; } - DiagnosticsDisposition::Purge => {} + DiagnosticsDisposition::Purge | DiagnosticsDisposition::Retain => {} } if plan.metadata_disposition==MetadataDisposition::Tombstone{ - tx.execute("INSERT OR IGNORE INTO worker_tombstones(workspace_id,runtime_id,worker_id,display_name,profile,worker_created_at,removed_at,archive_id,policy_id,policy_digest,operation_id) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,worker.display_name,worker.profile,worker.created_at,now,plan.archive_id,plan.policy_id,plan.policy_digest,operation_id])?; + tx.execute("INSERT OR IGNORE INTO worker_tombstones(workspace_id,runtime_id,worker_id,display_name,profile,worker_created_at,removed_at,archive_id,policy_id,policy_digest,operation_id) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,worker.display_name,worker.profile,worker.created_at,now,if erasure.is_some() { None } else { plan.archive_id.clone() },plan.policy_id,plan.policy_digest,operation_id])?; } tx.execute("UPDATE worker_workdir_links SET unlinked_at=?4 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND unlinked_at IS NULL",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,now])?; let deleted=tx.execute("DELETE FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND updated_at=?4",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,plan.worker_updated_at])?; @@ -595,6 +761,9 @@ impl SqliteWorkspaceStore { if deleted!=1{return Err(StoreError::InvalidInput(format!("stale:{}:removal fence changed",plan.plan_id)));} let catalog=commit_worker_catalog_removal(&tx,workspace_id,&plan.worker)?; tx.execute("UPDATE worker_removal_operations SET state='succeeded',failure_category=NULL,updated_at=?1 WHERE operation_id=?2",params![now,operation_id])?; + if let Some(ack) = erasure { + insert_erasure_reconciliation_audit(&tx, &plan, ack, Some(result))?; + } tx.execute("INSERT OR IGNORE INTO worker_retention_audit_events(event_id,operation_id,workspace_id,event_kind,detail,created_at) VALUES(?1,?2,?3,'worker_removed',?4,?5)",params![stable("wre",operation_id),operation_id,workspace_id,format!("runtime_id={} worker_id={} session={} metadata={} diagnostics={}",plan.worker.runtime_id,plan.worker.worker_id,sess(plan.session_disposition),meta(plan.metadata_disposition),diag(plan.diagnostics_disposition)),now])?; tx.commit()?;plan.state=WorkerRemovalPlanState::Succeeded;plan.updated_at=now;Ok(WorkerRemovalCommit { plan, catalog }) }).map_err(map_error) @@ -969,6 +1138,32 @@ fn load_plan_q(c: &Connection, key: &str, id: &str) -> crate::Result, + plan: &WorkerRemovalPlan, + ack: &RuntimeJournalErasureAck, + result: Option<&WorkerRetentionExecutionResult>, +) -> crate::Result<()> { + let proof = ack.domain_proof(); + let kind = if result.is_some() { + "erased_removal_receipt" + } else { + "erased_archive_superseded" + }; + let detail = serde_json::json!({ + "input_fingerprint": plan.input_fingerprint, + "session_id": proof.session_id, + "archive_id": plan.archive_id, + "source_removed": result.map(|r| r.source_removed), + "old_diagnostics_retained": result.map(|r| r.diagnostics_retained), + "runtime_erasure_ack": { "session_id": proof.session_id, "source": ack.source }, + "old_acceptance": if result.is_some() { "receipt_confirmed" } else { "unknown_archive_fenced" }, + }); + tx.execute("INSERT OR IGNORE INTO worker_retention_audit_events(event_id,operation_id,workspace_id,event_kind,detail,created_at) VALUES(?1,?2,?3,?4,?5,?6)", + params![stable("wre", &format!("{}:{kind}", plan.operation_id)), plan.operation_id, plan.workspace_id, kind, detail.to_string(), Utc::now().to_rfc3339()])?; + Ok(()) +} + fn mark_stale( tx: &rusqlite::Transaction<'_>, plan: &WorkerRemovalPlan, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 10ad743c6..8cb1b01a9 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1889,8 +1889,98 @@ impl WorkerRemovalService { &self, runtime: &RuntimeRegistry, target: &RuntimeWorkerRef, - prepared: crate::retention::PreparedWorkerRemoval, + mut prepared: crate::retention::PreparedWorkerRemoval, + parent_workspace_deletion_operation_id: Option<&str>, ) -> std::result::Result { + if prepared.plan.session_disposition + == worker_runtime::retention::SessionDisposition::Archive + { + let erasure_ack = match subjektiv_journal::erased_archive_removal_proof( + self, runtime, target, &prepared, + ) { + Ok(ack) => ack, + Err(error) => { + return Ok(worker_remove_error_response( + StatusCode::CONFLICT, + "erased_removal_reconciliation_unconfirmed", + &error.to_string(), + )); + } + }; + if let Some(ack) = erasure_ack { + // Replay EXACT old input under the acknowledged permanent fence. + // Runtime can return an existing receipt, but cannot re-Archive. + match runtime.execute_worker_retention(target, prepared.runtime_request.clone()) { + Ok(result) => { + if let Err(error) = self + .store + .audit_erased_worker_removal_receipt(&prepared, &ack, &result) + { + return Ok(worker_retention_error_response(error)); + } + return match self.worker_projection.publish_ordered(|| { + self.store + .commit_erased_worker_removal( + &self.workspace_id, + &prepared, + &result, + &ack, + ) + .map(|commit| ((), commit.catalog)) + }) { + Ok(()) => Ok(worker_remove_success_response(target)), + Err(error) => Ok(worker_retention_error_response(error)), + }; + } + Err(_) => { + // Errors are unknown acceptance, not failure. A missing + // aggregate without a receipt remains unresolved. + let inventory = match runtime.worker_retention_inventory(target) { + Ok(inventory) => inventory, + Err(_) => { + return Ok(worker_remove_error_response( + StatusCode::SERVICE_UNAVAILABLE, + "erased_removal_outcome_unknown", + "Old Runtime outcome cannot be reconciled; removal remains recoverable", + )); + } + }; + if let Err(error) = self + .store + .supersede_erased_worker_removal(&prepared, &ack, &inventory) + { + return Ok(worker_retention_error_response(error)); + } + let request = crate::retention::WorkerRemovalPlanRequest { + workspace_id: self.workspace_id.clone(), + worker: target.clone(), + reason: prepared.plan.reason.clone(), + journal_erasure: Some(ack.domain_proof().clone()), + }; + let plan = match self.store.plan_worker_removal(&request, &inventory) { + Ok(plan) => plan, + Err(error) => return Ok(worker_retention_error_response(error)), + }; + prepared = match self.store.prepare_worker_removal_execution( + &self.workspace_id, + &plan.plan_id, + &plan.input_fingerprint, + ) { + Ok(prepared) => prepared, + Err(error) => return Ok(worker_retention_error_response(error)), + }; + if let Some(parent) = parent_workspace_deletion_operation_id { + self.store + .append_workspace_deletion_child_operation( + parent, + &prepared.plan.operation_id, + ) + .map_err(|error| error.to_string())?; + } + } + } + } + } if let Err(error) = subjektiv_journal::validate_removal_retention(self, target, &prepared) { return Ok(worker_remove_error_response( StatusCode::CONFLICT, @@ -2168,7 +2258,12 @@ impl WorkerRemovalService { )); } return self - .resume_worker_retention(runtime, &target, prepared) + .resume_worker_retention( + runtime, + &target, + prepared, + parent_workspace_deletion_operation_id, + ) .await; } diff --git a/crates/workspace-server/src/server/subjektiv_journal.rs b/crates/workspace-server/src/server/subjektiv_journal.rs index 6de0b3e46..0ee0a75af 100644 --- a/crates/workspace-server/src/server/subjektiv_journal.rs +++ b/crates/workspace-server/src/server/subjektiv_journal.rs @@ -878,6 +878,51 @@ pub(super) fn prepare_worker_removal( Ok(None) } +/// A successful erase response proves the Runtime's permanent fence. The reserved +/// archive covers transfer-before-Backend-commit crashes too. Never admit or generate. +pub(super) fn erased_archive_removal_proof( + service: &WorkerRemovalService, + runtime: &RuntimeRegistry, + target: &RuntimeWorkerRef, + prepared: &crate::retention::PreparedWorkerRemoval, +) -> Result> { + let Some(attr) = removal_attribution(service, target)? else { + return Ok(None); + }; + let store = crate::subjektiv::SubjektivStore::open( + &service.journal_storage, + &service.journal_registration, + ) + .map_err(|e| Error::Store(e.to_string()))?; + let Some(proof) = + crate::retention::ErasedJournalSession::validated(&service.workspace_id, &store, &attr)? + else { + return Ok(None); + }; + let api = service.journal_dispatch_api.upgrade().ok_or_else(|| { + Error::Store("Workspace API unavailable for erased removal recovery".into()) + })?; + reconcile_session_jobs(&api, &attr, true)?; + let source = runtime_api::SessionPublicSource::Retained { + worker_id: prepared.runtime_request.worker_id, + }; + let ack = match proof.clone().erase_runtime_source(runtime, source) { + Ok(ack) => ack, + Err(original) => { + let Some(archive_id) = prepared.runtime_request.archive_id.clone() else { + return Err(original); + }; + proof + .erase_runtime_source( + runtime, + runtime_api::SessionPublicSource::Archived { archive_id }, + ) + .map_err(|_| original)? + } + }; + Ok(Some(ack)) +} + pub(super) fn validate_removal_retention( service: &WorkerRemovalService, target: &RuntimeWorkerRef, diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs index 061c29032..8b1bb7d33 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs @@ -2141,3 +2141,503 @@ async fn serialized_removal_input_cannot_supply_journal_erasure_authority() { assert!(!f.store.journal_session_erased(&f.host).unwrap()); assert!(session_root(f.dir.path(), &f.target).exists()); } + +impl SubjectRemovalFixture { + async fn reopen(self) -> Self { + let Self { + dir, + api, + dispatcher, + controller, + target, + attr, + host, + store, + } = self; + let config = api.config.clone(); + drop(api); + drop(dispatcher); + drop(store); + let control = Arc::new(test_control_store(&config)); + let dispatcher = Arc::new( + crate::worker_source::EmbeddedServerWorkerMutationDispatcher::new( + config.clone(), + control.clone(), + ), + ); + let api = WorkspaceApi::new_with_execution_backend_and_broker( + config, + control, + Arc::new(DeterministicExecutionBackend::default()), + BackendResourceBroker::default(), + HostExecutors::default(), + Some(dispatcher.clone()), + ) + .await + .unwrap(); + let store = open_subjektiv_store(&api).unwrap(); + Self { + dir, + api, + dispatcher, + controller, + target, + attr, + host, + store, + } + } + + fn assert_erased_removal(&self) { + assert!( + self.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &self.target) + .unwrap() + .is_none() + ); + assert!( + self.api + .config_store + .worker_session_archive(TEST_WORKSPACE_ID, &self.target, &self.attr.session_id) + .unwrap() + .is_none() + ); + assert!( + self.api + .store + .list_backend_journal_jobs(TEST_WORKSPACE_ID, &self.attr.session_id, None, 100) + .unwrap() + .is_empty() + ); + assert!(!session_root(self.dir.path(), &self.target).exists()); + assert!(matches!( + source_retention(self.dir.path()).ensure_live(self.attr.session_id.parse().unwrap()), + Err(session_store::StoreError::SessionErased(_)) + )); + } +} + +#[tokio::test] +async fn erased_archive_plan_resend_and_reopen_replan_without_republishing_sources() { + for stage in ["planned", "executing", "failed"] { + for restart in [false, true] { + let mut f = SubjectRemovalFixture::new().await; + f.disable(); + let policy = f + .api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap(); + let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); + let archive_parent = f + .dir + .path() + .join(".test-embedded-runtime-store/archives/workers"); + if stage == "planned" { + conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE UPDATE OF state ON worker_removal_operations WHEN NEW.state='executing' BEGIN SELECT RAISE(ABORT,'injected execution interruption'); END;").unwrap(); + } else { + // Fail Archive publication before any receipt. Suppress recording + // failure too to model the executing-at-Host-stop durable boundary. + std::fs::create_dir_all(archive_parent.parent().unwrap()).unwrap(); + std::fs::write(&archive_parent, "publication blocked").unwrap(); + if stage == "executing" { + conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE UPDATE OF state ON worker_removal_operations WHEN NEW.state='failed' BEGIN SELECT RAISE(ABORT,'injected failure-save interruption'); END;").unwrap(); + } + } + assert_ne!(f.remove().status, StatusCode::OK.as_u16()); + conn.execute_batch("DROP TRIGGER IF EXISTS interrupt_removal") + .unwrap(); + if stage != "planned" { + std::fs::remove_file(&archive_parent).unwrap(); + } + let old = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(serde_json::to_value(old.plan.state).unwrap(), stage); + assert_eq!( + old.plan.session_disposition, + worker_runtime::retention::SessionDisposition::Archive + ); + f.api + .erase_subjektiv_journal_session( + &f.attr.session_id, + "erase-interrupted-archive-plan", + ) + .unwrap(); + if restart { + f = f.reopen().await; + } + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{stage}/{restart}: {}", + response.body + ); + let saved = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!( + saved.plan.session_disposition, + worker_runtime::retention::SessionDisposition::Purge + ); + assert_eq!(saved.plan.reason, old.plan.reason); + assert_ne!(saved.plan.operation_id, old.plan.operation_id); + let (state, fingerprint, detail): (String, String, String) = conn.query_row( + "SELECT o.state,o.input_fingerprint,a.detail FROM worker_removal_operations o JOIN worker_retention_audit_events a USING(operation_id) WHERE o.operation_id=?1 AND a.event_kind='erased_archive_superseded'", + [&old.plan.operation_id], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?))).unwrap(); + assert_eq!(state, "stale"); + assert_eq!(fingerprint, old.plan.input_fingerprint); + let detail: Value = serde_json::from_str(&detail).unwrap(); + assert_eq!(detail["old_acceptance"], "unknown_archive_fenced"); + assert!(detail["source_removed"].is_null()); + f.assert_erased_removal(); + assert_eq!( + f.api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap(), + policy + ); + } + } +} + +#[tokio::test] +async fn erased_completed_archive_receipt_is_audited_without_republishing_on_resend_or_reopen() { + for restart in [false, true] { + let mut f = SubjectRemovalFixture::new().await; + f.disable(); + let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); + // Runtime has committed source_removed, but Backend commit/response is lost. + conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'injected metadata commit interruption'); END;").unwrap(); + assert_ne!(f.remove().status, StatusCode::OK.as_u16()); + conn.execute_batch("DROP TRIGGER interrupt_removal") + .unwrap(); + let old = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!( + old.plan.state, + crate::retention::WorkerRemovalPlanState::Failed + ); + let receipt = f + .api + .runtime + .execute_worker_retention(&f.target, old.runtime_request.clone()) + .unwrap(); + assert!(receipt.source_removed); + let archive_id = old.plan.archive_id.as_ref().unwrap(); + let archive_source = f + .dir + .path() + .join(".test-embedded-runtime-store/archives/workers") + .join(archive_id) + .join("session"); + assert!(archive_source.exists()); + // The domain tombstone commits even though retained-source resolution fails. + // Recovery must locate only this plan's exact reserved archive, not publish it. + assert!( + f.api + .erase_subjektiv_journal_session(&f.attr.session_id, "erase-transferred-original") + .is_err() + ); + assert!(f.store.journal_session_erased(&f.host).unwrap()); + if restart { + f = f.reopen().await; + } + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "{restart}: {}", + response.body + ); + let saved = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(saved.plan.operation_id, old.plan.operation_id); + assert_eq!(saved.plan.input_fingerprint, old.plan.input_fingerprint); + assert_eq!(saved.plan.session_disposition, old.plan.session_disposition); + let detail: String = conn.query_row("SELECT detail FROM worker_retention_audit_events WHERE operation_id=?1 AND event_kind='erased_removal_receipt'", [&old.plan.operation_id], |r| r.get(0)).unwrap(); + let detail: Value = serde_json::from_str(&detail).unwrap(); + assert_eq!(detail["source_removed"], true); + assert_eq!(detail["old_acceptance"], "receipt_confirmed"); + assert!(!archive_source.exists()); + f.assert_erased_removal(); + } +} + +#[tokio::test] +async fn erased_archive_without_receipt_or_inventory_remains_unknown_without_replanning() { + let f = SubjectRemovalFixture::new().await; + f.disable(); + let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); + conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE UPDATE OF state ON worker_removal_operations WHEN NEW.state='executing' BEGIN SELECT RAISE(ABORT,'injected interruption'); END;").unwrap(); + assert_ne!(f.remove().status, StatusCode::OK.as_u16()); + conn.execute_batch("DROP TRIGGER interrupt_removal") + .unwrap(); + let old = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + f.api + .erase_subjektiv_journal_session(&f.attr.session_id, "erase-with-unknown-outcome") + .unwrap(); + // Corrupt/missing aggregate after an unacknowledged operation is NOT evidence + // that source_removed completed. Runtime has no matching receipt here. + std::fs::remove_file( + session_root(f.dir.path(), &f.target) + .parent() + .unwrap() + .join("worker.json"), + ) + .unwrap(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::SERVICE_UNAVAILABLE.as_u16(), + "{}", + response.body + ); + let body: Value = serde_json::from_str(&response.body).unwrap(); + assert_eq!(body["code"], "erased_removal_outcome_unknown"); + let saved = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(saved.plan.operation_id, old.plan.operation_id); + assert_eq!( + saved.plan.state, + crate::retention::WorkerRemovalPlanState::Executing + ); + let count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM worker_removal_operations WHERE worker_id=?1", + [&f.target.worker_id], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(count, 1); +} + +#[tokio::test] +async fn erased_archive_retained_diagnostics_receipt_finalizes_only_after_physical_ack_on_resend_or_reopen() + { + for restart in [false, true] { + for interrupt_erase in [false, true] { + let mut f = SubjectRemovalFixture::new().await; + f.disable(); + let policy = f + .api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(); + let policy = f + .api + .config_store + .update_worker_retention_policy( + TEST_WORKSPACE_ID, + &policy.policy_digest, + &crate::retention::WorkerRetentionPolicyUpdate { + policy_id: "fixture-retained-diagnostics".into(), + session_disposition: policy.session_disposition, + metadata_disposition: policy.metadata_disposition, + archive_retention: policy.archive_retention, + diagnostics_disposition: + worker_runtime::retention::DiagnosticsDisposition::Retain, + diagnostics_retention_seconds: Some(3600), + }, + ) + .unwrap(); + let worker_root = session_root(f.dir.path(), &f.target) + .parent() + .unwrap() + .to_path_buf(); + std::fs::create_dir_all(worker_root.join("runs/fixture-run")).unwrap(); + std::fs::write( + worker_root.join("runs/fixture-run/worker.out.log"), + "diagnostic original", + ) + .unwrap(); + let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap(); + // Real Runtime receipt exists; Backend commit/response is interrupted. + conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'injected metadata commit interruption'); END;").unwrap(); + assert_ne!(f.remove().status, StatusCode::OK.as_u16()); + conn.execute_batch("DROP TRIGGER interrupt_removal") + .unwrap(); + let old = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + let runtime_root = f.dir.path().join(".test-embedded-runtime-store"); + let receipt_path = runtime_root + .join("retention/operations") + .join(format!("{}.json", old.plan.operation_id)); + let receipt_bytes = std::fs::read(&receipt_path).unwrap(); + let receipt = f + .api + .runtime + .execute_worker_retention(&f.target, old.runtime_request.clone()) + .unwrap(); + assert!(receipt.source_removed); + assert!(receipt.diagnostics_retained); + let diagnostics_root = runtime_root + .join("archives/diagnostics") + .join(&old.plan.operation_id); + let diagnostics_manifest = diagnostics_root.join("manifest.json"); + let diagnostics_bytes = std::fs::read(&diagnostics_manifest).unwrap(); + let archive_source = runtime_root + .join("archives/workers") + .join(old.plan.archive_id.as_ref().unwrap()) + .join("session"); + assert!(diagnostics_root.exists()); + assert!(archive_source.exists()); + // Domain erasure succeeds, but no physical acknowledgement can be + // claimed from the old retained locator after aggregate removal. + assert!( + f.api + .erase_subjektiv_journal_session( + &f.attr.session_id, + "erase-retained-diagnostics" + ) + .is_err() + ); + if interrupt_erase { + // Runtime must reject mismatched diagnostics identity rather than + // acknowledge removal of an unrelated operation. No fake Host ack. + let mut damaged: Value = serde_json::from_slice(&diagnostics_bytes).unwrap(); + damaged["input_fingerprint"] = json!("unrelated-operation"); + std::fs::write(&diagnostics_manifest, serde_json::to_vec(&damaged).unwrap()) + .unwrap(); + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::CONFLICT.as_u16(), + "{}", + response.body + ); + let body: Value = serde_json::from_str(&response.body).unwrap(); + assert_eq!(body["code"], "erased_removal_reconciliation_unconfirmed"); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .is_some() + ); + let count: i64 = conn.query_row("SELECT COUNT(*) FROM worker_retention_audit_events WHERE operation_id=?1 AND event_kind='erased_removal_receipt'", [&old.plan.operation_id], |r| r.get(0)).unwrap(); + assert_eq!( + count, 0, + "failed acknowledgement cannot authorize finalization/audit" + ); + assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt_bytes); + let waiting = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(waiting.plan.operation_id, old.plan.operation_id); + assert_eq!(waiting.plan.input_fingerprint, old.plan.input_fingerprint); + assert_eq!( + waiting.plan.state, + crate::retention::WorkerRemovalPlanState::Executing + ); + let plans: i64 = conn + .query_row( + "SELECT COUNT(*) FROM worker_removal_operations WHERE worker_id=?1", + [&f.target.worker_id], + |r| r.get(0), + ) + .unwrap(); + assert_eq!( + plans, 1, + "missing physical ack cannot authorize a replacement plan" + ); + std::fs::write(&diagnostics_manifest, &diagnostics_bytes).unwrap(); + } + if restart { + f = f.reopen().await; + } + let response = f.remove(); + assert_eq!( + response.status, + StatusCode::OK.as_u16(), + "restart={restart}, interrupted={interrupt_erase}: {}", + response.body + ); + assert!(!diagnostics_root.exists()); + assert!(!archive_source.exists()); + assert_eq!( + std::fs::read(&receipt_path).unwrap(), + receipt_bytes, + "historical Runtime receipt must not be rewritten" + ); + let saved = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); + assert_eq!(saved.plan.operation_id, old.plan.operation_id); + assert_eq!(saved.plan.input_fingerprint, old.plan.input_fingerprint); + assert_eq!( + saved.plan.diagnostics_disposition, + old.plan.diagnostics_disposition + ); + let detail: String = conn.query_row("SELECT detail FROM worker_retention_audit_events WHERE operation_id=?1 AND event_kind='erased_removal_receipt'", [&old.plan.operation_id], |r| r.get(0)).unwrap(); + let detail: Value = serde_json::from_str(&detail).unwrap(); + assert_eq!(detail["source_removed"], true); + assert_eq!(detail["old_diagnostics_retained"], true); + assert_eq!( + detail["runtime_erasure_ack"]["session_id"], + f.attr.session_id + ); + let diagnostics_catalog: i64 = conn + .query_row( + "SELECT COUNT(*) FROM worker_diagnostics_archives WHERE operation_id=?1", + [&old.plan.operation_id], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(diagnostics_catalog, 0); + let tombstone_archive: Option = conn + .query_row( + "SELECT archive_id FROM worker_tombstones WHERE operation_id=?1", + [&old.plan.operation_id], + |r| r.get(0), + ) + .unwrap(); + assert!(tombstone_archive.is_none()); + f.assert_erased_removal(); + assert_eq!( + f.api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap(), + Some(policy) + ); + } + } +} diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 6a6c3075e..dbbcc2a11 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -1451,6 +1451,40 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { "Worker retention authority is unavailable".to_string(), )) } + fn supersede_erased_worker_removal( + &self, + prepared: &crate::retention::PreparedWorkerRemoval, + proof: &crate::retention::RuntimeJournalErasureAck, + inventory: &worker_runtime::retention::WorkerRetentionInventory, + ) -> std::result::Result<(), crate::retention::WorkerRetentionError> { + let _ = (prepared, proof, inventory); + Err(crate::retention::WorkerRetentionError::Invalid( + "Erased removal reconciliation authority is unavailable".into(), + )) + } + fn audit_erased_worker_removal_receipt( + &self, + prepared: &crate::retention::PreparedWorkerRemoval, + proof: &crate::retention::RuntimeJournalErasureAck, + result: &worker_runtime::retention::WorkerRetentionExecutionResult, + ) -> std::result::Result<(), crate::retention::WorkerRetentionError> { + let _ = (prepared, proof, result); + Err(crate::retention::WorkerRetentionError::Invalid( + "Erased removal reconciliation authority is unavailable".into(), + )) + } + fn commit_erased_worker_removal( + &self, + workspace_id: &str, + prepared: &crate::retention::PreparedWorkerRemoval, + result: &worker_runtime::retention::WorkerRetentionExecutionResult, + proof: &crate::retention::RuntimeJournalErasureAck, + ) -> std::result::Result { + let _ = (workspace_id, prepared, result, proof); + Err(crate::retention::WorkerRetentionError::Invalid( + "Erased removal reconciliation authority is unavailable".into(), + )) + } fn list_workspaces(&self) -> Result>; fn upsert_repository(&self, record: &RepositoryRecord) -> Result<()>; fn insert_repository(&self, record: &RepositoryRecord) -> Result; @@ -5438,6 +5472,37 @@ impl ControlPlaneStore for SqliteWorkspaceStore { ) } + fn supersede_erased_worker_removal( + &self, + prepared: &crate::retention::PreparedWorkerRemoval, + proof: &crate::retention::RuntimeJournalErasureAck, + inventory: &worker_runtime::retention::WorkerRetentionInventory, + ) -> std::result::Result<(), crate::retention::WorkerRetentionError> { + SqliteWorkspaceStore::supersede_erased_worker_removal(self, prepared, proof, inventory) + } + fn audit_erased_worker_removal_receipt( + &self, + prepared: &crate::retention::PreparedWorkerRemoval, + proof: &crate::retention::RuntimeJournalErasureAck, + result: &worker_runtime::retention::WorkerRetentionExecutionResult, + ) -> std::result::Result<(), crate::retention::WorkerRetentionError> { + SqliteWorkspaceStore::audit_erased_worker_removal_receipt(self, prepared, proof, result) + } + fn commit_erased_worker_removal( + &self, + workspace_id: &str, + prepared: &crate::retention::PreparedWorkerRemoval, + result: &worker_runtime::retention::WorkerRetentionExecutionResult, + proof: &crate::retention::RuntimeJournalErasureAck, + ) -> std::result::Result { + SqliteWorkspaceStore::commit_erased_worker_removal( + self, + workspace_id, + prepared, + result, + proof, + ) + } fn list_workspaces(&self) -> Result> { self.with_conn(|conn| { let mut stmt = conn.prepare(