[Fix] Fast automation Run now requests time out

apps/api/src/handlers/custom-automations/index.ts

@@ -16,6 +16,7 @@ import {
   users,
 } from '@roomote/db/server';
 import {
+  getCustomAutomationRunStatus,
   listConnectedCommunicationProviders,
   resolveCustomAutomationSchedule,
   runCustomAutomationNow,
@@ -540,5 +541,22 @@ customAutomationsRouter.delete('/:id', async (c) => {
 
 customAutomationsRouter.post('/:id/run', async (c) => {
   const result = await runCustomAutomationNow(c.req.param('id'));
-  return c.json(result, result.outcome === 'failed' ? 400 : 200);
+  return c.json(
+    result,
+    result.outcome === 'accepted'
+      ? 202
+      : result.outcome === 'failed'
+        ? 400
+        : 200,
+  );
+});
+
+customAutomationsRouter.get('/:id/runs/:invocationId', async (c) => {
+  const result = await getCustomAutomationRunStatus({
+    automationId: c.req.param('id'),
+    invocationId: c.req.param('invocationId'),
+  });
+  return result
+    ? c.json(result)
+    : c.json({ error: 'Custom automation run was not found.' }, 404);
 });

apps/bullmq/src/custom-automation-run-queue.ts

@@ -0,0 +1,67 @@
+import { Queue, QueueEvents, Worker } from 'bullmq';
+
+import {
+  CUSTOM_AUTOMATION_RUN_JOB_NAME,
+  CUSTOM_AUTOMATION_RUN_QUEUE_NAME,
+  customAutomationRunJobSchema,
+  runClaimedFastCustomAutomation,
+  type CustomAutomationRunJob,
+} from '@roomote/sdk/server';
+import { db, recordCustomAutomationRunOutcome } from '@roomote/db/server';
+
+import { getRedis } from './redis';
+
+export function startCustomAutomationRunQueue() {
+  const connection = getRedis();
+  const queue = new Queue<CustomAutomationRunJob, void, string>(
+    CUSTOM_AUTOMATION_RUN_QUEUE_NAME,
+    { connection },
+  );
+  const worker = new Worker<CustomAutomationRunJob, void, string>(
+    CUSTOM_AUTOMATION_RUN_QUEUE_NAME,
+    async (job) => {
+      if (job.name !== CUSTOM_AUTOMATION_RUN_JOB_NAME) {
+        throw new Error(`Unknown custom automation job: ${job.name}`);
+      }
+      await runClaimedFastCustomAutomation(
+        customAutomationRunJobSchema.parse(job.data),
+      );
+    },
+    {
+      connection,
+      concurrency: 3,
+      autorun: true,
+      // A stalled Fast turn may already have posted externally; never replay it.
+      maxStalledCount: 0,
+    },
+  );
+  const queueEvents = new QueueEvents(CUSTOM_AUTOMATION_RUN_QUEUE_NAME, {
+    connection,
+  });
+
+  worker.on('failed', (job, error) => {
+    const parsed = customAutomationRunJobSchema.safeParse(job?.data);
+    if (parsed.success) {
+      void recordCustomAutomationRunOutcome(db, {
+        id: parsed.data.automationId,
+        status: 'failed',
+        error: error.message,
+        launchClaimedAt: new Date(parsed.data.launchClaimedAt),
+      }).catch((finalizeError) =>
+        console.error(
+          '[CustomAutomationRunQueue] failed to finalize invocation:',
+          finalizeError,
+        ),
+      );
+    }
+    console.error(
+      `[CustomAutomationRunQueue] job ${job?.id} failed:`,
+      error.message,
+    );
+  });
+  worker.on('error', (error) =>
+    console.error('[CustomAutomationRunQueue] worker error:', error),
+  );
+
+  return { queue, worker, queueEvents };
+}

apps/bullmq/src/index.ts

@@ -48,6 +48,7 @@ import { startSlackPrInactivityQueue } from './slack-pr-inactivity-queue';
 import { startPrReviewNotificationQueue } from './pr-review-notification-queue';
 import { startActivePrReviewFollowUpQueue } from './active-pr-review-follow-up-queue';
 import { startPullRequestMergeabilityCheckQueue } from './pull-request-mergeability-check-queue';
+import { startCustomAutomationRunQueue } from './custom-automation-run-queue';
 import { startTaskSleepQueue } from './task-sleep-queue';
 import { startAutomationRecommendationsQueue } from './automation-recommendations-queue';
 
@@ -187,6 +188,11 @@ const {
   worker: pullRequestMergeabilityCheckWorker,
   queueEvents: pullRequestMergeabilityCheckQueueEvents,
 } = startPullRequestMergeabilityCheckQueue();
+const {
+  queue: customAutomationRunQueue,
+  worker: customAutomationRunWorker,
+  queueEvents: customAutomationRunQueueEvents,
+} = startCustomAutomationRunQueue();
 
 const serverAdapter = new HonoAdapter(serveStatic);
 
@@ -226,6 +232,7 @@ createBullBoard({
     new BullMQAdapter(pullRequestMergeabilityCheckQueue, {
       readOnlyMode: false,
     }),
+    new BullMQAdapter(customAutomationRunQueue, { readOnlyMode: false }),
   ],
   serverAdapter,
 });
@@ -399,6 +406,9 @@ async function gracefulShutdown() {
     await pullRequestMergeabilityCheckWorker.close();
     await pullRequestMergeabilityCheckQueueEvents.close();
     await pullRequestMergeabilityCheckQueue.close();
+    await customAutomationRunWorker.close();
+    await customAutomationRunQueueEvents.close();
+    await customAutomationRunQueue.close();
     await discordGatewaySupervisor.stop();
     await closeRedis();
   } catch (error) {

apps/web/src/components/settings/automations/CustomAutomationsSection.tsx

@@ -139,6 +139,9 @@ function CustomAutomationRunButton({
     ...trpc.automations.triggerCustomAutomation.mutationOptions({
       onSuccess: (result) => {
         switch (result.outcome) {
+          case 'accepted':
+            toast.success(`Running ${automation.name} now`);
+            break;
           case 'launched':
             toast.success(`Running ${automation.name} now`, {
               action: {

apps/web/src/trpc/commands/automations/custom-automations.ts

@@ -18,7 +18,7 @@ import {
   resolveDeploymentTimeZone,
   runCustomAutomationNow,
   validateCronExpression,
-  type AutomationRunNowResult,
+  type CustomAutomationRunNowResult,
 } from '@roomote/sdk/server';
 import {
   ALL_REPOSITORIES,
@@ -346,7 +346,7 @@ export async function deleteCustomAutomationCommand(
 export async function triggerCustomAutomationCommand(
   auth: UserAuthSuccess,
   input: { id: string },
-): Promise<AutomationRunNowResult> {
+): Promise<CustomAutomationRunNowResult> {
   assertAdmin(auth);
   return runCustomAutomationNow(input.id);
 }

apps/worker/src/mcp/roomote-mcp-server/api-client.ts

@@ -57,8 +57,14 @@ export async function fetchWithTimeout(
     return await fetch(url, { ...options, signal });
   } catch (error) {
     if (timeoutSignal.aborted) {
+      const method = options.method?.toUpperCase() ?? 'GET';
+      const retrySafe = method === 'GET' || method === 'HEAD';
       throw new Error(
-        `${context.label}: no response from the Roomote API within ${timeoutMs}ms; the request was aborted and is safe to retry.`,
+        `${context.label}: no response from the Roomote API within ${timeoutMs}ms; the request was aborted${
+          retrySafe
+            ? ' and is safe to retry.'
+            : ', but the operation may still complete; check its status before retrying.'
+        }`,
       );
     }