"""Council mode orchestration: spawn, probe, validate, synthesis context."""

from __future__ import annotations

import json
import logging
import os
import re
import uuid
from datetime import datetime, timezone
from typing import Any

import kanban_storage as ks
from utils.kanban_context import FAILURE_VERIFIER_FAILED, verify_kanban_result
from utils.kanban_coordinator import TERMINAL_BOARD_STATUSES, probe_board_active
from utils.kanban_dispatcher import nudge_dispatcher
from utils.council_prompts import (
    COUNCIL_CONTRARIAN_HINT,
    COUNCIL_RESEARCHER_HINT,
    COUNCIL_RESEARCHER_RETRY_HINT,
    COUNCIL_VERIFIER_R2_HINT,
    council_verifier_hint_for_frame,
)

_log = logging.getLogger(__name__)

COUNCIL_ORIGIN = "council"
COUNCIL_BATCH_ID_META = "council_batch_id"
COUNCIL_SEAT_META = "council_seat"
COUNCIL_FRAME_META = "council_frame"
COUNCIL_ROUND_META = "council_round"
COUNCIL_PIPELINE_ATTEMPT_META = "council_pipeline_attempt"
COUNCIL_RESEARCHER_HAS_SEED_META = "council_researcher_has_seed"
COUNCIL_BATCH_ACTIVATED_MARKER = "Council batch:"
# Header for repo policy excerpts in synthesis context; delivery checks this to pick
# the code-grounded vs general Captain synthesis suffix.
COUNCIL_POLICY_HEADER = "## Implementation policy (from repo — authoritative for the verdict)"
_REPO_READ_PREFIXES = ("utils/", "docs/", "mesh/", "tools/", "tests/")
TERMINAL_FAILURE_STATUSES = frozenset({"blocked", "cancelled", "gave_up"})
# Council "open batch" probes must treat gave_up as terminal (unlike coordinator board).
COUNCIL_TERMINAL_STATUSES = frozenset(set(TERMINAL_BOARD_STATUSES) | {"gave_up"})

_SEAT_LABELS = {
    "researcher": "Researcher (Athena)",
    "verifier": "Verifier (Themis)",
    "contrarian": "Contrarian (Prometheus)",
}


def council_dev_repo_root() -> str:
    """Absolute path to the Mugi source tree when available (not ~/.mugi/workspace)."""
    env = (os.environ.get("MUGI_REPO_ROOT") or "").strip()
    if env:
        root = os.path.abspath(os.path.expanduser(env))
        if os.path.isdir(root):
            return root
    try:
        import mugi_backend

        root = os.path.dirname(os.path.abspath(mugi_backend.__file__))
        if os.path.isfile(os.path.join(root, "mugi_backend.py")):
            return root
    except Exception:
        pass
    return ""


_COUNCIL_RESEARCHER_SEED_RULES: list[tuple[tuple[str, ...], tuple[str, ...]]] = (
    (
        ("kanban", "dispatcher", "worker", "deadline", "cooperative", "completion", "slot", "backpressure"),
        (
            "utils/kanban_dispatcher.py",
            "docs/worker_completion_reliability_audit.md",
            "utils/kanban_context.py",
        ),
    ),
    (
        ("council", "synthesis", "seat", "deliberat"),
        (
            "utils/council_orchestrator.py",
            "utils/council_delivery.py",
        ),
    ),
    (
        ("idle_tick", "idle tick", "immediate", "watching", "delivery", "timing", "captain"),
        (
            "utils/council_delivery.py",
            "mugi_backend.py",
        ),
    ),
    (
        ("mesh", "peer", "federat"),
        (
            "mesh/__init__.py",
        ),
    ),
)


def _council_frame_matches_keywords(frame: str, keywords: tuple[str, ...]) -> bool:
    lower = (frame or "").lower()
    return any(kw in lower for kw in keywords)


_COUNCIL_POLICY_FRAME_KEYWORDS = (
    "council",
    "synthesis",
    "delivery",
    "idle_tick",
    "idle tick",
    "mugi",
    "implement",
    "orchestrat",
    "kanban",
    "captain",
    "immediate",
    "thread",
    "watching",
)

_COUNCIL_POLICY_SOURCES: list[tuple[str, tuple[str, ...]]] = (
    (
        "utils/council_delivery.py",
        (
            "def _deliver_council_synthesis",
            "def on_contrarian_done",
            "def on_round2_verifier_done",
            "if synthesis_delivery_pending(origin)",
            "def synthesis_delivery_pending",
            "def council_synthesis_already_answered",
            "def wake_council_thread",
            "def deliver_council_continuation",
        ),
    ),
    (
        "mugi_backend.py",
        (
            "council_synthesis_already_answered",
            "council_awaiting_synthesis",
            "synthesis_delivery_pending",
            'display_text="[Council synthesis]"',
        ),
    ),
)


def _read_dev_repo_anchor_excerpt(
    root: str,
    rel_path: str,
    anchors: tuple[str, ...],
    *,
    max_chars: int = 2800,
    context_before: int = 2,
    context_after: int = 22,
) -> str:
    """Pull fixed symbol/route anchors from a source file (policy excerpts)."""
    full = os.path.join(root, rel_path)
    if not os.path.isfile(full) or not anchors:
        return ""
    try:
        with open(full, encoding="utf-8", errors="replace") as fh:
            lines = fh.read().splitlines()
    except OSError:
        return ""
    hit_ranges: list[tuple[int, int]] = []
    lower_lines = [ln.lower() for ln in lines]
    for anchor in anchors:
        needle = (anchor or "").strip().lower()
        if not needle:
            continue
        for idx, lower in enumerate(lower_lines):
            if needle in lower:
                start = max(0, idx - context_before)
                end = min(len(lines), idx + context_after)
                hit_ranges.append((start, end))
                break
    if not hit_ranges:
        return ""
    hit_ranges.sort()
    merged: list[tuple[int, int]] = []
    for start, end in hit_ranges:
        if not merged or start > merged[-1][1]:
            merged.append((start, end))
        else:
            merged[-1] = (merged[-1][0], max(merged[-1][1], end))
    chunks: list[str] = []
    for start, end in merged:
        chunks.extend(lines[start:end])
        chunks.append("---")
    text = "\n".join(chunks).strip()
    if len(text) > max_chars:
        return text[:max_chars] + "\n… [truncated]"
    return text


def _read_dev_repo_excerpt(
    root: str,
    rel_path: str,
    *,
    max_chars: int = 2400,
    needles: list[str] | None = None,
) -> str:
    full = os.path.join(root, rel_path)
    if not os.path.isfile(full):
        return ""
    try:
        with open(full, encoding="utf-8", errors="replace") as fh:
            text = fh.read()
    except OSError:
        return ""
    if needles:
        lines = text.splitlines()
        hits: list[str] = []
        for idx, line in enumerate(lines):
            if any(n.lower() in line.lower() for n in needles):
                start = max(0, idx - 3)
                end = min(len(lines), idx + 14)
                hits.extend(lines[start:end])
                hits.append("---")
        text = "\n".join(hits).strip() if hits else text
    text = text.strip()
    if len(text) > max_chars:
        return text[:max_chars] + "\n… [truncated]"
    return text


def council_researcher_seed_excerpts(frame: str, *, max_total_chars: int = 6500) -> str:
    """Server-side product source injected into Researcher brief (read_file cannot reach dev checkout)."""
    root = council_dev_repo_root()
    frame_s = (frame or "").strip()
    if not root or not frame_s:
        return ""
    rel_paths: list[str] = []
    for keywords, paths in _COUNCIL_RESEARCHER_SEED_RULES:
        if _council_frame_matches_keywords(frame_s, keywords):
            for rel in paths:
                if rel not in rel_paths:
                    rel_paths.append(rel)
    needles = [w for w in re.split(r"[^\w]+", frame_s.lower()) if len(w) >= 5][:10]
    parts: list[str] = []
    used = 0
    policy = council_implementation_policy_excerpt(
        frame_s,
        max_total_chars=min(3200, max_total_chars),
    )
    if policy:
        parts.append(policy)
        used += len(policy)
    if not rel_paths and not parts:
        return ""
    for rel in rel_paths[:4]:
        if used >= max_total_chars:
            break
        budget = min(2400, max_total_chars - used)
        excerpt = _read_dev_repo_excerpt(root, rel, max_chars=budget, needles=needles or None)
        if excerpt:
            parts.append(f"### {rel}\n{excerpt}")
            used += len(excerpt)
    if not parts:
        return ""
    return (
        "## Pre-seeded product source (authoritative for Mugi implementation)\n"
        "STOP — do **not** call read_file or read_file_chunk on utils/, docs/, or mesh/ paths. "
        "The dev checkout is not reachable via read_file; the excerpts below are already loaded "
        "from the live repo. Synthesize claims from this section, cite file paths in evidence[], "
        "then emit ```kanban-result```. Use deep_web_search only for generic industry background "
        "the frame still needs.\n\n"
        + "\n\n".join(parts)
    )


def _frame_wants_implementation_policy(frame: str) -> bool:
    lower = (frame or "").strip().lower()
    if not lower:
        return False
    return any(kw in lower for kw in _COUNCIL_POLICY_FRAME_KEYWORDS)


def council_implementation_policy_excerpt(
    frame: str,
    *,
    max_total_chars: int = 6200,
) -> str:
    """Repo excerpts for Captain synthesis — authoritative over seat philosophy."""
    root = council_dev_repo_root()
    if not root or not _frame_wants_implementation_policy(frame):
        return ""
    parts: list[str] = []
    used = 0
    for rel, anchors in _COUNCIL_POLICY_SOURCES:
        if used >= max_total_chars:
            break
        budget = min(3600, max_total_chars - used)
        excerpt = _read_dev_repo_anchor_excerpt(root, rel, anchors, max_chars=budget)
        if excerpt:
            parts.append(f"### {rel}\n{excerpt}")
            used += len(excerpt)
    if not parts:
        return ""
    return (
        f"{COUNCIL_POLICY_HEADER}\n"
        "Answer the Captain frame from this section first. Seat summaries and Contrarian "
        "dissent add color but **cannot** override what these paths show. If the repo is "
        "silent on a branch of the question, say so — do not invent behavior or prefer "
        "philosophy over code.\n\n"
        + "\n\n".join(parts)
    )


