fix(p9): watcher folds gh exit into state; +abandon, +cleanup, --background alias

scripts/p9.py

@@ -160,11 +160,16 @@ class PRState(str, enum.Enum):
     (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
+    # 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),
 }
 
 
@@ -524,9 +529,13 @@ def _scalar(s: str) -> Any:
     return s
 
 
-def load_policy(path: Path | None = None) -> PolicyConfig:
-    """Load .control/policy.yaml. **Fail-closed** on missing/malformed blocks."""
-    p = path or policy_yaml_path()
+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:
@@ -1061,6 +1070,87 @@ def cmd_doctor(_args: argparse.Namespace) -> int:
     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.
@@ -1195,6 +1285,20 @@ def cmd_conformance(args: argparse.Namespace) -> int:
 
 
 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)
@@ -1205,26 +1309,49 @@ def cmd_watch(args: argparse.Namespace) -> int:
     watcher_id = uuid.uuid4().hex[:12]
     proc = spawn_watcher(pr, repo, dry_run=args.dry_run)
     pid = proc.pid if proc else 0
-    event = PRStateEvent(
+    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},
-    )
-    append_state_event(event)
+        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:
-        print(f"watcher_id={watcher_id} pid={pid} pr={pr} repo={repo}")
+        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
 
 
@@ -1378,11 +1505,24 @@ def build_parser() -> argparse.ArgumentParser:
     )
     sub = p.add_subparsers(dest="cmd", required=True)
 
-    pw = sub.add_parser("watch", help="Start a CI watcher for a PR")
+    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)
 
@@ -1424,6 +1564,19 @@ def build_parser() -> argparse.ArgumentParser:
     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")