"""LangChain tools for the kanban subsystem.

Two tiers of tools live here, with very different security postures:

  Planner-tier (registered in TIER_0 alongside `update_task_plan`):
    - ``kanban_create_task`` / ``kanban_create_batch`` — drop work onto the
      board.
    - ``kanban_list_tasks`` / ``kanban_show_task`` / ``kanban_list_runs`` —
      read paths.
    - ``kanban_cancel_task`` — terminate a task you own.

  Worker-tier (claim-scoped; only meaningful inside a kanban worker
  thread — :func:`utils.kanban_dispatcher.current_claim_token` must
  return a matching ``(task_id, claim_token)`` tuple):
    - ``kanban_post_summary`` — early hand-off summary write.
    - ``kanban_post_progress`` — appends a structured note to the active
      run's metadata for live progress.

Operator-only verbs (``block``, ``unblock``, ``set_assignee``,
``dispatcher_nudge``, ``dispatcher_health``) are deliberately **not**
registered as agent tools. They live on the HTTP / MCP surfaces with the
auth-token gate; see ``app_http_routes.py`` and ``mcp_server.py``.

The threat model is documented in ``docs/harness_threat_model.md`` —
search for the "Kanban" section.
"""

from __future__ import annotations

import json as _json
import logging
from typing import Any

from langchain_core.tools import tool
from pydantic import BaseModel, Field, field_validator

import kanban_storage as ks
from utils.kanban_dispatcher import current_claim_token, nudge_dispatcher

_log = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _origin_thread_id() -> str:
    try:
        from persistence import get_current_agent_thread_id
        return (get_current_agent_thread_id() or "").strip()
    except Exception:
        return ""


def _reject_unknown_assignee(assignee: str) -> str | None:
    """Reject personality names and other non-profile assignees at create time."""
    a = (assignee or "general").strip()
    if not a or a == "triage":
        return None
    if a.startswith("mesh:"):
        return None
    try:
        from subagent_runner import SUBAGENT_PROFILES
    except ImportError:
        SUBAGENT_PROFILES = {}
    if a in SUBAGENT_PROFILES:
        return None
    valid = ", ".join(sorted(SUBAGENT_PROFILES.keys())[:12])
    suffix = "…" if len(SUBAGENT_PROFILES) > 12 else ""
    return (
        f"Task rejected: unknown assignee profile '{a}'. "
        f"Use a kanban assignee profile (e.g. {valid}{suffix}), not a personality name."
    )


def _mesh_eligible_agent_creation_allowed() -> bool:
    """Plan B gate: when true, the agent is allowed to set ``mesh-eligible``
    via tools. Default false keeps the Plan A invariant (operator-only via
    REST/MCP)."""
    try:
        from core.config import get_kanban_config
        return bool(get_kanban_config().get("mesh_eligible_agent_creation", False))
    except Exception:
        return False


def _allow_shell_agent_creation_allowed() -> bool:
    """When true, the agent may set ``allow_shell`` on a task via tools.

    Default false keeps the per-task shell opt-in operator-only (spec §4): an
    injected or confused planner cannot grant a worker raw shell access.
    """
    try:
        from core.config import get_kanban_config
        return bool(get_kanban_config().get("allow_shell_agent_creation", False))
    except Exception:
        return False


def _strip_operator_only_metadata(metadata: Any) -> Any:
    """Drop operator-only metadata keys from agent-tier task creation.

    ``allow_shell`` grants a kanban worker raw shell; it is operator-only
    unless ``[kanban].allow_shell_agent_creation`` is set. Logs on strip so
    operators can detect agent attempts to flip it.
    """
    if not isinstance(metadata, dict) or "allow_shell" not in metadata:
        return metadata
    if _allow_shell_agent_creation_allowed():
        return metadata
    md = dict(metadata)
    md.pop("allow_shell", None)
    _log.debug("kanban_create: stripped operator-only metadata 'allow_shell' from agent input")
    return md


def _strip_operator_only_tags(tags: Any) -> list[str]:
    """Remove tags that only operators may set. Logs a debug line on strip
    so operators can detect agent attempts to flip mesh-eligible bits.

    When ``[kanban].mesh_eligible_agent_creation`` is true, ``mesh-eligible``
    is allowed through (Plan B opt-in promotion to agent-tier).
    """
    if tags is None:
        return []
    if isinstance(tags, str):
        try:
            parsed = _json.loads(tags)
        except Exception:
            return []
        tags = parsed
    if not isinstance(tags, list):
        return []
    mesh_allowed = _mesh_eligible_agent_creation_allowed()
    out: list[str] = []
    for t in tags:
        if not isinstance(t, str):
            continue
        s = t.strip()
        if not s:
            continue
        if s == "mesh-eligible" and mesh_allowed:
            out.append(s)
            continue
        if s in ks.OPERATOR_ONLY_TAGS:
            _log.debug("kanban_create: stripped operator-only tag %r from agent input", s)
            continue
        out.append(s)
    return out


