diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b6e941..be51e33 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,49 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Fixed +- **bob commit-lifecycle deadlock (mark/seal split).** Sealing the run inside + a pre-commit gate released the ACTIVE marker *before* the commit landed, so + a require-run policy blocked the very commit the gate had just validated + (and a failed commit left the run wrongly closed). The gate now splits the + lifecycle: `bob gate --mark-pass` (pre-commit) validates and records + `gate-pass.json` but keeps ACTIVE; the new `bob seal` (post-commit) + releases ACTIVE only when a gate pass is recorded, stamping the landed + commit hash. New `bob hook-install` writes all three hooks (pre-commit, + pre-merge-commit, post-commit); `AGENT_ULTRA_REQUIRE_RUN=1` makes the + pipeline mandatory for hook-level commits. `--complete-on-pass` remains + for closures outside a commit. `run_bob(..., seal_on_pass=False)` selects + the git-workflow lifecycle from Python. Regression test drives the full + cycle through real git: proven commit lands, post-commit seals, an + unproven follow-up commit is blocked. **If you had wired `bob gate + --complete-on-pass` into a pre-commit hook, replace it: run + `agent-ultra bob hook-install` to regenerate the hooks.** + ### Added +- **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 + panel, the broker's risk tiers, and signed receipts into one loop. Each + gated step leaves a hash-chained (`prev_sha256`), HMAC-signed step receipt + written from real execution: RED/GREEN receipts come from the pytest + runner's actual output (`writer="system"`); the two fan-out steps are + cross-checked against ultracode's own checksummed run receipts; ULTRA is + proven by the panel execution receipt (a self-review is not a panel); a + `passed` quiz needs a captured operator response. The commit/report gate + re-runs pytest live, re-hashes covered files (staleness), demands coverage + for staged files, and fail-closes on anything it cannot verify — a + skipped, fabricated, edited, re-ordered, or hand-authored step blocks. + Mock mode (`--mock`, no API key) swaps only the model content for a + bundled sample task; pytest, ultracode, the panel, and the gate all really + execute. CLI: `agent-ultra bob run|gate|status` (alias: `agent-ultra + build`); Python: `run_bob` / `gate_check` / `assert_bob_done`. `doctor` + and `demo` prove the pass AND three blocked-fraud scenarios; see + [docs/bob-the-builder.md](docs/bob-the-builder.md). Each cross-checked + artifact (the ultracode run receipt's file bytes, the panel receipt's + task/artifact hashes) is bound into the signed chain, and model-authored + file paths are validated against workspace escape; the docs state the + trust boundary honestly (no host middleware, unlike the private reference). + 29 offline tests. - **ultracode** (`agent_ultra.ultracode`) — deterministic multi-agent workflow engine. Workflows are plain Python modules (`META` + `async def run(wf)`) that fan work across bounded agents via `wf.agent` (one model call, optional diff --git a/README.md b/README.md index c1ea5a5..92873ac 100644 --- a/README.md +++ b/README.md @@ -72,11 +72,12 @@ $ agent-ultra doctor [PASS] panel demo: 3 findings, 3 accepted, decision produced [PASS] ULTRA demo: build->test->panel->fix loop->proof artifacts [PASS] ultracode: fan-out -> journal -> resume (cached) -> valid receipt + [PASS] bob: 10-step pipeline gate passes; skipped/faked steps block [PASS] receipts bus: HMAC authenticity, forgery rejected, audit chain [PASS] verifier: refute-first default, engine re-check, refuted!=gate [PASS] secret hygiene: redaction active, no secrets in artifacts [PASS] worker layer: router (default) available -Result: 12 pass, 1 warn, 0 fail +Result: 13 pass, 1 warn, 0 fail $ agent-ultra demo [ok] panel: 3 findings, 3 accepted -> Fix the empty-token acceptance before shipping... @@ -85,6 +86,7 @@ $ agent-ultra demo [ok] receipts: genuine authentic, forgery rejected, audit chain ok=True [ok] verifier: no-evidence refuted, re-check confirmed (cannot satisfy a gate) [ok] ultracode: 4 agents fan out -> COMPLETE -> receipt valid=True -> resume replays 4 cached (0 calls) +[ok] bob: 10-step pipeline gate passed=True; skipped step blocks=True, fake fan-out blocks=True, doctored panel blocks=True DEMO PASSED — the full loop works on this machine. $ agent-ultra --mock panel "Is this auth service safe?" --lenses security,correctness,failure-modes @@ -147,6 +149,50 @@ Commands: `ultracode run ` · `list` · `status` · `resume `. Bundled workflows: **smoke** (fan-out + pipeline) and **review** (finders → skeptic votes → synthesis). Full guide: [docs/ultracode.md](docs/ultracode.md). +## bob — the 10-step enforced build pipeline + +**bob** composes the whole kit into one enforced build loop (alias: +`agent-ultra build`): + +``` +SPEC -> RED -> GREEN -> REFACTOR -> CODE-QUALITY -> SECURITY-FANOUT + -> WORKFLOW -> ULTRA -> QUIZ -> COMMIT +``` + +Every gated step leaves a **hash-chained, HMAC-signed receipt** written from +real execution — the pytest runner's actual output (RED/GREEN), ultracode's +checksummed run receipts (the two fan-out steps), the panel's execution +receipt (ULTRA). The commit gate re-derives what it can (a live pytest +re-run, file staleness hashes) and cross-checks the rest. You cannot *claim* +a step ran: + +- a **skipped step** blocks (missing receipt + broken chain), +- a **fabricated fan-out** blocks (no ultracode run receipt backs the claim), +- a **doctored panel receipt** blocks (its checksum no longer matches), +- an **edited or hand-authored receipt** blocks (integrity/HMAC fail). + +```console +$ agent-ultra bob run "add a slugify helper" --mock # offline, no key +bob run 20260708T154002Z — add a slugify helper [mock: sample task, no key] + [step01_spec] slugify(text) lowercases the input, replaces runs of non-al... + [step02_red] RED: 4 failing / 4 collected + [step03_green] GREEN: 4 passing + [step04_refactor] no edits (hashes re-pinned) + [step05_codequality] checklist recorded + [step06_security] 4 agents, 4 finding(s) + [step07_workflow] 5 agents over 5 dimensions + [step08_panel] decision: Fix the empty-token acceptance before shipping... + [step09_quiz] outcome=skipped (3 questions) + +gate PASSED — receipt chain validates (9 receipts). +``` + +Mock mode swaps only the *model content* for a bundled sample task — pytest, +ultracode, the panel, and the gate all really execute, which is what makes +the no-key demo honest. Commands: `bob run ` · `bob gate` +(pre-commit chokepoint; `--complete-on-pass` seals the run) · `bob status`. +Full guide: [docs/bob-the-builder.md](docs/bob-the-builder.md). + ## What's in the box | module | what it does | @@ -154,6 +200,7 @@ skeptic votes → synthesis). Full guide: [docs/ultracode.md](docs/ultracode.md) | **panel** | Adversarial review: parallel critic *lenses* → steelman → judge cross-exam → synthesis. Verdicts: `real_now` / `real_later` / `theoretical` / `wrong`. Panel agents are roles, not models — one healthy route runs a whole panel. | | **ultra_loop** | build → test → panel → classify → fix → re-test → re-panel → **ship gate**. Red tests stop before the panel; low-context panels are refused; only proof-gated work ships. | | **ultracode** | Deterministic multi-agent workflows: `META` + `async run(wf)` scripts fan work across bounded agents (`parallel`/`pipeline`) under hard budgets, with a resumable journal, a checksummed receipt, and a terminal-safe status card. Fan-out → journal → resume → receipt → status. | +| **bob** | The 10-step enforced build pipeline (SPEC→RED→GREEN→…→COMMIT). Each gated step leaves a hash-chained, HMAC-signed receipt from real execution; the commit gate re-runs tests live and cross-checks ultracode/panel receipts. Skipped, faked, or edited steps block. | | **broker** | Every model-authored command classified SAFE / ELEVATED / DANGEROUS, ledgered, and executed only if its tier allows. DANGEROUS with no approval path is **denied by default**. | | **proof** | "Done" requires recorded evidence. Accepted findings become gates; `assert_shippable()` raises on unsupported completion claims. | | **evidence** | Bounded source gathering with secret redaction; low-context detection. | @@ -248,6 +295,7 @@ Full guide: [docs/adapter-guide.md](docs/adapter-guide.md). [architecture](docs/architecture.md) · [panel](docs/panel.md) · [ULTRA loop](docs/ultra-loop.md) · [ultracode](docs/ultracode.md) · +[bob pipeline](docs/bob-the-builder.md) · [command broker](docs/command-broker.md) · [proof gates](docs/proof-gates.md) · [adapter guide](docs/adapter-guide.md) · [security](docs/security.md) · [troubleshooting](docs/troubleshooting.md) · diff --git a/docs/bob-the-builder.md b/docs/bob-the-builder.md new file mode 100644 index 0000000..18bf8f4 --- /dev/null +++ b/docs/bob-the-builder.md @@ -0,0 +1,176 @@ +# bob — the 10-step enforced build pipeline + +`bob` composes the kit's pieces — ultracode fan-out, the adversarial panel, +the command broker's risk tiers, and signed receipts — into one enforced +build loop. The doctrine: + +``` +real execution -> receipt -> gate validates the chain -> commit/report +``` + +Ten steps, six of them gated. You cannot claim a step ran: the gate +re-derives what it can and cross-checks the rest, and every receipt is +hash-chained and HMAC-signed, so skipping, faking, editing, re-ordering, or +hand-authoring a step blocks the commit. + +## The pipeline + +| # | step | receipt | how it's enforced | +|---|------|---------|-------------------| +| 1 | SPEC | agent | one-paragraph spec recorded (judgment step) | +| 2 | RED | **system** | written by the pytest runner from actual output; needs ≥1 FAILED, 0 collection ERRORs | +| 3 | GREEN | **system** | all tests pass; every RED test id must appear in the GREEN run | +| 4 | REFACTOR | agent | judgment step; edited files re-pin their hashes | +| 5 | CODE-QUALITY | agent | judgment step; checklist recorded | +| 6 | SECURITY-FANOUT | **engine** | cross-checked: points at an ultracode run whose own checksummed receipt must re-verify, with ≥1 real agent call | +| 7 | WORKFLOW | **engine** | same cross-check + non-empty review dimensions | +| 8 | ULTRA | **engine** | the panel writes its execution receipt from real model calls — a self-review is not a panel | +| 9 | QUIZ | agent | `passed` requires a captured operator response; otherwise the honest outcome is `skipped` | +| 10 | COMMIT | gate | `gate-pass.json` written by the gate itself; ACTIVE marker released | + +`writer` records who produced each receipt's evidence. The gate trusts +nothing it cannot re-derive (live pytest re-run, file staleness hashes) or +cross-check (ultracode receipts, the panel execution receipt). + +## Quick start (no API key) + +```bash +agent-ultra bob run "add a slugify helper" --mock +``` + +Mock mode swaps only the MODEL CONTENT (spec text, tests, implementation) +for a bundled sample task. Everything else is real: pytest runs RED and +GREEN as subprocesses, ultracode fans out on the mock route and writes its +checksummed receipts, the panel executes and writes its execution receipt, +and the gate validates the whole chain (including a live pytest re-run). +`agent-ultra build run ...` is the same command under its alias. + +On a real endpoint (after `agent-ultra init` + key env var), drop `--mock`: +the spec, tests, and implementation come from your configured model. If the +model's tests don't genuinely fail at RED or its implementation doesn't +genuinely pass at GREEN, the run ends **blocked** — the pipeline never +pretends. + +## The receipt chain + +Receipts live in `/.agent-ultra/bob/runs//`, one JSON per +step, each carrying: + +- `receipt_sha256` — integrity: the body was not edited after writing; +- `receipt_hmac` — authenticity: written by something holding the + per-install key (created at first use under the agent-ultra home, never + inside the workspace); +- `prev_sha256` — the sha256 of the receipt before it. Editing or replacing + ANY earlier receipt breaks every later one; +- `files` — sha256 of every file the step covered. A covered file that + changes afterward is **stale** and blocks until the owning step re-runs; +- `mock` — set on mock-route evidence. The gate only accepts mock receipts + with `--allow-mock` (demo mode), never by default. + +## The gate (commit/report chokepoint) + +```bash +agent-ultra bob gate # validate the active run's chain +agent-ultra bob gate --mark-pass # validate + record the pass, keep ACTIVE +agent-ultra bob seal # release ACTIVE (needs a recorded pass) +agent-ultra bob gate --complete-on-pass # mark + seal in one step (no commit involved) +agent-ultra bob status # which steps have receipts +``` + +The gate fail-closes: a check it cannot perform (missing receipt, unreadable +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). + +### Commit lifecycle: mark at pre-commit, seal at post-commit + +Sealing inside pre-commit is a deadlock: the pass would release ACTIVE +*before* the commit lands, so any require-run policy then blocks the very +commit the gate just validated (and if the commit fails, the run is wrongly +closed). The split lifecycle avoids it: + +1. **pre-commit** — `bob gate --hook --mark-pass`: validates the chain; on + pass it writes `gate-pass.json` but **keeps ACTIVE**. +2. the commit lands. +3. **post-commit** — `bob seal --if-passed`: releases ACTIVE and stamps the + landed commit hash into `gate-pass.json`. `seal` refuses when no gate + pass is recorded — a run can never be sealed unproven. + +Install all three hooks (pre-commit, pre-merge-commit — git fires that one +for merge commits — and post-commit) in one command: + +```bash +agent-ultra bob hook-install +``` + +With no active run the gate exits 0 immediately (non-interference). To make +the pipeline mandatory — every commit needs an active, proven run — export +`AGENT_ULTRA_REQUIRE_RUN=1` in the agent's environment: hook-mode gates then +block commits that have no run at all. + +`--complete-on-pass` remains for closures that happen *outside* a commit +(the mock demo's report path, an operator retiring a run manually). + +From Python, `assert_bob_done(workspace)` raises `BobProofError` when a run +is active and unproven — wire it in front of your agent's completion claims: + +```python +from agent_ultra.bob import assert_bob_done, BobProofError +assert_bob_done(".") # raises unless the receipt chain validates +``` + +`assert_bob_done` passes `staged_files=[]` (at claim time nothing is staged), +so it validates the receipt chain but does **not** run the staged-file +coverage check — that lives in the `agent-ultra bob gate` pre-commit path, +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. + +## Trust boundary (read this) + +bob resists the failure modes an agent actually falls into: **skipping** the +expensive steps, **claiming** a step ran without running it, **borrowing** a +prior run's fan-out or panel, and **editing** a receipt after the fact. It +enforces those by re-deriving what it can (a live pytest re-run) and binding +every cross-checked artifact — the ultracode run's receipt bytes and the +panel receipt's task/artifact hashes — into an HMAC-signed, hash-chained +receipt. Change any pinned artifact and the signed chain breaks. + +What bob does **not** claim: to stop a fully adversarial process that has +arbitrary code execution *and* the per-install signing key. The key lives at +`/bob.key`, readable by the same process the gate runs in, +so an attacker willing to genuinely fabricate every artifact end-to-end (run +the real fan-out over decoy code, forge matching signed receipts) can pass. +The private reference implementation closes this by writing the delegation +and clarify logs **host-side**, outside the agent's reach; the public port +has no host middleware, so its guarantee is "honest work is cheap, faking it +convincingly is not," not "unforgeable." Treat the gate as a strong +tripwire against lazy or accidental skipping, not a cryptographic proof +against a determined local adversary. + +## What the proofs look like + +`agent-ultra doctor` and `agent-ultra demo` both run the pipeline offline +and then attack it three ways; all three must block: + +1. **skipped step** — a gated receipt is removed → missing receipt + broken + chain; +2. **fabricated fan-out** — a re-signed step-6 receipt names an invented + ultracode run → no run receipt on disk backs the claim; +3. **doctored panel receipt** — one field changed in the panel's execution + receipt → its checksum no longer matches. + +The offline test suite (`pytest -q`) additionally proves: unsigned receipts +are forgeries, a wrong key fails closed, re-ordered/removed receipts break +linkage, `passed` quiz outcomes need an operator response, stale files +block, and the live re-run catches a state that fails its own tests. + +## Troubleshooting + +| symptom | fix | +|---------|-----| +| `gate BLOCKED ... receipt missing` | That step never ran (or its receipt was deleted). Re-run the step; the chain rebuilds from there. | +| `stale: changed after its last covering receipt` | Re-run the step that owns the file (usually GREEN) so hashes re-pin. | +| `authenticity FAIL` on every receipt | The per-install key changed (e.g. new `AGENT_ULTRA_HOME`). Finish runs with the key that started them. | +| `receipt is marked mock` | You gated a demo run without `--allow-mock`. Mock evidence never satisfies a real gate. | +| blocked at RED on a real route | The model's tests didn't fail cleanly (or errored on import). Inspect the run dir's `step02_red.json` `output_tail`. | diff --git a/src/agent_ultra/adapters/cli.py b/src/agent_ultra/adapters/cli.py index e5c5060..81d626f 100644 --- a/src/agent_ultra/adapters/cli.py +++ b/src/agent_ultra/adapters/cli.py @@ -287,6 +287,102 @@ def _infer_workflow_for_run(args) -> str: return "" +# -------------------------------------------------------------------------- +# bob — the 10-step enforced build pipeline +# -------------------------------------------------------------------------- + +def _bob_run(args) -> int: + from ..bob import run_bob + pool = None + if not args.mock: + 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) + return 0 if outcome.passed else 1 + + +def _bob_gate(args) -> int: + from ..bob import BobRun, gate_check + run = BobRun.active(args.workspace) + if run is None: + # non-interference by default; AGENT_ULTRA_REQUIRE_RUN=1 makes the + # pipeline mandatory for hook-level commits (mirrors the private + # reference's require-run policy). + if args.hook and os.environ.get("AGENT_ULTRA_REQUIRE_RUN") == "1": + print("BLOCKED: no active bob run and AGENT_ULTRA_REQUIRE_RUN=1 " + "— the pipeline is mandatory; start a run first", + file=sys.stderr) + return 1 + print("no active bob run — nothing to gate (non-interference)") + return 0 + result = gate_check(run, allow_mock=args.allow_mock, + rerun_tests=not args.no_rerun) + if result.passed: + if args.complete_on_pass: + run.complete(result.to_dict()) + elif args.mark_pass: + run.mark_pass(result.to_dict()) + print(f"gate PASSED for run {run.run_id}") + for step, summary in result.checked.items(): + print(f" [ok] {step}: {summary}") + return 0 + print(f"gate BLOCKED run {run.run_id} — {len(result.errors)} failure(s):") + for e in result.errors: + print(f" ! {e}") + return 1 + + +def _bob_seal(args) -> int: + from ..bob import BobRun + run = BobRun.active(args.workspace) + if run is None: + if args.if_passed: + return 0 + print("no active bob run", file=sys.stderr) + return 2 + if run.seal(): + print(f"SEALED — run {run.run_id}") + return 0 + if args.if_passed: + return 0 + print(f"NOT SEALED — run {run.run_id} has no recorded gate pass " + "(run `agent-ultra bob gate --mark-pass` first)", file=sys.stderr) + return 1 + + +def _bob_hooks(args) -> int: + from ..bob.pipeline import install_hooks + try: + hook = install_hooks(args.workspace, python=args.python) + except FileNotFoundError as e: + print(str(e), file=sys.stderr) + return 2 + print(f"installed: {hook} (+ pre-merge-commit, post-commit)") + print("lifecycle: pre-commit validates + marks the pass; post-commit " + "seals after the commit lands.") + return 0 + + +def _bob_status(args) -> int: + from ..bob import BobRun, STEPS, GATED_STEPS + run = BobRun.active(args.workspace) + if run is None: + print("no active bob run") + return 0 + print(f"run {run.run_id} — {run.goal()}") + have = {r.get('step'): r for r in run.chain()} + for step in STEPS: + if step == "step10_commit": + continue + mark = ("done" if step in have + else "REQUIRED" if step in GATED_STEPS else "-") + print(f" {step:<20} {mark}") + missing = [s for s in GATED_STEPS if s not in have] + print(f"\n{'ready for `agent-ultra bob gate`' if not missing else 'missing: ' + ', '.join(missing)}") + return 0 + + def _receipts_bus(args): from ..receipts_bus import ReceiptsBus return ReceiptsBus(db_path=args.db or "", key_path=args.key or "") @@ -438,6 +534,68 @@ def _uc_common(sp): _uc_common(ucrs) ucrs.set_defaults(func=_ultracode_resume) + # -- bob: the 10-step enforced build pipeline -------------------------- + for alias in ("bob", "build"): + kwargs = {"help": "10-step enforced build pipeline: every step " + "leaves a receipt; commit/report only when " + "the chain validates (alias: build)"} \ + if alias == "bob" else {} + bp = sub.add_parser(alias, **kwargs) + bsub = bp.add_subparsers(dest="bob_cmd", required=True) + + br = bsub.add_parser("run", help="run the whole pipeline on a task") + br.add_argument("task") + 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("--mock", action="store_true", + help="offline demo on the bundled sample task (no key)") + br.add_argument("--mode", default="single", choices=["single", "mixed"]) + br.add_argument("--config", default="") + br.set_defaults(func=_bob_run) + + bg = bsub.add_parser("gate", help="validate the receipt chain of the " + "active run (pre-commit chokepoint)") + bg.add_argument("--workspace", default=".") + bg.add_argument("--allow-mock", action="store_true", + help="accept mock-route receipts (demo runs only)") + bg.add_argument("--no-rerun", action="store_true", + help="skip the live pytest re-run (chain checks only)") + bg.add_argument("--mark-pass", action="store_true", + help="record gate-pass.json on success, keep ACTIVE " + "(pre-commit hook mode — seal happens post-commit)") + bg.add_argument("--complete-on-pass", action="store_true", + help="mark pass AND release ACTIVE on success " + "(manual close outside a commit)") + bg.add_argument("--hook", action="store_true", + help="hook mode: honor AGENT_ULTRA_REQUIRE_RUN=1 " + "when no run is active") + bg.set_defaults(func=_bob_gate) + + bl = bsub.add_parser("seal", help="release ACTIVE after the commit " + "landed (post-commit hook; requires " + "a recorded gate pass)") + bl.add_argument("--workspace", default=".") + bl.add_argument("--if-passed", action="store_true", + help="exit 0 silently when no run is active or the " + "gate has not passed (hook mode)") + bl.set_defaults(func=_bob_seal) + + bh = bsub.add_parser("hook-install", + help="write the pre-commit (gate --mark-pass), " + "pre-merge-commit, and post-commit (seal) " + "hooks") + bh.add_argument("--workspace", default=".") + bh.add_argument("--python", default="", + help="interpreter to bake into the hooks") + bh.set_defaults(func=_bob_hooks) + + bs = bsub.add_parser("status", help="show which steps have receipts") + bs.add_argument("--workspace", default=".") + bs.set_defaults(func=_bob_status) + 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/adapters/doctor.py b/src/agent_ultra/adapters/doctor.py index 145c0ec..74508fd 100644 --- a/src/agent_ultra/adapters/doctor.py +++ b/src/agent_ultra/adapters/doctor.py @@ -253,6 +253,64 @@ def run_check(cmd, cwd): "refuted_verdict": refuted_receipt["verdict"]} +def bob_demo(out_dir: Path): + """Offline: the full 10-step pipeline passes its gate, then three faked + chains block — a skipped step, a fabricated fan-out claim, and a doctored + panel receipt. All state is isolated under *out_dir*.""" + import io + from ..bob import run_bob, gate_check + from ..bob.pipeline import BobRun + + ws = out_dir / "bob-demo" + key = out_dir / "bob-demo.key" + uc_home = out_dir / "bob-demo-ultracode" + sink = io.StringIO() + outcome = run_bob("add a slugify helper", ws, mock=True, key_path=key, + ultracode_home=uc_home, interactive=False, out=sink) + + run = BobRun(workspace=ws, run_dir=ws / ".agent-ultra" / "bob" / "runs" + / outcome.run_id, key=key.read_bytes()) + + # 1. a skipped step blocks: remove the workflow receipt + wf_receipt = run.receipt_path("step07_workflow") + saved = wf_receipt.read_text(encoding="utf-8") + wf_receipt.unlink() + skipped_blocks = not gate_check(run, allow_mock=True, + rerun_tests=False).passed + wf_receipt.write_text(saved, encoding="utf-8") + + # 2. a fabricated fan-out claim blocks: re-sign step06 naming an invented + # ultracode run (valid chain + HMAC, but nothing on disk backs the claim) + sec = run.load("step06_security") + run.receipt_path("step06_security").unlink() + run.write("step06_security", + {"ultracode": {"run_id": "20990101T000000-fake-000000", + "home": str(uc_home)}, + "lenses": ["invented"], "findings": []}, + writer="engine", mock=True) + fake_fanout_blocks = not gate_check(run, allow_mock=True, + rerun_tests=False).passed + run.receipt_path("step06_security").write_text( + json.dumps(sec, indent=2, ensure_ascii=False), encoding="utf-8") + + # 3. a doctored panel receipt blocks: flip one byte in the panel's own + # execution receipt (checksum no longer matches) + panel_receipt = run.run_dir / "panel" / "panel_execution_receipt.json" + body = json.loads(panel_receipt.read_text(encoding="utf-8")) + body["lens_count_executed"] = 99 + panel_receipt.write_text(json.dumps(body, indent=2), encoding="utf-8") + fake_panel_blocks = not gate_check(run, allow_mock=True, + rerun_tests=False).passed + + ok = (outcome.passed and skipped_blocks and fake_fanout_blocks + and fake_panel_blocks) + return ok, {"pipeline_passed": outcome.passed, + "skipped_step_blocks": skipped_blocks, + "fake_fanout_blocks": fake_fanout_blocks, + "fake_panel_blocks": fake_panel_blocks, + "run_id": outcome.run_id} + + def run_demo(args) -> int: out = Path(tempfile.mkdtemp(prefix="agent-ultra-demo-")) print("agent-ultra demo (offline, deterministic mock route)\n") @@ -277,8 +335,14 @@ def run_demo(args) -> int: print(f"[{'ok' if ok6 else 'FAIL'}] ultracode: {uc['calls']} agents fan out " f"-> {uc['final_state']} -> receipt valid={uc['receipt_valid']} " f"-> resume replays {uc['resume_cached']} cached ({uc['resume_calls']} calls)") + ok7, bb = bob_demo(out) + print(f"[{'ok' if ok7 else 'FAIL'}] bob: 10-step pipeline gate " + f"passed={bb['pipeline_passed']}; skipped step blocks=" + f"{bb['skipped_step_blocks']}, fake fan-out blocks=" + f"{bb['fake_fanout_blocks']}, doctored panel blocks=" + f"{bb['fake_panel_blocks']}") print(f"\nartifacts: {out}") - all_ok = ok1 and ok2 and ok3 and ok4 and ok5 and ok6 + all_ok = ok1 and ok2 and ok3 and ok4 and ok5 and ok6 and ok7 print("\nDEMO " + ("PASSED — the full loop works on this machine." if all_ok else "FAILED — run `agent-ultra doctor`.")) return 0 if all_ok else 1 @@ -381,6 +445,15 @@ def add(status, name, detail=""): except Exception as e: add(FAIL, "ultracode", str(e)[:160]) + # 7a3. bob: the 10-step pipeline passes its gate; faked chains block + try: + ok, detail = bob_demo(scratch) + add(PASS if ok else FAIL, + "bob: 10-step pipeline gate passes; skipped/faked steps block", + json.dumps(detail)[:140]) + except Exception as e: + add(FAIL, "bob pipeline", str(e)[:160]) + # 7b. receipts bus: authenticity separate from integrity, forgery rejected, # gate audit chain intact. try: diff --git a/src/agent_ultra/bob/__init__.py b/src/agent_ultra/bob/__init__.py new file mode 100644 index 0000000..86d293a --- /dev/null +++ b/src/agent_ultra/bob/__init__.py @@ -0,0 +1,32 @@ +"""bob — the 10-step enforced build pipeline, public edition. + + SPEC -> RED -> GREEN -> REFACTOR -> CODE-QUALITY -> SECURITY-FANOUT + -> WORKFLOW -> ULTRA -> QUIZ -> COMMIT + +Every gated step leaves a hash-chained, HMAC-signed receipt written from real +execution (pytest output, ultracode run receipts, the panel's execution +receipt). The commit/report gate re-derives what it can (live pytest) and +cross-checks the rest — a skipped, faked, or edited step blocks. + + agent-ultra bob run "add a slugify helper" --mock # no key needed + agent-ultra bob gate # validate the chain +""" + +from .receipts import ( + STEPS, 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, +) +from .runner import run_bob, 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", +] diff --git a/src/agent_ultra/bob/pipeline.py b/src/agent_ultra/bob/pipeline.py new file mode 100644 index 0000000..19a8531 --- /dev/null +++ b/src/agent_ultra/bob/pipeline.py @@ -0,0 +1,792 @@ +"""Bob pipeline core — runs, the RED/GREEN pytest runner, and the gate. + +Doctrine (public port of the 10-step enforced build pipeline): + + SPEC -> RED -> GREEN -> REFACTOR -> CODE-QUALITY -> SECURITY-FANOUT + -> WORKFLOW -> ULTRA -> QUIZ -> COMMIT + + real execution -> receipt -> gate validates the chain -> commit/report + +Enforcement comes from WHO writes a receipt and what the gate can RE-DERIVE +or CROSS-CHECK — never from trusting a claim: + + * steps 2/3 (RED/GREEN): receipts are written by ``run_pytest_step`` from + actual pytest output (``writer="system"``); the gate re-runs pytest live. + * steps 6/7 (fan-out): receipts point at an ultracode run; the gate loads + that run's own checksummed receipt, re-verifies it, and demands real + agent calls. An invented run_id, an empty run, or a doctored receipt + all block. + * step 8 (panel): the panel engine writes its execution receipt from real + model calls; the gate validates it structurally — a self-review is not + a panel. + * step 9 (quiz): ``passed`` requires a captured operator response; + otherwise the honest outcome is ``skipped``. + * every receipt is hash-chained and HMAC-signed (see receipts.py), so + skipping, re-ordering, editing, or hand-authoring a step breaks the + chain and the gate fails closed. + +Receipts live in ``/.agent-ultra/bob/runs//``, one JSON per +step. Stdlib only. +""" + +from __future__ import annotations + +import fnmatch +import hashlib +import json +import os +import re +import subprocess +import sys +from dataclasses import dataclass, field +from pathlib import Path + +from .receipts import ( + STEPS, GATED_STEPS, BobReceiptError, + build_step_receipt, load_receipt, validate_chain, utc_now, +) + +# staged files never produced by the build itself +DEFAULT_ALLOW_UNCOVERED = ("*.md", ".agent-ultra/*", ".gitignore") + + +class BobProofError(Exception): + """Raised when completion is claimed while an active run is unproven.""" + + +def default_key_path() -> Path: + home = os.environ.get("AGENT_ULTRA_HOME", "") + base = Path(home) if home else Path.home() / ".agent-ultra" + return base / "bob.key" + + +def load_or_create_key(key_path: str | Path | None = None) -> bytes: + """Per-install signing key. Created on first use, never committed + (it lives under the agent-ultra home, not the workspace).""" + p = Path(key_path) if key_path else default_key_path() + if p.is_file(): + return p.read_bytes() + p.parent.mkdir(parents=True, exist_ok=True) + key = os.urandom(32) + p.write_bytes(key) + return key + + +def sha256_file(path: Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(65536), b""): + h.update(chunk) + return h.hexdigest() + + +def safe_rel(workspace: Path, name: str) -> str: + """Return *name* as a workspace-relative POSIX path, or raise ValueError + if it is absolute or escapes the workspace via ``..``. Model-authored file + 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: + raise ValueError(f"{name!r} is absolute") + target = (workspace / cand).resolve() + try: + rel = target.relative_to(workspace) + except ValueError: + raise ValueError(f"{name!r} escapes the workspace") + if not rel.parts: + raise ValueError(f"{name!r} is the workspace root, not a file") + return rel.as_posix() + + +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 + detect a swapped or edited ultracode receipt regardless of that receipt's + own weaker self-checksum.""" + p = Path(home) / "runs" / str(run_id) / "receipt.json" + if not p.is_file(): + return "" + return sha256_file(p) + + +def utc_stamp() -> str: + import time + return time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) + + +# -------------------------------------------------------------------------- +# runs +# -------------------------------------------------------------------------- + +@dataclass +class BobRun: + """One pipeline run: a receipt directory plus the ACTIVE marker.""" + + workspace: Path + run_dir: Path + key: bytes = b"" + + @staticmethod + def _root(workspace: Path) -> Path: + return Path(workspace) / ".agent-ultra" / "bob" + + @classmethod + def start(cls, workspace: str | Path, goal: str = "", + key_path: str | Path | None = None, + force: bool = False) -> "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.""" + workspace = Path(workspace) + marker = cls._root(workspace) / "ACTIVE" + if marker.is_file() and 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 pass " + "force=True to abandon it. Starting over would orphan it.") + stamp = utc_stamp() + run_dir = cls._root(workspace) / "runs" / stamp + n = 0 + while run_dir.exists(): # same-second collision + n += 1 + run_dir = cls._root(workspace) / "runs" / f"{stamp}-{n}" + run_dir.mkdir(parents=True) + (run_dir / "run.json").write_text(json.dumps({ + "run_id": run_dir.name, "goal": goal, "started_utc": utc_now(), + }, indent=2, ensure_ascii=False), encoding="utf-8") + marker.write_text(run_dir.name, encoding="utf-8") + return cls(workspace=workspace, run_dir=run_dir, + key=load_or_create_key(key_path)) + + @classmethod + def active(cls, workspace: str | Path, + key_path: str | Path | None = None) -> "BobRun | None": + workspace = Path(workspace) + marker = cls._root(workspace) / "ACTIVE" + if not marker.is_file(): + return None + name = marker.read_text(encoding="utf-8").strip() + run_dir = cls._root(workspace) / "runs" / name + if not run_dir.is_dir(): + # A marker naming a missing run dir is TAMPERING, not "no run": + # complete() is the only code that removes the marker. Surface a + # sentinel run whose (empty) chain fails every gate, so the gate + # blocks instead of reporting "nothing to enforce". + return cls(workspace=workspace, run_dir=run_dir, + key=load_or_create_key(key_path)) + return cls(workspace=workspace, run_dir=run_dir, + key=load_or_create_key(key_path)) + + def marker_orphaned(self) -> bool: + return not self.run_dir.is_dir() + + @property + def run_id(self) -> str: + return self.run_dir.name + + def goal(self) -> str: + try: + meta = json.loads((self.run_dir / "run.json").read_text( + encoding="utf-8")) + return str(meta.get("goal", "")) + except (OSError, ValueError): + return "" + + def receipt_path(self, step: str) -> Path: + if step not in STEPS: + raise BobReceiptError(f"unknown pipeline step: {step}") + return self.run_dir / f"{step}.json" + + def load(self, step: str) -> dict | None: + return load_receipt(self.receipt_path(step)) + + def chain(self) -> list[dict]: + """All written receipts in seq order (the order they were written).""" + recs = [r for s in STEPS if (r := self.load(s)) is not None] + return sorted(recs, key=lambda r: r.get("seq", 0)) + + def _next_link(self) -> tuple[int, str]: + chain = self.chain() + if not chain: + return 0, "" + last = chain[-1] + return int(last.get("seq", 0)) + 1, str(last.get("receipt_sha256", "")) + + def hash_files(self, files) -> dict: + hashes: dict = {} + for f in files or []: + f = Path(f) + if not f.is_absolute(): + f = self.workspace / f + rel = str(f.resolve().relative_to(self.workspace.resolve())) + hashes[rel.replace("\\", "/")] = sha256_file(f) + return hashes + + def write(self, step: str, evidence: dict, writer: str, + files=None, mock: bool = False) -> dict: + seq, prev = self._next_link() + rec = build_step_receipt( + step=step, run_id=self.run_id, seq=seq, writer=writer, + evidence=evidence, files=self.hash_files(files), + prev_sha256=prev, mock=mock, key=self.key or None) + self.receipt_path(step).write_text( + json.dumps(rec, indent=2, ensure_ascii=False), encoding="utf-8") + return rec + + def mark_pass(self, gate_result: dict) -> None: + """Record the gate's pass log WITHOUT releasing the ACTIVE marker. + Called by the pre-commit gate: the run must stay active while the + commit it just validated is still in flight — releasing here is the + commit deadlock (pass -> ACTIVE gone -> a require-run policy blocks + the very commit that passed). ``seal()`` releases, after the commit + lands.""" + (self.run_dir / "gate-pass.json").write_text( + json.dumps({"completed_utc": utc_now(), + "git_head": git_head(self.workspace), + "gate": gate_result}, + indent=2, ensure_ascii=False), encoding="utf-8") + + def seal(self) -> bool: + """Release the ACTIVE marker after the commit landed (post-commit + hook). Refuses unless ``mark_pass()`` ran — a run must never be + sealed without a recorded gate pass. Records the landed commit hash; + that hash plus gate-pass.json is the step-10 record (no agent-written + step10 receipt on purpose). Returns True if sealed.""" + gp = self.run_dir / "gate-pass.json" + if not gp.is_file(): + return False + try: + data = json.loads(gp.read_text(encoding="utf-8")) + except (OSError, ValueError): + data = {} + data["sealed_utc"] = utc_now() + data["commit"] = git_head(self.workspace) + gp.write_text(json.dumps(data, indent=2, ensure_ascii=False), + encoding="utf-8") + marker = self._root(self.workspace) / "ACTIVE" + if marker.is_file() and marker.read_text( + encoding="utf-8").strip() == self.run_id: + marker.unlink() + return True + + def complete(self, gate_result: dict) -> None: + """mark_pass + seal in one step — for callers that validate and + finish OUTSIDE a commit (the mock demo's report path, an operator + closing a run manually). Inside a git workflow use the split: + pre-commit marks the pass, post-commit seals.""" + self.mark_pass(gate_result) + self.seal() + + +# -------------------------------------------------------------------------- +# system pytest runner (steps 2 and 3) +# -------------------------------------------------------------------------- + +# -v node lines: "tests/test_x.py::test_a PASSED [ 50%]" and parametrized ids +# that contain spaces: "test_x.py::test_a[hello world] PASSED [ 50%]". The id +# is everything up to the status word, which pytest right-pads before the +# result; require "::" so prose lines don't match. +_NODE_RE = re.compile( + r"^(\S+::.+?)\s+(PASSED|FAILED|ERROR|XFAIL|XPASS|SKIPPED)" + r"(?=\s|$|\s*\[)", re.M) + + +def _run_pytest(workspace: Path, test_file: str, timeout: int) -> tuple[int, str]: + cmd = [sys.executable, "-m", "pytest", test_file, "-v", "--tb=short", + "-p", "no:cacheprovider"] + try: + p = subprocess.run(cmd, cwd=str(workspace), capture_output=True, + text=True, timeout=timeout) + return p.returncode, (p.stdout or "") + "\n" + (p.stderr or "") + except (subprocess.TimeoutExpired, OSError, subprocess.SubprocessError): + return -1, f"TIMEOUT or spawn error after {timeout}s" + + +def parse_pytest_nodes(output: str) -> dict: + """{node_id: status} from verbose pytest output.""" + return {m.group(1): m.group(2) for m in _NODE_RE.finditer(output)} + + +def run_pytest_step(run: BobRun, step: str, test_file: str, + extra_files=None, timeout: int = 600, + mock: bool = False) -> tuple[bool, dict, str]: + """Execute pytest and write the RED or GREEN receipt from its ACTUAL + output (``writer="system"``). Returns (requirement_met, receipt, message). + The receipt is written either way — it records what really happened, and + the gate rejects a chain whose evidence misses the step's requirement. + The pytest invocation is classified through the command broker's risk + tiers and the tier is recorded in the evidence. + """ + if step not in ("red", "green"): + raise ValueError("step must be 'red' or 'green'") + step_name = "step02_red" if step == "red" else "step03_green" + + from ..broker.broker import classify + shown_cmd = f"{Path(sys.executable).name} -m pytest {test_file} -v" + tier, _why = classify(shown_cmd) + + exit_code, output = _run_pytest(run.workspace, test_file, timeout) + nodes = parse_pytest_nodes(output) + counts = {"passed": 0, "failed": 0, "error": 0, "skipped": 0, "other": 0} + for status in nodes.values(): + k = status.lower() + counts[k if k in counts else "other"] += 1 + + evidence = { + "test_file": test_file, + "command": shown_cmd, + "broker_tier": tier, + "exit_code": exit_code, + "test_ids": sorted(nodes), + "statuses": nodes, + "counts": counts, + "output_tail": output[-4000:], + "output_sha256": hashlib.sha256( + output.encode("utf-8", errors="ignore")).hexdigest(), + } + + if step == "red": + # failing tests, not broken ones: import/collection mistakes are + # ERRORs and they don't count as a proper RED + ok = bool(nodes) and counts["failed"] >= 1 and counts["error"] == 0 + msg = (f"RED: {counts['failed']} failing / {len(nodes)} collected" + if ok else + f"RED requirement NOT met: failed={counts['failed']} " + f"error={counts['error']} collected={len(nodes)} exit={exit_code}") + else: + ok = exit_code == 0 and counts["passed"] >= 1 and counts["failed"] == 0 + msg = (f"GREEN: {counts['passed']} passing" if ok else + f"GREEN requirement NOT met: passed={counts['passed']} " + f"failed={counts['failed']} exit={exit_code}") + + evidence["requirement_met"] = bool(ok) + files = [Path(test_file)] + [Path(f) for f in (extra_files or [])] + receipt = run.write(step_name, evidence, writer="system", files=files, + mock=mock) + return bool(ok), receipt, msg + + +# -------------------------------------------------------------------------- +# cross-checks for the fan-out and panel steps +# -------------------------------------------------------------------------- + +def _as_int(value) -> int: + try: + return int(value or 0) + except (TypeError, ValueError): + return 0 + + +def check_ultracode_evidence(evidence: dict, *, allow_mock: bool, + receipt_mock: bool, + expect_workflow: str = "") -> list[str]: + """Cross-check a step-6/7 receipt against the ultracode run it names. + + The bob step receipt (HMAC-signed) records the ultracode receipt's + file-bytes fingerprint and the workflow name; here we re-derive both and + demand the run actually fanned out. A fabricated run_id (no file), an + edited receipt (fingerprint mismatch), the wrong workflow (borrowed run), + zero agents, or a checksum failure all fail closed. + """ + from ..ultracode import verify_receipt + if not isinstance(evidence, dict): + return ["fan-out evidence malformed"] + errors: list[str] = [] + uc = evidence.get("ultracode") + uc = uc if isinstance(uc, dict) else {} + run_id = str(uc.get("run_id", "") or "") + home = str(uc.get("home", "") or "") + if not run_id or not home: + return ["no ultracode run recorded — fan-out may be fabricated"] + receipt_path = Path(home) / "runs" / run_id / "receipt.json" + if not receipt_path.is_file(): + return [f"ultracode run {run_id!r} has no receipt on disk — " + "fabricated run_id?"] + # the signed bob receipt pinned the ultracode receipt's file bytes: an + # edited or swapped receipt breaks this without needing the ultracode + # engine's (weaker, keyless) self-checksum to catch it. + recorded_fp = str(uc.get("receipt_file_sha256", "") or "") + actual_fp = sha256_file(receipt_path) + if not recorded_fp: + errors.append("fan-out receipt has no recorded ultracode fingerprint " + "— cannot bind it to this run (fail closed)") + elif recorded_fp != actual_fp: + errors.append(f"ultracode receipt for {run_id!r} does not match the " + "fingerprint pinned at run time — swapped or edited") + try: + rec = json.loads(receipt_path.read_text(encoding="utf-8")) + except (OSError, ValueError) as e: + return errors + [f"ultracode receipt unreadable: {e}"] + if not isinstance(rec, dict): + return errors + ["ultracode receipt is not an object"] + if not verify_receipt(rec): + errors.append(f"ultracode receipt for {run_id!r} fails its checksum " + "— edited after the run") + if str(rec.get("run_id", "")) != run_id: + errors.append(f"ultracode receipt's own run_id {rec.get('run_id')!r} " + f"!= claimed {run_id!r} — mismatched receipt") + recorded_name = str(uc.get("workflow_name", "") or "") + actual_name = str(rec.get("name", "") or "") + if expect_workflow and recorded_name != expect_workflow: + errors.append(f"fan-out claims workflow {recorded_name!r}, expected " + f"{expect_workflow!r} for this step") + if expect_workflow and actual_name != expect_workflow: + errors.append(f"ultracode run is workflow {actual_name!r}, not the " + f"expected {expect_workflow!r} — borrowed run?") + agents = rec.get("agents") + agents = agents if isinstance(agents, list) else [] + done = [a for a in agents if isinstance(a, dict) and a.get("status") == "ok"] + if not done: + errors.append(f"ultracode run {run_id!r} completed 0 agents — " + "an empty fan-out proves nothing") + 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: + errors.append("receipt is marked mock — mock evidence only satisfies " + "the gate in demo mode (--allow-mock)") + return errors + + +def check_panel_evidence(evidence: dict, workspace: Path) -> list[str]: + """Step 8: validate the panel's own execution receipt AND bind it to this + run — the signed step receipt pins the expected task_id and the sha256 of + the reviewed source, so a borrowed panel (different task/code) is rejected + even though its own checksum is intact. A self-review is not a panel; a + panel of someone else's code is not a review of this change.""" + from ..panel_receipt import validate_receipt + if not isinstance(evidence, dict): + return ["panel evidence malformed"] + art = str(evidence.get("panel_run_dir", "") or "") + if not art: + return ["no panel run dir recorded"] + path = Path(art) + if not path.is_absolute(): + path = workspace / path + expect_task = str(evidence.get("expect_task_id", "") or "") + expect_hash = str(evidence.get("expect_artifact_hash", "") or "") + if not expect_task or not expect_hash: + return ["panel evidence is not bound to this run (missing " + "expect_task_id/expect_artifact_hash) — fail closed"] + ok, msg = validate_receipt(path, expect_task_id=expect_task, + expect_artifact_hash=expect_hash) + return [] if ok else [msg] + + +# -------------------------------------------------------------------------- +# the gate +# -------------------------------------------------------------------------- + +@dataclass +class GateResult: + passed: bool + errors: list = field(default_factory=list) + checked: dict = field(default_factory=dict) # step -> one-line summary + + def to_dict(self) -> dict: + return {"passed": self.passed, "errors": self.errors, + "checked": self.checked} + + +def _staged_files(workspace: Path) -> "list[str] | None": + """Staged paths RELATIVE TO THE WORKSPACE (``git diff --cached + --relative``), or None when git cannot answer (not a repo, git missing) — + the caller fails that closed rather than treating it as 'nothing staged'.""" + try: + p = subprocess.run(["git", "-C", str(workspace), "diff", "--cached", + "--relative", "--name-only"], + capture_output=True, text=True, timeout=30) + except (OSError, subprocess.SubprocessError): + return None + if p.returncode != 0: + return None + return [ln.strip().replace("\\", "/") for ln in p.stdout.splitlines() + if ln.strip()] + + +def gate_check(run: BobRun, *, rerun_tests: bool = True, + allow_mock: bool = False, staged_files=None, + allow_uncovered: tuple = DEFAULT_ALLOW_UNCOVERED) -> GateResult: + """Validate the whole receipt chain. Fail-closed: a check that cannot be + performed (missing receipt, unreadable/malformed artifact, unverifiable + chain, git unavailable for coverage) is a failure, not a warning — any + unexpected error is caught and reported as a block, never a crash. + + ``staged_files=None`` reads git's staged list; pass ``[]`` to skip the + coverage check explicitly (report path, where nothing is staged but the + receipts themselves must still hold). + """ + try: + return _gate_check(run, rerun_tests=rerun_tests, allow_mock=allow_mock, + staged_files=staged_files, + allow_uncovered=allow_uncovered) + except Exception as e: # fail closed: a malformed receipt blocks, not crashes + return GateResult(passed=False, + errors=[f"gate error (fail-closed): " + f"{type(e).__name__}: {e}"]) + + +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 " + "the gate in demo mode (--allow-mock)") + return None + + +def _gate_check(run: BobRun, *, rerun_tests: bool, allow_mock: bool, + staged_files, allow_uncovered: tuple) -> GateResult: + errors: list[str] = [] + checked: dict = {} + + # An ACTIVE marker pointing at a missing run dir is tampering, not "done". + 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)"]) + + # ---- chain spine: order, hashes, signatures ------------------------- + chain = run.chain() + errors.extend(validate_chain(chain, run.key or None)) + checked["chain"] = f"{len(chain)} receipt(s), hash-chained + signed" + + # ---- completeness ---------------------------------------------------- + have = {r.get("step") for r in chain} + for step in 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 {} + + def _counts(ev) -> dict: + c = ev.get("counts") + return c if isinstance(c, dict) else {} + + # ---- step 2: RED ------------------------------------------------------ + red = run.load("step02_red") + red_ids: set = set() + if red is not None: + ev = _ev(red) + red_ids = set(ev.get("test_ids") or []) + if red.get("writer") != "system": + errors.append("step02_red: receipt not written by the pytest " + "runner (writer != system)") + if not ev.get("requirement_met"): + errors.append("step02_red: tests did not actually fail cleanly") + if _as_int(_counts(ev).get("failed")) < 1: + errors.append("step02_red: 0 test failures recorded") + if _as_int(_counts(ev).get("error")) > 0: + errors.append("step02_red: collection/import ERRORs recorded — " + "broken tests are not failing tests") + if (e := _mock_flag_error(red, "step02_red", allow_mock)): + errors.append(e) + checked["step02_red"] = (f"{len(red_ids)} tests, " + f"{_as_int(_counts(ev).get('failed'))} failing") + + # ---- step 3: GREEN ---------------------------------------------------- + green = run.load("step03_green") + test_file = "" + if green is not None: + ev = _ev(green) + test_file = str(ev.get("test_file", "") or "") + green_ids = set(ev.get("test_ids") or []) + if green.get("writer") != "system": + errors.append("step03_green: receipt not written by the pytest " + "runner (writer != system)") + if not ev.get("requirement_met"): + errors.append("step03_green: tests did not all pass") + if not test_file: + errors.append("step03_green: no test_file recorded — the live " + "re-run cannot be performed (fail closed)") + if red is not None and not red_ids <= green_ids: + missing = sorted(red_ids - green_ids)[:5] + errors.append(f"step03_green: RED tests missing from the GREEN " + f"run: {missing}") + if (e := _mock_flag_error(green, "step03_green", allow_mock)): + errors.append(e) + checked["step03_green"] = f"{len(green_ids)} tests passing" + + # ---- live re-run (re-derived, never trusted) --------------------------- + if rerun_tests and green is not None and test_file: + exit_code, output = _run_pytest(run.workspace, test_file, timeout=600) + live_ids = set(parse_pytest_nodes(output)) + if exit_code != 0: + errors.append(f"live re-run: pytest exit {exit_code} — the " + "current state does not pass its own tests") + if red is not None and not red_ids <= live_ids: + errors.append("live re-run: tests recorded at RED are no longer " + "collected — the test file changed after receipts") + checked["live_rerun"] = f"exit {exit_code}, {len(live_ids)} tests" + + # ---- steps 6/7: fan-out cross-checked against ultracode artifacts ----- + for step, extra, wf_name in (("step06_security", "", "bob-security"), + ("step07_workflow", "dimensions", "bob-review")): + rec = run.load(step) + if rec is None: + continue # missing already reported + ev = _ev(rec) + step_errors = check_ultracode_evidence( + ev, allow_mock=allow_mock, receipt_mock=bool(rec.get("mock")), + expect_workflow=wf_name) + errors.extend(f"{step}: {e}" for e in step_errors) + if extra and not ev.get(extra): + errors.append(f"{step}: no {extra} recorded") + checked[step] = ("ultracode run verified" if not step_errors + else "INVALID") + + # ---- step 8: panel ----------------------------------------------------- + panel = run.load("step08_panel") + if panel is not None: + perrs = check_panel_evidence(_ev(panel), run.workspace) + if (e := _mock_flag_error(panel, "panel", allow_mock)): + perrs = perrs + [e] + errors.extend(f"step08_panel: {e}" for e in perrs) + checked["step08_panel"] = "receipt valid" if not perrs else "INVALID" + + # ---- step 9: quiz ------------------------------------------------------- + quiz = run.load("step09_quiz") + if quiz is not None: + ev = _ev(quiz) + if (e := _mock_flag_error(quiz, "step09_quiz", allow_mock)): + errors.append(e) + outcome = ev.get("outcome") + if outcome not in ("passed", "skipped"): + errors.append(f"step09_quiz: outcome {outcome!r} is not " + "passed/skipped") + if not ev.get("questions"): + errors.append("step09_quiz: no comprehension questions recorded") + if outcome == "passed" and not str( + ev.get("operator_response", "") or "").strip(): + errors.append("step09_quiz: outcome 'passed' without a captured " + "operator response — a quiz nobody answered") + checked["step09_quiz"] = f"outcome={outcome}" + + # ---- staleness: receipts vs the working tree ----------------------------- + 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 step that owns it") + checked["staleness"] = f"{len(coverage)} file(s) covered" + + # ---- staged files must be covered ----------------------------------------- + # staged_files=[] (report/claim path) skips this deliberately. None means + # "read git": inside a work tree, a diff failure fails closed; outside any + # repo there is no `git commit` to smuggle into, so coverage is moot. + 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("coverage check: `git diff --cached` failed " + "inside a work tree — cannot verify staged " + "files (fail closed)") + staged_files = [] + else: + staged_files = [] + for rel in staged_files: + if rel in coverage: + continue + if any(fnmatch.fnmatch(rel, pat) for pat in allow_uncovered): + continue + errors.append(f"uncovered: staged file {rel} appears in no receipt — " + "it was never tested or fanned out for scrutiny") + + return GateResult(passed=not errors, errors=errors, checked=checked) + + +def _inside_work_tree(workspace: Path) -> bool: + try: + p = subprocess.run(["git", "-C", str(workspace), "rev-parse", + "--is-inside-work-tree"], capture_output=True, + text=True, timeout=30) + except (OSError, subprocess.SubprocessError): + return False + return p.returncode == 0 and p.stdout.strip() == "true" + + +def git_head(workspace: Path) -> str: + try: + p = subprocess.run(["git", "-C", str(workspace), "rev-parse", "HEAD"], + capture_output=True, text=True, timeout=30) + return p.stdout.strip() if p.returncode == 0 else "" + except (OSError, subprocess.SubprocessError): + return "" + + +# --------------------------------------------------------------------------- +# git hooks: pre-commit gates (mark only), post-commit seals +# --------------------------------------------------------------------------- + +# The lifecycle that avoids the commit deadlock: the pre-commit gate MARKS a +# pass but keeps ACTIVE (the run must stay active for the commit it just +# validated — sealing here would make any require-run policy block that very +# commit); the post-commit hook SEALS once the commit has landed. + +_PRE_HOOK = """#!/bin/sh +# bob pipeline gate (generated by agent-ultra bob hook-install). +# Blocks the commit while a bob run is active and unproven; on pass it only +# MARKS the pass — the post-commit hook seals the run after the commit lands. +# No active run -> exits 0 (non-interference) unless AGENT_ULTRA_REQUIRE_RUN=1. +"{python}" -m agent_ultra.adapters.cli bob gate --workspace "$(git rev-parse --show-toplevel)" --hook --mark-pass{flags} +""" + +_POST_HOOK = """#!/bin/sh +# bob pipeline seal (generated by agent-ultra bob hook-install). +# The commit has landed: seal the active run (requires a recorded gate pass; +# refuses otherwise). No active run -> no-op. +"{python}" -m agent_ultra.adapters.cli bob seal --workspace "$(git rev-parse --show-toplevel)" --if-passed +""" + + +def install_hooks(workspace: Path, python: str = "", + gate_flags: str = "") -> Path: + """Write the pre-commit (gate --mark-pass), pre-merge-commit (git fires + that hook, not pre-commit, for `git merge` commits), and post-commit + (seal) hooks in *workspace*. ``gate_flags`` appends extra flags to the + pre-commit gate (e.g. ``--allow-mock`` for demo workspaces). Returns the + pre-commit hook path.""" + import stat + hooks = Path(workspace) / ".git" / "hooks" + if not hooks.parent.is_dir(): + raise FileNotFoundError(f"not a git repo: {workspace}") + hooks.mkdir(exist_ok=True) + fmt = {"python": (python or sys.executable).replace("\\", "/"), + "flags": (" " + gate_flags.strip()) if gate_flags.strip() else ""} + written = [] + for name, tmpl in (("pre-commit", _PRE_HOOK), + ("pre-merge-commit", _PRE_HOOK), + ("post-commit", _POST_HOOK)): + hook = hooks / name + hook.write_text(tmpl.format(**fmt), encoding="utf-8", newline="\n") + hook.chmod(hook.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP + | stat.S_IXOTH) + written.append(hook) + return written[0] + + +def assert_bob_done(workspace: str | Path, *, + allow_mock: bool = False, + key_path: str | Path | None = None) -> "GateResult | None": + """Raise BobProofError if a bob run is active in *workspace* and its + receipt chain does not validate. Returns the GateResult (None when no run + is active). ``staged_files=[]`` on purpose: at claim time nothing need be + staged — the receipts themselves must hold.""" + run = BobRun.active(workspace, key_path=key_path) + if run is None: + return None + result = gate_check(run, allow_mock=allow_mock, staged_files=[]) + if not result.passed: + raise BobProofError( + f"bob run {run.run_id}: completion claimed without a valid " + f"receipt chain — {len(result.errors)} failure(s): " + + "; ".join(result.errors[:5]) + + ("; ..." if len(result.errors) > 5 else "")) + return result diff --git a/src/agent_ultra/bob/receipts.py b/src/agent_ultra/bob/receipts.py new file mode 100644 index 0000000..77a3265 --- /dev/null +++ b/src/agent_ultra/bob/receipts.py @@ -0,0 +1,134 @@ +"""Bob receipt schema — one hash-chained, HMAC-signed receipt per pipeline step. + +A bob receipt is a plain JSON dict (public-safe, stdlib-only) with three +tamper controls, reusing the receipts-bus hash helpers: + + * ``receipt_sha256`` — integrity: the body was not edited after writing. + * ``receipt_hmac`` — authenticity: it was written by something holding the + per-install key (a hand-authored receipt with a correct sha256 but no + valid HMAC is a forgery and the gate rejects it). + * ``prev_sha256`` — chaining: each receipt embeds the sha256 of the one + before it, so editing or re-writing ANY earlier step breaks every later + receipt. The chain is the pipeline's spine. + +``writer`` records WHO produced the evidence: ``system`` receipts are written +by the pytest runner from real subprocess output; ``engine`` receipts are +written by the bob runner from real ultracode / panel artifacts it can +re-verify; ``agent`` receipts carry judgment-step notes the gate does not +treat as proof. The gate trusts nothing it cannot re-derive or cross-check. +""" + +from __future__ import annotations + +import json +import time +import uuid +from pathlib import Path + +from ..receipts_bus.envelope import receipt_sha256, receipt_hmac, verify_hashes + +STEPS = ( + "step01_spec", "step02_red", "step03_green", "step04_refactor", + "step05_codequality", "step06_security", "step07_workflow", + "step08_panel", "step09_quiz", "step10_commit", +) + +# Steps the gate demands a receipt for. 1/4/5 are judgment steps where a +# mechanical gate would fake enforcement; 10's record is gate-pass.json itself. +GATED_STEPS = ("step02_red", "step03_green", "step06_security", + "step07_workflow", "step08_panel", "step09_quiz") + +WRITERS = frozenset({"system", "engine", "agent"}) + + +class BobReceiptError(ValueError): + """A bob receipt violated the schema (bad step, bad writer, bad chain).""" + + +def utc_now() -> str: + return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + + +def build_step_receipt( + *, + step: str, + run_id: str, + seq: int, + writer: str, + evidence: dict, + files: dict | None = None, + prev_sha256: str = "", + mock: bool = False, + key: bytes | None = None, + ts: str = "", +) -> dict: + """Build a fully-formed, hashed (and, with ``key``, signed) step receipt. + + ``files`` maps workspace-relative paths to their sha256 at receipt time — + the gate's staleness check re-hashes them later. ``mock`` marks evidence + produced on the keyless mock route; the gate only accepts mock receipts + when explicitly told to (demo mode), never by default. + """ + if step not in STEPS: + raise BobReceiptError(f"unknown step {step!r}; allowed: {list(STEPS)}") + if writer not in WRITERS: + raise BobReceiptError(f"unknown writer {writer!r}; allowed: {sorted(WRITERS)}") + rec: dict = { + "schema": "bob-step-receipt/1", + "receipt_id": uuid.uuid4().hex, + "step": step, + "run_id": run_id, + "seq": int(seq), + "ts": ts or utc_now(), + "writer": writer, + "mock": bool(mock), + "evidence": dict(evidence or {}), + "files": dict(files or {}), + "prev_sha256": prev_sha256, + } + rec["receipt_sha256"] = receipt_sha256(rec) + rec["receipt_hmac"] = receipt_hmac(rec, key) if key else "" + return rec + + +def load_receipt(path: str | Path) -> dict | None: + p = Path(path) + if not p.is_file(): + return None + try: + rec = json.loads(p.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + return rec if isinstance(rec, dict) else None + + +def validate_chain(receipts: list[dict], key: bytes | None) -> list[str]: + """Validate an ordered receipt chain. Returns a list of errors (empty = + chain holds). Checks, per receipt: schema fields, integrity sha256, + HMAC authenticity (fail closed — no key or empty HMAC is a failure), + strictly increasing ``seq``, and ``prev_sha256`` linkage. + """ + errors: list[str] = [] + prev_sha = "" + prev_seq = -1 + for rec in receipts: + label = rec.get("step", "?") + if rec.get("schema") != "bob-step-receipt/1": + errors.append(f"{label}: not a bob step receipt (schema field)") + continue + integrity, authentic = verify_hashes(rec, key) + if not integrity: + errors.append(f"{label}: integrity FAIL — receipt edited after writing") + if not authentic: + errors.append(f"{label}: authenticity FAIL — missing/invalid HMAC " + "(hand-authored receipt?)") + seq = rec.get("seq", -1) + if not isinstance(seq, int) or seq <= prev_seq: + errors.append(f"{label}: seq {seq!r} does not increase (chain " + "reordered or receipt re-written)") + if rec.get("prev_sha256", "") != prev_sha: + errors.append(f"{label}: prev_sha256 does not match the previous " + "receipt — the chain is broken") + prev_sha = rec.get("receipt_sha256", "") + prev_seq = seq if isinstance(seq, int) else prev_seq + return errors diff --git a/src/agent_ultra/bob/runner.py b/src/agent_ultra/bob/runner.py new file mode 100644 index 0000000..5f82372 --- /dev/null +++ b/src/agent_ultra/bob/runner.py @@ -0,0 +1,349 @@ +"""The bob runner — drives the 10 steps end to end and gates the report. + + from agent_ultra.bob import run_bob + outcome = run_bob("add a slugify helper", workspace, mock=True) + +Mock mode (no API key) uses the bundled sample task for MODEL CONTENT only; +pytest, ultracode, the panel, and the gate all really execute (see sample.py). +On a real route the spec/tests/implementation come from the configured model; +if the model's output does not genuinely fail at RED and pass at GREEN, the +run ends blocked — the pipeline never pretends. + +The final report is printed ONLY when ``gate_check`` passes; a failing chain +prints the exact errors and exits nonzero. Committing stays in the operator's +hands: the gate seals the run and tells you it is safe to commit. +""" + +from __future__ import annotations + +import hashlib +import json +import sys +from dataclasses import dataclass, field +from pathlib import Path + +from .pipeline import ( + BobRun, BobProofError, gate_check, run_pytest_step, GateResult, + sha256_file, safe_rel, ultracode_receipt_fingerprint, +) +from . import sample +from .workflows import SECURITY_WORKFLOW, REVIEW_WORKFLOW + + +@dataclass +class BobOutcome: + run_id: str + passed: bool + blocked_at: str = "" # step name when the pipeline stopped early + gate: "GateResult | None" = None + steps: list = field(default_factory=list) # (step, one-line summary) + report: dict = field(default_factory=dict) + + +def _say(out, line: str) -> None: + print(line, file=out or sys.stdout) + + +def _excerpt(workspace: Path, files, limit: int = 6000) -> str: + parts = [] + for rel in files: + p = workspace / rel + try: + parts.append(f"### {rel}\n" + p.read_text(encoding="utf-8")) + except OSError: + continue + return "\n".join(parts)[:limit] + + +# -- real-route content generation (mock mode uses sample.py instead) -------- + +def _model_json(pool, prompt: str, max_tokens: int = 4000): + from ..panel.json_extract import extract_json + model = pool.routes[0] + raw = pool.client_for(model).complete(model, prompt, max_tokens) + return extract_json(raw) + + +def _gen_spec(pool, task: str) -> str: + obj = _model_json(pool, + "Write a one-paragraph implementation spec for this task. Reply with " + f'ONLY JSON: {{"spec": ""}}.\nTask: {task}') + return str((obj or {}).get("spec", "") or "").strip() + + +def _gen_files(pool, task: str, spec: str) -> "tuple[str, str, str, str] | None": + """Ask the model for (test_file, tests, impl_file, stub) then later the + implementation. Returns None when the model output is unusable.""" + obj = _model_json(pool, + "Write pytest contract tests (RED-first) plus a stub for this task. " + 'Reply with ONLY JSON: {"test_file": "test_x.py", "tests": "", "impl_file": "x.py", "stub": ""}.\n' + f"Task: {task}\nSpec: {spec}", max_tokens=6000) + if not isinstance(obj, dict): + return None + keys = ("test_file", "tests", "impl_file", "stub") + if not all(str(obj.get(k, "") or "").strip() for k in keys): + return None + return tuple(str(obj[k]) for k in keys) # type: ignore[return-value] + + +def _gen_impl(pool, task: str, spec: str, tests: str, impl_file: str) -> str: + obj = _model_json(pool, + "Implement the module so ALL these tests pass. The tests are " + "READ-ONLY. Reply with ONLY JSON: " + f'{{"content": ""}}.\n' + f"Task: {task}\nSpec: {spec}\nTests:\n{tests}", max_tokens=8000) + return str((obj or {}).get("content", "") or "") + + +# -- the pipeline ------------------------------------------------------------- + +def run_bob(task: str, workspace: str | Path, *, mock: bool = False, + pool=None, key_path=None, ultracode_home=None, + interactive: "bool | None" = None, force: bool = False, + seal_on_pass: bool = True, out=None) -> BobOutcome: + """Run SPEC->...->COMMIT-gate in *workspace*. Returns a BobOutcome; the + caller maps ``passed`` to the exit code. ``pool`` is a RoutePool (mock + or real); ``ultracode_home`` and ``key_path`` exist so tests and the + doctor can isolate all state. ``force`` abandons an existing unproven + active run instead of refusing to start. + + ``seal_on_pass=False`` marks the gate pass but keeps the run ACTIVE — + the git-workflow lifecycle, where the post-commit hook seals after the + commit lands (see `bob hook-install`). The default seals immediately, + which is correct for the report path (mock demo, no commit follows).""" + from ..routes.pool import RoutePool + workspace = Path(workspace) + workspace.mkdir(parents=True, exist_ok=True) + if mock: + # two dialects of the offline route: the ultracode mock answers + # schema prompts; the panel mock answers challenge/verdict prompts + from ..ultracode import demo_pool + from ..routes.mock import demo_panel_client + pool = demo_pool() + panel_pool = RoutePool(["mock-a"], client=demo_panel_client()) + elif pool is None: + raise ValueError("a real run needs a configured RoutePool; " + "pass pool= or use mock=True") + else: + panel_pool = pool + if interactive is None: + interactive = (not mock) and sys.stdin.isatty() + + try: + run = BobRun.start(workspace, goal=task, key_path=key_path, force=force) + 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 run {run.run_id} — {task}" + + (" [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) + p.write_text(content, encoding="utf-8") + ok, _rec, msg = run_pytest_step(run, "red", test_file, + extra_files=[impl_file], mock=mock) + record("step02_red", msg) + if not ok: + return blocked("step02_red", msg) + + # ---- 3. GREEN ------------------------------------------------------------- + impl = sample.SAMPLE_IMPL if mock else _gen_impl(pool, task, spec, tests, + impl_file) + if not impl: + return blocked("step03_green", "model produced no implementation") + (workspace / impl_file).write_text(impl, encoding="utf-8") + ok, _rec, msg = run_pytest_step(run, "green", test_file, + extra_files=[impl_file], mock=mock) + record("step03_green", msg) + if not ok: + return blocked("step03_green", msg) + + # ---- 4/5. REFACTOR + CODE-QUALITY (judgment steps, notes only) ---------- + run.write("step04_refactor", + {"notes": "reviewed for structure; no edits needed" + if mock else "see step07 review notes"}, + writer="agent", files=[test_file, impl_file], mock=mock) + record("step04_refactor", "no edits (hashes re-pinned)") + run.write("step05_codequality", + {"checklist": ["names read clearly", "no dead code", + "errors surface, not swallowed"]}, + writer="agent", files=[test_file, impl_file], mock=mock) + record("step05_codequality", "checklist recorded") + + # ---- 6/7. fan-out steps via ultracode ------------------------------------ + from ..ultracode import UltracodeEngine + eng = UltracodeEngine(pool, home=ultracode_home) + excerpt = _excerpt(workspace, [test_file, impl_file]) + excerpt_sha = hashlib.sha256(excerpt.encode("utf-8")).hexdigest() + + def _fanout_evidence(rep, wf_name, extra): + """Bind the ultracode run to THIS bob run: the recorded fingerprint + (file-bytes sha of the run's receipt.json) is stamped into the + HMAC-signed bob receipt, so editing the ultracode receipt or pointing + at a different run breaks the signed chain (see gate cross-check).""" + fp = ultracode_receipt_fingerprint(eng.home, rep.run_id) + ev = {"ultracode": {"run_id": rep.run_id, "home": str(eng.home), + "workflow_name": wf_name, + "receipt_file_sha256": fp}, + "excerpt_sha256": excerpt_sha} + ev.update(extra) + return ev + + rep = eng.run_script(SECURITY_WORKFLOW, + args={"goal": task, "target_excerpt": excerpt}) + if rep.final_state != "COMPLETE": + return blocked("step06_security", + f"security fan-out ended {rep.final_state}") + result = rep.result or {} + run.write("step06_security", + _fanout_evidence(rep, "bob-security", + {"lenses": result.get("lenses", []), + "findings": result.get("findings", [])}), + writer="engine", files=[test_file, impl_file], mock=mock) + record("step06_security", + f"{rep.agents_run} agents, {len(result.get('findings', []))} finding(s)") + + rep = eng.run_script(REVIEW_WORKFLOW, + args={"goal": task, "spec": spec, + "target_excerpt": excerpt}) + if rep.final_state != "COMPLETE": + return blocked("step07_workflow", + f"review fan-out ended {rep.final_state}") + result = rep.result or {} + run.write("step07_workflow", + _fanout_evidence(rep, "bob-review", + {"dimensions": result.get("dimensions", []), + "reviews": result.get("reviews", {})}), + writer="engine", files=[test_file, impl_file], mock=mock) + record("step07_workflow", + f"{rep.agents_run} agents over {len(result.get('dimensions', []))} dimensions") + + # ---- 8. ULTRA (adversarial panel + its execution receipt) ---------------- + from ..panel.engine import PanelEngine, PanelError + from ..panel_receipt import build_receipt, write_receipt + engine = PanelEngine(panel_pool) + reviewed = _excerpt(workspace, [impl_file]) + try: + report = engine.run(f"Is this implementation of '{task}' safe and " + "correct for production?", + size="small", + lenses=["security", "correctness", "failure-modes"], + context=reviewed) + except PanelError as e: + return blocked("step08_panel", f"panel error: {e}") + panel_dir = run.run_dir / "panel" + reviewed_sha = hashlib.sha256(reviewed.encode("utf-8")).hexdigest() + panel_path = write_receipt(panel_dir, + build_receipt(report, task_id=run.run_id, + reviewed_input=reviewed)) + # cover the panel receipt file itself so the chain's staleness check + # re-hashes it — an edit-then-recompute of its own checksum still trips + # the signed bob receipt. + run.write("step08_panel", + {"panel_run_dir": panel_dir.relative_to(workspace).as_posix(), + "expect_task_id": run.run_id, + "expect_artifact_hash": reviewed_sha, + "decision": report.decision, + "accepted_findings": len(report.accepted)}, + writer="engine", + files=[test_file, impl_file, panel_path], mock=mock) + record("step08_panel", f"decision: {str(report.decision)[:60]}") + + # ---- 9. QUIZ --------------------------------------------------------------- + questions = list(sample.SAMPLE_QUIZ_QUESTIONS) if mock else [ + f"What behavior does {test_file} pin down, in your own words?", + "Which edge case would you add a test for next?", + "What would make you revert this change?", + ] + 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("step09_quiz", + {"questions": questions, "outcome": outcome, + "operator_response": operator_response}, + writer="agent" if outcome == "passed" else "engine", mock=mock) + record("step09_quiz", f"outcome={outcome} ({len(questions)} questions)") + + # ---- 10. COMMIT gate --------------------------------------------------------- + # report path: nothing is staged yet, so staged_files=[] — the receipts + # themselves must hold. The pre-commit `bob gate` enforces staged coverage. + gate = gate_check(run, allow_mock=mock, staged_files=[]) + if not gate.passed: + _say(out, "\ngate BLOCKED the report:") + for e in gate.errors: + _say(out, f" ! {e}") + return BobOutcome(run_id=run.run_id, passed=False, + blocked_at="step10_commit", gate=gate, steps=steps) + if seal_on_pass: + run.complete(gate.to_dict()) + else: + run.mark_pass(gate.to_dict()) + report_doc = { + "run_id": run.run_id, "task": task, "spec": spec, + "files": [test_file, impl_file], + "gate": gate.to_dict(), "mock": mock, + } + (run.run_dir / "report.json").write_text( + json.dumps(report_doc, indent=2, ensure_ascii=False), encoding="utf-8") + _say(out, f"\ngate PASSED — receipt chain validates " + f"({len(run.chain())} receipts).") + _say(out, f"report: {run.run_dir / 'report.json'}") + if seal_on_pass: + _say(out, "safe to commit: git add + git commit (re-run " + "`agent-ultra bob gate` after staging).") + else: + _say(out, "pass marked; run stays ACTIVE — the post-commit hook " + "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) diff --git a/src/agent_ultra/bob/sample.py b/src/agent_ultra/bob/sample.py new file mode 100644 index 0000000..690dd01 --- /dev/null +++ b/src/agent_ultra/bob/sample.py @@ -0,0 +1,70 @@ +"""The bundled sample build task for bob's mock mode. + +Mock mode swaps the MODEL CONTENT (spec text, test file, implementation) for +these deterministic fixtures — everything else is real: pytest really runs +RED and GREEN as subprocesses, ultracode really fans out on the mock route, +the panel really executes and writes its receipt, and the gate really +validates the chain. That's what makes the no-key demo honest proof of the +machinery rather than a canned transcript. + +The task: a tiny ``slugify`` helper. The stub raises, so RED genuinely fails; +the implementation genuinely passes. +""" + +SAMPLE_GOAL = "add a slugify(text) helper that makes URL-safe slugs" + +SAMPLE_SPEC = ( + "slugify(text) lowercases the input, replaces runs of non-alphanumeric " + "characters with single hyphens, and strips leading/trailing hyphens. " + "Empty or symbol-only input returns ''. Input is str; output is str." +) + +TEST_FILE = "test_slug_util.py" +IMPL_FILE = "slug_util.py" + +SAMPLE_TESTS = '''\ +"""Contract tests for slug_util.slugify (written before the implementation).""" +from slug_util import slugify + + +def test_lowercases_and_hyphenates(): + assert slugify("Hello World") == "hello-world" + + +def test_collapses_symbol_runs(): + assert slugify("a b!!c") == "a-b-c" + + +def test_strips_edge_hyphens(): + assert slugify("--Already--Slugged--") == "already-slugged" + + +def test_empty_and_symbol_only(): + assert slugify("") == "" + assert slugify("!!!") == "" +''' + +SAMPLE_STUB = '''\ +"""URL-slug helper (stub — implementation comes after RED).""" + + +def slugify(text): + raise NotImplementedError +''' + +SAMPLE_IMPL = '''\ +"""URL-slug helper.""" +import re + +_RUNS = re.compile(r"[^a-z0-9]+") + + +def slugify(text): + return _RUNS.sub("-", str(text).lower()).strip("-") +''' + +SAMPLE_QUIZ_QUESTIONS = [ + "What does slugify return for symbol-only input, and which test proves it?", + "Why are runs of symbols collapsed to a single hyphen instead of one each?", + "Which step would catch a regex that also stripped digits?", +] diff --git a/src/agent_ultra/bob/workflows/__init__.py b/src/agent_ultra/bob/workflows/__init__.py new file mode 100644 index 0000000..de94c24 --- /dev/null +++ b/src/agent_ultra/bob/workflows/__init__.py @@ -0,0 +1,8 @@ +"""Bundled ultracode workflow scripts used by bob steps 6 and 7.""" + +from pathlib import Path + +WORKFLOWS_DIR = Path(__file__).resolve().parent + +SECURITY_WORKFLOW = WORKFLOWS_DIR / "security_fanout.py" +REVIEW_WORKFLOW = WORKFLOWS_DIR / "review_fanout.py" diff --git a/src/agent_ultra/bob/workflows/review_fanout.py b/src/agent_ultra/bob/workflows/review_fanout.py new file mode 100644 index 0000000..49fc7d0 --- /dev/null +++ b/src/agent_ultra/bob/workflows/review_fanout.py @@ -0,0 +1,49 @@ +"""bob step 7 — WORKFLOW review as an ultracode workflow. + +One agent per review dimension (correctness, security, performance, style, +spec-compliance) examines the change independently; results are merged. The +bob gate cross-checks the ultracode run receipt and demands non-empty +dimensions — a claimed-but-never-run review blocks the commit. +""" + +META = { + "name": "bob-review", + "description": "bob WORKFLOW step: multi-dimension review fan-out", + "phases": ["Fanout", "Merge"], +} + +DIMENSIONS = ("correctness", "security", "performance", "style", + "spec-compliance") + +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 change") + spec = str(args.get("spec", "") or "")[:2000] + excerpt = str(args.get("target_excerpt", "") or "")[:6000] + + wf.phase("Fanout") + results = await wf.parallel([ + (lambda dim=dim: wf.agent( + f"Review this change on the {dim} dimension ONLY.\n" + f"Task: {goal}\nSpec: {spec}\nCode:\n{excerpt}\n" + "Return your verdict and notes.", + schema=REVIEW_SCHEMA, label=f"review:{dim}")) + for dim in DIMENSIONS + ]) + + wf.phase("Merge", kind="fan_in") + merged = {dim: (res or {"verdict": "concerns", + "notes": ["agent failed — treat as unreviewed"]}) + for dim, res in zip(DIMENSIONS, results)} + return {"dimensions": list(DIMENSIONS), "reviews": merged, + "agents": len(DIMENSIONS)} diff --git a/src/agent_ultra/bob/workflows/security_fanout.py b/src/agent_ultra/bob/workflows/security_fanout.py new file mode 100644 index 0000000..4d772e2 --- /dev/null +++ b/src/agent_ultra/bob/workflows/security_fanout.py @@ -0,0 +1,63 @@ +"""bob step 6 — SECURITY-FANOUT as an ultracode workflow. + +Independent security-focused agents (one per attack lens) each review the +same code and return structured findings. The ultracode engine journals every +call and writes its checksummed run receipt — that receipt is what the bob +gate cross-checks, so the fan-out cannot be claimed without having run. +""" + +META = { + "name": "bob-security", + "description": "bob SECURITY-FANOUT: independent security lenses over the change", + "phases": ["Fanout", "Merge"], +} + +LENSES = ("injection-and-input-handling", "secrets-and-data-exposure", + "resource-and-denial", "logic-and-authz") + +FINDINGS_SCHEMA = { + "type": "object", + "properties": { + "findings": { + "type": "array", + "items": { + "type": "object", + "properties": { + "id": {"type": "string"}, + "severity": {"type": "string", + "enum": ["low", "medium", "high", "critical"]}, + "triage": {"type": "string", + "enum": ["real_now", "real_later", "theoretical"]}, + "claim": {"type": "string"}, + }, + "required": ["id", "severity", "triage", "claim"], + }, + }, + }, + "required": ["findings"], +} + + +async def run(wf): + args = wf.args if isinstance(wf.args, dict) else {} + goal = str(args.get("goal", "") or "the change") + excerpt = str(args.get("target_excerpt", "") or "")[:6000] + + wf.phase("Fanout") + results = await wf.parallel([ + (lambda lens=lens: wf.agent( + f"You are a security reviewer using ONLY the {lens} lens.\n" + f"Task under review: {goal}\n" + f"Code:\n{excerpt}\n" + "Report concrete findings for this lens (empty list if none).", + schema=FINDINGS_SCHEMA, label=f"sec:{lens}")) + for lens in LENSES + ]) + + wf.phase("Merge", kind="fan_in") + findings = [] + for lens, res in zip(LENSES, results): + for f in ((res or {}).get("findings") or []): + findings.append({**f, "lens": lens}) + return {"lenses": list(LENSES), "findings": findings, + "agents": len(LENSES)} diff --git a/tests/test_bob.py b/tests/test_bob.py new file mode 100644 index 0000000..49bec1a --- /dev/null +++ b/tests/test_bob.py @@ -0,0 +1,424 @@ +"""bob pipeline: receipt chain, gate enforcement, and anti-fake proofs. + +The expensive full mock pipeline runs ONCE (module fixture); every tampering +scenario works on a disposable copy of that run so the proofs stay independent. +Fully offline: mock routes, real pytest subprocesses, real ultracode runs. +""" + +import io +import json +import shutil + +import pytest + +from agent_ultra.bob import ( + BobRun, BobProofError, GATED_STEPS, + build_step_receipt, validate_chain, gate_check, run_pytest_step, + run_bob, assert_bob_done, +) +from agent_ultra.bob.pipeline import ( + check_ultracode_evidence, safe_rel, parse_pytest_nodes, +) + +KEY = b"bob-test-key-0123456789abcdef" + + +# -------------------------------------------------------------------------- +# receipt chain (pure, no subprocess) +# -------------------------------------------------------------------------- + +def _chain(n=3, key=KEY): + steps = ["step01_spec", "step02_red", "step03_green"][:n] + out, prev = [], "" + for i, step in enumerate(steps): + rec = build_step_receipt(step=step, run_id="r1", seq=i, + writer="agent", evidence={"i": i}, + prev_sha256=prev, key=key) + prev = rec["receipt_sha256"] + out.append(rec) + return out + + +def test_valid_chain_has_no_errors(): + assert validate_chain(_chain(), KEY) == [] + + +def test_edited_receipt_breaks_integrity_and_chain(): + chain = _chain() + chain[1]["evidence"]["i"] = 99 # tamper after signing + errs = "\n".join(validate_chain(chain, KEY)) + assert "integrity FAIL" in errs + + +def test_unsigned_receipt_is_a_forgery(): + chain = _chain(key=None) # correct sha256, no HMAC + errs = "\n".join(validate_chain(chain, KEY)) + assert "authenticity FAIL" in errs + + +def test_wrong_key_fails_closed(): + errs = "\n".join(validate_chain(_chain(), b"some-other-install-key")) + assert "authenticity FAIL" in errs + + +def test_reordered_chain_breaks_linkage(): + chain = _chain() + chain[1], chain[2] = chain[2], chain[1] + errs = "\n".join(validate_chain(chain, KEY)) + assert "seq" in errs or "prev_sha256" in errs + + +def test_removed_middle_receipt_breaks_linkage(): + chain = _chain() + del chain[1] + errs = "\n".join(validate_chain(chain, KEY)) + assert "prev_sha256" in errs + + +# -------------------------------------------------------------------------- +# RED/GREEN system runner +# -------------------------------------------------------------------------- + +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") + 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 + + +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") + ok, _rec, msg = run_pytest_step(run, "red", "test_t.py") + assert not ok + + +# -------------------------------------------------------------------------- +# the full mock pipeline (one real run, many disposable copies) +# -------------------------------------------------------------------------- + +@pytest.fixture(scope="module") +def bob_world(tmp_path_factory): + base = tmp_path_factory.mktemp("bobworld") + outcome = run_bob("add a slugify helper", base / "ws", mock=True, + key_path=base / "bob.key", + ultracode_home=base / "uc", + interactive=False, out=io.StringIO()) + assert outcome.passed, f"mock pipeline must pass: {outcome.blocked_at}" + return base, outcome + + +@pytest.fixture() +def world_copy(bob_world, tmp_path): + """A disposable copy of the passing run for tampering scenarios.""" + base, outcome = bob_world + shutil.copytree(base / "ws", tmp_path / "ws") + shutil.copy(base / "bob.key", tmp_path / "bob.key") + ws = tmp_path / "ws" + run = BobRun(workspace=ws, + run_dir=ws / ".agent-ultra" / "bob" / "runs" / outcome.run_id, + key=(base / "bob.key").read_bytes()) + return run + + +def test_mock_run_completes_and_gate_passes(bob_world): + base, outcome = bob_world + assert outcome.passed and outcome.gate.passed + assert (base / "ws" / ".agent-ultra" / "bob" / "runs" / outcome.run_id + / "gate-pass.json").is_file() + assert (base / "ws" / ".agent-ultra" / "bob" / "runs" / outcome.run_id + / "report.json").is_file() + # the run was sealed: ACTIVE released + assert not (base / "ws" / ".agent-ultra" / "bob" / "ACTIVE").exists() + + +def test_every_gated_step_left_a_receipt(world_copy): + have = {r["step"] for r in world_copy.chain()} + assert set(GATED_STEPS) <= have + + +def test_mock_receipts_blocked_outside_demo_mode(world_copy): + result = gate_check(world_copy, allow_mock=False, rerun_tests=False) + assert not result.passed + assert any("mock" in e for e in result.errors) + + +def test_skipped_step_blocks(world_copy): + world_copy.receipt_path("step06_security").unlink() + result = gate_check(world_copy, allow_mock=True, rerun_tests=False) + assert not result.passed + assert any("step06_security" in e and "missing" in e + for e in result.errors) + + +def test_fake_fanout_claim_blocks(bob_world, world_copy): + """A re-signed receipt naming an invented ultracode run must block.""" + base, _ = bob_world + world_copy.receipt_path("step07_workflow").unlink() + world_copy.write("step07_workflow", + {"ultracode": {"run_id": "20990101T000000-fake-000000", + "home": str(base / "uc")}, + "dimensions": ["invented"]}, + writer="engine", mock=True) + result = gate_check(world_copy, allow_mock=True, rerun_tests=False) + assert not result.passed + assert any("fabricated" in e or "no receipt on disk" in e + for e in result.errors) + + +def test_doctored_ultracode_receipt_blocks(): + """check_ultracode_evidence itself rejects a nonexistent receipt.json.""" + errs = check_ultracode_evidence( + {"ultracode": {"run_id": "nope", "home": "does/not/exist"}}, + allow_mock=True, receipt_mock=True) + assert errs and ("fabricated" in errs[0] or "no receipt on disk" in errs[0]) + + +def test_fake_panel_claim_blocks(world_copy): + receipt = (world_copy.run_dir / "panel" / "panel_execution_receipt.json") + body = json.loads(receipt.read_text(encoding="utf-8")) + body["lens_count_executed"] = 99 # checksum now stale + receipt.write_text(json.dumps(body), encoding="utf-8") + result = gate_check(world_copy, allow_mock=True, rerun_tests=False) + assert not result.passed + assert any("step08_panel" in e for e in result.errors) + + +def test_quiz_passed_without_operator_response_blocks(world_copy): + world_copy.receipt_path("step09_quiz").unlink() + world_copy.write("step09_quiz", + {"questions": ["q1"], "outcome": "passed", + "operator_response": ""}, + writer="agent", mock=True) + result = gate_check(world_copy, allow_mock=True, rerun_tests=False) + assert not result.passed + assert any("quiz nobody answered" in e for e in result.errors) + + +def test_stale_file_blocks(world_copy): + impl = world_copy.workspace / "slug_util.py" + impl.write_text(impl.read_text(encoding="utf-8") + "\n# edited later\n", + encoding="utf-8") + result = gate_check(world_copy, allow_mock=True, rerun_tests=False) + assert not result.passed + assert any("stale" in e for e in result.errors) + + +def test_live_rerun_catches_broken_tests(world_copy): + """Even with pristine receipts, the gate re-runs pytest — a state that + fails its own tests blocks regardless of what the chain says. + (Here the receipts also go stale; the live re-run must ALSO fire.)""" + impl = world_copy.workspace / "slug_util.py" + impl.write_text("def slugify(text):\n return 'broken'\n", + encoding="utf-8") + result = gate_check(world_copy, allow_mock=True, rerun_tests=True) + assert not result.passed + assert any("live re-run" in e for e in result.errors) + + +def test_assert_bob_done_raises_while_unproven(tmp_path): + run = BobRun.start(tmp_path / "ws", goal="g", + key_path=tmp_path / "k.key") + with pytest.raises(BobProofError): + assert_bob_done(tmp_path / "ws", key_path=tmp_path / "k.key") + assert run.run_id # started, never proven + + +def test_assert_bob_done_none_when_no_active_run(tmp_path): + assert assert_bob_done(tmp_path, key_path=tmp_path / "k.key") is None + + +# -------------------------------------------------------------------------- +# hardening: findings from the adversarial review +# -------------------------------------------------------------------------- + +def test_safe_rel_rejects_absolute_and_traversal(tmp_path): + assert safe_rel(tmp_path, "sub/x.py") == "sub/x.py" + for bad in ("../evil.py", "../../x", "/etc/passwd", "C:/Windows/x"): + try: + safe_rel(tmp_path, bad) + except ValueError: + continue + raise AssertionError(f"{bad!r} should have been rejected") + + +def test_pytest_parser_handles_parametrized_space_ids(): + out = ("test_x.py::test_a[hello world] PASSED [ 50%]\n" + "test_x.py::test_b[a b] FAILED [100%]\n") + nodes = parse_pytest_nodes(out) + assert nodes == {"test_x.py::test_a[hello world]": "PASSED", + "test_x.py::test_b[a b]": "FAILED"} + + +def test_borrowed_ultracode_run_blocks(bob_world, world_copy): + """A valid receipt from the WRONG workflow (e.g. a trivial prior run) must + not satisfy a gated fan-out step.""" + base, _ = bob_world + # point step07 (expects bob-review) at the step06 security run instead + sec = world_copy.load("step06_security")["evidence"]["ultracode"] + world_copy.receipt_path("step07_workflow").unlink() + world_copy.write("step07_workflow", + {"ultracode": {**sec, "workflow_name": "bob-security"}, + "excerpt_sha256": "x", "dimensions": ["d"]}, + writer="engine", mock=True) + result = gate_check(world_copy, allow_mock=True, rerun_tests=False) + assert not result.passed + assert any("bob-review" in e for e in result.errors) + + +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) + assert not result.passed + assert any("fingerprint" in e for e in result.errors) + + +def test_borrowed_panel_receipt_blocks(bob_world, world_copy): + """A structurally valid panel receipt for a DIFFERENT task must fail the + step-8 binding (expect_task_id / expect_artifact_hash).""" + import json as _json + receipt = world_copy.run_dir / "panel" / "panel_execution_receipt.json" + body = _json.loads(receipt.read_text(encoding="utf-8")) + # forge a self-consistent receipt for a different task/artifact + from agent_ultra.panel_receipt.receipt import _sha256 + body["task_id"] = "some-other-run" + b = {k: v for k, v in body.items() if k != "receipt_sha256"} + body["receipt_sha256"] = _sha256(_json.dumps(b, sort_keys=True, + ensure_ascii=False)) + receipt.write_text(_json.dumps(body), encoding="utf-8") + result = gate_check(world_copy, allow_mock=True, rerun_tests=False) + assert not result.passed + assert any("step08_panel" in e for e in result.errors) + + +def test_malformed_receipt_fails_closed_not_crash(world_copy): + import json as _json + p = world_copy.receipt_path("step02_red") + body = _json.loads(p.read_text(encoding="utf-8")) + body["evidence"]["counts"] = "not-a-dict" + p.write_text(_json.dumps(body), encoding="utf-8") + # must return a blocked GateResult, never raise + result = gate_check(world_copy, allow_mock=True, rerun_tests=False) + assert not result.passed + + +def test_deleting_run_dir_blocks_via_orphaned_marker(bob_world, tmp_path): + """A sealed marker pointing at a missing run dir is tampering, not 'done'.""" + import shutil as _shutil + base, outcome = bob_world + _shutil.copytree(base / "ws", tmp_path / "ws") + _shutil.copy(base / "bob.key", tmp_path / "bob.key") + root = tmp_path / "ws" / ".agent-ultra" / "bob" + (root / "ACTIVE").write_text(outcome.run_id, encoding="utf-8") + _shutil.rmtree(root / "runs" / outcome.run_id) # delete the receipts + run = BobRun.active(tmp_path / "ws", key_path=tmp_path / "bob.key") + assert run is not None and run.marker_orphaned() + assert not gate_check(run, allow_mock=True, rerun_tests=False).passed + + +def test_start_refuses_to_orphan_an_active_run(tmp_path): + BobRun.start(tmp_path / "ws", goal="first", key_path=tmp_path / "k.key") + try: + BobRun.start(tmp_path / "ws", goal="second", key_path=tmp_path / "k.key") + 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", + key_path=tmp_path / "k.key", force=True) + assert r2.run_id + + +def test_mock_panel_receipt_blocked_without_allow_mock(world_copy): + """The mock flag must be enforced on step 8 too, not only steps 6/7.""" + result = gate_check(world_copy, allow_mock=False, rerun_tests=False) + assert not result.passed + assert any("panel" in e and "mock" in e for e in result.errors) + + +# -------------------------------------------------------------------------- +# commit lifecycle: mark at pre-commit, seal at post-commit (deadlock fix) +# -------------------------------------------------------------------------- + +import os +import subprocess +import sys + + +def _git(ws, *args, env=None): + e = dict(os.environ) + e.update(env or {}) + return subprocess.run(["git", "-C", str(ws)] + list(args), + capture_output=True, text=True, timeout=120, env=e) + + +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") + marker = tmp_path / "ws" / ".agent-ultra" / "bob" / "ACTIVE" + # seal without a recorded pass must refuse + assert run.seal() is False + assert marker.is_file() + run.mark_pass({"passed": True}) + assert marker.is_file(), "mark_pass must NOT release ACTIVE" + assert (run.run_dir / "gate-pass.json").is_file() + assert run.seal() is True + assert not marker.exists(), "seal releases ACTIVE after the commit lands" + + +def test_commit_deadlock_lifecycle(tmp_path, monkeypatch): + """The exact deadlock regression: a proven run must survive its own + pre-commit gate, the commit must land, post-commit must seal, and a + second commit without a new run must be blocked under REQUIRE_RUN.""" + home = tmp_path / "home" + monkeypatch.setenv("AGENT_ULTRA_HOME", str(home)) + ws = tmp_path / "ws" + + # a passing pipeline that stays ACTIVE (git-workflow lifecycle) + outcome = run_bob("add a slugify helper", ws, mock=True, + interactive=False, seal_on_pass=False, out=io.StringIO()) + assert outcome.passed + marker = ws / ".agent-ultra" / "bob" / "ACTIVE" + assert marker.is_file(), "run must stay ACTIVE until the commit lands" + + # a real git repo with the split hooks (demo mode: --allow-mock) + assert _git(ws, "init", "-q").returncode == 0 + _git(ws, "config", "user.email", "bob@example.com") + _git(ws, "config", "user.name", "bob") + from agent_ultra.bob import install_hooks + install_hooks(ws, python=sys.executable, gate_flags="--allow-mock") + + (ws / ".gitignore").write_text("__pycache__/\n", encoding="utf-8") + _git(ws, "add", "-A") + + env = {"AGENT_ULTRA_REQUIRE_RUN": "1", "AGENT_ULTRA_HOME": str(home)} + p = _git(ws, "commit", "-m", "feat: slugify (bob run proven)", env=env) + assert p.returncode == 0, ( + f"the proven commit must LAND (deadlock regression):\n" + f"stdout={p.stdout}\nstderr={p.stderr}") + + # post-commit sealed the run — ACTIVE gone only AFTER the commit + assert not marker.exists(), "post-commit must seal (release ACTIVE)" + gp = json.loads((ws / ".agent-ultra" / "bob" / "runs" / outcome.run_id + / "gate-pass.json").read_text(encoding="utf-8")) + head = _git(ws, "rev-parse", "HEAD").stdout.strip() + assert gp.get("commit") == head, "seal must stamp the landed commit hash" + assert gp.get("sealed_utc"), "seal must record when it sealed" + + # a second commit with NO active run is blocked under REQUIRE_RUN=1 + (ws / "unproven.py").write_text("x = 1\n", encoding="utf-8") + _git(ws, "add", "unproven.py") + p2 = _git(ws, "commit", "-m", "sneaky: no pipeline run", env=env) + assert p2.returncode != 0, "commit without an active run must be blocked" + assert "AGENT_ULTRA_REQUIRE_RUN" in (p2.stdout + p2.stderr) + # without the policy env, non-interference still applies + p3 = _git(ws, "commit", "-m", "operator commit (no policy)", + env={"AGENT_ULTRA_HOME": str(home)}) + assert p3.returncode == 0