a3s_code_core/tools/

task.rs

//! Task tools for delegated child runs.
//!
//! The Task tool allows the main agent to delegate one or more specialized
//! tasks to focused child runs. Each child run gets bounded context and the
//! permissions declared by its agent definition.
//!
//! ## Usage
//!
//! ```json
//! {"tasks": [{
//!   "agent": "explore",
//!   "description": "Find authentication code",
//!   "prompt": "Search for files related to user authentication..."
//! }]}
//! ```

use crate::agent::{AgentConfig, AgentEvent, AgentLoop};
use crate::llm::structured::{
    generate_blocking_with_cancellation, parse_validated_output, StructuredMode, StructuredRequest,
};
use crate::llm::{LlmClient, ToolDefinition};
use crate::mcp::{McpBinding, McpManager};
use crate::orchestration::{AgentExecutor, AgentStepSpec, StepOutcome, ToolSourceAnchor};
use crate::subagent::{AgentDefinition, AgentRegistry};
use crate::tools::types::{Tool, ToolContext, ToolOutput};
use anyhow::{Context, Result};
use async_trait::async_trait;
use futures::FutureExt;
use serde::{Deserialize, Serialize};
use std::any::Any;
use std::collections::HashSet;
use std::panic::AssertUnwindSafe;
use std::path::PathBuf;
use std::sync::{Arc, Mutex, MutexGuard};
use tokio::sync::broadcast;
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;

const TASK_OUTPUT_CONTEXT_LIMIT: usize = 4_000;
const TASK_OUTPUT_CONTEXT_HEAD: usize = 3_000;
const TASK_OUTPUT_CONTEXT_TAIL: usize = 800;
const MAX_TASK_SOURCE_ANCHORS: usize = 64;
const MAX_TASK_SOURCE_CANDIDATES: usize = MAX_TASK_SOURCE_ANCHORS * 4;
const MAX_TASK_SOURCE_TOOL_BYTES: usize = 64;
const MAX_TASK_SOURCE_VALUE_BYTES: usize = 4 * 1024;
const MAX_PARALLEL_TASK_SOURCE_ANCHORS: usize = MAX_TASK_SOURCE_ANCHORS;
const TASK_TOOL_DESCRIPTION: &str = "Delegate one or more bounded tasks to specialized child runs. Pass one item for a focused child run or multiple INDEPENDENT items for concurrent fan-out. A single item may run in the background; multi-item calls are collected by the parent. By default any failed child makes a multi-item call fail; evidence-gathering callers may set allow_partial_failure=true. Choose canonical worker names from the live agent catalog. Custom agents from agent_dirs and .a3s/agents are supported; .claude/agents is read for compatibility.";
const PARALLEL_TASK_TOOL_DESCRIPTION: &str = "REMOVED from the model-visible registry (`HARNESS-CONV4`). Prefer `task` with multiple `tasks[]` items. This type remains for focused unit tests that construct ParallelTaskTool directly.";

/// Task tool parameters
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TaskParams {
    /// Agent type to use (explore, general, plan, verification, review, etc.)
    pub agent: String,
    /// Short description of the task (for display)
    pub description: String,
    /// Detailed prompt for the agent
    pub prompt: String,
    /// Optional: run in background (default: false)
    #[serde(default)]
    pub background: bool,
    /// Optional: maximum steps for this task
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_steps: Option<usize>,
    /// Optional: JSON schema the child result must satisfy.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub output_schema: Option<serde_json::Value>,
}

/// Task tool result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskResult {
    /// Task output from the delegated child run.
    pub output: String,
    /// Child session ID
    pub session_id: String,
    /// Agent type used
    pub agent: String,
    /// Whether the task succeeded
    pub success: bool,
    /// Task ID for tracking
    pub task_id: String,
    /// Structured child output validated against an optional output schema.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub structured: Option<serde_json::Value>,
    /// Source locations observed by successful built-in child tool calls.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub source_anchors: Vec<ToolSourceAnchor>,
}

struct ScopedTaskExecution<'a> {
    event_tx: Option<broadcast::Sender<AgentEvent>>,
    parent_session_id: Option<&'a str>,
    emit_start: bool,
    parent_cancellation: Option<&'a CancellationToken>,
    admitted_capability_subtask: Option<crate::capability::AgentCapabilitySubtask>,
    parallel_lifecycle: Option<Arc<ParallelTaskLifecycle>>,
}

/// Coordinates terminal lifecycle events for one bounded parallel fan-out.
///
/// A child normally emits its own `SubagentEnd`, but an outer `JoinSet` may
/// have to abort a child that is stuck in a non-cooperative provider or
/// subprocess. The fan-out then emits a synthetic terminal event. Keeping
/// this state separate from the public task tracker lets the natural and
/// synthetic paths race safely while still guaranteeing exactly one end event
/// for every emitted start event.
#[derive(Default)]
pub(super) struct ParallelTaskLifecycle {
    state: Mutex<ParallelTaskLifecycleState>,
}

#[derive(Default)]
struct ParallelTaskLifecycleState {
    started: HashSet<String>,
    ended: HashSet<String>,
}

impl ParallelTaskLifecycle {
    fn lock_state(&self) -> MutexGuard<'_, ParallelTaskLifecycleState> {
        match self.state.lock() {
            Ok(guard) => guard,
            // A poisoned lifecycle state still contains the authoritative
            // event history. Recover it instead of turning cancellation
            // cleanup into a process panic.
            Err(poisoned) => poisoned.into_inner(),
        }
    }

    /// Record that the start event has been written to the tracker/stream.
    /// This method deliberately has no await point: once a start is observable
    /// the enclosing task cannot be aborted between the write and this mark.
    fn mark_started(&self, task_id: &str) {
        self.lock_state().started.insert(task_id.to_string());
    }

    fn is_started(&self, task_id: &str) -> bool {
        self.lock_state().started.contains(task_id)
    }

    /// Mark a started task as terminal after its terminal event has been
    /// written. The parallel fan-out drains every child join before synthetic
    /// cleanup, so natural and synthetic terminal emitters cannot overlap.
    fn mark_ended(&self, task_id: &str) {
        let mut state = self.lock_state();
        if state.started.contains(task_id) {
            state.ended.insert(task_id.to_string());
        }
    }

    fn is_ended(&self, task_id: &str) -> bool {
        self.lock_state().ended.contains(task_id)
    }
}

mod result_projection;
use result_projection::*;

mod parallel_execution;

const MAX_PARALLEL_TASKS_PER_CALL: usize = 32;

fn provider_quota_for_client(
    client: &dyn LlmClient,
) -> Option<crate::task_scheduler::TaskSchedulerQuota> {
    let pool = client.model_generation_pool()?;
    crate::task_scheduler::TaskSchedulerQuota::new(
        pool.identity.clone(),
        pool.max_concurrency().get(),
    )
    .ok()
}