def _format_task_summary(task: dict[str, Any]) -> str:
    title = task.get("title") or ""
    if len(title) > 160:
        title = title[:157] + "…"
    return (
        f"id={task.get('id', '?')} status={task.get('status', '?')} "
        f"assignee={task.get('assignee', '?')} priority={task.get('priority', '?')} "
        f"tenant={task.get('tenant', 'default')} title={title!r}"
    )


# ---------------------------------------------------------------------------
# Planner-tier: create a single task
# ---------------------------------------------------------------------------

class KanbanCreateTaskInput(BaseModel):
    title: str = Field(..., description="Short task title (<=240 chars).")
    body: str = Field(
        default="",
        description="Detailed instructions for the assigned worker (<=16000 chars).",
    )
    assignee: str = Field(
        default="general",
        description=(
            "Subagent profile name (e.g. 'research', 'code_readonly', 'extraction', "
            "'web_research_light', 'ops_cautious', 'general')."
        ),
    )
    parent_id: str = Field(
        default="",
        description=(
            "Optional id of a single parent task; this task waits until that "
            "parent is done. For fan-in / diamond DAGs, use parent_ids instead. "
            "Leave empty for a root/standalone task — do NOT pass '__root__' "
            "(that is only a kanban_list_tasks filter, not a real task id)."
        ),
    )
    parent_ids: Any = Field(
        default=None,
        description=(
            "Optional list of parent task ids for multi-parent (fan-in) shapes. "
            "The new task becomes ready only when ALL declared parents are done. "
            "JSON array of strings or a Python list. Mutually exclusive with "
            "parent_id when they disagree; pass parent_ids alone for clarity."
        ),
    )
    priority: int = Field(default=3, description="1 (highest) .. 5 (lowest).")
    tags: Any = Field(
        default=None,
        description="Optional list of strings or JSON array of strings.",
    )
    metadata: Any = Field(
        default=None,
        description=(
            "Optional JSON object with structured fields (acceptance criteria, "
            "links, etc). Stored encrypted; max 8 KB."
        ),
    )
    deadline_seconds: int = Field(
        default=0,
        description="Optional overall deadline in seconds (0 = none, max 7 days).",
    )
    specifier_profile: str = Field(
        default="",
        description=(
            "Optional target profile for triage tasks. When set, the task is "
            "assigned to 'triage' and the dispatcher routes it to the specified "
            "profile after the triage step. Requires triage_enabled in config."
        ),
    )
    tool_names: Any = Field(
        default=None,
        description=(
            "Optional list of tool names the worker should use. When set, the "
            "worker only sees these tools (plus kanban infrastructure tools). "
            "When omitted, the worker gets the full tool set for its profile. "
            "Example: ['write_text_file', 'convert_video']."
        ),
    )
    idempotency_key: str = Field(
        default="",
        description=(
            "Optional caller-chosen key. When provided, repeated calls with the same "
            "(tenant, idempotency_key) return the original task instead of duplicating."
        ),
    )

    @field_validator("tags", mode="before")
    @classmethod
    def _coerce_tags(cls, v: Any) -> Any:
        if v is None:
            return None
        if isinstance(v, str):
            s = v.strip()
            if not s:
                return None
            try:
                return _json.loads(s)
            except Exception:
                return None
        return v

    @field_validator("metadata", mode="before")
    @classmethod
    def _coerce_metadata(cls, v: Any) -> Any:
        if v is None:
            return None
        if isinstance(v, str):
            s = v.strip()
            if not s:
                return None
            try:
                return _json.loads(s)
            except Exception:
                return None
        return v

    @field_validator("parent_ids", mode="before")
    @classmethod
    def _coerce_parent_ids(cls, v: Any) -> Any:
        if v is None:
            return None
        if isinstance(v, str):
            s = v.strip()
            if not s:
                return None
            try:
                parsed = _json.loads(s)
                return parsed if isinstance(parsed, list) else None
            except Exception:
                return None
        if isinstance(v, list):
            return [str(x) for x in v if x]
        return None


SPAWN_DEPTH_META = "spawn_depth"


