feat(fleet): two-press Stop for writing agents and per-descendant receipts

crates/tui/src/runtime_api.rs

@@ -408,6 +408,25 @@ struct SkillsResponse {
 #[derive(Debug, Serialize)]
 struct AgentRunsResponse {
     runs: Vec<AgentWorkerRecord>,
+    /// Live launch-governor state for agents this runtime launches (Fleet
+    /// runs), so a client can say why a queued run waits (addendum F5).
+    governor: AgentRunsGovernor,
+}
+
+/// The rate-limit governor behind agent launches, as of this response.
+#[derive(Debug, Serialize)]
+struct AgentRunsGovernor {
+    /// Launch slots currently granted, after any rate-limit shrink.
+    launch_slots: usize,
+    /// Configured launch concurrency.
+    max_launch_slots: usize,
+    /// New launches are held entirely after sustained provider rate limits.
+    paused: bool,
+    /// Provider rate limits seen inside the governor's sliding window.
+    recent_rate_limits: usize,
+    /// One human line while launches are held back; absent at full speed.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    status: Option<String>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -2064,7 +2083,22 @@ async fn list_agent_runs(
     let runs = load_persisted_agent_worker_records(&state.workspace).map_err(|err| {
         ApiError::internal(format!("Failed to load persisted agent run records: {err}"))
     })?;
-    Ok(Json(AgentRunsResponse { runs }))
+    let snapshot = state
+        .sub_agent_manager
+        .read()
+        .await
+        .rate_limit_governor()
+        .snapshot(std::time::Instant::now());
+    Ok(Json(AgentRunsResponse {
+        runs,
+        governor: AgentRunsGovernor {
+            launch_slots: snapshot.launch_capacity,
+            max_launch_slots: snapshot.max_capacity,
+            paused: snapshot.paused,
+            recent_rate_limits: snapshot.window_limited,
+            status: snapshot.status_line(),
+        },
+    }))
 }
 
 async fn get_agent_run(

crates/tui/src/runtime_api/tests.rs

@@ -2822,6 +2822,14 @@ async fn agent_runs_runtime_api_exposes_persisted_worker_receipts() -> Result<()
         .json()
         .await?;
     assert_eq!(runs["runs"][0]["spec"]["run_id"], "run_receipt");
+    // F5: the payload carries the launch governor; a calm fleet has no line.
+    assert_eq!(runs["governor"]["paused"], false);
+    assert_eq!(runs["governor"]["recent_rate_limits"], 0);
+    assert_eq!(
+        runs["governor"]["launch_slots"],
+        runs["governor"]["max_launch_slots"]
+    );
+    assert!(runs["governor"].get("status").is_none());
     assert_eq!(runs["runs"][0]["follow_up"]["tool"], "handle_read");
     assert_eq!(
         runs["runs"][0]["verification"]["status"],

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

@@ -663,6 +663,91 @@ async fn cancel_appends_work_preservation_note_once() {
     assert_eq!(scout.result.as_deref(), Some(CANCELLED_BY_PARENT_RESULT));
 }
 
+/// F4: a Stop cascades to descendants, and each write-scoped descendant
+/// stopped with the parent gets its own receipt; a read-only one does not.
+#[tokio::test]
+async fn cancel_receipts_each_writing_descendant_stopped_with_its_parent() {
+    let tmp = tempdir().unwrap();
+    let root = tmp.path();
+    git(root, &["init", "--quiet"]);
+    git(root, &["config", "user.name", "Budget test"]);
+    git(root, &["config", "user.email", "budget@example.invalid"]);
+    fs::write(root.join("src.rs"), "baseline\n").unwrap();
+    git(root, &["add", "--", "src.rs"]);
+    git(root, &["commit", "--quiet", "-m", "baseline"]);
+
+    let manager = Arc::new(RwLock::new(SubAgentManager::new(root.to_path_buf(), 4)));
+    for (agent_id, write, parent) in [
+        ("tree-parent", true, None),
+        ("tree-writer", true, Some("tree-parent")),
+        ("tree-scout", false, Some("tree-parent")),
+        ("tree-stranger", true, None),
+    ] {
+        let mut spec = make_worker_spec(agent_id, root.to_path_buf());
+        spec.runtime_profile.permissions.write = write;
+        spec.parent_run_id = parent.map(str::to_string);
+        let mut guard = manager.write().await;
+        guard.register_worker(spec);
+        let (input_tx, _input_rx) = mpsc::unbounded_channel();
+        let mut agent = SubAgent::new(
+            agent_id.to_string(),
+            FleetRole::Worker,
+            "work that gets stopped".to_string(),
+            SubAgentAssignment {
+                objective: "edit".to_string(),
+                role: Some("worker".to_string()),
+            },
+            "deepseek-v4-flash".to_string(),
+            None,
+            None,
+            input_tx,
+            root.to_path_buf(),
+            guard.current_session_boot_id.clone(),
+        );
+        agent.task_handle = Some(tokio::spawn(async {
+            tokio::time::sleep(Duration::from_secs(60)).await;
+        }));
+        guard.agents.insert(agent_id.to_string(), agent);
+    }
+    fs::create_dir_all(root.join("scratch")).unwrap();
+    fs::write(root.join("scratch/half-done.rs"), "wip\n").unwrap();
+
+    // The cascade order of `cancel_agent_for_session`: descendants, then the
+    // target. The unrelated writer is stopped too but is not a descendant.
+    let parent = {
+        let mut guard = manager.write().await;
+        for id in ["tree-writer", "tree-scout", "tree-stranger"] {
+            guard.cancel_agent(id).unwrap();
+        }
+        guard.cancel_agent("tree-parent").unwrap()
+    };
+    let parent = preserve_cancelled_work(&manager, parent).await;
+    assert!(
+        parent
+            .result
+            .as_deref()
+            .is_some_and(|text| text.contains("scratch/half-done.rs")),
+        "{:?}",
+        parent.result
+    );
+
+    let guard = manager.read().await;
+    let writer = guard.get_result("tree-writer").unwrap();
+    let writer_text = writer.result.as_deref().unwrap_or_default();
+    assert!(
+        writer_text.starts_with(CANCELLED_BY_PARENT_RESULT),
+        "{writer_text}"
+    );
+    assert!(
+        writer_text.contains("scratch/half-done.rs"),
+        "{writer_text}"
+    );
+    let scout = guard.get_result("tree-scout").unwrap();
+    assert_eq!(scout.result.as_deref(), Some(CANCELLED_BY_PARENT_RESULT));
+    let stranger = guard.get_result("tree-stranger").unwrap();
+    assert_eq!(stranger.result.as_deref(), Some(CANCELLED_BY_PARENT_RESULT));
+}
+
 /// #5529: a budget death must name the work the worker left on disk. The
 /// spawn-time delivery baseline is what makes the inventory attributable to
 /// this worker rather than the parent's own dirty files.

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

@@ -3744,9 +3744,8 @@ impl SubAgentManager {
     }
 
     /// The rate-limit governor backing [`Self::launch_gate`]; exposed so the
-    /// engine can stamp it onto root runtimes and tests can drive the
-    /// adaptive scheduler. (Surfacing governor state in status events is a
-    /// parent-repo follow-up.)
+    /// engine can stamp it onto root runtimes, `GET /v1/agent-runs` can report
+    /// it (addendum F5), and tests can drive the adaptive scheduler.
     #[must_use]
     pub(crate) fn rate_limit_governor(&self) -> Arc<governor::RateLimitGovernor> {
         Arc::clone(&self.governor)
@@ -5747,6 +5746,32 @@ impl SubAgentManager {
             .map(|agent| self.snapshot_for_listing(agent))
     }
 
+    /// Write-scoped descendants of `ancestor` that the same Stop cancelled and
+    /// that carry no preservation receipt yet (F4 per-descendant receipts).
+    /// Read-only descendants are skipped: they have no baseline to inventory.
+    fn freshly_cancelled_writing_descendants(&self, ancestor: &str) -> Vec<String> {
+        self.agents
+            .values()
+            .filter(|agent| {
+                agent.id != ancestor
+                    && agent.status == SubAgentStatus::Cancelled
+                    && agent.result.as_deref() == Some(CANCELLED_BY_PARENT_RESULT)
+                    && self
+                        .worker_records
+                        .get(&agent.id)
+                        .is_some_and(|record| record.spec.runtime_profile.permissions.write)
+                    && self
+                        .ensure_caller_controls_descendant(
+                            &agent.id,
+                            Some(ancestor),
+                            "agent/cancel",
+                        )
+                        .is_ok()
+            })
+            .map(|agent| agent.id.clone())
+            .collect()
+    }
+
     /// Terminalize a child that already left `Running` but whose worker record
     /// never reached a terminal status — a child parked at the parent's turn
     /// end, or one waiting on an answer the parent has now decided not to give
@@ -11432,6 +11457,11 @@ fn budget_partial_result(
 /// isolated-worktree checkpoint a budget death gets, off the manager lock,
 /// and appends it to the child's result. Read-only children have no
 /// delivery baseline and are returned unchanged.
+///
+/// A Stop cascades to the child's descendants (`cancel_agent_for_session`),
+/// so each write-scoped descendant stopped with it gets its own receipt too:
+/// the work a grandchild left is named on the grandchild's record instead of
+/// vanishing behind the parent's single line.
 pub(crate) async fn preserve_cancelled_work(
     manager: &SharedSubAgentManager,
     snapshot: SubAgentResult,
@@ -11441,6 +11471,20 @@ pub(crate) async fn preserve_cancelled_work(
     {
         return snapshot;
     }
+    let descendants = manager
+        .read()
+        .await
+        .freshly_cancelled_writing_descendants(&snapshot.agent_id);
+    for descendant in descendants {
+        if let Some(note) =
+            budget_work_preservation_note(manager, &descendant, "cancelled with its parent").await
+        {
+            manager
+                .write()
+                .await
+                .append_cancel_preservation_note(&descendant, &note);
+        }
+    }
     let Some(note) =
         budget_work_preservation_note(manager, &snapshot.agent_id, "cancelled by parent").await
     else {