diff --git a/CHANGELOG.md b/CHANGELOG.md index a15926a41..1e56154da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,21 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - `zeph-core`: new `Channel::send_skill_catalog` method (default no-op) and `AgentEvent::SkillCatalog`, forwarded by `AnyChannel`, `GatewayChannel`, and `AppChannel`, overridden by `TuiChannel` to feed the mention picker's Skills tab. +- `zeph-subagent`, `zeph-config`, `zeph-core`: configurable, default-on session-wide + cumulative cap on total sub-agent spawns (issue #6545), independent of the existing + `max_concurrent` (in-flight) and `max_spawn_depth` (recursion) guardrails — catches a + shallow, low-concurrency but high-frequency sequential delegation loop that neither of + those bounds. New `[agents] max_spawns_per_session` config field (default `100`, `0` = + unlimited), new `SubAgentError::SessionSpawnLimit` variant, and a new + `zeph_subagent::SessionSpawnBudget` counter type (an uncloneable `AtomicUsize` newtype, + reached via an `Agent::session_budget()` accessor rather than a copied handle) so + `SubAgentManager::spawn`/`resume` and the ACP `/subagent spawn` path (which never touches + `SubAgentManager`) enforce the same cumulative counter. The cap is checked ahead of + `max_spawn_depth`/`max_concurrent` but + consumed only at each spawn path's true commit point, so a spawn rejected for any other + reason — including a transient `ConcurrencyLimit` the orchestration scheduler retries — + never burns budget it never used. `--migrate-config` populates the new key on existing + configs; `/agent status` and `/agent list` surface the running count. ### Changed diff --git a/book/src/reference/configuration.md b/book/src/reference/configuration.md index 14ae6dcf0..2b20b0591 100644 --- a/book/src/reference/configuration.md +++ b/book/src/reference/configuration.md @@ -766,6 +766,7 @@ enabled = false # Enable native worktree isolation for backgr [agents] enabled = false # Enable sub-agent system (default: false) max_concurrent = 1 # Max concurrent sub-agents (default: 1) +max_spawns_per_session = 100 # Cumulative spawn cap for the whole session (default: 100, 0 = unlimited) delegation_mode = "proactive" # Autonomous spawn control: "proactive" (default), "explicit_request_only", or "disabled" extra_dirs = [] # Additional directories to scan for agent definitions # default_memory_scope = "project" # Default memory scope for agents without explicit `memory` field diff --git a/config/default.toml b/config/default.toml index a92afc69a..0d2eff920 100644 --- a/config/default.toml +++ b/config/default.toml @@ -1194,6 +1194,11 @@ enabled = false delegation_mode = "proactive" # Maximum number of sub-agents that can run concurrently max_concurrent = 1 +# Maximum cumulative number of sub-agents that may be spawned within a single session, +# independent of max_concurrent (in-flight limit) and max_spawn_depth (recursion limit) — +# guards against a shallow, low-concurrency but high-frequency sequential delegation loop +# (issue #6545). 0 = unlimited. +max_spawns_per_session = 100 # Allow sub-agents to use bypass_permissions mode (enable only in trusted environments) allow_bypass_permissions = false # Enable writing JSONL transcripts for sub-agent sessions (required for /agent resume) diff --git a/crates/zeph-config/src/agent.rs b/crates/zeph-config/src/agent.rs index 98b7dca29..a5d8fb83b 100644 --- a/crates/zeph-config/src/agent.rs +++ b/crates/zeph-config/src/agent.rs @@ -226,6 +226,10 @@ fn default_max_concurrent() -> usize { 5 } +fn default_max_spawns_per_session() -> usize { + 100 +} + fn default_context_window_turns() -> usize { 10 } @@ -600,6 +604,7 @@ impl Default for TaskSupervisorConfig { /// delegation_mode = "explicit_request_only" /// max_concurrent = 3 /// max_spawn_depth = 2 +/// max_spawns_per_session = 50 /// ``` #[derive(Debug, Clone, Deserialize, Serialize)] #[serde(default)] @@ -620,6 +625,15 @@ pub struct SubAgentConfig { /// Maximum number of sub-agents that can run concurrently. #[serde(default = "default_max_concurrent")] pub max_concurrent: usize, + /// Maximum cumulative number of sub-agents that may be spawned within a single session, + /// independent of [`max_concurrent`][Self::max_concurrent] (in-flight limit) and + /// [`max_spawn_depth`][Self::max_spawn_depth] (recursion limit). Guards against a shallow, + /// low-concurrency but high-frequency sequential delegation loop that neither of those + /// limits would catch (issue #6545). `0` = unlimited. The counter resets at session start + /// and is shared across every spawn chokepoint (`SubAgentManager::spawn`/`resume` and the + /// ACP `/subagent spawn` path). Default: `100`. + #[serde(default = "default_max_spawns_per_session")] + pub max_spawns_per_session: usize, /// Additional directories to search for `.agent.md` definition files. pub extra_dirs: Vec, /// User-level agents directory. @@ -718,6 +732,7 @@ impl Default for SubAgentConfig { enabled: false, delegation_mode: DelegationMode::default(), max_concurrent: default_max_concurrent(), + max_spawns_per_session: default_max_spawns_per_session(), extra_dirs: Vec::new(), user_agents_dir: None, default_permission_mode: None, @@ -808,6 +823,31 @@ mod tests { ); } + #[test] + fn subagent_config_max_spawns_per_session_default_direct() { + // Direct-Default assertion (distinct from the serde round-trip below): catches a + // forgotten `impl Default` entry, which the per-field serde attribute would otherwise + // mask on every serde-loaded path while every non-serde caller silently got `0`. + assert_eq!(SubAgentConfig::default().max_spawns_per_session, 100); + } + + #[test] + fn subagent_config_max_spawns_per_session_omitted_key_defaults_100() { + // Serde round-trip: a present `[agents]` section with the key absent must not + // silently resolve to `0` (unlimited) — that would recreate the exact "safety net + // ships disabled" defect class #6469/PR #6528 fixed for the tool-call/cost guardrails. + let toml_str = "enabled = true\nmax_concurrent = 3"; + let cfg: SubAgentConfig = toml::from_str(toml_str).unwrap(); + assert_eq!(cfg.max_spawns_per_session, 100); + } + + #[test] + fn subagent_config_deserialize_max_spawns_per_session() { + let toml_str = "max_spawns_per_session = 50"; + let cfg: SubAgentConfig = toml::from_str(toml_str).unwrap(); + assert_eq!(cfg.max_spawns_per_session, 50); + } + #[test] fn subagent_config_delegation_mode_defaults_proactive() { let cfg = SubAgentConfig::default(); diff --git a/crates/zeph-config/src/migrate/mod.rs b/crates/zeph-config/src/migrate/mod.rs index b276fad96..2a38fc04e 100644 --- a/crates/zeph-config/src/migrate/mod.rs +++ b/crates/zeph-config/src/migrate/mod.rs @@ -44,7 +44,7 @@ pub use memory::*; pub use plugins::migrate_plugins_reputation_config; pub use serve::migrate_serve_config; pub use session::*; -pub use subagent::migrate_agents_delegation_mode; +pub use subagent::{migrate_agents_delegation_mode, migrate_agents_max_spawns_per_session}; pub use tools::*; /// Returns `true` when `name` is an active (non-commented) TOML section header in `src`. @@ -608,19 +608,19 @@ mod steps; use steps::{ MigrateA2aCardTrustConfig, MigrateA2aServerRemoveInertFields, MigrateAcpAuthClientsConfig, MigrateAcpSubagentsConfig, MigrateAgentBudgetHint, MigrateAgentRetryToToolsRetry, - MigrateAgentTimeReminder, MigrateAgentsDelegationMode, MigrateAutodreamConfig, - MigrateCavemanConfig, MigrateCocoonProviderNotice, MigrateCocoonShowBalance, - MigrateCompressionPredictorConfig, MigrateDatabaseUrl, MigrateDeepLinkConfig, - MigrateDurableConfig, MigrateDurableHwmAdvisory, MigrateDurableKeyRotation, - MigrateDurableSharedDb, MigrateDurableStaleRunningAfterSecs, MigrateEgressConfig, - MigrateEmbedProviderRename, MigrateEvalModelToProvider, MigrateFidelityTimeoutDefaults, - MigrateFiveSignalConfig, MigrateFocusAutoConsolidateMinWindow, MigrateForgettingConfig, - MigrateGoalsConfig, MigrateGonkagateToGonka, MigrateHooksPermissionDeniedConfig, - MigrateHooksTurnComplete, MigrateIntegrityConfig, MigrateKnowledgeConfig, - MigrateLlmStreamLimits, MigrateMagicDocsConfig, MigrateMcpElicitationConfig, - MigrateMcpMaxConnectAttempts, MigrateMcpMediaConfig, MigrateMcpRetryAndToolTimeout, - MigrateMcpTrustLevels, MigrateMemoryConsentGateConfig, MigrateMemoryGraph, - MigrateMemoryGraphRecallIncludeImported, MigrateMemoryHebbian, + MigrateAgentTimeReminder, MigrateAgentsDelegationMode, MigrateAgentsMaxSpawnsPerSession, + MigrateAutodreamConfig, MigrateCavemanConfig, MigrateCocoonProviderNotice, + MigrateCocoonShowBalance, MigrateCompressionPredictorConfig, MigrateDatabaseUrl, + MigrateDeepLinkConfig, MigrateDurableConfig, MigrateDurableHwmAdvisory, + MigrateDurableKeyRotation, MigrateDurableSharedDb, MigrateDurableStaleRunningAfterSecs, + MigrateEgressConfig, MigrateEmbedProviderRename, MigrateEvalModelToProvider, + MigrateFidelityTimeoutDefaults, MigrateFiveSignalConfig, MigrateFocusAutoConsolidateMinWindow, + MigrateForgettingConfig, MigrateGoalsConfig, MigrateGonkagateToGonka, + MigrateHooksPermissionDeniedConfig, MigrateHooksTurnComplete, MigrateIntegrityConfig, + MigrateKnowledgeConfig, MigrateLlmStreamLimits, MigrateMagicDocsConfig, + MigrateMcpElicitationConfig, MigrateMcpMaxConnectAttempts, MigrateMcpMediaConfig, + MigrateMcpRetryAndToolTimeout, MigrateMcpTrustLevels, MigrateMemoryConsentGateConfig, + MigrateMemoryGraph, MigrateMemoryGraphRecallIncludeImported, MigrateMemoryHebbian, MigrateMemoryHebbianConsolidation, MigrateMemoryHebbianSpread, MigrateMemoryPersonaConfig, MigrateMemoryReasoning, MigrateMemoryReasoningJudge, MigrateMemoryRetrieval, MigrateMemoryRetrievalQueryBias, MigrateMemoryStoreConfig, MigrateMemoryTypeAwareCompose, @@ -860,6 +860,9 @@ pub static MIGRATIONS: std::sync::LazyLock> // Step 104 — add expandable_blockquote_min_lines advisory comment to an // existing active [telegram] table (spec 007-3-telegram-rich-text, #6541) Box::new(MigrateTelegramExpandableBlockquoteConfig), + // Step 105 — insert active max_spawns_per_session = 100 into an existing + // [agents] table with enabled = true and no max_spawns_per_session key (#6545) + Box::new(MigrateAgentsMaxSpawnsPerSession), ] }); diff --git a/crates/zeph-config/src/migrate/steps.rs b/crates/zeph-config/src/migrate/steps.rs index 19224056d..3244e28f6 100644 --- a/crates/zeph-config/src/migrate/steps.rs +++ b/crates/zeph-config/src/migrate/steps.rs @@ -89,8 +89,9 @@ use super::{ MigrateError, Migration, MigrationResult, migrate_a2a_card_trust_config, migrate_a2a_server_remove_inert_fields, migrate_acp_auth_clients_config, migrate_acp_subagents_config, migrate_agent_budget_hint, migrate_agent_retry_to_tools_retry, - migrate_agent_time_reminder, migrate_agents_delegation_mode, migrate_autodream_config, - migrate_caveman_config, migrate_cocoon_provider_notice, migrate_cocoon_show_balance, + migrate_agent_time_reminder, migrate_agents_delegation_mode, + migrate_agents_max_spawns_per_session, migrate_autodream_config, migrate_caveman_config, + migrate_cocoon_provider_notice, migrate_cocoon_show_balance, migrate_compression_predictor_config, migrate_database_url, migrate_deep_link_config, migrate_durable_config, migrate_durable_hwm_advisory, migrate_durable_key_rotation, migrate_durable_shared_db, migrate_durable_stale_running_after_secs, migrate_egress_config, @@ -1341,3 +1342,16 @@ impl Migration for MigrateTelegramExpandableBlockquoteConfig { migrate_telegram_expandable_blockquote_config(toml_src) } } + +/// Step 105 — insert an active `max_spawns_per_session = 100` value into an existing +/// `[agents]` table with `enabled = true` and no `max_spawns_per_session` key (issue #6545). +pub(super) struct MigrateAgentsMaxSpawnsPerSession; +impl Migration for MigrateAgentsMaxSpawnsPerSession { + fn name(&self) -> &'static str { + "migrate_agents_max_spawns_per_session" + } + + fn apply(&self, toml_src: &str) -> Result { + migrate_agents_max_spawns_per_session(toml_src) + } +} diff --git a/crates/zeph-config/src/migrate/subagent.rs b/crates/zeph-config/src/migrate/subagent.rs index 9b5282b1a..48b05beb9 100644 --- a/crates/zeph-config/src/migrate/subagent.rs +++ b/crates/zeph-config/src/migrate/subagent.rs @@ -70,6 +70,67 @@ pub fn migrate_agents_delegation_mode(toml_src: &str) -> Result Result { + if toml_src.contains("max_spawns_per_session") || !section_header_present(toml_src, "agents") { + return Ok(MigrationResult { + output: toml_src.to_owned(), + changed_count: 0, + sections_changed: Vec::new(), + }); + } + + let mut doc = toml_src.parse::()?; + let Some(agents_table) = doc + .get_mut("agents") + .and_then(toml_edit::Item::as_table_mut) + else { + return Ok(MigrationResult { + output: toml_src.to_owned(), + changed_count: 0, + sections_changed: Vec::new(), + }); + }; + + let enabled = agents_table + .get("enabled") + .and_then(toml_edit::Item::as_value) + .and_then(toml_edit::Value::as_bool) + .unwrap_or(false); + + if !enabled { + return Ok(MigrationResult { + output: toml_src.to_owned(), + changed_count: 0, + sections_changed: Vec::new(), + }); + } + + agents_table.insert("max_spawns_per_session", toml_edit::value(100_i64)); + + Ok(MigrationResult { + output: doc.to_string(), + changed_count: 1, + sections_changed: vec!["agents.max_spawns_per_session".to_owned()], + }) +} + #[cfg(test)] mod tests { use super::*; @@ -122,4 +183,53 @@ mod tests { assert_eq!(twice.changed_count, 0); assert_eq!(twice.output, once.output); } + + #[test] + fn max_spawns_inserts_100_when_enabled_and_key_absent() { + let src = "[agents]\nenabled = true\nmax_concurrent = 3\n"; + let result = migrate_agents_max_spawns_per_session(src).expect("migrate"); + assert_eq!(result.changed_count, 1); + assert!(result.output.contains("max_spawns_per_session = 100")); + } + + #[test] + fn max_spawns_noop_when_disabled() { + let src = "[agents]\nenabled = false\n"; + let result = migrate_agents_max_spawns_per_session(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert!(!result.output.contains("max_spawns_per_session")); + } + + #[test] + fn max_spawns_noop_when_key_already_present() { + let src = "[agents]\nenabled = true\nmax_spawns_per_session = 50\n"; + let result = migrate_agents_max_spawns_per_session(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert_eq!(result.output, src); + } + + #[test] + fn max_spawns_noop_when_section_absent() { + let src = "[llm]\nprovider = \"claude\"\n"; + let result = migrate_agents_max_spawns_per_session(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert_eq!(result.output, src); + } + + #[test] + fn max_spawns_noop_when_section_only_commented_out() { + let src = "# [agents]\n# enabled = true\n"; + let result = migrate_agents_max_spawns_per_session(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert_eq!(result.output, src); + } + + #[test] + fn max_spawns_idempotent() { + let src = "[agents]\nenabled = true\n"; + let once = migrate_agents_max_spawns_per_session(src).expect("migrate"); + let twice = migrate_agents_max_spawns_per_session(&once.output).expect("migrate"); + assert_eq!(twice.changed_count, 0); + assert_eq!(twice.output, once.output); + } } diff --git a/crates/zeph-config/src/migrate/tests.rs b/crates/zeph-config/src/migrate/tests.rs index a8ec960ba..5068aa012 100644 --- a/crates/zeph-config/src/migrate/tests.rs +++ b/crates/zeph-config/src/migrate/tests.rs @@ -9,8 +9,8 @@ use super::*; fn migrations_registry_has_all_steps() { assert_eq!( MIGRATIONS.len(), - 104, - "MIGRATIONS registry must contain all 104 sequential steps" + 105, + "MIGRATIONS registry must contain all 105 sequential steps" ); for m in MIGRATIONS.iter() { assert!( @@ -2124,7 +2124,42 @@ fn migrate_focus_auto_consolidate_noop_when_only_commented_section() { #[test] fn registry_has_fifty_entries() { - assert_eq!(MIGRATIONS.len(), 104); + assert_eq!(MIGRATIONS.len(), 105); +} + +/// SC-003 (issue #6545): the isolated `migrate_agents_max_spawns_per_session` tests in +/// `migrate/subagent.rs` prove the function itself works, but not that it is actually wired +/// into `MIGRATIONS` — a function that exists but is never pushed into the registry (the +/// repo's recurring wire-X defect class) would pass those tests while silently doing nothing +/// for a real `--migrate-config` run. +/// +/// Deliberately does **not** use `ConfigMigrator::migrate` — that type is the separate +/// catch-all reference-merge pass (`migrate/mod.rs`, diffs against the embedded +/// `config/default.toml` and appends missing keys as *commented-out* advisory lines via +/// `merge_table_commented`); it never reads `MIGRATIONS` at all. Using it here would pass even +/// if `MigrateAgentsMaxSpawnsPerSession` were never pushed into the registry — the catch-all +/// pass would still add `# max_spawns_per_session = 100` as a comment, and a naive +/// `str::contains` assertion cannot tell an active key from a commented one. Instead this +/// mirrors the real `--migrate-config` flow's first phase (`src/commands/migrate.rs`, +/// `handle_migrate_config`): fold `MIGRATIONS.iter()` over the input, then parse the result and +/// assert the key is present as an *active*, typed value — a parsed-document assertion cannot +/// match a comment, unlike a substring check. +#[test] +fn full_registry_adds_max_spawns_per_session_to_legacy_agents_config() { + let legacy = "[agents]\nenabled = true\nmax_concurrent = 3\n"; + let mut current = legacy.to_owned(); + for m in MIGRATIONS.iter() { + current = m + .apply(¤t) + .expect("registry migration must not fail") + .output; + } + let doc: toml_edit::DocumentMut = current.parse().expect("migrated output must be valid TOML"); + assert_eq!( + doc["agents"]["max_spawns_per_session"].as_integer(), + Some(100), + "MIGRATIONS must add an active (non-commented) key, got: {current}" + ); } #[test] @@ -2270,6 +2305,7 @@ fn registry_preserves_order_matches_dispatch() { "migrate_agents_delegation_mode", "migrate_memory_consent_gate_config", "migrate_telegram_expandable_blockquote_config", + "migrate_agents_max_spawns_per_session", ]; let actual: Vec<&str> = MIGRATIONS.iter().map(|m| m.name()).collect(); assert_eq!(actual, expected); diff --git a/crates/zeph-core/config/default.toml b/crates/zeph-core/config/default.toml index 6838f98a2..0f97b1056 100644 --- a/crates/zeph-core/config/default.toml +++ b/crates/zeph-core/config/default.toml @@ -710,6 +710,9 @@ enabled = false delegation_mode = "proactive" # Maximum number of sub-agents that can run concurrently max_concurrent = 1 +# Maximum cumulative number of sub-agents that may be spawned within a single session, +# independent of max_concurrent and max_spawn_depth (issue #6545). 0 = unlimited. +max_spawns_per_session = 100 # Allow sub-agents to use bypass_permissions mode (enable only in trusted environments) allow_bypass_permissions = false # Enable writing JSONL transcripts for sub-agent sessions (required for /agent resume) diff --git a/crates/zeph-core/src/agent/slash_commands.rs b/crates/zeph-core/src/agent/slash_commands.rs index 6be208e23..a7541edd3 100644 --- a/crates/zeph-core/src/agent/slash_commands.rs +++ b/crates/zeph-core/src/agent/slash_commands.rs @@ -117,6 +117,22 @@ impl Agent { /// `#[non_exhaustive]` variant. `/subagent spawn` is itself an explicit user action, so it /// stays permitted under `explicit_request_only` and `proactive`, blocked only when /// `permits_explicit()` is `false` (currently just `disabled`). + /// + /// A second, independent policy is enforced at this same choke point: the session-wide + /// cumulative spawn budget (issue #6545, `Agent::session_budget`). Since this path spawns + /// an out-of-process ACP session and never touches `SubAgentManager`, it would otherwise + /// bypass the cap entirely. `AcpSubagentSpawnFn` is fallible — but unlike the manager path + /// (where `NotFound`/`ConcurrencyLimit` genuinely mean "no resources allocated"), the real + /// callback wired in `src/runner.rs` (`zeph_acp::run_session`) launches the child process + /// *before* it can fail: its `Err` arm covers three cases — the launch itself failing + /// (nothing ran), a live child that ran for up to `session_timeout_secs` and then timed + /// out, and a live child that ran and then failed post-launch I/O/protocol. Only the first + /// allocates zero resources; the other two already spent an OS process. Because + /// `Result` erases which case occurred by the time it crosses this + /// boundary, the budget is consumed unconditionally once `spawn_fn` returns at all (both + /// `Ok` and `Err`) — accepting a small overcount on launch-failure in exchange for never + /// undercounting a real, possibly-hung child process, which is the actual runaway-loop + /// threat this cap exists to bound. async fn handle_subagent_slash(&mut self, args: &str) -> Result<(), error::AgentError> { let msg: String = if args.is_empty() { "Usage: /subagent \n\nSubcommands:\n spawn Spawn an ACP sub-agent process".to_owned() @@ -126,6 +142,11 @@ impl Agent { "spawn" => { let cmd = rest.trim(); let effective_mode = self.effective_delegation_mode(); + let max_spawns = self + .services + .orchestration + .subagent_config + .max_spawns_per_session; if cmd.is_empty() { "Usage: /subagent spawn \n\nExample: /subagent spawn zeph --acp" .to_owned() @@ -137,10 +158,24 @@ impl Agent { "Sub-agent delegation is disabled by configuration \ ([agents].delegation_mode = \"disabled\" or [agents].enabled = false)." .to_owned() + } else if let Err(e) = self.session_budget().check(max_spawns) { + tracing::warn!( + error = %e, + "/subagent spawn rejected: session spawn budget exhausted" + ); + format!("Sub-agent error: {e}") } else if let Some(spawn_fn) = self.runtime.config.acp_subagent_spawn_fn.clone() { let cmd = cmd.to_owned(); - match spawn_fn(cmd).await { + let result = spawn_fn(cmd).await; + // Commit point (issue #6545): record unconditionally once `spawn_fn` + // has returned at all, in both the `Ok` and `Err` arms — see the doc + // comment above for why an `Err` here cannot be assumed to mean "never + // launched". Re-borrows `self` fresh (rather than holding a reference + // across the `.await` above) since `SessionSpawnBudget` has no + // `Clone`/`Arc` by design — see its own doc comment. + self.session_budget().record_spawn(); + match result { Ok(output) => output, Err(e) => format!("Sub-agent error: {e}"), } diff --git a/crates/zeph-core/src/agent/state/mod.rs b/crates/zeph-core/src/agent/state/mod.rs index 393eefa73..53159b439 100644 --- a/crates/zeph-core/src/agent/state/mod.rs +++ b/crates/zeph-core/src/agent/state/mod.rs @@ -741,6 +741,17 @@ pub(crate) struct OrchestrationState { /// Manages spawned sub-agents. pub(crate) subagent_manager: Option, pub(crate) subagent_config: crate::config::SubAgentConfig, + /// Fail-closed fallback session-wide subagent-spawn budget (issue #6545). + /// + /// Used only when no `subagent_manager` is wired (serve/daemon/acp bootstrap paths, or a + /// bare test harness) — see `Agent::session_budget` in `subagent_commands.rs`, which + /// prefers the manager's own budget (the origin/source of truth) whenever one exists and + /// falls back to this field otherwise, mirroring `effective_delegation_mode`'s existing + /// fallback-to-config precedent. Never copied into or out of `subagent_manager` — kept + /// deliberately independent so a future direct `subagent_manager = Some(...)` assignment + /// (several already exist in this crate's test helpers) can never desynchronize two + /// budgets that were supposed to be the same one. + pub(crate) session_spawn_budget: zeph_subagent::SessionSpawnBudget, pub(crate) orchestration_config: crate::config::OrchestrationConfig, /// Lazily initialized plan template cache. `None` until first use or when /// memory (`SQLite`) is unavailable. diff --git a/crates/zeph-core/src/agent/subagent_commands.rs b/crates/zeph-core/src/agent/subagent_commands.rs index 9d4e0e0b4..b199fd56c 100644 --- a/crates/zeph-core/src/agent/subagent_commands.rs +++ b/crates/zeph-core/src/agent/subagent_commands.rs @@ -264,6 +264,7 @@ impl Agent { fn handle_agent_list(&self) -> Option { use std::fmt::Write as _; let mgr = self.services.orchestration.subagent_manager.as_ref()?; + let spawns_line = self.format_session_spawns_line(); let mode_label = match mgr.delegation_mode() { zeph_config::DelegationMode::Disabled => "disabled", zeph_config::DelegationMode::ExplicitRequestOnly => "explicit_request_only", @@ -273,10 +274,11 @@ impl Agent { let defs = mgr.definitions(); if defs.is_empty() { return Some(format!( - "Delegation mode: {mode_label}\nNo sub-agent definitions found." + "{spawns_line}\nDelegation mode: {mode_label}\nNo sub-agent definitions found." )); } - let mut out = format!("Delegation mode: {mode_label}\nAvailable sub-agents:\n"); + let mut out = + format!("{spawns_line}\nDelegation mode: {mode_label}\nAvailable sub-agents:\n"); for d in defs { let memory_label = match d.memory { Some(zeph_subagent::MemoryScope::User) => " [memory:user]", @@ -300,11 +302,12 @@ impl Agent { fn handle_agent_status(&self) -> Option { use std::fmt::Write as _; let mgr = self.services.orchestration.subagent_manager.as_ref()?; + let spawns_line = self.format_session_spawns_line(); let statuses = mgr.statuses(); if statuses.is_empty() { - return Some("No active sub-agents.".into()); + return Some(format!("{spawns_line}\nNo active sub-agents.")); } - let mut out = String::from("Active sub-agents:\n"); + let mut out = format!("{spawns_line}\nActive sub-agents:\n"); for (id, s) in &statuses { let state = format!("{:?}", s.state).to_lowercase(); let elapsed = s.started_at.elapsed().as_secs(); @@ -828,6 +831,55 @@ impl Agent { .subagent_config .effective_delegation_mode() } + + /// The session-wide cumulative subagent-spawn budget in force for this session (issue + /// #6545). + /// + /// Returns the `SubAgentManager`'s own budget when a manager is wired (the common case: + /// CLI/TUI runner), so a manager-side spawn and the ACP `/subagent spawn` chokepoint in + /// `slash_commands.rs` observe and contribute to the exact same cumulative count. Falls + /// back to `OrchestrationState::session_spawn_budget` when no manager is wired + /// (serve/daemon/acp bootstrap paths, or a bare test harness) — fail-closed rather than + /// unenforced, mirroring [`effective_delegation_mode`][Self::effective_delegation_mode]'s + /// fallback-to-config precedent above. An accessor rather than a copied handle, so a + /// future direct `subagent_manager = Some(...)` assignment elsewhere can never + /// desynchronize two independent budgets. + pub(super) fn session_budget(&self) -> &zeph_subagent::SessionSpawnBudget { + self.services + .orchestration + .subagent_manager + .as_ref() + .map_or( + &self.services.orchestration.session_spawn_budget, + zeph_subagent::SubAgentManager::session_budget, + ) + } + + /// Format the `Session spawns: N/max` (or `N/unlimited`) line shared by + /// `handle_agent_status` and `handle_agent_list`. + /// + /// Must be called before either function's early "no active agents"/"no definitions" + /// return, not just the non-empty branch — that early return is precisely the state right + /// after the cap fires under the shipped `max_concurrent = 1` default, which is exactly + /// when an operator needs to see the count (issue #6545). Reads through + /// [`session_budget`][Self::session_budget] rather than a manager parameter's own + /// `session_budget()`, so this stays the only path that resolves which budget instance + /// applies — both callers happen to have a manager in hand already, but routing through + /// the accessor avoids a second, parallel resolution path to the same value. + fn format_session_spawns_line(&self) -> String { + let max = self + .services + .orchestration + .subagent_config + .max_spawns_per_session; + let spawned = self.session_budget().spawned(); + if max == 0 { + format!("Session spawns: {spawned}/unlimited") + } else { + format!("Session spawns: {spawned}/{max}") + } + } + /// Build a `SpawnContext` from current agent state for sub-agent spawning. pub(super) fn build_spawn_context( &self, diff --git a/crates/zeph-core/src/agent/tests/agent_tests/subagent_command_tests.rs b/crates/zeph-core/src/agent/tests/agent_tests/subagent_command_tests.rs index fd86a29cf..23d4a3f94 100644 --- a/crates/zeph-core/src/agent/tests/agent_tests/subagent_command_tests.rs +++ b/crates/zeph-core/src/agent/tests/agent_tests/subagent_command_tests.rs @@ -100,6 +100,39 @@ async fn agent_command_status_no_agents_returns_empty_message() { .await .unwrap(); assert!(resp.contains("No active sub-agents")); + // Issue #6545, S4 regression: the "Session spawns: N/max" line must survive the early + // return — this is exactly the state right after the cap fires, when an operator most + // needs to see the count. Default config's max_spawns_per_session is 100. + assert!( + resp.starts_with("Session spawns: 0/100"), + "expected the spawns line as the first line, got: {resp}" + ); +} + +/// Issue #6545, S4 regression: `handle_agent_list`'s *other* early return (no definitions +/// loaded at all, distinct from the "no active agents" branch above) must also carry the +/// spawns line — `make_agent_with_manager()` always pushes a "helper" definition, so this +/// test builds its own manager with none. +#[tokio::test] +async fn agent_command_list_no_definitions_still_shows_spawns_line() { + use zeph_subagent::SubAgentManager; + + let provider = mock_provider(vec![]); + let channel = MockChannel::new(vec![]); + let registry = create_test_registry(); + let executor = MockToolExecutor::no_tools(); + let mut agent = Agent::new(provider, channel, registry, None, 5, executor); + agent.services.orchestration.subagent_manager = Some(SubAgentManager::new(4)); + + let resp = agent + .handle_agent_command(AgentCommand::List) + .await + .unwrap(); + assert!(resp.contains("No sub-agent definitions found.")); + assert!( + resp.starts_with("Session spawns: 0/100"), + "expected the spawns line as the first line, got: {resp}" + ); } #[tokio::test] diff --git a/crates/zeph-core/src/agent/tests/small_misc_tests.rs b/crates/zeph-core/src/agent/tests/small_misc_tests.rs index b8319bb55..3c5c40ecf 100644 --- a/crates/zeph-core/src/agent/tests/small_misc_tests.rs +++ b/crates/zeph-core/src/agent/tests/small_misc_tests.rs @@ -170,6 +170,130 @@ async fn subagent_spawn_explicit_request_only_still_allows_acp_spawn() { ); } +/// Issue #6545, N1 (revised — see security audit `2026-07-28T00-38-48-security.md`): +/// `AcpSubagentSpawnFn` is fallible, but its `Err` arm cannot be assumed to mean "never +/// launched" — the real callback (`zeph_acp::run_session`) launches the child process before +/// it can time out or fail post-launch, and `Result` erases that distinction +/// by the time it crosses this boundary. So a *simulated post-launch failure* (as opposed to +/// the callback simply not existing, or the pre-flight budget `check()` rejecting the attempt +/// before `spawn_fn` is ever called) must still consume budget: undercounting a real process +/// launch is the actual vulnerability this cap exists to prevent, not the minor overcount of +/// the rarer "never launched" sub-case. Regression guard against reverting to the earlier, +/// incorrect `Ok`-arm-only commit point. +#[tokio::test] +async fn subagent_spawn_post_launch_failure_still_consumes_budget() { + let mut h = QuickTestAgent::minimal(""); + h.agent.services.orchestration.subagent_config.enabled = true; + h.agent.runtime.config.acp_subagent_spawn_fn = Some(std::sync::Arc::new(|_cmd: String| { + Box::pin(async move { Err("launch failed".to_owned()) }) + })); + + let result = h + .agent + .dispatch_slash_command("/subagent spawn my-command") + .await; + assert!(result.is_some(), "must be intercepted"); + let output = h.sent_messages().join("\n"); + assert!( + output.contains("launch failed"), + "expected the spawn_fn error, got: {output}" + ); + assert_eq!( + h.agent + .services + .orchestration + .session_spawn_budget + .spawned(), + 1, + "spawn_fn having returned at all — even Err — means it was invoked and must consume \ + budget, since its Err arm can mean a real child process ran and then failed" + ); +} + +/// Issue #6545, N3(b): with no `SubAgentManager` wired at all (the `QuickTestAgent::minimal` +/// harness default — matches serve/daemon/acp bootstrap paths per critic finding N2/R4), the +/// ACP chokepoint must still enforce the cap via `OrchestrationState`'s own fallback budget. +#[tokio::test] +async fn subagent_spawn_cap_reached_enforced_without_manager() { + let mut h = QuickTestAgent::minimal(""); + h.agent.services.orchestration.subagent_config.enabled = true; + h.agent + .services + .orchestration + .subagent_config + .max_spawns_per_session = 1; + h.agent.runtime.config.acp_subagent_spawn_fn = Some(std::sync::Arc::new(|cmd: String| { + Box::pin(async move { Ok(format!("spawned: {cmd}")) }) + })); + assert!( + h.agent.services.orchestration.subagent_manager.is_none(), + "precondition: this test exercises the no-manager fallback path" + ); + + let first = h + .agent + .dispatch_slash_command("/subagent spawn first-command") + .await; + assert!(first.is_some()); + assert!( + h.sent_messages() + .join("\n") + .contains("spawned: first-command") + ); + + let second = h + .agent + .dispatch_slash_command("/subagent spawn second-command") + .await; + assert!(second.is_some()); + let output = h.sent_messages().join("\n"); + assert!( + !output.contains("spawned: second-command"), + "cap must reject the second spawn, got: {output}" + ); + assert!( + output.contains("session spawn limit"), + "rejection must name the session spawn limit, got: {output}" + ); +} + +/// Issue #6545, N3(a): with a `SubAgentManager` wired (mirrors `runner.rs`'s production +/// `with_orchestration` path), an ACP spawn through the chokepoint must contribute to the +/// exact same cumulative count the manager itself sees — proving `Agent::session_budget`'s +/// accessor resolves to the manager's own budget, not a disconnected fallback, whenever a +/// manager exists. (The reverse — a manager-side spawn being observable through the ACP +/// chokepoint — is trivially true through the shared `&SessionSpawnBudget` reference and isn't +/// separately asserted here.) +#[tokio::test] +async fn subagent_spawn_visible_to_manager_budget_when_wired() { + let mut h = QuickTestAgent::minimal(""); + h.agent.services.orchestration.subagent_config.enabled = true; + h.agent.services.orchestration.subagent_manager = Some(zeph_subagent::SubAgentManager::new(4)); + h.agent.runtime.config.acp_subagent_spawn_fn = Some(std::sync::Arc::new(|cmd: String| { + Box::pin(async move { Ok(format!("spawned: {cmd}")) }) + })); + + let result = h + .agent + .dispatch_slash_command("/subagent spawn my-command") + .await; + assert!(result.is_some()); + assert!(h.sent_messages().join("\n").contains("spawned: my-command")); + + let mgr = h + .agent + .services + .orchestration + .subagent_manager + .as_ref() + .expect("manager must still be wired"); + assert_eq!( + mgr.session_budget().spawned(), + 1, + "the ACP spawn must be visible through the manager's own budget handle" + ); +} + #[tokio::test] async fn subagent_unknown_subcommand_returns_error() { let mut h = QuickTestAgent::minimal(""); diff --git a/crates/zeph-orchestration/src/scheduler/tick/tests.rs b/crates/zeph-orchestration/src/scheduler/tick/tests.rs index 9e66c9a30..e5fe0fd69 100644 --- a/crates/zeph-orchestration/src/scheduler/tick/tests.rs +++ b/crates/zeph-orchestration/src/scheduler/tick/tests.rs @@ -1383,6 +1383,98 @@ fn test_record_spawn_failure() { ); } +/// Issue #6545 (S2): unlike `ConcurrencyLimit`, `SessionSpawnLimit` is not special-cased as +/// transient inside `record_spawn_failure` — it falls through to `TaskStatus::Failed` + +/// `dag::propagate_failure`, exactly like any other `SubAgentError` variant. +/// `record_spawn_failure` performs no error-type inspection beyond the `ConcurrencyLimit` +/// check, so what happens next is entirely a function of the node's `FailureStrategy`. This +/// test pins that behavior under the shipped **default** (`FailureStrategy::Abort`): the +/// failure is terminal for the graph. It does **not** mean `SessionSpawnLimit` can never be +/// retried in general — under the opt-in `FailureStrategy::Retry`, `propagate_failure`'s own +/// `Retry` arm resurrects *any* `Failed` node (including one classified from +/// `SessionSpawnLimit`) back to `Ready`, bounded by `max_retries` — see +/// `test_record_spawn_failure_session_spawn_limit_bounded_retry_under_retry_strategy` below. +#[test] +fn test_record_spawn_failure_session_spawn_limit_marks_failed_not_ready() { + let graph = graph_from_nodes(vec![make_node(0, &[])]); + let mut scheduler = make_scheduler(graph); + + scheduler.graph.tasks[0].status = TaskStatus::Running; + + let error = SubAgentError::SessionSpawnLimit { + spawned: 100, + max: 100, + }; + let actions = scheduler.record_spawn_failure(TaskId(0), &error); + assert_eq!( + scheduler.graph.tasks[0].status, + TaskStatus::Failed, + "under the default FailureStrategy::Abort, SessionSpawnLimit is terminal" + ); + assert_eq!(scheduler.graph.status, GraphStatus::Failed); + assert!( + actions + .iter() + .any(|a| matches!(a, SchedulerAction::Done { .. })) + ); +} + +/// Issue #6545 (I6): pins the `FailureStrategy::Retry` case the test above explicitly does not +/// cover. `record_spawn_failure` classifies `SessionSpawnLimit` identically to every +/// non-`ConcurrencyLimit` error — it has no special "unretryable" marking — so under a node's +/// opt-in `Retry` strategy, `dag::propagate_failure` resurrects it back to `Ready` like any +/// other failure, bounded by `max_retries` (not unbounded — the guard itself is check-only and +/// never consumes budget, so these bounded retries cost nothing either way, but the node does +/// leave `Failed` and re-enter `Ready` rather than staying terminal). +#[test] +fn test_record_spawn_failure_session_spawn_limit_bounded_retry_under_retry_strategy() { + use crate::graph::FailureStrategy; + + let graph = graph_from_nodes(vec![make_node(0, &[])]); + let mut scheduler = make_scheduler(graph); + + scheduler.graph.tasks[0].failure_strategy = Some(FailureStrategy::Retry); + scheduler.graph.tasks[0].max_retries = Some(2); + scheduler.graph.tasks[0].retry_count = 0; + + let error = SubAgentError::SessionSpawnLimit { + spawned: 100, + max: 100, + }; + + // First two failures: resurrected to Ready, retry_count increments, no terminal actions. + for expected_retry_count in 1..=2 { + scheduler.graph.tasks[0].status = TaskStatus::Running; + let actions = scheduler.record_spawn_failure(TaskId(0), &error); + assert_eq!( + scheduler.graph.tasks[0].status, + TaskStatus::Ready, + "under FailureStrategy::Retry, SessionSpawnLimit is resurrected like any other \ + failure while retry_count < max_retries" + ); + assert_eq!(scheduler.graph.tasks[0].retry_count, expected_retry_count); + assert!( + actions.is_empty(), + "propagate_failure's Retry branch returns no cancel/done actions on resurrection" + ); + } + + // Third failure: max_retries (2) exhausted — falls through to terminal Abort behavior, + // proving the retry is bounded, not unbounded. + scheduler.graph.tasks[0].status = TaskStatus::Running; + let actions = scheduler.record_spawn_failure(TaskId(0), &error); + assert_eq!( + scheduler.graph.tasks[0].status, + TaskStatus::Failed, + "retries must be bounded by max_retries, not indefinite" + ); + assert!( + actions + .iter() + .any(|a| matches!(a, SchedulerAction::Done { .. })) + ); +} + #[test] fn test_record_spawn_failure_concurrency_limit_reverts_to_ready() { let graph = graph_from_nodes(vec![make_node(0, &[])]); diff --git a/crates/zeph-subagent/README.md b/crates/zeph-subagent/README.md index 589fb4a3f..c331e3a37 100644 --- a/crates/zeph-subagent/README.md +++ b/crates/zeph-subagent/README.md @@ -161,10 +161,21 @@ delegation_mode = "proactive" # "disabled" | "explicit_request_only" | "proacti - `disabled` — no spawn from any code path (slash command, orchestration scheduler, `/subagent spawn`); read-only operations (`/agent list`) still work. - `explicit_request_only` — only spawns attributable to a direct user action (`/agent spawn`, `/agent resume`, `/subagent spawn`) are permitted; the orchestration scheduler's autonomous DAG dispatch is rejected. -- `proactive` — both explicit and autonomous spawns are permitted, subject to the pre-existing `max_concurrent`/`max_spawn_depth`/permission-grant constraints. Matches the subsystem's behavior prior to this field's introduction. +- `proactive` — both explicit and autonomous spawns are permitted, subject to the pre-existing `max_concurrent`/`max_spawn_depth`/`max_spawns_per_session`/permission-grant constraints. Matches the subsystem's behavior prior to this field's introduction. Enforcement is fail-closed: every spawn is tagged with a `SpawnOrigin` (`Explicit` or `Autonomous`) on `SpawnContext`, and an untagged context defaults to `Autonomous` — the restrictive value — so a forgotten call site is denied under the restrictive modes rather than silently allowed. `SubAgentManager::spawn` (and `spawn_for_task`, which delegates to it) is the single chokepoint; a rejected spawn returns `SubAgentError::DelegationDenied` before any resource is allocated. Overridable via `ZEPH_AGENTS_DELEGATION_MODE` or `--delegation-mode`. +## Session-wide spawn cap + +Independent of `max_concurrent` (in-flight limit) and `max_spawn_depth` (recursion limit), `max_spawns_per_session` bounds the *cumulative* number of sub-agents spawned over a session's lifetime — catching a shallow, low-concurrency but high-frequency sequential delegation loop that neither of those limits would (issue #6545): + +```toml +[agents] +max_spawns_per_session = 100 # default: 100; 0 = unlimited +``` + +`SessionSpawnBudget` is a plain, uncloneable `AtomicUsize` counter — `SubAgentManager` owns the origin instance, `zeph-core`'s `OrchestrationState` owns an independent fallback used only when no manager is wired. `SubAgentManager::spawn`/`resume` and `zeph-core`'s ACP `/subagent spawn` chokepoint (which never touches `SubAgentManager` at all) reach whichever instance applies through an `Agent::session_budget()` accessor that hands out a `&SessionSpawnBudget` reference, so both paths enforce the exact same session-wide counter without ever copying it. Checked before `max_spawn_depth`/`max_concurrent`, but only *consumed* at each path's true commit point — a spawn rejected for any other reason (`NotFound`, a transient `ConcurrencyLimit` the orchestration scheduler will retry, a failed ACP process launch) never burns budget it never used. Reaching the cap returns `SubAgentError::SessionSpawnLimit`, whose `Display` names the config key directly. Resets at session start; never persisted. + ## Features | Feature | Description | diff --git a/crates/zeph-subagent/src/budget.rs b/crates/zeph-subagent/src/budget.rs new file mode 100644 index 000000000..ab9e46801 --- /dev/null +++ b/crates/zeph-subagent/src/budget.rs @@ -0,0 +1,165 @@ +// SPDX-FileCopyrightText: 2026 Andrei G +// SPDX-License-Identifier: MIT OR Apache-2.0 + +//! Session-wide cumulative subagent spawn budget (issue #6545). + +use std::sync::atomic::{AtomicUsize, Ordering}; + +use crate::error::SubAgentError; + +/// Session-wide cumulative counter of subagent spawns. +/// +/// Bounds the *total* number of subagents spawned over a session's lifetime, independent of +/// [`SubAgentManager::spawn`](crate::manager::SubAgentManager::spawn)'s existing +/// `max_concurrent` (in-flight) and `max_spawn_depth` (recursion) guardrails — a shallow, +/// low-concurrency but high-frequency sequential delegation loop trips neither of those. +/// +/// # Ownership, not a shared handle +/// +/// Deliberately a plain `AtomicUsize` newtype — no `Arc`, no `Clone`. Nothing in this crate +/// needs a shared, cloned handle to a budget: `SubAgentManager` owns one instance as the origin +/// of truth, and `zeph-core`'s `OrchestrationState` owns an independent fallback instance used +/// only when no manager is wired. Both the manager-side spawn path and the ACP `/subagent +/// spawn` chokepoint (which never touches `SubAgentManager` at all) reach whichever instance +/// applies through an accessor (`Agent::session_budget` in `zeph-core`) that hands out a plain +/// `&SessionSpawnBudget` reference, never a copy. +/// +/// This also makes a would-be TOCTOU hazard structurally impossible rather than merely +/// documented: such a hazard could only arise if `SubAgentManager` were ever shared (e.g. +/// behind an `Arc`) across concurrent tasks, and a future refactor down that path would have to +/// deliberately reintroduce `Clone`/`Arc` here — a change reviewable on its own, rather than one +/// hiding behind an innocuous `.clone()` call at some unrelated call site. +/// +/// # Concurrency +/// +/// `AtomicUsize` provides interior mutability so [`check`](Self::check) and +/// [`record_spawn`](Self::record_spawn) work through a shared `&self` reference (every access +/// goes through `&self` accessors, never `&mut self`), satisfying NFR-001's atomic-counter +/// requirement. Every call path that reaches either method — `SubAgentManager::spawn`/`resume`, +/// the orchestration scheduler, and the ACP chokepoint — is serialized behind `&mut Agent` on +/// the single agent task, so `Relaxed` ordering suffices. The check/consume split (budget is +/// checked at the spawn guard but only consumed at the true commit point, so a rejected or +/// transiently retried spawn never burns budget it never used) introduces no reachable race +/// under that serialization. +/// +/// # Examples +/// +/// ```rust +/// use zeph_subagent::SessionSpawnBudget; +/// +/// let budget = SessionSpawnBudget::default(); +/// assert_eq!(budget.spawned(), 0); +/// +/// budget.check(1).expect("budget not yet exhausted"); +/// budget.record_spawn(); +/// assert_eq!(budget.spawned(), 1); +/// assert!(budget.check(1).is_err(), "cap of 1 must now be exhausted"); +/// +/// // `0` is the unlimited sentinel: check() always succeeds regardless of count. +/// assert!(budget.check(0).is_ok()); +/// ``` +#[derive(Default)] +pub struct SessionSpawnBudget(AtomicUsize); + +impl SessionSpawnBudget { + /// Check the budget without consuming it. + /// + /// `max == 0` is the unlimited sentinel and always succeeds, so callers never need to + /// duplicate the sentinel check themselves (mirrors + /// [`DelegationMode::permits_explicit`](zeph_config::DelegationMode::permits_explicit)'s + /// anti-drift rationale for a check shared across multiple chokepoints). + /// + /// # Errors + /// + /// Returns [`SubAgentError::SessionSpawnLimit`] when the cumulative spawn count has + /// already reached `max`. + pub fn check(&self, max: usize) -> Result<(), SubAgentError> { + if max == 0 { + return Ok(()); + } + let spawned = self.0.load(Ordering::Relaxed); + if spawned >= max { + return Err(SubAgentError::SessionSpawnLimit { spawned, max }); + } + Ok(()) + } + + /// Record a successful spawn, incrementing the cumulative count by one. + /// + /// Must be called only at a spawn's true commit point — see the check/consume split + /// described in the type-level concurrency note. + pub fn record_spawn(&self) { + self.0.fetch_add(1, Ordering::Relaxed); + } + + /// Current cumulative spawn count. + #[must_use] + pub fn spawned(&self) -> usize { + self.0.load(Ordering::Relaxed) + } +} + +impl std::fmt::Debug for SessionSpawnBudget { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_tuple("SessionSpawnBudget") + .field(&self.spawned()) + .finish() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_mints_independent_counters() { + let a = SessionSpawnBudget::default(); + let b = SessionSpawnBudget::default(); + a.record_spawn(); + assert_eq!(a.spawned(), 1); + assert_eq!( + b.spawned(), + 0, + "default() must not share state across instances" + ); + } + + #[test] + fn zero_is_unlimited_sentinel() { + let budget = SessionSpawnBudget::default(); + for _ in 0..1000 { + budget.record_spawn(); + } + assert!(budget.check(0).is_ok()); + } + + #[test] + fn check_does_not_consume() { + let budget = SessionSpawnBudget::default(); + budget.check(5).unwrap(); + budget.check(5).unwrap(); + assert_eq!(budget.spawned(), 0, "check() must be read-only"); + } + + #[test] + fn cap_reached_returns_session_spawn_limit() { + let budget = SessionSpawnBudget::default(); + budget.record_spawn(); + let err = budget.check(1).unwrap_err(); + assert!(matches!( + err, + SubAgentError::SessionSpawnLimit { spawned: 1, max: 1 } + )); + } + + #[test] + fn debug_prints_count() { + let budget = SessionSpawnBudget::default(); + budget.record_spawn(); + let debug = format!("{budget:?}"); + assert!( + debug.contains('1'), + "Debug output must surface the count: {debug}" + ); + } +} diff --git a/crates/zeph-subagent/src/error.rs b/crates/zeph-subagent/src/error.rs index b6f944c0a..059717e2b 100644 --- a/crates/zeph-subagent/src/error.rs +++ b/crates/zeph-subagent/src/error.rs @@ -122,4 +122,39 @@ pub enum SubAgentError { origin: crate::manager::SpawnOrigin, def_name: String, }, + + /// The session-wide cumulative spawn budget has been exhausted (issue #6545). + /// + /// Distinct from [`SubAgentError::ConcurrencyLimit`] (bounds in-flight agents) and + /// [`SubAgentError::MaxDepthExceeded`] (bounds recursion depth): this bounds the total + /// number of subagents spawned over the session's lifetime, independent of both, so a + /// shallow, low-concurrency but high-frequency sequential delegation loop is still caught. + /// The `Display` string names the config key directly because the only user-visible + /// surface for most callers is `format!("Failed to spawn sub-agent: {e}")`. + #[error( + "session spawn limit reached (spawned: {spawned}, max: {max}) — raise \ + [agents].max_spawns_per_session in config.toml, or set it to 0 for unlimited" + )] + SessionSpawnLimit { spawned: usize, max: usize }, +} + +#[cfg(test)] +mod tests { + use super::*; + + /// NFR-004 (issue #6545): the only user-visible surface for most callers is + /// `format!("Failed to spawn sub-agent: {e}")`, so the remedy must live in `Display` + /// itself — verify the config key is named verbatim, not just implied. + #[test] + fn session_spawn_limit_display_names_config_key() { + let err = SubAgentError::SessionSpawnLimit { + spawned: 100, + max: 100, + }; + let msg = err.to_string(); + assert!( + msg.contains("[agents].max_spawns_per_session"), + "Display must name the config key verbatim, got: {msg}" + ); + } } diff --git a/crates/zeph-subagent/src/lib.rs b/crates/zeph-subagent/src/lib.rs index 559a1c1cb..94237e99d 100644 --- a/crates/zeph-subagent/src/lib.rs +++ b/crates/zeph-subagent/src/lib.rs @@ -39,6 +39,7 @@ //! ``` mod agent_loop; +pub mod budget; pub mod command; pub mod cwd_guard; pub mod def; @@ -55,6 +56,7 @@ pub mod resolve; pub mod state; pub mod transcript; +pub use budget::SessionSpawnBudget; pub use command::{AgentCommand, AgentsCommand}; pub use cwd_guard::CwdLock; pub use def::{ diff --git a/crates/zeph-subagent/src/manager/mod.rs b/crates/zeph-subagent/src/manager/mod.rs index 9833b0840..f345843e7 100644 --- a/crates/zeph-subagent/src/manager/mod.rs +++ b/crates/zeph-subagent/src/manager/mod.rs @@ -526,6 +526,14 @@ pub struct SubAgentManager { /// is authoritative at spawn time; `SubAgentManager` does not re-read `enabled` itself. /// Defaults to [`zeph_config::DelegationMode::default()`] (`Proactive`) until set. delegation_mode: zeph_config::DelegationMode, + /// Session-wide cumulative subagent-spawn budget (issue #6545). + /// + /// This manager owns the origin instance: [`session_budget`][Self::session_budget] exposes + /// a `&SessionSpawnBudget` reference so other chokepoints that never touch this manager — + /// e.g. the ACP `/subagent spawn` path in `zeph-core`'s `handle_subagent_slash` — can + /// enforce the same session-wide cap. See [`SessionSpawnBudget`]'s own doc comment for why + /// it is a plain, uncloned `AtomicUsize` newtype rather than a shared `Arc` handle. + session_spawn_budget: crate::budget::SessionSpawnBudget, } impl std::fmt::Debug for SubAgentManager { @@ -549,6 +557,7 @@ impl std::fmt::Debug for SubAgentManager { .field("secret_registry", &self.secret_registry.is_some()) .field("pii_filter", &self.pii_filter.is_some()) .field("delegation_mode", &self.delegation_mode) + .field("session_spawn_budget", &self.session_spawn_budget) .finish() } } @@ -576,9 +585,23 @@ impl SubAgentManager { secret_registry: None, pii_filter: None, delegation_mode: zeph_config::DelegationMode::default(), + session_spawn_budget: crate::budget::SessionSpawnBudget::default(), } } + /// The session-wide cumulative subagent-spawn budget this manager originates (issue + /// #6545). + /// + /// Returns a reference to this manager's own budget instance — not a fresh, independent + /// one — so a caller reading through this accessor observes the same cumulative count as + /// every spawn through this manager. See + /// [`SessionSpawnBudget`][crate::budget::SessionSpawnBudget]'s doc comment for why the type + /// itself has no `Clone`/`Arc`. + #[must_use] + pub fn session_budget(&self) -> &crate::budget::SessionSpawnBudget { + &self.session_spawn_budget + } + /// Inject a [`TaskSupervisor`] so subagent lifecycle tasks are registered and visible. /// /// Must be called before the first [`spawn`][Self::spawn]. When set, each spawned agent diff --git a/crates/zeph-subagent/src/manager/spawn.rs b/crates/zeph-subagent/src/manager/spawn.rs index ad4ab51ec..161c6dc54 100644 --- a/crates/zeph-subagent/src/manager/spawn.rs +++ b/crates/zeph-subagent/src/manager/spawn.rs @@ -570,9 +570,10 @@ impl SubAgentManager { /// # Errors /// /// Returns [`SubAgentError::NotFound`] if no definition with the given name exists, - /// [`SubAgentError::ConcurrencyLimit`] if the concurrency limit is exceeded, or - /// [`SubAgentError::Invalid`] if the agent requests `bypass_permissions` but the config - /// does not allow it (`allow_bypass_permissions: false`). + /// [`SubAgentError::ConcurrencyLimit`] if the concurrency limit is exceeded, + /// [`SubAgentError::SessionSpawnLimit`] if the session-wide cumulative spawn cap has been + /// reached, or [`SubAgentError::Invalid`] if the agent requests `bypass_permissions` but + /// the config does not allow it (`allow_bypass_permissions: false`). #[allow(clippy::too_many_arguments, clippy::too_many_lines)] // complex algorithm function; both suppressions justified until the function is decomposed in a future refactor #[tracing::instrument(name = "subagent.manager.spawn", skip_all, fields(def_name = def_name))] @@ -614,6 +615,26 @@ impl SubAgentManager { }); } + // Session-wide cumulative spawn cap (issue #6545): checked before the depth/concurrency + // checks below, same as the delegation gate above. Deliberately also precedes the + // max_spawn_depth check, one step further than issue #6545 formally requires (only + // priority over ConcurrencyLimit was required) — harmless today since `spawn_depth` is + // always 0 in production, and it keeps both "no resources allocated yet" guards + // adjacent. Read-only: budget is consumed only at the commit point below, not here, so + // a spawn rejected by a later check (NotFound, ConcurrencyLimit) never burns budget it + // never used. + if let Err(e) = self + .session_spawn_budget + .check(config.max_spawns_per_session) + { + tracing::warn!( + error = %e, + def_name, + "sub-agent spawn rejected: session spawn budget exhausted" + ); + return Err(e); + } + if ctx.spawn_depth >= config.max_spawn_depth { return Err(SubAgentError::MaxDepthExceeded { depth: ctx.spawn_depth, @@ -896,6 +917,11 @@ impl SubAgentManager { }; self.agents.insert(task_id.clone(), handle); + // Commit point for the session-wide spawn budget (issue #6545): the handle is now + // owned by the manager and every fallible step above has already succeeded, so this + // spawn is real and must count toward the cap. Must stay after the insert, not at the + // guard above — see the check/consume split note there. + self.session_spawn_budget.record_spawn(); if let Some(ref registry) = self.fleet_registry { let registry = Arc::clone(registry); @@ -1062,7 +1088,9 @@ impl SubAgentManager { /// [`SubAgentError::NotFound`] if no transcript with the given prefix exists, /// [`SubAgentError::AmbiguousId`] if the prefix matches multiple agents, /// [`SubAgentError::Transcript`] on I/O or parse failure, - /// [`SubAgentError::ConcurrencyLimit`] if the concurrency limit is exceeded. + /// [`SubAgentError::ConcurrencyLimit`] if the concurrency limit is exceeded, or + /// [`SubAgentError::SessionSpawnLimit`] if the session-wide cumulative spawn cap has been + /// reached. #[allow(clippy::too_many_lines, clippy::too_many_arguments)] #[tracing::instrument(name = "subagent.manager.resume", skip_all, fields(id_prefix = id_prefix))] pub async fn resume( @@ -1093,6 +1121,22 @@ impl SubAgentManager { }); } + // Session-wide cumulative spawn cap (issue #6545): `resume()` allocates the identical + // per-spawn resources `spawn()` does (transcript writer, agent loop task, handle), so + // an `/agent resume`-in-a-loop bypass would otherwise be uncapped. Read-only here; see + // the check/consume split note on the `spawn()` guard above. + if let Err(e) = self + .session_spawn_budget + .check(config.max_spawns_per_session) + { + tracing::warn!( + error = %e, + id_prefix, + "sub-agent resume rejected: session spawn budget exhausted" + ); + return Err(e); + } + let dir = self.effective_transcript_dir(config); let id_prefix_owned = id_prefix.to_owned(); let dir_clone = dir.clone(); @@ -1301,6 +1345,9 @@ impl SubAgentManager { }; self.agents.insert(new_task_id.clone(), handle); + // Commit point for the session-wide spawn budget (issue #6545) — see the matching + // note in `spawn()`. + self.session_spawn_budget.record_spawn(); tracing::info!( task_id = %new_task_id, original_id = %original_id, diff --git a/crates/zeph-subagent/src/manager/tests.rs b/crates/zeph-subagent/src/manager/tests.rs index b5f3631c4..bc95bddb8 100644 --- a/crates/zeph-subagent/src/manager/tests.rs +++ b/crates/zeph-subagent/src/manager/tests.rs @@ -504,6 +504,290 @@ async fn concurrency_limit_enforced() { assert_matches!(err, SubAgentError::ConcurrencyLimit { .. }); } +mod session_spawn_budget_gate { + //! Session-wide cumulative subagent spawn cap (issue #6545). + + use super::*; + + fn cfg_with_max_spawns(max: usize) -> SubAgentConfig { + SubAgentConfig { + max_spawns_per_session: max, + ..SubAgentConfig::default() + } + } + + #[tokio::test] + async fn cap_reached_returns_session_spawn_limit() { + let mut mgr = SubAgentManager::new(10); + mgr.definitions.push(sample_def()); + let cfg = cfg_with_max_spawns(1); + + mgr.spawn( + "bot", + "first", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + ) + .await + .unwrap(); + + let err = mgr + .spawn( + "bot", + "second", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + ) + .await + .unwrap_err(); + assert_matches!(err, SubAgentError::SessionSpawnLimit { spawned: 1, max: 1 }); + } + + #[tokio::test] + async fn zero_is_unlimited_sentinel() { + let mut mgr = SubAgentManager::new(10); + mgr.definitions.push(sample_def()); + let cfg = cfg_with_max_spawns(0); + + for i in 0..5 { + mgr.spawn( + "bot", + &format!("task-{i}"), + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + ) + .await + .unwrap(); + } + assert_eq!(mgr.session_budget().spawned(), 5); + } + + #[tokio::test] + async fn concurrency_limit_rejection_does_not_consume_budget() { + let mut mgr = SubAgentManager::new(1); + mgr.definitions.push(sample_def()); + let cfg = cfg_with_max_spawns(10); + + mgr.spawn( + "bot", + "first", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + ) + .await + .unwrap(); + + let err = mgr + .spawn( + "bot", + "second", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + ) + .await + .unwrap_err(); + assert_matches!(err, SubAgentError::ConcurrencyLimit { .. }); + assert_eq!( + mgr.session_budget().spawned(), + 1, + "a transient ConcurrencyLimit rejection must not consume budget \ + (DagScheduler::record_spawn_failure retries it)" + ); + } + + #[tokio::test] + async fn not_found_rejection_does_not_consume_budget() { + let mut mgr = SubAgentManager::new(10); + let cfg = cfg_with_max_spawns(10); + + let err = mgr + .spawn( + "missing", + "task", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + ) + .await + .unwrap_err(); + assert_matches!(err, SubAgentError::NotFound(_)); + assert_eq!(mgr.session_budget().spawned(), 0); + } + + #[tokio::test] + async fn session_spawn_limit_takes_priority_over_concurrency_limit() { + let mut mgr = SubAgentManager::new(1); + mgr.definitions.push(sample_def()); + let cfg = cfg_with_max_spawns(1); + + mgr.spawn( + "bot", + "first", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + ) + .await + .unwrap(); + + // Both the session cap (1) and the concurrency limit (1) are now exhausted. + let err = mgr + .spawn( + "bot", + "second", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + ) + .await + .unwrap_err(); + assert_matches!( + err, + SubAgentError::SessionSpawnLimit { .. }, + "the session cap must be checked before the concurrency check" + ); + } + + #[tokio::test] + async fn resume_is_gated_and_counts() { + let tmp = tempfile::tempdir().unwrap(); + let agent_id = "deadcode-0000-0000-0000-000000000000"; + write_completed_meta(tmp.path(), agent_id, "bot"); + + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + let cfg = SubAgentConfig { + transcript_dir: Some(tmp.path().to_path_buf()), + max_spawns_per_session: 1, + ..SubAgentConfig::default() + }; + + let (new_id, _def_name) = mgr + .resume( + "deadcode", + "continue the work", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + None, + ) + .await + .unwrap(); + assert_eq!(mgr.session_budget().spawned(), 1); + mgr.cancel(&new_id).unwrap(); + + let agent_id2 = "beefcode-0000-0000-0000-000000000000"; + write_completed_meta(tmp.path(), agent_id2, "bot"); + let err = mgr + .resume( + "beefcode", + "continue the work", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + None, + ) + .await + .unwrap_err(); + assert_matches!(err, SubAgentError::SessionSpawnLimit { .. }); + } + + /// `spawn_for_task` delegates straight to `spawn` (see `manager/spawn.rs`), so the same + /// budget gate must reject it too — mirrors `spawn_for_task_is_gated_identically` in + /// `delegation_mode_gate` below. + #[tokio::test] + async fn spawn_for_task_is_gated_and_counts() { + let mut mgr = SubAgentManager::new(10); + mgr.definitions.push(sample_def()); + let cfg = cfg_with_max_spawns(1); + + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + mgr.spawn_for_task( + "bot", + "task", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + move |id, result| { + let _ = tx.send((id, result)); + }, + ) + .await + .unwrap(); + assert_eq!(mgr.session_budget().spawned(), 1); + + let (tx2, _rx2) = tokio::sync::mpsc::unbounded_channel(); + let err = mgr + .spawn_for_task( + "bot", + "task2", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + move |id, result| { + let _ = tx2.send((id, result)); + }, + ) + .await + .unwrap_err(); + assert_matches!(err, SubAgentError::SessionSpawnLimit { .. }); + assert!( + rx.try_recv().is_err(), + "on_done must never fire for a rejected spawn" + ); + } + + #[tokio::test] + async fn session_budget_accessor_reflects_spawn_count() { + let mut mgr = SubAgentManager::new(10); + mgr.definitions.push(sample_def()); + let cfg = cfg_with_max_spawns(0); + + assert_eq!(mgr.session_budget().spawned(), 0); + for i in 0..3 { + mgr.spawn( + "bot", + &format!("task-{i}"), + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + SpawnContext::default(), + ) + .await + .unwrap(); + } + assert_eq!(mgr.session_budget().spawned(), 3); + } +} + // --- #1619 regression tests: reserved_slots --- #[tokio::test]