# agent_worker.py - Background thread and helpers for agent execution. Used by aiagent.
from __future__ import annotations

import base64
import json
import logging
import os
import re
import threading
import time
from threading import Thread
from activity_bus import emit_activity_enriched
from run_context import set_run_context, clear_run_context, sanitize_run_id

from utils.agent_loop_best_effort import best_effort, best_effort_value

from tools import (
    set_input_status_callback,
    set_live_view_callback,
    set_chat_event_callback,
    set_plugin_stream_callback,
    set_tool_stop_check_callback,
)
from agent_loop import run_raw_agent, _log_ts

_log = logging.getLogger(__name__)


def _extract_output_path_from_tool_result(tool_name, content):
    """Extract output file path from tool result text for ocr_to_file, write_text_file, save_to_file, Kraken."""
    if not content or not tool_name:
        return None
    name = (tool_name or "").lower()
    if name == "run_web_script":
        import re

        m = re.search(r"^run_dir:\s*(\S+)", str(content or ""), re.MULTILINE)
        if m:
            return m.group(1).strip()
        for line in str(content or "").splitlines():
            ls = line.strip()
            if ls.startswith("- ") and os.path.isabs(ls[2:].strip()):
                return ls[2:].strip()
        return None
    if name == "write_web_act_file":
        import re

        m = re.search(r"Wrote \d+ chars to (\S+)", str(content or ""))
        if m:
            return m.group(1).strip()
        return None
    if name not in (
        "ocr_scanned_pdf_to_file",
        "ocr_complex_document_to_file",
        "ocr_with_kraken_api",
        "ocr_with_kraken_api_to_file",
        "write_text_file",
        "save_to_file",
        "append_to_text_file",
    ):
        return None
    import re
    # "It has been saved to:\n/path" (Kraken/OCR auto-save message)
    m = re.search(r"has been saved to:\s*\n([^\s\n]+)", content, re.I)
    if m:
        return m.group(1).strip()
    # "Kraken OCR saved to:\n/path" or "saved to:\n/path"
    m = re.search(r"(?:Kraken OCR )?saved to:\s*\n([^\s\n]+)", content, re.I)
    if m:
        return m.group(1).strip()
    # "Appended N pages (...) to /path/to/file.md. File size now N bytes."
    m = re.search(r"\s+to\s+([^\n]+?)\s*\.\s*File\s", content)
    if m:
        return m.group(1).strip()
    # "Saved to /path/to/file" or "Written to /path"
    m = re.search(r"(?:saved|written|appended)\s+to\s+([^\s,]+)", content, re.I)
    if m:
        return m.group(1).strip()
    # "Output file: /path" or "output_file_path": "/path"
    m = re.search(r"(?:output\s+file|output_file_path)[:\s]+([^\s,]+)", content, re.I)
    if m:
        return m.group(1).strip().strip('"')
    # Fallback: any absolute path that looks like a file (e.g. /Users/.../file.md)
    m = re.search(r"(/\S+\.(?:md|txt|tex|json)\b)", content)
    if m:
        return m.group(1).strip()
    return None


def _short_path(path, max_len=50):
    """Return a display-friendly path (basename or truncated)."""
    if not path or not isinstance(path, str):
        return ""
    s = path.strip()
    if not s:
        return ""
    expanded = os.path.expanduser(s)
    base = os.path.basename(expanded.rstrip("/"))
    if len(base) <= max_len:
        return base
    return base[: max_len - 2] + "…"


