"""Kanban coordinator mode: batch-scoped parent monitoring after multi-task DAG create."""

from __future__ import annotations

import logging
import re
import threading
import uuid
from typing import Any

_log = logging.getLogger(__name__)

COORDINATOR_BATCH_ID_META = "coordinator_batch_id"
COORDINATOR_BATCH_ACTIVATED_MARKER = "Coordinator batch:"
BATCH_TERMINAL_BANNER_PREFIX = "Kanban batch finished"
STANDALONE_TERMINAL_BANNER_PREFIX = "Kanban task finished"
TERMINAL_BOARD_STATUSES = frozenset({"done", "blocked", "cancelled"})
FAILURE_BOARD_STATUSES = frozenset({"blocked", "cancelled"})

_batch_finalize_locks: dict[str, threading.Lock] = {}
_batch_finalize_meta_lock = threading.Lock()

# Parent coordinator allowlist (everything else blocked in agent loop).
COORDINATOR_CORE_TOOLS: frozenset[str] = frozenset({
    "find_tools",
    "get_current_time",
    "get_task_plan",
    "update_task_plan",
    "clear_task_plan",
    "kanban_create_task",
    "kanban_create_batch",
    "kanban_list_tasks",
    "kanban_show_task",
    "kanban_list_runs",
    "kanban_cancel_task",
})

COORDINATOR_READ_PROBE_TOOLS: frozenset[str] = frozenset({
    "read_file",
    "list_directory",
})

COORDINATOR_DENY_KANBAN: frozenset[str] = frozenset({
    "kanban_post_summary",
    "kanban_post_progress",
})

# Parent coordinator may poll with these; repeated-only rounds trigger idle honest-stop.
COORDINATOR_MONITOR_TOOLS: frozenset[str] = frozenset({
    "kanban_list_tasks",
    "kanban_show_task",
    "kanban_list_runs",
    "get_task_plan",
    "update_task_plan",
})

_BOARD_ABORT_REQUEST_RE = re.compile(
    r"\b(?:abort|cancel|stop|kill|tear[\s-]?down)\b",
    re.IGNORECASE,
)


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

    cfg = get_kanban_config()
    return {
        "enabled": bool(cfg.get("coordinator_mode_enabled", True)),
        "min_batch_size": max(1, int(cfg.get("coordinator_min_batch_size", 2) or 2)),
        "allow_read_probe": bool(cfg.get("coordinator_allow_read_probe", False)),
        "notify_desktop": bool(cfg.get("coordinator_notify_desktop", True)),
        "rounds_bonus": max(0, int(cfg.get("coordinator_max_tool_rounds_bonus", 15) or 0)),
        "poll_idle_rounds": max(3, int(cfg.get("coordinator_poll_idle_rounds", 8) or 8)),
        "coordinator_auto_repair": bool(cfg.get("coordinator_auto_repair", False)),
        "max_auto_repair_per_task": max(0, int(cfg.get("max_auto_repair_per_task", 2) or 2)),
        "notify_on_blocked": bool(cfg.get("coordinator_notify_on_blocked", True)),
    }


def should_activate_coordinator(*, origin_thread_id: str, batch_size: int) -> bool:
    if not (origin_thread_id or "").strip():
        return False
    cfg = coordinator_settings()
    if not cfg["enabled"] or batch_size < cfg["min_batch_size"]:
        return False
    try:
        if board_is_active(origin_thread_id):
            _log.info(
                "Skipping coordinator activation for %s: board already active",
                origin_thread_id,
            )
            return False
    except Exception:
        _log.debug("coordinator activation board probe failed", exc_info=True)
        return False
    return True


def coordinator_batch_overlap_error(*, origin_thread_id: str, batch_size: int) -> str:
    """Return a user-visible error when a coordinator batch cannot start."""
    if not (origin_thread_id or "").strip():
        return ""
    cfg = coordinator_settings()
    if not cfg["enabled"] or batch_size < cfg["min_batch_size"]:
        return ""
    try:
        from utils.council_orchestrator import probe_council_active

        if probe_council_active(origin_thread_id):
            return (
                "Council mode is running for this conversation. "
                "Wait for council seats to finish or stop the council "
                "before creating a coordinator batch."
            )
    except Exception:
        _log.debug("coordinator council overlap probe failed", exc_info=True)
    try:
        if probe_board_active(origin_thread_id):
            return (
                "Kanban coordinator board is still active for this conversation. "
                "Wait for workers to finish, cancel open tasks, or clear the board "
                "before creating another multi-task batch on the same thread."
            )
    except Exception:
        _log.debug("coordinator overlap probe failed", exc_info=True)
        return "Could not verify coordinator board state; try again."
    return ""


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


def set_active_batch(thread_id: str, batch_id: str) -> None:
    if not thread_id or not batch_id:
        return
    from chat_storage import set_thread_coordinator_batch

    set_thread_coordinator_batch(thread_id, batch_id)


def clear_active_batch(thread_id: str) -> None:
    if not thread_id:
        return
    from chat_storage import clear_thread_coordinator_batch

    clear_thread_coordinator_batch(thread_id)


def get_active_batch_id(thread_id: str) -> str:
    if not thread_id:
        return ""
    from chat_storage import get_thread_coordinator_batch

    return get_thread_coordinator_batch(thread_id) or ""