/// Task executor for delegated child runs.
#[derive(Clone)]
pub struct TaskExecutor {
    /// Agent registry for looking up agent definitions
    registry: Arc<AgentRegistry>,
    /// LLM client used to power child agent loops
    llm_client: Arc<dyn LlmClient>,
    /// Workspace path shared with child agents
    workspace: String,
    /// Ordered MCP managers for registering inherited tools in child sessions.
    mcp_managers: Vec<Arc<McpManager>>,
    /// Exact projected MCP bindings inherited from the admitted parent Run.
    mcp_bindings: Vec<Arc<McpBinding>>,
    /// Optional Tool presentation profile forced onto delegated children.
    /// Tests use Direct so Adaptive selection cannot hide manager-injected
    /// MCP tools and mask OPT-MCP1 regressions.
    child_tool_presentation: Option<crate::tools::ToolPresentationProfileV1>,
    /// Exact host tools owned by this executor. They are installed only in
    /// child executors and remain bounded by the composed parent/child
    /// governance context.
    scoped_tools: Vec<Arc<dyn Tool>>,
    /// Parent capabilities to inherit into child runs.
    parent_context: Option<crate::child_run::ChildRunContext>,
    /// Search configuration captured from the invoking parent context.
    search_config: Option<Arc<crate::config::SearchConfig>>,
    /// Agent-scoped search admission shared with delegated child runs.
    search_bulkhead: Option<a3s_search::Bulkhead>,
    /// Agent-scoped headless retry allowance shared with delegated child runs.
    search_retry_budget: Option<a3s_search::RetryBudget>,
    /// Parent-session request flights shared with delegated child runs.
    search_request_coalescer: Option<a3s_search::SearchCoalescer>,
    /// Weak capability parents captured from the invoking Tool Turn.
    capability_context: Option<crate::capability::AgentToolCapabilityContext>,
    /// Optional lifetime boundary inherited from the session that created this
    /// executor. Keeping it on the executor prevents cached workflow/executor
    /// handles from starting new child runs after their session is closed.
    parent_cancellation: Option<CancellationToken>,
    max_parallel_tasks: usize,
    /// Shared across every fan-out started by this executor. Per-call wave
    /// limits alone are insufficient when a dynamic workflow launches several
    /// `parallel_task` host steps concurrently.
    parallel_permits: Arc<tokio::sync::Semaphore>,
    /// Optional shared tracker — when present each task registers a
    /// `CancellationToken` so callers can cancel by `task_id`.
    subagent_tracker: Option<Arc<crate::subagent_task_tracker::InMemorySubagentTaskTracker>>,
    /// Agent-wide scheduler for independent delegated work.
    task_scheduler: Option<Arc<crate::task_scheduler::TaskScheduler>>,
    /// Host-started workflow executors have no parent lease, so their
    /// foreground steps must be admitted independently. Model-invoked task
    /// tools inherit the enclosing run's lease and leave this false.
    schedule_foreground: bool,
    /// Transient run/host scope used to derive the owner quota. Only the
    /// digest-derived identity crosses into the scheduler actor.
    admission_scope: Option<String>,
    /// Provider/model capacity projected into the same scheduler actor as
    /// owner admission. This is metadata only; the scheduler remains the
    /// single live reservation authority.
    provider_quota: Option<crate::task_scheduler::TaskSchedulerQuota>,
    /// Shared provider admission for foreground child runs. Background runs
    /// already hold `provider_quota` on their outer scheduler lease and use a
    /// local child gate to avoid recursively reserving that same dimension.
    provider_admission: Option<crate::llm::ModelGenerationAdmission>,
}

impl TaskExecutor {
    /// Create a new task executor
    pub fn new(
        registry: Arc<AgentRegistry>,
        llm_client: Arc<dyn LlmClient>,
        workspace: String,
    ) -> Self {
        let provider_quota = provider_quota_for_client(llm_client.as_ref());
        Self {
            registry,
            llm_client,
            workspace,
            mcp_managers: Vec::new(),
            mcp_bindings: Vec::new(),
            child_tool_presentation: None,
            scoped_tools: Vec::new(),
            parent_context: None,
            search_config: None,
            search_bulkhead: None,
            search_retry_budget: None,
            search_request_coalescer: None,
            capability_context: None,
            parent_cancellation: None,
            max_parallel_tasks: crate::agent::DEFAULT_MAX_PARALLEL_TASKS,
            parallel_permits: Arc::new(tokio::sync::Semaphore::new(
                crate::agent::DEFAULT_MAX_PARALLEL_TASKS,
            )),
            subagent_tracker: None,
            task_scheduler: None,
            schedule_foreground: false,
            admission_scope: None,
            provider_quota,
            provider_admission: None,
        }
    }

    /// Create a new task executor with MCP manager for tool inheritance
    pub fn with_mcp(
        registry: Arc<AgentRegistry>,
        llm_client: Arc<dyn LlmClient>,
        workspace: String,
        mcp_manager: Arc<McpManager>,
    ) -> Self {
        Self::with_mcp_managers(registry, llm_client, workspace, vec![mcp_manager])
    }

    /// Create a task executor with ordered MCP capability sources.
    pub fn with_mcp_managers(
        registry: Arc<AgentRegistry>,
        llm_client: Arc<dyn LlmClient>,
        workspace: String,
        mcp_managers: Vec<Arc<McpManager>>,
    ) -> Self {
        let provider_quota = provider_quota_for_client(llm_client.as_ref());
        Self {
            registry,
            llm_client,
            workspace,
            mcp_managers,
            mcp_bindings: Vec::new(),
            child_tool_presentation: None,
            scoped_tools: Vec::new(),
            parent_context: None,
            search_config: None,
            search_bulkhead: None,
            search_retry_budget: None,
            search_request_coalescer: None,
            capability_context: None,
            parent_cancellation: None,
            max_parallel_tasks: crate::agent::DEFAULT_MAX_PARALLEL_TASKS,
            parallel_permits: Arc::new(tokio::sync::Semaphore::new(
                crate::agent::DEFAULT_MAX_PARALLEL_TASKS,
            )),
            subagent_tracker: None,
            task_scheduler: None,
            schedule_foreground: false,
            admission_scope: None,
            provider_quota,
            provider_admission: None,
        }
    }

    /// Add immutable MCP bindings already admitted by the parent capability
    /// Run. These bindings are never rediscovered through a manager.
    pub(crate) fn with_projected_mcp_bindings(mut self, bindings: Vec<Arc<McpBinding>>) -> Self {
        self.mcp_bindings = bindings;
        self
    }

    /// Force a Tool presentation profile on delegated children.
    #[cfg(test)]
    pub(crate) fn with_child_tool_presentation(
        mut self,
        profile: crate::tools::ToolPresentationProfileV1,
    ) -> Self {
        self.child_tool_presentation = Some(profile);
        self
    }

    /// Install exact host-provided tools only in child runs created by this
    /// executor. Registration does not grant invocation authority: every call
    /// still crosses composed parent and child governance. A name collision
    /// with another child capability fails before model execution.
    pub fn with_scoped_tools(mut self, tools: Vec<Arc<dyn Tool>>) -> Self {
        self.scoped_tools = tools;
        self
    }

    /// Set parent session capabilities to inherit into child runs.
    pub fn with_parent_context(mut self, ctx: crate::child_run::ChildRunContext) -> Self {
        if let Some(max_parallel_tasks) = ctx.max_parallel_tasks {
            let max_parallel_tasks = max_parallel_tasks.max(1);
            self.max_parallel_tasks = max_parallel_tasks;
            self.parallel_permits = Arc::new(tokio::sync::Semaphore::new(max_parallel_tasks));
        }
        self.parent_context = Some(ctx);
        self
    }

    fn scoped_for_invocation(self: &Arc<Self>, ctx: &ToolContext) -> Arc<Self> {
        let mut scoped = self.as_ref().clone();
        scoped.search_config = ctx.search_config.clone();
        scoped.search_bulkhead = Some(ctx.search_bulkhead());
        scoped.search_retry_budget = Some(ctx.search_retry_budget());
        scoped.search_request_coalescer = Some(ctx.search_request_coalescer());
        scoped.capability_context = ctx.capability_context();
        scoped.admission_scope = ctx
            .run_id()
            .map(|run_id| format!("run:{run_id}"))
            .or_else(|| ctx.session_id.as_deref().map(|id| format!("session:{id}")));
        if ctx.has_run_governance() {
            scoped.parent_context = scoped.parent_context.take().map(|parent| {
                parent.with_run_governance(
                    ctx.run_permission_checker(),
                    ctx.run_confirmation_manager(),
                )
            });
        }
        Arc::new(scoped)
    }

    fn child_tool_context(
        &self,
        session_id: String,
        cancellation: CancellationToken,
    ) -> ToolContext {
        let mut context = ToolContext::new(PathBuf::from(&self.workspace))
            .with_session_id(session_id)
            .with_cancellation(cancellation);
        if let (Some(bulkhead), Some(retry_budget)) =
            (&self.search_bulkhead, &self.search_retry_budget)
        {
            context = context.with_search_runtime(bulkhead.clone(), retry_budget.clone());
        }
        if let Some(search_config) = &self.search_config {
            context = context.with_search_config(search_config.as_ref().clone());
        }
        if let Some(coalescer) = &self.search_request_coalescer {
            context = context.with_search_request_coalescer(coalescer.clone());
        }
        context
    }

