# subagent_runner.py — Subagent execution: profile-based tool access, cancellation,
# concurrency limits, retry with backoff, structured results, graceful shutdown.
from __future__ import annotations

import atexit
import contextvars
import logging
import threading
import time
import traceback
from dataclasses import dataclass
from typing import Any
from uuid import uuid4

from tools import get_tools
from utils.delegation_blackboard import (
    append_system_footer,
    get_handoff_auto_config,
    read_blackboard_excerpt,
    resolve_delegation_scratch_path,
)
from utils.handoff_critic import maybe_critique_handoff
from utils.handoff_envelope import HandoffEnvelope, envelope_to_prompt_block
from utils.agent_loop_best_effort import best_effort, best_effort_value

log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Profile registry — each profile declares allowed tools, system hint, and
# whether personality files should be injected (default: no).
# ---------------------------------------------------------------------------

SUBAGENT_PROFILES: dict[str, dict] = {
    # --- Legacy aliases (backward-compat with AGENT_TYPES) ---
    "research": {
        "tools": {
            "deep_web_search", "read_and_analyze_page", "analyze_multiple_sources",
            "append_research_scratch", "search_web", "search_web_public",
            "read_webpage", "save_research_findings", "generate_final_research_report",
            "suggest_follow_up_queries", "reflect_research_state", "plan_research_subquestions", "summarize_url",
            "query_knowledge", "ask_studied_corpus",
            "append_delegation_scratch", "read_delegation_scratch",
        },
        "research_mode": True,
        "inject_personality": False,
        "system_hint": (
            "You are a research specialist. Return concise, source-grounded findings. "
            "When an 'Ingested Knowledge' block appears in your context (or the user has a "
            "studied corpus on the topic), prefer query_knowledge / ask_studied_corpus before "
            "the web; otherwise rely on web search. "
            "If a kanban completion contract is present, keep the final visible summary and kanban-result "
            "fence outside <think> tags."
        ),
        "timeout_seconds": 600,
        "max_retries": None,
    },
    "bash": {
        "tools": {
            "shell_execute", "get_current_directory", "read_file", "read_file_chunk",
            "write_text_file", "list_directory",
            "append_delegation_scratch", "read_delegation_scratch",
        },
        "research_mode": False,
        "inject_personality": False,
        "system_hint": "You are a command-line specialist.",
        "timeout_seconds": None,
        "max_retries": None,
    },
    "general": {
        "tools": None,
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "You are a general-purpose assistant. "
            "When a kanban completion contract is present, keep the visible summary and "
            "trailing ```kanban-result``` JSON fence outside <think> tags."
        ),
        "timeout_seconds": 900,
        "max_retries": None,
    },
    # --- New expanded profiles ---
    "web_research_light": {
        "tools": {
            "search_web", "search_web_public", "read_webpage", "summarize_url",
            "read_and_analyze_page", "analyze_multiple_sources",
            "suggest_follow_up_queries", "reflect_research_state", "plan_research_subquestions", "download_file",
            "query_knowledge", "ask_studied_corpus",
            "append_delegation_scratch", "read_delegation_scratch",
        },
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "You are a focused web research assistant. Search the web, read pages, "
            "and return a concise summary with key findings and source URLs. "
            "When an 'Ingested Knowledge' block appears in your context (or the user has a "
            "studied corpus on the topic), prefer query_knowledge / ask_studied_corpus before "
            "the web; otherwise rely on web search. "
            "Do not speculate beyond what the sources state."
        ),
        "timeout_seconds": 180,
        "max_retries": 1,
    },
    "code_readonly": {
        "tools": {
            "read_file", "read_file_chunk", "list_directory", "search_filesystem",
            "get_current_directory", "get_file_metadata", "verify_file_exists",
            "append_delegation_scratch", "read_delegation_scratch",
        },
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "You are a code analysis assistant. Read files, explore directory structures, "
            "and return clear, structured answers about the codebase. Do not modify any files."
        ),
        "timeout_seconds": 90,
        "max_retries": 0,
    },
    "extraction": {
        "tools": {
            "extract_pdf_text", "read_file", "read_file_chunk", "read_webpage",
            "summarize_url", "download_file", "get_file_metadata",
            "ocr_scanned_pdf", "ocr_scanned_pdf_to_file",
            "ocr_image_text",
            "ocr_complex_document", "ocr_complex_document_to_file",
            "ocr_with_kraken_api", "ocr_with_kraken_api_to_file",
            "append_delegation_scratch", "read_delegation_scratch",
        },
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "You are a data extraction specialist. Extract structured information from "
            "documents, web pages, and files. Return results in a clean, organized format."
        ),
        "timeout_seconds": 180,
        "max_retries": 1,
    },
    "ops_cautious": {
        "tools": {
            "shell_execute", "get_current_directory", "read_file", "read_file_chunk",
            "write_text_file", "list_directory", "search_filesystem", "get_file_metadata",
            "verify_file_exists", "copy_file", "move_file",
            # Allow cautious workers to execute explicit Hyperframes render tasks.
            "hyperframes_create_video", "hyperframes_render",
            "hyperframes_lint", "hyperframes_doctor",
            "append_delegation_scratch", "read_delegation_scratch",
        },
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "You are a cautious operations assistant. Execute commands carefully, "
            "verify before modifying, and report results clearly."
        ),
        "timeout_seconds": 180,
        "max_retries": 0,
    },
    # --- Heartbeat profiles: proactive allowlist vs profile tools (intentional split) ---
    #
    # 1) heartbeat_standard and heartbeat_maintenance use their profile tool lists directly. Bounded
    #    autonomy_mode does not intersect those lists with the proactive allowlist.
    #
    # 2) heartbeat_drive_action uses its profile tools intersected with the proactive allowlist;
    #    enforcement lives in mugi_runner._run_heartbeat_sync.
    #
    # 3) Why: standard/growth ticks need drive-admin tools (list_drives, resolve_drive,
    #    dismiss_drive, create_drive) to manage drives. drive_action runs one bounded proactive
    #    step and should stay on workspace/research-style tools only—no drive CRUD.
    #
    "heartbeat_standard": {
        "tools": {
            "read_file", "read_file_chunk", "list_directory", "read_webpage", "append_to_memory",
            "write_session_state", "get_current_time", "get_upcoming_calendar_events",
            "get_weather", "check_feeds", "get_reminders",
            "query_knowledge",
            "ask_studied_corpus",
            "list_drives", "resolve_drive", "dismiss_drive", "create_drive",
            "list_watch_targets", "compare_watch_baseline",
            "list_feeds",
        },
        "research_mode": False,
        "inject_personality": True,
        "system_hint": "You are a background heartbeat assistant. Be concise and proactive.",
        "timeout_seconds": 120,
        "max_retries": 0,
    },
    "heartbeat_maintenance": {
        "tools": {
            "read_file", "read_file_chunk", "list_directory",
            "search_memory", "summarize_old_memory", "get_daily_memory_paths",
            "query_knowledge",
            "ask_studied_corpus", "refresh_memory",
            "list_drives", "resolve_drive", "dismiss_drive", "create_drive",
            "search_web", "read_webpage",
        },
        "research_mode": False,
        "inject_personality": True,
        "system_hint": (
            "You review what has happened since your last check-in. "
            "Write brief in-character observations with continuity and restraint. "
            "List [USER_FACTS] as bullets only — post-processing routes personal facts "
            "to reviewable proposals and project facts to long-term memory; do not call "
            "append_to_memory yourself. Cross-topic links belong in consolidation; "
            "when noting a thread across sessions, use concrete language the main agent can cite naturally."
        ),
        "timeout_seconds": 180,
        "max_retries": 0,
    },
    "heartbeat_drive_action": {
        "tools": {
            "read_file", "read_file_chunk", "search_web",
            "search_web_public", "read_webpage",
            "query_knowledge",
            "ask_studied_corpus",
            "append_to_memory", "write_session_state",
            "list_drives", "resolve_drive", "dismiss_drive",
        },
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "You execute exactly one bounded proactive action for a selected drive. "
            "Use proactive-safe tools only, provide concrete evidence, and stop. "
            "After research (search_web/read_webpage/query_knowledge), persist project/work "
            "findings via write_session_state (preferred) or append_to_memory for durable "
            "non-personal research outcomes before resolving. Do not infer or store personal "
            "preferences or identity facts here — those belong in User Model proposals, not "
            "drive_action persistence."
        ),
        "timeout_seconds": 150,
        "max_retries": 0,
    },
    "hooks": {
        "tools": {
            "read_file", "read_file_chunk", "search_web", "search_memory", "list_directory",
            "skills_list",
        },
        "research_mode": False,
        "inject_personality": False,
        "system_hint": "You are responding to an automated system event. Be concise and factual. Read-only tools only.",
        "timeout_seconds": 60,
        "max_retries": 0,
    },
    # Active Memory: a *read-only* pre-reply memory sub-agent. Run via
    # ``run_subagent(agent_type="active_memory", force_utility_model=True, ...)``;
    # the utility model (mlx-vlm) executes with this tight tool allowlist and
    # a hard timeout. Output is strictly "NONE" or up to ~5 fact bullets — see
    # ``utils/active_memory.py`` for the prompt templates and the untrusted
    # system-context injection in ``agent_loop.run_raw_agent``.
    "active_memory": {
        "tools": {"search_memory", "read_file", "read_file_chunk", "query_knowledge"},
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "Return at most 5 bullet facts about the user relevant to their latest message. "
            "Return exactly NONE when nothing is relevant. No prose, no questions."
        ),
        "timeout_seconds": 6,
        "max_retries": 0,
    },
    # /btw ephemeral side-question mode. See utils/btw.py. This profile is a
    # strict, stateless, read-only helper: no memory writes, no tool-augmented
    # research, and output is intentionally short. The btw handler prefers the
    # utility model and only falls back to a subagent when the utility model is
    # unavailable; this profile is what that fallback uses.
    "btw": {
        "tools": set(),
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "You are answering a one-off side question from the user. "
            "Use only the brief conversation snippet provided plus general knowledge. "
            "Be direct and helpful. No preamble. No tools. No follow-up questions. "
            "Keep the answer under ~400 words unless the question genuinely requires more."
        ),
        "timeout_seconds": 30,
        "max_retries": 0,
    },
    "council_researcher": {
        "tools": {
            "deep_web_search", "read_webpage", "read_file", "read_file_chunk",
            "query_knowledge", "ask_studied_corpus",
        },
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "Council Researcher (Athena). Pre-seeded product excerpts are authoritative — "
            "never read_file utils/, docs/, or mesh/ paths (dev checkout is not on disk for "
            "read_file). When an 'Ingested Knowledge' block appears in your context, a studied "
            "corpus is relevant — use query_knowledge / ask_studied_corpus first; otherwise do "
            "not query corpora. Use deep_web_search (not search_web) for web gather — it tries "
            "the configured SearXNG instance first, then public fallback. Otherwise ≤4 gather "
            "tools (search/read/knowledge), then STOP with kanban-result JSON: "
            "council_seat=researcher, claims[], uncertainties[]. "
            "No list_directory. No append_research_scratch."
        ),
        "timeout_seconds": 600,
        "max_retries": 1,
    },
    "council_verifier": {
        "tools": {
            "read_file", "read_file_chunk", "list_directory", "search_filesystem",
            "get_current_directory", "get_file_metadata", "verify_file_exists",
            "query_knowledge", "ask_studied_corpus",
        },
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "Council Verifier (Themis). Apply frame-fit rubric to each parent claim_id. "
            "Follow the task body's tool guidance for general vs product frames. "
            "On general frames you may read_webpage cited URLs; on product frames spot-check "
            "repo paths with read_file_chunk only. "
            "End with kanban-result JSON: council_seat=verifier, verified_claims[], rejections[]."
        ),
        "timeout_seconds": 300,
        "max_retries": 0,
    },
    "council_contrarian": {
        "tools": set(),
        "research_mode": False,
        "inject_personality": False,
        "system_hint": (
            "Council Contrarian (Prometheus) seat — dissent only, no tools. "
            "End with kanban-result JSON including council_seat=contrarian."
        ),
        "timeout_seconds": 180,
        "max_retries": 0,
    },
}

# Keep AGENT_TYPES as a backward-compat alias for external callers.
AGENT_TYPES = SUBAGENT_PROFILES


def profile_read_file_chunk_invariant_violations() -> list[str]:
    """Profiles with ``read_file`` must also expose ``read_file_chunk`` (finite tool sets only)."""
    violations: list[str] = []
    for name, profile in SUBAGENT_PROFILES.items():
        tools = profile.get("tools")
        if tools is None:
            continue
        if "read_file" in tools and "read_file_chunk" not in tools:
            violations.append(name)
    return violations

# Max output characters per profile — controls verbosity.  None = no limit.
PROFILE_MAX_OUTPUT: dict[str, int | None] = {
    "web_research_light": 4000,
    "code_readonly": 8000,
    "extraction": 12000,
    "ops_cautious": 6000,
    "heartbeat_standard": 2500,
    "heartbeat_maintenance": 3000,
    "heartbeat_drive_action": 2500,
    "active_memory": 800,
    "btw": 2400,
    "research": None,
    "bash": None,
    "general": None,
}

_COMPRESSED_PERSONALITY_HINT = (
    "You are a warm, curious, genuine assistant. Your name and boundaries live in "
    "PERSONA.md and USER.md — follow those when present. Be concise and caring."
)


# ---------------------------------------------------------------------------
# Handoff telemetry (parent-visible digest; no paths or envelope content)
# ---------------------------------------------------------------------------

