# app_http_routes.py - FastAPI routes for the chat app. No import of app_pyobjc2.
# create_fastapi_app(delegate, html_content, temp_upload_dir, get_voice_prefs=None) returns the FastAPI app.
# Delegate must have _main_thread_queue and the handler methods the routes invoke on the main thread.
from __future__ import annotations

import asyncio
import base64
import hashlib
import json
import logging
import os
import shlex
import shutil
import tempfile
import threading
import time
import uuid
from typing import Any, Callable
from urllib.parse import quote, unquote

_log = logging.getLogger(__name__)


def is_trusted_localhost_http_url(value: str) -> bool:
    """Allow empty; else http:// to loopback (localhost, 127.0.0.1, IPv6 [::1]).

    WKWebView may send Origin/Referer as ``http://[::1]:port/...``; treating that as
    non-local caused 403 on ``/plugin/...`` while ``http://localhost:port`` was fine.
    """
    v = (value or "").strip()
    if not v:
        return True
    lo = v.lower()
    return (
        lo.startswith("http://localhost:")
        or lo.startswith("http://127.0.0.1:")
        or lo.startswith("http://[::1]")
    )


from fastapi import FastAPI, Form, UploadFile, File, Request, Query
from fastapi.responses import HTMLResponse, JSONResponse, FileResponse, StreamingResponse, Response

from http_send_surface import parse_send_surface as _parse_send_surface
from utils.path_safety import is_dangerous_path
import attachment_registry as _attach_reg

# Post-onboarding discoverability: rotating one-line tips (voice, attachments, research, vault, tools)
POST_ONBOARDING_TIPS = [
    "Tip: Click the mic to speak your message or talk to the agent.",
    "Tip: Attach files or paste images for the agent to read.",
    "Tip: Turn on Research for deeper answers with sources.",
    "Tip: Open the Vault to keep notes and drafts organized.",
    "Tip: The agent can run tools (e.g. search, code) when you allow it.",
]


def _trust_summary_memory_snapshot() -> dict[str, Any]:
    """Bounded filesystem/config glance for `/activity/trust_summary` ``memory`` key.

    No SQLite or FTS probes — only workspace path sanity, MEMORY.md hub presence,
    and shard-file enumeration (already capped inside ``list_shard_rel_paths``).
    """
    out: dict[str, Any] = {"schema": "mugi.trust_summary.memory.v1", "ok": False}
    try:
        from core import get_workspace_path
        from core.config import get_memory_config
        from utils.memory_shard_layout import layout_is_sharded, list_shard_rel_paths

        ws = (get_workspace_path() or "").strip()
        mem_cfg = get_memory_config() or {}
        fts_on = bool(mem_cfg.get("memory_fts_enabled", True))
        reachable = bool(ws and os.path.isdir(ws))
        hub = bool(ws and os.path.isfile(os.path.join(ws, "MEMORY.md")))
        sharded = bool(ws and layout_is_sharded(ws))
        shard_files = len(list_shard_rel_paths(ws)) if reachable else 0
        folder = os.path.basename(os.path.normpath(ws)) if ws else ""
        out.update(
            {
                "ok": True,
                "workspace_reachable": reachable,
                "workspace_folder_name": folder,
                "memory_hub_present": hub,
                "sharded_layout": sharded,
                "shard_file_count": int(shard_files),
                "memory_fts_enabled": fts_on,
            }
        )
    except Exception:
        out["ok"] = False
    return out


def mount_plugin_http_routes(
    fastapi_app: Any,
    delegate: Any,
    auth_token: str | None,
) -> None:
    """Mount ``/plugin/<slug>`` sub-apps for each loaded plugin with HTTP route proxies.

    Safe to call multiple times: each slug is mounted at most once. Plugin discovery
    may finish *after* :func:`create_fastapi_app` returns (background thread), so
    :func:`mugi_backend._build_app` should call this again after
    ``PluginManager.wait_initial_discover()``.
    """
    pm = getattr(delegate, "_plugin_manager", None)
    if pm is None:
        return
    if getattr(delegate, "_plugin_http_mounted_slugs", None) is None:
        try:
            delegate._plugin_http_mounted_slugs = set()
        except Exception:
            return
    mounted: set = delegate._plugin_http_mounted_slugs
    if getattr(delegate, "_plugin_route_mount_errors", None) is None:
        try:
            delegate._plugin_route_mount_errors = {}
        except Exception:
            pass
    _plugin_auth_token = auth_token

    def _make_plugin_auth_dep(token: str | None):
        from fastapi import Depends, HTTPException as _HTTPException

        async def _plugin_auth(request: Request):
            _max_body = 1024 * 1024
            _content_len = request.headers.get("content-length")
            if _content_len:
                try:
                    if int(_content_len) > _max_body:
                        raise _HTTPException(status_code=413, detail="payload too large")
                except ValueError:
                    raise _HTTPException(status_code=400, detail="invalid content-length")
            if token:
                t = request.cookies.get("mugi_auth") or request.headers.get("x-auth-token")
                if t != token:
                    raise _HTTPException(status_code=403, detail="forbidden")
                origin = request.headers.get("origin") or ""
                referer = request.headers.get("referer") or ""
                if origin and not is_trusted_localhost_http_url(origin):
                    raise _HTTPException(status_code=403, detail="forbidden")
                if referer and not is_trusted_localhost_http_url(referer):
                    raise _HTTPException(status_code=403, detail="forbidden")
            else:
                host = (request.client.host if request.client else "") or ""
                if host not in ("127.0.0.1", "::1", "localhost"):
                    raise _HTTPException(status_code=403, detail="forbidden")

        return _plugin_auth

    for slug, route_setups in pm.get_route_setups():
        if slug in mounted:
            continue
        try:
            from fastapi import Depends as _Depends
            from fastapi import FastAPI as _FAPI

            _auth_dep = _make_plugin_auth_dep(_plugin_auth_token)
            sub_app = _FAPI(dependencies=[_Depends(_auth_dep)])
            for setup_fn in route_setups:
                setup_fn(sub_app)
            fastapi_app.mount(f"/plugin/{slug}", sub_app)
            mounted.add(slug)
            if isinstance(getattr(delegate, "_plugin_route_mount_errors", None), dict):
                delegate._plugin_route_mount_errors.pop(slug, None)
        except Exception as e:
            import sys as _sys

            _sys.stderr.write(f"[PLUGIN] Failed to mount routes for {slug}: {e}\n")
            if isinstance(getattr(delegate, "_plugin_route_mount_errors", None), dict):
                delegate._plugin_route_mount_errors[slug] = str(e)


