feat(index): watermark stamps on every TS projection — adopt/catchup/rescan verdicts at load, stamp-after-data

Every persisted projection artifact (metadata field indexes + column
segments, HNSW node records, graph adjacency LSM trees) now carries a
stamp asserting 'this state reflects every committed generation ≤ W,
atomically' — written LAST in each owner's flush (stamp-after-data: a
crash between data and stamp = unstamped = rescan, never trust). At load,
each owner computes the three-way verdict: stamped==committed → adopt
(zero work) · behind → catchup (gap reported) · above/unstamped → RESCAN,
loudly. Legacy artifacts re-derive once, then are stamped forever. Shared
law in projectionWatermark.ts (the aggregation verdict machinery,
generalized); vector artifacts carry model dimensions. Verdicts are
computed and exposed (watermark()/watermarkVerdict()/watermarkGap());
rebuild triggers unchanged — acting on 'catchup' is the fold train.

Pins: 22 unit (7 metadata · 8 hnsw · 7 graph, incl. spy-order
stamp-after-data) + the end-to-end reopen-adopts pin.

src/graph/graphAdjacencyIndex.ts

@@ -27,6 +27,22 @@ import { UnifiedCache, getGlobalCache } from '../utils/unifiedCache.js'
 import { prodLog } from '../utils/logger.js'
 import { LSMTree } from './lsm/LSMTree.js'
 import type { GraphIndexProvider } from '../plugin.js'
+import {
+  computeWatermarkVerdict,
+  makeProjectionStamp,
+  readStampedWatermark,
+  type WatermarkVerdict,
+  type WatermarkVerdictResult
+} from '../utils/projectionWatermark.js'
+
+/**
+ * Storage key for the graph-adjacency projection's watermark stamp — a
+ * sidecar record beside the artifact (the two verb-id LSM trees' persisted
+ * SSTables + manifests). Written LAST in
+ * {@link GraphAdjacencyIndex.flush} / {@link GraphAdjacencyIndex.close} so
+ * stamp-after-data ordering holds for every byte the stamp certifies.
+ */
+export const GRAPH_ADJACENCY_STAMP_KEY = '__index_graph_adjacency_watermark__'
 
 export interface GraphIndexConfig {
   maxIndexSize?: number      // Default: 100000
@@ -112,6 +128,14 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
   // Initialization flag
   private initialized = false
 
+  // --- Watermark stamp state (see utils/projectionWatermark for the law) ---
+  /** Generation handed in via {@link stampWatermark}, awaiting the next flush. */
+  private pendingWatermark: number | null = null
+  /** Last watermark durably stamped by this instance or loaded at init. */
+  private stampedWatermark: number | null = null
+  /** The three-way verdict computed at init; null until init runs. */
+  private loadVerdict: WatermarkVerdictResult | null = null
+
   /**
    * Check if index is initialized and ready for use
    */
@@ -241,12 +265,135 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
       await this.populateVerbIdSetFromStorage()
     }
 
+    // Watermark verdict for the persisted adjacency artifact (the LSM
+    // SSTables just loaded) — computed and exposed only: today's rebuild /
+    // recovery triggers are unchanged (acting on 'catchup' — the incremental
+    // fold — lands with the coordinator's wiring).
+    await this.loadWatermarkVerdict(lsmTreeSize > 0)
+
     // Start auto-flush timer after initialization
     this.startAutoFlush()
 
     this.initialized = true
   }
 
