From a0130127dc5bbf716332938a483badfadef9f339 Mon Sep 17 00:00:00 2001 From: CodeWhale Bot Date: Wed, 23 Sep 2026 09:46:58 -0700 Subject: [PATCH] feat(fleet): two-press Stop for writing agents and per-descendant receipts Addenda F4/F5 (0.10.1): - F4: stopping an agent that can change files takes two presses of `X` (or `X` then Enter) in /subagents; Esc disarms, and an armed Stop clears if the agent finishes first. A Stop cascades to descendants, and each write-scoped descendant stopped with it now gets its own work-preservation receipt instead of vanishing behind the parent's single line. Read-only descendants have no baseline and get none. - F5: `GET /v1/agent-runs` returns a `governor` object (launch slots, max slots, paused, recent rate limits, status line) so a client can say why a queued run waits. docs/SUBAGENTS.md documents the governor. Mined from the unreviewed 0.10.1 WIP branch. Correction: the WIP's /subagents header line for the governor had no production caller (the TUI gets agent lists from the Engine without governor state); rather than ship an unwired view hook, it is removed and the docs state that limit. The queued-row reason and the API object are what ships. Evidence: 1612 passed, 0 failed (11,651 skipped) across budget_handback, views::, agent_runs, runtime_api::tests, subagent, golden, phase_strip and settings selections. TUI all-target/all-feature Clippy with CI flags and fmt passed; dead-code 279 and blocking-call budgets unchanged. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/tui/src/runtime_api.rs | 36 +++- crates/tui/src/runtime_api/tests.rs | 8 + .../tools/subagent/budget_handback_tests.rs | 85 +++++++++ crates/tui/src/tools/subagent/mod.rs | 50 ++++- crates/tui/src/tui/views/mod.rs | 174 +++++++++++++++++- docs/SUBAGENTS.md | 33 +++- 6 files changed, 372 insertions(+), 14 deletions(-) diff --git a/crates/tui/src/runtime_api.rs b/crates/tui/src/runtime_api.rs index 495593e866..92015e946c 100644 --- a/crates/tui/src/runtime_api.rs +++ b/crates/tui/src/runtime_api.rs @@ -408,6 +408,25 @@ struct SkillsResponse { #[derive(Debug, Serialize)] struct AgentRunsResponse { runs: Vec, + /// 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, } #[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( diff --git a/crates/tui/src/runtime_api/tests.rs b/crates/tui/src/runtime_api/tests.rs index 85e61e6a54..8b3fd5863b 100644 --- a/crates/tui/src/runtime_api/tests.rs +++ b/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"], diff --git a/crates/tui/src/tools/subagent/budget_handback_tests.rs b/crates/tui/src/tools/subagent/budget_handback_tests.rs index 885bceb219..1e8bcf8942 100644 --- a/crates/tui/src/tools/subagent/budget_handback_tests.rs +++ b/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. diff --git a/crates/tui/src/tools/subagent/mod.rs b/crates/tui/src/tools/subagent/mod.rs index 851a9e0c2a..40468aa064 100644 --- a/crates/tui/src/tools/subagent/mod.rs +++ b/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 { 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 { + 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, ¬e); + } + } let Some(note) = budget_work_preservation_note(manager, &snapshot.agent_id, "cancelled by parent").await else { diff --git a/crates/tui/src/tui/views/mod.rs b/crates/tui/src/tui/views/mod.rs index 10133c292f..fd211cf8fb 100644 --- a/crates/tui/src/tui/views/mod.rs +++ b/crates/tui/src/tui/views/mod.rs @@ -5573,6 +5573,21 @@ pub struct SubAgentsView { /// the parked roster instead of stacking a second one. Direct entry /// (`/fleet workers`, the Work dock) leaves it false, so `Esc` closes. back_to_fleet_roster: bool, + /// Agent whose Stop is armed (addendum F4). Stopping an agent that can + /// change files takes two presses of `X` (or `X` then `Enter`); `Esc` + /// or moving the selection disarms it. + armed_stop: Option, +} + +/// Whether stopping this agent can strand file work, so `X` asks twice: it +/// is still running and may write files (write permission or a full shell). +/// Rows without a permission snapshot (live progress rows) stop on one press. +fn subagent_stop_needs_confirm(agent: &SubAgentResult) -> bool { + agent.status == SubAgentStatus::Running + && agent + .runtime_permissions + .as_ref() + .is_some_and(|permissions| permissions.write || permissions.shell == "full") } /// Build the agent rows shown by `/subagents`. @@ -5730,6 +5745,7 @@ impl SubAgentsView { locale: Locale::En, opened_at: std::time::Instant::now(), back_to_fleet_roster: false, + armed_stop: None, } } @@ -5742,6 +5758,26 @@ impl SubAgentsView { view } + fn selected_agent(&self) -> Option<&SubAgentResult> { + let id = self.ordered_agent_ids().get(self.selected).cloned()?; + self.agents.iter().find(|agent| agent.agent_id == id) + } + + /// `X`: stop the selected agent. One that can change files arms first and + /// stops on the second press, so a stray key cannot end a writer. + fn press_stop(&mut self) -> ViewAction { + let Some(agent) = self.selected_agent() else { + return ViewAction::None; + }; + let agent_id = agent.agent_id.clone(); + if subagent_stop_needs_confirm(agent) && self.armed_stop.as_deref() != Some(&agent_id) { + self.armed_stop = Some(agent_id); + return ViewAction::None; + } + self.armed_stop = None; + ViewAction::Emit(ViewEvent::SidebarAgentCancel { agent_id }) + } + /// Mark this view as pushed on top of the Fleet roster (#5954), so the /// footer says `back` and `F` pops to the parked roster. #[must_use] @@ -5839,6 +5875,20 @@ impl ModalView for SubAgentsView { fn handle_key(&mut self, key: KeyEvent) -> ViewAction { use crossterm::event::KeyCode; + // An armed Stop (F4) owns Esc and Enter: Esc disarms without closing, + // Enter confirms — the same two-step the Work inspector's Stop uses. + if self.armed_stop.is_some() { + match key.code { + KeyCode::Esc => { + self.armed_stop = None; + return ViewAction::None; + } + KeyCode::Enter => return self.press_stop(), + KeyCode::Char('x') | KeyCode::Char('X') => {} + _ => self.armed_stop = None, + } + } + match key.code { KeyCode::Esc | KeyCode::Char('q') => ViewAction::Close, // Enter opens the selected agent's transcript — the same primary @@ -5851,14 +5901,10 @@ impl ModalView for SubAgentsView { KeyCode::Char('r') | KeyCode::Char('R') => { ViewAction::Emit(ViewEvent::SubAgentsRefresh) } - // Manage: stop the selected worker. Terminal workers ignore the - // key; the cancel receipt names what happened either way. - KeyCode::Char('x') | KeyCode::Char('X') => { - match self.ordered_agent_ids().get(self.selected).cloned() { - Some(agent_id) => ViewAction::Emit(ViewEvent::SidebarAgentCancel { agent_id }), - None => ViewAction::None, - } - } + // Manage: stop the selected agent. Terminal agents ignore the + // key; the cancel receipt names what happened either way. A + // running agent that can change files asks twice (F4). + KeyCode::Char('x') | KeyCode::Char('X') => self.press_stop(), // The roster is the same destination either way: pop back to the // parked one when there is one (#5954) — re-running `/fleet` // would stack a duplicate roster and lose its cursor. @@ -5942,6 +5988,15 @@ impl ModalView for SubAgentsView { .position(|candidate| candidate == &id) }) .unwrap_or_else(|| self.selected.min(last)); + // An armed Stop only survives while its agent is still selected and + // still needs the confirm (it may have finished meanwhile). + let still_armed = self.armed_stop.as_deref().is_some_and(|armed| { + self.selected_agent() + .is_some_and(|agent| agent.agent_id == armed && subagent_stop_needs_confirm(agent)) + }); + if !still_armed { + self.armed_stop = None; + } true } @@ -6094,7 +6149,14 @@ impl ModalView for SubAgentsView { ActionHint::new("Esc", self.esc_hint_label()), ActionHint::new("↑/↓", tr(self.locale, MessageId::CtxInspActionSelect)), ActionHint::new("Enter", tr(self.locale, MessageId::ExtensionsActionFocus)), - ActionHint::new("X", tr(self.locale, MessageId::SidebarStopControl)), + if self.armed_stop.is_some() { + ActionHint::new( + "X/Enter", + tr(self.locale, MessageId::WorkSurfaceStopConfirmHint), + ) + } else { + ActionHint::new("X", tr(self.locale, MessageId::SidebarStopControl)) + }, ActionHint::new("R", tr(self.locale, MessageId::SubagentsActionRefresh)), ActionHint::new("F", tr(self.locale, MessageId::SubagentsActionRosterSetup)), ], @@ -7191,6 +7253,100 @@ mod tests { } } + fn writer_agent(id: &str) -> SubAgentResult { + let mut agent = manager_agent(id, SubAgentStatus::Running); + agent.runtime_permissions = Some(codewhale_protocol::fleet::FleetEffectivePermissions { + write: true, + network: false, + shell: "read_only".to_string(), + tool_scope: "inherit".to_string(), + tools: Vec::new(), + background: false, + max_spawn_depth: 0, + profile_id: None, + profile_origin: None, + source: "test".to_string(), + }); + agent + } + + fn press(view: &mut SubAgentsView, code: KeyCode) -> ViewAction { + view.handle_key(KeyEvent::new(code, KeyModifiers::NONE)) + } + + fn is_stop_of(action: &ViewAction, id: &str) -> bool { + matches!( + action, + ViewAction::Emit(ViewEvent::SidebarAgentCancel { agent_id }) if agent_id == id + ) + } + + #[test] + fn stopping_a_writing_agent_takes_two_presses_and_esc_disarms() { + let mut view = SubAgentsView::new(vec![writer_agent("w")]); + + // First X arms; nothing is stopped yet and the footer asks to confirm. + assert!(matches!( + press(&mut view, KeyCode::Char('x')), + ViewAction::None + )); + let area = Rect::new(0, 0, 100, 20); + let mut buf = Buffer::empty(area); + view.render(area, &mut buf); + assert!(buffer_text(&buf, area).contains("X/Enter")); + + // Esc disarms without closing the register. + assert!(matches!(press(&mut view, KeyCode::Esc), ViewAction::None)); + assert!(view.armed_stop.is_none()); + + // X, X stops; X then Enter stops too. + assert!(matches!( + press(&mut view, KeyCode::Char('X')), + ViewAction::None + )); + assert!(is_stop_of(&press(&mut view, KeyCode::Char('X')), "w")); + assert!(matches!( + press(&mut view, KeyCode::Char('x')), + ViewAction::None + )); + assert!(is_stop_of(&press(&mut view, KeyCode::Enter), "w")); + assert!(view.armed_stop.is_none()); + } + + #[test] + fn stopping_a_read_only_or_moved_selection_does_not_need_the_armed_press() { + // A read-only running agent (no write, no full shell) stops at once. + let mut read_only = writer_agent("r"); + if let Some(permissions) = read_only.runtime_permissions.as_mut() { + permissions.write = false; + } + let mut view = SubAgentsView::new(vec![read_only]); + assert!(is_stop_of(&press(&mut view, KeyCode::Char('x')), "r")); + + // Moving the selection disarms: the next X on the other writer arms + // afresh instead of stopping it. + let mut view = SubAgentsView::new(vec![writer_agent("a"), writer_agent("b")]); + assert!(matches!( + press(&mut view, KeyCode::Char('x')), + ViewAction::None + )); + press(&mut view, KeyCode::Down); + assert!(view.armed_stop.is_none()); + assert!(matches!( + press(&mut view, KeyCode::Char('x')), + ViewAction::None + )); + assert!(is_stop_of(&press(&mut view, KeyCode::Char('x')), "b")); + + // An armed agent that finishes before the confirm is disarmed. + let mut view = SubAgentsView::new(vec![writer_agent("w")]); + press(&mut view, KeyCode::Char('x')); + let mut done = writer_agent("w"); + done.status = SubAgentStatus::Completed; + view.update_subagents(&[done]); + assert!(view.armed_stop.is_none()); + } + #[test] fn worker_register_update_preserves_selected_agent_across_new_spawns() { let mut view = SubAgentsView::new(vec![manager_agent("b", SubAgentStatus::Running)]); diff --git a/docs/SUBAGENTS.md b/docs/SUBAGENTS.md index 253e3219c7..ffb3f5a959 100644 --- a/docs/SUBAGENTS.md +++ b/docs/SUBAGENTS.md @@ -430,7 +430,7 @@ request broad fan-out and let the manager drain it without creating an unbounded population. By default every admitted child may start immediately — there is no artificial -throttle. Request the fan-out the work actually needs and let the runtime +throttle beyond the rate-limit governor described below. Request the fan-out the work actually needs and let the runtime queue and drain it; the caps above are enforcement, not a reason to pre-refuse valid work. If you want gentler fan-out, lower `[subagents].launch_concurrency` (how many direct children start at once); children beyond that limit **queue** @@ -445,6 +445,29 @@ instantaneous execution bounded. Completed / failed / cancelled records persist for inspection but don't occupy an admission slot. Agents that lost their `task_handle` (e.g. across a process restart) also don't count against the cap. +### Rate-limit governor + +The one automatic throttle is the rate-limit governor. It watches provider +rate limits (HTTP 429) across a 60-second window. After repeated limits it +shrinks the number of launch slots; under a sustained burst it pauses new +launches entirely. Steady successes add slots back one at a time. It never +interrupts an agent that is already running, and quota exhaustion is not +treated as a throttle. + +While the governor is holding launches back, it says so in two places: + +- a queued agent's row gives the reason, for example + `launch slots throttled to 4/8 after 2 provider rate limit(s) in the last 60s` + or `launches paused after 4 provider rate limit(s) in the last 60s`, and the + time its wall budget ends; +- `GET /v1/agent-runs` returns a `governor` object next to `runs`, with + `launch_slots`, `max_launch_slots`, `paused`, `recent_rate_limits`, and a + `status` line while launches are held back. It describes launches made by the + runtime serving the request (Fleet runs). + +Known limitation: the `/subagents` register header does not show the governor +line yet; the TUI receives agent lists from the Engine without governor state. + Provider profiles let one config stay aggressive for direct API routes while keeping subscription or aggregator routes gentle. Every key under `[subagents.providers.]` inherits from `[subagents]` when omitted. @@ -575,6 +598,14 @@ zero representation for that default never cancels a finite inherited cap. 1800-second default. It includes admission queue time, model requests, and tools. The effective absolute deadline is persisted. +The wall clock starts when the agent is started, not when it gets a launch +slot. This is deliberate. The queue wait and the run share one deadline, so a +saturated or rate-limited fleet cannot keep an agent alive past the budget +you gave it. The cost is that time spent queued is time taken from the run. +The queued row says so instead of hiding it: it names the reason for the wait +and the time the wall budget ends. If agents regularly spend a large share of +their budget queued, start fewer at once or raise `wall_time_secs`. + For example, a focused review can request: ```json