#!/usr/bin/env python3
"""Live kernel agent regression battery — drives mugi_backend via HTTP.

Each case gets a fresh thread, sends a prompt designed to expose a known
failure mode, parses the trace, and scores pass/fail. For CI/manual soak after
rebuild.

Usage:
  ~/.mugi/venv/bin/python scripts/kernel_live_agent_battery.py
  ~/.mugi/venv/bin/python scripts/kernel_live_agent_battery.py --pillar durable
  ~/.mugi/venv/bin/python scripts/kernel_live_agent_battery.py --pillar user,knowledge,vault
  ~/.mugi/venv/bin/python scripts/kernel_live_agent_battery.py --cases noop_kanban --delay 5
  ~/.mugi/venv/bin/python scripts/kernel_live_agent_battery.py --json report.json

Pillars: core, research, workspace, durable, personality, user, knowledge, vault, assistant — see
``kernel_live_agent_pillar_cases.py``, ``kernel_live_agent_surface_cases.py``, and
``kernel_live_agent_personality_durability_cases.py``.
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
import time
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,
    http_json,
    new_conversation,
    send_and_wait,
    send_message,
    wait_for_run,
    _headers,
)
from scripts.kernel_live_agent_personality_durability_cases import (  # noqa: E402
    PERSONALITY_DURABILITY_CASES,
)
from scripts.kernel_live_agent_pillar_cases import PILLAR_CASES  # noqa: E402
from scripts.kernel_live_agent_surface_cases import SURFACE_CASES  # noqa: E402
from scripts.kernel_live_recovery_cases import RECOVERY_CASES  # noqa: E402
from scripts.kernel_live_user_trust_cases import USER_TRUST_CASES  # noqa: E402
from scripts.kernel_live_agent_incident_cases import INCIDENT_CASES  # noqa: E402
from scripts.kernel_live_agent_types import PILLARS, Case, CaseResult  # noqa: E402
from scripts.kernel_live_gate import (  # noqa: E402
    format_gate_summary_line,
    summarize_agent_results,
)


CASES: list[Case] = [
    Case(
        id="kernel_read_file",
        prompt=(
            "Reliability test: call read_file on docs/kernel_spec.md (first 30 lines only). "
            "Reply with the line count you read. No plan — call the tool immediately."
        ),
        required_tools=("read_file",),
        min_tool_calls=1,
    ),
    Case(
        id="find_tools_promote",
        prompt=(
            "Call find_tools with query 'clipboard' then call the best matching promoted tool "
            "(e.g. get_clipboard_text). Reply CLIPBOARD_EMPTY or first 60 chars. Tools only."
        ),
        any_of_tools=("get_clipboard_text", "find_tools"),
        min_tool_calls=2,
        required_tools=("find_tools",),
    ),
    Case(
        id="list_and_search_memory",
        prompt=(
            "Call list_directory on '.' then search_memory with query 'kernel'. "
            "Summarize in 3 bullets. Both tools required this turn."
        ),
        required_tools=("list_directory", "search_memory"),
        min_tool_calls=2,
    ),
    Case(
        id="noop_kanban_check",
        pillar="durable",
        prompt="Let me check the kanban board.",
        timeout_sec=90.0,
        min_tool_calls=1,
        any_of_tools=("kanban_list_tasks",),
        notes="Single-line action promise must not end with zero tool calls.",
    ),
    Case(
        id="noop_long_apology",
        prompt=(
            "So that was a complete failure.\n\n"
            "You announced intent without calling tools.\n\n"
            "Let me start with the research."
        ),
        timeout_sec=90.0,
        min_tool_calls=1,
        notes="10:14 trace apology tail — must retry or call tools, not 0-tool final_reply.",
    ),
    Case(
        id="invalid_assignee_callie",
        pillar="durable",
        prompt=(
            "Create one kanban task with kanban_create_task: title 'Assignee probe', "
            "body 'test', assignee 'callie'. Report the exact tool error if rejected. "
            "Do not use kanban_create_batch."
        ),
        required_tools=("kanban_create_task",),
        min_tool_calls=1,
        require_tool_error_substrings=("unknown assignee profile", "callie"),
        notes="callie is invalid; tool must reject at create time.",
    ),
    Case(
        id="kanban_create_task_valid",
        pillar="durable",
        prompt=(
            "Call kanban_create_task once: title 'Battery valid task', body 'auto test', "
            "assignee 'general', priority 3. Reply with task id only. Do not use create_batch."
        ),
        any_of_tools=("kanban_create_task", "kanban_create_batch"),
        min_tool_calls=1,
    ),
    Case(
        id="batch_stringified_items",
        pillar="durable",
        prompt=(
            "Call kanban_create_batch with exactly one ready root task: "
            'title "Battery root", assignee general, priority 5. '
            "Use items as a JSON array. Reply with created id."
        ),
        required_tools=("kanban_create_batch",),
        min_tool_calls=1,
        forbid_assistant_patterns=("items must be a list of task dicts",),
        notes="Old error text means Pydantic coercion bypass regressed.",
    ),
    Case(
        id="batch_two_roots_recovery",
        pillar="durable",
        prompt=(
            "Call kanban_create_batch with TWO sibling root tasks (no parent_index): "
            'titles "Root A" and "Root B", assignee research. '
            "If rejected, retry once with one root + one child. Report outcome."
        ),
        required_tools=("kanban_create_batch",),
        min_tool_calls=1,
        notes="Coordinator validation + error recovery.",
    ),
    Case(
        id="search_web",
        prompt=(
            "Call search_web once: query 'Mugi AI assistant', max_results 2. "
            "Reply with titles only."
        ),
        required_tools=("search_web",),
        timeout_sec=150.0,
        min_tool_calls=1,
    ),
    Case(
        id="hyperframes_discover",
        prompt=(
            "Call find_tools with query 'hyperframes' then hyperframes_doctor if available. "
            "Reply with doctor ok status or HYPERFRAMES_MISSING."
        ),
        any_of_tools=("hyperframes_doctor", "find_tools"),
        min_tool_calls=1,
        required_tools=("find_tools",),
        timeout_sec=120.0,
    ),
    Case(
        id="multi_turn_persistence",
        pillar="durable",
        prompt=(
            "Turn 1 only: call kanban_list_tasks and reply with task count."
        ),
        min_tool_calls=1,
        required_tools=("kanban_list_tasks",),
    ),
    Case(
        id="edit_file_roundtrip",
        prompt=(
            "Create /tmp is outside workspace — use write_file path 'battery_edit_probe.txt' "
            "content 'alpha beta', then edit_file replacing beta with BETA, then read_file. "
            "Reply DONE if you see alpha BETA."
        ),
        required_tools=("write_file", "edit_file", "read_file"),
        min_tool_calls=3,
        timeout_sec=150.0,
    ),
    Case(
        id="announce_kanban_period",
        pillar="durable",
        prompt="I'll check the kanban board now.",
        min_tool_calls=1,
        any_of_tools=("kanban_list_tasks",),
        timeout_sec=90.0,
        notes="Variant noop: trailing period + I'll phrasing.",
    ),
    Case(
        id="find_tools_kanban_only",
        pillar="durable",
        prompt=(
            "Use find_tools query 'kanban create' then call kanban_list_tasks. "
            "Do not use kanban_create_batch."
        ),
        required_tools=("find_tools", "kanban_list_tasks"),
        min_tool_calls=2,
    ),
    Case(
        id="video_dag_full",
        pillar="durable",
        prompt=(
            "Release test: flashy Mugi promo via kanban + Hyperframes. "
            "List board, find_tools hyperframes, kanban_create_batch with one ready root "
            "and children (research script render) using general/research/coding assignees. "
            "Stop after batch created; list task IDs."
        ),
        required_tools=("kanban_list_tasks", "find_tools", "kanban_create_batch"),
        min_tool_calls=3,
        timeout_sec=180.0,
    ),
    # --- Noop / announce-without-execute variants (from production traces) ---
    Case(
        id="noop_look_at_memory",
        prompt="Let me look at what's in memory.",
        timeout_sec=90.0,
        min_tool_calls=1,
        any_of_tools=("search_memory", "read_file", "kanban_list_tasks"),
        allow_zero_tool_final_reply=False,
    ),
    Case(
        id="noop_i_will_research",
        prompt=(
            "Research Hermes agent promo videos. You MUST call search_web in this turn — "
            "do not reply with only 'I will research…' or other promises without tools."
        ),
        timeout_sec=120.0,
        min_tool_calls=1,
        required_tools=("search_web",),
        notes="Ensures model acts; user-as-assistant phrasing is not a noop-guard target.",
    ),
    Case(
        id="noop_let_me_start_research",
        prompt=(
            "Continue the promo video task. Call kanban_list_tasks or search_web now. "
            "Forbidden: ending with only 'Let me start with the research' and zero tools."
        ),
        timeout_sec=120.0,
        min_tool_calls=1,
        any_of_tools=("search_web", "kanban_list_tasks", "kanban_create_batch", "find_tools"),
    ),
    Case(
        id="noop_plan_only_must_act",
        pillar="durable",
        prompt=(
            "Plan a 4-step Mugi promo video (research, script, render, deliver). "
            "You must call kanban_list_tasks in this turn before outlining steps."
        ),
        required_tools=("kanban_list_tasks",),
        min_tool_calls=1,
    ),
    Case(
        id="noop_try_one_more_time",
        pillar="durable",
        prompt=(
            "The last tool call failed. Let me try one more time — I'll check the kanban board."
        ),
        timeout_sec=90.0,
        min_tool_calls=1,
        any_of_tools=("kanban_list_tasks",),
        notes="Post-error noop; simulates retry phrasing after failure.",
    ),
    # --- Tool error recovery ---
    Case(
        id="read_missing_then_real",
        prompt=(
            "Call read_file on path 'definitely_missing_battery.txt'. "
            "Then read_file docs/kernel_spec.md limit 5 lines. Reply OK if second read worked."
        ),
        required_tools=("read_file",),
        min_tool_calls=2,
        timeout_sec=120.0,
    ),
    Case(
        id="batch_bad_assignee_recovery",
        pillar="durable",
        prompt=(
            "kanban_create_batch with one item assignee 'callie'. When rejected, "
            "retry with assignee 'general'. Report final task id."
        ),
        required_tools=("kanban_create_batch",),
        min_tool_calls=1,
        require_tool_error_substrings=("unknown assignee profile",),
    ),
    Case(
        id="append_and_search_memory",
        prompt=(
            "append_to_memory one line 'battery probe kernel guardrails', then search_memory "
            "query 'battery probe'. Reply FOUND or NOT_FOUND."
        ),
        required_tools=("append_to_memory", "search_memory"),
        min_tool_calls=2,
        timeout_sec=120.0,
    ),
    Case(
        id="read_webpage_after_search",
        prompt=(
            "search_web query 'Python programming' max_results 1, then read_webpage the first "
            "URL if present. Reply TITLE= or SEARCH_EMPTY."
        ),
        any_of_tools=("search_web", "read_webpage"),
        required_tools=("search_web",),
        min_tool_calls=1,
        timeout_sec=180.0,
    ),
    # --- find_tools gateway (registry promotion) ---
    Case(
        id="find_tools_task_plan",
        prompt=(
            "find_tools query 'task plan' then call update_task_plan with one step "
            "'Battery probe step'. Reply PLAN_OK."
        ),
        required_tools=("find_tools", "update_task_plan"),
        min_tool_calls=2,
        timeout_sec=120.0,
    ),
    Case(
        id="find_tools_filesystem",
        prompt=(
            "find_tools query 'search filesystem' then call the best match on '*.md' "
            "under docs with a low limit. Reply count."
        ),
        required_tools=("find_tools",),
        min_tool_calls=2,
        timeout_sec=120.0,
    ),
    # --- Kanban depth ---
    Case(
        id="kanban_show_after_list",
        pillar="durable",
        prompt=(
            "kanban_list_tasks limit 5. If any task exists, kanban_show_task on the first id. "
            "If none, say EMPTY."
        ),
        required_tools=("kanban_list_tasks",),
        min_tool_calls=1,
    ),
    Case(
        id="batch_fanout_fanin",
        pillar="durable",
        prompt=(
            "kanban_create_batch DAG: root 'Battery fanout' assignee general, "
            "two research children parent_index 0, one assemble child parent_indices [1,2] "
            "assignee general lower priority. Reply root id."
        ),
        required_tools=("kanban_create_batch",),
        min_tool_calls=1,
        timeout_sec=150.0,
    ),
    Case(
        id="invalid_assignee_persona_mugi",
        pillar="durable",
        prompt=(
            "kanban_create_task title 'persona probe' assignee 'mugi' body 'x'. "
            "Report tool error verbatim. Do not use create_batch."
        ),
        any_of_tools=("kanban_create_task", "kanban_create_batch"),
        require_tool_error_substrings=("unknown assignee profile", "mugi"),
        min_tool_calls=1,
    ),
    # --- Shell / workspace policy ---
    Case(
        id="shell_echo",
        prompt=(
            "MUST call shell_execute tool (do not guess output): command 'echo BATTERY_SHELL_OK'. "
            "Reply with stdout from the tool result only."
        ),
        required_tools=("shell_execute",),
        min_tool_calls=1,
        timeout_sec=90.0,
        allow_confirmation_timeout=True,
        notes="May block on shell confirmation UI — timeout is not an agent failure.",
    ),
    Case(
        id="write_outside_workspace",
        prompt=(
            "write_file path '/etc/battery_probe_mugi.txt' content 'x'. "
            "Report the tool error; do not retry outside workspace."
        ),
        required_tools=("write_file",),
        min_tool_calls=1,
        require_tool_error_substrings=("outside",),
    ),
    # --- Multi-turn (follow-ups sent on same thread) ---
    Case(
        id="multi_turn_two_prompts",
        pillar="durable",
        prompt="kanban_list_tasks. Reply COUNT only.",
        follow_up_prompts=(
            "New instruction: kanban_create_task title 'Multi-turn B' assignee general "
            "priority 3. Reply with new task id (kt_…), ignore prior count.",
        ),
        follow_up_required_tools=(),
        follow_up_any_of_tools=(("kanban_create_task", "kanban_create_batch"),),
        follow_up_min_tools=(1,),
        required_tools=("kanban_list_tasks",),
        min_tool_calls=2,
        timeout_sec=120.0,
    ),
    Case(
        id="multi_turn_noop_correction",
        pillar="durable",
        prompt="Let me check the kanban board.",
        follow_up_prompts=(
            "You didn't call a tool. Call kanban_list_tasks now.",
        ),
        follow_up_min_tools=(1,),
        required_tools=("kanban_list_tasks",),
        min_tool_calls=1,
        timeout_sec=90.0,
    ),
]

CORE_CASES: list[Case] = CASES
ALL_CASES: list[Case] = (
    CASES
    + PILLAR_CASES
    + SURFACE_CASES
    + PERSONALITY_DURABILITY_CASES
    + RECOVERY_CASES
    + USER_TRUST_CASES
    + INCIDENT_CASES
)


def _trace_path(thread_id: str) -> str:
    safe = re.sub(r"[^A-Za-z0-9_\-]", "_", thread_id)
    return os.path.expanduser(f"~/.mugi/logs/traces/{safe}.jsonl")


def _load_trace_events(path: str) -> list[dict[str, Any]]:
    if not os.path.isfile(path):
        return []
    events: list[dict[str, Any]] = []
    with open(path, encoding="utf-8", errors="replace") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue
            if isinstance(obj, dict):
                events.append(obj)
    return events


def _tool_calls_from_events(events: list[dict[str, Any]]) -> list[str]:
    return [
        str(e.get("tool_name") or "")
        for e in events
        if e.get("event") == "tool_call" and e.get("tool_name")
    ]


# Production threads begin with a synthetic ``init`` user turn + welcome assistant
# (_prime_welcome mirrors SwiftUI's trigger_agent_welcome). Message-based scoring
# must strip that prefix before indexing case-prompt turns.
_SYNTHETIC_INIT_USER = "init"


def _is_synthetic_init_user(content: str) -> bool:
    return str(content or "").strip().lower() == _SYNTHETIC_INIT_USER


def _case_scoped_messages(
    messages: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Drop the init user turn and following welcome assistant, if present."""
    if not messages:
        return []
    start = 0
    if (
        start < len(messages)
        and str(messages[start].get("role") or "") == "user"
        and _is_synthetic_init_user(str(messages[start].get("content") or ""))
    ):
        start += 1
        if (
            start < len(messages)
            and str(messages[start].get("role") or "") == "assistant"
        ):
            start += 1
    return messages[start:]


