P9

name: tests

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]
  workflow_dispatch:

concurrency:
  group: ${{ github.workflow }}-${{ github.ref }}
  cancel-in-progress: true

jobs:
  pytest:
    name: pytest (${{ matrix.python-version }})
    runs-on: ubuntu-latest
    strategy:
      fail-fast: false
      matrix:
        python-version: ["3.11", "3.12", "3.13"]
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python ${{ matrix.python-version }}
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}

      - name: Install dev dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r tests/requirements-dev.txt

      - name: Run conformance battery
        run: |
          python scripts/p9.py conformance -v

      - name: Smoke-test CLI surfaces
        run: |
          export BROOMVA_P9_HOME=/tmp/p9-smoke
          export BROOMVA_P9_POLICY=tests/fixtures/policy-good.yaml
          python scripts/p9.py --help
          python scripts/p9.py doctor || true   # gh may not be authed in CI
          python scripts/p9.py status
          python scripts/p9.py wait-queue list

__pycache__/
*.pyc
.pytest_cache/
.venv/
venv/
.env
.DS_Store

MIT License

Copyright (c) 2026 Carlos D. Escobar-Valbuena (broomva)

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

#!/bin/bash
# p9-escalate-notify.sh — default escalation notify hook.
#
# Invoked by P9 after a Linear ticket is created for an unclassified or
# evaluator-stalled CI failure. Receives JSON on stdin:
#   {pr: int, repo: str, failure_signature: str, linear_ticket: str, attempt: int}
#
# Default behavior: log to ~/.config/broomva/p9/escalations.log. Override
# this hook in .control/policy.yaml -> ci_heal.escalation_channel.notify_hook
# to wire Discord, Telegram, or any other channel via the existing
# claude-remote-sessions infra.

set -euo pipefail

LOG="${BROOMVA_P9_HOME:-${XDG_CONFIG_HOME:-$HOME/.config}/broomva/p9}/escalations.log"
mkdir -p "$(dirname "$LOG")"

PAYLOAD="$(cat)"
TS="$(date -u +%FT%TZ)"
echo "[$TS] $PAYLOAD" >> "$LOG"

exit 0

#!/usr/bin/env python3
"""
p9.py — Broomva CI watcher + productive-wait primitive (bstack P9).

Replaces sleep-based CI waits with an event-driven control loop:
  - `gh pr checks <pr> --watch` via run_in_background as the notification
    mechanism (not polling).
  - context-scoped wait-queue draining session/memory/graph/docs/Linear
    while CI runs.
  - classifier (fast regex filter) + evaluator (progress score) self-heal
    loop with stability-budget termination.
  - merge authorization delegated to existing control metalayer
    (.control/policy.yaml). P9 emits MERGE_READY; metalayer authorizes.

Stdlib-only at runtime. Test-only deps (pytest, vcrpy) live under
tests/requirements-dev.txt.

Spec: docs/superpowers/specs/2026-05-04-p9-ci-watcher-design.md
"""

from __future__ import annotations

import argparse
import contextlib
import dataclasses
import datetime as _dt
import enum
import errno
import fcntl
import hashlib
import json
import os
import re
import subprocess
import sys
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Iterable, Iterator

# ─────────────────────────────────────────────────────────────────────────────
# Exit codes (composable shell semantics)
# ─────────────────────────────────────────────────────────────────────────────
EXIT_OK = 0
EXIT_DEGRADED = 1                # recoverable: run again, may succeed
EXIT_POLICY_ERROR = 2            # policy.yaml missing/malformed (fail-closed)
EXIT_USAGE = 3                   # bad CLI args
EXIT_EXTERNAL_ERROR = 4          # gh/Linear/network failure
EXIT_CONCURRENCY_CEILING = 5     # max_concurrent_prs reached
EXIT_HEAL_LOCK_TIMEOUT = 6
EXIT_AUTO_MERGE_BLOCKED = 7      # auto_merge policy says require_human / notify
EXIT_INVARIANT_VIOLATION = 99    # cardinal-rule breach: cannot persist state

# ─────────────────────────────────────────────────────────────────────────────
# Paths
# ─────────────────────────────────────────────────────────────────────────────
def p9_home() -> Path:
    """State directory; overridable via BROOMVA_P9_HOME for tests."""
    override = os.environ.get("BROOMVA_P9_HOME")
    if override:
        return Path(override)
    xdg = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config")
    return Path(xdg) / "broomva" / "p9"


def state_jsonl() -> Path:
    return p9_home() / "state.jsonl"


def wait_queue_jsonl() -> Path:
    return p9_home() / "wait-queue.jsonl"


def pending_escalations_jsonl() -> Path:
    return p9_home() / "pending-escalations.jsonl"


def heal_lock_path() -> Path:
    return p9_home() / "heal.lock"


def state_lock_path() -> Path:
    return p9_home() / "state.lock"


def queue_lock_path() -> Path:
    return p9_home() / "queue.lock"


def policy_yaml_path() -> Path:
    """Resolve .control/policy.yaml.

    Honors BROOMVA_P9_POLICY env var (used by tests). Otherwise walks from
    cwd upward looking for `.control/policy.yaml`.
    """
    override = os.environ.get("BROOMVA_P9_POLICY")
    if override:
        return Path(override)
    cur = Path.cwd().resolve()
    for parent in [cur, *cur.parents]:
        p = parent / ".control" / "policy.yaml"
        if p.exists():
            return p
    return cur / ".control" / "policy.yaml"  # fail-closed marker


def rubric_md_path() -> Path:
    """Locate references/scoring-rubric.md alongside this script."""
    return Path(__file__).resolve().parent.parent / "references" / "scoring-rubric.md"


# ─────────────────────────────────────────────────────────────────────────────
# Errors
# ─────────────────────────────────────────────────────────────────────────────
class P9Error(Exception):
    """Base exception with an exit code."""
    code = EXIT_DEGRADED


class PolicyError(P9Error):
    code = EXIT_POLICY_ERROR


class IllegalTransitionError(P9Error):
    code = EXIT_INVARIANT_VIOLATION


class ConcurrencyCeilingError(P9Error):
    code = EXIT_CONCURRENCY_CEILING


# ─────────────────────────────────────────────────────────────────────────────
# State machine
# ─────────────────────────────────────────────────────────────────────────────
class PRState(str, enum.Enum):
    PUSHED = "PUSHED"
    WATCHING = "WATCHING"
    GREEN = "GREEN"
    RED_CLASSIFIED = "RED_CLASSIFIED"
    RED_UNCLASSIFIED = "RED_UNCLASSIFIED"
    HEALING = "HEALING"
    MERGE_READY = "MERGE_READY"
    MERGED = "MERGED"
    ESCALATED = "ESCALATED"
    ABANDONED = "ABANDONED"


