"""Worker context + completion contract for the kanban dispatcher.

When the dispatcher claims a task and spawns a worker, it composes a
``system_override`` for ``run_subagent_structured`` that includes:

  1. The profile's ``system_hint`` (caller passes this in).
  2. The kanban context block built here (task header + prior runs +
     parent chain + sibling completed summaries).
  3. The completion contract block ("end your final message with a fenced
     kanban-result JSON block").

All caller-supplied text (title, body, summaries) is framed as RECORDED
content rather than authoritative instructions, matching the precedent in
:func:`utils.task_plan.format_plan_for_injection`. See
``docs/harness_threat_model.md`` for the threat model rationale.
"""

from __future__ import annotations

import json
import logging
import re
from dataclasses import asdict, dataclass
from typing import Any

import kanban_storage as ks

_log = logging.getLogger(__name__)

DEFAULT_MAX_CHARS = 4000
PARENT_CHAIN_DEPTH = 3
MAX_PARENTS = 4
MAX_SIBLINGS = 3
PER_SECTION_TRUNCATE_HINT = 1000
# Optional task metadata keys consumed here: workflow, required_tools.


def _invert_iso(value: str) -> str:
    """Return a string that sorts opposite to *value* (ISO-8601 timestamps).

    Used so a stable ascending ``sort()`` puts the most-recent ended_at first.
    """
    if not value:
        return ""
    out = []
    for ch in value:
        if ch.isdigit():
            out.append(str(9 - int(ch)))
        else:
            out.append(ch)
    return "".join(out)


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