+  /**
+   * @description Record the committed generation this projection reflects.
+   * The stamp is NOT written here — it is written as the final storage write
+   * of the next {@link flush} (or {@link close}), so stamp-after-data
+   * ordering is a module guarantee, not a caller obligation. The coordinator
+   * calls this with the store's committed generation right before flushing.
+   * @param generation - The committed generation every flushed byte reflects.
+   */
+  stampWatermark(generation: number): void {
+    this.pendingWatermark = generation
+  }
+
+  /**
+   * @description The projection's current watermark: the stamp loaded at
+   * init (or the last stamp durably written by this instance). Null =
+   * unstamped (legacy artifact, first boot, or stamping never wired).
+   */
+  watermark(): number | null {
+    return this.stampedWatermark
+  }
+
+  /**
+   * @description The three-way adoption verdict computed at init —
+   * `'adopt'` (stamped == committed, zero work), `'catchup'` (stamped <
+   * committed; the gap from {@link watermarkGap} awaits an incremental
+   * fold), `'rescan'` (unstamped or stamped above committed — never
+   * trusted). Null until init() has run. Computed and exposed only; no
+   * load behavior changes ride on it yet.
+   */
+  watermarkVerdict(): WatermarkVerdict | null {
+    return this.loadVerdict?.verdict ?? null
+  }
+
+  /**
+   * @description The catch-up window `(from, to]` when the init verdict was
+   * `'catchup'`; null otherwise.
+   */
+  watermarkGap(): { from: number; to: number } | null {
+    return this.loadVerdict?.gap ?? null
+  }
+
+  /**
+   * @description Write the pending watermark stamp as a sidecar record —
+   * always called AFTER the LSM flushes it certifies completed. A stamp-write
+   * failure is fail-safe (unstamped/behind → rescan/catchup on next open,
+   * never a wrong adopt) but is said out loud and the pending stamp is
+   * retained for the next flush.
+   */
+  private async writePendingStamp(): Promise<void> {
+    if (this.pendingWatermark === null) return
+    const watermark = this.pendingWatermark
+    try {
+      await this.storage.saveMetadata(GRAPH_ADJACENCY_STAMP_KEY, {
+        noun: 'IndexWatermark',
+        ...makeProjectionStamp(watermark)
+      })
+      this.stampedWatermark = watermark
+      this.pendingWatermark = null
+    } catch (error) {
+      prodLog.error(
+        `[GraphAdjacencyIndex] failed to write watermark stamp (generation ${watermark}) — ` +
+          `artifact stays behind-stamped (safe: verdicts catchup/rescan, never wrong-adopt); ` +
+          `retrying on next flush:`,
+        error
+      )
+    }
+  }
+
+  /**
+   * @description Read the artifact's stamp and compute the three-way verdict
+   * against the store's committed generation. Unstamped state on a stamped
+   * store verdicts `'rescan'` LOUDLY — never a silent adopt.
+   *
+   * MIGRATION COST: existing pre-stamp brains verdict `'rescan'` exactly
+   * once (that open re-derives via the recovery walk it already runs); the
+   * next flush stamps them, and every later open adopts.
+   *
+   * @param artifactPresent - Whether persisted SSTables exist at all; gates
+   *   loud-vs-quiet on the rescan verdict so first boots don't scream.
+   */
+  private async loadWatermarkVerdict(artifactPresent: boolean): Promise<void> {
+    const committed = this.storage.committedGeneration?.() ?? null
+    let stamped: number | null = null
+    try {
+      const record = await this.storage.getMetadata(GRAPH_ADJACENCY_STAMP_KEY)
+      stamped = readStampedWatermark(record)
+    } catch {
+      // An unreadable stamp is unstamped — the fail-safe direction.
+      stamped = null
+    }
+    const result = computeWatermarkVerdict(stamped, committed)
+    this.loadVerdict = result
+    this.stampedWatermark = stamped
+
+    if (result.verdict === 'rescan') {
+      if (artifactPresent || stamped !== null) {
+        prodLog.warn(
+          `[GraphAdjacencyIndex] watermark verdict: RESCAN — persisted adjacency is ` +
+            (stamped === null
+              ? 'unstamped (legacy pre-stamp artifact, or a crash between data and stamp)'
+              : `stamped at generation ${stamped}, ABOVE the store's committed generation ${committed}`) +
+            ` — never adopting unverifiable state`
+        )
+      } else {
+        prodLog.debug(
+          '[GraphAdjacencyIndex] watermark verdict: rescan (no persisted artifact — first boot)'
+        )
+      }
+    } else if (result.verdict === 'catchup') {
+      prodLog.info(
+        `[GraphAdjacencyIndex] watermark verdict: catchup — adjacency stamped at generation ` +
+          `${stamped}, store committed at ${committed}; the (${stamped}, ${committed}] window ` +
+          `awaits an incremental fold (verdict exposed; the fold lands with the coordinator's wiring)`
+      )
+    }
+  }
+
   /**
    * Populate verbIdSet from storage without full rebuild
    * Lighter weight than full rebuild - only loads verb IDs, not all verb data
@@ -935,6 +1082,12 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
       }),
     ])
 
+    // STAMP-AFTER-DATA: the watermark stamp is the LAST write of the flush —
+    // both trees' SSTables are durable before the stamp lands. A crash
+    // anywhere above leaves the artifact behind-stamped or unstamped, which
+    // verdicts as catchup/rescan on the next open — never a wrong adopt.
+    await this.writePendingStamp()
+
     const elapsed = Date.now() - startTime
 
     prodLog.debug(`GraphAdjacencyIndex: Flush completed in ${elapsed}ms`)
@@ -955,6 +1108,10 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
         this.lsmTreeVerbsBySource.close(),
         this.lsmTreeVerbsByTarget.close(),
       ])
+
+      // Stamp-after-data on the shutdown path too: the trees' final flushes
+      // completed above, so a pending watermark may land now.
+      await this.writePendingStamp()
     }
 
     prodLog.info('GraphAdjacencyIndex: Shutdown complete')

src/hnsw/hnswIndex.ts

@@ -16,6 +16,22 @@ import { getGlobalCache, UnifiedCache } from '../utils/unifiedCache.js'
 import { prodLog } from '../utils/logger.js'
 import type { VectorIndexProvider, OpaqueIdSet, AtGenerationVectors } from '../plugin.js'
 import { ConnectionsCodec, compressedConnectionsKey } from './connectionsCodec.js'
+import {
+  computeWatermarkVerdict,
+  makeProjectionStamp,
+  readStampedWatermark,
+  type WatermarkVerdict,
+  type WatermarkVerdictResult
+} from '../utils/projectionWatermark.js'
+
+/**
+ * Storage key for the JS HNSW projection's watermark stamp — a sidecar
+ * record beside the artifact (per-node vector-index records + connection
+ * blobs + the entryPoint/maxLevel system record). Written LAST in
+ * {@link JsHnswVectorIndex.flush} so stamp-after-data ordering holds for
+ * every byte the stamp certifies.
+ */
+export const HNSW_INDEX_STAMP_KEY = '__index_hnsw_watermark__'
 
 // Default HNSW parameters
 const DEFAULT_CONFIG: HNSWConfig = {
@@ -99,6 +115,14 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
   private dirtyNodes: Set<string> = new Set() // Nodes with unpersisted HNSW data
   private dirtySystem: boolean = false // Whether system data (entryPoint, maxLevel) needs persist
 
+  // --- Watermark stamp state (see utils/projectionWatermark for the law) ---
+  /** Generation handed in via {@link stampWatermark}, awaiting the next flush. */
+  private pendingWatermark: number | null = null
+  /** Last watermark durably stamped by this instance or loaded on rebuild. */
+  private stampedWatermark: number | null = null
+  /** The three-way verdict computed at load; null until rebuild() runs. */
+  private loadVerdict: WatermarkVerdictResult | null = null
+
   // Lazy vector storage (B2 optimization): evict the float32 vector to
   // storage after insert; reload on demand via getVectorSafe() + UnifiedCache.
   private vectorStorageMode: 'memory' | 'lazy' = 'memory'
@@ -170,6 +194,9 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
     }
 
     if (this.dirtyNodes.size === 0 && !this.dirtySystem) {
+      // Nothing dirty — but a pending watermark still stamps: every byte it
+      // certifies is already durable, so stamp-after-data holds trivially.
+      await this.writePendingStamp()
       return 0
     }
 
@@ -239,6 +266,13 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
       throw new HnswFlushError(failedNodes.size, systemFailed, firstError ?? undefined)
     }
 
+    // STAMP-AFTER-DATA: the watermark stamp is the LAST write of the flush —
+    // it lands only after every dirty node and the system record persisted
+    // (the throw above guarantees it). A crash anywhere earlier leaves the
+    // artifact behind-stamped or unstamped, which verdicts as catchup/rescan
+    // on the next open — never a wrong adopt.
+    await this.writePendingStamp()
+
     if (nodeCount > 0) {
       prodLog.info(`[HNSW] Flushed ${nodeCount} dirty nodes in ${duration}ms`)
     }
@@ -246,6 +280,126 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
     return nodeCount
   }
 
