feat(log): system commits carry their origin; the attested per-id reconcile door

Two consumer-driven cures sharing one stamp. (1) TX-LOG ORIGIN: engine-
originated commits stamp an optional origin on their tx-log entry AND the
commit fact's meta — 'system:embed-landing' (the deferred vector landing),
'system:adoption-backfill' (baseline re-commits), 'system:reconcile'. A
downstream activity feed showed a double tick because the landing commit was
indistinguishable from a user save, and the consumer rightly refused a
time-window collapse as a quiet loss; feeds now filter on fact. User writes
stay unstamped — absent origin is the user shape, every existing consumer
unchanged. (2) reconcileLogDivergence(id, {attest}): the human's door for
log-live-canonical-absent, the one class adoption refuses by design because
a lost-tombstone deletion is indistinguishable from canonical loss.
'deleted' mints the missing tombstone (history keeps the earlier live
record); 'restore' folds the log's only copy back into canonical; wrong-
class calls refuse typed with nothing written. Loud, narrated, single-row,
origin-stamped. From a production adoption's one surviving divergence.

src/brainy.ts

@@ -2245,7 +2245,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
           },
           undefined,
           undefined,
-          [{ type: 'embed.landed', id, vector: newVector }]
+          [{ type: 'embed.landed', id, vector: newVector }],
+          'system:embed-landing'
         )
         this.clearPendingEmbed(id)
       } catch (err) {
@@ -2479,7 +2480,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
     run: TransactionFunction<void>,
     precommit?: (before: CommitBeforeImages) => void,
     pendingEvents?: PendingChangeEvent[],
-    records?: FactMarkerRecord[]
+    records?: FactMarkerRecord[],
+    origin?: string
   ): Promise<{ generation?: number; timestamp: number; degraded?: string[] }> {
     // Change-feed capture: when this write will emit, hold a reference to the
     // commit's before-images so `remove` events can carry the record's last
@@ -2542,6 +2544,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
         touched,
         precommit: captureAndCheck,
         ...(records && records.length > 0 ? { records } : {}),
+        ...(origin ? { origin } : {}),
         execute: () =>
           this.transactionManager.executeTransaction(run, {
             timeout: transactTimeoutBudget(
@@ -8358,7 +8361,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
               }
             }
           })
-        })
+        }, undefined, undefined, undefined, 'system:adoption-backfill')
       }
       const next = await this.runOracle({ listAll: true })
       // THE ONLY STOP: no progress. With uncapped listings both counts are
@@ -8392,6 +8395,137 @@ export class Brainy<T = any> implements BrainyInterface<T> {
     return report
   }
 
