fix: harden archival state boundaries

src/ledger/attest.py

@@ -34,6 +34,7 @@
 import os
 from pathlib import Path
 
+from ledger._filelock import file_lock
 from ledger.dualcontrol import ActionProposal, ProposalStore
 from ledger.metadata.premis import PremisLog
 from ledger.models import PremisEvent, PremisEventType, now_iso
@@ -51,6 +52,7 @@
 _PROPOSALS_FILENAME = "attestations.json"
 _ATTESTED_FILENAME = "attested-conditions.json"
 _PREMIS_FILENAME = "attestations.premis.json"
+_WORKFLOW_LOCK_FILENAME = "attestations.workflow"
 
 
 def _read_attested(path: Path) -> frozenset[str]:
@@ -134,12 +136,17 @@ def approve(
         exactly on the approval that tipped the proposal over the threshold — the
         approval that writes the condition into the attested set and records the
         ``POLICY_CHANGE`` event."""
-        proposal = self._proposals.approve(proposal_id, steward)
-        if not proposal.is_ready(threshold):
-            return proposal, False
-        self._record_attested(proposal, agent=steward, now=now or now_iso())
-        self._proposals.mark(proposal.proposal_id, "executed")
-        return proposal, True
+        # Serialize approval through terminal marking, not just each individual
+        # JSON rewrite. Otherwise two stewards can both observe the same proposal
+        # as ready before either marks it executed, recording the outcome twice.
+        workflow_lock = self._dir / _WORKFLOW_LOCK_FILENAME
+        with file_lock(workflow_lock):
+            proposal = self._proposals.approve(proposal_id, steward)
+            if not proposal.is_ready(threshold):
+                return proposal, False
+            attested_now = self._record_attested(proposal, agent=steward, now=now or now_iso())
+            self._proposals.mark(proposal.proposal_id, "executed")
+            return proposal, attested_now
 
     # --- reads --------------------------------------------------------------
 
@@ -157,15 +164,24 @@ def attested(self) -> frozenset[str]:
 
     # --- durable outcome ----------------------------------------------------
 
-    def _record_attested(self, proposal: ActionProposal, *, agent: str, now: str) -> None:
+    def _record_attested(self, proposal: ActionProposal, *, agent: str, now: str) -> bool:
         """Add the condition to the attested set and append a PREMIS ``POLICY_CHANGE``.
 
         The set is the machine-readable authority the access layer consults; the
         append-only PREMIS event is the accountable, replica-checkable *why*. Both are
-        no-outing safe — a condition string plus steward ids, never an identity."""
+        no-outing safe — a condition string plus steward ids, never an identity.
+
+        The attested-set read-modify-write is locked (:func:`ledger._filelock.file_lock`)
+        so two approvals that each cross the threshold for a *different* condition at
+        nearly the same moment cannot race and silently drop one (a lost attestation is
+        as serious a bug here as a lost consent withdrawal elsewhere in the archive)."""
         condition = proposal.target
         attested_path = self._dir / _ATTESTED_FILENAME
-        _write_attested(attested_path, _read_attested(attested_path) | {condition})
+        with file_lock(attested_path):
+            prior = _read_attested(attested_path)
+            attested_now = condition not in prior
+            if attested_now:
+                _write_attested(attested_path, prior | {condition})
 
         event = PremisEvent(
             event_type=PremisEventType.POLICY_CHANGE,
@@ -181,6 +197,19 @@ def _record_attested(self, proposal: ActionProposal, *, agent: str, now: str) ->
             event_datetime=now,
         )
         log_path = self._dir / _PREMIS_FILENAME
-        log = PremisLog.read(log_path) if log_path.exists() else PremisLog()
-        log.record(event)
-        log.write(log_path)
+        # PREMIS is another whole-file read-modify-write and needs its own stable
+        # sibling lock. Also make this append idempotent by condition: if the set
+        # write succeeded but the log write failed, retry repairs the missing event;
+        # if both succeeded but terminal marking failed, retry never duplicates it.
+        event_prefix = f"condition attested and now met: {condition} ("
+        with file_lock(log_path):
+            log = PremisLog.read(log_path) if log_path.exists() else PremisLog()
+            already_logged = any(
+                candidate.event_type is PremisEventType.POLICY_CHANGE
+                and candidate.detail.startswith(event_prefix)
+                for candidate in log.events
+            )
+            if not already_logged:
+                log.record(event)
+                log.write(log_path)
+        return attested_now

src/ledger/bag.py

@@ -26,7 +26,7 @@
 import shutil
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
-from pathlib import Path, PurePosixPath
+from pathlib import Path, PurePosixPath, PureWindowsPath
 
 from ledger.errors import BagValidationError
 from ledger.fixity import CHUNK_SIZE, AuditReport, hash_file, hash_file_multi, verify_file
@@ -49,12 +49,17 @@ def _reject_unsafe_relpath(relpath: str, *, context: str) -> None:
     or hash files outside the bag (a path-traversal vulnerability). Validation is
     purely lexical, so it holds before any file is touched (securability, safety).
     """
-    pure = PurePosixPath(relpath)
+    posix = PurePosixPath(relpath)
+    windows = PureWindowsPath(relpath)
     if (
         not relpath
         or relpath.startswith("/")
-        or pure.is_absolute()
-        or ".." in pure.parts
+        or "\\" in relpath
+        or posix.is_absolute()
+        or windows.is_absolute()
+        or bool(windows.drive)
+        or ".." in posix.parts
+        or ".." in windows.parts
         or "\x00" in relpath
     ):
         raise BagValidationError(f"unsafe path in {context}: {relpath!r}")