"""Kanban dashboard plugin — backend API routes, mounted at /api/plugins/kanban/.

Every handler is a thin wrapper around ``hermes_cli.kanban_db`` (the same code paths the CLI
and gateway ``/kanban`` command use, so the surfaces cannot drift). The ``/events`` WebSocket
tails the append-only ``task_events`` table on a short poll (WAL reads run alongside the
dispatcher's write txns); it carries its credential in the query string (browsers can't set
``Authorization`` on an upgrade) and is gated by the dashboard's canonical WS auth check.
"""

from __future__ import annotations

import asyncio
import importlib
import json
import logging
import re
import sqlite3
import time
from concurrent.futures import ThreadPoolExecutor
from contextlib import closing, contextmanager
from dataclasses import asdict
from functools import partial
from pathlib import Path
from typing import Any, Callable, Iterator, Optional

from fastapi import (
    APIRouter, File, Form, HTTPException, Query, UploadFile, WebSocket, WebSocketDisconnect, status as http_status)
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field

from hermes_cli import kanban_db
from hermes_cli.web_read_coalescing import coalesced_read
from hermes_cli import kanban_db_connect as kbc
from hermes_cli import kanban_db_notify as kbn
from hermes_cli import kanban_db_dispatch as kbd
from hermes_cli import kanban_db_workspace as kbw
from hermes_cli import kanban_diagnostics as kd
from hermes_cli.kanban_db import KANBAN_ATTACHMENT_MAX_BYTES, _collision_free_path, _safe_attachment_name

log = logging.getLogger(__name__)

router = APIRouter()

_BOARD_Q = Query(None, description="Kanban board slug (omit for current)")


# --- Connection / board helpers ---------------------------------------------

def _ws_upgrade_authorized(ws: "WebSocket") -> bool:
    """Authorize a WS upgrade via the dashboard's canonical gate (``web_server_chat._ws_auth_ok``:
    ``?token=`` / ``?ticket=`` / ``?internal=``) so this endpoint can never drift from core
    auth; accepts when the dashboard isn't importable (bare-FastAPI test harness)."""
    try:
        from hermes_cli import web_server_chat as _ws
    except Exception:
        return True
    return bool(_ws._ws_auth_ok(ws))


def _normalize_slug_or_400(slug: str) -> Optional[str]:
    with _value_error_400():
        return kanban_db._normalize_board_slug(slug)


def _resolve_board(board: Optional[str]) -> Optional[str]:
    """Validate/normalise a board slug query param (400 malformed, 404 unknown);
    ``None`` when omitted so ``kb.connect()`` falls through to the active board."""
    if board is None or board == "":
        return None
    normed = _normalize_slug_or_400(board)
    if normed and normed != kanban_db.DEFAULT_BOARD and not kanban_db.board_exists(normed):
        raise HTTPException(status_code=404, detail=f"board {normed!r} does not exist")
    return normed


def _existing_board_slug(slug: str) -> str:
    """Normalise a path slug and require the board to exist (400 / 404)."""
    normed = _normalize_slug_or_400(slug)
    if not normed or not kanban_db.board_exists(normed):
        raise HTTPException(status_code=404, detail=f"board {slug!r} does not exist")
    return normed


def _conn(board: Optional[str] = None):
    """Connect to the already-normalised ``board`` (``None`` = active). ``init_db`` is
    idempotent; running it here lets a fresh install self-heal if POST /tasks arrives first."""
    try:
        kanban_db.init_db(board=board)
    except Exception as exc:
        log.warning("kanban init_db failed: %s", exc)
    return kbc.connect(board=board)


@contextmanager
def _board_conn(board: Optional[str]) -> Iterator[tuple[Optional[str], sqlite3.Connection]]:
    """Resolve the ``board`` query param, open a connection, close it on exit."""
    board = _resolve_board(board)
    with closing(_conn(board=board)) as conn:
        yield board, conn


def _with_board_pinned(board: Optional[str], fn: Callable[[], Any]) -> Any:
    """Run ``fn`` with the board pinned context-locally, not via the process-global
    ``HERMES_KANBAN_BOARD`` env var (concurrent requests for different boards would cross-write)."""
    with kanban_db.scoped_current_board(_resolve_board(board) or kanban_db.DEFAULT_BOARD):
        return fn()


def _require(getter: Callable, conn: sqlite3.Connection, ident, label: str):
    obj = getter(conn, ident)
    if obj is None:
        raise HTTPException(status_code=404, detail=f"{label} {ident} not found")
    return obj


def _run_aux(board: Optional[str], module: str, fn: str, task_id: str, author: Optional[str]) -> Any:
    """Run a slow auxiliary-LLM task helper (``hermes_cli.<module>.<fn>``) with the board pinned;
    the module is imported lazily so a missing aux client can't break plugin load."""
    def _run():
        return getattr(importlib.import_module(f"hermes_cli.{module}"), fn)(task_id, author=(author or None))
    return _with_board_pinned(board, _run)


def _require_task(conn: sqlite3.Connection, task_id: str) -> kanban_db.Task:
    return _require(kanban_db.get_task, conn, task_id, "task")


def _require_run(conn: sqlite3.Connection, run_id: int) -> kanban_db.Run:
    return _require(kanban_db.get_run, conn, run_id, "run")


def _require_ok(ok: bool) -> None:
    """404 when a kanban_db mutator reports the task vanished mid-request."""
    if not ok:
        raise HTTPException(status_code=404, detail="task not found")


def _conflict(detail: str) -> HTTPException:
    return HTTPException(status_code=409, detail=detail)


@contextmanager
def _map_errors(status: int, *types: type[BaseException]) -> Iterator[None]:
    """Map the given exception types to ``HTTPException(status, str(exc))``."""
    try:
        yield
    except types as e:
        raise HTTPException(status_code=status, detail=str(e))


_value_error_400 = partial(_map_errors, 400, ValueError)  # domain-layer validation refusals


@contextmanager
def _errors_to_500(prefix: str) -> Iterator[None]:
    """Map any unexpected exception to ``500 "<prefix>: <exc>"``; HTTPExceptions pass through."""
    try:
        yield
    except HTTPException:
        raise
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f"{prefix}: {exc}")


# --- Serialization helpers --------------------------------------------------

# Dashboard columns, left-to-right ("archived" is a filter toggle, not a column). Keep in
# sync with kanban_db.VALID_STATUSES — a status missing here gets mis-bucketed into ``todo``.
BOARD_COLUMNS: list[str] = ["triage", "todo", "scheduled", "ready", "running", "blocked", "review", "done"]

_CARD_SUMMARY_PREVIEW_CHARS = 200


def _task_dict(task: kanban_db.Task, *, latest_summary: Optional[str] = None,
               current_run_started_at: Optional[int] = None) -> dict[str, Any]:
    d = asdict(task)
    # Derived age metrics so the UI can colour stale cards without client deltas.
    try:
        d["age"] = kanban_db.task_age(task)
    except Exception:
        d["age"] = {"created_age_seconds": None, "started_age_seconds": None, "time_to_complete_seconds": None}
    # Latest non-null run summary (workers hand off via ``task_runs.summary``, not ``tasks.result``).
    d["latest_summary"] = latest_summary
    # Start of the current task_runs row (``tasks.started_at`` is the FIRST-EVER
    # start; after a retry the run clock must tick from the fresh run).
    d["current_run_started_at"] = current_run_started_at
    return d


def _attachment_dict(a: kanban_db.Attachment) -> dict[str, Any]:
    """``stored_path`` is the absolute on-disk path workers read; UI downloads by ``id``."""
    return {
        "id": a.id, "task_id": a.task_id, "filename": a.filename, "content_type": a.content_type,
        "size": a.size, "uploaded_by": a.uploaded_by, "stored_path": a.stored_path, "created_at": a.created_at}


def _placeholders(ids: list) -> str:
    return ",".join(["?"] * len(ids))


def _compute_task_diagnostics(conn: sqlite3.Connection, task_ids: Optional[list[str]] = None) -> dict[str, list[dict]]:
    """``{task_id: [diagnostic_dict, ...]}`` (tasks with none omitted) via three aggregate
    queries (tasks, events, runs) — slurps the board; paginate if profiling shows a hotspot."""
    from hermes_cli.config import load_config

    if task_ids is not None and not task_ids:
        return {}
    diag_config = kd.config_from_runtime_config(load_config())
    if task_ids is not None:
        rows = conn.execute(f"SELECT * FROM tasks WHERE id IN ({_placeholders(task_ids)})", tuple(task_ids)).fetchall()
    else:
        rows = conn.execute("SELECT * FROM tasks WHERE status != 'archived'").fetchall()
    if not rows:
        return {}
    row_ids = [r["id"] for r in rows]

    def _rows_by_task(table: str) -> dict[str, list]:
        by_task: dict[str, list] = {tid: [] for tid in row_ids}
        for row in conn.execute(
            f"SELECT * FROM {table} WHERE task_id IN ({_placeholders(row_ids)}) ORDER BY id", tuple(row_ids)):
            by_task.setdefault(row["task_id"], []).append(row)
        return by_task

    events_by_task = _rows_by_task("task_events")
    runs_by_task = _rows_by_task("task_runs")
    graph_by_task = kanban_db.task_graph_contexts(conn, row_ids)
    out: dict[str, list[dict]] = {}
    for r in rows:
        tid = r["id"]
        diags = kd.compute_task_diagnostics(
            r, events_by_task[tid], runs_by_task[tid], config=diag_config, graph=graph_by_task.get(tid))
        if diags:
            out[tid] = [d.to_dict() for d in diags]
    return out


def _warnings_summary_from_diagnostics(diagnostics: list[dict]) -> Optional[dict]:
    """Compact card badge summary ``{count, kinds, latest_at, highest_severity}``; None when empty."""
    if not diagnostics:
        return None
    kinds: dict[str, int] = {}
    count = latest = 0
    highest_idx, highest_sev = -1, None
    for d in diagnostics:
        n = d.get("count", 1)
        kinds[d["kind"]] = kinds.get(d["kind"], 0) + n
        count += n
        latest = max(latest, d.get("last_seen_at") or 0)
        sev = d.get("severity")
        if sev in kd.SEVERITY_ORDER and kd.SEVERITY_ORDER.index(sev) > highest_idx:
            highest_idx, highest_sev = kd.SEVERITY_ORDER.index(sev), sev
    return {"count": count, "kinds": kinds, "latest_at": latest, "highest_severity": highest_sev}


