microclaw_engine/tools/

subagents.rs

use async_trait::async_trait;
use serde_json::json;
use std::collections::{BTreeSet, HashMap};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, LazyLock, Mutex};
use tokio::sync::Semaphore;
use tracing::{info, warn};

use super::{
    auth_context_from_input, authorize_chat_access, schema_object, Tool, ToolAuthContext,
    ToolRegistry, ToolResult,
};
use crate::config::{Config, ResolvedSubagentAcpTargetConfig};
use crate::internal::channels::channel::deliver_and_store_bot_message;
use crate::internal::channels::channel_adapter::ChannelRegistry;
use crate::internal::storage::db::{
    call_blocking, CreateSubagentRunParams, Database, FinishSubagentRunParams,
};
use microclaw_core::llm_types::{
    ContentBlock, Message, MessageContent, ResponseContentBlock, ToolDefinition,
};

const MAX_SUB_AGENT_ITERATIONS: usize = 16;

#[derive(Debug, Clone, Copy)]
enum SubagentExecutionRuntime {
    Native,
    Acp,
}

impl SubagentExecutionRuntime {
    fn from_raw(raw: Option<&str>) -> Result<Self, String> {
        match raw.unwrap_or("native").trim().to_ascii_lowercase().as_str() {
            "" | "native" => Ok(Self::Native),
            "acp" => Ok(Self::Acp),
            other => Err(format!(
                "Unsupported subagent runtime '{other}'. Expected 'native' or 'acp'."
            )),
        }
    }

    fn from_input(
        input: &serde_json::Value,
        parent_meta: Option<&SubagentRuntimeMeta>,
    ) -> Result<Self, String> {
        let runtime = input
            .get("runtime")
            .and_then(|v| v.as_str())
            .or_else(|| parent_meta.and_then(|meta| meta.runtime.as_deref()));
        Self::from_raw(runtime)
    }

    fn as_str(&self) -> &'static str {
        match self {
            Self::Native => "native",
            Self::Acp => "acp",
        }
    }
}

#[derive(Debug, Clone)]
struct SubagentRuntimeMeta {
    depth: i64,
    token_budget_remaining: Option<i64>,
    runtime: Option<String>,
    runtime_target: Option<String>,
}

fn subagent_runtime_meta_from_input(input: &serde_json::Value) -> Option<SubagentRuntimeMeta> {
    let meta = input.get("__subagent_runtime")?;
    let depth = meta.get("depth").and_then(|v| v.as_i64()).unwrap_or(0);
    let token_budget_remaining = meta
        .get("token_budget_remaining")
        .and_then(|v| v.as_i64())
        .filter(|v| *v > 0);
    let runtime = meta
        .get("runtime")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|v| !v.is_empty())
        .map(ToOwned::to_owned);
    let runtime_target = meta
        .get("runtime_target")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|v| !v.is_empty())
        .map(ToOwned::to_owned);
    Some(SubagentRuntimeMeta {
        depth,
        token_budget_remaining,
        runtime,
        runtime_target,
    })
}

fn compute_child_token_budget(
    requested_budget: Option<i64>,
    parent_budget_remaining: Option<i64>,
    configured_max: i64,
) -> Result<i64, String> {
    let configured_max = configured_max.clamp(2_000, 2_000_000);
    if let Some(parent_remaining) = parent_budget_remaining {
        if parent_remaining < 2_000 {
            return Err(format!(
                "subagent budget exhausted: parent remaining {} < 2000",
                parent_remaining
            ));
        }
        let desired = requested_budget.unwrap_or((parent_remaining / 2).max(2_000));
        return Ok(desired.clamp(2_000, parent_remaining.min(configured_max)));
    }
    let desired = requested_budget.unwrap_or(configured_max);
    Ok(desired.clamp(2_000, configured_max))
}

/// Name of the synthetic tool a sub-agent calls to return its final structured
/// result. Using a tool input schema is the provider-native way to constrain the
/// shape of the output (validated function/tool arguments on both Anthropic and
/// OpenAI), which is far more reliable than parsing free-text JSON.
const SUBMIT_RESULT_TOOL: &str = "submit_result";

/// The contract schema, expressed as a tool so the model emits it as validated
/// tool input rather than free text.
fn submit_result_tool_definition() -> ToolDefinition {
    ToolDefinition {
        name: SUBMIT_RESULT_TOOL.to_string(),
        description:
            "Submit your final structured result and end the run. Call this exactly once when the \
             task is complete, instead of writing the result as plain text."
                .to_string(),
        input_schema: json!({
            "type": "object",
            "properties": {
                "summary": {"type": "string", "description": "One-line summary of what you did."},
                "findings": {"type": "array", "items": {"type": "string"}, "description": "Key findings or results."},
                "artifacts": {"type": "array", "items": {"type": "object"}, "description": "Files/artifacts produced, each {type, path, description}."},
                "next_actions": {"type": "array", "items": {"type": "string"}, "description": "Suggested follow-up actions."},
                "final_answer": {"type": "string", "description": "The full answer delivered to the user."}
            },
            "required": ["summary", "final_answer"]
        }),
    }
}

/// Whether the sub-agent's terminal text actually carries the structured
/// output contract — a JSON object with a non-empty `summary` or `final_answer`
/// — as opposed to raw prose or leaked reasoning. Drives the one-shot repair
/// re-ask in the sub-agent loop before falling back to best-effort
/// normalization.
pub(crate) fn subagent_output_is_structured(raw_text: &str) -> bool {
    let cleaned = crate::agent_engine::sanitize_user_visible_text(raw_text);
    let text = cleaned.trim();
    let parsed = serde_json::from_str::<serde_json::Value>(text)
        .ok()
        .or_else(|| {
            let start = text.find('{')?;
            let end = text.rfind('}')?;
            if end > start {
                serde_json::from_str::<serde_json::Value>(&text[start..=end]).ok()
            } else {
                None
            }
        });
    parsed
        .as_ref()
        .and_then(|v| v.as_object())
        .map(|obj| {
            ["final_answer", "summary"].iter().any(|k| {
                obj.get(*k)
                    .and_then(|v| v.as_str())
                    .map(|s| !s.trim().is_empty())
                    .unwrap_or(false)
            })
        })
        .unwrap_or(false)
}

pub(crate) fn normalize_subagent_artifact_payload(raw_text: &str) -> (String, String) {
    // Strip any chain-of-thought the model emitted so it never leaks into the
    // chat announcement, and so a JSON object that follows a <think> block can
    // still be parsed (a leading reasoning block makes a whole-string parse fail).
    let cleaned = crate::agent_engine::sanitize_user_visible_text(raw_text);
    let text = cleaned.trim();
    // Prefer a clean whole-string parse; otherwise recover an embedded {...}
    // object (model wrapped the JSON in prose) so we don't fall back to dumping
    // raw reasoning as the answer.
    let parsed = serde_json::from_str::<serde_json::Value>(text)
        .ok()
        .or_else(|| {
            let start = text.find('{')?;
            let end = text.rfind('}')?;
            if end > start {
                serde_json::from_str::<serde_json::Value>(&text[start..=end]).ok()
            } else {
                None
            }
        });
    let mut summary = String::new();
    let mut final_answer = String::new();
    let mut findings: Vec<String> = Vec::new();
    let mut next_actions: Vec<String> = Vec::new();
    let mut artifacts: Vec<serde_json::Value> = Vec::new();
    if let Some(obj) = parsed.as_ref().and_then(|v| v.as_object()) {
        summary = obj
            .get("summary")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .trim()
            .to_string();
        final_answer = obj
            .get("final_answer")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .trim()
            .to_string();
        if let Some(arr) = obj.get("findings").and_then(|v| v.as_array()) {
            findings = arr
                .iter()
                .filter_map(|v| v.as_str())
                .map(str::trim)
                .filter(|v| !v.is_empty())
                .map(ToOwned::to_owned)
                .collect();
        }
        if let Some(arr) = obj.get("next_actions").and_then(|v| v.as_array()) {
            next_actions = arr
                .iter()
                .filter_map(|v| v.as_str())
                .map(str::trim)
                .filter(|v| !v.is_empty())
                .map(ToOwned::to_owned)
                .collect();
        }
        if let Some(arr) = obj.get("artifacts").and_then(|v| v.as_array()) {
            artifacts = arr.clone();
        }
    }
    if summary.is_empty() {
        summary = text
            .chars()
            .take(600)
            .collect::<String>()
            .trim()
            .to_string();
    }
    if final_answer.is_empty() {
        final_answer = summary.clone();
    }
    let envelope = json!({
        "protocol": "subagent_artifact_v1",
        "summary": summary,
        "findings": findings,
        "artifacts": artifacts,
        "next_actions": next_actions,
        "final_answer": final_answer,
        "raw_text": text,
    });
    let answer_text = envelope
        .get("final_answer")
        .and_then(|v| v.as_str())
        .unwrap_or("(sub-agent produced no output)")
        .to_string();
    (answer_text, envelope.to_string())
}

struct SubagentRuntime {
    semaphore: Semaphore,
    cancel_flags: Mutex<HashMap<String, Arc<AtomicBool>>>,
}

impl SubagentRuntime {
    fn new(max_concurrent: usize) -> Self {
        Self {
            semaphore: Semaphore::new(max_concurrent.max(1)),
            cancel_flags: Mutex::new(HashMap::new()),
        }
    }

    fn register_run(&self, run_id: &str) -> Arc<AtomicBool> {
        let flag = Arc::new(AtomicBool::new(false));
        if let Ok(mut guard) = self.cancel_flags.lock() {
            guard.insert(run_id.to_string(), flag.clone());
        }
        flag
    }

    fn cancel_run(&self, run_id: &str) {
        if let Ok(guard) = self.cancel_flags.lock() {
            if let Some(flag) = guard.get(run_id) {
                flag.store(true, Ordering::Relaxed);
            }
        }
    }

    fn remove_run(&self, run_id: &str) {
        if let Ok(mut guard) = self.cancel_flags.lock() {
            guard.remove(run_id);
        }
    }
}

static RUNTIME: LazyLock<Mutex<Option<Arc<SubagentRuntime>>>> = LazyLock::new(|| Mutex::new(None));

// Subagent runs must outlive the Tokio runtime of the foreground caller. Work
// intentionally creates a short-lived runtime for each turn; spawning on that
// runtime used to abort accepted background runs as soon as the Main Agent
// finished. A process-lifetime executor gives Server and Work the same durable
// in-process lifecycle while SQLite remains the restart recovery authority.
static TASK_RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .thread_name("microclaw-subagent")
        .build()
        .expect("failed to create process-lifetime subagent runtime")
});

fn subagent_runtime(config: &Config) -> Arc<SubagentRuntime> {
    let mut guard = match RUNTIME.lock() {
        Ok(g) => g,
        Err(poisoned) => poisoned.into_inner(),
    };
    if let Some(existing) = guard.as_ref() {
        return existing.clone();
    }
    let runtime = Arc::new(SubagentRuntime::new(config.subagents.max_concurrent));
    *guard = Some(runtime.clone());
    runtime
}

/// Shared cancellation boundary for product UIs that already resolved and
/// authorized the owning chat. The database flag survives process failure;
/// the in-memory flag interrupts a currently executing run promptly.
pub fn request_subagent_cancel(
    config: &Config,
    db: &Database,
    chat_id: i64,
    run_id: &str,
) -> Result<bool, microclaw_core::error::MicroClawError> {
    let requested = db.request_subagent_cancel(run_id, chat_id)?;
    if requested {
        subagent_runtime(config).cancel_run(run_id);
    }
    Ok(requested)
}

async fn log_subagent_event(
    db: Arc<Database>,
    run_id: &str,
    event_type: &str,
    detail: Option<String>,
) {
    let run_id = run_id.to_string();
    let event_type = event_type.to_string();
    let _ = call_blocking(db, move |db| {
        db.append_subagent_event(&run_id, &event_type, detail.as_deref())
    })
    .await;
}

pub(crate) async fn is_cancelled(
    db: Arc<Database>,
    run_id: &str,
    local_flag: &Arc<AtomicBool>,
) -> Result<bool, String> {
    if local_flag.load(Ordering::Relaxed) {
        return Ok(true);
    }
    let run_id_owned = run_id.to_string();
    let db_cancel = call_blocking(db, move |db| db.is_subagent_cancel_requested(&run_id_owned))
        .await
        .map_err(|e| format!("Failed checking cancel state: {e}"))?;
    Ok(db_cancel)
}

