[Fix] Fast turns killed mid-flight are closed out instead of vanishing

apps/api/src/graceful-shutdown.ts

@@ -3,6 +3,7 @@ import {
   abortActiveFastAgentTurns,
   beginFastAgentTurnDrain,
   FastAgentProcessShutdownError,
+  markActiveFastAgentTurnsShutdown,
   waitForActiveFastAgentTurnsToSettle,
 } from '@roomote/cloud-agents/server';
 
@@ -12,6 +13,15 @@ import {
 // typical 30s SIGTERM-to-SIGKILL grace window. R_API_SHUTDOWN_DRAIN_MS
 // overrides it; 0 restores the previous abort-immediately behavior.
 const DEFAULT_API_SHUTDOWN_DRAIN_MS = 20_000;
+// How long a still-pending shutdown stamp may hold the exit after the drain
+// and the abort have already given it time.
+const STAMP_SETTLE_GRACE_MS = 2_000;
+
+function delay(ms: number): Promise<void> {
+  return new Promise((resolve) => {
+    setTimeout(resolve, ms).unref?.();
+  });
+}
 
 export function resolveApiShutdownDrainMs(
   env: NodeJS.ProcessEnv = process.env,
@@ -27,6 +37,7 @@ export function resolveApiShutdownDrainMs(
 type ApiShutdownOptions = {
   abortTurns?: typeof abortActiveFastAgentTurns;
   beginDrain?: typeof beginFastAgentTurnDrain;
+  markTurnsShutdown?: typeof markActiveFastAgentTurnsShutdown;
   waitForTurns?: typeof waitForActiveFastAgentTurnsToSettle;
   drainMs?: number;
   exitProcess?: (code?: number) => never;
@@ -41,6 +52,7 @@ export async function gracefullyShutdownApi(
   {
     abortTurns = abortActiveFastAgentTurns,
     beginDrain = beginFastAgentTurnDrain,
+    markTurnsShutdown = markActiveFastAgentTurnsShutdown,
     waitForTurns = waitForActiveFastAgentTurnsToSettle,
     drainMs = resolveApiShutdownDrainMs(),
     exitProcess = process.exit,
@@ -54,6 +66,13 @@ export async function gracefullyShutdownApi(
   // give in-flight turns a bounded window to finish on their own. Only the
   // stragglers still active at the deadline are aborted.
   beginDrain(reason);
+  // The platform may SIGKILL right after SIGTERM (Railway's default grace is
+  // zero), so the evidence that these turns were cut short is written now,
+  // alongside the drain rather than after it. A turn that finishes during
+  // the drain settles its row and the stamp is moot.
+  const markPromise = markTurnsShutdown({
+    lockTtlSeconds: Math.ceil(drainMs / 1000) + 60,
+  }).catch(() => 0);
   const closePromise = new Promise<Error | null>((resolve) => {
     server.close((error) => resolve(error ?? null));
   });
@@ -66,6 +85,11 @@ export async function gracefullyShutdownApi(
   const abortPromise = abortTurns(reason);
   const closeError = await closePromise;
   await abortPromise;
+  // The stamps have had the drain and the abort to land. Give a slow one a
+  // last brief moment, but never let it hold up the exit: losing a stamp
+  // costs latency (the reconciler falls back to the lease), holding the
+  // process costs the platform's patience.
+  await Promise.race([markPromise, delay(STAMP_SETTLE_GRACE_MS)]);
   if (closeError) {
     logError('[api] Graceful shutdown failed', closeError);
   }

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

@@ -11,6 +11,7 @@ import {
   type DiscordCommunicationProvider,
 } from '@roomote/communication/discord-provider';
 import {
+  bindFastAgentTurnLockDurableRow,
   getOrCreateFastAgentSession,
   acquireFastAgentTurnLock,
   answerFastAgentQuestion,
@@ -241,12 +242,14 @@ export async function processDiscordFastAgentMessage(
     });
     const durableTurnForResume = durableTurn;
     if (durableTurnForResume) {
-      activeTurnLock.durableRowId = durableTurnForResume.id;
-      activeTurnLock.durableResume = () =>
-        wakeFastAgentParentEventNow({
-          conversationId: session.id,
-          eventKey: durableTurnForResume.eventKey,
-        });
+      await bindFastAgentTurnLockDurableRow(activeTurnLock, {
+        rowId: durableTurnForResume.id,
+        resume: () =>
+          wakeFastAgentParentEventNow({
+            conversationId: session.id,
+            eventKey: durableTurnForResume.eventKey,
+          }),
+      });
     }
     const footerContext = await resolveFastSessionReplyFooterContext({
       sessionId: session.id,

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

@@ -1,4 +1,5 @@
 import {
+  bindFastAgentTurnLockDurableRow,
   getOrCreateFastAgentSession,
   acquireFastAgentTurnLock,
   answerFastAgentQuestion,
@@ -289,12 +290,14 @@ export async function processFastAgentMessage(params: {
       return null;
     });
     if (durableTurn) {
-      activeTurnLock.durableRowId = durableTurn.id;
-      activeTurnLock.durableResume = () =>
-        wakeFastAgentParentEventNow({
-          conversationId: session.id,
-          eventKey: durableTurn.eventKey,
-        });
+      await bindFastAgentTurnLockDurableRow(activeTurnLock, {
+        rowId: durableTurn.id,
+        resume: () =>
+          wakeFastAgentParentEventNow({
+            conversationId: session.id,
+            eventKey: durableTurn.eventKey,
+          }),
+      });
     }
     params.onAccepted?.(() =>
       activeTurnLock.abort(

apps/bullmq/src/index.ts

@@ -1,6 +1,13 @@
 import { serve } from '@hono/node-server';
 import { serveStatic } from '@hono/node-server/serve-static';
 import { Hono } from 'hono';
+import {
+  abortActiveFastAgentTurns,
+  beginFastAgentTurnDrain,
+  FastAgentProcessShutdownError,
+  markActiveFastAgentTurnsShutdown,
+  waitForActiveFastAgentTurnsToSettle,
+} from '@roomote/cloud-agents/server';
 import { showRoutes } from 'hono/dev';
 import { createBullBoard } from '@bull-board/api';
 import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
@@ -373,9 +380,47 @@ app.route('/admin/queues', serverAdapter.registerPlugin());
 
 app.get('/', (c) => c.redirect('/admin/queues'));
 
-async function gracefulShutdown() {
+// Resumed Fast turns run in this process, so a restart gives them the same
+// bounded window the API gives its own turns. Matches the API's default;
+// both must stay inside the platform's SIGTERM-to-SIGKILL grace.
+const FAST_TURN_SHUTDOWN_DRAIN_MS = 20_000;
+
+async function gracefulShutdown(signal: NodeJS.Signals = 'SIGTERM') {
   console.log('[Shutdown] Starting graceful shutdown...');
 
+  // Stop admitting turns, then stamp the active ones before anything that
+  // takes time: a kill right after the signal still leaves evidence for the
+  // dead-turn reconciler, and the shortened lock keeps a dead turn from
+  // sitting behind its lock's full TTL. Turns that finish inside the drain
+  // settle and release their own locks, so the stamp is moot for them.
+  const shutdownReason = new FastAgentProcessShutdownError(signal);
+  beginFastAgentTurnDrain(shutdownReason);
+  // The stamp runs alongside the drain, not before it: one DB update per
+  // active turn must never extend the shutdown budget.
+  const stampPromise = markActiveFastAgentTurnsShutdown({
+    lockTtlSeconds: Math.ceil(FAST_TURN_SHUTDOWN_DRAIN_MS / 1000) + 60,
+  }).catch(() => undefined);
+  const remainingTurns = await waitForActiveFastAgentTurnsToSettle(
+    FAST_TURN_SHUTDOWN_DRAIN_MS,
+  ).catch(() => 0);
+  if (remainingTurns > 0) {
+    console.warn(
+      `[Shutdown] Aborting ${remainingTurns} Fast turn(s) still active after the ${FAST_TURN_SHUTDOWN_DRAIN_MS}ms drain.`,
+    );
+  }
+  // Aborting releases each turn's durable claim and wakes the queue, so an
+  // interrupted resumed turn goes back to recovery instead of waiting out
+  // its claim lease.
+  await abortActiveFastAgentTurns(shutdownReason).catch(() => undefined);
+  // The stamps have had the drain and the abort to land; a slow one gets a
+  // last brief moment and then stops holding up the shutdown.
+  await Promise.race([
+    stampPromise,
+    new Promise((resolve) => {
+      setTimeout(resolve, 2_000).unref?.();
+    }),
+  ]);
+
   try {
     await schedulerWorker.close();
     await schedulerQueueEvents.close();

apps/bullmq/src/scheduled-jobs/sessions-reconcile.ts

@@ -1,4 +1,8 @@
-import { reconcileExpiredFastAgentInferenceRetryNotices } from '@roomote/cloud-agents/server';
+import {
+  isFastAgentTurnLockHeld,
+  reconcileExpiredFastAgentInferenceRetryNotices,
+  reconcileFastAgentDeadTurns,
+} from '@roomote/cloud-agents/server';
 import {
   and,
   db,
@@ -208,6 +212,11 @@ async function reconcileRecentSessions(watermark: Date | null): Promise<void> {
   let orphanFailures = 0;
   const reconciledRetryNotices =
     await reconcileExpiredFastAgentInferenceRetryNotices(BATCH_SIZE);
+  // Turns whose owner was killed or crashed without closing out: give them
+  // the honest restart closeout and release the responding lease.
+  const reconciledDeadTurns = await reconcileFastAgentDeadTurns(BATCH_SIZE, {
+    isTurnLive: isFastAgentTurnLockHeld,
+  });
 
   // Fast conversations without a session row (e.g. created before this
   // release finished its backfill) are adopted here so the unified list
@@ -356,6 +365,7 @@ async function reconcileRecentSessions(watermark: Date | null): Promise<void> {
     refreshedSessions: recent.length,
     healedExpiredLeases: expiredLeases.length,
     reconciledRetryNotices,
+    reconciledDeadTurns,
   });
 }