fix(runtime): fence shared task-store execution ownership

Use fresh transactions in the existing shared store and retained Runtime
scope/generation leases so multiple TUIs preserve live foreign work and
merge durable cancellation. Only eligible workers claim queued tasks;
interrupted Running work is recovered once without replaying execution.

Bind automation definitions and immutable admissions to trusted execution
scope, preserving ambiguous legacy records until explicit adoption.
Retain actual Engine and accepted receipt completion through shutdown and
canceled drains. Show unavailable inventory without inferring empty state.

Validation on exact final source:
- Production cargo check: passed in 53.94s.
- Fresh TUI libtest build: passed in 2m48s.
- Focused acceptance: 470 passed, 0 failed, 4 ignored in 39.31s.
  Three ignored subprocess helpers are exercised by parent tests; the
  opt-in Agent Mail release matrix was not run.
- cargo fmt --all --check and git diff --check passed.

npm test and npm run check:web were not run in this source lane; the
coordinating integration task owns those gates. No hosted CI, provider,
deployment, publication, or customer acceptance claim.

crates/tui/locales/ca.json

@@ -1578,6 +1578,8 @@
   "WorkflowReceiptProviderReported": "informat pel proveïdor",
   "WorkflowReceiptEstimated": "estimat",
   "SidebarTasksLabel": "Tasques",
+  "TaskOwnershipUnverified": "Responsable de l’execució sense verificar",
+  "TaskInventoryUnavailable": "L’emmagatzematge de tasques no està disponible; es mostra l’últim estat conegut.",
   "SidebarTodoLabel": "Pendents",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · en queden {remaining}",
   "SidebarStopControl": "atura",

crates/tui/locales/de.json

@@ -1578,6 +1578,8 @@
   "WorkflowReceiptProviderReported": "vom Anbieter gemeldet",
   "WorkflowReceiptEstimated": "geschätzt",
   "SidebarTasksLabel": "Aufgaben",
+  "TaskOwnershipUnverified": "Ausführungsverantwortung ungeprüft",
+  "TaskInventoryUnavailable": "Der Aufgabenspeicher ist nicht verfügbar; der zuletzt bekannte Zustand wird angezeigt.",
   "SidebarTodoLabel": "Zu erledigen",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · {remaining} übrig",
   "SidebarStopControl": "Stopp",

crates/tui/locales/en.json

@@ -1601,6 +1601,8 @@
   "WorkflowReceiptProviderReported": "provider-reported",
   "WorkflowReceiptEstimated": "estimated",
   "SidebarTasksLabel": "Tasks",
+  "TaskOwnershipUnverified": "Execution ownership unverified",
+  "TaskInventoryUnavailable": "Task storage is unavailable; showing the last known state.",
   "SidebarTodoLabel": "To-do",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · {remaining} left",
   "SidebarStopControl": "stop",

crates/tui/locales/es-419.json

@@ -1601,6 +1601,8 @@
   "WorkflowReceiptProviderReported": "informado por proveedor",
   "WorkflowReceiptEstimated": "estimado",
   "SidebarTasksLabel": "Tareas",
+  "TaskOwnershipUnverified": "Responsable de ejecución sin verificar",
+  "TaskInventoryUnavailable": "El almacenamiento de tareas no está disponible; se muestra el último estado conocido.",
   "SidebarTodoLabel": "Pendientes",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · {remaining} restantes",
   "SidebarStopControl": "detener",

crates/tui/locales/fr.json

@@ -1578,6 +1578,8 @@
   "WorkflowReceiptProviderReported": "rapporté par le fournisseur",
   "WorkflowReceiptEstimated": "estimé",
   "SidebarTasksLabel": "Tâches",
+  "TaskOwnershipUnverified": "Responsable de l’exécution non vérifié",
+  "TaskInventoryUnavailable": "Le stockage des tâches est indisponible ; le dernier état connu est affiché.",
   "SidebarTodoLabel": "À faire",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · {remaining} restantes",
   "SidebarStopControl": "arrêter",

crates/tui/locales/hi.json

@@ -1578,6 +1578,8 @@
   "WorkflowReceiptProviderReported": "प्रदाता द्वारा बताया गया",
   "WorkflowReceiptEstimated": "अनुमानित",
   "SidebarTasksLabel": "कार्य",
+  "TaskOwnershipUnverified": "निष्पादन स्वामित्व सत्यापित नहीं है",
+  "TaskInventoryUnavailable": "कार्य संग्रह उपलब्ध नहीं है; अंतिम ज्ञात स्थिति दिखाई जा रही है।",
   "SidebarTodoLabel": "करने योग्य",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · {remaining} शेष",
   "SidebarStopControl": "रोकें",

crates/tui/locales/id.json

@@ -1578,6 +1578,8 @@
   "WorkflowReceiptProviderReported": "dilaporkan penyedia",
   "WorkflowReceiptEstimated": "perkiraan",
   "SidebarTasksLabel": "Tugas",
+  "TaskOwnershipUnverified": "Pemilik eksekusi belum diverifikasi",
+  "TaskInventoryUnavailable": "Penyimpanan tugas tidak tersedia; menampilkan status terakhir yang diketahui.",
   "SidebarTodoLabel": "Perlu dikerjakan",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · sisa {remaining}",
   "SidebarStopControl": "berhenti",

crates/tui/locales/ja.json

@@ -1601,6 +1601,8 @@
   "WorkflowReceiptProviderReported": "プロバイダー報告",
   "WorkflowReceiptEstimated": "推定",
   "SidebarTasksLabel": "タスク",
+  "TaskOwnershipUnverified": "実行主体を確認できません",
+  "TaskInventoryUnavailable": "タスクストレージにアクセスできないため、最後に確認した状態を表示しています。",
   "SidebarTodoLabel": "To-do",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · 残り{remaining}",
   "SidebarStopControl": "停止",

crates/tui/locales/ko.json

@@ -1601,6 +1601,8 @@
   "WorkflowReceiptProviderReported": "제공자 보고",
   "WorkflowReceiptEstimated": "추정",
   "SidebarTasksLabel": "작업",
+  "TaskOwnershipUnverified": "실행 주체 미확인",
+  "TaskInventoryUnavailable": "작업 저장소를 사용할 수 없어 마지막으로 확인된 상태를 표시합니다.",
   "SidebarTodoLabel": "할 일",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · {remaining}개 남음",
   "SidebarStopControl": "중지",

crates/tui/locales/pt-BR.json

@@ -1601,6 +1601,8 @@
   "WorkflowReceiptProviderReported": "informado pelo provedor",
   "WorkflowReceiptEstimated": "estimado",
   "SidebarTasksLabel": "Tarefas",
+  "TaskOwnershipUnverified": "Responsável pela execução não verificado",
+  "TaskInventoryUnavailable": "O armazenamento de tarefas está indisponível; exibindo o último estado conhecido.",
   "SidebarTodoLabel": "A fazer",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · {remaining} restantes",
   "SidebarStopControl": "parar",

crates/tui/locales/ru.json

@@ -1578,6 +1578,8 @@
   "WorkflowReceiptProviderReported": "по данным провайдера",
   "WorkflowReceiptEstimated": "оценка",
   "SidebarTasksLabel": "Задачи",
+  "TaskOwnershipUnverified": "Владелец выполнения не подтверждён",
+  "TaskInventoryUnavailable": "Хранилище задач недоступно; показано последнее известное состояние.",
   "SidebarTodoLabel": "Сделать",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · осталось {remaining}",
   "SidebarStopControl": "стоп",

crates/tui/locales/uk.json

@@ -1578,6 +1578,8 @@
   "WorkflowReceiptProviderReported": "за даними провайдера",
   "WorkflowReceiptEstimated": "оцінка",
   "SidebarTasksLabel": "Завдання",
+  "TaskOwnershipUnverified": "Власника виконання не підтверджено",
+  "TaskInventoryUnavailable": "Сховище завдань недоступне; показано останній відомий стан.",
   "SidebarTodoLabel": "До виконання",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · залишилося {remaining}",
   "SidebarStopControl": "стоп",

crates/tui/locales/vi.json

@@ -1601,6 +1601,8 @@
   "WorkflowReceiptProviderReported": "nhà cung cấp báo cáo",
   "WorkflowReceiptEstimated": "ước tính",
   "SidebarTasksLabel": "Nhiệm vụ",
+  "TaskOwnershipUnverified": "Chưa xác minh chủ thể thực thi",
+  "TaskInventoryUnavailable": "Không thể truy cập kho tác vụ; đang hiển thị trạng thái được ghi nhận gần nhất.",
   "SidebarTodoLabel": "Cần làm",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · còn {remaining}",
   "SidebarStopControl": "dừng",

crates/tui/locales/zh-Hans.json

@@ -1601,6 +1601,8 @@
   "WorkflowReceiptProviderReported": "提供商报告",
   "WorkflowReceiptEstimated": "估算",
   "SidebarTasksLabel": "任务",
+  "TaskOwnershipUnverified": "执行归属尚未验证",
+  "TaskInventoryUnavailable": "任务存储不可用，正在显示最后已知的状态。",
   "SidebarTodoLabel": "待办",
   "WorkSurfaceTodoProgress": "{label} {completed}/{total} · 剩余 {remaining}",
   "SidebarStopControl": "停止",

crates/tui/locales/zh-Hant.json

@@ -1619,6 +1619,8 @@
   "SidebarDestructiveArmed": "破壞性操作已就緒 — 按 Enter 確認，或清空草稿以取消",
   "SidebarStopControl": "停止",
   "SidebarTasksLabel": "任務",
+  "TaskOwnershipUnverified": "執行歸屬尚未驗證",
+  "TaskInventoryUnavailable": "任務儲存空間無法使用，正在顯示最後已知的狀態。",
   "SidebarTodoLabel": "待辦",
   "StartupDefaultNotSaved": "{setting}未能儲存：{error}",
   "StartupDefaultSubjectAll": "啟動預設值",

crates/tui/src/automation_manager.rs

@@ -27,9 +27,9 @@ use crate::utils::spawn_supervised;
 /// can build a fixed-id record directly (no create/delete id swap).
 // v2 pins provider identity. Older runtimes must reject a pinned definition
 // instead of silently sending its model through their current provider.
-pub(crate) const CURRENT_AUTOMATION_SCHEMA_VERSION: u32 = 2;
-const CURRENT_RUN_SCHEMA_VERSION: u32 = 2;
-const CURRENT_TRIGGER_SCHEMA_VERSION: u32 = 2;
+pub(crate) const CURRENT_AUTOMATION_SCHEMA_VERSION: u32 = 3;
+const CURRENT_RUN_SCHEMA_VERSION: u32 = 3;
+const CURRENT_TRIGGER_SCHEMA_VERSION: u32 = 3;
 const DEFAULT_AUTOMATION_MODE: &str = "agent";
 const DEFAULT_AUTOMATION_ALLOW_SHELL: bool = false;
 const DEFAULT_AUTOMATION_TRUST_MODE: bool = false;
@@ -102,6 +102,9 @@ pub struct DelayedTriggerRecord {
     pub parent_trigger_id: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub dispatch: Option<AutomationDispatch>,
+    /// Bound by the trusted service, independently of visibility ownership.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub execution_scope: Option<String>,
 }
 
 /// Input for creating a new delayed trigger.