struct RunSubAgentTaskParams {
    config: Config,
    db: Arc<Database>,
    channel_registry: Arc<ChannelRegistry>,
    auth_context: ToolAuthContext,
    run_id: String,
    runtime: SubagentExecutionRuntime,
    acp_target: Option<ResolvedSubagentAcpTargetConfig>,
    depth: i64,
    run_token_budget: i64,
    task: String,
    context: String,
    specialist: String,
    exit_criteria: Vec<crate::completion_contract::ExitCriterion>,
    local_cancel: Arc<AtomicBool>,
}

/// Bash-backed runner for `Command` exit criteria: routes through the
/// sub-agent's own tool registry, so the sandbox router, dangerous-pattern
/// checks, and tool_policy apply exactly as for agent-issued commands.
struct SubagentBashRunner<'a> {
    tools: &'a ToolRegistry,
    auth: &'a ToolAuthContext,
}

#[async_trait::async_trait]
impl crate::completion_contract::CommandRunner for SubagentBashRunner<'_> {
    async fn run(&self, command: &str) -> (bool, String) {
        let result = self
            .tools
            .execute_with_auth("bash", json!({ "command": command }), self.auth)
            .await;
        (!result.is_error, result.content)
    }
}

/// Result of verifying a finished sub-agent result against its contract.
struct ContractVerdict {
    /// Result text with the evidence report prepended.
    annotated: String,
    /// The report alone (used to prompt the bounded in-run retry).
    report: String,
    passed: bool,
}

/// Verify a finished result against its completion contract (if any) and
/// build the evidence report, so the parent loop and the user see
/// VERIFIED/FAILED with per-criterion evidence instead of trusting the
/// sub-agent's self-report. Returns `None` when no contract was declared.
#[allow(clippy::too_many_arguments)]
async fn apply_completion_contract(
    criteria: &[crate::completion_contract::ExitCriterion],
    final_text: &str,
    config: &Config,
    tools: &ToolRegistry,
    auth: &ToolAuthContext,
    db: Arc<Database>,
    run_id: &str,
) -> Option<ContractVerdict> {
    if criteria.is_empty() {
        return None;
    }
    let base = std::path::Path::new(&config.working_dir);
    let working_dir = match config.working_dir_isolation {
        crate::config::WorkingDirIsolation::Direct => base.to_path_buf(),
        crate::config::WorkingDirIsolation::Shared => base.join("shared"),
        crate::config::WorkingDirIsolation::Chat => {
            crate::internal::tool_runtime::runtime::chat_working_dir(
                base,
                &auth.caller_channel,
                auth.caller_chat_id,
            )
        }
    };
    let runner = SubagentBashRunner { tools, auth };
    let outcomes = crate::completion_contract::verify_criteria(
        criteria,
        final_text,
        &working_dir,
        Some(&runner),
    )
    .await;
    let passed = outcomes.iter().filter(|o| o.passed).count();
    log_subagent_event(
        db,
        run_id,
        "contract",
        Some(format!(
            "{} {passed}/{}",
            if passed == outcomes.len() {
                "verified"
            } else {
                "failed"
            },
            outcomes.len()
        )),
    )
    .await;
    let report = crate::completion_contract::render_report(&outcomes);
    Some(ContractVerdict {
        annotated: format!("{report}\n{final_text}"),
        passed: passed == outcomes.len(),
        report,
    })
}

/// Prompt sent back into the sub-agent conversation for the single bounded
/// contract retry.
fn contract_retry_prompt(report: &str) -> String {
    format!(
        "Completion contract verification FAILED:\n{report}\nThe failed criteria are unmet \
         work — complete them now (use tools), then submit your result again via `submit_result`. \
         This is your final attempt."
    )
}

async fn run_sub_agent_task(
    params: RunSubAgentTaskParams,
) -> Result<(String, String, i64, i64), String> {
    let RunSubAgentTaskParams {
        config,
        db,
        channel_registry,
        auth_context,
        run_id,
        runtime,
        acp_target,
        depth,
        run_token_budget,
        task,
        context,
        specialist,
        exit_criteria,
        local_cancel,
    } = params;
    if matches!(runtime, SubagentExecutionRuntime::Acp) {
        let Some(acp_target) = acp_target else {
            return Err("ACP runtime target was not resolved".into());
        };
        // Contracts on the ACP runtime: the external agent's conversation is
        // not ours to continue, so the bounded retry is a full re-run with
        // the failure evidence appended to the context.
        let mut context = context;
        if !exit_criteria.is_empty() {
            context.push_str(&crate::completion_contract::render_for_prompt(
                &exit_criteria,
            ));
        }
        let first = crate::acp_subagent::run_acp_subagent_task(
            crate::acp_subagent::AcpSubagentTaskParams {
                config: config.clone(),
                db: db.clone(),
                auth_context: auth_context.clone(),
                run_id: run_id.clone(),
                task: task.clone(),
                context: context.clone(),
                local_cancel: local_cancel.clone(),
                target: acp_target.clone(),
            },
        )
        .await?;
        if exit_criteria.is_empty() {
            return Ok(first);
        }
        // Verification (and its command runner) go through our own sub-agent
        // tool registry — same sandbox/policy guards as the native runtime.
        let tools = ToolRegistry::new_sub_agent(&config, db.clone(), Some(channel_registry), false);
        let (final_text, artifact_json, mut in_tok, mut out_tok) = first;
        let verdict = apply_completion_contract(
            &exit_criteria,
            &final_text,
            &config,
            &tools,
            &auth_context,
            db.clone(),
            &run_id,
        )
        .await
        .expect("criteria checked non-empty");
        if verdict.passed {
            return Ok((verdict.annotated, artifact_json, in_tok, out_tok));
        }
        log_subagent_event(db.clone(), &run_id, "contract_retry", None).await;
        let retry_context = format!("{context}\n\n{}", contract_retry_prompt(&verdict.report));
        let (retry_text, retry_artifact, retry_in, retry_out) =
            crate::acp_subagent::run_acp_subagent_task(
                crate::acp_subagent::AcpSubagentTaskParams {
                    config: config.clone(),
                    db: db.clone(),
                    auth_context: auth_context.clone(),
                    run_id: run_id.clone(),
                    task,
                    context: retry_context,
                    local_cancel,
                    target: acp_target,
                },
            )
            .await?;
        in_tok += retry_in;
        out_tok += retry_out;
        let verdict = apply_completion_contract(
            &exit_criteria,
            &retry_text,
            &config,
            &tools,
            &auth_context,
            db.clone(),
            &run_id,
        )
        .await
        .expect("criteria checked non-empty");
        return Ok((verdict.annotated, retry_artifact, in_tok, out_tok));
    }

    let llm = crate::llm::create_provider(&config);
    let allow_session_tools = depth < config.subagents.max_spawn_depth as i64;
    let tools = ToolRegistry::new_sub_agent(
        &config,
        db.clone(),
        Some(channel_registry),
        allow_session_tools,
    );
    // The sub-agent's normal tools plus a synthetic `submit_result` tool used to
    // return the final structured result as validated tool input (native
    // constrained decoding) rather than hand-written JSON text.
    let mut tool_defs = tools.definitions().to_vec();
    tool_defs.push(submit_result_tool_definition());

    let profile = crate::tools::specialists::resolve_specialist(Some(specialist.as_str()));
    let system_prompt = format!(
        "{persona}\n\nComplete the task thoroughly with tool use when needed.\nYou are a background worker: you have NO tools to message the chat, write memory, or schedule tasks (no `send_message`, `write_memory`, or `schedule` — do not look for them or stall when they are missing). `read_memory` and `structured_memory_search` are read-only lookups. Any durable facts you surface in `findings`/`final_answer` are persisted by the orchestrator automatically — just report them, don't try to save them yourself.\nFor long tasks, call `report_progress` at meaningful milestones with a one-line status so the user gets colleague-style updates while you work.\nIf a sub-problem falls outside your expertise, get a quick second opinion from the right specialist with `consult_specialist` (e.g. as a researcher, hand a draft to the writer; as a coder, ask the mathematician to check a formula) and weave their answer into your work — don't fake expertise you don't have. If a sub-problem is large enough to need its own run and you're allowed to spawn, delegate it with `sessions_spawn`; otherwise name the right specialist in next_actions.\nOutput contract (required): when the task is complete, call the `submit_result` tool exactly once with your structured result (summary, findings, artifacts, next_actions, final_answer). Prefer `submit_result` over writing the result as text. If you do answer in text instead, it MUST be a single JSON object with those same keys and nothing else.",
        persona = profile.persona
    );

    let mut user_content = if context.is_empty() {
        task.to_string()
    } else {
        format!("Context: {context}\n\nTask: {task}")
    };
    if !exit_criteria.is_empty() {
        user_content.push_str(&crate::completion_contract::render_for_prompt(
            &exit_criteria,
        ));
    }

    let mut messages = vec![Message {
        role: "user".into(),
        content: MessageContent::Text(user_content),
    }];
    let mut input_tokens_sum = 0_i64;
    let mut output_tokens_sum = 0_i64;
    // Bounded (one-shot) repair: if the model ends without the required JSON
    // contract, we re-ask exactly once before falling back to best-effort
    // normalization. Prevents an unbounded re-ask loop.
    let mut repair_attempted = false;
    // Bounded contract retry: when the completion contract fails, give the
    // sub-agent exactly ONE chance to finish the unmet criteria before the
    // failed result is returned as-is.
    let mut contract_retry_attempted = false;

    for _ in 0..MAX_SUB_AGENT_ITERATIONS {
        if is_cancelled(db.clone(), &run_id, &local_cancel).await? {
            return Err("cancelled".into());
        }

        let response = llm
            .send_message(&system_prompt, messages.clone(), Some(tool_defs.clone()))
            .await
            .map_err(|e| format!("Sub-agent API error: {e}"))?;

        if let Some(usage) = &response.usage {
            let input_tokens = i64::from(usage.input_tokens);
            let output_tokens = i64::from(usage.output_tokens);
            input_tokens_sum += input_tokens;
            output_tokens_sum += output_tokens;

            let channel = auth_context.caller_channel.clone();
            let provider = config.llm_provider.clone();
            let model = config.model.clone();
            let chat_id = auth_context.caller_chat_id;
            let _ = call_blocking(db.clone(), move |db| {
                db.log_llm_usage(
                    chat_id,
                    &channel,
                    &provider,
                    &model,
                    input_tokens,
                    output_tokens,
                    "subagent_run",
                )
                .map(|_| ())
            })
            .await;
            let total = input_tokens_sum + output_tokens_sum;
            if run_token_budget > 0 && total > run_token_budget {
                return Err(format!(
                    "budget_exceeded: total_tokens={} budget={}",
                    total, run_token_budget
                ));
            }
        }

        let stop_reason = response.stop_reason.as_deref().unwrap_or("end_turn");
        if stop_reason == "end_turn" || stop_reason == "max_tokens" {
            let text = response
                .content
                .iter()
                .filter_map(|block| match block {
                    ResponseContentBlock::Text { text } => Some(text.as_str()),
                    _ => None,
                })
                .collect::<Vec<_>>()
                .join("");
            // One-shot repair re-ask: the contract requires a JSON object, but
            // models sometimes stop on prose or leaked reasoning. Ask once for
            // clean JSON before falling back to best-effort normalization. Only
            // for a natural `end_turn`; a `max_tokens` stop is truncation, where
            // re-asking would just burn more budget without fixing the shape.
            if stop_reason == "end_turn"
                && !repair_attempted
                && !text.trim().is_empty()
                && !subagent_output_is_structured(&text)
            {
                repair_attempted = true;
                log_subagent_event(db.clone(), &run_id, "output_repair", None).await;
                messages.push(Message {
                    role: "assistant".into(),
                    content: MessageContent::Text(text),
                });
                messages.push(Message {
                    role: "user".into(),
                    content: MessageContent::Text(
                        "Your previous reply was not the required output. Respond with ONLY a single JSON object matching the contract: {\"summary\":string,\"findings\":string[],\"artifacts\":[],\"next_actions\":string[],\"final_answer\":string}. No prose, no markdown fences, no <think> tags."
                            .into(),
                    ),
                });
                continue;
            }
            let source = if text.is_empty() {
                "(sub-agent produced no output)".to_string()
            } else {
                text
            };
            let (final_text, artifact_json) = normalize_subagent_artifact_payload(&source);
            match apply_completion_contract(
                &exit_criteria,
                &final_text,
                &config,
                &tools,
                &auth_context,
                db.clone(),
                &run_id,
            )
            .await
            {
                Some(verdict) if !verdict.passed && !contract_retry_attempted => {
                    contract_retry_attempted = true;
                    log_subagent_event(db.clone(), &run_id, "contract_retry", None).await;
                    messages.push(Message {
                        role: "assistant".into(),
                        content: MessageContent::Text(verdict.annotated),
                    });
                    messages.push(Message {
                        role: "user".into(),
                        content: MessageContent::Text(contract_retry_prompt(&verdict.report)),
                    });
                    continue;
                }
                Some(verdict) => {
                    return Ok((
                        verdict.annotated,
                        artifact_json,
                        input_tokens_sum,
                        output_tokens_sum,
                    ));
                }
                None => {
                    return Ok((
                        final_text,
                        artifact_json,
                        input_tokens_sum,
                        output_tokens_sum,
                    ));
                }
            }
        }

        if stop_reason == "tool_use" {
            // If the model called `submit_result`, its (schema-validated) input
            // IS the final structured result — short-circuit and return it.
            if let Some(submit_input) = response.content.iter().find_map(|b| match b {
                ResponseContentBlock::ToolUse { name, input, .. } if name == SUBMIT_RESULT_TOOL => {
                    Some(input.clone())
                }
                _ => None,
            }) {
                log_subagent_event(db.clone(), &run_id, "submit_result", None).await;
                let (final_text, artifact_json) =
                    normalize_subagent_artifact_payload(&submit_input.to_string());
                match apply_completion_contract(
                    &exit_criteria,
                    &final_text,
                    &config,
                    &tools,
                    &auth_context,
                    db.clone(),
                    &run_id,
                )
                .await
                {
                    Some(verdict) if !verdict.passed && !contract_retry_attempted => {
                        contract_retry_attempted = true;
                        log_subagent_event(db.clone(), &run_id, "contract_retry", None).await;
                        // Keep the tool_use/tool_result pairing the provider
                        // requires: echo the assistant turn (including the
                        // submit_result call), then answer that call with the
                        // failure report.
                        let submit_id = response
                            .content
                            .iter()
                            .find_map(|b| match b {
                                ResponseContentBlock::ToolUse { id, name, .. }
                                    if name == SUBMIT_RESULT_TOOL =>
                                {
                                    Some(id.clone())
                                }
                                _ => None,
                            })
                            .unwrap_or_default();
                        let assistant_content: Vec<ContentBlock> = response
                            .content
                            .iter()
                            .filter_map(|block| match block {
                                ResponseContentBlock::Text { text } => {
                                    Some(ContentBlock::Text { text: text.clone() })
                                }
                                ResponseContentBlock::ToolUse {
                                    id,
                                    name,
                                    input,
                                    thought_signature,
                                } => Some(ContentBlock::ToolUse {
                                    id: id.clone(),
                                    name: name.clone(),
                                    input: input.clone(),
                                    thought_signature: thought_signature.clone(),
                                }),
                                ResponseContentBlock::Other => None,
                            })
                            .collect();
                        messages.push(Message {
                            role: "assistant".into(),
                            content: MessageContent::Blocks(assistant_content),
                        });
                        messages.push(Message {
                            role: "user".into(),
                            content: MessageContent::Blocks(vec![ContentBlock::ToolResult {
                                tool_use_id: submit_id,
                                content: contract_retry_prompt(&verdict.report),
                                is_error: Some(true),
                            }]),
                        });
                        continue;
                    }
                    Some(verdict) => {
                        return Ok((
                            verdict.annotated,
                            artifact_json,
                            input_tokens_sum,
                            output_tokens_sum,
                        ));
                    }
                    None => {
                        return Ok((
                            final_text,
                            artifact_json,
                            input_tokens_sum,
                            output_tokens_sum,
                        ));
                    }
                }
            }

            let assistant_content: Vec<ContentBlock> = response
                .content
                .iter()
                .filter_map(|block| match block {
                    ResponseContentBlock::Text { text } => {
                        Some(ContentBlock::Text { text: text.clone() })
                    }
                    ResponseContentBlock::ToolUse {
                        id,
                        name,
                        input,
                        thought_signature,
                    } => Some(ContentBlock::ToolUse {
                        id: id.clone(),
                        name: name.clone(),
                        input: input.clone(),
                        thought_signature: thought_signature.clone(),
                    }),
                    ResponseContentBlock::Other => None,
                })
                .collect();

            messages.push(Message {
                role: "assistant".into(),
                content: MessageContent::Blocks(assistant_content),
            });

            let mut tool_results = Vec::new();
            for block in &response.content {
                if let ResponseContentBlock::ToolUse {
                    id, name, input, ..
                } = block
                {
                    log_subagent_event(
                        db.clone(),
                        &run_id,
                        "tool_use",
                        Some(format!("tool={name}")),
                    )
                    .await;
                    let mut tool_input = input.clone();
                    if let Some(obj) = tool_input.as_object_mut() {
                        let remaining_budget = if run_token_budget > 0 {
                            Some((run_token_budget - input_tokens_sum - output_tokens_sum).max(0))
                        } else {
                            None
                        };
                        obj.insert(
                            "__subagent_runtime".to_string(),
                            json!({
                                "run_id": run_id.clone(),
                                "depth": depth,
                                "runtime": "native",
                                "runtime_target": serde_json::Value::Null,
                                "token_budget_remaining": remaining_budget,
                            }),
                        );
                    }
                    let result = tools
                        .execute_with_auth(name, tool_input, &auth_context)
                        .await;
                    tool_results.push(ContentBlock::ToolResult {
                        tool_use_id: id.clone(),
                        content: result.content,
                        is_error: if result.is_error { Some(true) } else { None },
                    });
                }
            }

            messages.push(Message {
                role: "user".into(),
                content: MessageContent::Blocks(tool_results),
            });
            continue;
        }

        let text = response
            .content
            .iter()
            .filter_map(|block| match block {
                ResponseContentBlock::Text { text } => Some(text.as_str()),
                _ => None,
            })
            .collect::<Vec<_>>()
            .join("");
        let source = if text.is_empty() {
            "(sub-agent produced no output)".to_string()
        } else {
            text
        };
        let (final_text, artifact_json) = normalize_subagent_artifact_payload(&source);
        match apply_completion_contract(
            &exit_criteria,
            &final_text,
            &config,
            &tools,
            &auth_context,
            db.clone(),
            &run_id,
        )
        .await
        {
            Some(verdict) if !verdict.passed && !contract_retry_attempted => {
                contract_retry_attempted = true;
                log_subagent_event(db.clone(), &run_id, "contract_retry", None).await;
                messages.push(Message {
                    role: "assistant".into(),
                    content: MessageContent::Text(verdict.annotated),
                });
                messages.push(Message {
                    role: "user".into(),
                    content: MessageContent::Text(contract_retry_prompt(&verdict.report)),
                });
                continue;
            }
            Some(verdict) => {
                return Ok((
                    verdict.annotated,
                    artifact_json,
                    input_tokens_sum,
                    output_tokens_sum,
                ));
            }
            None => {
                return Ok((
                    final_text,
                    artifact_json,
                    input_tokens_sum,
                    output_tokens_sum,
                ));
            }
        }
    }

    Err("Sub-agent reached maximum iterations without completing the task.".into())
}