    /// Bind every run started by this executor to a parent lifetime.
    ///
    /// A token that is already cancelled makes execution fail before emitting
    /// `SubagentStart` or performing MCP/LLM work. In-flight children derive
    /// their own token so cancellation still cascades without granting them the
    /// ability to cancel the parent.
    pub fn with_parent_cancellation(mut self, cancellation: CancellationToken) -> Self {
        self.parent_cancellation = Some(cancellation);
        self
    }

    pub fn with_max_parallel_tasks(mut self, max_parallel_tasks: usize) -> Self {
        let max_parallel_tasks = max_parallel_tasks.max(1);
        self.max_parallel_tasks = max_parallel_tasks;
        self.parallel_permits = Arc::new(tokio::sync::Semaphore::new(max_parallel_tasks));
        self
    }

    /// Share a tracker with this executor. When set, each task registers
    /// a `CancellationToken` against the tracker so the parent session
    /// can cancel by `task_id`.
    pub fn with_subagent_tracker(
        mut self,
        tracker: Arc<crate::subagent_task_tracker::InMemorySubagentTaskTracker>,
    ) -> Self {
        self.subagent_tracker = Some(tracker);
        self
    }

    /// Admit independent delegated work through the owning agent's scheduler.
    pub fn with_task_scheduler(
        mut self,
        scheduler: Arc<crate::task_scheduler::TaskScheduler>,
        schedule_foreground: bool,
    ) -> Self {
        // OPT-POOL1: shared provider capacity requires a typed
        // `ModelGenerationPool`. Without one, children keep a local-only gate
        // and must not attach a scheduler quota that would look like a product
        // shared pool.
        self.provider_admission = self.llm_client.model_generation_pool().and_then(|pool| {
            crate::llm::ModelGenerationAdmission::new(
                self.llm_client.model_generation_concurrency(),
            )
            .with_model_generation_pool(
                Arc::clone(&scheduler),
                pool,
                crate::task_scheduler::TaskPriority::Foreground,
                "task-child-model-generation",
            )
            .ok()
        });
        self.task_scheduler = Some(scheduler);
        self.schedule_foreground = schedule_foreground;
        self
    }

    #[cfg(test)]
    pub(crate) fn has_provider_model_generation_admission(&self) -> bool {
        self.provider_admission.is_some()
    }

    #[cfg(test)]
    pub(crate) fn provider_admission_publishes_typed_pool(&self) -> bool {
        self.provider_admission
            .as_ref()
            .is_some_and(|admission| admission.publishes_model_generation_pool())
    }

    fn visible_agents(&self) -> Vec<AgentDefinition> {
        self.registry.list_visible()
    }

    /// Execute a task by spawning an isolated child AgentLoop.
    ///
    /// `parent_session_id` flows into the emitted `SubagentStart`/`SubagentEnd`
    /// events so dashboards can associate child runs with the parent session.
    pub async fn execute(
        &self,
        params: TaskParams,
        event_tx: Option<broadcast::Sender<AgentEvent>>,
        parent_session_id: Option<&str>,
    ) -> Result<TaskResult> {
        self.execute_with_parent_cancellation(
            params,
            event_tx,
            parent_session_id,
            self.parent_cancellation.as_ref(),
        )
        .await
    }

    async fn execute_with_parent_cancellation(
        &self,
        params: TaskParams,
        event_tx: Option<broadcast::Sender<AgentEvent>>,
        parent_session_id: Option<&str>,
        parent_cancellation: Option<&CancellationToken>,
    ) -> Result<TaskResult> {
        let task_id = format!("task-{}", uuid::Uuid::new_v4());
        self.execute_with_task_id_scoped(
            task_id,
            params,
            ScopedTaskExecution {
                event_tx,
                parent_session_id,
                emit_start: true,
                parent_cancellation,
                admitted_capability_subtask: None,
                parallel_lifecycle: None,
            },
        )
        .await
    }

    /// Execute a task using a caller-supplied task id. Used by `execute_background`
    /// so the synchronously-returned task id matches the one in lifecycle events.
    /// When `emit_start` is `false` the caller is responsible for emitting
    /// `SubagentStart` themselves (e.g. to avoid a race against a tracker query).
    pub async fn execute_with_task_id(
        &self,
        task_id: String,
        params: TaskParams,
        event_tx: Option<broadcast::Sender<AgentEvent>>,
        parent_session_id: Option<&str>,
        emit_start: bool,
    ) -> Result<TaskResult> {
        self.execute_with_task_id_scoped(
            task_id,
            params,
            ScopedTaskExecution {
                event_tx,
                parent_session_id,
                emit_start,
                parent_cancellation: self.parent_cancellation.as_ref(),
                admitted_capability_subtask: None,
                parallel_lifecycle: None,
            },
        )
        .await
    }

    async fn execute_with_task_id_scoped(
        &self,
        task_id: String,
        params: TaskParams,
        execution: ScopedTaskExecution<'_>,
    ) -> Result<TaskResult> {
        let ScopedTaskExecution {
            event_tx,
            parent_session_id,
            emit_start,
            parent_cancellation,
            admitted_capability_subtask,
            parallel_lifecycle,
        } = execution;
        let was_promoted = admitted_capability_subtask.is_some();
        if !was_promoted && parent_cancellation.is_some_and(CancellationToken::is_cancelled) {
            anyhow::bail!("Operation cancelled by parent session");
        }

        let capability_subtask = match admitted_capability_subtask {
            Some(subtask) => Some(subtask),
            None => self
                .capability_context
                .as_ref()
                .map(|context| context.admit_subtask(task_id.clone(), params.background))
                .transpose()?,
        };
        let cancel_token = capability_subtask.as_ref().map_or_else(
            || {
                parent_cancellation
                    .map(CancellationToken::child_token)
                    .unwrap_or_default()
            },
            crate::capability::AgentCapabilitySubtask::cancellation,
        );
        let capability_runtime = capability_subtask
            .as_ref()
            .map(crate::capability::AgentCapabilitySubtask::runtime);
        let execution = self
            .execute_with_task_id_in_scope(
                task_id,
                params,
                event_tx,
                parent_session_id,
                emit_start,
                cancel_token,
                capability_runtime,
                parallel_lifecycle,
            )
            .await;
        let close = close_capability_subtask(capability_subtask.as_ref()).await;
        match (execution, close) {
            (Ok(result), Ok(())) => Ok(result),
            (Ok(_), Err(close_error)) => Err(close_error),
            (Err(error), Ok(())) => Err(error),
            (Err(error), Err(close_error)) => {
                tracing::warn!(
                    error = %close_error,
                    "Capability Subtask close also failed after delegated execution failure"
                );
                Err(error)
            }
        }
    }

