"""Flight Deck backend — Docker container & process management for Captain Claw agents."""

from __future__ import annotations

import os
import sys
import json
import secrets
import signal
import asyncio
import subprocess
import shutil
from pathlib import Path
from contextlib import asynccontextmanager
from typing import Any

import docker
import yaml
from fastapi import Depends, FastAPI, File, HTTPException, Request, UploadFile, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

# Load a project-local ``.env`` into ``os.environ`` early — before any
# captain_claw submodule imports — so env-driven bridges (Messenger,
# WhatsApp) and threshold-tunable subsystems (face_index) see values
# placed in a CWD-relative ``.env`` without needing the user to source
# the file from the shell or wire systemd ``EnvironmentFile=`` entries.
#
# python-dotenv is already a core dep. ``load_dotenv()`` with no args
# walks up from the CWD looking for ``.env`` and is a no-op if not found.
# ``override=False`` (the default) preserves any value already set in
# the real environment — shell exports win over the file, which is the
# behaviour everyone expects.
from dotenv import load_dotenv
load_dotenv()

import logging

from captain_claw.flight_deck.auth import get_current_user, get_optional_user, get_ws_user, set_auth_db
from captain_claw.flight_deck.db import FlightDeckDB


# ── Console logging: timestamps + ANSI colors ──────────────────────────
_ANSI = {
    "reset":   "\033[0m",
    "dim":     "\033[2m",
    "bold":    "\033[1m",
    "grey":    "\033[90m",
    "red":     "\033[31m",
    "green":   "\033[32m",
    "yellow":  "\033[33m",
    "blue":    "\033[34m",
    "magenta": "\033[35m",
    "cyan":    "\033[36m",
    "white":   "\033[37m",
    "br_red":  "\033[91m",
    "br_green":"\033[92m",
    "br_yellow":"\033[93m",
    "br_blue": "\033[94m",
    "br_cyan": "\033[96m",
}

_LEVEL_COLORS = {
    "DEBUG":    _ANSI["grey"],
    "INFO":     _ANSI["br_green"],
    "WARNING":  _ANSI["br_yellow"],
    "ERROR":    _ANSI["br_red"],
    "CRITICAL": _ANSI["bold"] + _ANSI["br_red"],
}


class _FDColorFormatter(logging.Formatter):
    """Date/time + colored level + dim logger name + message."""

    def __init__(self, use_color: bool):
        super().__init__()
        self.use_color = use_color

    def format(self, record: logging.LogRecord) -> str:
        ts = self.formatTime(record, "%Y-%m-%d %H:%M:%S")
        level = record.levelname
        name = record.name
        msg = record.getMessage()
        if record.exc_info:
            msg = msg + "\n" + self.formatException(record.exc_info)
        if self.use_color:
            lvl_col = _LEVEL_COLORS.get(level, "")
            R = _ANSI["reset"]
            return (
                f"{_ANSI['grey']}[{ts}]{R} "
                f"{lvl_col}{level:<8}{R} "
                f"{_ANSI['cyan']}{name}{R}  "
                f"{msg}"
            )
        return f"[{ts}] {level:<8} {name}  {msg}"


class _SuppressShutdownCancelFilter(logging.Filter):
    """Drop uvicorn's noisy "Exception in ASGI application" tracebacks
    whose root cause is a ``CancelledError`` raised because uvicorn was
    shutting down. Each open browser tab (SSE / long-poll / streaming
    response) produces one such traceback at Ctrl+C — they are expected
    and harmless, and they crowd out the real exit messages.

    Genuine ASGI exceptions (real bugs) are kept: they have a different
    root cause and pass through untouched.
    """

    def filter(self, record: logging.LogRecord) -> bool:
        if record.levelno < logging.ERROR:
            return True
        if "Exception in ASGI application" not in record.getMessage():
            return True
        exc = record.exc_info[1] if record.exc_info else None
        # Walk the exception chain to the root cause.
        seen: set[int] = set()
        while exc is not None and id(exc) not in seen:
            seen.add(id(exc))
            if isinstance(exc, asyncio.CancelledError):
                return False  # drop the record
            exc = exc.__cause__ or exc.__context__
        return True


def _configure_fd_logging() -> None:
    """Install our colored handler on the root logger and silence dupes."""
    use_color = sys.stderr.isatty() and os.environ.get("NO_COLOR", "") == ""
    handler = logging.StreamHandler(stream=sys.stderr)
    handler.setFormatter(_FDColorFormatter(use_color=use_color))
    root = logging.getLogger()
    # Replace any pre-existing handlers so we don't double-print.
    for h in list(root.handlers):
        root.removeHandler(h)
    root.addHandler(handler)
    root.setLevel(logging.INFO)
    # Make sure these loggers don't add their own handlers on top of ours.
    for name in ("uvicorn", "uvicorn.error", "uvicorn.access", "fastapi", "flight_deck"):
        lg = logging.getLogger(name)
        lg.handlers.clear()
        lg.propagate = True
    # Quiet down access log a touch — INFO level keeps it visible.
    logging.getLogger("uvicorn.access").setLevel(logging.INFO)
    # Suppress the per-stream CancelledError tracebacks during shutdown.
    logging.getLogger("uvicorn.error").addFilter(_SuppressShutdownCancelFilter())


_configure_fd_logging()

class _KwargLoggerAdapter:
    """Thin wrapper around stdlib Logger that accepts structured kwargs.

    Multiple call sites in this file use a structlog-like API
    (`log.info("event", key=val, ...)`). The stdlib logger raises
    ``TypeError: Logger._log() got an unexpected keyword argument ...`` for
    those, which historically masked bugs (e.g. /announce-port returning 500
    on every call, leaving the registry stale). This adapter converts the
    kwargs into ``key=value key2=value2`` and appends them to the message
    so the existing call sites Just Work without churn.

    Reserved kwargs that the stdlib logger understands (``exc_info``,
    ``stack_info``, ``stacklevel``, ``extra``) are passed through unchanged.
    """

    _RESERVED = {"exc_info", "stack_info", "stacklevel", "extra"}

    def __init__(self, logger: logging.Logger):
        self._logger = logger

    def _emit(self, level: int, msg, args, kwargs):
        passthrough = {k: kwargs.pop(k) for k in list(kwargs) if k in self._RESERVED}
        if kwargs:
            extras = " ".join(f"{k}={v}" for k, v in kwargs.items())
            if isinstance(msg, str) and args:
                # Preserve old %-style formatting if used
                try:
                    msg = msg % args
                    args = ()
                except Exception:
                    pass
            msg = f"{msg}  {extras}" if msg else extras
        self._logger.log(level, msg, *args, **passthrough)

    def debug(self, msg=None, *args, **kwargs): self._emit(logging.DEBUG, msg, args, kwargs)
    def info(self, msg=None, *args, **kwargs): self._emit(logging.INFO, msg, args, kwargs)
    def warning(self, msg=None, *args, **kwargs): self._emit(logging.WARNING, msg, args, kwargs)
    def error(self, msg=None, *args, **kwargs): self._emit(logging.ERROR, msg, args, kwargs)
    def critical(self, msg=None, *args, **kwargs): self._emit(logging.CRITICAL, msg, args, kwargs)
    # Alias used by some libs
    warn = warning

    def __getattr__(self, name):
        # Fall through to the underlying logger for anything we don't override
        return getattr(self._logger, name)


log = _KwargLoggerAdapter(logging.getLogger("flight_deck"))
from captain_claw.flight_deck.rate_limiter import (
    check_api_rate_limit, check_spawn_rate_limit, check_agent_count_limit,
    load_plan_limits_from_db_sync,
)

def _resolve_static_dir() -> Path:
    """Resolve static dir, handling PyInstaller bundles where __file__ is at _internal/ root."""
    normal = Path(__file__).parent / "static"
    if normal.is_dir():
        return normal
    # PyInstaller: entry script lands in _internal/, data is in _internal/captain_claw/flight_deck/static/
    if getattr(sys, "_MEIPASS", None):
        bundled = Path(sys._MEIPASS) / "captain_claw" / "flight_deck" / "static"
        if bundled.is_dir():
            return bundled
    return normal

STATIC_DIR = _resolve_static_dir()

# ── Config ──

def _default_data_dir() -> str:
    """Use ~/.captain-claw/fd-data for standalone builds, ./fd-data otherwise."""
    if getattr(sys, "_MEIPASS", None):
        return str(Path.home() / ".captain-claw" / "fd-data")
    return "./fd-data"

DATA_DIR = Path(os.environ.get("FD_DATA_DIR", _default_data_dir())).resolve()
CONTAINER_LABEL = "flight-deck.managed"
OWNER_LABEL = "flight-deck.owner"
CC_IMAGE_DEFAULT = "kstevica/captain-claw:latest"
AUTH_ENABLED = os.environ.get("FD_AUTH_ENABLED", "true").lower() in ("true", "1", "yes")


# ── Docker client ──

_client: docker.DockerClient | None = None


def get_docker() -> docker.DockerClient:
    global _client
    if _client is None:
        _client = docker.from_env()
    return _client


# ── Process registry ──

PROCESS_REGISTRY_FILE = DATA_DIR / ".processes.json"
_processes: dict[str, subprocess.Popen] = {}  # slug -> Popen

# Serialise process spawns so concurrent requests can't both pick the same
# port between the availability check and Popen. Lazily constructed in the
# running event loop (asyncio.Lock() at import time would bind to whichever
# loop happens to exist then).
_spawn_lock: asyncio.Lock | None = None

def _get_spawn_lock() -> asyncio.Lock:
    global _spawn_lock
    if _spawn_lock is None:
        _spawn_lock = asyncio.Lock()
    return _spawn_lock


class ProcessEntry(BaseModel):
    """Persisted metadata for a managed process agent."""
    slug: str
    name: str
    description: str = ""
    web_port: int
    web_auth: str = ""
    pid: int | None = None
    provider: str = ""
    model: str = ""


def _load_process_registry() -> dict[str, dict]:
    """Load process registry from disk.

    Uses an advisory shared flock so we don't observe a partially-written file
    while another coroutine / process is in the middle of `_save_process_registry`.
    Falls back to {} only if the file is genuinely missing or unparseable —
    NEVER on transient lock contention.
    """
    if not PROCESS_REGISTRY_FILE.is_file():
        return {}
    try:
        import fcntl as _fcntl
        with PROCESS_REGISTRY_FILE.open("r") as _f:
            try:
                _fcntl.flock(_f.fileno(), _fcntl.LOCK_SH)
            except OSError:
                pass  # flock unsupported on some FSes (NFS, etc.) — best effort
            data = _f.read()
        return json.loads(data) if data else {}
    except (json.JSONDecodeError, OSError) as exc:
        log.warning("process registry read failed", error=str(exc))
        return {}


def _save_process_registry(registry: dict[str, dict]):
    """Persist process registry to disk atomically.

    Writes to a sibling .tmp file then os.replace()s — a partial write can
    never leave the canonical file in a half-written state, and concurrent
    readers will always see either the previous or the new full snapshot.
    Also takes an exclusive flock around the rename for cross-process safety.
    """
    PROCESS_REGISTRY_FILE.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(registry, indent=2)
    tmp_path = PROCESS_REGISTRY_FILE.with_suffix(PROCESS_REGISTRY_FILE.suffix + ".tmp")
    try:
        import fcntl as _fcntl
        with tmp_path.open("w") as _f:
            try:
                _fcntl.flock(_f.fileno(), _fcntl.LOCK_EX)
            except OSError:
                pass
            _f.write(payload)
            _f.flush()
            try:
                os.fsync(_f.fileno())
            except OSError:
                pass
        os.replace(str(tmp_path), str(PROCESS_REGISTRY_FILE))
    except OSError as exc:
        log.error("process registry write failed", error=str(exc))
        raise


def _process_is_alive(slug: str) -> bool:
    """Check if a managed process is still running."""
    proc = _processes.get(slug)
    if proc and proc.poll() is None:
        return True
    # Also check by PID from registry
    registry = _load_process_registry()
    entry = registry.get(slug)
    if entry and entry.get("pid"):
        try:
            os.kill(entry["pid"], 0)
            return True
        except (OSError, ProcessLookupError):
            pass
    return False


def _kill_pid(pid: int, timeout: float = 5.0):
    """Send SIGTERM to a PID and wait for it to die; SIGKILL if needed."""
    import time
    try:
        os.kill(pid, signal.SIGTERM)
    except (OSError, ProcessLookupError):
        return
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            os.kill(pid, 0)
            time.sleep(0.3)
        except (OSError, ProcessLookupError):
            return
    # Still alive — force kill
    try:
        os.kill(pid, signal.SIGKILL)
    except (OSError, ProcessLookupError):
        pass


def _stop_all_processes():
    """Stop all managed process agents in parallel (called on FD shutdown)."""
    import threading
    registry = _load_process_registry()
    pids_to_kill: list[int] = []
    for slug, entry in registry.items():
        pid = entry.get("pid")
        if pid and _process_is_alive(slug):
            pids_to_kill.append(pid)
            entry["pid"] = None

    if pids_to_kill:
        # Send SIGTERM to all at once
        for pid in pids_to_kill:
            try:
                os.kill(pid, signal.SIGTERM)
            except (OSError, ProcessLookupError):
                pass

        # Wait for all in parallel threads
        def _wait_and_kill(pid: int, timeout: float = 5.0):
            import time
            deadline = time.monotonic() + timeout
            while time.monotonic() < deadline:
                try:
                    os.kill(pid, 0)
                    time.sleep(0.2)
                except (OSError, ProcessLookupError):
                    return
                try:
                    os.kill(pid, signal.SIGKILL)
                except (OSError, ProcessLookupError):
                    pass

        threads = [threading.Thread(target=_wait_and_kill, args=(pid,)) for pid in pids_to_kill]
        for t in threads:
            t.start()
        for t in threads:
            t.join(timeout=6)

    _save_process_registry(registry)
    _processes.clear()


def _resolve_cc_web_bin() -> str:
    """Resolve the ``captain-claw-web`` executable.

    In a PyInstaller standalone build the binary lives next to this process's
    executable (``sys._MEIPASS``'s parent) and is NOT on PATH, so a bare
    ``"captain-claw-web"`` lookup fails with FileNotFoundError. Resolve the
    bundled path when frozen; fall back to PATH for pip/dev installs.
    """
    if getattr(sys, "_MEIPASS", None):
        bundled = Path(sys._MEIPASS).parent / "captain-claw-web"
        if bundled.exists():
            return str(bundled)
    return "captain-claw-web"


def _start_registered_process(slug: str, entry: dict) -> bool:
    """Start a single process agent from its registry entry. Returns True on success."""
    agent_dir = DATA_DIR / slug
    if not agent_dir.is_dir():
        return False

    web_port = entry.get("web_port", 24080)

    # Rebuild environment from .env file
    environment = dict(os.environ)
    env_file = agent_dir / ".env"
    if env_file.is_file():
        content = env_file.read_text().strip()
        if content:
            for line in content.split("\n"):
                if "=" in line:
                    k, v = line.split("=", 1)
                    environment[k] = v

    # Share secrets from Flight Deck's own env with the agent when the agent
    # didn't set them (or set them blank). Lets one place hold the keys — e.g.
    # SONIOX for video/audio transcription, WhatsApp Cloud API creds for the
    # agent-side send path — instead of duplicating into every agent's .env.
    for _shared in (
        "SONIOX_API_KEY",
        "WHATSAPP_ACCESS_TOKEN", "WHATSAPP_PHONE_NUMBER_ID", "WHATSAPP_ALLOWED_WAIDS",
    ):
        if not str(environment.get(_shared, "")).strip():
            _fd_val = str(os.environ.get(_shared, "")).strip()
            if _fd_val:
                environment[_shared] = _fd_val

    environment["HOME"] = str(agent_dir / "data" / "home-config-parent")
    # Slug + URL for port-fallback callbacks. Without FD_URL the agent can't
    # announce a drifted port back to Flight Deck and the registry goes stale
    # (chat panel then 401s because FD proxies to the old port).
    environment["FD_AGENT_SLUG"] = slug
    if "FD_URL" not in environment:
        fd_port = os.environ.get("FD_PORT", "25080")
        environment["FD_URL"] = f"http://localhost:{fd_port}"

    log_file = agent_dir / "process.log"
    try:
        log_fh = open(log_file, "a")
        proc = subprocess.Popen(
            [_resolve_cc_web_bin(), "--port", str(web_port)],
            cwd=str(agent_dir),
            env=environment,
            stdout=log_fh,
            stderr=subprocess.STDOUT,
            start_new_session=True,
        )
        _processes[slug] = proc
        entry["pid"] = proc.pid
        return True
    except Exception as exc:
        log.error("Failed to start process agent", slug=slug, error=str(exc))
        return False


def _reattach_processes():
    """On startup, check registered processes and restart any that were running."""
    import time as _time

    registry = _load_process_registry()
    restarted = []
    skipped = []
    stagger_s = float(os.environ.get("FD_REATTACH_STAGGER_S", "0.3"))
    first_launch = True
    for slug, entry in registry.items():
        # Skip agents that were intentionally stopped by the user
        if entry.get("stopped"):
            skipped.append(slug)
            continue
        pid = entry.get("pid")
        if pid:
            try:
                os.kill(pid, 0)  # Still alive — just track it
                continue
            except (OSError, ProcessLookupError):
                pass
        # Process was registered but is dead — restart it. Stagger the
        # launches so concurrent port probes don't race and pick the same
        # fallback port.
        if entry.get("web_port"):
            if not first_launch and stagger_s > 0:
                _time.sleep(stagger_s)
            first_launch = False
            if _start_registered_process(slug, entry):
                restarted.append(slug)
            else:
                entry["pid"] = None
    _save_process_registry(registry)
    if restarted:
        print(f"Flight Deck: restarted {len(restarted)} process agent(s): {', '.join(restarted)}")
    if skipped:
        print(f"Flight Deck: skipped {len(skipped)} stopped process agent(s): {', '.join(skipped)}")


# ── App ──

@asynccontextmanager
async def lifespan(app: FastAPI):
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    # Initialize game registry with FD data dir
    from captain_claw.games.registry import set_games_dir
    games_dir = DATA_DIR / "games"
    games_dir.mkdir(parents=True, exist_ok=True)
    set_games_dir(games_dir)
    # Set up image generation provider (default from IMAGE_PROVIDERS order)
    from captain_claw.games.image_service import switch_provider, list_providers, get_provider_id
    providers = list_providers()
    default_id = get_provider_id()
    default_prov = next((p for p in providers if p["id"] == default_id and p["available"]), None)
    if not default_prov:
        default_prov = next((p for p in providers if p["available"]), None)
    if default_prov:
        switch_provider(default_prov["id"])
        print(f"Flight Deck: image provider '{default_prov['id']}' enabled")
    else:
        print("Flight Deck: no image provider available")
    # Defer reattach so uvicorn has time to actually bind its listening
    # socket. Otherwise the children we launch here race to call
    # /fd/processes/{slug}/announce-port before FD is accepting connections,
    # and their drift announcements get "connection refused".
    async def _deferred_reattach():
        await asyncio.sleep(1.0)
        await asyncio.to_thread(_reattach_processes)
    asyncio.create_task(_deferred_reattach())
    # Initialize database for auth & settings
    if AUTH_ENABLED:
        _fd_db = FlightDeckDB(DATA_DIR / "flight-deck.db")
        await _fd_db.init()
        set_auth_db(_fd_db)
        app.state.fd_db = _fd_db
        # Load admin-configured plan limits from DB
        plan_limits_raw = await _fd_db.get_system_setting("fd:plan-limits")
        load_plan_limits_from_db_sync(plan_limits_raw)
        # Initialize vast.ai GPU cloud integration.
        from captain_claw.vastai.manager import VastAIManager
        from captain_claw.flight_deck.vastai_routes import set_vastai_manager
        from captain_claw.vastai.wake import register_manager as _register_vastai_wake
        _vastai_mgr = VastAIManager(_fd_db)
        await _vastai_mgr.initialize()
        set_vastai_manager(_vastai_mgr)
        _register_vastai_wake(_vastai_mgr)
        app.state.vastai_manager = _vastai_mgr
    # ── Code-app subprocess runtime ──
    # Lazy-spawn per app on first request; reaper kicks in here.
    from captain_claw.flight_deck.app_runtime import get_runtime as _get_app_runtime
    _app_rt = _get_app_runtime()
    await _app_rt.start()
    app.state.app_runtime = _app_rt
    # ── FD scheduler (proactive push: agent prompts on a timer → WhatsApp /
    # channel). Disable with FD_SCHEDULER_DISABLED=true. ──
    if os.environ.get("FD_SCHEDULER_DISABLED", "").lower() not in ("true", "1", "yes"):
        from captain_claw.flight_deck.fd_scheduler import scheduler_loop as _sched_loop
        _sched_stop = asyncio.Event()
        app.state.scheduler_stop = _sched_stop
        app.state.scheduler_task = asyncio.create_task(_sched_loop(_sched_stop))
        print("Flight Deck: scheduler started")
    # ── Flow engine (process automations: trigger → steps on the agent pool) ──
    try:
        from captain_claw.flight_deck.flows_store import FlowStore
        from captain_claw.flight_deck.flow_runner import FlowRunner
        from captain_claw.flight_deck import flow_router
        _flow_store = FlowStore(DATA_DIR / "flows.db")
        _fd_port = os.environ.get("FD_PORT", "25080")
        _flow_runner = FlowRunner(
            _flow_store,
            get_agents=_running_agents,
            resolve_auth=_resolve_agent_auth,
            fd_self_base=f"http://localhost:{_fd_port}",
            fd_tools=_fd_internal_tools(),
            whatsapp_send=_flow_whatsapp_send,
            transfer_file=_transfer_file_to_agent,
        )
        app.state.flow_store = _flow_store
        app.state.flow_runner = _flow_runner
        flow_router.set_engine(_flow_store, _flow_runner)
        print("Flight Deck: flow engine ready")
    except Exception as _exc:
        print(f"Flight Deck: flow engine init failed: {_exc}")
    yield
    # Stop the scheduler loop first so it doesn't fire mid-shutdown.
    if hasattr(app.state, "scheduler_stop"):
        app.state.scheduler_stop.set()
        if hasattr(app.state, "scheduler_task"):
            try:
                await asyncio.wait_for(app.state.scheduler_task, timeout=5.0)
            except (asyncio.TimeoutError, asyncio.CancelledError, Exception):
                pass
    # Shutdown: stop all managed process agents
    print("Flight Deck: stopping managed process agents...")
    _stop_all_processes()
    # Stop every running code-app subprocess + cancel the idle reaper.
    if hasattr(app.state, "app_runtime"):
        try:
            await app.state.app_runtime.shutdown()
        except Exception as _exc:
            print(f"Flight Deck: app_runtime shutdown error: {_exc}")
    if hasattr(app.state, "vastai_manager"):
        await app.state.vastai_manager.shutdown()
    if AUTH_ENABLED and hasattr(app.state, "fd_db"):
        await app.state.fd_db.close()
    if _client:
        _client.close()


app = FastAPI(title="Flight Deck", lifespan=lifespan)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# Log validation errors with full detail for debugging
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    log.error("Validation error on %s %s: %s", request.method, request.url.path, exc.errors())
    return JSONResponse(status_code=422, content={"detail": exc.errors()})

# ── Auth dependency that always uses Depends ──
# When AUTH_ENABLED is False, using `= None` instead of `= Depends(...)` causes
# FastAPI to treat the parameter as a body field, breaking request parsing.
# This wrapper ensures we always use Depends regardless of auth state.

async def _no_user() -> None:
    return None

_optional_user_dep = Depends(get_optional_user) if AUTH_ENABLED else Depends(_no_user)
_required_user_dep = Depends(get_current_user) if AUTH_ENABLED else Depends(_no_user)

# ── Auth & user routes ──

from captain_claw.flight_deck.auth_routes import router as auth_router
from captain_claw.flight_deck.settings_routes import router as settings_router
from captain_claw.flight_deck.chat_routes import router as chat_router
from captain_claw.flight_deck.admin_routes import router as admin_router
from captain_claw.flight_deck.council_routes import router as council_router
from captain_claw.flight_deck.google_oauth_routes import router as google_oauth_router
from captain_claw.flight_deck.codex_oauth_routes import router as codex_oauth_router
from captain_claw.flight_deck.games_routes import router as games_router
from captain_claw.flight_deck.vastai_routes import router as vastai_router
from captain_claw.flight_deck.prompt_routes import router as prompt_router
from captain_claw.flight_deck.project_routes import router as project_router
from captain_claw.flight_deck.mcp_routes import router as mcp_router
from captain_claw.flight_deck.app_routes import router as apps_router
from captain_claw.flight_deck.app_files_routes import router as app_files_router
from captain_claw.flight_deck.app_builtin_routes import router as app_builtin_router
from captain_claw.flight_deck.app_code_routes import router as app_code_router
from captain_claw.flight_deck.glasses_bridge import router as glasses_router
from captain_claw.flight_deck.face_routes import router as face_router
from captain_claw.flight_deck.messenger_bridge import router as messenger_router
from captain_claw.flight_deck.whatsapp_bridge import router as whatsapp_router
from captain_claw.flight_deck.fd_scheduler import router as scheduler_router