/// When the last child of a parent finishes, post one consolidated summary of
/// the whole batch. Idempotent: the announce is keyed on `<parent>:fanin`, whose
/// UNIQUE constraint means only one fan-in message is ever enqueued even if
/// several children finish at once.
async fn maybe_post_fan_in_summary(
    config: &Config,
    channel_registry: Arc<ChannelRegistry>,
    db: Arc<Database>,
    chat_id: i64,
    parent_id: &str,
) {
    let parent_owned = parent_id.to_string();
    let active = call_blocking(db.clone(), {
        let p = parent_owned.clone();
        move |db| db.count_active_subagent_children(&p)
    })
    .await
    .unwrap_or(0);
    if active > 0 {
        return;
    }
    let children = match call_blocking(db.clone(), {
        let p = parent_owned.clone();
        move |db| db.list_subagent_children(&p)
    })
    .await
    {
        Ok(v) => v,
        Err(e) => {
            warn!("fan-in: failed to list children of {parent_id}: {e}");
            return;
        }
    };
    // Only summarize genuine batches (the single-child case is covered by its
    // own completion message).
    if children.len() < 2 {
        return;
    }

    let n = children.len();
    let mut lines = vec![format!("🧩 All {n} sub-tasks done:")];
    let mut caller_channel = String::new();
    for c in &children {
        if caller_channel.is_empty() {
            caller_channel = c.caller_channel.clone();
        }
        let emoji = match c.status.as_str() {
            "completed" => "✅",
            "cancelled" => "🛑",
            "timed_out" => "⏱️",
            _ => "❌",
        };
        let name = c
            .label
            .clone()
            .filter(|l| !l.trim().is_empty())
            .unwrap_or_else(|| c.task.chars().take(40).collect::<String>());
        let detail = c
            .result_text
            .clone()
            .or_else(|| c.error_text.clone())
            .map(|t| {
                let snippet: String = t.trim().chars().take(120).collect();
                format!(" — {snippet}")
            })
            .unwrap_or_default();
        lines.push(format!("{emoji} {name}{detail}"));
    }
    let summary = lines.join("\n");

    let announce_id = format!("{parent_owned}:fanin");
    let enqueue = call_blocking(db.clone(), {
        let channel = caller_channel.clone();
        let summary = summary.clone();
        move |db| db.enqueue_subagent_announce(&announce_id, chat_id, &channel, &summary)
    })
    .await;
    // A UNIQUE violation here just means another sibling already enqueued it.
    if enqueue.is_ok() {
        let _ = flush_pending_announces_once(config, channel_registry, db, 10).await;
    }
}

async fn build_announce_payload(
    db: Arc<Database>,
    chat_id: i64,
    run_id: &str,
) -> Result<String, String> {
    let run_id_owned = run_id.to_string();
    let run = match call_blocking(db.clone(), move |db| {
        db.get_subagent_run(&run_id_owned, chat_id)
    })
    .await
    {
        Ok(Some(run)) => run,
        Ok(None) => return Err("run_not_found".into()),
        Err(e) => return Err(format!("failed_loading_run: {e}")),
    };

    let status_emoji = match run.status.as_str() {
        "completed" => "✅",
        "cancelled" => "🛑",
        "timed_out" => "⏱️",
        _ => "❌",
    };

    let mut text = format!(
        "{status_emoji} Subagent `{}` finished\nstatus: {}\ninput_tokens: {}\noutput_tokens: {}\nbudget: {}",
        run.run_id, run.status, run.input_tokens, run.output_tokens, run.token_budget
    );
    if let Some(err) = &run.error_text {
        text.push_str(&format!("\nerror: {err}"));
    }
    if let Some(result) = &run.result_text {
        let visible = crate::agent_engine::sanitize_user_visible_text(result);
        let clipped: String = visible.chars().take(1200).collect();
        text.push_str("\nresult:\n");
        text.push_str(&clipped);
    }
    Ok(text)
}

pub async fn flush_pending_announces_once(
    config: &Config,
    channel_registry: Arc<ChannelRegistry>,
    db: Arc<Database>,
    max_batch: usize,
) -> usize {
    let now = chrono::Utc::now().to_rfc3339();
    let rows = match call_blocking(db.clone(), move |db| {
        db.list_due_subagent_announces(&now, max_batch)
    })
    .await
    {
        Ok(v) => v,
        Err(e) => {
            warn!("failed to list due subagent announces: {e}");
            return 0;
        }
    };

    let mut processed = 0usize;
    for row in rows {
        let bot_username = config.bot_username_for_channel(&row.caller_channel);
        let delivery = deliver_and_store_bot_message(
            channel_registry.as_ref(),
            db.clone(),
            &bot_username,
            row.chat_id,
            &row.payload_text,
        )
        .await;
        match delivery {
            Ok(_) => {
                let id = row.id;
                let _ =
                    call_blocking(db.clone(), move |db| db.mark_subagent_announce_sent(id)).await;
            }
            Err(err) => {
                let next_attempts = row.attempts + 1;
                let terminal = next_attempts >= 5;
                let delay_secs = 1_i64 << next_attempts.min(6);
                let next_at = if terminal {
                    None
                } else {
                    Some((chrono::Utc::now() + chrono::Duration::seconds(delay_secs)).to_rfc3339())
                };
                let id = row.id;
                let err_text = err;
                let _ = call_blocking(db.clone(), move |db| {
                    db.mark_subagent_announce_retry(
                        id,
                        next_attempts,
                        next_at.as_deref(),
                        &err_text,
                        terminal,
                    )
                })
                .await;
            }
        }
        processed += 1;
    }
    processed
}

