"""ConversationManager — runs the tool-loop and forwards events to a callback.

Lives on a worker thread (Qt bridge owns the QThread). The GUI never touches
this directly except through the bridge.

Cancellation: a threading.Event the GUI sets when Stop is clicked. We check
it between rounds and propagate it into client.stream() so the OpenAI iterator
short-circuits mid-chunk.

Roles: a manager instance is either a planner, a worker, or "single" (legacy
single-model chat). The role is implicit in the ToolRegistry's mode plus the
History's system prompt — the manager itself only branches when it sees a
`dispatch_to_worker` tool call: that path is intercepted and routed through
the supplied DispatchCallback rather than the registry.
"""
from __future__ import annotations

import hashlib
import json
import re
import subprocess
import threading
from pathlib import Path
from typing import Any, Callable

from aura.client import (
    ApiError,
    ContentDelta,
    Done,
    Event,
    ReasoningDelta,
    TerminalOutput,
    ToolCallArgsDelta,
    ToolCallEnd,
    ToolCallStart,
    ToolResult,
    Usage,
    WorkerDispatchRequested,
)
from aura.config import ModelId, ThinkingMode
from aura.conversation.dispatch import (
    DispatchCallback,
    WorkerDispatchRequest,
    WorkerDispatchResult,
    WorkerOutcomeStatus,
    infer_outcome_status,
)
from aura.conversation.history import History
from aura.conversation.loop_detection import LoopDetector
from aura.conversation.tool_limits import (
    MAX_WORKER_REDISPATCHES_PER_USER_TURN,
    WRITE_TOOLS,
    ToolLimitState,
    limit_reached_payload,
)
from aura.conversation.tool_runner import ToolRunner
from aura.conversation.tools._types import (
    ApprovalCallback,
    ApprovalDecision,
    ApprovalRequest,
)
from aura.conversation.tools.registry import ToolRegistry
from aura.hooks import hooks
from aura.project_env import preferred_python_for_compile, quote_command_arg

EventCallback = Callable[[Event], None]

EDIT_MECHANICS_FAILURE_CLASSES = {
    "edit_mechanics_symbol_not_found",
    "edit_mechanics_old_str_not_found",
    "edit_mechanics_ambiguous_match",
    "patch_hunk_not_found",
    "patch_hunk_ambiguous",
    "patch_file_hash_mismatch",
}

EDIT_TRANSACTION_FAILURE_CLASSES = {
    "edit_transaction_hash_mismatch",
    "edit_transaction_symbol_not_found",
    "edit_transaction_ambiguous_symbol",
    "edit_transaction_invalid_operation",
    "edit_transaction_invalid_syntax",
    "edit_transaction_not_applicable",
}

COMPLETION_PHRASE_MARKERS = (
    "all set",
    "staged and ready",
    "ready for you",
    "let me know",
    "if you need anything else",
    "committed and done",
    "everything else is in good shape",
    "when you want to commit",
    "no further action needed",
)

TASK_COMPLETION_TOOL_NAMES = {
    "run_terminal_command",
    "run_diagnostic_command",
    "git_status",
    "git_diff",
    "git_log",
    "git_show",
    "git_log_file",
}

WORKER_RECOVERY_ALWAYS_ALLOWED = {
    "read_file",
    "read_files",
    "read_file_range",
    "read_file_outline",
    "grep_search",
    "find_usages",
    "search_codebase",
    "list_directory",
    "glob",
    "git_status",
    "git_diff",
    "git_log",
    "git_show",
    "git_log_file",
    "run_diagnostic_command",
    "get_workspace_snapshot",
}

SYNTAX_REPAIR_ALWAYS_ALLOWED = WORKER_RECOVERY_ALWAYS_ALLOWED

ACTION_COMPLETION_TOOL_NAMES = TASK_COMPLETION_TOOL_NAMES | WRITE_TOOLS

WORKER_EDIT_RECOVERY_INSTRUCTION = (
    "Previous edit failed recoverably. Re-read the affected file with read_file or read_file_range, "
    "then retry patch_file once with corrected hunks and the current expected_file_hash. "
    "Do not use write_file as a fallback for an existing-file edit. "
    "Finish only after the edit is applied and touched Python files pass py_compile."
)

WORKER_AUTO_PY_COMPILE_INSTRUCTION = (
    "Focused py_compile failed on the following Python file(s). "
    "Re-read and repair the file(s), then run python -m py_compile again. "
    "Finish only after py_compile passes.\n\n"
    "Diagnostic output:\n{diagnostics}"
)

def _normalize_worker_path(path: str) -> str:
    normalized = str(path).replace("\\", "/")
    if normalized.startswith("./"):
        normalized = normalized[2:]
    while "//" in normalized:
        normalized = normalized.replace("//", "/")
    return normalized


def _unique_worker_paths(paths: list[str]) -> list[str]:
    unique: list[str] = []
    seen: set[str] = set()
    for path in paths:
        normalized = _normalize_worker_path(path).strip()
        if not normalized or normalized in seen:
            continue
        unique.append(normalized)
        seen.add(normalized)
    return unique


def _planner_stale_read_notice(modified_files: list[str]) -> str:
    files = _unique_worker_paths(modified_files)
    bullet_list = "\n".join(f"- {path}" for path in files)
    return (
        "Planner stale-read invalidation:\n"
        "The Worker modified these files:\n"
        f"{bullet_list}\n\n"
        "Any prior Planner reads of those paths are stale. "
        "Re-read the modified files before planning, dispatching, or reasoning "
        "about further edits involving them. "
        "If the Worker completed successfully, summarize or finish normally; "
        "do not redispatch because of this notice unless the user asks for more."
    )


def _is_validation_scratch_path(path: str) -> bool:
    normalized = _normalize_worker_path(path)
    name = normalized.rsplit("/", 1)[-1]
    if not name.endswith(".py"):
        return False
    if normalized.startswith(".aura/tmp/") or "/" not in normalized:
        return name.startswith(
            (
                "dump",
                "_check",
                "check",
                "tmp",
                "_tmp",
                "_inspect",
                "inspect",
                "diagnostic",
                "_diagnostic",
            )
        )
    return False


