feat(hub,web): support scheduling messages for future delivery (#590)

cli/src/api/apiSession.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from 'vitest'
-import { isExternalUserMessage } from './apiSession'
+import { isExternalUserMessage, IncomingMessageFilter } from './apiSession'
 
 describe('isExternalUserMessage', () => {
     const baseUserMsg = {
@@ -96,3 +96,70 @@ describe('isExternalUserMessage', () => {
         ).toBe(false)
     })
 })
+
+describe('IncomingMessageFilter (HAPI Bot R3 finding #1)', () => {
+    it('accepts a mature scheduled message whose seq is below the latest cursor', () => {
+        // schedule seq=10, immediate seq=11 acks first → cursor=11.
+        // seq=10 matures: seq-only dedup would drop it; id-based dedup must accept.
+        const filter = new IncomingMessageFilter()
+        expect(filter.accept({ id: 'msg-imm', seq: 11 })).toBe(true)
+        expect(filter.accept({ id: 'msg-sched', seq: 10 })).toBe(true)
+    })
+
+    it('rejects an exact id duplicate (re-emit on the next mature tick)', () => {
+        const filter = new IncomingMessageFilter()
+        expect(filter.accept({ id: 'msg-1', seq: 1 })).toBe(true)
+        expect(filter.accept({ id: 'msg-1', seq: 1 })).toBe(false)
+    })
+
+    it('falls back to seq-only dedup for messages without an id', () => {
+        const filter = new IncomingMessageFilter()
+        expect(filter.accept({ seq: 5 })).toBe(true)
+        // seq <= cursor and no id → drop (legacy behaviour preserved).
+        expect(filter.accept({ seq: 4 })).toBe(false)
+        expect(filter.accept({ seq: 5 })).toBe(false)
+    })
+
+    it('advances cursorSeq monotonically regardless of arrival order', () => {
+        const filter = new IncomingMessageFilter()
+        filter.accept({ id: 'a', seq: 11 })
+        filter.accept({ id: 'b', seq: 10 })
+        expect(filter.cursorSeq()).toBe(11)
+    })
+
+    it('bounds the seen-id set to the configured capacity (LRU eviction)', () => {
+        const filter = new IncomingMessageFilter(3)
+        filter.accept({ id: 'a', seq: 1 })
+        filter.accept({ id: 'b', seq: 2 })
+        filter.accept({ id: 'c', seq: 3 })
+        filter.accept({ id: 'd', seq: 4 })
+        // 'a' should have been evicted — re-presenting it is treated as new.
+        expect(filter.accept({ id: 'a', seq: 5 })).toBe(true)
+        // 'd' is still in the set.
+        expect(filter.accept({ id: 'd', seq: 6 })).toBe(false)
+    })
+
+    it('refreshes recency on dedup hit so re-emits survive bursts of unrelated ids', () => {
+        // Models the documented contract: the hub re-emits the same id every 5 s
+        // until the CLI acks.  If the dedup were FIFO (insert-order only), a
+        // burst of capacity-many unrelated ids between re-emits would evict the
+        // pending id and the next re-emit would double-deliver.
+        const filter = new IncomingMessageFilter(3)
+        // Pre-fill so 'pending' is not at the head.
+        filter.accept({ id: 'a', seq: 1 })
+        filter.accept({ id: 'pending', seq: 2 })
+        filter.accept({ id: 'b', seq: 3 })
+        // Re-emit pending → recency refresh moves it to the tail.
+        expect(filter.accept({ id: 'pending', seq: 4 })).toBe(false)
+        // Burst that evicts oldest entries.  Without the refresh 'pending' would
+        // be at insert position 2 and would be evicted; with the refresh it is
+        // now the newest entry and survives.
+        filter.accept({ id: 'c', seq: 5 })
+        filter.accept({ id: 'd', seq: 6 })
+        // 'a' (oldest) and then 'b' should have been evicted; 'pending' must
+        // still dedup.
+        expect(filter.accept({ id: 'pending', seq: 7 })).toBe(false)
+        expect(filter.accept({ id: 'a', seq: 8 })).toBe(true)
+        expect(filter.accept({ id: 'b', seq: 9 })).toBe(true)
+    })
+})

cli/src/api/apiSession.ts

@@ -71,6 +71,64 @@ export function isExternalUserMessage(body: RawJSONLines): body is Extract<RawJS
     return true
 }
 
