fix(runtime): retain worker ownership and late completion receipts (CW-7414)

A child can finish after its Runtime parent is idle or after a newer turn
starts. Share the actual worker manager and hold one permanent lifecycle
receiver, capture the dispatch owner and original Runtime turn, and publish
one durable terminal item/event to that origin. Retain owner admission policy,
shared write claims, process leases and pending terminal deliveries through
shutdown; protect live children from eviction and preserve journal failures.

Expose captured pending-approval input on authenticated canonical snapshots
and approval.required events, preserving missing versus explicit null values.
Runtime-minted single-use decision IDs, namespaced execution IDs, force/deny
routing and redacted global/legacy history remain unchanged. Existing fixtures
use actual origin-aware registration; no new test functions or code comments.

Local frozen source at 68f4dd3406, reconciled onto 601009c8b9:
- Upstream's 11 contributor-history commits have an identical committed tree;
  all 4092 tracked source inputs and eleven owned paths remained byte-identical.
- cargo fmt --all -- --check: exit 0.
- Existing focused Rust filters: 25 commands, 29 passed; 0 failed; every
  selection positive. Literal lines include:
  test result: ok. 1 passed; 0 failed; 0 ignored
  test result: ok. 3 passed; 0 failed; 0 ignored
- Scoped locked all-target Clippy: exit 0, using exactly canonical CI's three
  existing style allowances (uninlined_format_args, too_many_arguments,
  unnecessary_map_or) after fixing the new registry type complexity. The prior
  unconfigured strict failure is retained; no new lint allowance or test skip.
- npm test: 547 tests; 540 passed; 0 failed; 7 existing skipped.
- npm run check:web: exit 0 (facts/docs/tokens, lint, types, production web build).
- Actual CLI build: exit 0; SHA256
  458aaa00373b748df978a6281ae1ee75e32e1342dfdfd66c6163e14f75d01889.

Separate actual-process lanes, each with independent native identity/port
cleanup and unchanged source/binary receipts:
- Engine + PostgreSQL Work bridge: 8 named checks passed; 0 failed. Two owners,
  parent-first completion, original-turn exactly-once Runtime/PG/live/replay
  SSE delivery, producer reconnect, newer active turn and durable histories.
- Engine owner admission/lease/restart: 9 named checks passed; 0 failed. Real
  route refusal, independent-owner admission, both exclusive lease files,
  same-profile history recovery, current token 200/old token 401, no model replay.
- Engine scoped writers: 11 named checks passed; 0 failed. Normal three parent
  allow-once approvals, live overlapping-claim refusal, disjoint actual writes,
  exact nonce/readback hashes, paired durable ToolUse/ToolResult identities,
  captured Custom delegation and terminal-only release of claims.

All process model transport was scripted loopback. Earlier failed harness
attempts remain failed, including identity-domain/cursor/public-route/approval
note assumptions; they were corrected without relaxing product assertions.
Muse Wonder Full/max informed the earlier diagnosis; later timed-out requests
supplied no additional opinion. Dev linker oversized __eh_frame warning is
retained. This is source/local/process readiness, not hosted CI, a released
Engine, VM/provider isolation, paid billing, signing, deployment or production.

crates/tui/src/core/engine.rs

@@ -53,9 +53,9 @@ use crate::tools::spec::{
 };
 use crate::tools::subagent::{
     FleetRole, ForegroundChildRegistry, Mailbox, MailboxMessage, SharedSubAgentManager,
-    SubAgentCompletion, SubAgentForkContext, SubAgentManager, SubAgentResult, SubAgentRuntime,
-    SubAgentStatus, agent_worker_owner_snapshot,
-    new_shared_subagent_manager_with_state_root_and_timeout,
+    SubAgentCompletion, SubAgentExecutionLimits, SubAgentForkContext, SubAgentManager,
+    SubAgentResult, SubAgentRuntime, SubAgentStatus, TerminalEventDelivery,
+    agent_worker_owner_snapshot, new_shared_subagent_manager_with_state_root_and_timeout,
 };
 use crate::tools::todo::{SharedTodoList, new_shared_todo_list};
 use crate::tools::user_input::{UserInputRequest, UserInputResponse};
@@ -762,6 +762,7 @@ pub struct EngineHandle {
     /// straight to it: the engine may be streaming or running tools and not
     /// reading approvals for a long time.
     subagent_manager: SharedSubAgentManager,
+    terminal_worker_events: Option<Arc<TerminalEventDelivery>>,
 }
 
 const MAX_PENDING_COMPACTION_CANCELLATIONS: usize = 64;
