fix(lab): serialize CL-02 ledger mutations

src/lab/fabric/observe.ts

@@ -12,7 +12,7 @@ import { fixtureDigest, isSha256Hex, jcsStringify } from "../digest";
 import type { ObservationEvent, RouteSubjectV1, TaskSubjectV1 } from "../events/types";
 import { LabValidationError } from "../events/errors";
 import { assignEventId, validateSubject } from "../events/validate";
-import { appendLabEventIfAbsent } from "../ledger/store";
+import { withLedgerMutation } from "../ledger/store";
 import { ensureLabDirs } from "../paths";
 import {
   FABRIC_EVIDENCE_LAYER,
@@ -430,9 +430,11 @@ function persistFabricOutcome(
   const ownsStore = !opts.artifactStore;
   const store = opts.artifactStore ?? createArtifactStore(paths.artifactsDir);
   try {
-    const { event } = observationFromFabricOutcome(outcome, { ...opts, artifactStore: store });
-    appendLabEventIfAbsent(paths.ledgerPath, event);
-    return { event, ledgerPath: paths.ledgerPath };
+    return withLedgerMutation(paths.ledgerPath, (ledger) => {
+      const { event } = observationFromFabricOutcome(outcome, { ...opts, artifactStore: store });
+      ledger.appendIfAbsent(event);
+      return { event, ledgerPath: paths.ledgerPath };
+    });
   } finally {
     if (ownsStore) store.close();
   }

src/lab/ledger/purge.ts

@@ -18,7 +18,7 @@ import {
   expandSensitiveArtifactEventTargets,
 } from "./artifact-refs";
 import { buildInvalidationIndex } from "./invalidation";
-import { appendLabEvent, replayLabLedger } from "./store";
+import { withLedgerMutation } from "./store";
 import { ensureLabDirs } from "../paths";
 import { rebuildLabProjection } from "../projection/rebuild";
 import { jcsStringify } from "../digest";
@@ -153,73 +153,86 @@ export function purgeSensitiveEvidence(req: SensitivePurgeRequest): PurgeTombsto
   const targetArtifactDigests = [...(req.targetArtifactDigests ?? [])].sort();
   const purgeActions = [...(req.purgeActions ?? PURGE_ACTIONS)].sort();
   const explicitSensitive = new Set(targetArtifactDigests);
-
-  const replay = replayLabLedger(paths.ledgerPath);
-  const index = buildInvalidationIndex(replay.events);
-  const removeIds = expandSensitiveArtifactEventTargets(
-    replay.events,
-    index,
-    new Set(targetEventIds),
-    explicitSensitive,
-  );
-
-  const tombstonePayload = {
-    schemaVersion: LAB_EVENT_SCHEMA_VERSION,
-    eventKind: "purge_tombstone" as const,
-    recordedAt: req.recordedAt ?? Date.now(),
-    producer: LAB_PRODUCER,
-    producerVersion: req.producerVersion ?? LAB_PRODUCER_VERSION,
-    targetEventIds: [...removeIds].sort(),
-    targetArtifactDigests,
-    reason: "sensitive_evidence" as const,
-    purgeActions,
-  };
-  const tombstone = validateLabEvent(assignEventId(tombstonePayload)) as PurgeTombstoneEvent;
-
-  const deletionPlan = purgeActions.includes("artifact")
-    ? artifactDeletionPlan(replay.events, index, removeIds, targetArtifactDigests)
-    : { deletable: [], retainedExplicit: [] };
-
-  if (deletionPlan.retainedExplicit.length > 0) {
-    throw new PurgeError(
-      "sensitive_bytes_retained",
-      `explicit sensitive artifacts remain required: ${deletionPlan.retainedExplicit.join(",")}`,
-    );
-  }
-
-  let dir: TrustedArtifactDir | null = null;
   const completed: string[] = [];
+
   try {
-    if (purgeActions.includes("scratch")) {
-      purgeBoundedDirectory(paths.scratchDir);
-      completed.push("scratch");
-    }
-    if (purgeActions.includes("export")) {
-      purgeBoundedDirectory(paths.exportDir);
-      completed.push("export");
-    }
+    const tombstone = withLedgerMutation(paths.ledgerPath, (ledger) => {
+      // Replay and plan under the same lock as every append. Otherwise an event
+      // appended after this snapshot can be lost by the atomic rename or can
+      // start referencing an artifact after the deletion plan was calculated.
+      const replay = ledger.replay();
+      const index = buildInvalidationIndex(replay.events);
+      const removeIds = expandSensitiveArtifactEventTargets(
+        replay.events,
+        index,
+        new Set(targetEventIds),
+        explicitSensitive,
+      );
 
-    if (purgeActions.includes("artifact")) {
-      if (deletionPlan.deletable.length > 0) {
-        dir = openTrustedArtifactDir(paths.artifactsDir);
-        deleteArtifactsFailClosed(dir, deletionPlan.deletable);
+      const tombstonePayload = {
+        schemaVersion: LAB_EVENT_SCHEMA_VERSION,
+        eventKind: "purge_tombstone" as const,
+        recordedAt: req.recordedAt ?? Date.now(),
+        producer: LAB_PRODUCER,
+        producerVersion: req.producerVersion ?? LAB_PRODUCER_VERSION,
+        targetEventIds: [...removeIds].sort(),
+        targetArtifactDigests,
+        reason: "sensitive_evidence" as const,
+        purgeActions,
+      };
+      const tombstone = validateLabEvent(assignEventId(tombstonePayload)) as PurgeTombstoneEvent;
+
+      const deletionPlan = purgeActions.includes("artifact")
+        ? artifactDeletionPlan(replay.events, index, removeIds, targetArtifactDigests)
+        : { deletable: [], retainedExplicit: [] };
+
+      if (deletionPlan.retainedExplicit.length > 0) {
+        throw new PurgeError(
+          "sensitive_bytes_retained",
+          `explicit sensitive artifacts remain required: ${deletionPlan.retainedExplicit.join(",")}`,
+        );
       }
-      completed.push("artifact");
-    }
 
-    if (purgeActions.includes("ledger")) {
-      const kept: LabEvent[] = [];
-      for (const event of replay.events) {
-        if (removeIds.has(event.eventId)) continue;
-        kept.push(event);
+      if (purgeActions.includes("scratch")) {
+        purgeBoundedDirectory(paths.scratchDir);
+        completed.push("scratch");
       }
-      kept.push(tombstone);
-      atomicRewriteLedger(paths.ledgerPath, kept);
-    } else {
-      appendLabEvent(paths.ledgerPath, tombstone);
-    }
-    completed.push("ledger");
+      if (purgeActions.includes("export")) {
+        purgeBoundedDirectory(paths.exportDir);
+        completed.push("export");
+      }
+
+      let dir: TrustedArtifactDir | null = null;
+      try {
+        if (purgeActions.includes("artifact")) {
+          if (deletionPlan.deletable.length > 0) {
+            dir = openTrustedArtifactDir(paths.artifactsDir);
+            deleteArtifactsFailClosed(dir, deletionPlan.deletable);
+          }
+          completed.push("artifact");
+        }
+
+        if (purgeActions.includes("ledger")) {
+          const kept: LabEvent[] = [];
+          for (const event of replay.events) {
+            if (removeIds.has(event.eventId)) continue;
+            kept.push(event);
+          }
+          kept.push(tombstone);
+          atomicRewriteLedger(paths.ledgerPath, kept);
+        } else {
+          ledger.append(tombstone);
+        }
+        completed.push("ledger");
+      } finally {
+        if (dir) closeTrustedArtifactDir(dir);
+      }
+
+      return tombstone;
+    });
 
+    // SQLite is disposable and rebuildLabProjection replays the canonical
+    // ledger again, so it does not need to extend the mutation lock duration.
     if (purgeActions.includes("sqlite")) {
       rebuildLabProjection(req.configDir);
       completed.push("sqlite");
@@ -235,7 +248,5 @@ export function purgeSensitiveEvidence(req: SensitivePurgeRequest): PurgeTombsto
       err instanceof Error ? err.message : String(err),
       completed,
     );
-  } finally {
-    if (dir) closeTrustedArtifactDir(dir);
   }
 }

src/lab/ledger/store.ts

@@ -25,6 +25,12 @@ export interface LedgerStore {
   replay(): ReplayResult;
 }
 
+export interface LedgerMutationContext {
+  replay(): ReplayResult;
+  append(event: LabEvent): void;
+  appendIfAbsent(event: LabEvent): boolean;
+}
+
 const LEDGER_LOCK_STALE_MS = 60_000;
 const LEDGER_LOCK_WAIT_MS = 5_000;
 
@@ -36,10 +42,7 @@ interface LedgerLockMeta {
 
 /** Block synchronously for the given duration (ledger lock retry only). */
 function sleepSyncMs(ms: number): void {
-  const end = Date.now() + ms;
-  while (Date.now() < end) {
-    /* spin */
-  }
+  Bun.sleepSync(ms);
 }
 
 /** Read pid, createdAt, and token metadata from a ledger lock file, if well-formed. */
@@ -83,8 +86,7 @@ function isLedgerLockStale(lockPath: string): boolean {
       return false;
     }
   }
-  if (isLockHolderAlive(meta.pid)) return false;
-  return Date.now() - meta.createdAt > LEDGER_LOCK_STALE_MS;
+  return !isLockHolderAlive(meta.pid);
 }
 
 /** Create a ledger lock file exclusively, recovering stale locks when needed. */
@@ -156,11 +158,10 @@ function withLedgerLock<T>(ledgerPath: string, fn: () => T): T {
   }
 }
 
-/** Durable append of one validated event as a single JSONL line + fsync. */
-export function appendLabEvent(ledgerPath: string, event: LabEvent): void {
-  const validated = validateLabEvent(event);
+/** Durable append of one already-validated event as a single JSONL line + fsync. */
+function appendValidatedLabEvent(ledgerPath: string, event: LabEvent): void {
   mkdirSync(dirname(ledgerPath), { recursive: true, mode: 0o700 });
-  const line = `${jcsStringify(validated)}\n`;
+  const line = `${jcsStringify(event)}\n`;
   const bytes = new TextEncoder().encode(line);
   const fd = openSync(ledgerPath, "a", 0o600);
   try {
@@ -179,25 +180,45 @@ export function appendLabEvent(ledgerPath: string, event: LabEvent): void {
 }
 
 /**
- * Append only when eventId is absent. Uses an exclusive lock file plus a
- * process-local event-id index refreshed under the lock.
+ * Serialize a ledger read-modify-write transaction with all ordinary appends.
+ * The callback receives lock-aware operations so callers that also publish
+ * artifacts can keep artifact writes and the corresponding event atomic with
+ * respect to sensitive purge.
  */
-export function appendLabEventIfAbsent(ledgerPath: string, event: LabEvent): boolean {
-  const validated = validateLabEvent(event);
+export function withLedgerMutation<T>(
+  ledgerPath: string,
+  fn: (mutation: LedgerMutationContext) => T,
+): T {
   return withLedgerLock(ledgerPath, () => {
-    // Refresh from disk under the lock so concurrent writers are visible.
-    const fresh = new Set<string>();
-    if (existsSync(ledgerPath)) {
-      for (const row of replayLabLedger(ledgerPath).events) {
-        fresh.add(row.eventId);
-      }
-    }
-    if (fresh.has(validated.eventId)) return false;
-    appendLabEvent(ledgerPath, validated);
-    return true;
+    const replay = () => replayLabLedger(ledgerPath);
+    const append = (event: LabEvent) => appendValidatedLabEvent(ledgerPath, validateLabEvent(event));
+    const appendIfAbsent = (event: LabEvent): boolean => {
+      const validated = validateLabEvent(event);
+      if (replay().events.some((row) => row.eventId === validated.eventId)) return false;
+      appendValidatedLabEvent(ledgerPath, validated);
+      return true;
+    };
+    return fn({ replay, append, appendIfAbsent });
+  });
+}
+
+/** Durable append of one validated event as a single JSONL line + fsync. */
+export function appendLabEvent(ledgerPath: string, event: LabEvent): void {
+  const validated = validateLabEvent(event);
+  withLedgerMutation(ledgerPath, (mutation) => {
+    mutation.append(validated);
   });
 }
 
+/**
+ * Append only when eventId is absent. Uses the same mutation lock as every
+ * other ledger writer so the presence check and append are one transaction.
+ */
+export function appendLabEventIfAbsent(ledgerPath: string, event: LabEvent): boolean {
+  const validated = validateLabEvent(event);
+  return withLedgerMutation(ledgerPath, (mutation) => mutation.appendIfAbsent(validated));
+}
+
 function processLine(
   line: string,
   lineNumber: number,

src/lab/observe/from-conformance.ts

@@ -17,7 +17,7 @@ import {
 } from "../digest";
 import type { ObservationEvent } from "../events/types";
 import { assignEventId } from "../events/validate";
-import { appendLabEvent } from "../ledger/store";
+import { withLedgerMutation } from "../ledger/store";
 import { ensureLabDirs } from "../paths";
 import type {
   CaseAuthority,
@@ -285,12 +285,14 @@ export function persistConformanceResult(
   const ownsStore = !opts.artifactStore;
   const store = opts.artifactStore ?? createArtifactStore(paths.artifactsDir);
   try {
-    const { event } = observationFromConformanceResult(result, caseRecord, authority, {
-      ...opts,
-      artifactStore: store,
+    return withLedgerMutation(paths.ledgerPath, (ledger) => {
+      const { event } = observationFromConformanceResult(result, caseRecord, authority, {
+        ...opts,
+        artifactStore: store,
+      });
+      ledger.append(event);
+      return { event, ledgerPath: paths.ledgerPath };
     });
-    appendLabEvent(paths.ledgerPath, event);
-    return { event, ledgerPath: paths.ledgerPath };
   } finally {
     if (ownsStore) store.close();
   }

src/lab/observe/from-live.ts

@@ -5,7 +5,7 @@ import { LAB_EVENT_SCHEMA_VERSION, LAB_PRODUCER, LAB_PRODUCER_VERSION, OBSERVATI
 import { fixtureDigest, scenarioManifestDigest, subjectIdForSubject, suiteManifestDigest } from "../digest";
 import type { FailureRecordV1, ObservationEvent } from "../events/types";
 import { assignEventId } from "../events/validate";
-import { appendLabEvent } from "../ledger/store";
+import { withLedgerMutation } from "../ledger/store";
 import { ensureLabDirs } from "../paths";
 import type { CaseAuthority, CaseRecord } from "../conformance/types";
 import { trustedLiveResultRetryable } from "../live/executor";
@@ -106,6 +106,12 @@ export function observationFromLiveResult(result: LiveScenarioRunResult, caseRec
 
 export function persistLiveResult(result: LiveScenarioRunResult, caseRecord: CaseRecord, authority: CaseAuthority, opts: PersistLiveOptions = {}): PersistedLiveObservation {
   const paths = ensureLabDirs(opts.configDir); const ownsStore = !opts.artifactStore; const store = opts.artifactStore ?? createArtifactStore(paths.artifactsDir);
-  try { const { event } = observationFromLiveResult(result, caseRecord, authority, { ...opts, artifactStore: store }); appendLabEvent(paths.ledgerPath, event); return { event, ledgerPath: paths.ledgerPath }; }
+  try {
+    return withLedgerMutation(paths.ledgerPath, (ledger) => {
+      const { event } = observationFromLiveResult(result, caseRecord, authority, { ...opts, artifactStore: store });
+      ledger.append(event);
+      return { event, ledgerPath: paths.ledgerPath };
+    });
+  }
   finally { if (ownsStore) store.close(); }
 }