agent_runtime.py

from __future__ import annotations

from dataclasses import dataclass, field, replace
from datetime import datetime, timezone
import json
from pathlib import Path
from typing import Any
from uuid import uuid4

from .account_runtime import AccountRuntime
from .agent_manager import AgentManager
from .agent_context import clear_context_caches
from .agent_context import render_context_report as render_agent_context_report
from .agent_context_usage import collect_context_usage, estimate_tokens, format_context_usage
from .compact import compact_conversation
from .ask_user_runtime import AskUserRuntime
from .agent_registry import (
    delete_agent_definition,
    find_agent_definition,
    normalize_mutable_source,
    load_agent_registry,
    render_agent_mutation,
    render_agent_detail,
    render_agents_report,
    scaffold_agent_definition,
    update_agent_definition,
)
from .config_runtime import ConfigRuntime
from .hook_policy import HookPolicyRuntime
from .lsp_runtime import LSPRuntime
from .mcp_runtime import MCPRuntime
from .agent_prompting import (
    build_prompt_context,
    build_system_prompt_parts,
    render_system_prompt,
)
from .agent_session import AgentSessionState
from .agent_slash_commands import preprocess_slash_command
from .agent_tools import (
    AgentTool,
    build_tool_context,
    default_tool_registry,
    execute_tool_streaming,
    serialize_tool_result,
)
from .agent_types import (
    AgentRunResult,
    AgentPermissions,
    AgentRuntimeConfig,
    AssistantTurn,
    BudgetConfig,
    ModelConfig,
    OutputSchemaConfig,
    StreamEvent,
    ToolCall,
    ToolExecutionResult,
    UsageStats,
)
from .openai_compat import OpenAICompatClient, OpenAICompatError
from .plan_runtime import PlanRuntime
from .plugin_runtime import PluginRuntime
from .remote_runtime import RemoteRuntime
from .remote_trigger_runtime import RemoteTriggerRuntime
from .search_runtime import SearchRuntime
from .task_runtime import TaskRuntime
from .team_runtime import TeamRuntime
from .tokenizer_runtime import describe_token_counter
from .workflow_runtime import WorkflowRuntime
from .worktree_runtime import WorktreeRuntime
from .session_env_vars import clear_session_env_vars
from .session_store import (
    StoredAgentSession,
    load_agent_session,
    save_agent_session,
    serialize_model_config,
    serialize_runtime_config,
    usage_from_payload,
)
from .token_budget import calculate_token_budget, format_token_budget
from .builtin_agents import (
    AgentDefinition,
    ALL_AGENT_DISALLOWED_TOOLS,
    GENERAL_PURPOSE_AGENT,
)
from .microcompact import microcompact_messages as _microcompact_messages


@dataclass(frozen=True)
class BudgetDecision:
    exceeded: bool
    reason: str | None = None


@dataclass(frozen=True)
class PromptPreflightResult:
    usage_increment: UsageStats = field(default_factory=UsageStats)
    model_calls_increment: int = 0
    stop_reason: str | None = None
    reason: str | None = None