def _attach_diagnostics(task_d: dict, diags: Optional[list[dict]]) -> None:
    """Full list in the payload (drawer renders without a second round-trip); card badge gets the summary."""
    if diags:
        task_d["diagnostics"] = diags
        task_d["warnings"] = _warnings_summary_from_diagnostics(diags)


def _links_for(conn: sqlite3.Connection, task_id: str) -> dict[str, list[str]]:
    """Return {'parents': [...], 'children': [...]} for a task."""
    def _ids(col: str, other: str) -> list[str]:
        return [r[col] for r in conn.execute(f"SELECT {col} FROM task_links WHERE {other} = ? ORDER BY {col}", (task_id,))]
    return {"parents": _ids("parent_id", "child_id"), "children": _ids("child_id", "parent_id")}


def _link_tasks(conn: sqlite3.Connection, links: dict[str, list[str]]) -> list[dict]:
    """One {id, title, status} row per linked task, so UIs can render titles
    instead of raw ids. Dropped/foreign rows are simply absent — callers fall
    back to the id."""
    rows = []
    for task_id in dict.fromkeys([*links["parents"], *links["children"]]):
        task = kanban_db.get_task(conn, task_id)
        if task:
            rows.append({"id": task.id, "title": task.title, "status": task.status})
    return rows


# --- GET /board -------------------------------------------------------------

def get_board(
    tenant: Optional[str] = Query(None, description="Filter to a single tenant"),
    include_archived: bool = Query(False),
    board: Optional[str] = _BOARD_Q,
    workflow_template_id: Optional[str] = Query(None, description="Restrict to tasks using this workflow template id"),
    current_step_key: Optional[str] = Query(None, description="Restrict to tasks at this workflow step key")):
    """Full board grouped by status column; omitting ``board`` uses the active board
    (``HERMES_KANBAN_BOARD`` env → on-disk ``current`` pointer → ``default``)."""
    with _board_conn(board) as (board, conn):
        tasks = kanban_db.list_tasks(
            conn, tenant=tenant, include_archived=include_archived,
            workflow_template_id=workflow_template_id, current_step_key=current_step_key)
        # Link / comment / progress rollups are each one aggregate query rather than N per-task lookups.
        link_counts: dict[str, dict[str, int]] = {}
        for row in conn.execute("SELECT parent_id, child_id FROM task_links").fetchall():
            link_counts.setdefault(row["parent_id"], {"parents": 0, "children": 0})["children"] += 1
            link_counts.setdefault(row["child_id"], {"parents": 0, "children": 0})["parents"] += 1
        comment_counts: dict[str, int] = {
            r["task_id"]: r["n"] for r in conn.execute("SELECT task_id, COUNT(*) AS n FROM task_comments GROUP BY task_id")}
        progress: dict[str, dict[str, int]] = {}  # per parent: children done / total, rendered as "N/M"
        for row in conn.execute(
            "SELECT l.parent_id AS pid, t.status AS cstatus FROM task_links l JOIN tasks t ON t.id = l.child_id").fetchall():
            p = progress.setdefault(row["pid"], {"done": 0, "total": 0})
            p["total"] += 1
            p["done"] += row["cstatus"] == "done"
        diagnostics_per_task = _compute_task_diagnostics(conn, task_ids=None)
        latest_event_id = conn.execute("SELECT COALESCE(MAX(id), 0) AS m FROM task_events").fetchone()["m"]
        columns: dict[str, list[dict]] = {c: [] for c in BOARD_COLUMNS}
        if include_archived:
            columns["archived"] = []
        # One window-function query for latest summaries (avoids N+1); cards get a
        # truncated preview, the full text comes from /tasks/:id.
        summary_map = kanban_db.latest_summaries(conn, [t.id for t in tasks])
        # One query for the active run's start per card (avoids N+1); the run
        # clock ticks from this, not the task's first-ever start.
        run_start_map = kanban_db.current_run_started_ats(conn, [t.id for t in tasks])
        for t in tasks:
            full = summary_map.get(t.id)
            d = _task_dict(t, latest_summary=(full[:_CARD_SUMMARY_PREVIEW_CHARS] if full else None),
                           current_run_started_at=run_start_map.get(t.id))
            d["link_counts"] = link_counts.get(t.id, {"parents": 0, "children": 0})
            d["comment_count"] = comment_counts.get(t.id, 0)
            d["progress"] = progress.get(t.id)  # None when the task has no children
            _attach_diagnostics(d, diagnostics_per_task.get(t.id))
            columns[t.status if t.status in columns else "todo"].append(d)

        # Queue lanes keep the list_tasks dispatch order; the done column is
        # history, so order it newest-completed-first. Two stable sorts compose
        # into the "completed_at DESC NULLS LAST, id DESC" SQL key.
        columns["done"].sort(key=lambda d: d["id"], reverse=True)
        columns["done"].sort(key=lambda d: (d["completed_at"] is None, -(d["completed_at"] or 0)))

        # Queue columns keep list_tasks' dispatch order (priority DESC, created_at ASC).
        tenants = [r["tenant"] for r in conn.execute("SELECT DISTINCT tenant FROM tasks WHERE tenant IS NOT NULL ORDER BY tenant")]
        assignees = [r["assignee"] for r in conn.execute(
            "SELECT DISTINCT assignee FROM tasks WHERE assignee IS NOT NULL AND status != 'archived' ORDER BY assignee")]
        return {
            "columns": [{"name": name, "tasks": columns[name]} for name in columns], "tenants": tenants,
            "assignees": assignees, "latest_event_id": int(latest_event_id), "now": int(time.time())}


_read_board = coalesced_read(get_board)


@router.get("/board")
async def get_board_endpoint(
    tenant: Optional[str] = Query(None, description="Filter to a single tenant"),
    include_archived: bool = Query(False),
    board: Optional[str] = _BOARD_Q,
    workflow_template_id: Optional[str] = Query(None, description="Restrict to tasks using this workflow template id"),
    current_step_key: Optional[str] = Query(None, description="Restrict to tasks at this workflow step key"),
):
    # Resolve selection before keying so a board switch cannot join an older read.
    return await _read_board(
        tenant=tenant,
        include_archived=include_archived,
        board=board or kanban_db.get_current_board(),
        workflow_template_id=workflow_template_id,
        current_step_key=current_step_key,
    )


# --- GET /tasks/:id ---------------------------------------------------------

@router.get("/tasks/{task_id}")
def get_task(
    task_id: str,
    board: Optional[str] = Query(None),
    run_state_type: Optional[str] = Query(None, description="With run_state_name: filter runs by column 'status' or 'outcome'"),
    run_state_name: Optional[str] = Query(None, description="With run_state_type: exact value for that run column")):
    with _board_conn(board) as (board, conn):
        if (run_state_type is None) ^ (run_state_name is None):
            raise HTTPException(status_code=400, detail="run_state_type and run_state_name must be passed together or omitted")
        if run_state_type not in (None, "status", "outcome"):
            raise HTTPException(status_code=400, detail="run_state_type must be 'status' or 'outcome'")
        task = _require_task(conn, task_id)
        # Drawer returns the FULL summary (cards on /board carry a 200-char preview).
        task_d = _task_dict(
            task, latest_summary=kanban_db.latest_summary(conn, task_id),
            current_run_started_at=kanban_db.current_run_started_ats(conn, [task_id]).get(task_id))
        links = _links_for(conn, task_id)
        child_summaries = kanban_db.latest_summaries(conn, links["children"])
        children = filter(None, (kanban_db.get_task(conn, cid) for cid in links["children"]))
        _attach_diagnostics(task_d, _compute_task_diagnostics(conn, task_ids=[task_id]).get(task_id) or [])
        return {
            "task": task_d,
            "comments": [asdict(c) for c in kanban_db.list_comments(conn, task_id)],
            "events": [asdict(e) for e in kanban_db.list_events(conn, task_id)],
            "attachments": [_attachment_dict(a) for a in kanban_db.list_attachments(conn, task_id)],
            "links": links,
            "link_tasks": _link_tasks(conn, links),
            "child_results": [
                {"id": c.id, "title": c.title, "status": c.status, "latest_summary": child_summaries.get(c.id), "result": c.result}
                for c in children],
            "runs": [asdict(r) for r in kanban_db.list_runs(conn, task_id, state_type=run_state_type, state_name=run_state_name)]}


# --- POST /tasks ------------------------------------------------------------

class CreateTaskBody(BaseModel):
    title: str
    body: Optional[str] = None
    assignee: Optional[str] = None
    tenant: Optional[str] = None
    priority: int = 0
    workspace_kind: Optional[str] = None  # None = scratch, or the board project's worktree when scoped
    workspace_path: Optional[str] = None
    parents: list[str] = Field(default_factory=list)
    triage: bool = False
    idempotency_key: Optional[str] = None
    max_runtime_seconds: Optional[int] = None
    skills: Optional[list[str]] = None
    goal_mode: bool = False
    goal_max_turns: Optional[int] = None
    model_override: Optional[str] = None
    provider_override: Optional[str] = None
    reasoning_effort: Optional[str] = None  # none|minimal|…|ultra; None inherits the profile's level
    project_id: Optional[str] = None  # None inherits the board's scoped project (if any)


@router.post("/tasks")
def create_task(payload: CreateTaskBody, board: Optional[str] = Query(None)):
    with _board_conn(board) as (board, conn), _value_error_400():
        # CreateTaskBody field names match create_task's keyword parameters.
        task_id = kanban_db.create_task(conn, created_by="dashboard", board=board, **payload.model_dump())
        task = kanban_db.get_task(conn, task_id)
        body: dict[str, Any] = {"task": _task_dict(
            task, current_run_started_at=kanban_db.current_run_started_ats(conn, [task_id]).get(task_id)
        ) if task else None}
        # Dispatcher-presence warning so the UI can banner a ready+assigned task that would
        # otherwise sit idle (no gateway / dispatch_in_gateway=false); triage/todo are expected
        # to wait, unassigned tasks can't dispatch anyway. Probe the request's active home: the
        # dashboard backend may run under a different HERMES_HOME than the board's profile.
        if task and task.status == "ready" and task.assignee:
            try:
                from hermes_cli.kanban import _check_dispatcher_presence
                from hermes_constants import get_hermes_home
                running, message = _check_dispatcher_presence(hermes_home=get_hermes_home())
                if not running and message:
                    body["warning"] = message
            except Exception:
                pass  # probe failure must never block the create itself
        return body