+  /**
+   * @description Record the committed generation this projection reflects.
+   * The stamp is NOT written here — it is written as the final storage write
+   * of the next {@link flush} (stamp-after-data ordering is a module
+   * guarantee, not a caller obligation). The coordinator calls this with the
+   * store's committed generation right before flushing.
+   * @param generation - The committed generation every flushed byte reflects.
+   */
+  public stampWatermark(generation: number): void {
+    this.pendingWatermark = generation
+  }
+
+  /**
+   * @description The projection's current watermark: the stamp loaded at
+   * rebuild (or the last stamp durably written by this instance). Null =
+   * unstamped (legacy artifact, first boot, or stamping never wired).
+   */
+  public watermark(): number | null {
+    return this.stampedWatermark
+  }
+
+  /**
+   * @description The three-way adoption verdict computed at load —
+   * `'adopt'` (stamped == committed, zero work), `'catchup'` (stamped <
+   * committed; the gap from {@link watermarkGap} awaits an incremental
+   * fold), `'rescan'` (unstamped or stamped above committed — never
+   * trusted). Null until rebuild() has run. Computed and exposed only; no
+   * load behavior changes ride on it yet — today's rebuild triggers are
+   * unchanged.
+   */
+  public watermarkVerdict(): WatermarkVerdict | null {
+    return this.loadVerdict?.verdict ?? null
+  }
+
+  /**
+   * @description The catch-up window `(from, to]` when the load verdict was
+   * `'catchup'`; null otherwise.
+   */
+  public watermarkGap(): { from: number; to: number } | null {
+    return this.loadVerdict?.gap ?? null
+  }
+
+  /**
+   * @description Write the pending watermark stamp as a sidecar record —
+   * always called AFTER the data it certifies is durable. The stamp carries
+   * the vector-space identity this module can honestly assert: dimensions
+   * only (no embedding-model id is reachable from the index — it never sees
+   * the embedder). A stamp-write failure is fail-safe (unstamped/behind →
+   * rescan/catchup on next open, never a wrong adopt) but is said out loud
+   * and the pending stamp is retained for the next flush.
+   */
+  private async writePendingStamp(): Promise<void> {
+    if (this.pendingWatermark === null || !this.storage) return
+    const watermark = this.pendingWatermark
+    try {
+      await this.storage.saveMetadata(HNSW_INDEX_STAMP_KEY, {
+        noun: 'IndexWatermark',
+        ...makeProjectionStamp(watermark, { dimensions: this.dimension })
+      })
+      this.stampedWatermark = watermark
+      this.pendingWatermark = null
+    } catch (error) {
+      prodLog.error(
+        `[HNSW] failed to write watermark stamp (generation ${watermark}) — ` +
+          `artifact stays behind-stamped (safe: verdicts catchup/rescan, never wrong-adopt); ` +
+          `retrying on next flush:`,
+        error
+      )
+    }
+  }
+
+  /**
+   * @description Read the artifact's stamp and compute the three-way verdict
+   * against the store's committed generation. Unstamped state on a stamped
+   * store verdicts `'rescan'` LOUDLY — never a silent adopt.
+   *
+   * MIGRATION COST: existing pre-stamp brains verdict `'rescan'` exactly
+   * once (that open re-derives via the rebuild it is already running); the
+   * next flush stamps them, and every later open adopts.
+   *
+   * @param artifactPresent - Whether a persisted artifact exists at all (a
+   *   system record was found); gates loud-vs-quiet on the rescan verdict so
+   *   first boots don't scream.
+   */
+  private async loadWatermarkVerdict(artifactPresent: boolean): Promise<void> {
+    if (!this.storage) return
+    const committed = this.storage.committedGeneration?.() ?? null
+    let stamped: number | null = null
+    try {
+      const record = await this.storage.getMetadata(HNSW_INDEX_STAMP_KEY)
+      stamped = readStampedWatermark(record)
+    } catch {
+      // An unreadable stamp is unstamped — the fail-safe direction.
+      stamped = null
+    }
+    const result = computeWatermarkVerdict(stamped, committed)
+    this.loadVerdict = result
+    this.stampedWatermark = stamped
+
+    if (result.verdict === 'rescan') {
+      if (artifactPresent || stamped !== null) {
+        prodLog.warn(
+          `[HNSW] watermark verdict: RESCAN — persisted index is ` +
+            (stamped === null
+              ? 'unstamped (legacy pre-stamp artifact, or a crash between data and stamp)'
+              : `stamped at generation ${stamped}, ABOVE the store's committed generation ${committed}`) +
+            ` — never adopting unverifiable state`
+        )
+      } else {
+        prodLog.debug('[HNSW] watermark verdict: rescan (no persisted artifact — first boot)')
+      }
+    } else if (result.verdict === 'catchup') {
+      prodLog.info(
+        `[HNSW] watermark verdict: catchup — index stamped at generation ${stamped}, ` +
+          `store committed at ${committed}; the (${stamped}, ${committed}] window awaits ` +
+          `an incremental fold (verdict exposed; the fold lands with the coordinator's wiring)`
+      )
+    }
+  }
+
   /**
    * @description Persist one node's connections. When the connections codec is
    * wired AND the storage adapter exposes `saveBinaryBlob`, the per-level
@@ -1563,6 +1717,11 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
         this.maxLevel = systemData.maxLevel
       }
 
+      // Step 2b: Watermark verdict for the persisted artifact — computed and
+      // exposed only (today's rebuild flow is unchanged; this rebuild IS the
+      // re-derive a 'rescan' verdict asks for).
+      await this.loadWatermarkVerdict(systemData !== null)
+
       // Step 3: Determine preloading strategy (adaptive caching)
       // Check if vectors should be preloaded at init or loaded on-demand
       const stats = await this.storage.getStatistics()

src/utils/metadataIndex.ts

@@ -13,6 +13,13 @@ import { MetadataIndexCache, MetadataIndexCacheConfig } from './metadataIndexCac
 import { compareCodePoints } from './collation.js'
 import { prodLog } from './logger.js'
 import { getGlobalCache, UnifiedCache } from './unifiedCache.js'
+import {
+  computeWatermarkVerdict,
+  makeProjectionStamp,
+  readStampedWatermark,
+  type WatermarkVerdict,
+  type WatermarkVerdictResult
+} from './projectionWatermark.js'
 import {
   NounType,
   VerbType,
@@ -109,6 +116,15 @@ interface FieldStats {
   normalizationStrategy?: 'none' | 'precision' | 'bucket'
 }
 
+/**
+ * Storage key for the metadata projection's watermark stamp — a sidecar
+ * record beside the artifact (field registry + field indexes + chunked
+ * sparse indexes + column-store segments + id-mapper records). Written LAST
+ * in {@link MetadataIndexManager.flush} so stamp-after-data ordering holds
+ * for every byte the stamp certifies.
+ */
+export const METADATA_INDEX_STAMP_KEY = '__index_metadata_watermark__'
+
 /**
  * Implements {@link MetadataIndexProvider}: the metadata-index surface Brainy
  * calls on whatever the `'metadataIndex'` provider resolves to (its own
@@ -124,6 +140,14 @@ export class MetadataIndexManager implements MetadataIndexProvider {
   private lastFlushTime = Date.now()
   private autoFlushThreshold = 10 // Start with 10 for more frequent non-blocking flushes
 
+  // --- Watermark stamp state (see utils/projectionWatermark for the law) ---
+  /** Generation handed in via {@link stampWatermark}, awaiting the next flush. */
+  private pendingWatermark: number | null = null
+  /** Last watermark durably stamped by this instance or loaded at init. */
+  private stampedWatermark: number | null = null
+  /** The three-way verdict computed at init; null until init runs. */
+  private loadVerdict: WatermarkVerdictResult | null = null
+
   // Cardinality and field statistics tracking
   private fieldStats = new Map<string, FieldStats>()
   private cardinalityUpdateInterval = 100 // Update cardinality every N operations