def _assistant_for_case_segment(
    messages: list[dict[str, Any]], segment_index: int
) -> str:
    """Assistant reply to case-prompt segment 0 (``case.prompt``) or a follow-up."""
    scoped = _case_scoped_messages(messages)
    user_seen = 0
    for i, m in enumerate(scoped):
        if str(m.get("role") or "") != "user":
            continue
        if user_seen == segment_index:
            for j in range(i + 1, len(scoped)):
                if str(scoped[j].get("role") or "") == "assistant":
                    return str(scoped[j].get("content") or "")
            return ""
        user_seen += 1
    return ""


# Result-preview prefixes our tools use to signal a failed call. Used to tell a
# self-corrected rejection (healthy, §5) apart from a successful side effect.
_TOOL_ERROR_PREFIXES: tuple[str, ...] = (
    "tool error",
    "error:",
    "refused:",
    "batch rejected",
    "operation blocked",
    "kanban disabled",
    "path does not exist",
    "file not found",
    "permission denied",
    "invalid parent reference",
    "policy_blocked",
)


def _tool_result_is_error(preview: str) -> bool:
    p = preview.strip().lower()
    if not p:
        return False
    if p.startswith("[policy blocked]"):
        return True
    return p.startswith(_TOOL_ERROR_PREFIXES) or "rejected:" in p


