From eb154f01e66b2ac9ba03a6b9f7777f9673049ff0 Mon Sep 17 00:00:00 2001 From: Matt Rubens <2600+mrubens@users.noreply.github.com> Date: Fri, 4 Sep 2026 00:52:04 -0400 Subject: [PATCH 1/3] [Fix] Reactions and platform events no longer supersede parked Fast turns Every inline admission discarded the conversation's older pending inline rows on the assumption that a newer human message stands in for the earlier request. #2156 routed emoji reactions and web platform events through the same path, so a reaction while a turn was parked for a retry, or waiting to resume after an interruption, silently dropped that turn and turned its retry notice into an interruption message. Only typed human messages supersede now. The next-turn and settle reconciles also leave a retry notice alone while another durable row for the conversation is still pending, so the resumed run edits it into the answer instead of posting beside a false interruption. --- .changeset/fast-reaction-supersession.md | 5 + ...fast-agent-conversation-repository.test.ts | 104 ++++++++++++++++++ .../__tests__/fast-agent-service.test.ts | 14 +-- .../fast-agent-conversation-repository.ts | 34 +++++- .../server/fast-agent/fast-agent-service.ts | 2 + .../lib/fast-agent-human-follow-up.test.ts | 61 ++++++++++ .../server/lib/fast-agent-human-follow-up.ts | 57 ++++++---- 7 files changed, 249 insertions(+), 28 deletions(-) create mode 100644 .changeset/fast-reaction-supersession.md diff --git a/.changeset/fast-reaction-supersession.md b/.changeset/fast-reaction-supersession.md new file mode 100644 index 0000000000..df333f1aef --- /dev/null +++ b/.changeset/fast-reaction-supersession.md @@ -0,0 +1,5 @@ +--- +'@roomote/web': patch +--- + +Emoji reactions and web platform events (setup kickoffs, input responses) no longer supersede a Fast turn that is parked for an inference retry or waiting to resume after an interruption. Every inline admission used to discard the conversation's older pending turn rows on the assumption that a newer human message stands in for the earlier request; a reaction or platform event does not, so the earlier question was silently dropped and its retry notice was turned into an interruption message. Those turns now keep their row and resume once the conversation is idle again, and a turn's entry and settle reconciles leave a retry notice alone while another durable row for the conversation is still pending, so the resumed run edits it into the answer instead of posting beside a false interruption. Typed human messages still supersede as before. diff --git a/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts b/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts index 8eb0c74d4b..b7c2c7f831 100644 --- a/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts +++ b/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts @@ -1669,6 +1669,110 @@ describe('Fast conversation repository', () => { }); }); + it('does not stamp a retry notice whose turn still has a pending durable row', async () => { + const user = await createUser(); + const conversation = { + surface: 'web' as const, + workspaceId: user.id, + conversationId: crypto.randomUUID(), + }; + const session = await fastAgentConversationRepository.getOrCreate({ + userId: user.id, + conversation, + }); + await fastAgentConversationRepository.upsertMessage({ + conversationId: session.id, + message: { + eventId: 'turn-parked:retry-notice:0', + turnId: 'turn-parked', + turnSeq: 1, + ts: 100, + eventType: 'roomote_runtime.assistant_message', + role: 'assistant', + contentBlocks: [{ type: 'text', text: 'Retrying in 45s' }], + metadata: { + visibleInTranscript: true, + purpose: 'progress', + inferenceRetryNotice: true, + inferenceRetryActive: true, + }, + payload: { purpose: 'progress' }, + source: 'web', + }, + }); + const parent = { sessionId: session.id, conversation }; + // The parked turn that owns the notice: claim released, retry scheduled. + const [parked] = await db + .insert(fastAgentParentEvents) + .values({ + conversationId: session.id, + eventKey: `parked-${session.id}`, + parent, + event: { type: 'human_follow_up', eventId: 'parked' }, + admission: 'inline', + retryAt: new Date(Date.now() + 60_000), + }) + .returning({ id: fastAgentParentEvents.id }); + // A reaction turn admitted beside it, under its own claim. + const [reaction] = await db + .insert(fastAgentParentEvents) + .values({ + conversationId: session.id, + eventKey: `reaction-${session.id}`, + parent, + event: { type: 'human_follow_up', eventId: 'reaction' }, + admission: 'inline', + claimedUntil: new Date(Date.now() + 60_000), + }) + .returning({ id: fastAgentParentEvents.id }); + const readNotice = async () => { + const [notice] = await db + .select({ metadata: fastAgentMessages.metadata }) + .from(fastAgentMessages) + .where( + and( + eq(fastAgentMessages.conversationId, session.id), + eq(fastAgentMessages.eventId, 'turn-parked:retry-notice:0'), + ), + ); + return notice!.metadata; + }; + + // The reaction turn's entry reconcile sees the parked row and leaves the + // notice for the parked turn's resumed run to edit. + await expect( + reconcileFastAgentInferenceRetryNotices( + session.id, + 'next_turn_reconcile', + { + excludeEventId: reaction!.id, + }, + ), + ).resolves.toBe(0); + expect(await readNotice()).toMatchObject({ inferenceRetryActive: true }); + + // Once the parked turn settles, only the caller's own row remains, and + // that never counts as a pending turn that owns the notice. + await db + .update(fastAgentParentEvents) + .set({ deliveredAt: new Date() }) + .where(eq(fastAgentParentEvents.id, parked!.id)); + await expect( + reconcileFastAgentInferenceRetryNotices( + session.id, + 'next_turn_reconcile', + { + excludeEventId: reaction!.id, + }, + ), + ).resolves.toBe(1); + expect(await readNotice()).toMatchObject({ + inferenceRetryActive: false, + purpose: 'closeout', + interruptionReason: 'next_turn_reconcile', + }); + }); + it('renews only a live responding lease', async () => { const user = await createUser(); const session = await fastAgentConversationRepository.getOrCreate({ diff --git a/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-service.test.ts b/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-service.test.ts index b576f1ef31..d4bef71519 100644 --- a/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-service.test.ts +++ b/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-service.test.ts @@ -4470,10 +4470,9 @@ describe('answerFastAgentQuestion native OpenCode tools', () => { expect(mocks.markDurableDelivered).not.toHaveBeenCalled(); expect(mocks.releaseDurableClaim).not.toHaveBeenCalled(); expect(mocks.revokeDurableReplay).not.toHaveBeenCalled(); - expect(mocks.reconcileRetryNotices).not.toHaveBeenCalledWith( - expect.anything(), - 'turn_settled_reconcile', - ); + expect( + mocks.reconcileRetryNotices.mock.calls.map((call) => call[1]), + ).not.toContain('turn_settled_reconcile'); expect( mocks.touchSessionActivity.mock.calls.map((call) => call[3]), ).not.toContainEqual({ respondingUntil: null }); @@ -4881,10 +4880,9 @@ describe('answerFastAgentQuestion native OpenCode tools', () => { expect.any(String), '100.2', ); - expect(mocks.reconcileRetryNotices).not.toHaveBeenCalledWith( - expect.anything(), - 'next_turn_reconcile', - ); + expect( + mocks.reconcileRetryNotices.mock.calls.map((call) => call[1]), + ).not.toContain('next_turn_reconcile'); // The answer replaces the predecessor's visible notice in place and // retires the same canonical event instead of posting beside it. expect(replaceReply).toHaveBeenCalledWith( diff --git a/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts b/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts index b4f61c2d68..dec989e45c 100644 --- a/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts +++ b/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts @@ -17,6 +17,7 @@ import { isNotNull, isNull, lt, + ne, or, sessions, sql, @@ -112,6 +113,7 @@ async function reconcileInferenceRetryNotices( conversationId: string, requireExpiredLease: boolean, reason: FastAgentInterruptionReason, + options: { excludeEventId?: string } = {}, ): Promise { await database.execute( sql`select pg_advisory_xact_lock(hashtextextended(${`fast-agent-conversation:${conversationId}`}, 0))`, @@ -127,6 +129,31 @@ async function reconcileInferenceRetryNotices( } } + // An active notice whose turn still has a pending durable row is not + // orphaned: that turn is parked for a retry, waiting for the queue, or + // running elsewhere, and its resumed run edits the notice into the answer. + // Stamping it as interrupted here would post a false interruption and + // leave the eventual answer beside it. The caller's own row (a new turn + // that has not superseded the older one, such as a reaction) is excluded. + const [pendingTurn] = await database + .select({ id: fastAgentParentEvents.id }) + .from(fastAgentParentEvents) + .where( + and( + eq(fastAgentParentEvents.conversationId, conversationId), + eq(fastAgentParentEvents.admission, 'inline'), + isNull(fastAgentParentEvents.deliveredAt), + isNull(fastAgentParentEvents.discardedAt), + ...(options.excludeEventId + ? [ne(fastAgentParentEvents.id, options.excludeEventId)] + : []), + ), + ) + .limit(1); + if (pendingTurn) { + return 0; + } + // One set-based statement with no prior read: the terminal metadata is // derived from each row's current value under its row lock, so a cause an // interrupted owner commits concurrently (e.g. lock_lost) cannot be @@ -172,9 +199,14 @@ export async function reconcileFastAgentInferenceRetryNotices( FastAgentInterruptionReason, 'next_turn_reconcile' | 'turn_settled_reconcile' >, + options: { + /** The calling turn's own durable row, which must not count as a pending + * turn that owns the notices. */ + excludeEventId?: string; + } = {}, ): Promise { return db.transaction((tx) => - reconcileInferenceRetryNotices(tx, conversationId, false, reason), + reconcileInferenceRetryNotices(tx, conversationId, false, reason, options), ); } diff --git a/packages/cloud-agents/src/server/fast-agent/fast-agent-service.ts b/packages/cloud-agents/src/server/fast-agent/fast-agent-service.ts index dfced2e7f4..5009622b2f 100644 --- a/packages/cloud-agents/src/server/fast-agent/fast-agent-service.ts +++ b/packages/cloud-agents/src/server/fast-agent/fast-agent-service.ts @@ -2515,6 +2515,7 @@ export async function answerFastAgentQuestion({ await reconcileFastAgentInferenceRetryNotices( session.id, 'next_turn_reconcile', + durableAdmission ? { excludeEventId: durableAdmission.eventId } : {}, ).catch((error) => { console.warn( `[Fast Agent] Failed to reconcile interrupted inference retry notices: ${formatErrorForLog(error)}`, @@ -4758,6 +4759,7 @@ export async function answerFastAgentQuestion({ await reconcileFastAgentInferenceRetryNotices( canonicalConversationId, 'turn_settled_reconcile', + durableAdmission ? { excludeEventId: durableAdmission.eventId } : {}, ).catch((error) => { console.warn( `[Fast Agent] Failed to reconcile settled inference retry notices: ${formatErrorForLog(error)}`, diff --git a/packages/sdk/src/server/lib/fast-agent-human-follow-up.test.ts b/packages/sdk/src/server/lib/fast-agent-human-follow-up.test.ts index 1b137a96be..2c2dca2e98 100644 --- a/packages/sdk/src/server/lib/fast-agent-human-follow-up.test.ts +++ b/packages/sdk/src/server/lib/fast-agent-human-follow-up.test.ts @@ -133,6 +133,67 @@ describe('persistFastAgentInlineHumanTurn', () => { expect(mocks.updateWhere).toHaveBeenCalledOnce(); }); + it('does not let a reaction supersede a parked or interrupted turn', async () => { + mocks.findFirst.mockResolvedValue({ + id: 'row-1', + admission: 'inline', + deliveredAt: null, + discardedAt: null, + }); + + await expect( + persistFastAgentInlineHumanTurn({ + parent, + event: { + ...event, + input: { + type: 'reaction', + externalInput: { + type: 'reaction_added', + provider: 'slack', + reactions: [{ name: 'thumbsup' }], + reactor: { externalUserId: 'user-1' }, + message: { + workspaceId: 'team-1', + channelId: 'channel-1', + messageId: '100.1', + threadId: '100.1', + text: 'Earlier message', + }, + eventId: '100.3', + }, + }, + }, + }), + ).resolves.toEqual({ id: 'row-1', eventKey: 'stable-event-key' }); + // The reaction's own row is persisted, but the older pending inline row + // (a turn parked for a retry or waiting to resume) is left alone. + expect(mocks.insertOnConflict).toHaveBeenCalledOnce(); + expect(mocks.updateWhere).not.toHaveBeenCalled(); + }); + + it('does not let a platform event supersede a parked or interrupted turn', async () => { + mocks.findFirst.mockResolvedValue({ + id: 'row-1', + admission: 'inline', + deliveredAt: null, + discardedAt: null, + }); + + await expect( + persistFastAgentInlineHumanTurn({ + parent, + event: { + ...event, + turnSource: 'platform_event', + platformEventKind: 'delegated_task', + setupSession: true, + }, + }), + ).resolves.toEqual({ id: 'row-1', eventKey: 'stable-event-key' }); + expect(mocks.updateWhere).not.toHaveBeenCalled(); + }); + it('returns no durable handle when the same message already settled', async () => { mocks.findFirst.mockResolvedValue({ id: 'row-1', diff --git a/packages/sdk/src/server/lib/fast-agent-human-follow-up.ts b/packages/sdk/src/server/lib/fast-agent-human-follow-up.ts index d0a63af17e..a017588959 100644 --- a/packages/sdk/src/server/lib/fast-agent-human-follow-up.ts +++ b/packages/sdk/src/server/lib/fast-agent-human-follow-up.ts @@ -49,10 +49,27 @@ export type FastAgentHumanFollowUpAdmission = * row is persisted under a claim lease before any work starts, so the turn * survives the accepting process: if that process is interrupted before the * turn has posted its closeout, it releases the claim and the parent-event - * queue resumes the turn, telling it what the earlier attempt already did. The queue is not woken here; a live owner runs - * the turn itself. An older pending inline row for the same conversation is - * an interrupted turn this newer message supersedes. + * queue resumes the turn, telling it what the earlier attempt already did. + * The queue is not woken here; a live owner runs the turn itself. + * + * A typed human message supersedes an older pending inline row for the same + * conversation: that row is an interrupted or parked turn, and the new turn + * is told about the request it still owes. A reaction or a platform event + * admitted through this path does not supersede anything: neither answers + * the earlier request, so the earlier turn keeps its row and resumes once + * the conversation is idle again. */ +/** + * Only a typed human message stands in for the request an older pending turn + * still owes. A reaction (`input`) or a platform event admitted through the + * human path (`turnSource`) is a side conversation: discarding the older row + * for it would silently drop a question that was parked for a retry or + * waiting to resume. + */ +function supersedesPendingTurns(event: FastAgentHumanFollowUpEvent): boolean { + return !event.input && event.turnSource !== 'platform_event'; +} + export async function persistFastAgentInlineHumanTurn(params: { parent: FastAgentParent; event: FastAgentHumanFollowUpEvent; @@ -101,22 +118,24 @@ export async function persistFastAgentInlineHumanTurn(params: { .where(eq(fastAgentParentEvents.id, row.id)); } - await tx - .update(fastAgentParentEvents) - .set({ - discardedAt: new Date(), - lastError: 'Superseded by a newer human message.', - updatedAt: new Date(), - }) - .where( - and( - eq(fastAgentParentEvents.conversationId, params.parent.sessionId), - eq(fastAgentParentEvents.admission, 'inline'), - ne(fastAgentParentEvents.eventKey, eventKey), - isNull(fastAgentParentEvents.deliveredAt), - isNull(fastAgentParentEvents.discardedAt), - ), - ); + if (supersedesPendingTurns(params.event)) { + await tx + .update(fastAgentParentEvents) + .set({ + discardedAt: new Date(), + lastError: 'Superseded by a newer human message.', + updatedAt: new Date(), + }) + .where( + and( + eq(fastAgentParentEvents.conversationId, params.parent.sessionId), + eq(fastAgentParentEvents.admission, 'inline'), + ne(fastAgentParentEvents.eventKey, eventKey), + isNull(fastAgentParentEvents.deliveredAt), + isNull(fastAgentParentEvents.discardedAt), + ), + ); + } return { id: row.id, eventKey, ...(resumed ? { resumed: true } : {}) }; }); From 72a5b01cfa589f2a211f0092548aa6d543368dc1 Mon Sep 17 00:00:00 2001 From: Matt Rubens <2600+mrubens@users.noreply.github.com> Date: Fri, 4 Sep 2026 01:13:36 -0400 Subject: [PATCH 2/3] Keep the expired-lease sweep as the backstop for stalled hand-offs The pending-row guard now counts only a live claim or a scheduled retry on the expired-lease path, so a released or expired row whose queue wakeup never runs cannot leave a stale retry notice active forever. --- .../fast-agent-conversation-repository.test.ts | 6 ++++-- .../fast-agent-conversation-repository.ts | 13 +++++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts b/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts index b7c2c7f831..46ed7cbecc 100644 --- a/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts +++ b/packages/cloud-agents/src/server/fast-agent/__tests__/fast-agent-conversation-repository.test.ts @@ -1656,10 +1656,12 @@ describe('Fast conversation repository', () => { await reconcileExpiredFastAgentInferenceRetryNotices(); expect(await readNotice()).toMatchObject({ inferenceRetryActive: true }); - // Once the row is settled, the same notice is an orphan again. + // A hand-off whose claim was released and whose retry time has passed + // is only owned while its queue wakeup runs; the expired-lease sweep is + // the backstop for one that never does, so such a row does not block it. await db .update(fastAgentParentEvents) - .set({ deliveredAt: new Date() }) + .set({ retryAt: new Date(Date.now() - 1_000), claimedUntil: null }) .where(eq(fastAgentParentEvents.conversationId, session.id)); await reconcileExpiredFastAgentInferenceRetryNotices(); expect(await readNotice()).toMatchObject({ diff --git a/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts b/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts index dec989e45c..522c4c71ec 100644 --- a/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts +++ b/packages/cloud-agents/src/server/fast-agent/fast-agent-conversation-repository.ts @@ -135,6 +135,11 @@ async function reconcileInferenceRetryNotices( // Stamping it as interrupted here would post a false interruption and // leave the eventual answer beside it. The caller's own row (a new turn // that has not superseded the older one, such as a reaction) is excluded. + // + // The expired-lease sweep is the backstop for a hand-off whose queue + // wakeup never runs, so there only a live claim or a scheduled retry + // counts as owned; a released or expired row must not block it forever. + const now = new Date(); const [pendingTurn] = await database .select({ id: fastAgentParentEvents.id }) .from(fastAgentParentEvents) @@ -147,6 +152,14 @@ async function reconcileInferenceRetryNotices( ...(options.excludeEventId ? [ne(fastAgentParentEvents.id, options.excludeEventId)] : []), + ...(requireExpiredLease + ? [ + or( + gt(fastAgentParentEvents.claimedUntil, now), + gt(fastAgentParentEvents.retryAt, now), + ), + ] + : []), ), ) .limit(1); From a800c13f8046889ba1622bbb24852b625c10d683 Mon Sep 17 00:00:00 2001 From: Matt Rubens <2600+mrubens@users.noreply.github.com> Date: Fri, 4 Sep 2026 01:24:13 -0400 Subject: [PATCH 3/3] Re-run review