def resolve_coordinator_batch_id(thread_id: str) -> str:
    """Batch id for UI/API without mutating chat storage.

    Falls back to open tasks' ``coordinator_batch_id`` metadata when the chat
    DB pointer was lost but workers are still running.
    """
    bid = get_active_batch_id(thread_id)
    if bid:
        return bid
    return _find_open_batch_id(thread_id)


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


def tasks_for_batch(
    thread_id: str,
    batch_id: str,
    *,
    limit: int = 500,
) -> list[dict[str, Any]]:
    if not thread_id or not batch_id:
        return []
    import kanban_storage as ks

    cap = max(1, min(int(limit or 500), 500))
    matched: dict[str, dict[str, Any]] = {}
    for order in ("desc", "asc"):
        rows = ks.list_tasks(
            origin_thread_id=thread_id, limit=cap, max_limit=cap, order=order
        )
        for task in rows:
            if _batch_id_from_task(task) != batch_id:
                continue
            tid = str(task.get("id") or "").strip()
            if tid:
                matched[tid] = task
            elif task not in matched.values():
                matched[f"__anon_{len(matched)}"] = task
    return list(matched.values())


def _origin_thread_has_kanban_tasks(thread_id: str) -> bool:
    if not thread_id:
        return False
    import kanban_storage as ks

    return bool(
        ks.list_tasks(origin_thread_id=thread_id, limit=1, max_limit=1, order="desc")
    )


def _find_open_batch_id(thread_id: str) -> str:
    """Return batch id for open coordinator tasks without mutating chat storage."""
    batch_id = get_active_batch_id(thread_id)
    if batch_id:
        return batch_id
    if not thread_id:
        return ""
    import kanban_storage as ks

    for order in ("desc", "asc"):
        rows = ks.list_tasks(
            origin_thread_id=thread_id, limit=100, max_limit=100, order=order
        )
        for task in rows:
            candidate = _batch_id_from_task(task)
            if not candidate:
                continue
            if (task.get("status") or "") in TERMINAL_BOARD_STATUSES:
                continue
            return candidate
    return ""


def _recover_batch_pointer(thread_id: str) -> str:
    """Restore coordinator pointer when tasks carry batch metadata but chat DB lost it."""
    batch_id = _find_open_batch_id(thread_id)
    if not batch_id or get_active_batch_id(thread_id):
        return batch_id
    try:
        set_active_batch(thread_id, batch_id)
    except Exception:
        _log.warning(
            "Failed to recover coordinator pointer for %s batch %s",
            thread_id,
            batch_id,
            exc_info=True,
        )
    return batch_id


def probe_board_active(thread_id: str) -> bool:
    """Read-only coordinator probe for delivery/UI — never clears the batch pointer."""
    batch_id = _find_open_batch_id(thread_id)
    if not batch_id:
        return False
    tasks = tasks_for_batch(thread_id, batch_id)
    if not tasks:
        _log.warning(
            "Coordinator batch %s on %s has no visible tasks; treating board as inactive",
            batch_id,
            thread_id,
        )
        return False
    return any((t.get("status") or "") not in TERMINAL_BOARD_STATUSES for t in tasks)


def tasks_for_active_batch(thread_id: str) -> list[dict[str, Any]]:
    batch_id = _find_open_batch_id(thread_id)
    if not batch_id:
        return []
    return tasks_for_batch(thread_id, batch_id)


def count_coordinator_blocked_tasks(thread_id: str) -> int:
    """Read-only count of blocked tasks on the active coordinator batch."""
    tid = (thread_id or "").strip()
    if not tid or not probe_board_active(tid):
        return 0
    try:
        return sum(
            1
            for t in tasks_for_active_batch(tid)
            if (t.get("status") or "").strip() == "blocked"
        )
    except Exception:
        _log.debug("count_coordinator_blocked_tasks failed", exc_info=True)
        return 0


def board_is_active(thread_id: str) -> bool:
    """Reconcile coordinator pointer with board state (may clear stale pointers)."""
    batch_id = get_active_batch_id(thread_id)
    if not batch_id:
        batch_id = _recover_batch_pointer(thread_id)
    if not batch_id:
        return False
    tasks = tasks_for_batch(thread_id, batch_id)
    if not tasks:
        _log.warning(
            "Coordinator batch %s on %s has no visible tasks; clearing stale pointer",
            batch_id,
            thread_id,
        )
        clear_active_batch(thread_id)
        return False
    open_tasks = [t for t in tasks if (t.get("status") or "") not in TERMINAL_BOARD_STATUSES]
    if not open_tasks:
        maybe_finalize_coordinator_batch(thread_id, batch_id, tasks)
        return False
    return True


def _batch_finalize_lock(thread_id: str, batch_id: str) -> threading.Lock:
    key = f"{thread_id}:{batch_id}"
    with _batch_finalize_meta_lock:
        lock = _batch_finalize_locks.get(key)
        if lock is None:
            lock = threading.Lock()
            _batch_finalize_locks[key] = lock
        return lock