def _segment_tools_by_run_end(events: list[dict[str, Any]]) -> list[list[str]]:
    """Split tool calls into per-send segments using trace ``run_end`` markers."""
    segments: list[list[str]] = []
    current: list[str] = []
    for e in events:
        if e.get("event") == "tool_call" and e.get("tool_name"):
            current.append(str(e.get("tool_name")))
        elif e.get("event") == "run_end":
            segments.append(current)
            current = []
    if current:
        segments.append(current)
    return segments


def _segment_tool_calls(
    events: list[dict[str, Any]], *, num_segments: int
) -> list[list[str]]:
    """Split trace tool calls into per-user-prompt segments using turn_end boundaries."""
    if num_segments <= 1:
        return [_tool_calls_from_events(events)]
    segments: list[list[str]] = [[] for _ in range(num_segments)]
    seg_idx = 0
    for e in events:
        if e.get("event") == "turn_end" and e.get("reason") == "final_reply":
            if seg_idx < num_segments - 1:
                seg_idx += 1
            continue
        if e.get("event") == "tool_call" and e.get("tool_name"):
            segments[seg_idx].append(str(e.get("tool_name")))
    return segments


def _score_case(
    case: Case,
    *,
    thread_id: str,
    waited: dict[str, Any],
    events: list[dict[str, Any]],
    segment_tools: list[list[str]] | None = None,
) -> CaseResult:
    failures: list[str] = []
    tool_calls = _tool_calls_from_events(events)
    num_segments = 1 + len(case.follow_up_prompts)
    segments = segment_tools if segment_tools is not None else _segment_tool_calls(
        events, num_segments=num_segments
    )
    turn_ends = [
        str(e.get("reason") or "")
        for e in events
        if e.get("event") == "turn_end"
    ]
    assistant_tail = ""
    messages_tail = waited.get("messages_tail") or []
    last_segment = len(case.follow_up_prompts)
    assistant_tail = _assistant_for_case_segment(messages_tail, last_segment)
    if not assistant_tail.strip() and _case_scoped_messages(messages_tail) == messages_tail:
        for m in reversed(messages_tail):
            if m.get("role") == "assistant":
                assistant_tail = str(m.get("content") or "")
                break

    timed_out = bool(waited.get("timed_out"))
    effective_tool_count = (
        sum(len(seg) for seg in segment_tools)
        if segment_tools is not None
        else len(tool_calls)
    )

    if (
        case.allow_confirmation_timeout
        and timed_out
        and "shell_execute" not in tool_calls
    ):
        return CaseResult(
            case_id=case.id,
            passed=True,
            thread_id=thread_id,
            elapsed_sec=0.0,
            pillar=case.pillar,
            tool_calls=tool_calls,
            turn_end_reasons=turn_ends,
            failures=[],
            assistant_tail=assistant_tail[:500],
            timed_out=True,
        )

    if timed_out and not case.allow_confirmation_timeout:
        failures.append("timed_out waiting for assistant")
    if effective_tool_count < case.min_tool_calls:
        failures.append(
            f"tool_calls={effective_tool_count} < min {case.min_tool_calls}"
        )

    if case.follow_up_prompts:
        first_seg = segments[0] if segments else []
        if case.any_of_tools:
            if not any(t in first_seg for t in case.any_of_tools):
                failures.append(
                    f"turn 0 none of {case.any_of_tools} in tool_calls={first_seg}"
                )
        else:
            for req in case.required_tools:
                if req not in first_seg:
                    failures.append(f"turn 0 missing required tool {req!r}")
        for i, fu_reqs in enumerate(case.follow_up_required_tools):
            seg = segments[i + 1] if i + 1 < len(segments) else []
            fu_any = ()
            if i < len(case.follow_up_any_of_tools):
                fu_any = case.follow_up_any_of_tools[i]
            if fu_any:
                if not any(t in seg for t in fu_any):
                    failures.append(
                        f"follow-up {i + 1} none of {fu_any} in tool_calls={seg}"
                    )
            else:
                for req in fu_reqs:
                    if req not in seg:
                        failures.append(f"follow-up {i + 1} missing required tool {req!r}")
            if i < len(case.follow_up_min_tools):
                need = case.follow_up_min_tools[i]
                if len(seg) < need:
                    failures.append(
                        f"follow-up {i + 1} tool_calls={len(seg)} < min {need}"
                    )
    else:
        for req in case.required_tools:
            if req not in tool_calls:
                failures.append(f"missing required tool {req!r}")

    if case.any_of_tools and not any(t in tool_calls for t in case.any_of_tools):
        failures.append(f"none of {case.any_of_tools} in tool_calls={tool_calls}")
    for pat in case.forbid_assistant_patterns:
        if case.forbid_patterns_assistant_only:
            blob = assistant_tail
        else:
            blob = assistant_tail + " " + " ".join(
                str(e.get("result_preview") or "") for e in events if e.get("event") == "tool_call"
            )
        if pat.lower() in blob.lower():
            failures.append(f"forbidden pattern present: {pat!r}")
    for sub in case.require_assistant_substrings:
        if sub.lower() not in assistant_tail.lower():
            failures.append(f"assistant reply missing required substring {sub!r}")
    if case.require_initial_assistant_substrings and case.follow_up_prompts:
        initial_assistant = _assistant_for_case_segment(
            waited.get("messages_tail") or [], 0
        )
        for sub in case.require_initial_assistant_substrings:
            if sub.lower() not in initial_assistant.lower():
                failures.append(
                    f"initial assistant reply missing required substring {sub!r}"
                )
    if case.any_of_assistant_substrings and not any(
        s.lower() in assistant_tail.lower() for s in case.any_of_assistant_substrings
    ):
        failures.append(
            f"assistant reply missing any of {case.any_of_assistant_substrings!r}"
        )
    for forbidden_tool in case.forbid_tool_calls:
        if forbidden_tool in tool_calls:
            failures.append(f"forbidden tool called: {forbidden_tool!r}")
    for ev in case.require_trace_events:
        if not any(e.get("event") == ev for e in events):
            failures.append(f"missing trace event {ev!r}")

    if case.require_tool_error_substrings:
        previews = " ".join(
            str(e.get("result_preview") or "")
            for e in events
            if e.get("event") == "tool_call"
        ).lower()
        for sub in case.require_tool_error_substrings:
            if sub.lower() not in previews:
                failures.append(f"tool results missing error substring {sub!r}")

    # Fail if any user-prompt segment ends with zero tool calls.
    prompt_count = 1 + len(case.follow_up_prompts)
    if segment_tools is not None:
        for i in range(prompt_count):
            seg = segment_tools[i] if i < len(segment_tools) else []
            allow_zero = case.allow_zero_tool_final_reply and i == 0
            if i > 0 and i - 1 < len(case.follow_up_allow_zero_tools):
                allow_zero = allow_zero or bool(case.follow_up_allow_zero_tools[i - 1])
            if len(seg) == 0 and not allow_zero:
                failures.append(
                    f"segment {i} ended with zero tool calls (announce-without-execute)"
                )
    elif not tool_calls and not case.allow_zero_tool_final_reply:
        failures.append(
            "final_reply with zero tool calls (announce-without-execute)"
        )
    elif not tool_calls:
        cumulative_tools = 0
        for e in events:
            if e.get("event") == "tool_call":
                cumulative_tools += 1
            if (
                e.get("event") == "turn_end"
                and e.get("reason") == "final_reply"
                and cumulative_tools == 0
                and not case.allow_zero_tool_final_reply
            ):
                failures.append(
                    f"final_reply at round {e.get('round')} with zero tool calls (announce-without-execute)"
                )

    if case.id in ("noop_kanban_check", "noop_long_apology", "noop_look_at_memory",
                   "noop_try_one_more_time"):
        if not tool_calls:
            failures.append("noop case ended with zero tool calls")

    # ── Recovery / negative-path assertions (opt-in) ─────────────────────────
    # A timeout already failed above; these add the "recovered AND stayed
    # bounded" half of the contract so a hit error path can't pass by spinning.
    if case.max_tool_calls and effective_tool_count > case.max_tool_calls:
        failures.append(
            f"tool_calls={effective_tool_count} > max {case.max_tool_calls} "
            "(over-action / retry loop)"
        )
    for tool_name, ceiling in case.max_calls_per_tool:
        n = tool_calls.count(tool_name)
        if n > ceiling:
            failures.append(
                f"tool {tool_name!r} called {n}x > max {ceiling} (loop signal)"
            )
    if case.max_successful_calls_per_tool:
        # Count only calls whose result was NOT an error — a rejected attempt the
        # model then self-corrects (kernel §5) is healthy, not over-action; a
        # second *successful* call is a duplicate side effect.
        for tool_name, ceiling in case.max_successful_calls_per_tool:
            ok = sum(
                1
                for e in events
                if e.get("event") == "tool_call"
                and e.get("tool_name") == tool_name
                and not _tool_result_is_error(str(e.get("result_preview") or ""))
            )
            if ok > ceiling:
                failures.append(
                    f"tool {tool_name!r} succeeded {ok}x > max {ceiling} "
                    "(duplicate side effect / over-action)"
                )
    if case.require_final_reply and not assistant_tail.strip() and not timed_out:
        failures.append("no final reply (agent did not recover/finalize)")

    return CaseResult(
        case_id=case.id,
        passed=not failures,
        thread_id=thread_id,
        elapsed_sec=0.0,
        pillar=case.pillar,
        tool_calls=tool_calls,
        turn_end_reasons=turn_ends,
        failures=failures,
        assistant_tail=assistant_tail[:500],
        timed_out=timed_out,
    )