# --- Attachments — upload / list / download / delete ------------------------
# Size cap, filename sanitiser, and collision resolver live in ``kanban_db`` so the
# dashboard, agent toolset, and CLI share one implementation.

@router.get("/tasks/{task_id}/attachments")
def list_task_attachments(task_id: str, board: Optional[str] = Query(None)):
    with _board_conn(board) as (board, conn):
        _require_task(conn, task_id)
        return {"attachments": [_attachment_dict(a) for a in kanban_db.list_attachments(conn, task_id)]}


@router.post("/tasks/{task_id}/attachments")
async def upload_task_attachment(
    task_id: str,
    file: UploadFile = File(...),
    board: Optional[str] = Query(None),
    uploaded_by: Optional[str] = Form(None)):
    """Store an upload under ``attachments_root(board)/<task_id>/`` (sanitised,
    collision-resolved name; ``_safe_attachment_name`` ValueError → 400) and record it."""
    with _board_conn(board) as (board, conn), _value_error_400():
        _require_task(conn, task_id)
        safe_name = _safe_attachment_name(file.filename or "")
        dest_dir = kanban_db.task_attachments_dir(task_id, board=board)
        dest_dir.mkdir(parents=True, exist_ok=True)
        dest_path = _collision_free_path(dest_dir, safe_name)  # foo.pdf → foo (1).pdf …
        total = 0  # stream in chunks with a hard size cap so one upload can't fill the disk
        try:
            with open(dest_path, "wb") as out:
                while chunk := await file.read(1024 * 1024):
                    total += len(chunk)
                    if total > KANBAN_ATTACHMENT_MAX_BYTES:
                        out.close()
                        dest_path.unlink(missing_ok=True)
                        raise HTTPException(
                            status_code=413, detail=f"attachment exceeds {KANBAN_ATTACHMENT_MAX_BYTES // (1024 * 1024)} MB limit")
                    out.write(chunk)
        except OSError as exc:
            raise HTTPException(status_code=500, detail=f"failed to store attachment: {exc}")
        att_id = kanban_db.add_attachment(
            conn, task_id, filename=dest_path.name, stored_path=str(dest_path.resolve()),
            content_type=file.content_type, size=total, uploaded_by=(uploaded_by or "dashboard"))
        att = kanban_db.get_attachment(conn, att_id)
        return {"attachment": _attachment_dict(att) if att else None}


@router.get("/attachments/{attachment_id}")
def download_attachment(attachment_id: int, board: Optional[str] = Query(None)):
    with _board_conn(board) as (board, conn):
        att = kanban_db.get_attachment(conn, attachment_id)
        if att is None:
            raise HTTPException(status_code=404, detail="attachment not found")
        # Defense in depth against a tampered DB row: the blob must still live under the board's attachments root.
        root = kanban_db.attachments_root(board=board).resolve()
        try:
            stored = Path(att.stored_path).resolve()
            stored.relative_to(root)
        except (ValueError, OSError):
            raise HTTPException(status_code=404, detail="attachment file unavailable")
        if not stored.is_file():
            raise HTTPException(status_code=404, detail="attachment file missing on disk")
        return FileResponse(path=str(stored), filename=att.filename, media_type=att.content_type or "application/octet-stream")


@router.delete("/attachments/{attachment_id}")
def remove_attachment(attachment_id: int, board: Optional[str] = Query(None)):
    with _board_conn(board) as (board, conn):
        if kanban_db.delete_attachment(conn, attachment_id) is None:
            raise HTTPException(status_code=404, detail="attachment not found")
        return {"ok": True, "id": attachment_id}


# --- PATCH /tasks/:id  and  POST /tasks/bulk ---------------------------------

class UpdateTaskBody(BaseModel):
    status: Optional[str] = None
    assignee: Optional[str] = None
    priority: Optional[int] = None
    title: Optional[str] = None
    body: Optional[str] = None
    result: Optional[str] = None
    block_reason: Optional[str] = None
    # Handoff fields forwarded to complete_task on -> 'done' (parity with ``hermes kanban complete``).
    summary: Optional[str] = None
    metadata: Optional[dict] = None
    # In a PATCH ``None`` means "field not sent", so ``clear_*=True`` is the explicit clear signal.
    # ``reasoning_effort="none"`` is a VALUE (thinking off); it is cleared separately so
    # dropping a model override doesn't silently reset the depth.
    model_override: Optional[str] = None
    provider_override: Optional[str] = None
    clear_model_override: bool = False
    reasoning_effort: Optional[str] = None
    clear_reasoning_effort: bool = False


class BulkTaskBody(BaseModel):
    ids: list[str]
    status: Optional[str] = None
    assignee: Optional[str] = None  # "" or None = unassign
    priority: Optional[int] = None
    archive: bool = False
    result: Optional[str] = None
    summary: Optional[str] = None
    metadata: Optional[dict] = None
    reclaim_first: bool = False
    # Same semantics as UpdateTaskBody.
    model_override: Optional[str] = None
    provider_override: Optional[str] = None
    clear_model_override: bool = False
    reasoning_effort: Optional[str] = None
    clear_reasoning_effort: bool = False


class _StatusRejected(Exception):
    """A status the dashboard may not set via this path; the message is user-facing."""


_RUNNING_DIRECT_MSG = "Cannot set status to 'running' directly; use the dispatcher/claim path"


def _drag_to(conn, task_id: str, s: str) -> bool:
    """Drag-drop into ready/todo/triage: blocked/scheduled -> ready re-opens via ``unblock_task``;
    leaving ``review`` goes through ``reopen_review_task`` (stale-run recovery, parent re-gate,
    ``review_reopened`` event) instead of a raw write; ``triage`` needs no current-state query."""
    current = kanban_db.get_task(conn, task_id) if s != "triage" else None
    if s == "ready" and current and current.status in ("blocked", "scheduled"):
        return kanban_db.unblock_task(conn, task_id)
    if current is not None and current.status == "review":
        return kanban_db.reopen_review_task(conn, task_id)
    return _set_status_direct(conn, task_id, s)


# Status verb dispatch shared by PATCH /tasks/{id} and POST /tasks/bulk: (conn, task_id,
# payload) -> ok. ``review`` uses request_review (never a block, so it can't trip unblock-loop
# detection) and ``done`` pass ``force=True``: a dashboard action is a human override of a live worker claim.
_STATUS_HANDLERS: dict[str, Any] = {
    "done": lambda conn, tid, p: kanban_db.complete_task(
        conn, tid, result=p.result, summary=p.summary, metadata=p.metadata, force=True),
    "blocked": lambda conn, tid, p: kanban_db.block_task(conn, tid, reason=getattr(p, "block_reason", None)),
    "scheduled": lambda conn, tid, p: kanban_db.schedule_task(conn, tid, reason=getattr(p, "block_reason", None)),
    "review": lambda conn, tid, p: kanban_db.request_review(
        conn, tid, summary=p.summary, metadata=p.metadata, reviewer=(p.assignee or None), force=True),
    "ready": lambda conn, tid, p: _drag_to(conn, tid, "ready"),
    "todo": lambda conn, tid, p: _drag_to(conn, tid, "todo"),
    "triage": lambda conn, tid, p: _drag_to(conn, tid, "triage")}


def _apply_status(conn, task_id: str, s: str, p, unknown_detail: str) -> bool:
    """Dispatch a status verb; raises ``_StatusRejected`` (user-facing message)
    for ``running`` or an unknown status (``unknown_detail``)."""
    if s == "running":
        raise _StatusRejected(_RUNNING_DIRECT_MSG)
    handler = _STATUS_HANDLERS.get(s)
    if handler is None:
        raise _StatusRejected(unknown_detail)
    return handler(conn, task_id, p)


def _set_priority(conn, task_id: str, priority: int, board: Optional[str]) -> None:
    kanban_db.edit_task(conn, task_id, priority=int(priority), board=board)


def _apply_model_override(conn, task_id: str, p) -> bool:
    """Raises ValueError/RuntimeError from kanban_db for the caller to map."""
    new_model = None if p.clear_model_override else (p.model_override or "").strip() or None
    return kanban_db.set_model_override(conn, task_id, new_model, provider=p.provider_override)


def _apply_reasoning_effort(conn, task_id: str, p) -> bool:
    return kanban_db.set_reasoning_effort(conn, task_id, None if p.clear_reasoning_effort else p.reasoning_effort)


# Override knobs shared by PATCH and bulk: (payload wants it?, apply, bulk refusal message).
_OVERRIDE_OPS = (
    (lambda p: p.clear_model_override or p.model_override is not None, _apply_model_override, "model override refused"),
    (lambda p: p.clear_reasoning_effort or p.reasoning_effort is not None, _apply_reasoning_effort, "reasoning override refused"),
)


def _patch_status(conn, task_id: str, payload: UpdateTaskBody, review_assignee_deferred: bool) -> None:
    """PATCH status phase: 400 on a rejected verb, 409 when the transition is refused
    (naming the blocking parent(s) for ``ready``/``done``/``review`` so the UI renders an actionable toast)."""
    s = payload.status
    if s == "archived":
        ok = kanban_db.archive_task(conn, task_id)
    else:
        with _map_errors(400, _StatusRejected, ValueError):
            ok = _apply_status(conn, task_id, s, payload, f"unknown status: {s}")
        if s == "review" and ok and review_assignee_deferred and not payload.assignee:
            ok = kanban_db.assign_task(conn, task_id, None)
    if ok:
        return
    blockers = _parents_blocking_ready(conn, task_id) if s == "ready" else []
    if blockers:
        names = ", ".join(f"{p['title']!r} ({p['id']}, status={p['status']})" for p in blockers)
        raise _conflict(f"Cannot move to 'ready': blocked by parent(s) not done — {names}")
    raise _conflict(_open_parent_refusal(conn, task_id, s) or f"status transition to {s!r} not valid from current state")