pub struct SessionsSpawnTool {
    config: Config,
    db: Arc<Database>,
    channel_registry: Arc<ChannelRegistry>,
}

impl SessionsSpawnTool {
    pub fn new(config: &Config, db: Arc<Database>, channel_registry: Arc<ChannelRegistry>) -> Self {
        Self {
            config: config.clone(),
            db,
            channel_registry,
        }
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "sessions_spawn".into(),
            description: format!(
                "Spawn an asynchronous sub-agent run for long tasks. Returns immediately with a run id. Use subagents_list/subagents_info/subagents_kill to manage runs. Pick a `specialist` to route the work to a focused expert. Available specialists: {}.",
                crate::tools::specialists::specialist_catalog()
            ),
            input_schema: schema_object(
                json!({
                    "task": {
                        "type": "string",
                        "description": "Task for the spawned sub-agent"
                    },
                    "context": {
                        "type": "string",
                        "description": "Optional extra context passed to the sub-agent"
                    },
                    "specialist": {
                        "type": "string",
                        "enum": crate::tools::specialists::specialist_names(),
                        "description": "Which specialist persona the sub-agent adopts. Defaults to generalist."
                    },
                    "label": {
                        "type": "string",
                        "description": "Short human-friendly name for this task (e.g. 'competitor research'), shown when listing/reporting progress. Recommended when running several tasks at once."
                    },
                    "runtime": {
                        "type": "string",
                        "enum": ["native", "acp"],
                        "description": "Execution backend for the sub-agent. Defaults to native."
                    },
                    "runtime_target": {
                        "type": "string",
                        "description": "Optional named ACP target when runtime is 'acp'."
                    },
                    "chat_id": {
                        "type": "integer",
                        "description": "Target chat id. Defaults to current chat."
                    },
                    "token_budget": {
                        "type": "integer",
                        "description": "Optional token budget cap for this run."
                    },
                    "exit_criteria": {
                        "type": "array",
                        "maxItems": 8,
                        "description": "Optional completion contract: machine-checkable exit criteria the runtime VERIFIES when the run finishes (the result is marked VERIFIED/FAILED with evidence). Entries: {type:'file_exists', path} | {type:'file_contains', path, needle} | {type:'file_min_bytes', path, min_bytes} | {type:'result_contains', needle} | {type:'command', run, expect_contains?}. Paths are relative to the chat working dir; commands run through the sandboxed bash tool. Declare criteria whenever the task has an objectively checkable outcome.",
                        "items": {"type": "object"}
                    }
                }),
                &["task"],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth =
            match auth_context_from_input(&input) {
                Some(v) => v,
                None => return ToolResult::error(
                    "sessions_spawn requires caller auth context; run from an active chat session"
                        .into(),
                ),
            };

        let chat_id = input
            .get("chat_id")
            .and_then(|v| v.as_i64())
            .unwrap_or(auth.caller_chat_id);
        if let Err(e) = authorize_chat_access(&input, chat_id) {
            return ToolResult::error(e);
        }

        let task = match input.get("task").and_then(|v| v.as_str()) {
            Some(v) if !v.trim().is_empty() => v.trim().to_string(),
            _ => return ToolResult::error("Missing required parameter: task".into()),
        };
        let context = input
            .get("context")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .trim()
            .to_string();
        // Resolve the requested specialist (falls back to generalist for unknown/empty).
        let specialist = crate::tools::specialists::resolve_specialist(
            input.get("specialist").and_then(|v| v.as_str()),
        )
        .name
        .to_string();
        let exit_criteria =
            match crate::completion_contract::parse_exit_criteria(input.get("exit_criteria")) {
                Ok(v) => v,
                Err(e) => return ToolResult::error(e),
            };
        // Optional human-friendly label ("competitor research") for "what am I working on".
        let label = input
            .get("label")
            .and_then(|v| v.as_str())
            .map(str::trim)
            .filter(|v| !v.is_empty())
            .map(|v| v.chars().take(80).collect::<String>());
        let parent_meta = subagent_runtime_meta_from_input(&input);
        let execution_runtime =
            match SubagentExecutionRuntime::from_input(&input, parent_meta.as_ref()) {
                Ok(v) => v,
                Err(e) => return ToolResult::error(e),
            };
        let runtime_target = input
            .get("runtime_target")
            .and_then(|v| v.as_str())
            .map(str::trim)
            .filter(|v| !v.is_empty())
            .map(ToOwned::to_owned)
            .or_else(|| {
                parent_meta
                    .as_ref()
                    .and_then(|meta| meta.runtime_target.clone())
            });
        let parent_depth = parent_meta.as_ref().map(|m| m.depth).unwrap_or(0);
        let parent_budget_remaining = parent_meta.as_ref().and_then(|m| m.token_budget_remaining);
        let child_depth = if parent_depth > 0 {
            parent_depth + 1
        } else {
            1
        };
        if child_depth as usize > self.config.subagents.max_spawn_depth {
            return ToolResult::error(format!(
                "subagent spawn depth exceeded: requested depth {}, max {}",
                child_depth, self.config.subagents.max_spawn_depth
            ));
        }
        let parent_run_id = input
            .get("__subagent_runtime")
            .and_then(|v| v.get("run_id"))
            .and_then(|v| v.as_str())
            .map(str::to_string);
        let requested_budget = input
            .get("token_budget")
            .and_then(|v| v.as_i64())
            .filter(|v| *v > 0);
        let child_token_budget = match compute_child_token_budget(
            requested_budget,
            parent_budget_remaining,
            self.config.subagents.max_tokens_per_run,
        ) {
            Ok(v) => v,
            Err(e) => return ToolResult::error(e),
        };

        let db_for_count = self.db.clone();
        let active_count = match call_blocking(db_for_count, move |db| {
            db.count_active_subagent_runs_for_chat(chat_id)
        })
        .await
        {
            Ok(v) => v,
            Err(e) => {
                return ToolResult::error(format!("Failed checking active subagent runs: {e}"));
            }
        };
        if active_count as usize >= self.config.subagents.max_active_per_chat {
            return ToolResult::error(format!(
                "Too many active subagent runs for this chat (limit: {})",
                self.config.subagents.max_active_per_chat
            ));
        }
        if let Some(parent_id) = parent_run_id.as_ref() {
            let parent_id_for_count = parent_id.clone();
            let active_children = match call_blocking(self.db.clone(), move |db| {
                db.count_active_subagent_children(&parent_id_for_count)
            })
            .await
            {
                Ok(v) => v,
                Err(e) => {
                    return ToolResult::error(format!(
                        "Failed checking active subagent child runs: {e}"
                    ));
                }
            };
            if active_children as usize >= self.config.subagents.max_children_per_run {
                return ToolResult::error(format!(
                    "Too many active child runs for this parent (limit: {})",
                    self.config.subagents.max_children_per_run
                ));
            }
        }

        let run_id = format!("subrun-{}", uuid::Uuid::new_v4());
        let (provider, model, acp_target) = match execution_runtime {
            SubagentExecutionRuntime::Native => (
                self.config.llm_provider.clone(),
                self.config.model.clone(),
                None,
            ),
            SubagentExecutionRuntime::Acp => {
                if !self.config.subagents.acp.default_target.enabled {
                    return ToolResult::error(
                        "ACP runtime is disabled. Set `subagents.acp.enabled: true` and configure `subagents.acp.command` or `subagents.acp.targets` first."
                            .into(),
                    );
                }
                let resolved = match self
                    .config
                    .subagents
                    .acp
                    .resolve_target(runtime_target.as_deref())
                {
                    Ok(target) => target,
                    Err(err) => return ToolResult::error(err),
                };
                (
                    crate::acp_subagent::acp_runtime_provider(resolved.name.as_deref()),
                    crate::acp_subagent::acp_runtime_model(&resolved),
                    Some(resolved),
                )
            }
        };

        let run_id_for_insert = run_id.clone();
        let task_for_insert = task.clone();
        let context_for_insert = context.clone();
        let caller_channel_for_insert = auth.caller_channel.clone();
        let parent_for_insert = parent_run_id.clone();
        let label_for_insert = label.clone();
        if let Err(e) = call_blocking(self.db.clone(), move |db| {
            db.create_subagent_run(CreateSubagentRunParams {
                run_id: &run_id_for_insert,
                parent_run_id: parent_for_insert.as_deref(),
                depth: child_depth,
                token_budget: child_token_budget,
                chat_id,
                caller_channel: &caller_channel_for_insert,
                task: &task_for_insert,
                context: &context_for_insert,
                provider: &provider,
                model: &model,
                label: label_for_insert.as_deref(),
            })
        })
        .await
        {
            return ToolResult::error(format!("Failed creating subagent run: {e}"));
        }
        log_subagent_event(
            self.db.clone(),
            &run_id,
            "accepted",
            Some(format!(
                "depth={child_depth} runtime={} specialist={specialist}{}",
                execution_runtime.as_str(),
                runtime_target
                    .as_deref()
                    .map(|target| format!(" runtime_target={target}"))
                    .unwrap_or_default()
            )),
        )
        .await;

        let runtime = subagent_runtime(&self.config);
        let local_cancel = runtime.register_run(&run_id);
        let db = self.db.clone();
        let cfg = self.config.clone();
        let run_id_async = run_id.clone();
        let task_async = task.clone();
        let context_async = context.clone();
        let specialist_async = specialist.clone();
        let exit_criteria_async = exit_criteria.clone();
        let parent_run_id_async = parent_run_id.clone();
        let auth_async = ToolAuthContext {
            caller_channel: auth.caller_channel.clone(),
            caller_chat_id: chat_id,
            principal: format!("subagent:{run_id}"),
            control_chat_ids: auth.control_chat_ids.clone(),
            env_files: auth.env_files.clone(),
        };
        let channel_registry = self.channel_registry.clone();
        let subagent_channel_registry = self.channel_registry.clone();
        let fan_in_channel_registry = self.channel_registry.clone();
        let caller_channel_for_event = auth.caller_channel.clone();
        TASK_RUNTIME.spawn(async move {
            let run_id_for_finish = run_id_async.clone();
            let _ = call_blocking(db.clone(), {
                let run_id = run_id_async.clone();
                move |db| db.mark_subagent_queued(&run_id)
            })
            .await;
            log_subagent_event(db.clone(), &run_id_async, "queued", None).await;

            let _permit = match runtime.semaphore.acquire().await {
                Ok(p) => p,
                Err(_) => {
                    let _ = call_blocking(db.clone(), move |db| {
                        db.mark_subagent_finished(FinishSubagentRunParams {
                            run_id: &run_id_for_finish,
                            status: "failed",
                            error_text: Some("subagent runtime is shutting down"),
                            result_text: None,
                            artifact_json: None,
                            input_tokens: 0,
                            output_tokens: 0,
                        })
                    })
                    .await;
                    runtime.remove_run(&run_id_async);
                    return;
                }
            };

            let _ = call_blocking(db.clone(), {
                let run_id = run_id_async.clone();
                move |db| db.mark_subagent_running(&run_id)
            })
            .await;
            log_subagent_event(db.clone(), &run_id_async, "running", None).await;

            let timeout_secs = cfg.subagents.run_timeout_secs;
            let run_future = run_sub_agent_task(RunSubAgentTaskParams {
                config: cfg.clone(),
                db: db.clone(),
                channel_registry: subagent_channel_registry,
                auth_context: auth_async,
                run_id: run_id_async.clone(),
                runtime: execution_runtime,
                acp_target,
                depth: child_depth,
                run_token_budget: child_token_budget,
                task: task_async,
                context: context_async,
                specialist: specialist_async,
                exit_criteria: exit_criteria_async,
                local_cancel,
            });

            let final_outcome = if timeout_secs > 0 {
                match tokio::time::timeout(std::time::Duration::from_secs(timeout_secs), run_future)
                    .await
                {
                    Ok(result) => result,
                    Err(_) => Err("timed_out".to_string()),
                }
            } else {
                run_future.await
            };

            let finish_status = match &final_outcome {
                Ok(_) => "completed",
                Err(err) if err == "cancelled" => "cancelled",
                Err(err) if err == "timed_out" => "timed_out",
                Err(err) if err.contains("budget_exceeded:") => "budget_exceeded",
                Err(_) => "failed",
            };
            // Surface completion into the parent turn's heartbeat if it is
            // still running (no-op otherwise).
            crate::agent_engine::send_turn_event(
                &auth.caller_channel,
                chat_id,
                crate::agent_engine::AgentEvent::SubagentFinished {
                    run_id: run_id_async.clone(),
                    status: finish_status.to_string(),
                },
            );

            match final_outcome {
                Ok((result, artifact_json, input_tokens, output_tokens)) => {
                    let rid = run_id_for_finish.clone();
                    let _ = call_blocking(db.clone(), move |db| {
                        db.mark_subagent_finished(FinishSubagentRunParams {
                            run_id: &rid,
                            status: "completed",
                            error_text: None,
                            result_text: Some(&result),
                            artifact_json: Some(&artifact_json),
                            input_tokens,
                            output_tokens,
                        })
                    })
                    .await;
                    log_subagent_event(db.clone(), &run_id_for_finish, "completed", None).await;
                }
                Err(err) if err == "cancelled" => {
                    let rid = run_id_for_finish.clone();
                    let _ = call_blocking(db.clone(), move |db| {
                        db.mark_subagent_finished(FinishSubagentRunParams {
                            run_id: &rid,
                            status: "cancelled",
                            error_text: Some("Cancelled by user"),
                            result_text: None,
                            artifact_json: None,
                            input_tokens: 0,
                            output_tokens: 0,
                        })
                    })
                    .await;
                    log_subagent_event(db.clone(), &run_id_for_finish, "cancelled", None).await;
                }
                Err(err) if err == "timed_out" => {
                    let rid = run_id_for_finish.clone();
                    let _ = call_blocking(db.clone(), move |db| {
                        db.mark_subagent_finished(FinishSubagentRunParams {
                            run_id: &rid,
                            status: "timed_out",
                            error_text: Some("Sub-agent run exceeded configured timeout"),
                            result_text: None,
                            artifact_json: None,
                            input_tokens: 0,
                            output_tokens: 0,
                        })
                    })
                    .await;
                    log_subagent_event(db.clone(), &run_id_for_finish, "timed_out", None).await;
                }
                Err(err) => {
                    let rid = run_id_for_finish.clone();
                    let err_for_db = err.clone();
                    let status = if err_for_db.contains("budget_exceeded:") {
                        "budget_exceeded"
                    } else {
                        "failed"
                    };
                    let _ = call_blocking(db.clone(), move |db| {
                        db.mark_subagent_finished(FinishSubagentRunParams {
                            run_id: &rid,
                            status,
                            error_text: Some(&err_for_db),
                            result_text: None,
                            artifact_json: None,
                            input_tokens: 0,
                            output_tokens: 0,
                        })
                    })
                    .await;
                    log_subagent_event(db.clone(), &run_id_for_finish, "failed", Some(err)).await;
                }
            }

            runtime.remove_run(&run_id_async);

            let user_chat_announces_enabled = cfg.subagents.announce_to_chat
                && auth.caller_channel != "weixin"
                && !auth.caller_channel.starts_with("weixin.");
            if user_chat_announces_enabled {
                match build_announce_payload(db.clone(), chat_id, &run_id_async).await {
                    Ok(payload) => {
                        let rid = run_id_async.clone();
                        let caller_channel = auth.caller_channel.clone();
                        let _ = call_blocking(db.clone(), move |db| {
                            db.enqueue_subagent_announce(&rid, chat_id, &caller_channel, &payload)
                        })
                        .await;
                        let _ =
                            flush_pending_announces_once(&cfg, channel_registry, db.clone(), 10)
                                .await;
                    }
                    Err(e) => {
                        warn!("failed to build announce payload for run {run_id_async}: {e}");
                    }
                }
            }

            // Fan-in: when this was the last active child of a parent run, post one
            // consolidated summary of the whole batch (opt-in).
            if user_chat_announces_enabled && cfg.subagents.fan_in_summary {
                if let Some(parent_id) = parent_run_id_async.as_ref() {
                    maybe_post_fan_in_summary(
                        &cfg,
                        fan_in_channel_registry,
                        db,
                        chat_id,
                        parent_id,
                    )
                    .await;
                }
            }
        });

        info!("subagent accepted run_id={run_id} chat_id={chat_id}");
        crate::agent_engine::send_turn_event(
            &caller_channel_for_event,
            chat_id,
            crate::agent_engine::AgentEvent::SubagentStarted {
                run_id: run_id.clone(),
                label: label.clone().unwrap_or_else(|| specialist.clone()),
            },
        );
        ToolResult::success(
            json!({
                "status": "accepted",
                "run_id": run_id,
                "chat_id": chat_id,
                "depth": child_depth,
                "specialist": specialist,
                "label": label,
                "runtime": execution_runtime.as_str(),
                "runtime_target": runtime_target,
                "token_budget": child_token_budget,
                "parent_run_id": parent_run_id,
            })
            .to_string(),
        )
    }
}

pub struct SubagentsListTool {
    db: Arc<Database>,
}

impl SubagentsListTool {
    pub fn new(db: Arc<Database>) -> Self {
        Self { db }
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "subagents_list".into(),
            description: "List recent subagent runs for the current chat.".into(),
            input_schema: schema_object(
                json!({
                    "chat_id": {"type": "integer"},
                    "limit": {"type": "integer", "minimum": 1, "maximum": 100}
                }),
                &[],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth =
            match auth_context_from_input(&input) {
                Some(v) => v,
                None => return ToolResult::error(
                    "subagents_list requires caller auth context; run from an active chat session"
                        .into(),
                ),
            };
        let chat_id = input
            .get("chat_id")
            .and_then(|v| v.as_i64())
            .unwrap_or(auth.caller_chat_id);
        if let Err(e) = authorize_chat_access(&input, chat_id) {
            return ToolResult::error(e);
        }
        let limit = input
            .get("limit")
            .and_then(|v| v.as_u64())
            .unwrap_or(20)
            .clamp(1, 100) as usize;

        let rows = match call_blocking(self.db.clone(), move |db| {
            db.list_subagent_runs(chat_id, limit)
        })
        .await
        {
            Ok(v) => v,
            Err(e) => return ToolResult::error(format!("Failed listing subagent runs: {e}")),
        };
        let payload: Vec<serde_json::Value> = rows
            .into_iter()
            .map(|r| {
                json!({
                    "run_id": r.run_id,
                    "label": r.label,
                    "parent_run_id": r.parent_run_id,
                    "depth": r.depth,
                    "token_budget": r.token_budget,
                    "status": r.status,
                    "created_at": r.created_at,
                    "started_at": r.started_at,
                    "finished_at": r.finished_at,
                    "cancel_requested": r.cancel_requested,
                    "task": r.task,
                    "progress": r.progress_text,
                    "last_progress_at": r.last_progress_at,
                    "input_tokens": r.input_tokens,
                    "output_tokens": r.output_tokens,
                    "artifact_json": r.artifact_json,
                })
            })
            .collect();

        ToolResult::success(json!({"chat_id": chat_id, "runs": payload}).to_string())
    }
}

/// Resolve a run reference (exact run_id or human-friendly label) to a run_id
/// within the given chat. Shared by subagents_info / subagents_kill.
async fn resolve_subagent_ref(
    db: &Arc<Database>,
    chat_id: i64,
    run_ref: &str,
) -> Result<Option<String>, String> {
    let run_ref_owned = run_ref.to_string();
    call_blocking(db.clone(), move |db| {
        db.resolve_subagent_run_id(chat_id, &run_ref_owned)
    })
    .await
    .map_err(|e| format!("Failed resolving subagent reference: {e}"))
}

pub struct SubagentsInfoTool {
    db: Arc<Database>,
}

impl SubagentsInfoTool {
    pub fn new(db: Arc<Database>) -> Self {
        Self { db }
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "subagents_info".into(),
            description: "Get detailed information for one subagent run, by run id or task label."
                .into(),
            input_schema: schema_object(
                json!({
                    "run_id": {"type": "string", "description": "Run id or the task label given at spawn."},
                    "chat_id": {"type": "integer"}
                }),
                &["run_id"],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth =
            match auth_context_from_input(&input) {
                Some(v) => v,
                None => return ToolResult::error(
                    "subagents_info requires caller auth context; run from an active chat session"
                        .into(),
                ),
            };
        let chat_id = input
            .get("chat_id")
            .and_then(|v| v.as_i64())
            .unwrap_or(auth.caller_chat_id);
        if let Err(e) = authorize_chat_access(&input, chat_id) {
            return ToolResult::error(e);
        }
        let run_ref = match input.get("run_id").and_then(|v| v.as_str()) {
            Some(v) if !v.trim().is_empty() => v.trim().to_string(),
            _ => return ToolResult::error("Missing required parameter: run_id".into()),
        };
        let run_id = match resolve_subagent_ref(&self.db, chat_id, &run_ref).await {
            Ok(Some(v)) => v,
            Ok(None) => return ToolResult::error(format!("No subagent run matching '{run_ref}'")),
            Err(e) => return ToolResult::error(e),
        };

        let run = match call_blocking(self.db.clone(), move |db| {
            db.get_subagent_run(&run_id, chat_id)
        })
        .await
        {
            Ok(Some(v)) => v,
            Ok(None) => return ToolResult::error("Subagent run not found".into()),
            Err(e) => return ToolResult::error(format!("Failed reading subagent run: {e}")),
        };

        ToolResult::success(
            json!({
                "run_id": run.run_id,
                "label": run.label,
                "parent_run_id": run.parent_run_id,
                "depth": run.depth,
                "chat_id": run.chat_id,
                "caller_channel": run.caller_channel,
                "task": run.task,
                "context": run.context,
                "status": run.status,
                "created_at": run.created_at,
                "started_at": run.started_at,
                "finished_at": run.finished_at,
                "cancel_requested": run.cancel_requested,
                "progress": run.progress_text,
                "last_progress_at": run.last_progress_at,
                "error_text": run.error_text,
                "result_text": run.result_text,
                "input_tokens": run.input_tokens,
                "output_tokens": run.output_tokens,
                "total_tokens": run.total_tokens,
                "provider": run.provider,
                "model": run.model,
                "token_budget": run.token_budget,
                "artifact_json": run.artifact_json,
            })
            .to_string(),
        )
    }
}

pub struct SubagentsKillTool {
    config: Config,
    db: Arc<Database>,
}

impl SubagentsKillTool {
    pub fn new(config: &Config, db: Arc<Database>) -> Self {
        Self {
            config: config.clone(),
            db,
        }
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "subagents_kill".into(),
            description: "Request cancellation for one running subagent run (by run id or task label), or all active runs in current chat with run_id=all.".into(),
            input_schema: schema_object(
                json!({
                    "run_id": {"type": "string", "description": "Run id, task label, or 'all'"},
                    "chat_id": {"type": "integer"}
                }),
                &["run_id"],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth =
            match auth_context_from_input(&input) {
                Some(v) => v,
                None => return ToolResult::error(
                    "subagents_kill requires caller auth context; run from an active chat session"
                        .into(),
                ),
            };
        let chat_id = input
            .get("chat_id")
            .and_then(|v| v.as_i64())
            .unwrap_or(auth.caller_chat_id);
        if let Err(e) = authorize_chat_access(&input, chat_id) {
            return ToolResult::error(e);
        }
        let run_ref = match input.get("run_id").and_then(|v| v.as_str()) {
            Some(v) if !v.trim().is_empty() => v.trim().to_string(),
            _ => return ToolResult::error("Missing required parameter: run_id".into()),
        };

        let runtime = subagent_runtime(&self.config);

        if run_ref.eq_ignore_ascii_case("all") {
            let rows = match call_blocking(self.db.clone(), move |db| {
                db.list_subagent_runs(chat_id, 200)
            })
            .await
            {
                Ok(v) => v,
                Err(e) => return ToolResult::error(format!("Failed listing subagent runs: {e}")),
            };
            let mut cancelled = 0usize;
            for row in rows {
                if matches!(row.status.as_str(), "accepted" | "queued" | "running") {
                    let rid = row.run_id.clone();
                    let requested = call_blocking(self.db.clone(), move |db| {
                        db.request_subagent_cancel(&rid, chat_id)
                    })
                    .await
                    .unwrap_or(false);
                    if requested {
                        runtime.cancel_run(&row.run_id);
                        log_subagent_event(
                            self.db.clone(),
                            &row.run_id,
                            "cancel_requested",
                            Some("kill_all".to_string()),
                        )
                        .await;
                        cancelled += 1;
                    }
                }
            }
            return ToolResult::success(
                json!({"status": "ok", "cancelled": cancelled, "chat_id": chat_id}).to_string(),
            );
        }

        let run_id = match resolve_subagent_ref(&self.db, chat_id, &run_ref).await {
            Ok(Some(v)) => v,
            Ok(None) => return ToolResult::error(format!("No subagent run matching '{run_ref}'")),
            Err(e) => return ToolResult::error(e),
        };

        let run_id_for_db = run_id.clone();
        let requested = match call_blocking(self.db.clone(), move |db| {
            db.request_subagent_cancel(&run_id_for_db, chat_id)
        })
        .await
        {
            Ok(v) => v,
            Err(e) => {
                return ToolResult::error(format!("Failed requesting cancellation: {e}"));
            }
        };

        if !requested {
            return ToolResult::error("Subagent run not found or already finished".into());
        }
        runtime.cancel_run(&run_id);
        log_subagent_event(
            self.db.clone(),
            &run_id,
            "cancel_requested",
            Some("kill_one".to_string()),
        )
        .await;
        ToolResult::success(json!({"status": "ok", "run_id": run_id}).to_string())
    }
}

pub struct SubagentsRetryAnnouncesTool {
    config: Config,
    db: Arc<Database>,
    channel_registry: Arc<ChannelRegistry>,
}

pub struct SubagentsFocusTool {
    db: Arc<Database>,
}

impl SubagentsFocusTool {
    pub fn new(db: Arc<Database>) -> Self {
        Self { db }
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "subagents_focus".into(),
            description: "Bind the current chat to a subagent run for follow-up actions.".into(),
            input_schema: schema_object(
                json!({
                    "run_id": {"type":"string"},
                    "chat_id": {"type":"integer"}
                }),
                &["run_id"],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth = match auth_context_from_input(&input) {
            Some(v) => v,
            None => {
                return ToolResult::error("subagents_focus requires caller auth context".into())
            }
        };
        let chat_id = input
            .get("chat_id")
            .and_then(|v| v.as_i64())
            .unwrap_or(auth.caller_chat_id);
        if let Err(e) = authorize_chat_access(&input, chat_id) {
            return ToolResult::error(e);
        }
        let run_id = match input.get("run_id").and_then(|v| v.as_str()) {
            Some(v) if !v.trim().is_empty() => v.trim().to_string(),
            _ => return ToolResult::error("Missing required parameter: run_id".into()),
        };
        let run_id_for_check = run_id.clone();
        let exists = match call_blocking(self.db.clone(), move |db| {
            db.get_subagent_run(&run_id_for_check, chat_id)
        })
        .await
        {
            Ok(v) => v.is_some(),
            Err(e) => return ToolResult::error(format!("Failed reading subagent run: {e}")),
        };
        if !exists {
            return ToolResult::error("Subagent run not found".into());
        }
        let run_id_for_set = run_id.clone();
        if let Err(e) = call_blocking(self.db.clone(), move |db| {
            db.set_subagent_focus(chat_id, &run_id_for_set)
        })
        .await
        {
            return ToolResult::error(format!("Failed setting subagent focus: {e}"));
        }
        ToolResult::success(json!({"status":"ok","chat_id":chat_id,"run_id":run_id}).to_string())
    }
}

pub struct SubagentsUnfocusTool {
    db: Arc<Database>,
}

impl SubagentsUnfocusTool {
    pub fn new(db: Arc<Database>) -> Self {
        Self { db }
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "subagents_unfocus".into(),
            description: "Clear focused subagent binding for the current chat.".into(),
            input_schema: schema_object(
                json!({
                    "chat_id": {"type":"integer"}
                }),
                &[],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth = match auth_context_from_input(&input) {
            Some(v) => v,
            None => {
                return ToolResult::error("subagents_unfocus requires caller auth context".into())
            }
        };
        let chat_id = input
            .get("chat_id")
            .and_then(|v| v.as_i64())
            .unwrap_or(auth.caller_chat_id);
        if let Err(e) = authorize_chat_access(&input, chat_id) {
            return ToolResult::error(e);
        }
        if let Err(e) =
            call_blocking(self.db.clone(), move |db| db.clear_subagent_focus(chat_id)).await
        {
            return ToolResult::error(format!("Failed clearing subagent focus: {e}"));
        }
        ToolResult::success(json!({"status":"ok","chat_id":chat_id}).to_string())
    }
}

pub struct SubagentsFocusedTool {
    db: Arc<Database>,
}

impl SubagentsFocusedTool {
    pub fn new(db: Arc<Database>) -> Self {
        Self { db }
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "subagents_focused".into(),
            description: "Show focused subagent binding for current chat.".into(),
            input_schema: schema_object(
                json!({
                    "chat_id": {"type":"integer"}
                }),
                &[],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth = match auth_context_from_input(&input) {
            Some(v) => v,
            None => {
                return ToolResult::error("subagents_focused requires caller auth context".into())
            }
        };
        let chat_id = input
            .get("chat_id")
            .and_then(|v| v.as_i64())
            .unwrap_or(auth.caller_chat_id);
        if let Err(e) = authorize_chat_access(&input, chat_id) {
            return ToolResult::error(e);
        }
        let focused =
            match call_blocking(self.db.clone(), move |db| db.get_subagent_focus(chat_id)).await {
                Ok(v) => v,
                Err(e) => return ToolResult::error(format!("Failed loading subagent focus: {e}")),
            };
        ToolResult::success(json!({"chat_id":chat_id,"run_id":focused}).to_string())
    }
}

pub struct SubagentsSendTool {
    config: Config,
    db: Arc<Database>,
    channel_registry: Arc<ChannelRegistry>,
}

impl SubagentsSendTool {
    pub fn new(config: &Config, db: Arc<Database>, channel_registry: Arc<ChannelRegistry>) -> Self {
        Self {
            config: config.clone(),
            db,
            channel_registry,
        }
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "subagents_send".into(),
            description:
                "Send follow-up work to focused subagent by spawning a child continuation run."
                    .into(),
            input_schema: schema_object(
                json!({
                    "message": {"type":"string"},
                    "chat_id": {"type":"integer"}
                }),
                &["message"],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth = match auth_context_from_input(&input) {
            Some(v) => v,
            None => return ToolResult::error("subagents_send requires caller auth context".into()),
        };
        let chat_id = input
            .get("chat_id")
            .and_then(|v| v.as_i64())
            .unwrap_or(auth.caller_chat_id);
        if let Err(e) = authorize_chat_access(&input, chat_id) {
            return ToolResult::error(e);
        }
        let message = match input.get("message").and_then(|v| v.as_str()) {
            Some(v) if !v.trim().is_empty() => v.trim().to_string(),
            _ => return ToolResult::error("Missing required parameter: message".into()),
        };
        let focused_run =
            match call_blocking(self.db.clone(), move |db| db.get_subagent_focus(chat_id)).await {
                Ok(Some(v)) => v,
                Ok(None) => return ToolResult::error("No focused subagent for this chat".into()),
                Err(e) => return ToolResult::error(format!("Failed loading subagent focus: {e}")),
            };
        let focused_run_for_load = focused_run.clone();
        let parent = match call_blocking(self.db.clone(), move |db| {
            db.get_subagent_run(&focused_run_for_load, chat_id)
        })
        .await
        {
            Ok(Some(v)) => v,
            Ok(None) => return ToolResult::error("Focused subagent run not found".into()),
            Err(e) => return ToolResult::error(format!("Failed loading focused subagent: {e}")),
        };

        let spawn_tool =
            SessionsSpawnTool::new(&self.config, self.db.clone(), self.channel_registry.clone());
        let remaining_budget = (parent.token_budget - parent.total_tokens).max(0);
        let parent_runtime = if parent.provider.starts_with("acp") {
            "acp"
        } else {
            "native"
        };
        let spawn_input = json!({
            "task": format!("Continuation request: {message}"),
            "context": format!("This is a follow-up sent to focused run {}. Continue the work based on prior run context and produce actionable output.", focused_run),
            "__microclaw_auth": {
                "caller_channel": auth.caller_channel,
                "caller_chat_id": chat_id,
                "control_chat_ids": auth.control_chat_ids,
                "env_files": auth.env_files,
            },
            "__subagent_runtime": {
                "run_id": parent.run_id,
                "depth": parent.depth,
                "runtime": parent_runtime,
                "runtime_target": crate::acp_subagent::acp_runtime_target_from_provider(&parent.provider),
                "token_budget_remaining": remaining_budget,
            }
        });
        spawn_tool.execute(spawn_input).await
    }
}

pub struct SubagentsLogTool {
    db: Arc<Database>,
}

/// One orchestrate work package: a task plus its optional completion contract.
struct OrchestratePackage {
    task: String,
    exit_criteria: Vec<crate::completion_contract::ExitCriterion>,
}

/// Marker prefix a failed contract leaves at the head of result_text
/// (produced by completion_contract::render_report).
const CONTRACT_FAILED_MARKER: &str = "[completion contract] FAILED";

pub struct SubagentsOrchestrateTool {
    config: Config,
    db: Arc<Database>,
    channel_registry: Arc<ChannelRegistry>,
}

impl SubagentsOrchestrateTool {
    pub fn new(config: &Config, db: Arc<Database>, channel_registry: Arc<ChannelRegistry>) -> Self {
        Self {
            config: config.clone(),
            db,
            channel_registry,
        }
    }