def council_researcher_has_preseed(task_metadata: dict[str, Any] | None) -> bool:
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    return bool(tm.get(COUNCIL_RESEARCHER_HAS_SEED_META))


def is_council_repo_relative_read_path(path: str) -> bool:
    """True when a read path targets the dev repo, not workspace/Documents/Mugi."""
    p = str(path or "").strip().replace("\\", "/")
    if not p:
        return False
    lower = p.lower()
    if lower.startswith("~/documents/mugi/"):
        rest = lower[len("~/documents/mugi/") :]
        return rest.startswith(_REPO_READ_PREFIXES)
    if "documents/code/mugi" in lower and any(seg in lower for seg in _REPO_READ_PREFIXES):
        return True
    norm = p.lstrip("./")
    return norm.startswith(_REPO_READ_PREFIXES)


def council_researcher_blocked_repo_read_message(path: str) -> str:
    return (
        "[Council guidance] read_file cannot open the dev checkout at "
        f"{path!r}. The **Pre-seeded product source** section in your task already "
        "contains the excerpts — synthesize claims from there, cite paths in evidence[], "
        "and finish with ```kanban-result```. Do not retry read_file on utils/, docs/, "
        "or mesh/ paths."
    )


def council_researcher_source_paths_block() -> str:
    """Steer Researcher toward seeded product source + workspace docs; web is secondary."""
    lines = [
        "Source guidance:",
        "- **Mugi product implementation:** when pre-seeded excerpts appear below, they are "
        "authoritative — synthesize claims and cite paths like utils/council_delivery.py in "
        "evidence[]. **Do not** call read_file/read_file_chunk on utils/, docs/, or mesh/ paths "
        "(dev checkout; not in ~/.mugi/workspace/).",
        "- **read_file/read_file_chunk** only reach ~/.mugi/workspace/ and ~/Documents/Mugi/ "
        "(operator output — OCR, downloads — **not** the Python source tree).",
        "- **Workspace docs:** read_file_chunk once per path (chunk_index 0, 1, …) when needed.",
        "- Do NOT loop read_file on ~/.mugi/workspace/AGENTS.md — copies may be stale.",
        "- **Generic industry context only:** deep_web_search / read_webpage (max 2 when seed excerpts exist).",
        "- **Ingested corpora:** when an 'Ingested Knowledge' block appears below, a studied "
        "corpus is relevant — use query_knowledge / ask_studied_corpus first; otherwise do not "
        "query corpora.",
        "- Never re-search the same query; no list_directory.",
    ]
    root = council_dev_repo_root()
    if root:
        lines.append(f"- Dev checkout (pre-seeded server-side only): {root}")
    lines.append(
        "- ~/Documents/Mugi/ is operator artifacts, not the git repo — repo-relative paths "
        "do not exist there."
    )
    return "\n".join(lines)


def council_settings() -> dict[str, Any]:
    from core.config import get_council_config

    return get_council_config()


def council_enabled() -> bool:
    cfg = council_settings()
    mode = str(cfg.get("mode") or "off").strip().lower()
    if mode == "off":
        return False
    return bool(cfg.get("enabled", True))


def new_council_batch_id() -> str:
    return f"kcn_{uuid.uuid4().hex[:16]}"


def _batch_id_from_task(task: dict[str, Any]) -> str:
    meta = task.get("metadata")
    if isinstance(meta, dict):
        bid = meta.get(COUNCIL_BATCH_ID_META)
        if isinstance(bid, str):
            return bid.strip()
    return ""


def _seat_from_task(task: dict[str, Any]) -> str:
    meta = task.get("metadata")
    if isinstance(meta, dict):
        return str(meta.get(COUNCIL_SEAT_META) or "").strip().lower()
    return ""


def _round_from_task(task: dict[str, Any]) -> int:
    meta = task.get("metadata")
    if isinstance(meta, dict):
        try:
            return max(1, int(meta.get(COUNCIL_ROUND_META) or 1))
        except (TypeError, ValueError):
            return 1
    return 1


def council_max_rounds_config() -> int:
    try:
        raw = council_settings().get("max_rounds", 1)
        return max(1, min(2, int(raw)))
    except (TypeError, ValueError):
        return 1


def council_max_researcher_attempts_config() -> int:
    try:
        raw = council_settings().get("max_researcher_attempts", 2)
        return max(1, min(4, int(raw)))
    except (TypeError, ValueError):
        return 2


def _pipeline_attempt_from_task(task: dict[str, Any]) -> int:
    meta = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
    try:
        return max(1, int(meta.get(COUNCIL_PIPELINE_ATTEMPT_META) or 1))
    except (TypeError, ValueError):
        return 1


def _tasks_for_seat(
    thread_id: str,
    batch_id: str,
    seat: str,
    *,
    include_cancelled: bool = False,
) -> list[dict[str, Any]]:
    seat_key = (seat or "").strip().lower()
    out: list[dict[str, Any]] = []
    for row in tasks_for_council_batch(thread_id, batch_id):
        if _seat_from_task(row) != seat_key:
            continue
        if not include_cancelled and (row.get("status") or "").strip() == "cancelled":
            continue
        out.append(row)
    return out


def _latest_task_for_seat(
    thread_id: str,
    batch_id: str,
    seat: str,
    *,
    include_cancelled: bool = False,
) -> dict[str, Any] | None:
    rows = _tasks_for_seat(
        thread_id,
        batch_id,
        seat,
        include_cancelled=include_cancelled,
    )
    if not rows:
        return None
    return max(
        rows,
        key=lambda t: (
            _pipeline_attempt_from_task(t),
            str(t.get("updated_at") or t.get("created_at") or ""),
        ),
    )


def _max_pipeline_attempt_for_batch(thread_id: str, batch_id: str) -> int:
    attempts = [
        _pipeline_attempt_from_task(t)
        for t in tasks_for_council_batch(thread_id, batch_id)
    ]
    return max(attempts) if attempts else 1


def should_retry_researcher_after_verifier(
    verifier_meta: dict[str, Any],
    *,
    researcher_meta: dict[str, Any] | None = None,
) -> tuple[bool, str]:
    """True when the Verifier found zero supported claims — Researcher should gather again."""
    md = verifier_meta if isinstance(verifier_meta, dict) else {}
    verified = md.get("verified_claims") if isinstance(md.get("verified_claims"), list) else []
    rejections = md.get("rejections") if isinstance(md.get("rejections"), list) else []
    supported = [
        row
        for row in verified
        if isinstance(row, dict)
        and str(row.get("status") or "").strip().lower() == "supported"
    ]
    if supported:
        return False, ""
    if rejections and council_max_rounds_config() >= 2:
        # Disagreement gate schedules round-2 Verifier after Contrarian finishes.
        return False, "disagreement_gate_owns_follow_up"
    # Retry when every claim was rejected (rejections-only is valid Lucas output) or
    # verified_claims lists claims but none are supported.
    if rejections or verified:
        return True, "verifier found no supported claims"
    return False, ""


def batch_has_pending_researcher_retry(thread_id: str, batch_id: str) -> bool:
    for row in tasks_for_council_batch(thread_id, batch_id):
        if _seat_from_task(row) != "researcher":
            continue
        if _pipeline_attempt_from_task(row) <= 1:
            continue
        if (row.get("status") or "").strip() not in COUNCIL_TERMINAL_STATUSES:
            return True
    return False


def should_schedule_round2_verifier(
    *,
    verifier_meta: dict[str, Any],
    contrarian_meta: dict[str, Any],
    max_rounds: int | None = None,
) -> bool:
    """Disagreement gate: Verifier rejections or heavy Contrarian dissent."""
    cap = council_max_rounds_config() if max_rounds is None else max(1, min(2, int(max_rounds)))
    if cap < 2:
        return False
    rejections = (
        verifier_meta.get("rejections") if isinstance(verifier_meta.get("rejections"), list) else []
    )
    if len(rejections) >= 1:
        return True
    dissent = (
        contrarian_meta.get("dissent_items")
        if isinstance(contrarian_meta.get("dissent_items"), list)
        else []
    )
    return len(dissent) >= 3


def _researcher_task_for_batch(thread_id: str, batch_id: str) -> dict[str, Any] | None:
    return _latest_task_for_seat(thread_id, batch_id, "researcher")


def batch_has_round2_verifier(thread_id: str, batch_id: str) -> bool:
    for row in tasks_for_council_batch(thread_id, batch_id):
        if _seat_from_task(row) == "verifier" and _round_from_task(row) >= 2:
            return True
    return False


def _verifier_r1_task_for_batch(thread_id: str, batch_id: str) -> dict[str, Any] | None:
    rows = [
        row
        for row in tasks_for_council_batch(thread_id, batch_id)
        if _seat_from_task(row) == "verifier" and _round_from_task(row) <= 1
    ]
    if not rows:
        return None
    return max(
        rows,
        key=lambda t: (
            _pipeline_attempt_from_task(t),
            str(t.get("updated_at") or t.get("created_at") or ""),
        ),
    )


def _contrarian_task_for_batch(thread_id: str, batch_id: str) -> dict[str, Any] | None:
    return _latest_task_for_seat(thread_id, batch_id, "contrarian")


def _find_open_council_batch_id(thread_id: str) -> str:
    tid = (thread_id or "").strip()
    if not tid:
        return ""
    try:
        rows = ks.list_tasks(
            origin_thread_id=tid,
            limit=100,
            max_limit=100,
            order="desc",
        )
    except Exception:
        _log.debug("council open batch scan failed", exc_info=True)
        return ""
    batch_ids: list[str] = []
    for row in rows:
        meta = row.get("metadata") if isinstance(row.get("metadata"), dict) else {}
        if meta.get("origin") != COUNCIL_ORIGIN:
            continue
        bid = str(meta.get(COUNCIL_BATCH_ID_META) or "").strip()
        if not bid or bid in batch_ids:
            continue
        batch_ids.append(bid)
    for bid in batch_ids:
        open_for_batch = False
        for row in rows:
            meta = row.get("metadata") if isinstance(row.get("metadata"), dict) else {}
            if str(meta.get(COUNCIL_BATCH_ID_META) or "").strip() != bid:
                continue
            if (row.get("status") or "") not in COUNCIL_TERMINAL_STATUSES:
                open_for_batch = True
                break
        if open_for_batch:
            return bid
    return ""


def get_active_council_batch_id(thread_id: str) -> str:
    if not thread_id:
        return ""
    from chat_storage import get_thread_council_batch

    return get_thread_council_batch(thread_id) or ""


def set_active_council_batch(thread_id: str, batch_id: str) -> None:
    from chat_storage import set_thread_council_batch

    set_thread_council_batch(thread_id, batch_id)


def clear_active_council_batch(thread_id: str) -> None:
    from chat_storage import clear_thread_council_batch

    clear_thread_council_batch(thread_id)


def probe_council_batch_id(thread_id: str) -> str:
    """Read-only batch id (stored pointer or open-batch scan; never writes)."""
    bid = get_active_council_batch_id(thread_id)
    if bid:
        return bid
    return _find_open_council_batch_id(thread_id)


def resolve_council_batch_id(thread_id: str) -> str:
    """Batch id for delivery paths; recovers stored pointer when an open batch exists."""
    bid = probe_council_batch_id(thread_id)
    if bid and not get_active_council_batch_id(thread_id):
        try:
            set_active_council_batch(thread_id, bid)
        except Exception:
            _log.warning("council pointer recovery failed for %s", thread_id, exc_info=True)
    return bid


def tasks_for_council_batch(thread_id: str, batch_id: str) -> list[dict[str, Any]]:
    tid = (thread_id or "").strip()
    bid = (batch_id or "").strip()
    if not tid or not bid:
        return []
    try:
        rows = ks.list_tasks(
            origin_thread_id=tid,
            limit=100,
            max_limit=100,
            order="asc",
        )
    except Exception:
        return []
    out: list[dict[str, Any]] = []
    for row in rows:
        meta = row.get("metadata") if isinstance(row.get("metadata"), dict) else {}
        if str(meta.get(COUNCIL_BATCH_ID_META) or "").strip() == bid:
            out.append(row)
    return out


def probe_council_open_batch_id(thread_id: str) -> str:
    """Batch id only when at least one seat task is still non-terminal."""
    return _find_open_council_batch_id(thread_id)


def council_seat_in_progress_for_batch(thread_id: str, batch_id: str) -> str:
    """Seat key for the current non-terminal task (latest pipeline attempt per role)."""
    tid = (thread_id or "").strip()
    bid = (batch_id or "").strip()
    if not tid or not bid:
        return ""
    tasks = tasks_for_council_batch(tid, bid)
    if not tasks:
        return ""

    def _latest_for(
        seat: str,
        *,
        verifier_round: int | None = None,
    ) -> dict[str, Any] | None:
        rows: list[dict[str, Any]] = []
        for row in tasks:
            if _seat_from_task(row) != seat:
                continue
            if verifier_round is not None:
                rnd = _round_from_task(row)
                if verifier_round >= 2 and rnd < 2:
                    continue
                if verifier_round <= 1 and rnd >= 2:
                    continue
            if (row.get("status") or "").strip() == "cancelled":
                continue
            rows.append(row)
        if not rows:
            return None
        return max(
            rows,
            key=lambda t: (
                _pipeline_attempt_from_task(t),
                str(t.get("updated_at") or t.get("created_at") or ""),
            ),
        )

    for seat, verifier_round in (
        ("researcher", None),
        ("verifier", 1),
        ("contrarian", None),
        ("verifier", 2),
    ):
        latest = _latest_for(seat, verifier_round=verifier_round)
        if not latest:
            continue
        if (latest.get("status") or "").strip() not in COUNCIL_TERMINAL_STATUSES:
            return seat
    return ""


def probe_council_active(thread_id: str) -> bool:
    """Read-only probe — never clears the council pointer."""
    batch_id = probe_council_open_batch_id(thread_id)
    if not batch_id:
        return False
    tasks = tasks_for_council_batch(thread_id, batch_id)
    if not tasks:
        return False
    return any((t.get("status") or "") not in COUNCIL_TERMINAL_STATUSES for t in tasks)


def council_synthesis_ready(thread_id: str, batch_id: str) -> bool:
    """True when every seat in the batch is terminal and no retry chain is in flight."""
    tid = (thread_id or "").strip()
    bid = (batch_id or "").strip()
    if not tid or not bid:
        return False
    if batch_has_pending_researcher_retry(tid, bid):
        return False
    tasks = tasks_for_council_batch(tid, bid)
    if not tasks:
        return False
    return all((t.get("status") or "").strip() in COUNCIL_TERMINAL_STATUSES for t in tasks)


def council_batch_overlap_error(*, origin_thread_id: str) -> str:
    if not (origin_thread_id or "").strip():
        return ""
    if not council_enabled():
        return ""
    try:
        if probe_council_active(origin_thread_id):
            return (
                "Council mode is already running for this conversation. "
                "Wait for seats to finish or stop the current council before starting another."
            )
    except Exception:
        _log.debug("council overlap probe failed", exc_info=True)
        return "Could not verify council state; try again."
    return ""


def spawn_preflight_error(
    *,
    origin_thread_id: str,
    allow_in_flight_turn: bool = False,
    slash_override: bool = False,
) -> str:
    """Return user-visible error when council cannot start, else empty string."""
    if not slash_override and not council_enabled():
        return "Council mode is disabled. Enable it in Settings → General → Council mode."
    tid = (origin_thread_id or "").strip()
    if not allow_in_flight_turn:
        try:
            from utils.council_triggers import thread_agent_running

            if tid and thread_agent_running(tid):
                return (
                    "This conversation is still processing. "
                    "Wait for the current turn to finish, then approve council or try again."
                )
        except Exception:
            _log.debug("council thread-running check failed", exc_info=True)
    try:
        from core.config import get_kanban_config

        if not bool(get_kanban_config().get("enabled", True)):
            return "Kanban is disabled — council mode requires the task board."
    except Exception:
        _log.debug("kanban enabled check failed", exc_info=True)
        return "Could not verify Kanban status; try again."
    overlap = council_batch_overlap_error(origin_thread_id=origin_thread_id)
    if overlap:
        return overlap
    try:
        from utils.council_budgets import council_cap_refusal_message

        cap_msg = council_cap_refusal_message()
        if cap_msg:
            return cap_msg
    except Exception:
        _log.warning("council cap check failed", exc_info=True)
        return "Could not verify council limits; try again."
    if probe_board_active(origin_thread_id):
        return (
            "Kanban coordinator board is active for this conversation. "
            "Wait for it to finish or clear it before convening council."
        )
    return ""


def _acceptance_for_seat(seat: str, *, council_round: int = 1) -> dict[str, Any]:
    base = {"require_metadata_fields": ["council_seat"]}
    if seat == "researcher":
        base["require_metadata_fields"] = ["council_seat", "claims"]
        base["require_metadata_nonempty"] = ["claims"]
    elif seat == "verifier":
        base["require_metadata_fields"] = ["council_seat"]
        base["require_metadata_nonempty_any"] = ["verified_claims", "rejections"]
        if council_round >= 2:
            base["require_metadata_fields"].append("follow_up_notes")
            base["require_metadata_nonempty"] = ["follow_up_notes"]
    elif seat == "contrarian":
        base["require_metadata_fields"] = [
            "council_seat",
            "dissent_items",
            "assumptions_challenged",
            "strongest_counter",
            "failure_modes",
        ]
        base["require_metadata_nonempty"] = [
            "dissent_items",
            "assumptions_challenged",
            "strongest_counter",
            "failure_modes",
        ]
    return base


def build_council_batch_items(
    *,
    council_frame: str,
    batch_id: str,
    origin_thread_id: str,
) -> list[dict[str, Any]]:
    frame = (council_frame or "").strip()
    if not frame:
        raise ValueError("council_frame is required")
    shared = {
        "origin": COUNCIL_ORIGIN,
        COUNCIL_BATCH_ID_META: batch_id,
        COUNCIL_FRAME_META: frame,
        COUNCIL_ROUND_META: 1,
        COUNCIL_PIPELINE_ATTEMPT_META: 1,
        "origin_thread_id": origin_thread_id,
    }
    _seed = council_researcher_seed_excerpts(frame)
    researcher_body = (
        f"Captain frame (answer this, not raw chat drift):\n{frame}\n\n"
        f"{council_researcher_source_paths_block()}\n"
    )
    if _seed:
        researcher_body += f"\n{_seed}\n"
    researcher_body += f"\n{COUNCIL_RESEARCHER_HINT}"
    verifier_body = build_council_verifier_task_body(frame)
    contrarian_body = (
        f"Captain frame:\n{frame}\n\n"
        f"{COUNCIL_CONTRARIAN_HINT}"
    )
    return [
        {
            "title": "[Council] Researcher",
            "body": researcher_body,
            "assignee": "council_researcher",
            "priority": 1,
            "metadata": {
                **shared,
                COUNCIL_SEAT_META: "researcher",
                COUNCIL_RESEARCHER_HAS_SEED_META: bool(_seed),
                "acceptance": _acceptance_for_seat("researcher"),
            },
        },
        {
            "title": "[Council] Verifier",
            "body": verifier_body,
            "assignee": "council_verifier",
            "priority": 2,
            "parent_index": 0,
            "metadata": {
                **shared,
                COUNCIL_SEAT_META: "verifier",
                "acceptance": _acceptance_for_seat("verifier"),
            },
        },
        {
            "title": "[Council] Contrarian",
            "body": contrarian_body,
            "assignee": "council_contrarian",
            "priority": 3,
            # Parent is Verifier only (index 1): Verifier already depends on Researcher,
            # so AND-promotion is preserved without a diamond false-positive in cycle check.
            "parent_index": 1,
            "metadata": {
                **shared,
                COUNCIL_SEAT_META: "contrarian",
                "acceptance": _acceptance_for_seat("contrarian"),
            },
        },
    ]


def council_batch_created_user_message(batch_id: str) -> str:
    bid = (batch_id or "").strip()
    return (
        f"{COUNCIL_BATCH_ACTIVATED_MARKER} {bid}\n"
        f"Council convened: {council_seat_display_label('researcher')} → "
        f"{council_seat_display_label('verifier')} → {council_seat_display_label('contrarian')}. "
        "I'm pausing here until seats finish — I'll synthesize when the contrarian review completes."
    )


def spawn_council_batch(
    *,
    origin_thread_id: str,
    council_frame: str,
    allow_in_flight_turn: bool = False,
    slash_override: bool = False,
) -> dict[str, Any]:
    """Create the sequential council DAG and set the thread pointer."""
    tid = (origin_thread_id or "").strip()
    preflight = spawn_preflight_error(
        origin_thread_id=tid,
        allow_in_flight_turn=allow_in_flight_turn,
        slash_override=slash_override,
    )
    if preflight:
        raise ValueError(preflight)
    batch_id = new_council_batch_id()
    if not batch_id:
        raise ValueError("council batch id allocation failed")
    items = build_council_batch_items(
        council_frame=council_frame,
        batch_id=batch_id,
        origin_thread_id=tid,
    )
    rows = ks.create_batch(items, origin_thread_id=tid)
    if len(rows) != 3:
        raise RuntimeError(f"council batch expected 3 tasks, got {len(rows)}")
    set_active_council_batch(tid, batch_id)
    try:
        from utils.council_budgets import record_council_spawn

        record_council_spawn()
    except Exception:
        _log.warning("council cap record failed (spawn succeeded; caps may be understated)", exc_info=True)
    try:
        from chat_storage import clear_thread_council_pending_offer

        clear_thread_council_pending_offer(tid)
    except Exception:
        _log.debug("clear council pending offer on spawn failed", exc_info=True)
    try:
        nudge_dispatcher()
    except Exception:
        _log.debug("council dispatcher nudge failed", exc_info=True)
    return {
        "ok": True,
        "council_batch_id": batch_id,
        "researcher_task_id": rows[0]["id"],
        "verifier_task_id": rows[1]["id"],
        "contrarian_task_id": rows[2]["id"],
        "message": council_batch_created_user_message(batch_id),
    }


def build_researcher_retry_body(
    *,
    council_frame: str,
    verifier_meta: dict[str, Any],
    researcher_meta: dict[str, Any],
    pipeline_attempt: int,
) -> str:
    frame = (council_frame or "").strip()
    blocks: list[str] = []
    if frame:
        blocks.append(f"Captain frame (answer this, not raw chat drift):\n{frame}")
    blocks.append(COUNCIL_RESEARCHER_RETRY_HINT)
    blocks.append(f"Pipeline attempt: {pipeline_attempt}")
    prior_claims = researcher_meta.get("claims") if isinstance(researcher_meta.get("claims"), list) else []
    if prior_claims:
        blocks.append(
            "─── Prior Researcher claims (RECORDED — do not repeat) ───\n"
            + json.dumps(prior_claims, ensure_ascii=False, indent=2)[:4000]
        )
    rejections = verifier_meta.get("rejections") if isinstance(verifier_meta.get("rejections"), list) else []
    verified = (
        verifier_meta.get("verified_claims")
        if isinstance(verifier_meta.get("verified_claims"), list)
        else []
    )
    if rejections or verified:
        blocks.append(
            "─── Verifier rejection notes (RECORDED) ───\n"
            + json.dumps(
                {"rejections": rejections, "verified_claims": verified},
                ensure_ascii=False,
                indent=2,
            )[:4000]
        )
    blocks.append(council_researcher_source_paths_block())
    _seed = council_researcher_seed_excerpts(frame)
    if _seed:
        blocks.append(_seed)
    blocks.append(COUNCIL_RESEARCHER_HINT)
    return "\n\n".join(blocks)


def spawn_researcher_retry_pipeline(
    *,
    origin_thread_id: str,
    batch_id: str,
    verifier_task_id: str,
) -> dict[str, Any]:
    """After Verifier R1 rejects all claims, re-run Researcher → Verifier → Contrarian."""
    tid = (origin_thread_id or "").strip()
    bid = (batch_id or "").strip()
    vid = (verifier_task_id or "").strip()
    if not tid or not bid or not vid:
        raise ValueError("origin_thread_id, batch_id, and verifier_task_id are required")
    if batch_has_pending_researcher_retry(tid, bid):
        pending = _latest_task_for_seat(tid, bid, "researcher")
        return {
            "ok": True,
            "already_spawned": True,
            "researcher_task_id": pending["id"] if pending else "",
        }

    next_attempt = _max_pipeline_attempt_for_batch(tid, bid) + 1
    if next_attempt > council_max_researcher_attempts_config():
        return {"ok": False, "reason": "max_researcher_attempts", "attempt": next_attempt}

    verifier_task = ks.get_task(vid)
    if not verifier_task:
        raise ValueError("verifier task not found")
    vmeta = load_seat_result_metadata(vid)
    researcher = _researcher_task_for_batch(tid, bid)
    if not researcher:
        raise ValueError("researcher task not found for batch")
    rmeta = load_seat_result_metadata(researcher["id"])
    should, why = should_retry_researcher_after_verifier(vmeta, researcher_meta=rmeta)
    if not should:
        return {"ok": False, "reason": why or "verifier_did_not_request_retry"}

    contrarian = _contrarian_task_for_batch(tid, bid)
    if contrarian:
        cid = str(contrarian.get("id") or "").strip()
        cst = (contrarian.get("status") or "").strip()
        if cst == "done":
            return {"ok": False, "reason": "contrarian_already_done", "status": cst}
        if cst not in COUNCIL_TERMINAL_STATUSES:
            cancelled: list[str] = []
            try:
                from utils.kanban_dispatcher import request_worker_stop

                request_worker_stop(cid, join_seconds=3.0)
            except Exception:
                _log.debug("request_worker_stop before contrarian cancel failed", exc_info=True)
            try:
                cancelled = ks.cancel_task(cid, cascade=False)
                if cancelled:
                    _log.info(
                        "cancelled contrarian %s (%s) for researcher retry on batch %s",
                        cid,
                        cst,
                        bid,
                    )
            except Exception:
                _log.warning("cancel contrarian for researcher retry failed", exc_info=True)
            fresh = ks.get_task(cid) if cid else None
            fst = (fresh.get("status") or "").strip() if fresh else cst
            if fst == "done":
                return {"ok": False, "reason": "contrarian_already_done", "status": fst}
            if fst not in COUNCIL_TERMINAL_STATUSES and cid not in cancelled:
                return {
                    "ok": False,
                    "reason": "contrarian_cancel_failed",
                    "status": fst,
                    "task_id": cid,
                }

    frame = ""
    tmeta = verifier_task.get("metadata") if isinstance(verifier_task.get("metadata"), dict) else {}
    frame = str(tmeta.get(COUNCIL_FRAME_META) or "").strip()
    shared = {
        "origin": COUNCIL_ORIGIN,
        COUNCIL_BATCH_ID_META: bid,
        COUNCIL_FRAME_META: frame,
        COUNCIL_ROUND_META: 1,
        COUNCIL_PIPELINE_ATTEMPT_META: next_attempt,
        "origin_thread_id": tid,
        "council_researcher_retry_reason": why[:500],
    }
    researcher_body = build_researcher_retry_body(
        council_frame=frame,
        verifier_meta=vmeta,
        researcher_meta=rmeta,
        pipeline_attempt=next_attempt,
    )
    _retry_seed = council_researcher_seed_excerpts(frame)
    verifier_body = build_council_verifier_task_body(frame)
    contrarian_body = (
        f"Captain frame:\n{frame}\n\n"
        f"{COUNCIL_CONTRARIAN_HINT}"
    )
    researcher_row = ks.create_task(
        title=f"[Council] Researcher (retry {next_attempt})",
        body=researcher_body,
        assignee="council_researcher",
        priority=1,
        origin_thread_id=tid,
        metadata={
            **shared,
            COUNCIL_SEAT_META: "researcher",
            COUNCIL_RESEARCHER_HAS_SEED_META: bool(_retry_seed),
            "acceptance": _acceptance_for_seat("researcher"),
        },
        idempotency_key=f"council_retry_r_{next_attempt}_{bid}",
    )
    verifier_row = ks.create_task(
        title=f"[Council] Verifier (retry {next_attempt})",
        body=verifier_body,
        assignee="council_verifier",
        priority=2,
        parent_ids=[researcher_row["id"]],
        origin_thread_id=tid,
        metadata={
            **shared,
            COUNCIL_SEAT_META: "verifier",
            "acceptance": _acceptance_for_seat("verifier"),
        },
        idempotency_key=f"council_retry_v_{next_attempt}_{bid}",
    )
    contrarian_row = ks.create_task(
        title=f"[Council] Contrarian (retry {next_attempt})",
        body=contrarian_body,
        assignee="council_contrarian",
        priority=3,
        parent_ids=[verifier_row["id"]],
        origin_thread_id=tid,
        metadata={
            **shared,
            COUNCIL_SEAT_META: "contrarian",
            "acceptance": _acceptance_for_seat("contrarian"),
        },
        idempotency_key=f"council_retry_c_{next_attempt}_{bid}",
    )
    try:
        nudge_dispatcher()
    except Exception:
        _log.debug("council retry dispatcher nudge failed", exc_info=True)
    return {
        "ok": True,
        "pipeline_attempt": next_attempt,
        "researcher_task_id": researcher_row["id"],
        "verifier_task_id": verifier_row["id"],
        "contrarian_task_id": contrarian_row["id"],
        "retry_reason": why,
    }


def maybe_schedule_researcher_retry_after_verifier(
    *,
    thread_id: str,
    batch_id: str,
    verifier_task_id: str,
) -> bool:
    """Return True when a researcher retry chain was spawned (Contrarian deferred)."""
    try:
        out = spawn_researcher_retry_pipeline(
            origin_thread_id=thread_id,
            batch_id=batch_id,
            verifier_task_id=verifier_task_id,
        )
    except Exception:
        _log.warning("council researcher retry spawn failed", exc_info=True)
        return False
    if not out.get("ok"):
        if out.get("already_spawned"):
            return True
        _log.warning(
            "council researcher retry refused for verifier %s batch %s: %s",
            verifier_task_id,
            batch_id,
            out.get("reason") or "unknown",
        )
        return False
    try:
        from utils.kanban_activity import emit_council_seat_activity

        emit_council_seat_activity(
            event_type="council_researcher_retry",
            task_id=str(out.get("researcher_task_id") or ""),
            seat="researcher",
            batch_id=batch_id,
            origin_thread_id=thread_id,
            status="queued",
            assignee="council_researcher",
            summary=str(out.get("retry_reason") or "Researcher retry scheduled"),
        )
    except Exception:
        _log.debug("council retry activity emit failed", exc_info=True)
    return True


def load_seat_result_metadata(task_id: str) -> dict[str, Any]:
    """Load metadata from the latest completed run for a council seat task."""
    tid = (task_id or "").strip()
    if not tid:
        return {}
    try:
        runs = ks.list_runs(tid, limit=10)
    except Exception:
        _log.debug("list_runs failed for %s", tid, exc_info=True)
        return {}
    for run in runs:
        if (run.get("outcome") or "") != ks.OUTCOME_COMPLETED:
            continue
        md = run.get("metadata")
        if isinstance(md, dict):
            return md
    return {}


_FRAME_PRODUCT_KEYWORDS: tuple[str, ...] = (
    "mugi",
    "kanban",
    "dispatcher",
    "council",
    "mesh",
    "worker",
    "deadline",
    "subagent",
    "cooperative",
    "completion",
    "wcr",
    "harness",
)

_GENERIC_EVIDENCE_MARKERS: tuple[str, ...] = (
    "gathered search/read tool results",
    "gathered search results",
    "gathered search/read",
)


def _frame_expects_product_evidence(frame: str) -> bool:
    lower = (frame or "").lower()
    return any(kw in lower for kw in _FRAME_PRODUCT_KEYWORDS)


def council_verifier_tool_names_for_frame(frame: str) -> frozenset[str]:
    """Verifier allowlist: ``read_webpage`` on general frames only."""
    from subagent_runner import SUBAGENT_PROFILES

    profile = SUBAGENT_PROFILES.get("council_verifier") or {}
    raw = profile.get("tools")
    base = frozenset(str(t) for t in raw if str(t).strip()) if raw else frozenset()
    if _frame_expects_product_evidence(frame):
        return base
    return base | frozenset({"read_webpage"})


def _claim_looks_like_snippet_dump(text: str) -> bool:
    lower = (text or "").lower()
    return "title:" in lower and ("url:" in lower or "http" in lower)


def _evidence_entries(evidence: Any) -> list[str]:
    if not isinstance(evidence, list):
        return []
    return [str(e).strip() for e in evidence if str(e).strip()]


def _evidence_is_repo_path(entry: str) -> bool:
    s = (entry or "").strip().replace("\\", "/")
    if not s or s.startswith("http"):
        return False
    return "/" in s and any(s.endswith(ext) for ext in (".py", ".md", ".toml", ".swift"))


def _claim_evidence_web_only(evidence: list[str]) -> bool:
    if not evidence:
        return True
    return all(
        e.lower().startswith("http")
        or e.lower() in _GENERIC_EVIDENCE_MARKERS
        for e in evidence
    )


def _claim_addresses_product_frame(frame: str, claim_text: str, evidence: list[str]) -> bool:
    if any(_evidence_is_repo_path(e) for e in evidence):
        return True
    frame_words = {
        w
        for w in re.split(r"[^\w]+", (frame or "").lower())
        if len(w) >= 5 and w not in _FRAME_PRODUCT_KEYWORDS
    }
    text_lower = (claim_text or "").lower()
    if frame_words and any(w in text_lower for w in frame_words):
        return True
    return any(kw in text_lower for kw in _FRAME_PRODUCT_KEYWORDS if kw in (frame or "").lower())


def audit_researcher_claims_for_frame(
    frame: str,
    claims: list[dict[str, Any]],
    *,
    researcher_completion_salvage: bool = False,
) -> list[dict[str, str]]:
    """Heuristic frame-fit audit — returns suggested rejections [{claim_id, reason}]."""
    product_frame = _frame_expects_product_evidence(frame)
    rejections: list[dict[str, str]] = []
    for raw in claims:
        if not isinstance(raw, dict):
            continue
        cid = str(raw.get("id") or "").strip()
        if not cid:
            continue
        text = str(raw.get("text") or "").strip()
        evidence = _evidence_entries(raw.get("evidence"))
        if _claim_looks_like_snippet_dump(text):
            rejections.append(
                {
                    "claim_id": cid,
                    "reason": "claim text is search snippet boilerplate, not a synthesized finding",
                }
            )
            continue
        # Salvage + web-only evidence is only disqualifying for product frames: a
        # general-knowledge frame has no evidence channel other than the web, so
        # rejecting on the salvage flag alone made every salvaged general-frame
        # council run fail deterministically (kcn_c32156cfe1b340e6).
        if product_frame and researcher_completion_salvage and _claim_evidence_web_only(evidence):
            rejections.append(
                {
                    "claim_id": cid,
                    "reason": "Researcher completion salvage with web-only evidence — not frame-grounded",
                }
            )
            continue
        if product_frame and _claim_evidence_web_only(evidence):
            rejections.append(
                {
                    "claim_id": cid,
                    "reason": "product-specific frame but evidence is generic web URLs only",
                }
            )
            continue
        if product_frame and not _claim_addresses_product_frame(frame, text, evidence):
            rejections.append(
                {
                    "claim_id": cid,
                    "reason": "claim does not address the Captain's product frame",
                }
            )
    return rejections


def validate_verifier_frame_fit(
    frame: str,
    researcher_claims: list[dict[str, Any]],
    verifier_metadata: dict[str, Any],
    *,
    researcher_completion_salvage: bool = False,
) -> tuple[bool, str]:
    """Fail closed when Verifier marked supported but objective audit says unsupported."""
    md = verifier_metadata if isinstance(verifier_metadata, dict) else {}
    verified = md.get("verified_claims") if isinstance(md.get("verified_claims"), list) else []
    supported_ids = {
        _verifier_entry_claim_id(row)
        for row in verified
        if isinstance(row, dict)
        and str(row.get("status") or "").strip().lower() == "supported"
    }
    supported_ids.discard("")
    audit = audit_researcher_claims_for_frame(
        frame,
        researcher_claims,
        researcher_completion_salvage=researcher_completion_salvage,
    )
    for row in audit:
        cid = str(row.get("claim_id") or "").strip()
        if cid and cid in supported_ids:
            reason = str(row.get("reason") or "frame-fit audit failed").strip()
            return False, f"claim {cid} marked supported but fails frame-fit audit: {reason}"
    return True, ""


def council_verifier_context_block(frame: str) -> str:
    """Rubric + optional product seed so Verifier can judge frame fit."""
    if _frame_expects_product_evidence(frame):
        evidence_rule = (
            "2) Evidence fit — this is a **product/implementation frame**: repo paths "
            "(utils/…, docs/…) substantiate; bare blog URLs alone do not support "
            "implementation claims."
        )
        salvage_rule = (
            "4) Salvage handoffs — Researcher completion_salvage + web-only evidence → "
            "unsupported or unknown."
        )
    else:
        evidence_rule = (
            "2) Evidence fit — this is a **general (non-product) frame**: reputable web "
            "sources (official docs, benchmarks, vendor releases, reputable publications) "
            "are valid evidence. Do NOT reject a claim merely because its evidence is a "
            "URL. Reject when evidence is irrelevant, snippet boilerplate, or — for "
            "time-sensitive frames (latest/best X) — clearly stale relative to the "
            "current date."
        )
        salvage_rule = (
            "4) Salvage handoffs — for this general frame, a completion_salvage flag is "
            "NOT a rejection reason by itself; web evidence is the expected channel. "
            "Judge each claim on frame fit and evidence quality."
        )
    lines = [
        "Verification rubric (apply to EACH parent claim_id):",
        "1) Frame fit — claim must bear on the Captain's framed decision, not generic theory alone.",
        evidence_rule,
        "3) Snippet dumps — claim text embedding Title:/URL:/Snippet: → unsupported.",
        salvage_rule,
        "5) Definitions-only support is invalid — 'provides technical definitions' is not sufficient.",
        "Default to unsupported when evidence does not substantiate the claim for THIS frame.",
    ]
    seed = council_researcher_seed_excerpts(frame, max_total_chars=4000)
    if seed:
        lines.extend(["", "Reference product source (same seed Researcher received):", seed])
    return "\n".join(lines)


def build_council_verifier_task_body(frame: str) -> str:
    frame_s = (frame or "").strip()
    return "\n\n".join(
        part
        for part in (
            f"Captain frame:\n{frame_s}" if frame_s else "",
            council_verifier_context_block(frame_s),
            council_verifier_hint_for_frame(frame_s),
        )
        if part
    )


def _researcher_claims_from_parent_ids(parent_task_ids: list[str]) -> tuple[list[dict[str, Any]], bool]:
    claims: list[dict[str, Any]] = []
    salvage = False
    for pid in parent_task_ids:
        ptask = ks.get_task(pid)
        if not ptask or _seat_from_task(ptask) != "researcher":
            continue
        md = load_seat_result_metadata(pid)
        if bool(md.get("completion_salvage")):
            salvage = True
        raw_claims = md.get("claims") if isinstance(md.get("claims"), list) else []
        for raw in raw_claims:
            if isinstance(raw, dict):
                claims.append(raw)
    return claims, salvage


def build_parent_claim_index(parent_task_id: str) -> dict[str, Any]:
    md = load_seat_result_metadata(parent_task_id)
    claims = md.get("claims") if isinstance(md.get("claims"), list) else []
    index: list[dict[str, Any]] = []
    ids: set[str] = set()
    for raw in claims:
        if not isinstance(raw, dict):
            continue
        cid = str(raw.get("id") or "").strip()
        if not cid or cid in ids:
            continue
        ids.add(cid)
        evidence = raw.get("evidence") if isinstance(raw.get("evidence"), list) else []
        index.append(
            {
                "id": cid,
                "text": str(raw.get("text") or "")[:500],
                "evidence": [str(e) for e in evidence if str(e).strip()][:8],
                "confidence": str(raw.get("confidence") or "").strip(),
            }
        )
    out: dict[str, Any] = {"claim_ids": sorted(ids), "claims": index}
    if md.get("completion_salvage"):
        out["researcher_completion_salvage"] = True
    return out


def _verifier_entry_claim_id(entry: dict[str, Any]) -> str:
    """Canonical claim_id from verifier verified_claims / rejections rows."""
    if entry.get("claim_id"):
        return str(entry.get("claim_id") or "").strip()
    return str(entry.get("id") or "").strip()


def validate_parent_claim_refs(
    seat: str,
    result_metadata: dict[str, Any],
    parent_task_ids: list[str],
    *,
    task: dict[str, Any] | None = None,
) -> tuple[bool, str]:
    if seat != "verifier":
        return True, ""
    md = result_metadata if isinstance(result_metadata, dict) else {}
    allowed: set[str] = set()
    task_round = _round_from_task(task) if task else 1
    if task_round >= 2 and task:
        tmeta = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
        bid = str(tmeta.get(COUNCIL_BATCH_ID_META) or "").strip()
        tid = (task.get("origin_thread_id") or "").strip()
        researcher = _researcher_task_for_batch(tid, bid) if tid and bid else None
        if researcher:
            idx = build_parent_claim_index(researcher["id"])
            for cid in idx.get("claim_ids") or []:
                allowed.add(str(cid))
    if not allowed:
        for pid in parent_task_ids or []:
            ptask = ks.get_task(pid)
            if ptask and _seat_from_task(ptask) == "researcher":
                idx = build_parent_claim_index(pid)
                for cid in idx.get("claim_ids") or []:
                    allowed.add(str(cid))
            elif task_round <= 1:
                idx = build_parent_claim_index(pid)
                for cid in idx.get("claim_ids") or []:
                    allowed.add(str(cid))
    if not allowed:
        return False, "verifier requires a completed Researcher parent with claims"
    verified = md.get("verified_claims") if isinstance(md.get("verified_claims"), list) else []
    rejections = md.get("rejections") if isinstance(md.get("rejections"), list) else []
    if not verified and not rejections:
        return False, "verifier must emit verified_claims and/or rejections"
    for entry in verified:
        if not isinstance(entry, dict):
            continue
        if entry.get("id") and not entry.get("claim_id"):
            return False, "verified_claims must use claim_id (not id)"
    for entry in rejections:
        if not isinstance(entry, dict):
            continue
        if entry.get("id") and not entry.get("claim_id"):
            return False, "rejections must use claim_id (not id)"
    unknown: list[str] = []
    addressed: set[str] = set()
    for key in ("verified_claims", "rejections"):
        entries = md.get(key) if isinstance(md.get(key), list) else []
        for entry in entries:
            if not isinstance(entry, dict):
                continue
            cid = _verifier_entry_claim_id(entry)
            if not cid:
                continue
            addressed.add(cid)
            if cid not in allowed:
                unknown.append(cid)
    if unknown:
        return False, "unknown claim_id(s) not in Researcher output: " + ", ".join(sorted(set(unknown)))
    missing = sorted(allowed - addressed)
    if missing:
        return False, "verifier must address every parent claim_id; missing: " + ", ".join(missing)
    return True, ""


def _validate_researcher_claims(metadata: dict[str, Any]) -> tuple[bool, str]:
    claims = metadata.get("claims") if isinstance(metadata.get("claims"), list) else []
    if len(claims) < 1:
        return False, "researcher requires at least one claim"
    seen: set[str] = set()
    uncertainties = metadata.get("uncertainties") if isinstance(metadata.get("uncertainties"), list) else []
    for raw in claims:
        if not isinstance(raw, dict):
            return False, "claims entries must be objects"
        cid = str(raw.get("id") or "").strip()
        if not cid:
            return False, "each claim requires a non-empty id"
        if cid in seen:
            return False, f"duplicate claim id {cid!r}"
        seen.add(cid)
        evidence = raw.get("evidence") if isinstance(raw.get("evidence"), list) else []
        if not evidence and not uncertainties:
            return False, f"claim {cid!r} needs evidence or uncertainties must explain the gap"
    return True, ""


def _assumptions_challenged_as_list(raw: Any) -> list[str]:
    """Accept list or single non-empty string (common Gemma kanban-result shape)."""
    if isinstance(raw, list):
        return [str(x).strip() for x in raw if str(x).strip()]
    if isinstance(raw, str):
        text = raw.strip()
        return [text] if text else []
    return []


def _validate_contrarian_structure(metadata: dict[str, Any]) -> tuple[bool, str]:
    dissent = metadata.get("dissent_items") if isinstance(metadata.get("dissent_items"), list) else []
    if len(dissent) < 2:
        return False, "contrarian requires at least two dissent_items"
    assumptions = _assumptions_challenged_as_list(metadata.get("assumptions_challenged"))
    if len(assumptions) < 1:
        return False, "contrarian requires assumptions_challenged"
    counter = str(metadata.get("strongest_counter") or "").strip()
    if not counter:
        return False, "contrarian requires strongest_counter"
    failures = metadata.get("failure_modes") if isinstance(metadata.get("failure_modes"), list) else []
    if len(failures) < 1:
        return False, "contrarian requires failure_modes"
    return True, ""


def apply_council_seat_verify(
    verifier_ok: bool,
    verifier_reason: str,
    verifier_detail: str,
    parsed: dict[str, Any],
    *,
    task: dict[str, Any],
    task_metadata: dict[str, Any] | None = None,
) -> tuple[bool, str, str]:
    """Run council extended checks after base verify_kanban_result (local + mesh workers)."""
    task_meta = task_metadata if isinstance(task_metadata, dict) else {}
    if not verifier_ok or task_meta.get("origin") != COUNCIL_ORIGIN:
        return verifier_ok, verifier_reason, verifier_detail
    try:
        ext_ok, ext_reason, ext_detail = verify_council_seat_extended(
            parsed,
            task=task,
            task_metadata=task_meta,
        )
        if not ext_ok:
            return ext_ok, ext_reason, ext_detail
    except Exception:
        _log.debug("council seat extended verify failed", exc_info=True)
        return False, FAILURE_VERIFIER_FAILED, "council validation error"
    return verifier_ok, verifier_reason, verifier_detail


def verify_council_seat_extended(
    parsed: dict[str, Any],
    *,
    task: dict[str, Any],
    task_metadata: dict[str, Any] | None = None,
) -> tuple[bool, str, str]:
    """Declarative acceptance + council-specific structural checks."""
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    if tm.get("origin") != COUNCIL_ORIGIN:
        ok, reason, detail = verify_kanban_result(parsed, task_metadata=tm)
        return ok, reason, detail

    ok, reason, detail = verify_kanban_result(parsed, task_metadata=tm)
    if not ok:
        return ok, reason, detail

    md = parsed.get("metadata") if isinstance(parsed.get("metadata"), dict) else {}
    seat = str(tm.get(COUNCIL_SEAT_META) or md.get(COUNCIL_SEAT_META) or "").strip().lower()
    if seat == "researcher":
        rok, rdetail = _validate_researcher_claims(md)
        if not rok:
            return False, FAILURE_VERIFIER_FAILED, rdetail
    elif seat == "contrarian":
        cok, cdetail = _validate_contrarian_structure(md)
        if not cok:
            return False, FAILURE_VERIFIER_FAILED, cdetail
    elif seat == "verifier":
        parent_ids = task.get("parent_ids") if isinstance(task.get("parent_ids"), list) else []
        if not parent_ids:
            pid = str(task.get("parent_id") or "").strip()
            parent_ids = [pid] if pid else []
        vok, vdetail = validate_parent_claim_refs(seat, md, parent_ids, task=task)
        if not vok:
            return False, FAILURE_VERIFIER_FAILED, vdetail
        if _round_from_task(task) <= 1:
            frame = str(tm.get(COUNCIL_FRAME_META) or "").strip()
            researcher_claims, salvage = _researcher_claims_from_parent_ids(parent_ids)
            if frame and researcher_claims:
                fok, fdetail = validate_verifier_frame_fit(
                    frame,
                    researcher_claims,
                    md,
                    researcher_completion_salvage=salvage,
                )
                if not fok:
                    return False, FAILURE_VERIFIER_FAILED, fdetail
    return True, "ok", ""


def council_seat_display_label(seat: str) -> str:
    key = (seat or "").strip().lower()
    if key in _SEAT_LABELS:
        return _SEAT_LABELS[key]
    if key.startswith("council_"):
        return council_seat_display_label(key.removeprefix("council_"))
    return key or "Council"


def format_council_status_snapshot(tasks: list[dict[str, Any]]) -> str:
    lines: list[str] = []
    task_list = list(tasks or [])

    def _latest(seat: str, *, verifier_round: int | None = None) -> dict[str, Any] | None:
        rows = []
        for t in task_list:
            if _seat_from_task(t) != seat:
                continue
            if verifier_round is not None:
                if verifier_round >= 2 and _round_from_task(t) < 2:
                    continue
                if verifier_round <= 1 and _round_from_task(t) >= 2:
                    continue
            if (t.get("status") or "").strip() == "cancelled":
                continue
            rows.append(t)
        if not rows:
            return None
        return max(
            rows,
            key=lambda row: (
                _pipeline_attempt_from_task(row),
                str(row.get("updated_at") or row.get("created_at") or ""),
            ),
        )

    researcher = _latest("researcher")
    verifier_r1 = _latest("verifier", verifier_round=1)
    contrarian = _latest("contrarian")
    verifier_r2 = _latest("verifier", verifier_round=2)
    for t, label in (
        (researcher, council_seat_display_label("researcher")),
        (verifier_r1, council_seat_display_label("verifier")),
        (contrarian, council_seat_display_label("contrarian")),
        (verifier_r2, f"{council_seat_display_label('verifier')} (round 2)"),
    ):
        if not t:
            continue
        lines.append(f"- {label}: {t.get('status', '?')}")
    return "\n".join(lines)


