[Fix] Reactions and platform events no longer supersede parked Fast turns

.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.

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<number> {
   await database.execute(
     sql`select pg_advisory_xact_lock(hashtextextended(${`fast-agent-conversation:${conversationId}`}, 0))`,
@@ -127,6 +129,44 @@ 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.
+  //
+  // 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)
+    .where(
+      and(
+        eq(fastAgentParentEvents.conversationId, conversationId),
+        eq(fastAgentParentEvents.admission, 'inline'),
+        isNull(fastAgentParentEvents.deliveredAt),
+        isNull(fastAgentParentEvents.discardedAt),
+        ...(options.excludeEventId
+          ? [ne(fastAgentParentEvents.id, options.excludeEventId)]
+          : []),
+        ...(requireExpiredLease
+          ? [
+              or(
+                gt(fastAgentParentEvents.claimedUntil, now),
+                gt(fastAgentParentEvents.retryAt, now),
+              ),
+            ]
+          : []),
+      ),
+    )
+    .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 +212,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<number> {
   return db.transaction((tx) =>
-    reconcileInferenceRetryNotices(tx, conversationId, false, reason),
+    reconcileInferenceRetryNotices(tx, conversationId, false, reason, options),
   );
 }
 

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)}`,

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 } : {}) };
   });