Merge branch 'work/00001KZPQSFQD-worker-retention' into orchestration-merge-request-domain

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;

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<WorkerRetentionInventory, RuntimeError> {
+        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<WorkerRetentionExecutionResult, RuntimeError> {
+        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<MutexGuard<'_, RuntimeState>, RuntimeError> {
         self.inner.lock().map_err(|_| RuntimeError::StatePoisoned)
     }
@@ -1673,6 +1790,9 @@ struct SubscriptionSink {
 struct RuntimeState {
     display_name: Option<String>,
     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<String>,
     #[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 {

crates/workspace-server/src/hosts.rs

@@ -1497,6 +1497,9 @@ impl EmbeddedWorkerRuntime {
 
     pub fn from_runtime(workspace_id: impl AsRef<str>, 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),

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;

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<T>(&self, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
+    pub(crate) fn with_conn<T>(&self, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
         let conn = self
             .conn
             .lock()
@@ -780,6 +785,17 @@ impl SqliteWorkspaceStore {
         f(&conn)
     }
 
+    pub(crate) fn with_conn_mut<T>(
+        &self,
+        f: impl FnOnce(&mut Connection) -> Result<T>,
+    ) -> Result<T> {
+        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<TicketWorkerAssignmentUpdate> {
         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(),