def _batch_terminal_already_delivered(thread_id: str, batch_id: str) -> bool:
    """True when a batch-terminal guard message was already persisted."""
    banner = f"{BATCH_TERMINAL_BANNER_PREFIX} ({batch_id})"
    try:
        from chat_storage import get_messages

        user_checked = 0
        for msg in reversed(get_messages(thread_id) or []):
            if (msg.get("role") or "").strip() != "user":
                continue
            user_checked += 1
            content = msg.get("content") or ""
            if banner in content:
                return True
            if user_checked >= 12:
                break
    except Exception:
        _log.debug("batch terminal dedupe scan failed", exc_info=True)
    return False


def maybe_finalize_coordinator_batch(
    thread_id: str,
    batch_id: str,
    tasks: list[dict[str, Any]],
    *,
    triggering_task_id: str = "",
) -> bool:
    """Idempotent batch-terminal delivery + pointer clear.

    Returns True when a new batch-terminal message was delivered this call.
    """
    tid = (thread_id or "").strip()
    bid = (batch_id or "").strip()
    if not tid or not bid:
        return False
    open_tasks = [t for t in tasks if (t.get("status") or "") not in TERMINAL_BOARD_STATUSES]
    if open_tasks:
        return False

    lock = _batch_finalize_lock(tid, bid)
    with lock:
        open_tasks = [t for t in tasks if (t.get("status") or "") not in TERMINAL_BOARD_STATUSES]
        if open_tasks:
            return False
        if _batch_terminal_already_delivered(tid, bid):
            clear_active_batch(tid)
            return False
        try:
            from utils.kanban_coordinator_delivery import deliver_coordinator_batch_terminal

            delivered = deliver_coordinator_batch_terminal(
                tid,
                batch_id=bid,
                tasks=tasks,
                triggering_task_id=triggering_task_id,
            )
        except Exception:
            _log.warning("coordinator batch terminal delivery failed", exc_info=True)
            return False
        if not delivered:
            _log.warning("batch terminal delivery returned False for %s batch %s", tid, bid)
            return False
        clear_active_batch(tid)
        return True


def refresh_coordinator_state_for_task(task_id: str) -> None:
    """When the active batch has no open tasks, deliver batch-terminal summary.

    Standalone tasks (``kanban_create_task`` with ``origin_thread_id``, no
    ``coordinator_batch_id``) deliver a single-task summary + parent wake here.
    """
    if not task_id:
        return
    import kanban_storage as ks

    task = ks.get_task(task_id)
    if not task:
        return
    origin = (task.get("origin_thread_id") or "").strip()
    if not origin:
        return
    batch_id = _batch_id_from_task(task)
    if not batch_id:
        maybe_deliver_standalone_task_terminal(task_id)
        return
    active = get_active_batch_id(origin)
    if active and active != batch_id:
        return
    tasks = tasks_for_batch(origin, batch_id)
    maybe_finalize_coordinator_batch(
        origin,
        batch_id,
        tasks,
        triggering_task_id=task_id,
    )


def infer_coordinator_deliverable_type(hay: str) -> str | None:
    """Infer deliverable subtype from title/body text (lowercased haystack)."""
    if any(token in hay for token in ("ffmpeg", "encode", "transcode", "convert_video")):
        return "video_encode"
    has_video_signal = any(token in hay for token in ("mp4", "webm", "hyperframes", "video"))
    if has_video_signal and "render" in hay:
        return "video_render"
    if "hyperframes" in hay:
        return "video_render"
    if has_video_signal:
        return "video"
    if any(token in hay for token in ("pdf", "report", "document", "export", "deliverable")):
        return "document"
    return None


_WEB_ACT_INFER_SIGNALS: tuple[str, ...] = (
    "playwright",
    "run_web_script",
    "write_web_act_file",
    "web automation",
    "web_act",
    "web script",
)


def _hay_implies_web_act(hay: str) -> bool:
    lower = (hay or "").lower()
    return any(sig in lower for sig in _WEB_ACT_INFER_SIGNALS)


_WEB_ACT_WORKER_GUIDANCE: dict[str, str] = {
    "mutate": (
        "Write the Playwright script with write_web_act_file only (not write_text_file). "
        "Use paths relative to the web-act scope (e.g. explore.py from the explore.py template). "
        "Always pass relative_path and content; use safe_goto() for navigation."
    ),
    "verify": (
        "Run run_web_script with fresh_run=True after the script succeeds in-place. "
        "End with kanban-result including answer, web_act_run_dir, and artifact_paths."
    ),
    "deliver": (
        "Complete the interactive web task: write_web_act_file → run_web_script → "
        "run_web_script(fresh_run=True). Put run_dir and answer in kanban-result metadata."
    ),
}


def _stamp_web_act_worker_guidance(md: dict[str, Any]) -> None:
    if str(md.get("deliverable_type") or "").strip().lower() != "web_act":
        return
    if str(md.get("worker_guidance") or md.get("worker_hints") or "").strip():
        return
    kind = str(md.get("task_kind") or "").strip().lower() or "deliver"
    guidance = _WEB_ACT_WORKER_GUIDANCE.get(kind) or _WEB_ACT_WORKER_GUIDANCE["deliver"]
    md["worker_guidance"] = guidance


_WEB_ACT_TOOL_NAMES: tuple[str, ...] = ("write_web_act_file", "run_web_script")


def _stamp_web_act_allowed_hosts(md: dict[str, Any], body: str) -> None:
    if str(md.get("deliverable_type") or "").strip().lower() != "web_act":
        return
    if md.get("allowed_hosts"):
        return
    import re
    from urllib.parse import urlparse

    hosts: list[str] = []
    for match in re.findall(r"https?://[^\s<>\"']+", body or ""):
        host = (urlparse(match).hostname or "").strip().lower()
        if host and host not in hosts:
            hosts.append(host)
    if hosts:
        md["allowed_hosts"] = hosts


def _merge_web_act_tool_names(md: dict[str, Any]) -> None:
    if str(md.get("deliverable_type") or "").strip().lower() != "web_act":
        return
    try:
        from tools.web_act import web_act_enabled

        if not web_act_enabled():
            return
    except Exception:
        return
    for key in ("tool_names", "required_tools"):
        existing = md.get(key)
        if isinstance(existing, list):
            md[key] = list(dict.fromkeys([*existing, *_WEB_ACT_TOOL_NAMES]))
        elif key == "required_tools":
            md["required_tools"] = list(_WEB_ACT_TOOL_NAMES)


_VIDEO_RENDER_WORKER_GUIDANCE: dict[str, str] = {
    "mutate": (
        "Patch the project index.html in place (read_file first). Keep window.__hf, "
        "local ./vendor/gsap.min.js (never CDN script src), data-start=\"0\" on #root, "
        "and gsap.timeline({ paused: true }). One focused HTML edit, then hand off."
    ),
    "verify": (
        "Run the lint tool for this deliverable (e.g. hyperframes_lint with project slug). "
        "If lint reports errors, end blocked with blockers + next_action — never claim done."
    ),
    "deliver": (
        "Run hyperframes_render (or the project's render tool) after lint passes. "
        "Put the output path in kanban-result metadata.output or artifact_paths."
    ),
}


def _stamp_video_render_worker_guidance(md: dict[str, Any]) -> None:
    if str(md.get("deliverable_type") or "").strip().lower() != "video_render":
        return
    if str(md.get("worker_guidance") or md.get("worker_hints") or "").strip():
        return
    kind = str(md.get("task_kind") or "").strip().lower()
    guidance = _VIDEO_RENDER_WORKER_GUIDANCE.get(kind)
    if guidance:
        md["worker_guidance"] = guidance


def infer_coordinator_task_kind_metadata(
    metadata: dict[str, Any] | None,
    *,
    title: str = "",
    body: str = "",
) -> dict[str, Any]:
    """Stamp ``task_kind`` and ``deliverable_type`` on coordinator batch items when omitted."""
    md = dict(metadata or {})
    hay = f"{title} {body}".lower()

    if not str(md.get("task_kind") or "").strip():
        if "research" in hay:
            md["task_kind"] = "research"
        elif "lint" in hay or ("verify" in hay and "deliver" not in hay):
            md["task_kind"] = "verify"
        elif any(
            token in hay
            for token in ("render", "mp4", "webm", "pdf", "report", "document", "export", "deliver")
        ):
            md["task_kind"] = "deliver"
        elif any(
            token in hay
            for token in ("implement", "develop", "composition", "html", "mutate", "patch")
        ):
            md["task_kind"] = "mutate"

    task_kind = str(md.get("task_kind") or "").strip().lower()
    # Research tasks study a topic (e.g. "Research Hyperframes limits"); topic keywords
    # must not stamp video_render acceptance on workers that only search/read.
    if not str(md.get("deliverable_type") or "").strip() and task_kind != "research":
        if task_kind in ("deliver", "mutate", "") and _hay_implies_web_act(hay):
            md["deliverable_type"] = "web_act"
        else:
            inferred = infer_coordinator_deliverable_type(hay)
            if inferred:
                md["deliverable_type"] = inferred

    _stamp_video_render_worker_guidance(md)
    _stamp_web_act_worker_guidance(md)
    _stamp_web_act_allowed_hosts(md, body)
    _merge_web_act_tool_names(md)
    return md


def _effective_task_board_status(task: dict[str, Any]) -> str:
    """Detect false ``done`` rows whose terminal run is actually salvage-blocked."""
    status = str(task.get("status") or "").strip().lower()
    tid = str(task.get("id") or "").strip()
    if not tid:
        return status
    import kanban_storage as ks

    try:
        for run in ks.list_runs(tid, limit=4):
            outcome = str(run.get("outcome") or "").strip().lower()
            if outcome == ks.OUTCOME_ACTIVE:
                continue
            md = run.get("metadata") if isinstance(run.get("metadata"), dict) else {}
            salvage_status = str(md.get("status") or "").strip().lower()
            if salvage_status == "blocked" or outcome == ks.OUTCOME_BLOCKED:
                return "blocked"
            if salvage_status == "partial":
                return "blocked"
            break
    except Exception:
        _log.debug("effective board status lookup failed for %s", tid, exc_info=True)
    return status


def classify_batch_terminal(tasks: list[dict[str, Any]]) -> str:
    """Classify a terminal batch for coordinator delivery copy.

    Returns one of: ``all_done``, ``partial_failure``, ``all_blocked``.
    """
    if not tasks:
        return "all_done"
    statuses = [_effective_task_board_status(t) for t in tasks]
    if all(s == "done" for s in statuses):
        return "all_done"
    if statuses and all(s in FAILURE_BOARD_STATUSES for s in statuses):
        return "all_blocked"
    if any(s in FAILURE_BOARD_STATUSES for s in statuses):
        return "partial_failure"
    return "all_done"