app.include_router(auth_router)
app.include_router(settings_router)
app.include_router(chat_router)
app.include_router(admin_router)
app.include_router(council_router)
app.include_router(google_oauth_router)
app.include_router(codex_oauth_router)
app.include_router(games_router)
app.include_router(vastai_router)
app.include_router(prompt_router)
app.include_router(project_router)
app.include_router(mcp_router)
# Legacy manifest-based app routes. The agent-coded app runtime
# (``app_code_router``) is the going-forward path; the manifest
# routes are kept for migration but disabled by default. Re-enable
# by setting ``FD_LEGACY_MANIFEST_APPS=true``.
if os.environ.get("FD_LEGACY_MANIFEST_APPS", "false").lower() in ("true", "1", "yes"):
    app.include_router(apps_router)
    app.include_router(app_files_router)
    app.include_router(app_builtin_router)
app.include_router(app_code_router)
app.include_router(glasses_router)
app.include_router(face_router)
app.include_router(messenger_router)
app.include_router(whatsapp_router)
app.include_router(scheduler_router)


# ── Auth dependency helper ──

def _optional_user():
    """Return a dependency that requires auth when enabled, skips when disabled."""
    if AUTH_ENABLED:
        return Depends(get_current_user)
    return None


async def _get_user_id(request: Request) -> str:
    """Extract user_id from request state (set by auth middleware). Returns '' when auth disabled."""
    return getattr(request.state, "user_id", "")


def _require_auth():
    """FastAPI dependency that enforces auth when FD_AUTH_ENABLED=true."""
    async def _dep(user: dict = Depends(get_current_user)):
        return user
    if AUTH_ENABLED:
        return Depends(_dep)
    return None


# ── Models ──

class AgentConfig(BaseModel):
    """Agent spawn configuration — matches the frontend form."""
    # Identity
    name: str = ""
    description: str = ""
    hostname: str = "captain-claw"
    image: str = CC_IMAGE_DEFAULT

    # LLM
    provider: str = "ollama"
    model: str = "minimax-m2.7:cloud"
    temperature: float = 0.7
    max_tokens: int = 32768
    provider_api_key: str = ""
    base_url: str = ""
    # Optional per-session selectable model list (config.model.allowed). Used by
    # the free-OpenRouter "Freebie" spawn so all free models are available to
    # switch between at runtime, with `model` as the default.
    allowed_models: list[dict] = Field(default_factory=list)

    # BotPort
    botport_enabled: bool = True
    botport_url: str = ""
    botport_instance_name: str = ""
    botport_key: str = ""
    botport_secret: str = ""
    botport_max_concurrent: int = 5

    # Tools
    tools: list[str] = Field(default_factory=lambda: [
        "shell", "read", "write", "glob", "edit",
        "web_fetch", "web_search", "browser", "botport",
    ])

    # Web
    web_enabled: bool = True
    web_port: int = 24080
    web_auth_token: str = ""

    # Platforms
    telegram_enabled: bool = False
    telegram_bot_token: str = ""
    discord_enabled: bool = False
    discord_bot_token: str = ""
    slack_enabled: bool = False
    slack_bot_token: str = ""

    # Cognitive mode
    cognitive_mode: str = "neutra"

    # Docker
    network_mode: str = "host"
    restart_policy: str = "unless-stopped"
    extra_volumes: list[dict] = Field(default_factory=list)
    env_vars: list[dict] = Field(default_factory=list)

    # Ownership hint — used by internal callers (e.g. Old Man) that cannot
    # authenticate via JWT but need the spawned agent to inherit the owner.
    owner_hint: str = ""


class ContainerInfo(BaseModel):
    id: str
    name: str
    status: str
    image: str
    created: str
    agent_name: str = ""
    description: str = ""
    ports: dict = Field(default_factory=dict)
    web_port: int | None = None
    web_auth: str = ""


class ContainerActionResult(BaseModel):
    ok: bool
    container_id: str
    message: str = ""
    old_container_id: str = ""


# ── Helpers ──

def _slug(name: str) -> str:
    import re
    return re.sub(r"[^a-z0-9-]", "-", (name or "cc-agent").lower()).strip("-") or "cc-agent"


def _docker_host() -> str:
    """Return the hostname containers should use to reach the Docker host."""
    import platform
    # On macOS/Windows, host.docker.internal resolves to the host.
    # On Linux with host networking, localhost works; with bridge, use the gateway.
    if platform.system() in ("Darwin", "Windows"):
        return "host.docker.internal"
    return "127.0.0.1"


def _build_config_yaml(c: AgentConfig) -> str:
    """Generate config.yaml content from agent config."""
    dhost = _docker_host()
    cfg: dict = {
        "model": {
            "provider": c.provider,
            "model": c.model,
            "temperature": c.temperature,
            "max_tokens": c.max_tokens,
            "api_key": "",  # Key goes in .env
            "base_url": (
                c.base_url
                if c.base_url
                else (f"http://{dhost}:11434" if c.provider == "ollama" else "")
            ),
            **({"allowed": c.allowed_models} if c.allowed_models else {}),
        },
        "context": {
            "max_tokens": 160000,
            "compaction_threshold": 0.8,
            "compaction_ratio": 0.4,
        },
        "memory": {
            "enabled": True,
            "path": "/home/claw/.captain-claw/memory.db",
            "index_workspace": True,
            "index_sessions": True,
            "embeddings": {
                "provider": "auto",
                "ollama_model": "nomic-embed-text",
                "ollama_base_url": f"http://{dhost}:11434",
                "fallback_to_local_hash": True,
            },
        },
        "tools": {
            "enabled": c.tools,
            "shell": {"timeout": 120, "default_policy": "ask"},
            "browser": {"headless": True, "viewport_width": 1280, "viewport_height": 720},
            "web_search": {"provider": "brave", "max_results": 5},
            "require_confirmation": ["shell", "write", "edit"],
        },
        "session": {"storage": "sqlite", "path": "/data/sessions/sessions.db", "auto_save": True},
        "workspace": {"path": "/data/workspace"},
        "web": {
            "enabled": c.web_enabled,
            "host": "0.0.0.0",
            "port": c.web_port,
            "api_enabled": True,
            "auth_token": c.web_auth_token,
        },
        "botport": {
            "enabled": c.botport_enabled,
            "url": c.botport_url,
            "instance_name": c.botport_instance_name or c.name or "default",
            "key": c.botport_key,
            "secret": c.botport_secret,
            "advertise_personas": True,
            "advertise_tools": True,
            "advertise_models": True,
            "max_concurrent": c.botport_max_concurrent,
            "reconnect_delay_seconds": 5.0,
            "heartbeat_interval_seconds": 30.0,
        },
        "telegram": {"enabled": c.telegram_enabled, "bot_token": c.telegram_bot_token},
        "discord": {"enabled": c.discord_enabled, "bot_token": c.discord_bot_token},
        "slack": {"enabled": c.slack_enabled, "bot_token": c.slack_bot_token},
        "logging": {"level": "INFO", "format": "console"},
        "cognitive_mode": {
            "enabled": True,
            "default_mode": c.cognitive_mode,
        },
    }
    return yaml.dump(cfg, default_flow_style=False, sort_keys=False, allow_unicode=True)


def _build_env(c: AgentConfig) -> str:
    lines: list[str] = []
    if c.provider_api_key:
        # Map provider to env var name
        key_map = {
            "anthropic": "ANTHROPIC_API_KEY",
            "openai": "OPENAI_API_KEY",
            "gemini": "GEMINI_API_KEY",
            "xai": "XAI_API_KEY",
            "openrouter": "OPENROUTER_API_KEY",
        }
        env_name = key_map.get(c.provider, "API_KEY")
        lines.append(f"{env_name}={c.provider_api_key}")
    # Ollama inside Docker needs to reach the host
    if c.provider == "ollama":
        dhost = _docker_host()
        lines.append(f"OLLAMA_BASE_URL=http://{dhost}:11434")
    for ev in c.env_vars:
        if ev.get("key"):
            lines.append(f"{ev['key']}={ev.get('value', '')}")
    return "\n".join(lines) + "\n" if lines else ""


def _localize_url(url: str) -> str:
    """Rewrite Docker-internal hostnames to localhost for process agents."""
    return url.replace("host.docker.internal", "localhost").replace("host.docker.internal", "127.0.0.1") if url else url


def _build_process_config_yaml(c: AgentConfig, agent_dir: Path) -> str:
    """Generate config.yaml for a pip-installed process agent (local paths)."""
    home_config = agent_dir / "data" / "home-config"
    # For process agents, rewrite Docker-internal URLs to localhost
    botport_url = _localize_url(c.botport_url)
    cfg: dict = {
        "model": {
            "provider": c.provider,
            "model": c.model,
            "temperature": c.temperature,
            "max_tokens": c.max_tokens,
            "api_key": "",  # Key goes in .env
            "base_url": (
                c.base_url
                if c.base_url
                else ("http://127.0.0.1:11434" if c.provider == "ollama" else "")
            ),
            **({"allowed": c.allowed_models} if c.allowed_models else {}),
        },
        "context": {
            "max_tokens": 160000,
            "compaction_threshold": 0.8,
            "compaction_ratio": 0.4,
        },
        "memory": {
            "enabled": True,
            "path": str(home_config / "memory.db"),
            "index_workspace": True,
            "index_sessions": True,
            "embeddings": {
                "provider": "auto",
                "ollama_model": "nomic-embed-text",
                "ollama_base_url": "http://127.0.0.1:11434",
                "fallback_to_local_hash": True,
            },
        },
        "tools": {
            "enabled": c.tools,
            "shell": {"timeout": 120, "default_policy": "ask"},
            "browser": {"headless": True, "viewport_width": 1280, "viewport_height": 720},
            "web_search": {"provider": "brave", "max_results": 5},
            "require_confirmation": ["shell", "write", "edit"],
        },
        "session": {
            "storage": "sqlite",
            "path": str(agent_dir / "data" / "sessions" / "sessions.db"),
            "auto_save": True,
        },
        "workspace": {"path": str(agent_dir / "data" / "workspace")},
        "web": {
            "enabled": c.web_enabled,
            "host": "127.0.0.1",
            "port": c.web_port,
            "api_enabled": True,
            "auth_token": c.web_auth_token,
        },
        "botport": {
            "enabled": c.botport_enabled,
            "url": botport_url,
            "instance_name": c.botport_instance_name or c.name or "default",
            "key": c.botport_key,
            "secret": c.botport_secret,
            "advertise_personas": True,
            "advertise_tools": True,
            "advertise_models": True,
            "max_concurrent": c.botport_max_concurrent,
            "reconnect_delay_seconds": 5.0,
            "heartbeat_interval_seconds": 30.0,
        },
        "telegram": {"enabled": c.telegram_enabled, "bot_token": c.telegram_bot_token},
        "discord": {"enabled": c.discord_enabled, "bot_token": c.discord_bot_token},
        "slack": {"enabled": c.slack_enabled, "bot_token": c.slack_bot_token},
        "logging": {"level": "INFO", "format": "console"},
    }
    return yaml.dump(cfg, default_flow_style=False, sort_keys=False, allow_unicode=True)


def _container_info(c: docker.models.containers.Container) -> ContainerInfo:
    labels = c.labels or {}
    web_port_str = labels.get("flight-deck.web-port", "")
    # Try label first, then fall back to Docker port bindings
    web_port: int | None = int(web_port_str) if web_port_str else None
    if web_port is None:
        # Extract from Docker port mappings (e.g. {"24080/tcp": [{"HostPort": "24080"}]})
        docker_ports = c.attrs.get("NetworkSettings", {}).get("Ports", {}) or {}
        for container_port, bindings in docker_ports.items():
            if bindings:
                try:
                    web_port = int(bindings[0].get("HostPort", 0))
                    if web_port:
                        break
                except (ValueError, IndexError, TypeError):
                    pass
    return ContainerInfo(
        id=c.short_id,
        name=c.name,
        status=c.status,
        image=str(c.image.tags[0]) if c.image.tags else str(c.image.short_id),
        created=str(c.attrs.get("Created", "")),
        agent_name=labels.get("flight-deck.agent-name", ""),
        description=labels.get("flight-deck.description", ""),
        ports=c.attrs.get("NetworkSettings", {}).get("Ports", {}),
        web_port=web_port,
        web_auth=labels.get("flight-deck.web-auth", ""),
    )


def _find_container(container_id: str, owner_id: str = "") -> docker.models.containers.Container:
    client = get_docker()
    # Try by short ID, full ID, or name
    for c in client.containers.list(all=True, filters={"label": CONTAINER_LABEL}):
        if c.short_id == container_id or c.id == container_id or c.name == container_id:
            if AUTH_ENABLED and owner_id:
                if (c.labels or {}).get(OWNER_LABEL, "") != owner_id:
                    raise HTTPException(status_code=404, detail=f"Container {container_id} not found")
            return c
    raise HTTPException(status_code=404, detail=f"Container {container_id} not found")


# ── Endpoints ──

@app.get("/fd/containers", response_model=list[ContainerInfo])
async def list_containers(request: Request, user: dict | None = _required_user_dep):
    """List all Flight Deck managed containers (filtered by owner when auth enabled)."""
    try:
        client = get_docker()
        containers = client.containers.list(all=True, filters={"label": CONTAINER_LABEL})
    except Exception:
        return []  # Docker not available (e.g. running inside a container)
    user_id = getattr(request.state, "user_id", "")
    if AUTH_ENABLED and user_id:
        containers = [c for c in containers if (c.labels or {}).get(OWNER_LABEL, "") == user_id]
    return [_container_info(c) for c in containers]


def _is_port_available(port: int) -> bool:
    """Check if a TCP port is available on localhost."""
    import socket
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        try:
            s.bind(("127.0.0.1", port))
            return True
        except OSError:
            return False


def _schedule_fleet_notify(name: str, port: int, event: str = "joined", owner_id: str = ""):
    """Schedule a fleet notification as a background async task."""
    async def _run():
        # Give the new agent a moment to start up before notifying peers
        await asyncio.sleep(5)
        await _notify_fleet_change(name, port, event, owner_id=owner_id)
    try:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            loop.create_task(_run())
        else:
            asyncio.run(_run())
    except RuntimeError:
        pass  # Best-effort


@app.post("/fd/spawn", response_model=ContainerActionResult)
async def spawn_agent(config: AgentConfig, request: Request, user: dict | None = _optional_user_dep):
    """Spawn a new Captain Claw container."""
    # Check if docker spawn is allowed
    sys_cfg = await _get_system_config()
    docker_default = not os.environ.get("CAPTAIN_CLAW_DOCKER")
    if not sys_cfg.get("docker_spawn_enabled", docker_default):
        raise HTTPException(403, "Docker container spawning is disabled by the administrator.")
    # Rate limiting & agent count check
    if AUTH_ENABLED and user:
        check_api_rate_limit(user)
        check_spawn_rate_limit(user)
        # Count existing containers for this user
        client_tmp = get_docker()
        user_id = user["id"]
        owned = [c for c in client_tmp.containers.list(all=True, filters={"label": CONTAINER_LABEL})
                 if (c.labels or {}).get(OWNER_LABEL, "") == user_id]
        await check_agent_count_limit(user, len(owned))

    client = get_docker()
    slug = _slug(config.name)

    # Ensure port is available; find a free one if not
    if config.web_enabled and (config.web_port <= 0 or not _is_port_available(config.web_port)):
        config.web_port = _find_available_port(config.web_port if config.web_port > 0 else 24080)

    # Auto-generate auth token if none provided — prevents unauthenticated
    # direct access to agent ports bypassing Flight Deck.
    if config.web_enabled and not config.web_auth_token:
        config.web_auth_token = secrets.token_urlsafe(32)

    # Check for name collision
    try:
        existing = client.containers.get(slug)
        if existing.status == "running":
            raise HTTPException(400, f"Container '{slug}' already running. Stop it first or use a different name.")
        # Remove stopped container with same name
        existing.remove()
    except docker.errors.NotFound:
        pass

    # Prepare data directory
    agent_dir = DATA_DIR / slug
    agent_dir.mkdir(parents=True, exist_ok=True)
    (agent_dir / "data" / "workspace").mkdir(parents=True, exist_ok=True)
    (agent_dir / "data" / "sessions").mkdir(parents=True, exist_ok=True)
    (agent_dir / "data" / "skills").mkdir(parents=True, exist_ok=True)
    (agent_dir / "data" / "home-config").mkdir(parents=True, exist_ok=True)

    # Write config files
    config_yaml = _build_config_yaml(config)
    (agent_dir / "config.yaml").write_text(config_yaml)
    # Also write into home-config so it takes precedence over any stale config
    # that CC's settings page may have written into ~/.captain-claw/config.yaml
    (agent_dir / "data" / "home-config" / "config.yaml").write_text(config_yaml)

    env_content = _build_env(config)
    (agent_dir / ".env").write_text(env_content)

    # Write cognitive mode file for the agent (Docker spawn).
    if config.cognitive_mode and config.cognitive_mode != "neutra":
        mode_file = agent_dir / "data" / "home-config" / "cognitive_mode.txt"
        mode_file.write_text(config.cognitive_mode, encoding="utf-8")

    # New agents deploy in eco mode by default.
    _write_eco_flag_on_spawn(agent_dir)

    # Build volume mounts
    # CC WORKDIR is /app — it loads ./config.yaml from CWD (/app/config.yaml)
    # and ~/.captain-claw/config.yaml (home dir overlay, highest priority).
    # Mount config to /app/config.yaml (CWD) so CC finds it on startup.
    # Mount home-config dir for memory.db and other persistent state.
    config_file = str(agent_dir / "config.yaml")
    volumes = {
        config_file: {"bind": "/app/config.yaml", "mode": "ro"},
        str(agent_dir / ".env"): {"bind": "/app/.env", "mode": "ro"},
        str(agent_dir / "data" / "home-config"): {"bind": "/home/claw/.captain-claw", "mode": "rw"},
        str(agent_dir / "data" / "workspace"): {"bind": "/data/workspace", "mode": "rw"},
        str(agent_dir / "data" / "sessions"): {"bind": "/data/sessions", "mode": "rw"},
        str(agent_dir / "data" / "skills"): {"bind": "/data/skills", "mode": "rw"},
    }
    for ev in config.extra_volumes:
        host = ev.get("host", "")
        container = ev.get("container", "")
        if host and container:
            volumes[host] = {"bind": container, "mode": "rw"}

    # Build environment
    environment: dict[str, str] = {}
    if env_content:
        for line in env_content.strip().split("\n"):
            if "=" in line:
                k, v = line.split("=", 1)
                environment[k] = v
    for ev in config.env_vars:
        if ev.get("key"):
            environment[ev["key"]] = ev.get("value", "")

    # Resolve owner: authenticated user > owner_hint > infer from existing agents
    owner_id = getattr(request.state, "user_id", "") or config.owner_hint
    if not owner_id:
        # Fallback: inherit owner from an existing agent in the registry.
        registry = _load_process_registry()
        for _entry in registry.values():
            if _entry.get("owner"):
                owner_id = _entry["owner"]
                break
        if not owner_id:
            # Try Docker containers
            try:
                for c in client.containers.list(all=True, filters={"label": CONTAINER_LABEL}):
                    o = (c.labels or {}).get(OWNER_LABEL, "")
                    if o:
                        owner_id = o
                        break
            except Exception:
                pass

    # Pass owner ID so child agents can propagate ownership when spawning
    if owner_id:
        environment["FD_OWNER_ID"] = owner_id

    # Slug for port-fallback callbacks (Docker path)
    environment["FD_AGENT_SLUG"] = slug

    # Labels for tracking
    labels = {
        CONTAINER_LABEL: "true",
        OWNER_LABEL: owner_id,
        "flight-deck.agent-name": config.name or slug,
        "flight-deck.description": config.description or "",
        "flight-deck.image": config.image,
        "flight-deck.web-port": str(config.web_port) if config.web_enabled else "",
        "flight-deck.web-auth": config.web_auth_token or "",
    }

    # Security options
    security_opt = ["no-new-privileges:true", "seccomp:unconfined"]

    # Restart policy
    restart_map = {
        "unless-stopped": {"Name": "unless-stopped"},
        "always": {"Name": "always"},
        "on-failure": {"Name": "on-failure", "MaximumRetryCount": 5},
        "no": {"Name": ""},
    }
    restart = restart_map.get(config.restart_policy, {"Name": "unless-stopped"})

    # Port publishing — needed on macOS where host networking doesn't work.
    import platform
    ports: dict[str, int] = {}
    use_network: str | None = config.network_mode
    if platform.system() == "Darwin" and config.network_mode == "host":
        # macOS: host networking is a no-op, switch to default bridge + port mapping
        use_network = None
        if config.web_enabled:
            ports[f"{config.web_port}/tcp"] = config.web_port
    elif config.network_mode != "host":
        # Explicit bridge/custom network: publish web port
        if config.web_enabled:
            ports[f"{config.web_port}/tcp"] = config.web_port

    try:
        container = client.containers.run(
            image=config.image,
            name=slug,
            hostname=config.hostname or slug,
            detach=True,
            network_mode=use_network,
            ports=ports or None,
            volumes=volumes,
            environment=environment,
            labels=labels,
            security_opt=security_opt,
            cap_drop=["ALL"],
            cap_add=["CHOWN", "SETUID", "SETGID", "SYS_CHROOT"],
            tmpfs={"/tmp": "", "/run": ""},
            restart_policy=restart,
            stop_signal="SIGTERM",
        )
        port_info = f" on port {config.web_port}" if config.web_enabled else ""
        # Log usage
        if AUTH_ENABLED and user:
            db = app.state.fd_db
            await db.log_usage(user["id"], "agent_spawn", json.dumps({"agent": slug, "type": "container", "image": config.image}))
        # Notify other agents about the new peer (scoped to same owner)
        if config.web_enabled:
            _schedule_fleet_notify(config.name or slug, config.web_port, owner_id=owner_id)
        return ContainerActionResult(ok=True, container_id=container.short_id, message=f"Agent '{slug}' spawned{port_info}")
    except docker.errors.ImageNotFound:
        raise HTTPException(404, f"Docker image '{config.image}' not found. Pull it first.")
    except docker.errors.APIError as exc:
        raise HTTPException(500, f"Docker error: {exc.explanation or str(exc)}")


@app.post("/fd/containers/{container_id}/stop", response_model=ContainerActionResult)
async def stop_container(container_id: str, request: Request, user: dict | None = _required_user_dep):
    import asyncio
    c = _find_container(container_id, getattr(request.state, "user_id", ""))
    if c.status != "running":
        return ContainerActionResult(ok=True, container_id=c.short_id, message="Already stopped")
    await asyncio.get_event_loop().run_in_executor(None, lambda: c.stop(timeout=5))
    return ContainerActionResult(ok=True, container_id=c.short_id, message="Stopped")


@app.post("/fd/containers/{container_id}/start", response_model=ContainerActionResult)
async def start_container(container_id: str, request: Request, user: dict | None = _required_user_dep):
    import asyncio
    c = _find_container(container_id, getattr(request.state, "user_id", ""))
    if c.status == "running":
        return ContainerActionResult(ok=True, container_id=c.short_id, message="Already running")
    try:
        await asyncio.get_event_loop().run_in_executor(None, c.start)
    except docker.errors.APIError as exc:
        explanation = exc.explanation or str(exc)
        raise HTTPException(500, f"Docker start failed: {explanation}")
    return ContainerActionResult(ok=True, container_id=c.short_id, message="Started")


@app.post("/fd/containers/{container_id}/restart", response_model=ContainerActionResult)
async def restart_container(container_id: str, request: Request, user: dict | None = _required_user_dep):
    import asyncio
    c = _find_container(container_id, getattr(request.state, "user_id", ""))
    await asyncio.get_event_loop().run_in_executor(None, lambda: c.restart(timeout=5))
    return ContainerActionResult(ok=True, container_id=c.short_id, message="Restarted")