def humanize_council_failure_detail(raw: str, *, seat: str = "") -> str:
    """Map internal validator/dispatcher jargon to operator-friendly copy."""
    detail = (raw or "").strip()
    seat_label = council_seat_display_label(seat) if seat else "A council seat"
    lower = detail.lower()
    if not detail:
        return f"{seat_label} did not meet the completion bar."
    if "parse_failure" in lower or "parseable result" in lower or "kanban-result" in lower:
        return (
            f"{seat_label} finished without a valid handoff block. "
            "Open the Kanban board to inspect that seat's run."
        )
    if "unknown claim_id" in lower:
        return (
            f"{seat_label} referenced claims the Researcher did not produce. "
            "Open the Kanban board to compare Researcher and Verifier output."
        )
    if "contrarian requires assumptions_challenged" in lower:
        return (
            f"{council_seat_display_label('contrarian')} handoff used the wrong metadata shape "
            "(assumptions_challenged must be a JSON array, not a single string). "
            "Open the Kanban board to inspect that seat's run."
        )
    if "contrarian requires at least two dissent_items" in lower:
        return (
            f"{council_seat_display_label('contrarian')} did not produce enough structured dissent. "
            "No consensus synthesis was attempted."
        )
    if "contrarian requires failure_modes" in lower:
        return (
            f"{council_seat_display_label('contrarian')} handoff was missing failure_modes. "
            "Open the Kanban board to inspect that seat's run."
        )
    if "contrarian requires strongest_counter" in lower:
        return (
            f"{council_seat_display_label('contrarian')} handoff was missing strongest_counter. "
            "Open the Kanban board to inspect that seat's run."
        )
    if "contrarian requires" in lower and "falsifiers" in lower:
        return (
            f"{council_seat_display_label('contrarian')} handoff was missing falsifiers. "
            "Open the Kanban board to inspect that seat's run."
        )
    if "contrarian" in (seat or "").lower() and "dissent" in lower and "requires" not in lower:
        return (
            f"{council_seat_display_label('contrarian')} did not produce enough structured dissent. "
            "No consensus synthesis was attempted."
        )
    if "validation error" in lower or "validation failed" in lower:
        # Generic validation failure (e.g. a Verifier verdict whose structure/metadata could
        # not be validated). Distinguish this from a crash AND from a valid rejection: the
        # seat ran and produced output, but it failed verification, so the pipeline stopped
        # rather than act on an unverified result.
        return (
            f"{seat_label} produced output that could not be validated, so the pipeline "
            "stopped rather than act on an unverified result. This is not a crash — open "
            "the Kanban board to inspect that seat's run."
        )
    if len(detail) > 240:
        return detail[:237] + "…"
    return detail


_AUTH_REDACT_RE = re.compile(
    r"(?i)\b(authorization|x-auth-token|api[_-]?key|bearer)\b\s*[:=]\s*\S+"
)


def _redact_sensitive_text(text: str) -> str:
    return _AUTH_REDACT_RE.sub(r"\1: [redacted]", text or "")


def _redact_council_value(value: Any) -> Any:
    if isinstance(value, str):
        return _redact_sensitive_text(value)
    if isinstance(value, list):
        return [_redact_council_value(v) for v in value]
    if isinstance(value, dict):
        return {k: _redact_council_value(v) for k, v in value.items()}
    return value


def batch_belongs_to_thread(*, thread_id: str, batch_id: str) -> bool:
    tid = (thread_id or "").strip()
    bid = (batch_id or "").strip()
    if not tid or not bid:
        return False
    tasks = tasks_for_council_batch(tid, bid)
    return len(tasks) >= 1


def build_council_transcript(*, thread_id: str, batch_id: str) -> dict[str, Any]:
    """Structured council transcript for HTTP/API consumers (evidence redacted)."""
    tid = (thread_id or "").strip()
    bid = (batch_id or "").strip()
    if not tid or not bid:
        raise ValueError("thread_id and batch_id are required")
    if not batch_belongs_to_thread(thread_id=tid, batch_id=bid):
        raise ValueError("batch does not belong to thread")
    tasks = tasks_for_council_batch(tid, bid)
    frame = ""
    seats: list[dict[str, Any]] = []
    seat_tasks: list[tuple[str, dict[str, Any]]] = []
    researcher = _researcher_task_for_batch(tid, bid)
    if researcher:
        seat_tasks.append(("researcher", researcher))
    v1 = _verifier_r1_task_for_batch(tid, bid)
    if v1:
        seat_tasks.append(("verifier", v1))
    contrarian = _contrarian_task_for_batch(tid, bid)
    if contrarian:
        seat_tasks.append(("contrarian", contrarian))
    verifier_r2 = next(
        (t for t in tasks if _seat_from_task(t) == "verifier" and _round_from_task(t) >= 2),
        None,
    )
    if verifier_r2:
        seat_tasks.append(("verifier_r2", verifier_r2))
    for seat_key, task in seat_tasks:
        meta = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
        if not frame:
            frame = str(meta.get(COUNCIL_FRAME_META) or "").strip()
        run_md = load_seat_result_metadata(task["id"])
        summary = ""
        try:
            runs = ks.list_runs(task["id"], limit=3)
            for run in runs:
                if (run.get("outcome") or "") == ks.OUTCOME_COMPLETED:
                    summary = str(run.get("summary") or "").strip()
                    break
        except Exception:
            pass
        label = (
            f"{council_seat_display_label('verifier')} (round 2)"
            if seat_key == "verifier_r2"
            else council_seat_display_label(seat_key)
        )
        seats.append(
            {
                "seat": seat_key,
                "label": label,
                "task_id": task.get("id"),
                "status": task.get("status"),
                "summary": _redact_sensitive_text(summary),
                "metadata": _redact_council_value(run_md),
            }
        )
    return {
        "thread_id": tid,
        "batch_id": bid,
        "council_frame": frame,
        "seats": seats,
    }


def format_council_transcript_text(transcript: dict[str, Any]) -> str:
    lines: list[str] = []
    frame = str(transcript.get("council_frame") or "").strip()
    if frame:
        lines.extend(["Captain frame:", frame, ""])
    for seat in transcript.get("seats") or []:
        if not isinstance(seat, dict):
            continue
        label = str(seat.get("label") or seat.get("seat") or "Seat")
        status = str(seat.get("status") or "?")
        summary = str(seat.get("summary") or "").strip()
        lines.append(f"### {label} ({status})")
        if summary:
            lines.append(summary)
        lines.append("")
    return "\n".join(lines).strip()


def stop_council_batch(
    *,
    thread_id: str,
    reason: str = "operator_stop",
) -> dict[str, Any]:
    """Cancel an in-flight council batch and surface a failure delivery."""
    tid = (thread_id or "").strip()
    if not tid:
        return {"ok": False, "error": "thread_id required", "status": 400}
    bid = probe_council_batch_id(tid)
    if not bid:
        clear_active_council_batch(tid)
        return {"ok": False, "error": "no_active_council", "status": 404}
    if not probe_council_active(tid):
        clear_active_council_batch(tid)
        return {"ok": False, "error": "no_active_council", "status": 404}

    tasks = tasks_for_council_batch(tid, bid)
    researcher = next((t for t in tasks if _seat_from_task(t) == "researcher"), None)
    cancelled = False
    cancel_errors: list[str] = []
    if researcher and (researcher.get("status") or "") not in COUNCIL_TERMINAL_STATUSES:
        try:
            ks.cancel_task(researcher["id"], cascade=True)
            cancelled = True
        except Exception:
            _log.warning("council stop cancel failed for %s", researcher.get("id"), exc_info=True)
            cancel_errors.append("researcher")
    if not cancelled:
        for t in tasks:
            st = (t.get("status") or "").strip()
            if st in COUNCIL_TERMINAL_STATUSES:
                continue
            try:
                ks.cancel_task(t["id"], cascade=False)
                cancelled = True
            except Exception:
                _log.debug("council stop cancel %s failed", t.get("id"), exc_info=True)
                cancel_errors.append(str(t.get("id") or "task"))

    detail_by_reason = {
        "operator_stop": "You stopped the council run.",
        "deadline_exceeded": (
            "Council exceeded the configured time limit. "
            "Open the Kanban board to inspect partial seat output."
        ),
    }
    seat_label = "council"
    if reason == "operator_stop":
        seat_label = "operator"
    from utils.council_delivery import council_failure_already_delivered, on_pipeline_failed

    if not council_failure_already_delivered(tid):
        on_pipeline_failed(
            thread_id=tid,
            batch_id=bid,
            seat=seat_label,
            detail=detail_by_reason.get(reason, "Council was stopped before completion."),
        )
    try:
        nudge_dispatcher()
    except Exception:
        _log.debug("council stop dispatcher nudge failed", exc_info=True)
    result: dict[str, Any] = {
        "ok": True,
        "council_batch_id": bid,
        "reason": reason,
    }
    if cancel_errors:
        result["cancel_warning"] = (
            "Council was stopped, but some seat tasks may still be winding down on the board."
        )
    return result


def _parse_utc_iso_timestamp(raw: str) -> datetime | None:
    ts = (raw or "").strip()
    if not ts:
        return None
    try:
        if ts.endswith("Z"):
            ts = ts[:-1] + "+00:00"
        return datetime.fromisoformat(ts)
    except ValueError:
        return None


def _council_deadline_limit_seconds() -> int:
    cfg = council_settings()
    raw = cfg.get("council_batch_deadline_seconds", 3600)
    if raw is None:
        return 3600
    try:
        return max(0, int(raw))
    except (TypeError, ValueError):
        return 3600


def _iter_open_council_batches_for_deadline() -> list[tuple[str, str]]:
    """Return (thread_id, batch_id) for batches with at least one non-terminal council seat."""
    seen: set[tuple[str, str]] = set()
    out: list[tuple[str, str]] = []
    try:
        from chat_storage import list_thread_ids_with_council_state

        for tid in list_thread_ids_with_council_state():
            bid = probe_council_batch_id(tid)
            if not bid or not probe_council_active(tid):
                continue
            key = (tid, bid)
            if key not in seen:
                seen.add(key)
                out.append(key)
    except Exception:
        _log.debug("council deadline pointer scan failed", exc_info=True)
    try:
        rows = ks.list_tasks(limit=150, max_limit=150, order="desc")
    except Exception:
        rows = []
    for row in rows or []:
        meta = row.get("metadata") if isinstance(row.get("metadata"), dict) else {}
        if meta.get("origin") != COUNCIL_ORIGIN:
            continue
        if (row.get("status") or "") in COUNCIL_TERMINAL_STATUSES:
            continue
        tid = (row.get("origin_thread_id") or "").strip()
        bid = str(meta.get(COUNCIL_BATCH_ID_META) or "").strip()
        if not tid or not bid:
            continue
        key = (tid, bid)
        if key in seen:
            continue
        if probe_council_active(tid):
            seen.add(key)
            out.append(key)
    return out


def _batch_deadline_start_iso(thread_id: str, batch_id: str) -> str:
    try:
        from chat_storage import get_thread_council_activated_at

        ts = get_thread_council_activated_at(thread_id)
        if ts:
            return ts
    except Exception:
        pass
    earliest = ""
    for t in tasks_for_council_batch(thread_id, batch_id):
        created = str(t.get("created_at") or "").strip()
        if created and (not earliest or created < earliest):
            earliest = created
    return earliest


def reconcile_council_batch_deadlines() -> int:
    """Enforce ``council_batch_deadline_seconds`` for open batches. Returns kill count."""
    limit_sec = _council_deadline_limit_seconds()
    if limit_sec <= 0:
        return 0

    killed = 0
    now = datetime.now(tz=timezone.utc)
    for tid, bid in _iter_open_council_batches_for_deadline():
        activated_raw = _batch_deadline_start_iso(tid, bid)
        if not activated_raw:
            continue
        activated = _parse_utc_iso_timestamp(activated_raw)
        if activated is None:
            _log.debug("invalid council deadline start for %s: %s", tid, activated_raw)
            continue
        if activated.tzinfo is None:
            activated = activated.replace(tzinfo=timezone.utc)
        elapsed = (now - activated).total_seconds()
        if elapsed <= limit_sec:
            continue
        result = stop_council_batch(thread_id=tid, reason="deadline_exceeded")
        if result.get("ok"):
            killed += 1
    return killed


def build_round2_verifier_body(
    *,
    council_frame: str,
    thread_id: str,
    batch_id: str,
    contrarian_task_id: str,
) -> str:
    blocks: list[str] = []
    frame = (council_frame or "").strip()
    if frame:
        blocks.append(f"Captain frame:\n{frame}")
    researcher = _researcher_task_for_batch(thread_id, batch_id)
    if researcher:
        idx = build_parent_claim_index(researcher["id"])
        blocks.append(
            "─── Round 1 claim index (RECORDED, not authoritative) ───\n"
            + json.dumps(idx, ensure_ascii=False, indent=2)
        )
    contrarian_md = load_seat_result_metadata(contrarian_task_id)
    if contrarian_md:
        blocks.append(
            "─── Contrarian objections (RECORDED) ───\n"
            + json.dumps(
                {
                    "dissent_items": contrarian_md.get("dissent_items"),
                    "assumptions_challenged": contrarian_md.get("assumptions_challenged"),
                    "strongest_counter": contrarian_md.get("strongest_counter"),
                    "failure_modes": contrarian_md.get("failure_modes"),
                },
                ensure_ascii=False,
                indent=2,
            )
        )
    verifier_r1 = _verifier_r1_task_for_batch(thread_id, batch_id)
    if verifier_r1:
        v1_md = load_seat_result_metadata(verifier_r1["id"])
        blocks.append(
            "─── Verifier Round 1 result (RECORDED) ───\n"
            + json.dumps(
                {
                    "verified_claims": v1_md.get("verified_claims"),
                    "rejections": v1_md.get("rejections"),
                },
                ensure_ascii=False,
                indent=2,
            )
        )
    blocks.append(COUNCIL_VERIFIER_R2_HINT)
    return "\n\n".join(blocks)


def spawn_round2_verifier_task(
    *,
    contrarian_task_id: str,
    origin_thread_id: str,
    batch_id: str,
) -> dict[str, Any]:
    """Spawn disagreement-gated Verifier round 2 (parent = Contrarian R1)."""
    cid = (contrarian_task_id or "").strip()
    tid = (origin_thread_id or "").strip()
    bid = (batch_id or "").strip()
    if not cid or not tid or not bid:
        raise ValueError("contrarian_task_id, origin_thread_id, and batch_id are required")
    if batch_has_round2_verifier(tid, bid):
        existing = next(
            t
            for t in tasks_for_council_batch(tid, bid)
            if _seat_from_task(t) == "verifier" and _round_from_task(t) >= 2
        )
        return {"ok": True, "task_id": existing["id"], "already_spawned": True}
    contrarian = ks.get_task(cid)
    if not contrarian:
        raise ValueError("contrarian task not found")
    meta = contrarian.get("metadata") if isinstance(contrarian.get("metadata"), dict) else {}
    frame = str(meta.get(COUNCIL_FRAME_META) or "").strip()
    shared = {
        "origin": COUNCIL_ORIGIN,
        COUNCIL_BATCH_ID_META: bid,
        COUNCIL_FRAME_META: frame,
        COUNCIL_ROUND_META: 2,
        "origin_thread_id": tid,
    }
    body = build_round2_verifier_body(
        council_frame=frame,
        thread_id=tid,
        batch_id=bid,
        contrarian_task_id=cid,
    )
    row = ks.create_task(
        title="[Council] Verifier (round 2)",
        body=body,
        assignee="council_verifier",
        priority=4,
        parent_ids=[cid],
        origin_thread_id=tid,
        metadata={
            **shared,
            COUNCIL_SEAT_META: "verifier",
            "acceptance": _acceptance_for_seat("verifier", council_round=2),
        },
        idempotency_key=f"council_r2_{bid}",
    )
    try:
        nudge_dispatcher()
    except Exception:
        _log.debug("council r2 dispatcher nudge failed", exc_info=True)
    return {"ok": True, "task_id": row["id"], "already_spawned": False}


def _append_seat_synthesis_block(blocks: list[str], label: str, task: dict[str, Any]) -> None:
    md = load_seat_result_metadata(task["id"])
    summary = ""
    try:
        runs = ks.list_runs(task["id"], limit=3)
        for run in runs:
            if (run.get("outcome") or "") == ks.OUTCOME_COMPLETED:
                summary = str(run.get("summary") or "").strip()
                break
    except Exception:
        pass
    blocks.append(
        f"### {label}\nSummary: {summary[:2000]}\nMetadata: "
        f"{json.dumps(md, ensure_ascii=False)[:4000]}"
    )


def _council_frame_for_batch(thread_id: str, batch_id: str) -> str:
    for task in tasks_for_council_batch(thread_id, batch_id):
        meta = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
        frame = str(meta.get(COUNCIL_FRAME_META) or "").strip()
        if frame:
            return frame
    return ""


def build_synthesis_context(
    *,
    thread_id: str,
    batch_id: str,
    r2_incomplete: bool = False,
) -> str:
    """Assemble structured seat summaries for Captain synthesis delivery."""
    from utils.council_prompts import COUNCIL_R2_INCOMPLETE_BANNER

    tasks = tasks_for_council_batch(thread_id, batch_id)
    blocks: list[str] = []
    frame = _council_frame_for_batch(thread_id, batch_id)
    if frame:
        blocks.append(f"Captain frame (answer this for the operator):\n{frame}")
    policy = council_implementation_policy_excerpt(frame) if frame else ""
    if policy:
        blocks.append(policy)
    if r2_incomplete:
        blocks.append(COUNCIL_R2_INCOMPLETE_BANNER)
    sequence: list[tuple[str, dict[str, Any] | None]] = [
        (council_seat_display_label("researcher"), _researcher_task_for_batch(thread_id, batch_id)),
        (
            f"{council_seat_display_label('verifier')} — round 1",
            _verifier_r1_task_for_batch(thread_id, batch_id),
        ),
        (council_seat_display_label("contrarian"), _contrarian_task_for_batch(thread_id, batch_id)),
    ]
    if not r2_incomplete:
        verifier_r2 = next(
            (
                t
                for t in tasks
                if _seat_from_task(t) == "verifier"
                and _round_from_task(t) >= 2
                and (t.get("status") or "") == "done"
            ),
            None,
        )
        if verifier_r2:
            sequence.append(
                (f"{council_seat_display_label('verifier')} — round 2 follow-up", verifier_r2)
            )
    for label, task in sequence:
        if task:
            _append_seat_synthesis_block(blocks, label, task)
    return "\n\n".join(blocks)


def is_council_task_metadata(task_metadata: dict[str, Any] | None) -> bool:
    tm = task_metadata if isinstance(task_metadata, dict) else {}
    if tm.get("origin") == COUNCIL_ORIGIN:
        return True
    return bool(str(tm.get(COUNCIL_SEAT_META) or "").strip())


def build_council_context_supplement(task: dict[str, Any]) -> str:
    """RECORDED parent claim index for Verifier/Contrarian seats."""
    meta = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
    if meta.get("origin") != COUNCIL_ORIGIN:
        return ""
    seat = str(meta.get(COUNCIL_SEAT_META) or "").strip().lower()
    if seat not in ("verifier", "contrarian"):
        return ""
    parent_ids = task.get("parent_ids") if isinstance(task.get("parent_ids"), list) else []
    if not parent_ids:
        pid = str(task.get("parent_id") or "").strip()
        parent_ids = [pid] if pid else []
    blocks: list[str] = []
    task_round = _round_from_task(task)
    if seat == "verifier" and task_round >= 2:
        tid = (task.get("origin_thread_id") or "").strip()
        bid = str(meta.get(COUNCIL_BATCH_ID_META) or "").strip()
        researcher = _researcher_task_for_batch(tid, bid) if tid and bid else None
        if researcher:
            idx = build_parent_claim_index(researcher["id"])
            blocks.append(
                "─── Round 1 claim index (RECORDED, not authoritative) ───\n"
                + json.dumps(idx, ensure_ascii=False, indent=2)
            )
        for pid in parent_ids:
            ptask = ks.get_task(pid)
            if not ptask or _seat_from_task(ptask) != "contrarian":
                continue
            cmd = load_seat_result_metadata(pid)
            blocks.append(
                "─── Contrarian objections (RECORDED) ───\n"
                + json.dumps(
                    {
                        "dissent_items": cmd.get("dissent_items"),
                        "assumptions_challenged": cmd.get("assumptions_challenged"),
                        "strongest_counter": cmd.get("strongest_counter"),
                        "failure_modes": cmd.get("failure_modes"),
                    },
                    ensure_ascii=False,
                    indent=2,
                )
            )
        return "\n\n".join(blocks)
    for pid in parent_ids:
        ptask = ks.get_task(pid)
        if not ptask or _seat_from_task(ptask) != "researcher":
            continue
        idx = build_parent_claim_index(pid)
        blocks.append(
            "─── Parent claim index (RECORDED, not authoritative) ───\n"
            + json.dumps(idx, ensure_ascii=False, indent=2)
        )
    if not blocks:
        # Contrarian's parent is the Verifier (not the Researcher), so the parent walk
        # above never matches for that seat — resolve the Researcher from the batch.
        tid = (task.get("origin_thread_id") or "").strip()
        bid = str(meta.get(COUNCIL_BATCH_ID_META) or "").strip()
        researcher = _researcher_task_for_batch(tid, bid) if tid and bid else None
        if researcher:
            idx = build_parent_claim_index(researcher["id"])
            if idx.get("claim_ids"):
                blocks.append(
                    "─── Researcher claim index (RECORDED, not authoritative) ───\n"
                    + json.dumps(idx, ensure_ascii=False, indent=2)
                )
    return "\n\n".join(blocks)