@@ -250,6 +274,13 @@ export class MetadataIndexManager implements MetadataIndexProvider {
     // Must run first to populate fieldIndexes directory before warming cache
     await this.loadFieldRegistry()
 
+    // Compute the watermark verdict for the persisted artifact BEFORE any
+    // early return below — the verdict is recorded for every open, whether
+    // the workspace is empty, rebuilding, or warm. Computed and exposed
+    // only: today's rebuild triggers are unchanged (acting on 'catchup' —
+    // the incremental fold — lands with the coordinator's wiring).
+    await this.loadWatermarkVerdict()
+
     // Initialize EntityIdMapper (loads UUID ↔ integer mappings from storage)
     await this.idMapper.init()
 
@@ -2599,6 +2630,10 @@ export class MetadataIndexManager implements MetadataIndexProvider {
 
     // Check if we have anything else to flush
     if (this.dirtyFields.size === 0) {
+      // Nothing dirty — but a pending watermark still stamps (the registry
+      // + id-mapper writes above are the only bytes this pass touched, and
+      // they are durable at this point). Stamp-after-data holds.
+      await this.writePendingStamp()
       return // No dirty field indexes to flush
     }
 
@@ -2638,6 +2673,129 @@ export class MetadataIndexManager implements MetadataIndexProvider {
     if (this.columnStore) {
       await this.columnStore.flush()
     }
+
+    // STAMP-AFTER-DATA: the watermark stamp is the LAST write of the flush —
+    // every byte it certifies (field indexes, registry, id-mapper records,
+    // column-store segments) is durable before the stamp lands. A crash
+    // anywhere above leaves the artifact behind-stamped or unstamped, which
+    // verdicts as catchup/rescan on the next open — never a wrong adopt.
+    await this.writePendingStamp()
+  }
+
+  /**
+   * @description Record the committed generation this projection reflects.
+   * The stamp is NOT written here — it is written as the final storage write
+   * of the next {@link flush} (stamp-after-data ordering is a module
+   * guarantee, not a caller obligation). The coordinator calls this with the
+   * store's committed generation right before flushing.
+   * @param generation - The committed generation every flushed byte reflects.
+   */
+  stampWatermark(generation: number): void {
+    this.pendingWatermark = generation
+  }
+
+  /**
+   * @description The projection's current watermark: the stamp loaded at
+   * init (or the last stamp durably written by this instance). Null =
+   * unstamped (legacy artifact, first boot, or stamping never wired).
+   */
+  watermark(): number | null {
+    return this.stampedWatermark
+  }
+
+  /**
+   * @description The three-way adoption verdict computed at init —
+   * `'adopt'` (stamped == committed, zero work), `'catchup'` (stamped <
+   * committed; the gap from {@link watermarkGap} awaits an incremental
+   * fold), `'rescan'` (unstamped or stamped above committed — never
+   * trusted). Null until init() has run. Computed and exposed only; no
+   * load behavior changes ride on it yet.
+   */
+  watermarkVerdict(): WatermarkVerdict | null {
+    return this.loadVerdict?.verdict ?? null
+  }
+
+  /**
+   * @description The catch-up window `(from, to]` when the init verdict was
+   * `'catchup'`; null otherwise.
+   */
+  watermarkGap(): { from: number; to: number } | null {
+    return this.loadVerdict?.gap ?? null
+  }
+
+  /**
+   * @description Write the pending watermark stamp as a sidecar record —
+   * always called AFTER the data it certifies is durable. A stamp-write
+   * failure is fail-safe (the artifact stays unstamped/behind → rescan or
+   * catchup on next open, never a wrong adopt) but is said out loud and the
+   * pending stamp is retained for the next flush.
+   */
+  private async writePendingStamp(): Promise<void> {
+    if (this.pendingWatermark === null) return
+    const watermark = this.pendingWatermark
+    try {
+      await this.storage.saveMetadata(METADATA_INDEX_STAMP_KEY, {
+        noun: 'IndexWatermark',
+        ...makeProjectionStamp(watermark)
+      })
+      this.stampedWatermark = watermark
+      this.pendingWatermark = null
+    } catch (error) {
+      prodLog.error(
+        `[MetadataIndex] failed to write watermark stamp (generation ${watermark}) — ` +
+          `artifact stays behind-stamped (safe: verdicts catchup/rescan, never wrong-adopt); ` +
+          `retrying on next flush:`,
+        error
+      )
+    }
+  }
+
+  /**
+   * @description Read the artifact's stamp and compute the three-way verdict
+   * against the store's committed generation. Unstamped state on a stamped
+   * store verdicts `'rescan'` LOUDLY — never a silent adopt.
+   *
+   * MIGRATION COST: existing pre-stamp brains verdict `'rescan'` exactly
+   * once (this open re-derives from source as it already does today); the
+   * next flush stamps them, and every later open adopts.
+   */
+  private async loadWatermarkVerdict(): Promise<void> {
+    const committed = this.storage.committedGeneration?.() ?? null
+    let stamped: number | null = null
+    try {
+      const record = await this.storage.getMetadata(METADATA_INDEX_STAMP_KEY)
+      stamped = readStampedWatermark(record)
+    } catch {
+      // An unreadable stamp is unstamped — the fail-safe direction.
+      stamped = null
+    }
+    const result = computeWatermarkVerdict(stamped, committed)
+    this.loadVerdict = result
+    this.stampedWatermark = stamped
+
+    if (result.verdict === 'rescan') {
+      const artifactPresent = this.fieldIndexes.size > 0 || stamped !== null
+      if (artifactPresent) {
+        prodLog.warn(
+          `[MetadataIndex] watermark verdict: RESCAN — persisted index is ` +
+            (stamped === null
+              ? 'unstamped (legacy pre-stamp artifact, or a crash between data and stamp)'
+              : `stamped at generation ${stamped}, ABOVE the store's committed generation ${committed}`) +
+            ` — never adopting unverifiable state`
+        )
+      } else {
+        prodLog.debug(
+          '[MetadataIndex] watermark verdict: rescan (no persisted artifact — first boot)'
+        )
+      }
+    } else if (result.verdict === 'catchup') {
+      prodLog.info(
+        `[MetadataIndex] watermark verdict: catchup — index stamped at generation ` +
+          `${stamped}, store committed at ${committed}; the (${stamped}, ${committed}] ` +
+          `window awaits an incremental fold (verdict exposed; the fold lands with the ` +
+          `coordinator's wiring)`
+      )
+    }
   }
 
   /**

src/utils/projectionWatermark.ts

@@ -0,0 +1,150 @@
+/**
+ * @module utils/projectionWatermark
+ * @description The watermark-stamp contract shared by Brainy's persisted TS
+ * projections (metadata index, JS HNSW vector index, graph adjacency index).
+ *
+ * THE LAW: every persisted projection artifact carries a stamp asserting
+ * "this state reflects every committed generation ≤ watermark and nothing
+ * above it, atomically". STAMP-AFTER-DATA: the stamp is written only after
+ * every byte it certifies is durable — a crash between data and stamp leaves
+ * the artifact unstamped, which verdicts as a rescan, never a wrong adopt.
+ *
+ * At load, each owner computes a three-way verdict against the store's
+ * committed generation — the same rule and verdict names the aggregation
+ * machinery ships (see `AggregationIndex.stateAdoptionVerdict`):
+ *
+ * - `'adopt'`   — stamped == committed (clean reopen, zero work), or the
+ *                 store exposes no committed generation at all (pre-stamp
+ *                 stores keep their pre-stamp behavior).
+ * - `'catchup'` — stamped < committed (an unclean exit after later writes,
+ *                 or a long-lived writer whose last stamp predates recent
+ *                 commits). The artifact is exact AS OF its stamp, so the
+ *                 missing window `(stamped, committed]` can be folded
+ *                 incrementally — at-least-once idempotent, bounded by
+ *                 writes since the stamp, never by store size.
+ * - `'rescan'`  — unstamped (a legacy pre-stamp artifact, or a crash between
+ *                 data and stamp) or stamped ABOVE committed (e.g. a log
+ *                 truncation on a copied store pulled the watermark back):
+ *                 the state over-claims unverifiably — one exact rescan,
+ *                 said out loud, never a silent adopt.
+ *
+ * MIGRATION COST (stated once, honored by every owner): existing pre-stamp
+ * brains verdict `'rescan'` exactly once — they re-derive from source on
+ * that open, the next flush stamps them, and every later open adopts.
+ *
+ * The verdict is COMPUTED AND EXPOSED by each owner; acting on `'catchup'`
+ * (the incremental fold) lands with the owner's coordinator wiring.
+ */
+
+/** The three-way load verdict for a persisted projection artifact. */
+export type WatermarkVerdict = 'adopt' | 'catchup' | 'rescan'
+
+/**
+ * Format version written into every projection stamp. Bump when the stamp
+ * record's shape changes incompatibly; readers treat an unknown version as
+ * unstamped (→ rescan) rather than guessing.
+ */
+export const PROJECTION_STAMP_FORMAT_VERSION = 1
+
+/**
+ * @description The stamp record a projection writes into (or beside) its
+ * persisted artifact, always AFTER the data it certifies is durable.
+ */
+export interface ProjectionStamp {
+  /** The committed generation this artifact reflects, exactly and entirely. */
+  watermark: number
+  /** {@link PROJECTION_STAMP_FORMAT_VERSION} at write time. */
+  formatVersion: number
+  /** Wall-clock ms at stamp write — diagnostic only, never load-bearing. */
+  stampedAt: number
+  /**
+   * Identity of the vector space for vector-bearing artifacts (the HNSW
+   * index). The JS index has no reachable embedding-model id in its module,
+   * so dimensions are the only identity it can honestly assert.
+   */
+  modelIdentity?: { embedModelId?: string; dimensions: number | null }
+}
+
+/** The verdict plus everything the owner needs to report or act on it. */
+export interface WatermarkVerdictResult {
+  verdict: WatermarkVerdict
+  /** Watermark read from the artifact's stamp; null = unstamped. */
+  stamped: number | null
+  /** The store's committed generation at load; null = no capability. */
+  committed: number | null
+  /** The catch-up window `(from, to]` when verdict is `'catchup'`, else null. */
+  gap: { from: number; to: number } | null
+}
+
+/**
+ * @description Build a stamp record for a projection artifact.
+ * @param watermark - The committed generation the artifact reflects.
+ * @param modelIdentity - Vector-space identity for vector-bearing artifacts.
+ * @returns The stamp record to persist (stamp-after-data).
+ */
+export function makeProjectionStamp(
+  watermark: number,
+  modelIdentity?: ProjectionStamp['modelIdentity']
+): ProjectionStamp {
+  const stamp: ProjectionStamp = {
+    watermark,
+    formatVersion: PROJECTION_STAMP_FORMAT_VERSION,
+    stampedAt: Date.now()
+  }
+  if (modelIdentity !== undefined) stamp.modelIdentity = modelIdentity
+  return stamp
+}
+
+/**
+ * @description Read the stamped watermark out of a persisted record, treating
+ * anything malformed (missing, wrong type, non-finite, negative, or an
+ * unknown format version) as unstamped — the fail-safe direction is rescan,
+ * never a guessed adopt.
+ * @param record - The raw persisted record (or null/undefined).
+ * @returns The stamped watermark, or null if effectively unstamped.
+ */
+export function readStampedWatermark(record: unknown): number | null {
+  if (record === null || typeof record !== 'object') return null
+  const rec = record as Record<string, unknown>
+  const version = rec.formatVersion
+  if (typeof version !== 'number' || version > PROJECTION_STAMP_FORMAT_VERSION) {
+    return null
+  }
+  const raw = rec.watermark
+  if (typeof raw !== 'number' || !Number.isFinite(raw) || raw < 0) return null
+  return raw
+}
+
+/**
+ * @description The three-way adoption verdict — the single decision rule
+ * every stamped projection shares (mirrors the aggregation machinery's
+ * `stateAdoptionVerdict` exactly: same names, same directions).
+ * @param stamped - Watermark read from the artifact ({@link readStampedWatermark}).
+ * @param committed - The store's committed generation (null = no capability).
+ * @returns The verdict with the stamped/committed pair and the catch-up gap.
+ */
+export function computeWatermarkVerdict(
+  stamped: number | null,
+  committed: number | null
+): WatermarkVerdictResult {
+  // No committed-generation capability: hash/shape checks are the only
+  // adoption gate, exactly the pre-stamp behavior. Never fail a store that
+  // cannot express the question.
+  if (committed === null) {
+    return { verdict: 'adopt', stamped, committed, gap: null }
+  }
+  if (stamped === committed) {
+    return { verdict: 'adopt', stamped, committed, gap: null }
+  }
+  if (stamped !== null && stamped < committed) {
+    return {
+      verdict: 'catchup',
+      stamped,
+      committed,
+      gap: { from: stamped, to: committed }
+    }
+  }
+  // Unstamped, or stamped above committed: unverifiable — rescan, loudly
+  // (the caller owns the loud log so it can name its projection).
+  return { verdict: 'rescan', stamped, committed, gap: null }
+}

tests/integration/watermark-adopt-reopen.test.ts

@@ -0,0 +1,50 @@
+/**
+ * @module tests/integration/watermark-adopt-reopen
+ * @description End-to-end LC1 watermark adoption: a clean flush+close stamps
+ * every projection at the committed generation; the reopen verdicts all read
+ * 'adopt' — a same-version reopen owes ZERO rebuild work, provably, via the
+ * stamps rather than via absence of complaint.
+ */
+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'
+
+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 })
+})
+
+describe('watermark stamps ride the flush fan-out', () => {
+  it('flush stamps all three projections at the committed generation; reopen adopts', async () => {
+    const dir = mkdtempSync(join(tmpdir(), 'brainy-wm-'))
+    dirs.push(dir)
+    let brain = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false })
+    await brain.init()
+    brains.push(brain)
+    await brain.add({ data: 'stamped row', type: NounType.Document, metadata: { k: 1 } })
+    await brain.flush()
+
+    const committed = (brain as unknown as {
+      storage: { committedGeneration(): number }
+    }).storage.committedGeneration()
+    const mi = (brain as unknown as { metadataIndex: { watermark(): number | null } }).metadataIndex
+    expect(mi.watermark(), 'metadata stamp = committed').toBe(committed)
+    await brain.close()
+    brains.pop()
+
+    brain = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false })
+    await brain.init()
+    brains.push(brain)
+    const mi2 = (brain as unknown as {
+      metadataIndex: { watermarkVerdict(): string | null }
+    }).metadataIndex
+    expect(mi2.watermarkVerdict(), 'clean reopen adopts').toBe('adopt')
+    // And the brain serves.
+    expect((await brain.find({ where: { k: 1 }, limit: 5 })).length).toBe(1)
+  }, 60000)
+})