def _resolve_worker_spawn_depth() -> tuple[int, str]:
    """Spawn-depth budget for tasks created from inside a kanban worker.

    June 2026 incident: self-referential stress-test tasks made each worker
    create more tasks, whose workers created more tasks — unbounded board
    growth (8 → 29 tasks, 39 worker runs/hour) that saturated the local MLX
    server. Concurrency caps bound how fast the board drains, not how fast
    it breeds; this bounds the breeding.

    Returns ``(depth_for_new_tasks, rejection_message)``. Depth 0 means the
    creator is not a worker (operator / chat planner) — no stamp, no cap.
    A non-empty rejection message means creation must be refused.
    """
    bound = current_claim_token()
    if not (isinstance(bound, tuple) and len(bound) == 2 and bound[0]):
        return 0, ""
    current_depth = 1
    try:
        task = ks.get_task(bound[0]) or {}
        md = task.get("metadata")
        if isinstance(md, dict):
            current_depth = max(1, int(md.get(SPAWN_DEPTH_META) or 0) + 1)
    except Exception:
        _log.debug("spawn depth lookup failed; assuming depth 1", exc_info=True)
    try:
        from core.config import get_kanban_config

        cap = int(get_kanban_config().get("max_spawn_depth", 2))
    except Exception:
        cap = 2
    if cap > 0 and current_depth > cap:
        return current_depth, (
            f"Task creation rejected: spawn-depth limit reached "
            f"(this worker is already {current_depth - 1} level(s) deep; cap is {cap}). "
            "Do NOT delegate further — complete your own task directly and report "
            "your findings in the final kanban-result block."
        )
    return current_depth, ""


def _stamp_spawn_depth(metadata: Any, depth: int) -> Any:
    if depth <= 0:
        return metadata
    md = dict(metadata) if isinstance(metadata, dict) else {}
    md[SPAWN_DEPTH_META] = depth
    return md


# '__root__' is a kanban_list_tasks *filter* sentinel ("list root tasks"). Models
# routinely leak it into *creation* as parent_id="__root__" to mean "this is a
# root task", and storage rejects it with an unhelpful generic "does not exist",
# costing an extra self-correction round. No literal task id is ever one of these
# words, so flagging them here can never mask a real parent reference.
_ROOT_PARENT_SENTINELS = frozenset({"__root__", "root", "none", "null"})


def _reject_root_sentinel_parent(parent_id: Any, parent_ids: Any) -> str | None:
    """Targeted, actionable error when a root-sentinel is used as a *parent ref*.

    Returns a one-round-fixable hint (honoring kernel §5 — no silent coercion;
    the model must omit the field) or ``None`` when the references are clean.
    """
    refs: list[str] = []
    if isinstance(parent_id, str) and parent_id.strip():
        refs.append(parent_id.strip())
    if isinstance(parent_ids, (list, tuple)):
        refs.extend(str(x).strip() for x in parent_ids if str(x).strip())
    elif isinstance(parent_ids, str) and parent_ids.strip():
        refs.append(parent_ids.strip())
    for ref in refs:
        if ref.lower() in _ROOT_PARENT_SENTINELS:
            return (
                f"Invalid parent reference {ref!r}: '__root__' is only a "
                "kanban_list_tasks filter, not a real task id. For a root or "
                "standalone task, OMIT parent_id / parent_ids / parent_index "
                "entirely (a task with no parent is a root task)."
            )
    return None