def _last_run_next_action(task_id: str) -> str:
    import kanban_storage as ks
    from utils.kanban_dispatcher import enrich_kanban_run_row

    try:
        for run in ks.list_runs(task_id, limit=5):
            row = enrich_kanban_run_row(run)
            action = str(row.get("next_action") or "").strip()
            if action:
                return action
            md = run.get("metadata") if isinstance(run.get("metadata"), dict) else {}
            action = str(md.get("next_action") or "").strip()
            if action:
                return action
    except Exception:
        _log.debug("last run next_action lookup failed for %s", task_id, exc_info=True)
    return ""


def _standalone_terminal_already_delivered(thread_id: str, task_id: str) -> bool:
    banner = f"{STANDALONE_TERMINAL_BANNER_PREFIX} (`{task_id}`)"
    try:
        from chat_storage import get_messages

        user_checked = 0
        for msg in reversed(get_messages(thread_id) or []):
            if (msg.get("role") or "").strip() != "user":
                continue
            user_checked += 1
            content = msg.get("content") or ""
            if banner in content:
                return True
            if user_checked >= 12:
                break
    except Exception:
        _log.debug("standalone terminal dedupe scan failed", exc_info=True)
    return False


def format_standalone_task_terminal_body(task: dict[str, Any]) -> str:
    """Plain-text summary for a single root task dropped from chat (no coordinator batch)."""
    import kanban_storage as ks

    tid = str(task.get("id") or "?")
    status = _effective_task_board_status(task)
    title = (task.get("title") or "")[:120]
    lines = [
        f"Task `{tid}` reached terminal status `{status}`.",
        f"Title: {title}",
    ]
    summary_snippet = ""
    try:
        for run in ks.list_runs(tid, limit=3):
            snippet = (run.get("summary") or "").strip()
            if snippet:
                summary_snippet = snippet[:500]
                break
    except Exception:
        _log.debug("standalone terminal run summary lookup failed for %s", tid, exc_info=True)
    if summary_snippet:
        lines.extend(["", f"Worker summary: {summary_snippet}"])
    meta = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
    answer = str(meta.get("answer") or "").strip()
    if answer:
        lines.append(f"Answer: {answer[:300]}")
    run_dir = str(meta.get("web_act_run_dir") or "").strip()
    if run_dir:
        lines.append(f"Artifacts: {run_dir}")
    next_action = _last_run_next_action(tid)
    if next_action:
        lines.append(f"Next action: {next_action[:300]}")
    return "\n".join(lines)


def maybe_deliver_standalone_task_terminal(task_id: str) -> bool:
    """Deliver + wake origin chat when a standalone root task finishes.

    ``kanban_create_task`` from chat does not get coordinator batch metadata or
    auto thread subscriptions; this path reports outcomes back to the parent thread.
    """
    tid = (task_id or "").strip()
    if not tid:
        return False
    import kanban_storage as ks

    task = ks.get_task(tid)
    if not task:
        return False
    if _batch_id_from_task(task):
        return False
    if not _task_is_root(task):
        return False
    if _effective_task_board_status(task) not in TERMINAL_BOARD_STATUSES:
        return False
    origin = (task.get("origin_thread_id") or "").strip()
    if not origin:
        return False
    if probe_board_active(origin):
        return False
    try:
        rows = ks.list_tasks(origin_thread_id=origin, limit=100, max_limit=100, order="desc")
    except Exception:
        rows = []
    for row in rows:
        if (row.get("id") or "") == tid:
            continue
        if _batch_id_from_task(row):
            continue
        if (row.get("status") or "") not in TERMINAL_BOARD_STATUSES:
            return False
    if _standalone_terminal_already_delivered(origin, tid):
        return False
    status = _effective_task_board_status(task)
    banner = f"{STANDALONE_TERMINAL_BANNER_PREFIX} (`{tid}`)"
    body = format_standalone_task_terminal_body(task)
    try:
        from utils.kanban_coordinator_delivery import (
            deliver_coordinator_continuation,
            wake_coordinator_thread,
        )

        delivered = deliver_coordinator_continuation(
            origin,
            banner=banner,
            body=body,
        )
        if delivered:
            wake_coordinator_thread(origin, reason=status or "done", force=True)
        return delivered
    except Exception:
        _log.warning("standalone task terminal delivery failed for %s", tid, exc_info=True)
        return False