def _open_parent_refusal(conn, task_id: str, s: str) -> Optional[str]:
    """complete_task/request_review return bare False for a dependency refusal too;
    for a refused ``done``/``review`` name the open parents instead of the generic text."""
    if s not in ("done", "review"):
        return None
    blockers = kanban_db.unsatisfied_parents(conn, task_id)
    if not blockers:
        return None
    detail = ", ".join(f"{pid} ({status})" for pid, status in blockers)
    return f"cannot move {task_id} to {s!r}: unsatisfied parent dependencies: {detail}; complete the parents first (done or archived)"


def _patch_title_body(conn, task_id: str, payload: UpdateTaskBody, board: Optional[str]) -> None:
    """PATCH title/body phase: one UPDATE + ``edited`` event, then the post-commit observer
    (field names only — values never leave the DB via this payload)."""
    with kanban_db.write_txn(conn):
        sets, vals = [], []
        if payload.title is not None:
            if not payload.title.strip():
                raise HTTPException(status_code=400, detail="title cannot be empty")
            sets.append("title = ?")
            vals.append(payload.title.strip())
        if payload.body is not None:
            sets.append("body = ?")
            vals.append(payload.body)
        vals.append(task_id)
        conn.execute(f"UPDATE tasks SET {', '.join(sets)} WHERE id = ?", vals)
        conn.execute(
            "INSERT INTO task_events (task_id, kind, payload, created_at) VALUES (?, 'edited', NULL, ?)",
            (task_id, int(time.time())))
    kanban_db.notify_task_updated(
        conn, task_id, [f for f in ("title", "body") if getattr(payload, f) is not None], board=board)


@router.patch("/tasks/{task_id}")
def update_task(task_id: str, payload: UpdateTaskBody, board: Optional[str] = Query(None)):
    with _board_conn(board) as (board, conn):
        _require_task(conn, task_id)
        # For a combined assignee+review patch, request_review must capture the
        # current implementer before the task is routed to the reviewer.
        review_assignee_deferred = payload.status == "review" and payload.assignee is not None
        if payload.assignee is not None and not review_assignee_deferred:
            with _map_errors(409, RuntimeError):
                _require_ok(kanban_db.assign_task(conn, task_id, payload.assignee or None))
        if payload.status is not None:
            _patch_status(conn, task_id, payload, review_assignee_deferred)
        for wanted, apply, _refused in _OVERRIDE_OPS:
            if wanted(payload):
                with _map_errors(400, ValueError, RuntimeError):
                    ok = apply(conn, task_id, payload)
                _require_ok(ok)
        if payload.priority is not None:
            _set_priority(conn, task_id, payload.priority, board)
        if payload.title is not None or payload.body is not None:
            _patch_title_body(conn, task_id, payload, board)
        updated = kanban_db.get_task(conn, task_id)
        return {"task": _task_dict(
            updated, current_run_started_at=kanban_db.current_run_started_ats(conn, [task_id]).get(task_id)
        ) if updated else None}


@router.delete("/tasks/{task_id}")
def delete_task(task_id: str, board: Optional[str] = Query(None)):
    with _board_conn(board) as (board, conn):
        if not kanban_db.delete_task(conn, task_id):
            raise HTTPException(status_code=404, detail=f"task {task_id} not found")
        return {"deleted": True, "task_id": task_id}


def _parents_blocking_ready(conn: sqlite3.Connection, task_id: str) -> list:
    """Parent rows (id, title, status) not ``done`` that block promotion to ``ready``.

    Used to enrich the 409 response from :func:`update_task` so the dashboard can show an actionable toast
    (#26744) instead of a silent no-op. Returns ``[]`` when nothing blocks the transition (e.g. no parents,
    or all parents already done).
    """
    rows = conn.execute(
        "SELECT t.id, t.title, t.status FROM tasks t "
        "JOIN task_links l ON l.parent_id = t.id "
        "WHERE l.child_id = ? AND t.status != 'done'",
        (task_id,)).fetchall()
    return [{"id": r["id"], "title": r["title"], "status": r["status"]} for r in rows]


def _set_status_direct(conn: sqlite3.Connection, task_id: str, new_status: str) -> bool:
    """Direct status write for drag-drop moves without a structured verb (todo<->ready,
    running<->ready) + a ``status`` event. Leaving ``running`` closes the run as 'reclaimed'
    so attempt history isn't orphaned; the worker is killed only AFTER the txn commits."""
    terminations: list[tuple[Optional[int], Optional[str], Optional[int]]] = []
    effective_status = new_status
    with kanban_db.write_txn(conn):
        prev = conn.execute(
            "SELECT status, current_run_id, worker_pid, claim_lock, worker_started_at FROM tasks WHERE id = ?",
            (task_id,)).fetchone()
        if prev is None:
            return False
        if prev["status"] == "running" and new_status == "ready":
            resume_status = kanban_db._retry_status_for_run(conn, task_id, prev["current_run_id"])
            if resume_status == "review":
                effective_status = "review" if kanban_db._parents_satisfied(conn, task_id) else "todo"
        # Never promote to 'ready' unless all parents are done/archived — otherwise the
        # dispatcher spawns a child whose upstream work hasn't completed.
        if effective_status == "ready" and not kanban_db._parents_satisfied(conn, task_id):
            return False
        was_running = prev["status"] == "running"
        reopening_satisfied_parent = prev["status"] in {"done", "archived"} and effective_status not in {"done", "archived"}
        cur = conn.execute(
            "UPDATE tasks SET status = ?, "
            "  claim_lock = CASE WHEN ? = 'running' THEN claim_lock ELSE NULL END, "
            "  claim_expires = CASE WHEN ? = 'running' THEN claim_expires ELSE NULL END, "
            "  worker_pid = CASE WHEN ? = 'running' THEN worker_pid ELSE NULL END "
            "WHERE id = ?",
            (effective_status,) * 4 + (task_id,))
        if cur.rowcount != 1:
            return False
        run_id = None
        if was_running and effective_status != "running" and prev["current_run_id"]:
            run_id = kanban_db._end_run(
                conn, task_id, outcome="reclaimed", status="reclaimed",
                summary=f"status changed to {effective_status} (dashboard/direct)")
            terminations.append((prev["worker_pid"], prev["claim_lock"], prev["worker_started_at"]))
        conn.execute(
            "INSERT INTO task_events (task_id, run_id, kind, payload, created_at) VALUES (?, ?, 'status', ?, ?)",
            (task_id, run_id, json.dumps({"status": effective_status, "requested_status": new_status}), int(time.time())))
        if reopening_satisfied_parent:
            # Domain-layer invalidation composes via a savepoint inside our txn and hands
            # back worker terminations to perform post-commit.
            result = kanban_db.invalidate_descendants_for_parent_reopen(conn, task_id, author="dashboard")
            terminations.extend(result["terminations"])
    for pid, claim_lock, started_at in terminations:
        kanban_db._terminate_reclaimed_worker(pid, claim_lock, started_at=started_at)
    # Re-opening something may have made children stale.
    if effective_status in {"done", "ready", "review"}:
        kanban_db.recompute_ready(conn)
    return True


# --- Comments / links -------------------------------------------------------

class CommentBody(BaseModel):
    body: str
    author: Optional[str] = "dashboard"


@router.post("/tasks/{task_id}/comments")
def add_comment(task_id: str, payload: CommentBody, board: Optional[str] = Query(None)):
    if not payload.body.strip():
        raise HTTPException(status_code=400, detail="body is required")
    with _board_conn(board) as (board, conn):
        _require_task(conn, task_id)
        kanban_db.add_comment(conn, task_id, author=payload.author or "dashboard", body=payload.body)
        return {"ok": True}


class LinkBody(BaseModel):
    parent_id: str
    child_id: str


@router.post("/links")
def add_link(payload: LinkBody, board: Optional[str] = Query(None)):
    with _board_conn(board) as (board, conn), _value_error_400():
        gated = kanban_db.link_tasks(conn, payload.parent_id, payload.child_id)
        return {"ok": True, "gated": gated}


@router.delete("/links")
def delete_link(parent_id: str = Query(...), child_id: str = Query(...), board: Optional[str] = Query(None)):
    with _board_conn(board) as (board, conn):
        return {"ok": bool(kanban_db.unlink_tasks(conn, parent_id, child_id))}


def _bulk_apply_one(conn, tid: str, payload: BulkTaskBody, board: Optional[str], entry: dict) -> None:
    """Apply the bulk patch to one task, recording refusals in ``entry`` without aborting the
    remaining ops — except a rejected status verb (``_StatusRejected`` propagates)."""
    if payload.archive and not kanban_db.archive_task(conn, tid):
        entry.update(ok=False, error="archive refused")
    if payload.status is not None and not payload.archive:
        s = payload.status
        if not _apply_status(conn, tid, s, payload, f"unknown status {s!r}"):
            entry.update(ok=False, error=_open_parent_refusal(conn, tid, s) or f"transition to {s!r} refused")
    if payload.assignee is not None:
        try:
            ok = (kanban_db.reassign_task(conn, tid, payload.assignee or None, reclaim_first=True) if payload.reclaim_first
                  else kanban_db.assign_task(conn, tid, payload.assignee or None))
            if not ok:
                entry.update(ok=False, error="assign refused")
        except RuntimeError as e:
            entry.update(ok=False, error=str(e))
    if payload.priority is not None:
        _set_priority(conn, tid, payload.priority, board)
    for wanted, apply, refused in _OVERRIDE_OPS:
        if wanted(payload):
            try:
                if not apply(conn, tid, payload):
                    entry.update(ok=False, error=refused)
            except (ValueError, RuntimeError) as e:
                entry.update(ok=False, error=str(e))


