A rejected seed promise poisons a session into silent total row loss

hypaware-core/plugins-workspace/ai-gateway/src/message_projector.js

@@ -272,6 +272,11 @@ export function createAiGatewayMessageProjector(opts) {
  * `onExchangeFinished` without serializing (proxy.js stores the returned
  * promise but does not await it), so this overlap is real.
  *
+ * This is also where the seed path's "a seeding miss must never cost a row"
+ * guarantee is ENFORCED rather than merely documented: whatever the seed
+ * rejects with is absorbed here, for the whole path, including the parts of
+ * it that reach storage the leaf scan does not own.
+ *
  * @param {string} sessionId
  * @param {ReturnType<typeof createAiGatewayConversationState>} state
  * @param {Map<string, Promise<void>>} seedPromises
@@ -286,7 +291,36 @@ function seedSeenMessagesForSession(sessionId, state, seedPromises, storage, log
   if (!pending) {
     // This body runs to here synchronously (no prior await), so the map is
     // populated before any concurrent caller can observe a missing entry.
-    pending = seedSessionIfCommitted(sessionId, state, storage, log, sessionIndex)
+    //
+    // `projectExchange` awaits this and `source.js` drops the row for
+    // whatever it rejects with, so a seed that could not run must settle as
+    // "seeded nothing", never as a rejection: a seeding miss risks only the
+    // duplicate the seed exists to prevent (which settlement/compaction
+    // still collapse), while a rejection costs a real row.
+    //
+    // Nothing ever REWRITES a memo entry, so absorbing the rejection alone
+    // would cache "could not seed" as this session's verdict for the
+    // listener's lifetime. That is right for a scan that RAN and came back
+    // empty-handed, and wrong for one that broke: every later exchange
+    // would short-circuit onto the memo and inherit a verdict no scan ever
+    // produced. So drop the memo too, and let the next exchange retry and
+    // re-warn. The retry costs a scan that fails the way this one did (the
+    // caching in LLP 0204 is there to spare the daemon whole-table scans
+    // that SUCCEED), and buys back the operator signal whose absence made
+    // this silent: without it, one broken session logged once and then went
+    // quiet while every row for it was dropped.
+    // @ref LLP 0204#fix [constrained-by]: the per-session seed exists to
+    //   save a scan, so nothing in it may end an exchange or outlive itself
+    pending = seedSessionIfCommitted(sessionId, state, storage, log, sessionIndex).catch((err) => {
+      log?.warn?.('aigw.seed_seen_messages_failed', {
+        session_id: sessionId,
+        error_kind: 'seed_rejected',
+        error: err instanceof Error ? err.message : String(err),
+      })
+      // Guarded so a concurrent caller's newer memo is not evicted by this
+      // one's failure.
+      if (seedPromises.get(sessionId) === pending) seedPromises.delete(sessionId)
+    })
     seedPromises.set(sessionId, pending)
   }
   return pending
@@ -445,11 +479,21 @@ async function scanCommittedSessionIds(storage, log) {
  * their `message_id`s into `state.seenMessages`.
  *
  * Best-effort throughout: a missing storage handle (unit-test stubs), a
- * missing table, or an unreadable partition degrades to "not seeded" and
- * NEVER throws (a seeding miss only risks the duplicate this guards
- * against (which settlement/compaction can still collapse), whereas
- * throwing would drop a real row). The promise still resolves on a
- * partial/failed scan, so it is cached and not retried on every exchange.
+ * missing table, or an unreadable partition degrades to "not seeded" (a
+ * seeding miss only risks the duplicate this guards against (which
+ * settlement/compaction can still collapse), whereas failing the exchange
+ * would drop a real row). The promise still resolves on a partial/failed
+ * scan, so it is cached and not retried on every exchange.
+ *
+ * It does NOT, however, promise never to throw, and used to claim it did:
+ * only the `discoverCachePartitions` CALL is guarded below, not the walk
+ * over the answer, so a storage that resolves a truthy NON-iterable (a
+ * violation of its own declared return type) throws out of this function.
+ * The guarantee callers actually need is that seeding never costs a row,
+ * and that is enforced one level up in `seedSeenMessagesForSession`, which
+ * absorbs a rejection from anywhere in the seed path (this scan, or the
+ * committed-session index it consults first) rather than resting on a
+ * contract each leaf asserts about itself.
  *
  * @param {string} sessionId
  * @param {ReturnType<typeof createAiGatewayConversationState>} state
@@ -467,6 +511,12 @@ async function scanCommittedMessageIds(sessionId, state, storage, log) {
   } catch (err) {
     log?.warn?.('aigw.seed_seen_messages_failed', {
       session_id: sessionId,
+      // `error_kind` separates the two ways this message is reached,
+      // because they call for opposite responses: `discover_failed` is an
+      // I/O condition this scan handled and the next session may not hit,
+      // while `seed_rejected` (from the caller) means the seed path broke
+      // and will keep breaking until someone fixes it.
+      error_kind: 'discover_failed',
       error: err instanceof Error ? err.message : String(err),
     })
     return

test/plugins/ai-gateway-message-projector.test.js

@@ -1104,6 +1104,118 @@ test('committed-session index: a build that could not scan is not cached as "no
   )
 })
 
+test('seed failure: a storage that breaks its discover contract loses no rows and does not poison the session memo', async () => {
+  // Finding (#692), two halves of one silent failure:
+  //
+  //  - `scanCommittedMessageIds` guards only the `discoverCachePartitions`
+  //    CALL; the walk over the answer sits outside that try/catch, so a
+  //    storage resolving a truthy NON-iterable (a violation of its own
+  //    declared `CachePartitionMeta[]` return type) throws out of a
+  //    function whose whole point is to degrade rather than cost a row.
+  //  - `seedPromises` memoized that rejected promise and never removed it,
+  //    so `projectExchange` rejected, `source.js` caught it and dropped the
+  //    row, and EVERY later exchange for the session short-circuited onto
+  //    the poisoned memo and was dropped with no warn at all. Measured in
+  //    review: five exchanges, two warn lines, zero rows.
+  //
+  // So both properties are asserted per exchange: the row still lands, and
+  // the failure keeps saying so instead of going quiet after the first.
+  //
+  // The index build (discover call 1) is kept WELL-FORMED on purpose. A
+  // storage malformed on that call too rejects inside the committed-session
+  // index, which is issue #685 / PR #690's separate defect; keeping it
+  // well-formed isolates this one and keeps this test independent of that
+  // fix.
+  let discoverCalls = 0
+  const storage = /** @type {ExtendedQueryStorageService} */ (/** @type {unknown} */ ({
+    async discoverCachePartitions() {
+      discoverCalls++
+      if (discoverCalls === 1) {
+        return [{ dataset: 'ai_gateway_messages', partition: {}, path: '/p', epoch: 0, rowCount: 1 }]
+      }
+      return /** @type {never} */ ({ malformed: true })
+    },
+    async *readRows() {
+      // The index must place this session among the committed ones, or the
+      // per-session scan is skipped and the defect never fires.
+      yield { session_id: 'sess-broken-seed', message_id: 'uuid-committed' }
+    },
+  }))
+  /** @type {Array<{ level: string, message: string, fields: Record<string, unknown> }>} */
+  const logged = []
+  const projector = createAiGatewayMessageProjector({
+    gatewayId: 'gw-test',
+    projectors: [registered('native', { project: perExchangeMessage })],
+    storage,
+    log: collectingLogger(logged),
+  })
+
+  const first = await settledProjection(
+    projector.projectExchange({ ...exchange(), exchange_id: 'ex-1', path: 'sess-broken-seed' })
+  )
+  assert.equal(
+    first.error === undefined ? undefined : String(first.error),
+    undefined,
+    'exchange 1: a seed that could not run must not fail the projection'
+  )
+  assert.equal(first.rows?.length, 1, 'exchange 1: the row survives a seed that could not run')
+
+  const second = await settledProjection(
+    projector.projectExchange({ ...exchange(), exchange_id: 'ex-2', path: 'sess-broken-seed' })
+  )
+  assert.equal(
+    second.error === undefined ? undefined : String(second.error),
+    undefined,
+    'exchange 2: a memoized failure must not fail every later exchange for the session'
+  )
+  assert.equal(second.rows?.length, 1, 'exchange 2: the session is not permanently poisoned')
+
+  // 3 = one index build + one seed scan per exchange. Exchange 2 having
+  // re-run its scan is the direct evidence that no failed memo survived it.
+  assert.equal(discoverCalls, 3, 'a seed that broke is retried, not cached as this session verdict')
+
+  // Silence was the other half of the defect: a daemon dropping every row
+  // for a session while logging nothing is the failure operators could not
+  // see. Each failing exchange must carry its own signal.
+  const warns = logged.filter((e) => e.level === 'warn' && e.message === 'aigw.seed_seen_messages_failed')
+  assert.equal(warns.length, 2, 'each exchange whose seed failed emits its own operator signal')
+  for (const warn of warns) {
+    assert.equal(warn.fields.error_kind, 'seed_rejected')
+    assert.equal(warn.fields.session_id, 'sess-broken-seed')
+    assert.match(String(warn.fields.error), /not iterable/)
+  }
+})
+
+/**
+ * A projector whose message_id is unique per exchange. Reusing one id
+ * across exchanges would let the seen-set dedup (the very thing the seed
+ * feeds) suppress the second row legitimately, hiding a drop under a zero.
+ *
+ * @param {AiGatewayExchangeInput} input
+ * @returns {AiGatewayProjectedExchange}
+ */
+function perExchangeMessage(input) {
+  return {
+    provider: 'native',
+    session_id: String(input.path),
+    messages: [{ role: 'user', content: 'hi', message_id: `uuid-${input.exchange_id}` }],
+  }
+}
+
+/**
+ * Settle a projection without letting a rejection escape, so a test can
+ * assert on "it rejected" as evidence instead of dying on it.
+ *
+ * @param {Promise<unknown[]>} projecting
+ * @returns {Promise<{ rows: unknown[] | undefined, error: unknown }>}
+ */
+function settledProjection(projecting) {
+  return projecting.then(
+    (rows) => ({ rows, error: undefined }),
+    (/** @type {unknown} */ error) => ({ rows: undefined, error })
+  )
+}
+
 /**
  * Minimal `ExtendedQueryStorageService`-shaped stub exposing only the
  * committed-partition read surface the projector feature-detects: