"""Council thread delivery + agent wake after Kanban lifecycle events."""

from __future__ import annotations

import logging
import threading
import time
from typing import Callable

from utils.kanban_coordinator_delivery import GUARD_PREFIX
from utils.council_orchestrator import (
    COUNCIL_BATCH_ID_META,
    COUNCIL_ORIGIN,
    COUNCIL_POLICY_HEADER,
    COUNCIL_SEAT_META,
    COUNCIL_TERMINAL_STATUSES,
    TERMINAL_FAILURE_STATUSES,
    build_council_transcript,
    build_synthesis_context,
    clear_active_council_batch,
    council_settings,
    council_synthesis_ready,
    format_council_status_snapshot,
    format_council_transcript_text,
    humanize_council_failure_detail,
    probe_council_batch_id,
    probe_council_open_batch_id,
    resolve_council_batch_id,
    tasks_for_council_batch,
)
from utils.council_prompts import (
    COUNCIL_FAILURE_SUFFIX,
    COUNCIL_SYNTHESIS_SUFFIX,
    COUNCIL_SYNTHESIS_SUFFIX_GENERAL,
)

_log = logging.getLogger(__name__)

_wake_fn: Callable[[str, str], None] | None = None
_chat_reload_fn: Callable[[str], None] | None = None
_thread_state_refresh_fn: Callable[[str], None] | None = None
_wake_lock = threading.Lock()
_last_wake_mono: dict[str, float] = {}
_deferred_wake_timers: dict[str, threading.Timer] = {}
_pending_wake_after_run: dict[str, str] = {}
_pending_wake_fallback_timers: dict[str, threading.Timer] = {}
_WAKE_MIN_INTERVAL_SEC = 30.0
_PENDING_WAKE_TIMEOUT_SEC = 90.0
_WAKE_REASON_PRIORITY: dict[str, int] = {
    "council_synthesis": 3,
    "council_failed": 2,
    "council_honest_stop": 2,
    "council_skip": 2,
    "council_synthesis_delivery_failed": 1,
    "council_event": 0,
}
_SYNTHESIS_BANNER = "Council seats finished"
_FAILURE_BANNER = "Council could not complete"


def should_reuse_council_delivery_for_idle_wake(history: list) -> bool:
    """True when idle-tick wake should not append another guard-marked user bubble."""
    user_checked = 0
    for msg in reversed(history or []):
        if (msg.get("role") or "").strip() != "user":
            continue
        user_checked += 1
        content = msg.get("content")
        if isinstance(content, str) and GUARD_PREFIX in content and "Council" in content:
            return True
        if user_checked >= 8:
            break
    return False


def probe_council_delivery_wake_kind(history: list) -> str | None:
    """Return ``synthesis``, ``failure``, or None for the latest council delivery wake."""
    from utils.council_prompts import (
        COUNCIL_FAILURE_WAKE_PROMPT,
        COUNCIL_SYNTHESIS_WAKE_PROMPT,
    )

    # Only the *latest* user message can be a live delivery wake. A council wake always
    # injects its banner as the most recent user turn and runs immediately. Earlier
    # delivery bubbles ("Council seats finished" / "Council could not complete") are
    # persisted, operator-facing history that lingers in the thread forever — scanning
    # back over them wrongly classifies ordinary later chat as a council wake, which
    # strips every tool (the empty-tool-set thread-poisoning bug). So inspect the most
    # recent user message only.
    for msg in reversed(history or []):
        if (msg.get("role") or "").strip() != "user":
            continue
        content = msg.get("content")
        if not isinstance(content, str):
            return None
        if GUARD_PREFIX in content:
            if _FAILURE_BANNER in content:
                return "failure"
            if _SYNTHESIS_BANNER in content:
                return "synthesis"
        if content.strip().startswith("Council deliberation failed"):
            return "failure"
        if content.strip().startswith("Council deliberation finished"):
            return "synthesis"
        if COUNCIL_FAILURE_WAKE_PROMPT[:48] in content:
            return "failure"
        if COUNCIL_SYNTHESIS_WAKE_PROMPT[:48] in content:
            return "synthesis"
        return None
    return None


