diff --git a/crates/stella-cli/src/claims.rs b/crates/stella-cli/src/claims.rs index e0716bb8c..5352ae7dd 100644 --- a/crates/stella-cli/src/claims.rs +++ b/crates/stella-cli/src/claims.rs @@ -349,6 +349,14 @@ impl ToolExecutor for ClaimTap<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// Forwarded: letting the empty default stand would silently serialize the + /// inner executor's sibling spawns (see the port's contract). The spawn + /// tool names no mutating path and no transient lane, so concurrent + /// siblings pass through `execute` above without touching a claim. + fn parallel_safe_names(&self) -> std::collections::HashSet { + self.inner.parallel_safe_names() + } } #[cfg(test)] @@ -372,12 +380,28 @@ mod tests { content: "ok".into(), } } + fn parallel_safe_names(&self) -> std::collections::HashSet { + std::collections::HashSet::from(["task".to_string()]) + } } fn store() -> Arc { Arc::new(Store::in_memory().unwrap()) } + /// The tap sits directly under the engine in every deck lane, so a tap + /// that swallowed the claim (the default is empty) would serialize + /// sibling spawns no matter what the registry advertised. + #[test] + fn the_claim_tap_forwards_parallel_safe_names() { + let inner = Passthrough(Default::default()); + let tap = ClaimTap::new(&inner, None, "ses-1/lead"); + assert!( + tap.parallel_safe_names().contains("task"), + "the inner executor's concurrency claim must survive the tap" + ); + } + #[tokio::test] async fn first_write_claims_and_a_rival_is_refused_with_the_holder_named() { let store = store(); diff --git a/crates/stella-cli/src/command_deck.rs b/crates/stella-cli/src/command_deck.rs index ae71e4bb1..3ab7a34aa 100644 --- a/crates/stella-cli/src/command_deck.rs +++ b/crates/stella-cli/src/command_deck.rs @@ -88,7 +88,7 @@ use stella_pipeline::{ }; use stella_protocol::{ AgentEvent, CiStatus, CompletionMessage, CompletionRequest, ModelRef, PrStatus, TaskItem, - ToolOutput, ToolSchema, + ToolOutput, }; use stella_store::Store; use stella_tools::ToolRegistry; @@ -118,6 +118,7 @@ mod scope_gate; mod session_clear; mod sessions_view; mod settle; +mod task_tap; mod theme_cmd; use crate::memory::{SessionMemory, inject_recall_block}; use crate::runtime::{SystemClock, TokioSleeper}; @@ -126,6 +127,7 @@ use authoring::{agents_list_creating, agents_list_inbound, handle_agent_create}; pub(crate) use forwarder::spawn_forwarder; use scope_gate::DeckApprovalGate; use sessions_view::sessions_inbound; +use task_tap::TaskTap; /// The lead agent's id — the one conversation this driver runs. pub(crate) const LEAD: &str = "lead"; @@ -4641,51 +4643,5 @@ impl AskUserIo for DeckAskUserIo { } } -/// Mirrors the task board into the event stream: after any `task_*` tool -/// call the FULL board snapshot rides the turn's channel as -/// `AgentEvent::TaskUpdate` — persisted by the forwarder, so replay shows -/// the checklist exactly as it moved — and `task_assign`'s spawn requests -/// are handed to the driver's supervisor channel. `supervisor: None` is the -/// worker configuration (v1 delegation runs from the lead only; a worker's -/// stranded requests are reported on its lane by `crate::subsession`). -pub(crate) struct TaskTap<'a> { - pub(crate) inner: &'a dyn ToolExecutor, - pub(crate) events: UnboundedSender, - pub(crate) registry: &'a ToolRegistry, - pub(crate) supervisor: Option>, -} - -#[async_trait] -impl ToolExecutor for TaskTap<'_> { - fn schemas(&self) -> Vec { - self.inner.schemas() - } - - async fn execute(&self, name: &str, input: &Value) -> ToolOutput { - let output = self.inner.execute(name, input).await; - if name.starts_with("task_") { - let tasks: Vec = { - let board = self.registry.task_board(); - let guard = board.lock().unwrap_or_else(|p| p.into_inner()); - guard.items().to_vec() - }; - let _ = self.events.send(AgentEvent::TaskUpdate { tasks }); - if let Some(sup) = &self.supervisor { - for request in self.registry.take_spawn_requests() { - let _ = sup.send(SupervisorMsg::SpawnTask(request)); - } - } - } - output - } - - /// Forwarded: this is a decorator, and a decorator that let the default - /// `0.0` stand would silently drop sub-agent spend out of the parent's - /// budget (see the port's contract). - fn drain_sub_agent_spend_usd(&self) -> f64 { - self.inner.drain_sub_agent_spend_usd() - } -} - #[cfg(test)] mod tests; diff --git a/crates/stella-cli/src/command_deck/task_tap.rs b/crates/stella-cli/src/command_deck/task_tap.rs new file mode 100644 index 000000000..0096493f0 --- /dev/null +++ b/crates/stella-cli/src/command_deck/task_tap.rs @@ -0,0 +1,107 @@ +//! The deck's task-board tap — split out of `command_deck.rs` (a god file +//! closed to growth) when the tap grew its `parallel_safe_names` forward. + +use async_trait::async_trait; +use serde_json::Value; +use stella_core::ports::ToolExecutor; +use stella_protocol::{AgentEvent, TaskItem, ToolOutput, ToolSchema}; +use stella_tools::ToolRegistry; +use tokio::sync::mpsc::UnboundedSender; + +use crate::subsession::SupervisorMsg; + +/// Mirrors the task board into the event stream: after any `task_*` tool +/// call the FULL board snapshot rides the turn's channel as +/// `AgentEvent::TaskUpdate` — persisted by the forwarder, so replay shows +/// the checklist exactly as it moved — and `task_assign`'s spawn requests +/// are handed to the driver's supervisor channel. `supervisor: None` is the +/// worker configuration (v1 delegation runs from the lead only; a worker's +/// stranded requests are reported on its lane by `crate::subsession`). +pub(crate) struct TaskTap<'a> { + pub(crate) inner: &'a dyn ToolExecutor, + pub(crate) events: UnboundedSender, + pub(crate) registry: &'a ToolRegistry, + pub(crate) supervisor: Option>, +} + +#[async_trait] +impl ToolExecutor for TaskTap<'_> { + fn schemas(&self) -> Vec { + self.inner.schemas() + } + + async fn execute(&self, name: &str, input: &Value) -> ToolOutput { + let output = self.inner.execute(name, input).await; + if name.starts_with("task_") { + let tasks: Vec = { + let board = self.registry.task_board(); + let guard = board.lock().unwrap_or_else(|p| p.into_inner()); + guard.items().to_vec() + }; + let _ = self.events.send(AgentEvent::TaskUpdate { tasks }); + if let Some(sup) = &self.supervisor { + for request in self.registry.take_spawn_requests() { + let _ = sup.send(SupervisorMsg::SpawnTask(request)); + } + } + } + output + } + + /// Forwarded: this is a decorator, and a decorator that let the default + /// `0.0` stand would silently drop sub-agent spend out of the parent's + /// budget (see the port's contract). + fn drain_sub_agent_spend_usd(&self) -> f64 { + self.inner.drain_sub_agent_spend_usd() + } + + /// Forwarded: letting the empty default stand would silently serialize + /// the inner executor's sibling spawns (see the port's contract). The + /// spawn tool is `task`, not `task_*` — the tap never fires for it. + fn parallel_safe_names(&self) -> std::collections::HashSet { + self.inner.parallel_safe_names() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A leaf claiming one parallel-safe name, standing in for the registry. + struct Claiming; + + #[async_trait] + impl ToolExecutor for Claiming { + fn schemas(&self) -> Vec { + Vec::new() + } + async fn execute(&self, _name: &str, _input: &Value) -> ToolOutput { + ToolOutput::Ok { + content: String::new(), + } + } + fn parallel_safe_names(&self) -> std::collections::HashSet { + std::collections::HashSet::from(["task".to_string()]) + } + } + + /// The deck's lead lane wraps the whole stack in this tap last, so a tap + /// that swallowed the claim would kill concurrent sibling spawns for + /// every deck session no matter what the layers below advertised. + #[test] + fn the_task_tap_forwards_parallel_safe_names() { + let inner = Claiming; + let registry = ToolRegistry::with_issue_backend(std::path::PathBuf::from("."), None); + let (events, _rx) = tokio::sync::mpsc::unbounded_channel(); + let tap = TaskTap { + inner: &inner, + events, + registry: ®istry, + supervisor: None, + }; + assert!( + tap.parallel_safe_names().contains("task"), + "the tap must forward the inner executor's concurrency claims" + ); + } +} diff --git a/crates/stella-cli/src/discovery.rs b/crates/stella-cli/src/discovery.rs index 5875b1ab2..ab37390d5 100644 --- a/crates/stella-cli/src/discovery.rs +++ b/crates/stella-cli/src/discovery.rs @@ -785,6 +785,14 @@ impl ToolExecutor for DiscoveryToolSet<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// Forwarded: letting the empty default stand would silently serialize the + /// inner executor's sibling spawns (see the port's contract). No lean-mode + /// intersection — hiding is a prompt-budget measure, and a hidden tool + /// called by name still executes above. + fn parallel_safe_names(&self) -> std::collections::HashSet { + self.inner.parallel_safe_names() + } } /// Split an advertised `mcp__server__tool` name into (server, tool). diff --git a/crates/stella-cli/src/fleet_commits.rs b/crates/stella-cli/src/fleet_commits.rs index a2909b19a..fe9f389f7 100644 --- a/crates/stella-cli/src/fleet_commits.rs +++ b/crates/stella-cli/src/fleet_commits.rs @@ -131,6 +131,13 @@ impl ToolExecutor for CommitObserver<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// Forwarded: letting the empty default stand would silently serialize + /// the inner executor's sibling spawns (see the port's contract) — the + /// spawn tool is not commit-lane routed, so nothing is observed for it. + fn parallel_safe_names(&self) -> std::collections::HashSet { + self.inner.parallel_safe_names() + } } /// One attempt's commits, oldest first — and the two ways of knowing which @@ -317,6 +324,10 @@ mod tests { Vec::new() } + fn parallel_safe_names(&self) -> std::collections::HashSet { + std::collections::HashSet::from(["task".to_string()]) + } + async fn execute(&self, name: &str, _input: &Value) -> ToolOutput { if name == "repo_commit" { let mut script = self.script.lock().unwrap(); @@ -349,6 +360,21 @@ mod tests { commits.iter().map(|c| c.sha.as_str()).collect() } + /// The observer wraps every shared-tree fleet worker's stack, so one that + /// swallowed the claim (the default is empty) would serialize sibling + /// spawns for the whole fleet no matter what the registry advertised. + #[test] + fn the_commit_observer_forwards_parallel_safe_names() { + let tree = Arc::new(SharedTree::default()); + let tools = CommittingTools::new(tree.clone(), Vec::new()); + assert!( + observer(&tools, &tree, "t1") + .parallel_safe_names() + .contains("task"), + "the inner executor's concurrency claim must survive the observer" + ); + } + /// The witness for #1216's first half. Two shared-tree workers commit /// into ONE tree, interleaved — `t1`, then `t2`, then `t1` again. Every /// commit must appear under the task that made it and under no other. diff --git a/crates/stella-cli/src/interactive.rs b/crates/stella-cli/src/interactive.rs index d63fa7330..7d8c67f91 100644 --- a/crates/stella-cli/src/interactive.rs +++ b/crates/stella-cli/src/interactive.rs @@ -527,6 +527,13 @@ impl ToolExecutor for InteractiveToolSet<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// Forwarded: letting the empty default stand would silently serialize the + /// inner executor's sibling spawns (see the port's contract). The tools + /// added here (`ask_user`, skills) make no concurrency claim of their own. + fn parallel_safe_names(&self) -> std::collections::HashSet { + self.inner.parallel_safe_names() + } } #[cfg(test)] diff --git a/crates/stella-cli/src/subagent.rs b/crates/stella-cli/src/subagent.rs index 77492597b..c64cf867f 100644 --- a/crates/stella-cli/src/subagent.rs +++ b/crates/stella-cli/src/subagent.rs @@ -47,10 +47,12 @@ //! `task` calls from one step execute concurrently — reachable since the //! engine's dispatch scheduler groups the spawn tool with read-only calls //! (`Tool::parallel_safe`; before that claim existed, every non-read-only -//! call was its own barrier and this concurrency was dead code). Two siblings -//! can therefore each carve against the same headroom before either settles — -//! an overshoot bounded by one child's cap, and caught by the parent's guard -//! at the next step boundary regardless. +//! call was its own barrier and this concurrency was dead code). Every +//! sibling in one dispatch group can therefore carve against the same +//! headroom before any settles — an overshoot bounded by one child's cap per +//! concurrently-running sibling (up to the engine's dispatch cap of 8, so +//! seven extra caps in the worst case, not one), and caught by the parent's +//! guard at the next step boundary regardless. //! //! ## Settling is the child's job //! diff --git a/crates/stella-cli/src/subagent/tests.rs b/crates/stella-cli/src/subagent/tests.rs index 75cc9f42d..6dc89a2b9 100644 --- a/crates/stella-cli/src/subagent/tests.rs +++ b/crates/stella-cli/src/subagent/tests.rs @@ -34,6 +34,11 @@ impl ToolExecutor for LedgerBase { fn drain_sub_agent_spend_usd(&self) -> f64 { stella_core::subagent::drain_sub_agent_spend(&self.0) } + fn parallel_safe_names(&self) -> std::collections::HashSet { + // The registry claims exactly this for the `task` tool + // (`Tool::parallel_safe`); the stand-in reports it the same way. + std::collections::HashSet::from(["task".to_string()]) + } } struct NeverProvider; @@ -101,6 +106,39 @@ async fn the_production_tool_stack_forwards_sub_agent_spend() { ); } +/// **The parallel-dispatch counterpart of the spend witness above.** +/// +/// `parallel_safe_names` has an empty default, so any decorator that forgets +/// to forward it silently serializes sibling `task` calls in every real +/// session — the registry's claim never reaches the engine, and the feature +/// (#1776) is dead exactly where it shipped. Asserted through the shipped +/// composition for the same reason as the spend test: a future decorator +/// inserted into the real stack fails here, and nowhere else. +#[tokio::test] +async fn the_production_tool_stack_forwards_parallel_safe_names() { + let ledger: SubAgentSpendLedger = Arc::default(); + let base = LedgerBase(ledger); + + let customs = + stella_tools::custom::CustomToolSet::new(&base, Vec::new(), std::path::PathBuf::from(".")); + let (stub_tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + let interactive = crate::interactive::InteractiveToolSet::new( + &customs, + stub_tx, + crate::interactive::default_ask_io(false), + ); + let permitted = crate::agent::PolicyToolSet::new(&interactive, Default::default()); + let discovery = + crate::discovery::DiscoveryToolSet::new(&permitted, std::path::PathBuf::from(".")); + + assert!( + discovery.parallel_safe_names().contains("task"), + "the registry's concurrency claim must survive every decorator \ + between the engine and the registry — one that swallows it silently \ + serializes sibling spawns" + ); +} + /// A dispatcher whose registry has been dropped reports a refusal rather /// than panicking a torn-down session. #[tokio::test] diff --git a/crates/stella-cli/src/tool_policy.rs b/crates/stella-cli/src/tool_policy.rs index 1fa2361fc..ea492261e 100644 --- a/crates/stella-cli/src/tool_policy.rs +++ b/crates/stella-cli/src/tool_policy.rs @@ -102,6 +102,16 @@ impl ToolExecutor for PolicyToolSet<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.get().drain_sub_agent_spend_usd() } + + /// Forwarded minus what the policy withholds. Letting the empty default + /// stand would silently serialize the inner executor's sibling spawns — + /// but blindly delegating would advertise names `execute` above refuses, + /// an empty promise. Same two-sided shape as `schemas`/`execute`. + fn parallel_safe_names(&self) -> std::collections::HashSet { + let mut names = self.inner.get().parallel_safe_names(); + names.retain(|name| self.policy.allows(name)); + names + } } /// Which `"tools"` key withheld `name` — the exact name, its group, or the @@ -161,6 +171,9 @@ mod tests { content: format!("ran {name}"), } } + fn parallel_safe_names(&self) -> std::collections::HashSet { + std::collections::HashSet::from(["task".to_string()]) + } } fn names(set: &PolicyToolSet<'_>) -> Vec { @@ -259,6 +272,32 @@ mod tests { ); } + /// The concurrency claim survives the decorator when the policy allows + /// the tool — a set that forgot to forward would silently serialize + /// sibling spawns in every real session (the default is empty). + #[test] + fn parallel_safe_names_are_forwarded_when_the_policy_allows() { + let fake = Fake; + let set = PolicyToolSet::new(&fake, ToolPolicy::allow_all()); + assert!( + set.parallel_safe_names().contains("task"), + "the inner executor's claim must survive the policy layer" + ); + } + + /// The other half of the two-sided shape: a tool the policy withholds is + /// refused by `execute`, so advertising it as parallel-safe would claim + /// concurrency for a call this decorator will never run. + #[test] + fn a_disabled_tool_is_not_advertised_as_parallel_safe() { + let fake = Fake; + let set = PolicyToolSet::new(&fake, ToolPolicy::from_switches([("task".into(), false)])); + assert!( + set.parallel_safe_names().is_empty(), + "a withheld tool must not be advertised as parallel-safe" + ); + } + #[tokio::test] async fn deny_by_default_leaves_only_what_is_re_enabled() { let fake = Fake; diff --git a/crates/stella-mcp/src/toolset.rs b/crates/stella-mcp/src/toolset.rs index 5d1821eb2..cdcd91d4b 100644 --- a/crates/stella-mcp/src/toolset.rs +++ b/crates/stella-mcp/src/toolset.rs @@ -560,6 +560,16 @@ impl ToolExecutor for McpToolSet { .as_ref() .map_or(0.0, |native| native.drain_sub_agent_spend_usd()) } + + /// Forwarded from the native layer: letting the empty default stand would + /// silently serialize its sibling spawns (see the port's contract). + /// External MCP tools never carry this claim — the port pins their + /// concurrency to `read_only`, which they are advertised without. + fn parallel_safe_names(&self) -> std::collections::HashSet { + self.native + .as_ref() + .map_or_else(Default::default, |native| native.parallel_safe_names()) + } } /// A Best-of-N candidate's tool surface (issue #248 Phase 1): built by @@ -613,6 +623,13 @@ impl ToolExecutor for CandidateMcpView { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// Forwarded from the candidate's own `native` layer — the executor every + /// non-`mcp__` name routes to in `execute` above. `inner`'s set would name + /// tools this view never runs, and MCP tools never carry the claim. + fn parallel_safe_names(&self) -> std::collections::HashSet { + self.native.parallel_safe_names() + } } /// Compose the namespaced tool name for a server/tool pair. @@ -667,6 +684,34 @@ mod tests { content: format!("native ran {name}"), } } + fn parallel_safe_names(&self) -> HashSet { + HashSet::from(["task".to_string()]) + } + } + + /// Both wrappers forward the native layer's concurrency claim — the empty + /// default would silently serialize sibling spawns in any MCP-connected + /// session (and in every Best-of-N candidate). + #[tokio::test] + async fn parallel_safe_names_forward_from_the_native_layer() { + let client = connected_client("files", "read").await; + let set = Arc::new(McpToolSet::from_clients(vec![client]).wrapping(Arc::new(FakeNative))); + assert!( + set.parallel_safe_names().contains("task"), + "the native layer's claim must survive the MCP set" + ); + + let view = set.for_candidates(Arc::new(FakeNative)); + assert!( + view.parallel_safe_names().contains("task"), + "and the candidate view forwards its own native layer's claim" + ); + + let bare = McpToolSet::from_clients(Vec::new()); + assert!( + bare.parallel_safe_names().is_empty(), + "no native layer, no claims" + ); } /// A transport that never answers `tools/call` — used to prove the diff --git a/crates/stella-serve/src/subagents.rs b/crates/stella-serve/src/subagents.rs index 6d5733ed8..f5b4c686e 100644 --- a/crates/stella-serve/src/subagents.rs +++ b/crates/stella-serve/src/subagents.rs @@ -437,6 +437,14 @@ impl ToolExecutor for DelegatingTools<'_> { self.inner.drain_sub_agent_spend_usd() + stella_core::subagent::drain_sub_agent_spend(&self.spend) } + + /// Forwarded: letting the empty default stand would silently serialize + /// the host executor's sibling spawns (see the port's contract). The + /// `task` this wrapper itself implements is deliberately not added here — + /// claiming it needs a witness on the remoted dispatch path first. + fn parallel_safe_names(&self) -> std::collections::HashSet { + self.inner.parallel_safe_names() + } } /// Turn a child's outcome into a model-visible result. @@ -508,6 +516,55 @@ fn slug(description: &str) -> String { mod tests { use super::*; + /// A host executor claiming one parallel-safe name. + struct ClaimingHost; + + #[async_trait] + impl ToolExecutor for ClaimingHost { + fn schemas(&self) -> Vec { + Vec::new() + } + async fn execute(&self, _name: &str, _input: &Value) -> ToolOutput { + ToolOutput::Ok { + content: String::new(), + } + } + fn parallel_safe_names(&self) -> std::collections::HashSet { + std::collections::HashSet::from(["host_task".to_string()]) + } + } + + /// A dispatcher that refuses everything — the wrapper under test never + /// dispatches in this test, so any answer would do. + struct RefusingDispatcher; + + #[async_trait] + impl SubAgentDispatcher for RefusingDispatcher { + async fn dispatch(&self, _spec: SubAgentSpec) -> SubAgentOutcome { + SubAgentOutcome::Refused { + reason: "not in this test".into(), + } + } + } + + /// The wrapper sits between the served engine and the host's executor, so + /// one that swallowed the claim (the default is empty) would serialize + /// the host's sibling spawns on every served turn. + #[test] + fn delegating_tools_forward_parallel_safe_names() { + let host = ClaimingHost; + let tools = DelegatingTools::new( + &host, + Arc::new(RefusingDispatcher), + 4, + stella_core::subagent::SubAgentSpendLedger::default(), + ); + assert!( + tools.parallel_safe_names().contains("host_task"), + "the host executor's concurrency claim must survive the wrapper" + ); + } + /// The operator's ceilings bound every caller-settable knob, and a /// deployment that has not opted in gets no children whatever the request /// asks for. diff --git a/crates/stella-tools/src/custom.rs b/crates/stella-tools/src/custom.rs index 60327e788..6aaa33d8d 100644 --- a/crates/stella-tools/src/custom.rs +++ b/crates/stella-tools/src/custom.rs @@ -768,6 +768,13 @@ impl ToolExecutor for CustomToolSet<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.get().drain_sub_agent_spend_usd() } + + /// Forwarded: letting the empty default stand would silently serialize the + /// inner executor's sibling spawns (see the port's contract). Custom + /// script tools spawn workspace subprocesses and make no such claim. + fn parallel_safe_names(&self) -> std::collections::HashSet { + self.inner.get().parallel_safe_names() + } } #[cfg(test)] diff --git a/crates/stella-tools/src/hunk_review.rs b/crates/stella-tools/src/hunk_review.rs index 910d4348a..b06d58c17 100644 --- a/crates/stella-tools/src/hunk_review.rs +++ b/crates/stella-tools/src/hunk_review.rs @@ -456,6 +456,13 @@ impl ToolExecutor for HunkGate<'_> { fn drain_sub_agent_spend_usd(&self) -> f64 { self.inner.drain_sub_agent_spend_usd() } + + /// Forwarded — letting the empty default stand would silently serialize + /// the inner executor's sibling spawns (see the port's contract); the + /// spawn tool is not in `GATED_TOOLS`, so the gate never reviews it. + fn parallel_safe_names(&self) -> std::collections::HashSet { + self.inner.parallel_safe_names() + } } #[cfg(test)] @@ -531,6 +538,20 @@ mod tests { serde_json::json!({ "path": path, "old_string": old, "new_string": new }) } + /// The gate wraps the registry in every reviewed session, so one that + /// swallowed the claim (the default is empty) would serialize sibling + /// spawns exactly where review mode ships. Asserted over the real + /// registry, whose `task` tool is the one shipped claimant. + #[test] + fn the_gate_forwards_parallel_safe_names() { + let (dir, reg, _) = fixture(); + let gate = HunkGate::new(®, None, dir.path().to_path_buf()); + assert!( + gate.parallel_safe_names().contains("task"), + "the registry's concurrency claim must survive the gate" + ); + } + /// The issue's acceptance criterion, end to end and through the real /// `apply_edits` transaction: two hunks proposed, one declined, exactly one /// applied. diff --git a/crates/stella-tools/src/registry.rs b/crates/stella-tools/src/registry.rs index 1e46630fc..794f759fe 100644 --- a/crates/stella-tools/src/registry.rs +++ b/crates/stella-tools/src/registry.rs @@ -2169,13 +2169,13 @@ impl ToolExecutor for ToolRegistry { .filter(|(_, tool)| tool.parallel_safe()) .map(|(name, _)| name.clone()) .collect(); - if let Ok(late) = self.late_tools.read() { - names.extend( - late.iter() - .filter(|(_, tool)| tool.parallel_safe()) - .map(|(name, _)| name.clone()), - ); - } + // Poison-tolerant like every sibling `late_tools` read path. + let late = self.late_tools.read().unwrap_or_else(|p| p.into_inner()); + names.extend( + late.iter() + .filter(|(_, tool)| tool.parallel_safe()) + .map(|(name, _)| name.clone()), + ); names } } diff --git a/crates/stella-tools/src/registry/tests.rs b/crates/stella-tools/src/registry/tests.rs index 82b8948e4..b4af90413 100644 --- a/crates/stella-tools/src/registry/tests.rs +++ b/crates/stella-tools/src/registry/tests.rs @@ -1421,3 +1421,64 @@ fn a_beliefs_coverage_survives_the_snapshot_round_trip() { "a partial belief is still partial after a resume" ); } + +/// A minimal parallel-safe tool for the late-enabled overlay. +struct LateSpawn; + +#[async_trait] +impl Tool for LateSpawn { + fn schema(&self) -> ToolSchema { + ToolSchema { + name: "late_spawn".into(), + description: "late-enabled spawn stand-in".into(), + input_schema: serde_json::json!({ "type": "object" }), + read_only: false, + speculation_safe: false, + } + } + fn parallel_safe(&self) -> bool { + true + } + async fn execute(&self, _input: &Value, _root: &std::path::Path) -> ToolOutput { + ToolOutput::Ok { + content: String::new(), + } + } +} + +/// The witness for the poisoned-lock repair in +/// `ToolExecutor::parallel_safe_names`: every sibling `late_tools` read path +/// tolerates poison (`unwrap_or_else(|p| p.into_inner())`), but this one used +/// `if let Ok(..)` — after a panic elsewhere it silently dropped every +/// late-enabled claim while `schemas` kept advertising the tool, serializing +/// dispatch with nothing to say so. +#[test] +fn parallel_safe_names_survive_a_poisoned_late_overlay() { + let (_root, reg) = bare_registry(None); + reg.late_tools + .write() + .unwrap_or_else(|p| p.into_inner()) + .insert("late_spawn".to_string(), Arc::new(LateSpawn)); + + // Poison the lock the only way it happens in production: a panic while + // the guard is held. + let _ = std::thread::scope(|scope| { + scope + .spawn(|| { + let _guard = reg.late_tools.write().unwrap(); + panic!("poison the overlay"); + }) + .join() + }); + assert!(reg.late_tools.is_poisoned(), "the setup really poisoned it"); + + let names = stella_core::ports::ToolExecutor::parallel_safe_names(®); + assert!( + names.contains("late_spawn"), + "a poisoned overlay must not silently shrink the claim set" + ); + assert!( + names.contains("task"), + "and the primary map's claims are untouched" + ); +}