#!/usr/bin/env python """PRD Plugin autonomous run-until-done Stop hook for Claude Code. Wired to Stop in .claude/settings.json. When the configured autonomy tier is `autonomous`, this keeps the agent working toward the active goal instead of stopping for pointless checkpoints — the deterministic, plugin-managed cousin of the native /goal command (which is per-session and judged from the transcript; this reads real state). It blocks the stop (returns `{"decision":"block","reason":...}`) only when ALL of: - automation.autonomy_level == "autonomous", and - automation.autonomous_run_until_done is true (default true), and - there is an active goal: a TRK-* tracking record whose status is not closed, and - THIS session owns that goal (see below), and - no pause has been requested (.prd_plugin/local/autonomy-pause absent), and - the per-session continue count is under automation.autonomous_continue_cap. Multi-session safety (REQ-053, AI-Collab-v3 HLT-016): multi-agent repos spawn headless `claude -p` child sessions that inherit this hook. They must never be conscripted into a goal they don't own. Three layers: 0. Cross-host exclusion (REQ-105, ai-collab-v3 REQ-090): a goal whose owner_agent names a different host runtime (claude/codex/opencode, via the PRD_HOOK_HOST env the dispatcher exports from --host) is never a candidate — other hosts' sessions don't share this state file, so the first-blocked-session ownership below cannot see their claims. 0b. Staleness exclusion (REQ-114): an open goal not updated within automation.stop_guard_goal_max_age_days (default 2) is only a candidate for the session that already owns it. Run-until-done continues the work a session was last doing; abandoned goals belong to the staleness audit. 1. Hard opt-out for spawners: set PRD_STOP_GUARD=off (or 0/false/disabled) or PRD_WORKER_SESSION=1 in the child environment and the guard allows the stop unconditionally, whatever the goal state. 2. Session ownership by default: the first session the guard blocks toward a goal becomes that goal's owner (persisted in .prd_plugin/local/stop-guard-state.json, keyed by goal id). Only the owning session_id is ever blocked; every other session stops freely. The binding is pruned when the goal closes (complete/resolved/deferred/parked/ superseded/...) or a different goal becomes active. Otherwise it allows the stop. It NEVER forces continuation past the consent floor: when the agent genuinely needs the user (push/merge/publish/secrets, or a real blocker), it writes `.prd_plugin/local/autonomy-pause` and the guard lets it stop (then clears the marker). Any error allows the stop — it must never trap the session in a loop. """ import datetime import json import os import sys from pathlib import Path CLOSED_STATUSES = {"complete", "completed", "done", "closed", "cancelled", "canceled", "superseded", "abandoned", "resolved", # deferred/parked are explicit owner decisions to stop — # the guard must never demand work on a parked goal "deferred", "parked", "finished"} GOAL_TYPES = {"goal", "active_work"} DEFAULT_CAP = 25 # Freshness window (REQ-114): an open goal is only a run-until-done candidate # if it was updated within this many days — unless THIS session owns it. Stale # open goals are the staleness audit's territory; they never conscript. DEFAULT_GOAL_MAX_AGE_DAYS = 2 PAUSE_REL = ".prd_plugin/local/autonomy-pause" OPT_OUT_VALUES = {"off", "0", "false", "disable", "disabled"} # Host runtimes whose sessions never share a stop-guard state file. A goal whose # owner_agent names exactly one of these is bound to that runtime (REQ-105). KNOWN_HOSTS = ("claude", "codex", "opencode") def _read_json(path): return json.loads(Path(path).read_text(encoding="utf-8-sig")) def _automation(root): cfg = _read_json(Path(root) / ".prd_plugin" / "config.json") return cfg.get("automation", {}) if isinstance(cfg, dict) else {} def opted_out(): """Spawner contract: worker sessions are never conscripted.""" if os.environ.get("PRD_STOP_GUARD", "").strip().lower() in OPT_OUT_VALUES: return True if os.environ.get("PRD_WORKER_SESSION", "").strip().lower() in {"1", "true", "yes"}: return True return False def _owner_host(owner_agent): """The host runtime an owner_agent names, or None when it names none or is ambiguous. 'AGENT-CODEX'/'Codex' -> 'codex'; 'AGENT-001'/'AGENT-MCP' -> None.""" text = str(owner_agent or "").lower() matches = [h for h in KNOWN_HOSTS if h in text] return matches[0] if len(matches) == 1 else None def _is_fresh(rec, today, max_age_days): """A goal is fresh when its updated_at (or created_at) is within the window. Undated or unparseable records stay candidates (legacy shape; real records are always tool-stamped).""" stamp = rec.get("updated_at") or rec.get("created_at") if not stamp: return True try: when = datetime.date.fromisoformat(str(stamp)[:10]) except ValueError: return True return (today - when).days <= max_age_days def active_goal(root, host=None, owned=None, today=None, max_age_days=None): """Most-recently-updated candidate open TRK-* goal/active_work id, or None. Preferring the newest open goal avoids latching onto a stale record that an earlier session forgot to close. Two exclusions shape the candidate set: - Cross-host (REQ-105, ai-collab-v3 REQ-090): when the current host runtime is known, a goal whose owner_agent names a DIFFERENT host runtime is never a candidate — its sessions run under another host whose ownership registry this one never sees. Hostless owners stay candidates. - Staleness (REQ-114): when a freshness window is given (max_age_days not None), a goal not updated within it is only a candidate if THIS session already owns it (`owned`) — run-until-done continues the work a session was last doing, it does not adopt abandoned goals. """ data = _read_json(Path(root) / ".prd_plugin" / "state" / "tracking.json") open_goals = [ rec for rec in data.get("records", []) if isinstance(rec, dict) and rec.get("type") in GOAL_TYPES and str(rec.get("status", "")).lower() not in CLOSED_STATUSES ] if host: open_goals = [rec for rec in open_goals if _owner_host(rec.get("owner_agent")) in (None, host)] if max_age_days is not None: today = today or datetime.date.today() owned = owned or set() open_goals = [rec for rec in open_goals if rec.get("id") in owned or _is_fresh(rec, today, max_age_days)] if not open_goals: return None open_goals.sort(key=lambda r: (str(r.get("updated_at") or ""), str(r.get("created_at") or ""), str(r.get("id") or ""))) return open_goals[-1].get("id") def is_paused(root): return (Path(root) / PAUSE_REL).is_file() def clear_pause(root): try: (Path(root) / PAUSE_REL).unlink() except OSError: pass def _stop(reason): return {"action": "stop", "reason": reason, "goal": None} def decide(root, continue_count, session_id=None, owners=None, host=None, today=None): """Pure decision: continue or stop. Any failure resolves to stop.""" if opted_out(): return _stop("PRD_STOP_GUARD opt-out set — worker session, allowing stop") try: auto = _automation(root) except Exception: return _stop("could not read config") # Which tiers keep working instead of stopping. # # `autonomous` is run-until-done. `key_decision` is NOT run-until-done: it # still stops for key decisions and the consent floor — but "may I continue # the planned work?" was never a key decision, and the guard returning # "tier is not autonomous" let every such stop through (REQ-150). Both # tiers share one escape: declare the real need in the pause marker. tier = auto.get("autonomy_level") if tier == "autonomous": if not auto.get("autonomous_run_until_done", True): return _stop("autonomous_run_until_done is off") elif tier == "key_decision": if not auto.get("key_decision_continue_guard", True): return _stop("key_decision_continue_guard is off") else: return _stop(f"autonomy tier {tier!r} always allows a stop") if is_paused(root): return _stop("autonomy-pause requested — allowing stop for the consent floor / a blocker") try: max_age = int(auto.get("stop_guard_goal_max_age_days", DEFAULT_GOAL_MAX_AGE_DAYS)) except (TypeError, ValueError): max_age = DEFAULT_GOAL_MAX_AGE_DAYS owned = {g for g, s in (owners or {}).items() if s == session_id} try: goal = active_goal(root, host=host, owned=owned, today=today, max_age_days=max_age) if not goal: if active_goal(root, host=host, owned=owned): return _stop("every open goal is stale — run-until-done only " "continues a session's current work; stale goals " "are the staleness audit's job. Allowing stop") if host and active_goal(root): return _stop("every open goal is owned by another host agent — " "this session is free to stop") except Exception: return _stop("could not read tracking state") if not goal: return _stop("no active goal — nothing left to continue") # Session ownership: only the session bound to this goal may be blocked. # Everyone else (headless workers, advisor consults) stops freely. if session_id is not None and owners: owner = owners.get(goal) if owner and owner != session_id: return _stop(f"goal {goal} is owned by another session — this session is free to stop") try: cap = int(auto.get("autonomous_continue_cap", DEFAULT_CAP)) except (TypeError, ValueError): cap = DEFAULT_CAP if continue_count >= cap: return _stop(f"continue cap reached ({cap}) — allowing stop") if tier == "key_decision": reason = ( f"Goal {goal} is still active, and continuing planned work is NOT a decision. " "Do not stop to ask 'shall I proceed?', 'which would you prefer?', or anything " "the method, the plan, or the repo already answers — resolve it yourself first " "(capability discovery, the wiki, canonical state, then the code). " "In key_decision you DO stop for a genuine key decision (architecture or scope " "choice, anything irreversible, high severity or risk) and for the consent floor " "(push / merge to main / publish, secrets, another repo). When that is what you " "need, write the question into .prd_plugin/local/autonomy-pause and stop — the " "guard will let you through. Otherwise continue the work." ) else: reason = ( f"Autonomous run-until-done: goal {goal} is still active. Continue with the " "next incomplete step toward it — do not stop to ask 'shall I proceed?'. " "If you genuinely need the user (consent floor: push/merge/publish/secrets, " "or a real blocker), create the file .prd_plugin/local/autonomy-pause with a " "one-line reason and then stop; the guard will let you. Otherwise keep going." ) return {"action": "continue", "reason": reason, "goal": goal} # --- per-session state (runtime, gitignored) ------------------------------- # Schema: {"counters": {session_id: int}, "owners": {goal_id: session_id}}. # Legacy flat {session_id: int} files are migrated on read. def _state_path(root): return Path(root) / ".prd_plugin" / "local" / "stop-guard-state.json" def _load_state(root): p = _state_path(root) data = {} if p.is_file(): try: data = _read_json(p) except Exception: data = {} if not isinstance(data, dict): data = {} if "counters" not in data and "owners" not in data: # legacy shape: flat {session: count} counters = {k: v for k, v in data.items() if isinstance(v, int)} data = {"counters": counters, "owners": {}} data.setdefault("counters", {}) data.setdefault("owners", {}) if not isinstance(data["counters"], dict): data["counters"] = {} if not isinstance(data["owners"], dict): data["owners"] = {} return data def _save_state(root, data): p = _state_path(root) try: p.parent.mkdir(parents=True, exist_ok=True) tmp = p.with_suffix(p.suffix + ".tmp") tmp.write_text(json.dumps(data), encoding="utf-8") os.replace(tmp, p) except OSError: pass def _prune_owners(owners, current_goal): """Drop bindings for goals that are no longer the active goal (closed, superseded, or replaced). One goal is active at a time by design.""" return {g: s for g, s in owners.items() if g == current_goal} def main(): try: raw = sys.stdin.read() payload = json.loads(raw) if raw.strip() else {} except Exception: payload = {} if not isinstance(payload, dict): payload = {} root = payload.get("cwd") or str(Path.cwd()) session = payload.get("session_id") or "default" # Host runtime identity, exported by prd_hook_dispatch.py (--host). Absent # under direct/legacy wiring, which keeps the pre-REQ-105 behavior. host = os.environ.get("PRD_HOOK_HOST", "").strip().lower() or None state = _load_state(root) counters = state["counters"] owners = state["owners"] # Cumulative per-session count of forced continuations; reset only on an # allow-stop (counting only while stop_hook_active was true let the cap # reset every natural stop and never bite). count = int(counters.get(session, 0)) d = decide(root, count, session_id=session, owners=owners, host=host, today=None) if d["action"] == "continue": counters[session] = count + 1 owners.setdefault(d["goal"], session) # first blocked session owns the goal state["owners"] = _prune_owners(owners, d["goal"]) _save_state(root, state) print(json.dumps({"decision": "block", "reason": d["reason"]})) return 0 # allow stop: reset this session's counter, prune stale ownership bindings, # and clear any pause marker changed = False if session in counters: counters.pop(session, None) changed = True try: current = active_goal(root, host=host) except Exception: current = None pruned = _prune_owners(owners, current) if pruned != owners: state["owners"] = pruned changed = True if changed: _save_state(root, state) if is_paused(root): clear_pause(root) # silent allow (no output) — nothing to inject return 0 if __name__ == "__main__": main() sys.exit(0)