def _tool_status_for_chat(tool_name, args):
    """Return a precise, user-friendly status line for the input status bar (tool + key args)."""
    name = (tool_name or "").lower()
    arg = (args or {})
    if not isinstance(arg, dict):
        arg = {}
    fp = arg.get("file_path") or arg.get("filepath") or arg.get("path") or ""
    path = (fp or "").strip()
    query = (arg.get("query") or arg.get("url") or "").strip()
    cmd = (arg.get("command") or "").strip()
    short_path = _short_path(path) if path else ""
    short_query = (query[:40] + "…") if len(query) > 40 else query
    short_cmd = (cmd[:35] + "…") if len(cmd) > 35 else cmd

    if name == "read_file":
        return f"read_file: {short_path or '…'}"
    if name == "read_file_chunk":
        idx = arg.get("chunk_index", "?")
        n = arg.get("num_chunks", 1)
        if n and int(n) > 1:
            return f"read_file_chunk: {short_path or '…'} (chunks {idx}+{n})"
        return f"read_file_chunk: {short_path or '…'} (chunk {idx})"
    if name == "write_text_file":
        return f"write_text_file: {short_path or '…'}"
    if name == "save_to_file":
        return f"save_to_file: {short_path or '…'}"
    if name == "append_to_text_file":
        return f"append_to_text_file: {short_path or '…'}"
    if name == "delete_file":
        return f"delete_file: {short_path or '…'}"
    if name == "extract_pdf_text":
        return f"extract_pdf_text: {short_path or '…'}"
    if name == "ocr_scanned_pdf":
        return f"ocr_scanned_pdf: {short_path or '…'}"
    if name == "ocr_scanned_pdf_to_file":
        out = (arg.get("output_file_path") or "").strip()
        out_short = _short_path(out) if out else "…"
        return f"ocr_scanned_pdf_to_file: {short_path or '…'} → {out_short}"
    if name == "ocr_complex_document_to_file":
        out = (arg.get("output_file_path") or "").strip()
        out_short = _short_path(out) if out else "…"
        return f"ocr_complex_document_to_file: {short_path or '…'} → {out_short}"
    if name == "ocr_with_kraken_api":
        return f"ocr_with_kraken_api: {short_path or '…'}"
    if name == "ocr_with_kraken_api_to_file":
        out = (arg.get("output_file_path") or "").strip()
        out_short = _short_path(out) if out else "…"
        return f"ocr_with_kraken_api_to_file: {short_path or '…'} → {out_short}"
    if name == "split_pdf":
        return "split_pdf: splitting PDF…"
    if name == "merge_pdfs":
        return "merge_pdfs: merging PDFs…"
    if name == "split_text_file":
        return "split_text_file: splitting…"
    if name == "merge_text_files":
        return "merge_text_files: merging…"
    if name == "submit_ocr_chunk_corrections":
        return f"submit_ocr_chunk_corrections: {short_path or '…'}"
    if name == "list_directory":
        return f"list_directory: {short_path or '.'}"
    if name in ("search_web", "search_web_public"):
        return f"{name}: {short_query or '…'}"
    if name in ("search_files", "search_filesystem"):
        return "search_filesystem: searching…"
    if name == "read_webpage":
        return f"read_webpage: {short_query or '…'}"
    if name == "download_file":
        return "download_file: downloading…"
    if name == "run_python_code":
        return "run_python_code: running snippet…"
    if name == "shell_execute":
        return f"shell_execute: {short_cmd or '…'}"
    if name == "get_file_metadata":
        return f"get_file_metadata: {short_path or '…'}"
    if name == "get_current_directory":
        return "get_current_directory"
    if name == "verify_file_exists":
        return f"verify_file_exists: {short_path or '…'}"
    if name == "get_default_save_location":
        return "get_default_save_location"
    if name == "suggest_save_filename":
        return f"suggest_save_filename: {short_path or '…'}"
    # Research tools
    if name == "deep_web_search":
        return f"deep_web_search: {short_query or '…'}"
    if name == "read_and_analyze_page":
        return f"read_and_analyze_page: {short_query or '…'}"
    if name == "analyze_multiple_sources":
        n = len(arg.get("analysis_outputs") or [])
        return f"analyze_multiple_sources: {n} source(s)"
    if name == "save_research_findings":
        topic = (arg.get("topic") or "").strip()
        short_topic = (topic[:40] + "…") if len(topic) > 40 else (topic or "…")
        return f"save_research_findings: {short_topic}"
    if name == "suggest_follow_up_queries":
        return "suggest_follow_up_queries: …"
    if name == "generate_final_research_report":
        topic = (arg.get("topic") or "").strip()
        short_topic = (topic[:40] + "…") if len(topic) > 40 else (topic or "…")
        return f"generate_final_research_report: {short_topic}"
    if name.startswith("mcp__"):
        parts = name.split("__", 2)
        server = parts[1] if len(parts) > 1 else "mcp"
        tool = parts[2] if len(parts) > 2 else name
        first_arg = ""
        try:
            if arg:
                key = next(iter(arg.keys()))
                val = str(arg.get(key, "")).strip()
                first_arg = (val[:32] + "…") if len(val) > 32 else val
        except (TypeError, ValueError, StopIteration):
            _log.debug("MCP tool status: failed to peek first arg", exc_info=True)
            first_arg = ""
        return f"{server}: {tool}" + (f" ({first_arg})" if first_arg else "")
    if name == "spotlight_search":
        q = (arg.get("query") or "").strip()
        name_q = (arg.get("name_contains") or "").strip()
        within = arg.get("within_days") or 0
        scope = (arg.get("scope") or "").strip()
        bits: list[str] = []
        if q:
            bits.append((q[:40] + "…") if len(q) > 40 else q)
        if name_q:
            bits.append(f"name~{name_q}")
        try:
            if int(within) > 0:
                bits.append(f"{int(within)}d")
        except (TypeError, ValueError):
            _log.debug("spotlight_search status: invalid within_days", exc_info=True)
        if scope:
            scope_short = _short_path(scope) if scope.startswith("/") or scope.startswith("~") else scope
            bits.append(f"in {scope_short}")
        return "spotlight_search: " + (", ".join(bits) if bits else "searching…")
    if name == "get_recent_files":
        lim = arg.get("limit") or 5
        try:
            lim = int(lim)
        except (TypeError, ValueError):
            lim = 5
        return f"get_recent_files: last {lim}"
    if name == "list_inbox":
        bits: list[str] = []
        fc = (arg.get("from_contains") or "").strip()
        sc = (arg.get("subject_contains") or "").strip()
        if fc:
            bits.append(f"from ~{fc}")
        if sc:
            bits.append(f"subject ~{sc}")
        if arg.get("unread_only", True):
            bits.append("unread")
        return "list_inbox: " + (", ".join(bits) if bits else "recent")
    if name == "delegate_task":
        atype = arg.get("agent_type", "research")
        short_desc = (arg.get("task_description") or "")[:40]
        if len(arg.get("task_description", "")) > 40:
            short_desc += "…"
        return f"delegate_task ({atype}): {short_desc}"
    return f"tool: {name}"