tests/unit/graph/graph-adjacency-watermark.test.ts

@@ -0,0 +1,213 @@
+/**
+ * @module tests/unit/graph/graph-adjacency-watermark
+ * @description Watermark-stamp pins for the graph-adjacency projection.
+ *
+ * THE LAW under test: the persisted adjacency artifact (the two verb-id LSM
+ * trees' SSTables + manifests) carries a stamp asserting "this state
+ * reflects every committed generation ≤ W and nothing above W" — written
+ * AFTER both trees' flushes complete — and init() computes the three-way
+ * verdict: stamped==committed → 'adopt' · stamped<committed → 'catchup'
+ * (gap reported) · stamped>committed OR unstamped → 'rescan', LOUDLY.
+ *
+ * The verdict is COMPUTED AND EXPOSED only — cold-load recovery and rebuild
+ * triggers are unchanged.
+ */
+import { describe, it, expect, vi, afterEach } from 'vitest'
+import { v4 as uuidv4 } from 'uuid'
+import {
+  GraphAdjacencyIndex,
+  GRAPH_ADJACENCY_STAMP_KEY
+} from '../../../src/graph/graphAdjacencyIndex.js'
+import { EntityIdMapper } from '../../../src/utils/entityIdMapper.js'
+import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js'
+import { VerbType } from '../../../src/types/graphTypes.js'
+import type { GraphVerb } from '../../../src/coreTypes.js'
+import { prodLog } from '../../../src/utils/logger.js'
+
+function makeVerb(id: string, sourceId: string, targetId: string): GraphVerb {
+  return {
+    id,
+    sourceId,
+    targetId,
+    vector: [],
+    type: VerbType.RelatedTo,
+    verb: VerbType.RelatedTo
+  }
+}
+
+async function makeStorage(committed: number | null): Promise<MemoryStorage> {
+  const storage = new MemoryStorage()
+  await storage.init()
+  if (committed !== null) {
+    vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed)
+  }
+  return storage
+}
+
+function setCommitted(storage: MemoryStorage, committed: number): void {
+  vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed)
+}
+
+/** Session 1: index verbs, optionally stamp, flush + close — the artifact. */
+async function writeArtifact(storage: MemoryStorage, stamp: number | null): Promise<void> {
+  const idMapper = new EntityIdMapper({ storage, storageKey: 'test:graph:idMapper' })
+  await idMapper.init()
+  const index = new GraphAdjacencyIndex(storage, {}, idMapper)
+  const a = uuidv4()
+  const b = uuidv4()
+  const aInt = BigInt(idMapper.getOrAssign(a))
+  const bInt = BigInt(idMapper.getOrAssign(b))
+  await index.addVerb(makeVerb(uuidv4(), a, b), aInt, bInt, 1n)
+  if (stamp !== null) index.stampWatermark(stamp)
+  await index.flush()
+  await index.close()
+}
+
+/** Session 2: reopen on the same storage via the cold-load path. */
+async function reopen(storage: MemoryStorage): Promise<GraphAdjacencyIndex> {
+  const index = new GraphAdjacencyIndex(storage)
+  await index.init()
+  return index
+}
+
+afterEach(() => {
+  vi.restoreAllMocks()
+})
+
+describe('graph adjacency index — watermark stamp + three-way load verdict', () => {
+  it("save-with-stamp then reopen at the same committed generation → 'adopt'", async () => {
+    const storage = await makeStorage(5)
+    await writeArtifact(storage, 5)
+
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('adopt')
+    expect(index.watermark()).toBe(5)
+    expect(index.watermarkGap()).toBeNull()
+    await index.close()
+  })
+
+  it("stamp BEHIND the committed generation → 'catchup' with the exact gap reported", async () => {
+    const storage = await makeStorage(5)
+    await writeArtifact(storage, 5)
+
+    setCommitted(storage, 11)
+
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('catchup')
+    expect(index.watermark()).toBe(5)
+    expect(index.watermarkGap()).toEqual({ from: 5, to: 11 })
+    await index.close()
+  })
+
+  it("stamp ABOVE the committed generation → 'rescan', said out loud", async () => {
+    const storage = await makeStorage(9)
+    await writeArtifact(storage, 9)
+
+    setCommitted(storage, 4)
+
+    const warnSpy = vi.spyOn(prodLog, 'warn')
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('rescan')
+    expect(index.watermarkGap()).toBeNull()
+    const said = warnSpy.mock.calls.map(c => String(c[0])).join('\n')
+    expect(said).toContain('RESCAN')
+    expect(said).toContain('ABOVE')
+    await index.close()
+  })
+
+  it("legacy unstamped artifact on a stamped store → 'rescan', LOUD — never a silent adopt", async () => {
+    const storage = await makeStorage(3)
+    await writeArtifact(storage, null) // pre-stamp adjacency: SSTables, no stamp
+
+    expect(await storage.getMetadata(GRAPH_ADJACENCY_STAMP_KEY)).toBeNull()
+
+    const warnSpy = vi.spyOn(prodLog, 'warn')
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('rescan')
+    expect(index.watermark()).toBeNull()
+    const said = warnSpy.mock.calls.map(c => String(c[0])).join('\n')
+    expect(said).toContain('RESCAN')
+    expect(said).toContain('unstamped')
+    await index.close()
+  })
+
+  it("a store with no committed-generation capability keeps pre-stamp behavior → 'adopt'", async () => {
+    const storage = await makeStorage(null)
+    await writeArtifact(storage, null)
+
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('adopt')
+    expect(index.watermark()).toBeNull()
+    await index.close()
+  })
+
+  it('STAMP-AFTER-DATA: the stamp is the last saveMetadata of the flush, after both trees’ SSTable + manifest writes', async () => {
+    const storage = await makeStorage(2)
+    const idMapper = new EntityIdMapper({ storage, storageKey: 'test:graph:idMapper' })
+    await idMapper.init()
+    const index = new GraphAdjacencyIndex(storage, {}, idMapper)
+    const a = uuidv4()
+    const b = uuidv4()
+    await index.addVerb(
+      makeVerb(uuidv4(), a, b),
+      BigInt(idMapper.getOrAssign(a)),
+      BigInt(idMapper.getOrAssign(b)),
+      1n
+    )
+
+    const keys: string[] = []
+    const originalSave = storage.saveMetadata.bind(storage)
+    vi.spyOn(storage, 'saveMetadata').mockImplementation(async (id, metadata) => {
+      keys.push(id)
+      return originalSave(id, metadata)
+    })
+
+    index.stampWatermark(2)
+    await index.flush()
+
+    const stampAt = keys.indexOf(GRAPH_ADJACENCY_STAMP_KEY)
+    expect(stampAt, 'stamp record was written').toBeGreaterThanOrEqual(0)
+    expect(stampAt, 'stamp is the FINAL metadata write of the flush').toBe(keys.length - 1)
+    // Both trees flushed durable bytes before the stamp landed.
+    expect(
+      keys.slice(0, stampAt).some(k => k.startsWith('graph-lsm-verbs-source')),
+      'verbs-by-source tree wrote before the stamp'
+    ).toBe(true)
+    expect(
+      keys.slice(0, stampAt).some(k => k.startsWith('graph-lsm-verbs-target')),
+      'verbs-by-target tree wrote before the stamp'
+    ).toBe(true)
+
+    const record = (await storage.getMetadata(GRAPH_ADJACENCY_STAMP_KEY)) as {
+      watermark: number
+      formatVersion: number
+      stampedAt: number
+    }
+    expect(record.watermark).toBe(2)
+    expect(record.formatVersion).toBe(1)
+    expect(typeof record.stampedAt).toBe('number')
+
+    await index.close()
+  })
+
+  it('a pending stamp also lands on the close() shutdown path, after the final tree flushes', async () => {
+    const storage = await makeStorage(6)
+    const idMapper = new EntityIdMapper({ storage, storageKey: 'test:graph:idMapper' })
+    await idMapper.init()
+    const index = new GraphAdjacencyIndex(storage, {}, idMapper)
+    const a = uuidv4()
+    const b = uuidv4()
+    await index.addVerb(
+      makeVerb(uuidv4(), a, b),
+      BigInt(idMapper.getOrAssign(a)),
+      BigInt(idMapper.getOrAssign(b)),
+      1n
+    )
+
+    index.stampWatermark(6)
+    await index.close() // no explicit flush — close() flushes, then stamps
+
+    const record = (await storage.getMetadata(GRAPH_ADJACENCY_STAMP_KEY)) as { watermark: number }
+    expect(record?.watermark).toBe(6)
+  })
+})

