[Feat] Durable admission: interrupted Fast turns resume on the next process

.env.production.example

@@ -18,6 +18,9 @@ PREVIEW_PROXY_SUBDOMAIN_SUFFIX=preview
 # the remainder. Keep it under the platform's SIGTERM-to-SIGKILL grace
 # window; 0 aborts active turns immediately.
 # R_API_SHUTDOWN_DRAIN_MS=20000
+# Set to true to stop persisting human Fast turns for automatic resumption
+# after a restart (kill switch for durable admission).
+# R_FAST_DURABLE_ADMISSION_DISABLED=true
 
 # Replace local defaults before exposing a shared deployment.
 DATABASE_URL=postgres://postgres:password@postgres:5432/roomote_development

apps/api/src/handlers/discord/fast-agent.ts

@@ -27,8 +27,11 @@ import {
 } from '@roomote/communication';
 import {
   admitFastAgentHumanFollowUp,
+  persistFastAgentInlineHumanTurn,
   recordFastAgentConversationMessageBestEffort,
   resolveUserMcpServerConfigs,
+  wakeFastAgentParentEventNow,
+  type FastAgentDurableTurn,
 } from '@roomote/sdk/server';
 import { ALL_REPOSITORIES, type TaskInitiator } from '@roomote/types';
 
@@ -177,33 +180,56 @@ export async function processDiscordFastAgentMessage(
       userId: input.senderUserId,
       conversation,
     });
+    const humanFollowUpEvent = {
+      type: 'human_follow_up' as const,
+      eventId,
+      currentMessageId: anchorMessageId ?? eventId,
+      userId: input.senderUserId,
+      question: input.question,
+      senderDisplayName:
+        input.interaction?.interaction.member?.nick ??
+        input.sender.global_name ??
+        input.sender.username,
+      senderExternalId: input.sender.id,
+    };
+    let durableTurn: FastAgentDurableTurn | null = null;
     if (!releaseFastAgentLock) {
       const admission = await admitFastAgentHumanFollowUp({
         parent: { sessionId: session.id, conversation },
-        event: {
-          type: 'human_follow_up',
-          eventId,
-          currentMessageId: anchorMessageId ?? eventId,
-          userId: input.senderUserId,
-          question: input.question,
-          senderDisplayName:
-            input.interaction?.interaction.member?.nick ??
-            input.sender.global_name ??
-            input.sender.username,
-          senderExternalId: input.sender.id,
-        },
+        event: humanFollowUpEvent,
       });
       if (admission.kind !== 'turn') {
         input.onAccepted?.(admission.abort);
         return true;
       }
       releaseFastAgentLock = admission.turnLock;
+      durableTurn = admission.durable;
     }
     if (!releaseFastAgentLock) {
       input.onRejected?.();
       return false;
     }
     const activeTurnLock = releaseFastAgentLock;
+    // Durable admission: the turn is persisted under this process's claim
+    // before it runs, so an interruption hands it to the queue.
+    durableTurn ??= await persistFastAgentInlineHumanTurn({
+      parent: { sessionId: session.id, conversation },
+      event: humanFollowUpEvent,
+    }).catch((error) => {
+      console.error(
+        `[DiscordFastAgent] Failed to persist Fast turn admission: ${error instanceof Error ? error.message : String(error)}`,
+      );
+      return null;
+    });
+    const durableTurnForResume = durableTurn;
+    if (durableTurnForResume) {
+      activeTurnLock.durableRowId = durableTurnForResume.id;
+      activeTurnLock.durableResume = () =>
+        wakeFastAgentParentEventNow({
+          conversationId: session.id,
+          eventKey: durableTurnForResume.eventKey,
+        });
+    }
     const footerContext = await resolveFastSessionReplyFooterContext({
       sessionId: session.id,
     });
