diff --git a/docs/audit-panel.md b/docs/audit-panel.md new file mode 100644 index 0000000..1c880dc --- /dev/null +++ b/docs/audit-panel.md @@ -0,0 +1,27 @@ +# Three-auditor panel + fix loop + +**Runtime: droid** (+ the `no-mistakes` CLI). Not Command Code. + +When auditing a list of checks: + +1. **One shared prompt per check** (scope, intent, deliberate decisions, non-goals). +2. **Three seats, concurrent:** buggie · no-mistakes · ponytail ultra + (launch buggie and ponytail via **`droid exec`**; no-mistakes via `no-mistakes axi run`). + Skills may be operator-local; see `auditor-loop/AUDIT_PANEL.md` for the full local protocol. +3. **Synthesize → fix** agreed majors (verify before/after). +4. **Residual re-audit with droid** (same check, residual brief) — not `./auditor-loop/run.sh`. +5. **Repeat triad + droid residual** on the **same** check until no major issues. +6. **Only then** start the next check. Never fan out (N × 3) in one blast. + +```text +for each check: + while majors remain: + droid(buggie) + no-mistakes + droid(ponytail ultra) (wait) + fix + droid residual → fix + next check +``` + +Full protocol: local `auditor-loop/AUDIT_PANEL.md` (often git-excluded). + +**no-mistakes** needs a feature branch with real commits ahead of `main`. Same tip as main → skipped steps, not acceptance. diff --git a/observer_kit/cli.py b/observer_kit/cli.py index cd1664b..85657ed 100644 --- a/observer_kit/cli.py +++ b/observer_kit/cli.py @@ -89,8 +89,10 @@ def load_secret_entries(path: Path) -> list[tuple[str, str]]: if not path.is_file(): _secrets_exit(f"[observer] secrets: file not found: {path}") entries: list[tuple[str, str]] = [] + seen: set[str] = set() try: - text = path.read_text(encoding="utf-8") + # utf-8-sig strips a leading BOM so editor-saved files fail closed cleanly. + text = path.read_text(encoding="utf-8-sig") except OSError as exc: _secrets_exit(f"[observer] secrets: cannot read {path}: {exc}") for lineno, raw in enumerate(text.splitlines(), 1): @@ -124,17 +126,13 @@ def load_secret_entries(path: Path) -> list[tuple[str, str]]: "op://pointer (no $/{}/templates, no plain secrets) so the file " "stays committable and uninterpolated" ) - entries.append((key, value)) - if not entries: - _secrets_exit(f"[observer] secrets: no KEY=op:// entries in {path}") - seen: set[str] = set() - unique: list[tuple[str, str]] = [] - for key, value in entries: if key in seen: continue seen.add(key) - unique.append((key, value)) - return unique + entries.append((key, value)) + if not entries: + _secrets_exit(f"[observer] secrets: no KEY=op:// entries in {path}") + return entries def load_secret_pointers(path: Path) -> list[str]: @@ -142,12 +140,15 @@ def load_secret_pointers(path: Path) -> list[str]: return [key for key, _ in load_secret_entries(path)] -def resolve_op_bin() -> str: +def resolve_op_bin() -> tuple[str, bool]: """Locate the 1Password CLI. Prefer absolute ``OBSERVER_OP_BIN`` (operator-controlled). Otherwise ``PATH`` lookup via ``shutil.which`` — agents can poison PATH, so set OBSERVER_OP_BIN in production. + + Returns ``(absolute_path, from_path)`` where *from_path* is True when the + binary came from PATH rather than OBSERVER_OP_BIN. """ override = os.environ.get("OBSERVER_OP_BIN", "").strip() if override: @@ -161,14 +162,14 @@ def resolve_op_bin() -> str: _secrets_exit( f"[observer] secrets: OBSERVER_OP_BIN not executable: {candidate}" ) - return str(candidate.resolve()) + return str(candidate.resolve()), False found = shutil.which("op") if not found: _secrets_exit( "[observer] secrets: 1Password CLI 'op' not on PATH. " "Install it, set OBSERVER_OP_BIN to an absolute path, or omit --secrets." ) - return found + return found, True def snapshot_secrets_env(entries: list[tuple[str, str]]) -> Path: @@ -191,10 +192,14 @@ def snapshot_secrets_env(entries: list[tuple[str, str]]) -> Path: def command_implies_dry_run(command: list[str]) -> bool: - """True when argv looks like a dry-run / sample (secrets may resolve).""" + """True when argv looks like a dry-run / sample (secrets may resolve). + + Only ``--dry-run`` (and ``--dry-run=…``) count — bare ``-n`` is *not* + treated as dry-run (grep/head use ``-n`` for unrelated meanings). + """ for tok in command: low = str(tok).strip().lower() - if low in {"--dry-run", "-n"}: + if low == "--dry-run": return True if low.startswith("--dry-run="): val = low.split("=", 1)[1] @@ -214,21 +219,41 @@ def _secrets_approval_required() -> bool: return True +def _runguard_bind_state(state_dir: Path): + """Temporarily point runguard's module-level state dir at *state_dir*. + + ``runguard._STATE_DIR`` is set at import time; mutating only + ``os.environ['RUNGUARD_STATE_DIR']`` does not rebind it. + """ + import observer_kit.runguard as runguard + + state_dir = state_dir.expanduser().resolve() + prev_mod = runguard._STATE_DIR + prev_env = os.environ.get("RUNGUARD_STATE_DIR") + runguard._STATE_DIR = str(state_dir) + os.environ["RUNGUARD_STATE_DIR"] = str(state_dir) + return runguard, prev_mod, prev_env + + +def _runguard_restore_state(runguard, prev_mod: str, prev_env: str | None) -> None: + runguard._STATE_DIR = prev_mod + if prev_env is None: + os.environ.pop("RUNGUARD_STATE_DIR", None) + else: + os.environ["RUNGUARD_STATE_DIR"] = prev_env + + def list_pending_full_run_approvals(state_dir: Path) -> list[str]: """Run ids with a valid unacked ``approve_full_run`` under *state_dir*.""" state_dir = state_dir.expanduser().resolve() - prev = os.environ.get("RUNGUARD_STATE_DIR") - os.environ["RUNGUARD_STATE_DIR"] = str(state_dir) + runguard, prev_mod, prev_env = _runguard_bind_state(state_dir) try: - from observer_kit.runguard import pending_full_run_approvals - candidates: set[str] = set() runs = state_dir / "runs" if runs.is_dir(): for lane_dir in runs.iterdir(): if lane_dir.is_dir() and not lane_dir.name.startswith("."): candidates.add(f"runguard:{lane_dir.name}") - # Root / legacy controls may tag a run id without a lane folder yet. root_controls = state_dir / "controls.jsonl" if root_controls.is_file(): try: @@ -251,16 +276,13 @@ def list_pending_full_run_approvals(state_dir: Path) -> list[str]: pending: list[str] = [] for rid in sorted(candidates): try: - if pending_full_run_approvals(rid): + if runguard.pending_full_run_approvals(rid): pending.append(rid) except Exception: continue return pending finally: - if prev is None: - os.environ.pop("RUNGUARD_STATE_DIR", None) - else: - os.environ["RUNGUARD_STATE_DIR"] = prev + _runguard_restore_state(runguard, prev_mod, prev_env) def ensure_secrets_approval( @@ -274,6 +296,9 @@ def ensure_secrets_approval( Dry-run samples may still receive credentials. Full runs (no ``--dry-run``) require a pending ``approve_full_run`` unless ``OBSERVER_ALLOW_UNAPPROVED_FULL_RUN=1``. + + Without ``--run-id``, exactly one pending approval in the state dir is + required (multiple pending forces ``--run-id`` to avoid cross-lane bleed). """ if command_implies_dry_run(command): return @@ -281,39 +306,42 @@ def ensure_secrets_approval( return pending = list_pending_full_run_approvals(state_dir) if run_id: - if run_id in pending or any( - p == run_id or p.endswith(f":{run_id}") for p in pending - ): + if any(p == run_id or p.endswith(f":{run_id}") for p in pending): return - # Precise check for the requested lane even if scan missed it. - prev = os.environ.get("RUNGUARD_STATE_DIR") - os.environ["RUNGUARD_STATE_DIR"] = str(state_dir.expanduser().resolve()) + runguard, prev_mod, prev_env = _runguard_bind_state(state_dir) try: - from observer_kit.runguard import pending_full_run_approvals - - if pending_full_run_approvals(run_id): + if runguard.pending_full_run_approvals(run_id): return finally: - if prev is None: - os.environ.pop("RUNGUARD_STATE_DIR", None) - else: - os.environ["RUNGUARD_STATE_DIR"] = prev + _runguard_restore_state(runguard, prev_mod, prev_env) _secrets_exit( f"[observer] secrets: full-run refused — no approve_full_run for " f"{run_id}. Review a dry-run sample, approve on the dashboard, then " f"retry (exit {_SECRETS_APPROVAL_EXIT}).", _SECRETS_APPROVAL_EXIT, ) - if pending: - return - _secrets_exit( - "[observer] secrets: full-run refused — no pending approve_full_run in " - f"{state_dir}. Run a --dry-run sample first (secrets allowed for samples), " - "approve on the dashboard, then retry the full run " - f"(exit {_SECRETS_APPROVAL_EXIT}). " - "Or pass --run-id runguard:LANE after approval. " - "Opt out: OBSERVER_ALLOW_UNAPPROVED_FULL_RUN=1.", - _SECRETS_APPROVAL_EXIT, + if not pending: + _secrets_exit( + "[observer] secrets: full-run refused — no pending approve_full_run in " + f"{state_dir}. Run a --dry-run sample first (secrets allowed for samples), " + "approve on the dashboard, then retry the full run " + f"(exit {_SECRETS_APPROVAL_EXIT}). " + "Pass --run-id runguard:LANE after approval. " + "Opt out: OBSERVER_ALLOW_UNAPPROVED_FULL_RUN=1.", + _SECRETS_APPROVAL_EXIT, + ) + if len(pending) > 1: + _secrets_exit( + "[observer] secrets: full-run refused — multiple pending " + f"approve_full_run ({', '.join(pending)}). Pass " + f"--run-id runguard:LANE for the lane you approved " + f"(exit {_SECRETS_APPROVAL_EXIT}).", + _SECRETS_APPROVAL_EXIT, + ) + print( + f"[observer] secrets: using pending approval for {pending[0]} " + "(pass --run-id to pin a lane)", + flush=True, ) @@ -324,20 +352,20 @@ def _append_jsonl_record(path: Path, record: dict) -> None: fh.write(line + "\n") -def record_secrets_injection( +def record_secrets_event( state_dir: Path, keys: list[str], secrets_path: Path, *, + event: str, dry_run: bool, run_id: str | None = None, + write_lane: bool = False, ) -> None: - """Durable names-only audit: state-dir secrets_audit.jsonl + optional lane ledger.""" - from observer_kit._util import lane_from_run_id, timestamp as _ts - + """Durable names-only audit into secrets_audit.jsonl (+ optional lane ledger).""" rec = { - "event": "secrets_injected", - "ts": _ts(), + "event": event, + "ts": _timestamp(), "keys": list(keys), "source": str(secrets_path), "dry_run": bool(dry_run), @@ -347,28 +375,46 @@ def record_secrets_injection( rec["run"] = str(run_id) state_dir = state_dir.expanduser().resolve() _append_jsonl_record(state_dir / "secrets_audit.jsonl", rec) - if run_id: - lane = lane_from_run_id(run_id) + if write_lane and run_id: + lane = _lane_from_run_id(run_id) if lane: - _append_jsonl_record( - state_dir / "runs" / lane / "events.jsonl", rec - ) + _append_jsonl_record(state_dir / "runs" / lane / "events.jsonl", rec) + + +def record_secrets_injection( + state_dir: Path, + keys: list[str], + secrets_path: Path, + *, + dry_run: bool, + run_id: str | None = None, +) -> None: + """Compat: record a post-start ``secrets_injected`` event (names only).""" + record_secrets_event( + state_dir, + keys, + secrets_path, + event="secrets_injected", + dry_run=dry_run, + run_id=run_id, + write_lane=bool(run_id), + ) def wrap_command_with_secrets( command: list[str], secrets_path: Path -) -> tuple[list[str], list[str], Path]: +) -> tuple[list[str], list[str], Path, bool]: """Wrap *command* with ``op run --env-file`` so secrets never enter this process. - Returns ``(wrapped_command, key_names, snapshot_path)``. Caller must delete - *snapshot_path* after the child exits. Requires the 1Password CLI - (``OBSERVER_OP_BIN`` or PATH). + Returns ``(wrapped_command, key_names, snapshot_path, op_from_path)``. + Caller must delete *snapshot_path* after the child exits (use try/finally). """ secrets_path = secrets_path.expanduser().resolve() entries = load_secret_entries(secrets_path) keys = [key for key, _ in entries] + # Resolve op first so a missing CLI never leaves an orphaned snapshot. + op_bin, op_from_path = resolve_op_bin() snapshot = snapshot_secrets_env(entries) - op_bin = resolve_op_bin() wrapped = [ op_bin, "run", @@ -377,7 +423,7 @@ def wrap_command_with_secrets( "--", *command, ] - return wrapped, keys, snapshot + return wrapped, keys, snapshot, op_from_path def cmd_init(args: argparse.Namespace) -> int: @@ -1129,99 +1175,126 @@ def pump_watcher() -> None: secrets_arg = getattr(args, "secrets", None) secrets_meta: dict | None = None secrets_snapshot: Path | None = None - if secrets_arg: - secrets_path = Path(secrets_arg).expanduser().resolve() - child_argv = list(command) - dry_sample = command_implies_dry_run(child_argv) - ensure_secrets_approval( - state_dir, - child_argv, - run_id=getattr(args, "run_id", None) or None, - ) - command, secret_keys, secrets_snapshot = wrap_command_with_secrets( - child_argv, secrets_path - ) - # Fail closed: only op may supply declared keys (not a parent export). - for key in secret_keys: - env.pop(key, None) - # Hint for the child only — not proof of resolution (see OBSERVER_OP_BIN). - env["OBSERVER_SECRETS"] = ",".join(secret_keys) - record_secrets_injection( - state_dir, - secret_keys, - secrets_path, - dry_run=dry_sample, - run_id=None, - ) - secrets_meta = { - "keys": secret_keys, - "path": secrets_path, - "dry_run": dry_sample, - "lane_recorded": False, - } - print( - f"[observer] secrets: requesting {', '.join(secret_keys)} via op run " - f"(source={secrets_path})", - flush=True, - ) - - proc = subprocess.Popen( - command, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - env=env, - ) + rc = 1 + try: + if secrets_arg: + secrets_path = Path(secrets_arg).expanduser().resolve() + child_argv = list(command) + dry_sample = command_implies_dry_run(child_argv) + ensure_secrets_approval( + state_dir, + child_argv, + run_id=getattr(args, "run_id", None) or None, + ) + command, secret_keys, secrets_snapshot, op_from_path = ( + wrap_command_with_secrets(child_argv, secrets_path) + ) + if op_from_path: + print( + "[observer] secrets: warning: op resolved from PATH " + f"({command[0]}). Agents can plant a shim — set " + "OBSERVER_OP_BIN to an absolute 1Password CLI path.", + file=sys.stderr, + flush=True, + ) + # Fail closed: only op may supply declared keys (not a parent export). + for key in secret_keys: + env.pop(key, None) + # Hint for the child only — not proof of resolution. + env["OBSERVER_SECRETS"] = ",".join(secret_keys) + # Requested before op runs (not "injected" — that waits for run start). + record_secrets_event( + state_dir, + secret_keys, + secrets_path, + event="secrets_injection_requested", + dry_run=dry_sample, + run_id=None, + write_lane=False, + ) + secrets_meta = { + "keys": secret_keys, + "path": secrets_path, + "dry_run": dry_sample, + "lane_recorded": False, + } + print( + f"[observer] secrets: requesting {', '.join(secret_keys)} via op run " + f"(source={secrets_path})", + flush=True, + ) - def pump_stdout() -> None: - assert proc.stdout is not None - for line in proc.stdout: - sys.stdout.write(line) - sys.stdout.flush() - - def pump_stderr() -> None: - assert proc.stderr is not None - for line in proc.stderr: - if line.startswith("OBSERVER_RUN_STARTED "): - run_id = line.strip().split(maxsplit=1)[1] - run_id_holder["run_id"] = run_id - ensure_watcher(run_id) - if secrets_meta and not secrets_meta.get("lane_recorded"): - record_secrets_injection( - state_dir, - secrets_meta["keys"], - secrets_meta["path"], - dry_run=bool(secrets_meta.get("dry_run")), - run_id=run_id, - ) - secrets_meta["lane_recorded"] = True - sys.stderr.write(line) - sys.stderr.flush() + proc = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=env, + ) - out_thread = threading.Thread(target=pump_stdout) - err_thread = threading.Thread(target=pump_stderr) - out_thread.start() - err_thread.start() + def pump_stdout() -> None: + assert proc.stdout is not None + for line in proc.stdout: + sys.stdout.write(line) + sys.stdout.flush() + + def pump_stderr() -> None: + assert proc.stderr is not None + for line in proc.stderr: + if line.startswith("OBSERVER_RUN_STARTED "): + run_id = line.strip().split(maxsplit=1)[1] + run_id_holder["run_id"] = run_id + ensure_watcher(run_id) + if secrets_meta and not secrets_meta.get("lane_recorded"): + record_secrets_event( + state_dir, + secrets_meta["keys"], + secrets_meta["path"], + event="secrets_injected", + dry_run=bool(secrets_meta.get("dry_run")), + run_id=run_id, + write_lane=True, + ) + secrets_meta["lane_recorded"] = True + sys.stderr.write(line) + sys.stderr.flush() - try: - rc = proc.wait() - except KeyboardInterrupt: - proc.terminate() - rc = 130 - out_thread.join() - err_thread.join() + out_thread = threading.Thread(target=pump_stdout) + err_thread = threading.Thread(target=pump_stderr) + out_thread.start() + err_thread.start() - if secrets_snapshot is not None: try: - secrets_snapshot.unlink(missing_ok=True) - except OSError: - pass - if secrets_meta is not None and rc == 0: - print( - f"[observer] secrets: child exited 0 after op run " - f"({', '.join(secrets_meta['keys'])})", - flush=True, - ) + rc = proc.wait() + except KeyboardInterrupt: + proc.terminate() + rc = 130 + out_thread.join() + err_thread.join() + + if secrets_meta is not None: + if rc == 0: + print( + f"[observer] secrets: child exited 0 after op run " + f"({', '.join(secrets_meta['keys'])})", + flush=True, + ) + else: + record_secrets_event( + state_dir, + secrets_meta["keys"], + secrets_meta["path"], + event="secrets_injection_failed", + dry_run=bool(secrets_meta.get("dry_run")), + run_id=run_id_holder.get("run_id"), + write_lane=False, + ) + finally: + if secrets_snapshot is not None: + try: + secrets_snapshot.unlink(missing_ok=True) + except OSError: + pass if run_id_holder["run_id"] is None and args.watch != "none": print("[observer] no OBSERVER_RUN_STARTED marker was seen; watcher was not started", file=sys.stderr) diff --git a/tests/test_secrets.py b/tests/test_secrets.py index 2bfa894..e4f5e18 100644 --- a/tests/test_secrets.py +++ b/tests/test_secrets.py @@ -158,9 +158,49 @@ def expect_exit(fn, *args, **kwargs) -> tuple[int, str]: command_implies_dry_run(["python3", "w.py", "--dry-run", "--limit", "10"]) and not command_implies_dry_run(["python3", "w.py", "--full-run"]), ) + ok( + "command_implies_dry_run ignores bare -n (grep/head flag, not dry-run)", + not command_implies_dry_run(["grep", "-n", "pattern", "file.py"]), + ) + + # op missing must not leave observer-secrets-*.env in TMPDIR + import tempfile as _tf + + empty_bin2 = root / "empty-bin-snap" + empty_bin2.mkdir(exist_ok=True) + saved_path2 = os.environ.get("PATH") + saved_op2 = os.environ.pop("OBSERVER_OP_BIN", None) + os.environ["PATH"] = str(empty_bin2) + before_snaps = set(Path(_tf.gettempdir()).glob("observer-secrets-*.env")) + try: + expect_exit(wrap_command_with_secrets, ["python3", "w.py"], good) + finally: + if saved_path2 is None: + os.environ.pop("PATH", None) + else: + os.environ["PATH"] = saved_path2 + if saved_op2 is not None: + os.environ["OBSERVER_OP_BIN"] = saved_op2 + after_snaps = set(Path(_tf.gettempdir()).glob("observer-secrets-*.env")) + ok( + "wrap_command_with_secrets leaves no orphan snapshot when op missing", + after_snaps <= before_snaps, + f"new files: {after_snaps - before_snaps}", + ) audit_state = root / "audit-state" audit_state.mkdir() + from observer_kit.cli import record_secrets_event + + record_secrets_event( + audit_state, + ["HUBSPOT_TOKEN"], + good, + event="secrets_injection_requested", + dry_run=True, + run_id=None, + write_lane=False, + ) record_secrets_injection( audit_state, ["HUBSPOT_TOKEN"], @@ -172,17 +212,18 @@ def expect_exit(fn, *args, **kwargs) -> tuple[int, str]: lane_lines = ( audit_state / "runs" / "demo-lane" / "events.jsonl" ).read_text(encoding="utf-8").splitlines() - audit_rec = json.loads(audit_lines[0]) + req_rec = json.loads(audit_lines[0]) + inj_rec = json.loads(audit_lines[1]) lane_rec = json.loads(lane_lines[0]) ok( - "record_secrets_injection writes names-only audit + lane ledger events", - audit_rec.get("event") == "secrets_injected" - and audit_rec.get("keys") == ["HUBSPOT_TOKEN"] + "audit distinguishes requested vs injected; lane only for injected", + req_rec.get("event") == "secrets_injection_requested" + and inj_rec.get("event") == "secrets_injected" + and inj_rec.get("keys") == ["HUBSPOT_TOKEN"] and "sk-" not in audit_lines[0] - and "op://" not in json.dumps(audit_rec.get("keys")) and lane_rec.get("run") == "runguard:demo-lane" - and lane_rec.get("keys") == ["HUBSPOT_TOKEN"], - audit_lines[0] + " | " + lane_lines[0], + and lane_rec.get("event") == "secrets_injected", + str(audit_lines) + " | " + lane_lines[0], ) # wrap without op on PATH @@ -234,7 +275,9 @@ def expect_exit(fn, *args, **kwargs) -> tuple[int, str]: os.environ["PATH"] = str(fake_bin) + os.pathsep + (saved_path or os.defpath) os.environ["OBSERVER_OP_BIN"] = str(fake_op.resolve()) try: - wrapped, wrap_keys, snap = wrap_command_with_secrets(["echo", "hi"], good) + wrapped, wrap_keys, snap, from_path = wrap_command_with_secrets( + ["echo", "hi"], good + ) ok( "wrap_command_with_secrets prefixes op run --env-file snapshot --", wrap_keys == ["HUBSPOT_TOKEN", "CLAY_API_KEY"] @@ -244,6 +287,7 @@ def expect_exit(fn, *args, **kwargs) -> tuple[int, str]: and wrapped[4] == "--" and wrapped[5:] == ["echo", "hi"] and snap.is_file() + and from_path is False # OBSERVER_OP_BIN set and "op://vault/hubspot/credential" in snap.read_text(encoding="utf-8"), str(wrapped), ) @@ -396,13 +440,14 @@ def expect_exit(fn, *args, **kwargs) -> tuple[int, str]: ) audit = state / "secrets_audit.jsonl" + audit_text = audit.read_text(encoding="utf-8") if audit.is_file() else "" ok( - "run --secrets writes durable secrets_audit.jsonl (names only)", + "run --secrets writes secrets_injection_requested (names only, pre-op)", audit.is_file() - and "secrets_injected" in audit.read_text(encoding="utf-8") - and "HUBSPOT_TOKEN" in audit.read_text(encoding="utf-8") - and "test-token-from-op" not in audit.read_text(encoding="utf-8"), - audit.read_text(encoding="utf-8") if audit.is_file() else "missing", + and "secrets_injection_requested" in audit_text + and "HUBSPOT_TOKEN" in audit_text + and "test-token-from-op" not in audit_text, + audit_text or "missing", ) # Full-run without approval must not materialize credentials (exit 4). @@ -481,30 +526,46 @@ def expect_exit(fn, *args, **kwargs) -> tuple[int, str]: f"rc={proc5.returncode} {out5[-600:]}", ) - # Unit: ensure_secrets_approval with a posted control. - from observer_kit.runguard import post_control + # Unit: ensure_secrets_approval with a posted control (bind runguard state). + import observer_kit.runguard as rg + prev_mod = rg._STATE_DIR + rg._STATE_DIR = str(state) os.environ["RUNGUARD_STATE_DIR"] = str(state) try: - post_control("runguard:approved-lane", "approve_full_run", note="ok") - finally: - pass - try: - ensure_secrets_approval( + rg.post_control("runguard:approved-lane", "approve_full_run", note="ok") + try: + ensure_secrets_approval( + state, + ["python3", "w.py", "--full-run"], + run_id="runguard:approved-lane", + ) + approved_ok = True + approved_msg = "" + except SystemExit as exc: + approved_ok = False + approved_msg = str(exc) + ok( + "ensure_secrets_approval allows full-run when approve_full_run is pending", + approved_ok, + approved_msg, + ) + + # Multiple pending approvals without --run-id must fail closed. + rg.post_control("runguard:lane-a", "approve_full_run", note="a") + rg.post_control("runguard:lane-b", "approve_full_run", note="b") + multi_code, multi_msg = expect_exit( + ensure_secrets_approval, state, ["python3", "w.py", "--full-run"], - run_id="runguard:approved-lane", ) - approved_ok = True - approved_msg = "" - except SystemExit as exc: - approved_ok = False - approved_msg = str(exc) - ok( - "ensure_secrets_approval allows full-run when approve_full_run is pending", - approved_ok, - approved_msg, - ) + ok( + "ensure_secrets_approval refuses multiple pending without --run-id", + multi_code == 4 and "multiple pending" in multi_msg, + f"code={multi_code} msg={multi_msg}", + ) + finally: + rg._STATE_DIR = prev_mod finally: os.environ.pop("OBSERVER_OP_BIN", None) if saved_path is None: