fix(chat): close the re-audit's delivery gaps

- a message not delivered whose id became a turn in the chat is dropped, not shown
  twice, with or without a streamId, and a history read wakes the settle for it
- a fork neither cuts at nor copies a steering message no turn applied, so the
  worker's cut cannot 404 and a copy cannot hold a dead bubble
- a journal entry stamps writtenAt; one still on its way past the age limit restores
  as not delivered, with no network call
- a dismiss the worker answers with a hash conflict removes the record and the message
- the storage adapter names sessionStorage inside its try, including at module load
- disposition reads get their own counter again: a message read as not applied goes
  out without inheriting read backoff, and reads leave the send budget alone

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/home/hooks/outgoing-state.test.ts

@@ -21,6 +21,7 @@ import {
   fromJournaled,
   hasOutgoingFollowUp,
   holdAfterErroredTurn,
+  isDeliveredAsTurn,
   MAX_OUTGOING_DELIVERY_ATTEMPTS,
   nextOutgoingStep,
   ORPHAN_AUTO_SEND_MAX_AGE_MS,
@@ -40,6 +41,8 @@ function entry(id: string, patch: Partial<OutgoingMessage> = {}): OutgoingMessag
     display: { id, role: 'user', content: id },
     status: 'pending',
     attempts: 0,
+    reads: 0,
+    writtenAt: NOW,
     ...patch,
   }
 }
@@ -222,22 +225,67 @@ describe('transitions', () => {
     ).toBe('failed')
   })
 
