refactor(subagent): move delivery git work off the manager write lock (#6210)

Completion path: finish_terminal_result no longer runs the git trio +
fingerprints synchronously. ensure_worker_delivery_verified snapshots
inputs under a read lock, computes in spawn_blocking, and stores under a
follow-up write lock (idempotent via DeliveryEvidence.checked). The
natural/panic epilogues ensure before the lock so fan-in completions and
the terminal persist stay fresh; Stop/interrupt/close/stale commits heal
on the next detail projection.

Spawn path: spawn_subagent_from_input fingerprints the baseline pre-lock
(assuming write; registration discards it when the resolved spec is
read-only, and skips capture when the parent ceiling denies writes).
Resume/Fleet/test paths keep the inline capture.

Tests: deferred-semantics updates plus pending/ensure-idempotence/
no-op/heal/adopt/discard/fallback coverage.

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

@@ -174,7 +174,7 @@ async fn fixture(mode: &'static str, first_tokens: u64, max_steps: u32) -> Fixtu
     agent.status = SubAgentStatus::Running;
     {
         let mut guard = manager.write().await;
-        guard.register_worker_for_session(spec, &runtime.context.state_namespace);
+        guard.register_worker_for_session(spec, &runtime.context.state_namespace, None);
         guard.agents.insert("report-worker".to_string(), agent);
     }
     let task = tokio::spawn(run_subagent_task(SubAgentTask {

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

@@ -346,6 +346,7 @@ async fn completion_usage_counts_real_manifest_only_root_fork() {
                 checkpoint_continuation: false,
                 ..Default::default()
             },
+            None,
         )
         .unwrap();
     let record = guard.worker_records.get_mut(&fork.agent_id).unwrap();

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

@@ -616,7 +616,9 @@ impl ToolSpec for AgentsInterruptTool {
             let manager = self.manager.read().await;
             manager.get_worker_record_for_session(&context.state_namespace, &snapshot.agent_id)
         };
-        let projection = subagent_session_projection(snapshot, false, context, worker_record).await;
+        let projection =
+            subagent_session_projection(&self.manager, snapshot, false, context, worker_record)
+                .await;
         let payload = json!({
             "action": "interrupt",
             "agent_id": projection.agent_id,

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

@@ -1,7 +1,10 @@
 //! Delivery evidence replacing the old prose-verb/git-status heuristic.
 //! The worker ledger retains the spawn baseline; this module only reads files.
 
-use super::{AgentRunVerificationSummary, AgentWorkerSpec, normalize_claim_path};
+use super::{
+    AgentRunVerificationSummary, AgentWorkerSpec, default_agent_run_verification,
+    normalize_claim_path,
+};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest as _, Sha256};
 use std::collections::{BTreeMap, BTreeSet};
@@ -120,14 +123,19 @@ fn fingerprint(root: &Path, relative: &str) -> Option<String> {
 
 impl DeliveryEvidence {
     pub(super) fn capture(spec: &AgentWorkerSpec) -> Self {
-        let baseline = spec
-            .runtime_profile
-            .permissions
-            .write
+        Self::capture_for_handle(&spec.workspace, spec.runtime_profile.permissions.write)
+    }
+
+    /// Baseline capture that needs only the workspace and write permission —
+    /// the two spec fields the baseline actually reads. The async spawn path
+    /// calls this in `spawn_blocking` BEFORE the manager write lock (#6210)
+    /// and threads the evidence through registration, so git + file
+    /// fingerprints never run under the lock.
+    pub(super) fn capture_for_handle(workspace: &Path, write: bool) -> Self {
+        let baseline = write
             .then(|| {
                 let root =
-                    String::from_utf8(git(&spec.workspace, &["rev-parse", "--show-toplevel"])?)
-                        .ok()?;
+                    String::from_utf8(git(workspace, &["rev-parse", "--show-toplevel"])?).ok()?;
                 let root = PathBuf::from(root.trim());
                 let head = git(&root, &["rev-parse", "--verify", "HEAD"])
                     .and_then(|bytes| String::from_utf8(bytes).ok())
@@ -483,3 +491,61 @@ pub(super) fn verify_changes(
         deliverables: Vec::new(),
     })
 }
+
+/// Everything delivery verification needs, snapshotted under a read lock.
+/// `allowed[i]` is the write-scope verdict for `deliverables[i]`. The compute
+/// half runs in `spawn_blocking` with no manager lock held (#6210).
+#[derive(Debug, Clone)]
+pub(super) struct DeliveryVerificationInputs {
+    pub evidence: DeliveryEvidence,
+    pub workspace: PathBuf,
+    pub result_text: String,
+    pub write_perm: bool,
+    pub deliverables: Vec<String>,
+    pub allowed: Vec<bool>,
+}
+
+/// Pure compute half of worker delivery verification: the git trio +
+/// fingerprints (`changed_paths`), claim comparison, and per-deliverable
+/// presence checks. Runs off the manager lock; the caller stores the summary.
+pub(super) fn compute_delivery_verification(
+    inputs: &DeliveryVerificationInputs,
+) -> AgentRunVerificationSummary {
+    let changed = inputs.evidence.changed_paths(&inputs.workspace);
+    let mut verification = verify_changes(
+        &inputs.result_text,
+        inputs.write_perm,
+        &inputs.evidence,
+        changed.as_ref(),
+        &inputs.deliverables.iter().cloned().collect(),
+    )
+    .unwrap_or_else(default_agent_run_verification);
+    verification.deliverables = inputs
+        .deliverables
+        .iter()
+        .zip(inputs.allowed.iter())
+        .map(|(path, allowed)| check_deliverable(&inputs.workspace, path, *allowed))
+        .collect();
+    let missing = verification
+        .deliverables
+        .iter()
+        .filter(|verdict| verdict.status != "present")
+        .map(|verdict| format!("{} ({})", verdict.path, verdict.status))
+        .collect::<Vec<_>>();
+    if !missing.is_empty() {
+        let prior = if verification.status == "claim_mismatch" {
+            format!(" {}", verification.summary)
+        } else {
+            String::new()
+        };
+        verification.status = "deliverable_missing".to_string();
+        verification.summary = format!(
+            "Declared deliverables not produced as non-empty files in the worker write scope: {}.{prior}",
+            missing.join(", ")
+        );
+    } else if !inputs.deliverables.is_empty() && verification.status == "self_report_only" {
+        verification.status = "deliverables_present".to_string();
+        verification.summary = "Declared files exist and are non-empty inside the worker write scope; their contents remain a worker self-report.".to_string();
+    }
+    verification
+}

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

@@ -70,6 +70,14 @@ fn complete(manager: &mut SubAgentManager, id: &str, report: &str) -> AgentRunVe
     result.status = SubAgentStatus::Completed;
     result.result = Some(report.into());
     manager.complete_worker_from_result(id, &result);
+    // Deferred verification (#6210): the commit leaves verification pending;
+    // run the same snapshot→compute→store halves `ensure` runs off the lock.
+    if !manager.worker_records[id].delivery_evidence.checked
+        && let Some(inputs) = manager.delivery_verification_inputs(id, &result)
+    {
+        let verification = delivery::compute_delivery_verification(&inputs);
+        manager.store_delivery_verification(id, verification);
+    }
     manager.worker_records[id].verification.clone()
 }
 
@@ -540,3 +548,136 @@ async fn enforced_readonly_python_queries_sqlite_under_a_live_peer_write_claim()
         );
     }
 }
+
+/// Deferred-verification contract (#6210): the terminal commit stores the
+/// worker projection but leaves verification pending for `ensure`.
+#[test]
+fn terminal_commit_leaves_delivery_verification_pending() {
+    let tmp = tempdir().unwrap();
+    repository(tmp.path());
+    let (mut manager, id) = worker(tmp.path(), true, &["report.md"], &["."]);
+    let mut result = manager.get_result(&id).unwrap();
+    result.status = SubAgentStatus::Completed;
+    result.result = Some("Finished the research.".into());
+    manager.complete_worker_from_result(&id, &result);
+    let record = manager.worker_records.get(&id).unwrap();
+    assert!(!record.delivery_evidence.checked);
+    assert_eq!(record.verification.status, "self_report_only");
+    assert_eq!(
+        record.result_summary.as_deref(),
+        Some("Finished the research.")
+    );
+}
+
+#[tokio::test]
+async fn ensure_worker_delivery_verified_stores_verdicts_and_is_idempotent() {
+    let tmp = tempdir().unwrap();
+    repository(tmp.path());
+    let (manager, id) = worker(tmp.path(), true, &["report.md"], &["."]);
+    let manager = Arc::new(RwLock::new(manager));
+    let result = {
+        let mut guard = manager.write().await;
+        let mut result = guard.get_result(&id).unwrap();
+        result.status = SubAgentStatus::Completed;
+        result.result = Some("Finished the research.".into());
+        guard.complete_worker_from_result(&id, &result);
+        assert!(!guard.worker_records[&id].delivery_evidence.checked);
+        result
+    };
+    ensure_worker_delivery_verified(&manager, &id, &result).await;
+    let first = manager.read().await.worker_records[&id]
+        .verification
+        .clone();
+    assert_eq!(first.status, "deliverable_missing");
+    assert!(
+        manager.read().await.worker_records[&id]
+            .delivery_evidence
+            .checked
+    );
+    // A second call is a no-op even with a different report.
+    let mut other = result.clone();
+    other.result = Some("CHANGES: src/lib.rs".into());
+    ensure_worker_delivery_verified(&manager, &id, &other).await;
+    assert_eq!(manager.read().await.worker_records[&id].verification, first);
+}
+
+#[tokio::test]
+async fn ensure_worker_delivery_verified_ignores_running_missing_and_checked() {
+    let tmp = tempdir().unwrap();
+    repository(tmp.path());
+    let (manager, id) = worker(tmp.path(), true, &[], &["src"]);
+    let manager = Arc::new(RwLock::new(manager));
+    let running = manager.read().await.get_result(&id).unwrap();
+    assert_eq!(running.status, SubAgentStatus::Running);
+    ensure_worker_delivery_verified(&manager, &id, &running).await;
+    assert!(
+        !manager.read().await.worker_records[&id]
+            .delivery_evidence
+            .checked
+    );
+    // A missing worker id is a silent no-op.
+    let mut missing = running.clone();
+    missing.agent_id = "agent_missing".to_string();
+    missing.status = SubAgentStatus::Completed;
+    ensure_worker_delivery_verified(&manager, "agent_missing", &missing).await;
+    // A checked record keeps its stored verdict.
+    let mut done = running.clone();
+    done.status = SubAgentStatus::Completed;
+    done.result = Some("CHANGES: src/lib.rs".into());
+    ensure_worker_delivery_verified(&manager, &id, &done).await;
+    assert_eq!(
+        manager.read().await.worker_records[&id].verification.status,
+        "claim_mismatch"
+    );
+    let mut changed_mind = done.clone();
+    changed_mind.result = Some("CHANGES: None".into());
+    ensure_worker_delivery_verified(&manager, &id, &changed_mind).await;
+    assert_eq!(
+        manager.read().await.worker_records[&id].verification.status,
+        "claim_mismatch"
+    );
+}
+
+/// Read-side backstop (#6210): a terminal detail projection heals a
+/// verification left pending by a Stop/interrupt/close/stale commit.
+#[tokio::test]
+async fn detail_projection_heals_pending_delivery_verification() {
+    let tmp = tempdir().unwrap();
+    repository(tmp.path());
+    let (manager, id) = worker(tmp.path(), true, &["report.md"], &["."]);
+    let manager = Arc::new(RwLock::new(manager));
+    let result = {
+        let mut guard = manager.write().await;
+        let mut result = guard.get_result(&id).unwrap();
+        result.status = SubAgentStatus::Completed;
+        result.result = Some("Finished the research.".into());
+        guard.complete_worker_from_result(&id, &result);
+        result
+    };
+    let mut context = ToolContext::new(tmp.path());
+    context.state_namespace = "workspace".to_string();
+    let worker_record = manager
+        .read()
+        .await
+        .get_worker_record_for_session("workspace", &id);
+    assert!(
+        worker_record
+            .as_ref()
+            .is_some_and(|record| !record.delivery_evidence.checked)
+    );
+    let projection =
+        subagent_session_projection(&manager, result, false, &context, worker_record).await;
+    assert_eq!(projection.verification.status, "deliverable_missing");
+    assert!(
+        projection
+            .verification
+            .deliverables
+            .iter()
+            .any(|verdict| verdict.path == "report.md")
+    );
+    assert!(
+        manager.read().await.worker_records[&id]
+            .delivery_evidence
+            .checked
+    );
+}

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

@@ -714,6 +714,7 @@ async fn lifecycle_detail_budget_keeps_the_transcript_handle_retrievable() {
         &id, "retained", &messages, 1, true,
     ));
     let projection = subagent_session_projection(
+        &new_shared_subagent_manager(dir.path().to_path_buf(), 1),
         manager.get_result(&id).unwrap(),
         false,
         &context,

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

@@ -195,6 +195,7 @@ async fn launch_narrows_all_limits_and_continuation_cannot_restart_deadline() {
             SubAgentAssignment::new("inspect".to_string(), None),
             Some(vec![]),
             options,
+            None,
         )
         .unwrap();
     let profile = &guard.worker_records[&child.agent_id].spec.runtime_profile;
@@ -218,6 +219,7 @@ async fn launch_narrows_all_limits_and_continuation_cannot_restart_deadline() {
             resume_from_agent_id: Some(child.agent_id),
             ..Default::default()
         },
+        None,
     );
     assert!(
         refused
@@ -257,6 +259,7 @@ async fn resume_intersects_saved_write_shell_and_tool_permissions_with_current_c
                 preserve_runtime_profile: Some(saved),
                 ..Default::default()
             },
+            None,
         )
         .unwrap();
     let profile = &guard.worker_records[&child.agent_id].spec.runtime_profile;
@@ -300,6 +303,7 @@ async fn root_fork_of_depth_two_leaf_cannot_regain_a_generation() {
                 resume_from_agent_id: Some("leaf".to_string()),
                 ..Default::default()
             },
+            None,
         )
         .unwrap();
     let spec = &guard.worker_records[&child.agent_id].spec;

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

@@ -974,15 +974,27 @@ impl AgentWorkerRecord {
     /// a hand-rolled struct literal that would drift from this one.
     #[cfg(test)]
     pub(crate) fn new(spec: AgentWorkerSpec, now_ms: u64) -> Self {
-        Self::new_for_session(spec, now_ms, String::new())
+        Self::new_for_session(spec, now_ms, String::new(), None)
     }
 
-    fn new_for_session(spec: AgentWorkerSpec, now_ms: u64, owner_session_id: String) -> Self {
+    fn new_for_session(
+        spec: AgentWorkerSpec,
+        now_ms: u64,
+        owner_session_id: String,
+        precomputed_evidence: Option<DeliveryEvidence>,
+    ) -> Self {
         let run_id = agent_worker_run_id(&spec);
         let artifacts = default_subagent_artifacts(&run_id);
         let follow_up = follow_up_target_for_spec(&spec);
         let takeover = takeover_target_for_spec(&spec);
-        let delivery_evidence = DeliveryEvidence::capture(&spec);
+        // #6210: the async spawn path fingerprints pre-lock assuming write
+        // capability; the resolved spec permission is authoritative, so a
+        // baseline for a read-only worker is discarded, never stored.
+        let delivery_evidence = match precomputed_evidence {
+            Some(evidence) if spec.runtime_profile.permissions.write => evidence,
+            Some(_) => DeliveryEvidence::default(),
+            None => DeliveryEvidence::capture(&spec),
+        };
         let mut verification = default_agent_run_verification();
         if let Some(manifest) = spec.launch_manifest.as_ref() {
             verification.deliverables = manifest
@@ -4806,16 +4818,22 @@ impl SubAgentManager {
     }
 
     pub fn register_worker(&mut self, spec: AgentWorkerSpec) {
-        self.register_worker_for_session(spec, "");
+        self.register_worker_for_session(spec, "", None);
     }
 
-    fn register_worker_for_session(&mut self, spec: AgentWorkerSpec, owner_session_id: &str) {
+    fn register_worker_for_session(
+        &mut self,
+        spec: AgentWorkerSpec,
+        owner_session_id: &str,
+        precomputed_evidence: Option<DeliveryEvidence>,
+    ) {
         let worker_id = spec.worker_id.clone();
         let now_ms = epoch_millis_now();
         let mut record = AgentWorkerRecord::new_for_session(
             normalize_worker_spec(spec),
             now_ms,
             owner_session_id.to_string(),
+            precomputed_evidence,
         );
         self.push_worker_event(
             &mut record,
@@ -5444,73 +5462,66 @@ impl SubAgentManager {
         }
     }
 
-    fn verify_worker_delivery(&mut self, worker_id: &str, result: &SubAgentResult) {
-        let Some(record) = self.worker_records.get(worker_id) else {
-            return;
-        };
-        if record.delivery_evidence.checked || result.status == SubAgentStatus::Running {
-            return;
+    /// Store a verification computed off the lock. Re-checks `checked` so a
+    /// racing `ensure_worker_delivery_verified` cannot overwrite a stored
+    /// verdict (#6210). Returns whether this call stored.
+    fn store_delivery_verification(
+        &mut self,
+        worker_id: &str,
+        verification: AgentRunVerificationSummary,
+    ) -> bool {
+        if let Some(record) = self.worker_records.get_mut(worker_id)
+            && !record.delivery_evidence.checked
+        {
+            record.verification = verification;
+            record.delivery_evidence.checked = true;
+            return true;
         }
-        let workspace = &record.spec.workspace;
-        let changed = record.delivery_evidence.changed_paths(workspace);
-        let mut verification = delivery::verify_changes(
-            result.result.as_deref().unwrap_or_default(),
-            record.spec.runtime_profile.permissions.write,
-            &record.delivery_evidence,
-            changed.as_ref(),
-            &record
-                .spec
-                .launch_manifest
-                .as_ref()
-                .map(|manifest| manifest.deliverables.iter().cloned().collect())
-                .unwrap_or_default(),
-        )
-        .unwrap_or_else(default_agent_run_verification);
-        let paths = record
+        false
+    }
+
+    /// Snapshot everything delivery verification needs. `None` when there is
+    /// no record, verification already ran, or the result is not terminal.
+    /// Pure reads for the read lock in `ensure_worker_delivery_verified`
+    /// (#6210).
+    fn delivery_verification_inputs(
+        &self,
+        worker_id: &str,
+        result: &SubAgentResult,
+    ) -> Option<delivery::DeliveryVerificationInputs> {
+        let record = self.worker_records.get(worker_id)?;
+        if record.delivery_evidence.checked || result.status == SubAgentStatus::Running {
+            return None;
+        }
+        let deliverables: Vec<String> = record
             .spec
             .launch_manifest
             .as_ref()
-            .map(|manifest| manifest.deliverables.as_slice())
+            .map(|manifest| manifest.deliverables.clone())
             .unwrap_or_default();
-        verification.deliverables = paths
+        let allowed = deliverables
             .iter()
             .map(|path| {
-                let allowed = record.spec.runtime_profile.permissions.write
+                record.spec.runtime_profile.permissions.write
                     && self
                         .validate_write_scope(worker_id, std::slice::from_ref(path))
-                        .is_ok();
-                delivery::check_deliverable(workspace, path, allowed)
+                        .is_ok()
             })
             .collect();
-        let missing = verification
-            .deliverables
-            .iter()
-            .filter(|verdict| verdict.status != "present")
-            .map(|verdict| format!("{} ({})", verdict.path, verdict.status))
-            .collect::<Vec<_>>();
-        if !missing.is_empty() {
-            let prior = if verification.status == "claim_mismatch" {
-                format!(" {}", verification.summary)
-            } else {
-                String::new()
-            };
-            verification.status = "deliverable_missing".to_string();
-            verification.summary = format!(
-                "Declared deliverables not produced as non-empty files in the worker write scope: {}.{prior}",
-                missing.join(", ")
-            );
-        } else if !paths.is_empty() && verification.status == "self_report_only" {
-            verification.status = "deliverables_present".to_string();
-            verification.summary = "Declared files exist and are non-empty inside the worker write scope; their contents remain a worker self-report.".to_string();
-        }
-        if let Some(record) = self.worker_records.get_mut(worker_id) {
-            record.verification = verification;
-            record.delivery_evidence.checked = true;
-        }
+        Some(delivery::DeliveryVerificationInputs {
+            evidence: record.delivery_evidence.clone(),
+            workspace: record.spec.workspace.clone(),
+            result_text: result.result.clone().unwrap_or_default(),
+            write_perm: record.spec.runtime_profile.permissions.write,
+            deliverables,
+            allowed,
+        })
     }
 
     fn complete_worker_from_result(&mut self, worker_id: &str, result: &SubAgentResult) {
-        self.verify_worker_delivery(worker_id, result);
+        // Delivery verification is deferred: `ensure_worker_delivery_verified`
+        // computes it off the lock before the terminal commit (natural and
+        // panic paths) or heals it on the next detail read (#6210).
         let status = worker_status_from_subagent_result(result);
         let message = match &result.status {
             SubAgentStatus::Completed => Some("completed".to_string()),
@@ -6228,6 +6239,9 @@ impl SubAgentManager {
             assignment,
             allowed_tools,
             options,
+            // Checkpoint resume replays under the write lock; the baseline
+            // captures inline there, as before (#6210).
+            None,
         )?;
         Ok(resumed)
     }
@@ -6538,7 +6552,7 @@ impl SubAgentManager {
             child_route: None,
             launch_manifest: None,
         };
-        self.register_worker_for_session(spec, "workspace");
+        self.register_worker_for_session(spec, "workspace", None);
         (agent_id, input_rx)
     }
 
@@ -6713,6 +6727,7 @@ impl SubAgentManager {
         assignment: SubAgentAssignment,
         allowed_tools: Option<Vec<String>>,
         options: SubAgentSpawnOptions,
+        precomputed_delivery_evidence: Option<DeliveryEvidence>,
     ) -> Result<SubAgentResult> {
         self.cleanup(COMPLETED_AGENT_RETENTION);
 
@@ -7114,7 +7129,11 @@ impl SubAgentManager {
             };
         agent.owner_session_id = runtime.context.state_namespace.clone();
         agent.terminal_delivery = Some(SubAgentTerminalDeliveryContext::from_runtime(&runtime));
-        self.register_worker_for_session(worker_spec, &runtime.context.state_namespace);
+        self.register_worker_for_session(
+            worker_spec,
+            &runtime.context.state_namespace,
+            precomputed_delivery_evidence,
+        );
 
         // Shared-workspace writers may execute only after their exact worker
         // identity and claim are durably replayable. Persist a Starting record
@@ -7874,7 +7893,6 @@ impl SubAgentManager {
             handle.abort();
         }
 
-        self.verify_worker_delivery(agent_id, &result);
         let delivery = self
             .agents
             .get(agent_id)
@@ -8149,11 +8167,30 @@ fn subagent_checkpoint_is_continuable(snapshot: &SubAgentResult) -> bool {
 }
 
 async fn subagent_session_projection(
+    manager: &SharedSubAgentManager,
     snapshot: SubAgentResult,
     timed_out: bool,
     context: &ToolContext,
     worker_record: Option<AgentWorkerRecord>,
 ) -> SubAgentSessionProjection {
+    // Deferred-verification backstop (#6210): terminal commits outside the
+    // natural/panic epilogues (Stop, interrupt, session close, stale cleanup)
+    // leave verification pending, so the first detail read heals it before
+    // the projection reports it. Re-fetch under the session gate afterwards.
+    let worker_record = if snapshot.status != SubAgentStatus::Running
+        && worker_record
+            .as_ref()
+            .is_some_and(|record| !record.delivery_evidence.checked)
+    {
+        ensure_worker_delivery_verified(manager, &snapshot.agent_id, &snapshot).await;
+        manager
+            .read()
+            .await
+            .get_worker_record_for_session(&context.state_namespace, &snapshot.agent_id)
+            .or(worker_record)
+    } else {
+        worker_record
+    };
     let transcript_session_id = format!("agent:{}", snapshot.agent_id);
     let continuable = subagent_checkpoint_is_continuable(&snapshot);
     let transcript_payload = json!({
@@ -9672,7 +9709,9 @@ impl ToolSpec for AgentTool {
             let manager = self.manager.read().await;
             manager.get_worker_record_for_session(&context.state_namespace, &snapshot.agent_id)
         };
-        let projection = subagent_session_projection(snapshot, false, context, worker_record).await;
+        let projection =
+            subagent_session_projection(&self.manager, snapshot, false, context, worker_record)
+                .await;
         let mut value = serde_json::to_value(&projection)
             .map_err(|e| ToolError::execution_failed(e.to_string()))?;
         compact_spawn_receipt(&mut value, verbose);
@@ -9914,7 +9953,7 @@ async fn inspect_agent_from_input(
         }
 
         let mut projection =
-            subagent_session_projection(snapshot, false, context, worker_record).await;
+            subagent_session_projection(&manager, snapshot, false, context, worker_record).await;
         projection.resumed_from = compact
             .get("resumed_from")
             .and_then(Value::as_str)
@@ -9979,7 +10018,8 @@ async fn cancel_agent_from_input(
             manager.get_worker_record_for_session(&context.state_namespace, &snapshot.agent_id);
         (snapshot, worker_record)
     };
-    let projection = subagent_session_projection(snapshot, false, context, worker_record).await;
+    let projection =
+        subagent_session_projection(&manager, snapshot, false, context, worker_record).await;
     let mut tool_result = ToolResult::json(&projection)
         .map_err(|err| ToolError::execution_failed(err.to_string()))?;
     tool_result.metadata = Some(json!({
@@ -10438,6 +10478,21 @@ async fn spawn_subagent_from_input(
     if let Some((lease_key, display_path)) = resident_lease.as_ref() {
         reserve_resident_lease(lease_key, display_path)?;
     }
+    // #6210: fingerprint the delivery baseline BEFORE the manager write
+    // lock. The child workspace is final here (worktree creation already
+    // ran); write capability resolves under the lock, so capture assuming
+    // write and let registration discard it for readers. Skipped only when
+    // the parent ceiling already denies writes — `derive_child` intersects,
+    // so that child is definitely read-only. A join failure falls back to
+    // the inline capture, exactly as before.
+    let precomputed_delivery_evidence = if child_runtime.worker_profile.permissions.write {
+        let workspace = child_runtime.context.workspace.clone();
+        tokio::task::spawn_blocking(move || DeliveryEvidence::capture_for_handle(&workspace, true))
+            .await
+            .ok()
+    } else {
+        None
+    };
     let mut manager_guard = manager.write().await;
 
     let result = manager_guard.spawn_background_with_assignment_options(
@@ -10468,6 +10523,7 @@ async fn spawn_subagent_from_input(
             claim_pre_namespaced: false,
             preserve_runtime_profile: None,
         },
+        precomputed_delivery_evidence,
     );
     let result = match result {
         Ok(result) => result,
@@ -11109,28 +11165,29 @@ async fn supervise_subagent_task_body(
         return;
     };
     let message = crate::utils::panic_message(&*panic);
+    let mut result = match manager_handle.read().await.get_result(&agent_id) {
+        Ok(result) => result,
+        Err(err) => {
+            tracing::error!(
+                target: "subagent",
+                agent_id = %agent_id,
+                ?err,
+                "panicked task no longer has a manager record"
+            );
+            std::panic::resume_unwind(panic);
+        }
+    };
+    result.status = SubAgentStatus::Failed(format!("sub-agent task panicked: {message}"));
+    result.result = None;
+    result.needs_input = None;
+    // Verification precedes the write lock like the natural epilogue (#6210).
+    ensure_worker_delivery_verified(&manager_handle, &agent_id, &result).await;
     {
         let mut manager = manager_handle.write().await;
-        match manager.get_result(&agent_id) {
-            Ok(mut result) => {
-                result.status =
-                    SubAgentStatus::Failed(format!("sub-agent task panicked: {message}"));
-                result.result = None;
-                result.needs_input = None;
-                // Arbitrated exactly like the natural terminal commit: when a
-                // cancel or another terminal outcome already won, this is a
-                // no-op rather than a second result.
-                manager.finish_terminal_result(&agent_id, result, false, true);
-            }
-            Err(err) => {
-                tracing::error!(
-                    target: "subagent",
-                    agent_id = %agent_id,
-                    ?err,
-                    "panicked task no longer has a manager record"
-                );
-            }
-        }
+        // Arbitrated exactly like the natural terminal commit: when a
+        // cancel or another terminal outcome already won, this is a
+        // no-op rather than a second result.
+        manager.finish_terminal_result(&agent_id, result, false, true);
     }
     std::panic::resume_unwind(panic);
 }
@@ -11200,6 +11257,41 @@ async fn budget_work_preservation_note(
     })
 }
 
+/// Deferred delivery verification (#6210). Snapshots inputs under a read
+/// lock, runs the git-subprocess + fingerprint compute in `spawn_blocking`
+/// with no lock held, then stores under a follow-up write lock. Idempotent
+/// via `DeliveryEvidence.checked`: concurrent callers may duplicate the
+/// read-only computation, but only the first store wins.
+///
+/// A racing terminal commit can land between the compute and the store; the
+/// stored verdict is then based on this call's result text rather than the
+/// committed one. The deliverable-presence verdicts are text-independent, so
+/// only the claimed-path comparison can differ, and only in that race.
+async fn ensure_worker_delivery_verified(
+    manager: &SharedSubAgentManager,
+    worker_id: &str,
+    result: &SubAgentResult,
+) {
+    let inputs = {
+        let manager = manager.read().await;
+        let Some(inputs) = manager.delivery_verification_inputs(worker_id, result) else {
+            return;
+        };
+        inputs
+    };
+    let verification =
+        tokio::task::spawn_blocking(move || delivery::compute_delivery_verification(&inputs))
+            .await
+            .ok();
+    let Some(verification) = verification else {
+        return;
+    };
+    manager
+        .write()
+        .await
+        .store_delivery_verification(worker_id, verification);
+}
+
 fn budget_partial_result_with_note(
     mut result: SubAgentResult,
     cause: &str,
@@ -11384,40 +11476,48 @@ async fn run_subagent_task_inner(mut task: SubAgentTask) {
         None
     };
 
+    // The terminal result is built before the write lock: `get_result` takes
+    // `&self`, the error transforms are pure, and no live writer can change
+    // the agent's text between here and the commit — only a racing terminal
+    // claim, which leaves this epilogue with nothing to commit either way.
+    // Verification runs here too, so the fan-in completion and the terminal
+    // persist both carry fresh verdicts with no git subprocess under the lock
+    // (#6210).
+    let terminal = match result {
+        Ok(result) => result,
+        Err(_) => {
+            let mut result = match task.manager_handle.read().await.get_result(&agent_id) {
+                Ok(result) => result,
+                Err(err) => {
+                    tracing::error!(
+                        target: "subagent",
+                        agent_id = %agent_id,
+                        ?err,
+                        "failed task no longer has a manager record"
+                    );
+                    return;
+                }
+            };
+            let error = failure_error
+                .clone()
+                .expect("failed task should carry annotated error");
+            if error.contains("wall-time budget exhausted") {
+                budget_partial_result(result, &error, preservation_note.as_deref())
+            } else {
+                result.status = SubAgentStatus::Failed(error);
+                result.result = None;
+                result.needs_input = None;
+                result
+            }
+        }
+    };
+    ensure_worker_delivery_verified(&task.manager_handle, &agent_id, &terminal).await;
     // Every terminal path — successful/fatal model exit, explicit Stop,
     // coordination interrupt, and stale cleanup — arbitrates and publishes
     // through `finish_terminal_result`. Cancellation that already won leaves
     // this late epilogue with no claim and therefore no duplicate fan-in.
     let terminal_committed = {
         let mut manager = task.manager_handle.write().await;
-        let terminal = match result {
-            Ok(result) => result,
-            Err(_) => {
-                let mut result = match manager.get_result(&agent_id) {
-                    Ok(result) => result,
-                    Err(err) => {
-                        tracing::error!(
-                            target: "subagent",
-                            agent_id = %agent_id,
-                            ?err,
-                            "failed task no longer has a manager record"
-                        );
-                        return;
-                    }
-                };
-                let error = failure_error
-                    .clone()
-                    .expect("failed task should carry annotated error");
-                if error.contains("wall-time budget exhausted") {
-                    budget_partial_result(result, &error, preservation_note.as_deref())
-                } else {
-                    result.status = SubAgentStatus::Failed(error);
-                    result.result = None;
-                    result.needs_input = None;
-                    result
-                }
-            }
-        };
         manager.finish_terminal_result(&agent_id, terminal, false, true)
     };
     if !terminal_committed {

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

@@ -2181,6 +2181,7 @@ async fn detached_interactive_usage_after_mailbox_seal_reaches_session_accountin
         manager.write().await.register_worker_for_session(
             make_worker_spec(worker_id, tmp.path().to_path_buf()),
             "session-detached-usage",
+            None,
         );
     }
     let (release_usage_tx, release_usage_rx) = tokio::sync::oneshot::channel();
@@ -2266,6 +2267,7 @@ async fn child_guardian_usage_source_is_sanitized_and_replay_idempotent() {
     manager.write().await.register_worker_for_session(
         make_worker_spec("agent_guardian", tmp.path().to_path_buf()),
         "guardian-usage-session",
+        None,
     );
 
     let runtime_owner = "interactive:guardian-usage-session:turn-parent";
@@ -2352,6 +2354,7 @@ async fn ownerless_no_mailbox_provider_usage_reaches_accounting_once() {
     manager.write().await.register_worker_for_session(
         make_worker_spec("agent_direct", tmp.path().to_path_buf()),
         "direct-usage-session",
+        None,
     );
     let mut runtime = stub_runtime();
     runtime.manager = Arc::clone(&manager);
@@ -2446,6 +2449,7 @@ async fn ownerless_child_usage_crossing_new_settles_to_its_dispatch_origin_once(
     manager.write().await.register_worker_for_session(
         make_worker_spec(agent_id, tmp.path().to_path_buf()),
         origin_session_id,
+        None,
     );
 
     // Dispatch in the origin session: the engine's off-turn continuation
@@ -2567,6 +2571,7 @@ async fn provider_success_without_usage_records_one_route_aware_gap_and_no_zero_
     manager.write().await.register_worker_for_session(
         make_worker_spec("agent_missing_usage", tmp.path().to_path_buf()),
         "missing-usage-session",
+        None,
     );
 
     let runtime_owner = "interactive:missing-usage-session:turn-parent";
@@ -5158,7 +5163,8 @@ async fn session_projection_exposes_forked_prefix_cache_contract() {
     snapshot.fork_context = true;
 
     let ctx = ToolContext::new(".");
-    let projection = subagent_session_projection(snapshot, false, &ctx, None).await;
+    let manager = new_shared_subagent_manager(PathBuf::from("."), 1);
+    let projection = subagent_session_projection(&manager, snapshot, false, &ctx, None).await;
 
     assert_eq!(projection.name, "fanout_review");
     assert_eq!(projection.context_mode, "forked");
@@ -5204,7 +5210,8 @@ async fn terminal_session_projection_prefers_full_transcript_handle() {
         )
     };
 
-    let projection = subagent_session_projection(snapshot, false, &ctx, None).await;
+    let manager = new_shared_subagent_manager(PathBuf::from("."), 1);
+    let projection = subagent_session_projection(&manager, snapshot, false, &ctx, None).await;
 
     assert_eq!(projection.transcript_handle, full_handle);
     assert_eq!(projection.transcript_handle.name, "full_transcript");
@@ -5224,7 +5231,8 @@ async fn interrupted_projection_exposes_checkpoint_metadata_and_messages() {
     snapshot.checkpoint = Some(checkpoint.clone());
 
     let ctx = ToolContext::new(".");
-    let projection = subagent_session_projection(snapshot, false, &ctx, None).await;
+    let manager = new_shared_subagent_manager(PathBuf::from("."), 1);
+    let projection = subagent_session_projection(&manager, snapshot, false, &ctx, None).await;
 
     assert_eq!(projection.status, "waiting_for_user");
     assert!(projection.terminal);
@@ -5257,7 +5265,7 @@ async fn interrupted_projection_exposes_checkpoint_metadata_and_messages() {
     );
 
     let timed_out_projection =
-        subagent_session_projection(projection.snapshot.clone(), true, &ctx, None).await;
+        subagent_session_projection(&manager, projection.snapshot.clone(), true, &ctx, None).await;
     assert!(timed_out_projection.needs_continuation);
     assert!(timed_out_projection.timed_out);
     assert!(timed_out_projection.timed_out_with_checkpoint);
@@ -9274,7 +9282,8 @@ async fn api_timeout_preserves_checkpoint_and_returns_needs_input_without_parkin
         manager.get_worker_record(&agent_id)
     };
     let projection =
-        subagent_session_projection(interrupted.clone(), false, &ctx, worker_record).await;
+        subagent_session_projection(&manager, interrupted.clone(), false, &ctx, worker_record)
+            .await;
     assert_eq!(projection.status, "waiting_for_user");
     assert!(projection.continuable);
     assert!(projection.needs_continuation);
@@ -9665,6 +9674,7 @@ async fn spawn_duplicate_session_name_error_names_conflicting_agent() {
                     name: Some("researcher".to_string()),
                     ..Default::default()
                 },
+                None,
             )
             .expect_err("duplicate session name must error")
     };
@@ -9732,6 +9742,7 @@ async fn spawn_session_name_held_by_prior_session_agent_does_not_collide() {
                     name: Some("researcher".to_string()),
                     ..Default::default()
                 },
+                None,
             )
             .expect("a prior-session holder must not reject a fresh same-name spawn")
     };
@@ -9775,6 +9786,7 @@ async fn shared_write_claim_is_registered_before_parallel_launch_and_manifested(
                 make_assignment(),
                 Some(vec![]),
                 options,
+                None,
             )
             .expect("first writer admitted");
         let second = guard
@@ -9795,6 +9807,7 @@ async fn shared_write_claim_is_registered_before_parallel_launch_and_manifested(
                     }),
                     ..Default::default()
                 },
+                None,
             )
             .expect_err("overlapping live contract must contend");
         (first.agent_id, second.to_string())
@@ -9849,6 +9862,7 @@ async fn write_capable_agent_does_not_launch_when_durable_registration_fails() {
                 }),
                 ..Default::default()
             },
+            None,
         )
         .expect_err("writer must fail before spawn when its durable claim cannot commit")
         .to_string();
@@ -9930,6 +9944,7 @@ async fn write_scope_contention_covers_regular_agent_and_active_fleet_writer() {
                 }),
                 ..Default::default()
             },
+            None,
         )
         .expect_err("regular-agent launch must see active Fleet ownership");
     let launch = launch.to_string();
@@ -15013,6 +15028,7 @@ fn persist_round_trip_preserves_session_and_boot_ownership() {
         writer.register_worker_for_session(
             make_worker_spec("headless_persist", dir.path().to_path_buf()),
             "session-persist",
+            None,
         );
         writer
             .persist_state()
@@ -20301,8 +20317,8 @@ fn init_claim_repo(root: &Path) {
     assert!(output.status.success(), "git commit: {output:?}");
 }
 
-#[test]
-fn completed_claim_of_untouched_file_taints_verification() {
+#[tokio::test]
+async fn completed_claim_of_untouched_file_taints_verification() {
     // R7 (finish-operator 2026-08-02): the morning report caught a child
     // claiming edits git had never seen — by hand. At terminal delivery the
     // claimed changed-files are checked against git status in the child's
@@ -20313,15 +20329,29 @@ fn completed_claim_of_untouched_file_taints_verification() {
     let mut spec = make_worker_spec("agent_claims", tmp.path().to_path_buf());
     spec.runtime_profile.permissions.write = true;
     manager.register_worker(spec);
+    let manager = Arc::new(RwLock::new(manager));
 
     let mut snapshot = make_snapshot(SubAgentStatus::Completed);
     snapshot.agent_id = "agent_claims".to_string();
     snapshot.name = "agent_claims".to_string();
     snapshot.workspace = Some(tmp.path().to_path_buf());
     snapshot.result = Some("CHANGES: src/lib.rs".to_string());
-    manager.complete_worker_from_result("agent_claims", &snapshot);
+    // Deferred verification (#6210): the commit leaves it pending; `ensure`
+    // computes the taint off the lock.
+    {
+        let mut guard = manager.write().await;
+        guard.complete_worker_from_result("agent_claims", &snapshot);
+        assert!(
+            !guard.worker_records["agent_claims"]
+                .delivery_evidence
+                .checked
+        );
+    }
+    ensure_worker_delivery_verified(&manager, "agent_claims", &snapshot).await;
 
     let record = manager
+        .read()
+        .await
         .get_worker_record("agent_claims")
         .expect("worker record");
     assert_eq!(record.verification.status, "claim_mismatch");
@@ -20405,8 +20435,8 @@ fn resume_from_rejects_running_source() {
     );
 }
 
-#[test]
-fn completed_claim_matching_workspace_state_stays_untainted() {
+#[tokio::test]
+async fn completed_claim_matching_workspace_state_stays_untainted() {
     let tmp = tempdir().expect("tempdir");
     init_claim_repo(tmp.path());
 
@@ -20414,16 +20444,11 @@ fn completed_claim_matching_workspace_state_stays_untainted() {
     let mut manager = SubAgentManager::new(tmp.path().to_path_buf(), 2);
     manager.register_worker(make_worker_spec("agent_honest", tmp.path().to_path_buf()));
     std::fs::write(tmp.path().join("src/lib.rs"), "pub fn improved() {}\n").expect("edit file");
-    let mut snapshot = make_snapshot(SubAgentStatus::Completed);
-    snapshot.agent_id = "agent_honest".to_string();
-    snapshot.name = "agent_honest".to_string();
-    snapshot.workspace = Some(tmp.path().to_path_buf());
-    snapshot.result = Some("Updated src/lib.rs with the new implementation.".to_string());
-    manager.complete_worker_from_result("agent_honest", &snapshot);
-    let record = manager
-        .get_worker_record("agent_honest")
-        .expect("worker record");
-    assert_eq!(record.verification.status, "self_report_only");
+    let mut honest = make_snapshot(SubAgentStatus::Completed);
+    honest.agent_id = "agent_honest".to_string();
+    honest.name = "agent_honest".to_string();
+    honest.workspace = Some(tmp.path().to_path_buf());
+    honest.result = Some("Updated src/lib.rs with the new implementation.".to_string());
 
     // Honest committed claim: the child committed its work, so git status is
     // clean but the commit is newer than the worker record.
@@ -20447,20 +20472,33 @@ fn completed_claim_matching_workspace_state_stays_untainted() {
         // so the just-made commit is unambiguously after it.
         record.created_at_ms = record.created_at_ms.saturating_sub(60_000);
     }
-    let mut snapshot = make_snapshot(SubAgentStatus::Completed);
-    snapshot.agent_id = "agent_committer".to_string();
-    snapshot.name = "agent_committer".to_string();
-    snapshot.workspace = Some(tmp.path().to_path_buf());
-    snapshot.result = Some("Updated src/lib.rs and committed the change.".to_string());
-    manager.complete_worker_from_result("agent_committer", &snapshot);
-    let record = manager
-        .get_worker_record("agent_committer")
-        .expect("worker record");
-    assert_eq!(
-        record.verification.status, "self_report_only",
-        "{}",
-        record.verification.summary
-    );
+    let mut committer = make_snapshot(SubAgentStatus::Completed);
+    committer.agent_id = "agent_committer".to_string();
+    committer.name = "agent_committer".to_string();
+    committer.workspace = Some(tmp.path().to_path_buf());
+    committer.result = Some("Updated src/lib.rs and committed the change.".to_string());
+    let manager = Arc::new(RwLock::new(manager));
+    // Deferred verification (#6210): both commits leave verification
+    // pending; `ensure` computes the untainted verdicts off the lock.
+    {
+        let mut guard = manager.write().await;
+        guard.complete_worker_from_result("agent_honest", &honest);
+        guard.complete_worker_from_result("agent_committer", &committer);
+        for id in ["agent_honest", "agent_committer"] {
+            assert!(!guard.worker_records[id].delivery_evidence.checked);
+        }
+    }
+    ensure_worker_delivery_verified(&manager, "agent_honest", &honest).await;
+    ensure_worker_delivery_verified(&manager, "agent_committer", &committer).await;
+    let guard = manager.read().await;
+    for id in ["agent_honest", "agent_committer"] {
+        let record = guard.get_worker_record(id).expect("worker record");
+        assert_eq!(
+            record.verification.status, "self_report_only",
+            "{id}: {}",
+            record.verification.summary
+        );
+    }
 }
 
 #[tokio::test]
@@ -20496,8 +20534,9 @@ async fn spawn_receipt_compacts_and_verbose_restores_the_archive() {
     let snapshot = inner.get_result(&agent_id).expect("snapshot");
     let worker_record = inner.get_worker_record(&agent_id);
     let context = ToolContext::new(".");
+    let shared = new_shared_subagent_manager(PathBuf::from("."), 1);
     let mut projection =
-        subagent_session_projection(snapshot, false, &context, worker_record).await;
+        subagent_session_projection(&shared, snapshot, false, &context, worker_record).await;
     // The route receipt rides inside the budget rather than being exempt from
     // it (#5305), so measure the receipt that ships.
     let metadata = spawn_route_metadata("zai", "glm-5", "agent_profile.model");
@@ -20650,8 +20689,9 @@ async fn spawn_receipt_route_survives_compaction_for_a_type_only_spawn() {
     let snapshot = inner.get_result(&agent_id).expect("snapshot");
     let worker_record = inner.get_worker_record(&agent_id);
     let context = ToolContext::new(".");
+    let shared = new_shared_subagent_manager(PathBuf::from("."), 1);
     let mut projection =
-        subagent_session_projection(snapshot, false, &context, worker_record).await;
+        subagent_session_projection(&shared, snapshot, false, &context, worker_record).await;
     let metadata = spawn_route_metadata("deepseek", "deepseek-v4-flash", "run.model");
     projection.child_route = Some(spawn_child_route_projection(&metadata));
 
@@ -21681,6 +21721,7 @@ mod child_permission_gate {
             manager.register_worker_for_session(
                 make_worker_spec("agent_gate", workspace),
                 "guardian-test-session",
+                None,
             );
         }
         // Keep the tempdir alive for the registry's lifetime by leaking it
@@ -23076,3 +23117,53 @@ async fn test_disallowed_tools_resume_keeps_saved_and_current_ancestor_denials()
         vec!["mcp_saved_*"]
     );
 }
+
+/// Precomputed spawn evidence (#6210) is adopted for write-capable workers:
+/// a baseline captured before the lock answers changed-path queries.
+#[test]
+fn precomputed_delivery_evidence_is_adopted_for_write_workers() {
+    let tmp = tempdir().expect("tempdir");
+    init_claim_repo(tmp.path());
+    let evidence = DeliveryEvidence::capture_for_handle(tmp.path(), true);
+    let mut manager = SubAgentManager::new(tmp.path().to_path_buf(), 2);
+    let spec = make_write_worker_spec("agent_precomputed", tmp.path().to_path_buf(), ".");
+    assert!(spec.runtime_profile.permissions.write);
+    manager.register_worker_for_session(spec, "workspace", Some(evidence));
+    let record = manager
+        .get_worker_record("agent_precomputed")
+        .expect("worker record");
+    assert!(record.delivery_evidence.changed_paths(tmp.path()).is_some());
+}
+
+/// The resolved spec permission is authoritative: a precomputed baseline for
+/// a read-only worker is discarded, never stored (#6210).
+#[test]
+fn precomputed_delivery_evidence_is_discarded_for_read_only_workers() {
+    let tmp = tempdir().expect("tempdir");
+    init_claim_repo(tmp.path());
+    let evidence = DeliveryEvidence::capture_for_handle(tmp.path(), true);
+    assert!(evidence.changed_paths(tmp.path()).is_some());
+    let mut manager = SubAgentManager::new(tmp.path().to_path_buf(), 2);
+    let mut spec = make_worker_spec("agent_reader", tmp.path().to_path_buf());
+    spec.runtime_profile.permissions.write = false;
+    manager.register_worker_for_session(spec, "workspace", Some(evidence));
+    let record = manager
+        .get_worker_record("agent_reader")
+        .expect("worker record");
+    assert!(record.delivery_evidence.changed_paths(tmp.path()).is_none());
+}
+
+/// Paths without precomputed evidence (resume, Fleet, tests) capture inline
+/// at registration, exactly as before (#6210).
+#[test]
+fn missing_precomputed_evidence_falls_back_to_inline_capture() {
+    let tmp = tempdir().expect("tempdir");
+    init_claim_repo(tmp.path());
+    let mut manager = SubAgentManager::new(tmp.path().to_path_buf(), 2);
+    let spec = make_write_worker_spec("agent_inline", tmp.path().to_path_buf(), ".");
+    manager.register_worker_for_session(spec, "workspace", None);
+    let record = manager
+        .get_worker_record("agent_inline")
+        .expect("worker record");
+    assert!(record.delivery_evidence.changed_paths(tmp.path()).is_some());
+}