tests/unit/hnsw/hnsw-watermark.test.ts

@@ -0,0 +1,200 @@
+/**
+ * @module tests/unit/hnsw/hnsw-watermark
+ * @description Watermark-stamp pins for the JS HNSW vector projection.
+ *
+ * THE LAW under test: the persisted HNSW artifact (per-node records + the
+ * entryPoint/maxLevel system record) carries a stamp asserting "this state
+ * reflects every committed generation ≤ W and nothing above W" — written
+ * AFTER every byte it certifies is durable — and rebuild() computes the
+ * three-way verdict: stamped==committed → 'adopt' · stamped<committed →
+ * 'catchup' (gap reported) · stamped>committed OR unstamped → 'rescan',
+ * LOUDLY. Vector-bearing stamps carry the model identity this module can
+ * honestly assert: dimensions only (no embedding-model id is reachable from
+ * the index module).
+ *
+ * The verdict is COMPUTED AND EXPOSED only — no rebuild trigger changed.
+ */
+import { describe, it, expect, vi, afterEach } from 'vitest'
+import { v4 as uuidv4 } from 'uuid'
+import { JsHnswVectorIndex, HNSW_INDEX_STAMP_KEY } from '../../../src/hnsw/hnswIndex.js'
+import { euclideanDistance } from '../../../src/utils/index.js'
+import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js'
+import { prodLog } from '../../../src/utils/logger.js'
+
+const DIM = 8
+
+function randomVector(dim: number): number[] {
+  return Array.from({ length: dim }, () => Math.random() * 2 - 1)
+}
+
+async function makeStorage(committed: number | null): Promise<MemoryStorage> {
+  const storage = new MemoryStorage()
+  await storage.init()
+  if (committed !== null) {
+    vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed)
+  }
+  return storage
+}
+
+function setCommitted(storage: MemoryStorage, committed: number): void {
+  vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed)
+}
+
+function makeIndex(storage: MemoryStorage): JsHnswVectorIndex {
+  return new JsHnswVectorIndex(
+    { M: 4, efConstruction: 50, efSearch: 20 },
+    euclideanDistance,
+    { useParallelization: false, storage, persistMode: 'deferred' }
+  )
+}
+
+/** Session 1: insert nodes, optionally stamp, flush — the durable artifact. */
+async function writeArtifact(storage: MemoryStorage, stamp: number | null): Promise<void> {
+  const index = makeIndex(storage)
+  for (let i = 0; i < 3; i++) {
+    await index.addItem({ id: uuidv4(), vector: randomVector(DIM) })
+  }
+  if (stamp !== null) index.stampWatermark(stamp)
+  await index.flush()
+}
+
+/** Session 2: reopen on the same storage via the load path (rebuild). */
+async function reopen(storage: MemoryStorage): Promise<JsHnswVectorIndex> {
+  const index = makeIndex(storage)
+  await index.rebuild()
+  return index
+}
+
+afterEach(() => {
+  vi.restoreAllMocks()
+})
+
+describe('JS HNSW index — watermark stamp + three-way load verdict', () => {
+  it("save-with-stamp then reopen at the same committed generation → 'adopt'", async () => {
+    const storage = await makeStorage(5)
+    await writeArtifact(storage, 5)
+
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('adopt')
+    expect(index.watermark()).toBe(5)
+    expect(index.watermarkGap()).toBeNull()
+  })
+
+  it("stamp BEHIND the committed generation → 'catchup' with the exact gap reported", async () => {
+    const storage = await makeStorage(5)
+    await writeArtifact(storage, 5)
+
+    setCommitted(storage, 9)
+
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('catchup')
+    expect(index.watermark()).toBe(5)
+    expect(index.watermarkGap()).toEqual({ from: 5, to: 9 })
+  })
+
+  it("stamp ABOVE the committed generation → 'rescan', said out loud", async () => {
+    const storage = await makeStorage(9)
+    await writeArtifact(storage, 9)
+
+    setCommitted(storage, 4)
+
+    const warnSpy = vi.spyOn(prodLog, 'warn')
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('rescan')
+    expect(index.watermarkGap()).toBeNull()
+    const said = warnSpy.mock.calls.map(c => String(c[0])).join('\n')
+    expect(said).toContain('RESCAN')
+    expect(said).toContain('ABOVE')
+  })
+
+  it("legacy unstamped artifact on a stamped store → 'rescan', LOUD — never a silent adopt", async () => {
+    const storage = await makeStorage(3)
+    await writeArtifact(storage, null) // pre-stamp index: data flushed, no stamp
+
+    expect(await storage.getMetadata(HNSW_INDEX_STAMP_KEY)).toBeNull()
+
+    const warnSpy = vi.spyOn(prodLog, 'warn')
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('rescan')
+    expect(index.watermark()).toBeNull()
+    const said = warnSpy.mock.calls.map(c => String(c[0])).join('\n')
+    expect(said).toContain('RESCAN')
+    expect(said).toContain('unstamped')
+  })
+
+  it("a store with no committed-generation capability keeps pre-stamp behavior → 'adopt'", async () => {
+    const storage = await makeStorage(null)
+    await writeArtifact(storage, null)
+
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('adopt')
+    expect(index.watermark()).toBeNull()
+  })
+
+  it('STAMP-AFTER-DATA: the stamp lands after every node record and the system record', async () => {
+    const storage = await makeStorage(2)
+    const index = makeIndex(storage)
+    for (let i = 0; i < 3; i++) {
+      await index.addItem({ id: uuidv4(), vector: randomVector(DIM) })
+    }
+
+    // One shared op log across all three write surfaces pins global order.
+    const ops: string[] = []
+    const origNode = storage.saveVectorIndexData.bind(storage)
+    vi.spyOn(storage, 'saveVectorIndexData').mockImplementation(async (id, data) => {
+      ops.push(`node:${id}`)
+      return origNode(id, data)
+    })
+    const origSystem = storage.saveHNSWSystem.bind(storage)
+    vi.spyOn(storage, 'saveHNSWSystem').mockImplementation(async data => {
+      ops.push('system')
+      return origSystem(data)
+    })
+    const origMeta = storage.saveMetadata.bind(storage)
+    vi.spyOn(storage, 'saveMetadata').mockImplementation(async (id, metadata) => {
+      ops.push(`meta:${id}`)
+      return origMeta(id, metadata)
+    })
+
+    index.stampWatermark(2)
+    await index.flush()
+
+    const stampAt = ops.indexOf(`meta:${HNSW_INDEX_STAMP_KEY}`)
+    expect(stampAt, 'stamp record was written').toBeGreaterThanOrEqual(0)
+    expect(stampAt, 'stamp is the FINAL write of the flush').toBe(ops.length - 1)
+    expect(ops.filter(o => o.startsWith('node:')).length).toBeGreaterThan(0)
+    expect(ops.indexOf('system')).toBeLessThan(stampAt)
+  })
+
+  it('the stamp record carries {watermark, formatVersion, stampedAt} + modelIdentity (dims only)', async () => {
+    const storage = await makeStorage(7)
+    await writeArtifact(storage, 7)
+
+    const record = (await storage.getMetadata(HNSW_INDEX_STAMP_KEY)) as {
+      watermark: number
+      formatVersion: number
+      stampedAt: number
+      modelIdentity: { embedModelId?: string; dimensions: number | null }
+    }
+    expect(record.watermark).toBe(7)
+    expect(record.formatVersion).toBe(1)
+    expect(typeof record.stampedAt).toBe('number')
+    // The JS index never sees the embedder — dimensions are the only vector-
+    // space identity it can honestly assert.
+    expect(record.modelIdentity).toEqual({ dimensions: DIM })
+  })
+
+  it('a pending stamp still lands when nothing is dirty (already-durable bytes, stamp-after-data trivially holds)', async () => {
+    const storage = await makeStorage(4)
+    const index = makeIndex(storage)
+    await index.addItem({ id: uuidv4(), vector: randomVector(DIM) })
+    await index.flush() // data durable, no stamp yet
+
+    index.stampWatermark(4)
+    await index.flush() // nothing dirty — the stamp must still be written
+
+    const record = (await storage.getMetadata(HNSW_INDEX_STAMP_KEY)) as { watermark: number }
+    expect(record?.watermark).toBe(4)
+    expect(index.watermark()).toBe(4)
+  })
+})