    #[allow(clippy::too_many_arguments)]
    async fn execute_with_task_id_in_scope(
        &self,
        task_id: String,
        params: TaskParams,
        event_tx: Option<broadcast::Sender<AgentEvent>>,
        parent_session_id: Option<&str>,
        emit_start: bool,
        cancel_token: CancellationToken,
        capability_runtime: Option<crate::capability::AgentCapabilityRuntime>,
        parallel_lifecycle: Option<Arc<ParallelTaskLifecycle>>,
    ) -> Result<TaskResult> {
        // Background callers receive the task id before this future starts.
        // Register immediately so targeted cancellation also interrupts time
        // spent waiting in the global scheduler.
        if params.background {
            if let Some(ref tracker) = self.subagent_tracker {
                tracker
                    .register_canceller(&task_id, cancel_token.clone())
                    .await;
            }
        }
        let execution_identity =
            if (params.background || self.schedule_foreground) && self.task_scheduler.is_some() {
                let mut identity_spec = AgentStepSpec::new(
                    task_id.clone(),
                    params.agent.clone(),
                    params.description.clone(),
                    params.prompt.clone(),
                );
                if let Some(max_steps) = params.max_steps {
                    identity_spec = identity_spec.with_max_steps(max_steps);
                }
                if let Some(output_schema) = params.output_schema.clone() {
                    identity_spec = identity_spec.with_output_schema(output_schema);
                }
                if let Some(parent_session_id) = parent_session_id {
                    identity_spec = identity_spec.with_parent_session_id(parent_session_id);
                }
                Some(
                    crate::orchestration::workflow_step_execution_identity(
                        parent_session_id.unwrap_or("host"),
                        &identity_spec,
                    )
                    .map_err(|error| {
                        anyhow::anyhow!("derive delegated task execution identity: {error}")
                    })?,
                )
            } else {
                None
            };
        let admission_quota =
            if (params.background || self.schedule_foreground) && self.task_scheduler.is_some() {
                let scope = self.admission_scope.clone().unwrap_or_else(|| {
                    parent_session_id
                        .map(|session_id| format!("session:{session_id}"))
                        .unwrap_or_else(|| "host".to_string())
                });
                Some(
                    crate::task_scheduler::TaskSchedulerQuota::for_scope(
                        &scope,
                        self.max_parallel_tasks,
                    )
                    .map_err(|error| anyhow::anyhow!(error))?,
                )
            } else {
                None
            };
        let _task_lease = if params.background || self.schedule_foreground {
            match &self.task_scheduler {
                Some(scheduler) => {
                    let mut quotas = Vec::with_capacity(2);
                    if let Some(quota) = admission_quota.as_ref() {
                        quotas.push(quota.clone());
                    }
                    if let Some(quota) = self.provider_quota.as_ref() {
                        if !quotas
                            .iter()
                            .any(|candidate| candidate.identity == quota.identity)
                        {
                            quotas.push(quota.clone());
                        }
                    }
                    let priority = if params.background {
                        crate::task_scheduler::TaskPriority::Background
                    } else {
                        crate::task_scheduler::TaskPriority::Foreground
                    };
                    let label = format!(
                        "{}:subagent:{}",
                        parent_session_id.unwrap_or("host"),
                        task_id
                    );
                    Some(if quotas.is_empty() {
                        scheduler
                            .acquire_with_identity(
                                priority,
                                label,
                                execution_identity.clone(),
                                &cancel_token,
                            )
                            .await
                            .map_err(|error| anyhow::anyhow!(error))?
                    } else {
                        scheduler
                            .acquire_with_quotas(
                                priority,
                                label,
                                &quotas,
                                execution_identity.clone(),
                                &cancel_token,
                            )
                            .await
                            .map_err(|error| anyhow::anyhow!(error))?
                    })
                }
                None => None,
            }
        } else {
            None
        };

        let session_id = format!("task-run-{}", task_id);
        let started_ms = epoch_ms();
        let output_schema = params.output_schema.clone();

        let agent = self
            .registry
            .get_arc(&params.agent)
            .context(format!("Unknown agent type: '{}'", params.agent))?;
        let tool_free = agent.tool_free;
        let tool_free_system = agent.prompt.clone();
        let inherited_security_provider = self
            .parent_context
            .as_ref()
            .and_then(|context| context.security_provider.clone());

        if emit_start {
            let event = AgentEvent::SubagentStart {
                task_id: task_id.clone(),
                session_id: session_id.clone(),
                parent_session_id: parent_session_id.unwrap_or_default().to_string(),
                agent: params.agent.clone(),
                description: params.description.clone(),
                started_ms,
            };
            let event = inherited_security_provider
                .as_deref()
                .map(|provider| crate::security::sanitize_agent_event(provider, &event))
                .unwrap_or(event);
            if let Some(ref tracker) = self.subagent_tracker {
                tracker.record_event(&event).await;
            }
            if let Some(ref tx) = event_tx {
                let _ = tx.send(event);
            }
            if let Some(lifecycle) = &parallel_lifecycle {
                // Keep this after the last await and after the broadcast so an
                // abort can never suppress a start that the lifecycle thinks
                // was emitted.
                lifecycle.mark_started(&task_id);
            }
        }

        // Build a child ToolExecutor. Task tools are intentionally omitted
        // here to prevent unlimited delegation nesting.
        let child_executor = if let Some(ref parent_ctx) = self.parent_context {
            if let Some(ref services) = parent_ctx.workspace_services {
                crate::tools::ToolExecutor::new_with_workspace_services_artifact_limits_and_immutable_content_adapter(
                        self.workspace.clone(),
                        Arc::clone(services),
                        crate::tools::ArtifactStoreLimits::default(),
                        parent_ctx.immutable_content_adapter.clone(),
                    )
            } else if let Some(adapter) = parent_ctx.immutable_content_adapter.clone() {
                crate::tools::ToolExecutor::new_with_immutable_content_adapter(
                    self.workspace.clone(),
                    adapter,
                )
            } else {
                crate::tools::ToolExecutor::new(self.workspace.clone())
            }
        } else {
            crate::tools::ToolExecutor::new(self.workspace.clone())
        };

        // Register MCP tools so child agents can access MCP servers.
        // When the parent Run already projected exact McpBindings, those
        // bindings are the routing authority (OPT-MCP1). Manager snapshots are
        // mutable refresh caches and must not inject unbound tools into the
        // delegated child.
        // When the parent Run already projected exact McpBindings, those
        // bindings are the routing authority (OPT-MCP1). Manager snapshots are
        // mutable refresh caches and must not inject unbound tools into the
        // delegated child.
        if self.mcp_bindings.is_empty() {
            for mcp in &self.mcp_managers {
                let all_tools = tokio::select! {
                    biased;
                    _ = cancel_token.cancelled() => {
                        anyhow::bail!("Operation cancelled before child execution");
                    }
                    tools = mcp.get_all_tools() => tools,
                };
                let mut by_server: std::collections::HashMap<
                    String,
                    Vec<crate::mcp::protocol::McpTool>,
                > = std::collections::HashMap::new();
                for (server, tool) in all_tools {
                    by_server.entry(server).or_default().push(tool);
                }
                for (server_name, tools) in by_server {
                    let wrappers =
                        crate::mcp::tools::create_mcp_tools(&server_name, tools, Arc::clone(mcp));
                    for wrapper in wrappers {
                        child_executor.register_dynamic_tool(wrapper);
                    }
                }
            }
        }
        // Projected bindings are the Run-frozen MCP generation.
        for binding in &self.mcp_bindings {
            if cancel_token.is_cancelled() {
                anyhow::bail!("Operation cancelled before child execution");
            }
            for wrapper in binding.projected_tools() {
                child_executor.register_dynamic_tool(wrapper);
            }
        }

        // These exact Arc values belong only to this TaskExecutor. Install them
        // after all inherited sources so no scoped tool can shadow a built-in,
        // compatibility MCP tool, or Run-frozen projected binding.
        for tool in &self.scoped_tools {
            if !child_executor.register_dynamic_tool_if_absent(Arc::clone(tool)) {
                anyhow::bail!(
                    "Workflow-scoped tool '{}' conflicts with another child capability",
                    tool.name()
                );
            }
        }

        let child_executor = Arc::new(child_executor);

        let mut child_config = AgentConfig {
            tools: child_executor.definitions(),
            ..AgentConfig::default()
        };
        agent.apply_to(&mut child_config);
        if let Some(ref parent_ctx) = self.parent_context {
            parent_ctx.apply_to(&mut child_config);
        }
        if let Some(profile) = self.child_tool_presentation.clone() {
            child_config.tool_presentation_profile = profile;
        }
        // A delegated task is already the output of a parent planning
        // decision. Running the generic pre-analysis/planning classifier again
        // adds an unrelated LLM round to every child and can consume the whole
        // fan-out deadline before any task tool runs.
        child_config.planning_mode = crate::prompts::PlanningMode::Disabled;
        if let Some(max_steps) = params.max_steps {
            child_config.max_tool_rounds = max_steps;
        }
        let child_security_provider = child_config.security_provider.clone();
        let source_security_provider = child_security_provider.clone();

        let mut tool_context = self.child_tool_context(session_id.clone(), cancel_token.clone());
        if let Some(ref parent_ctx) = self.parent_context {
            if let Some(ref services) = parent_ctx.workspace_services {
                tool_context = tool_context.with_workspace_services(Arc::clone(services));
            }
            if let Some(ref sandbox) = parent_ctx.sandbox_handle {
                child_executor.registry().set_sandbox(Arc::clone(sandbox));
                tool_context = tool_context.with_sandbox(Arc::clone(sandbox));
            }
        }

        let source_context = tool_context.clone();
        let mut agent_loop = AgentLoop::new(
            Arc::clone(&self.llm_client),
            child_executor,
            tool_context,
            child_config,
        );
        if !params.background && !self.schedule_foreground {
            if let Some(admission) = &self.provider_admission {
                agent_loop = agent_loop.with_model_generation_admission(admission.clone());
            }
        }
        if let Some(runtime) = capability_runtime {
            agent_loop = agent_loop.with_capability_runtime(runtime);
        }

        // Always observe the child event stream so successful source tool calls
        // survive in TaskResult metadata even when nobody subscribed to live
        // progress. Forward the same events when a parent broadcast exists.
        let (mpsc_tx, mut mpsc_rx) = tokio::sync::mpsc::channel(100);
        let broadcast_tx = event_tx.clone();
        let progress_task_id = task_id.clone();
        let progress_session_id = session_id.clone();
        let child_event_forwarder = tokio::spawn(async move {
            let mut source_anchors = Vec::new();
            let mut seen_source_anchors = std::collections::HashSet::new();
            let mut scanned_source_candidates = 0usize;
            while let Some(event) = mpsc_rx.recv().await {
                let event = source_security_provider
                    .as_deref()
                    .map(|provider| crate::security::sanitize_agent_event(provider, &event))
                    .unwrap_or(event);
                collect_tool_source_anchors(
                    &event,
                    &source_context,
                    &mut source_anchors,
                    &mut seen_source_anchors,
                    &mut scanned_source_candidates,
                );
                if let Some(ref broadcast_tx) = broadcast_tx {
                    if let Some(progress) = synthesize_subagent_progress(
                        &event,
                        &progress_task_id,
                        &progress_session_id,
                    ) {
                        let _ = broadcast_tx.send(progress);
                    }
                    let _ = broadcast_tx.send(event);
                }
            }
            source_anchors
        });
        let child_event_tx = Some(mpsc_tx);
        let child_llm_event_tx = child_event_tx.clone();

        // Register a CancellationToken with the tracker (if shared) so the
        // parent session's `cancel_subagent_task` can interrupt this run.
        if !params.background {
            if let Some(ref tracker) = self.subagent_tracker {
                tracker
                    .register_canceller(&task_id, cancel_token.clone())
                    .await;
            }
        }

        let structured_prompt = output_schema
            .as_ref()
            .filter(|_| !tool_free)
            .map(|schema| structured_task_prompt(&params.prompt, schema));
        let execution_prompt = structured_prompt.as_deref().unwrap_or(&params.prompt);

        let mut structured = None;
        let (mut output, mut success, raw_output) = if tool_free && output_schema.is_some() {
            let operation = agent_loop.begin_capability_operation(
                0,
                &cancel_token,
                "structured task generation",
            )?;
            let llm_client = agent_loop.scoped_llm_client_for_parts(
                Some(&session_id),
                &child_llm_event_tx,
                operation.cancellation(),
            );
            let generation = Self::generate_structured_task(
                &*llm_client,
                &params.prompt,
                tool_free_system.as_deref(),
                output_schema.clone().expect("schema checked above"),
                operation.cancellation(),
            )
            .await;
            let generation = settle_task_capability_operation(
                generation,
                operation.close().await,
                "structured task generation",
            );
            match generation {
                Ok(object) => {
                    let output = serde_json::to_string_pretty(&object)
                        .unwrap_or_else(|_| object.to_string());
                    structured = Some(object);
                    (output, true, None)
                }
                Err(error) if cancel_token.is_cancelled() => {
                    (format!("Task cancelled by caller: {error}"), false, None)
                }
                Err(error) => (format!("Task failed: {error}"), false, None),
            }
        } else {
            match agent_loop
                .execute_with_session(
                    &[],
                    execution_prompt,
                    Some(&session_id),
                    child_event_tx.clone(),
                    Some(&cancel_token),
                )
                .await
            {
                Ok(_) if cancel_token.is_cancelled() => {
                    ("Task cancelled by caller".to_string(), false, None)
                }
                Ok(result) if result.text.trim().is_empty() => (
                    "Task failed: child agent returned no final output".to_string(),
                    false,
                    None,
                ),
                Ok(result) if AgentLoop::is_synthetic_failure_output(&result.text) => {
                    (format!("Task failed: {}", result.text), false, None)
                }
                Ok(result) => {
                    let raw_output = result
                        .messages
                        .last()
                        .filter(|message| message.role == "assistant")
                        .map(crate::llm::Message::text)
                        .filter(|text| !text.trim().is_empty());
                    (result.text, true, raw_output)
                }
                Err(e) if cancel_token.is_cancelled() => {
                    (format!("Task cancelled by caller: {}", e), false, None)
                }
                Err(e) => (format!("Task failed: {}", e), false, None),
            }
        };

        if success && !tool_free {
            if let Some(schema) = output_schema.as_ref() {
                if let Some(object) = raw_output
                    .as_deref()
                    .and_then(|raw| parse_validated_output(raw, schema))
                    .or_else(|| parse_validated_output(&output, schema))
                {
                    structured = Some(object);
                } else {
                    let operation = agent_loop.begin_capability_operation(
                        0,
                        &cancel_token,
                        "structured task coercion",
                    )?;
                    let llm_client = agent_loop.scoped_llm_client_for_parts(
                        Some(&session_id),
                        &child_llm_event_tx,
                        operation.cancellation(),
                    );
                    let coercion = Self::coerce_to_schema(
                        &*llm_client,
                        &output,
                        schema.clone(),
                        operation.cancellation(),
                    )
                    .await;
                    let coercion = settle_task_capability_operation(
                        coercion,
                        operation.close().await,
                        "structured task coercion",
                    );
                    match coercion {
                        Ok(object) => structured = Some(object),
                        Err(error) => {
                            success = false;
                            output = format!("{output}\n\n[structured output failed: {error}]");
                        }
                    }
                }
            }
        }
        if let Some(provider) = child_security_provider.as_deref() {
            output = crate::security::sanitize_text(provider, &output);
            if let Some(value) = structured.take() {
                let sanitized = output_schema.as_ref().map_or_else(
                    || sanitize_task_json(provider, &value),
                    |schema| sanitize_task_json_with_schema(provider, &value, schema),
                );
                if output_schema
                    .as_ref()
                    .is_none_or(|schema| value_matches_schema(&sanitized, schema))
                {
                    structured = Some(sanitized);
                } else {
                    success = false;
                }
            }
        }

        // The child loop and optional structured-output pass are the only
        // producers. Close their sender and drain the bridge before emitting
        // SubagentEnd so callers never observe a terminal event followed by
        // stale child deltas or progress events.
        drop(child_event_tx);
        drop(child_llm_event_tx);
        let source_anchors = match child_event_forwarder.await {
            Ok(source_anchors) => source_anchors,
            Err(error) => {
                tracing::warn!(%error, task_id = %task_id, "subagent event bridge failed");
                Vec::new()
            }
        };

        let end_event = AgentEvent::SubagentEnd {
            task_id: task_id.clone(),
            session_id: session_id.clone(),
            agent: params.agent.clone(),
            output: output.clone(),
            success,
            finished_ms: epoch_ms(),
        };
        if let Some(ref tracker) = self.subagent_tracker {
            // The tracker is authoritative even when a background child
            // finishes after the parent run's event forwarder has closed.
            if success {
                tracker
                    .record_source_anchors(&task_id, &source_anchors)
                    .await;
            }
            tracker.record_event(&end_event).await;
            tracker.clear_canceller(&task_id).await;
        }
        if let Some(ref tx) = event_tx {
            let _ = tx.send(end_event);
        }
        if let Some(lifecycle) = &parallel_lifecycle {
            // Keep this as the final synchronous operation. If the child was
            // aborted at an earlier await, synthetic cleanup can still fill in
            // the missing terminal event.
            lifecycle.mark_ended(&task_id);
        }

        Ok(TaskResult {
            output,
            session_id,
            agent: params.agent,
            success,
            task_id,
            structured,
            source_anchors,
        })
    }