def register_council_wake(fn: Callable[[str, str], None]) -> None:
    global _wake_fn
    _wake_fn = fn


def register_council_chat_reload(fn: Callable[[str], None]) -> None:
    global _chat_reload_fn
    _chat_reload_fn = fn


def register_council_thread_state_refresh(fn: Callable[[str], None]) -> None:
    global _thread_state_refresh_fn
    _thread_state_refresh_fn = fn


def notify_council_thread_state_refresh(thread_id: str) -> None:
    """Lightweight banner refresh (no chat history reload)."""
    tid = (thread_id or "").strip()
    if not tid or _thread_state_refresh_fn is None:
        return
    try:
        _thread_state_refresh_fn(tid)
    except Exception:
        _log.debug("council thread_state refresh failed", exc_info=True)


def notify_council_chat_reload(thread_id: str) -> None:
    tid = (thread_id or "").strip()
    if not tid or _chat_reload_fn is None:
        return
    try:
        _chat_reload_fn(tid)
    except Exception:
        _log.debug("council chat reload failed", exc_info=True)


def _cancel_trailing_wake(tid: str) -> None:
    timer = _deferred_wake_timers.pop(tid, None)
    if timer is not None:
        timer.cancel()


def _schedule_trailing_wake(tid: str, reason: str, delay_sec: float) -> None:
    delay = max(0.05, float(delay_sec))

    def _fire() -> None:
        _deferred_wake_timers.pop(tid, None)
        with _wake_lock:
            _last_wake_mono[tid] = time.monotonic()
        _invoke_wake(tid, reason or "council_event")

    _cancel_trailing_wake(tid)
    timer = threading.Timer(delay, _fire)
    timer.daemon = True
    _deferred_wake_timers[tid] = timer
    timer.start()


def _invoke_wake(tid: str, reason: str) -> None:
    if _wake_fn is None:
        return
    try:
        _wake_fn(tid, reason or "council_event")
    except Exception:
        _log.debug("wake_council_thread failed", exc_info=True)


def _merge_pending_wake_reason(prev: str, new: str) -> str:
    prev_v = (prev or "council_event").strip() or "council_event"
    new_v = (new or "council_event").strip() or "council_event"
    if _WAKE_REASON_PRIORITY.get(prev_v, 0) >= _WAKE_REASON_PRIORITY.get(new_v, 0):
        return prev_v
    return new_v


def _should_skip_council_wake(thread_id: str, reason: str) -> bool:
    """Skip duplicate or stale council wakes once synthesis has an assistant reply."""
    r = (reason or "").strip() or "council_event"
    if not council_synthesis_already_answered(thread_id):
        return False
    if r == "council_synthesis":
        _log.info(
            "council synthesis wake skipped for %s — already answered",
            thread_id,
        )
        return True
    # ``/council`` schedules ``council_honest_stop`` with a 90s fallback; that timer
    # must not fire after synthesis already landed (``1bee30d6`` duplicate verdict).
    if r in ("council_honest_stop", "council_skip"):
        _log.info(
            "council %s wake skipped for %s — synthesis already answered",
            r,
            thread_id,
        )
        return True
    return False


def _clear_pending_council_wake_state(tid: str) -> None:
    """Drop queued + fallback-timer wakes after a council wake actually fired."""
    with _wake_lock:
        _pending_wake_after_run.pop(tid, None)
    _cancel_pending_wake_fallback(tid)


def register_pending_council_wake(thread_id: str, *, reason: str = "") -> None:
    tid = (thread_id or "").strip()
    if not tid:
        return
    with _wake_lock:
        prev = _pending_wake_after_run.get(tid, "")
        merged = _merge_pending_wake_reason(prev, reason)
    if _should_skip_council_wake(tid, merged):
        with _wake_lock:
            _pending_wake_after_run.pop(tid, None)
        _cancel_pending_wake_fallback(tid)
        return
    with _wake_lock:
        _pending_wake_after_run[tid] = merged
    _schedule_pending_wake_fallback(tid)


def _cancel_pending_wake_fallback(tid: str) -> None:
    timer = _pending_wake_fallback_timers.pop(tid, None)
    if timer is not None:
        timer.cancel()


def _council_pending_wake_should_defer(tid: str, reason: str) -> bool:
    """Defer fallback while seats are still running (avoid premature synthesis/honest-stop)."""
    if probe_council_open_batch_id(tid):
        return True
    if reason == "council_synthesis":
        bid = probe_council_batch_id(tid)
        if bid and not council_synthesis_ready(tid, bid):
            return True
    return False


def _schedule_pending_wake_fallback(tid: str) -> None:
    def _fire() -> None:
        _pending_wake_fallback_timers.pop(tid, None)
        with _wake_lock:
            reason = _pending_wake_after_run.pop(tid, "")
        if not reason:
            return
        if _should_skip_council_wake(tid, reason):
            return
        if _council_pending_wake_should_defer(tid, reason):
            _log.info(
                "council pending wake fallback deferred for %s — seats still open",
                tid,
            )
            with _wake_lock:
                _pending_wake_after_run[tid] = reason
            _schedule_pending_wake_fallback(tid)
            return
        _log.info("council pending wake fallback for %s", tid)
        wake_council_thread(tid, reason=reason, force=True)

    _cancel_pending_wake_fallback(tid)
    timer = threading.Timer(_PENDING_WAKE_TIMEOUT_SEC, _fire)
    timer.daemon = True
    _pending_wake_fallback_timers[tid] = timer
    timer.start()


def flush_pending_council_wake(thread_id: str) -> None:
    tid = (thread_id or "").strip()
    if not tid:
        return
    _cancel_pending_wake_fallback(tid)
    with _wake_lock:
        reason = _pending_wake_after_run.pop(tid, "")
    if reason and not _should_skip_council_wake(tid, reason):
        wake_council_thread(tid, reason=reason, force=True)


def wake_council_thread(thread_id: str, *, reason: str = "", force: bool = False) -> None:
    tid = (thread_id or "").strip()
    if not tid or _wake_fn is None:
        return
    if _should_skip_council_wake(tid, reason):
        with _wake_lock:
            _pending_wake_after_run.pop(tid, None)
        _cancel_pending_wake_fallback(tid)
        return
    now = time.monotonic()
    with _wake_lock:
        if not force:
            prev = _last_wake_mono.get(tid)
            if prev is not None and (now - prev) < _WAKE_MIN_INTERVAL_SEC:
                _schedule_trailing_wake(tid, reason, _WAKE_MIN_INTERVAL_SEC - (now - prev))
                return
        _cancel_trailing_wake(tid)
        _last_wake_mono[tid] = now
    _invoke_wake(tid, reason)
    _clear_pending_council_wake_state(tid)


def deliver_council_continuation(
    thread_id: str,
    *,
    banner: str,
    body: str,
    suffix: str = COUNCIL_SYNTHESIS_SUFFIX,
    skip_dedupe: bool = False,
) -> bool:
    tid = (thread_id or "").strip()
    if not tid:
        return False
    banner_v = (banner or "Council update").strip()
    if not skip_dedupe and _council_delivery_already_persisted(tid, banner_v):
        return True

    snap = ""
    try:
        bid = probe_council_batch_id(tid) or resolve_council_batch_id(tid)
        if bid:
            snap = format_council_status_snapshot(tasks_for_council_batch(tid, bid))
    except Exception:
        _log.debug("council snapshot for delivery failed", exc_info=True)

    lines = [
        GUARD_PREFIX,
        banner_v,
        "",
        (body or "").strip(),
    ]
    suffix_v = (suffix or "").strip()
    if suffix_v:
        lines.extend(["", suffix_v])
    if snap:
        lines.extend(["", "Council snapshot:", snap])
    content = "\n".join(lines).strip()
    try:
        from chat_storage import append_messages

        last_id = append_messages(tid, [{"role": "user", "content": content}])
        if last_id is None:
            return False
        notify_council_chat_reload(tid)
        notify_council_thread_state_refresh(tid)
        return True
    except Exception:
        _log.warning("deliver_council_continuation append failed", exc_info=True)
        return False


def council_failure_already_delivered(thread_id: str) -> bool:
    """True when a council failure bubble was already appended for this thread."""
    return _council_delivery_already_persisted(thread_id, _FAILURE_BANNER)