+  /**
+   * @description THE ATTESTED PER-ID RECONCILE DOOR for the one divergence
+   * class the adoption backfill refuses BY DESIGN: `log-live-canonical-absent`
+   * — the log holds a live record for a row the canonical tree says does not
+   * exist. The engine cannot tell a legitimate pre-log deletion (the log
+   * missed the tombstone — the deferred-durability-era ack-window class) from
+   * canonical LOSS (the log holds the only surviving copy); auto-curing would
+   * silently destroy data in one of the two readings. A HUMAN attests which:
+   *
+   * - `attest: 'deleted'` — the row was legitimately deleted; mint the
+   *   tombstone fact the log always lacked (canonical stays absent). The
+   *   log's history keeps the old live record — as-of reads before the
+   *   tombstone still see it.
+   * - `attest: 'restore'` — canonical lost the row; fold the log's latest
+   *   after-image back into canonical (both sides now agree it lives).
+   *
+   * Loud, narrated, single-row, and stamped `origin: 'system:reconcile'` on
+   * both the tx-log entry and the commit fact. Refuses (typed) when the id's
+   * log and canonical already agree, when `restore` is attested but the log
+   * holds no record, and when canonical is PRESENT-but-different (that is
+   * `state-differs` — `adoptLogAuthority()`'s backfill owns it).
+   *
+   * @param id - The single entity id to reconcile.
+   * @param options.attest - The human's word on which reading is true.
+   * @returns What was done and the generation that recorded it.
+   * @throws When the divergence is not the attested class (nothing is written).
+   */
+  async reconcileLogDivergence(
+    id: string,
+    options: { attest: 'deleted' | 'restore' }
+  ): Promise<{ reconciled: 'tombstoned' | 'restored'; id: string; generation: number }> {
+    await this.ensureInitialized()
+    this.assertWritable('reconcileLogDivergence')
+
+    // Fold the log for THIS id (one scan; a rare operator door).
+    const scan = this.scanFacts()
+    if (!scan) {
+      throw new Error('reconcileLogDivergence: this store has no fact log — nothing to reconcile against')
+    }
+    let logLatest: { tombstoned: boolean; record: { metadata: unknown; vector: unknown } | null } | null = null
+    for await (const batch of scan.batches()) {
+      for (const fact of batch.facts) {
+        for (const op of fact.ops) {
+          if (op.kind === 'noun' && op.id === id) {
+            logLatest =
+              op.record === null
+                ? { tombstoned: true, record: null }
+                : { tombstoned: false, record: { metadata: op.record.metadata, vector: op.record.vector } }
+          }
+        }
+      }
+    }
+    const canonical = await this.storage.readNounRaw(id)
+    const canonicalAbsent = canonical.metadata === null && canonical.vector === null
+
+    // Only the log-live + canonical-absent shape passes; everything else
+    // names its actual state and the door that owns it.
+    if (!logLatest || logLatest.tombstoned) {
+      throw new Error(
+        `reconcileLogDivergence(${id}): the log's latest state is ` +
+          `${logLatest ? 'a tombstone' : 'no record at all'} — there is no ` +
+          `log-live-canonical-absent divergence here. If the oracle reports this id, ` +
+          `re-run verifyLogAuthority() for the current class.`
+      )
+    }
+    if (!canonicalAbsent) {
+      throw new Error(
+        `reconcileLogDivergence(${id}): canonical is PRESENT — this is not the ` +
+          `log-live-canonical-absent class. If canonical differs from the log ` +
+          `(state-differs), adoptLogAuthority()'s backfill cures it; nothing was written.`
+      )
+    }
+
+    if (options.attest === 'deleted') {
+      // Mint the tombstone fact the log always lacked. writeNounRaw with null
+      // parts is an idempotent delete; the commit fact reads canonical back
+      // after execute (absent) and records the tombstone.
+      const receipt = await this.persistSingleOp(
+        { nouns: [id] },
+        async (tx) => {
+          tx.addOperation({
+            name: 'ReconcileTombstone',
+            execute: async () => {
+              await this.storage.writeNounRaw(id, { metadata: null, vector: null })
+              return async () => {
+                // Undo of an idempotent delete of an absent row: nothing.
+              }
+            }
+          })
+        },
+        undefined,
+        undefined,
+        undefined,
+        'system:reconcile'
+      )
+      prodLog.warn(
+        `[Brainy] reconcileLogDivergence: ${id} attested DELETED — tombstone fact minted ` +
+          `at generation ${receipt.generation}; the log now agrees the row is gone ` +
+          `(its history keeps the earlier live record).`
+      )
+      return { reconciled: 'tombstoned', id, generation: receipt.generation! }
+    }
+
+    // attest: 'restore' — the log's copy is the survivor; fold it back.
+    const record = logLatest.record!
+    const receipt = await this.persistSingleOp(
+      { nouns: [id] },
+      async (tx) => {
+        tx.addOperation({
+          name: 'ReconcileRestore',
+          execute: async () => {
+            await this.storage.writeNounRaw(id, record)
+            return async () => {
+              await this.storage.writeNounRaw(id, { metadata: null, vector: null })
+            }
+          }
+        })
+      },
+      undefined,
+      undefined,
+      undefined,
+      'system:reconcile'
+    )
+    prodLog.warn(
+      `[Brainy] reconcileLogDivergence: ${id} attested RESTORE — the log's latest ` +
+        `after-image was folded back into canonical at generation ${receipt.generation}. ` +
+        `Derived indexes reconcile at next open/repairIndex; the row serves from canonical now.`
+    )
+    return { reconciled: 'restored', id, generation: receipt.generation! }
+  }
+
   /**
    * @description Read the reified transaction log — one entry per committed
    * generation, carrying the committed generation, the commit timestamp, and

src/db/generationStore.ts

@@ -428,7 +428,13 @@ export class GenerationStore {
   private pendingGens: number[] = []
   private readonly pendingBuffer = new Map<
     number,
-    { nouns: Map<string, GenerationRecord>; verbs: Map<string, GenerationRecord>; timestamp: number }
+    {
+      nouns: Map<string, GenerationRecord>
+      verbs: Map<string, GenerationRecord>
+      timestamp: number
+      /** Engine-origin stamp for the tx-log entry (absent = user write). */
+      origin?: string
+    }
   >()
   /** Pending timer-coalesced flush handle (cleared on flush/close). */
   private pendingFlushTimer: ReturnType<typeof setTimeout> | null = null