def format_batch_terminal_body(
    tasks: list[dict[str, Any]],
    *,
    triggering_task_id: str = "",
) -> str:
    """Plain-text summary for batch-terminal coordinator delivery."""
    import kanban_storage as ks

    classification = classify_batch_terminal(tasks)
    done_count = sum(1 for t in tasks if _effective_task_board_status(t) == "done")
    blocked_count = sum(
        1 for t in tasks if _effective_task_board_status(t) in FAILURE_BOARD_STATUSES
    )
    if classification == "all_done":
        headline = "All tasks in this coordinator batch completed successfully."
    elif classification == "all_blocked":
        headline = (
            f"Kanban batch finished with all {len(tasks)} task(s) blocked or cancelled."
        )
    else:
        headline = (
            f"Kanban batch finished with partial failure: {done_count} done, "
            f"{blocked_count} blocked/cancelled of {len(tasks)} total."
        )

    lines = [headline]
    if triggering_task_id:
        lines.append(f"Last update from task `{triggering_task_id}`.")
    lines.append("")
    for t in sorted(tasks, key=lambda row: (row.get("title") or "", row.get("id") or "")):
        tid = t.get("id") or "?"
        status = _effective_task_board_status(t)
        title = (t.get("title") or "")[:100]
        summary_snippet = ""
        try:
            for run in ks.list_runs(str(tid), limit=3):
                snippet = (run.get("summary") or "").strip()
                if snippet:
                    summary_snippet = snippet[:300]
                    break
        except Exception:
            _log.debug("batch terminal run summary lookup failed for %s", tid, exc_info=True)
        line = f"- `{tid}` [{status}] {title}"
        next_action = _last_run_next_action(str(tid))
        if next_action:
            line += f"\n  Next action: {next_action[:300]}"
        if summary_snippet:
            line += f"\n  Summary: {summary_snippet}"
        lines.append(line)
    return "\n".join(lines)


def format_board_status_snapshot(tasks: list[dict[str, Any]]) -> str:
    if not tasks:
        return "(no tasks)"
    lines: list[str] = []
    for t in tasks:
        tid = t.get("id") or "?"
        status = t.get("status") or "?"
        title = (t.get("title") or "")[:80]
        assignee = t.get("assignee") or "?"
        lines.append(f"- `{tid}` [{status}] ({assignee}) {title}")
    return "\n".join(lines)


def validate_coordinator_batch_structure(items: list[dict[str, Any]]) -> None:
    """Reject planner batches where every task is a root (no DAG edges).

    Without parent_index/parent_indices, all tasks become ``ready`` immediately
    so stages like "Final Review" can finish before upstream research completes.
    """
    if len(items) < 2:
        return
    for raw in items:
        md = raw.get("metadata") if isinstance(raw.get("metadata"), dict) else {}
        if md.get("allow_parallel_roots"):
            return
    roots = 0
    for idx, raw in enumerate(items):
        if not isinstance(raw, dict):
            continue
        parent_indices = raw.get("parent_indices")
        has_parent = (
            raw.get("parent_index") is not None
            or (isinstance(parent_indices, list) and bool(parent_indices))
            or (isinstance(raw.get("parent_id"), str) and str(raw.get("parent_id")).strip())
            or (isinstance(raw.get("parent_ids"), list) and bool(raw.get("parent_ids")))
        )
        if not has_parent:
            roots += 1
    if roots > 1:
        raise ValueError(
            f"coordinator batch has {roots} root tasks; wire dependencies with "
            "parent_index or parent_indices so only one task starts ready (unless "
            "metadata.allow_parallel_roots is true for intentional fan-out)."
        )
    _validate_composite_task_bodies(items)


_COMPOSITE_BODY_CHAR_LIMIT = 2400
_COMPOSITE_BODY_BULLET_MIN = 3


def _validate_composite_task_bodies(items: list[dict[str, Any]]) -> None:
    """Reject single tasks that bundle many deliverables without child tasks."""
    for raw in items:
        if not isinstance(raw, dict):
            continue
        body = str(raw.get("body") or "")
        if len(body) < _COMPOSITE_BODY_CHAR_LIMIT:
            continue
        bullet_lines = sum(
            1 for ln in body.splitlines()
            if ln.strip().startswith(("- ", "* ", "1.", "2.", "3."))
        )
        if bullet_lines >= _COMPOSITE_BODY_BULLET_MIN:
            title = str(raw.get("title") or "task")[:80]
            raise ValueError(
                f"coordinator batch task {title!r} body looks like multiple deliverables "
                f"({len(body)} chars, {bullet_lines} list items); split into dependent child tasks."
            )


def is_tool_allowed_in_coordinator(tool_name: str, *, thread_id: str = "") -> bool:
    name = (tool_name or "").strip()
    if not name:
        return False
    if name in COORDINATOR_DENY_KANBAN:
        return False
    if name in ("kanban_create_batch", "kanban_create_task"):
        tid = (thread_id or "").strip()
        if tid and probe_board_active(tid):
            return False
    if name in COORDINATOR_CORE_TOOLS:
        return True
    cfg = coordinator_settings()
    if cfg["allow_read_probe"] and name in COORDINATOR_READ_PROBE_TOOLS:
        return True
    return False


def coordinator_blocked_message(tool_name: str) -> str:
    return (
        f"Coordinator mode: tool `{tool_name}` is blocked while a Kanban board is executing "
        "this conversation's work. Use kanban_list_tasks / kanban_show_task to monitor progress. "
        "Do not duplicate worker implementation in this thread."
    )


def is_coordinator_monitor_only_round(tool_names: list[str]) -> bool:
    """True when every tool in the round is coordinator monitoring (poll/plan)."""
    names = [str(n or "").strip() for n in tool_names if str(n or "").strip()]
    if not names:
        return False
    return all(n in COORDINATOR_MONITOR_TOOLS for n in names)


def _user_message_text(content: Any) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for part in content:
            if isinstance(part, dict) and part.get("type") == "text":
                parts.append(str(part.get("text") or ""))
        return "\n".join(parts)
    return str(content or "")


