fix(plugins): a revocation blocks carrying built-in trust only until the next review (K4)

builtin_predecessor returned None whenever any same-named built-in entry
lacked a receipt. A user who revoked Computer Use on one build and then
reviewed and enabled a later build therefore lost it on every following
upgrade, forever; so did anyone with a disable-only (never reviewed) entry.

Now the most recently reviewed same-named entry decides. A revoked entry
is dated by its last review_history receipt, so a revocation still blocks
carrying until the user reviews a build again, and ties go to the
revocation. Entries that were never reviewed granted nothing and are
ignored. The K4 test gains the re-review step (v4 trusted and enabled,
v5 carries).

Checks:
- cargo test -p codewhale-tui --lib plugins::builtin: 19 passed, 0 failed,
  1 ignored
- cargo test -p codewhale-tui --lib plugins:: -- --test-threads=1:
  212 passed, 0 failed, 1 ignored (parallel runs flake on a pre-existing
  process-env race with managed_policy_path_env_override_is_honored)
- rustfmt --check on registry.rs and builtin.rs: clean

Refs #6303

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

crates/tui/src/plugins/builtin.rs

@@ -675,7 +675,7 @@ mod tests {
 
         const MANIFEST: &[u8] = br#"{"$schema":"https://agent-plugins.org/schemas/plugin.json","name":"fixture","version":"1.0.0"}"#;
         const SKILL: &[u8] = b"---\nname: extra\ndescription: An added skill.\n---\nBody.\n";
-        let builds: [&[(&str, &[u8])]; 4] = [
+        let builds: [&[(&str, &[u8])]; 5] = [
             &[("plugin.json", MANIFEST), ("body.txt", b"v1")],
             &[("plugin.json", MANIFEST), ("body.txt", b"v2")],
             &[
@@ -688,6 +688,11 @@ mod tests {
                 ("body.txt", b"v4"),
                 ("skills/extra/SKILL.md", SKILL),
             ],
+            &[
+                ("plugin.json", MANIFEST),
+                ("body.txt", b"v5"),
+                ("skills/extra/SKILL.md", SKILL),
+            ],
         ];
         let temp = tempfile::tempdir().unwrap();
         let cache = temp.path().join("cache");
@@ -747,10 +752,19 @@ mod tests {
         // A revocation anywhere in the line blocks carrying.
         let mut v3 = v3;
         v3.revoke_trust("fixture").unwrap();
-        let v4 = registry_for(builds[3]);
+        let mut v4 = registry_for(builds[3]);
         let plugin = v4.get("fixture").unwrap();
         assert_eq!(plugin.trust_status, PluginTrustStatus::NeverReviewed);
         assert!(!plugin.enabled);
+
+        // A revocation blocks only until the next review: once the user
+        // reviews and enables a later build, upgrades carry that review again.
+        v4.trust("fixture").unwrap();
+        v4.enable("fixture").unwrap();
+        let v5 = registry_for(builds[4]);
+        let plugin = v5.get("fixture").unwrap();
+        assert_eq!(plugin.trust_status, PluginTrustStatus::Trusted);
+        assert!(plugin.active());
     }
 
     #[test]

crates/tui/src/plugins/registry.rs

@@ -563,8 +563,8 @@ impl PluginRegistry {
     ///   user reviews the changes.
     ///
     /// Fail-closed: nothing is carried when the new id already has state,
-    /// when any same-named predecessor has no receipt (it was revoked, or
-    /// never reviewed), or when the state file is invalid. Older ids are left
+    /// when the most recently reviewed same-named predecessor has since been
+    /// revoked, or when the state file is invalid. Older ids are left
     /// untouched, so a still-running older binary keeps its own authority.
     /// Only the built-in scope is ever carried; user and workspace bundles
     /// still require review of the exact bytes on disk.
@@ -1227,9 +1227,13 @@ pub(crate) fn harden_plugin_state_file(_path: &Path) -> Result<(), String> {
     Ok(())
 }
 
-/// The newest persisted review of another built-in with this name, when one
-/// may be carried to `id`: `id` has no state yet, and every same-named
-/// built-in entry still holds a receipt (a revocation anywhere blocks it).
+/// The newest persisted review of another built-in with this name, when it
+/// may be carried to `id`: `id` has no state yet, and the same-named built-in
+/// entry with the most recent review still holds its receipt. A revoked entry
+/// is dated by its last review, so revoking blocks carrying until the user
+/// reviews a build again; ties go to the revocation. Entries that were never
+/// reviewed (for example, disabled before any review) granted nothing and are
+/// ignored.
 fn builtin_predecessor<'a>(
     state: &'a PluginStateFile,
     id: &PluginId,
@@ -1245,14 +1249,19 @@ fn builtin_predecessor<'a>(
         if parts.next() != Some(builtin) || parts.nth(1) != Some(name) {
             continue;
         }
-        let receipt = entry.trust.as_ref()?;
-        let reviewed = chrono::DateTime::parse_from_rfc3339(&receipt.reviewed_at)
+        let Some(last_review) = entry.trust.as_ref().or(entry.review_history.last()) else {
+            continue;
+        };
+        let reviewed = chrono::DateTime::parse_from_rfc3339(&last_review.reviewed_at)
             .map_or(i64::MIN, |at| at.timestamp_micros());
-        if newest.is_none_or(|(_, at)| reviewed > at) {
+        let revoked = entry.trust.is_none();
+        if newest.is_none_or(|(_, at)| reviewed > at || (reviewed == at && revoked)) {
             newest = Some((entry, reviewed));
         }
     }
-    newest.map(|(entry, _)| entry)
+    newest
+        .map(|(entry, _)| entry)
+        .filter(|entry| entry.trust.is_some())
 }
 
 fn runtime_stage_path(state_path: &Path, id: &PluginId, content_hash: &str) -> PathBuf {

crates/tui/src/runtime_api.rs

@@ -86,8 +86,8 @@ use crate::task_manager::{
     NewTaskRequest, SharedTaskManager, TaskManager, TaskManagerConfig, TaskRecord, TaskSummary,
 };
 use crate::tools::subagent::{
-    AgentWorkerRecord, SharedSubAgentManager, load_persisted_agent_worker_records,
-    new_shared_subagent_manager_with_timeout,
+    AgentWorkerRecord, AgentWorkerStatus, SharedSubAgentManager, SubAgentStatus,
+    load_persisted_agent_worker_records, new_shared_subagent_manager_with_timeout,
 };
 #[cfg(test)]
 pub(super) use codewhale_models::{ContentBlock, Message};
@@ -1205,6 +1205,7 @@ pub fn build_router(state: RuntimeApiState) -> Router {
         .route("/v1/workspace/instructions", get(workspace_instructions))
         .route("/v1/agent-runs", get(list_agent_runs))
         .route("/v1/agent-runs/{run_id}", get(get_agent_run))
+        .route("/v1/agent-runs/{run_id}/cancel", post(cancel_agent_run))
         .route("/v1/fleet/profiles", get(list_fleet_profiles))
         .route(
             "/v1/fleet/runs",
@@ -2075,18 +2076,145 @@ async fn get_agent_run(
     })?;
     let run = runs
         .into_iter()
-        .find(|record| {
-            let effective_run_id = if record.spec.run_id.is_empty() {
-                record.spec.worker_id.as_str()
-            } else {
-                record.spec.run_id.as_str()
-            };
-            effective_run_id == run_id || record.spec.worker_id == run_id
-        })
+        .find(|record| agent_run_matches(record, &run_id))
         .ok_or_else(|| ApiError::not_found(format!("agent run '{run_id}' not found")))?;
     Ok(Json(run))
 }
 
+/// A run is addressed by its run id, or by its worker id for records that
+/// predate run ids.
+fn agent_run_matches(record: &AgentWorkerRecord, run_id: &str) -> bool {
+    let effective_run_id = if record.spec.run_id.is_empty() {
+        record.spec.worker_id.as_str()
+    } else {
+        record.spec.run_id.as_str()
+    };
+    effective_run_id == run_id || record.spec.worker_id == run_id
+}
+
+/// How long a stop request waits for the owning engine to record the
+/// terminal receipt before answering `202 Accepted` with the live record.
+const AGENT_RUN_CANCEL_SETTLE: Duration = Duration::from_secs(3);
+
+/// `POST /v1/agent-runs/{run_id}/cancel`: stop a delegated agent run and
+/// answer with its receipt (addendum F2).
+///
+/// The stop goes through the same session-scoped path as the TUI's `X` and
+/// the `agent/cancel` tool, so descendants stop with it and a write-scoped
+/// child's work is inventoried rather than dropped. The answer is:
+/// - `200` with the terminal record once the run is stopped (or was already
+///   finished — stopping is idempotent);
+/// - `202` with the current record when the owning engine accepted the stop
+///   but has not recorded the terminal receipt yet;
+/// - `404` for an unknown run;
+/// - `409` when the run belongs to a session this runtime does not host, so
+///   nothing here can reach it.
+async fn cancel_agent_run(
+    State(state): State<RuntimeApiState>,
+    Path(run_id): Path<String>,
+) -> Result<(StatusCode, Json<AgentWorkerRecord>), ApiError> {
+    // Runs this runtime is executing itself (Fleet-launched children) stop
+    // in place. Only a child running in this process qualifies: records the
+    // manager loaded from disk belong to whichever process wrote them.
+    let owned = {
+        let manager = state.sub_agent_manager.read().await;
+        manager
+            .list_worker_records()
+            .into_iter()
+            .find(|record| agent_run_matches(record, &run_id))
+            .filter(|record| {
+                manager
+                    .get_result(&record.spec.worker_id)
+                    .is_ok_and(|agent| agent.status == SubAgentStatus::Running)
+            })
+    };
+    if let Some(record) = owned {
+        let agent_id = record.spec.worker_id.clone();
+        let cancelled = {
+            let mut manager = state.sub_agent_manager.write().await;
+            if record.owner_session_id.is_empty() {
+                manager.cancel_agent(&agent_id)
+            } else {
+                manager.cancel_agent_for_session(&record.owner_session_id, &agent_id)
+            }
+        }
+        .map_err(|err| {
+            ApiError::conflict(format!("agent run '{run_id}' could not be stopped: {err}"))
+        })?;
+        crate::tools::subagent::preserve_cancelled_work(&state.sub_agent_manager, cancelled).await;
+        let manager = state.sub_agent_manager.read().await;
+        let record = manager
+            .list_worker_records()
+            .into_iter()
+            .find(|record| record.spec.worker_id == agent_id)
+            .unwrap_or(record);
+        let status = if record.status.is_terminal() {
+            StatusCode::OK
+        } else {
+            StatusCode::ACCEPTED
+        };
+        return Ok((status, Json(record)));
+    }
+
+    let find_persisted = |workspace: &FsPath| -> Result<Option<AgentWorkerRecord>, ApiError> {
+        load_persisted_agent_worker_records(workspace)
+            .map(|runs| {
+                runs.into_iter()
+                    .find(|record| agent_run_matches(record, &run_id))
+            })
+            .map_err(|err| {
+                ApiError::internal(format!("Failed to load persisted agent run records: {err}"))
+            })
+    };
+    let record = find_persisted(&state.workspace)?
+        .ok_or_else(|| ApiError::not_found(format!("agent run '{run_id}' not found")))?;
+
+    // A runtime thread's session id is its thread id: its live engine owns
+    // the child and stops it through the session-scoped cancel path. The
+    // on-disk projection cannot tell a live child from an orphan (loading it
+    // marks every in-flight record interrupted), so a hosted thread is always
+    // asked, and only its own write settles the answer.
+    let engine = if record.owner_session_id.is_empty() {
+        None
+    } else {
+        state
+            .runtime_threads
+            .loaded_engine(&record.owner_session_id)
+            .await
+    };
+    let Some(engine) = engine else {
+        if record.status.is_terminal() {
+            return Ok((StatusCode::OK, Json(record)));
+        }
+        return Err(ApiError::conflict(format!(
+            "agent run '{run_id}' belongs to a session this runtime is not hosting; stop it from that session"
+        )));
+    };
+    engine
+        .send(crate::core::ops::Op::CancelSubAgent {
+            agent_id: record.spec.worker_id.clone(),
+        })
+        .await
+        .map_err(|err| ApiError::internal(format!("Failed to reach the run's engine: {err}")))?;
+
+    let settled = |current: &AgentWorkerRecord| {
+        current.status.is_terminal()
+            && (current.status != AgentWorkerStatus::Interrupted
+                || current.latest_message != record.latest_message)
+    };
+    let deadline = tokio::time::Instant::now() + AGENT_RUN_CANCEL_SETTLE;
+    loop {
+        let current = find_persisted(&state.workspace)?.unwrap_or_else(|| record.clone());
+        if settled(&current) {
+            return Ok((StatusCode::OK, Json(current)));
+        }
+        if tokio::time::Instant::now() >= deadline {
+            return Ok((StatusCode::ACCEPTED, Json(current)));
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+}
+
 async fn list_fleet_profiles(
     State(state): State<RuntimeApiState>,
 ) -> Result<Json<Value>, ApiError> {

crates/tui/src/runtime_api/tests.rs

@@ -2850,6 +2850,126 @@ async fn agent_runs_runtime_api_exposes_persisted_worker_receipts() -> Result<()
     Ok(())
 }
 
+#[tokio::test]
+async fn agent_run_cancel_stops_a_live_child_and_returns_its_receipt() -> Result<()> {
+    let root = std::env::temp_dir().join(format!("codewhale-agent-run-cancel-{}", Uuid::new_v4()));
+    let workspace = root.join("workspace");
+    fs::create_dir_all(&workspace)?;
+    let manager = crate::tools::subagent::new_shared_subagent_manager(workspace.clone(), 2);
+    let agent_id = {
+        let mut guard = manager.write().await;
+        let id = guard.insert_test_running_agent("stoppable", &workspace);
+        guard.assign_test_session_owner(&id, "session-stop");
+        id
+    };
+
+    let Some((addr, _runtime_threads, handle)) =
+        spawn_test_server_with_root_token_mobile_workspace_and_subagents(
+            root.clone(),
+            root.join("sessions"),
+            None,
+            false,
+            workspace,
+            Some(manager.clone()),
+            None,
+        )
+        .await?
+    else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+
+    let response = client
+        .post(format!("http://{addr}/v1/agent-runs/{agent_id}/cancel"))
+        .send()
+        .await?;
+    assert_eq!(response.status(), StatusCode::OK);
+    let receipt: serde_json::Value = response.json().await?;
+    assert_eq!(receipt["spec"]["worker_id"], agent_id.as_str());
+    assert_eq!(receipt["status"], "cancelled");
+    assert_eq!(
+        manager.read().await.get_result(&agent_id)?.status,
+        crate::tools::subagent::SubAgentStatus::Cancelled
+    );
+
+    // Stopping a stopped run is a no-op that answers with the same receipt.
+    let again = client
+        .post(format!("http://{addr}/v1/agent-runs/{agent_id}/cancel"))
+        .send()
+        .await?;
+    assert_eq!(again.status(), StatusCode::OK);
+    let again: serde_json::Value = again.json().await?;
+    assert_eq!(again["status"], "cancelled");
+
+    let missing = client
+        .post(format!("http://{addr}/v1/agent-runs/missing/cancel"))
+        .send()
+        .await?
+        .status();
+    assert_eq!(missing, StatusCode::NOT_FOUND);
+
+    handle.abort();
+    Ok(())
+}
+
+#[tokio::test]
+async fn agent_run_cancel_refuses_a_run_owned_by_a_session_it_does_not_host() -> Result<()> {
+    let root = std::env::temp_dir().join(format!(
+        "codewhale-agent-run-cancel-foreign-{}",
+        Uuid::new_v4()
+    ));
+    let workspace = root.join("workspace");
+    fs::create_dir_all(workspace.join(".codewhale/state"))?;
+    // A child parked on a question in another terminal session: in flight on
+    // disk, but no engine in this runtime owns it.
+    let mut record = {
+        let manager = crate::tools::subagent::new_shared_subagent_manager(workspace.clone(), 1);
+        let mut guard = manager.write().await;
+        let id = guard.insert_test_running_agent("elsewhere", &workspace);
+        guard.assign_test_session_owner(&id, "terminal-session");
+        guard
+            .list_worker_records()
+            .into_iter()
+            .find(|record| record.spec.worker_id == id)
+            .expect("seeded record")
+    };
+    record.status = crate::tools::subagent::AgentWorkerStatus::WaitingForUser;
+    fs::write(
+        workspace.join(".codewhale/state/subagents.v1.json"),
+        serde_json::to_vec_pretty(&json!({
+            "schema_version": 1,
+            "agents": [],
+            "workers": [record],
+        }))?,
+    )?;
+
+    let Some((addr, _runtime_threads, handle)) =
+        spawn_test_server_with_root_token_mobile_workspace(
+            root.clone(),
+            root.join("sessions"),
+            None,
+            false,
+            workspace,
+        )
+        .await?
+    else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+    let response = client
+        .post(format!(
+            "http://{addr}/v1/agent-runs/agent_elsewhere/cancel"
+        ))
+        .send()
+        .await?;
+    assert_eq!(response.status(), StatusCode::CONFLICT);
+    let body = response.text().await?;
+    assert!(body.contains("not hosting"), "{body}");
+
+    handle.abort();
+    Ok(())
+}
+
 #[tokio::test]
 async fn stream_requires_prompt() -> Result<()> {
     let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {

crates/tui/src/runtime_threads.rs

@@ -11215,6 +11215,18 @@ impl RuntimeThreadManager {
         }
     }
 
+    /// 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.
+    pub async fn loaded_engine(&self, thread_id: &str) -> Option<EngineHandle> {
+        self.active
+            .lock()
+            .await
+            .engines
+            .get(thread_id)
+            .map(|state| state.engine.clone())
+    }
+
     /// Get the engine handle for a thread, loading it if necessary.
     /// Public wrapper around the private `ensure_engine_loaded`.
     pub async fn get_engine(&self, thread_id: &str) -> Result<EngineHandle> {

docs/PLUGIN_MARKETPLACE.md

@@ -38,8 +38,9 @@ hash is unchanged: a bundle you trusted and enabled stays trusted and enabled
 on the new build. When the capabilities changed, it shows
 `capabilities-changed` and stays off until you review it again with
 `/plugin show computer-use` and `/plugin trust computer-use`. If you revoked
-trust on any earlier build, nothing carries and the new build waits for a
-fresh review. User and workspace plugins never carry trust: changed bytes
+trust after your most recent review, nothing carries and the new build waits
+for a fresh review; once you review a build again, later upgrades carry that
+review. User and workspace plugins never carry trust: changed bytes
 always need review.
 
 ## Chromewhale

docs/RUNTIME_API.md

@@ -1999,6 +1999,7 @@ a read-only inspection surface:
 |---|---|
 | List persisted agent runs | `GET /v1/agent-runs` |
 | Inspect one run | `GET /v1/agent-runs/{run_id}` |
+| Stop one run | `POST /v1/agent-runs/{run_id}/cancel` |
 
 The response is the same worker-record shape surfaced by `agent` receipts:
 `spec.run_id`, `actor_kind`, lifecycle `status`, bounded `events`,
@@ -2006,9 +2007,22 @@ The response is the same worker-record shape surfaced by `agent` receipts:
 falls back to the worker id for older records, and `{run_id}` may be either the
 run id or the worker id.
 
-These endpoints do not start, cancel, or steer sub-agents. The API surface
-exists so app/editor/headless clients can inspect the same handoff receipts that
-the TUI and parent model see.
+These endpoints do not start or steer sub-agents. The API surface exists so
+app/editor/headless clients can inspect the same handoff receipts that the TUI
+and parent model see, and stop a run they are showing.
+
+`POST /v1/agent-runs/{run_id}/cancel` takes no body. It stops the run through
+the same session-scoped path as the TUI's stop and the `agent/cancel` tool:
+descendants stop with it, and a write-scoped child's changed files are named in
+its result rather than dropped. It answers with the worker record:
+
+- `200` when the record is terminal (stopping an already-finished run is a
+  no-op that returns its receipt);
+- `202` when the owning engine accepted the stop but has not recorded the
+  terminal receipt within a few seconds; poll `GET /v1/agent-runs/{run_id}`;
+- `404` for an unknown run;
+- `409` when the run belongs to a session this runtime is not hosting (for
+  example a separate terminal session); stop it from that session.
 
 ## Session lifecycle (native UI supervision)