@tool(args_schema=KanbanCreateTaskInput)
def kanban_create_task(
    title: str,
    body: str = "",
    assignee: str = "general",
    parent_id: str = "",
    parent_ids: Any = None,
    priority: int = 3,
    tags: Any = None,
    metadata: Any = None,
    specifier_profile: str = "",
    deadline_seconds: int = 0,
    idempotency_key: str = "",
    tool_names: Any = None,
) -> str:
    """Drop a task on the durable kanban board. Returns its id and status.

    Use this for work that can run autonomously — research, extraction,
    code-readonly investigations, ops tasks. Pick `assignee` from the
    SUBAGENT_PROFILES (research, web_research_light, code_readonly,
    extraction, ops_cautious, general).

    Single-parent (sequential): set `parent_id`. Multi-parent (fan-in /
    diamond): set `parent_ids=[..., ...]` to make this task wait for ALL
    listed parents to finish. For short, single-thread, in-conversation
    work, use `update_task_plan` instead. For multi-step / multi-profile
    DAGs, prefer `kanban_create_batch`.

    Pass `tool_names` to **narrow** which tools the worker sees (e.g.
    ``['read_file']`` on code_readonly). Names outside the assignee profile
    are dropped; kanban infrastructure tools are unioned with any valid
    override (narrow-only — never widens the profile).
    """
    spawn_depth, depth_rejection = _resolve_worker_spawn_depth()
    if depth_rejection:
        return depth_rejection
    root_sentinel_err = _reject_root_sentinel_parent(parent_id, parent_ids)
    if root_sentinel_err:
        return root_sentinel_err
    safe_tags = _strip_operator_only_tags(tags)
    parent_ids_arg: list[str] | None = None
    if isinstance(parent_ids, list):
        parent_ids_arg = [s for s in (str(x).strip() for x in parent_ids) if s]
    spec = (specifier_profile or "").strip()

    original_tool_names = tool_names
    effective_metadata = _strip_operator_only_metadata(
        _stamp_spawn_depth(metadata, spawn_depth)
    )
    if tool_names and isinstance(tool_names, list):
        coerced = [str(t).strip() for t in tool_names if str(t).strip()]
        if coerced:
            effective_metadata = dict(effective_metadata or {})
            effective_metadata["tool_names"] = coerced

    assignee_rejection = _reject_unknown_assignee(assignee)
    if assignee_rejection:
        return assignee_rejection

    try:
        task = ks.create_task(
            title=title,
            body=body or "",
            assignee=assignee,
            parent_id=(parent_id or "") or None,
            parent_ids=parent_ids_arg,
            priority=priority,
            tags=safe_tags,
            metadata=effective_metadata,
            origin_thread_id=_origin_thread_id(),
            deadline_seconds=deadline_seconds,
            idempotency_key=idempotency_key,
            specifier_profile=spec,
        )
    except RuntimeError as exc:
        return f"Kanban disabled: {exc}"
    except (ValueError, PermissionError) as exc:
        return f"Task rejected: {exc}"
    except Exception as exc:
        _log.warning("kanban_create_task failed", exc_info=True)
        return f"Failed to create kanban task: {exc}"
    try:
        nudge_dispatcher()
    except Exception:
        pass
    result = "Created kanban task: " + _format_task_summary(task)
    if original_tool_names and not (task.get("metadata") or {}).get("tool_names"):
        result += "\n(note: all tool_names were invalid and were removed)"
    return result


# ---------------------------------------------------------------------------
# Planner-tier: create a batch (DAG) atomically
# ---------------------------------------------------------------------------

class KanbanCreateBatchInput(BaseModel):
    items: Any = Field(
        ...,
        description=(
            "List of task dicts. Each item: {title, body?, assignee?, priority?, "
            "tags?, metadata?, deadline_seconds?, idempotency_key?, parent_index?, "
            "parent_indices?, parent_id?, parent_ids?}. parent_index / parent_indices "
            "reference earlier items in the list (0-based); parent_id / parent_ids "
            "reference existing task ids. Indices and ids may be combined for "
            "diamond / fan-in shapes (e.g. parent_indices=[0, 1] + parent_ids=[...]). "
            "For a root item, OMIT all parent fields — do NOT set parent_id="
            "'__root__' (that is only a kanban_list_tasks filter). "
            "Pass at most 64 items."
        ),
    )

    @field_validator("items", mode="before")
    @classmethod
    def _coerce_items(cls, v: Any) -> Any:
        if v is None:
            return []
        if isinstance(v, str):
            s = v.strip()
            if not s:
                return []
            try:
                return _json.loads(s)
            except Exception as exc:
                raise ValueError(f"items is not valid JSON: {exc}") from exc
        return v