@app.delete("/fd/containers/{container_id}", response_model=ContainerActionResult)
async def remove_container(container_id: str, force: bool = False, request: Request = None, user: dict | None = _required_user_dep):
    import asyncio
    c = _find_container(container_id, getattr(request.state, "user_id", ""))
    name = c.name
    await asyncio.get_event_loop().run_in_executor(None, lambda: c.remove(force=force))
    return ContainerActionResult(ok=True, container_id=container_id, message=f"Removed '{name}'")


class RebuildRequest(BaseModel):
    description: str = ""  # Frontend sends current description override


class CloneRequest(BaseModel):
    new_name: str  # Name for the cloned agent


@app.post("/fd/containers/{container_id}/rebuild", response_model=ContainerActionResult)
async def rebuild_container(container_id: str, request: Request, req: RebuildRequest | None = None, user: dict | None = _required_user_dep):
    """Rebuild a container: stop, remove, pull latest image, re-spawn with same config."""
    import platform

    c = _find_container(container_id, getattr(request.state, "user_id", ""))
    old_short_id = c.short_id
    labels = c.labels or {}

    # If frontend sent a description override, update the label
    if req and req.description:
        labels["flight-deck.description"] = req.description

    # Read the original spawn config from labels + container inspection
    agent_name = labels.get("flight-deck.agent-name", c.name)
    slug = _slug(agent_name)
    image = labels.get("flight-deck.image", CC_IMAGE_DEFAULT)
    web_port_str = labels.get("flight-deck.web-port", "")
    web_port = int(web_port_str) if web_port_str else None
    web_auth = labels.get("flight-deck.web-auth", "")
    description = labels.get("flight-deck.description", "")

    # Read environment from the running container
    env_list = c.attrs.get("Config", {}).get("Env", [])
    environment: dict[str, str] = {}
    for e in env_list:
        if "=" in e:
            k, v = e.split("=", 1)
            environment[k] = v

    # Read mounts from container
    mounts = c.attrs.get("Mounts", [])
    volumes: dict[str, dict] = {}
    for m in mounts:
        src = m.get("Source", "")
        dst = m.get("Destination", "")
        mode = m.get("Mode", "rw")
        if src and dst:
            volumes[src] = {"bind": dst, "mode": mode}

    # Read restart policy
    host_config = c.attrs.get("HostConfig", {})
    restart_policy = host_config.get("RestartPolicy", {"Name": "unless-stopped"})

    # Read security opts, cap_drop, cap_add
    security_opt = host_config.get("SecurityOpt", ["no-new-privileges:true", "seccomp:unconfined"])
    cap_drop = host_config.get("CapDrop", ["ALL"])
    cap_add = host_config.get("CapAdd", ["CHOWN", "SETUID", "SETGID", "SYS_CHROOT"])

    # Read tmpfs
    tmpfs = host_config.get("Tmpfs", {"/tmp": "", "/run": ""})

    # Read network mode
    network_mode = host_config.get("NetworkMode", "host")

    # Read hostname
    hostname = c.attrs.get("Config", {}).get("Hostname", slug)

    # Port publishing
    ports: dict[str, int] = {}
    network_mode_use: str | None = network_mode
    if platform.system() == "Darwin" and network_mode == "host":
        network_mode_use = None
        if web_port:
            ports[f"{web_port}/tcp"] = web_port
    elif network_mode != "host":
        if web_port:
            ports[f"{web_port}/tcp"] = web_port

    # Stop and remove old container
    if c.status == "running":
        c.stop(timeout=5)
    c.remove(force=True)

    # Pull latest image
    client = get_docker()
    try:
        client.images.pull(image)
    except docker.errors.APIError:
        pass  # If pull fails, use whatever's cached locally

    # Re-create with same config
    try:
        new_container = client.containers.run(
            image=image,
            name=slug,
            hostname=hostname,
            detach=True,
            network_mode=network_mode_use,
            ports=ports or None,
            volumes=volumes,
            environment=environment,
            labels=labels,
            security_opt=security_opt,
            cap_drop=cap_drop,
            cap_add=cap_add,
            tmpfs=tmpfs,
            restart_policy=restart_policy,
            stop_signal="SIGTERM",
        )
        return ContainerActionResult(
            ok=True,
            container_id=new_container.short_id,
            old_container_id=old_short_id,
            message=f"Agent '{agent_name}' rebuilt with latest image",
        )
    except docker.errors.ImageNotFound:
        raise HTTPException(404, f"Docker image '{image}' not found.")
    except docker.errors.APIError as exc:
        raise HTTPException(500, f"Docker error: {exc.explanation or str(exc)}")


def _find_available_port(start: int) -> int:
    """Find first available TCP port starting from `start`, checking
    running containers, managed processes, and the host's listening sockets."""
    import socket

    used_ports: set[int] = set()

    # Collect ports from Docker containers
    try:
        client = get_docker()
        for c in client.containers.list(all=True, filters={"label": CONTAINER_LABEL}):
            lbl_port = (c.labels or {}).get("flight-deck.web-port", "")
            if lbl_port:
                try:
                    used_ports.add(int(lbl_port))
                except ValueError:
                    pass
            docker_ports = c.attrs.get("NetworkSettings", {}).get("Ports", {}) or {}
            for _cp, bindings in docker_ports.items():
                if bindings:
                    for b in bindings:
                        try:
                            used_ports.add(int(b.get("HostPort", 0)))
                        except (ValueError, TypeError):
                            pass
    except Exception:
        pass  # Docker may not be available

    # Collect ports from managed processes
    registry = _load_process_registry()
    for entry in registry.values():
        wp = entry.get("web_port")
        if wp:
            used_ports.add(wp)

    max_search = int(os.environ.get("FD_PORT_RANGE", "500"))
    port = start
    while port < start + max_search:
        if port not in used_ports:
            # Also check if host port is free
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                try:
                    s.bind(("127.0.0.1", port))
                    return port
                except OSError:
                    pass
        port += 1
    raise HTTPException(500, f"No available port found in range {start}-{start + max_search}")


@app.post("/fd/containers/{container_id}/clone", response_model=ContainerActionResult)
async def clone_container(container_id: str, req: CloneRequest, request: Request, user: dict | None = _required_user_dep):
    """Clone a container: create a new agent with same config but its own data folder."""
    import platform
    import shutil

    c = _find_container(container_id, getattr(request.state, "user_id", ""))
    labels = dict(c.labels or {})

    # Read the original config
    image = labels.get("flight-deck.image", CC_IMAGE_DEFAULT)
    web_port_str = labels.get("flight-deck.web-port", "")
    old_web_port = int(web_port_str) if web_port_str else None
    web_auth = labels.get("flight-deck.web-auth", "")
    old_agent_name = labels.get("flight-deck.agent-name", c.name)

    new_name = req.new_name.strip()
    if not new_name:
        raise HTTPException(400, "Name is required")
    new_slug = _slug(new_name)

    # Check name collision
    client = get_docker()
    try:
        existing = client.containers.get(new_slug)
        if existing.status == "running":
            raise HTTPException(400, f"Container '{new_slug}' already running.")
        existing.remove()
    except docker.errors.NotFound:
        pass

    # Determine old agent directory from the actual mount sources
    old_slug = _slug(old_agent_name)
    old_agent_dir = DATA_DIR / old_slug
    new_agent_dir = DATA_DIR / new_slug

    # Create new data directory structure
    new_agent_dir.mkdir(parents=True, exist_ok=True)
    (new_agent_dir / "data" / "workspace").mkdir(parents=True, exist_ok=True)
    (new_agent_dir / "data" / "sessions").mkdir(parents=True, exist_ok=True)
    (new_agent_dir / "data" / "skills").mkdir(parents=True, exist_ok=True)
    (new_agent_dir / "data" / "home-config").mkdir(parents=True, exist_ok=True)

    # Copy config.yaml and .env from original if they exist
    for fname in ("config.yaml", ".env"):
        src_file = old_agent_dir / fname
        if src_file.is_file():
            shutil.copy2(str(src_file), str(new_agent_dir / fname))
    # Copy home-config/config.yaml too
    src_hc = old_agent_dir / "data" / "home-config" / "config.yaml"
    if src_hc.is_file():
        shutil.copy2(str(src_hc), str(new_agent_dir / "data" / "home-config" / "config.yaml"))

    # Read environment from the source container
    env_list = c.attrs.get("Config", {}).get("Env", [])
    environment: dict[str, str] = {}
    for e in env_list:
        if "=" in e:
            k, v = e.split("=", 1)
            environment[k] = v

    # Build volume mounts for the clone — use the known structure
    # instead of trying to remap arbitrary paths from the old container.
    old_agent_dir_str = str(old_agent_dir)
    new_agent_dir_str = str(new_agent_dir)
    config_file = str(new_agent_dir / "config.yaml")
    env_file = str(new_agent_dir / ".env")

    # Start with the standard CC mounts pointing to new data dir
    volumes: dict[str, dict] = {
        config_file: {"bind": "/app/config.yaml", "mode": "ro"},
        env_file: {"bind": "/app/.env", "mode": "ro"},
        str(new_agent_dir / "data" / "home-config"): {"bind": "/home/claw/.captain-claw", "mode": "rw"},
        str(new_agent_dir / "data" / "workspace"): {"bind": "/data/workspace", "mode": "rw"},
        str(new_agent_dir / "data" / "sessions"): {"bind": "/data/sessions", "mode": "rw"},
        str(new_agent_dir / "data" / "skills"): {"bind": "/data/skills", "mode": "rw"},
    }
    # Carry over any extra volumes that weren't part of the agent data dir
    mounts = c.attrs.get("Mounts", [])
    known_dests = {"/app/config.yaml", "/app/.env", "/home/claw/.captain-claw",
                   "/data/workspace", "/data/sessions", "/data/skills"}
    for m in mounts:
        src = m.get("Source", "")
        dst = m.get("Destination", "")
        mode = m.get("Mode", "rw")
        if src and dst and dst not in known_dests:
            volumes[src] = {"bind": dst, "mode": mode}

    # Read host config
    host_config = c.attrs.get("HostConfig", {})
    restart_policy = host_config.get("RestartPolicy", {"Name": "unless-stopped"})
    security_opt = host_config.get("SecurityOpt", ["no-new-privileges:true", "seccomp:unconfined"])
    cap_drop = host_config.get("CapDrop", ["ALL"])
    cap_add = host_config.get("CapAdd", ["CHOWN", "SETUID", "SETGID", "SYS_CHROOT"])
    tmpfs = host_config.get("Tmpfs", {"/tmp": "", "/run": ""})
    network_mode = host_config.get("NetworkMode", "host")

    # Find first available port starting from original
    new_web_port: int | None = None
    if old_web_port:
        new_web_port = _find_available_port(old_web_port + 1)

    # Update labels for the clone
    labels["flight-deck.agent-name"] = new_name
    labels["flight-deck.description"] = ""
    if new_web_port:
        labels["flight-deck.web-port"] = str(new_web_port)

    # Update config.yaml with new web port and botport instance name
    cfg_path = new_agent_dir / "config.yaml"
    if cfg_path.is_file() and new_web_port and old_web_port:
        cfg_text = cfg_path.read_text()
        cfg_text = cfg_text.replace(f"port: {old_web_port}", f"port: {new_web_port}")
        # Update botport instance name
        if old_agent_name:
            cfg_text = cfg_text.replace(f"instance_name: {old_agent_name}", f"instance_name: {new_name}")
            cfg_text = cfg_text.replace(f"instance_name: '{old_agent_name}'", f"instance_name: '{new_name}'")
        cfg_path.write_text(cfg_text)
        # Also update home-config copy
        hc_path = new_agent_dir / "data" / "home-config" / "config.yaml"
        if hc_path.is_file():
            hc_path.write_text(cfg_text)

    # Make sure .env file exists (even if empty) so Docker doesn't create a directory
    env_path = new_agent_dir / ".env"
    if not env_path.is_file():
        env_path.write_text("")

    hostname = new_slug

    # Port publishing
    ports: dict[str, int] = {}
    network_mode_use: str | None = network_mode
    if platform.system() == "Darwin" and network_mode == "host":
        network_mode_use = None
        if new_web_port:
            ports[f"{new_web_port}/tcp"] = new_web_port
    elif network_mode != "host":
        if new_web_port:
            ports[f"{new_web_port}/tcp"] = new_web_port

    try:
        new_container = client.containers.run(
            image=image,
            name=new_slug,
            hostname=hostname,
            detach=True,
            network_mode=network_mode_use,
            ports=ports or None,
            volumes=volumes,
            environment=environment,
            labels=labels,
            security_opt=security_opt,
            cap_drop=cap_drop,
            cap_add=cap_add,
            tmpfs=tmpfs,
            restart_policy=restart_policy,
            stop_signal="SIGTERM",
        )
        return ContainerActionResult(
            ok=True,
            container_id=new_container.short_id,
            message=f"Agent '{new_name}' cloned from '{old_agent_name}' (port {new_web_port})",
        )
    except docker.errors.ImageNotFound:
        raise HTTPException(404, f"Docker image '{image}' not found.")
    except docker.errors.APIError as exc:
        raise HTTPException(500, f"Docker error: {exc.explanation or str(exc)}")


@app.get("/fd/containers/{container_id}/logs")
async def container_logs(container_id: str, tail: int = 200, since_ts: float = 0, follow: bool = False, request: Request = None, user: dict | None = _required_user_dep):
    """Fetch container logs.

    When *since_ts* > 0 the Docker ``since`` parameter is used so only
    new log lines written after that Unix timestamp are returned
    (incremental fetch).  The response includes ``timestamp`` – the
    current time – for the frontend to pass back on the next poll.
    """
    import time
    c = _find_container(container_id, getattr(request.state, "user_id", ""))
    now = time.time()
    if follow:
        def stream():
            for chunk in c.logs(stream=True, follow=True, tail=tail):
                yield chunk
        return StreamingResponse(stream(), media_type="text/plain")
    elif since_ts > 0:
        logs = c.logs(tail=0, since=since_ts, timestamps=False).decode("utf-8", errors="replace")
        return {"logs": logs, "timestamp": now}
    else:
        logs = c.logs(tail=tail).decode("utf-8", errors="replace")
        return {"logs": logs, "timestamp": now}


# ── Agent config editing ──

class AgentConfigUpdate(BaseModel):
    config_yaml: str | None = None
    env: str | None = None


def _resolve_agent_dir(identifier: str, kind: str, user_id: str) -> Path:
    """Resolve the on-disk data directory for a container or process agent."""
    if kind == "docker":
        c = _find_container(identifier, user_id)
        slug = c.name
    else:
        entry = _verify_process_owner(identifier, user_id)
        slug = identifier
    agent_dir = DATA_DIR / slug
    if not agent_dir.is_dir():
        raise HTTPException(404, f"Agent data directory not found for '{slug}'")
    return agent_dir


@app.get("/fd/agent-config/{kind}/{identifier}")
async def get_agent_config(
    kind: str, identifier: str, request: Request,
    user: dict | None = _required_user_dep,
):
    """Read an agent's config.yaml and .env files."""
    if kind not in ("docker", "process"):
        raise HTTPException(400, "kind must be 'docker' or 'process'")
    user_id = getattr(request.state, "user_id", "")
    agent_dir = _resolve_agent_dir(identifier, kind, user_id)

    config_yaml = ""
    env = ""
    config_file = agent_dir / "config.yaml"
    env_file = agent_dir / ".env"
    if config_file.is_file():
        config_yaml = config_file.read_text()
    if env_file.is_file():
        env = env_file.read_text()
    return {"config_yaml": config_yaml, "env": env}


@app.put("/fd/agent-config/{kind}/{identifier}")
async def update_agent_config(
    kind: str, identifier: str, body: AgentConfigUpdate, request: Request,
    user: dict | None = _required_user_dep,
):
    """Update an agent's config.yaml and/or .env files. Agent must be restarted for changes to take effect."""
    if kind not in ("docker", "process"):
        raise HTTPException(400, "kind must be 'docker' or 'process'")
    user_id = getattr(request.state, "user_id", "")
    agent_dir = _resolve_agent_dir(identifier, kind, user_id)

    updated = []
    if body.config_yaml is not None:
        config_file = agent_dir / "config.yaml"
        config_file.write_text(body.config_yaml)
        # Also update home-config copy so it takes precedence on restart
        home_config = agent_dir / "data" / "home-config" / "config.yaml"
        if home_config.parent.is_dir():
            home_config.write_text(body.config_yaml)
        # Also update home-config-parent/.captain-claw copy
        parent_config = agent_dir / "data" / "home-config-parent" / ".captain-claw" / "config.yaml"
        if parent_config.parent.is_dir():
            parent_config.write_text(body.config_yaml)
        updated.append("config.yaml")
    if body.env is not None:
        env_file = agent_dir / ".env"
        env_file.write_text(body.env)
        updated.append(".env")

    return {"ok": True, "updated": updated, "message": "Restart the agent for changes to take effect."}


class AgentModelUpdate(BaseModel):
    provider: str
    model: str
    api_key: str | None = None


@app.put("/fd/agent-model/{kind}/{identifier}")
async def update_agent_model(
    kind: str, identifier: str, body: AgentModelUpdate, request: Request,
    user: dict | None = _required_user_dep,
):
    """Quick-update an agent's provider, model, and optionally api_key in all config locations."""
    if kind not in ("docker", "process"):
        raise HTTPException(400, "kind must be 'docker' or 'process'")
    user_id = getattr(request.state, "user_id", "")
    agent_dir = _resolve_agent_dir(identifier, kind, user_id)

    # Gather all config file paths that may exist
    config_paths = [
        agent_dir / "config.yaml",
        agent_dir / "data" / "home-config" / "config.yaml",
        agent_dir / "data" / "home-config-parent" / ".captain-claw" / "config.yaml",
    ]

    updated_count = 0
    for cfg_path in config_paths:
        if not cfg_path.is_file():
            continue
        try:
            data = yaml.safe_load(cfg_path.read_text()) or {}
        except Exception:
            data = {}
        if not isinstance(data, dict):
            data = {}
        if "model" not in data or not isinstance(data.get("model"), dict):
            data["model"] = {}
        data["model"]["provider"] = body.provider
        data["model"]["model"] = body.model
        if body.api_key is not None:
            data["model"]["api_key"] = body.api_key
        cfg_path.write_text(yaml.dump(data, default_flow_style=False, sort_keys=False, allow_unicode=True))
        updated_count += 1

    return {"ok": True, "updated": updated_count, "message": "Restart the agent for changes to take effect."}


class AgentModeUpdate(BaseModel):
    mode: str = "neutra"


@app.put("/fd/agent-mode/{kind}/{identifier}")
async def update_agent_mode(
    kind: str, identifier: str, body: AgentModeUpdate, request: Request,
    user: dict | None = _required_user_dep,
):
    """Update an agent's cognitive mode at runtime (no restart needed).

    Writes the mode to cognitive_mode.txt — the agent picks it up
    on the next system prompt build via mtime-based cache invalidation.
    """
    if kind not in ("docker", "process"):
        raise HTTPException(400, "kind must be 'docker' or 'process'")

    # Validate mode name.
    from captain_claw.cognitive_mode import MODES
    mode_name = body.mode.lower().strip()
    if mode_name not in MODES:
        raise HTTPException(400, f"Unknown cognitive mode: {mode_name!r}. Valid: {', '.join(MODES)}")

    user_id = getattr(request.state, "user_id", "")
    agent_dir = _resolve_agent_dir(identifier, kind, user_id)

    # Write to all potential home config locations.
    for subdir in ("home-config", "home-config-parent"):
        cc_dir = agent_dir / "data" / subdir / ".captain-claw"
        if cc_dir.is_dir():
            mode_file = cc_dir / "cognitive_mode.txt"
            if mode_name == "neutra":
                # Remove file for neutra (default no-op).
                mode_file.unlink(missing_ok=True)
            else:
                mode_file.write_text(mode_name, encoding="utf-8")

    return {"ok": True, "mode": mode_name, "message": "Mode updated. Takes effect on the agent's next response."}


class AgentEcoModeUpdate(BaseModel):
    enabled: bool = False


@app.put("/fd/agent-eco-mode/{kind}/{identifier}")
async def update_agent_eco_mode(
    kind: str, identifier: str, body: AgentEcoModeUpdate, request: Request,
    user: dict | None = _required_user_dep,
):
    """Toggle eco mode (micro instructions + lazy tools) at runtime.

    Writes ``eco_mode.txt`` — the agent picks it up on the next system
    prompt build, just like cognitive mode.
    """
    if kind not in ("docker", "process"):
        raise HTTPException(400, "kind must be 'docker' or 'process'")

    user_id = getattr(request.state, "user_id", "")
    agent_dir = _resolve_agent_dir(identifier, kind, user_id)

    for subdir in ("home-config", "home-config-parent"):
        cc_dir = agent_dir / "data" / subdir / ".captain-claw"
        if cc_dir.is_dir():
            eco_file = cc_dir / "eco_mode.txt"
            if body.enabled:
                eco_file.write_text("on", encoding="utf-8")
            else:
                eco_file.unlink(missing_ok=True)

    return {"ok": True, "enabled": body.enabled, "message": "Eco mode updated. Takes effect on the agent's next response."}


@app.get("/fd/agent-eco-mode/{kind}/{identifier}")
async def get_agent_eco_mode(
    kind: str, identifier: str, request: Request,
    user: dict | None = _required_user_dep,
):
    """Read current eco mode state for an agent."""
    if kind not in ("docker", "process"):
        raise HTTPException(400, "kind must be 'docker' or 'process'")

    user_id = getattr(request.state, "user_id", "")
    agent_dir = _resolve_agent_dir(identifier, kind, user_id)

    for subdir in ("home-config", "home-config-parent"):
        eco_file = agent_dir / "data" / subdir / ".captain-claw" / "eco_mode.txt"
        if eco_file.is_file():
            return {"enabled": True}

    return {"enabled": False}


def _write_eco_flag_on_spawn(agent_dir: Path, enabled: bool = True) -> None:
    """Write the ``eco_mode.txt`` flag at spawn time so a new agent deploys
    in eco mode (micro instructions + lazy tools) from its very first response.

    Writes to every location an agent might read it from, covering both runtimes:

      * Process agents run with ``HOME=data/home-config-parent`` and read
        ``~/.captain-claw/eco_mode.txt`` → ``home-config-parent/.captain-claw/``.
      * Docker agents bind-mount ``data/home-config`` as ``/home/claw/.captain-claw``,
        so the agent reads ``home-config/eco_mode.txt`` directly.

    The nested ``.captain-claw/eco_mode.txt`` files also match what the
    ``/fd/agent-eco-mode`` GET endpoint reads, so the Flight Deck UI shows
    the toggle as ON immediately after spawn.
    """
    if not enabled:
        return
    # Agent-read locations + FD UI state locations.
    targets = [
        agent_dir / "data" / "home-config-parent" / ".captain-claw" / "eco_mode.txt",
        agent_dir / "data" / "home-config" / ".captain-claw" / "eco_mode.txt",
        # Docker: home-config IS the container's ~/.captain-claw.
        agent_dir / "data" / "home-config" / "eco_mode.txt",
    ]
    for target in targets:
        try:
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_text("on", encoding="utf-8")
        except Exception as exc:  # best-effort; never fail a spawn over this
            log.warning("Failed to write eco flag on spawn", path=str(target), error=str(exc))


class AgentNanoModeUpdate(BaseModel):
    enabled: bool = False


@app.put("/fd/agent-nano-mode/{kind}/{identifier}")
async def update_agent_nano_mode(
    kind: str, identifier: str, body: AgentNanoModeUpdate, request: Request,
    user: dict | None = _required_user_dep,
):
    """Toggle nano (barebone) mode at runtime.

    Writes ``nano_mode.txt`` — the agent picks it up on the next system
    prompt build.  Nano implies eco/micro: tool definitions are stripped
    to a barebone allowlist and prompts use nano_<name> templates.
    """
    if kind not in ("docker", "process"):
        raise HTTPException(400, "kind must be 'docker' or 'process'")

    user_id = getattr(request.state, "user_id", "")
    agent_dir = _resolve_agent_dir(identifier, kind, user_id)

    for subdir in ("home-config", "home-config-parent"):
        cc_dir = agent_dir / "data" / subdir / ".captain-claw"
        if cc_dir.is_dir():
            nano_file = cc_dir / "nano_mode.txt"
            if body.enabled:
                nano_file.write_text("on", encoding="utf-8")
            else:
                nano_file.unlink(missing_ok=True)

    return {"ok": True, "enabled": body.enabled, "message": "Nano mode updated. Takes effect on the agent's next response."}