# Max size per image for base64 embedding (backends that require base64 often have limits)
_MAX_IMAGE_BASE64_BYTES = 10 * 1024 * 1024  # 10 MB


_LOCAL_USER_NAME_QUERY_RE = re.compile(
    r"^\s*(?:(?:hey|hi|hello)\s+\w+\s*,?\s*)?"
    r"(?:"
    r"what(?:'|’| i)?s\s+my\s+name\??|"
    r"what\s+is\s+my\s+name\??|"
    r"who\s+am\s+i\??|"
    r"do\s+you\s+know\s+my\s+name\??|"
    r"what\s+do\s+you\s+call\s+me\??"
    r")\s*$",
    re.IGNORECASE,
)


_PLAIN_CHAT_TOOL_TRIGGER_RE = re.compile(
    r"\b("
    r"search|browse|google|look\s+up|web|online|current|latest|today|now|"
    r"weather|temperature|news|stock|price|score|"
    r"read|write|edit|create|delete|save|append|open|move|copy|rename|"
    r"file|folder|directory|pdf|image|screenshot|ocr|clipboard|"
    r"run|execute|shell|terminal|python|code|script|command|"
    r"email|mail|calendar|remind|schedule|timer|"
    r"research|investigate|compare|find|check|"
    r"remember|recall|memory|memories|previous|past|earlier|"
    r"conversation|conversations|"
    r"met\s+before|(?:we(?:'|’)ve|we\s+have)\s+met|know\s+me|"
    r"commit|push|git|plugin|workflow|skill|mesh|peer"
    r")\b",
    re.IGNORECASE,
)


def _should_force_plain_chat_run(
    text: str,
    *,
    has_media_attachment: bool = False,
    research_mode: bool = False,
    include_cli: bool = False,
    idle_tick: bool = False,
) -> bool:
    """Disabled worker-side fast routing.

    Normal user turns must remain tool-enabled. Keeping the helper as a stable
    symbol avoids breaking imports while preventing future no-tools bypasses.
    """
    return False


def _fast_local_identity_reply(
    text: str,
    *,
    has_media_attachment: bool = False,
    research_mode: bool = False,
    include_cli: bool = False,
    idle_tick: bool = False,
    quick_panel: bool = False,
    dashboard_surface: bool = False,
) -> str | None:
    """Answer trivial local identity questions without paying a full LLM prefill."""
    if has_media_attachment or research_mode or include_cli or idle_tick or quick_panel or dashboard_surface:
        return None
    if not isinstance(text, str):
        return None
    stripped = text.strip()
    if not stripped or len(stripped) > 120 or "\n" in stripped:
        return None
    if not _LOCAL_USER_NAME_QUERY_RE.match(stripped):
        return None
    def _load_user_display_name() -> str:
        from core.config import get_user_display_name

        return (get_user_display_name() or "").strip()

    user_name = best_effort_value(
        _load_user_display_name,
        log="Failed to read user display name for fast identity reply",
        default="",
        logger=_log,
    )
    if not user_name:
        return None
    return f"Your name is {user_name}."


def _image_paths_to_data_urls(image_paths, log_fn=None):
    """Convert local image paths to data:image/...;base64,... URLs for vision backends that require base64."""
    if not image_paths:
        return []
    data_urls = []
    ext_to_mime = {
        ".png": "image/png",
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".gif": "image/gif",
        ".bmp": "image/bmp",
        ".webp": "image/webp",
    }
    for item in image_paths:
        if isinstance(item, (list, tuple)) and len(item) >= 2:
            filename, path = item[0], item[1]
        else:
            path = item
            filename = os.path.basename(path) if isinstance(path, str) else ""
        path = os.path.expanduser(path) if isinstance(path, str) else path
        if not path or not os.path.isfile(path):
            if log_fn:
                log_fn(f"[vision] Skip missing/not-file: {path}", "DEBUG")
            continue
        try:
            size = os.path.getsize(path)
            if size > _MAX_IMAGE_BASE64_BYTES:
                if log_fn:
                    log_fn(f"[vision] Skip large image ({size} bytes): {path}", "WARN")
                continue
            with open(path, "rb") as f:
                raw = f.read()
            b64 = base64.standard_b64encode(raw).decode("ascii")
            ext = os.path.splitext(filename)[1].lower() if filename else os.path.splitext(str(path))[1].lower()
            mime = ext_to_mime.get(ext, "image/png")
            data_urls.append(f"data:{mime};base64,{b64}")
        except Exception as e:
            if log_fn:
                log_fn(f"[vision] Failed to read image {path}: {e}", "WARN")
    return data_urls


