From 58ead9efed3ef6bc7d40cc932b167425e79656f9 Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Fri, 31 Jul 2026 13:02:53 -0400 Subject: [PATCH 1/6] fix(run): retain durable journal enrollment Co-Authored-By: Codex --- src/brigade/aboyeur.py | 29 +++++++++++++-- tests/test_aboyeur.py | 84 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 4 deletions(-) diff --git a/src/brigade/aboyeur.py b/src/brigade/aboyeur.py index 9b3bf420..36cb87ad 100644 --- a/src/brigade/aboyeur.py +++ b/src/brigade/aboyeur.py @@ -2791,7 +2791,7 @@ def record_run_start( codex_transport: str | None = None, started_at: datetime | None = None, scheduler: str | None = None, -) -> None: +) -> bool: """Write the minimal typed receipt needed before optional or blocking work. The requested scheduler is recorded here so a run that dies before dispatch @@ -2870,6 +2870,7 @@ def record_run_start( except (OSError, run_lifecycle.LifecycleJournalError, run_checkpoint.CheckpointError) as exc: raise runguard.RetainRunLockError(f"failed to write initial run receipt: {exc}") from exc _write_json(output_dir / "roster.json", _roster_payload(roster)) + return lifecycle_requested or authority_requested @contextmanager @@ -3037,6 +3038,7 @@ def run( output_dir = output_dir.expanduser() if output_dir is not None else None handoff_inbox = handoff_inbox.expanduser() if handoff_inbox is not None else None direct_worker = worker is not None + durable_enrollment_expected = False # What the roster/flag asked for vs what dispatch actually ran. `used` stays # None until dispatch resolves it, so a run that dies before dispatch reads @@ -3058,11 +3060,30 @@ def _payload(**kwargs: Any) -> dict[str, object]: "lifecycle_journal_requested" not in kwargs or "run_journal_authority_requested" not in kwargs ): run_path = output_dir / "run.json" - if run_path.is_file(): + try: + run_info = os.lstat(run_path) + except FileNotFoundError as exc: + if durable_enrollment_expected: + raise runguard.RetainRunLockError("refusing to overwrite unknown durable enrollment state") from exc + run_info = None + except OSError as exc: + if durable_enrollment_expected: + raise runguard.RetainRunLockError("refusing to overwrite unknown durable enrollment state") from exc + run_info = None + if run_info is not None and (not os.path.isfile(run_path) or os.path.islink(run_path)): + if durable_enrollment_expected: + raise runguard.RetainRunLockError("refusing to overwrite unknown durable enrollment state") + elif run_info is not None: try: existing = json.loads(run_path.read_text()) - except (OSError, json.JSONDecodeError): + except (OSError, ValueError, RecursionError) as exc: + if durable_enrollment_expected: + raise runguard.RetainRunLockError( + "refusing to overwrite unknown durable enrollment state" + ) from exc existing = None + if not isinstance(existing, dict) and durable_enrollment_expected: + raise runguard.RetainRunLockError("refusing to overwrite unknown durable enrollment state") if isinstance(existing, dict): if ( "lifecycle_journal_requested" not in kwargs @@ -3139,7 +3160,7 @@ def _drift_failure_rc() -> int | None: print(f"error: {worker_error}", file=sys.stderr) return 2 if output_dir is not None: - record_run_start( + durable_enrollment_expected = record_run_start( output_dir, task=task, cwd=cwd, diff --git a/tests/test_aboyeur.py b/tests/test_aboyeur.py index 33f3a7d2..b3975654 100644 --- a/tests/test_aboyeur.py +++ b/tests/test_aboyeur.py @@ -5662,6 +5662,30 @@ def test_default_journal_authority_implies_lifecycle_without_lifecycle_environme assert meta[_LIFECYCLE_REQUEST_FIELD] is True +def test_record_run_start_treats_lifecycle_only_enrollment_as_durable(tmp_path): + workspace, run_dir = _authority_run_dir(tmp_path) + lifecycle_only = _legacy_status_payload("started") + lifecycle_only[_LIFECYCLE_REQUEST_FIELD] = True + (run_dir / "run.json").write_text(json.dumps(lifecycle_only)) + + with runguard.run_lock(workspace, run_dir=run_dir): + assert ( + aboyeur.record_run_start( + run_dir, + task="lifecycle only", + cwd=workspace, + roster=_roster(), + read_only=False, + lock_workspace=workspace, + ) + is True + ) + + receipt = json.loads((run_dir / "run.json").read_text()) + assert receipt[_LIFECYCLE_REQUEST_FIELD] is True + assert _AUTHORITY_REQUEST_FIELD not in receipt + + def test_existing_legacy_run_remains_snapshot_only_when_rerecorded(tmp_path, monkeypatch): """A pre-cutover receipt without enrollment fields stays snapshot-only.""" workspace, run_dir = _authority_run_dir(tmp_path) @@ -5932,6 +5956,66 @@ def capture_write_json(path, payload): assert aboyeur._resolve_authority_state(output_dir) == "authoritative" +@pytest.mark.parametrize( + ("case", "mutate"), + [ + pytest.param("invalid-json", lambda path: path.write_bytes(b"{not valid json"), id="invalid-json"), + pytest.param("json-array", lambda path: path.write_text("[]"), id="json-array"), + pytest.param("json-null", lambda path: path.write_text("null"), id="json-null"), + pytest.param("missing", lambda path: path.unlink(), id="missing"), + pytest.param("non-regular", lambda path: (path.unlink(), path.mkdir()), id="non-regular"), + pytest.param("symlink", lambda path: (path.unlink(), path.symlink_to("untrusted-run.json")), id="symlink"), + ], +) +def test_run_payload_fails_closed_when_enrolled_run_json_becomes_corrupt(monkeypatch, tmp_path, case, mutate): + """A later run status write must not downgrade a corrupt enrolled receipt.""" + monkeypatch.delenv(_LIFECYCLE_ENV, raising=False) + + def fake_run_agent(cli_ref, prompt, timeout=600.0, cwd=None, read_only=False): + if cli_ref == "ollama:llama3.3": + return agents.AgentResult(text="worker output", ok=True) + return agents.AgentResult( + text=json.dumps({"assignments": [{"worker": "coder", "task": "implement it"}]}), ok=True + ) + + run_cwd = tmp_path / "work" + run_cwd.mkdir() + _init_git_repo(run_cwd) + (run_cwd / "tracked.txt").write_text("initial\n") + _commit_all(run_cwd) + output_dir = tmp_path / "run" + real_write_json = aboyeur._write_json + run_json_writes = 0 + + def corrupt_after_enrollment(path, payload): + nonlocal run_json_writes + result = real_write_json(path, payload) + if Path(path).name == "run.json": + run_json_writes += 1 + if run_json_writes == 1: + mutate(Path(path)) + return result + + monkeypatch.setattr(aboyeur.agents, "run_agent", fake_run_agent) + monkeypatch.setattr(aboyeur, "_write_json", corrupt_after_enrollment) + + with runguard.run_lock(run_cwd, run_dir=output_dir): + with pytest.raises( + runguard.RetainRunLockError, + match="refusing to overwrite unknown durable enrollment state", + ) as exc_info: + run_aboyeur_guarded( + "build feature", + _roster(), + cwd=run_cwd, + output_dir=output_dir, + code_graph_enabled=False, + ) + + assert str(exc_info.value) == "refusing to overwrite unknown durable enrollment state" + assert run_json_writes == 1, f"{case} was overwritten by a legacy payload" + + # -- Issue #568 slice 7 assignment 3: durable enrollment fail-closed ------------ _CORRUPT_RUN_JSON_CASES = [ From 12e2a5f10b80dfd843b50a068b0ab252b549e7d1 Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Fri, 31 Jul 2026 13:03:00 -0400 Subject: [PATCH 2/6] fix(journal): anchor redaction records Co-Authored-By: Codex --- src/brigade/run_checkpoint.py | 8 + src/brigade/run_events.py | 9 + src/brigade/run_projector.py | 1 + src/brigade/run_redaction.py | 548 ++++++++++++++++++++++++++++------ tests/test_run_checkpoint.py | 29 ++ tests/test_run_redaction.py | 306 ++++++++++++++++++- 6 files changed, 811 insertions(+), 90 deletions(-) diff --git a/src/brigade/run_checkpoint.py b/src/brigade/run_checkpoint.py index 1ff8f347..d50a1f8d 100644 --- a/src/brigade/run_checkpoint.py +++ b/src/brigade/run_checkpoint.py @@ -980,6 +980,14 @@ def _verify_coverage( paired_event_type = latest.payload.get("paired_event_type") pairing_key = latest.payload.get("pairing_key") tail = events[-1] + trailing_redaction_anchors = 0 + while tail.event_type == "run.redaction.recorded": + trailing_redaction_anchors += 1 + if trailing_redaction_anchors == len(events): + raise CheckpointError( + _bound("journal tail is not covered by the latest checkpoint"), category="uncovered-tail" + ) + tail = events[-1 - trailing_redaction_anchors] if latest.sequence == tail.sequence: # A dispatch pairing key promises a specific identity-bearing fact. # A checkpoint at tail means that fact never committed, so recovery diff --git a/src/brigade/run_events.py b/src/brigade/run_events.py index a9fc09d9..c5a6d629 100644 --- a/src/brigade/run_events.py +++ b/src/brigade/run_events.py @@ -107,6 +107,15 @@ } ), "run.artifact_collection.started": frozenset({"detail"}), + "run.redaction.recorded": frozenset( + { + "operation_id", + "affected_first_sequence", + "affected_last_sequence", + "reason_class", + "record_sha256", + } + ), } APPROVAL_DECISION_STATES = frozenset({"pending", "approved", "rejected", "held", "consumed"}) APPROVAL_DECISION_EVENT_STATES = { diff --git a/src/brigade/run_projector.py b/src/brigade/run_projector.py index acbf9cbc..3a3f9b93 100644 --- a/src/brigade/run_projector.py +++ b/src/brigade/run_projector.py @@ -144,6 +144,7 @@ def _has_dispatch_identity(payload: Any) -> bool: "approval.rejected", "approval.held", "approval.consumed", + "run.redaction.recorded", } ) diff --git a/src/brigade/run_redaction.py b/src/brigade/run_redaction.py index 897d7a71..c7694d67 100644 --- a/src/brigade/run_redaction.py +++ b/src/brigade/run_redaction.py @@ -430,9 +430,12 @@ def _rewrite_events( idempotency_keys: set[str] = set() for event in events: affected = sequence_start <= event.sequence <= sequence_end - payload = _redacted_payload(event) if affected else dict(event.payload) + preserved_anchor = event.event_type == "run.redaction.recorded" + payload = _redacted_payload(event) if affected and not preserved_anchor else dict(event.payload) idempotency_key = ( - _redaction_idempotency_key(operation_id, event.sequence) if affected else event.idempotency_key + _redaction_idempotency_key(operation_id, event.sequence) + if affected and not preserved_anchor + else event.idempotency_key ) if idempotency_key in idempotency_keys: raise RedactionError("rewritten journal idempotency key collision") @@ -1122,10 +1125,15 @@ def _record_bytes( rewritten_sha256: str | None, quarantine_retained: bool, parent_operation_id: str | None, + parent_record_sha256: str | None = None, rewritten_digest_retired_by: str | None = None, ) -> bytes: if (rewritten_sha256 is None) == (rewritten_digest_retired_by is None): raise RedactionError("redaction record digest is invalid") + if (parent_operation_id is None) != (parent_record_sha256 is None): + raise RedactionError("redaction record parent reference is invalid") + if parent_record_sha256 is not None and not re.fullmatch(r"[0-9a-f]{64}", parent_record_sha256): + raise RedactionError("redaction record parent reference is invalid") payload = { "schema": REDACTION_SCHEMA, "schema_version": REDACTION_SCHEMA_VERSION, @@ -1139,6 +1147,8 @@ def _record_bytes( "projection_verified": True, "quarantine_retained": quarantine_retained, } + if parent_record_sha256 is not None: + payload["parent_record_sha256"] = parent_record_sha256 if rewritten_sha256 is not None: payload["rewritten_journal_sha256"] = rewritten_sha256 if rewritten_digest_retired_by is not None: @@ -1157,6 +1167,7 @@ def _write_redaction_record( rewritten_sha256: str | None, quarantine_retained: bool, parent_operation_id: str | None = None, + parent_record_sha256: str | None = None, rewritten_digest_retired_by: str | None = None, ) -> None: _atomic_write( @@ -1170,6 +1181,7 @@ def _write_redaction_record( rewritten_sha256=rewritten_sha256, quarantine_retained=quarantine_retained, parent_operation_id=parent_operation_id, + parent_record_sha256=parent_record_sha256, rewritten_digest_retired_by=rewritten_digest_retired_by, ), mode=_FILE_MODE, @@ -1177,6 +1189,16 @@ def _write_redaction_record( ) +def _redaction_record_sha256(record_path: Path) -> str: + return _digest( + _read_bounded_regular( + record_path, + limit=16 * 1024, + category="redaction record", + ) + ) + + def _validate_redaction_record( record_path: Path, *, @@ -1209,6 +1231,12 @@ def _validate_redaction_record( or not isinstance(record.get("quarantine_retained"), bool) ): raise RedactionError("redaction record verification failed") + parent_record_sha256 = record.get("parent_record_sha256") + if (parent_operation_id is None) != (parent_record_sha256 is None) or ( + parent_record_sha256 is not None + and (not isinstance(parent_record_sha256, str) or not re.fullmatch(r"[0-9a-f]{64}", parent_record_sha256)) + ): + raise RedactionError("redaction record verification failed") if rewritten_sha256 is not None: if ( not re.fullmatch(r"[0-9a-f]{64}", rewritten_sha256) @@ -1238,6 +1266,7 @@ def _parse_lineage_record(raw_record: bytes, *, operation_id: str, run_id: str) and rewritten_digest_retired_by != operation_id ) parent = record.get("parent_operation_id") + parent_record_sha256 = record.get("parent_record_sha256") if ( record.get("schema") != REDACTION_SCHEMA or record.get("schema_version") != REDACTION_SCHEMA_VERSION @@ -1250,6 +1279,11 @@ def _parse_lineage_record(raw_record: bytes, *, operation_id: str, run_id: str) or not isinstance(record.get("sequence_end"), int) or valid_rewritten == valid_retirement or (parent is not None and (not isinstance(parent, str) or not _OPERATION_RE.fullmatch(parent))) + or (parent is None) != (parent_record_sha256 is None) + or ( + parent_record_sha256 is not None + and (not isinstance(parent_record_sha256, str) or not re.fullmatch(r"[0-9a-f]{64}", parent_record_sha256)) + ) or record.get("chain_verified") is not True or record.get("projection_verified") is not True or not isinstance(record.get("quarantine_retained"), bool) @@ -1258,6 +1292,190 @@ def _parse_lineage_record(raw_record: bytes, *, operation_id: str, run_id: str) return record +def _parent_record_reference( + events: Sequence[run_journal.RunEvent], + parent_operation_id: str | None, +) -> str | None: + if parent_operation_id is None: + return None + parent_anchors = [ + event + for event in events + if event.event_type == "run.redaction.recorded" and event.payload.get("operation_id") == parent_operation_id + ] + if len(parent_anchors) != 1: + raise RedactionError("redaction lineage is incomplete") + payload = parent_anchors[0].payload + reference = payload.get("record_sha256") + if ( + set(payload) + != { + "operation_id", + "affected_first_sequence", + "affected_last_sequence", + "reason_class", + "record_sha256", + } + or not isinstance(reference, str) + or not re.fullmatch(r"[0-9a-f]{64}", reference) + ): + raise RedactionError("redaction lineage parent reference is invalid") + return reference + + +def _validate_chained_anchors( + events: Sequence[run_journal.RunEvent], + records: Mapping[str, Mapping[str, Any]], + record_digests: Mapping[str, str], + states: Mapping[str, Mapping[str, Any]] | None = None, + *, + resumable_operation_id: str | None = None, +) -> None: + seen: set[str] = set() + for event in events: + if event.event_type != "run.redaction.recorded": + continue + payload = event.payload + if set(payload) != { + "operation_id", + "affected_first_sequence", + "affected_last_sequence", + "reason_class", + "record_sha256", + }: + raise RedactionError("redaction chained anchor verification failed") + operation_id = payload.get("operation_id") + if not isinstance(operation_id, str) or operation_id in seen: + raise RedactionError("redaction chained anchor verification failed") + record = records.get(operation_id) + if record is None: + raise RedactionError("redaction chained anchor verification failed") + seen.add(operation_id) + expected_digest = record_digests.get(operation_id) + if expected_digest is None: + raise RedactionError("redaction chained anchor verification failed") + if ( + payload.get("affected_first_sequence") != record.get("sequence_start") + or payload.get("affected_last_sequence") != record.get("sequence_end") + or payload.get("reason_class") != record.get("reason_code") + ): + raise RedactionError("redaction chained anchor verification failed") + if payload.get("record_sha256") != expected_digest: + current = states.get(resumable_operation_id) if states is not None and resumable_operation_id else None + allowed_current = ( + operation_id == resumable_operation_id + and current is not None + and current.get("phase") + in { + "cleanup-authorized", + "cleaned", + } + ) + allowed_parent = ( + current is not None + and current.get("phase") in {"cleanup-authorized", "cleaned"} + and record.get("rewritten_digest_retired_by") == resumable_operation_id + and current.get("parent_operation_id") == operation_id + ) + if not (allowed_current or allowed_parent): + raise RedactionError("redaction chained anchor verification failed") + missing = set(records) - seen + if missing: + resumable_state = states.get(resumable_operation_id) if states is not None and resumable_operation_id else None + if missing != {resumable_operation_id} or resumable_state is None or resumable_state.get("phase") != "verified": + raise RedactionError("redaction chained anchor verification failed") + + +def _redaction_anchor_payload(record: Mapping[str, Any], *, record_sha256: str) -> dict[str, Any]: + return { + "operation_id": record["operation_id"], + "affected_first_sequence": record["sequence_start"], + "affected_last_sequence": record["sequence_end"], + "reason_class": record["reason_code"], + "record_sha256": record_sha256, + } + + +def _append_redaction_anchor( + journal_path: Path, + events: Sequence[run_journal.RunEvent], + record: Mapping[str, Any], + *, + record_sha256: str, +) -> run_journal.RunEvent: + return run_journal.append_event( + journal_path, + run_id=events[-1].run_id, + event_type="run.redaction.recorded", + payload=_redaction_anchor_payload(record, record_sha256=record_sha256), + idempotency_key=f"redaction-recorded:{record['operation_id']}", + expected_previous_sequence=events[-1].sequence, + ) + + +def _refresh_chained_anchors( + run_dir: Path, + *, + workspace: Path, + resumable_operation_id: str | None, +) -> str: + """Re-chain anchor payload hashes after a record lifecycle update.""" + journal_path = run_dir / "events" / "lifecycle.jsonl" + events = _verified_events(journal_path, category="journal") + records, states, record_digests = _load_operation_inventory( + run_dir / "events" / "redactions", run_dir.name, resumable_operation_id=resumable_operation_id + ) + _validate_chained_anchors( + events, + records, + record_digests, + states, + resumable_operation_id=resumable_operation_id, + ) + rewritten: list[run_journal.RunEvent] = [] + previous_digest: str | None = None + for event in events: + payload = ( + _redaction_anchor_payload( + records[event.payload["operation_id"]], + record_sha256=record_digests[event.payload["operation_id"]], + ) + if event.event_type == "run.redaction.recorded" + else event.payload + ) + envelope = run_events.build_event( + run_id=event.run_id, + sequence=event.sequence, + event_type=event.event_type, + payload=payload, + idempotency_key=event.idempotency_key, + recorded_at=event.recorded_at, + previous_digest=previous_digest, + ) + rewritten_event = run_journal.RunEvent( + schema=envelope["schema"], + schema_version=envelope["schema_version"], + event_id=envelope["event_id"], + run_id=envelope["run_id"], + sequence=envelope["sequence"], + event_type=envelope["event_type"], + recorded_at=envelope["recorded_at"], + idempotency_key=envelope["idempotency_key"], + request_digest=envelope["request_digest"], + previous_digest=envelope["previous_digest"], + event_digest=envelope["event_digest"], + payload=dict(envelope["payload"]), + ) + rewritten.append(rewritten_event) + previous_digest = rewritten_event.event_digest + data = _canonical_event_bytes(rewritten) + _assert_active_owner(workspace, run_dir) + _replace_journal(journal_path, data) + snapshot = _load_json_object(run_dir / "run.json", limit=MAX_RUN_JSON_BYTES, category="run projection") + _replace_projection(run_dir, _projection(snapshot, rewritten)) + return _digest(data) + + def _validate_lineage_graph(records: Mapping[str, Mapping[str, Any]]) -> None: if not records: return @@ -1300,12 +1518,13 @@ def _load_operation_inventory( run_id: str, *, resumable_operation_id: str | None, -) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]]: +) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]], dict[str, str]]: if not os.path.lexists(redactions_dir): - return {}, {} + return {}, {}, {} root_fd = _open_directory_handle(redactions_dir, category="redactions") records: dict[str, dict[str, Any]] = {} states: dict[str, dict[str, Any]] = {} + record_digests: dict[str, str] = {} incomplete: list[str] = [] try: try: @@ -1387,6 +1606,7 @@ def _load_operation_inventory( raise RedactionError("incomplete redaction transaction exists; retry its original request") incomplete.append(operation_id) records[operation_id] = record + record_digests[operation_id] = _digest(raw_record) finally: os.close(root_fd) @@ -1408,10 +1628,8 @@ def _load_operation_inventory( if state_digest != record_digest or state_retired_by != record_retired_by: if record_retired_by is not None and state_digest is not None: split_retired_by = record_retired_by - split_digest = state_digest elif state_retired_by is not None and record_digest is not None: split_retired_by = state_retired_by - split_digest = record_digest else: raise RedactionError("redaction transaction state/record mismatch") child_state = states.get(split_retired_by) @@ -1423,7 +1641,6 @@ def _load_operation_inventory( or child_state.get("phase") != "cleanup-authorized" or child_state.get("parent_operation_id") != operation_id or child_record.get("parent_operation_id") != operation_id - or child_state.get("original_sha256") != split_digest ): raise RedactionError("redaction transaction state/record mismatch") phase = state.get("phase") @@ -1432,92 +1649,109 @@ def _load_operation_inventory( phase == "cleaned" and quarantine_retained is not False ): raise RedactionError("redaction transaction state/record mismatch") + parent_operation_id = record.get("parent_operation_id") + parent_reference = record.get("parent_record_sha256") + if parent_operation_id is None: + if parent_reference is not None: + raise RedactionError("redaction lineage parent reference is invalid") + continue + if parent_operation_id not in states or not isinstance(parent_reference, str): + raise RedactionError("redaction lineage parent reference is invalid") + quarantine = redactions_dir / operation_id / "original.jsonl" + if os.path.lexists(quarantine): + original_sha256 = state.get("original_sha256") + if not isinstance(original_sha256, str) or not re.fullmatch(r"[0-9a-f]{64}", original_sha256): + raise RedactionError("redaction lineage child quarantine is invalid") + _verify_quarantine(quarantine, original_sha256) + if _parent_record_reference( + _verified_events(quarantine, category="redaction quarantine"), parent_operation_id + ) != (parent_reference): + raise RedactionError("redaction lineage parent reference is invalid") + elif phase not in {"cleanup-authorized", "cleaned"}: + raise RedactionError("redaction lineage child quarantine is missing") _validate_lineage_graph(records) - return records, states + return records, states, record_digests def _retire_rewritten_digest_aliases( redactions_dir: Path, *, run_id: str, - sensitive_digest: str, + parent_operation_id: str | None, retired_by_operation_id: str, records: Mapping[str, Mapping[str, Any]], ) -> None: - for operation_id, record in records.items(): - state_path = redactions_dir / operation_id / "state.json" - state = _load_and_validate_state( - state_path, - operation_id=operation_id, + if parent_operation_id is None: + return + record = records.get(parent_operation_id) + if record is None: + raise RedactionError("redaction lineage is incomplete") + operation_dir = redactions_dir / parent_operation_id + state_path = operation_dir / "state.json" + record_path = operation_dir / "record.json" + state = _load_and_validate_state( + state_path, + operation_id=parent_operation_id, + run_id=run_id, + sequence_start=record["sequence_start"], + sequence_end=record["sequence_end"], + reason=record["reason_code"], + ) + if state.get("parent_operation_id") != record.get("parent_operation_id"): + raise RedactionError("redaction lineage metadata mismatch") + if record.get("rewritten_digest_retired_by") not in {None, retired_by_operation_id} or state.get( + "rewritten_digest_retired_by" + ) not in {None, retired_by_operation_id}: + raise RedactionError("redaction lineage digest mismatch") + if record.get("rewritten_digest_retired_by") != retired_by_operation_id: + _write_redaction_record( + record_path, + operation_id=parent_operation_id, run_id=run_id, sequence_start=record["sequence_start"], sequence_end=record["sequence_end"], reason=record["reason_code"], + rewritten_sha256=None, + quarantine_retained=record["quarantine_retained"], + parent_operation_id=record.get("parent_operation_id"), + parent_record_sha256=record.get("parent_record_sha256"), + rewritten_digest_retired_by=retired_by_operation_id, ) - if state.get("parent_operation_id") != record.get("parent_operation_id"): - raise RedactionError("redaction lineage metadata mismatch") - - record_digest = record.get("rewritten_journal_sha256") - record_retired_by = record.get("rewritten_digest_retired_by") - state_digest = state.get("rewritten_sha256") - state_retired_by = state.get("rewritten_digest_retired_by") - record_matches = record_digest == sensitive_digest or record_retired_by == retired_by_operation_id - state_matches = state_digest == sensitive_digest or state_retired_by == retired_by_operation_id - if not record_matches and not state_matches: - continue - if (record_digest != sensitive_digest and record_retired_by != retired_by_operation_id) or ( - state_digest != sensitive_digest and state_retired_by != retired_by_operation_id - ): - raise RedactionError("redaction lineage digest mismatch") - operation_dir = redactions_dir / operation_id - record_path = operation_dir / "record.json" - if not (record_retired_by == retired_by_operation_id and state_retired_by == retired_by_operation_id): - _write_redaction_record( - record_path, - operation_id=operation_id, - run_id=run_id, - sequence_start=record["sequence_start"], - sequence_end=record["sequence_end"], - reason=record["reason_code"], - rewritten_sha256=None, - quarantine_retained=record["quarantine_retained"], - parent_operation_id=record.get("parent_operation_id"), - rewritten_digest_retired_by=retired_by_operation_id, - ) - _write_state( - state_path, - operation_id=operation_id, - run_id=run_id, - sequence_start=record["sequence_start"], - sequence_end=record["sequence_end"], - reason=record["reason_code"], - original_sha256=state.get("original_sha256"), - rewritten_sha256=None, - phase=state["phase"], - parent_operation_id=state.get("parent_operation_id"), - rewritten_digest_retired_by=retired_by_operation_id, - ) - operation_fd = _open_directory_handle(operation_dir, category="redaction operation") - try: - _remove_operation_temps( - operation_dir, - operation_fd, - patterns=(_STATE_TEMP_RE, _RECORD_TEMP_RE), - category="redaction lineage cleanup", - ) - finally: - os.close(operation_fd) - _validate_redaction_record( - record_path, - operation_id=operation_id, + if state.get("rewritten_digest_retired_by") != retired_by_operation_id: + _write_state( + state_path, + operation_id=parent_operation_id, run_id=run_id, sequence_start=record["sequence_start"], sequence_end=record["sequence_end"], reason=record["reason_code"], + original_sha256=state.get("original_sha256"), rewritten_sha256=None, - parent_operation_id=record.get("parent_operation_id"), + phase=state["phase"], + parent_operation_id=state.get("parent_operation_id"), rewritten_digest_retired_by=retired_by_operation_id, ) + operation_fd = _open_directory_handle(operation_dir, category="redaction operation") + try: + _remove_operation_temps( + operation_dir, + operation_fd, + patterns=(_STATE_TEMP_RE, _RECORD_TEMP_RE), + category="redaction lineage cleanup", + ) + finally: + os.close(operation_fd) + _validate_redaction_record( + record_path, + operation_id=parent_operation_id, + run_id=run_id, + sequence_start=record["sequence_start"], + sequence_end=record["sequence_end"], + reason=record["reason_code"], + rewritten_sha256=None, + parent_operation_id=record.get("parent_operation_id"), + rewritten_digest_retired_by=retired_by_operation_id, + ) def _lineage_parent_for_digest( @@ -1539,8 +1773,16 @@ def _lineage_contains( *, ancestor_operation_id: str, active_journal_digest: str, + active_events: Sequence[run_journal.RunEvent] | None = None, ) -> bool: - current = _lineage_parent_for_digest(records, active_journal_digest) + anchored_operations = ( + [event.payload["operation_id"] for event in active_events if event.event_type == "run.redaction.recorded"] + if active_events is not None + else [] + ) + current = ( + anchored_operations[-1] if anchored_operations else _lineage_parent_for_digest(records, active_journal_digest) + ) visited: set[str] = set() while current is not None: if current == ancestor_operation_id: @@ -1585,7 +1827,10 @@ def _assert_affected_values_removed( end: int, ) -> None: for prior, current in zip(original, rewritten, strict=True): - if not start <= prior.sequence <= end or prior.event_type == run_checkpoint.CHECKPOINT_EVENT_TYPE: + if not start <= prior.sequence <= end or prior.event_type in { + run_checkpoint.CHECKPOINT_EVENT_TYPE, + "run.redaction.recorded", + }: continue for key, value in prior.payload.items(): if key == "status" or value is None: @@ -1703,6 +1948,7 @@ def _post_replace_verify( run_dir: Path, *, expected_digest: str | None, + resumable_operation_id: str | None = None, ) -> tuple[dict[str, Any], list[run_journal.RunEvent], str]: journal_path = run_dir / "events" / "lifecycle.jsonl" active = _read_bounded_regular( @@ -1714,6 +1960,19 @@ def _post_replace_verify( if expected_digest is not None and active_digest != expected_digest: raise RedactionError("post-rewrite verification failed") events = _verified_events(journal_path, category="post-rewrite journal") + if any(event.event_type == "run.redaction.recorded" for event in events): + records, states, record_digests = _load_operation_inventory( + run_dir / "events" / "redactions", + run_dir.name, + resumable_operation_id=resumable_operation_id, + ) + _validate_chained_anchors( + events, + records, + record_digests, + states, + resumable_operation_id=resumable_operation_id, + ) snapshot = _load_json_object( run_dir / "run.json", limit=MAX_RUN_JSON_BYTES, @@ -1750,8 +2009,34 @@ def _resume_projection_after_rewrite( projected = _projection(snapshot, events) if snapshot == projected.snapshot: return - if _projection_semantics(snapshot) != _projection_semantics(projected.snapshot): - raise RedactionError("post-rewrite projection semantics changed") + last_sequence = snapshot.get("journal_last_sequence") + projected_last_sequence = projected.snapshot.get("journal_last_sequence") + if last_sequence == projected_last_sequence: + if _projection_semantics(snapshot) != _projection_semantics(projected.snapshot): + raise RedactionError("post-rewrite projection semantics changed") + else: + snapshot_without_sequence = { + key: value + for key, value in snapshot.items() + if key not in {_PROJECTION_DIGEST_FIELD, "journal_last_sequence"} + } + projected_without_sequence = { + key: value + for key, value in projected.snapshot.items() + if key not in {_PROJECTION_DIGEST_FIELD, "journal_last_sequence"} + } + if snapshot_without_sequence != projected_without_sequence: + raise RedactionError("post-rewrite projection semantics changed") + if ( + isinstance(last_sequence, bool) + or not isinstance(last_sequence, int) + or isinstance(projected_last_sequence, bool) + or not isinstance(projected_last_sequence, int) + or last_sequence < 0 + or last_sequence > projected_last_sequence + or any(event.sequence > last_sequence and event.event_type != "run.redaction.recorded" for event in events) + ): + raise RedactionError("post-rewrite projection sequence lag is invalid") _replace_projection(run_dir, projected) @@ -1803,13 +2088,31 @@ def redact_journal( if original_bytes != _canonical_event_bytes(events): raise RedactionError("journal changed during redaction preflight") active_digest = _digest(original_bytes) - lineage_records, inventory_states = _load_operation_inventory( + lineage_records, inventory_states, record_digests = _load_operation_inventory( redactions_dir, resolved_run_dir.name, resumable_operation_id=operation_id, ) + _validate_chained_anchors( + events, + lineage_records, + record_digests, + inventory_states, + resumable_operation_id=operation_id, + ) prior_state = inventory_states.get(operation_id) + if prior_state is not None and any( + event.event_type == "run.redaction.recorded" and event.payload.get("operation_id") == operation_id + for event in events + ): + _resume_projection_after_rewrite(resolved_run_dir, events) + snapshot = _load_json_object( + resolved_run_dir / "run.json", + limit=MAX_RUN_JSON_BYTES, + category="run projection", + ) + if prior_state is not None: if ( prior_state.get("sequence_start") != start @@ -1839,6 +2142,7 @@ def redact_journal( lineage_records, ancestor_operation_id=operation_id, active_journal_digest=active_digest, + active_events=events, ) if is_current_or_descendant: if prior_state["phase"] == "cleanup-authorized": @@ -1852,9 +2156,13 @@ def redact_journal( if not isinstance(original_sha256, str): raise RedactionError("redaction transaction digest is invalid") _verify_quarantine(quarantine_path, original_sha256) - if isinstance(rewritten_sha256, str) and active_digest == rewritten_sha256: + anchor_present = any( + event.event_type == "run.redaction.recorded" + and event.payload.get("operation_id") == operation_id + for event in events + ) + if anchor_present: _resume_projection_after_rewrite(resolved_run_dir, events) - _post_replace_verify(resolved_run_dir, expected_digest=None) if not os.path.lexists(record_path): if not isinstance(rewritten_sha256, str) or active_digest != rewritten_sha256 or cleaned: raise RedactionError("redaction lineage record is missing") @@ -1869,6 +2177,7 @@ def redact_journal( rewritten_sha256=rewritten_sha256, quarantine_retained=True, parent_operation_id=parent_operation_id, + parent_record_sha256=_parent_record_reference(events, parent_operation_id), ) except (OSError, RedactionError) as exc: raise RedactionError("redaction record write failed") from exc @@ -1895,6 +2204,27 @@ def redact_journal( parent_operation_id=parent_operation_id, rewritten_digest_retired_by=rewritten_digest_retired_by, ) + if not anchor_present: + record = _load_json_object(record_path, limit=16 * 1024, category="redaction record") + _assert_active_owner(workspace, resolved_run_dir) + anchor = _append_redaction_anchor( + journal_path, + events, + record, + record_sha256=_redaction_record_sha256(record_path), + ) + _replace_projection( + resolved_run_dir, + _projection( + _load_json_object( + resolved_run_dir / "run.json", + limit=MAX_RUN_JSON_BYTES, + category="run projection", + ), + [*events, anchor], + ), + ) + _post_replace_verify(resolved_run_dir, expected_digest=None) return RedactionReport( operation_id, start, @@ -1906,11 +2236,22 @@ def redact_journal( if active_digest != prior_state.get("original_sha256") or prior_state["phase"] != "quarantined": raise RedactionError("redaction transaction does not match the active journal") + active_anchor_ids = [ + event.payload["operation_id"] for event in events if event.event_type == "run.redaction.recorded" + ] parent_operation_id = ( prior_state.get("parent_operation_id") if prior_state is not None else _lineage_parent_for_digest(lineage_records, active_digest) + or (active_anchor_ids[-1] if active_anchor_ids else None) ) + if not any( + start <= event.sequence <= end + and event.event_type not in {run_checkpoint.CHECKPOINT_EVENT_TYPE, "run.redaction.recorded"} + for event in events + ): + raise RedactionError("redaction range contains no redactable payloads") + parent_record_sha256 = _parent_record_reference(events, parent_operation_id) rewritten_events, rewritten_bytes = _rewrite_events( events, sequence_start=start, @@ -1965,7 +2306,11 @@ def redact_journal( parent_operation_id=parent_operation_id, ) _replace_projection(resolved_run_dir, after_projection) - _post_replace_verify(resolved_run_dir, expected_digest=rewritten_sha256) + _post_replace_verify( + resolved_run_dir, + expected_digest=rewritten_sha256, + resumable_operation_id=operation_id, + ) try: _write_redaction_record( record_path, @@ -1977,6 +2322,7 @@ def redact_journal( rewritten_sha256=rewritten_sha256, quarantine_retained=True, parent_operation_id=parent_operation_id, + parent_record_sha256=parent_record_sha256, ) except (OSError, RedactionError) as exc: raise RedactionError("redaction record write failed") from exc @@ -2002,6 +2348,16 @@ def redact_journal( rewritten_sha256=rewritten_sha256, parent_operation_id=parent_operation_id, ) + record = _load_json_object(record_path, limit=16 * 1024, category="redaction record") + _assert_active_owner(workspace, resolved_run_dir) + anchor = _append_redaction_anchor( + journal_path, + rewritten_events, + record, + record_sha256=_redaction_record_sha256(record_path), + ) + _replace_projection(resolved_run_dir, _projection(after_projection.snapshot, [*rewritten_events, anchor])) + _post_replace_verify(resolved_run_dir, expected_digest=None) return RedactionReport(operation_id, start, end, quarantine_path, record_path) @@ -2029,7 +2385,7 @@ def cleanup_redaction_quarantine( if not _existing_operation_dir(resolved_run_dir, operation_id): raise RedactionError("redaction transaction does not exist") state_path = operation_dir / "state.json" - lineage_records, inventory_states = _load_operation_inventory( + lineage_records, inventory_states, _ = _load_operation_inventory( operation_dir.parent, resolved_run_dir.name, resumable_operation_id=operation_id, @@ -2037,10 +2393,14 @@ def cleanup_redaction_quarantine( state = inventory_states.get(operation_id) if state is None: raise RedactionError("redaction transaction metadata mismatch") + record = lineage_records.get(operation_id) + if record is None: + raise RedactionError("redaction transaction metadata mismatch") start = state.get("sequence_start") end = state.get("sequence_end") reason = state.get("reason_code") parent_operation_id = state.get("parent_operation_id") + parent_record_sha256 = record.get("parent_record_sha256") rewritten_sha256 = state.get("rewritten_sha256") rewritten_digest_retired_by = state.get("rewritten_digest_retired_by") if ( @@ -2055,12 +2415,19 @@ def cleanup_redaction_quarantine( journal_path = resolved_run_dir / "events" / "lifecycle.jsonl" events = _verified_events(journal_path, category="journal") + _resume_projection_after_rewrite(resolved_run_dir, events) + snapshot = _load_json_object( + resolved_run_dir / "run.json", + limit=MAX_RUN_JSON_BYTES, + category="run projection", + ) _verify_current_projection(snapshot, events) _validate_checkpoint_artifacts(resolved_run_dir, snapshot, events) try: _, _, active_digest = _post_replace_verify( resolved_run_dir, expected_digest=None, + resumable_operation_id=operation_id, ) except RedactionError as exc: raise RedactionError("cleanup verification failed") from exc @@ -2068,6 +2435,7 @@ def cleanup_redaction_quarantine( lineage_records, ancestor_operation_id=operation_id, active_journal_digest=active_digest, + active_events=events, ): raise RedactionError("cleanup verification failed") try: @@ -2088,6 +2456,12 @@ def cleanup_redaction_quarantine( if state["phase"] == "cleaned": if os.path.lexists(quarantine_path): raise RedactionError("cleanup state conflicts with retained quarantine") + refreshed_digest = _refresh_chained_anchors( + resolved_run_dir, + workspace=workspace, + resumable_operation_id=operation_id, + ) + _post_replace_verify(resolved_run_dir, expected_digest=refreshed_digest) return RedactionReport( operation_id, start, @@ -2120,6 +2494,7 @@ def cleanup_redaction_quarantine( rewritten_sha256=rewritten_sha256, quarantine_retained=True, parent_operation_id=parent_operation_id, + parent_record_sha256=parent_record_sha256, rewritten_digest_retired_by=rewritten_digest_retired_by, ) _write_state( @@ -2168,14 +2543,10 @@ def cleanup_redaction_quarantine( _retire_rewritten_digest_aliases( operation_dir.parent, run_id=resolved_run_dir.name, - sensitive_digest=original_sha256, + parent_operation_id=parent_operation_id, retired_by_operation_id=operation_id, records=lineage_records, ) - _post_replace_verify( - resolved_run_dir, - expected_digest=active_digest, - ) _write_redaction_record( record_path, operation_id=operation_id, @@ -2186,6 +2557,7 @@ def cleanup_redaction_quarantine( rewritten_sha256=rewritten_sha256, quarantine_retained=False, parent_operation_id=parent_operation_id, + parent_record_sha256=parent_record_sha256, rewritten_digest_retired_by=rewritten_digest_retired_by, ) _write_state( @@ -2201,6 +2573,12 @@ def cleanup_redaction_quarantine( parent_operation_id=parent_operation_id, rewritten_digest_retired_by=rewritten_digest_retired_by, ) + refreshed_digest = _refresh_chained_anchors( + resolved_run_dir, + workspace=workspace, + resumable_operation_id=operation_id, + ) + _post_replace_verify(resolved_run_dir, expected_digest=refreshed_digest) return RedactionReport( operation_id, start, diff --git a/tests/test_run_checkpoint.py b/tests/test_run_checkpoint.py index 4116bca2..7325da8d 100644 --- a/tests/test_run_checkpoint.py +++ b/tests/test_run_checkpoint.py @@ -1646,6 +1646,35 @@ def _journal_with_checkpoint_and_trailing( return checkpoint +def test_recover_from_checkpoint_accepts_trailing_redaction_anchor(tmp_path): + 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=None, + trailing_events=[ + ( + "run.redaction.recorded", + { + "operation_id": "redact-0123456789abcdef", + "affected_first_sequence": 1, + "affected_last_sequence": 1, + "reason_class": "credential-exposure", + "record_sha256": "a" * 64, + }, + "redaction-recorded-test", + "2026-07-27T15:30:46.000000Z", + ) + ], + ) + (run_dir / "run.json").unlink() + + assert run_checkpoint.recover_from_checkpoint(run_dir, None) == run_json_obj + + @pytest.mark.parametrize( ("payload", "pairing_seat", "pairing_attempt"), [ diff --git a/tests/test_run_redaction.py b/tests/test_run_redaction.py index c18a5805..ad6a28fb 100644 --- a/tests/test_run_redaction.py +++ b/tests/test_run_redaction.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import json import os import subprocess @@ -496,7 +497,7 @@ def test_redaction_quarantines_rewrites_rechains_and_reprojects(tmp_path): verified = run_journal.read_journal_bounded(_journal_path(run_dir)) assert verified.chain_errors == [] assert verified.partial_tail is None - assert len(verified.events) == len(original_events) + assert len(verified.events) == len(original_events) + 1 assert verified.events[1].payload == {"detail": "[REDACTED]"} assert verified.events[0].event_digest == original_events[0].event_digest assert verified.events[1].event_digest != original_events[1].event_digest @@ -505,10 +506,225 @@ def test_redaction_quarantines_rewrites_rechains_and_reprojects(tmp_path): current = json.loads((run_dir / "run.json").read_text()) after_projection = run_projector.project_run_snapshot(current, verified.events, journal_present=True).snapshot assert current == after_projection - assert _without_tail_digest(current) == _without_tail_digest(before_projection) + expected_projection = dict(before_projection) + expected_projection["journal_last_sequence"] = len(verified.events) + assert _without_tail_digest(current) == _without_tail_digest(expected_projection) assert current["journal_last_event_digest"] == verified.events[-1].event_digest +def test_overlapping_redactions_preserve_both_chained_operation_anchors(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + second = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=5, + reason=REASON_CODE, + operator_confirmed=True, + ) + + anchors = [ + event + for event in run_journal.read_journal_bounded(_journal_path(run_dir)).events + if event.event_type == "run.redaction.recorded" + ] + assert {anchor.payload["operation_id"] for anchor in anchors} == {first.operation_id, second.operation_id} + assert all( + set(anchor.payload) + == { + "operation_id", + "affected_first_sequence", + "affected_last_sequence", + "reason_class", + "record_sha256", + } + for anchor in anchors + ) + records = {report.operation_id: report.record_path.read_bytes() for report in (first, second)} + assert all( + anchor.payload["record_sha256"] == hashlib.sha256(records[anchor.payload["operation_id"]]).hexdigest() + for anchor in anchors + ) + second_record = json.loads(second.record_path.read_text()) + assert second_record["parent_operation_id"] == first.operation_id + + +def test_redaction_rejects_preserved_structural_only_range_without_mutation(tmp_path): + run_dir, _, events = _authority_run(tmp_path) + checkpoint = next(event for event in events if event.event_type == run_checkpoint.CHECKPOINT_EVENT_TYPE) + before = _artifact_snapshot(run_dir) + + with pytest.raises(run_redaction.RedactionError, match="redaction range contains no redactable payloads"): + run_redaction.redact_journal( + run_dir, + sequence_start=checkpoint.sequence, + sequence_end=checkpoint.sequence, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert _artifact_snapshot(run_dir) == before + assert not (run_dir / "events" / "redactions").exists() + + +def test_resume_projection_accepts_only_trailing_redaction_anchor_lag(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + events = run_journal.read_journal_bounded(_journal_path(run_dir)).events + projected_before_anchor = json.loads((run_dir / "run.json").read_text()) + projected_before_anchor["journal_last_sequence"] = events[-2].sequence + projected_before_anchor["journal_last_event_digest"] = events[-2].event_digest + (run_dir / "run.json").write_text(json.dumps(projected_before_anchor, indent=2, sort_keys=True) + "\n") + + run_redaction._resume_projection_after_rewrite(run_dir, events) + + assert ( + json.loads((run_dir / "run.json").read_text()) + == run_projector.project_run_snapshot(projected_before_anchor, events, journal_present=True).snapshot + ) + + +def test_resume_projection_rejects_non_anchor_sequence_lag(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + events = run_journal.read_journal_bounded(_journal_path(run_dir)).events + stale = json.loads((run_dir / "run.json").read_text()) + stale["journal_last_sequence"] = events[-3].sequence + stale["journal_last_event_digest"] = events[-3].event_digest + (run_dir / "run.json").write_text(json.dumps(stale, indent=2, sort_keys=True) + "\n") + + with pytest.raises(run_redaction.RedactionError, match="projection sequence lag"): + run_redaction._resume_projection_after_rewrite(run_dir, events) + + +def test_redaction_retry_appends_missing_anchor_after_verified_record(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + real_append = run_redaction._append_redaction_anchor + failed = False + + def fail_before_anchor(*args, **kwargs): + nonlocal failed + if not failed: + failed = True + raise run_redaction.RedactionError("simulated anchor append failure") + return real_append(*args, **kwargs) + + monkeypatch.setattr(run_redaction, "_append_redaction_anchor", fail_before_anchor) + with pytest.raises(run_redaction.RedactionError, match="anchor append"): + run_redaction.redact_journal( + run_dir, sequence_start=2, sequence_end=2, reason=REASON_CODE, operator_confirmed=True + ) + + monkeypatch.setattr(run_redaction, "_append_redaction_anchor", real_append) + report = run_redaction.redact_journal( + run_dir, sequence_start=2, sequence_end=2, reason=REASON_CODE, operator_confirmed=True + ) + events = run_journal.read_journal_bounded(_journal_path(run_dir)).events + anchors = [event for event in events if event.event_type == "run.redaction.recorded"] + assert [anchor.payload["operation_id"] for anchor in anchors] == [report.operation_id] + assert ( + json.loads((run_dir / "run.json").read_text()) + == run_projector.project_run_snapshot( + json.loads((run_dir / "run.json").read_text()), events, journal_present=True + ).snapshot + ) + + +def test_redaction_retry_reprojects_after_anchor_before_projection_failure(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + real_replace_projection = run_redaction._replace_projection + failed = False + + def fail_anchor_projection(path, projection): + nonlocal failed + if not failed and projection.snapshot["journal_last_sequence"] == 5: + failed = True + raise run_redaction.RedactionError("simulated anchor projection failure") + return real_replace_projection(path, projection) + + monkeypatch.setattr(run_redaction, "_replace_projection", fail_anchor_projection) + with pytest.raises(run_redaction.RedactionError, match="anchor projection"): + run_redaction.redact_journal( + run_dir, sequence_start=2, sequence_end=2, reason=REASON_CODE, operator_confirmed=True + ) + + monkeypatch.setattr(run_redaction, "_replace_projection", real_replace_projection) + report = run_redaction.redact_journal( + run_dir, sequence_start=2, sequence_end=2, reason=REASON_CODE, operator_confirmed=True + ) + events = run_journal.read_journal_bounded(_journal_path(run_dir)).events + anchors = [event for event in events if event.event_type == "run.redaction.recorded"] + assert [anchor.payload["operation_id"] for anchor in anchors] == [report.operation_id] + assert ( + json.loads((run_dir / "run.json").read_text()) + == run_projector.project_run_snapshot( + json.loads((run_dir / "run.json").read_text()), events, journal_present=True + ).snapshot + ) + + +def test_redaction_refuses_anchor_append_after_ownership_loss(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + real_assert_owner = run_redaction._assert_active_owner + calls = 0 + + def lose_owner(workspace, resolved_run_dir): + nonlocal calls + calls += 1 + if calls == 2: + raise run_redaction.RedactionError("redaction lost exclusive run lock ownership") + return real_assert_owner(workspace, resolved_run_dir) + + monkeypatch.setattr(run_redaction, "_assert_active_owner", lose_owner) + with pytest.raises(run_redaction.RedactionError, match="lost exclusive"): + run_redaction.redact_journal( + run_dir, sequence_start=2, sequence_end=2, reason=REASON_CODE, operator_confirmed=True + ) + + assert not [ + event + for event in run_journal.read_journal_bounded(_journal_path(run_dir)).events + if event.event_type == "run.redaction.recorded" + ] + + +def test_replaced_redaction_record_set_fails_chained_anchor_validation(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + record = json.loads(report.record_path.read_text()) + replacement = (json.dumps(record, separators=(",", ":")) + "\n").encode() + assert replacement != report.record_path.read_bytes() + report.record_path.write_bytes(replacement) + + with pytest.raises(run_redaction.RedactionError, match="anchor"): + run_redaction._post_replace_verify(run_dir, expected_digest=None) + + def test_redaction_refuses_live_lock_without_mutation(tmp_path): run_dir, _, _ = _authority_run(tmp_path) workspace = tmp_path / "workspace" @@ -1051,7 +1267,7 @@ def test_redacted_payloads_remain_valid_for_projection_sensitive_status_fields(t assert report.events[3].payload == {"status": "ok", "detail": "[REDACTED]"} current = json.loads((run_dir / "run.json").read_text()) assert current["status"] == before_projection["status"] == "ok" - assert current["journal_last_sequence"] == 4 + assert current["journal_last_sequence"] == 5 for event in report.events: assert run_events.validate_event(event.to_dict()) == [] @@ -1059,6 +1275,10 @@ def test_redacted_payloads_remain_valid_for_projection_sensitive_status_fields(t def test_two_operator_processes_cannot_publish_concurrent_rewrites(tmp_path): run_dir, _, _ = _authority_run(tmp_path) marker = tmp_path / "first-operator-inside-transaction" + child_env = dict(os.environ) + child_env["PYTHONPATH"] = os.pathsep.join( + filter(None, (str(Path(__file__).parents[1] / "src"), child_env.get("PYTHONPATH"))) + ) script = textwrap.dedent( """ import sys @@ -1098,12 +1318,15 @@ def paused_rewrite(*args, **kwargs): text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + env=child_env, ) for _ in range(100): if marker.exists(): break time.sleep(0.02) - assert marker.exists() + if not marker.exists(): + first_stdout, first_stderr = first.communicate(timeout=10) + pytest.fail(f"first operator never entered transaction\nstdout:\n{first_stdout}\nstderr:\n{first_stderr}") second = subprocess.run( [ @@ -1118,6 +1341,7 @@ def paused_rewrite(*args, **kwargs): text=True, capture_output=True, check=False, + env=child_env, ) first_stdout, first_stderr = first.communicate(timeout=10) @@ -1536,6 +1760,7 @@ def test_sequential_cleanup_removes_prior_rewritten_digest_alias_oracle(tmp_path first_state_path = first.record_path.parent / "state.json" first_state = json.loads(first_state_path.read_text()) first_rewritten_digest = first_state["rewritten_sha256"] + first_post_anchor_digest = hashlib.sha256(_journal_path(run_dir).read_bytes()).hexdigest() assert first_rewritten_digest in first.record_path.read_text() second = run_redaction.redact_journal( @@ -1546,7 +1771,14 @@ def test_sequential_cleanup_removes_prior_rewritten_digest_alias_oracle(tmp_path operator_confirmed=True, ) second_state = json.loads((second.record_path.parent / "state.json").read_text()) - assert second_state["original_sha256"] == first_rewritten_digest + assert second_state["original_sha256"] == first_post_anchor_digest + assert second_state["original_sha256"] != first_rewritten_digest + parent_anchor = next( + event + for event in run_journal.read_journal_bounded(_journal_path(run_dir)).events + if event.event_type == "run.redaction.recorded" and event.payload["operation_id"] == first.operation_id + ) + assert json.loads(second.record_path.read_text())["parent_record_sha256"] == parent_anchor.payload["record_sha256"] run_redaction.cleanup_redaction_quarantine( run_dir, @@ -1681,6 +1913,69 @@ def fail_child_cleaned_state(*args, **kwargs): assert second_original_digest.encode() not in path.read_bytes() +def test_redaction_after_cleaned_parent_uses_active_anchor_reference(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=first.operation_id, + operator_confirmed=True, + ) + parent_anchor = next( + event + for event in run_journal.read_journal_bounded(_journal_path(run_dir)).events + if event.event_type == "run.redaction.recorded" and event.payload["operation_id"] == first.operation_id + ) + + second = run_redaction.redact_journal( + run_dir, + sequence_start=4, + sequence_end=4, + reason="personal-data-exposure", + operator_confirmed=True, + ) + + second_record = json.loads(second.record_path.read_text()) + assert second_record["parent_operation_id"] == first.operation_id + assert second_record["parent_record_sha256"] == parent_anchor.payload["record_sha256"] + + +def test_redaction_inventory_rejects_tampered_parent_record_anchor_reference(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + second = run_redaction.redact_journal( + run_dir, + sequence_start=4, + sequence_end=4, + reason="personal-data-exposure", + operator_confirmed=True, + ) + second_record = json.loads(second.record_path.read_text()) + second_record["parent_record_sha256"] = "0" * 64 + second.record_path.write_text(json.dumps(second_record, indent=2, sort_keys=True) + "\n") + + with pytest.raises(run_redaction.RedactionError, match="parent reference"): + run_redaction.redact_journal( + run_dir, + sequence_start=3, + sequence_end=3, + reason="other-sensitive-data", + operator_confirmed=True, + ) + + def test_redaction_inventory_rejects_record_without_state(tmp_path): run_dir, _, _ = _authority_run(tmp_path) first = run_redaction.redact_journal( @@ -1724,6 +2019,7 @@ def test_redaction_inventory_rejects_forged_multiple_tip_graph(tmp_path): second_record = json.loads(second.record_path.read_text()) second_state["parent_operation_id"] = None second_record["parent_operation_id"] = None + second_record.pop("parent_record_sha256") second_state_path.write_text(json.dumps(second_state)) second.record_path.write_text(json.dumps(second_record)) From ff72dfa1c9f9a158b6070d496a1ddf8031fcf4f3 Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Fri, 31 Jul 2026 13:03:06 -0400 Subject: [PATCH 3/6] fix(io): fsync atomic write directories Co-Authored-By: Codex --- src/brigade/localio.py | 34 +++++++++++++++- tests/test_localio.py | 89 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+), 2 deletions(-) diff --git a/src/brigade/localio.py b/src/brigade/localio.py index 1c8ad090..9487c89c 100644 --- a/src/brigade/localio.py +++ b/src/brigade/localio.py @@ -11,6 +11,7 @@ import json import os import re +import stat import subprocess import tempfile from datetime import datetime, timezone @@ -47,8 +48,10 @@ def write_text_atomic(path: Path, data: str) -> None: The write goes to a temp file in the same directory and is swapped in with os.replace, so a reader (or a crashed writer) never observes a half-written - file: it sees either the old file or the complete new one. On failure the - temp file is removed and the existing file is left untouched. + file: it sees either the old file or the complete new one. On failure before + replacement the temp file is removed and the existing file is left untouched. + A directory-fsync failure occurs after replacement and means durable + publication is unconfirmed, although the new bytes are already present. """ path.parent.mkdir(parents=True, exist_ok=True) fd, tmp_name = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp") @@ -62,6 +65,33 @@ def write_text_atomic(path: Path, data: str) -> None: except BaseException: tmp_path.unlink(missing_ok=True) raise + _fsync_parent_directory(path.parent) + + +def _fsync_parent_directory(path: Path) -> None: + """Durably publish a replacement on platforms that support directory fsync.""" + if not _supports_directory_fsync(): + return + flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) + fd = os.open(path, flags) + primary: BaseException | None = None + try: + if not stat.S_ISDIR(os.fstat(fd).st_mode): + raise OSError("atomic-write parent is not a directory") + os.fsync(fd) + except BaseException as exc: + primary = exc + raise + finally: + try: + os.close(fd) + except BaseException: + if primary is None: + raise + + +def _supports_directory_fsync() -> bool: + return os.name == "posix" def write_bytes_atomic(path: Path, data: bytes) -> None: diff --git a/tests/test_localio.py b/tests/test_localio.py index 865aecfe..5bff3b9a 100644 --- a/tests/test_localio.py +++ b/tests/test_localio.py @@ -3,6 +3,8 @@ from __future__ import annotations import json +import os +import stat from pathlib import Path import pytest @@ -45,6 +47,93 @@ def _boom(src, dst): assert leftovers == [] +@pytest.mark.skipif(os.name != "posix", reason="directory fsync is a POSIX durability guarantee") +def test_write_text_atomic_fsyncs_parent_after_replace(tmp_path: Path, monkeypatch): + path = tmp_path / "receipt.txt" + calls: list[tuple[str, bool]] = [] + real_replace = localio.os.replace + real_fsync = localio.os.fsync + + def tracking_replace(source, destination): + calls.append(("replace", False)) + return real_replace(source, destination) + + def tracking_fsync(fd): + info = os.fstat(fd) + calls.append(("fsync", stat.S_ISDIR(info.st_mode))) + return real_fsync(fd) + + monkeypatch.setattr(localio.os, "replace", tracking_replace) + monkeypatch.setattr(localio.os, "fsync", tracking_fsync) + + localio.write_text_atomic(path, "durable\n") + + replace_index = calls.index(("replace", False)) + assert any(kind == "fsync" and is_directory for kind, is_directory in calls[replace_index + 1 :]) + + +@pytest.mark.skipif( + os.name != "posix" or not hasattr(os, "O_NOFOLLOW"), reason="O_NOFOLLOW is a POSIX symlink hardening flag" +) +def test_write_text_atomic_opens_parent_without_following_symlinks(tmp_path: Path, monkeypatch): + path = tmp_path / "receipt.txt" + real_open = localio.os.open + flags_seen: list[int] = [] + + def tracking_open(target, flags, *args, **kwargs): + if target == path.parent: + flags_seen.append(flags) + return real_open(target, flags, *args, **kwargs) + + monkeypatch.setattr(localio.os, "open", tracking_open) + + localio.write_text_atomic(path, "durable\n") + + assert flags_seen and flags_seen[-1] & os.O_NOFOLLOW + + +@pytest.mark.skipif(os.name != "posix", reason="directory fsync is a POSIX durability guarantee") +def test_write_text_atomic_reports_directory_fsync_failure_after_replace(tmp_path: Path, monkeypatch): + path = tmp_path / "receipt.txt" + real_fsync = localio.os.fsync + real_close = localio.os.close + + def fail_directory_fsync(fd): + if stat.S_ISDIR(os.fstat(fd).st_mode): + raise OSError("simulated directory fsync failure") + return real_fsync(fd) + + def also_fail_directory_close(fd): + if stat.S_ISDIR(os.fstat(fd).st_mode): + raise OSError("simulated directory close failure") + return real_close(fd) + + monkeypatch.setattr(localio.os, "fsync", fail_directory_fsync) + monkeypatch.setattr(localio.os, "close", also_fail_directory_close) + + with pytest.raises(OSError, match="simulated directory fsync failure"): + localio.write_text_atomic(path, "published\n") + + assert path.read_text() == "published\n" + + +def test_write_text_atomic_skips_parent_fsync_when_platform_has_no_support(tmp_path: Path, monkeypatch): + path = tmp_path / "receipt.txt" + monkeypatch.setattr(localio, "_supports_directory_fsync", lambda: False) + real_fsync = localio.os.fsync + + def reject_directory_fsync(fd): + if stat.S_ISDIR(os.fstat(fd).st_mode): + raise AssertionError("directory fsync should be skipped") + return real_fsync(fd) + + monkeypatch.setattr(localio.os, "fsync", reject_directory_fsync) + + localio.write_text_atomic(path, "portable\n") + + assert path.read_text() == "portable\n" + + def test_canonical_json_digest_excludes_top_level_keys_and_hashes_files(tmp_path: Path): payload = { "b": 2, From 744ee3c871e00088e774b995fbabf00ca25a55f3 Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Fri, 31 Jul 2026 13:03:16 -0400 Subject: [PATCH 4/6] fix(journal): serialize partial-tail recovery Co-Authored-By: Codex --- src/brigade/run_journal.py | 8 ++++++-- tests/test_run_journal.py | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/src/brigade/run_journal.py b/src/brigade/run_journal.py index 4d7c4066..bfb530eb 100644 --- a/src/brigade/run_journal.py +++ b/src/brigade/run_journal.py @@ -923,8 +923,12 @@ def recover_partial_tail(journal_path: Path, quarantine_dir: Path) -> RecoveryRe and ``quarantine_path`` is None. Normal readers must never call this; it is the only API that mutates the journal body. """ - journal_path = Path(journal_path) - quarantine_dir = Path(quarantine_dir) + with _append_critical_section(): + return _recover_partial_tail_locked(Path(journal_path), Path(quarantine_dir)) + + +def _recover_partial_tail_locked(journal_path: Path, quarantine_dir: Path) -> RecoveryReport: + """Perform recovery while the append critical section is held.""" if os.path.lexists(journal_path) and not journal_path.exists(): raise RunJournalError(_bound(f"journal path is a dangling symlink: {journal_path.name}")) if not journal_path.exists(): diff --git a/tests/test_run_journal.py b/tests/test_run_journal.py index cbf892a2..55e4fb5a 100644 --- a/tests/test_run_journal.py +++ b/tests/test_run_journal.py @@ -1564,6 +1564,42 @@ def nested(): assert rc == 0, f"child exited {rc}\nstdout:\n{stdout}\nstderr:\n{stderr}" +def test_recover_partial_tail_reentrancy_matches_append_event(tmp_path): + """Recovery must fail fast when signal delivery recursively enters it.""" + script = """ +from pathlib import Path +from brigade import run_journal + +journal_path = Path(__import__("sys").argv[1]) +quarantine_dir = Path(__import__("sys").argv[2]) +run_journal._HAS_PTHREAD_SIGMASK = False +original = run_journal._read_bytes_nofollow +entered = False +def recurse(path): + global entered + data = original(path) + if not entered: + entered = True + run_journal.recover_partial_tail(journal_path, quarantine_dir) + return data +run_journal._read_bytes_nofollow = recurse +try: + run_journal.recover_partial_tail(journal_path, quarantine_dir) +except run_journal.RunJournalError as exc: + if "recursive append" not in exc.diagnostic: + raise SystemExit(2) + raise SystemExit(0) +raise SystemExit(1) +""" + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + journal_path.write_bytes(journal_path.read_bytes() + b"partial") + + rc, stdout, stderr = _run_isolated_child(script, str(journal_path), str(tmp_path / "quarantine"), timeout=2.0) + + assert rc == 0, f"child exited {rc}\nstdout:\n{stdout}\nstderr:\n{stderr}" + + def test_append_event_process_lock_prevents_worker_main_sigterm_duplicate_in_subprocess(tmp_path): """Worker-thread append holds the process lock while SIGTERM runs on the main thread: the handler blocks until the worker finishes, then sees a fresh tail From b81ed9ac3a20b9c21494ad77a9ca1601198d4832 Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Fri, 31 Jul 2026 15:05:26 -0400 Subject: [PATCH 5/6] fix(io): support symlinked atomic-write parents Co-Authored-By: Codex --- src/brigade/localio.py | 2 +- tests/test_localio.py | 29 ++++++++++++++++++++++++++++- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/src/brigade/localio.py b/src/brigade/localio.py index 9487c89c..a00a6ec6 100644 --- a/src/brigade/localio.py +++ b/src/brigade/localio.py @@ -73,7 +73,7 @@ def _fsync_parent_directory(path: Path) -> None: if not _supports_directory_fsync(): return flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) - fd = os.open(path, flags) + fd = os.open(path.resolve(), flags) primary: BaseException | None = None try: if not stat.S_ISDIR(os.fstat(fd).st_mode): diff --git a/tests/test_localio.py b/tests/test_localio.py index 5bff3b9a..ff4a22a4 100644 --- a/tests/test_localio.py +++ b/tests/test_localio.py @@ -9,7 +9,7 @@ import pytest -from brigade import localio +from brigade import localio, run_journal def test_read_json_dict_invalid_utf8_returns_none(tmp_path: Path): @@ -92,6 +92,33 @@ def tracking_open(target, flags, *args, **kwargs): assert flags_seen and flags_seen[-1] & os.O_NOFOLLOW +@pytest.mark.skipif( + os.name != "posix" or not hasattr(os, "O_NOFOLLOW"), reason="O_NOFOLLOW is a POSIX symlink hardening flag" +) +def test_write_text_atomic_allows_directory_symlink_parent(tmp_path: Path): + real_parent = tmp_path / "real-parent" + real_parent.mkdir() + symlinked_parent = tmp_path / "symlinked-parent" + symlinked_parent.symlink_to(real_parent, target_is_directory=True) + path = symlinked_parent / "receipt.txt" + + localio.write_text_atomic(path, "durable through symlink\n") + + assert path.read_text() == "durable through symlink\n" + assert (real_parent / "receipt.txt").read_text() == "durable through symlink\n" + + +@pytest.mark.skipif(os.name != "posix", reason="directory fsync is a POSIX durability guarantee") +def test_run_journal_fsync_directory_refuses_directory_symlink(tmp_path: Path): + real_directory = tmp_path / "real-directory" + real_directory.mkdir() + symlinked_directory = tmp_path / "symlinked-directory" + symlinked_directory.symlink_to(real_directory, target_is_directory=True) + + with pytest.raises(run_journal.RunJournalError, match="refusing symlinked path"): + run_journal._fsync_directory(symlinked_directory) + + @pytest.mark.skipif(os.name != "posix", reason="directory fsync is a POSIX durability guarantee") def test_write_text_atomic_reports_directory_fsync_failure_after_replace(tmp_path: Path, monkeypatch): path = tmp_path / "receipt.txt" From 54bce3db55193c1be3a62f9510fcfa6ec35c0e9b Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Fri, 31 Jul 2026 15:05:36 -0400 Subject: [PATCH 6/6] test(redaction): pin overlapping anchor preservation Co-Authored-By: Codex --- tests/test_run_redaction.py | 33 ++++++++++++++++++++------------- 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/tests/test_run_redaction.py b/tests/test_run_redaction.py index ad6a28fb..20bb49f5 100644 --- a/tests/test_run_redaction.py +++ b/tests/test_run_redaction.py @@ -512,7 +512,7 @@ def test_redaction_quarantines_rewrites_rechains_and_reprojects(tmp_path): assert current["journal_last_event_digest"] == verified.events[-1].event_digest -def test_overlapping_redactions_preserve_both_chained_operation_anchors(tmp_path): +def test_redaction_preserves_first_anchor_when_second_range_contains_it(tmp_path): run_dir, _, _ = _authority_run(tmp_path) first = run_redaction.redact_journal( @@ -522,19 +522,27 @@ def test_overlapping_redactions_preserve_both_chained_operation_anchors(tmp_path reason=REASON_CODE, operator_confirmed=True, ) + first_anchor = next( + event + for event in run_journal.read_journal_bounded(_journal_path(run_dir)).events + if event.event_type == "run.redaction.recorded" and event.payload["operation_id"] == first.operation_id + ) + second_sequence_start = 2 + second_sequence_end = first_anchor.sequence + assert second_sequence_start <= first_anchor.sequence <= second_sequence_end + second = run_redaction.redact_journal( run_dir, - sequence_start=2, - sequence_end=5, + sequence_start=second_sequence_start, + sequence_end=second_sequence_end, reason=REASON_CODE, operator_confirmed=True, ) - anchors = [ - event - for event in run_journal.read_journal_bounded(_journal_path(run_dir)).events - if event.event_type == "run.redaction.recorded" - ] + verified = run_journal.read_journal_bounded(_journal_path(run_dir)) + assert verified.chain_errors == [] + assert verified.partial_tail is None + anchors = [event for event in verified.events if event.event_type == "run.redaction.recorded"] assert {anchor.payload["operation_id"] for anchor in anchors} == {first.operation_id, second.operation_id} assert all( set(anchor.payload) @@ -547,11 +555,10 @@ def test_overlapping_redactions_preserve_both_chained_operation_anchors(tmp_path } for anchor in anchors ) - records = {report.operation_id: report.record_path.read_bytes() for report in (first, second)} - assert all( - anchor.payload["record_sha256"] == hashlib.sha256(records[anchor.payload["operation_id"]]).hexdigest() - for anchor in anchors - ) + record_hashes = { + report.operation_id: hashlib.sha256(report.record_path.read_bytes()).hexdigest() for report in (first, second) + } + assert {anchor.payload["operation_id"]: anchor.payload["record_sha256"] for anchor in anchors} == record_hashes second_record = json.loads(second.record_path.read_text()) assert second_record["parent_operation_id"] == first.operation_id