HANDOFF_BLACKBOARD_SKIP_NOT_ATTEMPTED = "not_attempted"
HANDOFF_BLACKBOARD_SKIP_NONE = "none"
HANDOFF_BLACKBOARD_SKIP_AUTO_DISABLED = "auto_disabled"
HANDOFF_BLACKBOARD_SKIP_MISSING_JOB_ID = "missing_job_id"
HANDOFF_BLACKBOARD_SKIP_RESOLVE_ERROR = "resolve_error"
HANDOFF_BLACKBOARD_SKIP_EMPTY_EXCERPT = "empty_excerpt"


@dataclass
class HandoffTelemetry:
    delegation_job_id: str = ""
    blackboard_injected_chars: int = 0
    blackboard_skip_reason: str = HANDOFF_BLACKBOARD_SKIP_NOT_ATTEMPTED
    envelope_prompt_injected: bool = False


# ---------------------------------------------------------------------------
# Structured result
# ---------------------------------------------------------------------------

@dataclass
class SubagentResult:
    """Structured return from a subagent run."""
    success: bool
    output: str
    error: str | None = None
    timed_out: bool = False
    timed_out_partial: bool = False
    elapsed_seconds: float = 0.0
    delegation_job_id: str = ""
    blackboard_injected_chars: int = 0
    blackboard_skip_reason: str = HANDOFF_BLACKBOARD_SKIP_NOT_ATTEMPTED
    envelope_prompt_injected: bool = False

    def to_string(self) -> str:
        if self.timed_out:
            partial = self.output
            secs = int(self.elapsed_seconds)
            if partial:
                return f"Subagent timed out after {secs}s. Partial result: {partial}"
            return f"Subagent timed out after {secs}s with no output."
        if not self.success:
            return f"Subagent error: {self.error or 'unknown error'}"
        return self.output if self.output.strip() else "Subagent returned no output."


# ---------------------------------------------------------------------------
# Iteration budget — thread-safe shared counter for total tool rounds across
# parent and subagents. Prevents runaway chains consuming unlimited rounds.
# ---------------------------------------------------------------------------

class IterationBudget:
    """Thread-safe budget for total tool-calling rounds shared across agent hierarchy."""

    def __init__(self, total: int = 60):
        self._remaining = total
        self._total = total
        self._lock = threading.Lock()

    def consume(self, n: int = 1) -> bool:
        """Consume n iterations. Returns True when consumption succeeds."""
        if n <= 0:
            return True
        with self._lock:
            if self._remaining < n:
                return False
            self._remaining -= n
            return True

    @property
    def remaining(self) -> int:
        with self._lock:
            return self._remaining

    @property
    def exhausted(self) -> bool:
        with self._lock:
            return self._remaining <= 0

    def __repr__(self) -> str:
        return f"IterationBudget({self.remaining}/{self._total})"


_CURRENT_ITERATION_BUDGET: contextvars.ContextVar[IterationBudget | None] = contextvars.ContextVar(
    "mugi_current_iteration_budget",
    default=None,
)
_CURRENT_DELEGATION_PARENT_THREAD_ID: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "mugi_delegation_parent_thread_id",
    default=None,
)
_CURRENT_DELEGATION_JOB_ID: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "mugi_delegation_job_id",
    default=None,
)


def get_current_iteration_budget() -> IterationBudget | None:
    return _CURRENT_ITERATION_BUDGET.get()


def set_current_iteration_budget(budget: IterationBudget | None) -> contextvars.Token:
    return _CURRENT_ITERATION_BUDGET.set(budget)


def reset_current_iteration_budget(token: contextvars.Token) -> None:
    _CURRENT_ITERATION_BUDGET.reset(token)


def get_current_delegation_parent_thread_id() -> str | None:
    return _CURRENT_DELEGATION_PARENT_THREAD_ID.get()


def get_current_delegation_job_id() -> str | None:
    return _CURRENT_DELEGATION_JOB_ID.get()


def set_current_delegation_context(
    parent_thread_id: str | None,
    delegation_job_id: str | None,
) -> tuple[contextvars.Token, contextvars.Token]:
    tok_parent = _CURRENT_DELEGATION_PARENT_THREAD_ID.set(parent_thread_id)
    tok_job = _CURRENT_DELEGATION_JOB_ID.set(delegation_job_id)
    return tok_parent, tok_job


def reset_current_delegation_context(tokens: tuple[contextvars.Token, contextvars.Token]) -> None:
    tok_parent, tok_job = tokens
    _CURRENT_DELEGATION_PARENT_THREAD_ID.reset(tok_parent)
    _CURRENT_DELEGATION_JOB_ID.reset(tok_job)


# ---------------------------------------------------------------------------
# Config helper
# ---------------------------------------------------------------------------

def _get_subagent_config() -> dict:
    def _load() -> dict:
        from core.config import get_subagent_config

        cfg = get_subagent_config()
        return cfg if isinstance(cfg, dict) else {}

    return best_effort_value(
        _load,
        log="Failed to load subagent config",
        default={},
        logger=log,
    )


# ---------------------------------------------------------------------------
# Tool filtering
# ---------------------------------------------------------------------------

def _build_tools_override(allowed_names: set[str] | None) -> list | None:
    """Return a filtered tool list, or None to use the full set."""
    if allowed_names is None:
        return None
    all_tools = get_tools(include_media=True, include_cli=False)
    return [t for t in all_tools if getattr(t, "name", None) in allowed_names]