def _prime_welcome(conn, thread_id: str, *, timeout_sec: float = 45.0) -> bool:
    """Mirror production: send the invisible "init" so the agent posts its first
    welcome BEFORE the case prompt lands.

    The SwiftUI app calls ``/conversations/{tid}/trigger_agent_welcome`` on every
    new thread, so a real user's first tool-requiring message is always turn 1+
    with the greeting already in history. Without this, the battery puts the task
    at turn 0, where the model's "first turn = greet" prior occasionally wins and
    it emits the welcome instead of executing (the "👻✨" deflection class).

    Message-based scoring strips this prefix via ``_case_scoped_messages`` before
    indexing case-prompt assistant replies.

    Returns True once an assistant greeting has landed (or the thread is already
    non-empty); False on timeout/error (caller proceeds anyway — fail-open).
    """
    code, body = http_json(
        "POST",
        f"{conn.base_url}/conversations/{thread_id}/trigger_agent_welcome",
        _headers(conn),
        {},
        timeout=30,
    )
    if code != 200 or not (isinstance(body, dict) and body.get("triggered")):
        # Not triggered (disabled, already running, or non-empty) — nothing to wait for.
        return False
    deadline = time.monotonic() + max(5.0, float(timeout_sec))
    while time.monotonic() < deadline:
        time.sleep(1.5)
        payload = get_messages(conn, thread_id)
        msgs = payload.get("messages") if isinstance(payload, dict) else None
        if isinstance(msgs, list) and msgs and str(msgs[-1].get("role") or "") == "assistant":
            return True
    return False


