supervisor.py

#!/usr/bin/env python3
"""
supervisor.py - Own the relay lifecycle, tmux mirrors, runtime state, and
local web UI for clcod.
"""

from __future__ import annotations

import argparse
import asyncio
import copy
import hashlib
import json
import os
import secrets
import shlex
import signal
import socket
import subprocess
import sys
import queue
import threading
import time
from http import HTTPStatus
from http.cookies import SimpleCookie
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, urlparse

import relay
import dispatcher as dispatcher_mod
from event_store import EventStore, import_transcript_to_event_store
from task_state import TaskStateManager, atomic_write_json

SCRIPT_DIR = Path(__file__).resolve().parent
WEB_DIR = SCRIPT_DIR / "web"


class ReusableHTTPServer(ThreadingHTTPServer):
    allow_reuse_address = True


def utc_now() -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def build_ui_url(config: dict[str, Any]) -> str:
    ui = config["ui"]
    return f"http://{ui['host']}:{ui['port']}"


USAGE_WINDOW_SECONDS = 5 * 60 * 60  # 5-hour rolling window
DEFAULT_USAGE_LIMIT = 50000         # default token budget per window
ROUTE_HISTORY_LIMIT = 8


def build_usage_window(agent: dict[str, Any]) -> dict[str, Any]:
    return {
        "window_start": time.time(),
        "tokens_used": 0,
        "limit": agent.get("usage_limit", DEFAULT_USAGE_LIMIT),
    }


def build_agent_state(agent: dict[str, Any]) -> dict[str, Any]:
    return {
        "state": "starting",
        "session_id": None,
        "mirror_mode": agent["mirror_mode"],
        "mirror_view": "log",
        "pane_target": None,
        "pane_command": None,
        "last_error": None,
        "last_reply_at": None,
        "selected_model": agent.get("selected_model", "default"),
        "selected_effort": agent.get("selected_effort", "default"),
        "model_options": agent.get("model_options", []),
        "effort_options": agent.get("effort_options", []),
        "effort_matrix": agent.get("effort_matrix", {}),
        "usage_window": build_usage_window(agent),
    }


def build_initial_state(config: dict[str, Any]) -> dict[str, Any]:
    session = config["tmux"]["session"]
    agents = {
        agent["name"]: build_agent_state(agent)
        for agent in config["agents"]
        if agent["enabled"]
    }
    return {
        "app": {
            "phase": "booting",
            "sleeping": False,
            "ui_url": build_ui_url(config),
            "default_sender": config["ui"]["default_sender"],
        },
        "relay": {
            "state": "starting",
            "pid": os.getpid(),
            "last_error": None,
        },
        "tmux": {
            "session": session,
            "state": "starting",
            "attach_command": f"tmux attach -t {session}",
        },
        "agents": agents,
        "project": {
            "active": None,
            "name": None,
            "path": str(SCRIPT_DIR),
        },
        "workspace": {
            "repo_path": str(SCRIPT_DIR),
            "branch": None,
            "dirty": False,
            "dirty_files": 0,
            "sync_state": "idle",
            "last_sync_at": None,
            "compact_state": "idle",
            "last_compact_at": None,
            "last_archive_path": None,
        },
        "dispatcher": {
            "state": "disabled",
            "router_model": None,
            "routes_total": 0,
            "absorbs_total": 0,
            "tokens_saved": 0,
            "pane_target": None,
        },
        "routing": {
            "active": [],
            "recent": [],
            "last_route_at": None,
        },
        "tasks": {
            "total": 0,
            "pending": 0,
            "in_progress": 0,
            "done": 0,
            "last_created_at": None,
        },
        "transcript": {
            "path": str(config["workspace"]["log_path"]),
            "last_speaker": "",
            "last_updated_at": None,
            "rev": 0,
        },
        "queue": {
            "depth": 0,
            "active": 0,
            "total_processed": 0,
            "total_failed": 0,
            "last_job_id": None,
            "last_completed_at": None,
        },
    }


def parse_transcript_entries(text: str, limit: int) -> list[dict[str, str]]:
    entries: list[dict[str, str]] = []
    tagged_speaker: str | None = None
    tagged_lines: list[str] = []

    def flush_tagged() -> None:
        nonlocal tagged_speaker, tagged_lines
        if tagged_speaker and tagged_lines:
            entry = {"speaker": tagged_speaker, "text": "\n".join(tagged_lines).strip()}
            if entry["text"]:
                entries.append(entry)
        tagged_speaker = None
        tagged_lines = []

    for line in text.splitlines():
        raw_line = line.rstrip()
        line = raw_line.strip()
        if not line:
            flush_tagged()
            continue
        if line.startswith("[") and line.endswith("]") and len(line) > 2:
            flush_tagged()
            tagged_speaker = line[1:-1].strip()
            continue
        if tagged_speaker:
            tagged_lines.append(raw_line)
            continue
        try:
            payload = json.loads(line)
            if "speaker" in payload and "text" in payload:
                entry = {"speaker": payload["speaker"], "text": payload["text"]}
                if payload.get("ts"):
                    entry["ts"] = payload["ts"]
                if payload.get("seq"):
                    entry["seq"] = payload["seq"]
                entries.append(entry)
            elif "sender" in payload and "body" in payload:
                entry = {"speaker": payload["sender"], "text": payload["body"]}
                if payload.get("ts"):
                    entry["ts"] = payload["ts"]
                if payload.get("seq"):
                    entry["seq"] = payload["seq"]
                entries.append(entry)
        except json.JSONDecodeError:
            continue
    flush_tagged()
    return entries[-limit:]


def fallback_compact_summary(text: str) -> str:
    entries = parse_transcript_entries(text, 8)
    if not entries:
        return "Conversation compacted. No transcript content was available."
    fragments: list[str] = []
    for entry in entries[-6:]:
        speaker = str(entry.get("speaker") or "UNKNOWN")
        body = relay.truncate_text(str(entry.get("text") or ""), 72)
        fragments.append(f"{speaker}: {body}")
    return relay.truncate_text(
        "Conversation compacted. Recent context: " + " | ".join(fragments),
        700,
    )


def build_log_mirror_command(agent: dict[str, Any]) -> str:
    log_path = agent["io_log_path"]
    work_dir = agent.get("work_dir") or str(SCRIPT_DIR)
    script = (
        f"mkdir -p {shlex.quote(str(log_path.parent))} && "
        f"touch {shlex.quote(str(log_path))} && "
        f"printf '%s\\n\\n' {shlex.quote(f'[{agent['name']}] live log mirror')} && "
        f"exec tail -n 120 -F {shlex.quote(str(log_path))}"
    )
    return f"cd {shlex.quote(work_dir)} && bash -lc {shlex.quote(script)}"


def build_resume_mirror_command(agent: dict[str, Any], session_id: str) -> str:
    work_dir = agent.get("work_dir") or str(SCRIPT_DIR)
    args = [
        item.format_map({"session_id": session_id, "work_dir": work_dir, "script_dir": work_dir})
        for item in agent.get("mirror_resume_args", [])
    ]
    cmd = [agent["cmd"], *relay.build_selection_args(agent), *args]
    return "cd {} && exec {}".format(
        shlex.quote(work_dir),
        " ".join(shlex.quote(part) for part in cmd),
    )


def desired_mirror_view(agent: dict[str, Any], session_id: str | None) -> str:
    if agent["mirror_mode"] == "resume" and session_id and agent.get("mirror_resume_args"):
        return "resume"
    return "log"


def infer_agent_state(
    current_state: str,
    relay_state: str,
    mirror_view: str,
    pane_command: str | None,
) -> str:
    if current_state == "error":
        return "error"
    if not pane_command:
        return "starting"
    if relay_state != "running":
        return "auth"
    if mirror_view in {"resume", "log"}:
        return "ready"
    return "warming"


