diff --git a/docs/phase-568-slice-6-journal-authority-plan.md b/docs/phase-568-slice-6-journal-authority-plan.md index 9a978058..6f66da96 100644 --- a/docs/phase-568-slice-6-journal-authority-plan.md +++ b/docs/phase-568-slice-6-journal-authority-plan.md @@ -1,5 +1,9 @@ # Issue #568 Slice 6: Journal Authority Implementation Plan +> Historical slice note. This plan records the slice 6 opt-in rollout. Slice 11 +> superseded that rollout: new runs request journal authority by default, and +> `BRIGADE_RUN_JOURNAL_AUTHORITY` no longer controls runtime behavior. + Goal in one sentence: cut `run.json` over to a journal-authoritative projected snapshot for newly enrolled runs while leaving legacy and lifecycle-only runs byte-identical to slice 5. diff --git a/docs/phase-568-slice-6-journal-authority.md b/docs/phase-568-slice-6-journal-authority.md index a51a4a4d..1f10f388 100644 --- a/docs/phase-568-slice-6-journal-authority.md +++ b/docs/phase-568-slice-6-journal-authority.md @@ -1,5 +1,10 @@ # Issue #568 Slice 6: Journal Authority +> Historical slice note. This document records the slice 6 opt-in rollout. +> Slice 11 superseded that rollout: new runs request journal authority by +> default, and `BRIGADE_RUN_JOURNAL_AUTHORITY` no longer controls runtime +> behavior. + Reviewed design specification for the approved slice 6 boundary of GitHub issue #568 (https://github.com/escoffier-labs/brigade/issues/568). Slices 1 through 5 landed the append-only journal kernel (`run_journal`, `run_events`), opt-in lifecycle journaling diff --git a/docs/receipt-schemas.md b/docs/receipt-schemas.md index 88f571f4..e64d4083 100644 --- a/docs/receipt-schemas.md +++ b/docs/receipt-schemas.md @@ -243,6 +243,13 @@ original file is missing, corrupt, or not an object. | `artifact_collection` | object | no | Artifact-retention result | | `resumed_at` | array of string | no | ISO-8601 resume timestamps | | `recovery_history` | array of object | no | Prior failure objects retained after a successful resume | +| `lifecycle_journal_requested` | boolean | no | Durable enrollment request. Present and `true` on every new run | +| `run_journal_authority_requested` | boolean | no | Durable authority request. Present and `true` on every new run | +| `projector_version` | integer | no | Journal projector version used for the compatibility snapshot | +| `journal_present` | boolean | no | Whether the verified lifecycle journal exists | +| `journal_last_sequence` | integer | no | Last event sequence applied to this snapshot | +| `journal_last_event_digest` | string / null | no | Digest at `journal_last_sequence`. Null only at sequence zero | +| `approval_reference` | object | no | Redacted approval identity, source, fingerprints, and decision state | **Partial stale-recovery variant** @@ -317,10 +324,27 @@ writes this smaller receipt: | `worktree` | string | Detached worktree path when used | | `failure` | object | Same shape as `failure` when collection failed | -**Lifecycle:** this file is **updated in place** during a run. Treat each write as -the latest snapshot, not an append-only log. Sidecars (`roster.json`, `plan.json`, -`worker-results.json`, `synthesis.json`) are write-once per phase (resume salvage -and patch-ref binding may rewrite worker/synthesis artifacts). +**Lifecycle:** journal authority is the default for every new run. +`events/lifecycle.jsonl` is the append-only lifecycle record and `run.json` is its +latest `brigade.run.v1` compatibility projection. Existing run directories that +lack both durable request fields remain snapshot-only and are not migrated in +place. Readers must ignore additive keys. A paused approval projects `status: +running` so a previous version still sees a known nonterminal status. + +Do not create a journal or add durable request fields to a legacy receipt by +hand. Journal-aware writers verify the bounded chain and checkpoint before +replacing the snapshot. An older release may inspect the additive `run.json` +shape, but operators must roll forward before it writes, recovers, or resumes a +journal-authoritative run. A retry after an approval action redeemed its claim +but exited before outcome persistence verifies the same run and fingerprints. +Daily approvals also bind the redeemed claim to the exact completed Daily run +receipt. A missing or changed completion receipt is not recoverable by retry. +The source-store lock spans validation, any missing `approval.consumed` and +`run.resumed` facts, and the refreshed snapshot. Review writes wait until that +transaction finishes. Reconciliation does not execute the action again. +Sidecars (`roster.json`, `plan.json`, `worker-results.json`, `synthesis.json`) +are write-once per phase (resume salvage and patch-ref binding may rewrite +worker/synthesis artifacts). --- diff --git a/docs/technical-guide.md b/docs/technical-guide.md index 53e0b427..6eec5e77 100644 --- a/docs/technical-guide.md +++ b/docs/technical-guide.md @@ -373,6 +373,47 @@ Common `brigade run` flags: - `--read-only` tells the orchestrator and workers to inspect and recommend only. - `--sandbox {read-only,workspace-write,danger-full-access}` overrides the native Codex sandbox mode from the roster. +### Run journal authority and upgrade behavior + +Every new run directory records `lifecycle_journal_requested: true` and +`run_journal_authority_requested: true`. Once the run holds its matching lock, +`events/lifecycle.jsonl` becomes the lifecycle authority and `run.json` stays the +`brigade.run.v1` compatibility snapshot. There is no environment setting that +disables journal authority for a new run. + +Existing run directories are classified from their stored artifacts. A run whose +`run.json` lacks both durable request fields remains snapshot-only, even when a +newer Brigade release records another start attempt. Do not add the fields or +create `events/lifecycle.jsonl` by hand. Keep legacy runs on the snapshot path, or +start a new run after upgrading. + +The compatibility snapshot keeps schema version 1 and changes additively. Readers +must ignore unknown keys. A paused approval still projects the known nonterminal +status `running`. `approval_reference.decision_state` carries the pause state for +new readers. This lets the previous release inspect `run.json` during the +compatibility window, but it must not resume or write a journal-authoritative run. +Upgrade the writer before using `runs recover` or `runs resume` on one of those +runs. + +Use `brigade runs show ` or `brigade runs watch ` for inspection. Use +`brigade runs recover ` when the snapshot is missing, corrupt, or behind a +verified journal checkpoint. Recovery validates the bounded event chain and +checkpoint before replacing `run.json`. If rollback leaves an older Brigade +unable to interpret a journal event or projector version, stop that writer and +roll forward. The append-only journal format is a one-way storage boundary. + +If the approved action completed but the process exited before recording +`approval.consumed` and `run.resumed`, run `brigade runs resume ` again. +Brigade verifies that the redeemed claim belongs to the same run and still +matches the stored approval fingerprints. For a Daily action, it also requires +the lock-bound completion reference to match the exact successful receipt under +`.brigade/daily/runs/`. A missing, failed, or changed receipt fails closed. After +those checks, Brigade retains the source-store lock while it records only the +missing journal facts and refreshes `run.json`. Review changes wait for that +transaction and cannot rewrite a consumed Daily approval or completed Tool +call. Recovery does not consume the approval or run the action again. A +redeemed claim owned by another run still fails closed. + For `codex` agents, `--read-only` also passes `codex exec --sandbox read-only`. Combine `--sandbox` with `--read-only` to keep prompt-level read-only rules while overriding native Codex sandbox behavior. This override does not weaken adapters whose read-only flag takes precedence, including Cursor plan mode, Antigravity's sandbox, Kimi's plan mode, Aider's dry run, and Codex Cloud's remote isolation. Brigade's warning follows the command each adapter will execute. diff --git a/src/brigade/aboyeur.py b/src/brigade/aboyeur.py index 6418e020..9b3bf420 100644 --- a/src/brigade/aboyeur.py +++ b/src/brigade/aboyeur.py @@ -61,20 +61,11 @@ BRIEF_BUDGET_BYTES = 6000 NOOP_DETAIL = "no-op" -# Journal-authority opt-in (issue #568 slice 6). Per-run and durable, like the -# lifecycle-journal flag. The authority opt-in implies lifecycle journaling: -# enrolling a new run with BRIGADE_RUN_JOURNAL_AUTHORITY=1 sets BOTH request -# fields true even when BRIGADE_LIFECYCLE_JOURNAL is unset. Existing runs -# enroll only from their durable run.json field, never a later environment -# change. -_AUTHORITY_FLAG_ENV = "BRIGADE_RUN_JOURNAL_AUTHORITY" +# Journal authority is the default for every new run (issue #568 slice 11). +# Enrollment stays durable and per-run: existing runs are classified only from +# their stored run.json fields, so legacy snapshot-only runs are never migrated +# by a later Brigade release or environment change. _AUTHORITY_REQUEST_FIELD = "run_journal_authority_requested" -_AUTHORITY_TRUTHY = frozenset({"1", "true", "yes", "on"}) - - -def is_run_journal_authority_enabled() -> bool: - """True when the opt-in journal-authority flag is set in the environment.""" - return os.environ.get(_AUTHORITY_FLAG_ENV, "").strip().lower() in _AUTHORITY_TRUTHY # A plan-mode seat has no write tool, so any file it tries to create fails, and a @@ -2835,18 +2826,13 @@ def record_run_start( existing_lifecycle_requested = True if existing.get(_AUTHORITY_REQUEST_FIELD) is True: existing_authority_requested = True - # A new run enrolls in lifecycle journaling when the lifecycle-journal flag - # is set OR when the authority opt-in is set: BRIGADE_RUN_JOURNAL_AUTHORITY=1 - # implies lifecycle journaling even when BRIGADE_LIFECYCLE_JOURNAL is unset, - # so a new authority run carries BOTH durable request fields. An existing - # run enrolls only from its durable run.json fields: a later environment - # change never enrolls it. + # Every new run carries both durable request fields. Existing runs enroll + # only from their stored run.json fields, so legacy snapshot-only runs stay + # untouched. An authority request implies lifecycle journaling even when an + # older or repaired receipt retained only the authority field. new_run = not run_json_exists - authority_env = is_run_journal_authority_enabled() - lifecycle_requested = existing_lifecycle_requested or ( - new_run and (run_lifecycle.is_lifecycle_journaling_enabled() or authority_env) - ) - authority_requested = existing_authority_requested or (new_run and authority_env) + lifecycle_requested = existing_lifecycle_requested or existing_authority_requested or new_run + authority_requested = existing_authority_requested or new_run # The first run.json write activates the lifecycle journal and publishes a # recovery checkpoint BEFORE the atomic run.json replacement. If that final # replacement fails, durable journal/checkpoint state already exists without diff --git a/src/brigade/daily_cmd/approvals.py b/src/brigade/daily_cmd/approvals.py index f13a81df..6b2aa00e 100644 --- a/src/brigade/daily_cmd/approvals.py +++ b/src/brigade/daily_cmd/approvals.py @@ -89,11 +89,22 @@ def _approval_lock_path(target: Path, approval_id: str) -> Path: return _approvals_root(target).parent / ".approval-locks" / f"{approval_id}.lock" +def _acquire_approval_store_lock(path: Path) -> runguard._LockOwnership: + deadline = time.monotonic() + 5.0 + while True: + try: + return runguard._acquire_lock(path) + except runguard.RunLockError as exc: + if "another brigade run appears active" not in str(exc) or time.monotonic() >= deadline: + raise + time.sleep(0.01) + + @contextmanager def _approval_store_lock(target: Path, approval_id: str) -> Iterator[None]: path = _approval_lock_path(target, approval_id) path.parent.mkdir(parents=True, exist_ok=True) - ownership = runguard._acquire_lock(path) + ownership = _acquire_approval_store_lock(path) try: yield finally: @@ -367,6 +378,188 @@ def redeem_for_run(target: Path, approval_id: str, reference: Mapping[str, str], return ApprovalClaim(decided_at=decided_at, claimed_at=redeemed_at, already_claimed=False) +def record_redeemed_action_completed( + target: Path, + approval_id: str, + reference: Mapping[str, str], + run_receipt: Mapping[str, Any], +) -> bool: + """Bind a successful Daily action receipt to its redeemed approval claim.""" + with _approval_store_lock(target, approval_id): + approval = _find_approval(target, approval_id) + if approval is None: + raise ApprovalClaimError(f"daily approval not found: {approval_id}") + actual = _approval_claim_reference(approval) + if any(reference.get(field) != actual[field] for field in _APPROVAL_REFERENCE_FIELDS): + raise ApprovalClaimError("daily approval fingerprints changed after action") + claim = approval.get("approval_claim") + if claim is None: + return False + if not isinstance(claim, Mapping) or claim.get("state") != "redeemed": + raise ApprovalClaimError("daily approval claim changed after action") + owner_run_id = claim.get("run_id") + redeemed_at = claim.get("redeemed_at") + if not isinstance(owner_run_id, str) or not owner_run_id: + raise ApprovalClaimError("daily approval redeemed claim has no run id") + if not isinstance(redeemed_at, str) or not redeemed_at: + raise ApprovalClaimError("daily approval redeemed claim has no redemption timestamp") + if ( + approval.get("status") != "consumed" + or approval.get("consumed_run_id") != owner_run_id + or approval.get("consumed_at") != redeemed_at + ): + raise ApprovalClaimError(f"daily approval redeemed source state is {approval.get('status') or 'unknown'}") + + daily_run_id = run_receipt.get("run_id") + completed_at = run_receipt.get("completed_at") + action_id = approval.get("selected_action_id") + selected_action = ( + run_receipt.get("selected_action") if isinstance(run_receipt.get("selected_action"), Mapping) else {} + ) + if ( + run_receipt.get("status") != "completed" + or run_receipt.get("approval_id") != approval_id + or run_receipt.get("selected_action_id") != action_id + or selected_action.get("source_fingerprint") != approval.get("source_fingerprint") + or not isinstance(daily_run_id, str) + or not daily_run_id + or not isinstance(completed_at, str) + or not completed_at + ): + raise ApprovalClaimError("daily approval action receipt does not prove successful completion") + + completed = { + "state": "completed", + "owner_run_id": owner_run_id, + "daily_run_id": daily_run_id, + "action_id": action_id, + "source_fingerprint": approval.get("source_fingerprint"), + "completed_at": completed_at, + } + existing = approval.get("approval_action_receipt") + if existing is not None and existing != completed: + raise ApprovalClaimError("daily approval action receipt conflicts with stored completion") + approval["approval_action_receipt"] = completed + _write_approval_unlocked(target, approval) + return True + + +def _redeemed_reconciliation_blockers( + approval: Mapping[str, Any], + config: dict[str, Any], +) -> list[str]: + blockers: list[str] = [] + if approval.get("config_fingerprint") != _config_fingerprint(config): + blockers.append("daily config changed since approval") + selected_action = approval.get("selected_action") + if not isinstance(selected_action, dict) or not selected_action: + blockers.append("daily approval selected action is malformed") + elif approval.get("selected_adapter") != _adapter_for(selected_action): + blockers.append("daily approval adapter changed since approval") + return blockers + + +def _validate_redeemed_for_run_unlocked( + target: Path, + approval_id: str, + reference: Mapping[str, str], + run_id: str, +) -> ApprovalClaim: + """Validate one redeemed Daily claim while its source-store lock is held.""" + approval = _find_approval(target, approval_id) + if approval is None: + raise ApprovalClaimError(f"daily approval not found: {approval_id}") + actual = _approval_claim_reference(approval) + if any(reference.get(field) != actual[field] for field in _APPROVAL_REFERENCE_FIELDS): + raise ApprovalClaimError("daily approval fingerprints changed before reconciliation") + claim = approval.get("approval_claim") + if not isinstance(claim, Mapping) or claim.get("state") != "redeemed": + raise ApprovalClaimError("daily approval has no redeemed claim to reconcile") + claim_run_id = claim.get("run_id") + if claim_run_id != run_id: + owner = claim_run_id if isinstance(claim_run_id, str) and claim_run_id else "another run" + raise ApprovalClaimError(f"daily approval redeemed claim belongs to {owner}") + redeemed_at = claim.get("redeemed_at") + if not isinstance(redeemed_at, str) or not redeemed_at: + raise ApprovalClaimError("daily approval redeemed claim has no redemption timestamp") + if ( + approval.get("status") != "consumed" + or approval.get("consumed_run_id") != run_id + or approval.get("consumed_at") != redeemed_at + ): + raise ApprovalClaimError(f"daily approval redeemed source state is {approval.get('status') or 'unknown'}") + + action_receipt = approval.get("approval_action_receipt") + if not isinstance(action_receipt, Mapping) or action_receipt.get("state") != "completed": + raise ApprovalClaimError("daily approval has no completed action receipt to reconcile") + if action_receipt.get("owner_run_id") != run_id: + owner = action_receipt.get("owner_run_id") + owner_text = owner if isinstance(owner, str) and owner else "another run" + raise ApprovalClaimError(f"daily approval completed action receipt belongs to {owner_text}") + daily_run_id = action_receipt.get("daily_run_id") + completed_at = action_receipt.get("completed_at") + if ( + action_receipt.get("action_id") != approval.get("selected_action_id") + or action_receipt.get("source_fingerprint") != approval.get("source_fingerprint") + or not isinstance(daily_run_id, str) + or not daily_run_id + or not isinstance(completed_at, str) + or not completed_at + ): + raise ApprovalClaimError("daily approval completed action receipt is malformed") + run_receipt = _read_json(_runs_root(target) / daily_run_id / "run.json") + selected_action = ( + run_receipt.get("selected_action") + if isinstance(run_receipt, Mapping) and isinstance(run_receipt.get("selected_action"), Mapping) + else {} + ) + if ( + not isinstance(run_receipt, Mapping) + or run_receipt.get("status") != "completed" + or run_receipt.get("run_id") != daily_run_id + or run_receipt.get("approval_id") != approval_id + or run_receipt.get("selected_action_id") != approval.get("selected_action_id") + or selected_action.get("source_fingerprint") != approval.get("source_fingerprint") + or run_receipt.get("completed_at") != completed_at + ): + raise ApprovalClaimError("daily approval completed action receipt no longer matches its run") + config, _ = _load_config(target) + blockers = _redeemed_reconciliation_blockers(approval, config) + if blockers: + raise ApprovalClaimError(f"daily approval is stale or blocked: {blockers[0]}") + decided_at = approval.get("reviewed_at") + if not isinstance(decided_at, str) or not decided_at: + raise ApprovalClaimError("daily approval decision has no review timestamp") + return ApprovalClaim( + decided_at=decided_at, + claimed_at=redeemed_at, + already_claimed=True, + ) + + +@contextmanager +def redeemed_reconciliation_guard( + target: Path, + approval_id: str, + reference: Mapping[str, str], + run_id: str, +) -> Iterator[ApprovalClaim]: + """Hold the Daily source-store lock through outcome reconciliation.""" + with _approval_store_lock(target, approval_id): + yield _validate_redeemed_for_run_unlocked(target, approval_id, reference, run_id) + + +def validate_redeemed_for_run( + target: Path, + approval_id: str, + reference: Mapping[str, str], + run_id: str, +) -> ApprovalClaim: + """Validate one redeemed Daily claim for outcome-only reconciliation.""" + with redeemed_reconciliation_guard(target, approval_id, reference, run_id) as claim: + return claim + + def _consume_approval(target: Path, approval: dict[str, Any], run_id: str) -> None: claim = approval.get("approval_claim") if isinstance(claim, Mapping) and claim.get("state") == "reserved": diff --git a/src/brigade/daily_cmd/run_loop.py b/src/brigade/daily_cmd/run_loop.py index d6ea9a00..edf89553 100644 --- a/src/brigade/daily_cmd/run_loop.py +++ b/src/brigade/daily_cmd/run_loop.py @@ -461,6 +461,13 @@ def run( receipt["completed_at"] = _now().isoformat() receipt["next_recommended_command"] = "brigade daily closeout" _record_run(target, receipt) + if rc == 0 and approval is not None: + record_redeemed_action_completed( + target, + str(approval.get("approval_id") or ""), + _approval_claim_reference(approval), + receipt, + ) _record_telemetry_event( target, { diff --git a/src/brigade/dogfood_cmd.py b/src/brigade/dogfood_cmd.py index 2dc41eea..478f060a 100644 --- a/src/brigade/dogfood_cmd.py +++ b/src/brigade/dogfood_cmd.py @@ -11,7 +11,7 @@ from pathlib import Path from typing import Any -from . import aboyeur, localio, toml_compat +from . import aboyeur, localio, runguard, toml_compat from . import agents from . import runs_cmd from .roster import Agent, Roster @@ -526,16 +526,31 @@ def run( else default_handoff_inbox(effective_target, effective_agent_cli) ) - rc = aboyeur.run( - task or DEFAULT_TASK, - _dogfood_roster(effective_timeout, effective_agent_cli), - show_plan=True, - cwd=effective_target, - output_dir=chosen_output_dir, - handoff_inbox=chosen_handoff_inbox, - read_only=True, - sandbox="read-only" if effective_native else None, - ) + run_task = task or DEFAULT_TASK + roster = _dogfood_roster(effective_timeout, effective_agent_cli) + try: + aboyeur.record_run_start( + chosen_output_dir, + task=run_task, + cwd=effective_target, + roster=roster, + read_only=True, + lock_workspace=effective_target, + ) + with runguard.run_lock(effective_target, run_dir=chosen_output_dir): + rc = aboyeur.run( + run_task, + roster, + show_plan=True, + cwd=effective_target, + output_dir=chosen_output_dir, + handoff_inbox=chosen_handoff_inbox, + read_only=True, + sandbox="read-only" if effective_native else None, + ) + except runguard.RunGuardError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 if chosen_output_dir.is_dir(): _write_summary(chosen_output_dir) print(f"artifacts: {chosen_output_dir}", file=sys.stderr) diff --git a/src/brigade/model_trials.py b/src/brigade/model_trials.py index 87b92cb2..204e990c 100644 --- a/src/brigade/model_trials.py +++ b/src/brigade/model_trials.py @@ -688,17 +688,28 @@ def execute( print(f"error: {exc}", file=sys.stderr) return 2 try: - rc = aboyeur.run( - cell.prompt, - roster, - worker=cell.seat, + read_only = cell.execution_mode == "read-only" + aboyeur.record_run_start( + run_dir, + task=cell.prompt, cwd=cell_workspace, - output_dir=run_dir, - route_enabled=False, - read_only=cell.execution_mode == "read-only", - authorized_writable_worktree=cell.execution_mode == "writable-worktree", + roster=roster, + read_only=read_only, + worker=cell.seat, lock_workspace=workspace, ) + with runguard.run_lock(workspace, run_dir=run_dir): + rc = aboyeur.run( + cell.prompt, + roster, + worker=cell.seat, + cwd=cell_workspace, + output_dir=run_dir, + route_enabled=False, + read_only=read_only, + authorized_writable_worktree=cell.execution_mode == "writable-worktree", + lock_workspace=workspace, + ) try: text = (run_dir / "final.txt").read_text() except OSError: diff --git a/src/brigade/runs_cmd.py b/src/brigade/runs_cmd.py index 1e5b4685..9bf55dad 100644 --- a/src/brigade/runs_cmd.py +++ b/src/brigade/runs_cmd.py @@ -890,38 +890,18 @@ def _refresh_resumed_snapshot( aboyeur._write_json(run_dir / "run.json", receipt_schema.stamp_run_receipt(refreshed)) -def _finalize_redeemed_approval( +def _commit_redeemed_approval_facts( run_dir: Path, workspace: Path, reference: Mapping[str, str], + *, + decided_at: str, + claimed_at: str, ) -> None: - """Commit consumed/resumed facts only after the source CAS redeemed.""" + """Commit outcome facts while the caller retains the source-store lock.""" from . import run_lifecycle run_id = run_dir.name - source = reference["source"] - record: Mapping[str, Any] | None - if source == "daily": - from .daily_cmd import approvals as daily_approvals - - record = daily_approvals._find_approval(workspace, reference["approval_id"]) - elif source == "tool": - from .tools_cmd import calls as tool_calls - - record, _calls, _error = tool_calls._resolve_call(workspace, reference["approval_id"]) - else: - raise ApprovalResumeError(f"unsupported approval source: {source}") - if not isinstance(record, Mapping): - raise ApprovalResumeError("approval source record disappeared after resumed action") - _reference_matches_record(reference, record) - claim = record.get("approval_claim") - if ( - not isinstance(claim, Mapping) - or claim.get("run_id") != run_id - or claim.get("state") != "redeemed" - or not isinstance(claim.get("redeemed_at"), str) - ): - raise ApprovalResumeError("approval action did not redeem its reserved claim") run_lifecycle.record_lifecycle_event( run_dir, event_type="approval.consumed", @@ -950,16 +930,66 @@ def _finalize_redeemed_approval( meta = _read_json(run_dir / "run.json") if meta is None: raise ApprovalResumeError("approval run receipt disappeared after resumed action") - decided_at = record.get("reviewed_at") _refresh_resumed_snapshot( run_dir, meta, reference, - decided_at=decided_at if isinstance(decided_at, str) else None, - claimed_at=str(claim["redeemed_at"]), + decided_at=decided_at, + claimed_at=claimed_at, ) +def _finalize_redeemed_approval( + run_dir: Path, + workspace: Path, + reference: Mapping[str, str], +) -> None: + """Validate and commit outcome facts in one source-store transaction.""" + run_id = run_dir.name + source = reference["source"] + if source == "daily": + from .daily_cmd import approvals as daily_approvals + + try: + with daily_approvals.redeemed_reconciliation_guard( + workspace, + reference["approval_id"], + reference, + run_id, + ) as claim: + _commit_redeemed_approval_facts( + run_dir, + workspace, + reference, + decided_at=claim.decided_at, + claimed_at=claim.claimed_at, + ) + except daily_approvals.ApprovalClaimError as exc: + raise ApprovalResumeError(str(exc)) from exc + return + if source == "tool": + from .tools_cmd import calls as tool_calls + + try: + with tool_calls.redeemed_reconciliation_guard( + workspace, + reference["approval_id"], + reference, + run_id, + ) as claim: + _commit_redeemed_approval_facts( + run_dir, + workspace, + reference, + decided_at=claim.decided_at, + claimed_at=claim.claimed_at, + ) + except tool_calls.CallClaimError as exc: + raise ApprovalResumeError(str(exc)) from exc + return + raise ApprovalResumeError(f"unsupported approval source: {source}") + + def resume(run_dir: Path) -> int: """Resume a legacy run or consume an approval before a paused continuation.""" from . import run_lifecycle, run_resume, runguard @@ -1022,7 +1052,8 @@ def resume(run_dir: Path) -> int: ) raise ApprovalResumeError(f"approval is {authorization.status}") if authorization.status == "redeemed": - raise ApprovalResumeError("approval claim is already redeemed; outcome reconciliation is required") + _finalize_redeemed_approval(run_dir, workspace, reference) + return 0 if authorization.status in {"consumed", "reserved"}: if authorization.consumed_by != run_dir.name: owner = authorization.consumed_by or "another run" diff --git a/src/brigade/tools_cmd/calls.py b/src/brigade/tools_cmd/calls.py index d3a46761..365a4927 100644 --- a/src/brigade/tools_cmd/calls.py +++ b/src/brigade/tools_cmd/calls.py @@ -5,6 +5,7 @@ import json import re import sys +import time from collections.abc import Iterator, Mapping from contextlib import contextmanager from dataclasses import dataclass @@ -218,11 +219,22 @@ def _calls_lock_path(target: Path) -> Path: return path.with_name(f"{path.name}.lock") +def _acquire_calls_store_lock(path: Path) -> runguard._LockOwnership: + deadline = time.monotonic() + 5.0 + while True: + try: + return runguard._acquire_lock(path) + except runguard.RunLockError as exc: + if "another brigade run appears active" not in str(exc) or time.monotonic() >= deadline: + raise + time.sleep(0.01) + + @contextmanager def _calls_store_lock(target: Path) -> Iterator[None]: path = _calls_lock_path(target) path.parent.mkdir(parents=True, exist_ok=True) - ownership = runguard._acquire_lock(path) + ownership = _acquire_calls_store_lock(path) try: yield finally: @@ -461,6 +473,72 @@ def redeem_for_run_unlocked(call: dict[str, Any], token: str) -> str: return redeemed_at +def _validate_redeemed_for_run_unlocked( + target: Path, + approval_id: str, + reference: Mapping[str, str], + run_id: str, +) -> CallClaim: + """Validate one redeemed Tool claim while its source-store lock is held.""" + call, _calls, error = _resolve_call(target, approval_id) + if call is None: + raise CallClaimError(error or f"tool approval not found: {approval_id}") + actual = _approval_claim_reference(call) + if any(reference.get(field) != actual[field] for field in _APPROVAL_REFERENCE_FIELDS): + raise CallClaimError("tool approval fingerprints changed before reconciliation") + claim = call.get("approval_claim") + if not isinstance(claim, Mapping) or claim.get("state") != "redeemed": + raise CallClaimError("tool approval has no redeemed claim to reconcile") + claim_run_id = claim.get("run_id") + if claim_run_id != run_id: + owner = claim_run_id if isinstance(claim_run_id, str) and claim_run_id else "another run" + raise CallClaimError(f"tool approval redeemed claim belongs to {owner}") + redeemed_at = claim.get("redeemed_at") + if not isinstance(redeemed_at, str) or not redeemed_at: + raise CallClaimError("tool approval redeemed claim has no redemption timestamp") + if call.get("status") != "completed": + raise CallClaimError(f"tool approval redeemed source state is {call.get('status') or 'unknown'}") + revalidation = dict(call) + revalidation["status"] = "approved" + revalidation["run_id"] = None + revalidation["started_at"] = None + revalidation["completed_at"] = None + blockers = _call_run_blockers(target, revalidation) + if blockers: + raise CallClaimError(f"tool approval is stale or blocked: {blockers[0]}") + decided_at = call.get("reviewed_at") + if not isinstance(decided_at, str) or not decided_at: + raise CallClaimError("tool approval decision has no review timestamp") + return CallClaim( + decided_at=decided_at, + claimed_at=redeemed_at, + already_claimed=True, + ) + + +@contextmanager +def redeemed_reconciliation_guard( + target: Path, + approval_id: str, + reference: Mapping[str, str], + run_id: str, +) -> Iterator[CallClaim]: + """Hold the Tool source-store lock through outcome reconciliation.""" + with _calls_store_lock(target): + yield _validate_redeemed_for_run_unlocked(target, approval_id, reference, run_id) + + +def validate_redeemed_for_run( + target: Path, + approval_id: str, + reference: Mapping[str, str], + run_id: str, +) -> CallClaim: + """Validate one redeemed Tool claim for outcome-only reconciliation.""" + with redeemed_reconciliation_guard(target, approval_id, reference, run_id) as claim: + return claim + + def _make_call_record(plan_payload: dict[str, Any]) -> dict[str, Any]: fingerprint = _call_fingerprint(plan_payload) now = helpers._now().isoformat() @@ -942,12 +1020,17 @@ def _call_review( else: print(f"error: {error}", file=sys.stderr) return 1 - if call.get("status") == "running": - payload = {"target": str(target), "error": "running calls cannot be reviewed", "call": call} + current_status = str(call.get("status") or "") + if current_status in {"running", "completed"}: + payload = { + "target": str(target), + "error": f"{current_status} calls cannot be reviewed", + "call": call, + } if json_output: print(json.dumps(payload, indent=2, sort_keys=True)) else: - print("error: running calls cannot be reviewed", file=sys.stderr) + print(f"error: {current_status} calls cannot be reviewed", file=sys.stderr) return 1 if status == "approved" and call.get("blockers"): payload = {"target": str(target), "error": "blocked calls cannot be approved", "call": call} diff --git a/tests/run_test_helpers.py b/tests/run_test_helpers.py new file mode 100644 index 00000000..8cf6fd76 --- /dev/null +++ b/tests/run_test_helpers.py @@ -0,0 +1,71 @@ +from pathlib import Path +import subprocess +from typing import Any + +from brigade import aboyeur, runguard + + +def _ignore_brigade_runtime(workspace: Path) -> None: + """Keep harness artifacts out of git ground-truth comparisons in test repos.""" + if not runguard.is_git_worktree(workspace): + return + result = subprocess.run( + ["git", "rev-parse", "--git-path", "info/exclude"], + cwd=workspace, + check=True, + capture_output=True, + text=True, + ) + exclude_path = Path(result.stdout.strip()) + if not exclude_path.is_absolute(): + exclude_path = workspace / exclude_path + try: + existing = exclude_path.read_text() + except FileNotFoundError: + existing = "" + pattern = "/.brigade/" + if pattern in existing.splitlines(): + return + exclude_path.parent.mkdir(parents=True, exist_ok=True) + separator = "" if not existing or existing.endswith("\n") else "\n" + exclude_path.write_text(f"{existing}{separator}{pattern}\n") + + +def run_aboyeur_guarded(*args: Any, **kwargs: Any) -> int: + """Call ``aboyeur.run`` under the same run guard used by the CLI.""" + output_dir = kwargs.get("output_dir") + if output_dir is None: + return aboyeur.run(*args, **kwargs) + + run_dir = Path(output_dir).expanduser().resolve() + + explicit_workspace = kwargs.get("lock_workspace") + requested_workspace = explicit_workspace or kwargs.get("cwd") + workspace_path = ( + Path(explicit_workspace or requested_workspace).expanduser().resolve() + if explicit_workspace is not None or requested_workspace is not None + else run_dir.parent / f".{run_dir.name}-lock-workspace" + ) + workspace_path.mkdir(parents=True, exist_ok=True) + kwargs["lock_workspace"] = workspace_path + _ignore_brigade_runtime(workspace_path) + if runguard.is_active_run_owner(workspace_path, run_dir): + return aboyeur.run(*args, **kwargs) + task = args[0] if args else kwargs.get("task") + roster = args[1] if len(args) > 1 else kwargs.get("roster") + raw_cwd = kwargs.get("cwd") + cwd = Path(raw_cwd).expanduser().resolve() if raw_cwd is not None else None + aboyeur.record_run_start( + run_dir, + task=task, + cwd=cwd, + roster=roster, + read_only=bool(kwargs.get("read_only", False)), + worker=kwargs.get("worker"), + dry_run=bool(kwargs.get("dry_run", False)), + lock_workspace=workspace_path, + codex_transport=kwargs.get("codex_transport") or roster.codex_transport, + scheduler=kwargs.get("scheduler", "waves"), + ) + with runguard.run_lock(workspace_path, run_dir=run_dir): + return aboyeur.run(*args, **kwargs) diff --git a/tests/test_aboyeur.py b/tests/test_aboyeur.py index 683cce98..33f3a7d2 100644 --- a/tests/test_aboyeur.py +++ b/tests/test_aboyeur.py @@ -17,6 +17,7 @@ from brigade import proc from brigade import runguard from brigade.roster import Agent, Roster +from tests.run_test_helpers import run_aboyeur_guarded from tests.work_cmd_test_helpers import _init_git_repo @@ -876,7 +877,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), code_graph=brief, route_enabled=False) == 0 + assert run_aboyeur_guarded("build feature", _roster(), code_graph=brief, route_enabled=False) == 0 assert len(calls) == 3 @@ -905,7 +906,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), evidence=evidence, route_enabled=False) == 0 + assert run_aboyeur_guarded("build feature", _roster(), evidence=evidence, route_enabled=False) == 0 assert len(calls) == 3 @@ -940,7 +941,10 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): ) assert ( - aboyeur.run("build feature", _roster(), cwd=tmp_path / "work", output_dir=output_dir, drift_impact=drift) == 0 + run_aboyeur_guarded( + "build feature", _roster(), cwd=tmp_path / "work", output_dir=output_dir, drift_impact=drift + ) + == 0 ) run_meta = json.loads((output_dir / "run.json").read_text()) @@ -977,7 +981,10 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), cwd=tmp_path / "work", output_dir=output_dir, evidence=evidence) == 0 + assert ( + run_aboyeur_guarded("build feature", _roster(), cwd=tmp_path / "work", output_dir=output_dir, evidence=evidence) + == 0 + ) run_meta = json.loads((output_dir / "run.json").read_text()) assert run_meta["evidence_brief"] == { @@ -1016,7 +1023,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) assert ( - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), cwd=tmp_path / "work", @@ -1053,7 +1060,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) assert ( - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), cwd=tmp_path / "work", @@ -1105,7 +1112,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): ) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run("build feature", _roster(), dry_run=True, route_enabled=False) + rc = run_aboyeur_guarded("build feature", _roster(), dry_run=True, route_enabled=False) out = capsys.readouterr().out assert rc == 0 assert "implement it" in out @@ -1132,7 +1139,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): ) output_dir = tmp_path / "run" - assert aboyeur.run("build feature", _roster(), dry_run=True, output_dir=output_dir) == 0 + assert run_aboyeur_guarded("build feature", _roster(), dry_run=True, output_dir=output_dir) == 0 run_meta = json.loads((output_dir / "run.json").read_text()) assert run_meta["code_graph_delta"] == { @@ -1161,7 +1168,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): return agents.AgentResult(text="final answer", ok=True) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run("build feature", _roster(), route_enabled=False) + rc = run_aboyeur_guarded("build feature", _roster(), route_enabled=False) out = capsys.readouterr().out assert rc == 0 assert out.strip() == "final answer" @@ -1178,7 +1185,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "do exactly this", _roster(), worker="coder", @@ -1218,7 +1225,7 @@ def test_lifecycle_run_dispatch_emits_only_identified_worker_dispatch_facts(monk with runguard.run_lock(tmp_path, run_dir=output_dir): assert ( - aboyeur.run( + run_aboyeur_guarded( "do exactly this", _roster(), worker="coder", @@ -1261,7 +1268,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "do exactly this", _roster(), worker="coder", @@ -1301,7 +1308,7 @@ def test_run_direct_grok_progress_only_output_fails_with_honest_artifacts(monkey monkeypatch.setattr(agents.proc, "run", lambda argv, **kwargs: agents.proc.Result(0, output + "\n", "")) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "Review the current documentation diff and report only actionable findings.", _grok_roster(), cwd=tmp_path, @@ -1348,7 +1355,7 @@ def test_run_direct_grok_structured_final_succeeds_with_honest_artifacts(monkeyp monkeypatch.setattr(agents.proc, "run", lambda argv, **kwargs: agents.proc.Result(0, stdout + "\n", "")) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "Review the current documentation diff and report only actionable findings.", _grok_roster(), cwd=tmp_path, @@ -1382,7 +1389,7 @@ def test_run_direct_grok_continuation_recovery_preserves_both_attempts(monkeypat ) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "Review the current diff.", _grok_roster(), cwd=tmp_path, @@ -1428,7 +1435,7 @@ def test_run_direct_grok_fallback_recovery_selects_explicit_acpx_seat(monkeypatc ) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "Review the current diff.", _grok_roster(fallback=True), cwd=tmp_path, @@ -1463,7 +1470,7 @@ def test_run_direct_grok_missing_fallback_is_typed_after_continuation(monkeypatc ) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "Review the current diff.", _grok_roster(), cwd=tmp_path, @@ -1503,7 +1510,7 @@ def test_run_direct_grok_all_attempts_invalid_preserves_terminal_failure(monkeyp ) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "Review the current diff.", _grok_roster(fallback=True), cwd=tmp_path, @@ -1534,7 +1541,7 @@ def test_run_direct_grok_operational_envelope_never_enters_recovery(monkeypatch, calls = _stub_grok_process(monkeypatch, agents.proc.Result(0, envelope + "\n", "")) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "Review the current diff.", _grok_roster(fallback=True), cwd=tmp_path, @@ -1569,7 +1576,7 @@ def test_run_direct_grok_operational_diagnostic_outside_answer_never_enters_reco ) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "Review the current diff.", _grok_roster(fallback=True), cwd=tmp_path, @@ -1602,7 +1609,7 @@ def test_run_defers_success_until_worktree_artifact_collection(monkeypatch, tmp_ monkeypatch.setattr(agents.proc, "run", lambda argv, **kwargs: agents.proc.Result(0, stdout + "\n", "")) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "Implement the requested change.", _grok_roster(), cwd=tmp_path, @@ -1620,14 +1627,16 @@ def test_run_defers_success_until_worktree_artifact_collection(monkeypatch, tmp_ assert "duration_seconds" not in run_payload assert (output_dir / "final.txt").read_text().strip() == answer - aboyeur.record_artifact_collection( - output_dir, - status="ok", - patch_ref="changes.patch", - changed=False, - tracked_count=0, - untracked_count=0, - ) + lock_workspace = Path(run_payload["lock_workspace"]) + with runguard.run_lock(lock_workspace, run_dir=output_dir): + aboyeur.record_artifact_collection( + output_dir, + status="ok", + patch_ref="changes.patch", + changed=False, + tracked_count=0, + untracked_count=0, + ) run_payload = json.loads((output_dir / "run.json").read_text()) assert run_payload["status"] == "ok" @@ -1738,7 +1747,7 @@ def interrupt_after_dispatch(*args, **kwargs): # noqa: ARG001 expected_kind = "signal" if escape == "sigterm" else "keyboard-interrupt" expected_exception = SystemExit if escape == "sigterm" else KeyboardInterrupt with pytest.raises(expected_exception): - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), worker="coder", @@ -1831,7 +1840,7 @@ def fake_run_agent(cli_ref, prompt, **kwargs): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "do exactly this", _roster(), worker="coder", @@ -1865,7 +1874,7 @@ def fake_run_agent(cli_ref, prompt, **kwargs): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) assert ( - aboyeur.run( + run_aboyeur_guarded( "do exactly this", _roster(), worker="coder", @@ -1896,7 +1905,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "fix the bug", _roster(), worker="coder", @@ -1945,7 +1954,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): return agents.AgentResult(text="final answer", ok=True) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster()) == 0 + assert run_aboyeur_guarded("build feature", _roster()) == 0 worker_calls = [call for call in calls if "You are Brigade worker" in call[1]] assert [call[0] for call in worker_calls] == ["ollama:llama3.3", "codex"] @@ -1982,7 +1991,7 @@ def fake_run_agent(cli_ref, prompt, **kwargs): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) with pytest.raises(KeyboardInterrupt): - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), output_dir=output_dir, @@ -2012,7 +2021,7 @@ def interrupted_plan(*args, **kwargs): # noqa: ARG001 monkeypatch.setattr(aboyeur, "plan", interrupted_plan) with pytest.raises(KeyboardInterrupt): - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), output_dir=output_dir, @@ -2050,7 +2059,7 @@ def failed_synthesis(*args, **kwargs): # noqa: ARG001 monkeypatch.setattr(aboyeur, "_run_orchestrator", failed_synthesis) with pytest.raises(RuntimeError, match="synthesis exploded"): - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), output_dir=output_dir, @@ -2074,9 +2083,12 @@ def test_direct_run_terminalizes_sigterm_in_subprocess(tmp_path): script = f""" import time from pathlib import Path -from brigade import aboyeur +from brigade import aboyeur, runguard from brigade.roster import Agent, Roster +run_dir = Path({str(output_dir)!r}) +workspace = run_dir.parent / "workspace" +workspace.mkdir() roster = Roster( orchestrator="chef", agents={{ @@ -2090,13 +2102,16 @@ def blocked_plan(*args, **kwargs): time.sleep(0.05) aboyeur.plan = blocked_plan -aboyeur.run( - "build feature", - roster, - output_dir=Path({str(output_dir)!r}), - code_graph_enabled=False, - route_enabled=False, -) +with runguard.run_lock(workspace, run_dir=run_dir): + aboyeur.run( + "build feature", + roster, + cwd=workspace, + lock_workspace=workspace, + output_dir=run_dir, + code_graph_enabled=False, + route_enabled=False, + ) """ child_env = os.environ.copy() child_env["PYTHONPATH"] = str(Path(__file__).parents[1] / "src") @@ -2148,7 +2163,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): return agents.AgentResult(text="final answer", ok=True) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _timeout_roster(), route_enabled=False) == 0 + assert run_aboyeur_guarded("build feature", _timeout_roster(), route_enabled=False) == 0 assert calls == [("codex", 45.0), ("ollama:llama3.3", 12.0), ("codex", 45.0)] @@ -2167,7 +2182,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False, mo return agents.AgentResult(text="final answer", ok=True) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _model_roster(), route_enabled=False) == 0 + assert run_aboyeur_guarded("build feature", _model_roster(), route_enabled=False) == 0 assert calls == [ ("claude", "claude-fable-5"), ("codex", "gpt-5.5-codex"), @@ -2256,7 +2271,7 @@ def fake_cursor(prompt, **kwargs): ) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "inspect", roster, worker="composer", @@ -2330,7 +2345,7 @@ def fake_cursor(prompt, **kwargs): ) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "inspect", roster, worker="composer", @@ -2406,7 +2421,7 @@ def fake_cursor(prompt, **kwargs): ) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "inspect", roster, worker="composer", @@ -2460,7 +2475,7 @@ def test_roster_payload_includes_read_only_capability(): assert payload["agents"]["coder"]["read_only_capable"] is False -def test_direct_worker_rejects_incapable_read_only_seat_before_artifacts(monkeypatch, tmp_path, capsys): +def test_direct_worker_rejects_incapable_read_only_seat_after_bootstrap_only(monkeypatch, tmp_path, capsys): output_dir = tmp_path / "run" monkeypatch.setattr( aboyeur, @@ -2468,7 +2483,7 @@ def test_direct_worker_rejects_incapable_read_only_seat_before_artifacts(monkeyp lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("validation happened too late")), ) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "inspect feature", _roster_with_incapable_worker(), worker="coder", @@ -2478,7 +2493,12 @@ def test_direct_worker_rejects_incapable_read_only_seat_before_artifacts(monkeyp ) assert rc == 2 - assert not output_dir.exists() + bootstrap = json.loads((output_dir / "run.json").read_text()) + assert bootstrap["status"] == "started" + assert bootstrap["task"] == "inspect feature" + assert bootstrap[_AUTHORITY_REQUEST_FIELD] is True + assert not (output_dir / "plan.json").exists() + assert not (output_dir / "worker-results.json").exists() err = capsys.readouterr().err assert "coder" in err assert "agents.coder.read_only_capable is false" in err @@ -2509,7 +2529,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("inspect feature", _roster(), output_dir=output_dir, read_only=True) == 0 + assert run_aboyeur_guarded("inspect feature", _roster(), output_dir=output_dir, read_only=True) == 0 assert all("READ-ONLY MODE" in prompt for _, prompt, _ in calls) assert all("Do not modify files" in prompt for _, prompt, _ in calls) assert all(read_only for _, _, read_only in calls) @@ -2541,7 +2561,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): ) output_dir = tmp_path / "run" - assert aboyeur.run("inspect feature", _roster(), output_dir=output_dir, read_only=True) == 0 + assert run_aboyeur_guarded("inspect feature", _roster(), output_dir=output_dir, read_only=True) == 0 run_meta = json.loads((output_dir / "run.json").read_text()) assert run_meta["code_graph_delta"]["status"] == "skipped_read_only" @@ -2564,7 +2584,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): return agents.AgentResult(text="final answer", ok=True) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("inspect feature", _roster(), read_only=True, sandbox_read_only=False) == 0 + assert run_aboyeur_guarded("inspect feature", _roster(), read_only=True, sandbox_read_only=False) == 0 assert all("READ-ONLY MODE" in prompt for _, prompt, _ in calls) assert all(read_only is False for _, _, read_only in calls) @@ -2584,7 +2604,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False, sa return agents.AgentResult(text="final answer", ok=True) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("inspect feature", _roster(), read_only=True, sandbox="danger-full-access") == 0 + assert run_aboyeur_guarded("inspect feature", _roster(), read_only=True, sandbox="danger-full-access") == 0 assert all("READ-ONLY MODE" in prompt for _, prompt, _, _ in calls) assert all(sandbox == "danger-full-access" for _, _, _, sandbox in calls) @@ -2601,7 +2621,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): return agents.AgentResult(text="final answer", ok=True) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run("build feature", _roster(), show_plan=True) + rc = run_aboyeur_guarded("build feature", _roster(), show_plan=True) out = capsys.readouterr().out assert rc == 0 assert "plan:" in out @@ -2621,7 +2641,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): return agents.AgentResult(text="final answer", ok=True) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run("build feature", _roster(), verbose=True) + rc = run_aboyeur_guarded("build feature", _roster(), verbose=True) out = capsys.readouterr().out assert rc == 0 assert "workers:" in out @@ -2644,7 +2664,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): return agents.AgentResult(text="final answer", ok=True) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster()) == 3 + assert run_aboyeur_guarded("build feature", _roster()) == 3 assert "not installed" in calls[-1][1] @@ -2671,7 +2691,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 + assert run_aboyeur_guarded("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 assert (output_dir / "plan.json").is_file() assert (output_dir / "plan-attempts.json").is_file() assert (output_dir / "roster.json").is_file() @@ -2934,7 +2954,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) assert ( - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), cwd=run_cwd, @@ -2948,7 +2968,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): assert worker_results["results"][0]["ok"] is True assert worker_results["results"][0]["detail"] == "no-op" assert worker_results["ground_truth"]["changed_files"] == [] - assert worker_results["ground_truth"]["untracked_files"] == [".brigade/scratch.txt"] + assert worker_results["ground_truth"]["untracked_files"] == [] assert worker_results["ground_truth"]["suspected_noop"] is True assert json.loads((output_dir / "run.json").read_text())["suspected_noop"] is True @@ -2975,7 +2995,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("inspect feature", _roster(), cwd=run_cwd, output_dir=output_dir, read_only=True) == 0 + assert run_aboyeur_guarded("inspect feature", _roster(), cwd=run_cwd, output_dir=output_dir, read_only=True) == 0 worker_results = json.loads((output_dir / "worker-results.json").read_text()) assert worker_results["results"][0]["detail"] == "" @@ -3006,7 +3026,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 + assert run_aboyeur_guarded("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 worker_results = json.loads((output_dir / "worker-results.json").read_text()) assert worker_results["results"][0]["detail"] == "" @@ -3058,7 +3078,7 @@ def fake_capture_after(target, run_dir, before): monkeypatch.setattr(aboyeur.graphtrail_delta, "capture_before", fake_capture_before) monkeypatch.setattr(aboyeur.graphtrail_delta, "capture_after_and_diff", fake_capture_after) - assert aboyeur.run("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 + assert run_aboyeur_guarded("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 assert delta_calls == [ ("before", run_cwd, output_dir), @@ -3124,7 +3144,7 @@ def fake_capture_after(target, run_dir, before): monkeypatch.setattr(aboyeur.graphtrail_delta, "capture_before", lambda target, run_dir: before_payload) monkeypatch.setattr(aboyeur.graphtrail_delta, "capture_after_and_diff", fake_capture_after) - assert aboyeur.run("build feature", _roster(), cwd=tmp_path, output_dir=output_dir, code_graph=brief) == 0 + assert run_aboyeur_guarded("build feature", _roster(), cwd=tmp_path, output_dir=output_dir, code_graph=brief) == 0 expected = { "counts": { @@ -3168,7 +3188,7 @@ def fake_capture_after(target, run_dir, before): monkeypatch.setattr(aboyeur.graphtrail_delta, "capture_after_and_diff", fake_capture_after) assert ( - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), cwd=tmp_path, @@ -3210,7 +3230,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): lambda target, run_dir, before: {"status": "sync_failed", "ok": False, "summary": "failed"}, ) - assert aboyeur.run("build feature", _roster(), cwd=tmp_path, output_dir=output_dir, code_graph=brief) == 0 + assert run_aboyeur_guarded("build feature", _roster(), cwd=tmp_path, output_dir=output_dir, code_graph=brief) == 0 ground_truth = json.loads((output_dir / "worker-results.json").read_text())["ground_truth"] assert "context_eval" not in ground_truth @@ -3245,7 +3265,7 @@ def fake_capture_after(target, run_dir, before): monkeypatch.setattr(aboyeur.graphtrail_delta, "capture_before", lambda target, run_dir: before_payload) monkeypatch.setattr(aboyeur.graphtrail_delta, "capture_after_and_diff", fake_capture_after) - assert aboyeur.run("build feature", _roster(), cwd=tmp_path, output_dir=output_dir, code_graph=brief) == 0 + assert run_aboyeur_guarded("build feature", _roster(), cwd=tmp_path, output_dir=output_dir, code_graph=brief) == 0 ground_truth = json.loads((output_dir / "worker-results.json").read_text())["ground_truth"] assert "context_eval" not in ground_truth @@ -3308,7 +3328,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 + assert run_aboyeur_guarded("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 ground_truth = json.loads((output_dir / "worker-results.json").read_text())["ground_truth"] assert [receipt["run_id"] for receipt in ground_truth["verify_receipts"]] == ["99990101-000000-work-verify-abc123"] @@ -3343,7 +3363,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 + assert run_aboyeur_guarded("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 ground_truth = json.loads((output_dir / "worker-results.json").read_text())["ground_truth"] assert ground_truth["available"] is False @@ -3359,7 +3379,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), dry_run=True, output_dir=output_dir) == 0 + assert run_aboyeur_guarded("build feature", _roster(), dry_run=True, output_dir=output_dir) == 0 assert json.loads((output_dir / "plan.json").read_text())["assignments"][0]["worker"] == "coder" assert json.loads((output_dir / "plan-attempts.json").read_text())["attempts"][0]["parsed"] is True run_meta = json.loads((output_dir / "run.json").read_text()) @@ -3383,7 +3403,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), output_dir=output_dir) == 2 + assert run_aboyeur_guarded("build feature", _roster(), output_dir=output_dir) == 2 assert "invalid plan" in capsys.readouterr().err assert len(calls) == 2 run_meta = json.loads((output_dir / "run.json").read_text()) @@ -3511,7 +3531,7 @@ def fake_run_agent(cli_ref, prompt, **kwargs): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) assert ( - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), output_dir=output_dir, @@ -3548,7 +3568,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), output_dir=output_dir) == 2 + assert run_aboyeur_guarded("build feature", _roster(), output_dir=output_dir) == 2 attempt = json.loads((output_dir / "plan-attempts.json").read_text())["attempts"][0] assert attempt["failure_phase"] == "output-validation" assert attempt["failure_kind"] == "non-final-output" @@ -3589,7 +3609,7 @@ def should_not_run(*args, **kwargs): # noqa: ARG001 monkeypatch.delenv("CLOUDFLARE_GATEWAY_ID", raising=False) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "build feature", roster, output_dir=output_dir, @@ -3636,7 +3656,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), output_dir=output_dir) == 2 + assert run_aboyeur_guarded("build feature", _roster(), output_dir=output_dir) == 2 assert "synthesis failed" in capsys.readouterr().err run_meta = json.loads((output_dir / "run.json").read_text()) assert run_meta["status"] == "failed" @@ -3682,7 +3702,7 @@ def observed_write_handoff(*args, **kwargs): monkeypatch.setattr(aboyeur, "write_run_handoff", observed_write_handoff) assert ( - aboyeur.run( + run_aboyeur_guarded( "build feature\n## task heading", _roster(), output_dir=output_dir, @@ -3732,7 +3752,9 @@ def fail_handoff(*args, **kwargs): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) monkeypatch.setattr(aboyeur, "write_run_handoff", fail_handoff) - assert aboyeur.run("build feature", _roster(), output_dir=output_dir, handoff_inbox=tmp_path / "handoffs") == 2 + assert ( + run_aboyeur_guarded("build feature", _roster(), output_dir=output_dir, handoff_inbox=tmp_path / "handoffs") == 2 + ) captured = capsys.readouterr() assert captured.out.strip() == "final answer" assert "handoff failed: cannot write handoff" in captured.err @@ -3785,7 +3807,7 @@ def fail_handoff(*args, **kwargs): monkeypatch.setattr(aboyeur, "write_run_handoff", fail_handoff) with pytest.raises(type(exception)): - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), output_dir=output_dir, @@ -3810,9 +3832,12 @@ def test_handoff_sigterm_terminalizes_nonterminal_receipt(tmp_path): import json import time from pathlib import Path -from brigade import aboyeur, agents +from brigade import aboyeur, agents, runguard from brigade.roster import Agent, Roster +run_dir = Path({str(output_dir)!r}) +workspace = run_dir.parent / "workspace" +workspace.mkdir() roster = Roster( orchestrator="chef", agents={{ @@ -3836,14 +3861,17 @@ def blocked_handoff(*args, **kwargs): time.sleep(0.05) aboyeur.agents.run_agent = fake_run_agent aboyeur.write_run_handoff = blocked_handoff -aboyeur.run( - "build feature", - roster, - output_dir=Path({str(output_dir)!r}), - handoff_inbox=Path({str(tmp_path / "handoffs")!r}), - code_graph_enabled=False, - route_enabled=False, -) +with runguard.run_lock(workspace, run_dir=run_dir): + aboyeur.run( + "build feature", + roster, + cwd=workspace, + lock_workspace=workspace, + output_dir=run_dir, + handoff_inbox=Path({str(tmp_path / "handoffs")!r}), + code_graph_enabled=False, + route_enabled=False, + ) """ child_env = os.environ.copy() child_env["PYTHONPATH"] = str(Path(__file__).parents[1] / "src") @@ -3899,7 +3927,7 @@ def fake_orchestrator(*args, process_registry=None, **kwargs): monkeypatch.setattr(aboyeur, "_run_orchestrator", fake_orchestrator) assert ( - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), output_dir=tmp_path / "run", @@ -3928,7 +3956,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): return agents.AgentResult(text="final answer", ok=True) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _restricted_roster(), route_enabled=False) == 3 + assert run_aboyeur_guarded("build feature", _restricted_roster(), route_enabled=False) == 3 assert [call[0] for call in calls] == ["codex", "codex"] assert "not allowed by limits.allow_models" in calls[-1][1] @@ -4137,7 +4165,7 @@ def start_thread(self, **kwargs): # noqa: ARG002 output_dir = tmp_path / "run" assert ( - aboyeur.run( + run_aboyeur_guarded( "task", _appserver_roster(), worker="cook", @@ -4172,7 +4200,7 @@ def test_run_appserver_worker_rejects_non_final_output_in_receipts(monkeypatch, monkeypatch.setattr(aboyeur.codex_appserver, "AppServer", _IntentOnlyAppServer) output_dir = tmp_path / "run" - rc = aboyeur.run( + rc = run_aboyeur_guarded( "review the repository", _appserver_roster(), worker="cook", @@ -4216,7 +4244,7 @@ def fake_dispatch(*args, process_registry=None, **kwargs): monkeypatch.setattr(aboyeur, "dispatch", fake_dispatch) assert ( - aboyeur.run( + run_aboyeur_guarded( "task", _appserver_roster(), worker="cook", @@ -4264,7 +4292,7 @@ def fake_run_agent(cli_ref, prompt, **kwargs): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) out = tmp_path / "run" - rc = aboyeur.run("task", roster, cwd=tmp_path, output_dir=out) + rc = run_aboyeur_guarded("task", roster, cwd=tmp_path, output_dir=out) assert rc == 0 assert "falling back to exec" in capsys.readouterr().err run_json = json.loads((out / "run.json").read_text()) @@ -4293,7 +4321,7 @@ def fake_dispatch(*args, **kwargs): monkeypatch.setattr(aboyeur, "dispatch", fake_dispatch) assert ( - aboyeur.run( + run_aboyeur_guarded( "task", _appserver_roster(), worker="cook", @@ -4341,7 +4369,7 @@ def fake_dispatch(*args, **kwargs): monkeypatch.setattr(aboyeur, "dispatch", fake_dispatch) assert ( - aboyeur.run( + run_aboyeur_guarded( "task", _appserver_roster(), worker="cook", @@ -4388,7 +4416,7 @@ def close(self): output_dir = tmp_path / "run" with pytest.raises(KeyboardInterrupt): - aboyeur.run( + run_aboyeur_guarded( "task", _appserver_roster(), worker="cook", @@ -4522,7 +4550,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) output_dir = tmp_path / "out" - assert aboyeur.run("rename the config loader helper", _roster(), output_dir=output_dir) == 0 + assert run_aboyeur_guarded("rename the config loader helper", _roster(), output_dir=output_dir) == 0 payload = json.loads((output_dir / "run.json").read_text()) assert payload["route"]["signals"] == ["code"] assert payload["route"]["route"] == ["implement", "correctness-review", "verify"] @@ -4612,7 +4640,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) output_dir = tmp_path / "out" - assert aboyeur.run("tidy the session helper", _roster(), cwd=tmp_path, output_dir=output_dir) == 0 + assert run_aboyeur_guarded("tidy the session helper", _roster(), cwd=tmp_path, output_dir=output_dir) == 0 payload = json.loads((output_dir / "run.json").read_text()) assert "auth-surface" in payload["route"]["signals"] assert "security-review" in payload["route"]["route"] @@ -4628,7 +4656,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) output_dir = tmp_path / "out" assert ( - aboyeur.run( + run_aboyeur_guarded( "add pagination to the list endpoint", _roster(), output_dir=output_dir, @@ -4801,7 +4829,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 + assert run_aboyeur_guarded("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 snapshot_file = output_dir / "pre-run-snapshot.json" assert snapshot_file.is_file() @@ -4857,7 +4885,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run("build feature", _roster(), cwd=run_cwd, output_dir=output_dir, code_graph_enabled=False) + rc = run_aboyeur_guarded("build feature", _roster(), cwd=run_cwd, output_dir=output_dir, code_graph_enabled=False) assert rc == 2 err = capsys.readouterr().err @@ -4921,7 +4949,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "build feature", _roster(), cwd=run_cwd, @@ -4964,7 +4992,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run("build feature", _roster(), cwd=run_cwd, output_dir=output_dir, code_graph_enabled=False) + rc = run_aboyeur_guarded("build feature", _roster(), cwd=run_cwd, output_dir=output_dir, code_graph_enabled=False) assert rc == 2 err = capsys.readouterr().err @@ -4988,15 +5016,20 @@ def boom(_cwd): monkeypatch.setattr(aboyeur.runguard, "capture_pre_run_snapshot", boom) - rc = aboyeur.run("build feature", _roster(), cwd=run_cwd, output_dir=output_dir, code_graph_enabled=False) + rc = run_aboyeur_guarded("build feature", _roster(), cwd=run_cwd, output_dir=output_dir, code_graph_enabled=False) assert rc == 2 err = capsys.readouterr().err assert "pre-run snapshot failed" in err assert "could not read git state" in err - # No run artifacts should have been written: preflight failed before start. - assert not (output_dir / "run.json").exists() + # The CLI-equivalent bootstrap receipt precedes the guarded run preflight, + # but no pre-run snapshot or dispatch artifacts may be written. + bootstrap = json.loads((output_dir / "run.json").read_text()) + assert bootstrap["status"] == "started" + assert bootstrap["task"] == "build feature" + assert bootstrap[_AUTHORITY_REQUEST_FIELD] is True assert not (output_dir / "pre-run-snapshot.json").exists() + assert not (output_dir / "plan.json").exists() def test_run_persists_pre_run_snapshot_before_worker_dispatch(monkeypatch, tmp_path): @@ -5021,7 +5054,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - assert aboyeur.run("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 + assert run_aboyeur_guarded("build feature", _roster(), cwd=run_cwd, output_dir=output_dir) == 0 assert dispatched_with_snapshot["value"] is True @@ -5202,7 +5235,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "build feature", _roster(), cwd=worktree, @@ -5290,7 +5323,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): output_dir = tmp_path / "run" monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "build feature", _roster(), cwd=worktree, @@ -5398,53 +5431,244 @@ def _authority_run_dir(tmp_path: Path) -> tuple[Path, Path]: return workspace, run_dir -def test_new_run_authority_env_implies_both_durable_request_fields(tmp_path, monkeypatch): - """Blocker #1: BRIGADE_RUN_JOURNAL_AUTHORITY=1 alone enrolls a new run with - BOTH durable request fields true, even when BRIGADE_LIFECYCLE_JOURNAL is - unset. The authority opt-in implies lifecycle journaling. - """ - monkeypatch.setenv(_AUTHORITY_ENV, "1") +def test_new_run_defaults_to_journal_authority_without_environment_flags(tmp_path, monkeypatch): + monkeypatch.delenv(_AUTHORITY_ENV, raising=False) monkeypatch.delenv(_LIFECYCLE_ENV, raising=False) workspace, run_dir = _authority_run_dir(tmp_path) with runguard.run_lock(workspace, run_dir=run_dir): aboyeur.record_run_start( run_dir, - task="authority-implied lifecycle", + task="default journal authority", cwd=workspace, roster=_roster(), read_only=False, lock_workspace=workspace, ) - meta = json.loads((run_dir / "run.json").read_text()) - assert meta[_AUTHORITY_REQUEST_FIELD] is True - assert meta[_LIFECYCLE_REQUEST_FIELD] is True + receipt = json.loads((run_dir / "run.json").read_text()) + assert receipt[_AUTHORITY_REQUEST_FIELD] is True + assert receipt[_LIFECYCLE_REQUEST_FIELD] is True -def test_existing_legacy_run_not_enrolled_by_later_authority_env(tmp_path, monkeypatch): - """Blocker #1 guard: an existing legacy run never picks up journal authority - from a later environment change. The durable field is set at run creation - only; a later BRIGADE_RUN_JOURNAL_AUTHORITY=1 must not enroll it. - """ - monkeypatch.delenv(_AUTHORITY_ENV, raising=False) +def test_default_authority_run_json_remains_readable_by_previous_v1_reader(tmp_path): + workspace, run_dir = _authority_run_dir(tmp_path) + + with runguard.run_lock(workspace, run_dir=run_dir): + aboyeur.record_run_start( + run_dir, + task="previous reader compatibility", + cwd=workspace, + roster=_roster(), + read_only=False, + lock_workspace=workspace, + ) + + def previous_v1_reader(path: Path) -> dict[str, object]: + payload = json.loads(path.read_text()) + assert payload["schema"] == "brigade.run.v1" + assert payload["schema_version"] == 1 + assert payload["status"] in {"started", "planning", "dispatching", "running"} + return { + "task": payload["task"], + "status": payload["status"], + "started_at": payload["started_at"], + } + + previous_view = previous_v1_reader(run_dir / "run.json") + assert previous_view["task"] == "previous reader compatibility" + assert previous_view["status"] == "started" + + current_payload = json.loads((run_dir / "run.json").read_text()) + assert current_payload[_AUTHORITY_REQUEST_FIELD] is True + assert current_payload[_LIFECYCLE_REQUEST_FIELD] is True + assert (run_dir / "events" / "lifecycle.jsonl").is_file() + + +def test_default_authority_bootstrap_precedes_lock_and_dispatch(tmp_path, monkeypatch): + workspace = tmp_path / "workspace" + workspace.mkdir() + _init_git_repo(workspace) + (workspace / ".gitignore").write_text(".brigade/\n") + run_dir = workspace / ".brigade" / "runs" / "bootstrap-lock-dispatch" + observed = [] + + def fake_run_agent(cli_ref, prompt, **kwargs): + observed.append( + ( + runguard.run_lock_state(workspace, run_dir), + (run_dir / "events" / "lifecycle.jsonl").is_file(), + ) + ) + return agents.AgentResult(text="finished", ok=True) + + monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) + aboyeur.record_run_start( + run_dir, + task="bootstrap before lock", + cwd=workspace, + roster=_roster(), + read_only=False, + worker="coder", + lock_workspace=workspace, + ) + + pre_lock = json.loads((run_dir / "run.json").read_text()) + assert pre_lock[_AUTHORITY_REQUEST_FIELD] is True + assert pre_lock[_LIFECYCLE_REQUEST_FIELD] is True + assert not (run_dir / "events").exists() + + with runguard.run_lock(workspace, run_dir=run_dir): + assert ( + aboyeur.run( + "bootstrap before lock", + _roster(), + cwd=workspace, + output_dir=run_dir, + worker="coder", + lock_workspace=workspace, + code_graph_enabled=False, + evidence_enabled=False, + route_enabled=False, + ) + == 0 + ) + + assert observed == [("live", True)] + receipt = json.loads((run_dir / "run.json").read_text()) + assert receipt[_AUTHORITY_REQUEST_FIELD] is True + assert receipt[_LIFECYCLE_REQUEST_FIELD] is True + assert receipt["journal_present"] is True + + +def test_guarded_test_helper_bootstraps_before_its_lock_and_dispatches_with_journal( + tmp_path, + monkeypatch, +): + workspace = tmp_path / "workspace" + workspace.mkdir() + _init_git_repo(workspace) + run_dir = workspace / ".brigade" / "runs" / "helper-bootstrap-lock" + start_states = [] + dispatch_states = [] + original_record_run_start = aboyeur.record_run_start + + def tracked_record_run_start(*args, **kwargs): + start_states.append( + ( + runguard.run_lock_state(workspace, run_dir), + (run_dir / "events" / "lifecycle.jsonl").is_file(), + ) + ) + return original_record_run_start(*args, **kwargs) + + def fake_run_agent(cli_ref, prompt, **kwargs): + dispatch_states.append( + ( + runguard.run_lock_state(workspace, run_dir), + (run_dir / "events" / "lifecycle.jsonl").is_file(), + ) + ) + return agents.AgentResult(text="finished", ok=True) + + monkeypatch.setattr(aboyeur, "record_run_start", tracked_record_run_start) + monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) + + assert ( + run_aboyeur_guarded( + "helper bootstrap before lock", + _roster(), + cwd=workspace, + output_dir=run_dir, + worker="coder", + code_graph_enabled=False, + evidence_enabled=False, + route_enabled=False, + ) + == 0 + ) + + assert start_states == [("absent", False), ("live", False)] + assert dispatch_states == [("live", True)] + receipt = json.loads((run_dir / "run.json").read_text()) + assert receipt[_AUTHORITY_REQUEST_FIELD] is True + assert receipt[_LIFECYCLE_REQUEST_FIELD] is True + + +def test_removed_authority_environment_switch_cannot_disable_new_run_enrollment(tmp_path, monkeypatch): + monkeypatch.setenv(_AUTHORITY_ENV, "0") monkeypatch.delenv(_LIFECYCLE_ENV, raising=False) workspace, run_dir = _authority_run_dir(tmp_path) with runguard.run_lock(workspace, run_dir=run_dir): aboyeur.record_run_start( run_dir, - task="legacy run", + task="environment switch removed", cwd=workspace, roster=_roster(), read_only=False, lock_workspace=workspace, ) + + receipt = json.loads((run_dir / "run.json").read_text()) + assert receipt[_AUTHORITY_REQUEST_FIELD] is True + assert receipt[_LIFECYCLE_REQUEST_FIELD] is True + assert not hasattr(aboyeur, "is_run_journal_authority_enabled") + + +def test_unguarded_enrolled_direct_run_fails_closed_before_dispatch(tmp_path): + workspace = tmp_path / "workspace" + workspace.mkdir() + run_dir = tmp_path / "run" + + with pytest.raises( + runguard.RetainRunLockError, + match="failed to record dispatch lifecycle fact: enrolled lifecycle journal is missing", + ): + aboyeur.run( + "unguarded direct run", + _roster(), + cwd=workspace, + output_dir=run_dir, + worker="coder", + code_graph_enabled=False, + evidence_enabled=False, + route_enabled=False, + ) + + receipt = json.loads((run_dir / "run.json").read_text()) + assert receipt[_AUTHORITY_REQUEST_FIELD] is True + assert receipt[_LIFECYCLE_REQUEST_FIELD] is True + assert not (run_dir / "events" / "lifecycle.jsonl").exists() + + +def test_default_journal_authority_implies_lifecycle_without_lifecycle_environment(tmp_path, monkeypatch): + """Default authority enrollment carries both durable request fields.""" + monkeypatch.delenv(_LIFECYCLE_ENV, raising=False) + workspace, run_dir = _authority_run_dir(tmp_path) + + with runguard.run_lock(workspace, run_dir=run_dir): + aboyeur.record_run_start( + run_dir, + task="authority-implied lifecycle", + cwd=workspace, + roster=_roster(), + read_only=False, + lock_workspace=workspace, + ) + meta = json.loads((run_dir / "run.json").read_text()) - assert _AUTHORITY_REQUEST_FIELD not in meta - assert _LIFECYCLE_REQUEST_FIELD not in meta + assert meta[_AUTHORITY_REQUEST_FIELD] is True + assert meta[_LIFECYCLE_REQUEST_FIELD] is True + - # A later env change must not enroll the existing run. +def test_existing_legacy_run_remains_snapshot_only_when_rerecorded(tmp_path, monkeypatch): + """A pre-cutover receipt without enrollment fields stays snapshot-only.""" + workspace, run_dir = _authority_run_dir(tmp_path) + (run_dir / "run.json").write_text(json.dumps(_legacy_status_payload("started"))) + + # The removed authority setting and the still-supported lifecycle setting + # cannot migrate an existing snapshot-only run. monkeypatch.setenv(_AUTHORITY_ENV, "1") monkeypatch.setenv(_LIFECYCLE_ENV, "1") with runguard.run_lock(workspace, run_dir=run_dir): @@ -5637,19 +5861,9 @@ def failing_read_bytes(self): def test_run_preserves_authority_enrollment_through_first_follow_up_receipt_rewrite(monkeypatch, tmp_path): - """Blocker #2: a new run enrolled via BRIGADE_RUN_JOURNAL_AUTHORITY=1 (with - BRIGADE_LIFECYCLE_JOURNAL unset) carries BOTH durable request fields on its - initial run.json. The first follow-up receipt rewrite that aboyeur.run - performs after record_run_start (the nested run._payload rebuild) must - preserve BOTH durable fields, not just lifecycle_journal_requested. - Without that, _write_json classifies the rewrite as legacy and the legacy - fast path overwrites run_journal_authority_requested away, silently - dropping the run off the authority path before any status transition is - recorded. - """ + """The first follow-up receipt rewrite preserves default enrollment.""" import copy - monkeypatch.setenv(_AUTHORITY_ENV, "1") monkeypatch.delenv(_LIFECYCLE_ENV, raising=False) calls = [] @@ -5686,7 +5900,7 @@ def capture_write_json(path, payload): with runguard.run_lock(run_cwd, run_dir=output_dir): assert ( - aboyeur.run( + run_aboyeur_guarded( "build feature", _roster(), cwd=run_cwd, @@ -5780,10 +5994,9 @@ def failing_read_text(self, *args, **kwargs): (run_dir / "run.json").write_text(corrupt_content) original_bytes = corrupt_content - # Environment flags are set to prove they cannot infer unknown durable - # state and rescue a corrupt run.json. + # Default-on enrollment cannot infer unknown durable state and rescue a + # corrupt run.json. monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") with pytest.raises(runguard.RetainRunLockError): with runguard.run_lock(workspace, run_dir=run_dir): @@ -5813,10 +6026,9 @@ def test_record_run_start_retains_lock_on_invalid_utf8_existing_run_json(tmp_pat original_bytes = b'{"schema":"brigade.run.v1","lifecycle_journal_requested":true}\xff' (run_dir / "run.json").write_bytes(original_bytes) - # Neither environment flag may replace the unreadable durable state with - # inferred enrollment during a re-record attempt. + # Configuration cannot replace the unreadable durable state with inferred + # enrollment during a re-record attempt. monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") with pytest.raises( runguard.RetainRunLockError, @@ -5859,7 +6071,6 @@ def failing_loads(*args, **kwargs): monkeypatch.setattr(aboyeur.json, "loads", failing_loads) monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") with pytest.raises( runguard.RetainRunLockError, @@ -5891,10 +6102,8 @@ def test_record_run_start_preserves_valid_legacy_existing_run_enrollment(tmp_pat workspace, run_dir = _corrupt_run_dir(tmp_path) (run_dir / "run.json").write_text(json.dumps(_legacy_status_payload("started"))) - # Environment flags are set, but an existing run never enrolls from - # environment changes alone. + # The remaining lifecycle setting cannot migrate an existing run. monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") with runguard.run_lock(workspace, run_dir=run_dir): aboyeur.record_run_start( diff --git a/tests/test_aboyeur_status.py b/tests/test_aboyeur_status.py index 1bab0865..ed6cdf9b 100644 --- a/tests/test_aboyeur_status.py +++ b/tests/test_aboyeur_status.py @@ -14,6 +14,7 @@ from brigade import aboyeur from brigade import agents from brigade.roster import Agent, Roster +from tests.run_test_helpers import run_aboyeur_guarded def _roster(): @@ -61,7 +62,7 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False, ** ) monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) - return aboyeur.run( + return run_aboyeur_guarded( "build feature", _roster(), cwd=output_dir.parent, diff --git a/tests/test_daily_driver_cmd.py b/tests/test_daily_driver_cmd.py index c63a81a8..89311742 100644 --- a/tests/test_daily_driver_cmd.py +++ b/tests/test_daily_driver_cmd.py @@ -562,7 +562,6 @@ def test_daily_approval_pauses_real_run_and_cli_resume_consumes_store_once( } ], ) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") marker = "d" * 43 monkeypatch.setenv(runs_cmd._APPROVAL_CORRELATION_ENV, marker) @@ -647,6 +646,8 @@ def continue_daily(path, *, approval_resume, approval_token): assert consumed["status"] == "consumed" assert consumed["consumed_run_id"] == run_dir.name assert consumed["approval_claim"]["state"] == "redeemed" + assert consumed["approval_action_receipt"]["state"] == "completed" + assert consumed["approval_action_receipt"]["owner_run_id"] == run_dir.name event_types = [ event.event_type for event in run_journal.read_journal(run_dir / "events" / "lifecycle.jsonl").events ] @@ -750,7 +751,6 @@ def test_daily_rejected_pause_stays_rejected_after_store_is_approved( } ], ) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") marker = "e" * 43 monkeypatch.setenv(runs_cmd._APPROVAL_CORRELATION_ENV, marker) diff --git a/tests/test_dogfood_cmd.py b/tests/test_dogfood_cmd.py index aec7383c..07e78fbc 100644 --- a/tests/test_dogfood_cmd.py +++ b/tests/test_dogfood_cmd.py @@ -1,9 +1,11 @@ import json from brigade import aboyeur +from brigade import agents from brigade import cli from brigade import dogfood_cmd from brigade import localio +from brigade import runguard from brigade import runs_cmd @@ -103,6 +105,38 @@ def fake_run( assert "artifacts:" in capsys.readouterr().err +def test_dogfood_bootstraps_default_authority_before_locked_dispatch(tmp_path, monkeypatch): + run_dir = tmp_path / ".brigade" / "runs" / "dogfood-authority" + observed = [] + + def fake_run_agent(cli_ref, prompt, **kwargs): + observed.append( + ( + runguard.run_lock_state(tmp_path, run_dir), + json.loads((run_dir / "run.json").read_text()), + (run_dir / "events" / "lifecycle.jsonl").is_file(), + ) + ) + if len(observed) == 1: + return agents.AgentResult( + text=json.dumps({"assignments": [{"worker": "reviewer", "task": "inspect it"}]}), + ok=True, + ) + if len(observed) == 2: + return agents.AgentResult(text="review complete", ok=True) + return agents.AgentResult(text="dogfood complete", ok=True) + + monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) + + assert dogfood_cmd.run(None, target=tmp_path, output_dir=run_dir, inspect=False) == 0 + assert observed + for lock_state, receipt, journal_present in observed: + assert lock_state == "live" + assert receipt["run_journal_authority_requested"] is True + assert receipt["lifecycle_journal_requested"] is True + assert journal_present is True + + def test_dogfood_writes_summary_with_next_step(tmp_path, monkeypatch): def fake_run( task, @@ -117,7 +151,7 @@ def fake_run( sandbox_read_only=None, sandbox=None, ): - output_dir.mkdir(parents=True) + output_dir.mkdir(parents=True, exist_ok=True) _write_json( output_dir / "run.json", { diff --git a/tests/test_model_trials.py b/tests/test_model_trials.py index 7ea45726..e4348ca5 100644 --- a/tests/test_model_trials.py +++ b/tests/test_model_trials.py @@ -479,6 +479,7 @@ def fake_run(task, roster, **kwargs): return 0 monkeypatch.setattr(model_trials.runguard, "is_git_worktree", lambda path: True) + monkeypatch.setattr(model_trials.runguard, "git_root", lambda path: workspace.resolve()) monkeypatch.setattr( model_trials, "_trial_worktree_path", diff --git a/tests/test_route_policy.py b/tests/test_route_policy.py index ef33ac18..78b8de09 100644 --- a/tests/test_route_policy.py +++ b/tests/test_route_policy.py @@ -13,6 +13,7 @@ from brigade.roster import Agent, Roster from brigade.route_receipts import route_decision_payload, write_route_decision +from tests.run_test_helpers import run_aboyeur_guarded from tests.test_scorecard import ( _SCORECARD_MANIFEST_ID, _effectiveness_command, @@ -1045,7 +1046,7 @@ def fake_dispatch(assignments, roster, **kwargs): return [aboyeur.WorkerResult(worker="coder", task="implement helper in src", text="done", ok=True)] monkeypatch.setattr(aboyeur, "dispatch", fake_dispatch) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "implement helper in src", _plan_mode_roster("codex"), cwd=scoreable_target, @@ -1109,7 +1110,7 @@ def fake_dispatch(*args, **kwargs): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) monkeypatch.setattr(aboyeur, "dispatch", fake_dispatch) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "implement helper", _plan_mode_roster("codex"), cwd=scoreable_target, @@ -1164,7 +1165,7 @@ def fake_dispatch(*args, **kwargs): monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) monkeypatch.setattr(aboyeur, "dispatch", fake_dispatch) - rc = aboyeur.run( + rc = run_aboyeur_guarded( "implement helper", _plan_mode_roster("codex"), cwd=scoreable_target, diff --git a/tests/test_run_control.py b/tests/test_run_control.py index a7947878..8f2f95c7 100644 --- a/tests/test_run_control.py +++ b/tests/test_run_control.py @@ -12,6 +12,7 @@ from brigade import aboyeur, agents, cli, codex_appserver, run_control from brigade.roster import Agent, Roster +from tests.run_test_helpers import run_aboyeur_guarded FAKE = [sys.executable, str(Path(__file__).parent / "fake_appserver.py")] @@ -225,7 +226,7 @@ def fake_run_agent(cli_ref, prompt, **kwargs): result: dict[str, int] = {} thread = threading.Thread( - target=lambda: result.update(rc=aboyeur.run("do it", roster, cwd=tmp_path, output_dir=run_dir)), + target=lambda: result.update(rc=run_aboyeur_guarded("do it", roster, cwd=tmp_path, output_dir=run_dir)), daemon=True, ) thread.start() @@ -289,7 +290,7 @@ def fake_run_agent(cli_ref, prompt, **kwargs): result: dict[str, int] = {} thread = threading.Thread( - target=lambda: result.update(rc=aboyeur.run("do it", roster, cwd=tmp_path, output_dir=run_dir)), + target=lambda: result.update(rc=run_aboyeur_guarded("do it", roster, cwd=tmp_path, output_dir=run_dir)), daemon=True, ) thread.start() @@ -360,7 +361,7 @@ def register_and_signal(registry, worker, turn, turn_id): thread = threading.Thread( target=lambda: result.update( - rc=aboyeur.run( + rc=run_aboyeur_guarded( "HANG until interrupted", roster, worker="cook", diff --git a/tests/test_run_lifecycle.py b/tests/test_run_lifecycle.py index d63c3ebc..72cf2e1d 100644 --- a/tests/test_run_lifecycle.py +++ b/tests/test_run_lifecycle.py @@ -1052,7 +1052,6 @@ def _write_run_json_authority(run_dir: Path, status: str, **kwargs) -> None: def test_authority_enrollment_persists_run_journal_authority_requested(tmp_path, monkeypatch): repo = _repo(tmp_path) run_dir = _run_dir(repo) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") aboyeur.record_run_start( run_dir, @@ -1071,7 +1070,6 @@ def test_authority_enrollment_persists_run_journal_authority_requested(tmp_path, def test_not_yet_authoritative_run_writes_legacy_body_when_gate_not_ready(tmp_path, monkeypatch): repo = _repo(tmp_path) run_dir = _run_dir(repo) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") aboyeur.record_run_start( run_dir, @@ -1101,7 +1099,6 @@ def test_not_yet_authoritative_run_writes_legacy_body_when_gate_not_ready(tmp_pa def test_first_write_match_authorizes_first_projected_snapshot(tmp_path, monkeypatch): repo = _repo(tmp_path) run_dir = _run_dir(repo) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") aboyeur.record_run_start( run_dir, @@ -1123,7 +1120,6 @@ def test_first_write_match_authorizes_first_projected_snapshot(tmp_path, monkeyp def test_authoritative_run_fail_closed_on_projection_error(tmp_path, monkeypatch): repo = _repo(tmp_path) run_dir = _run_dir(repo) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") aboyeur.record_run_start( run_dir, @@ -1150,19 +1146,12 @@ def test_authoritative_run_fail_closed_on_projection_error(tmp_path, monkeypatch assert (run_dir / "run.json").read_bytes() == meta_before -def test_legacy_run_unchanged_under_authority_flag_off(tmp_path, monkeypatch): +def test_existing_lifecycle_only_run_remains_non_authoritative(tmp_path, monkeypatch): repo = _repo(tmp_path) run_dir = _run_dir(repo) - monkeypatch.delenv("BRIGADE_RUN_JOURNAL_AUTHORITY", raising=False) - monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") - aboyeur.record_run_start( - run_dir, - task="lifecycle only", - cwd=repo, - roster=_minimal_roster(), - read_only=False, - lock_workspace=repo, - ) + existing = _run_payload("started", lock_workspace=repo) + existing[_REQUEST_FIELD] = True + localio.write_json(run_dir / "run.json", existing) with runguard.run_lock(repo, run_dir=run_dir): _write_run_json(run_dir, "planning") meta = json.loads((run_dir / "run.json").read_text()) @@ -1171,7 +1160,6 @@ def test_legacy_run_unchanged_under_authority_flag_off(tmp_path, monkeypatch): def _enroll_and_authorize(repo, run_dir, monkeypatch): - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") aboyeur.record_run_start( run_dir, @@ -1517,7 +1505,6 @@ def test_authority_fail_closed_on_saved_digest_not_matching_verified_prefix(tmp_ def test_authoritative_write_order_checkpoint_lifecycle_parity_readiness_replace(tmp_path, monkeypatch): repo = _repo(tmp_path) run_dir = _run_dir(repo) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") aboyeur.record_run_start( run_dir, @@ -1587,7 +1574,6 @@ def atomic_spy(path, payload, **kw): def test_first_authority_write_checkpoint_seq1_then_status_seq2(tmp_path, monkeypatch): repo = _repo(tmp_path) run_dir = _run_dir(repo) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") aboyeur.record_run_start( run_dir, @@ -1715,7 +1701,6 @@ def test_first_authority_started_write_projects_on_first_write(tmp_path, monkeyp # checkpoint seq1 + run.created seq2. repo = _repo(tmp_path) run_dir = _run_dir(repo) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") aboyeur.record_run_start( run_dir, @@ -1743,7 +1728,6 @@ def test_authority_requested_post_parity_not_ready_falls_back_to_legacy(tmp_path # falls back to legacy (never fail-closed). repo = _repo(tmp_path) run_dir = _run_dir(repo) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") aboyeur.record_run_start( run_dir, @@ -1771,16 +1755,7 @@ def test_legacy_run_does_not_call_prior_authority_gate(tmp_path, monkeypatch): # bytes/order are preserved. repo = _repo(tmp_path) run_dir = _run_dir(repo) - monkeypatch.delenv("BRIGADE_RUN_JOURNAL_AUTHORITY", raising=False) - monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") - aboyeur.record_run_start( - run_dir, - task="legacy run", - cwd=repo, - roster=_minimal_roster(), - read_only=False, - lock_workspace=repo, - ) + localio.write_json(run_dir / "run.json", _run_payload("started", lock_workspace=repo)) def raise_if_called(run_dir): raise AssertionError("check_projection_readiness must not be called for legacy runs") diff --git a/tests/test_run_resume.py b/tests/test_run_resume.py index e4b904a6..eaeb39a4 100644 --- a/tests/test_run_resume.py +++ b/tests/test_run_resume.py @@ -629,7 +629,6 @@ def test_authority_resume_failed_synthesis_receipt_write_retains_lock(tmp_path, run_meta["cwd"] = str(tmp_path) run_meta["lock_workspace"] = str(tmp_path) (run_dir / "run.json").write_text(json.dumps(run_meta)) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") monkeypatch.setattr(run_resume.codex_appserver, "AppServer", _StubServer) monkeypatch.setattr( @@ -674,7 +673,6 @@ def test_authority_resume_successful_synthesis_receipt_write_retains_lock(tmp_pa run_meta["cwd"] = str(tmp_path) run_meta["lock_workspace"] = str(tmp_path) (run_dir / "run.json").write_text(json.dumps(run_meta)) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") monkeypatch.setattr(run_resume.codex_appserver, "AppServer", _StubServer) monkeypatch.setattr( diff --git a/tests/test_runs_cmd.py b/tests/test_runs_cmd.py index f819f784..fa60fbf8 100644 --- a/tests/test_runs_cmd.py +++ b/tests/test_runs_cmd.py @@ -128,6 +128,62 @@ def _patch_daily_store(monkeypatch, approval, *, blockers=None): "_approval_blockers", lambda target, record, config: list(blockers or []), ) + monkeypatch.setattr( + daily_cmd.approvals, + "_redeemed_reconciliation_blockers", + lambda record, config: list(blockers or []), + ) + + +def _patch_tool_store(monkeypatch, call, *, blockers=None): + monkeypatch.setattr( + tools_cmd.calls, + "_resolve_call", + lambda target, call_id: (call, [call], None), + ) + monkeypatch.setattr( + tools_cmd.calls, + "_call_run_blockers", + lambda target, record: list(blockers or []), + ) + + +def _mark_redeemed(record, run_id): + redeemed_at = "2026-07-30T18:01:00+00:00" + record["approval_claim"] = { + "run_id": run_id, + "state": "redeemed", + "reserved_at": "2026-07-30T18:00:30+00:00", + "token_fingerprint": "a" * 64, + "redeemed_at": redeemed_at, + } + return redeemed_at + + +def _mark_daily_action_completed(record, target, owner_run_id): + daily_run_id = "daily-action-run-1" + completed_at = "2026-07-30T18:02:00+00:00" + record["approval_action_receipt"] = { + "state": "completed", + "owner_run_id": owner_run_id, + "daily_run_id": daily_run_id, + "action_id": record["selected_action_id"], + "source_fingerprint": record["source_fingerprint"], + "completed_at": completed_at, + } + receipt_path = target / ".brigade" / "daily" / "runs" / daily_run_id / "run.json" + receipt_path.parent.mkdir(parents=True) + _write_json( + receipt_path, + { + "status": "completed", + "run_id": daily_run_id, + "approval_id": record["approval_id"], + "selected_action_id": record["selected_action_id"], + "selected_action": {"source_fingerprint": record["source_fingerprint"]}, + "completed_at": completed_at, + }, + ) def test_approval_pause_marker_binds_exact_event_and_redacts_artifact(tmp_path, monkeypatch): @@ -586,6 +642,234 @@ def blockers(target, record): assert "contract fingerprint is stale" in capsys.readouterr().err +@pytest.mark.parametrize( + ("source", "status"), + [ + ("daily", "rejected"), + ("daily", "held"), + ("tool", "rejected"), + ("tool", "held"), + ], +) +def test_redeemed_retry_refuses_changed_source_decision( + tmp_path, + monkeypatch, + capsys, + source, + status, +): + record = _daily_approval(status=status) if source == "daily" else _tool_call() + if source == "tool": + record["status"] = status + run_dir, _ = _approval_run(tmp_path, source=source, record=record) + _mark_redeemed(record, run_dir.name) + events = [] + + if source == "daily": + _patch_daily_store(monkeypatch, record) + else: + _patch_tool_store(monkeypatch, record) + monkeypatch.setattr( + run_lifecycle, + "record_lifecycle_event", + lambda *args, event_type, **kwargs: events.append(event_type), + ) + + assert runs_cmd.resume(run_dir) == 2 + assert events == [] + assert status in capsys.readouterr().err + + +@pytest.mark.parametrize("source", ["daily", "tool"]) +def test_redeemed_retry_revalidates_current_blockers(tmp_path, monkeypatch, capsys, source): + if source == "daily": + record = _daily_approval(status="consumed", consumed_run_id="run-approval-1") + else: + record = _tool_call() + record["status"] = "completed" + run_dir, _ = _approval_run(tmp_path, source=source, record=record) + redeemed_at = _mark_redeemed(record, run_dir.name) + if source == "daily": + record["consumed_at"] = redeemed_at + _mark_daily_action_completed(record, tmp_path, run_dir.name) + _patch_daily_store(monkeypatch, record, blockers=["daily config changed since approval"]) + else: + _patch_tool_store(monkeypatch, record, blockers=["contract fingerprint is stale"]) + events = [] + monkeypatch.setattr( + run_lifecycle, + "record_lifecycle_event", + lambda *args, event_type, **kwargs: events.append(event_type), + ) + + assert runs_cmd.resume(run_dir) == 2 + assert events == [] + assert "stale or blocked" in capsys.readouterr().err + + +def test_redeemed_daily_retry_requires_completed_action_receipt(tmp_path, monkeypatch, capsys): + record = _daily_approval(status="consumed", consumed_run_id="run-approval-1") + run_dir, _ = _approval_run(tmp_path, source="daily", record=record) + redeemed_at = _mark_redeemed(record, run_dir.name) + record["consumed_at"] = redeemed_at + _patch_daily_store(monkeypatch, record) + events = [] + monkeypatch.setattr( + run_lifecycle, + "record_lifecycle_event", + lambda *args, event_type, **kwargs: events.append(event_type), + ) + + assert runs_cmd.resume(run_dir) == 2 + assert events == [] + assert "completed action receipt" in capsys.readouterr().err + + +def test_redeemed_daily_retry_refuses_tampered_action_receipt(tmp_path, monkeypatch, capsys): + record = _daily_approval(status="consumed", consumed_run_id="run-approval-1") + run_dir, _ = _approval_run(tmp_path, source="daily", record=record) + redeemed_at = _mark_redeemed(record, run_dir.name) + record["consumed_at"] = redeemed_at + _mark_daily_action_completed(record, tmp_path, run_dir.name) + daily_run_id = record["approval_action_receipt"]["daily_run_id"] + action_receipt_path = tmp_path / ".brigade" / "daily" / "runs" / daily_run_id / "run.json" + action_receipt = json.loads(action_receipt_path.read_text()) + action_receipt["status"] = "failed" + _write_json(action_receipt_path, action_receipt) + _patch_daily_store(monkeypatch, record) + events = [] + monkeypatch.setattr( + run_lifecycle, + "record_lifecycle_event", + lambda *args, event_type, **kwargs: events.append(event_type), + ) + + assert runs_cmd.resume(run_dir) == 2 + assert events == [] + assert "no longer matches its run" in capsys.readouterr().err + + +@pytest.mark.parametrize("source", ["daily", "tool"]) +def test_redeemed_reconciliation_holds_source_lock_through_journal_commit( + tmp_path, + monkeypatch, + capsys, + source, +): + if source == "daily": + record = _daily_approval(status="consumed", consumed_run_id="run-approval-1") + else: + record = _tool_call() + record["status"] = "completed" + run_dir, _ = _approval_run(tmp_path, source=source, record=record) + redeemed_at = _mark_redeemed(record, run_dir.name) + if source == "daily": + record["consumed_at"] = redeemed_at + _mark_daily_action_completed(record, tmp_path, run_dir.name) + daily_cmd.approvals._write_approval(tmp_path, record) + monkeypatch.setattr( + daily_cmd.approvals, + "_redeemed_reconciliation_blockers", + lambda approval, config: [], + ) + else: + tools_cmd.calls._write_calls(tmp_path, [record]) + monkeypatch.setattr(tools_cmd.calls, "_call_run_blockers", lambda target, call: []) + + append_entered = threading.Event() + review_started = threading.Event() + review_finished = threading.Event() + review_result = [] + events = [] + + def review_source(): + assert append_entered.wait(2) + review_started.set() + if source == "daily": + rc = daily_cmd.approvals_hold( + target=tmp_path, + approval_id=record["approval_id"], + reason="concurrent hold", + ) + else: + rc = tools_cmd.call_reject( + target=tmp_path, + call_id=record["id"], + reason="concurrent rejection", + ) + review_result.append(rc) + review_finished.set() + + def append_fact(*args, event_type, **kwargs): + if not events: + append_entered.set() + assert review_started.wait(2) + assert not review_finished.wait(0.2) + events.append(event_type) + + monkeypatch.setattr(run_lifecycle, "record_lifecycle_event", append_fact) + reviewer = threading.Thread(target=review_source) + reviewer.start() + assert runs_cmd.resume(run_dir) == 0 + reviewer.join(2) + assert not reviewer.is_alive() + assert review_result == [1] + assert events == ["approval.consumed", "run.resumed"] + if source == "daily": + stored = daily_cmd.approvals._find_approval(tmp_path, record["approval_id"]) + assert stored is not None + assert stored["status"] == "consumed" + else: + stored, _calls, error = tools_cmd.calls._resolve_call(tmp_path, record["id"]) + assert error is None + assert stored is not None + assert stored["status"] == "completed" + capsys.readouterr() + + +@pytest.mark.parametrize("source", ["daily", "tool"]) +def test_redeemed_retry_refuses_claim_for_another_run(tmp_path, monkeypatch, capsys, source): + if source == "daily": + record = _daily_approval(status="consumed", consumed_run_id="other-run") + else: + record = _tool_call() + record["status"] = "completed" + run_dir, _ = _approval_run(tmp_path, source=source, record=record) + redeemed_at = _mark_redeemed(record, "other-run") + if source == "daily": + record["consumed_at"] = redeemed_at + _patch_daily_store(monkeypatch, record) + else: + _patch_tool_store(monkeypatch, record) + events = [] + monkeypatch.setattr( + run_lifecycle, + "record_lifecycle_event", + lambda *args, event_type, **kwargs: events.append(event_type), + ) + + assert runs_cmd.resume(run_dir) == 2 + assert events == [] + assert "belongs to other-run" in capsys.readouterr().err + + +def test_redeemed_retry_refuses_legacy_tool_consumption_without_claim(tmp_path, monkeypatch, capsys): + call = _tool_call() + call.update({"status": "running", "run_id": "run-approval-1"}) + run_dir, _ = _approval_run(tmp_path, source="tool", record=call) + _patch_tool_store(monkeypatch, call) + events = [] + monkeypatch.setattr( + run_lifecycle, + "record_lifecycle_event", + lambda *args, event_type, **kwargs: events.append(event_type), + ) + + assert runs_cmd.resume(run_dir) == 2 + assert events == [] + assert "tool approval is running" in capsys.readouterr().err + + def test_resume_nonapproval_run_uses_legacy_fallback(tmp_path, monkeypatch): run_dir = tmp_path / "run" run_dir.mkdir() diff --git a/tests/test_work_cmd_services.py b/tests/test_work_cmd_services.py index 8883aa1f..c2f1cb07 100644 --- a/tests/test_work_cmd_services.py +++ b/tests/test_work_cmd_services.py @@ -20,6 +20,7 @@ from brigade import runguard from brigade import run_journal from brigade import run_resume +from brigade import run_shadow from brigade import runs_cmd from brigade import security_cmd from brigade import tools_cmd @@ -1494,7 +1495,6 @@ def dispatch_with_approval_request(*args, **kwargs): ] monkeypatch.setattr(aboyeur, "dispatch", dispatch_with_approval_request) - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") assert ( @@ -1653,7 +1653,6 @@ def test_tools_resume_claim_rechecks_store_after_authorization( _write_script_tool_config(tmp_path, script='print("ok")\n') assert tools_cmd.call_queue(target=tmp_path, tool_id="runner", args='{"path":"pending"}', json_output=True) == 0 pending = json.loads(capsys.readouterr().out)["call"] - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") marker = "q" * 43 monkeypatch.setenv(runs_cmd._APPROVAL_CORRELATION_ENV, marker) @@ -1747,7 +1746,6 @@ def test_tools_reserved_claim_rechecks_contract_before_redemption_and_action( == 0 ) pending = json.loads(capsys.readouterr().out)["call"] - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") marker = "s" * 43 monkeypatch.setenv(runs_cmd._APPROVAL_CORRELATION_ENV, marker) @@ -1821,7 +1819,7 @@ def stale_before_redeem(path, *, approval_resume, approval_token): assert "run.resumed" not in event_types -def test_tools_redeemed_action_crash_fails_closed_without_second_execution( +def test_tools_redeemed_action_crash_reconciles_without_second_execution( tmp_path, monkeypatch, capsys, @@ -1847,7 +1845,6 @@ def test_tools_redeemed_action_crash_fails_closed_without_second_execution( == 0 ) pending = json.loads(capsys.readouterr().out)["call"] - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") marker = "r" * 43 monkeypatch.setenv(runs_cmd._APPROVAL_CORRELATION_ENV, marker) @@ -1915,13 +1912,36 @@ def redeem_then_crash(path, *, approval_resume, approval_token): assert cli.main(["runs", "resume", str(run_dir)]) == 2 assert (tmp_path / "redeemed-count").read_text() == "1" - assert cli.main(["runs", "resume", str(run_dir)]) == 2 + assert "injected outcome persistence crash" in capsys.readouterr().err + journal_path = run_dir / "events" / "lifecycle.jsonl" + before_reconciliation = run_journal.read_journal(journal_path).events + assert "approval.consumed" not in [event.event_type for event in before_reconciliation] + assert "run.resumed" not in [event.event_type for event in before_reconciliation] + + assert cli.main(["runs", "resume", str(run_dir)]) == 0 assert (tmp_path / "redeemed-count").read_text() == "1" assert provider_calls == [run_dir] stored = next(item for item in tools_cmd._read_calls(tmp_path) if item["id"] == pending["id"]) assert stored["approval_claim"]["state"] == "redeemed" + assert stored["approval_claim"]["run_id"] == run_dir.name assert raw_tokens[0] not in json.dumps(stored) - assert "outcome reconciliation is required" in capsys.readouterr().err + events = run_journal.read_journal(journal_path).events + event_types = [event.event_type for event in events] + for fact in ("approval.consumed", "run.resumed"): + assert event_types.count(fact) == 1 + assert event_types[event_types.index(fact) - 1] == "run.snapshot.checkpointed" + snapshot = json.loads((run_dir / "run.json").read_text()) + assert snapshot["approval_reference"]["decision_state"] == "consumed" + assert snapshot["approval_reference"]["consuming_run_id"] == run_dir.name + assert snapshot["journal_last_sequence"] == events[-1].sequence + assert snapshot["journal_last_event_digest"] == events[-1].event_digest + assert run_shadow.check_projection_readiness(run_dir).ready is True + + reconciled_journal = journal_path.read_bytes() + assert cli.main(["runs", "resume", str(run_dir)]) == 0 + assert journal_path.read_bytes() == reconciled_journal + assert (tmp_path / "redeemed-count").read_text() == "1" + assert provider_calls == [run_dir] def test_tools_approval_pauses_real_run_and_cli_resume_consumes_store_once( @@ -1950,7 +1970,6 @@ def test_tools_approval_pauses_real_run_and_cli_resume_consumes_store_once( == 0 ) pending = json.loads(capsys.readouterr().out)["call"] - monkeypatch.setenv("BRIGADE_RUN_JOURNAL_AUTHORITY", "1") monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") marker = "t" * 43 monkeypatch.setenv(runs_cmd._APPROVAL_CORRELATION_ENV, marker)