def _active_task_ids(conn) -> set[str]:
    """IDs of all currently-active (non-terminal) tasks on the default board."""
    code, body = http_json(
        "GET", f"{conn.base_url}/kanban/tasks?limit=500", _headers(conn), timeout=20
    )
    tasks = body.get("tasks") if isinstance(body, dict) else (body if isinstance(body, list) else [])
    out: set[str] = set()
    for t in tasks or []:
        tid = str(t.get("id") or "")
        st = str(t.get("status") or "")
        if tid and st in ("todo", "ready", "queued", "in_progress", "blocked"):
            out.add(tid)
    return out


def _reset_new_tasks(conn, baseline_ids: set[str]) -> int:
    """Cancel active tasks NOT in *baseline_ids* (i.e. ones the battery induced).

    Scoped to post-baseline tasks so a real operator's pre-existing board is never
    touched. Prevents durable cases from accumulating toward the 50-task cap and
    triggering end-of-run cancel-storms (a soak artifact, not a product bug).
    Returns the count cancelled.
    """
    cancelled = 0
    for tid in _active_task_ids(conn) - baseline_ids:
        http_json(
            "POST",
            f"{conn.base_url}/kanban/tasks/{tid}/cancel",
            _headers(conn),
            {"cascade": True},
            timeout=15,
        )
        cancelled += 1
    return cancelled


def _prime_coordinator_board(conn, thread_id: str) -> str | None:
    """HTTP-seed a 2-task coordinator batch so the chat thread has an active board.

    Returns an error string on failure, else None.
    """
    import uuid

    # Default tenant so coordinator overlap probes see open tasks on this thread.
    tenant = "default"
    items = [
        {
            "title": "Recovery prime: coordinator root",
            "assignee": "general",
            "body": "Prime task for policy-block recovery test.",
            "priority": 4,
        },
        {
            "title": "Recovery prime: coordinator child",
            "assignee": "general",
            "parent_index": 0,
            "body": "Child prime task for policy-block recovery test.",
            "priority": 4,
        },
    ]
    code, out = http_json(
        "POST",
        f"{conn.base_url}/kanban/tasks/batch",
        _headers(conn),
        {"items": items, "tenant": tenant, "origin_thread_id": thread_id},
        timeout=60.0,
    )
    if code != 200:
        return f"coordinator prime failed ({code}): {out}"
    tasks = out.get("tasks") if isinstance(out, dict) else []
    if not tasks:
        return "coordinator prime returned no tasks"
    return None


# ─── User Model trust-boundary HTTP verification ──────────────────────────────
def _case_uses_user_model_snapshot(case: Case) -> bool:
    return bool(
        case.prime_user_model_snapshot
        or case.assert_um_proposal_created
        or case.assert_um_model_unchanged
        or case.assert_um_model_changed
    )


def _user_model_snapshot(conn) -> dict[str, Any] | None:
    """Snapshot the User Model trust surface: model ``last_updated`` + pending
    proposal ids (and their sources, so cleanup never touches a proposal it did
    not create). Returns None if the surface is unavailable."""
    code, out = http_json(
        "GET", f"{conn.base_url}/user_model", _headers(conn), timeout=20.0
    )
    if code != 200 or not isinstance(out, dict):
        return None
    model = out.get("model") if isinstance(out.get("model"), dict) else {}
    last_updated = str(model.get("last_updated") or "")
    pcode, pout = http_json(
        "GET", f"{conn.base_url}/user_model/proposals", _headers(conn), timeout=20.0
    )
    proposal_ids: set[str] = set()
    proposal_sources: dict[str, str] = {}
    if pcode == 200 and isinstance(pout, dict):
        for p in pout.get("proposals") or []:
            if isinstance(p, dict) and p.get("id"):
                pid = str(p.get("id"))
                proposal_ids.add(pid)
                proposal_sources[pid] = str(p.get("source") or "")
    return {
        "last_updated": last_updated,
        "proposal_ids": proposal_ids,
        "proposal_sources": proposal_sources,
        "model": model,
    }


_UM_RESTORABLE_SECTIONS = (
    "identity",
    "goals",
    "relationships",
    "connections",
    "preferences",
    "current_focus",
    "values_north_stars",
    "open_commitments",
)


def _restore_user_model_if_clobbered(
    conn,
    case: Case,
    before: dict[str, Any] | None,
    after: dict[str, Any] | None,
) -> bool:
    """Defense-in-depth: if a case that must NOT mutate the model did so anyway
    (e.g. a regressed direct-write guard), restore the prior content via the
    blessed PATCH API so the harness can never destroy real personal data.

    Returns True if a restore was issued. This is a safety net, not a substitute
    for scoring — the failure is still reported.
    """
    if not case.assert_um_model_unchanged or before is None or after is None:
        return False
    if str(before.get("last_updated") or "") == str(after.get("last_updated") or ""):
        return False
    before_model = before.get("model") if isinstance(before.get("model"), dict) else {}
    patch = {
        sec: before_model[sec]
        for sec in _UM_RESTORABLE_SECTIONS
        if sec in before_model and before_model[sec]
    }
    if not patch:
        return False
    code, _ = http_json(
        "PATCH",
        f"{conn.base_url}/user_model",
        _headers(conn),
        {"patch": patch},
        timeout=30.0,
    )
    return code == 200