@@ -279,6 +305,9 @@ export async function processDiscordFastAgentMessage(
       conversation,
       currentMessageId: anchorMessageId ?? input.interaction?.interaction.id,
       signal: activeTurnLock.signal,
+      ...(durableTurnForResume
+        ? { durableAdmission: { eventId: durableTurnForResume.id } }
+        : {}),
       senderDisplayName:
         input.interaction?.interaction.member?.nick ??
         input.sender.global_name ??
@@ -293,6 +322,15 @@ export async function processDiscordFastAgentMessage(
             entry.user !== input.sender.id,
         ),
       adapter: {
+        ...(durableTurnForResume
+          ? {
+              requestDurableResume: () =>
+                wakeFastAgentParentEventNow({
+                  conversationId: session.id,
+                  eventKey: durableTurnForResume.eventKey,
+                }),
+            }
+          : {}),
         resolveMcpServerConfigs: () =>
           resolveUserMcpServerConfigs({
             userId: input.senderUserId,

apps/api/src/handlers/slack/events/fast-agent.ts

@@ -24,6 +24,9 @@ import {
 import { appendAttachmentTextsToPromptText } from '@roomote/cloud-agents';
 import {
   admitFastAgentHumanFollowUp,
+  persistFastAgentInlineHumanTurn,
+  wakeFastAgentParentEventNow,
+  type FastAgentDurableTurn,
   recordFastAgentConversationMessageBestEffort,
   resolveUserMcpServerConfigs,
 } from '@roomote/sdk/server';
@@ -220,34 +223,56 @@ export async function processFastAgentMessage(params: {
       conversation.surface !== incomingConversation.surface ||
       conversation.workspaceId !== incomingConversation.workspaceId ||
       conversation.conversationId !== incomingConversation.conversationId;
+    const humanFollowUpEvent = {
+      type: 'human_follow_up' as const,
+      eventId: event.ts,
+      currentMessageId: event.ts,
+      userId,
+      question,
+      ...(attachments.images.length ? { images: attachments.images } : {}),
+      ...(currentMessage?.username
+        ? { senderDisplayName: currentMessage.username }
+        : {}),
+      ...(event.user ? { senderExternalId: event.user } : {}),
+    };
+    let durableTurn: FastAgentDurableTurn | null = null;
     if (needsCanonicalAdmission) {
       const admission = await admitFastAgentHumanFollowUp({
         parent: { sessionId: session.id, conversation },
-        event: {
-          type: 'human_follow_up',
-          eventId: event.ts,
-          currentMessageId: event.ts,
-          userId,
-          question,
-          ...(attachments.images.length ? { images: attachments.images } : {}),
-          ...(currentMessage?.username
-            ? { senderDisplayName: currentMessage.username }
-            : {}),
-          ...(event.user ? { senderExternalId: event.user } : {}),
-        },
+        event: humanFollowUpEvent,
       });
       if (admission.kind !== 'turn') {
         params.onAccepted?.(admission.abort);
         return;
       }
       releaseCanonicalFastAgentLock = admission.turnLock;
+      durableTurn = admission.durable;
     }
     const activeTurnLock =
       releaseCanonicalFastAgentLock ?? releaseFastAgentLock;
     if (!activeTurnLock) {
       params.onRejected?.();
       return;
     }
+    // Durable admission: the turn is persisted under this process's claim
+    // before it runs, so an interruption hands it to the queue.
+    durableTurn ??= await persistFastAgentInlineHumanTurn({
+      parent: { sessionId: session.id, conversation },
+      event: humanFollowUpEvent,
+    }).catch((error) => {
+      console.error(
+        `[SlackWebhook] Failed to persist Fast turn admission: ${error instanceof Error ? error.message : String(error)}`,
+      );
+      return null;
+    });
+    if (durableTurn) {
+      activeTurnLock.durableRowId = durableTurn.id;
+      activeTurnLock.durableResume = () =>
+        wakeFastAgentParentEventNow({
+          conversationId: session.id,
+          eventKey: durableTurn.eventKey,
+        });
+    }
     params.onAccepted?.(() =>
       activeTurnLock.abort(
         new Error('Fast suggestion launch settlement failed.'),
@@ -264,6 +289,7 @@ export async function processFastAgentMessage(params: {
       conversation,
       currentMessageId: event.ts,
       signal: activeTurnLock.signal,
+      ...(durableTurn ? { durableAdmission: { eventId: durableTurn.id } } : {}),
       senderExternalId: event.user,
       senderDisplayName:
         currentMessage?.user === event.user
@@ -277,6 +303,15 @@ export async function processFastAgentMessage(params: {
         !directedAtRoomote,
       ...(roomoteSlackUserId ? { slackRoomoteUserId: roomoteSlackUserId } : {}),
       adapter: {
+        ...(durableTurn
+          ? {
+              requestDurableResume: () =>
+                wakeFastAgentParentEventNow({
+                  conversationId: session.id,
+                  eventKey: durableTurn.eventKey,
+                }),
+            }
+          : {}),
         activity: createFastAgentSlackSessionActivity({
           slack,
           workspaceId: teamId,

apps/web/src/trpc/commands/fast-sessions/index.ts

@@ -12,7 +12,9 @@ import {
 } from '@roomote/cloud-agents/server';
 import {
   buildFastAgentSurfaceReplyDelivery,
+  persistFastAgentInlineHumanTurn,
   resolveUserMcpServerConfigs,
+  wakeFastAgentParentEventNow,
   type FastAgentSurfaceReplyDelivery,
 } from '@roomote/sdk/server';
 import {
@@ -120,6 +122,9 @@ type WebFastAgentTurnInput = {
   /** Deterministic turn ID override. Canonical event IDs derive from it, so a
    * fixed value lets a turn be claimed idempotently across retries. */
   currentMessageId?: string;
+  /** Fast conversation id for durable admission of a human turn. Platform
+   * turns (kickoffs, artifact builds) omit it and stay non-replayable. */
+  durableSessionId?: string;
   /** Skip the turn if this exact canonical event row already exists when the
    * turn acquires its lock. This is the atomic claim for the setup kickoff:
    * concurrent submits can both pass the pre-schedule check, but the first
@@ -169,6 +174,7 @@ async function runWebFastAgentTurn({
   platformEventKind,
   currentMessageId,
   skipIfEventExists,
+  durableSessionId,
 }: WebFastAgentTurnInput): Promise<void> {
   const conversation = delivery.conversation;
   const release = await acquireFastAgentTurnLock({ conversation });
@@ -222,15 +228,47 @@ async function runWebFastAgentTurn({
       }
     }
 
+    const turnMessageId = currentMessageId ?? `web-${randomUUID()}`;
+    // Durable admission: a human web turn is persisted under this process's
+    // claim before it runs, so an interruption hands it to the queue.
+    const durableTurn =
+      durableSessionId && !platformEventKind
+        ? await persistFastAgentInlineHumanTurn({
+            parent: { sessionId: durableSessionId, conversation },
+            event: {
+              type: 'human_follow_up',
+              eventId: turnMessageId,
+              currentMessageId: turnMessageId,
+              userId,
+              question,
+              ...(images?.length ? { images } : {}),
+              ...(senderDisplayName ? { senderDisplayName } : {}),
+            },
+          }).catch((error) => {
+            console.error(
+              `[Fast Web] Failed to persist turn admission: ${formatErrorForLog(error)}`,
+            );
+            return null;
+          })
+        : null;
+    if (durableTurn && durableSessionId) {
+      release.durableRowId = durableTurn.id;
+      release.durableResume = () =>
+        wakeFastAgentParentEventNow({
+          conversationId: durableSessionId,
+          eventKey: durableTurn.eventKey,
+        });
+    }
     await answerFastAgentQuestion({
       question,
       images,
       attachmentTexts,
       userId,
       apiBaseUrl,
       conversation,
-      currentMessageId: currentMessageId ?? `web-${randomUUID()}`,
+      currentMessageId: turnMessageId,
       signal: release.signal,
+      ...(durableTurn ? { durableAdmission: { eventId: durableTurn.id } } : {}),
       model,
       reasoningEffort,
       senderDisplayName,
@@ -248,6 +286,15 @@ async function runWebFastAgentTurn({
             apiBaseUrl,
             includeRoomoteMemberTools: true,
           }),
+        ...(durableTurn && durableSessionId
+          ? {
+              requestDurableResume: () =>
+                wakeFastAgentParentEventNow({
+                  conversationId: durableSessionId,
+                  eventKey: durableTurn.eventKey,
+                }),
+            }
+          : {}),
         ...delivery.adapter,
       },
     });
@@ -463,6 +510,7 @@ export async function startFastSessionCommand(
       attachmentTexts: input.attachmentTexts,
       model: settings.model,
       reasoningEffort: settings.reasoningEffort,
+      durableSessionId: session.id,
       ...(kickoffTurnId && kickoffPromptEventId
         ? {
             currentMessageId: kickoffTurnId,
@@ -689,6 +737,7 @@ export async function replyToFastSessionCommand(
     model: settings.model,
     reasoningEffort: settings.reasoningEffort,
     ...(senderDisplayName ? { senderDisplayName } : {}),
+    durableSessionId: session.id,
   });
 
   return { success: true };