    fn merge_run_artifacts(
        runs: &[crate::internal::storage::db::SubagentRunRecord],
    ) -> serde_json::Value {
        let mut summaries = Vec::new();
        let mut findings = BTreeSet::new();
        let mut next_actions = BTreeSet::new();
        let mut artifacts = Vec::new();
        for run in runs {
            let Some(raw) = run.artifact_json.as_deref() else {
                continue;
            };
            let Ok(v) = serde_json::from_str::<serde_json::Value>(raw) else {
                continue;
            };
            if let Some(s) = v.get("summary").and_then(|x| x.as_str()) {
                if !s.trim().is_empty() {
                    summaries.push(s.trim().to_string());
                }
            }
            if let Some(arr) = v.get("findings").and_then(|x| x.as_array()) {
                for f in arr.iter().filter_map(|x| x.as_str()) {
                    if !f.trim().is_empty() {
                        findings.insert(f.trim().to_string());
                    }
                }
            }
            if let Some(arr) = v.get("next_actions").and_then(|x| x.as_array()) {
                for n in arr.iter().filter_map(|x| x.as_str()) {
                    if !n.trim().is_empty() {
                        next_actions.insert(n.trim().to_string());
                    }
                }
            }
            if let Some(arr) = v.get("artifacts").and_then(|x| x.as_array()) {
                artifacts.extend(arr.iter().cloned());
            }
        }
        json!({
            "protocol": "orchestrate_merge_v1",
            "summary": summaries,
            "findings": findings.into_iter().collect::<Vec<_>>(),
            "next_actions": next_actions.into_iter().collect::<Vec<_>>(),
            "artifacts": artifacts,
        })
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "subagents_orchestrate".into(),
            description: "Depth=2 orchestration template: spawn multiple worker runs and optionally wait+merge structured artifacts.".into(),
            input_schema: schema_object(
                json!({
                    "goal": {"type":"string"},
                    "work_packages": {"type":"array", "items":{}, "description":"Work packages: plain strings, or objects {task, exit_criteria?} where exit_criteria is a completion contract (same shape as sessions_spawn) verified when the worker finishes. With wait=true, a worker whose contract FAILED is re-spawned once with the failure evidence (contract-aware fan-in retry)."},
                    "chat_id": {"type":"integer"},
                    "wait": {"type":"boolean"},
                    "wait_timeout_secs": {"type":"integer", "minimum":1, "maximum":1200},
                    "token_budget_total": {"type":"integer", "minimum":2000}
                }),
                &["goal", "work_packages"],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth = match auth_context_from_input(&input) {
            Some(v) => v,
            None => {
                return ToolResult::error(
                    "subagents_orchestrate requires caller auth context".into(),
                )
            }
        };
        let chat_id = input
            .get("chat_id")
            .and_then(|v| v.as_i64())
            .unwrap_or(auth.caller_chat_id);
        if let Err(e) = authorize_chat_access(&input, chat_id) {
            return ToolResult::error(e);
        }
        let goal = match input.get("goal").and_then(|v| v.as_str()) {
            Some(v) if !v.trim().is_empty() => v.trim().to_string(),
            _ => return ToolResult::error("Missing required parameter: goal".into()),
        };
        let raw_packages = input
            .get("work_packages")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default();
        let mut packages: Vec<OrchestratePackage> = Vec::new();
        for (i, v) in raw_packages.iter().enumerate() {
            if let Some(task) = v.as_str() {
                if !task.trim().is_empty() {
                    packages.push(OrchestratePackage {
                        task: task.trim().to_string(),
                        exit_criteria: Vec::new(),
                    });
                }
                continue;
            }
            if v.is_object() {
                let task = v
                    .get("task")
                    .and_then(|t| t.as_str())
                    .map(str::trim)
                    .unwrap_or_default()
                    .to_string();
                if task.is_empty() {
                    return ToolResult::error(format!(
                        "work_packages[{i}] object is missing a non-empty `task`"
                    ));
                }
                let exit_criteria =
                    match crate::completion_contract::parse_exit_criteria(v.get("exit_criteria")) {
                        Ok(c) => c,
                        Err(e) => {
                            return ToolResult::error(format!("work_packages[{i}]: {e}"));
                        }
                    };
                packages.push(OrchestratePackage {
                    task,
                    exit_criteria,
                });
                continue;
            }
            return ToolResult::error(format!(
                "work_packages[{i}] must be a string or an object {{task, exit_criteria?}}"
            ));
        }
        if packages.is_empty() {
            return ToolResult::error("work_packages must include at least one item".into());
        }
        let max_workers = self
            .config
            .subagents
            .orchestrate_max_workers
            .min(self.config.subagents.max_children_per_run);
        let wait = input.get("wait").and_then(|v| v.as_bool()).unwrap_or(false);
        let wait_timeout_secs = input
            .get("wait_timeout_secs")
            .and_then(|v| v.as_u64())
            .unwrap_or(120)
            .clamp(1, 1200);
        let total_budget = input
            .get("token_budget_total")
            .and_then(|v| v.as_i64())
            .filter(|v| *v > 0)
            .unwrap_or(self.config.subagents.max_tokens_per_run);
        let parent_run_id = input
            .get("__subagent_runtime")
            .and_then(|v| v.get("run_id"))
            .and_then(|v| v.as_str())
            .map(str::to_string);
        let parent_meta = subagent_runtime_meta_from_input(&input);
        let parent_remaining = parent_meta.as_ref().and_then(|m| m.token_budget_remaining);
        let total_budget = if let Some(remaining) = parent_remaining {
            total_budget.min(remaining.max(0))
        } else {
            total_budget
        };
        if total_budget < 2_000 {
            return ToolResult::error(format!(
                "subagent budget exhausted for orchestration: remaining {} < 2000",
                total_budget
            ));
        }
        let max_workers_by_budget = (total_budget / 2_000).max(1) as usize;
        let allowed_workers = max_workers.min(max_workers_by_budget).max(1);
        if packages.len() > allowed_workers {
            packages.truncate(allowed_workers);
        }
        let each_budget = (total_budget / packages.len() as i64).clamp(2_000, total_budget);

        let spawn_tool =
            SessionsSpawnTool::new(&self.config, self.db.clone(), self.channel_registry.clone());
        let mut spawned = Vec::new();
        for (idx, pkg) in packages.iter().enumerate() {
            let mut spawn_input = json!({
                "task": format!("Work package {}: {}", idx + 1, pkg.task),
                "context": format!(
                    "Orchestration goal: {goal}\nOutput strictly in protocol subagent_artifact_v1 with keys summary/findings/artifacts/next_actions/final_answer."
                ),
                "chat_id": chat_id,
                "token_budget": each_budget,
                "__microclaw_auth": {
                    "caller_channel": auth.caller_channel.clone(),
                    "caller_chat_id": chat_id,
                    "control_chat_ids": auth.control_chat_ids.clone(),
                    "env_files": auth.env_files.clone(),
                }
            });
            if !pkg.exit_criteria.is_empty() {
                if let Some(obj) = spawn_input.as_object_mut() {
                    obj.insert(
                        "exit_criteria".to_string(),
                        serde_json::to_value(&pkg.exit_criteria).unwrap_or_default(),
                    );
                }
            }
            let spawn_input = if let (Some(meta), Some(parent_run_id)) =
                (parent_meta.as_ref(), parent_run_id.as_deref())
            {
                let mut obj = spawn_input
                    .as_object()
                    .cloned()
                    .unwrap_or_else(serde_json::Map::new);
                obj.insert(
                    "__subagent_runtime".to_string(),
                    json!({
                        "run_id": parent_run_id,
                        "depth": meta.depth,
                        "runtime": meta.runtime.as_deref().unwrap_or("native"),
                        "runtime_target": meta.runtime_target.clone(),
                        "token_budget_remaining": each_budget,
                    }),
                );
                serde_json::Value::Object(obj)
            } else {
                spawn_input
            };
            let res = spawn_tool.execute(spawn_input).await;
            if res.is_error {
                return res;
            }
            let parsed = match serde_json::from_str::<serde_json::Value>(&res.content) {
                Ok(v) => v,
                Err(e) => return ToolResult::error(format!("Invalid sessions_spawn output: {e}")),
            };
            let run_id = match parsed.get("run_id").and_then(|v| v.as_str()) {
                Some(v) => v.to_string(),
                None => return ToolResult::error("sessions_spawn did not return run_id".into()),
            };
            spawned.push(run_id);
        }

        if !wait {
            return ToolResult::success(
                json!({
                    "status": "accepted",
                    "chat_id": chat_id,
                    "goal": goal,
                    "workers": spawned.len(),
                    "run_ids": spawned,
                    "token_budget_total": total_budget,
                    "token_budget_each": each_budget,
                })
                .to_string(),
            );
        }

        let deadline =
            std::time::Instant::now() + std::time::Duration::from_secs(wait_timeout_secs);
        let mut runs = Vec::new();
        let mut wait_timed_out = false;
        while std::time::Instant::now() < deadline {
            let ids = spawned.clone();
            let rows = match call_blocking(self.db.clone(), move |db| {
                let mut out = Vec::new();
                for run_id in ids {
                    if let Some(row) = db.get_subagent_run(&run_id, chat_id)? {
                        out.push(row);
                    }
                }
                Ok::<_, microclaw_core::error::MicroClawError>(out)
            })
            .await
            {
                Ok(v) => v,
                Err(e) => return ToolResult::error(format!("Failed loading runs: {e}")),
            };
            let done = rows
                .iter()
                .all(|r| !matches!(r.status.as_str(), "accepted" | "queued" | "running"));
            runs = rows;
            if done {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        }
        if runs
            .iter()
            .any(|r| matches!(r.status.as_str(), "accepted" | "queued" | "running"))
        {
            wait_timed_out = true;
        }

        // Contract-aware fan-in retry: a completed worker whose result carries
        // the FAILED contract marker gets exactly ONE replacement run, spawned
        // with the failure evidence as extra context. (The in-run bounded
        // retry has already happened inside the worker, so this catches the
        // cases where the worker gave up.)
        let mut retried: Vec<serde_json::Value> = Vec::new();
        if !wait_timed_out {
            let mut replacements: Vec<(usize, String)> = Vec::new(); // (runs idx, new run_id)
            for (idx, run) in runs.iter().enumerate() {
                if run.status != "completed" {
                    continue;
                }
                let Some(result_text) = run.result_text.as_deref() else {
                    continue;
                };
                if !result_text.trim_start().starts_with(CONTRACT_FAILED_MARKER) {
                    continue;
                }
                // Recover the package (spawn order matches `packages`).
                let Some(pkg_idx) = spawned.iter().position(|id| *id == run.run_id) else {
                    continue;
                };
                let Some(pkg) = packages.get(pkg_idx) else {
                    continue;
                };
                let evidence_end = microclaw_core::text::floor_char_boundary(result_text, 600);
                let mut spawn_input = json!({
                    "task": format!("Work package {} (retry): {}", pkg_idx + 1, pkg.task),
                    "context": format!(
                        "Orchestration goal: {goal}\nA previous attempt FAILED its completion contract. Evidence:\n{}\nComplete the unmet criteria.\nOutput strictly in protocol subagent_artifact_v1 with keys summary/findings/artifacts/next_actions/final_answer.",
                        &result_text[..evidence_end]
                    ),
                    "chat_id": chat_id,
                    "token_budget": each_budget,
                    "exit_criteria": serde_json::to_value(&pkg.exit_criteria).unwrap_or_default(),
                    "__microclaw_auth": {
                        "caller_channel": auth.caller_channel.clone(),
                        "caller_chat_id": chat_id,
                        "control_chat_ids": auth.control_chat_ids.clone(),
                        "env_files": auth.env_files.clone(),
                    }
                });
                if let (Some(meta), Some(parent_run_id)) =
                    (parent_meta.as_ref(), parent_run_id.as_deref())
                {
                    if let Some(obj) = spawn_input.as_object_mut() {
                        obj.insert(
                            "__subagent_runtime".to_string(),
                            json!({
                                "run_id": parent_run_id,
                                "depth": meta.depth,
                                "runtime": meta.runtime.as_deref().unwrap_or("native"),
                                "runtime_target": meta.runtime_target.clone(),
                                "token_budget_remaining": each_budget,
                            }),
                        );
                    }
                }
                let res = spawn_tool.execute(spawn_input).await;
                if res.is_error {
                    // Budget/depth exhausted: keep the failed result as-is.
                    continue;
                }
                let new_run_id = serde_json::from_str::<serde_json::Value>(&res.content)
                    .ok()
                    .and_then(|v| v.get("run_id").and_then(|r| r.as_str()).map(str::to_string));
                if let Some(new_run_id) = new_run_id {
                    retried.push(json!({
                        "package": pkg_idx + 1,
                        "failed_run_id": run.run_id.clone(),
                        "retry_run_id": new_run_id.clone(),
                    }));
                    replacements.push((idx, new_run_id));
                }
            }

            if !replacements.is_empty() {
                // Wait for the replacement runs within the remaining deadline.
                let retry_ids: Vec<String> =
                    replacements.iter().map(|(_, id)| id.clone()).collect();
                loop {
                    let ids = retry_ids.clone();
                    let rows = match call_blocking(self.db.clone(), move |db| {
                        let mut out = Vec::new();
                        for run_id in ids {
                            if let Some(row) = db.get_subagent_run(&run_id, chat_id)? {
                                out.push(row);
                            }
                        }
                        Ok::<_, microclaw_core::error::MicroClawError>(out)
                    })
                    .await
                    {
                        Ok(v) => v,
                        Err(e) => return ToolResult::error(format!("Failed loading runs: {e}")),
                    };
                    let done = rows
                        .iter()
                        .all(|r| !matches!(r.status.as_str(), "accepted" | "queued" | "running"));
                    if done {
                        // Swap the finished replacements into the merge set.
                        for (idx, retry_id) in &replacements {
                            if let Some(row) = rows.iter().find(|r| r.run_id == *retry_id) {
                                runs[*idx] = row.clone();
                            }
                        }
                        break;
                    }
                    if std::time::Instant::now() >= deadline {
                        wait_timed_out = true;
                        break;
                    }
                    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
                }
            }
        }

        let merged = Self::merge_run_artifacts(&runs);
        ToolResult::success(
            json!({
                "status": if wait_timed_out { "timeout_partial" } else { "merged" },
                "chat_id": chat_id,
                "goal": goal,
                "workers": spawned.len(),
                "run_ids": spawned,
                "retried": retried,
                "runs": runs.into_iter().map(|r| json!({
                    "run_id": r.run_id,
                    "status": r.status,
                    "depth": r.depth,
                    "token_budget": r.token_budget,
                    "total_tokens": r.total_tokens,
                    "artifact_json": r.artifact_json,
                })).collect::<Vec<_>>(),
                "merged": merged,
            })
            .to_string(),
        )
    }
}

impl SubagentsLogTool {
    pub fn new(db: Arc<Database>) -> Self {
        Self { db }
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "subagents_log".into(),
            description: "Get timeline events for one subagent run.".into(),
            input_schema: schema_object(
                json!({
                    "run_id": {"type":"string"},
                    "chat_id": {"type":"integer"},
                    "limit": {"type":"integer", "minimum":1, "maximum":200}
                }),
                &["run_id"],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth = match auth_context_from_input(&input) {
            Some(v) => v,
            None => return ToolResult::error("subagents_log requires caller auth context".into()),
        };
        let chat_id = input
            .get("chat_id")
            .and_then(|v| v.as_i64())
            .unwrap_or(auth.caller_chat_id);
        if let Err(e) = authorize_chat_access(&input, chat_id) {
            return ToolResult::error(e);
        }
        let run_id = match input.get("run_id").and_then(|v| v.as_str()) {
            Some(v) if !v.trim().is_empty() => v.trim().to_string(),
            _ => return ToolResult::error("Missing required parameter: run_id".into()),
        };
        let run_id_for_check = run_id.clone();
        let run_exists = match call_blocking(self.db.clone(), move |db| {
            db.get_subagent_run(&run_id_for_check, chat_id)
        })
        .await
        {
            Ok(v) => v.is_some(),
            Err(e) => return ToolResult::error(format!("Failed reading subagent run: {e}")),
        };
        if !run_exists {
            return ToolResult::error("Subagent run not found".into());
        }
        let limit = input
            .get("limit")
            .and_then(|v| v.as_u64())
            .unwrap_or(50)
            .clamp(1, 200) as usize;
        let run_id_for_events = run_id.clone();
        let events = match call_blocking(self.db.clone(), move |db| {
            db.list_subagent_events(&run_id_for_events, limit)
        })
        .await
        {
            Ok(v) => v,
            Err(e) => return ToolResult::error(format!("Failed listing subagent events: {e}")),
        };
        let payload: Vec<serde_json::Value> = events
            .into_iter()
            .map(|e| {
                json!({
                    "id": e.id,
                    "run_id": e.run_id,
                    "event_type": e.event_type,
                    "detail": e.detail,
                    "created_at": e.created_at
                })
            })
            .collect();
        ToolResult::success(json!({"run_id": run_id, "events": payload}).to_string())
    }
}

impl SubagentsRetryAnnouncesTool {
    pub fn new(config: &Config, db: Arc<Database>, channel_registry: Arc<ChannelRegistry>) -> Self {
        Self {
            config: config.clone(),
            db,
            channel_registry,
        }
    }
}

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

    fn definition(&self) -> ToolDefinition {
        ToolDefinition {
            name: "subagents_retry_announces".into(),
            description:
                "Manually flush pending subagent completion announcements (control chats only)."
                    .into(),
            input_schema: schema_object(
                json!({
                    "batch": {"type": "integer", "minimum": 1, "maximum": 200}
                }),
                &[],
            ),
        }
    }

    async fn execute(&self, input: serde_json::Value) -> ToolResult {
        let auth = match auth_context_from_input(&input) {
            Some(v) => v,
            None => {
                return ToolResult::error(
                    "subagents_retry_announces requires caller auth context".into(),
                )
            }
        };
        if !auth.is_control_chat() {
            return ToolResult::error(
                "Permission denied: subagents_retry_announces requires control chat".into(),
            );
        }
        let batch = input
            .get("batch")
            .and_then(|v| v.as_u64())
            .unwrap_or(50)
            .clamp(1, 200) as usize;
        let _ = flush_pending_announces_once(
            &self.config,
            self.channel_registry.clone(),
            self.db.clone(),
            batch,
        )
        .await;
        ToolResult::success(json!({"status":"ok","batch":batch}).to_string())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::WorkingDirIsolation;

    fn test_config() -> Config {
        let mut cfg = Config::test_defaults();
        cfg.model = "claude-test".into();
        cfg.max_tokens = 2048;
        cfg.data_dir = "/tmp".into();
        cfg.working_dir = "/tmp".into();
        cfg.working_dir_isolation = WorkingDirIsolation::Shared;
        cfg.web_enabled = false;
        cfg
    }

    fn test_db() -> Arc<Database> {
        let dir = std::env::temp_dir().join(format!(
            "microclaw_subagents_tool_test_{}",
            uuid::Uuid::new_v4()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        Arc::new(Database::new(dir.to_str().unwrap()).unwrap())
    }

    #[tokio::test]
    async fn test_sessions_spawn_requires_task() {
        let tool =
            SessionsSpawnTool::new(&test_config(), test_db(), Arc::new(ChannelRegistry::new()));
        let result = tool
            .execute(json!({"__microclaw_auth": {"caller_channel":"web", "caller_chat_id": 1}}))
            .await;
        assert!(result.is_error);
        assert!(result.content.contains("task"));
    }

    #[tokio::test]
    async fn test_sessions_spawn_rejects_unknown_runtime() {
        let tool =
            SessionsSpawnTool::new(&test_config(), test_db(), Arc::new(ChannelRegistry::new()));
        let result = tool
            .execute(json!({
                "task": "run",
                "runtime": "mystery",
                "__microclaw_auth": {"caller_channel":"web", "caller_chat_id": 1}
            }))
            .await;
        assert!(result.is_error);
        assert!(result.content.contains("Unsupported subagent runtime"));
    }

    #[tokio::test]
    async fn test_sessions_spawn_rejects_disabled_acp_runtime() {
        let tool =
            SessionsSpawnTool::new(&test_config(), test_db(), Arc::new(ChannelRegistry::new()));
        let result = tool
            .execute(json!({
                "task": "run",
                "runtime": "acp",
                "__microclaw_auth": {"caller_channel":"web", "caller_chat_id": 1}
            }))
            .await;
        assert!(result.is_error);
        assert!(result.content.contains("ACP runtime is disabled"));
    }

    #[tokio::test]
    async fn test_sessions_spawn_rejects_unknown_acp_target() {
        let mut cfg = test_config();
        cfg.subagents.acp.default_target.enabled = true;
        cfg.subagents.acp.default_target.command = "codex".into();
        let tool = SessionsSpawnTool::new(&cfg, test_db(), Arc::new(ChannelRegistry::new()));
        let result = tool
            .execute(json!({
                "task": "run",
                "runtime": "acp",
                "runtime_target": "missing",
                "__microclaw_auth": {"caller_channel":"web", "caller_chat_id": 1}
            }))
            .await;
        assert!(result.is_error);
        assert!(result.content.contains("Unknown ACP runtime target"));
    }

    #[tokio::test]
    async fn test_sessions_spawn_inherits_acp_runtime_target_from_parent_meta() {
        let mut cfg = test_config();
        cfg.subagents.acp.default_target.enabled = true;
        cfg.subagents.acp.targets.insert(
            "worker".into(),
            crate::config::SubagentAcpTargetConfig {
                enabled: true,
                command: "codex".into(),
                ..crate::config::SubagentAcpTargetConfig::default()
            },
        );
        let tool = SessionsSpawnTool::new(&cfg, test_db(), Arc::new(ChannelRegistry::new()));
        let result = tool
            .execute(json!({
                "task": "run",
                "__microclaw_auth": {"caller_channel":"web", "caller_chat_id": 1},
                "__subagent_runtime": {
                    "run_id": "parent",
                    "depth": 0,
                    "runtime": "acp",
                    "runtime_target": "worker",
                    "token_budget_remaining": 8000
                }
            }))
            .await;
        assert!(!result.is_error);
        let parsed: serde_json::Value = serde_json::from_str(&result.content).unwrap();
        assert_eq!(parsed.get("runtime").and_then(|v| v.as_str()), Some("acp"));
        assert_eq!(
            parsed.get("runtime_target").and_then(|v| v.as_str()),
            Some("worker")
        );
    }

    #[tokio::test]
    async fn test_subagents_info_requires_run_id() {
        let tool = SubagentsInfoTool::new(test_db());
        let result = tool
            .execute(json!({"__microclaw_auth": {"caller_channel":"web", "caller_chat_id": 1}}))
            .await;
        assert!(result.is_error);
        assert!(result.content.contains("run_id"));
    }

    #[test]
    fn test_normalize_strips_leaked_thinking() {
        // A sub-agent that emits only chain-of-thought (no JSON) must not leak
        // the <think> block into the announced answer.
        let raw = "<think>The write_memory tool is not available - I see read_memory \
                   but not write_memory. Let me inform the user.</think>";
        let (answer, envelope) = normalize_subagent_artifact_payload(raw);
        assert!(!answer.contains("<think>"));
        assert!(!answer.contains("write_memory"));
        assert!(!envelope.contains("<think>"));
    }

    #[test]
    fn contract_failed_marker_matches_report_rendering() {
        // The fan-in retry detects failed contracts by this marker; it must
        // stay in sync with completion_contract::render_report.
        let report = crate::completion_contract::render_report(&[
            crate::completion_contract::CriterionOutcome {
                description: "file `x` exists".into(),
                passed: false,
                evidence: "missing".into(),
            },
        ]);
        assert!(report.starts_with(CONTRACT_FAILED_MARKER));
        // And the VERIFIED path must NOT match the failure marker.
        let ok_report = crate::completion_contract::render_report(&[
            crate::completion_contract::CriterionOutcome {
                description: "file `x` exists".into(),
                passed: true,
                evidence: "ok".into(),
            },
        ]);
        assert!(!ok_report.starts_with(CONTRACT_FAILED_MARKER));
    }

    #[test]
    fn test_submit_result_tool_definition_shape() {
        let def = submit_result_tool_definition();
        assert_eq!(def.name, SUBMIT_RESULT_TOOL);
        let props = def.input_schema.get("properties").unwrap();
        for key in [
            "summary",
            "findings",
            "artifacts",
            "next_actions",
            "final_answer",
        ] {
            assert!(props.get(key).is_some(), "missing schema property {key}");
        }
        let required = def
            .input_schema
            .get("required")
            .unwrap()
            .as_array()
            .unwrap();
        assert!(required.iter().any(|v| v == "final_answer"));
        // A submit_result payload normalizes into the announced answer.
        let payload = json!({"summary": "s", "final_answer": "the answer"}).to_string();
        let (answer, _envelope) = normalize_subagent_artifact_payload(&payload);
        assert_eq!(answer, "the answer");
    }

    #[test]
    fn test_subagent_output_is_structured() {
        // Prose / leaked reasoning is NOT structured (would trigger a repair).
        assert!(!subagent_output_is_structured(
            "<think>no write_memory tool</think> I can't save this."
        ));
        assert!(!subagent_output_is_structured(
            "Done, everything looks good."
        ));
        // A contract object (even after a thinking block) IS structured.
        assert!(subagent_output_is_structured(
            "<think>ok</think>{\"summary\":\"s\",\"final_answer\":\"done\"}"
        ));
        // An object with only empty contract fields is not structured.
        assert!(!subagent_output_is_structured(
            "{\"summary\":\"\",\"final_answer\":\"\"}"
        ));
    }

    #[test]
    fn test_normalize_recovers_json_after_thinking() {
        // Reasoning block followed by the contract JSON: the JSON must win, not
        // the raw text fallback.
        let raw = "<think>let me structure this</think>\n\
                   {\"summary\":\"done\",\"findings\":[\"a\"],\"artifacts\":[],\
                   \"next_actions\":[],\"final_answer\":\"All set.\"}";
        let (answer, _envelope) = normalize_subagent_artifact_payload(raw);
        assert_eq!(answer, "All set.");
    }

    #[test]
    fn test_compute_child_token_budget_respects_parent_remaining() {
        let budget = compute_child_token_budget(Some(5000), Some(3000), 120000).unwrap();
        assert_eq!(budget, 3000);
    }

    #[test]
    fn test_compute_child_token_budget_rejects_exhausted_parent() {
        let err = compute_child_token_budget(None, Some(1500), 120000).unwrap_err();
        assert!(err.contains("budget exhausted"));
    }

    #[test]
    fn process_lifetime_executor_outlives_foreground_runtime() {
        let (tx, rx) = std::sync::mpsc::channel();
        {
            let foreground = tokio::runtime::Runtime::new().unwrap();
            foreground.block_on(async {
                TASK_RUNTIME.spawn(async move {
                    tokio::time::sleep(std::time::Duration::from_millis(30)).await;
                    tx.send("finished").unwrap();
                });
            });
        }
        assert_eq!(
            rx.recv_timeout(std::time::Duration::from_secs(2)).unwrap(),
            "finished"
        );
    }
}