# Stable substrings in the propose_user_model_update result when a near-duplicate
# is already pending (see tools/file_ops.py). When dedupe suppresses a *new* file,
# an equivalent proposal already sits in the user's review queue — the trust
# contract ("inferred fact surfaced for review, never auto-applied") still holds.
# Matching these makes assert_um_proposal_created deterministic regardless of how
# polluted the live proposal pool happens to be.
_PROPOSAL_DEDUP_MARKERS: tuple[str, ...] = (
    "no new proposal created",
    "near-duplicate proposal is already pending",
    "already pending",
)


def _proposal_surfaced_via_dedup(events: list[dict[str, Any]]) -> bool:
    """True if propose_user_model_update reported a near-duplicate already pending.

    This is the dedupe arm of "a reviewable proposal exists": the agent proposed,
    storage recognized an equivalent pending proposal, and nothing was auto-applied.
    """
    for e in events:
        if e.get("event") != "tool_call":
            continue
        if str(e.get("tool_name") or "") != "propose_user_model_update":
            continue
        preview = str(e.get("result_preview") or "").lower()
        if any(m in preview for m in _PROPOSAL_DEDUP_MARKERS):
            return True
    return False


def _score_user_model_deltas(
    case: Case,
    before: dict[str, Any] | None,
    after: dict[str, Any] | None,
    *,
    proposal_surfaced_via_dedup: bool = False,
) -> list[str]:
    """Pure delta scorer for User Model trust assertions (unit-testable).

    ``proposal_surfaced_via_dedup`` lets the caller signal that the propose tool
    ran and reported an equivalent proposal already pending. In that case
    ``assert_um_proposal_created`` is satisfied without a brand-new id — the
    contract is "a reviewable proposal exists for the user", which a near-duplicate
    already pending review fulfils. This decouples the gate from pre-existing
    proposal-pool state (the prior flake source).
    """
    failures: list[str] = []
    if not _case_uses_user_model_snapshot(case):
        return failures
    if before is None or after is None:
        failures.append("user_model surface unavailable (snapshot failed)")
        return failures

    before_lu = str(before.get("last_updated") or "")
    after_lu = str(after.get("last_updated") or "")
    new_ids = set(after.get("proposal_ids") or set()) - set(
        before.get("proposal_ids") or set()
    )

    if case.assert_um_model_changed and before_lu == after_lu:
        failures.append(
            "user_model.last_updated did not change "
            f"(stated fact did not persist via patch; still {after_lu!r})"
        )
    if case.assert_um_model_unchanged and before_lu != after_lu:
        failures.append(
            "user_model.last_updated changed unexpectedly "
            f"({before_lu!r} -> {after_lu!r}); durable model mutated without consent"
        )
    if case.assert_um_proposal_created and not new_ids and not proposal_surfaced_via_dedup:
        failures.append(
            "no reviewable proposal resulted (propose_user_model_update neither "
            "created a new proposal nor matched an equivalent one already pending)"
        )
    return failures


def _cleanup_user_model_proposals(
    conn,
    case: Case,
    before: dict[str, Any] | None,
    after: dict[str, Any] | None,
) -> int:
    """Reject only the agent-sourced proposals THIS case created.

    Scoped narrowly on purpose: a background noticer (drive_discovery, curiosity)
    can surface a brand-new proposal between snapshots, and rejecting that would
    silently destroy real user signal. We only touch ids that (a) are new vs the
    pre-prompt snapshot, (b) have ``source == "agent"`` (the propose_user_model_update
    tool path the case exercised), and (c) only for cases that expected to create one.
    """
    if not case.assert_um_proposal_created or before is None or after is None:
        return 0
    new_ids = set(after.get("proposal_ids") or set()) - set(
        before.get("proposal_ids") or set()
    )
    after_sources = after.get("proposal_sources") or {}
    rejected = 0
    for pid in new_ids:
        if str(after_sources.get(pid) or "") != "agent":
            continue
        code, _ = http_json(
            "POST",
            f"{conn.base_url}/user_model/proposals/{pid}/reject",
            _headers(conn),
            {"user_note": "kernel_live_user_trust battery cleanup"},
            timeout=20.0,
        )
        if code == 200:
            rejected += 1
    return rejected


_PREFS_PROBE_KEY = "_kernel_live_battery_prefs_probe"


def _run_http_case(conn, case: Case) -> CaseResult:
    """Deterministic HTTP-only cases (sidebar, preferences) — no MLX."""
    t0 = time.monotonic()
    failures: list[str] = []
    thread_id = ""

    if case.id == "sidebar_conversation_lifecycle":
        code, out = http_json(
            "POST", f"{conn.base_url}/conversations/new", _headers(conn), timeout=20.0
        )
        if code != 200 or not isinstance(out, dict) or not out.get("thread_id"):
            failures.append(f"POST /conversations/new failed ({code}): {out}")
        else:
            thread_id = str(out["thread_id"])
            code, listed = http_json(
                "GET", f"{conn.base_url}/conversations", _headers(conn), timeout=20.0
            )
            ids = (
                {c["thread_id"] for c in listed if isinstance(c, dict) and c.get("thread_id")}
                if code == 200 and isinstance(listed, list)
                else set()
            )
            if thread_id not in ids:
                failures.append(f"new thread {thread_id!r} missing from GET /conversations")
            code, _ = http_json(
                "POST",
                f"{conn.base_url}/conversations/{thread_id}/rename",
                _headers(conn),
                {"title": "Gate Battery Probe"},
                timeout=20.0,
            )
            if code != 200:
                failures.append(f"rename failed ({code})")
            code, _ = http_json(
                "POST",
                f"{conn.base_url}/conversations/{thread_id}/trash",
                _headers(conn),
                timeout=20.0,
            )
            if code != 200:
                failures.append(f"trash failed ({code})")
            code, listed2 = http_json(
                "GET", f"{conn.base_url}/conversations", _headers(conn), timeout=20.0
            )
            ids2 = (
                {c["thread_id"] for c in listed2 if isinstance(c, dict)}
                if code == 200 and isinstance(listed2, list)
                else set()
            )
            if thread_id in ids2:
                failures.append("trashed thread still in default /conversations list")
            code, _ = http_json(
                "POST",
                f"{conn.base_url}/conversations/{thread_id}/restore",
                _headers(conn),
                timeout=20.0,
            )
            if code != 200:
                failures.append(f"restore failed ({code})")
            code, listed3 = http_json(
                "GET", f"{conn.base_url}/conversations", _headers(conn), timeout=20.0
            )
            ids3 = (
                {c["thread_id"] for c in listed3 if isinstance(c, dict)}
                if code == 200 and isinstance(listed3, list)
                else set()
            )
            if thread_id not in ids3:
                failures.append("restored thread missing from GET /conversations")

    elif case.id == "preferences_load_save_roundtrip":
        code, before = http_json(
            "GET", f"{conn.base_url}/preferences", _headers(conn), timeout=20.0
        )
        if code != 200 or not isinstance(before, dict):
            failures.append(f"GET /preferences failed ({code}): {before}")
        else:
            prior = before.get(_PREFS_PROBE_KEY)
            probe_val = f"gate-{int(time.time())}"
            code, saved = http_json(
                "POST",
                f"{conn.base_url}/preferences",
                _headers(conn),
                {_PREFS_PROBE_KEY: probe_val},
                timeout=20.0,
            )
            if code != 200:
                failures.append(f"POST /preferences failed ({code}): {saved}")
            else:
                code, after = http_json(
                    "GET", f"{conn.base_url}/preferences", _headers(conn), timeout=20.0
                )
                if code != 200 or not isinstance(after, dict):
                    failures.append(f"GET /preferences after save failed ({code})")
                elif after.get(_PREFS_PROBE_KEY) != probe_val:
                    failures.append(
                        f"preferences round-trip failed: "
                        f"expected {_PREFS_PROBE_KEY}={probe_val!r}, "
                        f"got {after.get(_PREFS_PROBE_KEY)!r}"
                    )
            # Restore prior probe value (or remove key).
            restore_body = {_PREFS_PROBE_KEY: prior} if prior is not None else {}
            if restore_body or prior is None:
                http_json(
                    "POST",
                    f"{conn.base_url}/preferences",
                    _headers(conn),
                    restore_body if restore_body else {_PREFS_PROBE_KEY: ""},
                    timeout=20.0,
                )
    else:
        failures.append(f"unknown http_only case {case.id!r}")

    elapsed = time.monotonic() - t0
    return CaseResult(
        case_id=case.id,
        passed=not failures,
        thread_id=thread_id,
        elapsed_sec=elapsed,
        pillar=case.pillar,
        failures=failures,
    )