@router.post("/tasks/bulk")
def bulk_update(payload: BulkTaskBody, board: Optional[str] = Query(None)):
    """Apply the same patch to every id. Independent iteration — per-task
    failures don't abort siblings; returns per-id outcome for partials."""
    ids = [i for i in (payload.ids or []) if i]
    if not ids:
        raise HTTPException(status_code=400, detail="ids is required")
    results: list[dict] = []
    with _board_conn(board) as (board, conn):
        for tid in ids:
            entry: dict[str, Any] = {"id": tid, "ok": True}
            try:
                if kanban_db.get_task(conn, tid) is None:
                    entry.update(ok=False, error="not found")
                else:
                    _bulk_apply_one(conn, tid, payload, board, entry)
            except Exception as e:  # one bad id shouldn't kill the batch (incl. _StatusRejected)
                entry.update(ok=False, error=str(e))
            results.append(entry)
        return {"results": results}


# --- Diagnostics — fleet-wide distress signals (see kanban_diagnostics) ------

@router.get("/diagnostics")
def list_diagnostics(
    board: Optional[str] = _BOARD_Q,
    severity: Optional[str] = Query(None, description="Filter by severity: warning|error|critical")):
    """Tasks with an active diagnostic, highest severity first then most recent; also
    consumed by ``hermes kanban diagnostics`` when the dashboard runs."""
    with _board_conn(board) as (board, conn):
        diags_by_task = _compute_task_diagnostics(conn, task_ids=None)
        if severity and diags_by_task:
            diags_by_task = {
                tid: keep
                for tid, dl in diags_by_task.items()
                if (keep := [d for d in dl if kd.severity_at_or_above(d.get("severity"), severity)])}
        if not diags_by_task:
            return {"diagnostics": [], "count": 0}
        ids = list(diags_by_task.keys())
        rows = {r["id"]: r for r in conn.execute(
            f"SELECT id, title, status, assignee FROM tasks WHERE id IN ({_placeholders(ids)})", tuple(ids)).fetchall()}
        out = []
        for tid, dl in diags_by_task.items():
            r = rows.get(tid) or {"title": None, "status": None, "assignee": None}
            out.append({
                "task_id": tid, "task_title": r["title"], "task_status": r["status"], "task_assignee": r["assignee"],
                "diagnostics": dl})
        sev_idx = {s: i for i, s in enumerate(kd.SEVERITY_ORDER)}
        out.sort(key=lambda row: (
            -sev_idx.get(row["diagnostics"][0].get("severity"), -1), -(row["diagnostics"][0].get("last_seen_at") or 0)))
        return {"diagnostics": out, "count": sum(len(d["diagnostics"]) for d in out)}


# --- Worker visibility — active-worker list, per-run inspect/terminate -------

try:
    import psutil as _psutil
except ImportError:
    _psutil = None  # type: ignore[assignment]


@router.get("/workers/active")
def list_active_workers(board: Optional[str] = _BOARD_Q):
    """Every running worker: an open ``task_runs`` row with a ``worker_pid`` whose
    task is ``running``. Returns ``{workers, count, checked_at}``."""
    with _board_conn(board) as (board, conn):
        rows = conn.execute(
            "SELECT r.id AS run_id, r.task_id, t.title AS task_title, t.status AS task_status, "
            "t.assignee AS task_assignee, r.profile, r.worker_pid, r.started_at, r.claim_lock, "
            "r.claim_expires, r.last_heartbeat_at, r.max_runtime_seconds "
            "FROM task_runs r JOIN tasks t ON t.id = r.task_id "
            "WHERE r.ended_at IS NULL AND r.worker_pid IS NOT NULL AND t.status = 'running' "
            "ORDER BY r.started_at ASC").fetchall()
        workers = [dict(row) for row in rows]
        return {"workers": workers, "count": len(workers), "checked_at": int(time.time())}


@router.get("/runs/{run_id}")
def get_run_endpoint(run_id: int, board: Optional[str] = _BOARD_Q):
    """``{run: {...}}`` with the same serialisation as ``GET /tasks/{id}``; 404 if unknown."""
    with _board_conn(board) as (board, conn):
        return {"run": asdict(_require_run(conn, run_id))}


@router.get("/runs/{run_id}/inspect")
def inspect_run_endpoint(run_id: int, board: Optional[str] = _BOARD_Q):
    """Live psutil stats for a run's worker; ``{alive: false, reason}`` when unavailable and
    access-denied reported inline rather than as a 500."""
    with _board_conn(board) as (board, conn):
        r = _require_run(conn, run_id)

    def _dead(reason: str, **extra) -> dict:
        return {"run_id": run_id, "alive": False, **extra, "reason": reason}

    if r.ended_at is not None:
        return _dead("run already ended")
    pid = r.worker_pid
    if pid is None:
        return _dead("no worker_pid recorded")
    if _psutil is None:
        return _dead("psutil not available", pid=pid)
    try:
        proc = _psutil.Process(pid)
        info = proc.as_dict(attrs=["cpu_percent", "memory_info", "num_threads", "status", "create_time", "cmdline"])
        try:
            num_fds = proc.num_fds()
        except AttributeError:  # POSIX-only
            num_fds = None
        mem = info.get("memory_info")
        return {
            "run_id": run_id, "alive": True, "pid": pid,
            "cpu_percent": info.get("cpu_percent"),
            "memory_rss_bytes": mem.rss if mem else None,
            "memory_vms_bytes": mem.vms if mem else None,
            "num_threads": info.get("num_threads"), "num_fds": num_fds,
            "status": info.get("status"), "create_time": info.get("create_time"), "cmdline": info.get("cmdline")}
    except _psutil.NoSuchProcess:
        return _dead("process not found", pid=pid)
    except _psutil.AccessDenied:
        return {"run_id": run_id, "alive": True, "pid": pid, "error": "access denied"}


class TerminateRunBody(BaseModel):
    reason: Optional[str] = None


@router.post("/runs/{run_id}/terminate")
def terminate_run_endpoint(run_id: int, payload: TerminateRunBody, board: Optional[str] = _BOARD_Q):
    """Terminate an in-flight run via ``reclaim_task`` (same SIGTERM->SIGKILL flow, bookkeeping
    and events as ``POST /tasks/{id}/reclaim``); 409 if already ended / not reclaimable.

    Closes the gap left by PR #28432, which shipped the read-only sibling endpoints (``/workers/active``,
    ``/runs/{run_id}``, ``/runs/{run_id}/inspect``) but no termination control surface.
    """
    with _board_conn(board) as (board, conn):
        r = _require_run(conn, run_id)
        if r.ended_at is not None:
            raise _conflict(f"run {run_id} already ended")
        if not kanban_db.reclaim_task(conn, r.task_id, reason=payload.reason):
            raise _conflict(f"cannot terminate run {run_id}: task {r.task_id} is no longer in a reclaimable state")
        return {"ok": True, "run_id": run_id, "task_id": r.task_id}


# --- Recovery actions — reclaim / specify / reassign / estimate -------------

class ReclaimBody(BaseModel):
    reason: Optional[str] = None


@router.post("/tasks/{task_id}/reclaim")
def reclaim_task_endpoint(task_id: str, payload: ReclaimBody, board: Optional[str] = Query(None)):
    """Release an active worker claim without waiting for the claim TTL
    (``hermes kanban reclaim <task_id> --reason ...``)."""
    with _board_conn(board) as (board, conn):
        if not kanban_db.reclaim_task(conn, task_id, reason=payload.reason):
            raise _conflict(f"cannot reclaim {task_id}: not in a claimable state (not running, or unknown id)")
        return {"ok": True, "task_id": task_id}


class SpecifyBody(BaseModel):
    """Only the author is configurable; model + prompt come from
    ``auxiliary.triage_specifier`` in config.yaml, same as the CLI."""

    author: Optional[str] = None


@router.post("/tasks/{task_id}/specify")
def specify_task_endpoint(task_id: str, payload: SpecifyBody, board: Optional[str] = Query(None)):
    """Flesh out a triage task via the auxiliary LLM (``hermes kanban specify``). Non-OK is NOT
    an HTTP error — the UI renders the reason inline. Sync ``def`` → runs in the threadpool."""
    outcome = _run_aux(board, "kanban_specify", "specify_task", task_id, payload.author)
    return {"ok": bool(outcome.ok), "task_id": outcome.task_id, "reason": outcome.reason, "new_title": outcome.new_title}


class ReassignBody(BaseModel):
    profile: Optional[str] = None  # "" or None = unassign
    reclaim_first: bool = False
    reason: Optional[str] = None


@router.post("/tasks/{task_id}/reassign")
def reassign_task_endpoint(task_id: str, payload: ReassignBody, board: Optional[str] = Query(None)):
    """Reassign to another profile, optionally reclaiming first
    (``hermes kanban reassign <task_id> <profile> [--reclaim]``)."""
    with _board_conn(board) as (board, conn):
        ok = kanban_db.reassign_task(
            conn, task_id, payload.profile or None, reclaim_first=bool(payload.reclaim_first), reason=payload.reason)
        if not ok:
            raise _conflict(
                f"cannot reassign {task_id}: unknown id, or still "
                "running (pass reclaim_first=true to release the claim first)")
        return {"ok": True, "task_id": task_id, "assignee": payload.profile or None}


# Estimate: rough token/complexity read via the auxiliary model. NOT a dollar cost.
_ESTIMATE_SYSTEM_PROMPT = (
    "You estimate how much work an autonomous coding agent will spend on a "
    "kanban task. Given the task title and description, respond with STRICT "
    "JSON only (no prose, no code fence):\n"
    '{"est_tokens": <integer total tokens across the whole run>, '
    '"complexity": "S"|"M"|"L", '
    '"rationale": "<one short sentence>"}\n'
    "Base the token figure on a realistic multi-turn agent run (reading files, "
    "tool calls, edits, retries) — not a single reply. S≈small/localized, "
    "M≈multi-file, L≈broad or ambiguous. Be honest that this is a rough guess.")


class EstimateBody(BaseModel):
    title: str = ""
    body: Optional[str] = None


@router.post("/estimate")
def estimate_text_endpoint(payload: EstimateBody):
    """Estimate from raw title/body (create dialog, before a task exists)."""
    return _run_estimate(payload.title, payload.body, task_id=None)