class ConversationManager:
    def __init__(
        self,
        history: History,
        tool_registry: ToolRegistry,
    ) -> None:
        self._history = history
        self._tools = tool_registry
        self._loop_detector = LoopDetector()
        self._tool_runner = ToolRunner(
            history=self._history,
            workspace_root=self._tools.workspace_root,
            loop_detector=self._loop_detector,
        )

    @property
    def history(self) -> History:
        return self._history

    @property
    def tools(self) -> ToolRegistry:
        return self._tools

    def set_workspace_root(self, root: Path) -> None:
        self._tool_runner.set_workspace_root(root)

    def send(
        self,
        on_event: EventCallback,
        approval_cb: ApprovalCallback,
        cancel_event: threading.Event,
        model: ModelId,
        thinking: ThinkingMode,
        dispatch_cb: DispatchCallback | None = None,
        temperature: float = 0.7,
        max_tool_rounds: int | None = None,
        hook_name: str = 'generate_planner_code',
        explicit_validation_commands: list[str] | None = None,
    ) -> None:
        """Run the model -> tool -> model loop until the model stops calling tools.

        Caller appends the user message to history before invoking this.

        `dispatch_cb` is required when the registry is in "planner" mode (the
        only mode that exposes the `dispatch_to_worker` tool). If the tool is
        called and `dispatch_cb` is None, the call returns an error result so
        the planner can recover rather than blocking forever.

        `hook_name` controls which hook to trigger for model generation.
        The planner uses `generate_planner_code`; workers use `generate_worker_code`.
        """
        import concurrent.futures

        reject_all_for_turn = False
        mode = getattr(self._tools, "mode", "single")
        limits = ToolLimitState(mode=mode)
        rounds_used = 0
        worker_needs_final_report = False
        worker_phase_boundary_info: dict[str, Any] | None = None
        worker_redispatches = 0
        worker_dispatch_failures: dict[str, int] = {}
        edit_failed_shapes: set[str] = set()
        edit_fallback_required: dict[str, dict[str, Any]] = {}
        recovery_block_counts: dict[str, int] = {}
        line_range_reread_required: dict[str, dict[str, Any]] = {}
        worker_file_state: dict[str, dict[str, Any]] = {}
        patch_failed_cycles: dict[str, int] = {}
        syntax_repair_required: dict[str, dict[str, Any]] = {}
        syntax_validation_required: set[str] = set()
        write_attempts_by_path: dict[str, int] = {}
        worker_recovery_nudge_sent = False
        stale_validation_notes: list[str] = []
        task_completion_context = False
        final_messages_after_completion = 0
        last_completion_final_text = ""

        while True:
            if (
                mode in {"planner", "single"}
                and task_completion_context
                and final_messages_after_completion >= 1
            ):
                return

            rounds_used += 1
            if max_tool_rounds is not None and rounds_used > max_tool_rounds:
                on_event(ApiError(status_code=None, message=f"Exceeded max tool rounds ({max_tool_rounds})."))
                return

            limits.begin_model_round()
            if cancel_event.is_set():
                self._cleanup_cancelled(on_event)
                return

            full_message: dict[str, Any] | None = None
            tool_defs = [] if worker_needs_final_report else self._tools.tool_defs()

            for ev in hooks.trigger(
                hook_name,
                messages=self._history.for_api(),
                tools=tool_defs,
                model=model,
                thinking=thinking,
                cancel_event=cancel_event,
                temperature=temperature,
            ):
                on_event(ev)
                if isinstance(ev, Done):
                    full_message = ev.full_message
                if isinstance(ev, ApiError):
                    return  # surface and stop

            if cancel_event.is_set():
                # If we have some content but no tool calls, we can keep it.
                # If it's empty or has orphaned tool calls, we must strip it.
                if full_message is not None:
                    # DeepSeek/OpenRouter specific: reasoning_content is NOT 'content' for the API.
                    # Standard APIs REQUIRE 'content' (string) or 'tool_calls' (list).
                    content = full_message.get("content")
                    reasoning = full_message.get("reasoning_content")

                    has_any_text = bool(content or reasoning)
                    if has_any_text:
                        full_message.pop("tool_calls", None)
                        # Normalize content to string so API doesn't reject it
                        if full_message.get("content") is None:
                            full_message["content"] = ""
                        self._history.append_assistant(full_message)
                    else:
                        self._cleanup_cancelled(on_event)
                else:
                    self._cleanup_cancelled(on_event)
                return

            if full_message is None:
                # Should not happen in normal stream completion
                return

            tool_calls = full_message.get("tool_calls") or []
            if (
                not tool_calls
                and mode in {"planner", "single"}
                and task_completion_context
            ):
                content_text = self._assistant_message_text(full_message)
                if final_messages_after_completion >= 1:
                    if self._is_repetitive_completion_final(
                        content_text,
                        last_completion_final_text,
                    ):
                        return
                    return
                self._history.append_assistant(full_message)
                final_messages_after_completion += 1
                last_completion_final_text = content_text
                return

            self._history.append_assistant(full_message)

            if worker_needs_final_report:
                for tc in tool_calls:
                    fn = tc["function"]
                    name = fn["name"]
                    tool_call_id = tc["id"]
                    reason = (
                        str(worker_phase_boundary_info.get("reason"))
                        if worker_phase_boundary_info
                        else "worker_phase_boundary"
                    )
                    message = (
                        str(worker_phase_boundary_info.get("message"))
                        if worker_phase_boundary_info
                        else (
                            "Worker reached a recoverable phase boundary for this pass. "
                            "Produce the continuation report now."
                        )
                    )
                    info = {
                        "ok": False,
                        "limit_reached": bool(
                            worker_phase_boundary_info
                            and worker_phase_boundary_info.get("limit_reached")
                        ),
                        "loop_detected": bool(
                            worker_phase_boundary_info
                            and worker_phase_boundary_info.get("loop_detected")
                        ),
                        "recoverable": True,
                        "phase_boundary": True,
                        "reason": reason,
                        "tool": name,
                        "message": message,
                        "counts": limits.to_dict(),
                    }
                    self._append_limit_tool_result(tool_call_id, name, info, on_event)
                return

            if not tool_calls:
                if mode == "worker":
                    if self._has_terminal_syntax_failure(syntax_repair_required):
                        if not worker_recovery_nudge_sent:
                            diagnostic_parts = []
                            for path, state in syntax_repair_required.items():
                                if state.get("repair_failed") and state.get("error"):
                                    diagnostic_parts.append(f"{path}:\n{state['error']}")
                            diagnostic_text = "\n\n".join(diagnostic_parts)
                            instruction = (
                                "Terminal py_compile still failing after repair. "
                                "Re-read the failing Python file, fix the syntax error, "
                                "then re-run python -m py_compile. "
                                "Finish only after py_compile passes."
                            )
                            if diagnostic_text:
                                instruction += f"\n\nDiagnostic output:\n{diagnostic_text}"
                            self._history.append_user_text(instruction)
                            worker_recovery_nudge_sent = True
                            continue
                        self._finish_worker_unrecoverable(
                            on_event,
                            failure_class="syntax_invalid",
                            error="Python syntax still fails after two repair attempts.",
                        )
                        return
                    edit_recovery_pending = bool(
                        edit_fallback_required or line_range_reread_required
                    )
                    syntax_repair_pending = bool(
                        self._syntax_repair_paths(syntax_repair_required)
                    )
                    if (
                        edit_recovery_pending
                        or syntax_repair_pending
                    ):
                        if not worker_recovery_nudge_sent:
                            if edit_recovery_pending:
                                instruction = WORKER_EDIT_RECOVERY_INSTRUCTION
                            else:
                                # Distinguish craft-gate failures from terminal py_compile failures
                                any_repair_failed = any(
                                    state.get("repair_failed")
                                    for state in syntax_repair_required.values()
                                )
                                if any_repair_failed:
                                    instruction = (
                                        "Previous py_compile failed. Re-read the touched "
                                        "Python file, repair it with patch_file, then run python -m py_compile again. "
                                        "Finish only after py_compile passes."
                                    )
                                else:
                                    instruction = (
                                        "Validation caught invalid Python in the following file(s). "
                                        "Re-read the file, repair with patch_file, "
                                        "then run python -m py_compile. "
                                        "Finish only after py_compile passes."
                                    )
                            if not edit_recovery_pending:
                                diagnostic_parts = []
                                for path, state in syntax_repair_required.items():
                                    if (
                                        not state.get("awaiting_validation")
                                        and not state.get("repair_failed")
                                        and state.get("error")
                                    ):
                                        diagnostic_parts.append(f"{path}:\n{state['error']}")
                                diagnostic_text = "\n\n".join(diagnostic_parts)
                                if diagnostic_text:
                                    instruction += f"\n\nDiagnostic output:\n{diagnostic_text}"
                            self._history.append_user_text(instruction)
                            worker_recovery_nudge_sent = True
                            continue
                        error_parts = [
                            "Worker stopped before recovering from a recoverable failure."
                        ]
                        details: dict[str, Any] = {}
                        if syntax_repair_pending:
                            sync_paths = sorted(
                                self._syntax_repair_paths(syntax_repair_required)
                            )
                            error_parts.append(
                                f" Syntax repair pending on: {', '.join(sync_paths)}."
                            )
                            details["syntax_paths"] = sync_paths
                        if edit_recovery_pending:
                            error_parts.append(
                                " Edit mechanics recovery pending."
                            )
                            details.update(self._edit_recovery_details(
                                edit_fallback_required,
                                line_range_reread_required,
                            ))
                        self._finish_worker_unrecoverable(
                            on_event,
                            failure_class="worker_recovery_exhausted",
                            error="".join(error_parts),
                            details=details or None,
                        )
                        return
                    syntax_validation_required.difference_update(
                        path for path in set(syntax_validation_required)
                        if _is_validation_scratch_path(path)
                    )
                    if syntax_validation_required:
                        product_paths = sorted(
                            _normalize_worker_path(path) for path in syntax_validation_required
                            if not _is_validation_scratch_path(path)
                        )
                        if product_paths:
                            all_ok, diagnostics = self._run_focused_py_compile(product_paths)
                            self._emit_auto_py_compile_result(
                                paths=product_paths,
                                ok=all_ok,
                                diagnostics=diagnostics,
                                on_event=on_event,
                            )
                            if all_ok:
                                syntax_validation_required.clear()
                            else:
                                # Auto-py_compile failed — feed diagnostics back for repair
                                for path in product_paths:
                                    self._set_syntax_repair_state(syntax_repair_required, path, {
                                        "error": diagnostics,
                                        "failed_repairs": 0,
                                    })
                                syntax_validation_required.clear()
                                instruction = WORKER_AUTO_PY_COMPILE_INSTRUCTION.format(
                                    diagnostics=diagnostics,
                                )
                                self._history.append_user_text(instruction)
                                continue
                    return
                return

            _terminal_dispatch = False
            _worker_phase_boundary_info: dict[str, Any] | None = None

            # Pre-process tools sequentially to handle limits check and identify parallelizable ones
            tasks = []
            for tc in tool_calls:
                fn = tc["function"]
                name = fn["name"]
                tool_call_id = tc["id"]
                try:
                    args = json.loads(fn.get("arguments") or "{}")
                except json.JSONDecodeError as exc:
                    err = f"failed to parse tool arguments as JSON: {exc}"
                    self._history.append_tool_result(
                        tool_call_id, json.dumps({"ok": False, "error": err})
                    )
                    on_event(
                        ToolResult(
                            tool_call_id=tool_call_id,
                            name=name,
                            ok=False,
                            result=err,
                        )
                    )
                    continue

                allowed, limit_info = limits.check(name)
                if not allowed:
                    self._append_limit_tool_result(tool_call_id, name, limit_info, on_event)
                    if self._is_recoverable_phase_boundary(limit_info):
                        _worker_phase_boundary_info = limit_info
                    continue
                limits.record(name)
                tasks.append({"id": tool_call_id, "name": name, "args": args})

            if cancel_event.is_set():
                self._cleanup_cancelled(on_event)
                return

            def process_task(task: dict[str, Any]) -> dict[str, Any]:
                nonlocal _terminal_dispatch, _worker_phase_boundary_info, reject_all_for_turn, worker_redispatches, stale_validation_notes
                tool_call_id = task["id"]
                name = task["name"]
                args = task["args"]

                if mode == "worker":
                    blocked = self._worker_recovery_block(
                        tool_call_id=tool_call_id,
                        name=name,
                        args=args,
                        edit_failed_shapes=edit_failed_shapes,
                        edit_fallback_required=edit_fallback_required,
                        recovery_block_counts=recovery_block_counts,
                        line_range_reread_required=line_range_reread_required,
                        worker_file_state=worker_file_state,
                        patch_failed_cycles=patch_failed_cycles,
                        syntax_repair_required=syntax_repair_required,
                        syntax_validation_required=syntax_validation_required,
                        write_attempts_by_path=write_attempts_by_path,
                    )
                    if blocked is not None:
                        blocked_payload = self._parse_tool_payload(str(blocked.get("result_payload", "")))
                        if self._is_recoverable_phase_boundary(blocked_payload):
                            _worker_phase_boundary_info = blocked_payload
                        return blocked

                if name == "dispatch_to_worker":
                    result = self._tool_runner.handle_dispatch(
                        tool_call_id=tool_call_id,
                        args=args,
                        on_event=on_event,
                        dispatch_cb=dispatch_cb,
                    )
                    if result is not None and not result.cancelled:
                        if result.ok:
                            _terminal_dispatch = True
                        else:
                            action = self._classify_failed_worker_dispatch(
                                args=args,
                                result=result,
                                failures=worker_dispatch_failures,
                                failed_attempts=worker_redispatches,
                            )
                            if action["counts_as_attempt"]:
                                worker_redispatches += 1
                            blocker_reason = action["blocker_reason"]
                            if blocker_reason:
                                return {
                                    "id": tool_call_id,
                                    "blocker": True,
                                    "result": result,
                                    "blocker_reason": blocker_reason,
                                    "planner_stale_read_files": (
                                        list(result.modified_files)
                                        if result.modified_files
                                        else []
                                    ),
                                }
                    return {
                        "id": tool_call_id,
                        "skip": True,
                        "completed_dispatch_for_final": self._is_completed_worker_result(result),
                        "planner_stale_read_files": (
                            list(result.modified_files) if result and result.modified_files else []
                        ),
                    }

                if name == "run_research":
                    ok = self._tool_runner.handle_research(
                        tool_call_id=tool_call_id,
                        args=args,
                        on_event=on_event,
                        model=model,
                        cancel_event=cancel_event,
                        temperature=temperature,
                    )
                    if ok:
                        _terminal_dispatch = True
                    return {"id": tool_call_id, "skip": True}

                if name == "run_terminal_command":
                    loop_info = self._tool_runner.handle_terminal_command(
                        tool_call_id=tool_call_id,
                        args=args,
                        on_event=on_event,
                        cancel_event=cancel_event,
                        mode=mode,
                        explicit_validation_commands=explicit_validation_commands,
                    )
                    if mode == "worker":
                        self._update_syntax_state_from_terminal(
                            args=args,
                            loop_info=loop_info,
                            syntax_repair_required=syntax_repair_required,
                            syntax_validation_required=syntax_validation_required,
                            stale_validation_notes=stale_validation_notes,
                        )
                    if self._is_recoverable_phase_boundary(loop_info):
                        _worker_phase_boundary_info = loop_info
                    return {
                        "id": tool_call_id,
                        "skip": True,
                        "completed_tool_result_for_final": self._terminal_result_completed(loop_info),
                    }

                if reject_all_for_turn and name in WRITE_TOOLS:
                    payload = json.dumps(
                        {
                            "ok": False,
                            "error": "User rejected all writes in this turn.",
                            "failure_class": "approval_rejected",
                            "applied": False,
                            "write_outcome": "not_applied_user_rejected",
                        }
                    )
                    return {
                        "id": tool_call_id,
                        "result_payload": payload,
                        "event": ToolResult(
                            tool_call_id=tool_call_id,
                            name=name,
                            ok=False,
                            result=payload,
                            extras={"approval": "reject_all"},
                        )
                    }

                exec_result = self._tools.execute(
                    name=name,
                    args=args,
                    approval_cb=approval_cb,
                    reject_all=False,
                )

                # Check rejection state after execute (approval_cb could set it)
                if exec_result.extras.get("approval") == "reject_all":
                    reject_all_for_turn = True

                tool_msg_content = exec_result.to_tool_message_content()
                if mode == "worker":
                    tool_msg_content = self._update_worker_recovery_state(
                        name=name,
                        args=args,
                        ok=exec_result.ok,
                        content=tool_msg_content,
                        edit_failed_shapes=edit_failed_shapes,
                        edit_fallback_required=edit_fallback_required,
                        line_range_reread_required=line_range_reread_required,
                        worker_file_state=worker_file_state,
                        patch_failed_cycles=patch_failed_cycles,
                        syntax_repair_required=syntax_repair_required,
                        syntax_validation_required=syntax_validation_required,
                        write_attempts_by_path=write_attempts_by_path,
                    )

                loop_result = self._apply_loop_detection(
                    mode=mode,
                    name=name,
                    args=args,
                    ok=exec_result.ok,
                    result_payload=tool_msg_content,
                )
                tool_msg_content = loop_result["content"]
                loop_info = loop_result["info"]

                if self._is_recoverable_phase_boundary(loop_info):
                    _worker_phase_boundary_info = loop_info

                return {
                    "id": tool_call_id,
                    "result_payload": tool_msg_content,
                    "event": ToolResult(
                        tool_call_id=tool_call_id,
                        name=name,
                        ok=exec_result.ok,
                        result=tool_msg_content,
                        extras=exec_result.extras,
                    ),
                    "completed_tool_result_for_final": (
                        mode in {"planner", "single"}
                        and self._tool_result_completes_action(name, exec_result.ok)
                    ),
                }

            # Only parallelize read-only tools to avoid race conditions.
            read_only_tools = {"read_file", "read_file_outline", "list_directory", "grep_search", "glob"}

            # Execute tasks
            results_to_append = []

            with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
                # We can map read tasks, and sequentialize others.
                futures = {}
                for task in tasks:
                    if cancel_event.is_set():
                        break

                    if task["name"] in read_only_tools:
                        futures[executor.submit(process_task, task)] = task
                    else:
                        # Ensure we wait for all pending reads before a write
                        for fut in concurrent.futures.as_completed(futures):
                            results_to_append.append(fut.result())
                        futures.clear()

                        if cancel_event.is_set():
                            break

                        results_to_append.append(process_task(task))

                # Wait for any remaining reads
                for fut in concurrent.futures.as_completed(futures):
                    results_to_append.append(fut.result())

            # History is not thread-safe. Reorder results by original tool_call_id order and append.
            results_by_id = {r.get("id"): r for r in results_to_append if r is not None}

            if stale_validation_notes:
                note_text = "\n".join(stale_validation_notes)
                self._history.append_user_text(note_text)

            completed_dispatch_for_final = False
            completed_tool_result_for_final = False
            planner_stale_read_files: list[str] = []
            for task in tasks:
                if cancel_event.is_set():
                    self._cleanup_cancelled(on_event)
                    return

                res = results_by_id.get(task["id"])
                if not res:
                    continue

                planner_stale_read_files.extend(
                    str(path) for path in res.get("planner_stale_read_files", [])
                )
                if res.get("blocker"):
                    if planner_stale_read_files:
                        self._history.append_user_text(
                            _planner_stale_read_notice(planner_stale_read_files)
                        )
                    self._append_dispatch_blocker_message(
                        res["result"], str(res.get("blocker_reason", "")), on_event
                    )
                    return
                if res.get("completed_dispatch_for_final"):
                    completed_dispatch_for_final = True
                if res.get("completed_tool_result_for_final"):
                    completed_tool_result_for_final = True
                if res.get("skip"):
                    continue

                if "result_payload" in res:
                    self._history.append_tool_result(task["id"], res["result_payload"])
                    on_event(res["event"])

            if planner_stale_read_files:
                self._history.append_user_text(
                    _planner_stale_read_notice(planner_stale_read_files)
                )

            if _worker_phase_boundary_info is not None:
                worker_phase_boundary_info = _worker_phase_boundary_info
                worker_needs_final_report = True
                continue

            if completed_dispatch_for_final or completed_tool_result_for_final:
                task_completion_context = True
                continue

            # If research completed, stop the loop.
            # The Research Completed card is the final user-facing result.
            if _terminal_dispatch:
                return

    @staticmethod
    def _assistant_message_text(message: dict[str, Any]) -> str:
        content = message.get("content")
        if isinstance(content, str):
            return content
        if isinstance(content, list):
            parts: list[str] = []
            for item in content:
                if isinstance(item, dict) and isinstance(item.get("text"), str):
                    parts.append(item["text"])
            return "\n".join(parts)
        return ""

    @staticmethod
    def _is_completed_worker_result(result: WorkerDispatchResult | None) -> bool:
        if result is None or result.cancelled:
            return False
        if result.needs_followup or result.recoverable or result.phase_boundary:
            return False
        status = infer_outcome_status(result)
        return status in {
            WorkerOutcomeStatus.completed.value,
            WorkerOutcomeStatus.completed_with_caveats.value,
        }

    @staticmethod
    def _terminal_result_completed(info: dict[str, Any] | None) -> bool:
        payload = info.get("_terminal_payload") if isinstance(info, dict) else None
        return isinstance(payload, dict) and payload.get("ok") is True

    @staticmethod
    def _tool_result_completes_action(name: str, ok: bool) -> bool:
        return ok and name in ACTION_COMPLETION_TOOL_NAMES

    @staticmethod
    def _completion_phrase_hits(text: str) -> set[str]:
        lowered = " ".join(str(text or "").lower().split())
        return {
            marker
            for marker in COMPLETION_PHRASE_MARKERS
            if marker in lowered
        }

    @classmethod
    def _is_completion_style_message(cls, text: str) -> bool:
        return bool(cls._completion_phrase_hits(text))

    @classmethod
    def _is_repetitive_completion_final(cls, current: str, previous: str) -> bool:
        current_hits = cls._completion_phrase_hits(current)
        previous_hits = cls._completion_phrase_hits(previous)
        if current_hits and (current_hits & previous_hits):
            return True
        return cls._text_overlap_ratio(current, previous) >= 0.7

    @staticmethod
    def _text_overlap_ratio(left: str, right: str) -> float:
        left_words = set(re.findall(r"[a-z0-9_]+", str(left).lower()))
        right_words = set(re.findall(r"[a-z0-9_]+", str(right).lower()))
        if not left_words or not right_words:
            return 0.0
        return len(left_words & right_words) / max(len(left_words), len(right_words))

    @staticmethod
    def _syntax_repair_paths(
        syntax_repair_required: dict[str, dict[str, Any]],
    ) -> set[str]:
        return {
            _normalize_worker_path(path)
            for path, state in syntax_repair_required.items()
            if not state.get("awaiting_validation")
        }

    @staticmethod
    def _syntax_repair_state_for_path(
        syntax_repair_required: dict[str, dict[str, Any]],
        path: str,
    ) -> dict[str, Any]:
        normalized = _normalize_worker_path(path)
        state = syntax_repair_required.get(normalized)
        if state is not None:
            return state
        for existing_path, existing_state in syntax_repair_required.items():
            if _normalize_worker_path(existing_path) == normalized:
                return existing_state
        return {}

    @staticmethod
    def _set_syntax_repair_state(
        syntax_repair_required: dict[str, dict[str, Any]],
        path: str,
        state: dict[str, Any],
    ) -> dict[str, Any]:
        normalized = _normalize_worker_path(path)
        for existing_path in list(syntax_repair_required):
            if _normalize_worker_path(existing_path) == normalized and existing_path != normalized:
                syntax_repair_required.pop(existing_path, None)
        syntax_repair_required[normalized] = state
        return state

    @staticmethod
    def _pop_syntax_repair_state(
        syntax_repair_required: dict[str, dict[str, Any]],
        path: str,
    ) -> None:
        normalized = _normalize_worker_path(path)
        syntax_repair_required.pop(normalized, None)
        for existing_path in list(syntax_repair_required):
            if _normalize_worker_path(existing_path) == normalized:
                syntax_repair_required.pop(existing_path, None)

    @staticmethod
    def _discard_syntax_validation_path(
        syntax_validation_required: set[str],
        path: str,
    ) -> None:
        normalized = _normalize_worker_path(path)
        syntax_validation_required.discard(normalized)
        for existing_path in set(syntax_validation_required):
            if _normalize_worker_path(existing_path) == normalized:
                syntax_validation_required.discard(existing_path)

    @staticmethod
    def _has_terminal_syntax_failure(
        syntax_repair_required: dict[str, dict[str, Any]],
    ) -> bool:
        return any(state.get("repair_failed") for state in syntax_repair_required.values())

    def _finish_worker_unrecoverable(
        self,
        on_event: EventCallback,
        *,
        failure_class: str,
        error: str,
        details: dict[str, Any] | None = None,
    ) -> None:
        payload = {
            "ok": False,
            "failure_class": failure_class,
            "error": error,
        }
        if details:
            payload["details"] = details
        content = json.dumps(payload, ensure_ascii=False)
        full_message = {
            "role": "assistant",
            "content": content,
            "reasoning_content": None,
        }
        self._history.append_assistant(full_message)
        on_event(ContentDelta(text=content))
        on_event(Done(finish_reason="stop", full_message=full_message))

    @staticmethod
    def _edit_recovery_details(
        edit_fallback_required: dict[str, dict[str, Any]],
        line_range_reread_required: dict[str, dict[str, Any]],
    ) -> dict[str, Any]:
        pending = edit_fallback_required or line_range_reread_required
        if not pending:
            return {}
        path = sorted(pending)[0]
        record = pending[path]
        details: dict[str, Any] = {
            "path": path,
            "tool": record.get("tool") or record.get("name") or "",
            "failure_class": record.get("failure_class") or "",
            "error": record.get("error") or "",
        }
        for key in (
            "operation_index",
            "failed_operation",
            "reason",
            "stale",
            "ambiguous",
            "not_found",
            "candidate_count",
            "candidates",
            "suggested_next_tool",
            "suggested_next_action",
        ):
            if key in record:
                details[key] = record[key]
        return details

    def _run_focused_py_compile(
        self,
        paths: list[str],
    ) -> tuple[bool, str]:
        """Run python -m py_compile on each touched product Python file.

        Returns (all_succeeded, combined_output).
        Uses sys.executable, cwd=workspace root, timeout=30s.
        Normalizes paths safely (backslash/slash, strip leading "./").
        Preserves dot-prefixed directories like .aura.
        """
        if not paths:
            return True, ""
        workspace_root = self._tools.workspace_root
        python_exe = str(preferred_python_for_compile(Path(workspace_root)))
        outputs: list[str] = []
        all_ok = True
        for path in sorted(paths):
            normalized = _normalize_worker_path(path)
            if _is_validation_scratch_path(normalized):
                continue
            full_path = Path(workspace_root) / normalized
            if not full_path.exists():
                outputs.append(f"{normalized}: file not found — cannot py_compile")
                all_ok = False
                continue
            try:
                result = subprocess.run(
                    [python_exe, "-m", "py_compile", str(full_path)],
                    capture_output=True,
                    text=True,
                    timeout=30,
                    cwd=workspace_root,
                )
                if result.returncode != 0:
                    all_ok = False
                    err = result.stderr.strip() or result.stdout.strip() or "py_compile failed"
                    outputs.append(f"{normalized}: {err}")
                else:
                    outputs.append(f"{normalized}: ok")
            except subprocess.TimeoutExpired:
                all_ok = False
                outputs.append(f"{normalized}: timed out after 30s")
            except FileNotFoundError:
                all_ok = False
                outputs.append(f"{normalized}: project Python interpreter not found")
            except OSError as exc:
                all_ok = False
                outputs.append(f"{normalized}: OSError: {exc}")
        combined = "\n".join(outputs)
        return all_ok, combined

    def _emit_auto_py_compile_result(
        self,
        *,
        paths: list[str],
        ok: bool,
        diagnostics: str,
        on_event: EventCallback,
    ) -> None:
        product_paths = [
            _normalize_worker_path(path)
            for path in paths
            if not _is_validation_scratch_path(path)
        ]
        if not product_paths:
            return
        python_exe = quote_command_arg(preferred_python_for_compile(Path(self._tools.workspace_root)))
        command = python_exe + " -m py_compile " + " ".join(product_paths)
        payload = {
            "ok": ok,
            "command": command,
            "exit_code": 0 if ok else 1,
            "output": diagnostics,
            "auto_validation": True,
        }
        content = json.dumps(payload, ensure_ascii=False)
        on_event(
            ToolResult(
                tool_call_id="auto_py_compile",
                name="run_terminal_command",
                ok=ok,
                result=content,
            )
        )

    def _worker_recovery_block(
        self,
        *,
        tool_call_id: str,
        name: str,
        args: dict[str, Any],
        edit_failed_shapes: set[str],
        edit_fallback_required: dict[str, dict[str, Any]],
        recovery_block_counts: dict[str, int],
        line_range_reread_required: dict[str, dict[str, Any]],
        syntax_repair_required: dict[str, dict[str, Any]],
        syntax_validation_required: set[str],
        write_attempts_by_path: dict[str, int],
        worker_file_state: dict[str, dict[str, Any]] | None = None,
        patch_failed_cycles: dict[str, int] | None = None,
    ) -> dict[str, Any] | None:
        raw_path = self._tool_path(name, args)
        path = _normalize_worker_path(raw_path) if raw_path else ""
        worker_file_state = worker_file_state if worker_file_state is not None else {}
        patch_failed_cycles = patch_failed_cycles if patch_failed_cycles is not None else {}
        if name in WORKER_RECOVERY_ALWAYS_ALLOWED:
            return None
        syntax_paths = self._syntax_repair_paths(syntax_repair_required)
        if syntax_paths and not self._syntax_repair_tool_allowed(name, args, syntax_paths):
            target = sorted(syntax_paths)[0]
            state = self._syntax_repair_state_for_path(syntax_repair_required, target)
            repair_failed = bool(state.get("repair_failed"))
            error_msg = (
                f"Python syntax is invalid in {target}. "
                + (
                    "Syntax still fails after one repair attempt."
                    if repair_failed
                    else "Repair that file and pass py_compile before any unrelated tool call."
                )
            )
            diagnostic = state.get("error", "")
            if diagnostic:
                error_msg += f"\n\nDiagnostic output:\n{diagnostic}"
            payload = self._recovery_payload(
                path=target,
                failure_class="syntax_invalid",
                error=error_msg,
                suggested_next_tool="patch_file",
                suggested_next_action=(
                    "Re-read the file, inspect proposed_context if present, then submit one corrected "
                    "patch_file transaction with the current expected_file_hash. Run py_compile after "
                    "the patch is applied."
                ),
                recoverable=not repair_failed,
            )
            self._record_recovery_block(payload, f"syntax:{target}:{name}", recovery_block_counts)
            return self._blocked_tool_result(tool_call_id, name, payload)

        if (
            path
            and name in {"edit_file", "edit_line_range"}
            and write_attempts_by_path.get(path, 0) > 3
        ):
            payload = self._recovery_payload(
                path=path,
                failure_class="edit_mechanics_multi_edit_spin",
                error=(
                    "Multiple failed or unapplied write attempts targeted the same file. "
                    "Use patch_file for this existing-file edit."
                ),
                suggested_next_tool="patch_file",
                suggested_next_action=(
                    "Re-read the file, then submit one patch_file containing all intended hunks."
                ),
            )
            self._record_recovery_block(payload, f"multi-edit-spin:{path}:{name}", recovery_block_counts)
            return self._blocked_tool_result(tool_call_id, name, payload)

        if name == "apply_edit_transaction" and path:
            shape = self._edit_shape_signature(name, args)
            if shape in edit_failed_shapes:
                if f"ambiguous-replace-text:{shape}" in edit_failed_shapes:
                    payload = self._recovery_payload(
                        path=path,
                        failure_class="edit_transaction_ambiguous_symbol",
                        error=(
                            "Repeated ambiguous replace_text_once transaction. "
                            "Do not retry the same exact text shape."
                        ),
                        suggested_next_tool="patch_file",
                        suggested_next_action=(
                            "Re-read the file, then retry patch_file with occurrence or allow_multiple."
                        ),
                        recoverable=False,
                    )
                    self._record_recovery_block(payload, shape, recovery_block_counts)
                    return self._blocked_tool_result(tool_call_id, name, payload)
                payload = self._recovery_payload(
                    path=path,
                    failure_class="edit_mechanics_blocked",
                    error="Repeated apply_edit_transaction failure. Re-read the file and return a concise blocker instead of switching between edit tools.",
                    suggested_next_tool="read_file",
                    suggested_next_action="Re-read the file, then report the typed transaction blocker if the structured operation still cannot be applied safely.",
                    recoverable=False,
                )
                self._record_recovery_block(payload, shape, recovery_block_counts)
                return self._blocked_tool_result(tool_call_id, name, payload)

        if name == "edit_line_range" and path in line_range_reread_required:
            payload = self._recovery_payload(
                path=path,
                failure_class="edit_mechanics_stale_line_range",
                error="Stale line range after a failed edit_line_range. Re-read the file before retrying line-range editing.",
                suggested_next_tool="read_file",
                suggested_next_action="Re-read the file before retrying an edit.",
            )
            self._record_recovery_block(payload, f"line-range-reread:{path}", recovery_block_counts)
            return self._blocked_tool_result(tool_call_id, name, payload)

        if name == "patch_file" and path and patch_failed_cycles.get(path, 0) >= 2:
            payload = self._recovery_payload(
                path=path,
                failure_class="patch_file_repeated_failure",
                error=(
                    "patch_file failed twice on the same file after fresh reads. "
                    "Stop editing this file and report the structured blocker instead of retrying."
                ),
                suggested_next_tool="read_file",
                suggested_next_action=(
                    "Do not retry patch_file for this path in this Worker run. "
                    "Report the repeated patch blocker to the Planner."
                ),
                recoverable=False,
            )
            payload["applied"] = False
            payload["write_outcome"] = "not_applied_edit_mechanics_blocked"
            payload["patch_failed_cycles"] = patch_failed_cycles.get(path, 0)
            return self._blocked_tool_result(tool_call_id, name, payload)

        if name == "patch_file" and path in edit_fallback_required:
            prior = edit_fallback_required[path]
            failure_class = str(prior.get("failure_class") or "patch_hunk_not_found")
            if failure_class == "patch_file_hash_mismatch":
                suggested_next_action = (
                    "Re-read the file with read_file or read_file_range, then retry patch_file once "
                    "with expected_file_hash set to the new content_hash."
                )
            else:
                suggested_next_action = "Re-read the file, then retry patch_file once with current exact text."
            payload = self._recovery_payload(
                path=path,
                failure_class=failure_class,
                error="Previous patch_file failed. Re-read the file before retrying the patch.",
                suggested_next_tool="read_file",
                suggested_next_action=suggested_next_action,
            )
            payload["previous_error"] = prior.get("error", "")
            self._record_recovery_block(payload, f"patch-reread:{path}", recovery_block_counts)
            if recovery_block_counts.get(f"patch-reread:{path}", 0) >= 2:
                payload["recoverable"] = False
                payload["error"] = (
                    "patch_file failed repeatedly on the same file after re-reading. "
                    "Stop editing this file and report the structured blocker instead of retrying."
                )
            return self._blocked_tool_result(tool_call_id, name, payload)

        if name == "patch_file" and path:
            blocked = self._worker_patch_file_state_block(
                tool_call_id=tool_call_id,
                name=name,
                path=path,
                args=args,
                worker_file_state=worker_file_state,
            )
            if blocked is not None:
                return blocked

        if name in ("edit_file", "edit_symbol") and path in edit_fallback_required:
            prior = edit_fallback_required[path]
            block_key = self._edit_shape_signature(name, args)
            payload = self._recovery_payload(
                path=path,
                failure_class=str(prior.get("failure_class") or self._default_edit_failure_class(name)),
                error="Repeated failed edit tactic. Do not retry this edit shape. Re-read the file and use patch_file for existing-file code changes.",
                suggested_next_tool="patch_file",
                suggested_next_action="Use read_file/read_file_outline, then submit one patch_file with exact hunks.",
            )
            payload["previous_error"] = prior.get("error", "")
            self._record_recovery_block(payload, block_key, recovery_block_counts)
            return self._blocked_tool_result(tool_call_id, name, payload)

        if name in ("edit_file", "edit_symbol", "edit_line_range"):
            shape = self._edit_shape_signature(name, args)
            if shape in edit_failed_shapes:
                payload = self._recovery_payload(
                    path=path,
                    failure_class=self._default_edit_failure_class(name),
                    error="Repeated failed edit tactic. Do not retry this edit shape. Re-read the file and use patch_file for existing-file code changes.",
                    suggested_next_tool="patch_file",
                    suggested_next_action="Use read_file/read_file_outline, then submit one patch_file with exact hunks.",
                )
                self._record_recovery_block(payload, shape, recovery_block_counts)
                return self._blocked_tool_result(tool_call_id, name, payload)

        return None

    def _worker_patch_file_state_block(
        self,
        *,
        tool_call_id: str,
        name: str,
        path: str,
        args: dict[str, Any],
        worker_file_state: dict[str, dict[str, Any]],
    ) -> dict[str, Any] | None:
        if not self._worker_path_is_existing_file(path):
            return None

        expected_hash = args.get("expected_file_hash")
        if not isinstance(expected_hash, str) or not expected_hash:
            payload = self._recovery_payload(
                path=path,
                failure_class="patch_file_missing_expected_hash",
                error=(
                    "Worker patch_file on an existing file requires expected_file_hash "
                    "from the latest successful read."
                ),
                suggested_next_tool="read_file",
                suggested_next_action=(
                    "Read the file, then retry patch_file with expected_file_hash "
                    "set to the returned content_hash."
                ),
            )
            payload["applied"] = False
            payload["write_outcome"] = "not_applied_edit_mechanics_blocked"
            return self._blocked_tool_result(tool_call_id, name, payload)

        state = self._worker_file_state_for_path(worker_file_state, path)
        known_hash = str(state.get("content_hash") or "") if state else ""
        if not state or known_hash != expected_hash or not state.get("fresh_for_patch"):
            payload = self._recovery_payload(
                path=path,
                failure_class="patch_file_hash_mismatch",
                error=(
                    "patch_file expected_file_hash does not match the Worker's "
                    "latest successful read for this file."
                ),
                suggested_next_tool="read_file",
                suggested_next_action=(
                    "Re-read the file with read_file or read_file_range, then retry patch_file once "
                    "with expected_file_hash set to the new content_hash."
                ),
            )
            payload["applied"] = False
            payload["write_outcome"] = "not_applied_edit_mechanics_blocked"
            payload["recoverable"] = True
            payload["stale"] = True
            payload["expected_file_hash"] = expected_hash
            if known_hash:
                payload["latest_read_content_hash"] = known_hash
            if state and state.get("last_read_tool"):
                payload["last_read_tool"] = state.get("last_read_tool")
            return self._blocked_tool_result(tool_call_id, name, payload)

        return None

    def _worker_path_is_existing_file(self, path: str) -> bool:
        try:
            root = Path(self._tools.workspace_root).resolve()
            candidate = Path(str(path).lstrip("/\\"))
            if not candidate.is_absolute():
                candidate = root / candidate
            resolved = candidate.resolve()
            resolved.relative_to(root)
            return resolved.is_file()
        except (OSError, ValueError):
            return False

    @staticmethod
    def _worker_file_state_for_path(
        worker_file_state: dict[str, dict[str, Any]],
        path: str,
    ) -> dict[str, Any] | None:
        normalized_candidates = {
            _normalize_worker_path(path),
            _normalize_worker_path(str(path).lstrip("/\\")),
        }
        for normalized in normalized_candidates:
            state = worker_file_state.get(normalized)
            if state is not None:
                return state
        for existing_path, existing_state in worker_file_state.items():
            if _normalize_worker_path(existing_path) in normalized_candidates:
                return existing_state
        return None

    def _update_worker_recovery_state(
        self,
        *,
        name: str,
        args: dict[str, Any],
        ok: bool,
        content: str,
        edit_failed_shapes: set[str],
        edit_fallback_required: dict[str, dict[str, Any]],
        line_range_reread_required: dict[str, dict[str, Any]],
        syntax_repair_required: dict[str, dict[str, Any]],
        syntax_validation_required: set[str],
        write_attempts_by_path: dict[str, int],
        worker_file_state: dict[str, dict[str, Any]] | None = None,
        patch_failed_cycles: dict[str, int] | None = None,
    ) -> str:
        worker_file_state = worker_file_state if worker_file_state is not None else {}
        patch_failed_cycles = patch_failed_cycles if patch_failed_cycles is not None else {}
        parsed = self._parse_tool_payload(content)
        self._record_reads_for_recovery(
            name,
            args,
            parsed,
            line_range_reread_required,
            edit_fallback_required,
            worker_file_state,
        )
        raw_path = self._tool_path(name, args, parsed)
        path = _normalize_worker_path(raw_path) if raw_path else ""
        if name in WRITE_TOOLS and path and (
            not ok
            or (
                isinstance(parsed, dict)
                and parsed.get("ok")
                and parsed.get("applied") is False
            )
        ):
            write_attempts_by_path[path] = write_attempts_by_path.get(path, 0) + 1

        if ok:
            if name in WRITE_TOOLS and path:
                self._pop_normalized_recovery_key(edit_fallback_required, path)
                self._pop_normalized_recovery_key(line_range_reread_required, path)
                self._pop_normalized_key(worker_file_state, path)
                self._pop_normalized_key(patch_failed_cycles, path)
                if self._is_python_path(path) and not _is_validation_scratch_path(path):
                    syntax_validation_required.add(path)
                state = self._syntax_repair_state_for_path(syntax_repair_required, path)
                if state:
                    state["repair_attempted"] = True
                    state["awaiting_validation"] = True
                    self._set_syntax_repair_state(syntax_repair_required, path, state)
                    if not _is_validation_scratch_path(path):
                        syntax_validation_required.add(path)
            return content

        if name in ("edit_file", "edit_symbol", "edit_line_range", "apply_edit_transaction"):
            edit_failed_shapes.add(self._edit_shape_signature(name, args))

        if not isinstance(parsed, dict):
            return content

        failure_class = str(parsed.get("failure_class", ""))
        shape = self._edit_shape_signature(name, args)
        if path and failure_class in EDIT_TRANSACTION_FAILURE_CLASSES:
            parsed.setdefault("applied", False)
            parsed.setdefault("write_outcome", "not_applied_edit_mechanics_blocked")
            parsed["recoverable"] = False
            parsed.pop("suggested_tool", None)
            parsed.pop("suggested_next_tool", None)
            if (
                failure_class == "edit_transaction_ambiguous_symbol"
                and self._has_replace_text_once_operation(args)
            ):
                edit_failed_shapes.add(f"ambiguous-replace-text:{shape}")
                parsed["suggested_next_action"] = (
                    "Re-read the file, then retry patch_file with occurrence "
                    "or allow_multiple."
                )
            else:
                parsed["suggested_next_action"] = (
                    "Transaction could not be applied safely. Re-read the file "
                    "and report this typed blocker if the operation is still not applicable."
                )
            content = json.dumps(parsed, ensure_ascii=False)
        elif path and name != "patch_file" and failure_class in EDIT_MECHANICS_FAILURE_CLASSES:
            parsed.setdefault("applied", False)
            parsed.setdefault("write_outcome", "not_applied_edit_mechanics_blocked")
            parsed.setdefault("tool", name)
            edit_fallback_required[path] = parsed
            parsed["recoverable"] = True
            parsed["suggested_next_tool"] = "patch_file"
            parsed["suggested_next_action"] = "Do not retry this low-level edit shape. Re-read the file and submit one patch_file for existing-file code changes."
            content = json.dumps(parsed, ensure_ascii=False)
        elif path and failure_class == "edit_mechanics_stale_line_range":
            parsed.setdefault("applied", False)
            parsed.setdefault("write_outcome", "not_applied_edit_mechanics_blocked")
            parsed.setdefault("tool", name)
            line_range_reread_required[path] = parsed
            parsed["recoverable"] = True
            parsed["suggested_next_tool"] = "read_file"
            parsed["suggested_next_action"] = "Re-read the file before retrying an edit."
            content = json.dumps(parsed, ensure_ascii=False)
        elif path and name == "patch_file" and failure_class in {"patch_hunk_not_found", "patch_hunk_ambiguous", "patch_file_hash_mismatch"}:
            parsed.setdefault("applied", False)
            parsed.setdefault("write_outcome", "not_applied_edit_mechanics_blocked")
            parsed.setdefault("tool", name)
            failed_cycles = patch_failed_cycles.get(path, 0) + 1
            patch_failed_cycles[path] = failed_cycles
            parsed["patch_failed_cycles"] = failed_cycles
            parsed["stale"] = True
            parsed["suggested_next_tool"] = "read_file"
            if failure_class == "patch_file_hash_mismatch":
                parsed["suggested_next_action"] = (
                    "Re-read the file with read_file or read_file_range, then retry patch_file once "
                    "with expected_file_hash set to the returned content_hash."
                )
            else:
                parsed["suggested_next_action"] = (
                    "Re-read the file, then retry patch_file once with current exact text "
                    "and expected_file_hash set to the returned content_hash."
                )
            if failed_cycles >= 2:
                parsed["recoverable"] = False
                parsed["failure_class"] = "patch_file_repeated_failure"
                parsed["error"] = (
                    "patch_file failed twice on the same file after a fresh read. "
                    "Stop editing this file and report the structured blocker instead of retrying."
                )
                self._pop_normalized_recovery_key(edit_fallback_required, path)
            else:
                edit_fallback_required[path] = parsed
                parsed["recoverable"] = True
            content = json.dumps(parsed, ensure_ascii=False)
        elif path and failure_class == "syntax_invalid":
            parsed.setdefault("applied", False)
            parsed.setdefault("write_outcome", "not_applied_craft_rejected")
            state = self._syntax_repair_state_for_path(syntax_repair_required, path)
            if not state:
                state = {"failed_repairs": 0}
            self._set_syntax_repair_state(syntax_repair_required, path, state)
            state["awaiting_validation"] = False
            if name in WRITE_TOOLS:
                state["failed_repairs"] = int(state.get("failed_repairs", 0)) + 1
            parsed["suggested_next_tool"] = "patch_file"
            parsed["suggested_next_action"] = (
                "Re-read the file, inspect proposed_context if present, then submit one corrected "
                "patch_file transaction with the current expected_file_hash. Run py_compile after "
                "the patch is applied."
            )
            if int(state.get("failed_repairs", 0)) > 1:
                parsed["recoverable"] = False
                parsed["error"] = "Syntax repair failed after one repair attempt. " + str(parsed.get("error", ""))
            content = json.dumps(parsed, ensure_ascii=False)

        return content

    @staticmethod
    def _is_py_compile_error(output: str) -> bool:
        """Check if output contains Python-level error markers (py_compile actually ran)."""
        return bool(
            "SyntaxError" in output
            or re.search(r'\bFile "', output)
            or "NameError" in output
            or "IndentationError" in output
            or "TypeError:" in output
        )

    @staticmethod
    def _is_shell_failure(output: str) -> bool:
        """Check if output indicates a shell-level failure (command never reached py_compile)."""
        shell_markers = [
            "cannot find the path specified",
            "not recognized as an internal or external command",
            "No such file or directory",
            "command not found",
            "not found",
        ]
        output_lower = output.lower()
        for marker in shell_markers:
            if marker in output_lower:
                return True
        first_lines = output.split("\n")[:3]
        for line in first_lines:
            stripped = line.strip()
            if stripped.startswith("cd:") or stripped.startswith("chdir:"):
                return True
        return False

    def _update_syntax_state_from_terminal(
        self,
        *,
        args: dict[str, Any],
        loop_info: dict[str, Any] | None,
        syntax_repair_required: dict[str, dict[str, Any]],
        syntax_validation_required: set[str],
        stale_validation_notes: list[str] | None = None,
    ) -> None:
        payload = loop_info.get("_terminal_payload") if isinstance(loop_info, dict) else None
        if not isinstance(payload, dict):
            return
        command = str(payload.get("command") or args.get("command") or "")
        workspace_root = Path(self._tools.workspace_root)
        targets = [
            _normalize_worker_path(path)
            for path in self._py_compile_targets(command)
            if not _is_validation_scratch_path(path)
            and not (
                "/" not in path
                and not (workspace_root / path).exists()
            )
        ]
        if not targets:
            return
        if payload.get("ok"):
            for path in targets:
                state = self._syntax_repair_state_for_path(syntax_repair_required, path)
                if state and state.get("awaiting_validation") is False:
                    if stale_validation_notes is not None:
                        stale_validation_notes.append(
                            "Stale validation cleared: "
                            f"py_compile passed for {path} after a prior craft-gate rejection."
                        )
                self._pop_syntax_repair_state(syntax_repair_required, path)
                self._discard_syntax_validation_path(syntax_validation_required, path)
            return
        # Shell-level failures (cd prefix, command not found) should not trigger
        # Python syntax repair — the command never reached py_compile.
        output = str(payload.get("output") or "")
        if not self._is_py_compile_error(output) and self._is_shell_failure(output):
            return
        for path in targets:
            prior = self._syntax_repair_state_for_path(syntax_repair_required, path)
            failed_after_repair = bool(
                prior.get("repair_attempted") or prior.get("awaiting_validation")
            )
            self._set_syntax_repair_state(syntax_repair_required, path, {
                "error": payload.get("output", ""),
                "failed_repairs": int(prior.get("failed_repairs", 0)) + (1 if failed_after_repair else 0),
                "repair_failed": failed_after_repair,
            })
            self._discard_syntax_validation_path(syntax_validation_required, path)

    @staticmethod
    def _record_reads_for_recovery(
        name: str,
        args: dict[str, Any],
        parsed: Any,
        line_range_reread_required: dict[str, dict[str, Any]],
        edit_fallback_required: dict[str, dict[str, Any]],
        worker_file_state: dict[str, dict[str, Any]] | None = None,
    ) -> None:
        if name == "read_file":
            path = str(args.get("path") or (parsed.get("path") if isinstance(parsed, dict) else ""))
            if (
                path
                and isinstance(parsed, dict)
                and parsed.get("ok") is True
                and parsed.get("truncated") is not True
            ):
                recorded = ConversationManager._record_worker_file_state(
                    worker_file_state,
                    path,
                    parsed,
                    "read_file",
                )
                if recorded:
                    ConversationManager._pop_normalized_recovery_key(line_range_reread_required, path)
                    ConversationManager._pop_normalized_recovery_key(edit_fallback_required, path)
        elif name == "read_files":
            files = parsed.get("files") if isinstance(parsed, dict) else None
            if isinstance(files, dict):
                for path_key, result in files.items():
                    if not isinstance(result, dict):
                        continue
                    path = str(result.get("path") or path_key)
                    if result.get("ok") is True and result.get("truncated") is not True:
                        recorded = ConversationManager._record_worker_file_state(
                            worker_file_state,
                            path,
                            result,
                            "read_files",
                        )
                        if recorded:
                            ConversationManager._pop_normalized_recovery_key(line_range_reread_required, path)
                            ConversationManager._pop_normalized_recovery_key(edit_fallback_required, path)
        elif name == "read_file_range":
            path = str(args.get("path") or (parsed.get("path") if isinstance(parsed, dict) else ""))
            if path and isinstance(parsed, dict) and parsed.get("ok") is True:
                recorded = ConversationManager._record_worker_file_state(
                    worker_file_state,
                    path,
                    parsed,
                    "read_file_range",
                )
                if recorded:
                    prior = ConversationManager._normalized_state_value(edit_fallback_required, path)
                    if (
                        isinstance(prior, dict)
                        and prior.get("failure_class") == "patch_file_hash_mismatch"
                    ):
                        ConversationManager._pop_normalized_recovery_key(edit_fallback_required, path)

    @staticmethod
    def _record_worker_file_state(
        worker_file_state: dict[str, dict[str, Any]] | None,
        path: str,
        result: dict[str, Any],
        tool_name: str,
    ) -> bool:
        if worker_file_state is None:
            return False
        content_hash = result.get("content_hash")
        file_size = result.get("file_size")
        if not isinstance(content_hash, str) or not content_hash:
            return False
        if not isinstance(file_size, int):
            return False
        normalized = _normalize_worker_path(str(result.get("path") or path))
        worker_file_state[normalized] = {
            "content_hash": content_hash,
            "file_size": file_size,
            "truncated": bool(result.get("truncated", False)),
            "last_read_tool": tool_name,
            "fresh_for_patch": result.get("truncated") is not True,
        }
        return True

    @staticmethod
    def _pop_normalized_recovery_key(
        state: dict[str, dict[str, Any]],
        path: str,
    ) -> None:
        ConversationManager._pop_normalized_key(state, path)

    @staticmethod
    def _normalized_state_value(
        state: dict[str, Any],
        path: str,
    ) -> Any:
        normalized = _normalize_worker_path(path)
        if normalized in state:
            return state[normalized]
        for existing_path, value in state.items():
            if _normalize_worker_path(existing_path) == normalized:
                return value
        return None

    @staticmethod
    def _pop_normalized_key(
        state: dict[str, Any],
        path: str,
    ) -> None:
        normalized = _normalize_worker_path(path)
        state.pop(normalized, None)
        for existing_path in list(state):
            if _normalize_worker_path(existing_path) == normalized:
                state.pop(existing_path, None)

    @staticmethod
    def _syntax_repair_tool_allowed(
        name: str,
        args: dict[str, Any],
        syntax_paths: set[str],
    ) -> bool:
        if name in SYNTAX_REPAIR_ALWAYS_ALLOWED:
            return True
        if name in WRITE_TOOLS:
            return _normalize_worker_path(str(args.get("path", ""))) in syntax_paths
        if name == "run_terminal_command":
            command = str(args.get("command", ""))
            # py_compile targeting a broken file
            if re.search(
                r"(?i)(?:^|[;&|]\s*)"
                r"(?:(?:\"[^\"]*python3?(?:\.exe)?\")|(?:'[^']*python3?(?:\.exe)?')|"
                r"(?:[A-Za-z]:)?[A-Za-z0-9_./\\\-]*python3?(?:\.exe)?|py)"
                r"\s+-m\s+py_compile\b",
                command,
            ):
                targets = ConversationManager._py_compile_targets(command)
                return bool(targets) and any(target in syntax_paths for target in targets)
            # pytest targeting a broken file (substring check)
            if re.search(
                r"(?i)(?:^|[;&|]\s*)"
                r"(?:(?:\"[^\"]*python3?(?:\.exe)?\")|(?:'[^']*python3?(?:\.exe)?')|"
                r"(?:[A-Za-z]:)?[A-Za-z0-9_./\\\-]*python3?(?:\.exe)?|py)"
                r"\s+-m\s+pytest\b",
                command,
            ):
                return any(path in command for path in syntax_paths)
            return False
        return False

    @staticmethod
    def _py_compile_targets(command: str) -> list[str]:
        if "py_compile" not in command:
            return []
        matches = re.findall(
            r"(?<![\w.-])([A-Za-z0-9_./\\:\-]+\.py)(?![\w.-])",
            command,
        )
        targets: list[str] = []
        for match in matches:
            target = _normalize_worker_path(match)
            if target.endswith("py_compile.py"):
                continue
            targets.append(target)
        return targets

    @staticmethod
    def _normalize_py_compile_path(raw: str) -> str:
        p = raw.strip().replace("\\", "/")
        if p.startswith("./"):
            p = p[2:]
        return p

    @staticmethod
    def _is_python_path(path: str) -> bool:
        return path.replace("\\\\", "/").endswith(".py")

    @staticmethod
    def _tool_path(name: str, args: dict[str, Any], parsed: Any = None) -> str:
        if isinstance(parsed, dict):
            value = parsed.get("path") or parsed.get("rel_path")
            if isinstance(value, str) and value:
                return value
        value = args.get("path")
        return str(value) if value is not None else ""

    @staticmethod
    def _edit_shape_signature(name: str, args: dict[str, Any]) -> str:
        path = str(args.get("path", ""))
        if name == "edit_file":
            raw = str(args.get("old_str", ""))
            marker = hashlib.sha256(raw.encode("utf-8", errors="replace")).hexdigest()
        elif name == "edit_symbol":
            marker = "|".join(
                str(args.get(key, ""))
                for key in ("symbol_type", "class_name", "symbol_name")
            )
        elif name == "edit_line_range":
            marker = f"{args.get('start_line')}:{args.get('end_line')}"
        elif name == "apply_edit_transaction":
            marker = ConversationManager._transaction_shape_marker(args)
        else:
            marker = json.dumps(args, sort_keys=True, ensure_ascii=False)
        return json.dumps({"tool": name, "path": path, "shape": marker}, sort_keys=True)

    @staticmethod
    def _transaction_shape_marker(args: dict[str, Any]) -> list[dict[str, Any]]:
        operations = args.get("operations")
        if not isinstance(operations, list):
            return []
        markers: list[dict[str, Any]] = []
        for op in operations:
            if not isinstance(op, dict):
                markers.append({"op": "invalid"})
                continue
            kind = str(op.get("op") or op.get("type") or "")
            marker: dict[str, Any] = {
                "op": kind,
                "symbol_type": op.get("symbol_type"),
                "symbol_name": op.get("symbol_name") or op.get("function_name") or op.get("method_name") or op.get("name"),
                "class_name": op.get("class_name"),
                "occurrence": op.get("occurrence"),
                "allow_multiple": op.get("allow_multiple"),
            }
            for source_key, marker_key in (
                ("old", "old_hash"),
                ("new", "new_hash"),
                ("text", "text_hash"),
                ("new_definition", "new_definition_hash"),
                ("content", "content_hash"),
                ("start_marker", "start_marker_hash"),
                ("end_marker", "end_marker_hash"),
            ):
                value = op.get(source_key)
                if isinstance(value, str):
                    marker[marker_key] = hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest()
            markers.append({k: v for k, v in marker.items() if v not in (None, "")})
        return markers

    @staticmethod
    def _has_replace_text_once_operation(args: dict[str, Any]) -> bool:
        operations = args.get("operations")
        if not isinstance(operations, list):
            return False
        return any(
            isinstance(op, dict)
            and str(op.get("op") or op.get("type") or "") == "replace_text_once"
            for op in operations
        )

    @staticmethod
    def _default_edit_failure_class(name: str) -> str:
        if name == "edit_symbol":
            return "edit_mechanics_symbol_not_found"
        if name == "edit_line_range":
            return "edit_mechanics_stale_line_range"
        return "edit_mechanics_old_str_not_found"

    @staticmethod
    def _parse_tool_payload(content: str) -> Any:
        try:
            return json.loads(content)
        except (TypeError, json.JSONDecodeError):
            return None

    @staticmethod
    def _recovery_payload(
        *,
        path: str,
        failure_class: str,
        error: str,
        suggested_next_tool: str,
        suggested_next_action: str,
        recoverable: bool = True,
    ) -> dict[str, Any]:
        return {
            "ok": False,
            "path": path,
            "rel_path": path,
            "error": error,
            "failure_class": failure_class,
            "recoverable": recoverable,
            "internal_recovery_steer": True,
            "suggested_tool": suggested_next_tool,
            "suggested_next_tool": suggested_next_tool,
            "suggested_next_action": suggested_next_action,
        }

    @staticmethod
    def _record_recovery_block(
        payload: dict[str, Any],
        key: str,
        recovery_block_counts: dict[str, int],
    ) -> None:
        count = recovery_block_counts.get(key, 0) + 1
        recovery_block_counts[key] = count
        payload["repeated_blocks"] = count

    @staticmethod
    def _blocked_tool_result(tool_call_id: str, name: str, payload: dict[str, Any]) -> dict[str, Any]:
        content = json.dumps(payload, ensure_ascii=False)
        return {
            "id": tool_call_id,
            "result_payload": content,
            "event": ToolResult(
                tool_call_id=tool_call_id,
                name=name,
                ok=False,
                result=content,
            ),
        }

    def _append_limit_tool_result(
        self,
        tool_call_id: str,
        name: str,
        info: dict[str, Any],
        on_event: EventCallback,
    ) -> None:
        payload = limit_reached_payload(info)
        self._history.append_tool_result(tool_call_id, payload)
        on_event(
            ToolResult(
                tool_call_id=tool_call_id,
                name=name,
                ok=False,
                result=payload,
                extras={
                    "limit_reached": bool(info.get("limit_reached")),
                    "recoverable": bool(info.get("recoverable")),
                    "phase_boundary": bool(info.get("phase_boundary")),
                    "reason": str(info.get("reason", "")),
                },
            )
        )

    def _classify_failed_worker_dispatch(
        self,
        *,
        args: dict[str, Any],
        result: WorkerDispatchResult,
        failures: dict[str, int],
        failed_attempts: int,
    ) -> dict[str, Any]:
        """Record a failed dispatch and decide whether the planner may continue."""
        if self._is_worker_internal_error(result):
            return {"counts_as_attempt": False, "blocker_reason": "internal"}

        if not self._failed_dispatch_allows_planner_continuation(result):
            return {"counts_as_attempt": False, "blocker_reason": "failed"}

        signature = self._worker_dispatch_failure_signature(args, result)
        repeated_count = failures.get(signature, 0) + 1
        failures[signature] = repeated_count

        if repeated_count >= 2:
            return {"counts_as_attempt": True, "blocker_reason": "repeated"}

        if failed_attempts + 1 >= MAX_WORKER_REDISPATCHES_PER_USER_TURN:
            return {"counts_as_attempt": True, "blocker_reason": "limit"}

        return {"counts_as_attempt": True, "blocker_reason": ""}

    @staticmethod
    def _failed_dispatch_allows_planner_continuation(
        result: WorkerDispatchResult,
    ) -> bool:
        if result.ok or result.cancelled:
            return False
        if result.extras.get("dispatch_spec_rejected"):
            return True
        if result.mismatch is not None:
            return True
        if result.extras.get("planner_resolution_needed"):
            return True
        if infer_outcome_status(result) == WorkerOutcomeStatus.needs_planner_resolution.value:
            return True
        return bool(result.needs_followup or result.recoverable or result.phase_boundary)

    @staticmethod
    def _is_worker_internal_error(result: WorkerDispatchResult) -> bool:
        return bool(
            result.extras.get("worker_internal_error")
            or result.extras.get("dispatch_internal_error")
        )

    def _worker_dispatch_failure_signature(
        self,
        args: dict[str, Any],
        result: WorkerDispatchResult,
    ) -> str:
        spec = {
            "goal": str(args.get("goal", "")),
            "files": [str(item) for item in args.get("files", [])]
            if isinstance(args.get("files"), list)
            else [],
            "spec": str(args.get("spec", "")),
            "acceptance": str(args.get("acceptance", "")),
            "summary": str(args.get("summary", "")),
        }
        payload = {
            "spec": spec,
            "error": self._worker_dispatch_error_signature(result),
        }
        return json.dumps(payload, sort_keys=True, ensure_ascii=False)

    @staticmethod
    def _worker_dispatch_error_signature(result: WorkerDispatchResult) -> str:
        extras = result.extras or {}
        if extras.get("dispatch_spec_rejected"):
            errors = extras.get("quality_errors")
            if isinstance(errors, list):
                return "dispatch_spec_rejected:" + "|".join(str(e) for e in errors)
            return "dispatch_spec_rejected"
        if extras.get("worker_internal_error"):
            return "worker_internal_error"

        parts: list[str] = []
        if result.followup_reason:
            parts.append(f"reason:{result.followup_reason}")
        for key in ("errors", "caveats"):
            values = extras.get(key)
            if isinstance(values, list) and values:
                parts.append(
                    f"{key}:"
                    + "|".join(
                        " ".join(str(value).split())[:160] for value in values[:3]
                    )
                )
        if result.needs_followup:
            parts.append("needs_followup")
        if result.recoverable:
            parts.append("recoverable")
        if result.phase_boundary:
            parts.append("phase_boundary")
        if result.mismatch is not None:
            parts.extend([
                result.mismatch.kind,
                result.mismatch.requested,
                result.mismatch.observed,
                result.mismatch.question_for_planner,
            ])
        if not parts:
            parts.append(" ".join(result.summary.split())[:240])
        return ";".join(parts)

    @staticmethod
    def _is_recoverable_phase_boundary(info: dict[str, Any] | None) -> bool:
        return bool(info and info.get("recoverable") and info.get("phase_boundary"))

    def _append_dispatch_blocker_message(
        self,
        result: WorkerDispatchResult,
        reason: str,
        on_event: EventCallback,
    ) -> None:
        if reason == "internal":
            message = (
                "Harness error due to an internal Worker exception. "
                "I stopped automatic redispatch to avoid repeating the same handoff."
            )
        elif reason == "repeated":
            if result.extras.get("dispatch_spec_rejected"):
                message = (
                    "Plan incomplete — missing required dispatch details. "
                    "The same Worker handoff was rejected twice, so I stopped automatic redispatch."
                )
            elif result.mismatch is not None or result.extras.get("planner_resolution_needed"):
                message = (
                    "The same Worker mismatch was returned twice. I stopped automatic redispatch "
                    "so the Planner handoff can change."
                )
            else:
                message = (
                    "The same Worker dispatch failed twice with the same result. "
                    "I stopped automatic redispatch so the plan can be corrected first."
                )
        elif reason == "limit":
            message = (
                "Worker dispatch did not complete after "
                f"{MAX_WORKER_REDISPATCHES_PER_USER_TURN} failed attempts this turn. "
                "I stopped automatic redispatch so the next handoff can change."
            )
        else:
            message = (
                "Harness error. I stopped automatic redispatch so the failure can be addressed first."
            )
        on_event(ContentDelta(text=message))
        full_message = {
            "role": "assistant",
            "content": message,
            "reasoning_content": None,
        }
        self._history.append_assistant(full_message)
        on_event(Done(finish_reason="stop", full_message=full_message))

    def _apply_loop_detection(
        self,
        *,
        mode: str,
        name: str,
        args: dict[str, Any],
        ok: bool,
        result_payload: str,
    ) -> dict[str, Any]:
        """Track repetitive failures and return annotated content plus metadata."""
        observed = self._loop_detector.observe(
            mode=mode,
            tool_name=name,
            args=args,
            ok=ok,
            content=result_payload,
        )
        return {"content": observed.content, "info": observed.info}

    def _cleanup_cancelled(self, on_event: EventCallback) -> None:
        """Call this when a turn is cancelled while waiting for model or tool.
        Ensure history doesn't contain an assistant message with pending tool calls
        that haven't been followed by tool result messages.
        """
        if not self._history.messages:
            on_event(ApiError(status_code=None, message="Cancelled."))
            return

        # We look for the MOST RECENT assistant message.
        # If it has tool calls that are missing results, we MUST clean it up.
        for i in range(len(self._history.messages) - 1, -1, -1):
            msg = self._history.messages[i]
            if msg.get("role") == "user":
                # If we hit a user message first, it means the turn was cancelled
                # before the assistant even started responding.
                break

            if msg.get("role") == "assistant":
                tool_calls = msg.get("tool_calls")
                if tool_calls:
                    call_ids = {tc["id"] for tc in tool_calls}
                    # Look at messages following this one.
                    for j in range(i + 1, len(self._history.messages)):
                        m = self._history.messages[j]
                        if m.get("role") == "tool":
                            call_ids.discard(m.get("tool_call_id"))

                    if call_ids:
                        # Incomplete! Truncate history back to BEFORE this assistant message.
                        # We find the user message that preceded it.
                        user_idx = -1
                        for k in range(i - 1, -1, -1):
                            if self._history.messages[k].get("role") == "user":
                                user_idx = k
                                break
                        if user_idx != -1:
                            self._history.truncate_after(user_idx + 1)
                        else:
                            self._history.truncate_after(i)
                elif not msg.get("content") and not msg.get("reasoning_content"):
                    # Empty assistant message — strip it.
                    self._history.truncate_after(i)
                break

        on_event(ApiError(status_code=None, message="Cancelled."))


__all__ = [
    "ConversationManager",
    "ApprovalCallback",
    "ApprovalDecision",
    "ApprovalRequest",
    "EventCallback",
    "Event",
    "ReasoningDelta",
    "ContentDelta",
    "ToolCallStart",
    "ToolCallArgsDelta",
    "ToolCallEnd",
    "Usage",
    "Done",
    "ApiError",
    "ToolResult",
    "WorkerDispatchRequested",
    "TerminalOutput",
    "DispatchCallback",
    "WorkerDispatchRequest",
    "WorkerDispatchResult",
]