@app.get("/fd/agent-nano-mode/{kind}/{identifier}")
async def get_agent_nano_mode(
    kind: str, identifier: str, request: Request,
    user: dict | None = _required_user_dep,
):
    """Read current nano mode state for an agent."""
    if kind not in ("docker", "process"):
        raise HTTPException(400, "kind must be 'docker' or 'process'")

    user_id = getattr(request.state, "user_id", "")
    agent_dir = _resolve_agent_dir(identifier, kind, user_id)

    for subdir in ("home-config", "home-config-parent"):
        nano_file = agent_dir / "data" / subdir / ".captain-claw" / "nano_mode.txt"
        if nano_file.is_file():
            return {"enabled": True}

    return {"enabled": False}


@app.get("/fd/cognitive-modes")
async def list_cognitive_modes():
    """Return all available cognitive modes for UI dropdowns."""
    from captain_claw.cognitive_mode import list_modes, mode_to_dict
    return {"modes": [mode_to_dict(m) for m in list_modes()]}


@app.get("/fd/containers/{container_id}")
async def get_container(container_id: str, request: Request, user: dict | None = _required_user_dep):
    c = _find_container(container_id, getattr(request.state, "user_id", ""))
    info = _container_info(c)
    # Add extra details
    return {
        **info.model_dump(),
        "labels": c.labels,
        "env": c.attrs.get("Config", {}).get("Env", []),
        "mounts": [
            {"source": m.get("Source", ""), "destination": m.get("Destination", ""), "mode": m.get("Mode", "")}
            for m in c.attrs.get("Mounts", [])
        ],
    }


@app.websocket("/fd/agent-ws/{host}/{port}")
async def agent_ws_proxy(ws: WebSocket, host: str, port: int, token: str = ""):
    """Proxy WebSocket to a CC agent — avoids browser CORS restrictions."""
    import websockets

    await ws.accept()
    # Auto-resolve auth token if the caller didn't provide one
    auth = token or _resolve_agent_auth(port)
    params = f"?token={auth}" if auth else ""
    agent_url = f"ws://{host}:{port}/ws{params}"

    try:
        # ping_interval/ping_timeout keep the upstream link alive through any
        # intermediaries and surface dead peers fast — without these the proxy
        # silently drops after a few minutes of idle, forcing the user to
        # re-click the chat button in FD.
        async with websockets.connect(
            agent_url,
            max_size=4 * 1024 * 1024,
            ping_interval=20,
            ping_timeout=10,
        ) as agent_ws:
            async def client_to_agent():
                try:
                    while True:
                        data = await ws.receive_text()
                        await agent_ws.send(data)
                except WebSocketDisconnect:
                    pass

            async def agent_to_client():
                try:
                    async for msg in agent_ws:
                        await ws.send_text(msg if isinstance(msg, str) else msg.decode())
                except Exception:
                    pass

            done, pending = await asyncio.wait(
                [asyncio.create_task(client_to_agent()), asyncio.create_task(agent_to_client())],
                return_when=asyncio.FIRST_COMPLETED,
            )
            for t in pending:
                t.cancel()
    except Exception as exc:
        try:
            await ws.send_text(f'{{"type":"error","message":"Connection failed: {exc}"}}')
            await ws.close()
        except Exception:
            pass


@app.get("/fd/probe")
async def probe_agent(host: str = "localhost", port: int = 23080):
    """Probe a CC agent's web server (server-side, avoids CORS)."""
    import httpx
    url = f"http://{host}:{port}/"
    try:
        async with httpx.AsyncClient(timeout=3.0) as client:
            resp = await client.get(url)
            return {"ok": resp.status_code < 500, "status": resp.status_code}
    except Exception:
        return {"ok": False, "status": 0}


class FleetAgent(BaseModel):
    name: str
    slug: str = ""  # FD_AGENT_SLUG — what the agent identifies itself as in X-Agent-Slug
    kind: str  # docker | process | local
    host: str = "localhost"
    port: int
    status: str
    description: str = ""


@app.get("/fd/fleet", response_model=list[FleetAgent])
async def get_fleet(request: Request, user: dict | None = _optional_user_dep):
    """Return all running/known agents across docker, process, and local stores."""
    fleet: list[FleetAgent] = []
    user_id = getattr(request.state, "user_id", "")

    # Docker containers
    try:
        client = get_docker()
        for c in client.containers.list(all=True, filters={"label": CONTAINER_LABEL}):
            labels = c.labels or {}
            if AUTH_ENABLED and user_id and labels.get(OWNER_LABEL, "") != user_id:
                continue
            wp = labels.get("flight-deck.web-port", "")
            _agent_name = labels.get("flight-deck.agent-name", c.name)
            fleet.append(FleetAgent(
                name=_agent_name,
                slug=_slug(_agent_name),
                kind="docker",
                host="localhost",
                port=int(wp) if wp else 0,
                status=c.status,
                description=labels.get("flight-deck.description", ""),
            ))
    except Exception:
        pass

    # Process agents
    registry = _load_process_registry()
    for slug, entry in registry.items():
        if AUTH_ENABLED and user_id and entry.get("owner", "") != user_id:
            continue
        alive = _process_is_alive(slug)
        fleet.append(FleetAgent(
            name=entry.get("name", slug),
            slug=slug,
            kind="process",
            host="localhost",
            port=entry.get("web_port", 0),
            status="running" if alive else "stopped",
            description=entry.get("description", ""),
        ))

    return fleet


def _resolve_agent_auth(port: int) -> str:
    """Look up the auth token for an agent by its web port from Docker labels or process registry."""
    # Check Docker containers
    try:
        client = get_docker()
        for c in client.containers.list(filters={"label": CONTAINER_LABEL}):
            labels = c.labels or {}
            wp = labels.get("flight-deck.web-port", "")
            if wp and int(wp) == port:
                return labels.get("flight-deck.web-auth", "")
    except Exception:
        pass

    # Check process registry. Multiple (stale) entries can share a web_port, so
    # prefer the one whose process is actually alive before falling back to any.
    registry = _load_process_registry()
    matches = [(slug, e) for slug, e in registry.items() if e.get("web_port") == port]
    for slug, entry in matches:
        if _process_is_alive(slug):
            return entry.get("web_auth", "")
    if matches:
        return matches[0][1].get("web_auth", "")

    return ""


# ── Flow engine helpers ────────────────────────────────────────────────

def _running_agents() -> list[dict[str, Any]]:
    """Lightweight pool snapshot for the FlowRunner's agent selector."""
    out: list[dict[str, Any]] = []
    for slug, entry in _load_process_registry().items():
        out.append({
            "name": entry.get("name", slug),
            "host": "localhost",
            "port": entry.get("web_port", 0),
            "status": "running" if _process_is_alive(slug) else "stopped",
            "description": entry.get("description", ""),
            # Carry the auth from the SAME entry so the consult uses a token that
            # matches THIS agent — port-keyed re-resolution can collide when stale
            # entries share a port.
            "auth": entry.get("web_auth", ""),
        })
    return out


def _fd_internal_tools() -> dict[str, Any]:
    """FD-side tools the FlowRunner can call directly (no agent), e.g. face_identify."""
    async def _face_identify(args: dict[str, Any]) -> str:
        """Recognize faces in an image, in-process in Flight Deck.

        ``args.image`` is a filesystem path the FD process can read — use the
        ``{{trigger.fd_image_path}}`` Flow variable (FD-local copy of the
        inbound photo), NOT ``{{trigger.image_path}}`` (which points at the
        agent host). Returns JSON the rest of the Flow can branch on:
          {name, person_id, confident, confidence, count, card}
        On any failure returns {error: "..."} with confident=false so a
        ``branch`` step still has a defined shape to test.
        """
        image = str(args.get("image") or "").strip()
        if not image:
            return json.dumps({"confident": False, "name": None, "error": "no image path provided"})
        try:
            from pathlib import Path as _Path

            p = _Path(image).expanduser()
            if not p.is_file():
                return json.dumps({
                    "confident": False, "name": None,
                    "error": f"image not readable by Flight Deck: {image} "
                             "(use {{trigger.fd_image_path}}, not {{trigger.image_path}})",
                })
            blob = p.read_bytes()

            from captain_claw.flight_deck import face_index  # type: ignore
            result = await face_index.get_index().recognize(image_blob=blob, channel="flow")
            return json.dumps({
                "confident": bool(result.name is not None),
                "name": result.name,
                "person_id": result.person_id,
                "confidence": round(float(result.confidence), 4),
                "count": len(result.faces),
                "card": result.card_markdown,
            })
        except RuntimeError as exc:
            # Missing 'faces' extra (insightface/sqlite-vec) surfaces here.
            return json.dumps({"confident": False, "name": None, "error": str(exc)})
        except Exception as exc:
            return json.dumps({"confident": False, "name": None, "error": f"face_identify failed: {exc}"})
    return {"face_identify": _face_identify}


async def _flow_whatsapp_send(waid: str, text: str) -> None:
    try:
        from captain_claw.flight_deck.whatsapp_bridge import _send_whatsapp_text
        await _send_whatsapp_text(waid, text, mirror=True)
    except Exception as exc:
        log.warning("flow whatsapp send failed: %s", exc)


# ── Flow engine API (/fd/flows) ────────────────────────────────────────

def _flow_store():
    store = getattr(app.state, "flow_store", None)
    if store is None:
        raise HTTPException(503, "Flow engine not ready")
    return store


@app.get("/fd/flows")
async def fd_flows_list(request: Request, user: dict | None = _optional_user_dep):
    return {"flows": await _flow_store().list_flows()}


@app.post("/fd/flows")
async def fd_flows_create(request: Request, user: dict | None = _optional_user_dep):
    spec = await request.json()
    fid = await _flow_store().create_flow(spec)
    return {"id": fid}


@app.get("/fd/flows/runs/{run_id}")
async def fd_flows_run_detail(run_id: str, request: Request, user: dict | None = _optional_user_dep):
    detail = await _flow_store().get_run(run_id)
    if not detail:
        raise HTTPException(404, "run not found")
    return detail


@app.post("/fd/flows/evaluate")
async def fd_flows_evaluate(request: Request, user: dict | None = _optional_user_dep):
    """Agent-handled channels (web/glasses) call this before a turn: classify the
    inbound, and if a Flow matches, run it and return its output to relay."""
    from captain_claw.flight_deck import flow_router
    if not flow_router.engine_ready():
        return {"matched": False}
    raw = await request.json()
    payload = flow_router.classify_payload(
        channel=str(raw.get("channel") or "web"),
        text=str(raw.get("text") or ""),
        mime=str(raw.get("mime") or ""),
        image_path=str(raw.get("image_path") or ""),
        video_path=str(raw.get("video_path") or ""),
        audio_path=str(raw.get("audio_path") or ""),
        waid=str(raw.get("waid") or ""),
        origin_host=str(raw.get("origin_host") or "localhost"),
        origin_port=int(raw.get("origin_port") or 0),
        origin_name=str(raw.get("origin_name") or ""),
    )
    # 0. Flow control command ('/flow stop|pause|resume', slash optional) —
    #    intercept before treating the message as input or a new trigger.
    if await flow_router.maybe_handle_flow_command(payload):
        return {"matched": True, "deferred": True}

    # 0b. Explicit start: '/flow run|start <name|id>' — launch a specific flow
    #     BOUND to this channel/origin (same path as a trigger match) so that
    #     /flow status|stop and input-resume all target it. Works for disabled
    #     flows too (enable/disable only gates automatic trigger firing).
    import re as _re
    _run_m = _re.match(r"^\s*/?flow\s+(?:run|start|launch)\s+(.+)$", str(payload.get("text") or ""), _re.I)
    if _run_m:
        _target = _run_m.group(1).strip()
        _store = flow_router._STORE
        _flow = None
        if _store is not None:
            _flow = await _store.get_flow(_target) or await _store.get_flow_by_name(_target)
        if _flow is None:
            return {"matched": True, "output": f"No flow named “{_target}”."}
        if flow_router._flow_needs_async(_flow):
            asyncio.create_task(flow_router._bg_run(_flow, payload))
            return {"matched": True, "flow": _flow.get("name"), "deferred": True}
        _result = await app.state.flow_runner.run(_flow, payload)
        return {
            "matched": True, "flow": _flow.get("name"),
            "run_id": _result.get("run_id"), "output": _result.get("output") or "",
        }

    # 1. Resume a paused flow first: if one is waiting on an `input` step for
    #    this channel+agent, this message is the reply — feed it and stop (the
    #    flow continues in the background and delivers via the channel).
    if flow_router.deliver_pending_input(
        waid=payload.get("waid", ""), channel=payload.get("channel", ""),
        origin_port=int(payload.get("origin_port") or 0), text=payload.get("text", ""),
    ):
        return {"matched": True, "deferred": True}

    flow = await flow_router.match_flow(payload)
    if not flow:
        return {"matched": False}

    # 2. Flows that can pause for input or consult the ORIGIN agent must run
    #    detached: the origin agent is the one blocked on THIS evaluate call, so
    #    a synchronous origin-consult would deadlock it. Background it and let it
    #    deliver via the channel (agent chat-push); the agent ends its turn now.
    if flow_router._flow_needs_async(flow):
        asyncio.create_task(flow_router._bg_run(flow, payload))
        return {"matched": True, "flow": flow.get("name"), "deferred": True}

    # 3. Simple flow → run inline and relay its output (unchanged behaviour).
    result = await app.state.flow_runner.run(flow, payload)
    return {
        "matched": True, "flow": flow.get("name"),
        "run_id": result.get("run_id"), "output": result.get("output") or "",
    }


# ── Flow DSL: text <-> flow, and agent-assisted NL -> flow ──────────────
# Registered BEFORE /fd/flows/{flow_id} so 'dsl'/'compile' aren't read as ids.

@app.get("/fd/flows/docs")
async def fd_flows_docs(request: Request, user: dict | None = _optional_user_dep):
    """Serve the Flow language reference (FLOWS.md) for the in-app docs viewer."""
    from pathlib import Path as _Path
    here = _Path(__file__).resolve()
    candidates = [
        here.parent.parent.parent / "FLOWS.md",   # repo root (../../ from this file)
        here.parent.parent / "FLOWS.md",
        _Path.cwd() / "FLOWS.md",
    ]
    for p in candidates:
        try:
            if p.is_file():
                return {"ok": True, "markdown": p.read_text(encoding="utf-8")}
        except Exception:
            continue
    return {"ok": False, "markdown": "# Flow docs\n\nFLOWS.md was not found on the server."}


@app.post("/fd/flows/dsl/compile")
async def fd_flows_dsl_compile(request: Request, user: dict | None = _optional_user_dep):
    """Deterministic: DSL text → flow dict (with structured errors)."""
    from captain_claw.flight_deck import flow_dsl
    body = await request.json()
    try:
        flow = flow_dsl.compile_dsl(str(body.get("dsl") or ""))
        return {"ok": True, "flow": flow}
    except flow_dsl.DSLError as exc:
        return {"ok": False, "error": exc.msg, "line": exc.line}
    except Exception as exc:
        return {"ok": False, "error": str(exc), "line": 0}


@app.post("/fd/flows/dsl/decompile")
async def fd_flows_dsl_decompile(request: Request, user: dict | None = _optional_user_dep):
    """Deterministic: flow dict → DSL text (for the code view / round-trip)."""
    from captain_claw.flight_deck import flow_dsl
    body = await request.json()
    flow = body.get("flow") or {}
    try:
        return {"ok": True, "dsl": flow_dsl.decompile(flow)}
    except Exception as exc:
        return {"ok": False, "error": str(exc)}


async def _ai_compile_flow(text: str, agent: str = "", current: str = "") -> dict:
    """Agent-assisted: free text / loose code → canonical DSL (via a pooled
    agent's model) → deterministic compile + validate. The model's output is
    always run through the real parser, so invalid output is rejected. Shared by
    /fd/flows/compile and /fd/flows/synthesize."""
    import re
    from captain_claw.flight_deck import flow_dsl
    text = str(text or "").strip()
    if not text:
        return {"ok": False, "error": "no text to compile"}
    current = str(current or "").strip()
    body = {"agent": agent}

    # Pick a RUNNING agent with a live web port to host the model call. Try
    # several (a registry "running" flag can lag a just-died web server).
    running = [a for a in _running_agents()
               if str(a.get("status")) == "running" and int(a.get("port") or 0) > 0]
    if not running:
        return {"ok": False, "error": "no running agent available to compile with"}
    # The caller may pick which agent compiles; prefer it, fall back to others.
    chosen = str(body.get("agent") or "").strip()
    if chosen:
        picked = [a for a in running if str(a.get("name")) == chosen]
        agents = picked + [a for a in running if str(a.get("name")) != chosen]
    else:
        agents = running

    system = (
        "You translate a user's request into a Captain Claw Flow DSL program. "
        "Output ONLY the DSL — no prose, no code fences.\n"
        "Grammar:" + flow_dsl.__doc__.split("Grammar", 1)[1] + "\n"
        "STEP TYPE RULES (important):\n"
        "• agent — open-ended work the model decides how to do: search the web, "
        "research, look something up, summarize, answer. Use `agent on origin` "
        "with a `prompt:`. THIS is what you use for 'search for X', 'find', "
        "'research', 'look up'.\n"
        "• tool — a SINGLE named, deterministic tool. ONLY use it when you know "
        "the exact tool name; it needs `tool: <name>` and `arg <k>: <v>`. The "
        "only `on fd` tool is `face_identify`. NEVER emit a `tool` step without "
        "a `tool:` name — use an `agent` step instead.\n"
        "• vision — describe/read an image (`on capability:vision`, `image:`).\n"
        "• input — ask the user something and wait (`prompt:`). The reply is "
        "{{steps.<id>.output}}.\n"
        "• emit — send a message (`emit \"...\"`).\n"
        "• gosub — call ANOTHER flow as a subroutine and wait for it: "
        "`gosub \"Other Flow\"` with optional `with <k>: <v>` argument lines. The "
        "child's return value is {{calls.<step_id>.output}} and its status is "
        "{{calls.<step_id>.status}}. Use it to reuse an existing flow.\n"
        "• return — end the flow now and hand a value back to the caller: "
        "`return {{steps.x.output}}` (works inside a branch path too). A flow that "
        "is meant to be gosub'd should end with `output -> return`.\n"
        "• spawn — start another flow in the BACKGROUND (don't wait): "
        "`spawn \"Other Flow\"` with optional `with <k>: <v>`. Later `join <step_id>` "
        "to collect it. Use it to run things in parallel.\n"
        "• join — wait for a spawned flow: `join <spawn_step_id>` with optional "
        "`timeout: <seconds>`. Result is {{joins.<spawn_step_id>.output}} and "
        "status {{joins.<spawn_step_id>.status}} (done/error/timeout).\n"
        "• error — an error-handler step that reports a problem: "
        "`error \"It failed: {{error.message}}\"`. Reach it from a failing call's "
        "`on error -> <step>` line, or by branching on {{calls.<id>.status}}.\n"
        "On a gosub/join/spawn step, add `on error -> <step>` to jump to a handler "
        "if that call fails. Add `retry: <N>` to a gosub/join/spawn to re-try on "
        "failure before on-error.\n"
        "• set — compute a value into {{vars.<name>}}: `set total = {{vars.total}} "
        "+ 1`. The expression supports + - * / (+ also concatenates strings/lists), "
        "list literals [a, b], {{path}} operands, and functions split, join, len, "
        "upper, lower, trim, first, last, append, int, str, contains. e.g. "
        "`set cities = split({{steps.ask.output}}, \",\")` turns text into a list.\n"
        "• foreach — run a flow once per list item: a `foreach <var> in <list>` "
        "header, then a `gosub \"Flow\"` (sequential) OR `spawn \"Flow\"` (parallel) "
        "line, then `with <k>: <v>` args that use {{<var>}}. {{steps.<id>.output}} "
        "is the list of each result. Use it instead of repeating near-identical "
        "steps.\n"
        "• while — loop: `while <condition> -> <target>` jumps to <target> (whose "
        "path loops back) while the condition holds; else falls through. Pair with "
        "`set` for a counter.\n"
        "• sleep — pause the run: `sleep 30s` / `5m` / `2h` / `1d`.\n"
        "• wait — pause until an inbound message matches: `wait until contains "
        "\"approved\"` (the flow parks; other messages go to the agent). The "
        "matching text is {{steps.<id>.output}}. Good for approvals.\n"
        "Lists: any value can be a list (from set/split/foreach); when shown in a "
        "string it joins by newlines.\n"
        "WHEN TO USE WHAT: open-ended thinking → agent; iterate over many items → "
        "foreach; parallel work → spawn+join or foreach+spawn; loop with a counter "
        "→ set+while; reuse an existing flow → gosub; wait for the user → input "
        "(asks now) or wait (gates on a condition).\n"
        "TRIGGER: default to `trigger any` unless the user explicitly names a "
        "channel (whatsapp/web/glasses). Add `when <rules>` only if they gave a "
        "condition. Rules are joined with `and` (all must match) OR `or` (any "
        "matches) — pick ONE, don't mix. For 'any of these words' use `or`, e.g. "
        "`trigger any when contains \"hungry\" or contains \"gladan\" or contains "
        "\"gladni\"`. Write quotes plainly — never backslash-escape them.\n"
        "Selectors: origin, fd, any, capability:vision, name:<agent>. "
        "Template with {{trigger.text}}, {{trigger.image_path}}, "
        "{{trigger.fd_image_path}}, {{steps.<id>.output}}. Always include a "
        "trigger, at least one step, and an `output -> same` line."
    )
    if current:
        system += (
            "\n\nEDIT MODE: The user already has a flow (given below) and wants "
            "to MODIFY it. Apply their requested change to that existing program "
            "and output the COMPLETE updated DSL — keep every other step, the "
            "trigger, and the output line unchanged unless the request says "
            "otherwise. Preserve existing step ids; give any new step a fresh, "
            "descriptive snake_case id. Do not drop or reorder unrelated steps."
        )
    import httpx

    async def _complete(messages: list[dict[str, Any]]) -> tuple[str, str]:
        """Return (dsl, error). Tries running agents until one responds."""
        last = "no reachable agent"
        for agent in agents:
            host, port, token = agent["host"], int(agent["port"]), agent.get("auth", "")
            try:
                async with httpx.AsyncClient(timeout=120.0) as client:
                    r = await client.post(
                        f"http://{host}:{port}/api/llm/complete",
                        params={"token": token} if token else {},
                        json={"messages": messages, "temperature": 0.1},
                    )
                d = r.json() or {}
                if not d.get("ok"):
                    last = f"model error on {agent.get('name')}: {d.get('error') or r.status_code}"
                    continue
                content = str(d.get("content") or "").strip()
                content = re.sub(r"^```[a-zA-Z]*\n?|\n?```$", "", content).strip()
                return content, ""
            except Exception as exc:
                last = f"{agent.get('name')}: {exc}"
        return "", last

    if current:
        user_msg = (
            "Here is the current flow DSL:\n\n"
            f"{current}\n\n"
            f"Requested change: {text}\n\n"
            "Output ONLY the complete updated DSL."
        )
    else:
        user_msg = text
    messages: list[dict[str, Any]] = [
        {"role": "system", "content": system},
        {"role": "user", "content": user_msg},
    ]
    dsl, err = await _complete(messages)
    if not dsl:
        return {"ok": False, "error": f"compile call failed ({err})"}

    # Up to two attempts: if the generated DSL fails to compile, feed the error
    # back to the model once and let it repair its own output.
    for attempt in range(2):
        try:
            flow = flow_dsl.compile_dsl(dsl)
            return {"ok": True, "flow": flow, "dsl": dsl}
        except flow_dsl.DSLError as exc:
            problem = f"line {exc.line}: {exc.msg}"
        except Exception as exc:
            problem = str(exc)
        if attempt == 0:
            messages += [
                {"role": "assistant", "content": dsl},
                {"role": "user", "content": (
                    f"That DSL failed to compile ({problem}). Fix it and output "
                    "ONLY the corrected DSL. Remember: open-ended work like "
                    "searching is an `agent on origin` step with a prompt, never "
                    "a bare `tool` step."
                )},
            ]
            fixed, ferr = await _complete(messages)
            if fixed:
                dsl = fixed
                continue
        return {"ok": False, "error": f"generated DSL invalid ({problem})", "dsl": dsl}
    return {"ok": False, "error": "generated DSL invalid", "dsl": dsl}


@app.post("/fd/flows/compile")
async def fd_flows_compile(request: Request, user: dict | None = _optional_user_dep):
    """Agent-assisted NL/loose-code → validated flow (for the Code view)."""
    body = await request.json()
    return await _ai_compile_flow(
        str(body.get("text") or ""), str(body.get("agent") or ""), str(body.get("current") or ""),
    )