@router.post("/tasks/{task_id}/estimate")
def estimate_task_endpoint(task_id: str, board: Optional[str] = Query(None)):
    """Estimate for an existing task; ``{ok, est_tokens, complexity, rationale, model}``."""
    with _board_conn(board) as (board, conn):
        task = _require_task(conn, task_id)
    return _run_estimate(task.title, task.body, task_id=task_id)


def _cap(s: Optional[str], n: int) -> str:
    s = (s or "").strip()
    return s if len(s) <= n else s[:n] + "…"


def _run_estimate(title: str, body: Optional[str], *, task_id: Optional[str]) -> dict:
    """Never raises — config/parse/API errors become ``{"ok": False, "reason"}`` so the UI renders them inline."""
    if not (title or "").strip():
        return {"ok": False, "reason": "a title is required to estimate"}
    try:
        from agent.auxiliary_client import call_llm
    except Exception:
        return {"ok": False, "reason": "auxiliary client unavailable"}
    user_msg = f"Title: {_cap(title, 400)}\n\nDescription:\n{_cap(body, 4000) or '(none)'}"
    # Headless like specify/decompose's _call_aux: without a bound affinity scope the relay-affinity
    # headers are omitted and the OpenCode Go relay answers 400 MissingSessionID (#112043). The
    # create dialog has no task yet, so it shares one stable key.
    from agent.portal_tags import get_affinity_scope, reset_affinity_scope, set_affinity_scope
    affinity_token = None if get_affinity_scope() else set_affinity_scope(f"kanban:{task_id or 'estimate'}")
    try:
        resp = call_llm(
            task="kanban_estimator",
            messages=[{"role": "system", "content": _ESTIMATE_SYSTEM_PROMPT}, {"role": "user", "content": user_msg}],
            temperature=0.0, max_tokens=300, timeout=60)
    except Exception as exc:
        return {"ok": False, "reason": f"LLM error: {type(exc).__name__}"}
    finally:
        if affinity_token is not None:
            reset_affinity_scope(affinity_token)
    try:
        raw = (resp.choices[0].message.content or "").strip()
        model = getattr(resp, "model", None)
    except Exception:
        raw, model = "", None

    # Same tolerant JSON-blob extraction the specifier uses.
    try:
        m = None if raw.lstrip().startswith("{") else re.search(r"\{.*\}", raw, re.DOTALL)
        obj = json.loads(m.group(0) if m else raw)
        parsed = obj if isinstance(obj, dict) else None
    except Exception:
        parsed = None
    if not parsed:
        return {"ok": False, "reason": "could not parse an estimate from the model"}
    try:
        est_tokens = int(parsed.get("est_tokens") or 0)
    except (TypeError, ValueError):
        est_tokens = 0
    complexity = str(parsed.get("complexity") or "").strip().upper()
    return {
        "ok": True, "est_tokens": est_tokens, "complexity": complexity if complexity in {"S", "M", "L"} else None,
        "rationale": str(parsed.get("rationale") or "").strip() or None, "model": model}


# --- Plugin config ----------------------------------------------------------

def _load_config_or_empty() -> dict:
    try:
        from hermes_cli.config import load_config
        return load_config() or {}
    except Exception:
        return {}


@router.get("/config")
def get_config():
    """Kanban dashboard preferences from the ``dashboard.kanban`` config section."""
    k_cfg = (_load_config_or_empty().get("dashboard") or {}).get("kanban") or {}
    return {
        "default_tenant": k_cfg.get("default_tenant") or "",
        "lane_by_profile": bool(k_cfg.get("lane_by_profile", True)),
        "include_archived_by_default": bool(k_cfg.get("include_archived_by_default", False)),
        "render_markdown": bool(k_cfg.get("render_markdown", True))}


# --- Home-channel subscriptions (per-task, per-platform toggles) -------------
# Each gateway platform has at most one "home" (chat_id, thread_id, name); a toggle-on writes
# exactly the notify_subs row ``/kanban create`` would, so the gateway notifier needs no plumbing.

def _configured_home_channels() -> list[dict]:
    """Every platform with a home_channel, from the live GatewayConfig (so env overlays
    like ``TELEGRAM_HOME_CHANNEL`` are honored), sorted by platform."""
    try:
        from gateway.config import load_gateway_config
        gw_cfg = load_gateway_config()
    except Exception:
        return []
    result = [
        {"platform": platform.value, "chat_id": pcfg.home_channel.chat_id,
         "thread_id": pcfg.home_channel.thread_id or "", "name": pcfg.home_channel.name or "Home"}
        for platform, pcfg in gw_cfg.platforms.items() if pcfg and pcfg.home_channel]
    result.sort(key=lambda r: r["platform"])
    return result


def _active_profile_name() -> str:
    """Current Hermes profile name for notify-sub ownership."""
    try:
        from hermes_cli.profiles import get_active_profile_name
        return get_active_profile_name() or "default"
    except Exception:
        return "default"


def _home_for_platform(platform: str, detail: str) -> dict:
    home = next((h for h in _configured_home_channels() if h["platform"] == platform), None)
    if not home:
        raise HTTPException(status_code=404, detail=detail)
    return home


@router.get("/home-channels")
def get_home_channels(task_id: Optional[str] = Query(None), board: Optional[str] = Query(None)):
    """Every platform with a home channel plus whether *task_id* (if given) is
    subscribed to it; without ``task_id`` every ``subscribed`` is false."""
    homes = _configured_home_channels()
    subscribed_homes: set[tuple[str, str, str]] = set()
    if task_id:
        with _board_conn(board) as (board, conn):
            subs = kbn.list_notify_subs(conn, task_id)
        subscribed_homes = {
            (str(sub.get("platform") or ""), str(sub.get("chat_id") or ""), str(sub.get("thread_id") or "")) for sub in subs}
    return {"home_channels": [
        {**home, "subscribed": (home["platform"], home["chat_id"], home["thread_id"]) in subscribed_homes} for home in homes]}


@router.post("/tasks/{task_id}/home-subscribe/{platform}")
def subscribe_home(task_id: str, platform: str, board: Optional[str] = Query(None)):
    """Subscribe *task_id* to *platform*'s home channel. Idempotent at the DB
    layer; 404 when the platform has no home or the task doesn't exist."""
    home = _home_for_platform(
        platform,
        f"No home channel configured for platform {platform!r}. "
        f"Set one from the messenger via /sethome, or configure "
        f"gateway.platforms.{platform}.home_channel in config.yaml.")
    with _board_conn(board) as (board, conn):
        _require_task(conn, task_id)
        kbn.add_notify_sub(
            conn, task_id=task_id, platform=platform, chat_id=home["chat_id"],
            thread_id=home["thread_id"] or None, notifier_profile=_active_profile_name())
        return {"ok": True, "task_id": task_id, "home_channel": home}


@router.delete("/tasks/{task_id}/home-subscribe/{platform}")
def unsubscribe_home(task_id: str, platform: str, board: Optional[str] = Query(None)):
    """Remove any notify subscription on *task_id* matching *platform*'s home."""
    home = _home_for_platform(platform, f"No home channel configured for platform {platform!r}.")
    with _board_conn(board) as (board, conn):
        kbn.remove_notify_sub(
            conn, task_id=task_id, platform=platform, chat_id=home["chat_id"], thread_id=home["thread_id"] or None)
        return {"ok": True, "task_id": task_id, "home_channel": home}


# --- Stats / assignees / worker log / dispatch / model options ---------------

@router.get("/stats")
def get_stats(board: Optional[str] = Query(None)):
    """Per-status + per-assignee counts + oldest-ready age (HUD and router profiles)."""
    with _board_conn(board) as (board, conn):
        return kanban_db.board_stats(conn)


@router.get("/assignees")
def get_assignees(board: Optional[str] = Query(None)):
    """Union of on-disk profiles and assignees used on the board, so a fresh
    profile appears in the picker before it has any task."""
    with _board_conn(board) as (board, conn):
        return {"assignees": kanban_db.known_assignees(conn)}


@router.get("/tasks/{task_id}/log")
def get_task_log(task_id: str, tail: Optional[int] = Query(None, ge=1, le=2_000_000), board: Optional[str] = Query(None)):
    """Worker stdout/stderr log. ``tail`` caps the response bytes; 404 if the
    task never spawned. On-disk log rotates at 2 MiB with one ``.log.1`` kept."""
    with _board_conn(board) as (board, conn):
        _require_task(conn, task_id)
    content = kanban_db.read_worker_log(task_id, tail_bytes=tail, board=board)
    log_path = kanban_db.worker_log_path(task_id, board=board)
    size = log_path.stat().st_size if log_path.exists() else 0
    return {
        "task_id": task_id, "path": str(log_path), "exists": content is not None,
        "size_bytes": size, "content": content or "", "truncated": bool(tail and size > tail)}


@router.post("/dispatch")
def dispatch(dry_run: bool = Query(False), max_n: int = Query(8, alias="max"), board: Optional[str] = Query(None)):
    """Dispatch nudge so the UI doesn't wait out the 60 s dispatcher tick."""
    with _board_conn(board) as (board, conn):
        result = kbd.dispatch_once(conn, dry_run=dry_run, max_spawn=max_n, board=board)
        try:
            return asdict(result)  # DispatchResult is a dataclass
        except TypeError:
            return {"result": str(result)}


@router.get("/model-options")
def model_options():
    """Providers + curated models for the override dropdown via ``inventory.build_models_payload``
    (same substrate as the Models page) so it can't offer a pair Hermes rejects. Skips pricing
    and custom-provider probes: a slow/offline local endpoint must not hang the drawer."""
    try:
        from hermes_cli.inventory import build_models_payload, load_picker_context

        payload = build_models_payload(
            load_picker_context(), explicit_only=True, canonical_order=True, probe_custom_providers=False)
        return {
            "providers": [
                {"slug": row.get("slug", ""), "label": row.get("label") or row.get("slug", ""),
                 "models": list(row.get("models") or [])}
                for row in payload.get("providers", [])
                if row.get("models")]}
    except Exception:
        log.exception("kanban model-options failed")
        return {"providers": []}  # empty catalog → the UI falls back to a free-text input


# --- Boards CRUD (multi-project support) --------------------------------------