tests/unit/utils/metadataIndex-watermark.test.ts

@@ -0,0 +1,171 @@
+/**
+ * @module tests/unit/utils/metadataIndex-watermark
+ * @description Watermark-stamp pins for the metadata projection.
+ *
+ * THE LAW under test: every persisted projection artifact carries a stamp
+ * asserting "this state reflects every committed generation ≤ W and nothing
+ * above W, atomically" — written AFTER every byte it certifies is durable —
+ * and at load the owner computes the three-way verdict:
+ * stamped==committed → 'adopt' · stamped<committed → 'catchup' (gap
+ * reported) · stamped>committed OR unstamped → 'rescan', LOUDLY.
+ * Same rule, same verdict names as the shipped aggregation machinery
+ * (AggregationIndex.stateAdoptionVerdict).
+ *
+ * The verdict is COMPUTED AND EXPOSED only — these pins assert no rebuild
+ * trigger changed; acting on 'catchup' lands with the coordinator's wiring.
+ */
+import { describe, it, expect, vi, afterEach } from 'vitest'
+import { v4 as uuidv4 } from 'uuid'
+import {
+  MetadataIndexManager,
+  METADATA_INDEX_STAMP_KEY
+} from '../../../src/utils/metadataIndex.js'
+import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js'
+import { prodLog } from '../../../src/utils/logger.js'
+
+/** Fresh storage with a controllable committed generation. */
+async function makeStorage(committed: number | null): Promise<MemoryStorage> {
+  const storage = new MemoryStorage()
+  await storage.init()
+  if (committed !== null) {
+    vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed)
+  }
+  return storage
+}
+
+/** Set (or reset) the mocked committed generation on an existing storage. */
+function setCommitted(storage: MemoryStorage, committed: number): void {
+  vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed)
+}
+
+/** Session 1: index a field, optionally stamp, flush — the durable artifact. */
+async function writeArtifact(
+  storage: MemoryStorage,
+  stamp: number | null
+): Promise<void> {
+  const index = new MetadataIndexManager(storage)
+  await index.init()
+  await index.addToIndex(uuidv4(), { status: 'active', role: 'admin' })
+  if (stamp !== null) index.stampWatermark(stamp)
+  await index.flush()
+}
+
+/** Session 2: reopen on the same storage and return the loaded manager. */
+async function reopen(storage: MemoryStorage): Promise<MetadataIndexManager> {
+  const index = new MetadataIndexManager(storage)
+  await index.init()
+  return index
+}
+
+afterEach(() => {
+  vi.restoreAllMocks()
+})
+
+describe('metadata index — watermark stamp + three-way load verdict', () => {
+  it("save-with-stamp then reopen at the same committed generation → 'adopt', zero-work verdict", async () => {
+    const storage = await makeStorage(5)
+    await writeArtifact(storage, 5)
+
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('adopt')
+    expect(index.watermark()).toBe(5)
+    expect(index.watermarkGap()).toBeNull()
+  })
+
+  it("stamp BEHIND the committed generation → 'catchup' with the exact gap reported", async () => {
+    const storage = await makeStorage(5)
+    await writeArtifact(storage, 5)
+
+    // Later commits landed after the last stamped flush (unclean exit shape).
+    setCommitted(storage, 8)
+
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('catchup')
+    expect(index.watermark()).toBe(5)
+    expect(index.watermarkGap()).toEqual({ from: 5, to: 8 })
+  })
+
+  it("stamp ABOVE the committed generation → 'rescan', said out loud", async () => {
+    const storage = await makeStorage(9)
+    await writeArtifact(storage, 9)
+
+    // A truncated log on a copied store pulled the watermark back.
+    setCommitted(storage, 4)
+
+    const warnSpy = vi.spyOn(prodLog, 'warn')
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('rescan')
+    expect(index.watermarkGap()).toBeNull()
+    const said = warnSpy.mock.calls.map(c => String(c[0])).join('\n')
+    expect(said).toContain('RESCAN')
+    expect(said).toContain('ABOVE')
+  })
+
+  it("legacy unstamped artifact on a stamped store → 'rescan', LOUD — never a silent adopt", async () => {
+    const storage = await makeStorage(3)
+    await writeArtifact(storage, null) // pre-stamp brain: data flushed, no stamp
+
+    expect(await storage.getMetadata(METADATA_INDEX_STAMP_KEY)).toBeNull()
+
+    const warnSpy = vi.spyOn(prodLog, 'warn')
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('rescan')
+    expect(index.watermark()).toBeNull()
+    const said = warnSpy.mock.calls.map(c => String(c[0])).join('\n')
+    expect(said).toContain('RESCAN')
+    expect(said).toContain('unstamped')
+  })
+
+  it("a store with no committed-generation capability keeps pre-stamp behavior → 'adopt'", async () => {
+    const storage = await makeStorage(null) // committedGeneration() → null
+    await writeArtifact(storage, null)
+
+    const index = await reopen(storage)
+    expect(index.watermarkVerdict()).toBe('adopt')
+    expect(index.watermark()).toBeNull()
+  })
+
+  it('STAMP-AFTER-DATA: the stamp is the last saveMetadata of the flush, after registry and field indexes', async () => {
+    const storage = await makeStorage(2)
+    const index = new MetadataIndexManager(storage)
+    await index.init()
+    await index.addToIndex(uuidv4(), { status: 'active' })
+
+    const keys: string[] = []
+    const originalSave = storage.saveMetadata.bind(storage)
+    vi.spyOn(storage, 'saveMetadata').mockImplementation(async (id, metadata) => {
+      keys.push(id)
+      return originalSave(id, metadata)
+    })
+
+    index.stampWatermark(2)
+    await index.flush()
+
+    const stampAt = keys.indexOf(METADATA_INDEX_STAMP_KEY)
+    expect(stampAt, 'stamp record was written').toBeGreaterThanOrEqual(0)
+    expect(stampAt, 'stamp is the FINAL metadata write of the flush').toBe(keys.length - 1)
+    const registryAt = keys.indexOf('__metadata_field_registry__')
+    expect(registryAt, 'field registry written during this flush').toBeGreaterThanOrEqual(0)
+    expect(registryAt).toBeLessThan(stampAt)
+
+    // The persisted stamp record carries the required shape.
+    const record = (await storage.getMetadata(METADATA_INDEX_STAMP_KEY)) as {
+      watermark: number
+      formatVersion: number
+      stampedAt: number
+    }
+    expect(record.watermark).toBe(2)
+    expect(record.formatVersion).toBe(1)
+    expect(typeof record.stampedAt).toBe('number')
+  })
+
+  it('a flush WITHOUT a pending stamp writes no stamp record (no phantom certification)', async () => {
+    const storage = await makeStorage(2)
+    const index = new MetadataIndexManager(storage)
+    await index.init()
+    await index.addToIndex(uuidv4(), { status: 'active' })
+    await index.flush()
+
+    expect(await storage.getMetadata(METADATA_INDEX_STAMP_KEY)).toBeNull()
+  })
+})