def operator_requested_board_abort(messages: list[Any]) -> bool:
    """True when a recent user message explicitly asks to abort/cancel the board."""
    if not messages:
        return False
    for msg in reversed(messages[-12:]):
        if not isinstance(msg, dict) or msg.get("role") != "user":
            continue
        text = _user_message_text(msg.get("content")).strip()
        if not text or text.startswith("[Internal"):
            continue
        lower = text.lower()
        if not _BOARD_ABORT_REQUEST_RE.search(text):
            continue
        if any(
            kw in lower
            for kw in (
                "kanban",
                "board",
                "batch",
                "workers",
                "worker",
                "tasks",
                "everything",
                "all of",
            )
        ):
            return True
    return False


def foreground_kanban_cancel_allowed(
    messages: list[Any],
    *,
    user_text: str = "",
) -> bool:
    """True when foreground chat may call ``kanban_cancel_task``."""
    user = (user_text or "").strip()
    if not user and messages:
        for msg in reversed(messages[-12:]):
            if not isinstance(msg, dict) or msg.get("role") != "user":
                continue
            user = _user_message_text(msg.get("content")).strip()
            if user and not user.startswith("[Internal"):
                break
    if not user:
        return False
    try:
        from agent_loop_guardrails import _mandated_tool_names_from_user

        if "kanban_cancel_task" in _mandated_tool_names_from_user(user):
            return True
    except Exception:
        _log.debug("foreground cancel mandate check failed", exc_info=True)
    if operator_requested_board_abort(messages):
        return True
    if re.search(r"(?i)\bcancel\s+(?:task\s+)?kt_[a-f0-9]", user):
        return True
    return False


def coordinator_cancel_blocked(thread_id: str, messages: list[Any]) -> tuple[bool, str]:
    """Return (blocked, message) for kanban_cancel_task during active coordinator mode."""
    if not thread_id or not probe_board_active(thread_id):
        return False, ""
    if operator_requested_board_abort(messages):
        return False, ""
    return True, (
        "Coordinator mode: kanban_cancel_task is blocked while workers are executing this "
        "board. Wait for workers to finish, or ask the operator to explicitly abort/cancel "
        "the Kanban board if they want everything stopped."
    )


def coordinator_mode_suffix() -> str:
    return (
        "\n\n─── Kanban coordinator mode ───\n"
        "A multi-task Kanban board is executing this goal. You are the coordinator: "
        "monitor board state with kanban_list_tasks / kanban_show_task, update the in-thread "
        "task plan for tracking only, and summarize when workers finish. Do NOT call "
        "implementation tools (file writes, shell, delegate_task, domain-specific tool families) — "
        "Kanban workers own execution.\n"
        "Do NOT call kanban_cancel_task unless the operator explicitly asks to abort the board. "
        "After one brief status check, stop and wait — you will resume when Kanban delivery "
        "wakes this thread. Do not burn rounds re-polling the same in_progress task.\n"
        "When creating batches: wire dependencies (parent_index/parent_indices) so only one root "
        "starts ready; assign one verifiable mutation per worker (research vs write vs verify vs "
        "deliver); stamp acceptance.require_tools_any on mutate/verify tasks when you know the "
        "required tool family.\n"
    )


def _repair_tasks_for(task_id: str) -> list[dict[str, Any]]:
    """Existing auto-repair rows linked to *task_id*."""
    import kanban_storage as ks

    out: list[dict[str, Any]] = []
    try:
        for row in ks.list_tasks(limit=200):
            md = row.get("metadata") if isinstance(row.get("metadata"), dict) else {}
            if str(md.get("repair_of") or "").strip() == task_id:
                out.append(row)
    except Exception:
        _log.debug("repair task scan failed for %s", task_id, exc_info=True)
    return out