@tool(args_schema=KanbanCreateBatchInput)
def kanban_create_batch(items: Any) -> str:
    """Atomically insert a DAG of kanban tasks.

    Use when a goal naturally splits into multiple coordinated tasks
    (research fan-out + assembly, multi-stage codebase audit, etc).
    `parent_index` and `parent_indices` make intra-batch dependencies easy
    to express (0-based, must reference earlier items in the list).

    Diamond / fan-in example — task 3 waits for both 1 and 2:

      [
        {"title": "Plan", "assignee": "general"},
        {"title": "Subtopic A", "assignee": "research", "parent_index": 0},
        {"title": "Subtopic B", "assignee": "research", "parent_index": 0},
        {"title": "Assemble final report", "assignee": "general",
         "parent_indices": [1, 2],
         "metadata": {"acceptance": ["covers A and B with citations"]}}
      ]

    All items are inserted in a single transaction; cycle detection,
    idempotency, and backpressure caps apply across the batch.
    """
    if not isinstance(items, list):
        if isinstance(items, str) and items.strip().startswith("["):
            return (
                "items must be a JSON array of task dicts, not a string. "
                'Pass items as an array: {"items": [{"title": "...", '
                '"assignee": "general"}]}, not a stringified array: '
                '{"items": "[{...}]"}.'
            )
        return "items must be a list of task dicts"
    spawn_depth, depth_rejection = _resolve_worker_spawn_depth()
    if depth_rejection:
        return depth_rejection
    safe_items: list[dict[str, Any]] = []
    for raw in items:
        if not isinstance(raw, dict):
            return "each item must be a dict"
        item = dict(raw)
        assignee_rejection = _reject_unknown_assignee(str(item.get("assignee") or "general"))
        if assignee_rejection:
            return assignee_rejection
        root_sentinel_err = _reject_root_sentinel_parent(
            item.get("parent_id"), item.get("parent_ids")
        )
        if root_sentinel_err:
            return root_sentinel_err
        item["tags"] = _strip_operator_only_tags(item.get("tags"))
        item["metadata"] = _strip_operator_only_metadata(
            _stamp_spawn_depth(item.get("metadata"), spawn_depth)
        )
        safe_items.append(item)
    try:
        rows = ks.create_batch(
            safe_items,
            origin_thread_id=_origin_thread_id(),
        )
    except RuntimeError as exc:
        return f"Kanban disabled: {exc}"
    except (ValueError, PermissionError) as exc:
        return f"Tool error: Batch rejected: {exc}"
    except Exception as exc:
        _log.warning("kanban_create_batch failed", exc_info=True)
        return f"Failed to create batch: {exc}"
    try:
        nudge_dispatcher()
    except Exception:
        pass
    lines = [f"Created {len(rows)} kanban tasks:"]
    batch_meta = (rows[0].get("metadata") or {}) if rows else {}
    from utils.kanban_coordinator import COORDINATOR_BATCH_ACTIVATED_MARKER, COORDINATOR_BATCH_ID_META

    coord_batch = batch_meta.get(COORDINATOR_BATCH_ID_META)
    if isinstance(coord_batch, str) and coord_batch.strip():
        lines.append(f"{COORDINATOR_BATCH_ACTIVATED_MARKER} {coord_batch.strip()}")
    for r in rows:
        lines.append("  " + _format_task_summary(r))
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Planner-tier: list / show / list runs
# ---------------------------------------------------------------------------

class KanbanListTasksInput(BaseModel):
    status: str = Field(
        default="",
        description=(
            "Optional filter: comma-separated subset of "
            "todo|ready|in_progress|blocked|done|cancelled. Empty = all."
        ),
    )
    assignee: str = Field(default="", description="Optional profile filter.")
    tenant: str = Field(default="", description="Optional tenant filter.")
    parent_id: str = Field(
        default="",
        description="Optional parent_id filter. Pass '__root__' to list root tasks.",
    )
    origin_thread_id: str = Field(
        default="",
        description="Optional origin conversation thread id filter.",
    )
    limit: int = Field(default=20, description="Max rows to return (1..100).")


@tool(args_schema=KanbanListTasksInput)
def kanban_list_tasks(
    status: str = "",
    assignee: str = "",
    tenant: str = "",
    parent_id: str = "",
    origin_thread_id: str = "",
    limit: int = 20,
) -> str:
    """List kanban tasks matching the filters; ordered by priority, then
    age. Default returns the next ~20 tasks across all statuses. Use to
    see the board state before creating new work."""
    statuses: list[str] | None = None
    if status:
        statuses = [s.strip().lower() for s in status.split(",") if s.strip()]
    parent_filter: str | None
    if not parent_id:
        parent_filter = None
    elif parent_id == "__root__":
        parent_filter = ""
    else:
        parent_filter = parent_id
    origin_filter: str | None = (origin_thread_id or "").strip() or None
    try:
        rows = ks.list_tasks(
            status=statuses,
            assignee=assignee or None,
            tenant=tenant or None,
            parent_id=parent_filter,
            origin_thread_id=origin_filter,
            limit=max(1, min(int(limit or 20), 100)),
        )
    except Exception as exc:
        return f"Failed to list tasks: {exc}"
    if not rows:
        return "No kanban tasks match."
    lines = [f"Found {len(rows)} kanban task(s):"]
    for r in rows:
        lines.append("  " + _format_task_summary(r))
    return "\n".join(lines)


class KanbanShowTaskInput(BaseModel):
    task_id: str = Field(..., description="Task id (e.g. 'kt_abc...').")


