#!/usr/bin/env python3
"""Live coordinator/council delivery E2E battery.

Exercises the async delivery seam end-to-end over HTTP against a running
mugi_backend:

  1. Seed work on a fresh chat thread (coordinator batch or /council slash).
  2. Poll kanban until all tasks reach a terminal state (real MLX workers).
  3. Assert exactly one guard-marked delivery user bubble lands on the parent.
  4. Optionally wait for parent synthesis (assistant reply after delivery wake).
  5. Assert coordinator/council board pointers are inactive via thread_state.

Usage:
  ~/.mugi/venv/bin/python scripts/kernel_live_delivery_battery.py
  ~/.mugi/venv/bin/python scripts/kernel_live_delivery_battery.py --cases coordinator_batch_delivery
  make live-delivery-battery ARGS="--cases coordinator_batch_delivery"

Requires a listening mugi_backend (``make backend`` or Swift app).
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
import time
import uuid
from typing import Any

_REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _REPO not in sys.path:
    sys.path.insert(0, _REPO)

from utils.mcp_backend_client import (  # noqa: E402
    discover_backend,
    get_messages,
    get_thread_state,
    http_json,
    new_conversation,
    send_message,
    wait_for_run,
    _headers,
)
from scripts.kernel_live_delivery_cases import DELIVERY_CASES  # noqa: E402
from scripts.kernel_live_delivery_types import (  # noqa: E402
    GUARD_PREFIX,
    DeliveryCase,
    DeliveryCaseResult,
    _FAILURE_TERMINAL,
)
from scripts.kernel_live_gate import format_gate_summary_line  # noqa: E402

_TERMINAL = frozenset({"done", "blocked", "gave_up", "cancelled"})


def _kanban_get(conn, path: str, *, timeout: float = 30.0) -> tuple[int, Any]:
    return http_json("GET", f"{conn.base_url}{path}", _headers(conn), timeout=timeout)


def _kanban_post(conn, path: str, body: dict[str, Any] | None = None, *, timeout: float = 30.0) -> tuple[int, Any]:
    return http_json("POST", f"{conn.base_url}{path}", _headers(conn), body, timeout=timeout)


def kanban_create_batch(
    conn,
    *,
    items: list[dict[str, Any]],
    tenant: str,
    origin_thread_id: str,
) -> dict[str, Any]:
    code, out = _kanban_post(
        conn,
        "/kanban/tasks/batch",
        {"items": items, "tenant": tenant, "origin_thread_id": origin_thread_id},
        timeout=60.0,
    )
    if code != 200:
        return {"error": f"POST /kanban/tasks/batch failed ({code})", "body": out}
    tasks = out.get("tasks") if isinstance(out, dict) else (out if isinstance(out, list) else [])
    ids = [str(t.get("id")) for t in tasks if isinstance(t, dict) and t.get("id")]
    if not ids:
        return {"error": "batch returned no task ids", "body": out}
    return {"task_ids": ids, "tasks": tasks}


def kanban_list_by_origin(conn, origin_thread_id: str, *, tenant: str = "") -> list[dict[str, Any]]:
    qs = f"/kanban/tasks?origin_thread_id={origin_thread_id}&limit=50"
    if tenant:
        qs += f"&tenant={tenant}"
    code, out = _kanban_get(conn, qs)
    if code != 200:
        return []
    if isinstance(out, dict):
        return out.get("tasks") or []
    return out if isinstance(out, list) else []


def kanban_show_task(conn, task_id: str) -> dict[str, Any]:
    code, out = _kanban_get(conn, f"/kanban/tasks/{task_id}")
    if code != 200 or not isinstance(out, dict):
        return {"error": f"GET /kanban/tasks/{task_id} failed ({code})", "body": out}
    return out


def kanban_list_runs(conn, task_id: str, *, limit: int = 2) -> list[dict[str, Any]]:
    code, out = _kanban_get(conn, f"/kanban/tasks/{task_id}/runs?limit={limit}")
    if code != 200:
        return []
    if isinstance(out, list):
        return out
    return out.get("runs") or [] if isinstance(out, dict) else []


def kanban_nudge(conn) -> None:
    _kanban_post(conn, "/kanban/dispatcher/nudge", {})


def kanban_block_task(conn, task_id: str, *, reason: str) -> dict[str, Any]:
    code, out = _kanban_post(
        conn,
        f"/kanban/tasks/{task_id}/block",
        {"reason": reason},
        timeout=20.0,
    )
    if code != 200 or not isinstance(out, dict):
        return {"error": f"POST block failed ({code})", "body": out}
    return out


def _summarize_terminal(conn, task_id: str) -> dict[str, Any]:
    task = kanban_show_task(conn, task_id)
    if task.get("error"):
        return {"task_id": task_id, "error": task.get("error")}
    runs = kanban_list_runs(conn, task_id, limit=1)
    latest = runs[0] if runs else {}
    return {
        "task_id": task_id,
        "status": task.get("status"),
        "failure_count": task.get("failure_count"),
        "run_outcome": latest.get("outcome"),
        "run_summary": str(latest.get("summary") or ""),
    }


def wait_for_tasks(
    conn,
    task_ids: list[str],
    *,
    timeout_sec: float,
    poll_sec: float,
    origin_thread_id: str = "",
    case: DeliveryCase | None = None,
) -> tuple[list[dict[str, Any]], bool, list[str]]:
    """Poll until every task is terminal or timeout."""
    mid_failures: list[str] = []
    deadline = time.monotonic() + max(1.0, timeout_sec)
    last_nudge = 0.0
    while time.monotonic() < deadline:
        summaries = [_summarize_terminal(conn, tid) for tid in task_ids]
        if all(str(s.get("status") or "") in _TERMINAL for s in summaries):
            return summaries, False, mid_failures

        if (
            case
            and case.assert_no_delivery_before_all_terminal
            and origin_thread_id
        ):
            done = sum(1 for s in summaries if str(s.get("status") or "") == "done")
            if 0 < done < len(task_ids):
                early = _guard_delivery_messages(
                    _messages_list(conn, origin_thread_id),
                    banner_substrings=case.require_delivery_banner_substrings
                    or ("batch finished",),
                )
                if early:
                    mid_failures.append(
                        "premature guard delivery while child task still open "
                        f"(root done, {len(task_ids) - done} task(s) not terminal)"
                    )
                    return summaries, False, mid_failures

        now = time.monotonic()
        if now - last_nudge >= 30.0:
            kanban_nudge(conn)
            last_nudge = now
        time.sleep(max(1.0, poll_sec))
    summaries = [_summarize_terminal(conn, tid) for tid in task_ids]
    timed_out = not all(str(s.get("status") or "") in _TERMINAL for s in summaries)
    return summaries, timed_out, mid_failures


def _messages_list(conn, thread_id: str) -> list[dict[str, Any]]:
    payload = get_messages(conn, thread_id)
    if isinstance(payload, dict) and payload.get("error"):
        return []
    msgs = payload.get("messages") if isinstance(payload, dict) else None
    return msgs if isinstance(msgs, list) else []


def _guard_delivery_messages(
    messages: list[dict[str, Any]],
    *,
    banner_substrings: tuple[str, ...],
) -> list[str]:
    hits: list[str] = []
    for m in messages:
        if str(m.get("role") or "") != "user":
            continue
        content = str(m.get("content") or "")
        if GUARD_PREFIX not in content:
            continue
        blob = content.lower()
        if banner_substrings and not all(s.lower() in blob for s in banner_substrings):
            continue
        hits.append(content)
    return hits


def wait_for_guard_delivery(
    conn,
    thread_id: str,
    *,
    banner_substrings: tuple[str, ...],
    timeout_sec: float,
    poll_sec: float,
) -> tuple[list[str], bool]:
    deadline = time.monotonic() + max(5.0, timeout_sec)
    last_nudge = 0.0
    while time.monotonic() < deadline:
        hits = _guard_delivery_messages(
            _messages_list(conn, thread_id),
            banner_substrings=banner_substrings,
        )
        if hits:
            return hits, False
        now = time.monotonic()
        if now - last_nudge >= 25.0:
            kanban_nudge(conn)
            last_nudge = now
        time.sleep(max(1.0, poll_sec))
    hits = _guard_delivery_messages(
        _messages_list(conn, thread_id),
        banner_substrings=banner_substrings,
    )
    return hits, not bool(hits)


def _tenant_for_case(case_id: str) -> str:
    return f"delivery_battery_{case_id}_{uuid.uuid4().hex[:8]}"


def score_delivery_case(
    case: DeliveryCase,
    *,
    origin_thread_id: str,
    conn,
    summaries: list[dict[str, Any]],
    tasks_timed_out: bool,
    delivery_messages: list[str],
    delivery_timed_out: bool,
    synthesis_ok: bool,
    synthesis_error: str,
    mid_failures: list[str] | None = None,
) -> list[str]:
    failures: list[str] = list(mid_failures or [])
    if tasks_timed_out:
        failures.append("timed_out waiting for terminal task state")
    if delivery_timed_out:
        failures.append("timed_out waiting for guard-marked delivery bubble")

    done_count = sum(1 for s in summaries if str(s.get("status") or "") == "done")
    if done_count < case.expect_tasks_done:
        statuses = [str(s.get("status") or "?") for s in summaries]
        failures.append(
            f"expected {case.expect_tasks_done} done task(s), got {done_count}; "
            f"statuses={statuses}"
        )

    failed_count = sum(
        1 for s in summaries if str(s.get("status") or "") in _FAILURE_TERMINAL
    )
    if case.expect_tasks_failed and failed_count < case.expect_tasks_failed:
        statuses = [str(s.get("status") or "?") for s in summaries]
        failures.append(
            f"expected {case.expect_tasks_failed} failed task(s), got {failed_count}; "
            f"statuses={statuses}"
        )

    if len(delivery_messages) > case.max_guard_deliveries:
        failures.append(
            f"guard delivery count {len(delivery_messages)} > max "
            f"{case.max_guard_deliveries} (double-wake / duplicate delivery)"
        )
    if not delivery_messages and not delivery_timed_out:
        failures.append("no guard-marked delivery user message found")

    if delivery_messages:
        blob = delivery_messages[-1].lower()
        for sub in case.require_delivery_body_substrings:
            if sub.lower() not in blob:
                failures.append(
                    f"delivery body missing required substring {sub!r}"
                )
        for forbidden in case.forbid_delivery_banner_substrings:
            if forbidden.lower() in blob:
                failures.append(
                    f"forbidden delivery banner substring present: {forbidden!r}"
                )

    if case.expect_coordinator_inactive and origin_thread_id:
        state = get_thread_state(conn, origin_thread_id)
        if isinstance(state, dict) and "error" not in state:
            if state.get("coordinator_mode"):
                failures.append("coordinator_mode still active after batch terminal")
            open_tasks = state.get("coordinator_open_tasks")
            if isinstance(open_tasks, int) and open_tasks > 0:
                failures.append(
                    f"coordinator_open_tasks={open_tasks} after batch terminal"
                )

    if case.expect_council_inactive and origin_thread_id:
        state = get_thread_state(conn, origin_thread_id)
        if isinstance(state, dict) and "error" not in state:
            if state.get("council_active"):
                failures.append("council_active still true after seats finished")

    if case.wait_parent_synthesis and not synthesis_ok:
        failures.append(synthesis_error or "parent synthesis did not complete")

    return failures


def seed_coordinator_batch(
    conn,
    case: DeliveryCase,
    *,
    origin_thread_id: str,
    tenant: str,
) -> tuple[list[str], list[str]]:
    failures: list[str] = []
    if not case.batch_items:
        failures.append("coordinator_batch seed requires batch_items")
        return [], failures
    created = kanban_create_batch(
        conn,
        items=case.batch_items,
        tenant=tenant,
        origin_thread_id=origin_thread_id,
    )
    if created.get("error"):
        failures.append(str(created.get("error")))
        return [], failures
    ids = list(created["task_ids"])
    if len(ids) != case.expect_task_count:
        failures.append(
            f"expected {case.expect_task_count} tasks, got {len(ids)}"
        )
    return ids, failures


def seed_coordinator_block_child(
    conn,
    case: DeliveryCase,
    *,
    origin_thread_id: str,
    tenant: str,
) -> tuple[list[str], list[str]]:
    """Seed batch, wait for root done, HTTP-block child (deterministic partial fail)."""
    task_ids, failures = seed_coordinator_batch(
        conn, case, origin_thread_id=origin_thread_id, tenant=tenant
    )
    if failures or len(task_ids) < 2:
        return task_ids, failures
    root_id, child_id = task_ids[0], task_ids[1]
    deadline = time.monotonic() + min(300.0, case.timeout_sec * 0.5)
    while time.monotonic() < deadline:
        root = kanban_show_task(conn, root_id)
        if str(root.get("status") or "") == "done":
            break
        kanban_nudge(conn)
        time.sleep(case.poll_sec)
    else:
        failures.append(f"root task {root_id} did not reach done before child block")
        return task_ids, failures

    blocked = kanban_block_task(
        conn,
        child_id,
        reason="delivery_battery: block child for partial-failure delivery test",
    )
    if blocked.get("error"):
        failures.append(str(blocked.get("error")))
    kanban_nudge(conn)
    return task_ids, failures


def seed_council_slash(
    conn,
    case: DeliveryCase,
    *,
    origin_thread_id: str,
) -> tuple[list[str], list[str], bool, str]:
    """Returns (task_ids, failures, skipped, skip_reason)."""
    failures: list[str] = []
    frame = (case.council_frame or "").strip()
    if not frame:
        failures.append("council_slash seed requires council_frame")
        return [], failures, False, ""

    sent = send_message(conn, text=f"/council {frame}", thread_id=origin_thread_id)
    if sent.get("error"):
        failures.append(str(sent.get("error")))
        return [], failures, False, ""

    # Slash handler returns immediately; poll until tasks materialize.
    deadline = time.monotonic() + 45.0
    task_ids: list[str] = []
    while time.monotonic() < deadline:
        rows = kanban_list_by_origin(conn, origin_thread_id)
        task_ids = [str(t.get("id")) for t in rows if t.get("id")]
        if len(task_ids) >= case.expect_task_count:
            break
        time.sleep(2.0)

    if not task_ids:
        msgs = _messages_list(conn, origin_thread_id)
        refusal = [
            m
            for m in msgs
            if m.get("role") == "user"
            and GUARD_PREFIX in str(m.get("content") or "")
            and "council" in str(m.get("content") or "").lower()
        ]
        if refusal:
            return [], [], True, "council spawn refused (disabled or capped)"
        failures.append("council slash did not create tasks within 45s")
        return [], failures, False, ""

    if len(task_ids) != case.expect_task_count:
        failures.append(
            f"expected {case.expect_task_count} council tasks, got {len(task_ids)}"
        )
    return task_ids, failures, False, ""


def _task_seat(task: dict[str, Any]) -> str:
    md = task.get("metadata") if isinstance(task.get("metadata"), dict) else {}
    return str(md.get("council_seat") or "").strip().lower()


def seed_council_block_contrarian(
    conn,
    case: DeliveryCase,
    *,
    origin_thread_id: str,
) -> tuple[list[str], list[str], bool, str]:
    """Spawn council, wait for R+V, block Contrarian → pipeline failure delivery."""
    task_ids, failures, skipped, skip_reason = seed_council_slash(
        conn, case, origin_thread_id=origin_thread_id
    )
    if failures or skipped or not task_ids:
        return task_ids, failures, skipped, skip_reason

    deadline = time.monotonic() + min(600.0, case.timeout_sec * 0.6)
    contrarian_id = ""
    while time.monotonic() < deadline:
        rows = kanban_list_by_origin(conn, origin_thread_id)
        by_seat = {_task_seat(t): str(t.get("id") or "") for t in rows if t.get("id")}
        researcher = by_seat.get("researcher", "")
        verifier = by_seat.get("verifier", "")
        contrarian_id = by_seat.get("contrarian", "")
        if researcher and verifier and contrarian_id:
            r = kanban_show_task(conn, researcher)
            v = kanban_show_task(conn, verifier)
            if (
                str(r.get("status") or "") == "done"
                and str(v.get("status") or "") == "done"
            ):
                break
        kanban_nudge(conn)
        time.sleep(case.poll_sec)

    if not contrarian_id:
        failures.append("could not resolve contrarian seat task id")
        return task_ids, failures, False, ""

    blocked = kanban_block_task(
        conn,
        contrarian_id,
        reason="delivery_battery: intentional contrarian block for failure path",
    )
    if blocked.get("error"):
        failures.append(str(blocked.get("error")))
        return task_ids, failures, False, ""

    kanban_nudge(conn)
    time.sleep(2.0)
    return task_ids, failures, False, ""


def wait_for_parent_synthesis(
    conn,
    thread_id: str,
    *,
    messages_before: int,
    timeout_sec: float,
) -> tuple[bool, str]:
    """Wait until an assistant message appears after delivery (parent wake)."""
    deadline = time.monotonic() + max(10.0, timeout_sec)
    while time.monotonic() < deadline:
        msgs = _messages_list(conn, thread_id)
        if len(msgs) > messages_before:
            for m in reversed(msgs):
                if str(m.get("role") or "") == "assistant":
                    content = str(m.get("content") or "").strip()
                    if content:
                        return True, ""
        state = get_thread_state(conn, thread_id)
        if isinstance(state, dict) and state.get("running"):
            waited = wait_for_run(
                conn,
                thread_id,
                timeout_sec=min(60.0, deadline - time.monotonic()),
                poll_sec=2.0,
            )
            if not waited.get("timed_out") and not waited.get("error"):
                msgs = _messages_list(conn, thread_id)
                for m in reversed(msgs):
                    if str(m.get("role") or "") == "assistant":
                        if str(m.get("content") or "").strip():
                            return True, ""
        time.sleep(2.0)
    return False, "timed_out waiting for parent synthesis assistant reply"


def run_delivery_case(conn, case: DeliveryCase) -> DeliveryCaseResult:
    tenant = _tenant_for_case(case.id)
    t0 = time.monotonic()
    conv = new_conversation(conn)
    if conv.get("error"):
        return DeliveryCaseResult(
            case_id=case.id,
            passed=False,
            elapsed_sec=time.monotonic() - t0,
            pillar=case.pillar,
            failures=[str(conv.get("error"))],
            monitor_only=case.monitor_only,
        )
    origin_tid = str(conv["thread_id"])

    seed_failures: list[str] = []
    skipped = False
    skip_reason = ""
    task_ids: list[str] = []

    if case.seed == "coordinator_batch":
        task_ids, seed_failures = seed_coordinator_batch(
            conn, case, origin_thread_id=origin_tid, tenant=tenant
        )
    elif case.seed == "coordinator_block_child":
        task_ids, seed_failures = seed_coordinator_block_child(
            conn, case, origin_thread_id=origin_tid, tenant=tenant
        )
    elif case.seed == "council_slash":
        task_ids, seed_failures, skipped, skip_reason = seed_council_slash(
            conn, case, origin_thread_id=origin_tid
        )
    elif case.seed == "council_block_contrarian":
        task_ids, seed_failures, skipped, skip_reason = seed_council_block_contrarian(
            conn, case, origin_thread_id=origin_tid
        )
    else:
        seed_failures.append(f"unknown seed {case.seed!r}")

    if skipped:
        return DeliveryCaseResult(
            case_id=case.id,
            passed=True,
            elapsed_sec=time.monotonic() - t0,
            pillar=case.pillar,
            origin_thread_id=origin_tid,
            tenant=tenant,
            skipped=True,
            skip_reason=skip_reason,
            monitor_only=case.monitor_only,
        )

    if seed_failures or not task_ids:
        return DeliveryCaseResult(
            case_id=case.id,
            passed=False,
            elapsed_sec=time.monotonic() - t0,
            pillar=case.pillar,
            origin_thread_id=origin_tid,
            tenant=tenant,
            task_ids=task_ids,
            failures=seed_failures or ["no tasks seeded"],
            monitor_only=case.monitor_only,
        )

    msgs_before_tasks = len(_messages_list(conn, origin_tid))
    kanban_nudge(conn)
    summaries, tasks_timed_out, mid_failures = wait_for_tasks(
        conn,
        task_ids,
        timeout_sec=case.timeout_sec,
        poll_sec=case.poll_sec,
        origin_thread_id=origin_tid,
        case=case,
    )

    delivery_messages, delivery_timed_out = wait_for_guard_delivery(
        conn,
        origin_tid,
        banner_substrings=case.require_delivery_banner_substrings,
        timeout_sec=min(
            300.0 if case.expect_tasks_failed else 180.0,
            case.timeout_sec * 0.4,
        ),
        poll_sec=case.delivery_poll_sec,
    )

    synthesis_ok = True
    synthesis_error = ""
    if case.wait_parent_synthesis and delivery_messages and not delivery_timed_out:
        synthesis_ok, synthesis_error = wait_for_parent_synthesis(
            conn,
            origin_tid,
            messages_before=msgs_before_tasks,
            timeout_sec=case.synthesis_timeout_sec,
        )

    failures = seed_failures + score_delivery_case(
        case,
        origin_thread_id=origin_tid,
        conn=conn,
        summaries=summaries,
        tasks_timed_out=tasks_timed_out,
        delivery_messages=delivery_messages,
        delivery_timed_out=delivery_timed_out,
        synthesis_ok=synthesis_ok,
        synthesis_error=synthesis_error,
        mid_failures=mid_failures,
    )

    return DeliveryCaseResult(
        case_id=case.id,
        passed=not failures,
        elapsed_sec=time.monotonic() - t0,
        pillar=case.pillar,
        origin_thread_id=origin_tid,
        tenant=tenant,
        task_ids=task_ids,
        terminal=summaries,
        delivery_messages=[m[:400] for m in delivery_messages],
        failures=failures,
        timed_out=tasks_timed_out or delivery_timed_out,
        monitor_only=case.monitor_only,
    )


def cleanup_tenant(conn, tenant: str) -> int:
    code, body = _kanban_get(conn, f"/kanban/tasks?tenant={tenant}&limit=200")
    tasks = body.get("tasks") if isinstance(body, dict) else (body if isinstance(body, list) else [])
    cancelled = 0
    for t in tasks or []:
        tid = t.get("id")
        st = t.get("status")
        if tid and st in ("todo", "ready", "in_progress", "blocked", "queued"):
            _kanban_post(conn, f"/kanban/tasks/{tid}/cancel", {"cascade": True})
            cancelled += 1
    for terminal in ("done", "cancelled"):
        _kanban_post(conn, "/kanban/tasks/clear", {"status": terminal, "tenant": tenant})
    return cancelled


def summarize_delivery_results(
    results: list[DeliveryCaseResult],
) -> dict[str, Any]:
    """Split results into gating vs monitor-only and compute the exit code.

    Monitor-only cases (model-dependent happy paths) are reported but never affect
    the exit code — only gating-case failures fail the battery. Pure + unit-testable.
    """
    gating = [r for r in results if not r.monitor_only]
    monitor = [r for r in results if r.monitor_only]
    passed = sum(1 for r in gating if r.passed and not r.skipped)
    skipped = sum(1 for r in gating if r.skipped)
    failed = sum(1 for r in gating if not r.passed and not r.skipped)
    return {
        "gating": gating,
        "monitor": monitor,
        "passed": passed,
        "skipped": skipped,
        "failed": failed,
        "monitor_passed": sum(1 for r in monitor if r.passed and not r.skipped),
        "monitor_failed": sum(1 for r in monitor if not r.passed and not r.skipped),
        "exit_code": 0 if failed == 0 else 1,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Live coordinator/council delivery E2E battery")
    parser.add_argument("--cases", default="", help="Comma-separated case ids (default: all)")
    parser.add_argument("--pillar", default="", help="Comma-separated pillars (coordinator,council)")
    parser.add_argument("--delay", type=float, default=15.0, help="Seconds between cases (default 15)")
    parser.add_argument("--json", default="", help="Write JSON report to path")
    parser.add_argument("--cleanup", action="store_true", help="Clear isolated tenant tasks after each case")
    parser.add_argument(
        "--gate",
        action="store_true",
        help=(
            "Release-gate mode: monitor-only cases are non-gating (reported but exit 0)."
        ),
    )
    args = parser.parse_args()

    conn = discover_backend()
    if isinstance(conn, dict):
        print("BACKEND:", json.dumps(conn, indent=2), file=sys.stderr)
        return 2

    selected = list(DELIVERY_CASES)
    if args.cases.strip():
        want = {c.strip() for c in args.cases.split(",") if c.strip()}
        selected = [c for c in DELIVERY_CASES if c.id in want]
    if args.pillar.strip():
        pillars = {p.strip() for p in args.pillar.split(",") if p.strip()}
        selected = [c for c in selected if c.pillar in pillars]

    if not selected:
        print("No cases selected.", file=sys.stderr)
        return 2

    by_pillar = {}
    for c in selected:
        by_pillar[c.pillar] = by_pillar.get(c.pillar, 0) + 1

    print(
        f"backend pid={conn.pid} port={conn.port} cases={len(selected)} "
        f"pillars={by_pillar} delay={args.delay}s cleanup={'on' if args.cleanup else 'off'}\n"
    )

    results: list[DeliveryCaseResult] = []
    for i, case in enumerate(selected, 1):
        print(f"[{i}/{len(selected)}] [{case.pillar}] {case.id} …", flush=True)
        res = run_delivery_case(conn, case)
        results.append(res)
        if res.skipped:
            status = f"SKIP ({res.skip_reason})"
        elif res.passed:
            status = "PASS" + (" [monitor]" if res.monitor_only else "")
        elif res.monitor_only:
            # Non-gating: model-dependent happy path; reported, never fails the run.
            status = "MONITOR-FAIL (non-gating)"
        else:
            status = "FAIL"
        print(f"  {status} ({res.elapsed_sec:.1f}s) tasks={res.task_ids}")
        if res.delivery_messages:
            print(f"    delivery: {res.delivery_messages[0][:120]}…")
        for f in res.failures:
            print(f"    - {f}")
        print()
        if args.cleanup and res.tenant:
            n = cleanup_tenant(conn, res.tenant)
            if n:
                print(f"    [cleanup] cancelled {n} task(s) in {res.tenant}\n", flush=True)
        if i < len(selected) and args.delay > 0:
            time.sleep(args.delay)

    # Monitor-only cases are reported but never gate the run (model-dependent).
    summary = summarize_delivery_results(results)
    gating = summary["gating"]
    monitor = summary["monitor"]
    passed, skipped, failed = summary["passed"], summary["skipped"], summary["failed"]
    if args.gate:
        print(format_gate_summary_line("delivery", {
            "passed": passed,
            "failed": failed,
            "total": len(gating),
            "flaky": 0,
        }))
        if monitor:
            print(
                f"  monitor-only (non-gating): {summary['monitor_passed']} pass, "
                f"{summary['monitor_failed']} fail"
            )
    print(f"SUMMARY: {passed}/{len(gating)} passed, {failed} failed, {skipped} skipped")
    if monitor:
        print(
            f"  monitor-only (non-gating): {summary['monitor_passed']} pass, "
            f"{summary['monitor_failed']} fail "
            f"[{', '.join(r.case_id for r in monitor)}]"
        )
    for pillar in ("coordinator", "council"):
        sub = [r for r in gating if r.pillar == pillar]
        if sub:
            p = sum(1 for r in sub if r.passed and not r.skipped)
            print(f"  {pillar}: {p}/{len(sub)} passed")

    report = {
        "backend": {"pid": conn.pid, "port": conn.port, "repo_root": conn.repo_root},
        "summary": {
            "passed": passed,
            "failed": failed,
            "skipped": skipped,
            "total": len(gating),
            "monitor_passed": summary["monitor_passed"],
            "monitor_failed": summary["monitor_failed"],
            "monitor_total": len(monitor),
        },
        "results": [
            {
                "case_id": r.case_id,
                "pillar": r.pillar,
                "passed": r.passed,
                "monitor_only": r.monitor_only,
                "skipped": r.skipped,
                "skip_reason": r.skip_reason,
                "origin_thread_id": r.origin_thread_id,
                "tenant": r.tenant,
                "task_ids": r.task_ids,
                "elapsed_sec": round(r.elapsed_sec, 2),
                "delivery_messages": r.delivery_messages,
                "terminal": r.terminal,
                "failures": r.failures,
                "timed_out": r.timed_out,
            }
            for r in results
        ],
    }
    if args.json:
        with open(args.json, "w", encoding="utf-8") as f:
            json.dump(report, f, indent=2)
        print(f"Wrote {args.json}")

    exit_code = summary["exit_code"] if args.gate else (0 if failed == 0 else 1)
    return exit_code


if __name__ == "__main__":
    raise SystemExit(main())