@app.post("/fd/flows/synthesize")
async def fd_flows_synthesize(request: Request, user: dict | None = _optional_user_dep):
    """Agent synthesis: a natural-language goal → a validated, **call-only**
    flow stored in the SCRATCH space (origin=agent). Dedups by canonical hash
    (retrieve-before-generate), optionally runs it, and returns its handle/name.

    Body: {goal, agent?, author?, run?(bool), payload?(dict)}."""
    from captain_claw.flight_deck import flow_dsl
    body = await request.json()
    goal = str(body.get("goal") or body.get("text") or "").strip()
    if not goal:
        return {"ok": False, "error": "no goal to synthesize"}
    author = str(body.get("author") or body.get("agent") or "").strip()
    store = _flow_store()

    # 1. Compile the goal to a validated flow via a pooled model.
    res = await _ai_compile_flow(goal, str(body.get("agent") or ""))
    if not res.get("ok"):
        return {"ok": False, "error": res.get("error") or "synthesis failed", "dsl": res.get("dsl")}
    flow = res["flow"]
    dsl = res.get("dsl") or flow_dsl.decompile(flow)
    flow["origin"] = "agent"

    # 2. Dedup: a structurally-identical scratch flow already exists → reuse it.
    #    A quarantined one is negative memory — don't re-create the same bad flow.
    h = flow_dsl.canonical_hash(flow)
    existing = await store.find_scratch_by_hash(h)
    if existing and existing.get("state") == "quarantined":
        return {"ok": False, "quarantined": True,
                "error": "This flow pattern failed repeatedly before (quarantined) — refine the goal."}
    if existing:
        fid, name, reused = existing["id"], existing["name"], True
    else:
        fid = await store.create_scratch_flow(flow, author=author, dsl_hash=h)
        name, reused = flow.get("name") or "Synthesized flow", False

    run = bool(body.get("run"))
    # A selection without a run counts as a use; a run records its own outcome.
    if reused and not run:
        await store.bump_use(fid)

    out: dict = {"ok": True, "flow_id": fid, "name": name, "reused": reused, "dsl": dsl}

    # 3. Optionally run it now (this records the outcome → drives promotion/quarantine).
    if run:
        target = await store.get_flow(fid)
        result = await app.state.flow_runner.run(target, body.get("payload") or {})
        out["run_id"] = result.get("run_id")
        out["status"] = result.get("status")
        out["output"] = result.get("output") or ""
    return out


@app.get("/fd/flows/scratch")
async def fd_flows_scratch(request: Request, user: dict | None = _optional_user_dep):
    """List the scratch space (synthesized flows) with provenance + lifecycle.
    Self-maintains: reclassifies states and GCs expired flows on view."""
    store = _flow_store()
    try:
        await store.maintain_scratch()
    except Exception as exc:
        log.warning("scratch maintain failed: %s", exc)
    return {"flows": await store.list_scratch_flows()}


@app.post("/fd/flows/scratch/maintain")
async def fd_flows_scratch_maintain(request: Request, user: dict | None = _optional_user_dep):
    """Janitor: reclassify every scratch flow (candidate/quarantined) and GC the
    expired ones. Returns a summary. Safe to call on a schedule."""
    return {"ok": True, **(await _flow_store().maintain_scratch())}


@app.post("/fd/flows/{flow_id}/promote")
async def fd_flows_promote(flow_id: str, request: Request, user: dict | None = _optional_user_dep):
    """Promote a scratch flow into the permanent space (optionally rename)."""
    try:
        body = await request.json()
    except Exception:
        body = {}
    ok = await _flow_store().promote_flow(flow_id, name=str(body.get("name") or "") or None)
    if not ok:
        raise HTTPException(404, "scratch flow not found")
    return {"ok": True}


@app.get("/fd/flows/{flow_id}")
async def fd_flows_get(flow_id: str, request: Request, user: dict | None = _optional_user_dep):
    flow = await _flow_store().get_flow(flow_id)
    if not flow:
        raise HTTPException(404, "flow not found")
    return flow


@app.put("/fd/flows/{flow_id}")
async def fd_flows_update(flow_id: str, request: Request, user: dict | None = _optional_user_dep):
    spec = await request.json()
    ok = await _flow_store().update_flow(flow_id, spec)
    if not ok:
        raise HTTPException(404, "flow not found")
    return {"ok": True}


@app.delete("/fd/flows/{flow_id}")
async def fd_flows_delete(flow_id: str, request: Request, user: dict | None = _optional_user_dep):
    ok = await _flow_store().delete_flow(flow_id)
    return {"ok": ok}


@app.post("/fd/flows/{flow_id}/enable")
async def fd_flows_enable(flow_id: str, request: Request, user: dict | None = _optional_user_dep):
    body = await request.json()
    ok = await _flow_store().set_enabled(flow_id, bool(body.get("enabled", True)))
    if not ok:
        raise HTTPException(404, "flow not found")
    return {"ok": True}


@app.post("/fd/flows/{flow_id}/run")
async def fd_flows_run(flow_id: str, request: Request, user: dict | None = _optional_user_dep):
    store = _flow_store()
    flow = await store.get_flow(flow_id)
    if not flow:
        raise HTTPException(404, "flow not found")
    try:
        body = await request.json()
    except Exception:
        body = {}
    payload = body.get("payload") or {}
    run_id = await store.start_run(flow_id, flow.get("name", ""), payload)
    # Run in the background; the UI polls /fd/flows/runs/{run_id} for the log.
    asyncio.create_task(app.state.flow_runner.run(flow, payload, run_id=run_id))
    return {"run_id": run_id}


@app.post("/fd/flows/runs/{run_id}/pause")
async def fd_flows_run_pause(run_id: str, request: Request, user: dict | None = _optional_user_dep):
    from captain_claw.flight_deck import flow_runner
    ok = flow_runner.request_pause(run_id)
    if ok:
        try:
            await _flow_store().set_run_status(run_id, "paused")
        except Exception:
            pass
    return {"ok": ok, "status": "paused" if ok else "not_running"}


@app.post("/fd/flows/runs/{run_id}/resume")
async def fd_flows_run_resume(run_id: str, request: Request, user: dict | None = _optional_user_dep):
    from captain_claw.flight_deck import flow_runner
    ok = flow_runner.request_resume(run_id)
    if ok:
        try:
            await _flow_store().set_run_status(run_id, "running")
        except Exception:
            pass
    return {"ok": ok, "status": "running" if ok else "not_running"}


@app.post("/fd/flows/runs/{run_id}/stop")
async def fd_flows_run_stop(run_id: str, request: Request, user: dict | None = _optional_user_dep):
    try:
        body = await request.json()
    except Exception:
        body = {}
    message = str(body.get("message") or "")
    from captain_claw.flight_deck import flow_runner
    ok = flow_runner.request_stop(run_id, message)
    return {"ok": ok, "status": "stopping" if ok else "not_running"}


@app.post("/fd/flows/{flow_id}/test")
async def fd_flows_test(flow_id: str, request: Request, user: dict | None = _optional_user_dep):
    store = _flow_store()
    flow = await store.get_flow(flow_id)
    if not flow:
        raise HTTPException(404, "flow not found")
    try:
        body = await request.json()
    except Exception:
        body = {}
    result = await app.state.flow_runner.run(flow, body.get("payload") or {}, dry=True)
    return {"steps": result.get("steps", []), "status": result.get("status")}


@app.get("/fd/flows/{flow_id}/runs")
async def fd_flows_runs(flow_id: str, request: Request, user: dict | None = _optional_user_dep):
    return {"runs": await _flow_store().list_runs(flow_id)}


async def _notify_fleet_change(new_agent_name: str, new_agent_port: int, event: str = "joined", owner_id: str = ""):
    """Notify running agents about a fleet change. When auth is enabled, only notify agents owned by the same user."""
    import websockets as _ws
    import json as _json

    # Build list of running agent WebSocket endpoints (excluding the new one),
    # scoped to the same owner when auth is enabled.
    targets: list[tuple[str, int, str]] = []  # (host, port, auth)

    try:
        client = get_docker()
        for c in client.containers.list(filters={"label": CONTAINER_LABEL}):
            labels = c.labels or {}
            if AUTH_ENABLED and owner_id and labels.get(OWNER_LABEL, "") != owner_id:
                continue
            wp = labels.get("flight-deck.web-port", "")
            if wp and int(wp) != new_agent_port:
                targets.append(("localhost", int(wp), labels.get("flight-deck.web-auth", "")))
    except Exception:
        pass

    registry = _load_process_registry()
    for slug, entry in registry.items():
        if AUTH_ENABLED and owner_id and entry.get("owner", "") != owner_id:
            continue
        wp = entry.get("web_port", 0)
        if wp and wp != new_agent_port and _process_is_alive(slug):
            targets.append(("localhost", wp, entry.get("web_auth", "")))

    if not targets:
        return

    # Build fleet list scoped to the same owner
    fleet: list[dict] = []
    try:
        client = get_docker()
        for c in client.containers.list(all=True, filters={"label": CONTAINER_LABEL}):
            labels = c.labels or {}
            if AUTH_ENABLED and owner_id and labels.get(OWNER_LABEL, "") != owner_id:
                continue
            wp = labels.get("flight-deck.web-port", "")
            fleet.append({"name": labels.get("flight-deck.agent-name", c.name), "status": c.status, "port": int(wp) if wp else 0})
    except Exception:
        pass
    reg = _load_process_registry()
    for slug, entry in reg.items():
        if AUTH_ENABLED and owner_id and entry.get("owner", "") != owner_id:
            continue
        fleet.append({"name": entry.get("name", slug), "status": "running" if _process_is_alive(slug) else "stopped", "port": entry.get("web_port", 0)})

    notification = (
        f"[Flight Deck] Agent '{new_agent_name}' has {event} the fleet on port {new_agent_port}. "
        f"Current fleet: {', '.join(a['name'] + ' (' + a['status'] + ', :' + str(a['port']) + ')' for a in fleet)}"
    )

    async def _send_to(host: str, port: int, auth: str):
        params = f"?token={auth}" if auth else ""
        url = f"ws://{host}:{port}/ws{params}"
        try:
            async with _ws.connect(url, open_timeout=5, close_timeout=3) as ws:
                # Wait for welcome
                raw = await asyncio.wait_for(ws.recv(), timeout=5)
                welcome = _json.loads(raw)
                if welcome.get("type") != "welcome":
                    return
                # Skip replay
                while True:
                    raw = await asyncio.wait_for(ws.recv(), timeout=5)
                    msg = _json.loads(raw)
                    if msg.get("type") == "replay_done":
                        break
                # Send as notification (injected into session, no LLM response triggered)
                await ws.send(_json.dumps({"type": "notification", "content": notification}))
                # Wait briefly then disconnect
                try:
                    await asyncio.wait_for(ws.recv(), timeout=5)
                except asyncio.TimeoutError:
                    pass
        except Exception:
            pass  # Best-effort; don't fail spawn if notification fails

    # Fire all notifications concurrently
    await asyncio.gather(*[_send_to(h, p, a) for h, p, a in targets], return_exceptions=True)


_IMAGE_EXTS_TRANSFER = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp"}


async def _transfer_file_to_agent(host: str, port: int, abs_path: str) -> tuple[list[str], list[str]]:
    """Upload a sender-local file to the TARGET agent on the operator's behalf.

    Flight Deck holds every agent's web-auth token (``_resolve_agent_auth``), so
    it can authenticate the upload even though the sending agent can't. Reads the
    file from disk (same-host fleet) and uploads to the target's
    /api/image|file/upload. Returns (image_paths, file_paths) on the TARGET.
    """
    import httpx
    from pathlib import Path

    p = Path(abs_path)
    if not abs_path or not p.is_file():
        return [], []
    is_img = p.suffix.lower() in _IMAGE_EXTS_TRANSFER
    endpoint = "/api/image/upload" if is_img else "/api/file/upload"
    auth = _resolve_agent_auth(port)
    params = {"token": auth} if auth else {}
    try:
        blob = p.read_bytes()
        async with httpx.AsyncClient(timeout=120.0) as client:
            resp = await client.post(
                f"http://{host}:{port}{endpoint}", params=params,
                files={"file": (p.name, blob)},
            )
        if resp.status_code == 200:
            tgt = str((resp.json() or {}).get("path") or "")
            if tgt:
                return ([tgt], []) if is_img else ([], [tgt])
        else:
            log.warning("agent file transfer rejected", status=resp.status_code, body=resp.text[:200])
    except Exception as exc:
        log.warning("agent file transfer failed: %s", exc)
    return [], []


class ConsultPeerRequest(BaseModel):
    host: str = "localhost"
    port: int
    auth: str = ""
    message: str
    source_name: str = "another agent"
    timeout: float = Field(default=480.0, le=600.0)
    # Agent-to-agent file transfer. The sender passes a sender-local absolute
    # path in ``attach_path``; Flight Deck uploads it to the target (with the
    # target's resolved auth token) and forwards the resulting target-local
    # path into the chat payload. image_paths/file_paths may also be passed
    # directly when already uploaded.
    attach_path: str = ""
    image_paths: list[str] = Field(default_factory=list)
    file_paths: list[str] = Field(default_factory=list)
    # When set, the target must NOT evaluate Flow triggers for this message
    # (loop guard: a Flow's agent-step consult shouldn't re-trigger a Flow).
    no_flow: bool = False
    # Tools the target must NOT use for this consult (deterministic guardrail —
    # e.g. an image-describe step denies shell/scripts/read so the model uses
    # the attached image instead of operating on the path).
    deny_tools: list[str] = Field(default_factory=list)
    # When set, the target replies ONLY to this consult (does NOT broadcast its
    # reply to its own channels/UI). Prevents double-delivery when a flow step
    # runs on a channel-connected agent (e.g. the WhatsApp origin agent).
    no_broadcast: bool = False


# Track active consultations to prevent duplicate requests to the same target
_active_consults: dict[int, str] = {}  # target_port -> source_name
_active_delegates: set[tuple[int, int]] = set()  # (source_port, target_port) in-flight


@app.post("/fd/consult-peer")
async def consult_peer(req: ConsultPeerRequest, request: Request, user: dict | None = _optional_user_dep):
    """Send a message to a peer agent and stream back intermediate events + final response as ndjson."""
    import websockets
    import json

    # NB: a peer serves one consult at a time. Rather than reject when it's
    # busy (which made flows/agents fail or loop), we QUEUE: _event_stream waits
    # for the in-flight consult to finish, then acquires the lock.

    # Event types we forward as peer activity so the caller can show progress
    _FORWARD_TYPES = {"status", "thinking", "monitor", "tool_stream"}

    # Resolve auth token from Fleet Deck records if not provided by caller.
    auth = req.auth
    if not auth:
        auth = _resolve_agent_auth(req.port)

    params = f"?token={auth}" if auth else ""
    agent_url = f"ws://{req.host}:{req.port}/ws{params}"

    async def _event_stream():
        # Queue behind any in-flight consult to this agent (it serves one at a
        # time). Bounded by the request timeout. The check-then-set has no await
        # between, so only one waiter acquires per loop tick (no race).
        _waited = 0.0
        _cap = min(float(req.timeout), 180.0)
        while _active_consults.get(req.port) and _waited < _cap:
            if _waited == 0.0:
                yield json.dumps({"event": "status", "data": {"status": f"Agent on port {req.port} busy — queuing…"}}) + "\n"
            await asyncio.sleep(1.0)
            _waited += 1.0
        if _active_consults.get(req.port):
            yield json.dumps({"ok": False, "error": f"Agent on port {req.port} still busy after {int(_waited)}s — try again."}) + "\n"
            return
        _active_consults[req.port] = req.source_name
        try:
            async with websockets.connect(agent_url, max_size=4 * 1024 * 1024) as ws:
                # Wait for welcome
                welcome = json.loads(await asyncio.wait_for(ws.recv(), timeout=10))
                if welcome.get("type") != "welcome":
                    yield json.dumps({"ok": False, "error": "Unexpected handshake"}) + "\n"
                    return

                # Skip replay messages
                while True:
                    raw = await asyncio.wait_for(ws.recv(), timeout=10)
                    msg = json.loads(raw)
                    if msg.get("type") == "replay_done":
                        break
                    if msg.get("type") not in ("chat_message",) or not msg.get("replay"):
                        break

                # Transfer any attached file to the target (FD resolves its auth).
                _img, _fil = list(req.image_paths), list(req.file_paths)
                if req.attach_path:
                    _ti, _tf = await _transfer_file_to_agent(req.host, req.port, req.attach_path)
                    _img += _ti
                    _fil += _tf
                _chat_payload: dict[str, Any] = {"type": "chat", "content": req.message}
                if _img:
                    _chat_payload["image_paths"] = _img
                if _fil:
                    _chat_payload["file_paths"] = _fil
                if req.no_flow:
                    _chat_payload["no_flow"] = True
                if req.deny_tools:
                    _chat_payload["deny_tools"] = list(req.deny_tools)
                if req.no_broadcast:
                    _chat_payload["no_broadcast"] = True
                await ws.send(json.dumps(_chat_payload))

                # Stream events until we get the final assistant response
                response_parts: list[str] = []
                deadline = asyncio.get_event_loop().time() + req.timeout
                recv_interval = 15.0  # heartbeat every 15s of silence
                _busy_retries = 0     # peer is single-threaded; wait it out
                while True:
                    remaining = deadline - asyncio.get_event_loop().time()
                    if remaining <= 0:
                        if not response_parts:
                            yield json.dumps({"ok": False, "error": "Timed out waiting for response"}) + "\n"
                            return
                        break
                    try:
                        raw = await asyncio.wait_for(ws.recv(), timeout=min(remaining, recv_interval))
                    except asyncio.TimeoutError:
                        # No message in recv_interval — send heartbeat and keep waiting
                        elapsed = int(req.timeout - remaining)
                        yield json.dumps({"event": "heartbeat", "data": {"elapsed": elapsed, "timeout": int(req.timeout)}}) + "\n"
                        continue
                    msg = json.loads(raw)
                    msg_type = msg.get("type", "")

                    # Forward interesting intermediate events
                    if msg_type in _FORWARD_TYPES:
                        yield json.dumps({"event": msg_type, "data": msg}) + "\n"

                    if msg_type == "chat_message" and msg.get("role") == "assistant" and not msg.get("replay"):
                        content = msg.get("content", "")
                        if content:
                            response_parts.append(content)
                        break
                    elif msg_type == "error":
                        _err = str(msg.get("message", "Agent error"))
                        # Transient "busy" → wait and re-send on the same socket.
                        if _busy_retries < 8 and ("busy processing" in _err.lower() or "session is busy" in _err.lower()):
                            _busy_retries += 1
                            _wait = min(2 + _busy_retries * 2, 12)
                            yield json.dumps({"event": "status", "data": {"status": f"Peer busy, retrying ({_busy_retries})…"}}) + "\n"
                            await asyncio.sleep(_wait)
                            try:
                                await ws.send(json.dumps(_chat_payload))
                            except Exception:
                                yield json.dumps({"ok": False, "error": _err}) + "\n"
                                return
                            continue
                        yield json.dumps({"ok": False, "error": _err}) + "\n"
                        return

            yield json.dumps({
                "ok": True,
                "done": True,
                "response": "\n".join(response_parts) if response_parts else "(no response)",
            }) + "\n"
        except Exception as exc:
            yield json.dumps({"ok": False, "error": f"Connection failed: {exc}"}) + "\n"
        finally:
            _active_consults.pop(req.port, None)

    return StreamingResponse(_event_stream(), media_type="application/x-ndjson")


class DelegatePeerRequest(BaseModel):
    target_host: str = "localhost"
    target_port: int
    target_name: str = ""
    source_host: str = "localhost"
    source_port: int
    source_name: str = "another agent"
    message: str
    timeout: float = Field(default=600.0, le=1800.0)
    # Origin platform tracking — so results are delivered to the correct session
    origin_platform: str = "web"       # "web" or "telegram"
    origin_user_id: str = ""           # telegram user id
    origin_chat_id: int = 0            # telegram chat id
    # Agent-to-agent file transfer (FD uploads attach_path to the target).
    attach_path: str = ""
    image_paths: list[str] = Field(default_factory=list)
    file_paths: list[str] = Field(default_factory=list)


@app.post("/fd/delegate-peer")
async def delegate_peer(req: DelegatePeerRequest, request: Request, user: dict | None = _optional_user_dep):
    """Fire-and-forget: send a task to a peer agent. When the peer finishes, deliver the result back to the source agent as a chat message."""
    import websockets
    import json

    _FORWARD_TYPES = {"status", "thinking", "monitor", "tool_stream"}

    target_auth = _resolve_agent_auth(req.target_port)
    source_auth = _resolve_agent_auth(req.source_port)

    peer_display = req.target_name or f"agent@{req.target_port}"

    # Guard: an agent must not delegate to itself (e.g. a vision agent that
    # tried image_vision, failed, then "delegated" the image to its own name).
    if req.source_port and req.target_port and int(req.source_port) == int(req.target_port):
        return {
            "ok": False,
            "message": f"{peer_display} is the requesting agent itself — handle the task directly, do not delegate to yourself.",
        }

    # Guard: if this source already has a delegation in flight to this target,
    # don't pile on another (the model sometimes re-delegates the same task with
    # a reworded message, which dodges arg-based dedup and storms the peer).
    _deleg_key = (req.source_port, req.target_port)
    if _deleg_key in _active_delegates:
        return {
            "ok": True,
            "message": (
                f"A task is already being processed by {peer_display} for you — "
                f"waiting on that result. Do NOT delegate again; tell the user you're waiting."
            ),
        }

    async def _background():
        log.info("delegate_background: started", target=peer_display, source=req.source_name,
                 target_port=req.target_port, source_port=req.source_port)

        # Phase 1: send task to target agent and wait for response
        t_params = f"?token={target_auth}" if target_auth else ""
        target_url = f"ws://{req.target_host}:{req.target_port}/ws{t_params}"
        response_text = ""
        try:
            async with websockets.connect(target_url, max_size=4 * 1024 * 1024) as ws:
                welcome = json.loads(await asyncio.wait_for(ws.recv(), timeout=10))
                if welcome.get("type") != "welcome":
                    response_text = f"[Error] Unexpected handshake from {peer_display}"
                    log.error("delegate_background: bad handshake from target", target=peer_display)
                else:
                    # Skip replay
                    while True:
                        raw = await asyncio.wait_for(ws.recv(), timeout=10)
                        msg = json.loads(raw)
                        if msg.get("type") == "replay_done":
                            break
                        if msg.get("type") not in ("chat_message",) or not msg.get("replay"):
                            break

                    _img, _fil = list(req.image_paths), list(req.file_paths)
                    if req.attach_path:
                        _ti, _tf = await _transfer_file_to_agent(req.target_host, req.target_port, req.attach_path)
                        _img += _ti
                        _fil += _tf
                    _payload: dict[str, Any] = {"type": "chat", "content": req.message}
                    if _img:
                        _payload["image_paths"] = _img
                    if _fil:
                        _payload["file_paths"] = _fil
                    await ws.send(json.dumps(_payload))
                    log.info("delegate_background: task sent to target", target=peer_display)

                    # Wait for the final response
                    deadline = asyncio.get_event_loop().time() + req.timeout
                    recv_interval = 30.0
                    _busy_retries = 0  # peer is single-threaded; wait it out instead of erroring back
                    while True:
                        remaining = deadline - asyncio.get_event_loop().time()
                        if remaining <= 0:
                            response_text = f"[Timeout] {peer_display} did not finish within {int(req.timeout)}s"
                            log.warning("delegate_background: target timed out", target=peer_display, timeout=req.timeout)
                            break
                        try:
                            raw = await asyncio.wait_for(ws.recv(), timeout=min(remaining, recv_interval))
                        except asyncio.TimeoutError:
                            continue
                        msg = json.loads(raw)
                        msg_type = msg.get("type", "")
                        if msg_type == "chat_message" and msg.get("role") == "assistant" and not msg.get("replay"):
                            response_text = msg.get("content", "(no response)")
                            log.info("delegate_background: got response from target",
                                     target=peer_display, response_len=len(response_text))
                            break
                        elif msg_type == "error":
                            _err = str(msg.get("message", "Unknown error"))
                            # "Agent is busy" is transient (peer serves one request
                            # at a time). Wait and re-send on the SAME connection
                            # instead of bouncing a fake "result" back to the caller,
                            # which made the caller re-delegate in a loop.
                            if _busy_retries < 8 and ("busy processing" in _err.lower() or "session is busy" in _err.lower()):
                                _busy_retries += 1
                                _wait = min(2 + _busy_retries * 2, 12)
                                log.info("delegate_background: target busy, waiting then retrying",
                                         target=peer_display, attempt=_busy_retries, wait=_wait)
                                await asyncio.sleep(_wait)
                                try:
                                    await ws.send(json.dumps(_payload))
                                except Exception:
                                    response_text = f"[Error from {peer_display}] {_err}"
                                    break
                                continue
                            response_text = f"[Error from {peer_display}] {_err}"
                            log.error("delegate_background: target returned error", target=peer_display, error=_err)
                            break
        except Exception as exc:
            response_text = f"[Error] Could not reach {peer_display}: {exc}"
            log.error("delegate_background: phase 1 failed", target=peer_display, error=str(exc))

        if not response_text:
            response_text = "(no response received)"

        # Phase 2: deliver the result back to the source agent
        s_params = f"?token={source_auth}" if source_auth else ""
        source_url = f"ws://{req.source_host}:{req.source_port}/ws{s_params}"
        callback_msg = (
            f"[Delegated result from {peer_display}] This is the ANSWER you were "
            f"waiting for. Relay it to the user now, concisely, in their language. "
            f"Do NOT delegate again, do NOT call any tool, and do NOT say you are "
            f"still waiting — you already have the result below:\n\n{response_text}"
        )
        log.info("delegate_background: delivering result to source",
                 source=req.source_name, source_port=req.source_port, result_len=len(response_text))
        try:
            async with websockets.connect(source_url, open_timeout=10, close_timeout=5) as ws:
                welcome = json.loads(await asyncio.wait_for(ws.recv(), timeout=10))
                if welcome.get("type") != "welcome":
                    log.error("delegate_callback: unexpected handshake from source", source=req.source_name)
                    return
                # Skip replay
                while True:
                    raw = await asyncio.wait_for(ws.recv(), timeout=10)
                    msg = json.loads(raw)
                    if msg.get("type") == "replay_done":
                        break
                notification_payload = {
                    "type": "notification",
                    "content": callback_msg,
                    "trigger_response": True,
                }
                # Include origin platform info so the agent routes the result correctly
                if req.origin_platform and req.origin_platform != "web":
                    notification_payload["origin_platform"] = req.origin_platform
                    notification_payload["origin_user_id"] = req.origin_user_id
                    notification_payload["origin_chat_id"] = req.origin_chat_id
                await ws.send(json.dumps(notification_payload))
                log.info("delegate_callback: result delivered to source", source=req.source_name,
                         origin_platform=req.origin_platform)
                # Wait briefly for acknowledgment
                try:
                    await asyncio.wait_for(ws.recv(), timeout=30)
                except asyncio.TimeoutError:
                    pass
        except Exception as exc:
            log.error("delegate_callback: failed to deliver result to source", source=req.source_name, error=str(exc))

    # Keep a strong reference so the task doesn't get garbage-collected
    _active_delegates.add(_deleg_key)
    task = asyncio.create_task(_background())
    if not hasattr(app.state, "_delegate_tasks"):
        app.state._delegate_tasks = set()
    app.state._delegate_tasks.add(task)

    def _on_delegate_done(t: Any) -> None:
        app.state._delegate_tasks.discard(t)
        _active_delegates.discard(_deleg_key)

    task.add_done_callback(_on_delegate_done)

    return {"ok": True, "message": f"Task delegated to {peer_display}. Results will be delivered to {req.source_name} when ready."}