def _augment_kanban_handoff_tools(effective_tool_names, kanban_claim):
    """Grain-reframe Stage B (flag-gated): guarantee the claim-scoped handoff tools for kanban workers.

    When the native result tool is promoted to the primary handoff channel
    (``[kanban].native_result_handoff``), every kanban worker must actually have
    ``kanban_post_summary`` available — narrowed workers already get the infra tools via
    ``KANBAN_WORKER_INFRA_TOOLS``, but a full-profile worker (``effective_tool_names`` ==
    ``profile["tools"]``) could otherwise be taught to call a tool it lacks. The handoff tools
    self-reject outside their claim scope, so this is safe.

    No-op when there is no kanban claim (non-kanban subagents) or the flag is off — preserving the
    exact pre-Stage-B tool set. Returns ``effective_tool_names`` unchanged when it is ``None``
    (full-set sentinel) so we never accidentally narrow an otherwise-unrestricted worker.
    """
    if not kanban_claim or effective_tool_names is None:
        return effective_tool_names
    try:
        from utils.kanban_context import _native_result_handoff_enabled

        if _native_result_handoff_enabled():
            return set(effective_tool_names) | {"kanban_post_summary", "kanban_post_progress"}
    except Exception:
        log.debug("Stage B handoff-tool union skipped", exc_info=True)
    return effective_tool_names


def get_profile_tools_override(agent_type: str) -> list | None:
    """Public helper for callers that need profile-filtered tool sets."""
    profile = SUBAGENT_PROFILES.get(agent_type, SUBAGENT_PROFILES["general"])
    return _build_tools_override(profile.get("tools"))


def get_profile_personality_hint(agent_type: str) -> str | None:
    """Return compact personality hint when profile requests in-character writing."""
    profile = SUBAGENT_PROFILES.get(agent_type, SUBAGENT_PROFILES["general"])
    return _COMPRESSED_PERSONALITY_HINT if profile.get("inject_personality") else None


# ---------------------------------------------------------------------------
# Concurrency guard
# ---------------------------------------------------------------------------

_semaphore: threading.Semaphore | None = None
_semaphore_limit: int | None = None
_semaphore_lock = threading.Lock()


def _configured_max_concurrent() -> int:
    cfg = _get_subagent_config()
    try:
        return max(1, int(cfg.get("max_concurrent", 3)))
    except Exception:
        return 3


def reset_subagent_semaphore_for_tests() -> None:
    """Test hook: drop cached concurrency semaphore."""
    global _semaphore, _semaphore_limit
    with _semaphore_lock:
        _semaphore = None
        _semaphore_limit = None


def _get_semaphore() -> threading.Semaphore:
    global _semaphore, _semaphore_limit
    limit = _configured_max_concurrent()
    if _semaphore is not None and _semaphore_limit == limit:
        return _semaphore
    with _semaphore_lock:
        if _semaphore is None or _semaphore_limit != limit:
            _semaphore = threading.Semaphore(limit)
            _semaphore_limit = limit
            log.debug("Subagent concurrency semaphore (re)initialized max_concurrent=%s", limit)
    return _semaphore


def _resolve_effective_timeout(
    *,
    timeout_seconds: int | None,
    profile: dict,
    cfg: dict,
    force_utility_model: bool,
) -> tuple[int, str]:
    """Return (seconds, source_label) documenting timeout precedence.

    Main LLM path: caller override → profile default → [subagent].timeout_seconds.
    Utility-model path: caller override → profile default (or 10 s fallback).
    """
    if timeout_seconds is not None:
        return int(timeout_seconds), "caller_override"
    profile_timeout = profile.get("timeout_seconds")
    if profile_timeout is not None:
        return int(profile_timeout), "profile_default"
    if force_utility_model:
        return 10, "utility_fallback"
    return int(cfg.get("timeout_seconds", 480)), "subagent_config"


# ---------------------------------------------------------------------------
# Shutdown tracking
# ---------------------------------------------------------------------------

_active_entries_lock = threading.Lock()
_active_entries: list[tuple[threading.Thread, threading.Event]] = []
_atexit_registered = False


def _register_atexit() -> None:
    global _atexit_registered
    if _atexit_registered:
        return
    _atexit_registered = True

    def _shutdown():
        with _active_entries_lock:
            entries = list(_active_entries)
        for _thread, stop_event in entries:
            stop_event.set()
        for thread, _stop in entries:
            thread.join(timeout=5)

    atexit.register(_shutdown)


def _track_thread(thread: threading.Thread, stop_event: threading.Event) -> None:
    _register_atexit()
    with _active_entries_lock:
        _active_entries.append((thread, stop_event))


def _untrack_thread(thread: threading.Thread) -> None:
    with _active_entries_lock:
        _active_entries[:] = [(t, e) for t, e in _active_entries if t is not thread]


# ---------------------------------------------------------------------------
# Core runner — structured version
# ---------------------------------------------------------------------------