# Allowed (from -> to) transitions. All edges in spec §5.1.
_TRANSITIONS: set[tuple[PRState, PRState]] = {
    (PRState.PUSHED, PRState.WATCHING),
    (PRState.WATCHING, PRState.GREEN),
    (PRState.WATCHING, PRState.RED_CLASSIFIED),
    (PRState.WATCHING, PRState.RED_UNCLASSIFIED),
    (PRState.RED_CLASSIFIED, PRState.HEALING),
    (PRState.RED_CLASSIFIED, PRState.ESCALATED),  # evaluator-stalled
    (PRState.HEALING, PRState.WATCHING),          # heal pushed; new watch cycle
    (PRState.HEALING, PRState.ESCALATED),         # heal corruption / scope violation
    (PRState.RED_UNCLASSIFIED, PRState.ESCALATED),
    (PRState.GREEN, PRState.MERGE_READY),
    (PRState.MERGE_READY, PRState.MERGED),
    (PRState.MERGE_READY, PRState.WATCHING),       # rare: human pushed amend post-green
    # Terminal "abandoned" reachable from any non-terminal — needed for
    # `p9 abandon` and `p9 cleanup` to drain orphans regardless of the
    # state they're parked in.
    (PRState.PUSHED, PRState.ABANDONED),
    (PRState.WATCHING, PRState.ABANDONED),
    (PRState.HEALING, PRState.ABANDONED),
    (PRState.RED_CLASSIFIED, PRState.ABANDONED),
    (PRState.RED_UNCLASSIFIED, PRState.ABANDONED),
    (PRState.GREEN, PRState.ABANDONED),
    (PRState.MERGE_READY, PRState.ABANDONED),
}


def assert_legal_transition(curr: PRState, nxt: PRState) -> None:
    if curr == nxt:
        return  # idempotent self-event allowed (e.g., status refresh)
    if (curr, nxt) not in _TRANSITIONS:
        raise IllegalTransitionError(
            f"Illegal PR state transition: {curr.value} -> {nxt.value}"
        )


def is_terminal(state: PRState) -> bool:
    return state in {PRState.MERGED, PRState.ESCALATED, PRState.ABANDONED}


# ─────────────────────────────────────────────────────────────────────────────
# Dataclasses
# ─────────────────────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class IsolationTierMap:
    research: str = "none"
    docs: str = "none"
    code_independent: str = "worktree"
    code_dependent: str = "stacked_branch"
    governance: str = "blocked"


@dataclass(frozen=True)
class CIWatchPolicy:
    enabled: bool = True
    max_concurrent_prs: int = 1
    isolation_tier_map: IsolationTierMap = field(default_factory=IsolationTierMap)


@dataclass(frozen=True)
class EscalationChannel:
    linear_team: str = "BRO"
    linear_label: str = "ci-heal-escalation"
    notify_hook: str = "skills/p9/scripts/p9-escalate-notify.sh"


@dataclass(frozen=True)
class CIHealPolicy:
    enabled: bool = True
    max_attempts: int = 5
    stability_floor: float = 0.3
    classified_failure_types: tuple[str, ...] = (
        "lint", "format", "test_flaky", "codegen_drift", "import_missing",
    )
    escalation_channel: EscalationChannel = field(default_factory=EscalationChannel)


@dataclass(frozen=True)
class AutoMergeRule:
    """One auto-merge rule. Either branch_pattern or path_touched (not both).

    branch_pattern: fnmatch glob against the PR's head branch (e.g. "docs/*").
    path_touched:   substring match against any file in the PR diff
                    (e.g. "CLAUDE.md" matches the literal path).
    action:         "auto" → run gh pr merge; "require_human" → hard-block,
                    exit 7; "notify" → emit MERGE_NOTIFIED event, exit 7.

    Rules are evaluated in declaration order; first match wins.
    Matches are short-circuited: `require_human` and `notify` are blocking
    states even if a later rule would auto-merge.
    """
    branch_pattern: str | None = None
    path_touched: str | None = None
    action: str = "notify"


@dataclass(frozen=True)
class AutoMergePolicy:
    enabled: bool = False
    require_no_requested_changes: bool = True
    require_branch_up_to_date: bool = True
    merge_method: str = "squash"          # squash | merge | rebase
    delete_branch: bool = True
    rules: tuple[AutoMergeRule, ...] = ()
    default_action: str = "notify"        # fail-safe default


@dataclass(frozen=True)
class PolicyConfig:
    ci_watch: CIWatchPolicy
    ci_heal: CIHealPolicy
    auto_merge: AutoMergePolicy = field(default_factory=AutoMergePolicy)


@dataclass
class PRStateEvent:
    """One row of state.jsonl."""
    ts: str
    pr: int
    repo: str
    from_state: str
    to_state: str
    watcher_id: str
    attempt: int = 0
    evaluator_score: float | None = None
    extra: dict[str, Any] = field(default_factory=dict)

    def to_jsonl(self) -> str:
        return json.dumps(dataclasses.asdict(self), separators=(",", ":"))


@dataclass(frozen=True)
class ClassifierResult:
    failure_type: str
    classified: bool
    confidence: float
    heal_command: str | None
    signature_hash: str
    rationale: str


@dataclass(frozen=True)
class EvaluatorResult:
    progress_score: float
    signature_changed: bool
    failures_decreased: bool
    budget_remaining: float
    classifier_confidence: float
    stalled: bool


_QUEUE_PRIORITY = ("session", "memory", "graph", "docs", "linear")


@dataclass
class WaitQueueItem:
    """One row of wait-queue.jsonl."""
    id: str
    source: str
    item: str
    created_at: str
    pr: int | None = None
    isolation_tier: str = "none"

    def to_jsonl(self) -> str:
        return json.dumps(dataclasses.asdict(self), separators=(",", ":"))


# ─────────────────────────────────────────────────────────────────────────────
# Filesystem helpers — JSONL append, locks, corruption recovery
# ─────────────────────────────────────────────────────────────────────────────
@contextlib.contextmanager
def file_lock(lock_path: Path, timeout_s: float = 30.0) -> Iterator[None]:
    """Cross-process flock with timeout. Creates lock file if absent."""
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(str(lock_path), os.O_CREAT | os.O_RDWR, 0o644)
    deadline = time.monotonic() + timeout_s
    try:
        while True:
            try:
                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                break
            except OSError as e:
                if e.errno not in (errno.EAGAIN, errno.EACCES):
                    raise
                if time.monotonic() >= deadline:
                    raise P9Error(
                        f"flock timeout after {timeout_s}s on {lock_path}"
                    ) from e
                time.sleep(0.05)
        yield
    finally:
        with contextlib.suppress(OSError):
            fcntl.flock(fd, fcntl.LOCK_UN)
        os.close(fd)


