"""Kanban task board storage — durable, encrypted, profile-shared.

Lives next to :mod:`chat_storage` and reuses its connection / Fernet stack.
Tables: ``kanban_tasks`` (one row per task) and ``kanban_task_runs`` (one row
per attempt; outcome / summary / metadata / error). Schema is created in
``chat_storage._get_connection()``.

Public surface (module-level functions):

- ``create_task`` / ``create_batch`` — DAG-aware insert with cycle detection,
  idempotency, kill-switch + backpressure gates.
- ``get_task`` / ``list_tasks`` / ``list_children`` — read paths (decrypted).
- ``claim_task`` — atomic claim using ``UPDATE ... RETURNING *`` (SQLite
  >= 3.35) with a select-then-update fallback for older runtimes. Both
  paths bump ``last_attempt`` in the same transaction.
- ``release_claim`` / ``record_completion`` / ``record_failure`` — terminal
  paths that also write the corresponding ``kanban_task_runs`` row, promote
  ready children, and emit (ciphertext-free) activity events.
- ``block_task`` / ``unblock_task`` / ``cancel_task`` — operator paths.
- ``set_assignee`` — operator-only patch; requires ``force=True`` to seize an
  in-flight task.
- ``start_run`` / ``close_run`` / ``list_runs`` — low-level run helpers used
  by the dispatcher when the spawn path is split from the claim.
- ``find_stale_claims`` / ``find_overdue_tasks`` / ``mark_inflight_cancelled``
  — dispatcher and shutdown plumbing.
- ``delete_task`` / ``clear_terminal_tasks`` — operator-triggered cleanup.
- ``purge_terminal`` / ``trim_runs`` — retention helpers.
- ``health_summary`` — counts for ``GET /kanban/dispatcher/health``.

Encryption: title, body, tags, metadata, summary, error are Fernet
ciphertext (Keychain-backed key). Activity events emitted from this module
contain only ids and transitions — never decrypted content.

See ``.cursor/plans/mugi_kanban_v1_backend_29b4b5cf.plan.md`` for the full
design and threat model anchors.
"""

from __future__ import annotations

import json
import logging
import secrets
import sqlite3
import threading
import time
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any, Iterable

from chat_storage import _get_connection, _encrypt, _decrypt, ensure_encryption_ready
from utils.kanban_activity import (
    emit_kanban_event_activity,
    try_acquire_kanban_create_slot,
)
from utils.kanban_events import notify_kanban_event

_log = logging.getLogger(__name__)


def _bump_search_generation() -> None:
    """Lazy import to avoid circular dependency with kanban_search."""
    try:
        import kanban_search
        kanban_search.bump_generation()
    except Exception:
        pass

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

SCHEMA_VERSION = 2  # Plan B: kanban_task_parents join table; v1 rows backfilled.

STATUS_TODO = "todo"
STATUS_READY = "ready"
STATUS_IN_PROGRESS = "in_progress"
STATUS_BLOCKED = "blocked"
STATUS_DONE = "done"
STATUS_CANCELLED = "cancelled"
ALL_STATUSES = frozenset({
    STATUS_TODO, STATUS_READY, STATUS_IN_PROGRESS,
    STATUS_BLOCKED, STATUS_DONE, STATUS_CANCELLED,
})
TERMINAL_STATUSES = frozenset({STATUS_DONE, STATUS_CANCELLED})
ACTIVE_STATUSES = frozenset({
    STATUS_TODO, STATUS_READY, STATUS_IN_PROGRESS, STATUS_BLOCKED,
})

OUTCOME_ACTIVE = "active"
OUTCOME_COMPLETED = "completed"
OUTCOME_BLOCKED = "blocked"
OUTCOME_SPAWN_FAILED = "spawn_failed"
OUTCOME_CRASHED = "crashed"
OUTCOME_TIMED_OUT = "timed_out"
OUTCOME_GAVE_UP = "gave_up"
OUTCOME_CANCELLED = "cancelled"
TERMINAL_OUTCOMES = frozenset({
    OUTCOME_COMPLETED, OUTCOME_BLOCKED, OUTCOME_SPAWN_FAILED,
    OUTCOME_CRASHED, OUTCOME_TIMED_OUT, OUTCOME_GAVE_UP, OUTCOME_CANCELLED,
})

# Field caps. These are deliberately tight to bound prompt-injection blast
# radius and DB size — task body lives in the worker's system prompt.
MAX_TITLE_CHARS = 240
MAX_BODY_CHARS = 16_000
MAX_TAGS = 16
MAX_TAG_CHARS = 64
MAX_METADATA_BYTES = 8 * 1024
MAX_SUMMARY_CHARS = 8_000
MAX_ERROR_CHARS = 2_000
MAX_TENANT_CHARS = 64
MAX_ASSIGNEE_CHARS = 64
MAX_THREAD_ID_CHARS = 128
MAX_IDEMPOTENCY_CHARS = 128
MAX_PROGRESS_NOTE_CHARS = 256
MAX_PROGRESS_ARGS_CHARS = 256
MAX_PROGRESS_RESULT_CHARS = 512
MAX_PROGRESS_THUMB_B64_CHARS = 8192
MAX_PROGRESS_CALL_ID_CHARS = 128
MAX_PROGRESS_LOG_ENTRIES = 50
PARENT_CHAIN_HOP_CAP = 64
PARENT_DAG_EDGE_CAP = 256
MAX_PARENTS_PER_TASK = 16  # bounds the initial parent set passed to create_task

# Operator-only tags stripped from agent-tier creation (see plan §security).
# Plan B introduces [kanban].mesh_eligible_agent_creation = false to optionally
# allow the agent to set this; until then the strip is the floor.
OPERATOR_ONLY_TAGS = frozenset({"mesh-eligible"})


def _now_iso() -> str:
    return datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _new_id(prefix: str) -> str:
    return f"{prefix}_{uuid.uuid4().hex[:14]}"


def _new_token() -> str:
    return secrets.token_hex(16)


def _supports_returning() -> bool:
    """SQLite >= 3.35 supports UPDATE ... RETURNING. Defaults true; falls
    back transparently when it doesn't."""
    return sqlite3.sqlite_version_info >= (3, 35)


# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------

def _validate_text(name: str, value: Any, *, max_chars: int, allow_empty: bool = False) -> str:
    if value is None:
        if allow_empty:
            return ""
        raise ValueError(f"{name} is required")
    if not isinstance(value, str):
        value = str(value)
    s = value.strip()
    if not s and not allow_empty:
        raise ValueError(f"{name} must be non-empty")
    if len(s) > max_chars:
        raise ValueError(f"{name} exceeds {max_chars}-char limit")
    return s


def _validate_assignee(value: Any) -> str:
    s = _validate_text("assignee", value, max_chars=MAX_ASSIGNEE_CHARS)
    # Mesh peer assignees use 'mesh:<fp>' shape (Plan B); accept syntactically.
    return s


def _validate_priority(value: Any) -> int:
    try:
        p = int(value)
    except (TypeError, ValueError):
        raise ValueError("priority must be an integer in [1, 5]")
    if p < 1 or p > 5:
        raise ValueError("priority must be in [1, 5]")
    return p


def _validate_tags(value: Any) -> list[str]:
    if value in (None, ""):
        return []
    if isinstance(value, str):
        try:
            value = json.loads(value)
        except Exception as exc:
            raise ValueError(f"tags is not valid JSON: {exc}")
    if not isinstance(value, list):
        raise ValueError("tags must be a JSON array of strings")
    out: list[str] = []
    for t in value:
        if not isinstance(t, str):
            raise ValueError("tags must be strings")
        s = t.strip()
        if not s:
            continue
        if len(s) > MAX_TAG_CHARS:
            raise ValueError(f"tag exceeds {MAX_TAG_CHARS}-char limit")
        out.append(s)
        if len(out) > MAX_TAGS:
            raise ValueError(f"more than {MAX_TAGS} tags")
    return out


def _validate_tool_names_in_metadata(metadata: dict) -> dict:
    """Sanitize ``metadata.tool_names`` if present.

    Valid shape: list of up to 30 non-empty alphanumeric+underscore strings,
    each at most 64 chars.  Invalid entries are silently stripped; if nothing
    survives, the key is removed entirely (falls back to profile default).
    """
    raw = metadata.get("tool_names")
    if raw is None:
        return metadata
    if not isinstance(raw, list):
        return {k: v for k, v in metadata.items() if k != "tool_names"}
    clean = [
        s for s in (str(x).strip() for x in raw)
        if s and len(s) <= 64 and s.replace("_", "").isalnum()
    ][:30]
    if clean:
        return dict(metadata, tool_names=clean)
    return {k: v for k, v in metadata.items() if k != "tool_names"}


def _validate_metadata(value: Any) -> dict[str, Any]:
    if value in (None, ""):
        return {}
    if isinstance(value, str):
        try:
            value = json.loads(value)
        except Exception as exc:
            raise ValueError(f"metadata is not valid JSON: {exc}")
    if not isinstance(value, dict):
        raise ValueError("metadata must be a JSON object")
    payload = json.dumps(value, separators=(",", ":"), ensure_ascii=False)
    if len(payload.encode("utf-8")) > MAX_METADATA_BYTES:
        raise ValueError(f"metadata exceeds {MAX_METADATA_BYTES}-byte limit")
    return _validate_tool_names_in_metadata(value)


def _validate_terminal_metadata(value: Any) -> dict[str, Any]:
    """Validate metadata for a terminal recorder, shrinking (not raising) when it
    exceeds the byte budget.

    Terminal recording (completion / blocked-salvage / failure) must never be
    lost to a size error, regardless of which caller assembled the metadata
    (incidents kt_1de7d1f3d14b4e and kt_6cc4cd8e862947, where an oversized
    handoff threw, was swallowed, and pinned the task ``in_progress`` with a live
    claim). Non-size validation errors (bad JSON, non-dict) still raise.
    """
    try:
        return _validate_metadata(value or {})
    except ValueError as exc:
        if "byte limit" not in str(exc):
            raise
        return _validate_metadata(shrink_metadata_to_limit(value or {}))


def metadata_within_limit(value: dict[str, Any]) -> bool:
    """True when ``value`` serializes under the metadata byte budget."""
    try:
        payload = json.dumps(value, separators=(",", ":"), ensure_ascii=False)
        return len(payload.encode("utf-8")) <= MAX_METADATA_BYTES
    except Exception:
        return False


def shrink_metadata_to_limit(value: Any) -> dict[str, Any]:
    """Deterministically reduce a metadata dict until it fits the byte budget.

    Terminal outcome recording must never be lost to a validation error
    (incident kt_1de7d1f3d14b4e: oversized metadata made record_completion —
    and its record_failure fallback — raise, pinning the task in_progress
    forever). Reduction order: drop dispatcher bookkeeping (``progress_log``),
    truncate long string values, then drop the largest remaining values.
    Stamps ``metadata_truncated: true`` whenever anything was removed.
    Never raises; returns {} for non-dict input.
    """
    if not isinstance(value, dict):
        return {}
    md = dict(value)
    if metadata_within_limit(md):
        return md
    md.pop("progress_log", None)
    md["metadata_truncated"] = True
    if metadata_within_limit(md):
        return md
    for key, val in list(md.items()):
        if isinstance(val, str) and len(val) > 500:
            md[key] = val[:500] + "…[truncated]"
    if metadata_within_limit(md):
        return md
    # Drop largest values until it fits (keep small scalars and markers).
    def _size(item: tuple[str, Any]) -> int:
        try:
            return len(json.dumps(item[1], separators=(",", ":"), ensure_ascii=False))
        except Exception:
            return 1 << 20
    for key, _ in sorted(md.items(), key=_size, reverse=True):
        if key == "metadata_truncated":
            continue
        md.pop(key, None)
        if metadata_within_limit(md):
            return md
    return {"metadata_truncated": True}


def _validate_deadline(seconds: Any) -> str:
    if seconds in (None, "", 0):
        return ""
    try:
        s = int(seconds)
    except (TypeError, ValueError):
        raise ValueError("deadline_seconds must be an integer")
    if s <= 0:
        return ""
    if s > 7 * 24 * 3600:
        raise ValueError("deadline_seconds may not exceed 7 days")
    deadline = datetime.now(tz=timezone.utc) + timedelta(seconds=s)
    return deadline.strftime("%Y-%m-%dT%H:%M:%SZ")


def _validate_idempotency_key(value: Any) -> str:
    if value in (None, ""):
        return ""
    s = _validate_text("idempotency_key", value, max_chars=MAX_IDEMPOTENCY_CHARS)
    return s


def _kanban_enabled() -> bool:
    """Master kill switch — :mod:`core.config` ``[kanban].enabled``."""
    try:
        from core.config import get_kanban_config
        return bool(get_kanban_config().get("enabled", True))
    except Exception:
        _log.error("Kanban config load failed; treating kanban as disabled (fail-closed)", exc_info=True)
        return False


def _kanban_caps() -> dict[str, int]:
    try:
        from core.config import get_kanban_config
        cfg = get_kanban_config()
    except Exception:
        cfg = {}
    return {
        "max_active_per_tenant": int(cfg.get("max_active_tasks_per_tenant", 50)),
        "max_active_global": int(cfg.get("max_active_tasks_global", 200)),
        "rate_per_minute": int(cfg.get("create_rate_per_minute_per_tenant", 60)),
        "default_claim_ttl": int(cfg.get("default_claim_ttl_seconds", 900)),
        "default_deadline_seconds": int(cfg.get("default_deadline_seconds", 0)),
        "failure_limit": int(cfg.get("failure_limit", 3)),
        "retention_days": int(cfg.get("retention_days", 14)),
        "max_runs_per_task": int(cfg.get("max_runs_per_task", 50)),
    }


# ---------------------------------------------------------------------------
# Connection / retry helpers
# ---------------------------------------------------------------------------

class _BusyRetry:
    """Tiny SQLITE_BUSY backoff context manager; 3 attempts at 25/100/400 ms."""
    _delays = (0.025, 0.1, 0.4)

    def __init__(self) -> None:
        self.attempts = 0

    def __enter__(self) -> "_BusyRetry":
        return self

    def __exit__(self, exc_type, exc, tb) -> bool:
        if exc_type is sqlite3.OperationalError and "locked" in str(exc).lower():
            if self.attempts < len(self._delays):
                time.sleep(self._delays[self.attempts])
                self.attempts += 1
                return True
        return False


_migration_lock = threading.Lock()
_migrations_applied: bool = False


def _apply_kanban_migrations(conn: sqlite3.Connection) -> None:
    """Bring the kanban tables up to ``SCHEMA_VERSION``.

    v1 -> v2: backfill ``kanban_task_parents`` from the legacy single-parent
    column ``kanban_tasks.parent_id`` and stamp ``schema_version=2`` on
    every row. Idempotent — uses ``INSERT OR IGNORE`` and a guarded UPDATE.
    """
    cur = conn.execute(
        "SELECT MIN(schema_version) FROM kanban_tasks"
    ).fetchone()
    min_v = int(cur[0]) if cur and cur[0] is not None else SCHEMA_VERSION
    if min_v >= SCHEMA_VERSION:
        return
    if min_v < 2:
        # Backfill join rows for existing single-parent data. The PRIMARY
        # KEY (task_id, parent_id) means INSERT OR IGNORE makes this a
        # no-op when the row already exists from a prior migration pass.
        conn.execute(
            """
            INSERT OR IGNORE INTO kanban_task_parents (task_id, parent_id, ordinal, created_at)
            SELECT id, parent_id, 0, created_at
              FROM kanban_tasks
             WHERE parent_id IS NOT NULL AND parent_id != ''
            """
        )
        conn.execute(
            "UPDATE kanban_tasks SET schema_version=? WHERE schema_version < ?",
            (SCHEMA_VERSION, SCHEMA_VERSION),
        )


def _open_conn() -> sqlite3.Connection:
    """Return a fresh DB connection (the existing chat_storage._get_connection
    pattern; new connection per call, WAL mode is set on first connect).

    On the first call after process start, apply pending kanban schema
    migrations exactly once (gated by ``_migrations_applied``).
    """
    ensure_encryption_ready()
    conn = _get_connection()
    conn.row_factory = sqlite3.Row
    global _migrations_applied
    if not _migrations_applied:
        with _migration_lock:
            if not _migrations_applied:
                try:
                    conn.execute("BEGIN IMMEDIATE")
                    _apply_kanban_migrations(conn)
                    conn.commit()
                except sqlite3.OperationalError:
                    # Another process / thread won the race; their migration
                    # is correct, so we drop our transaction and proceed.
                    try:
                        conn.rollback()
                    except Exception:
                        pass
                _migrations_applied = True
    return conn