@tool(args_schema=KanbanShowTaskInput)
def kanban_show_task(task_id: str) -> str:
    """Show the full details of a single kanban task, including the body,
    metadata, and recent run summaries (terminal outcomes only)."""
    try:
        task = ks.get_task(task_id)
    except Exception as exc:
        return f"Failed to read task: {exc}"
    if not task:
        return f"No task with id={task_id!r}"
    runs = ks.list_runs(task_id, limit=5)
    parent_ids = task.get("parent_ids") or []
    if isinstance(parent_ids, list) and parent_ids:
        if len(parent_ids) == 1:
            parents_repr = f"parents={parent_ids[0]}"
        else:
            parents_repr = "parents=[" + ", ".join(parent_ids) + "]"
    else:
        legacy = task.get("parent_id") or ""
        parents_repr = f"parents={legacy or 'None'}"
    lines = [
        _format_task_summary(task),
        f"  {parents_repr}",
        f"  failure_count={task.get('failure_count', 0)} "
        f"deadline_at={task.get('deadline_at') or 'None'} "
        f"completed_at={task.get('completed_at') or 'None'}",
    ]
    body = (task.get("body") or "").strip()
    if body:
        if len(body) > 800:
            body = body[:800] + "…"
        lines.append("  body:")
        for ln in body.splitlines():
            lines.append("    " + ln)
    metadata = task.get("metadata") or {}
    if isinstance(metadata, dict) and metadata:
        try:
            md = _json.dumps(metadata, ensure_ascii=False)
            if len(md) > 600:
                md = md[:600] + "…"
            lines.append(f"  metadata: {md}")
        except Exception:
            pass
    if runs:
        lines.append(f"  runs (most recent {len(runs)}):")
        for r in runs:
            outcome = r.get("outcome") or "?"
            attempt = r.get("attempt", "?")
            summary = (r.get("summary") or "").strip()
            if len(summary) > 200:
                summary = summary[:200] + "…"
            error = (r.get("error") or "").strip()
            if len(error) > 200:
                error = error[:200] + "…"
            lines.append(
                f"    attempt {attempt}: {outcome} "
                + (f"summary={summary!r}" if summary else "")
                + (f" error={error!r}" if error else "")
            )
    return "\n".join(lines)


class KanbanListRunsInput(BaseModel):
    task_id: str = Field(..., description="Task id.")
    limit: int = Field(default=20, description="Max rows (1..100).")


@tool(args_schema=KanbanListRunsInput)
def kanban_list_runs(task_id: str, limit: int = 20) -> str:
    """List run history (attempts) for a kanban task."""
    try:
        runs = ks.list_runs(task_id, limit=max(1, min(int(limit or 20), 100)))
    except Exception as exc:
        return f"Failed to list runs: {exc}"
    if not runs:
        return f"No runs recorded for task {task_id}."
    try:
        from utils.kanban_dispatcher import enrich_kanban_run_row
    except Exception:
        enrich_kanban_run_row = None
    lines = [f"Runs for {task_id} ({len(runs)}):"]
    for raw in runs:
        r = enrich_kanban_run_row(raw) if enrich_kanban_run_row else dict(raw)
        outcome = r.get("outcome") or "?"
        attempt = r.get("attempt", "?")
        started_at = r.get("started_at") or ""
        ended_at = r.get("ended_at") or ""
        lines.append(f"  attempt {attempt}: {outcome} started={started_at} ended={ended_at}")
        summary = (r.get("summary") or "").strip()
        if summary:
            if len(summary) > 200:
                summary = summary[:200] + "…"
            lines.append(f"    summary: {summary}")
        error = (r.get("error") or "").strip()
        if error:
            if len(error) > 200:
                error = error[:200] + "…"
            lines.append(f"    error: {error}")
        md = r.get("metadata") or {}
        if isinstance(md, dict):
            typed = md.get("failure_reason") or r.get("typed_reason")
            if isinstance(typed, str) and typed.strip():
                lines.append(f"    typed_reason: {typed.strip()}")
            failure_class = md.get("failure_class") or r.get("failure_class")
            if isinstance(failure_class, str) and failure_class.strip():
                lines.append(f"    failure_class: {failure_class.strip()}")
            elif isinstance(typed, str) and typed.strip():
                from utils.kanban_context import classify_failure_class

                lines.append(
                    f"    failure_class: {classify_failure_class(typed.strip())}"
                )
            next_action = md.get("next_action") or r.get("next_action")
            if isinstance(next_action, str) and next_action.strip():
                na = next_action.strip()
                if len(na) > 200:
                    na = na[:200] + "…"
                lines.append(f"    next_action: {na}")
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Planner-tier: cancel
# ---------------------------------------------------------------------------

class KanbanCancelTaskInput(BaseModel):
    task_id: str = Field(..., description="Task id to cancel.")
    cascade: bool = Field(
        default=True,
        description="Cancel descendants too (default true).",
    )