def jsonl_append(path: Path, payload: str, lock: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with file_lock(lock):
        with path.open("a", encoding="utf-8") as f:
            f.write(payload)
            if not payload.endswith("\n"):
                f.write("\n")


def jsonl_read_all(path: Path) -> tuple[list[dict[str, Any]], int]:
    """Read JSONL, skipping the last line if it's a partial/corrupt write.

    Returns (rows, dropped) where dropped is the number of lines we skipped
    (currently 0 or 1).
    """
    if not path.exists():
        return [], 0
    raw = path.read_text(encoding="utf-8")
    if not raw:
        return [], 0
    lines = raw.splitlines()
    rows: list[dict[str, Any]] = []
    dropped = 0
    for i, line in enumerate(lines):
        if not line.strip():
            continue
        try:
            rows.append(json.loads(line))
        except json.JSONDecodeError:
            # JSONL append-only design: only the last line can be partial.
            # If the corrupt line is not the last, that's an invariant violation.
            if i == len(lines) - 1:
                dropped = 1
            else:
                raise IllegalTransitionError(
                    f"Mid-file JSON corruption in {path} at line {i + 1}"
                )
    return rows, dropped


# ─────────────────────────────────────────────────────────────────────────────
# Policy loader (strict; fail-closed)
# ─────────────────────────────────────────────────────────────────────────────
def _yaml_loader():
    """Lazy-import PyYAML; fall back to a tiny parser for the keys we need."""
    try:
        import yaml  # type: ignore
        return yaml.safe_load
    except ImportError:
        return _minimal_yaml_load


def _minimal_yaml_load(text: str) -> dict[str, Any]:
    """Tiny YAML subset: top-level mappings, nested mappings, scalars,
    inline lists `[a,b]`, and block-style list-of-dicts `- key: val`.

    Sufficient for `.control/policy.yaml` ci_watch / ci_heal / auto_merge
    blocks. Not a general YAML parser — intentionally narrow so it fails
    noisily on anything unexpected.
    """
    out: dict[str, Any] = {}
    # stack tracks (indent, dict) frames for nested mappings.
    stack: list[tuple[int, dict[str, Any]]] = [(-1, out)]
    # list_stack tracks (indent, list) frames for active block lists.
    list_stack: list[tuple[int, list[Any]]] = []

    for raw_line in text.splitlines():
        line = raw_line.rstrip()
        if not line or line.lstrip().startswith("#"):
            continue
        indent = len(line) - len(line.lstrip(" "))
        body = line.strip()

        if body.startswith("- "):
            # block list item; pop deeper frames first
            while list_stack and list_stack[-1][0] >= indent:
                list_stack.pop()
            while stack and stack[-1][0] >= indent:
                stack.pop()
            if not list_stack:
                continue  # orphan list item — ignore
            target_list = list_stack[-1][1]
            inner = body[2:].strip()
            if ":" in inner and not (inner.startswith("[") or inner.startswith('"')):
                # block-style list-of-dicts: `- key: val` starts a new dict
                # and any subsequent more-indented `key: val` lines populate it
                key, _, value_str = inner.partition(":")
                value_str = value_str.strip()
                new_dict: dict[str, Any] = {}
                if value_str:
                    new_dict[key.strip()] = _scalar(value_str)
                target_list.append(new_dict)
                # Subsequent same-indent `- ` items pop this frame; nested
                # `key: val` lines at deeper indent populate new_dict.
                stack.append((indent, new_dict))
            else:
                # scalar list item
                target_list.append(_scalar(inner))
            continue

        if ":" not in body:
            continue
        key, _, value_str = body.partition(":")
        value_str = value_str.strip()
        # Pop deeper frames
        while stack and stack[-1][0] >= indent:
            stack.pop()
        while list_stack and list_stack[-1][0] >= indent:
            list_stack.pop()
        parent = stack[-1][1] if stack else out

        if value_str == "":
            # Mapping or list — we don't know yet. Default to mapping;
            # if the next non-blank line is a `- `, the open list_stack
            # frame below catches it.
            new_map: dict[str, Any] = {}
            new_list: list[Any] = []
            parent[key.strip()] = new_map
            stack.append((indent, new_map))
            # Tentatively register a list at deeper indent; whichever
            # the next line uses (mapping vs list) wins via stack pop.
            # We need a sentinel: register the list only if the next
            # `-` arrives at deeper indent. Easiest: pre-register both
            # but only commit to one once content arrives.
            # Concretely: replace mapping with list lazily on first `-`.
            list_stack.append((indent, new_list))
            # If the next content is a `- ` deeper than `indent`, we'll
            # convert: replace parent[key] with new_list and pop the dict.
            stack[-1] = (indent, new_map)
            # Stash a reference so a `- ` line can swap mapping → list
            new_map.setdefault("__p9_yaml_pending_list_holder__",
                               (parent, key.strip(), new_list))
        elif value_str.startswith("[") and value_str.endswith("]"):
            inner = value_str[1:-1].strip()
            parent[key.strip()] = (
                [_scalar(p.strip()) for p in inner.split(",") if p.strip()]
                if inner else []
            )
        else:
            parent[key.strip()] = _scalar(value_str)

    # Second pass: any dict that still holds the pending-list sentinel
    # AND has no other keys is actually an empty dict; any dict whose
    # corresponding list got populated had its mapping replaced.
    _resolve_pending_lists(out)
    return out


def _resolve_pending_lists(node: Any) -> None:
    """Walk the parsed tree; convert mapping→list where a `-` block was
    used. Sentinel removal must come before semantic validation."""
    if isinstance(node, dict):
        sentinel_key = "__p9_yaml_pending_list_holder__"
        if sentinel_key in node:
            holder = node[sentinel_key]
            del node[sentinel_key]
            parent, key, the_list = holder
            if the_list:
                # `-` lines populated the list; replace mapping with list
                parent[key] = the_list
        for v in list(node.values()):
            _resolve_pending_lists(v)
    elif isinstance(node, list):
        for item in node:
            _resolve_pending_lists(item)


def _scalar(s: str) -> Any:
    s = s.strip().strip('"').strip("'")
    if s.lower() == "true":
        return True
    if s.lower() == "false":
        return False
    if s.lower() in ("null", "none", "~"):
        return None
    try:
        return int(s)
    except ValueError:
        pass
    try:
        return float(s)
    except ValueError:
        pass
    return s


def load_policy(path: Path | str | None = None) -> PolicyConfig:
    """Load .control/policy.yaml. **Fail-closed** on missing/malformed blocks.

    Accepts both `Path` and `str` (str is coerced — historic callers were
    inconsistent). None falls back to `policy_yaml_path()`.
    """
    p = Path(path) if path is not None else policy_yaml_path()
    if not p.exists():
        raise PolicyError(f"policy.yaml not found at {p}")
    try:
        loader = _yaml_loader()
        data = loader(p.read_text(encoding="utf-8"))
    except Exception as e:
        raise PolicyError(f"policy.yaml malformed: {e}") from e
    if not isinstance(data, dict):
        raise PolicyError("policy.yaml must be a mapping at the top level")
    if "ci_watch" not in data:
        raise PolicyError("policy.yaml missing required block: ci_watch")
    if "ci_heal" not in data:
        raise PolicyError("policy.yaml missing required block: ci_heal")
    return _parse_policy(data)


def _parse_policy(data: dict[str, Any]) -> PolicyConfig:
    cw_raw = data.get("ci_watch") or {}
    ch_raw = data.get("ci_heal") or {}
    iso_raw = cw_raw.get("isolation_tier_map") or {}
    esc_raw = ch_raw.get("escalation_channel") or {}
    types_raw = ch_raw.get("classified_failure_types") or ()
    if not isinstance(types_raw, (list, tuple)):
        raise PolicyError("ci_heal.classified_failure_types must be a list")
    return PolicyConfig(
        ci_watch=CIWatchPolicy(
            enabled=bool(cw_raw.get("enabled", True)),
            max_concurrent_prs=int(cw_raw.get("max_concurrent_prs", 1)),
            isolation_tier_map=IsolationTierMap(
                research=str(iso_raw.get("research", "none")),
                docs=str(iso_raw.get("docs", "none")),
                code_independent=str(iso_raw.get("code_independent", "worktree")),
                code_dependent=str(iso_raw.get("code_dependent", "stacked_branch")),
                governance=str(iso_raw.get("governance", "blocked")),
            ),
        ),
        ci_heal=CIHealPolicy(
            enabled=bool(ch_raw.get("enabled", True)),
            max_attempts=int(ch_raw.get("max_attempts", 5)),
            stability_floor=float(ch_raw.get("stability_floor", 0.3)),
            classified_failure_types=tuple(str(t) for t in types_raw),
            escalation_channel=EscalationChannel(
                linear_team=str(esc_raw.get("linear_team", "BRO")),
                linear_label=str(esc_raw.get("linear_label", "ci-heal-escalation")),
                notify_hook=str(
                    esc_raw.get(
                        "notify_hook",
                        "skills/p9/scripts/p9-escalate-notify.sh",
                    )
                ),
            ),
        ),
        auto_merge=_parse_auto_merge(data.get("auto_merge")),
    )


_AUTO_MERGE_ACTIONS = ("auto", "require_human", "notify")


def _parse_auto_merge(raw: Any) -> AutoMergePolicy:
    """Parse the optional auto_merge: block. Absence is *not* an error —
    auto-merge is opt-in; missing block defaults to disabled (fail-safe)."""
    if raw is None:
        return AutoMergePolicy()
    if not isinstance(raw, dict):
        raise PolicyError("auto_merge must be a mapping if present")
    rules_raw = raw.get("rules") or []
    if not isinstance(rules_raw, list):
        raise PolicyError("auto_merge.rules must be a list")
    rules: list[AutoMergeRule] = []
    for i, r in enumerate(rules_raw):
        if not isinstance(r, dict):
            raise PolicyError(f"auto_merge.rules[{i}] must be a mapping")
        action = str(r.get("action", "notify"))
        if action not in _AUTO_MERGE_ACTIONS:
            raise PolicyError(
                f"auto_merge.rules[{i}].action must be one of {_AUTO_MERGE_ACTIONS}, "
                f"got {action!r}"
            )
        bp = r.get("branch_pattern")
        pt = r.get("path_touched")
        if bp and pt:
            raise PolicyError(
                f"auto_merge.rules[{i}] cannot set both branch_pattern and path_touched"
            )
        if not bp and not pt:
            raise PolicyError(
                f"auto_merge.rules[{i}] must set either branch_pattern or path_touched"
            )
        rules.append(AutoMergeRule(
            branch_pattern=str(bp) if bp else None,
            path_touched=str(pt) if pt else None,
            action=action,
        ))
    default = str(raw.get("default_action", "notify"))
    if default not in _AUTO_MERGE_ACTIONS:
        raise PolicyError(
            f"auto_merge.default_action must be one of {_AUTO_MERGE_ACTIONS}, "
            f"got {default!r}"
        )
    method = str(raw.get("merge_method", "squash"))
    if method not in ("squash", "merge", "rebase"):
        raise PolicyError(
            f"auto_merge.merge_method must be squash|merge|rebase, got {method!r}"
        )
    return AutoMergePolicy(
        enabled=bool(raw.get("enabled", False)),
        require_no_requested_changes=bool(raw.get("require_no_requested_changes", True)),
        require_branch_up_to_date=bool(raw.get("require_branch_up_to_date", True)),
        merge_method=method,
        delete_branch=bool(raw.get("delete_branch", True)),
        rules=tuple(rules),
        default_action=default,
    )


# ─────────────────────────────────────────────────────────────────────────────
# Auto-merge matcher
# ─────────────────────────────────────────────────────────────────────────────
import fnmatch  # noqa: E402  (kept near use site for clarity)


def match_auto_merge_action(
    policy: AutoMergePolicy,
    *,
    branch: str,
    paths_touched: Iterable[str],
) -> tuple[str, str]:
    """First-match-wins evaluator. Returns (action, reason).

    Path rules are evaluated FIRST regardless of order — a path-touched
    `require_human` rule (e.g. CLAUDE.md) is a hard block that cannot be
    overridden by a later branch rule. This implements the "governance
    paths always block" invariant from the original brainstorming.
    """
    paths = list(paths_touched)

    # Pass 1: any path rule with require_human is a hard block.
    for rule in policy.rules:
        if rule.path_touched and rule.action == "require_human":
            for p in paths:
                if rule.path_touched in p:
                    return ("require_human",
                            f"path rule blocks: {rule.path_touched!r} in {p}")

    # Pass 2: first match wins (path or branch).
    for rule in policy.rules:
        if rule.path_touched:
            for p in paths:
                if rule.path_touched in p:
                    return (rule.action,
                            f"path rule matched: {rule.path_touched!r} in {p}")
        if rule.branch_pattern and fnmatch.fnmatch(branch, rule.branch_pattern):
            return (rule.action, f"branch rule matched: {rule.branch_pattern!r}")

    return (policy.default_action, "no rule matched; using default_action")


# ─────────────────────────────────────────────────────────────────────────────
# Failure classifier (rubric-driven regex matcher)
# ─────────────────────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class RubricEntry:
    failure_type: str
    patterns: tuple[re.Pattern[str], ...]
    heal_command: str | None      # None => escalate (cannot auto-fix)
    confidence_floor: float = 0.7  # below this, drop to "unclassified"


def _builtin_rubric() -> tuple[RubricEntry, ...]:
    """Default rubric. Mirrors references/scoring-rubric.md.

    Rubric markdown is the authoritative source for humans; this constant is
    the authoritative source for code. Tests assert the two stay in sync.
    """
    return (
        RubricEntry(
            failure_type="lint",
            patterns=(
                re.compile(r"biome\s+(check|lint).*(found|error)", re.IGNORECASE),
                re.compile(r"eslint.*\d+\s+(error|problem)", re.IGNORECASE),
                re.compile(r"clippy.*::error", re.IGNORECASE),
                re.compile(r"::error.*lint", re.IGNORECASE),
            ),
            heal_command="bun run lint:fix",
        ),
        RubricEntry(
            failure_type="format",
            patterns=(
                re.compile(r"prettier.*--check.*would reformat", re.IGNORECASE),
                re.compile(r"rustfmt.*Diff in", re.IGNORECASE),
                re.compile(r"would reformat", re.IGNORECASE),
            ),
            heal_command="bun run format",
        ),
        RubricEntry(
            failure_type="type",
            patterns=(
                re.compile(r"\berror TS\d+\b"),
                re.compile(r"\berror\[E\d+\]"),
                re.compile(r"type error.*at\s+[\w./]+:\d+", re.IGNORECASE),
            ),
            heal_command=None,  # escalate — type errors need human reasoning
        ),
        RubricEntry(
            failure_type="test_flaky",
            patterns=(
                # Detected by signature-history rather than text — see
                # classify_with_history. The pattern below matches generic
                # test failure for fallback recognition only.
                re.compile(r"^\s*FAIL\s+", re.MULTILINE),
            ),
            heal_command="gh run rerun --failed",
            confidence_floor=0.9,  # only confident if history confirms flakiness
        ),
        RubricEntry(
            failure_type="codegen_drift",
            patterns=(
                re.compile(r"(generated|codegen).*(out of (date|sync)|stale)", re.IGNORECASE),
                re.compile(r"schema mismatch", re.IGNORECASE),
                re.compile(r"graphql codegen.*diff", re.IGNORECASE),
            ),
            heal_command="bun run codegen",
        ),
        RubricEntry(
            failure_type="import_missing",
            patterns=(
                re.compile(r"Cannot find module ['\"]([^'\"]+)['\"]", re.IGNORECASE),
                re.compile(r"unresolved import\s+`([^`]+)`", re.IGNORECASE),
                re.compile(r"Module not found", re.IGNORECASE),
            ),
            heal_command=None,  # complex — escalate v1
        ),
    )


def _signature_hash(log: str) -> str:
    """Stable signature for a failure log: first error-ish line, normalized."""
    candidates = [
        ln for ln in log.splitlines()
        if re.search(r"\b(error|FAIL|fatal)\b", ln, re.IGNORECASE)
    ]
    src = candidates[0] if candidates else log[:200]
    norm = re.sub(r"\s+", " ", src).strip()
    return hashlib.sha256(norm.encode("utf-8")).hexdigest()[:16]


def classify(
    log: str,
    rubric: tuple[RubricEntry, ...] | None = None,
) -> ClassifierResult:
    """Pure classifier: regex match against the rubric.

    Returns the best match. If best confidence < that entry's floor, returns
    an `unclassified` result (which the caller must treat as `escalate`).
    """
    rubric = rubric or _builtin_rubric()
    best: tuple[RubricEntry, float] | None = None
    for entry in rubric:
        score = _entry_score(entry, log)
        if score == 0.0:
            continue
        if best is None or score > best[1]:
            best = (entry, score)
    sig = _signature_hash(log)
    if best is None or best[1] < best[0].confidence_floor:
        return ClassifierResult(
            failure_type="unclassified",
            classified=False,
            confidence=best[1] if best else 0.0,
            heal_command=None,
            signature_hash=sig,
            rationale=(
                f"no rubric entry above floor (best={best[0].failure_type}@{best[1]:.2f})"
                if best else "no rubric pattern matched"
            ),
        )
    entry, conf = best
    return ClassifierResult(
        failure_type=entry.failure_type,
        classified=entry.heal_command is not None,
        confidence=conf,
        heal_command=entry.heal_command,
        signature_hash=sig,
        rationale=f"matched {entry.failure_type} at confidence {conf:.2f}",
    )


def _entry_score(entry: RubricEntry, log: str) -> float:
    """How well does this rubric entry match? 0.0–1.0."""
    matched = sum(1 for p in entry.patterns if p.search(log))
    if matched == 0:
        return 0.0
    # Scale by fraction of patterns hit, then cap at 1.0. Multi-pattern hits
    # boost confidence; single-pattern hits sit at ~0.7 — exactly the default
    # confidence_floor, so single-pattern matches *just* qualify.
    base = 0.7 + 0.1 * (matched - 1)
    return min(base, 1.0)


# ─────────────────────────────────────────────────────────────────────────────
# Evaluator (progress-score brain)
# ─────────────────────────────────────────────────────────────────────────────
def evaluate(
    *,
    attempt: int,
    max_attempts: int,
    classifier_confidence: float,
    prev_signature: str | None,
    curr_signature: str,
    prev_failure_count: int | None,
    curr_failure_count: int,
    stability_floor: float,
) -> EvaluatorResult:
    """Compute progress_score per spec §6.3.

    progress_score =
       0.4 × signature_changed?   +
       0.3 × failures_decreased?  +
       0.2 × (1 - attempt/max)    +
       0.1 × classifier_confidence
    """
    sig_changed = prev_signature is not None and curr_signature != prev_signature
    failures_dec = (
        prev_failure_count is not None and curr_failure_count < prev_failure_count
    )
    budget = max(0.0, 1.0 - (attempt / max(1, max_attempts)))
    score = (
        0.4 * (1.0 if sig_changed else 0.0)
        + 0.3 * (1.0 if failures_dec else 0.0)
        + 0.2 * budget
        + 0.1 * classifier_confidence
    )
    return EvaluatorResult(
        progress_score=round(score, 4),
        signature_changed=sig_changed,
        failures_decreased=failures_dec,
        budget_remaining=round(budget, 4),
        classifier_confidence=classifier_confidence,
        stalled=score < stability_floor,
    )


def stalled_for_two_cycles(
    history: Iterable[float], stability_floor: float
) -> bool:
    """True iff the last two evaluator scores are both below floor."""
    last_two = list(history)[-2:]
    return len(last_two) >= 2 and all(s < stability_floor for s in last_two)


# ─────────────────────────────────────────────────────────────────────────────
# Wait queue
# ─────────────────────────────────────────────────────────────────────────────
def _validate_source(source: str) -> str:
    if source not in _QUEUE_PRIORITY:
        raise P9Error(
            f"invalid source '{source}'; must be one of {_QUEUE_PRIORITY}"
        )
    return source


def queue_push(item: str, source: str, *, pr: int | None = None,
               isolation_tier: str = "none") -> WaitQueueItem:
    src = _validate_source(source)
    entry = WaitQueueItem(
        id=uuid.uuid4().hex[:12],
        source=src,
        item=item,
        created_at=_utcnow(),
        pr=pr,
        isolation_tier=isolation_tier,
    )
    jsonl_append(wait_queue_jsonl(), entry.to_jsonl(), queue_lock_path())
    return entry


def queue_list() -> list[WaitQueueItem]:
    rows, _ = jsonl_read_all(wait_queue_jsonl())
    items = [WaitQueueItem(**r) for r in rows]
    items.sort(key=lambda it: (_QUEUE_PRIORITY.index(it.source), it.created_at))
    return items


def queue_pop() -> WaitQueueItem | None:
    """Atomic pop: re-write the queue without the highest-priority entry."""
    with file_lock(queue_lock_path()):
        rows, _ = jsonl_read_all(wait_queue_jsonl())
        if not rows:
            return None
        items = [WaitQueueItem(**r) for r in rows]
        items.sort(key=lambda it: (_QUEUE_PRIORITY.index(it.source), it.created_at))
        head, *rest = items
        wait_queue_jsonl().write_text(
            "".join(it.to_jsonl() + "\n" for it in rest), encoding="utf-8",
        )
        return head


def queue_clear() -> int:
    with file_lock(queue_lock_path()):
        if not wait_queue_jsonl().exists():
            return 0
        before = len(wait_queue_jsonl().read_text(encoding="utf-8").splitlines())
        wait_queue_jsonl().write_text("", encoding="utf-8")
        return before


# ─────────────────────────────────────────────────────────────────────────────
# State store
# ─────────────────────────────────────────────────────────────────────────────
def _utcnow() -> str:
    return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds")


def append_state_event(event: PRStateEvent) -> None:
    assert_legal_transition(PRState(event.from_state), PRState(event.to_state))
    jsonl_append(state_jsonl(), event.to_jsonl(), state_lock_path())


def current_pr_state(pr: int) -> PRState | None:
    rows, _ = jsonl_read_all(state_jsonl())
    last: PRState | None = None
    for r in rows:
        if r.get("pr") == pr:
            last = PRState(r["to_state"])
    return last


def open_prs() -> list[dict[str, Any]]:
    """All PRs that haven't reached a terminal state."""
    rows, _ = jsonl_read_all(state_jsonl())
    seen: dict[int, dict[str, Any]] = {}
    for r in rows:
        seen[r["pr"]] = r
    return [r for r in seen.values() if not is_terminal(PRState(r["to_state"]))]


# ─────────────────────────────────────────────────────────────────────────────
# Watcher manager (subprocess control for `gh pr checks --watch`)
# ─────────────────────────────────────────────────────────────────────────────
def spawn_watcher(pr: int, repo: str | None = None,
                  *, dry_run: bool = False) -> subprocess.Popen[bytes] | None:
    """Spawn `gh pr checks <pr> --watch` detached. Returns the Popen or None
    in dry_run mode (used by tests and `--background --dry-run`)."""
    if dry_run:
        return None
    cmd = ["gh", "pr", "checks", str(pr), "--watch"]
    if repo:
        cmd += ["--repo", repo]
    return subprocess.Popen(
        cmd,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )


def is_watcher_alive(pid: int) -> bool:
    if pid <= 0:
        return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True


# ─────────────────────────────────────────────────────────────────────────────
# Concurrency
# ─────────────────────────────────────────────────────────────────────────────
def enforce_concurrency_ceiling(policy: PolicyConfig) -> None:
    open_count = len(open_prs())
    if open_count >= policy.ci_watch.max_concurrent_prs:
        raise ConcurrencyCeilingError(
            f"max_concurrent_prs={policy.ci_watch.max_concurrent_prs} "
            f"already in flight ({open_count} open)"
        )


# ─────────────────────────────────────────────────────────────────────────────
# Subcommand handlers
# ─────────────────────────────────────────────────────────────────────────────
def cmd_doctor(_args: argparse.Namespace) -> int:
    problems: list[str] = []
    # 1. gh present + authed
    try:
        out = subprocess.run(
            ["gh", "auth", "status"],
            capture_output=True, text=True, timeout=10, check=False,
        )
        if out.returncode != 0:
            problems.append(f"gh auth status non-zero: {out.stderr.strip()[:200]}")
    except FileNotFoundError:
        problems.append("gh CLI not installed")
    except subprocess.TimeoutExpired:
        problems.append("gh auth status timed out")

    # 2. State directory writable
    try:
        p9_home().mkdir(parents=True, exist_ok=True)
        probe = p9_home() / ".doctor-probe"
        probe.write_text("ok", encoding="utf-8")
        probe.unlink()
    except OSError as e:
        problems.append(f"state directory not writable: {e}")

    # 3. Policy blocks present
    try:
        load_policy()
    except PolicyError as e:
        problems.append(f"policy: {e}")

    # 4. Rubric file present (if absent, builtin still works but we warn)
    if not rubric_md_path().exists():
        problems.append(
            f"references/scoring-rubric.md missing at {rubric_md_path()} "
            f"(builtin rubric still available but markdown is the human-canonical source)"
        )

    if not problems:
        print("p9 doctor: ok")
        return EXIT_OK
    print("p9 doctor: degraded")
    for p in problems:
        print(f"  - {p}")
    # Policy errors are exit 2 (fail-closed); other degradations exit 1.
    if any(p.startswith("policy:") for p in problems):
        return EXIT_POLICY_ERROR
    return EXIT_DEGRADED


def cmd_abandon(args: argparse.Namespace) -> int:
    """Mark a PR as ABANDONED.

    Idempotent on already-terminal states. Emits a terminal-state event so
    `max_concurrent_prs` accounting is freed and `p9 cleanup` doesn't
    re-flag it.
    """
    pr = int(args.pr)
    state = current_pr_state(pr)
    if state is None:
        print(f"PR #{pr}: no state to abandon", file=sys.stderr)
        return EXIT_DEGRADED
    if is_terminal(state):
        print(f"PR #{pr}: already terminal ({state.value}); no-op")
        return EXIT_OK
    repo = args.repo or _detect_repo() or ""
    append_state_event(PRStateEvent(
        ts=_utcnow(), pr=pr, repo=repo,
        from_state=state.value,
        to_state=PRState.ABANDONED.value,
        watcher_id="abandon",
        extra={"reason": args.reason or "manual abandon"},
    ))
    print(f"PR #{pr}: {state.value} → ABANDONED")
    return EXIT_OK


def cmd_cleanup(args: argparse.Namespace) -> int:
    """Drain orphan watchers by polling GitHub for each open row's true state.

    For every PR in a non-terminal local state, queries
    `gh pr view --json state,mergedAt`. If GitHub reports MERGED or CLOSED,
    appends a terminal ABANDONED event with the reason. If GitHub reports
    OPEN, leaves the row alone. PRs that fail to query are reported but
    not abandoned (no false-positive cleanup).
    """
    rows = open_prs()
    if not rows:
        print("p9 cleanup: no open PRs")
        return EXIT_OK
    cleaned = 0
    skipped = 0
    for row in rows:
        pr = row["pr"]
        repo = row.get("repo") or _detect_repo() or ""
        cmd = ["gh", "pr", "view", str(pr), "--json", "state,mergedAt"]
        if repo:
            cmd += ["--repo", repo]
        result = subprocess.run(cmd, capture_output=True, text=True,
                                timeout=30, check=False)
        if result.returncode != 0:
            print(f"  #{pr}: cannot query gh; leaving as {row['to_state']} "
                  f"({result.stderr.strip()[:80]})")
            skipped += 1
            continue
        try:
            data = json.loads(result.stdout)
        except json.JSONDecodeError:
            print(f"  #{pr}: gh returned non-JSON; leaving as {row['to_state']}")
            skipped += 1
            continue
        gh_state = (data.get("state") or "").upper()
        from_state = PRState(row["to_state"])
        if gh_state in ("MERGED", "CLOSED"):
            reason = ("merged outside p9" if gh_state == "MERGED"
                      else "closed outside p9")
            append_state_event(PRStateEvent(
                ts=_utcnow(), pr=pr, repo=repo,
                from_state=from_state.value,
                to_state=PRState.ABANDONED.value,
                watcher_id="cleanup",
                extra={"reason": reason, "gh_state": gh_state},
            ))
            print(f"  #{pr}: {from_state.value} → ABANDONED ({reason})")
            cleaned += 1
        else:
            print(f"  #{pr}: still OPEN; leaving as {from_state.value}")
    print(f"p9 cleanup: drained {cleaned}, skipped {skipped}")
    return EXIT_OK


def cmd_auto_merge(args: argparse.Namespace) -> int:
    """Auto-merge actuator. Closes the gap between MERGE_READY signal and
    actual `gh pr merge` execution.

    Flow:
      1. Load policy → bail if auto_merge.enabled is false.
      2. Verify PR is in MERGE_READY state.
      3. Fetch branch + touched paths via `gh pr view`.
      4. Match against policy rules → action ∈ {auto, require_human, notify}.
      5. auto → run `gh pr merge` (or print plan in --dry-run mode), transition
         MERGE_READY → MERGED, return 0.
         require_human / notify → idempotent self-transition with extra payload
         indicating block reason, return 7 (EXIT_AUTO_MERGE_BLOCKED).
    """
    pr = int(args.pr)
    repo = args.repo or _detect_repo() or ""
    policy = load_policy()
    if not policy.auto_merge.enabled:
        print("auto_merge.enabled=false in policy; refusing to merge", file=sys.stderr)
        return EXIT_POLICY_ERROR

    state = current_pr_state(pr)
    if state != PRState.MERGE_READY:
        print(
            f"PR #{pr} not in MERGE_READY (current={state.value if state else 'UNKNOWN'}); "
            f"call `p9 merge-ready` first",
            file=sys.stderr,
        )
        return EXIT_DEGRADED

    branch, paths = _gh_pr_branch_and_paths(pr, repo)
    action, reason = match_auto_merge_action(
        policy.auto_merge, branch=branch, paths_touched=paths,
    )

    if action != "auto":
        # Block: emit idempotent state event with rationale; never merge.
        append_state_event(PRStateEvent(
            ts=_utcnow(),
            pr=pr, repo=repo,
            from_state=PRState.MERGE_READY.value,
            to_state=PRState.MERGE_READY.value,
            watcher_id="auto-merge",
            extra={"auto_merge": {"action": action, "reason": reason,
                                  "branch": branch, "paths": list(paths)[:20]}},
        ))
        print(f"auto-merge blocked: action={action}; reason={reason}",
              file=sys.stderr)
        return EXIT_AUTO_MERGE_BLOCKED

    # Auto path
    if args.dry_run:
        print(f"auto-merge dry-run: would merge PR #{pr} ({branch}) via "
              f"`gh pr merge --{policy.auto_merge.merge_method}`")
        return EXIT_OK

    rc = _gh_pr_merge(
        pr, repo,
        method=policy.auto_merge.merge_method,
        delete_branch=policy.auto_merge.delete_branch,
    )
    if rc != 0:
        print(f"gh pr merge exited {rc}; PR not merged", file=sys.stderr)
        return EXIT_EXTERNAL_ERROR

    append_state_event(PRStateEvent(
        ts=_utcnow(),
        pr=pr, repo=repo,
        from_state=PRState.MERGE_READY.value,
        to_state=PRState.MERGED.value,
        watcher_id="auto-merge",
        extra={"auto_merge": {"action": "auto", "reason": reason,
                              "branch": branch,
                              "method": policy.auto_merge.merge_method}},
    ))
    print(f"auto-merge: PR #{pr} merged ({branch})")
    return EXIT_OK


def _gh_pr_branch_and_paths(pr: int, repo: str) -> tuple[str, list[str]]:
    """Return (head_branch, [files_touched]) for a PR via `gh pr view`."""
    cmd = ["gh", "pr", "view", str(pr),
           "--json", "headRefName,files",
           "-q", '{branch: .headRefName, files: [.files[].path]}']
    if repo:
        cmd += ["--repo", repo]
    out = subprocess.run(cmd, capture_output=True, text=True, timeout=30, check=False)
    if out.returncode != 0:
        raise P9Error(f"gh pr view failed: {out.stderr.strip()[:200]}")
    try:
        data = json.loads(out.stdout)
    except json.JSONDecodeError as e:
        raise P9Error(f"gh pr view returned non-JSON: {e}") from e
    return str(data.get("branch", "")), list(data.get("files") or [])


def _gh_pr_merge(pr: int, repo: str, *, method: str, delete_branch: bool) -> int:
    """Invoke `gh pr merge` with the configured method. Returns exit code."""
    cmd = ["gh", "pr", "merge", str(pr), f"--{method}"]
    if delete_branch:
        cmd.append("--delete-branch")
    if repo:
        cmd += ["--repo", repo]
    return subprocess.run(cmd, check=False).returncode


def cmd_conformance(args: argparse.Namespace) -> int:
    """Run the full pytest battery (unit + integration + chaos).

    Used as the CI-lane validator and as a local pre-merge check. Honors
    BROOMVA_P9_PYTEST env var (default: `python3 -m pytest`) so callers can
    pin a specific interpreter or test runner.
    """
    runner = os.environ.get("BROOMVA_P9_PYTEST", f"{sys.executable} -m pytest")
    tests_dir = Path(__file__).resolve().parent.parent / "tests"
    if not tests_dir.exists():
        print(f"p9 conformance: tests directory not found at {tests_dir}",
              file=sys.stderr)
        return EXIT_DEGRADED
    cmd = runner.split() + [str(tests_dir)]
    if args.verbose:
        cmd.append("-v")
    if args.k:
        cmd += ["-k", args.k]
    print(f"p9 conformance: running {' '.join(cmd)}")
    rc = subprocess.run(cmd, check=False).returncode
    if rc == 0:
        print("p9 conformance: ok")
        return EXIT_OK
    print(f"p9 conformance: failed (pytest exit {rc})", file=sys.stderr)
    return EXIT_DEGRADED


def cmd_watch(args: argparse.Namespace) -> int:
    """Watch CI on a PR.

    Default behavior (PR E onwards): foreground — block on
    `gh pr checks --watch`, then fold the subprocess exit code into a state
    transition (WATCHING → GREEN on exit 0, WATCHING → RED_UNCLASSIFIED
    otherwise). Callers (the agent) wrap this in `run_in_background` so the
    bg-task notification fires when the *whole* watch+fold has finished —
    which is what the cardinal protocol actually wants.

    --detach reverts to the old fire-and-forget behavior (no fold; the
    caller is responsible for polling state). --background and --block are
    aliases for the default; they exist so historic AGENTS.md guidance
    using `p9 watch <pr> --background` keeps working.
    """
    policy = load_policy()
    if not policy.ci_watch.enabled:
        print("ci_watch.enabled=false in policy; refusing to watch", file=sys.stderr)
        return EXIT_POLICY_ERROR
    enforce_concurrency_ceiling(policy)
    pr = int(args.pr)
    repo = args.repo or _detect_repo()
    watcher_id = uuid.uuid4().hex[:12]
    proc = spawn_watcher(pr, repo, dry_run=args.dry_run)
    pid = proc.pid if proc else 0
    append_state_event(PRStateEvent(
        ts=_utcnow(),
        pr=pr,
        repo=repo or "",
        from_state=PRState.PUSHED.value,
        to_state=PRState.WATCHING.value,
        watcher_id=watcher_id,
        attempt=0,
        extra={"pid": pid, "dry_run": args.dry_run, "detach": args.detach},
    ))
    if args.json:
        print(json.dumps({
            "watcher_id": watcher_id,
            "pid": pid,
            "pr": pr,
            "repo": repo,
            "mode": "detach" if args.detach else "foreground",
        }))
    else:
        mode = "detach" if args.detach else "foreground"
        print(f"watcher_id={watcher_id} pid={pid} pr={pr} repo={repo} mode={mode}")

    # Detach / dry-run: do NOT block; caller polls state.jsonl.
    if args.detach or args.dry_run or proc is None:
        return EXIT_OK

    # Foreground: block on subprocess, then fold result into a state event.
    rc = proc.wait()
    next_state = PRState.GREEN if rc == 0 else PRState.RED_UNCLASSIFIED
    append_state_event(PRStateEvent(
        ts=_utcnow(),
        pr=pr,
        repo=repo or "",
        from_state=PRState.WATCHING.value,
        to_state=next_state.value,
        watcher_id=watcher_id,
        attempt=0,
        extra={"gh_exit_code": rc, "folded_by": "p9 watch"},
    ))
    if args.json:
        print(json.dumps({"watcher_id": watcher_id, "result": next_state.value, "gh_exit_code": rc}))
    else:
        print(f"folded: {next_state.value} (gh exit {rc})")
    return EXIT_OK


def cmd_status(args: argparse.Namespace) -> int:
    rows = open_prs()
    if args.pr is not None:
        rows = [r for r in rows if r["pr"] == int(args.pr)]
    if args.json:
        print(json.dumps({"open_prs": rows}, indent=2))
    else:
        if not rows:
            print("no PRs in flight")
            return EXIT_OK
        for r in rows:
            print(
                f"#{r['pr']:<5} {r['to_state']:<18} "
                f"watcher={r['watcher_id']} attempt={r.get('attempt', 0)}"
            )
    return EXIT_OK


def cmd_wait_queue(args: argparse.Namespace) -> int:
    sub = args.action
    if sub == "push":
        item = queue_push(
            args.item, args.source, pr=args.pr, isolation_tier=args.tier or "none",
        )
        print(item.id)
        return EXIT_OK
    if sub == "pop":
        head = queue_pop()
        if not head:
            print("(empty)")
            return EXIT_OK
        print(json.dumps(dataclasses.asdict(head)))
        return EXIT_OK
    if sub == "list":
        items = queue_list()
        if args.json:
            print(json.dumps([dataclasses.asdict(it) for it in items], indent=2))
        else:
            for it in items:
                print(f"[{it.source:<7}] {it.id}  {it.item}")
        return EXIT_OK
    if sub == "clear":
        n = queue_clear()
        print(f"cleared {n} item(s)")
        return EXIT_OK
    print(f"unknown wait-queue action: {sub}", file=sys.stderr)
    return EXIT_USAGE


def cmd_heal(args: argparse.Namespace) -> int:
    if not args.classify:
        print("p9 heal currently only supports --classify (read-only)", file=sys.stderr)
        return EXIT_USAGE
    if args.log_file:
        log = Path(args.log_file).read_text(encoding="utf-8")
    elif not sys.stdin.isatty():
        log = sys.stdin.read()
    else:
        # Live mode: pull from gh
        log = _gh_log_failed(args.pr, args.repo)
    result = classify(log)
    print(json.dumps(dataclasses.asdict(result), indent=2))
    return EXIT_OK


def cmd_events_tail(args: argparse.Namespace) -> int:
    rows, _ = jsonl_read_all(state_jsonl())
    if args.since:
        cutoff = _parse_duration_ago(args.since)
        rows = [r for r in rows if r["ts"] >= cutoff]
    for r in rows:
        print(json.dumps(r))
    return EXIT_OK


def cmd_merge_ready(args: argparse.Namespace) -> int:
    pr = int(args.pr)
    state = current_pr_state(pr)
    if state != PRState.GREEN:
        print(
            f"PR #{pr} not GREEN (current={state.value if state else 'UNKNOWN'})",
            file=sys.stderr,
        )
        return EXIT_DEGRADED
    repo = args.repo or _detect_repo()
    event = PRStateEvent(
        ts=_utcnow(),
        pr=pr,
        repo=repo or "",
        from_state=PRState.GREEN.value,
        to_state=PRState.MERGE_READY.value,
        watcher_id="merge-ready",
        attempt=0,
    )
    append_state_event(event)
    print(f"PR #{pr} marked MERGE_READY (control metalayer authorizes merge)")
    return EXIT_OK


# ─────────────────────────────────────────────────────────────────────────────
# Helpers (gh integration, cwd repo detection, time parsing)
# ─────────────────────────────────────────────────────────────────────────────
def _detect_repo() -> str | None:
    try:
        out = subprocess.run(
            ["gh", "repo", "view", "--json", "nameWithOwner", "-q", ".nameWithOwner"],
            capture_output=True, text=True, timeout=10, check=False,
        )
        if out.returncode == 0:
            return out.stdout.strip() or None
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return None
    return None


def _gh_log_failed(pr: int, repo: str | None) -> str:
    cmd = ["gh", "run", "view", "--log-failed"]
    if repo:
        cmd += ["--repo", repo]
    out = subprocess.run(cmd, capture_output=True, text=True, timeout=60, check=False)
    return out.stdout or out.stderr or ""


_DURATION_RE = re.compile(r"^(\d+)([smhd])$")


def _parse_duration_ago(spec: str) -> str:
    m = _DURATION_RE.match(spec)
    if not m:
        return spec  # treat as ISO timestamp passthrough
    n, unit = int(m.group(1)), m.group(2)
    secs = n * {"s": 1, "m": 60, "h": 3600, "d": 86400}[unit]
    return (
        _dt.datetime.now(_dt.timezone.utc) - _dt.timedelta(seconds=secs)
    ).isoformat(timespec="seconds")


# ─────────────────────────────────────────────────────────────────────────────
# CLI dispatch
# ─────────────────────────────────────────────────────────────────────────────
def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="p9",
        description=(
            "Broomva CI watcher + productive-wait primitive. "
            "See docs/superpowers/specs/2026-05-04-p9-ci-watcher-design.md"
        ),
    )
    sub = p.add_subparsers(dest="cmd", required=True)

    pw = sub.add_parser("watch", help="Watch CI on a PR (foreground; folds result into state)")
    pw.add_argument("pr", help="PR number")
    pw.add_argument("--repo", help="OWNER/REPO (auto-detected if omitted)")
    pw.add_argument("--dry-run", action="store_true",
                    help="Do not actually spawn `gh pr checks --watch` (test mode)")
    pw.add_argument("--detach", action="store_true",
                    help="Fire-and-forget: spawn the watcher but do not block "
                         "or fold its exit into a state event. The caller is "
                         "responsible for finalizing state. Default is "
                         "foreground (block + fold).")
    # `--background` and `--block` are aliases for the default foreground
    # behavior. They exist so historic AGENTS.md guidance using
    # `p9 watch <pr> --background` keeps working without surprise errors.
    pw.add_argument("--background", action="store_true",
                    help="Alias for default foreground behavior (kept for "
                         "backwards compatibility with reflexive-rule guidance)")
    pw.add_argument("--block", action="store_true",
                    help="Alias for default foreground behavior")
    pw.add_argument("--json", action="store_true")
    pw.set_defaults(func=cmd_watch)

    ps = sub.add_parser("status", help="Show in-flight PRs")
    ps.add_argument("--pr", help="filter by PR number")
    ps.add_argument("--json", action="store_true")
    ps.set_defaults(func=cmd_status)

    pq = sub.add_parser("wait-queue", help="Manage the productive-wait queue")
    pq.add_argument("action", choices=["push", "pop", "list", "clear"])
    pq.add_argument("--source", default="session",
                    choices=list(_QUEUE_PRIORITY))
    pq.add_argument("--item", default="", help="(push only) free-text item")
    pq.add_argument("--pr", type=int, default=None,
                    help="(push only) tag for this PR")
    pq.add_argument("--tier", default=None,
                    help="(push only) isolation tier")
    pq.add_argument("--json", action="store_true")
    pq.set_defaults(func=cmd_wait_queue)

    ph = sub.add_parser("heal", help="Classify a CI failure (read-only)")
    ph.add_argument("pr", help="PR number")
    ph.add_argument("--repo", default=None)
    ph.add_argument("--classify", action="store_true",
                    help="Required: pure classifier read-out (no heal action)")
    ph.add_argument("--log-file", default=None,
                    help="Read failure log from file instead of `gh run view`")
    ph.set_defaults(func=cmd_heal)

    pe = sub.add_parser("events", help="Stream P9 events")
    pe_sub = pe.add_subparsers(dest="events_cmd", required=True)
    pet = pe_sub.add_parser("tail")
    pet.add_argument("--since", default=None,
                     help="Filter events newer than DURATION (e.g. 6h, 30m)")
    pet.set_defaults(func=cmd_events_tail)

    pm = sub.add_parser("merge-ready", help="Mark PR as ready for metalayer-authorized merge")
    pm.add_argument("pr")
    pm.add_argument("--repo", default=None)
    pm.set_defaults(func=cmd_merge_ready)

    pab = sub.add_parser("abandon",
                         help="Mark a PR as ABANDONED (frees concurrency slot)")
    pab.add_argument("pr")
    pab.add_argument("--repo", default=None)
    pab.add_argument("--reason", default=None,
                     help="Free-text reason recorded in extra.reason")
    pab.set_defaults(func=cmd_abandon)

    pcu = sub.add_parser("cleanup",
                         help="Drain orphan WATCHING/HEALING rows by polling "
                              "GitHub for each open PR's true state")
    pcu.set_defaults(func=cmd_cleanup)

    pa = sub.add_parser("auto-merge",
                        help="Run policy-gated auto-merge on a MERGE_READY PR")
    pa.add_argument("pr")
    pa.add_argument("--repo", default=None)
    pa.add_argument("--dry-run", action="store_true",
                    help="Print the planned merge instead of executing it")
    pa.set_defaults(func=cmd_auto_merge)

    pd = sub.add_parser("doctor", help="Health-check P9 dependencies")
    pd.set_defaults(func=cmd_doctor)

    pc = sub.add_parser("conformance",
                        help="Run the full pytest battery (unit + integration + chaos)")
    pc.add_argument("-v", "--verbose", action="store_true",
                    help="Verbose pytest output (-v)")
    pc.add_argument("-k", default=None,
                    help="Filter expression passed through to pytest -k")
    pc.set_defaults(func=cmd_conformance)

    return p


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        return args.func(args)
    except P9Error as e:
        print(f"p9: {e}", file=sys.stderr)
        return e.code


if __name__ == "__main__":
    sys.exit(main())