def build_friendly_agent_error_message(exc: BaseException) -> str:
    """Map an agent-run exception to a calm, plain-language message for the chat bubble.

    This is the only text the user sees when a run fails, so it must NEVER embed raw
    exception text, tracebacks, or "Error type:" — the full technical detail stays in
    the worker logs and on the activity event (`meta.detail`). Pure and side-effect
    free (aside from a best-effort backend-URL lookup) so it can be unit-tested.
    """
    error_str = str(exc).lower()
    error_type = type(exc).__name__

    def _is_local_mlx_url(url: str) -> bool:
        u = url or ""
        return ":8000" in u or "127.0.0.1" in u or "localhost" in u

    # Server returned HTML instead of a JSON API response.
    if "<!doctype html>" in error_str or "<html" in error_str:
        if "internal server error" in error_str:
            return (
                "The AI model ran into a problem with that request.\n\n"
                "This usually happens when a request gets too large — for example after "
                "reading several files or a very long conversation.\n\n"
                "What helps:\n"
                "• Start a new chat, or ask about a smaller piece at a time.\n"
                "• If it keeps happening, reload your model in Settings → LLMs."
            )
        return (
            "I couldn't get a valid response from the AI server.\n\n"
            "It may not be running, or it may need to be reconfigured. "
            "You can review your model in Settings → LLMs and try again."
        )
    # Connection errors (can't reach the LLM endpoint at all).
    if (
        "apiconnectionerror" in error_type.lower()
        or "connection error" in error_str
        or "connection reset" in error_str
        or "connection refused" in error_str
        or "readerror" in error_str
    ):
        try:
            from agent_loop import get_backend_base_url
            configured_url = get_backend_base_url()
        except Exception:
            configured_url = ""
        if _is_local_mlx_url(configured_url):
            return (
                "I can't reach the local AI model just yet.\n\n"
                "It may still be starting up, or it might not be set up. Give it a few "
                "seconds and try again — or open Settings → LLMs to finish setting up a model."
            )
        return (
            "I can't reach the AI server right now.\n\n"
            "Please make sure it's running and reachable, then try again. "
            "You can review the connection in Settings → LLMs."
        )
    # InternalServerError from the OpenAI client (often a 500 when context is too large).
    if "internalservererror" in error_type.lower() or "internal server error" in error_str:
        return (
            "The AI model ran into a problem with that request.\n\n"
            "This usually happens when a request gets too large — for example after "
            "reading several files or a very long conversation.\n\n"
            "What helps:\n"
            "• Start a new chat, or ask about a smaller piece at a time.\n"
            "• If it keeps happening, reload your model in Settings → LLMs."
        )
    # MLX chat-template / runtime mismatch.
    if (
        "apply_chat_template" in error_str
        or "chat template" in error_str
        or "multimodal stream" in error_str
    ):
        return (
            "This model needs a different setup to work with Mugi.\n\n"
            "Mugi will try a fallback model during setup. If you're seeing this "
            "outside of setup, open Settings → LLMs and reload a model."
        )
    # Timeouts.
    if "timeout" in error_str or "readtimeout" in error_str or "timed out" in error_str:
        return (
            "That request took too long and timed out.\n\n"
            "This can happen with very large files or when the model is busy.\n\n"
            "What helps:\n"
            "• Try a smaller file or a shorter question.\n"
            "• Check that the model is responding, then try again."
        )
    return (
        "Something went wrong while I was working on that.\n\n"
        "Please try again. If it keeps happening, restart Mugi or "
        "check your model in Settings → LLMs."
    )


class AgentWorker(Thread):
    """Background thread for agent execution."""
    
    def __init__(self, user_text, on_chunk, on_finished, on_error, on_done, on_log=None, on_status=None, on_live_view=None, on_chat_event=None, on_plugin_stream=None, on_tool_event=None, on_agent_message=None, on_new_assistant_content=None, on_replace_last_content=None, thread_id="default", image_urls=None, image_paths=None, vision_image_format=None, research_mode=False, has_media_attachment=False, include_cli=False, display_text=None, file_attachments=None, run_id="", parent_run_id="", idle_tick=False, quick_panel=False, dashboard_surface=False):
        super().__init__(daemon=True)
        self.user_text = user_text
        self.display_text = display_text  # optional: short text for storage/UI; user_text is used for the model
        self.file_attachments = list(file_attachments or [])  # [{filename, url}, ...] for non-image chips
        self.idle_tick = bool(idle_tick)
        self.on_chunk = on_chunk
        self.on_finished = on_finished
        self.on_error = on_error
        self.on_done = on_done
        self.on_log = on_log
        self.on_status = on_status  # optional: status lines (e.g. "Writing OCR to file…") go here instead of on_chunk
        self.on_live_view = on_live_view  # optional: rich Live View event callback (one dict arg); set on tools layer during run
        self.on_chat_event = on_chat_event  # optional: push structured events (e.g. inline images) into chat stream
        self.on_plugin_stream = on_plugin_stream  # optional: push structured plugin stream events into realtime dashboard SSE
        self.on_tool_event = on_tool_event  # optional: push structured tool start/result events into chat stream
        self.on_agent_message = on_agent_message  # optional: (text, is_thinking) -> show in chat as agent/thinking bubble
        self.on_new_assistant_content = on_new_assistant_content  # optional: no-arg callback when starting new assistant segment (round_ > 1)
        self.on_replace_last_content = on_replace_last_content  # optional: (normalized_text) -> replace current bubble so streamed text normalizes per round
        self.thread_id = thread_id  # for multi-conversation: config["configurable"]["thread_id"]
        self.image_urls = image_urls if image_urls else []  # for UI bubble display (localhost URLs)
        self.image_paths = image_paths if image_paths else []  # list of (filename, path) for vision
        self.vision_image_format = (vision_image_format or "base64").strip().lower()  # "base64" (default) or "url"
        if self.vision_image_format not in ("base64", "url"):
            self.vision_image_format = "base64"
        self.research_mode = bool(research_mode)
        self.has_media_attachment = bool(has_media_attachment)
        self.include_cli = bool(include_cli)
        self.run_id = sanitize_run_id(run_id)
        self.parent_run_id = sanitize_run_id(parent_run_id)
        self.quick_panel = bool(quick_panel)
        self.dashboard_surface = bool(dashboard_surface)
        self._stop_requested = False
        self._status_timer = None  # for "still working" ping during long tool runs
        self._last_output_path = None  # last file path from ocr_to_file / write_text_file / save_to_file
        self._memory_turn_summary_ref: list | None = [None]
        self._knowledge_turn_summary_ref: list | None = [None]
    
    def request_stop(self):
        self._stop_requested = True

        def _terminate_subprocesses() -> None:
            from tools._logging import terminate_cancellable_subprocesses_for_thread

            terminate_cancellable_subprocesses_for_thread(self.ident)

        best_effort(
            _terminate_subprocesses,
            log="Failed to terminate cancellable subprocesses on stop request",
            logger=_log,
        )

    def _bubble_text_for_storage(self) -> str | None:
        """User-visible bubble text for chat_storage — never the agent attachment preamble."""
        if self.display_text is not None:
            return (self.display_text or "").strip() or None
        if self.file_attachments or self.image_urls or self.image_paths:
            return None
        return None

    def _worker_log(self, msg, level="DEBUG"):
        """Print to stdout and send full message to terminal window (no truncation)."""
        prefix = f"[run_id={self.run_id}] " if self.run_id else ""
        out = f"{prefix}{msg}"
        print(f"{_log_ts()} {out}", flush=True)
        if self.on_log:
            best_effort(
                lambda: self.on_log(level, out),
                log="Failed to forward worker log to on_log callback",
                logger=_log,
            )
    
    def run(self):
        _run_ctx_token = set_run_context(
            run_id=self.run_id,
            thread_id=self.thread_id,
            parent_run_id=self.parent_run_id,
        )
        self._worker_log("[AGENT WORKER] Starting run() method", "AGENT")
        try:
            self._worker_log(f"[AGENT WORKER] Thread name: {self.name}", "DEBUG")
            self._worker_log(f"[AGENT WORKER] Input text length: {len(self.user_text)}", "DEBUG")
            
            if self.on_log:
                try:
                    self.on_log("AGENT", f"Starting agent execution with input: {self.user_text[:200]}...")
                    self.on_log("DEBUG", f"AgentWorker thread: {self.name}")
                except Exception as log_err:
                    self._worker_log(f"[ERROR] Failed to log initial message: {log_err}", "ERROR")
            # Allow tools (e.g. read_webpage) to send status to the input bar during this run
            if self.on_status:
                set_input_status_callback(self.on_status)
            if self.on_live_view:
                set_live_view_callback(self.on_live_view)
            if self.on_chat_event:
                set_chat_event_callback(self.on_chat_event)
            if self.on_plugin_stream:
                set_plugin_stream_callback(self.on_plugin_stream)
            set_tool_stop_check_callback(lambda: self._stop_requested)

            # Build user message (OpenAI format; vision as content parts)
            use_urls = self.vision_image_format == "url" and self.image_urls
            image_data_urls = () if use_urls else _image_paths_to_data_urls(self.image_paths, self._worker_log)
            if use_urls:
                content = []
                if self.user_text:
                    content.append({"type": "text", "text": self.user_text})
                for url in self.image_urls:
                    content.append({"type": "image_url", "image_url": {"url": url}})
                user_msg = {"role": "user", "content": content}
            elif image_data_urls:
                content = []
                if self.user_text:
                    content.append({"type": "text", "text": self.user_text})
                for data_url in image_data_urls:
                    content.append({"type": "image_url", "image_url": {"url": data_url}})
                user_msg = {"role": "user", "content": content}
            elif self.image_urls:
                content = []
                if self.user_text:
                    content.append({"type": "text", "text": self.user_text})
                for url in self.image_urls:
                    content.append({"type": "image_url", "image_url": {"url": url}})
                user_msg = {"role": "user", "content": content}
            else:
                user_msg = {"role": "user", "content": self.user_text}

            # Thread history + new user message (run_raw_agent will prepend system prompt)
            from chat_storage import get_messages as _get_messages, append_messages as _append_messages
            history = _get_messages(self.thread_id, include_row_id=True) or []
            history = [
                m for m in history
                if isinstance(m, dict) and (m.get("role") in ("user", "assistant"))
            ]
            last_path_ref = [None]
            self._memory_turn_summary_ref = [None]
            self._knowledge_turn_summary_ref = [None]

            # Persist user message for UI/history: use display_text (short, chip-style) so bubble and reload show icon + filename only.
            _new_user_row_id: int | None = None
            _reuse_coordinator_delivery = False
            if self.idle_tick:
                try:
                    from utils.kanban_coordinator_delivery import (
                        should_reuse_coordinator_delivery_for_idle_wake,
                    )

                    if should_reuse_coordinator_delivery_for_idle_wake(history):
                        _reuse_coordinator_delivery = True
                    else:
                        from utils.council_delivery import (
                            should_reuse_council_delivery_for_idle_wake,
                        )

                        if should_reuse_council_delivery_for_idle_wake(history):
                            _reuse_coordinator_delivery = True
                    if _reuse_coordinator_delivery:
                        for _m in reversed(history):
                            if (_m.get("role") or "").strip() == "user":
                                user_msg = _m
                                break
                except Exception:
                    _log.debug("coordinator/council delivery reuse probe failed", exc_info=True)

            def _persist_user_message() -> None:
                nonlocal _new_user_row_id
                to_store = user_msg
                if self.idle_tick and isinstance(to_store, dict):
                    to_store = dict(to_store)
                    to_store["_idle_tick"] = True
                if isinstance(user_msg.get("content"), list):
                    stored_content = []
                    bubble_text = self._bubble_text_for_storage()
                    image_idx = [0]
                    for part in user_msg["content"]:
                        if isinstance(part, dict) and part.get("type") == "text":
                            if bubble_text is not None:
                                stored_content.append({"type": "text", "text": bubble_text})
                            elif not (self.file_attachments or self.image_urls or self.image_paths):
                                stored_content.append({"type": "text", "text": part.get("text") or ""})
                        elif isinstance(part, dict) and (part.get("type") == "image_url" or "image_url" in part):
                            url = (part.get("image_url") or {}).get("url") or ""
                            if isinstance(url, str) and url.strip().lower().startswith("data:"):
                                stored_content.append(part)
                            elif image_idx[0] < len(self.image_urls):
                                stored_content.append({"type": "image_url", "image_url": {"url": self.image_urls[image_idx[0]]}})
                                image_idx[0] += 1
                            else:
                                stored_content.append(part)
                        else:
                            stored_content.append(part)
                    for fa in self.file_attachments:
                        if isinstance(fa, dict):
                            fn = str(fa.get("filename") or "").strip()
                            url = str(fa.get("url") or "").strip()
                            if fn and url:
                                stored_content.append({"type": "file_attachment", "filename": fn, "url": url})
                    to_store = dict(to_store)
                    to_store["content"] = stored_content
                elif self.file_attachments:
                    stored_content = []
                    bubble_text = self._bubble_text_for_storage()
                    if bubble_text:
                        stored_content.append({"type": "text", "text": bubble_text})
                    for fa in self.file_attachments:
                        if isinstance(fa, dict):
                            fn = str(fa.get("filename") or "").strip()
                            url = str(fa.get("url") or "").strip()
                            if fn and url:
                                stored_content.append({"type": "file_attachment", "filename": fn, "url": url})
                    to_store = dict(user_msg)
                    to_store["content"] = stored_content
                elif self.display_text is not None and self.idle_tick:
                    to_store = dict(to_store)
                    to_store["content"] = self.display_text
                _new_user_row_id = _append_messages(self.thread_id, [to_store])

            if _reuse_coordinator_delivery:
                if isinstance(user_msg, dict) and user_msg.get("_row_id") is not None:
                    _new_user_row_id = user_msg.get("_row_id")
            else:
                try:
                    _persist_user_message()
                except Exception:
                    _log.error("Failed to persist user message to thread history", exc_info=True)
                    if self.on_error:
                        try:
                            self.on_error(
                                "Could not save your message to conversation history. "
                                "Please try sending again."
                            )
                        except Exception:
                            _log.debug("on_error callback failed after user persist failure", exc_info=True)
                    try:
                        self.on_done()
                    finally:
                        return
                if _new_user_row_id is None:
                    _log.error("User message persist returned no row id for thread %s", self.thread_id)
                    if self.on_error:
                        try:
                            self.on_error(
                                "Could not save your message to conversation history. "
                                "Please try sending again."
                            )
                        except Exception:
                            _log.debug("on_error callback failed after empty row id", exc_info=True)
                    try:
                        self.on_done()
                    finally:
                        return

            # Attach _row_id to the new user message so anchor computation can include it.
            if _new_user_row_id is not None:
                user_msg["_row_id"] = _new_user_row_id

            _fast_reply = _fast_local_identity_reply(
                self.user_text,
                has_media_attachment=self.has_media_attachment or bool(self.image_urls) or bool(self.image_paths),
                research_mode=self.research_mode,
                include_cli=self.include_cli,
                idle_tick=self.idle_tick,
                quick_panel=self.quick_panel,
                dashboard_surface=self.dashboard_surface,
            )
            if _fast_reply is not None:
                from utils.knowledge_turn_trace import (
                    knowledge_turn_trace_begin,
                    knowledge_turn_trace_finalize,
                )
                from utils.memory_turn_trace import memory_turn_trace_begin, memory_turn_trace_finalize

                memory_turn_trace_begin(self.thread_id, self._memory_turn_summary_ref)
                knowledge_turn_trace_begin(self.thread_id, self._knowledge_turn_summary_ref)
                try:
                    _append_messages(
                        self.thread_id, [{"role": "assistant", "content": _fast_reply}]
                    )
                except Exception:
                    _log.error("Failed to persist fast-path assistant reply", exc_info=True)
                    if self.on_error:
                        try:
                            self.on_error(
                                "Could not save the assistant reply to conversation history. "
                                "It may be missing after reload."
                            )
                        except Exception:
                            _log.debug(
                                "on_error callback failed after fast-path persist failure",
                                exc_info=True,
                            )
                    try:
                        self.on_done()
                    finally:
                        return
                if self.on_status:
                    best_effort(
                        lambda: self.on_status(""),
                        log="Failed to clear status on fast-path reply",
                        logger=_log,
                    )
                if self.on_chunk:
                    best_effort(
                        lambda: self.on_chunk(_fast_reply),
                        log="Failed to emit fast-path chunk",
                        logger=_log,
                    )
                memory_turn_trace_finalize(self._memory_turn_summary_ref)
                knowledge_turn_trace_finalize(self._knowledge_turn_summary_ref)
                if self.on_finished:
                    best_effort(
                        lambda: self.on_finished(_fast_reply),
                        log="Failed to emit fast-path finished callback",
                        logger=_log,
                    )
                try:
                    self.on_done()
                finally:
                    return

            # Normal user turns must stay tool-enabled. The only no-tools fast
            # path left is the synthetic startup greeting inside run_raw_agent.
            _force_plain_chat = False

            # Build merged history: literal messages + context anchor for the LLM.
            from context_summarizer import build_merged_history_for_main_agent
            if _reuse_coordinator_delivery:
                messages = build_merged_history_for_main_agent(self.thread_id, list(history))
            else:
                messages = build_merged_history_for_main_agent(self.thread_id, list(history) + [user_msg])

            self._worker_log("[AGENT WORKER] Calling run_raw_agent()", "AGENT")
            if self.on_status:
                best_effort(
                    lambda: self.on_status("Connecting…"),
                    log="Failed to emit connecting status",
                    logger=_log,
                )

            # Elapsed-time ticker: fires every 2 s while the LLM is evaluating the prompt.
            # Cancelled as soon as the first chunk or an explicit on_status update arrives.
            _ticker_cancelled = threading.Event()
            _ticker_start = time.monotonic()
            _TICKER_MESSAGES = [
                "Processing…",
                "Processing…",
                "Thinking…",
                "Thinking…",
                "Almost there…",
            ]

            def _ticker_fire():
                if _ticker_cancelled.is_set():
                    return
                elapsed = int(time.monotonic() - _ticker_start)
                phase = min(elapsed // 4, len(_TICKER_MESSAGES) - 1)
                label = _TICKER_MESSAGES[phase]
                if self.on_status:
                    _ticker_status = f"{label} ({elapsed}s)"
                    best_effort(
                        lambda: self.on_status(_ticker_status),
                        log="Failed to emit elapsed-time status ticker",
                        logger=_log,
                    )
                if not _ticker_cancelled.is_set():
                    self._status_timer = threading.Timer(2.0, _ticker_fire)
                    self._status_timer.daemon = True
                    self._status_timer.start()

            self._status_timer = threading.Timer(2.0, _ticker_fire)
            self._status_timer.daemon = True
            self._status_timer.start()

            # Wrap on_status so ticker stops the moment agent_loop sends a real status.
            _original_on_status = self.on_status

            def _on_status_with_cancel(status, _orig=_original_on_status):
                _ticker_cancelled.set()
                if self._status_timer is not None:
                    self._status_timer.cancel()
                    self._status_timer = None
                if _orig:
                    best_effort(
                        lambda: _orig(status),
                        log="Failed to forward on_status from agent loop",
                        logger=_log,
                    )

            # Wrap on_chunk so ticker stops the moment the first token arrives.
            _original_on_chunk = self.on_chunk

            def _on_chunk_with_cancel(chunk, _orig=_original_on_chunk):
                _ticker_cancelled.set()
                if self._status_timer is not None:
                    self._status_timer.cancel()
                    self._status_timer = None
                if _orig:
                    best_effort(
                        lambda: _orig(chunk),
                        log="Failed to forward on_chunk from agent loop",
                        logger=_log,
                    )

            # run_raw_agent loads config itself so personality is fresh every turn; no need to load here.
            _policy_ctx = None
            if self.idle_tick or self.quick_panel or self.dashboard_surface:
                from tool_policy import context_from_prefs_and_flags
                _policy_ctx = context_from_prefs_and_flags(
                    idle_tick=self.idle_tick,
                    quick_panel=self.quick_panel,
                    dashboard_surface=self.dashboard_surface,
                )
            run_raw_agent(
                messages=messages,
                thread_id=self.thread_id,
                config=None,
                on_chunk=_on_chunk_with_cancel,
                on_log=self.on_log,
                on_error=self.on_error,
                on_finished=self.on_finished,
                on_status=_on_status_with_cancel,
                on_tool_event=self.on_tool_event,
                on_agent_message=self.on_agent_message,
                on_new_assistant_content=self.on_new_assistant_content,
                on_replace_last_content=self.on_replace_last_content,
                _last_output_path_ref=last_path_ref,
                stop_check=lambda: self._stop_requested,
                research_mode=self.research_mode,
                has_media_attachment=self.has_media_attachment,
                include_cli=self.include_cli,
                policy_context=_policy_ctx,
                quick_panel=self.quick_panel,
                dashboard_surface=self.dashboard_surface,
                force_plain_chat=_force_plain_chat,
                _memory_turn_summary_ref=self._memory_turn_summary_ref,
                _knowledge_turn_summary_ref=self._knowledge_turn_summary_ref,
            )
            if last_path_ref and last_path_ref[0]:
                self._last_output_path = last_path_ref[0]

            self._worker_log("[AGENT WORKER] Calling on_done", "AGENT")
            print(f"{_log_ts()} [AGENT_DEBUG] About to call on_done (worker thread)", flush=True)
            try:
                self.on_done()
                print(f"{_log_ts()} [AGENT_DEBUG] on_done returned", flush=True)
            except Exception as done_err:
                print(f"{_log_ts()} [AGENT_DEBUG] on_done raised: {done_err}", flush=True)
                self._worker_log(f"[ERROR] on_done failed: {done_err}", "ERROR")
                import traceback
                self._worker_log(f"[ERROR] Traceback:\n{traceback.format_exc()}", "ERROR")
        except Exception as e:
            import traceback
            error_msg = traceback.format_exc()
            # Log to console first (in case terminal logging fails)
            print(f"{_log_ts()} [ERROR] Agent execution failed: {str(e)}")
            print(f"{_log_ts()} [ERROR] Full traceback:\n{error_msg}")
            
            # Map the exception to a calm, plain-language chat message. Raw detail
            # (str(e), traceback) is preserved in the logs above and on the activity
            # event below (`meta.detail`) — never in the user-facing bubble.
            technical_detail = str(e)[:600]
            user_friendly_msg = build_friendly_agent_error_message(e)
            
            # Try to log to terminal (may fail if UI is broken)
            try:
                if self.on_log:
                    self.on_log("ERROR", f"Agent execution failed: {str(e)}")
                    self.on_log("ERROR", f"Exception type: {type(e).__name__}")
                    self.on_log("ERROR", f"Full traceback:\n{error_msg}")
            except Exception as log_error:
                print(f"{_log_ts()} [ERROR] Failed to log error to terminal: {log_error}")
            
            # Try to call error handler
            try:
                self.on_error(user_friendly_msg)
            except Exception as error_handler_error:
                print(f"{_log_ts()} [ERROR] Failed to call on_error handler: {error_handler_error}")
            
            # Try to call done handler
            try:
                self.on_done()
            except Exception as done_error:
                print(f"{_log_ts()} [ERROR] Failed to call on_done handler: {done_error}")
            best_effort(
                lambda: emit_activity_enriched(
                    {
                        "source": "agent",
                        "state": "error",
                        "summary": f"Agent run failed: {type(e).__name__}",
                        "thread_id": self.thread_id,
                        "run_id": self.run_id,
                        "meta": {"error_type": type(e).__name__, "detail": technical_detail},
                    },
                    tags=["user_initiated"],
                ),
                log="agent worker error activity emit failed",
                logger=_log,
            )

        finally:
            clear_run_context(_run_ctx_token)
            best_effort(
                lambda: _ticker_cancelled.set(),
                log="Failed to cancel status ticker event",
                logger=_log,
            )
            if self._status_timer is not None:

                def _cancel_status_timer() -> None:
                    if self._status_timer is not None:
                        self._status_timer.cancel()

                best_effort(
                    _cancel_status_timer,
                    log="Failed to cancel status timer",
                    logger=_log,
                )
                self._status_timer = None
            set_input_status_callback(None)
            set_live_view_callback(None)
            set_chat_event_callback(None)
            set_plugin_stream_callback(None)
            set_tool_stop_check_callback(None)

            def _clear_subprocess_registry() -> None:
                from tools._logging import clear_cancellable_subprocess_registry_for_thread

                clear_cancellable_subprocess_registry_for_thread(self.ident)

            best_effort(
                _clear_subprocess_registry,
                log="Failed to clear cancellable subprocess registry for worker thread",
                logger=_log,
            )


# ============================================================================

# Raw OpenAI-compatible agent; no LangGraph. Run SwiftUI app or python app_pyobjc2.py (legacy) for the UI.
if __name__ == "__main__":
    print("Raw agent (no LangGraph). Run the SwiftUI app (MugiApp) or python app_pyobjc2.py for the legacy UI.", flush=True)