@@ -1642,6 +1648,12 @@ export class GenerationStore {
      * surfacing that honestly.
      */
     records?: FactMarkerRecord[]
+    /**
+     * Engine-origin stamp (`'system:embed-landing'`, `'system:adoption-backfill'`,
+     * `'system:reconcile'`). Rides the tx-log entry AND the commit fact's meta,
+     * so both records agree about WHO committed. Absent = user write.
+     */
+    origin?: string
   }): Promise<{ generation: number; timestamp: number; degraded?: string[] }> {
     return this.withMutex(async () => {
       // Refuse to accept a write whose history we cannot make durable: if the
@@ -1710,7 +1722,7 @@ export class GenerationStore {
             // incomplete for these ids until the next rebuild/repairIndex (the
             // egress guard prevents wrong results meanwhile). Loud, honest,
             // no double-write.
-            this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp })
+            this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp, ...(args.origin ? { origin: args.origin } : {}) })
             this.pendingGens.push(gen)
             this.extendChains(gen, nouns, verbs)
             // The adopted generation is committed — it gets its fact like any
@@ -1723,6 +1735,7 @@ export class GenerationStore {
                   timestamp,
                   nouns,
                   verbs,
+                  ...(args.origin ? { meta: { origin: args.origin } } : {}),
                   ...(args.records && args.records.length > 0 ? { records: args.records } : {})
                 })
               )
@@ -1763,7 +1776,7 @@ export class GenerationStore {
       if (this.commitFaultInjector) this.commitFaultInjector('singleop-after-execute')
 
       // Buffer the pending generation + make it instantly visible to reads.
-      this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp })
+      this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp, ...(args.origin ? { origin: args.origin } : {}) })
       this.pendingGens.push(gen)
       this.extendChains(gen, nouns, verbs)
       // Fact log (dual-write): the acked write's AFTER-IMAGE fact, appended
@@ -1802,6 +1815,7 @@ export class GenerationStore {
               timestamp,
               nouns,
               verbs,
+              ...(args.origin ? { meta: { origin: args.origin } } : {}),
               ...(args.records && args.records.length > 0 ? { records: args.records } : {})
             })
           )
@@ -1958,7 +1972,7 @@ export class GenerationStore {
         const deltaPath = `${dir}/tx.json`
         await this.storage.writeRawObject(deltaPath, delta)
         stagedPaths.push(deltaPath)
-        logEntries.push({ generation: gen, timestamp: buf.timestamp })
+        logEntries.push({ generation: gen, timestamp: buf.timestamp, ...(buf.origin ? { origin: buf.origin } : {}) })
       }
 
       // Test-only crash simulation. A crash here must cost only the window's

src/db/types.ts

@@ -412,6 +412,17 @@ export interface TxLogEntry {
   timestamp: number
   /** Transaction metadata, when supplied to `transact()`. */
   meta?: Record<string, unknown>
+  /**
+   * WHO committed. Absent = a user write (every pre-existing consumer's
+   * reading stays exact). Engine-originated commits stamp themselves —
+   * `'system:embed-landing'` (the deferred vector landing),
+   * `'system:adoption-backfill'` (baseline re-commits), `'system:reconcile'`
+   * (the attested per-id divergence door) — so activity feeds can filter on
+   * fact instead of collapsing near-in-time entries (a consumer refused that
+   * heuristic as a quiet loss, correctly; this field is the honest cure).
+   * The same stamp rides the commit fact's meta, so log and tx-log agree.
+   */
+  origin?: string
 }
 
 // ============================================================================

tests/integration/txlog-origin-and-reconcile.test.ts