@dataclass
class LocalCodingAgent:
    model_config: ModelConfig
    runtime_config: AgentRuntimeConfig
    custom_system_prompt: str | None = None
    append_system_prompt: str | None = None
    override_system_prompt: str | None = None
    tool_registry: dict[str, AgentTool] | None = None
    agent_manager: AgentManager | None = None
    parent_agent_id: str | None = None
    managed_group_id: str | None = None
    managed_child_index: int | None = None
    managed_label: str | None = None
    plugin_runtime: PluginRuntime | None = None
    hook_policy_runtime: HookPolicyRuntime | None = None
    mcp_runtime: MCPRuntime | None = None
    remote_runtime: RemoteRuntime | None = None
    remote_trigger_runtime: RemoteTriggerRuntime | None = None
    search_runtime: SearchRuntime | None = None
    account_runtime: AccountRuntime | None = None
    ask_user_runtime: AskUserRuntime | None = None
    config_runtime: ConfigRuntime | None = None
    lsp_runtime: LSPRuntime | None = None
    plan_runtime: PlanRuntime | None = None
    task_runtime: TaskRuntime | None = None
    team_runtime: TeamRuntime | None = None
    workflow_runtime: WorkflowRuntime | None = None
    worktree_runtime: WorktreeRuntime | None = None
    last_session: AgentSessionState | None = field(default=None, init=False, repr=False)
    last_run_result: AgentRunResult | None = field(default=None, init=False, repr=False)
    cumulative_usage: UsageStats = field(default_factory=UsageStats, init=False, repr=False)
    cumulative_cost_usd: float = field(default=0.0, init=False, repr=False)
    _compact_consecutive_failures: int = field(default=0, init=False, repr=False)
    active_session_id: str | None = field(default=None, init=False, repr=False)
    last_session_path: str | None = field(default=None, init=False, repr=False)
    managed_agent_id: str | None = field(default=None, init=False, repr=False)
    resume_source_session_id: str | None = field(default=None, init=False, repr=False)

    def __post_init__(self) -> None:
        if self.tool_registry is None:
            self.tool_registry = default_tool_registry()
        if self.agent_manager is None:
            self.agent_manager = AgentManager()
        if self.plugin_runtime is None:
            self.plugin_runtime = PluginRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.hook_policy_runtime is None:
            self.hook_policy_runtime = HookPolicyRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.mcp_runtime is None:
            self.mcp_runtime = MCPRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.remote_runtime is None:
            self.remote_runtime = RemoteRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.remote_trigger_runtime is None:
            self.remote_trigger_runtime = RemoteTriggerRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.search_runtime is None:
            self.search_runtime = SearchRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.account_runtime is None:
            self.account_runtime = AccountRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.ask_user_runtime is None:
            self.ask_user_runtime = AskUserRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.config_runtime is None:
            self.config_runtime = ConfigRuntime.from_workspace(self.runtime_config.cwd)
        if self.lsp_runtime is None:
            self.lsp_runtime = LSPRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.plan_runtime is None:
            self.plan_runtime = PlanRuntime.from_workspace(self.runtime_config.cwd)
        if self.task_runtime is None:
            self.task_runtime = TaskRuntime.from_workspace(self.runtime_config.cwd)
        if self.team_runtime is None:
            self.team_runtime = TeamRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.workflow_runtime is None:
            self.workflow_runtime = WorkflowRuntime.from_workspace(
                self.runtime_config.cwd,
                tuple(str(path) for path in self.runtime_config.additional_working_directories),
            )
        if self.worktree_runtime is None:
            self.worktree_runtime = WorktreeRuntime.from_workspace(self.runtime_config.cwd)
        self.runtime_config = self._apply_hook_policy_budget_overrides(self.runtime_config)
        registry = dict(self.tool_registry)
        plugin_tools = self.plugin_runtime.register_tool_aliases(registry)
        if plugin_tools:
            registry = {**registry, **plugin_tools}
        virtual_tools = self.plugin_runtime.register_virtual_tools(registry)
        if virtual_tools:
            registry = {**registry, **virtual_tools}
        self.tool_registry = registry
        self.client = OpenAICompatClient(self.model_config)
        self.tool_context = build_tool_context(
            self.runtime_config,
            tool_registry=self.tool_registry,
            extra_env=(
                self.hook_policy_runtime.safe_env()
                if self.hook_policy_runtime is not None
                else None
            ),
            search_runtime=self.search_runtime,
            account_runtime=self.account_runtime,
            ask_user_runtime=self.ask_user_runtime,
            config_runtime=self.config_runtime,
            lsp_runtime=self.lsp_runtime,
            mcp_runtime=self.mcp_runtime,
            remote_runtime=self.remote_runtime,
            remote_trigger_runtime=self.remote_trigger_runtime,
            plan_runtime=self.plan_runtime,
            task_runtime=self.task_runtime,
            team_runtime=self.team_runtime,
            workflow_runtime=self.workflow_runtime,
            worktree_runtime=self.worktree_runtime,
        )

    def set_model(self, model: str) -> None:
        self.model_config = replace(self.model_config, model=model)
        self.client = OpenAICompatClient(self.model_config)

    def clear_runtime_state(self) -> None:
        self.last_session = None
        self.last_run_result = None
        self.active_session_id = None
        self.last_session_path = None
        self.resume_source_session_id = None
        if self.plugin_runtime is not None:
            self.plugin_runtime.restore_session_state({})
        # Mirror commands/clear/caches.ts: drop session-scoped env vars on /clear.
        clear_session_env_vars()

    def build_prompt_context(self, scratchpad_directory: Path | None = None):
        return build_prompt_context(
            self.runtime_config,
            self.model_config,
            scratchpad_directory=scratchpad_directory,
        )

    def build_system_prompt_parts(self, prompt_context=None) -> list[str]:
        if prompt_context is None:
            prompt_context = self.build_prompt_context()
        return build_system_prompt_parts(
            prompt_context=prompt_context,
            runtime_config=self.runtime_config,
            tools=self.tool_registry,
            available_agents=self.available_agents(),
            custom_system_prompt=self.custom_system_prompt,
            append_system_prompt=self.append_system_prompt,
            override_system_prompt=self.override_system_prompt,
        )

    def load_agent_registry(self):
        return load_agent_registry(self.runtime_config.cwd)

    def available_agents(self) -> tuple[AgentDefinition, ...]:
        return self.load_agent_registry().active_agents

    def build_session(
        self,
        user_prompt: str | None = None,
        *,
        scratchpad_directory: Path | None = None,
    ) -> AgentSessionState:
        prompt_context = self.build_prompt_context(scratchpad_directory)
        system_prompt_parts = self.build_system_prompt_parts(prompt_context)
        return AgentSessionState.create(
            system_prompt_parts,
            user_prompt,
            user_context=prompt_context.user_context,
            system_context=prompt_context.system_context,
        )

    def _apply_hook_policy_budget_overrides(
        self,
        runtime_config: AgentRuntimeConfig,
    ) -> AgentRuntimeConfig:
        if self.hook_policy_runtime is None or not self.hook_policy_runtime.manifests:
            return runtime_config
        overrides = self.hook_policy_runtime.budget_overrides()
        if not overrides:
            return runtime_config
        budget = runtime_config.budget_config
        return replace(
            runtime_config,
            budget_config=BudgetConfig(
                max_total_tokens=(
                    budget.max_total_tokens
                    if budget.max_total_tokens is not None
                    else _optional_policy_int(overrides.get('max_total_tokens'))
                ),
                max_input_tokens=(
                    budget.max_input_tokens
                    if budget.max_input_tokens is not None
                    else _optional_policy_int(overrides.get('max_input_tokens'))
                ),
                max_output_tokens=(
                    budget.max_output_tokens
                    if budget.max_output_tokens is not None
                    else _optional_policy_int(overrides.get('max_output_tokens'))
                ),
                max_reasoning_tokens=(
                    budget.max_reasoning_tokens
                    if budget.max_reasoning_tokens is not None
                    else _optional_policy_int(overrides.get('max_reasoning_tokens'))
                ),
                max_total_cost_usd=(
                    budget.max_total_cost_usd
                    if budget.max_total_cost_usd is not None
                    else _optional_policy_float(overrides.get('max_total_cost_usd'))
                ),
                max_tool_calls=(
                    budget.max_tool_calls
                    if budget.max_tool_calls is not None
                    else _optional_policy_int(overrides.get('max_tool_calls'))
                ),
                max_delegated_tasks=(
                    budget.max_delegated_tasks
                    if budget.max_delegated_tasks is not None
                    else _optional_policy_int(overrides.get('max_delegated_tasks'))
                ),
                max_model_calls=(
                    budget.max_model_calls
                    if budget.max_model_calls is not None
                    else _optional_policy_int(overrides.get('max_model_calls'))
                ),
                max_session_turns=(
                    budget.max_session_turns
                    if budget.max_session_turns is not None
                    else _optional_policy_int(overrides.get('max_session_turns'))
                ),
            ),
        )

    def run(self, prompt: str) -> AgentRunResult:
        self.managed_agent_id = None
        self.resume_source_session_id = None
        if self.plugin_runtime is not None:
            self.plugin_runtime.restore_session_state({})
        session_id = uuid4().hex
        scratchpad_directory = self._ensure_scratchpad_directory(session_id)
        result = self._run_prompt(
            prompt,
            base_session=None,
            session_id=session_id,
            scratchpad_directory=scratchpad_directory,
            existing_file_history=(),
        )
        self._accumulate_usage(result)
        self._finalize_managed_agent(result)
        return result

    def resume(self, prompt: str, stored_session: StoredAgentSession) -> AgentRunResult:
        self.managed_agent_id = None
        self.resume_source_session_id = stored_session.session_id
        session = AgentSessionState.from_persisted(
            system_prompt_parts=stored_session.system_prompt_parts,
            user_context=stored_session.user_context,
            system_context=stored_session.system_context,
            messages=stored_session.messages,
        )
        self._append_file_history_replay_if_needed(
            session,
            stored_session.file_history,
        )
        self._append_compaction_replay_if_needed(session)
        self.active_session_id = stored_session.session_id
        self.last_session = session
        self.last_session_path = str(
            self.runtime_config.session_directory / f'{stored_session.session_id}.json'
        )
        if self.plugin_runtime is not None:
            self.plugin_runtime.restore_session_state(stored_session.plugin_state)
        scratchpad_directory = (
            Path(stored_session.scratchpad_directory)
            if stored_session.scratchpad_directory
            else self._ensure_scratchpad_directory(stored_session.session_id)
        )
        result = self._run_prompt(
            prompt,
            base_session=session,
            session_id=stored_session.session_id,
            scratchpad_directory=scratchpad_directory,
            existing_file_history=stored_session.file_history,
        )
        self._accumulate_usage(result)
        self._finalize_managed_agent(result)
        return result

    def _run_prompt(
        self,
        prompt: str,
        *,
        base_session: AgentSessionState | None,
        session_id: str,
        scratchpad_directory: Path | None,
        existing_file_history: tuple[dict[str, object], ...],
    ) -> AgentRunResult:
        slash_result = preprocess_slash_command(self, prompt)
        if slash_result.handled and not slash_result.should_query:
            return AgentRunResult(
                final_output=slash_result.output,
                turns=0,
                tool_calls=0,
                transcript=slash_result.transcript,
                session_id=self.active_session_id,
                session_path=self.last_session_path,
                scratchpad_directory=(
                    str(scratchpad_directory) if scratchpad_directory is not None else None
                ),
            )

        effective_prompt = self._apply_hook_policy_before_prompt_hooks(
            slash_result.prompt or prompt
        )
        effective_prompt = self._apply_plugin_before_prompt_hooks(effective_prompt)
        effective_prompt = self._apply_plugin_resume_hooks(
            effective_prompt,
            resumed=base_session is not None,
        )
        self.managed_agent_id = self.agent_manager.start_agent(
            prompt=effective_prompt,
            parent_agent_id=self.parent_agent_id,
            group_id=self.managed_group_id,
            child_index=self.managed_child_index,
            label=self.managed_label or ('root' if base_session is None else 'resume'),
            resumed_from_session_id=self.resume_source_session_id,
        )
        session = (
            base_session
            if base_session is not None
            else self.build_session(
                None,
                scratchpad_directory=scratchpad_directory,
            )
        )
        session.append_user(effective_prompt)
        self.last_session = session
        self.active_session_id = session_id
        tool_specs = [tool.to_openai_tool() for tool in self.tool_registry.values()]
        starting_usage = UsageStats()
        starting_cost_usd = 0.0
        starting_tool_calls = 0
        starting_session_turns = 0
        starting_model_calls = 0
        if base_session is not None and self.resume_source_session_id:
            try:
                stored_resume_state = load_agent_session(
                    self.resume_source_session_id,
                    directory=self.runtime_config.session_directory,
                )
            except OSError:
                stored_resume_state = None
            if stored_resume_state is not None:
                starting_usage = usage_from_payload(stored_resume_state.usage)
                starting_cost_usd = stored_resume_state.total_cost_usd
                starting_tool_calls = stored_resume_state.tool_calls
                starting_session_turns = stored_resume_state.turns
                budget_state = (
                    stored_resume_state.budget_state
                    if isinstance(stored_resume_state.budget_state, dict)
                    else {}
                )
                starting_model_calls = int(budget_state.get('model_calls', 0)) if isinstance(budget_state.get('model_calls', 0), int) else 0
        tool_calls = starting_tool_calls
        last_content = ''
        total_usage = starting_usage
        total_cost_usd = starting_cost_usd
        file_history = list(existing_file_history)
        stream_events: list[dict[str, object]] = []
        assistant_response_segments: list[str] = []
        completed_tool_call_ids: set[str] = set()
        delegated_tasks = sum(
            1 for entry in file_history if entry.get('action') in ('delegate_agent', 'Agent')
        )
        model_calls = starting_model_calls

        initial_budget = self._check_budget(
            total_usage,
            total_cost_usd,
            tool_calls=tool_calls,
            delegated_tasks=delegated_tasks,
            model_calls=model_calls,
            session_turns=starting_session_turns,
        )
        if initial_budget.exceeded:
            result = AgentRunResult(
                final_output=initial_budget.reason or 'Stopped before the first model call.',
                turns=0,
                tool_calls=0,
                transcript=session.transcript(),
                session_id=session_id,
                usage=total_usage,
                total_cost_usd=total_cost_usd,
                stop_reason='budget_exceeded',
                file_history=tuple(file_history),
                scratchpad_directory=(
                    str(scratchpad_directory) if scratchpad_directory is not None else None
                ),
            )
            result = self._persist_session(session, result)
            self.last_run_result = result
            return result

        for turn_index in range(1, self.runtime_config.max_turns + 1):
            self._microcompact_session_if_needed(
                session,
                stream_events,
                turn_index=turn_index,
            )
            self._snip_session_if_needed(
                session,
                stream_events,
                turn_index=turn_index,
            )
            self._compact_session_if_needed(
                session,
                stream_events,
                turn_index=turn_index,
            )
            preflight = self._preflight_prompt_length(
                session,
                stream_events,
                turn_index=turn_index,
            )
            if preflight.usage_increment.total_tokens or preflight.model_calls_increment:
                total_usage = total_usage + preflight.usage_increment
                total_cost_usd = self.model_config.pricing.estimate_cost_usd(total_usage)
                model_calls += preflight.model_calls_increment
                budget_after_preflight = self._check_budget(
                    total_usage,
                    total_cost_usd,
                    tool_calls=tool_calls,
                    delegated_tasks=delegated_tasks,
                    model_calls=model_calls,
                    session_turns=starting_session_turns + turn_index,
                )
                if budget_after_preflight.exceeded:
                    result = AgentRunResult(
                        final_output=(
                            budget_after_preflight.reason
                            or 'Stopped because the runtime budget was exceeded.'
                        ),
                        turns=turn_index,
                        tool_calls=tool_calls,
                        transcript=session.transcript(),
                        events=tuple(stream_events),
                        usage=total_usage,
                        total_cost_usd=total_cost_usd,
                        stop_reason='budget_exceeded',
                        file_history=tuple(file_history),
                        session_id=session_id,
                        scratchpad_directory=(
                            str(scratchpad_directory) if scratchpad_directory is not None else None
                        ),
                    )
                    result = self._persist_session(session, result)
                    self.last_run_result = result
                    return result
            if preflight.stop_reason is not None:
                result = AgentRunResult(
                    final_output=preflight.reason or 'Stopped before the next model call.',
                    turns=max(turn_index - 1, 0),
                    tool_calls=tool_calls,
                    transcript=session.transcript(),
                    events=tuple(stream_events),
                    usage=total_usage,
                    total_cost_usd=total_cost_usd,
                    stop_reason=preflight.stop_reason,
                    file_history=tuple(file_history),
                    session_id=session_id,
                    scratchpad_directory=(
                        str(scratchpad_directory) if scratchpad_directory is not None else None
                    ),
                )
                result = self._append_runtime_after_turn_events(
                    result,
                    prompt=effective_prompt,
                    turn_index=max(turn_index - 1, 0),
                )
                result = self._persist_session(session, result)
                self.last_run_result = result
                return result
            try:
                turn, turn_events = self._query_model(session, tool_specs)
            except OpenAICompatError as exc:
                if self._is_prompt_too_long_error(exc) and self._reactive_compact_session(
                    session,
                    stream_events,
                    turn_index=turn_index,
                ):
                    try:
                        turn, turn_events = self._query_model(session, tool_specs)
                    except OpenAICompatError as retry_exc:
                        exc = retry_exc
                    else:
                        stream_events.extend(
                            {
                                'type': 'reactive_compact_retry',
                                'turn_index': turn_index,
                            }
                            for _ in [0]
                        )
                        stream_events.extend(event.to_dict() for event in turn_events)
                        model_calls += 1
                        total_usage = total_usage + turn.usage
                        total_cost_usd = self.model_config.pricing.estimate_cost_usd(total_usage)
                        last_content = turn.content

                        budget_after_model = self._check_budget(
                            total_usage,
                            total_cost_usd,
                            tool_calls=tool_calls,
                            delegated_tasks=delegated_tasks,
                            model_calls=model_calls,
                            session_turns=starting_session_turns + turn_index,
                        )
                        if budget_after_model.exceeded:
                            result = AgentRunResult(
                                final_output=(
                                    budget_after_model.reason
                                    or 'Stopped because the runtime budget was exceeded.'
                                ),
                                turns=turn_index,
                                tool_calls=tool_calls,
                                transcript=session.transcript(),
                                events=tuple(stream_events),
                                usage=total_usage,
                                total_cost_usd=total_cost_usd,
                                stop_reason='budget_exceeded',
                                file_history=tuple(file_history),
                                session_id=session_id,
                                scratchpad_directory=(
                                    str(scratchpad_directory) if scratchpad_directory is not None else None
                                ),
                            )
                            result = self._persist_session(session, result)
                            self.last_run_result = result
                            return result

                        if not turn.tool_calls:
                            assistant_response_segments.append(turn.content)
                            if self._should_continue_response(turn):
                                session.append_user(
                                    self._build_continuation_prompt(),
                                    metadata={
                                        'kind': 'continuation_request',
                                        'continuation_index': len(assistant_response_segments),
                                    },
                                    message_id=f'continuation_{turn_index}',
                                )
                                stream_events.append(
                                    {
                                        'type': 'continuation_request',
                                        'reason': turn.finish_reason,
                                        'continuation_index': len(assistant_response_segments),
                                    }
                                )
                                last_content = ''.join(assistant_response_segments)
                                continue
                            result = AgentRunResult(
                                final_output=''.join(assistant_response_segments),
                                turns=turn_index,
                                tool_calls=tool_calls,
                                transcript=session.transcript(),
                                events=tuple(stream_events),
                                usage=total_usage,
                                total_cost_usd=total_cost_usd,
                                stop_reason=turn.finish_reason,
                                file_history=tuple(file_history),
                                session_id=session_id,
                                scratchpad_directory=(
                                    str(scratchpad_directory) if scratchpad_directory is not None else None
                                ),
                            )
                            result = self._persist_session(session, result)
                            self.last_run_result = result
                            return result
                        # fall through to the normal tool-call branch below
                # normal error path if not recovered
                result = AgentRunResult(
                    final_output=str(exc),
                    turns=max(turn_index - 1, 0),
                    tool_calls=tool_calls,
                    transcript=session.transcript(),
                    events=tuple(stream_events),
                    usage=total_usage,
                    total_cost_usd=total_cost_usd,
                    stop_reason='backend_error',
                    file_history=tuple(file_history),
                    session_id=session_id,
                    scratchpad_directory=(
                        str(scratchpad_directory) if scratchpad_directory is not None else None
                    ),
                )
                result = self._append_runtime_after_turn_events(
                    result,
                    prompt=effective_prompt,
                    turn_index=turn_index,
                )
                result = self._persist_session(session, result)
                self.last_run_result = result
                return result

            stream_events.extend(event.to_dict() for event in turn_events)
            model_calls += 1
            total_usage = total_usage + turn.usage
            total_cost_usd = self.model_config.pricing.estimate_cost_usd(total_usage)
            last_content = turn.content

            budget_after_model = self._check_budget(
                total_usage,
                total_cost_usd,
                tool_calls=tool_calls,
                delegated_tasks=delegated_tasks,
                model_calls=model_calls,
                session_turns=starting_session_turns + turn_index,
            )
            if budget_after_model.exceeded:
                result = AgentRunResult(
                    final_output=(
                        budget_after_model.reason
                        or 'Stopped because the runtime budget was exceeded.'
                    ),
                    turns=turn_index,
                    tool_calls=tool_calls,
                    transcript=session.transcript(),
                    events=tuple(stream_events),
                    usage=total_usage,
                    total_cost_usd=total_cost_usd,
                    stop_reason='budget_exceeded',
                    file_history=tuple(file_history),
                    session_id=session_id,
                    scratchpad_directory=(
                        str(scratchpad_directory) if scratchpad_directory is not None else None
                    ),
                )
                result = self._persist_session(session, result)
                self.last_run_result = result
                return result

            if not turn.tool_calls:
                assistant_response_segments.append(turn.content)
                if self._should_continue_response(turn):
                    session.append_user(
                        self._build_continuation_prompt(),
                        metadata={
                            'kind': 'continuation_request',
                            'continuation_index': len(assistant_response_segments),
                        },
                        message_id=f'continuation_{turn_index}',
                    )
                    stream_events.append(
                        {
                            'type': 'continuation_request',
                            'reason': turn.finish_reason,
                            'continuation_index': len(assistant_response_segments),
                        }
                    )
                    last_content = ''.join(assistant_response_segments)
                    continue
                result = AgentRunResult(
                    final_output=''.join(assistant_response_segments),
                    turns=turn_index,
                    tool_calls=tool_calls,
                    transcript=session.transcript(),
                    events=tuple(stream_events),
                    usage=total_usage,
                    total_cost_usd=total_cost_usd,
                    stop_reason=turn.finish_reason,
                    file_history=tuple(file_history),
                    session_id=session_id,
                    scratchpad_directory=(
                        str(scratchpad_directory) if scratchpad_directory is not None else None
                    ),
                )
                result = self._append_runtime_after_turn_events(
                    result,
                    prompt=effective_prompt,
                    turn_index=turn_index,
                )
                result = self._persist_session(session, result)
                self.last_run_result = result
                return result

            for tool_call in turn.tool_calls:
                assistant_response_segments.clear()
                if tool_call.id in completed_tool_call_ids:
                    session.append_tool(
                        name=tool_call.name,
                        tool_call_id=tool_call.id,
                        content='[duplicate tool_call_id, already executed]',
                    )
                    continue
                tool_calls += 1
                if tool_call.name in ('Agent', 'delegate_agent'):
                    delegated_tasks += self._delegated_task_units(tool_call.arguments)
                budget_after_tool_request = self._check_budget(
                    total_usage,
                    total_cost_usd,
                    tool_calls=tool_calls,
                    delegated_tasks=delegated_tasks,
                    model_calls=model_calls,
                    session_turns=starting_session_turns + turn_index,
                )
                if budget_after_tool_request.exceeded:
                    stream_events.append(
                        {
                            'type': 'task_budget_exceeded',
                            'turn_index': turn_index,
                            'tool_name': tool_call.name,
                            'tool_call_id': tool_call.id,
                            'reason': budget_after_tool_request.reason,
                        }
                    )
                    result = AgentRunResult(
                        final_output=(
                            budget_after_tool_request.reason
                            or 'Stopped because the runtime budget was exceeded.'
                        ),
                        turns=turn_index,
                        tool_calls=tool_calls,
                        transcript=session.transcript(),
                        events=tuple(stream_events),
                        usage=total_usage,
                        total_cost_usd=total_cost_usd,
                        stop_reason='budget_exceeded',
                        file_history=tuple(file_history),
                        session_id=session_id,
                        scratchpad_directory=(
                            str(scratchpad_directory) if scratchpad_directory is not None else None
                        ),
                    )
                    result = self._persist_session(session, result)
                    self.last_run_result = result
                    return result
                tool_result = None
                tool_message_index = session.start_tool(
                    name=tool_call.name,
                    tool_call_id=tool_call.id,
                    message_id=f'tool_{len(session.messages)}',
                    metadata={'phase': 'starting'},
                )
                stream_events.append(
                    {
                        'type': 'tool_start',
                        'tool_name': tool_call.name,
                        'tool_call_id': tool_call.id,
                        'message_id': session.messages[tool_message_index].message_id,
                    }
                )
                if self.plugin_runtime is not None:
                    self.plugin_runtime.record_tool_attempt(tool_call.name, blocked=False)
                plugin_preflight_messages = self._plugin_tool_preflight_messages(tool_call.name)
                policy_preflight_messages = self._hook_policy_tool_preflight_messages(
                    tool_call.name
                )
                if plugin_preflight_messages:
                    stream_events.append(
                        {
                            'type': 'plugin_tool_preflight',
                            'tool_name': tool_call.name,
                            'tool_call_id': tool_call.id,
                            'message_id': session.messages[tool_message_index].message_id,
                            'message_count': len(plugin_preflight_messages),
                        }
                    )
                if policy_preflight_messages:
                    stream_events.append(
                        {
                            'type': 'hook_policy_tool_preflight',
                            'tool_name': tool_call.name,
                            'tool_call_id': tool_call.id,
                            'message_id': session.messages[tool_message_index].message_id,
                            'message_count': len(policy_preflight_messages),
                        }
                    )
                plugin_block_message = self._plugin_block_message(tool_call.name)
                policy_block_message = self._hook_policy_block_message(tool_call.name)
                if plugin_block_message is not None:
                    if self.plugin_runtime is not None:
                        blocked_attempts = int(
                            self.plugin_runtime.session_state.get('blocked_tool_attempts', 0)
                        )
                        self.plugin_runtime.session_state['blocked_tool_attempts'] = (
                            blocked_attempts + 1
                        )
                    tool_result = ToolExecutionResult(
                        name=tool_call.name,
                        ok=False,
                        content=plugin_block_message,
                        metadata={
                            'action': 'plugin_block',
                            'plugin_blocked': True,
                            'plugin_block_message': plugin_block_message,
                        },
                    )
                    stream_events.append(
                        {
                            'type': 'plugin_tool_block',
                            'tool_name': tool_call.name,
                            'tool_call_id': tool_call.id,
                            'message_id': session.messages[tool_message_index].message_id,
                            'message': plugin_block_message,
                        }
                    )
                if policy_block_message is not None:
                    tool_result = ToolExecutionResult(
                        name=tool_call.name,
                        ok=False,
                        content=policy_block_message,
                        metadata={
                            'action': 'hook_policy_block',
                            'hook_policy_blocked': True,
                            'hook_policy_block_message': policy_block_message,
                            'error_kind': 'permission_denied',
                        },
                    )
                    stream_events.append(
                        {
                            'type': 'hook_policy_tool_block',
                            'tool_name': tool_call.name,
                            'tool_call_id': tool_call.id,
                            'message_id': session.messages[tool_message_index].message_id,
                            'message': policy_block_message,
                        }
                    )
                if tool_call.name in ('Agent', 'delegate_agent'):
                    if tool_result is None:
                        tool_result = self._execute_delegate_agent(tool_call.arguments)
                elif tool_call.name == 'Skill':
                    if tool_result is None:
                        tool_result = self._execute_skill(tool_call.arguments)
                elif tool_result is None:
                    for update in execute_tool_streaming(
                        self.tool_registry,
                        tool_call.name,
                        tool_call.arguments,
                        self.tool_context,
                    ):
                        if update.kind == 'delta':
                            session.append_tool_delta(
                                tool_message_index,
                                update.content,
                                metadata={'last_stream': update.stream or 'tool'},
                            )
                            stream_events.append(
                                {
                                    'type': 'tool_delta',
                                    'tool_name': tool_call.name,
                                    'tool_call_id': tool_call.id,
                                    'message_id': session.messages[tool_message_index].message_id,
                                    'stream': update.stream,
                                    'delta': update.content,
                                }
                            )
                            continue
                        tool_result = update.result
                if tool_result is None:
                    raise RuntimeError(f'Tool executor returned no final result for {tool_call.name}')
                if self.plugin_runtime is not None:
                    self.plugin_runtime.record_tool_result(
                        tool_call.name,
                        ok=tool_result.ok,
                        metadata=tool_result.metadata,
                    )
                plugin_messages = self._plugin_tool_result_messages(tool_call.name)
                policy_messages = self._hook_policy_tool_result_messages(tool_call.name)
                if plugin_messages:
                    merged_metadata = dict(tool_result.metadata)
                    merged_metadata['plugin_messages'] = list(plugin_messages)
                    tool_result = ToolExecutionResult(
                        name=tool_result.name,
                        ok=tool_result.ok,
                        content=tool_result.content,
                        metadata=merged_metadata,
                    )
                    for message in plugin_messages:
                        stream_events.append(
                            {
                                'type': 'plugin_tool_hook',
                                'tool_name': tool_call.name,
                                'tool_call_id': tool_call.id,
                                'message_id': session.messages[tool_message_index].message_id,
                                'message': message,
                            }
                        )
                if policy_messages:
                    merged_metadata = dict(tool_result.metadata)
                    merged_metadata['hook_policy_messages'] = list(policy_messages)
                    tool_result = ToolExecutionResult(
                        name=tool_result.name,
                        ok=tool_result.ok,
                        content=tool_result.content,
                        metadata=merged_metadata,
                    )
                    for message in policy_messages:
                        stream_events.append(
                            {
                                'type': 'hook_policy_tool_hook',
                                'tool_name': tool_call.name,
                                'tool_call_id': tool_call.id,
                                'message_id': session.messages[tool_message_index].message_id,
                                'message': message,
                            }
                        )
                if tool_result.metadata.get('error_kind') == 'permission_denied':
                    stream_events.append(
                        {
                            'type': 'tool_permission_denial',
                            'tool_name': tool_call.name,
                            'tool_call_id': tool_call.id,
                            'message_id': session.messages[tool_message_index].message_id,
                            'reason': tool_result.content,
                            'source': (
                                'hook_policy'
                                if tool_result.metadata.get('action') == 'hook_policy_block'
                                else 'tool_runtime'
                            ),
                        }
                    )
                session.finalize_tool(
                    tool_message_index,
                    content=serialize_tool_result(tool_result),
                    metadata={
                        'phase': 'completed',
                        'plugin_preflight_messages': list(plugin_preflight_messages),
                        'hook_policy_preflight_messages': list(policy_preflight_messages),
                        **dict(tool_result.metadata),
                    },
                    stop_reason='tool_completed',
                )
                completed_tool_call_ids.add(tool_call.id)
                stream_events.append(
                    {
                        'type': 'tool_result',
                        'tool_name': tool_call.name,
                        'tool_call_id': tool_call.id,
                        'message_id': session.messages[tool_message_index].message_id,
                        'ok': tool_result.ok,
                        'metadata': dict(tool_result.metadata),
                    }
                )
                self._append_runtime_tool_followup_events(
                    stream_events,
                    tool_call=tool_call,
                    tool_result=tool_result,
                )
                plugin_runtime_message = self._build_plugin_tool_runtime_message(
                    tool_name=tool_call.name,
                    preflight_messages=plugin_preflight_messages,
                    block_message=plugin_block_message,
                    plugin_messages=plugin_messages,
                    hook_policy_preflight_messages=policy_preflight_messages,
                    hook_policy_block_message=policy_block_message,
                    hook_policy_messages=policy_messages,
                    delegate_preflight_messages=tuple(
                        message
                        for message in tool_result.metadata.get(
                            'plugin_delegate_preflight_messages',
                            [],
                        )
                        if isinstance(message, str) and message
                    ),
                    delegate_after_messages=tuple(
                        message
                        for message in tool_result.metadata.get(
                            'plugin_delegate_after_messages',
                            [],
                        )
                        if isinstance(message, str) and message
                    ),
                )
                if plugin_runtime_message is not None:
                    session.append_user(
                        plugin_runtime_message,
                        metadata={
                            'kind': 'plugin_tool_runtime',
                            'tool_name': tool_call.name,
                            'tool_call_id': tool_call.id,
                            'plugin_blocked': plugin_block_message is not None,
                            'plugin_message_count': len(plugin_messages),
                            'plugin_preflight_count': len(plugin_preflight_messages),
                        },
                        message_id=f'plugin_tool_runtime_{tool_call.id}',
                    )
                    stream_events.append(
                        {
                            'type': 'plugin_tool_context',
                            'tool_name': tool_call.name,
                            'tool_call_id': tool_call.id,
                            'message_id': f'plugin_tool_runtime_{tool_call.id}',
                            'blocked': plugin_block_message is not None,
                            'message_count': len(plugin_messages),
                            'preflight_count': len(plugin_preflight_messages),
                        }
                    )
                self._refresh_runtime_views_for_tool_result(tool_call.name, tool_result)
                history_entry = self._build_file_history_entry(
                    tool_call=tool_call,
                    tool_result=tool_result,
                    turn_index=turn_index,
                )
                if history_entry is not None:
                    file_history.append(history_entry)

        result = AgentRunResult(
            final_output=(
                last_content
                or 'Stopped: max turns reached before the model produced a final answer.'
            ),
            turns=self.runtime_config.max_turns,
            tool_calls=tool_calls,
            transcript=session.transcript(),
            events=tuple(stream_events),
            usage=total_usage,
            total_cost_usd=total_cost_usd,
            stop_reason='max_turns',
            file_history=tuple(file_history),
            session_id=session_id,
            scratchpad_directory=(
                str(scratchpad_directory) if scratchpad_directory is not None else None
            ),
        )
        result = self._append_runtime_after_turn_events(
            result,
            prompt=effective_prompt,
            turn_index=self.runtime_config.max_turns,
        )
        result = self._persist_session(session, result)
        self.last_run_result = result
        return result

    def _query_model(
        self,
        session: AgentSessionState,
        tool_specs: list[dict[str, object]],
    ) -> tuple[AssistantTurn, tuple[StreamEvent, ...]]:
        if not self.runtime_config.stream_model_responses:
            turn = self.client.complete(
                session.to_openai_messages(),
                tool_specs,
                output_schema=self.runtime_config.output_schema,
            )
            assistant_tool_calls = tuple(
                {
                    'id': tool_call.id,
                    'type': 'function',
                    'function': {
                        'name': tool_call.name,
                        'arguments': json.dumps(
                            tool_call.arguments,
                            ensure_ascii=True,
                        ),
                    },
                }
                for tool_call in turn.tool_calls
            )
            session.append_assistant(
                turn.content,
                assistant_tool_calls,
                message_id=f'assistant_{len(session.messages)}',
                stop_reason=turn.finish_reason,
                usage=turn.usage,
            )
            return turn, ()

        assistant_index = session.start_assistant(
            message_id=f'assistant_{len(session.messages)}'
        )
        usage = UsageStats()
        finish_reason: str | None = None
        events: list[StreamEvent] = []
        for event in self.client.stream(
            session.to_openai_messages(),
            tool_specs,
            output_schema=self.runtime_config.output_schema,
        ):
            events.append(event)
            if event.type == 'content_delta':
                session.append_assistant_delta(assistant_index, event.delta)
            elif event.type == 'tool_call_delta':
                session.merge_assistant_tool_call_delta(
                    assistant_index,
                    tool_call_index=event.tool_call_index or 0,
                    tool_call_id=event.tool_call_id,
                    tool_name=event.tool_name,
                    arguments_delta=event.arguments_delta,
                )
            elif event.type == 'usage':
                usage = usage + event.usage
            elif event.type == 'message_stop':
                finish_reason = event.finish_reason

        session.finalize_assistant(
            assistant_index,
            finish_reason=finish_reason,
            usage=usage,
        )
        assistant_message = session.messages[assistant_index]
        turn = AssistantTurn(
            content=assistant_message.content,
            tool_calls=self._tool_calls_from_message(assistant_message.tool_calls),
            finish_reason=finish_reason,
            raw_message=assistant_message.to_openai_message(),
            usage=usage,
        )
        return turn, tuple(events)

    def _tool_calls_from_message(
        self,
        tool_calls: tuple[dict[str, object], ...],
    ) -> tuple[ToolCall, ...]:
        parsed: list[ToolCall] = []
        for index, raw_tool_call in enumerate(tool_calls):
            function_block = raw_tool_call.get('function')
            if not isinstance(function_block, dict):
                continue
            name = function_block.get('name')
            if not isinstance(name, str) or not name:
                continue
            raw_arguments = function_block.get('arguments', '')
            if isinstance(raw_arguments, str) and raw_arguments.strip():
                arguments = json.loads(raw_arguments)
                if not isinstance(arguments, dict):
                    raise OpenAICompatError(
                        f'Tool arguments must decode to an object, got {type(arguments).__name__}'
                    )
            else:
                arguments = {}
            call_id = raw_tool_call.get('id')
            if not isinstance(call_id, str) or not call_id:
                call_id = f'call_{index}'
            parsed.append(
                ToolCall(
                    id=call_id,
                    name=name,
                    arguments=arguments,
                )
            )
        return tuple(parsed)

    def _should_continue_response(self, turn: AssistantTurn) -> bool:
        return turn.finish_reason in {'length', 'max_tokens'}

    def _build_continuation_prompt(self) -> str:
        return (
            '<system-reminder>\n'
            'Your previous answer was truncated because the model stopped early. '
            'Continue exactly where you left off. Do not repeat completed text.\n'
            '</system-reminder>'
        )

    def _check_budget(
        self,
        usage: UsageStats,
        total_cost_usd: float,
        *,
        tool_calls: int,
        delegated_tasks: int,
        model_calls: int,
        session_turns: int,
    ) -> BudgetDecision:
        budget = self.runtime_config.budget_config
        token_reason = self._check_token_budget(usage, budget)
        if token_reason is not None:
            return BudgetDecision(exceeded=True, reason=token_reason)
        if (
            budget.max_total_cost_usd is not None
            and total_cost_usd > budget.max_total_cost_usd
        ):
            return BudgetDecision(
                exceeded=True,
                reason=(
                    'Stopped because the total estimated cost '
                    f'(${total_cost_usd:.6f}) exceeded the configured budget '
                    f'(${budget.max_total_cost_usd:.6f}).'
                ),
            )
        if (
            budget.max_tool_calls is not None
            and tool_calls > budget.max_tool_calls
        ):
            return BudgetDecision(
                exceeded=True,
                reason=(
                    'Stopped because the tool-call budget was exceeded '
                    f'({tool_calls} > {budget.max_tool_calls}).'
                ),
            )
        if (
            budget.max_delegated_tasks is not None
            and delegated_tasks > budget.max_delegated_tasks
        ):
            return BudgetDecision(
                exceeded=True,
                reason=(
                    'Stopped because the delegated-task budget was exceeded '
                    f'({delegated_tasks} > {budget.max_delegated_tasks}).'
                ),
            )
        if (
            budget.max_model_calls is not None
            and model_calls > budget.max_model_calls
        ):
            return BudgetDecision(
                exceeded=True,
                reason=(
                    'Stopped because the model-call budget was exceeded '
                    f'({model_calls} > {budget.max_model_calls}).'
                ),
            )
        if (
            budget.max_session_turns is not None
            and session_turns > budget.max_session_turns
        ):
            return BudgetDecision(
                exceeded=True,
                reason=(
                    'Stopped because the session-turn budget was exceeded '
                    f'({session_turns} > {budget.max_session_turns}).'
                ),
            )
        return BudgetDecision(exceeded=False)

    def _preflight_prompt_length(
        self,
        session: AgentSessionState,
        stream_events: list[dict[str, object]],
        *,
        turn_index: int,
    ) -> PromptPreflightResult:
        snapshot = calculate_token_budget(
            session=session,
            model=self.model_config.model,
            budget_config=self.runtime_config.budget_config,
            output_schema=self.runtime_config.output_schema,
        )
        if not snapshot.exceeds_soft_limit and not snapshot.exceeds_hard_limit:
            return PromptPreflightResult()

        stream_events.append(
            {
                'type': 'prompt_length_check',
                'turn_index': turn_index,
                'projected_input_tokens': snapshot.projected_input_tokens,
                'soft_input_limit_tokens': snapshot.soft_input_limit_tokens,
                'hard_input_limit_tokens': snapshot.hard_input_limit_tokens,
                'soft_overflow_tokens': snapshot.soft_overflow_tokens,
                'overflow_tokens': snapshot.overflow_tokens,
                'exceeds_hard_limit': snapshot.exceeds_hard_limit,
            }
        )

        target_tokens = snapshot.soft_input_limit_tokens
        if snapshot.exceeds_hard_limit:
            target_tokens = snapshot.hard_input_limit_tokens
        if target_tokens < 0:
            target_tokens = 0

        if self._reduce_context_pressure(
            session,
            stream_events,
            turn_index=turn_index,
            target_tokens=target_tokens,
            allow_compaction=True,
        ):
            recovered = calculate_token_budget(
                session=session,
                model=self.model_config.model,
                budget_config=self.runtime_config.budget_config,
                output_schema=self.runtime_config.output_schema,
            )
            stream_events.append(
                {
                    'type': 'prompt_length_recovery',
                    'turn_index': turn_index,
                    'strategy': 'heuristic',
                    'projected_input_tokens': recovered.projected_input_tokens,
                    'soft_input_limit_tokens': recovered.soft_input_limit_tokens,
                    'hard_input_limit_tokens': recovered.hard_input_limit_tokens,
                    'exceeds_hard_limit': recovered.exceeds_hard_limit,
                    'exceeds_soft_limit': recovered.exceeds_soft_limit,
                }
            )
            if not recovered.exceeds_soft_limit and not recovered.exceeds_hard_limit:
                return PromptPreflightResult()
            snapshot = recovered

        # Circuit-breaker: skip auto-compact after MAX_COMPACT_FAILURES consecutive failures
        from .compact import MAX_COMPACT_FAILURES
        if self._compact_consecutive_failures >= MAX_COMPACT_FAILURES:
            stream_events.append(
                {
                    'type': 'auto_compact_circuit_breaker',
                    'turn_index': turn_index,
                    'consecutive_failures': self._compact_consecutive_failures,
                }
            )
        elif self._can_auto_compact_with_summary(session):
            compact_result = compact_conversation(
                self,
                custom_instructions=(
                    'Automatically collapse earlier conversation context to fit the next model '
                    'turn. Preserve the active task, recent file changes, failures, pending work, '
                    'and exact next step.'
                ),
            )
            if compact_result.error is None:
                self._compact_consecutive_failures = 0  # Reset on success
                recovered = calculate_token_budget(
                    session=session,
                    model=self.model_config.model,
                    budget_config=self.runtime_config.budget_config,
                    output_schema=self.runtime_config.output_schema,
                )
                stream_events.append(
                    {
                        'type': 'auto_compact_summary',
                        'turn_index': turn_index,
                        'pre_compact_token_count': compact_result.pre_compact_token_count,
                        'post_compact_token_count': compact_result.post_compact_token_count,
                        'true_post_compact_token_count': compact_result.true_post_compact_token_count,
                        'summary_usage_tokens': compact_result.usage.total_tokens,
                        'ptl_retries': compact_result.ptl_retries,
                        'projected_input_tokens': recovered.projected_input_tokens,
                        'soft_input_limit_tokens': recovered.soft_input_limit_tokens,
                        'hard_input_limit_tokens': recovered.hard_input_limit_tokens,
                        'exceeds_hard_limit': recovered.exceeds_hard_limit,
                        'exceeds_soft_limit': recovered.exceeds_soft_limit,
                    }
                )
                if not recovered.exceeds_soft_limit and not recovered.exceeds_hard_limit:
                    return PromptPreflightResult(
                        usage_increment=compact_result.usage,
                        model_calls_increment=1,
                    )
                snapshot = recovered
                if compact_result.usage.total_tokens:
                    return PromptPreflightResult(
                        usage_increment=compact_result.usage,
                        model_calls_increment=1,
                        stop_reason=(
                            'prompt_too_long'
                            if recovered.exceeds_hard_limit
                            else None
                        ),
                        reason=(
                            self._build_prompt_length_error(recovered)
                            if recovered.exceeds_hard_limit
                            else None
                        ),
                    )
            else:
                self._compact_consecutive_failures += 1
                stream_events.append(
                    {
                        'type': 'auto_compact_failed',
                        'turn_index': turn_index,
                        'reason': compact_result.error,
                        'consecutive_failures': self._compact_consecutive_failures,
                    }
                )

        if snapshot.exceeds_hard_limit:
            return PromptPreflightResult(
                stop_reason='prompt_too_long',
                reason=self._build_prompt_length_error(snapshot),
            )

        stream_events.append(
            {
                'type': 'prompt_length_warning',
                'turn_index': turn_index,
                'projected_input_tokens': snapshot.projected_input_tokens,
                'soft_input_limit_tokens': snapshot.soft_input_limit_tokens,
                'hard_input_limit_tokens': snapshot.hard_input_limit_tokens,
                'soft_overflow_tokens': snapshot.soft_overflow_tokens,
            }
        )
        return PromptPreflightResult()

    def _can_auto_compact_with_summary(self, session: AgentSessionState) -> bool:
        prefix_count = self._compact_prefix_count(session)
        preserve_count = max(self.runtime_config.compact_preserve_messages, 1)
        return len(session.messages) - prefix_count > preserve_count

    def _build_prompt_length_error(self, snapshot) -> str:
        return (
            'Stopped before the next model call because the prompt would exceed the '
            'effective input budget. '
            f'Projected prompt tokens: {snapshot.projected_input_tokens:,}; '
            f'hard input limit: {snapshot.hard_input_limit_tokens:,}; '
            f'soft input limit: {snapshot.soft_input_limit_tokens:,}.'
        )

    def _microcompact_session_if_needed(
        self,
        session: AgentSessionState,
        stream_events: list[dict[str, object]],
        *,
        turn_index: int,
    ) -> None:
        """Run time-based microcompaction to clear old tool results.

        Fires when the gap since the last assistant message exceeds the
        threshold (default 60 minutes), indicating the server-side cache
        has expired and the full prefix will be rewritten anyway.
        """
        if not session.messages:
            return
        result = _microcompact_messages(
            session.messages,
            model=self.model_config.model,
        )
        if not result.triggered:
            return
        session.messages = result.messages
        stream_events.append(
            {
                'type': 'microcompact',
                'turn_index': turn_index,
                'cleared_tool_count': result.cleared_tool_count,
                'kept_tool_count': result.kept_tool_count,
                'estimated_tokens_saved': result.estimated_tokens_saved,
                'gap_minutes': round(result.gap_minutes, 1),
            }
        )

    def _snip_session_if_needed(
        self,
        session: AgentSessionState,
        stream_events: list[dict[str, object]],
        *,
        turn_index: int,
    ) -> None:
        threshold = self.runtime_config.auto_snip_threshold_tokens
        if threshold is None or threshold <= 0:
            return
        self._reduce_context_pressure(
            session,
            stream_events,
            turn_index=turn_index,
            target_tokens=threshold,
            allow_compaction=False,
        )

    def _compact_session_if_needed(
        self,
        session: AgentSessionState,
        stream_events: list[dict[str, object]],
        *,
        turn_index: int,
    ) -> None:
        threshold = self.runtime_config.auto_compact_threshold_tokens
        if threshold is None or threshold <= 0:
            return
        self._reduce_context_pressure(
            session,
            stream_events,
            turn_index=turn_index,
            target_tokens=threshold,
            allow_compaction=True,
        )

    def _reactive_compact_session(
        self,
        session: AgentSessionState,
        stream_events: list[dict[str, object]],
        *,
        turn_index: int,
    ) -> bool:
        return self._reduce_context_pressure(
            session,
            stream_events,
            turn_index=turn_index,
            target_tokens=0,
            allow_compaction=True,
            reactive=True,
        )

    def _reduce_context_pressure(
        self,
        session: AgentSessionState,
        stream_events: list[dict[str, object]],
        *,
        turn_index: int,
        target_tokens: int,
        allow_compaction: bool,
        reactive: bool = False,
    ) -> bool:
        changed = False
        for _ in range(6):
            usage_report = collect_context_usage(
                session=session,
                model=self.model_config.model,
                strategy='reactive_compact' if reactive else 'context_pressure',
            )
            if usage_report.total_tokens <= target_tokens:
                break
            if self._snip_session_pass(
                session,
                stream_events,
                turn_index=turn_index,
                target_tokens=target_tokens,
                current_total=usage_report.total_tokens,
                reactive=reactive,
            ):
                changed = True
                continue
            if allow_compaction and self._compact_session_pass(
                session,
                stream_events,
                turn_index=turn_index,
                usage_total=usage_report.total_tokens,
                reactive=reactive,
            ):
                changed = True
                if reactive:
                    continue
                break
            break
        return changed

    def _snip_session_pass(
        self,
        session: AgentSessionState,
        stream_events: list[dict[str, object]],
        *,
        turn_index: int,
        target_tokens: int,
        current_total: int,
        reactive: bool,
    ) -> bool:
        prefix_count = self._compact_prefix_count(session)
        tail_count = min(
            max(self.runtime_config.compact_preserve_messages, 0),
            max(len(session.messages) - prefix_count, 0),
        )
        candidate_indexes = [
            index
            for index in range(prefix_count, max(len(session.messages) - tail_count, prefix_count))
            if self._message_can_be_snipped(session.messages[index])
        ]
        if not candidate_indexes:
            return False
        snipped_count = 0
        tokens_removed = 0
        snipped_message_ids: list[str] = []
        for index in candidate_indexes:
            if current_total <= target_tokens and not reactive:
                break
            message = session.messages[index]
            original_tokens = estimate_tokens(message.content, self.model_config.model)
            replacement = self._build_snipped_message_content(message)
            replacement_tokens = estimate_tokens(replacement, self.model_config.model)
            if replacement_tokens >= original_tokens:
                continue
            session.tombstone_message(
                index,
                summary=replacement,
                stop_reason='snipped_for_context',
                mutation_kind='snip_tombstone',
                metadata={
                    'kind': 'snipped_message',
                    'original_token_estimate': original_tokens,
                    'replacement_token_estimate': replacement_tokens,
                    'snipped_turn_index': turn_index,
                    'snipped_from_role': message.role,
                    'snipped_from_message_id': message.message_id,
                    'snipped_from_kind': message.metadata.get('kind'),
                    'snipped_from_lineage_id': message.metadata.get('lineage_id'),
                    'snipped_from_revision': message.metadata.get('revision'),
                },
            )
            delta = original_tokens - replacement_tokens
            current_total -= delta
            tokens_removed += delta
            snipped_count += 1
            if session.messages[index].message_id:
                snipped_message_ids.append(session.messages[index].message_id)
            if reactive and snipped_count >= 3:
                break
        if not snipped_count:
            return False
        stream_events.append(
            {
                'type': 'reactive_snip_boundary' if reactive else 'snip_boundary',
                'turn_index': turn_index,
                'snipped_message_count': snipped_count,
                'estimated_tokens_removed': tokens_removed,
                'snipped_message_ids': snipped_message_ids,
            }
        )
        return True

    def _compact_session_pass(
        self,
        session: AgentSessionState,
        stream_events: list[dict[str, object]],
        *,
        turn_index: int,
        usage_total: int,
        reactive: bool,
    ) -> bool:
        prefix_count = self._compact_prefix_count(session)
        preserve_messages = max(self.runtime_config.compact_preserve_messages, 0)
        if reactive:
            preserve_messages = max(preserve_messages // 2, 1)
        tail_count = min(
            preserve_messages,
            max(len(session.messages) - prefix_count, 0),
        )
        compact_end = len(session.messages) - tail_count
        if compact_end <= prefix_count:
            return False
        candidates = session.messages[prefix_count:compact_end]
        preserved_tail = list(session.messages[compact_end:])
        if not candidates:
            return False
        compacted_tokens = sum(
            usage.tokens
            for usage in (
                collect_context_usage(
                    session=AgentSessionState(
                        system_prompt_parts=session.system_prompt_parts,
                        user_context=session.user_context,
                        system_context=session.system_context,
                        messages=list(candidates),
                    ),
                    model=self.model_config.model,
                    strategy='compacted_segment',
                ).categories
            )
            if usage.name != 'Free space'
        )
        compact_message = self._build_compact_boundary_message(
            candidates,
            turn_index=turn_index,
            estimated_tokens_before=usage_total,
            estimated_tokens_removed=compacted_tokens,
            preserved_tail_count=tail_count,
            preserved_tail=preserved_tail,
        )
        session.messages = (
            session.messages[:prefix_count]
            + [compact_message]
            + session.messages[compact_end:]
        )
        stream_events.append(
            {
                'type': 'reactive_compact_boundary' if reactive else 'compact_boundary',
                'turn_index': turn_index,
                'compacted_message_count': len(candidates),
                'estimated_tokens_before': usage_total,
                'estimated_tokens_removed': compacted_tokens,
                'preserved_tail_count': tail_count,
                'preserved_tail_ids': [
                    message.message_id for message in preserved_tail if message.message_id
                ],
                'compaction_depth': compact_message.metadata.get('compaction_depth'),
                'nested_compaction_count': compact_message.metadata.get('nested_compaction_count'),
                'compacted_message_ids': [
                    message.message_id for message in candidates if message.message_id
                ],
            }
        )
        return True

    def _check_token_budget(
        self,
        usage: UsageStats,
        budget: BudgetConfig,
    ) -> str | None:
        if budget.max_total_tokens is not None and usage.total_tokens > budget.max_total_tokens:
            return (
                'Stopped because the total token budget was exceeded '
                f'({usage.total_tokens} > {budget.max_total_tokens}).'
            )
        if budget.max_input_tokens is not None and usage.input_tokens > budget.max_input_tokens:
            return (
                'Stopped because the input token budget was exceeded '
                f'({usage.input_tokens} > {budget.max_input_tokens}).'
            )
        if budget.max_output_tokens is not None and usage.output_tokens > budget.max_output_tokens:
            return (
                'Stopped because the output token budget was exceeded '
                f'({usage.output_tokens} > {budget.max_output_tokens}).'
            )
        if (
            budget.max_reasoning_tokens is not None
            and usage.reasoning_tokens > budget.max_reasoning_tokens
        ):
            return (
                'Stopped because the reasoning token budget was exceeded '
                f'({usage.reasoning_tokens} > {budget.max_reasoning_tokens}).'
            )
        return None

    def _build_file_history_entry(
        self,
        *,
        tool_call: ToolCall,
        tool_result,
        turn_index: int,
    ) -> dict[str, object] | None:
        if not tool_result.metadata:
            return None
        if (
            'path' not in tool_result.metadata
            and 'command' not in tool_result.metadata
            and tool_result.metadata.get('action') not in ('delegate_agent', 'Agent')
        ):
            return None
        metadata = dict(tool_result.metadata)
        entry: dict[str, object] = {
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'turn_index': turn_index,
            'tool_call_id': tool_call.id,
            'tool_name': tool_call.name,
            'ok': tool_result.ok,
            'history_entry_id': f'{turn_index}:{tool_call.id}:{tool_call.name}',
            'result_preview': self._preview_text(tool_result.content, 220),
            **metadata,
        }
        action = metadata.get('action')
        path = metadata.get('path')
        if isinstance(path, str) and path:
            entry['history_kind'] = 'file_change'
            entry['changed_paths'] = [path]
            before_sha256 = metadata.get('before_sha256')
            if isinstance(before_sha256, str) and before_sha256:
                entry['before_snapshot_id'] = f'{path}:{before_sha256[:12]}'
            after_sha256 = metadata.get('after_sha256')
            if isinstance(after_sha256, str) and after_sha256:
                entry['after_snapshot_id'] = f'{path}:{after_sha256[:12]}'
        elif isinstance(metadata.get('command'), str):
            entry['history_kind'] = 'shell'
        elif action in ('delegate_agent', 'Agent'):
            entry['history_kind'] = 'delegation'
            delegate_batches = metadata.get('delegate_batches')
            if isinstance(delegate_batches, list):
                entry['delegate_batch_count'] = len(delegate_batches)
            dependency_skips = metadata.get('dependency_skips')
            if isinstance(dependency_skips, int) and not isinstance(dependency_skips, bool):
                entry['dependency_skips'] = dependency_skips
        else:
            entry['history_kind'] = 'tool'
        return entry

    def _compact_prefix_count(self, session: AgentSessionState) -> int:
        prefix_count = 0
        for message in session.messages:
            if prefix_count == 0 and message.role == 'system':
                prefix_count += 1
                continue
            if (
                prefix_count == 1
                and message.role == 'user'
                and message.content.startswith('<system-reminder>')
            ):
                prefix_count += 1
                continue
            break
        return prefix_count

    def _message_can_be_snipped(self, message) -> bool:
        if message.metadata.get('kind') in {
            'compact_boundary',
            'snipped_message',
            'file_history_replay',
        }:
            return False
        if message.role == 'tool':
            return True
        if message.role == 'assistant' and (message.tool_calls or len(message.content) > 600):
            return True
        if (
            message.role == 'user'
            and message.metadata.get('kind') in {'continuation_request', 'file_history_replay'}
        ):
            return True
        return False

    def _build_snipped_message_content(self, message) -> str:
        preview = ' '.join(message.content.split())
        if len(preview) > 120:
            preview = preview[:117] + '...'
        if message.role == 'tool':
            label = f'tool result ({message.name or "tool"})'
        elif message.role == 'assistant':
            label = 'assistant message with tool calls'
        else:
            label = message.role
        return (
            '<system-reminder>\n'
            f'Older {label} was snipped to save context.\n'
            f'Message id: {message.message_id or "(none)"}\n'
            f'Preview: {preview or "(empty)"}\n'
            '</system-reminder>'
        )

    def _build_compact_boundary_message(
        self,
        messages,
        *,
        turn_index: int,
        estimated_tokens_before: int,
        estimated_tokens_removed: int,
        preserved_tail_count: int,
        preserved_tail,
    ):
        summary_lines = [
            '<system-reminder>',
            'Earlier conversation history was compacted to keep the session within the context budget.',
            '',
            'Compacted summary:',
        ]
        remaining = 24
        for message in messages:
            if remaining <= 0:
                break
            label = message.role
            if message.role == 'tool' and message.name:
                label = f'tool:{message.name}'
            snippet = ' '.join(message.content.split())
            if len(snippet) > 160:
                snippet = snippet[:157] + '...'
            if not snippet:
                snippet = '(empty)'
            summary_lines.append(f'- {label}: {snippet}')
            remaining -= 1
        if len(messages) > 24:
            summary_lines.append(f'- ... plus {len(messages) - 24} older messages')
        summary_lines.extend(
            [
                '',
                'Keep using the preserved recent tail as the active working set.',
                '</system-reminder>',
            ]
        )
        from .agent_session import AgentMessage

        nested_compaction_count = sum(
            1 for message in messages if message.metadata.get('kind') == 'compact_boundary'
        )
        prior_depths = [
            int(message.metadata.get('compaction_depth', 0))
            for message in messages
            if isinstance(message.metadata.get('compaction_depth', 0), int)
        ]
        compaction_depth = (max(prior_depths) if prior_depths else 0) + 1
        compacted_kinds: dict[str, int] = {}
        source_mutation_totals: dict[str, int] = {}
        compacted_lineage_ids: list[str] = []
        preserved_tail_lineage_ids = [
            lineage_id
            for lineage_id in (
                message.metadata.get('lineage_id') for message in preserved_tail
            )
            if isinstance(lineage_id, str) and lineage_id
        ]
        max_source_revision = 0
        max_source_mutation_serial = 0
        compacted_revision_total = 0
        for message in messages:
            kind = message.metadata.get('kind')
            label = str(kind) if isinstance(kind, str) and kind else message.role
            compacted_kinds[label] = compacted_kinds.get(label, 0) + 1
            lineage_id = message.metadata.get('lineage_id')
            if isinstance(lineage_id, str) and lineage_id:
                compacted_lineage_ids.append(lineage_id)
            revision = message.metadata.get('revision')
            if isinstance(revision, int) and not isinstance(revision, bool):
                max_source_revision = max(max_source_revision, revision)
                compacted_revision_total += revision
            max_mutation_serial = message.metadata.get('max_mutation_serial')
            if isinstance(max_mutation_serial, int) and not isinstance(max_mutation_serial, bool):
                max_source_mutation_serial = max(
                    max_source_mutation_serial,
                    max_mutation_serial,
                )
            mutation_totals = message.metadata.get('mutation_totals')
            if isinstance(mutation_totals, dict):
                for mutation_kind, count in mutation_totals.items():
                    if (
                        not isinstance(mutation_kind, str)
                        or not mutation_kind
                        or isinstance(count, bool)
                        or not isinstance(count, int)
                        or count <= 0
                    ):
                        continue
                    source_mutation_totals[mutation_kind] = (
                        source_mutation_totals.get(mutation_kind, 0) + count
                    )

        compact_boundary_id = f'compact_boundary_{turn_index}_{len(messages)}'

        return AgentMessage(
            role='system',
            content='\n'.join(summary_lines),
            message_id=compact_boundary_id,
            metadata={
                'kind': 'compact_boundary',
                'lineage_id': compact_boundary_id,
                'revision': 0,
                'revision_count': 1,
                'message_role': 'system',
                'turn_index': turn_index,
                'compacted_message_count': len(messages),
                'estimated_tokens_before': estimated_tokens_before,
                'estimated_tokens_removed': estimated_tokens_removed,
                'preserved_tail_count': preserved_tail_count,
                'preserved_tail_ids': [
                    message.message_id for message in preserved_tail if message.message_id
                ],
                'compaction_depth': compaction_depth,
                'nested_compaction_count': nested_compaction_count,
                'compacted_kinds': compacted_kinds,
                'compacted_lineage_ids': compacted_lineage_ids,
                'preserved_tail_lineage_ids': preserved_tail_lineage_ids,
                'max_source_revision': max_source_revision,
                'max_source_mutation_serial': max_source_mutation_serial,
                'source_mutation_totals': source_mutation_totals,
                'compacted_revision_total': compacted_revision_total,
                'compacted_message_ids': [
                    message.message_id for message in messages if message.message_id
                ],
            },
        )

    def _is_prompt_too_long_error(self, exc: Exception) -> bool:
        text = str(exc).lower()
        patterns = (
            'prompt is too long',
            'maximum context length',
            'context length exceeded',
            'too many tokens',
            'input too long',
            'context window',
        )
        return any(pattern in text for pattern in patterns)

    def _execute_skill(
        self,
        arguments: dict[str, object],
    ) -> ToolExecutionResult:
        """Execute a skill through the Skill tool.

        Checks bundled skills first, then falls back to slash commands.
        """
        from .agent_slash_commands import find_slash_command, get_slash_command_specs
        from .bundled_skills import find_bundled_skill, get_bundled_skills

        skill_name = arguments.get('skill')
        if not isinstance(skill_name, str) or not skill_name.strip():
            return ToolExecutionResult(
                name='Skill',
                ok=False,
                content='skill must be a non-empty string',
            )

        # Normalize: strip leading '/' if present
        skill_name = skill_name.strip().lstrip('/')
        args = arguments.get('args', '')
        if not isinstance(args, str):
            args = str(args) if args is not None else ''

        # 1. Check bundled skills first
        bundled = find_bundled_skill(skill_name)
        if bundled is not None:
            prompt = bundled.get_prompt(self, args.strip())
            return ToolExecutionResult(
                name='Skill',
                ok=True,
                content=prompt,
                metadata={
                    'action': 'skill',
                    'skill_name': skill_name,
                    'source': 'bundled',
                    'should_query': True,
                },
            )

        # 2. Fall back to slash commands
        spec = find_slash_command(skill_name)
        if spec is None:
            available_cmds = sorted(
                name
                for s in get_slash_command_specs()
                for name in s.names
            )
            available_skills = [sk.name for sk in get_bundled_skills()]
            all_available = sorted(set(available_cmds + available_skills))
            return ToolExecutionResult(
                name='Skill',
                ok=False,
                content=(
                    f'Unknown skill: {skill_name}. '
                    f'Available skills: {", ".join(all_available[:30])}'
                ),
                metadata={'action': 'skill_not_found', 'skill_name': skill_name},
            )

        # Invoke the slash command handler
        input_text = f'/{skill_name} {args}'.strip()
        result = spec.handler(self, args.strip(), input_text)

        if result.output:
            content = result.output
        elif result.prompt:
            content = result.prompt
        else:
            content = f'Skill /{skill_name} completed.'

        return ToolExecutionResult(
            name='Skill',
            ok=True,
            content=content,
            metadata={
                'action': 'skill',
                'skill_name': skill_name,
                'command_name': spec.names[0],
                'handled': result.handled,
                'should_query': result.should_query,
            },
        )

    def _resolve_agent_definition(
        self,
        arguments: dict[str, object],
    ) -> AgentDefinition:
        """Resolve the agent definition from subagent_type or default to general-purpose."""
        subagent_type = arguments.get('subagent_type')
        if isinstance(subagent_type, str) and subagent_type:
            agent_def = find_agent_definition(self.runtime_config.cwd, subagent_type)
            if agent_def is not None:
                return agent_def
        return GENERAL_PURPOSE_AGENT

    def _resolve_child_model_config(
        self,
        arguments: dict[str, object],
        agent_def: AgentDefinition,
    ) -> ModelConfig:
        """Resolve model config for a child agent based on explicit override or agent definition."""
        model_override = arguments.get('model')
        agent_model = agent_def.model

        # Explicit model param in arguments takes priority
        if isinstance(model_override, str) and model_override.strip():
            return replace(self.model_config, model=model_override.strip())

        # Agent definition model
        if agent_model and agent_model != 'inherit':
            return replace(self.model_config, model=agent_model)

        return self.model_config

    def _filter_tools_for_agent(
        self,
        agent_def: AgentDefinition,
    ) -> dict[str, AgentTool]:
        """Build the tool registry for a child agent based on its definition."""
        # Start from parent tools, remove Agent/delegate_agent to prevent recursive spawning
        base_tools = {
            name: tool
            for name, tool in self.tool_registry.items()
            if name not in ('delegate_agent', 'Agent')
        }

        # Apply agent-specific tool allow-list
        if agent_def.tools is not None:
            allowed = set(agent_def.tools)
            base_tools = {
                name: tool
                for name, tool in base_tools.items()
                if name in allowed
            }

        # Apply agent-specific disallowed tools
        if agent_def.disallowed_tools:
            denied = set(agent_def.disallowed_tools)
            base_tools = {
                name: tool
                for name, tool in base_tools.items()
                if name not in denied
            }

        # Apply universal agent disallowed tools
        base_tools = {
            name: tool
            for name, tool in base_tools.items()
            if name not in ALL_AGENT_DISALLOWED_TOOLS
        }

        return base_tools

    def _execute_delegate_agent(
        self,
        arguments: dict[str, object],
    ) -> ToolExecutionResult:
        tool_name = 'Agent'
        agent_def = self._resolve_agent_definition(arguments)
        max_turns = arguments.get('max_turns')
        if max_turns is not None and (isinstance(max_turns, bool) or not isinstance(max_turns, int) or max_turns < 1):
            return ToolExecutionResult(
                name=tool_name,
                ok=False,
                content='max_turns must be an integer >= 1',
            )
        subtasks = self._normalize_delegate_subtasks(arguments)
        if not subtasks:
            return ToolExecutionResult(
                name=tool_name,
                ok=False,
                content='prompt must be a non-empty string or subtasks must contain at least one prompt',
            )

        # Resolve child permissions — read-only agents get no write/shell
        if agent_def.disallowed_tools and (
            'edit_file' in agent_def.disallowed_tools
            or 'write_file' in agent_def.disallowed_tools
        ):
            # Read-only agent (Explore, Plan, verification)
            child_permissions = AgentPermissions(
                allow_file_write=False,
                allow_shell_commands=self.runtime_config.permissions.allow_shell_commands,
                allow_destructive_shell_commands=False,
            )
        else:
            child_permissions = AgentPermissions(
                allow_file_write=(
                    self.runtime_config.permissions.allow_file_write
                    and bool(arguments.get('allow_write', False))
                ),
                allow_shell_commands=(
                    self.runtime_config.permissions.allow_shell_commands
                    and bool(arguments.get('allow_shell', False))
                ),
                allow_destructive_shell_commands=False,
            )

        # Resolve max_turns — agent definition or explicit param
        effective_max_turns = max_turns or agent_def.max_turns or min(self.runtime_config.max_turns, 6)

        child_runtime_config = replace(
            self.runtime_config,
            max_turns=effective_max_turns,
            permissions=child_permissions,
            auto_compact_threshold_tokens=self.runtime_config.auto_compact_threshold_tokens,
        )

        child_model_config = self._resolve_child_model_config(arguments, agent_def)
        child_tools = self._filter_tools_for_agent(agent_def)
        include_parent_context = bool(arguments.get('include_parent_context', True))
        continue_on_error = bool(arguments.get('continue_on_error', True))
        max_failures = arguments.get('max_failures')
        if isinstance(max_failures, bool) or (max_failures is not None and not isinstance(max_failures, int)):
            max_failures = None
        if isinstance(max_failures, int) and max_failures < 0:
            max_failures = None
        strategy = self._normalize_delegate_strategy(arguments.get('strategy'))
        child_summaries: list[dict[str, object]] = []
        child_session_ids: list[str] = []
        prior_results: list[dict[str, str]] = []
        completed_labels: set[str] = set()
        failed_labels: set[str] = set()
        delegate_preflight_messages = (
            self.plugin_runtime.before_delegate_injections()
            if self.plugin_runtime is not None
            else ()
        )
        delegate_after_messages: tuple[str, ...] = ()
        group_id: str | None = None
        if self.agent_manager is not None and len(subtasks) > 1:
            group_id = self.agent_manager.start_group(
                label=str(arguments.get('label') or 'delegated_group'),
                parent_agent_id=self.managed_agent_id,
                strategy=strategy,
            )
        planned_batches = self._plan_delegate_batches(subtasks, strategy)
        batch_summaries: list[dict[str, object]] = []
        failed_children = 0
        dependency_skips = 0
        child_result = None
        stop_processing = False
        for batch_index, batch in enumerate(planned_batches, start=1):
            if stop_processing:
                break
            batch_completed = 0
            batch_failed = 0
            batch_skipped = 0
            batch_labels: list[str] = []
            for subtask in batch:
                index = int(subtask.get('_delegate_index', len(child_summaries) + 1))
                subtask_label = str(subtask.get('label') or f'subtask_{index}')
                batch_labels.append(subtask_label)
                dependencies = tuple(
                    item
                    for item in subtask.get('depends_on', ())
                    if isinstance(item, str) and item
                )
                unmet_dependencies = [
                    dependency
                    for dependency in dependencies
                    if dependency not in completed_labels
                ]
                blocked_dependencies = [
                    dependency
                    for dependency in dependencies
                    if dependency in failed_labels
                ]
                if unmet_dependencies:
                    skip_reason = (
                        'skipped_dependency'
                        if blocked_dependencies
                        else 'pending_dependency'
                    )
                    child_result = AgentRunResult(
                        final_output=(
                            'Skipped delegated subtask because dependencies were not satisfied: '
                            + ', '.join(unmet_dependencies)
                        ),
                        turns=0,
                        tool_calls=0,
                        transcript=(),
                        stop_reason=skip_reason,
                    )
                    summary = {
                        'index': index,
                        'label': subtask_label,
                        'session_id': '',
                        'turns': child_result.turns,
                        'tool_calls': child_result.tool_calls,
                        'stop_reason': skip_reason,
                        'output_preview': self._preview_text(child_result.final_output, 220),
                        'resume_used': False,
                        'resumed_from_session_id': '',
                        'depends_on': list(dependencies),
                        'batch_index': batch_index,
                    }
                    child_summaries.append(summary)
                    failed_children += 1
                    batch_failed += 1
                    batch_skipped += 1
                    dependency_skips += 1
                    failed_labels.add(subtask_label)
                    if isinstance(max_failures, int) and failed_children > max_failures:
                        stop_processing = True
                        break
                    if not continue_on_error:
                        stop_processing = True
                        break
                    continue
                # Use agent definition's system prompt if available
                child_system_prompt = agent_def.system_prompt or self.custom_system_prompt
                child_override_prompt = None
                if agent_def.system_prompt:
                    child_override_prompt = agent_def.system_prompt
                else:
                    child_override_prompt = self.override_system_prompt

                # Inject critical system reminder if agent definition has one
                child_append_prompt = self.append_system_prompt
                if agent_def.critical_system_reminder:
                    reminder = f'\n\n<system-reminder>\n{agent_def.critical_system_reminder}\n</system-reminder>'
                    child_append_prompt = (
                        (child_append_prompt or '') + reminder
                    )

                child_agent = LocalCodingAgent(
                    model_config=child_model_config,
                    runtime_config=replace(
                        child_runtime_config,
                        max_turns=subtask.get('max_turns', child_runtime_config.max_turns),
                        disable_claude_md_discovery=agent_def.omit_claude_md,
                    ),
                    custom_system_prompt=child_system_prompt if not child_override_prompt else None,
                    append_system_prompt=child_append_prompt,
                    override_system_prompt=child_override_prompt,
                    tool_registry=child_tools,
                    agent_manager=self.agent_manager,
                    parent_agent_id=self.managed_agent_id,
                    managed_group_id=group_id,
                    managed_child_index=index,
                    managed_label=subtask_label,
                )
                if group_id is not None and child_agent.managed_agent_id is not None:
                    self.agent_manager.register_group_child(
                        group_id,
                        child_agent.managed_agent_id,
                        child_index=index,
                    )
                resume_session_id = subtask.get('resume_session_id')
                child_prompt = str(subtask['prompt'])
                if agent_def.initial_prompt and not (
                    isinstance(resume_session_id, str) and resume_session_id
                ):
                    child_prompt = f'{agent_def.initial_prompt.strip()}\n\n{child_prompt}'.strip()
                if delegate_preflight_messages:
                    child_prompt = self._prepend_plugin_delegate_context(
                        child_prompt,
                        delegate_preflight_messages,
                    )
                if include_parent_context and prior_results:
                    child_prompt = self._prepend_delegate_context(child_prompt, prior_results)
                resume_used = False
                if isinstance(resume_session_id, str) and resume_session_id:
                    try:
                        stored_child_session = load_agent_session(
                            resume_session_id,
                            directory=child_runtime_config.session_directory,
                        )
                    except OSError:
                        child_result = AgentRunResult(
                            final_output=f'Unable to load delegated session {resume_session_id}.',
                            turns=0,
                            tool_calls=0,
                            transcript=(),
                            stop_reason='resume_load_error',
                            session_id=resume_session_id,
                        )
                        failed_children += 1
                        batch_failed += 1
                        summary = {
                            'index': index,
                            'label': subtask_label,
                            'session_id': resume_session_id,
                            'turns': child_result.turns,
                            'tool_calls': child_result.tool_calls,
                            'stop_reason': child_result.stop_reason or 'resume_load_error',
                            'output_preview': self._preview_text(child_result.final_output, 220),
                            'resume_used': True,
                            'resumed_from_session_id': resume_session_id,
                            'depends_on': list(dependencies),
                            'batch_index': batch_index,
                        }
                        child_summaries.append(summary)
                        prior_results.append(
                            {
                                'label': summary['label'],
                                'output_preview': str(summary['output_preview']),
                            }
                        )
                        failed_labels.add(subtask_label)
                        if isinstance(max_failures, int) and failed_children > max_failures:
                            stop_processing = True
                            break
                        if not continue_on_error:
                            stop_processing = True
                            break
                        continue
                    child_result = child_agent.resume(child_prompt, stored_child_session)
                    resume_used = True
                else:
                    child_result = child_agent.run(child_prompt)
                if group_id is not None and child_agent.managed_agent_id is not None:
                    self.agent_manager.register_group_child(
                        group_id,
                        child_agent.managed_agent_id,
                        child_index=index,
                    )
                summary = {
                    'index': index,
                    'label': subtask_label,
                    'session_id': child_result.session_id or '',
                    'turns': child_result.turns,
                    'tool_calls': child_result.tool_calls,
                    'stop_reason': child_result.stop_reason or 'stop',
                    'output_preview': self._preview_text(child_result.final_output, 220),
                    'resume_used': resume_used,
                    'resumed_from_session_id': (
                        str(resume_session_id)
                        if isinstance(resume_session_id, str) and resume_session_id
                        else ''
                    ),
                    'depends_on': list(dependencies),
                    'batch_index': batch_index,
                }
                child_summaries.append(summary)
                if child_result.session_id:
                    child_session_ids.append(child_result.session_id)
                prior_results.append(
                    {
                        'label': summary['label'],
                        'output_preview': str(summary['output_preview']),
                    }
                )
                if child_result.stop_reason in {'backend_error', 'budget_exceeded'}:
                    failed_children += 1
                    batch_failed += 1
                    failed_labels.add(subtask_label)
                    if isinstance(max_failures, int) and failed_children > max_failures:
                        stop_processing = True
                        break
                    if not continue_on_error:
                        stop_processing = True
                        break
                else:
                    batch_completed += 1
                    completed_labels.add(subtask_label)
            batch_status = 'completed'
            if batch_failed and batch_completed:
                batch_status = 'partial'
            elif batch_failed:
                batch_status = 'failed'
            batch_summaries.append(
                {
                    'batch_index': batch_index,
                    'labels': batch_labels,
                    'completed_children': batch_completed,
                    'failed_children': batch_failed,
                    'skipped_children': batch_skipped,
                    'status': batch_status,
                }
            )
        assert child_result is not None
        completed_children = len(child_summaries) - failed_children
        resumed_children = sum(
            1 for summary in child_summaries if summary.get('resume_used')
        )
        group_status = 'completed'
        if failed_children and completed_children:
            group_status = 'partial'
        elif failed_children:
            group_status = 'failed'
        delegate_after_messages = (
            self.plugin_runtime.after_delegate_injections()
            if self.plugin_runtime is not None
            else ()
        )
        if group_id is not None and self.agent_manager is not None:
            self.agent_manager.finish_group(
                group_id,
                status=group_status,
                completed_children=completed_children,
                failed_children=failed_children,
                batch_count=len(batch_summaries),
                max_batch_size=max((len(batch['labels']) for batch in batch_summaries), default=0),
                dependency_skips=dependency_skips,
            )
        summary_lines = [
            (
                'Delegated agent completed the subtask.'
                if len(child_summaries) == 1
                else f'Delegated agent completed {len(child_summaries)} sequential subtasks.'
            ),
        ]
        if group_id is not None:
            summary_lines.append(f'group_id={group_id}')
            summary_lines.append(f'group_status={group_status}')
            summary_lines.append(f'resumed_children={resumed_children}')
            summary_lines.append(f'strategy={strategy}')
            summary_lines.append(f'batch_count={len(batch_summaries)}')
            summary_lines.append(f'dependency_skips={dependency_skips}')
            summary_lines.append('')
        if delegate_preflight_messages:
            summary_lines.append('Plugin delegate preflight:')
            summary_lines.extend(f'- {message}' for message in delegate_preflight_messages)
            summary_lines.append('')
        for batch in batch_summaries:
            summary_lines.append(
                f"[batch {batch['batch_index']}] status={batch['status']} "
                f"labels={','.join(batch['labels']) or '(none)'} "
                f"completed={batch['completed_children']} failed={batch['failed_children']} "
                f"skipped={batch['skipped_children']}"
            )
        if batch_summaries:
            summary_lines.append('')
        for summary in child_summaries:
            summary_lines.extend(
                [
                    f"[{summary['label']}]",
                    f"batch_index={summary['batch_index']}",
                    f"session_id={summary['session_id']}",
                    f"turns={summary['turns']}",
                    f"tool_calls={summary['tool_calls']}",
                    f"stop_reason={summary['stop_reason']}",
                    f"resume_used={summary['resume_used']}",
                    f"resumed_from_session_id={summary['resumed_from_session_id']}",
                    f"depends_on={','.join(summary.get('depends_on', [])) or '(none)'}",
                    f"output_preview={summary['output_preview']}",
                    '',
                ]
            )
        if delegate_after_messages:
            summary_lines.append('Plugin delegate completion:')
            summary_lines.extend(f'- {message}' for message in delegate_after_messages)
            summary_lines.append('')
        summary_lines.append('Final delegated output:')
        summary_lines.append(child_result.final_output)
        return ToolExecutionResult(
            name=tool_name,
            ok=True,
            content='\n'.join(summary_lines).strip(),
            metadata={
                # Canonical name regardless of the alias the model invoked
                # the tool with ('Agent' or 'delegate_agent'). Downstream
                # consumers (file_history, observability) expect the stable
                # historical identifier. Tests pin this contract.
                'action': 'delegate_agent',
                'subagent_type': agent_def.agent_type,
                'child_session_id': child_result.session_id,
                'child_session_ids': child_session_ids,
                'child_turns': child_result.turns,
                'child_tool_calls': child_result.tool_calls,
                'child_stop_reason': child_result.stop_reason,
                'child_results': child_summaries,
                'subtask_count': len(child_summaries),
                'group_id': group_id,
                'group_status': group_status,
                'failed_children': failed_children,
                'completed_children': completed_children,
                'resumed_children': resumed_children,
                'strategy': strategy,
                'max_failures': max_failures,
                'delegate_batches': batch_summaries,
                'dependency_skips': dependency_skips,
                'plugin_delegate_preflight_messages': list(delegate_preflight_messages),
                'plugin_delegate_after_messages': list(delegate_after_messages),
            },
        )

    def _normalize_delegate_subtasks(
        self,
        arguments: dict[str, object],
    ) -> list[dict[str, object]]:
        subtasks: list[dict[str, object]] = []
        raw_subtasks = arguments.get('subtasks')
        if isinstance(raw_subtasks, list):
            for index, item in enumerate(raw_subtasks, start=1):
                if isinstance(item, str) and item.strip():
                    subtasks.append(
                        {
                            'prompt': item.strip(),
                            'label': f'subtask_{index}',
                            '_delegate_index': index,
                        }
                    )
                    continue
                if isinstance(item, dict):
                    prompt = item.get('prompt')
                    if not isinstance(prompt, str) or not prompt.strip():
                        continue
                    label = item.get('label')
                    max_turns = item.get('max_turns')
                    task: dict[str, object] = {
                        'prompt': prompt.strip(),
                        'label': label if isinstance(label, str) and label.strip() else f'subtask_{index}',
                    }
                    resume_session_id = item.get('resume_session_id')
                    if resume_session_id is None:
                        resume_session_id = item.get('session_id')
                    if isinstance(resume_session_id, str) and resume_session_id.strip():
                        task['resume_session_id'] = resume_session_id.strip()
                    depends_on = item.get('depends_on')
                    if isinstance(depends_on, list):
                        task['depends_on'] = tuple(
                            dependency.strip()
                            for dependency in depends_on
                            if isinstance(dependency, str) and dependency.strip()
                        )
                    if isinstance(max_turns, int) and not isinstance(max_turns, bool) and max_turns > 0:
                        task['max_turns'] = max_turns
                    task['_delegate_index'] = index
                    subtasks.append(task)
        prompt = arguments.get('prompt')
        if isinstance(prompt, str) and prompt.strip():
            if not subtasks:
                task: dict[str, object] = {'prompt': prompt.strip(), 'label': 'subtask_1'}
                resume_session_id = arguments.get('resume_session_id')
                if resume_session_id is None:
                    resume_session_id = arguments.get('session_id')
                if isinstance(resume_session_id, str) and resume_session_id.strip():
                    task['resume_session_id'] = resume_session_id.strip()
                task['_delegate_index'] = 1
                subtasks.append(task)
        return [
            {
                **task,
                '_delegate_index': int(task.get('_delegate_index', index)),
            }
            for index, task in enumerate(subtasks[:8], start=1)
        ]

    def _normalize_delegate_strategy(self, strategy: object) -> str:
        if not isinstance(strategy, str) or not strategy.strip():
            return 'serial'
        normalized = strategy.strip().lower().replace('-', '_')
        if normalized in {'graph', 'topological', 'dependency_graph', 'parallel', 'parallel_batches'}:
            return 'topological'
        return 'serial'

    def _plan_delegate_batches(
        self,
        subtasks: list[dict[str, object]],
        strategy: str,
    ) -> list[list[dict[str, object]]]:
        if strategy != 'topological':
            return [subtasks]
        remaining = list(subtasks)
        scheduled_labels: set[str] = set()
        known_labels = {
            str(task.get('label'))
            for task in subtasks
            if isinstance(task.get('label'), str) and str(task.get('label')).strip()
        }
        batches: list[list[dict[str, object]]] = []
        while remaining:
            ready: list[dict[str, object]] = []
            blocked: list[dict[str, object]] = []
            for task in remaining:
                dependencies = tuple(
                    item
                    for item in task.get('depends_on', ())
                    if isinstance(item, str) and item
                )
                if any(dependency not in known_labels for dependency in dependencies):
                    blocked.append(task)
                    continue
                if all(dependency in scheduled_labels for dependency in dependencies):
                    ready.append(task)
                else:
                    blocked.append(task)
            if not ready:
                batches.append(blocked)
                break
            batches.append(
                sorted(
                    ready,
                    key=lambda task: int(task.get('_delegate_index', 0)),
                )
            )
            scheduled_labels.update(
                str(task.get('label'))
                for task in ready
                if isinstance(task.get('label'), str) and str(task.get('label')).strip()
            )
            remaining = blocked
        return batches

    def _delegated_task_units(
        self,
        arguments: dict[str, object],
    ) -> int:
        subtasks = arguments.get('subtasks')
        if isinstance(subtasks, list):
            count = sum(
                1
                for item in subtasks
                if (
                    isinstance(item, str)
                    and item.strip()
                ) or (
                    isinstance(item, dict)
                    and isinstance(item.get('prompt'), str)
                    and item.get('prompt', '').strip()
                )
            )
            if count:
                return count
        return 1

    def _prepend_delegate_context(
        self,
        prompt: str,
        prior_results: list[dict[str, str]],
    ) -> str:
        lines = [
            '<system-reminder>',
            'Prior delegated subtask summaries:',
        ]
        for result in prior_results[-4:]:
            lines.append(f"- {result['label']}: {result['output_preview']}")
        lines.extend(['</system-reminder>', '', prompt])
        return '\n'.join(lines)

    def _prepend_plugin_delegate_context(
        self,
        prompt: str,
        messages: tuple[str, ...],
    ) -> str:
        if not messages:
            return prompt
        lines = [
            '<system-reminder>',
            'Plugin delegate guidance:',
        ]
        lines.extend(f'- {message}' for message in messages)
        lines.extend(['</system-reminder>', '', prompt])
        return '\n'.join(lines)

    def _append_runtime_tool_followup_events(
        self,
        stream_events: list[dict[str, object]],
        *,
        tool_call: ToolCall,
        tool_result: ToolExecutionResult,
    ) -> None:
        metadata = tool_result.metadata
        if metadata.get('action') == 'plugin_virtual_tool':
            stream_events.append(
                {
                    'type': 'plugin_virtual_tool_result',
                    'tool_call_id': tool_call.id,
                    'tool_name': tool_call.name,
                    'plugin_name': metadata.get('plugin_name'),
                    'virtual_tool': metadata.get('virtual_tool'),
                }
            )
        plugin_delegate_preflight = metadata.get('plugin_delegate_preflight_messages')
        if isinstance(plugin_delegate_preflight, list) and plugin_delegate_preflight:
            stream_events.append(
                {
                    'type': 'plugin_delegate_preflight',
                    'tool_call_id': tool_call.id,
                    'tool_name': tool_call.name,
                    'message_count': len(plugin_delegate_preflight),
                }
            )
        plugin_delegate_after = metadata.get('plugin_delegate_after_messages')
        if isinstance(plugin_delegate_after, list) and plugin_delegate_after:
            stream_events.append(
                {
                    'type': 'plugin_delegate_after',
                    'tool_call_id': tool_call.id,
                    'tool_name': tool_call.name,
                    'message_count': len(plugin_delegate_after),
                }
            )
        if tool_call.name not in ('Agent', 'delegate_agent'):
            return
        delegate_batches = metadata.get('delegate_batches')
        if isinstance(delegate_batches, list):
            for batch in delegate_batches:
                if not isinstance(batch, dict):
                    continue
                stream_events.append(
                    {
                        'type': 'delegate_batch_result',
                        'tool_call_id': tool_call.id,
                        'group_id': metadata.get('group_id'),
                        'batch_index': batch.get('batch_index'),
                        'status': batch.get('status'),
                        'labels': batch.get('labels'),
                        'completed_children': batch.get('completed_children'),
                        'failed_children': batch.get('failed_children'),
                        'skipped_children': batch.get('skipped_children'),
                    }
                )
        child_results = metadata.get('child_results')
        if isinstance(child_results, list):
            for child in child_results:
                if not isinstance(child, dict):
                    continue
                stream_events.append(
                    {
                        'type': 'delegate_subtask_result',
                        'tool_call_id': tool_call.id,
                        'group_id': metadata.get('group_id'),
                        'label': child.get('label'),
                        'index': child.get('index'),
                        'session_id': child.get('session_id'),
                        'stop_reason': child.get('stop_reason'),
                        'tool_calls': child.get('tool_calls'),
                        'turns': child.get('turns'),
                        'resume_used': child.get('resume_used'),
                        'resumed_from_session_id': child.get('resumed_from_session_id'),
                        'depends_on': child.get('depends_on'),
                        'batch_index': child.get('batch_index'),
                    }
                )
        if metadata.get('group_id') is not None:
            stream_events.append(
                {
                    'type': 'delegate_group_result',
                    'tool_call_id': tool_call.id,
                    'group_id': metadata.get('group_id'),
                    'group_status': metadata.get('group_status'),
                    'subtask_count': metadata.get('subtask_count'),
                    'completed_children': metadata.get('completed_children'),
                    'failed_children': metadata.get('failed_children'),
                    'resumed_children': metadata.get('resumed_children'),
                    'strategy': metadata.get('strategy'),
                    'max_failures': metadata.get('max_failures'),
                    'batch_count': len(delegate_batches) if isinstance(delegate_batches, list) else 0,
                    'dependency_skips': metadata.get('dependency_skips'),
                }
            )

    def _preview_text(self, text: str, limit: int) -> str:
        normalized = ' '.join(text.split())
        if len(normalized) <= limit:
            return normalized
        return normalized[: limit - 3] + '...'

    def _ensure_scratchpad_directory(self, session_id: str) -> Path:
        scratchpad_directory = (self.runtime_config.scratchpad_root / session_id).resolve()
        scratchpad_directory.mkdir(parents=True, exist_ok=True)
        return scratchpad_directory

    def _append_file_history_replay_if_needed(
        self,
        session: AgentSessionState,
        file_history: tuple[dict[str, object], ...],
    ) -> None:
        if not file_history:
            return
        replay_count = len(file_history)
        unique_paths = sorted(
            {
                path
                for entry in file_history
                for path in (
                    entry.get('changed_paths')
                    if isinstance(entry.get('changed_paths'), list)
                    else ([entry.get('path')] if isinstance(entry.get('path'), str) else [])
                )
                if isinstance(path, str) and path
            }
        )
        snapshot_count = sum(
            1
            for entry in file_history
            for key in ('before_snapshot_id', 'after_snapshot_id')
            if isinstance(entry.get(key), str) and entry.get(key)
        )
        for message in reversed(session.messages):
            if message.metadata.get('kind') != 'file_history_replay':
                continue
            if message.metadata.get('file_history_count') == replay_count:
                return
            break
        session.append_user(
            self._render_file_history_replay(file_history),
            metadata={
                'kind': 'file_history_replay',
                'file_history_count': replay_count,
                'file_history_unique_paths': len(unique_paths),
                'file_history_snapshot_count': snapshot_count,
            },
            message_id=f'file_history_replay_{replay_count}',
        )

    def _render_file_history_replay(
        self,
        file_history: tuple[dict[str, object], ...],
    ) -> str:
        unique_paths = sorted(
            {
                path
                for entry in file_history
                for path in (
                    entry.get('changed_paths')
                    if isinstance(entry.get('changed_paths'), list)
                    else ([entry.get('path')] if isinstance(entry.get('path'), str) else [])
                )
                if isinstance(path, str) and path
            }
        )
        snapshot_count = sum(
            1
            for entry in file_history
            for key in ('before_snapshot_id', 'after_snapshot_id')
            if isinstance(entry.get(key), str) and entry.get(key)
        )
        lines = [
            '<system-reminder>',
            'Recent file history from this saved session:',
            f'- History entries: {len(file_history)}',
            f'- Unique changed paths: {len(unique_paths)}',
            f'- Snapshot ids: {snapshot_count}',
        ]
        if unique_paths:
            preview_paths = ', '.join(unique_paths[:4])
            if len(unique_paths) > 4:
                preview_paths += f', ... (+{len(unique_paths) - 4} more)'
            lines.append(f'- Changed path preview: {preview_paths}')
        for entry in file_history[-10:]:
            action = str(entry.get('action', entry.get('tool_name', 'tool')))
            turn = entry.get('turn_index')
            path = entry.get('path')
            command = entry.get('command')
            details = [f'action={action}']
            history_entry_id = entry.get('history_entry_id')
            if isinstance(history_entry_id, str) and history_entry_id:
                details.append(f'entry_id={history_entry_id}')
            if turn is not None:
                details.append(f'turn={turn}')
            if path:
                details.append(f'path={path}')
            if command:
                details.append(f'command={command}')
            child_session_ids = entry.get('child_session_ids')
            if isinstance(child_session_ids, list) and child_session_ids:
                details.append(f'child_sessions={len(child_session_ids)}')
            delegate_batch_count = entry.get('delegate_batch_count')
            if isinstance(delegate_batch_count, int) and not isinstance(delegate_batch_count, bool):
                details.append(f'batches={delegate_batch_count}')
            dependency_skips = entry.get('dependency_skips')
            if isinstance(dependency_skips, int) and not isinstance(dependency_skips, bool):
                details.append(f'dependency_skips={dependency_skips}')
            lines.append(f"- {'; '.join(details)}")
            before_snapshot_id = entry.get('before_snapshot_id')
            if isinstance(before_snapshot_id, str) and before_snapshot_id:
                lines.append(f'  before_snapshot: {before_snapshot_id}')
            after_snapshot_id = entry.get('after_snapshot_id')
            if isinstance(after_snapshot_id, str) and after_snapshot_id:
                lines.append(f'  after_snapshot: {after_snapshot_id}')
            before_preview = entry.get('before_preview')
            if isinstance(before_preview, str) and before_preview:
                lines.append(f'  before: {before_preview}')
            after_preview = entry.get('after_preview')
            if isinstance(after_preview, str) and after_preview:
                lines.append(f'  after: {after_preview}')
            result_preview = entry.get('result_preview')
            if isinstance(result_preview, str) and result_preview:
                lines.append(f'  result: {result_preview}')
        if len(file_history) > 10:
            lines.append(f'- ... plus {len(file_history) - 10} older file-history entries')
        lines.extend(
            [
                '',
                'Use this replayed history when continuing the task so you avoid repeating prior edits or commands.',
                '</system-reminder>',
            ]
        )
        return '\n'.join(lines)

    def _append_compaction_replay_if_needed(
        self,
        session: AgentSessionState,
    ) -> None:
        compact_messages = [
            message for message in session.messages
            if message.metadata.get('kind') == 'compact_boundary'
        ]
        snipped_messages = [
            message for message in session.messages
            if message.metadata.get('kind') == 'snipped_message'
        ]
        if not compact_messages and not snipped_messages:
            return
        for message in reversed(session.messages):
            if message.metadata.get('kind') != 'compaction_replay':
                continue
            return
        session.append_user(
            self._render_compaction_replay(compact_messages, snipped_messages),
            metadata={
                'kind': 'compaction_replay',
                'compact_boundary_count': len(compact_messages),
                'snipped_message_count': len(snipped_messages),
            },
            message_id=(
                f'compaction_replay_{len(compact_messages)}_{len(snipped_messages)}'
            ),
        )

    def _render_compaction_replay(
        self,
        compact_messages,
        snipped_messages,
    ) -> str:
        lines = [
            '<system-reminder>',
            'This resumed session already contains compacted or snipped history.',
            f'- Compact boundaries: {len(compact_messages)}',
            f'- Snipped/tombstoned messages: {len(snipped_messages)}',
        ]
        latest_boundary = compact_messages[-1] if compact_messages else None
        if latest_boundary is not None:
            lines.append(
                f"- Latest compact boundary id: {latest_boundary.message_id or '(none)'}"
            )
            depth = latest_boundary.metadata.get('compaction_depth')
            if isinstance(depth, int) and not isinstance(depth, bool):
                lines.append(f'- Latest compaction depth: {depth}')
            compacted_lineages = latest_boundary.metadata.get('compacted_lineage_ids')
            if isinstance(compacted_lineages, list) and compacted_lineages:
                lines.append(f'- Latest compacted lineages: {len(compacted_lineages)}')
            max_source_mutation_serial = latest_boundary.metadata.get('max_source_mutation_serial')
            if (
                isinstance(max_source_mutation_serial, int)
                and not isinstance(max_source_mutation_serial, bool)
                and max_source_mutation_serial > 0
            ):
                lines.append(
                    f'- Latest source mutation serial: {max_source_mutation_serial}'
                )
            source_mutation_totals = latest_boundary.metadata.get('source_mutation_totals')
            if isinstance(source_mutation_totals, dict) and source_mutation_totals:
                rendered = ', '.join(
                    f'{name}:{count}'
                    for name, count in sorted(source_mutation_totals.items())
                    if isinstance(name, str)
                    and name
                    and isinstance(count, int)
                    and not isinstance(count, bool)
                    and count > 0
                )
                if rendered:
                    lines.append(f'- Latest compacted mutations: {rendered}')
            preserved_tail = latest_boundary.metadata.get('preserved_tail_ids')
            if isinstance(preserved_tail, list) and preserved_tail:
                lines.append(
                    '- Latest preserved tail ids: '
                    + ', '.join(str(item) for item in preserved_tail[:4])
                )
        if snipped_messages:
            last_ids = [
                message.message_id or '(none)'
                for message in snipped_messages[-3:]
            ]
            lines.append(f"- Recent snipped ids: {', '.join(last_ids)}")
            snipped_lineages = [
                str(message.metadata.get('snipped_from_lineage_id'))
                for message in snipped_messages[-3:]
                if isinstance(message.metadata.get('snipped_from_lineage_id'), str)
            ]
            if snipped_lineages:
                lines.append(f"- Recent snipped lineages: {', '.join(snipped_lineages)}")
        lines.extend(
            [
                '',
                'Use the surviving transcript plus the compacted summaries as the authoritative context when continuing.',
                '</system-reminder>',
            ]
        )
        return '\n'.join(lines)

    def _apply_hook_policy_before_prompt_hooks(self, prompt: str) -> str:
        if self.hook_policy_runtime is None or not self.hook_policy_runtime.manifests:
            return prompt
        injections = self.hook_policy_runtime.before_prompt_messages()
        managed_settings = self.hook_policy_runtime.managed_settings()
        safe_env = self.hook_policy_runtime.safe_env()
        trusted = self.hook_policy_runtime.is_trusted()
        if not injections and not managed_settings and not safe_env and trusted:
            return prompt
        lines = ['<system-reminder>', 'Workspace hook/policy guidance:']
        lines.append(
            f'- Trust mode: {"trusted" if trusted else "untrusted"}'
        )
        if not trusted:
            lines.append(
                '- Untrusted workspaces should favor inspection-first behavior. '
                'Avoid unnecessary writes or shell actions unless the task clearly requires them.'
            )
        for entry in injections:
            lines.append(f'- Before prompt: {entry}')
        if managed_settings:
            lines.append(
                '- Managed settings: '
                + ', '.join(f'{key}={value}' for key, value in sorted(managed_settings.items()))
            )
        if safe_env:
            lines.append(
                '- Safe environment values loaded for tools: '
                + ', '.join(sorted(safe_env))
            )
        lines.extend(['</system-reminder>', '', prompt])
        return '\n'.join(lines)

    def _build_plugin_tool_runtime_message(
        self,
        *,
        tool_name: str,
        preflight_messages: tuple[str, ...],
        block_message: str | None,
        plugin_messages: tuple[str, ...],
        hook_policy_preflight_messages: tuple[str, ...] = (),
        hook_policy_block_message: str | None = None,
        hook_policy_messages: tuple[str, ...] = (),
        delegate_preflight_messages: tuple[str, ...] = (),
        delegate_after_messages: tuple[str, ...] = (),
    ) -> str | None:
        if (
            block_message is None
            and not plugin_messages
            and not preflight_messages
            and hook_policy_block_message is None
            and not hook_policy_preflight_messages
            and not hook_policy_messages
            and not delegate_preflight_messages
            and not delegate_after_messages
        ):
            return None
        plugin_only = (
            hook_policy_block_message is None
            and not hook_policy_preflight_messages
            and not hook_policy_messages
        )
        lines = [
            '<system-reminder>',
            (
                f'Plugin tool runtime guidance for `{tool_name}`:'
                if plugin_only
                else f'Runtime tool guidance for `{tool_name}`:'
            ),
        ]
        for message in preflight_messages:
            lines.append(f'- Before tool: {message}')
        for message in hook_policy_preflight_messages:
            lines.append(f'- Hook/policy before tool: {message}')
        for message in delegate_preflight_messages:
            lines.append(f'- Before delegate: {message}')
        if block_message is not None:
            lines.append(f'- Blocked: {block_message}')
        if hook_policy_block_message is not None:
            lines.append(f'- Hook/policy blocked: {hook_policy_block_message}')
        for message in plugin_messages:
            lines.append(f'- After result: {message}')
        for message in hook_policy_messages:
            lines.append(f'- Hook/policy after result: {message}')
        for message in delegate_after_messages:
            lines.append(f'- After delegate: {message}')
        lines.extend(
            [
                '',
                'Use this runtime guidance when deciding the next tool call or assistant response.',
                '</system-reminder>',
            ]
        )
        return '\n'.join(lines)

    def _plugin_tool_preflight_messages(self, tool_name: str) -> tuple[str, ...]:
        if self.plugin_runtime is None:
            return ()
        return self.plugin_runtime.tool_preflight_injections(tool_name)

    def _plugin_block_message(self, tool_name: str) -> str | None:
        if self.plugin_runtime is None:
            return None
        return self.plugin_runtime.blocked_tool_message(tool_name)

    def _plugin_tool_result_messages(self, tool_name: str) -> tuple[str, ...]:
        if self.plugin_runtime is None:
            return ()
        return self.plugin_runtime.tool_result_injections(tool_name)

    def _hook_policy_tool_preflight_messages(self, tool_name: str) -> tuple[str, ...]:
        if self.hook_policy_runtime is None:
            return ()
        return self.hook_policy_runtime.before_tool_messages(tool_name)

    def _hook_policy_block_message(self, tool_name: str) -> str | None:
        if self.hook_policy_runtime is None:
            return None
        return self.hook_policy_runtime.denied_tool_message(tool_name)

    def _hook_policy_tool_result_messages(self, tool_name: str) -> tuple[str, ...]:
        if self.hook_policy_runtime is None:
            return ()
        return self.hook_policy_runtime.after_tool_messages(tool_name)

    def _persist_session(
        self,
        session: AgentSessionState,
        result: AgentRunResult,
    ) -> AgentRunResult:
        if result.session_id is None:
            return result
        persist_events = list(result.events)
        if self.plugin_runtime is not None:
            persist_messages = self.plugin_runtime.before_persist_injections()
            if persist_messages:
                session.append_user(
                    self._render_plugin_persist_message(persist_messages),
                    metadata={
                        'kind': 'plugin_persist',
                        'message_count': len(persist_messages),
                    },
                    message_id=f'plugin_persist_{result.session_id}',
                )
                persist_events.append(
                    {
                        'type': 'plugin_before_persist',
                        'session_id': result.session_id,
                        'message_count': len(persist_messages),
                    }
                )
        previous_turns = 0
        previous_tool_calls = 0
        previous_budget_state: dict[str, object] = {}
        existing_path = self.runtime_config.session_directory / f'{result.session_id}.json'
        if existing_path.exists():
            try:
                previous = load_agent_session(
                    result.session_id,
                    directory=self.runtime_config.session_directory,
                )
            except OSError:
                previous = None
            if previous is not None:
                previous_turns = previous.turns
                previous_tool_calls = previous.tool_calls
                if isinstance(previous.budget_state, dict):
                    previous_budget_state = dict(previous.budget_state)
        budget_state = {
            'model_calls': int(previous_budget_state.get('model_calls', 0))
            + max(result.turns, 0),
            'session_turns': previous_turns + result.turns,
            'tool_calls': previous_tool_calls + result.tool_calls,
            'delegated_tasks': sum(
                1 for entry in result.file_history if entry.get('action') in ('delegate_agent', 'Agent')
            ),
        }
        stored = StoredAgentSession(
            session_id=result.session_id,
            model_config=serialize_model_config(self.model_config),
            runtime_config=serialize_runtime_config(self.runtime_config),
            system_prompt_parts=session.system_prompt_parts,
            user_context=dict(session.user_context),
            system_context=dict(session.system_context),
            messages=session.transcript(),
            turns=previous_turns + result.turns,
            tool_calls=previous_tool_calls + result.tool_calls,
            usage=result.usage.to_dict(),
            total_cost_usd=result.total_cost_usd,
            file_history=result.file_history,
            budget_state=budget_state,
            plugin_state=(
                self.plugin_runtime.export_session_state()
                if self.plugin_runtime is not None
                else {}
            ),
            scratchpad_directory=result.scratchpad_directory,
        )
        path = save_agent_session(
            stored,
            directory=self.runtime_config.session_directory,
        )
        self.last_session_path = str(path)
        return replace(
            result,
            session_path=self.last_session_path,
            events=tuple(persist_events),
            transcript=session.transcript(),
        )

    def render_system_prompt(self) -> str:
        prompt_context = self.build_prompt_context()
        parts = self.build_system_prompt_parts(prompt_context)
        return render_system_prompt(parts)

    def render_context_report(self, prompt: str | None = None) -> str:
        session = self.last_session if prompt is None else None
        strategy = 'current Python session'
        if session is None:
            session = self.build_session(prompt)
            strategy = 'one-shot Python session preview'
        report = collect_context_usage(
            session=session,
            model=self.model_config.model,
            strategy=strategy,
        )
        return format_context_usage(report)

    def render_context_snapshot_report(self) -> str:
        prompt_context = self.build_prompt_context()
        return render_agent_context_report(prompt_context, self.model_config.model)

    def render_permissions_report(self) -> str:
        permissions = self.runtime_config.permissions
        lines = [
            '# Permissions',
            '',
            f'- File write tools: {"enabled" if permissions.allow_file_write else "disabled"}',
            f'- Shell commands: {"enabled" if permissions.allow_shell_commands else "disabled"}',
            f'- Destructive shell commands: {"enabled" if permissions.allow_destructive_shell_commands else "disabled"}',
        ]
        if self.hook_policy_runtime is not None and self.hook_policy_runtime.manifests:
            lines.append(
                f'- Workspace trust mode: {"trusted" if self.hook_policy_runtime.is_trusted() else "untrusted"}'
            )
            denied_tools = sorted(
                {
                    name
                    for manifest in self.hook_policy_runtime.manifests
                    for name in manifest.deny_tools
                }
            )
            if denied_tools:
                lines.append('- Policy-denied tools: ' + ', '.join(denied_tools))
        return '\n'.join(lines)

    def render_tools_report(self) -> str:
        permissions = self.runtime_config.permissions
        lines = ['# Tools', '']
        for tool in self.tool_registry.values():
            state = 'enabled'
            if tool.name == 'bash' and not permissions.allow_shell_commands:
                state = 'blocked by permissions'
            if tool.name in {'write_file', 'edit_file'} and not permissions.allow_file_write:
                state = 'blocked by permissions'
            if (
                self.hook_policy_runtime is not None
                and self.hook_policy_runtime.denied_tool_message(tool.name) is not None
            ):
                state = 'blocked by hook policy'
            lines.append(f'- `{tool.name}`: {tool.description} [{state}]')
        return '\n'.join(lines)

    def render_agents_report(self, *, show_all: bool = False) -> str:
        snapshot = self.load_agent_registry()
        return render_agents_report(
            snapshot,
            cwd=self.runtime_config.cwd,
            show_all=show_all,
        )

    def render_agent_detail_report(self, agent_type: str) -> str:
        snapshot = self.load_agent_registry()
        return render_agent_detail(snapshot, agent_type)

    def render_agent_create_report(
        self,
        agent_type: str,
        *,
        source: str = 'projectSettings',
        description: str | None = None,
        system_prompt: str | None = None,
        overwrite: bool = False,
    ) -> str:
        result = scaffold_agent_definition(
            self.runtime_config.cwd,
            agent_type=agent_type,
            source=normalize_mutable_source(source),
            overwrite=overwrite,
            description=description,
            system_prompt=system_prompt,
        )
        return render_agent_mutation(result)

    def render_agent_update_report(
        self,
        agent_type: str,
        *,
        source: str = 'auto',
        description: str | None = None,
        system_prompt: str | None = None,
    ) -> str:
        update_kwargs: dict[str, object] = {}
        if description is not None:
            update_kwargs['description'] = description
        if system_prompt is not None:
            update_kwargs['system_prompt'] = system_prompt
        result = update_agent_definition(
            self.runtime_config.cwd,
            agent_type=agent_type,
            source=normalize_mutable_source(source, allow_auto=True),
            **update_kwargs,
        )
        return render_agent_mutation(result)

    def render_agent_delete_report(
        self,
        agent_type: str,
        *,
        source: str = 'auto',
    ) -> str:
        result = delete_agent_definition(
            self.runtime_config.cwd,
            agent_type=agent_type,
            source=normalize_mutable_source(source, allow_auto=True),
        )
        return render_agent_mutation(result)

    def render_memory_report(self) -> str:
        prompt_context = self.build_prompt_context()
        claude_md = prompt_context.user_context.get('claudeMd')
        if not claude_md:
            return '# Memory\n\nNo CLAUDE.md memory files are currently loaded.'
        return '\n'.join(['# Memory', '', claude_md])

    def render_account_report(self, profile: str | None = None) -> str:
        if self.account_runtime is None:
            return '# Account\n\nNo local account runtime is available.'
        if profile:
            return self.account_runtime.render_profile(profile)
        return '\n'.join(['# Account', '', self.account_runtime.render_summary()])

    def render_search_report(
        self,
        query: str | None = None,
        *,
        provider: str | None = None,
        max_results: int = 5,
        domains: tuple[str, ...] = (),
    ) -> str:
        if self.search_runtime is None or not self.search_runtime.has_search_runtime():
            return (
                '# Search\n\nNo local search provider is available. '
                'Add a .claw-search.json or .claude/search.json manifest, '
                'or set SEARXNG_BASE_URL, BRAVE_SEARCH_API_KEY, or TAVILY_API_KEY.'
            )
        if query:
            try:
                return self.search_runtime.render_search_results(
                    query,
                    provider_name=provider,
                    max_results=max_results,
                    domains=domains,
                    timeout_seconds=self.runtime_config.command_timeout_seconds,
                )
            except (KeyError, LookupError, OSError, ValueError) as exc:
                return f'# Search\n\nSearch failed: {exc}'
        if provider:
            return self.search_runtime.render_provider(provider)
        return '\n'.join(['# Search', '', self.search_runtime.render_summary()])

    def render_search_providers_report(self, query: str | None = None) -> str:
        if self.search_runtime is None or not self.search_runtime.has_search_runtime():
            return '# Search Providers\n\nNo local search providers discovered.'
        return self.search_runtime.render_providers_index(query=query)

    def render_search_activate_report(self, provider: str) -> str:
        if self.search_runtime is None or not self.search_runtime.has_search_runtime():
            return '# Search\n\nNo local search provider is available.'
        try:
            report = self.search_runtime.activate_provider(provider)
        except KeyError:
            return f'# Search\n\nUnknown search provider: {provider}'
        clear_context_caches()
        self.tool_context = replace(
            self.tool_context,
            search_runtime=self.search_runtime,
        )
        return '\n'.join(['# Search', '', report.as_text()])

    def render_account_profiles_report(self, query: str | None = None) -> str:
        if self.account_runtime is None:
            return '# Account Profiles\n\nNo local account runtime is available.'
        return self.account_runtime.render_profiles_index(query=query)

    def render_account_login_report(
        self,
        target: str,
        *,
        provider: str | None = None,
        auth_mode: str | None = None,
    ) -> str:
        if self.account_runtime is None:
            return '# Account\n\nNo local account runtime is available.'
        report = self.account_runtime.login(target, provider=provider, auth_mode=auth_mode)
        clear_context_caches()
        return '\n'.join(['# Account', '', report.as_text()])

    def render_account_logout_report(self) -> str:
        if self.account_runtime is None:
            return '# Account\n\nNo local account runtime is available.'
        report = self.account_runtime.logout(reason='slash_or_cli_logout')
        clear_context_caches()
        return '\n'.join(['# Account', '', report.as_text()])

    def render_config_report(self) -> str:
        if self.config_runtime is None:
            return '# Config\n\nNo local config runtime is available.'
        return '\n'.join(['# Config', '', self.config_runtime.render_summary()])

    def render_lsp_report(self) -> str:
        if self.lsp_runtime is None or not self.lsp_runtime.has_lsp_support():
            return '# LSP\n\nNo local LSP runtime is available.'
        return '\n'.join(['# LSP', '', self.lsp_runtime.render_summary()])

    def render_lsp_document_symbols_report(self, file_path: str) -> str:
        if self.lsp_runtime is None or not self.lsp_runtime.has_lsp_support():
            return '# LSP Document Symbols\n\nNo local LSP runtime is available.'
        try:
            return self.lsp_runtime.render_document_symbols(file_path)
        except KeyError as exc:
            return f'# LSP Document Symbols\n\n{exc}'

    def render_lsp_workspace_symbols_report(
        self,
        query: str,
        *,
        max_results: int = 50,
    ) -> str:
        if self.lsp_runtime is None or not self.lsp_runtime.has_lsp_support():
            return '# LSP Workspace Symbols\n\nNo local LSP runtime is available.'
        return self.lsp_runtime.render_workspace_symbols(query, max_results=max_results)

    def render_lsp_definition_report(
        self,
        file_path: str,
        line: int,
        character: int,
        *,
        max_results: int = 20,
    ) -> str:
        if self.lsp_runtime is None or not self.lsp_runtime.has_lsp_support():
            return '# LSP Definition\n\nNo local LSP runtime is available.'
        try:
            return self.lsp_runtime.render_definition(
                file_path,
                line,
                character,
                max_results=max_results,
            )
        except KeyError as exc:
            return f'# LSP Definition\n\n{exc}'

    def render_lsp_references_report(
        self,
        file_path: str,
        line: int,
        character: int,
        *,
        max_results: int = 50,
    ) -> str:
        if self.lsp_runtime is None or not self.lsp_runtime.has_lsp_support():
            return '# LSP References\n\nNo local LSP runtime is available.'
        try:
            return self.lsp_runtime.render_references(
                file_path,
                line,
                character,
                max_results=max_results,
            )
        except KeyError as exc:
            return f'# LSP References\n\n{exc}'

    def render_lsp_hover_report(self, file_path: str, line: int, character: int) -> str:
        if self.lsp_runtime is None or not self.lsp_runtime.has_lsp_support():
            return '# LSP Hover\n\nNo local LSP runtime is available.'
        try:
            return self.lsp_runtime.render_hover(file_path, line, character)
        except KeyError as exc:
            return f'# LSP Hover\n\n{exc}'

    def render_lsp_diagnostics_report(self, file_path: str | None = None) -> str:
        if self.lsp_runtime is None or not self.lsp_runtime.has_lsp_support():
            return '# LSP Diagnostics\n\nNo local LSP runtime is available.'
        try:
            return self.lsp_runtime.render_diagnostics(file_path)
        except KeyError as exc:
            return f'# LSP Diagnostics\n\n{exc}'

    def render_lsp_prepare_call_hierarchy_report(
        self,
        file_path: str,
        line: int,
        character: int,
    ) -> str:
        if self.lsp_runtime is None or not self.lsp_runtime.has_lsp_support():
            return '# LSP Call Hierarchy\n\nNo local LSP runtime is available.'
        try:
            return self.lsp_runtime.render_prepare_call_hierarchy(file_path, line, character)
        except KeyError as exc:
            return f'# LSP Call Hierarchy\n\n{exc}'

    def render_lsp_incoming_calls_report(
        self,
        file_path: str,
        line: int,
        character: int,
        *,
        max_results: int = 50,
    ) -> str:
        if self.lsp_runtime is None or not self.lsp_runtime.has_lsp_support():
            return '# LSP Incoming Calls\n\nNo local LSP runtime is available.'
        try:
            return self.lsp_runtime.render_incoming_calls(
                file_path,
                line,
                character,
                max_results=max_results,
            )
        except KeyError as exc:
            return f'# LSP Incoming Calls\n\n{exc}'

    def render_lsp_outgoing_calls_report(
        self,
        file_path: str,
        line: int,
        character: int,
        *,
        max_results: int = 50,
    ) -> str:
        if self.lsp_runtime is None or not self.lsp_runtime.has_lsp_support():
            return '# LSP Outgoing Calls\n\nNo local LSP runtime is available.'
        try:
            return self.lsp_runtime.render_outgoing_calls(
                file_path,
                line,
                character,
                max_results=max_results,
            )
        except KeyError as exc:
            return f'# LSP Outgoing Calls\n\n{exc}'

    def render_config_effective_report(self) -> str:
        if self.config_runtime is None:
            return '# Config Effective\n\nNo local config runtime is available.'
        return '\n'.join(['# Config Effective', '', self.config_runtime.render_effective_config()])

    def render_config_source_report(self, source: str) -> str:
        if self.config_runtime is None:
            return '# Config Source\n\nNo local config runtime is available.'
        return '\n'.join(['# Config Source', '', self.config_runtime.render_source(source)])

    def render_config_value_report(self, key_path: str, source: str | None = None) -> str:
        if self.config_runtime is None:
            return '# Config Value\n\nNo local config runtime is available.'
        try:
            rendered = self.config_runtime.render_value(key_path, source=source)
        except KeyError as exc:
            label = source if source is not None else key_path
            return f'# Config Value\n\nUnknown config key or source: {label or exc.args[0]}'
        return '\n'.join(['# Config Value', '', rendered])

    def render_mcp_report(self, query: str | None = None) -> str:
        if self.mcp_runtime is None:
            return '# MCP\n\nNo local MCP manifests, servers, or resources discovered.'
        if query:
            return self.mcp_runtime.render_resource_index(query=query)
        return '\n'.join(['# MCP', '', self.mcp_runtime.render_summary()])

    def render_remote_report(self, target: str | None = None) -> str:
        if self.remote_runtime is None:
            return '# Remote\n\nNo local remote runtime is available.'
        if target:
            report = self.remote_runtime.connect(target)
            clear_context_caches()
            return '\n'.join(['# Remote', '', report.as_text()])
        return '\n'.join(['# Remote', '', self.remote_runtime.render_summary()])

    def render_remote_mode_report(self, target: str, *, mode: str) -> str:
        if self.remote_runtime is None:
            return '# Remote\n\nNo local remote runtime is available.'
        report = self.remote_runtime.connect(target, mode=mode)
        clear_context_caches()
        return '\n'.join(['# Remote', '', report.as_text()])

    def render_remote_profiles_report(self, query: str | None = None) -> str:
        if self.remote_runtime is None:
            return '# Remote Profiles\n\nNo local remote runtime is available.'
        return self.remote_runtime.render_profiles_index(query=query)

    def render_remote_disconnect_report(self) -> str:
        if self.remote_runtime is None:
            return '# Remote\n\nNo local remote runtime is available.'
        report = self.remote_runtime.disconnect()
        clear_context_caches()
        return '\n'.join(['# Remote', '', report.as_text()])

    def render_worktree_report(self) -> str:
        if self.worktree_runtime is None:
            return '# Worktree\n\nNo local worktree runtime is available.'
        return '\n'.join(['# Worktree', '', self.worktree_runtime.render_summary()])

    def render_worktree_enter_report(self, name: str | None = None) -> str:
        if self.worktree_runtime is None:
            return '# Worktree\n\nNo local worktree runtime is available.'
        try:
            report = self.worktree_runtime.enter(name=name)
        except (RuntimeError, ValueError) as exc:
            return f'# Worktree\n\n{exc}'
        self._apply_runtime_cwd_update(Path(report.worktree_path or self.runtime_config.cwd))
        return '\n'.join(['# Worktree', '', report.as_text()])

    def render_worktree_exit_report(
        self,
        *,
        action: str = 'keep',
        discard_changes: bool = False,
    ) -> str:
        if self.worktree_runtime is None:
            return '# Worktree\n\nNo local worktree runtime is available.'
        try:
            report = self.worktree_runtime.exit(
                action=action,
                discard_changes=discard_changes,
            )
        except (RuntimeError, ValueError) as exc:
            return f'# Worktree\n\n{exc}'
        target_cwd = report.original_cwd or report.current_cwd or str(self.runtime_config.cwd)
        self._apply_runtime_cwd_update(Path(target_cwd))
        return '\n'.join(['# Worktree', '', report.as_text()])

    def render_worktree_history_report(self) -> str:
        if self.worktree_runtime is None:
            return '# Worktree History\n\nNo local worktree runtime is available.'
        return self.worktree_runtime.render_history()

    def render_mcp_resources_report(self, query: str | None = None) -> str:
        if self.mcp_runtime is None:
            return '# MCP Resources\n\nNo local MCP manifests, servers, or resources discovered.'
        return self.mcp_runtime.render_resource_index(query=query)

    def render_mcp_resource_report(self, uri: str) -> str:
        if self.mcp_runtime is None:
            return '# MCP Resource\n\nNo local MCP manifests, servers, or resources discovered.'
        return self.mcp_runtime.render_resource(uri)

    def render_mcp_tools_report(
        self,
        query: str | None = None,
        *,
        server: str | None = None,
    ) -> str:
        if self.mcp_runtime is None:
            return '# MCP Tools\n\nNo local MCP manifests, servers, or resources discovered.'
        return self.mcp_runtime.render_tool_index(query=query, server_name=server)

    def render_mcp_call_tool_report(
        self,
        tool_name: str,
        *,
        arguments: dict[str, Any] | None = None,
        server: str | None = None,
    ) -> str:
        if self.mcp_runtime is None:
            return '# MCP Tool Result\n\nNo local MCP manifests, servers, or resources discovered.'
        try:
            return self.mcp_runtime.render_tool_call(
                tool_name,
                arguments=arguments,
                server_name=server,
            )
        except FileNotFoundError as exc:
            return f'# MCP Tool Result\n\n{exc}'

    def render_tasks_report(self, status: str | None = None) -> str:
        if self.task_runtime is None:
            return '# Tasks\n\nNo local task runtime is available.'
        return self.task_runtime.render_tasks(status=status)

    def render_next_tasks_report(self) -> str:
        if self.task_runtime is None:
            return '# Next Tasks\n\nNo local task runtime is available.'
        return self.task_runtime.render_next_tasks()

    def render_plan_report(self) -> str:
        if self.plan_runtime is None:
            return '# Plan\n\nNo local plan runtime is available.'
        return self.plan_runtime.render_plan()

    def render_task_report(self, task_id: str) -> str:
        if self.task_runtime is None:
            return '# Task\n\nNo local task runtime is available.'
        return self.task_runtime.render_task(task_id)

    def render_ask_user_report(self) -> str:
        if self.ask_user_runtime is None:
            return '# Ask User\n\nNo local ask-user runtime is available.'
        return '\n'.join(['# Ask User', '', self.ask_user_runtime.render_summary()])

    def render_ask_user_history_report(self) -> str:
        if self.ask_user_runtime is None:
            return '# Ask User History\n\nNo local ask-user runtime is available.'
        return self.ask_user_runtime.render_history()

    def render_teams_report(self, query: str | None = None) -> str:
        if self.team_runtime is None:
            return '# Teams\n\nNo local team runtime is available.'
        return self.team_runtime.render_teams_index(query=query)

    def render_team_report(self, team_name: str) -> str:
        if self.team_runtime is None:
            return '# Team\n\nNo local team runtime is available.'
        try:
            return self.team_runtime.render_team(team_name)
        except KeyError:
            return f'# Team\n\nUnknown team: {team_name}'

    def render_team_messages_report(self, team_name: str | None = None) -> str:
        if self.team_runtime is None:
            return '# Team Messages\n\nNo local team runtime is available.'
        try:
            return self.team_runtime.render_messages(team_name=team_name)
        except KeyError:
            return f'# Team Messages\n\nUnknown team: {team_name}'

    def render_workflows_report(self, query: str | None = None) -> str:
        if self.workflow_runtime is None or not self.workflow_runtime.has_workflows():
            return '# Workflows\n\nNo local workflow runtime is available.'
        return self.workflow_runtime.render_workflows_index(query=query)

    def render_workflow_report(self, workflow_name: str) -> str:
        if self.workflow_runtime is None or not self.workflow_runtime.has_workflows():
            return '# Workflow\n\nNo local workflow runtime is available.'
        try:
            return self.workflow_runtime.render_workflow(workflow_name)
        except KeyError:
            return f'# Workflow\n\nUnknown workflow: {workflow_name}'

    def render_workflow_run_report(
        self,
        workflow_name: str,
        *,
        arguments: dict[str, Any] | None = None,
    ) -> str:
        if self.workflow_runtime is None or not self.workflow_runtime.has_workflows():
            return '# Workflow Run\n\nNo local workflow runtime is available.'
        try:
            return self.workflow_runtime.render_run_report(
                workflow_name,
                arguments=arguments,
            )
        except KeyError:
            return f'# Workflow Run\n\nUnknown workflow: {workflow_name}'

    def render_remote_triggers_report(self, query: str | None = None) -> str:
        if self.remote_trigger_runtime is None or not self.remote_trigger_runtime.has_state():
            return '# Remote Triggers\n\nNo local remote trigger runtime is available.'
        return self.remote_trigger_runtime.render_trigger_index(query=query)

    def render_remote_trigger_report(self, trigger_id: str) -> str:
        if self.remote_trigger_runtime is None or not self.remote_trigger_runtime.has_state():
            return '# Remote Trigger\n\nNo local remote trigger runtime is available.'
        try:
            return self.remote_trigger_runtime.render_trigger(trigger_id)
        except KeyError:
            return f'# Remote Trigger\n\nUnknown remote trigger: {trigger_id}'

    def render_remote_trigger_action_report(
        self,
        action: str,
        *,
        trigger_id: str | None = None,
        body: dict[str, Any] | None = None,
    ) -> str:
        if self.remote_trigger_runtime is None:
            return '# Remote Trigger\n\nNo local remote trigger runtime is available.'
        normalized = action.strip().lower()
        try:
            if normalized == 'list':
                return self.remote_trigger_runtime.render_trigger_index()
            if normalized == 'get':
                if not trigger_id:
                    return '# Remote Trigger\n\ntrigger_id is required for get'
                return self.remote_trigger_runtime.render_trigger(trigger_id)
            if normalized == 'create':
                created = self.remote_trigger_runtime.create_trigger(body or {})
                return self.remote_trigger_runtime.render_trigger(created.trigger_id)
            if normalized == 'update':
                if not trigger_id:
                    return '# Remote Trigger\n\ntrigger_id is required for update'
                updated = self.remote_trigger_runtime.update_trigger(trigger_id, body or {})
                return self.remote_trigger_runtime.render_trigger(updated.trigger_id)
            if normalized == 'run':
                if not trigger_id:
                    return '# Remote Trigger Run\n\ntrigger_id is required for run'
                return self.remote_trigger_runtime.render_run_report(trigger_id, body=body)
        except (KeyError, TypeError, ValueError) as exc:
            return f'# Remote Trigger\n\n{exc}'
        return '# Remote Trigger\n\naction must be one of list, get, create, update, or run'

    def render_hook_policy_report(self) -> str:
        if self.hook_policy_runtime is None:
            return '# Hook Policy\n\nNo local hook or policy manifests discovered.'
        return '\n'.join(['# Hook Policy', '', self.hook_policy_runtime.render_summary()])

    def render_trust_report(self) -> str:
        trusted = True
        settings: dict[str, Any] = {}
        env_values: dict[str, str] = {}
        if self.hook_policy_runtime is not None:
            trusted = self.hook_policy_runtime.is_trusted()
            settings = self.hook_policy_runtime.managed_settings()
            env_values = self.hook_policy_runtime.safe_env()
        lines = [
            '# Trust',
            '',
            f'- Workspace trust mode: {"trusted" if trusted else "untrusted"}',
        ]
        if settings:
            lines.append('- Managed settings:')
            lines.extend(f'  - {key}={value}' for key, value in sorted(settings.items()))
        if env_values:
            lines.append('- Safe environment values:')
            lines.extend(f'  - {key}={value}' for key, value in sorted(env_values.items()))
        return '\n'.join(lines)

    def render_status_report(self) -> str:
        token_counter = describe_token_counter(self.model_config.model)
        lines = [
            '# Status',
            '',
            f'- Model: {self.model_config.model}',
            f'- Token counter: {token_counter.backend} ({token_counter.source})',
            f'- Registered tools: {len(self.tool_registry)}',
            f'- Streaming model responses: {self.runtime_config.stream_model_responses}',
            f'- Session ID: {self.active_session_id or "none"}',
            f'- Last session loaded: {"yes" if self.last_session is not None else "no"}',
        ]
        if self.last_session is not None:
            token_budget = calculate_token_budget(
                session=self.last_session,
                model=self.model_config.model,
                budget_config=self.runtime_config.budget_config,
                output_schema=self.runtime_config.output_schema,
            )
            lines.append(
                f'- Prompt budget: {token_budget.projected_input_tokens:,} / {token_budget.soft_input_limit_tokens:,} soft'
            )
            lines.append(
                f'- Prompt hard limit: {token_budget.hard_input_limit_tokens:,}'
            )
        if self.hook_policy_runtime is not None and self.hook_policy_runtime.manifests:
            lines.append(
                f'- Workspace trust mode: {"trusted" if self.hook_policy_runtime.is_trusted() else "untrusted"}'
            )
        if self.mcp_runtime is not None:
            if self.mcp_runtime.resources:
                lines.append(f'- MCP local resources: {len(self.mcp_runtime.resources)}')
            if self.mcp_runtime.servers:
                lines.append(f'- MCP servers: {len(self.mcp_runtime.servers)}')
        if self.remote_runtime is not None and self.remote_runtime.has_remote_config():
            lines.append(f'- Remote profiles: {len(self.remote_runtime.profiles)}')
            if self.remote_runtime.active_connection is not None:
                connection = self.remote_runtime.active_connection
                lines.append(
                    f'- Active remote: {connection.mode} -> {connection.target}'
                )
        if self.search_runtime is not None and self.search_runtime.has_search_runtime():
            lines.append(f'- Search providers: {len(self.search_runtime.providers)}')
            active_provider = self.search_runtime.current_provider()
            if active_provider is not None:
                lines.append(
                    f'- Active search provider: {active_provider.name} ({active_provider.provider})'
                )
        if self.account_runtime is not None and self.account_runtime.has_account_state():
            lines.append(f'- Account profiles: {len(self.account_runtime.profiles)}')
            if self.account_runtime.active_session is not None:
                session = self.account_runtime.active_session
                lines.append(
                    f'- Active account: {session.provider} -> {session.identity}'
                )
        if self.ask_user_runtime is not None and self.ask_user_runtime.has_state():
            lines.append(f'- Ask-user queued answers: {len(self.ask_user_runtime.queued_answers)}')
            lines.append(f'- Ask-user history: {len(self.ask_user_runtime.history)}')
        if self.config_runtime is not None and self.config_runtime.has_config():
            lines.append(f'- Config sources: {len(self.config_runtime.sources)}')
            lines.append(
                f'- Effective config keys: {len(self.config_runtime.list_keys())}'
            )
        if self.lsp_runtime is not None and self.lsp_runtime.has_lsp_support():
            lines.append(
                f'- LSP indexed files: {len(self.lsp_runtime._workspace_files())}'
            )
        if self.plan_runtime is not None and self.plan_runtime.steps:
            lines.append(f'- Local plan steps: {len(self.plan_runtime.steps)}')
        if self.task_runtime is not None and self.task_runtime.tasks:
            lines.append(f'- Local tasks: {len(self.task_runtime.tasks)}')
        if self.team_runtime is not None and self.team_runtime.has_team_state():
            lines.append(f'- Local teams: {len(self.team_runtime.teams)}')
            lines.append(f'- Team messages: {len(self.team_runtime.messages)}')
        if self.last_session_path is not None:
            lines.append(f'- Session path: {self.last_session_path}')
        if self.last_run_result is not None:
            lines.extend(
                [
                    f'- Last run turns: {self.last_run_result.turns}',
                    f'- Last run tool calls: {self.last_run_result.tool_calls}',
                    f'- Last run total tokens: {self.last_run_result.usage.total_tokens}',
                    f'- Last run total cost: ${self.last_run_result.total_cost_usd:.6f}',
                ]
            )
            if self.last_run_result.scratchpad_directory is not None:
                lines.append(
                    f'- Scratchpad directory: {self.last_run_result.scratchpad_directory}'
                )
        else:
            lines.append('- Last run: none')
        if self.agent_manager is not None:
            lines.extend(self.agent_manager.summary_lines())
        return '\n'.join(lines)

    def render_token_budget_report(self) -> str:
        session = self.last_session or self.build_session()
        snapshot = calculate_token_budget(
            session=session,
            model=self.model_config.model,
            budget_config=self.runtime_config.budget_config,
            output_schema=self.runtime_config.output_schema,
        )
        return format_token_budget(snapshot)

    def _finalize_managed_agent(self, result: AgentRunResult) -> None:
        if self.managed_agent_id is None or self.agent_manager is None:
            self.resume_source_session_id = None
            return
        self.agent_manager.finish_agent(
            self.managed_agent_id,
            session_id=result.session_id,
            session_path=result.session_path,
            turns=result.turns,
            tool_calls=result.tool_calls,
            stop_reason=result.stop_reason,
        )
        self.resume_source_session_id = None

    def _accumulate_usage(self, result: AgentRunResult) -> None:
        """Add a run's usage to the cumulative session totals."""
        self.cumulative_usage = self.cumulative_usage + result.usage
        self.cumulative_cost_usd += result.total_cost_usd

    def _refresh_runtime_views_for_tool_result(
        self,
        tool_name: str,
        tool_result: ToolExecutionResult,
    ) -> None:
        if not tool_result.ok:
            return
        cwd_update = tool_result.metadata.get('cwd_update')
        if isinstance(cwd_update, str) and cwd_update:
            self._apply_runtime_cwd_update(Path(cwd_update))
        refresh_tool_names = {
            'update_plan',
            'plan_clear',
            'task_create',
            'task_update',
            'task_start',
            'task_complete',
            'task_block',
            'task_cancel',
            'todo_write',
            'search_activate_provider',
            'remote_connect',
            'remote_disconnect',
            'account_login',
            'account_logout',
            'config_set',
            'ask_user_question',
            'team_create',
            'team_delete',
            'send_message',
            'workflow_run',
            'remote_trigger',
            'worktree_enter',
            'worktree_exit',
        }
        if tool_name not in refresh_tool_names:
            return
        clear_context_caches()
        additional_dirs = tuple(
            str(path) for path in self.runtime_config.additional_working_directories
        )
        if tool_name.startswith('remote_'):
            self.remote_runtime = RemoteRuntime.from_workspace(
                self.runtime_config.cwd,
                additional_working_directories=additional_dirs,
            )
        if tool_name == 'remote_trigger':
            self.remote_trigger_runtime = RemoteTriggerRuntime.from_workspace(
                self.runtime_config.cwd,
                additional_working_directories=additional_dirs,
            )
        if tool_name.startswith('search_'):
            self.search_runtime = SearchRuntime.from_workspace(
                self.runtime_config.cwd,
                additional_working_directories=additional_dirs,
            )
        if tool_name.startswith('account_'):
            self.account_runtime = AccountRuntime.from_workspace(
                self.runtime_config.cwd,
                additional_working_directories=additional_dirs,
            )
        if tool_name == 'ask_user_question':
            self.ask_user_runtime = AskUserRuntime.from_workspace(
                self.runtime_config.cwd,
                additional_working_directories=additional_dirs,
            )
        if tool_name == 'config_set':
            self.config_runtime = ConfigRuntime.from_workspace(self.runtime_config.cwd)
        if tool_name.startswith('task_') or tool_name == 'todo_write':
            self.task_runtime = TaskRuntime.from_workspace(self.runtime_config.cwd)
        if tool_name.startswith('plan_') or tool_name == 'update_plan':
            self.plan_runtime = PlanRuntime.from_workspace(self.runtime_config.cwd)
        if tool_name.startswith('team_') or tool_name == 'send_message':
            self.team_runtime = TeamRuntime.from_workspace(
                self.runtime_config.cwd,
                additional_working_directories=additional_dirs,
            )
        if tool_name.startswith('workflow_'):
            self.workflow_runtime = WorkflowRuntime.from_workspace(
                self.runtime_config.cwd,
                additional_working_directories=additional_dirs,
            )
        if tool_name.startswith('worktree_'):
            self.worktree_runtime = WorktreeRuntime.from_workspace(self.runtime_config.cwd)
        self.tool_context = replace(
            self.tool_context,
            tool_registry=self.tool_registry,
            search_runtime=self.search_runtime,
            account_runtime=self.account_runtime,
            ask_user_runtime=self.ask_user_runtime,
            config_runtime=self.config_runtime,
            lsp_runtime=self.lsp_runtime,
            remote_runtime=self.remote_runtime,
            remote_trigger_runtime=self.remote_trigger_runtime,
            plan_runtime=self.plan_runtime,
            task_runtime=self.task_runtime,
            team_runtime=self.team_runtime,
            workflow_runtime=self.workflow_runtime,
            worktree_runtime=self.worktree_runtime,
        )

    def _apply_runtime_cwd_update(self, new_cwd: Path) -> None:
        resolved_cwd = new_cwd.resolve()
        if resolved_cwd == self.runtime_config.cwd.resolve():
            return
        self.runtime_config = replace(self.runtime_config, cwd=resolved_cwd)
        clear_context_caches()
        additional_dirs = tuple(
            str(path) for path in self.runtime_config.additional_working_directories
        )
        self.plugin_runtime = PluginRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.hook_policy_runtime = HookPolicyRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.mcp_runtime = MCPRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.remote_runtime = RemoteRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.remote_trigger_runtime = RemoteTriggerRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.search_runtime = SearchRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.account_runtime = AccountRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.ask_user_runtime = AskUserRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.config_runtime = ConfigRuntime.from_workspace(self.runtime_config.cwd)
        self.lsp_runtime = LSPRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.task_runtime = TaskRuntime.from_workspace(self.runtime_config.cwd)
        self.plan_runtime = PlanRuntime.from_workspace(self.runtime_config.cwd)
        self.team_runtime = TeamRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.workflow_runtime = WorkflowRuntime.from_workspace(
            self.runtime_config.cwd,
            additional_dirs,
        )
        self.worktree_runtime = WorktreeRuntime.from_workspace(self.runtime_config.cwd)
        self.runtime_config = self._apply_hook_policy_budget_overrides(self.runtime_config)
        registry = dict(default_tool_registry())
        if self.plugin_runtime is not None:
            alias_tools = self.plugin_runtime.register_tool_aliases(registry)
            if alias_tools:
                registry = {**registry, **alias_tools}
            virtual_tools = self.plugin_runtime.register_virtual_tools(registry)
            if virtual_tools:
                registry = {**registry, **virtual_tools}
        self.tool_registry = registry
        self.tool_context = build_tool_context(
            self.runtime_config,
            tool_registry=self.tool_registry,
            extra_env=(
                self.hook_policy_runtime.safe_env()
                if self.hook_policy_runtime is not None
                else None
            ),
            search_runtime=self.search_runtime,
            account_runtime=self.account_runtime,
            ask_user_runtime=self.ask_user_runtime,
            config_runtime=self.config_runtime,
            lsp_runtime=self.lsp_runtime,
            mcp_runtime=self.mcp_runtime,
            remote_runtime=self.remote_runtime,
            remote_trigger_runtime=self.remote_trigger_runtime,
            plan_runtime=self.plan_runtime,
            task_runtime=self.task_runtime,
            team_runtime=self.team_runtime,
            workflow_runtime=self.workflow_runtime,
            worktree_runtime=self.worktree_runtime,
        )

    def _apply_plugin_before_prompt_hooks(self, prompt: str) -> str:
        if self.plugin_runtime is None:
            return prompt
        injections = self.plugin_runtime.before_prompt_injections()
        state_reminder = self.plugin_runtime.runtime_state_reminder()
        if not injections and not state_reminder:
            return prompt
        lines = ['<system-reminder>', 'Plugin before-prompt hooks:']
        lines.extend(f'- {entry}' for entry in injections)
        if state_reminder:
            lines.extend(['', state_reminder])
        lines.extend(['</system-reminder>', '', prompt])
        return '\n'.join(lines)

    def _apply_plugin_resume_hooks(
        self,
        prompt: str,
        *,
        resumed: bool,
    ) -> str:
        if not resumed or self.plugin_runtime is None:
            return prompt
        injections = self.plugin_runtime.on_resume_injections()
        if not injections:
            return prompt
        lines = ['<system-reminder>', 'Plugin resume hooks:']
        lines.extend(f'- {entry}' for entry in injections)
        lines.extend(['</system-reminder>', '', prompt])
        return '\n'.join(lines)

    def _render_plugin_persist_message(
        self,
        messages: tuple[str, ...],
    ) -> str:
        lines = ['<system-reminder>', 'Plugin persist hooks:']
        lines.extend(f'- {entry}' for entry in messages)
        lines.extend(
            [
                '',
                'This session state was persisted with plugin lifecycle guidance.',
                '</system-reminder>',
            ]
        )
        return '\n'.join(lines)

    def _append_plugin_after_turn_events(
        self,
        result: AgentRunResult,
        *,
        prompt: str,
        turn_index: int,
    ) -> AgentRunResult:
        if self.plugin_runtime is None:
            return result
        injections = self.plugin_runtime.after_turn_injections()
        if not injections:
            return result
        appended = list(result.events)
        for entry in injections:
            appended.append(
                {
                    'type': 'plugin_after_turn',
                    'turn_index': turn_index,
                    'message': entry,
                    'prompt_preview': self._preview_text(prompt, 120),
                    'stop_reason': result.stop_reason,
                }
            )
        return replace(result, events=tuple(appended))

    def _append_runtime_after_turn_events(
        self,
        result: AgentRunResult,
        *,
        prompt: str,
        turn_index: int,
    ) -> AgentRunResult:
        updated = self._append_plugin_after_turn_events(
            result,
            prompt=prompt,
            turn_index=turn_index,
        )
        if self.hook_policy_runtime is None:
            return updated
        injections = self.hook_policy_runtime.after_turn_messages()
        if not injections:
            return updated
        appended = list(updated.events)
        for entry in injections:
            appended.append(
                {
                    'type': 'hook_policy_after_turn',
                    'turn_index': turn_index,
                    'message': entry,
                    'prompt_preview': self._preview_text(prompt, 120),
                    'stop_reason': updated.stop_reason,
                    'trusted': self.hook_policy_runtime.is_trusted(),
                }
            )
        return replace(updated, events=tuple(appended))


def _optional_policy_int(value: object) -> int | None:
    if isinstance(value, bool) or not isinstance(value, int):
        return None
    return value


def _optional_policy_float(value: object) -> float | None:
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return float(value)
    return None