From da8313fa1a6a14c34e009fc8501ba067d657cd41 Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 11 Aug 2026 23:57:37 +0900 Subject: [PATCH] runtime: add revisioned worker retention authority --- crates/worker-runtime/src/lib.rs | 2 + crates/worker-runtime/src/retention.rs | 1184 +++++++++++++++++++++ crates/worker-runtime/src/runtime.rs | 135 +++ crates/workspace-server/src/hosts.rs | 3 + crates/workspace-server/src/lib.rs | 1 + crates/workspace-server/src/retention.rs | 1194 ++++++++++++++++++++++ crates/workspace-server/src/store.rs | 92 +- 7 files changed, 2599 insertions(+), 12 deletions(-) create mode 100644 crates/worker-runtime/src/retention.rs create mode 100644 crates/workspace-server/src/retention.rs diff --git a/crates/worker-runtime/src/lib.rs b/crates/worker-runtime/src/lib.rs index 43d226bf..001a102a 100644 --- a/crates/worker-runtime/src/lib.rs +++ b/crates/worker-runtime/src/lib.rs @@ -22,6 +22,8 @@ pub mod management; pub mod observation; pub mod profile_archive; pub mod resource; +#[cfg(feature = "fs-store")] +pub mod retention; mod runtime; pub mod worker_backend; pub mod working_directory; diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs new file mode 100644 index 00000000..e77a8372 --- /dev/null +++ b/crates/worker-runtime/src/retention.rs @@ -0,0 +1,1184 @@ +//! Runtime-owned execution of Backend-resolved Worker retention dispositions. +//! +//! This boundary deliberately accepts only stable ids and resolved dispositions. +//! Host paths and provider handles never cross it. + +use crate::error::RuntimeError; +use crate::identity::WorkerId; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::collections::BTreeSet; +use std::fs::{self, File, OpenOptions}; +use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +const ARCHIVE_SCHEMA_VERSION: u32 = 1; +const OPERATION_SCHEMA_VERSION: u32 = 1; +const RETENTION_LOCK: &str = ".worker-retention.lock"; +static NEXT_TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(1); + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum SessionDisposition { + Archive, + Purge, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum DiagnosticsDisposition { + Purge, + /// The Backend catalog owns the expiry. Runtime keeps only bounded stdout/stderr + /// evidence and never mixes it into the Session archive. + Retain, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerRetentionInventory { + pub workspace_id: String, + pub runtime_id: String, + pub worker_id: WorkerId, + pub run_generation: u64, + pub session_id: Option, + pub segment_ids: Vec, + pub session_bytes: u64, + pub diagnostics_bytes: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerRetentionExecutionRequest { + pub operation_id: String, + pub input_fingerprint: String, + pub archive_id: Option, + pub workspace_id: String, + pub source_runtime_id: String, + pub worker_id: WorkerId, + pub expected_run_generation: u64, + pub source_created_at: String, + pub removed_at: String, + pub effective_profile: Option, + pub retention_class: Option, + pub policy_id: String, + pub policy_revision: u64, + pub session_disposition: SessionDisposition, + pub diagnostics_disposition: DiagnosticsDisposition, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerSessionArchiveManifest { + pub schema_version: u32, + pub archive_id: String, + pub workspace_id: String, + pub source_runtime_id: String, + pub source_worker_id: WorkerId, + pub source_session_id: String, + pub segment_ids: Vec, + pub source_created_at: String, + pub removed_at: String, + pub archived_at_unix_seconds: u64, + pub effective_profile: Option, + pub retention_class: Option, + pub content_checksum_sha256: String, + pub content_bytes: u64, + pub content_file_count: u64, + pub policy_id: String, + pub policy_revision: u64, + pub operation_id: String, + pub input_fingerprint: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerRetentionExecutionResult { + pub operation_id: String, + pub input_fingerprint: String, + pub worker_id: WorkerId, + pub session_disposition: SessionDisposition, + pub diagnostics_disposition: DiagnosticsDisposition, + pub archive: Option, + pub source_removed: bool, + pub diagnostics_retained: bool, +} + +pub(crate) trait WorkerRetentionProvider: Send + Sync { + fn inventory( + &self, + workspace_id: &str, + runtime_id: &str, + worker_id: WorkerId, + run_generation: u64, + ) -> Result; + + fn execute( + &self, + request: &WorkerRetentionExecutionRequest, + ) -> Result; + + fn completed( + &self, + operation_id: &str, + input_fingerprint: &str, + ) -> Result, RuntimeError>; +} + +/// Filesystem provider for the canonical Runtime Worker aggregate. +#[derive(Clone, Debug)] +pub(crate) struct FsWorkerRetentionProvider { + runtime_root: PathBuf, +} + +impl FsWorkerRetentionProvider { + pub(crate) fn new(runtime_root: impl Into) -> Self { + Self { + runtime_root: runtime_root.into(), + } + } + + pub(crate) fn recover_after_source_removal( + &self, + request: &WorkerRetentionExecutionRequest, + ) -> Result { + if self.worker_dir(request.worker_id).exists() { + return Err(RuntimeError::WorkerNotFound { + worker_id: request.worker_id, + }); + } + self.execute(request) + } + + fn worker_dir(&self, worker_id: WorkerId) -> PathBuf { + self.runtime_root + .join("workers") + .join(worker_id.to_string()) + } + + fn operation_path(&self, operation_id: &str) -> Result { + validate_id("operation_id", operation_id)?; + Ok(self + .runtime_root + .join("retention") + .join("operations") + .join(format!("{operation_id}.json"))) + } + + fn archive_dir(&self, archive_id: &str) -> Result { + validate_id("archive_id", archive_id)?; + Ok(self + .runtime_root + .join("archives") + .join("workers") + .join(archive_id)) + } + + fn diagnostics_dir(&self, operation_id: &str) -> Result { + validate_id("operation_id", operation_id)?; + Ok(self + .runtime_root + .join("archives") + .join("diagnostics") + .join(operation_id)) + } +} + +impl WorkerRetentionProvider for FsWorkerRetentionProvider { + fn inventory( + &self, + workspace_id: &str, + runtime_id: &str, + worker_id: WorkerId, + run_generation: u64, + ) -> Result { + let worker_dir = self.worker_dir(worker_id); + if !worker_dir.is_dir() { + return Err(RuntimeError::WorkerNotFound { worker_id }); + } + let session_dir = worker_dir.join("session"); + let (session_id, segment_ids, session_bytes) = if session_dir.is_dir() { + let manifest: CanonicalSessionManifest = read_json( + &session_dir.join("session.json"), + "inventory Worker retention", + )?; + let files = collect_files(&session_dir, "inventory Worker retention")?; + let mut segment_ids = BTreeSet::new(); + let mut bytes = 0_u64; + for (relative, path) in files { + bytes = bytes.saturating_add(file_len(&path, "inventory Worker retention")?); + if let Some(name) = relative.file_name().and_then(|value| value.to_str()) { + let segment = name + .strip_suffix(".trace.jsonl") + .or_else(|| name.strip_suffix(".jsonl")); + if let Some(segment) = segment { + segment_ids.insert(segment.to_string()); + } + } + } + ( + Some(manifest.session_id), + segment_ids.into_iter().collect(), + bytes, + ) + } else { + (None, Vec::new(), 0) + }; + let diagnostics_bytes = diagnostics_files(&worker_dir, "inventory Worker retention")? + .into_iter() + .try_fold(0_u64, |total, path| { + file_len(&path, "inventory Worker retention").map(|size| total.saturating_add(size)) + })?; + Ok(WorkerRetentionInventory { + workspace_id: workspace_id.to_string(), + runtime_id: runtime_id.to_string(), + worker_id, + run_generation, + session_id, + segment_ids, + session_bytes, + diagnostics_bytes, + }) + } + + fn execute( + &self, + request: &WorkerRetentionExecutionRequest, + ) -> Result { + validate_request(request)?; + fs::create_dir_all(&self.runtime_root).map_err(|source| RuntimeError::StoreIo { + operation: "prepare Worker retention", + path: self.runtime_root.clone(), + source, + })?; + 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 Worker retention", + path: lock_path.clone(), + source, + })?; + lock.lock().map_err(|source| RuntimeError::StoreIo { + operation: "lock Worker retention", + path: lock_path, + source, + })?; + + let pending = read_operation_receipt( + &self.operation_path(&request.operation_id)?, + &request.input_fingerprint, + )?; + if let Some(receipt) = &pending { + if receipt.result.source_removed { + return Ok(receipt.result.clone()); + } + } + + let worker_dir = self.worker_dir(request.worker_id); + if !worker_dir.is_dir() { + if let Some(mut receipt) = pending { + // A prior attempt durably committed all disposition evidence and + // removed the aggregate, then stopped before finalizing its receipt. + receipt.result.source_removed = true; + atomic_write_json( + &self.operation_path(&request.operation_id)?, + &receipt, + "recover Worker retention receipt", + )?; + return Ok(receipt.result); + } + return Err(RuntimeError::WorkerNotFound { + worker_id: request.worker_id, + }); + } + let snapshot: WorkerGenerationSnapshot = + read_json(&worker_dir.join("worker.json"), "execute Worker retention")?; + if snapshot.run_generation != request.expected_run_generation { + return Err(RuntimeError::InvalidRequest(format!( + "Worker retention plan expected generation {}, current generation is {}", + request.expected_run_generation, snapshot.run_generation + ))); + } + + let archive = match request.session_disposition { + SessionDisposition::Archive => { + Some(commit_session_archive(self, request, &worker_dir)?) + } + SessionDisposition::Purge => None, + }; + let diagnostics_retained = match request.diagnostics_disposition { + DiagnosticsDisposition::Purge => false, + DiagnosticsDisposition::Retain => { + commit_diagnostics_archive(self, request, &worker_dir)?; + true + } + }; + + let mut result = 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, + source_removed: false, + diagnostics_retained, + }; + let mut receipt = RetentionOperationReceipt { + schema_version: OPERATION_SCHEMA_VERSION, + result: result.clone(), + }; + // Pending receipt makes the delete/final-receipt crash window + // recoverable without treating an uncommitted archive as completion. + atomic_write_json( + &self.operation_path(&request.operation_id)?, + &receipt, + "commit pending Worker retention receipt", + )?; + + fs::remove_dir_all(&worker_dir).map_err(|source| RuntimeError::StoreIo { + operation: "remove retained Worker aggregate", + path: worker_dir.clone(), + source, + })?; + sync_directory( + worker_dir.parent().unwrap_or(&self.runtime_root), + "remove retained Worker aggregate", + )?; + + result.source_removed = true; + receipt.result = result.clone(); + atomic_write_json( + &self.operation_path(&request.operation_id)?, + &receipt, + "commit Worker retention receipt", + )?; + Ok(result) + } + + fn completed( + &self, + operation_id: &str, + input_fingerprint: &str, + ) -> Result, RuntimeError> { + let path = self.operation_path(operation_id)?; + let Some(receipt) = read_operation_receipt(&path, input_fingerprint)? else { + return Ok(None); + }; + Ok(receipt.result.source_removed.then_some(receipt.result)) + } +} + +#[derive(Deserialize)] +struct WorkerGenerationSnapshot { + #[serde(default)] + run_generation: u64, +} + +#[derive(Deserialize)] +struct CanonicalSessionManifest { + session_id: String, +} + +#[derive(Serialize, Deserialize)] +struct RetentionOperationReceipt { + schema_version: u32, + result: WorkerRetentionExecutionResult, +} + +fn read_operation_receipt( + path: &Path, + input_fingerprint: &str, +) -> Result, RuntimeError> { + if !path.is_file() { + return Ok(None); + } + let receipt: RetentionOperationReceipt = read_json(path, "read Worker retention receipt")?; + if receipt.schema_version != OPERATION_SCHEMA_VERSION { + return Err(RuntimeError::StoreCorrupt { + operation: "read Worker retention receipt", + path: path.to_path_buf(), + message: format!( + "unsupported operation receipt schema {}", + receipt.schema_version + ), + }); + } + if receipt.result.input_fingerprint != input_fingerprint { + return Err(RuntimeError::InvalidRequest(format!( + "retention operation {} was already used with different input", + receipt.result.operation_id + ))); + } + Ok(Some(receipt)) +} + +#[derive(Serialize, Deserialize)] +struct DiagnosticsArchiveManifest { + schema_version: u32, + operation_id: String, + workspace_id: String, + source_runtime_id: String, + source_worker_id: WorkerId, + input_fingerprint: String, + content_checksum_sha256: String, + content_bytes: u64, + content_file_count: u64, +} + +fn validate_request(request: &WorkerRetentionExecutionRequest) -> Result<(), RuntimeError> { + validate_id("operation_id", &request.operation_id)?; + validate_id("workspace_id", &request.workspace_id)?; + validate_id("source_runtime_id", &request.source_runtime_id)?; + validate_id("policy_id", &request.policy_id)?; + if request.input_fingerprint.trim().is_empty() { + return Err(RuntimeError::InvalidRequest( + "retention input fingerprint must not be empty".to_string(), + )); + } + match (request.session_disposition, request.archive_id.as_deref()) { + (SessionDisposition::Archive, Some(id)) => validate_id("archive_id", id), + (SessionDisposition::Archive, None) => Err(RuntimeError::InvalidRequest( + "archive disposition requires archive_id".to_string(), + )), + (SessionDisposition::Purge, None) => Ok(()), + (SessionDisposition::Purge, Some(_)) => Err(RuntimeError::InvalidRequest( + "purge disposition must not include archive_id".to_string(), + )), + } +} + +fn validate_id(kind: &str, value: &str) -> Result<(), RuntimeError> { + if value.is_empty() + || value.len() > 160 + || value == "." + || value == ".." + || !value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':')) + { + return Err(RuntimeError::InvalidRequest(format!( + "invalid retention {kind}" + ))); + } + Ok(()) +} + +fn commit_session_archive( + provider: &FsWorkerRetentionProvider, + request: &WorkerRetentionExecutionRequest, + worker_dir: &Path, +) -> Result { + let archive_id = request.archive_id.as_deref().ok_or_else(|| { + RuntimeError::InvalidRequest("archive disposition requires archive_id".to_string()) + })?; + let session_dir = worker_dir.join("session"); + if !session_dir.is_dir() { + return Err(RuntimeError::StoreMissing { + operation: "archive Worker Session", + path: session_dir, + }); + } + let session: CanonicalSessionManifest = + read_json(&session_dir.join("session.json"), "archive Worker Session")?; + let source_files = collect_files(&session_dir, "archive Worker Session")?; + let (checksum, bytes, count) = checksum_files(&source_files, "archive Worker Session")?; + let mut segment_ids = BTreeSet::new(); + for (relative, _) in &source_files { + if let Some(name) = relative.file_name().and_then(|name| name.to_str()) { + if let Some(segment) = name + .strip_suffix(".trace.jsonl") + .or_else(|| name.strip_suffix(".jsonl")) + { + segment_ids.insert(segment.to_string()); + } + } + } + let manifest = WorkerSessionArchiveManifest { + schema_version: ARCHIVE_SCHEMA_VERSION, + archive_id: archive_id.to_string(), + workspace_id: request.workspace_id.clone(), + source_runtime_id: request.source_runtime_id.clone(), + source_worker_id: request.worker_id, + source_session_id: session.session_id, + segment_ids: segment_ids.into_iter().collect(), + source_created_at: request.source_created_at.clone(), + removed_at: request.removed_at.clone(), + archived_at_unix_seconds: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + effective_profile: request.effective_profile.clone(), + retention_class: request.retention_class.clone(), + content_checksum_sha256: checksum, + content_bytes: bytes, + content_file_count: count, + policy_id: request.policy_id.clone(), + policy_revision: request.policy_revision, + operation_id: request.operation_id.clone(), + input_fingerprint: request.input_fingerprint.clone(), + }; + let archive_dir = provider.archive_dir(archive_id)?; + if archive_dir.exists() { + return validate_existing_archive(&archive_dir, &manifest); + } + let parent = archive_dir + .parent() + .ok_or_else(|| RuntimeError::StoreCorrupt { + operation: "archive Worker Session", + path: archive_dir.clone(), + message: "archive target has no parent".to_string(), + })?; + fs::create_dir_all(parent).map_err(|source| RuntimeError::StoreIo { + operation: "archive Worker Session", + path: parent.to_path_buf(), + source, + })?; + let staging = temporary_path(&archive_dir); + let result = (|| { + fs::create_dir(&staging).map_err(|source| RuntimeError::StoreIo { + operation: "archive Worker Session", + path: staging.clone(), + source, + })?; + let target_session = staging.join("session"); + fs::create_dir(&target_session).map_err(|source| RuntimeError::StoreIo { + operation: "archive Worker Session", + path: target_session.clone(), + source, + })?; + copy_files(&source_files, &target_session, "archive Worker Session")?; + atomic_write_json( + &staging.join("manifest.json"), + &manifest, + "archive Worker Session", + )?; + sync_tree(&staging, "archive Worker Session")?; + fs::rename(&staging, &archive_dir).map_err(|source| RuntimeError::StoreIo { + operation: "archive Worker Session", + path: archive_dir.clone(), + source, + })?; + sync_directory(parent, "archive Worker Session")?; + validate_existing_archive(&archive_dir, &manifest) + })(); + if result.is_err() { + let _ = fs::remove_dir_all(&staging); + } + result +} + +fn validate_existing_archive( + archive_dir: &Path, + expected: &WorkerSessionArchiveManifest, +) -> Result { + let existing: WorkerSessionArchiveManifest = read_json( + &archive_dir.join("manifest.json"), + "verify Worker Session archive", + )?; + let mut comparable_expected = expected.clone(); + comparable_expected.archived_at_unix_seconds = existing.archived_at_unix_seconds; + if existing != comparable_expected { + return Err(RuntimeError::StoreCorrupt { + operation: "verify Worker Session archive", + path: archive_dir.join("manifest.json"), + message: "archive id collision or manifest mismatch".to_string(), + }); + } + let files = collect_files( + &archive_dir.join("session"), + "verify Worker Session archive", + )?; + let (checksum, bytes, count) = checksum_files(&files, "verify Worker Session archive")?; + if checksum != existing.content_checksum_sha256 + || bytes != existing.content_bytes + || count != existing.content_file_count + { + return Err(RuntimeError::StoreCorrupt { + operation: "verify Worker Session archive", + path: archive_dir.to_path_buf(), + message: "archive checksum or content summary mismatch".to_string(), + }); + } + Ok(existing) +} + +fn commit_diagnostics_archive( + provider: &FsWorkerRetentionProvider, + request: &WorkerRetentionExecutionRequest, + worker_dir: &Path, +) -> Result<(), RuntimeError> { + let files = diagnostics_files(worker_dir, "archive Worker diagnostics")?; + let target = provider.diagnostics_dir(&request.operation_id)?; + if target.exists() { + let manifest: DiagnosticsArchiveManifest = read_json( + &target.join("manifest.json"), + "verify Worker diagnostics archive", + )?; + if manifest.input_fingerprint != request.input_fingerprint { + return Err(RuntimeError::InvalidRequest(format!( + "diagnostics archive operation {} was reused with different input", + request.operation_id + ))); + } + return Ok(()); + } + let parent = target.parent().ok_or_else(|| RuntimeError::StoreCorrupt { + operation: "archive Worker diagnostics", + path: target.clone(), + message: "diagnostics archive target has no parent".to_string(), + })?; + fs::create_dir_all(parent).map_err(|source| RuntimeError::StoreIo { + operation: "archive Worker diagnostics", + path: parent.to_path_buf(), + source, + })?; + let staging = temporary_path(&target); + let result = (|| { + fs::create_dir(&staging).map_err(|source| RuntimeError::StoreIo { + operation: "archive Worker diagnostics", + path: staging.clone(), + source, + })?; + let source_files = files + .iter() + .map(|path| { + let relative = + path.strip_prefix(worker_dir) + .map_err(|_| RuntimeError::StoreCorrupt { + operation: "archive Worker diagnostics", + path: path.clone(), + message: "diagnostics path escaped Worker aggregate".to_string(), + })?; + Ok((relative.to_path_buf(), path.clone())) + }) + .collect::, RuntimeError>>()?; + copy_files(&source_files, &staging, "archive Worker diagnostics")?; + let (checksum, bytes, count) = checksum_files(&source_files, "archive Worker diagnostics")?; + let manifest = DiagnosticsArchiveManifest { + schema_version: ARCHIVE_SCHEMA_VERSION, + operation_id: request.operation_id.clone(), + workspace_id: request.workspace_id.clone(), + source_runtime_id: request.source_runtime_id.clone(), + source_worker_id: request.worker_id, + input_fingerprint: request.input_fingerprint.clone(), + content_checksum_sha256: checksum, + content_bytes: bytes, + content_file_count: count, + }; + atomic_write_json( + &staging.join("manifest.json"), + &manifest, + "archive Worker diagnostics", + )?; + sync_tree(&staging, "archive Worker diagnostics")?; + fs::rename(&staging, &target).map_err(|source| RuntimeError::StoreIo { + operation: "archive Worker diagnostics", + path: target.clone(), + source, + })?; + sync_directory(parent, "archive Worker diagnostics") + })(); + if result.is_err() { + let _ = fs::remove_dir_all(&staging); + } + result +} + +fn diagnostics_files( + worker_dir: &Path, + operation: &'static str, +) -> Result, RuntimeError> { + let runs = worker_dir.join("runs"); + if !runs.is_dir() { + return Ok(Vec::new()); + } + let mut files = Vec::new(); + for (_, path) in collect_files(&runs, operation)? { + let name = path.file_name().and_then(|name| name.to_str()); + if matches!(name, Some("worker.out.log" | "worker.err.log")) { + files.push(path); + } + } + files.sort(); + Ok(files) +} + +fn collect_files( + root: &Path, + operation: &'static str, +) -> Result, RuntimeError> { + fn visit( + root: &Path, + current: &Path, + operation: &'static str, + files: &mut Vec<(PathBuf, PathBuf)>, + ) -> Result<(), RuntimeError> { + let mut entries = fs::read_dir(current) + .map_err(|source| RuntimeError::StoreIo { + operation, + path: current.to_path_buf(), + source, + })? + .collect::, _>>() + .map_err(|source| RuntimeError::StoreIo { + operation, + path: current.to_path_buf(), + source, + })?; + entries.sort_by_key(|entry| entry.file_name()); + for entry in entries { + let path = entry.path(); + let file_type = entry.file_type().map_err(|source| RuntimeError::StoreIo { + operation, + path: path.clone(), + source, + })?; + if file_type.is_symlink() { + return Err(RuntimeError::StoreCorrupt { + operation, + path, + message: "symlinks are not allowed in retained Worker evidence".to_string(), + }); + } + if file_type.is_dir() { + visit(root, &path, operation, files)?; + } else if file_type.is_file() { + let relative = path + .strip_prefix(root) + .map_err(|_| RuntimeError::StoreCorrupt { + operation, + path: path.clone(), + message: "retention source escaped its aggregate root".to_string(), + })? + .to_path_buf(); + files.push((relative, path)); + } else { + return Err(RuntimeError::StoreCorrupt { + operation, + path, + message: "unsupported retained Worker evidence entry".to_string(), + }); + } + } + Ok(()) + } + if !root.is_dir() { + return Ok(Vec::new()); + } + let mut files = Vec::new(); + visit(root, root, operation, &mut files)?; + files.sort_by(|left, right| left.0.cmp(&right.0)); + Ok(files) +} + +fn checksum_files( + files: &[(PathBuf, PathBuf)], + operation: &'static str, +) -> Result<(String, u64, u64), RuntimeError> { + let mut hasher = Sha256::new(); + let mut total = 0_u64; + for (relative, path) in files { + let relative = relative.to_string_lossy(); + hasher.update((relative.len() as u64).to_be_bytes()); + hasher.update(relative.as_bytes()); + let mut file = File::open(path).map_err(|source| RuntimeError::StoreIo { + operation, + path: path.clone(), + source, + })?; + let length = file_len(path, operation)?; + hasher.update(length.to_be_bytes()); + let mut buffer = [0_u8; 8192]; + loop { + let read = file + .read(&mut buffer) + .map_err(|source| RuntimeError::StoreIo { + operation, + path: path.clone(), + source, + })?; + if read == 0 { + break; + } + hasher.update(&buffer[..read]); + } + total = total.saturating_add(length); + } + let digest = hasher.finalize(); + let checksum = digest + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + Ok((checksum, total, files.len() as u64)) +} + +fn copy_files( + files: &[(PathBuf, PathBuf)], + target_root: &Path, + operation: &'static str, +) -> Result<(), RuntimeError> { + for (relative, source_path) in files { + let target = target_root.join(relative); + let parent = target.parent().ok_or_else(|| RuntimeError::StoreCorrupt { + operation, + path: target.clone(), + message: "retention copy target has no parent".to_string(), + })?; + fs::create_dir_all(parent).map_err(|source| RuntimeError::StoreIo { + operation, + path: parent.to_path_buf(), + source, + })?; + let bytes = fs::read(source_path).map_err(|source| RuntimeError::StoreIo { + operation, + path: source_path.clone(), + source, + })?; + let mut file = OpenOptions::new() + .write(true) + .create_new(true) + .open(&target) + .map_err(|source| RuntimeError::StoreIo { + operation, + path: target.clone(), + source, + })?; + file.write_all(&bytes) + .and_then(|_| file.sync_all()) + .map_err(|source| RuntimeError::StoreIo { + operation, + path: target, + source, + })?; + } + Ok(()) +} + +fn atomic_write_json( + path: &Path, + value: &T, + operation: &'static str, +) -> Result<(), RuntimeError> { + let parent = path.parent().ok_or_else(|| RuntimeError::StoreCorrupt { + operation, + path: path.to_path_buf(), + message: "retention record has no parent".to_string(), + })?; + fs::create_dir_all(parent).map_err(|source| RuntimeError::StoreIo { + operation, + path: parent.to_path_buf(), + source, + })?; + let temporary = temporary_path(path); + let result = (|| { + let mut file = OpenOptions::new() + .write(true) + .create_new(true) + .open(&temporary) + .map_err(|source| RuntimeError::StoreIo { + operation, + path: temporary.clone(), + source, + })?; + serde_json::to_writer_pretty(&mut file, value).map_err(|source| { + RuntimeError::StoreCorrupt { + operation, + path: temporary.clone(), + message: source.to_string(), + } + })?; + file.write_all(b"\n") + .and_then(|_| file.sync_all()) + .map_err(|source| RuntimeError::StoreIo { + operation, + path: temporary.clone(), + source, + })?; + drop(file); + fs::rename(&temporary, path).map_err(|source| RuntimeError::StoreIo { + operation, + path: path.to_path_buf(), + source, + })?; + sync_directory(parent, operation) + })(); + if result.is_err() { + let _ = fs::remove_file(temporary); + } + result +} + +fn read_json Deserialize<'de>>( + path: &Path, + operation: &'static str, +) -> Result { + let file = File::open(path).map_err(|source| match source.kind() { + std::io::ErrorKind::NotFound => RuntimeError::StoreMissing { + operation, + path: path.to_path_buf(), + }, + _ => RuntimeError::StoreIo { + operation, + path: path.to_path_buf(), + source, + }, + })?; + serde_json::from_reader(file).map_err(|source| RuntimeError::StoreCorrupt { + operation, + path: path.to_path_buf(), + message: source.to_string(), + }) +} + +fn sync_tree(path: &Path, operation: &'static str) -> Result<(), RuntimeError> { + let mut directories = vec![path.to_path_buf()]; + let mut index = 0; + while index < directories.len() { + let current = directories[index].clone(); + index += 1; + for entry in fs::read_dir(¤t).map_err(|source| RuntimeError::StoreIo { + operation, + path: current.clone(), + source, + })? { + let entry = entry.map_err(|source| RuntimeError::StoreIo { + operation, + path: current.clone(), + source, + })?; + if entry.path().is_dir() { + directories.push(entry.path()); + } + } + } + for directory in directories.into_iter().rev() { + sync_directory(&directory, operation)?; + } + Ok(()) +} + +fn sync_directory(path: &Path, operation: &'static str) -> Result<(), RuntimeError> { + File::open(path) + .and_then(|file| file.sync_all()) + .map_err(|source| RuntimeError::StoreIo { + operation, + path: path.to_path_buf(), + source, + }) +} + +fn file_len(path: &Path, operation: &'static str) -> Result { + fs::metadata(path) + .map(|metadata| metadata.len()) + .map_err(|source| RuntimeError::StoreIo { + operation, + path: path.to_path_buf(), + source, + }) +} + +fn temporary_path(path: &Path) -> PathBuf { + let sequence = NEXT_TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed); + let name = path + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or("retention"); + path.with_file_name(format!(".{name}.tmp-{}-{sequence}", std::process::id())) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, Barrier}; + + fn write_json(path: &Path, value: &impl Serialize) { + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write(path, serde_json::to_vec_pretty(value).unwrap()).unwrap(); + } + + fn source(root: &Path, worker_id: WorkerId, generation: u64) { + let worker = root.join("workers").join(worker_id.to_string()); + write_json( + &worker.join("worker.json"), + &serde_json::json!({"run_generation": generation}), + ); + write_json( + &worker.join("session/session.json"), + &serde_json::json!({"schema_version": 1, "session_id": "session-a"}), + ); + fs::create_dir_all(worker.join("session/segments")).unwrap(); + fs::write(worker.join("session/segments/segment-a.jsonl"), b"one\n").unwrap(); + fs::create_dir_all(worker.join(format!("runs/{generation}"))).unwrap(); + fs::write( + worker.join(format!("runs/{generation}/worker.out.log")), + b"diagnostic\n", + ) + .unwrap(); + fs::write( + worker.join(format!("runs/{generation}/worker.sock")), + b"not retained", + ) + .unwrap(); + } + + fn request( + worker_id: WorkerId, + generation: u64, + disposition: SessionDisposition, + ) -> WorkerRetentionExecutionRequest { + WorkerRetentionExecutionRequest { + operation_id: "operation-a".to_string(), + input_fingerprint: "fingerprint-a".to_string(), + archive_id: (disposition == SessionDisposition::Archive) + .then(|| "archive-a".to_string()), + workspace_id: "workspace-a".to_string(), + source_runtime_id: "runtime-a".to_string(), + worker_id, + expected_run_generation: generation, + source_created_at: "2026-01-01T00:00:00Z".to_string(), + removed_at: "2026-01-02T00:00:00Z".to_string(), + effective_profile: Some("builtin:coder".to_string()), + retention_class: None, + policy_id: "policy-a".to_string(), + policy_revision: 3, + session_disposition: disposition, + diagnostics_disposition: DiagnosticsDisposition::Purge, + } + } + + #[test] + fn archive_is_verified_before_source_removal_and_retry_converges() { + let temp = tempfile::tempdir().unwrap(); + let worker_id = WorkerId::new(7); + source(temp.path(), worker_id, 4); + let provider = FsWorkerRetentionProvider::new(temp.path()); + let request = request(worker_id, 4, SessionDisposition::Archive); + + let first = provider.execute(&request).unwrap(); + assert!(first.source_removed); + let archive = first.archive.as_ref().unwrap(); + assert_eq!(archive.source_session_id, "session-a"); + assert_eq!(archive.segment_ids, vec!["segment-a"]); + assert!(!temp.path().join("workers/7").exists()); + assert!( + temp.path() + .join("archives/workers/archive-a/session/segments/segment-a.jsonl") + .is_file() + ); + assert!(!temp.path().join("archives/workers/archive-a/runs").exists()); + + let retry = provider.execute(&request).unwrap(); + assert_eq!(retry, first); + assert_eq!( + fs::read_dir(temp.path().join("archives/workers")) + .unwrap() + .count(), + 1 + ); + } + + #[test] + fn archive_failure_keeps_live_source_for_retry() { + let temp = tempfile::tempdir().unwrap(); + let worker_id = WorkerId::new(8); + source(temp.path(), worker_id, 2); + let collision = temp.path().join("archives/workers/archive-a"); + fs::create_dir_all(&collision).unwrap(); + fs::write(collision.join("manifest.json"), b"not-json").unwrap(); + let provider = FsWorkerRetentionProvider::new(temp.path()); + + assert!( + provider + .execute(&request(worker_id, 2, SessionDisposition::Archive)) + .is_err() + ); + assert!(temp.path().join("workers/8/session").is_dir()); + assert!( + !temp + .path() + .join("retention/operations/operation-a.json") + .exists() + ); + } + + #[test] + fn purge_removes_aggregate_and_rejects_stale_generation() { + let temp = tempfile::tempdir().unwrap(); + let provider = FsWorkerRetentionProvider::new(temp.path()); + let worker_id = WorkerId::new(9); + source(temp.path(), worker_id, 5); + let stale = request(worker_id, 4, SessionDisposition::Purge); + assert!(provider.execute(&stale).is_err()); + assert!(temp.path().join("workers/9/session").is_dir()); + + let mut current = request(worker_id, 5, SessionDisposition::Purge); + current.operation_id = "operation-current".to_string(); + current.input_fingerprint = "fingerprint-current".to_string(); + let result = provider.execute(¤t).unwrap(); + assert!(result.archive.is_none()); + assert!(!temp.path().join("workers/9").exists()); + assert!( + temp.path() + .join("retention/operations/operation-current.json") + .is_file() + ); + } + + #[test] + fn pending_receipt_recovers_delete_to_receipt_crash_window() { + let temp = tempfile::tempdir().unwrap(); + let worker_id = WorkerId::new(11); + source(temp.path(), worker_id, 1); + let provider = FsWorkerRetentionProvider::new(temp.path()); + let request = request(worker_id, 1, SessionDisposition::Archive); + let completed = provider.execute(&request).unwrap(); + let receipt_path = temp.path().join("retention/operations/operation-a.json"); + let mut receipt: RetentionOperationReceipt = + serde_json::from_slice(&fs::read(&receipt_path).unwrap()).unwrap(); + receipt.result.source_removed = false; + fs::write(&receipt_path, serde_json::to_vec_pretty(&receipt).unwrap()).unwrap(); + + let recovered = provider.execute(&request).unwrap(); + assert_eq!(recovered, completed); + assert!( + provider + .completed("operation-a", "fingerprint-a") + .unwrap() + .is_some() + ); + } + + #[test] + fn concurrent_retry_produces_one_archive() { + let temp = tempfile::tempdir().unwrap(); + let worker_id = WorkerId::new(10); + source(temp.path(), worker_id, 1); + let provider = Arc::new(FsWorkerRetentionProvider::new(temp.path())); + let request = Arc::new(request(worker_id, 1, SessionDisposition::Archive)); + let barrier = Arc::new(Barrier::new(3)); + let handles = (0..2) + .map(|_| { + let provider = provider.clone(); + let request = request.clone(); + let barrier = barrier.clone(); + std::thread::spawn(move || { + barrier.wait(); + provider.execute(&request) + }) + }) + .collect::>(); + barrier.wait(); + let results = handles + .into_iter() + .map(|handle| handle.join().unwrap().unwrap()) + .collect::>(); + assert_eq!(results[0], results[1]); + assert_eq!( + fs::read_dir(temp.path().join("archives/workers")) + .unwrap() + .count(), + 1 + ); + } +} diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 336bf86d..e93a421d 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -26,6 +26,11 @@ use crate::management::{ }; #[cfg(feature = "ws-server")] use crate::observation::{WorkerObservationCursor, WorkerObservationEvent}; +#[cfg(feature = "fs-store")] +use crate::retention::{ + FsWorkerRetentionProvider, WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult, + WorkerRetentionInventory, WorkerRetentionProvider, +}; use protocol::subscription::{ EventSubscriptionSelector, SubscriptionEventPayload, SubscriptionSnapshot, SubscriptionValidationError, SubscriptionWorkdirId, SubscriptionWorker, SubscriptionWorkerId, @@ -1648,6 +1653,118 @@ impl Runtime { Ok(()) } + /// Bind the Backend registry identity once. Retention evidence fails closed + /// until the Runtime host supplies this trusted configuration. + pub fn bind_runtime_identity(&self, runtime_id: &str) -> Result<(), RuntimeError> { + if runtime_id.trim().is_empty() || runtime_id.len() > 160 { + return Err(RuntimeError::InvalidRequest( + "Runtime identity must be non-empty and bounded".to_string(), + )); + } + let mut state = self.lock()?; + match state.runtime_identity.as_deref() { + Some(current) if current == runtime_id => Ok(()), + Some(_) => Err(RuntimeError::InvalidRequest( + "Runtime identity is already bound".to_string(), + )), + None => { + state.runtime_identity = Some(runtime_id.to_string()); + Ok(()) + } + } + } + + /// Read canonical aggregate facts needed by a Backend removal plan. + #[cfg(feature = "fs-store")] + pub fn worker_retention_inventory( + &self, + workspace_id: &str, + worker_ref: &WorkerRef, + ) -> Result { + let state = self.lock()?; + let runtime_id = state.runtime_identity.as_deref().ok_or_else(|| { + RuntimeError::InvalidRequest( + "Runtime identity is not bound for Worker retention".to_string(), + ) + })?; + let worker = state.worker(worker_ref)?; + if worker.workspace_id.as_deref() != Some(workspace_id) { + return Err(RuntimeError::WorkerNotFound { + worker_id: worker_ref.worker_id, + }); + } + let store = state.fs_store().ok_or_else(|| { + RuntimeError::InvalidRequest( + "Worker retention archive authority requires an fs-backed Runtime".to_string(), + ) + })?; + FsWorkerRetentionProvider::new(store.runtime_dir()).inventory( + workspace_id, + runtime_id, + worker.worker_id, + worker.run_generation, + ) + } + + /// Execute a Backend-resolved retention plan. Only stopped Workers are + /// eligible. Provider receipt lookup happens before live lookup so exact + /// retries converge after aggregate removal. + #[cfg(feature = "fs-store")] + pub fn execute_worker_retention( + &self, + request: &WorkerRetentionExecutionRequest, + ) -> Result { + let mut state = self.lock()?; + let runtime_id = state.runtime_identity.clone().ok_or_else(|| { + RuntimeError::InvalidRequest( + "Runtime identity is not bound for Worker retention".to_string(), + ) + })?; + if request.source_runtime_id != runtime_id { + return Err(RuntimeError::InvalidRequest( + "Worker retention Runtime identity mismatch".to_string(), + )); + } + let store = state.fs_store().ok_or_else(|| { + RuntimeError::InvalidRequest( + "Worker retention execution requires an fs-backed Runtime".to_string(), + ) + })?; + let provider = FsWorkerRetentionProvider::new(store.runtime_dir()); + if let Some(completed) = + provider.completed(&request.operation_id, &request.input_fingerprint)? + { + state.workers.remove(&request.worker_id); + state.persist_runtime_snapshot()?; + return Ok(completed); + } + let Some(worker) = state.workers.get(&request.worker_id) else { + // Recover a pending receipt after a crash between aggregate removal + // and final receipt/Runtime catalog commit. + return provider.recover_after_source_removal(request); + }; + if worker.workspace_id.as_deref() != Some(request.workspace_id.as_str()) { + return Err(RuntimeError::WorkerNotFound { + worker_id: request.worker_id, + }); + } + if worker.status != WorkerStatus::Stopped { + return Err(RuntimeError::InvalidRequest( + "Worker retention requires a stopped Worker".to_string(), + )); + } + if worker.run_generation != request.expected_run_generation { + return Err(RuntimeError::InvalidRequest(format!( + "Worker retention plan expected generation {}, current generation is {}", + request.expected_run_generation, worker.run_generation + ))); + } + let result = provider.execute(request)?; + state.workers.remove(&request.worker_id); + state.persist_runtime_snapshot()?; + Ok(result) + } + fn lock(&self) -> Result, RuntimeError> { self.inner.lock().map_err(|_| RuntimeError::StatePoisoned) } @@ -1673,6 +1790,9 @@ struct SubscriptionSink { struct RuntimeState { display_name: Option, backend: RuntimeBackendKind, + /// Backend-bound stable identity used for cross-boundary retention evidence. + /// It is configured once by the Runtime host and never model input. + runtime_identity: Option, #[cfg_attr(not(feature = "fs-store"), allow(dead_code))] persistence: RuntimePersistence, status: RuntimeStatus, @@ -1701,6 +1821,7 @@ impl RuntimeState { Self { display_name, backend: RuntimeBackendKind::Memory, + runtime_identity: None, persistence: RuntimePersistence::Memory, status: RuntimeStatus::Running, execution_backend: None, @@ -1729,6 +1850,7 @@ impl RuntimeState { Self { display_name, backend: RuntimeBackendKind::FsStore, + runtime_identity: None, persistence: RuntimePersistence::Fs(store), status: RuntimeStatus::Running, execution_backend: None, @@ -1779,6 +1901,7 @@ impl RuntimeState { Ok(Self { display_name: persisted.display_name, backend: RuntimeBackendKind::FsStore, + runtime_identity: None, persistence: RuntimePersistence::Fs(store), status: persisted.status, execution_backend: None, @@ -2550,6 +2673,18 @@ mod tests { use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; + #[test] + fn runtime_identity_binding_is_immutable_and_host_owned() { + let runtime = Runtime::new_memory(); + runtime.bind_runtime_identity("runtime-a").unwrap(); + runtime.bind_runtime_identity("runtime-a").unwrap(); + assert!(runtime.bind_runtime_identity("runtime-b").is_err()); + assert_eq!( + runtime.lock().unwrap().runtime_identity.as_deref(), + Some("runtime-a") + ); + } + #[test] fn typed_segments_allow_empty_flat_content() { let input = WorkerInput { diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index b2046416..46aecd87 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1497,6 +1497,9 @@ impl EmbeddedWorkerRuntime { pub fn from_runtime(workspace_id: impl AsRef, runtime: worker_runtime::Runtime) -> Self { let workspace_id = workspace_id.as_ref().to_string(); + runtime + .bind_runtime_identity(EMBEDDED_RUNTIME_ID) + .expect("fresh embedded Runtime must accept its Backend-owned identity"); Self { runtime_id: EMBEDDED_RUNTIME_ID.to_string(), host_id: host_id_for_embedded_workspace(&workspace_id), diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index 32708b85..544f38fb 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -19,6 +19,7 @@ pub mod records; pub use records::ticket_api_typescript; pub mod repositories; pub mod resource_broker; +pub mod retention; pub mod runtime_subscription; pub mod server; pub mod skills; diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs new file mode 100644 index 00000000..b3b21ff6 --- /dev/null +++ b/crates/workspace-server/src/retention.rs @@ -0,0 +1,1194 @@ +//! Workspace DB authority for deterministic Worker retention planning. +//! Runtime receives only resolved dispositions and stable ids; policy authority +//! never comes from prompts, profiles, or model input. + +use crate::{Error as StoreError, store::SqliteWorkspaceStore}; +use chrono::Utc; +use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use worker_runtime::identity::RuntimeWorkerRef; +use worker_runtime::retention::{ + DiagnosticsDisposition, SessionDisposition, WorkerRetentionExecutionRequest, + WorkerRetentionExecutionResult, WorkerRetentionInventory, +}; + +pub const CONSERVATIVE_POLICY_ID: &str = "workspace-default-conservative"; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum MetadataDisposition { + Tombstone, + Purge, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ArchiveRetention { + Forever, + ForSeconds { seconds: u64 }, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerRetentionPolicy { + pub workspace_id: String, + pub policy_id: String, + pub revision: u64, + pub session_disposition: SessionDisposition, + pub metadata_disposition: MetadataDisposition, + pub archive_retention: ArchiveRetention, + pub diagnostics_disposition: DiagnosticsDisposition, + pub diagnostics_retention_seconds: Option, + pub created_at: String, + pub updated_at: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerRetentionPolicyUpdate { + pub policy_id: String, + pub session_disposition: SessionDisposition, + pub metadata_disposition: MetadataDisposition, + pub archive_retention: ArchiveRetention, + pub diagnostics_disposition: DiagnosticsDisposition, + pub diagnostics_retention_seconds: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerRemovalPlanRequest { + pub workspace_id: String, + pub worker: RuntimeWorkerRef, + pub expected_worker_revision: String, + pub reason: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum WorkerRemovalBlocker { + Hold, + CurrentAssignment { + assignment_id: String, + ticket_id: String, + }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkerRemovalPlanState { + Planned, + Blocked, + Executing, + Failed, + Stale, + Succeeded, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerRemovalPlan { + pub plan_id: String, + pub operation_id: String, + pub input_fingerprint: String, + pub workspace_id: String, + pub worker: RuntimeWorkerRef, + pub worker_revision: String, + pub run_generation: u64, + pub policy_id: String, + pub policy_revision: u64, + pub session_disposition: SessionDisposition, + pub metadata_disposition: MetadataDisposition, + pub archive_retention: ArchiveRetention, + pub diagnostics_disposition: DiagnosticsDisposition, + pub diagnostics_retention_seconds: Option, + pub archive_id: Option, + pub blockers: Vec, + pub state: WorkerRemovalPlanState, + pub reason: String, + pub created_at: String, + pub updated_at: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PreparedWorkerRemoval { + pub plan: WorkerRemovalPlan, + pub runtime_request: WorkerRetentionExecutionRequest, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerTombstone { + pub workspace_id: String, + pub worker: RuntimeWorkerRef, + pub display_name: String, + pub profile: Option, + pub created_at: String, + pub removed_at: String, + pub archive_id: Option, + pub policy_id: String, + pub policy_revision: u64, + pub operation_id: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerOrphanDiagnostic { + pub diagnostic_id: String, + pub workspace_id: String, + pub runtime_id: String, + pub worker_id: String, + pub category: String, + pub detail: String, + pub observed_at: String, +} + +#[derive(thiserror::Error, Debug)] +pub enum WorkerRetentionError { + #[error(transparent)] + Store(#[from] StoreError), + #[error("Worker retention policy is not configured for Workspace {workspace_id}")] + PolicyMissing { workspace_id: String }, + #[error("Worker retention policy revision conflict: expected {expected}, current {actual}")] + PolicyRevisionConflict { expected: u64, actual: u64 }, + #[error("Worker was not found in the requested Workspace")] + WorkerNotFound, + #[error("Worker belongs to a different Workspace")] + CrossWorkspace, + #[error("Worker revision changed: expected {expected}, current {actual}")] + WorkerRevisionConflict { expected: String, actual: String }, + #[error("Worker removal is blocked: {0:?}")] + Blocked(Vec), + #[error("Worker removal plan {plan_id} is stale: {reason}")] + StalePlan { plan_id: String, reason: String }, + #[error("Worker removal operation {operation_id} was reused with different input")] + OperationFingerprintConflict { operation_id: String }, + #[error("invalid Worker retention input: {0}")] + Invalid(String), +} + +pub(crate) fn create_worker_retention_tables(conn: &Connection) -> crate::Result<()> { + conn.execute_batch(r#" + CREATE TABLE workspace_worker_retention_policy_revisions ( + workspace_id TEXT NOT NULL, policy_id TEXT NOT NULL, revision INTEGER NOT NULL CHECK(revision>0), + session_disposition TEXT NOT NULL CHECK(session_disposition IN ('archive','purge')), + metadata_disposition TEXT NOT NULL CHECK(metadata_disposition IN ('tombstone','purge')), + archive_retention_kind TEXT NOT NULL CHECK(archive_retention_kind IN ('forever','for_seconds')), + archive_retention_seconds INTEGER, + diagnostics_disposition TEXT NOT NULL CHECK(diagnostics_disposition IN ('purge','retain')), + diagnostics_retention_seconds INTEGER, created_at TEXT NOT NULL, + PRIMARY KEY(workspace_id,policy_id,revision), + FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE); + CREATE TABLE workspace_worker_retention_policies ( + workspace_id TEXT PRIMARY KEY, policy_id TEXT NOT NULL, revision INTEGER NOT NULL, updated_at TEXT NOT NULL, + FOREIGN KEY(workspace_id,policy_id,revision) REFERENCES workspace_worker_retention_policy_revisions(workspace_id,policy_id,revision), + FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE); + CREATE TABLE worker_removal_operations ( + operation_id TEXT PRIMARY KEY, plan_id TEXT NOT NULL UNIQUE, input_fingerprint TEXT NOT NULL, + workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL, + worker_revision TEXT NOT NULL, run_generation INTEGER NOT NULL CHECK(run_generation>=0), + policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL, + session_disposition TEXT NOT NULL, metadata_disposition TEXT NOT NULL, + archive_retention_kind TEXT NOT NULL, archive_retention_seconds INTEGER, + diagnostics_disposition TEXT NOT NULL, + diagnostics_retention_seconds INTEGER, archive_id TEXT UNIQUE, blockers_json TEXT NOT NULL, + state TEXT NOT NULL CHECK(state IN ('planned','blocked','executing','failed','stale','succeeded')), + reason TEXT NOT NULL, failure_category TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, + FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE); + CREATE INDEX worker_removal_operations_worker_idx ON worker_removal_operations(workspace_id,runtime_id,worker_id,created_at); + CREATE TABLE worker_session_archives ( + archive_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL, + session_id TEXT NOT NULL, checksum_sha256 TEXT NOT NULL, content_bytes INTEGER NOT NULL, + policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL, operation_id TEXT NOT NULL UNIQUE, + committed_at TEXT NOT NULL, expires_at TEXT, + FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE, + FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id)); + CREATE TABLE worker_diagnostics_archives ( + operation_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL, + committed_at TEXT NOT NULL, expires_at TEXT NOT NULL, + FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id), + FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE); + CREATE TABLE worker_tombstones ( + workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL, + display_name TEXT NOT NULL, profile TEXT, worker_created_at TEXT NOT NULL, removed_at TEXT NOT NULL, + archive_id TEXT, policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL, operation_id TEXT NOT NULL UNIQUE, + PRIMARY KEY(workspace_id,runtime_id,worker_id), + FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE, + FOREIGN KEY(archive_id) REFERENCES worker_session_archives(archive_id), + FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id)); + CREATE TABLE worker_orphan_diagnostics ( + diagnostic_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL, + category TEXT NOT NULL, detail TEXT NOT NULL, observed_at TEXT NOT NULL, + FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE); + CREATE TABLE worker_retention_audit_events ( + event_id TEXT PRIMARY KEY, operation_id TEXT NOT NULL, workspace_id TEXT NOT NULL, + event_kind TEXT NOT NULL, detail TEXT NOT NULL, created_at TEXT NOT NULL, + FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id), + FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE); + CREATE TRIGGER seed_worker_retention_policy_after_workspace_insert AFTER INSERT ON workspaces BEGIN + INSERT INTO workspace_worker_retention_policy_revisions + (workspace_id,policy_id,revision,session_disposition,metadata_disposition,archive_retention_kind,archive_retention_seconds,diagnostics_disposition,diagnostics_retention_seconds,created_at) + VALUES(NEW.workspace_id,'workspace-default-conservative',1,'archive','tombstone','forever',NULL,'purge',NULL,NEW.created_at); + INSERT INTO workspace_worker_retention_policies(workspace_id,policy_id,revision,updated_at) + VALUES(NEW.workspace_id,'workspace-default-conservative',1,NEW.created_at); + END; + "#)?; + let now = Utc::now().to_rfc3339(); + conn.execute("INSERT OR IGNORE INTO workspace_worker_retention_policy_revisions + (workspace_id,policy_id,revision,session_disposition,metadata_disposition,archive_retention_kind,archive_retention_seconds,diagnostics_disposition,diagnostics_retention_seconds,created_at) + SELECT workspace_id,?1,1,'archive','tombstone','forever',NULL,'purge',NULL,?2 FROM workspaces", params![CONSERVATIVE_POLICY_ID,now])?; + conn.execute("INSERT OR IGNORE INTO workspace_worker_retention_policies(workspace_id,policy_id,revision,updated_at) + SELECT workspace_id,?1,1,?2 FROM workspaces", params![CONSERVATIVE_POLICY_ID,now])?; + Ok(()) +} + +impl SqliteWorkspaceStore { + pub fn worker_retention_policy( + &self, + workspace_id: &str, + ) -> crate::Result> { + self.with_conn(|conn| load_policy(conn, workspace_id)) + } + + pub fn update_worker_retention_policy( + &self, + workspace_id: &str, + expected: u64, + update: &WorkerRetentionPolicyUpdate, + ) -> Result { + validate_policy(update)?; + self.with_conn_mut(|conn| { + let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let current=load_policy(&tx,workspace_id)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?; + if current.revision!=expected { return Err(StoreError::InvalidInput(format!("policy-conflict:{expected}:{}",current.revision))); } + let revision=current.revision+1; let now=Utc::now().to_rfc3339(); + tx.execute("INSERT INTO workspace_worker_retention_policy_revisions + (workspace_id,policy_id,revision,session_disposition,metadata_disposition,archive_retention_kind,archive_retention_seconds,diagnostics_disposition,diagnostics_retention_seconds,created_at) + VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)",params![workspace_id,update.policy_id,revision,sess(update.session_disposition),meta(update.metadata_disposition),archive_kind(update.archive_retention),archive_seconds(update.archive_retention),diag(update.diagnostics_disposition),update.diagnostics_retention_seconds,now])?; + let changed=tx.execute("UPDATE workspace_worker_retention_policies SET policy_id=?1,revision=?2,updated_at=?3 WHERE workspace_id=?4 AND revision=?5", + params![update.policy_id,revision,now,workspace_id,expected])?; + if changed!=1 { return Err(StoreError::InvalidInput(format!("policy-conflict:{expected}:{revision}"))); } + tx.commit()?; load_policy(conn,workspace_id)?.ok_or_else(|| StoreError::InvalidInput("updated policy missing".into())) + }).map_err(map_error) + } + + pub fn plan_worker_removal( + &self, + req: &WorkerRemovalPlanRequest, + inv: &WorkerRetentionInventory, + ) -> Result { + validate_plan(req, inv)?; + let now = Utc::now().to_rfc3339(); + self.with_conn_mut(|conn| { + let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let policy=load_policy(&tx,&req.workspace_id)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?; + let worker=match load_worker(&tx,&req.workspace_id,&req.worker)? { + Some(v)=>v, + None=>{ + let other:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM worker_registry WHERE runtime_id=?1 AND runtime_worker_id=?2 AND workspace_id!=?3)",params![req.worker.runtime_id,req.worker.worker_id,req.workspace_id],|r|r.get(0))?; + return Err(StoreError::InvalidInput(if other{"cross-workspace".into()}else{"worker-missing".into()})); + } + }; + if worker.updated_at!=req.expected_worker_revision { return Err(StoreError::InvalidInput(format!("worker-conflict:{}:{}",req.expected_worker_revision,worker.updated_at))); } + let mut blockers=Vec::new(); + if worker.retention_state=="pinned" { blockers.push(WorkerRemovalBlocker::Hold); } + if let Some((assignment_id,ticket_id))=tx.query_row("SELECT a.assignment_id,a.ticket_id FROM ticket_current_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",params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()? { + blockers.push(WorkerRemovalBlocker::CurrentAssignment{assignment_id,ticket_id}); + } + let fp=fingerprint(req,inv,&policy,&blockers)?; + let plan_id=stable("wrp",&fp); let operation_id=stable("wro",&fp); + let archive_id=(policy.session_disposition==SessionDisposition::Archive).then(||stable("wra",&fp)); + let state=if blockers.is_empty(){WorkerRemovalPlanState::Planned}else{WorkerRemovalPlanState::Blocked}; + tx.execute("INSERT OR IGNORE INTO worker_removal_operations(operation_id,plan_id,input_fingerprint,workspace_id,runtime_id,worker_id,worker_revision,run_generation,policy_id,policy_revision,session_disposition,metadata_disposition,archive_retention_kind,archive_retention_seconds,diagnostics_disposition,diagnostics_retention_seconds,archive_id,blockers_json,state,reason,created_at,updated_at) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?21)",params![operation_id,plan_id,fp,req.workspace_id,req.worker.runtime_id,req.worker.worker_id,req.expected_worker_revision,inv.run_generation,policy.policy_id,policy.revision,sess(policy.session_disposition),meta(policy.metadata_disposition),archive_kind(policy.archive_retention),archive_seconds(policy.archive_retention),diag(policy.diagnostics_disposition),policy.diagnostics_retention_seconds,archive_id,serde_json::to_string(&blockers).map_err(|e|StoreError::InvalidInput(e.to_string()))?,state_s(state),req.reason,now])?; + let plan=load_plan(&tx,&plan_id)?.ok_or_else(||StoreError::InvalidInput("plan missing".into()))?; + if plan.input_fingerprint!=fp{return Err(StoreError::InvalidInput(format!("fingerprint:{}",plan.operation_id)));} + tx.commit()?; Ok(plan) + }).map_err(map_error) + } + + pub fn begin_worker_removal( + &self, + workspace_id: &str, + plan_id: &str, + fp: &str, + ) -> Result { + self.with_conn_mut(|conn|{ + let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let mut plan=load_plan(&tx,plan_id)?.ok_or_else(||StoreError::InvalidInput(format!("stale:{plan_id}:plan missing")))?; + if plan.workspace_id!=workspace_id{return Err(StoreError::InvalidInput("cross-workspace".into()));} + if plan.input_fingerprint!=fp{return Err(StoreError::InvalidInput(format!("fingerprint:{}",plan.operation_id)));} + if plan.state==WorkerRemovalPlanState::Succeeded{tx.commit()?;return Ok(plan);} + if plan.state==WorkerRemovalPlanState::Blocked { + return Err(StoreError::InvalidInput(format!("blocked:{}",serde_json::to_string(&plan.blockers).unwrap()))); + } + if !matches!(plan.state, WorkerRemovalPlanState::Planned | WorkerRemovalPlanState::Failed | WorkerRemovalPlanState::Executing) { + 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_policy(&tx,workspace_id)?.ok_or_else(||StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?; + if policy.policy_id!=plan.policy_id||policy.revision!=plan.policy_revision{ + mark_stale(&tx,&plan,"policy revision changed")?;tx.commit()?; + return Err(stale_error(&plan,"policy revision 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_revision{ + mark_stale(&tx,&plan,"Worker revision changed")?;tx.commit()?; + return Err(stale_error(&plan,"Worker revision changed")); + } + if worker.retention_state=="pinned"{ + 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_current_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)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|r|r.get(0))?; + if assigned{ + mark_stale(&tx,&plan,"current assignment added")?;tx.commit()?; + return Err(stale_error(&plan,"current assignment added")); + } + let now=Utc::now().to_rfc3339(); + tx.execute("UPDATE worker_removal_operations SET state='executing',failure_category=NULL,updated_at=?1 WHERE operation_id=?2",params![now,plan.operation_id])?; + plan.state=WorkerRemovalPlanState::Executing;plan.updated_at=now;tx.commit()?;Ok(plan) + }).map_err(map_error) + } + + /// Revalidates Backend authority and derives the complete Runtime request + /// from the immutable plan. Callers cannot substitute generation or + /// dispositions without causing a fingerprint/manifest mismatch. + pub fn prepare_worker_removal_execution( + &self, + workspace_id: &str, + plan_id: &str, + input_fingerprint: &str, + ) -> Result { + let plan = self.begin_worker_removal(workspace_id, plan_id, input_fingerprint)?; + let worker = self + .with_conn(|conn| load_worker(conn, workspace_id, &plan.worker))? + .ok_or_else(|| WorkerRetentionError::StalePlan { + plan_id: plan.plan_id.clone(), + reason: "Worker disappeared after execution fence".to_string(), + })?; + let worker_number = plan.worker.worker_id.parse::().map_err(|_| { + WorkerRetentionError::Invalid( + "Runtime Worker id is not a canonical unsigned integer".to_string(), + ) + })?; + let removed_at = Utc::now().to_rfc3339(); + Ok(PreparedWorkerRemoval { + runtime_request: WorkerRetentionExecutionRequest { + operation_id: plan.operation_id.clone(), + input_fingerprint: plan.input_fingerprint.clone(), + archive_id: plan.archive_id.clone(), + workspace_id: plan.workspace_id.clone(), + source_runtime_id: plan.worker.runtime_id.clone(), + worker_id: worker_runtime::identity::WorkerId::new(worker_number), + expected_run_generation: plan.run_generation, + source_created_at: worker.created_at, + removed_at, + effective_profile: worker.profile, + retention_class: None, + policy_id: plan.policy_id.clone(), + policy_revision: plan.policy_revision, + session_disposition: plan.session_disposition, + diagnostics_disposition: plan.diagnostics_disposition, + }, + plan, + }) + } + + pub fn fail_worker_removal( + &self, + workspace_id: &str, + operation_id: &str, + fp: &str, + category: &str, + ) -> Result<(), WorkerRetentionError> { + bounded("failure category", category, 160)?; + self.with_conn_mut(|conn|{ + let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let changed=tx.execute("UPDATE worker_removal_operations SET state='failed',failure_category=?1,updated_at=?2 WHERE workspace_id=?3 AND operation_id=?4 AND input_fingerprint=?5 AND state IN ('planned','executing','failed')",params![category,Utc::now().to_rfc3339(),workspace_id,operation_id,fp])?; + if changed!=1{return Err(StoreError::InvalidInput("active operation mismatch".into()));} + tx.commit()?;Ok(()) + })?; + Ok(()) + } + + pub fn commit_worker_removal( + &self, + workspace_id: &str, + operation_id: &str, + fp: &str, + result: &WorkerRetentionExecutionResult, + ) -> Result { + self.with_conn_mut(|conn|{ + let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let mut plan=load_plan_op(&tx,operation_id)?.ok_or_else(||StoreError::InvalidInput("operation missing".into()))?; + if plan.workspace_id!=workspace_id{return Err(StoreError::InvalidInput("cross-workspace".into()));} + if plan.input_fingerprint!=fp||result.input_fingerprint!=fp||result.operation_id!=operation_id{return Err(StoreError::InvalidInput(format!("fingerprint:{operation_id}")));} + if plan.state==WorkerRemovalPlanState::Succeeded{tx.commit()?;return Ok(plan);} + if plan.state != WorkerRemovalPlanState::Executing { + return Err(StoreError::InvalidInput(format!("stale:{}:plan state {} is not committable", plan.plan_id, state_s(plan.state)))); + } + if result.worker_id.to_string() != plan.worker.worker_id + || result.session_disposition != plan.session_disposition + || result.diagnostics_disposition != plan.diagnostics_disposition + { + return Err(StoreError::InvalidInput("Runtime retention result does not match removal plan".into())); + } + if !result.source_removed{return Err(StoreError::InvalidInput("Runtime source was not removed".into()));} + 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_revision{return Err(StoreError::InvalidInput(format!("stale:{}:Worker revision 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_current_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)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|row|row.get(0))?; + if assigned { return Err(StoreError::InvalidInput(format!("stale:{}:current assignment added",plan.plan_id))); } + let now=Utc::now().to_rfc3339(); + if let Some(a)=&result.archive{ + if plan.archive_id.as_deref()!=Some(&a.archive_id)||a.workspace_id!=workspace_id||a.source_runtime_id!=plan.worker.runtime_id||a.source_worker_id.to_string()!=plan.worker.worker_id||a.policy_id!=plan.policy_id||a.policy_revision!=plan.policy_revision{return Err(StoreError::InvalidInput("archive manifest mismatch".into()));} + let expires_at=match plan.archive_retention { ArchiveRetention::Forever=>None, ArchiveRetention::ForSeconds{seconds}=>{let seconds=i64::try_from(seconds).map_err(|_|StoreError::InvalidInput("archive retention deadline overflow".into()))?;Some((Utc::now()+chrono::Duration::seconds(seconds)).to_rfc3339())} }; + tx.execute("INSERT OR IGNORE INTO worker_session_archives(archive_id,workspace_id,runtime_id,worker_id,session_id,checksum_sha256,content_bytes,policy_id,policy_revision,operation_id,committed_at,expires_at) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12)",params![a.archive_id,workspace_id,plan.worker.runtime_id,plan.worker.worker_id,a.source_session_id,a.content_checksum_sha256,a.content_bytes,plan.policy_id,plan.policy_revision,operation_id,now,expires_at])?; + }else if plan.session_disposition==SessionDisposition::Archive{return Err(StoreError::InvalidInput("archive manifest missing".into()));} + match plan.diagnostics_disposition { + DiagnosticsDisposition::Purge if result.diagnostics_retained => return Err(StoreError::InvalidInput("Runtime retained diagnostics for purge disposition".into())), + DiagnosticsDisposition::Retain if !result.diagnostics_retained => return Err(StoreError::InvalidInput("Runtime did not retain diagnostics".into())), + DiagnosticsDisposition::Retain => { + let seconds=plan.diagnostics_retention_seconds.ok_or_else(||StoreError::InvalidInput("diagnostics retention deadline missing".into()))?; + let seconds=i64::try_from(seconds).map_err(|_|StoreError::InvalidInput("diagnostics retention deadline overflow".into()))?; + let expires_at=(Utc::now()+chrono::Duration::seconds(seconds)).to_rfc3339(); + tx.execute("INSERT OR IGNORE INTO worker_diagnostics_archives(operation_id,workspace_id,runtime_id,worker_id,policy_id,policy_revision,committed_at,expires_at) VALUES(?1,?2,?3,?4,?5,?6,?7,?8)",params![operation_id,workspace_id,plan.worker.runtime_id,plan.worker.worker_id,plan.policy_id,plan.policy_revision,now,expires_at])?; + } + DiagnosticsDisposition::Purge => {} + } + if plan.metadata_disposition==MetadataDisposition::Tombstone{ + tx.execute("INSERT OR IGNORE INTO worker_tombstones(workspace_id,runtime_id,worker_id,display_name,profile,worker_created_at,removed_at,archive_id,policy_id,policy_revision,operation_id) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,worker.display_name,worker.profile,worker.created_at,now,plan.archive_id,plan.policy_id,plan.policy_revision,operation_id])?; + } + let deleted=tx.execute("DELETE FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND runtime_worker_id=?3 AND updated_at=?4",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,plan.worker_revision])?; + if deleted!=1{return Err(StoreError::InvalidInput(format!("stale:{}:removal fence changed",plan.plan_id)));} + tx.execute("UPDATE worker_removal_operations SET state='succeeded',failure_category=NULL,updated_at=?1 WHERE operation_id=?2",params![now,operation_id])?; + tx.execute("INSERT OR IGNORE INTO worker_retention_audit_events(event_id,operation_id,workspace_id,event_kind,detail,created_at) VALUES(?1,?2,?3,'worker_removed',?4,?5)",params![stable("wre",operation_id),operation_id,workspace_id,format!("runtime_id={} worker_id={} session={} metadata={} diagnostics={}",plan.worker.runtime_id,plan.worker.worker_id,sess(plan.session_disposition),meta(plan.metadata_disposition),diag(plan.diagnostics_disposition)),now])?; + tx.commit()?;plan.state=WorkerRemovalPlanState::Succeeded;plan.updated_at=now;Ok(plan) + }).map_err(map_error) + } + + pub fn record_worker_orphan_diagnostic( + &self, + d: &WorkerOrphanDiagnostic, + ) -> Result<(), WorkerRetentionError> { + bounded("orphan category", &d.category, 160)?; + bounded("orphan detail", &d.detail, 2000)?; + self.with_conn(|conn|{conn.execute("INSERT OR IGNORE INTO worker_orphan_diagnostics(diagnostic_id,workspace_id,runtime_id,worker_id,category,detail,observed_at) VALUES(?1,?2,?3,?4,?5,?6,?7)",params![d.diagnostic_id,d.workspace_id,d.runtime_id,d.worker_id,d.category,d.detail,d.observed_at])?;Ok(())})?; + Ok(()) + } + + pub fn worker_tombstone( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> crate::Result> { + self.with_conn(|conn|conn.query_row("SELECT display_name,profile,worker_created_at,removed_at,archive_id,policy_id,policy_revision,operation_id FROM worker_tombstones WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![workspace_id,worker.runtime_id,worker.worker_id],|r|Ok(WorkerTombstone{workspace_id:workspace_id.into(),worker:worker.clone(),display_name:r.get(0)?,profile:r.get(1)?,created_at:r.get(2)?,removed_at:r.get(3)?,archive_id:r.get(4)?,policy_id:r.get(5)?,policy_revision:r.get::<_,i64>(6)? as u64,operation_id:r.get(7)?})).optional().map_err(StoreError::from)) + } +} + +#[derive(Clone)] +struct WorkerRow { + display_name: String, + profile: Option, + retention_state: String, + created_at: String, + updated_at: String, +} +fn load_worker(c: &Connection, w: &str, r: &RuntimeWorkerRef) -> crate::Result> { + c.query_row("SELECT display_name,profile,retention_state,created_at,updated_at FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND runtime_worker_id=?3",params![w,r.runtime_id,r.worker_id],|x|Ok(WorkerRow{display_name:x.get(0)?,profile:x.get(1)?,retention_state:x.get(2)?,created_at:x.get(3)?,updated_at:x.get(4)?})).optional().map_err(StoreError::from) +} +fn load_policy(c: &Connection, w: &str) -> crate::Result> { + c.query_row( + "SELECT p.policy_id,p.revision,r.session_disposition,r.metadata_disposition, + r.archive_retention_kind,r.archive_retention_seconds, + r.diagnostics_disposition,r.diagnostics_retention_seconds,r.created_at,p.updated_at + FROM workspace_worker_retention_policies p + JOIN workspace_worker_retention_policy_revisions r + ON r.workspace_id=p.workspace_id AND r.policy_id=p.policy_id AND r.revision=p.revision + WHERE p.workspace_id=?1", + params![w], + |row| { + let session: String = row.get(2)?; + let metadata: String = row.get(3)?; + let archive_kind: String = row.get(4)?; + let archive_seconds: Option = row.get(5)?; + let diagnostics: String = row.get(6)?; + Ok(WorkerRetentionPolicy { + workspace_id: w.into(), + policy_id: row.get(0)?, + revision: row.get::<_, i64>(1)? as u64, + session_disposition: parse_s(&session)?, + metadata_disposition: parse_m(&metadata)?, + archive_retention: parse_archive(&archive_kind, archive_seconds)?, + diagnostics_disposition: parse_d(&diagnostics)?, + diagnostics_retention_seconds: row.get::<_, Option>(7)?.map(|v| v as u64), + created_at: row.get(8)?, + updated_at: row.get(9)?, + }) + }, + ) + .optional() + .map_err(StoreError::from) +} +fn load_plan(c: &Connection, id: &str) -> crate::Result> { + load_plan_q(c, "plan_id", id) +} +fn load_plan_op(c: &Connection, id: &str) -> crate::Result> { + load_plan_q(c, "operation_id", id) +} +fn load_plan_q(c: &Connection, key: &str, id: &str) -> crate::Result> { + let query = format!( + "SELECT plan_id,operation_id,input_fingerprint,workspace_id,runtime_id,worker_id, + worker_revision,run_generation,policy_id,policy_revision,session_disposition, + metadata_disposition,archive_retention_kind,archive_retention_seconds, + diagnostics_disposition,diagnostics_retention_seconds,archive_id,blockers_json, + state,reason,created_at,updated_at + FROM worker_removal_operations WHERE {key}=?1" + ); + c.query_row(&query, params![id], |row| { + let session: String = row.get(10)?; + let metadata: String = row.get(11)?; + let archive_kind: String = row.get(12)?; + let archive_seconds: Option = row.get(13)?; + let diagnostics: String = row.get(14)?; + let blockers: String = row.get(17)?; + let state: String = row.get(18)?; + Ok(WorkerRemovalPlan { + plan_id: row.get(0)?, + operation_id: row.get(1)?, + input_fingerprint: row.get(2)?, + workspace_id: row.get(3)?, + worker: RuntimeWorkerRef { + runtime_id: row.get(4)?, + worker_id: row.get(5)?, + }, + worker_revision: row.get(6)?, + run_generation: row.get::<_, i64>(7)? as u64, + policy_id: row.get(8)?, + policy_revision: row.get::<_, i64>(9)? as u64, + session_disposition: parse_s(&session)?, + metadata_disposition: parse_m(&metadata)?, + archive_retention: parse_archive(&archive_kind, archive_seconds)?, + diagnostics_disposition: parse_d(&diagnostics)?, + diagnostics_retention_seconds: row.get::<_, Option>(15)?.map(|v| v as u64), + archive_id: row.get(16)?, + blockers: serde_json::from_str(&blockers).map_err(|error| { + rusqlite::Error::FromSqlConversionFailure( + 17, + rusqlite::types::Type::Text, + Box::new(error), + ) + })?, + state: parse_state(&state)?, + reason: row.get(19)?, + created_at: row.get(20)?, + updated_at: row.get(21)?, + }) + }) + .optional() + .map_err(StoreError::from) +} +fn mark_stale( + tx: &rusqlite::Transaction<'_>, + plan: &WorkerRemovalPlan, + reason: &str, +) -> crate::Result<()> { + tx.execute("UPDATE worker_removal_operations SET state='stale',failure_category=?1,updated_at=?2 WHERE operation_id=?3",params![reason,Utc::now().to_rfc3339(),plan.operation_id])?; + Ok(()) +} +fn stale_error(plan: &WorkerRemovalPlan, reason: &str) -> StoreError { + StoreError::InvalidInput(format!("stale:{}:{reason}", plan.plan_id)) +} +fn fingerprint( + r: &WorkerRemovalPlanRequest, + i: &WorkerRetentionInventory, + p: &WorkerRetentionPolicy, + b: &[WorkerRemovalBlocker], +) -> crate::Result { + serde_json::to_vec(&serde_json::json!([ + r.workspace_id, + r.worker.runtime_id, + r.worker.worker_id, + r.expected_worker_revision, + i.run_generation, + i.session_id, + i.segment_ids, + p.policy_id, + p.revision, + p.session_disposition, + p.metadata_disposition, + p.archive_retention, + p.diagnostics_disposition, + p.diagnostics_retention_seconds, + b, + r.reason + ])) + .map(|v| hash(&v)) + .map_err(|e| StoreError::InvalidInput(e.to_string())) +} +fn hash(b: &[u8]) -> String { + Sha256::digest(b) + .iter() + .map(|v| format!("{v:02x}")) + .collect() +} +fn stable(p: &str, v: &str) -> String { + format!("{p}_{}", &hash(v.as_bytes())[..32]) +} +fn validate_plan( + r: &WorkerRemovalPlanRequest, + i: &WorkerRetentionInventory, +) -> Result<(), WorkerRetentionError> { + bounded("workspace", &r.workspace_id, 160)?; + bounded("revision", &r.expected_worker_revision, 256)?; + bounded("reason", &r.reason, 2000)?; + if i.workspace_id != r.workspace_id + || i.runtime_id != r.worker.runtime_id + || i.worker_id.to_string() != r.worker.worker_id + { + return Err(WorkerRetentionError::CrossWorkspace); + } + Ok(()) +} +fn validate_policy(u: &WorkerRetentionPolicyUpdate) -> Result<(), WorkerRetentionError> { + bounded("policy id", &u.policy_id, 160)?; + if matches!( + u.archive_retention, + ArchiveRetention::ForSeconds { seconds: 0 } + ) { + return Err(WorkerRetentionError::Invalid( + "archive retention seconds must be positive".to_string(), + )); + } + match (u.diagnostics_disposition, u.diagnostics_retention_seconds) { + (DiagnosticsDisposition::Purge, None) | (DiagnosticsDisposition::Retain, Some(1..)) => { + Ok(()) + } + _ => Err(WorkerRetentionError::Invalid( + "diagnostics retention/disposition mismatch".into(), + )), + } +} +fn bounded(k: &str, v: &str, n: usize) -> Result<(), WorkerRetentionError> { + if v.trim().is_empty() || v.len() > n { + Err(WorkerRetentionError::Invalid(format!( + "{k} must be non-empty and at most {n} bytes" + ))) + } else { + Ok(()) + } +} +fn map_error(e: StoreError) -> WorkerRetentionError { + let StoreError::InvalidInput(m) = &e else { + return WorkerRetentionError::Store(e); + }; + if let Some(x) = m.strip_prefix("policy-missing:") { + return WorkerRetentionError::PolicyMissing { + workspace_id: x.into(), + }; + } + if let Some(x) = m.strip_prefix("policy-conflict:") { + let mut s = x.split(':'); + return WorkerRetentionError::PolicyRevisionConflict { + expected: s.next().and_then(|v| v.parse().ok()).unwrap_or(0), + actual: s.next().and_then(|v| v.parse().ok()).unwrap_or(0), + }; + } + if m == "cross-workspace" { + return WorkerRetentionError::CrossWorkspace; + } + if m == "worker-missing" { + return WorkerRetentionError::WorkerNotFound; + } + if let Some(x) = m.strip_prefix("worker-conflict:") { + let mut s = x.splitn(2, ':'); + return WorkerRetentionError::WorkerRevisionConflict { + expected: s.next().unwrap_or_default().into(), + actual: s.next().unwrap_or_default().into(), + }; + } + if let Some(x) = m.strip_prefix("fingerprint:") { + return WorkerRetentionError::OperationFingerprintConflict { + operation_id: x.into(), + }; + } + if let Some(x) = m.strip_prefix("blocked:") { + return WorkerRetentionError::Blocked(serde_json::from_str(x).unwrap_or_default()); + } + if let Some(x) = m.strip_prefix("stale:") { + let mut s = x.splitn(2, ':'); + return WorkerRetentionError::StalePlan { + plan_id: s.next().unwrap_or_default().into(), + reason: s.next().unwrap_or_default().into(), + }; + } + WorkerRetentionError::Store(e) +} +fn archive_kind(value: ArchiveRetention) -> &'static str { + match value { + ArchiveRetention::Forever => "forever", + ArchiveRetention::ForSeconds { .. } => "for_seconds", + } +} +fn archive_seconds(value: ArchiveRetention) -> Option { + match value { + ArchiveRetention::Forever => None, + ArchiveRetention::ForSeconds { seconds } => Some(seconds), + } +} +fn parse_archive(kind: &str, seconds: Option) -> rusqlite::Result { + match (kind, seconds) { + ("forever", None) => Ok(ArchiveRetention::Forever), + ("for_seconds", Some(seconds)) if seconds > 0 => Ok(ArchiveRetention::ForSeconds { + seconds: seconds as u64, + }), + _ => Err(bad("archive retention", kind)), + } +} +fn sess(v: SessionDisposition) -> &'static str { + match v { + SessionDisposition::Archive => "archive", + SessionDisposition::Purge => "purge", + } +} +fn meta(v: MetadataDisposition) -> &'static str { + match v { + MetadataDisposition::Tombstone => "tombstone", + MetadataDisposition::Purge => "purge", + } +} +fn diag(v: DiagnosticsDisposition) -> &'static str { + match v { + DiagnosticsDisposition::Purge => "purge", + DiagnosticsDisposition::Retain => "retain", + } +} +fn state_s(v: WorkerRemovalPlanState) -> &'static str { + match v { + WorkerRemovalPlanState::Planned => "planned", + WorkerRemovalPlanState::Blocked => "blocked", + WorkerRemovalPlanState::Executing => "executing", + WorkerRemovalPlanState::Failed => "failed", + WorkerRemovalPlanState::Stale => "stale", + WorkerRemovalPlanState::Succeeded => "succeeded", + } +} +fn bad(k: &str, v: &str) -> rusqlite::Error { + rusqlite::Error::FromSqlConversionFailure( + 0, + rusqlite::types::Type::Text, + format!("invalid {k}: {v}").into(), + ) +} +fn parse_s(v: &str) -> rusqlite::Result { + match v { + "archive" => Ok(SessionDisposition::Archive), + "purge" => Ok(SessionDisposition::Purge), + _ => Err(bad("session", v)), + } +} +fn parse_m(v: &str) -> rusqlite::Result { + match v { + "tombstone" => Ok(MetadataDisposition::Tombstone), + "purge" => Ok(MetadataDisposition::Purge), + _ => Err(bad("metadata", v)), + } +} +fn parse_d(v: &str) -> rusqlite::Result { + match v { + "purge" => Ok(DiagnosticsDisposition::Purge), + "retain" => Ok(DiagnosticsDisposition::Retain), + _ => Err(bad("diagnostics", v)), + } +} +fn parse_state(v: &str) -> rusqlite::Result { + match v { + "planned" => Ok(WorkerRemovalPlanState::Planned), + "blocked" => Ok(WorkerRemovalPlanState::Blocked), + "executing" => Ok(WorkerRemovalPlanState::Executing), + "failed" => Ok(WorkerRemovalPlanState::Failed), + "stale" => Ok(WorkerRemovalPlanState::Stale), + "succeeded" => Ok(WorkerRemovalPlanState::Succeeded), + _ => Err(bad("state", v)), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::store::{ControlPlaneStore, TicketWorkerAssignmentRecord, WorkerRegistryRecord}; + use worker_runtime::identity::WorkerId; + fn setup() -> SqliteWorkspaceStore { + let s = SqliteWorkspaceStore::in_memory().unwrap(); + s.with_conn(|c|{c.execute("INSERT INTO workspaces(workspace_id,display_name,state,created_at,updated_at)VALUES('w','W','active','t','t')",[])?;c.execute("INSERT INTO worker_registry(workspace_id,runtime_id,runtime_worker_id,display_name,profile,retention_state,created_at,updated_at)VALUES('w','r',1,'one','builtin:coder','normal','created','rev1')",[])?;Ok(())}).unwrap(); + s + } + fn inv() -> WorkerRetentionInventory { + WorkerRetentionInventory { + workspace_id: "w".into(), + runtime_id: "r".into(), + worker_id: WorkerId::new(1), + run_generation: 2, + session_id: Some("s".into()), + segment_ids: vec!["a".into()], + session_bytes: 1, + diagnostics_bytes: 0, + } + } + fn req() -> WorkerRemovalPlanRequest { + WorkerRemovalPlanRequest { + workspace_id: "w".into(), + worker: RuntimeWorkerRef { + runtime_id: "r".into(), + worker_id: "1".into(), + }, + expected_worker_revision: "rev1".into(), + reason: "cleanup".into(), + } + } + #[test] + fn conservative_seed_and_policy_conflict() { + let s = setup(); + let p = s.worker_retention_policy("w").unwrap().unwrap(); + assert_eq!(p.session_disposition, SessionDisposition::Archive); + assert_eq!(p.archive_retention, ArchiveRetention::Forever); + let u = WorkerRetentionPolicyUpdate { + policy_id: "p".into(), + session_disposition: SessionDisposition::Archive, + metadata_disposition: MetadataDisposition::Purge, + archive_retention: ArchiveRetention::ForSeconds { seconds: 3_600 }, + diagnostics_disposition: DiagnosticsDisposition::Purge, + diagnostics_retention_seconds: None, + }; + let updated = s.update_worker_retention_policy("w", 1, &u).unwrap(); + assert_eq!(updated.revision, 2); + assert_eq!( + updated.archive_retention, + ArchiveRetention::ForSeconds { seconds: 3_600 } + ); + assert!(matches!( + s.update_worker_retention_policy("w", 1, &u), + Err(WorkerRetentionError::PolicyRevisionConflict { .. }) + )); + } + #[test] + fn deterministic_plan_hold_and_cross_workspace() { + let s = setup(); + let a = s.plan_worker_removal(&req(), &inv()).unwrap(); + let b = s.plan_worker_removal(&req(), &inv()).unwrap(); + assert_eq!(a.plan_id, b.plan_id); + s.with_conn(|c| { + c.execute( + "UPDATE worker_registry SET retention_state='pinned' WHERE workspace_id='w'", + [], + )?; + Ok(()) + }) + .unwrap(); + let mut q = req(); + q.expected_worker_revision = "rev1".into(); + let p = s.plan_worker_removal(&q, &inv()).unwrap(); + assert_eq!(p.blockers, vec![WorkerRemovalBlocker::Hold]); + assert!(matches!( + s.begin_worker_removal("w", &p.plan_id, &p.input_fingerprint), + Err(WorkerRetentionError::Blocked(_)) + )); + let mut i = inv(); + i.workspace_id = "other".into(); + assert!(matches!( + s.plan_worker_removal(&req(), &i), + Err(WorkerRetentionError::CrossWorkspace) + )); + i.workspace_id = "w".into(); + i.runtime_id = "other-runtime".into(); + assert!(matches!( + s.plan_worker_removal(&req(), &i), + Err(WorkerRetentionError::CrossWorkspace) + )); + } + #[test] + fn stale_policy_and_failed_retry_restore_fence() { + let s = setup(); + let p = s.plan_worker_removal(&req(), &inv()).unwrap(); + let u = WorkerRetentionPolicyUpdate { + policy_id: "new".into(), + session_disposition: SessionDisposition::Purge, + metadata_disposition: MetadataDisposition::Purge, + archive_retention: ArchiveRetention::Forever, + diagnostics_disposition: DiagnosticsDisposition::Purge, + diagnostics_retention_seconds: None, + }; + s.update_worker_retention_policy("w", 1, &u).unwrap(); + assert!(matches!( + s.begin_worker_removal("w", &p.plan_id, &p.input_fingerprint), + Err(WorkerRetentionError::StalePlan { .. }) + )); + let state: String = s + .with_conn(|conn| { + conn.query_row( + "SELECT state FROM worker_removal_operations WHERE plan_id=?1", + params![p.plan_id], + |row| row.get(0), + ) + .map_err(StoreError::from) + }) + .unwrap(); + assert_eq!(state, "stale"); + assert!(matches!( + s.begin_worker_removal("w", &p.plan_id, &p.input_fingerprint), + Err(WorkerRetentionError::StalePlan { .. }) + )); + } + #[test] + fn prepared_execution_is_derived_from_pinned_plan_generation() { + let s = setup(); + let plan = s.plan_worker_removal(&req(), &inv()).unwrap(); + let prepared = s + .prepare_worker_removal_execution("w", &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + assert_eq!(prepared.runtime_request.expected_run_generation, 2); + assert_eq!( + prepared.runtime_request.session_disposition, + SessionDisposition::Archive + ); + assert_eq!(prepared.runtime_request.policy_revision, 1); + assert_eq!(prepared.runtime_request.worker_id, WorkerId::new(1)); + } + + #[test] + fn purge_tombstone_commit_is_idempotent() { + let s = setup(); + s.with_conn(|conn| { + conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,runtime_id,worker_id,assigned_by,assigned_at) VALUES('w','ticket-old','assignment-old','r','1','test','t')", [])?; + Ok(()) + }).unwrap(); + let p = s.plan_worker_removal(&req(), &inv()).unwrap(); + s.begin_worker_removal("w", &p.plan_id, &p.input_fingerprint) + .unwrap(); + let result = WorkerRetentionExecutionResult { + operation_id: p.operation_id.clone(), + input_fingerprint: p.input_fingerprint.clone(), + worker_id: WorkerId::new(1), + session_disposition: p.session_disposition, + diagnostics_disposition: p.diagnostics_disposition, + archive: Some(worker_runtime::retention::WorkerSessionArchiveManifest { + schema_version: 1, + archive_id: p.archive_id.clone().unwrap(), + workspace_id: "w".into(), + source_runtime_id: "r".into(), + source_worker_id: WorkerId::new(1), + source_session_id: "s".into(), + segment_ids: vec!["a".into()], + source_created_at: "created".into(), + removed_at: "removed".into(), + archived_at_unix_seconds: 1, + effective_profile: None, + retention_class: None, + content_checksum_sha256: "sum".into(), + content_bytes: 1, + content_file_count: 1, + policy_id: p.policy_id.clone(), + policy_revision: p.policy_revision, + operation_id: p.operation_id.clone(), + input_fingerprint: p.input_fingerprint.clone(), + }), + source_removed: true, + diagnostics_retained: false, + }; + assert_eq!( + s.commit_worker_removal("w", &p.operation_id, &p.input_fingerprint, &result) + .unwrap() + .state, + WorkerRemovalPlanState::Succeeded + ); + assert!(s.worker_tombstone("w", &p.worker).unwrap().is_some()); + assert_eq!( + s.commit_worker_removal("w", &p.operation_id, &p.input_fingerprint, &result) + .unwrap() + .state, + WorkerRemovalPlanState::Succeeded + ); + let historical: i64 = s.with_conn(|conn| conn.query_row( + "SELECT COUNT(*) FROM ticket_worker_assignments WHERE workspace_id='w' AND assignment_id='assignment-old'", + [], + |row| row.get(0), + ).map_err(StoreError::from)).unwrap(); + assert_eq!(historical, 1); + } + #[test] + fn assignment_and_orphan_are_authoritative() { + let s = setup(); + s.with_conn(|c|{c.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,runtime_id,worker_id,assigned_by,assigned_at)VALUES('w','ticket','assignment','r','1','test','t')",[])?;c.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,runtime_id,worker_id,updated_at)VALUES('w','ticket','assignment','r','1','t')",[])?;Ok(())}).unwrap(); + let p = s.plan_worker_removal(&req(), &inv()).unwrap(); + assert!( + matches!(&p.blockers[..],[WorkerRemovalBlocker::CurrentAssignment{assignment_id,ticket_id}] if assignment_id=="assignment"&&ticket_id=="ticket") + ); + let d = WorkerOrphanDiagnostic { + diagnostic_id: "orphan".into(), + workspace_id: "w".into(), + runtime_id: "r".into(), + worker_id: "missing".into(), + category: "runtime_without_catalog".into(), + detail: "bounded diagnostic".into(), + observed_at: "t".into(), + }; + s.record_worker_orphan_diagnostic(&d).unwrap(); + let n: i64 = s + .with_conn(|c| { + c.query_row( + "SELECT COUNT(*) FROM worker_orphan_diagnostics WHERE diagnostic_id='orphan'", + [], + |r| r.get(0), + ) + .map_err(StoreError::from) + }) + .unwrap(); + assert_eq!(n, 1); + } + #[test] + fn concurrent_plan_converges_and_purge_omits_tombstone() { + let s = std::sync::Arc::new(setup()); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(3)); + let handles = (0..2) + .map(|_| { + let s = s.clone(); + let b = barrier.clone(); + std::thread::spawn(move || { + b.wait(); + s.plan_worker_removal(&req(), &inv()).unwrap() + }) + }) + .collect::>(); + barrier.wait(); + let plans = handles + .into_iter() + .map(|h| h.join().unwrap()) + .collect::>(); + assert_eq!(plans[0].plan_id, plans[1].plan_id); + let s = setup(); + let u = WorkerRetentionPolicyUpdate { + policy_id: "purge".into(), + session_disposition: SessionDisposition::Purge, + metadata_disposition: MetadataDisposition::Purge, + archive_retention: ArchiveRetention::Forever, + diagnostics_disposition: DiagnosticsDisposition::Purge, + diagnostics_retention_seconds: None, + }; + s.update_worker_retention_policy("w", 1, &u).unwrap(); + let p = s.plan_worker_removal(&req(), &inv()).unwrap(); + s.begin_worker_removal("w", &p.plan_id, &p.input_fingerprint) + .unwrap(); + let r = WorkerRetentionExecutionResult { + operation_id: p.operation_id.clone(), + input_fingerprint: p.input_fingerprint.clone(), + worker_id: WorkerId::new(1), + session_disposition: SessionDisposition::Purge, + diagnostics_disposition: DiagnosticsDisposition::Purge, + archive: None, + source_removed: true, + diagnostics_retained: false, + }; + s.commit_worker_removal("w", &p.operation_id, &p.input_fingerprint, &r) + .unwrap(); + assert!(s.worker_tombstone("w", &p.worker).unwrap().is_none()); + } + #[test] + fn commit_requires_executing_state_and_exact_runtime_result() { + let store = setup(); + let plan = store.plan_worker_removal(&req(), &inv()).unwrap(); + let mut result = WorkerRetentionExecutionResult { + operation_id: plan.operation_id.clone(), + input_fingerprint: plan.input_fingerprint.clone(), + worker_id: WorkerId::new(1), + session_disposition: plan.session_disposition, + diagnostics_disposition: plan.diagnostics_disposition, + archive: None, + source_removed: true, + diagnostics_retained: false, + }; + assert!(matches!( + store.commit_worker_removal("w", &plan.operation_id, &plan.input_fingerprint, &result), + Err(WorkerRetentionError::StalePlan { .. }) + )); + store + .begin_worker_removal("w", &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + result.worker_id = WorkerId::new(2); + assert!( + store + .commit_worker_removal("w", &plan.operation_id, &plan.input_fingerprint, &result) + .is_err() + ); + let count: i64 = store.with_conn(|conn| conn.query_row( + "SELECT COUNT(*) FROM worker_registry WHERE workspace_id='w' AND runtime_id='r' AND runtime_worker_id=1", + [], + |row| row.get(0), + ).map_err(StoreError::from)).unwrap(); + assert_eq!(count, 1); + } + + #[test] + fn execution_fence_blocks_stale_upsert_and_new_assignment() { + let store = setup(); + let plan = store.plan_worker_removal(&req(), &inv()).unwrap(); + store + .begin_worker_removal("w", &plan.plan_id, &plan.input_fingerprint) + .unwrap(); + let stale = WorkerRegistryRecord { + workspace_id: "w".into(), + worker: RuntimeWorkerRef { + runtime_id: "r".into(), + worker_id: "1".into(), + }, + display_name: "stale".into(), + profile: None, + retention_state: "normal".into(), + transcript_ref: None, + session_ref: None, + summary_ref: None, + diagnostics_ref: None, + created_at: "created".into(), + updated_at: "rev2".into(), + }; + store.upsert_worker_registry(&stale).unwrap(); + let revision: String = store.with_conn(|conn| conn.query_row( + "SELECT updated_at FROM worker_registry WHERE workspace_id='w' AND runtime_id='r' AND runtime_worker_id=1", + [], + |row| row.get(0), + ).map_err(StoreError::from)).unwrap(); + assert_eq!(revision, "rev1"); + + let assignment = TicketWorkerAssignmentRecord { + workspace_id: "w".into(), + ticket_id: "new-ticket".into(), + assignment_id: "new-assignment".into(), + worker: RuntimeWorkerRef { + runtime_id: "r".into(), + worker_id: "1".into(), + }, + assigned_by: "test".into(), + assigned_at: "t".into(), + }; + assert!( + store + .set_current_ticket_worker_assignment( + &assignment, + None, + "event", + "assignment-operation", + false, + ) + .is_err() + ); + } + + #[test] + fn old_schema_upgrade_seeds_existing_workspace() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("server.db"); + { + let s = SqliteWorkspaceStore::open(&path).unwrap(); + s.with_conn(|c|{c.execute("INSERT INTO workspaces(workspace_id,display_name,state,created_at,updated_at)VALUES('legacy','Legacy','active','old','old')",[])?;c.execute_batch("DROP TRIGGER seed_worker_retention_policy_after_workspace_insert;DROP TABLE worker_retention_audit_events;DROP TABLE worker_tombstones;DROP TABLE worker_session_archives;DROP TABLE worker_diagnostics_archives;DROP TABLE worker_orphan_diagnostics;DROP TABLE worker_removal_operations;DROP TABLE workspace_worker_retention_policies;DROP TABLE workspace_worker_retention_policy_revisions;DELETE FROM __yoi_schema_migrations WHERE version=28;")?;Ok(())}).unwrap(); + } + let reopened = SqliteWorkspaceStore::open(&path).unwrap(); + let p = reopened.worker_retention_policy("legacy").unwrap().unwrap(); + assert_eq!(p.policy_id, CONSERVATIVE_POLICY_ID); + assert_eq!(p.session_disposition, SessionDisposition::Archive); + assert_eq!(p.metadata_disposition, MetadataDisposition::Tombstone); + } +} diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 92cdfd03..8b3786e8 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -156,6 +156,11 @@ const MIGRATIONS: &[Migration] = &[ name: "scope Repository identity and references by Workspace", apply: scope_repository_identity_by_workspace, }, + Migration { + version: 28, + name: "create Worker retention authority", + apply: crate::retention::create_worker_retention_tables, + }, ]; struct Migration { @@ -772,7 +777,7 @@ impl SqliteWorkspaceStore { }) } - fn with_conn(&self, f: impl FnOnce(&Connection) -> Result) -> Result { + pub(crate) fn with_conn(&self, f: impl FnOnce(&Connection) -> Result) -> Result { let conn = self .conn .lock() @@ -780,6 +785,17 @@ impl SqliteWorkspaceStore { f(&conn) } + pub(crate) fn with_conn_mut( + &self, + f: impl FnOnce(&mut Connection) -> Result, + ) -> Result { + let mut conn = self + .conn + .lock() + .map_err(|_| Error::Store("sqlite connection lock poisoned".to_string()))?; + f(&mut conn) + } + pub fn upsert_trusted_runtime(&self, record: &TrustedRuntimeRecord) -> Result<()> { self.with_conn(|conn| { conn.execute( @@ -1919,6 +1935,23 @@ impl ControlPlaneStore for SqliteWorkspaceStore { fn upsert_worker_registry(&self, record: &WorkerRegistryRecord) -> Result<()> { self.with_conn(|conn| { + let removal_blocks_upsert: bool = conn.query_row( + "SELECT EXISTS( + SELECT 1 FROM worker_removal_operations + WHERE workspace_id = ?1 AND runtime_id = ?2 + AND CAST(worker_id AS INTEGER) = ?3 + AND state IN ('executing', 'failed', 'succeeded') + )", + params![ + record.workspace_id, + record.worker.runtime_id, + record.worker.worker_id + ], + |row| row.get(0), + )?; + if removal_blocks_upsert { + return Ok(()); + } conn.execute( r#"INSERT INTO worker_registry ( workspace_id, runtime_id, runtime_worker_id, display_name, profile, @@ -1937,7 +1970,14 @@ impl ControlPlaneStore for SqliteWorkspaceStore { session_ref = excluded.session_ref, summary_ref = excluded.summary_ref, diagnostics_ref = excluded.diagnostics_ref, - updated_at = excluded.updated_at"#, + updated_at = excluded.updated_at + WHERE NOT EXISTS ( + SELECT 1 FROM worker_removal_operations retention + WHERE retention.workspace_id = excluded.workspace_id + AND retention.runtime_id = excluded.runtime_id + AND CAST(retention.worker_id AS INTEGER) = excluded.runtime_worker_id + AND retention.state IN ('executing', 'failed', 'succeeded') + )"#, params![ record.workspace_id, record.worker.runtime_id, @@ -2006,7 +2046,13 @@ impl ControlPlaneStore for SqliteWorkspaceStore { let changed = conn.execute( r#"UPDATE worker_registry SET retention_state = ?4, updated_at = ?5 - WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3"#, + WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3 + AND NOT EXISTS ( + SELECT 1 FROM worker_removal_operations retention + WHERE retention.workspace_id = ?1 AND retention.runtime_id = ?2 + AND CAST(retention.worker_id AS INTEGER) = ?3 + AND retention.state IN ('executing', 'failed') + )"#, params![ workspace_id, worker.runtime_id, @@ -2192,6 +2238,28 @@ impl ControlPlaneStore for SqliteWorkspaceStore { ) -> Result { self.with_conn(|conn| { let tx = conn.unchecked_transaction()?; + let removal_blocks_assignment: bool = tx.query_row( + "SELECT EXISTS( + SELECT 1 FROM worker_removal_operations + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 + AND state IN ('executing', 'failed', 'succeeded') + UNION ALL + SELECT 1 FROM worker_tombstones + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 + )", + params![ + record.workspace_id, + record.worker.runtime_id, + record.worker.worker_id + ], + |row| row.get(0), + )?; + if removal_blocks_assignment { + return Err(Error::TicketAssignmentConflict(format!( + "Worker {}/{} is being retained or has been removed", + record.worker.runtime_id, record.worker.worker_id + ))); + } let mut reserved_operation = false; if let Some(existing) = read_assignment_operation(&tx, &record.workspace_id, operation_id)? @@ -4847,7 +4915,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 27); + assert_eq!(current_schema_version(&conn).unwrap(), 28); assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap()); } @@ -4880,7 +4948,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 27); + assert_eq!(current_schema_version(&conn).unwrap(), 28); assert!(table_exists(&conn, "flow_sources").unwrap()); assert!(table_exists(&conn, "flow_source_revisions").unwrap()); assert!(!table_exists(&conn, "flow_instances").unwrap()); @@ -4947,7 +5015,7 @@ INSERT INTO worker_workdir_attachment_reservations ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 27); + assert_eq!(current_schema_version(&conn).unwrap(), 28); let repositories_sql: String = conn .query_row( "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'", @@ -5127,7 +5195,7 @@ INSERT INTO workdir_registry ( let db = dir.path().join("control-plane.sqlite"); let store = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 27); + assert_eq!(store.schema_version().await.unwrap(), 28); assert!( !store .with_conn(|conn| table_exists(conn, "worker_workspace_credentials")) @@ -5144,7 +5212,7 @@ INSERT INTO workdir_registry ( store.upsert_workspace(&record).await.unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(reopened.schema_version().await.unwrap(), 27); + assert_eq!(reopened.schema_version().await.unwrap(), 28); assert_eq!( reopened.get_workspace("local-dev").await.unwrap(), Some(record) @@ -5691,7 +5759,7 @@ INSERT INTO workdir_registry ( .unwrap(); let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 27); + assert_eq!(store.schema_version().await.unwrap(), 28); store .with_conn(|conn| { @@ -5880,7 +5948,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn repository_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 27); + assert_eq!(store.schema_version().await.unwrap(), 28); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -5946,7 +6014,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn memory_authority_records_round_trip_and_close_staging() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 27); + assert_eq!(store.schema_version().await.unwrap(), 28); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -6209,7 +6277,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn account_and_login_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 27); + assert_eq!(store.schema_version().await.unwrap(), 28); let now = "2026-07-22T00:00:00Z".to_string(); let account = AccountRecord { account_id: "acct-user-alice".to_string(),