def council_synthesis_already_answered(thread_id: str) -> bool:
    """True when synthesis delivery already has a following assistant reply.

    Blocks generic idle_tick from re-elaborating the verdict on the same thread.
    """
    tid = (thread_id or "").strip()
    if not tid:
        return False
    try:
        from chat_storage import get_messages

        rows = get_messages(tid) or []
        synthesis_idx: int | None = None
        for idx, msg in enumerate(rows):
            if (msg.get("role") or "").strip() != "user":
                continue
            content = str(msg.get("content") or "")
            if GUARD_PREFIX in content and _SYNTHESIS_BANNER in content:
                synthesis_idx = idx
        if synthesis_idx is None:
            return False
        for msg in rows[synthesis_idx + 1 :]:
            role = (msg.get("role") or "").strip()
            if role == "assistant" and str(msg.get("content") or "").strip():
                return True
            if role == "user":
                content = str(msg.get("content") or "").strip()
                if not content or content.startswith("init"):
                    continue
                if GUARD_PREFIX in content:
                    continue
                if content.startswith("[") and content.endswith("]"):
                    continue
                return False
    except Exception:
        _log.warning("council_synthesis_already_answered probe failed", exc_info=True)
        # Fail-closed: block idle_tick when synthesis-answered state cannot be verified.
        return True
    return False


def synthesis_delivery_pending(thread_id: str) -> bool:
    """True when all council seats are terminal but synthesis is not yet in chat."""
    tid = (thread_id or "").strip()
    if not tid:
        return False
    try:
        bid = probe_council_batch_id(tid)
        if not bid:
            return False
        tasks = tasks_for_council_batch(tid, bid)
        if not tasks:
            return False
        if any((t.get("status") or "").strip() not in COUNCIL_TERMINAL_STATUSES for t in tasks):
            return False
        banner = f"{_SYNTHESIS_BANNER} ({bid})"
        return not _council_delivery_already_persisted(tid, banner)
    except Exception:
        _log.warning("synthesis_delivery_pending probe failed", exc_info=True)
        # Fail-closed: block idle_tick when synthesis state cannot be verified.
        return True


def _extract_failure_batch_id(content: str) -> str:
    import re

    m = re.search(
        rf"{re.escape(_FAILURE_BANNER)}\s*\(([^)]+)\)",
        content or "",
    )
    return (m.group(1) or "").strip() if m else ""


def _failure_summary_from_guard_content(content: str) -> tuple[str, str]:
    """Extract (summary, batch_id) from a pipeline failure guard bubble."""
    batch_id = _extract_failure_batch_id(content)
    for line in content.splitlines():
        ln = line.strip()
        if not ln or GUARD_PREFIX in ln or _FAILURE_BANNER in ln:
            continue
        if ln.startswith("(") and ln.endswith(")"):
            continue
        return ln[:240], batch_id
    return "Council could not complete.", batch_id


def probe_council_thread_failure_summary(thread_id: str) -> tuple[bool, str, str]:
    """Return (failed, brief_summary, batch_id) for chat thread_state / failure banner.

    Newest synthesis supersedes older failures. Spawn-refusal bubbles are skipped
    so a later cap refusal does not hide an earlier pipeline failure.
    """
    tid = (thread_id or "").strip()
    if not tid:
        return False, "", ""
    try:
        from chat_storage import get_messages

        checked = 0
        for msg in reversed(get_messages(tid) or []):
            if (msg.get("role") or "").strip() != "user":
                continue
            checked += 1
            content = str(msg.get("content") or "")
            if GUARD_PREFIX not in content:
                continue
            if _SYNTHESIS_BANNER in content:
                return False, "", ""
            if _FAILURE_BANNER in content:
                summary, batch_id = _failure_summary_from_guard_content(content)
                return True, summary, batch_id
            # Spawn refusal and other council guards: keep scanning older messages.
            if checked >= 24:
                break
    except Exception:
        _log.warning("probe_council_thread_failure_summary failed", exc_info=True)
        return False, "", ""
    return False, "", ""


def _council_delivery_already_persisted(thread_id: str, banner: str) -> bool:
    try:
        from chat_storage import get_messages

        prefix = (banner or "").strip()[:48]
        if not prefix:
            return False
        checked = 0
        for msg in reversed(get_messages(thread_id) or []):
            if (msg.get("role") or "").strip() != "user":
                continue
            checked += 1
            content = msg.get("content") or ""
            if prefix in content and GUARD_PREFIX in content:
                return True
            if checked >= 12:
                break
    except Exception:
        _log.debug("council delivery dedupe scan failed", exc_info=True)
    return False


def _deliver_council_synthesis(
    *,
    thread_id: str,
    batch_id: str,
    r2_incomplete: bool = False,
) -> bool:
    """Shared synthesis delivery path (R1-only or R1+R2 merged)."""
    banner = f"{_SYNTHESIS_BANNER} ({batch_id})"
    if _council_delivery_already_persisted(thread_id, banner):
        clear_active_council_batch(thread_id)
        notify_council_thread_state_refresh(thread_id)
        # Bubble may exist without a following assistant reply (append ok, wake failed).
        if not council_synthesis_already_answered(thread_id):
            wake_council_thread(thread_id, reason="council_synthesis", force=True)
        return False
    body = build_synthesis_context(
        thread_id=thread_id,
        batch_id=batch_id,
        r2_incomplete=r2_incomplete,
    )
    # Code-grounded suffix only when repo policy excerpts back the verdict; general
    # frames (no implementation policy section) must not demand repo citations.
    suffix = (
        COUNCIL_SYNTHESIS_SUFFIX
        if COUNCIL_POLICY_HEADER in body
        else COUNCIL_SYNTHESIS_SUFFIX_GENERAL
    )
    try:
        cfg = council_settings()
        if bool(cfg.get("show_transcript_in_chat", False)):
            transcript = build_council_transcript(thread_id=thread_id, batch_id=batch_id)
            appendix = format_council_transcript_text(transcript)
            if appendix:
                body = f"{body}\n\n## Council transcript\n{appendix}"
    except Exception:
        _log.debug("council transcript appendix failed", exc_info=True)
    delivered = deliver_council_continuation(
        thread_id,
        banner=banner,
        body=body,
        suffix=suffix,
    )
    if not delivered:
        _log.warning("council synthesis delivery failed for %s; retrying once", thread_id)
        delivered = deliver_council_continuation(
            thread_id,
            banner=banner,
            body=body,
            suffix=suffix,
            skip_dedupe=True,
        )
    clear_active_council_batch(thread_id)
    if delivered:
        wake_council_thread(thread_id, reason="council_synthesis", force=True)
    else:
        _log.error("council synthesis delivery failed after retry for %s", thread_id)
        wake_council_thread(
            thread_id,
            reason="council_synthesis_delivery_failed",
            force=True,
        )
    return delivered


def maybe_schedule_round2_after_contrarian(
    *,
    thread_id: str,
    batch_id: str,
    contrarian_task_id: str,
) -> bool:
    """Return True when round 2 Verifier was scheduled (synthesis deferred)."""
    from utils.council_orchestrator import (
        _contrarian_task_for_batch,
        _verifier_r1_task_for_batch,
        batch_has_round2_verifier,
        council_max_rounds_config,
        load_seat_result_metadata,
        should_schedule_round2_verifier,
        spawn_round2_verifier_task,
    )

    tid = (thread_id or "").strip()
    bid = (batch_id or "").strip()
    if not tid or not bid:
        return False
    if batch_has_round2_verifier(tid, bid):
        from utils.council_orchestrator import _round_from_task, _seat_from_task

        verifier_r2 = next(
            (
                t
                for t in tasks_for_council_batch(tid, bid)
                if _seat_from_task(t) == "verifier" and _round_from_task(t) >= 2
            ),
            None,
        )
        if verifier_r2 and not council_failure_already_delivered(tid):
            st = (verifier_r2.get("status") or "").strip()
            if st == "done":
                on_round2_verifier_done(thread_id=tid, batch_id=bid, r2_incomplete=False)
            elif st in TERMINAL_FAILURE_STATUSES:
                on_round2_verifier_done(thread_id=tid, batch_id=bid, r2_incomplete=True)
        return True
    verifier_r1 = _verifier_r1_task_for_batch(tid, bid)
    contrarian = _contrarian_task_for_batch(tid, bid)
    if not verifier_r1 or not contrarian:
        return False
    v1_md = load_seat_result_metadata(verifier_r1["id"])
    c_md = load_seat_result_metadata(contrarian["id"])
    if not should_schedule_round2_verifier(
        verifier_meta=v1_md,
        contrarian_meta=c_md,
        max_rounds=council_max_rounds_config(),
    ):
        return False
    try:
        spawn_round2_verifier_task(
            contrarian_task_id=contrarian_task_id,
            origin_thread_id=tid,
            batch_id=bid,
        )
        return True
    except Exception:
        _log.warning("council round 2 spawn failed for %s", tid, exc_info=True)
        if not council_failure_already_delivered(tid):
            on_round2_verifier_done(
                thread_id=tid,
                batch_id=bid,
                r2_incomplete=True,
            )
        return True


def on_round2_verifier_done(
    *,
    thread_id: str,
    batch_id: str,
    r2_incomplete: bool = False,
) -> bool:
    """Synthesis after disagreement-gated Verifier round 2."""
    return _deliver_council_synthesis(
        thread_id=thread_id,
        batch_id=batch_id,
        r2_incomplete=r2_incomplete,
    )


def on_contrarian_done(*, thread_id: str, batch_id: str) -> bool:
    return _deliver_council_synthesis(thread_id=thread_id, batch_id=batch_id)


def on_pipeline_failed(
    *,
    thread_id: str,
    batch_id: str,
    seat: str,
    detail: str = "",
) -> bool:
    banner = f"{_FAILURE_BANNER} ({batch_id})"
    if _council_delivery_already_persisted(thread_id, banner):
        clear_active_council_batch(thread_id)
        notify_council_thread_state_refresh(thread_id)
        return False
    friendly = humanize_council_failure_detail(detail, seat=seat)
    if seat in ("operator", "council"):
        label = "Council"
    else:
        from utils.council_orchestrator import council_seat_display_label

        label = council_seat_display_label(seat) if seat else "Council"
    body = (
        f"{label} could not finish. {friendly} "
        f"No synthesized consensus was produced. "
        f"(Batch {batch_id} — open Kanban to inspect seat tasks.)"
    )
    delivered = deliver_council_continuation(
        thread_id,
        banner=banner,
        body=body,
        suffix=COUNCIL_FAILURE_SUFFIX,
    )
    clear_active_council_batch(thread_id)
    if delivered:
        wake_council_thread(thread_id, reason="council_failed", force=True)
    else:
        _log.error("council failure delivery append failed for %s", thread_id)
        wake_council_thread(thread_id, reason="council_failed", force=True)
    return delivered


def _task_meta(task: dict) -> dict:
    md = task.get("metadata")
    return md if isinstance(md, dict) else {}


def _council_seat_next_action(task_id: str) -> str:
    """Best-effort next_action from the latest enriched run row."""
    import kanban_storage as ks

    tid = (task_id or "").strip()
    if not tid:
        return ""
    try:
        from utils.kanban_dispatcher import enrich_kanban_run_row

        runs = ks.list_runs(tid, limit=1)
        if not runs:
            return ""
        row = enrich_kanban_run_row(runs[0])
        return str(row.get("next_action") or "").strip()
    except Exception:
        _log.debug("_council_seat_next_action failed", exc_info=True)
        return ""


def _seat_from_task(task: dict) -> str:
    return str(_task_meta(task).get(COUNCIL_SEAT_META) or "").strip().lower()


def _council_terminal_failure_should_ignore(
    *,
    task: dict,
    thread_id: str,
    batch_id: str,
) -> bool:
    """Skip pipeline failure when an older seat was superseded by researcher retry."""
    from utils.council_orchestrator import (
        _max_pipeline_attempt_for_batch,
        _pipeline_attempt_from_task,
        batch_has_pending_researcher_retry,
    )

    tid = (thread_id or "").strip()
    bid = (batch_id or "").strip()
    if not tid or not bid:
        return False
    if not batch_has_pending_researcher_retry(tid, bid):
        return False
    attempt = _pipeline_attempt_from_task(task)
    max_attempt = _max_pipeline_attempt_for_batch(tid, bid)
    if attempt < max_attempt:
        return True
    status = (task.get("status") or "").strip()
    return status == "cancelled"


def _cancel_open_council_batch_tasks(*, thread_id: str, batch_id: str) -> None:
    import kanban_storage as ks

    for t in tasks_for_council_batch(thread_id, batch_id):
        st = (t.get("status") or "").strip()
        if st in COUNCIL_TERMINAL_STATUSES:
            continue
        try:
            ks.cancel_task(t["id"], cascade=False)
        except Exception:
            _log.debug("cancel council task %s failed", t.get("id"), exc_info=True)


def maybe_handle_council_task_terminal(task_id: str) -> None:
    """Route council batch terminal events after dispatcher persistence."""
    import kanban_storage as ks

    tid = (task_id or "").strip()
    if not tid:
        return
    task = ks.get_task(tid)
    if not task:
        return
    meta = _task_meta(task)
    if meta.get("origin") != COUNCIL_ORIGIN:
        return

    origin = (task.get("origin_thread_id") or "").strip()
    batch_id = str(meta.get(COUNCIL_BATCH_ID_META) or "").strip()
    if not origin or not batch_id:
        _log.warning("council task %s missing origin_thread_id or batch_id", tid)
        return

    try:
        _route_council_task_terminal_locked(
            task_id=tid,
            task=task,
            origin=origin,
            batch_id=batch_id,
        )
    finally:
        notify_council_thread_state_refresh(origin)