-  it('reads a held message again later, counting toward the one delivery cap', () => {
+  it('reads a held message again later, and gives up at its own cap', () => {
     const context = { streamId: 'turn', turnErrored: false, now: NOW }
     const steered = entry('a', { status: 'steered', streamId: 'turn' })
     const later = afterDispositionRead(steered, { status: 'queued' }, context)
-    expect(later).toMatchObject({ attempts: 1, unconfirmed: false })
+    expect(later).toMatchObject({ reads: 1, unconfirmed: false })
     expect(later.notBefore).toBeGreaterThan(NOW)
     expect(afterDispositionRead(steered, { status: 'unknown' }, context).unconfirmed).toBe(true)
     expect(
       afterDispositionRead(
-        { ...steered, attempts: MAX_OUTGOING_DELIVERY_ATTEMPTS - 1 },
+        { ...steered, reads: MAX_OUTGOING_DELIVERY_ATTEMPTS - 1 },
         { status: 'queued' },
         context
       ).status
     ).toBe('failed')
   })
 
+  it('sends a message read as not applied at once, whatever the reads before it took', () => {
+    const steered = entry('a', { status: 'steered', streamId: 'turn', reads: 6 })
+    const patch = afterDispositionRead(
+      steered,
+      { status: 'not_applied' },
+      { streamId: 'turn', turnErrored: false, now: NOW }
+    )
+    expect(patch.status).toBe('pending')
+    expect(patch.notBefore).toBeUndefined()
+  })
+
+  it('still backs off a message the chat kept refusing as busy', () => {
+    const steered = entry('a', { status: 'steered', streamId: 'turn', attempts: 2 })
+    const { notBefore } = afterDispositionRead(
+      steered,
+      { status: 'not_applied' },
+      { streamId: 'turn', turnErrored: false, now: NOW }
+    )
+    expect(notBefore).toBeGreaterThan(NOW)
+  })
+
+  it('leaves reads out of the budget for sends', () => {
+    const read = afterDispositionRead(
+      entry('a', { status: 'steered', streamId: 'turn' }),
+      { status: 'queued' },
+      { streamId: 'turn', turnErrored: false, now: NOW }
+    )
+    expect(read.attempts).toBeUndefined()
+    const sent = afterUnsentAttempt(entry('a', { reads: MAX_OUTGOING_DELIVERY_ATTEMPTS }), {
+      online: true,
+      now: NOW,
+    })
+    expect(sent).toMatchObject({ status: 'pending', attempts: 1 })
+  })
+
+  it('counts a message not delivered or steered as delivered once its id is a turn', () => {
+    const admitted = new Set(['a'])
+    expect(isDeliveredAsTurn(entry('a', { status: 'failed' }), admitted)).toBe(true)
+    expect(isDeliveredAsTurn(entry('a', { status: 'failed', streamId: 'turn' }), admitted)).toBe(
+      true
+    )
+    expect(isDeliveredAsTurn(entry('a', { status: 'steered' }), admitted)).toBe(true)
+    expect(isDeliveredAsTurn(entry('b', { status: 'failed' }), admitted)).toBe(false)
+  })
+
   it('reads a steered message again on Retry, and sends one that never reached a turn', () => {
     expect(afterRetry(entry('a', { status: 'failed', streamId: 'turn' })).status).toBe('steered')
     expect(afterRetry(entry('a', { status: 'failed', refusedBy: 'turn' })).status).toBe('pending')
@@ -268,7 +316,7 @@ describe('transitions', () => {
 describe('journal', () => {
   it('restores a send or steer that was out as possibly held, so it cannot steer again', () => {
     for (const status of ['sending', 'steering'] as const) {
-      const restored = fromJournaled('chat', toJournaled(entry('a', { status })))
+      const restored = fromJournaled('chat', toJournaled(entry('a', { status })), NOW)
       expect(restored.unconfirmed).toBe(true)
     }
   })
@@ -278,7 +326,8 @@ describe('journal', () => {
       'chat',
       toJournaled(
         entry('a', { status: 'applied', streamId: 'turn', textOffset: 4, refusedBy: 'earlier' })
-      )
+      ),
+      NOW
     )
     expect(restored).toMatchObject({
       status: 'applied',
@@ -289,6 +338,27 @@ describe('journal', () => {
   })
 })
 
+describe('a journal restored past the age limit', () => {
+  const restore = (status: OutgoingMessage['status'], age: number) =>
+    fromJournaled(
+      'chat',
+      toJournaled(entry('a', { status, streamId: 'turn', writtenAt: NOW - age })),
+      NOW
+    ).status
+
+  it('shows a message on its way as not delivered', () => {
+    expect(restore('pending', ORPHAN_AUTO_SEND_MAX_AGE_MS)).toBe('failed')
+    expect(restore('steered', ORPHAN_AUTO_SEND_MAX_AGE_MS)).toBe('failed')
+    expect(restore('sending', ORPHAN_AUTO_SEND_MAX_AGE_MS)).toBe('failed')
+  })
+
+  it('resumes a recent one where it stood, and leaves one already applied alone', () => {
+    expect(restore('pending', ORPHAN_AUTO_SEND_MAX_AGE_MS - 1)).toBe('pending')
+    expect(restore('steered', ORPHAN_AUTO_SEND_MAX_AGE_MS - 1)).toBe('steered')
+    expect(restore('applied', ORPHAN_AUTO_SEND_MAX_AGE_MS)).toBe('applied')
+  })
+})
+
 describe('deliveryOf', () => {
   it('says a message waits for the turn only when it cannot steer it', () => {
     expect(deliveryOf(entry('a', { unconfirmed: true }), true)).toBe('waiting')

apps/sim/app/workspace/[workspaceId]/home/hooks/outgoing-state.ts

@@ -13,8 +13,10 @@ import type { ChatContext } from '@/stores/panel'
 
 /** The worker's cap on steering text; a longer message goes out as a turn. */
 const MAX_STEERING_CONTENT_LENGTH = 32_768
-/** Rounds that came back without delivering a message before it shows as not delivered. */
+/** Sends that came back without delivering a message before it shows as not delivered. */
 export const MAX_OUTGOING_DELIVERY_ATTEMPTS = 8
+/** Reads of a turn's end that still found a message held before it shows as not delivered. */
+const MAX_DISPOSITION_READS = 8
 const OUTGOING_RETRY_BASE_MS = 1000
 const OUTGOING_RETRY_MAX_MS = 30_000
 /**
@@ -68,8 +70,12 @@ export interface OutgoingMessage {
   textOffset?: number
   /** A turn that refused it: it waits for that turn to end rather than steering it again. */
   refusedBy?: string
-  /** Delivery rounds that came back without delivering it, whichever way they did. */
+  /** Sends that came back without delivering it (no answer, or the chat was busy). */
   attempts: number
+  /** Reads after its turn ended that found it still held; they say nothing about a send. */
+  reads: number
+  /** Epoch ms it was written, which decides whether it may still go out on its own. */
+  writtenAt: number
   /** Epoch ms before which it is not tried again. */
   notBefore?: number
 }
@@ -141,15 +147,30 @@ export function toJournaled(message: OutgoingMessage): JournaledOutgoing {
       ? { unconfirmed: true }
       : {}),
     ...(message.textOffset !== undefined ? { textOffset: message.textOffset } : {}),
+    writtenAt: message.writtenAt,
   }
 }
 
-export function fromJournaled(chatKey: string, entry: JournaledOutgoing): OutgoingMessage {
+/**
+ * The entry a journal restores. One on its way that has sat past the age limit (a
+ * browser session restored days later) shows as not delivered, with no network call
+ * and nothing going out until the user asks.
+ */
+export function fromJournaled(
+  chatKey: string,
+  entry: JournaledOutgoing,
+  now: number
+): OutgoingMessage {
+  const stillOnItsWay = entry.status === 'pending' || entry.status === 'steered'
   return {
     ...entry,
+    ...(stillOnItsWay && now - entry.writtenAt >= ORPHAN_AUTO_SEND_MAX_AGE_MS
+      ? { status: 'failed' as const }
+      : {}),
     chatKey,
     display: optimisticUserMessageFor(entry.id, entry.payload),
     attempts: 0,
+    reads: 0,
   }
 }
 
@@ -176,6 +197,19 @@ export function admittedMessageIds(history: MothershipChatHistory | undefined):
   return admitted
 }
 
+/**
+ * Whether the chat's history shows this message is a turn of its own, whatever this
+ * tab last knew: another tab sent it, or a send that got no answer was admitted.
+ */
+export function isDeliveredAsTurn(
+  message: OutgoingMessage,
+  admittedIds: ReadonlySet<string>
+): boolean {
+  return (
+    (message.status === 'failed' || message.status === 'steered') && admittedIds.has(message.id)
+  )
+}
+
 export type OutgoingStep =
   /** Nothing can go now: nothing is pending, or what is first must wait. */
   | { kind: 'idle' }
@@ -281,13 +315,13 @@ export function afterDispositionRead(
   }
   /* Still held (or no answer): read again later. A run that never settles it, such
      as one whose owner died, does not keep it waiting forever. */
-  const attempts = message.attempts + 1
-  return attempts >= MAX_OUTGOING_DELIVERY_ATTEMPTS
-    ? { status: 'failed', attempts }
+  const reads = message.reads + 1
+  return reads >= MAX_DISPOSITION_READS
+    ? { status: 'failed', reads }
     : {
-        attempts,
+        reads,
         unconfirmed: outcome.status === 'unknown',
-        notBefore: context.now + outgoingRetryDelay(attempts),
+        notBefore: context.now + outgoingRetryDelay(reads),
       }
 }
 
@@ -300,6 +334,7 @@ export function afterRetry(message: OutgoingMessage): Partial<OutgoingMessage> {
   return {
     status: message.streamId ? 'steered' : 'pending',
     attempts: 0,
+    reads: 0,
     notBefore: undefined,
   }
 }
@@ -363,16 +398,18 @@ export function adoptOrphanRow(row: PersistedRow, chatKey: string, now: number):
     ...(contexts?.length ? { contexts } : {}),
     ...(row.requestMode ? { requestMode: row.requestMode } : {}),
   })
-  const age = now - Date.parse(row.timestamp)
+  const writtenAt = Date.parse(row.timestamp) || 0
   return {
     id: row.id,
     chatKey,
     payload,
     display: optimisticUserMessageFor(row.id, payload),
-    status: age < ORPHAN_AUTO_SEND_MAX_AGE_MS ? 'steered' : 'failed',
+    status: now - writtenAt < ORPHAN_AUTO_SEND_MAX_AGE_MS ? 'steered' : 'failed',
     streamId: row.steering?.streamId,
     unconfirmed: true,
     attempts: 0,
+    reads: 0,
+    writtenAt,
   }
 }