diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index 225045481..6e4243cc8 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -1219,8 +1219,34 @@ pub struct SubjectDeleteWorkerResponse { pub originals_absent: bool, } +/// Additional erasure of a Subject Job's newly reconciled removal copies. +/// Both identities are exact saved requests; this never changes the old receipt. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct SubjectJobRemovalPurgeRequest { + pub deletion: SubjectDeleteWorkerRequest, + pub removal: WorkerRetentionExecutionRequest, +} + +/// Confirms current physical erasure under the permanent deletion fence. +/// Operation identity belongs to `removal`, not the Subject deletion operation. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct SubjectJobRemovalPurgeResponse { + pub workspace_id: String, + pub worker_id: WorkerId, + pub operation_id: String, + pub input_fingerprint: String, +} + #[api(reqwest, axum)] pub trait RuntimeApi { + #[post("/v1/subject-job-removal-purge", status = 200, error_status = 400)] + async fn purge_subject_job_removal( + &self, + #[body] request: SubjectJobRemovalPurgeRequest, + ) -> Result; + #[post("/v1/subject-delete-worker", status = 200, error_status = 400)] async fn subject_delete_worker( &self, @@ -1787,6 +1813,18 @@ mod tests { ) -> Result { Err(test_error(501)) } + async fn purge_subject_job_removal( + &self, + request: SubjectJobRemovalPurgeRequest, + ) -> Result { + Ok(SubjectJobRemovalPurgeResponse { + workspace_id: request.removal.workspace_id, + worker_id: request.removal.worker_id, + operation_id: request.removal.operation_id, + input_fingerprint: request.removal.input_fingerprint, + }) + } + async fn subject_delete_worker( &self, request: SubjectDeleteWorkerRequest, @@ -1989,9 +2027,35 @@ mod tests { .subject_delete_worker(deletion.clone()) .await .unwrap(); - let mut expected = serde_json::to_value(deletion).unwrap(); + let mut expected = serde_json::to_value(&deletion).unwrap(); expected["originals_absent"] = true.into(); assert_eq!(serde_json::to_value(receipt).unwrap(), expected); + let purge = SubjectJobRemovalPurgeRequest { + removal: WorkerRetentionExecutionRequest { + operation_id: "old-removal".into(), + input_fingerprint: "old-removal-fingerprint".into(), + archive_id: Some("old-archive".into()), + workspace_id: deletion.workspace_id.clone(), + source_runtime_id: "runtime-a".into(), + worker_id: deletion.worker_id, + source_created_at: "2026-01-01T00:00:00Z".into(), + removed_at: "2026-01-02T00:00:00Z".into(), + effective_profile: None, + retention_class: None, + policy_id: "old-policy".into(), + session_disposition: SessionDisposition::Archive, + diagnostics_disposition: DiagnosticsDisposition::Retain, + }, + deletion, + }; + let receipt = client + .purge_subject_job_removal(purge.clone()) + .await + .unwrap(); + assert_eq!(receipt.workspace_id, purge.removal.workspace_id); + assert_eq!(receipt.worker_id, purge.removal.worker_id); + assert_eq!(receipt.operation_id, purge.removal.operation_id); + assert_eq!(receipt.input_fingerprint, purge.removal.input_fingerprint); let summary = client.runtime_summary().await.unwrap(); assert_eq!(summary.runtime.display_name.as_deref(), Some("round-trip")); let workers = client diff --git a/crates/subjektiv/README.md b/crates/subjektiv/README.md index 5212d8572..c7a413cb7 100644 --- a/crates/subjektiv/README.md +++ b/crates/subjektiv/README.md @@ -35,6 +35,14 @@ originals and pre-existing archives have independent retention lifecycles: the W Host archives source originals rather than erasing them. Those archives may contain previously injected Memory; Workdirs, Tickets, exports and external provider copies are also outside Subject content erasure. These boundaries must be explained to the user. +An unfinished, previously authorized Worker retention operation is snapshotted separately +from prior successful removals. Its exact original request is reconciled under the permanent +fence; its disposition, fingerprint and canonical result are never rewritten. Body originals +follow that earlier independent request. For a Subject Job, any archive/diagnostic copies +produced by the unfinished operation require a distinct Runtime physical-erasure acknowledgement +before deletion can complete. Lost responses or failed receipt commits retain the same request +and incomplete state across restart. Successes that already existed at deletion start remain +under their independent archive lifecycle; a later success is not inserted into that snapshot. The domain rechecks after out-of-DB capture callbacks and before returning Job read responses; already returned DTOs cannot be revoked by SQLite. Host lifecycle locking must cover delivery/reinjection of previously fetched content as well as finish. diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 7fd7880aa..06f551100 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2104,7 +2104,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s } if matches!( path, - "/v1/subject-create-recovery" | "/v1/subject-delete-worker" + "/v1/subject-create-recovery" + | "/v1/subject-delete-worker" + | "/v1/subject-job-removal-purge" ) && *method == Method::POST { return Some("workers:delete"); diff --git a/crates/worker-runtime/src/http_server/runtime_management_api.rs b/crates/worker-runtime/src/http_server/runtime_management_api.rs index 3ac26499b..8d3a1157e 100644 --- a/crates/worker-runtime/src/http_server/runtime_management_api.rs +++ b/crates/worker-runtime/src/http_server/runtime_management_api.rs @@ -406,6 +406,18 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi { } } + async fn purge_subject_job_removal( + &self, + request: runtime_api::SubjectJobRemovalPurgeRequest, + ) -> Result { + let scope = material_scope(&self.state, &request.deletion.workspace_id)?; + self.state + .runtime + .purge_subject_job_removal_scoped(&scope, request) + .map_err(RuntimeHttpRestError::runtime) + .map_err(api_error) + } + async fn subject_delete_worker( &self, request: runtime_api::SubjectDeleteWorkerRequest, 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 0a89c9c53..50949ff17 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 @@ -3,6 +3,8 @@ //! verifier are connected. Only controller execution is replaced by AcceptingBackend; //! no model, external server, user credentials or running Runtime is used. use super::*; +#[path = "subject_job_removal_purge_tests.rs"] +mod subject_job_removal_purge_tests; use crate::retention::{DiagnosticsDisposition, SessionDisposition, runtime_session_retention}; use runtime_api::{ SessionMaterialCatalogAvailability as Catalog, SessionMaterialReadAvailability as Read, diff --git a/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs b/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs new file mode 100644 index 000000000..f757b690c --- /dev/null +++ b/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs @@ -0,0 +1,117 @@ +//! Real filesystem failures and reopened durable fences for the additional purge. +//! Uses the same production stores/router fixture as the copy-erasure suite. +use super::*; + +#[test] +fn subject_job_removal_purge_partial_physical_failure_keeps_fence_and_retry_erases_after_reopen() { + let f = Fixture::new(); + let mut request = f.removal_purge_request(SessionDisposition::Archive); + request.removal.diagnostics_disposition = runtime_api::DiagnosticsDisposition::Retain; + f.complete_removal(&request); + let receipt_path = f.receipt_path(); + let receipt = std::fs::read(&receipt_path).unwrap(); + let session_root = f + .temp + .path() + .join("archives/workers/canonical-archive/session"); + let parked = f.temp.path().join("interrupted-session-fixture"); + std::fs::rename(&session_root, &parked).unwrap(); + std::fs::write(&session_root, b"deterministic directory removal failure").unwrap(); + assert!( + f.runtime + .purge_subject_job_removal_scoped(&f.scope, request.clone()) + .is_err() + ); + assert!(matches!( + runtime_session_retention(f.temp.path()).ensure_live(f.session), + Err(session_store::StoreError::SessionErased(_)) + )); + assert!( + !f.temp + .path() + .join("archives/diagnostics/remove-source") + .exists() + ); + assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt); + std::fs::remove_file(&session_root).unwrap(); + std::fs::rename(parked, &session_root).unwrap(); + let Fixture { + temp, + runtime, + scope, + store, + session, + .. + } = f; + drop(store); + drop(runtime); + let runtime = Runtime::with_fs_store_and_execution_backend( + crate::FsRuntimeStoreOptions::new(temp.path()).with_runtime_id("runtime-test"), + Arc::new(AcceptingBackend), + ) + .unwrap(); + runtime.bind_runtime_identity("runtime-test").unwrap(); + runtime + .purge_subject_job_removal_scoped(&scope, request) + .unwrap(); + assert!(!session_root.exists()); + assert!(matches!( + runtime_session_retention(temp.path()).lock_live(session), + Err(session_store::StoreError::SessionErased(_)) + )); + assert_eq!(std::fs::read(receipt_path).unwrap(), receipt); +} + +#[test] +fn subject_job_removal_purge_unconfirmed_durable_deletion_fence_preserves_copies_until_reconciled() +{ + let f = Fixture::new(); + let request = f.removal_purge_request(SessionDisposition::Archive); + f.complete_removal(&request); + let copies = f.copies(); + let receipt_path = f.receipt_path(); + let receipt = std::fs::read(&receipt_path).unwrap(); + let Fixture { + temp, + runtime, + scope, + store, + .. + } = f; + drop(store); + drop(runtime); + // Frozen durable crash-window fixture: permanent admission exclusion was + // saved, but no confirmed execution-shutdown response was committed. + let snapshot_path = temp.path().join("runtime.json"); + let mut snapshot: serde_json::Value = + serde_json::from_slice(&std::fs::read(&snapshot_path).unwrap()).unwrap(); + snapshot["subject_delete_worker_fences"][request.deletion.worker_id.to_string()]["response"] = + serde_json::Value::Null; + std::fs::write(&snapshot_path, serde_json::to_vec(&snapshot).unwrap()).unwrap(); + let runtime = Runtime::with_fs_store_and_execution_backend( + crate::FsRuntimeStoreOptions::new(temp.path()).with_runtime_id("runtime-test"), + Arc::new(AcceptingBackend), + ) + .unwrap(); + runtime.bind_runtime_identity("runtime-test").unwrap(); + assert!( + runtime + .purge_subject_job_removal_scoped(&scope, request.clone()) + .is_err() + ); + assert_eq!(tree(&temp.path().join("archives")), copies); + assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt); + runtime + .subject_delete_worker_scoped(&scope, request.deletion.clone()) + .unwrap(); + runtime + .purge_subject_job_removal_scoped(&scope, request) + .unwrap(); + assert!( + !temp + .path() + .join("archives/workers/canonical-archive/session") + .exists() + ); + assert_eq!(std::fs::read(receipt_path).unwrap(), receipt); +} diff --git a/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_tests.rs b/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_tests.rs new file mode 100644 index 000000000..e90337ce1 --- /dev/null +++ b/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_tests.rs @@ -0,0 +1,486 @@ +//! Additional removal-copy erasure uses the real Runtime, retention provider, +//! Session tombstone, filesystem and signed HTTP router. Only execution is fake +//! (the parent fixture's AcceptingBackend); no Backend recovery is simulated here. +use super::*; +#[path = "subject_job_removal_purge_recovery_tests.rs"] +mod recovery; +use runtime_api::{SubjectDeleteWorkerRequest, SubjectJobRemovalPurgeRequest}; +use std::collections::BTreeMap; +use std::path::Path; + +fn tree(root: &Path) -> BTreeMap> { + fn walk(root: &Path, at: &Path, files: &mut BTreeMap>) { + if !at.exists() { + return; + } + for entry in std::fs::read_dir(at).unwrap() { + let entry = entry.unwrap(); + if entry.file_type().unwrap().is_dir() { + walk(root, &entry.path(), files); + } else { + files.insert( + entry.path().strip_prefix(root).unwrap().into(), + std::fs::read(entry.path()).unwrap(), + ); + } + } + } + let mut files = BTreeMap::new(); + walk(root, root, &mut files); + files +} + +impl Fixture { + fn deletion_request(&self) -> SubjectDeleteWorkerRequest { + SubjectDeleteWorkerRequest { + workspace_id: self.scope.workspace_id.clone(), + worker_id: self.worker.worker_id, + create_fingerprint: task_request("identity only").create_fingerprint, + operation_id: "subject-deletion".into(), + } + } + + fn removal_purge_request( + &self, + disposition: SessionDisposition, + ) -> SubjectJobRemovalPurgeRequest { + let removal = self.retention_request(disposition); + SubjectJobRemovalPurgeRequest { + deletion: self.deletion_request(), + removal: serde_json::from_value(serde_json::to_value(removal).unwrap()).unwrap(), + } + } + + fn complete_removal( + &self, + request: &SubjectJobRemovalPurgeRequest, + ) -> crate::retention::WorkerRetentionExecutionResult { + let diagnostics = self.root().parent().unwrap().join("runs/attempt"); + std::fs::create_dir_all(&diagnostics).unwrap(); + std::fs::write(diagnostics.join("worker.out.log"), b"Subject Job output").unwrap(); + self.runtime + .subject_delete_worker_scoped(&self.scope, request.deletion.clone()) + .unwrap(); + let removal = + serde_json::from_value(serde_json::to_value(&request.removal).unwrap()).unwrap(); + self.runtime.execute_worker_retention(&removal).unwrap() + } + + fn receipt_path(&self) -> PathBuf { + self.temp + .path() + .join("retention/operations/remove-source.json") + } + + fn copies(&self) -> BTreeMap> { + tree(&self.temp.path().join("archives")) + } + + fn independent_worker(&self) -> WorkerRef { + let mut request = task_request("independent Worker"); + request.config_bundle = None; + request.memory_settings.as_mut().unwrap().workspace_id = self.scope.workspace_id.clone(); + self.runtime + .create_worker_scoped(&self.scope, request) + .unwrap() + .worker_ref + } + + // Independent successful snapshot made before the Subject deletion starts. + fn prior_archive(&self) -> WorkerRef { + let worker = self.independent_worker(); + let aggregate = self + .temp + .path() + .join("workers") + .join(worker.worker_id.to_string()); + let session = session_store::new_session_id(); + let segment = session_store::new_segment_id(); + WorkerSessionStore::new(aggregate.join("session")) + .unwrap() + .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(); + WorkerAggregateStore::new(&aggregate, format!("worker-runtime-{}", worker.worker_id)) + .unwrap() + .write( + &session_store::WorkerMetadata::new( + format!("worker-runtime-{}", worker.worker_id), + Some(session_store::WorkerActiveSegmentRef::active_segment( + session, segment, + )), + ) + .with_workspace_id(&self.scope.workspace_id), + ) + .unwrap(); + self.runtime + .stop_worker_scoped(&self.scope, &worker, None) + .unwrap(); + let mut removal = self.retention_request(SessionDisposition::Archive); + removal.worker_id = worker.worker_id; + removal.operation_id = "prior-removal".into(); + removal.archive_id = Some("prior-archive".into()); + removal.input_fingerprint = "prior-fingerprint".into(); + self.runtime.execute_worker_retention(&removal).unwrap(); + worker + } +} + +#[test] +fn subject_job_removal_purge_erases_copies_preserves_receipt_and_prior_success_after_reopen() { + for disposition in [SessionDisposition::Archive, SessionDisposition::Purge] { + let f = Fixture::new(); + f.prior_archive(); + let prior = tree(&f.temp.path().join("archives/workers/prior-archive")); + assert!(!prior.is_empty()); + let unrelated = f.independent_worker(); + let unrelated_root = f + .temp + .path() + .join("workers") + .join(unrelated.worker_id.to_string()); + let unrelated_before = tree(&unrelated_root); + let mut request = f.removal_purge_request(disposition); + request.removal.diagnostics_disposition = runtime_api::DiagnosticsDisposition::Retain; + let original_result = f.complete_removal(&request); + let receipt = std::fs::read(f.receipt_path()).unwrap(); + let archive = f.temp.path().join("archives/workers/canonical-archive"); + let diagnostics = f.temp.path().join("archives/diagnostics/remove-source"); + assert!(diagnostics.is_dir()); + let manifest = (disposition == SessionDisposition::Archive) + .then(|| std::fs::read(archive.join("manifest.json")).unwrap()); + let archive_staging = f + .temp + .path() + .join("archives/workers/.canonical-archive.tmp-123-1"); + if let Some(manifest) = &manifest { + std::fs::create_dir_all(archive_staging.join("session/segments")).unwrap(); + std::fs::write(archive_staging.join("manifest.json"), manifest).unwrap(); + std::fs::write( + archive_staging.join("session/segments/interrupted.jsonl"), + b"interrupted Subject Job copy", + ) + .unwrap(); + } + // Diagnostics copying precedes publication of its manifest. Interrupted + // staging is still bound to the exact removal operation name. + let diagnostics_staging = f + .temp + .path() + .join("archives/diagnostics/.remove-source.tmp-123-2"); + std::fs::create_dir_all(&diagnostics_staging).unwrap(); + std::fs::write( + diagnostics_staging.join("worker.out.log"), + b"interrupted diagnostic copy", + ) + .unwrap(); + for _ in 0..2 { + let response = f + .runtime + .purge_subject_job_removal_scoped(&f.scope, request.clone()) + .unwrap(); + assert_eq!(response.operation_id, request.removal.operation_id); + assert_eq!( + response.input_fingerprint, + request.removal.input_fingerprint + ); + assert_eq!(response.worker_id, request.removal.worker_id); + assert_eq!(response.workspace_id, request.removal.workspace_id); + assert!(!diagnostics.exists()); + assert!(!diagnostics_staging.exists()); + assert!(!archive_staging.join("session").exists()); + assert!(!archive.join("session").exists()); + assert_eq!(std::fs::read(f.receipt_path()).unwrap(), receipt); + if let Some(manifest) = &manifest { + assert_eq!( + std::fs::read(archive.join("manifest.json")).unwrap(), + *manifest + ); + assert!(matches!( + runtime_session_retention(f.temp.path()).ensure_live(f.session), + Err(session_store::StoreError::SessionErased(_)) + )); + } + } + assert_eq!( + tree(&f.temp.path().join("archives/workers/prior-archive")), + prior + ); + assert_eq!(tree(&unrelated_root), unrelated_before); + let Fixture { + temp, + runtime, + scope, + store, + session, + .. + } = f; + drop(store); + drop(runtime); + let runtime = Runtime::with_fs_store_and_execution_backend( + crate::FsRuntimeStoreOptions::new(temp.path()).with_runtime_id("runtime-test"), + Arc::new(AcceptingBackend), + ) + .unwrap(); + runtime.bind_runtime_identity("runtime-test").unwrap(); + // Lost response/restart is an exact retry, not a new erasure operation. + runtime + .purge_subject_job_removal_scoped(&scope, request.clone()) + .unwrap(); + let removal = + serde_json::from_value(serde_json::to_value(&request.removal).unwrap()).unwrap(); + assert_eq!( + runtime.execute_worker_retention(&removal).unwrap(), + original_result + ); + assert!(!archive.join("session").exists()); + assert!(!diagnostics.exists()); + assert_eq!( + std::fs::read(temp.path().join("retention/operations/remove-source.json")).unwrap(), + receipt + ); + assert_eq!( + tree(&temp.path().join("archives/workers/prior-archive")), + prior + ); + assert_eq!(tree(&unrelated_root), unrelated_before); + if disposition == SessionDisposition::Archive { + assert!(matches!( + runtime_session_retention(temp.path()).lock_live(session), + Err(session_store::StoreError::SessionErased(_)) + )); + assert!( + runtime + .session_public_material_reader_scoped( + &scope, + &Source::Archived { + archive_id: "canonical-archive".into() + }, + &session.to_string() + ) + .is_err() + ); + } + } +} + +#[test] +fn subject_job_removal_purge_requires_complete_old_request_and_confirmed_exact_fence() { + let f = Fixture::new(); + let request = f.removal_purge_request(SessionDisposition::Archive); + let copies = f.copies(); + assert!( + f.runtime + .purge_subject_job_removal_scoped(&f.scope, request.clone()) + .is_err() + ); + assert_eq!(f.copies(), copies); + f.complete_removal(&request); + let copies = f.copies(); + let receipt = std::fs::read(f.receipt_path()).unwrap(); + let original = serde_json::to_value(&request).unwrap(); + // Every saved request field is bound, not just operation/fingerprint. These + // are external serialized requests, including the pre-policy-change values. + for (section, field, value) in [ + ( + "deletion", + "workspace_id", + serde_json::json!("foreign-workspace"), + ), + ( + "deletion", + "worker_id", + serde_json::json!(WorkerId::now_v7()), + ), + ( + "deletion", + "create_fingerprint", + serde_json::json!("foreign-create"), + ), + ( + "deletion", + "operation_id", + serde_json::json!("foreign-deletion"), + ), + ( + "removal", + "operation_id", + serde_json::json!("unknown-removal"), + ), + ( + "removal", + "input_fingerprint", + serde_json::json!("foreign-fingerprint"), + ), + ( + "removal", + "archive_id", + serde_json::json!("foreign-archive"), + ), + ( + "removal", + "workspace_id", + serde_json::json!("foreign-workspace"), + ), + ( + "removal", + "source_runtime_id", + serde_json::json!("foreign-runtime"), + ), + ( + "removal", + "worker_id", + serde_json::json!(WorkerId::now_v7()), + ), + ("removal", "source_created_at", serde_json::json!("changed")), + ("removal", "removed_at", serde_json::json!("changed")), + ("removal", "effective_profile", serde_json::json!("changed")), + ("removal", "retention_class", serde_json::json!("changed")), + ("removal", "policy_id", serde_json::json!("new-policy")), + ("removal", "session_disposition", serde_json::json!("purge")), + ( + "removal", + "diagnostics_disposition", + serde_json::json!("retain"), + ), + ] { + let mut changed = original.clone(); + changed[section][field] = value; + let changed = serde_json::from_value(changed).unwrap(); + assert!( + f.runtime + .purge_subject_job_removal_scoped(&f.scope, changed) + .is_err(), + "{section}.{field}" + ); + assert_eq!(f.copies(), copies, "{section}.{field}"); + assert_eq!(std::fs::read(f.receipt_path()).unwrap(), receipt); + } + for scope in [ + RuntimeWorkspaceScope::new("workspace-b", "https://backend.test"), + RuntimeWorkspaceScope::new("workspace-a", "https://other-backend.test"), + ] { + assert!( + f.runtime + .purge_subject_job_removal_scoped(&scope, request.clone()) + .is_err() + ); + assert_eq!(f.copies(), copies); + } +} + +#[test] +fn subject_job_removal_purge_missing_pending_or_foreign_receipt_preserves_copies() { + for kind in [ + "missing", + "pending", + "foreign-worker", + "foreign-disposition", + ] { + let f = Fixture::new(); + let mut request = f.removal_purge_request(SessionDisposition::Archive); + request.removal.diagnostics_disposition = runtime_api::DiagnosticsDisposition::Retain; + f.complete_removal(&request); + let path = f.receipt_path(); + let mut receipt: serde_json::Value = + serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); + match kind { + "missing" => std::fs::remove_file(&path).unwrap(), + _ => { + match kind { + "pending" => receipt["result"]["source_removed"] = false.into(), + "foreign-worker" => { + receipt["result"]["worker_id"] = serde_json::json!(WorkerId::now_v7()) + } + _ => receipt["result"]["session_disposition"] = "purge".into(), + } + std::fs::write(&path, serde_json::to_vec(&receipt).unwrap()).unwrap(); + } + } + let copies = f.copies(); + let retained_receipt = std::fs::read(&path).ok(); + assert!( + f.runtime + .purge_subject_job_removal_scoped(&f.scope, request) + .is_err(), + "{kind}" + ); + assert_eq!(f.copies(), copies, "{kind}"); + assert_eq!(std::fs::read(path).ok(), retained_receipt); + runtime_session_retention(f.temp.path()) + .ensure_live(f.session) + .unwrap(); + } +} + +#[tokio::test] +async fn subject_job_removal_purge_signed_http_requires_delete_and_matching_scope() { + let f = Fixture::new(); + let mut request = f.removal_purge_request(SessionDisposition::Archive); + request.removal.diagnostics_disposition = runtime_api::DiagnosticsDisposition::Retain; + f.complete_removal(&request); + let identity = RuntimeIdentityMaterial::generate("purge-key").unwrap(); + let trust = workspace_trust(&identity); + let app = runtime_http_router_with_workspace_auth( + f.runtime.clone(), + None, + workspace_test_auth(vec![trust.clone()]), + ); + let path = "/v1/subject-job-removal-purge"; + let copies = f.copies(); + for operation in ["workers:read", "workers:stop", "workers:create"] { + let body = serde_json::to_vec(&request).unwrap(); + let mut claims = signed_claims(&trust, "POST", path, &body); + claims.operation = operation.into(); + let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert_eq!(f.copies(), copies); + } + let mut foreign = request.clone(); + foreign.deletion.workspace_id = "foreign-workspace".into(); + let body = serde_json::to_vec(&foreign).unwrap(); + let claims = signed_claims(&trust, "POST", path, &body); + let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await; + assert_eq!(response.status(), StatusCode::FORBIDDEN); + assert_eq!(f.copies(), copies); + for _ in 0..2 { + let body = serde_json::to_vec(&request).unwrap(); + let claims = signed_claims(&trust, "POST", path, &body); + assert_eq!(claims.operation, "workers:delete"); + let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await; + assert_eq!(response.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(response.into_body(), 16384) + .await + .unwrap(); + let response: runtime_api::SubjectJobRemovalPurgeResponse = + serde_json::from_slice(&bytes).unwrap(); + assert_eq!(response.operation_id, request.removal.operation_id); + assert_eq!( + response.input_fingerprint, + request.removal.input_fingerprint + ); + assert!( + !f.temp + .path() + .join("archives/workers/canonical-archive/session") + .exists() + ); + assert!( + !f.temp + .path() + .join("archives/diagnostics/remove-source") + .exists() + ); + } +} diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index c11802fd4..29296d3f4 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -359,28 +359,91 @@ fn erase_archived_diagnostics( return Err(invalid()); } } + erase_operation_diagnostics( + runtime_root, + &archive.workspace_id, + &archive.source_runtime_id, + archive.source_worker_id, + &archive.operation_id, + &archive.input_fingerprint, + ) +} + +fn erase_removal_session_copies( + runtime_root: &Path, + archive: &WorkerSessionArchiveManifest, +) -> Result<(), session_store::StoreError> { + let parent = runtime_root.join("archives/workers"); + let prefix = format!(".{}.tmp-", archive.archive_id); + for entry in fs::read_dir(&parent)? { + let entry = entry?; + let name = entry.file_name(); + let Some(name) = name.to_str() else { continue }; + if name != archive.archive_id && !is_operation_staging(name, &prefix) { + continue; + } + if !entry.file_type()?.is_dir() { + return Err(std::io::Error::other("Removal archive source is not a directory").into()); + } + let mut saved: WorkerSessionArchiveManifest = + serde_json::from_slice(&fs::read(entry.path().join("manifest.json"))?)?; + // Each interrupted attempt attests the same input, with its own time. + saved.archived_at_unix_seconds = archive.archived_at_unix_seconds; + if saved != *archive { + return Err(std::io::Error::other("Removal archive copy identity mismatch").into()); + } + match fs::remove_dir_all(entry.path().join("session")) { + Ok(()) => (), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => (), + Err(error) => return Err(error.into()), + } + sync_directory(&entry.path(), "erase removal Session") + .map_err(|error| std::io::Error::other(error.to_string()))?; + } + Ok(()) +} + +fn is_operation_staging(name: &str, prefix: &str) -> bool { + name.strip_prefix(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()) + }) + }) +} + +fn erase_operation_diagnostics( + runtime_root: &Path, + workspace_id: &str, + source_runtime_id: &str, + worker_id: WorkerId, + operation_id: &str, + input_fingerprint: &str, +) -> Result<(), session_store::StoreError> { + let invalid = || { + session_store::StoreError::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "Archived diagnostics erasure identity mismatch", + )) + }; + validate_id("operation_id", operation_id).map_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); + let staging_prefix = format!(".{operation_id}.tmp-"); 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 { + let staged = is_operation_staging(name, &staging_prefix); + if name != operation_id && !staged { continue; } if !entry.file_type()?.is_dir() { @@ -395,11 +458,11 @@ fn erase_archived_diagnostics( }; 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 + || manifest.workspace_id != workspace_id + || manifest.source_runtime_id != source_runtime_id + || manifest.source_worker_id != worker_id + || manifest.operation_id != operation_id + || manifest.input_fingerprint != input_fingerprint { return Err(invalid()); } @@ -480,6 +543,109 @@ impl FsWorkerRetentionProvider { Ok(receipt.result.source_removed.then_some(receipt.result)) } + /// Caller holds the Worker operation lock and has verified the durable + /// Subject deletion fence. Never execute/recover a missing or pending receipt + /// here: the original canonical operation must have completed first. + pub(crate) fn purge_subject_job_removal( + &self, + request: &WorkerRetentionExecutionRequest, + ) -> Result<(), RuntimeError> { + validate_request(request)?; + let lock_path = self.runtime_root.join(RETENTION_LOCK); + let lock = OpenOptions::new() + .create(true) + .read(true) + .write(true) + .open(&lock_path) + .map_err(|source| RuntimeError::StoreIo { + operation: "lock Subject Job removal purge", + path: lock_path.clone(), + source, + })?; + lock.lock().map_err(|source| RuntimeError::StoreIo { + operation: "lock Subject Job removal purge", + path: lock_path, + source, + })?; + let receipt = read_operation_receipt(&self.operation_path(&request.operation_id)?)? + .ok_or_else(|| { + RuntimeError::InvalidRequest("Canonical removal receipt missing".into()) + })?; + validate_receipt_request(&receipt, request)?; + let result = &receipt.result; + if !result.source_removed + || result.worker_id != request.worker_id + || result.session_disposition != request.session_disposition + || result.diagnostics_disposition != request.diagnostics_disposition + || result.diagnostics_retained + != (request.diagnostics_disposition == DiagnosticsDisposition::Retain) + || result.archive.is_some() + != (request.session_disposition == SessionDisposition::Archive) + || self.worker_dir(request.worker_id).exists() + { + return Err(RuntimeError::InvalidRequest( + "Canonical removal is not an exact successful receipt".into(), + )); + } + let archive = if let Some(manifest) = &result.archive { + if request.archive_id.as_ref() != Some(&manifest.archive_id) + || manifest.workspace_id != request.workspace_id + || manifest.source_runtime_id != request.source_runtime_id + || manifest.source_worker_id != request.worker_id + || manifest.operation_id != request.operation_id + || manifest.input_fingerprint != request.input_fingerprint + { + return Err(RuntimeError::InvalidRequest( + "Removal archive identity mismatch".into(), + )); + } + let (saved, _) = self.session_archive_identity(&manifest.archive_id)?; + if saved != *manifest { + return Err(RuntimeError::InvalidRequest( + "Removal archive receipt mismatch".into(), + )); + } + let session = manifest + .source_session_id + .parse::() + .map_err(|_| { + RuntimeError::InvalidRequest("Invalid removal Session identity".into()) + })?; + Some(session) + } else { + None + }; + let erase = || -> Result<(), session_store::StoreError> { + erase_operation_diagnostics( + &self.runtime_root, + &request.workspace_id, + &request.source_runtime_id, + request.worker_id, + &request.operation_id, + &request.input_fingerprint, + )?; + if let Some(manifest) = &result.archive { + erase_removal_session_copies(&self.runtime_root, manifest)?; + } + Ok(()) + }; + if let Some(session) = &archive { + runtime_session_retention(&self.runtime_root) + .delete( + *session, + session_store::retention::SessionDeletionIntent::ErasePermanently, + erase, + ) + .map_err(retention_error)?; + } else { + // Diagnostics-only old Purge operations have no archive Session ID. + // The existing permanent Worker fence plus immutable completed receipt + // prevents execution and re-publication; hold the retention lock here. + erase().map_err(retention_error)?; + } + Ok(()) + } + pub(crate) fn snapshot( &self, workspace_id: &str, diff --git a/crates/worker-runtime/src/runtime/subject_delete.rs b/crates/worker-runtime/src/runtime/subject_delete.rs index 485faf258..6dcecaf3d 100644 --- a/crates/worker-runtime/src/runtime/subject_delete.rs +++ b/crates/worker-runtime/src/runtime/subject_delete.rs @@ -10,6 +10,79 @@ pub(crate) struct SubjectDeleteWorkerFence { } impl Runtime { + /// Erase only copies bound to an exact, completed canonical removal. The + /// Subject fence already excludes execution; this does not mint erase authority + /// for an arbitrary Worker or reinterpret the historical retention result. + pub fn purge_subject_job_removal_scoped( + &self, + scope: &RuntimeWorkspaceScope, + request: runtime_api::SubjectJobRemovalPurgeRequest, + ) -> Result { + let deletion = &request.deletion; + let removal = &request.removal; + if deletion.workspace_id != scope.workspace_id + || removal.workspace_id != deletion.workspace_id + || removal.worker_id != deletion.worker_id + { + return Err(RuntimeError::InvalidRequest( + "Subject Job removal scope mismatch".into(), + )); + } + let lease = self.worker_operation_lock(deletion.worker_id)?; + let _operation = lease.lock().map_err(|_| RuntimeError::StatePoisoned)?; + let mut state = self.lock()?; + state.ensure_running()?; + state.ensure_workspace_owner(scope, false)?; + if state.runtime_identity.as_deref() != Some(removal.source_runtime_id.as_str()) { + return Err(RuntimeError::InvalidRequest( + "Subject Job removal Runtime identity mismatch".into(), + )); + } + state + .subject_delete_worker_fences + .get(&deletion.worker_id) + .filter(|fence| fence.request == *deletion && fence.response.is_some()) + .ok_or_else(|| { + RuntimeError::InvalidRequest( + "Exact confirmed Subject deletion fence required".into(), + ) + })?; + if state.workers.contains_key(&deletion.worker_id) + || state + .unreconciled_worker_writes + .contains(&deletion.worker_id) + { + return Err(RuntimeError::InvalidRequest( + "Subject Job canonical removal must be reconciled first".into(), + )); + } + #[cfg(feature = "fs-store")] + { + let store = state.fs_store().ok_or_else(|| { + RuntimeError::InvalidRequest("Subject Job purge requires filesystem storage".into()) + })?; + let removal: WorkerRetentionExecutionRequest = serde_json::from_value( + serde_json::to_value(removal) + .map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?, + ) + .map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?; + // Worker -> Runtime -> retention -> Session is the canonical removal + // lock order. The provider validates the receipt under its disk lock. + FsWorkerRetentionProvider::new(store.runtime_dir()) + .purge_subject_job_removal(&removal)?; + Ok(runtime_api::SubjectJobRemovalPurgeResponse { + workspace_id: removal.workspace_id, + worker_id: removal.worker_id, + operation_id: removal.operation_id, + input_fingerprint: removal.input_fingerprint, + }) + } + #[cfg(not(feature = "fs-store"))] + Err(RuntimeError::InvalidRequest( + "Subject Job purge requires filesystem storage".into(), + )) + } + pub fn subject_delete_worker_scoped( &self, scope: &RuntimeWorkspaceScope, diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 8ba623282..e72990b20 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1292,6 +1292,13 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { }) } + fn purge_subject_job_removal( + &self, + _request: runtime_api::SubjectJobRemovalPurgeRequest, + ) -> Result { + Err("Subject Job removal source erasure is unavailable on this Runtime".into()) + } + fn subject_delete_worker( &self, _request: runtime_api::SubjectDeleteWorkerRequest, @@ -2276,6 +2283,32 @@ impl RuntimeRegistry { runtime.worker_session_attachment(workspace_id, &worker_ref, session_id, attachment_id) } + pub fn purge_subject_job_removal( + &self, + runtime_id: &str, + deletion: runtime_api::SubjectDeleteWorkerRequest, + removal: WorkerRetentionExecutionRequest, + ) -> Result { + validate_backend_identifier("runtime_id", runtime_id)?; + let removal = runtime_contract_convert(removal).map_err(|diagnostic| { + RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: runtime_id.into(), + code: "subject_job_removal_request_invalid".into(), + message: diagnostic.message, + } + })?; + self.runtime(runtime_id)? + .purge_subject_job_removal(runtime_api::SubjectJobRemovalPurgeRequest { + deletion, + removal, + }) + .map_err(|message| RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: runtime_id.into(), + code: "subject_job_removal_purge_unconfirmed".into(), + message, + }) + } + pub fn subject_delete_worker( &self, runtime_id: &str, @@ -3402,6 +3435,19 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { .map_err(|e| e.to_string()) } + fn purge_subject_job_removal( + &self, + request: runtime_api::SubjectJobRemovalPurgeRequest, + ) -> Result { + if request.deletion.workspace_id != self.workspace_id { + return Err("Subject Job removal erasure Workspace mismatch".into()); + } + let scope = RuntimeWorkspaceScope::new(&request.deletion.workspace_id, "embedded-backend"); + self.runtime + .purge_subject_job_removal_scoped(&scope, request) + .map_err(|e| e.to_string()) + } + fn recover_subject_create( &self, request: runtime_api::SubjectCreateRecoveryRequest, @@ -5694,6 +5740,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { }) } + fn purge_subject_job_removal( + &self, + request: runtime_api::SubjectJobRemovalPurgeRequest, + ) -> Result { + self.run_runtime_api( + self.request_timeout, + MAX_REMOTE_RUNTIME_RESPONSE_BYTES, + move |client| async move { client.purge_subject_job_removal(request).await }, + ) + .map_err(|diagnostic| diagnostic.message) + } + fn subject_delete_worker( &self, request: runtime_api::SubjectDeleteWorkerRequest, diff --git a/crates/workspace-server/src/implementation_result_tests.rs b/crates/workspace-server/src/implementation_result_tests.rs index baf355717..216c55787 100644 --- a/crates/workspace-server/src/implementation_result_tests.rs +++ b/crates/workspace-server/src/implementation_result_tests.rs @@ -407,7 +407,8 @@ fn reconstruct_implementation_schema96(conn: &Connection) -> Result<()> { } // Children before parents; dropping the tables also drops snapshot guards. conn.execute_batch( - "DROP TABLE subject_deletion_allocations; + "DROP TABLE subject_deletion_removals; + DROP TABLE subject_deletion_allocations; DROP TABLE subject_deletion_jobs; DROP TABLE subject_deletion_workers; DROP TABLE subject_deletions; diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index b727b0eef..c0dc1e64a 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -333,6 +333,8 @@ impl SqliteWorkspaceStore { return Err(StoreError::InvalidInput(if other{"cross-workspace".into()}else{"worker-missing".into()})); } }; + let pending:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM subject_deletion_removals WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND completed=0)",params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id],|row|row.get(0))?; + if pending {return Err(StoreError::InvalidInput("Pinned pre-deletion removal must reconcile before another retention plan".into()));} if crate::store::subject_deletion_worker_kind(&tx, &req.workspace_id, &req.worker)?.is_some() { let prior = tx.prepare("SELECT plan_id FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state IN ('executing','failed')")? .query_map(params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id],|r|r.get::<_,String>(0))? @@ -381,23 +383,26 @@ impl SqliteWorkspaceStore { return Err(StoreError::InvalidInput(format!("stale:{plan_id}:plan state {} is not executable", state_s(plan.state)))); } if !plan.blockers.is_empty(){return Err(StoreError::InvalidInput(format!("blocked:{}",serde_json::to_string(&plan.blockers).unwrap())));} - let policy=load_removal_policy(&tx,workspace_id,&plan.worker,plan.policy_id==ERASED_JOURNAL_POLICY_ID)?.ok_or_else(||StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?; - if policy.policy_id!=plan.policy_id||policy.policy_digest!=plan.policy_digest{ - mark_stale(&tx,&plan,"policy content changed")?;tx.commit()?; - return Err(stale_error(&plan,"policy content changed")); + let reconciliation=crate::store::subject_deletion_removal_matches(&tx,workspace_id,&plan.worker,&plan.operation_id,&plan.input_fingerprint)?; + if !reconciliation { + let policy=load_removal_policy(&tx,workspace_id,&plan.worker,plan.policy_id==ERASED_JOURNAL_POLICY_ID)?.ok_or_else(||StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?; + if policy.policy_id!=plan.policy_id||policy.policy_digest!=plan.policy_digest{ + mark_stale(&tx,&plan,"policy content changed")?;tx.commit()?; + return Err(stale_error(&plan,"policy content changed")); + } } let worker=load_worker(&tx,workspace_id,&plan.worker)?.ok_or_else(||StoreError::InvalidInput(format!("stale:{plan_id}:Worker missing")))?; - if worker.updated_at!=plan.worker_updated_at{ + if !reconciliation && worker.updated_at!=plan.worker_updated_at{ mark_stale(&tx,&plan,"Worker updated_at changed")?;tx.commit()?; return Err(stale_error(&plan,"Worker updated_at changed")); } if worker.retention_state=="pinned"{ - mark_stale(&tx,&plan,"hold added")?;tx.commit()?; + if !reconciliation {mark_stale(&tx,&plan,"hold added")?;}tx.commit()?; return Err(stale_error(&plan,"hold added")); } let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3 UNION ALL SELECT 1 FROM ticket_assignment_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND claim_state='pending' AND action IN ('assign','reassign'))",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|r|r.get(0))?; if assigned{ - mark_stale(&tx,&plan,"unfinished work added")?;tx.commit()?; + if !reconciliation {mark_stale(&tx,&plan,"unfinished work added")?;}tx.commit()?; return Err(stale_error(&plan,"unfinished work added")); } let now=Utc::now().to_rfc3339(); @@ -416,6 +421,15 @@ impl SqliteWorkspaceStore { input_fingerprint: &str, ) -> Result { let plan = self.begin_worker_removal(workspace_id, plan_id, input_fingerprint)?; + if let Some(reconciliation) = self.subject_deletion_removal(workspace_id, &plan.worker)? + && reconciliation.request.operation_id == plan.operation_id + { + return Ok(PreparedWorkerRemoval { + runtime_request: reconciliation.request, + prior_failure_category: plan.failure_category.clone(), + plan, + }); + } let worker = self .with_conn(|conn| load_worker(conn, workspace_id, &plan.worker))? .ok_or_else(|| WorkerRetentionError::StalePlan { @@ -466,12 +480,18 @@ impl SqliteWorkspaceStore { WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state IN ('planned','executing','failed','succeeded') AND ( - state='succeeded' OR worker_updated_at=( + state='succeeded' OR EXISTS( + SELECT 1 FROM subject_deletion_removals d + WHERE d.workspace_id=worker_removal_operations.workspace_id + AND d.runtime_id=worker_removal_operations.runtime_id + AND d.worker_id=worker_removal_operations.worker_id + AND d.operation_id=worker_removal_operations.operation_id + ) OR worker_updated_at=( SELECT updated_at FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 ) ) - ORDER BY CASE WHEN EXISTS(SELECT 1 FROM subject_deletion_workers d WHERE d.workspace_id=worker_removal_operations.workspace_id AND d.runtime_id=worker_removal_operations.runtime_id AND d.worker_id=worker_removal_operations.worker_id AND d.prior_removal_operation_id=worker_removal_operations.operation_id) THEN 0 WHEN state='succeeded' THEN 1 ELSE 2 END, + ORDER BY CASE WHEN EXISTS(SELECT 1 FROM subject_deletion_workers d WHERE d.workspace_id=worker_removal_operations.workspace_id AND d.runtime_id=worker_removal_operations.runtime_id AND d.worker_id=worker_removal_operations.worker_id AND d.prior_removal_operation_id=worker_removal_operations.operation_id) THEN 0 WHEN EXISTS(SELECT 1 FROM subject_deletion_removals d WHERE d.workspace_id=worker_removal_operations.workspace_id AND d.runtime_id=worker_removal_operations.runtime_id AND d.worker_id=worker_removal_operations.worker_id AND d.operation_id=worker_removal_operations.operation_id) THEN 1 WHEN state='succeeded' THEN 2 ELSE 3 END, created_at DESC,operation_id DESC LIMIT 1", params![workspace_id, worker.runtime_id, worker.worker_id], |row| row.get::<_, String>(0), @@ -489,6 +509,15 @@ impl SqliteWorkspaceStore { self.with_conn(|conn| require_current_subject_deletion_plan(conn, &plan)) .map_err(map_error)?; let prior_failure_category = plan.failure_category.clone(); + if let Some(reconciliation) = self.subject_deletion_removal(workspace_id, &plan.worker)? + && reconciliation.request.operation_id == plan.operation_id + { + return Ok(Some(PreparedWorkerRemoval { + runtime_request: reconciliation.request, + prior_failure_category, + plan, + })); + } let worker_id = plan .worker .worker_id @@ -698,8 +727,12 @@ impl SqliteWorkspaceStore { return Err(StoreError::InvalidInput("Erased removal receipt requires matching source and physical Runtime erasure acknowledgement".into())); } } + let reconciliation=crate::store::subject_deletion_removal_matches(&tx,workspace_id,&plan.worker,&plan.operation_id,&plan.input_fingerprint)?; 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)));} + // A pinned pre-deletion request is reconciled unchanged under the permanent + // Subject/Runtime exclusion. Its Worker identity cannot be reused. Metadata + // version movement must not discard an already-applied canonical result. + if !reconciliation && 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))); } let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3 UNION ALL SELECT 1 FROM ticket_assignment_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND claim_state='pending' AND action IN ('assign','reassign'))",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|row|row.get(0))?; if assigned { return Err(StoreError::InvalidInput(format!("stale:{}:unfinished work added",plan.plan_id))); } @@ -771,7 +804,7 @@ impl SqliteWorkspaceStore { 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])?; + let deleted=tx.execute("DELETE FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND (updated_at=?4 OR ?5)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,plan.worker_updated_at,reconciliation])?; tx.execute("UPDATE worker_create_reservations SET state='removed',updated_at=?4 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state IN ('reserved','created')",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,now])?; 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)?; @@ -1109,6 +1142,18 @@ fn require_current_subject_deletion_plan( c: &Connection, plan: &WorkerRemovalPlan, ) -> crate::Result<()> { + if crate::store::subject_deletion_removal_matches( + c, + &plan.workspace_id, + &plan.worker, + &plan.operation_id, + &plan.input_fingerprint, + )? { + // An unfinished canonical request is pinned separately from successes. Only + // its exact original input may reconcile; any required Job copy erasure is + // a subsequent physical operation and never changes this historical result. + return Ok(()); + } if plan.state == WorkerRemovalPlanState::Succeeded && crate::store::subject_deletion_prior_removal_matches( c, @@ -2831,7 +2876,7 @@ mod tests { #[test] fn subject_deletion_snapshot_overrides_policy_without_changing_normal_removal() { let store = setup(); - let ordinary = store.plan_worker_removal(&req(), &inv()).unwrap(); + let ordinary = setup().plan_worker_removal(&req(), &inv()).unwrap(); store .prepare_subject_deletion("w", "subject", "delete", &[req().worker.clone()]) .unwrap(); @@ -2842,12 +2887,8 @@ mod tests { assert_eq!(body.diagnostics_disposition, DiagnosticsDisposition::Retain); assert!(body.diagnostics_retention_seconds.is_some()); assert_ne!(body.policy_id, ordinary.policy_id); - // A saved plan that would discard diagnostics cannot silently stand in - // for the Subject conversation archive plan. - assert!(matches!( - store.begin_worker_removal("w", &ordinary.plan_id, &ordinary.input_fingerprint), - Err(WorkerRetentionError::StalePlan { .. }) - )); + // A new deletion-owned plan uses the archival policy; independently saved + // old requests are reconciled separately, without borrowing this policy. let ordinary_store = setup(); assert_eq!( ordinary_store @@ -2859,10 +2900,10 @@ mod tests { } #[test] - fn subject_deletion_job_purge_rejects_recovery_of_old_executing_archive() { + fn subject_deletion_pins_old_archive_input_without_counting_it_as_job_purge() { let store = setup(); let old = store.plan_worker_removal(&req(), &inv()).unwrap(); - store + let original = store .prepare_worker_removal_execution("w", &old.plan_id, &old.input_fingerprint) .unwrap(); store @@ -2877,14 +2918,21 @@ mod tests { Ok(()) }) .unwrap(); - assert!(matches!( - store.recover_worker_removal_execution("w", &req().worker), - Err(WorkerRetentionError::StalePlan { .. }) - )); - assert!(matches!( - store.plan_worker_removal(&req(), &inv()), - Err(WorkerRetentionError::StalePlan { .. }) - )); + let recovered = store + .recover_worker_removal_execution("w", &req().worker) + .unwrap() + .unwrap(); + assert_eq!(recovered.runtime_request, original.runtime_request); + let prepared = store + .prepare_worker_removal_execution("w", &old.plan_id, &old.input_fingerprint) + .unwrap(); + assert_eq!(prepared.runtime_request, original.runtime_request); + assert!(store.plan_worker_removal(&req(), &inv()).is_err()); + assert!( + store + .erase_subject_deleted_jobs("w", "subject", "delete") + .is_err() + ); let fresh = store .with_conn(|c| load_removal_policy(c, "w", &req().worker, false)) .unwrap() diff --git a/crates/workspace-server/src/server/subjektiv_deletion.rs b/crates/workspace-server/src/server/subjektiv_deletion.rs index 1c380eee6..9063f3b90 100644 --- a/crates/workspace-server/src/server/subjektiv_deletion.rs +++ b/crates/workspace-server/src/server/subjektiv_deletion.rs @@ -182,6 +182,45 @@ async fn remove_worker( ) .into()); } + if let Some(reconciliation) = api + .config_store + .subject_deletion_removal(api.workspace_id(), worker)? + { + if reconciliation.completed { + return Ok(()); + } + // First reconcile the original canonical result, without changing any old + // disposition, fingerprint or success snapshot. A lost Runtime response or + // Backend commit remains unknown until that same child succeeds. + execute_removal(api, operation, worker).await?; + let result = api + .runtime + .execute_worker_retention(worker, reconciliation.request.clone()) + .map_err(RuntimeRegistryError::into_error)?; + let purge = if reconciliation.is_job { + Some( + api.runtime + .purge_subject_job_removal( + &worker.runtime_id, + request, + reconciliation.request.clone(), + ) + .map_err(RuntimeRegistryError::into_error)?, + ) + } else { + None + }; + // The extra physical Job erasure is a distinct receipt. An old Archive + // success is never counted as a Purge, nor pinned as an older success later. + api.config_store.confirm_subject_deletion_removal( + api.workspace_id(), + worker, + &reconciliation.request, + &result, + purge.as_ref(), + )?; + return Ok(()); + } // Replay the canonical child even when physical removal already happened: // its retained receipt can finish the Backend catalog midpoint. The Runtime // fence above also prevents an older create from arriving after that removal. diff --git a/crates/workspace-server/src/server/tests/subject_deletion_tests.rs b/crates/workspace-server/src/server/tests/subject_deletion_tests.rs index 298eea8a0..8ddca16ad 100644 --- a/crates/workspace-server/src/server/tests/subject_deletion_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_deletion_tests.rs @@ -727,3 +727,473 @@ async fn unmaterialized_body_and_job_reservations_finish_only_with_runtime_stora } } } + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn old_job_removal_reconciles_exact_input_then_erases_copies_across_http_retry_and_restart() { + use worker_runtime::retention::{DiagnosticsDisposition, SessionDisposition}; + for phase in [ + "planned", + "executing", + "failed", + "response_lost", + "catalog_failure", + "purge_ack_commit_failure", + ] { + let dir = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(dir.path()).await; + let store = open_subjektiv_store(&api).unwrap(); + let subject = store + .create_subject( + crate::subjektiv::SubjectRole::new(format!("old plan {phase}")).unwrap(), + ) + .unwrap(); + let other = store + .create_subject_with_behavior( + crate::subjektiv::SubjectRole::new("keep other").unwrap(), + "untouched".into(), + ) + .unwrap(); + let candidate = serde_json::from_value(json!({ + "schema_version": 2, "id": "old-job-candidate", "extract_run_id": "extract", + "source": {"segment_id":"segment", "range":[1,2]}, "kind":"constraint", + "claim":"private candidate", "why_useful":"erase after explicit deletion", + "evidence": [], "source_refs": [] + })) + .unwrap(); + store + .stage_candidate(crate::subjektiv::SubjectStagingRecord::attach( + &subject.id, + candidate, + )) + .unwrap(); + start_subjektiv_staging_consolidation( + api.clone(), + &subject.id, + MemoryConsolidateStagingOperation { force: true }, + ) + .await + .unwrap(); + let job = api + .store + .find_active_backend_job_for_resource( + TEST_WORKSPACE_ID, + &crate::backend_job::subjektiv_serialization_key(&subject.id), + ) + .unwrap() + .unwrap(); + let attempt = api + .store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &job.request.job_id, + &crate::backend_job::attempt_id(&job.request.job_id, job.current_attempt), + ) + .unwrap() + .unwrap(); + let worker = attempt.worker.clone().unwrap(); + seed_worker_session_for_cleanup(dir.path(), &worker); + api.runtime + .stop_worker( + &worker, + WorkerLifecycleRequest { + reason: Some("Prior independent removal".into()), + ticket_assignment: None, + }, + ) + .unwrap(); + let diagnostic = dir + .path() + .join(".test-embedded-runtime-store/workers") + .join(&worker.worker_id) + .join("runs/old/worker.out.log"); + std::fs::create_dir_all(diagnostic.parent().unwrap()).unwrap(); + std::fs::write(&diagnostic, "private old Job diagnostic").unwrap(); + let policy = api + .config_store + .worker_retention_policy(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(); + api.config_store + .update_worker_retention_policy( + TEST_WORKSPACE_ID, + &policy.policy_digest, + &crate::retention::WorkerRetentionPolicyUpdate { + policy_id: "prior-job-archive-policy".into(), + session_disposition: SessionDisposition::Archive, + metadata_disposition: crate::retention::MetadataDisposition::Tombstone, + archive_retention: crate::retention::ArchiveRetention::ForSeconds { + seconds: 600, + }, + diagnostics_disposition: DiagnosticsDisposition::Retain, + diagnostics_retention_seconds: Some(600), + }, + ) + .unwrap(); + let inventory = api.runtime.worker_retention_inventory(&worker).unwrap(); + assert!(inventory.diagnostics_bytes > 0); + let old = api + .config_store + .plan_worker_removal( + &crate::retention::WorkerRemovalPlanRequest { + workspace_id: TEST_WORKSPACE_ID.into(), + worker: worker.clone(), + reason: "Previously requested retention".into(), + journal_erasure: None, + }, + &inventory, + ) + .unwrap(); + assert_eq!(old.session_disposition, SessionDisposition::Archive); + let saved = if phase == "planned" { + api.config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &worker) + .unwrap() + .unwrap() + } else { + api.config_store + .prepare_worker_removal_execution( + TEST_WORKSPACE_ID, + &old.plan_id, + &old.input_fingerprint, + ) + .unwrap() + }; + if phase == "failed" { + api.config_store + .fail_worker_removal( + TEST_WORKSPACE_ID, + &old.operation_id, + &old.input_fingerprint, + "old_unknown_outcome", + ) + .unwrap(); + } + if phase == "catalog_failure" { + api.config_store.with_conn(|c| { c.execute_batch("CREATE TRIGGER injected_old_catalog_failure BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'catalog failure'); END;")?; Ok(()) }).unwrap(); + } + let old_result = if matches!(phase, "response_lost" | "catalog_failure") { + let result = api + .runtime + .execute_worker_retention(&worker, saved.runtime_request.clone()) + .unwrap(); + if phase == "catalog_failure" { + assert!( + api.config_store + .commit_worker_removal( + TEST_WORKSPACE_ID, + &old.operation_id, + &old.input_fingerprint, + &result + ) + .is_err() + ); + } + Some(result) + } else { + None + }; + if phase == "purge_ack_commit_failure" { + api.config_store.with_conn(|c| { c.execute_batch("CREATE TRIGGER injected_purge_ack_commit_failure BEFORE UPDATE OF completed ON subject_deletion_removals WHEN NEW.completed=1 BEGIN SELECT RAISE(ABORT,'purge receipt commit failure'); END;")?; Ok(()) }).unwrap(); + } + // Force a midpoint AFTER domain/snapshot commit but BEFORE any deletion-owned + // Runtime effect. It must retain exactly the original canonical input. + *execution.subject_delete_failure.lock().unwrap() = Some("shutdown outcome unknown".into()); + let owner = + seed_test_workspace_owner_token(api.store.as_ref(), &format!("old-removal-{phase}")) + .await; + let (status, pending) = delete_request( + &api, + &subject.id, + "same-delete", + Some(&owner), + TEST_WORKSPACE_ID, + ) + .await; + assert_eq!(status, StatusCode::OK, "{phase}: {pending}"); + assert_eq!(pending["completed"], false, "{phase}"); + let pinned = api + .config_store + .subject_deletion_removal(TEST_WORKSPACE_ID, &worker) + .unwrap() + .unwrap(); + assert_eq!(pinned.request, saved.runtime_request, "{phase}"); + assert!(!pinned.completed); + assert!(pinned.is_job); + assert!( + api.config_store + .erase_subject_deleted_jobs(TEST_WORKSPACE_ID, &subject.id, "same-delete") + .is_err() + ); + *execution.subject_delete_failure.lock().unwrap() = None; + let mut reopened = api.clone(); + let reopened_store = + Arc::new(SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap()); + reopened.config_store = reopened_store.clone(); + reopened.store = reopened_store; + if matches!(phase, "catalog_failure" | "purge_ack_commit_failure") { + let (status, pending) = delete_request( + &reopened, + &subject.id, + "same-delete", + Some(&owner), + TEST_WORKSPACE_ID, + ) + .await; + assert_eq!(status, StatusCode::OK, "{phase}: {pending}"); + assert_eq!(pending["completed"], false, "{phase}"); + assert!( + !store + .subject_deletion(&subject.id) + .unwrap() + .unwrap() + .completed + ); + assert!( + reopened + .store + .get_backend_job(TEST_WORKSPACE_ID, &job.request.job_id) + .unwrap() + .is_some() + ); + reopened + .config_store + .with_conn(|c| { + c.execute_batch(if phase == "catalog_failure" { + "DROP TRIGGER injected_old_catalog_failure;" + } else { + "DROP TRIGGER injected_purge_ack_commit_failure;" + })?; + Ok(()) + }) + .unwrap(); + } + // Exercise the same production restart recovery that starts from the saved + // domain operation, then HTTP replay of the same full request. + let recovery = reopened.clone(); + let handle = tokio::runtime::Handle::current(); + reopened + .host_executors + .host() + .unwrap() + .run(move || handle.block_on(subjektiv_deletion::recover(&recovery))) + .await + .unwrap(); + let (status, done) = delete_request( + &reopened, + &subject.id, + "same-delete", + Some(&owner), + TEST_WORKSPACE_ID, + ) + .await; + assert_eq!(status, StatusCode::OK, "{phase}: {done}"); + assert_eq!(done["completed"], true, "{phase}: {done}"); + let reconciled = reopened + .config_store + .subject_deletion_removal(TEST_WORKSPACE_ID, &worker) + .unwrap() + .unwrap(); + assert!(reconciled.completed); + assert_eq!(reconciled.request, saved.runtime_request); + let canonical = reopened + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &worker) + .unwrap() + .unwrap(); + assert_eq!( + canonical.plan.state, + crate::retention::WorkerRemovalPlanState::Succeeded + ); + assert_eq!(canonical.runtime_request, saved.runtime_request); + assert_eq!( + canonical.plan.session_disposition, + SessionDisposition::Archive, + "old result was not relabeled Purge" + ); + let result = reopened + .runtime + .execute_worker_retention(&worker, saved.runtime_request.clone()) + .unwrap(); + if let Some(old_result) = old_result { + assert_eq!(result, old_result); + } + assert_eq!(result.session_disposition, SessionDisposition::Archive); + assert!(result.source_removed); + let archive = result.archive.as_ref().unwrap(); + let root = dir.path().join(".test-embedded-runtime-store"); + assert!( + !root + .join("archives/workers") + .join(&archive.archive_id) + .join("session") + .exists(), + "new Job archive copy survives {phase}" + ); + if result.diagnostics_disposition == DiagnosticsDisposition::Retain { + assert!( + !root + .join("archives/diagnostics") + .join(&old.operation_id) + .exists(), + "Job diagnostics survive {phase}" + ); + } + assert!( + reopened + .store + .get_backend_job(TEST_WORKSPACE_ID, &job.request.job_id) + .unwrap() + .is_none() + ); + assert!( + reopened + .store + .current_worker_singleton_owner( + TEST_WORKSPACE_ID, + &crate::backend_job::subjektiv_serialization_key(&subject.id) + ) + .unwrap() + .is_none() + ); + assert!(store.subject(&subject.id).unwrap().is_none()); + assert_eq!( + store.subject(&other.id).unwrap().unwrap().behavior_md, + "untouched" + ); + let (status, replay) = delete_request( + &reopened, + &subject.id, + "same-delete", + Some(&owner), + TEST_WORKSPACE_ID, + ) + .await; + assert_eq!(status, StatusCode::OK); + assert_eq!(done, replay); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn old_body_archive_reconciles_unchanged_and_remains_retained_after_subject_deletion() { + let dir = tempfile::tempdir().unwrap(); + init_clean_git_workspace(dir.path()); + let (api, execution) = test_api_with_recording_backend(dir.path()).await; + let store = open_subjektiv_store(&api).unwrap(); + let subject = store + .create_subject(crate::subjektiv::SubjectRole::new("old body archive").unwrap()) + .unwrap(); + let Json(created) = create_workspace_worker( + State(api.clone()), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(), + display_name: "Existing body removal".into(), + singleton_key: None, + profile: Some("builtin:companion".into()), + ticket_assignment: None, + initial_submit: vec![], + workdir_attachments: vec![], + feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest { + subjektiv: Some(server_api::SubjektivWorkerConnectionRequest { + subject_id: subject.id.clone(), + }), + }, + control_operation_id: None, + execution_grant: None, + }), + ) + .await + .unwrap(); + let worker = RuntimeWorkerRef::new(created.runtime_id, created.worker_id); + seed_worker_session_for_cleanup(dir.path(), &worker); + api.runtime + .stop_worker( + &worker, + WorkerLifecycleRequest { + reason: Some("Earlier removal".into()), + ticket_assignment: None, + }, + ) + .unwrap(); + let inventory = api.runtime.worker_retention_inventory(&worker).unwrap(); + let old = api + .config_store + .plan_worker_removal( + &crate::retention::WorkerRemovalPlanRequest { + workspace_id: TEST_WORKSPACE_ID.into(), + worker: worker.clone(), + reason: "Earlier authorized archive".into(), + journal_erasure: None, + }, + &inventory, + ) + .unwrap(); + let saved = api + .config_store + .prepare_worker_removal_execution(TEST_WORKSPACE_ID, &old.plan_id, &old.input_fingerprint) + .unwrap(); + let result = api + .runtime + .execute_worker_retention(&worker, saved.runtime_request.clone()) + .unwrap(); + // Simulate a lost result before canonical Backend commit. Deletion must pin an + // unfinished request, not retrospectively classify its archive as a prior success. + *execution.subject_delete_failure.lock().unwrap() = Some("unknown shutdown".into()); + let owner = seed_test_workspace_owner_token(api.store.as_ref(), "old-body-removal").await; + let (status, pending) = delete_request( + &api, + &subject.id, + "same-body-delete", + Some(&owner), + TEST_WORKSPACE_ID, + ) + .await; + assert_eq!(status, StatusCode::OK, "{pending}"); + assert_eq!(pending["completed"], false); + let snapshot = api + .config_store + .subject_deletion_removal(TEST_WORKSPACE_ID, &worker) + .unwrap() + .unwrap(); + assert!(!snapshot.is_job && !snapshot.completed); + assert_eq!(snapshot.request, saved.runtime_request); + *execution.subject_delete_failure.lock().unwrap() = None; + let (status, done) = delete_request( + &api, + &subject.id, + "same-body-delete", + Some(&owner), + TEST_WORKSPACE_ID, + ) + .await; + assert_eq!(status, StatusCode::OK, "{done}"); + assert_eq!(done["completed"], true, "{done}"); + let completed = api + .config_store + .subject_deletion_removal(TEST_WORKSPACE_ID, &worker) + .unwrap() + .unwrap(); + assert!(completed.completed); + assert_eq!(completed.request, saved.runtime_request); + let replay = api + .runtime + .execute_worker_retention(&worker, saved.runtime_request) + .unwrap(); + assert_eq!(replay, result); + let archive = result.archive.unwrap(); + assert!( + dir.path() + .join(".test-embedded-runtime-store/archives/workers") + .join(&archive.archive_id) + .join("session") + .is_dir() + ); + assert!(api.config_store.with_conn(|c| Ok(c.query_row("SELECT prior_removal_operation_id IS NULL FROM subject_deletion_workers WHERE workspace_id=?1 AND subject_id=?2 AND runtime_id=?3 AND worker_id=?4", rusqlite::params![TEST_WORKSPACE_ID,subject.id,worker.runtime_id,worker.worker_id], |r|r.get::<_,bool>(0))?)).unwrap()); + assert!( + store + .subject_deletion(&subject.id) + .unwrap() + .unwrap() + .completed + ); +} diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 3fb7ea555..815583e7d 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -95,7 +95,8 @@ use crate::{Error, Result}; #[path = "subject_deletion_store.rs"] mod subject_deletion_store; pub(crate) use subject_deletion_store::{ - prior_removal_matches as subject_deletion_prior_removal_matches, + SubjectDeletionRemoval, prior_removal_matches as subject_deletion_prior_removal_matches, + removal_reconciliation_matches as subject_deletion_removal_matches, worker_kind as subject_deletion_worker_kind, }; @@ -16255,7 +16256,8 @@ fn current_worker_singleton_owner( let owner = conn .query_row( "SELECT owner.runtime_id, owner.worker_id, owner.generation, \ - CASE WHEN substr(owner.singleton_key,1,10)='subjektiv:' AND reservation.state='removed' \ + CASE WHEN reservation.state='removed' AND EXISTS(SELECT 1 FROM subject_deletion_removals d WHERE d.workspace_id=owner.workspace_id AND d.runtime_id=owner.runtime_id AND d.worker_id=owner.worker_id AND owner.singleton_key IN ('subjektiv:'||d.subject_id,'subjektiv-consolidation:'||d.subject_id)) THEN 'created' \ + WHEN substr(owner.singleton_key,1,10)='subjektiv:' AND reservation.state='removed' \ AND NOT EXISTS(SELECT 1 FROM subject_worker_releases r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.singleton_key=owner.singleton_key AND r.singleton_generation=owner.generation AND r.completed=1) \ AND NOT EXISTS(SELECT 1 FROM subject_create_recoveries r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.singleton_key=owner.singleton_key AND r.singleton_generation=owner.generation AND r.create_fingerprint=reservation.create_fingerprint AND r.completed=1) \ AND NOT EXISTS(SELECT 1 FROM worker_removal_operations r JOIN subject_deletion_workers d ON d.workspace_id=r.workspace_id AND d.runtime_id=r.runtime_id AND d.worker_id=r.worker_id WHERE r.workspace_id=owner.workspace_id AND r.runtime_id=owner.runtime_id AND r.worker_id=owner.worker_id AND r.state='succeeded' AND owner.singleton_key='subjektiv:'||d.subject_id AND (d.prior_removal_operation_id=r.operation_id OR (d.prior_removal_operation_id IS NULL AND d.is_job=0 AND r.session_disposition='archive' AND r.archive_retention_kind='forever' AND r.metadata_disposition='tombstone'))) \ @@ -16278,7 +16280,15 @@ fn current_worker_singleton_owner( ) .optional()?; owner - .map(|(runtime_id, worker_id, generation, state)| { + .map(|(runtime_id, worker_id, generation, mut state)| { + let worker = RuntimeWorkerRef::new(runtime_id, worker_id); + if let Some(subject) = singleton_key.strip_prefix("subjektiv:").or_else(|| singleton_key.strip_prefix("subjektiv-consolidation:")) + && state.as_deref() == Some("created") + && subject_deletion_store::completed_removal_matches(conn, workspace_id, subject, &worker)? + { + let removed: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM worker_create_reservations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state='removed')",params![workspace_id,worker.runtime_id,worker.worker_id],|r|r.get(0))?; + if removed { state = Some("removed".into()); } + } let generation = generation.try_into().map_err(|_| { Error::Store(format!( "Worker singleton key {singleton_key:?} has an invalid generation" @@ -16288,7 +16298,7 @@ fn current_worker_singleton_owner( WorkerSingletonLease { key: singleton_key.to_string(), generation, - worker: RuntimeWorkerRef::new(runtime_id, worker_id), + worker, }, state, )) @@ -26903,7 +26913,7 @@ INSERT INTO worker_registry ( .query_map([],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?)))?.collect::>>()?; for (name,table) in triggers {conn.execute_batch(&format!("DROP TRIGGER {name}; CREATE TRIGGER {name} BEFORE DELETE ON {table} BEGIN SELECT RAISE(ABORT,'Notification evidence is immutable'); END;"))?;} for name in ["subject_deleted_job_insert","subject_deleted_job_result","subject_deleted_attempt_insert","subject_deleted_attempt_result","subject_deleted_worker_reservation","subject_deleted_notification_source","subject_deleted_notification_payload","subject_deleted_notification_send","subject_deletion_worker_prior_insert","subject_deletion_worker_identity_immutable"] {conn.execute_batch(&format!("DROP TRIGGER {name}"))?;} - conn.execute_batch("DROP TABLE subject_deletion_allocations; DROP TABLE subject_deletion_jobs; DROP TABLE subject_deletion_workers; DROP TABLE subject_deletions;")?; + conn.execute_batch("DROP TABLE subject_deletion_removals; DROP TABLE subject_deletion_allocations; DROP TABLE subject_deletion_jobs; DROP TABLE subject_deletion_workers; DROP TABLE subject_deletions;")?; conn.execute_batch("DROP TRIGGER backend_job_input_retirement; DROP TRIGGER backend_job_identity_immutable; ALTER TABLE backend_jobs DROP COLUMN input_disposition; DROP TRIGGER passkey_metadata_insert; diff --git a/crates/workspace-server/src/subject_deletion_schema.sql b/crates/workspace-server/src/subject_deletion_schema.sql index 610f29253..0d766bffd 100644 --- a/crates/workspace-server/src/subject_deletion_schema.sql +++ b/crates/workspace-server/src/subject_deletion_schema.sql @@ -35,6 +35,41 @@ BEGIN SELECT RAISE(ABORT,'Subject deletion Worker snapshot identity is immutable CREATE TRIGGER subject_deletion_original_absence_monotonic BEFORE UPDATE OF originals_absent ON subject_deletion_workers WHEN OLD.originals_absent=1 AND NEW.originals_absent=0 BEGIN SELECT RAISE(ABORT,'Subject deletion original absence proof is permanent'); END; +-- In-flight removals are distinct from successful prior receipts. Pin the exact +-- old Runtime request once; Subject deletion never rewrites its retention policy. +CREATE TABLE subject_deletion_removals ( + workspace_id TEXT NOT NULL, + subject_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + operation_id TEXT NOT NULL REFERENCES worker_removal_operations(operation_id), + input_fingerprint TEXT NOT NULL, + request_json TEXT NOT NULL CHECK(json_valid(request_json)), + completed INTEGER NOT NULL DEFAULT 0 CHECK(completed IN (0,1)), + job_purge_receipt_json TEXT CHECK(job_purge_receipt_json IS NULL OR json_valid(job_purge_receipt_json)), + PRIMARY KEY(workspace_id,subject_id,runtime_id,worker_id), + FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE, + FOREIGN KEY(workspace_id,subject_id,runtime_id,worker_id) + REFERENCES subject_deletion_workers(workspace_id,subject_id,runtime_id,worker_id) ON DELETE CASCADE +); +CREATE TRIGGER subject_deletion_removal_insert BEFORE INSERT ON subject_deletion_removals +WHEN NEW.completed!=0 OR NEW.job_purge_receipt_json IS NOT NULL OR NOT EXISTS( + SELECT 1 FROM subject_deletion_workers w JOIN worker_removal_operations r + ON r.workspace_id=w.workspace_id AND r.runtime_id=w.runtime_id AND r.worker_id=w.worker_id + WHERE w.workspace_id=NEW.workspace_id AND w.subject_id=NEW.subject_id + AND w.runtime_id=NEW.runtime_id AND w.worker_id=NEW.worker_id AND w.prior_removal_operation_id IS NULL + AND r.operation_id=NEW.operation_id AND r.input_fingerprint=NEW.input_fingerprint + AND r.state IN ('planned','executing','failed') + AND json_extract(NEW.request_json,'$.operation_id')=r.operation_id + AND json_extract(NEW.request_json,'$.input_fingerprint')=r.input_fingerprint) +BEGIN SELECT RAISE(ABORT,'Subject deletion reconciliation must pin an unfinished canonical removal'); END; +CREATE TRIGGER subject_deletion_removal_identity_immutable BEFORE UPDATE OF + workspace_id,subject_id,runtime_id,worker_id,operation_id,input_fingerprint,request_json ON subject_deletion_removals +BEGIN SELECT RAISE(ABORT,'Subject deletion removal request is immutable'); END; +CREATE TRIGGER subject_deletion_removal_completion_monotonic BEFORE UPDATE ON subject_deletion_removals +WHEN (OLD.completed=1 AND NEW.completed!=1) + OR (OLD.job_purge_receipt_json IS NOT NULL AND NEW.job_purge_receipt_json IS NOT OLD.job_purge_receipt_json) +BEGIN SELECT RAISE(ABORT,'Subject deletion removal receipt is permanent'); END; -- Job/attempt identities remain after payload erasure, preventing stale allocation -- or caller-selected Job-id reuse from recreating the deleted Subject's inputs. CREATE TABLE subject_deletion_jobs ( diff --git a/crates/workspace-server/src/subject_deletion_store.rs b/crates/workspace-server/src/subject_deletion_store.rs index 61db2bfb2..275f81bb9 100644 --- a/crates/workspace-server/src/subject_deletion_store.rs +++ b/crates/workspace-server/src/subject_deletion_store.rs @@ -5,14 +5,184 @@ //! All historical Job attempts and dispatch allocations are snapshotted, not just //! the current attempt. Pending/unknown work becomes failed/subject_deleted (not a //! successful Job). Originals in body/delegated Sessions are archived forever; -//! Existing canonical Worker removals and archives are outside this deletion: +//! Successful canonical Worker removals and archives are outside this deletion: //! prepare pins their exact successful receipt, regardless of its disposition. +//! Unfinished removals instead pin their immutable old execution request; exact +//! reconciliation and (for Jobs) additional physical erasure must complete first. //! Newly removed Job Sessions/diagnostics need canonical Purge before input, //! result, notification payloads and live Job/cache references can be erased. //! This DB has no separate Job content cache. Runtime caches are the canonical //! Worker removal's responsibility. Notification text already delivered into an //! ordinary conversation follows that conversation's archive lifecycle. use super::*; +use worker_runtime::retention::{ + DiagnosticsDisposition, WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult, +}; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct SubjectDeletionRemoval { + pub request: WorkerRetentionExecutionRequest, + pub completed: bool, + pub is_job: bool, +} + +fn decode_request(json: &str) -> Result { + serde_json::from_str(json).map_err(|e| Error::Store(e.to_string())) +} + +// Only source metadata comes from the saved Worker snapshot. Everything owned by +// the canonical plan is rechecked, even after its success/catalog removal. +fn canonical_request( + conn: &Connection, + operation: &str, + source_created_at: &str, + profile: Option<&str>, +) -> Result> { + let row: Option<(String, String)> = conn.query_row( + "SELECT json_object('operation_id',operation_id,'input_fingerprint',input_fingerprint, + 'archive_id',archive_id,'workspace_id',workspace_id,'source_runtime_id',runtime_id, + 'worker_id',worker_id,'source_created_at',?2,'removed_at',created_at, + 'effective_profile',?3,'retention_class',NULL,'policy_id',policy_id, + 'session_disposition',session_disposition,'diagnostics_disposition',diagnostics_disposition),state + FROM worker_removal_operations WHERE operation_id=?1", + params![operation,source_created_at,profile], |r| Ok((r.get(0)?,r.get(1)?)), + ).optional()?; + row.map(|(json, state)| Ok((decode_request(&json)?, state))) + .transpose() +} + +fn saved_removal( + conn: &Connection, + workspace: &str, + subject: Option<&str>, + worker: &RuntimeWorkerRef, +) -> Result> { + let rows = conn.prepare("SELECT d.request_json,d.completed,w.is_job FROM subject_deletion_removals d + JOIN subject_deletion_workers w USING(workspace_id,subject_id,runtime_id,worker_id) + WHERE d.workspace_id=?1 AND d.runtime_id=?2 AND d.worker_id=?3 AND (?4 IS NULL OR d.subject_id=?4) + ORDER BY d.subject_id")? + .query_map(params![workspace,worker.runtime_id,worker.worker_id,subject], |r| Ok((r.get::<_,String>(0)?,r.get::<_,bool>(1)?,r.get::<_,bool>(2)?)))? + .collect::>>()?; + let mut saved: Option = None; + for (json, completed, is_job) in rows { + let request = decode_request(&json)?; + if let Some(saved) = saved.as_mut() { + if saved.request != request { + return Err(Error::RepositoryConflict( + "Conflicting Subject deletion removal snapshots".into(), + )); + } + saved.completed &= completed; + saved.is_job |= is_job; + } else { + saved = Some(SubjectDeletionRemoval { + request, + completed, + is_job, + }); + } + } + Ok(saved) +} + +pub(crate) fn removal_reconciliation_matches( + conn: &Connection, + workspace: &str, + worker: &RuntimeWorkerRef, + operation: &str, + fp: &str, +) -> Result { + let Some(saved) = saved_removal(conn, workspace, None, worker)? else { + return Ok(false); + }; + let request = &saved.request; + if request.workspace_id != workspace + || request.source_runtime_id != worker.runtime_id + || request.worker_id.to_string() != worker.worker_id + || request.operation_id != operation + || request.input_fingerprint != fp + { + return Ok(false); + } + Ok(canonical_request( + conn, + operation, + &request.source_created_at, + request.effective_profile.as_deref(), + )? + .is_some_and(|(canonical, state)| { + canonical == *request + && matches!( + state.as_str(), + "planned" | "executing" | "failed" | "succeeded" + ) + })) +} + +pub(super) fn completed_removal_matches( + conn: &Connection, + workspace: &str, + subject: &str, + worker: &RuntimeWorkerRef, +) -> Result { + let Some(saved) = saved_removal(conn, workspace, Some(subject), worker)? else { + return Ok(false); + }; + if !saved.completed + || !removal_reconciliation_matches( + conn, + workspace, + worker, + &saved.request.operation_id, + &saved.request.input_fingerprint, + )? + { + return Ok(false); + } + Ok(conn.query_row( + "SELECT EXISTS(SELECT 1 FROM subject_deletion_removals d + JOIN worker_removal_operations r ON r.operation_id=d.operation_id AND r.input_fingerprint=d.input_fingerprint + WHERE d.workspace_id=?1 AND d.subject_id=?2 AND d.runtime_id=?3 AND d.worker_id=?4 + AND r.state='succeeded' AND d.completed=1 AND (?5=0 OR ( + json_extract(d.job_purge_receipt_json,'$.workspace_id')=d.workspace_id + AND json_extract(d.job_purge_receipt_json,'$.worker_id')=d.worker_id + AND json_extract(d.job_purge_receipt_json,'$.operation_id')=d.operation_id + AND json_extract(d.job_purge_receipt_json,'$.input_fingerprint')=d.input_fingerprint)))", + params![workspace,subject,worker.runtime_id,worker.worker_id,saved.is_job], + |r| r.get(0), + )?) +} + +fn pin_removal( + conn: &Connection, + workspace: &str, + subject: &str, + worker: &RuntimeWorkerRef, +) -> Result<()> { + let operation: Option = conn.query_row("SELECT operation_id FROM worker_removal_operations + WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state IN ('planned','executing','failed') + AND (state IN ('executing','failed') OR worker_updated_at=(SELECT updated_at FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3)) + ORDER BY created_at DESC,operation_id DESC LIMIT 1", + params![workspace,worker.runtime_id,worker.worker_id],|r|r.get(0)).optional()?; + let Some(operation) = operation else { + return Ok(()); + }; + let source: Option<(String,Option)> = conn.query_row("SELECT created_at,profile FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3", + params![workspace,worker.runtime_id,worker.worker_id], |r| Ok((r.get(0)?,r.get(1)?))).optional()?; + let (created_at, profile) = source.ok_or_else(|| { + Error::RepositoryConflict( + "Unfinished Worker removal has no source identity to snapshot".into(), + ) + })?; + let (request, _) = canonical_request(conn, &operation, &created_at, profile.as_deref())? + .ok_or_else(|| { + Error::RepositoryConflict("Canonical removal disappeared while snapshotting".into()) + })?; + let json = serde_json::to_string(&request).map_err(|e| Error::Store(e.to_string()))?; + conn.execute("INSERT INTO subject_deletion_removals(workspace_id,subject_id,runtime_id,worker_id,operation_id,input_fingerprint,request_json) + VALUES(?1,?2,?3,?4,?5,?6,?7)",params![workspace,subject,worker.runtime_id,worker.worker_id,operation,request.input_fingerprint,json])?; + Ok(()) +} const MIGRATION_NAME: &str = "Durable Subject deletion snapshots and Job admission fences"; @@ -103,6 +273,10 @@ pub(super) fn verify(conn: &Connection) -> Result<()> { for name in [ "subject_deletions", "subject_deletion_workers", + "subject_deletion_removals", + "subject_deletion_removal_insert", + "subject_deletion_removal_identity_immutable", + "subject_deletion_removal_completion_monotonic", "subject_deletion_jobs", "subject_deletion_allocations", "subject_deleted_job_insert", @@ -251,6 +425,72 @@ fn worker_fence_request( } impl SqliteWorkspaceStore { + pub(crate) fn subject_deletion_removal( + &self, + workspace: &str, + worker: &RuntimeWorkerRef, + ) -> Result> { + self.with_conn(|conn| saved_removal(conn, workspace, None, worker)) + } + + /// A canonical success alone is not proof of additional Job erasure. Keep the + /// old receipt/dispositions intact and only complete this separate snapshot + /// after the exact Runtime result and (for Jobs) physical purge response. + pub(crate) fn confirm_subject_deletion_removal( + &self, + workspace: &str, + worker: &RuntimeWorkerRef, + request: &WorkerRetentionExecutionRequest, + result: &WorkerRetentionExecutionResult, + purge: Option<&runtime_api::SubjectJobRemovalPurgeResponse>, + ) -> Result<()> { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let saved = saved_removal(&tx,workspace,None,worker)? + .ok_or_else(|| Error::RepositoryConflict("No saved Subject deletion removal".into()))?; + if saved.request != *request + || !removal_reconciliation_matches(&tx,workspace,worker,&request.operation_id,&request.input_fingerprint)? + || !tx.query_row("SELECT state='succeeded' FROM worker_removal_operations WHERE operation_id=?1",[&request.operation_id],|r|r.get::<_,bool>(0))? + || result.operation_id != request.operation_id || result.input_fingerprint != request.input_fingerprint + || result.worker_id != request.worker_id || result.session_disposition != request.session_disposition + || result.diagnostics_disposition != request.diagnostics_disposition || !result.source_removed + || result.diagnostics_retained != (request.diagnostics_disposition == DiagnosticsDisposition::Retain) { + return Err(Error::RepositoryConflict("Subject deletion removal requires exact saved request and canonical Runtime success".into())); + } + let archive_matches = match (&result.archive, request.session_disposition) { + (Some(archive), worker_runtime::retention::SessionDisposition::Archive) => { + request.archive_id.as_deref() == Some(archive.archive_id.as_str()) + && archive.workspace_id == workspace + && archive.source_runtime_id == worker.runtime_id + && archive.source_worker_id == request.worker_id + && archive.operation_id == request.operation_id + && archive.input_fingerprint == request.input_fingerprint + && archive.policy_id == request.policy_id + && archive.source_created_at == request.source_created_at + } + (None, worker_runtime::retention::SessionDisposition::Purge) => true, + _ => false, + }; + if !archive_matches { + return Err(Error::RepositoryConflict("Subject deletion removal archive identity mismatch".into())); + } + if saved.is_job && purge.is_none() { + return Err(Error::RepositoryConflict("Subject Job removal requires an additional physical purge receipt".into())); + } + if let Some(purge) = purge { + if purge.workspace_id != workspace || purge.worker_id != request.worker_id + || purge.operation_id != request.operation_id || purge.input_fingerprint != request.input_fingerprint { + return Err(Error::RepositoryConflict("Subject Job physical purge receipt identity mismatch".into())); + } + } + let purge_json = purge.map(serde_json::to_string).transpose().map_err(|e| Error::Store(e.to_string()))?; + tx.execute("UPDATE subject_deletion_removals SET completed=1,job_purge_receipt_json=coalesce(job_purge_receipt_json,?4) + WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![workspace,worker.runtime_id,worker.worker_id,purge_json])?; + tx.commit()?; + Ok(()) + }) + } + pub(crate) fn subject_deletion_worker_request( &self, workspace: &str, @@ -316,6 +556,12 @@ impl SqliteWorkspaceStore { tx.execute("INSERT INTO subject_deletion_workers(workspace_id,subject_id,runtime_id,worker_id,is_job,prior_removal_operation_id) VALUES(?1,?2,?3,?4,?5,?6) ON CONFLICT(workspace_id,subject_id,runtime_id,worker_id) DO UPDATE SET is_job=max(is_job,excluded.is_job)",params![workspace,subject,worker.runtime_id,worker.worker_id,is_job,prior])?; } + // Deduplicate workers before pinning: an identity can be both a + // delegated Worker and a Job attempt; the final is_job wins. + let pending = tx.prepare("SELECT runtime_id,worker_id FROM subject_deletion_workers WHERE workspace_id=?1 AND subject_id=?2 AND prior_removal_operation_id IS NULL")? + .query_map(params![workspace,subject],|r|Ok(RuntimeWorkerRef::new(r.get::<_,String>(0)?,r.get::<_,String>(1)?)))? + .collect::>>()?; + for worker in pending { pin_removal(&tx,workspace,subject,&worker)?; } let now = chrono::Utc::now().to_rfc3339(); tx.execute("UPDATE backend_job_attempts SET state='failed',failure_category='subject_deleted',failure_detail=NULL,completed_at=?3,updated_at=?3, worker_cleanup_state=CASE WHEN worker_id IS NULL THEN 'completed' ELSE 'pending' END, @@ -362,6 +608,12 @@ impl SqliteWorkspaceStore { || fence.create_fingerprint != request.create_fingerprint || fence.operation_id != request.operation_id { return Err(Error::RepositoryConflict("Subject deletion absence fence identity mismatch".into())); } + // An in-flight pin cannot be bypassed by UnknownWorker/absence or + // a different later canonical success, even if that success is Purge. + if saved_removal(conn,workspace,Some(subject),worker)?.is_some() { + if completed_removal_matches(conn,workspace,subject,worker)? { return Ok(()); } + return Err(Error::RepositoryConflict("Subject deletion removal reconciliation is incomplete".into())); + } if fence.originals_absent { let registered: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3)", params![workspace,worker.runtime_id,worker.worker_id], |r|r.get(0))?; if registered { return Err(Error::RepositoryConflict("Registered Worker requires canonical original retention".into())); } @@ -474,7 +726,19 @@ fn require_worker_receipts( subject: &str, jobs_only: bool, ) -> Result<()> { + let pinned = conn.prepare("SELECT d.runtime_id,d.worker_id FROM subject_deletion_removals d JOIN subject_deletion_workers w USING(workspace_id,subject_id,runtime_id,worker_id) + WHERE d.workspace_id=?1 AND d.subject_id=?2 AND (?3=0 OR w.is_job=1)")? + .query_map(params![workspace,subject,jobs_only],|r|Ok(RuntimeWorkerRef::new(r.get::<_,String>(0)?,r.get::<_,String>(1)?)))? + .collect::>>()?; + for worker in pinned { + if !completed_removal_matches(conn, workspace, subject, &worker)? { + return Err(Error::RepositoryConflict( + "Subject deletion removal reconciliation receipt is incomplete".into(), + )); + } + } let missing: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM subject_deletion_workers w WHERE w.workspace_id=?1 AND w.subject_id=?2 AND (?3=0 OR w.is_job=1) AND w.originals_absent=0 + AND NOT EXISTS(SELECT 1 FROM subject_deletion_removals d WHERE d.workspace_id=w.workspace_id AND d.subject_id=w.subject_id AND d.runtime_id=w.runtime_id AND d.worker_id=w.worker_id) AND NOT ( (w.prior_removal_operation_id IS NOT NULL AND EXISTS(SELECT 1 FROM worker_removal_operations r WHERE r.operation_id=w.prior_removal_operation_id AND r.workspace_id=w.workspace_id AND r.runtime_id=w.runtime_id AND r.worker_id=w.worker_id AND r.state='succeeded')) OR (w.prior_removal_operation_id IS NULL AND ( @@ -1108,6 +1372,8 @@ mod tests { "job-owner", Some("subjektiv-consolidation:subject"), ); + registry(&s, &job_worker); + registry(&s, &body); receipt(&s, &job_worker, "executing", "archive", "retain"); receipt(&s, &body, "failed", "purge", "purge"); s.prepare_subject_deletion("w", "subject", "delete", std::slice::from_ref(&body)) @@ -1134,8 +1400,11 @@ mod tests { c.execute("UPDATE worker_create_reservations SET state='removed' WHERE workspace_id='w'",[])?; Ok(()) }).unwrap(); - s.erase_subject_deleted_jobs("w", "subject", "delete") - .unwrap(); + assert!( + s.erase_subject_deleted_jobs("w", "subject", "delete") + .is_err(), + "replacement canonical Purge cannot bypass the pinned unfinished removal" + ); assert!( s.finish_subject_deletion_record("w", "subject", "delete") .is_err(), @@ -1143,6 +1412,389 @@ mod tests { ); } + fn in_flight(s: &SqliteWorkspaceStore, w: &RuntimeWorkerRef, state: &str, is_job: bool) { + registry(s, w); + reservation( + s, + w, + "pending", + Some(if is_job { + "subjektiv-consolidation:subject" + } else { + "subjektiv:subject" + }), + ); + receipt(s, w, state, "archive", "retain"); + s.with_conn(|c| { + c.execute("UPDATE worker_registry SET profile='builtin:old-profile' WHERE worker_id=?1",[&w.worker_id])?; + c.execute("UPDATE worker_removal_operations SET archive_id='old-archive',metadata_disposition='purge',archive_retention_kind='for_seconds',archive_retention_seconds=60 WHERE worker_id=?1",[&w.worker_id])?; + Ok(()) + }).unwrap(); + } + + fn execution_result( + request: &WorkerRetentionExecutionRequest, + ) -> WorkerRetentionExecutionResult { + WorkerRetentionExecutionResult { + operation_id: request.operation_id.clone(), + input_fingerprint: request.input_fingerprint.clone(), + worker_id: request.worker_id, + session_disposition: request.session_disposition, + diagnostics_disposition: request.diagnostics_disposition, + archive: request.archive_id.as_ref().map(|archive_id| { + worker_runtime::retention::WorkerSessionArchiveManifest { + schema_version: 1, + archive_id: archive_id.clone(), + workspace_id: request.workspace_id.clone(), + source_runtime_id: request.source_runtime_id.clone(), + source_worker_id: request.worker_id, + source_session_id: "old-session".into(), + source_active_segment_id: None, + segment_ids: vec![], + source_created_at: request.source_created_at.clone(), + removed_at: request.removed_at.clone(), + archived_at_unix_seconds: 0, + effective_profile: request.effective_profile.clone(), + retention_class: request.retention_class.clone(), + content_checksum_sha256: "fixture-checksum".into(), + content_bytes: 0, + content_file_count: 0, + policy_id: request.policy_id.clone(), + operation_id: request.operation_id.clone(), + input_fingerprint: request.input_fingerprint.clone(), + } + }), + source_removed: true, + diagnostics_retained: request.diagnostics_disposition == DiagnosticsDisposition::Retain, + } + } + + fn succeed(s: &SqliteWorkspaceStore, request: &WorkerRetentionExecutionRequest) { + // Isolated store proof fixture. HTTP/Runtime tests exercise canonical + // commit; here success alone must still leave reconciliation incomplete. + s.with_conn(|c| { + c.execute("UPDATE worker_removal_operations SET state='succeeded' WHERE operation_id=?1",[&request.operation_id])?; + c.execute("DELETE FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![request.workspace_id,request.source_runtime_id,request.worker_id.to_string()])?; + c.execute("UPDATE worker_create_reservations SET state='removed' WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![request.workspace_id,request.source_runtime_id,request.worker_id.to_string()])?; + Ok(()) + }).unwrap(); + } + + #[test] + fn unfinished_removal_pins_full_identity_and_requires_exact_success_before_release() { + let s = setup(); + let w = worker(71); + in_flight(&s, &w, "executing", false); + // An executing plan may have an older registry epoch. Its exact Runtime + // request is still authoritative; do not turn this into permanent stale. + s.with_conn(|c| { + c.execute( + "UPDATE worker_registry SET updated_at='later' WHERE worker_id=?1", + [&w.worker_id], + )?; + Ok(()) + }) + .unwrap(); + s.prepare_subject_deletion("w", "subject", "delete", &[]) + .unwrap(); + let saved = s.subject_deletion_removal("w", &w).unwrap().unwrap(); + let req = &saved.request; + assert!(!saved.completed); + assert!(!saved.is_job); + assert_eq!(req.source_created_at, NOW); + assert_eq!(req.removed_at, NOW); + assert_eq!( + req.effective_profile.as_deref(), + Some("builtin:old-profile") + ); + assert_eq!(req.retention_class, None); + assert_eq!(req.archive_id.as_deref(), Some("old-archive")); + assert_eq!(req.policy_id, "test"); + assert_eq!(saved_prior(&s, "subject", &w), None); + s.with_conn(|c| { + assert!(removal_reconciliation_matches(c,"w",&w,&req.operation_id,&req.input_fingerprint)?); + assert!(!removal_reconciliation_matches(c,"w",&w,&req.operation_id,"changed")?); + assert!(!removal_reconciliation_matches(c,"other",&w,&req.operation_id,&req.input_fingerprint)?); + assert!(c.execute("UPDATE subject_deletion_removals SET request_json=request_json WHERE workspace_id='w'",[]).is_err()); + Ok(()) + }).unwrap(); + let result = execution_result(req); + assert!( + s.confirm_subject_deletion_removal("w", &w, req, &result, None) + .is_err() + ); + succeed(&s, req); + assert!( + s.current_worker_singleton_owner("w", "subjektiv:subject") + .unwrap() + .is_some() + ); + assert!( + s.finish_subject_deletion_record("w", "subject", "delete") + .is_err() + ); + let fence_request = s + .subject_deletion_worker_request("w", "subject", "delete", &w) + .unwrap(); + let fence = runtime_api::SubjectDeleteWorkerResponse { + workspace_id: fence_request.workspace_id, + worker_id: fence_request.worker_id, + create_fingerprint: fence_request.create_fingerprint, + operation_id: fence_request.operation_id, + originals_absent: true, + }; + assert!( + s.confirm_subject_deletion_absence("w", "subject", "delete", &w, &fence) + .is_err() + ); + for (field, value) in [ + ("operation_id", serde_json::json!("changed")), + ("input_fingerprint", serde_json::json!("changed")), + ("archive_id", serde_json::json!("changed")), + ("workspace_id", serde_json::json!("other")), + ("source_runtime_id", serde_json::json!("other-runtime")), + ("worker_id", serde_json::json!(worker(72).worker_id)), + ("source_created_at", serde_json::json!("changed")), + ("removed_at", serde_json::json!("changed")), + ("effective_profile", serde_json::json!("changed")), + ("retention_class", serde_json::json!("changed")), + ("policy_id", serde_json::json!("changed")), + ("session_disposition", serde_json::json!("purge")), + ("diagnostics_disposition", serde_json::json!("purge")), + ] { + let mut changed = serde_json::to_value(req).unwrap(); + changed[field] = value; + let changed: WorkerRetentionExecutionRequest = serde_json::from_value(changed).unwrap(); + assert!( + s.confirm_subject_deletion_removal("w", &w, &changed, &result, None) + .is_err(), + "{field}" + ); + assert!( + !s.subject_deletion_removal("w", &w) + .unwrap() + .unwrap() + .completed + ); + } + for (field, value) in [ + ("operation_id", serde_json::json!("changed")), + ("input_fingerprint", serde_json::json!("changed")), + ("worker_id", serde_json::json!(worker(72).worker_id)), + ("session_disposition", serde_json::json!("purge")), + ("diagnostics_disposition", serde_json::json!("purge")), + ("source_removed", serde_json::json!(false)), + ("diagnostics_retained", serde_json::json!(false)), + ("archive", serde_json::Value::Null), + ] { + let mut changed = serde_json::to_value(&result).unwrap(); + changed[field] = value; + let changed = serde_json::from_value(changed).unwrap(); + assert!( + s.confirm_subject_deletion_removal("w", &w, req, &changed, None) + .is_err(), + "{field}" + ); + } + s.confirm_subject_deletion_removal("w", &w, req, &result, None) + .unwrap(); + assert_eq!( + saved_prior(&s, "subject", &w), + None, + "not a late prior success" + ); + assert!( + s.current_worker_singleton_owner("w", "subjektiv:subject") + .unwrap() + .is_none() + ); + s.confirm_subject_deletion_absence("w", "subject", "delete", &w, &fence) + .unwrap(); + s.erase_subject_deleted_jobs("w", "subject", "delete") + .unwrap(); + s.finish_subject_deletion_record("w", "subject", "delete") + .unwrap(); + s.with_conn(|c| { + assert!(removal_reconciliation_matches( + c, + "w", + &w, + &req.operation_id, + &req.input_fingerprint + )?); + c.execute( + "UPDATE worker_removal_operations SET policy_id='changed' WHERE operation_id=?1", + [&req.operation_id], + )?; + assert!(!completed_removal_matches(c, "w", "subject", &w)?); + Ok(()) + }) + .unwrap(); + assert!( + s.finish_subject_deletion_record("w", "subject", "delete") + .is_err() + ); + } + + #[test] + fn job_reconciliation_requires_additional_purge_and_survives_reopen_and_retry() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("snapshot.db"); + let s = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + seed(&s); + let w = worker(81); + in_flight(&s, &w, "failed", true); + let other = worker(82); + registry(&s, &other); + reservation(&s, &other, "other-allocation", Some("subjektiv:another")); + receipt(&s, &other, "executing", "archive", "retain"); + s.prepare_subject_deletion("w", "another", "other-delete", &[]) + .unwrap(); + let other_before = s.subject_deletion_removal("w", &other).unwrap(); + s.prepare_subject_deletion("w", "subject", "delete", &[]) + .unwrap(); + let req = s + .subject_deletion_removal("w", &w) + .unwrap() + .unwrap() + .request; + drop(s); + let s = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + s.prepare_subject_deletion("w", "subject", "delete", &[]) + .unwrap(); + assert_eq!( + s.subject_deletion_removal("w", &w) + .unwrap() + .unwrap() + .request, + req + ); + assert!(s.subject_deletion_removal("other", &w).unwrap().is_none()); + let result = execution_result(&req); + let purge = runtime_api::SubjectJobRemovalPurgeResponse { + workspace_id: "w".into(), + worker_id: req.worker_id, + operation_id: req.operation_id.clone(), + input_fingerprint: req.input_fingerprint.clone(), + }; + assert!( + s.confirm_subject_deletion_removal("w", &w, &req, &result, Some(&purge)) + .is_err(), + "failed is not success" + ); + succeed(&s, &req); + assert!( + s.current_worker_singleton_owner("w", "subjektiv-consolidation:subject") + .unwrap() + .is_some(), + "canonical success cannot release Job occupancy before physical purge" + ); + assert!( + s.confirm_subject_deletion_removal("w", &w, &req, &result, None) + .is_err() + ); + assert!( + s.erase_subject_deleted_jobs("w", "subject", "delete") + .is_err() + ); + for (field, value) in [ + ("workspace_id", serde_json::json!("other")), + ("worker_id", serde_json::json!(worker(82).worker_id)), + ("operation_id", serde_json::json!("changed")), + ("input_fingerprint", serde_json::json!("changed")), + ] { + let mut changed = serde_json::to_value(&purge).unwrap(); + changed[field] = value; + let changed = serde_json::from_value(changed).unwrap(); + assert!( + s.confirm_subject_deletion_removal("w", &w, &req, &result, Some(&changed)) + .is_err(), + "{field}" + ); + } + assert!( + !s.subject_deletion_removal("w", &w) + .unwrap() + .unwrap() + .completed + ); + s.confirm_subject_deletion_removal("w", &w, &req, &result, Some(&purge)) + .unwrap(); + drop(s); + let s = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + s.confirm_subject_deletion_removal("w", &w, &req, &result, Some(&purge)) + .unwrap(); + assert!( + s.subject_deletion_removal("w", &w) + .unwrap() + .unwrap() + .completed + ); + assert_eq!(saved_prior(&s, "subject", &w), None); + assert!( + s.current_worker_singleton_owner("w", "subjektiv-consolidation:subject") + .unwrap() + .is_none() + ); + s.with_conn(|c| { + let old:(String,String,String) = c.query_row("SELECT state,session_disposition,metadata_disposition FROM worker_removal_operations WHERE operation_id=?1",[&req.operation_id],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?)))?; + assert_eq!(old,("succeeded".into(),"archive".into(),"purge".into())); + assert!(c.execute("UPDATE subject_deletion_removals SET completed=0",[]).is_err()); + assert_eq!(c.query_row("SELECT count(*) FROM pragma_foreign_key_check",[],|r|r.get::<_,i64>(0))?,0); + Ok(()) + }).unwrap(); + assert_eq!( + s.subject_deletion_removal("w", &other).unwrap(), + other_before + ); + assert!( + s.current_worker_singleton_owner("w", "subjektiv:another") + .unwrap() + .is_some() + ); + s.erase_subject_deleted_jobs("w", "subject", "delete") + .unwrap(); + s.finish_subject_deletion_record("w", "subject", "delete") + .unwrap(); + } + + #[test] + fn planned_pin_uses_current_epoch_and_never_replaces_prior_success() { + for (state, current, expected) in [ + ("planned", true, true), + ("planned", false, false), + ("failed", false, true), + ] { + let s = setup(); + let w = worker(91); + in_flight(&s, &w, state, false); + if !current { + s.with_conn(|c| { + c.execute("UPDATE worker_registry SET updated_at='later'", [])?; + Ok(()) + }) + .unwrap(); + } + s.prepare_subject_deletion("w", "subject", "delete", &[]) + .unwrap(); + assert_eq!( + s.subject_deletion_removal("w", &w).unwrap().is_some(), + expected, + "{state}/{current}" + ); + } + let s = setup(); + let w = worker(92); + in_flight(&s, &w, "executing", true); + receipt(&s, &w, "succeeded", "archive", "retain"); + s.prepare_subject_deletion("w", "subject", "delete", &[]) + .unwrap(); + assert!(saved_prior(&s, "subject", &w).is_some()); + assert!(s.subject_deletion_removal("w", &w).unwrap().is_none()); + s.erase_subject_deleted_jobs("w", "subject", "delete") + .unwrap(); + } + #[test] fn saved_prior_receipt_must_remain_the_exact_success_not_a_replacement() { let s = setup(); diff --git a/crates/workspace-server/src/workspace_deletion.rs b/crates/workspace-server/src/workspace_deletion.rs index 3de7c89cb..6cea92adb 100644 --- a/crates/workspace-server/src/workspace_deletion.rs +++ b/crates/workspace-server/src/workspace_deletion.rs @@ -96,6 +96,7 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[ "worker_control_grants", // Snapshot children precede their fence parent and canonical Worker removal // receipts (subject_deletion_workers.prior_removal_operation_id). + "subject_deletion_removals", "subject_deletion_allocations", "subject_deletion_jobs", "subject_deletion_workers", @@ -1286,6 +1287,7 @@ mod tests { fn workspace_purge_removes_subject_deletion_snapshots_and_preserves_other_workspace() { let (store, owner, workspace_id) = setup(); let tables = [ + "subject_deletion_removals", "subject_deletion_allocations", "subject_deletion_jobs", "subject_deletion_workers", @@ -1299,9 +1301,11 @@ mod tests { .position(|name| *name == table) .unwrap() }; - for child in &tables[..3] { + for child in &tables[..4] { assert!(position(*child) < position("subject_deletions")); } + assert!(position("subject_deletion_removals") < position("subject_deletion_workers")); + assert!(position("subject_deletion_removals") < position("worker_removal_operations")); assert!(position("subject_deletion_workers") < position("worker_removal_operations")); for workspace in ["workspace-a", "workspace-b"] { store.with_conn(|conn| { @@ -1310,11 +1314,16 @@ mod tests { worker_updated_at,policy_id,policy_digest,session_disposition,metadata_disposition, archive_retention_kind,diagnostics_disposition,blockers_json,state,reason,created_at,updated_at) VALUES(?1,?1,'fingerprint',?1,'runtime','worker','1','policy','digest', - 'purge','purge','forever','purge','[]','succeeded','removed','1','1')", [workspace])?; + 'purge','purge','forever','purge','[]','executing','removed','1','1')", [workspace])?; conn.execute("INSERT INTO subject_deletions(workspace_id,subject_id,operation_id,completed,jobs_erased) VALUES(?1,'subject','subject-delete',1,1)", [workspace])?; conn.execute("INSERT INTO subject_deletion_workers(workspace_id,subject_id,runtime_id,worker_id,is_job,prior_removal_operation_id,originals_absent) - VALUES(?1,'subject','runtime','worker',1,?1,1)", [workspace])?; + VALUES(?1,'subject','runtime','worker',1,NULL,0)", [workspace])?; + conn.execute("INSERT INTO subject_deletion_removals(workspace_id,subject_id,runtime_id,worker_id,operation_id,input_fingerprint,request_json) + VALUES(?1,'subject','runtime','worker',?1,'fingerprint',json_object('operation_id',?1,'input_fingerprint','fingerprint'))", [workspace])?; + // Settle canonical removal so Workspace deletion's independent + // preflight is unblocked; the reconciliation row still needs purge. + conn.execute("UPDATE worker_removal_operations SET state='succeeded' WHERE operation_id=?1", [workspace])?; conn.execute("INSERT INTO subject_deletion_jobs VALUES(?1,'subject','job')", [workspace])?; conn.execute("INSERT INTO subject_deletion_allocations VALUES(?1,'subject','allocation')", [workspace])?; Ok(()) diff --git a/web/workspace/src/lib/workspace/memory/SubjectDelete.svelte b/web/workspace/src/lib/workspace/memory/SubjectDelete.svelte index b6a35b541..577dc135c 100644 --- a/web/workspace/src/lib/workspace/memory/SubjectDelete.svelte +++ b/web/workspace/src/lib/workspace/memory/SubjectDelete.svelte @@ -91,7 +91,7 @@ {#if phase === 'confirming'}
{ event.preventDefault(); void submit(); }}>

This is irreversible. This Subject, its behavior, Memory, change history, staging, Surface, Journal, permissions, delegations, Writer/Session associations, and Subject Job inputs/results will be deleted. Connected and delegated Workers and Jobs must stop before completion can be confirmed.

-

Original Worker/Session text, existing archives and Workdirs, Tickets, and other separately retained resources are not deleted. Minimal deletion retry/audit metadata is retained. Other Subjects and Workspaces are not affected.

+

Original Worker/Session text, existing archives and Workdirs, Tickets, and other separately retained resources are not deleted. Minimal deletion retry/audit metadata is retained. Any earlier Worker retention request still follows its original disposition. Other Subjects and Workspaces are not affected.