"""Kanban dispatcher daemon.

Single background thread that drives the kanban board:

  - Reclaims stale ``in_progress`` claims (TTL exceeded → run is recorded
    as ``crashed`` and the task drops back to ``ready``).
  - Promotes ``todo`` children whose parents have just become ``done``
    (defensive: the main promotion path runs inside
    :func:`kanban_storage.record_completion`).
  - Atomically claims the highest-priority ``ready`` task per profile
    (subject to per-profile concurrency limits).
  - Spawns a worker thread that:
      * Sets a per-task ``claim_token`` contextvar (so claim-scoped tools
        in :mod:`tools.kanban_tools` can verify they're inside the right
        worker).
      * Builds a ``system_override`` from the profile system hint, the
        kanban context block, and the completion contract.
      * Calls :func:`subagent_runner.run_subagent_structured` with a
        :class:`tool_policy.ToolPolicyContext` that has
        ``background_automation=True``, ``subagent_mode=True``,
        ``scheduled_job=True``.
      * On return: parses the trailing ``kanban-result`` block (or reads
        the in-band ``kanban_post_summary`` write), then calls
        :func:`kanban_storage.record_completion` or
        :func:`kanban_storage.record_failure`.
  - Drains completed worker threads, enforces deadlines, and runs
    retention (purge / trim_runs) on idle ticks.

Public surface (module-level):

  - ``start_dispatcher()`` / ``stop_dispatcher()`` — daemon lifecycle
    (idempotent). Called from :mod:`mugi_runner` after the chat encryption
    key is initialized, and from :mod:`mugi_shutdown_cleanup` on teardown.
  - ``nudge_dispatcher()`` — wake the loop early (used by the create_task
    HTTP / agent path so a freshly-created ready task starts immediately).
    Restarts the loop automatically when it has died but the board still
    has work.
  - ``ensure_dispatcher_running()`` — idempotent (re)start when the loop
    thread is down; used by nudge/health and safe to call from ops hooks.
  - ``dispatcher_health()`` — combined ``kanban_storage.health_summary``
    plus dispatcher-internal counters (worker counts, last tick).
  - ``tick_once()`` — single-shot tick; returned by tests instead of the
    full daemon loop.
  - ``current_claim_token()`` — used by claim-scoped tools to verify the
    caller is inside a kanban worker thread.
"""

from __future__ import annotations

import contextvars
import logging
import os
import threading
import time
from datetime import datetime, timezone
from typing import Any

import kanban_storage as ks
from utils.path_safety import _is_path_under_root, is_dangerous_path
from utils.kanban_activity import emit_kanban_result_parse_failure_activity
from utils.kanban_context import (
    FAILURE_CONTRACT_EMPTY,
    FAILURE_CONTRACT_INVALID_JSON,
    FAILURE_CONTRACT_MISSING,
    FAILURE_CONTRACT_SCHEMA_INVALID,
    FAILURE_CONTRACT_UNCLOSED,
    FAILURE_MISSING_CAPABILITY,
    FAILURE_POLICY_BLOCKED_CAPABILITY,
    FAILURE_TOOL_RUNTIME_ERROR,
    FAILURE_TOOL_TIMEOUT,
    FAILURE_UNKNOWN_TOOL,
    FAILURE_UNKNOWN_ASSIGNEE,
    FAILURE_DEADLINE_EXPIRED,
    FAILURE_MESH_PEER_ERROR,
    FAILURE_SPAWN_FAILED,
    FAILURE_VERIFIER_FAILED,
    KANBAN_WORKER_INFRA_TOOLS as _KANBAN_INFRA_TOOLS,
    build_completion_contract,
    build_retry_worker_brief,
    build_worker_context,
    parse_kanban_result_detailed,
    normalize_kanban_result_metadata,
    verify_kanban_result,
    salvage_handoff_action,
    merge_kanban_handoff,
    normalize_salvage_handoff,
)

_log = logging.getLogger(__name__)

# PID of the single process that is allowed to RUN the claiming dispatcher
# loop. Set when ``start()`` is invoked (the backend via ``start_dispatcher``;
# tests via a direct ``start()``). Any other process — e.g. a long-lived MCP
# server or CLI that calls ``nudge_dispatcher()`` — must NOT lazily spin up a
# competing dispatcher against the shared DB: doing so races task claims and
# can run council seats with stale code whose terminal hooks never fire,
# stranding synthesis until the honest-stop fallback. Non-owners simply rely
# on the owner's loop, which already polls every tick.
_owner_pid: int | None = None


def resolve_kanban_worker_cwd(task: dict[str, Any]) -> str:
    """Preferred working directory for a kanban worker (thread-safe via contextvar)."""
    task_meta = task.get("metadata") or {}
    if not isinstance(task_meta, dict):
        task_meta = {}

    def _validate_worker_cwd(candidate: str) -> str:
        """Return canonical path when safe; empty string when rejected (DEF-088)."""
        if not candidate or not os.path.isdir(candidate):
            return ""
        try:
            real = os.path.realpath(candidate)
        except OSError:
            _log.warning("kanban worker_cwd resolution failed for %r", candidate, exc_info=True)
            return ""
        if is_dangerous_path(real):
            _log.warning("kanban worker_cwd rejected dangerous path: %s", real)
            return ""
        try:
            from core.config import get_workspace_path

            ws = get_workspace_path()
            if ws:
                ws_real = os.path.realpath(os.path.expanduser(ws))
                if not _is_path_under_root(real, ws_real):
                    _log.warning(
                        "kanban worker_cwd rejected path outside workspace: %s (workspace=%s)",
                        real,
                        ws_real,
                    )
                    return ""
        except Exception:
            _log.warning("kanban worker_cwd workspace check failed", exc_info=True)
            return ""
        return real

    raw_cwd = str(task_meta.get("worker_cwd") or "").strip()
    if raw_cwd:
        expanded = os.path.expanduser(raw_cwd)
        if os.path.isabs(expanded):
            validated = _validate_worker_cwd(expanded)
            if validated:
                return validated
        else:
            try:
                from core.config import get_workspace_path

                ws = get_workspace_path()
                if ws:
                    candidate = os.path.join(ws, raw_cwd.lstrip("/"))
                    validated = _validate_worker_cwd(candidate)
                    if validated:
                        return validated
            except Exception:
                _log.debug("kanban worker_cwd relative resolve failed", exc_info=True)

    subpath = str(task_meta.get("workspace_subpath") or "").strip()
    if subpath:
        try:
            from core.config import get_workspace_path

            ws = get_workspace_path()
            if ws:
                candidate = os.path.join(ws, subpath.lstrip("/"))
                if os.path.isdir(candidate):
                    return candidate
        except Exception:
            pass

    try:
        from core.config import get_workspace_path

        ws = get_workspace_path()
        if ws and os.path.isdir(ws):
            return ws
    except Exception:
        pass
    return ""


