From ef52b10eccd3263fb59b8a5cd38d7dc13aea7a81 Mon Sep 17 00:00:00 2001 From: "@mrubens" <2600+mrubens@users.noreply.github.com> Date: Tue, 25 Aug 2026 18:24:15 +0000 Subject: [PATCH 1/2] fix: make Fast automation run now asynchronous --- .../custom-automations-routes.test.ts | 46 +++++++ .../src/handlers/custom-automations/index.ts | 20 ++- .../src/custom-automation-run-queue.test.ts | 82 +++++++++++++ .../bullmq/src/custom-automation-run-queue.ts | 67 ++++++++++ apps/bullmq/src/index.ts | 10 ++ .../automations/CustomAutomationsSection.tsx | 3 + .../automations/custom-automations.ts | 4 +- .../__tests__/api-client.test.ts | 17 ++- .../__tests__/tasks-api-client.test.ts | 2 +- .../src/mcp/roomote-mcp-server/api-client.ts | 8 +- .../lib/__tests__/custom-automations.test.ts | 44 +++++++ packages/db/src/lib/custom-automations.ts | 24 +++- .../custom-automation-run-queue.test.ts | 67 ++++++++++ .../__tests__/custom-automations.test.ts | 103 ++++++++++++++++ .../custom-automation-run-queue.ts | 80 ++++++++++++ .../server/automations/custom-automations.ts | 114 ++++++++++++++++-- packages/sdk/src/server/automations/index.ts | 11 ++ packages/sdk/src/server/automations/types.ts | 4 + .../manage-custom-automations-tool.test.ts | 17 +++ .../src/manage-custom-automations-tool.ts | 29 ++++- 20 files changed, 730 insertions(+), 22 deletions(-) create mode 100644 apps/bullmq/src/custom-automation-run-queue.test.ts create mode 100644 apps/bullmq/src/custom-automation-run-queue.ts create mode 100644 packages/sdk/src/server/automations/__tests__/custom-automation-run-queue.test.ts create mode 100644 packages/sdk/src/server/automations/custom-automation-run-queue.ts diff --git a/apps/api/src/handlers/custom-automations/__tests__/custom-automations-routes.test.ts b/apps/api/src/handlers/custom-automations/__tests__/custom-automations-routes.test.ts index 2680752b6..8588e329d 100644 --- a/apps/api/src/handlers/custom-automations/__tests__/custom-automations-routes.test.ts +++ b/apps/api/src/handlers/custom-automations/__tests__/custom-automations-routes.test.ts @@ -32,6 +32,7 @@ const { mockListConnectedCommunicationProviders, mockResolveCustomAutomationSchedule, mockRunCustomAutomationNow, + mockGetCustomAutomationRunStatus, mockCaptureActivationCustomAutomationChanged, } = vi.hoisted(() => ({ mockUsersFindFirst: vi.fn(), @@ -45,6 +46,7 @@ const { mockListConnectedCommunicationProviders: vi.fn(), mockResolveCustomAutomationSchedule: vi.fn(), mockRunCustomAutomationNow: vi.fn(), + mockGetCustomAutomationRunStatus: vi.fn(), mockCaptureActivationCustomAutomationChanged: vi.fn(), })); @@ -63,6 +65,7 @@ vi.mock('@roomote/db/server', () => ({ })); vi.mock('@roomote/sdk/server', () => ({ + getCustomAutomationRunStatus: mockGetCustomAutomationRunStatus, listConnectedCommunicationProviders: mockListConnectedCommunicationProviders, resolveCustomAutomationSchedule: mockResolveCustomAutomationSchedule, runCustomAutomationNow: mockRunCustomAutomationNow, @@ -768,6 +771,49 @@ describe('custom-automations MCP routes', () => { }); }); + describe('manual runs', () => { + it('returns 202 with the accepted Fast invocation identifier', async () => { + const { app } = createApp(); + mockRunCustomAutomationNow.mockResolvedValue({ + outcome: 'accepted', + invocationId: 'invocation-1', + }); + + const res = await app.request('/custom-automations/automation-1/run', { + method: 'POST', + }); + + expect(res.status).toBe(202); + await expect(res.json()).resolves.toEqual({ + outcome: 'accepted', + invocationId: 'invocation-1', + }); + }); + + it.each([ + { status: 'succeeded' as const }, + { status: 'failed' as const, error: 'provider unavailable' }, + ])('returns $status terminal status', async (terminal) => { + const { app } = createApp(); + mockGetCustomAutomationRunStatus.mockResolvedValue({ + automationId: 'automation-1', + invocationId: 'invocation-1', + ...terminal, + }); + + const res = await app.request( + '/custom-automations/automation-1/runs/invocation-1', + ); + + expect(res.status).toBe(200); + await expect(res.json()).resolves.toEqual({ + automationId: 'automation-1', + invocationId: 'invocation-1', + ...terminal, + }); + }); + }); + describe('POST /resolve-schedule', () => { it('returns 400 with the message for a known schedule validation failure', async () => { const { app } = createApp(); diff --git a/apps/api/src/handlers/custom-automations/index.ts b/apps/api/src/handlers/custom-automations/index.ts index df2a929e8..b4f0f8511 100644 --- a/apps/api/src/handlers/custom-automations/index.ts +++ b/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); }); diff --git a/apps/bullmq/src/custom-automation-run-queue.test.ts b/apps/bullmq/src/custom-automation-run-queue.test.ts new file mode 100644 index 000000000..9d0b67626 --- /dev/null +++ b/apps/bullmq/src/custom-automation-run-queue.test.ts @@ -0,0 +1,82 @@ +const mocks = vi.hoisted(() => ({ + workerOptions: null as Record | null, + on: vi.fn(), + recordOutcome: vi.fn(), +})); + +vi.mock('bullmq', () => ({ + Queue: class Queue {}, + QueueEvents: class QueueEvents {}, + Worker: class Worker { + constructor( + _name: string, + _handler: unknown, + options: Record, + ) { + mocks.workerOptions = options; + } + on = mocks.on; + }, +})); + +vi.mock('@roomote/sdk/server', () => ({ + CUSTOM_AUTOMATION_RUN_JOB_NAME: 'run-custom-automation', + CUSTOM_AUTOMATION_RUN_QUEUE_NAME: 'custom-automation-runs', + customAutomationRunJobSchema: { + parse: (value: unknown) => value, + safeParse: (value: unknown) => ({ success: true, data: value }), + }, + runClaimedFastCustomAutomation: vi.fn(), +})); + +vi.mock('@roomote/db/server', () => ({ + db: {}, + recordCustomAutomationRunOutcome: mocks.recordOutcome, +})); + +vi.mock('./redis', () => ({ getRedis: vi.fn(() => ({})) })); + +import { startCustomAutomationRunQueue } from './custom-automation-run-queue'; + +describe('custom automation run worker', () => { + beforeEach(() => { + vi.clearAllMocks(); + mocks.workerOptions = null; + mocks.recordOutcome.mockResolvedValue(true); + }); + + it('disables stalled-job replay and clears the fenced claim on failure', async () => { + startCustomAutomationRunQueue(); + + expect(mocks.workerOptions).toMatchObject({ maxStalledCount: 0 }); + const failedHandler = mocks.on.mock.calls.find( + ([event]) => event === 'failed', + )?.[1] as + | ((job: { id: string; data: unknown }, error: Error) => void) + | undefined; + expect(failedHandler).toBeDefined(); + + const launchClaimedAt = '2026-08-25T17:41:38.469Z'; + failedHandler?.( + { + id: 'invocation-1', + data: { + automationId: '11111111-1111-4111-8111-111111111111', + launchClaimedAt, + }, + }, + new Error('job stalled'), + ); + + await vi.waitFor(() => + expect(mocks.recordOutcome).toHaveBeenCalledWith( + {}, + expect.objectContaining({ + status: 'failed', + error: 'job stalled', + launchClaimedAt: new Date(launchClaimedAt), + }), + ), + ); + }); +}); diff --git a/apps/bullmq/src/custom-automation-run-queue.ts b/apps/bullmq/src/custom-automation-run-queue.ts new file mode 100644 index 000000000..12dc6932f --- /dev/null +++ b/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( + CUSTOM_AUTOMATION_RUN_QUEUE_NAME, + { connection }, + ); + const worker = new Worker( + 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 }; +} diff --git a/apps/bullmq/src/index.ts b/apps/bullmq/src/index.ts index 8e5fc9172..88137503a 100644 --- a/apps/bullmq/src/index.ts +++ b/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) { diff --git a/apps/web/src/components/settings/automations/CustomAutomationsSection.tsx b/apps/web/src/components/settings/automations/CustomAutomationsSection.tsx index a3308a61c..885d43d7d 100644 --- a/apps/web/src/components/settings/automations/CustomAutomationsSection.tsx +++ b/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: { diff --git a/apps/web/src/trpc/commands/automations/custom-automations.ts b/apps/web/src/trpc/commands/automations/custom-automations.ts index e3a4645ec..7cb74ace1 100644 --- a/apps/web/src/trpc/commands/automations/custom-automations.ts +++ b/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 { +): Promise { assertAdmin(auth); return runCustomAutomationNow(input.id); } diff --git a/apps/worker/src/mcp/roomote-mcp-server/__tests__/api-client.test.ts b/apps/worker/src/mcp/roomote-mcp-server/__tests__/api-client.test.ts index 510da4679..63400e8d7 100644 --- a/apps/worker/src/mcp/roomote-mcp-server/__tests__/api-client.test.ts +++ b/apps/worker/src/mcp/roomote-mcp-server/__tests__/api-client.test.ts @@ -37,7 +37,7 @@ describe('fetchWithTimeout', () => { delete process.env.ROOMOTE_MCP_PLATFORM_API_TIMEOUT_MS; }); - it('rejects with a retryable error when the API never responds', async () => { + it('does not describe a timed-out POST as safe to retry', async () => { process.env.ROOMOTE_MCP_PLATFORM_API_TIMEOUT_MS = '25'; global.fetch = neverRespondingFetch() as unknown as typeof fetch; @@ -48,10 +48,23 @@ describe('fetchWithTimeout', () => { { label: 'Failed to manage source control' }, ), ).rejects.toThrow( - 'Failed to manage source control: no response from the Roomote API within 25ms; the request was aborted and is safe to retry.', + 'Failed to manage source control: no response from the Roomote API within 25ms; the request was aborted, but the operation may still complete; check its status before retrying.', ); }); + it('describes a timed-out GET as safe to retry', async () => { + process.env.ROOMOTE_MCP_PLATFORM_API_TIMEOUT_MS = '25'; + global.fetch = neverRespondingFetch() as unknown as typeof fetch; + + await expect( + fetchWithTimeout( + 'https://test-api.example.com/api/mcp/tasks', + { method: 'GET' }, + { label: 'Failed to search tasks' }, + ), + ).rejects.toThrow('the request was aborted and is safe to retry.'); + }); + it('honors an explicit timeoutMs over the environment default', async () => { process.env.ROOMOTE_MCP_PLATFORM_API_TIMEOUT_MS = '60000'; global.fetch = neverRespondingFetch() as unknown as typeof fetch; diff --git a/apps/worker/src/mcp/roomote-mcp-server/__tests__/tasks-api-client.test.ts b/apps/worker/src/mcp/roomote-mcp-server/__tests__/tasks-api-client.test.ts index 23e152de9..e57fb9bba 100644 --- a/apps/worker/src/mcp/roomote-mcp-server/__tests__/tasks-api-client.test.ts +++ b/apps/worker/src/mcp/roomote-mcp-server/__tests__/tasks-api-client.test.ts @@ -844,7 +844,7 @@ describe('manageSourceControl timeout', () => { body: 'Body', }), ).rejects.toThrow( - 'Failed to manage source control: no response from the Roomote API within 25ms; the request was aborted and is safe to retry.', + 'Failed to manage source control: no response from the Roomote API within 25ms; the request was aborted, but the operation may still complete; check its status before retrying.', ); }); }); diff --git a/apps/worker/src/mcp/roomote-mcp-server/api-client.ts b/apps/worker/src/mcp/roomote-mcp-server/api-client.ts index ebcfbec0d..1c2934957 100644 --- a/apps/worker/src/mcp/roomote-mcp-server/api-client.ts +++ b/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.' + }`, ); } diff --git a/packages/db/src/lib/__tests__/custom-automations.test.ts b/packages/db/src/lib/__tests__/custom-automations.test.ts index 8557cb615..d8e2390f3 100644 --- a/packages/db/src/lib/__tests__/custom-automations.test.ts +++ b/packages/db/src/lib/__tests__/custom-automations.test.ts @@ -8,6 +8,7 @@ import { listCustomAutomations, recordCustomAutomationRunOutcome, releaseCustomAutomationLaunchClaim, + renewCustomAutomationLaunchClaim, tryClaimCustomAutomationLaunch, updateCustomAutomation, } from '../custom-automations'; @@ -277,6 +278,49 @@ describe('custom automations helpers', () => { await deleteCustomAutomation(created.id); }); + it('does not reclaim an old claim while its lease heartbeat is fresh', async () => { + const [environment] = await db + .insert(environments) + .values({ + name: `custom-auto-env-lease-${Date.now()}`, + config: { name: 'test', repositories: [] }, + }) + .returning(); + const created = await createCustomAutomation({ + name: `Lease gate ${Date.now()}`, + prompt: 'Run a long automation.', + enabled: true, + scheduleMode: 'daily', + environmentId: environment!.id, + target: {}, + }); + const oldClaim = new Date(Date.now() - 11 * 60 * 1_000); + await db + .update(customAutomations) + .set({ launchClaimedAt: oldClaim, updatedAt: oldClaim }) + .where(eq(customAutomations.id, created.id)); + + await expect( + renewCustomAutomationLaunchClaim(created.id, oldClaim), + ).resolves.toBe(true); + await expect( + tryClaimCustomAutomationLaunch(created.id, created.lastRunAt), + ).resolves.toBeNull(); + + await db + .update(customAutomations) + .set({ updatedAt: oldClaim }) + .where(eq(customAutomations.id, created.id)); + const reclaimed = await tryClaimCustomAutomationLaunch( + created.id, + created.lastRunAt, + ); + expect(reclaimed).toBeInstanceOf(Date); + + await releaseCustomAutomationLaunchClaim(created.id, reclaimed!); + await deleteCustomAutomation(created.id); + }); + it('rejects a partially specified report destination', async () => { await expect( createCustomAutomation({ diff --git a/packages/db/src/lib/custom-automations.ts b/packages/db/src/lib/custom-automations.ts index 2a0b0236e..583c1756a 100644 --- a/packages/db/src/lib/custom-automations.ts +++ b/packages/db/src/lib/custom-automations.ts @@ -391,7 +391,10 @@ export async function tryClaimCustomAutomationLaunch( eq(customAutomations.id, id), or( isNull(customAutomations.launchClaimedAt), - lt(customAutomations.launchClaimedAt, staleBefore), + and( + lt(customAutomations.launchClaimedAt, staleBefore), + lt(customAutomations.updatedAt, staleBefore), + ), )!, expectedLastRunAt ? eq(customAutomations.lastRunAt, expectedLastRunAt) @@ -403,6 +406,25 @@ export async function tryClaimCustomAutomationLaunch( return claimed?.launchClaimedAt ?? null; } +export async function renewCustomAutomationLaunchClaim( + id: string, + launchClaimedAt: Date, + client: DatabaseOrTransaction = db, +): Promise { + const renewed = await client + .update(customAutomations) + .set({ updatedAt: new Date() }) + .where( + and( + eq(customAutomations.id, id), + eq(customAutomations.launchClaimedAt, launchClaimedAt), + ), + ) + .returning({ id: customAutomations.id }); + + return renewed.length > 0; +} + export async function releaseCustomAutomationLaunchClaim( id: string, launchClaimedAt: Date, diff --git a/packages/sdk/src/server/automations/__tests__/custom-automation-run-queue.test.ts b/packages/sdk/src/server/automations/__tests__/custom-automation-run-queue.test.ts new file mode 100644 index 000000000..8f7147d18 --- /dev/null +++ b/packages/sdk/src/server/automations/__tests__/custom-automation-run-queue.test.ts @@ -0,0 +1,67 @@ +const mocks = vi.hoisted(() => ({ + add: vi.fn(), + getJob: vi.fn(), +})); + +vi.mock('bullmq', () => ({ + Queue: class Queue { + add = mocks.add; + getJob = mocks.getJob; + }, +})); + +vi.mock('@roomote/redis', () => ({ getRedis: vi.fn(() => ({})) })); + +import { + CUSTOM_AUTOMATION_RUN_JOB_NAME, + enqueueCustomAutomationRun, + getCustomAutomationRunStatus, +} from '../custom-automation-run-queue'; + +const automationId = '11111111-1111-4111-8111-111111111111'; +const launchClaimedAt = new Date('2026-08-25T17:41:38.469Z'); +const invocationId = `${automationId}-${launchClaimedAt.getTime()}`; + +describe('custom automation run queue', () => { + beforeEach(() => vi.clearAllMocks()); + + it('enqueues a durable invocation with a deterministic identifier', async () => { + await expect( + enqueueCustomAutomationRun({ automationId, launchClaimedAt }), + ).resolves.toBe(invocationId); + + expect(mocks.add).toHaveBeenCalledWith( + CUSTOM_AUTOMATION_RUN_JOB_NAME, + { automationId, launchClaimedAt: launchClaimedAt.toISOString() }, + { jobId: invocationId }, + ); + }); + + it('reports successful terminal status', async () => { + mocks.getJob.mockResolvedValue({ + data: { automationId }, + getState: vi.fn().mockResolvedValue('completed'), + }); + + await expect( + getCustomAutomationRunStatus({ automationId, invocationId }), + ).resolves.toEqual({ automationId, invocationId, status: 'succeeded' }); + }); + + it('reports failed terminal status', async () => { + mocks.getJob.mockResolvedValue({ + data: { automationId }, + failedReason: 'provider unavailable', + getState: vi.fn().mockResolvedValue('failed'), + }); + + await expect( + getCustomAutomationRunStatus({ automationId, invocationId }), + ).resolves.toEqual({ + automationId, + invocationId, + status: 'failed', + error: 'provider unavailable', + }); + }); +}); diff --git a/packages/sdk/src/server/automations/__tests__/custom-automations.test.ts b/packages/sdk/src/server/automations/__tests__/custom-automations.test.ts index 8e7b44c94..2e655c705 100644 --- a/packages/sdk/src/server/automations/__tests__/custom-automations.test.ts +++ b/packages/sdk/src/server/automations/__tests__/custom-automations.test.ts @@ -5,6 +5,11 @@ const fastMocks = vi.hoisted(() => ({ deliverParentEvent: vi.fn(), slackPostMessage: vi.fn(), })); +const runQueueMocks = vi.hoisted(() => ({ enqueue: vi.fn() })); + +vi.mock('../custom-automation-run-queue', () => ({ + enqueueCustomAutomationRun: runQueueMocks.enqueue, +})); vi.mock('@roomote/cloud-agents/server', () => ({ enqueueTask: vi.fn(), @@ -50,6 +55,7 @@ vi.mock('@roomote/db/server', () => ({ listEnabledCustomAutomations: vi.fn(), recordCustomAutomationRunOutcome: vi.fn(), releaseCustomAutomationLaunchClaim: vi.fn(), + renewCustomAutomationLaunchClaim: vi.fn(), tryClaimCustomAutomationLaunch: vi.fn(), slackInstallations: {}, })); @@ -93,6 +99,7 @@ import { listEnabledCustomAutomations, recordCustomAutomationRunOutcome, releaseCustomAutomationLaunchClaim, + renewCustomAutomationLaunchClaim, tryClaimCustomAutomationLaunch, } from '@roomote/db/server'; import { ALL_REPOSITORIES, TaskPayloadKind } from '@roomote/types'; @@ -100,6 +107,7 @@ import { findUserDirectMessageDestination } from '../../lib/user-direct-message' import { customAutomationsJob, + runClaimedFastCustomAutomation, runCustomAutomationNow, } from '../custom-automations'; import { @@ -166,6 +174,7 @@ describe('customAutomationsJob', () => { }); fastMocks.deliverParentEvent.mockResolvedValue('delivered'); fastMocks.slackPostMessage.mockResolvedValue('100.001'); + runQueueMocks.enqueue.mockResolvedValue('invocation-1'); }); it('runs a channel-less Fast automation without enqueueing a task', async () => { @@ -275,6 +284,7 @@ describe('customAutomationsJob', () => { target: {}, createdByUserId: 'user-1', launchClaimedAt: staleClaim, + updatedAt: staleClaim, } as never, ]); @@ -757,6 +767,43 @@ describe('runCustomAutomationNow', () => { ); }); + it('accepts a Fast manual run after claiming and enqueueing it', async () => { + const claimAt = new Date('2026-08-25T17:41:38.469Z'); + vi.mocked(getCustomAutomationById).mockResolvedValue({ + ...automation, + executionMode: 'fast', + environmentId: null, + createdByUserId: 'user-1', + } as never); + vi.mocked(tryClaimCustomAutomationLaunch).mockResolvedValue(claimAt); + + await expect(runCustomAutomationNow(automation.id)).resolves.toEqual({ + outcome: 'accepted', + invocationId: 'invocation-1', + }); + expect(runQueueMocks.enqueue).toHaveBeenCalledWith({ + automationId: automation.id, + launchClaimedAt: claimAt, + }); + expect(fastMocks.deliverParentEvent).not.toHaveBeenCalled(); + }); + + it('does not enqueue a duplicate Fast manual run while claimed', async () => { + vi.mocked(getCustomAutomationById).mockResolvedValue({ + ...automation, + executionMode: 'fast', + environmentId: null, + createdByUserId: 'user-1', + } as never); + vi.mocked(tryClaimCustomAutomationLaunch).mockResolvedValue(null); + + await expect(runCustomAutomationNow(automation.id)).resolves.toEqual({ + outcome: 'skipped', + reason: 'Another launch is already in progress.', + }); + expect(runQueueMocks.enqueue).not.toHaveBeenCalled(); + }); + it('skips manual run when a concurrent launch holds the claim', async () => { vi.mocked(tryClaimCustomAutomationLaunch).mockResolvedValue(null); @@ -795,3 +842,59 @@ describe('runCustomAutomationNow', () => { expect(enqueueTask).not.toHaveBeenCalled(); }); }); + +describe('runClaimedFastCustomAutomation', () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(recordCustomAutomationRunOutcome).mockResolvedValue(true); + vi.mocked(renewCustomAutomationLaunchClaim).mockResolvedValue(true); + }); + + it('clears the fenced claim when the execution mode changed', async () => { + const claimAt = new Date('2026-08-25T17:41:38.469Z'); + vi.mocked(getCustomAutomationById).mockResolvedValue({ + ...automation, + launchClaimedAt: claimAt, + } as never); + + await expect( + runClaimedFastCustomAutomation({ + automationId: automation.id, + launchClaimedAt: claimAt.toISOString(), + }), + ).rejects.toThrow('invocation is no longer active'); + expect(recordCustomAutomationRunOutcome).toHaveBeenCalledWith( + db, + expect.objectContaining({ + id: automation.id, + status: 'failed', + launchClaimedAt: claimAt, + }), + ); + }); + + it('rejects an invocation that outlived its claim window', async () => { + const claimAt = new Date(Date.now() - 10 * 60 * 1_000 - 1); + vi.mocked(getCustomAutomationById).mockResolvedValue({ + ...automation, + executionMode: 'fast', + environmentId: null, + createdByUserId: 'user-1', + launchClaimedAt: claimAt, + } as never); + + await expect( + runClaimedFastCustomAutomation({ + automationId: automation.id, + launchClaimedAt: claimAt.toISOString(), + }), + ).rejects.toThrow('invocation expired before it started'); + expect(recordCustomAutomationRunOutcome).toHaveBeenCalledWith( + db, + expect.objectContaining({ + status: 'failed', + launchClaimedAt: claimAt, + }), + ); + }); +}); diff --git a/packages/sdk/src/server/automations/custom-automation-run-queue.ts b/packages/sdk/src/server/automations/custom-automation-run-queue.ts new file mode 100644 index 000000000..40c208cef --- /dev/null +++ b/packages/sdk/src/server/automations/custom-automation-run-queue.ts @@ -0,0 +1,80 @@ +import { Queue } from 'bullmq'; +import { z } from 'zod'; + +import { getRedis } from '@roomote/redis'; + +export const CUSTOM_AUTOMATION_RUN_QUEUE_NAME = 'custom-automation-runs'; +export const CUSTOM_AUTOMATION_RUN_JOB_NAME = 'run-custom-automation'; + +export const customAutomationRunJobSchema = z.object({ + automationId: z.string().uuid(), + launchClaimedAt: z.string().datetime(), +}); + +export type CustomAutomationRunJob = z.infer< + typeof customAutomationRunJobSchema +>; + +export type CustomAutomationRunStatus = { + automationId: string; + invocationId: string; + status: 'queued' | 'running' | 'succeeded' | 'failed'; + error?: string; +}; + +let queue: Queue | null = null; + +function getQueue(): Queue { + queue ??= new Queue( + CUSTOM_AUTOMATION_RUN_QUEUE_NAME, + { + connection: getRedis(), + defaultJobOptions: { + attempts: 1, + removeOnComplete: { age: 24 * 60 * 60, count: 500 }, + removeOnFail: { age: 7 * 24 * 60 * 60, count: 500 }, + }, + }, + ); + return queue; +} + +export async function enqueueCustomAutomationRun(params: { + automationId: string; + launchClaimedAt: Date; +}): Promise { + const invocationId = `${params.automationId}-${params.launchClaimedAt.getTime()}`; + await getQueue().add( + CUSTOM_AUTOMATION_RUN_JOB_NAME, + { + automationId: params.automationId, + launchClaimedAt: params.launchClaimedAt.toISOString(), + }, + { jobId: invocationId }, + ); + return invocationId; +} + +export async function getCustomAutomationRunStatus(params: { + automationId: string; + invocationId: string; +}): Promise { + const job = await getQueue().getJob(params.invocationId); + if (!job || job.data.automationId !== params.automationId) return null; + + const state = await job.getState(); + if (state === 'completed') { + return { ...params, status: 'succeeded' }; + } + if (state === 'failed') { + return { + ...params, + status: 'failed', + error: job.failedReason || 'Custom automation run failed.', + }; + } + return { + ...params, + status: state === 'active' ? 'running' : 'queued', + }; +} diff --git a/packages/sdk/src/server/automations/custom-automations.ts b/packages/sdk/src/server/automations/custom-automations.ts index 56f8d7d23..dc5540884 100644 --- a/packages/sdk/src/server/automations/custom-automations.ts +++ b/packages/sdk/src/server/automations/custom-automations.ts @@ -15,6 +15,7 @@ import { listEnabledCustomAutomations, recordCustomAutomationRunOutcome, releaseCustomAutomationLaunchClaim, + renewCustomAutomationLaunchClaim, tryClaimCustomAutomationLaunch, type CustomAutomation, slackInstallations, @@ -48,9 +49,10 @@ import { DAILY_WEEKLY_SCHEDULE_HOUR_LOCAL, isRunDue } from './scheduling-utils'; import { emptyJobResult, type AutomationJobResult, - type AutomationRunNowResult, type AutomationRunOpts, + type CustomAutomationRunNowResult, } from './types'; +import { enqueueCustomAutomationRun } from './custom-automation-run-queue'; import { findUserDirectMessageDestination } from '../lib/user-direct-message'; import { createDiscordCommunicationProviderFromRuntimeCredentials } from '../lib/discord-communication'; import { buildCustomAutomationSlackMessage } from '../lib/manager-slack'; @@ -61,6 +63,7 @@ import { } from '../lib/fast-agent-parent-event'; const LOG_PREFIX = '[custom-automations]'; +const CLAIM_HEARTBEAT_INTERVAL_MS = 60_000; const PROVIDER_LABELS: Record = { discord: 'Discord', @@ -424,6 +427,7 @@ async function launchCustomAutomationRow( automation: CustomAutomation, opts: AutomationRunOpts, scheduleContext?: ResolvedDeploymentTimeZone, + preclaimedLaunchAt?: Date, ): Promise { const result = emptyJobResult(); const frequency = getCustomAutomationFrequency(automation); @@ -481,9 +485,10 @@ async function launchCustomAutomationRow( } if ( + !preclaimedLaunchAt && fastExecution && automation.launchClaimedAt && - Date.now() - automation.launchClaimedAt.getTime() >= + Date.now() - automation.updatedAt.getTime() >= CUSTOM_AUTOMATION_LAUNCH_STALE_CLAIM_MS ) { const message = 'The previous Fast automation run was interrupted.'; @@ -580,10 +585,9 @@ async function launchCustomAutomationRow( // The short claim fence prevents concurrent launchers from double-launching // without blocking a due run behind a previous task that still appears active. - const launchClaimedAt = await tryClaimCustomAutomationLaunch( - automation.id, - automation.lastRunAt, - ); + const launchClaimedAt = + preclaimedLaunchAt ?? + (await tryClaimCustomAutomationLaunch(automation.id, automation.lastRunAt)); if (!launchClaimedAt) { result.skippedReason = 'Another launch is already in progress.'; return result; @@ -699,7 +703,74 @@ async function launchCustomAutomationRow( result.completed = true; return result; } catch (error) { - await releaseCustomAutomationLaunchClaim(automation.id, launchClaimedAt); + if (!preclaimedLaunchAt) { + await releaseCustomAutomationLaunchClaim(automation.id, launchClaimedAt); + } + throw error; + } +} + +export async function runClaimedFastCustomAutomation(params: { + automationId: string; + launchClaimedAt: string; +}): Promise { + const launchClaimedAt = new Date(params.launchClaimedAt); + const automation = await getCustomAutomationById(params.automationId); + + try { + if ( + !automation || + automation.executionMode !== 'fast' || + automation.launchClaimedAt?.getTime() !== launchClaimedAt.getTime() + ) { + throw new Error('Custom automation invocation is no longer active.'); + } + if ( + Date.now() - launchClaimedAt.getTime() >= + CUSTOM_AUTOMATION_LAUNCH_STALE_CLAIM_MS + ) { + throw new Error( + 'Custom automation invocation expired before it started.', + ); + } + + const heartbeat = setInterval(() => { + void renewCustomAutomationLaunchClaim( + automation.id, + launchClaimedAt, + ).catch((error) => + console.warn( + `${LOG_PREFIX} Failed to renew invocation ${params.automationId}: ${error instanceof Error ? error.message : String(error)}`, + ), + ); + }, CLAIM_HEARTBEAT_INTERVAL_MS); + heartbeat.unref(); + + const result = await (async () => { + try { + return await launchCustomAutomationRow( + automation, + { manualTrigger: true }, + undefined, + launchClaimedAt, + ); + } finally { + clearInterval(heartbeat); + } + })(); + const error = result.errors.join('; ') || result.skippedReason; + if (error) throw new Error(error); + if (!result.completed) { + throw new Error('Custom automation invocation did not complete.'); + } + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + await recordCustomAutomationRunOutcome(db, { + id: params.automationId, + status: 'failed', + error: message, + launchClaimedAt, + }); throw error; } } @@ -764,7 +835,7 @@ export async function customAutomationsJob( export async function runCustomAutomationNow( id: string, -): Promise { +): Promise { const automation = await getCustomAutomationById(id); if (!automation) { @@ -779,6 +850,33 @@ export async function runCustomAutomationNow( }; } + if (automation.executionMode === 'fast') { + const launchClaimedAt = await tryClaimCustomAutomationLaunch( + automation.id, + automation.lastRunAt, + ); + if (!launchClaimedAt) { + return { + outcome: 'skipped', + reason: 'Another launch is already in progress.', + }; + } + + try { + const invocationId = await enqueueCustomAutomationRun({ + automationId: automation.id, + launchClaimedAt, + }); + return { outcome: 'accepted', invocationId }; + } catch (error) { + await releaseCustomAutomationLaunchClaim(automation.id, launchClaimedAt); + return { + outcome: 'failed', + error: error instanceof Error ? error.message : String(error), + }; + } + } + try { const result = await launchCustomAutomationRow(automation, { manualTrigger: true, diff --git a/packages/sdk/src/server/automations/index.ts b/packages/sdk/src/server/automations/index.ts index 7c848aaf9..020e4e870 100644 --- a/packages/sdk/src/server/automations/index.ts +++ b/packages/sdk/src/server/automations/index.ts @@ -1,8 +1,18 @@ export { announcerJob } from './announcer'; export { customAutomationsJob, + runClaimedFastCustomAutomation, runCustomAutomationNow, } from './custom-automations'; +export { + CUSTOM_AUTOMATION_RUN_JOB_NAME, + CUSTOM_AUTOMATION_RUN_QUEUE_NAME, + customAutomationRunJobSchema, + enqueueCustomAutomationRun, + getCustomAutomationRunStatus, + type CustomAutomationRunJob, + type CustomAutomationRunStatus, +} from './custom-automation-run-queue'; export * from './custom-automation-schedule'; export { ciFailureTriageJob } from './ci-failure-triage'; export { @@ -50,5 +60,6 @@ export { export type { AutomationJobResult, AutomationRunNowResult, + CustomAutomationRunNowResult, AutomationRunOpts, } from './types'; diff --git a/packages/sdk/src/server/automations/types.ts b/packages/sdk/src/server/automations/types.ts index b19fe847e..2700ac24f 100644 --- a/packages/sdk/src/server/automations/types.ts +++ b/packages/sdk/src/server/automations/types.ts @@ -40,3 +40,7 @@ export type AutomationRunNowResult = | { outcome: 'completed' } | { outcome: 'skipped'; reason: string } | { outcome: 'failed'; error: string }; + +export type CustomAutomationRunNowResult = + | AutomationRunNowResult + | { outcome: 'accepted'; invocationId: string }; diff --git a/packages/types/src/manage-custom-automations-tool.test.ts b/packages/types/src/manage-custom-automations-tool.test.ts index 80ca11422..e29c39f4f 100644 --- a/packages/types/src/manage-custom-automations-tool.test.ts +++ b/packages/types/src/manage-custom-automations-tool.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from 'vitest'; import { MANAGE_CUSTOM_AUTOMATIONS_ACTIONS, MANAGE_CUSTOM_AUTOMATIONS_TOOL, + buildManageCustomAutomationsRequest, manageCustomAutomationsInputSchema, } from './manage-custom-automations-tool'; @@ -35,4 +36,20 @@ describe('manage custom automations tool contract', () => { MANAGE_CUSTOM_AUTOMATIONS_TOOL.inputSchema.prompt.description, ).toContain('Do not mention internal tool names or parameters.'); }); + + it('builds the accepted-run polling request', () => { + expect( + buildManageCustomAutomationsRequest({ + action: 'run_status', + automationId: 'automation-1', + invocationId: 'invocation-1', + }), + ).toEqual({ + ok: true, + request: { + path: '/automation-1/runs/invocation-1', + method: 'GET', + }, + }); + }); }); diff --git a/packages/types/src/manage-custom-automations-tool.ts b/packages/types/src/manage-custom-automations-tool.ts index eb33abf3d..16cb9c529 100644 --- a/packages/types/src/manage-custom-automations-tool.ts +++ b/packages/types/src/manage-custom-automations-tool.ts @@ -11,6 +11,7 @@ export const MANAGE_CUSTOM_AUTOMATIONS_ACTIONS = [ 'update', 'delete', 'run_now', + 'run_status', ] as const; export const manageCustomAutomationsFieldSchemas = { @@ -18,7 +19,11 @@ export const manageCustomAutomationsFieldSchemas = { automationId: z .string() .optional() - .describe('Required for update, delete, and run_now.'), + .describe('Required for update, delete, run_now, and run_status.'), + invocationId: z + .string() + .optional() + .describe('Required for run_status after run_now returns accepted.'), name: z.string().optional(), prompt: z .string() @@ -155,19 +160,31 @@ export function buildManageCustomAutomationsRequest( } case 'delete': case 'run_now': + case 'run_status': if (!params.automationId) { return { ok: false, error: `automationId is required for ${params.action}`, }; } + if (params.action === 'run_status' && !params.invocationId) { + return { ok: false, error: 'invocationId is required for run_status' }; + } return { ok: true, request: { - path: `/${encodeURIComponent(params.automationId)}${ - params.action === 'run_now' ? '/run' : '' - }`, - method: params.action === 'delete' ? 'DELETE' : 'POST', + path: + params.action === 'run_status' + ? `/${encodeURIComponent(params.automationId)}/runs/${encodeURIComponent(params.invocationId!)}` + : `/${encodeURIComponent(params.automationId)}${ + params.action === 'run_now' ? '/run' : '' + }`, + method: + params.action === 'delete' + ? 'DELETE' + : params.action === 'run_status' + ? 'GET' + : 'POST', }, }; } @@ -176,7 +193,7 @@ export function buildManageCustomAutomationsRequest( export const MANAGE_CUSTOM_AUTOMATIONS_TOOL = { name: 'manage_custom_automations', title: 'Manage Custom Automations', - description: `Admin-only management of deployment custom automations. List existing automations or enabled task models, resolve a cron or natural-language schedule, create or update an automation, delete an automation by exact ID, or run an enabled automation now. Pass environmentId "${FAST_EXECUTION}" to run the automation in Fast mode without starting a sandbox; Fast may still delegate a task when repository or workspace execution is required. Use list_models before setting a model override; create and update accept only exact model IDs returned by that action. Model IDs encode the inference route: for example, openrouter/... targets OpenRouter, while openai/... uses the deployment OpenAI route, including a connected ChatGPT subscription when configured. When the user asks an automation to DM them, set their preferred connected targetProvider and targetMode to direct_message; no targetChannelId is needed. Natural-language schedules are converted to validated five-field cron in the deployment scheduling timezone. Keep cadence only in the schedule field; do not repeat it in the stored prompt. When a user asks an automation to offer help, suggest tasks, make follow-ups actionable or launchable, or turn findings or action items into tasks, encode that intent in product language by instructing the automation to post concrete actions as launchable suggested tasks alongside its report. Do not expose runtime tool names or parameter syntax in the stored prompt. A request only to summarize or list action items is not suggested-task intent. Only promise launchable suggested tasks when the automation has both a configured chat report destination and a repository or environment for executable work; otherwise keep actions as report text and explain the missing capability. After successfully creating an automation in response to a conversational request, ask the user whether they want to run it now to test it.`, + description: `Admin-only management of deployment custom automations. List existing automations or enabled task models, resolve a cron or natural-language schedule, create or update an automation, delete an automation by exact ID, run an enabled automation now, or poll an accepted Fast run with run_status. Pass environmentId "${FAST_EXECUTION}" to run the automation in Fast mode without starting a sandbox; Fast may still delegate a task when repository or workspace execution is required. Use list_models before setting a model override; create and update accept only exact model IDs returned by that action. Model IDs encode the inference route: for example, openrouter/... targets OpenRouter, while openai/... uses the deployment OpenAI route, including a connected ChatGPT subscription when configured. When the user asks an automation to DM them, set their preferred connected targetProvider and targetMode to direct_message; no targetChannelId is needed. Natural-language schedules are converted to validated five-field cron in the deployment scheduling timezone. Keep cadence only in the schedule field; do not repeat it in the stored prompt. When a user asks an automation to offer help, suggest tasks, make follow-ups actionable or launchable, or turn findings or action items into tasks, encode that intent in product language by instructing the automation to post concrete actions as launchable suggested tasks alongside its report. Do not expose runtime tool names or parameter syntax in the stored prompt. A request only to summarize or list action items is not suggested-task intent. Only promise launchable suggested tasks when the automation has both a configured chat report destination and a repository or environment for executable work; otherwise keep actions as report text and explain the missing capability. After successfully creating an automation in response to a conversational request, ask the user whether they want to run it now to test it.`, inputSchema: manageCustomAutomationsFieldSchemas, annotations: { readOnlyHint: false, From e0be5868dfd36f4e1bcd9b42407c30050fea5208 Mon Sep 17 00:00:00 2001 From: "@daniel-lxs" <57051444+daniel-lxs@users.noreply.github.com> Date: Tue, 25 Aug 2026 18:44:57 +0000 Subject: [PATCH 2/2] test: isolate Brain backfill idempotency --- packages/db/src/lib/__tests__/brain.test.ts | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/packages/db/src/lib/__tests__/brain.test.ts b/packages/db/src/lib/__tests__/brain.test.ts index 8ed8fe580..8ae24a02c 100644 --- a/packages/db/src/lib/__tests__/brain.test.ts +++ b/packages/db/src/lib/__tests__/brain.test.ts @@ -25,6 +25,7 @@ import { releaseBrainMemoryEvents, maybeEnqueueBrainMemoryEvent, saveBrainAgentSummary, + sql, resetBrainIngestionState, canonicalizeBrainCollectorItemSlugs, deleteBrainCollectorItems, @@ -513,8 +514,15 @@ describe('backfillBrainMemoryEvents', () => { }) .returning(); - const first = await backfillBrainMemoryEvents(db); - const second = await backfillBrainMemoryEvents(db); + const [first, second] = await db.transaction(async (tx) => { + await tx.execute(sql`SET TRANSACTION ISOLATION LEVEL REPEATABLE READ`); + const firstResult = await backfillBrainMemoryEvents(tx); + // Root CI runs package suites concurrently against one database. A run + // completed by another package must not invalidate this idempotency pass. + await makeCompletedRun(); + const secondResult = await backfillBrainMemoryEvents(tx); + return [firstResult, secondResult]; + }); expect(first).toBeGreaterThanOrEqual(1); expect(second).toBe(0);