def create_fastapi_app(
    delegate: Any,
    html_content: str,
    temp_upload_dir: str,
    get_voice_prefs: Callable[[], Any] | None = None,
    get_theme: Callable[[], str] | None = None,
    live_view_event_bus: Any = None,
    activity_bus: Any = None,
    auth_token: str | None = None,
    on_user_input_start: Callable[[], None] | None = None,
) -> FastAPI:
    """Build FastAPI app with chat routes. Delegate is the app (e.g. ChatAppDelegate). get_theme() -> 'light'|'dark'|'system'."""
    app_ref = delegate
    fastapi_app = FastAPI()
    _PREVIEW_MEDIA_TYPES = {
        ".png": "image/png",
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".gif": "image/gif",
        ".svg": "image/svg+xml",
        ".webp": "image/webp",
        ".mp4": "video/mp4",
        ".webm": "video/webm",
        ".mov": "video/quicktime",
        ".avi": "video/x-msvideo",
        ".m4v": "video/x-m4v",
        ".html": "text/html; charset=utf-8",
        ".htm": "text/html; charset=utf-8",
        ".txt": "text/plain; charset=utf-8",
    }
    _PREVIEW_DISPLAY_TYPES = {
        ".png": "image",
        ".jpg": "image",
        ".jpeg": "image",
        ".gif": "image",
        ".svg": "image",
        ".webp": "image",
        ".mp4": "video",
        ".webm": "video",
        ".mov": "video",
        ".avi": "video",
        ".m4v": "video",
        ".html": "html",
        ".htm": "html",
        ".txt": "text",
    }

    if not (auth_token or "").strip():
        from persistence import generate_secure_token

        auth_token = generate_secure_token()
        _log.warning(
            "create_fastapi_app: no auth_token provided — generated ephemeral token (fail-closed)",
        )
    fastapi_app.state.auth_token = auth_token

    if auth_token:
        _AUTH_EXEMPT_PATHS = frozenset({"/", "/live_view", "/health"})
        # Mesh peer file serve uses route-local auth; do not skip the global token gate for
        # /files/ or /attachment/ (DEF-032 — LAN exposure when bound 0.0.0.0).
        _AUTH_EXEMPT_PREFIXES = ("/mesh/file_transfers/serve/",)
        _AUTH_EXEMPT_MESH_POST_PATHS = frozenset({"/mesh/handshake", "/mesh/message"})
        # Binary/media GETs may be requested by native clients or embedded web views that
        # set non-local Origin/Referer (or file://). Token/cookie is still required; only
        # the localhost Origin/Referer gate is skipped for these prefixes.
        _AUTH_RELAX_ORIGIN_CHECK_GET_PREFIXES = (
            "/vault/attachments/",
            "/files/",
            "/attachment/",
        )
        # Webhooks carry their own token+HMAC auth (see utils/webhooks.py) and
        # are exempt from the primary cookie/Origin/Referer checks (external
        # services cannot set those). Route itself 404s when webhooks are disabled.
        # These prefixes perform their own route-level authentication. Browser
        # extension capture also needs route-local CORS handling because its
        # Origin is chrome-extension://..., not localhost.
        _AUTH_EXEMPT_POST_PREFIXES = ("/webhook/", "/browser_extension/")

        @fastapi_app.middleware("http")
        async def _auth_middleware(request: Request, call_next):
            path = request.url.path
            if request.method == "GET":
                if path in _AUTH_EXEMPT_PATHS or any(path.startswith(prefix) for prefix in _AUTH_EXEMPT_PREFIXES):
                    return await call_next(request)
                if any(path.startswith(prefix) for prefix in _AUTH_EXEMPT_POST_PREFIXES):
                    return await call_next(request)
            if request.method == "POST":
                if request.url.path in _AUTH_EXEMPT_MESH_POST_PATHS:
                    return await call_next(request)
                if any(path.startswith(prefix) for prefix in _AUTH_EXEMPT_POST_PREFIXES):
                    return await call_next(request)
            if request.method == "OPTIONS":
                if any(path.startswith(prefix) for prefix in _AUTH_EXEMPT_POST_PREFIXES):
                    return await call_next(request)
            if request.method == "DELETE":
                if any(path.startswith(prefix) for prefix in _AUTH_EXEMPT_POST_PREFIXES):
                    return await call_next(request)
            token = request.cookies.get("mugi_auth") or request.headers.get("x-auth-token")
            if token != auth_token:
                return JSONResponse(status_code=403, content={"error": "forbidden"})
            if request.method == "GET" and any(
                path.startswith(prefix) for prefix in _AUTH_RELAX_ORIGIN_CHECK_GET_PREFIXES
            ):
                return await call_next(request)
            origin = request.headers.get("origin") or ""
            referer = request.headers.get("referer") or ""
            if origin and not is_trusted_localhost_http_url(origin):
                return JSONResponse(status_code=403, content={"error": "forbidden"})
            if referer and not is_trusted_localhost_http_url(referer):
                return JSONResponse(status_code=403, content={"error": "forbidden"})
            return await call_next(request)

    def _set_attached_and_send(
        ar,
        file_list,
        message,
        research_mode=False,
        quick_panel=False,
        dashboard_surface=False,
    ):
        ar._attached_files = list(file_list)
        try:
            ar.onSend_text_(
                None,
                message,
                research_mode=research_mode,
                quick_panel=quick_panel,
                dashboard_surface=dashboard_surface,
            )
        except TypeError:
            try:
                ar.onSend_text_(None, message, research_mode=research_mode, quick_panel=quick_panel)
            except TypeError:
                ar.onSend_text_(None, message, research_mode=research_mode)

    # ---- Slash command dispatch ----
    _SLASH_COMMANDS: dict[str, callable] = {}
    _SLASH_HELP: dict[str, str] = {}

    def _register_slash(name, help_text: str = ""):
        def decorator(fn):
            _SLASH_COMMANDS[name] = fn
            if (help_text or "").strip():
                _SLASH_HELP[name] = (help_text or "").strip()
            return fn
        return decorator

    def _slash_message_timestamp() -> str:
        from datetime import datetime

        return datetime.now().strftime("%I:%M %p")

    def _persist_slash_user_message(delegate, thread_id, text: str) -> None:
        """Mirror a normal send: persist the operator's slash line and show it in chat."""
        tid = thread_id or getattr(delegate, "_current_thread_id", None) or "default"
        content = (text or "").strip()
        if not content:
            return
        try:
            from chat_storage import append_messages

            append_messages(tid, [{"role": "user", "content": content}])
        except Exception:
            _log.debug("slash user persist failed", exc_info=True)
        try:
            delegate._push_event(
                {
                    "type": "message_add",
                    "text": content,
                    "is_user": True,
                    "timestamp": _slash_message_timestamp(),
                    "image_urls": [],
                    "file_attachments": [],
                },
                tid,
            )
        except Exception:
            _log.debug("slash user push_event failed", exc_info=True)

    def _push_slash_message(delegate, thread_id, text: str) -> None:
        tid = thread_id or getattr(delegate, "_current_thread_id", None) or "default"
        try:
            delegate._push_event(
                {"type": "message_add", "text": text, "is_user": False, "timestamp": ""},
                tid,
            )
        except Exception:
            _log.debug("slash push_event failed", exc_info=True)

    def _remember_display_window_event(delegate, thread_id: str, event: dict) -> None:
        tid = thread_id or getattr(delegate, "_current_thread_id", None) or "default"
        try:
            by_thread = getattr(delegate, "_last_display_window_events", None)
            if not isinstance(by_thread, dict):
                by_thread = {}
            by_thread[tid] = dict(event)
            delegate._last_display_window_events = by_thread
            delegate._last_display_window_event = dict(event)
        except Exception:
            _log.debug("failed to remember display_window event", exc_info=True)

    def _push_display_window_event(delegate, thread_id: str, filename: str, title: str, content_type: str) -> None:
        port = getattr(delegate, "_http_port", None)
        if not port:
            raise RuntimeError("backend port is unavailable")
        safe_name = os.path.basename((filename or "").strip())
        if not safe_name or safe_name != filename:
            raise ValueError("invalid preview filename")
        event = {
            "type": "display_window",
            "url": f"http://localhost:{int(port)}/files/{quote(safe_name)}",
            "content_type": content_type,
            "title": title or "Preview",
        }
        _remember_display_window_event(delegate, thread_id, event)
        delegate._push_event(event, thread_id or getattr(delegate, "_current_thread_id", "default"))

    def _preview_output_dir() -> str:
        out = os.path.join(os.path.expanduser("~"), "Documents", "Mugi")
        os.makedirs(out, exist_ok=True)
        return out

    def _preview_content_type_for_filename(filename: str) -> str | None:
        return _PREVIEW_DISPLAY_TYPES.get(os.path.splitext(filename)[1].lower())

    def _stage_preview_file(requested: str) -> tuple[str | None, str | None, str | None]:
        """Return (basename_in_output_dir, content_type, error) for a supported preview file."""
        raw = (requested or "").strip()
        if not raw:
            return None, None, "Usage: `/preview [filename-or-path]`."
        output_dir = _preview_output_dir()
        expanded = os.path.expanduser(raw)
        if os.path.isabs(expanded):
            source = expanded
        else:
            source = os.path.join(output_dir, raw)
        source = os.path.abspath(os.path.normpath(source))
        if is_dangerous_path(source):
            return None, None, "That path is in a protected location and cannot be previewed."
        if not os.path.isfile(source):
            return None, None, f"Preview file not found: {raw}"
        base = os.path.basename(source).strip()
        if not base:
            return None, None, "Invalid preview filename."
        content_type = _preview_content_type_for_filename(base)
        if content_type is None:
            return None, None, "Unsupported preview type. Use HTML, text, image, or video files."
        real_source = os.path.realpath(source)
        output_real = os.path.realpath(output_dir)
        direct_output_path = os.path.realpath(os.path.join(output_dir, base))
        if real_source == direct_output_path:
            return base, content_type, None
        try:
            stat = os.stat(real_source)
            digest_src = f"{real_source}|{stat.st_mtime_ns}|{stat.st_size}".encode("utf-8", errors="replace")
            digest = hashlib.sha256(digest_src).hexdigest()[:12]
            staged_name = f"_preview_{digest}_{base}"
            staged_path = os.path.realpath(os.path.join(output_dir, staged_name))
            if not staged_path.startswith(output_real + os.sep):
                return None, None, "Could not create a safe preview path."
            shutil.copy2(real_source, staged_path)
            return staged_name, content_type, None
        except Exception as e:
            _log.warning("Failed to stage preview file %s: %s", source, e)
            return None, None, f"Could not stage preview file: {e}"

    def _stage_preview_upload(source_path: str, original_name: str) -> tuple[str | None, str | None, str | None]:
        """Stage an already-accepted multipart upload for preview using its original filename."""
        source = os.path.abspath(os.path.normpath(source_path or ""))
        if not os.path.isfile(source):
            return None, None, "Preview attachment is missing."
        safe_base = os.path.basename((original_name or "").strip())
        if not safe_base:
            safe_base = os.path.basename(source)
        safe_base = safe_base.replace("/", "_").replace("\\", "_")
        content_type = _preview_content_type_for_filename(safe_base)
        if content_type is None:
            return None, None, "Unsupported preview type. Use HTML, text, image, or video files."
        try:
            output_dir = _preview_output_dir()
            output_real = os.path.realpath(output_dir)
            stat = os.stat(source)
            digest_src = f"{source}|{stat.st_mtime_ns}|{stat.st_size}|{safe_base}".encode("utf-8", errors="replace")
            digest = hashlib.sha256(digest_src).hexdigest()[:12]
            staged_name = f"_preview_{digest}_{safe_base}"
            staged_path = os.path.realpath(os.path.join(output_dir, staged_name))
            if not staged_path.startswith(output_real + os.sep):
                return None, None, "Could not create a safe preview path."
            shutil.copy2(source, staged_path)
            return staged_name, content_type, None
        except Exception as e:
            _log.warning("Failed to stage preview upload %s: %s", source, e)
            return None, None, f"Could not stage preview attachment: {e}"

    def _is_bare_preview_command(text: str) -> bool:
        return (text or "").strip().lower() == "/preview"

    def _queue_slash_send(thread_id, delegate, text: str, *, research_mode: bool = False) -> None:
        """Queue onSend_text_ on the delegate's main thread (same as /send JSON path)."""

        def _run():
            tid = (thread_id or "").strip()
            if tid and hasattr(delegate, "_current_thread_id"):
                delegate._current_thread_id = tid
            delegate.onSend_text_(None, text, research_mode=research_mode)

        try:
            delegate._main_thread_queue.put(_run)
        except Exception:
            _log.warning("slash queue send failed", exc_info=True)

    def _format_doctor_status(report: dict) -> str:
        """Short plain-text summary of build_doctor_report() for chat."""
        lines: list[str] = []
        ok = bool(report.get("ok"))
        elapsed = report.get("elapsed_ms", 0)
        lines.append(f"Doctor: {'OK' if ok else 'issues'} ({elapsed} ms)")
        ap = report.get("autonomy_posture") or {}
        level = (ap.get("level") or "ok").strip()
        notes = ap.get("notes") or []
        if notes:
            lines.append(f"Autonomy posture: {level}")
            for n in notes[:6]:
                lines.append(f"  • {n}")
            if len(notes) > 6:
                lines.append(f"  … ({len(notes) - 6} more)")
        else:
            lines.append(f"Autonomy posture: {level}")
        warns = report.get("warnings") or []
        if warns:
            lines.append("Notes:")
            for w in warns[:6]:
                lines.append(f"  • {w}")
            if len(warns) > 6:
                lines.append(f"  … ({len(warns) - 6} more)")
        failed = [c for c in (report.get("checks") or []) if not c.get("ok")]
        if failed:
            lines.append("Failed checks:")
            for chk in failed[:8]:
                name = chk.get("name", "?")
                detail = chk.get("detail") or {}
                err = detail.get("error", "failed")
                lines.append(f"  • {name}: {err}")
            if len(failed) > 8:
                lines.append(f"  … ({len(failed) - 8} more)")
        return "\n".join(lines)

    def _handle_slash_command(text, thread_id, delegate):
        """Check if text is a slash command. Returns JSONResponse or None."""
        if not text or not text.startswith("/"):
            return None
        parts = text.split(None, 1)
        cmd = parts[0].lower()
        arg = parts[1] if len(parts) > 1 else ""
        handler = _SLASH_COMMANDS.get(cmd)
        if handler is None:
            return None
        try:
            handler(thread_id or getattr(delegate, "_current_thread_id", "default"), arg, delegate)
        except Exception as e:
            _log.warning("Slash command %s error: %s", cmd, e)
            try:
                delegate._push_event(
                    {"type": "message_add", "text": f"Command error: {e}", "is_user": False, "timestamp": ""},
                    thread_id or getattr(delegate, "_current_thread_id", "default"),
                )
            except Exception:
                _log.debug("Slash command push_event failed", exc_info=True)
            return JSONResponse(
                content={"ok": False, "command": cmd, "error": str(e)[:200]},
                status_code=500,
            )
        return JSONResponse(content={"ok": True, "command": cmd})

    def _maybe_slash(text, tid, delegate):
        """If text is a slash command, handle it and return JSONResponse. Else None."""
        return _handle_slash_command(text, tid, delegate)

    @_register_slash(
        "/preview",
        help_text="Reopen the last Results window, or open a supported local file by name/path.",
    )
    def _cmd_preview(thread_id, arg, delegate):
        body = (arg or "").strip()
        if not body:
            by_thread = getattr(delegate, "_last_display_window_events", None)
            event = None
            if isinstance(by_thread, dict):
                event = by_thread.get(thread_id)
            if event is None:
                event = getattr(delegate, "_last_display_window_event", None)
            if isinstance(event, dict) and event.get("type") == "display_window":
                delegate._push_event(dict(event), thread_id)
            else:
                _push_slash_message(delegate, thread_id, "No previous Results preview is available yet.")
            return
        try:
            parts = shlex.split(body)
        except ValueError as e:
            _push_slash_message(delegate, thread_id, f"Could not parse preview path: {e}")
            return
        if len(parts) != 1:
            _push_slash_message(delegate, thread_id, "Usage: `/preview [filename-or-path]`.")
            return
        filename, content_type, error = _stage_preview_file(parts[0])
        if error or not filename or not content_type:
            _push_slash_message(delegate, thread_id, error or "Could not open preview.")
            return
        title = os.path.basename(parts[0]).strip() or "Preview"
        try:
            _push_display_window_event(delegate, thread_id, filename, title, content_type)
        except Exception as e:
            _push_slash_message(delegate, thread_id, f"Could not open preview: {e}")

    # ---- /compact slash command ----
    _compacting_threads: set[str] = set()
    _compacting_lock = threading.Lock()

    @_register_slash(
        "/compact",
        help_text="Summarize older messages into a context anchor (transcript unchanged; saves context).",
    )
    def _cmd_compact(thread_id, arg, delegate):
        with _compacting_lock:
            if thread_id in _compacting_threads:
                delegate._push_event(
                    {"type": "message_add", "text": "Compaction already in progress for this thread.", "is_user": False, "timestamp": ""},
                    thread_id,
                )
                return
            running_ids = getattr(delegate, "_running_thread_ids", set())
            if thread_id in running_ids:
                delegate._push_event(
                    {"type": "message_add", "text": "Cannot compact while the agent is active. Wait for the current turn to finish.", "is_user": False, "timestamp": ""},
                    thread_id,
                )
                return
            _compacting_threads.add(thread_id)

        def _do_compact():
            try:
                delegate._push_event(
                    {"type": "message_add", "text": "Compacting conversation...", "is_user": False, "timestamp": ""},
                    thread_id,
                )
                from chat_storage import get_messages, upsert_anchor
                msgs = get_messages(thread_id, include_row_id=True)
                ua_msgs = [m for m in msgs if m.get("role") in ("user", "assistant")]
                if len(ua_msgs) < 6:
                    delegate._push_event(
                        {"type": "message_add", "text": "Not enough messages to compact.", "is_user": False, "timestamp": ""},
                        thread_id,
                    )
                    return
                from context_summarizer import compute_anchor_update
                result = compute_anchor_update(ua_msgs, keep_recent=20)
                if result is None:
                    delegate._push_event(
                        {"type": "message_add", "text": "Compaction produced no result (conversation may be too short).", "is_user": False, "timestamp": ""},
                        thread_id,
                    )
                    return
                upsert_anchor(thread_id, result["through_message_id"], result["summary_text"])
                delegate._push_event(
                    {"type": "message_add", "text": "Context anchor updated (transcript unchanged).", "is_user": False, "timestamp": ""},
                    thread_id,
                )
            except Exception as e:
                _log.error("Compact failed for %s: %s", thread_id, e)
                try:
                    delegate._push_event(
                        {"type": "message_add", "text": f"Compaction failed: {e}", "is_user": False, "timestamp": ""},
                        thread_id,
                    )
                except Exception:
                    pass
            finally:
                with _compacting_lock:
                    _compacting_threads.discard(thread_id)

        threading.Thread(target=_do_compact, daemon=True, name=f"compact-{thread_id[:8]}").start()

    # ---- /btw ephemeral side-question ----
    # Per-origin-thread in-flight registry: a second /btw on the same origin
    # cancels the first so compounding utility-model load can't build up.
    # See utils/btw.py for the invisibility contract.
    _btw_inflight: dict[str, dict] = {}
    _btw_lock = threading.Lock()

    @_register_slash(
        "/btw",
        help_text="Ephemeral side question — not saved to chat, session search, or long-term memory.",
    )
    def _cmd_btw(thread_id, arg, delegate):
        """`/btw <question>` — one-off side question.

        The question and answer are never persisted to chat_storage /
        session_search / active_memory / consolidator / dreaming. See
        utils/btw.py for the full invisibility contract.

        Events emitted (all transient; none hit chat_storage):
            btw_echo    — user's question, immediately
            btw_pending — placeholder while the model runs
            btw_result  — final answer (or failure message)
        """
        question = (arg or "").strip()
        if not question:
            try:
                delegate._push_event(
                    {
                        "type": "btw_result",
                        "text": (
                            "Usage: `/btw <question>` — ask an ephemeral side "
                            "question that won't be saved to chat or memory."
                        ),
                        "is_user": False,
                        "is_btw": True,
                        "status": "help",
                        "timestamp": "",
                    },
                    thread_id,
                )
            except Exception:
                pass
            return

        from utils.btw import handle_btw_async, new_btw_thread_id

        origin = thread_id or "default"
        btw_tid = new_btw_thread_id()

        # Cancel any in-flight /btw for the same origin thread.
        with _btw_lock:
            prior = _btw_inflight.get(origin)
            if prior is not None:
                prior["cancelled"] = True
                try:
                    from utils.task_ledger import record_run as _ledger_record
                    _ledger_record(
                        kind="btw",
                        task_name="side_question",
                        thread_id=str(prior.get("btw_thread_id") or ""),
                        status="skipped",
                        skip_reason="cancelled_by_next_btw",
                        duration_ms=0,
                        meta={"origin_thread_id": origin},
                    )
                except Exception:
                    pass

        state: dict = {"cancelled": False, "btw_thread_id": btw_tid}
        with _btw_lock:
            _btw_inflight[origin] = state

        def _push(event: dict) -> None:
            try:
                delegate._push_event(event, thread_id)
            except Exception:
                _log.debug("btw push_event failed", exc_info=True)

        def _deliver(text, *, is_btw: bool = True):
            with _btw_lock:
                if state.get("cancelled"):
                    return
                if _btw_inflight.get(origin) is state:
                    _btw_inflight.pop(origin, None)
            _push({
                "type": "btw_result",
                "text": str(text or ""),
                "is_user": False,
                "is_btw": bool(is_btw),
                "timestamp": "",
            })

        _push({
            "type": "btw_echo",
            "text": question,
            "is_user": True,
            "is_btw": True,
            "timestamp": "",
        })
        _push({
            "type": "btw_pending",
            "text": "Thinking…",
            "is_user": False,
            "is_btw": True,
            "timestamp": "",
        })

        def _is_cancelled() -> bool:
            with _btw_lock:
                return bool(state.get("cancelled"))

        handle_btw_async(
            question=question,
            parent_thread_id=thread_id,
            deliver=_deliver,
            btw_thread_id=btw_tid,
            cancellation_check=_is_cancelled,
        )

    @_register_slash(
        "/mesh-send",
        help_text="Open Mesh encrypted file transfer (pick peer and files; deterministic, no agent).",
    )
    def _cmd_mesh_send(thread_id, arg, delegate):
        try:
            from core.config import get_mesh_config

            if not bool(get_mesh_config().get("file_transfer_enabled", True)):
                _push_slash_message(
                    delegate,
                    thread_id,
                    "Mesh file transfer is disabled in config (`[mesh] file_transfer_enabled = false`).",
                )
                return
            delegate._push_event({"type": "ui_open_mesh_send"}, thread_id)
        except Exception as e:
            _push_slash_message(delegate, thread_id, f"Could not open Mesh file send: {e}")

    @_register_slash(
        "/research",
        help_text="Send the rest of the line with Research mode on (research tools enabled).",
    )
    def _cmd_research(thread_id, arg, delegate):
        body = (arg or "").strip()
        if not body:
            _push_slash_message(
                delegate,
                thread_id,
                "Usage: `/research <question>` — sends your message with Research mode enabled.",
            )
            return
        _queue_slash_send(thread_id, delegate, body, research_mode=True)

    @_register_slash("/help", help_text="List available slash commands and short descriptions.")
    def _cmd_help(thread_id, arg, delegate):
        # Markdown `##` + `- ` lines: MarkdownNSRenderer merges plain lines with spaces;
        # bullet lines render as a readable list with one item per row.
        lines = ["## Slash commands", ""]
        for cmd in sorted(_SLASH_HELP.keys()):
            desc = (_SLASH_HELP.get(cmd) or "").strip()
            if desc:
                lines.append(f"- `{cmd}` — {desc}")
            else:
                lines.append(f"- `{cmd}`")
        _push_slash_message(delegate, thread_id, "\n".join(lines))

    # Prevents a rapid `/status` double-submit (or `/status` while the
    # Settings Diagnostics pane is already running doctor) from kicking off
    # two `build_doctor_report()` calls in parallel. doctor touches the LLM
    # backend, filesystem, keychain, and optional mesh peers; overlapping
    # runs are expensive and can race each other. The lock is cheap — each
    # run is short — and if the lock is already held we immediately reply
    # in-thread so the operator isn't left wondering whether their message
    # got lost.
    _status_lock = threading.Lock()

    @_register_slash(
        "/status",
        help_text="Run a quick doctor report (health checks, autonomy posture) in chat.",
    )
    def _cmd_status(thread_id, arg, delegate):
        if not _status_lock.acquire(blocking=False):
            _push_slash_message(
                delegate,
                thread_id,
                "Status is already running — the result will appear here shortly.",
            )
            return

        def _run():
            try:
                from diagnostics import build_doctor_report

                report = build_doctor_report()
                text = _format_doctor_status(report)
            except Exception as e:
                text = f"Status failed: {e}"
            finally:
                _status_lock.release()
            _push_slash_message(delegate, thread_id, text)

        threading.Thread(target=_run, daemon=True, name="slash-status").start()

    @_register_slash(
        "/plan",
        help_text="Task planning: ask the agent to use structured task plan tools for this thread.",
    )
    def _cmd_plan(thread_id, arg, delegate):
        goal = (arg or "").strip()
        if not goal:
            instruction = (
                "[/plan] The user invoked /plan. Call get_task_plan to read the current plan. "
                "If they have no concrete goal yet, ask briefly what they want broken into steps. "
                "Otherwise use update_task_plan (replace or merge) with clear step statuses."
            )
        else:
            instruction = (
                f"[/plan] The user wants a structured task plan for: {goal}. "
                "Use update_task_plan with concrete steps; mark the first actionable step "
                "in_progress when you start it."
            )
        _queue_slash_send(thread_id, delegate, instruction, research_mode=False)

    @_register_slash(
        "/council",
        help_text=(
            "Convene council on a question (Researcher → Verifier → Contrarian). "
            "Works even when Council mode is off in Settings."
        ),
    )
    def _cmd_council(thread_id, arg, delegate):
        tid = str(thread_id or "").strip() or "(default)"
        frame = (arg or "").strip()
        if not frame:
            _push_slash_message(
                delegate,
                thread_id,
                "Usage: `/council <question>` — convenes the council pipeline for this thread. "
                "Example: `/council Should we adopt async I/O for the HTTP layer?` "
                "Works even when **Settings → General → Council mode** is off.",
            )
            return
        try:
            from utils.council_offers import convene_council_via_slash

            _persist_slash_user_message(delegate, thread_id, f"/council {frame}")
            out = convene_council_via_slash(thread_id=tid, council_frame=frame)
            try:
                delegate._push_event({"type": "thread_state_reload"}, thread_id)
            except Exception:
                _log.debug("slash /council thread_state_reload failed", exc_info=True)
            honest = str(out.get("message") or "").strip()
            if honest:
                _push_slash_message(delegate, thread_id, honest)
            else:
                _push_slash_message(delegate, thread_id, "Council convened.")
        except ValueError as exc:
            err = str(exc).strip() or "preflight refused"
            _log.info("slash /council refused thread_id=%s: %s", tid, err)
            try:
                from utils.council_offers import deliver_council_spawn_refusal

                deliver_council_spawn_refusal(tid, reason=err)
            except Exception:
                _log.debug("slash /council refusal bubble failed", exc_info=True)
            _push_slash_message(delegate, thread_id, err)
        except Exception as exc:
            err = f"Could not convene council: {type(exc).__name__}: {exc}"
            _log.warning(
                "slash /council failed thread_id=%s frame=%r: %s",
                tid,
                frame[:120],
                exc,
                exc_info=True,
            )
            try:
                from utils.council_offers import deliver_council_spawn_refusal

                deliver_council_spawn_refusal(tid, reason=err)
            except Exception:
                _log.debug("slash /council refusal bubble failed", exc_info=True)
            _push_slash_message(delegate, thread_id, err)

    @_register_slash(
        "/kanban",
        help_text="Add a durable Kanban task from chat (`/kanban <title>` — assignee general, priority P3).",
    )
    def _cmd_kanban(thread_id, arg, delegate):
        tid = str(thread_id or "").strip() or "(default)"
        raw = (arg or "").strip()
        title_excerpt = raw.replace("\n", " ").strip()
        if len(title_excerpt) > 160:
            title_excerpt = title_excerpt[:159] + "…"

        if not raw:
            _log.info("slash /kanban help (no title) thread_id=%s", tid)
            _push_slash_message(
                delegate,
                thread_id,
                "Usage: `/kanban <title>` — adds a task to the encrypted Kanban board "
                "(assignee **general**, priority **P3**). "
                "Use **View → Kanban Board** or Settings → Kanban to manage tasks.",
            )
            return
        try:
            import kanban_storage as _ks

            task = _ks.create_task(
                raw,
                "",
                assignee="general",
                tenant="default",
                priority=3,
                tags=None,
                metadata=None,
                origin_thread_id=tid if tid != "(default)" else "",
            )
        except RuntimeError as exc:
            _log.warning(
                "slash /kanban failed (kanban unavailable) thread_id=%s title=%r error=%s",
                tid,
                title_excerpt,
                exc,
            )
            _push_slash_message(delegate, thread_id, f"Kanban is unavailable: {exc}")
            return
        except ValueError as exc:
            _log.warning(
                "slash /kanban failed (validation/rate-limit) thread_id=%s title=%r error=%s",
                tid,
                title_excerpt,
                exc,
            )
            _push_slash_message(delegate, thread_id, str(exc))
            return
        except Exception as exc:
            _log.warning(
                "slash /kanban failed (unexpected) thread_id=%s title=%r error=%s",
                tid,
                title_excerpt,
                exc,
                exc_info=True,
            )
            _push_slash_message(delegate, thread_id, f"Could not create Kanban task: {exc}")
            return
        try:
            from utils.kanban_dispatcher import nudge_dispatcher as _nudge

            _nudge()
        except Exception as nudge_exc:
            _log.warning(
                "slash /kanban create ok but nudge_dispatcher failed task_id=%s: %s",
                task.get("id"),
                nudge_exc,
                exc_info=True,
            )
        kt_id = task.get("id") or "?"
        kt_title = task.get("title") or raw
        kt_status = task.get("status") or "?"
        _log.info(
            "slash /kanban created task_id=%s status=%s assignee=%s tenant=%s "
            "origin_thread_id=%s title=%r",
            kt_id,
            kt_status,
            task.get("assignee"),
            task.get("tenant"),
            tid if tid != "(default)" else "",
            title_excerpt,
        )
        _push_slash_message(
            delegate,
            thread_id,
            f"Added Kanban task **{kt_title}** (`{kt_id}`), status **{kt_status}**.",
        )

    @_register_slash(
        "/neko",
        help_text="Neko shared browser: dashboard, link, install, or open an https URL.",
    )
    def _cmd_neko(thread_id, arg, delegate):
        rest = (arg or "").strip()
        if not rest:
            instruction = (
                "[/neko] The user invoked /neko. If open_neko_dashboard is in your tool list, "
                "call it once to open the Neko panel. Do not call it repeatedly in the same flow."
            )
        else:
            parts = rest.split(None, 1)
            first = parts[0].lower()
            tail = parts[1].strip() if len(parts) > 1 else ""
            if tail.startswith("http://") or tail.startswith("https://"):
                instruction = (
                    "[/neko] Navigate the shared Neko browser to this external URL with neko_navigate "
                    f"(never navigate to the Neko room URL or the Neko UI on loopback): {tail}"
                )
            elif rest.startswith("http://") or rest.startswith("https://"):
                instruction = (
                    "[/neko] Navigate the shared Neko browser to this external URL with neko_navigate "
                    f"(never navigate to the Neko room URL or the Neko UI on loopback): {rest}"
                )
            elif first in ("link", "room", "url"):
                instruction = (
                    "[/neko] The user wants the Neko room URL for their Mac browser: call get_neko_room_link."
                )
            elif first == "install":
                instruction = (
                    "[/neko] The user wants to install or start Neko: call install_neko if it is in your tool list, "
                    "or open_neko_dashboard once so they can use Install from the panel."
                )
            elif first in ("dashboard", "panel", "ui"):
                instruction = (
                    "[/neko] The user wants the Neko dashboard: call open_neko_dashboard once if available."
                )
            else:
                instruction = (
                    f"[/neko] Assist with Neko: {rest}. Use only Neko-related tools that appear in your tool list. "
                    "Never neko_navigate to the room URL or the Neko UI on loopback."
                )
        _queue_slash_send(thread_id, delegate, instruction, research_mode=False)

    @fastapi_app.get("/", response_class=HTMLResponse)
    def serve_html():
        resp = HTMLResponse(content=html_content, media_type="text/html; charset=utf-8")
        resp.headers["Content-Security-Policy"] = (
            "default-src 'self'; script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; "
            "style-src 'self' 'unsafe-inline'; img-src 'self' data: blob:; "
            "connect-src 'self'; frame-src 'none'; object-src 'none'"
        )
        if auth_token:
            resp.set_cookie("mugi_auth", auth_token, httponly=True, samesite="strict", path="/")
        return resp

    @fastapi_app.get("/open")
    async def open_url(url: str = Query(..., alias="url")):
        import webbrowser
        try:
            decoded = unquote(url)
            if not (decoded.startswith("http://") or decoded.startswith("https://")):
                return JSONResponse(content={"ok": False, "error": "invalid_url"}, status_code=400)
            loop = asyncio.get_running_loop()
            opened = await loop.run_in_executor(None, lambda: webbrowser.open(decoded))
            if opened is False:
                return JSONResponse(content={"ok": False, "error": "open_failed"}, status_code=500)
            return JSONResponse(content={"ok": True})
        except Exception as exc:
            _log.debug("Failed to open URL in browser", exc_info=True)
            return JSONResponse(
                content={"ok": False, "error": str(exc)[:200]},
                status_code=500,
            )

    @fastapi_app.get("/open_in_finder")
    async def open_in_finder(path: str = Query(..., alias="path")):
        """Open a path in Finder (folder or reveal file). Path must pass is_dangerous_path."""
        import subprocess
        try:
            decoded = unquote(path)
            if not decoded or is_dangerous_path(decoded):
                return JSONResponse(content={"ok": False, "error": "forbidden"}, status_code=403)
            expanded = os.path.expanduser(decoded)
            loop = asyncio.get_running_loop()

            def _run_open() -> int:
                proc = subprocess.run(["open", "--", expanded], check=False, timeout=5)
                return int(proc.returncode)

            rc = await loop.run_in_executor(None, _run_open)
            if rc != 0:
                return JSONResponse(content={"ok": False, "error": "open_failed"}, status_code=500)
            return JSONResponse(content={"ok": True})
        except Exception as exc:
            _log.debug("Failed to open path in Finder", exc_info=True)
            return JSONResponse(
                content={"ok": False, "error": str(exc)[:200]},
                status_code=500,
            )

    @fastapi_app.get("/attachment/id/{attachment_id}")
    def serve_attachment_by_id(attachment_id: str):
        """Serve an attached file by its registry attachment_id."""
        rec = _attach_reg.get_record(attachment_id)
        if rec is None:
            return JSONResponse(content={"error": "unknown attachment_id"}, status_code=404)
        real_path = os.path.realpath(rec.abs_path)
        if not os.path.isfile(real_path):
            return JSONResponse(content={"error": "file missing"}, status_code=404)
        return FileResponse(real_path)

    @fastapi_app.get("/attachment/{filename}")
    def serve_attachment(filename: str):
        """Serve an attached image file from temp_uploads for display in user bubbles."""
        safe_name = os.path.basename(filename).strip()
        if not safe_name or safe_name != filename:
            return JSONResponse(content={"error": "invalid"}, status_code=400)
        path = os.path.join(temp_upload_dir, safe_name)
        try:
            real_path = os.path.realpath(path)
            real_dir = os.path.realpath(temp_upload_dir)
            if not real_path.startswith(real_dir + os.sep):
                return JSONResponse(content={"error": "forbidden"}, status_code=403)
            if not os.path.isfile(real_path):
                return JSONResponse(content={"error": "not found"}, status_code=404)
            return FileResponse(real_path)
        except Exception:
            return JSONResponse(content={"error": "not found"}, status_code=404)

    _files_dir = os.path.join(os.path.expanduser("~"), "Documents", "Mugi")

    @fastapi_app.get("/files/{filename}")
    def serve_output_file(filename: str):
        """Serve an agent-generated preview file from ~/Documents/Mugi/."""
        safe_name = os.path.basename(filename).strip()
        if not safe_name or safe_name != filename:
            return JSONResponse(content={"error": "invalid"}, status_code=400)
        ext = os.path.splitext(safe_name)[1].lower()
        media_type = _PREVIEW_MEDIA_TYPES.get(ext)
        if media_type is None:
            return JSONResponse(content={"error": "forbidden file type"}, status_code=403)
        path = os.path.join(_files_dir, safe_name)
        try:
            real_path = os.path.realpath(path)
            real_dir = os.path.realpath(_files_dir)
            if not real_path.startswith(real_dir + os.sep):
                return JSONResponse(content={"error": "forbidden"}, status_code=403)
            if not os.path.isfile(real_path):
                return JSONResponse(content={"error": "not found"}, status_code=404)
            return FileResponse(
                real_path,
                media_type=media_type,
                headers={"X-Content-Type-Options": "nosniff"},
            )
        except Exception:
            return JSONResponse(content={"error": "not found"}, status_code=404)

    # Upload size limit: configurable via config, default 500 MB (supports large scanned PDFs/videos).
    def _get_max_upload_bytes() -> int:
        try:
            from core.config import get_config
            mb = int((get_config().get("general") or {}).get("max_upload_mb", 500))
            return mb * 1024 * 1024
        except Exception:
            return 500 * 1024 * 1024

    @fastapi_app.post("/send")
    async def send(request: Request):
        if callable(on_user_input_start):
            try:
                on_user_input_start()
            except Exception:
                _log.debug("on_user_input_start callback failed in /send", exc_info=True)
        content_type = (request.headers.get("content-type") or "").lower()
        if "multipart/form-data" in content_type:
            try:
                form = await request.form()
                message = (form.get("message") or "").strip()
                rm = (form.get("research_mode") or "").lower() == "true"
                quick, dash_surf = _parse_send_surface(form.get("surface"))
                slash_result = _maybe_slash(message, "", app_ref)
                if slash_result is not None:
                    return slash_result
                file_list = []
                for key in form.keys():
                    if key in ("message", "research_mode", "surface"):
                        continue
                    parts = form.getlist(key) if hasattr(form, "getlist") else [form.get(key)]
                    for part in parts:
                        if getattr(part, "filename", None) and hasattr(part, "read"):
                            max_bytes = _get_max_upload_bytes()
                            data = await part.read(max_bytes + 1)
                            if len(data) > max_bytes:
                                return JSONResponse(status_code=413, content={"error": "file too large"})
                            # Sanitize filename: strip path separators, keep only safe characters.
                            import re as _re_routes
                            raw_name = part.filename or "upload"
                            safe_filename = _re_routes.sub(r'[^\w.\-]', '_', os.path.basename(raw_name))[:255] or "upload"
                            ext = os.path.splitext(safe_filename)[1] or ""
                            fd, tmp_path = tempfile.mkstemp(suffix=ext, prefix="chat_attach_")
                            try:
                                os.write(fd, data)
                            finally:
                                os.close(fd)
                            file_list.append((safe_filename, tmp_path))
                            _attach_reg.register(
                                abs_path=os.path.abspath(tmp_path),
                                original_name=safe_filename,
                                thread_id="",
                                source_route="/send",
                            )
                app_ref._main_thread_queue.put(
                    lambda m=message, fl=file_list, r=rm, q=quick, ds=dash_surf: _set_attached_and_send(
                        app_ref, fl, m, research_mode=r, quick_panel=q, dashboard_surface=ds
                    )
                )
            except Exception as exc:
                _log.debug("Failed to process multipart /send request", exc_info=True)
                return JSONResponse(content={"ok": False, "error": f"multipart send failed: {exc}"}, status_code=500)
        else:
            try:
                body = await request.json()
            except Exception:
                _log.debug("Failed to parse /send JSON body", exc_info=True)
                return JSONResponse(content={"ok": False, "error": "invalid JSON body"}, status_code=400)
            text = (body.get("text") or "").strip()
            rm = bool(body.get("research_mode", False))
            tid = (body.get("thread_id") or "").strip()
            quick, dash_surf = _parse_send_surface(body.get("surface"))
            slash_result = _maybe_slash(text, tid, app_ref)
            if slash_result is not None:
                return slash_result
            def _do_send(t=text, r=rm, thread_id=tid, q=quick, ds=dash_surf):
                if thread_id and hasattr(app_ref, "_current_thread_id"):
                    app_ref._current_thread_id = thread_id
                try:
                    app_ref.onSend_text_(None, t, research_mode=r, quick_panel=q, dashboard_surface=ds)
                except TypeError:
                    try:
                        app_ref.onSend_text_(None, t, research_mode=r, quick_panel=q)
                    except TypeError:
                        app_ref.onSend_text_(None, t, research_mode=r)
            app_ref._main_thread_queue.put(_do_send)
        return JSONResponse(content={"ok": True})

    @fastapi_app.post("/message")
    async def handle_message(
        text: str = Form("", alias="text"),
        thread_id: str = Form(""),
        research_mode: str = Form(""),
        surface: str = Form(""),
        files: list[UploadFile] = File(default=[]),
    ):
        # `/message` is the multipart-capable counterpart of `/send`. Historically
        # it diverged: it ignored `research_mode`, skipped the slash-command
        # dispatcher, and swallowed errors silently. We now keep behavioural
        # parity with `/send` for text/thread/research/slash handling — the only
        # remaining difference is attachment intake, which is `/message`'s whole
        # reason to exist. Prefer `/send` for text-only paths because its
        # response shape is simpler.
        if callable(on_user_input_start):
            try:
                on_user_input_start()
            except Exception:
                _log.debug("on_user_input_start callback failed in /message", exc_info=True)
        tid = (thread_id or "").strip()
        rm = str(research_mode or "").strip().lower() in {"1", "true", "yes", "on"}
        body_text = (text or "").strip()
        quick, dash_surf = _parse_send_surface(surface)
        # Slash-command short-circuit: mirrors `/send` so users can type
        # `/compact` or `/summary` from any surface without files being
        # required to exist. Running before attachment processing avoids
        # unnecessary disk writes when the slash handler intercepts. Bare
        # `/preview` is the one exception: a dropped file is the command target.
        preview_attachment_mode = _is_bare_preview_command(body_text) and bool(files)
        if not preview_attachment_mode:
            slash_result = _maybe_slash(body_text, tid, app_ref)
            if slash_result is not None:
                return slash_result
        file_paths = []
        file_original_names: list[str] = []
        file_errors: list[str] = []
        for file in files:
            if file.filename:
                safe_name = (file.filename or "").replace("/", "_").replace("\\", "_").strip() or "upload"
                ext = os.path.splitext(safe_name)[1] or ""
                path = os.path.join(temp_upload_dir, str(uuid.uuid4()) + ext)
                try:
                    max_bytes = _get_max_upload_bytes()
                    data = await file.read(max_bytes + 1)
                    if len(data) > max_bytes:
                        return JSONResponse(status_code=413, content={"error": "file too large"})
                    with open(path, "wb") as f:
                        f.write(data)
                    file_paths.append(path)
                    _attach_reg.register(
                        abs_path=os.path.abspath(path),
                        original_name=safe_name,
                        thread_id=tid,
                        source_route="/message",
                    )
                    file_original_names.append(safe_name)
                except Exception as exc:
                    _log.debug("Failed to process uploaded file in /message", exc_info=True)
                    file_errors.append(f"{safe_name}: {exc}")
        if preview_attachment_mode:
            if file_errors and not file_paths:
                return JSONResponse(
                    status_code=500,
                    content={"ok": False, "command": "/preview", "error": "attachment processing failed", "details": file_errors},
                )
            if not file_paths:
                _push_slash_message(app_ref, tid, "Drop a supported file with `/preview` to open it in Results.")
                return JSONResponse(content={"ok": True, "command": "/preview"})
            filename, content_type, error = _stage_preview_upload(file_paths[0], file_original_names[0] if file_original_names else os.path.basename(file_paths[0]))
            if error or not filename or not content_type:
                _push_slash_message(app_ref, tid, error or "Could not open preview attachment.")
                return JSONResponse(content={"ok": True, "command": "/preview"})
            title = file_original_names[0] if file_original_names else os.path.basename(filename)
            try:
                _push_display_window_event(app_ref, tid, filename, title, content_type)
            except Exception as e:
                _push_slash_message(app_ref, tid, f"Could not open preview attachment: {e}")
            return JSONResponse(content={"ok": True, "command": "/preview", "file_paths": file_paths, "file_errors": file_errors})
        file_list = [(os.path.basename(p), p) for p in file_paths]
        if file_list or body_text:
            def _do_send(fl=file_list, m=body_text, thread_id=tid, r=rm, q=quick, ds=dash_surf):
                if thread_id and hasattr(app_ref, "_current_thread_id"):
                    app_ref._current_thread_id = thread_id
                if fl:
                    app_ref._attached_files = list(fl)
                try:
                    app_ref.onSend_text_(None, m, research_mode=r, quick_panel=q, dashboard_surface=ds)
                except TypeError:
                    try:
                        app_ref.onSend_text_(None, m, research_mode=r, quick_panel=q)
                    except TypeError:
                        try:
                            app_ref.onSend_text_(None, m, research_mode=r)
                        except TypeError:
                            app_ref.onSend_text_(None, m)
            app_ref._main_thread_queue.put(_do_send)
        if file_errors and not file_list and not body_text:
            return JSONResponse(
                status_code=500,
                content={"ok": False, "error": "attachment processing failed", "details": file_errors},
            )
        return JSONResponse(content={
            "ok": len(file_errors) == 0,
            "status": "received",
            "message": text,
            "research_mode": rm,
            "file_paths": file_paths,
            "file_errors": file_errors,
        })

    @fastapi_app.post("/sync_input")
    async def sync_input(request: Request):
        await request.body()
        return JSONResponse(content={"ok": True})

    @fastapi_app.post("/trigger_attach")
    async def trigger_attach(request: Request):
        await request.body()
        app_ref._main_thread_queue.put(lambda: app_ref.onAttachFile_(None))
        return JSONResponse(content={"ok": True})

    @fastapi_app.post("/remove_attach")
    async def remove_attach(request: Request):
        try:
            body = await request.json()
        except Exception:
            body = {}
        idx = body.get("index", 0)
        app_ref._main_thread_queue.put(lambda i=idx: app_ref.removeAttachmentAtIndex_(i))
        return JSONResponse(content={"ok": True})

    @fastapi_app.post("/trigger_stop")
    async def trigger_stop(request: Request):
        try:
            body = await request.json()
        except Exception:
            body = {}
        tid = (body.get("thread_id") or "").strip() if isinstance(body, dict) else ""
        def _do_stop(_tid=tid):
            try:
                app_ref.onStop_(None, thread_id=_tid)
            except TypeError:
                app_ref.onStop_(None)
        app_ref._main_thread_queue.put(_do_stop)
        return JSONResponse(content={"ok": True})

    @fastapi_app.post("/regenerate")
    async def regenerate(request: Request):
        await request.body()
        app_ref._main_thread_queue.put(lambda: app_ref.onRegenerateLastReply_(None))
        return JSONResponse(content={"ok": True})

    @fastapi_app.post("/recents")
    async def recents_snapshot(request: Request):
        """Swift client posts the user's recent documents here so the
        ``get_recent_files`` tool can return them without scraping disk.

        Body: ``{"items": [{"path": "/abs/path", "name": "…", "last_opened": "iso8601", "is_icloud_placeholder": bool}], "limit": int}``.
        The server trims to at most 50 and expects ``X-Auth-Token`` (enforced
        by the auth middleware); there is no unauthenticated fallback.
        """
        try:
            body = await request.json()
        except Exception:
            return JSONResponse(content={"ok": False, "error": "invalid JSON body"}, status_code=400)
        if not isinstance(body, dict):
            return JSONResponse(content={"ok": False, "error": "body must be an object"}, status_code=400)
        raw = body.get("items")
        if not isinstance(raw, list):
            return JSONResponse(content={"ok": False, "error": "items must be a list"}, status_code=400)
        cleaned: list[dict] = []
        for entry in raw[:50]:
            if not isinstance(entry, dict):
                continue
            path = str(entry.get("path") or "").strip()
            if not path:
                continue
            item = {"path": path}
            name = str(entry.get("name") or "").strip()
            if name:
                item["name"] = name[:255]
            last_opened = str(entry.get("last_opened") or "").strip()
            if last_opened:
                item["last_opened"] = last_opened[:64]
            if bool(entry.get("is_icloud_placeholder")):
                item["is_icloud_placeholder"] = True
            cleaned.append(item)
        if not hasattr(app_ref, "_recents_lock") or not hasattr(app_ref, "_recents_items"):
            return JSONResponse(content={"ok": False, "error": "recents store unavailable"}, status_code=503)
        with app_ref._recents_lock:
            app_ref._recents_items = cleaned
            import time as _t
            app_ref._recents_updated_at = _t.time()
        return JSONResponse(content={"ok": True, "count": len(cleaned)})

    @fastapi_app.post("/message_action")
    async def message_action(request: Request):
        try:
            body = await request.json()
        except Exception:
            body = {}
        action = (body.get("action") or "").strip() or "copy"
        message_index = body.get("message_index")
        if message_index is not None and not isinstance(message_index, int):
            try:
                message_index = int(message_index)
            except (TypeError, ValueError):
                message_index = None
        app_ref._main_thread_queue.put(lambda a=action, idx=message_index: app_ref.handleMessageAction_withMessageIndex_(a, idx))
        return JSONResponse(content={"ok": True})

    @fastapi_app.get("/voice_prefs")
    async def voice_prefs():
        if get_voice_prefs is not None:
            prefs_dict = get_voice_prefs()
        else:
            prefs_dict = {
                "voice_insert": True,
                "voice_push_to_talk": False,
                "voice_prefer_on_device": False,
                "supports_on_device": False,
            }
        return JSONResponse(content=prefs_dict)

    @fastapi_app.post("/voice_toggle")
    async def voice_toggle(request: Request):
        try:
            body = await request.json()
        except Exception:
            body = {}
        send_on_release = body.get("sendOnRelease", False)
        app_ref._main_thread_queue.put(lambda s=send_on_release: app_ref.onVoiceToggle_(s))
        return JSONResponse(content={"ok": True})

    @fastapi_app.post("/voice_start")
    async def voice_start(request: Request):
        try:
            body = await request.json()
        except Exception:
            body = {}
        send_on_release = body.get("sendOnRelease", False)
        app_ref._main_thread_queue.put(lambda s=send_on_release: app_ref.onVoiceStart_(s))
        return JSONResponse(content={"ok": True})

    @fastapi_app.post("/voice_stop")
    async def voice_stop(request: Request):
        app_ref._main_thread_queue.put(lambda: app_ref.onVoiceStop_())
        return JSONResponse(content={"ok": True})

    @fastapi_app.get("/theme")
    async def get_theme_route():
        """Return current app theme so the chat UI can match (light/dark/system)."""
        theme = "system"
        if get_theme is not None:
            try:
                t = get_theme()
                if t in ("light", "dark", "system"):
                    theme = t
            except Exception:
                _log.debug("Failed to resolve theme", exc_info=True)
        return JSONResponse(content={"theme": theme})

    @fastapi_app.get("/post_onboarding_tip")
    async def get_post_onboarding_tip():
        """Return current post-onboarding tip if onboarding is done and tips not all dismissed."""
        try:
            from core import reload_config
            config = reload_config()
            onboarding = config.get("onboarding") or {}
            onboarding_done = onboarding.get("done", False)
            next_index = onboarding.get("post_onboarding_tips_next_index", 0)
            if not onboarding_done or next_index >= len(POST_ONBOARDING_TIPS):
                return JSONResponse(content={"show": False})
            return JSONResponse(content={
                "show": True,
                "index": next_index,
                "text": POST_ONBOARDING_TIPS[next_index],
            })
        except Exception:
            _log.debug("Failed to load post-onboarding tip state", exc_info=True)
            return JSONResponse(content={"show": False})

    @fastapi_app.post("/post_onboarding_tip_dismiss")
    async def post_onboarding_tip_dismiss(request: Request):
        """Mark current tip as dismissed; advance to next tip."""
        try:
            body = await request.json()
        except Exception:
            _log.debug("Failed to parse /post_onboarding_tip_dismiss JSON body", exc_info=True)
            body = {}
        index = body.get("index", 0)
        try:
            from core import reload_config, save_config
            config = reload_config()
            next_index = min(len(POST_ONBOARDING_TIPS), int(index) + 1)
            config.setdefault("onboarding", {})["post_onboarding_tips_next_index"] = next_index
            save_config(config)
        except Exception:
            _log.debug("Failed to persist post-onboarding tip dismissal", exc_info=True)
        return JSONResponse(content={"ok": True})

    # ── Live View: SSE stream, HTML page, and demo endpoint ──────────────
    if live_view_event_bus is not None:
        _lv_bus = live_view_event_bus

        @fastapi_app.get("/live_view/stream")
        async def live_view_stream(thread_id: str = Query("", alias="thread_id")):
            """SSE endpoint: streams Live View events for a given thread_id."""
            tid = (thread_id or "").strip()
            if not tid:
                return JSONResponse(content={"error": "thread_id required"}, status_code=400)

            async def _event_generator():
                try:
                    gen = _lv_bus.subscribe(tid)
                    # Wrap the async generator to interleave heartbeats every 15s
                    aiter = gen.__aiter__()
                    while True:
                        try:
                            event = await asyncio.wait_for(aiter.__anext__(), timeout=15.0)
                        except StopAsyncIteration:
                            return
                        except asyncio.TimeoutError:
                            yield ": heartbeat\n\n"
                            continue
                        yield f"data: {json.dumps(event, default=str)}\n\n"
                        if event.get("type") == "done":
                            return
                except asyncio.CancelledError:
                    return
                except Exception as exc:
                    err = {"type": "error", "message": str(exc)}
                    yield f"data: {json.dumps(err)}\n\n"

            return StreamingResponse(
                _event_generator(),
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )

        @fastapi_app.get("/live_view", response_class=HTMLResponse)
        async def live_view_page(thread_id: str = Query("", alias="thread_id")):
            """Serve the Live View HTML UI. thread_id is passed through to the JS."""
            from live_view_html import get_live_view_html
            plugin_js = ""
            pm = getattr(app_ref, "_plugin_manager", None)
            if pm is not None:
                try:
                    plugin_js = pm.get_live_view_js()
                except Exception:
                    _log.debug("Failed to load plugin Live View JS", exc_info=True)
            resp = HTMLResponse(content=get_live_view_html(plugin_js), media_type="text/html; charset=utf-8")
            if auth_token:
                resp.set_cookie("mugi_auth", auth_token, httponly=True, samesite="strict", path="/")
            return resp

        @fastapi_app.get("/live_view/history")
        async def live_view_history(thread_id: str = Query("", alias="thread_id")):
            """Return stored Live View events for a thread_id (for replaying history on switch)."""
            tid = (thread_id or "").strip()
            if not tid:
                return JSONResponse(content=[], status_code=200)
            events = _lv_bus.get_history(tid)
            health = _lv_bus.get_persist_health()
            if health.get("persist_decrypt_failed"):
                return JSONResponse(
                    content={
                        "events": events,
                        "health": health,
                        "warning": "Live View history could not be restored (encryption key may have changed).",
                    },
                    status_code=200,
                )
            return JSONResponse(content=events)

        @fastapi_app.get("/live_view/health")
        async def live_view_health():
            """Return Live View persistence health (decrypt failures after key rotation)."""
            return JSONResponse(content=_lv_bus.get_persist_health())

        @fastapi_app.post("/live_view/clear_history")
        async def live_view_clear_history(request: Request):
            """Clear stored Live View events for a thread_id."""
            try:
                body = await request.json()
            except Exception:
                _log.debug("Failed to parse /live_view/clear_history JSON body", exc_info=True)
                body = {}
            tid = (body.get("thread_id") or "").strip()
            if tid:
                _lv_bus.clear_history(tid)
            return JSONResponse(content={"ok": True})

        @fastapi_app.post("/live_view/delete_image")
        async def live_view_delete_image(request: Request):
            """Delete a Live View image by its content hash."""
            try:
                body = await request.json()
            except Exception:
                _log.debug("Failed to parse /live_view/delete_image JSON body", exc_info=True)
                body = {}
            img_hash = (body.get("hash") or "").strip()
            if not img_hash:
                return JSONResponse(content={"error": "hash required"}, status_code=400)
            _lv_bus.delete_image(img_hash)
            return JSONResponse(content={"ok": True})

        @fastapi_app.get("/live_view/image/{img_hash}")
        async def live_view_image(img_hash: str):
            """Serve a decrypted Live View preview image by its content hash."""
            b64 = _lv_bus.load_image(img_hash)
            if not b64:
                return JSONResponse(content={"error": "not found"}, status_code=404)
            try:
                raw = base64.b64decode(b64)
            except Exception:
                _log.debug("Failed to decode Live View image base64 payload", exc_info=True)
                return JSONResponse(content={"error": "decode error"}, status_code=500)
            content_type = "image/png" if raw[:4] == b"\x89PNG" else "image/jpeg"
            return Response(content=raw, media_type=content_type)

        @fastapi_app.post("/live_view/demo")
        async def live_view_demo(request: Request):
            """Start a fake multi-page OCR demo that emits events to the Live View bus."""
            try:
                body = await request.json()
            except Exception:
                body = {}
            tid = (body.get("thread_id") or "").strip()
            if not tid:
                return JSONResponse(content={"error": "thread_id required"}, status_code=400)
            from live_view_demo import start_demo_in_background
            start_demo_in_background(_lv_bus, tid)
            return JSONResponse(content={"ok": True, "thread_id": tid})

    if activity_bus is not None:
        _activity_bus = activity_bus

        @fastapi_app.get("/activity/stream")
        async def activity_stream():
            """SSE endpoint: stream global activity events."""
            sub_queue = _activity_bus.subscribe()

            async def _gen():
                try:
                    while True:
                        try:
                            evt = await asyncio.wait_for(sub_queue.get(), timeout=15.0)
                        except asyncio.TimeoutError:
                            yield ": heartbeat\n\n"
                            continue
                        yield f"data: {json.dumps(evt, default=str)}\n\n"
                except asyncio.CancelledError:
                    return
                finally:
                    _activity_bus.unsubscribe(sub_queue)

            return StreamingResponse(
                _gen(),
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )

        @fastapi_app.get("/activity/status")
        async def activity_status():
            """Snapshot of active tasks + recent history for reconnect hydration."""
            return JSONResponse(content=_activity_bus.snapshot())

        @fastapi_app.get("/activity/trust_summary")
        async def activity_trust_summary():
            """Allowlisted trust/observability summary for compact status card."""
            from tools.system import _load_app_prefs_tools
            from core import get_search_config

            summary: dict[str, Any] = {
                "schema": "mugi.trust_summary.v1",
                "generated_at_ms": int(time.time() * 1000),
            }
            try:
                app_prefs = _load_app_prefs_tools() or {}
            except Exception:
                app_prefs = {}
            summary["cli_tools_enabled"] = bool(app_prefs.get("include_cli_tools", False))
            summary["activity_audit_verbose"] = bool(app_prefs.get("activity_audit_verbose", False))
            summary["swift_bridge_configured"] = bool(os.environ.get("MUGI_CALENDAR_URL"))

            try:
                search_cfg = get_search_config() or {}
            except Exception:
                search_cfg = {}
            try:
                search_port = int(search_cfg.get("port", 0) or 0)
            except Exception:
                search_port = 0
            summary["search"] = {
                "use_local_docker": bool(search_cfg.get("use_local_docker", True)),
                "port": search_port,
                "public_fallback_configured": bool(search_cfg.get("public_fallback_url")),
            }

            try:
                mesh_mgr = getattr(delegate, "_mesh_manager", None)
                peer_count = 0
                if mesh_mgr is not None:
                    peers = getattr(mesh_mgr, "_peer_store", None)
                    if peers is not None and hasattr(peers, "all"):
                        peer_count = len(peers.all())
                summary["mesh"] = {
                    "mesh_listening": bool(mesh_mgr is not None),
                    "mesh_peer_count": int(peer_count),
                }
            except Exception:
                summary["mesh"] = {"mesh_listening": False, "mesh_peer_count": 0}

            try:
                from chat_storage import _get_connection
                conn = _get_connection()
                try:
                    row = conn.execute(
                        "SELECT COUNT(1) FROM scheduled_jobs WHERE status = 'active'"
                    ).fetchone()
                    summary["scheduled_jobs_active"] = int((row or [0])[0] or 0)
                    # Include a small projection of upcoming/last-run hints so
                    # the macOS Activity pane can show "Next: 2h" without needing
                    # a separate list endpoint. We redact the prompt body; the
                    # Swift side only needs id/schedule/next_run_at/last_run_at.
                    hint_rows = conn.execute(
                        "SELECT id, schedule, next_run_at, last_run_at, runs_completed, repeat_count "
                        "FROM scheduled_jobs WHERE status = 'active' "
                        "ORDER BY (next_run_at IS NULL OR next_run_at = ''), next_run_at ASC, created_at ASC "
                        "LIMIT 3"
                    ).fetchall()
                    summary["schedule_hints"] = [
                        {
                            "id": str(r[0] or ""),
                            "schedule": str(r[1] or ""),
                            "next_run_at": str(r[2] or ""),
                            "last_run_at": str(r[3] or ""),
                            "runs_completed": int(r[4] or 0),
                            "repeat_count": int(r[5] or 0),
                        }
                        for r in hint_rows
                    ]
                finally:
                    conn.close()
            except Exception:
                summary["scheduled_jobs_active"] = 0
                summary["schedule_hints"] = []

            snapshot = _activity_bus.snapshot(recent_limit=120)
            active = snapshot.get("active") if isinstance(snapshot, dict) else []
            active_rows = active if isinstance(active, list) else []
            active_run_ids = {
                str(row.get("run_id") or "").strip()
                for row in active_rows
                if isinstance(row, dict) and str(row.get("run_id") or "").strip()
            }
            active_thread_ids = {
                str(tid or "").strip()
                for tid in getattr(delegate, "_running_thread_ids", set())
                if str(tid or "").strip()
            }
            summary["activity"] = {
                "active_count": len(active_rows),
                "active_run_ids": len(active_run_ids),
                # Estimate active runs from both the activity bus and live running threads.
                "active_runs_estimate": max(len(active_run_ids), len(active_thread_ids)),
                "active_thread_ids": len(active_thread_ids),
            }
            summary["memory"] = _trust_summary_memory_snapshot()
            return JSONResponse(content=summary)

    @fastapi_app.get("/debug/background_gate")
    async def debug_background_gate():
        """Debug endpoint: report background-runner idle gate state."""
        try:
            from persistence import MAIN_ACTIVITY_STATE_PATH, MAIN_ACTIVITY_TS_PATH
            threshold_min = 4.0  # Must match mugi_runner.IDLE_THRESHOLD_MINUTES
            now = time.time()

            state_obj = None
            state_error = None
            if os.path.isfile(MAIN_ACTIVITY_STATE_PATH):
                try:
                    with open(MAIN_ACTIVITY_STATE_PATH, "r", encoding="utf-8") as f:
                        loaded = json.load(f)
                    if isinstance(loaded, dict):
                        state_obj = loaded
                except Exception as e:
                    state_error = str(e)

            ts_value = None
            ts_source = None
            if isinstance(state_obj, dict):
                try:
                    v = state_obj.get("ts")
                    if v is not None:
                        ts_value = float(v)
                        ts_source = "state"
                except Exception:
                    _log.debug("Failed to parse ts from main activity state", exc_info=True)
            if ts_value is None and os.path.isfile(MAIN_ACTIVITY_TS_PATH):
                try:
                    with open(MAIN_ACTIVITY_TS_PATH, "r", encoding="utf-8") as f:
                        raw = (f.read() or "").strip()
                    if raw:
                        ts_value = float(raw)
                        ts_source = "fallback_ts"
                except Exception:
                    _log.debug("Failed to parse fallback main activity timestamp", exc_info=True)

            busy = bool(state_obj.get("busy", False)) if isinstance(state_obj, dict) else None
            idle_min = max(0.0, (now - ts_value) / 60.0) if ts_value is not None else None

            if state_obj is None:
                can_run = False
                reason = "main activity state unavailable"
            elif busy:
                can_run = False
                reason = "main chat busy"
            elif idle_min is None:
                can_run = False
                reason = "missing timestamp"
            elif idle_min < threshold_min:
                can_run = False
                reason = f"main idle {idle_min:.1f} min < {threshold_min:.1f} min"
            else:
                can_run = True
                reason = f"main idle {idle_min:.1f} min"

            return JSONResponse(content={
                "ok": True,
                "now_ts": now,
                "idle_threshold_minutes": threshold_min,
                "state_path": MAIN_ACTIVITY_STATE_PATH,
                "state_exists": os.path.isfile(MAIN_ACTIVITY_STATE_PATH),
                "state_parse_error": state_error,
                "raw_state": state_obj,
                "fallback_ts_path": MAIN_ACTIVITY_TS_PATH,
                "fallback_ts_exists": os.path.isfile(MAIN_ACTIVITY_TS_PATH),
                "resolved_ts": ts_value,
                "resolved_ts_source": ts_source,
                "busy": busy,
                "idle_minutes": idle_min,
                "can_run_background": can_run,
                "decision_reason": reason,
            })
        except Exception as e:
            return JSONResponse(status_code=500, content={"ok": False, "error": str(e)})

    # Mount plugin HTTP sub-applications under /plugin/<slug>/
    # Plugin sub-apps are mounted as standalone FastAPI instances which do NOT inherit the
    # parent middleware. Auth is enforced via a FastAPI dependency on each sub-app.
    # Discovery may still be in progress; mugi_backend._build_app calls mount_plugin_http_routes
    # again after PluginManager.wait_initial_discover().
    mount_plugin_http_routes(fastapi_app, delegate, auth_token)

    @fastapi_app.get("/attachment_telemetry")
    async def attachment_telemetry():
        """Return attachment registry counters and current attachment count."""
        try:
            counters = _attach_reg.get_counters()
            from core.config import get_attachment_mode
            mode = get_attachment_mode()
        except Exception:
            counters = {}
            mode = "dual"
        return JSONResponse(content={"mode": mode, "counters": counters})

    # ---- Phase 2a: Inbound webhooks -------------------------------------
    _webhook_hits: dict[str, list[float]] = {}
    _WEBHOOK_RATE_WINDOW = 60.0

    def _webhook_rate_limited(ip: str, limit: int) -> bool:
        import time as _t
        bucket = _webhook_hits.setdefault(ip, [])
        now = _t.monotonic()
        bucket[:] = [t for t in bucket if now - t < _WEBHOOK_RATE_WINDOW]
        if len(bucket) >= limit:
            return True
        bucket.append(now)
        return False

    @fastapi_app.post("/webhook/{source}")
    async def webhook_endpoint(source: str, request: Request):
        from utils import webhooks as _wh
        try:
            from core.config import get_webhooks_config
            cfg = get_webhooks_config() or {}
        except Exception:
            cfg = {}
        rate_limit = int(cfg.get("rate_limit_per_min", 10) or 10)
        remote_ip = ""
        try:
            remote_ip = str(request.client.host or "")
        except Exception:
            pass
        if _webhook_rate_limited(remote_ip, rate_limit):
            return JSONResponse(status_code=429, content={"error": "rate_limited"})
        try:
            body_bytes = await request.body()
        except Exception:
            body_bytes = b""
        status, body = _wh.process_request(
            source,
            body_bytes,
            token_header=request.headers.get("x-mugi-webhook-token"),
            timestamp_header=request.headers.get("x-mugi-timestamp"),
            signature_header=request.headers.get("x-mugi-signature"),
        )
        return JSONResponse(status_code=status, content=body)

    # ---- Phase 2a: task_ledger read route -------------------------------
    @fastapi_app.get("/task_ledger/recent")
    async def task_ledger_recent(
        limit: int = 200,
        kind: str | None = None,
        status: str | None = None,
        since_ts: str | None = None,
    ):
        from utils import task_ledger as _tl
        rows = _tl.recent(limit=limit, kind=kind, status=status, since_ts=since_ts)
        return JSONResponse(content={"rows": rows})

    # ---- Structured task plan (right-panel rendering for the macOS client) ----
    @fastapi_app.get("/task_plan")
    async def get_task_plan_route(thread_id: str = ""):
        from utils.task_plan import is_valid_thread_id
        tid = (thread_id or "").strip()
        if not tid or not is_valid_thread_id(tid):
            return JSONResponse(status_code=400, content={"error": "invalid_thread_id"})
        try:
            from chat_storage import ensure_encryption_ready, get_task_plan_with_meta, probe_encryption_ready
            from utils.task_plan_reconcile import assess_plan_staleness, format_stale_hint

            if not probe_encryption_ready():
                return JSONResponse(
                    status_code=503,
                    content={"error": "encryption_not_ready", "thread_id": tid},
                )
            ensure_encryption_ready()
            plan, updated_at = get_task_plan_with_meta(tid)
            stale_hint: str = ""
            if plan:
                assessment = assess_plan_staleness(tid)
                stale_hint = format_stale_hint(assessment) if assessment else ""
        except Exception as exc:
            _log.debug("get_task_plan failed", exc_info=True)
            return JSONResponse(status_code=500, content={"error": str(exc)})
        payload: dict = {"thread_id": tid, "plan": plan, "updated_at": updated_at}
        if stale_hint:
            payload["stale_hint"] = stale_hint
        return JSONResponse(content=payload)

    @fastapi_app.delete("/task_plan")
    async def delete_task_plan_route(thread_id: str = ""):
        from utils.task_plan import is_valid_thread_id, notify_task_plan_changed
        tid = (thread_id or "").strip()
        if not tid or not is_valid_thread_id(tid):
            return JSONResponse(status_code=400, content={"error": "invalid_thread_id"})
        try:
            from chat_storage import ensure_encryption_ready, delete_task_plan
            ensure_encryption_ready()
            existed = delete_task_plan(tid)
        except Exception as exc:
            _log.debug("delete_task_plan failed", exc_info=True)
            return JSONResponse(status_code=500, content={"error": str(exc)})
        try:
            notify_task_plan_changed(tid, cleared=True)
        except Exception:
            pass
        return JSONResponse(content={"ok": True, "existed": bool(existed)})

    # ---- Kanban: durable, encrypted task board ---------------------------
    #
    # All routes are gated by the global auth middleware (above). Operator
    # verbs (block / unblock / set_assignee / dispatcher/nudge) are exposed
    # ONLY here and in mcp_server (never as agent tools). SSE events are
    # ciphertext-free by contract; clients fetch decrypted bodies through
    # /kanban/{id}.
    _kanban_nudge_lock = threading.Lock()
    _kanban_nudge_state = {"window_start": 0.0, "count": 0}
    _KANBAN_NUDGE_RATE = 30  # max nudges per minute

    def _rate_limit_kanban_nudge() -> bool:
        now = time.monotonic()
        with _kanban_nudge_lock:
            window_start = _kanban_nudge_state["window_start"]
            if now - window_start > 60.0:
                _kanban_nudge_state["window_start"] = now
                _kanban_nudge_state["count"] = 0
            if _kanban_nudge_state["count"] >= _KANBAN_NUDGE_RATE:
                return False
            _kanban_nudge_state["count"] += 1
            return True

    @fastapi_app.post("/kanban/tasks")
    async def kanban_create_task_route(request: Request):
        try:
            body = await request.json()
        except Exception:
            return JSONResponse(status_code=400, content={"error": "invalid_json"})
        if not isinstance(body, dict):
            return JSONResponse(status_code=400, content={"error": "expected_object"})
        parent_ids_raw = body.get("parent_ids")
        parent_ids: list[str] | None = None
        if isinstance(parent_ids_raw, list):
            parent_ids = [
                str(x).strip() for x in parent_ids_raw
                if isinstance(x, str) and x.strip()
            ] or None
        http_metadata = body.get("metadata")
        http_tool_names = body.get("tool_names")
        if isinstance(http_tool_names, list) and http_tool_names:
            http_metadata = dict(http_metadata) if isinstance(http_metadata, dict) else {}
            http_metadata["tool_names"] = http_tool_names
        try:
            import kanban_storage as _ks
            task = _ks.create_task(
                title=str(body.get("title") or "").strip(),
                body=str(body.get("body") or ""),
                assignee=str(body.get("assignee") or "general"),
                parent_id=(body.get("parent_id") or None),
                parent_ids=parent_ids,
                tenant=str(body.get("tenant") or "default"),
                priority=int(body.get("priority", 3)),
                tags=body.get("tags"),
                metadata=http_metadata,
                origin_thread_id=str(body.get("origin_thread_id") or ""),
                deadline_seconds=body.get("deadline_seconds"),
                idempotency_key=str(body.get("idempotency_key") or ""),
                model_override=str(body.get("model_override") or ""),
                max_retries_override=body.get("max_retries_override"),
                specifier_profile=str(body.get("specifier_profile") or ""),
            )
        except RuntimeError as exc:
            return JSONResponse(status_code=503, content={"error": str(exc)})
        except (ValueError, PermissionError) as exc:
            return JSONResponse(status_code=400, content={"error": str(exc)})
        except Exception as exc:
            _log.warning("kanban_create_task_route failed", exc_info=True)
            return JSONResponse(status_code=500, content={"error": str(exc)})
        dispatcher_nudged = True
        try:
            from utils.kanban_dispatcher import nudge_dispatcher as _nudge
            _nudge()
        except Exception:
            _log.warning("kanban_create_task_route: dispatcher nudge failed", exc_info=True)
            dispatcher_nudged = False
        out = dict(task)
        out["dispatcher_nudged"] = dispatcher_nudged
        return JSONResponse(content=out)

    @fastapi_app.post("/kanban/tasks/batch")
    async def kanban_create_batch_route(request: Request):
        try:
            body = await request.json()
        except Exception:
            return JSONResponse(status_code=400, content={"error": "invalid_json"})
        if not isinstance(body, dict):
            return JSONResponse(status_code=400, content={"error": "expected_object"})
        items = body.get("items")
        try:
            import kanban_storage as _ks
            rows = _ks.create_batch(
                items if isinstance(items, list) else [],
                tenant=str(body.get("tenant") or "default"),
                origin_thread_id=str(body.get("origin_thread_id") or ""),
            )
        except RuntimeError as exc:
            return JSONResponse(status_code=503, content={"error": str(exc)})
        except (ValueError, PermissionError) as exc:
            return JSONResponse(status_code=400, content={"error": str(exc)})
        except Exception as exc:
            _log.warning("kanban_create_batch_route failed", exc_info=True)
            return JSONResponse(status_code=500, content={"error": str(exc)})
        dispatcher_nudged = True
        try:
            from utils.kanban_dispatcher import nudge_dispatcher as _nudge
            _nudge()
        except Exception:
            _log.warning("kanban_create_batch_route: dispatcher nudge failed", exc_info=True)
            dispatcher_nudged = False
        return JSONResponse(content={"tasks": rows, "dispatcher_nudged": dispatcher_nudged})

    @fastapi_app.get("/kanban/tasks")
    async def kanban_list_tasks_route(
        status: str = "",
        assignee: str = "",
        tenant: str = "",
        parent_id: str = "",
        origin_thread_id: str = "",
        limit: int = 50,
    ):
        statuses: list[str] | None = None
        if status:
            statuses = [s.strip().lower() for s in status.split(",") if s.strip()]
        if not parent_id:
            parent_filter: str | None = None
        elif parent_id == "__root__":
            parent_filter = ""
        else:
            parent_filter = parent_id
        origin_filter: str | None = (origin_thread_id or "").strip() or None
        try:
            import kanban_storage as _ks
            rows = _ks.list_tasks(
                status=statuses,
                assignee=assignee or None,
                tenant=tenant or None,
                parent_id=parent_filter,
                origin_thread_id=origin_filter,
                limit=limit,
            )
        except Exception as exc:
            _log.debug("kanban_list_tasks_route failed", exc_info=True)
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(content={"tasks": rows})

    @fastapi_app.get("/kanban/tasks/search")
    async def kanban_search_tasks_route(
        q: str = "",
        tenant: str = "",
        limit: int = 20,
        semantic: bool = False,
    ):
        if not q.strip():
            return JSONResponse(status_code=400, content={"error": "query parameter 'q' is required"})
        try:
            import kanban_search as _ksearch
            results = _ksearch.search(q, tenant=tenant or None, limit=limit, semantic=semantic)
        except Exception as exc:
            _log.debug("kanban_search_tasks_route failed", exc_info=True)
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(content={"tasks": results})

    @fastapi_app.get("/kanban/tasks/{task_id}/similar")
    async def kanban_similar_tasks_route(
        task_id: str,
        tenant: str = "",
        limit: int = 5,
    ):
        try:
            import kanban_search as _ksearch
            results = _ksearch.find_similar(task_id, tenant=tenant or None, limit=limit)
        except Exception as exc:
            _log.debug("kanban_similar_tasks_route failed", exc_info=True)
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(content={"tasks": results})

    @fastapi_app.get("/kanban/tasks/{task_id}")
    async def kanban_get_task_route(task_id: str):
        try:
            import kanban_storage as _ks
            task = _ks.get_task(task_id)
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        if not task:
            return JSONResponse(status_code=404, content={"error": "not_found"})
        return JSONResponse(content=task)

    @fastapi_app.get("/kanban/tasks/{task_id}/runs")
    async def kanban_list_runs_route(task_id: str, limit: int = 50):
        try:
            import kanban_storage as _ks
            from utils.kanban_dispatcher import enrich_kanban_run_row

            runs = _ks.list_runs(task_id, limit=limit)
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(
            content={"runs": [enrich_kanban_run_row(r) for r in runs if isinstance(r, dict)]}
        )

    @fastapi_app.post("/kanban/tasks/{task_id}/cancel")
    async def kanban_cancel_route(task_id: str, request: Request):
        try:
            body = await request.json()
        except Exception:
            body = {}
        cascade = bool(body.get("cascade", True)) if isinstance(body, dict) else True
        try:
            import kanban_storage as _ks
            cancelled = _ks.cancel_task(task_id, cascade=cascade)
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(content={"cancelled": cancelled})

    @fastapi_app.post("/kanban/tasks/{task_id}/block")
    async def kanban_block_route(task_id: str, request: Request):
        try:
            body = await request.json()
        except Exception:
            body = {}
        reason = str((body or {}).get("reason") or "").strip()
        if not reason:
            return JSONResponse(status_code=400, content={"error": "reason_required"})
        try:
            import kanban_storage as _ks
            task = _ks.block_task(task_id, reason=reason)
        except (ValueError, PermissionError) as exc:
            return JSONResponse(status_code=400, content={"error": str(exc)})
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(content=task)

    @fastapi_app.post("/kanban/tasks/{task_id}/unblock")
    async def kanban_unblock_route(task_id: str):
        try:
            import kanban_storage as _ks
            task = _ks.unblock_task(task_id)
        except (ValueError, PermissionError) as exc:
            return JSONResponse(status_code=400, content={"error": str(exc)})
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(content=task)

    @fastapi_app.patch("/kanban/tasks/{task_id}")
    async def kanban_patch_route(task_id: str, request: Request):
        try:
            body = await request.json()
        except Exception:
            return JSONResponse(status_code=400, content={"error": "invalid_json"})
        if not isinstance(body, dict):
            return JSONResponse(status_code=400, content={"error": "expected_object"})
        import kanban_storage as _ks
        task = None
        force = bool(body.get("force", False))
        try:
            if "assignee" in body:
                new_assignee = str(body["assignee"]).strip()
                if not new_assignee:
                    return JSONResponse(status_code=400, content={"error": "assignee must be non-empty"})
                task = _ks.set_assignee(task_id, new_assignee, force=force)
            if "status" in body:
                reason = str(body.get("reason") or "").strip()
                task = _ks.set_status(task_id, str(body["status"]), reason=reason)
            if "priority" in body:
                task = _ks.set_priority(task_id, int(body["priority"]))
            if "parent_ids" in body:
                raw = body["parent_ids"]
                pids = [str(p).strip() for p in raw if isinstance(p, str) and p.strip()] if isinstance(raw, list) else []
                task = _ks.set_parent_ids(task_id, pids)
        except (ValueError, PermissionError) as exc:
            return JSONResponse(status_code=400, content={"error": str(exc)})
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        if task is None:
            return JSONResponse(status_code=400, content={"error": "no patchable fields (assignee, priority, parent_ids)"})
        try:
            from utils.kanban_dispatcher import nudge_dispatcher as _nudge
            _nudge()
        except Exception:
            pass
        return JSONResponse(content=task)

    @fastapi_app.delete("/kanban/tasks/{task_id}")
    async def kanban_delete_task_route(task_id: str):
        try:
            import kanban_storage as _ks
            _ks.delete_task(task_id)
        except ValueError as exc:
            return JSONResponse(status_code=400, content={"error": str(exc)})
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(content={"deleted": True})

    @fastapi_app.post("/kanban/tasks/clear")
    async def kanban_clear_tasks_route(request: Request):
        try:
            body = await request.json()
        except Exception:
            return JSONResponse(status_code=400, content={"error": "invalid_json"})
        if not isinstance(body, dict):
            return JSONResponse(status_code=400, content={"error": "expected_object"})
        status = str(body.get("status") or "").strip()
        if status not in ("done", "cancelled"):
            return JSONResponse(
                status_code=400,
                content={"error": "status must be 'done' or 'cancelled'"},
            )
        tenant = str(body.get("tenant") or "").strip()
        assignee = str(body.get("assignee") or "").strip()
        try:
            import kanban_storage as _ks
            count = _ks.clear_terminal_tasks(status, tenant=tenant, assignee=assignee)
        except ValueError as exc:
            return JSONResponse(status_code=400, content={"error": str(exc)})
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(content={"deleted_count": count})

    @fastapi_app.post("/kanban/dispatcher/nudge")
    async def kanban_dispatcher_nudge_route():
        if not _rate_limit_kanban_nudge():
            return JSONResponse(status_code=429, content={"error": "rate_limited"})
        try:
            from utils.kanban_dispatcher import nudge_dispatcher as _nudge
            _nudge()
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(content={"ok": True})

    @fastapi_app.get("/kanban/dispatcher/health")
    async def kanban_dispatcher_health_route():
        try:
            from utils.kanban_dispatcher import dispatcher_health as _h
            return JSONResponse(content=_h())
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})

    @fastapi_app.post("/kanban/tasks/rebalance")
    async def kanban_rebalance_route(request: Request):
        try:
            data = await request.json()
        except Exception:
            return JSONResponse(status_code=400, content={"error": "invalid JSON"})
        status = data.get("status", "")
        if not status:
            return JSONResponse(status_code=400, content={"error": "status required"})
        try:
            import kanban_storage as _ks
            count = _ks.rebalance_lane_priorities(
                status,
                assignee=data.get("assignee", ""),
                tenant=data.get("tenant", ""),
            )
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(content={"updated": count})

    @fastapi_app.post("/kanban/tasks/{task_id}/subscribe")
    async def kanban_task_subscribe_route(task_id: str, request: Request):
        """Subscribe to terminal lifecycle events on a task.

        Body: ``{"channel": "thread|notification|webhook|discord|slack",
        "target": "...", "fire_on": ["gave_up", "done"]}``.
        Returns the new subscription id.
        """
        try:
            body = await request.json()
        except Exception:
            body = {}
        channel = str((body or {}).get("channel") or "notification").strip().lower()
        target = str((body or {}).get("target") or "").strip()
        raw_fire = (body or {}).get("fire_on")
        fire_on: list[str]
        if isinstance(raw_fire, list):
            fire_on = [str(x).strip().lower() for x in raw_fire if isinstance(x, (str, bytes))]
        elif isinstance(raw_fire, str):
            fire_on = [s.strip().lower() for s in raw_fire.split(",") if s.strip()]
        else:
            fire_on = ["gave_up", "done"]
        try:
            from utils.kanban_notifications import kanban_notify_subscribe as _sub
            sub_id = _sub(task_id, channel=channel, target=target, fire_on=fire_on)
        except ValueError as exc:
            return JSONResponse(status_code=400, content={"error": str(exc)})
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        return JSONResponse(status_code=201, content={"subscription_id": sub_id})

    @fastapi_app.delete("/kanban/subscriptions/{subscription_id}")
    async def kanban_subscription_delete_route(subscription_id: str):
        try:
            from utils.kanban_notifications import kanban_notify_unsubscribe as _unsub
            ok = _unsub(subscription_id)
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        if not ok:
            return JSONResponse(status_code=404, content={"error": "not_found"})
        return JSONResponse(content={"ok": True})

    @fastapi_app.get("/kanban/tasks/{task_id}/subscriptions")
    async def kanban_task_subscriptions_route(task_id: str):
        try:
            from utils.kanban_notifications import list_subscriptions_for_task as _ls
            subs = _ls(task_id)
        except Exception as exc:
            return JSONResponse(status_code=500, content={"error": str(exc)})
        for s in subs:
            s.pop("target", None)
        return JSONResponse(content={"subscriptions": subs})

    @fastapi_app.get("/kanban/stream")
    async def kanban_stream_route():
        """SSE: ciphertext-free kanban lifecycle events. Heartbeats every 15s."""
        from utils.kanban_events import subscribe as _subscribe, unsubscribe as _unsubscribe
        queue = _subscribe()

        async def _gen():
            try:
                while True:
                    try:
                        evt = await asyncio.wait_for(queue.get(), timeout=15.0)
                    except asyncio.TimeoutError:
                        yield ": heartbeat\n\n"
                        continue
                    yield f"data: {json.dumps(evt, default=str)}\n\n"
            except asyncio.CancelledError:
                return
            finally:
                _unsubscribe(queue)

        return StreamingResponse(
            _gen(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
        )

    # ---- Phase 2a: webhook source CRUD + rotation -----------------------
    @fastapi_app.post("/preferences/webhooks/sources")
    async def create_webhook_source(request: Request):
        import json as _json
        try:
            body = await request.json()
        except Exception:
            body = {}
        source = str((body or {}).get("source") or "").strip()
        capability = str((body or {}).get("capability") or "notify_only").strip().lower()
        if capability not in ("notify_only", "full_turn"):
            return JSONResponse(status_code=400, content={"error": "invalid_capability"})
        from utils.webhooks import validate_source_name as _vs
        if not _vs(source):
            return JSONResponse(status_code=400, content={"error": "invalid_source_name"})
        from persistence import (
            keychain_get_webhook_hmac,
            keychain_set_webhook_hmac,
            generate_secure_token,
        )
        existing = keychain_get_webhook_hmac(source)
        if existing:
            return JSONResponse(status_code=409, content={"error": "source_exists"})
        secret = generate_secure_token(32)
        keychain_set_webhook_hmac(source, secret)
        try:
            from persistence import webhook_sources_add as _wsa
            _wsa(source, capability=capability)
        except Exception:
            _log.error("webhook_sources_add failed for %s; rolling back keychain entry", source, exc_info=True)
            try:
                from persistence import keychain_delete_webhook_hmac
                keychain_delete_webhook_hmac(source)
            except Exception:
                _log.debug("Rollback: keychain_delete_webhook_hmac failed", exc_info=True)
            return JSONResponse(status_code=500, content={"error": "failed_to_register_source"})
        return JSONResponse(status_code=201, content={
            "source": source,
            "capability": capability,
            "hmac_secret": secret,
        })

    @fastapi_app.post("/preferences/webhooks/sources/{source}/capability")
    async def set_webhook_source_capability(source: str, request: Request):
        try:
            body = await request.json()
        except Exception:
            body = {}
        capability = str((body or {}).get("capability") or "").strip().lower()
        from utils.webhooks import validate_source_name as _vs
        if not _vs(source):
            return JSONResponse(status_code=400, content={"error": "invalid_source_name"})
        if capability not in ("notify_only", "full_turn"):
            return JSONResponse(status_code=400, content={"error": "invalid_capability"})
        from persistence import webhook_sources_set_capability
        ok = webhook_sources_set_capability(source, capability)
        if not ok:
            return JSONResponse(status_code=404, content={"error": "not_found"})
        return JSONResponse(content={"source": source, "capability": capability})

    @fastapi_app.delete("/preferences/webhooks/sources/{source}")
    async def delete_webhook_source(source: str):
        from utils.webhooks import validate_source_name as _vs
        if not _vs(source):
            return JSONResponse(status_code=400, content={"error": "invalid_source_name"})
        from persistence import keychain_delete_webhook_hmac
        keychain_delete_webhook_hmac(source)
        try:
            from persistence import webhook_sources_remove as _wsr
            _wsr(source)
        except Exception:
            _log.warning("webhook_sources_remove failed for %s; keychain already cleared", source, exc_info=True)
        return JSONResponse(content={"ok": True, "source": source})

    @fastapi_app.post("/preferences/webhooks/sources/{source}/rotate")
    async def rotate_webhook_source(source: str):
        from utils.webhooks import validate_source_name as _vs
        if not _vs(source):
            return JSONResponse(status_code=400, content={"error": "invalid_source_name"})
        from persistence import keychain_set_webhook_hmac, generate_secure_token
        secret = generate_secure_token(32)
        keychain_set_webhook_hmac(source, secret)
        return JSONResponse(content={"source": source, "hmac_secret": secret})

    @fastapi_app.post("/preferences/webhooks/rotate_primary")
    async def rotate_webhook_primary():
        from persistence import keychain_set_webhook_token, generate_secure_token
        tok = generate_secure_token(32)
        keychain_set_webhook_token(tok)
        return JSONResponse(content={"primary_token": tok})

    # ---- Phase 2b: outbound webhook CRUD + rotation ---------------------
    _OUTBOUND_SLUG_RE = __import__("re").compile(r"^[a-zA-Z0-9_-]{1,32}$")

    @fastapi_app.get("/preferences/outbound_webhooks")
    async def list_outbound_webhooks():
        from persistence import outbound_webhooks_list, keychain_get_outbound_webhook_hmac
        rows = []
        for row in outbound_webhooks_list():
            slug = str(row.get("slug") or "")
            rows.append({
                **row,
                "has_secret": bool(keychain_get_outbound_webhook_hmac(slug)),
            })
        return JSONResponse(content={"outbound_webhooks": rows})

    @fastapi_app.post("/preferences/outbound_webhooks")
    async def create_outbound_webhook(request: Request):
        try:
            body = await request.json()
        except Exception:
            body = {}
        slug = str((body or {}).get("slug") or "").strip()
        url = str((body or {}).get("url") or "").strip()
        label = str((body or {}).get("label") or "").strip()
        if not _OUTBOUND_SLUG_RE.match(slug):
            return JSONResponse(status_code=400, content={"error": "invalid_slug"})
        from persistence import (
            outbound_webhooks_add,
            keychain_set_outbound_webhook_hmac,
            generate_secure_token,
        )
        try:
            row = outbound_webhooks_add(slug, url, label=label)
        except ValueError as e:
            msg = str(e)
            status = 409 if "already registered" in msg else 400
            return JSONResponse(status_code=status, content={"error": msg})
        secret = generate_secure_token(32)
        keychain_set_outbound_webhook_hmac(slug, secret)
        return JSONResponse(status_code=201, content={**row, "hmac_secret": secret})

    @fastapi_app.delete("/preferences/outbound_webhooks/{slug}")
    async def delete_outbound_webhook(slug: str):
        if not _OUTBOUND_SLUG_RE.match(slug or ""):
            return JSONResponse(status_code=400, content={"error": "invalid_slug"})
        from persistence import outbound_webhooks_remove
        removed = outbound_webhooks_remove(slug)
        return JSONResponse(content={"ok": removed, "slug": slug})

    @fastapi_app.post("/preferences/outbound_webhooks/{slug}/rotate")
    async def rotate_outbound_webhook(slug: str):
        if not _OUTBOUND_SLUG_RE.match(slug or ""):
            return JSONResponse(status_code=400, content={"error": "invalid_slug"})
        from persistence import (
            outbound_webhooks_get,
            keychain_set_outbound_webhook_hmac,
            generate_secure_token,
        )
        if outbound_webhooks_get(slug) is None:
            return JSONResponse(status_code=404, content={"error": "not_found"})
        secret = generate_secure_token(32)
        keychain_set_outbound_webhook_hmac(slug, secret)
        return JSONResponse(content={"slug": slug, "hmac_secret": secret})

    @fastapi_app.get("/preferences/webhooks/status")
    async def webhook_status():
        from utils.webhooks import is_enabled as _en
        from persistence import keychain_get_webhook_token, webhook_sources_list
        enabled = _en()
        has_token = bool(keychain_get_webhook_token())
        return JSONResponse(content={
            "enabled": enabled,
            "has_primary_token": has_token,
            "sources": webhook_sources_list(),
        })

    # ---- Phase 2b: ledger reflection status + manual run ---------------
    @fastapi_app.get("/background/reflection/status")
    async def reflection_status():
        from utils import ledger_reflection as _lr
        try:
            state = _lr.read_last_reflection()
            return JSONResponse(content={
                "enabled": _lr.is_enabled(),
                "last_run": state,
            })
        except Exception as e:
            return JSONResponse(status_code=500, content={"error": str(e)})

    @fastapi_app.post("/background/reflection/run_now")
    async def reflection_run_now():
        """Force a single ledger reflection pass regardless of schedule.

        Used by the SwiftUI "Run now" button. Returns the summary dict
        from ``utils.ledger_reflection.run_reflection(force=True)``.
        """
        from utils import ledger_reflection as _lr
        try:
            summary = _lr.run_reflection(force=True)
            return JSONResponse(content=summary)
        except Exception as e:
            return JSONResponse(status_code=500, content={"error": str(e)})

    return fastapi_app
