fix(open): pending-embed recovery keeps the crash-recovery contract — foreground, bounded by the mark

The delta gate caught the backgrounded fold breaking six pinned
crash-recovery cases: a reopened brain must have its markers re-armed
when open() returns, and a background latch races every consumer of that
contract. The backgrounding is reverted; the low-water mark stays — it
is the part that kills the whole-history scan, and with it the
foreground fold costs the log's tail on any brain that has ever drained.
The unmarked first open after upgrade pays one full scan, once, and the
open narrates it as its own step.

src/brainy.ts

@@ -1820,16 +1820,22 @@ export class Brainy<T = any> implements BrainyInterface<T> {
       // a deferred write's ack and its background embed DELAYED a vector;
       // this is where it lands.
       if (!this.isReadOnly) {
-        // BEHIND THE DOORS (the open pays nothing here): the bridge + the
-        // recovery fold run as one latched background task; the embed worker
-        // starts when it settles. A pending embed's outcome was always
-        // eventual — moving its recovery off the open's foreground changes
-        // when the worker starts, never whether a marker is honored.
-        // awaitPendingEmbeds() and close() wait on the latch first.
-        this._pendingEmbedRecovery = (async () => {
+        // Foreground, as the crash-recovery contract pins it: a reopened brain
+        // has its markers re-armed when open() returns. The low-water mark
+        // bounds this to the log's tail on any brain that has ever drained —
+        // milliseconds — so the foreground cost is the unmarked first open
+        // only, once per upgraded brain.
         try {
-            await this.bridgeLegacyPendingEmbedSidecars()
-            await this.recoverPendingEmbedsFromLog()
+          await step(
+            'bridge-pending-embed-sidecars',
+            'migrating any pre-log deferred-embed marker files into the generation log',
+            () => this.bridgeLegacyPendingEmbedSidecars()
+          )
+          await step(
+            'recover-pending-embeds',
+            'folding the generation log\'s deferred-embed markers (from the low-water mark) into the pending set',
+            () => this.recoverPendingEmbedsFromLog()
+          )
           if (this._pendingEmbedIds.size > 0) {
             prodLog.info(
               `[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` +
@@ -1844,7 +1850,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
               `the log's markers remain durable; recovery retries next open`
           )
         }
-        })()
       }
 
       // PHASE 4 of 5 — "VFS bootstrap": shutdown-hook registration, blob
@@ -2418,8 +2423,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    */
   private static readonly PENDING_EMBED_LOWWATER_PATH = '_system/pending_embeds_lowwater.json'
 
-  /** Resolves when the background pending-embed recovery fold has settled (open arms it). */
-  private _pendingEmbedRecovery: Promise<void> | null = null
 
   /**
    * @description Mark a deferred embed pending (MT5): the id joins the
@@ -2497,9 +2500,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    * pending set last drained to empty — so a settled brain reads only the
    * facts since then, not its whole history. Without a mark (first open
    * after upgrade) it scans from generation 1, once; a stale-low mark costs
-   * a longer scan, never a marker. The fold runs BEHIND the doors (open
-   * arms it as a background task and the embed worker starts when it
-   * settles); {@link awaitPendingEmbeds} and close() wait for it first.
+   * a longer scan, never a marker. The fold stays on the open's foreground —
+   * the crash-recovery contract pins that a reopened brain has its markers
+   * re-armed when open() returns — and the mark is what makes that cheap.
    * It is SKIPPED WHOLESALE when the log has never had a v2 tail
    * ({@link FactLog.hasV2History} — v1 facts cannot carry marker records),
    * so pre-cutover brains pay nothing; on a mixed log the scan still reads
@@ -2710,7 +2713,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    * before I proceed" callers use this; nothing else ever needs to wait.
    */
   public async awaitPendingEmbeds(): Promise<void> {
-    if (this._pendingEmbedRecovery) await this._pendingEmbedRecovery
     while (this._pendingEmbedIds.size > 0 || this._embedWorkerFlight) {
       this.kickEmbedWorker()
       await (this._embedWorkerFlight ?? Promise.resolve())
@@ -19507,18 +19509,6 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    *   terminal releases have run.
    */
   async close(): Promise<void> {
-    if (this._pendingEmbedRecovery) {
-      // Settle the background marker fold before the durable steps — its scan
-      // is bounded by the low-water mark (a full scan happens at most once,
-      // on the first open after upgrade).
-      const settleStart = Date.now()
-      await this._pendingEmbedRecovery
-      const settleMs = Date.now() - settleStart
-      if (settleMs >= 1000) {
-        prodLog.info(`[Brainy] close: pending-embed recovery settled in ${settleMs}ms`)
-      }
-      this._pendingEmbedRecovery = null
-    }
     if (this._pendingEmbedIds.size === 0) await this.writeEmbedLowWater()
     let closeFailure: unknown = null
     try {

tests/integration/pending-embed-low-water.test.ts

@@ -6,9 +6,8 @@
  * on the open's foreground — O(whole history) per open on long-lived brains.
  * Now: an advisory low-water mark (`_system/pending_embeds_lowwater.json`)
  * records the committed generation whenever the pending set drains to empty,
- * recovery scans from `mark + 1`, and the fold runs behind the doors as a
- * latched background task the worker, `awaitPendingEmbeds()` and `close()`
- * wait on. The mark is advisory: stale-low costs a longer scan, never a
+ * recovery scans from `mark + 1` on the open's foreground — the crash-recovery
+ * contract keeps markers re-armed when open() returns. The mark is advisory: stale-low costs a longer scan, never a
  * marker — a pending embed enqueued before a crash is still recovered.
  */
 import { describe, it, expect, afterEach, vi } from 'vitest'
@@ -20,7 +19,7 @@ import { NounType } from '../../src/types/graphTypes'
 
 const LOWWATER_PATH = '_system/pending_embeds_lowwater.json'
 
-describe('pending-embed recovery: bounded by the low-water mark, behind the doors', () => {
+describe('pending-embed recovery: bounded by the low-water mark', () => {
   const roots: string[] = []
   const dir = (): string => {
     const d = mkdtempSync(join(tmpdir(), 'brainy-lowwater-'))
@@ -100,7 +99,6 @@ describe('pending-embed recovery: bounded by the low-water mark, behind the door
     await (brain as any).storage.releaseWriterLock()
 
     const brain2 = await open(root)
-    await (brain2 as any)._pendingEmbedRecovery
     expect(brain2.pendingEmbedCount()).toBeGreaterThan(0)
     await brain2.awaitPendingEmbeds()
     expect(brain2.pendingEmbedCount()).toBe(0)
@@ -110,7 +108,7 @@ describe('pending-embed recovery: bounded by the low-water mark, behind the door
     await brain.close().catch(() => undefined)
   })
 
-  it('open arms the fold as a background latch; awaitPendingEmbeds waits on it', async () => {
+  it('a reopened brain has its pending set settled when open() returns', async () => {
     const root = dir()
     const brain = await open(root)
     await brain.add({ id: 'a-row', data: 'some data', type: NounType.Thing })
@@ -118,10 +116,8 @@ describe('pending-embed recovery: bounded by the low-water mark, behind the door
     await brain.close()
 
     const brain2 = await open(root)
-    // The latch exists the moment init() returns (writable filesystem brain)…
-    expect((brain2 as any)._pendingEmbedRecovery).not.toBeNull()
-    // …and the barrier settles it before answering.
-    await brain2.awaitPendingEmbeds()
+    // The crash-recovery contract: markers are re-armed by open itself —
+    // no latch, no background race. (Here the drain landed, so zero.)
     expect(brain2.pendingEmbedCount()).toBe(0)
     await brain2.close()
   })