def resolve_kanban_worker_timeout_seconds(
    task: dict[str, Any],
    *,
    deadline_mono: float | None = None,
) -> int:
    """Runtime budget for a kanban worker subagent.

    Kanban workers run full agent loops (many tool rounds, renders, file edits).
    They must not inherit short profile defaults such as ``ops_cautious`` (180 s).
    """
    if deadline_mono is not None:
        return max(15, int(deadline_mono - time.monotonic()))

    task_meta = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
    configured: int | None = None
    raw = task_meta.get("worker_timeout_seconds")
    if raw is not None:
        try:
            configured = max(15, int(raw))
        except (TypeError, ValueError):
            configured = None

    if configured is None:
        try:
            from core.config import get_kanban_config

            configured = max(
                15,
                int(get_kanban_config().get("default_worker_timeout_seconds") or 900),
            )
        except Exception:
            configured = 900

    claim_exp = str(task.get("claim_expires_at") or "").strip()
    if claim_exp:
        try:
            from datetime import datetime, timezone

            exp = datetime.strptime(claim_exp, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
            remaining = int((exp - datetime.now(timezone.utc)).total_seconds())
            if remaining > 0:
                return min(configured, remaining)
        except Exception:
            pass

    return configured


def _verify_task_metadata(task: dict[str, Any]) -> dict[str, Any]:
    from utils.kanban_context import task_metadata_for_verification

    return task_metadata_for_verification(task)


_CONTRACT_REQUEUE_REASONS = frozenset({
    FAILURE_CONTRACT_MISSING,
    FAILURE_CONTRACT_UNCLOSED,
    FAILURE_CONTRACT_EMPTY,
    FAILURE_CONTRACT_INVALID_JSON,
    FAILURE_CONTRACT_SCHEMA_INVALID,
    FAILURE_VERIFIER_FAILED,
})


def _strict_cfg() -> dict[str, bool]:
    try:
        from core.config import get_kanban_config

        cfg = get_kanban_config()
    except Exception:
        cfg = {}
    return {
        "strict_mode": bool(cfg.get("strict_mode_enabled", False)),
        "shadow_mode": bool(cfg.get("strict_shadow_mode_enabled", False)),
        "legacy_mapping": bool(cfg.get("strict_legacy_outcome_mapping", True)),
    }


def _native_result_handoff_enabled() -> bool:
    """Stage B flag (fail-closed): promote the native result tool to first-class handoff.

    On any config read error returns False so the pre-Stage-B fence/in-band behavior stands.
    """
    try:
        from core.config import get_kanban_config

        return bool(get_kanban_config().get("native_result_handoff", False))
    except Exception:
        _log.debug("native_result_handoff flag read failed", exc_info=True)
        return False


def _reason_to_outcome(reason: str) -> str:
    """Strict-mode outcome mapping — contract misses requeue; policy misses terminal."""
    if reason in _CONTRACT_REQUEUE_REASONS:
        return ks.OUTCOME_CRASHED
    if reason in {
        FAILURE_MISSING_CAPABILITY,
        FAILURE_POLICY_BLOCKED_CAPABILITY,
        FAILURE_UNKNOWN_TOOL,
        FAILURE_UNKNOWN_ASSIGNEE,
        FAILURE_DEADLINE_EXPIRED,
    }:
        return ks.OUTCOME_GAVE_UP
    if reason == FAILURE_TOOL_TIMEOUT:
        return ks.OUTCOME_TIMED_OUT
    if reason == FAILURE_TOOL_RUNTIME_ERROR:
        return ks.OUTCOME_CRASHED
    return ks.OUTCOME_CRASHED


def _resolve_typed_outcome(reason: str, *, legacy_default: str, cfg: dict[str, bool]) -> str:
    """Rollout-aware outcome mapping for typed failures."""
    if cfg.get("shadow_mode") and not cfg.get("strict_mode"):
        return legacy_default
    if cfg.get("strict_mode") and not cfg.get("legacy_mapping"):
        return _reason_to_outcome(reason)
    return legacy_default


def _legacy_outcome_for_typed_reason(reason: str) -> str:
    """Default storage outcome before strict-mode remapping."""
    if reason in _CONTRACT_REQUEUE_REASONS:
        return ks.OUTCOME_CRASHED
    if reason in {
        FAILURE_MISSING_CAPABILITY,
        FAILURE_POLICY_BLOCKED_CAPABILITY,
        FAILURE_UNKNOWN_TOOL,
        FAILURE_UNKNOWN_ASSIGNEE,
        FAILURE_DEADLINE_EXPIRED,
    }:
        return ks.OUTCOME_GAVE_UP
    if reason == FAILURE_TOOL_TIMEOUT:
        return ks.OUTCOME_TIMED_OUT
    return ks.OUTCOME_CRASHED


def enrich_kanban_run_row(run: dict[str, Any]) -> dict[str, Any]:
    """Add API-facing typed_reason / failure_class / next_action / tail_excerpt."""
    from utils.kanban_context import FAILURE_CLASS_OPERATOR, classify_failure_class

    row = dict(run)
    md = row.get("metadata")
    if isinstance(md, dict):
        failure_reason = md.get("failure_reason")
        if isinstance(failure_reason, str) and failure_reason.strip():
            reason_v = failure_reason.strip()
            row["typed_reason"] = reason_v
            from utils.kanban_context import humanize_typed_failure_reason

            label = humanize_typed_failure_reason(reason_v)
            if label:
                row["typed_reason_label"] = label
            stored_class = md.get("failure_class")
            if isinstance(stored_class, str) and stored_class.strip():
                row["failure_class"] = stored_class.strip()
            else:
                row["failure_class"] = classify_failure_class(reason_v)
        next_action = md.get("next_action")
        if isinstance(next_action, str) and next_action.strip():
            row["next_action"] = next_action.strip()
        elif isinstance(failure_reason, str) and failure_reason.strip():
            fallback = _next_action_for_reason(failure_reason.strip())
            if fallback:
                row["next_action"] = fallback
        tail_excerpt = md.get("tail_excerpt")
        if isinstance(tail_excerpt, str) and tail_excerpt.strip():
            row["tail_excerpt"] = tail_excerpt.strip()
        # Surface salvage provenance so operators/assembly can tell a harness-synthesized or
        # in-band handoff apart from native trailing-fence compliance (salvage is a backup,
        # never the silent happy path).
        handoff_source = md.get("handoff_source")
        if isinstance(handoff_source, str) and handoff_source.strip():
            src = handoff_source.strip()
            row["handoff_source"] = src
            # A first-class native-tool submission (Stage B "tool_submit") is the worker
            # completing cleanly, NOT a harness fallback — keep harness_synthesized False so it
            # doesn't contradict the flag's documented meaning in persisted metadata + Swift UI.
            # salvage_stats applies the same rule via is_first_class_handoff_source.
            row["harness_synthesized"] = not ks.is_first_class_handoff_source(src)
        elif md.get("completion_salvage") is True:
            row["handoff_source"] = "council_researcher_salvage"
            row["harness_synthesized"] = True
    if "failure_class" not in row and str(row.get("outcome") or "").strip() == ks.OUTCOME_CANCELLED:
        row["failure_class"] = FAILURE_CLASS_OPERATOR
    return row


def _try_inband_completion(
    *,
    inband_summary: str,
    inband_metadata: dict[str, Any],
    task_meta: dict[str, Any],
) -> tuple[dict[str, Any] | None, str, dict[str, Any]]:
    """When the fence is missing but kanban_post_summary succeeded, salvage handoff."""
    summary = (inband_summary or "").strip()
    if not summary:
        return None, "", {}
    md = dict(inband_metadata or {})
    # The run row's metadata carries dispatcher-owned bookkeeping (progress_log
    # rows with per-tool result previews — easily >8KB on a chatty run). That
    # is NOT part of the worker's handoff: passing it through would round-trip
    # it into record_completion's metadata validation and deterministically
    # fail the 8192-byte limit (incident kt_1de7d1f3d14b4e: completion lost,
    # root pinned in_progress, 7 children frozen). The terminal merge in
    # kanban_storage preserves progress_log on the run row independently.
    md.pop("progress_log", None)
    # Stage B: a deliberate native-tool submission (kanban_post_summary stamps _handoff_channel
    # ="tool") standing in for a missing/truncated fence is a first-class completion, not salvage —
    # but only when the flag is on. The internal channel marker never persists into the handoff.
    deliberate_tool = str(md.pop("_handoff_channel", "") or "").strip().lower() == "tool"
    if deliberate_tool and _native_result_handoff_enabled():
        source = "tool_submit"
    else:
        # Provenance: this handoff substitutes an in-band summary for the missing trailing fence.
        # Stamped so salvage stays observable and never indistinguishable from native compliance.
        source = "inband_summary"
    synthetic = {
        "summary": summary,
        "metadata": {**md, "handoff_source": source},
    }
    ok, reason, detail = verify_kanban_result(synthetic, task_metadata=task_meta)
    if not ok:
        return None, reason, {"verifier_detail": detail}
    return synthetic, "ok", synthetic.get("metadata") or {}


def _resolve_kanban_verifier_failure(
    *,
    verifier_reason: str,
    verifier_detail: str,
    metadata: dict[str, Any],
    strict_cfg: dict[str, bool],
    kanban_worker: bool = False,
) -> tuple[bool, dict[str, Any]]:
    """Return (should_fail, metadata_updates). Kanban workers always fail on verifier miss."""
    if (
        not kanban_worker
        and strict_cfg.get("shadow_mode")
        and not strict_cfg.get("strict_mode")
    ):
        updates = dict(metadata)
        updates["verifier_shadow_reason"] = verifier_reason
        updates["verifier_shadow_detail"] = verifier_detail
        return False, updates
    return True, metadata


def _next_action_for_reason(reason: str) -> str:
    return {
        FAILURE_CONTRACT_MISSING: "Return a trailing ```kanban-result fenced JSON block.",
        FAILURE_CONTRACT_UNCLOSED: "Close the kanban-result fence and ensure valid JSON object.",
        FAILURE_CONTRACT_EMPTY: "Populate summary and metadata in the trailing kanban-result block.",
        FAILURE_CONTRACT_INVALID_JSON: "Fix invalid JSON syntax in kanban-result payload.",
        FAILURE_CONTRACT_SCHEMA_INVALID: "Use {'summary': string, 'metadata': object} in kanban-result.",
        FAILURE_VERIFIER_FAILED: "Provide a dense summary and required metadata fields.",
        FAILURE_MISSING_CAPABILITY: "Recreate task with required tools in metadata.tool_names.",
        FAILURE_POLICY_BLOCKED_CAPABILITY: "Enable required policy capability or use allowed tools.",
        FAILURE_UNKNOWN_TOOL: "Use find_tools and tool index to pick a valid tool.",
        FAILURE_TOOL_TIMEOUT: "Split work and retry with shorter bounded substeps.",
        FAILURE_TOOL_RUNTIME_ERROR: "Adjust tool args or choose a different compatible tool.",
        FAILURE_UNKNOWN_ASSIGNEE: "Recreate the task with a valid kanban assignee profile.",
        FAILURE_DEADLINE_EXPIRED: "Extend the deadline or recreate the task with more time.",
        FAILURE_MESH_PEER_ERROR: "Check mesh peer connectivity and retry the task.",
        FAILURE_SPAWN_FAILED: "Retry the task; if it persists, check dispatcher logs.",
    }.get(reason, "Retry with corrected inputs.")


def _typed_failure_payload(reason: str, detail: str) -> tuple[str, dict[str, Any]]:
    from utils.kanban_context import classify_failure_class

    msg = f"[typed_failure:{reason}] {detail}".strip()
    return msg[:1000], {
        "failure_reason": reason,
        "failure_class": classify_failure_class(reason),
        "next_action": _next_action_for_reason(reason),
    }


def _cancelled_failure_payload(
    reason_key: str,
    detail: str,
    *,
    extra: dict[str, Any] | None = None,
) -> tuple[str, dict[str, Any]]:
    from utils.kanban_context import classify_failure_class

    msg = detail.strip()[:1000]
    meta: dict[str, Any] = {
        "failure_reason": reason_key,
        "failure_class": classify_failure_class(reason_key),
    }
    if extra:
        meta.update(extra)
    return msg, meta


# ---------------------------------------------------------------------------
# Worker-scoped claim token (claim-scoped tool authorization)
# ---------------------------------------------------------------------------
#
# Worker threads call .run() inside a Context that sets _claim_token_var to
# the active claim_token. Claim-scoped tools (kanban_post_summary,
# kanban_post_progress) read this via `current_claim_token()` and refuse
# when it's missing or doesn't match the task they were called against.
# ---------------------------------------------------------------------------
_claim_token_var: contextvars.ContextVar[tuple[str, str] | None] = contextvars.ContextVar(
    "mugi_kanban_claim_token", default=None
)
_kanban_worker_cwd_var: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "mugi_kanban_worker_cwd", default=None
)
_kanban_worker_search_brief_var: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "mugi_kanban_worker_search_brief", default=None
)


def current_kanban_worker_cwd() -> str | None:
    """Preferred CWD for the active kanban worker thread (empty when unset)."""
    value = _kanban_worker_cwd_var.get()
    return value if value else None


def _set_kanban_worker_cwd(path: str) -> contextvars.Token:
    return _kanban_worker_cwd_var.set(path)


def _reset_kanban_worker_cwd(token: contextvars.Token) -> None:
    try:
        _kanban_worker_cwd_var.reset(token)
    except Exception:
        pass


def current_kanban_worker_search_brief() -> str | None:
    """Compact search/topic string pinned at kanban worker start (survives summarization)."""
    value = _kanban_worker_search_brief_var.get()
    return value if value else None


def _set_kanban_worker_search_brief(text: str) -> contextvars.Token:
    brief = (text or "").strip()[:500]
    return _kanban_worker_search_brief_var.set(brief or None)


def _reset_kanban_worker_search_brief(token: contextvars.Token) -> None:
    try:
        _kanban_worker_search_brief_var.reset(token)
    except Exception:
        pass


def current_claim_token() -> tuple[str, str] | None:
    """Return ``(task_id, claim_token)`` when called from inside a kanban
    worker thread, else None.
    """
    return _claim_token_var.get()


def _set_claim_token(task_id: str, token: str) -> contextvars.Token:
    return _claim_token_var.set((task_id, token))


def _reset_claim_token(token: contextvars.Token) -> None:
    try:
        _claim_token_var.reset(token)
    except Exception:
        pass


# ---------------------------------------------------------------------------
# In-flight worker registry (per-task stop flags + threads)
# ---------------------------------------------------------------------------