def _route_council_task_terminal_locked(
    *,
    task_id: str,
    task: dict,
    origin: str,
    batch_id: str,
) -> None:
    import kanban_storage as ks

    tid = task_id
    status = (task.get("status") or "").strip()
    seat = _seat_from_task(task)

    if status == "done" and seat == "verifier":
        from utils.council_orchestrator import (
            _round_from_task,
            load_seat_result_metadata,
            maybe_schedule_researcher_retry_after_verifier,
            should_retry_researcher_after_verifier,
            _researcher_task_for_batch,
        )

        if _round_from_task(task) >= 2:
            if not council_failure_already_delivered(origin):
                on_round2_verifier_done(
                    thread_id=origin,
                    batch_id=batch_id,
                    r2_incomplete=False,
                )
            return
        vmeta = load_seat_result_metadata(tid)
        researcher = _researcher_task_for_batch(origin, batch_id)
        rmeta = load_seat_result_metadata(researcher["id"]) if researcher else {}
        retry_wanted, _why = should_retry_researcher_after_verifier(
            vmeta,
            researcher_meta=rmeta,
        )
        if retry_wanted:
            if maybe_schedule_researcher_retry_after_verifier(
                thread_id=origin,
                batch_id=batch_id,
                verifier_task_id=tid,
            ):
                return
            _log.warning(
                "council researcher retry wanted but not scheduled for batch %s thread %s",
                batch_id,
                origin,
            )
        try:
            from utils.kanban_activity import emit_council_seat_activity

            emit_council_seat_activity(
                event_type="council_seat_done",
                task_id=tid,
                seat=seat,
                batch_id=batch_id,
                origin_thread_id=origin,
                status=status,
                assignee=str(task.get("assignee") or ""),
                summary=f"{seat} seat finished",
            )
        except Exception:
            _log.debug("council seat done activity failed", exc_info=True)
        return

    if status == "done" and seat == "researcher":
        # Honesty: a salvaged researcher handoff (harness synthesized it from gathered tool
        # text — the model never wrote a valid trailing fence) must not be framed as a clean
        # seat completion. Surface the salvage in the activity summary so operators can see
        # the council leaned on backup rather than native compliance.
        salvaged = False
        try:
            runs = ks.list_runs(tid, limit=1)
            if runs:
                rmd = runs[0].get("metadata")
                if isinstance(rmd, dict):
                    salvaged = bool(rmd.get("completion_salvage")) or str(
                        rmd.get("handoff_source") or ""
                    ).strip().startswith(("council_researcher_salvage", "ledger", "inband"))
        except Exception:
            _log.debug("council researcher salvage probe failed", exc_info=True)
        try:
            from utils.kanban_activity import emit_council_seat_activity

            emit_council_seat_activity(
                event_type="council_seat_done",
                task_id=tid,
                seat=seat,
                batch_id=batch_id,
                origin_thread_id=origin,
                status=status,
                assignee=str(task.get("assignee") or ""),
                summary=(
                    f"{seat} seat finished (harness-salvaged handoff)"
                    if salvaged
                    else f"{seat} seat finished"
                ),
            )
        except Exception:
            _log.debug("council seat done activity failed", exc_info=True)
        return

    if status in TERMINAL_FAILURE_STATUSES:
        from utils.council_orchestrator import _round_from_task

        if _council_terminal_failure_should_ignore(
            task=task,
            thread_id=origin,
            batch_id=batch_id,
        ):
            _log.info(
                "ignoring superseded council seat terminal %s (%s/%s) — retry in flight",
                tid,
                seat,
                status,
            )
            return

        if seat == "verifier" and _round_from_task(task) >= 2:
            try:
                from utils.kanban_activity import emit_council_seat_activity

                emit_council_seat_activity(
                    event_type="council_failed",
                    task_id=tid,
                    seat="verifier_r2",
                    batch_id=batch_id,
                    origin_thread_id=origin,
                    status=status,
                    assignee=str(task.get("assignee") or ""),
                    summary=f"Council round 2 verifier stopped ({status})",
                    next_action=_council_seat_next_action(tid),
                )
            except Exception:
                _log.debug("council r2 failed activity failed", exc_info=True)
            if not council_failure_already_delivered(origin):
                on_round2_verifier_done(
                    thread_id=origin,
                    batch_id=batch_id,
                    r2_incomplete=True,
                )
            return
        try:
            from utils.kanban_activity import emit_council_seat_activity

            emit_council_seat_activity(
                event_type="council_failed",
                task_id=tid,
                seat=seat or "pipeline",
                batch_id=batch_id,
                origin_thread_id=origin,
                status=status,
                assignee=str(task.get("assignee") or ""),
                summary=f"Council stopped at {seat or 'pipeline'} ({status})",
                next_action=_council_seat_next_action(tid),
            )
        except Exception:
            _log.debug("council failed activity failed", exc_info=True)
        _cancel_open_council_batch_tasks(thread_id=origin, batch_id=batch_id)
        detail = ""
        runs = ks.list_runs(tid, limit=1)
        if runs:
            rmd = runs[0].get("metadata")
            if isinstance(rmd, dict):
                detail = str(rmd.get("council_validation_error") or rmd.get("failure_reason") or "")
        if seat == "contrarian":
            detail = detail or "The contrarian review did not meet the bar."
        if not council_failure_already_delivered(origin):
            on_pipeline_failed(
                thread_id=origin,
                batch_id=batch_id,
                seat=seat,
                detail=detail,
            )
        return

    if status == "done" and seat == "contrarian":
        from utils.council_orchestrator import (
            _contrarian_task_for_batch,
            batch_has_pending_researcher_retry,
        )

        latest_contrarian = _contrarian_task_for_batch(origin, batch_id)
        latest_id = str((latest_contrarian or {}).get("id") or "").strip()
        if latest_id and latest_id != tid:
            _log.info(
                "ignoring superseded contrarian terminal %s (latest=%s)",
                tid,
                latest_id,
            )
            return
        if batch_has_pending_researcher_retry(origin, batch_id):
            _log.info(
                "deferring contrarian terminal %s — researcher retry in flight",
                tid,
            )
            return
        if maybe_schedule_round2_after_contrarian(
            thread_id=origin,
            batch_id=batch_id,
            contrarian_task_id=tid,
        ):
            return
        if not council_synthesis_ready(origin, batch_id):
            _log.info(
                "deferring synthesis for %s — batch %s not ready",
                origin,
                batch_id,
            )
            return
        on_contrarian_done(thread_id=origin, batch_id=batch_id)
        if (
            not council_synthesis_already_answered(origin)
            and synthesis_delivery_pending(origin)
        ):
            wake_council_thread(origin, reason="council_synthesis", force=True)
