diff --git a/.zeph/skills/setup-guide/SKILL.md b/.zeph/skills/setup-guide/SKILL.md index f513ab61d..b18d33abb 100644 --- a/.zeph/skills/setup-guide/SKILL.md +++ b/.zeph/skills/setup-guide/SKILL.md @@ -264,6 +264,18 @@ export ZEPH_AGENTS_FORWARD_TRANSCRIPT=true ``` Opt-in, default `false` (`agents.forward_transcript` in config). Also settable via `--forward-subagent-text`. No effect unless a consumer surface (`--tui` or `--bare`) is active for the session. +Delegation mode — control whether the main agent may spawn sub-agents, and who may trigger it: +```bash +export ZEPH_AGENTS_DELEGATION_MODE=explicit_request_only +``` +One of `disabled` (no spawn from any code path), `explicit_request_only` (only direct user +actions — `/agent spawn`, `/agent resume`, `/subagent spawn` — the main agent's own planner/ +scheduler may never spawn autonomously), or `proactive` (default; both explicit and autonomous +spawns allowed). Orthogonal to `agents.enabled`, which remains the outer kill switch: +`enabled = false` always behaves as `disabled` regardless of this setting. Also settable via +`--delegation-mode`. Useful to restrict in semi-trusted channels (Telegram/Discord/webhook +ingestion) where prompt-injected input could otherwise trigger unsupervised delegation. + ## Security Secret redaction: diff --git a/CHANGELOG.md b/CHANGELOG.md index cb72d69da..61084525f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,49 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). existed to support — `zeph-agent-tools` now has zero workspace or external dependencies. Behavior of `doom_loop_hash` and its call site are unchanged. +### Added + +- `zeph-subagent`: a running subagent's live transcript forwarding (`forward_transcript`, + issue #6359) now streams text and thinking output as token-level deltas *within* a turn + instead of only once the turn completes (#6456, FR-002b), on providers whose native + streaming-with-tools path is available (currently Claude, via the existing + `AnyProvider::chat_with_tools_stream`/`ToolSseStream`). `agent_loop.rs` drives the stream + and forwards each `ContentChunk`/`ThinkingChunk` immediately through the same tail-drop + forwarding channel `send_text`/`send_thinking` already used for FR-002a, while assembling + the identical `ChatResponse` shape `chat_with_tools` would have returned. No config change + is required — this is strictly finer-grained delivery under the existing + `forward_transcript` flag. Deltas remain ephemeral and display-only (tail-droppable under + backpressure, same as before): the accumulated response text and the guaranteed terminal + chunk remain the sole source of truth for the next turn's LLM context. Providers without a + native tool-streaming implementation continue to forward the full per-turn text once, at + turn end, exactly as before (zero-regression fallback, no new code path per backend). The + forwarding drain (`forward.rs`) now buffers each streamed delta with a bounded 256-byte + holdback window before sanitizing and emitting it, instead of sanitizing every delta in + complete isolation — closes a masking gap where a secret or PII pattern split across two + delta boundaries matched neither fragment individually and reached `--bare` stdout / the + TUI ring buffer unmasked. Buffered content is only ever delayed, never dropped: any + remaining held-back tail is flushed in full immediately before the terminal chunk. + +- `zeph-config`/`zeph-subagent`/`zeph-core`: `SubAgentConfig` gains a tri-state + `delegation_mode` (`disabled` / `explicit_request_only` / `proactive`, spec + `042-subagent-delegation-mode-parity`, issue #5857), so an operator can keep sub-agents + enabled and useful while forbidding the main agent from autonomously deciding to spawn one — + e.g. in a Telegram/Discord/webhook-facing channel where prompt-injected input may reach the + agent. `explicit_request_only` permits only spawns attributable to a direct user action + (`/agent spawn`, `/agent resume`, `/subagent spawn`) and rejects the orchestration + scheduler's autonomous DAG dispatch; `disabled` rejects every spawn path (read-only + operations like `/agent list` still work); `proactive` (the default) preserves the + subsystem's prior unconstrained behavior. Enforcement is fail-closed: a new + `SpawnContext.origin: SpawnOrigin` field defaults to `Autonomous` (the restrictive value), + so an untagged spawn site is denied under the restrictive modes rather than silently + allowed — every real `.spawn()`/`.spawn_for_task()` call site was audited and explicitly + tagged. `SubAgentConfig.enabled` becomes the outer kill switch: `enabled = false` always + resolves to the `disabled` behavior regardless of `delegation_mode`'s configured value. + Rejected spawns return the new `SubAgentError::DelegationDenied` and log a distinguishable + `tracing::warn!`. Overridable via `ZEPH_AGENTS_DELEGATION_MODE` or `--delegation-mode`; the + `--init` wizard and `--migrate-config` (existing `[agents] enabled = true` configs gain an + explicit `delegation_mode = "proactive"`) are updated accordingly. + ### Changed - `zeph-skills`: `embed_skills_with_timeout` now tries `LlmProvider::embed_batch` first diff --git a/Cargo.lock b/Cargo.lock index a8003b107..a8aec0308 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11989,6 +11989,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", + "tokio-stream", "tokio-util", "toml 1.1.2+spec-1.1.0", "tracing", diff --git a/config/default.toml b/config/default.toml index be39c96a5..c43afc4cd 100644 --- a/config/default.toml +++ b/config/default.toml @@ -1184,6 +1184,11 @@ notify_ack_timeout_ms = 5000 [agents] # Enable sub-agent spawning (required for /agent commands and multi-agent workflows) enabled = false +# Whether the main agent may spawn sub-agents, and who may trigger it: "disabled" / +# "explicit_request_only" / "proactive". Orthogonal to `enabled` above, which remains the +# outer kill switch (enabled = false always resolves to the "disabled" behavior regardless +# of this value). Env override: ZEPH_AGENTS_DELEGATION_MODE. CLI override: --delegation-mode. +delegation_mode = "proactive" # Maximum number of sub-agents that can run concurrently max_concurrent = 1 # Allow sub-agents to use bypass_permissions mode (enable only in trusted environments) diff --git a/crates/zeph-config/src/agent.rs b/crates/zeph-config/src/agent.rs index 2b4056ac7..98b7dca29 100644 --- a/crates/zeph-config/src/agent.rs +++ b/crates/zeph-config/src/agent.rs @@ -91,6 +91,61 @@ pub enum ContextInjectionMode { Summary, } +/// Tri-state control over whether the main agent may spawn sub-agents, and who may trigger it +/// (spec `042-subagent-delegation-mode-parity`, issue #5857). +/// +/// Orthogonal to [`SubAgentConfig::enabled`], which remains the outer kill switch: when +/// `enabled = false`, the effective mode is always [`DelegationMode::Disabled`] regardless of +/// this field's value (FR-002). Also orthogonal to [`PermissionMode`] — that governs what a +/// spawned sub-agent may *do*; this governs whether a spawn may happen *at all* and who may +/// trigger it. +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +#[non_exhaustive] +pub enum DelegationMode { + /// No spawn may proceed from any code path (slash command, orchestration planner/scheduler, + /// scheduled task). Read-only operations (`/agent list`, definition inspection, status + /// queries) remain available. + Disabled, + /// Only spawns attributable to a direct, explicit user action (e.g. `/agent spawn`) are + /// permitted; spawns originating from autonomous planner/scheduler decision-making are + /// rejected. + ExplicitRequestOnly, + /// Both explicit and autonomous spawn paths are permitted, subject to the pre-existing + /// constraints (`max_concurrent`, `max_spawn_depth`, permission grants, worktree isolation). + /// Matches the subsystem's behavior prior to this field's introduction. + #[default] + Proactive, +} + +impl DelegationMode { + /// Whether a spawn/resume attempt attributable to a direct, explicit user action (e.g. + /// `/agent spawn`, `/agent resume`, `/subagent spawn`) is permitted under this mode. + /// + /// Expressed as an allow-list (`Proactive` and `ExplicitRequestOnly` match; every other + /// value, including any future `#[non_exhaustive]` variant, does not) rather than a + /// deny-list (`self != Disabled`), so it fails closed automatically on a variant this + /// crate doesn't yet recognize instead of silently permitting it. Shared by every + /// origin-agnostic "is this explicit action even allowed at all" check — the ACP + /// `/subagent spawn` gate and `SubAgentManager::resume` — so the two enforcement points + /// cannot drift out of sync with each other. Does **not** cover `Autonomous`-origin spawns; + /// `SubAgentManager::spawn`'s own origin-aware gate handles that distinction directly. + /// + /// # Examples + /// + /// ```rust + /// use zeph_config::DelegationMode; + /// + /// assert!(DelegationMode::Proactive.permits_explicit()); + /// assert!(DelegationMode::ExplicitRequestOnly.permits_explicit()); + /// assert!(!DelegationMode::Disabled.permits_explicit()); + /// ``` + #[must_use] + pub fn permits_explicit(self) -> bool { + matches!(self, Self::Proactive | Self::ExplicitRequestOnly) + } +} + fn default_max_parent_messages() -> usize { 20 } @@ -542,6 +597,7 @@ impl Default for TaskSupervisorConfig { /// ```toml /// [agents] /// enabled = true +/// delegation_mode = "explicit_request_only" /// max_concurrent = 3 /// max_spawn_depth = 2 /// ``` @@ -550,7 +606,17 @@ impl Default for TaskSupervisorConfig { #[allow(clippy::struct_excessive_bools)] // independent config toggles; bitflags or enum would obscure semantics without reducing complexity pub struct SubAgentConfig { /// Enable the sub-agent subsystem. Default: `false`. + /// + /// Outer kill switch: when `false`, the effective [`delegation_mode`][Self::delegation_mode] + /// is always [`DelegationMode::Disabled`] regardless of that field's configured value + /// (spec `042-subagent-delegation-mode-parity` FR-002). pub enabled: bool, + /// Whether the main agent may spawn sub-agents, and who may trigger it: `disabled` / + /// `explicit_request_only` / `proactive`. Default: [`DelegationMode::Proactive`] (preserves + /// the subsystem's unconstrained behavior prior to this field's introduction, per FR-008). + /// Overridable via `ZEPH_AGENTS_DELEGATION_MODE` or the `--delegation-mode` CLI flag. + #[serde(default)] + pub delegation_mode: DelegationMode, /// Maximum number of sub-agents that can run concurrently. #[serde(default = "default_max_concurrent")] pub max_concurrent: usize, @@ -650,6 +716,7 @@ impl Default for SubAgentConfig { fn default() -> Self { Self { enabled: false, + delegation_mode: DelegationMode::default(), max_concurrent: default_max_concurrent(), extra_dirs: Vec::new(), user_agents_dir: None, @@ -674,6 +741,39 @@ impl Default for SubAgentConfig { } } +impl SubAgentConfig { + /// Resolve [`enabled`][Self::enabled] and [`delegation_mode`][Self::delegation_mode] into + /// the single effective mode that must be enforced at every spawn call site (spec + /// `042-subagent-delegation-mode-parity` FR-002, issue #5857). + /// + /// `enabled` is the outer kill switch: `enabled = false` always resolves to + /// [`DelegationMode::Disabled`], regardless of the configured `delegation_mode` value. + /// + /// # Examples + /// + /// ```rust + /// use zeph_config::{DelegationMode, SubAgentConfig}; + /// + /// let mut cfg = SubAgentConfig { + /// enabled: false, + /// delegation_mode: DelegationMode::Proactive, + /// ..SubAgentConfig::default() + /// }; + /// assert_eq!(cfg.effective_delegation_mode(), DelegationMode::Disabled); + /// + /// cfg.enabled = true; + /// assert_eq!(cfg.effective_delegation_mode(), DelegationMode::Proactive); + /// ``` + #[must_use] + pub fn effective_delegation_mode(&self) -> DelegationMode { + if self.enabled { + self.delegation_mode + } else { + DelegationMode::Disabled + } + } +} + /// Config-level lifecycle hooks fired when any sub-agent starts or stops. #[derive(Debug, Clone, Default, Deserialize, Serialize)] #[serde(default)] @@ -708,6 +808,66 @@ mod tests { ); } + #[test] + fn subagent_config_delegation_mode_defaults_proactive() { + let cfg = SubAgentConfig::default(); + assert_eq!(cfg.delegation_mode, DelegationMode::Proactive); + } + + #[test] + fn subagent_config_deserialize_delegation_mode() { + let toml_str = "enabled = true\ndelegation_mode = \"explicit_request_only\""; + let cfg: SubAgentConfig = toml::from_str(toml_str).unwrap(); + assert_eq!(cfg.delegation_mode, DelegationMode::ExplicitRequestOnly); + } + + #[test] + fn subagent_config_delegation_mode_omitted_defaults_proactive() { + let toml_str = "enabled = true"; + let cfg: SubAgentConfig = toml::from_str(toml_str).unwrap(); + assert_eq!(cfg.delegation_mode, DelegationMode::Proactive); + } + + #[test] + fn subagent_config_delegation_mode_rejects_unknown_value() { + let toml_str = "delegation_mode = \"sometimes\""; + let result: Result = toml::from_str(toml_str); + assert!(result.is_err(), "unrecognized value must fail to parse"); + } + + #[test] + fn permits_explicit_allow_list() { + assert!(DelegationMode::Proactive.permits_explicit()); + assert!(DelegationMode::ExplicitRequestOnly.permits_explicit()); + assert!(!DelegationMode::Disabled.permits_explicit()); + } + + #[test] + fn effective_delegation_mode_disabled_when_not_enabled() { + let cfg = SubAgentConfig { + enabled: false, + delegation_mode: DelegationMode::Proactive, + ..SubAgentConfig::default() + }; + assert_eq!(cfg.effective_delegation_mode(), DelegationMode::Disabled); + } + + #[test] + fn effective_delegation_mode_passes_through_when_enabled() { + for mode in [ + DelegationMode::Disabled, + DelegationMode::ExplicitRequestOnly, + DelegationMode::Proactive, + ] { + let cfg = SubAgentConfig { + enabled: true, + delegation_mode: mode, + ..SubAgentConfig::default() + }; + assert_eq!(cfg.effective_delegation_mode(), mode); + } + } + #[test] fn subagent_config_deserialize_forward_transcript() { let toml_str = "forward_transcript = true"; diff --git a/crates/zeph-config/src/env.rs b/crates/zeph-config/src/env.rs index cd04bcf09..ed18df64b 100644 --- a/crates/zeph-config/src/env.rs +++ b/crates/zeph-config/src/env.rs @@ -361,6 +361,20 @@ impl Config { { self.agents.forward_transcript = enabled; } + if let Ok(v) = std::env::var("ZEPH_AGENTS_DELEGATION_MODE") { + match v.as_str() { + "disabled" => self.agents.delegation_mode = crate::DelegationMode::Disabled, + "explicit_request_only" => { + self.agents.delegation_mode = crate::DelegationMode::ExplicitRequestOnly; + } + "proactive" => self.agents.delegation_mode = crate::DelegationMode::Proactive, + other => tracing::warn!( + value = other, + "ZEPH_AGENTS_DELEGATION_MODE: invalid value, ignoring \ + (expected disabled|explicit_request_only|proactive)" + ), + } + } } fn apply_env_overrides_security(&mut self) { diff --git a/crates/zeph-config/src/lib.rs b/crates/zeph-config/src/lib.rs index 95d40b03b..10d311beb 100644 --- a/crates/zeph-config/src/lib.rs +++ b/crates/zeph-config/src/lib.rs @@ -115,8 +115,9 @@ pub mod vigil; pub mod worktree; pub use agent::{ - AgentConfig, ContextInjectionMode, FocusConfig, GoalConfig, ModelSpec, ParentContextPolicy, - SubAgentConfig, SubAgentLifecycleHooks, TaskSupervisorConfig, ToolFilterConfig, + AgentConfig, ContextInjectionMode, DelegationMode, FocusConfig, GoalConfig, ModelSpec, + ParentContextPolicy, SubAgentConfig, SubAgentLifecycleHooks, TaskSupervisorConfig, + ToolFilterConfig, }; pub use channels::{ A2aClientConfig, A2aServerConfig, CardTrustPolicy, ChannelSkillsConfig, DiscordConfig, diff --git a/crates/zeph-config/src/migrate/mod.rs b/crates/zeph-config/src/migrate/mod.rs index 0903cf31a..09538f9b0 100644 --- a/crates/zeph-config/src/migrate/mod.rs +++ b/crates/zeph-config/src/migrate/mod.rs @@ -20,6 +20,7 @@ mod memory; mod plugins; mod serve; mod session; +mod subagent; mod tools; pub use features::{ @@ -42,6 +43,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 tools::*; /// Returns `true` when `name` is an active (non-commented) TOML section header in `src`. @@ -605,24 +607,24 @@ mod steps; use steps::{ MigrateA2aCardTrustConfig, MigrateA2aServerRemoveInertFields, MigrateAcpAuthClientsConfig, MigrateAcpSubagentsConfig, MigrateAgentBudgetHint, MigrateAgentRetryToToolsRetry, - MigrateAgentTimeReminder, 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, MigrateMemoryGraph, - MigrateMemoryGraphRecallIncludeImported, MigrateMemoryHebbian, - MigrateMemoryHebbianConsolidation, MigrateMemoryHebbianSpread, MigrateMemoryPersonaConfig, - MigrateMemoryReasoning, MigrateMemoryReasoningJudge, MigrateMemoryRetrieval, - MigrateMemoryRetrievalQueryBias, MigrateMemoryStoreConfig, MigrateMemoryTypeAwareCompose, - MigrateMicrocompactConfig, MigrateNliConfig, MigrateOrchestrationAssetSensitivity, - MigrateOrchestrationCommandConfig, MigrateOrchestrationEnsemble, - MigrateOrchestrationIdleTimeout, MigrateOrchestrationPersistence, + 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, MigrateMemoryGraph, MigrateMemoryGraphRecallIncludeImported, + MigrateMemoryHebbian, MigrateMemoryHebbianConsolidation, MigrateMemoryHebbianSpread, + MigrateMemoryPersonaConfig, MigrateMemoryReasoning, MigrateMemoryReasoningJudge, + MigrateMemoryRetrieval, MigrateMemoryRetrievalQueryBias, MigrateMemoryStoreConfig, + MigrateMemoryTypeAwareCompose, MigrateMicrocompactConfig, MigrateNliConfig, + MigrateOrchestrationAssetSensitivity, MigrateOrchestrationCommandConfig, + MigrateOrchestrationEnsemble, MigrateOrchestrationIdleTimeout, MigrateOrchestrationPersistence, MigrateOrchestrationWholePlanVerifierTimeout, MigrateOrchestratorProvider, MigrateOtelFilter, MigrateOverflowMaxPerCallOverride, MigratePiiFilterNames, MigratePlannerModelToProvider, MigratePluginsReputationConfig, MigratePolicyProviderAndUtilityWindow, @@ -847,6 +849,9 @@ pub static MIGRATIONS: std::sync::LazyLock> Box::new(MigrateIntegrityConfig), // Step 101 — advisory notice for [security.rate_limit] default-on posture (#6469) Box::new(MigrateRateLimitAdvisory), + // Step 102 — insert active delegation_mode = "proactive" into an existing + // [agents] table with enabled = true and no delegation_mode key (#5857) + Box::new(MigrateAgentsDelegationMode), ] }); diff --git a/crates/zeph-config/src/migrate/steps.rs b/crates/zeph-config/src/migrate/steps.rs index 1f50fbd3f..019b49493 100644 --- a/crates/zeph-config/src/migrate/steps.rs +++ b/crates/zeph-config/src/migrate/steps.rs @@ -89,8 +89,8 @@ 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_autodream_config, migrate_caveman_config, - migrate_cocoon_provider_notice, migrate_cocoon_show_balance, + migrate_agent_time_reminder, migrate_agents_delegation_mode, 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, @@ -1299,3 +1299,17 @@ impl Migration for MigrateRateLimitAdvisory { migrate_rate_limit_advisory(toml_src) } } + +/// Step 102 — insert an active `delegation_mode = "proactive"` value into an existing +/// `[agents]` table with `enabled = true` and no `delegation_mode` key (spec +/// `042-subagent-delegation-mode-parity` FR-010, issue #5857). +pub(super) struct MigrateAgentsDelegationMode; +impl Migration for MigrateAgentsDelegationMode { + fn name(&self) -> &'static str { + "migrate_agents_delegation_mode" + } + + fn apply(&self, toml_src: &str) -> Result { + migrate_agents_delegation_mode(toml_src) + } +} diff --git a/crates/zeph-config/src/migrate/subagent.rs b/crates/zeph-config/src/migrate/subagent.rs new file mode 100644 index 000000000..9b5282b1a --- /dev/null +++ b/crates/zeph-config/src/migrate/subagent.rs @@ -0,0 +1,125 @@ +// SPDX-FileCopyrightText: 2026 Andrei G +// SPDX-License-Identifier: MIT OR Apache-2.0 + +//! Sub-agent delegation-mode config migration step. +//! +//! Extracted as its own module (rather than folded into `tools.rs`, which owns the +//! `[agent]` — singular, main-agent — migrations) because `[agents]` (plural, +//! `SubAgentConfig`) is a distinct TOML section. + +use super::{MigrateError, MigrationResult, section_header_present}; + +/// Insert `delegation_mode = "proactive"` under `[agents]` when `enabled = true` and the key +/// is absent (spec `042-subagent-delegation-mode-parity` FR-010, issue #5857). +/// +/// Unlike most migration steps in this module, which surface a *commented-out* example because +/// `#[serde(default)]` already makes the field's absence harmless, this step inserts a real, +/// active value. `DelegationMode::default()` already resolves to `Proactive` on load, so the +/// insertion is behaviorally a no-op — its purpose is discoverability: an operator who already +/// opted into `enabled = true` should see the explicit autonomy setting in their config file +/// rather than have it resolved silently, since narrowing or widening this trust boundary later +/// is a security-relevant change (NFR-001). +/// +/// No-op when `[agents]` is absent, not active (only commented-out), `enabled` is not `true`, +/// or `delegation_mode` is already present. +/// +/// # Errors +/// +/// Returns `MigrateError::Parse` if the TOML cannot be parsed. +pub fn migrate_agents_delegation_mode(toml_src: &str) -> Result { + if toml_src.contains("delegation_mode") || !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("delegation_mode", toml_edit::value("proactive")); + + Ok(MigrationResult { + output: doc.to_string(), + changed_count: 1, + sections_changed: vec!["agents.delegation_mode".to_owned()], + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn inserts_proactive_when_enabled_and_key_absent() { + let src = "[agents]\nenabled = true\nmax_concurrent = 3\n"; + let result = migrate_agents_delegation_mode(src).expect("migrate"); + assert_eq!(result.changed_count, 1); + assert!(result.output.contains("delegation_mode = \"proactive\"")); + } + + #[test] + fn noop_when_disabled() { + let src = "[agents]\nenabled = false\n"; + let result = migrate_agents_delegation_mode(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert!(!result.output.contains("delegation_mode")); + } + + #[test] + fn noop_when_key_already_present() { + let src = "[agents]\nenabled = true\ndelegation_mode = \"explicit_request_only\"\n"; + let result = migrate_agents_delegation_mode(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert_eq!(result.output, src); + } + + #[test] + fn noop_when_section_absent() { + let src = "[llm]\nprovider = \"claude\"\n"; + let result = migrate_agents_delegation_mode(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert_eq!(result.output, src); + } + + #[test] + fn noop_when_section_only_commented_out() { + let src = "# [agents]\n# enabled = true\n"; + let result = migrate_agents_delegation_mode(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert_eq!(result.output, src); + } + + #[test] + fn idempotent() { + let src = "[agents]\nenabled = true\n"; + let once = migrate_agents_delegation_mode(src).expect("migrate"); + let twice = migrate_agents_delegation_mode(&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 99500fa0b..6af79264d 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(), - 101, - "MIGRATIONS registry must contain all 101 sequential steps" + 102, + "MIGRATIONS registry must contain all 102 sequential steps" ); for m in MIGRATIONS.iter() { assert!( @@ -2093,7 +2093,7 @@ fn migrate_focus_auto_consolidate_noop_when_only_commented_section() { #[test] fn registry_has_fifty_entries() { - assert_eq!(MIGRATIONS.len(), 101); + assert_eq!(MIGRATIONS.len(), 102); } #[test] @@ -2236,6 +2236,7 @@ fn registry_preserves_order_matches_dispatch() { "migrate_durable_hwm_advisory", "migrate_integrity_config", "migrate_rate_limit_advisory", + "migrate_agents_delegation_mode", ]; 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 551fcca7b..6838f98a2 100644 --- a/crates/zeph-core/config/default.toml +++ b/crates/zeph-core/config/default.toml @@ -704,6 +704,10 @@ request_timeout_secs = 10 [agents] # Enable sub-agent spawning (required for /agent commands and multi-agent workflows) enabled = false +# Whether the main agent may spawn sub-agents, and who may trigger it: "disabled" / +# "explicit_request_only" / "proactive". Orthogonal to `enabled` above, which remains the +# outer kill switch. Env override: ZEPH_AGENTS_DELEGATION_MODE. +delegation_mode = "proactive" # Maximum number of sub-agents that can run concurrently max_concurrent = 1 # Allow sub-agents to use bypass_permissions mode (enable only in trusted environments) diff --git a/crates/zeph-core/src/agent/scheduler_loop.rs b/crates/zeph-core/src/agent/scheduler_loop.rs index 28fc52ae7..ce7e55379 100644 --- a/crates/zeph-core/src/agent/scheduler_loop.rs +++ b/crates/zeph-core/src/agent/scheduler_loop.rs @@ -536,6 +536,10 @@ impl Agent { let mut spawn_ctx = self.build_spawn_context(&cfg); spawn_ctx.network_denied = network_denied; spawn_ctx.inherited_tool_allowlist = task_allowlist; + // Autonomous DAG dispatch (spec 042, #5857): `build_spawn_context` sets `Explicit` as + // its base value (correct for its other, user-command-driven callers) — this is the + // one caller that isn't user-driven, so override it back to `Autonomous` here. + spawn_ctx.origin = zeph_subagent::SpawnOrigin::Autonomous; // Idle-timeout progress heartbeat (issue #6245, Alt-A): the driver owns creation of // the Arc. One clone flows into the sub-agent loop via `spawn_ctx.progress_at` diff --git a/crates/zeph-core/src/agent/slash_commands.rs b/crates/zeph-core/src/agent/slash_commands.rs index ea1687b7c..8446ea5ff 100644 --- a/crates/zeph-core/src/agent/slash_commands.rs +++ b/crates/zeph-core/src/agent/slash_commands.rs @@ -97,6 +97,18 @@ impl Agent { /// Routes `/subagent spawn ` through the ACP spawn callback when available. /// Returns a usage hint when no sub-command or command string is given, and a /// "not available" message when the ACP spawn callback has not been injected. + /// + /// This path launches an external ACP subagent process (`zeph_acp::run_session` via + /// `spawn_fn`, wired in `src/runner.rs`) and never touches `SubAgentManager` or + /// `SpawnContext` — the `delegation_mode` gate inside `SubAgentManager::spawn` does not + /// see it at all. Spec 042 FR-003 requires `disabled` mode to reject *every* spawn path, + /// so the effective-mode check below is an explicit, separate gate at this choke point + /// (issue #5857). Uses `DelegationMode::permits_explicit()` — the same allow-list predicate + /// `SubAgentManager::resume` uses — rather than a hand-written `== Disabled` deny-list, so + /// the two enforcement points cannot drift apart and neither fails open on a future + /// `#[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`). 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() @@ -105,9 +117,18 @@ impl Agent { match subcmd { "spawn" => { let cmd = rest.trim(); + let effective_mode = self.effective_delegation_mode(); if cmd.is_empty() { "Usage: /subagent spawn \n\nExample: /subagent spawn zeph --acp" .to_owned() + } else if !effective_mode.permits_explicit() { + tracing::warn!( + mode = ?effective_mode, + "/subagent spawn rejected: delegation disabled by configuration" + ); + "Sub-agent delegation is disabled by configuration \ + ([agents].delegation_mode = \"disabled\" or [agents].enabled = false)." + .to_owned() } else if let Some(spawn_fn) = self.runtime.config.acp_subagent_spawn_fn.clone() { let cmd = cmd.to_owned(); diff --git a/crates/zeph-core/src/agent/subagent_commands.rs b/crates/zeph-core/src/agent/subagent_commands.rs index ef7da4b2e..a079cb5a9 100644 --- a/crates/zeph-core/src/agent/subagent_commands.rs +++ b/crates/zeph-core/src/agent/subagent_commands.rs @@ -245,11 +245,19 @@ impl Agent { fn handle_agent_list(&self) -> Option { use std::fmt::Write as _; let mgr = self.services.orchestration.subagent_manager.as_ref()?; + let mode_label = match mgr.delegation_mode() { + zeph_config::DelegationMode::Disabled => "disabled", + zeph_config::DelegationMode::ExplicitRequestOnly => "explicit_request_only", + zeph_config::DelegationMode::Proactive => "proactive", + _ => "unknown", + }; let defs = mgr.definitions(); if defs.is_empty() { - return Some("No sub-agent definitions found.".into()); + return Some(format!( + "Delegation mode: {mode_label}\nNo sub-agent definitions found." + )); } - let mut out = String::from("Available sub-agents:\n"); + let mut out = format!("Delegation mode: {mode_label}\nAvailable sub-agents:\n"); for d in defs { let memory_label = match d.memory { Some(zeph_subagent::MemoryScope::User) => " [memory:user]", @@ -787,6 +795,20 @@ impl Agent { Some(bodies) } + /// The effective delegation mode currently in force (spec 042, issue #5857). + /// + /// Reads directly from `subagent_config` (always present, independent of whether a + /// `SubAgentManager` happens to be constructed) via + /// [`zeph_config::SubAgentConfig::effective_delegation_mode`], which folds in the + /// `enabled` outer kill switch. This is the same fold `src/runner.rs` bootstrap applies + /// before calling `SubAgentManager::set_delegation_mode` — reading it here independently + /// keeps this choke point correct even where no manager is wired up (e.g. a test harness). + pub(super) fn effective_delegation_mode(&self) -> zeph_config::DelegationMode { + self.services + .orchestration + .subagent_config + .effective_delegation_mode() + } /// Build a `SpawnContext` from current agent state for sub-agent spawning. pub(super) fn build_spawn_context( &self, @@ -838,6 +860,14 @@ impl Agent { // narrow against (see issue tracking the follow-up per-task `TaskNode` allowlist // plumbing for the orchestration layer). max_trust_level: Some(self.parent_effective_trust_level()), + // This helper's own three callers (`handle_agent_background`, + // `handle_agent_spawn_foreground`, `handle_agent_resume`) are all dispatched from + // the explicit `/agent spawn`/`/agent resume` slash command, so `Explicit` is the + // correct base value here (spec 042, issue #5857). `handle_scheduler_spawn_action` + // is the sole caller that needs `Autonomous` — it overrides `spawn_ctx.origin` + // immediately after calling this helper, mirroring how it already overrides + // `network_denied`/`progress_at` post-construction. + origin: zeph_subagent::SpawnOrigin::Explicit, ..Default::default() } } 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 b605ce359..b8319bb55 100644 --- a/crates/zeph-core/src/agent/tests/small_misc_tests.rs +++ b/crates/zeph-core/src/agent/tests/small_misc_tests.rs @@ -80,6 +80,11 @@ async fn subagent_spawn_no_command_returns_usage() { #[tokio::test] async fn subagent_spawn_without_callback_returns_not_available() { let mut h = QuickTestAgent::minimal(""); + // Delegation gate (spec 042, issue #5857): a fresh default config has `agents.enabled = + // false`, which now resolves to `Disabled` and would short-circuit before ever reaching + // the "callback missing" branch this test exercises. Opt in so the pre-existing behavior + // under test is reachable. + h.agent.services.orchestration.subagent_config.enabled = true; let result = h .agent .dispatch_slash_command("/subagent spawn cargo run -- --acp") @@ -95,6 +100,7 @@ async fn subagent_spawn_without_callback_returns_not_available() { #[tokio::test] async fn subagent_spawn_with_callback_returns_output() { 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 { Ok(format!("spawned: {cmd}")) }) })); @@ -110,6 +116,60 @@ async fn subagent_spawn_with_callback_returns_output() { ); } +/// Spec 042 FR-003 (issue #5857): `delegation_mode = "disabled"` (or, as here, the `enabled` +/// outer kill switch left at its default `false`) must reject the ACP `/subagent spawn` path +/// too — even though it never touches `SubAgentManager`/`SpawnContext` at all (critic's +/// corrected finding; see `handle_subagent_slash`'s doc comment). A configured callback must +/// never be invoked while disabled. +#[tokio::test] +async fn subagent_spawn_disabled_by_delegation_mode_returns_disabled_message() { + let mut h = QuickTestAgent::minimal(""); + // Default config: `agents.enabled = false` → effective mode `Disabled`. + 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(), "must be intercepted"); + let output = h.sent_messages().join("\n"); + assert!( + output.to_lowercase().contains("disabled"), + "expected a disabled-by-configuration message, got: {output}" + ); + assert!( + !output.contains("spawned: my-command"), + "callback must never run while delegation is disabled, got: {output}" + ); +} + +/// `delegation_mode = "explicit_request_only"` with `enabled = true` must still permit +/// `/subagent spawn` — it is itself an explicit user action (spec 042, issue #5857). +#[tokio::test] +async fn subagent_spawn_explicit_request_only_still_allows_acp_spawn() { + let mut h = QuickTestAgent::minimal(""); + h.agent.services.orchestration.subagent_config.enabled = true; + h.agent + .services + .orchestration + .subagent_config + .delegation_mode = zeph_config::DelegationMode::ExplicitRequestOnly; + 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(), "must be intercepted"); + let output = h.sent_messages().join("\n"); + assert!( + output.contains("spawned: my-command"), + "expected callback output under explicit_request_only, got: {output}" + ); +} + #[tokio::test] async fn subagent_unknown_subcommand_returns_error() { let mut h = QuickTestAgent::minimal(""); diff --git a/crates/zeph-subagent/Cargo.toml b/crates/zeph-subagent/Cargo.toml index 9664bf320..cf9b51c16 100644 --- a/crates/zeph-subagent/Cargo.toml +++ b/crates/zeph-subagent/Cargo.toml @@ -26,6 +26,7 @@ serde_norway.workspace = true tempfile.workspace = true thiserror.workspace = true tokio = { workspace = true, features = ["fs", "macros", "process", "rt-multi-thread", "sync", "time"] } +tokio-stream.workspace = true tokio-util.workspace = true toml.workspace = true tracing.workspace = true diff --git a/crates/zeph-subagent/README.md b/crates/zeph-subagent/README.md index 1bdda07e6..056929528 100644 --- a/crates/zeph-subagent/README.md +++ b/crates/zeph-subagent/README.md @@ -144,6 +144,22 @@ forward_transcript = false # default: false; also settable via --forward-subag Forwarding is structurally non-blocking on the sub-agent's own turn loop: `agent_loop.rs` does a non-blocking `try_send` of a `RawChunk` into a bounded per-task `mpsc` (capacity 128, tail-drop on full), and a manager-owned drain performs the one sanitize step before dispatching to whichever consumer surfaces (`ForwardSurfaces`) are active for the session. +## Delegation mode + +Tri-state control over whether the main agent may spawn sub-agents, and who may trigger it (spec `042-subagent-delegation-mode-parity`, issue #5857): + +```toml +[agents] +enabled = true # outer kill switch: false always resolves to "disabled" below +delegation_mode = "proactive" # "disabled" | "explicit_request_only" | "proactive" (default) +``` + +- `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. + +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`. + ## Features | Feature | Description | diff --git a/crates/zeph-subagent/src/agent_loop.rs b/crates/zeph-subagent/src/agent_loop.rs index c0a9a05dd..2ba6da85b 100644 --- a/crates/zeph-subagent/src/agent_loop.rs +++ b/crates/zeph-subagent/src/agent_loop.rs @@ -7,12 +7,14 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Instant; use tokio::sync::{mpsc, watch}; +use tokio_stream::StreamExt; use tokio_util::sync::CancellationToken; use zeph_llm::any::AnyProvider; use zeph_llm::provider::{ ChatResponse, LlmProvider, Message, MessageMetadata, MessagePart, Role, ThinkingBlock, - ToolDefinition, + ToolDefinition, ToolUseRequest, }; +use zeph_llm::sse::{ToolSseEvent, ToolSseStream}; use zeph_sanitizer::{ContentSanitizer, ContentSource, ContentSourceKind}; use zeph_tools::executor::{ErasedToolExecutor, ToolCall}; @@ -85,11 +87,13 @@ pub(super) struct AgentLoopArgs { /// Cross-crate debug-dump sink, threaded down from `SpawnContext::debug_dump_sink` /// (issue #6391). `None` when debug dumps are disabled. pub(super) debug_dump_sink: Option>, - /// Live transcript forwarding sender (issue #6359). `None` when forwarding is disabled, - /// no consumer surface is active, or no `TaskSupervisor` is wired — the `if let Some(f)` - /// gate at every call site is then a genuine no-op (FR-007): no allocation, no clone, - /// nothing sent. Owned exclusively by this run's own turn loop for its lifetime; never - /// clone the inner sender into a longer-lived struct (see `forward::ForwardSender` docs). + /// Live transcript forwarding sender (issue #6359 FR-002a per-turn; issue #6456 FR-002b + /// token-level intra-turn deltas on providers with native tool-streaming support). `None` + /// when forwarding is disabled, no consumer surface is active, or no `TaskSupervisor` is + /// wired — the `if let Some(f)` gate at every call site is then a genuine no-op (FR-007): + /// no allocation, no clone, nothing sent. Owned exclusively by this run's own turn loop + /// for its lifetime; never clone the inner sender into a longer-lived struct (see + /// `forward::ForwardSender` docs). pub(super) forward: Option, /// Shared secret-mask registry (issue #6492), the same `Arc` used for the parent's /// outbound-LLM masking and the forwarding drain's `SanitizeLayers`. Applied to every @@ -174,6 +178,122 @@ fn build_effective_system_prompt( effective } +/// Drive a `ToolSseStream` from a provider's native streaming-with-tools path (issue #6456, +/// FR-002b), forwarding each text/thinking delta through `forward` as it arrives and +/// assembling the same [`ChatResponse`] shape [`LlmProvider::chat_with_tools`] would return. +/// +/// Forwarded deltas are display-only (see `forward.rs` module docs' "Design contract" +/// section): `text_buf`/`thinking_blocks`/`tool_calls` are accumulated locally regardless of +/// whether any individual delta actually made it onto the (tail-drop) forward channel, so a +/// dropped delta only ever produces a display gap — the `ChatResponse` this function returns +/// is unaffected and remains the sole input to the next turn's LLM context. +/// +/// Mirrors `zeph-core`'s `SpeculativeStreamDrainer::drive` accumulation logic (same +/// `ToolSseStream`/`ToolSseEvent` contract), minus speculative dispatch, which is out of +/// scope for the subagent loop. +/// +/// # Errors +/// +/// Returns the first `LlmError` reported by the stream. +#[tracing::instrument(name = "subagent.agent_loop.stream_turn", skip_all)] +async fn drive_tool_stream( + mut stream: ToolSseStream, + forward: &ForwardSender, +) -> Result { + let mut tool_calls: Vec = Vec::new(); + let mut thinking_blocks: Vec = Vec::new(); + let mut text_buf = String::new(); + + while let Some(event) = stream.next().await { + match event { + ToolSseEvent::ContentChunk(text) => { + forward.send_text(&text); + text_buf.push_str(&text); + } + ToolSseEvent::ThinkingChunk(text) => { + forward.send_thinking(&text); + } + ToolSseEvent::ThinkingBlockDone(block) => { + thinking_blocks.push(block); + } + ToolSseEvent::ToolCallComplete { + id, + name, + full_json, + .. + } => { + let input = serde_json::from_str(&full_json) + .unwrap_or(serde_json::Value::Object(serde_json::Map::new())); + tool_calls.push(ToolUseRequest { + id, + name: name.into(), + input, + }); + } + ToolSseEvent::Error(e) => return Err(e), + // `ToolBlockStart`/`InputJsonDelta` only matter for the speculative-dispatch + // drainer (tool metadata ahead of the final JSON, used for early dispatch); + // `Compaction` summaries are a known TODO (matches `SpeculativeStreamDrainer:: + // drive`'s same limitation — not yet surfaced to the caller). Any future + // `#[non_exhaustive]` variant is likewise a no-op here rather than a compile + // error on an upstream addition. + _ => {} + } + } + + let text = if text_buf.is_empty() { + None + } else { + Some(text_buf) + }; + if tool_calls.is_empty() { + Ok(ChatResponse::Text(text.unwrap_or_default())) + } else { + Ok(ChatResponse::ToolUse { + text, + tool_calls, + thinking_blocks, + }) + } +} + +/// Attempt the streaming-with-tools path when forwarding is active, falling back to the +/// plain [`LlmProvider::chat_with_tools`] call otherwise or on any streaming-setup error +/// (issue #6456, FR-002b). +/// +/// Returns `(response, streamed)`; `streamed == true` means `forward` already received this +/// turn's text/thinking as deltas via [`drive_tool_stream`], so the caller (`run_turn`) must +/// not forward the same content again as a whole-turn chunk. +/// +/// Providers without a native tool-streaming implementation (every backend except Claude, +/// see `AnyProvider::chat_with_tools_stream`) return `Err(LlmError::Unavailable)` immediately +/// — no network round-trip is attempted before falling back, so this costs nothing beyond a +/// synchronous match for non-Claude providers, and per-turn forwarding behaves exactly as it +/// did before this issue (FR-002a, unchanged). +async fn call_provider_streaming_or_plain( + provider: &AnyProvider, + messages: &[Message], + tool_defs: &[ToolDefinition], + forward: Option<&ForwardSender>, +) -> Result<(ChatResponse, bool), zeph_llm::LlmError> { + if let Some(f) = forward { + match provider.chat_with_tools_stream(messages, tool_defs).await { + Ok(stream) => return drive_tool_stream(stream, f).await.map(|r| (r, true)), + Err(e) => { + tracing::debug!( + error = %e, + "provider has no native tool-streaming support (or the streaming request \ + failed); falling back to non-streaming chat_with_tools" + ); + } + } + } + provider + .chat_with_tools(messages, tool_defs) + .await + .map(|r| (r, false)) +} + #[tracing::instrument(name = "subagent.agent_loop.call_provider", skip_all, err)] #[allow(clippy::too_many_arguments)] async fn call_provider_with_status( @@ -186,7 +306,7 @@ async fn call_provider_with_status( llm_timeout: std::time::Duration, debug_dump_sink: Option<&dyn zeph_llm::debug_dump::DebugDumpSink>, forward: Option<&ForwardSender>, -) -> Result { +) -> Result<(ChatResponse, bool), super::error::SubAgentError> { // Mirrors `zeph-core`'s `prepare_chat_debug_dump`/`write_chat_debug_dump` pair so // sub-agent LLM calls are captured through the same `--debug-dump` pipeline as the // top-level agent loop (#6391). `None` when debug dumps are disabled. @@ -199,36 +319,38 @@ async fn call_provider_with_status( sink.dump_request(provider.name(), messages, tool_defs, provider_request) }); - let llm_result = - tokio::time::timeout(llm_timeout, provider.chat_with_tools(messages, tool_defs)) - .await - .map_err(|_| { - tracing::warn!( - timeout_secs = llm_timeout.as_secs(), - "sub-agent LLM call timed out" - ); - let timeout_err = super::error::SubAgentError::Llm("LLM call timed out".to_owned()); - // Without this, status_tx stays frozen at its last `Working` value forever — - // the TUI sidebar and `collect_finished_subagents()` never see a terminal - // state, so the handle is never reaped (#6381, same defect class as #6257's - // setup-phase fix). - let _ = status_tx.send(SubAgentStatus { - state: SubAgentState::Failed, - last_message: Some(timeout_err.to_string()), - turns_used: turns, - started_at, - }); - if let Some(f) = forward { - f.send_terminal(SubAgentState::Failed); - } - timeout_err - })?; + let llm_result = tokio::time::timeout( + llm_timeout, + call_provider_streaming_or_plain(provider, messages, tool_defs, forward), + ) + .await + .map_err(|_| { + tracing::warn!( + timeout_secs = llm_timeout.as_secs(), + "sub-agent LLM call timed out" + ); + let timeout_err = super::error::SubAgentError::Llm("LLM call timed out".to_owned()); + // Without this, status_tx stays frozen at its last `Working` value forever — + // the TUI sidebar and `collect_finished_subagents()` never see a terminal + // state, so the handle is never reaped (#6381, same defect class as #6257's + // setup-phase fix). + let _ = status_tx.send(SubAgentStatus { + state: SubAgentState::Failed, + last_message: Some(timeout_err.to_string()), + turns_used: turns, + started_at, + }); + if let Some(f) = forward { + f.send_terminal(SubAgentState::Failed); + } + timeout_err + })?; match llm_result { - Ok(r) => { + Ok((r, streamed)) => { if let (Some(sink), Some(id)) = (debug_dump_sink, dump_id) { sink.dump_response(id, &r); } - Ok(r) + Ok((r, streamed)) } Err(e) => { tracing::error!(error = %e, "sub-agent LLM call failed"); @@ -517,7 +639,7 @@ async fn run_turn( forward: Option<&ForwardSender>, secret_registry: Option<&zeph_sanitizer::secret_mask::SecretMaskRegistry>, ) -> Result { - let response = call_provider_with_status( + let (response, streamed) = call_provider_with_status( provider, messages, tool_defs, @@ -540,11 +662,15 @@ async fn run_turn( last_result.clone_from(&response_text); emit_working_status(status_tx, &response_text, *turns, started_at); - // FR-002a/FR-007: forward the turn's full text + any visible thinking blocks the - // instant this turn's response arrives. The `if let Some(f)` gate wraps the thinking - // extraction itself, not just the send — zero allocation when forwarding is inactive - // (critic M2). Must run before `response` is moved into `handle_tool_step` below. - if let Some(f) = forward { + // FR-002a/FR-002b/FR-007: forward this turn's text + any visible thinking blocks the + // instant they are available. When `streamed` is true, `call_provider_with_status` already + // forwarded every delta incrementally via `drive_tool_stream` while the turn was still in + // flight — forwarding the same content again here as one whole-turn chunk would just + // double it on the display surfaces, so this block is skipped in that case. The `if let + // Some(f)` gate wraps the thinking extraction itself, not just the send — zero allocation + // when forwarding is inactive (critic M2). Must run before `response` is moved into + // `handle_tool_step` below. + if !streamed && let Some(f) = forward { f.send_text(&response_text); if let ChatResponse::ToolUse { thinking_blocks, .. @@ -1745,3 +1871,206 @@ mod build_effective_system_prompt_tests { ); } } + +// ── #6456: token-level intra-turn transcript streaming (FR-002b) ────────── +#[cfg(test)] +mod token_streaming_tests { + use zeph_llm::LlmError; + use zeph_llm::mock::MockProvider; + + use super::*; + use crate::forward::new_channel; + + fn tool_stream_from(events: Vec) -> ToolSseStream { + Box::pin(tokio_stream::iter(events)) + } + + #[tokio::test] + async fn drive_tool_stream_assembles_text_and_forwards_every_delta() { + let (sender, mut rx) = new_channel(Arc::from("task-1"), Arc::from("agent-1")); + let stream = tool_stream_from(vec![ + ToolSseEvent::ContentChunk("Hello".into()), + ToolSseEvent::ContentChunk(", world".into()), + ToolSseEvent::ContentChunk("!".into()), + ]); + + let response = drive_tool_stream(stream, &sender) + .await + .expect("stream must assemble successfully"); + + match response { + ChatResponse::Text(t) => assert_eq!(t, "Hello, world!"), + other => panic!("expected ChatResponse::Text, got {other:?}"), + } + + // Every content delta must have been forwarded individually (not just the final + // accumulated text) — this is the actual FR-002b behavior under test. + let mut received = Vec::new(); + while let Ok(chunk) = rx.try_recv() { + received.push(format!("{chunk:?}")); + } + assert_eq!(received.len(), 3, "each delta must be forwarded separately"); + assert!(received[0].contains("Hello")); + assert!(received[1].contains(", world")); + assert!(received[2].contains('!')); + } + + #[tokio::test] + async fn drive_tool_stream_forwards_thinking_deltas_and_keeps_final_block() { + let (sender, mut rx) = new_channel(Arc::from("task-2"), Arc::from("agent-2")); + let stream = tool_stream_from(vec![ + ToolSseEvent::ThinkingChunk("step one".into()), + ToolSseEvent::ThinkingChunk(" step two".into()), + ToolSseEvent::ThinkingBlockDone(ThinkingBlock::Thinking { + thinking: "step one step two".into(), + signature: "sig".into(), + }), + ]); + + let response = drive_tool_stream(stream, &sender) + .await + .expect("stream must assemble successfully"); + + match response { + ChatResponse::Text(t) => assert!(t.is_empty(), "no content chunks were sent"), + other => panic!("expected empty ChatResponse::Text, got {other:?}"), + } + + let mut forwarded_thinking = 0; + while let Ok(chunk) = rx.try_recv() { + if format!("{chunk:?}").contains("Thinking") { + forwarded_thinking += 1; + } + } + assert_eq!( + forwarded_thinking, 2, + "both thinking deltas must be forwarded incrementally" + ); + } + + #[tokio::test] + async fn drive_tool_stream_assembles_tool_use_from_complete_events() { + let (sender, _rx) = new_channel(Arc::from("task-3"), Arc::from("agent-3")); + let stream = tool_stream_from(vec![ + ToolSseEvent::ToolBlockStart { + index: 0, + id: "call-1".into(), + name: "shell".into(), + }, + ToolSseEvent::InputJsonDelta { + index: 0, + delta: r#"{"cmd":"ls"}"#.into(), + }, + ToolSseEvent::ToolCallComplete { + index: 0, + id: "call-1".into(), + name: "shell".into(), + full_json: r#"{"cmd":"ls"}"#.into(), + }, + ]); + + let response = drive_tool_stream(stream, &sender) + .await + .expect("stream must assemble successfully"); + + match response { + ChatResponse::ToolUse { tool_calls, .. } => { + assert_eq!(tool_calls.len(), 1); + assert_eq!(tool_calls[0].name.as_str(), "shell"); + assert_eq!(tool_calls[0].input["cmd"], "ls"); + } + other => panic!("expected ChatResponse::ToolUse, got {other:?}"), + } + } + + #[tokio::test] + async fn drive_tool_stream_propagates_stream_error() { + let (sender, _rx) = new_channel(Arc::from("task-4"), Arc::from("agent-4")); + let stream = tool_stream_from(vec![ + ToolSseEvent::ContentChunk("partial".into()), + ToolSseEvent::Error(LlmError::Unavailable), + ]); + + let result = drive_tool_stream(stream, &sender).await; + assert!(result.is_err(), "a stream Error event must propagate"); + } + + /// Critic concern (c): the guaranteed source of truth is the accumulated `ChatResponse`, + /// never the forwarded deltas. Send far more chunks than the forward channel's capacity + /// without ever draining the receiver, so every `try_send` past capacity silently + /// tail-drops — and confirm the returned `ChatResponse` is still complete regardless. + #[tokio::test] + async fn drive_tool_stream_accumulation_survives_forward_channel_backpressure() { + let (sender, _rx) = new_channel(Arc::from("task-5"), Arc::from("agent-5")); + let mut events = Vec::new(); + let mut expected = String::new(); + for i in 0..300 { + let piece = format!("chunk{i} "); + expected.push_str(&piece); + events.push(ToolSseEvent::ContentChunk(piece)); + } + let stream = tool_stream_from(events); + + let response = drive_tool_stream(stream, &sender) + .await + .expect("stream must assemble successfully even under channel backpressure"); + + match response { + ChatResponse::Text(t) => assert_eq!( + t, expected, + "accumulated response text must be complete even though most forwarded \ + deltas were tail-dropped (never drained)" + ), + other => panic!("expected ChatResponse::Text, got {other:?}"), + } + } + + #[tokio::test] + async fn call_provider_streaming_or_plain_falls_back_for_non_streaming_provider() { + // MockProvider has no `AnyProvider::chat_with_tools_stream` branch — it hits the + // `_ => Err(Unavailable)` arm — so this must fall back to `chat_with_tools` + // unchanged, exactly like the pre-#6456 behavior (FR-002a). + let (mock, _counter) = + MockProvider::default().with_tool_use(vec![ChatResponse::Text("plain reply".into())]); + let provider = AnyProvider::Mock(mock); + let (sender, mut rx) = new_channel(Arc::from("task-6"), Arc::from("agent-6")); + let messages = vec![make_message(Role::User, "hi".into())]; + + let (response, streamed) = + call_provider_streaming_or_plain(&provider, &messages, &[], Some(&sender)) + .await + .expect("fallback call must succeed"); + + assert!( + !streamed, + "a non-streaming provider must report streamed = false" + ); + match response { + ChatResponse::Text(t) => assert_eq!(t, "plain reply"), + other => panic!("expected ChatResponse::Text, got {other:?}"), + } + // No deltas were ever streamed for this provider — the forward channel must stay + // empty here; `run_turn` is responsible for the single whole-turn forward in this + // case (unchanged FR-002a call site, tested separately at the loop level). + assert!(rx.try_recv().is_err(), "no deltas forwarded on fallback"); + } + + #[tokio::test] + async fn call_provider_streaming_or_plain_skips_streaming_attempt_when_forward_is_none() { + let (mock, _counter) = + MockProvider::default().with_tool_use(vec![ChatResponse::Text("no forward".into())]); + let provider = AnyProvider::Mock(mock); + let messages = vec![make_message(Role::User, "hi".into())]; + + let (response, streamed) = + call_provider_streaming_or_plain(&provider, &messages, &[], None) + .await + .expect("call must succeed with forwarding disabled"); + + assert!(!streamed); + match response { + ChatResponse::Text(t) => assert_eq!(t, "no forward"), + other => panic!("expected ChatResponse::Text, got {other:?}"), + } + } +} diff --git a/crates/zeph-subagent/src/error.rs b/crates/zeph-subagent/src/error.rs index 1b8ffed7b..b6f944c0a 100644 --- a/crates/zeph-subagent/src/error.rs +++ b/crates/zeph-subagent/src/error.rs @@ -106,4 +106,20 @@ pub enum SubAgentError { /// only called when `durable.enabled && durable.subagent`). #[error("durable error: {0}")] Durable(String), + + /// The spawn attempt was rejected by `delegation_mode` (spec + /// `042-subagent-delegation-mode-parity`, issue #5857): either `delegation_mode = + /// "disabled"` (all spawns rejected) or `delegation_mode = "explicit_request_only"` and + /// `origin` was [`SpawnOrigin::Autonomous`](crate::manager::SpawnOrigin). Distinct from + /// [`SubAgentError::ConcurrencyLimit`] and [`SubAgentError::MaxDepthExceeded`] so the + /// rejection reason is unambiguous in logs (FR-007). + #[error( + "delegation denied: mode={mode:?} origin={origin:?} agent='{def_name}' \ + (see [agents].delegation_mode / [agents].enabled in config.toml)" + )] + DelegationDenied { + mode: zeph_config::DelegationMode, + origin: crate::manager::SpawnOrigin, + def_name: String, + }, } diff --git a/crates/zeph-subagent/src/forward.rs b/crates/zeph-subagent/src/forward.rs index 8317e37dd..9b3a5dba5 100644 --- a/crates/zeph-subagent/src/forward.rs +++ b/crates/zeph-subagent/src/forward.rs @@ -1,10 +1,16 @@ // SPDX-FileCopyrightText: 2026 Andrei G // SPDX-License-Identifier: MIT OR Apache-2.0 -//! Live subagent transcript forwarding (issue #6359, spec `068-subagent-transcript-forward`). +//! Live subagent transcript forwarding (issue #6359, spec `068-subagent-transcript-forward`; +//! token-level intra-turn streaming, issue #6456, FR-002b). //! -//! Opt-in, per-turn forwarding of a running subagent's full text/thinking output to the -//! TUI runtime detail view and/or a `--bare` stdout sink. Pipeline shape: +//! Opt-in forwarding of a running subagent's text/thinking output to the TUI runtime detail +//! view and/or a `--bare` stdout sink, under the single `forward_transcript` config flag. +//! Granularity depends on provider support: when the provider's native streaming-with-tools +//! path is available (`agent_loop.rs` drives it), text/thinking chunks are forwarded as +//! partial deltas *within* a turn; otherwise (or when streaming fails) the full, untruncated +//! text/thinking output of one completed LLM turn is forwarded once the turn completes +//! (FR-002a, unchanged). Pipeline shape: //! //! ```text //! agent_loop.rs (sync, non-blocking) --try_send(RawChunk)--> per-task mpsc (cap 128) @@ -14,6 +20,20 @@ //! `RawChunk` only ever travels on the ingress channel; `SanitizedChunk` is constructed //! exclusively by the drain's sanitize step and is the only type any sink can receive //! (NFR-005 enforced structurally, not by convention). +//! +//! # Design contract: deltas are ephemeral, display-only (FR-002b) +//! +//! Every chunk sent through `ForwardSender::send_text` / `ForwardSender::send_thinking` — +//! whether it carries a whole turn's text or one streamed delta — travels on the same +//! tail-drop `mpsc` and MUST be treated as **display-only**. A dropped chunk is a display +//! gap, never a correctness error: the loop's own accumulated response text (returned from +//! `run_agent_loop`'s LLM call and pushed into `messages`) is assembled independently of +//! whether any given delta was actually forwarded, and the guaranteed terminal chunk (see +//! `ForwardSender::send_terminal`) marks the one point a consumer may treat as authoritative +//! for "this run reached a terminal state". No consumer (TUI ring buffer, `--bare` sink, a +//! future sink) may reconstruct the subagent's conversational state — let alone feed it back +//! into the parent's LLM context — by concatenating forwarded chunks; deltas never enter any +//! LLM context, they exist purely for live human-facing display. use std::collections::{HashMap, VecDeque}; use std::sync::Arc; @@ -65,9 +85,6 @@ impl ForwardSurfaces { /// Only ever travels on the per-task ingress `mpsc` — never exposed outside this module. #[derive(Debug, Clone)] pub(crate) struct RawChunk { - task_id: Arc, - def_name: Arc, - seq: u64, kind: ForwardChunkKind, } @@ -145,24 +162,129 @@ fn sanitize_text(raw_text: &str, def_name: &str, layers: &SanitizeLayers) -> Str body } -fn sanitize_chunk(raw: RawChunk, layers: &SanitizeLayers) -> SanitizedChunk { - let kind = match raw.kind { - ForwardChunkKind::Text(text) => { - SanitizedChunkKind::Text(sanitize_text(&text, raw.def_name.as_ref(), layers)) - } - ForwardChunkKind::Thinking(text) => { - SanitizedChunkKind::Thinking(sanitize_text(&text, raw.def_name.as_ref(), layers)) - } - ForwardChunkKind::Terminal(state) => SanitizedChunkKind::Terminal(state), - }; +/// Bounded lookback window (bytes) held back from the tail of a pending `Text`/`Thinking` +/// buffer before sanitizing and emitting its safe prefix (review Critical Issue #2, #6456 +/// follow-up). +/// +/// Without this, each streamed delta (FR-002b) was sanitized in complete isolation — a +/// secret or PII pattern split across two `ToolSseEvent` chunk boundaries matched neither +/// fragment individually and reached `--bare` stdout / the TUI ring buffer unmasked. Holding +/// back this many trailing bytes on every partial flush guarantees any pattern whose two +/// halves arrive within this window of each other is always sanitized as one contiguous +/// string before being released. +/// +/// Chosen generously above [`crate::grants::GrantedSecret`]-delivered or vault-registered +/// secret lengths seen in practice and every PII pattern in `zeph_sanitizer::pii` (email/ +/// phone/SSN/credit-card are all well under 80 bytes). A secret whose split fragments are +/// separated by *more* than this many bytes of other already-flushed content is a residual +/// limitation inherent to any bounded-window approach — not eliminated, only made +/// practically unreachable for realistic secret/PII lengths. +const SANITIZE_HOLDBACK_BYTES: usize = 256; + +/// Per-task raw text accumulated but not yet sanitized/emitted (review Critical Issue #2). +/// +/// Kept separate for the `Text` and `Thinking` streams since they are independent logical +/// channels that must never be concatenated with each other. +#[derive(Default)] +struct PendingSanitizeBuffers { + text: String, + thinking: String, +} + +/// Split off `buf`'s sanitizable prefix, leaving the last `holdback` bytes (rounded down to +/// the nearest UTF-8 char boundary, same class of problem as UTF-8 chunk-boundary handling) +/// in place for a future call to potentially combine with. Pass `holdback = 0` to flush the +/// entire remaining buffer — used once no more data for this task is coming (an explicit +/// `Terminal` chunk or the hard-abort backstop), so buffered content is only ever delayed, +/// never silently dropped. Returns `None` when there is nothing new to emit yet. +fn split_off_safe_prefix(buf: &mut String, holdback: usize) -> Option { + if buf.is_empty() { + return None; + } + let target = buf.len().saturating_sub(holdback); + let boundary = buf.floor_char_boundary(target); + if boundary == 0 { + return None; + } + let prefix = buf[..boundary].to_owned(); + buf.drain(..boundary); + Some(prefix) +} + +/// Attempt to flush a pending buffer's safe prefix, sanitize it, and wrap the result via +/// `wrap_kind` (`SanitizedChunkKind::Text` or `::Thinking`, both valid as a +/// `fn(String) -> SanitizedChunkKind` since each is a single-field tuple variant). Returns +/// `None` when [`split_off_safe_prefix`] found nothing new to emit yet. +fn try_flush_kind( + buf: &mut String, + holdback: usize, + def_name: &str, + layers: &SanitizeLayers, + wrap_kind: fn(String) -> SanitizedChunkKind, +) -> Option { + let safe = split_off_safe_prefix(buf, holdback)?; + Some(wrap_kind(sanitize_text(&safe, def_name, layers))) +} + +fn make_sanitized_chunk( + task_id: &Arc, + def_name: &Arc, + seq: u64, + kind: SanitizedChunkKind, +) -> SanitizedChunk { SanitizedChunk { - task_id: raw.task_id, - def_name: raw.def_name, - seq: raw.seq, + task_id: Arc::clone(task_id), + def_name: Arc::clone(def_name), + seq, kind, } } +/// Flush both pending buffers in full (no holdback — nothing more is coming for this task) +/// and dispatch any resulting chunk(s). Called immediately before an explicit `Terminal` +/// chunk or the hard-abort backstop, so buffered content is only ever delayed until the +/// run's very end, never silently dropped. +#[allow(clippy::too_many_arguments)] +fn flush_all_pending( + pending: &mut PendingSanitizeBuffers, + task_id: &Arc, + def_name: &Arc, + layers: &SanitizeLayers, + surfaces: ForwardSurfaces, + buffer: &ForwardBuffer, + dispatch: &mut impl FnMut(&SanitizedChunk, ForwardSurfaces, &ForwardBuffer), + emit_seq: &mut u64, +) { + if let Some(kind) = try_flush_kind( + &mut pending.text, + 0, + def_name.as_ref(), + layers, + SanitizedChunkKind::Text, + ) { + dispatch( + &make_sanitized_chunk(task_id, def_name, *emit_seq, kind), + surfaces, + buffer, + ); + *emit_seq += 1; + } + if let Some(kind) = try_flush_kind( + &mut pending.thinking, + 0, + def_name.as_ref(), + layers, + SanitizedChunkKind::Thinking, + ) { + dispatch( + &make_sanitized_chunk(task_id, def_name, *emit_seq, kind), + surfaces, + buffer, + ); + *emit_seq += 1; + } +} + /// Sender-side handle held by a single subagent's own turn loop for the lifetime of its /// run only. /// @@ -192,15 +314,11 @@ impl ForwardSender { fn try_send(&self, kind: ForwardChunkKind) { let seq = self.seq.fetch_add(1, Ordering::Relaxed); - let chunk = RawChunk { - task_id: Arc::clone(&self.task_id), - def_name: Arc::clone(&self.def_name), - seq, - kind, - }; + let chunk = RawChunk { kind }; if self.tx.try_send(chunk).is_ok() { tracing::debug!( task_id = %self.task_id, + def_name = %self.def_name, seq, "subagent.forward.emit" ); @@ -208,6 +326,7 @@ impl ForwardSender { let dropped = self.dropped.fetch_add(1, Ordering::Relaxed) + 1; tracing::warn!( task_id = %self.task_id, + def_name = %self.def_name, seq, dropped, "subagent.forward.drop: ingress channel full, chunk dropped" @@ -215,9 +334,17 @@ impl ForwardSender { } } - /// Forward one turn's full, untruncated text output. Call only from behind an + /// Forward a piece of assistant text output. Call only from behind an /// `if let Some(f) = forward` guard — the caller (`agent_loop.rs`) must never construct /// or clone the text ahead of that guard (FR-007). + /// + /// `text` may be a whole turn's full, untruncated text (FR-002a, the non-streaming or + /// stream-fallback path) or one incremental delta from a native streaming response + /// (FR-002b) — both are display-only chunks tail-dropped under backpressure identically; + /// see the module-level "Design contract" section. Callers must not send both the + /// streamed deltas and the final whole-turn text for the same turn — that would double- + /// forward the same content (see `agent_loop.rs::call_provider_with_status`'s `streamed` + /// flag). pub(crate) fn send_text(&self, text: &str) { if text.is_empty() { return; @@ -225,8 +352,9 @@ impl ForwardSender { self.try_send(ForwardChunkKind::Text(text.to_owned())); } - /// Forward one visible thinking block's text. Same no-op-behind-`Some` contract as - /// [`send_text`][Self::send_text]. + /// Forward a piece of visible thinking output — a whole completed thinking block + /// (FR-002a) or one incremental thinking delta (FR-002b). Same no-op-behind-`Some` + /// contract and no-double-forward caller responsibility as [`send_text`][Self::send_text]. pub(crate) fn send_thinking(&self, text: &str) { if text.is_empty() { return; @@ -382,16 +510,66 @@ async fn run_forward_drain_with( buffer: Arc, mut dispatch: impl FnMut(&SanitizedChunk, ForwardSurfaces, &ForwardBuffer), ) { - let mut next_seq: u64 = 0; + let mut pending = PendingSanitizeBuffers::default(); + let mut emit_seq: u64 = 0; loop { if let Some(raw) = rx.recv().await { - next_seq = raw.seq + 1; - let is_terminal = matches!(raw.kind, ForwardChunkKind::Terminal(_)); - let chunk = sanitize_chunk(raw, &layers); - dispatch(&chunk, surfaces, &buffer); - if is_terminal { - break; + match raw.kind { + ForwardChunkKind::Text(delta) => { + pending.text.push_str(&delta); + if let Some(kind) = try_flush_kind( + &mut pending.text, + SANITIZE_HOLDBACK_BYTES, + def_name.as_ref(), + &layers, + SanitizedChunkKind::Text, + ) { + dispatch( + &make_sanitized_chunk(&task_id, &def_name, emit_seq, kind), + surfaces, + &buffer, + ); + emit_seq += 1; + } + } + ForwardChunkKind::Thinking(delta) => { + pending.thinking.push_str(&delta); + if let Some(kind) = try_flush_kind( + &mut pending.thinking, + SANITIZE_HOLDBACK_BYTES, + def_name.as_ref(), + &layers, + SanitizedChunkKind::Thinking, + ) { + dispatch( + &make_sanitized_chunk(&task_id, &def_name, emit_seq, kind), + surfaces, + &buffer, + ); + emit_seq += 1; + } + } + ForwardChunkKind::Terminal(state) => { + flush_all_pending( + &mut pending, + &task_id, + &def_name, + &layers, + surfaces, + &buffer, + &mut dispatch, + &mut emit_seq, + ); + let chunk = make_sanitized_chunk( + &task_id, + &def_name, + emit_seq, + SanitizedChunkKind::Terminal(state), + ); + dispatch(&chunk, surfaces, &buffer); + break; + } } } else { tracing::warn!( @@ -399,12 +577,22 @@ async fn run_forward_drain_with( "subagent.forward.terminal: ingress channel closed without an explicit \ terminal chunk — synthesizing hard-abort backstop" ); - let synthesized = SanitizedChunk { - task_id: Arc::clone(&task_id), - def_name: Arc::clone(&def_name), - seq: next_seq, - kind: SanitizedChunkKind::Terminal(SubAgentState::Canceled), - }; + flush_all_pending( + &mut pending, + &task_id, + &def_name, + &layers, + surfaces, + &buffer, + &mut dispatch, + &mut emit_seq, + ); + let synthesized = make_sanitized_chunk( + &task_id, + &def_name, + emit_seq, + SanitizedChunkKind::Terminal(SubAgentState::Canceled), + ); dispatch(&synthesized, surfaces, &buffer); break; } @@ -686,6 +874,196 @@ mod tests { ); } + // --- Review Critical Issue #2: cross-delta secret/PII masking gap --- + + fn collect_forwarded_text(chunks: &[SanitizedChunk]) -> String { + chunks + .iter() + .filter_map(|c| match &c.kind { + SanitizedChunkKind::Text(t) => Some(t.as_str()), + _ => None, + }) + .collect() + } + + #[tokio::test(start_paused = true)] + async fn secret_split_across_two_deltas_is_still_masked() { + // A secret whose bytes are split across two separate `send_text` calls — simulating + // two ToolSseEvent::ContentChunk deltas arriving back-to-back during FR-002b + // streaming — must still be masked once both fragments have been buffered. Neither + // fragment alone contains the full registered secret value, so per-delta-isolated + // sanitization (the pre-fix behavior) would have let it straight through. + use zeph_sanitizer::secret_mask::{SecretCategory, SecretMaskRegistry}; + + let secret_value = "sk-live-topsecretvalue123456789"; + let registry = Arc::new(SecretMaskRegistry::new()); + registry.register("MY_KEY", secret_value, SecretCategory::ApiKey); + let (first_half, second_half) = secret_value.split_at(secret_value.len() / 2); + + let task_id: Arc = Arc::from("task-split"); + let def_name: Arc = Arc::from("agent-split"); + let (sender, rx) = new_channel(Arc::clone(&task_id), Arc::clone(&def_name)); + let buffer = new_buffer(); + + sender.send_text(&format!("the key is {first_half}")); + sender.send_text(&format!("{second_half}, use it wisely")); + sender.send_terminal(SubAgentState::Completed); + drop(sender); + + let seen = Arc::new(std::sync::Mutex::new(Vec::::new())); + let collected = Arc::clone(&seen); + let layers = SanitizeLayers { + sanitizer: ContentSanitizer::new(&zeph_config::ContentIsolationConfig::default()), + secret_registry: Some(registry), + pii_filter: None, + }; + run_forward_drain_with( + task_id, + def_name, + rx, + layers, + ForwardSurfaces { + tui: true, + bare: false, + }, + buffer, + move |chunk, surfaces, buffer| { + collected.lock().unwrap().push(chunk.clone()); + dispatch_chunk(chunk, surfaces, buffer); + }, + ) + .await; + + let combined = collect_forwarded_text(&seen.lock().unwrap()); + assert!( + !combined.contains(secret_value), + "secret split across two forwarded deltas must still be masked: {combined}" + ); + assert!( + combined.contains(" = Arc::from("task-split-pii"); + let def_name: Arc = Arc::from("agent-split-pii"); + let (sender, rx) = new_channel(Arc::clone(&task_id), Arc::clone(&def_name)); + let buffer = new_buffer(); + + sender.send_text(&format!("contact me at {first_half}")); + sender.send_text(&format!("{second_half} for details")); + sender.send_terminal(SubAgentState::Completed); + drop(sender); + + let seen = Arc::new(std::sync::Mutex::new(Vec::::new())); + let collected = Arc::clone(&seen); + let layers = SanitizeLayers { + sanitizer: ContentSanitizer::new(&zeph_config::ContentIsolationConfig::default()), + secret_registry: None, + pii_filter: Some(PiiFilter::new(PiiFilterConfig::default())), + }; + run_forward_drain_with( + task_id, + def_name, + rx, + layers, + ForwardSurfaces { + tui: true, + bare: false, + }, + buffer, + move |chunk, surfaces, buffer| { + collected.lock().unwrap().push(chunk.clone()); + dispatch_chunk(chunk, surfaces, buffer); + }, + ) + .await; + + let combined = collect_forwarded_text(&seen.lock().unwrap()); + assert!( + !combined.contains(email), + "email split across two forwarded deltas must still be scrubbed: {combined}" + ); + } + + #[tokio::test(start_paused = true)] + async fn secret_split_across_progressive_flush_boundary_is_still_masked() { + // Stronger test of the holdback *window* itself (not just "buffer until terminal"): + // enough filler precedes the secret's two fragments to force at least one + // progressive flush mid-stream (SANITIZE_HOLDBACK_BYTES is well under the total + // filler size), proving flushing genuinely happens before the terminal event, yet + // the secret's fragments — arriving back-to-back right after the filler — must still + // land inside the held-back tail and be masked as one contiguous string once + // fully buffered. + use zeph_sanitizer::secret_mask::{SecretCategory, SecretMaskRegistry}; + + let secret_value = "sk-live-anothersecretvalue987654321"; + let registry = Arc::new(SecretMaskRegistry::new()); + registry.register("MY_KEY", secret_value, SecretCategory::ApiKey); + let (first_half, second_half) = secret_value.split_at(secret_value.len() / 2); + + let task_id: Arc = Arc::from("task-window"); + let def_name: Arc = Arc::from("agent-window"); + let (sender, rx) = new_channel(Arc::clone(&task_id), Arc::clone(&def_name)); + let buffer = new_buffer(); + + for i in 0..40 { + sender.send_text(&format!("filler-chunk-{i:03} ")); + } + sender.send_text(first_half); + sender.send_text(second_half); + sender.send_terminal(SubAgentState::Completed); + drop(sender); + + let seen = Arc::new(std::sync::Mutex::new(Vec::::new())); + let collected = Arc::clone(&seen); + let layers = SanitizeLayers { + sanitizer: ContentSanitizer::new(&zeph_config::ContentIsolationConfig::default()), + secret_registry: Some(registry), + pii_filter: None, + }; + run_forward_drain_with( + task_id, + def_name, + rx, + layers, + ForwardSurfaces { + tui: true, + bare: false, + }, + buffer, + move |chunk, surfaces, buffer| { + collected.lock().unwrap().push(chunk.clone()); + dispatch_chunk(chunk, surfaces, buffer); + }, + ) + .await; + + let seen = seen.lock().unwrap(); + let text_chunk_count = seen + .iter() + .filter(|c| matches!(c.kind, SanitizedChunkKind::Text(_))) + .count(); + assert!( + text_chunk_count > 1, + "filler well over the holdback window must have produced at least one \ + progressive flush before the terminal-triggered final flush, got \ + {text_chunk_count} text chunk(s)" + ); + let combined = collect_forwarded_text(&seen); + assert!( + !combined.contains(secret_value), + "secret split across the streaming boundary must still be masked: {combined}" + ); + } + #[tokio::test(start_paused = true)] async fn buffer_entry_survives_during_grace_window_then_evicted() { // S3: the grace window's entire purpose is that a TUI view opened just after diff --git a/crates/zeph-subagent/src/lib.rs b/crates/zeph-subagent/src/lib.rs index 50274262f..5c6ad5c18 100644 --- a/crates/zeph-subagent/src/lib.rs +++ b/crates/zeph-subagent/src/lib.rs @@ -75,7 +75,7 @@ pub use hooks::{ PostToolUseHookInput, SubagentHooks, TOOL_ARGS_JSON_LIMIT, fire_hooks, hook_if_matches, make_base_hook_env, matching_hooks, }; -pub use manager::{SpawnContext, SubAgentHandle, SubAgentManager, SubAgentStatus}; +pub use manager::{SpawnContext, SpawnOrigin, SubAgentHandle, SubAgentManager, SubAgentStatus}; pub use memory::{ensure_memory_dir, load_memory_content}; pub use resolve::resolve_agent_paths; pub use state::SubAgentState; diff --git a/crates/zeph-subagent/src/manager/mod.rs b/crates/zeph-subagent/src/manager/mod.rs index d9db71d1c..02e8020ce 100644 --- a/crates/zeph-subagent/src/manager/mod.rs +++ b/crates/zeph-subagent/src/manager/mod.rs @@ -29,6 +29,41 @@ use crate::forward::ForwardSurfaces; use crate::grants::{GrantedSecret, PermissionGrants, SecretRequest}; use crate::state::SubAgentState; +/// Classifies what triggered a given spawn attempt, for [`DelegationMode`] enforcement +/// (spec `042-subagent-delegation-mode-parity`, issue #5857). +/// +/// # Fail-closed default +/// +/// [`SpawnOrigin::default`] is [`SpawnOrigin::Autonomous`] — the *restrictive* value, not the +/// permissive one. An untagged or forgotten [`SpawnContext`] therefore reads as `Autonomous` +/// and is denied under [`DelegationMode::ExplicitRequestOnly`]/[`DelegationMode::Disabled`], +/// never silently allowed. Only a caller that explicitly sets `origin = SpawnOrigin::Explicit` +/// (after auditing that the trigger really is a direct, attributable user action) can bypass +/// the `explicit_request_only` restriction. A mistagged legitimate spawn fails visibly (a +/// `tracing::warn!` plus a denied spawn); a mistagged illegitimate one can never be silently +/// let through. +/// +/// [`DelegationMode`]: zeph_config::DelegationMode +/// [`DelegationMode::ExplicitRequestOnly`]: zeph_config::DelegationMode::ExplicitRequestOnly +/// [`DelegationMode::Disabled`]: zeph_config::DelegationMode::Disabled +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SpawnOrigin { + /// A direct, attributable user action in the current turn (e.g. `/agent spawn`, + /// `/agent resume`) — always permitted except under `delegation_mode = "disabled"`. + Explicit, + /// Autonomous planner/scheduler decision-making with no corresponding explicit user + /// request in the current turn (e.g. the orchestration scheduler's DAG dispatch). Denied + /// under `delegation_mode = "explicit_request_only"` and `"disabled"`. + Autonomous, +} + +impl Default for SpawnOrigin { + /// Fail-closed: see the type-level doc comment. + fn default() -> Self { + Self::Autonomous + } +} + /// Parent-derived state propagated to a spawned sub-agent at spawn time. /// /// All fields default to empty/`None`, preserving existing behavior when callers @@ -48,7 +83,7 @@ use crate::state::SubAgentState; /// # Examples /// /// ```rust -/// use zeph_subagent::manager::SpawnContext; +/// use zeph_subagent::manager::{SpawnContext, SpawnOrigin}; /// /// // Minimal context — all fields use their defaults. /// let ctx = SpawnContext::default(); @@ -56,6 +91,8 @@ use crate::state::SubAgentState; /// assert_eq!(ctx.spawn_depth, 0); /// assert!(ctx.max_trust_level.is_none()); /// assert!(ctx.inherited_tool_allowlist.is_none()); +/// // Fail-closed: an untagged context reads as Autonomous, not Explicit. +/// assert_eq!(ctx.origin, SpawnOrigin::Autonomous); /// ``` #[derive(Default)] pub struct SpawnContext { @@ -186,6 +223,17 @@ pub struct SpawnContext { /// automatically — a sub-agent that spawns its own children must copy this field from /// its received `SpawnContext` into the child's, or grandchild LLM calls go undumped. pub debug_dump_sink: Option>, + + /// What triggered this spawn attempt — enforced against `delegation_mode` at the top of + /// [`SubAgentManager::spawn`] (spec `042-subagent-delegation-mode-parity`, issue #5857). + /// + /// Defaults to [`SpawnOrigin::Autonomous`] (fail-closed — see [`SpawnOrigin`]'s doc + /// comment). `zeph-core`'s `build_spawn_context` (the base used by the explicit + /// `/agent spawn`/`/agent resume` commands) sets this to [`SpawnOrigin::Explicit`]; the + /// orchestration scheduler overrides it back to `Autonomous` on its own call, mirroring + /// the existing post-construction override pattern used for + /// [`network_denied`][Self::network_denied] and [`progress_at`][Self::progress_at]. + pub origin: SpawnOrigin, } /// Live status snapshot of a running sub-agent. @@ -403,6 +451,13 @@ pub struct SubAgentManager { /// `PiiScrubbingDumpSink` (#6407). `None` (the default) means no PII scrubbing beyond the /// baseline `ContentSanitizer` pass — set via [`set_pii_filter`][Self::set_pii_filter]. pii_filter: Option, + /// Effective delegation mode gating every spawn (spec `042-subagent-delegation-mode-parity`, + /// issue #5857). Already folds in the `enabled` outer kill switch — the bootstrap caller + /// computes `if !config.agents.enabled { Disabled } else { config.agents.delegation_mode }` + /// before calling [`set_delegation_mode`][Self::set_delegation_mode], so this field alone + /// 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, } impl std::fmt::Debug for SubAgentManager { @@ -425,6 +480,7 @@ impl std::fmt::Debug for SubAgentManager { .field("forward_buffer", &"") .field("secret_registry", &self.secret_registry.is_some()) .field("pii_filter", &self.pii_filter.is_some()) + .field("delegation_mode", &self.delegation_mode) .finish() } } @@ -451,6 +507,7 @@ impl SubAgentManager { forward_buffer: crate::forward::new_buffer(), secret_registry: None, pii_filter: None, + delegation_mode: zeph_config::DelegationMode::default(), } } @@ -475,6 +532,27 @@ impl SubAgentManager { self.forward_surfaces = surfaces; } + /// Set the effective delegation mode gating every future [`spawn`][Self::spawn] call + /// (spec `042-subagent-delegation-mode-parity`, issue #5857). + /// + /// Call during bootstrap, before the first [`spawn`][Self::spawn]. The caller MUST fold + /// in the `enabled` outer kill switch before calling this: pass + /// `zeph_config::DelegationMode::Disabled` when `config.agents.enabled == false`, + /// regardless of `config.agents.delegation_mode`'s configured value (FR-002). The manager + /// itself performs no `enabled` check — it only sees the already-resolved effective mode. + pub fn set_delegation_mode(&mut self, mode: zeph_config::DelegationMode) { + self.delegation_mode = mode; + } + + /// The effective delegation mode currently in force, as set by + /// [`set_delegation_mode`][Self::set_delegation_mode]. Exposed for status/observability + /// surfaces (e.g. `/agent list`) so operators can confirm the active mode without + /// inspecting `config.toml` directly (spec 042 NFR-004). + #[must_use] + pub fn delegation_mode(&self) -> zeph_config::DelegationMode { + self.delegation_mode + } + /// Wire the bootstrap-level secret-mask registry into the forwarding pipeline (issue /// #6359, security Finding 1 / NFR-005). /// diff --git a/crates/zeph-subagent/src/manager/spawn.rs b/crates/zeph-subagent/src/manager/spawn.rs index 5a41fc4bd..f0752657e 100644 --- a/crates/zeph-subagent/src/manager/spawn.rs +++ b/crates/zeph-subagent/src/manager/spawn.rs @@ -586,6 +586,34 @@ impl SubAgentManager { config: &SubAgentConfig, ctx: SpawnContext, ) -> Result { + // Delegation-mode gate (spec 042, issue #5857): checked first, before any resource + // allocation (NFR-002) — a rejected spawn must have zero side effects (no worktree, + // no transcript file, no consumed concurrency slot). Expressed as an explicit allow-list + // (rather than a `match` computing `denied`) so that `DelegationMode` being + // `#[non_exhaustive]` fails closed automatically: any future variant this crate + // doesn't yet recognize matches neither arm below and is denied, not silently allowed. + let allowed = matches!( + (self.delegation_mode, ctx.origin), + (zeph_config::DelegationMode::Proactive, _) + | ( + zeph_config::DelegationMode::ExplicitRequestOnly, + super::SpawnOrigin::Explicit + ) + ); + if !allowed { + tracing::warn!( + mode = ?self.delegation_mode, + origin = ?ctx.origin, + def_name, + "sub-agent spawn rejected by delegation_mode" + ); + return Err(SubAgentError::DelegationDenied { + mode: self.delegation_mode, + origin: ctx.origin, + def_name: def_name.to_owned(), + }); + } + if ctx.spawn_depth >= config.max_spawn_depth { return Err(SubAgentError::MaxDepthExceeded { depth: ctx.spawn_depth, @@ -1040,6 +1068,24 @@ impl SubAgentManager { config: &SubAgentConfig, spawn_context: Option<&SpawnContext>, ) -> Result<(String, String), SubAgentError> { + // Delegation-mode gate (spec 042, issue #5857): `resume` is its own chokepoint, + // distinct from `spawn` — checked first, before any resource allocation (NFR-002). + // Resuming a sub-agent is inherently an explicit, attributable user action (there is + // no autonomous-resume path in this codebase), so it only needs the mode-only + // allow-list, not the origin-aware check `spawn` uses. + if !self.delegation_mode.permits_explicit() { + tracing::warn!( + mode = ?self.delegation_mode, + id_prefix, + "sub-agent resume rejected by delegation_mode" + ); + return Err(SubAgentError::DelegationDenied { + mode: self.delegation_mode, + origin: super::SpawnOrigin::Explicit, + def_name: id_prefix.to_owned(), + }); + } + let dir = self.effective_transcript_dir(config); let id_prefix_owned = id_prefix.to_owned(); let dir_clone = dir.clone(); diff --git a/crates/zeph-subagent/src/manager/tests.rs b/crates/zeph-subagent/src/manager/tests.rs index 57e05cabf..c163cbffc 100644 --- a/crates/zeph-subagent/src/manager/tests.rs +++ b/crates/zeph-subagent/src/manager/tests.rs @@ -3011,6 +3011,280 @@ fn spawn_context_default_is_empty() { assert!(ctx.mcp_tool_names.is_empty()); } +#[test] +fn spawn_context_default_origin_is_autonomous() { + // Fail-closed (spec 042, issue #5857): an untagged SpawnContext must read as the + // restrictive value, not the permissive one. + assert_eq!(SpawnContext::default().origin, SpawnOrigin::Autonomous); +} + +/// Delegation-mode × spawn-origin matrix (spec `042-subagent-delegation-mode-parity`, +/// issue #5857): every combination of the three `DelegationMode` values and the two +/// `SpawnOrigin` values, verifying `SubAgentManager::spawn`'s gate at the exact chokepoint +/// both `spawn` and `spawn_for_task` share. +mod delegation_mode_gate { + use super::*; + + async fn try_spawn( + mgr: &mut SubAgentManager, + mode: zeph_config::DelegationMode, + origin: SpawnOrigin, + ) -> Result { + mgr.set_delegation_mode(mode); + let ctx = SpawnContext { + origin, + ..SpawnContext::default() + }; + mgr.spawn( + "bot", + "task", + mock_provider(vec!["done"]), + noop_executor(), + None, + &SubAgentConfig::default(), + ctx, + ) + .await + } + + #[tokio::test] + async fn disabled_denies_explicit() { + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + let err = try_spawn( + &mut mgr, + zeph_config::DelegationMode::Disabled, + SpawnOrigin::Explicit, + ) + .await + .unwrap_err(); + assert!(matches!(err, SubAgentError::DelegationDenied { .. })); + } + + #[tokio::test] + async fn disabled_denies_autonomous() { + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + let err = try_spawn( + &mut mgr, + zeph_config::DelegationMode::Disabled, + SpawnOrigin::Autonomous, + ) + .await + .unwrap_err(); + assert!(matches!(err, SubAgentError::DelegationDenied { .. })); + } + + #[tokio::test] + async fn explicit_request_only_allows_explicit() { + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + let result = try_spawn( + &mut mgr, + zeph_config::DelegationMode::ExplicitRequestOnly, + SpawnOrigin::Explicit, + ) + .await; + assert!(result.is_ok(), "expected allow, got {result:?}"); + } + + #[tokio::test] + async fn explicit_request_only_denies_autonomous() { + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + let err = try_spawn( + &mut mgr, + zeph_config::DelegationMode::ExplicitRequestOnly, + SpawnOrigin::Autonomous, + ) + .await + .unwrap_err(); + assert!(matches!(err, SubAgentError::DelegationDenied { .. })); + } + + #[tokio::test] + async fn proactive_allows_explicit() { + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + let result = try_spawn( + &mut mgr, + zeph_config::DelegationMode::Proactive, + SpawnOrigin::Explicit, + ) + .await; + assert!(result.is_ok(), "expected allow, got {result:?}"); + } + + #[tokio::test] + async fn proactive_allows_autonomous() { + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + let result = try_spawn( + &mut mgr, + zeph_config::DelegationMode::Proactive, + SpawnOrigin::Autonomous, + ) + .await; + assert!(result.is_ok(), "expected allow, got {result:?}"); + } + + /// `spawn_for_task` delegates straight to `spawn` (see `manager/spawn.rs`), so the same + /// gate must reject it too — regression guard against the two entry points drifting apart. + #[tokio::test] + async fn spawn_for_task_is_gated_identically() { + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + mgr.set_delegation_mode(zeph_config::DelegationMode::ExplicitRequestOnly); + let ctx = SpawnContext { + origin: SpawnOrigin::Autonomous, + ..SpawnContext::default() + }; + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let err = mgr + .spawn_for_task( + "bot", + "task", + mock_provider(vec!["done"]), + noop_executor(), + None, + &SubAgentConfig::default(), + ctx, + move |id, result| { + let _ = tx.send((id, result)); + }, + ) + .await + .unwrap_err(); + assert!(matches!(err, SubAgentError::DelegationDenied { .. })); + assert!( + rx.try_recv().is_err(), + "on_done must never fire for a rejected spawn" + ); + } + + /// `enabled = false` must resolve to `Disabled` regardless of `delegation_mode`'s + /// configured value (FR-002) — exercised end-to-end via + /// `SubAgentConfig::effective_delegation_mode` (the exact fold `src/runner.rs` bootstrap + /// applies before calling `set_delegation_mode`). + #[tokio::test] + async fn enabled_false_kill_switch_overrides_proactive() { + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + let cfg = zeph_config::SubAgentConfig { + enabled: false, + delegation_mode: zeph_config::DelegationMode::Proactive, + ..zeph_config::SubAgentConfig::default() + }; + let err = try_spawn( + &mut mgr, + cfg.effective_delegation_mode(), + SpawnOrigin::Explicit, + ) + .await + .unwrap_err(); + assert!(matches!(err, SubAgentError::DelegationDenied { .. })); + } + + /// Regression test for the review finding that `resume()` had no delegation gate at all + /// (issue #5857 review round): `/agent resume` must be rejected under `disabled`, exactly + /// like `spawn()`. `resume` has no `Autonomous` origin path in this codebase, so it is + /// gated by the mode-only `DelegationMode::permits_explicit()` allow-list, not the + /// origin-aware matrix `spawn()` uses. + #[tokio::test] + async fn resume_denied_when_disabled() { + let tmp = tempfile::tempdir().unwrap(); + let agent_id = "aaaa1111-0000-0000-0000-000000000000"; + write_completed_meta(tmp.path(), agent_id, "bot"); + + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + mgr.set_delegation_mode(zeph_config::DelegationMode::Disabled); + let cfg = make_cfg_with_dir(tmp.path()); + + let err = mgr + .resume( + "aaaa1111", + "continue the work", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + None, + ) + .await + .unwrap_err(); + assert!( + matches!(err, SubAgentError::DelegationDenied { .. }), + "expected DelegationDenied, got {err:?}" + ); + assert!( + !mgr.agents.contains_key(agent_id), + "a denied resume must not register any agent" + ); + } + + /// `enabled = false` (the kill switch) must also reject `resume`, not only `spawn`. + #[tokio::test] + async fn resume_denied_when_enabled_false_kill_switch() { + let tmp = tempfile::tempdir().unwrap(); + let agent_id = "bbbb2222-0000-0000-0000-000000000000"; + write_completed_meta(tmp.path(), agent_id, "bot"); + + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + let cfg = { + let mut c = make_cfg_with_dir(tmp.path()); + c.enabled = false; + c.delegation_mode = zeph_config::DelegationMode::Proactive; + c + }; + mgr.set_delegation_mode(cfg.effective_delegation_mode()); + + let err = mgr + .resume( + "bbbb2222", + "continue the work", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + None, + ) + .await + .unwrap_err(); + assert!(matches!(err, SubAgentError::DelegationDenied { .. })); + } + + /// `explicit_request_only` must still permit `resume` — it is inherently an explicit + /// user action. + #[tokio::test] + async fn resume_allowed_under_explicit_request_only() { + let tmp = tempfile::tempdir().unwrap(); + let agent_id = "cccc3333-0000-0000-0000-000000000000"; + write_completed_meta(tmp.path(), agent_id, "bot"); + + let mut mgr = make_manager(); + mgr.definitions.push(sample_def()); + mgr.set_delegation_mode(zeph_config::DelegationMode::ExplicitRequestOnly); + let cfg = make_cfg_with_dir(tmp.path()); + + let result = mgr + .resume( + "cccc3333", + "continue the work", + mock_provider(vec!["done"]), + noop_executor(), + None, + &cfg, + None, + ) + .await; + assert!(result.is_ok(), "expected allow, got {result:?}"); + let (new_id, _) = result.unwrap(); + mgr.cancel(&new_id).unwrap(); + } +} + #[test] fn context_injection_none_passes_raw_prompt() { use zeph_config::ContextInjectionMode; diff --git a/docker/docker-compose.dev.yml b/docker/docker-compose.dev.yml index acf26a577..8d03ffaf9 100644 --- a/docker/docker-compose.dev.yml +++ b/docker/docker-compose.dev.yml @@ -119,6 +119,7 @@ services: ZEPH_LOG_FILE: ${ZEPH_LOG_FILE:-} ZEPH_LOG_LEVEL: ${ZEPH_LOG_LEVEL:-} ZEPH_AGENTS_FORWARD_TRANSCRIPT: ${ZEPH_AGENTS_FORWARD_TRANSCRIPT:-} + ZEPH_AGENTS_DELEGATION_MODE: ${ZEPH_AGENTS_DELEGATION_MODE:-} ZEPH_AUTO_UPDATE_CHECK: ${ZEPH_AUTO_UPDATE_CHECK:-false} ports: - "${ZEPH_A2A_PORT:-8080}:${ZEPH_A2A_PORT:-8080}" diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 1d6019709..9478d0e8e 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -140,6 +140,7 @@ services: ZEPH_LOG_FILE: ${ZEPH_LOG_FILE:-} ZEPH_LOG_LEVEL: ${ZEPH_LOG_LEVEL:-} ZEPH_AGENTS_FORWARD_TRANSCRIPT: ${ZEPH_AGENTS_FORWARD_TRANSCRIPT:-} + ZEPH_AGENTS_DELEGATION_MODE: ${ZEPH_AGENTS_DELEGATION_MODE:-} ZEPH_AUTO_UPDATE_CHECK: ${ZEPH_AUTO_UPDATE_CHECK:-false} ports: - "${ZEPH_A2A_PORT:-8080}:${ZEPH_A2A_PORT:-8080}" diff --git a/specs/044-subagent-lifecycle/spec.md b/specs/044-subagent-lifecycle/spec.md index b28fb86b4..0d9e99aa6 100644 --- a/specs/044-subagent-lifecycle/spec.md +++ b/specs/044-subagent-lifecycle/spec.md @@ -426,6 +426,78 @@ interleaves two different JSON schemas on stdout — unsupported for now; use `- --- +## 15. Delegation Mode Gate (spec `042-subagent-delegation-mode-parity`, issue #5857) + +### Problem + +Prior to this addition, `SubAgentConfig.enabled: bool` was the only lever governing sub-agent +spawning, and it was **inert** — no code path actually read it. An operator had no way to keep +the sub-agent subsystem enabled and useful while forbidding the main agent from autonomously +deciding to spawn one (e.g. in a semi-trusted channel where prompt-injected input could reach +the agent) — see `[[042-subagent-delegation-mode-parity/spec]]` for the full motivation. + +### Mechanism + +`SubAgentConfig` gains `delegation_mode: DelegationMode` (`disabled` / `explicit_request_only` / +`proactive`, `#[serde(default)]` → `Proactive`, preserving prior unconstrained behavior). Every +spawn attempt carries a `SpawnOrigin` (`Explicit` / `Autonomous`) on `SpawnContext`, enforced at +the top of `SubAgentManager::spawn` (the single chokepoint — `spawn_for_task` delegates to it, +so both share the gate automatically): + +- `disabled` rejects every spawn regardless of origin (read-only ops — `/agent list`, status + queries — are unaffected, since they never call `spawn`). +- `explicit_request_only` rejects `Autonomous`-origin spawns, permits `Explicit`. +- `proactive` permits both, unchanged from pre-existing behavior. + +**Fail-closed default**: `SpawnOrigin::default() = Autonomous` — the *restrictive* value, not +the permissive one. An untagged or forgotten `SpawnContext` is therefore denied under the +restrictive modes rather than silently allowed. Every real spawn call site was audited and +explicitly tagged: + +- `build_spawn_context` (`crates/zeph-core/src/agent/subagent_commands.rs`, shared by + `/agent spawn` foreground/background and `/agent resume`) sets `Explicit` — all three of its + callers are dispatched from the explicit `/agent` slash command. +- `handle_scheduler_spawn_action` (`crates/zeph-core/src/agent/scheduler_loop.rs`) overrides + `spawn_ctx.origin = Autonomous` immediately after calling `build_spawn_context`, mirroring the + existing post-construction override pattern already used for `network_denied`/`progress_at`. + This is the orchestration scheduler's autonomous DAG dispatch — the concrete threat this gate + protects against, since it spawns without any turn-level user confirmation. + +`SubAgentConfig.enabled` becomes the outer kill switch via +`SubAgentConfig::effective_delegation_mode()`: `enabled = false` always resolves to `Disabled` +regardless of `delegation_mode`'s configured value. `src/runner.rs` bootstrap calls +`mgr.set_delegation_mode(agents_config.effective_delegation_mode())` once, before any spawn can +occur; the manager itself never re-reads `enabled`. + +A rejected spawn returns `SubAgentError::DelegationDenied { mode, origin, def_name }` before any +resource is allocated (no worktree, no transcript file, no consumed concurrency slot), and logs +a `tracing::warn!` distinguishable from `ConcurrencyLimit`/`MaxDepthExceeded` rejections. + +### ACP `/subagent spawn` — a separate gate, not a `SpawnContext` tag + +`/subagent spawn ` (`crates/zeph-core/src/agent/slash_commands.rs::handle_subagent_slash`) +launches an **external ACP process** via `zeph_acp::run_session` and never constructs a +`SpawnContext` or touches `SubAgentManager` at all — the gate above does not see it. Spec 042 +FR-003 requires `disabled` mode to reject every spawn path, so `handle_subagent_slash` carries +its own explicit check against `SubAgentConfig::effective_delegation_mode()` before invoking the +spawn callback, rejecting only under `Disabled` (the command is itself an explicit user action, +so it remains permitted under `explicit_request_only` and `proactive`). + +### Key Invariants + +- The gate check runs before any resource allocation (NFR-002) — no partial worktree, transcript, + or concurrency-slot side effect on a rejected spawn. +- `SpawnOrigin`'s fail-closed default means a forgotten call site is a visible, safe denial under + restrictive modes, never a silent bypass. +- `delegation_mode` is orthogonal to `default_permission_mode`/`PermissionMode` — the former + governs *whether* a spawn may happen and *who* may trigger it; the latter governs what a + spawned sub-agent is allowed to *do* once running. Never merge the two. +- Out of scope for this addition (deferred, see spec 042 Open Questions): per-turn/per-session + override of `delegation_mode` (FR-011); a dedicated TUI status-bar/sidebar indicator beyond + the `/agent list` header line (NFR-004 partial). + +--- + ## 10. See Also - [[constitution]] — project principles diff --git a/src/cli.rs b/src/cli.rs index 9a08c5101..3babeed76 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -338,6 +338,15 @@ pub(crate) struct Cli { #[arg(long, value_name = "REF")] pub(crate) worktree_base_ref: Option, + /// Override the `agents.delegation_mode` config for this session. + /// + /// Accepted values: `disabled`, `explicit_request_only`, `proactive`. Still subject to + /// `agents.enabled` as the outer kill switch — `enabled = false` always resolves to + /// `disabled` regardless of this flag (spec `042-subagent-delegation-mode-parity`, + /// issue #5857). + #[arg(long, value_name = "MODE")] + pub(crate) delegation_mode: Option, + #[command(subcommand)] pub(crate) command: Option, diff --git a/src/init/agents.rs b/src/init/agents.rs index cbcdd9a4e..f2c6c3866 100644 --- a/src/init/agents.rs +++ b/src/init/agents.rs @@ -179,6 +179,39 @@ fn step_ensemble_verify(state: &mut WizardState) -> anyhow::Result<()> { pub(super) fn step_agents(state: &mut WizardState) -> anyhow::Result<()> { println!("== Step 9/10: Sub-Agent Defaults ==\n"); + state.agents_enabled = Confirm::new() + .with_prompt("Enable the sub-agent subsystem? (required for /agent commands and multi-agent workflows)") + .default(true) + .interact()?; + + if state.agents_enabled { + let delegation_items = [ + "proactive (main agent may decide on its own to delegate to a sub-agent)", + "explicit_request_only (only /agent spawn — the main agent may never decide on its own)", + "disabled (no spawn from any code path; definitions remain listable)", + ]; + let delegation_sel = Select::new() + .with_prompt( + "Delegation mode — who may trigger a sub-agent spawn? (see spec \ + 042-subagent-delegation-mode-parity; useful to restrict in \ + semi-trusted channels such as Telegram/Discord/webhook ingestion)", + ) + .items(delegation_items) + // Default to explicit_request_only (index 1), not proactive: a brand-new + // interactive wizard run has no prior deployment behavior to preserve, so the + // suggested answer for a first-time operator should be the more conservative + // option. `DelegationMode::default() = Proactive` remains correct for the + // struct-level default (FR-008, preserves existing `enabled=true` deployments) — + // this only changes the wizard's suggested answer. + .default(1) + .interact()?; + state.agents_delegation_mode = match delegation_sel { + 1 => zeph_config::DelegationMode::ExplicitRequestOnly, + 2 => zeph_config::DelegationMode::Disabled, + _ => zeph_config::DelegationMode::Proactive, + }; + } + let modes = ["default", "accept_edits", "dont_ask"]; let sel = Select::new() .with_prompt("Default permission mode for sub-agents") diff --git a/src/init/mod.rs b/src/init/mod.rs index 62b4738a8..e38c7443d 100644 --- a/src/init/mod.rs +++ b/src/init/mod.rs @@ -99,6 +99,16 @@ pub(crate) struct WizardState { /// `gemini_thinking_level` above — this field is `OpenAI`-specific to avoid two /// conflicting wizard prompts for the same knob on those providers. pub(crate) reasoning_effort: Option, + /// Enable the sub-agent subsystem — outer kill switch (spec 042, issue #5857). Default + /// `true` in the wizard: prior to this field's introduction, `enabled` was never wired to + /// any gate, so wizard-produced configs behaved as fully unconstrained regardless of its + /// value. Defaulting to `true` here preserves that observed behavior for fresh installs; + /// operators who want the new kill switch to actually restrict spawning can say "no" or + /// dial `agents_delegation_mode` down instead. + pub(crate) agents_enabled: bool, + /// Tri-state delegation-mode control (spec 042, issue #5857). Default `Proactive` — + /// preserves the subsystem's pre-existing unconstrained behavior. + pub(crate) agents_delegation_mode: zeph_config::DelegationMode, pub(crate) agents_default_permission_mode: Option, pub(crate) agents_default_disallowed_tools: Vec, pub(crate) agents_allow_bypass_permissions: bool, @@ -429,6 +439,8 @@ impl Default for WizardState { thinking: None, enable_extended_context: false, reasoning_effort: None, + agents_enabled: true, + agents_delegation_mode: zeph_config::DelegationMode::default(), agents_default_permission_mode: None, agents_default_disallowed_tools: Vec::new(), agents_allow_bypass_permissions: false, @@ -1206,6 +1218,8 @@ pub(crate) fn build_config(state: &WizardState) -> Config { ..zeph_tools::SearchConfig::default() }; + config.agents.enabled = state.agents_enabled; + config.agents.delegation_mode = state.agents_delegation_mode; config.agents.default_permission_mode = state.agents_default_permission_mode; config .agents diff --git a/src/runner.rs b/src/runner.rs index 726327ef7..4b06cde9a 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -3270,6 +3270,29 @@ pub(crate) async fn run(mut cli: Cli) -> anyhow::Result<()> { zeph_config::WorktreeBaseRef::Head }; } + if let Some(ref mode_str) = cli.delegation_mode { + match mode_str.as_str() { + "disabled" => agents_config.delegation_mode = zeph_config::DelegationMode::Disabled, + "explicit_request_only" => { + agents_config.delegation_mode = + zeph_config::DelegationMode::ExplicitRequestOnly; + } + "proactive" => { + agents_config.delegation_mode = zeph_config::DelegationMode::Proactive; + } + other => { + tracing::warn!( + value = other, + "--delegation-mode: invalid value, ignoring \ + (expected disabled|explicit_request_only|proactive)" + ); + } + } + } + // Sub-agent delegation gate (spec 042, issue #5857): `enabled` remains the outer kill + // switch — `effective_delegation_mode` folds it in so `SubAgentManager` itself only + // ever sees the already-resolved value and does not need to re-read `enabled` (FR-002). + mgr.set_delegation_mode(agents_config.effective_delegation_mode()); // Bootstrap the worktree subsystem when enabled (hard-fail per spec NEVER #92). if agents_config.worktree.enabled && !exec_mode.bare {