def run_case(conn, case: Case, *, prime_welcome: bool = True) -> CaseResult:
    if case.http_only:
        return _run_http_case(conn, case)
    tid = new_conversation(conn)["thread_id"]
    if prime_welcome:
        # Land the production init-welcome turn before the real prompt so the
        # case's first tool-requiring message is turn 1+, exactly like the app.
        _prime_welcome(conn, tid)
    if case.prime_coordinator_batch:
        prime_err = _prime_coordinator_board(conn, tid)
        if prime_err:
            return CaseResult(
                case_id=case.id,
                passed=False,
                thread_id=tid,
                elapsed_sec=0.0,
                pillar=case.pillar,
                failures=[prime_err],
            )
    um_before: dict[str, Any] | None = None
    if _case_uses_user_model_snapshot(case):
        um_before = _user_model_snapshot(conn)

    trace = _trace_path(tid)
    t0 = time.monotonic()

    prompts = [case.prompt, *case.follow_up_prompts]
    out: dict[str, Any] = {}
    segment_tools: list[list[str]] = []
    for i, text in enumerate(prompts):
        events_before = len(_load_trace_events(trace))
        if i == 0:
            out = send_and_wait(
                conn,
                text=text,
                thread_id=tid,
                timeout_sec=case.timeout_sec,
                poll_sec=2.0,
                trace_path=trace,
            )
        else:
            pre_payload = get_messages(conn, tid)
            pre_messages = pre_payload.get("messages") if isinstance(pre_payload, dict) else None
            pre_count = len(pre_messages) if isinstance(pre_messages, list) else 0
            sent = send_message(conn, text=text, thread_id=tid)
            if sent.get("error"):
                out = {"error": sent.get("error")}
                break
            waited = wait_for_run(
                conn,
                tid,
                timeout_sec=case.timeout_sec,
                poll_sec=2.0,
                trace_path=trace,
                baseline_message_count=pre_count,
            )
            out = {"send": sent, "wait": waited}
        events_after = _load_trace_events(trace)
        segment_tools.append(_tool_calls_from_events(events_after[events_before:]))
        if out.get("error"):
            break

    elapsed = time.monotonic() - t0
    if out.get("error"):
        return CaseResult(
            case_id=case.id,
            passed=False,
            thread_id=tid,
            elapsed_sec=elapsed,
            pillar=case.pillar,
            failures=[str(out.get("error"))],
        )
    waited = out.get("wait") or out
    events = _load_trace_events(trace)
    result = _score_case(
        case, thread_id=tid, waited=waited, events=events, segment_tools=segment_tools
    )
    if _case_uses_user_model_snapshot(case):
        um_after = _user_model_snapshot(conn)
        um_failures = _score_user_model_deltas(
            case,
            um_before,
            um_after,
            proposal_surfaced_via_dedup=_proposal_surfaced_via_dedup(events),
        )
        if um_failures:
            result.failures.extend(um_failures)
            result.passed = False
            if _restore_user_model_if_clobbered(conn, case, um_before, um_after):
                result.failures.append(
                    "[harness restored the User Model from snapshot after an "
                    "unexpected mutation — investigate the direct-write guard]"
                )
        _cleanup_user_model_proposals(conn, case, um_before, um_after)
    result.elapsed_sec = elapsed
    return result


def _is_timeout_failure(res: CaseResult) -> bool:
    """A failure dominated by a run timeout (load/model-spin flake), not a real
    assertion miss.

    When a run never produces its final assistant message, the tool-requirement
    checks fail as a *consequence* of the timeout, not because the agent did the
    wrong thing. Those are the cases worth one clean retry. A case that completed
    but failed an assertion (wrong tool, forbidden pattern) is NOT retried — that
    is a genuine signal we must not paper over.
    """
    if res.passed:
        return False
    if not res.timed_out and not any(
        "timed_out waiting for assistant" in f for f in res.failures
    ):
        return False
    # Every failure must be timeout-attributable. A forbidden-tool / wrong-output
    # failure alongside the timeout is still a real signal -> do not retry.
    real_signal_markers = (
        "forbidden tool called",
        "forbidden pattern present",
        "missing required substring",
        "tool results missing error substring",
    )
    for f in res.failures:
        if any(m in f for m in real_signal_markers):
            return False
    return True


def run_case_with_retry(
    conn,
    case: Case,
    *,
    prime_welcome: bool,
    max_attempts: int,
    cooldown_sec: float,
    on_retry=None,
) -> CaseResult:
    """Run a case, retrying once (up to *max_attempts*) on a timeout-class failure.

    A later attempt that passes marks the result ``flaky=True`` and preserves the
    first attempt's failures for reporting. Genuine assertion failures are never
    retried (see ``_is_timeout_failure``).
    """
    res = run_case(conn, case, prime_welcome=prime_welcome)
    res.attempts = 1
    if max_attempts <= 1 or not _is_timeout_failure(res):
        return res
    first_failures = list(res.failures)
    for attempt in range(2, max_attempts + 1):
        if on_retry is not None:
            on_retry(attempt)
        if cooldown_sec > 0:
            time.sleep(cooldown_sec)
        retry = run_case(conn, case, prime_welcome=prime_welcome)
        retry.attempts = attempt
        retry.first_attempt_failures = first_failures
        if retry.passed:
            retry.flaky = True
            return retry
        if not _is_timeout_failure(retry):
            # A non-timeout failure surfaced on retry — real signal, stop early.
            retry.first_attempt_failures = first_failures
            return retry
        res = retry
    res.first_attempt_failures = first_failures
    return res