@@ -181,6 +184,9 @@ pub struct AutomationRecord {
     pub next_run_at: Option<DateTime<Utc>>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub last_run_at: Option<DateTime<Utc>>,
+    /// Bound by the trusted service, independently of visibility ownership.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub execution_scope: Option<String>,
 }
 
 impl AutomationRecord {
@@ -240,6 +246,8 @@ pub struct AutomationRunRecord {
 /// `accepted` records task promotion, not provider execution or completion.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct AutomationDispatch {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    execution_scope: Option<String>,
     request: NewTaskRequest,
     task_data_dir: PathBuf,
     #[serde(default)]
@@ -937,6 +945,7 @@ fn days_in_month(year: i32, month: u32) -> u32 {
 
 #[derive(Debug, Clone)]
 pub struct AutomationManager {
+    execution_scope: Option<String>,
     automations_dir: PathBuf,
     runs_dir: PathBuf,
     triggers_dir: PathBuf,
@@ -1028,12 +1037,74 @@ impl AutomationManager {
         fs::create_dir_all(&triggers_dir)
             .with_context(|| format!("Failed to create {}", triggers_dir.display()))?;
         Ok(Self {
+            execution_scope: None,
             automations_dir,
             runs_dir,
             triggers_dir,
         })
     }
 
+    #[cfg(test)]
+    pub(crate) fn open_for_test(root: PathBuf) -> Result<Self> {
+        let mut manager = Self::open(root)?;
+        manager.execution_scope = Some(crate::task_manager::test_execution_scope("test"));
+        Ok(manager)
+    }
+
+    pub(crate) fn bind_task_manager(
+        &mut self,
+        tasks: &crate::task_manager::TaskManager,
+    ) -> Result<()> {
+        if self
+            .execution_scope
+            .as_deref()
+            .is_some_and(|scope| scope != tasks.execution_scope())
+        {
+            bail!("Automation service belongs to another Runtime scope");
+        }
+        self.execution_scope = Some(tasks.execution_scope().to_string());
+        Ok(())
+    }
+
+    pub(crate) fn execution_scope(&self) -> Option<&str> {
+        self.execution_scope.as_deref()
+    }
+
+    fn eligible_scope(&self, scope: Option<&str>) -> bool {
+        scope.is_some() && scope == self.execution_scope()
+    }
+
+    /// Explicit control may bind an unbound definition; saved admissions never
+    /// read this field back from the definition during recovery.
+    fn adopt_for_run(&self, automation: &mut AutomationRecord) -> Result<()> {
+        let scope = self
+            .execution_scope()
+            .context("Automation execution ownership is unverified")?;
+        if let Some(bound) = &automation.execution_scope {
+            if bound != scope {
+                bail!("Automation belongs to another Runtime execution scope");
+            }
+        } else {
+            automation.execution_scope = Some(scope.to_string());
+            automation.schema_version = CURRENT_AUTOMATION_SCHEMA_VERSION;
+            automation.updated_at = Utc::now();
+            if automation.status == AutomationStatus::Active {
+                let schedule = AutomationSchedule::parse_rrule(&automation.rrule)?;
+                automation.next_run_at =
+                    match schedule.next_after_with_anchor(Utc::now(), automation.created_at) {
+                        Ok(next) => Some(next),
+                        Err(_) if matches!(schedule, AutomationSchedule::Once { .. }) => {
+                            automation.status = AutomationStatus::Paused;
+                            None
+                        }
+                        Err(error) => return Err(error),
+                    };
+            }
+            self.save_automation_unlocked(automation)?;
+        }
+        Ok(())
+    }
+
     pub fn default_location() -> Result<Self> {
         Self::open(default_automations_dir())
     }
@@ -1087,6 +1158,7 @@ impl AutomationManager {
 
         let record = AutomationRecord {
             schema_version: CURRENT_AUTOMATION_SCHEMA_VERSION,
+            execution_scope: self.execution_scope.clone(),
             id: Uuid::new_v4().to_string(),
             name: req.name.trim().to_string(),
             prompt: req.prompt.trim().to_string(),
@@ -1188,7 +1260,13 @@ impl AutomationManager {
         req: UpdateAutomationRequest,
     ) -> Result<AutomationRecord> {
         let mut existing = self.get_automation(id)?;
-        let schedule_changed = req.rrule.is_some() || req.status.is_some();
+        let adopting = existing.execution_scope.is_none()
+            && self.execution_scope.is_some()
+            && req.status != Some(AutomationStatus::Paused);
+        if adopting {
+            existing.execution_scope = self.execution_scope.clone();
+        }
+        let schedule_changed = adopting || req.rrule.is_some() || req.status.is_some();
 
         if let Some(name) = req.name {
             if name.trim().is_empty() {
@@ -1249,7 +1327,10 @@ impl AutomationManager {
             }
         }
 
-        if existing.model_provider.is_some() || existing.model_provider_id.is_some() {
+        if existing.execution_scope.is_some()
+            || existing.model_provider.is_some()
+            || existing.model_provider_id.is_some()
+        {
             existing.schema_version = CURRENT_AUTOMATION_SCHEMA_VERSION;
         }
 
@@ -1492,7 +1573,9 @@ impl AutomationManager {
         self.with_transaction(|| {
             let mut due = Vec::new();
             for mut automation in self.list_automations()? {
-                if automation.status != AutomationStatus::Active {
+                if automation.status != AutomationStatus::Active
+                    || !self.eligible_scope(automation.execution_scope.as_deref())
+                {
                     continue;
                 }
                 let schedule = AutomationSchedule::parse_rrule(&automation.rrule)?;
@@ -1537,7 +1620,8 @@ impl AutomationManager {
                 return Ok(None);
             }
             let mut current = self.get_automation(&observed.id)?;
-            if current != *observed
+            if !self.eligible_scope(current.execution_scope.as_deref())
+                || current != *observed
                 || current.status != AutomationStatus::Active
                 || current.next_run_at != Some(run.scheduled_for)
             {
@@ -1694,6 +1778,7 @@ impl AutomationManager {
         }
         let record = DelayedTriggerRecord {
             schema_version: CURRENT_TRIGGER_SCHEMA_VERSION,
+            execution_scope: self.execution_scope.clone(),
             trigger_id: format!("trig_{}", Uuid::new_v4().simple()),
             fire_at: req.fire_at,
             message: req.message.trim().to_string(),
@@ -1889,9 +1974,13 @@ fn bind_run_dispatch(
     task_data_dir: &Path,
     scheduled: bool,
 ) -> Result<()> {
+    if automation.execution_scope.is_none() {
+        bail!("Automation execution ownership is unverified");
+    }
     run.schema_version = CURRENT_RUN_SCHEMA_VERSION;
     run.task_id = Some(crate::task_manager::TaskManager::new_task_id());
     run.dispatch = Some(AutomationDispatch {
+        execution_scope: automation.execution_scope.clone(),
         request: automation_task_request(automation),
         task_data_dir: task_data_dir
             .canonicalize()
@@ -1920,6 +2009,11 @@ async fn dispatch_bound_task(
     tasks: &SharedTaskManager,
 ) -> Result<crate::task_manager::TaskRecord> {
     check_dispatch_store(dispatch, tasks)?;
+    if dispatch.execution_scope.as_deref() != Some(tasks.execution_scope()) {
+        bail!(
+            "Automation admission execution ownership is unverified or belongs to another Runtime"
+        );
+    }
     let task = if dispatch.accepted {
         tasks
             .read_bound_task(task_id)?
@@ -1972,6 +2066,7 @@ pub async fn run_now_shared(
     automation_id: &str,
     task_manager: &SharedTaskManager,
 ) -> Result<AutomationRunRecord> {
+    automations.lock().await.bind_task_manager(task_manager)?;
     let task_manager = Arc::clone(task_manager);
     let task_data_dir = task_manager.data_dir();
     run_now_with(
@@ -2004,7 +2099,8 @@ where
     let (automation, run) = {
         let manager = automations.lock().await;
         manager.with_transaction(|| {
-            let automation = manager.get_automation(automation_id)?;
+            let mut automation = manager.get_automation(automation_id)?;
+            manager.adopt_for_run(&mut automation)?;
             let now = Utc::now();
             let mut run = new_run_record(&automation.id, now, now);
             bind_run_dispatch(&mut run, &automation, task_data_dir, false)?;
@@ -2024,6 +2120,7 @@ async fn scheduler_tick_shared(
     automations: &SharedAutomationManager,
     task_manager: &SharedTaskManager,
 ) -> Result<()> {
+    automations.lock().await.bind_task_manager(task_manager)?;
     let tasks = Arc::clone(task_manager);
     scheduler_tick_with(automations, &tasks.data_dir(), move |mut run| {
         let tasks = Arc::clone(&tasks);
@@ -2058,6 +2155,13 @@ where
             .as_ref()
             .is_some_and(|dispatch| !dispatch.accepted)
     }) {
+        if !automations.lock().await.eligible_scope(
+            run.dispatch
+                .as_ref()
+                .and_then(|dispatch| dispatch.execution_scope.as_deref()),
+        ) {
+            continue;
+        }
         automations.lock().await.recover_schedule_advance(&run)?;
         let run = enqueue(run).await;
         automations
@@ -2068,6 +2172,13 @@ where
     let now = Utc::now();
     let due = automations.lock().await.collect_due_runs(now)?;
     for (observed, proposed) in due {
+        if !automations
+            .lock()
+            .await
+            .eligible_scope(observed.execution_scope.as_deref())
+        {
+            continue;
+        }
         let run =
             automations
                 .lock()
@@ -2086,6 +2197,7 @@ async fn fire_due_triggers_shared(
     automations: &SharedAutomationManager,
     task_manager: &SharedTaskManager,
 ) -> Result<()> {
+    automations.lock().await.bind_task_manager(task_manager)?;
     let tasks = Arc::clone(task_manager);
     fire_due_triggers_with(automations, &tasks.data_dir(), move |trigger| {
         let tasks = Arc::clone(&tasks);
@@ -2138,6 +2250,17 @@ where
     let now = Utc::now();
     let candidates = automations.lock().await.collect_due_triggers(now)?;
     for candidate in candidates {
+        let scope = if candidate.status == DelayedTriggerStatus::Dispatching {
+            candidate
+                .dispatch
+                .as_ref()
+                .and_then(|d| d.execution_scope.as_deref())
+        } else {
+            candidate.execution_scope.as_deref()
+        };
+        if !automations.lock().await.eligible_scope(scope) {
+            continue;
+        }
         if !(candidate.status == DelayedTriggerStatus::Dispatching
             || (candidate.status == DelayedTriggerStatus::Pending && candidate.fire_at <= now))
         {
@@ -2148,12 +2271,21 @@ where
             manager.with_transaction(|| {
                 let mut current = manager.get_trigger(&candidate.trigger_id)?;
                 if current.status == DelayedTriggerStatus::Dispatching {
+                    if !manager.eligible_scope(
+                        current
+                            .dispatch
+                            .as_ref()
+                            .and_then(|d| d.execution_scope.as_deref()),
+                    ) {
+                        return Ok(None);
+                    }
                     if current.dispatch.is_none() || current.task_id.is_none() {
                         bail!("Claimed delayed trigger has no durable task binding");
                     }
                     return Ok(Some(current));
                 }
-                if current.status != DelayedTriggerStatus::Pending
+                if !manager.eligible_scope(current.execution_scope.as_deref())
+                    || current.status != DelayedTriggerStatus::Pending
                     || current.fire_at > now
                     || current.owner_session_id.is_none()
                 {
@@ -2163,6 +2295,7 @@ where
                 current.status = DelayedTriggerStatus::Dispatching;
                 current.task_id = Some(crate::task_manager::TaskManager::new_task_id());
                 current.dispatch = Some(AutomationDispatch {
+                    execution_scope: current.execution_scope.clone(),
                     request: NewTaskRequest {
                         prompt: current.message.clone(),
                         model: None,
@@ -2305,6 +2438,7 @@ async fn reconcile_run_statuses_shared(
             continue;
         }
         let dispatch = run.dispatch.get_or_insert_with(|| AutomationDispatch {
+            execution_scope: task.execution_scope.clone(),
             request: NewTaskRequest::from_task(&task),
             task_data_dir: task_manager.data_dir(),
             accepted: true,
@@ -2580,7 +2714,7 @@ mod tests {
             let root = tempfile::tempdir()?;
             let receipts = root.path().join("executions");
             let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
-            let manager = AutomationManager::open(root.path().join("automations"))?;
+            let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
             let automation = fixture_due_automation(&manager, "recover", 1);
             let shared = Arc::new(Mutex::new(manager));
             let (at_cut, reached_cut) = tokio::sync::oneshot::channel();
@@ -2642,7 +2776,7 @@ mod tests {
             // Simulate a dropped request/process before the final run receipt.
             tick.abort();
             assert!(tick.await.unwrap_err().is_cancelled());
-            let reopened = AutomationManager::open(root.path().join("automations"))?;
+            let reopened = AutomationManager::open_for_test(root.path().join("automations"))?;
             let future = Utc::now() + Duration::hours(4);
             let edited = reopened.update_automation(
                 &automation.id,
@@ -2699,7 +2833,7 @@ mod tests {
         let root = tempfile::tempdir()?;
         let receipts = root.path().join("executions");
         let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
-        let manager = AutomationManager::open(root.path().join("automations"))?;
+        let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
         let first = fixture_due_automation(&manager, "first", 1);
         let paused = fixture_due_automation(&manager, "paused", 2);
         let deleted = fixture_due_automation(&manager, "deleted", 3);
@@ -2731,7 +2865,7 @@ mod tests {
             admitted.automation_id, first.id,
             "ordered first claim really entered"
         );
-        let other = AutomationManager::open(root.path().join("automations"))?;
+        let other = AutomationManager::open_for_test(root.path().join("automations"))?;
         other.pause_automation(&paused.id)?;
         other.delete_automation(&deleted.id)?;
         // Pause after durable admission affects future runs; this run remains owned.
@@ -2767,7 +2901,7 @@ mod tests {
             let root = tempfile::tempdir()?;
             let receipts = root.path().join("executions");
             let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
-            let manager = AutomationManager::open(root.path().join("automations"))?;
+            let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
             let automation = fixture_due_automation(&manager, "history", 1);
             let shared = Arc::new(Mutex::new(manager));
             let run = run_now_shared(&shared, &automation.id, &tasks).await?;
@@ -2833,7 +2967,7 @@ mod tests {
         let other_receipts = root.path().join("foreign-executions");
         let other_tasks =
             fixture_tasks(&root.path().join("foreign-tasks"), &other_receipts).await?;
-        let manager = AutomationManager::open(root.path().join("automations"))?;
+        let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
         let automation = fixture_due_automation(&manager, "bound-store", 1);
         let proposed = manager.collect_due_runs(Utc::now())?.remove(0);
         let mut run = manager
@@ -2888,7 +3022,7 @@ mod tests {
         let root = tempfile::tempdir()?;
         let receipts = root.path().join("executions");
         let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
-        let manager = AutomationManager::open(root.path().join("automations"))?;
+        let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
         let mut triggers = Vec::new();
         for index in 0..2 {
             let mut trigger = manager.create_trigger(CreateDelayedTriggerRequest {
@@ -2928,7 +3062,7 @@ mod tests {
         let admitted = tokio::time::timeout(std::time::Duration::from_secs(5), reached).await??;
         assert_eq!(admitted.trigger_id, triggers[0].trigger_id);
         assert_eq!(admitted.status, DelayedTriggerStatus::Dispatching);
-        let other = AutomationManager::open(root.path().join("automations"))?;
+        let other = AutomationManager::open_for_test(root.path().join("automations"))?;
         other.cancel_trigger_for_owner(&triggers[1].trigger_id, "fixture-owner")?;
         assert!(
             other
@@ -3035,10 +3169,20 @@ mod tests {
         if !matches!(role.as_str(), "first" | "second") {
             bail!("invalid fixture role");
         }
-        let tasks = fixture_tasks(&root.join("tasks"), &root.join("executions")).await?;
-        let shared = Arc::new(Mutex::new(AutomationManager::open(
-            root.join("automations"),
-        )?));
+        let scope = if role == "first" {
+            "test"
+        } else {
+            "foreign-scheduler"
+        };
+        let tasks = TaskManager::start_with_executor_in_scope(
+            automation_task_config(root.join("tasks")),
+            Arc::new(AutomationRecordingExecutor(root.join("executions"))),
+            scope,
+        )
+        .await?;
+        let mut service = AutomationManager::open(root.join("automations"))?;
+        service.bind_task_manager(&tasks)?;
+        let shared = Arc::new(Mutex::new(service));
         crate::utils::write_atomic(&root.join(format!("{role}-ready")), b"ready")?;
         wait_fixture_path(&root.join(format!("{role}-go"))).await?;
         let observations = Arc::new(std::sync::Mutex::new(Vec::new()));
@@ -3093,15 +3237,14 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn two_scheduler_processes_share_one_claim_and_execute_one_bound_task() -> Result<()> {
+    async fn two_scheduler_processes_preserve_the_bound_scope_and_execute_one_task() -> Result<()> {
         let root = tempfile::tempdir()?;
-        let manager = AutomationManager::open(root.path().join("automations"))?;
+        let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
         let automation = fixture_due_automation(&manager, "cross-process", 1);
         let mut first = spawn_scheduler_fixture(root.path(), "first")?;
         let mut second = spawn_scheduler_fixture(root.path(), "second")?;
-        // Both independent TaskManagers open the empty task store before the
-        // first task is accepted; this isolates scheduler ownership from the
-        // separate pre-existing TaskManager-startup ownership limitation.
+        // Separate Runtime scopes share storage. The foreign scheduler cannot
+        // adopt the first scope's immutable dispatch even before acceptance.
         wait_fixture_path(&root.path().join("first-ready")).await?;
         wait_fixture_path(&root.path().join("second-ready")).await?;
         crate::utils::write_atomic(&root.path().join("first-go"), b"go")?;
@@ -3281,6 +3424,7 @@ mod tests {
         let now = Utc::now();
         AutomationRecord {
             schema_version: CURRENT_AUTOMATION_SCHEMA_VERSION,
+            execution_scope: Some(crate::task_manager::test_execution_scope("test")),
             id: Uuid::new_v4().to_string(),
             name: "Test automation".to_string(),
             prompt: "Run the automation".to_string(),
@@ -3537,11 +3681,13 @@ mod tests {
             .expect("reset-anchor schedule");
         assert_ne!(expected, reset_anchor, "fixture must detect anchor resets");
 
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
         manager.save_automation(&automation).expect("save");
         drop(manager);
 
-        let restarted = AutomationManager::open(tempdir.path().to_path_buf()).expect("reopen");
+        let restarted =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("reopen");
         assert!(
             restarted
                 .collect_due_runs(now)
@@ -3558,7 +3704,8 @@ mod tests {
     #[test]
     fn resume_uses_persisted_creation_anchor() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
         let before = Utc::now();
         let created_at = before - Duration::hours(51);
         let automation = anchored_automation(created_at, AutomationStatus::Paused);
@@ -3679,7 +3826,8 @@ mod tests {
     #[test]
     fn automation_model_round_trips_through_create_and_update() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
 
         let created = manager
             .create_automation(CreateAutomationRequest {
@@ -3723,7 +3871,8 @@ mod tests {
     #[test]
     fn deletes_definition_and_settled_runs_but_retains_unfinished_receipts() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
 
         let created = manager
             .create_automation(CreateAutomationRequest {
@@ -3788,7 +3937,8 @@ mod tests {
     #[test]
     fn automation_storage_rejects_traversal_ids() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().join("root")).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().join("root")).expect("manager");
         let escaped_file = tempdir.path().join("escape.json");
         let escaped_runs = tempdir.path().join("escape-runs");
 
@@ -3921,7 +4071,7 @@ api_key = "fixture-second"
 model = "private-model"
 "#,
         )?;
-        let automations = AutomationManager::open(root.path().join("schedules"))?;
+        let automations = AutomationManager::open_for_test(root.path().join("schedules"))?;
         let mut record = automation_record_with_settings(None, None, None, None);
         record.schema_version = 1;
         record.model = Some("private-model".to_string());
@@ -3930,7 +4080,7 @@ model = "private-model"
         automations.save_automation(&record)?;
         let record = automations.get_automation(&record.id)?;
         assert_eq!(
-            record.schema_version, 2,
+            record.schema_version, CURRENT_AUTOMATION_SCHEMA_VERSION,
             "a pin cannot be ignored by an older reader"
         );
 
@@ -3956,7 +4106,7 @@ model = "private-model"
             .await?;
         let task: crate::task_manager::TaskRecord =
             serde_json::from_slice(&serde_json::to_vec(&task)?)?;
-        assert_eq!(task.schema_version, 3);
+        assert_eq!(task.schema_version, 4);
         let thread = runtime
             .create_thread(ExecutionTask::from(&task).thread_request())
             .await?;
@@ -3999,7 +4149,7 @@ model = "private-model"
             std::sync::Arc::new(AutomationNoopExecutor),
         )
         .await?;
-        let manager = AutomationManager::open(tempdir.path().join("automations"))?;
+        let manager = AutomationManager::open_for_test(tempdir.path().join("automations"))?;
 
         let mut owned = manager.create_trigger(CreateDelayedTriggerRequest {
             fire_at: Utc::now() + Duration::hours(1),
@@ -4084,7 +4234,7 @@ model = "private-model"
     fn interrupted_watcher_migration_deduplicates_before_visibility() -> Result<()> {
         for suppressed in [false, true] {
             let root = tempfile::tempdir()?;
-            let manager = AutomationManager::open(root.path().join("automations"))?;
+            let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
             let automation = automation_record_with_settings(None, None, None, None);
             let mut run = queued_run_for(&automation);
             write_legacy_run_file(&manager, &run);
@@ -4123,7 +4273,8 @@ model = "private-model"
     #[test]
     fn save_run_uses_sortable_names_and_migrates_legacy_files() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
         let automation = automation_record_with_settings(None, None, None, None);
         let run = queued_run_for(&automation);
 
@@ -4151,7 +4302,8 @@ model = "private-model"
     #[test]
     fn finish_scheduled_run_persists_run_when_automation_deleted_mid_enqueue() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
         let automation = automation_record_with_settings(None, None, None, None);
         manager.save_automation(&automation).expect("save");
         let run = queued_run_for(&automation);
@@ -4180,7 +4332,8 @@ model = "private-model"
     #[test]
     fn once_schedule_fires_once_and_auto_completes() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
         let due_at = Utc::now() - Duration::minutes(1);
         let automation = AutomationRecord {
             rrule: due_at
@@ -4225,7 +4378,8 @@ model = "private-model"
     #[test]
     fn get_runs_by_ids_finds_live_runs_past_the_newest_window() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
         let automation = automation_record_with_settings(None, None, None, None);
         let base = Utc::now();
 
@@ -4264,7 +4418,8 @@ model = "private-model"
     #[test]
     fn get_runs_by_ids_ignores_non_json_noise() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
         let automation = automation_record_with_settings(None, None, None, None);
         let run = queued_run_for(&automation);
         manager.save_run(&run).expect("save json run");
@@ -4295,7 +4450,8 @@ model = "private-model"
     #[test]
     fn list_runs_merges_legacy_and_sortable_files_newest_first() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
         let automation = automation_record_with_settings(None, None, None, None);
         let base = Utc::now();
 
@@ -4334,7 +4490,8 @@ model = "private-model"
     #[test]
     fn list_runs_with_limit_skips_older_sortable_files_entirely() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
         let automation = automation_record_with_settings(None, None, None, None);
         let base = Utc::now();
 
@@ -4363,7 +4520,8 @@ model = "private-model"
     #[tokio::test]
     async fn list_automations_completes_during_slow_enqueue() {
         let tempdir = tempfile::tempdir().expect("tempdir");
-        let manager = AutomationManager::open(tempdir.path().to_path_buf()).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().to_path_buf()).expect("manager");
         let created = manager
             .create_automation(CreateAutomationRequest {
                 name: "Slow enqueue".to_string(),
@@ -4445,7 +4603,8 @@ model = "private-model"
         let mut automation = automation_record_with_settings(None, None, None, None);
         automation.delivery_mode = Some(AutomationDeliveryMode::Watcher);
         automation.next_run_at = Some(Utc::now() - Duration::seconds(1));
-        let manager = AutomationManager::open(tempdir.path().join("automations")).expect("manager");
+        let manager =
+            AutomationManager::open_for_test(tempdir.path().join("automations")).expect("manager");
         manager
             .save_automation(&automation)
             .expect("save automation");
@@ -4494,7 +4653,7 @@ model = "private-model"
         manager.save_automation(&updated)?;
         drop(manager);
         scheduler_tick_shared(&shared, &task_manager).await?;
-        assert_eq!(task_manager.list_tasks(None).await.len(), 1);
+        assert_eq!(task_manager.list_tasks(None).await?.len(), 1);
         assert!(
             shared
                 .lock()
@@ -4541,4 +4700,5 @@ model = "private-model"
             std::env::remove_var("CODEWHALE_HOME");
         }
     }
+    mod ownership;
 }

crates/tui/src/automation_manager/tests/ownership.rs

@@ -0,0 +1,121 @@
+use super::*;
+
+#[test]
+fn foreign_and_unbound_definitions_are_not_repaired_or_paused_by_a_tick() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let owner = AutomationManager::open_for_test(root.path().to_path_buf())?;
+    let mut foreign = automation_record_with_settings(None, None, None, None);
+    foreign.id = "foreign".into();
+    foreign.next_run_at = None;
+    owner.save_automation(&foreign)?;
+    let mut legacy = foreign.clone();
+    legacy.id = "legacy".into();
+    legacy.execution_scope = None;
+    legacy.rrule = "unparseable legacy schedule".into();
+    owner.save_automation(&legacy)?;
+    let before = [
+        fs::read(owner.automation_path(&foreign.id)?)?,
+        fs::read(owner.automation_path(&legacy.id)?)?,
+    ];
+    let mut other = AutomationManager::open(root.path().to_path_buf())?;
+    other.execution_scope = Some(crate::task_manager::test_execution_scope("other"));
+    assert!(other.collect_due_runs(Utc::now())?.is_empty());
+    assert_eq!(fs::read(owner.automation_path(&foreign.id)?)?, before[0]);
+    assert_eq!(fs::read(owner.automation_path(&legacy.id)?)?, before[1]);
+    Ok(())
+}
+
+#[tokio::test]
+async fn explicit_run_now_adopts_expired_legacy_once_but_does_not_rebind_old_admissions()
+-> Result<()> {
+    let root = tempfile::tempdir()?;
+    let receipts = root.path().join("executions");
+    let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
+    let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
+    let mut legacy = automation_record_with_settings(None, None, None, None);
+    legacy.execution_scope = None;
+    legacy.rrule = format!(
+        "FREQ=ONCE;AT={}",
+        (Utc::now() - Duration::hours(1)).to_rfc3339()
+    );
+    legacy.next_run_at = None;
+    manager.save_automation(&legacy)?;
+    let mut old = queued_run_for(&legacy);
+    let mut formerly_bound = legacy.clone();
+    formerly_bound.execution_scope = Some(crate::task_manager::test_execution_scope("test"));
+    bind_run_dispatch(&mut old, &formerly_bound, &tasks.data_dir(), false)?;
+    old.dispatch.as_mut().unwrap().execution_scope = None;
+    manager.save_run(&old)?;
+    let old_path = manager.run_path(&old)?;
+    let before = fs::read(&old_path)?;
+    let shared = Arc::new(Mutex::new(manager));
+    let explicit = run_now_shared(&shared, &legacy.id, &tasks).await?;
+    let id = explicit.task_id.context("explicit task")?;
+    let task = crate::task_manager::wait_for_terminal_state(
+        &tasks,
+        &id,
+        std::time::Duration::from_secs(10),
+    )
+    .await?;
+    assert_eq!(task.status, TaskStatus::Completed);
+    assert!(
+        task.owner_session_id.is_none(),
+        "None visibility is valid for a new scoped automation"
+    );
+    assert_eq!(
+        task.execution_scope.as_deref(),
+        Some(tasks.execution_scope())
+    );
+    assert_eq!(
+        tasks.get_task_for_active_runtime(&task.id).await?.id,
+        task.id
+    );
+    scheduler_tick_shared(&shared, &tasks).await?;
+    assert_eq!(fixture_executions(&receipts), vec![id]);
+    assert_eq!(
+        fs::read(old_path)?,
+        before,
+        "definition adoption cannot adopt an old occurrence"
+    );
+    let adopted = shared.lock().await.get_automation(&legacy.id)?;
+    assert_eq!(
+        adopted.execution_scope.as_deref(),
+        Some(tasks.execution_scope())
+    );
+    assert_eq!(adopted.status, AutomationStatus::Paused);
+    tasks.shutdown_and_wait().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn foreign_scoped_automation_and_unbound_trigger_never_dispatch_through_current_service()
+-> Result<()> {
+    let root = tempfile::tempdir()?;
+    let receipts = root.path().join("executions");
+    let tasks = fixture_tasks(&root.path().join("tasks"), &receipts).await?;
+    let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
+    let mut foreign = fixture_due_automation(&manager, "foreign", 1);
+    foreign.execution_scope = Some(crate::task_manager::test_execution_scope("foreign"));
+    manager.save_automation(&foreign)?;
+    let mut trigger = manager.create_trigger(CreateDelayedTriggerRequest {
+        fire_at: Utc::now() + Duration::hours(1),
+        message: "legacy continuation".into(),
+        workspace: Some(root.path().to_path_buf()),
+        owner_session_id: None,
+        parent_trigger_id: None,
+    })?;
+    trigger.execution_scope = None;
+    trigger.fire_at = Utc::now() - Duration::seconds(1);
+    manager.save_trigger(&trigger)?;
+    let path = manager.trigger_path(&trigger.trigger_id)?;
+    let before = fs::read(&path)?;
+    let shared = Arc::new(Mutex::new(manager));
+    scheduler_tick_shared(&shared, &tasks).await?;
+    fire_due_triggers_shared(&shared, &tasks).await?;
+    assert!(run_now_shared(&shared, &foreign.id, &tasks).await.is_err());
+    assert!(fixture_executions(&receipts).is_empty());
+    assert!(tasks.list_tasks(None).await?.is_empty());
+    assert_eq!(fs::read(path)?, before);
+    tasks.shutdown_and_wait().await?;
+    Ok(())
+}

crates/tui/src/core/engine.rs

@@ -6366,33 +6366,46 @@ impl Engine {
         let checked_at = chrono::Utc::now().timestamp_millis();
 
         let mut seen_tasks = HashSet::new();
+        let mut task_inventory_available = false;
         if let Some(task_manager) = self.config.runtime_services.task_manager.as_ref() {
-            for task in task_manager
+            match task_manager
                 .list_tasks_for_owner(None, None, session_id)
                 .await
             {
-                let external = format!("task:{}", task.id);
-                if !candidates.contains(&external) {
-                    continue;
+                Ok(tasks) => {
+                    task_inventory_available = true;
+                    for task in tasks {
+                        let external = format!("task:{}", task.id);
+                        if !candidates.contains(&external) {
+                            continue;
+                        }
+                        seen_tasks.insert(external.clone());
+                        if !task.execution_binding_known {
+                            continue;
+                        }
+                        if let Err(err) = work.reconcile_operation(
+                            session_id,
+                            crate::work_graph::task_owner_snapshot(
+                                &task.id,
+                                task.status,
+                                task.lifecycle_seq,
+                                task.created_at,
+                                task.started_at,
+                                task.ended_at,
+                            ),
+                        ) {
+                            tracing::warn!(task_id = %task.id, error = %err, "failed to reconcile restored task owner");
+                        }
+                    }
                 }
-                seen_tasks.insert(external.clone());
-                if let Err(err) = work.reconcile_operation(
-                    session_id,
-                    crate::work_graph::task_owner_snapshot(
-                        &task.id,
-                        task.status,
-                        task.lifecycle_seq,
-                        task.created_at,
-                        task.started_at,
-                        task.ended_at,
-                    ),
-                ) {
-                    tracing::warn!(task_id = %task.id, error = %err, "failed to reconcile restored task owner");
+                Err(error) => {
+                    tracing::warn!(%error, "Task owner inventory unavailable; retaining Work bindings")
                 }
             }
         }
         for external in candidates
             .iter()
+            .filter(|_| task_inventory_available)
             .filter(|external| external.starts_with("task:"))
             .filter(|external| !seen_tasks.contains(*external))
         {
@@ -7594,11 +7607,11 @@ pub(crate) fn spawn_engine_with_authoritative_route_config(
     config: EngineConfig,
     api_config: &Config,
     authoritative_route_config: Arc<parking_lot::RwLock<Config>>,
-) -> EngineHandle {
+) -> (EngineHandle, tokio::task::JoinHandle<()>) {
     let (mut engine, handle) = Engine::new(config, api_config);
     engine.authoritative_route_config = Some(authoritative_route_config);
 
-    spawn_supervised(
+    let worker = spawn_supervised(
         "engine-event-loop",
         std::panic::Location::caller(),
         async move {
@@ -7606,7 +7619,7 @@ pub(crate) fn spawn_engine_with_authoritative_route_config(
         },
     );
 
-    handle
+    (handle, worker)
 }
 
 #[cfg(test)]

crates/tui/src/core/engine/tests.rs

@@ -23075,3 +23075,79 @@ fn engine_adopts_host_owned_session_id_from_config() {
         engine.session_id()
     );
 }
+
+#[tokio::test]
+async fn restored_task_binding_is_not_missing_when_its_inventory_is_unavailable()
+-> anyhow::Result<()> {
+    use crate::task_manager::{TaskExecutionResult, TaskManager, TaskManagerConfig};
+    struct Unused;
+    #[async_trait::async_trait]
+    impl crate::task_manager::TaskExecutor for Unused {
+        async fn execute(
+            &self,
+            _: crate::task_manager::ExecutionTask,
+            _: tokio::sync::mpsc::Sender<crate::task_manager::TaskExecutionEvent>,
+            _: tokio_util::sync::CancellationToken,
+        ) -> TaskExecutionResult {
+            panic!("this fixture must never execute a task")
+        }
+    }
+    let (mut engine, _handle, _todos, work, root) = todo_engine();
+    let tasks = TaskManager::start_with_executor(
+        TaskManagerConfig {
+            data_dir: root.path().join("tasks"),
+            worker_count: 1,
+            default_workspace: root.path().into(),
+            default_model: "fixture".into(),
+            default_mode: "plan".into(),
+            allow_shell: false,
+            trust_mode: false,
+            execution_limits: crate::task_manager::TaskExecutionLimits::default(),
+        },
+        Arc::new(Unused),
+    )
+    .await?;
+    engine.config.runtime_services.task_manager = Some(tasks.clone());
+    let session = engine.session.id.clone();
+    let id = work
+        .register_operation(
+            &session,
+            crate::work_graph::OperationIntent::new(
+                "task:task_0123456789abcdef",
+                "restored task",
+                true,
+                "tasks",
+                "fixture",
+            ),
+        )
+        .map_err(anyhow::Error::msg)?;
+    let before = work
+        .capture(Some(&session))
+        .map_err(anyhow::Error::msg)?
+        .unwrap();
+    let queue = tasks.data_dir().join("queue.json");
+    let saved = std::fs::read(&queue)?;
+    std::fs::write(&queue, b"{corrupt")?;
+    engine.reconcile_restored_work_bindings().await;
+    let unavailable = work
+        .capture(Some(&session))
+        .map_err(anyhow::Error::msg)?
+        .unwrap();
+    assert_eq!(
+        serde_json::to_value(unavailable.graph.node(&id))?,
+        serde_json::to_value(before.graph.node(&id))?
+    );
+    std::fs::write(&queue, saved)?;
+    engine.reconcile_restored_work_bindings().await;
+    let available = work
+        .capture(Some(&session))
+        .map_err(anyhow::Error::msg)?
+        .unwrap();
+    assert_ne!(
+        serde_json::to_value(available.graph.node(&id))?,
+        serde_json::to_value(before.graph.node(&id))?,
+        "healthy absence must still reconcile OwnerMissing"
+    );
+    tasks.shutdown_and_wait().await?;
+    Ok(())
+}

crates/tui/src/localization.rs

@@ -1938,6 +1938,8 @@ pub enum MessageId {
     WorkflowReceiptEstimated,
     // Sidebar work strip.
     SidebarTasksLabel,
+    TaskOwnershipUnverified,
+    TaskInventoryUnavailable,
     SidebarTodoLabel,
     SidebarStopControl,
     SidebarDestructiveArmed,
@@ -4114,6 +4116,8 @@ pub const ALL_MESSAGE_IDS: &[MessageId] = &[
     MessageId::WorkflowReceiptProviderReported,
     MessageId::WorkflowReceiptEstimated,
     MessageId::SidebarTasksLabel,
+    MessageId::TaskOwnershipUnverified,
+    MessageId::TaskInventoryUnavailable,
     MessageId::SidebarTodoLabel,
     MessageId::SidebarStopControl,
     MessageId::SidebarDestructiveArmed,

crates/tui/src/operate.rs

@@ -10,7 +10,7 @@ use std::fs;
 use std::path::{Component, Path, PathBuf};
 use std::process::{Command, Stdio};
 
-use anyhow::{Context, Result};
+use anyhow::{Context, Result, bail};
 use chrono::Utc;
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
@@ -719,6 +719,13 @@ pub(crate) fn keepalive_readiness(
 ) -> Result<(String, bool)> {
     let mut readiness = (String::new(), false);
     manager.edit_automation(OPERATE_KEEPALIVE_ID, |current| {
+        if current
+            .as_ref()
+            .and_then(|record| record.execution_scope.as_deref())
+            .is_some_and(|scope| Some(scope) != manager.execution_scope())
+        {
+            bail!("Operate keepalive belongs to another Runtime execution scope");
+        }
         let (_, model, credentials) = keepalive_route(config, current.as_ref(), selection)?;
         readiness = (model, credentials);
         Ok(None)
@@ -1210,10 +1217,21 @@ pub(crate) fn upsert_keepalive(
     );
     let mut readiness = (String::new(), false);
     manager.edit_automation(OPERATE_KEEPALIVE_ID, |current| {
+        let scope = manager
+            .execution_scope()
+            .context("Operate execution ownership is unverified")?;
+        if current
+            .as_ref()
+            .and_then(|record| record.execution_scope.as_deref())
+            .is_some_and(|bound| bound != scope)
+        {
+            bail!("Operate keepalive belongs to another Runtime execution scope");
+        }
         let (identity, model, ready) = keepalive_route(config, current.as_ref(), selection)?;
         readiness = (model.clone(), ready);
         let mut record = current.unwrap_or_else(|| AutomationRecord {
             schema_version: crate::automation_manager::CURRENT_AUTOMATION_SCHEMA_VERSION,
+            execution_scope: Some(scope.to_string()),
             id: OPERATE_KEEPALIVE_ID.to_string(),
             name: "Operate keep-alive".to_string(),
             prompt: prompt.clone(),
@@ -1233,6 +1251,10 @@ pub(crate) fn upsert_keepalive(
             next_run_at: None,
             last_run_at: None,
         });
+        record.schema_version = crate::automation_manager::CURRENT_AUTOMATION_SCHEMA_VERSION;
+        if record.execution_scope.is_none() {
+            record.execution_scope = Some(scope.to_string());
+        }
         record.name = "Operate keep-alive".to_string();
         record.prompt = prompt;
         record.rrule = OPERATE_KEEPALIVE_RRULE.to_string();
@@ -1329,7 +1351,7 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
         let _cli = crate::test_support::EnvVarGuard::remove("CODEWHALE_CLI_API_KEY");
         let _missing = crate::test_support::EnvVarGuard::remove("CW_OPERATE_MISSING_TEST_KEY");
         let root = TempDir::new()?;
-        let manager = AutomationManager::open(root.path().join("automations"))?;
+        let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
         let mut config = route_fixture_config();
         assert!(upsert_keepalive(&manager, root.path(), true, &config, None)?.1);
         let first = manager.get_automation(OPERATE_KEEPALIVE_ID)?;
@@ -1388,7 +1410,7 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
         let _env = crate::test_support::lock_test_env();
         let _cli = crate::test_support::EnvVarGuard::remove("CODEWHALE_CLI_API_KEY");
         let root = TempDir::new()?;
-        let manager = AutomationManager::open(root.path().join("automations"))?;
+        let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
         let config = crate::config::Config {
             provider: Some("custom".into()),
             base_url: Some("https://legacy.example.test/v1".into()),
@@ -1435,7 +1457,7 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
     #[tokio::test]
     async fn keepalive_pin_reaches_task_and_survives_parent_change() -> Result<()> {
         let root = TempDir::new()?;
-        let manager = AutomationManager::open(root.path().join("automations"))?;
+        let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
         let mut config = route_fixture_config();
         upsert_keepalive(&manager, root.path(), false, &config, None)?;
         pause_keepalive(&manager)?;
@@ -1476,7 +1498,7 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
         assert_eq!(observed[0].model_provider_id, task.model_provider_id);
         assert_eq!(observed[0].auto_approve, Some(false));
         drop(observed);
-        let reopened = AutomationManager::open(root.path().join("automations"))?;
+        let reopened = AutomationManager::open_for_test(root.path().join("automations"))?;
         assert!(keepalive_readiness(&reopened, &config, None)?.1);
         assert_eq!(
             reopened.list_runs(OPERATE_KEEPALIVE_ID, None)?[0]
@@ -1831,7 +1853,7 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
     #[test]
     fn keepalive_reuse_refreshes_cwds_and_kicks_first_lead_run() {
         let dir = TempDir::new().expect("temp");
-        let manager = AutomationManager::open(dir.path().to_path_buf()).expect("manager");
+        let manager = AutomationManager::open_for_test(dir.path().to_path_buf()).expect("manager");
         let workspace_a = dir.path().join("workspace-a");
         let workspace_b = dir.path().join("workspace-b");
         fs::create_dir_all(&workspace_a).expect("dir a");
@@ -1878,7 +1900,7 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
     #[test]
     fn cancel_pauses_keepalive_so_no_cost_accrues() {
         let dir = TempDir::new().expect("temp");
-        let manager = AutomationManager::open(dir.path().to_path_buf()).expect("manager");
+        let manager = AutomationManager::open_for_test(dir.path().to_path_buf()).expect("manager");
         upsert_keepalive(&manager, dir.path(), false, &route_fixture_config(), None)
             .expect("upsert");
         pause_keepalive(&manager).expect("pause");
@@ -1890,7 +1912,8 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
 
         // Pausing is idempotent and a missing keepalive is not an error.
         pause_keepalive(&manager).expect("pause again");
-        let empty = AutomationManager::open(dir.path().join("empty")).expect("empty manager");
+        let empty =
+            AutomationManager::open_for_test(dir.path().join("empty")).expect("empty manager");
         pause_keepalive(&empty).expect("missing keepalive is a no-op");
         assert!(!kick_keepalive(&empty).expect("kick missing"));
 
@@ -2070,7 +2093,7 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
     #[test]
     fn keepalive_automation_and_defaults() {
         let dir = TempDir::new().expect("temp");
-        let manager = AutomationManager::open(dir.path().to_path_buf()).expect("manager");
+        let manager = AutomationManager::open_for_test(dir.path().to_path_buf()).expect("manager");
         upsert_keepalive(&manager, dir.path(), false, &route_fixture_config(), None)
             .expect("upsert");
         let record = manager

crates/tui/src/runtime_api.rs

@@ -863,7 +863,10 @@ pub async fn run_http_server(
     let task_manager =
         TaskManager::start_with_runtime_manager(task_cfg, config.clone(), runtime_threads.clone())
             .await?;
-    let automations = Arc::new(Mutex::new(AutomationManager::default_location()?));
+    let _task_shutdown = task_manager.shutdown_guard();
+    let mut automation_service = AutomationManager::default_location()?;
+    automation_service.bind_task_manager(&task_manager)?;
+    let automations = Arc::new(Mutex::new(automation_service));
     runtime_threads.attach_automation_manager(automations.clone());
     let scheduler_cancel = CancellationToken::new();
     let scheduler_handle = spawn_scheduler(
@@ -906,7 +909,7 @@ pub async fn run_http_server(
         config: Arc::new(parking_lot::RwLock::new(config.clone())),
         workspace,
         plugin_discovery,
-        task_manager,
+        task_manager: task_manager.clone(),
         runtime_threads,
         cors_origins: options.cors_origins.clone(),
         sessions_dir,
@@ -992,6 +995,7 @@ pub async fn run_http_server(
     .map_err(|e| anyhow!("Runtime API server error: {e}"));
     scheduler_cancel.cancel();
     scheduler_handle.abort();
+    task_manager.shutdown_and_wait().await?;
     serve_result
 }
 
@@ -5061,8 +5065,13 @@ async fn list_tasks(
                 .await
         }
         None => state.task_manager.list_tasks(query.limit).await,
-    };
-    let counts = state.task_manager.counts().await;
+    }
+    .map_err(|error| ApiError::internal(format!("Task inventory unavailable: {error}")))?;
+    let counts = state
+        .task_manager
+        .counts()
+        .await
+        .map_err(|error| ApiError::internal(format!("Task inventory unavailable: {error}")))?;
     Ok(Json(TasksResponse { tasks, counts }))
 }
 

crates/tui/src/runtime_api/tests.rs

@@ -1171,7 +1171,7 @@ async fn build_test_server(
         RuntimeThreadManagerConfig::from_task_data_dir(root.join("runtime")),
     )?);
     runtime_threads.attach_task_manager(manager.clone());
-    let automations = Arc::new(Mutex::new(AutomationManager::open(
+    let automations = Arc::new(Mutex::new(AutomationManager::open_for_test(
         root.join("automations"),
     )?));
     runtime_threads.attach_automation_manager(automations.clone());
@@ -8473,7 +8473,7 @@ api_key_env = "CW_OPERATE_MISSING_TEST_KEY"
         .await?;
     assert_eq!(started["operation"]["credentialsPresent"], true);
     assert_eq!(started["operation"]["leadOperator"]["model"], "fleet-model");
-    let manager = AutomationManager::open(root.path().join("automations"))?;
+    let manager = AutomationManager::open_for_test(root.path().join("automations"))?;
     let mut record = manager.get_automation(crate::operate::OPERATE_KEEPALIVE_ID)?;
     assert_eq!(record.model.as_deref(), Some("fleet-model"));
     assert_eq!(record.model_provider.as_deref(), Some("custom"));

crates/tui/src/runtime_threads.rs

@@ -3823,6 +3823,30 @@ struct ActiveThreads {
     lru: VecDeque<String>,
 }
 
+/// Shared ownership of an existing task's join. A canceled drain drops only
+/// its await/lock guard; the unfinished join stays available to the next drain.
+type RuntimeCompletion = Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>;
+
+fn retained_completion(worker: tokio::task::JoinHandle<()>) -> RuntimeCompletion {
+    Arc::new(Mutex::new(Some(worker)))
+}
+
+fn completion_finished(completion: &RuntimeCompletion) -> bool {
+    completion
+        .try_lock()
+        .is_ok_and(|worker| worker.as_ref().is_none_or(|worker| worker.is_finished()))
+}
+
+async fn await_retained_completion(completion: &RuntimeCompletion) -> Result<()> {
+    let mut worker = completion.lock().await;
+    let Some(join) = worker.as_mut() else {
+        return Ok(());
+    };
+    let result = join.await;
+    *worker = None;
+    result.context("Runtime worker shutdown failed")
+}
+
 pub type SharedRuntimeThreadManager = Arc<RuntimeThreadManager>;
 
 #[derive(Clone)]
@@ -3876,13 +3900,18 @@ pub struct RuntimeThreadManager {
     /// This orders both route changes and saved-history boundaries with dispatch.
     config_admission: Arc<AsyncRwLock<()>>,
     engine_load: Arc<Mutex<()>>,
+    shutdown_drain: Arc<Mutex<()>>,
+    engine_workers: Arc<parking_lot::Mutex<Vec<(EngineHandle, RuntimeCompletion)>>>,
+    turn_monitors: Arc<parking_lot::Mutex<Vec<RuntimeCompletion>>>,
     active: Arc<Mutex<ActiveThreads>>,
     event_emit: Arc<Mutex<()>>,
     projection_locks: Arc<parking_lot::Mutex<HashMap<String, Arc<Mutex<()>>>>>,
     event_tx: broadcast::Sender<RuntimeEventRecord>,
     manager_cfg: RuntimeThreadManagerConfig,
     cancel_token: CancellationToken,
-    task_manager: Arc<parking_lot::Mutex<Option<crate::task_manager::SharedTaskManager>>>,
+    task_manager: Arc<parking_lot::Mutex<std::sync::Weak<crate::task_manager::TaskManager>>>,
+    task_execution_lease:
+        Arc<parking_lot::Mutex<Option<Arc<crate::task_manager::TaskExecutionLease>>>>,
     automations:
         Arc<parking_lot::Mutex<Option<crate::automation_manager::SharedAutomationManager>>>,
     pending_approvals: Arc<parking_lot::Mutex<HashMap<String, PendingApprovalEntry>>>,
@@ -3895,11 +3924,34 @@ pub struct RuntimeThreadManager {
 }
 
 #[derive(Debug)]
-struct RuntimeProcessOwnerLock {
+pub(crate) struct RuntimeProcessOwnerLock {
     _file: File,
 }
 
 impl RuntimeProcessOwnerLock {
+    /// Reuse the Runtime's protected OS lease for task-store ownership. A
+    /// missing existing lease is uncertainty, never proof that an owner died.
+    pub(crate) fn try_acquire_file(path: &Path, create: bool) -> Result<Option<Self>> {
+        let root = checked_runtime_store_root(
+            path.parent()
+                .context("Owner lease has no parent")?
+                .to_path_buf(),
+        )?;
+        ensure_runtime_store_dir(&root)?;
+        let file = open_runtime_store_file(path, "Execution owner lease", |options| {
+            options
+                .create(create)
+                .truncate(false)
+                .read(true)
+                .write(true);
+        })?;
+        match Self::try_lock_exclusive(&file) {
+            Ok(()) => Ok(Some(Self { _file: file })),
+            Err(error) if Self::is_contention(&error) => Ok(None),
+            Err(error) => Err(error).context("Failed to acquire execution owner lease"),
+        }
+    }
+
     fn acquire(root: &Path) -> Result<Self> {
         let root = checked_runtime_store_root(root.to_path_buf())?;
         ensure_runtime_store_dir(&root)?;
@@ -4371,13 +4423,17 @@ impl RuntimeThreadManager {
             _process_owner_lock: process_owner_lock,
             config_admission: Arc::new(AsyncRwLock::new(())),
             engine_load: Arc::new(Mutex::new(())),
+            shutdown_drain: Arc::new(Mutex::new(())),
+            engine_workers: Arc::new(parking_lot::Mutex::new(Vec::new())),
+            turn_monitors: Arc::new(parking_lot::Mutex::new(Vec::new())),
             active: Arc::new(Mutex::new(ActiveThreads::default())),
             event_emit: Arc::new(Mutex::new(())),
             projection_locks: Arc::new(parking_lot::Mutex::new(HashMap::new())),
             event_tx,
             manager_cfg,
             cancel_token: CancellationToken::new(),
-            task_manager: Arc::new(parking_lot::Mutex::new(None)),
+            task_manager: Arc::new(parking_lot::Mutex::new(std::sync::Weak::new())),
+            task_execution_lease: Arc::new(parking_lot::Mutex::new(None)),
             automations: Arc::new(parking_lot::Mutex::new(None)),
             pending_approvals: Arc::new(parking_lot::Mutex::new(HashMap::new())),
             pending_user_inputs: Arc::new(parking_lot::Mutex::new(HashMap::new())),
@@ -4394,7 +4450,140 @@ impl RuntimeThreadManager {
     /// Attach the durable task manager so model-visible task tools work inside
     /// runtime thread turns as well as interactive TUI turns.
     pub fn attach_task_manager(&self, task_manager: crate::task_manager::SharedTaskManager) {
-        *self.task_manager.lock() = Some(task_manager);
+        *self.task_manager.lock() = Arc::downgrade(&task_manager);
+    }
+
+    /// Identity of the actual Runtime store, not a model-supplied session label.
+    pub(crate) fn task_execution_identity(&self) -> (String, Arc<RuntimeProcessOwnerLock>) {
+        let mut digest = Sha256::new();
+        digest.update(self.store.owner_id.as_bytes());
+        digest.update([0]);
+        digest.update(self.store.event_lock_path.as_os_str().as_encoded_bytes());
+        (
+            digest
+                .finalize()
+                .iter()
+                .map(|byte| format!("{byte:02x}"))
+                .collect(),
+            self._process_owner_lock.clone(),
+        )
+    }
+
+    pub(crate) async fn close_execution_admission(&self) {
+        let _admission = self.config_admission.write().await;
+        self.cancel_token.cancel();
+        let _loading = self.engine_load.lock().await;
+    }
+
+    /// Close admission, then drain the existing engines and terminal receipt
+    /// monitors. Ownership stays attached to this Runtime until its last user
+    /// drops it, including any actual execution still awaiting shutdown.
+    pub(crate) async fn shutdown_and_wait(&self) -> Result<()> {
+        let _drain = self.shutdown_drain.lock().await;
+        self.close_execution_admission().await;
+        let (engines, active_turns) = {
+            let active = self.active.lock().await;
+            (
+                active
+                    .engines
+                    .values()
+                    .map(|state| state.engine.clone())
+                    .collect::<Vec<_>>(),
+                active
+                    .engines
+                    .iter()
+                    .filter_map(|(id, state)| {
+                        state
+                            .active_turn
+                            .as_ref()
+                            .map(|turn| (id.clone(), turn.turn_id.clone()))
+                    })
+                    .collect::<Vec<_>>(),
+            )
+        };
+        let mut failure = None;
+        for (thread_id, turn_id) in active_turns {
+            if let Err(error) = self.interrupt_turn(&thread_id, &turn_id).await {
+                if !self
+                    .store
+                    .load_turn(&turn_id)
+                    .is_ok_and(|turn| turn.status != RuntimeTurnStatus::InProgress)
+                {
+                    failure = Some(anyhow!("Runtime turn interruption failed: {error}"));
+                }
+            }
+        }
+        let workers = self.engine_workers.lock().clone();
+        for engine in engines
+            .iter()
+            .chain(workers.iter().map(|(engine, _)| engine))
+        {
+            engine.cancel_with_reason(crate::core::engine::CancelReason::Internal);
+        }
+        for engine in engines
+            .iter()
+            .chain(workers.iter().map(|(engine, _)| engine))
+        {
+            let _ = engine.send(Op::Shutdown).await;
+        }
+        for (_, worker) in workers {
+            if let Err(error) = await_retained_completion(&worker).await {
+                failure = Some(anyhow!("Runtime engine shutdown failed: {error}"));
+            }
+        }
+        // Recovery reads can publish receipts too. Flushes that already passed
+        // their admission check retain this mutex through all receipt writes;
+        // later readers see the closed latch under the same mutex.
+        {
+            let _recovery = self.recovery_flush.lock().await;
+        }
+        loop {
+            let monitors = self.turn_monitors.lock().clone();
+            for monitor in monitors {
+                if let Err(error) = await_retained_completion(&monitor).await {
+                    failure = Some(anyhow!("Runtime terminal receipt monitor failed: {error}"));
+                }
+            }
+            let mut monitors = self.turn_monitors.lock();
+            monitors.retain(|monitor| !completion_finished(monitor));
+            if monitors.is_empty() {
+                break;
+            }
+        }
+        self.engine_workers
+            .lock()
+            .retain(|(_, worker)| !completion_finished(worker));
+        let mut active = self.active.lock().await;
+        active.engines.clear();
+        active.lru.clear();
+        failure.map_or(Ok(()), Err)
+    }
+
+    fn track_receipt_worker(&self, worker: tokio::task::JoinHandle<()>) {
+        let mut monitors = self.turn_monitors.lock();
+        monitors.retain(|monitor| !completion_finished(monitor));
+        monitors.push(retained_completion(worker));
+    }
+
+    fn ensure_accepting_execution(&self) -> Result<()> {
+        if self.cancel_token.is_cancelled() {
+            bail!("Runtime is shutting down; execution admission is closed");
+        }
+        Ok(())
+    }
+
+    /// Runtime clones retained by live Engine/tool work retain this lease too.
+    /// A TaskManager cancellation timeout cannot release execution ownership.
+    pub(crate) fn retain_task_execution_lease(
+        &self,
+        lease: Arc<crate::task_manager::TaskExecutionLease>,
+    ) -> Result<()> {
+        let mut current = self.task_execution_lease.lock();
+        if current.is_some() {
+            bail!("This Runtime already owns a task execution manager");
+        }
+        *current = Some(lease);
+        Ok(())
     }
 
     /// Attach the automation manager for model-visible scheduling tools.
@@ -4766,6 +4955,8 @@ impl RuntimeThreadManager {
         call_id: &str,
         result: DynamicToolCallResult,
     ) -> Result<bool> {
+        let admission = self.config_admission.read().await;
+        self.ensure_accepting_execution()?;
         let claim = match self.claim_pending_dynamic_tool(thread_id, turn_id, call_id) {
             PendingDynamicToolClaim::Claimed(claim) => claim,
             PendingDynamicToolClaim::Settling(_) | PendingDynamicToolClaim::Missing => {
@@ -4779,6 +4970,7 @@ impl RuntimeThreadManager {
         };
         let ack =
             self.spawn_dynamic_tool_settlement(claim, DynamicToolTerminalOutcome::Resolved(result));
+        drop(admission);
         Ok(Self::await_dynamic_tool_settlement(ack)
             .await?
             .result_accepted)
@@ -4790,6 +4982,8 @@ impl RuntimeThreadManager {
         input_id: &str,
         response: crate::tools::user_input::UserInputResponse,
     ) -> Result<bool> {
+        let admission = self.config_admission.read().await;
+        self.ensure_accepting_execution()?;
         let engine = {
             let active = self.active.lock().await;
             let Some(state) = active.engines.get(thread_id) else {
@@ -4814,22 +5008,27 @@ impl RuntimeThreadManager {
         // between durable acceptance and engine delivery.
         let manager = self.clone();
         let thread_id = thread_id.to_string();
-        tokio::spawn(async move {
-            manager
+        let (ack_tx, ack_rx) = oneshot::channel();
+        let worker = tokio::spawn(async move {
+            let result = manager
                 .settle_claimed_user_input(
                     &thread_id,
                     Some(engine),
                     request,
                     UserInputTerminalOutcome::Answered(response),
                 )
-                .await
-        })
-        .await
-        .context("User-input settlement task failed")?
+                .await;
+            let _ = ack_tx.send(result);
+        });
+        self.track_receipt_worker(worker);
+        drop(admission);
+        ack_rx.await.context("User-input settlement task failed")?
     }
 
     #[allow(dead_code)]
     pub async fn cancel_user_input(&self, thread_id: &str, input_id: &str) -> Result<bool> {
+        let admission = self.config_admission.read().await;
+        self.ensure_accepting_execution()?;
         let engine = {
             let active = self.active.lock().await;
             let Some(state) = active.engines.get(thread_id) else {
@@ -4850,18 +5049,23 @@ impl RuntimeThreadManager {
         };
         let manager = self.clone();
         let thread_id = thread_id.to_string();
-        tokio::spawn(async move {
-            manager
+        let (ack_tx, ack_rx) = oneshot::channel();
+        let worker = tokio::spawn(async move {
+            let result = manager
                 .settle_claimed_user_input(
                     &thread_id,
                     Some(engine),
                     request,
                     UserInputTerminalOutcome::Canceled { terminal: false },
                 )
-                .await
-        })
-        .await
-        .context("User-input cancellation task failed")?
+                .await;
+            let _ = ack_tx.send(result);
+        });
+        self.track_receipt_worker(worker);
+        drop(admission);
+        ack_rx
+            .await
+            .context("User-input cancellation task failed")?
     }
 
     async fn settle_claimed_user_input(
@@ -5357,28 +5561,39 @@ impl RuntimeThreadManager {
     }
 
     /// Arm one goal continuation pass to run after the quiet period. The
-    /// sleep is deliberately never cancelled: a pause, clear, completion, or
+    /// sleep is interrupted by Runtime shutdown. A pause, clear, completion, or
     /// cap that lands while the timer runs is honored by the re-read inside
     /// `run_goal_continuation`, which is the cancellation path — `DELETE
     /// /goal` and status syncs therefore do not need to interrupt this task.
     fn spawn_goal_continuation(&self, thread_id: String, delay_seconds: u64) {
+        if self.cancel_token.is_cancelled() {
+            return;
+        }
         let manager = self.clone();
-        tokio::spawn(async move {
+        let worker = tokio::spawn(async move {
             // Quiet period between passes, mirroring the interactive
             // engine's `goal_continuation_delay_seconds` behavior (#5508).
             if delay_seconds > 0 {
-                tokio::time::sleep(std::time::Duration::from_secs(delay_seconds)).await;
+                tokio::select! {
+                    _ = manager.cancel_token.cancelled() => return,
+                    _ = tokio::time::sleep(std::time::Duration::from_secs(delay_seconds)) => {}
+                }
+            }
+            if manager.cancel_token.is_cancelled() {
+                return;
             }
             if let Err(err) = manager.run_goal_continuation(&thread_id).await {
                 tracing::warn!("goal continuation for {thread_id} failed: {err}");
             }
         });
+        self.track_receipt_worker(worker);
     }
 
     /// Dispatch one goal continuation pass after the quiet period. Every
     /// guard re-reads durable state: the goal may have been paused, cleared,
     /// completed, or capped while the timer ran.
     async fn run_goal_continuation(&self, thread_id: &str) -> Result<()> {
+        self.ensure_accepting_execution()?;
         let Some(goal) = self.store.load_goal(thread_id)? else {
             return Ok(());
         };
@@ -5552,6 +5767,7 @@ impl RuntimeThreadManager {
         thread_id: &str,
         message_id: &AgentMailMessageId,
     ) -> Result<(AgentMailEnvelope, Option<TurnRecord>)> {
+        self.ensure_accepting_execution()?;
         let thread = self.get_thread(thread_id).await?;
         let address = agent_mail_address(&self.store.owner_id, &thread)?;
         {
@@ -5574,6 +5790,7 @@ impl RuntimeThreadManager {
 
         let (claimed, terminal) = {
             let _mail_mutation = self.store.mail_mutation.lock();
+            self.ensure_accepting_execution()?;
             let mut envelope = self.store.load_agent_mail(message_id)?;
             if envelope.destination != address {
                 bail!("Agent Mail ownership denied: message does not belong to this destination");
@@ -5773,6 +5990,7 @@ impl RuntimeThreadManager {
     }
 
     async fn deliver_next_wake_agent_mail(&self, thread_id: &str) -> Result<()> {
+        self.ensure_accepting_execution()?;
         let next = self
             .list_agent_mail_for_thread(thread_id)
             .await?
@@ -5794,12 +6012,16 @@ impl RuntimeThreadManager {
     }
 
     fn spawn_agent_mail_safe_boundary_delivery(&self, thread_id: String) {
+        if self.cancel_token.is_cancelled() {
+            return;
+        }
         let manager = self.clone();
-        tokio::spawn(async move {
+        let worker = tokio::spawn(async move {
             if let Err(error) = Box::pin(manager.deliver_next_wake_agent_mail(&thread_id)).await {
                 tracing::warn!(thread_id, %error, "Failed to deliver queued Agent Mail");
             }
         });
+        self.track_receipt_worker(worker);
     }
 
     fn projection_lock(&self, thread_id: &str) -> Arc<Mutex<()>> {
@@ -5922,6 +6144,9 @@ impl RuntimeThreadManager {
             return Ok(());
         }
         let _recovery_flush = self.recovery_flush.lock().await;
+        if self.cancel_token.is_cancelled() {
+            bail!("Runtime is shutting down; recovery receipts remain pending");
+        }
         loop {
             let next = self
                 .recovery_receipts
@@ -6046,7 +6271,7 @@ impl RuntimeThreadManager {
     ) -> oneshot::Receiver<std::result::Result<DynamicToolSettlementAck, String>> {
         let (ack_tx, ack_rx) = oneshot::channel();
         let manager = self.clone();
-        tokio::spawn(async move {
+        let worker = tokio::spawn(async move {
             use futures_util::FutureExt;
 
             let mut claim = Some(claim);
@@ -6184,6 +6409,7 @@ impl RuntimeThreadManager {
             };
             let _ = ack_tx.send(result);
         });
+        self.track_receipt_worker(worker);
         ack_rx
     }
 
@@ -8005,7 +8231,7 @@ impl RuntimeThreadManager {
     ) -> oneshot::Receiver<std::result::Result<TurnRecord, String>> {
         let (acceptance_tx, acceptance_rx) = oneshot::channel();
         let manager = Arc::new(self.clone());
-        tokio::spawn(async move {
+        let monitor = tokio::spawn(async move {
             use futures_util::FutureExt;
             let start_events = std::panic::AssertUnwindSafe(manager.emit_claimed_turn_started(
                 &turn,
@@ -8034,6 +8260,7 @@ impl RuntimeThreadManager {
                 .monitor_claimed_turn(turn.thread_id.clone(), turn.id.clone(), engine, kind)
                 .await;
         });
+        self.track_receipt_worker(monitor);
         acceptance_rx
     }
 
@@ -8045,7 +8272,7 @@ impl RuntimeThreadManager {
     ) -> oneshot::Receiver<TurnRecord> {
         let (receipt_tx, receipt_rx) = oneshot::channel();
         let manager = Arc::new(self.clone());
-        tokio::spawn(async move {
+        let worker = tokio::spawn(async move {
             use futures_util::FutureExt;
             let receipts = std::panic::AssertUnwindSafe(async {
                 if let Err(err) = manager
@@ -8087,6 +8314,7 @@ impl RuntimeThreadManager {
             }
             let _ = receipt_tx.send(turn);
         });
+        self.track_receipt_worker(worker);
         receipt_rx
     }
 
@@ -8150,6 +8378,7 @@ impl RuntimeThreadManager {
         // engine handoff, so a completed reload is a hard boundary: no later
         // dispatch can carry its predecessor's URL, key, model, or policy.
         let _config_admission = self.config_admission.read().await;
+        self.ensure_accepting_execution()?;
         let prompt = req.prompt.trim().to_string();
         if prompt.is_empty() {
             bail!("prompt is required");
@@ -8662,6 +8891,8 @@ impl RuntimeThreadManager {
         turn_id: &str,
         req: SteerTurnRequest,
     ) -> Result<TurnRecord> {
+        let _admission = self.config_admission.read().await;
+        self.ensure_accepting_execution()?;
         let prompt = req.prompt.trim().to_string();
         if prompt.is_empty() {
             bail!("prompt is required");
@@ -8771,6 +9002,7 @@ impl RuntimeThreadManager {
         // handoff so it cannot dispatch an old credential or endpoint after a
         // successful config reload.
         let _config_admission = self.config_admission.read().await;
+        self.ensure_accepting_execution()?;
         let thread = self.get_thread(thread_id).await?;
         let engine = self.ensure_engine_loaded(&thread).await?;
 
@@ -9009,8 +9241,10 @@ impl RuntimeThreadManager {
     }
 
     async fn ensure_engine_loaded(&self, thread_hint: &ThreadRecord) -> Result<EngineHandle> {
+        self.ensure_accepting_execution()?;
         {
             let mut active = self.active.lock().await;
+            self.ensure_accepting_execution()?;
             if let Some(engine) = active
                 .engines
                 .get(thread_hint.id.as_str())
@@ -9024,9 +9258,11 @@ impl RuntimeThreadManager {
         // Only one cache-miss build may run at a time. Recheck after taking
         // the build lock because another caller may already have won.
         let _engine_load = self.engine_load.lock().await;
+        self.ensure_accepting_execution()?;
         loop {
             {
                 let mut active = self.active.lock().await;
+                self.ensure_accepting_execution()?;
                 if let Some(engine) = active
                     .engines
                     .get(thread_hint.id.as_str())
@@ -9176,7 +9412,7 @@ impl RuntimeThreadManager {
                     .saturating_mul(1024 * 1024 * 1024),
                 lsp_config,
                 runtime_services: crate::tools::spec::RuntimeToolServices {
-                    task_manager: self.task_manager.lock().clone(),
+                    task_manager: self.task_manager.lock().upgrade(),
                     automations: self.automations.lock().clone(),
                     task_data_dir: Some(self.manager_cfg.task_data_dir.clone()),
                     active_task_id: thread.task_id.clone(),
@@ -9275,11 +9511,16 @@ impl RuntimeThreadManager {
 
             // Verify the persisted history before spawning an Engine task.
             let session_messages = self.restore_thread_messages(&thread)?;
-            let engine = spawn_engine_with_authoritative_route_config(
+            let (engine, worker) = spawn_engine_with_authoritative_route_config(
                 engine_cfg,
                 &cfg,
                 Arc::clone(&self.config),
             );
+            {
+                let mut workers = self.engine_workers.lock();
+                workers.retain(|(_, worker)| !completion_finished(worker));
+                workers.push((engine.clone(), retained_completion(worker)));
+            }
 
             let sys_prompt = thread
                 .system_prompt

crates/tui/src/runtime_threads/tests.rs

@@ -14237,3 +14237,5 @@ async fn report_store_failure_names_the_item_file_without_terminalizing() -> Res
     let _ = std::fs::remove_dir_all(dir);
     Ok(())
 }
+
+mod task_ownership;

crates/tui/src/runtime_threads/tests/task_ownership.rs

@@ -0,0 +1,354 @@
+use super::*;
+use crate::task_manager::{TaskManager, TaskManagerConfig};
+
+fn fixture_config() -> Config {
+    let mut config = Config {
+        api_key: Some("local-fixture-key".into()),
+        base_url: Some("http://127.0.0.1:1/v1".into()),
+        ..Config::default()
+    };
+    config.set_feature("mcp", false).unwrap();
+    config.set_feature("subagents", false).unwrap();
+    config
+}
+
+#[tokio::test]
+async fn idle_runtime_engine_cycles_are_drained_before_same_scope_reopen() -> Result<()> {
+    let _env = crate::test_support::lock_test_env();
+    let root = tempfile::tempdir()?;
+    let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", root.path());
+    let cfg = fixture_config();
+    let runtime_config = test_manager_config(root.path().join("runtime"));
+    let runtime = Arc::new(RuntimeThreadManager::open(
+        cfg.clone(),
+        root.path().into(),
+        runtime_config.clone(),
+    )?);
+    let tasks = TaskManager::start_with_runtime_manager(
+        TaskManagerConfig::from_runtime(&cfg, root.path().into(), None, Some(1)),
+        cfg.clone(),
+        runtime.clone(),
+    )
+    .await?;
+    let thread = runtime
+        .create_thread(CreateThreadRequest::default())
+        .await?;
+    // Load the real production Engine with its actual strong Runtime services.
+    // No turn is submitted, so constructing this Engine makes no provider call.
+    let engine = runtime.get_engine(&thread.id).await?;
+    engine.get_session_snapshot().await?;
+    runtime.spawn_goal_continuation(thread.id.clone(), 3_600);
+    let weak = Arc::downgrade(&tasks);
+    tasks.shutdown_and_wait().await?;
+    assert!(runtime.active.lock().await.engines.is_empty());
+    assert!(runtime.get_engine(&thread.id).await.is_err());
+    assert!(
+        RuntimeThreadManager::open(cfg.clone(), root.path().into(), runtime_config.clone())
+            .is_err(),
+        "retained Runtime handle still owns its scope"
+    );
+    drop(engine);
+    drop(tasks);
+    drop(runtime);
+    assert!(
+        weak.upgrade().is_none(),
+        "idle Engine service cycle must have ended"
+    );
+    let reopened = RuntimeThreadManager::open(cfg, root.path().into(), runtime_config)?;
+    reopened.shutdown_and_wait().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn shutdown_waits_actual_engine_exit_and_terminal_monitor_before_releasing_scope()
+-> Result<()> {
+    use crate::core::engine::Engine;
+    use crate::llm_client::mock::{MockLlmClient, canned};
+    let _env = crate::test_support::lock_test_env();
+    let root = tempfile::tempdir()?;
+    let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", root.path());
+    let cfg = fixture_config();
+    let manager_cfg = test_manager_config(root.path().join("runtime"));
+    let runtime = Arc::new(RuntimeThreadManager::open(
+        cfg.clone(),
+        root.path().into(),
+        manager_cfg.clone(),
+    )?);
+    let tasks = TaskManager::start_with_runtime_manager(
+        TaskManagerConfig::from_runtime(&cfg, root.path().into(), None, Some(1)),
+        cfg.clone(),
+        runtime.clone(),
+    )
+    .await?;
+    let thread = runtime
+        .create_thread(CreateThreadRequest::default())
+        .await?;
+    let model = Arc::new(MockLlmClient::new(vec![canned::simple_text_turn(
+        "owned fixture",
+    )]));
+    let (engine, handle) = Engine::new_with_model_client(
+        EngineConfig {
+            workspace: root.path().into(),
+            model: thread.model.clone(),
+            subagents_enabled: false,
+            snapshots_enabled: false,
+            memory_enabled: false,
+            terminal_chrome_enabled: false,
+            runtime_services: crate::tools::spec::RuntimeToolServices {
+                task_manager: Some(tasks.clone()),
+                active_thread_id: Some(thread.id.clone()),
+                dynamic_tool_executor: Some(Arc::new(runtime.as_ref().clone())),
+                ..Default::default()
+            },
+            ..EngineConfig::default()
+        },
+        &cfg,
+        model.clone(),
+    );
+    runtime
+        .install_test_engine(&thread.id, handle.clone())
+        .await?;
+    let (release, blocked) = oneshot::channel();
+    let worker = tokio::spawn(async move {
+        blocked.await.unwrap();
+        engine.run().await;
+    });
+    runtime
+        .engine_workers
+        .lock()
+        .push((handle.clone(), retained_completion(worker)));
+    let turn = runtime
+        .start_turn(
+            &thread.id,
+            StartTurnRequest {
+                prompt: "fixture admission".into(),
+                ..Default::default()
+            },
+        )
+        .await?;
+    let drain = tokio::spawn({
+        let tasks = tasks.clone();
+        async move { tasks.shutdown_and_wait().await }
+    });
+    tokio::time::timeout(Duration::from_secs(5), runtime.cancel_token.cancelled()).await?;
+    assert!(
+        !drain.is_finished(),
+        "a cancellation request cannot prove Engine exit"
+    );
+    assert!(
+        RuntimeThreadManager::open(cfg.clone(), root.path().into(), manager_cfg.clone()).is_err()
+    );
+    assert!(
+        runtime
+            .start_turn(
+                &thread.id,
+                StartTurnRequest {
+                    prompt: "late admission".into(),
+                    ..Default::default()
+                }
+            )
+            .await
+            .is_err()
+    );
+    let completion = runtime.engine_workers.lock()[0].1.clone();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while completion.try_lock().is_ok() {
+            sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await?;
+    drain.abort();
+    assert!(drain.await.unwrap_err().is_cancelled());
+    let retry = tokio::spawn({
+        let tasks = tasks.clone();
+        async move { tasks.shutdown_and_wait().await }
+    });
+    sleep(Duration::from_millis(25)).await;
+    assert!(
+        !retry.is_finished(),
+        "retry must still own and await the original Engine join"
+    );
+    release.send(()).unwrap();
+    tokio::time::timeout(Duration::from_secs(15), retry).await???;
+    drop(completion);
+    assert!(runtime.store.load_turn(&turn.id)?.status != RuntimeTurnStatus::InProgress);
+    drop(handle);
+    drop(tasks);
+    drop(runtime);
+    let reopened = RuntimeThreadManager::open(cfg, root.path().into(), manager_cfg)?;
+    reopened.shutdown_and_wait().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn shutdown_drains_accepted_user_input_receipt_after_caller_disconnects() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let runtime = Arc::new(test_manager(root.path().join("runtime"))?);
+    let thread = runtime
+        .create_thread(CreateThreadRequest::default())
+        .await?;
+    let harness = mock_engine_handle();
+    runtime
+        .install_test_engine(&thread.id, harness.handle.clone())
+        .await?;
+    runtime.register_pending_user_input(
+        &thread.id,
+        PendingUserInputRequest {
+            id: "input_drain".into(),
+            turn_id: "turn_drain".into(),
+            request: crate::tools::user_input::UserInputRequest {
+                questions: Vec::new(),
+            },
+        },
+    );
+    let hold_receipt = runtime.event_emit.lock().await;
+    let submission = tokio::spawn({
+        let runtime = runtime.clone();
+        let thread_id = thread.id.clone();
+        async move {
+            runtime
+                .submit_user_input(
+                    &thread_id,
+                    "input_drain",
+                    crate::tools::user_input::UserInputResponse {
+                        answers: Vec::new(),
+                    },
+                )
+                .await
+        }
+    });
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while runtime.turn_monitors.lock().is_empty() {
+            sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await?;
+    submission.abort();
+    assert!(submission.await.unwrap_err().is_cancelled());
+    let drain = tokio::spawn({
+        let runtime = runtime.clone();
+        async move { runtime.shutdown_and_wait().await }
+    });
+    tokio::time::timeout(Duration::from_secs(5), runtime.cancel_token.cancelled()).await?;
+    sleep(Duration::from_millis(25)).await;
+    assert!(
+        !drain.is_finished(),
+        "accepted detached receipt is still part of shutdown"
+    );
+    drop(hold_receipt);
+    tokio::time::timeout(Duration::from_secs(5), drain).await???;
+    let events = runtime.events_since(&thread.id, None)?;
+    assert_eq!(
+        events
+            .iter()
+            .filter(|event| event.event == "user_input.answered")
+            .count(),
+        1
+    );
+    assert!(runtime.pending_user_inputs.lock().is_empty());
+    Ok(())
+}
+
+#[tokio::test]
+async fn shutdown_fences_recovery_readers_without_publishing_late_receipts() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let runtime = Arc::new(test_manager(root.path().join("runtime"))?);
+    let thread = runtime
+        .create_thread(CreateThreadRequest::default())
+        .await?;
+    let turn = sample_turn(&thread.id, "turn_recovery_drain", RuntimeTurnStatus::Failed);
+    runtime.store.save_turn(&turn)?;
+    runtime.queue_recovery_receipt(RecoveredTurnReceipt {
+        turn,
+        unresolved_dynamic_tools: Vec::new(),
+    });
+    let hold = runtime.recovery_flush.lock().await;
+    let reader = tokio::spawn({
+        let runtime = runtime.clone();
+        let id = thread.id.clone();
+        async move { runtime.get_thread(&id).await }
+    });
+    tokio::task::yield_now().await;
+    let shutdown = tokio::spawn({
+        let runtime = runtime.clone();
+        async move { runtime.shutdown_and_wait().await }
+    });
+    tokio::time::timeout(Duration::from_secs(5), runtime.cancel_token.cancelled()).await?;
+    assert!(
+        !shutdown.is_finished(),
+        "shutdown must fence pending recovery producers"
+    );
+    drop(hold);
+    assert!(
+        tokio::time::timeout(Duration::from_secs(5), reader)
+            .await??
+            .is_err()
+    );
+    tokio::time::timeout(Duration::from_secs(5), shutdown).await???;
+    assert!(
+        runtime.recovery_receipts.lock().contains_key(&thread.id),
+        "unadmitted recovery remains queued for the next owner"
+    );
+    assert!(
+        runtime
+            .events_since(&thread.id, None)?
+            .iter()
+            .all(|event| event.event != "turn.completed")
+    );
+    assert!(runtime.get_thread_detail(&thread.id).await.is_err());
+    Ok(())
+}
+
+#[tokio::test]
+async fn shutdown_drains_an_already_started_recovery_flush() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let runtime = Arc::new(test_manager(root.path().join("runtime"))?);
+    let thread = runtime
+        .create_thread(CreateThreadRequest::default())
+        .await?;
+    let turn = sample_turn(
+        &thread.id,
+        "turn_started_recovery",
+        RuntimeTurnStatus::Failed,
+    );
+    runtime.store.save_turn(&turn)?;
+    runtime.queue_recovery_receipt(RecoveredTurnReceipt {
+        turn,
+        unresolved_dynamic_tools: Vec::new(),
+    });
+    let hold = runtime.event_emit.lock().await;
+    let reader = tokio::spawn({
+        let runtime = runtime.clone();
+        let id = thread.id.clone();
+        async move { runtime.get_thread(&id).await }
+    });
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while runtime.recovery_flush.try_lock().is_ok() {
+            sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await?;
+    let shutdown = tokio::spawn({
+        let runtime = runtime.clone();
+        async move { runtime.shutdown_and_wait().await }
+    });
+    tokio::time::timeout(Duration::from_secs(5), runtime.cancel_token.cancelled()).await?;
+    sleep(Duration::from_millis(25)).await;
+    assert!(
+        !shutdown.is_finished(),
+        "accepted recovery write must settle before drain succeeds"
+    );
+    drop(hold);
+    tokio::time::timeout(Duration::from_secs(5), reader).await???;
+    tokio::time::timeout(Duration::from_secs(5), shutdown).await???;
+    assert!(!runtime.recovery_receipts.lock().contains_key(&thread.id));
+    assert_eq!(
+        runtime
+            .events_since(&thread.id, None)?
+            .iter()
+            .filter(|event| event.event == "turn.completed")
+            .count(),
+        1
+    );
+    Ok(())
+}

crates/tui/src/task_manager/ownership_tests.rs

@@ -0,0 +1,575 @@
+use super::*;
+use std::io::Write;
+use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+fn config(root: &Path, scope: &str) -> TaskManagerConfig {
+    TaskManagerConfig {
+        data_dir: root.to_path_buf(),
+        worker_count: 1,
+        default_workspace: root.join(scope),
+        default_model: format!("{scope}-model"),
+        default_mode: "plan".into(),
+        allow_shell: false,
+        trust_mode: false,
+        execution_limits: TaskExecutionLimits::default(),
+    }
+}
+
+struct RecordingExecutor {
+    root: PathBuf,
+    scope: String,
+}
+
+#[async_trait]
+impl TaskExecutor for RecordingExecutor {
+    async fn execute(
+        &self,
+        task: ExecutionTask,
+        events: mpsc::Sender<TaskExecutionEvent>,
+        cancel: CancellationToken,
+    ) -> TaskExecutionResult {
+        let mut receipt = fs::OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(self.root.join(format!("{}.executions", self.scope)))
+            .unwrap();
+        writeln!(
+            receipt,
+            "{}",
+            serde_json::json!({
+                "id": task.id, "scope": self.scope, "model": task.model,
+                "provider": task.model_provider, "provider_id": task.model_provider_id,
+                "workspace": task.workspace, "prompt": task.prompt,
+                "allow_shell": task.allow_shell, "trust_mode": task.trust_mode,
+            })
+        )
+        .unwrap();
+        if task.prompt.ends_with("hold") {
+            cancel.cancelled().await;
+            fs::write(self.root.join(format!("{}.canceled", self.scope)), &task.id).unwrap();
+        }
+        events
+            .send(TaskExecutionEvent::MessageDelta {
+                content: format!("{} result", self.scope),
+            })
+            .await
+            .ok();
+        TaskExecutionResult {
+            status: TaskStatus::Completed,
+            result_text: Some(format!("{} completed", self.scope)),
+            error: None,
+            terminal_reason: TaskTerminalReason::Completed,
+        }
+    }
+}
+
+async fn manager(root: &Path, scope: &str) -> Result<SharedTaskManager> {
+    TaskManager::start_with_executor_in_scope(
+        config(root, scope),
+        Arc::new(RecordingExecutor {
+            root: root.to_path_buf(),
+            scope: scope.into(),
+        }),
+        scope,
+    )
+    .await
+}
+
+async fn wait_file(path: &Path) -> Result<Vec<u8>> {
+    tokio::time::timeout(Duration::from_secs(15), async {
+        loop {
+            match fs::read(path) {
+                Ok(bytes) if !bytes.is_empty() => return Ok(bytes),
+                Ok(_) => {}
+                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+                Err(error) => return Err(error.into()),
+            }
+            sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .context("fixture receipt timeout")?
+}
+
+fn executions(root: &Path, scope: &str) -> Vec<Value> {
+    fs::read_to_string(root.join(format!("{scope}.executions")))
+        .unwrap_or_default()
+        .lines()
+        .map(|line| serde_json::from_str(line).unwrap())
+        .collect()
+}
+
+struct ChildOwner(Child);
+impl Drop for ChildOwner {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+        let _ = self.0.wait();
+    }
+}
+fn child_owner(root: &Path) -> Result<ChildOwner> {
+    let log = fs::File::create(root.join("child.log"))?;
+    Ok(ChildOwner(
+        Command::new(std::env::current_exe()?)
+            .args([
+                "--exact",
+                "task_manager::ownership_tests::ownership_process_child",
+                "--ignored",
+                "--nocapture",
+            ])
+            .env("CW_TASK_OWNER_FIXTURE_ROOT", root)
+            .stdout(Stdio::from(log.try_clone()?))
+            .stderr(Stdio::from(log))
+            .spawn()?,
+    ))
+}
+
+#[tokio::test]
+#[ignore = "subprocess entry; invoked by shared-store ownership fixtures"]
+async fn ownership_process_child() -> Result<()> {
+    let root = PathBuf::from(std::env::var("CW_TASK_OWNER_FIXTURE_ROOT")?);
+    let owner = manager(&root, "A").await?;
+    let active = owner
+        .add_task(NewTaskRequest {
+            owner_session_id: Some("session-a".into()),
+            ..NewTaskRequest::from_prompt("A hold")
+        })
+        .await?;
+    wait_file(&root.join("A.executions")).await?;
+    let queued = owner
+        .add_task(NewTaskRequest {
+            owner_session_id: Some("session-a".into()),
+            ..NewTaskRequest::from_prompt("A queued")
+        })
+        .await?;
+    fs::write(
+        root.join("ready"),
+        serde_json::to_vec(&[active.id, queued.id])?,
+    )?;
+    wait_file(&root.join("stop")).await?;
+    owner.shutdown_and_wait().await?;
+    drop(owner);
+    Ok(())
+}
+
+#[tokio::test]
+async fn second_process_preserves_live_owner_and_durable_cancel_reaches_that_owner() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let mut child = child_owner(root.path())?;
+    let ids: Vec<String> = serde_json::from_slice(&wait_file(&root.path().join("ready")).await?)?;
+    let active_path = root.path().join("tasks").join(format!("{}.json", ids[0]));
+    let before = fs::read(&active_path)?;
+    let other = manager(root.path(), "B").await?;
+    assert_eq!(other.get_task(&ids[0]).await?.status, TaskStatus::Running);
+    assert_eq!(
+        fs::read(&active_path)?,
+        before,
+        "opening a foreign scope cannot rewrite live work"
+    );
+    assert!(
+        other
+            .get_task_for_owner(&ids[0], "session-b")
+            .await
+            .is_err()
+    );
+    assert!(
+        other
+            .cancel_task_for_owner(&ids[0][..12], "session-b")
+            .await
+            .is_err()
+    );
+    assert!(
+        other
+            .list_tasks_for_owner(None, None, "session-b")
+            .await?
+            .is_empty()
+    );
+    let own = other
+        .add_task(NewTaskRequest::from_prompt("B quick"))
+        .await?;
+    let done = wait_for_terminal_state(&other, &own.id, Duration::from_secs(10)).await?;
+    assert_eq!(done.status, TaskStatus::Completed);
+    assert_eq!(executions(root.path(), "B").len(), 1);
+    assert_eq!(executions(root.path(), "B")[0]["model"], "B-model");
+    assert_eq!(
+        executions(root.path(), "A").len(),
+        1,
+        "B cannot claim A's queued request"
+    );
+    other.cancel_task(&ids[1]).await?;
+    other.cancel_task(&ids[0]).await?;
+    wait_file(&root.path().join("A.canceled")).await?;
+    let canceled = wait_for_terminal_state(&other, &ids[0], Duration::from_secs(10)).await?;
+    assert_eq!(canceled.status, TaskStatus::Canceled);
+    assert!(canceled.cancel_requested_seq > 0);
+    assert_eq!(other.get_task(&ids[1]).await?.status, TaskStatus::Canceled);
+    assert_eq!(executions(root.path(), "A").len(), 1);
+    fs::write(root.path().join("stop"), "stop")?;
+    assert!(child.0.wait()?.success());
+    other.shutdown_and_wait().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn killed_process_is_reconciled_once_and_only_its_scope_resumes_queued_work() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let mut child = child_owner(root.path())?;
+    let ids: Vec<String> = serde_json::from_slice(&wait_file(&root.path().join("ready")).await?)?;
+    let observer = manager(root.path(), "B").await?;
+    assert!(
+        manager(root.path(), "A").await.is_err(),
+        "live scope cannot have a second executor"
+    );
+    child.0.kill()?;
+    child.0.wait()?;
+    let restarted = manager(root.path(), "A").await?;
+    let interrupted = restarted.get_task(&ids[0]).await?;
+    assert_eq!(interrupted.status, TaskStatus::Failed);
+    assert!(
+        interrupted
+            .error
+            .as_deref()
+            .unwrap()
+            .contains("Interrupted by process restart")
+    );
+    let terminal = wait_for_terminal_state(&restarted, &ids[1], Duration::from_secs(10)).await?;
+    assert_eq!(terminal.status, TaskStatus::Completed);
+    let receipt = executions(root.path(), "A");
+    assert_eq!(receipt.len(), 2);
+    assert_eq!(
+        receipt.iter().filter(|r| r["id"] == ids[0]).count(),
+        1,
+        "accepted Running work cannot replay"
+    );
+    assert_eq!(receipt[1]["id"], ids[1]);
+    assert_eq!(receipt[1]["model"], "A-model");
+    assert!(executions(root.path(), "B").is_empty());
+    restarted.shutdown_and_wait().await?;
+    drop(restarted);
+    let again = manager(root.path(), "A").await?;
+    assert_eq!(
+        again.get_task(&ids[0]).await?.lifecycle_seq,
+        interrupted.lifecycle_seq
+    );
+    assert_eq!(executions(root.path(), "A").len(), 2);
+    again.shutdown_and_wait().await?;
+    observer.shutdown_and_wait().await?;
+    Ok(())
+}
+
+async fn stop_idle_workers(manager: &TaskManager) {
+    for worker in std::mem::take(&mut *manager.workers.lock().await) {
+        worker.abort();
+        let _ = worker.await;
+    }
+}
+
+struct CountExecutor(Arc<AtomicUsize>);
+#[async_trait]
+impl TaskExecutor for CountExecutor {
+    async fn execute(
+        &self,
+        _: ExecutionTask,
+        _: mpsc::Sender<TaskExecutionEvent>,
+        _: CancellationToken,
+    ) -> TaskExecutionResult {
+        self.0.fetch_add(1, Ordering::SeqCst);
+        TaskExecutionResult::from_reason(TaskTerminalReason::Completed, None)
+    }
+}
+
+#[tokio::test]
+async fn cancellation_after_claim_before_first_poll_never_invokes_executor() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let calls = Arc::new(AtomicUsize::new(0));
+    let owner = TaskManager::start_with_executor_in_scope(
+        config(root.path(), "A"),
+        Arc::new(CountExecutor(calls.clone())),
+        "A",
+    )
+    .await?;
+    stop_idle_workers(&owner).await;
+    let task = owner
+        .add_task(NewTaskRequest::from_prompt("claim barrier"))
+        .await?;
+    let (id, request, cancel) = owner.claim_next_task().await?.context("claim")?;
+    let other = manager(root.path(), "B").await?;
+    other.cancel_task(&task.id).await?;
+    owner.run_task(id, request, cancel).await;
+    assert_eq!(calls.load(Ordering::SeqCst), 0);
+    assert_eq!(owner.get_task(&task.id).await?.status, TaskStatus::Canceled);
+    let healthy = owner
+        .add_task(NewTaskRequest::from_prompt("uncanceled control"))
+        .await?;
+    let (id, request, cancel) = owner.claim_next_task().await?.context("control claim")?;
+    owner.run_task(id, request, cancel).await;
+    assert_eq!(
+        calls.load(Ordering::SeqCst),
+        1,
+        "the same executor must run without cancellation"
+    );
+    assert_eq!(
+        owner.get_task(&healthy.id).await?.status,
+        TaskStatus::Completed
+    );
+    owner.shutdown_and_wait().await?;
+    other.shutdown_and_wait().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn pending_flush_and_finish_merge_fresh_metadata_and_monotonic_cancel() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let owner = manager(root.path(), "A").await?;
+    stop_idle_workers(&owner).await;
+    let task = owner
+        .add_task(NewTaskRequest::from_prompt("bounded deltas"))
+        .await?;
+    let (_, request, cancel) = owner.claim_next_task().await?.context("claim")?;
+    owner
+        .apply_execution_event(
+            &task.id,
+            TaskExecutionEvent::MessageDelta {
+                content: "accepted progress".into(),
+            },
+        )
+        .await?;
+    let other = manager(root.path(), "B").await?;
+    other
+        .record_tool_metadata(
+            &task.id,
+            &serde_json::json!({"task_updates": {"hunt_verdict": "wounded"}}),
+        )
+        .await?;
+    other.cancel_task(&task.id).await?;
+    let seq = other.get_task(&task.id).await?.cancel_requested_seq;
+    owner.flush_task(&task.id).await?;
+    owner
+        .finish_task(
+            &task.id,
+            TaskExecutionResult::from_reason(TaskTerminalReason::Completed, None),
+            cancel,
+            &request.mode_label,
+        )
+        .await?;
+    let final_task = other.get_task(&task.id).await?;
+    assert_eq!(final_task.status, TaskStatus::Canceled);
+    assert_eq!(final_task.cancel_requested_seq, seq);
+    assert_eq!(final_task.hunt_verdict.as_deref(), Some("wounded"));
+    assert_eq!(
+        final_task
+            .timeline
+            .iter()
+            .filter(|event| event.summary == "accepted progress")
+            .count(),
+        1
+    );
+    owner.shutdown_and_wait().await?;
+    other.shutdown_and_wait().await?;
+    Ok(())
+}
+
+#[cfg(unix)]
+struct ReadOnlyFixtureDir(PathBuf);
+#[cfg(unix)]
+impl ReadOnlyFixtureDir {
+    fn new(path: &Path) -> Result<Self> {
+        use std::os::unix::fs::PermissionsExt;
+        fs::set_permissions(path, fs::Permissions::from_mode(0o500))?;
+        Ok(Self(path.to_path_buf()))
+    }
+}
+#[cfg(unix)]
+impl Drop for ReadOnlyFixtureDir {
+    fn drop(&mut self) {
+        use std::os::unix::fs::PermissionsExt;
+        let _ = fs::set_permissions(&self.0, fs::Permissions::from_mode(0o700));
+    }
+}
+
+#[cfg(unix)]
+#[tokio::test]
+async fn failed_running_persistence_leaves_queued_work_without_executor_poll() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let calls = Arc::new(AtomicUsize::new(0));
+    let owner = TaskManager::start_with_executor_in_scope(
+        config(root.path(), "A"),
+        Arc::new(CountExecutor(calls.clone())),
+        "A",
+    )
+    .await?;
+    stop_idle_workers(&owner).await;
+    let task = owner
+        .add_task(NewTaskRequest::from_prompt("claim write failure"))
+        .await?;
+    let blocked = ReadOnlyFixtureDir::new(&owner.tasks_dir)?;
+    assert!(owner.claim_next_task().await.is_err());
+    assert_eq!(calls.load(Ordering::SeqCst), 0);
+    assert_eq!(owner.get_task(&task.id).await?.status, TaskStatus::Queued);
+    drop(blocked);
+    let (id, request, cancel) = owner.claim_next_task().await?.context("retry claim")?;
+    owner.run_task(id, request, cancel).await;
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+    owner.shutdown_and_wait().await?;
+    Ok(())
+}
+
+#[cfg(unix)]
+#[tokio::test]
+async fn failed_event_persistence_is_bounded_and_retains_each_accepted_delta_once() -> Result<()> {
+    let root = tempfile::tempdir()?;
+    let owner = manager(root.path(), "A").await?;
+    stop_idle_workers(&owner).await;
+    let task = owner
+        .add_task(NewTaskRequest::from_prompt("bounded write failure"))
+        .await?;
+    let (_, request, cancel) = owner.claim_next_task().await?.context("claim")?;
+    let blocked = ReadOnlyFixtureDir::new(&owner.tasks_dir)?;
+    for seq in 0..TASK_EVENT_CHANNEL_CAPACITY {
+        owner
+            .apply_execution_event(
+                &task.id,
+                TaskExecutionEvent::RuntimeEvent {
+                    seq: seq as u64,
+                    event: "fixture".into(),
+                    summary: seq.to_string(),
+                },
+            )
+            .await?;
+    }
+    let rejected = TaskExecutionEvent::RuntimeEvent {
+        seq: 999,
+        event: "fixture".into(),
+        summary: "after recovery".into(),
+    };
+    assert!(
+        owner
+            .apply_execution_event(&task.id, rejected.clone())
+            .await
+            .is_err()
+    );
+    assert_eq!(
+        owner.state.lock().await.pending_events[&task.id].len(),
+        TASK_EVENT_CHANNEL_CAPACITY
+    );
+    assert!(
+        cancel.is_cancelled(),
+        "failed persistence requests actual-owner cancellation"
+    );
+    drop(blocked);
+    owner.apply_execution_event(&task.id, rejected).await?;
+    owner.flush_task(&task.id).await?;
+    assert_eq!(
+        owner.get_task(&task.id).await?.runtime_event_count,
+        TASK_EVENT_CHANNEL_CAPACITY + 1
+    );
+    assert!(
+        !owner
+            .state
+            .lock()
+            .await
+            .pending_events
+            .contains_key(&task.id)
+    );
+    owner
+        .finish_task(
+            &task.id,
+            TaskExecutionResult::from_reason(TaskTerminalReason::Canceled, None),
+            cancel,
+            &request.mode_label,
+        )
+        .await?;
+    owner.shutdown_and_wait().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn legacy_queued_and_running_records_remain_visible_unverified_and_byte_identical()
+-> Result<()> {
+    let root = tempfile::tempdir()?;
+    let owner = manager(root.path(), "A").await?;
+    stop_idle_workers(&owner).await;
+    let first = owner
+        .add_task(NewTaskRequest {
+            owner_session_id: Some("legacy-owner".into()),
+            ..NewTaskRequest::from_prompt("legacy running")
+        })
+        .await?;
+    owner.claim_next_task().await?.context("claim")?;
+    let second = owner
+        .add_task(NewTaskRequest::from_prompt("legacy queued"))
+        .await?;
+    let mut snapshots = Vec::new();
+    for id in [&first.id, &second.id] {
+        let path = owner.tasks_dir.join(format!("{id}.json"));
+        let mut legacy: Value = serde_json::from_slice(&fs::read(&path)?)?;
+        let object = legacy.as_object_mut().unwrap();
+        object.remove("execution_scope");
+        object.remove("execution_generation");
+        object.remove("cancel_requested_seq");
+        object.insert("schema_version".into(), Value::from(3));
+        let bytes = serde_json::to_vec(&legacy)?;
+        fs::write(&path, &bytes)?;
+        snapshots.push((path, bytes));
+    }
+    let other = manager(root.path(), "B").await?;
+    sleep(STORE_REFRESH_INTERVAL * 2).await;
+    let rows = other.list_tasks(None).await?;
+    assert_eq!(rows.len(), 2);
+    assert!(rows.iter().all(|row| !row.execution_binding_known));
+    for (path, bytes) in snapshots {
+        assert_eq!(fs::read(path)?, bytes);
+    }
+    assert_eq!(
+        other
+            .get_task_for_owner(&first.id, "legacy-owner")
+            .await?
+            .status,
+        TaskStatus::Running
+    );
+    assert!(executions(root.path(), "B").is_empty());
+    owner.shutdown_and_wait().await?;
+    other.shutdown_and_wait().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn unreadable_owner_before_first_poll_waits_for_storage_without_executing() -> Result<()> {
+    for already_canceled in [false, true] {
+        let root = tempfile::tempdir()?;
+        let calls = Arc::new(AtomicUsize::new(0));
+        let owner = TaskManager::start_with_executor_in_scope(
+            config(root.path(), "A"),
+            Arc::new(CountExecutor(calls.clone())),
+            "A",
+        )
+        .await?;
+        stop_idle_workers(&owner).await;
+        let task = owner
+            .add_task(NewTaskRequest::from_prompt("unavailable before poll"))
+            .await?;
+        let (id, request, cancel) = owner.claim_next_task().await?.context("claim")?;
+        if already_canceled {
+            cancel.cancel();
+        }
+        let path = owner.tasks_dir.join(format!("{}.json", task.id));
+        let stored = fs::read(&path)?;
+        fs::write(&path, b"{corrupt")?;
+        let running = tokio::spawn({
+            let owner = owner.clone();
+            async move { owner.run_task(id, request, cancel).await }
+        });
+        sleep(STORE_REFRESH_INTERVAL * 2).await;
+        assert_eq!(calls.load(Ordering::SeqCst), 0);
+        assert!(
+            !running.is_finished(),
+            "unpersisted terminal state is still pending recovery"
+        );
+        fs::write(&path, stored)?;
+        tokio::time::timeout(Duration::from_secs(5), running).await??;
+        assert_eq!(owner.get_task(&task.id).await?.status, TaskStatus::Canceled);
+        assert_eq!(calls.load(Ordering::SeqCst), 0);
+        owner.shutdown_and_wait().await?;
+    }
+    Ok(())
+}

crates/tui/src/tools/send_later.rs

@@ -469,7 +469,7 @@ mod tests {
     }
 
     fn make_context_for_session(tmp: &TempDir, session_id: &str) -> ToolContext {
-        let manager = AutomationManager::open(tmp.path().to_path_buf()).unwrap();
+        let manager = AutomationManager::open_for_test(tmp.path().to_path_buf()).unwrap();
         let shared = Arc::new(Mutex::new(manager));
         ToolContext::new(".")
             .with_state_namespace(session_id)
@@ -875,7 +875,7 @@ mod tests {
             .unwrap()
             .to_string();
 
-        let manager = AutomationManager::open(tmp.path().to_path_buf()).unwrap();
+        let manager = AutomationManager::open_for_test(tmp.path().to_path_buf()).unwrap();
         let mut legacy = manager
             .create_trigger(crate::automation_manager::CreateDelayedTriggerRequest {
                 fire_at: chrono::Utc::now() + Duration::hours(1),
@@ -952,7 +952,7 @@ mod tests {
     #[tokio::test]
     async fn collect_due_triggers_returns_past_pending() {
         let tmp = TempDir::new().unwrap();
-        let manager = AutomationManager::open(tmp.path().to_path_buf()).unwrap();
+        let manager = AutomationManager::open_for_test(tmp.path().to_path_buf()).unwrap();
 
         // Create a trigger with fire_at one hour from now — not due yet.
         let req = crate::automation_manager::CreateDelayedTriggerRequest {

crates/tui/src/tools/tasks.rs

@@ -538,7 +538,8 @@ impl TasksTool {
         let limit = optional_u64(input, "limit", 20)?.clamp(1, 100) as usize;
         let tasks = manager
             .list_tasks_for_owner(Some(limit), None, &context.state_namespace)
-            .await;
+            .await
+            .map_err(|error| ToolError::execution_failed(error.to_string()))?;
         ToolResult::json(&json!({
             "summary": format!("{} durable task(s)", tasks.len()),
             "tasks": tasks,
@@ -1084,6 +1085,7 @@ fn task_result_with_lifecycle_warning(
         "summary": format!("{label}: {} ({:?})", task.id, task.status),
         "task": task,
         "lifecycle_warning": lifecycle_warning,
+        "execution_ownership": if task.execution_scope.is_some() { "scope_bound" } else { "unverified" },
     }))
     .map_err(|e| ToolError::execution_failed(e.to_string()))
 }

crates/tui/src/tui/app.rs

@@ -2334,6 +2334,8 @@ pub struct App {
     pub memory_size_hint: Option<String>,
     /// Cached background tasks for sidebar rendering.
     pub task_panel: Vec<TaskPanelEntry>,
+    pub task_panel_session_id: Option<String>,
+    pub task_panel_unavailable: bool,
     /// Live scheduled-work projection for the activity band
     /// (AUTOMATION-VISIBILITY-SPEC §2.1), refreshed on the task-panel cadence
     /// by `refresh_automation_panel`. The band reads it;

crates/tui/src/tui/app/init.rs

@@ -1082,6 +1082,8 @@ impl App {
             workspace_context_refreshed_at: None,
             memory_size_hint: None,
             task_panel: Vec::new(),
+            task_panel_session_id: None,
+            task_panel_unavailable: false,
             automation_panel: crate::tui::automation_panel::AutomationPanelState::default(),
             automation_scan: None,
             behavioral_tips: crate::tui::behavioral_tips::BehavioralTipState::new(

crates/tui/src/tui/automation_panel.rs

@@ -310,6 +310,7 @@ mod tests {
         let now = Utc::now();
         AutomationRecord {
             schema_version: 1,
+            execution_scope: Some(crate::task_manager::test_execution_scope("test")),
             id: id.to_string(),
             name: id.to_string(),
             prompt: "prompt".to_string(),

crates/tui/src/tui/automation_routing.rs

@@ -544,6 +544,7 @@ mod tests {
         let now = Utc::now();
         AutomationRecord {
             schema_version: 1,
+            execution_scope: Some(crate::task_manager::test_execution_scope("test")),
             id: "auto_1".to_string(),
             name: "Nightly checks".to_string(),
             prompt: "Run checks".to_string(),
@@ -715,7 +716,7 @@ mod tests {
     #[tokio::test]
     async fn pause_and_resume_emit_typed_mutation_receipts() {
         let temp = TempDir::new().expect("temp dir");
-        let manager = AutomationManager::open(temp.path().to_path_buf()).expect("manager");
+        let manager = AutomationManager::open_for_test(temp.path().to_path_buf()).expect("manager");
         let automation = manager
             .create_automation(CreateAutomationRequest {
                 name: "Nightly checks".to_string(),
@@ -764,7 +765,7 @@ mod tests {
     #[tokio::test]
     async fn delete_is_a_noop_until_snapshot_confirmation_then_removes_definition_and_runs() {
         let temp = TempDir::new().expect("temp dir");
-        let manager = AutomationManager::open(temp.path().to_path_buf()).expect("manager");
+        let manager = AutomationManager::open_for_test(temp.path().to_path_buf()).expect("manager");
         let automation = manager
             .create_automation(CreateAutomationRequest {
                 name: "Nightly checks".to_string(),

crates/tui/src/tui/subagent_routing.rs

@@ -1126,6 +1126,7 @@ mod tests {
 
     fn task_summary(id: &str, status: TaskStatus, duration_ms: Option<u64>) -> TaskSummary {
         TaskSummary {
+            execution_binding_known: true,
             id: id.to_string(),
             status,
             prompt_summary: "Fix task list output".to_string(),

crates/tui/src/tui/ui/apply.rs

@@ -2298,11 +2298,18 @@ pub(crate) async fn apply_command_result(
                             .list_tasks_for_owner(Some(30), None, session_id)
                             .await
                     }
-                    None => Vec::new(),
+                    None => Ok(Vec::new()),
                 };
                 refresh_active_task_panel(app, task_manager).await;
                 app.add_message(HistoryCell::System {
-                    content: format_task_list(&tasks),
+                    content: match tasks {
+                        Ok(tasks) => format_task_list(&tasks),
+                        Err(_) => crate::localization::tr(
+                            app.ui_locale,
+                            crate::localization::MessageId::TaskInventoryUnavailable,
+                        )
+                        .to_string(),
+                    },
                 });
             }
             AppAction::RemoteControl(action) => match action {

crates/tui/src/tui/ui/event_loop.rs

@@ -915,9 +915,10 @@ pub async fn run_tui(
         &session_id,
     )
     .await?;
-    let automations = std::sync::Arc::new(tokio::sync::Mutex::new(
-        AutomationManager::default_location()?,
-    ));
+    let _task_shutdown = task_manager.shutdown_guard();
+    let mut automation_service = AutomationManager::default_location()?;
+    automation_service.bind_task_manager(&task_manager)?;
+    let automations = std::sync::Arc::new(tokio::sync::Mutex::new(automation_service));
     let automation_cancel = tokio_util::sync::CancellationToken::new();
     let automation_scheduler = spawn_scheduler(
         automations.clone(),
@@ -1095,7 +1096,7 @@ pub async fn run_tui(
         &mut app,
         config,
         engine_handle,
-        task_manager,
+        task_manager.clone(),
         &event_broker,
         translation_client,
         pending_telemetry_notice,
@@ -1104,6 +1105,9 @@ pub async fn run_tui(
     .await;
     automation_cancel.cancel();
     automation_scheduler.abort();
+    if let Err(error) = task_manager.shutdown_and_wait().await {
+        tracing::error!(%error, "Task manager shutdown remains incomplete");
+    }
 
     // Join the startup-default writer before anything else tears down.
     //

crates/tui/src/tui/ui/task_projection.rs

@@ -9,14 +9,39 @@ pub(super) async fn refresh_active_task_panel(
     app: &mut App,
     task_manager: &SharedTaskManager,
 ) -> bool {
+    let namespace_changed = app.task_panel_session_id != app.current_session_id;
+    if namespace_changed {
+        app.task_panel.clear();
+        app.task_panel_session_id = app.current_session_id.clone();
+        app.task_panel_unavailable = false;
+    }
     let tasks = match app.current_session_id.as_deref() {
-        Some(session_id) => {
-            task_manager
-                .list_tasks_for_owner(None, None, session_id)
-                .await
-        }
+        Some(session_id) => match task_manager
+            .list_tasks_for_owner(None, None, session_id)
+            .await
+        {
+            Ok(tasks) => tasks,
+            Err(error) => {
+                let changed = namespace_changed || !app.task_panel_unavailable;
+                if !app.task_panel_unavailable {
+                    app.push_status_toast(
+                        crate::localization::tr(
+                            app.ui_locale,
+                            crate::localization::MessageId::TaskInventoryUnavailable,
+                        )
+                        .to_string(),
+                        crate::tui::app::StatusToastLevel::Warning,
+                        Some(8_000),
+                    );
+                    tracing::warn!(%error, "Task inventory unavailable; preserving scoped snapshot");
+                }
+                app.task_panel_unavailable = true;
+                return changed;
+            }
+        },
         None => Vec::new(),
     };
+    let was_unavailable = std::mem::replace(&mut app.task_panel_unavailable, false);
     let previously_active_durable_ids = app
         .task_panel
         .iter()
@@ -36,6 +61,9 @@ pub(super) async fn refresh_active_task_panel(
         app.current_session_id.as_deref(),
     ) {
         for task in &tasks {
+            if !task.execution_binding_known {
+                continue;
+            }
             let external = format!("task:{}", task.id);
             if !work.has_operation_binding(Some(session_id), &external) {
                 continue;
@@ -65,7 +93,22 @@ pub(super) async fn refresh_active_task_panel(
     let mut entries: Vec<TaskPanelEntry> =
         select_work_sidebar_tasks(tasks, session_started_at, app.current_session_id.as_deref())
             .into_iter()
-            .map(task_summary_to_panel_entry)
+            .map(|summary| {
+                let unverified = !summary.execution_binding_known
+                    && matches!(summary.status, TaskStatus::Queued | TaskStatus::Running);
+                let mut entry = task_summary_to_panel_entry(summary);
+                if unverified {
+                    entry.stale = true;
+                    entry.role = Some(
+                        crate::localization::tr(
+                            app.ui_locale,
+                            crate::localization::MessageId::TaskOwnershipUnverified,
+                        )
+                        .to_string(),
+                    );
+                }
+                entry
+            })
             .collect();
 
     entries.extend(active_rlm_task_entries(app));
@@ -136,7 +179,8 @@ pub(super) async fn refresh_active_task_panel(
     // Report whether anything visible changed so the idle tick can skip the
     // redraw: an unconditional 2.5 s repaint kept the app from ever going
     // quiescent (#3757).
-    let changed = lifecycle_changed || app.task_panel != entries;
+    let changed =
+        namespace_changed || was_unavailable || lifecycle_changed || app.task_panel != entries;
     app.task_panel = entries;
     let tip_shown = (durable_background_completed || shell_background_completed)
         && app.maybe_show_behavioral_tip(

crates/tui/src/tui/ui/tests.rs

@@ -8383,9 +8383,10 @@ async fn operate_entry_preserves_live_custom_identity_and_auto_over_startup_rout
         );
         let mut config = operate_tui_route_config();
         config.fleet_operator_route_applied = true;
-        let manager =
-            crate::automation_manager::AutomationManager::open(root.path().join("automations"))
-                .expect("manager");
+        let manager = crate::automation_manager::AutomationManager::open_for_test(
+            root.path().join("automations"),
+        )
+        .expect("manager");
         let automations = Arc::new(tokio::sync::Mutex::new(manager));
         let mut app = create_test_app();
         app.workspace = root.path().to_path_buf();
@@ -8448,9 +8449,10 @@ async fn operate_rejected_attach_does_not_reactivate_saved_keepalive() {
     let operate_dir = root.path().join("operate");
     let _operate = crate::test_support::EnvVarGuard::set("CODEWHALE_OPERATE_DIR", &operate_dir);
     let config = operate_tui_route_config();
-    let manager =
-        crate::automation_manager::AutomationManager::open(root.path().join("automations"))
-            .expect("manager");
+    let manager = crate::automation_manager::AutomationManager::open_for_test(
+        root.path().join("automations"),
+    )
+    .expect("manager");
     crate::operate::upsert_keepalive(&manager, root.path(), true, &config, None).expect("upsert");
     crate::operate::pause_keepalive(&manager).expect("pause");
     let before = serde_json::to_value(
@@ -8521,7 +8523,7 @@ async fn operate_mode_entry_attaches_to_recorded_operation() {
 
     let mut app = create_test_app();
     app.runtime_services.automations = Some(Arc::new(tokio::sync::Mutex::new(
-        crate::automation_manager::AutomationManager::open(dir.path().join("automations"))
+        crate::automation_manager::AutomationManager::open_for_test(dir.path().join("automations"))
             .expect("automations"),
     )));
     let engine = crate::core::engine::mock_engine_handle();
@@ -23631,6 +23633,7 @@ mod work_sidebar_projection_tests {
         owner_session_id: Option<&str>,
     ) -> TaskSummary {
         TaskSummary {
+            execution_binding_known: true,
             id: id.to_string(),
             status,
             prompt_summary: format!("task {id}"),
@@ -26213,3 +26216,75 @@ fn stale_session_projection_cannot_rewind_a_host_launch() {
     );
     assert_eq!(app.input, "typing remains responsive");
 }
+
+#[tokio::test]
+async fn task_inventory_failure_preserves_only_the_same_session_snapshot() -> anyhow::Result<()> {
+    use crate::task_manager::{
+        NewTaskRequest, TaskExecutionResult, TaskManager, TaskManagerConfig, TaskStatus,
+        TaskTerminalReason,
+    };
+    struct Done;
+    #[async_trait::async_trait]
+    impl crate::task_manager::TaskExecutor for Done {
+        async fn execute(
+            &self,
+            _: crate::task_manager::ExecutionTask,
+            _: tokio::sync::mpsc::Sender<crate::task_manager::TaskExecutionEvent>,
+            _: tokio_util::sync::CancellationToken,
+        ) -> TaskExecutionResult {
+            TaskExecutionResult {
+                status: TaskStatus::Completed,
+                result_text: Some("fixture".into()),
+                error: None,
+                terminal_reason: TaskTerminalReason::Completed,
+            }
+        }
+    }
+    let root = TempDir::new()?;
+    let tasks = TaskManager::start_with_executor(
+        TaskManagerConfig {
+            data_dir: root.path().into(),
+            worker_count: 1,
+            default_workspace: root.path().into(),
+            default_model: "fixture".into(),
+            default_mode: "plan".into(),
+            allow_shell: false,
+            trust_mode: false,
+            execution_limits: crate::task_manager::TaskExecutionLimits::default(),
+        },
+        Arc::new(Done),
+    )
+    .await?;
+    let record = tasks
+        .add_task(NewTaskRequest {
+            owner_session_id: Some("session-a".into()),
+            ..NewTaskRequest::from_prompt("session A receipt")
+        })
+        .await?;
+    crate::task_manager::wait_for_terminal_state(&tasks, &record.id, Duration::from_secs(5))
+        .await?;
+    let mut app = create_test_app();
+    app.current_session_id = Some("session-a".into());
+    app.session_started_at = chrono::Utc::now() - chrono::Duration::minutes(1);
+    super::task_projection::refresh_active_task_panel(&mut app, &tasks).await;
+    assert!(app.task_panel.iter().any(|row| row.id == record.id));
+    let queue = root.path().join("queue.json");
+    let saved = std::fs::read(&queue)?;
+    std::fs::write(&queue, b"{corrupt")?;
+    assert!(super::task_projection::refresh_active_task_panel(&mut app, &tasks).await);
+    assert!(app.task_panel_unavailable);
+    assert!(app.task_panel.iter().any(|row| row.id == record.id));
+    app.current_session_id = Some("session-b".into());
+    assert!(super::task_projection::refresh_active_task_panel(&mut app, &tasks).await);
+    assert!(app.task_panel_unavailable);
+    assert!(
+        app.task_panel.is_empty(),
+        "session A's stale rows cannot cross into B"
+    );
+    std::fs::write(queue, saved)?;
+    super::task_projection::refresh_active_task_panel(&mut app, &tasks).await;
+    assert!(!app.task_panel_unavailable);
+    assert!(app.task_panel.is_empty());
+    tasks.shutdown_and_wait().await?;
+    Ok(())
+}

crates/tui/src/tui/views/automations.rs

@@ -816,6 +816,7 @@ mod tests {
         let at = Utc.with_ymd_and_hms(2026, 9, 4, 17, 0, 0).unwrap();
         AutomationRecord {
             schema_version: 1,
+            execution_scope: Some(crate::task_manager::test_execution_scope("test")),
             id: id.to_string(),
             name: format!("cwc daily {id}"),
             prompt: "patrol".to_string(),
@@ -939,7 +940,8 @@ mod tests {
         let _env = crate::test_support::lock_test_env();
         let root = tempfile::tempdir().unwrap();
         let manager = std::sync::Arc::new(tokio::sync::Mutex::new(
-            crate::automation_manager::AutomationManager::open(root.path().join("store")).unwrap(),
+            crate::automation_manager::AutomationManager::open_for_test(root.path().join("store"))
+                .unwrap(),
         ));
         let mut view = AutomationsView::from_rows(Vec::new(), Locale::En);
         view.workspace = root.path().to_path_buf();

crates/tui/src/tui/views/automations/editor.rs

@@ -1257,7 +1257,7 @@ mod tests {
     fn editor_preserves_unknown_schedule_permissions_models_and_unedited_workspace() {
         let _env = crate::test_support::lock_test_env();
         let root = tempfile::tempdir().unwrap();
-        let manager = AutomationManager::open(root.path().join("store")).unwrap();
+        let manager = AutomationManager::open_for_test(root.path().join("store")).unwrap();
         let mut draft = editor(root.path());
         draft.name.insert("original", false);
         draft.prompt.insert("line one\nline two", true);
@@ -1296,7 +1296,7 @@ mod tests {
     fn editor_rejects_invalid_and_conflicting_saves_without_mutation() {
         let _env = crate::test_support::lock_test_env();
         let root = tempfile::tempdir().unwrap();
-        let manager = AutomationManager::open(root.path().join("store")).unwrap();
+        let manager = AutomationManager::open_for_test(root.path().join("store")).unwrap();
         let mut draft = editor(root.path());
         assert!(draft.save(&manager).is_err());
         assert!(manager.list_automations().unwrap().is_empty());
@@ -1343,7 +1343,7 @@ mod tests {
     fn editor_presets_use_canonical_schedules_and_atomic_pause() {
         let _env = crate::test_support::lock_test_env();
         let root = tempfile::tempdir().unwrap();
-        let manager = AutomationManager::open(root.path().join("store")).unwrap();
+        let manager = AutomationManager::open_for_test(root.path().join("store")).unwrap();
         let mut draft = editor(root.path());
         draft.name.insert("scheduled", false);
         draft.prompt.insert("prompt", true);
@@ -1390,7 +1390,7 @@ model = "same-model"
 "#,
         )
         .unwrap();
-        let manager = AutomationManager::open(root.path().join("store")).unwrap();
+        let manager = AutomationManager::open_for_test(root.path().join("store")).unwrap();
         let mut draft = AutomationEditor::new(&config, root.path(), Locale::En, None);
         draft.name.insert("name", false);
         draft.prompt.insert("prompt", true);