def maybe_auto_repair_blocked_task(task_id: str) -> dict[str, Any] | None:
    """When enabled, spawn a ready repair task from a blocked task's salvage metadata."""
    tid = (task_id or "").strip()
    if not tid:
        return None
    cfg = coordinator_settings()
    if not cfg.get("coordinator_auto_repair"):
        return None

    import kanban_storage as ks

    task = ks.get_task(tid)
    if not task or (task.get("status") or "") != "blocked":
        return None

    repairs = _repair_tasks_for(tid)
    max_repairs = int(cfg.get("max_auto_repair_per_task") or 2)
    if len(repairs) >= max_repairs:
        return None

    last_run = None
    for run in ks.list_runs(tid, limit=5):
        if (run.get("outcome") or "") != ks.OUTCOME_ACTIVE:
            last_run = run
            break
    if not last_run:
        return None

    md = last_run.get("metadata") if isinstance(last_run.get("metadata"), dict) else {}
    next_action = str(md.get("next_action") or "").strip()
    if not next_action:
        return None

    blockers = md.get("blockers")
    blocker_key = ""
    if isinstance(blockers, list):
        blocker_key = ",".join(sorted(str(x).strip() for x in blockers if str(x).strip()))
    elif isinstance(blockers, str):
        blocker_key = blockers.strip()

    if repairs and blocker_key:
        for prior in repairs:
            prior_md = prior.get("metadata") if isinstance(prior.get("metadata"), dict) else {}
            if str(prior_md.get("source_blocker_key") or "").strip() == blocker_key:
                _log.info(
                    "auto-repair skipped for %s: duplicate blocker %s",
                    tid,
                    blocker_key,
                )
                return None

    title = (task.get("title") or tid)[:80]
    parent_md = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
    repair_md: dict[str, Any] = {
        "repair_of": tid,
        "coordinator_auto_repair": True,
        "source_run_id": last_run.get("id"),
        "worker_guidance": next_action,
    }
    for key in ("task_kind", "deliverable_type"):
        val = str(parent_md.get(key) or "").strip()
        if val:
            repair_md[key] = val
    parent_acceptance = parent_md.get("acceptance")
    if isinstance(parent_acceptance, dict) and parent_acceptance:
        repair_md["acceptance"] = dict(parent_acceptance)
    if blocker_key:
        repair_md["source_blocker_key"] = blocker_key
    tools_next = md.get("tools_required_next")
    acceptance: dict[str, Any] = {}
    if isinstance(tools_next, list) and tools_next:
        acceptance["require_tools_any"] = [
            str(x).strip() for x in tools_next if str(x).strip()
        ]
    if acceptance:
        repair_md["acceptance"] = acceptance

    try:
        repair = ks.create_task(
            title=f"Repair: {title}",
            body=next_action,
            assignee=task.get("assignee") or "general",
            tenant=task.get("tenant") or "default",
            priority=max(1, int(task.get("priority") or 3) - 1),
            origin_thread_id=(task.get("origin_thread_id") or "").strip(),
            metadata=repair_md,
        )
    except Exception:
        _log.warning("auto-repair create_task failed for %s", tid, exc_info=True)
        return None
    _log.info(
        "auto-repair spawned task %s for blocked %s (blockers=%s)",
        repair.get("id"),
        tid,
        blocker_key or "-",
    )
    return repair


def _task_is_root(task: dict[str, Any]) -> bool:
    parent_ids = task.get("parent_ids") or []
    if parent_ids:
        return False
    return not (task.get("parent_id") or "").strip()


def auto_subscribe_coordinator_batch(
    tasks: list[dict[str, Any]],
    *,
    origin_thread_id: str,
) -> None:
    """Subscribe root tasks to thread (+ optional desktop) notifications."""
    if not origin_thread_id or not tasks:
        return
    cfg = coordinator_settings()
    from utils import kanban_notifications as kn

    roots = [t for t in tasks if _task_is_root(t)]
    if not roots:
        roots = [tasks[0]]

    for task in roots:
        task_id = task.get("id") or ""
        if not task_id:
            continue
        task_origin = (task.get("origin_thread_id") or "").strip()
        if task_origin != origin_thread_id:
            _log.warning(
                "Skipping coordinator subscribe: task origin %r != %r",
                task_origin,
                origin_thread_id,
            )
            continue
        try:
            thread_fire_on: tuple[str, ...] = ("done", "gave_up")
            if cfg.get("notify_on_blocked"):
                thread_fire_on = ("done", "gave_up", "blocked")
            kn.kanban_notify_subscribe_if_absent(
                task_id,
                channel="thread",
                target=origin_thread_id,
                fire_on=thread_fire_on,
            )
        except Exception:
            _log.debug("coordinator thread subscribe failed for %s", task_id, exc_info=True)
        if cfg["notify_desktop"]:
            try:
                kn.kanban_notify_subscribe_if_absent(
                    task_id,
                    channel="notification",
                    target=f"Kanban: {task.get('title') or task_id}",
                    fire_on=("done",),
                )
            except Exception:
                _log.debug("coordinator desktop subscribe failed for %s", task_id, exc_info=True)


_COORDINATOR_MONITOR_MARKERS: tuple[str, ...] = (
    "kanban",
    "board",
    "worker",
    "workers",
    "research task",
    "in progress",
    "in_progress",
    "monitor",
    "coordinating",
    "kt_",
)

_COORDINATOR_FALSE_DONE_MARKERS: tuple[str, ...] = (
    "all done",
    "everything is done",
    "fully complete",
    "task is complete",
    "video is ready",
    "rendered successfully",
    "finished the entire",
)


def is_coordinator_monitoring_prose(text: str) -> bool:
    """True when coordinator prose describes waiting/monitoring, not whole-goal completion."""
    t = (text or "").strip().lower()
    if len(t) < 24:
        return False
    if any(marker in t for marker in _COORDINATOR_FALSE_DONE_MARKERS):
        return False
    if not any(marker in t for marker in _COORDINATOR_MONITOR_MARKERS):
        return False
    if re.search(r"\b(done|finished|completed|delivered|ready for you)\b", t):
        if not re.search(r"\b(still|while|waiting|monitor|in progress|in_progress)\b", t):
            return False
    return True


def coordinator_plan_nudge() -> str:
    return (
        "\n\n─── Coordinator nudge ───\n"
        "Kanban coordinator mode is active. Poll the board with kanban_list_tasks "
        "(filter by this conversation if needed) and keep the operator informed. "
        "Do not duplicate worker implementation in this thread.\n"
    )


def coordinator_batch_created_user_message(batch_id: str) -> str:
    return (
        "[Kanban / recorded, not authoritative]\n"
        f"Coordinator mode activated for batch `{batch_id}`. "
        "Kanban workers own implementation; monitor the board and summarize outcomes "
        "when tasks complete. Do not duplicate Hyperframes or file-write work in this thread."
    )
