diff --git a/CHANGELOG.md b/CHANGELOG.md index be51e33..ad94114 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,59 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). `agent-ultra bob hook-install` to regenerate the hooks.** ### Added +- **bob surgical lane** (`agent-ultra bob surgical "" --files ...`, + `run_surgical` from Python) — the sanctioned lightweight tier for INERT + doc/config edits: a real two-lens review fan-out (ultracode-backed, + fingerprint-pinned like steps 6/7) + an operator quiz (`passed` requires + the captured answer) + its own fail-closed gate. Entry criteria are + objective and validated at start: inert types only (prose + known-inert + dotfiles; structured config needs the operator's explicit + `AGENT_ULTRA_SURGICAL_ALLOW_CONFIG=1`), a deny list of + executable-by-context files (CI workflows, hook managers, task runners, + package manifests, tool configs — by name, path, and shape), declared + files as the exact scope, and a 100-line/256-KiB text-only diff budget. + **Surgical narrows Bob, it never replaces it (C1):** `run.json`'s mode is + a request the gate verifies — the surgical gate is used only when the + actual staged set independently qualifies; code staged under a + "surgical" run routes to the FULL gate and its full receipt-chain + demands, and a full-mode run never auto-downgrades. + +- **bob run lifecycle + fan-out integrity hardening.** A run can no longer + be dropped quietly, and fan-out proof got stricter: + - *No silent orphaning.* `BobRun.start(force=True)` (CLI: + `bob run --operator-force`, renamed from `--force`) now leaves a + durable abandonment record instead of silently superseding the + unproven run. New `bob abandon --operator-abandon [--reason]` releases + an unproven run explicitly — refused without the flag — logging to + `.agent-ultra/bob/abandoned.jsonl` plus `abandoned.json` in the run + dir, and it announces loudly on stderr. + - *Operator escapes are agent-unreachable.* The command broker + classifies `--operator-abandon` / `--operator-force` / + `--operator-unbounded` as DANGEROUS, so a broker-mediated agent shell + cannot auto-run them. + - *Cloned fan-out rejected.* A step-6/7 ultracode run whose completed + agents all produced IDENTICAL output (per-agent `output_sha256` in the + run receipt) is one review wearing N hats — the gate now blocks it. + Demo mode (`--allow-mock`) is exempt: the offline mock answers every + agent identically by design, and mock evidence is already only + acceptable there. + - **Breaking:** `bob run --force` is now `bob run --operator-force`. + +- **bob run scope — a run is bound to what it declared.** Every bob run now + declares its target scope at start (`BobRun.start(..., scope=[...])`; + `run_bob` declares the generated test+impl files automatically) and the + gate rejects staged files outside it, so a run opened for module A cannot + smuggle in edits to module B. Scope is mandatory and fail-closed: a + no-scope start refuses, and a scope stripped from `run.json` afterwards + authorizes nothing. Expansion goes only through the validated, logged + mechanism (`agent-ultra bob scope-add`, recorded in `scope-log.jsonl`). + `.git/hooks` and `.agent-ultra` paths are rejected as scope entries + always — a run cannot declare authority over its own enforcement. The + only unbounded form is `operator_unbounded=True`, a Python-API parameter + deliberately absent from the CLI (an agent driving the CLI cannot reach + it), loudly announced and durably logged. **Breaking:** `BobRun.start` + now requires `scope=`; `run_bob` callers are unaffected. + - **bob** (`agent_ultra.bob`) — the 10-step enforced build pipeline (SPEC → RED → GREEN → REFACTOR → CODE-QUALITY → SECURITY-FANOUT → WORKFLOW → ULTRA → QUIZ → COMMIT) composing ultracode fan-out, the adversarial diff --git a/docs/bob-the-builder.md b/docs/bob-the-builder.md index c05ad50..718689b 100644 --- a/docs/bob-the-builder.md +++ b/docs/bob-the-builder.md @@ -83,6 +83,45 @@ artifact, unverifiable chain) is a failure, not a warning. On block, the errors name exactly which step to redo. Staged files that appear in no receipt also block (`*.md`, `.agent-ultra/*`, and `.gitignore` are exempt). +### Run scope: a run is FOR something + +Every run declares its target scope at start — the files/dirs/globs the run +is for. `run_bob` declares the generated test + implementation files +automatically; from Python pass `scope=[...]` to `BobRun.start`. The gate +then rejects any staged file **outside** the declared scope, so a run opened +for module A cannot smuggle in edits to module B: + +``` +out-of-scope: staged file other/thing.py is outside the run's declared scope +``` + +Rules: + +- **Scope is mandatory.** `BobRun.start` refuses a run with no scope, and a + scope stripped from `run.json` after the fact unbinds nothing — a no-scope + run authorizes NO staged files (fail closed). +- **Expansion is explicit, validated, and logged.** If the work legitimately + grows, `agent-ultra bob scope-add ` appends to the scope and + records the expansion in `scope-log.jsonl`. Nothing else widens a run. +- **Infrastructure is never in scope.** `.git/hooks` and `.agent-ultra` + paths are rejected at start and at scope-add, always — a run cannot + declare authority over the machinery that enforces it. +- **A run cannot be dropped by starting another.** `start` refuses while + an unproven run is ACTIVE, and a marker naming a deleted run directory is + a tampering sentinel (the gate fails closed), not "no run". The only + sanctioned drop is `agent-ultra bob abandon --operator-abandon + [--reason]` — durably recorded in `.agent-ultra/bob/abandoned.jsonl`. + The command broker classifies all `--operator-*` escape flags as + DANGEROUS, so a broker-mediated agent shell cannot auto-run them. +- **A cloned fan-out is one review.** If every completed agent of a + step-6/7 ultracode run produced identical output, the gate blocks — the + per-agent `output_sha256` in the run receipt exposes the clone. +- **The operator escape is not a CLI flag.** `BobRun.start(..., + operator_unbounded=True)` opens a run with no scope binding; it is a + Python-API parameter on purpose (no CLI verb exposes it, so an agent + driving the CLI cannot reach it), it is announced loudly on stderr, and + it is durably recorded in `scope-log.jsonl`. + ### Commit lifecycle: mark at pre-commit, seal at post-commit Sealing inside pre-commit is a deadlock: the pass would release ACTIVE @@ -127,6 +166,48 @@ which reads git's staged list. Install both: `assert_bob_done` guards completion claims, the pre-commit hook guards the commit itself and catches any staged file that no receipt covers. +## The surgical lane (inert edits only) + +Typo-class doc/config edits do not need RED/GREEN and a five-agent sweep. +The surgical lane is the sanctioned lightweight tier — and it NARROWS Bob, +it never replaces it: + +```bash +# edit the files first, then: +agent-ultra bob surgical "fix readme wording" --files README.md +agent-ultra bob surgical "fix readme wording" --files README.md --mock # offline demo +``` + +The lane runs a real two-lens review fan-out (accuracy + safety, ultracode- +backed and fingerprint-pinned exactly like steps 6/7), an operator quiz +(`passed` requires your captured answer; an unanswered quiz is honestly +`skipped`), and its own gate. + +Entry criteria are objective and validated at start — anything else raises +and the answer is the full pipeline: + +- **Inert types only.** Prose (`.md`/`.rst`/`.txt`/...) and known-inert + dotfiles auto-qualify. Structured config (yaml/json/toml/ini/cfg) is + dual-use: it needs the operator's explicit + `AGENT_ULTRA_SURGICAL_ALLOW_CONFIG=1` AND must clear the denylist. +- **The denylist is executable-by-context files** — CI workflows, git-hook + managers, task runners, package manifests, tool configs (matched by name, + path, and shape) — regardless of extension. A doc ABOUT Dockerfiles is a + doc; a `Dockerfile` is not. +- **Declared files are the exact scope.** The gate rejects any staged file + that was not declared, is not inert, or is denied; `scope-add` on a + surgical run refuses risky paths. +- **A small diff budget** (100 lines / 256 KiB, binary edits rejected), + fail-closed on any git error. + +**Mode downgrade cannot launder code (C1).** `run.json`'s mode field is a +REQUEST the gate verifies, never a trusted assertion: `bob gate` routes to +the surgical gate ONLY when the actual staged set independently qualifies +(all inert, none denied, all declared). Flip a run's mode to "surgical" +with code staged and you get the FULL gate — which demands the full receipt +chain that run never produced. A full-mode run never auto-downgrades +either, even when only inert files are staged: surgical is opt-in at start. + ## Trust boundary (read this) bob resists the failure modes an agent actually falls into: **skipping** the diff --git a/src/agent_ultra/adapters/cli.py b/src/agent_ultra/adapters/cli.py index 81d626f..30dee5d 100644 --- a/src/agent_ultra/adapters/cli.py +++ b/src/agent_ultra/adapters/cli.py @@ -298,7 +298,7 @@ def _bob_run(args) -> int: pool = _build_pool(args) outcome = run_bob(args.task, args.workspace, mock=args.mock, pool=pool, interactive=False if args.no_quiz else None, - force=args.force) + force=args.operator_force) return 0 if outcome.passed else 1 @@ -351,6 +351,53 @@ def _bob_seal(args) -> int: return 1 +def _bob_scope_add(args) -> int: + from ..bob import BobRun + run = BobRun.active(args.workspace) + if run is None or run.marker_orphaned(): + print("no active bob run — nothing to expand", file=sys.stderr) + return 2 + added, errors = run.add_scope(args.paths) + for e in errors: + print(f"SCOPE REJECTED — {e}", file=sys.stderr) + if added: + print(f"scope expanded (logged): +{added}") + print(f"scope now: {run.scope()}") + return 1 if errors else 0 + + +def _bob_surgical(args) -> int: + from ..bob import run_surgical + pool = None + if not args.mock: + pool = _build_pool(args) + outcome = run_surgical(args.task, args.workspace, args.files, + mock=args.mock, pool=pool, + interactive=not args.no_quiz) + return 0 if outcome.passed else 1 + + +def _bob_abandon(args) -> int: + from ..bob import BobRun + if not args.operator_abandon: + print("ABANDON REFUSED — requires the explicit --operator-abandon " + "flag (operator only; an agent cannot abandon an unproven " + "run)", file=sys.stderr) + return 1 + try: + name = BobRun.abandon(args.workspace, operator=True, + reason=args.reason) + except ValueError as e: + print(f"ABANDON REFUSED — {e}", file=sys.stderr) + return 1 + if name is None: + print("no active bob run — nothing to abandon") + return 0 + print(f"run {name} abandoned (logged to .agent-ultra/bob/" + "abandoned.jsonl)") + return 0 + + def _bob_hooks(args) -> int: from ..bob.pipeline import install_hooks try: @@ -548,8 +595,9 @@ def _uc_common(sp): br.add_argument("--workspace", default=".") br.add_argument("--no-quiz", action="store_true", help="record the quiz step as skipped (non-interactive)") - br.add_argument("--force", action="store_true", - help="abandon an existing unproven active run and start fresh") + br.add_argument("--operator-force", action="store_true", + help="OPERATOR ONLY: abandon an existing unproven " + "active run (durably logged) and start fresh") br.add_argument("--mock", action="store_true", help="offline demo on the bundled sample task (no key)") br.add_argument("--mode", default="single", choices=["single", "mixed"]) @@ -596,6 +644,46 @@ def _uc_common(sp): bs.add_argument("--workspace", default=".") bs.set_defaults(func=_bob_status) + ba = bsub.add_parser("scope-add", + help="expand the active run's declared scope — " + "the approved, validated, logged mechanism " + "(infrastructure paths always rejected)") + ba.add_argument("paths", nargs="+", + help="files/dirs/globs to add to the run's scope") + ba.add_argument("--workspace", default=".") + ba.set_defaults(func=_bob_scope_add) + + bab = bsub.add_parser("abandon", + help="OPERATOR ONLY: release an unproven " + "ACTIVE run with a durable abandonment " + "record — the only sanctioned way to " + "drop a run without proving it") + bab.add_argument("--operator-abandon", action="store_true", + help="required confirmation flag (operator only)") + bab.add_argument("--reason", default="", + help="why the run is abandoned") + bab.add_argument("--workspace", default=".") + bab.set_defaults(func=_bob_abandon) + + bsg = bsub.add_parser( + "surgical", + help="lightweight lane for INERT doc/config edits: review " + "fan-out + operator quiz + gate. Code never qualifies — " + "it routes to the full pipeline") + bsg.add_argument("task") + bsg.add_argument("--files", nargs="+", required=True, + help="the inert files this run may edit (its " + "exact scope)") + bsg.add_argument("--workspace", default=".") + bsg.add_argument("--mock", action="store_true", + help="offline demo (no key)") + bsg.add_argument("--no-quiz", action="store_true", + help="record the quiz as skipped (non-interactive)") + bsg.add_argument("--mode", default="single", + choices=["single", "mixed"]) + bsg.add_argument("--config", default="") + bsg.set_defaults(func=_bob_surgical) + gp = sub.add_parser("panel-gate", help="block REPORT unless a valid panel execution " "receipt proves real panel-agent calls happened") diff --git a/src/agent_ultra/bob/__init__.py b/src/agent_ultra/bob/__init__.py index 86d293a..ccee4c2 100644 --- a/src/agent_ultra/bob/__init__.py +++ b/src/agent_ultra/bob/__init__.py @@ -13,20 +13,23 @@ """ from .receipts import ( - STEPS, GATED_STEPS, BobReceiptError, + STEPS, GATED_STEPS, SURGICAL_GATED_STEPS, BobReceiptError, build_step_receipt, load_receipt, validate_chain, ) from .pipeline import ( BobRun, BobProofError, GateResult, gate_check, run_pytest_step, assert_bob_done, load_or_create_key, - install_hooks, + install_hooks, scope_contains, scope_entry_is_infra, + surgical_gate_check, ) -from .runner import run_bob, BobOutcome +from .runner import run_bob, run_surgical, BobOutcome __all__ = [ "STEPS", "GATED_STEPS", "BobReceiptError", "build_step_receipt", "load_receipt", "validate_chain", "BobRun", "BobProofError", "GateResult", "gate_check", "run_pytest_step", "assert_bob_done", "load_or_create_key", - "install_hooks", "run_bob", "BobOutcome", + "install_hooks", "scope_contains", "scope_entry_is_infra", + "surgical_gate_check", "SURGICAL_GATED_STEPS", "run_surgical", + "run_bob", "BobOutcome", ] diff --git a/src/agent_ultra/bob/pipeline.py b/src/agent_ultra/bob/pipeline.py index 73d3bba..6450ecc 100644 --- a/src/agent_ultra/bob/pipeline.py +++ b/src/agent_ultra/bob/pipeline.py @@ -39,12 +39,16 @@ import subprocess import sys from dataclasses import dataclass, field -from pathlib import Path +from pathlib import Path, PureWindowsPath from .receipts import ( - STEPS, GATED_STEPS, BobReceiptError, + STEPS, GATED_STEPS, SURGICAL_GATED_STEPS, BobReceiptError, build_step_receipt, load_receipt, validate_chain, utc_now, ) +from .surgical import ( + surgical_ext_allowed, surgical_file_denied, staged_set_qualifies, + diff_budget_error, +) # staged files never produced by the build itself DEFAULT_ALLOW_UNCOVERED = ("*.md", ".agent-ultra/*", ".gitignore") @@ -86,7 +90,13 @@ def safe_rel(workspace: Path, name: str) -> str: names pass through here before anything is written.""" workspace = Path(workspace).resolve() cand = Path(name) - if cand.is_absolute() or cand.drive or cand.anchor: + # judge absoluteness under BOTH path flavors: on POSIX, "C:/Windows/x" + # has no drive under native Path semantics and would silently become a + # literal "C:" directory inside the workspace (Ubuntu CI caught this) — + # a model-authored Windows-absolute name must be rejected everywhere. + win = PureWindowsPath(str(name)) + if (cand.is_absolute() or cand.drive or cand.anchor + or win.is_absolute() or win.drive or win.anchor): raise ValueError(f"{name!r} is absolute") target = (workspace / cand).resolve() try: @@ -98,6 +108,49 @@ def safe_rel(workspace: Path, name: str) -> str: return rel.as_posix() +def _norm_scope_entry(s: str) -> str: + """Normalize a scope entry to a forward-slash, workspace-relative path + with no leading ./ (a trailing / marks a directory).""" + r = str(s).strip().replace("\\", "/") + while r.startswith("./"): + r = r[2:] + return r + + +def scope_entry_is_infra(entry: str) -> bool: + """True when a scope entry names enforcement infrastructure — the git + hooks or bob's own receipt/marker machinery. Such paths are never a + legitimate run target: a run must not be able to declare authority over + the thing that enforces it.""" + segs = _norm_scope_entry(entry).lower().rstrip("/").split("/") + if ".agent-ultra" in segs: + return True + if ".git" in segs: + i = segs.index(".git") + return i + 1 >= len(segs) or segs[i + 1] == "hooks" + return False + + +def scope_contains(scope, target_rel: str) -> bool: + """True when *target_rel* (workspace-relative) falls inside *scope* — + an exact file, anything under a declared directory, or an fnmatch glob. + Case-folded with .lower() on purpose: os.path.normcase would flip / to + \\ on Windows and break the forward-slash prefix checks.""" + t = _norm_scope_entry(target_rel).lower() + for raw in scope or []: + e = _norm_scope_entry(raw).lower() + if not e: + continue + ed = e.rstrip("/") + if t == ed: + return True + if t.startswith(ed + "/"): + return True + if fnmatch.fnmatch(t, e) or fnmatch.fnmatch(t, ed): + return True + return False + + def ultracode_receipt_fingerprint(home, run_id: str) -> str: """The sha256 of an ultracode run's receipt.json FILE BYTES, or '' when it does not exist. Recorded into the (signed) bob receipt so the gate can @@ -133,18 +186,86 @@ def _root(workspace: Path) -> Path: @classmethod def start(cls, workspace: str | Path, goal: str = "", key_path: str | Path | None = None, - force: bool = False) -> "BobRun": + force: bool = False, + scope: "list | None" = None, + operator_unbounded: bool = False, + mode: str = "full", + surgical_files: "list | None" = None) -> "BobRun": """Open a run. Refuses (ValueError) if another run is already ACTIVE and unproven — starting over would silently orphan it and release it - from enforcement. Pass ``force=True`` to abandon it deliberately.""" + from enforcement. Pass ``force=True`` to abandon it deliberately. + + ``scope`` — the files/dirs/globs this run is FOR — is MANDATORY: the + gate rejects staged files outside the declared scope, so a run + opened for module A cannot smuggle in edits to module B. A run with + no declared scope fails closed (start refuses; a scope stripped from + run.json later blocks every staged file at the gate). Expand a live + run's scope only through ``add_scope`` (validated + logged). + + ``operator_unbounded=True`` is the explicit, operator-only escape: + no scope binding for this run. It is a Python-API parameter on + purpose — no CLI verb exposes it, so an agent driving the CLI cannot + reach it. It is loudly announced and durably logged, and it never + covers the git hooks or bob's own machinery.""" workspace = Path(workspace) - marker = cls._root(workspace) / "ACTIVE" - if marker.is_file() and not force: - prev = marker.read_text(encoding="utf-8").strip() + if mode not in ("full", "surgical"): + raise ValueError(f"unknown bob mode: {mode!r}") + if mode == "surgical": + # the surgical lane is for INERT edits only — objective entry + # criteria validated up front; anything else is full-pipeline + # work. Its scope IS its declared files, exactly. + if operator_unbounded: + raise ValueError( + "operator_unbounded is NEVER valid for surgical mode — " + "a surgical run's scope is exactly its declared inert " + "files") + if not surgical_files: + raise ValueError( + "surgical_files is required for surgical mode — " + "declare the files you intend to edit") + for f in surgical_files: + if surgical_file_denied(f): + raise ValueError( + f"surgical mode denied_path {f!r} — this file can " + "execute in CI/build/install contexts; run the " + "full pipeline") + if not surgical_ext_allowed(f): + raise ValueError( + f"surgical mode rejects code file {f!r} — only " + "non-executable text types qualify; run the full " + "pipeline") + scope = list(surgical_files) + scope = [_norm_scope_entry(s) for s in (scope or []) if str(s).strip()] + for s in scope: + if scope_entry_is_infra(s): + raise ValueError( + f"scope {s!r} names enforcement infrastructure " + "(.git/hooks or .agent-ultra) — never a run target") + if operator_unbounded and scope: + raise ValueError( + "operator_unbounded cannot be combined with a declared " + "scope — declare the files OR the operator escape, not both") + if not scope and not operator_unbounded: raise ValueError( - f"a bob run ({prev}) is already active and unproven in this " - "workspace — finish it (`agent-ultra bob gate`) or pass " - "force=True to abandon it. Starting over would orphan it.") + "a bob run must declare its target scope (the files/dirs " + "this run is for) — no-scope runs fail closed; expand later " + "with add_scope if the work legitimately grows") + marker = cls._root(workspace) / "ACTIVE" + if marker.is_file(): + if not force: + prev = marker.read_text(encoding="utf-8").strip() + raise ValueError( + f"a bob run ({prev}) is already active and unproven in " + "this workspace — finish it (`agent-ultra bob gate`) or " + "have the operator abandon it explicitly (`agent-ultra " + "bob abandon --operator-abandon`, logged). A run cannot " + "be dropped by starting another.") + # force=True is the OPERATOR escape (CLI: --operator-force; the + # command broker classifies operator flags as dangerous, so an + # agent-driven shell cannot auto-run them). Never silent: the + # abandonment is durably logged. + cls.abandon(workspace, operator=True, + reason="superseded via start(force=True)") stamp = utc_stamp() run_dir = cls._root(workspace) / "runs" / stamp n = 0 @@ -152,10 +273,33 @@ def start(cls, workspace: str | Path, goal: str = "", n += 1 run_dir = cls._root(workspace) / "runs" / f"{stamp}-{n}" run_dir.mkdir(parents=True) - (run_dir / "run.json").write_text(json.dumps({ + meta = { "run_id": run_dir.name, "goal": goal, "started_utc": utc_now(), - }, indent=2, ensure_ascii=False), encoding="utf-8") + "scope": scope, "mode": mode, + } + if mode == "surgical": + meta["surgical_files"] = [_norm_scope_entry(f) + for f in surgical_files] + if operator_unbounded: + meta["operator_unbounded"] = True + (run_dir / "run.json").write_text(json.dumps( + meta, indent=2, ensure_ascii=False), encoding="utf-8") marker.write_text(run_dir.name, encoding="utf-8") + if operator_unbounded: + try: + with open(run_dir / "scope-log.jsonl", "a", + encoding="utf-8") as fh: + fh.write(json.dumps({ + "ts": utc_now(), "operator_unbounded": True, + "note": "UNBOUNDED run opened via the operator " + "escape — no scope binding for this run", + }) + "\n") + except OSError: + pass + print("!!! bob: OPERATOR-UNBOUNDED run — no scope binding; the " + "gate will not reject out-of-scope staged files for this " + "run (hooks/.agent-ultra stay off limits). Logged.", + file=sys.stderr) return cls(workspace=workspace, run_dir=run_dir, key=load_or_create_key(key_path)) @@ -181,17 +325,121 @@ def active(cls, workspace: str | Path, def marker_orphaned(self) -> bool: return not self.run_dir.is_dir() + @classmethod + def abandon(cls, workspace: str | Path, *, operator: bool = False, + reason: str = "") -> "str | None": + """OPERATOR-ONLY: release the ACTIVE marker of an unproven run, + leaving a durable abandonment record. This is the only sanctioned + way to drop a run without proving it — start() refuses to orphan, + and the command broker classifies the CLI's operator flags as + dangerous so an agent-driven shell cannot auto-run them. Returns + the abandoned run id, or None when no run is active. Works on an + orphaned marker too (releases it, still on the record).""" + if not operator: + raise ValueError( + "abandon requires the explicit operator flag — an agent " + "cannot abandon an unproven run") + workspace = Path(workspace) + marker = cls._root(workspace) / "ACTIVE" + if not marker.is_file(): + return None + name = marker.read_text(encoding="utf-8").strip() + rec = {"ts": utc_now(), "run_id": name, "reason": reason, + "operator_abandon": True} + try: + with open(cls._root(workspace) / "abandoned.jsonl", "a", + encoding="utf-8") as fh: + fh.write(json.dumps(rec) + "\n") + except OSError: + pass + run_dir = cls._root(workspace) / "runs" / name + if run_dir.is_dir(): + try: + (run_dir / "abandoned.json").write_text( + json.dumps(rec, indent=2), encoding="utf-8") + except OSError: + pass + marker.unlink() + print(f"!!! bob: run {name} ABANDONED by operator (unproven; " + "logged to .agent-ultra/bob/abandoned.jsonl)", + file=sys.stderr) + return name + @property def run_id(self) -> str: return self.run_dir.name - def goal(self) -> str: + def _meta(self) -> dict: try: meta = json.loads((self.run_dir / "run.json").read_text( encoding="utf-8")) - return str(meta.get("goal", "")) + return meta if isinstance(meta, dict) else {} except (OSError, ValueError): - return "" + return {} + + def goal(self) -> str: + return str(self._meta().get("goal", "")) + + def scope(self) -> list: + """The run's declared target scope. Scope is mandatory: an empty + scope authorizes NOTHING at the gate (fail closed) unless the run + was opened with the explicit operator_unbounded escape.""" + return list(self._meta().get("scope") or []) + + def unbounded(self) -> bool: + """True only for a run opened via the explicit, operator-only + operator_unbounded escape. run.json is plaintext, so this flag is + tamper-EVIDENT (loudly logged at start), not tamper-proof — same + trust ceiling as the rest of the run metadata.""" + return bool(self._meta().get("operator_unbounded")) + + def mode(self) -> str: + """Run mode from run.json; absent field = full (backward compat). + run.json is plaintext, so the mode is a REQUEST the gate verifies + (see gate_check's C1 dispatch), never a trusted assertion.""" + return str(self._meta().get("mode") or "full") + + def surgical_files(self) -> list: + return list(self._meta().get("surgical_files") or []) + + def add_scope(self, paths) -> "tuple[list, list]": + """Expand the run's scope — the approved, validated, logged + mechanism. Rejects infrastructure paths always. Appends to run.json + and records each expansion in scope-log.jsonl. Returns + (added, errors).""" + meta = self._meta() + cur = list(meta.get("scope") or []) + surgical = str(meta.get("mode") or "full") == "surgical" + added, errors = [], [] + for raw in paths or []: + e = _norm_scope_entry(raw) + if not e: + continue + if scope_entry_is_infra(e): + errors.append(f"{e}: enforcement infrastructure — never a " + "run target") + continue + if surgical and (surgical_file_denied(e) + or not surgical_ext_allowed(e)): + errors.append(f"{e}: not an inert/allowed type — a surgical " + "run cannot expand into risky paths (use the " + "full pipeline)") + continue + if e not in cur: + cur.append(e) + added.append(e) + if added: + meta["scope"] = cur + (self.run_dir / "run.json").write_text(json.dumps( + meta, indent=2, ensure_ascii=False), encoding="utf-8") + try: + with open(self.run_dir / "scope-log.jsonl", "a", + encoding="utf-8") as fh: + fh.write(json.dumps({"ts": utc_now(), + "added": added}) + "\n") + except OSError: + pass + return added, errors def receipt_path(self, step: str) -> Path: if step not in STEPS: @@ -395,6 +643,21 @@ def _as_int(value) -> int: return 0 +def cloned_fanout_error(run_id: str, done_agents) -> "str | None": + """A fan-out of N agents that all produced the SAME output is one + review wearing N hats — the per-agent output_sha256 in the ultracode + receipt exposes the clone. Only the all-identical case is flagged + (>=2 agents, one distinct digest): a single coincidental pair of short + verdicts stays legal, a copy-pasted sweep does not.""" + digests = {str(a.get("output_sha256") or "") for a in done_agents + if isinstance(a, dict) and a.get("output_sha256")} + if len(done_agents) >= 2 and len(digests) == 1: + return (f"ultracode run {run_id!r}: all {len(done_agents)} agents " + "produced IDENTICAL output — a cloned fan-out is one " + "review, not an independent sweep") + return None + + def check_ultracode_evidence(evidence: dict, *, allow_mock: bool, receipt_mock: bool, expect_workflow: str = "") -> list[str]: @@ -457,6 +720,13 @@ def check_ultracode_evidence(evidence: dict, *, allow_mock: bool, if not done: errors.append(f"ultracode run {run_id!r} completed 0 agents — " "an empty fan-out proves nothing") + # the offline mock route answers every agent identically by design, so + # the cloned-fanout check would always trip in demo mode; mock receipts + # are only acceptable under --allow-mock anyway, which is the boundary + # that already scopes what a demo may prove + if not (receipt_mock and allow_mock): + if (err := cloned_fanout_error(run_id, done)): + errors.append(err) if _as_int(rec.get("model_calls")) + _as_int(rec.get("cached_hits")) <= 0: errors.append(f"ultracode run {run_id!r} made 0 model calls") if receipt_mock and not allow_mock: @@ -534,6 +804,22 @@ def gate_check(run: BobRun, *, rerun_tests: bool = True, receipts themselves must still hold). """ try: + # ---- mode dispatch (C1: mode-downgrade defense) -------------------- + # run.json's mode field is a REQUEST the gate verifies, never a + # trusted assertion. The commit is routed to the surgical gate ONLY + # when the run asks for it AND the actual staged set independently + # qualifies (all inert, none denied, all declared). Flipping a + # run's mode to "surgical" with code staged therefore routes it to + # the FULL gate — which demands the full receipt chain and blocks. + if run.mode() == "surgical" and not run.marker_orphaned(): + routed = staged_files + if routed is None and _inside_work_tree(Path(run.workspace)): + routed = _staged_files(Path(run.workspace)) + if routed is not None and staged_set_qualifies( + routed, run.surgical_files()): + return surgical_gate_check(run, allow_mock=allow_mock, + staged_files=routed) + # falls through: the staged set does not qualify -> FULL gate return _gate_check(run, rerun_tests=rerun_tests, allow_mock=allow_mock, staged_files=staged_files, allow_uncovered=allow_uncovered) @@ -543,6 +829,137 @@ def gate_check(run: BobRun, *, rerun_tests: bool = True, f"{type(e).__name__}: {e}"]) +def surgical_gate_check(run: BobRun, *, allow_mock: bool = False, + staged_files=None) -> GateResult: + """The surgical lane's gate — fail-closed throughout, and NARROWER than + full mode on every axis it covers: declared-file allowlist, inert types + only, a denylist, and a small diff budget. It can never loosen the full + gate: gate_check routes here only when the staged set independently + qualifies; everything else gets the full 10-step demands.""" + try: + return _surgical_gate_check(run, allow_mock=allow_mock, + staged_files=staged_files) + except Exception as e: + return GateResult(passed=False, + errors=[f"surgical gate error (fail-closed): " + f"{type(e).__name__}: {e}"]) + + +def _surgical_gate_check(run: BobRun, *, allow_mock: bool, + staged_files) -> GateResult: + errors: list[str] = [] + checked: dict = {} + + if run.marker_orphaned(): + return GateResult(passed=False, errors=[ + f"run {run.run_id}: ACTIVE marker names a run directory that no " + "longer exists — the run was deleted while unproven (fail closed)"]) + + declared = run.surgical_files() + if not declared: + errors.append("surgical: no surgical_files declared in run.json — " + "fail closed") + for f in declared: + if surgical_file_denied(f) or not surgical_ext_allowed(f): + errors.append(f"surgical: declared file {f} is not an inert/" + "allowed type — use the full pipeline") + + # chain spine: same signed/hash-chained demands as full mode + chain = run.chain() + errors.extend(validate_chain(chain, run.key or None)) + checked["chain"] = f"{len(chain)} receipt(s), hash-chained + signed" + + have = {r.get("step") for r in chain} + for step in SURGICAL_GATED_STEPS: + if step not in have: + errors.append(f"{step}: receipt missing — the step never ran " + "(or its receipt was deleted)") + + def _ev(rec) -> dict: + ev = rec.get("evidence") if isinstance(rec, dict) else None + return ev if isinstance(ev, dict) else {} + + # ---- review: cross-checked against the ultracode run it names -------- + review = run.load("step_surgical_review") + if review is not None: + ev = _ev(review) + rerrs = check_ultracode_evidence( + ev, allow_mock=allow_mock, receipt_mock=bool(review.get("mock")), + expect_workflow="bob-surgical-review") + errors.extend(f"step_surgical_review: {e}" for e in rerrs) + checked["step_surgical_review"] = ("ultracode run verified" + if not rerrs else "INVALID") + + # ---- quiz: passed requires the captured operator response ------------ + quiz = run.load("step_surgical_quiz") + if quiz is not None: + ev = _ev(quiz) + if (e := _mock_flag_error(quiz, "step_surgical_quiz", allow_mock)): + errors.append(e) + outcome = ev.get("outcome") + if outcome not in ("passed", "skipped"): + errors.append(f"step_surgical_quiz: outcome {outcome!r} is not " + "passed/skipped") + if not ev.get("questions"): + errors.append("step_surgical_quiz: no comprehension questions " + "recorded") + if outcome == "passed" and not str( + ev.get("operator_response", "") or "").strip(): + errors.append("step_surgical_quiz: outcome 'passed' without a " + "captured operator response — a quiz nobody " + "answered (self-graded quizzes do not count)") + checked["step_surgical_quiz"] = f"outcome={outcome}" + + # ---- staleness: covered files must match disk ------------------------- + coverage: dict = {} + for rec in chain: + coverage.update(rec.get("files") or {}) + for rel, expected in sorted(coverage.items()): + f = Path(run.workspace) / rel + if not f.is_file(): + errors.append(f"stale: {rel} covered by a receipt but no longer " + "exists") + elif sha256_file(f) != expected: + errors.append(f"stale: {rel} changed after its last covering " + "receipt — re-run the review") + checked["staleness"] = f"{len(coverage)} file(s) covered" + + # ---- staged set: declared + inert only, fail-closed -------------------- + if staged_files is None: + if _inside_work_tree(Path(run.workspace)): + staged_files = _staged_files(Path(run.workspace)) + if staged_files is None: + errors.append("surgical: could not read the staged list " + "(git error) — fail closed") + staged_files = [] + else: + staged_files = [] + declared_norm = {os.path.normpath(str(d)).replace("\\", "/").lower() + for d in declared} + for rel in staged_files: + norm = os.path.normpath(str(rel)).replace("\\", "/").lower() + if norm.startswith(".agent-ultra/"): + continue # run machinery rides along + if surgical_file_denied(rel): + errors.append(f"surgical: staged denied_path {rel} — this file " + "executes; use the full pipeline") + elif not surgical_ext_allowed(rel): + errors.append(f"surgical: staged {rel} is not an inert type — " + "the surgical lane is docs/config only") + elif norm not in declared_norm: + errors.append(f"surgical: staged {rel} was not declared in " + "surgical_files") + + # ---- diff budget: small edits only, fail-closed ------------------------- + if _inside_work_tree(Path(run.workspace)): + if (berr := diff_budget_error(Path(run.workspace))): + errors.append(berr) + checked["diff_budget"] = "within budget" if not any( + "budget" in e or "BINARY" in e for e in errors) else "EXCEEDED" + + return GateResult(passed=not errors, errors=errors, checked=checked) + + def _mock_flag_error(rec: dict, step: str, allow_mock: bool) -> "str | None": if rec.get("mock") and not allow_mock: return (f"{step}: receipt marked mock — mock evidence only satisfies " @@ -718,6 +1135,31 @@ def _counts(ev) -> dict: errors.append(f"uncovered: staged file {rel} appears in no receipt — " "it was never tested or fanned out for scrutiny") + # ---- staged files must be WITHIN the run's declared scope ------------------ + # A run opened for module A must not authorize commits touching module B. + # Scope is mandatory: no scope + not the operator escape = every staged + # file blocks (fail closed) and the missing declaration is itself an + # error. Bob's own machinery (.agent-ultra receipts) is always exempt. + scope = run.scope() + if run.unbounded(): + checked["scope"] = "OPERATOR-UNBOUNDED (no scope binding — logged)" + else: + if not scope: + errors.append( + "scope: run declares no target scope — scope is mandatory " + "and a no-scope run authorizes nothing (fail closed)") + for rel in staged_files: + if rel.startswith(".agent-ultra/") or rel == ".gitignore": + continue + if scope and scope_contains(scope, rel): + continue + errors.append( + f"out-of-scope: staged file {rel} is outside the run's " + "declared scope — if this edit is really part of the run, " + "expand the scope explicitly (`agent-ultra bob scope-add`)") + checked["scope"] = (f"{len(scope)} declared entr" + + ("y" if len(scope) == 1 else "ies")) + return GateResult(passed=not errors, errors=errors, checked=checked) diff --git a/src/agent_ultra/bob/receipts.py b/src/agent_ultra/bob/receipts.py index 77a3265..9f5cdc2 100644 --- a/src/agent_ultra/bob/receipts.py +++ b/src/agent_ultra/bob/receipts.py @@ -31,6 +31,9 @@ "step01_spec", "step02_red", "step03_green", "step04_refactor", "step05_codequality", "step06_security", "step07_workflow", "step08_panel", "step09_quiz", "step10_commit", + # surgical lane (lightweight tier for INERT doc/config edits — never a + # substitute for the full pipeline; code always routes to full) + "step_surgical_review", "step_surgical_quiz", ) # Steps the gate demands a receipt for. 1/4/5 are judgment steps where a @@ -38,6 +41,9 @@ GATED_STEPS = ("step02_red", "step03_green", "step06_security", "step07_workflow", "step08_panel", "step09_quiz") +# the surgical lane's own (narrower, harder) receipt demands +SURGICAL_GATED_STEPS = ("step_surgical_review", "step_surgical_quiz") + WRITERS = frozenset({"system", "engine", "agent"}) diff --git a/src/agent_ultra/bob/runner.py b/src/agent_ultra/bob/runner.py index 5f82372..ed69b4b 100644 --- a/src/agent_ultra/bob/runner.py +++ b/src/agent_ultra/bob/runner.py @@ -27,7 +27,8 @@ sha256_file, safe_rel, ultracode_receipt_fingerprint, ) from . import sample -from .workflows import SECURITY_WORKFLOW, REVIEW_WORKFLOW +from .workflows import (SECURITY_WORKFLOW, REVIEW_WORKFLOW, + SURGICAL_REVIEW_WORKFLOW) @dataclass @@ -131,8 +132,38 @@ def run_bob(task: str, workspace: str | Path, *, mock: bool = False, if interactive is None: interactive = (not mock) and sys.stdin.isatty() + # ---- content generation happens BEFORE the run opens, because the run + # must declare its target scope at start (the files this run is FOR). + # The gate then rejects any staged file outside that scope, so a run + # opened for one module cannot smuggle in edits to another. + spec = sample.SAMPLE_SPEC if mock else _gen_spec(pool, task) + if not spec: + _say(out, "cannot start: no spec produced") + return BobOutcome(run_id="", passed=False, blocked_at="step01_spec") + if mock: + test_file, tests, impl_file, stub = (sample.TEST_FILE, + sample.SAMPLE_TESTS, + sample.IMPL_FILE, + sample.SAMPLE_STUB) + else: + gen = _gen_files(pool, task, spec) + if gen is None: + _say(out, "cannot start: model produced no usable tests") + return BobOutcome(run_id="", passed=False, blocked_at="step02_red") + test_file, tests, impl_file, stub = gen + # model-authored file names are untrusted: they must land INSIDE the + # workspace (no absolute paths, no `..` traversal). This mirrors the rest + # of the kit's "model output is untrusted input" stance. try: - run = BobRun.start(workspace, goal=task, key_path=key_path, force=force) + test_file = safe_rel(workspace, test_file) + impl_file = safe_rel(workspace, impl_file) + except ValueError as e: + _say(out, f"cannot start: unsafe model-authored path: {e}") + return BobOutcome(run_id="", passed=False, blocked_at="step02_red") + + try: + run = BobRun.start(workspace, goal=task, key_path=key_path, + force=force, scope=[test_file, impl_file]) except ValueError as e: _say(out, f"cannot start: {e}") return BobOutcome(run_id="", passed=False, blocked_at="start") @@ -153,32 +184,11 @@ def blocked(step, why) -> BobOutcome: + (" [mock: sample task, no key]" if mock else "")) # ---- 1. SPEC ----------------------------------------------------------- - spec = sample.SAMPLE_SPEC if mock else _gen_spec(pool, task) - if not spec: - return blocked("step01_spec", "no spec produced") run.write("step01_spec", {"goal": task, "spec": spec}, writer="agent", mock=mock) record("step01_spec", spec[:70] + ("..." if len(spec) > 70 else "")) # ---- 2. RED ------------------------------------------------------------- - if mock: - test_file, tests, impl_file, stub = (sample.TEST_FILE, - sample.SAMPLE_TESTS, - sample.IMPL_FILE, - sample.SAMPLE_STUB) - else: - gen = _gen_files(pool, task, spec) - if gen is None: - return blocked("step02_red", "model produced no usable tests") - test_file, tests, impl_file, stub = gen - # model-authored file names are untrusted: they must land INSIDE the - # workspace (no absolute paths, no `..` traversal). This mirrors the rest - # of the kit's "model output is untrusted input" stance. - try: - test_file = safe_rel(workspace, test_file) - impl_file = safe_rel(workspace, impl_file) - except ValueError as e: - return blocked("step02_red", f"unsafe model-authored path: {e}") for rel, content in ((test_file, tests), (impl_file, stub)): p = workspace / rel p.parent.mkdir(parents=True, exist_ok=True) @@ -347,3 +357,118 @@ def _fanout_evidence(rep, wf_name, extra): "seals it once your commit lands (`bob hook-install`).") return BobOutcome(run_id=run.run_id, passed=True, gate=gate, steps=steps, report=report_doc) + + +# -- the surgical lane --------------------------------------------------------- + +def run_surgical(task: str, workspace: str | Path, files, *, + mock: bool = False, pool=None, key_path=None, + ultracode_home=None, interactive: "bool | None" = None, + seal_on_pass: bool = True, out=None) -> BobOutcome: + """The lightweight lane for INERT doc/config edits: review fan-out + + operator quiz + gate. Edit the files FIRST, then run this — the review + examines and pins the current content, and any later edit trips the + staleness check. + + Entry criteria are objective and validated at start (inert types only, + denylist, no infrastructure); anything that does not qualify raises and + the answer is the FULL pipeline (`run_bob`). The gate additionally + re-verifies the staged set at commit time, so the surgical lane can + never launder code changes — it narrows Bob, it never replaces it.""" + from ..routes.pool import RoutePool + workspace = Path(workspace) + if mock: + from ..ultracode import demo_pool + pool = demo_pool() + elif pool is None: + raise ValueError("a real run needs a configured RoutePool; " + "pass pool= or use mock=True") + if interactive is None: + interactive = (not mock) and sys.stdin.isatty() + + try: + run = BobRun.start(workspace, goal=task, key_path=key_path, + mode="surgical", surgical_files=list(files)) + except ValueError as e: + _say(out, f"cannot start: {e}") + return BobOutcome(run_id="", passed=False, blocked_at="start") + steps: list = [] + + def record(step, summary): + steps.append((step, summary)) + _say(out, f" [{step}] {summary}") + + def blocked(step, why) -> BobOutcome: + _say(out, f"\nBLOCKED at {step}: {why}") + _say(out, f"run {run.run_id} stays ACTIVE — finish or redo the " + "step, then `agent-ultra bob gate`.") + return BobOutcome(run_id=run.run_id, passed=False, blocked_at=step, + steps=steps) + + _say(out, f"bob surgical run {run.run_id} — {task}" + + (" [mock: no key]" if mock else "")) + _say(out, f" files: {list(files)}") + + # ---- review: real ultracode fan-out over the current content ---------- + from ..ultracode import UltracodeEngine + eng = UltracodeEngine(pool, home=ultracode_home) + excerpt = _excerpt(workspace, list(files)) + rep = eng.run_script(SURGICAL_REVIEW_WORKFLOW, + args={"goal": task, "files": list(files), + "target_excerpt": excerpt}) + if rep.final_state != "COMPLETE": + return blocked("step_surgical_review", + f"review fan-out ended {rep.final_state}") + result = rep.result or {} + fp = ultracode_receipt_fingerprint(eng.home, rep.run_id) + run.write("step_surgical_review", + {"ultracode": {"run_id": rep.run_id, "home": str(eng.home), + "workflow_name": "bob-surgical-review", + "receipt_file_sha256": fp}, + "lenses": result.get("lenses", []), + "reviews": result.get("reviews", {})}, + writer="engine", files=list(files), mock=mock) + record("step_surgical_review", + f"{rep.agents_run} agents over {len(result.get('lenses', []))} lenses") + + # ---- quiz: operator answer or the honest 'skipped' -------------------- + questions = [ + f"What exactly changed in {list(files)}, in your own words?", + "Could this edit alter any executable behavior? Why not?", + "What would make you revert it?", + ] + operator_response = "" + outcome = "skipped" + if interactive: + _say(out, "\nquiz — answer to claim 'passed' (empty answer skips):") + for q in questions: + _say(out, f" Q: {q}") + try: + operator_response = input(" A: ").strip() + except (EOFError, OSError): + operator_response = "" + outcome = "passed" if operator_response else "skipped" + run.write("step_surgical_quiz", + {"questions": questions, "outcome": outcome, + "operator_response": operator_response}, + writer="agent" if outcome == "passed" else "engine", mock=mock) + record("step_surgical_quiz", f"outcome={outcome}") + + # ---- gate --------------------------------------------------------------- + gate = gate_check(run, allow_mock=mock, staged_files=[]) + if not gate.passed: + _say(out, "\ngate BLOCKED:") + for e in gate.errors: + _say(out, f" ! {e}") + return BobOutcome(run_id=run.run_id, passed=False, + blocked_at="step_surgical_gate", gate=gate, + steps=steps) + if seal_on_pass: + run.complete(gate.to_dict()) + else: + run.mark_pass(gate.to_dict()) + _say(out, f"\ngate PASSED — surgical chain validates " + f"({len(run.chain())} receipts).") + _say(out, "safe to commit the declared files: git add + git commit " + "(the pre-commit gate re-verifies the staged set).") + return BobOutcome(run_id=run.run_id, passed=True, gate=gate, steps=steps) diff --git a/src/agent_ultra/bob/surgical.py b/src/agent_ultra/bob/surgical.py new file mode 100644 index 0000000..2d2d6ec --- /dev/null +++ b/src/agent_ultra/bob/surgical.py @@ -0,0 +1,261 @@ +"""Surgical lane entry criteria — the lightweight tier for INERT edits. + +The surgical lane exists for typo-class doc/config edits where the full +10-step pipeline is overkill. It never replaces Bob: every surgical check +is a HARDER, NARROWER constraint than full mode (declared-file allowlist, +inert file types, a denylist, a small diff budget), and anything that does +not qualify falls back to the FULL pipeline — the heavier lane is never +weakened by the lighter one existing. + +Qualification is allowlist-first: + + * truly-inert PROSE (.md/.rst/.txt/...) and known-inert dotfiles + auto-qualify; + * STRUCTURED CONFIG (yaml/json/toml/ini/cfg) is dual-use and does NOT + auto-qualify — it needs the operator's explicit opt-in + (AGENT_ULTRA_SURGICAL_ALLOW_CONFIG=1) and must still clear the + denylist + shape checks; + * everything else is full-pipeline work. + +The denylist is files whose CONTENT executes in CI/build/install/runtime +contexts regardless of extension (workflows, hook managers, task runners, +package manifests, tool configs). Matching is path-aware (exact basename / +basename prefix / path prefix / shape), never substring — a doc ABOUT +Dockerfiles is a doc, not a Dockerfile. Stdlib only. +""" + +from __future__ import annotations + +import os +import re +import subprocess +from pathlib import Path + +SURGICAL_DOC_EXTS = frozenset({ + ".md", ".markdown", ".rst", ".txt", ".adoc", +}) +SURGICAL_STRUCTURED_CONFIG_EXTS = frozenset({ + ".json", ".yaml", ".yml", ".toml", ".cfg", ".ini", +}) +SURGICAL_ALLOWED_EXTENSIONS = (SURGICAL_DOC_EXTS + | SURGICAL_STRUCTURED_CONFIG_EXTS) +# dotfiles have no suffix in pathlib — check by basename instead +SURGICAL_ALLOWED_DOTFILES = frozenset({ + ".gitignore", ".editorconfig", ".gitattributes", +}) + +# diff budget: surgical is for SMALL inert edits; anything bigger is full +# pipeline work by definition +SURGICAL_MAX_DIFF_LINES = 100 +SURGICAL_MAX_DIFF_BYTES = 256 * 1024 + + +def _config_opt_in() -> bool: + return os.environ.get("AGENT_ULTRA_SURGICAL_ALLOW_CONFIG", "0") == "1" + + +# Paths ALWAYS denied in surgical mode even when the extension is allowed — +# these execute code in CI/build/install/runtime contexts. +SURGICAL_DENIED_BASENAMES = frozenset({ + # CI/CD (many providers) + ".gitlab-ci.yml", "azure-pipelines.yml", "jenkinsfile", ".drone.yml", + ".travis.yml", "bitbucket-pipelines.yml", "appveyor.yml", + ".appveyor.yml", "cloudbuild.yaml", "cloudbuild.yml", + ".woodpecker.yml", ".woodpecker.yaml", "buildkite.yml", ".semaphore.yml", + "wercker.yml", "codeship-services.yml", "codeship-steps.yml", + # git-hook MANAGERS (execute arbitrary shell on git ops) + ".pre-commit-config.yaml", ".pre-commit-config.yml", + "lefthook.yml", "lefthook.yaml", ".lefthook.yml", ".lefthook.yaml", + ".overcommit.yml", ".huskyrc", ".huskyrc.json", ".huskyrc.yml", + # task runners / build DSLs + "taskfile.yml", "taskfile.yaml", "justfile", ".justfile", + "makefile", "gnumakefile", "rakefile", "snakefile", "cmakelists.txt", + # IaC / deploy (run code on apply/deploy) + "serverless.yml", "serverless.yaml", "netlify.toml", "fly.toml", + "vercel.json", "now.json", "pulumi.yaml", "template.yaml", + "samconfig.toml", "app.yaml", "cloudformation.yaml", "cloudformation.yml", + # container/build/devcontainer + "dockerfile", "containerfile", ".dockerignore", "procfile", + "devcontainer.json", "compose.yaml", "compose.yml", + "skaffold.yaml", "skaffold.yml", "wrangler.toml", "turbo.json", + "nx.json", "firebase.json", "deno.json", "deno.jsonc", "pubspec.yaml", + "supervisord.ini", "supervisord.conf", + # task-runner / tooling ecosystems that execute shell + "mise.toml", ".mise.toml", ".mise.local.toml", "dvc.yaml", + "moon.yml", "bunfig.toml", "biome.json", "earthly.toml", "earthfile", + "goreleaser.yml", "goreleaser.yaml", "release-please-config.json", + # test-runner / tool configs the shape regex doesn't catch by name + "pytest.ini", "nodemon.json", ".flake8", ".pylintrc", + # more code-executing tool configs (opt-in path) + "helmfile.yaml", "helmfile.yml", "chezmoi.toml", ".chezmoi.toml", + "dagger.json", "invoke.yaml", "invoke.yml", "tasks.py", "fabfile.py", + "helmfile.d", + # package / dependency manifests (code execution on install) + "package.json", "package-lock.json", "composer.json", "composer.lock", + "pyproject.toml", "setup.cfg", "setup.py", "tox.ini", "noxfile.py", + "requirements.txt", "pipfile", "pipfile.lock", "conanfile.txt", + "conanfile.py", "meta.yaml", "environment.yml", "environment.yaml", + "cargo.toml", "cargo.lock", "go.mod", "go.sum", + "gemfile", "gemfile.lock", "podfile", "cartfile", + "pom.xml", "build.gradle", "build.gradle.kts", "build.sbt", "build.xml", + "build.boot", "project.clj", "deps.edn", "mix.exs", "rebar.config", + ".npmrc", ".yarnrc", ".yarnrc.yml", "yarn.lock", "pnpm-lock.yaml", + "renovate.json", ".renovaterc", ".renovaterc.json", "dependabot.yml", + # environment injection + ".env", ".env.local", ".env.production", ".env.development", + # editor/runtime autoloaders that execute + ".vimrc", ".exrc", ".ideavimrc", "conftest.py", +}) +SURGICAL_DENIED_BASENAME_PREFIXES = ("docker-compose", "compose.", + "taskfile", ".env.", "cloudbuild.") +SURGICAL_DENIED_PATH_PREFIXES = ( + ".github/workflows/", ".github/actions/", ".circleci/", + ".gitea/workflows/", ".forgejo/workflows/", ".woodpecker/", + ".buildkite/", ".husky/", ".teamcity/", ".azuredevops/", + ".devcontainer/", "k8s/", "helm/", "kustomize/", ".config/systemd/", + ".vscode/", ".moon/") +# shape-based detection so novel/unlisted CI/hook/deploy tools are caught +# structurally (not by name). Any yml/yaml/json/toml whose path looks like +# a workflow/pipeline/CI/deploy/hook config falls back to the FULL +# pipeline. Doc-about-CI (.md/.txt) is unaffected — this only gates the +# executable-config extensions. +_EXEC_CONFIG_EXTS = frozenset( + {".yml", ".yaml", ".json", ".toml", ".ini", ".cfg", ".conf"}) +_EXEC_PATH_RE = re.compile( + r"(?:^|/)(?:\.?[a-z]*ci|workflows?|pipelines?|\.?deploy|hooks?|" + r"actions?|compose|devcontainer|container|k8s|kube|helm[a-z]*|kustomiz|" + r"serverless|terraform|ansible|chezmoi|dagger)(?:/|$|[._-])", + re.IGNORECASE) +# the whole TOOL-CONFIG class (test runners, linters, transpilers, +# bundlers) executes arbitrary code when its tool runs — a name census +# never converges, so match the SHAPE: .config., +# .rc[.], .conf.. Denied -> falls back to full Bob. +_TOOL_CONFIG_RE = re.compile( + r"(?:\.config\.[a-z0-9]+$)" + r"|(?:\.conf\.[a-z0-9]+$)" + r"|(?:^\.[a-z][a-z0-9_-]*rc$)" + r"|(?:^\.[a-z][a-z0-9_-]*rc\.[a-z0-9]+$)", + re.IGNORECASE) +# enforcement infrastructure is NEVER surgical-editable — the gate cannot +# be reconfigured through its own lightweight lane. Matched as path +# segments. +SURGICAL_DENIED_PATH_SEGMENTS = (".git", ".agent-ultra") + + +def surgical_file_denied(path: str) -> bool: + """True when *path* is on the surgical deny list — a file whose content + executes in CI/build/install contexts regardless of extension. + Path-aware matching (exact basename / basename prefix / path prefix / + shape), NOT substring.""" + norm = os.path.normpath(str(path)).replace("\\", "/").lower() + base = norm.rsplit("/", 1)[-1] + if base in SURGICAL_DENIED_BASENAMES: + return True + if any(base.startswith(pfx) for pfx in SURGICAL_DENIED_BASENAME_PREFIXES): + return True + if _TOOL_CONFIG_RE.search(base): + return True + segments = set(norm.split("/")) + if any(seg in segments for seg in SURGICAL_DENIED_PATH_SEGMENTS): + return True + if any(norm.startswith(pfx) or f"/{pfx}" in f"/{norm}" + for pfx in SURGICAL_DENIED_PATH_PREFIXES): + return True + ext = ("." + base.rsplit(".", 1)[-1]) if "." in base else "" + if ext in _EXEC_CONFIG_EXTS and _EXEC_PATH_RE.search(norm): + return True + return False + + +def surgical_ext_allowed(path: str) -> bool: + """True when *path* auto-qualifies for the surgical lane: truly-inert + PROSE and known-inert dotfiles always; STRUCTURED CONFIG only when the + operator opts in (AGENT_ULTRA_SURGICAL_ALLOW_CONFIG=1) AND it clears + the denylist/shape checks. Otherwise: full pipeline.""" + p = Path(str(path)) + if p.name.lower() in SURGICAL_ALLOWED_DOTFILES: + return True + ext = p.suffix.lower() + if ext in SURGICAL_DOC_EXTS: + return True + if ext in SURGICAL_STRUCTURED_CONFIG_EXTS: + return _config_opt_in() and not surgical_file_denied(path) + return False + + +def staged_set_qualifies(staged_files, declared) -> bool: + """True when every staged file independently qualifies for the surgical + lane AND was declared up front. This is the C1 mode-downgrade defense's + routing test: run.json's mode field is a REQUEST the gate verifies, + never a trusted assertion — flipping a run's mode to surgical with code + staged routes the commit to the FULL gate, which then demands the full + receipt chain.""" + declared_norm = {os.path.normpath(str(d)).replace("\\", "/").lower() + for d in (declared or [])} + for rel in staged_files or []: + norm = os.path.normpath(str(rel)).replace("\\", "/").lower() + if norm.startswith(".agent-ultra/"): + continue # run machinery rides along + if surgical_file_denied(rel) or not surgical_ext_allowed(rel): + return False + if norm not in declared_norm: + return False + return True + + +def _git_diff_stat(workspace: Path) -> "int | None": + """Total staged lines changed (added+deleted). Fail-closed: None on any + git error so the diff-budget check blocks rather than skips. Returns -1 + if any staged file is BINARY (numstat '-') — surgical is text-only.""" + try: + p = subprocess.run(["git", "-C", str(workspace), "diff", "--cached", + "--numstat"], capture_output=True, text=True, + timeout=30) + except (OSError, subprocess.SubprocessError): + return None + if p.returncode != 0: + return None + total = 0 + for ln in p.stdout.splitlines(): + parts = ln.split("\t") + if len(parts) >= 2: + if parts[0] == "-" or parts[1] == "-": + return -1 # binary staged edit + for n in parts[:2]: + if n.isdigit(): + total += int(n) + return total + + +def _git_diff_bytes(workspace: Path) -> "int | None": + """Total bytes in the staged diff. Fail-closed: None on git error.""" + try: + p = subprocess.run(["git", "-C", str(workspace), "diff", "--cached"], + capture_output=True, timeout=30) + except (OSError, subprocess.SubprocessError): + return None + if p.returncode != 0: + return None + return len(p.stdout or b"") + + +def diff_budget_error(workspace: Path) -> "str | None": + """None if the staged diff fits the surgical budget; else the blocking + message. Fail-closed on any git error, over-line, binary, or over-byte.""" + lines = _git_diff_stat(Path(workspace)) + if lines is None: + return ("surgical: could not compute staged diff size (git error) " + "— fail-closed") + if lines == -1: + return "surgical: staged diff contains a BINARY file — use full mode" + if lines > SURGICAL_MAX_DIFF_LINES: + return (f"surgical: staged diff {lines} lines exceeds the " + f"{SURGICAL_MAX_DIFF_LINES}-line budget — use full mode") + nbytes = _git_diff_bytes(Path(workspace)) + if nbytes is None: + return ("surgical: could not measure staged diff bytes (git error) " + "— fail-closed") + if nbytes > SURGICAL_MAX_DIFF_BYTES: + return (f"surgical: staged diff {nbytes} bytes exceeds the " + f"{SURGICAL_MAX_DIFF_BYTES}-byte budget — use full mode") + return None diff --git a/src/agent_ultra/bob/workflows/__init__.py b/src/agent_ultra/bob/workflows/__init__.py index de94c24..5735a70 100644 --- a/src/agent_ultra/bob/workflows/__init__.py +++ b/src/agent_ultra/bob/workflows/__init__.py @@ -6,3 +6,4 @@ SECURITY_WORKFLOW = WORKFLOWS_DIR / "security_fanout.py" REVIEW_WORKFLOW = WORKFLOWS_DIR / "review_fanout.py" +SURGICAL_REVIEW_WORKFLOW = WORKFLOWS_DIR / "surgical_review.py" diff --git a/src/agent_ultra/bob/workflows/surgical_review.py b/src/agent_ultra/bob/workflows/surgical_review.py new file mode 100644 index 0000000..2d66aad --- /dev/null +++ b/src/agent_ultra/bob/workflows/surgical_review.py @@ -0,0 +1,48 @@ +"""bob surgical lane — the review step as an ultracode workflow. + +Two agents examine the inert edit independently: one for accuracy (does the +text/config say what it should), one for safety (does the change smuggle in +anything with side effects — links, commands, config keys). The bob gate +cross-checks this run's receipt exactly like steps 6/7: fingerprint-pinned, +workflow-name-bound, and a claimed-but-never-run review blocks the commit. +""" + +META = { + "name": "bob-surgical-review", + "description": "bob surgical lane: accuracy + safety review of an inert edit", + "phases": ["Review", "Merge"], +} + +LENSES = ("accuracy", "safety") + +REVIEW_SCHEMA = { + "type": "object", + "properties": { + "verdict": {"type": "string", "enum": ["pass", "concerns"]}, + "notes": {"type": "array", "items": {"type": "string"}}, + }, + "required": ["verdict", "notes"], +} + + +async def run(wf): + args = wf.args if isinstance(wf.args, dict) else {} + goal = str(args.get("goal", "") or "the edit") + excerpt = str(args.get("target_excerpt", "") or "")[:6000] + files = args.get("files") or [] + + wf.phase("Review") + results = await wf.parallel([ + (lambda lens=lens: wf.agent( + f"Review this doc/config edit on the {lens} lens ONLY.\n" + f"Task: {goal}\nFiles: {files}\nContent:\n{excerpt}\n" + "Return your verdict and notes.", + schema=REVIEW_SCHEMA, label=f"surgical:{lens}")) + for lens in LENSES + ]) + + wf.phase("Merge", kind="fan_in") + merged = {lens: (res or {"verdict": "concerns", + "notes": ["agent failed — treat as unreviewed"]}) + for lens, res in zip(LENSES, results)} + return {"lenses": list(LENSES), "reviews": merged, "agents": len(LENSES)} diff --git a/src/agent_ultra/broker/broker.py b/src/agent_ultra/broker/broker.py index 43e299d..e1912cf 100644 --- a/src/agent_ultra/broker/broker.py +++ b/src/agent_ultra/broker/broker.py @@ -91,6 +91,11 @@ "destructive system/registry op"), (re.compile(r"(?i)\bstripe\b|\bbilling\b|\bcharge\b|\bpurchase\b"), "may spend money"), + # bob's operator-only escape flags (abandon an unproven run / + # force-supersede one / start unbounded). Operator decisions, + # never an agent's -- an agent-driven shell must not auto-run them. + (re.compile(r"(?i)--operator-(abandon|force|unbounded)"), + "operator-only pipeline escape flag"), ] # -- ELEVATED: normal local dev with side effects --------------------------- diff --git a/tests/test_bob.py b/tests/test_bob.py index 49bec1a..833e63b 100644 --- a/tests/test_bob.py +++ b/tests/test_bob.py @@ -18,6 +18,7 @@ ) from agent_ultra.bob.pipeline import ( check_ultracode_evidence, safe_rel, parse_pytest_nodes, + scope_contains, scope_entry_is_infra, ) KEY = b"bob-test-key-0123456789abcdef" @@ -82,7 +83,8 @@ def test_removed_middle_receipt_breaks_linkage(): def test_red_rejects_passing_tests(tmp_path): (tmp_path / "test_t.py").write_text("def test_a():\n assert True\n", encoding="utf-8") - run = BobRun.start(tmp_path, goal="g", key_path=tmp_path / "k.key") + run = BobRun.start(tmp_path, goal="g", key_path=tmp_path / "k.key", + scope=["test_t.py"]) ok, rec, msg = run_pytest_step(run, "red", "test_t.py") assert not ok and rec["writer"] == "system" assert rec["evidence"]["requirement_met"] is False @@ -91,7 +93,8 @@ def test_red_rejects_passing_tests(tmp_path): def test_red_rejects_collection_errors(tmp_path): (tmp_path / "test_t.py").write_text("import missing_module_xyz\n", encoding="utf-8") - run = BobRun.start(tmp_path, goal="g", key_path=tmp_path / "k.key") + run = BobRun.start(tmp_path, goal="g", key_path=tmp_path / "k.key", + scope=["test_t.py"]) ok, _rec, msg = run_pytest_step(run, "red", "test_t.py") assert not ok @@ -221,7 +224,7 @@ def test_live_rerun_catches_broken_tests(world_copy): def test_assert_bob_done_raises_while_unproven(tmp_path): run = BobRun.start(tmp_path / "ws", goal="g", - key_path=tmp_path / "k.key") + key_path=tmp_path / "k.key", scope=["x.py"]) with pytest.raises(BobProofError): assert_bob_done(tmp_path / "ws", key_path=tmp_path / "k.key") assert run.run_id # started, never proven @@ -273,9 +276,15 @@ def test_edited_ultracode_receipt_breaks_fingerprint(bob_world, world_copy): base, _ = bob_world uc = world_copy.load("step06_security")["evidence"]["ultracode"] rec_path = (base / "uc" / "runs" / uc["run_id"] / "receipt.json") - rec_path.write_text(rec_path.read_text(encoding="utf-8") + "\n", - encoding="utf-8") # one byte changes the file sha - result = gate_check(world_copy, allow_mock=True, rerun_tests=False) + original = rec_path.read_text(encoding="utf-8") + # the ultracode home is SHARED module state — restore it afterwards so + # later tests see the untampered world + try: + rec_path.write_text(original + "\n", + encoding="utf-8") # one byte changes the file sha + result = gate_check(world_copy, allow_mock=True, rerun_tests=False) + finally: + rec_path.write_text(original, encoding="utf-8") assert not result.passed assert any("fingerprint" in e for e in result.errors) @@ -324,17 +333,22 @@ def test_deleting_run_dir_blocks_via_orphaned_marker(bob_world, tmp_path): def test_start_refuses_to_orphan_an_active_run(tmp_path): - BobRun.start(tmp_path / "ws", goal="first", key_path=tmp_path / "k.key") + BobRun.start(tmp_path / "ws", goal="first", key_path=tmp_path / "k.key", + scope=["a.py"]) try: - BobRun.start(tmp_path / "ws", goal="second", key_path=tmp_path / "k.key") + BobRun.start(tmp_path / "ws", goal="second", + key_path=tmp_path / "k.key", scope=["a.py"]) except ValueError: pass else: raise AssertionError("second start should refuse while one is active") - # force= abandons it deliberately - r2 = BobRun.start(tmp_path / "ws", goal="second", + # force= is the operator escape — it abandons WITH a durable record + r2 = BobRun.start(tmp_path / "ws", goal="second", scope=["a.py"], key_path=tmp_path / "k.key", force=True) assert r2.run_id + log = (tmp_path / "ws" / ".agent-ultra" / "bob" + / "abandoned.jsonl").read_text(encoding="utf-8") + assert "operator_abandon" in log # never silent def test_mock_panel_receipt_blocked_without_allow_mock(world_copy): @@ -361,7 +375,8 @@ def _git(ws, *args, env=None): def test_mark_pass_keeps_active_and_seal_releases(tmp_path): - run = BobRun.start(tmp_path / "ws", goal="g", key_path=tmp_path / "k.key") + run = BobRun.start(tmp_path / "ws", goal="g", key_path=tmp_path / "k.key", + scope=["a.py"]) marker = tmp_path / "ws" / ".agent-ultra" / "bob" / "ACTIVE" # seal without a recorded pass must refuse assert run.seal() is False @@ -422,3 +437,161 @@ def test_commit_deadlock_lifecycle(tmp_path, monkeypatch): p3 = _git(ws, "commit", "-m", "operator commit (no policy)", env={"AGENT_ULTRA_HOME": str(home)}) assert p3.returncode == 0 + + +# -------------------------------------------------------------------------- +# run scope: a run opened for module A cannot authorize commits to module B +# -------------------------------------------------------------------------- + +def test_scope_start_without_scope_is_rejected(tmp_path): + with pytest.raises(ValueError, match="must declare its target scope"): + BobRun.start(tmp_path / "ws", goal="g", key_path=tmp_path / "k.key") + assert BobRun.active(tmp_path / "ws") is None # nothing half-started + + +def test_scope_infrastructure_entries_rejected(tmp_path): + for bad in (".git/hooks/pre-commit", ".agent-ultra/bob/ACTIVE", + "sub/.git/hooks", ".agent-ultra"): + with pytest.raises(ValueError, match="enforcement infrastructure"): + BobRun.start(tmp_path / "ws", goal="g", + key_path=tmp_path / "k.key", scope=[bad]) + assert scope_entry_is_infra(".git/hooks/pre-commit") + assert scope_entry_is_infra(".agent-ultra/bob/runs") + assert not scope_entry_is_infra("src/app.py") + + +def test_scope_matching(): + assert scope_contains(["mods/a"], "mods/a/impl.py") + assert scope_contains(["src/app.py"], "src/app.py") + assert scope_contains(["src/*.py"], "src/main.py") + assert not scope_contains(["mods/a"], "mods/ab/x.py") # not a prefix + assert not scope_contains(["mods/a"], "mods/b/x.py") + assert not scope_contains([], "anything.py") # empty = nothing + + +def test_scope_staged_file_outside_scope_blocks_gate(world_copy): + """The passing world's run was scoped to its test+impl files; a staged + file elsewhere must block even though every receipt is intact.""" + res = gate_check(world_copy, rerun_tests=False, allow_mock=True, + staged_files=["unrelated/other.py"]) + assert not res.passed + assert any("out-of-scope" in e for e in res.errors) + + +def test_scope_staged_in_scope_files_pass(world_copy): + scope = world_copy.scope() + assert scope, "run_bob must declare the generated files as scope" + res = gate_check(world_copy, rerun_tests=False, allow_mock=True, + staged_files=list(scope)) + assert res.passed, res.errors + + +def test_scope_add_is_validated_and_logged(world_copy): + res = gate_check(world_copy, rerun_tests=False, allow_mock=True, + staged_files=["docs/notes.md"]) + assert not res.passed # outside scope before + added, errors = world_copy.add_scope(["docs/notes.md", + ".git/hooks/pre-commit"]) + assert added == ["docs/notes.md"] + assert errors and "infrastructure" in errors[0] + log = (world_copy.run_dir / "scope-log.jsonl").read_text(encoding="utf-8") + assert "docs/notes.md" in log # expansion is durable + res = gate_check(world_copy, rerun_tests=False, allow_mock=True, + staged_files=["docs/notes.md"]) + assert res.passed, res.errors # explicit expansion works + + +def test_scope_stripped_from_run_json_fails_closed(world_copy): + """run.json is not HMAC-protected; deleting the scope must not unbind + the run — a no-scope run authorizes nothing.""" + meta = json.loads((world_copy.run_dir / "run.json").read_text( + encoding="utf-8")) + meta.pop("scope", None) + (world_copy.run_dir / "run.json").write_text( + json.dumps(meta), encoding="utf-8") + res = gate_check(world_copy, rerun_tests=False, allow_mock=True, + staged_files=["anything.py"]) + assert not res.passed + assert any("scope is mandatory" in e for e in res.errors) + assert any("out-of-scope" in e for e in res.errors) + + +def test_scope_operator_unbounded_is_explicit_and_logged(tmp_path, capsys): + # combined with a scope: contradiction, rejected + with pytest.raises(ValueError, match="cannot be combined"): + BobRun.start(tmp_path / "w1", goal="g", key_path=tmp_path / "k.key", + scope=["a.py"], operator_unbounded=True) + run = BobRun.start(tmp_path / "w2", goal="g", + key_path=tmp_path / "k.key", operator_unbounded=True) + assert run.unbounded() and run.scope() == [] + assert "OPERATOR-UNBOUNDED" in capsys.readouterr().err # loud + log = (run.run_dir / "scope-log.jsonl").read_text(encoding="utf-8") + assert "operator_unbounded" in log # durable + # infrastructure still can never be scope-added, even unbounded + added, errors = run.add_scope([".git/hooks/pre-commit"]) + assert not added and errors + + +def test_scope_operator_unbounded_has_no_cli_flag(): + """The unbounded escape is a Python-API parameter only: no CLI verb + exposes it, so an agent driving the CLI cannot reach it.""" + from agent_ultra.adapters import cli + src = open(cli.__file__, encoding="utf-8").read() + assert "operator-unbounded" not in src + assert "operator_unbounded" not in src + + +# -------------------------------------------------------------------------- +# abandon is explicit, operator-only, logged; escapes are agent-unreachable +# -------------------------------------------------------------------------- + +def test_abandon_is_operator_only_and_logged(tmp_path): + run = BobRun.start(tmp_path / "ws", goal="g", + key_path=tmp_path / "k.key", scope=["a.py"]) + with pytest.raises(ValueError, match="operator"): + BobRun.abandon(tmp_path / "ws") # agent path: refused + name = BobRun.abandon(tmp_path / "ws", operator=True, reason="stuck") + assert name == run.run_id + assert BobRun.active(tmp_path / "ws") is None # released + log = (tmp_path / "ws" / ".agent-ultra" / "bob" + / "abandoned.jsonl").read_text(encoding="utf-8") + assert run.run_id in log and "stuck" in log + assert (run.run_dir / "abandoned.json").exists() + # a fresh run may start after the explicit abandon + r2 = BobRun.start(tmp_path / "ws", goal="next", + key_path=tmp_path / "k.key", scope=["a.py"]) + assert BobRun.active(tmp_path / "ws").run_id == r2.run_id + + +def test_cli_abandon_requires_operator_flag(tmp_path, monkeypatch): + from agent_ultra.adapters import cli + BobRun.start(tmp_path / "ws", goal="g", key_path=tmp_path / "k.key", + scope=["a.py"]) + monkeypatch.setenv("AGENT_ULTRA_HOME", str(tmp_path / "home")) + assert cli.main(["bob", "abandon", + "--workspace", str(tmp_path / "ws")]) == 1 + assert BobRun.active(tmp_path / "ws") is not None # still enforced + assert cli.main(["bob", "abandon", "--operator-abandon", + "--workspace", str(tmp_path / "ws")]) == 0 + assert BobRun.active(tmp_path / "ws") is None + + +def test_broker_denies_operator_escape_flags(): + """An agent-driven shell (broker-mediated) cannot auto-run the + operator escapes: they classify DANGEROUS.""" + from agent_ultra.broker.broker import classify as _classify + for cmd in ("agent-ultra bob abandon --operator-abandon", + "agent-ultra bob run \"t\" --operator-force", + "python x.py --operator-unbounded"): + tier, why = _classify(cmd) + assert tier == "dangerous", (cmd, tier, why) + assert "operator-only" in why + + +def test_cloned_fanout_is_rejected(): + from agent_ultra.bob.pipeline import cloned_fanout_error + same = [{"status": "ok", "output_sha256": "aaa"} for _ in range(4)] + assert "IDENTICAL" in cloned_fanout_error("r1", same) + mixed = [{"status": "ok", "output_sha256": h} for h in ("a", "b", "a")] + assert cloned_fanout_error("r1", mixed) is None # pair is legal + assert cloned_fanout_error("r1", same[:1]) is None # single agent diff --git a/tests/test_bob_surgical.py b/tests/test_bob_surgical.py new file mode 100644 index 0000000..ff862e1 --- /dev/null +++ b/tests/test_bob_surgical.py @@ -0,0 +1,234 @@ +"""bob surgical lane: entry criteria, C1 mode-downgrade defense, gate. + +The lane's contract: INERT edits only, narrower than full mode on every +axis, and anything code-shaped routes to the FULL gate — surgical can never +launder a code change past Bob. +""" + +import io +import json + +import pytest + +from agent_ultra.bob import ( + BobRun, gate_check, run_surgical, surgical_gate_check, +) +from agent_ultra.bob.surgical import ( + surgical_ext_allowed, surgical_file_denied, staged_set_qualifies, + SURGICAL_MAX_DIFF_LINES, +) + + +# -------------------------------------------------------------------------- +# entry criteria (pure) +# -------------------------------------------------------------------------- + +def test_inert_prose_and_dotfiles_qualify(): + for p in ("README.md", "docs/guide.rst", "notes.txt", ".gitignore", + ".editorconfig"): + assert surgical_ext_allowed(p), p + assert not surgical_file_denied(p), p + + +def test_code_never_qualifies(): + for p in ("app.py", "src/x.js", "lib.rs", "run.sh", "mod.go"): + assert not surgical_ext_allowed(p), p + + +def test_structured_config_needs_operator_opt_in(monkeypatch): + monkeypatch.delenv("AGENT_ULTRA_SURGICAL_ALLOW_CONFIG", raising=False) + assert not surgical_ext_allowed("settings.yaml") + monkeypatch.setenv("AGENT_ULTRA_SURGICAL_ALLOW_CONFIG", "1") + assert surgical_ext_allowed("settings.yaml") + # even opted in, denied names stay denied + assert not surgical_ext_allowed("package.json") + assert not surgical_ext_allowed(".github/workflows/ci.yml") + + +def test_executing_configs_are_denied(): + for p in ("package.json", "pyproject.toml", "Makefile", "Dockerfile", + ".github/workflows/ci.yml", ".pre-commit-config.yaml", + "docker-compose.yml", "jest.config.json", ".mocharc.json", + "conftest.py", ".env"): + assert surgical_file_denied(p), p + + +def test_docs_about_tools_are_not_denied(): + # path-aware matching, not substring: a doc ABOUT Dockerfiles is a doc + for p in ("docs/Dockerfile-explained.md", "docs/makefile-tips.md"): + assert not surgical_file_denied(p), p + assert surgical_ext_allowed(p), p + + +def test_enforcement_infra_is_never_surgical(): + assert surgical_file_denied(".git/hooks/pre-commit") + assert surgical_file_denied(".agent-ultra/bob/ACTIVE") + + +def test_staged_set_qualification_is_strict(): + declared = ["README.md"] + assert staged_set_qualifies(["README.md"], declared) + assert staged_set_qualifies(["README.md", ".agent-ultra/bob/x.json"], + declared) # machinery rides along + assert not staged_set_qualifies(["app.py"], declared) # code + assert not staged_set_qualifies(["docs/x.md"], declared) # undeclared + + +# -------------------------------------------------------------------------- +# start: objective entry criteria, validated up front +# -------------------------------------------------------------------------- + +def test_surgical_start_rejects_code_and_denied(tmp_path): + with pytest.raises(ValueError, match="rejects code file"): + BobRun.start(tmp_path / "ws", goal="g", key_path=tmp_path / "k.key", + mode="surgical", surgical_files=["app.py"]) + with pytest.raises(ValueError, match="denied_path"): + BobRun.start(tmp_path / "ws", goal="g", key_path=tmp_path / "k.key", + mode="surgical", surgical_files=["package.json"]) + with pytest.raises(ValueError, match="surgical_files is required"): + BobRun.start(tmp_path / "ws", goal="g", key_path=tmp_path / "k.key", + mode="surgical") + with pytest.raises(ValueError, match="NEVER valid for surgical"): + BobRun.start(tmp_path / "ws", goal="g", key_path=tmp_path / "k.key", + mode="surgical", surgical_files=["a.md"], + operator_unbounded=True) + + +def test_surgical_scope_is_exactly_its_files(tmp_path): + run = BobRun.start(tmp_path / "ws", goal="g", + key_path=tmp_path / "k.key", + mode="surgical", surgical_files=["docs/a.md"]) + assert run.mode() == "surgical" + assert run.scope() == ["docs/a.md"] + # scope-add on a surgical run rejects risky paths, accepts inert ones + added, errors = run.add_scope(["evil.py"]) + assert not added and any("inert" in e for e in errors) + added, errors = run.add_scope(["docs/b.md"]) + assert added == ["docs/b.md"] and not errors + + +# -------------------------------------------------------------------------- +# the full lane end to end (mock, offline) +# -------------------------------------------------------------------------- + +@pytest.fixture() +def surgical_world(tmp_path, monkeypatch): + monkeypatch.setenv("AGENT_ULTRA_HOME", str(tmp_path / "home")) + ws = tmp_path / "ws" + ws.mkdir() + (ws / "README.md").write_text("# hello\nfixed wording\n", + encoding="utf-8") + outcome = run_surgical("fix readme wording", ws, ["README.md"], + mock=True, key_path=tmp_path / "k.key", + ultracode_home=tmp_path / "uc", + interactive=False, seal_on_pass=False, + out=io.StringIO()) + assert outcome.passed, f"surgical mock must pass: {outcome.blocked_at}" + run = BobRun.active(ws, key_path=tmp_path / "k.key") + assert run is not None + return ws, run + + +def test_surgical_mock_lane_passes(surgical_world): + ws, run = surgical_world + res = surgical_gate_check(run, allow_mock=True, staged_files=[]) + assert res.passed, res.errors + + +def test_surgical_gate_blocks_without_review(surgical_world): + ws, run = surgical_world + run.receipt_path("step_surgical_review").unlink() + res = surgical_gate_check(run, allow_mock=True, staged_files=[]) + assert not res.passed + assert any("step_surgical_review" in e and "missing" in e + for e in res.errors) + + +def test_surgical_quiz_passed_without_response_blocks(surgical_world): + ws, run = surgical_world + run.receipt_path("step_surgical_quiz").unlink() + run.write("step_surgical_quiz", + {"questions": ["q"], "outcome": "passed", + "operator_response": ""}, writer="agent", mock=True) + res = surgical_gate_check(run, allow_mock=True, staged_files=[]) + assert not res.passed + assert any("self-graded" in e or "operator response" in e + for e in res.errors) + + +def test_surgical_staged_undeclared_or_risky_blocks(surgical_world): + ws, run = surgical_world + res = surgical_gate_check(run, allow_mock=True, + staged_files=["other.md"]) + assert not res.passed + assert any("not declared" in e for e in res.errors) + res = surgical_gate_check(run, allow_mock=True, + staged_files=["app.py"]) + assert not res.passed + assert any("not an inert type" in e for e in res.errors) + res = surgical_gate_check(run, allow_mock=True, + staged_files=["package.json"]) + assert not res.passed + assert any("denied_path" in e for e in res.errors) + + +def test_surgical_staleness_blocks_post_review_edit(surgical_world): + ws, run = surgical_world + (ws / "README.md").write_text("# edited AFTER the review\n", + encoding="utf-8") + res = surgical_gate_check(run, allow_mock=True, staged_files=[]) + assert not res.passed + assert any("stale" in e for e in res.errors) + + +# -------------------------------------------------------------------------- +# C1: mode downgrade cannot launder code past the full gate +# -------------------------------------------------------------------------- + +def test_forged_surgical_mode_with_code_routes_to_full_gate(surgical_world): + """Flipping run.json's mode is a REQUEST, not a trusted assertion: with + code staged, gate_check routes to the FULL gate, which demands the full + receipt chain this run never produced.""" + ws, run = surgical_world + res = gate_check(run, rerun_tests=False, allow_mock=True, + staged_files=["app.py"]) + assert not res.passed + # blocked by FULL-pipeline demands, not by a surgical pass + assert any("step02_red" in e for e in res.errors) + assert any("step08_panel" in e for e in res.errors) + + +def test_qualifying_surgical_run_routes_to_surgical_gate(surgical_world): + ws, run = surgical_world + res = gate_check(run, rerun_tests=False, allow_mock=True, + staged_files=["README.md"]) + assert res.passed, res.errors + assert "step_surgical_review" in res.checked + + +def test_full_run_never_routes_to_surgical(tmp_path): + """A FULL-mode run with only inert files staged still gets the full + gate — surgical is opt-in at start, never an automatic downgrade.""" + run = BobRun.start(tmp_path / "ws", goal="g", + key_path=tmp_path / "k.key", scope=["README.md"]) + res = gate_check(run, rerun_tests=False, staged_files=["README.md"]) + assert not res.passed + assert any("step02_red" in e for e in res.errors) + + +def test_diff_budget_blocks_oversized_edit(tmp_path, monkeypatch): + """An in-repo surgical run with a staged diff over the line budget + blocks at the gate (fail-closed budget check).""" + import subprocess + ws = tmp_path / "ws" + ws.mkdir() + subprocess.run(["git", "init", "-q"], cwd=ws, check=True) + big = "\n".join(f"line {i}" for i in range(SURGICAL_MAX_DIFF_LINES + 20)) + (ws / "README.md").write_text(big + "\n", encoding="utf-8") + subprocess.run(["git", "add", "README.md"], cwd=ws, check=True) + run = BobRun.start(ws, goal="g", key_path=tmp_path / "k.key", + mode="surgical", surgical_files=["README.md"]) + res = surgical_gate_check(run, allow_mock=True, + staged_files=["README.md"]) + assert not res.passed + assert any("budget" in e for e in res.errors)