def main() -> int:
    parser = argparse.ArgumentParser(description="Kernel live agent regression battery")
    parser.add_argument("--cases", default="", help="Comma-separated case ids (default: all)")
    parser.add_argument(
        "--pillar",
        default="",
        help=f"Comma-separated pillars to run (default: all). Choices: {', '.join(PILLARS)}",
    )
    parser.add_argument(
        "--delay",
        type=float,
        default=3.0,
        help="Seconds to pause between cases (reduces backend saturation). Default 3.",
    )
    parser.add_argument("--json", default="", help="Write JSON report to path")
    parser.add_argument("--keep-threads", action="store_true", help="Do not delete probe threads")
    parser.add_argument(
        "--no-welcome",
        action="store_true",
        help=(
            "Skip the production init-welcome turn before each case prompt. By "
            "default the battery mirrors the app (trigger_agent_welcome), so the "
            "case prompt lands at turn 1+ with the greeting in history."
        ),
    )
    parser.add_argument(
        "--reset-board",
        action="store_true",
        help=(
            "Cancel battery-induced kanban tasks between cases (scoped to tasks "
            "created during the run; pre-existing board untouched). Avoids the "
            "50-task cap / cancel-storm soak artifact in durable cases."
        ),
    )
    parser.add_argument(
        "--max-attempts",
        type=int,
        default=2,
        help=(
            "Max attempts per case (default 2). A timeout-class failure is retried "
            "once; a later pass is reported as flaky. Set 1 to disable retries. "
            "Genuine assertion failures (wrong/forbidden tool, bad output) never retry."
        ),
    )
    parser.add_argument(
        "--retry-cooldown",
        type=float,
        default=8.0,
        help="Seconds to pause before a retry attempt (lets MLX/GPU settle). Default 8.",
    )
    parser.add_argument(
        "--gate",
        action="store_true",
        help=(
            "Release-gate mode: flaky passes are non-gating (reported but exit 0). "
            "Hard failures still fail the run."
        ),
    )
    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(ALL_CASES)
    if args.pillar.strip():
        want_pillars = {x.strip() for x in args.pillar.split(",") if x.strip()}
        selected = [c for c in ALL_CASES if c.pillar in want_pillars]
        if not selected:
            print("No cases for pillar(s)", want_pillars, file=sys.stderr)
            return 2
    if args.cases.strip():
        want = {x.strip() for x in args.cases.split(",") if x.strip()}
        selected = [c for c in selected if c.id in want]
        if not selected:
            print("No matching cases", file=sys.stderr)
            return 2

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

    # Snapshot the pre-existing board so --reset-board only ever cancels tasks
    # the battery itself induces (protects a real operator's open work).
    board_baseline: set[str] = _active_task_ids(conn) if args.reset_board else set()

    print(
        f"backend pid={conn.pid} port={conn.port} cases={len(selected)} "
        f"pillars={by_pillar} delay={args.delay}s welcome={'off' if args.no_welcome else 'on'} "
        f"reset_board={'on' if args.reset_board else 'off'} max_attempts={args.max_attempts}\n"
    )
    results: list[CaseResult] = []
    for i, case in enumerate(selected, 1):
        print(f"[{i}/{len(selected)}] [{case.pillar}] {case.id} …", flush=True)

        def _on_retry(attempt: int, _case=case):
            # Clean the board before the retry so a saturated board from this
            # attempt's own task churn doesn't bias the rerun.
            if args.reset_board:
                _reset_new_tasks(conn, board_baseline)
            print(
                f"    [retry] timeout-class failure; attempt {attempt}/{args.max_attempts}…",
                flush=True,
            )

        res = run_case_with_retry(
            conn,
            case,
            prime_welcome=not args.no_welcome,
            max_attempts=args.max_attempts,
            cooldown_sec=args.retry_cooldown,
            on_retry=_on_retry,
        )
        results.append(res)
        if res.passed and res.flaky:
            status = f"PASS (flaky, {res.attempts} tries)"
        elif res.passed:
            status = "PASS"
        else:
            status = f"FAIL ({res.attempts}x)" if res.attempts > 1 else "FAIL"
        print(f"  {status} ({res.elapsed_sec:.1f}s) tools={res.tool_calls}")
        if res.flaky and res.first_attempt_failures:
            print(f"    (first attempt: {'; '.join(res.first_attempt_failures)})")
        if res.failures:
            for f in res.failures:
                print(f"    - {f}")
        if not res.passed and res.assistant_tail:
            print(f"    assistant: {res.assistant_tail[:200]}…")
        print()
        if args.reset_board:
            n = _reset_new_tasks(conn, board_baseline)
            if n:
                print(f"    [reset-board] cancelled {n} battery task(s)\n", flush=True)
        if i < len(selected) and args.delay > 0:
            time.sleep(args.delay)

    passed = sum(1 for r in results if r.passed)
    failed = len(results) - passed
    pillar_summary: dict[str, dict[str, int]] = {}
    for r in results:
        bucket = pillar_summary.setdefault(r.pillar, {"passed": 0, "failed": 0, "total": 0})
        bucket["total"] += 1
        if r.passed:
            bucket["passed"] += 1
        else:
            bucket["failed"] += 1

    flaky = sum(1 for r in results if r.flaky)
    gate_summary = summarize_agent_results(results) if args.gate else None
    flaky_note = f" ({flaky} flaky)" if flaky else ""
    if args.gate and gate_summary:
        print(format_gate_summary_line("agent", gate_summary))
    print(f"SUMMARY: {passed}/{len(results)} passed, {failed} failed{flaky_note}")
    for pillar in PILLARS:
        if pillar in pillar_summary:
            s = pillar_summary[pillar]
            print(f"  {pillar}: {s['passed']}/{s['total']} passed")
    if flaky:
        names = ", ".join(r.case_id for r in results if r.flaky)
        print(f"  flaky (passed on retry): {names}")

    report = {
        "backend": {"pid": conn.pid, "port": conn.port, "repo_root": conn.repo_root},
        "summary": {
            "passed": passed,
            "failed": failed,
            "total": len(results),
            "flaky": flaky,
            "by_pillar": pillar_summary,
            "gate": gate_summary,
        },
        "results": [
            {
                "case_id": r.case_id,
                "pillar": r.pillar,
                "passed": r.passed,
                "thread_id": r.thread_id,
                "elapsed_sec": round(r.elapsed_sec, 2),
                "tool_calls": r.tool_calls,
                "turn_end_reasons": r.turn_end_reasons,
                "failures": r.failures,
                "timed_out": r.timed_out,
                "attempts": r.attempts,
                "flaky": r.flaky,
                "first_attempt_failures": r.first_attempt_failures,
            }
            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 = gate_summary["exit_code"] if gate_summary else (0 if failed == 0 else 1)
    return exit_code


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