class CreateBoardBody(BaseModel):
    slug: str
    name: Optional[str] = None
    description: Optional[str] = None
    icon: Optional[str] = None
    color: Optional[str] = None
    default_workdir: Optional[str] = None
    # Project (id or slug) scoping the board: default_workdir mirrors its primary repo, tasks inherit it.
    project_id: Optional[str] = None
    switch: bool = False


class RenameBoardBody(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    icon: Optional[str] = None
    color: Optional[str] = None
    # For both fields: ``None`` = leave unchanged; "" = clear; value = validate/resolve + set.
    default_workdir: Optional[str] = None
    project_id: Optional[str] = None


# Board transfer exchanges filesystem PATHS, not bytes (same contract as profile export/import):
# clients run the native save/open dialog on the machine hosting the backend.

class ExportBoardBody(BaseModel):
    output: str = ""  # empty → staging path under the kanban root
    attachments: bool = True
    logs: bool = False


class ImportBoardBody(BaseModel):
    archive: str  # path to a board .tar.gz on the backend's filesystem
    slug: Optional[str] = None  # override the archive's slug; collisions auto-suffix
    switch: bool = False


def _board_display_kwargs(p: BaseModel) -> dict[str, Any]:
    """Display-metadata fields shared by create_board / write_board_metadata."""
    return {"name": p.name, "description": p.description, "icon": p.icon, "color": p.color}


def _resolve_project(ref: Optional[str]) -> tuple[Optional[str], Optional[str], Optional[str]]:
    """Resolve a project id/slug to ``(id, name, primary_path)``; ``(None,)*3``
    for a falsy ref, 400 when a non-empty ref doesn't resolve."""
    if not ref or not ref.strip():
        return None, None, None
    with _errors_to_500("projects unavailable"):
        from hermes_cli import projects_db as pdb
        with pdb.connect_closing() as pconn:
            proj = pdb.get_project(pconn, ref.strip())
    if proj is None:
        raise HTTPException(status_code=400, detail=f"project {ref!r} does not exist")
    return proj.id, proj.name, (proj.primary_path or None)


def _projects_by_id() -> dict[str, Any]:
    """Map every project id -> Project (archived included) for annotation."""
    try:
        from hermes_cli import projects_db as pdb
        with pdb.connect_closing() as pconn:
            return {p.id: p for p in pdb.list_projects(pconn, include_archived=True)}
    except Exception:
        return {}


def _board_counts(slug: str) -> dict[str, int]:
    """``{status: count}`` for a board; ``{}`` on a missing/empty DB."""
    try:
        if not kanban_db.kanban_db_path(board=slug).exists():
            return {}
        with closing(kbc.connect(board=slug)) as conn:
            rows = conn.execute("SELECT status, COUNT(*) AS n FROM tasks GROUP BY status").fetchall()
            return {r["status"]: int(r["n"]) for r in rows}
    except Exception:
        return {}


def _default_workspace_kind(board: dict[str, Any]) -> str:
    """Recommend a non-destructive task workspace from board metadata."""
    workdir = str(board.get("default_workdir") or "").strip()
    if not workdir:
        return "scratch"
    try:
        return "worktree" if kbw._git_toplevel(Path(workdir)) else "dir"
    except (OSError, ValueError):
        return "dir"


def _annotate_board_meta(meta: dict) -> dict:
    meta["default_workspace_kind"] = _default_workspace_kind(meta)
    _, meta["project_name"], _ = _resolve_project(meta.get("project_id"))
    return meta


@router.get("/projects")
def list_kanban_projects():
    """Live (non-archived) projects available for board scoping."""
    with _errors_to_500("failed to list projects"):
        from hermes_cli import projects_db as pdb
        with pdb.connect_closing() as pconn:
            projects = pdb.list_projects(pconn, include_archived=False)
    return {"projects": [
        {"id": p.id, "slug": p.slug, "name": p.name,
         "primary_path": p.primary_path or "", "icon": p.icon or "", "color": p.color or ""}
        for p in projects]}


@router.get("/boards")
def list_boards(include_archived: bool = Query(False)):
    """Every board on disk with task counts and the active slug."""
    boards = kanban_db.list_boards(include_archived=include_archived)
    current = kanban_db.get_current_board()
    proj_map = _projects_by_id()
    for b in boards:
        b["is_current"] = (b["slug"] == current)
        b["counts"] = _board_counts(b["slug"])
        # Live cards only — archived tasks are hidden from every default board view,
        # so counting them in the switcher badge would visibly disagree.
        b["total"] = sum(n for status, n in b["counts"].items() if status != "archived")
        b["default_workspace_kind"] = _default_workspace_kind(b)
        pid = b["project_id"] = b.get("project_id") or None
        proj = proj_map.get(pid) if pid else None
        b["project_name"] = proj.name if proj else None
    return {"boards": boards, "current": current}


def _validate_workdir(raw: str) -> str:
    """Board default_workdir must be an absolute, existing directory (400 otherwise)."""
    requested = Path(raw).expanduser()
    if not requested.is_absolute():
        raise HTTPException(status_code=400, detail="Project directory must be an absolute path.")
    if not requested.is_dir():
        raise HTTPException(status_code=400, detail="Project directory must be an existing directory.")
    return str(requested.resolve())


@router.post("/boards")
def create_board_endpoint(payload: CreateBoardBody):
    """Create a board. Idempotent — ``slug`` collision returns the existing one."""
    default_workdir = _validate_workdir(payload.default_workdir) if payload.default_workdir else None
    # A chosen project's primary repo becomes the default workdir unless one was passed explicitly.
    project_id, _pname, primary_path = _resolve_project(payload.project_id)
    if primary_path and not default_workdir:
        default_workdir = primary_path
    with _value_error_400():
        meta = kanban_db.create_board(
            payload.slug, default_workdir=default_workdir, project_id=project_id, **_board_display_kwargs(payload))
    if payload.switch:
        with _value_error_400():
            kanban_db.set_current_board(meta["slug"])
    return {"board": _annotate_board_meta(meta), "current": kanban_db.get_current_board()}


@router.patch("/boards/{slug}")
def rename_board(slug: str, payload: RenameBoardBody):
    """Update display metadata / default workdir / project scope (slug is immutable)."""
    normed = _existing_board_slug(slug)
    # write_board_metadata treats a falsy value as "clear", so pass "" through.
    default_workdir: Optional[str] = None
    if payload.default_workdir is not None:
        raw = payload.default_workdir.strip()
        default_workdir = _validate_workdir(raw) if raw else ""
    # A resolved project mirrors its repo into default_workdir unless the caller set it explicitly.
    project_id: Optional[str] = None
    if payload.project_id is not None:
        if payload.project_id.strip():
            project_id, _pname, primary_path = _resolve_project(payload.project_id)
            if primary_path and default_workdir is None:
                default_workdir = primary_path
        else:
            project_id = ""  # clear the scope
    meta = kanban_db.write_board_metadata(
        normed, default_workdir=default_workdir, project_id=project_id, **_board_display_kwargs(payload))
    return {"board": _annotate_board_meta(meta)}


@router.delete("/boards/{slug}")
def delete_board(slug: str, delete: bool = Query(False, description="Hard-delete instead of archive")):
    """Archive (default) or hard-delete a board."""
    with _value_error_400():
        res = kanban_db.remove_board(slug, archive=not delete)
    return {"result": res, "current": kanban_db.get_current_board()}


async def _run_transfer(fn, log_label: str):
    """Run a blocking kanban_transfer call off the event loop, mapping its errors
    to 404 (missing path) / 400 (invalid) / 500 (logged)."""
    try:
        return await asyncio.get_running_loop().run_in_executor(None, fn)
    except FileNotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc))
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    except Exception as exc:
        log.exception("%s failed", log_label)
        raise HTTPException(status_code=500, detail=str(exc))


@router.post("/boards/{slug}/export")
async def export_board_endpoint(slug: str, body: ExportBoardBody):
    """Write ``slug`` to a portable archive; return the path written."""
    from hermes_cli import kanban_transfer

    output = (body.output or "").strip()
    if not output:
        staging = kanban_db.kanban_home() / "kanban" / "board-exports"
        try:
            staging.mkdir(parents=True, exist_ok=True)
        except OSError as exc:
            raise HTTPException(status_code=500, detail=f"Could not create export directory: {exc}")
        output = str(staging / f"{slug}-{time.strftime('%Y%m%d-%H%M%S')}.tar.gz")
    return await _run_transfer(
        lambda: kanban_transfer.export_board(slug, output, include_attachments=body.attachments, include_logs=body.logs),
        f"POST /boards/{slug}/export")


@router.post("/boards/import")
async def import_board_endpoint(body: ImportBoardBody):
    """Import a board archive as a NEW board; return the landed board."""
    from hermes_cli import kanban_transfer

    archive = (body.archive or "").strip()
    if not archive:
        raise HTTPException(status_code=400, detail="archive path is required")
    result = await _run_transfer(
        lambda: kanban_transfer.import_board(archive, (body.slug or "").strip() or None, activate=body.switch),
        "POST /boards/import")
    return {**result, "current": kanban_db.get_current_board()}


@router.post("/boards/{slug}/switch")
def switch_board(slug: str):
    """Persist ``slug`` as the active board for CLI / slash-command parity
    (dashboard users pick boards client-side via localStorage)."""
    normed = _existing_board_slug(slug)
    kanban_db.set_current_board(normed)
    return {"current": normed}


# --- Profile metadata & description editing (kanban orchestrator) ------------

class DescribeBody(BaseModel):
    description: Optional[str] = None  # explicit user-authored text


class DescribeAutoBody(BaseModel):
    overwrite: bool = False


@router.get("/profiles")
def list_profile_roster():
    """Every installed profile with its description (profiles without one are
    still routable on name alone, just less precisely)."""
    with _errors_to_500("failed to list profiles"):
        from hermes_cli import profiles as profiles_mod
        profiles = profiles_mod.list_profiles()
    return {"profiles": [
        {"name": p.name, "is_default": bool(p.is_default), "model": p.model or "", "provider": p.provider or "",
         "description": p.description or "", "description_auto": bool(p.description_auto),
         "skill_count": int(p.skill_count or 0)}
        for p in profiles]}