@tool(args_schema=KanbanCancelTaskInput)
def kanban_cancel_task(task_id: str, cascade: bool = True) -> str:
    """Cancel a kanban task (and by default its descendants). The dispatcher
    notices on its next tick and stops the worker if one is running."""
    try:
        cancelled = ks.cancel_task(task_id, cascade=bool(cascade))
    except Exception as exc:
        return f"Failed to cancel task: {exc}"
    if not cancelled:
        return f"No active task to cancel for id={task_id!r}."
    if cascade and len(cancelled) > 1:
        return f"Cancelled {len(cancelled)} tasks (root + descendants): {', '.join(cancelled)}"
    return f"Cancelled task {cancelled[0]}."


# ---------------------------------------------------------------------------
# Worker-tier: claim-scoped tools
# ---------------------------------------------------------------------------

class KanbanPostSummaryInput(BaseModel):
    task_id: str = Field(..., description="Task id you currently hold the claim for.")
    summary: str = Field(..., description="Dense, factual handoff summary (<=8000 chars).")
    metadata: Any = Field(
        default=None,
        description=(
            "Optional JSON object of structured handoff fields "
            "(changed_files, decisions, etc). Replaces any previously posted metadata."
        ),
    )
    status: str = Field(
        default="",
        description=(
            "Optional completion status of this handoff: 'success' (the task is done — this is "
            "the default when omitted), 'partial' (some deliverable exists but verification "
            "failed), or 'blocked' (you cannot finish). For 'partial' or 'blocked' also include "
            "metadata.blockers and metadata.next_action. The dispatcher still independently "
            "verifies the result and owns the completion decision — this only declares your "
            "honest status."
        ),
    )

    @field_validator("metadata", mode="before")
    @classmethod
    def _coerce_metadata(cls, v: Any) -> Any:
        if v is None:
            return None
        if isinstance(v, str):
            s = v.strip()
            if not s:
                return None
            try:
                return _json.loads(s)
            except Exception:
                return None
        return v


def _verify_claim_scope(task_id: str) -> str | None:
    """Return a string error when the caller is not running inside the
    kanban worker for *task_id*; None when authorized."""
    bound = current_claim_token()
    if bound is None:
        return "kanban worker tools may only be called from inside a kanban worker"
    bound_task, _bound_token = bound
    if bound_task != task_id:
        return (
            f"claim mismatch: this worker holds {bound_task!r}, not {task_id!r}; "
            "you may only act on the task you currently own"
        )
    return None


def _coerce_owned_task(task_id: str) -> tuple[str, str] | str:
    """Resolve the task a worker may write to, coercing a mismatched ``task_id``
    to the task it actually owns.

    A worker holds exactly one claim, so a ``task_id`` that does not match is
    model label-confusion (it commonly grabs the parent/coordinator id from its
    context) — NOT an authorization event. Refusing here discards the worker's
    completed handoff (incident kt_0fe2af9ad5f245: a finished research worker
    posted to its parent's id, got refused, and crashed as contract_missing).

    Coercion stays safe: the write always lands on the worker's OWN task and
    never touches the foreign id, so the "a worker can only affect the task it
    owns" invariant is preserved. Returns ``(owned_task_id, claim_token)`` or an
    error string when no claim is held at all.
    """
    bound = current_claim_token()
    if bound is None:
        return "kanban worker tools may only be called from inside a kanban worker"
    bound_task, bound_token = bound
    if task_id and task_id != bound_task:
        _log.warning(
            "kanban worker tool called with task_id=%r but worker holds %r; "
            "coercing to the owned task",
            task_id,
            bound_task,
        )
    return bound_task, bound_token