@@ -0,0 +1,142 @@
+/**
+ * @module tests/integration/txlog-origin-and-reconcile
+ * @description Two consumer-driven cures, pinned together because they share
+ * the origin stamp:
+ *
+ * 1. TX-LOG ORIGIN — engine-originated commits stamp `origin` on their
+ *    tx-log entry (and the commit fact's meta) so activity feeds filter on
+ *    fact: a downstream feed showed a "double tick" because the deferred
+ *    vector-landing commit was indistinguishable from a user save, and the
+ *    consumer rightly refused a time-window collapse as a quiet loss. User
+ *    writes stay UNSTAMPED (absent origin) — the pre-existing reading of
+ *    every consumer is exact.
+ *
+ * 2. THE RECONCILE DOOR — `log-live-canonical-absent` refuses auto-cure by
+ *    design (a legitimate lost-tombstone deletion is indistinguishable from
+ *    canonical loss); `reconcileLogDivergence(id, {attest})` is the human's
+ *    door: 'deleted' mints the missing tombstone, 'restore' folds the log's
+ *    copy back, wrong-class calls refuse typed with nothing written.
+ */
+import { describe, it, expect, afterEach } from 'vitest'
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { Brainy } from '../../src/index.js'
+import { NounType } from '../../src/types/graphTypes.js'
+
+type RawBox = {
+  storage: {
+    readNounRaw(id: string): Promise<{ metadata: unknown; vector: unknown }>
+    writeNounRaw(id: string, r: { metadata: unknown; vector: unknown }): Promise<void>
+  }
+}
+
+const dirs: string[] = []
+const brains: Brainy[] = []
+afterEach(async () => {
+  for (const b of brains.splice(0)) await b.close().catch(() => {})
+  for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
+})
+
+async function fsBrain(): Promise<Brainy> {
+  const dir = mkdtempSync(join(tmpdir(), 'brainy-origin-reconcile-'))
+  dirs.push(dir)
+  const brain = new Brainy({
+    storage: { type: 'filesystem', path: dir },
+    requireSubtype: false
+  })
+  await brain.init()
+  brains.push(brain)
+  return brain
+}
+
+describe('tx-log origin stamp', () => {
+  it('the deferred-embed landing commit is stamped system:embed-landing; the user write is not', async () => {
+    const brain = await fsBrain()
+    await brain.add({
+      data: 'a row whose vector lands later',
+      type: NounType.Document,
+      metadata: { k: 1 },
+      deferEmbedding: true
+    })
+    await brain.awaitPendingEmbeds()
+    await brain.flush()
+
+    const entries = await brain.transactionLog()
+    const system = entries.filter((e) => (e as { origin?: string }).origin === 'system:embed-landing')
+    const user = entries.filter((e) => !(e as { origin?: string }).origin)
+    expect(system.length, 'the landing commit is stamped').toBeGreaterThanOrEqual(1)
+    expect(user.length, 'the user add stays unstamped').toBeGreaterThanOrEqual(1)
+    // The feed cure in one line: filtering !origin removes the double tick.
+    expect(user.length).toBeLessThan(entries.length)
+  }, 120000)
+})
+
+describe('reconcileLogDivergence — the attested door', () => {
+  /** Manufacture the class: a live log record whose canonical row is gone. */
+  async function manufactureDivergence(brain: Brainy): Promise<string> {
+    const id = await brain.add({
+      data: 'pre-era row whose deletion the log never saw',
+      type: NounType.Document,
+      metadata: { era: 'pre-spine' }
+    })
+    await brain.flush()
+    // Delete canonical BEHIND the log's back (raw write, no generation) —
+    // exactly the shape a deferred-durability-era crash left behind.
+    const storage = (brain as unknown as RawBox).storage
+    await storage.writeNounRaw(id, { metadata: null, vector: null })
+    return id
+  }
+
+  it("attest:'deleted' mints the missing tombstone — the oracle goes green and the commit is stamped system:reconcile", async () => {
+    const brain = await fsBrain()
+    const id = await manufactureDivergence(brain)
+    const before = await brain.verifyLogAuthority()
+    expect(
+      before.mismatches.some((m) => m.id === id && m.reason === 'log-live-canonical-absent'),
+      'the manufactured divergence is oracle-visible as the refused class'
+    ).toBe(true)
+
+    const result = await brain.reconcileLogDivergence(id, { attest: 'deleted' })
+    expect(result.reconciled).toBe('tombstoned')
+
+    const after = await brain.verifyLogAuthority()
+    expect(after.mismatches.some((m) => m.id === id), 'the id no longer diverges').toBe(false)
+    expect(await brain.get(id), 'canonical stays absent').toBeNull()
+
+    await brain.flush()
+    const entries = await brain.transactionLog()
+    expect(
+      entries.some((e) => (e as { origin?: string }).origin === 'system:reconcile'),
+      'the reconcile commit is origin-stamped'
+    ).toBe(true)
+  }, 120000)
+
+  it("attest:'restore' folds the log's copy back into canonical", async () => {
+    const brain = await fsBrain()
+    const id = await manufactureDivergence(brain)
+
+    const result = await brain.reconcileLogDivergence(id, { attest: 'restore' })
+    expect(result.reconciled).toBe('restored')
+
+    const row = await brain.get(id)
+    expect(row, 'the log’s only copy lives again').not.toBeNull()
+    expect((row!.metadata as { era: string }).era).toBe('pre-spine')
+    expect((await brain.verifyLogAuthority()).mismatches.some((m) => m.id === id)).toBe(false)
+  }, 120000)
+
+  it('wrong-class calls refuse typed with nothing written', async () => {
+    const brain = await fsBrain()
+    const id = await brain.add({ data: 'healthy row', type: NounType.Document, metadata: { n: 1 } })
+    await brain.flush()
+    // Canonical present + log agrees: not the class — refuse, name the state.
+    await expect(brain.reconcileLogDivergence(id, { attest: 'deleted' })).rejects.toThrow(
+      /canonical is PRESENT/
+    )
+    expect(await brain.get(id), 'nothing was written').not.toBeNull()
+    // Unknown id: no log record at all — refuse, name it.
+    await expect(
+      brain.reconcileLogDivergence('00000000-0000-7000-8000-00000000dead', { attest: 'restore' })
+    ).rejects.toThrow(/no record at all/)
+  }, 120000)
+})