+/**
+ * Dedup filter for messages arriving on the realtime socket and via reconnect
+ * backfill.  Keyed by message id (with a bounded LRU) and falls back to the
+ * legacy seq cursor for messages that lack an id.
+ *
+ * Why id-first: scheduled messages keep the seq assigned at insertion time, so
+ * a row scheduled for T+1h (seq=10) can be released after a later immediate
+ * message (seq=11) has already advanced the cursor.  A pure seq <= cursor
+ * filter would silently drop the mature emit.  See HAPI Bot R3 finding #1.
+ */
+export class IncomingMessageFilter {
+    private readonly seenIds = new Set<string>()
+    private readonly capacity: number
+    private lastSeenSeq: number | null = null
+
+    constructor(capacity = 256) {
+        this.capacity = capacity
+    }
+
+    cursorSeq(): number | null {
+        return this.lastSeenSeq
+    }
+
+    /** Returns true if this message should be processed; false to drop as a duplicate. */
+    accept(message: { id?: string | null; seq?: number | null }): boolean {
+        const id = typeof message.id === 'string' && message.id.length > 0 ? message.id : null
+        if (id && this.seenIds.has(id)) {
+            // Refresh recency: the hub re-emits the same id every 5 s until the
+            // CLI acks (releaseMatureScheduledMessages contract).  Without a
+            // delete+re-add the entry stays at its first-insert position and can
+            // be evicted by a burst of unrelated ids before the ack lands —
+            // the next re-emit would then be treated as new and double-deliver.
+            this.seenIds.delete(id)
+            this.seenIds.add(id)
+            return false
+        }
+
+        const seq = typeof message.seq === 'number' ? message.seq : null
+        if (!id && seq !== null && this.lastSeenSeq !== null && seq <= this.lastSeenSeq) {
+            return false
+        }
+
+        if (id) {
+            this.seenIds.add(id)
+            if (this.seenIds.size > this.capacity) {
+                // Set iteration is insertion-ordered; with delete+re-add on dedup hit
+                // (above) this becomes a true LRU eviction.
+                const oldest = this.seenIds.values().next().value
+                if (oldest !== undefined) this.seenIds.delete(oldest)
+            }
+        }
+        if (seq !== null) {
+            this.lastSeenSeq = Math.max(this.lastSeenSeq ?? 0, seq)
+        }
+        return true
+    }
+}
+
 export class ApiSessionClient extends EventEmitter {
     private readonly token: string
     readonly sessionId: string
@@ -82,7 +140,7 @@ export class ApiSessionClient extends EventEmitter {
     private pendingMessages: { message: UserMessage; localId?: string }[] = []
     private pendingMessageCallback: ((message: UserMessage, localId?: string) => void) | null = null
     private cancelQueuedMessageCallback: ((localId: string) => boolean) | null = null
-    private lastSeenMessageSeq: number | null = null
+    private readonly incomingFilter = new IncomingMessageFilter()
     private backfillInFlight: Promise<void> | null = null
     private needsBackfill = false
     private hasConnectedOnce = false
@@ -274,13 +332,9 @@ export class ApiSessionClient extends EventEmitter {
         }
     }
 
-    private handleIncomingMessage(message: { seq?: number; localId?: string | null; content: unknown }): void {
-        const seq = typeof message.seq === 'number' ? message.seq : null
-        if (seq !== null) {
-            if (this.lastSeenMessageSeq !== null && seq <= this.lastSeenMessageSeq) {
-                return
-            }
-            this.lastSeenMessageSeq = seq
+    private handleIncomingMessage(message: { id?: string; seq?: number; localId?: string | null; content: unknown }): void {
+        if (!this.incomingFilter.accept({ id: message.id, seq: message.seq })) {
+            return
         }
 
         const userResult = UserMessageSchema.safeParse(message.content)
@@ -311,7 +365,7 @@ export class ApiSessionClient extends EventEmitter {
             return
         }
 
-        const startSeq = this.lastSeenMessageSeq
+        const startSeq = this.incomingFilter.cursorSeq()
         if (startSeq === null) {
             logger.debug('[API] Skipping backfill because no last-seen message sequence is available')
             return
@@ -353,7 +407,7 @@ export class ApiSessionClient extends EventEmitter {
                     this.handleIncomingMessage(message)
                 }
 
-                const observedSeq = this.lastSeenMessageSeq ?? maxSeq
+                const observedSeq = this.incomingFilter.cursorSeq() ?? maxSeq
                 const nextCursor = Math.max(maxSeq, observedSeq)
                 if (nextCursor <= cursor) {
                     logger.debug('[API] Backfill stopped due to non-advancing cursor', {