_WORKSPACE_ALLOWED_EXTS: set[str] = {
    # Documents
    ".txt", ".md", ".markdown", ".rst", ".html", ".htm", ".pdf",
    ".doc", ".docx", ".odt", ".rtf",
    # Presentations & spreadsheets
    ".ppt", ".pptx", ".xls", ".xlsx", ".ods", ".odp",
    # Data
    ".csv", ".tsv", ".json", ".jsonl", ".yaml", ".yml", ".toml", ".xml",
    # Scripts & code
    ".py", ".js", ".ts", ".jsx", ".tsx", ".sh", ".bash", ".sql",
    ".rb", ".go", ".rs", ".java", ".c", ".cpp", ".h", ".css", ".scss",
    # Images
    ".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".bmp", ".ico",
    # Audio / video
    ".mp3", ".wav", ".ogg", ".mp4", ".webm",
    # Archives
    ".zip", ".tar", ".gz",
    # Config / misc text
    ".env", ".ini", ".cfg", ".conf", ".log",
}

_TEXT_EXTS: set[str] = {
    ".txt", ".md", ".markdown", ".rst", ".json", ".jsonl",
    ".yaml", ".yml", ".csv", ".tsv", ".toml", ".xml",
    ".py", ".js", ".ts", ".jsx", ".tsx", ".html", ".htm",
    ".css", ".scss", ".sh", ".bash", ".sql", ".log",
    ".env", ".ini", ".cfg", ".conf", ".rb", ".go", ".rs",
    ".java", ".c", ".cpp", ".h",
}


def _scan_workspace_files(agent_dir: Path) -> list[dict]:
    """Scan an agent's workspace directory for user-facing files."""
    import mimetypes
    results: list[dict] = []
    workspace = agent_dir / "data" / "workspace"
    if not workspace.is_dir():
        return results
    for f in workspace.rglob("*"):
        if not f.is_file():
            continue
        ext = f.suffix.lower()
        if ext not in _WORKSPACE_ALLOWED_EXTS:
            continue
        try:
            stat = f.stat()
            mt, _ = mimetypes.guess_type(f.name)
            results.append({
                "logical": "workspace/" + str(f.relative_to(workspace)),
                "physical": str(f),
                "filename": f.name,
                "extension": ext,
                "exists": True,
                "size": stat.st_size,
                "modified": stat.st_mtime,
                "mime_type": mt or "application/octet-stream",
                "is_text": ext in _TEXT_EXTS,
                "source": "workspace",
            })
        except OSError:
            continue
    return results


# ── Agent Datastore proxy ─────────────────────────────────────────────

@app.get("/fd/agent-datastore/{host}/{port}/tables")
async def agent_datastore_tables(
    host: str, port: int, token: str = "",
    user: dict | None = _required_user_dep,
):
    """List datastore tables from a CC agent (proxied)."""
    import httpx
    auth = token or _resolve_agent_auth(port)
    params = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/datastore/tables{params}"
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.get("/fd/agent-datastore/{host}/{port}/tables/{table_name}/rows")
async def agent_datastore_rows(
    host: str, port: int, table_name: str,
    limit: int = 100, offset: int = 0,
    order_by: str = "", order_dir: str = "asc",
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """Query rows from a datastore table on a CC agent (proxied)."""
    import httpx
    auth = token or _resolve_agent_auth(port)
    qs = f"token={auth}&" if auth else ""
    qs += f"limit={limit}&offset={offset}"
    if order_by:
        qs += f"&order_by={order_by}&order_dir={order_dir}"
    url = f"http://{host}:{port}/api/datastore/tables/{table_name}/rows?{qs}"
    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.get("/fd/agent-datastore/{host}/{port}/tables/{table_name}/export")
async def agent_datastore_export(
    host: str, port: int, table_name: str,
    format: str = "csv", token: str = "",
    user: dict | None = _required_user_dep,
):
    """Proxy datastore export (csv/json/xlsx) from a CC agent."""
    import httpx
    from fastapi.responses import Response as FastAPIResponse
    if format not in ("csv", "json", "xlsx"):
        raise HTTPException(400, f"Unsupported format: {format}")
    auth = token or _resolve_agent_auth(port)
    qs = f"token={auth}&" if auth else ""
    qs += f"format={format}"
    url = f"http://{host}:{port}/api/datastore/tables/{table_name}/export?{qs}"
    try:
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                return FastAPIResponse(
                    content=resp.content,
                    media_type=resp.headers.get("content-type", "application/octet-stream"),
                    headers={"Content-Disposition": resp.headers.get("content-disposition", f'attachment; filename="{table_name}.{format}"')},
                )
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.post("/fd/agent-rephrase/{host}/{port}")
async def agent_rephrase(
    host: str, port: int, request: Request,
    user: dict | None = _required_user_dep,
):
    """Proxy a rephrase request to a CC agent's LLM."""
    import httpx
    auth = _resolve_agent_auth(port)
    params = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/llm/complete{params}"
    body = await request.json()
    prompt_text = body.get("content", "")
    payload = {
        "messages": [
            {"role": "system", "content": "You are a prompt engineer. Rephrase the following user prompt to be clearer, more specific, and more effective for an AI agent. Keep the same intent but improve clarity, structure, and specificity. Return only the improved prompt text, nothing else."},
            {"role": "user", "content": prompt_text},
        ],
        "temperature": 0.7,
        "max_tokens": 4096,
    }
    try:
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.post(
                url, json=payload,
                headers={"Content-Type": "application/json"},
            )
            if resp.status_code == 200:
                data = resp.json()
                return {"content": data.get("content", prompt_text)}
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.get("/fd/agent-files/{host}/{port}")
async def agent_files(host: str, port: int, token: str = "", since: str = "", request: Request = None, user: dict | None = _required_user_dep):
    """List files from a CC agent (proxied to avoid CORS), merged with workspace scan.

    The optional ``since=<unix_seconds>`` query param is forwarded to the
    agent's ``/api/files`` endpoint, which filters by mtime. Used by the
    plan-monitor "Files" tab to show files touched during a plan run; the
    workspace-scan merge is skipped when ``since`` is set so unrelated old
    files don't leak through.
    """
    import httpx
    registered: list[dict] = []
    # Auto-resolve auth token if the caller didn't provide one
    auth = token or _resolve_agent_auth(port)
    qparts = []
    if auth:
        qparts.append(f"token={auth}")
    if since:
        qparts.append(f"since={since}")
    params = ("?" + "&".join(qparts)) if qparts else ""
    url = f"http://{host}:{port}/api/files{params}"
    agent_reachable = False
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                registered = resp.json()
                agent_reachable = True
    except (httpx.ConnectError, Exception):
        pass

    # Also scan the workspace directory for any unregistered files
    workspace_files: list[dict] = []
    user_id = getattr(request.state, "user_id", "") if request else ""
    # Try to find the agent directory by matching port to a process or container
    registry = _load_process_registry()
    for slug, entry in registry.items():
        if entry.get("web_port") == port:
            if AUTH_ENABLED and user_id and entry.get("owner", "") != user_id:
                continue
            agent_dir = DATA_DIR / slug
            if agent_dir.is_dir():
                workspace_files = _scan_workspace_files(agent_dir)
            break

    if not workspace_files:
        if not registered and not agent_reachable:
            raise HTTPException(502, "Cannot connect to agent")
        return registered

    # When the caller filtered by `since`, the agent already applied the
    # mtime filter to its own registry. Merging in unfiltered workspace
    # files would defeat that filter, so we return the agent's view as-is.
    if since:
        return registered

    # Merge: use registered files as base, add workspace files not already listed
    registered_physicals = {f.get("physical", "") for f in registered}
    registered_filenames = {f.get("filename", "") for f in registered}
    for wf in workspace_files:
        if wf["physical"] not in registered_physicals and wf["filename"] not in registered_filenames:
            registered.append(wf)
    return registered


# ── Intentions proxy (FD panel → agent /api/intentions*) ──────────────


async def _proxy_agent_intentions(
    method: str, host: str, port: int, token: str, path: str,
    *, query: dict | None = None, body: dict | None = None,
):
    """Forward an intentions request to an agent, preserving its status code."""
    import httpx
    from fastapi.responses import JSONResponse

    params = dict(query or {})
    auth = token or _resolve_agent_auth(port)
    if auth:
        params["token"] = auth
    url = f"http://{host}:{port}{path}"
    try:
        async with httpx.AsyncClient(timeout=15.0) as client:
            resp = await client.request(method, url, params=params, json=body)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")
    except Exception as exc:
        raise HTTPException(502, f"Agent error: {exc}")
    try:
        payload = resp.json()
    except Exception:
        payload = {"error": resp.text[:500]}
    return JSONResponse(payload, status_code=resp.status_code)


@app.get("/fd/agent-intentions/{host}/{port}")
async def agent_intentions(
    host: str, port: int, token: str = "", status: str = "", origin: str = "",
    user: dict | None = _required_user_dep,
):
    q: dict = {}
    if status:
        q["status"] = status
    if origin:
        q["origin"] = origin
    return await _proxy_agent_intentions("GET", host, port, token, "/api/intentions", query=q)


@app.get("/fd/agent-intentions-decisions/{host}/{port}")
async def agent_intention_decisions(
    host: str, port: int, token: str = "", user: dict | None = _required_user_dep,
):
    return await _proxy_agent_intentions(
        "GET", host, port, token, "/api/intentions/decisions"
    )


@app.post("/fd/agent-intentions-decision/{host}/{port}/{decision_id}/resolve")
async def agent_resolve_intention_decision(
    host: str, port: int, decision_id: str, request: Request,
    token: str = "", user: dict | None = _required_user_dep,
):
    try:
        body = await request.json()
    except Exception:
        body = {}
    return await _proxy_agent_intentions(
        "POST", host, port, token,
        f"/api/intentions/decisions/{decision_id}/resolve", body=body,
    )


@app.post("/fd/agent-intention/{host}/{port}/{intention_id}/status")
async def agent_set_intention_status(
    host: str, port: int, intention_id: str, request: Request,
    token: str = "", user: dict | None = _required_user_dep,
):
    try:
        body = await request.json()
    except Exception:
        body = {}
    return await _proxy_agent_intentions(
        "POST", host, port, token,
        f"/api/intentions/{intention_id}/status", body=body,
    )


class TransferRequest(BaseModel):
    src_host: str
    src_port: int
    src_auth: str = ""
    src_path: str
    dst_host: str
    dst_port: int
    dst_auth: str = ""


@app.post("/fd/transfer")
async def transfer_file(req: TransferRequest, request: Request, user: dict | None = _required_user_dep):
    """Download a file from one agent and upload it to another."""
    import httpx

    # Build query params properly: a path with `?` collisions plus an
    # unencoded source path used to silently corrupt both URLs and the source
    # agent ended up parsing `path=/foo/bar?token=xyz` as a single value.
    src_params: dict[str, str] = {"path": req.src_path}
    if req.src_auth:
        src_params["token"] = req.src_auth
    dst_params: dict[str, str] = {}
    if req.dst_auth:
        dst_params["token"] = req.dst_auth

    async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client:
        # Download from source
        dl_url = f"http://{req.src_host}:{req.src_port}/api/files/download"
        dl_resp = await client.get(dl_url, params=src_params)
        if dl_resp.status_code != 200:
            raise HTTPException(
                502,
                f"Source agent download failed: {dl_resp.status_code} "
                f"({dl_resp.text[:200]})",
            )

        # Get filename from content-disposition or path
        cd = dl_resp.headers.get("content-disposition", "")
        if "filename=" in cd:
            filename = cd.split("filename=")[-1].strip('" ')
        else:
            filename = req.src_path.rsplit("/", 1)[-1]

        # Upload to destination
        up_url = f"http://{req.dst_host}:{req.dst_port}/api/file/upload"
        files = {"file": (filename, dl_resp.content, dl_resp.headers.get("content-type", "application/octet-stream"))}
        up_resp = await client.post(up_url, params=dst_params, files=files)
        if up_resp.status_code != 200:
            raise HTTPException(
                502,
                f"Destination agent upload failed: {up_resp.status_code} "
                f"({up_resp.text[:200]})",
            )

        result = up_resp.json()
        return {"ok": True, "filename": filename, "dest_path": result.get("path", ""), "size": result.get("size", 0)}


@app.get("/fd/agent-file-download/{host}/{port}")
async def agent_file_download(host: str, port: int, path: str, token: str = "", request: Request = None, user: dict | None = _required_user_dep):
    """Proxy file download from a CC agent."""
    import httpx
    import urllib.parse
    auth = token or _resolve_agent_auth(port)
    params = f"path={urllib.parse.quote(path)}"
    if auth:
        params += f"&token={urllib.parse.quote(auth)}"
    url = f"http://{host}:{port}/api/files/download?{params}"
    try:
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.get(url)
            if resp.status_code != 200:
                raise HTTPException(resp.status_code, f"Agent returned {resp.status_code}")
            cd = resp.headers.get("content-disposition", "")
            ct = resp.headers.get("content-type", "application/octet-stream")
            headers = {"Content-Type": ct}
            if cd:
                headers["Content-Disposition"] = cd
            else:
                filename = path.rsplit("/", 1)[-1]
                headers["Content-Disposition"] = f'attachment; filename="{filename}"'
            from starlette.responses import Response
            return Response(content=resp.content, headers=headers)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.get("/fd/agent-file-view/{host}/{port}")
async def agent_file_view(host: str, port: int, path: str, token: str = "", request: Request = None, user: dict | None = _required_user_dep):
    """Proxy file view from a CC agent (inline, no download header)."""
    import httpx
    import urllib.parse
    auth = token or _resolve_agent_auth(port)
    params = f"path={urllib.parse.quote(path)}"
    if auth:
        params += f"&token={urllib.parse.quote(auth)}"
    # Try the /api/files/view endpoint first, fall back to /download
    url = f"http://{host}:{port}/api/files/view?{params}"
    try:
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.get(url)
            if resp.status_code != 200:
                # Fallback to download endpoint
                url2 = f"http://{host}:{port}/api/files/download?{params}"
                resp = await client.get(url2)
                if resp.status_code != 200:
                    raise HTTPException(resp.status_code, f"Agent returned {resp.status_code}")
            ct = resp.headers.get("content-type", "text/plain")
            from starlette.responses import Response
            return Response(content=resp.content, headers={
                "Content-Type": ct,
                "Content-Disposition": "inline",
            })
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.post("/fd/agent-file-upload/{host}/{port}")
async def agent_file_upload(host: str, port: int, token: str = "", file: UploadFile = File(...), request: Request = None, user: dict | None = _required_user_dep):
    """Proxy file upload to a CC agent."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    params = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/file/upload{params}"
    content = await file.read()
    try:
        async with httpx.AsyncClient(timeout=120.0) as client:
            files = {"file": (file.filename or "upload", content, file.content_type or "application/octet-stream")}
            resp = await client.post(url, files=files)
            if resp.status_code != 200:
                # Surface the agent's actual error (e.g. "Unsupported file type
                # '.mov'") instead of a bare status code — invaluable for debugging.
                detail = resp.text[:500] if resp.text else ""
                raise HTTPException(resp.status_code, f"Agent upload failed ({resp.status_code}): {detail}")
            # Log usage
            if AUTH_ENABLED and user:
                db = app.state.fd_db
                await db.log_usage(user["id"], "file_upload", json.dumps({"filename": file.filename, "size": len(content), "agent_port": port}))
            return resp.json()
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.post("/fd/agent-file-save/{host}/{port}")
async def agent_file_save(host: str, port: int, token: str = "", request: Request = None, user: dict | None = _required_user_dep):
    """Proxy a text-file save to a CC agent (POST /api/files/content).

    Body: ``{"path": "<physical>", "content": "<text>"}``. The agent applies
    the same path-sandbox + token auth as its read endpoints."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    params = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/files/content{params}"
    body = await request.json()
    try:
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.post(url, json=body)
            if resp.status_code != 200:
                detail = resp.text[:500] if resp.text else ""
                raise HTTPException(resp.status_code, f"Agent save failed ({resp.status_code}): {detail}")
            if AUTH_ENABLED and user:
                db = app.state.fd_db
                await db.log_usage(user["id"], "file_save", json.dumps({"path": body.get("path", ""), "agent_port": port}))
            return resp.json()
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.get("/fd/agent-usage/{host}/{port}")
async def agent_usage(host: str, port: int, token: str = "", period: str = "today", request: Request = None, user: dict | None = _required_user_dep):
    """Proxy /api/usage from a CC agent."""
    import httpx
    auth = token or _resolve_agent_auth(port)
    params = f"?period={period}"
    if auth:
        params += f"&token={auth}"
    url = f"http://{host}:{port}/api/usage{params}"
    try:
        # First request with token may return a 302 redirect that sets a cookie.
        # Use follow_redirects=False to capture the cookie, then retry.
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.get(url, follow_redirects=False)

            if resp.status_code in (301, 302, 303, 307, 308):
                # Token was accepted and a cookie was set — retry with the cookie
                cookies = resp.cookies
                retry_url = f"http://{host}:{port}/api/usage?period={period}"
                resp = await client.get(retry_url, cookies=cookies)

            if resp.status_code != 200:
                raise HTTPException(resp.status_code, f"Agent returned {resp.status_code}")
            return resp.json()
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


# ── Orchestrator proxy endpoints ──


@app.post("/fd/orchestrator/{host}/{port}/prepare-tasks")
async def proxy_prepare_tasks(
    host: str, port: int, request: Request,
    user: dict | None = _required_user_dep,
):
    """Proxy prepare-tasks to a CC agent's orchestrator."""
    import httpx

    auth = _resolve_agent_auth(port)
    params = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/orchestrator/prepare-tasks{params}"
    body = await request.body()
    try:
        async with httpx.AsyncClient(timeout=120.0) as client:
            resp = await client.post(url, content=body, headers={"Content-Type": "application/json"})
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.post("/fd/orchestrator/{host}/{port}/run-tasks")
async def proxy_run_tasks(
    host: str, port: int, request: Request,
    user: dict | None = _required_user_dep,
):
    """Proxy run-tasks to a CC agent's orchestrator.

    Note: This is a long-running request. Real-time progress comes
    via the agent WebSocket (orchestrator_event messages).
    """
    import httpx

    auth = _resolve_agent_auth(port)
    params = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/orchestrator/run-tasks{params}"
    body = await request.body()
    try:
        async with httpx.AsyncClient(timeout=600.0) as client:
            resp = await client.post(url, content=body, headers={"Content-Type": "application/json"})
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.get("/fd/orchestrator/{host}/{port}/workspace")
async def proxy_workspace_snapshot(
    host: str, port: int, token: str = "",
    user: dict | None = _required_user_dep,
):
    """Proxy workspace snapshot from a CC agent's orchestrator."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    params = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/orchestrator/workspace{params}"
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.get("/fd/orchestrator/{host}/{port}/traces")
async def proxy_traces(
    host: str, port: int, token: str = "",
    user: dict | None = _required_user_dep,
):
    """Proxy trace spans from a CC agent's orchestrator."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    params = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/orchestrator/traces{params}"
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


# ── Memory transfer (curated insights + reflections) ──


@app.get("/fd/agent-memory/{host}/{port}/export")
async def proxy_memory_export(
    host: str, port: int,
    token: str = "",
    min_importance: int = 0,
    include_expired: bool = False,
    reflection_limit: int = 50,
    include_semantic: bool = False,
    semantic_limit: int = 1000,
    semantic_min_chars: int = 100,
    semantic_sources: str = "",
    semantic_include_imported: bool = False,
    user: dict | None = _required_user_dep,
):
    """Proxy a curated-memory bundle download from a CC agent.

    Streams the agent's ``/api/memory/export`` response back to the
    Flight Deck client, preserving the file download headers so the
    browser saves it as a JSON file.

    When ``include_semantic`` is true, the bundle additionally contains
    a text-only dump of the agent's semantic memory chunks (no vectors;
    the importing agent re-embeds with its own provider).
    """
    import httpx
    from fastapi.responses import Response

    auth = token or _resolve_agent_auth(port)
    qs_parts = []
    if auth:
        qs_parts.append(f"token={auth}")
    qs_parts.append(f"min_importance={int(min_importance)}")
    if include_expired:
        qs_parts.append("include_expired=1")
    qs_parts.append(f"reflection_limit={int(reflection_limit)}")
    if include_semantic:
        qs_parts.append("include_semantic=1")
        qs_parts.append(f"semantic_limit={int(semantic_limit)}")
        qs_parts.append(f"semantic_min_chars={int(semantic_min_chars)}")
        if semantic_sources:
            qs_parts.append(f"semantic_sources={semantic_sources}")
        if semantic_include_imported:
            qs_parts.append("semantic_include_imported=1")
    url = f"http://{host}:{port}/api/memory/export?{'&'.join(qs_parts)}"

    try:
        async with httpx.AsyncClient(timeout=120.0) as client:
            resp = await client.get(url)
            if resp.status_code != 200:
                raise HTTPException(resp.status_code, resp.text)
            disposition = resp.headers.get(
                "content-disposition",
                f'attachment; filename="captain-claw-memory-{host}-{port}.json"',
            )
            return Response(
                content=resp.content,
                media_type="application/json",
                headers={"Content-Disposition": disposition},
            )
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.post("/fd/agent-memory/{host}/{port}/import")
async def proxy_memory_import(
    host: str, port: int,
    request: Request,
    token: str = "",
    min_importance: int = 0,
    source_label: str = "",
    stage_conflicts: bool = False,
    skip_semantic: bool = False,
    user: dict | None = _required_user_dep,
):
    """Proxy a curated-memory bundle into a CC agent.

    Body: the JSON bundle previously downloaded via ``proxy_memory_export``
    (or any compatible bundle from another agent). When ``stage_conflicts``
    is true, conflicting decision/preference/workflow insights are routed
    to the agent's pending-review queue instead of being silently deduped.
    When ``skip_semantic`` is true, any ``semantic_chunks`` present in the
    bundle are dropped on the server side (curated-only import).
    """
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs_parts = []
    if auth:
        qs_parts.append(f"token={auth}")
    qs_parts.append(f"min_importance={int(min_importance)}")
    if source_label:
        qs_parts.append(f"source_label={source_label}")
    if stage_conflicts:
        qs_parts.append("stage_conflicts=1")
    if skip_semantic:
        qs_parts.append("skip_semantic=1")
    url = f"http://{host}:{port}/api/memory/import?{'&'.join(qs_parts)}"
    body = await request.body()

    try:
        async with httpx.AsyncClient(timeout=120.0) as client:
            resp = await client.post(
                url,
                content=body,
                headers={"Content-Type": "application/json"},
            )
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.get("/fd/agent-memory/{host}/{port}/reflections/imported")
async def proxy_list_imported_reflections(
    host: str, port: int,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """List imported (staged) reflections on a CC agent for the merge picker."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/memory/reflections/imported{qs}"

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.post("/fd/agent-memory/{host}/{port}/reflections/merge")
async def proxy_merge_reflection(
    host: str, port: int,
    request: Request,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """Trigger a personality-preserving reflection merge on a CC agent.

    Body: ``{"label": "<imported-subdir>", "filename": "<optional>"}``.
    The agent runs its reflection-merge LLM flow and writes the result as
    the new active reflection.
    """
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/memory/reflections/merge{qs}"
    body = await request.body()

    try:
        async with httpx.AsyncClient(timeout=300.0) as client:
            resp = await client.post(
                url,
                content=body,
                headers={"Content-Type": "application/json"},
            )
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


# ── Semantic memory import management ──


@app.get("/fd/agent-memory/{host}/{port}/semantic/labels")
async def proxy_list_semantic_imports(
    host: str, port: int,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """List imported semantic sources on a CC agent."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/memory/semantic/labels{qs}"

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.delete("/fd/agent-memory/{host}/{port}/semantic/labels/{label}")
async def proxy_delete_semantic_import(
    host: str, port: int, label: str,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """Purge one imported semantic source on a CC agent."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/memory/semantic/labels/{label}{qs}"

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.delete(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


# ── Pending-review insights (stage_conflicts queue) ──


@app.get("/fd/agent-insights/{host}/{port}/pending")
async def proxy_list_pending_insights(
    host: str, port: int,
    token: str = "",
    category: str = "",
    limit: int = 100,
    user: dict | None = _required_user_dep,
):
    """List the agent's staged insight conflicts awaiting review."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs_parts = []
    if auth:
        qs_parts.append(f"token={auth}")
    if category:
        qs_parts.append(f"category={category}")
    qs_parts.append(f"limit={int(limit)}")
    url = f"http://{host}:{port}/api/insights/pending?{'&'.join(qs_parts)}"

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.get("/fd/agent-insights/{host}/{port}/pending/count")
async def proxy_count_pending_insights(
    host: str, port: int,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """Cheap poll for the UI badge."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/insights/pending/count{qs}"

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.post("/fd/agent-insights/{host}/{port}/pending/{pending_id}/approve")
async def proxy_approve_pending_insight(
    host: str, port: int, pending_id: str,
    token: str = "",
    supersede: bool = True,
    user: dict | None = _required_user_dep,
):
    """Promote a staged insight into the live table."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs_parts = []
    if auth:
        qs_parts.append(f"token={auth}")
    qs_parts.append(f"supersede={'1' if supersede else '0'}")
    url = (
        f"http://{host}:{port}/api/insights/pending/{pending_id}/approve"
        f"?{'&'.join(qs_parts)}"
    )

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.post(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.post("/fd/agent-insights/{host}/{port}/pending/{pending_id}/reject")
async def proxy_reject_pending_insight(
    host: str, port: int, pending_id: str,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """Drop a staged insight without promoting it."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/insights/pending/{pending_id}/reject{qs}"

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.post(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


# ── Today aggregator proxy endpoints (reflections / cron / intuitions / skills) ──


async def _proxy_get_json(host: str, port: int, path: str, token: str = "", timeout: float = 15.0):
    """Helper: GET an agent endpoint and return parsed JSON or raise HTTPException."""
    import httpx
    auth = token or _resolve_agent_auth(port)
    sep = "&" if "?" in path else "?"
    url = f"http://{host}:{port}{path}"
    if auth:
        url = f"{url}{sep}token={auth}"
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            resp = await client.get(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.get("/fd/agent-reflections/{host}/{port}")
async def proxy_list_reflections(
    host: str, port: int,
    token: str = "",
    limit: int = 20,
    user: dict | None = _required_user_dep,
):
    """List recent reflections on a CC agent."""
    return await _proxy_get_json(host, port, f"/api/reflections?limit={int(limit)}", token=token)


@app.get("/fd/agent-reflections/{host}/{port}/latest")
async def proxy_latest_reflection(
    host: str, port: int,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """Get the latest active reflection from a CC agent."""
    return await _proxy_get_json(host, port, "/api/reflections/latest", token=token)


@app.get("/fd/agent-cron/{host}/{port}")
async def proxy_list_cron(
    host: str, port: int,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """List cron jobs on a CC agent."""
    return await _proxy_get_json(host, port, "/api/cron/jobs", token=token)


@app.get("/fd/agent-intuitions/{host}/{port}")
async def proxy_list_intuitions(
    host: str, port: int,
    token: str = "",
    limit: int = 20,
    user: dict | None = _required_user_dep,
):
    """List recent intuitions (nervous system) on a CC agent."""
    return await _proxy_get_json(
        host, port, f"/api/nervous-system?limit={int(limit)}", token=token
    )


# ── Skills proxy endpoints ──


@app.get("/fd/agent-skills/{host}/{port}")
async def proxy_list_skills(
    host: str, port: int,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """List installed skills on a CC agent."""
    return await _proxy_get_json(host, port, "/api/skills", token=token, timeout=30.0)


@app.post("/fd/agent-skills/{host}/{port}/install")
async def proxy_install_skill(
    host: str, port: int,
    request: Request,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """Install a skill on a CC agent from a GitHub URL."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/skills/install{qs}"
    body = await request.body()
    try:
        async with httpx.AsyncClient(timeout=180.0) as client:
            resp = await client.post(
                url, content=body, headers={"Content-Type": "application/json"}
            )
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.post("/fd/agent-skills/{host}/{port}/install-upload")
async def proxy_install_skill_upload(
    host: str, port: int,
    request: Request,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """Install a skill on a CC agent from an uploaded .md or .zip file."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/skills/install-upload{qs}"
    body = await request.body()
    content_type = request.headers.get("content-type", "application/octet-stream")
    try:
        async with httpx.AsyncClient(timeout=180.0) as client:
            resp = await client.post(
                url, content=body, headers={"Content-Type": content_type}
            )
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


@app.post("/fd/agent-skills/{host}/{port}/toggle")
async def proxy_toggle_skill(
    host: str, port: int,
    request: Request,
    token: str = "",
    user: dict | None = _required_user_dep,
):
    """Enable or disable a skill on a CC agent."""
    import httpx

    auth = token or _resolve_agent_auth(port)
    qs = f"?token={auth}" if auth else ""
    url = f"http://{host}:{port}/api/skills/toggle{qs}"
    body = await request.body()
    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.post(
                url, content=body, headers={"Content-Type": "application/json"}
            )
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


# ── Captain Claw Game proxy endpoints ──


async def _proxy_post_json(
    host: str, port: int, path: str, token: str, body: bytes, timeout: float = 30.0,
):
    """Helper: POST a JSON body to an agent endpoint and return parsed JSON."""
    import httpx
    auth = token or _resolve_agent_auth(port)
    sep = "&" if "?" in path else "?"
    url = f"http://{host}:{port}{path}"
    if auth:
        url = f"{url}{sep}token={auth}"
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            resp = await client.post(
                url, content=body or b"{}",
                headers={"Content-Type": "application/json"},
            )
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")


async def _proxy_delete_json(
    host: str, port: int, path: str, token: str, timeout: float = 15.0,
):
    """Helper: DELETE an agent endpoint and return parsed JSON."""
    import httpx
    auth = token or _resolve_agent_auth(port)
    sep = "&" if "?" in path else "?"
    url = f"http://{host}:{port}{path}"
    if auth:
        url = f"{url}{sep}token={auth}"
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            resp = await client.delete(url)
            if resp.status_code == 200:
                return resp.json()
            raise HTTPException(resp.status_code, resp.text)
    except httpx.ConnectError:
        raise HTTPException(502, "Cannot connect to agent")



# (Game proxy routes removed — games hosted directly by FD via games_routes.py)


# ── Process agent endpoints ──


class ProcessInfo(BaseModel):
    slug: str
    name: str
    description: str = ""
    status: str  # running | stopped
    web_port: int
    web_auth: str = ""
    pid: int | None = None
    provider: str = ""
    model: str = ""
    freebie: bool = False


class ProcessActionResult(BaseModel):
    ok: bool
    slug: str
    message: str = ""


@app.get("/fd/processes", response_model=list[ProcessInfo])
async def list_processes(request: Request, user: dict | None = _required_user_dep):
    """List all Flight Deck managed process agents."""
    registry = _load_process_registry()
    user_id = getattr(request.state, "user_id", "")
    result = []
    for slug, entry in registry.items():
        if AUTH_ENABLED and user_id and entry.get("owner", "") != user_id:
            continue
        alive = _process_is_alive(slug)
        result.append(ProcessInfo(
            slug=slug,
            name=entry.get("name", slug),
            description=entry.get("description", ""),
            status="running" if alive else "stopped",
            web_port=entry.get("web_port", 0),
            web_auth=entry.get("web_auth", ""),
            pid=entry.get("pid") if alive else None,
            provider=entry.get("provider", ""),
            model=entry.get("model", ""),
            freebie=bool(entry.get("freebie", False)),
        ))
    return result


@app.post("/fd/spawn-process", response_model=ProcessActionResult)
async def spawn_process(config: AgentConfig, request: Request, user: dict | None = _optional_user_dep):
    """Spawn a new Captain Claw process agent (pip-installed, no Docker)."""
    # Serialise spawns so two concurrent requests can't both land on the same
    # port between the port-pick and the Popen. The lock also covers a small
    # settle delay after launch to let the child's TCPSite.bind succeed (or
    # drift + announce) before the next spawn's port probe runs.
    async with _get_spawn_lock():
        return await _spawn_process_locked(config, request, user)


async def _spawn_process_locked(config: AgentConfig, request: Request, user: dict | None):
    # Rate limiting & agent count check
    if AUTH_ENABLED and user:
        check_api_rate_limit(user)
        check_spawn_rate_limit(user)
        # Count existing processes for this user
        registry = _load_process_registry()
        user_id = user["id"]
        owned = [e for e in registry.values() if e.get("owner") == user_id]
        await check_agent_count_limit(user, len(owned))

    slug = _slug(config.name)

    # Check if already running
    if _process_is_alive(slug):
        raise HTTPException(400, f"Process '{slug}' is already running. Stop it first or use a different name.")

    # Ensure port is available; find a free one if not
    if config.web_enabled and (config.web_port <= 0 or not _is_port_available(config.web_port)):
        config.web_port = _find_available_port(config.web_port if config.web_port > 0 else 24080)

    # Auto-generate auth token if none provided — prevents unauthenticated
    # direct access to agent ports bypassing Flight Deck.
    if config.web_enabled and not config.web_auth_token:
        config.web_auth_token = secrets.token_urlsafe(32)

    # Prepare data directory
    agent_dir = DATA_DIR / slug
    agent_dir.mkdir(parents=True, exist_ok=True)
    (agent_dir / "data" / "workspace").mkdir(parents=True, exist_ok=True)
    (agent_dir / "data" / "sessions").mkdir(parents=True, exist_ok=True)
    (agent_dir / "data" / "skills").mkdir(parents=True, exist_ok=True)
    (agent_dir / "data" / "home-config").mkdir(parents=True, exist_ok=True)

    # Write config files with local paths
    config_yaml = _build_process_config_yaml(config, agent_dir)
    (agent_dir / "config.yaml").write_text(config_yaml)
    (agent_dir / "data" / "home-config" / "config.yaml").write_text(config_yaml)

    env_content = _build_env(config)
    (agent_dir / ".env").write_text(env_content)

    # Build environment variables
    environment = dict(os.environ)
    if env_content:
        for line in env_content.strip().split("\n"):
            if "=" in line:
                k, v = line.split("=", 1)
                environment[k] = v
    for ev in config.env_vars:
        if ev.get("key"):
            environment[ev["key"]] = ev.get("value", "")

    # Resolve owner: authenticated user > owner_hint > infer from existing registry
    owner_id = getattr(request.state, "user_id", "") or config.owner_hint
    if not owner_id:
        # Fallback: inherit owner from an existing agent in the registry.
        # Covers the case where an internal caller (e.g. Old Man spawned before
        # the FD_OWNER_ID env var was added) doesn't have owner info.
        _reg = _load_process_registry()
        for _entry in _reg.values():
            if _entry.get("owner"):
                owner_id = _entry["owner"]
                break

    # Tell agents how to reach Flight Deck internally (for Telegram, Discord, etc.)
    if "FD_URL" not in environment:
        fd_port = os.environ.get("FD_PORT", "25080")
        environment["FD_URL"] = f"http://localhost:{fd_port}"

    # Pass owner ID so child agents can propagate ownership when spawning
    if owner_id:
        environment["FD_OWNER_ID"] = owner_id

    # Slug for port-fallback callbacks: lets the agent tell FD which actual
    # port it bound to if the requested one was already in use.
    environment["FD_AGENT_SLUG"] = slug

    # Set HOME to the agent's home-config directory so captain-claw
    # picks up ~/.captain-claw/config.yaml from there
    environment["HOME"] = str(agent_dir / "data" / "home-config-parent")
    home_cc_dir = agent_dir / "data" / "home-config-parent" / ".captain-claw"
    home_cc_dir.mkdir(parents=True, exist_ok=True)
    # Symlink or copy home-config -> ~/.captain-claw
    hc_config = agent_dir / "data" / "home-config" / "config.yaml"
    hc_target = home_cc_dir / "config.yaml"
    if hc_target.exists() or hc_target.is_symlink():
        hc_target.unlink()
    shutil.copy2(str(hc_config), str(hc_target))

    # Write cognitive mode file for the agent.
    if config.cognitive_mode and config.cognitive_mode != "neutra":
        mode_file = home_cc_dir / "cognitive_mode.txt"
        mode_file.write_text(config.cognitive_mode, encoding="utf-8")

    # New agents deploy in eco mode by default.
    _write_eco_flag_on_spawn(agent_dir)

    # Open log file
    log_file = agent_dir / "process.log"
    log_fh = open(log_file, "a")

    # Resolve captain-claw-web binary: bundled (PyInstaller) or PATH
    cc_web_bin = _resolve_cc_web_bin()

    # IMPORTANT: write the registry entry BEFORE we Popen the child. The
    # child can bind, drift to a fallback port, and POST to /announce-port
    # in the milliseconds between Popen and the post-Popen registry write —
    # if we wrote the registry after Popen we'd race the announce and
    # silently overwrite the drifted port back to the original (stale)
    # value. Pre-writing the entry also guarantees the announce-port
    # endpoint can find the slug.
    registry = _load_process_registry()
    registry[slug] = {
        "slug": slug,
        "name": config.name or slug,
        "description": config.description,
        "web_port": config.web_port,
        "web_auth": config.web_auth_token,
        "pid": None,  # filled in below once Popen returns
        "provider": config.provider,
        "model": config.model,
        "owner": owner_id,
    }
    _save_process_registry(registry)

    try:
        proc = subprocess.Popen(
            [cc_web_bin, "--port", str(config.web_port)],
            cwd=str(agent_dir),
            env=environment,
            stdout=log_fh,
            stderr=subprocess.STDOUT,
            start_new_session=True,  # Detach from parent
        )
    except FileNotFoundError:
        log_fh.close()
        raise HTTPException(500, f"captain-claw-web not found at '{cc_web_bin}'. Install captain-claw via pip first.")
    except Exception as exc:
        log_fh.close()
        raise HTTPException(500, f"Failed to start process: {exc}")

    _processes[slug] = proc

    # Stamp the live PID. Read-modify-write so we don't clobber a port the
    # child may have already announced between Popen and now.
    registry = _load_process_registry()
    if slug in registry:
        registry[slug]["pid"] = proc.pid
        _save_process_registry(registry)

    # Log usage
    if AUTH_ENABLED and user:
        db = app.state.fd_db
        await db.log_usage(user["id"], "agent_spawn", json.dumps({"agent": slug, "type": "process", "provider": config.provider, "model": config.model}))

    # Let the child actually bind its TCP port (and potentially announce a
    # drift back to us) before the spawn lock is released and the next queued
    # spawn runs its port probe. Without this window two back-to-back spawns
    # can both see the same port as "free".
    settle_s = float(os.environ.get("FD_SPAWN_SETTLE_S", "0.3"))
    if settle_s > 0:
        await asyncio.sleep(settle_s)

    # After settle, re-read the registry: if the child drifted and announced
    # a new port, we want to surface the actual bound port in our response
    # (and to peers we notify) instead of the originally requested one.
    final_registry = _load_process_registry()
    actual_port = final_registry.get(slug, {}).get("web_port", config.web_port)
    log.info(
        "spawn-process: post-settle registry read",
        slug=slug,
        requested=config.web_port,
        actual=actual_port,
        drifted=actual_port != config.web_port,
    )
    if actual_port != config.web_port:
        log.info(
            "spawn-process: child drifted to a fallback port",
            slug=slug,
            requested=config.web_port,
            actual=actual_port,
        )
        config.web_port = actual_port
    else:
        # No drift detected yet — but the child's announce might still be in
        # flight (network/scheduler latency, slow startup). Schedule a deferred
        # re-check that won't block the spawn response but will still surface
        # the corrected port to the FD UI on its next poll.
        async def _late_drift_check():
            for _ in range(10):  # up to ~5 s
                await asyncio.sleep(0.5)
                _later = _load_process_registry().get(slug, {}).get("web_port", 0)
                if _later and _later != config.web_port:
                    log.info(
                        "spawn-process: late drift detected after spawn returned",
                        slug=slug,
                        was=config.web_port,
                        now=_later,
                    )
                    return
        try:
            asyncio.get_event_loop().create_task(_late_drift_check())
        except RuntimeError:
            pass

    # Notify other agents about the new peer (scoped to same owner). Done
    # AFTER the settle so the announced port is used, not the stale one.
    if config.web_enabled:
        _schedule_fleet_notify(config.name or slug, config.web_port, owner_id=owner_id)

    return ProcessActionResult(ok=True, slug=slug, message=f"Process agent '{slug}' spawned (PID {proc.pid}, port {config.web_port})")


def _verify_process_owner(slug: str, user_id: str) -> dict:
    """Check that a process exists and belongs to the user. Returns the registry entry."""
    registry = _load_process_registry()
    entry = registry.get(slug)
    if not entry:
        raise HTTPException(status_code=404, detail=f"Process '{slug}' not found")
    if AUTH_ENABLED and user_id and entry.get("owner", "") != user_id:
        raise HTTPException(status_code=404, detail=f"Process '{slug}' not found")
    return entry


def _do_stop_process(slug: str) -> ProcessActionResult:
    """Internal helper to stop a process agent (no auth check)."""
    if not _process_is_alive(slug):
        return ProcessActionResult(ok=True, slug=slug, message="Already stopped")

    proc = _processes.get(slug)
    registry = _load_process_registry()
    pid = proc.pid if proc else registry.get(slug, {}).get("pid")

    if pid:
        _kill_pid(pid)

    # Update registry — mark as intentionally stopped
    if slug in registry:
        registry[slug]["pid"] = None
        registry[slug]["stopped"] = True
        _save_process_registry(registry)

    _processes.pop(slug, None)
    return ProcessActionResult(ok=True, slug=slug, message="Stopped")


def _do_start_process(slug: str) -> ProcessActionResult:
    """Internal helper to start a process agent (no auth check)."""
    registry = _load_process_registry()
    entry = registry.get(slug)
    if not entry:
        raise HTTPException(404, f"Process '{slug}' not found in registry")

    if _process_is_alive(slug):
        return ProcessActionResult(ok=True, slug=slug, message="Already running")

    if not (DATA_DIR / slug).is_dir():
        raise HTTPException(404, f"Agent directory not found: {DATA_DIR / slug}")

    # Clear stopped flag — user is intentionally starting this agent
    entry.pop("stopped", None)

    if not _start_registered_process(slug, entry):
        raise HTTPException(500, "Failed to start process (captain-claw-web not found?)")

    _save_process_registry(registry)

    return ProcessActionResult(ok=True, slug=slug, message=f"Started (PID {entry.get('pid', '?')})")


class AnnouncePortRequest(BaseModel):
    """Agent → FD callback: actual port the agent successfully bound to."""
    port: int
    auth: str = ""  # web_auth_token used as a shared secret to authenticate the callback


@app.post("/fd/processes/{slug}/announce-port", response_model=ProcessActionResult)
async def announce_process_port(slug: str, body: AnnouncePortRequest):
    """Update a process's actual web port after the agent fell back to a free
    port (because the requested one was already in use). Authenticated via the
    process's existing web_auth token, so no user session is required.

    Side-effects: persists the new port in the registry and re-broadcasts the
    fleet membership so peer agents and Flight Deck UI clients learn the new
    address.
    """
    log.info("announce-port: received", slug=slug, new_port=body.port)
    registry = _load_process_registry()
    entry = registry.get(slug)
    if not entry:
        log.warning("announce-port: slug not in registry", slug=slug, registry_slugs=list(registry.keys()))
        raise HTTPException(404, f"Process '{slug}' not found")

    expected_auth = entry.get("web_auth", "")
    # Compare with constant-time helper to avoid trivial timing leaks.
    import hmac
    if expected_auth and not hmac.compare_digest(expected_auth, body.auth or ""):
        log.warning("announce-port: auth mismatch", slug=slug)
        raise HTTPException(401, "Invalid auth token for port announce")

    if body.port <= 0 or body.port > 65535:
        raise HTTPException(400, f"Invalid port: {body.port}")

    old_port = entry.get("web_port", 0)
    if old_port == body.port:
        log.info("announce-port: unchanged", slug=slug, port=body.port)
        return ProcessActionResult(ok=True, slug=slug, message=f"Port unchanged ({body.port})")

    entry["web_port"] = body.port
    _save_process_registry(registry)
    log.info("announce-port: registry updated", slug=slug, old_port=old_port, new_port=body.port)

    # Verify the write actually landed by re-reading. Catches the (theoretical)
    # case where another writer raced us between save and return.
    _verify = _load_process_registry().get(slug, {}).get("web_port", 0)
    if _verify != body.port:
        log.error(
            "announce-port: registry write was clobbered immediately after save!",
            slug=slug,
            wrote=body.port,
            now_reads=_verify,
        )

    # Re-notify the fleet so peers update their connection details.
    owner_id = entry.get("owner", "")
    name = entry.get("name", slug)
    _schedule_fleet_notify(name, body.port, event="rebound", owner_id=owner_id)

    return ProcessActionResult(
        ok=True,
        slug=slug,
        message=f"Updated port: {old_port} → {body.port}",
    )


@app.post("/fd/processes/{slug}/stop", response_model=ProcessActionResult)
async def stop_process(slug: str, request: Request, user: dict | None = _required_user_dep):
    """Stop a running process agent."""
    _verify_process_owner(slug, getattr(request.state, "user_id", ""))
    import asyncio
    return await asyncio.get_event_loop().run_in_executor(None, _do_stop_process, slug)


@app.post("/fd/processes/{slug}/start", response_model=ProcessActionResult)
async def start_process(slug: str, request: Request, user: dict | None = _required_user_dep):
    """Start a stopped process agent."""
    _verify_process_owner(slug, getattr(request.state, "user_id", ""))
    import asyncio
    return await asyncio.get_event_loop().run_in_executor(None, _do_start_process, slug)


@app.post("/fd/processes/{slug}/restart", response_model=ProcessActionResult)
async def restart_process(slug: str, request: Request, user: dict | None = _required_user_dep):
    """Restart a process agent."""
    _verify_process_owner(slug, getattr(request.state, "user_id", ""))
    _do_stop_process(slug)
    import time
    time.sleep(1)
    return _do_start_process(slug)


@app.delete("/fd/processes/{slug}", response_model=ProcessActionResult)
async def remove_process(slug: str, force: bool = False, request: Request = None, user: dict | None = _required_user_dep):
    _verify_process_owner(slug, getattr(request.state, "user_id", ""))
    """Remove a process agent from the registry. Stops it first if running."""
    if _process_is_alive(slug):
        _do_stop_process(slug)

    registry = _load_process_registry()
    registry.pop(slug, None)
    _save_process_registry(registry)
    _processes.pop(slug, None)

    return ProcessActionResult(ok=True, slug=slug, message=f"Removed '{slug}' from registry")


@app.get("/fd/processes/{slug}/logs")
async def process_logs(slug: str, tail: int = 200, since_byte: int = 0, request: Request = None, user: dict | None = _required_user_dep):
    _verify_process_owner(slug, getattr(request.state, "user_id", ""))
    """Read logs from a process agent's log file.

    When *since_byte* > 0 the response only contains bytes written after
    that offset (incremental fetch).  The response always includes
    ``byte_offset`` – the file size at the time of reading – so the
    frontend can pass it back on the next poll.
    """
    log_file = DATA_DIR / slug / "process.log"
    if not log_file.is_file():
        return {"logs": "(no logs yet)", "byte_offset": 0}
    try:
        file_size = log_file.stat().st_size
        if since_byte > 0 and since_byte <= file_size:
            # Incremental: read only new bytes since last poll
            with open(log_file, "rb") as f:
                f.seek(since_byte)
                new_bytes = f.read()
            text = new_bytes.decode("utf-8", errors="replace")
            # Strip leading partial line if we didn't land on a boundary
            if since_byte > 0 and text and text[0] != "\n" and since_byte < file_size:
                nl = text.find("\n")
                if nl >= 0:
                    text = text[nl + 1:]
            return {"logs": text, "byte_offset": file_size}
        # Initial fetch: efficiently read last N lines from end of file
        chunk_size = max(4096, tail * 200)  # reasonable estimate
        with open(log_file, "rb") as f:
            f.seek(0, 2)  # seek to end
            end_pos = f.tell()
            start = max(0, end_pos - chunk_size)
            f.seek(start)
            data = f.read()
        text = data.decode("utf-8", errors="replace")
        lines = text.splitlines()
        # If we didn't read from start, first line may be partial – drop it
        if start > 0 and len(lines) > tail:
            lines = lines[1:]
        tail_lines = lines[-tail:] if len(lines) > tail else lines
        return {"logs": "\n".join(tail_lines), "byte_offset": end_pos}
    except Exception as exc:
        return {"logs": f"(error reading logs: {exc})", "byte_offset": 0}


@app.post("/fd/processes/{slug}/clone", response_model=ProcessActionResult)
async def clone_process(slug: str, req: CloneRequest, request: Request, user: dict | None = _required_user_dep):
    _verify_process_owner(slug, getattr(request.state, "user_id", ""))
    """Clone a process agent with a new name and port."""
    registry = _load_process_registry()
    entry = registry.get(slug)
    if not entry:
        raise HTTPException(404, f"Process '{slug}' not found")

    new_name = req.new_name.strip()
    if not new_name:
        raise HTTPException(400, "Name is required")
    new_slug = _slug(new_name)

    if new_slug in registry:
        if _process_is_alive(new_slug):
            raise HTTPException(400, f"Process '{new_slug}' already running.")

    old_agent_dir = DATA_DIR / slug
    new_agent_dir = DATA_DIR / new_slug

    # Create new data directory
    new_agent_dir.mkdir(parents=True, exist_ok=True)
    (new_agent_dir / "data" / "workspace").mkdir(parents=True, exist_ok=True)
    (new_agent_dir / "data" / "sessions").mkdir(parents=True, exist_ok=True)
    (new_agent_dir / "data" / "skills").mkdir(parents=True, exist_ok=True)
    (new_agent_dir / "data" / "home-config").mkdir(parents=True, exist_ok=True)

    # Copy config files
    for fname in ("config.yaml", ".env"):
        src = old_agent_dir / fname
        if src.is_file():
            shutil.copy2(str(src), str(new_agent_dir / fname))
    src_hc = old_agent_dir / "data" / "home-config" / "config.yaml"
    if src_hc.is_file():
        shutil.copy2(str(src_hc), str(new_agent_dir / "data" / "home-config" / "config.yaml"))

    # Find available port
    old_port = entry.get("web_port", 24080)
    new_port = _find_available_port(old_port + 1)

    # Update config.yaml with new port and name
    cfg_path = new_agent_dir / "config.yaml"
    if cfg_path.is_file():
        cfg_text = cfg_path.read_text()
        cfg_text = cfg_text.replace(f"port: {old_port}", f"port: {new_port}")
        old_name = entry.get("name", slug)
        if old_name:
            cfg_text = cfg_text.replace(f"instance_name: {old_name}", f"instance_name: {new_name}")
        # Update local paths to point to new agent dir
        cfg_text = cfg_text.replace(str(old_agent_dir), str(new_agent_dir))
        cfg_path.write_text(cfg_text)
        hc_path = new_agent_dir / "data" / "home-config" / "config.yaml"
        if hc_path.is_file():
            hc_path.write_text(cfg_text)

    # Register clone (but don't start it)
    registry[new_slug] = {
        "slug": new_slug,
        "name": new_name,
        "description": "",
        "web_port": new_port,
        "web_auth": entry.get("web_auth", ""),
        "pid": None,
        "provider": entry.get("provider", ""),
        "model": entry.get("model", ""),
    }
    _save_process_registry(registry)

    return ProcessActionResult(ok=True, slug=new_slug, message=f"Cloned '{slug}' → '{new_slug}' (port {new_port})")


async def _get_system_config() -> dict:
    """Load system config from DB. Returns defaults when auth disabled."""
    from captain_claw.flight_deck.admin_routes import _load_system_config, SYSTEM_CONFIG_DEFAULTS
    if not AUTH_ENABLED or not hasattr(app.state, "fd_db"):
        return {**SYSTEM_CONFIG_DEFAULTS}
    raw = await app.state.fd_db.get_system_setting("fd:system-config")
    return _load_system_config(raw)


@app.get("/fd/auth/status")
async def auth_status():
    """Check if auth is enabled and return system config flags (public endpoint)."""
    cfg = await _get_system_config()
    # When running inside a container, default docker spawn to False (no Docker socket)
    docker_default = not os.environ.get("CAPTAIN_CLAW_DOCKER")
    # Provide internal FD URL for agent-to-FD calls (inside container, use localhost)
    internal_fd_url = os.environ.get("FD_INTERNAL_URL", "")
    if not internal_fd_url and os.environ.get("CAPTAIN_CLAW_DOCKER"):
        internal_fd_url = "http://localhost:25080"
    return {
        "auth_enabled": AUTH_ENABLED,
        "docker_spawn_enabled": cfg.get("docker_spawn_enabled", docker_default),
        "internal_fd_url": internal_fd_url,
    }


# ── Agent Forge: LLM-powered team decomposition ──────────────────────────

class ForgeRequest(BaseModel):
    prompt: str
    provider: str = "anthropic"
    model: str = "claude-sonnet-4-20250514"
    api_key: str = ""
    project_id: str = ""  # optional: forge agents for an existing project


@app.post("/fd/forge")
async def forge_decompose(
    body: ForgeRequest, request: Request,
    user: dict | None = _required_user_dep,
):
    """Use an LLM to decompose a user objective into a team of specialized agents."""
    if not body.prompt.strip():
        raise HTTPException(400, "prompt is required")

    # Load the forge system prompt from instructions
    instructions_dir = Path(__file__).parent.parent / "instructions"
    system_prompt_file = instructions_dir / "forge_decompose_system_prompt.md"
    if not system_prompt_file.is_file():
        raise HTTPException(500, "Forge system prompt not found")
    system_prompt = system_prompt_file.read_text()

    # If forging for a project, augment the prompt with project context.
    user_prompt = body.prompt.strip()
    if body.project_id:
        try:
            from captain_claw.projects import get_project_manager
            pm = get_project_manager()
            if pm:
                ctx = await pm.get_project_context(body.project_id)
                if ctx:
                    project = ctx["project"]
                    project_context_parts = [
                        "\n\n## Project Context",
                        f"Project: {project['name']}",
                        f"Status: {project['status']}",
                        f"Description: {project['description']}",
                    ]
                    goals = project.get("goals", [])
                    if goals:
                        project_context_parts.append("\nGoals:")
                        for g in goals:
                            project_context_parts.append(f"- [{g.get('status', 'pending')}] {g['goal']}")
                    members = ctx.get("members", [])
                    if members:
                        project_context_parts.append("\nExisting team members:")
                        for m in members:
                            tags = ", ".join(m.get("expertise_tags", []))
                            project_context_parts.append(
                                f"- {m.get('agent_name') or m['agent_id']} ({m['role']})"
                                + (f" — {tags}" if tags else "")
                            )
                    decisions = ctx.get("decisions", [])
                    if decisions:
                        project_context_parts.append("\nPrior decisions:")
                        for d in decisions[:8]:
                            project_context_parts.append(f"- {d['title']}: {d.get('content', '')[:200]}")
                    project_context_parts.append(
                        "\nDesign the team considering existing members and prior decisions. "
                        "Do not duplicate roles already filled."
                    )
                    user_prompt += "\n".join(project_context_parts)
                    # Log activity.
                    await pm.log_activity(
                        body.project_id, "forge_run",
                        detail={"prompt": body.prompt[:200]},
                    )
        except Exception as exc:
            log.warning("Failed to load project context for forge", error=str(exc))

    # Create an LLM provider and make the decomposition call
    try:
        from captain_claw.llm import create_provider, Message
        provider = create_provider(
            provider=body.provider,
            model=body.model,
            api_key=body.api_key or None,
            temperature=0.7,
            max_tokens=16384,
        )
        response = await provider.complete(
            messages=[
                Message(role="system", content=system_prompt),
                Message(role="user", content=user_prompt),
            ],
            temperature=0.7,
            max_tokens=16384,
        )
    except Exception as e:
        log.error("Forge LLM call failed", exc_info=True)
        raise HTTPException(502, f"LLM call failed: {e}")

    # Parse the JSON response
    content = response.content.strip()
    # Strip markdown fences if present
    if content.startswith("```"):
        lines = content.split("\n")
        lines = [l for l in lines if not l.strip().startswith("```")]
        content = "\n".join(lines)
    try:
        result = json.loads(content)
    except json.JSONDecodeError:
        raise HTTPException(502, f"LLM returned invalid JSON: {content[:500]}")

    # Tag result with project_id so spawned agents can be auto-joined.
    if body.project_id:
        result["project_id"] = body.project_id

    return result


@app.get("/fd/health")
def health():
    try:
        client = get_docker()
        client.ping()
        docker_ok = True
    except Exception:
        docker_ok = False
    # Always report ok if at least the server is running
    # (processes don't need Docker)
    return {"ok": True, "docker": docker_ok, "processes": True}


# ── Old Man preset ──

OLD_MAN_TOOLS = [
    "shell", "read", "write", "glob", "edit",
    "web_fetch", "web_search", "browser",
    "pdf_extract", "docx_extract", "xlsx_extract", "pptx_extract",
    "pocket_tts", "send_mail", "clipboard",
    "screen_capture", "desktop_action",
    "scripts", "playbooks", "personality", "datastore", "insights",
    "cron", "summarize_files", "gws", "direct_api",
    "flight_deck",
]


def _build_old_man_config(
    *,
    name: str = "Old Man",
    description: str = "",
    provider: str = "ollama",
    model: str = "minimax-m2.7:cloud",
    api_key: str = "",
    base_url: str = "",
    web_port: int = 24080,
) -> AgentConfig:
    """Return an AgentConfig pre-filled for supervisor mode.

    The agent keeps the Old Man supervisor tooling (hotkey listener, fleet
    delegation), but its public identity (name/description) is taken from the
    first-run onboarding wizard so the fleet's seed agent can be named by the
    user instead of always being "Old Man".
    """
    return AgentConfig(
        name=(name or "Old Man").strip(),
        description=(description.strip() or "Desktop supervisor — triages requests, delegates to fleet agents"),
        provider=provider,
        model=model,
        provider_api_key=api_key,
        base_url=base_url.strip(),
        tools=OLD_MAN_TOOLS,
        web_port=web_port,
        # Old Man adds old_man.enabled + hotkey overrides via env var
        env_vars=[
            {"key": "CLAW_OLD_MAN__ENABLED", "value": "true"},
            {"key": "CLAW_TOOLS__SCREEN_CAPTURE__HOTKEY_ENABLED", "value": "true"},
            {"key": "FD_URL", "value": f"http://localhost:{os.environ.get('FD_PORT', '25080')}"},
        ],
    )


class OldManSpawnRequest(BaseModel):
    """Request body for the supervisor quick-spawn endpoint.

    ``name``/``description`` let the first-run onboarding wizard name the seed
    supervisor agent; both default to the classic "Old Man" identity.
    """
    name: str = "Old Man"
    description: str = ""
    provider: str = "ollama"
    model: str = "minimax-m2.7:cloud"
    api_key: str = ""
    base_url: str = ""
    web_port: int = 24080
    mode: str = "auto"  # "docker", "process", or "auto" (try docker first)


@app.post("/fd/spawn-old-man")
async def spawn_old_man(
    body: OldManSpawnRequest,
    request: Request,
    user: dict | None = _optional_user_dep,
):
    """One-click Old Man spawn — creates a supervisor agent with sane defaults.

    Tries Docker first (if available), falls back to process spawn.
    """
    config = _build_old_man_config(
        name=body.name,
        description=body.description,
        provider=body.provider,
        model=body.model,
        api_key=body.api_key,
        base_url=body.base_url,
        web_port=body.web_port,
    )

    use_docker = body.mode == "docker"
    use_process = body.mode == "process"

    if body.mode == "auto":
        # Prefer docker if available
        try:
            get_docker()
            sys_cfg = await _get_system_config()
            docker_default = not os.environ.get("CAPTAIN_CLAW_DOCKER")
            use_docker = sys_cfg.get("docker_spawn_enabled", docker_default)
            use_process = not use_docker
        except Exception:
            use_process = True

    if use_docker:
        return await spawn_agent(config, request, user)
    else:
        return await spawn_process(config, request, user)


# ── Free OpenRouter ("Freebie") agents ──

OPENROUTER_API_BASE = "https://openrouter.ai/api/v1"


def _is_free_openrouter_model(m: dict) -> bool:
    mid = str(m.get("id", ""))
    if mid.endswith(":free"):
        return True
    pricing = m.get("pricing", {}) or {}

    def _zero(v) -> bool:
        try:
            return float(v) == 0.0
        except Exception:
            return False

    return _zero(pricing.get("prompt")) and _zero(pricing.get("completion")) and _zero(pricing.get("request", "0"))


async def _fetch_openrouter_free_models() -> list[dict]:
    """Fetch OpenRouter's public model list (no key needed) and return the free
    models that support tool calling — Captain Claw agents are tool-heavy, so a
    model without ``tools`` in its supported parameters is useless here."""
    import httpx
    async with httpx.AsyncClient(timeout=20.0) as client:
        r = await client.get(f"{OPENROUTER_API_BASE}/models")
        r.raise_for_status()
        data = r.json().get("data", []) or []
    out: list[dict] = []
    for m in data:
        if not _is_free_openrouter_model(m):
            continue
        if "tools" not in (m.get("supported_parameters") or []):
            continue
        mid = str(m.get("id", ""))
        ctx = m.get("context_length") or (m.get("top_provider", {}) or {}).get("context_length") or 0
        try:
            ctx = int(ctx or 0)
        except Exception:
            ctx = 0
        out.append({"id": mid, "name": str(m.get("name", mid)), "context_length": ctx})
    out.sort(key=lambda x: x["id"])
    return out


def _free_allowed_entries(model_ids: list[str]) -> list[dict]:
    """Build config.model.allowed entries for a set of free OpenRouter model ids."""
    return [
        {"id": mid, "provider": "openrouter", "model": mid,
         "base_url": OPENROUTER_API_BASE, "model_type": "llm"}
        for mid in model_ids if mid
    ]


def _apply_free_models_to_configs(
    agent_dir: Path, *, default_model: str, model_ids: list[str], set_api_key: str | None = None,
) -> int:
    """Patch model.{provider,model,base_url,allowed} (+ api_key if given) across all
    3 config files for a freebie agent. Returns the number of files updated."""
    allowed = _free_allowed_entries(model_ids)
    config_paths = [
        agent_dir / "config.yaml",
        agent_dir / "data" / "home-config" / "config.yaml",
        agent_dir / "data" / "home-config-parent" / ".captain-claw" / "config.yaml",
    ]
    count = 0
    for cfg_path in config_paths:
        if not cfg_path.is_file():
            continue
        try:
            data = yaml.safe_load(cfg_path.read_text()) or {}
        except Exception:
            data = {}
        if not isinstance(data, dict):
            data = {}
        if not isinstance(data.get("model"), dict):
            data["model"] = {}
        data["model"]["provider"] = "openrouter"
        data["model"]["model"] = default_model
        data["model"]["base_url"] = OPENROUTER_API_BASE
        if set_api_key is not None:
            data["model"]["api_key"] = set_api_key
        data["model"]["allowed"] = allowed
        cfg_path.write_text(yaml.dump(data, default_flow_style=False, sort_keys=False, allow_unicode=True))
        count += 1
    return count


@app.get("/fd/openrouter/free-models")
async def openrouter_free_models(user: dict | None = _optional_user_dep):
    """List currently-free OpenRouter models (public list, no key required)."""
    try:
        return {"models": await _fetch_openrouter_free_models()}
    except Exception as exc:
        raise HTTPException(502, f"Failed to fetch OpenRouter models: {exc}")


class FreeAgentSpawnRequest(BaseModel):
    name: str = "Freebie"
    description: str = ""
    api_key: str = ""
    default_model: str = ""
    models: list[str] = []  # free model ids to allow (empty = fetch all free)
    web_port: int = 24080


@app.post("/fd/spawn-free")
async def spawn_free_agent(body: FreeAgentSpawnRequest, request: Request, user: dict | None = _optional_user_dep):
    """Spawn a free OpenRouter ("Freebie") process agent: default model + all free
    models as the per-session allowed list, marked freebie for later refresh."""
    if not body.api_key.strip():
        raise HTTPException(400, "An OpenRouter API key is required.")

    model_ids = [m for m in (body.models or []) if m]
    if not model_ids:
        model_ids = [m["id"] for m in await _fetch_openrouter_free_models()]
    if not model_ids:
        raise HTTPException(502, "No free OpenRouter models are available right now.")

    default_model = (body.default_model or model_ids[0]).strip()
    if default_model not in model_ids:
        model_ids = [default_model, *model_ids]

    config = AgentConfig(
        name=body.name or "Freebie",
        description=body.description or "Free OpenRouter agent",
        provider="openrouter",
        model=default_model,
        provider_api_key=body.api_key.strip(),
        base_url=OPENROUTER_API_BASE,
        allowed_models=_free_allowed_entries(model_ids),
        web_port=body.web_port,
    )

    result = await spawn_process(config, request, user)

    # Mark the agent as a freebie so the card can offer "Refresh free models".
    registry = _load_process_registry()
    if result.slug in registry:
        registry[result.slug]["freebie"] = True
        _save_process_registry(registry)

    return result


@app.post("/fd/agent-refresh-free-models/{kind}/{identifier}")
async def refresh_free_models(
    kind: str, identifier: str, request: Request, user: dict | None = _required_user_dep,
):
    """Re-fetch the current free OpenRouter models and rewrite the agent's
    model.allowed list (and default if it dropped off) across all 3 config files.
    The agent must be stopped — config is only read at startup."""
    if kind not in ("docker", "process"):
        raise HTTPException(400, "kind must be 'docker' or 'process'")

    # Guard: agent must be stopped.
    if kind == "process":
        if _process_is_alive(identifier):
            raise HTTPException(409, "Stop the agent before refreshing free models.")
    else:
        try:
            c = get_docker().containers.get(identifier)
            if c.status == "running":
                raise HTTPException(409, "Stop the agent before refreshing free models.")
        except HTTPException:
            raise
        except Exception:
            pass

    user_id = getattr(request.state, "user_id", "")
    agent_dir = _resolve_agent_dir(identifier, kind, user_id)

    free = await _fetch_openrouter_free_models()
    model_ids = [m["id"] for m in free]
    if not model_ids:
        raise HTTPException(502, "No free OpenRouter models are available right now.")

    # Keep the current default if it's still free, otherwise pick the first.
    current_default = ""
    cfg_file = agent_dir / "config.yaml"
    if cfg_file.is_file():
        try:
            current_default = ((yaml.safe_load(cfg_file.read_text()) or {}).get("model", {}) or {}).get("model", "")
        except Exception:
            current_default = ""
    default_model = current_default if current_default in model_ids else model_ids[0]

    updated = _apply_free_models_to_configs(agent_dir, default_model=default_model, model_ids=model_ids)
    return {
        "ok": True,
        "updated": updated,
        "count": len(model_ids),
        "default_model": default_model,
        "message": f"Refreshed {len(model_ids)} free models. Start the agent to use them.",
    }


# ── Static frontend serving ──

if STATIC_DIR.is_dir():
    # Serve built React assets
    app.mount("/assets", StaticFiles(directory=STATIC_DIR / "assets"), name="assets")

    # Prefixes that belong to the backend API surface. If a request
    # targets any of these and we got here, it means no real route
    # matched — return a proper 404 instead of falling through to
    # ``index.html``. Serving HTML with a 200 status for unknown API
    # paths silently breaks clients that ``resp.json()`` the body
    # (e.g. the agent's app_runner tool), producing opaque
    # "Expecting value: line 1 column 1 (char 0)" errors that look
    # like the API returned something nonsensical when really the
    # route just wasn't registered. Fail loudly here so missing /
    # stale routes are obvious.
    _API_PREFIXES = ("fd/", "api/", "ws/")

    @app.get("/{path:path}")
    async def spa_catch_all(path: str):
        """Serve the SPA — any non-API path returns index.html."""
        # Don't shadow unknown API routes with HTML.
        if path.startswith(_API_PREFIXES):
            raise HTTPException(status_code=404, detail=f"No route for /{path}")
        file = STATIC_DIR / path
        if file.is_file():
            return FileResponse(file)
        return FileResponse(STATIC_DIR / "index.html")


def main():
    """CLI entry point for Flight Deck."""
    import argparse
    import uvicorn

    parser = argparse.ArgumentParser(description="Flight Deck — Captain Claw agent management UI")
    parser.add_argument("--host", default="0.0.0.0", help="Bind host (default: 0.0.0.0)")
    parser.add_argument("--port", type=int, default=25080, help="Bind port (default: 25080)")
    parser.add_argument("--dev", action="store_true", help="Development mode (no static serving)")
    args = parser.parse_args()

    # Record the actual bound port so spawned agents get a correct FD_URL
    # callback. The FD_URL auto-injection (when an agent's env lacks it) reads
    # FD_PORT; without this it would fall back to a hardcoded 25080 guess.
    # setdefault: don't clobber an externally-provided value (e.g. a
    # reverse-proxy public port that differs from the bind port).
    os.environ.setdefault("FD_PORT", str(args.port))

    if not args.dev and not STATIC_DIR.is_dir():
        print(f"Warning: Static files not found at {STATIC_DIR}")
        print("Run 'cd flight-deck && npm run build' to build the frontend first.")
        print("Starting API-only mode (use --dev with Vite dev server).\n")

    log.info("Flight Deck starting on http://%s:%s", args.host, args.port)
    # log_config=None: keep our colored formatter from _configure_fd_logging()
    # instead of letting uvicorn slap its default handlers back on.
    # timeout_graceful_shutdown: cap the "waiting for connections to close"
    # phase on Ctrl+C. Without this, in-flight StreamingResponse handlers
    # (consult SSE, ndjson event streams) keep uvicorn waiting indefinitely
    # — the user then has to hit Ctrl+C several times to force-exit.
    uvicorn.run(
        app,
        host=args.host,
        port=args.port,
        log_config=None,
        timeout_graceful_shutdown=3,
    )


if __name__ == "__main__":
    main()
