fix(journal): close run-journal hardening gaps

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,

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.resolve(), 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:

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

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 = {

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():

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",
     }
 )