def _row_to_dict(row: sqlite3.Row | None) -> dict[str, Any] | None:
    if row is None:
        return None
    out: dict[str, Any] = {}
    for k in row.keys():
        out[k] = row[k]
    decrypt_failed_fields: list[str] = []
    # Decrypt blobs.
    for blob_field, target in (
        ("title_ciphertext", "title"),
        ("body_ciphertext", "body"),
        ("tags_ciphertext", "tags"),
        ("metadata_ciphertext", "metadata"),
        ("summary_ciphertext", "summary"),
        ("error_ciphertext", "error"),
    ):
        if blob_field in out:
            try:
                raw = _decrypt(out.pop(blob_field))
            except Exception:
                raw = ""
                decrypt_failed_fields.append(target)
                _log.warning(
                    "kanban decrypt failed for task %s field %s",
                    out.get("id"),
                    target,
                    exc_info=True,
                )
            if target == "tags":
                try:
                    out[target] = json.loads(raw) if raw else []
                except Exception:
                    out[target] = []
            elif target == "metadata":
                try:
                    out[target] = json.loads(raw) if raw else {}
                except Exception:
                    out[target] = {}
            else:
                out[target] = raw if raw else ("[decrypt failed]" if target in decrypt_failed_fields else raw)
    if decrypt_failed_fields:
        out["decrypt_error"] = True
        out["decrypt_failed_fields"] = list(decrypt_failed_fields)
    return out


# ---------------------------------------------------------------------------
# Parent-set helpers + cycle detection (multi-parent DAG)
# ---------------------------------------------------------------------------

def _get_parent_ids(conn: sqlite3.Connection, task_id: str) -> list[str]:
    """Return all parent ids for *task_id* in ordinal order. Falls back to
    the legacy single-parent column when no join rows are present (defensive
    against a partial migration; the v1 -> v2 backfill normally covers it).
    """
    if not task_id:
        return []
    rows = conn.execute(
        "SELECT parent_id FROM kanban_task_parents WHERE task_id=? "
        "ORDER BY ordinal ASC, parent_id ASC",
        (task_id,),
    ).fetchall()
    if rows:
        return [r[0] for r in rows if r[0]]
    legacy = conn.execute(
        "SELECT parent_id FROM kanban_tasks WHERE id=?", (task_id,)
    ).fetchone()
    if legacy and legacy[0]:
        return [legacy[0]]
    return []


def _get_child_ids(conn: sqlite3.Connection, parent_id: str) -> list[str]:
    """Return distinct child ids that depend on *parent_id*. Unions the join
    table with the legacy single-parent column to handle pre-migration rows.
    """
    if not parent_id:
        return []
    seen: set[str] = set()
    out: list[str] = []
    for row in conn.execute(
        "SELECT task_id FROM kanban_task_parents WHERE parent_id=?", (parent_id,)
    ).fetchall():
        cid = row[0]
        if cid and cid not in seen:
            seen.add(cid)
            out.append(cid)
    for row in conn.execute(
        "SELECT id FROM kanban_tasks WHERE parent_id=?", (parent_id,)
    ).fetchall():
        cid = row[0]
        if cid and cid not in seen:
            seen.add(cid)
            out.append(cid)
    return out


def _walk_parent_dag(
    conn: sqlite3.Connection,
    parent_ids: list[str] | str,
    *,
    extra_edges: dict[str, list[str]] | None = None,
) -> list[str]:
    """BFS over the parent DAG starting from *parent_ids* (inclusive of
    those roots). Returns the unique ancestor set in BFS order.

    Raises ``ValueError`` on:
      - any starting parent_id that does not exist;
      - dependency cycle reachable from the starting set;
      - node-cap (``PARENT_CHAIN_HOP_CAP``) or edge-cap
        (``PARENT_DAG_EDGE_CAP``) exceeded.

    *extra_edges* supplies in-flight edges that have not yet been written
    (used by ``create_task`` / ``create_batch`` so cycle detection runs
    against the merged DAG before any INSERT touches the DB).
    """
    if isinstance(parent_ids, str):
        roots = [parent_ids] if parent_ids else []
    else:
        roots = [p for p in parent_ids if p]
    if not roots:
        return []
    extra_edges = extra_edges or {}

    seen: set[str] = set()
    order: list[str] = []
    edges_walked = 0
    frontier: list[str] = list(roots)
    while frontier:
        cur = frontier.pop(0)
        if cur in seen:
            # A re-encounter is benign for a DAG; the cycle predicate fires
            # only when we re-encounter via the live frontier, which is
            # detected by the seen-check on append below.
            continue
        if len(seen) >= PARENT_CHAIN_HOP_CAP:
            raise ValueError(
                f"dependency DAG exceeds {PARENT_CHAIN_HOP_CAP} ancestor nodes"
            )
        # Existence: the row must be in kanban_tasks (extra_edges nodes are
        # by definition not yet in the DB but are seeded into ``seen``
        # before the walk for the in-batch case).
        if cur in extra_edges:
            parents = list(extra_edges.get(cur, []))
        else:
            row = conn.execute(
                "SELECT id FROM kanban_tasks WHERE id=?", (cur,)
            ).fetchone()
            if row is None:
                raise ValueError(f"parent_id {cur} does not exist")
            parents = _get_parent_ids(conn, cur)
        seen.add(cur)
        order.append(cur)
        for p in parents:
            edges_walked += 1
            if edges_walked > PARENT_DAG_EDGE_CAP:
                raise ValueError(
                    f"dependency DAG exceeds {PARENT_DAG_EDGE_CAP} edges"
                )
            if p in seen:
                # A direct cycle back into something already on this walk
                # via the same root set. We treat that as a cycle.
                if p in roots or p in order:
                    raise ValueError(f"dependency cycle detected at {p}")
                continue
            frontier.append(p)
    return order


def _walk_parent_chain(conn: sqlite3.Connection, parent_id: str) -> list[str]:
    """Backward-compatible wrapper. ``_walk_parent_dag`` is the canonical
    walker for cycle detection; this preserves the historic single-parent
    chain shape for any caller that explicitly wants the linear chain
    (currently none — kept for API stability)."""
    return _walk_parent_dag(conn, [parent_id])


def _validate_parent_id_list(
    parent_id: str | None,
    parent_ids: Iterable[str] | None,
) -> list[str]:
    """Normalize the ``parent_id`` / ``parent_ids`` overload into a single
    ordered, de-duplicated list. Rejects mixed inputs that disagree, an
    over-cap list, or non-string entries.
    """
    if parent_id and parent_ids:
        ids_list = [p for p in parent_ids if p]
        if not ids_list or ids_list[0] != parent_id and parent_id not in ids_list:
            raise ValueError(
                "parent_id and parent_ids both set and disagree; "
                "use parent_ids alone for multi-parent or parent_id alone for single."
            )
        # When both are set with parent_id mirrored as ids_list[0] (or
        # present in ids_list), prefer the plural form.
    out: list[str] = []
    seen: set[str] = set()
    if parent_ids:
        for raw in parent_ids:
            if not isinstance(raw, str):
                raise ValueError("parent_ids must be a list of strings")
            s = raw.strip()
            if not s:
                continue
            if s in seen:
                continue
            seen.add(s)
            out.append(s)
    elif parent_id:
        s = parent_id.strip()
        if s:
            out.append(s)
            seen.add(s)
    if len(out) > MAX_PARENTS_PER_TASK:
        raise ValueError(
            f"more than {MAX_PARENTS_PER_TASK} parents declared "
            f"(got {len(out)})"
        )
    return out


# ---------------------------------------------------------------------------
# Public API — create
# ---------------------------------------------------------------------------

def create_task(
    title: str,
    body: str,
    *,
    assignee: str = "general",
    parent_id: str | None = None,
    parent_ids: Iterable[str] | None = None,
    tenant: str = "default",
    priority: int = 3,
    tags: Any = None,
    metadata: Any = None,
    origin_thread_id: str = "",
    deadline_seconds: int | None = None,
    idempotency_key: str = "",
    model_override: str = "",
    max_retries_override: int | None = None,
    specifier_profile: str = "",
    _internal_allow_disabled: bool = False,
) -> dict[str, Any]:
    """Insert a new task. Returns the full task dict.

    Multi-parent: pass ``parent_ids=[...]`` for a fan-in / diamond shape.
    ``parent_id=`` (single) is a backward-compatible shorthand. Setting
    both is allowed only when they agree; disagreeing values raise
    ``ValueError``. The ``kanban_tasks.parent_id`` column stores the first
    parent (canonical "primary") for fast-path single-parent reads; the
    full set lives in ``kanban_task_parents``.

    Status auto-set: ``ready`` when no parents are declared, or when **all**
    declared parents are already ``done``; otherwise ``todo``. Idempotency:
    when ``idempotency_key`` is non-empty and a row already exists for
    ``(tenant, idempotency_key)``, returns the existing row instead of
    duplicating.
    """
    if not _kanban_enabled() and not _internal_allow_disabled:
        raise RuntimeError("kanban disabled in config")

    title_v = _validate_text("title", title, max_chars=MAX_TITLE_CHARS)
    body_v = _validate_text("body", body, max_chars=MAX_BODY_CHARS, allow_empty=True)
    assignee_v = _validate_assignee(assignee)
    tenant_v = _validate_text("tenant", tenant or "default", max_chars=MAX_TENANT_CHARS)
    priority_v = _validate_priority(priority)
    tags_v = _validate_tags(tags)
    metadata_v = _validate_metadata(metadata)
    from utils.kanban_context import stamp_default_task_acceptance

    metadata_v = stamp_default_task_acceptance(metadata_v if isinstance(metadata_v, dict) else {})

    specifier_v = specifier_profile.strip() if specifier_profile else ""
    if specifier_v:
        from core.config import get_kanban_config
        if not get_kanban_config().get("triage_enabled", True):
            raise ValueError("triage is disabled in config")
        metadata_v = dict(metadata_v) if metadata_v else {}
        metadata_v["specifier_profile"] = specifier_v
        assignee_v = "triage"
    elif assignee_v == "triage":
        from core.config import get_kanban_config
        if not get_kanban_config().get("triage_enabled", True):
            raise ValueError("triage is disabled in config")
    from utils.kanban_context import normalize_worker_task_metadata

    metadata_v = normalize_worker_task_metadata(metadata_v, assignee_v)
    origin_thread_v = _validate_text(
        "origin_thread_id", origin_thread_id or "",
        max_chars=MAX_THREAD_ID_CHARS, allow_empty=True,
    )
    deadline_at = _validate_deadline(deadline_seconds)
    idem_v = _validate_idempotency_key(idempotency_key)
    model_override_v = (model_override or "").strip()[:128]
    max_retries_v = -1 if max_retries_override is None else max(-1, int(max_retries_override))
    parent_ids_v = _validate_parent_id_list(parent_id, parent_ids)

    caps = _kanban_caps()
    if caps["rate_per_minute"] > 0 and not _internal_allow_disabled:
        # Idempotent replays return the existing row without creating anything;
        # they must not burn rate-limit capacity (a retry storm with the same
        # key could otherwise exhaust the tenant budget and start failing).
        is_replay = False
        if idem_v:
            conn = _open_conn()
            try:
                is_replay = conn.execute(
                    "SELECT 1 FROM kanban_tasks WHERE tenant=? AND idempotency_key=?",
                    (tenant_v, idem_v),
                ).fetchone() is not None
            finally:
                conn.close()
        if not is_replay and not try_acquire_kanban_create_slot(
            tenant_v, cap_per_minute=caps["rate_per_minute"]
        ):
            raise ValueError(f"rate limit exceeded for tenant '{tenant_v}'")

    retry = _BusyRetry()
    while True:
        with retry:
            return _create_task_inner(
                title=title_v,
                body=body_v,
                assignee=assignee_v,
                parent_ids=parent_ids_v,
                tenant=tenant_v,
                priority=priority_v,
                tags=tags_v,
                metadata=metadata_v,
                origin_thread_id=origin_thread_v,
                deadline_at=deadline_at,
                idempotency_key=idem_v,
                model_override=model_override_v,
                max_retries_override=max_retries_v,
                caps=caps,
            )


def _create_task_inner(
    *,
    title: str,
    body: str,
    assignee: str,
    parent_ids: list[str],
    tenant: str,
    priority: int,
    tags: list[str],
    metadata: dict,
    origin_thread_id: str,
    deadline_at: str,
    idempotency_key: str,
    model_override: str,
    max_retries_override: int,
    caps: dict[str, int],
) -> dict[str, Any]:
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")

        # Idempotency replay.
        if idempotency_key:
            row = conn.execute(
                "SELECT * FROM kanban_tasks WHERE tenant=? AND idempotency_key=?",
                (tenant, idempotency_key),
            ).fetchone()
            if row is not None:
                conn.commit()
                replayed = _row_to_dict(row)
                if replayed and replayed.get("decrypt_error"):
                    raise ValueError(
                        f"idempotency replay for key {idempotency_key!r} refused: "
                        f"existing task {replayed.get('id')} has decrypt errors "
                        f"({replayed.get('decrypt_failed_fields')})"
                    )
                if replayed:
                    # Mirror the fresh-create return shape so callers see the
                    # task's DAG edges on replay too.
                    existing_parents = _get_parent_ids(conn, replayed.get("id") or "")
                    if existing_parents:
                        replayed["parent_ids"] = existing_parents
                return replayed or {}

        # Backpressure: active counts (todo/ready/in_progress/blocked).
        if caps["max_active_per_tenant"] > 0:
            active_t = conn.execute(
                "SELECT COUNT(*) FROM kanban_tasks WHERE tenant=? AND status IN "
                "('todo','ready','in_progress','blocked')",
                (tenant,),
            ).fetchone()[0]
            if active_t >= caps["max_active_per_tenant"]:
                conn.rollback()
                raise ValueError(
                    f"tenant '{tenant}' has too many active tasks "
                    f"({active_t} >= {caps['max_active_per_tenant']})"
                )
        if caps["max_active_global"] > 0:
            active_g = conn.execute(
                "SELECT COUNT(*) FROM kanban_tasks WHERE status IN "
                "('todo','ready','in_progress','blocked')"
            ).fetchone()[0]
            if active_g >= caps["max_active_global"]:
                conn.rollback()
                raise ValueError(
                    f"global active task cap reached "
                    f"({active_g} >= {caps['max_active_global']})"
                )

        # Parent existence, cycle check, AND-all-parents-done evaluation.
        all_parents_done = True
        if parent_ids:
            _walk_parent_dag(conn, parent_ids)  # raises on cycle / missing
            placeholders = ",".join(["?"] * len(parent_ids))
            statuses = conn.execute(
                f"SELECT status FROM kanban_tasks WHERE id IN ({placeholders})",
                parent_ids,
            ).fetchall()
            if len(statuses) != len(parent_ids):
                # Should be caught by the walk; defensive.
                conn.rollback()
                raise ValueError("one or more declared parents do not exist")
            all_parents_done = all(s[0] == STATUS_DONE for s in statuses)

        # Apply default deadline if config has one and the caller didn't.
        if not deadline_at and caps["default_deadline_seconds"] > 0:
            deadline_at = _validate_deadline(caps["default_deadline_seconds"])

        task_id = _new_id("kt")
        now = _now_iso()
        primary_parent: str | None = parent_ids[0] if parent_ids else None
        status = STATUS_READY if (not parent_ids or all_parents_done) else STATUS_TODO

        conn.execute(
            """
            INSERT INTO kanban_tasks (
                id, title_ciphertext, body_ciphertext,
                status, assignee, tenant, priority,
                parent_id, origin_thread_id,
                tags_ciphertext, metadata_ciphertext,
                failure_count, last_attempt,
                claim_handle, claim_token, claim_expires_at,
                deadline_at, idempotency_key,
                model_override, max_retries_override,
                created_at, updated_at, completed_at, schema_version
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, '', '', '', ?, ?, ?, ?, ?, ?, '', ?)
            """,
            (
                task_id,
                _encrypt(title),
                _encrypt(body),
                status,
                assignee,
                tenant,
                priority,
                primary_parent,
                origin_thread_id,
                _encrypt(json.dumps(tags, ensure_ascii=False)),
                _encrypt(json.dumps(metadata, ensure_ascii=False)),
                deadline_at,
                idempotency_key,
                model_override,
                max_retries_override,
                now,
                now,
                SCHEMA_VERSION,
            ),
        )
        for ordinal, pid in enumerate(parent_ids):
            conn.execute(
                "INSERT INTO kanban_task_parents (task_id, parent_id, ordinal, created_at) "
                "VALUES (?, ?, ?, ?)",
                (task_id, pid, ordinal, now),
            )
        conn.commit()
        row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
    finally:
        conn.close()

    out = _row_to_dict(row) or {}
    if parent_ids:
        out["parent_ids"] = list(parent_ids)
    _emit_lifecycle_event(
        event_type="created",
        task_id=task_id,
        from_status="",
        to_status=status,
        assignee=assignee,
        tenant=tenant,
        run_id="",
        attempt=None,
    )
    _bump_search_generation()
    return out