def run_subagent_structured(
    task_description: str,
    agent_type: str = "research",
    context_snippet: str | None = None,
    timeout_seconds: int | None = None,
    model_override: str | None = None,
    base_url_override: str | None = None,
    api_key_override: str | None = None,
    policy_context: Any | None = None,
    iteration_budget: IterationBudget | None = None,
    parent_thread_id: str | None = None,
    delegation_job_id: str | None = None,
    handoff_envelope: HandoffEnvelope | None = None,
    *,
    force_utility_model: bool = False,
    system_override: str | None = None,
    tool_names_override: set[str] | None = None,
    kanban_claim: tuple[str, str] | None = None,
    kanban_worker_cwd: str | None = None,
    prefer_final_output: bool = False,
    skip_handoff_critic: bool = False,
    external_stop: threading.Event | None = None,
) -> SubagentResult:
    """Run an isolated, personality-free agent and return a structured result.

    The subagent has no identity, memory, or evolution context. It gets a
    task-only system prompt, a restricted tool set (based on agent_type/profile),
    and runs with subagent_mode=True in run_raw_agent.

    When ``force_utility_model=True`` the subagent short-circuits the main-LLM
    agent loop and runs as a single ``utility_model.utility_complete`` call
    with no tool use. This is the "pre-reply memory sub-agent" path used by
    Active Memory — it respects the profile's timeout, tool allowlist (as
    metadata/documentation only — the utility model does not invoke tools),
    and ``PROFILE_MAX_OUTPUT`` cap. ``system_override`` replaces the profile
    ``system_hint`` for this run.
    """
    t_start = time.monotonic()
    if policy_context is None:

        def _load_policy_context() -> Any:
            from tool_policy import get_current_policy_context

            return get_current_policy_context()

        policy_context = best_effort_value(
            _load_policy_context,
            log="Failed to read current policy context for subagent run",
            default=None,
            logger=log,
        )

    cfg = _get_subagent_config()

    def _record_subagent_ledger(res: SubagentResult, *, reason: str | None = None) -> None:
        from utils.task_ledger import record_run as _ledger_record

        if res.timed_out:
            status = "timeout"
        elif res.success:
            status = "ok"
        else:
            status = "error"
        _ledger_record(
            kind="subagent",
            task_name=f"subagent:{agent_type}",
            thread_id=parent_thread_id,
            status=status,
            skip_reason=reason if not res.success and status == "skipped" else None,
            duration_ms=int((res.elapsed_seconds or 0) * 1000),
            error_summary=(res.error or None) if not res.success else None,
            meta={
                "agent_type": str(agent_type or ""),
                "force_utility_model": bool(force_utility_model),
                "delegation_job_id": (res.delegation_job_id or delegation_job_id or "") or None,
            },
        )

    def _try_record_subagent_ledger(res: SubagentResult, *, reason: str | None = None) -> None:
        best_effort(
            lambda: _record_subagent_ledger(res, reason=reason),
            log="Failed to record subagent run in task ledger",
            logger=log,
        )

    # Respect enabled flag.
    if not cfg.get("enabled", True):
        res = SubagentResult(success=False, output="", error="Subagent execution is disabled in config.")
        _try_record_subagent_ledger(res, reason="disabled")
        return res

    # Concurrency guard.
    sem = _get_semaphore()
    if not sem.acquire(timeout=30):
        res = SubagentResult(success=False, output="", error="Too many subagents running concurrently. Try again later.")
        _try_record_subagent_ledger(res, reason="concurrency_limit")
        return res

    try:
        if force_utility_model:
            profile = SUBAGENT_PROFILES.get(agent_type, SUBAGENT_PROFILES["general"])
            effective_timeout, timeout_source = _resolve_effective_timeout(
                timeout_seconds=timeout_seconds,
                profile=profile,
                cfg=cfg,
                force_utility_model=True,
            )
            log.debug(
                "Subagent utility timeout agent_type=%s seconds=%s source=%s",
                agent_type,
                effective_timeout,
                timeout_source,
            )
            res = _run_utility_single(
                agent_type=agent_type,
                profile=profile,
                task_description=task_description,
                context_snippet=context_snippet,
                system_override=system_override,
                timeout_seconds=effective_timeout,
                t_start=t_start,
                parent_thread_id=parent_thread_id,
                delegation_job_id=delegation_job_id,
            )
            _try_record_subagent_ledger(res)
            return res

        profile = SUBAGENT_PROFILES.get(agent_type, SUBAGENT_PROFILES["general"])
        effective_timeout, timeout_source = _resolve_effective_timeout(
            timeout_seconds=timeout_seconds,
            profile=profile,
            cfg=cfg,
            force_utility_model=False,
        )
        log.debug(
            "Subagent run timeout agent_type=%s seconds=%s source=%s",
            agent_type,
            effective_timeout,
            timeout_source,
        )

        res = _run_with_retries(
            task_description,
            agent_type,
            context_snippet,
            effective_timeout,
            model_override,
            base_url_override,
            api_key_override,
            cfg,
            t_start,
            policy_context=policy_context,
            iteration_budget=iteration_budget,
            parent_thread_id=parent_thread_id,
            delegation_job_id=delegation_job_id,
            handoff_envelope=handoff_envelope,
            tool_names_override=tool_names_override,
            system_override=system_override,
            kanban_claim=kanban_claim,
            kanban_worker_cwd=kanban_worker_cwd,
            prefer_final_output=prefer_final_output,
            skip_handoff_critic=skip_handoff_critic,
            external_stop=external_stop,
        )
        _try_record_subagent_ledger(res)
        return res
    finally:
        sem.release()