@@ -814,6 +815,35 @@ impl CompactionCancellationState {
 }
 
 impl EngineHandle {
+    #[must_use]
+    pub(crate) fn subagent_manager(&self) -> SharedSubAgentManager {
+        self.subagent_manager.clone()
+    }
+
+    pub(crate) fn mark_worker_event_delivery_failed(&self) {
+        if let Some(events) = self.terminal_worker_events.as_ref() {
+            events.mark_failed();
+        }
+    }
+
+    pub(crate) async fn wait_terminal_worker_events(&self) -> Result<()> {
+        if let Some(events) = self.terminal_worker_events.as_ref() {
+            tokio::time::timeout(FOREGROUND_CHILD_SETTLE_GRACE, events.wait())
+                .await
+                .map_err(|_| {
+                    anyhow::anyhow!(
+                        "terminal worker event delivery timed out during Runtime shutdown"
+                    )
+                })??;
+        }
+        Ok(())
+    }
+
+    #[cfg(test)]
+    pub(crate) fn set_subagent_manager(&mut self, manager: SharedSubAgentManager) {
+        self.subagent_manager = manager;
+    }
+
     /// Publish typed compaction cancellation immediately, then enqueue the
     /// matching operation when capacity permits. The shared authority is what
     /// stops a running provider future; the operation keeps the mailbox
@@ -943,6 +973,7 @@ pub struct Engine {
     /// state stays alive across model turns.
     repl_kernel: Option<crate::repl::PythonRuntime>,
     subagent_manager: SharedSubAgentManager,
+    subagent_execution_limits: Option<SubAgentExecutionLimits>,
     /// The deterministic Auto-Review policy shared with every child runtime
     /// so children are gated by the same rules as the parent turn.
     shared_auto_review_policy: Arc<crate::core::authority::auto_review::AutoReviewPolicy>,
@@ -1874,9 +1905,18 @@ impl Engine {
     }
 
     fn new_admitted(
+        config: EngineConfig,
+        api_config: &Config,
+        host: Option<EngineHostSetup>,
+    ) -> (Self, EngineHandle) {
+        Self::new_admitted_with_subagent_manager(config, api_config, host, None)
+    }
+
+    fn new_admitted_with_subagent_manager(
         mut config: EngineConfig,
         api_config: &Config,
         mut host: Option<EngineHostSetup>,
+        supplied_subagent_manager: Option<SharedSubAgentManager>,
     ) -> (Self, EngineHandle) {
         crate::tls::ensure_rustls_crypto_provider();
 
@@ -2125,28 +2165,50 @@ impl Engine {
             codewhale_core::prefix_cache::PrefixStabilityManager::new_unpinned()
         });
 
-        let subagent_manager = host.as_ref().map_or_else(
-            || {
-                let subagent_state_root = config
-                    .subagent_state_root
-                    .clone()
-                    .unwrap_or_else(|| config.workspace.clone());
-                new_shared_subagent_manager_with_state_root_and_timeout(
-                    config.workspace.clone(),
-                    subagent_state_root,
+        let subagent_execution_limits = supplied_subagent_manager
+            .as_ref()
+            .map(|_| {
+                SubAgentExecutionLimits::new(
                     config.max_subagents,
                     config.max_admitted_subagents,
                     config.subagent_heartbeat_timeout,
                     config.launch_concurrency,
-                    // #5324: defaults are captured operator config, not call fields.
                     api_config.subagent_default_max_steps(),
                     api_config
                         .subagent_default_wall_time_secs()
-                        .map(std::time::Duration::from_secs),
+                        .map(Duration::from_secs),
                 )
-            },
-            |host| Arc::clone(host.subagent_manager()),
-        );
+            })
+            .or_else(|| match host.as_ref() {
+                Some(EngineHostSetup::Child(child)) => {
+                    child.authority.runtime.execution_limits.clone()
+                }
+                _ => None,
+            });
+        let subagent_manager = supplied_subagent_manager.unwrap_or_else(|| {
+            host.as_ref().map_or_else(
+                || {
+                    let subagent_state_root = config
+                        .subagent_state_root
+                        .clone()
+                        .unwrap_or_else(|| config.workspace.clone());
+                    new_shared_subagent_manager_with_state_root_and_timeout(
+                        config.workspace.clone(),
+                        subagent_state_root,
+                        config.max_subagents,
+                        config.max_admitted_subagents,
+                        config.subagent_heartbeat_timeout,
+                        config.launch_concurrency,
+                        // #5324: defaults are captured operator config, not call fields.
+                        api_config.subagent_default_max_steps(),
+                        api_config
+                            .subagent_default_wall_time_secs()
+                            .map(std::time::Duration::from_secs),
+                    )
+                },
+                |host| Arc::clone(host.subagent_manager()),
+            )
+        });
         // The OS wrappers below only cover child processes. Codewhale's own
         // `read_file`/`read`/`read_media` tools read in-process, so the same
         // deny-list is installed process-wide for them to consult (S1).
@@ -2263,6 +2325,7 @@ impl Engine {
             session,
             repl_kernel: None,
             subagent_manager,
+            subagent_execution_limits,
             shared_auto_review_policy,
             shell_manager,
             file_read_tracker,
@@ -2342,6 +2405,10 @@ impl Engine {
             compaction_cancellation,
             turn_heartbeat: Arc::clone(&engine.turn_heartbeat),
             subagent_manager: Arc::clone(&engine.subagent_manager),
+            terminal_worker_events: engine
+                .subagent_execution_limits
+                .as_ref()
+                .map(|limits| limits.terminal_events.clone()),
         };
 
         (engine, handle)
@@ -2351,6 +2418,7 @@ impl Engine {
     /// client. The event loop, prompt assembly, tool registry/execution,
     /// cancellation, and session projection are unchanged; only the model I/O
     /// boundary is replaced.
+    #[cfg(test)]
     pub fn new_with_model_client(
         config: EngineConfig,
         api_config: &Config,
@@ -3837,15 +3905,24 @@ impl Engine {
                         self.config.subagent_api_timeout = Duration::from_secs(api_timeout_secs);
                         self.config.subagent_heartbeat_timeout =
                             Duration::from_secs(heartbeat_timeout_secs);
-                        let launch_gate_applied = {
-                            let mut manager = self.subagent_manager.write().await;
-                            manager.update_runtime_limits(
-                                self.config.max_subagents,
-                                self.config.max_admitted_subagents,
-                                self.config.subagent_heartbeat_timeout,
-                                self.config.launch_concurrency,
-                            )
-                        };
+                        let launch_gate_applied =
+                            if let Some(limits) = self.subagent_execution_limits.as_mut() {
+                                limits.update(
+                                    self.config.max_subagents,
+                                    self.config.max_admitted_subagents,
+                                    self.config.subagent_heartbeat_timeout,
+                                    self.config.launch_concurrency,
+                                );
+                                true
+                            } else {
+                                let mut manager = self.subagent_manager.write().await;
+                                manager.update_runtime_limits(
+                                    self.config.max_subagents,
+                                    self.config.max_admitted_subagents,
+                                    self.config.subagent_heartbeat_timeout,
+                                    self.config.launch_concurrency,
+                                )
+                            };
                         let launch_note = if launch_gate_applied {
                             ""
                         } else {
@@ -5659,6 +5736,7 @@ impl Engine {
                 .with_locale_tag(route.locale_tag.clone())
                 .with_role_models(route.role_models.clone())
                 .with_api_config((*route.api_config).clone())
+                .with_execution_limits(self.subagent_execution_limits.clone())
                 .with_auto_model(route.auto_model)
                 .with_reasoning_effort(route.reasoning_effort.clone(), route.reasoning_effort_auto)
                 .with_agent_tool_surface_options(
@@ -7127,6 +7205,7 @@ impl Engine {
         .with_locale_tag(self.config.locale_tag.clone())
         .with_role_models(self.subagent_role_models())
         .with_api_config(self.api_config.clone())
+        .with_execution_limits(self.subagent_execution_limits.clone())
         .with_auto_model(self.session.auto_model)
         .with_reasoning_effort(
             self.session.reasoning_effort.clone(),
@@ -9343,6 +9422,8 @@ pub(crate) fn spawn_engine_with_authoritative_route_config(
     host_profile: EngineHostProfile,
     authoritative_route_config: Arc<parking_lot::RwLock<Config>>,
     model_client: Option<SharedModelClient>,
+    subagent_manager: Option<SharedSubAgentManager>,
+    execution_admission: CancellationToken,
 ) -> (EngineHandle, tokio::task::JoinHandle<()>) {
     // `model_client` replaces only the model I/O boundary (see
     // `Engine::new_with_model_client`); hosts pass `None` for the provider
@@ -9352,10 +9433,17 @@ pub(crate) fn spawn_engine_with_authoritative_route_config(
         config.goal_max_steps = None;
         config.subagents_enabled = false;
     }
-    let (mut engine, handle) = match model_client {
-        Some(client) => Engine::new_with_model_client(config, api_config, client),
-        None => Engine::new(config, api_config),
-    };
+    let (mut engine, mut handle) =
+        Engine::new_admitted_with_subagent_manager(config, api_config, None, subagent_manager);
+    if let Some(limits) = engine.subagent_execution_limits.as_mut() {
+        limits.admission = execution_admission;
+    }
+    if let Some(client) = model_client {
+        engine.model_client = Some(client);
+        engine.model_client_injected = true;
+        engine.codewhale_client_error = None;
+        handle.client_preflight_required = false;
+    }
     engine.host_profile = host_profile;
     engine.authoritative_route_config = Some(authoritative_route_config);
 
@@ -9520,6 +9608,7 @@ pub(crate) fn mock_engine_handle() -> MockEngineHandle {
             std::env::temp_dir(),
             1,
         ),
+        terminal_worker_events: Some(Arc::new(TerminalEventDelivery::default())),
     };
 
     MockEngineHandle {

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

@@ -892,6 +892,7 @@ fn engine_handle_try_send_does_not_block_when_op_channel_is_full() {
             std::env::temp_dir(),
             1,
         ),
+        terminal_worker_events: None,
     };
 
     // Fill the op channel with one message (capacity = 1).

crates/tui/src/runtime_api.rs

@@ -88,7 +88,7 @@ use crate::task_manager::{
 };
 use crate::tools::subagent::{
     AgentWorkerRecord, AgentWorkerStatus, SharedSubAgentManager, SubAgentStatus,
-    new_shared_subagent_manager_with_timeout,
+    new_shared_runtime_subagent_manager,
 };
 #[cfg(test)]
 pub(super) use codewhale_models::{ContentBlock, Message};
@@ -1369,15 +1369,11 @@ fn default_runtime_capabilities() -> RuntimeCapabilities {
     }
 }
 
-fn runtime_api_sub_agent_manager(workspace: &FsPath, workers: usize) -> SharedSubAgentManager {
-    let max_agents = workers.max(1);
-    new_shared_subagent_manager_with_timeout(
-        workspace.to_path_buf(),
-        max_agents,
-        max_agents,
-        Duration::from_secs(crate::config::DEFAULT_SUBAGENT_HEARTBEAT_TIMEOUT_SECS),
-        max_agents,
-    )
+fn runtime_api_sub_agent_manager(
+    workspace: &FsPath,
+    workers: usize,
+) -> Result<SharedSubAgentManager> {
+    new_shared_runtime_subagent_manager(workspace.to_path_buf(), workers)
 }
 
 #[derive(Debug, Serialize)]
@@ -1971,6 +1967,13 @@ pub async fn run_http_server(
         existing_binding.as_ref(),
     )?;
     let sessions_dir = runtime_threads.sessions_dir().to_path_buf();
+    let sub_agent_manager = runtime_api_sub_agent_manager(&workspace, options.workers)?;
+    runtime_threads
+        .attach_subagent_manager(sub_agent_manager.clone())
+        .await?;
+    let workspace_scopes =
+        RuntimeWorkspaceScopes::new(runtime_threads.clone(), sub_agent_manager.clone());
+    let workspace_scope = workspace_scopes.admit(workspace.clone()).await?;
     if let Some(tap) = options.runtime_event_tap {
         tap(runtime_threads.subscribe_events());
     }
@@ -2020,7 +2023,6 @@ pub async fn run_http_server(
     };
     let skill_state = SkillStateStore::load_default()
         .context("load persistent Skill activation state for Runtime API")?;
-    let sub_agent_manager = runtime_api_sub_agent_manager(&workspace, options.workers);
     let shutdown = RuntimeServerShutdown::default();
     // Opening a thread is every client's first read, and the store can only
     // answer it after one pass over the whole items directory (an item's
@@ -2033,9 +2035,7 @@ pub async fn run_http_server(
             tracing::warn!(%error, "thread item index warm-up failed");
         }
     });
-    let workspace_scopes =
-        RuntimeWorkspaceScopes::new(runtime_threads.clone(), sub_agent_manager.clone());
-    let workspace_scope = workspace_scopes.admit(workspace.clone()).await?;
+
     let state = RuntimeApiState {
         config: Arc::new(parking_lot::RwLock::new(config.clone())),
         workspace,
@@ -13184,7 +13184,10 @@ base_url = "http://127.0.0.1:9/v1"
         .await?;
         let listener = TcpListener::bind("127.0.0.1:0").await?;
         let addr = listener.local_addr()?;
-        let sub_agent_manager = runtime_api_sub_agent_manager(&workspace, 2);
+        let sub_agent_manager = runtime_api_sub_agent_manager(&workspace, 2)?;
+        runtime_threads
+            .attach_subagent_manager(sub_agent_manager.clone())
+            .await?;
         let workspace_scopes =
             RuntimeWorkspaceScopes::new(runtime_threads.clone(), sub_agent_manager.clone());
         let workspace_scope = workspace_scopes.admit(workspace.clone()).await?;

crates/tui/src/runtime_api/tests.rs

@@ -1288,9 +1288,13 @@ async fn build_test_server(
     }
 
     let auth_required = runtime_token.is_some();
-    let sub_agent_manager = overrides
-        .sub_agent_manager
-        .unwrap_or_else(|| runtime_api_sub_agent_manager(&workspace, 2));
+    let sub_agent_manager = match overrides.sub_agent_manager {
+        Some(manager) => manager,
+        None => runtime_api_sub_agent_manager(&workspace, 2)?,
+    };
+    runtime_threads
+        .attach_subagent_manager(sub_agent_manager.clone())
+        .await?;
     // A std listener: the server thread registers it with its own runtime
     // after setup is reported (see `spawn_product_stack_server`).
     let listener = match std::net::TcpListener::bind("127.0.0.1:0") {
@@ -2657,7 +2661,7 @@ async fn fleet_status_runtime_api_exposes_state_and_actions() -> Result<()> {
     let root = std::env::temp_dir().join(format!("codewhale-fleet-api-{}", Uuid::new_v4()));
     let workspace = root.join("workspace");
     fs::create_dir_all(&workspace)?;
-    let sub_agent_manager = runtime_api_sub_agent_manager(&workspace, 2);
+    let sub_agent_manager = runtime_api_sub_agent_manager(&workspace, 2)?;
     let manager = FleetManager::open(&workspace)?
         .with_sub_agent_manager(sub_agent_manager.clone())
         .with_session_model(DEFAULT_TEXT_MODEL)
@@ -2972,6 +2976,7 @@ async fn agent_runs_runtime_api_exposes_persisted_worker_receipts() -> Result<()
             launch_manifest: None,
         },
         owner_session_id: "session-receipt".to_string(),
+        runtime_origin_turn_id: None,
         actor_kind: "subagent".to_string(),
         parent_run_id: Some("parent_run".to_string()),
         follow_up: AgentRunFollowUpTarget {
@@ -3051,15 +3056,14 @@ async fn agent_runs_runtime_api_exposes_persisted_worker_receipts() -> Result<()
     )?;
 
     let sessions_dir = root.join("sessions");
-    let Some((addr, _runtime_threads, handle)) =
-        spawn_test_server_with_root_token_mobile_workspace(
-            root.clone(),
-            sessions_dir,
-            None,
-            false,
-            workspace,
-        )
-        .await?
+    let Some((addr, runtime_threads, handle)) = spawn_test_server_with_root_token_mobile_workspace(
+        root.clone(),
+        sessions_dir,
+        None,
+        false,
+        workspace.clone(),
+    )
+    .await?
     else {
         return Ok(());
     };
@@ -3105,6 +3109,60 @@ async fn agent_runs_runtime_api_exposes_persisted_worker_receipts() -> Result<()
         .status();
     assert_eq!(missing, StatusCode::NOT_FOUND);
 
+    assert!(run.get("runtime_origin_turn_id").is_none());
+    assert!(
+        serde_json::from_value::<AgentWorkerRecord>(run.clone())?
+            .runtime_origin_turn_id
+            .is_none()
+    );
+    let thread = runtime_threads
+        .create_thread(crate::runtime_threads::CreateThreadRequest::default())
+        .await?;
+    let engine = runtime_threads.get_engine(&thread.id).await?;
+    let actual = engine.subagent_manager();
+    let mut spec = serde_json::from_value::<AgentWorkerSpec>(run["spec"].clone())?;
+    spec.worker_id = "ordinary_receipt".into();
+    spec.run_id = "ordinary_run".into();
+    spec.parent_run_id = None;
+    actual
+        .write()
+        .await
+        .register_worker_for_session_with_origin(
+            spec,
+            &thread.id,
+            None,
+            Some("turn_original_receipt".into()),
+        );
+    let listed: serde_json::Value = client
+        .get(format!("http://{addr}/v1/agent-runs"))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    let records = listed["runs"].as_array().unwrap();
+    assert!(
+        records
+            .iter()
+            .any(|record| record["spec"]["worker_id"] == "agent_receipt")
+    );
+    assert_eq!(
+        records
+            .iter()
+            .filter(|record| record["spec"]["worker_id"] == "ordinary_receipt")
+            .count(),
+        1
+    );
+    let ordinary: serde_json::Value = client
+        .get(format!("http://{addr}/v1/agent-runs/ordinary_run"))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(ordinary["owner_session_id"], thread.id);
+    assert_eq!(ordinary["runtime_origin_turn_id"], "turn_original_receipt");
+    runtime_threads.shutdown_and_wait().await?;
     handle.abort();
     Ok(())
 }

crates/tui/src/runtime_api/tests/runtime_store_convergence.rs

@@ -777,7 +777,7 @@ async fn real_owner_two_workspace_services_share_scope_caches_and_settle_global_
     #[cfg(windows)]
     let socket = PathBuf::from(format!(r"\\.\pipe\scope-owner-{}", Uuid::new_v4()));
     let token = "cwrt_scope_operator_only";
-    let workers = runtime_api_sub_agent_manager(&original_workspace, 2);
+    let workers = runtime_api_sub_agent_manager(&original_workspace, 2)?;
     let (scope_tx, scope_rx) = oneshot::channel();
     let (original, manager, server) =
         spawn_test_server_with_root_token_mobile_workspace_and_overrides(

crates/tui/src/runtime_threads.rs

@@ -23,7 +23,7 @@ use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 use sha2::{Digest, Sha256};
-use tokio::sync::{Mutex, RwLock as AsyncRwLock, broadcast, mpsc, oneshot, watch};
+use tokio::sync::{Mutex, Notify, RwLock as AsyncRwLock, broadcast, mpsc, oneshot, watch};
 use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
@@ -50,7 +50,9 @@ use crate::route_runtime::{ResolvedRuntimeRoute, resolve_runtime_route_for_ident
 use crate::runtime_policy::RuntimePolicyProjection;
 use crate::tools::plan::new_shared_plan_state;
 use crate::tools::shell::{SharedShellManager, new_shared_shell_manager};
-use crate::tools::subagent::SubAgentStatus;
+use crate::tools::subagent::{
+    SharedSubAgentManager, SubAgentStatus, new_shared_runtime_subagent_manager,
+};
 use crate::tools::todo::new_shared_todo_list;
 #[cfg(test)]
 use codewhale_config::AppMode;
@@ -2370,6 +2372,11 @@ enum RuntimeEventMatch {
         message_id: String,
         attempt_count: u8,
     },
+    WorkerLifecycle {
+        turn_id: String,
+        agent_id: String,
+        event_name: String,
+    },
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -4372,6 +4379,16 @@ impl RuntimeThreadStore {
                     event.event == "turn.completed"
                         && event.turn_id.as_deref() == Some(turn_id.as_str())
                 }
+                RuntimeEventMatch::WorkerLifecycle {
+                    turn_id,
+                    agent_id,
+                    event_name,
+                } => {
+                    event.event == *event_name
+                        && event.turn_id.as_deref() == Some(turn_id.as_str())
+                        && event.payload.get("agent_id").and_then(Value::as_str)
+                            == Some(agent_id.as_str())
+                }
                 RuntimeEventMatch::DynamicTerminal { turn_id, call_id } => {
                     matches!(
                         event.event.as_str(),
@@ -5324,6 +5341,12 @@ pub struct ThreadDetail {
     pub approval_grants: Vec<RuntimeApprovalGrant>,
 }
 
+fn deserialize_present_approval_input<'de, D: serde::Deserializer<'de>>(
+    deserializer: D,
+) -> Result<Option<Value>, D::Error> {
+    Value::deserialize(deserializer).map(Some)
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PendingApprovalRequest {
     /// Runtime-owned, single-use approval ID. Clients echo this value; it is
@@ -5331,12 +5354,18 @@ pub struct PendingApprovalRequest {
     pub id: String,
     pub turn_id: String,
     pub tool_name: String,
+    #[serde(
+        default,
+        skip_serializing_if = "Option::is_none",
+        deserialize_with = "deserialize_present_approval_input"
+    )]
+    pub input: Option<Value>,
     pub description: String,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub intent_summary: Option<String>,
-    /// The provider tool-call ID this approval gates, mirroring the
+    /// The Engine tool-execution ID this approval gates, mirroring the
     /// `tool_call_id` on `approval.required`. A client resuming from a
-    /// snapshot needs it to attach the prompt to the tool row it belongs to.
+    /// snapshot needs it to attach the prompt to the execution it belongs to.
     /// It is a correlator, never a capability: `deliver_external_approval`
     /// matches `id` only, so this value settles nothing.
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -6222,6 +6251,323 @@ async fn await_retained_completion(completion: &RuntimeCompletion) -> Result<()>
 
 pub type SharedRuntimeThreadManager = Arc<RuntimeThreadManager>;
 
+fn push_runtime_notice(
+    notices: &parking_lot::Mutex<HashMap<String, Vec<ActiveNotice>>>,
+    thread_id: &str,
+    kind: &str,
+    turn_id: &str,
+    subject: &str,
+    detail: String,
+) -> String {
+    debug_assert!(
+        codewhale_config::notifications::NotificationEvent::parse(kind).is_some(),
+        "notice kind must be a NotificationEvent name: {kind}"
+    );
+    let id = runtime_record_id("notice");
+    let mut notices = notices.lock();
+    let list = notices.entry(thread_id.to_string()).or_default();
+    list.push(ActiveNotice {
+        id: id.clone(),
+        kind: kind.to_string(),
+        turn_id: turn_id.to_string(),
+        subject: subject.to_string(),
+        detail,
+        raised_at: Utc::now(),
+    });
+    if list.len() > MAX_NOTICES_PER_THREAD {
+        list.drain(..list.len() - MAX_NOTICES_PER_THREAD);
+    }
+    id
+}
+
+struct RuntimeWorkerEventSink {
+    thread_id: String,
+    workspace: PathBuf,
+    manager: SharedSubAgentManager,
+    store: RuntimeThreadStore,
+    projection: Arc<Mutex<()>>,
+    event_emit: Arc<Mutex<()>>,
+    event_tx: broadcast::Sender<RuntimeEventRecord>,
+    notices: Arc<parking_lot::Mutex<HashMap<String, Vec<ActiveNotice>>>>,
+    owner: std::sync::Weak<RuntimeProcessOwnerLock>,
+}
+
+impl RuntimeWorkerEventSink {
+    async fn publish(&self, event: EngineEvent) -> Result<Option<String>> {
+        let (owner, agent_id, event_name) = match &event {
+            EngineEvent::AgentSpawned {
+                owner_session_id,
+                id,
+                ..
+            } => (owner_session_id, id.clone(), "agent.spawned"),
+            EngineEvent::AgentProgress {
+                owner_session_id,
+                id,
+                ..
+            } => (owner_session_id, id.clone(), "agent.progress"),
+            EngineEvent::AgentComplete {
+                owner_session_id,
+                id,
+                ..
+            } => (owner_session_id, id.clone(), "agent.completed"),
+            _ => return Ok(None),
+        };
+        if owner != &self.thread_id {
+            return Ok(None);
+        }
+        let _owner = self
+            .owner
+            .upgrade()
+            .context("Runtime event owner has closed")?;
+        let record = self
+            .manager
+            .read()
+            .await
+            .worker_records_for_workspace(&self.workspace)
+            .map_err(anyhow::Error::msg)?
+            .into_iter()
+            .map(|(record, _)| record)
+            .find(|record| {
+                record.owner_session_id == self.thread_id && record.spec.worker_id == agent_id
+            });
+        let Some(record) = record else {
+            return Ok(None);
+        };
+        let Some(turn_id) = record.runtime_origin_turn_id else {
+            return Ok(None);
+        };
+        let thread_id = self.thread_id.clone();
+        let _projection = self.projection.lock().await;
+        let store = self.store.clone();
+        let origin = turn_id.clone();
+        let selected_thread = thread_id.clone();
+        let selected_workspace = self.workspace.clone();
+        let selected_agent = agent_id.clone();
+        let duplicate = tokio::task::spawn_blocking(move || -> Result<bool> {
+            let thread = store.load_thread(&selected_thread)?;
+            let turn = store.load_turn(&origin)?;
+            anyhow::ensure!(
+                turn.thread_id == selected_thread && thread.workspace == selected_workspace,
+                "child Runtime turn owner has changed"
+            );
+            if event_name == "agent.progress" {
+                return Ok(false);
+            }
+            store.contains_event(
+                &selected_thread,
+                &RuntimeEventMatch::WorkerLifecycle {
+                    turn_id: origin,
+                    agent_id: selected_agent,
+                    event_name: event_name.to_string(),
+                },
+            )
+        })
+        .await
+        .context("child receipt origin check failed")??;
+        if duplicate {
+            return Ok(Some(turn_id));
+        }
+        let item_id = if event_name == "agent.progress" {
+            runtime_record_id("item")
+        } else {
+            format!(
+                "item_{}",
+                crate::hashing::sha256_hex(format!("{turn_id}\0{agent_id}\0{event_name}"))
+            )
+        };
+        match event {
+            EngineEvent::AgentSpawned {
+                owner_session_id,
+                id,
+                prompt,
+                worker_status,
+                parent_run_id,
+                spawn_depth,
+                display_name,
+                ..
+            } if owner_session_id == thread_id => {
+                // Hosts name the agent the way the TUI does (#6565).
+                let name = display_name.as_deref().unwrap_or(&id);
+                let message = format!(
+                    "Sub-agent {name} spawned: {}",
+                    summarize_text(&prompt, SUMMARY_LIMIT)
+                );
+                let item = TurnItemRecord {
+                    schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
+                    id: item_id,
+                    turn_id: turn_id.clone(),
+                    kind: TurnItemKind::Status,
+                    status: TurnItemLifecycleStatus::Completed,
+                    summary: summarize_text(&message, SUMMARY_LIMIT),
+                    detail: Some(message),
+                    metadata: None,
+                    artifact_refs: Vec::new(),
+                    artifacts: Vec::new(),
+                    started_at: Some(Utc::now()),
+                    ended_at: Some(Utc::now()),
+                };
+                self.save_item(&item).await?;
+                self.emit_event(
+                    &thread_id,
+                    Some(&turn_id),
+                    Some(&item.id),
+                    "agent.spawned",
+                    json!({ "item": item, "agent_id": id, "agent_name": display_name,
+                            "worker_status": worker_status, "parent_run_id": parent_run_id,
+                            "spawn_depth": spawn_depth }),
+                )
+                .await?;
+            }
+            EngineEvent::AgentProgress {
+                owner_session_id,
+                id,
+                status,
+                activity,
+                parent_run_id,
+                spawn_depth,
+            } if owner_session_id == thread_id => {
+                let message = format!("Sub-agent {id}: {status}");
+                let item = TurnItemRecord {
+                    schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
+                    id: item_id,
+                    turn_id: turn_id.clone(),
+                    kind: TurnItemKind::Status,
+                    status: TurnItemLifecycleStatus::Completed,
+                    summary: summarize_text(&message, SUMMARY_LIMIT),
+                    detail: Some(message),
+                    metadata: None,
+                    artifact_refs: Vec::new(),
+                    artifacts: Vec::new(),
+                    started_at: Some(Utc::now()),
+                    ended_at: Some(Utc::now()),
+                };
+                self.save_item(&item).await?;
+                self.emit_event(
+                    &thread_id,
+                    Some(&turn_id),
+                    Some(&item.id),
+                    "agent.progress",
+                    json!({ "item": item, "agent_id": id,
+                            "worker_status": activity.worker_status, "step": activity.step,
+                            "parent_run_id": parent_run_id, "spawn_depth": spawn_depth }),
+                )
+                .await?;
+            }
+            EngineEvent::AgentComplete {
+                owner_session_id,
+                id,
+                result,
+                outcome,
+                parent_run_id,
+                spawn_depth,
+                continuable,
+                usage,
+                display_name,
+            } if owner_session_id == thread_id => {
+                let worker_status = outcome
+                    .as_ref()
+                    .map(crate::tools::subagent::subagent_status_name);
+                let name = display_name.as_deref().unwrap_or(&id);
+                let message = format!(
+                    "Sub-agent {name} {}: {}",
+                    worker_status.unwrap_or("settled (outcome unconfirmed)"),
+                    summarize_text(&result, SUMMARY_LIMIT)
+                );
+                let item = TurnItemRecord {
+                    schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
+                    id: item_id,
+                    turn_id: turn_id.clone(),
+                    kind: TurnItemKind::Status,
+                    status: TurnItemLifecycleStatus::Completed,
+                    summary: summarize_text(&message, SUMMARY_LIMIT),
+                    detail: Some(message),
+                    metadata: None,
+                    artifact_refs: Vec::new(),
+                    artifacts: Vec::new(),
+                    started_at: Some(Utc::now()),
+                    ended_at: Some(Utc::now()),
+                };
+                self.save_item(&item).await?;
+                self.emit_event(
+                    &thread_id,
+                    Some(&turn_id),
+                    Some(&item.id),
+                    "agent.completed",
+                    json!({ "item": item, "agent_id": id, "agent_name": display_name,
+                            "worker_status": worker_status, "parent_run_id": parent_run_id,
+                            "spawn_depth": spawn_depth, "continuable": continuable,
+                            "usage": usage }),
+                )
+                .await?;
+                self.raise_notice(
+                    &thread_id,
+                    codewhale_config::notifications::NotificationEvent::SubagentTerminal.as_str(),
+                    &turn_id,
+                    &id,
+                    format!(
+                        "{name} {}",
+                        match outcome.as_ref() {
+                            Some(SubAgentStatus::Completed) => "finished",
+                            Some(SubAgentStatus::Failed(_) | SubAgentStatus::BudgetExhausted) =>
+                                "failed",
+                            Some(SubAgentStatus::Cancelled | SubAgentStatus::Interrupted(_)) =>
+                                "stopped",
+                            Some(SubAgentStatus::Running) | None => "settled (outcome unconfirmed)",
+                        }
+                    ),
+                );
+            }
+            _ => {}
+        }
+        Ok(Some(turn_id))
+    }
+
+    async fn save_item(&self, item: &TurnItemRecord) -> Result<()> {
+        let store = self.store.clone();
+        let item = item.clone();
+        tokio::task::spawn_blocking(move || -> Result<()> {
+            store.save_item(&item)?;
+            let _turn_mutation = store.turn_mutation.lock();
+            let mut turn = store.load_turn(&item.turn_id)?;
+            if !turn.item_ids.contains(&item.id) {
+                turn.item_ids.push(item.id);
+                store.save_turn(&turn)?;
+            }
+            Ok(())
+        })
+        .await
+        .context("child receipt persistence failed")?
+    }
+
+    async fn emit_event(
+        &self,
+        thread_id: &str,
+        turn_id: Option<&str>,
+        item_id: Option<&str>,
+        event: impl Into<String>,
+        payload: Value,
+    ) -> Result<RuntimeEventRecord> {
+        let _emit = self.event_emit.lock().await;
+        let record = self
+            .store
+            .append_event(thread_id, turn_id, item_id, event, payload)
+            .await?;
+        let _ = self.event_tx.send(record.clone());
+        Ok(record)
+    }
+
+    fn raise_notice(
+        &self,
+        thread_id: &str,
+        kind: &str,
+        turn_id: &str,
+        subject: &str,
+        detail: String,
+    ) -> String {
+        push_runtime_notice(&self.notices, thread_id, kind, turn_id, subject, detail)
+    }
+}
+
 #[derive(Clone)]
 struct RecoveredTurnReceipt {
     turn: TurnRecord,
@@ -6275,8 +6621,13 @@ pub struct RuntimeThreadManager {
     config_admission: Arc<AsyncRwLock<()>>,
     engine_load: Arc<Mutex<()>>,
     shutdown_drain: Arc<Mutex<()>>,
-    engine_workers: Arc<parking_lot::Mutex<Vec<(EngineHandle, RuntimeCompletion)>>>,
+    engine_workers: Arc<parking_lot::Mutex<Vec<(String, EngineHandle, RuntimeCompletion)>>>,
+    subagent_manager: Arc<Mutex<Option<SharedSubAgentManager>>>,
     turn_monitors: Arc<parking_lot::Mutex<Vec<RuntimeCompletion>>>,
+    event_monitors: Arc<parking_lot::Mutex<Vec<RuntimeCompletion>>>,
+    event_reader_wake: Arc<Notify>,
+    event_pump_stop: CancellationToken,
+    _event_pump_owner: Arc<tokio_util::sync::DropGuard>,
     active: Arc<Mutex<ActiveThreads>>,
     event_emit: Arc<Mutex<()>>,
     projection_locks: Arc<parking_lot::Mutex<HashMap<String, Arc<Mutex<()>>>>>,
@@ -7718,6 +8069,7 @@ impl RuntimeThreadManager {
         if let Some(reason) = grants.degraded_reason() {
             tracing::warn!(%reason, "approval grants are unavailable; remembered approvals will not apply");
         }
+        let event_pump_stop = CancellationToken::new();
         let manager = Self {
             host_profile,
             config: Arc::new(parking_lot::RwLock::new(config)),
@@ -7729,7 +8081,12 @@ impl RuntimeThreadManager {
             engine_load: Arc::new(Mutex::new(())),
             shutdown_drain: Arc::new(Mutex::new(())),
             engine_workers: Arc::new(parking_lot::Mutex::new(Vec::new())),
+            subagent_manager: Arc::new(Mutex::new(None)),
             turn_monitors: Arc::new(parking_lot::Mutex::new(Vec::new())),
+            event_monitors: Arc::new(parking_lot::Mutex::new(Vec::new())),
+            event_reader_wake: Arc::new(Notify::new()),
+            event_pump_stop: event_pump_stop.clone(),
+            _event_pump_owner: Arc::new(event_pump_stop.drop_guard()),
             active: Arc::new(Mutex::new(ActiveThreads::default())),
             event_emit: Arc::new(Mutex::new(())),
             projection_locks: Arc::new(parking_lot::Mutex::new(HashMap::new())),
@@ -7856,18 +8213,18 @@ impl RuntimeThreadManager {
         let workers = self.engine_workers.lock().clone();
         for engine in engines
             .iter()
-            .chain(workers.iter().map(|(engine, _)| engine))
+            .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))
+            .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 {
+        for (_, _, worker) in &workers {
+            if let Err(error) = await_retained_completion(worker).await {
                 failure = Some(anyhow!("Runtime engine shutdown failed: {error}"));
             }
         }
@@ -7890,9 +8247,35 @@ impl RuntimeThreadManager {
                 break;
             }
         }
+        for (owner, engine, _) in &workers {
+            let authority = engine.subagent_manager();
+            let live = authority.read().await.live_count_for_session(owner);
+            if live > 0 {
+                failure = Some(anyhow!(
+                    "Runtime shutdown left {live} unsettled workers for {owner}"
+                ));
+            }
+        }
+        for engine in engines
+            .iter()
+            .chain(workers.iter().map(|(_, engine, _)| engine))
+        {
+            if let Err(error) = engine.wait_terminal_worker_events().await {
+                failure = Some(anyhow!("Runtime terminal worker delivery failed: {error}"));
+            }
+        }
+        self.event_pump_stop.cancel();
+        self.event_reader_wake.notify_waiters();
+        let pumps = self.event_monitors.lock().clone();
+        for pump in pumps {
+            if let Err(error) = await_retained_completion(&pump).await {
+                failure = Some(anyhow!("Runtime Engine event receiver failed: {error}"));
+            }
+        }
+        self.event_monitors.lock().clear();
         self.engine_workers
             .lock()
-            .retain(|(_, worker)| !completion_finished(worker));
+            .retain(|(_, _, worker)| !completion_finished(worker));
         let mut active = self.active.lock().await;
         active.engines.clear();
         active.lru.clear();
@@ -8628,6 +9011,7 @@ impl RuntimeThreadManager {
                 id: String::new(),
                 turn_id: "test-turn".to_string(),
                 tool_name: "test-tool".to_string(),
+                input: None,
                 description: format!("test approval {label}"),
                 intent_summary: None,
                 // Stands in for the provider's raw call ID so tests can prove
@@ -10520,25 +10904,7 @@ impl RuntimeThreadManager {
         subject: &str,
         detail: String,
     ) -> String {
-        debug_assert!(
-            codewhale_config::notifications::NotificationEvent::parse(kind).is_some(),
-            "notice kind must be a NotificationEvent name: {kind}"
-        );
-        let id = runtime_record_id("notice");
-        let mut notices = self.notices.lock();
-        let list = notices.entry(thread_id.to_string()).or_default();
-        list.push(ActiveNotice {
-            id: id.clone(),
-            kind: kind.to_string(),
-            turn_id: turn_id.to_string(),
-            subject: subject.to_string(),
-            detail,
-            raised_at: Utc::now(),
-        });
-        if list.len() > MAX_NOTICES_PER_THREAD {
-            list.drain(..list.len() - MAX_NOTICES_PER_THREAD);
-        }
-        id
+        push_runtime_notice(&self.notices, thread_id, kind, turn_id, subject, detail)
     }
 
     /// Notices currently raised on a thread, oldest first. Unknown threads
@@ -14073,6 +14439,7 @@ impl RuntimeThreadManager {
                 "Runtime shutdown requested before turn monitoring started",
             )
             .await;
+            self.event_reader_wake.notify_waiters();
             return;
         }
 
@@ -14089,7 +14456,10 @@ impl RuntimeThreadManager {
         .catch_unwind()
         .await;
         let failure = match result {
-            Ok(Ok(())) => return,
+            Ok(Ok(())) => {
+                self.event_reader_wake.notify_waiters();
+                return;
+            }
             Ok(Err(error)) => {
                 let failure = format!("Failed to monitor {}: {error}", kind.label());
                 // An unreadable item or turn under the monitor is the
@@ -14111,6 +14481,7 @@ impl RuntimeThreadManager {
         engine.cancel_with_reason(crate::core::engine::CancelReason::Internal);
         self.settle_claimed_turn_failure(&thread_id, &turn_id, &failure)
             .await;
+        self.event_reader_wake.notify_waiters();
     }
 
     fn spawn_claimed_turn_monitor(
@@ -14143,6 +14514,7 @@ impl RuntimeThreadManager {
                 manager
                     .settle_claimed_turn_failure(&turn.thread_id, &turn.id, &failure)
                     .await;
+                manager.event_reader_wake.notify_waiters();
                 return;
             }
 
@@ -15808,17 +16180,33 @@ impl RuntimeThreadManager {
             let model_client = self.test_model_client.lock().clone();
             #[cfg(not(test))]
             let model_client = None;
+            let subagent_manager = self.subagent_manager().await?;
+            let previous = self.engine_workers.lock().clone();
+            for (owner, old_engine, completion) in previous {
+                if owner == thread.id {
+                    old_engine.cancel_with_reason(crate::core::engine::CancelReason::Internal);
+                    let _ = old_engine.send(Op::Shutdown).await;
+                    await_retained_completion(&completion).await?;
+                }
+            }
             let (engine, worker) = spawn_engine_with_authoritative_route_config(
                 engine_cfg,
                 &cfg,
                 self.host_profile,
                 Arc::clone(&self.config),
                 model_client,
+                Some(subagent_manager),
+                self.cancel_token.clone(),
             );
+            let engine = self.route_engine_events(&thread, engine).await;
             {
                 let mut workers = self.engine_workers.lock();
-                workers.retain(|(_, worker)| !completion_finished(worker));
-                workers.push((engine.clone(), retained_completion(worker)));
+                workers.retain(|(_, _, worker)| !completion_finished(worker));
+                workers.push((
+                    thread.id.clone(),
+                    engine.clone(),
+                    retained_completion(worker),
+                ));
             }
 
             let sys_prompt = thread
@@ -15897,6 +16285,159 @@ impl RuntimeThreadManager {
         }
     }
 
+    async fn route_engine_events(
+        &self,
+        thread: &ThreadRecord,
+        mut engine: EngineHandle,
+    ) -> EngineHandle {
+        let original = engine.rx_event.clone();
+        let capacity = original.read().await.max_capacity();
+        let (forward, receiver) = mpsc::channel(capacity);
+        engine.rx_event = Arc::new(AsyncRwLock::new(receiver));
+        let identity = Arc::downgrade(&engine.rx_event);
+        let active = Arc::downgrade(&self.active);
+        let wake = self.event_reader_wake.clone();
+        let stop = self.event_pump_stop.clone();
+        let sink = RuntimeWorkerEventSink {
+            thread_id: thread.id.clone(),
+            workspace: thread.workspace.clone(),
+            manager: engine.subagent_manager(),
+            store: self.store.clone(),
+            projection: self.projection_lock(&thread.id),
+            event_emit: self.event_emit.clone(),
+            event_tx: self.event_tx.clone(),
+            notices: self.notices.clone(),
+            owner: Arc::downgrade(&self._process_owner_lock),
+        };
+        let handle = engine.clone();
+        let thread_id = thread.id.clone();
+        let pump = tokio::spawn(async move {
+            'pump: loop {
+                let event = tokio::select! {
+                    biased;
+                    event = async { original.write().await.recv().await } => match event { Some(event) => event, None => break },
+                    _ = stop.cancelled() => break,
+                };
+                if matches!(
+                    &event,
+                    EngineEvent::AgentSpawned { .. }
+                        | EngineEvent::AgentProgress { .. }
+                        | EngineEvent::AgentComplete { .. }
+                ) {
+                    let origin = match sink.publish(event.clone()).await {
+                        Ok(origin) => origin,
+                        Err(error) => {
+                            handle.mark_worker_event_delivery_failed();
+                            tracing::error!(thread_id = %thread_id, "Runtime child event publication failed: {error:#}");
+                            if let Some(_owner) = sink.owner.upgrade() {
+                                if let Some(failure) = RuntimeStoreRecordFailure::from_error(&error)
+                                {
+                                    let _ = sink
+                                        .emit_event(
+                                            &thread_id,
+                                            None,
+                                            None,
+                                            RUNTIME_STORE_FAILURE_EVENT,
+                                            json!(failure.notice(&error, false)),
+                                        )
+                                        .await;
+                                } else {
+                                    let _ = sink.emit_event(&thread_id, None, None, "error", json!({"message": "A background agent receipt could not be recorded."})).await;
+                                }
+                            }
+                            break;
+                        }
+                    };
+                    let owns_origin =
+                        if let (Some(origin), Some(active)) = (origin, active.upgrade()) {
+                            active
+                                .lock()
+                                .await
+                                .engines
+                                .get(&thread_id)
+                                .and_then(|state| state.active_turn.as_ref())
+                                .is_some_and(|turn| turn.turn_id == origin)
+                        } else {
+                            false
+                        };
+                    if !owns_origin {
+                        continue;
+                    }
+                }
+                loop {
+                    let notified = wake.notified();
+                    tokio::pin!(notified);
+                    notified.as_mut().enable();
+                    let Some(active) = active.upgrade() else {
+                        break 'pump;
+                    };
+                    let owns_turn = {
+                        let active = active.lock().await;
+                        active.engines.get(&thread_id).is_some_and(|state| {
+                            state.active_turn.is_some()
+                                && identity.upgrade().is_some_and(|receiver| {
+                                    Arc::ptr_eq(&receiver, &state.engine.rx_event)
+                                })
+                        })
+                    };
+                    if !owns_turn {
+                        break;
+                    }
+                    tokio::select! {
+                        permit = forward.reserve() => match permit {
+                            Ok(permit) => { permit.send(event); break; },
+                            Err(_) => break 'pump,
+                        },
+                        _ = &mut notified => {},
+                        _ = stop.cancelled() => break 'pump,
+                    }
+                }
+            }
+            handle.cancel_with_reason(crate::core::engine::CancelReason::Internal);
+            let _ = handle.try_send(Op::Shutdown);
+        });
+        let mut monitors = self.event_monitors.lock();
+        monitors.retain(|monitor| !completion_finished(monitor));
+        monitors.push(retained_completion(pump));
+        engine
+    }
+
+    pub(crate) async fn attach_subagent_manager(
+        &self,
+        manager: SharedSubAgentManager,
+    ) -> Result<()> {
+        let _loading = self.engine_load.lock().await;
+        let mut selected = self.subagent_manager.lock().await;
+        if let Some(current) = selected.as_ref() {
+            anyhow::ensure!(
+                Arc::ptr_eq(current, &manager),
+                "Runtime already has another worker authority"
+            );
+        } else {
+            anyhow::ensure!(
+                self.active.lock().await.engines.is_empty(),
+                "Runtime engines started before worker authority admission"
+            );
+            *selected = Some(manager);
+        }
+        Ok(())
+    }
+
+    async fn subagent_manager(&self) -> Result<SharedSubAgentManager> {
+        let mut selected = self.subagent_manager.lock().await;
+        if let Some(manager) = selected.as_ref() {
+            return Ok(manager.clone());
+        }
+        let workspace = self.workspace.clone();
+        let manager = tokio::task::spawn_blocking(move || {
+            new_shared_runtime_subagent_manager(workspace, MAX_SUBAGENTS)
+        })
+        .await
+        .context("Runtime worker authority admission failed")??;
+        *selected = Some(manager.clone());
+        Ok(manager)
+    }
+
     /// The thread's engine only when it is already live in this process.
     /// Control routes that act on in-flight work (stopping an agent run) use
     /// this: loading a cold engine cannot reach work that is not running here.
@@ -17505,154 +18046,6 @@ impl RuntimeThreadManager {
                         .await?;
                     }
                 }
-                EngineEvent::AgentSpawned {
-                    owner_session_id,
-                    id,
-                    prompt,
-                    worker_status,
-                    parent_run_id,
-                    spawn_depth,
-                    display_name,
-                    ..
-                } if owner_session_id == thread_id => {
-                    // Hosts name the agent the way the TUI does (#6565).
-                    let name = display_name.as_deref().unwrap_or(&id);
-                    let message = format!(
-                        "Sub-agent {name} spawned: {}",
-                        summarize_text(&prompt, SUMMARY_LIMIT)
-                    );
-                    let item = TurnItemRecord {
-                        schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
-                        id: runtime_record_id("item"),
-                        turn_id: turn_id.clone(),
-                        kind: TurnItemKind::Status,
-                        status: TurnItemLifecycleStatus::Completed,
-                        summary: summarize_text(&message, SUMMARY_LIMIT),
-                        detail: Some(message),
-                        metadata: None,
-                        artifact_refs: Vec::new(),
-                        artifacts: Vec::new(),
-                        started_at: Some(Utc::now()),
-                        ended_at: Some(Utc::now()),
-                    };
-                    self.store.save_item(&item)?;
-                    self.attach_item_to_turn(&turn_id, &item.id)?;
-                    self.emit_event(
-                        &thread_id,
-                        Some(&turn_id),
-                        Some(&item.id),
-                        "agent.spawned",
-                        json!({ "item": item, "agent_id": id, "agent_name": display_name,
-                            "worker_status": worker_status, "parent_run_id": parent_run_id,
-                            "spawn_depth": spawn_depth }),
-                    )
-                    .await?;
-                }
-                EngineEvent::AgentProgress {
-                    owner_session_id,
-                    id,
-                    status,
-                    activity,
-                    parent_run_id,
-                    spawn_depth,
-                } if owner_session_id == thread_id => {
-                    let message = format!("Sub-agent {id}: {status}");
-                    let item = TurnItemRecord {
-                        schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
-                        id: runtime_record_id("item"),
-                        turn_id: turn_id.clone(),
-                        kind: TurnItemKind::Status,
-                        status: TurnItemLifecycleStatus::Completed,
-                        summary: summarize_text(&message, SUMMARY_LIMIT),
-                        detail: Some(message),
-                        metadata: None,
-                        artifact_refs: Vec::new(),
-                        artifacts: Vec::new(),
-                        started_at: Some(Utc::now()),
-                        ended_at: Some(Utc::now()),
-                    };
-                    self.store.save_item(&item)?;
-                    self.attach_item_to_turn(&turn_id, &item.id)?;
-                    self.emit_event(
-                        &thread_id,
-                        Some(&turn_id),
-                        Some(&item.id),
-                        "agent.progress",
-                        json!({ "item": item, "agent_id": id,
-                            "worker_status": activity.worker_status, "step": activity.step,
-                            "parent_run_id": parent_run_id, "spawn_depth": spawn_depth }),
-                    )
-                    .await?;
-                }
-                EngineEvent::AgentComplete {
-                    owner_session_id,
-                    id,
-                    result,
-                    outcome,
-                    parent_run_id,
-                    spawn_depth,
-                    continuable,
-                    usage,
-                    display_name,
-                } if owner_session_id == thread_id => {
-                    let worker_status = outcome
-                        .as_ref()
-                        .map(crate::tools::subagent::subagent_status_name);
-                    let name = display_name.as_deref().unwrap_or(&id);
-                    let message = format!(
-                        "Sub-agent {name} {}: {}",
-                        worker_status.unwrap_or("settled (outcome unconfirmed)"),
-                        summarize_text(&result, SUMMARY_LIMIT)
-                    );
-                    let item = TurnItemRecord {
-                        schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
-                        id: runtime_record_id("item"),
-                        turn_id: turn_id.clone(),
-                        kind: TurnItemKind::Status,
-                        status: TurnItemLifecycleStatus::Completed,
-                        summary: summarize_text(&message, SUMMARY_LIMIT),
-                        detail: Some(message),
-                        metadata: None,
-                        artifact_refs: Vec::new(),
-                        artifacts: Vec::new(),
-                        started_at: Some(Utc::now()),
-                        ended_at: Some(Utc::now()),
-                    };
-                    self.store.save_item(&item)?;
-                    self.attach_item_to_turn(&turn_id, &item.id)?;
-                    self.emit_event(
-                        &thread_id,
-                        Some(&turn_id),
-                        Some(&item.id),
-                        "agent.completed",
-                        json!({ "item": item, "agent_id": id, "agent_name": display_name,
-                            "worker_status": worker_status, "parent_run_id": parent_run_id,
-                            "spawn_depth": spawn_depth, "continuable": continuable,
-                            "usage": usage }),
-                    )
-                    .await?;
-                    self.raise_notice(
-                        &thread_id,
-                        codewhale_config::notifications::NotificationEvent::SubagentTerminal
-                            .as_str(),
-                        &turn_id,
-                        &id,
-                        format!(
-                            "{name} {}",
-                            match outcome.as_ref() {
-                                Some(SubAgentStatus::Completed) => "finished",
-                                Some(
-                                    SubAgentStatus::Failed(_) | SubAgentStatus::BudgetExhausted,
-                                ) => "failed",
-                                Some(
-                                    SubAgentStatus::Cancelled | SubAgentStatus::Interrupted(_),
-                                ) => "stopped",
-                                Some(SubAgentStatus::Running) | None =>
-                                    "settled (outcome unconfirmed)",
-                            }
-                        ),
-                    );
-                }
                 EngineEvent::AgentList {
                     owner_session_id,
                     agents,
@@ -17734,13 +18127,10 @@ impl RuntimeThreadManager {
                     );
 
                     let pending_request = PendingApprovalRequest {
-                        // Replaced by the minted ID at registration. The raw
-                        // provider call ID travels in `tool_call_id`, where it
-                        // is a correlator a snapshot client can match against
-                        // the tool row and never a value the API will accept.
                         id: String::new(),
                         turn_id: turn_id.clone(),
                         tool_name: tool_name.clone(),
+                        input: Some(input.clone()),
                         description: description.clone(),
                         intent_summary: intent_summary.clone(),
                         tool_call_id: Some(id.clone()),
@@ -17830,6 +18220,7 @@ impl RuntimeThreadManager {
                                 "approval_id": approval_id,
                                 "tool_call_id": id,
                                 "tool_name": tool_name,
+                                "input": input,
                                 "summary": summary,
                                 "description": description,
                                 "intent_summary": intent_summary,
@@ -17933,6 +18324,7 @@ impl RuntimeThreadManager {
                                 "approval_id": approval_id,
                                 "tool_call_id": id,
                                 "tool_name": tool_name,
+                                "input": input,
                                 "summary": summary,
                                 "description": description,
                                 "intent_summary": intent_summary,
@@ -18000,6 +18392,7 @@ impl RuntimeThreadManager {
                                 "approval_id": approval_id,
                                 "tool_call_id": id,
                                 "tool_name": tool_name,
+                                "input": input,
                                 "summary": summary,
                                 "description": description,
                                 "intent_summary": intent_summary,
@@ -19259,6 +19652,18 @@ impl RuntimeThreadManager {
         let thread = self.get_thread(thread_id).await?;
         let config = self.read_config().clone();
         let route = self.resolved_route_for_thread(&config, &thread)?;
+        {
+            let mut selected = self.subagent_manager.lock().await;
+            if let Some(manager) = selected.as_ref() {
+                anyhow::ensure!(
+                    Arc::ptr_eq(manager, &engine.subagent_manager()),
+                    "fixture worker authority differs from Runtime owner"
+                );
+            } else {
+                *selected = Some(engine.subagent_manager());
+            }
+        }
+        let engine = self.route_engine_events(&thread, engine).await;
         let mut active = self.active.lock().await;
         active.engines.insert(
             thread_id.to_string(),
@@ -19499,7 +19904,11 @@ fn enforce_lru_capacity(
         .engines
         .iter()
         .filter_map(|(thread_id, state)| {
-            if state.active_turn.is_some() {
+            let manager = state.engine.subagent_manager();
+            let live_children = manager.try_read().map_or(true, |manager| {
+                manager.live_count_for_session(thread_id) > 0
+            });
+            if state.active_turn.is_some() || live_children {
                 Some(thread_id.clone())
             } else {
                 None

crates/tui/src/runtime_threads/tests.rs

@@ -6708,7 +6708,10 @@ async fn install_mock_engine(
     manager: &RuntimeThreadManager,
     thread_id: &str,
 ) -> crate::core::engine::MockEngineHandle {
-    let harness = mock_engine_handle();
+    let mut harness = mock_engine_handle();
+    if let Some(authority) = manager.subagent_manager.lock().await.as_ref() {
+        harness.handle.set_subagent_manager(authority.clone());
+    }
     manager
         .install_test_engine(thread_id, harness.handle.clone())
         .await
@@ -8104,6 +8107,23 @@ fn enforce_lru_capacity_does_not_loop_when_all_threads_are_active() {
     assert!(evicted.is_empty(), "no idle threads should be evicted");
     assert_eq!(active.engines.len(), 2);
     assert_eq!(active.lru.len(), 2);
+    active.engines.get_mut("thr_a").unwrap().active_turn = None;
+    let authority = active.engines["thr_a"].engine.subagent_manager();
+    let child = {
+        let mut workers = authority.try_write().unwrap();
+        let child = workers.insert_test_running_agent("lru_child", &PathBuf::from("."));
+        workers.assign_test_session_owner(&child, "thr_a");
+        child
+    };
+    assert!(enforce_lru_capacity(&mut active, 2).is_empty());
+    let _ = authority
+        .try_write()
+        .unwrap()
+        .cancel_agent_for_session("thr_a", &child);
+    let evicted = enforce_lru_capacity(&mut active, 2);
+    assert_eq!(evicted.len(), 1);
+    assert!(!active.engines.contains_key("thr_a"));
+    assert!(active.engines.contains_key("thr_b"));
 }
 
 #[test]
@@ -9944,7 +9964,10 @@ async fn completed_turn_without_engine_output_fails() -> Result<()> {
 #[tokio::test]
 async fn worker_lifecycle_receipts_preserve_owner_outcome_and_durable_replay() -> Result<()> {
     use crate::core::events::AgentProgressEventMeta;
-    use crate::tools::subagent::AgentWorkerStatus;
+    use crate::tools::subagent::{
+        AgentWorkerSpec, AgentWorkerStatus, AgentWorkerToolProfile, FleetRole, SubAgentManager,
+    };
+    use crate::worker_profile::WorkerRuntimeProfile;
 
     let runtime_dir = test_runtime_dir();
     let manager = test_manager(runtime_dir.clone())?;
@@ -9954,89 +9977,159 @@ async fn worker_lifecycle_receipts_preserve_owner_outcome_and_durable_replay() -
     let foreign = manager
         .create_thread(CreateThreadRequest::default())
         .await?;
-    let harness = install_mock_engine(&manager, &thread.id).await;
+    let mut harness = mock_engine_handle();
+    harness
+        .handle
+        .set_subagent_manager(Arc::new(tokio::sync::RwLock::new(SubAgentManager::new(
+            thread.workspace.clone(),
+            1,
+        ))));
+    manager
+        .install_test_engine(&thread.id, harness.handle.clone())
+        .await?;
+    let worker_manager = harness.handle.subagent_manager();
     let mut rx_op = harness.rx_op;
     let tx_event = harness.tx_event;
     let thread_id = thread.id.clone();
     let foreign_id = foreign.id.clone();
+    let (registered_tx, registered_rx) = oneshot::channel::<()>();
+    let (late_tx, late_rx) = oneshot::channel::<()>();
+    let mut events_rx = manager.subscribe_events();
+    let completed = EngineEvent::AgentComplete {
+        display_name: Some("audit docs".into()),
+        owner_session_id: thread.id.clone(),
+        id: "worker_completed".into(),
+        result: "Completed successfully".into(),
+        outcome: Some(SubAgentStatus::Completed),
+        parent_run_id: Some("parent".into()),
+        spawn_depth: Some(2),
+        continuable: Some(false),
+        usage: None,
+    };
     tokio::spawn(async move {
-        if matches!(rx_op.recv().await, Some(Op::SendMessage(TurnSpec { .. }))) {
+        if !matches!(rx_op.recv().await, Some(Op::SendMessage(TurnSpec { .. }))) {
+            return;
+        }
+        if registered_rx.await.is_err() {
+            return;
+        }
+        let _ = tx_event
+            .send(EngineEvent::TurnStarted {
+                turn_id: "engine_worker_lifecycle".into(),
+                created_at: Utc::now(),
+                route: None,
+                submission_id: None,
+            })
+            .await;
+        for owner in [&thread_id, &foreign_id] {
             let _ = tx_event
-                .send(EngineEvent::TurnStarted {
-                    turn_id: "engine_worker_lifecycle".into(),
-                    created_at: Utc::now(),
-                    route: None,
-                    submission_id: None,
+                .send(EngineEvent::AgentSpawned {
+                    display_name: None,
+                    owner_session_id: owner.to_string(),
+                    id: "worker_spawn".into(),
+                    prompt: "private prompt".into(),
+                    worker_status: Some(AgentWorkerStatus::Queued),
+                    parent_run_id: Some("parent".into()),
+                    spawn_depth: 2,
+                    model: "fixture".into(),
+                    route_source: None,
                 })
                 .await;
-            for owner in [thread_id.clone(), foreign_id] {
-                let _ = tx_event
-                    .send(EngineEvent::AgentSpawned {
-                        display_name: None,
-                        owner_session_id: owner.clone(),
-                        id: "worker_spawn".into(),
-                        prompt: "private prompt".into(),
-                        worker_status: Some(AgentWorkerStatus::Queued),
-                        parent_run_id: Some("parent".into()),
-                        spawn_depth: 2,
-                        model: "fixture".into(),
-                        route_source: None,
-                    })
-                    .await;
-                let _ = tx_event
-                    .send(EngineEvent::AgentProgress {
-                        owner_session_id: owner.clone(),
-                        id: "worker_spawn".into(),
-                        status: "private progress".into(),
-                        activity: AgentProgressEventMeta::new(AgentWorkerStatus::RunningTool)
-                            .with_step(3)
-                            .with_tool("private_tool"),
-                        parent_run_id: Some("parent".into()),
-                        spawn_depth: 2,
-                    })
-                    .await;
-                for (id, outcome) in [
-                    ("worker_completed", Some(SubAgentStatus::Completed)),
-                    (
-                        "worker_failed",
-                        Some(SubAgentStatus::Failed("private failure".into())),
-                    ),
-                    (
-                        "worker_interrupted",
-                        Some(SubAgentStatus::Interrupted("private reason".into())),
-                    ),
-                    ("worker_cancelled", Some(SubAgentStatus::Cancelled)),
-                    ("worker_budget", Some(SubAgentStatus::BudgetExhausted)),
-                    ("worker_legacy", None),
-                ] {
-                    let _ = tx_event
-                        .send(EngineEvent::AgentComplete {
-                            // #6565: hosts name the agent the way the TUI does.
-                            display_name: (id == "worker_completed").then(|| "audit docs".into()),
-                            owner_session_id: owner.clone(),
-                            id: id.into(),
-                            result: "Completed successfully".into(),
-                            outcome,
-                            parent_run_id: Some("parent".into()),
-                            spawn_depth: Some(2),
-                            continuable: Some(id == "worker_interrupted"),
-                            usage: None,
-                        })
-                        .await;
-                }
-            }
             let _ = tx_event
-                .send(EngineEvent::TurnComplete {
-                    usage: Usage::default(),
-                    parent_route_usage: Usage::default(),
-                    routed_usage_dropped_records: 0,
-                    status: TurnOutcomeStatus::Completed,
-                    error: None,
-                    tool_catalog: None,
-                    base_url: None,
+                .send(EngineEvent::AgentProgress {
+                    owner_session_id: owner.to_string(),
+                    id: "worker_spawn".into(),
+                    status: "private progress".into(),
+                    activity: AgentProgressEventMeta::new(AgentWorkerStatus::RunningTool)
+                        .with_step(3)
+                        .with_tool("private_tool"),
+                    parent_run_id: Some("parent".into()),
+                    spawn_depth: 2,
                 })
                 .await;
         }
+        let _ = tx_event
+            .send(EngineEvent::TurnComplete {
+                usage: Usage::default(),
+                parent_route_usage: Usage::default(),
+                routed_usage_dropped_records: 0,
+                status: TurnOutcomeStatus::Completed,
+                error: None,
+                tool_catalog: None,
+                base_url: None,
+            })
+            .await;
+        if late_rx.await.is_err() {
+            return;
+        }
+        let _ = tx_event.send(completed.clone()).await;
+        let _ = tx_event.send(completed.clone()).await;
+        if !matches!(rx_op.recv().await, Some(Op::SendMessage(TurnSpec { .. }))) {
+            return;
+        }
+        let _ = tx_event
+            .send(EngineEvent::TurnStarted {
+                turn_id: "engine_next_parent".into(),
+                created_at: Utc::now(),
+                route: None,
+                submission_id: None,
+            })
+            .await;
+        let _ = tx_event
+            .send(EngineEvent::MessageStarted { index: 0 })
+            .await;
+        let _ = tx_event
+            .send(EngineEvent::MessageDelta {
+                index: 0,
+                content: "Second parent response".into(),
+            })
+            .await;
+        let _ = tx_event
+            .send(EngineEvent::MessageComplete { index: 0 })
+            .await;
+
+        for owner in [&thread_id, &foreign_id] {
+            for (id, outcome) in [
+                (
+                    "worker_failed",
+                    Some(SubAgentStatus::Failed("private failure".into())),
+                ),
+                (
+                    "worker_interrupted",
+                    Some(SubAgentStatus::Interrupted("private reason".into())),
+                ),
+                ("worker_cancelled", Some(SubAgentStatus::Cancelled)),
+                ("worker_budget", Some(SubAgentStatus::BudgetExhausted)),
+                ("worker_legacy", None),
+                ("worker_unknown_origin", Some(SubAgentStatus::Completed)),
+            ] {
+                let _ = tx_event
+                    .send(EngineEvent::AgentComplete {
+                        display_name: None,
+                        owner_session_id: owner.to_string(),
+                        id: id.into(),
+                        result: "Completed successfully".into(),
+                        continuable: Some(id == "worker_interrupted"),
+                        outcome,
+                        parent_run_id: Some("parent".into()),
+                        spawn_depth: Some(2),
+                        usage: None,
+                    })
+                    .await;
+            }
+        }
+        let _ = tx_event.send(completed).await;
+        let _ = tx_event
+            .send(EngineEvent::TurnComplete {
+                usage: Usage::default(),
+                parent_route_usage: Usage::default(),
+                routed_usage_dropped_records: 0,
+                status: TurnOutcomeStatus::Completed,
+                error: None,
+                tool_catalog: None,
+                base_url: None,
+            })
+            .await;
     });
     let turn = manager
         .start_turn(
@@ -10047,7 +10140,87 @@ async fn worker_lifecycle_receipts_preserve_owner_outcome_and_durable_replay() -
             },
         )
         .await?;
-    let _ = wait_for_terminal_turn(&manager, &turn.id).await?;
+    {
+        let mut workers = worker_manager.write().await;
+        for id in [
+            "worker_spawn",
+            "worker_completed",
+            "worker_failed",
+            "worker_interrupted",
+            "worker_cancelled",
+            "worker_budget",
+            "worker_legacy",
+            "worker_unknown_origin",
+        ] {
+            workers.register_worker_for_session_with_origin(
+                AgentWorkerSpec {
+                    worker_id: id.into(),
+                    run_id: id.into(),
+                    parent_run_id: Some("parent".into()),
+                    workflow_run_id: None,
+                    session_name: None,
+                    objective: "fixture".into(),
+                    role: None,
+                    agent_type: FleetRole::Verifier,
+                    model: "fixture".into(),
+                    workspace: if id == "worker_completed" {
+                        thread.workspace.join("isolated-child")
+                    } else {
+                        thread.workspace.clone()
+                    },
+                    git_branch: None,
+                    context_mode: "fresh".into(),
+                    fork_context: false,
+                    tool_profile: AgentWorkerToolProfile::Inherited,
+                    runtime_profile: WorkerRuntimeProfile::default(),
+                    max_steps: 2,
+                    spawn_depth: 2,
+                    max_spawn_depth: 3,
+                    child_route: None,
+                    launch_manifest: None,
+                },
+                &thread.id,
+                None,
+                (id != "worker_unknown_origin").then(|| turn.id.clone()),
+            );
+        }
+    }
+    registered_tx
+        .send(())
+        .map_err(|_| anyhow!("fixture registration receiver closed"))?;
+    let original = wait_for_terminal_turn(&manager, &turn.id).await?;
+    assert_eq!(original.status, RuntimeTurnStatus::Completed);
+    assert!(
+        manager.active.lock().await.engines[&thread.id]
+            .active_turn
+            .is_none()
+    );
+    late_tx
+        .send(())
+        .map_err(|_| anyhow!("fixture completion receiver closed"))?;
+    tokio::time::timeout(TURN_SETTLEMENT_DEADLOCK_TIMEOUT, async {
+        loop {
+            let event = events_rx.recv().await?;
+            if event.event == "agent.completed" && event.payload["agent_id"] == "worker_completed" {
+                anyhow::ensure!(
+                    event.turn_id.as_deref() == Some(turn.id.as_str()),
+                    "idle completion changed parent"
+                );
+                return Ok::<_, anyhow::Error>(());
+            }
+        }
+    })
+    .await??;
+    let next = manager
+        .start_turn(
+            &thread.id,
+            StartTurnRequest {
+                prompt: "another parent".into(),
+                ..Default::default()
+            },
+        )
+        .await?;
+    let _ = wait_for_terminal_turn(&manager, &next.id).await?;
     let events = manager.events_since(&thread.id, None)?;
     let workers: Vec<_> = events
         .iter()
@@ -10061,7 +10234,12 @@ async fn worker_lifecycle_receipts_preserve_owner_outcome_and_durable_replay() -
     assert_eq!(
         workers.len(),
         8,
-        "foreign-owner envelopes must never enter this journal"
+        "foreign, unknown-origin and duplicate envelopes must not enter this journal"
+    );
+    assert!(
+        workers
+            .iter()
+            .all(|event| event.turn_id.as_deref() == Some(turn.id.as_str()))
     );
     assert_eq!(workers[0].payload["worker_status"], "queued");
     assert_eq!(workers[1].payload["worker_status"], "running_tool");
@@ -10096,6 +10274,27 @@ async fn worker_lifecycle_receipts_preserve_owner_outcome_and_durable_replay() -
             .unwrap()
             .starts_with("Sub-agent audit docs completed")
     );
+    let original = manager.store.load_turn(&turn.id)?;
+    assert_eq!(original.status, RuntimeTurnStatus::Completed);
+    assert!(
+        workers
+            .iter()
+            .all(|event| original.item_ids.contains(event.item_id.as_ref().unwrap()))
+    );
+    let current = manager.store.load_turn(&next.id)?;
+    assert!(
+        workers
+            .iter()
+            .all(|event| !current.item_ids.contains(event.item_id.as_ref().unwrap()))
+    );
+    assert_eq!(
+        manager
+            .store
+            .load_thread(&thread.id)?
+            .latest_turn_id
+            .as_deref(),
+        Some(next.id.as_str())
+    );
     let notices = manager.list_notices(&thread.id);
     assert!(
         notices
@@ -10115,9 +10314,6 @@ async fn worker_lifecycle_receipts_preserve_owner_outcome_and_durable_replay() -
             .iter()
             .all(|event| !event.event.starts_with("agent."))
     );
-
-    // Reopen only the existing journal authority. This proves persisted event
-    // replay, not a fresh worker-state snapshot or process resurrection.
     let reopened = RuntimeThreadStore::open(runtime_dir)?;
     let replay = reopened.events_since(&thread.id, None)?;
     assert_eq!(
@@ -10131,7 +10327,6 @@ async fn worker_lifecycle_receipts_preserve_owner_outcome_and_durable_replay() -
     );
     Ok(())
 }
-
 #[tokio::test]
 async fn preturn_control_status_does_not_make_empty_turn_succeed() -> Result<()> {
     let manager = test_manager(test_runtime_dir())?;
@@ -12683,144 +12878,169 @@ async fn await_approval_identity(
 
 #[tokio::test]
 async fn approval_required_awaits_external_decision_allow() -> Result<()> {
-    let manager = test_manager(test_runtime_dir())?;
-    let thread = manager
-        .create_thread(CreateThreadRequest {
-            model: None,
-            workspace: None,
-            mode: None,
-            allow_shell: None,
-            trust_mode: None,
-            auto_approve: None,
-            archived: false,
-            system_prompt: None,
-            task_id: None,
-            ..Default::default()
-        })
-        .await?;
-
-    let mut harness = install_mock_engine(&manager, &thread.id).await;
-    let _turn = manager
-        .start_turn(
-            &thread.id,
-            StartTurnRequest {
-                prompt: "needs approval".to_string(),
-                input_summary: None,
+    for (call_id, tool_name, input) in [
+        (
+            "tool_external_allow",
+            "exec_command",
+            json!({"command":"echo fixture","metadata":{"values":[1,true,null]}}),
+        ),
+        (
+            "agent:worker_fixture:approval:00000000-0000-4000-8000-000000000001",
+            "write",
+            json!({"path":"proof/shared.txt","content":"fixture\n"}),
+        ),
+        ("tool_external_null", "fixture_null", Value::Null),
+    ] {
+        let manager = test_manager(test_runtime_dir())?;
+        let thread = manager
+            .create_thread(CreateThreadRequest {
                 model: None,
+                workspace: None,
                 mode: None,
                 allow_shell: None,
                 trust_mode: None,
                 auto_approve: None,
+                archived: false,
+                system_prompt: None,
+                task_id: None,
                 ..Default::default()
-            },
-        )
-        .await?;
-    assert!(matches!(
-        harness.rx_op.recv().await,
-        Some(Op::SendMessage(TurnSpec { .. }))
-    ));
+            })
+            .await?;
 
-    harness
-        .tx_event
-        .send(EngineEvent::ApprovalRequired {
-            approval_key: "key1".to_string(),
-            approval_grouping_key: "key1".to_string(),
-            id: "tool_external_allow".to_string(),
-            tool_name: "exec_command".to_string(),
-            description: "external allow".to_string(),
-            input: serde_json::json!({}),
-            intent_summary: Some("I will update the config file.".to_string()),
-            approval_force_prompt: false,
-        })
-        .await?;
+        let mut harness = install_mock_engine(&manager, &thread.id).await;
+        let _turn = manager
+            .start_turn(
+                &thread.id,
+                StartTurnRequest {
+                    prompt: "needs approval".to_string(),
+                    input_summary: None,
+                    model: None,
+                    mode: None,
+                    allow_shell: None,
+                    trust_mode: None,
+                    auto_approve: None,
+                    ..Default::default()
+                },
+            )
+            .await?;
+        assert!(matches!(
+            harness.rx_op.recv().await,
+            Some(Op::SendMessage(TurnSpec { .. }))
+        ));
 
-    let deadline = Instant::now() + APPROVAL_READINESS_TIMEOUT;
-    while Instant::now() < deadline && manager.pending_approvals_count() == 0 {
-        sleep(Duration::from_millis(20)).await;
+        harness
+            .tx_event
+            .send(EngineEvent::ApprovalRequired {
+                approval_key: "key1".to_string(),
+                approval_grouping_key: "key1".to_string(),
+                id: call_id.to_string(),
+                tool_name: tool_name.to_string(),
+                description: "external allow".to_string(),
+                input: input.clone(),
+                intent_summary: Some("I will update the config file.".to_string()),
+                approval_force_prompt: false,
+            })
+            .await?;
+
+        let deadline = Instant::now() + APPROVAL_READINESS_TIMEOUT;
+        while Instant::now() < deadline && manager.pending_approvals_count() == 0 {
+            sleep(Duration::from_millis(20)).await;
+        }
+        assert_eq!(manager.pending_approvals_count(), 1);
+
+        let approval_id = await_approval_identity(&manager, &thread.id, call_id).await?;
+
+        // The snapshot a reconnecting client resumes from carries the same opaque
+        // capability as the event, so both routes to the ID agree.
+        let detail = manager.get_thread_detail(&thread.id).await?;
+        assert_eq!(detail.pending_approvals.len(), 1);
+        assert_eq!(detail.pending_approvals[0].id, approval_id);
+        assert_eq!(detail.pending_approvals[0].turn_id, _turn.id);
+        assert_eq!(detail.pending_approvals[0].tool_name, tool_name);
+        // The snapshot keeps the raw correlator so a client that reloaded past the
+        // event can still attach this prompt to the tool row it gates.
+        assert_eq!(
+            detail.pending_approvals[0].tool_call_id.as_deref(),
+            Some(call_id)
+        );
+        assert_eq!(detail.pending_approvals[0].input.as_ref(), Some(&input));
+        let serialized = serde_json::to_value(&detail.pending_approvals[0])?;
+        assert_eq!(serialized.get("input"), Some(&input));
+        let reloaded: PendingApprovalRequest = serde_json::from_value(serialized.clone())?;
+        assert_eq!(reloaded.input, Some(input.clone()));
+        let mut legacy = serialized;
+        legacy
+            .as_object_mut()
+            .expect("approval object")
+            .remove("input");
+        let legacy: PendingApprovalRequest = serde_json::from_value(legacy)?;
+        assert_eq!(legacy.input, None);
+        assert!(serde_json::to_value(legacy)?.get("input").is_none());
+        assert_eq!(detail.pending_user_inputs.len(), 0);
+
+        let events = manager.events_since(&thread.id, None)?;
+        let approval_event = events
+            .iter()
+            .rev()
+            .find(|event| event.event == "approval.required")
+            .context("missing approval.required event")?;
+        assert_eq!(approval_event.payload.get("input"), Some(&input));
+        assert_eq!(
+            approval_event
+                .payload
+                .get("intent_summary")
+                .and_then(Value::as_str),
+            Some("I will update the config file.")
+        );
+
+        assert!(!manager.deliver_external_approval(
+            call_id,
+            ExternalApprovalDecision::Allow { remember: false },
+        ));
+        assert_eq!(manager.pending_approvals_count(), 1);
+
+        assert!(manager.deliver_external_approval(
+            &approval_id,
+            ExternalApprovalDecision::Allow { remember: false },
+        ));
+        assert_eq!(
+            harness.recv_approval_event().await,
+            Some(MockApprovalEvent::Approved {
+                id: call_id.to_string(),
+            })
+        );
+        assert_eq!(manager.pending_approvals_count(), 0);
+        assert!(
+            manager
+                .get_thread_detail(&thread.id)
+                .await?
+                .pending_approvals
+                .is_empty()
+        );
+        assert!(
+            manager.events_since(&thread.id, None)?.iter().any(|event| {
+                event.event == "approval.decided"
+                    && event.payload.get("approval_id").and_then(Value::as_str)
+                        == Some(approval_id.as_str())
+                    && event.payload.get("tool_call_id").and_then(Value::as_str) == Some(call_id)
+                    && event.payload.get("decision").and_then(Value::as_str) == Some("allow")
+            }),
+            "approval.decided must correlate the opaque id with the tool call it settled"
+        );
+
+        harness
+            .tx_event
+            .send(EngineEvent::TurnComplete {
+                usage: Usage::default(),
+                parent_route_usage: Usage::default(),
+                routed_usage_dropped_records: 0,
+                status: TurnOutcomeStatus::Completed,
+                error: None,
+                tool_catalog: None,
+                base_url: None,
+            })
+            .await?;
     }
-    assert_eq!(manager.pending_approvals_count(), 1);
-
-    let approval_id = await_approval_identity(&manager, &thread.id, "tool_external_allow").await?;
-
-    // The snapshot a reconnecting client resumes from carries the same opaque
-    // capability as the event, so both routes to the ID agree.
-    let detail = manager.get_thread_detail(&thread.id).await?;
-    assert_eq!(detail.pending_approvals.len(), 1);
-    assert_eq!(detail.pending_approvals[0].id, approval_id);
-    assert_eq!(detail.pending_approvals[0].turn_id, _turn.id);
-    assert_eq!(detail.pending_approvals[0].tool_name, "exec_command");
-    // The snapshot keeps the raw correlator so a client that reloaded past the
-    // event can still attach this prompt to the tool row it gates.
-    assert_eq!(
-        detail.pending_approvals[0].tool_call_id.as_deref(),
-        Some("tool_external_allow")
-    );
-    assert_eq!(detail.pending_user_inputs.len(), 0);
-
-    let events = manager.events_since(&thread.id, None)?;
-    let approval_event = events
-        .iter()
-        .rev()
-        .find(|event| event.event == "approval.required")
-        .context("missing approval.required event")?;
-    assert_eq!(
-        approval_event
-            .payload
-            .get("intent_summary")
-            .and_then(Value::as_str),
-        Some("I will update the config file.")
-    );
-
-    // The raw provider ID is not a credential: only the minted one settles.
-    assert!(!manager.deliver_external_approval(
-        "tool_external_allow",
-        ExternalApprovalDecision::Allow { remember: false },
-    ));
-    assert_eq!(manager.pending_approvals_count(), 1);
-
-    assert!(manager.deliver_external_approval(
-        &approval_id,
-        ExternalApprovalDecision::Allow { remember: false },
-    ));
-    // The engine is still addressed by the provider's own call ID.
-    assert_eq!(
-        harness.recv_approval_event().await,
-        Some(MockApprovalEvent::Approved {
-            id: "tool_external_allow".to_string(),
-        })
-    );
-    assert_eq!(manager.pending_approvals_count(), 0);
-    assert!(
-        manager
-            .get_thread_detail(&thread.id)
-            .await?
-            .pending_approvals
-            .is_empty()
-    );
-    assert!(
-        manager.events_since(&thread.id, None)?.iter().any(|event| {
-            event.event == "approval.decided"
-                && event.payload.get("approval_id").and_then(Value::as_str)
-                    == Some(approval_id.as_str())
-                && event.payload.get("tool_call_id").and_then(Value::as_str)
-                    == Some("tool_external_allow")
-                && event.payload.get("decision").and_then(Value::as_str) == Some("allow")
-        }),
-        "approval.decided must correlate the opaque id with the tool call it settled"
-    );
-
-    harness
-        .tx_event
-        .send(EngineEvent::TurnComplete {
-            usage: Usage::default(),
-            parent_route_usage: Usage::default(),
-            routed_usage_dropped_records: 0,
-            status: TurnOutcomeStatus::Completed,
-            error: None,
-            tool_catalog: None,
-            base_url: None,
-        })
-        .await?;
     Ok(())
 }
 
@@ -15923,6 +16143,7 @@ async fn identical_raw_tool_call_ids_on_two_threads_stay_independently_gated() -
     let manager = test_manager(test_runtime_dir())?;
 
     let mut gated = Vec::new();
+    let mut inputs = std::collections::HashMap::new();
     for prompt in ["first gated turn", "second gated turn"] {
         let thread = manager
             .create_thread(CreateThreadRequest::default())
@@ -15941,6 +16162,8 @@ async fn identical_raw_tool_call_ids_on_two_threads_stay_independently_gated() -
             harness.rx_op.recv().await,
             Some(Op::SendMessage(TurnSpec { .. }))
         ));
+        let input =
+            json!({"command":"echo fixture","owner_prompt":prompt,"nested":{"values":[null,true]}});
         harness
             .tx_event
             .send(EngineEvent::ApprovalRequired {
@@ -15949,12 +16172,13 @@ async fn identical_raw_tool_call_ids_on_two_threads_stay_independently_gated() -
                 id: RAW_CALL_ID.to_string(),
                 tool_name: "exec_command".to_string(),
                 description: format!("gated {prompt}"),
-                input: serde_json::json!({}),
+                input: input.clone(),
                 intent_summary: None,
                 approval_force_prompt: false,
             })
             .await?;
         let approval_id = await_approval_identity(&manager, &thread.id, RAW_CALL_ID).await?;
+        inputs.insert(thread.id.clone(), input);
         gated.push((thread, harness, approval_id));
     }
 
@@ -15976,6 +16200,10 @@ async fn identical_raw_tool_call_ids_on_two_threads_stay_independently_gated() -
         // provider behavior this fix survives — while the capability that
         // settles each one is distinct.
         assert_eq!(pending[0].tool_call_id.as_deref(), Some(RAW_CALL_ID));
+        assert_eq!(pending[0].input.as_ref(), inputs.get(&thread.id));
+        let reloaded: PendingApprovalRequest =
+            serde_json::from_value(serde_json::to_value(&pending[0])?)?;
+        assert_eq!(reloaded.input.as_ref(), inputs.get(&thread.id));
     }
 
     // The shared correlator is visible to both clients and settles neither.
@@ -16016,6 +16244,7 @@ async fn identical_raw_tool_call_ids_on_two_threads_stay_independently_gated() -
         .pending_approvals;
     assert_eq!(still_pending.len(), 1);
     assert_eq!(still_pending[0].id, approval_b);
+    assert_eq!(still_pending[0].input.as_ref(), inputs.get(&thread_b.id));
     assert!(
         manager
             .get_thread_detail(&thread_a.id)
@@ -16124,6 +16353,7 @@ async fn runtime_approval_disposition_auto_paths_obey_force_and_posture() -> Res
         } else {
             None
         };
+        let input = json!({"approval_key":"extcall:ext:claimed","command":"echo fixture","nested":{"values":[null,true,3]}});
         harness
             .tx_event
             .send(EngineEvent::ApprovalRequired {
@@ -16133,7 +16363,7 @@ async fn runtime_approval_disposition_auto_paths_obey_force_and_posture() -> Res
                 tool_name: "bash".to_string(),
                 // Text and input cannot impersonate a Rust-minted origin key.
                 description: "Requested by extension:claimed".to_string(),
-                input: json!({"approval_key": "extcall:ext:claimed", "command": "echo fixture"}),
+                input: input.clone(),
                 intent_summary: None,
                 approval_force_prompt: forced,
             })
@@ -16165,6 +16395,17 @@ async fn runtime_approval_disposition_auto_paths_obey_force_and_posture() -> Res
         let detail = manager.get_thread_detail(&thread.id).await?;
         assert!(detail.pending_approvals.is_empty());
         assert_eq!(detail.approval_grants.len(), usize::from(remembered));
+        let required = manager
+            .events_since(&thread.id, None)?
+            .into_iter()
+            .find(|event| event.event == "approval.required");
+        if approved {
+            let required = required.context("automatic approval request receipt")?;
+            assert_eq!(required.payload.get("input"), Some(&input));
+            assert_eq!(required.payload["tool_call_id"], "policy-call");
+        } else {
+            assert!(required.is_none());
+        }
         let decided = manager
             .events_since(&thread.id, None)?
             .into_iter()

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

@@ -91,6 +91,7 @@ async fn shutdown_waits_actual_engine_exit_and_terminal_monitor_before_releasing
     let (engine, handle) = Engine::new_with_model_client(
         EngineConfig {
             workspace: root.path().into(),
+            session_id: Some(thread.id.clone()),
             model: thread.model.clone(),
             subagents_enabled: false,
             snapshots_enabled: false,
@@ -115,10 +116,11 @@ async fn shutdown_waits_actual_engine_exit_and_terminal_monitor_before_releasing
         blocked.await.unwrap();
         engine.run().await;
     });
-    runtime
-        .engine_workers
-        .lock()
-        .push((handle.clone(), retained_completion(worker)));
+    runtime.engine_workers.lock().push((
+        thread.id.clone(),
+        handle.clone(),
+        retained_completion(worker),
+    ));
     let turn = runtime
         .start_turn(
             &thread.id,
@@ -152,7 +154,7 @@ async fn shutdown_waits_actual_engine_exit_and_terminal_monitor_before_releasing
             .await
             .is_err()
     );
-    let completion = runtime.engine_workers.lock()[0].1.clone();
+    let completion = runtime.engine_workers.lock()[0].2.clone();
     tokio::time::timeout(Duration::from_secs(5), async {
         while completion.try_lock().is_ok() {
             sleep(Duration::from_millis(5)).await;

crates/tui/src/tools/subagent/completion_usage_tests.rs

@@ -87,6 +87,7 @@ async fn completion_usage_live_terminal_counts_grandchildren_and_continuations_o
                 parent_completion_tx: Some(tx),
                 mailbox: None,
                 event_tx: Some(event_tx),
+                event_delivery: None,
                 session_id: "workspace".into(),
             });
         let mut result = terminal_result(&manager, &root);

crates/tui/src/tools/subagent/mod.rs

@@ -1029,6 +1029,8 @@ pub struct AgentWorkerRecord {
     /// session-addressable without being guessed into a later conversation.
     #[serde(default, skip_serializing_if = "String::is_empty")]
     pub owner_session_id: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub runtime_origin_turn_id: Option<String>,
     #[serde(default = "default_subagent_actor_kind")]
     pub actor_kind: String,
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -1152,6 +1154,7 @@ impl AgentWorkerRecord {
             parent_run_id: spec.parent_run_id.clone(),
             spec,
             owner_session_id,
+            runtime_origin_turn_id: None,
             actor_kind: default_subagent_actor_kind(),
             follow_up,
             takeover,
@@ -2407,6 +2410,55 @@ impl SubAgentCompletion {
         })
     }
 }
+#[derive(Default)]
+pub(crate) struct TerminalEventDelivery {
+    pending: std::sync::atomic::AtomicUsize,
+    failed: std::sync::atomic::AtomicBool,
+    settled: tokio::sync::Notify,
+}
+
+impl TerminalEventDelivery {
+    pub(crate) fn mark_failed(&self) {
+        self.failed
+            .store(true, std::sync::atomic::Ordering::Release);
+    }
+
+    pub(crate) async fn wait(&self) -> Result<()> {
+        loop {
+            let settled = self.settled.notified();
+            tokio::pin!(settled);
+            settled.as_mut().enable();
+            if self.pending.load(std::sync::atomic::Ordering::Acquire) == 0 {
+                break;
+            }
+            settled.await;
+        }
+        anyhow::ensure!(
+            !self.failed.load(std::sync::atomic::Ordering::Acquire),
+            "a background worker lifecycle event could not reach its receiver or durable journal"
+        );
+        Ok(())
+    }
+}
+
+struct PendingTerminalEvent {
+    owner: Arc<TerminalEventDelivery>,
+    delivered: bool,
+}
+
+impl Drop for PendingTerminalEvent {
+    fn drop(&mut self) {
+        if !self.delivered {
+            self.owner
+                .failed
+                .store(true, std::sync::atomic::Ordering::Release);
+        }
+        self.owner
+            .pending
+            .fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
+        self.owner.settled.notify_waiters();
+    }
+}
 
 /// Live-only sinks needed to publish one terminal child outcome.
 ///
@@ -2422,6 +2474,7 @@ struct SubAgentTerminalDeliveryContext {
     parent_completion_tx: Option<mpsc::Sender<SubAgentCompletion>>,
     mailbox: Option<Mailbox>,
     event_tx: Option<mpsc::Sender<Event>>,
+    event_delivery: Option<Arc<TerminalEventDelivery>>,
     /// Shared session namespace (root session id), cloned down the spawn
     /// tree; over-budget final reports are spilled under it so the truncation
     /// footer can name a retrievable artifact.
@@ -2435,6 +2488,10 @@ impl SubAgentTerminalDeliveryContext {
             parent_completion_tx: runtime.parent_completion_tx.clone(),
             mailbox: runtime.mailbox.clone(),
             event_tx: runtime.event_tx.clone(),
+            event_delivery: runtime
+                .execution_limits
+                .as_ref()
+                .map(|limits| limits.terminal_events.clone()),
             session_id: runtime.context.state_namespace.clone(),
         }
     }
@@ -2465,7 +2522,7 @@ impl SubAgentTerminalDeliveryContext {
         }
 
         if let Some(event_tx) = self.event_tx.as_ref() {
-            send_terminal_event(
+            send_terminal_event_with_delivery(
                 event_tx,
                 Event::AgentComplete {
                     owner_session_id: self.session_id.clone(),
@@ -2478,6 +2535,7 @@ impl SubAgentTerminalDeliveryContext {
                     usage: result.usage.clone(),
                     display_name: Some(subagent_result_display_name(result)),
                 },
+                self.event_delivery.as_ref(),
             );
         }
     }
@@ -2494,20 +2552,53 @@ impl SubAgentTerminalDeliveryContext {
 /// lossy: delivered late it could land after the completion and resurrect a
 /// Running row, while a lost one is recovered by the completion itself.
 pub(crate) fn send_terminal_event<T: Send + 'static>(event_tx: &mpsc::Sender<T>, event: T) {
+    send_terminal_event_with_delivery(event_tx, event, None);
+}
+
+fn send_terminal_event_with_delivery<T: Send + 'static>(
+    event_tx: &mpsc::Sender<T>,
+    event: T,
+    delivery: Option<&Arc<TerminalEventDelivery>>,
+) {
     let event = match event_tx.try_send(event) {
-        Ok(()) | Err(mpsc::error::TrySendError::Closed(_)) => return,
+        Ok(()) => return,
+        Err(mpsc::error::TrySendError::Closed(_)) => {
+            if let Some(delivery) = delivery {
+                delivery
+                    .failed
+                    .store(true, std::sync::atomic::Ordering::Release);
+            }
+            return;
+        }
         Err(mpsc::error::TrySendError::Full(event)) => event,
     };
+    let pending = delivery.map(|owner| {
+        owner
+            .pending
+            .fetch_add(1, std::sync::atomic::Ordering::AcqRel);
+        PendingTerminalEvent {
+            owner: owner.clone(),
+            delivered: false,
+        }
+    });
     let tx = event_tx.clone();
     match tokio::runtime::Handle::try_current() {
         Ok(handle) => {
             handle.spawn(async move {
-                let _ = tx.send(event).await;
+                let mut pending = pending;
+                let delivered = tx.send(event).await.is_ok();
+                if let Some(pending) = pending.as_mut() {
+                    pending.delivered = delivered;
+                }
             });
         }
         Err(_) => {
             std::thread::spawn(move || {
-                let _ = tx.blocking_send(event);
+                let mut pending = pending;
+                let delivered = tx.blocking_send(event).is_ok();
+                if let Some(pending) = pending.as_mut() {
+                    pending.delivered = delivered;
+                }
             });
         }
     }
@@ -2799,6 +2890,63 @@ impl Drop for ForegroundChildRegistration {
     }
 }
 
+#[derive(Clone)]
+pub(crate) struct SubAgentExecutionLimits {
+    pub max_admitted: usize,
+    pub heartbeat_timeout: Duration,
+    pub default_max_steps: Option<u32>,
+    pub default_wall_time: Option<Duration>,
+    pub launch_gate: Arc<governor::DynamicGate>,
+    pub terminal_events: Arc<TerminalEventDelivery>,
+    pub admission: CancellationToken,
+}
+
+impl SubAgentExecutionLimits {
+    pub(crate) fn new(
+        max_agents: usize,
+        max_admitted: usize,
+        heartbeat_timeout: Duration,
+        launch_concurrency: usize,
+        default_max_steps: Option<u32>,
+        default_wall_time: Option<Duration>,
+    ) -> Self {
+        let max_agents = max_agents.clamp(1, MAX_SUBAGENTS);
+        Self {
+            max_admitted: max_admitted.clamp(max_agents, crate::config::MAX_SUBAGENT_ADMISSION),
+            heartbeat_timeout: if heartbeat_timeout.is_zero() {
+                Duration::from_secs(crate::config::DEFAULT_SUBAGENT_HEARTBEAT_TIMEOUT_SECS)
+            } else {
+                heartbeat_timeout
+            },
+            default_max_steps,
+            default_wall_time,
+            launch_gate: Arc::new(governor::DynamicGate::new(
+                launch_concurrency.clamp(1, max_agents),
+            )),
+            terminal_events: Arc::new(TerminalEventDelivery::default()),
+            admission: CancellationToken::new(),
+        }
+    }
+
+    pub(crate) fn update(
+        &mut self,
+        max_agents: usize,
+        max_admitted: usize,
+        heartbeat_timeout: Duration,
+        launch_concurrency: usize,
+    ) {
+        let max_agents = max_agents.clamp(1, MAX_SUBAGENTS);
+        self.max_admitted = max_admitted.clamp(max_agents, crate::config::MAX_SUBAGENT_ADMISSION);
+        self.heartbeat_timeout = if heartbeat_timeout.is_zero() {
+            Duration::from_secs(crate::config::DEFAULT_SUBAGENT_HEARTBEAT_TIMEOUT_SECS)
+        } else {
+            heartbeat_timeout
+        };
+        self.launch_gate
+            .set_capacity(launch_concurrency.clamp(1, max_agents));
+    }
+}
+
 #[derive(Clone)]
 pub struct SubAgentRuntime {
     pub client: CodewhaleClient,
@@ -2853,6 +3001,7 @@ pub struct SubAgentRuntime {
     /// Manager handle so children can recurse via `agent`. All agents
     /// at every depth share the same manager.
     pub manager: SharedSubAgentManager,
+    pub(crate) execution_limits: Option<SubAgentExecutionLimits>,
     /// Depth in the spawn tree. 0 = top-level user turn; 1 = direct child;
     /// etc. Children clone the parent runtime and increment this on spawn.
     pub spawn_depth: u32,
@@ -2999,6 +3148,7 @@ impl SubAgentRuntime {
             worker_profile: WorkerRuntimeProfile::for_role(FleetRole::Worker),
             event_tx,
             manager,
+            execution_limits: None,
             spawn_depth: 0,
             parent_agent_id: None,
             max_spawn_depth: DEFAULT_MAX_SPAWN_DEPTH,
@@ -3236,6 +3386,12 @@ impl SubAgentRuntime {
         self
     }
 
+    #[must_use]
+    pub(crate) fn with_execution_limits(mut self, limits: Option<SubAgentExecutionLimits>) -> Self {
+        self.execution_limits = limits;
+        self
+    }
+
     /// Preserve whether the parent session is using per-turn model routing.
     #[must_use]
     pub fn with_auto_model(mut self, auto_model: bool) -> Self {
@@ -3325,6 +3481,7 @@ impl SubAgentRuntime {
             worker_profile: self.worker_profile.clone(),
             event_tx: self.event_tx.clone(),
             manager: self.manager.clone(),
+            execution_limits: self.execution_limits.clone(),
             spawn_depth: self.spawn_depth.saturating_add(1),
             parent_agent_id: self.parent_agent_id.clone(),
             max_spawn_depth: self
@@ -3514,6 +3671,7 @@ pub struct SubAgent {
     pub needs_input: Option<SubAgentNeedsInput>,
     pub started_at: Instant,
     pub last_activity_at: Instant,
+    heartbeat_timeout: Option<Duration>,
     /// `None` = full registry inheritance, with approval-gated tools still
     /// blocked unless the parent runtime is auto-approved.
     /// `Some(list)` = explicit narrow allowlist (Custom agents, legacy).
@@ -3544,7 +3702,7 @@ pub struct SubAgent {
     requested_stop: Option<(SubAgentStatus, Option<String>)>,
     // An unresolved forced-stop retains the exact existing capacity/barrier
     // guards until owner teardown; it cannot be replayed as completed work.
-    unsettled_launch: Option<governor::DynamicGatePermit>,
+    unsettled_launch: Option<Vec<governor::DynamicGatePermit>>,
     unsettled_foreground: Option<ForegroundChildRegistration>,
 }
 
@@ -3583,6 +3741,7 @@ impl SubAgent {
             needs_input: None,
             started_at,
             last_activity_at: started_at,
+            heartbeat_timeout: None,
             allowed_tools,
             session_boot_id,
             owner_session_id: String::new(),
@@ -3661,7 +3820,7 @@ pub const COORDINATION_LOCK_TIMEOUT_MARKER: &str =
     "timed out acquiring delegated coordination lock";
 
 impl CoordinationProcessLock {
-    fn acquire(state_root: &Path) -> Result<Self> {
+    fn acquire(state_root: &Path, exclusive: bool) -> Result<Self> {
         let requested_root = normalize_subagent_workspace(state_root);
         let lock_dir = requested_root.join(".codewhale").join("state");
         // Refuse a linked `.codewhale` or `state` before creating anything:
@@ -3701,22 +3860,22 @@ impl CoordinationProcessLock {
         let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel(1);
         let (release_tx, release_rx) = std::sync::mpsc::sync_channel(0);
         let thread = std::thread::spawn(move || {
-            let lock = fd_lock::RwLock::new(lock_file);
-            // Changed from try_write (exclusive) to try_read (shared) so two
-            // sessions in the same state root can coexist without a sticky
-            // error — job rows still settle via atomic rename + sequence guard.
-            // Matches user intent: "no reason to have a lock like that".
-            match lock.try_read() {
-                Ok(guard) => {
-                    // Shared lock: multiple sessions coexist, no pid stamping needed.
-                    // The exclusive pid dance was for telling same-process handover
-                    // from foreign owner — with shared locks every holder is equal
-                    // and job rows settle via atomic rename, so we just hold the
-                    // shared flock for the session lifetime.
-                    let _guard = guard;
+            let mut lock = fd_lock::RwLock::new(lock_file);
+            let acquired = if exclusive {
+                lock.try_write().map(|guard| {
                     let _ = ready_tx.send(Ok::<(), String>(()));
+                    let _guard = guard;
                     let _ = release_rx.recv();
-                }
+                })
+            } else {
+                lock.try_read().map(|guard| {
+                    let _ = ready_tx.send(Ok::<(), String>(()));
+                    let _guard = guard;
+                    let _ = release_rx.recv();
+                })
+            };
+            match acquired {
+                Ok(()) => {}
                 Err(error) => {
                     let _ = ready_tx.send(Err(error.to_string()));
                 }
@@ -3810,8 +3969,10 @@ pub struct SubAgentManager {
     /// isolate it without changing child execution cwd or file authority.
     state_root: PathBuf,
     state_path: Option<PathBuf>,
-    coordination_process_lock: std::sync::Mutex<Option<CoordinationProcessLock>>,
+    coordination_process_lock: std::sync::Mutex<Option<Arc<CoordinationProcessLock>>>,
     coordination_process_lock_required: bool,
+    coordination_exclusive: bool,
+    workspace_writer_locks: HashMap<PathBuf, Arc<CoordinationProcessLock>>,
     /// Configured default per-child model-turn budget (`[subagents]
     /// default_max_steps`, #5324). `None` keeps the unbounded Fleet default;
     /// an explicit spawn `max_steps` still wins. Zero means unbounded.
@@ -4323,6 +4484,8 @@ impl SubAgentManager {
             state_path: None,
             coordination_process_lock: std::sync::Mutex::new(None),
             coordination_process_lock_required: false,
+            coordination_exclusive: false,
+            workspace_writer_locks: HashMap::new(),
             max_steps: None,
             wall_time: None,
             max_agents,
@@ -4387,6 +4550,16 @@ impl SubAgentManager {
         return Err("coordination workspace identity is unsupported on this platform".into());
         #[cfg(any(unix, windows))]
         {
+            if self.coordination_exclusive
+                && normalize_subagent_workspace(&lexical)
+                    != normalize_subagent_workspace(&self.state_root)
+                && !self.workspace_writer_locks.contains_key(&canonical_root)
+            {
+                let lock = CoordinationProcessLock::acquire(&lexical, true)
+                    .map_err(|error| error.to_string())?;
+                self.workspace_writer_locks
+                    .insert(canonical_root.clone(), Arc::new(lock));
+            }
             let receipt = CoordinationClaimScope {
                 canonical_root,
                 platform: platform.into(),
@@ -5480,7 +5653,9 @@ impl SubAgentManager {
             .coordination_process_lock
             .get_mut()
             .expect("coordination lock slot poisoned") =
-            CoordinationProcessLock::acquire(&self.state_root).ok();
+            CoordinationProcessLock::acquire(&self.state_root, self.coordination_exclusive)
+                .ok()
+                .map(Arc::new);
         self
     }
 
@@ -5499,9 +5674,9 @@ impl SubAgentManager {
         if slot.is_some() {
             return Ok(());
         }
-        match CoordinationProcessLock::acquire(&self.state_root) {
+        match CoordinationProcessLock::acquire(&self.state_root, self.coordination_exclusive) {
             Ok(lock) => {
-                *slot = Some(lock);
+                *slot = Some(Arc::new(lock));
                 Ok(())
             }
             Err(error) => Err(error.to_string()),
@@ -5648,9 +5823,18 @@ impl SubAgentManager {
             return Ok(std::thread::spawn(|| {}));
         };
         let state_root = self.state_root.clone();
+        let writer_locks = self
+            .coordination_process_lock
+            .lock()
+            .expect("coordination lock slot poisoned")
+            .iter()
+            .cloned()
+            .chain(self.workspace_writer_locks.values().cloned())
+            .collect::<Vec<_>>();
         // Spawn disk I/O off the write-lock hot path.  `payload` is fully
         // owned (cloned from `self.agents`) so it is `Send` and safe to move.
         let handle = std::thread::spawn(move || {
+            let _writer_locks = writer_locks;
             if let Err(err) = write_json_atomic(&state_root, &path, &payload) {
                 tracing::warn!(target: "subagent", ?err, "failed to persist sub-agent state");
             }
@@ -5845,6 +6029,7 @@ impl SubAgentManager {
                 needs_input: persisted.needs_input,
                 started_at,
                 last_activity_at: started_at,
+                heartbeat_timeout: None,
                 allowed_tools,
                 // Empty string when loading pre-#405 records; the
                 // manager treats that the same as a non-matching id —
@@ -6064,6 +6249,21 @@ impl SubAgentManager {
         spec: AgentWorkerSpec,
         owner_session_id: &str,
         precomputed_evidence: Option<DeliveryEvidence>,
+    ) {
+        self.register_worker_for_session_with_origin(
+            spec,
+            owner_session_id,
+            precomputed_evidence,
+            None,
+        );
+    }
+
+    pub(crate) fn register_worker_for_session_with_origin(
+        &mut self,
+        spec: AgentWorkerSpec,
+        owner_session_id: &str,
+        precomputed_evidence: Option<DeliveryEvidence>,
+        runtime_origin_turn_id: Option<String>,
     ) {
         let worker_id = spec.worker_id.clone();
         let now_ms = epoch_millis_now();
@@ -6073,6 +6273,7 @@ impl SubAgentManager {
             owner_session_id.to_string(),
             precomputed_evidence,
         );
+        record.runtime_origin_turn_id = runtime_origin_turn_id;
         self.push_worker_event(
             &mut record,
             AgentWorkerStatus::Starting,
@@ -8393,7 +8594,10 @@ impl SubAgentManager {
         agent.status == SubAgentStatus::Running
             && agent.task_handle.is_some()
             && agent.requested_stop.is_none()
-            && agent.last_activity_at.elapsed() >= self.running_heartbeat_timeout
+            && agent.last_activity_at.elapsed()
+                >= agent
+                    .heartbeat_timeout
+                    .unwrap_or(self.running_heartbeat_timeout)
     }
 
     pub fn touch(&mut self, agent_id: &str) -> bool {
@@ -8428,6 +8632,13 @@ impl SubAgentManager {
         options: SubAgentSpawnOptions,
         precomputed_delivery_evidence: Option<DeliveryEvidence>,
     ) -> Result<SubAgentResult> {
+        if runtime
+            .execution_limits
+            .as_ref()
+            .is_some_and(|limits| limits.admission.is_cancelled())
+        {
+            return Err(anyhow!("Runtime worker admission is closed"));
+        }
         let mut origin = match options.origin.as_ref() {
             Some(origin) => origin.clone(),
             None => self
@@ -8469,6 +8680,16 @@ impl SubAgentManager {
         self.cleanup(COMPLETED_AGENT_RETENTION);
 
         self.check_admission_capacity()?;
+        if let Some(limits) = runtime.execution_limits.as_ref() {
+            let admitted = self.live_count_for_session(&runtime.context.state_namespace);
+            if admitted >= limits.max_admitted {
+                return Err(anyhow!(
+                    "Sub-agent route admission limit reached (max_admitted {}, admitted {}). Wait for queued/running agents to finish or cancel unneeded agents.",
+                    limits.max_admitted,
+                    admitted
+                ));
+            }
+        }
         runtime.max_spawn_depth = runtime
             .max_spawn_depth
             .min(runtime.worker_profile.max_spawn_depth);
@@ -8501,7 +8722,14 @@ impl SubAgentManager {
             narrow_optional_limit(runtime.max_output_tokens, options.max_output_tokens);
         let mut max_steps = narrow_model_steps(
             runtime.worker_profile.max_steps,
-            resolve_max_steps(agent_type.clone(), options.max_steps, self.max_steps),
+            resolve_max_steps(
+                agent_type.clone(),
+                options.max_steps,
+                runtime
+                    .execution_limits
+                    .as_ref()
+                    .map_or(self.max_steps, |limits| limits.default_max_steps),
+            ),
         );
         if let Some(source) = options
             .resume_from_agent_id
@@ -8537,10 +8765,14 @@ impl SubAgentManager {
         // `wall_time_secs` may raise it up to MAX_CHILD_WALL_TIME. An operator
         // `[subagents] default_wall_time_secs` stays a ceiling, and inherited
         // profiles and deadlines still only narrow.
+        let default_wall_time = runtime
+            .execution_limits
+            .as_ref()
+            .map_or(self.wall_time, |limits| limits.default_wall_time);
         let wall_time = options
             .wall_time
-            .unwrap_or_else(|| self.wall_time.unwrap_or(DEFAULT_CHILD_WALL_TIME))
-            .min(self.wall_time.unwrap_or(MAX_CHILD_WALL_TIME))
+            .unwrap_or_else(|| default_wall_time.unwrap_or(DEFAULT_CHILD_WALL_TIME))
+            .min(default_wall_time.unwrap_or(MAX_CHILD_WALL_TIME))
             .min(
                 runtime
                     .worker_profile
@@ -8639,6 +8871,7 @@ impl SubAgentManager {
             // live holder, so a reused name never resolves ambiguously.
             if let Some(existing) = self.agents.values().find(|existing| {
                 existing.session_name == name
+                    && self.agent_is_owned_by_session(existing, &runtime.context.state_namespace)
                     && !self.is_from_prior_session(existing)
                     && !subagent_status_is_settled(&existing.status)
             }) {
@@ -8733,7 +8966,18 @@ impl SubAgentManager {
                 ));
             }
         }
+        let runtime_origin_turn_id = if runtime.context.runtime.active_thread_id.as_deref()
+            == Some(runtime.context.state_namespace.as_str())
+        {
+            runtime
+                .runtime_usage_lease
+                .as_ref()
+                .map(|lease| lease.owner().to_string())
+        } else {
+            None
+        };
         let durable_registration = write_capable
+            || runtime_origin_turn_id.is_some()
             || continuation_from.is_some()
             || origin
                 .selected_workspace
@@ -8918,12 +9162,17 @@ impl SubAgentManager {
                 }
             };
         agent.owner_session_id = runtime.context.state_namespace.clone();
+        agent.heartbeat_timeout = runtime
+            .execution_limits
+            .as_ref()
+            .map(|limits| limits.heartbeat_timeout);
         agent.stop_token = Some(runtime.cancel_token.clone());
         agent.terminal_delivery = Some(SubAgentTerminalDeliveryContext::from_runtime(&runtime));
-        self.register_worker_for_session(
+        self.register_worker_for_session_with_origin(
             worker_spec,
             &runtime.context.state_namespace,
             precomputed_delivery_evidence,
+            runtime_origin_turn_id,
         );
         self.install_worker_origin(&agent_id, origin);
 
@@ -9315,7 +9564,11 @@ impl SubAgentManager {
             let millis =
                 |duration: Duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX);
             snap.idle_ms = Some(millis(agent.last_activity_at.elapsed()));
-            snap.heartbeat_timeout_ms = Some(millis(self.running_heartbeat_timeout));
+            snap.heartbeat_timeout_ms = Some(millis(
+                agent
+                    .heartbeat_timeout
+                    .unwrap_or(self.running_heartbeat_timeout),
+            ));
         }
         if let Some(record) = self.worker_records.get(&agent.id) {
             snap.usage = Some(record.usage.clone());
@@ -9502,11 +9755,17 @@ impl SubAgentManager {
                 agent.status == SubAgentStatus::Running
                     && !agent.completion_claimed
                     && agent.requested_stop.is_none()
-                    && agent.last_activity_at.elapsed() >= timeout
+                    && agent.last_activity_at.elapsed()
+                        >= agent.heartbeat_timeout.unwrap_or(timeout)
             })
             .map(|agent| agent.id.clone())
             .collect::<Vec<_>>();
         for agent_id in stale_agent_ids {
+            let timeout = self
+                .agents
+                .get(&agent_id)
+                .and_then(|agent| agent.heartbeat_timeout)
+                .unwrap_or(timeout);
             let orphan = self
                 .agents
                 .get(&agent_id)
@@ -10972,6 +11231,44 @@ pub fn new_shared_subagent_manager_with_timeout(
     )
 }
 
+pub(crate) fn new_shared_runtime_subagent_manager(
+    workspace: PathBuf,
+    workers: usize,
+) -> Result<SharedSubAgentManager> {
+    type RuntimeManagerRegistry =
+        parking_lot::Mutex<HashMap<PathBuf, (usize, std::sync::Weak<RwLock<SubAgentManager>>)>>;
+    static MANAGERS: std::sync::OnceLock<RuntimeManagerRegistry> = std::sync::OnceLock::new();
+    let workers = workers.clamp(1, MAX_SUBAGENTS);
+    let state_path = default_state_path(&workspace)?;
+    let key = checked_subagent_state_path(&workspace, &state_path)?;
+    let mut managers = MANAGERS
+        .get_or_init(|| parking_lot::Mutex::new(HashMap::new()))
+        .lock();
+    managers.retain(|_, (_, manager)| manager.strong_count() > 0);
+    if let Some((capacity, manager)) = managers.get(&key)
+        && let Some(manager) = manager.upgrade()
+    {
+        anyhow::ensure!(
+            *capacity == workers,
+            "Runtime worker authority already has aggregate capacity {capacity}; requested {workers}. Attach to its owner or stop it before changing workers."
+        );
+        return Ok(manager);
+    }
+    let mut manager = SubAgentManager::new(workspace.clone(), workers)
+        .with_admission_limit(workers)
+        .with_launch_concurrency(workers)
+        .with_state_path(state_path);
+    manager.coordination_exclusive = true;
+    manager = manager.require_coordination_process_lock();
+    manager
+        .ensure_coordination_process_lock()
+        .map_err(anyhow::Error::msg)?;
+    manager.load_state()?;
+    let manager = Arc::new(RwLock::new(manager));
+    managers.insert(key, (workers, Arc::downgrade(&manager)));
+    Ok(manager)
+}
+
 /// Create a shared sub-agent manager with an explicit control-plane state root.
 ///
 /// `workspace` remains the child execution and file-authority root. The worker
@@ -11023,8 +11320,8 @@ pub fn new_shared_subagent_manager_with_state_root_and_timeout(
     // empty ledger that would be written straight over the real one the
     // moment the first session exited and the flock became acquirable
     // (owner report, 2026-08-04). Read unconditionally; the write gate on
-    // `persist_state`/`persist_state_synchronously` is what keeps the file
-    // single-writer.
+    // `persist_state`/`persist_state_synchronously` still checks this
+    // manager's coordinator admission before publishing.
     if let Err(err) = manager.load_state() {
         // Routed through tracing instead of stderr — see comment in
         // `persist_state_best_effort` above.
@@ -13727,12 +14024,20 @@ async fn run_subagent_task_inner(mut task: SubAgentTask) {
     // slot after waiting starts its work clock then (#6015), still bounded by
     // any inherited deadline. Cancellation while queued is handled by
     // the admitted child actor before it emits Started/Starting.
-    let mut _launch_permit = None;
+    let mut _launch_permits = Vec::new();
     let mut launch_wait_timed_out = false;
     let mut launched_from_queue = false;
-    if let Some(gate) = task.launch_gate.as_ref() {
+    let owner_gate = (task.runtime.spawn_depth == 1)
+        .then(|| {
+            task.runtime
+                .execution_limits
+                .as_ref()
+                .map(|limits| &limits.launch_gate)
+        })
+        .flatten();
+    for gate in owner_gate.into_iter().chain(task.launch_gate.as_ref()) {
         match Arc::clone(gate).try_acquire() {
-            Some(permit) => _launch_permit = Some(permit),
+            Some(permit) => _launch_permits.push(permit),
             None => {
                 match tokio::time::timeout_at(
                     deadline.into(),
@@ -13741,10 +14046,17 @@ async fn run_subagent_task_inner(mut task: SubAgentTask) {
                 .await
                 {
                     Ok(permit) => {
-                        launched_from_queue = permit.is_some();
-                        _launch_permit = permit;
+                        if let Some(permit) = permit {
+                            launched_from_queue = true;
+                            _launch_permits.push(permit);
+                        } else {
+                            break;
+                        }
+                    }
+                    Err(_) => {
+                        launch_wait_timed_out = true;
+                        break;
                     }
-                    Err(_) => launch_wait_timed_out = true,
                 }
             }
         }
@@ -13883,7 +14195,7 @@ async fn run_subagent_task_inner(mut task: SubAgentTask) {
                     Some(note.clone()),
                 )
             });
-            agent.unsettled_launch = _launch_permit.take();
+            agent.unsettled_launch = Some(std::mem::take(&mut _launch_permits));
             agent.unsettled_foreground = task._foreground_child_registration.take();
         }
         if let Some(record) = manager.worker_records.get_mut(&agent_id) {
@@ -19421,9 +19733,14 @@ mod owner_scoped_coordination_tests {
         let selected = root.join("selected");
         fs::create_dir_all(original.join("src")).unwrap();
         fs::create_dir_all(selected.join("src")).unwrap();
-        let mut manager = SubAgentManager::new(original.clone(), 2);
+        let authority = new_shared_runtime_subagent_manager(original.clone(), 2).unwrap();
+        let mut manager = authority.try_write().unwrap();
         admit(&mut manager, &original);
         admit(&mut manager, &selected);
+        let independent =
+            SubAgentManager::new(selected.clone(), 2).require_coordination_process_lock();
+        assert!(independent.ensure_coordination_process_lock().is_err());
+        assert!(independent.persist_state_synchronously().is_err());
         manager
             .preflight_worker_coordination(&writer("original", &original, "src"))
             .unwrap();

crates/tui/src/tools/subagent/tests.rs

@@ -1197,7 +1197,12 @@ fn coordination_hot_paths_count_only_active_authoritative_owners() {
 #[test]
 fn headless_worker_registration_enforces_live_claims_and_projects_context() {
     let tmp = tempdir().expect("tempdir");
-    let mut manager = SubAgentManager::new(tmp.path().to_path_buf(), 8);
+    let authority_a = new_shared_runtime_subagent_manager(tmp.path().to_path_buf(), 8)
+        .expect("admit Runtime writer");
+    let authority_b = new_shared_runtime_subagent_manager(tmp.path().join("."), 8)
+        .expect("reuse canonical Runtime writer");
+    assert!(Arc::ptr_eq(&authority_a, &authority_b));
+    let mut manager = authority_a.try_write().expect("first authority lock");
     manager
         .record_coordination_decision(DecisionRecord {
             decision_id: "decision-src-a".into(),
@@ -1218,7 +1223,7 @@ fn headless_worker_registration_enforces_live_claims_and_projects_context() {
         spec.role = Some("worker".into());
         spec.runtime_profile = WorkerRuntimeProfile::for_role(FleetRole::Builder);
         spec.launch_manifest = Some(ChildLaunchManifest {
-            owner_session: "fleet-run".into(),
+            owner_session: format!("owner-{id}"),
             child_id: id.into(),
             profile: spec.runtime_profile.clone(),
             prompt: spec.objective.clone(),
@@ -1237,11 +1242,14 @@ fn headless_worker_registration_enforces_live_claims_and_projects_context() {
     };
 
     let worker_a = worker("worker-a", "src/a");
+    let origin_a = manager
+        .capture_direct_worker_origin(tmp.path(), "owner-worker-a")
+        .expect("capture first owner");
     manager
-        .preflight_worker_coordination(&worker_a)
+        .preflight_worker_coordination_in_origin(&worker_a, &origin_a)
         .expect("first worker preflight");
     manager
-        .register_worker_with_coordination(worker_a)
+        .register_worker_with_coordination_in_origin(worker_a, origin_a)
         .expect("first worker registration");
     let record = manager
         .get_worker_record("worker-a")
@@ -1257,21 +1265,82 @@ fn headless_worker_registration_enforces_live_claims_and_projects_context() {
             .contains("src-a-contract")
     );
 
+    let origin_b = manager
+        .capture_direct_worker_origin(tmp.path(), "owner-worker-b")
+        .expect("capture overlapping owner");
     let overlap = manager
-        .preflight_worker_coordination(&worker("worker-b", "src/a/nested"))
+        .preflight_worker_coordination_in_origin(&worker("worker-b", "src/a/nested"), &origin_b)
         .expect_err("overlapping live Fleet writer must remain queued");
     assert!(overlap.contains("worker-a"), "{overlap}");
     assert_eq!(manager.coordination.contentions.len(), 1);
     assert!(manager.get_worker_record("worker-b").is_none());
 
     let worker_c = worker("worker-c", "src/b");
+    let origin_c = manager
+        .capture_direct_worker_origin(tmp.path(), "owner-worker-c")
+        .expect("capture non-overlapping owner");
     manager
-        .preflight_worker_coordination(&worker_c)
+        .preflight_worker_coordination_in_origin(&worker_c, &origin_c)
         .expect("non-overlapping worker preflight");
     manager
-        .register_worker_with_coordination(worker_c)
+        .register_worker_with_coordination_in_origin(worker_c, origin_c)
         .expect("non-overlapping workers may proceed concurrently");
     assert_eq!(manager.coordination.write_claims.len(), 2);
+    manager
+        .worker_records
+        .get_mut("worker-a")
+        .unwrap()
+        .runtime_origin_turn_id = Some("turn-owner-a".into());
+    manager
+        .worker_records
+        .get_mut("worker-c")
+        .unwrap()
+        .runtime_origin_turn_id = Some("turn-owner-c".into());
+    manager
+        .persist_state_synchronously()
+        .expect("persist both owners");
+    drop(manager);
+    let persisted = load_persisted_agent_worker_records_with_state_root(tmp.path(), tmp.path())
+        .expect("read shared durable history");
+    assert_eq!(persisted.len(), 2);
+    assert!(
+        persisted
+            .iter()
+            .any(|record| record.owner_session_id == "owner-worker-a")
+    );
+    assert!(
+        persisted
+            .iter()
+            .any(|record| record.owner_session_id == "owner-worker-c")
+    );
+    let independent =
+        SubAgentManager::new(tmp.path().to_path_buf(), 8).require_coordination_process_lock();
+    assert!(independent.ensure_coordination_process_lock().is_err());
+    assert!(independent.persist_state_synchronously().is_err());
+    drop(independent);
+    drop(authority_a);
+    drop(authority_b);
+    let reopened = new_shared_runtime_subagent_manager(tmp.path().to_path_buf(), 8)
+        .expect("reopen after actual writer release");
+    let reopened = reopened.try_read().expect("reopened owner lock");
+    assert!(reopened.get_worker_record("worker-a").is_some());
+    assert!(reopened.get_worker_record("worker-c").is_some());
+    assert_eq!(
+        reopened
+            .get_worker_record("worker-a")
+            .unwrap()
+            .runtime_origin_turn_id
+            .as_deref(),
+        Some("turn-owner-a")
+    );
+    assert_eq!(
+        reopened
+            .get_worker_record("worker-c")
+            .unwrap()
+            .runtime_origin_turn_id
+            .as_deref(),
+        Some("turn-owner-c")
+    );
 }
 
 #[test]
@@ -6952,6 +7021,9 @@ async fn full_event_channel_still_delivers_agent_complete() {
         .expect("fill the only slot");
     let mut runtime = runtime_with_depth(1, None);
     runtime.event_tx = Some(event_tx);
+    let limits = SubAgentExecutionLimits::new(2, 2, Duration::from_secs(60), 2, None, None);
+    let delivery = limits.terminal_events.clone();
+    runtime.execution_limits = Some(limits);
     agent.terminal_delivery = Some(SubAgentTerminalDeliveryContext::from_runtime(&runtime));
     manager.agents.insert(agent_id.clone(), agent);
     manager.register_worker(make_worker_spec(&agent_id, tmp.path().to_path_buf()));
@@ -6964,6 +7036,9 @@ async fn full_event_channel_still_delivers_agent_complete() {
         "the roster leaves Running"
     );
 
+    let waiting = delivery.wait();
+    tokio::pin!(waiting);
+    assert!(futures_util::poll!(&mut waiting).is_pending());
     let filler = event_rx.recv().await.expect("filler event");
     assert!(matches!(filler, Event::Status { .. }));
     let delivered = tokio::time::timeout(Duration::from_secs(5), event_rx.recv())
@@ -6974,6 +7049,18 @@ async fn full_event_channel_still_delivers_agent_complete() {
         &delivered,
         Event::AgentComplete { id, outcome: Some(SubAgentStatus::Cancelled), .. } if id == &agent_id
     ));
+    waiting.await.expect("actual terminal send completed");
+    let failed = Arc::new(TerminalEventDelivery::default());
+    let (tx, rx) = mpsc::channel(1);
+    tx.try_send(Event::status("full")).unwrap();
+    send_terminal_event_with_delivery(&tx, Event::status("terminal"), Some(&failed));
+    drop(rx);
+    assert!(
+        tokio::time::timeout(Duration::from_secs(5), failed.wait())
+            .await
+            .expect("closed sender settles without hanging")
+            .is_err()
+    );
 }
 
 /// #6560 D02-06: a nested parent's completion inbox is bounded and has no
@@ -11853,7 +11940,7 @@ fn coordination_lock_refuses_linked_codewhale_without_writing_through_it() {
     std::fs::create_dir_all(&outside).expect("mkdir outside");
     std::os::unix::fs::symlink(&outside, workspace.join(".codewhale")).expect("link .codewhale");
 
-    let Err(err) = CoordinationProcessLock::acquire(&workspace) else {
+    let Err(err) = CoordinationProcessLock::acquire(&workspace, false) else {
         panic!("a linked .codewhale must be refused");
     };
     let message = format!("{err:#}");
@@ -15518,6 +15605,7 @@ pub(crate) fn stub_runtime() -> SubAgentRuntime {
         worker_profile: WorkerRuntimeProfile::for_role(FleetRole::Worker),
         event_tx: None,
         manager: new_shared_subagent_manager(workspace, 5),
+        execution_limits: None,
         spawn_depth: 0,
         max_spawn_depth: DEFAULT_MAX_SPAWN_DEPTH,
         cancel_token: CancellationToken::new(),
@@ -18207,10 +18295,14 @@ async fn launch_gate_queues_extra_direct_children() {
     runtime.context = ToolContext::new(tmp.path());
     runtime.mailbox = Some(mailbox);
 
-    let gate = Arc::new(governor::DynamicGate::new(1));
-    let held_launch_permit = Arc::clone(&gate)
+    runtime.spawn_depth = 1;
+    let limits = SubAgentExecutionLimits::new(1, 4, Duration::from_secs(60), 1, None, None);
+    let held_launch_permit = limits
+        .launch_gate
         .try_acquire()
-        .expect("test holds the single launch permit");
+        .expect("hold owner route slot");
+    runtime.execution_limits = Some(limits);
+    let gate = manager.read().await.launch_gate.clone();
     let spawn = |agent_id: &str, gate: Option<Arc<governor::DynamicGate>>| {
         let (input_tx, input_rx) = mpsc::unbounded_channel();
         let agent = SubAgent::new(
@@ -18286,6 +18378,11 @@ async fn launch_gate_queues_extra_direct_children() {
         queued.is_ok(),
         "second child must publish a visible queued reason: {messages:?}"
     );
+    assert_eq!(
+        gate.available_permits(),
+        4,
+        "a route-blocked owner must not hold an aggregate slot"
+    );
     drop(held_launch_permit);
 
     let collected = tokio::time::timeout(Duration::from_secs(5), async {