class _InFlight:
    """Bookkeeping for a single running worker."""

    __slots__ = (
        "task_id",
        "claim_token",
        "assignee",
        "thread",
        "stop_flag",
        "started_mono",
        "deadline_mono",
        "deadline_stop_mono",
    )

    def __init__(
        self,
        *,
        task_id: str,
        claim_token: str,
        assignee: str,
        thread: threading.Thread,
        stop_flag: threading.Event,
        deadline_mono: float | None,
    ) -> None:
        self.task_id = task_id
        self.claim_token = claim_token
        self.assignee = assignee
        self.thread = thread
        self.stop_flag = stop_flag
        self.started_mono = time.monotonic()
        self.deadline_mono = deadline_mono
        self.deadline_stop_mono: float | None = None


# ---------------------------------------------------------------------------
# The dispatcher
# ---------------------------------------------------------------------------

def _deadline_join_grace_seconds(cfg: dict[str, Any] | None) -> float:
    if cfg is None:
        try:
            from core.config import get_kanban_config

            cfg = get_kanban_config()
        except Exception:
            cfg = {}
    try:
        return max(0.0, float(cfg.get("deadline_join_grace_seconds", 120)))
    except (TypeError, ValueError):
        return 120.0


class _Dispatcher:
    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._inflight: dict[str, _InFlight] = {}  # task_id -> _InFlight
        self._deadline_forced_release: set[str] = set()
        self._stop_event = threading.Event()
        self._wake_event = threading.Event()
        self._loop_thread: threading.Thread | None = None
        self._tick_count = 0
        self._last_tick_at: str = ""
        self._last_retention_mono = 0.0
        self._retention_interval = 300.0  # seconds

    # ------------------------------ lifecycle ---------------------------------

    def _loop_alive(self) -> bool:
        thread = self._loop_thread
        return thread is not None and thread.is_alive()

    @staticmethod
    def _board_needs_dispatcher(summary: dict[str, Any]) -> bool:
        """True when durable board work exists and the daemon should be ticking."""
        counts = summary.get("counts") if isinstance(summary.get("counts"), dict) else {}
        if int(summary.get("in_flight_total") or 0) > 0:
            return True
        if int(summary.get("queue_depth") or 0) > 0:
            return True
        for key in (ks.STATUS_READY, ks.STATUS_IN_PROGRESS, ks.STATUS_TODO):
            if int(counts.get(key) or 0) > 0:
                return True
        return False

    def ensure_running(self) -> bool:
        """Start (or restart) the daemon loop when kanban is enabled.

        Returns True when the loop thread is alive after this call.
        """
        if self._stop_event.is_set():
            return False
        if self._loop_alive():
            return True
        # Only the process that explicitly started the dispatcher may (re)start
        # the claiming loop. A nudge/health-check from a process that never
        # called start_dispatcher (MCP server, CLI) must NOT lazily spin up a
        # competing dispatcher against the shared DB. ``_owner_pid is None``
        # (nobody opted in here) is therefore also refused — only an explicit
        # ``start()`` claims ownership.
        if _owner_pid != os.getpid():
            return False
        if self._loop_thread is not None and not self._loop_thread.is_alive():
            _log.warning("Kanban dispatcher loop thread died; restarting")
            self._loop_thread = None
        started = self.start()
        if started:
            _log.info("Kanban dispatcher loop restarted")
        return self._loop_alive()

    def start(self) -> bool:
        """Start the daemon loop. Returns True when freshly started, False
        when already running. Idempotent."""
        if self._loop_alive():
            return False
        try:
            from core.config import get_kanban_config
            cfg = get_kanban_config()
        except Exception:
            cfg = {"enabled": True}
        if not cfg.get("enabled", True):
            _log.info("Kanban dispatcher disabled in config; not starting")
            return False
        # Claim process ownership: this is now the one process permitted to run
        # the claiming loop (see _owner_pid docstring above).
        global _owner_pid
        _owner_pid = os.getpid()
        self._stop_event.clear()
        self._wake_event.clear()
        self._loop_thread = threading.Thread(
            target=self._loop, daemon=True, name="kanban-dispatcher"
        )
        self._loop_thread.start()
        _log.info("Kanban dispatcher started")
        return True

    def stop(self, *, join_seconds: float = 5.0) -> None:
        """Stop the daemon loop and signal all workers. Idempotent."""
        self._stop_event.set()
        self._wake_event.set()
        thread = self._loop_thread
        if thread is not None and thread.is_alive():
            thread.join(timeout=join_seconds)
        self._loop_thread = None
        # Signal all workers and give them a brief grace period.
        with self._lock:
            inflight = list(self._inflight.values())
        for w in inflight:
            w.stop_flag.set()
        deadline = time.monotonic() + join_seconds
        for w in inflight:
            remaining = max(0.0, deadline - time.monotonic())
            if remaining <= 0:
                break
            try:
                w.thread.join(timeout=remaining)
            except Exception:
                pass
        # Mark anything still pending as cancelled (storage helper).
        try:
            ks.mark_inflight_cancelled()
        except Exception:
            _log.debug("mark_inflight_cancelled raised", exc_info=True)
        _log.info("Kanban dispatcher stopped")

    def nudge(self) -> None:
        """Wake the loop early; restart the loop first if it died."""
        self.ensure_running()
        self._wake_event.set()

    # ------------------------------ tick / loop -------------------------------

    def _loop(self) -> None:
        try:
            from core.config import get_kanban_config
        except Exception:
            get_kanban_config = lambda: {}  # noqa: E731
        try:
            while not self._stop_event.is_set():
                try:
                    cfg = get_kanban_config()
                except Exception:
                    cfg = {}
                tick_seconds = max(1.0, float(cfg.get("tick_seconds", 5)))
                try:
                    self.tick_once(cfg=cfg)
                except Exception:
                    _log.debug("kanban dispatcher tick raised", exc_info=True)
                # Wait for the next tick OR a nudge.
                self._wake_event.wait(timeout=tick_seconds)
                self._wake_event.clear()
        except Exception:
            _log.error("Kanban dispatcher loop exited unexpectedly", exc_info=True)
        finally:
            if not self._stop_event.is_set():
                _log.error(
                    "Kanban dispatcher loop stopped while kanban was still enabled; "
                    "call ensure_dispatcher_running() or nudge_dispatcher() to recover"
                )

    def tick_once(self, *, cfg: dict[str, Any] | None = None) -> dict[str, int]:
        """Run a single dispatcher tick. Returns a small counters dict for
        tests / health introspection."""
        if cfg is None:
            try:
                from core.config import get_kanban_config
                cfg = get_kanban_config()
            except Exception:
                cfg = {}

        counters = {
            "reclaimed": 0,
            "promoted": 0,
            "claimed": 0,
            "drained": 0,
            "deadline_killed": 0,
            "deadline_slot_freed": 0,
            "council_deadline_killed": 0,
            "retention_purged": 0,
            "retention_runs_trimmed": 0,
        }
        self._tick_count += 1
        self._last_tick_at = datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

        try:
            from utils.council_orchestrator import reconcile_council_batch_deadlines

            counters["council_deadline_killed"] = reconcile_council_batch_deadlines()
        except Exception:
            _log.debug("council deadline reconcile failed", exc_info=True)

        # 1) Reclaim stale claims.
        counters["reclaimed"] = self._reclaim_stale(cfg)

        # 2) Drain completed/finished workers (also enforces deadlines).
        drained, killed, slot_freed = self._drain_workers(cfg)
        counters["drained"] = drained
        counters["deadline_killed"] = killed
        counters["deadline_slot_freed"] = slot_freed

        # 3a) Local-first mesh routing: rewrite mesh-eligible tasks to a
        # trusted peer when the matching local profile is at concurrency cap.
        try:
            self._route_mesh_eligible(cfg)
        except Exception:
            _log.debug("mesh-eligible routing tick raised", exc_info=True)

        # 3b) Try to claim new tasks per profile, up to per-profile cap.
        counters["claimed"] = self._claim_round(cfg)

        # 4) Promote ready children defensively.
        # (Main path is inside record_completion; this is a safety net for
        # crash-recovered runs that flipped a parent to done outside the
        # normal flow, e.g. via direct DB tooling.)
        counters["promoted"] = self._opportunistic_promote()

        # 5) Periodic retention.
        now = time.monotonic()
        if now - self._last_retention_mono >= self._retention_interval:
            counters["retention_purged"], counters["retention_runs_trimmed"] = self._retention(cfg)
            self._last_retention_mono = now

        return counters

    # ------------------------------ tick helpers ------------------------------

    def _reclaim_stale(self, cfg: dict[str, Any]) -> int:
        try:
            stale = ks.find_stale_claims(limit=20)
        except Exception:
            _log.debug("find_stale_claims failed", exc_info=True)
            return 0
        if not stale:
            return 0
        reclaimed = 0
        for task in stale:
            tid = task.get("id") or ""
            tok = task.get("claim_token") or ""
            if not tid or not tok:
                continue
            # If we still own this in-process, the worker is alive — skip.
            with self._lock:
                if tid in self._inflight:
                    continue
            try:
                from utils.kanban_context import classify_failure_class

                ks.record_failure(
                    tid,
                    claim_token=tok,
                    outcome=ks.OUTCOME_CRASHED,
                    error="claim TTL exceeded; worker presumed crashed",
                    handle_id=task.get("claim_handle") or "",
                    metadata={
                        "failure_reason": "claim_ttl_exceeded",
                        "failure_class": classify_failure_class("claim_ttl_exceeded"),
                    },
                )
                reclaimed += 1
            except Exception:
                _log.debug("reclaim record_failure failed for %s", tid, exc_info=True)
        return reclaimed

    def _opportunistic_promote(self) -> int:
        # Walk recently-completed parents. We use a small bounded sweep — the
        # primary promote happens inside record_completion's transaction.
        # This catches DAG operations done outside that path.
        try:
            recently = ks.list_tasks(status=ks.STATUS_DONE, limit=20)
        except Exception:
            return 0
        promoted = 0
        for parent in recently:
            try:
                p = ks.promote_ready_children(parent.get("id") or "")
                promoted += len(p)
            except Exception:
                _log.debug("promote_ready_children failed", exc_info=True)
        return promoted

    def _route_mesh_eligible(self, cfg: dict[str, Any]) -> int:
        """Local-first router: rewrite ``mesh-eligible`` ready tasks whose
        local profile is at concurrency cap to a trusted peer. Returns the
        number of tasks rewritten. No-op when no trusted peers are
        available, when the local profile still has capacity, or when the
        task is already mesh-routed.
        """
        per_profile = cfg.get("concurrency") or {}
        if not isinstance(per_profile, dict):
            per_profile = {}
        try:
            ready = ks.list_tasks(status=ks.STATUS_READY, limit=200)
        except Exception:
            return 0
        candidates = []
        for t in ready:
            if (t.get("assignee") or "").startswith("mesh:"):
                continue
            tags = t.get("tags") or []
            if not isinstance(tags, list) or "mesh-eligible" not in tags:
                continue
            candidates.append(t)
        if not candidates:
            return 0

        per_profile_inflight: dict[str, int] = {}
        with self._lock:
            for w in self._inflight.values():
                per_profile_inflight[w.assignee] = per_profile_inflight.get(w.assignee, 0) + 1

        try:
            from mesh import get_mesh_manager
            manager = get_mesh_manager()
        except Exception:
            manager = None
        if manager is None:
            return 0
        try:
            sessions = manager.list_sessions()
        except Exception:
            sessions = []
        peers = [
            s for s in sessions
            if s.get("trusted_for_tasks") and s.get("peer_fingerprint")
        ]
        if not peers:
            return 0

        rewritten = 0
        for task in candidates:
            assignee = (task.get("assignee") or "general").strip() or "general"
            cap = int(per_profile.get(assignee, 1))
            if cap <= 0:
                continue
            inflight = per_profile_inflight.get(assignee, 0)
            if inflight < cap:
                continue
            peer_fp = peers[rewritten % len(peers)].get("peer_fingerprint") or ""
            if not peer_fp:
                continue
            try:
                ks.set_assignee(task["id"], f"mesh:{peer_fp}", force=False)
                rewritten += 1
            except Exception:
                _log.debug("mesh route set_assignee failed", exc_info=True)
        return rewritten

    def _claim_round(self, cfg: dict[str, Any]) -> int:
        per_profile = cfg.get("concurrency") or {}
        if not isinstance(per_profile, dict):
            per_profile = {}
        # Profiles that have ready work. We don't need to enumerate every
        # profile; we only consider those that appear in the ready queue.
        try:
            ready = ks.list_tasks(status=ks.STATUS_READY, limit=200)
        except Exception:
            return 0
        if not ready:
            return 0
        per_profile_inflight: dict[str, int] = {}
        with self._lock:
            for w in self._inflight.values():
                per_profile_inflight[w.assignee] = per_profile_inflight.get(w.assignee, 0) + 1

        seen_assignees: set[str] = set()
        claimed = 0
        for task in ready:
            assignee = (task.get("assignee") or "general").strip() or "general"
            if assignee in seen_assignees:
                # The next ready task for this assignee will be picked up on
                # a subsequent tick. We process at most one per profile per
                # tick to keep round-robin fairness.
                continue
            cap = int(per_profile.get(assignee, 1))
            if cap <= 0:
                seen_assignees.add(assignee)
                continue
            inflight = per_profile_inflight.get(assignee, 0)
            if inflight >= cap:
                seen_assignees.add(assignee)
                continue
            seen_assignees.add(assignee)
            ttl = int(cfg.get("default_claim_ttl_seconds", 900))
            try:
                claimed_task = ks.claim_task(
                    assignee=assignee,
                    handle_id="",  # storage will allocate one
                    ttl_seconds=ttl,
                )
            except Exception:
                _log.debug("claim_task raised", exc_info=True)
                continue
            if claimed_task is None:
                continue
            self._spawn_worker(claimed_task, cfg)
            claimed += 1
            per_profile_inflight[assignee] = inflight + 1
        return claimed

    def _worker_outcome_suppressed(self, task_id: str) -> bool:
        tid = (task_id or "").strip()
        if not tid:
            return False
        with self._lock:
            return tid in self._deadline_forced_release

    def _clear_deadline_force_release(self, task_id: str) -> None:
        tid = (task_id or "").strip()
        if not tid:
            return
        with self._lock:
            self._deadline_forced_release.discard(tid)

    def _force_deadline_slot_release(self, worker: _InFlight, cfg: dict[str, Any]) -> bool:
        """Free a per-profile slot when a deadline worker ignores cooperative stop."""
        tid = (worker.task_id or "").strip()
        if not tid:
            return False
        with self._lock:
            if tid not in self._inflight:
                return False
            self._inflight.pop(tid, None)
            self._deadline_forced_release.add(tid)
        try:
            task = ks.get_task(tid)
        except Exception:
            task = None
        if not task or (task.get("status") or "").strip() != ks.STATUS_IN_PROGRESS:
            return True
        claim_token = str(task.get("claim_token") or "").strip()
        if not claim_token:
            return True
        handle_id = str(task.get("claim_handle") or "")
        strict_cfg = _strict_cfg()
        err, meta = _typed_failure_payload(
            FAILURE_DEADLINE_EXPIRED,
            (
                "task deadline exceeded; worker thread did not exit within "
                f"{int(_deadline_join_grace_seconds(cfg))}s grace"
            ),
        )
        meta["deadline_slot_force_released"] = True
        self._record_outcome_locked(
            task_id=tid,
            claim_token=claim_token,
            outcome=_resolve_typed_outcome(
                FAILURE_DEADLINE_EXPIRED,
                legacy_default=_legacy_outcome_for_typed_reason(FAILURE_DEADLINE_EXPIRED),
                cfg=strict_cfg,
            ),
            error=err,
            handle_id=handle_id,
            run_id="",
            summary="",
            metadata=meta,
        )
        return True

    def _drain_workers(self, cfg: dict[str, Any] | None = None) -> tuple[int, int, int]:
        """Reap finished worker threads and enforce deadlines (WCR-07 slot release)."""
        drained = 0
        killed = 0
        slot_freed = 0
        now = time.monotonic()
        grace = _deadline_join_grace_seconds(cfg)
        to_force: list[_InFlight] = []
        with self._lock:
            for tid, w in list(self._inflight.items()):
                if not w.thread.is_alive():
                    self._inflight.pop(tid, None)
                    drained += 1
                    continue
                if w.deadline_mono is None or now < w.deadline_mono:
                    continue
                if not w.stop_flag.is_set():
                    w.stop_flag.set()
                    w.deadline_stop_mono = now
                    killed += 1
                    continue
                stop_mono = w.deadline_stop_mono
                if stop_mono is not None and now >= stop_mono + grace:
                    to_force.append(w)
        for worker in to_force:
            if self._force_deadline_slot_release(worker, cfg or {}):
                slot_freed += 1
        return drained, killed, slot_freed

    def _retention(self, cfg: dict[str, Any]) -> tuple[int, int]:
        rd = int(cfg.get("retention_days", 14))
        mr = int(cfg.get("max_runs_per_task", 50))
        purged = 0
        trimmed = 0
        try:
            purged = ks.purge_terminal(rd)
        except Exception:
            _log.debug("purge_terminal failed", exc_info=True)
        try:
            trimmed = ks.trim_runs(mr)
        except Exception:
            _log.debug("trim_runs failed", exc_info=True)
        return purged, trimmed

    # ------------------------------ worker spawn ------------------------------

    def _spawn_worker(self, task: dict[str, Any], cfg: dict[str, Any]) -> None:
        task_id = task.get("id") or ""
        claim_token = task.get("claim_token") or ""
        assignee = (task.get("assignee") or "general").strip() or "general"
        if not task_id or not claim_token:
            return

        # Validate assignee against SUBAGENT_PROFILES; unknown profile is a
        # final spawn_failed (not retried).
        try:
            from subagent_runner import SUBAGENT_PROFILES
        except Exception:
            SUBAGENT_PROFILES = {}
        if assignee not in SUBAGENT_PROFILES and not assignee.startswith("mesh:"):
            from utils.kanban_context import FAILURE_UNKNOWN_ASSIGNEE

            err, meta = _typed_failure_payload(
                FAILURE_UNKNOWN_ASSIGNEE,
                f"unknown assignee profile '{assignee}'",
            )
            strict_cfg = _strict_cfg()
            self._record_outcome_locked(
                task_id=task_id,
                claim_token=claim_token,
                outcome=_resolve_typed_outcome(
                    FAILURE_UNKNOWN_ASSIGNEE,
                    legacy_default=_legacy_outcome_for_typed_reason(
                        FAILURE_UNKNOWN_ASSIGNEE
                    ),
                    cfg=strict_cfg,
                ),
                error=err,
                handle_id="",
                run_id="",
                summary="",
                metadata=meta,
            )
            return

        # Compute deadline.
        deadline_mono: float | None = None
        deadline_str = task.get("deadline_at") or ""
        if deadline_str:
            try:
                dl = datetime.strptime(deadline_str, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
                seconds_remaining = (dl - datetime.now(tz=timezone.utc)).total_seconds()
                if seconds_remaining <= 0:
                    err, meta = _typed_failure_payload(
                        FAILURE_DEADLINE_EXPIRED,
                        "deadline already passed when claim was attempted",
                    )
                    strict_cfg = _strict_cfg()
                    self._record_outcome_locked(
                        task_id=task_id,
                        claim_token=claim_token,
                        outcome=_resolve_typed_outcome(
                            FAILURE_DEADLINE_EXPIRED,
                            legacy_default=_legacy_outcome_for_typed_reason(
                                FAILURE_DEADLINE_EXPIRED
                            ),
                            cfg=strict_cfg,
                        ),
                        error=err,
                        handle_id=str(task.get("claim_handle") or ""),
                        run_id="",
                        summary="",
                        metadata=meta,
                    )
                    return
                deadline_mono = time.monotonic() + seconds_remaining
            except Exception:
                deadline_mono = None

        stop_flag = threading.Event()
        worker_args = {
            "task": task,
            "stop_flag": stop_flag,
            "deadline_mono": deadline_mono,
        }

        # Spawn in a Context so contextvars are isolated and the claim token
        # set in the worker flow doesn't leak out.
        ctx = contextvars.copy_context()
        is_mesh = assignee.startswith("mesh:")
        target_fn = self._run_mesh_worker_sync if is_mesh else self._run_worker_sync

        def _runner() -> None:
            try:
                ctx.run(target_fn, worker_args)
            except Exception:
                _log.debug("kanban worker terminated abnormally", exc_info=True)

        thread = threading.Thread(
            target=_runner,
            daemon=True,
            name=f"kanban-{'mesh-' if is_mesh else ''}worker-{task_id}",
        )
        with self._lock:
            self._inflight[task_id] = _InFlight(
                task_id=task_id,
                claim_token=claim_token,
                assignee=assignee,
                thread=thread,
                stop_flag=stop_flag,
                deadline_mono=deadline_mono,
            )
        thread.start()

    def _run_worker_sync(self, args: dict[str, Any]) -> None:
        # Uses run_subagent_structured directly rather than subagent_registry.
        # This is intentional: kanban workers run under subagent_mode=True, but
        # the registry path in agent_loop is gated by `not subagent_mode`, so
        # registry dispatch would never apply. Direct invocation also lets us
        # pass system_override (worker context + completion contract) and
        # policy_context explicitly without coupling to registry internals.
        task: dict[str, Any] = args["task"]
        stop_flag: threading.Event = args["stop_flag"]
        deadline_mono: float | None = args["deadline_mono"]
        task_id: str = task["id"]
        claim_token: str = task["claim_token"]
        assignee: str = task["assignee"]
        attempt: int = int(task.get("last_attempt") or 1)
        handle_id: str = task.get("claim_handle") or ""
        origin_thread_id: str = (task.get("origin_thread_id") or "").strip()
        run_id = ""

        # Bind claim token contextvar so claim-scoped tools can verify.
        token_handle = _set_claim_token(task_id, claim_token)
        try:
            run_id = ks.start_run(
                task_id,
                attempt=attempt,
                assignee=assignee,
                handle_id=handle_id,
                claim_token=claim_token,
            )
        except Exception:
            _log.debug("start_run failed", exc_info=True)
            run_id = ""

        try:
            from utils.kanban_worker_progress import emit_kanban_worker_started

            emit_kanban_worker_started(task_id)
        except Exception:
            _log.debug("kanban worker started activity emit failed", exc_info=True)

        ledger_started_at = time.monotonic()
        strict_cfg = _strict_cfg()
        try:
            from subagent_runner import SUBAGENT_PROFILES, run_subagent_structured
            profile = SUBAGENT_PROFILES.get(assignee, SUBAGENT_PROFILES.get("general", {}))
            base_hint = (profile.get("system_hint") or "").strip()
            retry_brief = build_retry_worker_brief(task_id)
            kanban_block = build_worker_context(task_id)
            try:
                from utils.council_orchestrator import build_council_context_supplement

                council_sup = build_council_context_supplement(task)
                if council_sup:
                    kanban_block = "\n".join(p for p in (kanban_block, council_sup) if p)
            except Exception:
                _log.debug("council context supplement failed", exc_info=True)
            contract = build_completion_contract()

            task_meta = task.get("metadata") or {}
            from utils.kanban_context import resolve_effective_worker_tools

            effective_frozen = resolve_effective_worker_tools(assignee, task_meta)
            if effective_frozen:
                effective_tools: set[str] | None = set(effective_frozen)
                tool_hint = "Available tools for this task: " + ", ".join(sorted(effective_tools))
            else:
                effective_tools = None
                tool_hint = ""
            system_override = "\n".join(p for p in (base_hint, retry_brief, kanban_block, tool_hint, contract) if p)

            try:
                from tool_policy import ToolPolicyContext
                from utils.kanban_context import kanban_task_is_web_act_deliverable

                _web_act_task = kanban_task_is_web_act_deliverable(task_meta)
                _web_act_on = False
                if _web_act_task:
                    try:
                        from tools.web_act import web_act_enabled

                        _web_act_on = web_act_enabled()
                    except Exception:
                        _web_act_on = False
                if _web_act_task and not _web_act_on:
                    err, meta = _typed_failure_payload(
                        FAILURE_MISSING_CAPABILITY,
                        (
                            "deliverable_type web_act requires allow_web_act_scripts pref "
                            "and [web_act].enabled"
                        ),
                    )
                    self._record_outcome_locked(
                        task_id=task_id,
                        claim_token=claim_token,
                        outcome=_resolve_typed_outcome(
                            FAILURE_MISSING_CAPABILITY,
                            legacy_default=ks.OUTCOME_GAVE_UP,
                            cfg=strict_cfg,
                        ),
                        error=err,
                        handle_id=handle_id,
                        run_id=run_id,
                        summary="",
                        metadata=meta,
                    )
                    return
                # Per-task shell opt-in (spec §4): only honored for non-mesh
                # workers, and only when operator-set metadata enables it.
                _allow_shell_task = bool(task_meta.get("allow_shell")) and not str(
                    assignee or ""
                ).startswith("mesh:")
                policy_context = ToolPolicyContext(
                    background_automation=True,
                    subagent_mode=True,
                    scheduled_job=True,
                    kanban_worker=True,
                    kanban_web_act=bool(_web_act_task and _web_act_on),
                    kanban_allow_shell=_allow_shell_task,
                )
            except Exception:
                _log.exception("kanban worker policy context build failed")
                policy_context = ToolPolicyContext(
                    background_automation=True,
                    subagent_mode=True,
                    scheduled_job=True,
                    kanban_worker=True,
                    kanban_web_act=False,
                )

            timeout_seconds = resolve_kanban_worker_timeout_seconds(
                task,
                deadline_mono=deadline_mono,
            )

            task_description = task.get("title") or "kanban task"
            body = task.get("body") or ""
            context_snippet = body if isinstance(body, str) else str(body or "")

            model_override = (task.get("model_override") or "").strip() or None

            if self._worker_outcome_suppressed(task_id):
                return
            if stop_flag.is_set():
                if deadline_mono is not None and time.monotonic() >= deadline_mono:
                    err, meta = _typed_failure_payload(
                        FAILURE_DEADLINE_EXPIRED,
                        "task deadline exceeded before worker spawn",
                    )
                    outcome = _resolve_typed_outcome(
                        FAILURE_DEADLINE_EXPIRED,
                        legacy_default=_legacy_outcome_for_typed_reason(
                            FAILURE_DEADLINE_EXPIRED
                        ),
                        cfg=strict_cfg,
                    )
                elif not _task_was_cancelled(task_id):
                    # Dispatcher shutdown, not an operator cancel: requeue so
                    # the task survives an app restart instead of landing
                    # terminally in the Cancelled column.
                    ks.record_shutdown_requeue(
                        task_id,
                        claim_token=claim_token,
                        error="dispatcher shutdown before spawn; run cancelled",
                    )
                    return
                else:
                    err, meta = _cancelled_failure_payload(
                        "cancelled_before_spawn",
                        "cancelled before spawn",
                    )
                    outcome = ks.OUTCOME_CANCELLED
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=outcome,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                )
                return

            # Deterministic capability preflight before expensive execution.
            required_tools_raw = task_meta.get("required_tools") if isinstance(task_meta, dict) else None
            required_tools = [
                str(x).strip() for x in (required_tools_raw or [])
                if str(x).strip()
            ] if isinstance(required_tools_raw, list) else []
            if required_tools:
                if effective_tools is None:
                    try:
                        from tools import get_tools as _get_tools
                        all_visible_tools = _get_tools(include_media=True, include_cli=False)
                    except Exception:
                        all_visible_tools = []
                visible_tool_names = (
                    set(effective_tools)
                    if effective_tools is not None
                    else {t.name for t in all_visible_tools}
                )
                missing = sorted([name for name in required_tools if name not in visible_tool_names])
                if missing:
                    err, meta = _typed_failure_payload(
                        FAILURE_MISSING_CAPABILITY,
                        "missing required tools: " + ", ".join(missing),
                    )
                    self._record_outcome_locked(
                        task_id=task_id, claim_token=claim_token,
                    outcome=_resolve_typed_outcome(
                        FAILURE_MISSING_CAPABILITY,
                        legacy_default=ks.OUTCOME_GAVE_UP,
                        cfg=strict_cfg,
                    ),
                        error=err,
                        handle_id=handle_id, run_id=run_id,
                        summary="", metadata=meta,
                    )
                    return
                try:
                    from tool_policy import policy_gate_tool_call

                    blocked = [
                        name for name in required_tools
                        if policy_gate_tool_call(name, {}, ctx=policy_context) is not None
                    ]
                except Exception:
                    blocked = []
                if blocked:
                    err, meta = _typed_failure_payload(
                        FAILURE_POLICY_BLOCKED_CAPABILITY,
                        "policy blocked required tools: " + ", ".join(sorted(blocked)),
                    )
                    self._record_outcome_locked(
                        task_id=task_id, claim_token=claim_token,
                    outcome=_resolve_typed_outcome(
                        FAILURE_POLICY_BLOCKED_CAPABILITY,
                        legacy_default=ks.OUTCOME_GAVE_UP,
                        cfg=strict_cfg,
                    ),
                        error=err,
                        handle_id=handle_id, run_id=run_id,
                        summary="", metadata=meta,
                    )
                    return

            result = run_subagent_structured(
                task_description=str(task_description),
                agent_type=assignee,
                context_snippet=context_snippet,
                timeout_seconds=timeout_seconds,
                model_override=model_override,
                policy_context=policy_context,
                parent_thread_id=origin_thread_id or None,
                system_override=system_override,
                tool_names_override=effective_tools,
                kanban_claim=(task_id, claim_token),
                kanban_worker_cwd=resolve_kanban_worker_cwd(task) or None,
                prefer_final_output=True,
                skip_handoff_critic=True,
                external_stop=stop_flag,
            )
            duration_ms = int((time.monotonic() - ledger_started_at) * 1000)

            if self._worker_outcome_suppressed(task_id):
                return
            if stop_flag.is_set() or _task_was_cancelled(task_id):
                if (
                    not _task_was_cancelled(task_id)
                    and deadline_mono is not None
                    and time.monotonic() >= deadline_mono
                ):
                    err, meta = _typed_failure_payload(
                        FAILURE_DEADLINE_EXPIRED,
                        "task deadline exceeded during worker run",
                    )
                    outcome = _resolve_typed_outcome(
                        FAILURE_DEADLINE_EXPIRED,
                        legacy_default=_legacy_outcome_for_typed_reason(
                            FAILURE_DEADLINE_EXPIRED
                        ),
                        cfg=strict_cfg,
                    )
                elif not _task_was_cancelled(task_id):
                    # Dispatcher shutdown mid-run: the worker's partial output
                    # is lost either way, but the TASK is healthy — requeue it
                    # for the next backend instead of terminal cancellation.
                    ks.record_shutdown_requeue(
                        task_id,
                        claim_token=claim_token,
                        error="dispatcher shutdown during worker run; run cancelled",
                    )
                    return
                else:
                    err, meta = _cancelled_failure_payload(
                        "cancelled_during_worker_run",
                        "cancelled during worker run",
                    )
                    outcome = ks.OUTCOME_CANCELLED
                self._record_outcome_locked(
                    task_id=task_id,
                    claim_token=claim_token,
                    outcome=outcome,
                    error=err,
                    handle_id=handle_id,
                    run_id=run_id,
                    summary="",
                    metadata=meta,
                    duration_ms=duration_ms,
                )
                return

            # Reload task to pick up any in-band kanban_post_summary writes
            # the worker may have made via the claim-scoped tool.
            inband_summary = ""
            inband_metadata: dict[str, Any] = {}
            try:
                for r in ks.list_runs(task_id, limit=5):
                    if r.get("outcome") != ks.OUTCOME_ACTIVE:
                        continue
                    summary = (r.get("summary") or "").strip()
                    if not summary:
                        continue
                    inband_summary = summary
                    md = r.get("metadata") or {}
                    if isinstance(md, dict):
                        inband_metadata = md
                    break
            except Exception:
                pass

            if not result.success:
                partial_output = (result.output or "").strip()
                if getattr(result, "timed_out", False):
                    reason = FAILURE_TOOL_TIMEOUT
                    err_detail = result.error or "subagent timed out"
                    if partial_output:
                        err_detail = (
                            f"{err_detail}; partial_output_chars={len(partial_output)}"
                        )
                else:
                    reason = FAILURE_TOOL_RUNTIME_ERROR
                    err_detail = result.error or "subagent failed"
                err, meta = _typed_failure_payload(reason, err_detail)
                if partial_output:
                    meta["tail_excerpt"] = partial_output[-400:]
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=_resolve_typed_outcome(
                        reason,
                        legacy_default=(
                            ks.OUTCOME_TIMED_OUT if reason == FAILURE_TOOL_TIMEOUT else ks.OUTCOME_CRASHED
                        ),
                        cfg=strict_cfg,
                    ),
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                    duration_ms=duration_ms,
                )
                return

            output = (result.output or "").strip()
            parsed_detail = parse_kanban_result_detailed(output)
            parsed = parsed_detail.parsed
            if parsed is None:
                inband_parsed, inband_reason, inband_md = _try_inband_completion(
                    inband_summary=inband_summary,
                    inband_metadata=inband_metadata,
                    task_meta=task_meta if isinstance(task_meta, dict) else {},
                )
                if inband_parsed is not None:
                    parsed = inband_parsed
                elif inband_summary.strip() and inband_reason:
                    err, meta = _typed_failure_payload(
                        inband_reason,
                        "[kanban] in-band summary present but failed verification.",
                    )
                    meta["tail_excerpt"] = output[-400:] if output else "(empty)"
                    self._record_outcome_locked(
                        task_id=task_id, claim_token=claim_token,
                        outcome=_resolve_typed_outcome(
                            inband_reason,
                            legacy_default=_legacy_outcome_for_typed_reason(inband_reason),
                            cfg=strict_cfg,
                        ),
                        error=err,
                        handle_id=handle_id, run_id=run_id,
                        summary="", metadata=meta,
                        duration_ms=duration_ms,
                    )
                    return
            if parsed is None:
                emit_kanban_result_parse_failure_activity(
                    task_id=task_id,
                    raw_excerpt=parsed_detail.excerpt,
                    assignee=assignee,
                    tenant=task.get("tenant") or "default",
                )
                err, meta = _typed_failure_payload(
                    parsed_detail.reason,
                    "[kanban] worker did not return a parseable kanban-result block.",
                )
                meta["tail_excerpt"] = output[-400:] if output else "(empty)"
                task_kind = str((task_meta or {}).get("task_kind") or "").strip().lower()
                if (
                    parsed_detail.reason == FAILURE_CONTRACT_MISSING
                    and task_kind == "research"
                    and len(output) > 80
                ):
                    meta["blockers"] = ["contract_missing_after_research"]
                    meta["next_action"] = (
                        "Summarize findings already gathered in this run via save_research_findings, "
                        "then end with a trailing kanban-result fenced JSON block (summary + metadata)."
                    )
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=_resolve_typed_outcome(
                        parsed_detail.reason,
                        legacy_default=_legacy_outcome_for_typed_reason(parsed_detail.reason),
                        cfg=strict_cfg,
                    ),
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                    duration_ms=duration_ms,
                )
                return
            summary = parsed.get("summary") or inband_summary or ""
            effective_parsed, metadata = merge_kanban_handoff(
                parsed,
                inband_summary=inband_summary,
                inband_metadata=inband_metadata,
            )
            effective_parsed = normalize_salvage_handoff(
                effective_parsed,
                worker_transcript=output,
                task_metadata=_verify_task_metadata(task),
            )
            metadata = (
                effective_parsed.get("metadata")
                if isinstance(effective_parsed.get("metadata"), dict)
                else metadata
            )
            metadata = normalize_kanban_result_metadata(
                metadata,
                task_metadata=_verify_task_metadata(task),
            )
            effective_parsed = dict(effective_parsed)
            effective_parsed["metadata"] = metadata
            summary = effective_parsed.get("summary") or summary

            task_meta = _verify_task_metadata(task)
            verifier_ok, verifier_reason, verifier_detail = verify_kanban_result(
                effective_parsed,
                task_metadata=task_meta,
                worker_transcript=output,
            )
            if verifier_ok:
                try:
                    from utils.council_orchestrator import apply_council_seat_verify

                    verifier_ok, verifier_reason, verifier_detail = apply_council_seat_verify(
                        verifier_ok,
                        verifier_reason,
                        verifier_detail,
                        effective_parsed,
                        task=task,
                        task_metadata=task_meta,
                    )
                except Exception:
                    _log.debug("council seat extended verify failed", exc_info=True)
                    if task_meta.get("origin") == "council":
                        verifier_ok = False
                        verifier_reason = FAILURE_VERIFIER_FAILED
                        verifier_detail = "council validation error"
            if not verifier_ok:
                salvage = salvage_handoff_action(metadata)
                if salvage in ("block", "retry"):
                    self._record_verified_handoff_locked(
                        task_id=task_id,
                        claim_token=claim_token,
                        handle_id=handle_id,
                        run_id=run_id,
                        summary=summary,
                        metadata=metadata,
                        duration_ms=duration_ms,
                    )
                    return
                should_fail, metadata = _resolve_kanban_verifier_failure(
                    verifier_reason=verifier_reason,
                    verifier_detail=verifier_detail,
                    metadata=metadata,
                    strict_cfg=strict_cfg,
                    kanban_worker=True,
                )
                if should_fail:
                    err, fail_meta = _typed_failure_payload(verifier_reason, verifier_detail)
                    fail_meta["parsed_metadata"] = metadata
                    if task_meta.get("origin") == "council":
                        fail_meta["council_validation_error"] = verifier_detail
                    self._record_outcome_locked(
                        task_id=task_id, claim_token=claim_token,
                        outcome=_resolve_typed_outcome(
                            verifier_reason,
                            legacy_default=_legacy_outcome_for_typed_reason(verifier_reason),
                            cfg=strict_cfg,
                        ),
                        error=err,
                        handle_id=handle_id, run_id=run_id,
                        summary="", metadata=fail_meta,
                        duration_ms=duration_ms,
                    )
                    return

            self._record_verified_handoff_locked(
                task_id=task_id,
                claim_token=claim_token,
                handle_id=handle_id,
                run_id=run_id,
                summary=summary,
                metadata=metadata,
                duration_ms=duration_ms,
            )
        except Exception as exc:
            _log.debug("kanban worker raised", exc_info=True)
            if not self._worker_outcome_suppressed(task_id):
                try:
                    err, meta = _typed_failure_payload(
                        FAILURE_TOOL_RUNTIME_ERROR,
                        f"{type(exc).__name__}: {exc}",
                    )
                    self._record_outcome_locked(
                        task_id=task_id, claim_token=claim_token,
                        outcome=ks.OUTCOME_CRASHED,
                        error=err,
                        handle_id=handle_id, run_id=run_id,
                        summary="", metadata=meta,
                    )
                except Exception:
                    _log.debug("record_failure on worker exception failed", exc_info=True)
        finally:
            try:
                from utils.kanban_worker_progress import (
                    emit_kanban_worker_completed,
                    flush_pending_sse,
                )

                flush_pending_sse(task_id)
                emit_kanban_worker_completed(task_id)
            except Exception:
                _log.debug("kanban worker activity cleanup failed", exc_info=True)
            _reset_claim_token(token_handle)
            self._clear_deadline_force_release(task_id)
            with self._lock:
                self._inflight.pop(task_id, None)

    # ------------------------------ mesh worker -------------------------------

    def _run_mesh_worker_sync(self, args: dict[str, Any]) -> None:
        """Plan B consultative mesh delegation.

        Home retains the ``claim_token`` and the run row; the peer runs the
        work as a one-shot inbound subagent under the existing ``mesh_sourced``
        policy. We ship the task body + worker context + completion contract
        as a ``task_delegate`` message and parse the peer's ``task_result.text``
        identically to a local worker's stdout. The claim never crosses the
        wire, and there is no remote completion verb.
        """
        task: dict[str, Any] = args["task"]
        stop_flag: threading.Event = args["stop_flag"]
        deadline_mono: float | None = args["deadline_mono"]
        task_id: str = task["id"]
        claim_token: str = task["claim_token"]
        assignee: str = task["assignee"]
        attempt: int = int(task.get("last_attempt") or 1)
        handle_id: str = task.get("claim_handle") or ""
        peer_fp = assignee[len("mesh:"):].strip()
        run_id = ""

        token_handle = _set_claim_token(task_id, claim_token)
        ledger_started_at = time.monotonic()
        strict_cfg = _strict_cfg()
        try:
            try:
                run_id = ks.start_run(
                    task_id,
                    attempt=attempt,
                    assignee=assignee,
                    handle_id=handle_id,
                    claim_token=claim_token,
                )
            except Exception:
                _log.debug("start_run (mesh) failed", exc_info=True)
                run_id = ""

            try:
                from utils.kanban_worker_progress import emit_kanban_worker_started

                emit_kanban_worker_started(task_id)
            except Exception:
                _log.debug("kanban mesh worker started activity emit failed", exc_info=True)

            if not peer_fp:
                err, meta = _typed_failure_payload(
                    FAILURE_MESH_PEER_ERROR,
                    "mesh assignee missing peer fingerprint",
                )
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=ks.OUTCOME_GAVE_UP,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                )
                return

            try:
                from mesh import get_mesh_manager
                manager = get_mesh_manager()
            except Exception:
                manager = None
            if manager is None:
                err, meta = _typed_failure_payload(
                    FAILURE_MESH_PEER_ERROR,
                    "mesh manager not initialized",
                )
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=ks.OUTCOME_GAVE_UP,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                )
                return

            session_id = ""
            try:
                for sess in manager.list_sessions():
                    if (sess.get("peer_fingerprint") or "") == peer_fp:
                        session_id = sess.get("session_id") or ""
                        break
            except Exception:
                _log.debug("list_sessions raised in mesh worker", exc_info=True)
            if not session_id:
                err, meta = _typed_failure_payload(
                    FAILURE_MESH_PEER_ERROR,
                    f"mesh peer {peer_fp} not reachable",
                )
                meta["mesh_peer"] = peer_fp
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=ks.OUTCOME_GAVE_UP,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                )
                return
            try:
                trusted = bool(manager.peers.trusted_for_tasks(peer_fp))
            except Exception:
                trusted = False
            if not trusted:
                err, meta = _typed_failure_payload(
                    FAILURE_POLICY_BLOCKED_CAPABILITY,
                    "mesh peer not trusted_for_tasks",
                )
                meta["mesh_peer"] = peer_fp
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=ks.OUTCOME_GAVE_UP,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                )
                return

            base_hint = "Mesh task delegation: run as a one-shot subagent."
            retry_brief = build_retry_worker_brief(task_id)
            kanban_block = build_worker_context(task_id)
            contract = build_completion_contract()
            body_text = task.get("body") or ""
            outbound = "\n".join(
                p for p in (base_hint, retry_brief, kanban_block, contract, body_text) if p
            )

            timeout_seconds = 600
            if deadline_mono is not None:
                timeout_seconds = max(15, int(deadline_mono - time.monotonic()))

            conversation_id = f"kanban:{task_id}"
            reg = manager.register_pending_result(
                conversation_id=conversation_id,
                session_id=session_id,
                peer_fingerprint=peer_fp,
                timeout_seconds=timeout_seconds,
            )
            if not isinstance(reg, dict) or not reg.get("ok"):
                reg_err = (reg or {}).get("error") if isinstance(reg, dict) else "register_pending_result failed"
                err, meta = _typed_failure_payload(
                    FAILURE_MESH_PEER_ERROR,
                    f"mesh register_pending_result: {reg_err}",
                )
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=ks.OUTCOME_CRASHED,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                )
                return
            pending_key = str(reg.get("key") or "")

            try:
                from mesh.protocol import MeshMessage
            except Exception as exc:
                manager.unregister_pending_result(pending_key)
                err, meta = _typed_failure_payload(
                    FAILURE_MESH_PEER_ERROR,
                    f"mesh protocol import failed: {exc}",
                )
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=ks.OUTCOME_CRASHED,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                )
                return

            send_payload = {
                "text": outbound,
                "job_id": task_id,
                "kind": "kanban",
                "max_response_chars": 16_000,
            }
            send_msg = MeshMessage(
                type="task_delegate",
                conversation_id=conversation_id,
                conversation_type="task",
                payload=send_payload,
            )
            try:
                send_out = manager.send_message(
                    session_id=session_id,
                    message=send_msg,
                    source="kanban_dispatcher",
                )
            except Exception as exc:
                manager.unregister_pending_result(pending_key)
                err, meta = _typed_failure_payload(
                    FAILURE_MESH_PEER_ERROR,
                    f"mesh send raised: {exc}",
                )
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=ks.OUTCOME_CRASHED,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                )
                return
            if not isinstance(send_out, dict) or not send_out.get("ok"):
                send_err = (send_out or {}).get("error") if isinstance(send_out, dict) else "send failed"
                manager.unregister_pending_result(pending_key)
                err, meta = _typed_failure_payload(
                    FAILURE_MESH_PEER_ERROR,
                    f"mesh send: {send_err}",
                )
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=ks.OUTCOME_CRASHED,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                )
                return

            wait_out = manager.wait_for_pending_result(
                key=pending_key,
                timeout_seconds=timeout_seconds,
                stop_check=lambda: stop_flag.is_set(),
            )
            duration_ms = int((time.monotonic() - ledger_started_at) * 1000)

            if self._worker_outcome_suppressed(task_id):
                return

            if not isinstance(wait_out, dict):
                wait_out = {"ok": False, "error": "invalid wait result"}
            if wait_out.get("error") == "timeout":
                err, meta = _typed_failure_payload(
                    FAILURE_TOOL_TIMEOUT,
                    "mesh peer reply deadline exceeded",
                )
                meta["mesh_peer"] = peer_fp
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=ks.OUTCOME_TIMED_OUT,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                    duration_ms=duration_ms,
                )
                return
            if wait_out.get("error") == "cancelled":
                if deadline_mono is not None and time.monotonic() >= deadline_mono:
                    err, meta = _typed_failure_payload(
                        FAILURE_DEADLINE_EXPIRED,
                        "task deadline exceeded before mesh peer responded",
                    )
                    meta["mesh_peer"] = peer_fp
                    outcome = _resolve_typed_outcome(
                        FAILURE_DEADLINE_EXPIRED,
                        legacy_default=_legacy_outcome_for_typed_reason(
                            FAILURE_DEADLINE_EXPIRED
                        ),
                        cfg=strict_cfg,
                    )
                elif not _task_was_cancelled(task_id):
                    # Dispatcher shutdown while waiting on a mesh peer:
                    # requeue rather than terminally cancel.
                    ks.record_shutdown_requeue(
                        task_id,
                        claim_token=claim_token,
                        error="dispatcher shutdown before mesh peer responded; run cancelled",
                    )
                    return
                else:
                    err, meta = _cancelled_failure_payload(
                        "cancelled_before_mesh_response",
                        "cancelled before mesh peer responded",
                        extra={"mesh_peer": peer_fp},
                    )
                    outcome = ks.OUTCOME_CANCELLED
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=outcome,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                    duration_ms=duration_ms,
                )
                return
            if not wait_out.get("ok"):
                peer_err = str(wait_out.get("error") or "mesh peer error")
                err, meta = _typed_failure_payload(
                    FAILURE_MESH_PEER_ERROR,
                    f"mesh peer: {peer_err}",
                )
                meta["mesh_peer"] = peer_fp
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=ks.OUTCOME_CRASHED,
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                    duration_ms=duration_ms,
                )
                return

            reply_text = str(wait_out.get("text") or "")
            parsed_detail = parse_kanban_result_detailed(reply_text)
            parsed = parsed_detail.parsed
            tenant = task.get("tenant") or "default"
            if parsed is None:
                emit_kanban_result_parse_failure_activity(
                    task_id=task_id,
                    raw_excerpt=parsed_detail.excerpt,
                    assignee=assignee,
                    tenant=tenant,
                )
                err, meta = _typed_failure_payload(
                    parsed_detail.reason,
                    "[kanban] mesh peer did not return a parseable kanban-result block.",
                )
                meta["mesh_peer"] = peer_fp
                meta["tail_excerpt"] = reply_text[-400:] if reply_text else "(empty)"
                self._record_outcome_locked(
                    task_id=task_id, claim_token=claim_token,
                    outcome=_resolve_typed_outcome(
                        parsed_detail.reason,
                        legacy_default=_legacy_outcome_for_typed_reason(parsed_detail.reason),
                        cfg=strict_cfg,
                    ),
                    error=err,
                    handle_id=handle_id, run_id=run_id,
                    summary="", metadata=meta,
                    duration_ms=duration_ms,
                )
                return
            else:
                effective_parsed, metadata = merge_kanban_handoff(parsed)
                effective_parsed = normalize_salvage_handoff(
                    effective_parsed,
                    worker_transcript=reply_text,
                    task_metadata=_verify_task_metadata(task),
                )
                metadata = (
                    effective_parsed.get("metadata")
                    if isinstance(effective_parsed.get("metadata"), dict)
                    else metadata
                )
                summary = effective_parsed.get("summary") or ""
                metadata = normalize_kanban_result_metadata(
                    metadata,
                    task_metadata=_verify_task_metadata(task),
                )
                metadata = {**metadata, "mesh_peer": peer_fp}
                effective_parsed = dict(effective_parsed)
                effective_parsed["metadata"] = metadata

            task_meta = _verify_task_metadata(task)
            verifier_ok, verifier_reason, verifier_detail = verify_kanban_result(
                effective_parsed,
                task_metadata=task_meta,
                worker_transcript=reply_text,
            )
            try:
                from utils.council_orchestrator import apply_council_seat_verify

                verifier_ok, verifier_reason, verifier_detail = apply_council_seat_verify(
                    verifier_ok,
                    verifier_reason,
                    verifier_detail,
                    effective_parsed,
                    task=task,
                    task_metadata=task_meta,
                )
            except Exception:
                _log.debug("council mesh seat verify failed", exc_info=True)
                if task_meta.get("origin") == "council":
                    verifier_ok = False
                    verifier_reason = FAILURE_VERIFIER_FAILED
                    verifier_detail = "council validation error"
            if not verifier_ok:
                salvage = salvage_handoff_action(metadata)
                if salvage in ("block", "retry"):
                    self._record_verified_handoff_locked(
                        task_id=task_id,
                        claim_token=claim_token,
                        handle_id=handle_id,
                        run_id=run_id,
                        summary=summary,
                        metadata=metadata,
                        duration_ms=duration_ms,
                    )
                    return
                should_fail, metadata = _resolve_kanban_verifier_failure(
                    verifier_reason=verifier_reason,
                    verifier_detail=verifier_detail,
                    metadata=metadata,
                    strict_cfg=strict_cfg,
                    kanban_worker=True,
                )
                if should_fail:
                    err, meta = _typed_failure_payload(verifier_reason, verifier_detail)
                    meta["mesh_peer"] = peer_fp
                    meta["parsed_metadata"] = metadata
                    if task_meta.get("origin") == "council":
                        meta["council_validation_error"] = verifier_detail
                    self._record_outcome_locked(
                        task_id=task_id, claim_token=claim_token,
                        outcome=_resolve_typed_outcome(
                            verifier_reason,
                            legacy_default=_legacy_outcome_for_typed_reason(verifier_reason),
                            cfg=strict_cfg,
                        ),
                        error=err,
                        handle_id=handle_id, run_id=run_id,
                        summary="", metadata=meta,
                        duration_ms=duration_ms,
                    )
                    return

            self._record_verified_handoff_locked(
                task_id=task_id,
                claim_token=claim_token,
                handle_id=handle_id,
                run_id=run_id,
                summary=summary,
                metadata=metadata,
                duration_ms=duration_ms,
            )
        except Exception as exc:
            _log.debug("kanban mesh worker raised", exc_info=True)
            if not self._worker_outcome_suppressed(task_id):
                try:
                    err, meta = _typed_failure_payload(
                        FAILURE_TOOL_RUNTIME_ERROR,
                        f"{type(exc).__name__}: {exc}",
                    )
                    meta["mesh_peer"] = peer_fp
                    self._record_outcome_locked(
                        task_id=task_id, claim_token=claim_token,
                        outcome=ks.OUTCOME_CRASHED,
                        error=err,
                        handle_id=handle_id, run_id=run_id,
                        summary="", metadata=meta,
                    )
                except Exception:
                    _log.debug("record_failure on mesh worker exception failed", exc_info=True)
        finally:
            try:
                from utils.kanban_worker_progress import (
                    emit_kanban_worker_completed,
                    flush_pending_sse,
                )

                flush_pending_sse(task_id)
                emit_kanban_worker_completed(task_id)
            except Exception:
                _log.debug("kanban mesh worker activity cleanup failed", exc_info=True)
            _reset_claim_token(token_handle)
            self._clear_deadline_force_release(task_id)
            with self._lock:
                self._inflight.pop(task_id, None)

    def _record_verified_handoff_locked(
        self,
        *,
        task_id: str,
        claim_token: str,
        handle_id: str,
        run_id: str,
        summary: str,
        metadata: dict[str, Any],
        duration_ms: int | None = None,
    ) -> None:
        """Route verifier-success handoffs — salvage statuses must not mark ``done``."""
        # Decide routing from the full metadata first, then shrink for
        # persistence. Every branch below persists this metadata: the
        # success/retry branches shrink via _record_outcome_locked, but the
        # ``block`` branch calls ks.record_salvage_blocked directly, which
        # validates (and raises) on the byte budget. An oversized blocked-salvage
        # handoff therefore threw, was swallowed as a WARNING, and pinned the
        # task in_progress with a live claim forever (incident kt_6cc4cd8e862947:
        # research over a 213K-char MEMORY.md). Shrinking after routing keeps the
        # decision intact while guaranteeing all routes persist bounded metadata.
        salvage = salvage_handoff_action(metadata)
        try:
            shrunk = ks.shrink_metadata_to_limit(metadata or {})
            if shrunk.get("metadata_truncated") and not (metadata or {}).get("metadata_truncated"):
                _log.warning(
                    "verified handoff metadata for %s exceeded the byte budget; truncated",
                    task_id,
                )
            metadata = shrunk
        except Exception:
            _log.debug("verified handoff metadata shrink failed; continuing", exc_info=True)
        if salvage == "block":
            try:
                ks.record_salvage_blocked(
                    task_id,
                    claim_token=claim_token,
                    summary=summary,
                    metadata=metadata,
                    handle_id=handle_id,
                )
            except Exception:
                _log.warning("record salvage blocked for %s failed", task_id, exc_info=True)
            else:
                try:
                    from utils.kanban_coordinator import maybe_auto_repair_blocked_task

                    maybe_auto_repair_blocked_task(task_id)
                except Exception:
                    _log.debug("auto-repair after blocked salvage failed", exc_info=True)
                try:
                    from utils.council_delivery import maybe_handle_council_task_terminal

                    maybe_handle_council_task_terminal(task_id)
                except Exception:
                    _log.debug("council terminal hook after blocked salvage failed", exc_info=True)
            return
        if salvage == "retry":
            fail_meta = dict(metadata)
            fail_meta["salvage_status"] = "partial"
            if summary.strip():
                fail_meta["salvage_summary"] = summary.strip()
            self._record_outcome_locked(
                task_id=task_id,
                claim_token=claim_token,
                outcome=ks.OUTCOME_CRASHED,
                error=(summary or "partial salvage handoff")[:1000],
                handle_id=handle_id,
                run_id=run_id,
                summary="",
                metadata=fail_meta,
                duration_ms=duration_ms,
            )
            return
        self._record_outcome_locked(
            task_id=task_id,
            claim_token=claim_token,
            outcome=ks.OUTCOME_COMPLETED,
            error="",
            handle_id=handle_id,
            run_id=run_id,
            summary=summary,
            metadata=metadata,
            duration_ms=duration_ms,
        )

    def _record_outcome_locked(
        self,
        *,
        task_id: str,
        claim_token: str,
        outcome: str,
        error: str,
        handle_id: str,
        run_id: str,
        summary: str,
        metadata: dict[str, Any],
        duration_ms: int | None = None,
    ) -> None:
        """Record completion or failure plus the unified ledger row."""
        import time

        from utils.kanban_context import classify_failure_class

        # The outcome write is dispatcher-owned and must be durable: a metadata
        # validation rejection here loses the terminal state and pins the task
        # in_progress until claim expiry — then the retry repeats the loss
        # (incident kt_1de7d1f3d14b4e). Shrink to the storage budget up front.
        try:
            shrunk = ks.shrink_metadata_to_limit(metadata or {})
            if shrunk.get("metadata_truncated") and not (metadata or {}).get("metadata_truncated"):
                _log.warning(
                    "record outcome metadata for %s exceeded the byte budget; truncated",
                    task_id,
                )
            metadata = shrunk
        except Exception:
            _log.debug("metadata shrink failed; continuing with original", exc_info=True)

        last_exc: Exception | None = None
        for attempt in range(3):
            try:
                if outcome == ks.OUTCOME_COMPLETED:
                    ks.record_completion(
                        task_id,
                        claim_token=claim_token,
                        summary=summary,
                        metadata=metadata,
                        handle_id=handle_id,
                    )
                else:
                    ks.record_failure(
                        task_id,
                        claim_token=claim_token,
                        outcome=outcome,
                        error=error,
                        handle_id=handle_id,
                        metadata=metadata,
                    )
                last_exc = None
                break
            except Exception as exc:
                if (
                    outcome == ks.OUTCOME_CANCELLED
                    and isinstance(exc, ValueError)
                    and "not in_progress" in str(exc)
                ):
                    # Benign shutdown race: mark_inflight_cancelled already
                    # recorded the run as cancelled and reset the task to
                    # ready while this worker thread was still unwinding. The
                    # terminal state is durable — re-recording would fail
                    # forever and the old fallback spammed scary warnings +
                    # a spurious parse-failure activity event.
                    _log.debug(
                        "cancelled outcome for %s already recorded by shutdown reset",
                        task_id,
                    )
                    last_exc = None
                    break
                last_exc = exc
                if attempt < 2:
                    time.sleep(0.05 * (2 ** attempt))
        if last_exc is not None:
            _log.warning(
                "record outcome %s for %s failed after retries",
                outcome,
                task_id,
                exc_info=last_exc,
            )
            try:
                emit_kanban_result_parse_failure_activity(
                    task_id=task_id,
                    raw_excerpt=f"record_{outcome}_failed",
                    assignee="",
                    tenant="default",
                )
            except Exception:
                _log.debug("failed to emit record outcome failure activity", exc_info=True)
            try:
                # Last resort: minimal metadata only. Re-using the full blob
                # here re-triggers the very validation failure being handled
                # (the incident's fallback failed for the same reason).
                ks.record_failure(
                    task_id,
                    claim_token=claim_token,
                    outcome=ks.OUTCOME_CRASHED,
                    error=f"persistence failed after verified handoff: {last_exc}"[:1000],
                    handle_id=handle_id,
                    metadata={
                        "failure_reason": "record_outcome_persistence_failed",
                        "failure_class": classify_failure_class(
                            "record_outcome_persistence_failed"
                        ),
                        "intended_outcome": outcome,
                    },
                )
            except Exception:
                _log.warning(
                    "fallback record_failure after persistence failure also failed for %s",
                    task_id,
                    exc_info=True,
                )
            return
        # Unified ledger row (redacted summary/error per redaction policy).
        try:
            from utils.task_ledger import record_run as _ledger_record
            from utils.redaction import cap_str
            status = "ok" if outcome == ks.OUTCOME_COMPLETED else (
                "timeout" if outcome == ks.OUTCOME_TIMED_OUT else "error"
            )
            _ledger_record(
                kind="subagent",
                task_name=f"kanban:{task_id}",
                thread_id=None,
                status=status,
                duration_ms=duration_ms,
                error_summary=cap_str(error, 240) if error else None,
                meta={
                    "kanban_task_id": task_id,
                    "kanban_run_id": run_id,
                    "kanban_outcome": outcome,
                    "failure_reason": (metadata or {}).get("failure_reason"),
                    "failure_class": (metadata or {}).get("failure_class"),
                    "next_action": (metadata or {}).get("next_action"),
                    "summary_excerpt": cap_str(summary, 240) if summary else None,
                },
            )
        except Exception:
            _log.debug("task_ledger.record_run for kanban run failed", exc_info=True)
        try:
            from utils.council_delivery import maybe_handle_council_task_terminal

            maybe_handle_council_task_terminal(task_id)
        except Exception:
            _log.debug("council terminal hook failed", exc_info=True)

    # ------------------------------ health ------------------------------------

    def health(self) -> dict[str, Any]:
        try:
            base = ks.health_summary()
        except Exception:
            base = {}
        if self._board_needs_dispatcher(base):
            self.ensure_running()
        with self._lock:
            inflight_count = len(self._inflight)
            in_flight_tasks = list(self._inflight.keys())
        base["dispatcher"] = {
            "running": self._loop_alive(),
            "tick_count": self._tick_count,
            "last_tick_at": self._last_tick_at,
            "inflight_workers": inflight_count,
            "inflight_task_ids": in_flight_tasks,
        }
        return base