    /// Execute a task in the background.
    ///
    /// Returns immediately with the task ID; the same id is used in the emitted
    /// `SubagentStart`/`SubagentEnd` events so callers can correlate. Pre-emits
    /// `SubagentStart` synchronously when an event channel is available so a
    /// caller that queries the subagent task tracker right after this call
    /// observes the task in `Running` state without a race window.
    pub fn execute_background(
        self: Arc<Self>,
        params: TaskParams,
        event_tx: Option<broadcast::Sender<AgentEvent>>,
        parent_session_id: Option<String>,
    ) -> String {
        let parent_cancellation = self.parent_cancellation.clone();
        self.execute_background_with_parent_cancellation(
            params,
            event_tx,
            parent_session_id,
            parent_cancellation,
        )
    }

    fn execute_background_with_parent_cancellation(
        self: Arc<Self>,
        params: TaskParams,
        event_tx: Option<broadcast::Sender<AgentEvent>>,
        parent_session_id: Option<String>,
        parent_cancellation: Option<CancellationToken>,
    ) -> String {
        let task_id = format!("task-{}", uuid::Uuid::new_v4());
        let session_id = format!("task-run-{}", task_id);
        let failure_session_id = session_id.clone();
        let failure_agent = params.agent.clone();
        let start_event = AgentEvent::SubagentStart {
            task_id: task_id.clone(),
            session_id,
            parent_session_id: parent_session_id.clone().unwrap_or_default(),
            agent: params.agent.clone(),
            description: params.description.clone(),
            started_ms: epoch_ms(),
        };
        let security_provider = self
            .parent_context
            .as_ref()
            .and_then(|context| context.security_provider.clone());
        let start_event = security_provider
            .as_deref()
            .map(|provider| crate::security::sanitize_agent_event(provider, &start_event))
            .unwrap_or(start_event);

        if let Some(ref tx) = event_tx {
            let _ = tx.send(start_event.clone());
        }

        let capability_admission = self
            .capability_context
            .as_ref()
            .map(|context| {
                context
                    .admit_subtask(task_id.clone(), true)
                    .map(|subtask| (context.background_scope().clone(), subtask))
            })
            .transpose();
        let (capability_run, admitted_capability_subtask) = match capability_admission {
            Ok(Some((run, subtask))) => (Some(run), Some(subtask)),
            Ok(None) => (None, None),
            Err(error) => {
                let message = format!("Background task capability admission failed: {error}");
                let end_event = AgentEvent::SubagentEnd {
                    task_id: task_id.clone(),
                    session_id: failure_session_id,
                    agent: failure_agent,
                    output: message.clone(),
                    success: false,
                    finished_ms: epoch_ms(),
                };
                let end_event = security_provider
                    .as_deref()
                    .map(|provider| crate::security::sanitize_agent_event(provider, &end_event))
                    .unwrap_or(end_event);
                if let Some(tx) = event_tx {
                    let _ = tx.send(end_event);
                }
                tracing::error!(task_id = %task_id, "{message}");
                return task_id;
            }
        };

        let task_id_for_spawn = task_id.clone();
        let task_id_for_log = task_id.clone();
        let admission_failure_task_id = task_id.clone();
        let admission_failure_session_id = failure_session_id.clone();
        let admission_failure_agent = failure_agent.clone();
        let admission_failure_events = event_tx.clone();
        let admission_failure_security = security_provider.clone();
        let background = async move {
            if let Some(ref tracker) = self.subagent_tracker {
                tracker.record_event(&start_event).await;
            }
            let failure_event_tx = event_tx.clone();
            if let Err(error) = self
                .execute_with_task_id_scoped(
                    task_id_for_spawn,
                    params,
                    ScopedTaskExecution {
                        event_tx,
                        parent_session_id: parent_session_id.as_deref(),
                        emit_start: false,
                        parent_cancellation: parent_cancellation.as_ref(),
                        admitted_capability_subtask,
                        parallel_lifecycle: None,
                    },
                )
                .await
            {
                let end_event = AgentEvent::SubagentEnd {
                    task_id: task_id_for_log.clone(),
                    session_id: failure_session_id,
                    agent: failure_agent,
                    output: format!("Task failed before child execution started: {error}"),
                    success: false,
                    finished_ms: epoch_ms(),
                };
                let end_event = security_provider
                    .as_deref()
                    .map(|provider| crate::security::sanitize_agent_event(provider, &end_event))
                    .unwrap_or(end_event);
                if let Some(ref tracker) = self.subagent_tracker {
                    tracker.record_event(&end_event).await;
                    tracker.clear_canceller(&task_id_for_log).await;
                }
                if let Some(tx) = failure_event_tx {
                    let _ = tx.send(end_event);
                }
                tracing::error!("Background task {} failed: {}", task_id_for_log, error);
            }
        };
        if let Some(run) = capability_run {
            let task_name = format!("subagent.{task_id}");
            if let Err(error) = run.spawn_task(task_name, async move {
                background.await;
                Ok(())
            }) {
                // Admission races with Run close fail closed: no detached
                // child work may escape after the exact generation lease is
                // released.
                let message = format!("Background task capability admission failed: {error}");
                let end_event = AgentEvent::SubagentEnd {
                    task_id: admission_failure_task_id.clone(),
                    session_id: admission_failure_session_id,
                    agent: admission_failure_agent,
                    output: message.clone(),
                    success: false,
                    finished_ms: epoch_ms(),
                };
                let end_event = admission_failure_security
                    .as_deref()
                    .map(|provider| crate::security::sanitize_agent_event(provider, &end_event))
                    .unwrap_or(end_event);
                if let Some(tx) = admission_failure_events {
                    let _ = tx.send(end_event);
                }
                tracing::error!(task_id = %admission_failure_task_id, "{message}");
            }
        } else {
            tokio::spawn(background);
        }

        task_id
    }
}