def _run_utility_single(
    *,
    agent_type: str,
    profile: dict,
    task_description: str,
    context_snippet: str | None,
    system_override: str | None,
    timeout_seconds: int,
    t_start: float,
    parent_thread_id: str | None,
    delegation_job_id: str | None,
) -> SubagentResult:
    """Single-shot utility-model run for profiles that don't need the main LLM.

    No tool-calling loop, no personality, no retries. Used by Active Memory.
    Uses a worker thread so the hard timeout survives a wedged utility backend.

    Broad ``except Exception`` on import/availability returns a structured
    ``SubagentResult`` failure — intentional (DEF-031); do not narrow without
    preserving that contract.
    """
    try:
        from utility_model import is_available as _util_available, utility_complete
    except Exception as e:
        return SubagentResult(
            success=False,
            output="",
            error=f"utility_model unavailable: {e}",
            elapsed_seconds=time.monotonic() - t_start,
            delegation_job_id=(delegation_job_id or "").strip(),
        )
    try:
        if not _util_available():
            return SubagentResult(
                success=False,
                output="",
                error="utility_model not available",
                elapsed_seconds=time.monotonic() - t_start,
                delegation_job_id=(delegation_job_id or "").strip(),
            )
    except Exception:
        log.debug("utility_model availability check failed", exc_info=True)
        return SubagentResult(
            success=False,
            output="",
            error="utility_model availability check failed",
            elapsed_seconds=time.monotonic() - t_start,
            delegation_job_id=(delegation_job_id or "").strip(),
        )

    system_text = (system_override or profile.get("system_hint") or "").strip()
    user_parts: list[str] = []
    if context_snippet:
        user_parts.append(str(context_snippet))
    user_parts.append(str(task_description or ""))
    user_text = "\n\n".join(p for p in user_parts if p).strip()

    messages: list[dict[str, str]] = []
    if system_text:
        messages.append({"role": "system", "content": system_text})
    messages.append({"role": "user", "content": user_text})

    cap = PROFILE_MAX_OUTPUT.get(agent_type)
    max_tokens = 512
    if isinstance(cap, int) and cap > 0:
        # ~1 token ≈ 4 chars; keep a small floor to avoid starving short replies.
        max_tokens = max(128, min(1024, cap // 2 + 64))

    result_box: dict[str, str | None] = {"out": None}

    def _worker() -> None:
        try:
            result_box["out"] = utility_complete(
                messages,
                max_tokens=max_tokens,
                temperature=0.2,
                activity_operation=f"subagent:{agent_type}",
                activity_thread_id=parent_thread_id,
            )
        except Exception:
            log.debug("utility_complete failed in _run_utility_single", exc_info=True)
            result_box["out"] = None

    worker = threading.Thread(
        target=_worker, daemon=True, name=f"subagent-utility-{agent_type}"
    )
    worker.start()
    worker.join(timeout=max(0.5, float(timeout_seconds)))
    elapsed = time.monotonic() - t_start

    if worker.is_alive():
        return SubagentResult(
            success=False,
            output="",
            timed_out=True,
            elapsed_seconds=elapsed,
            delegation_job_id=(delegation_job_id or "").strip(),
        )

    raw = result_box["out"]
    if not raw:
        return SubagentResult(
            success=False,
            output="",
            error="utility_model returned no output",
            elapsed_seconds=elapsed,
            delegation_job_id=(delegation_job_id or "").strip(),
        )

    output = _enforce_output_limit(str(raw), agent_type)
    return SubagentResult(
        success=True,
        output=output,
        elapsed_seconds=elapsed,
        delegation_job_id=(delegation_job_id or "").strip(),
    )


def _run_with_retries(
    task_description: str,
    agent_type: str,
    context_snippet: str | None,
    timeout_seconds: int | None,
    model_override: str | None,
    base_url_override: str | None,
    api_key_override: str | None,
    cfg: dict,
    t_start: float,
    policy_context: Any | None = None,
    iteration_budget: IterationBudget | None = None,
    parent_thread_id: str | None = None,
    delegation_job_id: str | None = None,
    handoff_envelope: HandoffEnvelope | None = None,
    tool_names_override: set[str] | None = None,
    system_override: str | None = None,
    kanban_claim: tuple[str, str] | None = None,
    kanban_worker_cwd: str | None = None,
    prefer_final_output: bool = False,
    skip_handoff_critic: bool = False,
    external_stop: threading.Event | None = None,
) -> SubagentResult:
    profile = SUBAGENT_PROFILES.get(agent_type, SUBAGENT_PROFILES["general"])

    # Resolve overrides from config, then from profile defaults.
    if timeout_seconds is None:
        timeout_seconds, _ = _resolve_effective_timeout(
            timeout_seconds=None,
            profile=profile,
            cfg=cfg,
            force_utility_model=False,
        )
    if model_override is None:
        v = (cfg.get("model") or "").strip()
        model_override = v or None
    if base_url_override is None:
        v = (cfg.get("base_url") or "").strip()
        base_url_override = v or None
    if api_key_override is None:
        v = (cfg.get("api_key") or "").strip()
        api_key_override = v or None

    max_retries = profile.get("max_retries")
    if max_retries is None:
        max_retries = int(cfg.get("max_retries", 1))

    last_result: SubagentResult | None = None
    for attempt in range(max_retries + 1):
        if attempt > 0:
            time.sleep(min(2 ** attempt, 8))

        result = _single_run(
            task_description,
            agent_type,
            context_snippet,
            timeout_seconds,
            model_override,
            base_url_override,
            api_key_override,
            profile,
            t_start,
            policy_context=policy_context,
            iteration_budget=iteration_budget,
            parent_thread_id=parent_thread_id,
            delegation_job_id=delegation_job_id,
            handoff_envelope=handoff_envelope,
            tool_names_override=tool_names_override,
            system_override=system_override,
            kanban_claim=kanban_claim,
            kanban_worker_cwd=kanban_worker_cwd,
            prefer_final_output=prefer_final_output,
            skip_handoff_critic=skip_handoff_critic,
            external_stop=external_stop,
        )

        if result.success or result.timed_out:
            return result

        last_result = result
        log.info("Subagent attempt %d/%d failed: %s", attempt + 1, max_retries + 1,
                 (result.error or "")[:200])

    return last_result or SubagentResult(success=False, output="", error="All retry attempts exhausted.")


def _enforce_output_limit(output: str, agent_type: str) -> str:
    """Trim output to the profile's max length if configured."""
    limit = PROFILE_MAX_OUTPUT.get(agent_type)
    if limit is None or len(output) <= limit:
        return output
    return output[:limit] + f"\n\n[Output truncated at {limit} characters]"


def _build_subagent_user_content(
    *,
    task_description: str,
    context_snippet: str | None,
    handoff_envelope: HandoffEnvelope | None,
    delegation_job_id: str | None,
    parent_thread_id: str | None,
) -> tuple[str, str, str, str, HandoffTelemetry]:
    """Assemble first user message for subagent; returns telemetry for parent digest."""
    job = (delegation_job_id or "").strip()
    telemetry = HandoffTelemetry(
        delegation_job_id=job,
        blackboard_skip_reason=HANDOFF_BLACKBOARD_SKIP_NOT_ATTEMPTED,
    )

    scratch_path = ""
    blackboard_excerpt = ""
    envelope_summary = ""
    parts: list[str] = []

    handoff_auto, inject_chars, _post_append = get_handoff_auto_config()

    if not handoff_auto:
        telemetry.blackboard_skip_reason = HANDOFF_BLACKBOARD_SKIP_AUTO_DISABLED
    elif not job:
        telemetry.blackboard_skip_reason = HANDOFF_BLACKBOARD_SKIP_MISSING_JOB_ID
    else:
        path, safe_parent, safe_job, err = resolve_delegation_scratch_path(
            parent_thread_id or "", job, allow_write=False
        )
        if err or not path:
            telemetry.blackboard_skip_reason = HANDOFF_BLACKBOARD_SKIP_RESOLVE_ERROR
            log.debug(
                "Skipping blackboard inject (reason=%s parent=%s job=%s)",
                err,
                safe_parent,
                safe_job,
            )
        else:
            scratch_path = path
            blackboard_excerpt = read_blackboard_excerpt(path, max_chars=inject_chars).strip()
            if blackboard_excerpt:
                telemetry.blackboard_injected_chars = len(blackboard_excerpt)
                telemetry.blackboard_skip_reason = HANDOFF_BLACKBOARD_SKIP_NONE
                parts.append(
                    "### Delegation blackboard (read first)\n```text\n"
                    f"{blackboard_excerpt}\n```"
                )
            else:
                telemetry.blackboard_skip_reason = HANDOFF_BLACKBOARD_SKIP_EMPTY_EXCERPT

    envelope_block = envelope_to_prompt_block(handoff_envelope)
    telemetry.envelope_prompt_injected = bool(envelope_block)
    if envelope_block:
        envelope_summary = envelope_block
        parts.append(envelope_block)

    if context_snippet:
        parts.append(f"Context: {context_snippet}")
        parts.append(f"Task: {task_description}")
    else:
        parts.append(task_description)

    user_content = "\n\n".join(p for p in parts if p).strip()
    return user_content, blackboard_excerpt, envelope_summary, scratch_path, telemetry


def _single_run(
    task_description: str,
    agent_type: str,
    context_snippet: str | None,
    timeout_seconds: int,
    model_override: str | None,
    base_url_override: str | None,
    api_key_override: str | None,
    profile: dict,
    t_start: float,
    policy_context: Any | None = None,
    iteration_budget: IterationBudget | None = None,
    parent_thread_id: str | None = None,
    delegation_job_id: str | None = None,
    handoff_envelope: HandoffEnvelope | None = None,
    tool_names_override: set[str] | None = None,
    system_override: str | None = None,
    kanban_claim: tuple[str, str] | None = None,
    kanban_worker_cwd: str | None = None,
    prefer_final_output: bool = False,
    skip_handoff_critic: bool = False,
    external_stop: threading.Event | None = None,
) -> SubagentResult:
    effective_tool_names = tool_names_override if tool_names_override is not None else profile["tools"]
    effective_tool_names = _augment_kanban_handoff_tools(effective_tool_names, kanban_claim)
    tools_override = _build_tools_override(effective_tool_names)
    if tools_override is not None and len(tools_override) == 0:
        log.warning("tool_names_override resolved to empty tool set; falling back to profile default")
        tools_override = _build_tools_override(profile["tools"])
    personality_hint = _COMPRESSED_PERSONALITY_HINT if profile.get("inject_personality") else None

    user_content, blackboard_excerpt, envelope_summary, scratch_path, handoff_tel = _build_subagent_user_content(
        task_description=task_description,
        context_snippet=context_snippet,
        handoff_envelope=handoff_envelope,
        delegation_job_id=delegation_job_id,
        parent_thread_id=parent_thread_id,
    )
    log.debug(
        "Subagent handoff context job=%s envelope_present=%s blackboard_chars=%d",
        (delegation_job_id or "").strip() or "-",
        bool(envelope_summary),
        len(blackboard_excerpt),
    )
    messages = [{"role": "user", "content": user_content}]

    sub_thread_id = f"subagent_{uuid4().hex[:12]}"
    stop_event = threading.Event()

    collector: list[str] = []
    final_output_ref: list[str] = []
    error_ref: list[str] = []

    def _on_chunk(text):
        if text:
            collector.append(text)

    def _on_finished(text):
        if text:
            final_output_ref.clear()
            final_output_ref.append(text)
            if not prefer_final_output and not collector:
                collector.append(text)

    def _on_error(err):
        error_ref.append(str(err))

    def _run():
        from agent_loop import run_raw_agent

        _ctx_tokens = set_current_delegation_context(parent_thread_id, delegation_job_id)
        claim_token_handle = None
        cwd_token_handle = None
        search_brief_handle = None
        if kanban_claim:
            try:
                from utils.kanban_dispatcher import _reset_claim_token, _set_claim_token

                task_id, token = kanban_claim
                claim_token_handle = _set_claim_token(task_id, token)
            except Exception:
                log.debug("Failed to bind kanban claim in subagent thread", exc_info=True)
        if kanban_claim:
            try:
                from utils.kanban_dispatcher import _set_kanban_worker_search_brief

                task_line = (task_description or "").strip()
                if task_line:
                    search_brief_handle = _set_kanban_worker_search_brief(task_line)
            except Exception:
                log.debug("Failed to bind kanban worker search brief", exc_info=True)
        live_view_handle = None
        if kanban_claim:
            try:
                from tools._logging import set_live_view_callback
                from utils.kanban_worker_progress import record_live_preview

                _lv_task_id = kanban_claim[0]

                def _on_live_view(evt):
                    record_live_preview(_lv_task_id, evt)

                set_live_view_callback(_on_live_view)
                live_view_handle = True
            except Exception:
                log.debug("Failed to bind kanban live-view callback", exc_info=True)
        if kanban_worker_cwd:
            try:
                from utils.kanban_dispatcher import _reset_kanban_worker_cwd, _set_kanban_worker_cwd

                cwd_token_handle = _set_kanban_worker_cwd(kanban_worker_cwd)
            except Exception:
                log.debug("Failed to bind kanban worker cwd in subagent thread", exc_info=True)
        try:
            run_raw_agent(
                messages=messages,
                thread_id=sub_thread_id,
                config=None,
                on_chunk=_on_chunk,
                on_finished=_on_finished,
                on_error=_on_error,
                stop_check=lambda: stop_event.is_set() or bool(external_stop and external_stop.is_set()),
                research_mode=profile["research_mode"],
                subagent_mode=True,
                model_override=model_override,
                base_url_override=base_url_override,
                api_key_override=api_key_override,
                tools_override=tools_override,
                subagent_personality_hint=personality_hint,
                subagent_system_override=system_override,
                kanban_worker_mode=bool(kanban_claim),
                policy_context=policy_context,
                iteration_budget=iteration_budget,
            )
        except Exception:
            # Intentional: surface full traceback to parent via error_ref (DEF-031).
            error_ref.append(traceback.format_exc())
        finally:
            if live_view_handle is not None:
                try:
                    from tools._logging import set_live_view_callback

                    set_live_view_callback(None)
                except Exception:
                    log.debug("Failed to reset kanban live-view callback", exc_info=True)
            if kanban_claim:
                try:
                    from utils.kanban_worker_progress import flush_pending_sse

                    flush_pending_sse(kanban_claim[0])
                except Exception:
                    log.debug("Failed to flush kanban progress SSE", exc_info=True)
            if claim_token_handle is not None:
                best_effort(
                    lambda: _reset_claim_token(claim_token_handle),
                    log="Failed to reset kanban claim in subagent thread",
                    logger=log,
                )
            if search_brief_handle is not None:
                best_effort(
                    lambda: __import__(
                        "utils.kanban_dispatcher",
                        fromlist=["_reset_kanban_worker_search_brief"],
                    )._reset_kanban_worker_search_brief(search_brief_handle),
                    log="Failed to reset kanban worker search brief",
                    logger=log,
                )
            if cwd_token_handle is not None:
                best_effort(
                    lambda: _reset_kanban_worker_cwd(cwd_token_handle),
                    log="Failed to reset kanban worker cwd in subagent thread",
                    logger=log,
                )
            best_effort(
                lambda: reset_current_delegation_context(_ctx_tokens),
                log="Failed to reset delegation context",
                logger=log,
            )

    ctx = contextvars.copy_context()
    thread = threading.Thread(
        target=lambda: ctx.run(_run),
        daemon=True,
        name=f"subagent-{agent_type}",
    )
    _track_thread(thread, stop_event)
    timed_out = False

    def _resolve_output_text(*, partial_only: bool = False) -> str:
        joined = "".join(collector)
        if prefer_final_output:
            if partial_only:
                if joined:
                    return joined
                if final_output_ref:
                    return final_output_ref[-1]
                return ""
            if final_output_ref:
                return final_output_ref[-1]
            return joined
        if joined:
            return joined
        if final_output_ref:
            return final_output_ref[-1]
        return ""

    def _maybe_apply_handoff_critic(output: str) -> tuple[str, str | None]:
        if skip_handoff_critic:
            return output, None
        return maybe_critique_handoff(
            kind="delegation",
            task_text=task_description,
            context_text=context_snippet or "",
            output_text=output,
            blackboard_excerpt=blackboard_excerpt,
            envelope_summary=envelope_summary,
        )

    try:
        thread.start()
        thread.join(timeout=timeout_seconds)

        elapsed = time.monotonic() - t_start

        if thread.is_alive():
            timed_out = True
            stop_event.set()
            thread.join(timeout=5)
            partial = _resolve_output_text(partial_only=True)
            if not kanban_claim:
                output = _enforce_output_limit(partial, agent_type)
            else:
                output = partial
            output, _warning = _maybe_apply_handoff_critic(output)
            log.debug(
                "Subagent handoff critic job=%s state=%s",
                (delegation_job_id or "").strip() or "-",
                "fail" if _warning else "pass_or_skipped",
            )
            return SubagentResult(
                success=False,
                output=output,
                timed_out=True,
                timed_out_partial=bool(str(output or "").strip()),
                elapsed_seconds=elapsed,
                delegation_job_id=handoff_tel.delegation_job_id,
                blackboard_injected_chars=handoff_tel.blackboard_injected_chars,
                blackboard_skip_reason=handoff_tel.blackboard_skip_reason,
                envelope_prompt_injected=handoff_tel.envelope_prompt_injected,
            )

        if error_ref and not collector and not final_output_ref:
            return SubagentResult(
                success=False,
                output="",
                error=error_ref[0],
                elapsed_seconds=elapsed,
                delegation_job_id=handoff_tel.delegation_job_id,
                blackboard_injected_chars=handoff_tel.blackboard_injected_chars,
                blackboard_skip_reason=handoff_tel.blackboard_skip_reason,
                envelope_prompt_injected=handoff_tel.envelope_prompt_injected,
            )

        output = _resolve_output_text()
        if not kanban_claim:
            output = _enforce_output_limit(output, agent_type)
        output, _warning = _maybe_apply_handoff_critic(output)
        log.debug(
            "Subagent handoff critic job=%s state=%s",
            (delegation_job_id or "").strip() or "-",
            "fail" if _warning else "pass_or_skipped",
        )
        return SubagentResult(
            success=True,
            output=output,
            elapsed_seconds=elapsed,
            delegation_job_id=handoff_tel.delegation_job_id,
            blackboard_injected_chars=handoff_tel.blackboard_injected_chars,
            blackboard_skip_reason=handoff_tel.blackboard_skip_reason,
            envelope_prompt_injected=handoff_tel.envelope_prompt_injected,
        )
    finally:
        handoff_auto, _inject_chars, post_append = get_handoff_auto_config()
        if handoff_auto and post_append and scratch_path and (delegation_job_id or "").strip():
            elapsed_total = time.monotonic() - t_start
            status = "success"
            if timed_out:
                status = "timed_out"
            elif error_ref and not collector:
                status = "error"
            path, safe_parent, safe_job, err = resolve_delegation_scratch_path(
                parent_thread_id or "", delegation_job_id or "", allow_write=True
            )
            if not err and path:
                append_system_footer(
                    abs_path=path,
                    safe_parent=safe_parent,
                    safe_job=safe_job,
                    status=status,
                    elapsed_seconds=elapsed_total,
                    output_chars=len("".join(collector)),
                )
            else:
                log.debug("Skipped handoff footer append: %s", err)
        _untrack_thread(thread)


# ---------------------------------------------------------------------------
# Public string API (backward-compat for delegate_task and external callers)
# ---------------------------------------------------------------------------

def run_subagent(
    task_description: str,
    agent_type: str = "research",
    context_snippet: str | None = None,
    timeout_seconds: int | None = None,
    model_override: str | None = None,
    base_url_override: str | None = None,
    api_key_override: str | None = None,
    policy_context: Any | None = None,
    iteration_budget: IterationBudget | None = None,
    parent_thread_id: str | None = None,
    delegation_job_id: str | None = None,
    handoff_envelope: HandoffEnvelope | None = None,
    *,
    force_utility_model: bool = False,
    system_override: str | None = None,
    tool_names_override: set[str] | None = None,
) -> str:
    """Run an isolated, personality-free agent and return its reply as a string.

    Wraps run_subagent_structured() for backward compatibility. See that
    function for ``force_utility_model`` / ``system_override`` /
    ``tool_names_override`` semantics.
    """
    result = run_subagent_structured(
        task_description=task_description,
        agent_type=agent_type,
        context_snippet=context_snippet,
        timeout_seconds=timeout_seconds,
        model_override=model_override,
        base_url_override=base_url_override,
        api_key_override=api_key_override,
        policy_context=policy_context,
        iteration_budget=iteration_budget,
        parent_thread_id=parent_thread_id,
        delegation_job_id=delegation_job_id,
        handoff_envelope=handoff_envelope,
        force_utility_model=force_utility_model,
        system_override=system_override,
        tool_names_override=tool_names_override,
    )
    return result.to_string()
