fix(durability): three block-layer power-loss findings from the first fault-injection box run — all cured, matrix 15/15

An internal cross-engine fault-injection run (frozen-platter power-loss
capture) surfaced three release-gating findings; each cured in its owning
layer, each pinned:

1. WHOLE-LOG REPLAY ON UNCLEAN OPEN (the big one): log-authority replay
   only covered facts ABOVE the manifest — but live canonical entity
   writes are tmp+rename without per-file fsync, and the group-commit
   flush syncs staging + manifest, never the live tree. Power loss could
   therefore vaporize acked canonical bytes BELOW the manifest while the
   log held every fact scan-clean (measured: 299 of 301 acks lost).
   Now: a clean close stamps a clean-shutdown marker (fsynced, written
   last); every open consumes it; an UNCLEAN open under log authority
   folds the ENTIRE log into canonical — whole-entity after-images make
   the re-apply idempotent and byte-safe. Zero cost on the happy path;
   crash recovery pays one narrated fold. Recovery is replay: a crash is
   just bigger lag.

2. TORN WRITER LOCK: power loss legally leaves the lock file present but
   empty; the parse failure read as 'no holder' while the O_EXCL claim
   EEXISTed forever — a PERMANENT lockout no staleness check could clear.
   An unparseable lock is stale by definition (no live holder has one):
   unlink loudly and re-loop; a racer rewriting a valid lock first wins.

3. PAIR GUARD: flush() called metadataIndex.stampWatermark unguarded;
   a replacement metadata provider without the method killed the pair at
   first flush. All three stamp calls are optional-chained — a missing
   stamp is a verdict-side rescan, never a flush crash.

Pins: whole-log fold restores rows vanished below the manifest ·
clean-shutdown marker lifecycle (stamp/consume/re-stamp) · torn-lock
recovery with a fresh write after · stampless-provider flush.
Gates: unit 2055/2055 · integration 824 · kill-matrix 15/15.

tests/integration/durability-kill-matrix.test.ts

@@ -37,6 +37,7 @@
  */
 import { describe, it, expect, afterEach } from 'vitest'
 import * as fs from 'node:fs'
+import { join } from 'node:path'
 import { Brainy } from '../../src/brainy.js'
 import { NounType } from '../../src/types/graphTypes.js'
 import {
@@ -630,4 +631,71 @@ describe('durability kill matrix — crash at every commit-path step, recover by
     expect(storeOf(brain).committedGeneration()).toBe(floor)
     expect(await factGenerations(brain)).toEqual([floor])
   })
+
+  // ==========================================================================
+  // Block-layer power-loss findings (first dm-flakey run) — the three cures
+  // ==========================================================================
+
+  it('at-ack POWER LOSS BELOW THE MANIFEST — an unclean open folds the WHOLE log; acked writes committed before the flush still survive vanished canonical', async () => {
+    const { dir, brain, baselineId } = await arrangeBaseline('wlf')
+    await flipToAtAck(brain)
+    const ackedA = uid('wlf-a')
+    const ackedB = uid('wlf-b')
+    await brain.add({ id: ackedA, data: 'below manifest one', type: NounType.Document, vector: vec(2), metadata: { v: 2 } })
+    await brain.add({ id: ackedB, data: 'below manifest two', type: NounType.Document, vector: vec(3), metadata: { v: 3 } })
+    // The group-commit flush advances the manifest OVER these generations —
+    // but live canonical bytes are tmp+rename without per-file fsync, so a
+    // power cut can still take them. The fsynced facts are the durable copy.
+    await (brain as unknown as { flush(): Promise<void> }).flush()
+    await abandonAsCrashed(brain) // no clean close → no clean-shutdown marker
+    dropCanonicalNoun(dir, ackedA)
+    dropCanonicalNoun(dir, ackedB)
+
+    const reopened = await openLive(dir)
+    // The whole-log fold restores BOTH rows from facts ≤ manifest.
+    expect(((await reopened.get(ackedA)) as { metadata: { v: number } }).metadata.v).toBe(2)
+    expect(((await reopened.get(ackedB)) as { metadata: { v: number } }).metadata.v).toBe(3)
+    expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1)
+  })
+
+  it('clean-shutdown marker: a clean close writes it, the next open consumes it (no fold on the happy path)', async () => {
+    const { dir, brain } = await arrangeBaseline('csm')
+    await flipToAtAck(brain)
+    await brain.close()
+    liveBrains.splice(liveBrains.indexOf(brain), 1)
+    // The adapter stores raw objects gzipped — accept either spelling.
+    const markerExists = () =>
+      fs.existsSync(join(dir, '_system', 'clean-shutdown.json')) ||
+      fs.existsSync(join(dir, '_system', 'clean-shutdown.json.gz'))
+    expect(markerExists(), 'clean close stamps the marker').toBe(true)
+
+    const reopened = await openLive(dir)
+    expect(markerExists(), 'open consumes the marker').toBe(false)
+    await reopened.close()
+    liveBrains.splice(liveBrains.indexOf(reopened), 1)
+    expect(markerExists(), 'the next clean close re-stamps it').toBe(true)
+  })
+
+  it('torn writer lock (empty file) — open treats it as stale and recovers; never a permanent lockout', async () => {
+    const { dir, brain } = await arrangeBaseline('tlk')
+    await brain.close()
+    liveBrains.splice(liveBrains.indexOf(brain), 1)
+    // The power-loss shape: the lock file exists but is EMPTY (torn write).
+    fs.writeFileSync(join(dir, 'locks', '_writer.lock'), '')
+
+    const reopened = await openLive(dir) // must not throw 'contended'
+    const fresh = uid('tlk-fresh')
+    await reopened.add({ id: fresh, data: 'lock recovered', type: NounType.Document, vector: vec(4), metadata: { v: 4 } })
+    expect(await reopened.get(fresh)).not.toBeNull()
+  })
+
+  it('pair guard: a metadata index without stampWatermark never crashes flush', async () => {
+    const { brain } = await arrangeBaseline('psg')
+    liveBrains.push(brain)
+    // The native pair swaps the metadata manager; the replacement may not
+    // carry the stamp method — flush must treat that as verdict-side rescan,
+    // never a TypeError at the fan-out.
+    ;(brain as unknown as { metadataIndex: { stampWatermark?: unknown } }).metadataIndex.stampWatermark = undefined
+    await expect((brain as unknown as { flush(): Promise<void> }).flush()).resolves.toBeUndefined()
+  })
 })