async fn close_capability_subtask(
    subtask: Option<&crate::capability::AgentCapabilitySubtask>,
) -> Result<()> {
    let Some(subtask) = subtask else {
        return Ok(());
    };
    let report = subtask.close().await?;
    if !report.is_clean() {
        anyhow::bail!(
            "Capability Subtask close was incomplete (tasks failed: {}, tasks timed out: {}, child scopes failed: {}, child scopes timed out: {}, effects failed: {}, effects timed out: {})",
            report.tasks_failed,
            report.tasks_timed_out,
            report.child_scopes_failed,
            report.child_scopes_timed_out,
            report.effects_failed,
            report.effects_timed_out,
        );
    }
    Ok(())
}

fn settle_task_capability_operation<T>(
    execution: Result<T>,
    close: Result<()>,
    label: &str,
) -> Result<T> {
    match (execution, close) {
        (Ok(result), Ok(())) => Ok(result),
        (Ok(_), Err(close_error)) => Err(close_error),
        (Err(error), Ok(())) => Err(error),
        (Err(error), Err(close_error)) => {
            tracing::warn!(
                error = %close_error,
                operation = label,
                "Capability orchestration Turn close also failed after model failure"
            );
            Err(error)
        }
    }
}

fn structured_task_prompt(prompt: &str, schema: &serde_json::Value) -> String {
    let schema = serde_json::to_string_pretty(schema).unwrap_or_else(|_| schema.to_string());
    format!(
        "{prompt}\n\n\
         FINAL OUTPUT CONTRACT\n\
         Complete the requested investigation before answering. Your final response must contain \
         exactly one JSON value matching the JSON Schema below, with no Markdown fence or prose \
         outside the JSON. This contract applies to the final response only; use the available \
         tools as needed before finalizing.\n\n\
         {schema}"
    )
}