def sort_routes(routes: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return sorted(
        routes,
        key=lambda item: item.get("updated_at") or item.get("started_at") or "",
        reverse=True,
    )


class StateStore:
    def __init__(self, config: dict[str, Any]) -> None:
        self.path: Path = config["workspace"]["state_path"]
        self._lock = threading.Lock()
        self.state = build_initial_state(config)
        self.write()

    def _write_locked(self) -> None:
        atomic_write_json(self.path, self.state)

    def write(self) -> None:
        with self._lock:
            self._write_locked()

    def snapshot(self) -> dict[str, Any]:
        with self._lock:
            return copy.deepcopy(self.state)

    def patch(self, section: str, values: dict[str, Any]) -> None:
        with self._lock:
            self.state[section].update(values)
            self._write_locked()

    def patch_agent(self, name: str, values: dict[str, Any]) -> None:
        with self._lock:
            self.state["agents"][name].update(values)
            self._write_locked()

    def record_agent_usage(self, name: str, tokens: int) -> None:
        """Increment token usage for an agent, resetting the window if expired."""
        with self._lock:
            agent = self.state["agents"].get(name)
            if not agent:
                return
            window = agent.setdefault("usage_window", {
                "window_start": time.time(),
                "tokens_used": 0,
                "limit": DEFAULT_USAGE_LIMIT,
            })
            now = time.time()
            if now - window["window_start"] >= USAGE_WINDOW_SECONDS:
                window["window_start"] = now
                window["tokens_used"] = 0
            window["tokens_used"] += tokens
            self._write_locked()

    def patch_tasks_summary(self, values: dict[str, Any]) -> None:
        """Persist the replayed task summary only.

        Task counters are durable/replayed state. Other sections in state.json
        remain volatile runtime materialized views owned by the supervisor.
        """
        with self._lock:
            self.state["tasks"].update(values)
            self._write_locked()

    def fuel_for_agent(self, name: str) -> dict[str, Any]:
        """Return fuel gauge data for a single agent."""
        with self._lock:
            agent = self.state["agents"].get(name, {})
            window = agent.get("usage_window", {})
            window_start = window.get("window_start", time.time())
            tokens_used = window.get("tokens_used", 0)
            limit = window.get("limit", DEFAULT_USAGE_LIMIT)
            now = time.time()
            # Auto-reset expired windows
            if now - window_start >= USAGE_WINDOW_SECONDS:
                tokens_used = 0
                window_start = now
            remaining = max(0, limit - tokens_used)
            pct = round((remaining / limit) * 100, 1) if limit > 0 else 0
            return {
                "window_start": window_start,
                "tokens_used": tokens_used,
                "limit": limit,
                "remaining": remaining,
                "pct_remaining": pct,
            }


class RuntimeSupervisor:
    def __init__(self, config: dict[str, Any]) -> None:
        self.config = config
        self.session = config["tmux"]["session"]
        self.workspace = config["workspace"]
        self.settings_lock = threading.Lock()
        self.event_store = EventStore(config["workspace"]["events_db_path"])
        self.apply_saved_preferences()
        self.state = StateStore(config)
        self.task_state = TaskStateManager(
            event_store=self.event_store,
            tasks_path=config["workspace"]["tasks_path"],
            state_store=self.state,
            event_callback=self.handle_relay_event,
        )
        self._sleeping = False
        self._sleep_lock = threading.Lock()
        self.stop_event = asyncio.Event()
        self._password_hash: str = self._hash_password(self.password())
        self.auth_tokens: set[str] = set()
        self._load_auth_tokens()
        self.rebuild_state_from_events()
        self.http_server: ThreadingHTTPServer | None = None
        self.http_thread: threading.Thread | None = None
        self.mirror_keys: dict[str, tuple[str, str | None]] = {}
        self.pane_targets: dict[str, str] = {}
        self.projects_path: Path = config["workspace"]["projects_path"]
        self._sse_clients: list[queue.Queue] = []
        self._sse_lock = threading.Lock()
        self.max_sse_subscribers: int = config.get("ui", {}).get("max_sse_subscribers", 32)

    @property
    def _auth_tokens_path(self) -> Path:
        return self.workspace["log_path"].parent / "auth_tokens.json"

    def _load_auth_tokens(self) -> None:
        data = relay.read_json(self._auth_tokens_path, [])
        if isinstance(data, list):
            self.auth_tokens = set(data)

    def _save_auth_tokens(self) -> None:
        relay.write_json(self._auth_tokens_path, list(self.auth_tokens))

    def rebuild_state_from_events(self) -> None:
        """Replay durable task state into the runtime projections on startup.

        Durable/replayed state:
        - task board contents
        - task counters and last_created_at summary

        Volatile/runtime-only state:
        - tmux pane data
        - queue and routing activity
        - live agent process metadata
        """
        self.task_state.rebuild_from_events()

    def sse_subscribe(self) -> queue.Queue | None:
        """Subscribe to SSE events. Returns None if at capacity."""
        q: queue.Queue = queue.Queue(maxsize=64)
        with self._sse_lock:
            if len(self._sse_clients) >= self.max_sse_subscribers:
                return None
            self._sse_clients.append(q)
        return q

    def sse_unsubscribe(self, q: queue.Queue) -> None:
        with self._sse_lock:
            try:
                self._sse_clients.remove(q)
            except ValueError:
                pass

    def sse_client_count(self) -> int:
        with self._sse_lock:
            return len(self._sse_clients)

    def sse_broadcast(self, event_type: str, data: dict[str, Any], event_id: int | None = None) -> None:
        payload = {"type": event_type, **data}
        with self._sse_lock:
            dead: list[queue.Queue] = []
            for q in self._sse_clients:
                try:
                    q.put_nowait({"event_id": event_id, "payload": payload})
                except queue.Full:
                    dead.append(q)
            for q in dead:
                try:
                    self._sse_clients.remove(q)
                except ValueError:
                    pass
            if dead:
                print(
                    f"[supervisor] SSE: dropped {len(dead)} full queues, "
                    f"{len(self._sse_clients)} clients remain",
                    file=sys.stderr,
                    flush=True,
                )

    def refresh_task_state(self) -> None:
        self.state.patch_tasks_summary(self.task_state.summary())

    def is_sleeping(self) -> bool:
        with self._sleep_lock:
            return self._sleeping

    def set_sleeping(self, value: bool) -> None:
        with self._sleep_lock:
            self._sleeping = value
        self.state.patch("app", {"sleeping": value})

    def password(self) -> str:
        env_name = self.config["ui"]["password_env"]
        return os.environ.get(env_name) or self.config["ui"]["password"]

    @staticmethod
    def _hash_password(plaintext: str) -> str:
        return hashlib.sha256(plaintext.encode("utf-8")).hexdigest()

    def verify_password(self, candidate: str) -> bool:
        """Timing-safe password check. Hashes both sides before comparing."""
        return secrets.compare_digest(
            self._hash_password(candidate),
            self._password_hash,
        )

    def current_repo_path(self) -> Path:
        project_path = self.state.snapshot().get("project", {}).get("path")
        if project_path:
            return Path(project_path)
        return SCRIPT_DIR

    def refresh_workspace_state(self) -> None:
        repo_path = self.current_repo_path()
        branch = None
        dirty = False
        dirty_files = 0
        try:
            branch_proc = subprocess.run(
                ["git", "rev-parse", "--abbrev-ref", "HEAD"],
                cwd=str(repo_path),
                text=True,
                capture_output=True,
                timeout=5,
            )
            if branch_proc.returncode == 0:
                branch = branch_proc.stdout.strip() or None
            status_proc = subprocess.run(
                ["git", "status", "--porcelain"],
                cwd=str(repo_path),
                text=True,
                capture_output=True,
                timeout=5,
            )
            if status_proc.returncode == 0:
                dirty_files = len([line for line in status_proc.stdout.splitlines() if line.strip()])
                dirty = dirty_files > 0
        except Exception:
            branch = None
            dirty = False
            dirty_files = 0
        self.state.patch(
            "workspace",
            {
                "repo_path": str(repo_path),
                "branch": branch,
                "dirty": dirty,
                "dirty_files": dirty_files,
            },
        )

    def emit_local_event(self, event: dict[str, Any]) -> dict[str, Any]:
        payload = dict(event)
        stored = self.event_store.append_event(payload)
        payload["event_id"] = stored["id"]
        self.handle_relay_event(payload)
        return payload

    @staticmethod
    def sse_event_payload(event: dict[str, Any]) -> dict[str, Any]:
        payload = dict(event)
        if payload.get("type") == "tasks_bulk_updated":
            payload["type"] = "tasks_updated"
        return payload

    def persist_system_message(
        self,
        text: str,
        *,
        sender: str = "SYSTEM",
        message_type: str = "system",
    ) -> dict[str, Any]:
        return relay.persist_transcript_message(
            self.workspace["log_path"],
            sender,
            text,
            event_callback=self.handle_relay_event,
            event_store=self.event_store,
            message_type=message_type,
        )

    def preferences_payload(self) -> dict[str, Any]:
        data = relay.read_json(self.workspace["preferences_path"], {"agents": {}})
        if not isinstance(data, dict):
            return {"agents": {}}
        agents = data.get("agents", {})
        if not isinstance(agents, dict):
            agents = {}
        return {"agents": agents}

    def save_preferences_payload(self, payload: dict[str, Any]) -> None:
        relay.write_json(self.workspace["preferences_path"], payload)

    def apply_saved_preferences(self) -> None:
        preferences = self.preferences_payload()
        agent_preferences = preferences.get("agents", {})
        for agent in self.config["agents"]:
            if not agent["enabled"]:
                continue
            saved = agent_preferences.get(agent["name"], {})
            if not isinstance(saved, dict):
                saved = {}
            agent["selected_model"] = relay.resolve_selected_option(
                saved.get("selected_model", agent.get("selected_model", "default")),
                agent.get("model_options", []),
                str(agent.get("selected_model", "default")),
            )
            if agent.get("effort_options"):
                selected_effort = relay.resolve_selected_option(
                    saved.get("selected_effort", agent.get("selected_effort", "default")),
                    agent["effort_options"],
                    str(agent.get("selected_effort", "default")),
                )
                allowed_efforts = set(self.allowed_efforts_for(agent, agent["selected_model"]))
                if allowed_efforts and selected_effort not in allowed_efforts:
                    selected_effort = "default" if "default" in allowed_efforts else sorted(allowed_efforts)[0]
                agent["selected_effort"] = selected_effort
            else:
                agent["selected_effort"] = "default"

    def find_agent(self, name: str) -> dict[str, Any] | None:
        target = name.strip().upper()
        for agent in self.config["agents"]:
            if agent["name"] == target and agent["enabled"]:
                return agent
        return None

    def persist_agent_preferences(self, agent: dict[str, Any]) -> None:
        preferences = self.preferences_payload()
        preferences.setdefault("agents", {})
        preferences["agents"][agent["name"]] = {
            "selected_model": agent.get("selected_model", "default"),
            "selected_effort": agent.get("selected_effort", "default"),
        }
        self.save_preferences_payload(preferences)

    def allowed_efforts_for(self, agent: dict[str, Any], selected_model: str) -> list[str]:
        matrix = agent.get("effort_matrix", {})
        if isinstance(matrix, dict):
            allowed = matrix.get(selected_model) or matrix.get("default")
            if isinstance(allowed, list) and allowed:
                return [str(item) for item in allowed]
        return [str(item["id"]) for item in agent.get("effort_options", [])]

    def restart_agent(self, name: str) -> dict[str, Any]:
        """Kill the process in an agent's pane and respawn its mirror command."""
        agent = self.find_agent(name)
        if not agent:
            raise KeyError(name)
        pane_target = self.pane_targets.get(agent["name"])
        if not pane_target:
            raise RuntimeError(f"no pane target for {name}")
        # Force-respawn the mirror (kills whatever is running in the pane)
        self.mirror_keys.pop(agent["name"], None)
        self.sync_agent_mirrors(force=True)
        return self.state.snapshot()["agents"][agent["name"]]

    # ── Project management ──────────────────────────────────────────

    def list_projects(self) -> dict[str, Any]:
        return relay.load_projects(self.projects_path)

    def lock_project(self, path: str | None = None, url: str | None = None, name: str | None = None) -> dict[str, Any]:
        """Lock agents to a local path or clone a repo and lock to it."""
        projects = relay.load_projects(self.projects_path)
        if url:
            target_name = name or url.rstrip("/").rsplit("/", 1)[-1].removesuffix(".git")
            target_dir = SCRIPT_DIR / "projects" / target_name
            if not target_dir.exists():
                target_dir.parent.mkdir(parents=True, exist_ok=True)
                subprocess.run(
                    ["git", "clone", url, str(target_dir)],
                    check=True, timeout=120, capture_output=True, text=True,
                )
            project_path = str(target_dir.resolve())
            project_type = "cloned"
        elif path:
            resolved = Path(path).expanduser().resolve()
            if not resolved.is_dir():
                raise ValueError(f"path does not exist or is not a directory: {path}")
            project_path = str(resolved)
            project_type = "local"
            target_name = name or resolved.name
        else:
            raise ValueError("either path or url is required")

        project_id = target_name.lower().replace(" ", "-")
        projects["projects"][project_id] = {
            "name": target_name,
            "path": project_path,
            "type": project_type,
            "locked_at": utc_now(),
            "agents_allowed": [a["name"] for a in self.config["agents"] if a["enabled"]],
        }
        projects["active"] = project_id
        relay.save_projects(self.projects_path, projects)

        # Inject work_dir into runtime agent configs
        for agent in self.config["agents"]:
            if agent["enabled"]:
                agent["work_dir"] = project_path
        self.reset_all_agent_sessions()
        self.sync_agent_mirrors(force=True)

        lock_body = f"[PROJECT] Locked to: {target_name} ({project_path})"
        self.persist_system_message(lock_body, message_type="project")
        self.refresh_workspace_state()
        return projects

    def unlock_project(self) -> dict[str, Any]:
        """Release current project lock, agents return to home repo."""
        projects = relay.load_projects(self.projects_path)
        projects["active"] = None
        relay.save_projects(self.projects_path, projects)

        for agent in self.config["agents"]:
            agent.pop("work_dir", None)
        self.reset_all_agent_sessions()
        self.sync_agent_mirrors(force=True)

        self.persist_system_message(
            f"[PROJECT] Unlocked — agents returned to {SCRIPT_DIR}",
            message_type="project",
        )
        self.refresh_workspace_state()
        return projects

    def delete_project(self, project_id: str) -> dict[str, Any]:
        projects = relay.load_projects(self.projects_path)
        if project_id not in projects["projects"]:
            raise KeyError(project_id)
        if projects["active"] == project_id:
            projects["active"] = None
            for agent in self.config["agents"]:
                agent.pop("work_dir", None)
            self.reset_all_agent_sessions()
            self.sync_agent_mirrors(force=True)
        del projects["projects"][project_id]
        relay.save_projects(self.projects_path, projects)
        return projects

    def reset_agent_session(self, name: str) -> None:
        sessions = relay.load_sessions(self.workspace["sessions_path"])
        if name in sessions:
            del sessions[name]
            relay.save_sessions(self.workspace["sessions_path"], sessions)
        self.mirror_keys.pop(name, None)

    def reset_all_agent_sessions(self) -> None:
        relay.save_sessions(self.workspace["sessions_path"], {})
        self.mirror_keys.clear()

    def update_agent_settings(
        self,
        name: str,
        selected_model: str | None,
        selected_effort: str | None,
    ) -> dict[str, Any]:
        agent = self.find_agent(name)
        if not agent:
            raise KeyError(name)

        with self.settings_lock:
            next_model = relay.resolve_selected_option(
                selected_model or agent.get("selected_model", "default"),
                agent.get("model_options", []),
                str(agent.get("selected_model", "default")),
            )
            if agent.get("effort_options"):
                next_effort = relay.resolve_selected_option(
                    selected_effort or agent.get("selected_effort", "default"),
                    agent["effort_options"],
                    str(agent.get("selected_effort", "default")),
                )
                allowed_efforts = set(self.allowed_efforts_for(agent, next_model))
                if allowed_efforts and next_effort not in allowed_efforts:
                    next_effort = "default" if "default" in allowed_efforts else sorted(allowed_efforts)[0]
            else:
                next_effort = "default"

            changed = (
                next_model != agent.get("selected_model")
                or next_effort != agent.get("selected_effort")
            )
            agent["selected_model"] = next_model
            agent["selected_effort"] = next_effort
            self.persist_agent_preferences(agent)
            if changed:
                self.reset_agent_session(agent["name"])

        self.state.patch_agent(
            agent["name"],
            {
                "selected_model": agent["selected_model"],
                "selected_effort": agent["selected_effort"],
                "session_id": None if changed else self.state.snapshot()["agents"][agent["name"]].get("session_id"),
                "mirror_view": "log" if changed else self.state.snapshot()["agents"][agent["name"]].get("mirror_view", "log"),
                "model_options": agent.get("model_options", []),
                "effort_options": agent.get("effort_options", []),
                "effort_matrix": agent.get("effort_matrix", {}),
            },
        )
        if changed:
            self.sync_agent_mirrors(force=True)
        return self.state.snapshot()["agents"][agent["name"]]

    def tmux(self, *args: str, capture: bool = False, check: bool = True) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            ["tmux", *args],
            cwd=str(SCRIPT_DIR),
            text=True,
            capture_output=capture,
            check=check,
        )

    def tmux_session_exists(self) -> bool:
        result = subprocess.run(
            ["tmux", "has-session", "-t", self.session],
            cwd=str(SCRIPT_DIR),
            text=True,
            capture_output=True,
        )
        return result.returncode == 0

    def prepare_runtime(self) -> None:
        for path in (
            self.workspace["log_path"],
            self.workspace["relay_log_path"],
            self.workspace["state_path"],
            self.workspace["sessions_path"],
            self.workspace["preferences_path"],
            self.workspace["projects_path"],
            self.workspace["tasks_path"],
            self.workspace["events_db_path"],
        ):
            path.parent.mkdir(parents=True, exist_ok=True)
        self.workspace["archives_dir"].mkdir(parents=True, exist_ok=True)
        self.workspace["log_path"].touch(exist_ok=True)
        if not self.workspace["sessions_path"].exists():
            relay.save_sessions(self.workspace["sessions_path"], {})
        if not self.workspace["preferences_path"].exists():
            self.save_preferences_payload({"agents": {}})
        for agent in self.config["agents"]:
            if agent["enabled"]:
                agent["io_log_path"].parent.mkdir(parents=True, exist_ok=True)
                agent["io_log_path"].touch(exist_ok=True)
        relay.write_text(self.workspace["pid_path"], f"{os.getpid()}\n")
        import_transcript_to_event_store(self.event_store, self.workspace["log_path"])
        self.refresh_workspace_state()

    def ensure_tmux_layout(self) -> None:
        if self.tmux_session_exists():
            self.tmux("kill-session", "-t", self.session, check=False)

        self.tmux("new-session", "-d", "-s", self.session, "-x", "220", "-y", "60")
        # Window 0: transcript log watcher
        self.tmux("rename-window", "-t", f"{self.session}:0", "log")
        self.tmux("set-window-option", "-t", f"{self.session}:0", "remain-on-exit", "on")
        self.tmux(
            "respawn-pane",
            "-k",
            "-t",
            f"{self.session}:0.0",
            f"cd {shlex.quote(str(SCRIPT_DIR))} && exec bash watch-log.sh --config {shlex.quote(str(self.config['config_path']))}",
        )

        # One dedicated full-screen window per enabled agent
        enabled_agents = [agent for agent in self.config["agents"] if agent["enabled"]]
        for agent in enabled_agents:
            name = agent["name"]
            self.tmux("new-window", "-d", "-t", self.session, "-n", name)
            pane_target = f"{self.session}:{name}.0"
            self.pane_targets[name] = pane_target
            self.state.patch_agent(
                name,
                {
                    "pane_target": pane_target,
                    "mirror_mode": agent["mirror_mode"],
                    "mirror_view": "log",
                    "selected_model": agent.get("selected_model", "default"),
                    "selected_effort": agent.get("selected_effort", "default"),
                    "model_options": agent.get("model_options", []),
                    "effort_options": agent.get("effort_options", []),
                    "effort_matrix": agent.get("effort_matrix", {}),
                },
            )

        # Dispatcher monitor window — shows Ollama router status
        dispatcher_script = (
            "echo '[DISPATCHER] Ollama router monitor'; "
            "while true; do "
            "echo \"--- $(date) ---\"; "
            "curl -s http://localhost:11434/api/tags 2>/dev/null "
            "| python3 -c \""
            "import json,sys; d=json.load(sys.stdin); "
            "[print(' ', m['name']) for m in d.get('models',[])]"
            "\" 2>/dev/null || echo '  Ollama not reachable'; "
            "sleep 10; "
            "done"
        )
        dispatcher_pane = f"{self.session}:DISPATCHER.0"
        self.tmux(
            "new-window", "-d", "-t", self.session, "-n", "DISPATCHER",
            f"cd {shlex.quote(str(SCRIPT_DIR))} && bash -lc {shlex.quote(dispatcher_script)}",
        )
        self.pane_targets["DISPATCHER"] = dispatcher_pane
        self.state.patch("dispatcher", {"pane_target": dispatcher_pane})

        # Runtime log window
        self.tmux(
            "new-window",
            "-d",
            "-t",
            self.session,
            "-n",
            "runtime",
            f"cd {shlex.quote(str(SCRIPT_DIR))} && touch {shlex.quote(str(self.workspace['relay_log_path']))} && exec tail -n 120 -F {shlex.quote(str(self.workspace['relay_log_path']))}",
        )
        self.state.patch("tmux", {"state": "running"})
        self.sync_agent_mirrors(force=True)

    def sync_agent_mirrors(self, force: bool = False) -> None:
        sessions = relay.load_sessions(self.workspace["sessions_path"])
        for agent in [item for item in self.config["agents"] if item["enabled"]]:
            name = agent["name"]
            pane_target = self.pane_targets.get(name)
            if not pane_target:
                continue
            session_id = sessions.get(name)
            mirror_view = desired_mirror_view(agent, session_id)
            mirror_key = (mirror_view, session_id)
            if force or self.mirror_keys.get(name) != mirror_key:
                if mirror_view == "resume" and session_id:
                    cmd = build_resume_mirror_command(agent, session_id)
                else:
                    cmd = build_log_mirror_command(agent)
                self.tmux("respawn-pane", "-k", "-t", pane_target, cmd)
                self.mirror_keys[name] = mirror_key
            self.state.patch_agent(
                name,
                {
                    "session_id": session_id,
                    "mirror_view": mirror_view,
                    "pane_target": pane_target,
                    "mirror_mode": agent["mirror_mode"],
                    "selected_model": agent.get("selected_model", "default"),
                    "selected_effort": agent.get("selected_effort", "default"),
                    "model_options": agent.get("model_options", []),
                    "effort_options": agent.get("effort_options", []),
                    "effort_matrix": agent.get("effort_matrix", {}),
                },
            )

    def collect_pane_commands(self) -> dict[str, str]:
        if not self.tmux_session_exists():
            return {}
        # Query all windows in the session (each agent now has its own window).
        # Map both the raw tmux pane id (%7) and window-style targets
        # (triagent:CODEX.0) so health checks keep working across layout changes.
        result = self.tmux(
            "list-panes",
            "-s",
            "-t",
            self.session,
            "-F",
            "#{pane_id}\t#{session_name}:#{window_name}.#{pane_index}\t#{session_name}:#{window_index}.#{pane_index}\t#{pane_current_command}",
            capture=True,
        )
        pane_commands: dict[str, str] = {}
        for raw_line in result.stdout.splitlines():
            line = raw_line.strip()
            if not line:
                continue
            parts = line.split("\t", 3)
            if len(parts) != 4:
                continue
            pane_id, named_target, indexed_target, pane_command = parts
            normalized_command = pane_command.strip()
            for key in (pane_id.strip(), named_target.strip(), indexed_target.strip()):
                if key:
                    pane_commands[key] = normalized_command
        return pane_commands

    def _send_to_socket(self, message: dict[str, Any]) -> bool:
        """Send a message to the relay socket. Returns True if sent successfully."""
        socket_path = self.workspace.get("socket_path")
        if not socket_path:
            return False
        try:
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            try:
                sock.connect(str(socket_path))
                sock.sendall(json.dumps(message).encode("utf-8"))
                return True
            finally:
                sock.close()
        except Exception as e:
            print(f"[supervisor] socket error: {e}", file=sys.stderr)
            return False

    def _pick_compact_agent(self) -> str:
        """Pick the agent with the most remaining fuel capacity. Tie-break: alphabetical."""
        best_name = ""
        best_remaining = -1
        for name in sorted(self.state.snapshot()["agents"]):
            fuel = self.state.fuel_for_agent(name)
            if fuel["remaining"] > best_remaining:
                best_remaining = fuel["remaining"]
                best_name = name
        return best_name

    def compact_context(self) -> dict[str, Any]:
        """Archive transcript context locally and inject a compact summary without dispatching."""
        self.state.patch("workspace", {"compact_state": "running"})
        cleared: list[str] = []
        now = utc_now()
        log_path = self.workspace["log_path"]
        transcript = relay.read_text(log_path) if log_path.exists() else ""
        entries = parse_transcript_entries(transcript, 500)
        archive_stamp = time.strftime("%Y%m%d-%H%M%S", time.gmtime())
        archive_path = self.workspace["archives_dir"] / f"transcript-{archive_stamp}.json"

        for path in [self.workspace["relay_log_path"]] + [
            a["io_log_path"] for a in self.config["agents"] if a.get("io_log_path")
        ]:
            try:
                relay.write_text(Path(path), "")
                cleared.append(str(path))
            except OSError:
                pass

        runtime_dir = self.workspace["log_path"].parent
        for tmp in runtime_dir.glob("**/*.tmp"):
            tmp.unlink(missing_ok=True)
            cleared.append(str(tmp))

        dispatcher_config = self.config.get("dispatcher", {})
        summary = ""
        if transcript.strip() and dispatcher_config.get("enabled", False):
            try:
                summary = asyncio.run(dispatcher_mod.summarize_context(transcript[-12000:], dispatcher_config))
            except Exception:
                summary = ""
        if not summary:
            summary = fallback_compact_summary(transcript)
        summary = relay.truncate_text(summary, 800)

        archive_payload = {
            "created_at": now,
            "transcript_path": str(log_path),
            "project": self.state.snapshot().get("project", {}),
            "entries": entries,
            "summary": summary,
            "transcript": transcript,
        }
        relay.write_json(archive_path, archive_payload)
        self.emit_local_event(
            {
                "type": "transcript_compacted",
                "archive_path": str(archive_path),
                "created_at": now,
                "entry_count": len(entries),
                "char_count": len(transcript),
                "status": "archived",
            }
        )

        relay.write_text(log_path, "")
        summary_message = self.persist_system_message(
            f"[COMPACT SUMMARY] {summary}",
            message_type="compact_summary",
        )
        self.emit_local_event(
            {
                "type": "transcript_summary_inserted",
                "archive_path": str(archive_path),
                "created_at": now,
                "summary": summary,
                "message_id": summary_message.get("id"),
                "status": "inserted",
            }
        )

        self.state.patch(
            "workspace",
            {
                "compact_state": "archived",
                "last_compact_at": now,
                "last_archive_path": str(archive_path),
            },
        )
        return {
            "ok": True,
            "cleared": cleared,
            "injected": True,
            "archive_path": str(archive_path),
            "state": self.state.snapshot(),
        }

    def refresh_transcript_state(self) -> None:
        log_path = self.workspace["log_path"]
        text = relay.read_text(log_path) if log_path.exists() else ""
        self.state.patch(
            "transcript",
            {
                "last_speaker": relay.last_speaker(text),
                "last_updated_at": utc_now() if text else None,
            },
        )

    def refresh_tmux_state(self) -> None:
        if not self.tmux_session_exists():
            self.state.patch("tmux", {"state": "missing"})
            self.state.patch("app", {"phase": "error"})
            return

        self.state.patch("tmux", {"state": "running"})
        self.sync_agent_mirrors(force=False)
        pane_commands = self.collect_pane_commands()
        snapshot = self.state.snapshot()
        relay_state = snapshot["relay"]["state"]
        for name, agent_state in snapshot["agents"].items():
            pane_target = agent_state.get("pane_target")
            pane_command = pane_commands.get(pane_target or "")
            inferred_state = infer_agent_state(
                current_state=agent_state["state"],
                relay_state=relay_state,
                mirror_view=agent_state.get("mirror_view", "log"),
                pane_command=pane_command,
            )
            values = {"pane_command": pane_command, "state": inferred_state}
            if pane_command is None and agent_state["state"] != "error":
                values["state"] = "starting"
            self.state.patch_agent(name, values)

        app_phase = "booting"
        if snapshot["relay"]["state"] == "error":
            app_phase = "error"
        elif snapshot["relay"]["state"] == "running" and self.tmux_session_exists():
            app_phase = "ready"
        self.state.patch("app", {"phase": app_phase})

    def handle_relay_event(self, event: dict[str, Any]) -> None:
        event_id = event.get("event_id")
        if event["type"] == "relay_state":
            self.state.patch(
                "relay",
                {
                    "state": event["state"],
                    "pid": os.getpid(),
                    "last_error": event.get("last_error"),
                },
            )
            self.sse_broadcast("relay_state", {"state": event["state"]}, event_id)
            return

        if event["type"] == "transcript":
            current_rev = self.state.state.get("transcript", {}).get("rev", 0)
            self.state.patch(
                "transcript",
                {
                    "last_speaker": event.get("last_speaker", ""),
                    "last_updated_at": event.get("last_updated_at"),
                    "rev": current_rev + 1,
                },
            )
            # Track usage: use character count as token proxy (~4 chars ≈ 1 token)
            speaker = event.get("last_speaker", "").strip().upper()
            char_count = event.get("char_count", 0)
            if speaker and speaker in self.state.snapshot()["agents"] and char_count > 0:
                estimated_tokens = max(1, char_count // 4)
                self.state.record_agent_usage(speaker, estimated_tokens)
            msg = event.get("message")
            self.sse_broadcast(
                "transcript",
                {
                    "last_speaker": event.get("last_speaker", ""),
                    "rev": current_rev + 1,
                    **({"message": msg} if msg else {}),
                },
                event_id,
            )
            return

        if event["type"] == "agent_state":
            values = {
                "state": event["state"],
                "last_error": event.get("last_error"),
            }
            if "session_id" in event:
                values["session_id"] = event["session_id"]
            if "last_reply_at" in event:
                values["last_reply_at"] = event["last_reply_at"]
                values["last_activity_ts"] = event["last_reply_at"]
            if "tokens_delta" in event:
                values["tokens_delta"] = event["tokens_delta"]
                # Accumulate reply tokens into the agent's fuel gauge
                tokens = event["tokens_delta"]
                if tokens > 0:
                    self.state.record_agent_usage(event["agent"], tokens)

            # ── Pressure metrics ──
            agent_name = event.get("agent", "")
            now = time.time()
            snapshot = self.state.snapshot()
            agent_snap = snapshot.get("agents", {}).get(agent_name, {})
            pressure = agent_snap.get("pressure", {
                "queue_depth": 0,
                "last_latency_ms": 0,
                "tokens_per_sec": 0,
                "error_rate_5m": 0,
                "dispatch_ts": 0,
            })
            if event["state"] == "warming":
                pressure["dispatch_ts"] = now
                pressure["queue_depth"] = pressure.get("queue_depth", 0) + 1
            elif event["state"] == "ready":
                dispatch_ts = pressure.get("dispatch_ts", 0)
                if dispatch_ts > 0:
                    latency_ms = int((now - dispatch_ts) * 1000)
                    pressure["last_latency_ms"] = latency_ms
                    # Rolling average tokens/sec from latency + token delta
                    td = event.get("tokens_delta", 0)
                    elapsed = max(0.1, now - dispatch_ts)
                    if td > 0:
                        new_tps = td / elapsed
                        old_tps = pressure.get("tokens_per_sec", 0)
                        pressure["tokens_per_sec"] = round(old_tps * 0.6 + new_tps * 0.4, 1)
                pressure["queue_depth"] = max(0, pressure.get("queue_depth", 0) - 1)
                pressure["dispatch_ts"] = 0
            elif event["state"] == "error":
                pressure["queue_depth"] = max(0, pressure.get("queue_depth", 0) - 1)
                # Track errors in a simple counter (decay handled in JS)
                pressure["error_rate_5m"] = pressure.get("error_rate_5m", 0) + 1
            values["pressure"] = pressure

            self.state.patch_agent(event["agent"], values)
            self.sse_broadcast(
                "agent_state",
                {
                    "agent": agent_name,
                    "state": event["state"],
                    "last_error": event.get("last_error"),
                },
                event_id,
            )
            return

        if event["type"] == "dispatcher":
            action = event.get("action", "route")
            snapshot = self.state.snapshot()
            dispatcher_state = snapshot.get("dispatcher", {})
            dispatcher_config = self.config.get("dispatcher", {})
            routes = dispatcher_state.get("routes_total", 0)
            absorbs = dispatcher_state.get("absorbs_total", 0)
            tokens_saved = dispatcher_state.get("tokens_saved", 0)
            if action == "absorb":
                absorbs += 1
                tokens_saved += 500  # estimated tokens saved per absorbed message
            elif action == "route":
                routes += 1
                # Calculate tokens saved by not broadcasting to all agents
                all_agents = len([a for a in self.config["agents"] if a["enabled"]])
                routed = len(event.get("targets", []))
                if routed < all_agents:
                    tokens_saved += (all_agents - routed) * 200
            self.state.patch("dispatcher", {
                "state": "active",
                "router_model": dispatcher_config.get("router_model"),
                "routes_total": routes,
                "absorbs_total": absorbs,
                "tokens_saved": tokens_saved,
                "last_action": action,
                "last_targets": event.get("targets", []),
            })
            self.sse_broadcast(
                "dispatcher",
                {
                    "action": action,
                    "routes_total": routes,
                    "absorbs_total": absorbs,
                    "tokens_saved": tokens_saved,
                },
                event_id,
            )
            return

        if event["type"] == "route_state":
            snapshot = self.state.snapshot().get("routing", {})
            route = {key: value for key, value in event.items() if key != "type"}
            route_id = route.get("route_id")
            active = [
                item for item in snapshot.get("active", [])
                if item.get("route_id") != route_id
            ]
            recent = [
                item for item in snapshot.get("recent", [])
                if item.get("route_id") != route_id
            ]

            status = route.get("status", "transmitting")
            if status in {"complete", "error"}:
                route.setdefault("completed_at", route.get("updated_at"))
                recent = sort_routes([route, *recent])[:ROUTE_HISTORY_LIMIT]
            else:
                active = sort_routes([route, *active])[:ROUTE_HISTORY_LIMIT]

            routing_state = {
                "active": active,
                "recent": recent,
                "last_route_at": route.get("updated_at") or route.get("started_at"),
            }
            self.state.patch("routing", routing_state)
            self.sse_broadcast(
                "route_state",
                {
                    "route": route,
                    **routing_state,
                },
                event_id,
            )
            return

        if event["type"] == "task_created":
            self.sse_broadcast("task_created", event.get("task", {}), event_id)
            return

        if event["type"] == "task_updated":
            self.sse_broadcast("task_updated", {"task": event.get("task", {})}, event_id)
            return

        if event["type"] == "tasks_updated":
            self.sse_broadcast(
                "tasks_updated",
                {
                    "tasks": event.get("tasks", []),
                    "new_status": event.get("new_status", ""),
                },
                event_id,
            )
            return

        if event["type"] == "tasks_cleared":
            self.sse_broadcast("tasks_cleared", {"tasks": []}, event_id)
            return

        if event["type"] == "dispatch_skipped":
            self.state.patch("relay", {"last_error": event.get("reason")})
            self.sse_broadcast(
                "dispatch_skipped",
                {
                    "reason": event.get("reason"),
                    "targets": event.get("targets", []),
                    "message_id": event.get("message_id"),
                    "batch_ids": event.get("batch_ids", []),
                },
                event_id,
            )
            return

        if event["type"] == "dispatch_queued":
            depth = event.get("queue_depth", self.event_store.queue_depth())
            self.state.patch("queue", {"depth": depth, "last_job_id": event.get("job_id")})
            self.sse_broadcast(
                "dispatch_queued",
                {
                    "job_id": event.get("job_id"),
                    "targets": event.get("targets", []),
                    "sender": event.get("sender"),
                    "message_kind": event.get("message_kind"),
                    "queue_depth": depth,
                },
                event_id,
            )
            return

        if event["type"] == "dispatch_started":
            snapshot = self.state.snapshot().get("queue", {})
            depth = event.get("queue_depth", max(0, snapshot.get("depth", 1) - 1))
            self.state.patch("queue", {
                "depth": depth,
                "active": snapshot.get("active", 0) + 1,
            })
            self.sse_broadcast(
                "dispatch_started",
                {
                    "job_id": event.get("job_id"),
                    "targets": event.get("targets", []),
                    "sender": event.get("sender"),
                },
                event_id,
            )
            return

        if event["type"] == "dispatch_completed":
            snapshot = self.state.snapshot().get("queue", {})
            depth = event.get("queue_depth", max(0, snapshot.get("depth", 0)))
            self.state.patch("queue", {
                "depth": depth,
                "active": max(0, snapshot.get("active", 1) - 1),
                "total_processed": snapshot.get("total_processed", 0) + 1,
                "last_job_id": event.get("job_id"),
                "last_completed_at": utc_now(),
            })
            self.sse_broadcast(
                "dispatch_completed",
                {
                    "job_id": event.get("job_id"),
                    "targets": event.get("targets", []),
                    "queue_depth": depth,
                },
                event_id,
            )
            return

        if event["type"] == "dispatch_failed":
            snapshot = self.state.snapshot().get("queue", {})
            depth = event.get("queue_depth", snapshot.get("depth", 0))
            self.state.patch("queue", {
                "depth": depth,
                "active": max(0, snapshot.get("active", 1) - 1),
                "total_failed": snapshot.get("total_failed", 0) + 1,
            })
            self.sse_broadcast(
                "dispatch_failed",
                {
                    "job_id": event.get("job_id"),
                    "targets": event.get("targets", []),
                    "error": event.get("error"),
                },
                event_id,
            )
            return

        if event["type"] == "transcript_compacted":
            self.state.patch(
                "workspace",
                {
                    "compact_state": event.get("status", "archived"),
                    "last_compact_at": event.get("created_at"),
                    "last_archive_path": event.get("archive_path"),
                },
            )
            self.sse_broadcast(
                "transcript_compacted",
                {
                    "archive_path": event.get("archive_path"),
                    "entry_count": event.get("entry_count", 0),
                },
                event_id,
            )
            return

        if event["type"] == "transcript_summary_inserted":
            self.sse_broadcast(
                "transcript_summary_inserted",
                {
                    "archive_path": event.get("archive_path"),
                    "message_id": event.get("message_id"),
                },
                event_id,
            )
            return

        if event["type"] == "token":
            # Broadcast-only: token deltas must NOT be persisted via emit_event /
            # event_store.append_event.  Persisting them would corrupt Last-Event-ID
            # replay — the replay must reconstruct the full reply from the single
            # consolidated transcript event, not from thousands of delta fragments.
            self.sse_broadcast(
                "token",
                {
                    "agent": event.get("agent", ""),
                    "delta": event.get("delta", ""),
                    "seq": event.get("seq", 0),
                },
                # Pass None for event_id so the SSE frame carries no id: line —
                # clients must not track streaming deltas as their Last-Event-ID.
                None,
            )
            return

    def make_handler(self) -> type[BaseHTTPRequestHandler]:
        supervisor = self

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, format: str, *args: Any) -> None:
                return

            def _cookie_token(self) -> str | None:
                raw = self.headers.get("Cookie")
                if not raw:
                    return None
                cookie = SimpleCookie()
                cookie.load(raw)
                morsel = cookie.get("clcod_session")
                return morsel.value if morsel else None

            def _authorized(self) -> bool:
                token = self._cookie_token()
                return bool(token and token in supervisor.auth_tokens)

            def _json(self, payload: Any, status: int = HTTPStatus.OK, headers: dict[str, str] | None = None) -> None:
                body = json.dumps(payload).encode("utf-8")
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(body)))
                if headers:
                    for key, value in headers.items():
                        self.send_header(key, value)
                self.end_headers()
                self.wfile.write(body)

            def _payload(self) -> dict[str, Any]:
                try:
                    length = int(self.headers.get("Content-Length", "0"))
                except ValueError:
                    length = 0
                raw = self.rfile.read(length).decode("utf-8") if length else ""
                parsed = json.loads(raw or "{}")
                return parsed if isinstance(parsed, dict) else {}

            def _file(self, path: Path, content_type: str) -> None:
                if not path.exists():
                    self.send_error(HTTPStatus.NOT_FOUND)
                    return
                body = path.read_bytes()
                self.send_response(HTTPStatus.OK)
                self.send_header("Content-Type", content_type)
                self.send_header("Content-Length", str(len(body)))
                self.send_header("Cache-Control", "no-cache, no-store, must-revalidate")
                self.send_header("Pragma", "no-cache")
                self.send_header("Expires", "0")
                self.end_headers()
                self.wfile.write(body)

            def do_GET(self) -> None:
                parsed = urlparse(self.path)
                if parsed.path == "/":
                    return self._file(WEB_DIR / "index.html", "text/html; charset=utf-8")
                if parsed.path == "/app.js":
                    return self._file(WEB_DIR / "app.js", "application/javascript; charset=utf-8")
                if parsed.path == "/styles.css":
                    return self._file(WEB_DIR / "styles.css", "text/css; charset=utf-8")
                if parsed.path == "/api/state":
                    if not self._authorized():
                        return self._json({"locked": True, "app": {"phase": "locked"}})
                    supervisor.refresh_workspace_state()
                    snapshot = supervisor.state.snapshot()
                    for agent_name in snapshot.get("agents", {}):
                        snapshot["agents"][agent_name]["fuel"] = supervisor.state.fuel_for_agent(agent_name)
                    snapshot["events"] = {"latest_id": supervisor.event_store.latest_event_id()}
                    snapshot["queue"] = {
                        **snapshot.get("queue", {}),
                        "depth": supervisor.event_store.queue_depth(),
                        "active": supervisor.event_store.active_dispatch_count(),
                    }
                    return self._json(snapshot)
                if parsed.path == "/api/transcript":
                    if not self._authorized():
                        return self._json({"error": "locked"}, status=HTTPStatus.UNAUTHORIZED)
                    query = parse_qs(parsed.query)
                    limit = int(query.get("limit", ["120"])[0])
                    transcript = relay.read_text(supervisor.workspace["log_path"])
                    current_rev = supervisor.state.state.get("transcript", {}).get("rev", 0)
                    return self._json({"entries": parse_transcript_entries(transcript, max(1, min(limit, 500))), "rev": current_rev})
                if parsed.path == "/api/projects":
                    if not self._authorized():
                        return self._json({"error": "locked"}, status=HTTPStatus.UNAUTHORIZED)
                    return self._json(supervisor.list_projects())
                if parsed.path == "/api/tasks":
                    if not self._authorized():
                        return self._json({"error": "locked"}, status=HTTPStatus.UNAUTHORIZED)
                    query = parse_qs(parsed.query)
                    status_filter = query.get("status", [None])[0]
                    tasks_list = supervisor.task_state.list_tasks(status_filter=status_filter)
                    return self._json({"tasks": tasks_list[-100:]})
                if parsed.path == "/api/dispatcher/health":
                    if not self._authorized():
                        return self._json({"error": "locked"}, status=HTTPStatus.UNAUTHORIZED)
                    dispatcher_config = supervisor.config.get("dispatcher", {})
                    host = dispatcher_config.get("ollama_host", "http://localhost:11434")
                    try:
                        health = asyncio.run(dispatcher_mod.health_check(host))
                    except Exception:
                        health = {"available": False, "models": []}
                    snapshot = supervisor.state.snapshot()
                    d = snapshot.get("dispatcher", {})
                    return self._json({
                        "ok": True,
                        "available": health.get("available", False),
                        "models": health.get("models", []),
                        "router_model": dispatcher_config.get("router_model"),
                        "state": d.get("state", "disabled"),
                        "routes_total": d.get("routes_total", 0),
                        "absorbs_total": d.get("absorbs_total", 0),
                        "tokens_saved": d.get("tokens_saved", 0),
                        "last_action": d.get("last_action"),
                        "last_targets": d.get("last_targets", []),
                    })
                if parsed.path.startswith("/api/agents/") and parsed.path.endswith("/logs"):
                    parts = [p for p in parsed.path.split("/") if p]
                    if len(parts) != 4:
                        self.send_error(HTTPStatus.NOT_FOUND)
                        return
                    if not self._authorized():
                        return self._json({"error": "locked"}, status=HTTPStatus.UNAUTHORIZED)
                    agent_name = parts[2]
                    query = parse_qs(parsed.query)
                    tail = min(int(query.get("tail", ["30"])[0]), 200)
                    agent = supervisor.find_agent(agent_name)
                    if not agent:
                        return self._json({"ok": False, "error": "unknown agent"}, status=HTTPStatus.NOT_FOUND)
                    io_log_path = agent.get("io_log_path")
                    lines: list[str] = []
                    if io_log_path and Path(io_log_path).exists():
                        text = relay.read_text(Path(io_log_path))
                        lines = text.splitlines()[-tail:]
                    snapshot = supervisor.state.snapshot()
                    agent_state = snapshot.get("agents", {}).get(agent["name"], {})
                    return self._json({
                        "ok": True,
                        "agent": agent["name"],
                        "lines": lines,
                        "state": agent_state,
                    })
                if parsed.path == "/api/events":
                    if not self._authorized():
                        self.send_error(HTTPStatus.UNAUTHORIZED)
                        return
                    query = parse_qs(parsed.query)
                    after_id = int(query.get("after_id", ["0"])[0] or "0")
                    if after_id > 0 or query.get("format", [""])[0] == "json":
                        events = supervisor.event_store.list_events(
                            after_id=after_id,
                            limit=min(int(query.get("limit", ["200"])[0]), 1000),
                        )
                        return self._json(
                            {
                                "events": events,
                                "latest_id": supervisor.event_store.latest_event_id(),
                            }
                        )
                    q = supervisor.sse_subscribe()
                    if q is None:
                        self.send_error(
                            HTTPStatus.SERVICE_UNAVAILABLE,
                            "Too many SSE subscribers",
                        )
                        return
                    self.send_response(HTTPStatus.OK)
                    self.send_header("Content-Type", "text/event-stream")
                    self.send_header("Cache-Control", "no-cache, no-store, must-revalidate")
                    self.send_header("Connection", "keep-alive")
                    self.send_header("X-Accel-Buffering", "no")
                    self.end_headers()
                    try:
                        supervisor.refresh_workspace_state()
                        snapshot = supervisor.state.snapshot()
                        for agent_name in snapshot.get("agents", {}):
                            snapshot["agents"][agent_name]["fuel"] = supervisor.state.fuel_for_agent(agent_name)
                        snapshot["events"] = {"latest_id": supervisor.event_store.latest_event_id()}
                        init_data = json.dumps({"type": "init", **snapshot})
                        self.wfile.write(f"data: {init_data}\n\n".encode())
                        self.wfile.flush()
                        last_event_id = self.headers.get("Last-Event-ID")
                        if last_event_id and last_event_id.isdigit():
                            replay = supervisor.event_store.list_events(after_id=int(last_event_id), limit=500)
                            for item in replay:
                                event_id = item.get("id")
                                payload = json.dumps(supervisor.sse_event_payload(item))
                                if event_id is not None:
                                    self.wfile.write(f"id: {event_id}\n".encode())
                                self.wfile.write(f"data: {payload}\n\n".encode())
                            self.wfile.flush()
                        while True:
                            try:
                                frame = q.get(timeout=15)
                                payload = json.dumps(frame["payload"])
                                if frame.get("event_id") is not None:
                                    self.wfile.write(f"id: {frame['event_id']}\n".encode())
                                self.wfile.write(f"data: {payload}\n\n".encode())
                                self.wfile.flush()
                            except queue.Empty:
                                self.wfile.write(b": keepalive\n\n")
                                self.wfile.flush()
                    except (BrokenPipeError, ConnectionResetError, OSError):
                        pass
                    finally:
                        supervisor.sse_unsubscribe(q)
                    return
                self.send_error(HTTPStatus.NOT_FOUND)

            def do_POST(self) -> None:
                parsed = urlparse(self.path)
                if parsed.path == "/api/unlock":
                    payload = self._payload()
                    if not supervisor.verify_password(str(payload.get("password", ""))):
                        return self._json({"ok": False, "error": "invalid password"}, status=HTTPStatus.UNAUTHORIZED)

                    token = secrets.token_urlsafe(24)
                    supervisor.auth_tokens.add(token)
                    supervisor._save_auth_tokens()
                    headers = {
                        "Set-Cookie": f"clcod_session={token}; Path=/; HttpOnly; SameSite=Strict; Max-Age=604800"
                    }
                    return self._json({"ok": True, "state": supervisor.state.snapshot()}, headers=headers)

                if not self._authorized():
                    return self._json({"error": "locked"}, status=HTTPStatus.UNAUTHORIZED)

                if parsed.path == "/api/chat":
                    payload = self._payload()
                    raw_name = str(payload.get("name") or supervisor.config["ui"]["default_sender"]).strip()
                    raw_message = str(payload.get("message") or "").strip()
                    if not raw_name:
                        return self._json({"ok": False, "error": "name is required"}, status=HTTPStatus.BAD_REQUEST)
                    if not raw_message:
                        return self._json({"ok": False, "error": "message is required"}, status=HTTPStatus.BAD_REQUEST)
                    
                    speaker = raw_name[:40]
                    message = {
                        "id": secrets.token_urlsafe(16),
                        "sender": speaker,
                        "seq": int(time.time() * 1000),
                        "type": "message",
                        "body": raw_message[:8000],
                        "ts": utc_now(),
                    }
                    
                    if not supervisor._send_to_socket(message):
                        relay.persist_transcript_message(
                            supervisor.workspace["log_path"],
                            speaker,
                            raw_message[:8000],
                            event_callback=supervisor.handle_relay_event,
                            event_store=supervisor.event_store,
                            message=message,
                            message_type="message",
                        )
                    supervisor.refresh_workspace_state()
                    return self._json({"ok": True, "state": supervisor.state.snapshot()})

                if parsed.path.startswith("/api/agents/") and parsed.path.endswith("/settings"):
                    parts = [part for part in parsed.path.split("/") if part]
                    if len(parts) != 4:
                        self.send_error(HTTPStatus.NOT_FOUND)
                        return
                    agent_name = parts[2]
                    payload = self._payload()
                    try:
                        agent_state = supervisor.update_agent_settings(
                            agent_name,
                            str(payload.get("selected_model") or "").strip() or None,
                            str(payload.get("selected_effort") or "").strip() or None,
                        )
                    except KeyError:
                        return self._json({"ok": False, "error": "unknown agent"}, status=HTTPStatus.NOT_FOUND)
                    return self._json({"ok": True, "agent": agent_state, "state": supervisor.state.snapshot()})

                if parsed.path.startswith("/api/agents/") and parsed.path.endswith("/restart"):
                    parts = [part for part in parsed.path.split("/") if part]
                    if len(parts) != 4:
                        self.send_error(HTTPStatus.NOT_FOUND)
                        return
                    agent_name = parts[2]
                    try:
                        agent_state = supervisor.restart_agent(agent_name)
                    except KeyError:
                        return self._json({"ok": False, "error": "unknown agent"}, status=HTTPStatus.NOT_FOUND)
                    except RuntimeError as exc:
                        return self._json({"ok": False, "error": str(exc)}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
                    return self._json({"ok": True, "agent": agent_state, "state": supervisor.state.snapshot()})

                if parsed.path == "/api/compact":
                    result = supervisor.compact_context()
                    status = HTTPStatus.OK if result.get("ok", False) else HTTPStatus.INTERNAL_SERVER_ERROR
                    return self._json(result, status=status)

                if parsed.path == "/api/repo/pull":
                    supervisor.state.patch("workspace", {"sync_state": "running"})
                    try:
                        git_proc = subprocess.run(
                            ["git", "pull"],
                            cwd=str(SCRIPT_DIR),
                            text=True,
                            capture_output=True,
                            timeout=30,
                        )
                        chmod_proc = subprocess.run(
                            ["chmod", "-R", "u+rw,g+rw", "."],
                            cwd=str(SCRIPT_DIR),
                            text=True,
                            capture_output=True,
                            timeout=15,
                        )
                        ok = git_proc.returncode == 0
                        sync_body = (
                            f"[SYNC] Repository synced. All agents now share read-write access to {SCRIPT_DIR}."
                            if ok
                            else f"[SYNC ERROR] git pull failed for {SCRIPT_DIR}. Review the latest sync output before dispatching more work."
                        )
                        supervisor.persist_system_message(sync_body, message_type="sync")
                        supervisor.refresh_workspace_state()
                        supervisor.state.patch(
                            "workspace",
                            {
                                "sync_state": "synced" if ok else "error",
                                "last_sync_at": utc_now(),
                            },
                        )
                        return self._json({
                            "ok": ok,
                            "stdout": git_proc.stdout,
                            "stderr": git_proc.stderr,
                            "chmod_ok": chmod_proc.returncode == 0,
                            "chmod_stderr": chmod_proc.stderr,
                            "sync_path": str(SCRIPT_DIR),
                            "state": supervisor.state.snapshot(),
                        })
                    except Exception as exc:
                        supervisor.state.patch("workspace", {"sync_state": "error"})
                        return self._json({"ok": False, "error": str(exc)}, status=HTTPStatus.INTERNAL_SERVER_ERROR)

                if parsed.path == "/api/sleep":
                    payload = self._payload()
                    want_sleep = bool(payload.get("sleep", not supervisor.is_sleeping()))
                    supervisor.set_sleeping(want_sleep)
                    if not want_sleep:
                        wake_body = (
                            "[WAKE] All agents are back online. Review the messages above that "
                            "arrived during sleep and continue where we left off."
                        )
                        supervisor.persist_system_message(wake_body, message_type="wake")
                    return self._json({
                        "ok": True,
                        "sleeping": want_sleep,
                        "state": supervisor.state.snapshot(),
                    })

                # ── Project endpoints ──────────────────────────────
                if parsed.path == "/api/projects/lock":
                    payload = self._payload()
                    try:
                        projects = supervisor.lock_project(
                            path=str(payload.get("path") or "").strip() or None,
                            url=str(payload.get("url") or "").strip() or None,
                            name=str(payload.get("name") or "").strip() or None,
                        )
                        supervisor.state.patch("project", {
                            "active": projects["active"],
                            "name": projects["projects"].get(projects["active"], {}).get("name"),
                            "path": projects["projects"].get(projects["active"], {}).get("path", str(SCRIPT_DIR)),
                        })
                        return self._json({"ok": True, "projects": projects})
                    except (ValueError, subprocess.CalledProcessError) as exc:
                        return self._json({"ok": False, "error": str(exc)}, status=HTTPStatus.BAD_REQUEST)

                if parsed.path == "/api/projects/unlock":
                    projects = supervisor.unlock_project()
                    supervisor.state.patch("project", {
                        "active": None,
                        "name": None,
                        "path": str(SCRIPT_DIR),
                    })
                    return self._json({"ok": True, "projects": projects})

                # ── Task endpoints ────────────────────────────────
                if parsed.path == "/api/tasks":
                    payload = self._payload()
                    title = str(payload.get("title") or "").strip()
                    if not title:
                        return self._json({"ok": False, "error": "title is required"}, status=HTTPStatus.BAD_REQUEST)
                    task = supervisor.task_state.create_task_command(
                        title=title,
                        task_type=str(payload.get("type") or "general"),
                        priority=str(payload.get("priority") or "normal"),
                        assigned_to=payload.get("assigned_to") if isinstance(payload.get("assigned_to"), list) else None,
                        source_message=str(payload.get("source_message") or ""),
                    )
                    return self._json({"ok": True, "task": task})

                if parsed.path.startswith("/api/tasks/") and not parsed.path.endswith("/"):
                    parts = [p for p in parsed.path.split("/") if p]
                    if len(parts) == 3:
                        try:
                            task_id = int(parts[2])
                        except ValueError:
                            return self._json({"ok": False, "error": "invalid task id"}, status=HTTPStatus.BAD_REQUEST)
                        payload = self._payload()
                        task = supervisor.task_state.get_task(task_id)
                        if not task:
                            return self._json({"ok": False, "error": "task not found"}, status=HTTPStatus.NOT_FOUND)
                        kwargs: dict[str, Any] = {}
                        if "status" in payload and payload["status"] in relay.TASK_STATUSES:
                            kwargs["status"] = payload["status"]
                        if "assigned_to" in payload and isinstance(payload["assigned_to"], list):
                            kwargs["assigned_to"] = payload["assigned_to"]
                        if "priority" in payload:
                            kwargs["priority"] = str(payload["priority"])
                        if kwargs:
                            task = supervisor.task_state.update_task_command(task_id, **kwargs)
                        return self._json({"ok": True, "task": task})

                self.send_error(HTTPStatus.NOT_FOUND)

        return Handler

    def start_http_server(self) -> None:
        handler = self.make_handler()
        self.http_server = ReusableHTTPServer(
            (self.config["ui"]["host"], self.config["ui"]["port"]),
            handler,
        )
        self.http_thread = threading.Thread(target=self.http_server.serve_forever, daemon=True)
        self.http_thread.start()

    async def refresh_loop(self) -> None:
        tick = 0
        while not self.stop_event.is_set():
            self.refresh_transcript_state()
            self.refresh_tmux_state()
            if tick % 5 == 0:
                self.refresh_workspace_state()
            tick += 1
            if tick % 3 == 0 and self._sse_clients:
                snapshot = self.state.snapshot()
                for agent_name in snapshot.get("agents", {}):
                    snapshot["agents"][agent_name]["fuel"] = self.state.fuel_for_agent(agent_name)
                snapshot["events"] = {"latest_id": self.event_store.latest_event_id()}
                self.sse_broadcast("state_refresh", snapshot)
            await asyncio.sleep(1.0)

    async def run(self) -> int:
        self.prepare_runtime()
        self.start_http_server()
        self.ensure_tmux_layout()

        # Check dispatcher (Ollama) health on startup
        dispatcher_config = self.config.get("dispatcher", {})
        if dispatcher_config.get("enabled"):
            health = await dispatcher_mod.health_check(dispatcher_config.get("ollama_host", "http://localhost:11434"))
            if health["available"]:
                self.state.patch("dispatcher", {
                    "state": "active",
                    "router_model": dispatcher_config.get("router_model"),
                    "models_loaded": health["models"],
                })
                print(f"[supervisor] dispatcher active — Ollama models: {health['models']}", flush=True)
            else:
                self.state.patch("dispatcher", {"state": "unavailable"})
                print("[supervisor] dispatcher unavailable — Ollama not reachable, falling back to broadcast", flush=True)

        loop = asyncio.get_running_loop()
        for sig in (signal.SIGINT, signal.SIGTERM):
            try:
                loop.add_signal_handler(sig, self.stop_event.set)
            except NotImplementedError:
                pass

        refresh_task = asyncio.create_task(self.refresh_loop())
        relay_task = asyncio.create_task(
            relay.run_relay(
                self.config,
                event_callback=self.handle_relay_event,
                stop_event=self.stop_event,
                is_sleeping=self.is_sleeping,
                event_store=self.event_store,
                task_manager=self.task_state,
            )
        )

        result = 0
        try:
            result = await relay_task
        except Exception as exc:
            self.state.patch("relay", {"state": "error", "last_error": str(exc)})
            self.state.patch("app", {"phase": "error"})
            raise
        finally:
            self.stop_event.set()
            refresh_task.cancel()
            try:
                await refresh_task
            except asyncio.CancelledError:
                pass
            if self.http_server is not None:
                self.http_server.shutdown()
                self.http_server.server_close()
            self.event_store.close()
        return result


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="clcod supervisor")
    parser.add_argument("--config", default=str(relay.DEFAULT_CONFIG_PATH), help="Path to config.json")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    config = relay.load_config(args.config)
    supervisor = RuntimeSupervisor(config)
    return asyncio.run(supervisor.run())


if __name__ == "__main__":
    raise SystemExit(main())