@tool(args_schema=KanbanPostSummaryInput)
def kanban_post_summary(task_id: str, summary: str, metadata: Any = None, status: str = "") -> str:
    """Worker-only: write your handoff summary + metadata to the active run.

    The dispatcher reads this when your run finishes and merges it with any trailing
    kanban-result fence in your final message: a fence summary takes precedence when present,
    but an honest ``partial``/``blocked`` status posted here is preserved over a success-shaped
    fence. The dispatcher then independently verifies the result and owns the completion
    decision — this tool never marks a task done.
    Calling this from outside a kanban worker is rejected.
    """
    resolved = _coerce_owned_task(task_id)
    if isinstance(resolved, str):
        return f"Refused: {resolved}"
    task_id, claim_token = resolved
    summary_v = (summary or "").strip()
    if not summary_v:
        return "Refused: summary must be non-empty"
    if len(summary_v) > ks.MAX_SUMMARY_CHARS:
        return f"Refused: summary exceeds {ks.MAX_SUMMARY_CHARS}-char limit"
    metadata_v: dict[str, Any] = {}
    if isinstance(metadata, dict):
        metadata_v = dict(metadata)
    elif isinstance(metadata, str) and metadata.strip():
        try:
            parsed = _json.loads(metadata)
            if isinstance(parsed, dict):
                metadata_v = dict(parsed)
        except Exception:
            pass
    # Translate the explicit status enum into the verifier's metadata.status contract.
    # Success is signalled by an ABSENT metadata.status (verify_kanban_result rejects an
    # explicit success/done/completed), so an explicit success enum must clear it; the
    # dedicated status field wins over any status carried inside metadata.
    status_v = (status or "").strip().lower()
    if status_v:
        if status_v in ("success", "done", "completed", "complete", "ok"):
            metadata_v.pop("status", None)
        elif status_v in ("partial", "blocked"):
            metadata_v["status"] = status_v
        else:
            return "Refused: status must be one of success, partial, blocked"
    # Grain-reframe Stage B: mark this as a *deliberate* native-tool handoff. The dispatcher uses
    # this (only when [kanban].native_result_handoff is on) to classify a fence-absent completion
    # as first-class ("tool_submit") rather than the salvage "inband_summary" substitution. The
    # marker is harmless provenance when the flag is off.
    metadata_v["_handoff_channel"] = "tool"
    try:
        from chat_storage import _encrypt
        from kanban_storage import _open_conn, list_runs as _list_runs
        runs = _list_runs(task_id, limit=3)
        active_id = ""
        existing_md: dict[str, Any] = {}
        for r in runs:
            if r.get("outcome") == ks.OUTCOME_ACTIVE:
                active_id = r.get("id") or ""
                if isinstance(r.get("metadata"), dict):
                    existing_md = dict(r["metadata"])
                break
        if not active_id:
            return "Refused: no active run for this task; cannot post summary"
        # Merge over the run's existing metadata instead of replacing it —
        # a plain overwrite destroys the dispatcher-owned ``progress_log``
        # (mission log) accumulated by kanban_post_progress / auto progress
        # capture. Tool-supplied fields win; progress_log always survives.
        merged_md = {**existing_md, **metadata_v}
        if "progress_log" in existing_md:
            merged_md["progress_log"] = existing_md["progress_log"]
        if status_v in ("success", "done", "completed", "complete", "ok"):
            # An explicit success enum must clear any stale partial/blocked
            # status carried over from a previous post on the same run.
            merged_md.pop("status", None)
        metadata_v = merged_md
        conn = _open_conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            verify = conn.execute(
                "SELECT claim_token, status FROM kanban_tasks WHERE id=?",
                (task_id,),
            ).fetchone()
            if verify is None or verify["claim_token"] != claim_token or verify["status"] != ks.STATUS_IN_PROGRESS:
                conn.rollback()
                return "Refused: claim_token mismatch or task not in_progress"
            conn.execute(
                "UPDATE kanban_task_runs SET summary_ciphertext=?, metadata_ciphertext=? WHERE id=?",
                (
                    _encrypt(summary_v),
                    _encrypt(_json.dumps(metadata_v, ensure_ascii=False)),
                    active_id,
                ),
            )
            conn.commit()
        finally:
            conn.close()
    except Exception as exc:
        _log.warning("kanban_post_summary persist failed", exc_info=True)
        return f"Failed to post summary: {exc}"
    return "Summary posted to the active run; will be used as the worker's handoff."


class KanbanPostProgressInput(BaseModel):
    task_id: str = Field(..., description="Task id you currently hold the claim for.")
    note: str = Field(..., description="Short progress note (<=256 chars).")


@tool(args_schema=KanbanPostProgressInput)
def kanban_post_progress(task_id: str, note: str) -> str:
    """Worker-only: append a structured progress note to the active run.

    Notes are appended to ``metadata.progress_log`` on the run row and
    are visible in `kanban_show_task`. Use sparingly — long-running tasks
    that need to surface live state for operators or sibling workers.
    """
    resolved = _coerce_owned_task(task_id)
    if isinstance(resolved, str):
        return f"Refused: {resolved}"
    task_id, claim_token = resolved
    try:
        ok = ks.append_progress_note(task_id, note, expected_token=claim_token)
    except PermissionError as exc:
        return f"Refused: {exc}"
    except (ValueError, RuntimeError) as exc:
        return f"Failed to post progress: {exc}"
    except Exception as exc:
        _log.warning("kanban_post_progress failed", exc_info=True)
        return f"Failed to post progress: {exc}"
    return "Progress note appended." if ok else "No active run; progress not appended."