@router.patch("/profiles/{profile_name}")
def update_profile_description(profile_name: str, payload: DescribeBody):
    """Set (``description_auto: false`` so the auto-describer won't overwrite it
    without ``--overwrite``) or clear (empty string) a profile's description."""
    with _errors_to_500("failed to update profile"):
        from hermes_cli import profiles as profiles_mod
        canon = profiles_mod.normalize_profile_name(profile_name)
        if canon == "default":
            from hermes_constants import get_hermes_home  # type: ignore
            profile_dir = Path(get_hermes_home())
        else:
            profile_dir = profiles_mod.get_profile_dir(canon)
        if not profile_dir.is_dir():
            raise HTTPException(status_code=404, detail=f"profile '{profile_name}' not found")
        text = (payload.description or "").strip()
        profiles_mod.write_profile_meta(profile_dir, description=text, description_auto=False)
    return {"ok": True, "profile": canon, "description": text}


@router.post("/profiles/{profile_name}/describe-auto")
def auto_describe_profile(profile_name: str, payload: DescribeAutoBody):
    """``hermes profile describe <name> --auto``: persist with ``description_auto: true``.
    Non-OK outcomes are NOT HTTP errors — the UI renders the reason inline."""
    with _errors_to_500("describer crashed"):
        from hermes_cli import profile_describer
        outcome = profile_describer.describe_profile(profile_name, overwrite=bool(payload.overwrite))
    return {"ok": bool(outcome.ok), "profile": outcome.profile_name, "reason": outcome.reason, "description": outcome.description}


# --- Decompose (built-in decomposer fan-out) ----------------------------------

class DecomposeBody(BaseModel):
    author: Optional[str] = None


@router.post("/tasks/{task_id}/decompose")
def decompose_task_endpoint(task_id: str, payload: DecomposeBody, board: Optional[str] = Query(None)):
    """Fan a triage task out into child tasks via the auxiliary LLM (``hermes kanban decompose``).
    Non-OK is NOT an HTTP error. Sync ``def`` → runs in the threadpool."""
    outcome = _run_aux(board, "kanban_decompose", "decompose_task", task_id, payload.author)
    return {
        "ok": bool(outcome.ok), "task_id": outcome.task_id, "reason": outcome.reason,
        "fanout": bool(outcome.fanout), "child_ids": outcome.child_ids or [], "new_title": outcome.new_title}


# --- Orchestration settings (kanban.orchestrator_profile / default_assignee /
#     auto_decompose / auto_promote_children) ----------------------------------

class OrchestrationSettingsBody(BaseModel):
    orchestrator_profile: Optional[str] = None
    default_assignee: Optional[str] = None
    auto_decompose: Optional[bool] = None
    auto_promote_children: Optional[bool] = None


_PROFILE_SETTINGS = ("orchestrator_profile", "default_assignee")


@router.get("/orchestration")
def get_orchestration_settings():
    """Current orchestration knobs from config.yaml plus the resolved effective
    values. An unset/unknown profile resolves to the active profile here; the
    decomposer prefers the root card's assignee in that case and uses the active
    profile only for cards with no assignee."""
    cfg = _load_config_or_empty()
    kanban_cfg = (cfg.get("kanban") or {}) if isinstance(cfg, dict) else {}
    explicit = {k: (kanban_cfg.get(k) or "").strip() for k in _PROFILE_SETTINGS}
    resolved = dict(explicit)
    try:
        from hermes_cli import profiles as profiles_mod
        active_default = profiles_mod.get_active_profile_name() or "default"
        for k, v in explicit.items():
            if not v or not profiles_mod.profile_exists(v):
                resolved[k] = active_default
    except Exception:
        active_default = "default"
        resolved = {k: v or active_default for k, v in resolved.items()}
    return {
        "orchestrator_profile": explicit["orchestrator_profile"],
        "default_assignee": explicit["default_assignee"],
        "auto_decompose": bool(kanban_cfg.get("auto_decompose", True)),
        "auto_promote_children": bool(kanban_cfg.get("auto_promote_children", True)),
        "resolved_orchestrator_profile": resolved["orchestrator_profile"],
        "resolved_default_assignee": resolved["default_assignee"],
        "active_profile": active_default}


def _validated_profile_name(raw: Optional[str], profiles_mod) -> str:
    """Strip a profile name; 400 if non-empty and unknown. Fails open when the lookup itself errors."""
    name = (raw or "").strip()
    if name and profiles_mod is not None:
        try:
            exists = profiles_mod.profile_exists(name)
        except Exception:
            exists = True
        if not exists:
            raise HTTPException(status_code=400, detail=f"profile '{name}' does not exist")
    return name


@router.put("/orchestration")
def set_orchestration_settings(payload: OrchestrationSettingsBody):
    """Update orchestration knobs in config.yaml. Only fields explicitly passed
    are written; empty profile strings clear the override."""
    with _errors_to_500("failed to load config"):
        from hermes_cli.config import load_config, save_config
        cfg = load_config() or {}
    kanban_section = cfg.setdefault("kanban", {})
    if not isinstance(kanban_section, dict):
        kanban_section = cfg["kanban"] = {}
    try:
        from hermes_cli import profiles as profiles_mod
    except Exception:
        profiles_mod = None  # type: ignore
    # Field order == write order (profiles validated first, then the booleans).
    for key, value in payload.model_dump(exclude_none=True).items():
        kanban_section[key] = _validated_profile_name(value, profiles_mod) if key in _PROFILE_SETTINGS else bool(value)
    with _errors_to_500("failed to save config"):
        save_config(cfg)
    return get_orchestration_settings()  # callers re-render from the resolved state


# --- WebSocket: /events?since=<event_id>&board=<slug> ------------------------

# Event tail poll interval: WAL + 300 ms polling is the simplest robust approach (negligible CPU).
_EVENT_POLL_SECONDS = 0.3


def _since_param(ws: WebSocket) -> Optional[int]:
    """The client's event cursor, or None when it sent none (or garbage).

    None starts the stream at the board's current tail. Only an explicit
    ``since`` replays history — including ``since=0``. A client that just
    opened the board already holds the snapshot; replaying every
    ``task_events`` row (200 per 300 ms) turned each open into a refetch storm.
    """
    raw = ws.query_params.get("since")
    if raw is None or not str(raw).strip():
        return None
    try:
        return max(0, int(raw))
    except (TypeError, ValueError):
        return None


def _ws_board(raw: Optional[str]) -> Optional[str]:
    try:
        return kanban_db._normalize_board_slug(raw) if raw else None
    except ValueError:
        return None


class _EventTail:
    """Per-socket ``task_events`` tailer. One SQLite connection, used/closed only on a
    dedicated single-thread executor (connections are thread-affine); reusing it avoids
    churning WAL/SHM sidecars while an idle dashboard polls."""

    def __init__(self, board: Optional[str]) -> None:
        self._board = board
        self._conn: Optional[sqlite3.Connection] = None
        self._executor: Optional[ThreadPoolExecutor] = None

    def _latest(self) -> int:
        """The board's current tail: the cursor a client without one starts from."""
        if self._conn is None:
            self._conn = kbc.connect(board=self._board)
        rows = self._conn.execute(
            "SELECT COALESCE(MAX(id), 0) AS m FROM task_events", ()
        ).fetchall()
        return int(rows[0]["m"]) if rows else 0

    def _fetch(self, cursor: int) -> tuple[int, list[dict]]:
        if self._conn is None:
            self._conn = kbc.connect(board=self._board)
        rows = self._conn.execute(
            "SELECT id, task_id, run_id, kind, payload, created_at "
            "FROM task_events WHERE id > ? ORDER BY id ASC LIMIT 200",
            (cursor,)).fetchall()
        out: list[dict] = []
        for r in rows:
            try:
                payload = json.loads(r["payload"]) if r["payload"] else None
            except Exception:
                payload = None
            out.append({**dict(r), "payload": payload})
        return (rows[-1]["id"] if rows else cursor), out

    def _close(self) -> None:
        if self._conn is not None:
            self._conn.close()
            self._conn = None

    def _run(self, fn, *args):
        if self._executor is None:
            self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="kanban-events")
        return asyncio.get_running_loop().run_in_executor(self._executor, fn, *args)

    async def latest(self) -> int:
        return await self._run(self._latest)

    async def poll(self, cursor: int) -> tuple[int, list[dict]]:
        return await self._run(self._fetch, cursor)

    async def shutdown(self) -> None:
        if self._executor is None:
            return
        try:
            await asyncio.get_running_loop().run_in_executor(self._executor, self._close)
        except Exception as exc:
            log.warning("Kanban event stream connection cleanup failed: %s", exc)
        finally:
            self._executor.shutdown(wait=True, cancel_futures=True)


@router.websocket("/events")
async def stream_events(ws: WebSocket):
    if not _ws_upgrade_authorized(ws):
        await ws.close(code=http_status.WS_1008_POLICY_VIOLATION)
        return
    await ws.accept()
    # Board is pinned at the handshake; the UI opens a new WS on board change
    # rather than reconciling two cursors mid-stream.
    tail = _EventTail(_ws_board(ws.query_params.get("board")))
    since = _since_param(ws)
    try:
        # Capture the tail at accept, before the first wait, so an event that
        # lands in that window is still delivered. A missing cursor must not
        # mean 0 — that replays the whole history.
        cursor = since if since is not None else await tail.latest()
        while True:
            # Race receive() against the poll interval so a disconnect is detected even when no
            # events flow (else idle boards leak poll tasks). Other client messages are ignored.
            try:
                msg = await asyncio.wait_for(ws.receive(), timeout=_EVENT_POLL_SECONDS)
                if msg["type"] == "websocket.disconnect":
                    return
            except asyncio.TimeoutError:
                pass  # no client message — poll the DB
            cursor, events = await tail.poll(cursor)
            if events:
                await ws.send_json({"events": events, "cursor": cursor})
    except WebSocketDisconnect:
        return
    except asyncio.CancelledError:
        return  # normal shutdown; CancelledError is a BaseException the handler below wouldn't quiet
    except Exception as exc:  # never crash the dashboard worker
        log.warning("Kanban event stream error: %s", exc)
        try:
            await ws.close()
        except Exception:
            pass
    finally:
        await tail.shutdown()