fix(chat): hold a steering message a live run has not accepted yet

A read can land between steerRun recording the message and the worker accepting
it, and read 'unknown' then. Dropping the record there left a message the model
would still apply with no row. 'unknown' now reads as held while Sim's run is
live, as 'queued' does, and as not held once it ended, so nothing is stranded:
the sender reads again after the turn ends.

A message the user dismissed while a turn still held it shows as not delivered
again if that turn leaves it, rather than going out as a turn on its own.

Author: waleedlatif1

apps/sim/app/api/copilot/chat/steer/disposition/route.test.ts

@@ -107,17 +107,30 @@ describe('reading or dismissing a steering message', () => {
     })
   })
 
-  it('reads a message a live run never received as not delivered, without delivering it', async () => {
+  it('holds a message a live run has not accepted yet, since its steer may be on its way', async () => {
     recorded()
     worker.answers.push(Response.json({ steeringId: STEERING_ID, status: 'unknown' }))
 
     const response = await read()
 
-    expect(await response.json()).toEqual({ steeringId: STEERING_ID, status: 'not_applied' })
+    expect(await response.json()).toEqual({ steeringId: STEERING_ID, status: 'queued' })
     expect(worker.calls).toEqual([READ])
     expect(mockRecordDisposition).not.toHaveBeenCalled()
   })
 
+  it('reads a message a run ended without accepting as not delivered, for good', async () => {
+    recorded()
+    runIs('complete')
+    worker.answers.push(Response.json({ steeringId: STEERING_ID, status: 'unknown' }))
+
+    const response = await read()
+
+    expect(await response.json()).toEqual({ steeringId: STEERING_ID, status: 'not_applied' })
+    expect(mockRecordDisposition).toHaveBeenCalledWith('chat-1', [STEERING_ID], {
+      status: 'not_applied',
+    })
+  })
+
   it('reads an id the chat holds no record of as not delivered, asking the worker nothing', async () => {
     const response = await read()
 
@@ -155,8 +168,20 @@ describe('reading or dismissing a steering message', () => {
     }
   )
 
-  it('never delivers a message it dismisses, even one a live run never received', async () => {
+  it('keeps a message a live run has not accepted yet, and never delivers it', async () => {
+    recorded()
+    worker.answers.push(Response.json({ steeringId: STEERING_ID, status: 'unknown' }))
+
+    const response = await dismiss()
+
+    expect(await response.json()).toEqual({ status: 'busy', streamId: TURN_ID })
+    expect(worker.calls).toEqual([READ])
+    expect(dbChainMockFns.delete).not.toHaveBeenCalled()
+  })
+
+  it('dismisses a message once its run ended without accepting it', async () => {
     recorded()
+    runIs('complete')
     worker.answers.push(Response.json({ steeringId: STEERING_ID, status: 'unknown' }))
 
     const response = await dismiss()

apps/sim/app/api/copilot/chat/steer/route.test.ts

@@ -336,8 +336,17 @@ describe('POST /api/copilot/chat/steer', () => {
       expect(mockAppend).not.toHaveBeenCalled()
     })
 
-    it('steers it afresh once no run holds it', async () => {
+    it('waits on a live run that has not accepted it yet, never sending it twice', async () => {
       worker.read.push(Response.json({ steeringId: STEERING_ID, status: 'unknown' }))
+
+      const response = await POST(steerRequest())
+
+      expect(await response.json()).toEqual({ steeringId: STEERING_ID, status: 'queued' })
+      expect(worker.calls).toEqual([READ])
+    })
+
+    it('steers it afresh once no run holds it', async () => {
+      worker.read.push(Response.json({ steeringId: STEERING_ID, status: 'not_applied' }))
       worker.v2.push(Response.json({ steeringId: STEERING_ID, status: 'queued' }))
 
       const response = await POST(steerRequest())

apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.dom.test.tsx

@@ -4062,6 +4062,24 @@ describe('useChat remount send recovery', () => {
         expect(deliveryOf(view, id)).not.toBe('failed')
       })
 
+      it('shows a dismissed message held by a turn as not delivered once that turn leaves it', async () => {
+        state.dismissAnswers = [{ status: 'busy', streamId: 'earlier-turn' }]
+        state.steerDispositions = ['not_applied']
+        const view = await restoredChat([failedEntry])
+
+        await act(async () => {
+          view.getResult().dismissMessage(id)
+          await settle()
+        })
+        await waitFor(() => state.readBodies.length === 1)
+        await act(async () => {
+          await settle()
+        })
+
+        expect(deliveryOf(view, id)).toBe('failed')
+        expect(state.postBodies).toEqual([])
+      })
+
       it('stays not delivered when the server gave no answer', async () => {
         state.dismissAnswers = [500]
         const view = await restoredChat([failedEntry])

apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts

@@ -4787,11 +4787,14 @@ export function useChat(
           throw error
         })
         if (outcome.status === 'busy') {
+          /* The user gave it up: if that turn ends without applying it, it shows as
+             not delivered again rather than going out as a turn on its own. */
           updateOutgoing(id, {
             status: 'steered',
             streamId: outcome.streamId,
             attempts: 0,
             reads: 0,
+            noAutoSend: true,
           })
           void deliverOutgoing()
           return

apps/sim/lib/mothership/chat/application/steering-fallback.integration.ts

@@ -490,16 +490,32 @@ describe('steering fallback gate in PostgreSQL', () => {
       expect(await storedSteering(chatId, steeringId)).toEqual({ streamId: laterTurn })
     })
 
-    it('admits a turn for a message a live run never received, without delivering it', async () => {
-      const { chatId, turnId, steeringId } = await chatWithSteeredTurn()
-      await runFor(chatId, turnId, 'active')
-      worker.answers.push(Response.json({ steeringId, status: 'unknown' }))
+    it('holds a message a live run has not accepted yet, and admits it once the run ended', async () => {
+      const live = await chatWithSteeredTurn()
+      await runFor(live.chatId, live.turnId, 'active')
+      const ended = await chatWithSteeredTurn()
+      await runFor(ended.chatId, ended.turnId, 'complete')
+      worker.answers.push(
+        Response.json({ steeringId: live.steeringId, status: 'unknown' }),
+        Response.json({ steeringId: ended.steeringId, status: 'unknown' })
+      )
 
-      expect(await resolveSteeringFallback({ chatId, messageId: steeringId, userId })).toEqual({
-        kind: 'send',
+      const whileLive = await resolveSteeringFallback({
+        chatId: live.chatId,
+        messageId: live.steeringId,
+        userId,
       })
-      expect(worker.calls).toEqual([READ])
-      expect(await transcriptIds(chatId)).not.toContain(steeringId)
+      const onceEnded = await resolveSteeringFallback({
+        chatId: ended.chatId,
+        messageId: ended.steeringId,
+        userId,
+      })
+
+      expect(whileLive).toEqual({ kind: 'busy', streamId: live.turnId })
+      expect(await transcriptIds(live.chatId)).toContain(live.steeringId)
+      expect(onceEnded).toEqual({ kind: 'send' })
+      expect(await transcriptIds(ended.chatId)).not.toContain(ended.steeringId)
+      expect(worker.calls).toEqual([READ, READ])
     })
 
     it("decides from Sim's run with a worker that predates the read, never re-sending", async () => {
@@ -580,7 +596,7 @@ describe('steering fallback gate in PostgreSQL', () => {
       expect(await transcriptIds(chatId)).toContain(turnId)
     })
 
-    it('never delivers a message it dismisses, even one a live run never received', async () => {
+    it('keeps a message a live run has not accepted yet, and never delivers it', async () => {
       const { chatId, turnId, steeringId } = await chatWithSteeredTurn()
       await runFor(chatId, turnId, 'active')
       worker.answers.push(Response.json({ steeringId, status: 'unknown' }))
@@ -590,8 +606,22 @@ describe('steering fallback gate in PostgreSQL', () => {
         input: { chatId, messageId: steeringId },
       })
 
-      expect(answer).toEqual({ status: 'dismissed' })
+      expect(answer).toEqual({ status: 'busy', streamId: turnId })
       expect(worker.calls).toEqual([READ])
+      expect(await transcriptIds(chatId)).toContain(steeringId)
+    })
+
+    it('dismisses a message once its run ended without accepting it', async () => {
+      const { chatId, turnId, steeringId } = await chatWithSteeredTurn()
+      await runFor(chatId, turnId, 'complete')
+      worker.answers.push(Response.json({ steeringId, status: 'unknown' }))
+
+      const answer = await dismissSteeringMessage.execute({
+        principal: owner,
+        input: { chatId, messageId: steeringId },
+      })
+
+      expect(answer).toEqual({ status: 'dismissed' })
       expect(await transcriptIds(chatId)).not.toContain(steeringId)
     })
 

apps/sim/lib/mothership/chat/application/steering-fallback.ts

@@ -106,7 +106,9 @@ export async function steeredRunHasEnded(streamId: string, userId: string): Prom
  * can deliver it. A final answer is kept on the record and answers later reads. A
  * `queued` answer counts only while Sim still runs that turn: a run Sim recorded as
  * ended (one whose worker owner died, say) never applies it, and waiting on it
- * would hold the message forever.
+ * would hold the message forever. `unknown` (nothing accepted yet) is held the same
+ * way: the steer that recorded it may still be on its way to the worker, and a
+ * record dropped under it would leave a message the model has with no row.
  *
  * A worker without the read decides from Sim's run alone, never by re-sending the
  * message: a live run may hold it, and an ended one can no longer apply it. If it
@@ -133,8 +135,7 @@ export async function readSteeringRecord(params: {
       ? { status: 'not_held' }
       : { status: 'held' }
   }
-  if (answer.status === 'unknown') return { status: 'not_held' }
-  if (answer.status === 'queued') {
+  if (answer.status === 'queued' || answer.status === 'unknown') {
     if (!(await steeredRunHasEnded(record.streamId, userId))) return { status: 'held' }
     await recordSteeringDisposition(chatId, [messageId], { status: 'not_applied' })
     return { status: 'not_held' }