def create_batch(
    items: list[dict[str, Any]],
    *,
    tenant: str = "default",
    origin_thread_id: str = "",
    _internal_allow_disabled: bool = False,
) -> list[dict[str, Any]]:
    """Insert N tasks in a single transaction with parent resolution.

    Each item may declare its dependencies via any of:

    - ``parent_index``: 0-based int referencing an earlier item in the
      same batch. Single-parent shorthand.
    - ``parent_indices``: list of 0-based ints referencing earlier items.
      Multi-parent (fan-in / diamond) shape.
    - ``parent_id``: existing row id (single).
    - ``parent_ids``: existing row ids (multi).

    Indices and ids may be mixed (an item may declare both a batch index
    and an external id). Cycle check runs over the merged DAG (in-batch
    edges + each external parent's ancestors). On any error the entire
    batch is rolled back.

    Returns the inserted rows in input order.
    """
    if not _kanban_enabled() and not _internal_allow_disabled:
        raise RuntimeError("kanban disabled in config")
    if not isinstance(items, list) or not items:
        raise ValueError("items must be a non-empty list")
    if len(items) > 64:
        raise ValueError("create_batch supports at most 64 items per call")

    tenant_v = _validate_text("tenant", tenant or "default", max_chars=MAX_TENANT_CHARS)
    origin_v = _validate_text(
        "origin_thread_id", origin_thread_id or "",
        max_chars=MAX_THREAD_ID_CHARS, allow_empty=True,
    )

    is_council_batch = False
    for raw in items:
        if isinstance(raw, dict):
            md = raw.get("metadata") or raw.get("metadata_json") or {}
            if isinstance(md, dict) and md.get("origin") == "council":
                is_council_batch = True
                break

    if origin_v:
        if is_council_batch:
            try:
                from utils.council_orchestrator import council_batch_overlap_error

                council_overlap = council_batch_overlap_error(origin_thread_id=origin_v)
                if council_overlap:
                    raise ValueError(council_overlap)
            except ValueError:
                raise
            except Exception as exc:
                _log.error("council overlap check failed", exc_info=True)
                raise ValueError(f"council overlap check failed: {exc}") from exc
        else:
            try:
                from utils.kanban_coordinator import coordinator_batch_overlap_error

                overlap_err = coordinator_batch_overlap_error(
                    origin_thread_id=origin_v,
                    batch_size=len(items),
                )
                if overlap_err:
                    raise ValueError(overlap_err)
            except ValueError:
                raise
            except Exception as exc:
                _log.error("coordinator overlap check failed", exc_info=True)
                raise ValueError(f"coordinator overlap check failed: {exc}") from exc

    coordinator_batch_id = ""
    coordinator_activate = False
    try:
        from utils.kanban_coordinator import (
            COORDINATOR_BATCH_ID_META,
            new_coordinator_batch_id,
            should_activate_coordinator,
        )

        coordinator_activate = bool(
            not is_council_batch
            and should_activate_coordinator(origin_thread_id=origin_v, batch_size=len(items))
        )
        if coordinator_activate:
            coordinator_batch_id = new_coordinator_batch_id()
            if not str(coordinator_batch_id or "").strip():
                raise ValueError("coordinator batch id allocation returned empty id")
    except ValueError:
        raise
    except Exception as exc:
        if coordinator_activate:
            _log.error("coordinator batch id allocation failed", exc_info=True)
            raise ValueError(f"coordinator batch id allocation failed: {exc}") from exc
        _log.warning("coordinator batch setup skipped: %s", exc)

    caps = _kanban_caps()

    # Coordinator batches: validate DAG shape before persisting.
    if coordinator_batch_id:
        from utils.kanban_coordinator import validate_coordinator_batch_structure

        validate_coordinator_batch_structure(list(items))

    from utils.kanban_context import normalize_worker_task_metadata

    # Pre-validate every item so we don't open a DB connection for malformed input.
    normalized: list[dict[str, Any]] = []
    for idx, raw in enumerate(items):
        if not isinstance(raw, dict):
            raise ValueError(f"items[{idx}] must be a dict")
        title = _validate_text(f"items[{idx}].title", raw.get("title"), max_chars=MAX_TITLE_CHARS)
        body = _validate_text(
            f"items[{idx}].body", raw.get("body") or "",
            max_chars=MAX_BODY_CHARS, allow_empty=True,
        )
        assignee = _validate_assignee(raw.get("assignee") or "general")
        priority = _validate_priority(raw.get("priority") if raw.get("priority") is not None else 3)
        tags = _validate_tags(raw.get("tags") or raw.get("tags_json") or [])
        metadata = _validate_metadata(raw.get("metadata") or raw.get("metadata_json") or {})
        if coordinator_batch_id:
            from utils.kanban_coordinator import infer_coordinator_task_kind_metadata

            metadata = infer_coordinator_task_kind_metadata(
                metadata,
                title=title,
                body=body,
            )
        from utils.kanban_context import stamp_default_task_acceptance

        metadata = stamp_default_task_acceptance(metadata)
        if coordinator_batch_id:
            from utils.kanban_coordinator import COORDINATOR_BATCH_ID_META

            metadata = dict(metadata)
            metadata[COORDINATOR_BATCH_ID_META] = coordinator_batch_id
        raw_tool_names = raw.get("tool_names")
        if raw_tool_names is not None:
            metadata = dict(metadata) if metadata else {}
            metadata["tool_names"] = raw_tool_names
            metadata = _validate_tool_names_in_metadata(metadata)
        specifier = (raw.get("specifier_profile") or "").strip()
        if specifier:
            from core.config import get_kanban_config
            if not get_kanban_config().get("triage_enabled", True):
                raise ValueError("triage is disabled in config")
            metadata = dict(metadata) if metadata else {}
            metadata["specifier_profile"] = specifier
            assignee = "triage"
        elif assignee == "triage":
            from core.config import get_kanban_config
            if not get_kanban_config().get("triage_enabled", True):
                raise ValueError("triage is disabled in config")
        deadline_at = _validate_deadline(raw.get("deadline_seconds"))
        idem = _validate_idempotency_key(raw.get("idempotency_key") or "")
        model_override = (raw.get("model_override") or "").strip()[:128]
        max_retries_v = raw.get("max_retries_override")
        max_retries = -1 if max_retries_v is None else max(-1, int(max_retries_v))

        parent_indices_raw = raw.get("parent_indices")
        parent_ids_raw = raw.get("parent_ids")
        parent_index_raw = raw.get("parent_index")
        parent_id_raw = raw.get("parent_id")

        index_set: list[int] = []
        if parent_indices_raw is not None:
            if not isinstance(parent_indices_raw, list):
                raise ValueError(f"items[{idx}].parent_indices must be a list of integers")
            for pi in parent_indices_raw:
                try:
                    pi_i = int(pi)
                except (TypeError, ValueError):
                    raise ValueError(
                        f"items[{idx}].parent_indices entries must be integers"
                    )
                if pi_i < 0 or pi_i >= idx:
                    raise ValueError(
                        f"items[{idx}].parent_indices={pi_i} must reference an earlier item"
                    )
                if pi_i not in index_set:
                    index_set.append(pi_i)
        if parent_index_raw is not None:
            try:
                pi_i = int(parent_index_raw)
            except (TypeError, ValueError):
                raise ValueError(f"items[{idx}].parent_index must be an integer")
            if pi_i < 0 or pi_i >= idx:
                raise ValueError(
                    f"items[{idx}].parent_index={pi_i} must reference an earlier item"
                )
            if pi_i not in index_set:
                index_set.append(pi_i)

        ids_set: list[str] = _validate_parent_id_list(
            parent_id_raw if isinstance(parent_id_raw, str) else None,
            parent_ids_raw if isinstance(parent_ids_raw, list) else None,
        )

        if len(index_set) + len(ids_set) > MAX_PARENTS_PER_TASK:
            raise ValueError(
                f"items[{idx}] declares more than {MAX_PARENTS_PER_TASK} parents"
            )

        metadata = normalize_worker_task_metadata(metadata, assignee)

        normalized.append({
            "title": title,
            "body": body,
            "assignee": assignee,
            "priority": priority,
            "tags": tags,
            "metadata": metadata,
            "deadline_at": deadline_at,
            "idempotency_key": idem,
            "model_override": model_override,
            "max_retries_override": max_retries,
            "parent_indices": index_set,
            "parent_ids_external": ids_set,
        })

    # Rate check AFTER pre-validation, all-or-nothing. A malformed batch must
    # cost zero quota: the June 2026 stress test showed an agent iterating on
    # payload mistakes (priority range, parent_indices) burning the entire
    # tenant budget on rejected attempts, then being rate-limited out of the
    # corrected call. One ticket per item — a 50-task DAG still cannot bypass
    # the per-tenant minute budget.
    if caps["rate_per_minute"] > 0:
        from utils.kanban_activity import try_acquire_kanban_create_slots

        if not try_acquire_kanban_create_slots(
            tenant_v, count=len(items), cap_per_minute=caps["rate_per_minute"]
        ):
            raise ValueError(f"rate limit exceeded for tenant '{tenant_v}'")

    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        # Backpressure (count once, treat batch as a single atomic add).
        if caps["max_active_per_tenant"] > 0:
            active_t = conn.execute(
                "SELECT COUNT(*) FROM kanban_tasks WHERE tenant=? AND status IN "
                "('todo','ready','in_progress','blocked')",
                (tenant_v,),
            ).fetchone()[0]
            if active_t + len(items) > caps["max_active_per_tenant"]:
                conn.rollback()
                raise ValueError(
                    f"tenant '{tenant_v}' batch would exceed active cap "
                    f"({active_t} + {len(items)} > {caps['max_active_per_tenant']})"
                )
        if caps["max_active_global"] > 0:
            active_g = conn.execute(
                "SELECT COUNT(*) FROM kanban_tasks WHERE status IN "
                "('todo','ready','in_progress','blocked')"
            ).fetchone()[0]
            if active_g + len(items) > caps["max_active_global"]:
                conn.rollback()
                raise ValueError(
                    f"global batch would exceed active cap "
                    f"({active_g} + {len(items)} > {caps['max_active_global']})"
                )

        # Allocate ids up front so parent_index can be resolved.
        ids: list[str] = [_new_id("kt") for _ in items]

        # Build merged DAG edges for the cycle check (in-batch + external).
        # extra_edges maps every node (in-batch + external) to its parent set.
        extra_edges: dict[str, list[str]] = {}
        external_parent_done: dict[str, bool] = {}
        for idx, item in enumerate(normalized):
            parents_for_idx: list[str] = []
            for pi in item["parent_indices"]:
                parents_for_idx.append(ids[pi])
            for pid in item["parent_ids_external"]:
                if pid not in external_parent_done:
                    row = conn.execute(
                        "SELECT status FROM kanban_tasks WHERE id=?", (pid,)
                    ).fetchone()
                    if row is None:
                        conn.rollback()
                        raise ValueError(f"parent_id {pid} does not exist")
                    external_parent_done[pid] = (row[0] == STATUS_DONE)
                parents_for_idx.append(pid)
            extra_edges[ids[idx]] = parents_for_idx

        # Cycle / cap check across the merged DAG.
        for idx in range(len(normalized)):
            roots = extra_edges[ids[idx]]
            if roots:
                _walk_parent_dag(conn, roots, extra_edges=extra_edges)

        now = _now_iso()
        for idx, item in enumerate(normalized):
            parents_for_idx = extra_edges[ids[idx]]
            in_batch_parent_count = len(item["parent_indices"])
            # An in-batch parent is by definition not yet done.
            external_done_flags = [
                external_parent_done.get(pid, False)
                for pid in item["parent_ids_external"]
            ]
            all_external_done = all(external_done_flags) if external_done_flags else True
            all_done = (
                in_batch_parent_count == 0 and all_external_done
                if parents_for_idx
                else True
            )
            primary_parent: str | None = parents_for_idx[0] if parents_for_idx else None
            status = STATUS_READY if (not parents_for_idx or all_done) else STATUS_TODO

            conn.execute(
                """
                INSERT INTO kanban_tasks (
                    id, title_ciphertext, body_ciphertext,
                    status, assignee, tenant, priority,
                    parent_id, origin_thread_id,
                    tags_ciphertext, metadata_ciphertext,
                    failure_count, last_attempt,
                    claim_handle, claim_token, claim_expires_at,
                    deadline_at, idempotency_key,
                    model_override, max_retries_override,
                    created_at, updated_at, completed_at, schema_version
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, '', '', '', ?, ?, ?, ?, ?, ?, '', ?)
                """,
                (
                    ids[idx],
                    _encrypt(item["title"]),
                    _encrypt(item["body"]),
                    status,
                    item["assignee"],
                    tenant_v,
                    item["priority"],
                    primary_parent,
                    origin_v,
                    _encrypt(json.dumps(item["tags"], ensure_ascii=False)),
                    _encrypt(json.dumps(item["metadata"], ensure_ascii=False)),
                    item["deadline_at"],
                    item["idempotency_key"],
                    item["model_override"],
                    item["max_retries_override"],
                    now,
                    now,
                    SCHEMA_VERSION,
                ),
            )
            for ordinal, pid in enumerate(parents_for_idx):
                conn.execute(
                    "INSERT INTO kanban_task_parents (task_id, parent_id, ordinal, created_at) "
                    "VALUES (?, ?, ?, ?)",
                    (ids[idx], pid, ordinal, now),
                )
        conn.commit()
        out_rows = list(
            conn.execute(
                f"SELECT * FROM kanban_tasks WHERE id IN ({','.join(['?'] * len(ids))})",
                ids,
            ).fetchall()
        )
    finally:
        conn.close()

    # Re-sort to input order so callers can match by index.
    by_id = {r["id"]: r for r in out_rows}
    out: list[dict[str, Any]] = []
    for ord_idx, tid in enumerate(ids):
        d = _row_to_dict(by_id.get(tid))
        if d is None:
            continue
        d["parent_ids"] = list(extra_edges.get(tid, []))
        out.append(d)
        _emit_lifecycle_event(
            event_type="created",
            task_id=tid,
            from_status="",
            to_status=d.get("status", STATUS_TODO),
            assignee=d.get("assignee", ""),
            tenant=tenant_v,
            run_id="",
            attempt=None,
        )
    _bump_search_generation()

    if coordinator_batch_id and origin_v:
        try:
            from utils.kanban_coordinator import (
                auto_subscribe_coordinator_batch,
                set_active_batch,
            )

            set_active_batch(origin_v, coordinator_batch_id)
            auto_subscribe_coordinator_batch(out, origin_thread_id=origin_v)
        except Exception:
            _log.error(
                "Coordinator activation failed for %s batch %s after task insert",
                origin_v,
                coordinator_batch_id,
                exc_info=True,
            )

    return out


# ---------------------------------------------------------------------------
# Public API — read
# ---------------------------------------------------------------------------

def get_task(task_id: str) -> dict[str, Any] | None:
    if not task_id:
        return None
    conn = _open_conn()
    try:
        row = conn.execute(
            "SELECT * FROM kanban_tasks WHERE id=?", (task_id,)
        ).fetchone()
        if row is None:
            return None
        d = _row_to_dict(row)
        if d is not None:
            d["parent_ids"] = _get_parent_ids(conn, task_id)
        return d
    finally:
        conn.close()