# ---------------------------------------------------------------------------
# Module-level singleton
# ---------------------------------------------------------------------------

def _task_was_cancelled(task_id: str) -> bool:
    tid = str(task_id or "").strip()
    if not tid:
        return False
    try:
        row = ks.get_task(tid) or {}
        return (row.get("status") or "").strip() == ks.STATUS_CANCELLED
    except Exception:
        _log.debug("cancelled-task probe failed for %s", tid, exc_info=True)
        return False


_singleton_lock = threading.Lock()
_singleton: _Dispatcher | None = None


def _get_dispatcher() -> _Dispatcher:
    global _singleton
    with _singleton_lock:
        if _singleton is None:
            _singleton = _Dispatcher()
        return _singleton


def start_dispatcher() -> bool:
    """Start the kanban dispatcher daemon. Idempotent."""
    return _get_dispatcher().start()


def stop_dispatcher(*, join_seconds: float = 5.0) -> None:
    """Stop the kanban dispatcher daemon. Idempotent."""
    if _singleton is None:
        return
    _get_dispatcher().stop(join_seconds=join_seconds)


def ensure_dispatcher_running() -> bool:
    """Ensure the kanban dispatcher loop thread is alive (restart if needed)."""
    return _get_dispatcher().ensure_running()


def nudge_dispatcher() -> None:
    """Wake the kanban dispatcher loop early. Restarts the loop if it died."""
    _get_dispatcher().nudge()


def dispatcher_health() -> dict[str, Any]:
    """Combined storage + dispatcher health snapshot.

    Auto-restarts the dispatcher when the loop is down but the board still
    has ready/in_progress/todo work.
    """
    return _get_dispatcher().health()


def tick_once() -> dict[str, int]:
    """Single-shot tick (used by tests)."""
    return _get_dispatcher().tick_once()


def request_worker_stop(task_id: str, *, join_seconds: float = 2.0) -> bool:
    """Cooperatively stop an in-flight kanban worker before cancel/reassign."""
    tid = str(task_id or "").strip()
    if not tid:
        return False
    disp = _get_dispatcher()
    with disp._lock:
        worker = disp._inflight.get(tid)
    if worker is None:
        return False
    worker.stop_flag.set()
    try:
        worker.thread.join(timeout=max(0.0, float(join_seconds)))
    except Exception:
        _log.debug("request_worker_stop join failed for %s", tid, exc_info=True)
    return True
