feat(runs): graduate journal authority

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.

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

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).
 
 ---
 

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 <run>` or `brigade runs watch <run>` for inspection. Use
+`brigade runs recover <run>` 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 <run>` 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.
 

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

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":