fn value_matches_schema(value: &serde_json::Value, schema: &serde_json::Value) -> bool {
    serde_json::to_string(value)
        .ok()
        .and_then(|encoded| parse_validated_output(&encoded, schema))
        .is_some()
}

#[derive(Debug, Clone)]
struct AgentCatalogEntry {
    name: String,
    description: String,
}

fn agent_catalog_entries(agents: &[AgentDefinition]) -> Vec<AgentCatalogEntry> {
    let mut entries = agents
        .iter()
        .map(|agent| AgentCatalogEntry {
            name: agent.name.clone(),
            description: agent
                .description
                .split_whitespace()
                .collect::<Vec<_>>()
                .join(" "),
        })
        .collect::<Vec<_>>();
    entries.sort_by(|left, right| left.name.cmp(&right.name));
    entries
}

fn agent_catalog_text(agents: &[AgentDefinition]) -> String {
    agent_catalog_entries(agents)
        .into_iter()
        .map(|entry| format!("{}: {}", entry.name, entry.description))
        .collect::<Vec<_>>()
        .join("\n")
}

fn delegation_tool_description(base: &str, agents: &[AgentDefinition]) -> String {
    format!(
        "{base}\n\nAvailable agents (live catalog; use canonical names):\n{}",
        agent_catalog_text(agents)
    )
}

pub(super) fn task_agent_parameter_schema(agents: &[AgentDefinition]) -> serde_json::Value {
    let entries = agent_catalog_entries(agents);
    let examples = entries
        .iter()
        .map(|entry| serde_json::Value::String(entry.name.clone()))
        .collect::<Vec<_>>();
    let catalog = entries
        .into_iter()
        .map(|entry| format!("{}: {}", entry.name, entry.description))
        .collect::<Vec<_>>()
        .join("\n");
    serde_json::json!({
        "type": "string",
        "description": format!(
            "Required. Canonical agent type to use. Always provide this exact field name: 'agent'. Live agent catalog:\n{catalog}"
        ),
        "examples": examples
    })
}

/// Get the compatibility JSON schema accepted by the `task` executor.
///
/// The model sees the more compact array-only schema. This public schema also
/// accepts the pre-6.8 single-task object so persisted and host-direct calls do
/// not break during the tool-surface migration.
pub fn task_params_schema() -> serde_json::Value {
    task_params_schema_for_agents(&AgentRegistry::new().list_visible())
}

fn task_params_schema_for_agents(agents: &[AgentDefinition]) -> serde_json::Value {
    serde_json::json!({
        "oneOf": [
            legacy_task_params_schema_for_agents(agents),
            task_model_params_schema_for_agents(agents)
        ]
    })
}

fn legacy_task_params_schema_for_agents(agents: &[AgentDefinition]) -> serde_json::Value {
    let mut schema = task_item_params_schema_for_agents(agents, true);
    schema["examples"] = serde_json::json!([
        {
            "agent": "explore",
            "description": "Find Rust files",
            "prompt": "Search the workspace for Rust files and summarize the layout."
        },
        {
            "agent": "general",
            "description": "Investigate test failure",
            "prompt": "Inspect the failing tests and explain the root cause.",
            "max_steps": 6
        }
    ]);
    schema
}

fn task_model_params_schema_for_agents(agents: &[AgentDefinition]) -> serde_json::Value {
    parallel_params::task_tool_params_schema_for_agents(agents)
}

pub(super) fn task_item_params_schema_for_agents(
    agents: &[AgentDefinition],
    include_background: bool,
) -> serde_json::Value {
    let mut properties = serde_json::Map::from_iter([
        ("agent".to_string(), task_agent_parameter_schema(agents)),
        (
            "description".to_string(),
            serde_json::json!({
                "type": "string",
                "description": "Required. Short task label for display and tracking. Always provide this exact field name: 'description'."
            }),
        ),
        (
            "prompt".to_string(),
            serde_json::json!({
                "type": "string",
                "description": "Required. Detailed instruction for the delegated child run. Always provide this exact field name: 'prompt'."
            }),
        ),
        (
            "max_steps".to_string(),
            serde_json::json!({
                "type": "integer",
                "description": "Optional. Maximum number of steps for this task."
            }),
        ),
        (
            "output_schema".to_string(),
            serde_json::json!({
                "type": "object",
                "description": "Optional. JSON Schema object the delegated result must satisfy. When provided, the child output is coerced into a validated structured object and returned in metadata."
            }),
        ),
    ]);
    if include_background {
        properties.insert(
            "background".to_string(),
            serde_json::json!({
                "type": "boolean",
                "description": "Optional. Run this task in the background. Only valid when the outer tasks array contains one item. Default: false.",
                "default": false
            }),
        );
    }

    serde_json::json!({
        "type": "object",
        "additionalProperties": false,
        "properties": properties,
        "required": ["agent", "description", "prompt"]
    })
}

/// TaskTool wraps TaskExecutor as a Tool for registration in ToolExecutor.
/// This allows the LLM to delegate tasks through the standard tool interface.
pub struct TaskTool {
    executor: Arc<TaskExecutor>,
}

impl TaskTool {
    /// Create a new TaskTool
    pub fn new(executor: Arc<TaskExecutor>) -> Self {
        Self { executor }
    }

    async fn execute_single(&self, params: TaskParams, ctx: &ToolContext) -> Result<ToolOutput> {
        let parent_cancellation = ctx.cancellation_token();
        let executor = self.executor.scoped_for_invocation(ctx);

        if params.background {
            let task_id = executor.execute_background_with_parent_cancellation(
                params,
                ctx.agent_event_tx.clone(),
                ctx.session_id.clone(),
                Some(parent_cancellation),
            );
            return Ok(ToolOutput::success(format!(
                "Task started in background. Task ID: {}",
                task_id
            )));
        }

        let result = executor
            .execute_with_parent_cancellation(
                params,
                ctx.agent_event_tx.clone(),
                ctx.session_id.as_deref(),
                Some(&parent_cancellation),
            )
            .await?;
        let (content, truncated) = format_task_result_for_context(&result);
        let metadata = serde_json::json!({
            "task_id": result.task_id,
            "session_id": result.session_id,
            "agent": result.agent,
            "success": result.success,
            "output_bytes": result.output.len(),
            "truncated_for_context": truncated,
            "artifact_id": task_artifact_id(&result),
            "artifact_uri": task_artifact_uri(&result),
            "structured": result.structured,
            "source_anchors": result.source_anchors,
        });

        if result.success {
            Ok(ToolOutput::success(content).with_metadata(metadata))
        } else {
            Ok(ToolOutput::error(content).with_metadata(metadata))
        }
    }
}

#[async_trait]
impl Tool for TaskTool {
    fn name(&self) -> &str {
        "task"
    }

    fn description(&self) -> &str {
        TASK_TOOL_DESCRIPTION
    }

    fn parameters(&self) -> serde_json::Value {
        task_params_schema_for_agents(&self.executor.visible_agents())
    }

    fn definition(&self) -> ToolDefinition {
        let agents = self.executor.visible_agents();
        ToolDefinition {
            name: self.name().to_string(),
            description: delegation_tool_description(self.description(), &agents),
            parameters: task_model_params_schema_for_agents(&agents),
        }
    }

    async fn execute(&self, args: &serde_json::Value, ctx: &ToolContext) -> Result<ToolOutput> {
        if args.get("tasks").is_some() {
            let mut params: ParallelTaskParams = match serde_json::from_value(args.clone()) {
                Ok(params) => params,
                Err(error) => {
                    return Ok(invalid_delegation_argument(format!(
                        "Invalid task parameters: {error}"
                    )));
                }
            };
            if params.tasks.is_empty() {
                return Ok(invalid_delegation_argument(
                    "task requires at least 1 task".to_string(),
                ));
            }

            let has_fanout_options = params.allow_partial_failure
                || params.timeout_ms.is_some()
                || params.min_success_count.is_some();
            if params.tasks.len() == 1 && !has_fanout_options {
                return self.execute_single(params.tasks.remove(0), ctx).await;
            }

            return ParallelTaskTool::new(Arc::clone(&self.executor))
                .execute_params(params, ctx, "task", 1)
                .await;
        }

        let params: TaskParams = match serde_json::from_value(args.clone()) {
            Ok(params) => params,
            Err(error) => {
                return Ok(invalid_delegation_argument(format!(
                    "Invalid task parameters: {error}"
                )));
            }
        };
        self.execute_single(params, ctx).await
    }
}

mod parallel_params;
pub use parallel_params::{parallel_task_params_schema, ParallelTaskParams};

/// ParallelTaskTool allows the LLM to fan out multiple delegated tasks concurrently.
///
/// All tasks execute in parallel and the tool returns when all complete.
pub struct ParallelTaskTool {
    executor: Arc<TaskExecutor>,
}

impl ParallelTaskTool {
    /// Create a new ParallelTaskTool
    pub fn new(executor: Arc<TaskExecutor>) -> Self {
        Self { executor }
    }

    async fn execute_params(
        &self,
        params: ParallelTaskParams,
        ctx: &ToolContext,
        tool_name: &str,
        min_tasks: usize,
    ) -> Result<ToolOutput> {
        let started_at = std::time::Instant::now();
        let parent_cancellation = ctx.cancellation_token();
        let executor = self.executor.scoped_for_invocation(ctx);

        if params.tasks.len() < min_tasks {
            return Ok(invalid_delegation_argument(format!(
                "{tool_name} requires at least {min_tasks} task{}",
                if min_tasks == 1 { "" } else { "s" }
            )));
        }
        if params.tasks.len() > MAX_PARALLEL_TASKS_PER_CALL {
            return Ok(invalid_delegation_argument(format!(
                "{tool_name} accepts at most {MAX_PARALLEL_TASKS_PER_CALL} tasks"
            )));
        }
        if let Some((index, _)) = params
            .tasks
            .iter()
            .enumerate()
            .find(|(_, task)| task.background)
        {
            return Ok(invalid_delegation_argument(format!(
                "{tool_name} task {} cannot set background=true when fan-out options are used or multiple tasks are submitted; every branch is collected by the parent call",
                index + 1
            )));
        }
        if params.timeout_ms == Some(0) {
            return Ok(invalid_delegation_argument(format!(
                "{tool_name} timeout_ms must be at least 1"
            )));
        }
        if let Some(min_success_count) = params.min_success_count {
            if !params.allow_partial_failure {
                return Ok(invalid_delegation_argument(format!(
                    "{tool_name} min_success_count requires allow_partial_failure=true"
                )));
            }
            if min_success_count == 0 || min_success_count > params.tasks.len() {
                return Ok(invalid_delegation_argument(format!(
                    "{tool_name} min_success_count must be between 1 and the task count ({})",
                    params.tasks.len()
                )));
            }
        }

        let task_count = params.tasks.len();
        let run = executor
            .execute_parallel_for_tool(
                params.tasks.clone(),
                ctx.agent_event_tx.clone(),
                parallel_execution::ParallelToolOptions {
                    parent_session_id: ctx.session_id.as_deref(),
                    timeout_ms: params.timeout_ms,
                    min_success_count: params.min_success_count,
                    allow_partial_failure: params.allow_partial_failure,
                    parent_cancellation: Some(&parent_cancellation),
                },
            )
            .await;
        let results = run.results;

        let mut output = format!("Executed {} tasks concurrently:\n\n", task_count);
        let mut metadata_results = Vec::new();
        let source_anchor_counts = parallel_source_anchor_counts(&results);
        for (i, result) in results.iter().enumerate() {
            let status = if result.success { "[OK]" } else { "[ERR]" };
            let (formatted, truncated) = format_task_result_for_context(result);
            let (output_excerpt, _) = compact_task_output(&result.output);
            let source_anchors = &result.source_anchors[..source_anchor_counts[i]];
            metadata_results.push(serde_json::json!({
                "task_id": result.task_id,
                "session_id": result.session_id,
                "agent": result.agent,
                "success": result.success,
                "error_message": (!result.success).then(|| {
                    crate::text::truncate_utf8(&result.output, 1024).to_string()
                }),
                "output_excerpt": output_excerpt,
                "structured": result.structured,
                "source_anchors": source_anchors,
                "output_bytes": result.output.len(),
                "truncated_for_context": truncated,
                "artifact_id": task_artifact_id(result),
                "artifact_uri": task_artifact_uri(result),
            }));
            output.push_str(&format!(
                "--- Task {} ({}) {} ---\n{}\n\n",
                i + 1,
                result.agent,
                status,
                formatted
            ));
        }

        let success_count = results.iter().filter(|result| result.success).count();
        let failed_count = results.len().saturating_sub(success_count);
        let all_success = failed_count == 0;
        let partial_failure = failed_count > 0 && success_count > 0;
        if params.allow_partial_failure && partial_failure {
            output.push_str(&format!(
                "Partial failure tolerated: {success_count} succeeded, {failed_count} failed.\n"
            ));
        }
        if run.timed_out {
            output.push_str(&format!(
                "Task fan-out timed out after {} ms; returned completed child results and marked unfinished children failed.\n",
                run.timeout_ms.unwrap_or_default()
            ));
        } else if run.returned_early {
            output.push_str(&format!(
                "Task fan-out returned after reaching min_success_count={}; unfinished children were marked failed.\n",
                run.min_success_count.unwrap_or_default()
            ));
        }

        let tool_success = all_success || (params.allow_partial_failure && success_count > 0);
        let mut output = if tool_success {
            ToolOutput::success(output)
        } else {
            ToolOutput::error(output)
        };
        if !tool_success && failed_count > 0 {
            output.error_kind = Some(crate::tools::ToolErrorKind::PartialFailure {
                failed: failed_count,
                total: results.len(),
            });
        }

        Ok(output.with_metadata(serde_json::json!({
            "task_count": task_count,
            "result_count": results.len(),
            "success_count": success_count,
            "failed_count": failed_count,
            "all_success": all_success,
            "partial_failure": partial_failure,
            "allow_partial_failure": params.allow_partial_failure,
            "timeout_ms": params.timeout_ms,
            "timed_out": run.timed_out,
            "min_success_count": params.min_success_count,
            "returned_early": run.returned_early,
            "duration_ms": started_at.elapsed().as_millis().min(u128::from(u64::MAX)) as u64,
            "results": metadata_results,
        })))
    }
}

#[async_trait]
impl Tool for ParallelTaskTool {
    fn name(&self) -> &str {
        "parallel_task"
    }

    fn description(&self) -> &str {
        PARALLEL_TASK_TOOL_DESCRIPTION
    }

    fn parameters(&self) -> serde_json::Value {
        parallel_params::parallel_task_params_schema_for_agents(&self.executor.visible_agents())
    }

    fn definition(&self) -> ToolDefinition {
        let agents = self.executor.visible_agents();
        ToolDefinition {
            name: self.name().to_string(),
            description: delegation_tool_description(self.description(), &agents),
            parameters: parallel_params::parallel_task_params_schema_for_agents(&agents),
        }
    }

    fn is_model_visible(&self) -> bool {
        false
    }

    async fn execute(&self, args: &serde_json::Value, ctx: &ToolContext) -> Result<ToolOutput> {
        let params: ParallelTaskParams = match serde_json::from_value(args.clone()) {
            Ok(params) => params,
            Err(error) => {
                return Ok(invalid_delegation_argument(format!(
                    "Invalid parallel_task parameters: {error}"
                )));
            }
        };
        self.execute_params(params, ctx, "parallel_task", 2).await
    }
}

fn invalid_delegation_argument(message: String) -> ToolOutput {
    ToolOutput::error(&message)
        .with_error_kind(crate::tools::ToolErrorKind::InvalidArgument { message })
}

#[cfg(test)]
mod tests;