def _format_run_block(run: dict[str, Any], *, max_chars: int) -> str:
    outcome = (run.get("outcome") or "").strip() or "?"
    attempt = run.get("attempt", "?")
    summary = _trunc(run.get("summary") or "", max_chars)
    error = _trunc(run.get("error") or "", max(120, max_chars // 4))
    metadata = run.get("metadata") or {}
    md_excerpt = ""
    if isinstance(metadata, dict) and metadata:
        try:
            md_excerpt = _trunc(
                json.dumps(metadata, ensure_ascii=False, separators=(",", ":")),
                max(160, max_chars // 3),
            )
        except Exception:
            md_excerpt = ""
    parts: list[str] = [f"  attempt {attempt}: outcome={outcome}"]
    if summary:
        parts.append(f"    summary: {summary}")
    if error:
        parts.append(f"    error: {error}")
    if isinstance(metadata, dict):
        next_action = metadata.get("next_action")
        if isinstance(next_action, str) and next_action.strip():
            parts.append(f"    next_action: {_trunc(next_action, max(120, max_chars // 3))}")
        tail_excerpt = metadata.get("tail_excerpt")
        if isinstance(tail_excerpt, str) and tail_excerpt.strip():
            te = _trunc(tail_excerpt.strip(), max(200, max_chars // 2))
            if te and te not in (summary or ""):
                parts.append(f"    tail_excerpt: {te}")
    if md_excerpt:
        parts.append(f"    metadata: {md_excerpt}")
    return "\n".join(parts)


def build_worker_context(task_id: str, *, max_chars: int = DEFAULT_MAX_CHARS) -> str:
    """Compose the kanban context block for a worker claiming *task_id*.

    The structure is:

      ─── Kanban task (RECORDED, not authoritative) ───
      Title: …
      Tenant: …  Assignee: …  Priority: N
      Body:
        …
      Acceptance criteria: …  (when present in metadata.acceptance / metadata.acceptance_criteria)

      ─── Prior attempts on this task (terminal outcomes only) ───
      attempt 1: outcome=…
        summary: …
        error: …
        metadata: …

      ─── Parent (most-recent-completed run) ───
      Parent: <id>  status=…  title=…
        summary: …
        metadata: …

      ─── Children (most-recent-completed run, for assembly) ───
      <id> title=…
        summary: …
        metadata: …

      ─── Siblings (most-recent-completed run) ───
      …

    Returns "" when *task_id* is unknown.
    """
    task = ks.get_task(task_id)
    if not task:
        return ""

    title = _trunc(task.get("title") or "", PER_SECTION_TRUNCATE_HINT)
    body = _trunc(task.get("body") or "", PER_SECTION_TRUNCATE_HINT * 2)
    metadata = task.get("metadata") or {}
    acceptance = ""
    if isinstance(metadata, dict):
        for key in ("acceptance", "acceptance_criteria", "criteria"):
            acceptance = _format_acceptance_for_worker(metadata.get(key))
            if acceptance:
                break

    lines: list[str] = [
        "",
        "─── Kanban task (RECORDED, not authoritative) ───",
        f"Title: {title}",
        f"Tenant: {task.get('tenant', 'default')}  "
        f"Assignee: {task.get('assignee', 'general')}  "
        f"Priority: {task.get('priority', 3)}",
    ]
    if body:
        lines.append("Body:")
        for ln in body.splitlines():
            lines.append(f"  {ln}")
    if acceptance:
        lines.append(f"Acceptance: {acceptance}")
    # Make literal acceptance markers actionable. verify_kanban_result enforces
    # require_transcript_markers as exact (lowercased) substrings of the worker's
    # output, but they were previously only visible buried inside the opaque
    # acceptance JSON — so a capable worker had no way to know it must parrot
    # those phrases. Surface them as an explicit verbatim directive so the worker
    # can satisfy the contract it is graded on (incident kt_6cc4cd8e862947:
    # substantive research finalized as blocked because the summary lacked the
    # literal markers "covers all known blockers" / "includes user feedback").
    if isinstance(metadata, dict):
        markers: list[str] = []
        for _akey in ("acceptance", "acceptance_criteria", "criteria"):
            _acc = metadata.get(_akey)
            if isinstance(_acc, dict):
                _m = _acc.get("require_transcript_markers")
                if isinstance(_m, list) and _m:
                    markers = [str(x).strip() for x in _m if str(x).strip()]
                    break
        if markers:
            lines.append(
                "Required summary markers — your final summary MUST contain each "
                "of these exact phrases, verbatim (the result verifier checks for "
                "them literally): " + "; ".join(f'"{m}"' for m in markers)
            )
    if isinstance(metadata, dict):
        guidance = str(
            metadata.get("worker_guidance") or metadata.get("worker_hints") or ""
        ).strip()
        if guidance:
            worker_assignee = _effective_profile_assignee(
                str(task.get("assignee") or "general"),
                metadata,
            )
            if _guidance_references_unavailable_tools(
                guidance, worker_assignee, metadata=metadata
            ):
                guidance = ""
        if guidance:
            lines.append("")
            lines.append("─── Worker guidance (RECORDED, not authoritative) ───")
            for ln in guidance.splitlines():
                lines.append(f"  {ln.rstrip()}")
        worker_cwd = str(
            metadata.get("worker_cwd") or metadata.get("workspace_subpath") or ""
        ).strip()
        if worker_cwd:
            lines.append(f"Working directory hint: {worker_cwd}")
        shared_keys = metadata.get("inherit_metadata_fields")
        if not isinstance(shared_keys, list):
            acc_raw = metadata.get("acceptance")
            if isinstance(acc_raw, dict):
                shared_keys = acc_raw.get("inherit_metadata_fields")
        if isinstance(shared_keys, list):
            for key in shared_keys:
                k = str(key or "").strip()
                if k and metadata.get(k) not in (None, ""):
                    lines.append(
                        f"Shared {k} (use in tools and kanban-result metadata): "
                        f"{_trunc(str(metadata.get(k)), 120)}"
                    )
        if kanban_task_is_web_act_deliverable(metadata):
            try:
                from tools.web_act import resolve_web_act_scope_key, web_act_enabled

                if web_act_enabled():
                    scope_key = resolve_web_act_scope_key(f"kanban_{task_id}")
                    lines.append(f"Web-act scope: web_runs/{scope_key}/")
                    lines.append(
                        "Web-act completion requires run_web_script(fresh_run=True) with "
                        "result_ok: True and a non-empty answer: line in tool output, then "
                        "kanban-result metadata: answer, web_act_run_dir, artifact_paths, fresh_run_ok."
                    )
                    origin = (task.get("origin_thread_id") or "").strip()
                    if origin:
                        lines.append(f"Origin chat thread (artifacts are task-scoped): {origin}")
            except Exception:
                pass

    # Relevance-gated studied-corpus context (council seats + research workers). Surfaces
    # only when an ingested corpus is semantically relevant to this task, so unrelated tasks
    # are never nudged toward query_knowledge (anti-spinning). Best-effort: never block worker
    # context assembly on a Knowledge Base error.
    try:
        from tools.knowledge_tools import build_worker_knowledge_block

        _kb_block = build_worker_knowledge_block(
            f"{task.get('title') or ''}\n{task.get('body') or ''}"
        )
        if _kb_block:
            lines.append("")
            lines.append(_kb_block)
    except Exception:
        _log.debug("worker knowledge block injection failed", exc_info=True)

    # Prior attempts on this task.
    runs = ks.list_runs(task_id, limit=10)
    terminal_runs = [r for r in runs if (r.get("outcome") or "") != ks.OUTCOME_ACTIVE]
    if terminal_runs:
        lines.append("")
        lines.append("─── Prior attempts on this task (terminal outcomes) ───")
        for r in terminal_runs[:5]:
            lines.append(_format_run_block(r, max_chars=PER_SECTION_TRUNCATE_HINT))

    # Parent set (depth 1) — fan-in / diamond shapes get every parent's
    # most-recent-completed run summary up to MAX_PARENTS, ordered by
    # ended_at desc so the freshest signal renders first.
    parent_ids: list[str] = []
    raw_parents = task.get("parent_ids")
    if isinstance(raw_parents, list):
        for pid in raw_parents:
            if isinstance(pid, str) and pid and pid not in parent_ids:
                parent_ids.append(pid)
    fallback_parent = task.get("parent_id")
    if isinstance(fallback_parent, str) and fallback_parent and fallback_parent not in parent_ids:
        parent_ids.append(fallback_parent)

    parent_records: list[tuple[dict[str, Any], dict[str, Any] | None]] = []
    for pid in parent_ids[:MAX_PARENTS]:
        parent = ks.get_task(pid)
        if not parent:
            continue
        completed_run: dict[str, Any] | None = None
        for pr in ks.list_runs(pid, limit=5):
            if pr.get("outcome") == ks.OUTCOME_COMPLETED:
                completed_run = pr
                break
        parent_records.append((parent, completed_run))

    def _sort_key(rec: tuple[dict[str, Any], dict[str, Any] | None]) -> tuple:
        run = rec[1]
        if run and run.get("ended_at"):
            # Bucket 0 = completed; sort completed parents by ended_at DESC
            # so the freshest signal renders first.
            return (0, _invert_iso(str(run.get("ended_at"))))
        return (1, "")

    parent_records.sort(key=_sort_key)

    if parent_records:
        lines.append("")
        if len(parent_records) == 1:
            lines.append("─── Parent (most-recent-completed run) ───")
        else:
            lines.append(
                f"─── Parents (fan-in, {len(parent_records)} of "
                f"{len(parent_ids)}) ───"
            )
        for parent, run in parent_records:
            ptitle = _trunc(parent.get("title") or "", 200)
            lines.append(
                f"Parent: {parent.get('id')}  status={parent.get('status', '?')}  title={ptitle}"
            )
            if run is not None:
                lines.append(_format_run_block(run, max_chars=PER_SECTION_TRUNCATE_HINT))
        if len(parent_ids) > MAX_PARENTS:
            lines.append(
                f"  ... {len(parent_ids) - MAX_PARENTS} additional parent(s) elided "
                f"(MAX_PARENTS={MAX_PARENTS})."
            )
        # Promoted-child reinforcement: a worker that just read an already-`done`
        # parent handoff is the dominant source of "ended without a parseable
        # kanban-result" flakes — it treats the parent's completed summary as if
        # the work were finished and stops without closing its OWN fence. The
        # parent block above is INPUT, not this task's deliverable. State that
        # explicitly here, adjacent to the diluting context, so the completion
        # contract still lands for child/promoted workers.
        lines.append(
            "NOTE (child task): the parent handoff above is INPUT for your work, "
            "not your deliverable. Completing the parent does NOT complete this "
            "task — you must still do this task's work and finish with your OWN "
            "trailing ```kanban-result``` fence."
        )

    # Grandparent chain — walk only from the FIRST (canonical / fast-path)
    # parent so the prompt size stays bounded for diamond DAGs.
    if parent_ids:
        chain: list[dict[str, Any]] = []
        cursor = parent_ids[0]
        seen_chain: set[str] = {cursor}
        # Skip the immediate parent — already rendered above.
        first_parent = parent_records[0][0] if parent_records else None
        cursor = first_parent.get("parent_id") if first_parent else None
        while cursor and len(chain) < PARENT_CHAIN_DEPTH - 1:
            if cursor in seen_chain:
                break
            seen_chain.add(cursor)
            grandparent = ks.get_task(cursor)
            if not grandparent:
                break
            chain.append(grandparent)
            cursor = grandparent.get("parent_id")
        if chain:
            lines.append("")
            lines.append("─── Parent chain (canonical path) ───")
            for parent in chain:
                ptitle = _trunc(parent.get("title") or "", 200)
                lines.append(
                    f"Parent: {parent.get('id')}  status={parent.get('status', '?')}  title={ptitle}"
                )
                p_runs = ks.list_runs(parent.get("id") or "", limit=5)
                for pr in p_runs:
                    if pr.get("outcome") == ks.OUTCOME_COMPLETED:
                        lines.append(_format_run_block(pr, max_chars=PER_SECTION_TRUNCATE_HINT))
                        break

    # Children completed (assembly mode — when this is the parent and its
    # children have already finished).
    children = ks.list_children(task_id)
    if children:
        completed_kids = [c for c in children if c.get("status") == ks.STATUS_DONE]
        if completed_kids:
            lines.append("")
            lines.append("─── Completed children (for assembly) ───")
            for child in completed_kids:
                ctitle = _trunc(child.get("title") or "", 200)
                lines.append(f"{child.get('id')} title={ctitle}")
                c_runs = ks.list_runs(child.get("id") or "", limit=5)
                for cr in c_runs:
                    if cr.get("outcome") == ks.OUTCOME_COMPLETED:
                        lines.append(_format_run_block(cr, max_chars=PER_SECTION_TRUNCATE_HINT))
                        break

    # Sibling completed runs (depth-1 only, capped).
    parent_id = task.get("parent_id")
    if parent_id:
        siblings = ks.list_children(parent_id)
        sib_completed = [
            s for s in siblings
            if s.get("status") == ks.STATUS_DONE and s.get("id") != task_id
        ]
        sib_completed = sib_completed[:MAX_SIBLINGS]
        if sib_completed:
            lines.append("")
            lines.append("─── Sibling completed work (most recent) ───")
            for sib in sib_completed:
                stitle = _trunc(sib.get("title") or "", 160)
                lines.append(f"{sib.get('id')} title={stitle}")
                s_runs = ks.list_runs(sib.get("id") or "", limit=3)
                for sr in s_runs:
                    if sr.get("outcome") == ks.OUTCOME_COMPLETED:
                        lines.append(_format_run_block(sr, max_chars=PER_SECTION_TRUNCATE_HINT // 2))
                        break

    block = "\n".join(lines)
    if len(block) > max_chars:
        block = block[: max_chars - 1] + "…"
    return block


def _last_terminal_run(task_id: str) -> dict[str, Any] | None:
    """Most recent terminal run row for *task_id*, or None."""
    for run in ks.list_runs(task_id, limit=10):
        if (run.get("outcome") or "") != ks.OUTCOME_ACTIVE:
            return run
    return None


def _metadata_blockers_list(metadata: dict[str, Any]) -> list[str]:
    blockers = metadata.get("blockers")
    if isinstance(blockers, list):
        return [str(x).strip() for x in blockers if str(x).strip()]
    if isinstance(blockers, str) and blockers.strip():
        return [blockers.strip()]
    return []


def build_retry_worker_brief(task_id: str, *, max_chars: int = 800) -> str:
    """Structured retry brief from the last terminal run (attempt 2+ workers).

    Returns empty string when there is no prior failure signal.

    When the prior terminal run recorded durable artifacts (``metadata.artifact_paths`` —
    which already folds in the scratch path; see ``KanbanRunLedger`` finalize), the brief
    emits an explicit **resume-from-artifact** directive so the retried worker verifies and
    continues from the existing artifacts instead of blindly redoing the work. This closes the
    data-duplication seam that honest-partial preservation introduces (more retries → more risk
    of duplicate file writes, re-posts, or re-created child tasks). The directive is placed ahead
    of the prose summary/error so it survives end-truncation at ``max_chars``.
    """
    task = ks.get_task(task_id)
    if not task:
        return ""

    run = _last_terminal_run(task_id)
    if not run:
        return ""

    metadata = run.get("metadata") if isinstance(run.get("metadata"), dict) else {}
    summary = _trunc(str(run.get("summary") or ""), max(160, max_chars // 3))
    error = _trunc(str(run.get("error") or ""), max(120, max_chars // 4))
    next_action = str(metadata.get("next_action") or "").strip()
    failure_reason = str(metadata.get("failure_reason") or metadata.get("typed_reason") or "").strip()
    tail_excerpt = str(metadata.get("tail_excerpt") or "").strip()
    blockers = _metadata_blockers_list(metadata)
    artifact_paths = metadata.get("artifact_paths")
    paths: list[str] = []
    if isinstance(artifact_paths, list):
        paths = [str(p).strip() for p in artifact_paths if str(p).strip()][:4]
    elif isinstance(artifact_paths, str) and artifact_paths.strip():
        paths = [artifact_paths.strip()]

    if not any((summary, error, next_action, failure_reason, blockers, tail_excerpt, paths)):
        return ""

    lines = [
        "",
        "─── Retry brief (prior attempt failed — RECORDED, not authoritative) ───",
        f"Attempt: {run.get('attempt', '?')}  Outcome: {run.get('outcome', '?')}",
    ]
    if failure_reason:
        lines.append(f"Failure reason: {failure_reason}")
    if blockers:
        lines.append("Blockers: " + ", ".join(blockers[:6]))
    if next_action:
        lines.append(f"Required next action: {_trunc(next_action, max(220, max_chars // 2))}")
    if paths:
        lines.append(
            "Artifact paths (already produced by the prior attempt): "
            + ", ".join(_trunc(p, 80) for p in paths)
        )
        lines.append(
            "RESUME, do not restart: read/verify these existing artifacts and continue from them "
            "to finish. Do NOT re-create files, re-post results, or re-create child tasks that "
            "already exist — redoing completed work produces duplicate side effects."
        )
    if summary:
        lines.append(f"Prior summary: {summary}")
    if error:
        lines.append(f"Prior error: {error}")
    if tail_excerpt and not summary:
        lines.append(f"Prior output excerpt: {_trunc(tail_excerpt, max(280, max_chars // 2))}")
    lines.append(
        "Do not repeat the same observe-only loop. Mutate the artifact or end with "
        "metadata.status=\"blocked\" and a specific next_action if still stuck."
    )

    block = "\n".join(lines)
    if len(block) > max_chars:
        block = block[: max_chars - 1] + "…"
    return block


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

    On any config read error returns False so the fence-primary contract stands.
    """
    try:
        from core.config import get_kanban_config

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


def build_completion_contract() -> str:
    """The standardized 'how to finish' block appended to every kanban
    worker's system prompt.

    The dispatcher parses ``parse_kanban_result`` from the worker's output and
    then runs a strict verifier. Workers may also call the
    ``kanban_post_summary`` claim-scoped tool for in-band progress, but a
    parseable trailing ``kanban-result`` block is required for deterministic
    completion.

    Stage B (``[kanban].native_result_handoff``, default off): when on, the
    contract teaches the native ``kanban_post_summary`` tool as the *preferred*
    handoff channel (structured, truncation-robust) and keeps the trailing
    ``kanban-result`` fence as a fully-supported backup. Both channels complete;
    the dispatcher independently verifies and owns the completion decision. When
    off, the fence-primary contract below is emitted verbatim.
    """
    if _native_result_handoff_enabled():
        return (
            "\n\n"
            "─── How to finish (kanban completion contract) ───\n"
            "Use <think>...</think> only for short private scratch notes.\n"
            "Everything user-visible must be outside thinking tags.\n"
            "PREFERRED — finish by calling the kanban_post_summary tool with your handoff:\n"
            "  kanban_post_summary(summary=\"<dense, factual handoff for downstream workers>\", "
            "metadata={\"changed_files\": [...], \"decisions\": [...]}, status=\"success\")\n"
            "This native tool call is the primary, most reliable handoff: it is structured and "
            "survives output truncation. Call it once you have your final result.\n"
            "BACKUP — you may also (or instead) end your final assistant message with a fenced JSON "
            "block in this exact format (same shape; used automatically if the tool was unavailable):\n"
            "```kanban-result\n"
            '{"summary": "<dense, factual handoff for downstream workers>", '
            '"metadata": {"<free-form keys>": "<values>"}}\n'
            "```\n"
            "- summary: what downstream workers need (decisions, outputs, links, etc.). Concise.\n"
            "- metadata: any structured fields useful for the next stage "
            "(e.g. changed_files, tests_run, tokens_used, decisions). Free-form JSON object.\n"
            "Do not put the kanban-result fence inside <think> tags.\n"
            "Avoid long reasoning preambles like 'Thinking:' before the fence; sanitizer may drop them.\n"
            "The dispatcher independently verifies your handoff and owns the completion decision "
            "either way.\n"
            "\n"
            "When work is incomplete (either channel):\n"
            '- status: \"blocked\" when you cannot finish; \"partial\" when some deliverable exists but verification failed.\n'
            "  (Pass it as the kanban_post_summary status= argument, or as metadata.status in the fence.)\n"
            "- blockers: non-empty array of short machine-readable reason codes (required for salvage).\n"
            "- next_action: one concrete sentence the next worker should execute (required for salvage).\n"
            "- artifact_paths: optional array of primary file/project paths involved.\n"
            "- tools_required_next: optional array of tool names the next attempt should use.\n"
        )
    return (
        "\n\n"
        "─── How to finish (kanban completion contract) ───\n"
        "Use <think>...</think> only for short private scratch notes.\n"
        "Everything user-visible must be outside thinking tags.\n"
        "End your final assistant message with a fenced JSON block in this exact format:\n"
        "```kanban-result\n"
        '{"summary": "<dense, factual handoff for downstream workers>", '
        '"metadata": {"<free-form keys>": "<values>"}}\n'
        "```\n"
        "- summary: what downstream workers need (decisions, outputs, links, etc.). Concise.\n"
        "- metadata: any structured fields useful for the next stage "
        "(e.g. changed_files, tests_run, tokens_used, decisions). Free-form JSON object.\n"
        "Do not put the kanban-result fence inside <think> tags.\n"
        "Avoid long reasoning preambles like 'Thinking:' before the fence; sanitizer may drop them.\n"
        "You may also call kanban_post_summary(summary, metadata_json) for in-band handoff state.\n"
        "Important: a parseable trailing kanban-result block is still required for completion.\n"
        "\n"
        "When work is incomplete, use salvage metadata (still requires the fence above):\n"
        '- metadata.status: \"blocked\" when you cannot finish; \"partial\" when some deliverable exists but verification failed.\n'
        "- blockers: non-empty array of short machine-readable reason codes (required for salvage).\n"
        "- next_action: one concrete sentence the next worker should execute (required for salvage).\n"
        "- artifact_paths: optional array of primary file/project paths involved.\n"
        "- tools_required_next: optional array of tool names the next attempt should use.\n"
    )


# ---------------------------------------------------------------------------
# Result block parser (used by dispatcher to extract worker handoff)
# ---------------------------------------------------------------------------

_FENCE_OPENS = ("```kanban-result", "```kanban_result", "```KANBAN-RESULT")
_FENCE_CLOSE = "```"

# Canonical typed failure reasons shared by parser/dispatcher/loop.
FAILURE_CONTRACT_MISSING = "contract_missing"
FAILURE_CONTRACT_UNCLOSED = "contract_unclosed"
FAILURE_CONTRACT_EMPTY = "contract_empty"
FAILURE_CONTRACT_INVALID_JSON = "contract_invalid_json"
FAILURE_CONTRACT_SCHEMA_INVALID = "contract_schema_invalid"
FAILURE_VERIFIER_FAILED = "verifier_failed"
FAILURE_MISSING_CAPABILITY = "missing_capability"
FAILURE_POLICY_BLOCKED_CAPABILITY = "policy_blocked_capability"
FAILURE_TOOL_RUNTIME_ERROR = "tool_runtime_error"
FAILURE_TOOL_TIMEOUT = "tool_timeout"
FAILURE_UNKNOWN_TOOL = "unknown_tool"
FAILURE_UNKNOWN_ASSIGNEE = "unknown_assignee"
FAILURE_DEADLINE_EXPIRED = "deadline_expired"
FAILURE_MESH_PEER_ERROR = "mesh_peer_error"
FAILURE_SPAWN_FAILED = "spawn_failed"

# Run metadata: distinguish infra lies from model/policy mistakes (WCR-11).
FAILURE_CLASS_MODEL = "model"
FAILURE_CLASS_POLICY = "policy"
FAILURE_CLASS_INFRA = "infra"
FAILURE_CLASS_OPERATOR = "operator"

_FAILURE_CLASS_BY_REASON: dict[str, str] = {
    FAILURE_CONTRACT_MISSING: FAILURE_CLASS_MODEL,
    FAILURE_CONTRACT_UNCLOSED: FAILURE_CLASS_MODEL,
    FAILURE_CONTRACT_EMPTY: FAILURE_CLASS_MODEL,
    FAILURE_CONTRACT_INVALID_JSON: FAILURE_CLASS_MODEL,
    FAILURE_CONTRACT_SCHEMA_INVALID: FAILURE_CLASS_MODEL,
    FAILURE_VERIFIER_FAILED: FAILURE_CLASS_MODEL,
    FAILURE_UNKNOWN_TOOL: FAILURE_CLASS_MODEL,
    FAILURE_MISSING_CAPABILITY: FAILURE_CLASS_POLICY,
    FAILURE_POLICY_BLOCKED_CAPABILITY: FAILURE_CLASS_POLICY,
    FAILURE_UNKNOWN_ASSIGNEE: FAILURE_CLASS_POLICY,
    FAILURE_DEADLINE_EXPIRED: FAILURE_CLASS_POLICY,
    FAILURE_MESH_PEER_ERROR: FAILURE_CLASS_INFRA,
    FAILURE_SPAWN_FAILED: FAILURE_CLASS_INFRA,
    FAILURE_TOOL_RUNTIME_ERROR: FAILURE_CLASS_INFRA,
    FAILURE_TOOL_TIMEOUT: FAILURE_CLASS_INFRA,
}


_TYPED_REASON_LABELS: dict[str, str] = {
    FAILURE_CONTRACT_MISSING: "Missing completion block",
    FAILURE_CONTRACT_UNCLOSED: "Unclosed completion block",
    FAILURE_CONTRACT_EMPTY: "Empty completion block",
    FAILURE_CONTRACT_INVALID_JSON: "Invalid completion JSON",
    FAILURE_CONTRACT_SCHEMA_INVALID: "Invalid completion schema",
    # NB: this reason fires for *validation* failures of a seat's handoff (malformed
    # summary/metadata, unknown claim_id, bad contrarian shape, validation exceptions) —
    # NOT for a valid semantic rejection (those reach `done`). Keep the label about the
    # handoff failing verification, never "the Verifier rejected", which implies a verdict.
    FAILURE_VERIFIER_FAILED: "Handoff failed verification",
    FAILURE_MISSING_CAPABILITY: "Missing required tool capability",
    FAILURE_POLICY_BLOCKED_CAPABILITY: "Policy blocked a required tool",
    FAILURE_UNKNOWN_TOOL: "Unknown tool referenced",
    FAILURE_TOOL_TIMEOUT: "Worker timed out",
    FAILURE_TOOL_RUNTIME_ERROR: "Tool runtime error",
    FAILURE_UNKNOWN_ASSIGNEE: "Unknown worker assignee",
    FAILURE_DEADLINE_EXPIRED: "Task deadline expired",
    FAILURE_MESH_PEER_ERROR: "Mesh peer error",
    FAILURE_SPAWN_FAILED: "Worker failed to start",
    "cancelled": "Cancelled by operator",
    "cancelled_before_spawn": "Cancelled before worker started",
    "cancelled_during_worker_run": "Cancelled during worker run",
    "cancelled_before_mesh_response": "Cancelled before mesh response",
}


def humanize_typed_failure_reason(reason: str) -> str:
    """Operator-facing label for ``typed_reason`` / ``failure_reason`` codes."""
    key = str(reason or "").strip()
    if not key:
        return ""
    if key in _TYPED_REASON_LABELS:
        return _TYPED_REASON_LABELS[key]
    if key.startswith("[typed_failure:"):
        inner = key.split("]", 1)[-1].strip()
        if inner:
            return _trunc(inner, 120)
    return key.replace("_", " ").strip().capitalize()


def classify_failure_class(failure_reason: str) -> str:
    """Map ``failure_reason`` to model / policy / infra / operator."""
    key = str(failure_reason or "").strip()
    if not key:
        return FAILURE_CLASS_INFRA
    if key in (
        "cancelled",
        "cancelled_before_spawn",
        "cancelled_during_worker_run",
        "cancelled_before_mesh_response",
    ):
        return FAILURE_CLASS_OPERATOR
    if key in (
        "record_outcome_persistence_failed",
        "claim_ttl_exceeded",
        FAILURE_SPAWN_FAILED,
        FAILURE_MESH_PEER_ERROR,
    ):
        return FAILURE_CLASS_INFRA
    return _FAILURE_CLASS_BY_REASON.get(key, FAILURE_CLASS_INFRA)


@dataclass(frozen=True)
class KanbanParseResult:
    parsed: dict[str, Any] | None
    excerpt: str
    reason: str
    retryable: bool


def parse_kanban_result(text: str) -> tuple[dict[str, Any] | None, str]:
    """Parse the trailing ``kanban-result`` fenced block out of *text*.

    Returns (parsed_obj, raw_excerpt). ``parsed_obj`` is None when no block
    is present or the block isn't valid JSON. ``raw_excerpt`` is a
    cap-trimmed snippet of the worker's output for activity-bus debugging
    when parsing fails.
    """
    result = parse_kanban_result_detailed(text)
    return result.parsed, result.excerpt


def parse_kanban_result_detailed(text: str) -> KanbanParseResult:
    """Detailed parser variant that returns typed failure reasons."""
    if not isinstance(text, str) or not text:
        return KanbanParseResult(
            parsed=None,
            excerpt="",
            reason=FAILURE_CONTRACT_MISSING,
            retryable=False,
        )
    excerpt = text[-500:]

    # Find the LAST opening fence (worker may have an example earlier).
    last_open_idx = -1
    last_open_marker = ""
    for marker in _FENCE_OPENS:
        idx = text.rfind(marker)
        if idx > last_open_idx:
            last_open_idx = idx
            last_open_marker = marker
    if last_open_idx < 0:
        return KanbanParseResult(
            parsed=None,
            excerpt=excerpt,
            reason=FAILURE_CONTRACT_MISSING,
            retryable=False,
        )

    body_start = last_open_idx + len(last_open_marker)
    # Skip the trailing newline after the open fence.
    if body_start < len(text) and text[body_start] == "\n":
        body_start += 1

    close_len = len(_FENCE_CLOSE)
    close_indices: list[int] = []
    search_from = body_start
    while True:
        idx = text.find(_FENCE_CLOSE, search_from)
        if idx < 0:
            break
        close_indices.append(idx)
        search_from = idx + close_len
    if not close_indices:
        return KanbanParseResult(
            parsed=None,
            excerpt=excerpt,
            reason=FAILURE_CONTRACT_UNCLOSED,
            retryable=False,
        )

    # Prefer the last closing fence whose body parses as kanban-result JSON.
    # Global rfind breaks when prose after the block contains triple backticks;
    # first-close breaks when the body itself contains inline fenced examples.
    parsed: dict[str, Any] | None = None
    for close_idx in reversed(close_indices):
        payload = text[body_start:close_idx].strip()
        if not payload:
            continue
        try:
            candidate = json.loads(payload)
        except Exception:
            continue
        if not isinstance(candidate, dict):
            continue
        summary = candidate.get("summary")
        metadata = candidate.get("metadata")
        parsed = {
            "summary": str(summary or "").strip(),
            "metadata": metadata if isinstance(metadata, dict) else {},
        }
        break

    if parsed is None:
        first_payload = text[body_start:close_indices[0]].strip()
        if not first_payload:
            return KanbanParseResult(
                parsed=None,
                excerpt=excerpt,
                reason=FAILURE_CONTRACT_EMPTY,
                retryable=False,
            )
        try:
            bad = json.loads(first_payload)
        except Exception:
            return KanbanParseResult(
                parsed=None,
                excerpt=excerpt,
                reason=FAILURE_CONTRACT_INVALID_JSON,
                retryable=False,
            )
        if not isinstance(bad, dict):
            return KanbanParseResult(
                parsed=None,
                excerpt=excerpt,
                reason=FAILURE_CONTRACT_SCHEMA_INVALID,
                retryable=False,
            )
        return KanbanParseResult(
            parsed=None,
            excerpt=excerpt,
            reason=FAILURE_CONTRACT_INVALID_JSON,
            retryable=False,
        )

    return KanbanParseResult(
        parsed=parsed,
        excerpt=excerpt,
        reason="ok",
        retryable=False,
    )



def _inherit_metadata_keys(acceptance: dict[str, Any] | None) -> frozenset[str]:
    """Keys from task metadata that satisfy require_metadata_fields when absent in result."""
    if not isinstance(acceptance, dict):
        return frozenset()
    raw = acceptance.get("inherit_metadata_fields")
    if isinstance(raw, list):
        return frozenset(str(x).strip() for x in raw if str(x).strip())
    return frozenset()


def _format_acceptance_for_worker(raw: Any) -> str:
    if isinstance(raw, list) and raw:
        return "; ".join(_trunc(str(x), 160) for x in raw)
    if isinstance(raw, str) and raw.strip():
        return _trunc(raw, 480)
    if isinstance(raw, dict) and raw:
        try:
            return _trunc(json.dumps(raw, sort_keys=True), 480)
        except (TypeError, ValueError):
            return _trunc(str(raw), 480)
    return ""


def _result_metadata_has(
    key: str,
    result_md: dict[str, Any],
    task_md: dict[str, Any],
    inherit_keys: frozenset[str],
) -> bool:
    k = str(key or "").strip()
    if not k:
        return True
    if k in result_md and result_md.get(k) not in (None, ""):
        return True
    if k in inherit_keys and k in task_md and task_md.get(k) not in (None, ""):
        return True
    return False


def _effective_result_metadata(
    result_md: dict[str, Any],
    task_md: dict[str, Any],
    inherit_keys: frozenset[str],
) -> dict[str, Any]:
    merged = dict(result_md)
    for key in inherit_keys:
        if key not in merged and key in task_md:
            merged[key] = task_md[key]
    return merged


_DEFERRED_WORK_PHRASES: tuple[str, ...] = (
    "ready to proceed",
    "ready to begin",
    # "phase 2" must be anchored to a deferral verb: a bare substring collides
    # with deliverable summaries that legitimately enumerate documented "Phase 2"
    # sections (e.g. a doc's "Phase 2 — Tool lookup"). Only flag the worker
    # deferring its OWN work to a later phase.
    "to phase 2",
    "begin phase 2",
    "beginning phase 2",
    "start phase 2",
    "starting phase 2",
    "next step:",
    "next steps:",
    "will now",
    "about to start",
    "proceeding to step",
    "once approved",
    "starting investigation",
    "locating files",
    "current status:",
)

_NON_TERMINAL_META_STATUSES = frozenset({
    "in_progress",
    "pending",
    "starting",
    "locating",
    "rendering",
    "running",
    "queued",
    "waiting",
    "processing",
})

_TERMINAL_SALVAGE_META_STATUSES = frozenset({
    "partial",
    "blocked",
})

_COUNCIL_VERIFIER_VERDICT_FAILURE_STATUSES = frozenset({
    "failed",
    "error",
    "crashed",
    "gave_up",
    "rejected",
    "incomplete",
})

_FAILURE_META_STATUSES = frozenset({
    "failed",
    "error",
    "crashed",
    "gave_up",
})

DEFAULT_KANBAN_WORKER_ACCEPTANCE: dict[str, Any] = {
    "min_summary_chars": 48,
    "reject_deferred_prose": True,
    "forbid_status": sorted(_NON_TERMINAL_META_STATUSES | _FAILURE_META_STATUSES),
}

_TASK_KIND_ACCEPTANCE_HINTS: dict[str, dict[str, Any]] = {
    "research": {
        "require_tools_any": [
            "read_file",
            "read_file_chunk",
            "deep_web_search",
            "web_search",
            "read_webpage",
            "query_knowledge",
            "analyze_multiple_sources",
        ],
    },
    "mutate": {
        "require_tools_any": [
            "write_text_file",
            "save_to_file",
            "append_to_text_file",
            "save_research_findings",
        ],
    },
    "verify": {
        "require_tools_any_suffix": ["_lint"],
    },
    "deliver": {
        "require_metadata_nonempty_any": ["output", "artifact_paths"],
    },
}

# ``video`` / ``video_encode`` inherit deliver metadata proof only.
# ``video_render`` adds ``*_render`` tool evidence (Hyperframes and similar).
_DELIVERABLE_TYPE_ACCEPTANCE_HINTS: dict[str, dict[str, Any]] = {
    "video_render": {
        "require_tools_any_suffix": ["_render"],
    },
    "web_act": {
        "require_tools_any": ["write_web_act_file", "run_web_script"],
        "require_metadata_nonempty_any": ["output", "artifact_paths", "web_act_run_dir"],
    },
}


def task_metadata_for_verification(task: dict[str, Any]) -> dict[str, Any]:
    """Task metadata dict passed to ``verify_kanban_result`` / salvage helpers."""
    meta = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
    verify_meta = dict(meta)
    verify_meta.setdefault("title", task.get("title") or "")
    verify_meta.setdefault("body", task.get("body") or "")
    tid = str(task.get("id") or "").strip()
    if tid:
        verify_meta.setdefault("_task_id", tid)
    assignee = str(task.get("assignee") or "").strip()
    if assignee:
        verify_meta["_assignee"] = assignee
    return verify_meta


KANBAN_WORKER_INFRA_TOOLS = frozenset({
    "kanban_post_progress",
    "kanban_post_summary",
    "append_delegation_scratch",
    "read_delegation_scratch",
})

_REGISTERED_TOOL_NAMES_CACHE: frozenset[str] | None = None

_COUNCIL_ASSIGNEE_PREFIX = "council_"


def _registered_tool_names() -> frozenset[str]:
    global _REGISTERED_TOOL_NAMES_CACHE
    if _REGISTERED_TOOL_NAMES_CACHE is not None:
        return _REGISTERED_TOOL_NAMES_CACHE
    try:
        from tools import get_tools

        _REGISTERED_TOOL_NAMES_CACHE = frozenset(
            str(getattr(t, "name", "") or "").strip()
            for t in get_tools(include_media=True, include_cli=False)
            if str(getattr(t, "name", "") or "").strip()
        )
    except Exception:
        _log.warning("registered tool names lookup failed", exc_info=True)
        _REGISTERED_TOOL_NAMES_CACHE = frozenset()
    return _REGISTERED_TOOL_NAMES_CACHE


def _intersect_registered_tool_names(names: set[str]) -> set[str]:
    if not names:
        return set()
    registered = _registered_tool_names()
    if not registered:
        return set(names)
    return names & registered


def is_council_assignee(assignee: str) -> bool:
    return str(assignee or "").strip().startswith(_COUNCIL_ASSIGNEE_PREFIX)


def _is_council_metadata(metadata: dict[str, Any] | None) -> bool:
    try:
        from utils.council_orchestrator import is_council_task_metadata

        return is_council_task_metadata(metadata)
    except Exception:
        return False


def _effective_profile_assignee(assignee: str, metadata: dict[str, Any] | None = None) -> str:
    """Resolve triage → specifier profile for tool/profile lookups."""
    a = str(assignee or "general").strip() or "general"
    if a != "triage":
        return a
    md = metadata if isinstance(metadata, dict) else {}
    spec = str(md.get("specifier_profile") or "").strip()
    return spec or a


def profile_tool_names(
    assignee: str,
    *,
    metadata: dict[str, Any] | None = None,
) -> frozenset[str] | None:
    """Profile tool allowlist for *assignee*; ``None`` means full tool surface (``general``)."""
    from subagent_runner import SUBAGENT_PROFILES

    effective = _effective_profile_assignee(assignee, metadata)
    if effective == "council_verifier":
        md = metadata if isinstance(metadata, dict) else {}
        from utils.council_orchestrator import (
            COUNCIL_FRAME_META,
            council_verifier_tool_names_for_frame,
        )

        frame = str(md.get(COUNCIL_FRAME_META) or "").strip()
        return council_verifier_tool_names_for_frame(frame)
    if effective not in SUBAGENT_PROFILES:
        # mesh:* and other non-profile assignees — fail-closed for prompt/tool checks.
        return frozenset()
    profile = SUBAGENT_PROFILES[effective]
    tools = profile.get("tools")
    if tools is None:
        return None
    return frozenset(str(t) for t in tools if str(t).strip())


def assignee_has_tool(
    assignee: str,
    tool_name: str,
    *,
    metadata: dict[str, Any] | None = None,
) -> bool:
    """True when *assignee* may invoke *tool_name* (``general`` → any tool)."""
    tools = profile_tool_names(assignee, metadata=metadata)
    if tools is None:
        return True
    return str(tool_name or "").strip() in tools


def clamp_metadata_tool_names(
    metadata: dict[str, Any],
    assignee: str,
    *,
    log: logging.Logger | None = None,
) -> dict[str, Any]:
    """Narrow-only ``metadata.tool_names``; council seats ignore widening overrides."""
    md = dict(metadata or {})
    if is_council_assignee(assignee) or _is_council_metadata(md):
        if md.get("tool_names"):
            (log or _log).info(
                "ignoring metadata.tool_names for council assignee %s",
                assignee,
            )
            md.pop("tool_names", None)
        return md
    raw = md.get("tool_names")
    if not isinstance(raw, list) or not raw:
        return md
    requested = {str(x).strip() for x in raw if str(x).strip()}
    if not requested:
        md.pop("tool_names", None)
        return md
    profile_tools = profile_tool_names(assignee, metadata=md)
    if profile_tools is None:
        clamped = sorted(_intersect_registered_tool_names(requested))
    else:
        clamped = sorted(requested & profile_tools)
    if not clamped:
        (log or _log).warning(
            "metadata.tool_names had no overlap with %s profile; dropping override",
            assignee,
        )
        md.pop("tool_names", None)
        return md
    if set(clamped) != requested:
        (log or _log).info(
            "metadata.tool_names narrowed for %s: %s → %s",
            assignee,
            sorted(requested),
            clamped,
        )
    md["tool_names"] = clamped
    return md


def reconcile_worker_task_kind(
    metadata: dict[str, Any],
    assignee: str,
    *,
    log: logging.Logger | None = None,
) -> dict[str, Any]:
    """Drop incoherent ``task_kind: research`` when assignee is not the research profile."""
    md = dict(metadata or {})
    kind = str(md.get("task_kind") or "").strip().lower()
    if kind != "research":
        return md
    effective = _effective_profile_assignee(assignee, md)
    if effective == "research":
        return md
    if is_council_assignee(effective) or _is_council_metadata(md):
        return md
    (log or _log).info(
        "dropping task_kind=research for assignee %s (effective %s)",
        assignee,
        effective,
    )
    md.pop("task_kind", None)
    return md


def _guidance_references_unavailable_tools(
    guidance: str,
    assignee: str,
    *,
    metadata: dict[str, Any] | None = None,
) -> bool:
    """True when *guidance* names tools the effective assignee cannot invoke."""
    wg_lower = str(guidance or "").lower()
    if not wg_lower:
        return False
    checks: tuple[tuple[str, str], ...] = (
        ("hyperframes_render", "hyperframes_render"),
        ("hyperframes_lint", "hyperframes_lint"),
        ("hyperframes lint", "hyperframes_lint"),
        ("write_web_act_file", "write_web_act_file"),
        ("run_web_script", "run_web_script"),
        ("deep_web_search", "deep_web_search"),
        ("read_and_analyze_page", "read_and_analyze_page"),
        ("append_research_scratch", "append_research_scratch"),
    )
    for needle, tool in checks:
        if needle in wg_lower and not assignee_has_tool(
            assignee, tool, metadata=metadata
        ):
            return True
    return False


def reconcile_worker_guidance(
    metadata: dict[str, Any],
    assignee: str,
    *,
    log: logging.Logger | None = None,
) -> dict[str, Any]:
    """Drop ``worker_guidance`` that references tools outside the assignee profile."""
    md = dict(metadata or {})
    wg = str(md.get("worker_guidance") or md.get("worker_hints") or "").strip()
    if not wg:
        return md
    worker_assignee = _effective_profile_assignee(assignee, md)
    if _guidance_references_unavailable_tools(wg, worker_assignee, metadata=md):
        (log or _log).info(
            "dropping worker_guidance for assignee %s (references unavailable tools)",
            worker_assignee,
        )
        md.pop("worker_guidance", None)
        md.pop("worker_hints", None)
    return md


def normalize_worker_task_metadata(
    metadata: dict[str, Any],
    assignee: str,
    *,
    log: logging.Logger | None = None,
) -> dict[str, Any]:
    """Clamp tool overrides and reconcile task_kind at task create time."""
    md = clamp_metadata_tool_names(metadata, assignee, log=log)
    md = reconcile_worker_task_kind(md, assignee, log=log)
    return reconcile_worker_guidance(md, assignee, log=log)


def resolve_effective_worker_tools(
    assignee: str,
    metadata: dict[str, Any] | None,
    *,
    infra_tools: frozenset[str] | None = None,
) -> frozenset[str] | None:
    """Narrowed tool override for dispatcher spawn, or ``None`` for profile default."""
    infra = infra_tools or KANBAN_WORKER_INFRA_TOOLS
    md = metadata if isinstance(metadata, dict) else {}
    assignee_s = str(assignee or "").strip()
    if assignee_s == "council_verifier":
        from utils.council_orchestrator import (
            COUNCIL_FRAME_META,
            council_verifier_tool_names_for_frame,
        )

        frame = str(md.get(COUNCIL_FRAME_META) or "").strip()
        return council_verifier_tool_names_for_frame(frame)
    if is_council_assignee(assignee) or _is_council_metadata(md):
        return None
    raw = md.get("tool_names")
    if not isinstance(raw, list) or not raw:
        return None
    requested = {str(x).strip() for x in raw if str(x).strip()}
    if not requested:
        return None
    profile_tools = profile_tool_names(assignee, metadata=md)
    if profile_tools is None:
        narrowed = _intersect_registered_tool_names(requested)
    else:
        narrowed = requested & profile_tools
    if not narrowed:
        return None
    return frozenset(narrowed) | infra


def worker_core_diet_extra_tools(already_loaded: "set[str] | None" = None) -> list[str]:
    """Extra tool names a dieted kanban worker must carry natively from round 1.

    The core diet (tool-schema subsetting) drops everything outside CORE_TOOLS
    plus the working set, but a worker cannot afford a find_tools hop for
    capabilities its task contractually needs: the claim-scoped worker verbs
    (KANBAN_WORKER_INFRA_TOOLS) and the task's preflight-guaranteed
    ``metadata.required_tools``. Reads the bound claim token; returns [] when
    no claim is bound. Never raises.
    """
    seen = set(already_loaded or ())
    extras: list[str] = []
    for name in sorted(KANBAN_WORKER_INFRA_TOOLS):
        if name not in seen:
            extras.append(name)
            seen.add(name)
    try:
        from utils.kanban_dispatcher import current_claim_token

        claim = current_claim_token()
        if isinstance(claim, tuple) and len(claim) == 2 and claim[0]:
            import kanban_storage as ks

            task = ks.get_task(claim[0]) or {}
            md = task.get("metadata")
            req = md.get("required_tools") if isinstance(md, dict) else None
            for raw in (req if isinstance(req, list) else []):
                name = str(raw or "").strip()
                if name and name not in seen:
                    extras.append(name)
                    seen.add(name)
    except Exception:
        _log.debug("worker diet required_tools merge failed", exc_info=True)
    return extras


def _acceptance_requires_tool_suffix(
    task_metadata: dict[str, Any] | None,
    suffix: str,
) -> bool:
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    acceptance = tm.get("acceptance") if isinstance(tm.get("acceptance"), dict) else {}
    sfx = str(suffix or "").strip().lower()
    if not sfx.startswith("_"):
        sfx = f"_{sfx.lstrip('_')}"
    for raw in acceptance.get("require_tools_any_suffix") or []:
        token = str(raw or "").strip().lower()
        if token == sfx or token.lstrip("_") == sfx.lstrip("_"):
            return True
    for key in ("require_tools_any", "require_tools"):
        for raw in acceptance.get(key) or []:
            name = str(raw or "").strip().lower()
            if name.endswith(sfx):
                return True
    return False


def kanban_task_expects_lint(task_metadata: dict[str, Any] | None) -> bool:
    """True when this task's acceptance implies lint/verify tooling."""
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    if str(tm.get("task_kind") or "").strip().lower() == "verify":
        return True
    if str(tm.get("deliverable_type") or "").strip().lower() == "video_render":
        return True
    return _acceptance_requires_tool_suffix(tm, "_lint")


def kanban_task_expects_render(task_metadata: dict[str, Any] | None) -> bool:
    """True when this task's acceptance implies a render deliverable."""
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    if str(tm.get("deliverable_type") or "").strip().lower() == "video_render":
        return True
    return _acceptance_requires_tool_suffix(tm, "_render")


def kanban_task_is_web_act_deliverable(task_metadata: dict[str, Any] | None) -> bool:
    """True when task metadata declares deliverable_type web_act."""
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    return str(tm.get("deliverable_type") or "").strip().lower() == "web_act"


def kanban_task_expects_web_verify(task_metadata: dict[str, Any] | None) -> bool:
    """True when this task requires a fresh_run web-act verification pass."""
    return kanban_task_is_web_act_deliverable(task_metadata)


def kanban_worker_prefers_verify_loop(
    task_metadata: dict[str, Any] | None,
    *,
    wrote_text: bool,
    tools_executed: set[str] | frozenset[str] | list[str],
) -> bool:
    """Use verify-loop hard-stop once lint/render tools ran on a video deliverable.

    Render workers often probe with read_file before patching HTML; counting those
    as observe-only caused read_file×6 hard-stops after hyperframes_lint.
    """
    if wrote_text:
        return True
    executed = {str(n).strip() for n in tools_executed if str(n).strip()}
    if kanban_task_is_web_act_deliverable(task_metadata):
        return "run_web_script" in executed or "write_web_act_file" in executed
    if not kanban_task_is_video_render_deliverable(task_metadata):
        return False
    return any(n.endswith("_lint") or n.endswith("_render") for n in executed)


def kanban_task_is_video_render_deliverable(task_metadata: dict[str, Any] | None) -> bool:
    """True only for Hyperframes-style deliver tasks (not PDF/document deliverables)."""
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    if str(tm.get("task_kind") or "").strip().lower() != "deliver":
        return False
    return str(tm.get("deliverable_type") or "").strip().lower() == "video_render"


def _acceptance_hints_for_task_kind(task_kind: str) -> dict[str, Any]:
    kind = str(task_kind or "").strip().lower()
    hints = _TASK_KIND_ACCEPTANCE_HINTS.get(kind)
    return dict(hints) if isinstance(hints, dict) else {}


def _acceptance_hints_for_deliverable_type(deliverable_type: str) -> dict[str, Any]:
    dt = str(deliverable_type or "").strip().lower()
    hints = _DELIVERABLE_TYPE_ACCEPTANCE_HINTS.get(dt)
    return dict(hints) if isinstance(hints, dict) else {}


def _merge_acceptance_list_keys(
    merged: dict[str, Any],
    *,
    base: dict[str, Any],
    kind_hints: dict[str, Any],
    deliverable_hints: dict[str, Any],
    existing: dict[str, Any],
) -> None:
    for key in (
        "forbid_status",
        "require_metadata_fields",
        "require_tools",
        "require_tools_any",
        "require_tools_any_suffix",
        "require_metadata_nonempty",
        "require_metadata_nonempty_any",
    ):
        base_vals = (
            base.get(key)
            or kind_hints.get(key)
            or deliverable_hints.get(key)
        )
        over_vals = existing.get(key)
        if isinstance(base_vals, list) and isinstance(over_vals, list):
            merged[key] = list(dict.fromkeys([*base_vals, *over_vals]))


def stamp_default_task_acceptance(metadata: dict[str, Any] | None) -> dict[str, Any]:
    """Merge baseline acceptance criteria into task metadata (caller wins on keys)."""
    md = dict(metadata or {})
    base = dict(DEFAULT_KANBAN_WORKER_ACCEPTANCE)
    kind_hints = _acceptance_hints_for_task_kind(str(md.get("task_kind") or ""))
    deliverable_hints = _acceptance_hints_for_deliverable_type(str(md.get("deliverable_type") or ""))
    existing = md.get("acceptance")
    if isinstance(existing, list) and existing:
        merged = {**base, **kind_hints, **deliverable_hints}
        markers = [str(x).strip() for x in existing if str(x).strip()]
        if markers:
            merged["require_transcript_markers"] = markers
        md["acceptance"] = merged
    elif isinstance(existing, dict) and existing:
        merged = {**base, **kind_hints, **deliverable_hints, **existing}
        _merge_acceptance_list_keys(
            merged,
            base=base,
            kind_hints=kind_hints,
            deliverable_hints=deliverable_hints,
            existing=existing,
        )
        md["acceptance"] = merged
    else:
        merged = {**base, **kind_hints, **deliverable_hints}
        md["acceptance"] = merged
    dt = str(md.get("deliverable_type") or "").strip().lower()
    if dt == "web_act" and not md.get("result_required_metadata_fields"):
        md["result_required_metadata_fields"] = [
            "answer",
            "web_act_run_dir",
            "artifact_paths",
        ]
    return md

# Backward-compatible alias for callers/tests.
_IN_PROGRESS_META_STATUSES = _NON_TERMINAL_META_STATUSES

_SUMMARY_FAILURE_PHRASES: tuple[str, ...] = (
    "could not be completed",
    "could not complete",
    "was not completed",
    "unable to complete",
    "unable to finish",
    "did not complete",
    "never completed",
    "without completing",
    "could not lint",
    "linting could not",
    "cannot pass lint",
    "failed to lint",
    "failed to render",
    "render failed",
    "render was not",
    "not successfully rendered",
    "render blocked",
    "validation errors (render blocked)",
)


def _summary_defers_work(summary: str) -> bool:
    lower = summary.lower()
    return any(p in lower for p in _DEFERRED_WORK_PHRASES)


def _summary_admits_failure(summary: str) -> bool:
    lower = (summary or "").strip().lower()
    if not lower:
        return False
    return any(p in lower for p in _SUMMARY_FAILURE_PHRASES)


def _metadata_value_nonempty(val: Any) -> bool:
    if val is None or val == "" or val == {}:
        return False
    if isinstance(val, list):
        return any(
            _metadata_value_nonempty(item)
            if isinstance(item, (list, dict))
            else bool(str(item).strip())
            for item in val
        )
    if isinstance(val, str):
        return bool(val.strip())
    return True


def _metadata_key_nonempty(md: dict[str, Any], key: str) -> bool:
    k = str(key or "").strip()
    if not k or k not in md:
        return False
    return _metadata_value_nonempty(md.get(k))


_OUTPUT_PATH_METADATA_ALIASES: tuple[str, ...] = (
    "output_path",
    "output_file",
    "render_output",
    "file_path",
)


def normalize_kanban_result_metadata(
    metadata: dict[str, Any] | None,
    *,
    task_metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Map common render handoff aliases onto ``output`` / ``artifact_paths``.

    Workers often put the MP4 path in ``output_path`` (matching tool prose) while
    acceptance checks ``require_metadata_nonempty_any: [output, artifact_paths]``.
    """
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    md = dict(metadata) if isinstance(metadata, dict) else {}
    seat = str(tm.get("council_seat") or md.get("council_seat") or "").strip().lower()
    if tm.get("origin") == "council" and seat == "contrarian":
        raw_assumptions = md.get("assumptions_challenged")
        if isinstance(raw_assumptions, str):
            text = raw_assumptions.strip()
            if text:
                md["assumptions_challenged"] = [text]
    if tm.get("origin") == "council" and seat == "verifier":
        # Verifier models describe an all-claims-rejected VERDICT as status "failed";
        # that is a successful verification run, not a worker failure. Strip the
        # failure-ish status whenever a structured verdict is present so the seat is
        # not failed at the contract gate (kcn_c32156cfe1b340e6). "blocked" is kept —
        # it genuinely means the claims could not be evaluated.
        v_status = str(md.get("status") or "").strip().lower()
        has_verdict = isinstance(md.get("verified_claims"), list) or (
            isinstance(md.get("rejections"), list) and md.get("rejections")
        )
        if has_verdict and v_status in _COUNCIL_VERIFIER_VERDICT_FAILURE_STATUSES:
            md.pop("status", None)
    path = ""
    for key in _OUTPUT_PATH_METADATA_ALIASES:
        val = str(md.get(key) or "").strip()
        if val:
            path = val
            break
    if not path:
        return md
    if not _metadata_key_nonempty(md, "output"):
        md["output"] = path
    if not _metadata_key_nonempty(md, "artifact_paths"):
        md["artifact_paths"] = [path]
    elif isinstance(md.get("artifact_paths"), list):
        existing = [str(x).strip() for x in md["artifact_paths"] if str(x).strip()]
        if path not in existing:
            md["artifact_paths"] = [path, *existing][:4]
    # Web-act handoff aliases
    if str(md.get("web_act_answer") or "").strip() and not _metadata_key_nonempty(md, "answer"):
        md["answer"] = str(md.get("web_act_answer") or "").strip()
    run_dir = str(md.get("web_act_run_dir") or md.get("run_dir") or "").strip()
    if run_dir:
        if not _metadata_key_nonempty(md, "web_act_run_dir"):
            md["web_act_run_dir"] = run_dir
        if not _metadata_key_nonempty(md, "output"):
            md["output"] = run_dir
        if not _metadata_key_nonempty(md, "artifact_paths"):
            md["artifact_paths"] = [run_dir]
    return md


def _transcript_has_tool_suffix(tools_blob: str, suffix: str) -> bool:
    sfx = str(suffix or "").strip().lower()
    if not sfx:
        return False
    if not sfx.startswith("_"):
        sfx = f"_{sfx}"
    for token in re.findall(r"[a-z0-9_]+", tools_blob.lower()):
        if token.endswith(sfx):
            return True
    return False


def _tool_names_from_metadata_value(val: Any) -> list[str]:
    if isinstance(val, list):
        return [str(x).strip() for x in val if str(x or "").strip()]
    if isinstance(val, str) and val.strip():
        return [val.strip()]
    return []


def merge_tools_used_metadata(
    metadata: dict[str, Any] | None,
    tool_names: Any,
) -> dict[str, Any]:
    """Union tool names into ``metadata.tools_used`` for acceptance verification."""
    md = dict(metadata) if isinstance(metadata, dict) else {}
    names: set[str] = set()
    for key in ("tools_used", "tools_executed"):
        for name in _tool_names_from_metadata_value(md.get(key)):
            names.add(name.lower())
    if tool_names is not None:
        if isinstance(tool_names, (str, bytes)):
            tool_names = [tool_names]
        for raw in tool_names:
            name = str(raw or "").strip()
            if name:
                names.add(name.lower())
    if names:
        md["tools_used"] = sorted(names)
    return md


def _acceptance_tools_evidence_blob(
    *,
    worker_transcript: str,
    summary: str,
    metadata: dict[str, Any],
) -> str:
    """Transcript + summary + structured tool lists used for acceptance checks."""
    blob = (worker_transcript + " " + summary).lower()
    extra: list[str] = []
    for key in ("tools_used", "tools_executed"):
        for name in _tool_names_from_metadata_value(metadata.get(key)):
            extra.append(name.lower())
    if extra:
        blob = f"{blob} {' '.join(extra)}"
    return blob


def _verify_acceptance_criteria(
    *,
    acceptance: dict[str, Any],
    summary: str,
    metadata: dict[str, Any],
    effective_metadata: dict[str, Any],
    task_metadata: dict[str, Any],
    worker_transcript: str,
) -> tuple[bool, str]:
    """Declarative acceptance checks. Returns ``(ok, detail)``."""
    inherit_keys = _inherit_metadata_keys(acceptance)
    tools_blob = _acceptance_tools_evidence_blob(
        worker_transcript=worker_transcript,
        summary=summary,
        metadata=metadata,
    )
    meta_status = str(metadata.get("status") or "").strip().lower()
    is_terminal_salvage = meta_status in ("partial", "blocked")

    forbid = acceptance.get("forbid_status")
    if isinstance(forbid, list):
        forbidden = {str(x).strip().lower() for x in forbid if str(x).strip()}
        if meta_status in forbidden:
            return False, f"metadata.status={meta_status!r} is not allowed for this handoff"

    min_chars = int(acceptance.get("min_summary_chars") or 0)
    if min_chars > 0 and len(summary) < min_chars:
        return False, f"summary shorter than acceptance min_summary_chars ({min_chars})"

    req_meta = acceptance.get("require_metadata_fields")
    if isinstance(req_meta, list):
        missing = [
            str(k) for k in req_meta
            if str(k) and not _result_metadata_has(str(k), metadata, task_metadata, inherit_keys)
        ]
        if missing:
            return False, "missing required metadata fields: " + ", ".join(missing)

    require_nonempty = acceptance.get("require_metadata_nonempty")
    if isinstance(require_nonempty, list):
        for key in require_nonempty:
            k = str(key or "").strip()
            if not k:
                continue
            if not _metadata_key_nonempty(effective_metadata, k):
                return False, f"metadata.{k} must be non-empty"

    require_nonempty_any = acceptance.get("require_metadata_nonempty_any")
    if isinstance(require_nonempty_any, list) and require_nonempty_any:
        keys = [str(k).strip() for k in require_nonempty_any if str(k).strip()]
        if keys and not any(_metadata_key_nonempty(effective_metadata, k) for k in keys):
            return False, "requires non-empty metadata field (any of): " + ", ".join(keys)

    require_markers = acceptance.get("require_transcript_markers")
    if isinstance(require_markers, list) and require_markers:
        markers = [str(m).strip().lower() for m in require_markers if str(m).strip()]
        if markers and not any(m in tools_blob for m in markers):
            return (
                False,
                "missing required transcript/summary markers: "
                + ", ".join(str(m) for m in require_markers),
            )

    require_all = acceptance.get("require_tools") or []
    require_any = acceptance.get("require_tools_any") or []
    require_any_suffix = acceptance.get("require_tools_any_suffix") or []
    if not is_terminal_salvage:
        if isinstance(require_all, list):
            for tool in require_all:
                tname = str(tool or "").strip()
                if tname and tname not in tools_blob:
                    return False, f"missing required tool evidence for `{tname}`"
        if isinstance(require_any, list) and require_any:
            if not any(str(t or "").strip() in tools_blob for t in require_any):
                return False, "missing required tool evidence (require_tools_any)"
        if isinstance(require_any_suffix, list) and require_any_suffix:
            suffixes = [str(s or "").strip() for s in require_any_suffix if str(s or "").strip()]
            if suffixes and not any(_transcript_has_tool_suffix(tools_blob, s) for s in suffixes):
                return False, "missing required tool evidence (require_tools_any_suffix)"

    reject_deferred = acceptance.get("reject_deferred_prose", True)
    if is_terminal_salvage:
        reject_deferred = False
    if reject_deferred and _summary_defers_work(summary):
        if isinstance(require_any, list) and require_any:
            pass
        elif isinstance(require_all, list) and require_all:
            pass
        elif isinstance(require_any_suffix, list) and require_any_suffix:
            pass
        elif isinstance(require_nonempty_any, list) and require_nonempty_any:
            pass
        else:
            return False, "summary reads as deferred work without deliverable evidence"

    return True, ""


def _verify_web_act_artifacts(
    metadata: dict[str, Any],
    task_metadata: dict[str, Any],
    *,
    worker_transcript: str = "",
) -> tuple[bool, str]:
    """Filesystem + fresh_run checks for web_act deliverable handoffs."""
    import os

    from tools.web_act import validate_web_act_result

    run_dir = str(
        metadata.get("web_act_run_dir") or metadata.get("run_dir") or metadata.get("output") or ""
    ).strip()
    if not run_dir and isinstance(metadata.get("artifact_paths"), list):
        for raw in metadata["artifact_paths"]:
            ps = str(raw or "").strip()
            if not ps:
                continue
            run_dir = os.path.dirname(ps) if ps.endswith((".json", ".png")) else ps
            break
    if not run_dir:
        return False, "web_act handoff missing web_act_run_dir or artifact_paths"

    try:
        from core.config import get_workspace_path

        ws = os.path.abspath(os.path.expanduser(get_workspace_path()))
    except Exception:
        ws = os.path.abspath(os.path.expanduser("~/.mugi/workspace"))
    run_dir_abs = os.path.abspath(os.path.expanduser(run_dir))
    web_runs_root = os.path.join(ws, "web_runs")
    if not run_dir_abs.startswith(web_runs_root + os.sep):
        return False, f"web_act_run_dir must be under {web_runs_root}"
    if "/web_runs/kanban_" not in run_dir_abs.replace("\\", "/"):
        return False, "web_act_run_dir must be under web_runs/kanban_{task_id}/"
    task_id = str(task_metadata.get("_task_id") or task_metadata.get("task_id") or "").strip()
    if task_id:
        expected_scope = f"/web_runs/kanban_{task_id}/"
        if expected_scope not in run_dir_abs.replace("\\", "/") + "/":
            return False, f"web_act_run_dir must belong to task kanban_{task_id}"

    validated = validate_web_act_result(run_dir_abs, fresh_run_expected=True)
    if not validated.get("ok"):
        return False, validated.get("detail") or "web_act result.json validation failed"

    fresh_ok = bool(metadata.get("fresh_run_ok"))
    transcript = worker_transcript or ""
    try:
        from tools.web_act import web_act_tool_output_indicates_fresh_success

        fresh_from_transcript = web_act_tool_output_indicates_fresh_success(transcript)
    except Exception:
        fresh_from_transcript = False
    has_fresh_evidence = fresh_ok or fresh_from_transcript
    if not has_fresh_evidence:
        return False, (
            "web_act completion requires fresh_run_ok in metadata or fresh_run run_web_script "
            "success in worker transcript"
        )
    return True, ""


def verify_kanban_result(
    parsed: dict[str, Any],
    *,
    task_metadata: dict[str, Any] | None = None,
    worker_transcript: str = "",
) -> tuple[bool, str, str]:
    """Strict verifier stage after parser success.

    Returns ``(ok, reason, detail)`` where ``reason`` is typed and stable.
    """
    summary = str((parsed or {}).get("summary") or "").strip()
    raw_metadata = (parsed or {}).get("metadata") or {}
    if not summary:
        return False, FAILURE_VERIFIER_FAILED, "summary is required"
    if len(summary) < 12:
        return False, FAILURE_VERIFIER_FAILED, "summary is too short for downstream handoff"
    if not isinstance(raw_metadata, dict):
        return False, FAILURE_CONTRACT_SCHEMA_INVALID, "metadata must be a JSON object"

    tm = task_metadata if isinstance(task_metadata, dict) else {}
    metadata = normalize_kanban_result_metadata(raw_metadata, task_metadata=tm)
    inherit_keys = _inherit_metadata_keys(
        tm.get("acceptance") if isinstance(tm.get("acceptance"), dict) else None
    )
    effective_metadata = _effective_result_metadata(metadata, tm, inherit_keys)
    required_fields = tm.get("result_required_metadata_fields")
    if isinstance(required_fields, list):
        missing = [
            str(k) for k in required_fields
            if str(k) and not _result_metadata_has(str(k), metadata, tm, inherit_keys)
        ]
        if missing:
            return (
                False,
                FAILURE_VERIFIER_FAILED,
                "missing required metadata fields: " + ", ".join(missing),
            )

    acceptance = tm.get("acceptance")
    meta_status = str(metadata.get("status") or "").strip().lower()
    if meta_status in ("completed", "done", "success"):
        return (
            False,
            FAILURE_VERIFIER_FAILED,
            "metadata.status must be empty for successful completion, not "
            f"{meta_status!r}; use partial or blocked for salvage handoffs",
        )
    if meta_status in _NON_TERMINAL_META_STATUSES:
        return (
            False,
            FAILURE_VERIFIER_FAILED,
            f"kanban-result metadata.status={meta_status!r} is not a completion handoff",
        )
    if meta_status in _FAILURE_META_STATUSES:
        return (
            False,
            FAILURE_VERIFIER_FAILED,
            f"kanban-result metadata.status={meta_status!r} is a failure handoff, not done",
        )
    blockers = metadata.get("blockers")
    has_blockers = (
        (isinstance(blockers, list) and len(blockers) > 0)
        or (isinstance(blockers, str) and blockers.strip())
    )
    if has_blockers and meta_status not in _TERMINAL_SALVAGE_META_STATUSES:
        return (
            False,
            FAILURE_VERIFIER_FAILED,
            "kanban-result metadata.blockers requires metadata.status partial or blocked",
        )
    if meta_status in _TERMINAL_SALVAGE_META_STATUSES:
        next_action = str(metadata.get("next_action") or "").strip()
        if not next_action:
            return (
                False,
                FAILURE_VERIFIER_FAILED,
                "salvage handoff requires metadata.next_action",
            )
        salvage_blockers = metadata.get("blockers")
        has_salvage_blockers = (
            (isinstance(salvage_blockers, list) and len(salvage_blockers) > 0)
            or (isinstance(salvage_blockers, str) and salvage_blockers.strip())
        )
        if not has_salvage_blockers:
            return (
                False,
                FAILURE_VERIFIER_FAILED,
                "salvage handoff requires metadata.blockers",
            )
    if _summary_admits_failure(summary) and meta_status not in _TERMINAL_SALVAGE_META_STATUSES:
        return (
            False,
            FAILURE_VERIFIER_FAILED,
            "summary admits incomplete/failed work without a partial/blocked salvage handoff",
        )

    if isinstance(acceptance, dict):
        ok, detail = _verify_acceptance_criteria(
            acceptance=acceptance,
            summary=summary,
            metadata=metadata,
            effective_metadata=effective_metadata,
            task_metadata=tm,
            worker_transcript=worker_transcript or "",
        )
        if not ok:
            return False, FAILURE_VERIFIER_FAILED, detail

    if kanban_task_is_web_act_deliverable(tm) and meta_status not in _TERMINAL_SALVAGE_META_STATUSES:
        ok_art, art_detail = _verify_web_act_artifacts(
            metadata,
            tm,
            worker_transcript=worker_transcript or "",
        )
        if not ok_art:
            return False, FAILURE_VERIFIER_FAILED, art_detail

    return True, "ok", ""


KANBAN_PHASE_GATHER = "GATHER"
KANBAN_PHASE_MUTATE = "MUTATE"
KANBAN_PHASE_VERIFY = "VERIFY"
KANBAN_PHASE_CLOSE = "CLOSE"


@dataclass(frozen=True)
class KanbanWorkerPhaseSignals:
    """Run-time signals for ``compute_kanban_worker_phase`` (single caller contract)."""

    append_scratch_count: int = 0
    report_generated: bool = False
    wrote_text: bool = False
    render_tools_succeeded: bool = False
    min_scratch_appends: int = 2
    gather_limit_hit: bool = False
    research_gather_chars: int = 0
    research_tools_used: bool = False
    research_gather_tool_count: int = 0
    search_budget_exhausted: bool = False


def resolve_kanban_worker_phase(
    task_metadata: dict[str, Any] | None,
    signals: KanbanWorkerPhaseSignals,
) -> str | None:
    """Derive worker phase for per-round nudges. Never raises — returns None on failure."""
    import inspect

    try:
        sig = inspect.signature(compute_kanban_worker_phase)
        payload = asdict(signals)
        filtered = {key: value for key, value in payload.items() if key in sig.parameters}
        dropped = sorted(set(payload) - set(filtered))
        if dropped:
            _log.warning(
                "kanban worker phase: dropped unknown signals %s (caller/signature drift)",
                dropped,
            )
        return compute_kanban_worker_phase(task_metadata, **filtered)
    except Exception:
        _log.warning("kanban worker phase resolution failed", exc_info=True)
        return None


def compute_kanban_worker_phase(
    task_metadata: dict[str, Any] | None,
    *,
    append_scratch_count: int = 0,
    report_generated: bool = False,
    wrote_text: bool = False,
    render_tools_succeeded: bool = False,
    min_scratch_appends: int = 2,
    gather_limit_hit: bool = False,
    research_gather_chars: int = 0,
    research_tools_used: bool = False,
    research_gather_tool_count: int = 0,
    search_budget_exhausted: bool = False,
) -> str:
    """Derive a coarse worker phase for per-round steering nudges."""
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    try:
        from utils.council_orchestrator import is_council_task_metadata

        if is_council_task_metadata(tm):
            if wrote_text or gather_limit_hit:
                return KANBAN_PHASE_CLOSE
            if str(tm.get("council_seat") or "").strip().lower() == "researcher":
                from agent_loop_guardrails import council_researcher_gather_sufficient
                from utils.council_orchestrator import council_researcher_has_preseed

                if council_researcher_gather_sufficient(
                    research_gather_tool_count,
                    search_budget_exhausted=search_budget_exhausted,
                    has_preseed=council_researcher_has_preseed(tm),
                ):
                    return KANBAN_PHASE_CLOSE
            return KANBAN_PHASE_GATHER
    except Exception:
        pass
    kind = str(tm.get("task_kind") or "").strip().lower()
    expects_lint = kanban_task_expects_lint(tm)
    expects_render = kanban_task_expects_render(tm)
    expects_web = kanban_task_expects_web_verify(tm)
    if kanban_task_is_web_act_deliverable(tm):
        if not wrote_text:
            return KANBAN_PHASE_MUTATE
        if expects_web and not render_tools_succeeded:
            return KANBAN_PHASE_VERIFY
        return KANBAN_PHASE_CLOSE
    worker_assignee = _resolved_worker_assignee("", tm)
    if kind == "research" and worker_assignee == "research":
        if report_generated or append_scratch_count >= max(1, int(min_scratch_appends)):
            return KANBAN_PHASE_CLOSE
        if gather_limit_hit:
            return KANBAN_PHASE_CLOSE
        return KANBAN_PHASE_GATHER
    if kind == "mutate":
        if not wrote_text:
            return KANBAN_PHASE_MUTATE
        if (expects_lint or expects_render) and not render_tools_succeeded:
            return KANBAN_PHASE_VERIFY
        return KANBAN_PHASE_CLOSE
    if kind in ("verify", "deliver"):
        if render_tools_succeeded:
            return KANBAN_PHASE_CLOSE
        if kind == "deliver" and not wrote_text:
            return KANBAN_PHASE_MUTATE
        return KANBAN_PHASE_VERIFY
    if wrote_text and (expects_lint or expects_render) and not render_tools_succeeded:
        return KANBAN_PHASE_VERIFY
    if wrote_text:
        return KANBAN_PHASE_CLOSE
    return KANBAN_PHASE_GATHER


def observe_loop_salvage_next_action(
    *,
    task_metadata: dict[str, Any] | None,
    primary_path: str,
    verify_mode: bool = False,
) -> str:
    """``next_action`` for observe/verify loop circuit-break handoffs."""
    path = str(primary_path or "").strip() or "the primary artifact for this task"
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    worker_assignee = _resolved_worker_assignee("", tm)
    if kanban_task_is_web_act_deliverable(task_metadata):
        if verify_mode:
            return (
                f"On {path}, run run_web_script with fresh_run=True once; "
                "do not repeat the same script/read signature."
            )
        return (
            f"Stop re-reading {path}. Patch the script with write_web_act_file, "
            "then run_web_script once, then fresh_run=True."
        )
    if kanban_task_is_video_render_deliverable(task_metadata):
        if assignee_has_tool(worker_assignee, "hyperframes_render", metadata=tm):
            if verify_mode:
                return (
                    f"On {path}, run hyperframes_lint then hyperframes_render once; "
                    "do not repeat the same lint/read signature."
                )
            return (
                f"Stop re-reading {path}. Patch the HTML/GSAP composition with write_text_file, "
                "then run hyperframes_lint and hyperframes_render once."
            )
        if verify_mode:
            return (
                f"Inspect and verify {path} once instead of repeating the same tool signature."
            )
        return (
            f"Inspect and mutate {path}, then re-run verification once instead of "
            "repeating the same tool signature."
        )
    if verify_mode:
        return (
            f"Inspect and verify {path} once instead of repeating the same tool signature."
        )
    return (
        f"Inspect and mutate {path}, then re-run verification once instead of "
        "repeating the same tool signature."
    )


def _resolved_worker_assignee(
    assignee: str = "",
    task_metadata: dict[str, Any] | None = None,
) -> str:
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    raw = str(assignee or "").strip() or str(tm.get("_assignee") or "").strip()
    if not raw:
        return ""
    return _effective_profile_assignee(raw, tm)


def format_kanban_phase_nudge(
    phase: str,
    *,
    task_kind: str = "",
    task_metadata: dict[str, Any] | None = None,
    assignee: str = "",
    gather_evidence_excerpt: str = "",
) -> str:
    """One-line phase directive for the worker transcript."""
    phase_u = str(phase or "").strip().upper()
    kind = str(task_kind or "").strip().lower()
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    worker_assignee = _resolved_worker_assignee(assignee, tm)
    try:
        from utils.council_orchestrator import is_council_task_metadata

        if is_council_task_metadata(tm):
            if phase_u == KANBAN_PHASE_CLOSE:
                evidence = str(gather_evidence_excerpt or "").strip()
                evidence_block = f"{evidence}\n\n" if evidence else ""
                return (
                    "Kanban phase: CLOSE — STOP calling tools. Your very next assistant message "
                    "must end with a trailing ```kanban-result``` JSON fence (council_seat + this "
                    "seat's required metadata: claims, verified_claims, or dissent fields).\n"
                    f"{evidence_block}"
                    "Synthesize claims from gathered evidence above — do NOT claim no research was done."
                )
            try:
                from utils.council_orchestrator import council_researcher_has_preseed

                if (
                    str(tm.get("council_seat") or "").strip().lower() == "researcher"
                    and council_researcher_has_preseed(tm)
                ):
                    return (
                        "Kanban phase: GATHER — pre-seeded product excerpts are in your task. "
                        "Synthesize claims from them (cite utils/… paths in evidence[]); do NOT "
                        "read_file repo paths. Optional deep_web_search for generic context only. "
                        "Then CLOSE with ```kanban-result``` (council_seat + structured metadata)."
                    )
            except Exception:
                pass
            return (
                "Kanban phase: GATHER — use deep_web_search/read_file/read_webpage (2–4 gather calls "
                "total; at least two before CLOSE) to ground claims. No list_directory. "
                "No append_research_scratch. Then CLOSE with ```kanban-result``` "
                "(council_seat + structured metadata)."
            )
    except Exception:
        pass
    research_profile = worker_assignee == "research"
    if phase_u == KANBAN_PHASE_CLOSE:
        if kind == "research" and research_profile:
            return (
                "Kanban phase: CLOSE — use append_research_scratch / save or report tools, "
                "then end with a trailing kanban-result fenced JSON block."
            )
        return (
            "Kanban phase: CLOSE — complete verification if needed, "
            "then end with a trailing kanban-result fenced JSON block."
        )
    if phase_u == KANBAN_PHASE_VERIFY:
        if kanban_task_is_web_act_deliverable(tm):
            return (
                "Kanban phase: VERIFY — run run_web_script(script_path=…, fresh_run=True); "
                "then kanban-result with answer, web_act_run_dir, artifact_paths."
            )
        if kanban_task_is_video_render_deliverable(tm):
            if assignee_has_tool(worker_assignee, "hyperframes_render", metadata=tm):
                return (
                    "Kanban phase: VERIFY — run hyperframes_lint, then hyperframes_render; "
                    "then kanban-result fence."
                )
            return (
                "Kanban phase: VERIFY — run verification on the changed artifact, "
                "then kanban-result fence."
            )
        if kanban_task_expects_lint(tm):
            return (
                "Kanban phase: VERIFY — run the project lint tool on the changed artifact, "
                "then kanban-result fence."
            )
        return (
            "Kanban phase: VERIFY — run verification on the changed artifact, "
            "then kanban-result fence."
        )
    if phase_u == KANBAN_PHASE_MUTATE:
        if kanban_task_is_web_act_deliverable(tm):
            return (
                "Kanban phase: MUTATE — if scope is empty, write_web_act_file('explore.py', "
                "template). Then run_web_script. Do not use write_text_file for scripts."
            )
        if (
            kanban_task_is_video_render_deliverable(tm)
            and assignee_has_tool(worker_assignee, "hyperframes_render", metadata=tm)
        ):
            return (
                "Kanban phase: MUTATE — edit the HTML/GSAP composition (write_text_file); "
                "do not loop read_file. Then lint and render."
            )
        return "Kanban phase: MUTATE — write or patch the deliverable, then verify."
    if kind == "research" and research_profile:
        return (
            "Kanban phase: GATHER — use deep_web_search/read_and_analyze_page, then "
            "append_research_scratch with dense synthesis (required before more searches)."
        )
    return "Kanban phase: GATHER — gather evidence until scratch or artifact progress exists."


def _salvage_retry_hint(
    *,
    task_metadata: dict[str, Any] | None = None,
    summary: str = "",
    slug: str = "",
) -> str:
    """Tailored retry guidance for salvage ``next_action`` (never Hyperframes-specific by default)."""
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    task_kind = str(tm.get("task_kind") or "").strip().lower()
    deliverable_type = str(tm.get("deliverable_type") or "").strip().lower()
    hay = " ".join(
        str(tm.get(key) or "")
        for key in ("title", "body")
    ).lower()
    hay = f"{hay} {summary.lower()} {slug.lower()}".strip()

    worker_assignee = _resolved_worker_assignee("", tm)
    if deliverable_type == "video_render":
        if assignee_has_tool(worker_assignee, "hyperframes_render", metadata=tm):
            return "re-run lint and render verification once"
        return "re-run verification once"
    if deliverable_type == "web_act":
        return "re-run run_web_script with fresh_run=True once"
    if deliverable_type == "video_encode" or any(
        token in hay for token in ("ffmpeg", "encode", "transcode", "convert_video")
    ):
        return "re-run video encoding once"
    if task_kind == "verify" or " lint" in f" {hay}":
        return "re-run verification once"
    if task_kind == "deliver":
        return "retry deliverable production once"
    if task_kind == "research":
        return "retry research after resolving the blocker"
    if task_kind == "mutate":
        return "retry the mutation after applying the fix"
    return "retry once after applying the fix"


def _synthesize_salvage_next_action(
    *,
    blockers: list[str],
    path: str = "",
    slug: str = "",
    summary: str = "",
    task_metadata: dict[str, Any] | None = None,
    salvage_metadata: dict[str, Any] | None = None,
) -> str:
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    deliverable_type = str(tm.get("deliverable_type") or "").strip().lower()
    if deliverable_type == "web_act" and blockers:
        try:
            from utils.web_act_diagnostics import web_act_blocker_next_action_hint

            hint = web_act_blocker_next_action_hint(blockers, task_metadata=tm)
            sm = salvage_metadata if isinstance(salvage_metadata, dict) else {}
            tools_next = sm.get("tools_required_next")
            tool_names: list[str] = []
            if isinstance(tools_next, list):
                tool_names = [str(x).strip() for x in tools_next if str(x).strip()][:4]
            if tool_names:
                tools_phrase = ", ".join(f"`{n}`" for n in tool_names)
                return f"{hint} Then run {tools_phrase}."
            return hint
        except Exception:
            pass
    blocker_hint = blockers[0] if blockers else "reported blockers"
    sm = salvage_metadata if isinstance(salvage_metadata, dict) else {}
    tools_next = sm.get("tools_required_next")
    tool_names: list[str] = []
    if isinstance(tools_next, list):
        tool_names = [str(x).strip() for x in tools_next if str(x).strip()][:4]
    tools_phrase = ", ".join(f"`{n}`" for n in tool_names) if tool_names else ""

    retry_hint = _salvage_retry_hint(
        task_metadata=task_metadata,
        summary=summary,
        slug=slug,
    )
    if tools_phrase:
        action_tail = f"run {tools_phrase}"
        if path:
            return f"Patch {path} to resolve {blocker_hint}, then {action_tail}."
        if slug:
            return f"Fix {slug} to resolve {blocker_hint}, then {action_tail}."
        if summary:
            return _trunc(
                f"{summary} Apply the concrete fix, then {action_tail}.",
                240,
            )
        return f"Resolve {blocker_hint} on the primary artifact, then {action_tail}."

    if path:
        return f"Patch {path} to resolve {blocker_hint}, then {retry_hint}."
    if slug:
        tm = task_metadata if isinstance(task_metadata, dict) else {}
        deliverable_type = str(tm.get("deliverable_type") or "").strip().lower()
        if deliverable_type == "video_render":
            return (
                f"Fix the {slug} render project to resolve {blocker_hint}, then {retry_hint}."
            )
        if deliverable_type == "web_act":
            return (
                f"Fix the Playwright script ({slug}) to resolve {blocker_hint}, then {retry_hint}."
            )
        return f"Fix {slug} to resolve {blocker_hint}, then {retry_hint}."
    if summary:
        return _trunc(
            f"{summary} Apply the concrete fix, then {retry_hint}.",
            240,
        )
    return f"Resolve {blocker_hint} on the primary artifact, then {retry_hint}."


def normalize_salvage_handoff(
    parsed: dict[str, Any],
    *,
    worker_transcript: str = "",
    task_metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Fill salvage ``next_action`` when the model omitted it but blockers exist."""
    if not isinstance(parsed, dict):
        return parsed
    md = dict(parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {})
    status = str(md.get("status") or "").strip().lower()
    if status not in _TERMINAL_SALVAGE_META_STATUSES:
        return parsed
    if str(md.get("next_action") or "").strip():
        out = dict(parsed)
        out["metadata"] = md
        return out

    blockers = _metadata_blockers_list(md)
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    if kanban_task_is_web_act_deliverable(tm):
        try:
            from utils.web_act_diagnostics import merge_web_act_blockers, parse_blockers_from_tool_message

            transcript_blockers: list[str] = []
            for line in (worker_transcript or "").splitlines():
                if line.strip().lower().startswith("blockers:"):
                    transcript_blockers.extend(parse_blockers_from_tool_message(line))
            blockers = merge_web_act_blockers(transcript_blockers, blockers)
            if blockers:
                md["blockers"] = blockers
        except Exception:
            pass
    path = str(md.get("file_path") or md.get("artifact_paths") or "").strip()
    if isinstance(md.get("artifact_paths"), list) and md["artifact_paths"]:
        path = str(md["artifact_paths"][0] or "").strip() or path
    slug = str(md.get("project_slug") or md.get("project") or "").strip()
    summary = str(parsed.get("summary") or "").strip()

    md["next_action"] = _synthesize_salvage_next_action(
        blockers=blockers,
        path=path,
        slug=slug,
        summary=summary,
        task_metadata=task_metadata,
        salvage_metadata=md,
    )
    out = dict(parsed)
    out["metadata"] = md
    return out


def merge_kanban_handoff(
    parsed: dict[str, Any],
    *,
    inband_summary: str = "",
    inband_metadata: dict[str, Any] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Merge trailing fence with in-band ``kanban_post_summary`` writes.

    Salvage ``metadata.status`` (``partial``/``blocked``) from either source
    wins over a contradictory success-shaped fence so the verifier and
    dispatcher see the honest handoff.
    """
    inband_md = dict(inband_metadata or {})
    # Internal channel marker (Stage B) must never persist into the merged handoff metadata.
    inband_md.pop("_handoff_channel", None)
    fence_md = parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {}
    summary = str(parsed.get("summary") or inband_summary or "").strip()

    inband_salvage = salvage_handoff_action(inband_md)
    fence_salvage = salvage_handoff_action(fence_md)

    merged_md = dict(inband_md)
    if inband_salvage:
        merged_md.update({k: v for k, v in fence_md.items() if k not in ("status", "blockers")})
    elif fence_salvage:
        merged_md.update(fence_md)
    else:
        merged_md.update(fence_md)

    if inband_salvage and inband_md.get("blockers") is not None:
        merged_md["blockers"] = inband_md["blockers"]
    elif fence_salvage and fence_md.get("blockers") is not None and "blockers" not in merged_md:
        merged_md["blockers"] = fence_md["blockers"]

    if not summary and inband_salvage:
        summary = str(inband_md.get("salvage_summary") or inband_summary or "").strip()

    effective = {"summary": summary, "metadata": merged_md}
    return effective, merged_md


def salvage_handoff_action(metadata: dict[str, Any] | None) -> str | None:
    """Map salvage ``metadata.status`` to a dispatcher terminal action.

    Verifier success with ``partial``/``blocked`` means the worker reported
    honestly — not that the task succeeded. The dispatcher must not call
    :func:`kanban_storage.record_completion` for these handoffs.
    """
    status = str((metadata or {}).get("status") or "").strip().lower()
    if status == "blocked":
        return "block"
    if status == "partial":
        return "retry"
    return None