def list_tasks(
    *,
    status: str | Iterable[str] | None = None,
    assignee: str | None = None,
    tenant: str | None = None,
    parent_id: str | None = None,
    origin_thread_id: str | None = None,
    limit: int = 50,
    max_limit: int = 200,
    order: str = "asc",
) -> list[dict[str, Any]]:
    """List tasks for the board view. Returns decrypted rows ordered by
    (priority asc, created_at asc) which matches the dispatcher's claim
    order — the board therefore shows what will run next at the top."""
    limit = max(1, min(int(limit or 50), int(max_limit or 200)))
    where: list[str] = []
    params: list[Any] = []
    if status:
        if isinstance(status, str):
            statuses = [status]
        else:
            statuses = [s for s in status if isinstance(s, str)]
        if statuses:
            where.append(f"status IN ({','.join(['?'] * len(statuses))})")
            params.extend(statuses)
    if assignee:
        where.append("assignee=?")
        params.append(assignee)
    if tenant:
        where.append("tenant=?")
        params.append(tenant)
    if parent_id is not None:
        if parent_id == "":
            where.append("(parent_id IS NULL OR parent_id='')")
        else:
            where.append("parent_id=?")
            params.append(parent_id)
    if origin_thread_id is not None:
        where.append("origin_thread_id=?")
        params.append(origin_thread_id)
    sql = "SELECT * FROM kanban_tasks"
    if where:
        sql += " WHERE " + " AND ".join(where)
    order_cl = (order or "asc").strip().lower()
    if order_cl == "desc":
        sql += " ORDER BY priority ASC, created_at DESC LIMIT ?"
    else:
        sql += " ORDER BY priority ASC, created_at ASC LIMIT ?"
    params.append(limit)

    conn = _open_conn()
    try:
        rows = conn.execute(sql, params).fetchall()
        out: list[dict[str, Any]] = []
        # Bulk-fetch parent_ids per row (one query per row is fine at the
        # current scale — list endpoints are capped at 200; if this ever
        # becomes hot we can switch to a single grouped query).
        for row in rows:
            d = _row_to_dict(row)
            if d is None:
                continue
            d["parent_ids"] = _get_parent_ids(conn, d.get("id") or "")
            out.append(d)
    finally:
        conn.close()
    return out


def search_tasks(
    query: str,
    *,
    tenant: str | None = None,
    limit: int = 20,
) -> list[dict[str, Any]]:
    """Full-text search across decrypted title, body, and tags.

    Since the underlying store is encrypted, this reads all rows
    (capped at 500) into memory, decrypts them, and does case-insensitive
    substring matching. Results are scored by number of field matches and
    returned in descending relevance order.
    """
    q = (query or "").strip().lower()
    if not q:
        return []
    where: list[str] = []
    params: list[Any] = []
    if tenant:
        where.append("tenant=?")
        params.append(tenant)
    sql = "SELECT * FROM kanban_tasks"
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " ORDER BY priority ASC, created_at ASC LIMIT 500"

    conn = _open_conn()
    try:
        rows = conn.execute(sql, params).fetchall()
        candidates: list[tuple[int, dict[str, Any]]] = []
        for row in rows:
            d = _row_to_dict(row)
            if d is None:
                continue
            score = 0
            title = (d.get("title") or "").lower()
            body = (d.get("body") or "").lower()
            tags_str = " ".join(d.get("tags") or []).lower()
            if q in title:
                score += 3
            if q in body:
                score += 2
            if q in tags_str:
                score += 1
            if q in (d.get("id") or "").lower():
                score += 4
            if score > 0:
                d["parent_ids"] = _get_parent_ids(conn, d.get("id") or "")
                d["_search_score"] = score
                candidates.append((score, d))
        candidates.sort(key=lambda x: -x[0])
    finally:
        conn.close()
    return [c[1] for c in candidates[:limit]]


def list_children(parent_id: str) -> list[dict[str, Any]]:
    """Return all tasks that depend on *parent_id*. Unions the join table
    with the legacy single-parent column so multi-parent and migrated rows
    surface alike. Sorted by (priority asc, created_at asc).
    """
    if not parent_id:
        return []
    conn = _open_conn()
    try:
        rows = conn.execute(
            "SELECT t.* FROM kanban_tasks t "
            "WHERE t.id IN ("
            "    SELECT task_id FROM kanban_task_parents WHERE parent_id=?"
            "    UNION "
            "    SELECT id FROM kanban_tasks WHERE parent_id=? AND parent_id != ''"
            ") "
            "ORDER BY t.priority ASC, t.created_at ASC LIMIT 200",
            (parent_id, parent_id),
        ).fetchall()
        out: list[dict[str, Any]] = []
        for row in rows:
            d = _row_to_dict(row)
            if d is None:
                continue
            d["parent_ids"] = _get_parent_ids(conn, d.get("id") or "")
            out.append(d)
    finally:
        conn.close()
    return out


def get_parent_ids(task_id: str) -> list[str]:
    """Public read of a task's full parent set (ordered)."""
    if not task_id:
        return []
    conn = _open_conn()
    try:
        return _get_parent_ids(conn, task_id)
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# Public API — claim / release / completion
# ---------------------------------------------------------------------------

def claim_task(
    *,
    assignee: str,
    handle_id: str = "",
    ttl_seconds: int | None = None,
) -> dict[str, Any] | None:
    """Atomically claim the highest-priority ready task for *assignee*.

    Returns the claimed task dict (with new ``last_attempt``, ``claim_token``)
    or ``None`` when no work is available. Two concurrent calls cannot both
    win — guaranteed by SQLite's row-level UPDATE atomicity.
    """
    assignee_v = _validate_assignee(assignee)
    if assignee_v == "triage":
        raise ValueError("triage tasks cannot be claimed; reassign first")
    handle_id_v = (handle_id or _new_id("kh"))[:64]
    caps = _kanban_caps()
    ttl = int(ttl_seconds if ttl_seconds is not None else caps["default_claim_ttl"])
    if ttl <= 0:
        ttl = caps["default_claim_ttl"]
    expires_at = (
        datetime.now(tz=timezone.utc) + timedelta(seconds=ttl)
    ).strftime("%Y-%m-%dT%H:%M:%SZ")
    token = _new_token()
    now = _now_iso()

    conn = _open_conn()
    try:
        if _supports_returning():
            try:
                conn.execute("BEGIN IMMEDIATE")
                row = conn.execute(
                    """
                    UPDATE kanban_tasks
                       SET status='in_progress',
                           claim_handle=?,
                           claim_token=?,
                           claim_expires_at=?,
                           last_attempt = last_attempt + 1,
                           updated_at=?
                     WHERE id = (
                         SELECT id FROM kanban_tasks
                          WHERE status='ready' AND assignee=?
                          ORDER BY priority ASC, created_at ASC
                          LIMIT 1
                     )
                       AND status='ready'
                    RETURNING *
                    """,
                    (handle_id_v, token, expires_at, now, assignee_v),
                ).fetchone()
                conn.commit()
            except sqlite3.OperationalError:
                conn.rollback()
                raise
        else:
            conn.execute("BEGIN IMMEDIATE")
            picked = conn.execute(
                "SELECT id FROM kanban_tasks "
                "WHERE status='ready' AND assignee=? "
                "ORDER BY priority ASC, created_at ASC LIMIT 1",
                (assignee_v,),
            ).fetchone()
            if picked is None:
                conn.commit()
                row = None
            else:
                rid = picked[0]
                cur = conn.execute(
                    """
                    UPDATE kanban_tasks
                       SET status='in_progress',
                           claim_handle=?,
                           claim_token=?,
                           claim_expires_at=?,
                           last_attempt = last_attempt + 1,
                           updated_at=?
                     WHERE id=? AND status='ready'
                    """,
                    (handle_id_v, token, expires_at, now, rid),
                )
                if cur.rowcount != 1:
                    conn.commit()
                    row = None
                else:
                    row = conn.execute(
                        "SELECT * FROM kanban_tasks WHERE id=?", (rid,)
                    ).fetchone()
                    conn.commit()
    finally:
        conn.close()

    out = _row_to_dict(row)
    if out is None:
        return None
    _emit_lifecycle_event(
        event_type="claimed",
        task_id=out["id"],
        from_status=STATUS_READY,
        to_status=STATUS_IN_PROGRESS,
        assignee=out.get("assignee", ""),
        tenant=out.get("tenant", "default"),
        run_id="",
        attempt=int(out.get("last_attempt") or 0),
    )
    return out


def release_claim(task_id: str, *, expected_token: str) -> bool:
    """Atomically release a claim back to ``ready``. Returns True when the
    release happened (the row was in_progress with the given token)."""
    if not task_id or not expected_token:
        return False
    now = _now_iso()
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        cur = conn.execute(
            """
            UPDATE kanban_tasks
               SET status='ready',
                   claim_handle='',
                   claim_token='',
                   claim_expires_at='',
                   updated_at=?
             WHERE id=? AND status='in_progress' AND claim_token=?
            """,
            (now, task_id, expected_token),
        )
        conn.commit()
        if cur.rowcount == 1:
            row = conn.execute("SELECT assignee, tenant FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
            assignee = row[0] if row else ""
            tenant = row[1] if row else "default"
        else:
            assignee = ""
            tenant = "default"
    finally:
        conn.close()
    if cur.rowcount == 1:
        _emit_lifecycle_event(
            event_type="released",
            task_id=task_id,
            from_status=STATUS_IN_PROGRESS,
            to_status=STATUS_READY,
            assignee=assignee,
            tenant=tenant,
        )
        return True
    return False


def _decode_run_metadata_blob(ciphertext: str | None) -> dict[str, Any]:
    """Best-effort decrypt+parse of a run metadata ciphertext blob."""
    if not ciphertext:
        return {}
    try:
        parsed = json.loads(_decrypt(ciphertext) or "{}")
        return parsed if isinstance(parsed, dict) else {}
    except Exception:
        return {}


def _merge_terminal_run_metadata(
    existing: dict[str, Any] | None,
    incoming: dict[str, Any],
) -> dict[str, Any]:
    """Merge terminal outcome metadata without dropping live ``progress_log`` rows."""
    merged = dict(incoming)
    if isinstance(existing, dict):
        progress = existing.get("progress_log")
        if isinstance(progress, list):
            merged["progress_log"] = progress
    return merged


def _apply_terminal_progress_log(
    final_md: dict[str, Any],
    *,
    summary: str,
    success: bool,
) -> dict[str, Any]:
    """Close a live ``status`` tail (e.g. Generating response…) when the run ends."""
    try:
        from utils.kanban_worker_progress import finalize_progress_log_for_terminal

        out = dict(final_md)
        out["progress_log"] = finalize_progress_log_for_terminal(
            out.get("progress_log"),
            summary=summary,
            success=success,
        )
        return out
    except Exception:
        _log.debug("finalize progress_log on terminal run failed", exc_info=True)
        return final_md


def record_completion(
    task_id: str,
    *,
    claim_token: str,
    summary: str,
    metadata: Any = None,
    handle_id: str = "",
) -> dict[str, Any]:
    """Terminal completion path: closes the open run, sets the task to
    ``done``, resets failure_count, and promotes ready children. Returns the
    final task dict.
    """
    if not task_id:
        raise ValueError("task_id is required")
    if not claim_token:
        raise ValueError("claim_token is required")
    summary_v = _validate_text("summary", summary or "", max_chars=MAX_SUMMARY_CHARS, allow_empty=True)
    metadata_v = _validate_terminal_metadata(metadata or {})
    now = _now_iso()

    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT id, status, claim_token, last_attempt, assignee, tenant, parent_id "
            "FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None:
            conn.rollback()
            raise ValueError(f"task {task_id} not found")
        if row["status"] != STATUS_IN_PROGRESS:
            conn.rollback()
            raise ValueError(f"task {task_id} is not in_progress (got {row['status']})")
        if row["claim_token"] != claim_token:
            conn.rollback()
            raise PermissionError("claim_token mismatch")

        # Close active run (or insert a synthetic one if none was started).
        run_row = conn.execute(
            "SELECT id, metadata_ciphertext FROM kanban_task_runs WHERE task_id=? AND outcome='active' "
            "ORDER BY started_at DESC LIMIT 1",
            (task_id,),
        ).fetchone()
        if run_row is not None:
            existing_md = _decode_run_metadata_blob(run_row["metadata_ciphertext"])
            final_md = _apply_terminal_progress_log(
                _merge_terminal_run_metadata(existing_md, metadata_v),
                summary=summary_v,
                success=True,
            )
            conn.execute(
                "UPDATE kanban_task_runs SET outcome=?, summary_ciphertext=?, "
                "metadata_ciphertext=?, error_ciphertext=?, ended_at=? WHERE id=?",
                (
                    OUTCOME_COMPLETED,
                    _encrypt(summary_v),
                    _encrypt(json.dumps(final_md, ensure_ascii=False)),
                    _encrypt(""),
                    now,
                    run_row["id"],
                ),
            )
            run_id = run_row["id"]
        else:
            run_id = _new_id("kr")
            conn.execute(
                """
                INSERT INTO kanban_task_runs (
                    id, task_id, attempt, outcome, assignee, handle_id, claim_token,
                    summary_ciphertext, metadata_ciphertext, error_ciphertext,
                    started_at, ended_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    run_id,
                    task_id,
                    int(row["last_attempt"] or 1),
                    OUTCOME_COMPLETED,
                    row["assignee"],
                    handle_id,
                    claim_token,
                    _encrypt(summary_v),
                    _encrypt(json.dumps(metadata_v, ensure_ascii=False)),
                    _encrypt(""),
                    now,
                    now,
                ),
            )

        conn.execute(
            """
            UPDATE kanban_tasks
               SET status='done',
                   claim_handle='', claim_token='', claim_expires_at='',
                   failure_count=0,
                   updated_at=?, completed_at=?
             WHERE id=?
            """,
            (now, now, task_id),
        )

        # Promote children of this task whose other parents are also done.
        promoted = _promote_children_locked(conn, task_id)
        conn.commit()
        promoted_ids = list(promoted)
        final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
    finally:
        conn.close()

    final = _row_to_dict(final_row) or {}
    _emit_lifecycle_event(
        event_type="completed",
        task_id=task_id,
        from_status=STATUS_IN_PROGRESS,
        to_status=STATUS_DONE,
        assignee=final.get("assignee", ""),
        tenant=final.get("tenant", "default"),
        run_id=run_id,
        attempt=int(final.get("last_attempt") or 0),
    )
    for cid in promoted_ids:
        _emit_lifecycle_event(
            event_type="promoted",
            task_id=cid,
            from_status=STATUS_TODO,
            to_status=STATUS_READY,
            assignee="",
            tenant=final.get("tenant", "default"),
        )
    _bump_search_generation()
    return final


def record_salvage_blocked(
    task_id: str,
    *,
    claim_token: str,
    summary: str,
    metadata: Any = None,
    handle_id: str = "",
) -> dict[str, Any]:
    """Honest blocked salvage handoff: closes the run, sets task ``blocked``.

    Unlike :func:`record_completion`, this does **not** promote children.
    """
    if not task_id:
        raise ValueError("task_id is required")
    if not claim_token:
        raise ValueError("claim_token is required")
    summary_v = _validate_text("summary", summary or "", max_chars=MAX_SUMMARY_CHARS, allow_empty=True)
    metadata_v = _validate_terminal_metadata(metadata or {})
    now = _now_iso()

    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT id, status, claim_token, last_attempt, assignee, tenant, parent_id "
            "FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None:
            conn.rollback()
            raise ValueError(f"task {task_id} not found")
        if row["status"] != STATUS_IN_PROGRESS:
            conn.rollback()
            raise ValueError(f"task {task_id} is not in_progress (got {row['status']})")
        if row["claim_token"] != claim_token:
            conn.rollback()
            raise PermissionError("claim_token mismatch")

        run_row = conn.execute(
            "SELECT id, metadata_ciphertext FROM kanban_task_runs WHERE task_id=? AND outcome='active' "
            "ORDER BY started_at DESC LIMIT 1",
            (task_id,),
        ).fetchone()
        if run_row is not None:
            existing_md = _decode_run_metadata_blob(run_row["metadata_ciphertext"])
            final_md = _apply_terminal_progress_log(
                _merge_terminal_run_metadata(existing_md, metadata_v),
                summary=summary_v,
                success=False,
            )
            conn.execute(
                "UPDATE kanban_task_runs SET outcome=?, summary_ciphertext=?, "
                "metadata_ciphertext=?, error_ciphertext=?, ended_at=? WHERE id=?",
                (
                    OUTCOME_BLOCKED,
                    _encrypt(summary_v),
                    _encrypt(json.dumps(final_md, ensure_ascii=False)),
                    _encrypt(""),
                    now,
                    run_row["id"],
                ),
            )
            run_id = run_row["id"]
        else:
            run_id = _new_id("kr")
            conn.execute(
                """
                INSERT INTO kanban_task_runs (
                    id, task_id, attempt, outcome, assignee, handle_id, claim_token,
                    summary_ciphertext, metadata_ciphertext, error_ciphertext,
                    started_at, ended_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    run_id,
                    task_id,
                    int(row["last_attempt"] or 1),
                    OUTCOME_BLOCKED,
                    row["assignee"],
                    handle_id,
                    claim_token,
                    _encrypt(summary_v),
                    _encrypt(json.dumps(metadata_v, ensure_ascii=False)),
                    _encrypt(""),
                    now,
                    now,
                ),
            )

        conn.execute(
            """
            UPDATE kanban_tasks
               SET status='blocked',
                   claim_handle='', claim_token='', claim_expires_at='',
                   updated_at=?
             WHERE id=?
            """,
            (now, task_id),
        )
        conn.commit()
        final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
    finally:
        conn.close()

    final = _row_to_dict(final_row) or {}
    _emit_lifecycle_event(
        event_type="blocked",
        task_id=task_id,
        from_status=STATUS_IN_PROGRESS,
        to_status=STATUS_BLOCKED,
        assignee=final.get("assignee", ""),
        tenant=final.get("tenant", "default"),
        run_id=run_id,
        attempt=int(final.get("last_attempt") or 0),
    )
    _bump_search_generation()
    return final


def record_failure(
    task_id: str,
    *,
    claim_token: str,
    outcome: str,
    error: str = "",
    handle_id: str = "",
    metadata: Any = None,
) -> dict[str, Any]:
    """Record a non-success run result. Returns the updated task dict.

    The caller (dispatcher) decides whether to retry, gave_up, or block based
    on the returned ``failure_count`` and the configured ``failure_limit``.
    """
    if not task_id:
        raise ValueError("task_id is required")
    if not claim_token:
        raise ValueError("claim_token is required")
    if outcome not in {
        OUTCOME_BLOCKED, OUTCOME_SPAWN_FAILED, OUTCOME_CRASHED,
        OUTCOME_TIMED_OUT, OUTCOME_GAVE_UP, OUTCOME_CANCELLED,
    }:
        raise ValueError(f"invalid failure outcome: {outcome}")
    error_v = _validate_text(
        "error", (error or ""), max_chars=MAX_ERROR_CHARS, allow_empty=True
    )
    metadata_v = _validate_terminal_metadata(metadata or {})
    now = _now_iso()
    caps = _kanban_caps()
    failure_limit = caps["failure_limit"]

    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT id, status, claim_token, last_attempt, failure_count, max_retries_override, "
            "assignee, tenant, parent_id FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None:
            conn.rollback()
            raise ValueError(f"task {task_id} not found")
        if row["status"] != STATUS_IN_PROGRESS:
            conn.rollback()
            raise ValueError(f"task {task_id} is not in_progress (got {row['status']})")
        if row["claim_token"] != claim_token:
            conn.rollback()
            raise PermissionError("claim_token mismatch")

        # Close active run.
        run_row = conn.execute(
            "SELECT id, metadata_ciphertext FROM kanban_task_runs WHERE task_id=? AND outcome='active' "
            "ORDER BY started_at DESC LIMIT 1",
            (task_id,),
        ).fetchone()
        if run_row is not None:
            existing_md = _decode_run_metadata_blob(run_row["metadata_ciphertext"])
            final_md = _apply_terminal_progress_log(
                _merge_terminal_run_metadata(existing_md, metadata_v),
                summary=error_v or str(metadata_v.get("summary") or ""),
                success=False,
            )
            conn.execute(
                "UPDATE kanban_task_runs SET outcome=?, metadata_ciphertext=?, error_ciphertext=?, ended_at=? WHERE id=?",
                (outcome, _encrypt(json.dumps(final_md, ensure_ascii=False)), _encrypt(error_v), now, run_row["id"]),
            )
            run_id = run_row["id"]
        else:
            run_id = _new_id("kr")
            conn.execute(
                """
                INSERT INTO kanban_task_runs (
                    id, task_id, attempt, outcome, assignee, handle_id, claim_token,
                    summary_ciphertext, metadata_ciphertext, error_ciphertext,
                    started_at, ended_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    run_id,
                    task_id,
                    int(row["last_attempt"] or 1),
                    outcome,
                    row["assignee"],
                    handle_id,
                    claim_token,
                    _encrypt(""),
                    _encrypt(json.dumps(metadata_v, ensure_ascii=False)),
                    _encrypt(error_v),
                    now,
                    now,
                ),
            )

        new_failure_count = int(row["failure_count"] or 0) + 1
        # Per-task override caps below (never above) the global ceiling.
        max_retries_override = int(row["max_retries_override"] or -1)
        effective_limit = failure_limit
        if max_retries_override >= 0:
            effective_limit = min(failure_limit, max_retries_override + 1)

        # Final outcomes (gave_up / cancelled) always terminate the task.
        gave_up = (
            outcome == OUTCOME_GAVE_UP
            or outcome == OUTCOME_CANCELLED
            or new_failure_count >= effective_limit
        )

        if gave_up:
            new_status = STATUS_BLOCKED if outcome != OUTCOME_CANCELLED else STATUS_CANCELLED
            conn.execute(
                """
                UPDATE kanban_tasks
                   SET status=?,
                       claim_handle='', claim_token='', claim_expires_at='',
                       failure_count=?,
                       updated_at=?,
                       completed_at=CASE WHEN ?='cancelled' THEN ? ELSE completed_at END
                 WHERE id=?
                """,
                (new_status, new_failure_count, now, new_status, now, task_id),
            )
            # Synthesize a final gave_up run row when the underlying outcome
            # crossed the failure_limit (separate from the failed attempt
            # we just closed). This matches Hermes' "tail" gave_up row.
            if outcome != OUTCOME_GAVE_UP and outcome != OUTCOME_CANCELLED:
                tail_id = _new_id("kr")
                conn.execute(
                    """
                    INSERT INTO kanban_task_runs (
                        id, task_id, attempt, outcome, assignee, handle_id, claim_token,
                        summary_ciphertext, metadata_ciphertext, error_ciphertext,
                        started_at, ended_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        tail_id,
                        task_id,
                        int(row["last_attempt"] or 1),
                        OUTCOME_GAVE_UP,
                        row["assignee"],
                        handle_id,
                        claim_token,
                        _encrypt(""),
                        _encrypt(json.dumps(metadata_v, ensure_ascii=False)),
                        _encrypt(error_v),
                        now,
                        now,
                    ),
                )
        else:
            new_status = STATUS_READY  # retryable: dispatcher will reclaim
            conn.execute(
                """
                UPDATE kanban_tasks
                   SET status='ready',
                       claim_handle='', claim_token='', claim_expires_at='',
                       failure_count=?,
                       updated_at=?
                 WHERE id=?
                """,
                (new_failure_count, now, task_id),
            )
        conn.commit()
        final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
    finally:
        conn.close()

    final = _row_to_dict(final_row) or {}
    _emit_lifecycle_event(
        event_type="failed" if not gave_up else ("gave_up" if outcome != OUTCOME_CANCELLED else "cancelled"),
        task_id=task_id,
        from_status=STATUS_IN_PROGRESS,
        to_status=new_status,
        assignee=final.get("assignee", ""),
        tenant=final.get("tenant", "default"),
        run_id=run_id,
        attempt=int(final.get("last_attempt") or 0),
        extra={"failure_count": new_failure_count, "outcome": outcome},
    )
    _bump_search_generation()
    return final


# ---------------------------------------------------------------------------
# Operator paths — block / unblock / cancel / set_assignee
# ---------------------------------------------------------------------------

def block_task(task_id: str, *, reason: str) -> dict[str, Any]:
    """Operator-only: move a task to ``blocked`` regardless of status (except
    cancelled/done). Records a synthetic run with outcome='blocked'."""
    reason_v = _validate_text("reason", reason or "", max_chars=MAX_ERROR_CHARS, allow_empty=False)
    now = _now_iso()
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT id, status, last_attempt, assignee, tenant FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None:
            conn.rollback()
            raise ValueError(f"task {task_id} not found")
        if row["status"] in (STATUS_DONE, STATUS_CANCELLED):
            conn.rollback()
            raise ValueError(f"task {task_id} is already terminal ({row['status']})")
        prev_status = row["status"]
        conn.execute(
            """
            UPDATE kanban_tasks
               SET status='blocked',
                   claim_handle='', claim_token='', claim_expires_at='',
                   updated_at=?
             WHERE id=?
            """,
            (now, task_id),
        )
        run_id = _new_id("kr")
        conn.execute(
            """
            INSERT INTO kanban_task_runs (
                id, task_id, attempt, outcome, assignee, handle_id, claim_token,
                summary_ciphertext, metadata_ciphertext, error_ciphertext,
                started_at, ended_at
            ) VALUES (?, ?, ?, ?, ?, '', '', ?, ?, ?, ?, ?)
            """,
            (
                run_id, task_id, int(row["last_attempt"] or 0), OUTCOME_BLOCKED,
                row["assignee"], _encrypt(""), _encrypt("{}"), _encrypt(reason_v), now, now,
            ),
        )
        # Close any active run from a worker that may be in flight.
        conn.execute(
            "UPDATE kanban_task_runs SET outcome='cancelled', ended_at=? "
            "WHERE task_id=? AND outcome='active'",
            (now, task_id),
        )
        conn.commit()
        final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
    finally:
        conn.close()
    final = _row_to_dict(final_row) or {}
    _emit_lifecycle_event(
        event_type="blocked",
        task_id=task_id,
        from_status=prev_status,
        to_status=STATUS_BLOCKED,
        assignee=final.get("assignee", ""),
        tenant=final.get("tenant", "default"),
        run_id=run_id,
        attempt=int(final.get("last_attempt") or 0),
    )
    _bump_search_generation()
    return final


def unblock_task(task_id: str) -> dict[str, Any]:
    """Operator-only: move a blocked task back to ``ready`` (or ``todo`` if
    its parent is not yet done). Resets ``failure_count``."""
    now = _now_iso()
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT id, status, parent_id, assignee, tenant FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None:
            conn.rollback()
            raise ValueError(f"task {task_id} not found")
        if row["status"] != STATUS_BLOCKED:
            conn.rollback()
            raise ValueError(f"task {task_id} is not blocked (got {row['status']})")
        target = STATUS_READY
        # Multi-parent: the task can return to ready only when ALL of its
        # parents are done. If any parent is still active, return it to
        # todo so the promotion engine can re-fire it cleanly.
        parent_ids = _get_parent_ids(conn, task_id)
        if parent_ids:
            placeholders = ",".join(["?"] * len(parent_ids))
            statuses = conn.execute(
                f"SELECT status FROM kanban_tasks WHERE id IN ({placeholders})",
                parent_ids,
            ).fetchall()
            if (
                len(statuses) != len(parent_ids)
                or not all(s[0] == STATUS_DONE for s in statuses)
            ):
                target = STATUS_TODO
        conn.execute(
            "UPDATE kanban_tasks SET status=?, failure_count=0, updated_at=? WHERE id=?",
            (target, now, task_id),
        )
        conn.commit()
        final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
    finally:
        conn.close()
    final = _row_to_dict(final_row) or {}
    _emit_lifecycle_event(
        event_type="unblocked",
        task_id=task_id,
        from_status=STATUS_BLOCKED,
        to_status=target,
        assignee=final.get("assignee", ""),
        tenant=final.get("tenant", "default"),
    )
    _bump_search_generation()
    return final


def cancel_task(task_id: str, *, cascade: bool = True) -> list[str]:
    """Cancel a task. When ``cascade`` is True, recursively cancels all
    descendants reachable through either the join table or the legacy
    single-parent column. Diamond / fan-in shapes are de-duplicated, so a
    descendant reachable via multiple paths is cancelled exactly once.
    Returns the list of cancelled ids (including ``task_id``).

    Any active runs are closed with ``outcome='cancelled'``. The dispatcher
    detects the cancellation when its tracked handle's task disappears or
    when its drain pass observes the cancelled status.
    """
    now = _now_iso()
    cancelled: list[str] = []
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        # Collect the set of ids to cancel (BFS over the join table union
        # legacy parent_id). Multi-parent descendants are visited once.
        targets: list[str] = []
        frontier: list[str] = [task_id]
        seen: set[str] = set()
        while frontier:
            cur = frontier.pop()
            if cur in seen:
                continue
            seen.add(cur)
            row = conn.execute(
                "SELECT id, status FROM kanban_tasks WHERE id=?", (cur,)
            ).fetchone()
            if row is None:
                continue
            if row["status"] in (STATUS_DONE, STATUS_CANCELLED):
                continue
            targets.append(cur)
            if cascade:
                for cid in _get_child_ids(conn, cur):
                    if cid not in seen:
                        frontier.append(cid)
        if not targets:
            conn.commit()
            return []
        placeholders = ",".join(["?"] * len(targets))
        conn.execute(
            f"UPDATE kanban_tasks SET status='cancelled', "
            f"claim_handle='', claim_token='', claim_expires_at='', "
            f"updated_at=?, completed_at=? "
            f"WHERE id IN ({placeholders})",
            [now, now, *targets],
        )
        conn.execute(
            f"UPDATE kanban_task_runs SET outcome='cancelled', ended_at=? "
            f"WHERE outcome='active' AND task_id IN ({placeholders})",
            [now, *targets],
        )
        # Fan-in children that can now never promote: AND-promotion requires
        # *every* parent to reach done, and a cancelled parent never will.
        # Without cascade (or with a diamond shape cancelled mid-DAG) such
        # children would sit in `todo` forever. Surface them as `blocked`
        # with a typed reason run so the operator can re-parent, unblock,
        # or cancel them instead of inheriting a silent wedge.
        # (Runs after the cancel UPDATE, so cascade-cancelled children are
        # already 'cancelled' and naturally excluded.)
        wedged: dict[str, dict[str, Any]] = {}
        for r in conn.execute(
            f"SELECT DISTINCT t.id, t.assignee, t.tenant, t.last_attempt, p.parent_id "
            f"FROM kanban_tasks t JOIN kanban_task_parents p ON p.task_id = t.id "
            f"WHERE p.parent_id IN ({placeholders}) AND t.status='todo'",
            targets,
        ).fetchall():
            wedged.setdefault(r["id"], dict(r))
        for r in conn.execute(
            f"SELECT id, assignee, tenant, last_attempt, parent_id "
            f"FROM kanban_tasks WHERE parent_id IN ({placeholders}) AND status='todo'",
            targets,
        ).fetchall():
            wedged.setdefault(r["id"], dict(r))
        blocked_events: list[dict[str, Any]] = []
        for cid, info in wedged.items():
            cur = conn.execute(
                "UPDATE kanban_tasks SET status='blocked', updated_at=? "
                "WHERE id=? AND status='todo'",
                (now, cid),
            )
            if cur.rowcount != 1:
                continue
            run_id = _new_id("kr")
            reason = (
                f"parent task {info.get('parent_id') or '?'} was cancelled; this task "
                "can never become ready (all parents must reach done). Re-parent, "
                "unblock after fixing the DAG, or cancel it."
            )
            conn.execute(
                """
                INSERT INTO kanban_task_runs (
                    id, task_id, attempt, outcome, assignee, handle_id, claim_token,
                    summary_ciphertext, metadata_ciphertext, error_ciphertext,
                    started_at, ended_at
                ) VALUES (?, ?, ?, ?, ?, '', '', ?, ?, ?, ?, ?)
                """,
                (
                    run_id, cid, int(info.get("last_attempt") or 0), OUTCOME_BLOCKED,
                    info.get("assignee") or "", _encrypt(""), _encrypt("{}"),
                    _encrypt(reason), now, now,
                ),
            )
            blocked_events.append(
                {
                    "task_id": cid,
                    "assignee": info.get("assignee") or "",
                    "tenant": info.get("tenant") or "default",
                    "run_id": run_id,
                    "attempt": int(info.get("last_attempt") or 0),
                }
            )
        conn.commit()
        cancelled = list(targets)
    finally:
        conn.close()

    for tid in cancelled:
        _emit_lifecycle_event(
            event_type="cancelled",
            task_id=tid,
            from_status="",
            to_status=STATUS_CANCELLED,
            assignee="",
            tenant="default",
        )
    for evt in blocked_events:
        _emit_lifecycle_event(
            event_type="blocked",
            task_id=evt["task_id"],
            from_status=STATUS_TODO,
            to_status=STATUS_BLOCKED,
            assignee=evt["assignee"],
            tenant=evt["tenant"],
            run_id=evt["run_id"],
            attempt=evt["attempt"],
        )
    _bump_search_generation()
    return cancelled


def set_assignee(
    task_id: str,
    new_assignee: str,
    *,
    force: bool = False,
) -> dict[str, Any]:
    """Operator-only: re-route a task to a different profile.

    When the task is in_progress, ``force=True`` is required; the caller
    (HTTP layer) should signal the dispatcher's stop flag for the current
    handle separately. This function only mutates the row + closes the open
    run with outcome='cancelled' and resets to 'ready' under the new
    assignee.
    """
    new_assignee_v = _validate_assignee(new_assignee)
    now = _now_iso()
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT id, status, assignee, tenant, metadata_ciphertext FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None:
            conn.rollback()
            raise ValueError(f"task {task_id} not found")
        if row["status"] in (STATUS_DONE, STATUS_CANCELLED):
            conn.rollback()
            raise ValueError(f"task {task_id} is terminal ({row['status']})")
        old_assignee = row["assignee"]
        if old_assignee == new_assignee_v:
            conn.commit()
            final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
            return _row_to_dict(final_row) or {}
        if row["status"] == STATUS_IN_PROGRESS and not force:
            conn.rollback()
            raise ValueError("task is in_progress; pass force=True to seize")
        # Close active run as cancelled.
        conn.execute(
            "UPDATE kanban_task_runs SET outcome='cancelled', ended_at=? "
            "WHERE task_id=? AND outcome='active'",
            (now, task_id),
        )
        # Move to ready (or todo if any parent not done) under the new assignee.
        target = STATUS_READY
        parent_ids = _get_parent_ids(conn, task_id)
        if parent_ids:
            placeholders = ",".join(["?"] * len(parent_ids))
            statuses = conn.execute(
                f"SELECT status FROM kanban_tasks WHERE id IN ({placeholders})",
                parent_ids,
            ).fetchall()
            if (
                len(statuses) != len(parent_ids)
                or not all(s[0] == STATUS_DONE for s in statuses)
            ):
                target = STATUS_TODO
        from utils.kanban_context import normalize_worker_task_metadata

        metadata_v: dict[str, Any] = {}
        try:
            meta_raw = _decrypt(row["metadata_ciphertext"])
            metadata_v = json.loads(meta_raw) if meta_raw else {}
        except Exception:
            metadata_v = {}
        if not isinstance(metadata_v, dict):
            metadata_v = {}
        metadata_v = normalize_worker_task_metadata(metadata_v, new_assignee_v)
        metadata_cipher = _encrypt(json.dumps(metadata_v, ensure_ascii=False))
        conn.execute(
            """
            UPDATE kanban_tasks
               SET assignee=?,
                   status=?,
                   claim_handle='', claim_token='', claim_expires_at='',
                   metadata_ciphertext=?,
                   updated_at=?
             WHERE id=?
            """,
            (new_assignee_v, target, metadata_cipher, now, task_id),
        )
        conn.commit()
        final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
    finally:
        conn.close()
    final = _row_to_dict(final_row) or {}
    _emit_lifecycle_event(
        event_type="reassigned",
        task_id=task_id,
        from_status=row["status"],
        to_status=target,
        assignee=new_assignee_v,
        tenant=final.get("tenant", "default"),
        extra={"prev_assignee": old_assignee},
    )
    _bump_search_generation()
    return final


_OPERATOR_ALLOWED_TRANSITIONS = frozenset({
    ("todo", "ready"),
    ("todo", "blocked"),
    ("todo", "cancelled"),
    ("ready", "todo"),
    ("ready", "blocked"),
    ("ready", "cancelled"),
    ("blocked", "todo"),
    ("blocked", "ready"),
    ("blocked", "cancelled"),
    ("in_progress", "blocked"),
    ("in_progress", "cancelled"),
    # Operator restart of a cancelled task (e.g. cancelled by a dispatcher
    # shutdown race). Cancellation must never be a dead end the operator
    # cannot reverse from the board.
    ("cancelled", "ready"),
    ("cancelled", "todo"),
})


def set_status(
    task_id: str,
    new_status: str,
    *,
    reason: str = "",
) -> dict[str, Any]:
    """Operator-only: move a task between statuses.

    Only transitions listed in ``_OPERATOR_ALLOWED_TRANSITIONS`` are
    accepted. Delegates blocked→block_task and unblock→unblock_task
    for consistency. Cancelled transitions cascade to children.
    """
    if not task_id:
        raise ValueError("task_id required")
    new_status = new_status.strip().lower()

    if new_status == STATUS_BLOCKED:
        return block_task(task_id, reason=reason or "operator override")
    if new_status == STATUS_CANCELLED:
        cancel_task(task_id, cascade=True)
        conn = _open_conn()
        try:
            final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
        finally:
            conn.close()
        return _row_to_dict(final_row) or {}

    now = _now_iso()
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT id, status, assignee, tenant FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None:
            conn.rollback()
            raise ValueError(f"task {task_id} not found")
        old_status = row["status"]
        if old_status == new_status:
            conn.commit()
            final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
            return _row_to_dict(final_row) or {}

        if old_status == STATUS_BLOCKED and new_status in (STATUS_READY, STATUS_TODO):
            conn.rollback()
            return unblock_task(task_id)

        if (old_status, new_status) not in _OPERATOR_ALLOWED_TRANSITIONS:
            conn.rollback()
            raise ValueError(
                f"transition {old_status}→{new_status} not allowed via operator override"
            )
        if old_status == STATUS_IN_PROGRESS:
            conn.execute(
                "UPDATE kanban_task_runs SET outcome='cancelled', ended_at=? "
                "WHERE task_id=? AND outcome='active'",
                (now, task_id),
            )
            conn.execute(
                "UPDATE kanban_tasks SET claim_handle='', claim_token='', claim_expires_at='' WHERE id=?",
                (task_id,),
            )
        if old_status == STATUS_CANCELLED:
            # Restarting a cancelled task: clear the terminal timestamp so the
            # task reads as live again everywhere (UI, health, retention).
            conn.execute(
                "UPDATE kanban_tasks SET completed_at='' WHERE id=?",
                (task_id,),
            )
        conn.execute(
            "UPDATE kanban_tasks SET status=?, updated_at=? WHERE id=?",
            (new_status, now, task_id),
        )
        conn.commit()
        final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
    finally:
        conn.close()
    final = _row_to_dict(final_row) or {}
    _emit_lifecycle_event(
        event_type="status_changed",
        task_id=task_id,
        from_status=old_status,
        to_status=new_status,
        assignee=row["assignee"] or "",
        tenant=final.get("tenant", "default"),
        extra={"reason": reason} if reason else {},
    )
    _bump_search_generation()
    return final


def set_priority(task_id: str, priority: int) -> dict[str, Any]:
    """Operator-only: change a task's priority (1–9). Terminal tasks rejected."""
    priority_v = _validate_priority(priority)
    now = _now_iso()
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT id, status, priority, assignee, tenant FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None:
            conn.rollback()
            raise ValueError(f"task {task_id} not found")
        if row["status"] in (STATUS_DONE, STATUS_CANCELLED):
            conn.rollback()
            raise ValueError(f"task {task_id} is terminal ({row['status']})")
        old_prio = row["priority"]
        conn.execute(
            "UPDATE kanban_tasks SET priority=?, updated_at=? WHERE id=?",
            (priority_v, now, task_id),
        )
        conn.commit()
        final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
    finally:
        conn.close()
    final = _row_to_dict(final_row) or {}
    if old_prio != priority_v:
        _emit_lifecycle_event(
            event_type="priority_changed",
            task_id=task_id,
            from_status=row["status"],
            to_status=row["status"],
            assignee=row["assignee"] or "",
            tenant=final.get("tenant", "default"),
            extra={"old_priority": old_prio, "new_priority": priority_v},
        )
    _bump_search_generation()
    return final


def rebalance_lane_priorities(
    status: str,
    *,
    assignee: str = "",
    tenant: str = "",
) -> int:
    """Redistribute priority values evenly (1–9) for all non-terminal tasks
    in a given lane (status + optional assignee/tenant filter).

    Returns the count of tasks whose priority was updated. This is useful
    after drag-and-drop reordering to renormalize gaps.
    """
    if not status:
        raise ValueError("status required")
    now = _now_iso()
    conn = _open_conn()
    try:
        where = "status=?"
        params: list[Any] = [status]
        if assignee:
            where += " AND assignee=?"
            params.append(assignee)
        if tenant:
            where += " AND tenant=?"
            params.append(tenant)
        rows = conn.execute(
            f"SELECT id, priority FROM kanban_tasks WHERE {where} ORDER BY priority ASC, created_at ASC",
            params,
        ).fetchall()
        if len(rows) <= 1:
            return 0
        count = len(rows)
        updated = 0
        conn.execute("BEGIN IMMEDIATE")
        for idx, row in enumerate(rows):
            new_prio = max(1, min(9, 1 + int(idx * 8 / max(count - 1, 1))))
            if new_prio != row["priority"]:
                conn.execute(
                    "UPDATE kanban_tasks SET priority=?, updated_at=? WHERE id=?",
                    (new_prio, now, row["id"]),
                )
                updated += 1
        conn.commit()
    finally:
        conn.close()
    if updated:
        _bump_search_generation()
    return updated


def set_parent_ids(
    task_id: str,
    parent_ids: list[str],
) -> dict[str, Any]:
    """Operator-only: replace the parent set of a non-terminal task.

    Recalculates status: if all parents are done → ``ready``; if some
    parents are not done → ``todo``; no parents → ``ready``.
    Active/in-progress tasks raise unless they have no open runs.
    """
    if not task_id:
        raise ValueError("task_id required")
    parent_ids_v = _validate_parent_id_list(None, parent_ids)
    now = _now_iso()
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT id, status, assignee, tenant FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None:
            conn.rollback()
            raise ValueError(f"task {task_id} not found")
        if row["status"] in (STATUS_DONE, STATUS_CANCELLED):
            conn.rollback()
            raise ValueError(f"task {task_id} is terminal ({row['status']})")
        if row["status"] == STATUS_IN_PROGRESS:
            conn.rollback()
            raise ValueError("cannot re-parent an in_progress task")
        for pid in parent_ids_v:
            if pid == task_id:
                conn.rollback()
                raise ValueError("a task cannot be its own parent")
            prow = conn.execute(
                "SELECT id FROM kanban_tasks WHERE id=?", (pid,)
            ).fetchone()
            if prow is None:
                conn.rollback()
                raise ValueError(f"parent {pid} not found")

        conn.execute(
            "DELETE FROM kanban_task_parents WHERE task_id=?", (task_id,)
        )
        primary = parent_ids_v[0] if parent_ids_v else None
        now_ts = _now_iso()
        for idx, pid in enumerate(parent_ids_v):
            conn.execute(
                "INSERT INTO kanban_task_parents (task_id, parent_id, ordinal, created_at) "
                "VALUES (?,?,?,?)",
                (task_id, pid, idx, now_ts),
            )
        conn.execute(
            "UPDATE kanban_tasks SET parent_id=?, updated_at=? WHERE id=?",
            (primary, now, task_id),
        )

        new_status = row["status"]
        if parent_ids_v:
            placeholders = ",".join(["?"] * len(parent_ids_v))
            statuses = conn.execute(
                f"SELECT status FROM kanban_tasks WHERE id IN ({placeholders})",
                parent_ids_v,
            ).fetchall()
            if len(statuses) == len(parent_ids_v) and all(
                s[0] == STATUS_DONE for s in statuses
            ):
                new_status = STATUS_READY
            else:
                new_status = STATUS_TODO
        else:
            if row["status"] == STATUS_TODO:
                new_status = STATUS_READY
        if new_status != row["status"]:
            conn.execute(
                "UPDATE kanban_tasks SET status=? WHERE id=?",
                (new_status, task_id),
            )
        conn.commit()
        final_row = conn.execute("SELECT * FROM kanban_tasks WHERE id=?", (task_id,)).fetchone()
    finally:
        conn.close()
    final = _row_to_dict(final_row) or {}
    _emit_lifecycle_event(
        event_type="reparented",
        task_id=task_id,
        from_status=row["status"],
        to_status=final.get("status", row["status"]),
        assignee=row["assignee"] or "",
        tenant=final.get("tenant", "default"),
        extra={"parent_ids": parent_ids_v},
    )
    _bump_search_generation()
    return final


# ---------------------------------------------------------------------------
# Public API — runs (low-level; dispatcher uses these when the spawn path
# is split from the claim).
# ---------------------------------------------------------------------------

def start_run(
    task_id: str,
    *,
    attempt: int,
    assignee: str,
    handle_id: str = "",
    claim_token: str = "",
) -> str:
    """Insert an ``active`` run row. Returns run_id."""
    if not task_id:
        raise ValueError("task_id is required")
    run_id = _new_id("kr")
    now = _now_iso()
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        stale_rows = conn.execute(
            "SELECT id FROM kanban_task_runs WHERE task_id=? AND outcome=?",
            (task_id, OUTCOME_ACTIVE),
        ).fetchall()
        for stale in stale_rows:
            conn.execute(
                "UPDATE kanban_task_runs SET outcome=?, error_ciphertext=?, ended_at=? WHERE id=?",
                (
                    OUTCOME_CRASHED,
                    _encrypt("superseded by new attempt"),
                    now,
                    stale["id"],
                ),
            )
        conn.execute(
            """
            INSERT INTO kanban_task_runs (
                id, task_id, attempt, outcome, assignee, handle_id, claim_token,
                summary_ciphertext, metadata_ciphertext, error_ciphertext,
                started_at, ended_at
            ) VALUES (?, ?, ?, 'active', ?, ?, ?, ?, ?, ?, ?, '')
            """,
            (
                run_id,
                task_id,
                int(attempt),
                assignee,
                handle_id,
                claim_token,
                _encrypt(""),
                _encrypt("{}"),
                _encrypt(""),
                now,
            ),
        )
        conn.commit()
    finally:
        conn.close()
    return run_id


def close_run(
    run_id: str,
    *,
    outcome: str,
    summary: str = "",
    metadata: Any = None,
    error: str = "",
) -> None:
    if outcome not in (OUTCOME_COMPLETED, OUTCOME_BLOCKED, OUTCOME_SPAWN_FAILED,
                       OUTCOME_CRASHED, OUTCOME_TIMED_OUT, OUTCOME_GAVE_UP, OUTCOME_CANCELLED):
        raise ValueError(f"invalid outcome: {outcome}")
    summary_v = _validate_text("summary", summary or "", max_chars=MAX_SUMMARY_CHARS, allow_empty=True)
    metadata_v = _validate_metadata(metadata or {})
    error_v = _validate_text("error", error or "", max_chars=MAX_ERROR_CHARS, allow_empty=True)
    now = _now_iso()
    conn = _open_conn()
    try:
        run_row = conn.execute(
            "SELECT metadata_ciphertext FROM kanban_task_runs WHERE id=?",
            (run_id,),
        ).fetchone()
        existing_md = _decode_run_metadata_blob(
            run_row["metadata_ciphertext"] if run_row is not None else None
        )
        final_md = _merge_terminal_run_metadata(existing_md, metadata_v)
        conn.execute(
            "UPDATE kanban_task_runs SET outcome=?, summary_ciphertext=?, "
            "metadata_ciphertext=?, error_ciphertext=?, ended_at=? WHERE id=?",
            (
                outcome,
                _encrypt(summary_v),
                _encrypt(json.dumps(final_md, ensure_ascii=False)),
                _encrypt(error_v),
                now,
                run_id,
            ),
        )
        conn.commit()
    finally:
        conn.close()


def list_runs(task_id: str, *, limit: int = 50) -> list[dict[str, Any]]:
    if not task_id:
        return []
    conn = _open_conn()
    try:
        rows = conn.execute(
            "SELECT * FROM kanban_task_runs WHERE task_id=? "
            "ORDER BY started_at DESC LIMIT ?",
            (task_id, max(1, min(int(limit or 50), 200))),
        ).fetchall()
    finally:
        conn.close()
    return [r for r in (_row_to_dict(row) for row in rows) if r is not None]


# ---------------------------------------------------------------------------
# Worker progress channel (claim-token-scoped)
# ---------------------------------------------------------------------------

def _sanitize_progress_entry(entry: dict[str, Any]) -> dict[str, Any]:
    """Redact and cap structured progress-log fields before persistence."""
    from utils.redaction import cap_str

    if not isinstance(entry, dict):
        return {"ts": _now_iso(), "note": "", "phase": "status"}
    out: dict[str, Any] = {"ts": cap_str(entry.get("ts") or _now_iso(), 32)}
    for key, max_len in (
        ("call_id", MAX_PROGRESS_CALL_ID_CHARS),
        ("phase", 32),
        ("tool", 128),
        ("note", MAX_PROGRESS_NOTE_CHARS),
        ("url", 512),
        ("args_preview", MAX_PROGRESS_ARGS_CHARS),
        ("result_preview", MAX_PROGRESS_RESULT_CHARS),
        ("preview_thumb", MAX_PROGRESS_THUMB_B64_CHARS),
    ):
        raw = entry.get(key)
        if raw is None:
            continue
        text = cap_str(str(raw), max_len)
        if text:
            out[key] = text
    paths_raw = entry.get("paths")
    if isinstance(paths_raw, list):
        paths = [cap_str(str(p), 512) for p in paths_raw[:8] if p]
        paths = [p for p in paths if p]
        if paths:
            out["paths"] = paths
    if not out.get("note") and not out.get("phase"):
        out["phase"] = "status"
    return out


def _mutate_active_run_progress_log(
    task_id: str,
    *,
    expected_token: str,
    mutator,
    fail_open: bool = False,
) -> tuple[bool, str, int]:
    """Apply *mutator* to the active run's ``progress_log``.

    *mutator* receives ``(log_arr: list)`` and may mutate in place.
    Returns ``(ok, run_id, log_len)``. When *fail_open* is True, token
    mismatches return ``(False, "", 0)`` instead of raising.
    """
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        task_row = conn.execute(
            "SELECT claim_token, status FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if task_row is None:
            conn.rollback()
            if fail_open:
                return False, "", 0
            raise ValueError(f"task {task_id} not found")
        if task_row["claim_token"] != expected_token or task_row["status"] != STATUS_IN_PROGRESS:
            conn.rollback()
            if fail_open:
                return False, "", 0
            raise PermissionError("claim_token mismatch or task not in_progress")
        run_row = conn.execute(
            "SELECT id, metadata_ciphertext FROM kanban_task_runs "
            "WHERE task_id=? AND outcome=? ORDER BY started_at DESC LIMIT 1",
            (task_id, OUTCOME_ACTIVE),
        ).fetchone()
        if run_row is None:
            conn.rollback()
            return False, "", 0
        try:
            existing = json.loads(_decrypt(run_row["metadata_ciphertext"]) or "{}")
        except Exception:
            conn.rollback()
            _log.warning(
                "Refusing progress_log mutate: run metadata decrypt failed task_id=%s run_id=%s",
                task_id,
                run_row["id"],
                exc_info=True,
            )
            return False, "", 0
        if not isinstance(existing, dict):
            existing = {}
        log_arr = existing.get("progress_log")
        if not isinstance(log_arr, list):
            log_arr = []
        mutator(log_arr)
        if len(log_arr) > MAX_PROGRESS_LOG_ENTRIES:
            log_arr[:] = log_arr[-MAX_PROGRESS_LOG_ENTRIES:]
        existing["progress_log"] = log_arr
        conn.execute(
            "UPDATE kanban_task_runs SET metadata_ciphertext=? WHERE id=?",
            (_encrypt(json.dumps(existing, ensure_ascii=False)), run_row["id"]),
        )
        conn.commit()
        return True, str(run_row["id"]), len(log_arr)
    except PermissionError:
        raise
    except ValueError:
        raise
    except Exception:
        _log.debug("mutate_active_run_progress_log failed", exc_info=True)
        try:
            conn.rollback()
        except Exception:
            pass
        return False, "", 0
    finally:
        conn.close()


def append_progress_entry(
    task_id: str,
    entry: dict[str, Any],
    *,
    expected_token: str,
) -> tuple[bool, str, int]:
    """Append a structured progress entry to the active run (fail-open).

    Returns ``(ok, run_id, log_len)``.
    """
    try:
        clean = _sanitize_progress_entry(entry if isinstance(entry, dict) else {})

        def _append(log_arr: list) -> None:
            log_arr.append(clean)

        return _mutate_active_run_progress_log(
            task_id,
            expected_token=expected_token,
            mutator=_append,
            fail_open=True,
        )
    except Exception:
        _log.debug("append_progress_entry failed", exc_info=True)
        return False, "", 0


def upsert_progress_entry(
    task_id: str,
    call_id: str,
    patch: dict[str, Any],
    *,
    expected_token: str,
) -> tuple[bool, str, int]:
    """Merge *patch* into the progress row matching *call_id*, or append.

    Returns ``(ok, run_id, log_len)``.
    """
    cid = (call_id or "").strip()
    if not cid:
        ok, run_id, ln = append_progress_entry(task_id, patch, expected_token=expected_token)
        return ok, run_id, ln
    try:
        clean_patch = _sanitize_progress_entry({**patch, "call_id": cid})

        def _upsert(log_arr: list) -> None:
            for idx in range(len(log_arr) - 1, -1, -1):
                row = log_arr[idx]
                if isinstance(row, dict) and str(row.get("call_id") or "") == cid:
                    merged = dict(row)
                    merged.update(clean_patch)
                    if not merged.get("ts"):
                        merged["ts"] = row.get("ts") or _now_iso()
                    log_arr[idx] = merged
                    return
            if not clean_patch.get("ts"):
                clean_patch["ts"] = _now_iso()
            log_arr.append(clean_patch)

        return _mutate_active_run_progress_log(
            task_id,
            expected_token=expected_token,
            mutator=_upsert,
            fail_open=True,
        )
    except Exception:
        _log.debug("upsert_progress_entry failed", exc_info=True)
        return False, "", 0


def patch_active_run_metadata(
    task_id: str,
    *,
    expected_token: str,
    updates: dict[str, Any],
    fail_open: bool = True,
) -> bool:
    """Merge *updates* into the active run metadata dict (best-effort)."""
    if not isinstance(updates, dict) or not updates:
        return False

    def _patch(existing: dict) -> None:
        for key, val in updates.items():
            if key and val is not None:
                existing[str(key)] = val

    try:
        ok, _, _ = _mutate_active_run_metadata(
            task_id,
            expected_token=expected_token,
            mutator=_patch,
            fail_open=fail_open,
        )
        return ok
    except Exception:
        _log.debug("patch_active_run_metadata failed", exc_info=True)
        return False


def _mutate_active_run_metadata(
    task_id: str,
    *,
    expected_token: str,
    mutator,
    fail_open: bool = False,
) -> tuple[bool, str, int]:
    """Apply *mutator* to the active run metadata dict (not only progress_log)."""
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        task_row = conn.execute(
            "SELECT claim_token, status FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if task_row is None:
            conn.rollback()
            if fail_open:
                return False, "", 0
            raise ValueError(f"task {task_id} not found")
        if task_row["claim_token"] != expected_token or task_row["status"] != STATUS_IN_PROGRESS:
            conn.rollback()
            if fail_open:
                return False, "", 0
            raise PermissionError("claim_token mismatch or task not in_progress")
        run_row = conn.execute(
            "SELECT id, metadata_ciphertext FROM kanban_task_runs "
            "WHERE task_id=? AND outcome=? ORDER BY started_at DESC LIMIT 1",
            (task_id, OUTCOME_ACTIVE),
        ).fetchone()
        if run_row is None:
            conn.rollback()
            return False, "", 0
        try:
            existing = json.loads(_decrypt(run_row["metadata_ciphertext"]) or "{}")
        except Exception:
            conn.rollback()
            _log.warning(
                "Refusing run metadata mutate: decrypt failed task_id=%s run_id=%s",
                task_id,
                run_row["id"],
                exc_info=True,
            )
            return False, "", 0
        if not isinstance(existing, dict):
            existing = {}
        mutator(existing)
        conn.execute(
            "UPDATE kanban_task_runs SET metadata_ciphertext=? WHERE id=?",
            (_encrypt(json.dumps(existing, ensure_ascii=False)), run_row["id"]),
        )
        conn.commit()
        return True, str(run_row["id"]), 0
    except PermissionError:
        raise
    except ValueError:
        raise
    except Exception:
        _log.debug("_mutate_active_run_metadata failed", exc_info=True)
        try:
            conn.rollback()
        except Exception:
            pass
        return False, "", 0
    finally:
        conn.close()


def append_progress_note(
    task_id: str,
    note: str,
    *,
    expected_token: str,
) -> bool:
    """Append a progress note to the open run's metadata under the
    'progress_log' key. Used by the worker-tier ``kanban_post_progress``.

    Returns True on success. Raises PermissionError on token mismatch (the
    tool layer surfaces this to the worker).
    """
    note_v = _validate_text("note", note or "", max_chars=MAX_PROGRESS_NOTE_CHARS, allow_empty=False)
    try:
        ok, _, _ = _mutate_active_run_progress_log(
            task_id,
            expected_token=expected_token,
            mutator=lambda log_arr: log_arr.append({"ts": _now_iso(), "note": note_v, "phase": "status"}),
            fail_open=False,
        )
        return ok
    except (PermissionError, ValueError):
        raise
    except Exception:
        _log.debug("append_progress_note failed", exc_info=True)
        return False


# ---------------------------------------------------------------------------
# Promotion engine
# ---------------------------------------------------------------------------

def _promote_children_locked(conn: sqlite3.Connection, parent_id: str) -> list[str]:
    """Internal helper: callable inside an open transaction. Returns the list
    of child ids that were promoted from ``todo`` to ``ready`` because
    *parent_id* just transitioned to ``done``.

    Multi-parent rule: a child becomes ready only when **all** of its
    declared parents are ``done``. Children with any non-done parent stay
    in ``todo``.
    """
    if not parent_id:
        return []
    # Pull every todo child of this parent — both via the join table and
    # the legacy single-parent column (in case of a partial migration).
    candidate_ids: list[str] = []
    seen: set[str] = set()
    for row in conn.execute(
        "SELECT t.id FROM kanban_tasks t "
        "JOIN kanban_task_parents p ON p.task_id = t.id "
        "WHERE p.parent_id=? AND t.status='todo'",
        (parent_id,),
    ).fetchall():
        cid = row[0]
        if cid and cid not in seen:
            seen.add(cid)
            candidate_ids.append(cid)
    for row in conn.execute(
        "SELECT id FROM kanban_tasks WHERE parent_id=? AND status='todo'",
        (parent_id,),
    ).fetchall():
        cid = row[0]
        if cid and cid not in seen:
            seen.add(cid)
            candidate_ids.append(cid)
    if not candidate_ids:
        return []

    promoted: list[str] = []
    now_iso = _now_iso()
    for cid in candidate_ids:
        # All parents must be done (none in todo/ready/in_progress/blocked).
        # Look in the join table first; fall back to the single-parent
        # column when the join is empty (pre-migration case).
        join_rows = conn.execute(
            "SELECT pt.status FROM kanban_task_parents kp "
            "JOIN kanban_tasks pt ON pt.id = kp.parent_id "
            "WHERE kp.task_id=?",
            (cid,),
        ).fetchall()
        if join_rows:
            all_parents_done = all(r[0] == STATUS_DONE for r in join_rows)
        else:
            legacy = conn.execute(
                "SELECT pt.status FROM kanban_tasks t "
                "LEFT JOIN kanban_tasks pt ON pt.id = t.parent_id "
                "WHERE t.id=?",
                (cid,),
            ).fetchone()
            if legacy is None or legacy[0] is None:
                # Orphan parent reference; treat as ready (no blocking parent).
                all_parents_done = True
            else:
                all_parents_done = (legacy[0] == STATUS_DONE)
        if not all_parents_done:
            continue
        cur = conn.execute(
            "UPDATE kanban_tasks SET status='ready', updated_at=? WHERE id=? AND status='todo'",
            (now_iso, cid),
        )
        if cur.rowcount == 1:
            promoted.append(cid)
    return promoted


def promote_ready_children(parent_id: str) -> list[str]:
    """Public wrapper for the dispatcher. Opens its own transaction."""
    if not parent_id:
        return []
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        promoted = _promote_children_locked(conn, parent_id)
        conn.commit()
    finally:
        conn.close()
    for cid in promoted:
        _emit_lifecycle_event(
            event_type="promoted",
            task_id=cid,
            from_status=STATUS_TODO,
            to_status=STATUS_READY,
            assignee="",
            tenant="default",
        )
    return promoted


# ---------------------------------------------------------------------------
# Stale claims / overdue tasks / shutdown helpers
# ---------------------------------------------------------------------------

def find_stale_claims(*, now: str | None = None, limit: int = 50) -> list[dict[str, Any]]:
    now_iso = now or _now_iso()
    conn = _open_conn()
    try:
        rows = conn.execute(
            "SELECT * FROM kanban_tasks WHERE status='in_progress' AND claim_expires_at != '' "
            "AND claim_expires_at < ? ORDER BY claim_expires_at ASC LIMIT ?",
            (now_iso, max(1, min(int(limit), 500))),
        ).fetchall()
    finally:
        conn.close()
    return [r for r in (_row_to_dict(row) for row in rows) if r is not None]


def find_overdue_tasks(*, now: str | None = None, limit: int = 50) -> list[dict[str, Any]]:
    now_iso = now or _now_iso()
    conn = _open_conn()
    try:
        rows = conn.execute(
            "SELECT * FROM kanban_tasks WHERE deadline_at != '' AND deadline_at < ? "
            "AND status NOT IN ('done','cancelled') ORDER BY deadline_at ASC LIMIT ?",
            (now_iso, max(1, min(int(limit), 500))),
        ).fetchall()
    finally:
        conn.close()
    return [r for r in (_row_to_dict(row) for row in rows) if r is not None]


def record_shutdown_requeue(
    task_id: str,
    *,
    claim_token: str,
    error: str = "dispatcher shutdown; run cancelled",
) -> bool:
    """Worker-thread shutdown path: close the active run as ``cancelled`` and
    return the task to ``ready`` — WITHOUT bumping ``failure_count`` and
    WITHOUT terminally cancelling the task.

    June 2026 incident: on app quit, worker threads raced
    :func:`mark_inflight_cancelled`. When the worker won, it recorded
    ``record_failure(outcome=cancelled)`` which terminally cancelled the task
    (and bumped the failure count), so a plain app restart stranded healthy
    tasks in the Cancelled column with no operator path back. A dispatcher
    shutdown is an operational event, not a verdict on the task.

    Tolerant by design: returns False when the task is no longer
    ``in_progress`` or the token does not match (``mark_inflight_cancelled``
    or an operator cancel already handled it).
    """
    if not task_id or not claim_token:
        return False
    now = _now_iso()
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute(
            "SELECT id, status, claim_token, assignee, tenant, last_attempt "
            "FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None or row["status"] != STATUS_IN_PROGRESS or row["claim_token"] != claim_token:
            conn.rollback()
            return False
        conn.execute(
            "UPDATE kanban_task_runs SET outcome='cancelled', error_ciphertext=?, ended_at=? "
            "WHERE task_id=? AND outcome='active'",
            (_encrypt(error), now, task_id),
        )
        conn.execute(
            """
            UPDATE kanban_tasks
               SET status='ready',
                   claim_handle='', claim_token='', claim_expires_at='',
                   updated_at=?
             WHERE id=?
            """,
            (now, task_id),
        )
        conn.commit()
    finally:
        conn.close()
    _emit_lifecycle_event(
        event_type="released",
        task_id=task_id,
        from_status=STATUS_IN_PROGRESS,
        to_status=STATUS_READY,
        assignee=row["assignee"] or "",
        tenant=row["tenant"] or "default",
        attempt=int(row["last_attempt"] or 0),
        extra={"reason": "shutdown_requeue"},
    )
    _bump_search_generation()
    return True


def mark_inflight_cancelled() -> int:
    """Shutdown helper: open runs -> cancelled, in_progress tasks -> ready.
    Returns the number of tasks reset. Called from
    :func:`mugi_shutdown_cleanup`."""
    now = _now_iso()
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        cur_runs = conn.execute(
            "UPDATE kanban_task_runs SET outcome='cancelled', error_ciphertext=?, ended_at=? WHERE outcome='active'",
            (_encrypt("dispatcher shutdown; run cancelled"), now),
        )
        cur_tasks = conn.execute(
            """
            UPDATE kanban_tasks
               SET status='ready',
                   claim_handle='', claim_token='', claim_expires_at='',
                   updated_at=?
             WHERE status='in_progress'
            """,
            (now,),
        )
        conn.commit()
        n = cur_tasks.rowcount or 0
        _ = cur_runs.rowcount
    finally:
        conn.close()
    return n


# ---------------------------------------------------------------------------
# Retention
# ---------------------------------------------------------------------------

def purge_terminal(retention_days: int) -> int:
    """Delete done/cancelled tasks (and their runs) older than
    *retention_days*. Returns the number of tasks deleted."""
    if retention_days <= 0:
        return 0
    cutoff = (
        datetime.now(tz=timezone.utc) - timedelta(days=int(retention_days))
    ).strftime("%Y-%m-%dT%H:%M:%SZ")
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        rows = conn.execute(
            "SELECT id FROM kanban_tasks WHERE status IN ('done','cancelled') "
            "AND completed_at != '' AND completed_at < ?",
            (cutoff,),
        ).fetchall()
        ids = [r[0] for r in rows]
        if ids:
            placeholders = ",".join(["?"] * len(ids))
            conn.execute(
                f"DELETE FROM kanban_task_runs WHERE task_id IN ({placeholders})", ids
            )
            # Mirror delete_task: drop DAG edges, subscriptions, and legacy
            # parent pointers so retention cannot leave orphan parent links.
            # (An orphan edge silently disappears from the AND-promotion JOIN,
            # which would let a child promote as if its purged parent's
            # precondition were satisfied.)
            conn.execute(
                f"DELETE FROM kanban_task_parents "
                f"WHERE task_id IN ({placeholders}) OR parent_id IN ({placeholders})",
                [*ids, *ids],
            )
            conn.execute(
                f"UPDATE kanban_tasks SET parent_id=NULL "
                f"WHERE parent_id IN ({placeholders})",
                ids,
            )
            try:
                conn.execute(
                    f"DELETE FROM kanban_subscriptions WHERE task_id IN ({placeholders})",
                    ids,
                )
            except sqlite3.OperationalError:
                pass  # subscriptions table may not exist on older schemas
            conn.execute(
                f"DELETE FROM kanban_tasks WHERE id IN ({placeholders})", ids
            )
        conn.commit()
    finally:
        conn.close()
    return len(ids)


def delete_task(task_id: str) -> bool:
    """Permanently delete a single **terminal** task and its runs, parent
    edges, and subscription rows.

    Returns ``True`` on success.  Raises ``ValueError`` when the task does
    not exist or is not in a terminal status (``done`` / ``cancelled``).
    """
    if not task_id:
        raise ValueError("task_id required")
    conn = _open_conn()
    try:
        row = conn.execute(
            "SELECT status, assignee, tenant FROM kanban_tasks WHERE id=?",
            (task_id,),
        ).fetchone()
        if row is None:
            raise ValueError("task_not_found")
        status = row[0]
        if status not in (STATUS_DONE, STATUS_CANCELLED):
            raise ValueError(
                f"only done/cancelled tasks can be deleted (current: {status})"
            )
        assignee = row[1] or ""
        tenant = row[2] or "default"
        conn.execute("BEGIN IMMEDIATE")
        conn.execute(
            "DELETE FROM kanban_task_runs WHERE task_id=?", (task_id,)
        )
        conn.execute(
            "DELETE FROM kanban_task_parents WHERE task_id=? OR parent_id=?",
            (task_id, task_id),
        )
        conn.execute(
            "UPDATE kanban_tasks SET parent_id=NULL WHERE parent_id=?",
            (task_id,),
        )
        try:
            conn.execute(
                "DELETE FROM kanban_subscriptions WHERE task_id=?", (task_id,)
            )
        except sqlite3.OperationalError:
            pass
        conn.execute("DELETE FROM kanban_tasks WHERE id=?", (task_id,))
        conn.commit()
    finally:
        conn.close()
    _emit_lifecycle_event(
        event_type="deleted",
        task_id=task_id,
        from_status=status,
        to_status="",
        assignee=assignee,
        tenant=tenant,
    )
    _bump_search_generation()
    return True


def clear_terminal_tasks(
    status: str,
    *,
    tenant: str = "",
    assignee: str = "",
) -> int:
    """Bulk-delete all tasks with the given terminal *status*.

    Optional *tenant* / *assignee* filters narrow the set.  Returns the
    count of deleted tasks.  Raises ``ValueError`` if *status* is not
    terminal.  Emits a single ``bulk_cleared`` lifecycle event (not
    per-task) to avoid SSE storms.
    """
    if status not in (STATUS_DONE, STATUS_CANCELLED):
        raise ValueError(f"only done/cancelled statuses accepted (got: {status})")
    conn = _open_conn()
    try:
        where = "status=?"
        params: list[str] = [status]
        if tenant:
            where += " AND tenant=?"
            params.append(tenant)
        if assignee:
            where += " AND assignee=?"
            params.append(assignee)
        rows = conn.execute(
            f"SELECT id FROM kanban_tasks WHERE {where}", params
        ).fetchall()
        ids = [r[0] for r in rows]
        if ids:
            conn.execute("BEGIN IMMEDIATE")
            placeholders = ",".join(["?"] * len(ids))
            conn.execute(
                f"DELETE FROM kanban_task_runs WHERE task_id IN ({placeholders})",
                ids,
            )
            conn.execute(
                f"DELETE FROM kanban_task_parents WHERE task_id IN ({placeholders})"
                f" OR parent_id IN ({placeholders})",
                ids + ids,
            )
            conn.execute(
                f"UPDATE kanban_tasks SET parent_id=NULL "
                f"WHERE parent_id IN ({placeholders})",
                ids,
            )
            try:
                conn.execute(
                    f"DELETE FROM kanban_subscriptions WHERE task_id IN ({placeholders})",
                    ids,
                )
            except sqlite3.OperationalError:
                pass
            conn.execute(
                f"DELETE FROM kanban_tasks WHERE id IN ({placeholders})", ids
            )
            conn.commit()
    finally:
        conn.close()
    if ids:
        _emit_lifecycle_event(
            event_type="bulk_cleared",
            task_id="",
            from_status=status,
            to_status="",
            assignee=assignee,
            tenant=tenant or "default",
            extra={"deleted_count": len(ids)},
        )
    _bump_search_generation()
    return len(ids)


def trim_runs(max_runs_per_task: int) -> int:
    """For every task with > max_runs_per_task run rows, keep the first run
    (attempt 1) plus the most recent N-1; delete the rest. Returns total
    rows deleted."""
    if max_runs_per_task <= 0:
        return 0
    deleted = 0
    conn = _open_conn()
    try:
        conn.execute("BEGIN IMMEDIATE")
        over_ids = [
            r[0]
            for r in conn.execute(
                "SELECT task_id FROM kanban_task_runs GROUP BY task_id "
                "HAVING COUNT(*) > ?",
                (int(max_runs_per_task),),
            ).fetchall()
        ]
        for tid in over_ids:
            run_rows = conn.execute(
                "SELECT id, attempt, started_at FROM kanban_task_runs "
                "WHERE task_id=? ORDER BY started_at ASC",
                (tid,),
            ).fetchall()
            if len(run_rows) <= max_runs_per_task:
                continue
            keep_ids: set[str] = set()
            keep_ids.add(run_rows[0]["id"])  # first attempt
            for r in run_rows[-(max_runs_per_task - 1):]:
                keep_ids.add(r["id"])
            to_delete = [r["id"] for r in run_rows if r["id"] not in keep_ids]
            if to_delete:
                placeholders = ",".join(["?"] * len(to_delete))
                cur = conn.execute(
                    f"DELETE FROM kanban_task_runs WHERE id IN ({placeholders})",
                    to_delete,
                )
                deleted += cur.rowcount or 0
        conn.commit()
    finally:
        conn.close()
    return deleted


# ---------------------------------------------------------------------------
# Health summary
# ---------------------------------------------------------------------------

# Outcomes that represent a worker handoff attempt — where harness salvage/synthesis can
# occur. Operator cancels and spawn failures never produced a handoff, so they are excluded
# from the salvage-rate denominator (they would otherwise dilute the metric).
_HANDOFF_OUTCOMES = frozenset({
    OUTCOME_COMPLETED, OUTCOME_BLOCKED, OUTCOME_CRASHED, OUTCOME_GAVE_UP, OUTCOME_TIMED_OUT,
})

# Grain-reframe Stage B: handoff_source values that denote a FIRST-CLASS native-tool completion
# (a deliberate kanban_post_summary submission), not a harness salvage. These are surfaced in
# by_source for visibility but excluded from the salvaged count / salvage_rate — the whole point
# of the reframe is that a native-tool handoff is the worker completing cleanly, not the harness
# compensating. New first-class sources must be added here or they will inflate salvage_rate.
_FIRST_CLASS_HANDOFF_SOURCES = frozenset({"tool_submit"})


def is_first_class_handoff_source(source: str | None) -> bool:
    """True when ``handoff_source`` denotes a first-class native-tool completion, not salvage.

    Single source of truth shared by ``salvage_stats`` (which keeps these out of the salvaged
    count / salvage_rate) and the dispatcher run-row builder (which must NOT mark these as
    ``harness_synthesized``). A first-class source means the *worker* completed cleanly via a
    deliberate native tool submission — the harness did not lean on a fallback.
    """
    return isinstance(source, str) and source.strip() in _FIRST_CLASS_HANDOFF_SOURCES


def salvage_stats(*, window: int = 200) -> dict[str, Any]:
    """Aggregate harness-salvage provenance over the most recent terminal worker runs.

    Read-only telemetry that turns "salvage is a backup" into a measured number. A run counts
    as *salvaged* when its persisted handoff metadata carries ``handoff_source`` (stamped by the
    synthesis / in-band paths) or the legacy ``completion_salvage`` flag — i.e. the harness
    assembled the handoff rather than the worker emitting a clean one.

    Returns a stable-schema dict::

        {"window", "terminal_runs", "salvaged_runs", "salvage_rate", "by_source"}

    where ``salvage_rate`` is ``salvaged_runs / terminal_runs`` (0.0 when there are no terminal
    runs). Never raises — on query failure it returns a zeroed report so health/doctor callers
    stay alive (this is informational telemetry, not a terminal-state path).
    """
    window = max(1, min(int(window or 200), 2000))
    placeholders = ",".join("?" for _ in _HANDOFF_OUTCOMES)
    outcomes = sorted(_HANDOFF_OUTCOMES)
    try:
        conn = _open_conn()
        try:
            rows = conn.execute(
                f"SELECT * FROM kanban_task_runs WHERE outcome IN ({placeholders}) "
                "ORDER BY started_at DESC LIMIT ?",
                (*outcomes, window),
            ).fetchall()
        finally:
            conn.close()
    except Exception:
        _log.warning("salvage_stats query failed", exc_info=True)
        return {
            "window": window,
            "terminal_runs": 0,
            "salvaged_runs": 0,
            "salvage_rate": 0.0,
            "by_source": {},
            "salvaged_with_repair": 0,
            "salvaged_without_repair": 0,
        }
    total = 0
    salvaged = 0
    by_source: dict[str, int] = {}
    # Phase 2c: among salvaged runs, partition by whether the run carried evidence that the
    # harness attempted an in-context repair turn before synthesizing (metadata.repair_attempts
    # > 0). salvaged_without_repair is the class to watch — either no repair was attempted, or
    # the salvage happened dispatcher-side (in-band) where the worker's ledger was unavailable to
    # stamp the count. It is a lower bound on "synthesized after repair", not a proof of absence.
    salvaged_with_repair = 0
    salvaged_without_repair = 0
    for row in rows:
        rec = _row_to_dict(row)
        if rec is None:
            continue
        total += 1
        md = rec.get("metadata") if isinstance(rec.get("metadata"), dict) else {}
        source = md.get("handoff_source")
        if not (isinstance(source, str) and source.strip()):
            source = "council_researcher_salvage" if md.get("completion_salvage") is True else ""
        source = source.strip() if isinstance(source, str) else ""
        if source:
            by_source[source] = by_source.get(source, 0) + 1
            # First-class native-tool handoffs are completions, not salvage: record provenance
            # in by_source but keep them out of the salvaged count and the repair partition.
            if is_first_class_handoff_source(source):
                continue
            salvaged += 1
            try:
                attempts = int(md.get("repair_attempts") or 0)
            except (TypeError, ValueError):
                attempts = 0
            if attempts > 0:
                salvaged_with_repair += 1
            else:
                salvaged_without_repair += 1
    rate = (salvaged / total) if total else 0.0
    return {
        "window": window,
        "terminal_runs": total,
        "salvaged_runs": salvaged,
        "salvage_rate": round(rate, 4),
        "by_source": by_source,
        "salvaged_with_repair": salvaged_with_repair,
        "salvaged_without_repair": salvaged_without_repair,
    }


def health_summary() -> dict[str, Any]:
    conn = _open_conn()
    try:
        counts: dict[str, int] = {}
        for status in ALL_STATUSES:
            row = conn.execute(
                "SELECT COUNT(*) FROM kanban_tasks WHERE status=?", (status,)
            ).fetchone()
            counts[status] = int(row[0])
        in_flight_by_profile: dict[str, int] = {}
        for r in conn.execute(
            "SELECT assignee, COUNT(*) FROM kanban_tasks WHERE status='in_progress' "
            "GROUP BY assignee"
        ).fetchall():
            in_flight_by_profile[r[0]] = int(r[1])
        oldest_pending = conn.execute(
            "SELECT MIN(created_at) FROM kanban_tasks WHERE status IN ('todo','ready')"
        ).fetchone()[0]
        gave_up_count = conn.execute(
            "SELECT COUNT(*) FROM kanban_task_runs WHERE outcome='gave_up'"
        ).fetchone()[0]
    finally:
        conn.close()
    now = datetime.now(tz=timezone.utc)
    oldest_age = 0
    if oldest_pending:
        try:
            then = datetime.strptime(oldest_pending, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
            oldest_age = max(0, int((now - then).total_seconds()))
        except Exception:
            oldest_age = 0
    return {
        "queue_depth": counts.get(STATUS_TODO, 0) + counts.get(STATUS_READY, 0),
        "oldest_pending_age_s": oldest_age,
        "in_flight_by_profile": in_flight_by_profile,
        "in_flight_total": counts.get(STATUS_IN_PROGRESS, 0),
        "blocked_count": counts.get(STATUS_BLOCKED, 0),
        "gave_up_count": int(gave_up_count),
        "counts": counts,
        "salvage": salvage_stats(),
    }


# ---------------------------------------------------------------------------
# Internal: emit lifecycle event (activity bus + process-wide event sink)
# ---------------------------------------------------------------------------

def _emit_lifecycle_event(
    *,
    event_type: str,
    task_id: str,
    from_status: str,
    to_status: str,
    assignee: str,
    tenant: str,
    run_id: str = "",
    attempt: int | None = None,
    extra: dict[str, Any] | None = None,
) -> None:
    """Emit the canonical kanban event to both the activity bus (for sidebar
    rendering) and the process-wide event sink (for SSE multiplexing).

    Payload is ciphertext-free by contract.
    """
    payload = {
        "type": "kanban_event",
        "kanban_event": event_type,
        "task_id": task_id,
        "from_status": from_status,
        "to_status": to_status,
        "assignee": assignee,
        "tenant": tenant,
    }
    if run_id:
        payload["run_id"] = run_id
    if attempt is not None:
        payload["attempt"] = int(attempt)
    if extra:
        for k, v in extra.items():
            if k in payload:
                continue
            payload[k] = v
    origin_tid = ""
    try:
        row = get_task(task_id)
        if row:
            origin_tid = (row.get("origin_thread_id") or "").strip()
    except Exception:
        _log.debug("origin_thread_id lookup for kanban event failed", exc_info=True)
    if origin_tid:
        payload["origin_thread_id"] = origin_tid
    emit_kanban_event_activity(
        event_type=event_type,
        task_id=task_id,
        from_status=from_status,
        to_status=to_status,
        assignee=assignee,
        tenant=tenant,
        run_id=run_id,
        attempt=attempt,
        origin_thread_id=origin_tid,
        extra_meta=extra,
    )
    notify_kanban_event(payload, thread_id=origin_tid or None)

    if event_type in {"completed", "gave_up"}:
        try:
            from utils.kanban_notifications import maybe_dispatch_lifecycle
            maybe_dispatch_lifecycle(
                event_type=event_type,
                task_id=task_id,
                assignee=assignee,
                tenant=tenant,
                run_id=run_id or "",
                extra=extra,
            )
        except Exception:
            _log.debug("kanban_notifications dispatch failed", exc_info=True)

    if event_type in {"completed", "gave_up", "cancelled"}:
        try:
            from utils.kanban_observation_hooks import maybe_reconcile_observation_drive_on_lifecycle

            maybe_reconcile_observation_drive_on_lifecycle(
                event_type=event_type,
                task_id=task_id,
                run_id=run_id or "",
            )
        except Exception:
            _log.debug("kanban observation drive reconcile failed", exc_info=True)

    if event_type in {"completed", "gave_up", "cancelled", "failed", "blocked"}:
        try:
            from utils.kanban_coordinator import refresh_coordinator_state_for_task

            refresh_coordinator_state_for_task(task_id)
        except Exception:
            _log.debug("coordinator refresh failed", exc_info=True)

    if event_type == "blocked":
        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 on blocked failed", exc_info=True)

    if event_type in {
        "completed",
        "blocked",
        "claimed",
        "released",
        "failed",
        "gave_up",
        "cancelled",
        "unblocked",
        "promoted",
        "status_changed",
    }:
        try:
            from utils.task_plan_kanban_sync import maybe_sync_task_plan_for_task

            maybe_sync_task_plan_for_task(task_id)
        except Exception:
            _log.debug("task plan kanban sync failed", exc_info=True)

    if event_type == "bulk_cleared":
        try:
            from chat_storage import list_coordinator_thread_ids
            from utils.kanban_coordinator import board_is_active

            for tid in list_coordinator_thread_ids():
                board_is_active(tid)
        except Exception:
            _log.debug("coordinator refresh after bulk_cleared failed", exc_info=True)
