diff --git a/src/brigade/aboyeur.py b/src/brigade/aboyeur.py index b652efe7..a7cf5142 100644 --- a/src/brigade/aboyeur.py +++ b/src/brigade/aboyeur.py @@ -629,17 +629,15 @@ def _payload_requests_authority(payload: dict[str, object]) -> bool: def _genuine_journal_ahead(run_dir: Path) -> bool: - """Return True only when the journal tail has genuinely advanced past a - real recorded comparison (not a forged cursor). + """Return True only for one verified checkpoint/event pair ahead. ``check_projection_readiness`` reports ``REASON_JOURNAL_AHEAD`` whenever the artifact's ``last_compared_sequence``/``last_compared_event_digest`` does not match the journal tail, which conflates a real committed - transition whose shadow step has not run yet with a forged or stale - cursor. The journal-ahead exception may only authorize projection when the - artifact's cursor verifies against an actual event in a clean journal - prefix AND the tail has advanced past it; otherwise the evidence is - forged and the gate must fail closed. + checkpoint/event pair whose shadow step has not run yet with a forged, + stale, or multi-pair cursor. Catch-up is safe only when the artifact cursor + verifies against an actual event in a clean journal and the tail is exactly + one structurally covered pair ahead. """ artifact_path = run_shadow.shadow_artifact_path(run_dir) try: @@ -664,26 +662,53 @@ def _genuine_journal_ahead(run_dir: Path) -> bool: return False if not journal_report.events: return False - return journal_report.events[-1].sequence > baseline[0] + tail_seq = journal_report.events[-1].sequence + return run_shadow._checkpoint_status_write_gap(run_dir, prior_seq=baseline[0], tail_seq=tail_seq) -def _authoritative_prior_decision(run_dir: Path) -> str: +def _authoritative_prior_decision(run_dir: Path, prior_snapshot: dict[str, object]) -> str: """Resolve the prior authority gate for an authoritative run. - Returns ``"ready"`` when the prior committed state is ready, or - ``"journal-ahead"`` for the approved prior journal-ahead exception (a - genuine committed transition whose shadow step has not run yet). Any - other prior state raises a bounded ``LifecycleJournalError`` BEFORE the - checkpoint/lifecycle append, so forged evidence cannot authorize - projection and no unrelated prior reason piggybacks on the journal-ahead - exception. The decision is based on ``run_shadow.check_projection_readiness`` - on the prior committed state, not this write's pre-parity ad hoc parsing. + A ready prior returns immediately. A prior that is not ready may catch up + exactly one verified checkpoint/event pair by recording parity against the + persisted pre-write snapshot and then requiring readiness to become fully + green. Forged cursors, multi-pair gaps, mismatches, errors, and a catch-up + that does not restore readiness all raise before a new pair is appended. """ report = run_shadow.check_projection_readiness(run_dir) if report.ready: return "ready" if report.reasons == (run_shadow.REASON_JOURNAL_AHEAD,) and _genuine_journal_ahead(run_dir): - return "journal-ahead" + run_shadow.record_shadow_comparison(run_dir, prior_snapshot) + caught_up = run_shadow.check_projection_readiness(run_dir) + if caught_up.ready: + artifact_path = run_shadow.shadow_artifact_path(run_dir) + try: + artifact = json.loads(artifact_path.read_text()) + except (OSError, ValueError) as exc: + raise run_lifecycle.LifecycleJournalError( + run_events._bound("authoritative run prior catch-up evidence is unreadable") + ) from exc + mismatches = artifact.get("mismatches") if isinstance(artifact, dict) else None + errors = artifact.get("errors") if isinstance(artifact, dict) else None + if ( + not isinstance(artifact, dict) + or artifact.get("last_outcome") != run_shadow.OUTCOME_MATCH + or isinstance(mismatches, bool) + or not isinstance(mismatches, int) + or mismatches != 0 + or isinstance(errors, bool) + or not isinstance(errors, int) + or errors != 0 + or artifact.get("last_error_category") is not None + ): + raise run_lifecycle.LifecycleJournalError( + run_events._bound("authoritative run prior catch-up evidence is not a clean match") + ) + return "ready" + raise run_lifecycle.LifecycleJournalError( + run_events._bound("authoritative run prior catch-up not ready: " + ",".join(sorted(caught_up.reasons))) + ) raise run_lifecycle.LifecycleJournalError( run_events._bound("authoritative run prior gate not ready: " + ",".join(sorted(report.reasons))) ) @@ -705,117 +730,7 @@ def _project_authority_candidate(path: Path, run_dir: Path, candidate: dict[str, localio.write_text_atomic(path, projection.to_bytes().decode("utf-8")) -def _validate_journal_ahead_projection(run_dir: Path, readiness: run_shadow.ReadinessReport) -> None: - """Validate the post-parity shadow artifact for the approved prior - journal-ahead exception (finding 3). - - The exception may project only when the post-parity gate failure is - exactly the one comparison-gap error created by catching up a genuinely - ahead journal AND the current parity outcome is match. Any other - post-parity mismatch or error must fail closed with a bounded - ``LifecycleJournalError`` and no run.json replace, so the exception cannot - ignore an arbitrary post-parity defect. - - The post-parity ``ReadinessReport`` is checked explicitly: its reasons must - be exactly ``(REASON_ERROR_RECORDED,)``. A forged extra reason such as - ``REASON_JOURNAL_UNREADABLE`` (set by forging the aggregate - ``last_error_category`` to ``"journal-unreadable"``) must fail closed even - when the recent_records pair still looks like comparison-gap then match. - - The current shadow artifact is validated after parity: current - schema/version/run_id, current projector version, ``mismatches`` a - non-bool int equal to 0, ``errors`` a non-bool int equal to 1 (the - comparison-gap), ``last_error_category`` None on the final match - aggregate, last_outcome match, and the final two recent records are - comparison-gap error then match for the same current tail - sequence/digest. Strict integer semantics reject bool counters: ``False`` - must not satisfy ``mismatches == 0`` and ``True`` must not satisfy - ``errors == 1``. - """ - if readiness.reasons != (run_shadow.REASON_ERROR_RECORDED,): - raise run_lifecycle.LifecycleJournalError( - run_events._bound( - "journal-ahead projection post-parity reasons are not exactly error-recorded: " - + ",".join(sorted(readiness.reasons)) - ) - ) - artifact_path = run_shadow.shadow_artifact_path(run_dir) - try: - data = json.loads(artifact_path.read_text()) - except (OSError, ValueError): - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection artifact unreadable") - ) from None - if not isinstance(data, dict): - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection artifact is not an object") - ) - if ( - data.get("schema") != run_shadow.SHADOW_SCHEMA - or data.get("schema_version") != run_shadow.SHADOW_SCHEMA_VERSION - or data.get("run_id") != run_dir.name - ): - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection artifact schema/run_id mismatch") - ) - if data.get("projector_version") != run_projector.PROJECTOR_VERSION: - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection artifact projector version is not current") - ) - mismatches = data.get("mismatches") - if isinstance(mismatches, bool) or not isinstance(mismatches, int) or mismatches != 0: - raise run_lifecycle.LifecycleJournalError(run_events._bound("journal-ahead projection recorded a mismatch")) - errors = data.get("errors") - if isinstance(errors, bool) or not isinstance(errors, int) or errors != 1: - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection must record exactly one comparison-gap error") - ) - if data.get("last_error_category") is not None: - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection aggregate last_error_category is not none") - ) - if data.get("last_outcome") != run_shadow.OUTCOME_MATCH: - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection current parity is not a match") - ) - records = data.get("recent_records") - if not isinstance(records, list) or len(records) < 2: - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection artifact lacks the gap-then-match record pair") - ) - gap_record = records[-2] - match_record = records[-1] - if ( - not isinstance(gap_record, dict) - or gap_record.get("outcome") != run_shadow.OUTCOME_ERROR - or gap_record.get("category") != "comparison-gap" - ): - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection prior record is not a comparison-gap error") - ) - if not isinstance(match_record, dict) or match_record.get("outcome") != run_shadow.OUTCOME_MATCH: - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection current record is not a match") - ) - try: - journal_report = run_journal.read_journal_bounded(run_lifecycle._journal_path(run_dir)) - except (OSError, run_journal.RunJournalError) as exc: - raise run_lifecycle._bound_journal_failure(exc) from exc - if journal_report.partial_tail is not None or journal_report.chain_errors or not journal_report.events: - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection journal is not chain-valid") - ) - tail = journal_report.events[-1] - tail_seq = tail.sequence - tail_digest = tail.event_digest - for record in (gap_record, match_record): - if record.get("sequence") != tail_seq or record.get("event_digest") != tail_digest: - raise run_lifecycle.LifecycleJournalError( - run_events._bound("journal-ahead projection gap/match records do not match the current tail") - ) - - -def _write_json(path: Path, payload: object) -> None: +def _write_json_inner(path: Path, payload: object) -> None: # run.json is polled by `brigade runs watch/steer/interrupt` while the run # rewrites it, so the write must be atomic or a concurrent reader can # observe a truncated file. @@ -860,15 +775,22 @@ def _write_json(path: Path, payload: object) -> None: ) # Prior authority gate (authoritative only): fail closed BEFORE # the checkpoint/lifecycle append when the prior committed state - # is not ready and not a genuine journal-ahead. Legacy and - # authority-requested runs skip this gate; authority-requested - # bases projection on the post-parity readiness report, and legacy - # skips all prior-decision work (no shadow/journal reads). - prior_decision: str | None + # is not ready and not exactly one recoverable checkpoint/event + # pair ahead. The catch-up uses the persisted pre-write snapshot, + # never the incoming next status. Legacy and authority-requested + # runs skip this gate. if authority_state == "authoritative": - prior_decision = _authoritative_prior_decision(run_dir) - else: - prior_decision = None + try: + prior_snapshot = json.loads(path.read_bytes()) + except (OSError, ValueError, UnicodeDecodeError) as exc: + raise run_lifecycle.LifecycleJournalError( + run_events._bound("authoritative prior snapshot is unreadable") + ) from exc + if not isinstance(prior_snapshot, dict): + raise run_lifecycle.LifecycleJournalError( + run_events._bound("authoritative prior snapshot is not an object") + ) + _authoritative_prior_decision(run_dir, prior_snapshot) # Exact order: activate the journal, publish the recovery # checkpoint, append the lifecycle status transition, record the # shadow parity, consult the post-parity readiness veto, then @@ -912,24 +834,12 @@ def _write_json(path: Path, payload: object) -> None: else: localio.write_text_atomic(path, encoded_candidate) return - # authoritative: prior_decision is "ready" or "journal-ahead". + # Authoritative writes require both the prior gate and this + # post-parity gate to be fully ready. No comparison-gap override + # survives: the only recoverable lag was consumed before append. if readiness.ready: _project_authority_candidate(path, run_dir, candidate) return - # Post-parity gate is not ready. For the approved prior - # journal-ahead exception, allow projection despite the current - # parity gap ONLY when the post-parity gate failure is exactly - # the one comparison-gap error created by catching up a genuinely - # ahead journal AND the current parity outcome is match (finding - # 3). Any other post-parity mismatch or error fails closed with no - # run.json replace. For a prior-ready normal write, require the - # post-parity report to remain ready; any other prior reason - # already failed closed at the prior gate, so nothing unrelated - # piggybacks here. - if prior_decision == "journal-ahead": - _validate_journal_ahead_projection(run_dir, readiness) - _project_authority_candidate(path, run_dir, candidate) - return raise run_lifecycle.LifecycleJournalError( run_events._bound("authoritative run gate not ready: " + ",".join(sorted(readiness.reasons))) ) @@ -938,6 +848,20 @@ def _write_json(path: Path, payload: object) -> None: run_shadow.record_shadow_comparison(path.parent, payload) +def _write_json(path: Path, payload: object) -> None: + """Write JSON, serializing every run.json checkpoint/event transaction.""" + if ( + path.name == "run.json" + and isinstance(payload, dict) + and isinstance(payload.get("status"), str) + and payload["status"] + ): + with run_lifecycle.checkpoint_event_pair(): + _write_json_inner(path, payload) + return + _write_json_inner(path, payload) + + def _revision_contains(revisions_dir: Path, projection: bytes) -> bool: """Return whether a preserved sidecar revision matches ``projection``.""" if not revisions_dir.is_dir(): @@ -1768,6 +1692,10 @@ def dispatch( on_stage_start: Callable[[int, tuple[str, ...]], None] | None = None, on_interrupt: Callable[[], None] | None = None, on_scheduler_resolved: Callable[[str, str | None], None] | None = None, + on_dispatch_requested: Callable[[Agent], int | None] | None = None, + on_dispatch_observed: Callable[[Agent, int], None] | None = None, + on_dispatch_completed: Callable[[Agent, int], None] | None = None, + on_dispatch_failed: Callable[[Agent, int], None] | None = None, process_registry: proc.ProcessRegistry | None = None, build_prompt: Callable[..., str] | None = None, ) -> list[WorkerResult]: @@ -1799,6 +1727,10 @@ def dispatch( on_stage_start=on_stage_start, on_interrupt=on_interrupt, on_scheduler_resolved=on_scheduler_resolved, + on_dispatch_requested=on_dispatch_requested, + on_dispatch_observed=on_dispatch_observed, + on_dispatch_completed=on_dispatch_completed, + on_dispatch_failed=on_dispatch_failed, process_registry=process_registry, ) @@ -2488,8 +2420,8 @@ def record_dispatch_stage(output_dir: Path, *, stage: int, seats: tuple[str, ... raise runguard.RetainRunLockError(f"failed to write dispatch stage receipt: {exc}") from exc -def record_result_processing(output_dir: Path, *, seat: str) -> None: - """Record post-dispatch result processing and clear completed worker ownership.""" +def record_result_processing(output_dir: Path, *, seat: str | None = None) -> None: + """Record aggregate post-dispatch result processing and clear worker ownership.""" run_path = output_dir / "run.json" try: @@ -2502,13 +2434,11 @@ def record_result_processing(output_dir: Path, *, seat: str) -> None: raise runguard.RetainRunLockError("run receipt must contain an object during result processing") if payload.get("finished_at"): return - payload.update( - { - "status": "result-processing", - "status_started_at": _utc_iso(datetime.now(timezone.utc)), - "phase_owner": seat, - } - ) + payload.update({"status": "result-processing", "status_started_at": _utc_iso(datetime.now(timezone.utc))}) + if seat is None: + payload.pop("phase_owner", None) + else: + payload["phase_owner"] = seat payload.pop("active_stage", None) payload.pop("active_seats", None) try: @@ -3485,6 +3415,36 @@ def dispatch_interrupted() -> None: active_seats=active_seats, ) + def dispatch_fact(event_type: str, agent: Agent, attempt: int | None = None) -> int | None: + if output_dir is None: + return None + try: + event = run_lifecycle.record_dispatch_fact( + output_dir, + workspace=lock_workspace, + event_type=event_type, + seat=agent.name, + attempt=attempt, + ) + except (OSError, run_lifecycle.LifecycleJournalError, run_checkpoint.CheckpointError) as exc: + raise runguard.RetainRunLockError(f"failed to record dispatch lifecycle fact: {exc}") from exc + if event is None: + return None + recorded_attempt = event.payload.get("attempt") + return recorded_attempt if isinstance(recorded_attempt, int) else None + + def dispatch_requested(agent: Agent) -> int | None: + return dispatch_fact("run.dispatch.requested", agent) + + def dispatch_observed(agent: Agent, attempt: int) -> None: + dispatch_fact("run.dispatch.observed", agent, attempt) + + def dispatch_completed(agent: Agent, attempt: int) -> None: + dispatch_fact("run.dispatch.completed", agent, attempt) + + def dispatch_failed(agent: Agent, attempt: int) -> None: + dispatch_fact("run.dispatch.failed", agent, attempt) + active_seat = active_seats[0] if len(active_seats) == 1 else None worker_prompt_builder = partial(_worker_prompt, skill_policy=skill_policy) try: @@ -3513,6 +3473,10 @@ def dispatch_interrupted() -> None: on_stage_start=stage_started, on_interrupt=dispatch_interrupted, on_scheduler_resolved=scheduler_resolved, + on_dispatch_requested=dispatch_requested, + on_dispatch_observed=dispatch_observed, + on_dispatch_completed=dispatch_completed, + on_dispatch_failed=dispatch_failed, process_registry=process_registry, build_prompt=worker_prompt_builder, ) @@ -3568,7 +3532,7 @@ def dispatch_interrupted() -> None: if drift_rc is not None: return drift_rc if output_dir is not None: - record_result_processing(output_dir, seat=roster.orchestrator) + record_result_processing(output_dir) if output_dir is not None and code_graph_delta_before is not None and cwd is not None: code_graph_delta = graphtrail_delta.capture_after_and_diff(cwd, output_dir, code_graph_delta_before) context_eval_payload = _context_eval_for_run(code_graph, code_graph_delta) diff --git a/src/brigade/doctor.py b/src/brigade/doctor.py index 6388aa12..f7d17ef4 100644 --- a/src/brigade/doctor.py +++ b/src/brigade/doctor.py @@ -268,13 +268,22 @@ def _recovery_checkpoint_run_verdict(target: Path, run_dir: Path) -> tuple[str, except run_projector.ProjectionError: return "fail", "projection failed" + pending_dispatch = run_lifecycle.pending_dispatch_requests(report.events) + + def warn_with_pending(reason: str) -> tuple[str, str]: + if not pending_dispatch: + return "warn", reason + seat, attempt = pending_dispatch[0] + attempt_text = str(attempt) if attempt is not None else "unknown" + return "warn", f"{reason}; at-least-once dispatch recovery required (seat={seat}, attempt={attempt_text})" + run_json_path = run_dir / "run.json" try: run_json_present = run_json_path.is_file() except OSError: run_json_present = False if not run_json_present: - return "warn", "run.json missing with valid checkpoint" + return warn_with_pending("run.json missing with valid checkpoint") try: run_bytes = run_json_path.read_bytes() @@ -284,30 +293,32 @@ def _recovery_checkpoint_run_verdict(target: Path, run_dir: Path) -> tuple[str, try: parsed = json.loads(run_bytes.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError, ValueError, RecursionError): - return "warn", "run.json unparseable with valid checkpoint" + return warn_with_pending("run.json unparseable with valid checkpoint") run_meta = parsed if isinstance(parsed, dict) else None if run_meta is None: - return "warn", "run.json unparseable with valid checkpoint" + return warn_with_pending("run.json unparseable with valid checkpoint") if run_bytes == compare_bytes: + success_reason = "projected snapshot matches run.json" if is_authority else "checkpoint bytes match run.json" + else: + expected = _reconstruct_stale_lock_recovery_receipt(compare_obj, run_meta) + if expected is not None and expected == run_meta: + success_reason = ( + "stale-lock-recovery receipt matches projection reconstruction" + if is_authority + else "stale-lock-recovery receipt matches checkpoint reconstruction" + ) + else: + success_reason = None + if success_reason is None: return ( - "ok", - "projected snapshot matches run.json" if is_authority else "checkpoint bytes match run.json", - ) - - expected = _reconstruct_stale_lock_recovery_receipt(compare_obj, run_meta) - if expected is not None and expected == run_meta: - return ( - "ok", - "stale-lock-recovery receipt matches projection reconstruction" - if is_authority - else "stale-lock-recovery receipt matches checkpoint reconstruction", + "fail", + "run.json does not match projected snapshot" if is_authority else "run.json does not match checkpoint", ) - return ( - "fail", - "run.json does not match projected snapshot" if is_authority else "run.json does not match checkpoint", - ) + if pending_dispatch: + return warn_with_pending(success_reason) + return "ok", success_reason def _read_run_meta_fail_safe(run_json_path: Path) -> dict[str, object] | None: diff --git a/src/brigade/run_checkpoint.py b/src/brigade/run_checkpoint.py index 96340906..a0067be0 100644 --- a/src/brigade/run_checkpoint.py +++ b/src/brigade/run_checkpoint.py @@ -37,7 +37,7 @@ _CHECKPOINT_PAYLOAD_REQUIRED_KEYS = frozenset( {"path", "sha256", "media_type", "byte_size", "privacy_class", "paired_event_type"} ) -_CHECKPOINT_PAYLOAD_OPTIONAL_KEYS = frozenset({"body_kind"}) +_CHECKPOINT_PAYLOAD_OPTIONAL_KEYS = frozenset({"body_kind", "pairing_key"}) # Backwards-compatible alias kept for any external reader of the closed key set; # the validation logic now treats body_kind as optional. _CHECKPOINT_PAYLOAD_KEYS = _CHECKPOINT_PAYLOAD_REQUIRED_KEYS | _CHECKPOINT_PAYLOAD_OPTIONAL_KEYS @@ -63,6 +63,14 @@ _DURABLE_REQUEST_FIELDS = frozenset({"lifecycle_journal_requested", "run_journal_authority_requested"}) _BODY_KIND_BASE_STRIPPED = "base-stripped" +_DISPATCH_FACT_EVENT_TYPES = frozenset( + { + "run.dispatch.requested", + "run.dispatch.observed", + "run.dispatch.completed", + "run.dispatch.failed", + } +) _CHECKPOINT_IDEMPOTENCY_BASE_STRIPPED_PREFIX = f"{_CHECKPOINT_IDEMPOTENCY_PREFIX}:base-stripped" @@ -122,6 +130,11 @@ def _validate_payload(payload: Any) -> None: if not isinstance(body_kind, str) or body_kind != _BODY_KIND_BASE_STRIPPED: raise CheckpointError(_bound("body_kind must be base-stripped"), category="body-kind") + if "pairing_key" in payload: + pairing_key = payload["pairing_key"] + if not isinstance(pairing_key, str) or not _HEX64.match(pairing_key): + raise CheckpointError(_bound("pairing_key must be 64-char lowercase hex"), category="pairing-key") + if payload["media_type"] != CHECKPOINT_MEDIA_TYPE: raise CheckpointError(_bound("media_type mismatch"), category="media-type") @@ -161,9 +174,9 @@ def _validate_payload(payload: Any) -> None: raise CheckpointError(_bound("paired_event_type must be null or a string"), category="paired-event-type") elif paired not in run_events.EVENT_TYPES: raise CheckpointError(_bound("paired_event_type not in registry"), category="paired-event-type") - elif paired not in _mapped_lifecycle_event_types(): + elif paired not in _mapped_lifecycle_event_types() and paired not in _DISPATCH_FACT_EVENT_TYPES: raise CheckpointError( - _bound("paired_event_type is not a mapped lifecycle status event"), + _bound("paired_event_type is not a mapped lifecycle or dispatch fact event"), category="paired-event-type", ) @@ -608,6 +621,7 @@ def _checkpoint_payload( *, paired_event_type: str | None, body_kind: str | None = None, + pairing_key: str | None = None, ) -> dict[str, Any]: sha = hashlib.sha256(run_json_bytes).hexdigest() payload: dict[str, Any] = { @@ -620,32 +634,52 @@ def _checkpoint_payload( } if body_kind is not None: payload["body_kind"] = body_kind + if pairing_key is not None: + payload["pairing_key"] = pairing_key return payload +def dispatch_pairing_key(event_type: str, seat: str, attempt: int) -> str: + """Return the stable dispatch checkpoint identity for one worker action.""" + return hashlib.sha256( + run_events.canonical_bytes({"event_type": event_type, "seat": seat, "attempt": attempt}) + ).hexdigest() + + def _checkpoint_idempotency_key( sha: str, *, paired_event_type: str | None, body_kind: str | None = None, + pairing_key: str | None = None, ) -> str: paired = paired_event_type if paired_event_type is not None else "none" if body_kind == _BODY_KIND_BASE_STRIPPED: prefix = _CHECKPOINT_IDEMPOTENCY_BASE_STRIPPED_PREFIX else: prefix = _CHECKPOINT_IDEMPOTENCY_PREFIX - key = f"{prefix}:{sha}:{paired}" + if pairing_key is None: + key = f"{prefix}:{sha}:{paired}" + if len(key) <= run_events.MAX_IDEMPOTENCY_KEY_LEN: + return key + budget = run_events.MAX_IDEMPOTENCY_KEY_LEN - len(prefix) - 1 - len(sha) - 1 + if budget < 0: + paired_digest = hashlib.sha256(paired.encode("utf-8")).hexdigest()[:16] + return f"{prefix}:{sha}:{paired_digest}" + return f"{prefix}:{sha}:{paired[:budget]}" + + key = f"{prefix}:{sha}:{paired}:{pairing_key}" if len(key) <= run_events.MAX_IDEMPOTENCY_KEY_LEN: return key - # Bound the paired-event-type tail so the key stays within the envelope - # limit regardless of event_type length; the prefix and sha are fixed. - budget = run_events.MAX_IDEMPOTENCY_KEY_LEN - len(prefix) - 1 - len(sha) - 1 + # Bound the pairing identity so the key stays within the envelope limit + # regardless of event type length; the prefix and snapshot digest are fixed. + budget = run_events.MAX_IDEMPOTENCY_KEY_LEN - len(prefix) - 1 - len(sha) - 1 - len(paired) - 1 if budget < 0: # Pathological: prefix+sha already overflow the bound. Fall back to a # digest of the paired type so the key is still unique and bounded. - paired_digest = hashlib.sha256(paired.encode("utf-8")).hexdigest()[:16] - return f"{prefix}:{sha}:{paired_digest}" - return f"{prefix}:{sha}:{paired[:budget]}" + identity_digest = hashlib.sha256(pairing_key.encode("utf-8")).hexdigest()[:16] + return f"{prefix}:{sha}:{identity_digest}" + return f"{prefix}:{sha}:{paired}:{pairing_key[:budget]}" def write_checkpoint( @@ -655,6 +689,7 @@ def write_checkpoint( workspace: Path | None = None, paired_event_type: str | None, body_kind: str | None = None, + pairing_key: str | None = None, ) -> "run_journal.RunEvent | None": """Publish a crash-safe recovery checkpoint and append the checkpoint event. @@ -708,6 +743,8 @@ def write_checkpoint( run_dir = Path(run_dir).expanduser().resolve() run_json_bytes = bytes(run_json_bytes) + if pairing_key is not None and (not isinstance(pairing_key, str) or not _HEX64.match(pairing_key)): + raise CheckpointError(_bound("pairing_key must be 64-char lowercase hex"), category="pairing-key") # Validate body_kind and, for base-stripped, strip the journal-derived # metadata fields BEFORE SHA/payload/publish AND before # prepare_lifecycle_journal, so the content-addressed snapshot is stable @@ -767,7 +804,12 @@ def write_checkpoint( run_events._bound("lifecycle journal append requires the active run lock for this run") ) sha = hashlib.sha256(publish_bytes).hexdigest() - payload = _checkpoint_payload(publish_bytes, paired_event_type=paired_event_type, body_kind=body_kind) + payload = _checkpoint_payload( + publish_bytes, + paired_event_type=paired_event_type, + body_kind=body_kind, + pairing_key=pairing_key, + ) # Crash-safe publish FIRST. A CheckpointError here fails before the # lifecycle status append and before run.json replacement. publish_checkpoint_file(run_dir, publish_bytes) @@ -775,7 +817,12 @@ def write_checkpoint( report = run_journal.read_journal(journal_path) if report.partial_tail is not None or report.chain_errors: raise run_journal.ChainIntegrityError(run_events._bound(run_lifecycle._CHAIN_CATEGORY)) - idempotency_key = _checkpoint_idempotency_key(sha, paired_event_type=paired_event_type, body_kind=body_kind) + idempotency_key = _checkpoint_idempotency_key( + sha, + paired_event_type=paired_event_type, + body_kind=body_kind, + pairing_key=pairing_key, + ) return run_journal.append_event( journal_path, run_id=run_lifecycle._run_id_from_dir(run_dir), @@ -890,6 +937,8 @@ def _paired_event_derived_status(event: "run_journal.RunEvent") -> str | None: from brigade import run_projector # lazy: avoid import cycle event_type = event.event_type + if event_type == "run.dispatch.completed" and not run_projector._has_dispatch_identity(event.payload): + return "result-processing" if event_type in run_projector.EVENT_STATUS: return run_projector.EVENT_STATUS[event_type] rule = run_projector._PAYLOAD_STATUS_RULES.get(event_type) @@ -919,10 +968,20 @@ def _verify_coverage( status the paired event must derive to. """ paired_event_type = latest.payload.get("paired_event_type") + pairing_key = latest.payload.get("pairing_key") tail = events[-1] if latest.sequence == tail.sequence: - # Checkpoint is the tail. Covered for both null and non-null - # paired_event_type (a crash before the paired status append). + # A dispatch pairing key promises a specific identity-bearing fact. + # A checkpoint at tail means that fact never committed, so recovery + # must surface the incomplete pair instead of treating the checkpoint + # as covered. + if pairing_key is not None: + raise CheckpointError( + _bound("dispatch checkpoint pair is incomplete"), + category="incomplete-pair", + ) + # Legacy lifecycle checkpoints remain recoverable at tail after a + # crash before their paired status append. return if paired_event_type is None: # Null pairing covers only checkpoint-at-tail; any following event @@ -936,6 +995,24 @@ def _verify_coverage( paired = following[0] if paired.event_type != paired_event_type: raise CheckpointError(_bound("journal tail is not covered by the latest checkpoint"), category="uncovered-tail") + if pairing_key is not None: + seat = paired.payload.get("seat") + attempt = paired.payload.get("attempt") + if ( + paired.event_type not in _DISPATCH_FACT_EVENT_TYPES + or not isinstance(seat, str) + or not seat + or isinstance(attempt, bool) + or not isinstance(attempt, int) + or attempt < 1 + ): + raise CheckpointError(_bound("checkpoint pairing identity is invalid"), category="pairing") + if pairing_key != dispatch_pairing_key(paired.event_type, seat, attempt): + raise CheckpointError(_bound("checkpoint pairing identity does not match event"), category="pairing") + # Identity-bearing dispatch facts are status-neutral. Their paired + # checkpoint preserves the aggregate status already in run.json, so + # no event-derived status comparison applies. + return derived = _paired_event_derived_status(paired) checkpoint_status = checkpoint_obj.get("status") if not isinstance(checkpoint_status, str) or derived != checkpoint_status: diff --git a/src/brigade/run_events.py b/src/brigade/run_events.py index 209f5708..cdb68680 100644 --- a/src/brigade/run_events.py +++ b/src/brigade/run_events.py @@ -73,10 +73,12 @@ "run.planning.started": frozenset({"detail"}), "run.planning.completed": frozenset({"detail"}), "run.planning.failed": frozenset({"detail"}), + "run.dispatching.started": frozenset({"detail"}), "run.dispatch.requested": frozenset({"seat", "attempt", "detail"}), "run.dispatch.observed": frozenset({"seat", "attempt", "detail"}), "run.dispatch.completed": frozenset({"seat", "attempt", "detail"}), "run.dispatch.failed": frozenset({"seat", "attempt", "detail"}), + "run.result-processing.started": frozenset({"detail"}), "run.synthesis.started": frozenset({"detail"}), "run.synthesis.completed": frozenset({"detail"}), "run.synthesis.failed": frozenset({"detail"}), @@ -93,7 +95,16 @@ "run.failed": frozenset({"status", "detail"}), "run.interrupted": frozenset({"status", "detail"}), "run.snapshot.checkpointed": frozenset( - {"path", "sha256", "media_type", "byte_size", "privacy_class", "paired_event_type", "body_kind"} + { + "path", + "sha256", + "media_type", + "byte_size", + "privacy_class", + "paired_event_type", + "body_kind", + "pairing_key", + } ), "run.artifact_collection.started": frozenset({"detail"}), } diff --git a/src/brigade/run_lifecycle.py b/src/brigade/run_lifecycle.py index 21a86793..fae42012 100644 --- a/src/brigade/run_lifecycle.py +++ b/src/brigade/run_lifecycle.py @@ -74,7 +74,12 @@ import hashlib import json import os +import signal +import stat +import threading +from contextlib import contextmanager from collections.abc import Mapping +from collections.abc import Iterator from pathlib import Path from typing import Any @@ -97,8 +102,8 @@ STATUS_EVENT_TYPE: dict[str, str] = { "started": "run.created", "planning": "run.planning.started", - "dispatching": "run.dispatch.requested", - "result-processing": "run.dispatch.completed", + "dispatching": "run.dispatching.started", + "result-processing": "run.result-processing.started", "synthesizing": "run.synthesis.started", "handoff": "run.synthesis.completed", "ok": "run.completed", @@ -110,11 +115,70 @@ "artifact-collection": "run.artifact_collection.started", } +_CHECKPOINT_EVENT_PAIR_LOCK = threading.Lock() +_CHECKPOINT_EVENT_PAIR_STATE = threading.local() +_DISPATCH_FACT_TYPES = frozenset( + { + "run.dispatch.requested", + "run.dispatch.observed", + "run.dispatch.completed", + "run.dispatch.failed", + } +) + class LifecycleJournalError(RuntimeError): """Bounded lifecycle-journal failure; run.json must not advance past it.""" +@contextmanager +def checkpoint_event_pair() -> Iterator[None]: + """Serialize one checkpoint/event pair and defer SIGTERM across its gap. + + The lock is separate from ``run_journal``'s append lock. It spans the + checkpoint and paired event at the process level, so worker threads and + interruption writers cannot interleave pair members. SIGTERM is blocked + in the entering thread before it waits for the lock and restored to the + exact prior mask only after the pair lock is released. That ordering lets + a pending main-thread handler enter the writer normally instead of + reentering the protected region while its lock is still held. + """ + if getattr(_CHECKPOINT_EVENT_PAIR_STATE, "active", False): + raise LifecycleJournalError(run_events._bound("checkpoint/event pair reentry is not allowed")) + + previous_mask: set[int | signal.Signals] | None = None + if hasattr(signal, "pthread_sigmask"): + try: + previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGTERM}) + except (OSError, ValueError) as exc: + raise LifecycleJournalError(run_events._bound("checkpoint/event pair signal mask failed")) from exc + + acquired = False + restore_error: BaseException | None = None + try: + # Mark the whole wait-and-hold interval active. On platforms where + # SIGTERM cannot be masked, a same-thread signal handler that runs + # while lock acquisition is waiting must fail bounded instead of + # trying to acquire this non-reentrant lock again. + _CHECKPOINT_EVENT_PAIR_STATE.active = True + _CHECKPOINT_EVENT_PAIR_LOCK.acquire() + acquired = True + yield + finally: + _CHECKPOINT_EVENT_PAIR_STATE.active = False + if acquired: + _CHECKPOINT_EVENT_PAIR_LOCK.release() + if previous_mask is not None: + try: + signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask) + except (OSError, ValueError) as exc: + restore_error = exc + if restore_error is not None: + raise LifecycleJournalError( + run_events._bound("checkpoint/event pair signal mask restoration failed") + ) from restore_error + + def is_lifecycle_journaling_enabled() -> bool: """True when the opt-in lifecycle journal flag is set in the environment.""" return os.environ.get(_FLAG_ENV, "").strip().lower() in _TRUTHY @@ -128,6 +192,55 @@ def _journal_path(run_dir: Path) -> Path: return run_dir / "events" / _JOURNAL_NAME +def _dispatch_journal_path(run_dir: Path) -> Path | None: + """Return the active regular journal, or ``None`` only for true legacy. + + Dispatch is the last boundary before an external worker invocation. A + missing, renamed, or substituted journal after durable lifecycle or + authority enrollment must fail closed there. A snapshot is considered + truly unenrolled only when it is a regular JSON object and carries none + of the durable request or projection metadata fields. + """ + journal_path = _journal_path(run_dir) + try: + journal_mode = journal_path.lstat().st_mode + except FileNotFoundError: + journal_mode = None + except OSError as exc: + raise LifecycleJournalError(run_events._bound("lifecycle journal enrollment check failed")) from exc + if journal_mode is not None: + if not stat.S_ISREG(journal_mode): + raise LifecycleJournalError(run_events._bound("enrolled lifecycle journal is not a regular file")) + return journal_path + + run_json = run_dir / "run.json" + try: + run_mode = run_json.lstat().st_mode + except FileNotFoundError as exc: + raise LifecycleJournalError(run_events._bound("dispatch run receipt is missing")) from exc + except OSError as exc: + raise LifecycleJournalError(run_events._bound("dispatch run receipt enrollment check failed")) from exc + if not stat.S_ISREG(run_mode): + raise LifecycleJournalError(run_events._bound("dispatch run receipt is not a regular file")) + try: + meta = json.loads(run_json.read_bytes()) + except (OSError, ValueError, UnicodeDecodeError) as exc: + raise LifecycleJournalError(run_events._bound("dispatch run receipt enrollment is unreadable")) from exc + if not isinstance(meta, dict): + raise LifecycleJournalError(run_events._bound("dispatch run receipt enrollment is unreadable")) + enrollment_fields = { + _REQUEST_FIELD, + "run_journal_authority_requested", + "projector_version", + "journal_present", + "journal_last_sequence", + "journal_last_event_digest", + } + if any(field in meta for field in enrollment_fields): + raise LifecycleJournalError(run_events._bound("enrolled lifecycle journal is missing")) + return None + + def _journal_requested( run_dir: Path, *, @@ -172,6 +285,151 @@ def _run_snapshot_state(run_dir: Path) -> tuple[str | None, str | None]: return status, digest +def pending_dispatch_requests(events: list[run_journal.RunEvent]) -> list[tuple[str, int | None]]: + """Return dispatch requests without a terminal observation. + + A completed, failed, or observed fact closes the matching seat/attempt. + Missing or malformed legacy payloads remain explicit unknown + at-least-once work instead of being silently discarded. + """ + pending: list[tuple[str, int | None]] = [] + for event in events: + if event.event_type == "run.dispatch.requested": + seat = event.payload.get("seat") + attempt = event.payload.get("attempt") + # Historical aggregate dispatch status events carried only a + # detail field. They do not identify a worker action and must not + # manufacture perpetual recovery work for every legacy run. + if seat is None and attempt is None: + continue + valid_seat = seat if isinstance(seat, str) and seat else "unknown" + valid_attempt = ( + attempt if isinstance(attempt, int) and not isinstance(attempt, bool) and attempt > 0 else None + ) + pending.append((valid_seat, valid_attempt)) + continue + if event.event_type not in {"run.dispatch.observed", "run.dispatch.completed", "run.dispatch.failed"}: + continue + seat = event.payload.get("seat") + attempt = event.payload.get("attempt") + if ( + not isinstance(seat, str) + or not seat + or isinstance(attempt, bool) + or not isinstance(attempt, int) + or attempt < 1 + ): + continue + pending = [item for item in pending if item != (seat, attempt)] + return pending + + +def record_dispatch_fact( + run_dir: Path, + *, + workspace: Path | None, + event_type: str, + seat: str, + attempt: int | None = None, +) -> run_journal.RunEvent | None: + """Append one paired per-invocation dispatch fact under a process lock. + + Each worker transport call receives its own checkpoint-event pair. The + lock covers both appends so wave and DAG workers cannot interleave a + checkpoint with another seat's fact. The event payload deliberately holds + only the selected seat and its monotonically allocated attempt identity. + """ + if event_type not in _DISPATCH_FACT_TYPES: + raise LifecycleJournalError(run_events._bound("invalid dispatch fact event type")) + if not isinstance(seat, str) or not seat: + raise LifecycleJournalError(run_events._bound("dispatch fact seat must be non-empty")) + if attempt is None and event_type != "run.dispatch.requested": + raise LifecycleJournalError(run_events._bound("only dispatch requested may allocate an attempt")) + if attempt is not None and (isinstance(attempt, bool) or not isinstance(attempt, int) or attempt < 1): + raise LifecycleJournalError(run_events._bound("dispatch fact attempt must be a positive integer")) + + run_dir = Path(run_dir).expanduser().resolve() + with checkpoint_event_pair(): + journal_path = _dispatch_journal_path(run_dir) + if journal_path is None: + return None + if workspace is None or not runguard.is_active_run_owner(workspace, run_dir): + raise LifecycleJournalError("lifecycle journal append requires the active run lock for this run") + try: + # Import lazily to avoid the aboyeur -> lifecycle module cycle. + # These are the same authority gates used by run.json status + # writers, now applied before every dispatch pair. + from brigade import aboyeur, run_shadow + + snapshot = (run_dir / "run.json").read_bytes() + try: + snapshot_obj = json.loads(snapshot) + except (ValueError, UnicodeDecodeError) as exc: + raise LifecycleJournalError(run_events._bound("dispatch run receipt is not valid JSON")) from exc + if not isinstance(snapshot_obj, dict): + raise LifecycleJournalError(run_events._bound("dispatch run receipt is not a JSON object")) + authority_state = aboyeur._resolve_authority_state(run_dir) + if authority_state == "authoritative": + # The shared prior gate catches up exactly one verified + # checkpoint/event pair before any new dispatch or aggregate + # status pair can advance the journal again. + aboyeur._authoritative_prior_decision(run_dir, snapshot_obj) + if attempt is None: + report = run_journal.read_journal(journal_path) + if report.partial_tail is not None or report.chain_errors: + raise run_journal.ChainIntegrityError(run_events._bound(_CHAIN_CATEGORY)) + attempts: list[int] = [] + for event in report.events: + candidate = event.payload.get("attempt") + if ( + event.payload.get("seat") == seat + and isinstance(candidate, int) + and not isinstance(candidate, bool) + ): + attempts.append(candidate) + attempt = max(attempts, default=0) + 1 + assert attempt is not None + pairing_key = run_checkpoint.dispatch_pairing_key(event_type, seat, attempt) + checkpoint = run_checkpoint.write_checkpoint( + run_dir, + snapshot, + workspace=workspace, + paired_event_type=event_type, + body_kind=( + run_checkpoint._BODY_KIND_BASE_STRIPPED + if snapshot_obj.get("run_journal_authority_requested") is True + else None + ), + pairing_key=pairing_key, + ) + if checkpoint is None: + raise LifecycleJournalError(run_events._bound("enrolled lifecycle journal checkpoint was not recorded")) + report = run_journal.read_journal(journal_path) + if report.partial_tail is not None or report.chain_errors: + raise run_journal.ChainIntegrityError(run_events._bound(_CHAIN_CATEGORY)) + key_digest = hashlib.sha256( + run_events.canonical_bytes( + {"event_type": event_type, "seat": seat, "attempt": attempt, "pairing_key": pairing_key} + ) + ).hexdigest() + event = run_journal.append_event( + journal_path, + run_id=_run_id_from_dir(run_dir), + event_type=event_type, + payload={"seat": seat, "attempt": attempt, "detail": event_type.rsplit(".", 1)[-1]}, + idempotency_key=f"dispatch:{key_digest[:32]}", + expected_previous_sequence=report.events[-1].sequence if report.events else 0, + ) + run_shadow.record_shadow_comparison(run_dir, snapshot_obj) + return event + except run_journal.RunJournalError as exc: + raise _bound_journal_failure(exc) from exc + except run_events.CanonicalizationError as exc: + raise _bound_journal_failure(exc) from exc + except OSError as exc: + raise _bound_journal_failure(exc) from exc + + def _allowlisted_payload(event_type: str, *, status: str) -> dict[str, Any]: """Build a payload carrying only keys in the closed per-type allowlist.""" allowed = run_events.EVENT_TYPES[event_type] @@ -345,7 +603,10 @@ def record_lifecycle_transition( __all__ = [ "LifecycleJournalError", "STATUS_EVENT_TYPE", + "checkpoint_event_pair", "is_lifecycle_journaling_enabled", + "pending_dispatch_requests", "prepare_lifecycle_journal", + "record_dispatch_fact", "record_lifecycle_transition", ] diff --git a/src/brigade/run_projector.py b/src/brigade/run_projector.py index ec8a2e3e..b2d8d177 100644 --- a/src/brigade/run_projector.py +++ b/src/brigade/run_projector.py @@ -26,7 +26,7 @@ from brigade import run_checkpoint, run_events, run_journal -PROJECTOR_VERSION: int = 3 +PROJECTOR_VERSION: int = 4 # Field ownership over the run.json contract. Every current run.json key is # in exactly one of these two sets; see the ownership inventory in @@ -106,13 +106,29 @@ # the event-to-status table). EVENT_STATUS: dict[str, str] = { "run.planning.started": "planning", + "run.dispatching.started": "dispatching", "run.dispatch.requested": "dispatching", - "run.dispatch.completed": "result-processing", + "run.dispatch.observed": "dispatching", + "run.dispatch.completed": "dispatching", + "run.dispatch.failed": "dispatching", + "run.result-processing.started": "result-processing", "run.synthesis.started": "synthesizing", "run.synthesis.completed": "handoff", "run.artifact_collection.started": "artifact-collection", } + +def _has_dispatch_identity(payload: Any) -> bool: + return ( + isinstance(payload, Mapping) + and isinstance(payload.get("seat"), str) + and bool(payload["seat"]) + and isinstance(payload.get("attempt"), int) + and not isinstance(payload["attempt"], bool) + and payload["attempt"] > 0 + ) + + # Payload-driven rows: event_type -> (allowed payload status values, derived # status). A None derived status means the payload status value is used # directly as the derived status (run.failed: "failed" vs "timeout"). @@ -258,6 +274,15 @@ def _derive_status(envelopes: list[Mapping[str, Any]], base_status: Any) -> str: sequence = env["sequence"] if event_type == run_checkpoint.CHECKPOINT_EVENT_TYPE: continue + # Slice-9 per-worker facts are status-neutral: a worker terminal + # result must not advance the aggregate run while other seats remain + # active. Older aggregate envelopes lacked an identity and retain + # their original status semantics below. + if event_type.startswith("run.dispatch.") and _has_dispatch_identity(env["payload"]): + continue + if event_type == "run.dispatch.completed" and not _has_dispatch_identity(env["payload"]): + status = "result-processing" + continue if event_type in EVENT_STATUS: status = EVENT_STATUS[event_type] continue diff --git a/src/brigade/run_shadow.py b/src/brigade/run_shadow.py index bed2ac26..b8005ee3 100644 --- a/src/brigade/run_shadow.py +++ b/src/brigade/run_shadow.py @@ -83,7 +83,7 @@ def _valid_counter(value: object) -> bool: def _has_invalid_counters(data: object) -> bool: - """True when a current-v3 artifact carries an untrustworthy counter. + """True when a current-version artifact carries an untrustworthy counter. ``_record_outcome`` accumulates ``comparisons``, ``matches``, ``mismatches``, ``lags``, and ``errors``. A bool, negative, or non-int @@ -118,7 +118,7 @@ def _has_invalid_recent_records(data: object) -> bool: def _has_invalid_artifact_structure(data: object) -> bool: - """True when a current-v3 artifact is structurally untrustworthy. + """True when a current-version artifact is structurally untrustworthy. Invalid gated counters and malformed ``recent_records`` both close the carry path: quarantine the prior, start a fresh artifact, and record an @@ -330,7 +330,24 @@ def _checkpoint_status_write_gap( return False if status_event.event_type != paired_event_type: return False - return paired_event_type in run_lifecycle.STATUS_EVENT_TYPE.values() + pairing_key = checkpoint.payload.get("pairing_key") + if status_event.event_type in run_checkpoint._DISPATCH_FACT_EVENT_TYPES and pairing_key is None: + return False + if pairing_key is not None: + seat = status_event.payload.get("seat") + attempt = status_event.payload.get("attempt") + if ( + status_event.event_type not in run_checkpoint._DISPATCH_FACT_EVENT_TYPES + or not isinstance(pairing_key, str) + or not isinstance(seat, str) + or not seat + or isinstance(attempt, bool) + or not isinstance(attempt, int) + or attempt < 1 + ): + return False + return pairing_key == run_checkpoint.dispatch_pairing_key(status_event.event_type, seat, attempt) + return run_checkpoint._paired_event_derived_status(status_event) is not None def _append_gap_record(data: dict[str, Any], tail_seq: int, tail_digest: str | None) -> None: @@ -395,8 +412,8 @@ def _record_outcome( data["last_compared_event_digest"] = baseline[1] else: # Current-version artifact: counters and records only accumulate, - # never reset (mismatches/errors from a current-v3 artifact are - # preserved across later comparisons). A current-v3 artifact with + # never reset (mismatches/errors from a current-version artifact + # are preserved across later comparisons). A current-version artifact with # invalid counters (bool, negative, or non-int for comparisons, # matches, mismatches, lags, or errors) or malformed # ``recent_records`` is structurally broken and must not be @@ -424,7 +441,7 @@ def _record_outcome( return # idempotent: same tail sequence, digests, outcome, and category # A present-but-unparseable prior artifact is quarantined and treated as # absent; record an evidence-unreadable error before the main record. A - # current-v3 prior with invalid structure is treated the same way: forged + # current-version prior with invalid structure is treated the same way: forged # counters or malformed recent_records are dropped and an # evidence-unreadable side record explains the fresh start. if was_corrupt or structurally_invalid: @@ -673,7 +690,7 @@ def check_projection_readiness(run_dir: Path) -> ReadinessReport: or data.get("run_id") != run_dir.name ): return ReadinessReport(ready=False, reasons=(REASON_EVIDENCE_SCHEMA_MISMATCH,)) - # Strict structure validation: a current-v3 artifact accepts only + # Strict structure validation: a current-version artifact accepts only # non-bool nonnegative counters and a list of mapping-shaped recent # records. A forged counter or malformed/missing ``recent_records`` # is structurally broken and closes the gate as evidence-unreadable diff --git a/src/brigade/run_transport.py b/src/brigade/run_transport.py index 788fd08e..3c4d66a7 100644 --- a/src/brigade/run_transport.py +++ b/src/brigade/run_transport.py @@ -13,7 +13,7 @@ from typing import Any, Callable, Protocol from urllib.parse import urlparse -from . import agents, proc, run_control +from . import agents, proc, run_control, runguard from .roster import Agent, Roster, is_cli_allowed, timeout_for _GROK_CONTINUATION_PROMPT = ( @@ -265,6 +265,10 @@ def dispatch( on_stage_start: Callable[[int, tuple[str, ...]], None] | None = None, on_interrupt: Callable[[], None] | None = None, on_scheduler_resolved: Callable[[str, str | None], None] | None = None, + on_dispatch_requested: Callable[[Agent], int | None] | None = None, + on_dispatch_observed: Callable[[Agent, int], None] | None = None, + on_dispatch_completed: Callable[[Agent, int], None] | None = None, + on_dispatch_failed: Callable[[Agent, int], None] | None = None, process_registry: proc.ProcessRegistry | None = None, ) -> list[WorkerResult]: """Dispatch staged assignments while keeping transport policy in one module. @@ -335,7 +339,7 @@ def run_one(assignment: Assignment, prior_results: list[WorkerResult]) -> Worker started = time.monotonic() effective_read_only = read_only if sandbox_read_only is None else sandbox_read_only - def invoke( + def _invoke_external( selected_agent: Agent, selected_prompt: str, *, @@ -493,6 +497,30 @@ def invoke( process_registry=process_registry, ) + def invoke( + selected_agent: Agent, + selected_prompt: str, + *, + resume_session_id: str | None = None, + ) -> agents.AgentResult: + """Record transport facts around exactly one real external call.""" + attempt = on_dispatch_requested(selected_agent) if on_dispatch_requested is not None else None + try: + result = _invoke_external(selected_agent, selected_prompt, resume_session_id=resume_session_id) + except BaseException: + if attempt is not None and on_dispatch_failed is not None: + on_dispatch_failed(selected_agent, attempt) + raise + if attempt is not None: + if on_dispatch_observed is not None: + on_dispatch_observed(selected_agent, attempt) + if result.ok: + if on_dispatch_completed is not None: + on_dispatch_completed(selected_agent, attempt) + elif on_dispatch_failed is not None: + on_dispatch_failed(selected_agent, attempt) + return result + def finish( result: agents.AgentResult, terminal_agent: Agent, @@ -690,6 +718,8 @@ def finish( index = future_to_index[future] try: stage_results_by_index[index] = future.result() + except runguard.RetainRunLockError: + raise except Exception as exc: # pragma: no cover - defensive boundary assignment = stage_assignments[index] stage_results_by_index[index] = WorkerResult( @@ -848,6 +878,8 @@ def doomed(i: int) -> bool: try: finished = done.result() results[i] = finished + except runguard.RetainRunLockError: + raise except Exception as exc: finished = WorkerResult( worker=assignments[i].worker, diff --git a/src/brigade/runs_cmd.py b/src/brigade/runs_cmd.py index 101e230d..106b0bde 100644 --- a/src/brigade/runs_cmd.py +++ b/src/brigade/runs_cmd.py @@ -655,6 +655,19 @@ def _journal_active(run_dir: Path) -> bool: return (run_dir / "events" / "lifecycle.jsonl").is_file() +def _print_pending_dispatch_recovery(run_dir: Path) -> None: + """Surface unobserved external dispatches as at-least-once recovery work.""" + from . import run_journal, run_lifecycle + + try: + events = run_journal.read_journal(run_lifecycle._journal_path(run_dir)).events + except (OSError, run_journal.RunJournalError): + return + for seat, attempt in run_lifecycle.pending_dispatch_requests(events): + attempt_text = str(attempt) if attempt is not None else "unknown" + print(f"dispatch recovery: at-least-once work required (seat={seat}, attempt={attempt_text})") + + def _read_run_json_state(run_dir: Path) -> tuple[bool, dict[str, Any] | None, str | None, bool]: """Return (parseable, run_meta, read_error, read_oserror) for run.json. @@ -844,7 +857,10 @@ def recover(run: str | Path, *, cwd: Path, runs_dir: Path | None = None) -> int: return 2 if _journal_active(run_dir): - return _recover_from_checkpoint(run_dir, workspace, parseable, run_meta, read_error) + result = _recover_from_checkpoint(run_dir, workspace, parseable, run_meta, read_error) + if result == 0: + _print_pending_dispatch_recovery(run_dir) + return result return _recover_legacy(run_dir, workspace, parseable, run_meta, read_error) diff --git a/tests/test_aboyeur.py b/tests/test_aboyeur.py index 56767266..c9d6322d 100644 --- a/tests/test_aboyeur.py +++ b/tests/test_aboyeur.py @@ -1158,6 +1158,48 @@ def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): assert run_meta["status"] == "ok" +def test_lifecycle_run_dispatch_emits_only_identified_worker_dispatch_facts(monkeypatch, tmp_path): + from brigade import run_journal, run_lifecycle + + monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") + monkeypatch.setattr( + aboyeur.agents, + "run_agent", + lambda *args, **kwargs: agents.AgentResult(text="worker final output", ok=True), + ) + output_dir = tmp_path / "run" + + with runguard.run_lock(tmp_path, run_dir=output_dir): + assert ( + aboyeur.run( + "do exactly this", + _roster(), + worker="coder", + cwd=tmp_path, + output_dir=output_dir, + route_enabled=False, + ) + == 0 + ) + + events = run_journal.read_journal(run_lifecycle._journal_path(output_dir)).events + worker_facts = [event for event in events if event.event_type.startswith("run.dispatch.")] + assert {event.event_type for event in worker_facts} == { + "run.dispatch.requested", + "run.dispatch.observed", + "run.dispatch.completed", + } + assert all( + isinstance(event.payload.get("seat"), str) + and event.payload["seat"] + and isinstance(event.payload.get("attempt"), int) + and not isinstance(event.payload["attempt"], bool) + and event.payload["attempt"] > 0 + for event in worker_facts + ) + assert any(event.event_type == "run.dispatching.started" for event in events) + + def test_run_direct_worker_failure_reports_and_records(monkeypatch, capsys, tmp_path): def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): return agents.AgentResult( @@ -1620,7 +1662,7 @@ def test_artifact_collection_failure_preserves_terminal_primary_receipt(tmp_path @pytest.mark.parametrize("escape", ["keyboard", "sigterm"]) -def test_post_dispatch_escape_uses_result_processing_phase_owner(monkeypatch, tmp_path, escape): +def test_post_dispatch_escape_clears_worker_phase_owner(monkeypatch, tmp_path, escape): output_dir = tmp_path / "run" monkeypatch.setattr( aboyeur, @@ -1637,7 +1679,7 @@ def test_post_dispatch_escape_uses_result_processing_phase_owner(monkeypatch, tm def interrupt_after_dispatch(*args, **kwargs): # noqa: ARG001 receipt = json.loads((output_dir / "run.json").read_text()) assert receipt["status"] == "result-processing" - assert receipt["phase_owner"] == "chef" + assert "phase_owner" not in receipt assert "active_stage" not in receipt assert "active_seats" not in receipt if escape == "sigterm": @@ -1662,7 +1704,7 @@ def interrupt_after_dispatch(*args, **kwargs): # noqa: ARG001 assert receipt["status"] == "canceled" assert receipt["failure"]["phase"] == "result-processing" assert receipt["failure"]["kind"] == expected_kind - assert receipt["failure"]["seat"] == "chef" + assert receipt["failure"]["seat"] == "coder" @pytest.mark.skipif(not hasattr(time, "tzset"), reason="requires POSIX timezone control") @@ -5595,16 +5637,17 @@ def capture_write_json(path, payload): monkeypatch.setattr(aboyeur, "_write_json", capture_write_json) - assert ( - aboyeur.run( - "build feature", - _roster(), - cwd=run_cwd, - output_dir=output_dir, - code_graph_enabled=False, + with runguard.run_lock(run_cwd, run_dir=output_dir): + assert ( + aboyeur.run( + "build feature", + _roster(), + cwd=run_cwd, + output_dir=output_dir, + code_graph_enabled=False, + ) + == 0 ) - == 0 - ) # record_run_start produces the first run.json write; the next one is the # first follow-up receipt rewrite performed by aboyeur.run itself. @@ -5622,10 +5665,10 @@ def capture_write_json(path, payload): assert final_meta[_LIFECYCLE_REQUEST_FIELD] is True assert final_meta[_AUTHORITY_REQUEST_FIELD] is True - # The run remains on the authority path: the durable authority request is - # still true and no projection metadata has been published yet, so - # _resolve_authority_state classifies it as authority-requested. - assert aboyeur._resolve_authority_state(output_dir) == "authority-requested" + # The run remains on the authority path. Under the real run lock, the + # lifecycle journal activates and the ready first comparison publishes + # projection metadata, so the completed run is authoritative. + assert aboyeur._resolve_authority_state(output_dir) == "authoritative" # -- Issue #568 slice 7 assignment 3: durable enrollment fail-closed ------------ diff --git a/tests/test_doctor.py b/tests/test_doctor.py index 25f622e8..96cab9f0 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -6,6 +6,7 @@ from pathlib import Path import json +import pytest from brigade import cli from brigade import doctor as doctor_mod @@ -1361,6 +1362,7 @@ def _activate_recovery_journal_with_checkpoint( run_json_obj: dict, *, paired_event_type: str | None = "run.planning.started", + pairing_key: str | None = None, leave_stale_lock: bool = False, ) -> bytes: import shutil @@ -1381,6 +1383,7 @@ def _activate_recovery_journal_with_checkpoint( checkpoint_bytes, workspace=workspace, paired_event_type=paired_event_type, + pairing_key=pairing_key, ) lock_path = workspace / ".brigade" / "run.lock" if leave_stale_lock: @@ -1421,6 +1424,80 @@ def _recovery_check(target: Path, *, full: bool = False) -> tuple[str, str, str] return doctor_mod._check_recovery_checkpoints(target, full=full) +def test_doctor_validates_checkpoint_before_reporting_pending_dispatch(tmp_path: Path): + from brigade import run_checkpoint, run_journal, run_lifecycle, runguard + + workspace = tmp_path / "workspace" + workspace.mkdir() + run_dir = workspace / ".brigade" / "runs" / "dispatch-pending" + run_meta = {"status": "dispatching", "task": "demo", "cwd": str(workspace)} + _activate_recovery_journal_with_checkpoint(workspace, run_dir, run_meta) + with runguard.run_lock(workspace, run_dir=run_dir): + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=workspace, + event_type="run.dispatch.requested", + seat="coder", + ) + + verdict, reason = doctor_mod._recovery_checkpoint_run_verdict(workspace, run_dir) + assert verdict == "warn" + assert "at-least-once dispatch recovery required" in reason + + (run_dir / "run.json").unlink() + verdict, reason = doctor_mod._recovery_checkpoint_run_verdict(workspace, run_dir) + assert verdict == "warn" + assert "run.json missing with valid checkpoint" in reason + assert "at-least-once dispatch recovery required" in reason + + (run_dir / "run.json").write_text("not json") + verdict, reason = doctor_mod._recovery_checkpoint_run_verdict(workspace, run_dir) + assert verdict == "warn" + assert "run.json unparseable with valid checkpoint" in reason + assert "at-least-once dispatch recovery required" in reason + + latest = run_checkpoint.latest_checkpoint_event( + run_journal.read_journal(run_lifecycle._journal_path(run_dir)).events + ) + assert latest is not None + checkpoint_path = run_checkpoint.checkpoint_path(run_dir, latest.payload["sha256"]) + checkpoint_path.write_bytes(b"x" * latest.payload["byte_size"]) + + verdict, reason = doctor_mod._recovery_checkpoint_run_verdict(workspace, run_dir) + assert verdict == "fail" + assert "at-least-once" not in reason + + +@pytest.mark.parametrize( + "event_type", + [ + "run.dispatch.requested", + "run.dispatch.observed", + "run.dispatch.completed", + "run.dispatch.failed", + ], +) +def test_doctor_fails_dispatch_pairing_checkpoint_at_tail_as_incomplete(tmp_path: Path, event_type: str): + from brigade import run_checkpoint + + workspace = tmp_path / "workspace" + workspace.mkdir() + run_dir = workspace / ".brigade" / "runs" / f"dispatch-incomplete-{event_type.rsplit('.', 1)[-1]}" + run_meta = {"status": "dispatching", "task": "demo", "cwd": str(workspace)} + _activate_recovery_journal_with_checkpoint( + workspace, + run_dir, + run_meta, + paired_event_type=event_type, + pairing_key=run_checkpoint.dispatch_pairing_key(event_type, "coder", 1), + ) + + verdict, reason = doctor_mod._recovery_checkpoint_run_verdict(workspace, run_dir) + + assert verdict == "fail" + assert "incomplete" in reason + + def _snapshot_run_tree(run_dir: Path) -> dict[str, object]: from brigade import run_checkpoint diff --git a/tests/test_run_checkpoint.py b/tests/test_run_checkpoint.py index 634e5d84..c0d1e4f9 100644 --- a/tests/test_run_checkpoint.py +++ b/tests/test_run_checkpoint.py @@ -84,7 +84,16 @@ def test_checkpoint_dir_and_path_helpers(): def test_checkpoint_event_type_registered_with_closed_payload_keys(): assert "run.snapshot.checkpointed" in run_events.EVENT_TYPES assert run_events.EVENT_TYPES["run.snapshot.checkpointed"] == frozenset( - {"path", "sha256", "media_type", "byte_size", "privacy_class", "paired_event_type", "body_kind"} + { + "path", + "sha256", + "media_type", + "byte_size", + "privacy_class", + "paired_event_type", + "body_kind", + "pairing_key", + } ) @@ -1580,6 +1589,7 @@ def _journal_with_checkpoint_and_trailing( run_json_obj: dict, *, paired_event_type: str | None, + pairing_key: str | None = None, trailing_events: list[tuple[str, dict, str, str]], ) -> run_journal.RunEvent: """Activate the journal, write one checkpoint, then append trailing events. @@ -1593,7 +1603,11 @@ def _journal_with_checkpoint_and_trailing( run_lifecycle.prepare_lifecycle_journal(run_dir, workspace=workspace) run_json_bytes = _writer_bytes(run_json_obj) checkpoint = run_checkpoint.write_checkpoint( - run_dir, run_json_bytes, workspace=workspace, paired_event_type=paired_event_type + run_dir, + run_json_bytes, + workspace=workspace, + paired_event_type=paired_event_type, + pairing_key=pairing_key, ) assert checkpoint is not None prev_seq = checkpoint.sequence @@ -1611,6 +1625,71 @@ def _journal_with_checkpoint_and_trailing( return checkpoint +@pytest.mark.parametrize( + ("payload", "pairing_seat", "pairing_attempt"), + [ + ({"seat": "other", "attempt": 1, "detail": "completed"}, "coder", 1), + ({"seat": "coder", "attempt": 2, "detail": "completed"}, "coder", 1), + ({"seat": "coder", "detail": "completed"}, "coder", 1), + ], + ids=("wrong-seat", "wrong-attempt", "missing-attempt"), +) +def test_recover_from_checkpoint_rejects_mismatched_dispatch_pairing_identity( + tmp_path, payload, pairing_seat, pairing_attempt +): + workspace = _workspace(tmp_path) + run_dir = _run_dir(tmp_path) + run_json_obj = {"schema": "brigade.run.v1", "status": "dispatching", "task": "demo"} + _journal_with_checkpoint_and_trailing( + workspace, + run_dir, + run_json_obj, + paired_event_type="run.dispatch.completed", + pairing_key=run_checkpoint.dispatch_pairing_key("run.dispatch.completed", pairing_seat, pairing_attempt), + trailing_events=[ + ("run.dispatch.completed", payload, "dispatch-completed-1", "2026-07-27T15:30:46.000000Z"), + ], + ) + (run_dir / "run.json").unlink() + + with pytest.raises(run_checkpoint.CheckpointError) as excinfo: + run_checkpoint.recover_from_checkpoint(run_dir, None) + + assert excinfo.value.category == "pairing" + assert not (run_dir / "run.json").exists() + + +@pytest.mark.parametrize( + "event_type", + [ + "run.dispatch.requested", + "run.dispatch.observed", + "run.dispatch.completed", + "run.dispatch.failed", + ], +) +def test_recover_rejects_dispatch_pairing_checkpoint_at_tail_as_incomplete(tmp_path, event_type): + workspace = _workspace(tmp_path) + run_dir = _run_dir(tmp_path) + run_json_obj = {"schema": "brigade.run.v1", "status": "dispatching", "task": "demo"} + _journal_with_checkpoint_and_trailing( + workspace, + run_dir, + run_json_obj, + paired_event_type=event_type, + pairing_key=run_checkpoint.dispatch_pairing_key(event_type, "coder", 1), + trailing_events=[], + ) + (run_dir / "run.json").unlink() + + with pytest.raises(run_checkpoint.CheckpointError) as excinfo: + run_checkpoint.recover_from_checkpoint(run_dir, None) + + assert excinfo.value.category == "incomplete-pair" + assert len(str(excinfo.value)) <= run_events.MAX_DIAGNOSTIC_LEN + assert not (run_dir / "run.json").exists() + + def test_recover_from_checkpoint_fails_on_wrong_paired_event_type(tmp_path): """N+1 event_type differs from the checkpoint's paired_event_type -> uncovered.""" workspace = _workspace(tmp_path) diff --git a/tests/test_run_events.py b/tests/test_run_events.py index 6da74861..b4ea3f82 100644 --- a/tests/test_run_events.py +++ b/tests/test_run_events.py @@ -245,7 +245,16 @@ def test_event_type_registry_includes_run_created_with_status_only(): def test_checkpoint_event_type_registered_with_closed_payload_keys(): assert "run.snapshot.checkpointed" in run_events.EVENT_TYPES assert run_events.EVENT_TYPES["run.snapshot.checkpointed"] == frozenset( - {"path", "sha256", "media_type", "byte_size", "privacy_class", "paired_event_type", "body_kind"} + { + "path", + "sha256", + "media_type", + "byte_size", + "privacy_class", + "paired_event_type", + "body_kind", + "pairing_key", + } ) diff --git a/tests/test_run_lifecycle.py b/tests/test_run_lifecycle.py index 65e7cb56..7086b91d 100644 --- a/tests/test_run_lifecycle.py +++ b/tests/test_run_lifecycle.py @@ -16,20 +16,25 @@ from __future__ import annotations import json +import signal import stat +import threading from pathlib import Path import pytest from brigade import ( aboyeur, + agents, localio, proc, run_checkpoint, run_events, run_journal, run_lifecycle, + run_projector, runguard, + run_transport, ) from brigade import roster as roster_mod @@ -425,9 +430,9 @@ def test_aba_recurrence_appends_all_three_occurrences(enabled, tmp_path): assert [e.event_type for e in _status_events(run_dir)] == [ "run.created", - "run.dispatch.requested", - "run.dispatch.completed", - "run.dispatch.requested", + "run.dispatching.started", + "run.result-processing.started", + "run.dispatching.started", ] # Checkpoints interleave before each status event, but a checkpoint # replays when the run.json bytes + paired_event_type repeat. The second @@ -438,16 +443,76 @@ def test_aba_recurrence_appends_all_three_occurrences(enabled, tmp_path): "run.snapshot.checkpointed", "run.created", "run.snapshot.checkpointed", - "run.dispatch.requested", + "run.dispatching.started", "run.snapshot.checkpointed", - "run.dispatch.completed", - "run.dispatch.requested", + "run.result-processing.started", + "run.dispatching.started", ] assert [e.sequence for e in events] == [1, 2, 3, 4, 5, 6, 7] assert events[3].previous_digest == events[2].event_digest assert events[6].previous_digest == events[5].event_digest +def test_dispatch_facts_pair_each_real_attempt_without_reusing_pending_identity(enabled, tmp_path): + repo = _repo(tmp_path) + run_dir = _run_dir(repo) + + _write_run_json(run_dir, "started", lock_workspace=repo) + _write_run_json_locked(repo, run_dir, "started", lock_workspace=repo) + _write_run_json_locked(repo, run_dir, "dispatching", lock_workspace=repo) + + with runguard.run_lock(repo, run_dir=run_dir): + first = run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.requested", + seat="coder", + ) + second = run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.requested", + seat="coder", + ) + assert first is not None + assert second is not None + assert first.payload["attempt"] == 1 + assert second.payload["attempt"] == 2 + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.observed", + seat="coder", + attempt=1, + ) + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.failed", + seat="coder", + attempt=1, + ) + + events = _events(run_dir) + facts = [event for event in events if event.event_type.startswith("run.dispatch.")] + assert [(event.event_type, event.payload["attempt"]) for event in facts[-4:]] == [ + ("run.dispatch.requested", 1), + ("run.dispatch.requested", 2), + ("run.dispatch.observed", 1), + ("run.dispatch.failed", 1), + ] + dispatch_checkpoints = [ + event + for event in events + if event.event_type == run_checkpoint.CHECKPOINT_EVENT_TYPE + and event.payload.get("paired_event_type", "").startswith("run.dispatch.") + and "pairing_key" in event.payload + ] + assert len(dispatch_checkpoints) == 4 + assert len({event.payload["pairing_key"] for event in dispatch_checkpoints}) == 4 + assert run_lifecycle.pending_dispatch_requests(events) == [("coder", 2)] + + def test_artifact_collection_intermediate_appends_the_second_a(enabled, tmp_path): repo = _repo(tmp_path) run_dir = _run_dir(repo) @@ -468,9 +533,9 @@ def test_artifact_collection_intermediate_appends_the_second_a(enabled, tmp_path assert meta["status"] == "dispatching" assert [e.event_type for e in _status_events(run_dir)] == [ "run.created", - "run.dispatch.requested", + "run.dispatching.started", "run.artifact_collection.started", - "run.dispatch.requested", + "run.dispatching.started", ] events = _events(run_dir) # The second "dispatching" produces identical run.json bytes to the first, @@ -479,13 +544,13 @@ def test_artifact_collection_intermediate_appends_the_second_a(enabled, tmp_path "run.snapshot.checkpointed", "run.created", "run.snapshot.checkpointed", - "run.dispatch.requested", + "run.dispatching.started", "run.snapshot.checkpointed", "run.artifact_collection.started", - "run.dispatch.requested", + "run.dispatching.started", ] assert [e.sequence for e in events] == [1, 2, 3, 4, 5, 6, 7] - # The second run.dispatch.requested links to the artifact-collection + # The second run.dispatching.started links to the artifact-collection # status event immediately before it (its checkpoint replayed, so no # checkpoint sits between them). assert events[6].previous_digest == events[5].event_digest @@ -518,7 +583,7 @@ def test_retry_after_interruption_reuses_committed_event(enabled, tmp_path): assert [e.event_type for e in events] == [ "run.snapshot.checkpointed", "run.created", - "run.dispatch.completed", + "run.result-processing.started", ] assert replay.event_id == events[2].event_id assert _journal_path(run_dir).read_bytes() == journal_before @@ -1050,7 +1115,7 @@ def test_first_write_match_authorizes_first_projected_snapshot(tmp_path, monkeyp _write_run_json_authority(run_dir, "started") _write_run_json_authority(run_dir, "planning") meta = json.loads((run_dir / "run.json").read_text()) - assert meta["projector_version"] == 3 + assert meta["projector_version"] == run_projector.PROJECTOR_VERSION assert meta["journal_present"] is True assert meta["status"] == "planning" @@ -1122,7 +1187,7 @@ def _enroll_and_authorize(repo, run_dir, monkeypatch): return json.loads((run_dir / "run.json").read_text()) -def test_journal_ahead_remains_authoritative(tmp_path, monkeypatch): +def test_unpaired_journal_ahead_fails_closed_before_status_append(tmp_path, monkeypatch): repo = _repo(tmp_path) run_dir = _run_dir(repo) _enroll_and_authorize(repo, run_dir, monkeypatch) @@ -1137,12 +1202,13 @@ def test_journal_ahead_remains_authoritative(tmp_path, monkeypatch): expected_previous_sequence=tail, recorded_at="2026-07-27T15:31:46.000000Z", ) + run_json_before = (run_dir / "run.json").read_bytes() + journal_before = _journal_path(run_dir).read_bytes() with runguard.run_lock(repo, run_dir=run_dir): - _write_run_json_authority(run_dir, "ok") - meta = json.loads((run_dir / "run.json").read_text()) - assert meta["projector_version"] == 3 - assert meta["status"] == "ok" - assert meta["journal_present"] is True + with pytest.raises(run_lifecycle.LifecycleJournalError, match="journal-ahead"): + _write_run_json_authority(run_dir, "ok") + assert (run_dir / "run.json").read_bytes() == run_json_before + assert _journal_path(run_dir).read_bytes() == journal_before def test_current_version_mismatch_fail_closed(tmp_path, monkeypatch): @@ -1458,7 +1524,7 @@ def test_first_authority_started_write_projects_on_first_write(tmp_path, monkeyp with runguard.run_lock(repo, run_dir=run_dir): _write_run_json_authority(run_dir, "started") meta = json.loads((run_dir / "run.json").read_text()) - assert meta["projector_version"] == 3 + assert meta["projector_version"] == run_projector.PROJECTOR_VERSION assert meta["journal_present"] is True events = _events(run_dir) assert [e.event_type for e in events] == [ @@ -1551,16 +1617,23 @@ def two_phase(run_dir): assert (run_dir / "run.json").read_bytes() == run_json_before -def test_authoritative_prior_journal_ahead_allows_projection_despite_parity_gap(tmp_path, monkeypatch): - # Finding 4: for the approved prior journal-ahead exception, allow - # projection despite the current parity gap. A direct append advances the - # journal tail past the last recorded comparison; the prior gate approves - # the genuine journal-ahead, and this write's parity records a comparison - # gap (post-parity not ready) but the write still projects. +def test_authoritative_prior_status_changing_pair_catch_up_fails_closed(tmp_path, monkeypatch): + # A structurally covered checkpoint/status pair is still unsafe to catch + # up against the persisted pre-write snapshot when the event changes the + # projected status. The catch-up records a mismatch and fails before the + # next status pair is appended. repo = _repo(tmp_path) run_dir = _run_dir(repo) _enroll_and_authorize(repo, run_dir, monkeypatch) with runguard.run_lock(repo, run_dir=run_dir): + snapshot = (run_dir / "run.json").read_bytes() + run_checkpoint.write_checkpoint( + run_dir, + snapshot, + workspace=repo, + paired_event_type="run.completed", + body_kind=run_checkpoint._BODY_KIND_BASE_STRIPPED, + ) tail = run_journal.read_journal(_journal_path(run_dir)).events[-1].sequence run_journal.append_event( _journal_path(run_dir), @@ -1571,12 +1644,13 @@ def test_authoritative_prior_journal_ahead_allows_projection_despite_parity_gap( expected_previous_sequence=tail, recorded_at="2026-07-27T15:31:46.000000Z", ) + run_json_before = (run_dir / "run.json").read_bytes() + journal_before = _journal_path(run_dir).read_bytes() with runguard.run_lock(repo, run_dir=run_dir): - _write_run_json_authority(run_dir, "ok") - meta = json.loads((run_dir / "run.json").read_text()) - assert meta["projector_version"] == 3 - assert meta["journal_present"] is True - assert meta["status"] == "ok" + with pytest.raises(run_lifecycle.LifecycleJournalError, match="catch-up not ready"): + _write_run_json_authority(run_dir, "ok") + assert (run_dir / "run.json").read_bytes() == run_json_before + assert _journal_path(run_dir).read_bytes() == journal_before # -- Issue #568 slice 6, Task 7 final corrections: findings 1-3 regressions ----- @@ -1666,27 +1740,22 @@ def test_authority_fail_closed_on_mixed_run_id_chain_valid_prefix(tmp_path, monk assert journal.read_bytes() == journal_before -def test_authoritative_prior_journal_ahead_with_current_mismatch_fails_closed(tmp_path, monkeypatch): - # Finding 3: the approved prior journal-ahead exception must not ignore an - # arbitrary post-parity mismatch. Prior journal-ahead is genuine (the - # journal tail advanced past the last recorded comparison), but the - # current parity is forced to record a mismatch instead of a match. The - # gate must fail closed (raise, no run.json replace) rather than project. +def test_authoritative_pair_catch_up_with_current_mismatch_fails_closed(tmp_path, monkeypatch): + # A valid one-pair lag is eligible for catch-up, but the catch-up must not + # ignore an arbitrary parity mismatch. repo = _repo(tmp_path) run_dir = _run_dir(repo) _enroll_and_authorize(repo, run_dir, monkeypatch) with runguard.run_lock(repo, run_dir=run_dir): - tail = run_journal.read_journal(_journal_path(run_dir)).events[-1].sequence - run_journal.append_event( - _journal_path(run_dir), - run_id=run_dir.name, - event_type="run.completed", - payload={"status": "ok", "detail": "ok"}, - idempotency_key="complete-1", - expected_previous_sequence=tail, - recorded_at="2026-07-27T15:31:46.000000Z", + _append_dispatch_pair_without_parity( + run_dir, + repo, + event_type="run.dispatch.completed", + seat="coder", + attempt=1, ) run_json_before = (run_dir / "run.json").read_bytes() + journal_before = _journal_path(run_dir).read_bytes() real_record_shadow = run_shadow.record_shadow_comparison def force_mismatch(run_dir_arg, legacy_snapshot): @@ -1706,11 +1775,9 @@ def force_mismatch(run_dir_arg, legacy_snapshot): monkeypatch.setattr(run_shadow, "record_shadow_comparison", force_mismatch) with runguard.run_lock(repo, run_dir=run_dir): with pytest.raises(run_lifecycle.LifecycleJournalError): - _write_run_json_authority(run_dir, "ok") - # Finding 3: the validation runs AFTER the checkpoint/lifecycle append - # and parity, so the journal advances; the contract is "no run.json - # replace" (the prior gate already passed, so the append is expected). + _write_run_json_authority(run_dir, "result-processing") assert (run_dir / "run.json").read_bytes() == run_json_before + assert _journal_path(run_dir).read_bytes() == journal_before # -- Task 7 review correction: post-parity reasons / strict counters / aggregate @@ -1719,9 +1786,9 @@ def force_mismatch(run_dir_arg, legacy_snapshot): def _forge_after_parity(run_dir_arg, legacy_snapshot, *, forge): """Run the real shadow comparison, then forge the artifact per ``forge``. - Used by the journal-ahead correction regressions: the genuine parity pass - produces the approved comparison-gap-then-match record pair, and the forge - layer mutates only the aggregate fields the validator must reject. + Used by the one-pair catch-up regressions. The real comparison catches up + the eligible dispatch pair, then the forge layer mutates only the aggregate + fields the post-catch-up readiness check must reject. """ real_record_shadow = run_shadow.record_shadow_comparison @@ -1736,31 +1803,21 @@ def _wrapper(run_dir_inner, legacy_inner): def _enroll_authorize_and_advance(repo, run_dir, monkeypatch): - """Enroll, authorize, then append a direct ``run.completed`` event so the - journal tail is genuinely ahead of the last recorded comparison.""" + """Enroll, authorize, then append one dispatch pair without parity.""" _enroll_and_authorize(repo, run_dir, monkeypatch) with runguard.run_lock(repo, run_dir=run_dir): - tail = run_journal.read_journal(_journal_path(run_dir)).events[-1].sequence - run_journal.append_event( - _journal_path(run_dir), - run_id=run_dir.name, - event_type="run.completed", - payload={"status": "ok", "detail": "ok"}, - idempotency_key="complete-ahead", - expected_previous_sequence=tail, - recorded_at="2026-07-27T15:31:46.000000Z", + _append_dispatch_pair_without_parity( + run_dir, + repo, + event_type="run.dispatch.completed", + seat="coder", + attempt=1, ) return (run_dir / "run.json").read_bytes() -def test_journal_ahead_override_rejects_forged_extra_journal_unreadable_reason(tmp_path, monkeypatch): - # Correction 1: the journal-ahead override must accept the post-parity - # readiness report ONLY when its reasons are exactly - # (REASON_ERROR_RECORDED,). A forged aggregate ``last_error_category`` of - # "journal-unreadable" makes the post-parity report carry - # (REASON_ERROR_RECORDED, REASON_JOURNAL_UNREADABLE); even though the - # recent_records pair still looks like comparison-gap then match, the gate - # must fail closed with no run.json replace. +def test_pair_catch_up_rejects_forged_extra_journal_unreadable_reason(tmp_path, monkeypatch): + # A forged aggregate error after catch-up must fail closed. repo = _repo(tmp_path) run_dir = _run_dir(repo) run_json_before = _enroll_authorize_and_advance(repo, run_dir, monkeypatch) @@ -1779,7 +1836,7 @@ def forge(data): assert (run_dir / "run.json").read_bytes() == run_json_before -def test_journal_ahead_override_rejects_bool_mismatches_counter(tmp_path, monkeypatch): +def test_pair_catch_up_rejects_bool_mismatches_counter(tmp_path, monkeypatch): # Correction 2: ``mismatches`` must be a non-bool int equal to 0. A bool # ``False`` satisfies ``False == 0`` and must NOT be accepted. repo = _repo(tmp_path) @@ -1800,7 +1857,7 @@ def forge(data): assert (run_dir / "run.json").read_bytes() == run_json_before -def test_journal_ahead_override_rejects_bool_errors_counter(tmp_path, monkeypatch): +def test_pair_catch_up_rejects_bool_errors_counter(tmp_path, monkeypatch): # Correction 2: ``errors`` must be a non-bool int equal to 1. A bool # ``True`` satisfies ``True == 1`` and must NOT be accepted. repo = _repo(tmp_path) @@ -1821,7 +1878,7 @@ def forge(data): assert (run_dir / "run.json").read_bytes() == run_json_before -def test_journal_ahead_override_rejects_non_none_last_error_category(tmp_path, monkeypatch): +def test_pair_catch_up_rejects_non_none_last_error_category(tmp_path, monkeypatch): # Correction 3: the final match aggregate's ``last_error_category`` must # be None. A forger sets it to "comparison-gap" (which is NOT # "journal-unreadable", so the post-parity reasons stay exactly @@ -1845,3 +1902,431 @@ def forge(data): with pytest.raises(run_lifecycle.LifecycleJournalError): _write_run_json_authority(run_dir, "ok") assert (run_dir / "run.json").read_bytes() == run_json_before + + +@pytest.mark.parametrize("sabotage", ["remove", "rename", "symlink"]) +def test_dispatch_fact_fails_closed_when_enrolled_journal_is_not_regular( + tmp_path, + monkeypatch, + sabotage, +): + repo = _repo(tmp_path) + run_dir = _run_dir(repo) + monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") + _write_run_json(run_dir, "started", lock_workspace=repo) + _write_run_json_locked(repo, run_dir, "dispatching", lock_workspace=repo) + journal = _journal_path(run_dir) + displaced = journal.with_name("lifecycle.displaced") + if sabotage == "remove": + journal.unlink() + elif sabotage == "rename": + journal.rename(displaced) + else: + journal.rename(displaced) + journal.symlink_to(displaced.name) + + with runguard.run_lock(repo, run_dir=run_dir): + with pytest.raises(run_lifecycle.LifecycleJournalError, match="journal"): + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.requested", + seat="coder", + ) + + +def test_transport_does_not_invoke_after_enrolled_journal_disappears(tmp_path, monkeypatch): + repo = _repo(tmp_path) + run_dir = _run_dir(repo) + monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") + _write_run_json(run_dir, "started", lock_workspace=repo) + _write_run_json_locked(repo, run_dir, "dispatching", lock_workspace=repo) + _journal_path(run_dir).unlink() + external_calls: list[str] = [] + + def fake_run_agent(cli_ref, prompt, **kwargs): # noqa: ARG001 + external_calls.append(cli_ref) + return agents.AgentResult(text="must not run", ok=True) + + monkeypatch.setattr(agents, "run_agent", fake_run_agent) + roster = roster_mod.Roster( + orchestrator="chef", + agents={ + "chef": roster_mod.Agent("chef", "codex", "plan"), + "coder": roster_mod.Agent("coder", "codex", "code"), + }, + max_workers=1, + ) + + def requested(agent): + try: + return run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.requested", + seat=agent.name, + ) + except run_lifecycle.LifecycleJournalError as exc: + raise runguard.RetainRunLockError("dispatch fact unavailable") from exc + + with runguard.run_lock(repo, run_dir=run_dir): + with pytest.raises(runguard.RetainRunLockError, match="dispatch fact unavailable"): + run_transport.dispatch( + [run_transport.Assignment(worker="coder", task="do work")], + roster, + build_prompt=lambda agent, assignment, **kwargs: assignment.task, + run_appserver_worker=lambda *args, **kwargs: agents.AgentResult(text="", ok=False), + event_writer=lambda *args, **kwargs: None, + cwd=repo, + on_dispatch_requested=requested, + ) + + assert external_calls == [] + + +@pytest.mark.parametrize( + ("failure_kind", "detail"), + [ + ("keyboard-interrupt", "run canceled by user"), + ("signal-15", "run terminated by SIGTERM"), + ], +) +def test_dispatch_pair_is_adjacent_when_interrupt_writer_crosses_barrier( + tmp_path, + monkeypatch, + failure_kind, + detail, +): + repo = _repo(tmp_path) + run_dir = _run_dir(repo) + monkeypatch.setenv("BRIGADE_LIFECYCLE_JOURNAL", "1") + aboyeur.record_run_start( + run_dir, + task="barrier run", + cwd=repo, + roster=_minimal_roster(), + read_only=False, + lock_workspace=repo, + ) + with runguard.run_lock(repo, run_dir=run_dir): + _write_run_json(run_dir, "dispatching", lock_workspace=repo) + + checkpoint_written = threading.Event() + release_worker = threading.Event() + interrupt_done = threading.Event() + errors: list[BaseException] = [] + real_write_checkpoint = run_checkpoint.write_checkpoint + + def checkpoint_barrier(*args, **kwargs): + event = real_write_checkpoint(*args, **kwargs) + if kwargs.get("pairing_key") is not None: + checkpoint_written.set() + assert release_worker.wait(timeout=5) + return event + + monkeypatch.setattr(run_checkpoint, "write_checkpoint", checkpoint_barrier) + + def worker_writer(): + try: + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.requested", + seat="coder", + ) + except BaseException as exc: + errors.append(exc) + + def interrupt_writer(): + try: + aboyeur.record_run_termination( + run_dir, + status="canceled", + failure_phase="dispatch", + failure_kind=failure_kind, + detail=detail, + seat="coder", + ) + except BaseException as exc: + errors.append(exc) + finally: + interrupt_done.set() + + with runguard.run_lock(repo, run_dir=run_dir): + worker = threading.Thread(target=worker_writer) + worker.start() + assert checkpoint_written.wait(timeout=5) + interrupt = threading.Thread(target=interrupt_writer) + interrupt.start() + assert not interrupt_done.wait(timeout=0.1) + release_worker.set() + worker.join(timeout=5) + interrupt.join(timeout=5) + + assert not worker.is_alive() + assert not interrupt.is_alive() + assert errors == [] + events = _events(run_dir) + dispatch_index = next( + index + for index, event in enumerate(events) + if event.event_type == run_checkpoint.CHECKPOINT_EVENT_TYPE and event.payload.get("pairing_key") + ) + assert events[dispatch_index + 1].event_type == "run.dispatch.requested" + assert events[dispatch_index + 1].payload["seat"] == "coder" + assert events[dispatch_index + 1].payload["attempt"] == 1 + (run_dir / "run.json").unlink() + repaired = run_checkpoint.recover_from_checkpoint(run_dir, None) + assert repaired["status"] == "canceled" + + +@pytest.mark.skipif(not hasattr(signal, "pthread_sigmask"), reason="pthread signal masks unavailable") +def test_checkpoint_event_pair_restores_exact_signal_mask_and_rejects_reentry(monkeypatch): + previous_mask = {signal.SIGINT} + calls: list[tuple[int, set[signal.Signals]]] = [] + + def fake_sigmask(operation, mask): + calls.append((operation, set(mask))) + return previous_mask + + monkeypatch.setattr(signal, "pthread_sigmask", fake_sigmask) + with run_lifecycle.checkpoint_event_pair(): + with pytest.raises(run_lifecycle.LifecycleJournalError, match="reentry"): + with run_lifecycle.checkpoint_event_pair(): + pass + + assert calls == [ + (signal.SIG_BLOCK, {signal.SIGTERM}), + (signal.SIG_SETMASK, previous_mask), + ] + + +def test_authoritative_dispatch_pairs_keep_shadow_ready_through_synthesis(tmp_path, monkeypatch): + repo = _repo(tmp_path) + run_dir = _run_dir(repo) + _enroll_and_authorize(repo, run_dir, monkeypatch) + + with runguard.run_lock(repo, run_dir=run_dir): + requested = run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.requested", + seat="coder", + ) + assert requested is not None + attempt = requested.payload["attempt"] + assert isinstance(attempt, int) + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.observed", + seat="coder", + attempt=attempt, + ) + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.completed", + seat="coder", + attempt=attempt, + ) + _write_run_json_authority(run_dir, "result-processing") + _write_run_json_authority(run_dir, "synthesizing") + + shadow = json.loads(run_shadow.shadow_artifact_path(run_dir).read_text()) + assert shadow["errors"] == 0 + assert shadow["mismatches"] == 0 + assert shadow["last_outcome"] == run_shadow.OUTCOME_MATCH + readiness = run_shadow.check_projection_readiness(run_dir) + assert readiness.ready is True + assert readiness.reasons == () + + +def test_authoritative_dispatch_recovers_one_pair_parity_crash_before_next_pair(tmp_path, monkeypatch): + repo = _repo(tmp_path) + run_dir = _run_dir(repo) + _enroll_and_authorize(repo, run_dir, monkeypatch) + real_shadow = run_shadow.record_shadow_comparison + + with runguard.run_lock(repo, run_dir=run_dir): + monkeypatch.setattr(run_shadow, "record_shadow_comparison", lambda *args, **kwargs: None) + requested = run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.requested", + seat="coder", + ) + assert requested is not None + attempt = requested.payload["attempt"] + assert isinstance(attempt, int) + monkeypatch.setattr(run_shadow, "record_shadow_comparison", real_shadow) + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.observed", + seat="coder", + attempt=attempt, + ) + + shadow = json.loads(run_shadow.shadow_artifact_path(run_dir).read_text()) + assert shadow["errors"] == 0 + assert shadow["mismatches"] == 0 + assert shadow["last_outcome"] == run_shadow.OUTCOME_MATCH + assert run_shadow.check_projection_readiness(run_dir).ready is True + + +@pytest.mark.parametrize( + "terminal_event_type", + ["run.dispatch.completed", "run.dispatch.failed"], +) +def test_authoritative_status_write_catches_up_final_dispatch_pair_before_append( + tmp_path, + monkeypatch, + terminal_event_type, +): + repo = _repo(tmp_path) + run_dir = _run_dir(repo) + _enroll_and_authorize(repo, run_dir, monkeypatch) + real_shadow = run_shadow.record_shadow_comparison + + with runguard.run_lock(repo, run_dir=run_dir): + requested = run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.requested", + seat="coder", + ) + assert requested is not None + attempt = requested.payload["attempt"] + assert isinstance(attempt, int) + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.observed", + seat="coder", + attempt=attempt, + ) + monkeypatch.setattr(run_shadow, "record_shadow_comparison", lambda *args, **kwargs: None) + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type=terminal_event_type, + seat="coder", + attempt=attempt, + ) + monkeypatch.setattr(run_shadow, "record_shadow_comparison", real_shadow) + + _write_run_json_authority(run_dir, "result-processing") + _write_run_json_authority(run_dir, "synthesizing") + + events = _events(run_dir) + tail = events[-1] + shadow = json.loads(run_shadow.shadow_artifact_path(run_dir).read_text()) + receipt = json.loads((run_dir / "run.json").read_text()) + assert shadow["errors"] == 0 + assert shadow["mismatches"] == 0 + assert shadow["last_outcome"] == run_shadow.OUTCOME_MATCH + assert shadow["last_compared_sequence"] == tail.sequence + assert shadow["last_compared_event_digest"] == tail.event_digest + assert receipt["journal_last_sequence"] == tail.sequence + assert receipt["journal_last_event_digest"] == tail.event_digest + assert run_shadow.check_projection_readiness(run_dir).ready is True + + +def _append_dispatch_pair_without_parity( + run_dir: Path, + repo: Path, + *, + event_type: str, + seat: str, + attempt: int, + include_pairing_key: bool = True, +) -> None: + snapshot = (run_dir / "run.json").read_bytes() + pairing_key = run_checkpoint.dispatch_pairing_key(event_type, seat, attempt) + run_checkpoint.write_checkpoint( + run_dir, + snapshot, + workspace=repo, + paired_event_type=event_type, + body_kind=run_checkpoint._BODY_KIND_BASE_STRIPPED, + pairing_key=pairing_key if include_pairing_key else None, + ) + report = run_journal.read_journal(_journal_path(run_dir)) + run_journal.append_event( + _journal_path(run_dir), + run_id=run_dir.name, + event_type=event_type, + payload={"seat": seat, "attempt": attempt, "detail": event_type.rsplit(".", 1)[-1]}, + idempotency_key=f"test-uncompared:{event_type}:{seat}:{attempt}", + expected_previous_sequence=report.events[-1].sequence, + ) + + +@pytest.mark.parametrize("gap_kind", ["two-pair", "forged-cursor", "missing-pairing-key"]) +def test_authoritative_status_write_rejects_uncovered_or_forged_dispatch_gap( + tmp_path, + monkeypatch, + gap_kind, +): + repo = _repo(tmp_path) + run_dir = _run_dir(repo) + _enroll_and_authorize(repo, run_dir, monkeypatch) + + with runguard.run_lock(repo, run_dir=run_dir): + _append_dispatch_pair_without_parity( + run_dir, + repo, + event_type="run.dispatch.requested", + seat="coder", + attempt=1, + include_pairing_key=gap_kind != "missing-pairing-key", + ) + if gap_kind == "two-pair": + _append_dispatch_pair_without_parity( + run_dir, + repo, + event_type="run.dispatch.observed", + seat="coder", + attempt=1, + ) + elif gap_kind == "forged-cursor": + artifact_path = run_shadow.shadow_artifact_path(run_dir) + artifact = json.loads(artifact_path.read_text()) + artifact["last_compared_event_digest"] = "f" * 64 + artifact_path.write_text(json.dumps(artifact, indent=2, sort_keys=True) + "\n") + + run_json_before = (run_dir / "run.json").read_bytes() + journal_before = _journal_path(run_dir).read_bytes() + with pytest.raises(run_lifecycle.LifecycleJournalError): + _write_run_json_authority(run_dir, "result-processing") + + assert (run_dir / "run.json").read_bytes() == run_json_before + assert _journal_path(run_dir).read_bytes() == journal_before + + +def test_authority_dispatch_checkpoint_is_base_stripped_and_recovers_exact_tail(tmp_path, monkeypatch): + repo = _repo(tmp_path) + run_dir = _run_dir(repo) + _enroll_and_authorize(repo, run_dir, monkeypatch) + + with runguard.run_lock(repo, run_dir=run_dir): + requested = run_lifecycle.record_dispatch_fact( + run_dir, + workspace=repo, + event_type="run.dispatch.requested", + seat="coder", + ) + assert requested is not None + events = _events(run_dir) + dispatch_checkpoint = events[-2] + assert dispatch_checkpoint.event_type == run_checkpoint.CHECKPOINT_EVENT_TYPE + assert dispatch_checkpoint.payload["body_kind"] == "base-stripped" + assert dispatch_checkpoint.payload["pairing_key"] + + (run_dir / "run.json").unlink() + repaired = run_checkpoint.recover_from_checkpoint(run_dir, None) + events = _events(run_dir) + assert repaired["journal_last_sequence"] == events[-1].sequence + assert repaired["journal_last_event_digest"] == events[-1].event_digest + assert repaired["projector_version"] == run_projector.PROJECTOR_VERSION diff --git a/tests/test_run_projector.py b/tests/test_run_projector.py index e47996c7..bdbc2f54 100644 --- a/tests/test_run_projector.py +++ b/tests/test_run_projector.py @@ -432,8 +432,12 @@ def test_full_field_fixture_preserves_deep_equality_and_copies_nested_values(): } assert EVENT_STATUS == { "run.planning.started": "planning", + "run.dispatching.started": "dispatching", "run.dispatch.requested": "dispatching", - "run.dispatch.completed": "result-processing", + "run.dispatch.observed": "dispatching", + "run.dispatch.completed": "dispatching", + "run.dispatch.failed": "dispatching", + "run.result-processing.started": "result-processing", "run.synthesis.started": "synthesizing", "run.synthesis.completed": "handoff", "run.artifact_collection.started": "artifact-collection", @@ -449,6 +453,21 @@ def test_full_field_fixture_preserves_deep_equality_and_copies_nested_values(): assert set(projection.snapshot.keys()) <= OWNED_FIELDS +def test_identified_dispatch_fact_does_not_advance_aggregate_status(): + completed = _build_event( + 1, + "run.dispatch.completed", + {"seat": "coder", "attempt": 1, "detail": "completed"}, + "dispatch-worker-1", + RECORDED_AT, + None, + ) + + projection = project_run_snapshot(_minimal_base_snapshot("result-processing"), [completed], journal_present=True) + + assert projection.snapshot["status"] == "result-processing" + + def test_reprojection_is_byte_idempotent(): base = _minimal_base_snapshot() events = _golden_events() @@ -539,10 +558,10 @@ def test_checkpoint_after_unmapped_status_preserves_last_mapped_status(): assert projection.last_event_digest == unmapped_checkpoint["event_digest"] -def test_projector_version_is_three(): +def test_projector_version_is_four(): projection = project_run_snapshot(_minimal_base_snapshot(), [], journal_present=False) - assert PROJECTOR_VERSION == 3 - assert projection.snapshot["projector_version"] == 3 + assert PROJECTOR_VERSION == 4 + assert projection.snapshot["projector_version"] == 4 def _events_ending_with_completed(*, status: str | None) -> list[dict]: diff --git a/tests/test_run_shadow.py b/tests/test_run_shadow.py index 5931113e..abf4ba92 100644 --- a/tests/test_run_shadow.py +++ b/tests/test_run_shadow.py @@ -1113,7 +1113,7 @@ def test_version_one_shadow_artifact_is_stale(enabled, tmp_path): # Quarantine: the next shadow comparison atomically renames the stale # artifact to a private .stale-projector-v2- sibling under - # events before a fresh current-v3 artifact is written. + # events before a fresh current-version artifact is written. _write_run_json_locked(repo, run_dir, "planning") quarantined = list((run_dir / "events").glob(".stale-projector-v2-*")) @@ -1133,6 +1133,31 @@ def test_version_one_shadow_artifact_is_stale(enabled, tmp_path): assert fresh["last_compared_sequence"] != stale_seq +def test_dispatch_semantics_v4_quarantines_old_v3_shadow_artifact(enabled, tmp_path): + repo = _repo(tmp_path) + run_dir = _run_dir(repo) + assert run_projector.PROJECTOR_VERSION == 4 + + _write_run_json(run_dir, "started") + _write_run_json_locked(repo, run_dir, "started") + artifact = run_shadow.shadow_artifact_path(run_dir) + stale = json.loads(artifact.read_text()) + stale["projector_version"] = 3 + artifact.write_text(json.dumps(stale, indent=2, sort_keys=True) + "\n") + stale_bytes = artifact.read_bytes() + + report = run_shadow.check_projection_readiness(run_dir) + assert report.reasons == (REASON_EVIDENCE_PROJECTOR_VERSION_STALE,) + _write_run_json_locked(repo, run_dir, "planning") + + quarantined = list((run_dir / "events").glob(".stale-projector-v2-*")) + assert quarantined + assert quarantined[0].read_bytes() == stale_bytes + fresh = json.loads(artifact.read_text()) + assert fresh["projector_version"] == 4 + assert fresh["errors"] == 0 + + def test_current_projector_version_artifact_with_checkpoint_tail_reads_ready_when_bytes_match(enabled, tmp_path): repo = _repo(tmp_path) run_dir = _run_dir(repo) @@ -1234,7 +1259,7 @@ def test_stale_quarantine_writes_fresh_v3_counters(enabled, tmp_path): _make_stale_artifact(run_dir, stale_version=2) stale_bytes = artifact.read_bytes() - _write_run_json_locked(repo, run_dir, "dispatching") # triggers quarantine + fresh v3 + _write_run_json_locked(repo, run_dir, "dispatching") # triggers quarantine + fresh artifact quarantined = list((run_dir / "events").glob(".stale-projector-v2-*")) assert quarantined @@ -1264,7 +1289,7 @@ def test_crash_window_after_quarantine_reads_as_no_evidence(enabled, tmp_path): artifact = run_shadow.shadow_artifact_path(run_dir) # Simulate the crash window: quarantine the stale artifact (atomic rename - # to the private sibling) but do NOT write a fresh current-v3 artifact. + # to the private sibling) but do NOT write a fresh current-version artifact. stamp = "20260730T000000000000Z" artifact.replace(artifact.with_name(f".stale-projector-v2-{stamp}")) assert not artifact.exists() @@ -1336,7 +1361,7 @@ def test_stale_quarantine_larger_advance_records_comparison_gap(enabled, tmp_pat # Later locked write (result-processing) runs the hook: tail seq 7 vs the # reused stale baseline 4 is an unexplained advance (7 != 4+2), so a - # fresh comparison-gap error is recorded on the fresh v3 artifact. + # fresh comparison-gap error is recorded on the fresh current-version artifact. _write_run_json_locked(repo, run_dir, "result-processing") fresh = json.loads(run_shadow.shadow_artifact_path(run_dir).read_text()) @@ -1354,7 +1379,7 @@ def test_current_v3_mismatch_is_never_reset_by_later_match(enabled, tmp_path): _write_run_json(run_dir, "started") _write_run_json_locked(repo, run_dir, "started") # match, seq 2 - # Record a mismatch against the current v3 artifact (forged status + # Record a mismatch against the current-version artifact (forged status # divergence with a mapped legacy status) at the same journal tail. run_shadow.record_shadow_comparison( run_dir, @@ -1371,7 +1396,7 @@ def test_current_v3_mismatch_is_never_reset_by_later_match(enabled, tmp_path): # A subsequent matching comparison (normal checkpoint+status advance from # seq 2 to seq 4, no gap) must NOT reset the recorded mismatch: counters on - # a current-v3 artifact only accumulate, never zero out. + # a current-version artifact only accumulate, never zero out. _write_run_json_locked(repo, run_dir, "planning") # match, seq 4 data = json.loads(run_shadow.shadow_artifact_path(run_dir).read_text()) @@ -1387,7 +1412,7 @@ def test_current_v3_error_is_never_reset_by_later_match(enabled, tmp_path): _write_run_json(run_dir, "started") _write_run_json_locked(repo, run_dir, "started") # match, seq 2 - # Record a journal-unreadable error against the current v3 artifact via a + # Record a journal-unreadable error against the current-version artifact via a # direct call (a locked write would fail inside write_checkpoint before # the shadow hook runs, so the error path is exercised directly). The # journal tail stays at seq 2 throughout so the later clean comparison @@ -1434,7 +1459,7 @@ def test_bounded_journal_reads_map_bound_failure_to_journal_unreadable(enabled, repo = _repo(tmp_path) run_dir = _run_dir(repo) _write_run_json(run_dir, "started") - _write_run_json_locked(repo, run_dir, "started") # clean current-v3 artifact + _write_run_json_locked(repo, run_dir, "started") # clean current-version artifact # A bound-exceeded journal read must map to journal-unreadable on both the # comparison path and the readiness path. read_journal_bounded is the @@ -1913,7 +1938,7 @@ def test_gate_accepts_zero_mismatches_and_zero_errors_as_valid(tmp_path): def test_gate_rejects_malformed_recent_records_as_evidence_unreadable(enabled, tmp_path, malformed_recent_records): """Readiness must reject malformed records before the tail check. - The artifact starts as a real current-v3 match with its cursor equal to + The artifact starts as a real current-version match with its cursor equal to the journal tail. Only ``recent_records`` is forged, so the expected result cannot be attributed to an earlier readiness gate. """ @@ -1936,7 +1961,7 @@ def test_gate_rejects_malformed_recent_records_as_evidence_unreadable(enabled, t def test_record_outcome_does_not_trust_invalid_counters_on_carry(enabled, tmp_path): - """Blocker #2 carry side: a current-v3 prior artifact with invalid counters + """Blocker #2 carry side: a current-version prior artifact with invalid counters (bool / negative / non-int) must not be trusted when a new comparison accumulates. The prior is quarantined and a fresh artifact starts; the invalid counters never flow into the new artifact's accumulators. @@ -1958,7 +1983,7 @@ def test_record_outcome_does_not_trust_invalid_counters_on_carry(enabled, tmp_pa artifact.write_text(json.dumps(forged, indent=2, sort_keys=True) + "\n") # A new locked status write runs the shadow hook, which loads the prior - # current-v3 artifact, detects the invalid counters, quarantines it, and + # current-version artifact, detects the invalid counters, quarantines it, and # starts fresh with an evidence-unreadable side record plus the new match. _write_run_json_locked(repo, run_dir, "planning") # ck 3, planning 4 @@ -1987,7 +2012,7 @@ def test_record_outcome_does_not_trust_invalid_counters_on_carry(enabled, tmp_pa def test_record_outcome_does_not_trust_negative_counters_on_carry(enabled, tmp_path): - """Blocker #2 carry side: a current-v3 prior with negative comparisons must + """Blocker #2 carry side: a current-version prior with negative comparisons must not produce a negative accumulator (which would silently bypass the no-comparisons gate).""" repo = _repo(tmp_path) @@ -2069,7 +2094,7 @@ def test_record_shadow_comparison_quarantines_invalid_accumulated_counter(enable ], ) def test_malformed_recent_records_quarantined_via_record_shadow_comparison(enabled, tmp_path, malformed_recent_records): - """A current-v3 artifact with valid counters but malformed recent_records + """A current-version artifact with valid counters but malformed recent_records must not crash record_shadow_comparison. The prior is quarantined, counters are not carried, evidence-unreadable precedes the new outcome, and recent_records is a bounded list of dict records.""" @@ -2175,7 +2200,7 @@ def test_non_object_shadow_artifact_quarantined_via_record_shadow_comparison(ena _write_run_json_locked(repo, run_dir, "started") # ck 1, run.created 2 artifact = run_shadow.shadow_artifact_path(run_dir) - # Overwrite the valid current-v3 artifact with a non-object JSON payload. + # Overwrite the valid current-version artifact with a non-object JSON payload. artifact.write_text(non_object_payload + "\n") # Exercise the public writer path through a valid journal/snapshot. diff --git a/tests/test_run_transport_failfast.py b/tests/test_run_transport_failfast.py index 1219febd..a19469d0 100644 --- a/tests/test_run_transport_failfast.py +++ b/tests/test_run_transport_failfast.py @@ -110,3 +110,52 @@ def test_all_ok_stages_succeed_under_fail_fast(dispatch_harness): assert direct_calls == ["a", "b"] assert [result.ok for result in results] == [True, True] assert [result.text for result in results] == ["a ok", "b ok"] + + +def test_dispatch_callbacks_follow_each_external_result_and_keep_attempt_identity(dispatch_harness): + assignments = [ + Assignment(worker="a", task="first", stage=1), + Assignment(worker="a", task="second", stage=1), + ] + facts: list[tuple[str, str, int]] = [] + next_attempt = 0 + + def requested(agent): + nonlocal next_attempt + next_attempt += 1 + facts.append(("requested", agent.name, next_attempt)) + return next_attempt + + def observed(agent, attempt): + facts.append(("observed", agent.name, attempt)) + + def completed(agent, attempt): + facts.append(("completed", agent.name, attempt)) + + def failed(agent, attempt): + facts.append(("failed", agent.name, attempt)) + + results, direct_calls = dispatch_harness( + assignments, + { + "a": [ + agents.AgentResult(text="first", ok=True), + agents.AgentResult(text="second", ok=False, detail="non-zero"), + ] + }, + on_dispatch_requested=requested, + on_dispatch_observed=observed, + on_dispatch_completed=completed, + on_dispatch_failed=failed, + ) + + assert direct_calls == ["a", "a"] + assert [result.ok for result in results] == [True, False] + assert facts == [ + ("requested", "a", 1), + ("observed", "a", 1), + ("completed", "a", 1), + ("requested", "a", 2), + ("observed", "a", 2), + ("failed", "a", 2), + ] diff --git a/tests/test_runs_cmd.py b/tests/test_runs_cmd.py index 2b21e8d2..cf6c7837 100644 --- a/tests/test_runs_cmd.py +++ b/tests/test_runs_cmd.py @@ -1308,6 +1308,51 @@ def test_runs_recover_fails_closed_on_invalid_latest_checkpoint(tmp_path, capsys assert lock_path.is_dir() +def _activate_pending_dispatch_recovery_run(workspace: Path, run_dir: Path) -> None: + run_json_obj = {"status": "dispatching", "task": "demo", "cwd": str(workspace)} + _activate_journal_with_checkpoint(workspace, run_dir, run_json_obj) + shutil.rmtree(workspace / ".brigade" / "run.lock") + with runguard.run_lock(workspace, run_dir=run_dir): + run_lifecycle.record_dispatch_fact( + run_dir, + workspace=workspace, + event_type="run.dispatch.requested", + seat="coder", + ) + _write_lock_owner(workspace, run_dir, pid=99999999) + + +def test_runs_recover_prints_pending_dispatch_only_after_validated_recovery(tmp_path, capsys): + workspace = tmp_path / "workspace" + workspace.mkdir() + run_dir = workspace / ".brigade" / "runs" / _RUN_ID + _activate_pending_dispatch_recovery_run(workspace, run_dir) + + rc = runs_cmd.recover(str(run_dir), cwd=workspace) + + assert rc == 0 + out = capsys.readouterr().out + assert f"recovered: {run_dir}" in out + assert "dispatch recovery: at-least-once work required (seat=coder, attempt=1)" in out + + +def test_runs_recover_invalid_pending_dispatch_checkpoint_prints_no_recovery_hint(tmp_path, capsys): + workspace = tmp_path / "workspace" + workspace.mkdir() + run_dir = workspace / ".brigade" / "runs" / _RUN_ID + _activate_pending_dispatch_recovery_run(workspace, run_dir) + latest = run_checkpoint.latest_checkpoint_event(run_journal.read_journal(_journal_path(run_dir)).events) + assert latest is not None + checkpoint_path = run_checkpoint.checkpoint_path(run_dir, latest.payload["sha256"]) + checkpoint_path.write_bytes(b"x" * latest.payload["byte_size"]) + + rc = runs_cmd.recover(str(run_dir), cwd=workspace) + + captured = capsys.readouterr() + assert rc == 2 + assert "dispatch recovery:" not in captured.out + + def test_runs_recover_accepts_covered_paired_status_event_and_preserves_fields(tmp_path, capsys): """CLI recovery accepts a checkpoint N plus matching status N+1 and preserves fields.