From c785fca3d3f9f2c126f91c60f53699aca6bf10e4 Mon Sep 17 00:00:00 2001 From: "@daniel-lxs" <57051444+daniel-lxs@users.noreply.github.com> Date: Thu, 3 Sep 2026 17:57:47 +0000 Subject: [PATCH 1/6] fix: durably queue parent session notifications --- .../notify-fast-agent-parent.test.ts | 48 ++++++++++---- .../lib/artifacts/notify-fast-agent-parent.ts | 46 +++++++------ .../lib/fast-agent-parent-event-queue.test.ts | 58 ++++++++++++++++ .../lib/fast-agent-parent-event-queue.ts | 38 +++++++++++ ...y-fast-agent-parent-on-pr-feedback.test.ts | 61 +++++++++-------- ...ent-on-pull-request-status-changed.test.ts | 37 ++++++----- ...notify-fast-agent-parent-on-pr-feedback.ts | 66 ++++++++++--------- ...t-agent-parent-on-pull-request-conflict.ts | 50 +++++++------- 8 files changed, 269 insertions(+), 135 deletions(-) diff --git a/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts b/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts index 36ed296b1..c7feec0c1 100644 --- a/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts +++ b/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts @@ -18,7 +18,7 @@ const mocks = vi.hoisted(() => { claimReturning: vi.fn(), updateSet: vi.fn(), recordLifecycle: vi.fn(), - deliverParentEvent: vi.fn(), + enqueueParentEventAndWait: vi.fn(), FastAgentParentEventDeliveryError, }; }); @@ -54,10 +54,13 @@ vi.mock('@roomote/env', () => ({ })); vi.mock('../../fast-agent-parent-event', () => ({ - deliverFastAgentParentEvent: mocks.deliverParentEvent, FastAgentParentEventDeliveryError: mocks.FastAgentParentEventDeliveryError, })); +vi.mock('../../fast-agent-parent-event-queue', () => ({ + enqueueFastAgentParentEventAndWait: mocks.enqueueParentEventAndWait, +})); + import { notifyFastAgentParentOnArtifact } from '../notify-fast-agent-parent'; const fastParent = { @@ -97,7 +100,7 @@ describe('notifyFastAgentParentOnArtifact', () => { result: {}, }); mocks.claimReturning.mockResolvedValue([{ id: 200 }]); - mocks.deliverParentEvent.mockResolvedValue(undefined); + mocks.enqueueParentEventAndWait.mockResolvedValue('delivered'); mocks.recordLifecycle.mockResolvedValue(undefined); }); @@ -106,10 +109,9 @@ describe('notifyFastAgentParentOnArtifact', () => { 'delivered', ); - expect(mocks.deliverParentEvent).toHaveBeenCalledWith( - expect.objectContaining({ + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledWith( + { parent: fastParent, - lockWaitMs: expect.any(Number), event: expect.objectContaining({ type: 'artifact_published', taskId: 'child-task', @@ -122,7 +124,8 @@ describe('notifyFastAgentParentOnArtifact', () => { 'https://roomote.example/task/child-task/artifacts/proof/result.png?v=1', }), }), - }), + }, + { timeoutMs: 30_000 }, ); expect(mocks.recordLifecycle).toHaveBeenCalledWith( expect.anything(), @@ -140,11 +143,13 @@ describe('notifyFastAgentParentOnArtifact', () => { await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( 'already_delivered', ); - expect(mocks.deliverParentEvent).not.toHaveBeenCalled(); + expect(mocks.enqueueParentEventAndWait).not.toHaveBeenCalled(); }); it('releases a failed orchestrator delivery for retry', async () => { - mocks.deliverParentEvent.mockRejectedValueOnce(new Error('model offline')); + mocks.enqueueParentEventAndWait.mockRejectedValueOnce( + new Error('model offline'), + ); await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( 'failed', @@ -171,11 +176,11 @@ describe('notifyFastAgentParentOnArtifact', () => { await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( 'in_progress', ); - expect(mocks.deliverParentEvent).not.toHaveBeenCalled(); + expect(mocks.enqueueParentEventAndWait).not.toHaveBeenCalled(); }); it('keeps the claim when the failure happened after the Slack post', async () => { - mocks.deliverParentEvent.mockRejectedValueOnce( + mocks.enqueueParentEventAndWait.mockRejectedValueOnce( new mocks.FastAgentParentEventDeliveryError('lifecycle write failed', { replyPosted: true, }), @@ -193,7 +198,7 @@ describe('notifyFastAgentParentOnArtifact', () => { }); it('settles the claim as skipped when no retry can ever succeed', async () => { - mocks.deliverParentEvent.mockRejectedValueOnce( + mocks.enqueueParentEventAndWait.mockRejectedValueOnce( new mocks.FastAgentParentEventDeliveryError('parent session gone', { replyPosted: false, permanent: true, @@ -211,6 +216,21 @@ describe('notifyFastAgentParentOnArtifact', () => { ).toBe(true); }); + it('settles the claim when the durable queue discards the event', async () => { + mocks.enqueueParentEventAndWait.mockResolvedValueOnce('skipped'); + + await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( + 'skipped', + ); + expect( + mocks.updateSet.mock.calls.some(([values]) => { + const result = (values as { result?: { values?: unknown[] } }).result; + return result?.values?.includes('skipped') === true; + }), + ).toBe(true); + expect(mocks.recordLifecycle).not.toHaveBeenCalled(); + }); + it('uses inherited Fast parent metadata on resumed runs', async () => { mocks.findRun.mockResolvedValueOnce({ id: 200, @@ -226,7 +246,7 @@ describe('notifyFastAgentParentOnArtifact', () => { await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( 'delivered', ); - expect(mocks.deliverParentEvent).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); }); it('does nothing for standalone artifacts', async () => { @@ -240,6 +260,6 @@ describe('notifyFastAgentParentOnArtifact', () => { await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( 'not_applicable', ); - expect(mocks.deliverParentEvent).not.toHaveBeenCalled(); + expect(mocks.enqueueParentEventAndWait).not.toHaveBeenCalled(); }); }); diff --git a/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts b/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts index 5020d1e52..1e85fbde1 100644 --- a/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts +++ b/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts @@ -9,10 +9,8 @@ import { } from '@roomote/db/server'; import { Env } from '@roomote/env'; -import { - FastAgentParentEventDeliveryError, - deliverFastAgentParentEvent, -} from '../fast-agent-parent-event'; +import { FastAgentParentEventDeliveryError } from '../fast-agent-parent-event'; +import { enqueueFastAgentParentEventAndWait } from '../fast-agent-parent-event-queue'; import { buildFastAgentDeliveringMarker, buildFastAgentDeliveryClaimPredicate, @@ -27,9 +25,9 @@ export type FastArtifactNotificationResult = | 'skipped' | 'failed'; -/** Fail the turn-lock wait well below the worker's request timeout so the - * caller can 503 and the worker's confirmUpload retry does the waiting. */ -const ARTIFACT_DELIVERY_LOCK_WAIT_MS = 30_000; +/** Bound queue completion below the worker's request timeout so the caller can + * return 503 and let confirmUpload retry while durable delivery continues. */ +const ARTIFACT_DELIVERY_WAIT_TIMEOUT_MS = 30_000; function buildArtifactViewUrl(input: { taskId: string; @@ -107,22 +105,28 @@ export async function notifyFastAgentParentOnArtifact(input: { let delivered = false; try { - await deliverFastAgentParentEvent({ - parent, - event: { - type: 'artifact_published', - taskId: input.taskId, - runId: run.id, - artifact: { - id: input.id, - path: input.path, - version: input.version, - contentType: input.contentType, - viewUrl: buildArtifactViewUrl(input), + const delivery = await enqueueFastAgentParentEventAndWait( + { + parent, + event: { + type: 'artifact_published', + taskId: input.taskId, + runId: run.id, + artifact: { + id: input.id, + path: input.path, + version: input.version, + contentType: input.contentType, + viewUrl: buildArtifactViewUrl(input), + }, }, }, - lockWaitMs: ARTIFACT_DELIVERY_LOCK_WAIT_MS, - }); + { timeoutMs: ARTIFACT_DELIVERY_WAIT_TIMEOUT_MS }, + ); + if (delivery === 'skipped') { + await writeDeliveryMarker('skipped'); + return 'skipped'; + } delivered = true; await writeDeliveryMarker('delivered'); diff --git a/packages/sdk/src/server/lib/fast-agent-parent-event-queue.test.ts b/packages/sdk/src/server/lib/fast-agent-parent-event-queue.test.ts index 3aa002b3e..0e1687ad4 100644 --- a/packages/sdk/src/server/lib/fast-agent-parent-event-queue.test.ts +++ b/packages/sdk/src/server/lib/fast-agent-parent-event-queue.test.ts @@ -124,6 +124,7 @@ import { buildFastAgentParentEventKey, drainFastAgentParentEvents, enqueueFastAgentParentEvent, + enqueueFastAgentParentEventAndWait, enqueueFastAgentParentEventForRun, FastAgentParentBusyError, recoverPendingFastAgentParentEvents, @@ -234,6 +235,63 @@ describe('Fast parent event durable queue', () => { expect(mocks.queueAdd).toHaveBeenCalledOnce(); }); + it('waits for a durably admitted event to reach its delivered state', async () => { + mocks.findPending.mockResolvedValueOnce({ + deliveredAt: new Date(), + discardedAt: null, + }); + + await expect( + enqueueFastAgentParentEventAndWait( + { parent, event }, + { timeoutMs: 30_000 }, + ), + ).resolves.toBe('delivered'); + + expect(mocks.insertOnConflict).toHaveBeenCalledOnce(); + expect(mocks.queueAdd).toHaveBeenCalledOnce(); + }); + + it('reports a durably admitted event discarded by the queue as skipped', async () => { + mocks.findPending.mockResolvedValueOnce({ + deliveredAt: null, + discardedAt: new Date(), + }); + + await expect( + enqueueFastAgentParentEventAndWait( + { parent, event }, + { timeoutMs: 30_000 }, + ), + ).resolves.toBe('skipped'); + }); + + it('times out without withdrawing the durable event', async () => { + vi.useFakeTimers(); + mocks.findPending.mockResolvedValue({ + deliveredAt: null, + discardedAt: null, + }); + + try { + const delivery = enqueueFastAgentParentEventAndWait( + { parent, event }, + { timeoutMs: 100, pollIntervalMs: 25 }, + ); + const rejected = expect(delivery).rejects.toMatchObject({ + message: + 'Timed out waiting for the queued Fast parent event to be delivered.', + replyPosted: false, + }); + await vi.advanceTimersByTimeAsync(100); + + await rejected; + expect(mocks.insertOnConflict).toHaveBeenCalledOnce(); + } finally { + vi.useRealTimers(); + } + }); + it('builds a stable BullMQ-safe idempotency key', () => { const first = buildFastAgentParentEventKey({ parent, event }); expect(buildFastAgentParentEventKey({ parent, event })).toBe(first); diff --git a/packages/sdk/src/server/lib/fast-agent-parent-event-queue.ts b/packages/sdk/src/server/lib/fast-agent-parent-event-queue.ts index 6b4d75693..acb8e8e73 100644 --- a/packages/sdk/src/server/lib/fast-agent-parent-event-queue.ts +++ b/packages/sdk/src/server/lib/fast-agent-parent-event-queue.ts @@ -163,6 +163,44 @@ export async function enqueueFastAgentParentEvent(params: { return { eventKey, queued: true }; } +/** + * Durably admit an event, then wait for the ordered queue drain to settle it. + * Callers that must distinguish presentation from admission can retain their + * bounded retry contract without bypassing the shared parent-event queue. + */ +export async function enqueueFastAgentParentEventAndWait( + params: { + parent: FastAgentParent; + event: FastAgentParentEvent; + retryTaskStartRunId?: number; + }, + options: { timeoutMs: number; pollIntervalMs?: number }, +): Promise<'delivered' | 'skipped'> { + const { eventKey } = await enqueueFastAgentParentEvent(params); + const deadline = Date.now() + options.timeoutMs; + const pollIntervalMs = options.pollIntervalMs ?? 100; + + for (;;) { + const row = await db.query.fastAgentParentEvents.findFirst({ + where: eq(fastAgentParentEvents.eventKey, eventKey), + columns: { deliveredAt: true, discardedAt: true }, + }); + if (!row || row.discardedAt) return 'skipped'; + if (row.deliveredAt) return 'delivered'; + + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0) { + throw new FastAgentParentEventDeliveryError( + 'Timed out waiting for the queued Fast parent event to be delivered.', + { replyPosted: false }, + ); + } + await new Promise((resolve) => + setTimeout(resolve, Math.min(pollIntervalMs, remainingMs)), + ); + } +} + /** Serialize PR-open admission with terminal run updates on the same row. */ export async function enqueueFastAgentParentEventForRun(params: { parent: FastAgentParent; diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts index c6cc8dbe3..1ce41bca4 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts @@ -24,7 +24,7 @@ const mocks = vi.hoisted(() => { findClaimRun: vi.fn(), updateSet: vi.fn(), recordLifecycle: vi.fn(), - deliverParentEvent: vi.fn(), + enqueueParentEventAndWait: vi.fn(), getTaskUrl: vi.fn( ({ taskId }: { taskId: string }) => `https://roomote.example/task/${taskId}`, @@ -73,10 +73,13 @@ vi.mock('@roomote/cloud-agents/server', () => ({ })); vi.mock('../../fast-agent-parent-event', () => ({ - deliverFastAgentParentEvent: mocks.deliverParentEvent, FastAgentParentEventDeliveryError: mocks.FastAgentParentEventDeliveryError, })); +vi.mock('../../fast-agent-parent-event-queue', () => ({ + enqueueFastAgentParentEventAndWait: mocks.enqueueParentEventAndWait, +})); + import { notifyFastAgentParentOnPrFeedback } from '../notify-fast-agent-parent-on-pr-feedback'; const fastParent = { @@ -148,7 +151,7 @@ describe('notifyFastAgentParentOnPrFeedback', () => { runId: 200, }); mocks.findClaimRun.mockResolvedValue({ id: 200 }); - mocks.deliverParentEvent.mockResolvedValue('delivered'); + mocks.enqueueParentEventAndWait.mockResolvedValue('delivered'); mocks.recordLifecycle.mockResolvedValue(undefined); }); @@ -160,21 +163,23 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }), ).resolves.toBe(true); - expect(mocks.deliverParentEvent).toHaveBeenCalledWith({ - parent: fastParent, - lockWaitMs: 30_000, - event: { - type: 'pull_request_feedback', - feedbackId: expect.stringMatching(/^[a-f0-9]{24}$/), - taskId: 'child-task', - runId: 200, - taskUrl: 'https://roomote.example/task/child-task', - pullRequest: input.pullRequest, - summary: input.summary, - suggestedActionQuestion: input.suggestedActionQuestion, - suggestedActionPrompt: input.suggestedActionPrompt, + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledWith( + { + parent: fastParent, + event: { + type: 'pull_request_feedback', + feedbackId: expect.stringMatching(/^[a-f0-9]{24}$/), + taskId: 'child-task', + runId: 200, + taskUrl: 'https://roomote.example/task/child-task', + pullRequest: input.pullRequest, + summary: input.summary, + suggestedActionQuestion: input.suggestedActionQuestion, + suggestedActionPrompt: input.suggestedActionPrompt, + }, }, - }); + { timeoutMs: 30_000 }, + ); expect(mocks.recordLifecycle).toHaveBeenCalledWith( expect.anything(), expect.objectContaining({ @@ -194,7 +199,7 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }), ).resolves.toBe(false); - expect(mocks.deliverParentEvent).not.toHaveBeenCalled(); + expect(mocks.enqueueParentEventAndWait).not.toHaveBeenCalled(); }); it('uses the same feedback identity regardless of source event order', async () => { @@ -211,7 +216,7 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }); expect(claimedFeedbackIds()[1]).toBe(claimedFeedbackIds()[0]); - expect(mocks.deliverParentEvent).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); }); it('uses stable source events instead of generated summary text for fallback identity', async () => { @@ -230,7 +235,7 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }); expect(claimedFeedbackIds()[1]).toBe(claimedFeedbackIds()[0]); - expect(mocks.deliverParentEvent).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); }); it('delivers once across linked tasks sharing a conversation', async () => { @@ -250,7 +255,7 @@ describe('notifyFastAgentParentOnPrFeedback', () => { await notifyFastAgentParentOnPrFeedback({ run: olderRun, ...input }); await notifyFastAgentParentOnPrFeedback({ run: newerRun, ...input }); - expect(mocks.deliverParentEvent).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); expect(mocks.claimConversationDelivery).toHaveBeenCalledTimes(2); }); @@ -272,8 +277,8 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }), ).resolves.toBe(true); - expect(mocks.deliverParentEvent).toHaveBeenCalledOnce(); - expect(mocks.deliverParentEvent).toHaveBeenCalledWith( + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledWith( expect.objectContaining({ event: expect.objectContaining({ taskId: 'newer-task', @@ -281,6 +286,7 @@ describe('notifyFastAgentParentOnPrFeedback', () => { taskUrl: 'https://roomote.example/task/newer-task', }), }), + { timeoutMs: 30_000 }, ); }); @@ -294,7 +300,7 @@ describe('notifyFastAgentParentOnPrFeedback', () => { ...input, }); - expect(mocks.deliverParentEvent).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); expect(mocks.completeConversationDelivery).not.toHaveBeenCalled(); }); @@ -316,7 +322,7 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }); expect(claimedFeedbackIds()[1]).toBe(claimedFeedbackIds()[0]); - expect(mocks.deliverParentEvent).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); }); it('preserves structured terminal review metadata in the Fast event', async () => { @@ -334,16 +340,17 @@ describe('notifyFastAgentParentOnPrFeedback', () => { reviewResult, }); - expect(mocks.deliverParentEvent).toHaveBeenCalledWith( + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledWith( expect.objectContaining({ event: expect.objectContaining({ reviewResult }), }), + { timeoutMs: 30_000 }, ); }); it('does nothing for a task without a Fast parent', async () => { await notifyFastAgentParentOnPrFeedback({ run: makeRun({}), ...input }); - expect(mocks.deliverParentEvent).not.toHaveBeenCalled(); + expect(mocks.enqueueParentEventAndWait).not.toHaveBeenCalled(); }); }); diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts index 8e3091e03..138bb4049 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts @@ -22,6 +22,7 @@ const mocks = vi.hoisted(() => { recordLifecycle: vi.fn(), deliverParentEvent: vi.fn(), enqueueParentEvent: vi.fn(), + enqueueParentEventAndWait: vi.fn(), getTaskUrl: vi.fn(() => 'https://roomote.example/task/child-task'), FastAgentParentEventDeliveryError, }; @@ -69,6 +70,7 @@ vi.mock('../../fast-agent-parent-event', () => ({ vi.mock('../../fast-agent-parent-event-queue', () => ({ enqueueFastAgentParentEvent: mocks.enqueueParentEvent, + enqueueFastAgentParentEventAndWait: mocks.enqueueParentEventAndWait, })); import { notifyFastAgentParentOnPullRequestStatusChanged } from '../notify-fast-agent-parent-on-pull-request-status-changed'; @@ -115,6 +117,7 @@ describe('notifyFastAgentParentOnPullRequestStatusChanged', () => { eventKey: 'pr-status-event', queued: true, }); + mocks.enqueueParentEventAndWait.mockResolvedValue('delivered'); mocks.recordLifecycle.mockResolvedValue(undefined); }); @@ -177,7 +180,7 @@ describe('notifyFastAgentParentOnPullRequestConflict', () => { vi.clearAllMocks(); mocks.claimReturning.mockResolvedValue([{ id: 200 }]); mocks.findClaimRun.mockResolvedValue({ id: 200 }); - mocks.deliverParentEvent.mockResolvedValue('delivered'); + mocks.enqueueParentEventAndWait.mockResolvedValue('delivered'); mocks.recordLifecycle.mockResolvedValue(undefined); }); @@ -197,21 +200,23 @@ describe('notifyFastAgentParentOnPullRequestConflict', () => { }); expect(delivered).toBe(true); - expect(mocks.deliverParentEvent).toHaveBeenCalledWith({ - parent: fastParent, - lockWaitMs: 30_000, - event: expect.objectContaining({ - type: 'pull_request_conflict_detected', - taskId: 'child-task', - runId: 200, - conflictDetectedAt: conflictDetectedAt.toISOString(), - message: - '[Fix review feedback](https://github.com/acme/web/pull/42) now has merge conflicts. Update the branch or ask Roomote to resolve them.', - pullRequest: expect.objectContaining({ - repository: 'acme/web', - number: 42, + expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledWith( + { + parent: fastParent, + event: expect.objectContaining({ + type: 'pull_request_conflict_detected', + taskId: 'child-task', + runId: 200, + conflictDetectedAt: conflictDetectedAt.toISOString(), + message: + '[Fix review feedback](https://github.com/acme/web/pull/42) now has merge conflicts. Update the branch or ask Roomote to resolve them.', + pullRequest: expect.objectContaining({ + repository: 'acme/web', + number: 42, + }), }), - }), - }); + }, + { timeoutMs: 30_000 }, + ); }); }); diff --git a/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pr-feedback.ts b/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pr-feedback.ts index a5e6d2e95..e942d1a1d 100644 --- a/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pr-feedback.ts +++ b/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pr-feedback.ts @@ -13,13 +13,11 @@ import { type SourceControlProvider, } from '@roomote/types'; -import { - deliverFastAgentParentEvent, - type FastAgentPullRequestContext, -} from '../fast-agent-parent-event'; +import { type FastAgentPullRequestContext } from '../fast-agent-parent-event'; +import { enqueueFastAgentParentEventAndWait } from '../fast-agent-parent-event-queue'; import { deliverFastAgentParentPrEvent } from './deliver-fast-agent-parent-pr-event'; -const PR_FEEDBACK_DELIVERY_LOCK_WAIT_MS = 30_000; +const PR_FEEDBACK_DELIVERY_WAIT_TIMEOUT_MS = 30_000; function buildFeedbackId(params: { conversation: { @@ -162,35 +160,39 @@ export async function notifyFastAgentParentOnPrFeedback(params: { }, canonicalDeliveryOwned: params.canonicalDeliveryOwned, deliver: () => - deliverFastAgentParentEvent({ - parent, - event: { - type: 'pull_request_feedback', - feedbackId, - taskId: attributedTaskId, - runId: attributedRunId, - taskUrl: getTaskUrl({ + enqueueFastAgentParentEventAndWait( + { + parent, + event: { + type: 'pull_request_feedback', + feedbackId, taskId: attributedTaskId, - utm: { - source: parent.conversation.surface, - campaign: 'fast-delegation-pr-feedback', - }, - }), - pullRequest, - summary: params.summary, - ...(params.reviewResult ? { reviewResult: params.reviewResult } : {}), - ...(params.suggestedActionQuestion - ? { suggestedActionQuestion: params.suggestedActionQuestion } - : {}), - ...(params.suggestedActionPrompt - ? { suggestedActionPrompt: params.suggestedActionPrompt } - : {}), - ...(params.reviewActionDeliveryId - ? { reviewActionDeliveryId: params.reviewActionDeliveryId } - : {}), + runId: attributedRunId, + taskUrl: getTaskUrl({ + taskId: attributedTaskId, + utm: { + source: parent.conversation.surface, + campaign: 'fast-delegation-pr-feedback', + }, + }), + pullRequest, + summary: params.summary, + ...(params.reviewResult + ? { reviewResult: params.reviewResult } + : {}), + ...(params.suggestedActionQuestion + ? { suggestedActionQuestion: params.suggestedActionQuestion } + : {}), + ...(params.suggestedActionPrompt + ? { suggestedActionPrompt: params.suggestedActionPrompt } + : {}), + ...(params.reviewActionDeliveryId + ? { reviewActionDeliveryId: params.reviewActionDeliveryId } + : {}), + }, }, - lockWaitMs: PR_FEEDBACK_DELIVERY_LOCK_WAIT_MS, - }), + { timeoutMs: PR_FEEDBACK_DELIVERY_WAIT_TIMEOUT_MS }, + ), recordLifecycle: () => recordTaskRunLifecycleEvent(db, { runId: params.run.id, diff --git a/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-conflict.ts b/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-conflict.ts index cc32d0f99..5a205f8b0 100644 --- a/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-conflict.ts +++ b/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-conflict.ts @@ -11,14 +11,12 @@ import { type SourceControlProvider, } from '@roomote/types'; -import { - deliverFastAgentParentEvent, - type FastAgentPullRequestContext, -} from '../fast-agent-parent-event'; +import { type FastAgentPullRequestContext } from '../fast-agent-parent-event'; +import { enqueueFastAgentParentEventAndWait } from '../fast-agent-parent-event-queue'; import { deliverFastAgentParentPrEvent } from './deliver-fast-agent-parent-pr-event'; import { buildPullRequestConflictMessage } from './pull-request-mergeability-check'; -const PR_CONFLICT_DELIVERY_LOCK_WAIT_MS = 30_000; +const PR_CONFLICT_DELIVERY_WAIT_TIMEOUT_MS = 30_000; function buildNotifiedResultKey(params: { prUrl: string; @@ -66,28 +64,30 @@ export async function notifyFastAgentParentOnPullRequestConflict(params: { deliveryKey, logPrefix: 'notifyFastAgentParentOnPullRequestConflict', deliver: () => - deliverFastAgentParentEvent({ - parent, - event: { - type: 'pull_request_conflict_detected', - taskId: params.run.taskId, - runId: params.run.id, - taskUrl: getTaskUrl({ + enqueueFastAgentParentEventAndWait( + { + parent, + event: { + type: 'pull_request_conflict_detected', taskId: params.run.taskId, - utm: { - source: parent.conversation.surface, - campaign: 'fast-delegation-pr-conflict', - }, - }), - pullRequest, - conflictDetectedAt: params.conflictDetectedAt.toISOString(), - message: buildPullRequestConflictMessage({ - title: params.pullRequest.title, - url: params.pullRequest.url, - }), + runId: params.run.id, + taskUrl: getTaskUrl({ + taskId: params.run.taskId, + utm: { + source: parent.conversation.surface, + campaign: 'fast-delegation-pr-conflict', + }, + }), + pullRequest, + conflictDetectedAt: params.conflictDetectedAt.toISOString(), + message: buildPullRequestConflictMessage({ + title: params.pullRequest.title, + url: params.pullRequest.url, + }), + }, }, - lockWaitMs: PR_CONFLICT_DELIVERY_LOCK_WAIT_MS, - }), + { timeoutMs: PR_CONFLICT_DELIVERY_WAIT_TIMEOUT_MS }, + ), recordLifecycle: () => recordTaskRunLifecycleEvent(db, { runId: params.run.id, From 8a75ee92fbb7e0c348194414ed8304b865112fd3 Mon Sep 17 00:00:00 2001 From: "@daniel-lxs" <57051444+daniel-lxs@users.noreply.github.com> Date: Thu, 3 Sep 2026 19:08:13 +0000 Subject: [PATCH 2/6] fix: treat parent notification admission as handoff --- .../__tests__/upload-complete.test.ts | 8 +- .../src/handlers/artifacts/upload-complete.ts | 10 - .../src/jobs/pr-review-notification.test.ts | 1 - .../bullmq/src/jobs/pr-review-notification.ts | 9 +- .../jobs/pull-request-mergeability-check.ts | 9 +- .../notify-fast-agent-parent.test.ts | 171 ++--------- .../lib/artifacts/notify-fast-agent-parent.ts | 156 +++------- .../lib/fast-agent-parent-event-queue.test.ts | 58 ---- .../lib/fast-agent-parent-event-queue.ts | 38 --- ...deliver-fast-agent-parent-pr-event.test.ts | 285 ------------------ ...y-fast-agent-parent-on-pr-feedback.test.ts | 205 +++++-------- ...ent-on-pull-request-status-changed.test.ts | 152 +++++----- .../deliver-fast-agent-parent-pr-event.ts | 192 ------------ ...notify-fast-agent-parent-on-pr-feedback.ts | 110 +++---- ...t-agent-parent-on-pull-request-conflict.ts | 103 +++---- ...t-parent-on-pull-request-status-changed.ts | 96 +++--- 16 files changed, 349 insertions(+), 1254 deletions(-) delete mode 100644 packages/sdk/src/server/lib/task-runs/__tests__/deliver-fast-agent-parent-pr-event.test.ts delete mode 100644 packages/sdk/src/server/lib/task-runs/deliver-fast-agent-parent-pr-event.ts diff --git a/apps/api/src/handlers/artifacts/__tests__/upload-complete.test.ts b/apps/api/src/handlers/artifacts/__tests__/upload-complete.test.ts index f2dd5ee75..825708792 100644 --- a/apps/api/src/handlers/artifacts/__tests__/upload-complete.test.ts +++ b/apps/api/src/handlers/artifacts/__tests__/upload-complete.test.ts @@ -56,7 +56,7 @@ describe('markArtifactUploadComplete', () => { version: 1, uploaded: false, }); - mocks.notifyParent.mockResolvedValue('delivered'); + mocks.notifyParent.mockResolvedValue('queued'); }); it('notifies the Fast parent immediately after upload publication', async () => { @@ -73,11 +73,7 @@ describe('markArtifactUploadComplete', () => { }); }); - it('replays publication through the idempotent notifier', async () => { - mocks.notifyParent - .mockResolvedValueOnce('delivered') - .mockResolvedValueOnce('already_delivered'); - + it('replays publication through idempotent durable admission', async () => { expect((await markArtifactUploadComplete(context())).status).toBe(200); expect((await markArtifactUploadComplete(context())).status).toBe(200); expect(mocks.notifyParent).toHaveBeenCalledTimes(2); diff --git a/apps/api/src/handlers/artifacts/upload-complete.ts b/apps/api/src/handlers/artifacts/upload-complete.ts index 725d63595..14ca9e3dd 100644 --- a/apps/api/src/handlers/artifacts/upload-complete.ts +++ b/apps/api/src/handlers/artifacts/upload-complete.ts @@ -68,15 +68,5 @@ export async function markArtifactUploadComplete( 503, ); } - if (notification === 'in_progress') { - // Another request is mid-delivery; 503 keeps the worker retrying until - // that delivery settles instead of reporting success while it can still - // fail and release its claim. - return c.json( - { error: 'Artifact published; parent notification is in progress' }, - 503, - ); - } - return new Response(null, { status: 200 }); } diff --git a/apps/bullmq/src/jobs/pr-review-notification.test.ts b/apps/bullmq/src/jobs/pr-review-notification.test.ts index 614ec7c25..0972f701e 100644 --- a/apps/bullmq/src/jobs/pr-review-notification.test.ts +++ b/apps/bullmq/src/jobs/pr-review-notification.test.ts @@ -520,7 +520,6 @@ describe('prReviewNotificationJob', () => { ); expect(mockNotifyFastAgentParent).toHaveBeenCalledWith({ - canonicalDeliveryOwned: false, run: expect.objectContaining({ id: 1, taskId: 'task-1' }), feedbackSourceIds: [expect.any(String)], pullRequest: { diff --git a/apps/bullmq/src/jobs/pr-review-notification.ts b/apps/bullmq/src/jobs/pr-review-notification.ts index 5726cdd4e..f43047a3f 100644 --- a/apps/bullmq/src/jobs/pr-review-notification.ts +++ b/apps/bullmq/src/jobs/pr-review-notification.ts @@ -784,19 +784,18 @@ export const prReviewNotificationJob = async ( suggestedActionPrompt: followUp.prompt, } : {}), - canonicalDeliveryOwned: data.ownershipVersion === 'canonical', ...(options.reviewActionDeliveryId ? { reviewActionDeliveryId: options.reviewActionDeliveryId } : {}), }); - const deliveredToFastParent = await notifyFastParent({ + const admittedToFastParent = await notifyFastParent({ includeSuggestedAction: Boolean(followUp && !autoHandleUserId), ...(webReviewActionDeliveryId ? { reviewActionDeliveryId: webReviewActionDeliveryId } : {}), }); - if (deliveredToFastParent && webReviewActionDeliveryId) { + if (admittedToFastParent && webReviewActionDeliveryId) { const { attached } = await attachPendingPrReviewActionMessageWithRetirement( webReviewActionDeliveryId, @@ -818,7 +817,7 @@ export const prReviewNotificationJob = async ( let autoHandledText: string | null = null; const ownsAutoHandleDispatch = - directAutoHandleRoute !== null || deliveredToFastParent; + directAutoHandleRoute !== null || admittedToFastParent; if ( followUp && autoHandlePreference && @@ -953,7 +952,7 @@ ${delivery.text}`; } } - if (deliveredToFastParent && (!autoHandleUserId || autoHandledText)) { + if (admittedToFastParent && (!autoHandleUserId || autoHandledText)) { await recordPrReviewNotificationDeliveryBestEffort({ runId: latestJob.id, taskId: data.taskId, diff --git a/apps/bullmq/src/jobs/pull-request-mergeability-check.ts b/apps/bullmq/src/jobs/pull-request-mergeability-check.ts index ae4565b34..f3f96e746 100644 --- a/apps/bullmq/src/jobs/pull-request-mergeability-check.ts +++ b/apps/bullmq/src/jobs/pull-request-mergeability-check.ts @@ -102,8 +102,8 @@ async function postConflictNotification(params: { const title = params.candidate.prTitle ?? `Pull request #${params.candidate.prNumber}`; - const deliveredToFastParent = - await notifyFastAgentParentOnPullRequestConflict({ + const admittedToFastParent = await notifyFastAgentParentOnPullRequestConflict( + { run: latestRun, pullRequest: { provider: 'github', @@ -114,8 +114,9 @@ async function postConflictNotification(params: { url: params.candidate.prUrl, }, conflictDetectedAt: params.conflictDetectedAt, - }); - if (deliveredToFastParent) return true; + }, + ); + if (admittedToFastParent) return true; const route = await resolvePrReviewNotificationRoute(latestRun); const text = buildCandidateConflictText(params.candidate); diff --git a/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts b/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts index c7feec0c1..f7c4014c2 100644 --- a/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts +++ b/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts @@ -1,47 +1,18 @@ const mocks = vi.hoisted(() => { - class FastAgentParentEventDeliveryError extends Error { - readonly replyPosted: boolean; - readonly permanent: boolean; - - constructor( - message: string, - options: { replyPosted: boolean; permanent?: boolean }, - ) { - super(message); - this.replyPosted = options.replyPosted; - this.permanent = options.permanent ?? false; - } - } - return { findRun: vi.fn(), - claimReturning: vi.fn(), - updateSet: vi.fn(), recordLifecycle: vi.fn(), - enqueueParentEventAndWait: vi.fn(), - FastAgentParentEventDeliveryError, + enqueueParentEvent: vi.fn(), }; }); vi.mock('@roomote/db/server', () => ({ db: { query: { taskRuns: { findFirst: mocks.findRun } }, - update: vi.fn(() => ({ - set: vi.fn((values: unknown) => { - mocks.updateSet(values); - return { - where: vi.fn(() => ({ returning: mocks.claimReturning })), - }; - }), - })), }, and: vi.fn((...args: unknown[]) => args), eq: vi.fn((...args: unknown[]) => args), recordTaskRunLifecycleEvent: mocks.recordLifecycle, - sql: vi.fn((strings: TemplateStringsArray, ...values: unknown[]) => ({ - strings: [...strings], - values, - })), taskRuns: { id: 'task_runs.id', taskId: 'task_runs.task_id', @@ -53,12 +24,8 @@ vi.mock('@roomote/env', () => ({ Env: { R_APP_URL: 'https://roomote.example' }, })); -vi.mock('../../fast-agent-parent-event', () => ({ - FastAgentParentEventDeliveryError: mocks.FastAgentParentEventDeliveryError, -})); - vi.mock('../../fast-agent-parent-event-queue', () => ({ - enqueueFastAgentParentEventAndWait: mocks.enqueueParentEventAndWait, + enqueueFastAgentParentEvent: mocks.enqueueParentEvent, })); import { notifyFastAgentParentOnArtifact } from '../notify-fast-agent-parent'; @@ -99,34 +66,33 @@ describe('notifyFastAgentParentOnArtifact', () => { payload: { fastAgentParent: fastParent }, result: {}, }); - mocks.claimReturning.mockResolvedValue([{ id: 200 }]); - mocks.enqueueParentEventAndWait.mockResolvedValue('delivered'); + mocks.enqueueParentEvent.mockResolvedValue({ + eventKey: 'artifact-event', + queued: true, + }); mocks.recordLifecycle.mockResolvedValue(undefined); }); it('passes structured artifact metadata to the Fast orchestrator', async () => { await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( - 'delivered', + 'queued', ); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledWith( - { - parent: fastParent, - event: expect.objectContaining({ - type: 'artifact_published', - taskId: 'child-task', - runId: 200, - artifact: expect.objectContaining({ - id: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', - path: 'proof/result.png', - contentType: 'image/png', - viewUrl: - 'https://roomote.example/task/child-task/artifacts/proof/result.png?v=1', - }), + expect(mocks.enqueueParentEvent).toHaveBeenCalledWith({ + parent: fastParent, + event: expect.objectContaining({ + type: 'artifact_published', + taskId: 'child-task', + runId: 200, + artifact: expect.objectContaining({ + id: 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa', + path: 'proof/result.png', + contentType: 'image/png', + viewUrl: + 'https://roomote.example/task/child-task/artifacts/proof/result.png?v=1', }), - }, - { timeoutMs: 30_000 }, - ); + }), + }); expect(mocks.recordLifecycle).toHaveBeenCalledWith( expect.anything(), expect.objectContaining({ @@ -137,97 +103,14 @@ describe('notifyFastAgentParentOnArtifact', () => { ); }); - it('deduplicates an event already claimed by another delivery', async () => { - mocks.claimReturning.mockResolvedValueOnce([]); - - await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( - 'already_delivered', - ); - expect(mocks.enqueueParentEventAndWait).not.toHaveBeenCalled(); - }); - - it('releases a failed orchestrator delivery for retry', async () => { - mocks.enqueueParentEventAndWait.mockRejectedValueOnce( - new Error('model offline'), + it('reports a durable enqueue failure for retry', async () => { + mocks.enqueueParentEvent.mockRejectedValueOnce( + new Error('database offline'), ); await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( 'failed', ); - expect( - mocks.updateSet.mock.calls.some(([values]) => { - const result = (values as { result?: { strings?: string[] } }).result; - return result?.strings?.join('').includes(' - ') === true; - }), - ).toBe(true); - }); - - it('reports an in-flight delivery as in_progress instead of delivered', async () => { - mocks.claimReturning.mockResolvedValueOnce([]); - mocks.findRun.mockResolvedValue({ - id: 200, - taskId: 'child-task', - payload: { fastAgentParent: fastParent }, - result: { - 'fastAgentArtifact:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa': `delivering:${Date.now()}`, - }, - }); - - await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( - 'in_progress', - ); - expect(mocks.enqueueParentEventAndWait).not.toHaveBeenCalled(); - }); - - it('keeps the claim when the failure happened after the Slack post', async () => { - mocks.enqueueParentEventAndWait.mockRejectedValueOnce( - new mocks.FastAgentParentEventDeliveryError('lifecycle write failed', { - replyPosted: true, - }), - ); - - await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( - 'delivered', - ); - expect( - mocks.updateSet.mock.calls.some(([values]) => { - const result = (values as { result?: { strings?: string[] } }).result; - return result?.strings?.join('').includes(' - ') === true; - }), - ).toBe(false); - }); - - it('settles the claim as skipped when no retry can ever succeed', async () => { - mocks.enqueueParentEventAndWait.mockRejectedValueOnce( - new mocks.FastAgentParentEventDeliveryError('parent session gone', { - replyPosted: false, - permanent: true, - }), - ); - - await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( - 'skipped', - ); - expect( - mocks.updateSet.mock.calls.some(([values]) => { - const result = (values as { result?: { values?: unknown[] } }).result; - return result?.values?.includes('skipped') === true; - }), - ).toBe(true); - }); - - it('settles the claim when the durable queue discards the event', async () => { - mocks.enqueueParentEventAndWait.mockResolvedValueOnce('skipped'); - - await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( - 'skipped', - ); - expect( - mocks.updateSet.mock.calls.some(([values]) => { - const result = (values as { result?: { values?: unknown[] } }).result; - return result?.values?.includes('skipped') === true; - }), - ).toBe(true); expect(mocks.recordLifecycle).not.toHaveBeenCalled(); }); @@ -244,9 +127,9 @@ describe('notifyFastAgentParentOnArtifact', () => { }); await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( - 'delivered', + 'queued', ); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEvent).toHaveBeenCalledOnce(); }); it('does nothing for standalone artifacts', async () => { @@ -260,6 +143,6 @@ describe('notifyFastAgentParentOnArtifact', () => { await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( 'not_applicable', ); - expect(mocks.enqueueParentEventAndWait).not.toHaveBeenCalled(); + expect(mocks.enqueueParentEvent).not.toHaveBeenCalled(); }); }); diff --git a/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts b/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts index 1e85fbde1..1fe112243 100644 --- a/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts +++ b/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts @@ -4,31 +4,17 @@ import { db, eq, recordTaskRunLifecycleEvent, - sql, taskRuns, } from '@roomote/db/server'; import { Env } from '@roomote/env'; -import { FastAgentParentEventDeliveryError } from '../fast-agent-parent-event'; -import { enqueueFastAgentParentEventAndWait } from '../fast-agent-parent-event-queue'; -import { - buildFastAgentDeliveringMarker, - buildFastAgentDeliveryClaimPredicate, - isFastAgentDeliveringMarker, -} from '../task-runs/fast-agent-delivery-claim'; +import { enqueueFastAgentParentEvent } from '../fast-agent-parent-event-queue'; export type FastArtifactNotificationResult = | 'not_applicable' - | 'already_delivered' - | 'in_progress' - | 'delivered' - | 'skipped' + | 'queued' | 'failed'; -/** Bound queue completion below the worker's request timeout so the caller can - * return 503 and let confirmUpload retry while durable delivery continues. */ -const ARTIFACT_DELIVERY_WAIT_TIMEOUT_MS = 30_000; - function buildArtifactViewUrl(input: { taskId: string; path: string; @@ -58,127 +44,57 @@ export async function notifyFastAgentParentOnArtifact(input: { const run = await db.query.taskRuns.findFirst({ where: and(eq(taskRuns.id, input.runId), eq(taskRuns.taskId, input.taskId)), - columns: { id: true, taskId: true, payload: true, result: true }, + columns: { id: true, taskId: true, payload: true }, }); const parent = getFastAgentParentFromPayload(run?.payload); if (!run || !parent) { return 'not_applicable'; } - const deliveryKey = `fastAgentArtifact:${input.id}`; - const writeDeliveryMarker = async (marker: string) => { - await db - .update(taskRuns) - .set({ - result: sql`coalesce(${taskRuns.result}, '{}'::jsonb) || jsonb_build_object(${deliveryKey}::text, ${marker}::text)`, - }) - .where(eq(taskRuns.id, run.id)); - }; - const claimed = await db - .update(taskRuns) - .set({ - result: sql`coalesce(${taskRuns.result}, '{}'::jsonb) || jsonb_build_object(${deliveryKey}::text, ${buildFastAgentDeliveringMarker()}::text)`, - }) - .where( - and( - eq(taskRuns.id, run.id), - buildFastAgentDeliveryClaimPredicate(deliveryKey), - ), - ) - .returning({ id: taskRuns.id }); - - if (claimed.length === 0) { - // Distinguish a live in-flight delivery (the caller should keep - // retrying) from a settled one (the caller must stop). - const current = await db.query.taskRuns.findFirst({ - where: eq(taskRuns.id, run.id), - columns: { result: true }, - }); - const marker = (current?.result as Record | null)?.[ - deliveryKey - ]; - return isFastAgentDeliveringMarker(marker) - ? 'in_progress' - : 'already_delivered'; - } - - let delivered = false; - try { - const delivery = await enqueueFastAgentParentEventAndWait( - { - parent, - event: { - type: 'artifact_published', - taskId: input.taskId, - runId: run.id, - artifact: { - id: input.id, - path: input.path, - version: input.version, - contentType: input.contentType, - viewUrl: buildArtifactViewUrl(input), - }, + await enqueueFastAgentParentEvent({ + parent, + event: { + type: 'artifact_published', + taskId: input.taskId, + runId: run.id, + artifact: { + id: input.id, + path: input.path, + version: input.version, + contentType: input.contentType, + viewUrl: buildArtifactViewUrl(input), }, }, - { timeoutMs: ARTIFACT_DELIVERY_WAIT_TIMEOUT_MS }, - ); - if (delivery === 'skipped') { - await writeDeliveryMarker('skipped'); - return 'skipped'; - } - delivered = true; - - await writeDeliveryMarker('delivered'); - - await recordTaskRunLifecycleEvent(db, { - runId: run.id, - taskId: run.taskId, - eventType: 'decision', - message: `Passed artifact ${input.id} version ${input.version} to the Fast parent orchestrator.`, - details: { - reason: 'fast_agent_parent_artifact_event', - artifactId: input.id, - artifactPath: input.path, - artifactVersion: input.version, - fastAgentSessionId: parent.sessionId, - }, }); - return 'delivered'; + try { + await recordTaskRunLifecycleEvent(db, { + runId: run.id, + taskId: run.taskId, + eventType: 'decision', + message: `Queued artifact ${input.id} version ${input.version} for the Fast parent orchestrator.`, + details: { + reason: 'fast_agent_parent_artifact_event', + artifactId: input.id, + artifactPath: input.path, + artifactVersion: input.version, + fastAgentSessionId: parent.sessionId, + }, + }); + } catch (error) { + console.error( + `[notifyFastAgentParentOnArtifact] Failed to record queue admission for artifact ${input.id}: ${error instanceof Error ? error.message : String(error)}`, + ); + } + + return 'queued'; } catch (error) { console.error( `[notifyFastAgentParentOnArtifact] Failed for artifact ${input.id}: ${ error instanceof Error ? error.message : String(error) }`, ); - const deliveryError = - error instanceof FastAgentParentEventDeliveryError ? error : null; - - if (delivered || deliveryError?.replyPosted) { - // The parent thread already saw the event; releasing the claim would - // make a retry double-post. Settle the marker best-effort instead. - await writeDeliveryMarker('delivered').catch(() => {}); - return 'delivered'; - } - - if (deliveryError?.permanent) { - // No retry can succeed (parent session or installation gone). Settle - // the key so the upload confirmation is not stuck returning 503. - await writeDeliveryMarker('skipped').catch(() => {}); - return 'skipped'; - } - - try { - await db - .update(taskRuns) - .set({ - result: sql`coalesce(${taskRuns.result}, '{}'::jsonb) - ${deliveryKey}`, - }) - .where(eq(taskRuns.id, run.id)); - } catch { - // Best-effort claim release for retry. - } return 'failed'; } } diff --git a/packages/sdk/src/server/lib/fast-agent-parent-event-queue.test.ts b/packages/sdk/src/server/lib/fast-agent-parent-event-queue.test.ts index 0e1687ad4..3aa002b3e 100644 --- a/packages/sdk/src/server/lib/fast-agent-parent-event-queue.test.ts +++ b/packages/sdk/src/server/lib/fast-agent-parent-event-queue.test.ts @@ -124,7 +124,6 @@ import { buildFastAgentParentEventKey, drainFastAgentParentEvents, enqueueFastAgentParentEvent, - enqueueFastAgentParentEventAndWait, enqueueFastAgentParentEventForRun, FastAgentParentBusyError, recoverPendingFastAgentParentEvents, @@ -235,63 +234,6 @@ describe('Fast parent event durable queue', () => { expect(mocks.queueAdd).toHaveBeenCalledOnce(); }); - it('waits for a durably admitted event to reach its delivered state', async () => { - mocks.findPending.mockResolvedValueOnce({ - deliveredAt: new Date(), - discardedAt: null, - }); - - await expect( - enqueueFastAgentParentEventAndWait( - { parent, event }, - { timeoutMs: 30_000 }, - ), - ).resolves.toBe('delivered'); - - expect(mocks.insertOnConflict).toHaveBeenCalledOnce(); - expect(mocks.queueAdd).toHaveBeenCalledOnce(); - }); - - it('reports a durably admitted event discarded by the queue as skipped', async () => { - mocks.findPending.mockResolvedValueOnce({ - deliveredAt: null, - discardedAt: new Date(), - }); - - await expect( - enqueueFastAgentParentEventAndWait( - { parent, event }, - { timeoutMs: 30_000 }, - ), - ).resolves.toBe('skipped'); - }); - - it('times out without withdrawing the durable event', async () => { - vi.useFakeTimers(); - mocks.findPending.mockResolvedValue({ - deliveredAt: null, - discardedAt: null, - }); - - try { - const delivery = enqueueFastAgentParentEventAndWait( - { parent, event }, - { timeoutMs: 100, pollIntervalMs: 25 }, - ); - const rejected = expect(delivery).rejects.toMatchObject({ - message: - 'Timed out waiting for the queued Fast parent event to be delivered.', - replyPosted: false, - }); - await vi.advanceTimersByTimeAsync(100); - - await rejected; - expect(mocks.insertOnConflict).toHaveBeenCalledOnce(); - } finally { - vi.useRealTimers(); - } - }); - it('builds a stable BullMQ-safe idempotency key', () => { const first = buildFastAgentParentEventKey({ parent, event }); expect(buildFastAgentParentEventKey({ parent, event })).toBe(first); diff --git a/packages/sdk/src/server/lib/fast-agent-parent-event-queue.ts b/packages/sdk/src/server/lib/fast-agent-parent-event-queue.ts index 82b108ef9..8927a6a51 100644 --- a/packages/sdk/src/server/lib/fast-agent-parent-event-queue.ts +++ b/packages/sdk/src/server/lib/fast-agent-parent-event-queue.ts @@ -163,44 +163,6 @@ export async function enqueueFastAgentParentEvent(params: { return { eventKey, queued: true }; } -/** - * Durably admit an event, then wait for the ordered queue drain to settle it. - * Callers that must distinguish presentation from admission can retain their - * bounded retry contract without bypassing the shared parent-event queue. - */ -export async function enqueueFastAgentParentEventAndWait( - params: { - parent: FastAgentParent; - event: FastAgentParentEvent; - retryTaskStartRunId?: number; - }, - options: { timeoutMs: number; pollIntervalMs?: number }, -): Promise<'delivered' | 'skipped'> { - const { eventKey } = await enqueueFastAgentParentEvent(params); - const deadline = Date.now() + options.timeoutMs; - const pollIntervalMs = options.pollIntervalMs ?? 100; - - for (;;) { - const row = await db.query.fastAgentParentEvents.findFirst({ - where: eq(fastAgentParentEvents.eventKey, eventKey), - columns: { deliveredAt: true, discardedAt: true }, - }); - if (!row || row.discardedAt) return 'skipped'; - if (row.deliveredAt) return 'delivered'; - - const remainingMs = deadline - Date.now(); - if (remainingMs <= 0) { - throw new FastAgentParentEventDeliveryError( - 'Timed out waiting for the queued Fast parent event to be delivered.', - { replyPosted: false }, - ); - } - await new Promise((resolve) => - setTimeout(resolve, Math.min(pollIntervalMs, remainingMs)), - ); - } -} - /** Serialize PR-open admission with terminal run updates on the same row. */ export async function enqueueFastAgentParentEventForRun(params: { parent: FastAgentParent; diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/deliver-fast-agent-parent-pr-event.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/deliver-fast-agent-parent-pr-event.test.ts deleted file mode 100644 index b8f347f00..000000000 --- a/packages/sdk/src/server/lib/task-runs/__tests__/deliver-fast-agent-parent-pr-event.test.ts +++ /dev/null @@ -1,285 +0,0 @@ -const mocks = vi.hoisted(() => { - class FastAgentParentEventDeliveryError extends Error { - readonly replyPosted: boolean; - readonly permanent: boolean; - - constructor( - message: string, - options: { replyPosted: boolean; permanent?: boolean }, - ) { - super(message); - this.replyPosted = options.replyPosted; - this.permanent = options.permanent ?? false; - } - } - - return { - claimReturning: vi.fn(), - claimConversationDelivery: vi.fn(), - completeConversationDelivery: vi.fn(), - releaseConversationDelivery: vi.fn(), - findClaimRun: vi.fn(), - updateSet: vi.fn(), - sql: vi.fn((strings: TemplateStringsArray, ...values: unknown[]) => ({ - strings: [...strings], - values, - })), - FastAgentParentEventDeliveryError, - }; -}); - -vi.mock('@roomote/db/server', () => ({ - db: { - query: { - taskRuns: { findFirst: mocks.findClaimRun }, - }, - update: vi.fn(() => ({ - set: vi.fn((values: unknown) => { - mocks.updateSet(values); - return { - where: vi.fn(() => ({ returning: mocks.claimReturning })), - }; - }), - })), - }, - claimFastAgentPrFeedbackDelivery: mocks.claimConversationDelivery, - completeFastAgentPrFeedbackDelivery: mocks.completeConversationDelivery, - releaseFastAgentPrFeedbackDelivery: mocks.releaseConversationDelivery, - and: vi.fn((...args: unknown[]) => args), - asc: vi.fn((value: unknown) => value), - desc: vi.fn((value: unknown) => value), - eq: vi.fn((...args: unknown[]) => args), - sql: mocks.sql, - taskRuns: { - id: 'task_runs.id', - taskId: 'task_runs.task_id', - createdAt: 'task_runs.created_at', - result: 'task_runs.result', - }, -})); - -vi.mock('../../fast-agent-parent-event', () => ({ - FastAgentParentEventDeliveryError: mocks.FastAgentParentEventDeliveryError, -})); - -import { deliverFastAgentParentPrEvent } from '../deliver-fast-agent-parent-pr-event'; - -const run = { id: 200, taskId: 'child-task' }; -const deliveryKey = 'fastAgentParentPr:test'; - -function deliver(params?: { - deliver?: () => Promise<'delivered' | 'skipped'>; - recordLifecycle?: () => Promise; -}) { - return deliverFastAgentParentPrEvent({ - run, - deliveryKey, - logPrefix: 'testFastParentPrEvent', - deliver: params?.deliver ?? vi.fn().mockResolvedValue('delivered'), - recordLifecycle: - params?.recordLifecycle ?? vi.fn().mockResolvedValue(undefined), - }); -} - -function hasResultSql(fragment: string): boolean { - return mocks.updateSet.mock.calls.some(([values]) => { - const result = (values as { result?: { strings?: string[] } }).result; - return result?.strings?.join('').includes(fragment) === true; - }); -} - -describe('deliverFastAgentParentPrEvent', () => { - beforeEach(() => { - vi.clearAllMocks(); - vi.spyOn(console, 'error').mockImplementation(() => {}); - mocks.claimReturning.mockResolvedValue([{ id: run.id }]); - mocks.findClaimRun.mockResolvedValue({ id: run.id }); - mocks.claimConversationDelivery.mockResolvedValue({ - id: 'conversation-claim', - leaseToken: 'lease-token', - }); - mocks.completeConversationDelivery.mockResolvedValue(undefined); - mocks.releaseConversationDelivery.mockResolvedValue(undefined); - }); - - it('claims the event with stale-lease recovery before delivering it', async () => { - vi.useFakeTimers(); - vi.setSystemTime(new Date('2026-08-21T12:00:00Z')); - const deliverEvent = vi.fn().mockResolvedValue('delivered'); - const recordLifecycle = vi.fn().mockResolvedValue(undefined); - - await deliver({ deliver: deliverEvent, recordLifecycle }); - - expect(deliverEvent).toHaveBeenCalledOnce(); - expect(recordLifecycle).toHaveBeenCalledOnce(); - expect(hasResultSql('to_jsonb(now())')).toBe(true); - expect( - mocks.sql.mock.calls.some(([, ...values]) => - values.includes(Date.now() - 15 * 60 * 1000), - ), - ).toBe(true); - vi.useRealTimers(); - }); - - it('does not deliver when another caller owns or settled the claim', async () => { - mocks.claimReturning.mockResolvedValue([]); - const deliverEvent = vi.fn().mockResolvedValue('delivered'); - - await deliver({ deliver: deliverEvent }); - - expect(deliverEvent).not.toHaveBeenCalled(); - }); - - it('uses a conversation-scoped claim when feedback can arrive through multiple tasks', async () => { - const deliverEvent = vi.fn().mockResolvedValue('delivered'); - mocks.claimConversationDelivery - .mockResolvedValueOnce({ - status: 'claimed', - claim: { id: 'conversation-claim', leaseToken: 'lease-token' }, - }) - .mockResolvedValueOnce({ status: 'already_claimed' }); - const conversationClaim = { - conversation: { - surface: 'slack' as const, - workspaceId: 'T123', - conversationId: '100.001', - }, - feedbackId: 'feedback-1', - }; - - await deliverFastAgentParentPrEvent({ - run, - deliveryKey, - logPrefix: 'testFastParentPrEvent', - conversationClaim, - deliver: deliverEvent, - recordLifecycle: vi.fn().mockResolvedValue(undefined), - }); - await deliverFastAgentParentPrEvent({ - run: { id: 201, taskId: 'sibling-task' }, - deliveryKey, - logPrefix: 'testFastParentPrEvent', - conversationClaim, - deliver: deliverEvent, - recordLifecycle: vi.fn().mockResolvedValue(undefined), - }); - - expect(mocks.claimConversationDelivery).toHaveBeenNthCalledWith(1, { - conversation: { - surface: 'slack', - workspaceId: 'T123', - conversationId: '100.001', - }, - feedbackId: 'feedback-1', - taskId: 'child-task', - }); - expect(mocks.claimConversationDelivery).toHaveBeenNthCalledWith(2, { - conversation: { - surface: 'slack', - workspaceId: 'T123', - conversationId: '100.001', - }, - feedbackId: 'feedback-1', - taskId: 'sibling-task', - }); - expect(deliverEvent).toHaveBeenCalledOnce(); - expect(mocks.completeConversationDelivery).toHaveBeenCalledOnce(); - // The conversation claim is the sole arbiter, so the task-scoped claim - // predicate must never gate it. - expect(mocks.claimReturning).not.toHaveBeenCalled(); - }); - - it('falls back to the task-scoped claim when the conversation row is missing', async () => { - const deliverEvent = vi.fn().mockResolvedValue('delivered'); - mocks.claimConversationDelivery.mockResolvedValue({ - status: 'no_conversation', - }); - - await deliverFastAgentParentPrEvent({ - run, - deliveryKey, - logPrefix: 'testFastParentPrEvent', - conversationClaim: { - conversation: { - surface: 'slack' as const, - workspaceId: 'T123', - conversationId: '100.001', - }, - feedbackId: 'feedback-1', - }, - deliver: deliverEvent, - recordLifecycle: vi.fn().mockResolvedValue(undefined), - }); - - expect(deliverEvent).toHaveBeenCalledOnce(); - expect(mocks.completeConversationDelivery).not.toHaveBeenCalled(); - expect(mocks.findClaimRun).toHaveBeenCalled(); - }); - - it('stores the claim on the canonical task row across resumed runs', async () => { - mocks.findClaimRun.mockResolvedValue({ id: 100 }); - - await deliver(); - - expect(mocks.findClaimRun).toHaveBeenCalledWith( - expect.objectContaining({ columns: { id: true } }), - ); - }); - - it('settles a skipped delivery without recording lifecycle history', async () => { - const recordLifecycle = vi.fn().mockResolvedValue(undefined); - - await deliver({ - deliver: vi.fn().mockResolvedValue('skipped'), - recordLifecycle, - }); - - expect(recordLifecycle).not.toHaveBeenCalled(); - expect(hasResultSql('to_jsonb(now())')).toBe(true); - }); - - it('releases a transient failure so a later caller can retry', async () => { - await expect( - deliver({ - deliver: vi.fn().mockRejectedValue(new Error('model offline')), - }), - ).rejects.toThrow('model offline'); - - expect(hasResultSql(' - ')).toBe(true); - }); - - it.each([ - { replyPosted: true, permanent: false }, - { replyPosted: false, permanent: true }, - ])( - 'settles a non-retryable delivery error: %o', - async ({ replyPosted, permanent }) => { - await expect( - deliver({ - deliver: vi.fn().mockRejectedValue( - new mocks.FastAgentParentEventDeliveryError('delivery failed', { - replyPosted, - permanent, - }), - ), - }), - ).resolves.toBeUndefined(); - - expect(hasResultSql('to_jsonb(now())')).toBe(true); - expect(hasResultSql(' - ')).toBe(false); - }, - ); - - it('does not retry after delivery succeeds but lifecycle recording fails', async () => { - await expect( - deliver({ - recordLifecycle: vi - .fn() - .mockRejectedValue(new Error('lifecycle unavailable')), - }), - ).resolves.toBeUndefined(); - - expect(hasResultSql('to_jsonb(now())')).toBe(true); - expect(hasResultSql(' - ')).toBe(false); - }); -}); diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts index 1ce41bca4..471d315c1 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts @@ -1,83 +1,29 @@ import type { TaskRun } from '@roomote/db/server'; const mocks = vi.hoisted(() => { - class FastAgentParentEventDeliveryError extends Error { - readonly replyPosted: boolean; - readonly permanent: boolean; - - constructor( - message: string, - options: { replyPosted: boolean; permanent?: boolean }, - ) { - super(message); - this.replyPosted = options.replyPosted; - this.permanent = options.permanent ?? false; - } - } - return { - claimReturning: vi.fn(), - claimConversationDelivery: vi.fn(), - completeConversationDelivery: vi.fn(), - releaseConversationDelivery: vi.fn(), findReusableOwner: vi.fn(), - findClaimRun: vi.fn(), - updateSet: vi.fn(), recordLifecycle: vi.fn(), - enqueueParentEventAndWait: vi.fn(), + enqueueParentEvent: vi.fn(), getTaskUrl: vi.fn( ({ taskId }: { taskId: string }) => `https://roomote.example/task/${taskId}`, ), - FastAgentParentEventDeliveryError, }; }); vi.mock('@roomote/db/server', () => ({ - db: { - query: { - taskRuns: { findFirst: mocks.findClaimRun }, - }, - update: vi.fn(() => ({ - set: vi.fn((values: unknown) => { - mocks.updateSet(values); - return { - where: vi.fn(() => ({ returning: mocks.claimReturning })), - }; - }), - })), - }, - claimFastAgentPrFeedbackDelivery: mocks.claimConversationDelivery, - completeFastAgentPrFeedbackDelivery: mocks.completeConversationDelivery, - releaseFastAgentPrFeedbackDelivery: mocks.releaseConversationDelivery, + db: {}, findReusableGitHubPrFollowUpOwner: mocks.findReusableOwner, - and: vi.fn((...args: unknown[]) => args), - asc: vi.fn((value: unknown) => value), - desc: vi.fn((value: unknown) => value), - eq: vi.fn((...args: unknown[]) => args), recordTaskRunLifecycleEvent: mocks.recordLifecycle, - sql: vi.fn((strings: TemplateStringsArray, ...values: unknown[]) => ({ - strings: [...strings], - values, - })), - taskRuns: { - id: 'task_runs.id', - taskId: 'task_runs.task_id', - createdAt: 'task_runs.created_at', - result: 'task_runs.result', - }, })); vi.mock('@roomote/cloud-agents/server', () => ({ getTaskUrl: mocks.getTaskUrl, })); -vi.mock('../../fast-agent-parent-event', () => ({ - FastAgentParentEventDeliveryError: mocks.FastAgentParentEventDeliveryError, -})); - vi.mock('../../fast-agent-parent-event-queue', () => ({ - enqueueFastAgentParentEventAndWait: mocks.enqueueParentEventAndWait, + enqueueFastAgentParentEvent: mocks.enqueueParentEvent, })); import { notifyFastAgentParentOnPrFeedback } from '../notify-fast-agent-parent-on-pr-feedback'; @@ -121,37 +67,24 @@ const input = { suggestedActionPrompt: 'Address the requested changes.', }; -function claimedFeedbackIds(): string[] { - return mocks.claimConversationDelivery.mock.calls.map( - (call: unknown[]) => (call[0] as { feedbackId: string }).feedbackId, +function enqueuedFeedbackIds(): string[] { + return mocks.enqueueParentEvent.mock.calls.map( + (call: unknown[]) => + (call[0] as { event: { feedbackId: string } }).event.feedbackId, ); } describe('notifyFastAgentParentOnPrFeedback', () => { beforeEach(() => { vi.clearAllMocks(); - mocks.claimReturning.mockResolvedValue([{ id: 200 }]); - const claimed = new Set(); - mocks.claimConversationDelivery.mockImplementation( - async ({ feedbackId }: { feedbackId: string }) => { - if (claimed.has(feedbackId)) { - return { status: 'already_claimed' as const }; - } - claimed.add(feedbackId); - return { - status: 'claimed' as const, - claim: { id: `claim-${feedbackId}`, leaseToken: 'lease-token' }, - }; - }, - ); - mocks.completeConversationDelivery.mockResolvedValue(undefined); - mocks.releaseConversationDelivery.mockResolvedValue(undefined); mocks.findReusableOwner.mockResolvedValue({ taskId: 'child-task', runId: 200, }); - mocks.findClaimRun.mockResolvedValue({ id: 200 }); - mocks.enqueueParentEventAndWait.mockResolvedValue('delivered'); + mocks.enqueueParentEvent.mockResolvedValue({ + eventKey: 'feedback-event', + queued: true, + }); mocks.recordLifecycle.mockResolvedValue(undefined); }); @@ -163,23 +96,20 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }), ).resolves.toBe(true); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledWith( - { - parent: fastParent, - event: { - type: 'pull_request_feedback', - feedbackId: expect.stringMatching(/^[a-f0-9]{24}$/), - taskId: 'child-task', - runId: 200, - taskUrl: 'https://roomote.example/task/child-task', - pullRequest: input.pullRequest, - summary: input.summary, - suggestedActionQuestion: input.suggestedActionQuestion, - suggestedActionPrompt: input.suggestedActionPrompt, - }, + expect(mocks.enqueueParentEvent).toHaveBeenCalledWith({ + parent: fastParent, + event: { + type: 'pull_request_feedback', + feedbackId: expect.stringMatching(/^[a-f0-9]{24}$/), + taskId: 'child-task', + runId: 200, + taskUrl: 'https://roomote.example/task/child-task', + pullRequest: input.pullRequest, + summary: input.summary, + suggestedActionQuestion: input.suggestedActionQuestion, + suggestedActionPrompt: input.suggestedActionPrompt, }, - { timeoutMs: 30_000 }, - ); + }); expect(mocks.recordLifecycle).toHaveBeenCalledWith( expect.anything(), expect.objectContaining({ @@ -199,7 +129,50 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }), ).resolves.toBe(false); - expect(mocks.enqueueParentEventAndWait).not.toHaveBeenCalled(); + expect(mocks.enqueueParentEvent).not.toHaveBeenCalled(); + }); + + it('suppresses review-pipeline runs inherited from the parent Session', async () => { + await expect( + notifyFastAgentParentOnPrFeedback({ + run: makeRun({ + type: 'github_pr_review', + fastAgentParent: fastParent, + }), + ...input, + }), + ).resolves.toBe(false); + + expect(mocks.enqueueParentEvent).not.toHaveBeenCalled(); + }); + + it('surfaces durable admission failures to the caller', async () => { + mocks.enqueueParentEvent.mockRejectedValueOnce( + new Error('database offline'), + ); + + await expect( + notifyFastAgentParentOnPrFeedback({ + run: makeRun({ fastAgentParent: fastParent }), + ...input, + }), + ).rejects.toThrow('database offline'); + expect(mocks.recordLifecycle).not.toHaveBeenCalled(); + }); + + it('keeps successful admission when lifecycle logging fails', async () => { + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + mocks.recordLifecycle.mockRejectedValueOnce(new Error('logging offline')); + + await expect( + notifyFastAgentParentOnPrFeedback({ + run: makeRun({ fastAgentParent: fastParent }), + ...input, + }), + ).resolves.toBe(true); + expect(mocks.enqueueParentEvent).toHaveBeenCalledOnce(); + expect(errorSpy).toHaveBeenCalledOnce(); + errorSpy.mockRestore(); }); it('uses the same feedback identity regardless of source event order', async () => { @@ -215,8 +188,8 @@ describe('notifyFastAgentParentOnPrFeedback', () => { feedbackSourceIds: ['delivery-1', 'delivery-2'], }); - expect(claimedFeedbackIds()[1]).toBe(claimedFeedbackIds()[0]); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); + expect(enqueuedFeedbackIds()[1]).toBe(enqueuedFeedbackIds()[0]); + expect(mocks.enqueueParentEvent).toHaveBeenCalledTimes(2); }); it('uses stable source events instead of generated summary text for fallback identity', async () => { @@ -234,11 +207,11 @@ describe('notifyFastAgentParentOnPrFeedback', () => { feedbackSourceIds: ['github-review:123'], }); - expect(claimedFeedbackIds()[1]).toBe(claimedFeedbackIds()[0]); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); + expect(enqueuedFeedbackIds()[1]).toBe(enqueuedFeedbackIds()[0]); + expect(mocks.enqueueParentEvent).toHaveBeenCalledTimes(2); }); - it('delivers once across linked tasks sharing a conversation', async () => { + it('uses one stable queue identity across linked tasks sharing a conversation', async () => { const olderRun = makeRun( { fastAgentParent: fastParent }, { id: 100, taskId: 'older-task' }, @@ -255,8 +228,8 @@ describe('notifyFastAgentParentOnPrFeedback', () => { await notifyFastAgentParentOnPrFeedback({ run: olderRun, ...input }); await notifyFastAgentParentOnPrFeedback({ run: newerRun, ...input }); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); - expect(mocks.claimConversationDelivery).toHaveBeenCalledTimes(2); + expect(enqueuedFeedbackIds()[1]).toBe(enqueuedFeedbackIds()[0]); + expect(mocks.enqueueParentEvent).toHaveBeenCalledTimes(2); }); it('still delivers when a different task is the reusable owner', async () => { @@ -277,8 +250,8 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }), ).resolves.toBe(true); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledWith( + expect(mocks.enqueueParentEvent).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEvent).toHaveBeenCalledWith( expect.objectContaining({ event: expect.objectContaining({ taskId: 'newer-task', @@ -286,24 +259,9 @@ describe('notifyFastAgentParentOnPrFeedback', () => { taskUrl: 'https://roomote.example/task/newer-task', }), }), - { timeoutMs: 30_000 }, ); }); - it('falls back to the task-scoped claim when the conversation row is missing', async () => { - mocks.claimConversationDelivery.mockResolvedValue({ - status: 'no_conversation' as const, - }); - - await notifyFastAgentParentOnPrFeedback({ - run: makeRun({ fastAgentParent: fastParent }), - ...input, - }); - - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); - expect(mocks.completeConversationDelivery).not.toHaveBeenCalled(); - }); - it('shares a feedback identity between direct review handoff and webhook delivery', async () => { const run = makeRun({ fastAgentParent: fastParent }); await notifyFastAgentParentOnPrFeedback({ @@ -321,8 +279,8 @@ describe('notifyFastAgentParentOnPrFeedback', () => { reviewHeadSha: 'abc123', }); - expect(claimedFeedbackIds()[1]).toBe(claimedFeedbackIds()[0]); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledOnce(); + expect(enqueuedFeedbackIds()[1]).toBe(enqueuedFeedbackIds()[0]); + expect(mocks.enqueueParentEvent).toHaveBeenCalledTimes(2); }); it('preserves structured terminal review metadata in the Fast event', async () => { @@ -340,17 +298,16 @@ describe('notifyFastAgentParentOnPrFeedback', () => { reviewResult, }); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledWith( + expect(mocks.enqueueParentEvent).toHaveBeenCalledWith( expect.objectContaining({ event: expect.objectContaining({ reviewResult }), }), - { timeoutMs: 30_000 }, ); }); it('does nothing for a task without a Fast parent', async () => { await notifyFastAgentParentOnPrFeedback({ run: makeRun({}), ...input }); - expect(mocks.enqueueParentEventAndWait).not.toHaveBeenCalled(); + expect(mocks.enqueueParentEvent).not.toHaveBeenCalled(); }); }); diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts index 138bb4049..e9dcfef43 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts @@ -1,76 +1,22 @@ import type { TaskRun } from '@roomote/db/server'; -const mocks = vi.hoisted(() => { - class FastAgentParentEventDeliveryError extends Error { - readonly replyPosted: boolean; - readonly permanent: boolean; - - constructor( - message: string, - options: { replyPosted: boolean; permanent?: boolean }, - ) { - super(message); - this.replyPosted = options.replyPosted; - this.permanent = options.permanent ?? false; - } - } - - return { - claimReturning: vi.fn(), - findClaimRun: vi.fn(), - updateSet: vi.fn(), - recordLifecycle: vi.fn(), - deliverParentEvent: vi.fn(), - enqueueParentEvent: vi.fn(), - enqueueParentEventAndWait: vi.fn(), - getTaskUrl: vi.fn(() => 'https://roomote.example/task/child-task'), - FastAgentParentEventDeliveryError, - }; -}); +const mocks = vi.hoisted(() => ({ + recordLifecycle: vi.fn(), + enqueueParentEvent: vi.fn(), + getTaskUrl: vi.fn(() => 'https://roomote.example/task/child-task'), +})); vi.mock('@roomote/db/server', () => ({ - db: { - query: { - taskRuns: { findFirst: mocks.findClaimRun }, - }, - update: vi.fn(() => ({ - set: vi.fn((values: unknown) => { - mocks.updateSet(values); - return { - where: vi.fn(() => ({ returning: mocks.claimReturning })), - }; - }), - })), - }, - and: vi.fn((...args: unknown[]) => args), - asc: vi.fn((value: unknown) => value), - desc: vi.fn((value: unknown) => value), - eq: vi.fn((...args: unknown[]) => args), + db: {}, recordTaskRunLifecycleEvent: mocks.recordLifecycle, - sql: vi.fn((strings: TemplateStringsArray, ...values: unknown[]) => ({ - strings: [...strings], - values, - })), - taskRuns: { - id: 'task_runs.id', - taskId: 'task_runs.task_id', - createdAt: 'task_runs.created_at', - result: 'task_runs.result', - }, })); vi.mock('@roomote/cloud-agents/server', () => ({ getTaskUrl: mocks.getTaskUrl, })); -vi.mock('../../fast-agent-parent-event', () => ({ - deliverFastAgentParentEvent: mocks.deliverParentEvent, - FastAgentParentEventDeliveryError: mocks.FastAgentParentEventDeliveryError, -})); - vi.mock('../../fast-agent-parent-event-queue', () => ({ enqueueFastAgentParentEvent: mocks.enqueueParentEvent, - enqueueFastAgentParentEventAndWait: mocks.enqueueParentEventAndWait, })); import { notifyFastAgentParentOnPullRequestStatusChanged } from '../notify-fast-agent-parent-on-pull-request-status-changed'; @@ -110,14 +56,10 @@ const pullRequest = { describe('notifyFastAgentParentOnPullRequestStatusChanged', () => { beforeEach(() => { vi.clearAllMocks(); - mocks.claimReturning.mockResolvedValue([{ id: 200 }]); - mocks.findClaimRun.mockResolvedValue({ id: 200 }); - mocks.deliverParentEvent.mockResolvedValue('delivered'); mocks.enqueueParentEvent.mockResolvedValue({ eventKey: 'pr-status-event', queued: true, }); - mocks.enqueueParentEventAndWait.mockResolvedValue('delivered'); mocks.recordLifecycle.mockResolvedValue(undefined); }); @@ -171,16 +113,17 @@ describe('notifyFastAgentParentOnPullRequestStatusChanged', () => { actorLogin: 'alice', }); - expect(mocks.deliverParentEvent).not.toHaveBeenCalled(); + expect(mocks.enqueueParentEvent).not.toHaveBeenCalled(); }); }); describe('notifyFastAgentParentOnPullRequestConflict', () => { beforeEach(() => { vi.clearAllMocks(); - mocks.claimReturning.mockResolvedValue([{ id: 200 }]); - mocks.findClaimRun.mockResolvedValue({ id: 200 }); - mocks.enqueueParentEventAndWait.mockResolvedValue('delivered'); + mocks.enqueueParentEvent.mockResolvedValue({ + eventKey: 'pr-conflict-event', + queued: true, + }); mocks.recordLifecycle.mockResolvedValue(undefined); }); @@ -200,23 +143,64 @@ describe('notifyFastAgentParentOnPullRequestConflict', () => { }); expect(delivered).toBe(true); - expect(mocks.enqueueParentEventAndWait).toHaveBeenCalledWith( - { - parent: fastParent, - event: expect.objectContaining({ - type: 'pull_request_conflict_detected', - taskId: 'child-task', - runId: 200, - conflictDetectedAt: conflictDetectedAt.toISOString(), - message: - '[Fix review feedback](https://github.com/acme/web/pull/42) now has merge conflicts. Update the branch or ask Roomote to resolve them.', - pullRequest: expect.objectContaining({ - repository: 'acme/web', - number: 42, - }), + expect(mocks.enqueueParentEvent).toHaveBeenCalledWith({ + parent: fastParent, + event: expect.objectContaining({ + type: 'pull_request_conflict_detected', + taskId: 'child-task', + runId: 200, + conflictDetectedAt: conflictDetectedAt.toISOString(), + message: + '[Fix review feedback](https://github.com/acme/web/pull/42) now has merge conflicts. Update the branch or ask Roomote to resolve them.', + pullRequest: expect.objectContaining({ + repository: 'acme/web', + number: 42, }), - }, - { timeoutMs: 30_000 }, + }), + }); + }); + + it('suppresses conflict notifications from review-pipeline runs', async () => { + await expect( + notifyFastAgentParentOnPullRequestConflict({ + run: makeRun({ + type: 'github_pr_review_sync', + fastAgentParent: fastParent, + }), + pullRequest: { + provider: pullRequest.provider, + host: pullRequest.host, + repository: pullRequest.repository, + number: pullRequest.number, + title: pullRequest.title, + url: pullRequest.url, + }, + conflictDetectedAt: new Date('2026-08-24T23:00:00.000Z'), + }), + ).resolves.toBe(false); + + expect(mocks.enqueueParentEvent).not.toHaveBeenCalled(); + }); + + it('surfaces durable conflict admission failures', async () => { + mocks.enqueueParentEvent.mockRejectedValueOnce( + new Error('database offline'), ); + + await expect( + notifyFastAgentParentOnPullRequestConflict({ + run: makeRun({ fastAgentParent: fastParent }), + pullRequest: { + provider: pullRequest.provider, + host: pullRequest.host, + repository: pullRequest.repository, + number: pullRequest.number, + title: pullRequest.title, + url: pullRequest.url, + }, + conflictDetectedAt: new Date('2026-08-24T23:00:00.000Z'), + }), + ).rejects.toThrow('database offline'); + expect(mocks.recordLifecycle).not.toHaveBeenCalled(); }); }); diff --git a/packages/sdk/src/server/lib/task-runs/deliver-fast-agent-parent-pr-event.ts b/packages/sdk/src/server/lib/task-runs/deliver-fast-agent-parent-pr-event.ts deleted file mode 100644 index 27892f455..000000000 --- a/packages/sdk/src/server/lib/task-runs/deliver-fast-agent-parent-pr-event.ts +++ /dev/null @@ -1,192 +0,0 @@ -import { - type FastAgentPrFeedbackDeliveryClaim, - claimFastAgentPrFeedbackDelivery, - completeFastAgentPrFeedbackDelivery, - type SQL, - and, - asc, - db, - desc, - eq, - releaseFastAgentPrFeedbackDelivery, - sql, - taskRuns, -} from '@roomote/db/server'; -import type { FastAgentConversation } from '@roomote/types'; - -import { FastAgentParentEventDeliveryError } from '../fast-agent-parent-event'; -import { - buildFastAgentDeliveringMarker, - buildFastAgentDeliveryClaimPredicate, -} from './fast-agent-delivery-claim'; - -/** Own the shared claim and retry lifecycle for a Fast parent PR event. */ -export async function deliverFastAgentParentPrEvent(params: { - run: { id: number; taskId: string }; - deliveryKey: string; - claimCondition?: SQL; - deliver: () => Promise<'delivered' | 'skipped'>; - recordLifecycle: () => Promise; - logPrefix: string; - conversationClaim?: { - conversation: Pick< - FastAgentConversation, - 'surface' | 'workspaceId' | 'conversationId' - >; - feedbackId: string; - }; - /** Canonical destination delivery already owns the external side effect. */ - canonicalDeliveryOwned?: boolean; -}): Promise { - if (params.canonicalDeliveryOwned) { - return deliverClaimedFastAgentParentPrEvent({ - ...params, - markDelivered: async () => true, - releaseClaim: async () => undefined, - }); - } - - let conversationClaim: FastAgentPrFeedbackDeliveryClaim | null = null; - if (params.conversationClaim) { - const claimResult = await claimFastAgentPrFeedbackDelivery({ - conversation: params.conversationClaim.conversation, - feedbackId: params.conversationClaim.feedbackId, - taskId: params.run.taskId, - }); - if (claimResult.status === 'already_claimed') { - return; - } - if (claimResult.status === 'no_conversation') { - // The conversation row is the dedupe scope. Without it there is nothing - // to deduplicate against, so fall through to the task-scoped claim - // rather than silently dropping the event. - console.warn( - `[${params.logPrefix}] No Fast conversation row for ${params.conversationClaim.conversation.surface}:${params.conversationClaim.conversation.workspaceId}:${params.conversationClaim.conversation.conversationId}; falling back to the task-scoped delivery claim.`, - ); - } else { - conversationClaim = claimResult.claim; - } - } - - // Keep one claim row per task so a resume between two delivery paths cannot - // make the same Fast event look new. Prefer a row that already owns this key - // for compatibility with claims written before task-scoped delivery. - const claimRun = await db.query.taskRuns.findFirst({ - where: eq(taskRuns.taskId, params.run.taskId), - orderBy: [ - desc( - sql`coalesce(${taskRuns.result}, '{}'::jsonb) ? ${params.deliveryKey}`, - ), - asc(taskRuns.createdAt), - asc(taskRuns.id), - ], - columns: { id: true }, - }); - if (!claimRun) { - return; - } - - const markDelivered = async () => { - await db - .update(taskRuns) - .set({ - result: sql`coalesce(${taskRuns.result}, '{}'::jsonb) || jsonb_build_object(${params.deliveryKey}::text, to_jsonb(now()))`, - }) - .where(eq(taskRuns.id, claimRun.id)); - }; - - // A held conversation claim is already the single arbiter for this identity, - // so it must not be double-gated on the task-scoped claim predicate. Stamp - // the run row on success anyway so the delivery stays visible on the task. - if (conversationClaim) { - const claim = conversationClaim; - await deliverClaimedFastAgentParentPrEvent({ - ...params, - markDelivered: async () => { - await completeFastAgentPrFeedbackDelivery(claim); - await markDelivered(); - return true; - }, - releaseClaim: () => releaseFastAgentPrFeedbackDelivery(claim), - }); - return; - } - - const claimRows = await db - .update(taskRuns) - .set({ - result: sql`coalesce(${taskRuns.result}, '{}'::jsonb) || jsonb_build_object(${params.deliveryKey}::text, ${buildFastAgentDeliveringMarker()}::text)`, - }) - .where( - and( - eq(taskRuns.id, claimRun.id), - params.claimCondition, - buildFastAgentDeliveryClaimPredicate(params.deliveryKey), - ), - ) - .returning({ id: taskRuns.id }); - - if (claimRows.length === 0) { - return; - } - - await deliverClaimedFastAgentParentPrEvent({ - ...params, - markDelivered: async () => { - await markDelivered(); - return true; - }, - releaseClaim: async () => { - await db - .update(taskRuns) - .set({ - result: sql`coalesce(${taskRuns.result}, '{}'::jsonb) - ${params.deliveryKey}`, - }) - .where(eq(taskRuns.id, claimRun.id)); - }, - }); - return; -} - -async function deliverClaimedFastAgentParentPrEvent(params: { - run: { id: number }; - deliver: () => Promise<'delivered' | 'skipped'>; - recordLifecycle: () => Promise; - logPrefix: string; - markDelivered: () => Promise; - releaseClaim: () => Promise; -}): Promise { - let delivered = false; - try { - const delivery = await params.deliver(); - if (delivery === 'skipped') { - return params.markDelivered(); - } - delivered = true; - - if (!(await params.markDelivered())) { - return false; - } - await params.recordLifecycle(); - return true; - } catch (error) { - console.error( - `[${params.logPrefix}] Failed for run ${params.run.id}: ${ - error instanceof Error ? error.message : String(error) - }`, - ); - const deliveryError = - error instanceof FastAgentParentEventDeliveryError ? error : null; - - if (delivered || deliveryError?.replyPosted || deliveryError?.permanent) { - return params.markDelivered().catch(() => false); - } - - try { - await params.releaseClaim(); - } catch { - // Best-effort claim release for a later retry. - } - throw error; - } -} diff --git a/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pr-feedback.ts b/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pr-feedback.ts index 7e55f90ae..03175b281 100644 --- a/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pr-feedback.ts +++ b/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pr-feedback.ts @@ -15,10 +15,7 @@ import { } from '@roomote/types'; import { type FastAgentPullRequestContext } from '../fast-agent-parent-event'; -import { enqueueFastAgentParentEventAndWait } from '../fast-agent-parent-event-queue'; -import { deliverFastAgentParentPrEvent } from './deliver-fast-agent-parent-pr-event'; - -const PR_FEEDBACK_DELIVERY_WAIT_TIMEOUT_MS = 30_000; +import { enqueueFastAgentParentEvent } from '../fast-agent-parent-event-queue'; function buildFeedbackId(params: { conversation: { @@ -97,7 +94,6 @@ export async function notifyFastAgentParentOnPrFeedback(params: { feedbackSourceIds?: string[]; suggestedActionQuestion?: string; suggestedActionPrompt?: string; - canonicalDeliveryOwned?: boolean; reviewActionDeliveryId?: string; reviewResult?: { reviewKind: 'initial' | 'sync' | null; @@ -147,7 +143,6 @@ export async function notifyFastAgentParentOnPrFeedback(params: { reviewHeadSha: params.reviewHeadSha, reviewResult: params.reviewResult, }); - const notifiedResultKey = `fastAgentParentPrFeedback:${feedbackId}`; const pullRequest: FastAgentPullRequestContext = { provider: params.pullRequest.provider, host: params.pullRequest.host ?? null, @@ -158,64 +153,55 @@ export async function notifyFastAgentParentOnPrFeedback(params: { status: params.pullRequest.status ?? null, }; - const delivered = await deliverFastAgentParentPrEvent({ - run: params.run, - deliveryKey: notifiedResultKey, - logPrefix: 'notifyFastAgentParentOnPrFeedback', - conversationClaim: { - conversation: parent.conversation, + await enqueueFastAgentParentEvent({ + parent, + event: { + type: 'pull_request_feedback', feedbackId, - }, - canonicalDeliveryOwned: params.canonicalDeliveryOwned, - deliver: () => - enqueueFastAgentParentEventAndWait( - { - parent, - event: { - type: 'pull_request_feedback', - feedbackId, - taskId: attributedTaskId, - runId: attributedRunId, - taskUrl: getTaskUrl({ - taskId: attributedTaskId, - utm: { - source: parent.conversation.surface, - campaign: 'fast-delegation-pr-feedback', - }, - }), - pullRequest, - summary: params.summary, - ...(params.reviewResult - ? { reviewResult: params.reviewResult } - : {}), - ...(params.suggestedActionQuestion - ? { suggestedActionQuestion: params.suggestedActionQuestion } - : {}), - ...(params.suggestedActionPrompt - ? { suggestedActionPrompt: params.suggestedActionPrompt } - : {}), - ...(params.reviewActionDeliveryId - ? { reviewActionDeliveryId: params.reviewActionDeliveryId } - : {}), - }, - }, - { timeoutMs: PR_FEEDBACK_DELIVERY_WAIT_TIMEOUT_MS }, - ), - recordLifecycle: () => - recordTaskRunLifecycleEvent(db, { - runId: params.run.id, - taskId: params.run.taskId, - eventType: 'decision', - message: `Passed pull request feedback for ${pullRequest.repository ?? 'unknown'}#${pullRequest.number ?? 'unknown'} to the Fast parent orchestrator.`, - details: { - reason: 'fast_agent_parent_pr_feedback_event', - fastAgentSessionId: parent.sessionId, - provider: pullRequest.provider, - repository: pullRequest.repository, - prNumber: pullRequest.number, - prUrl: pullRequest.url, + taskId: attributedTaskId, + runId: attributedRunId, + taskUrl: getTaskUrl({ + taskId: attributedTaskId, + utm: { + source: parent.conversation.surface, + campaign: 'fast-delegation-pr-feedback', }, }), + pullRequest, + summary: params.summary, + ...(params.reviewResult ? { reviewResult: params.reviewResult } : {}), + ...(params.suggestedActionQuestion + ? { suggestedActionQuestion: params.suggestedActionQuestion } + : {}), + ...(params.suggestedActionPrompt + ? { suggestedActionPrompt: params.suggestedActionPrompt } + : {}), + ...(params.reviewActionDeliveryId + ? { reviewActionDeliveryId: params.reviewActionDeliveryId } + : {}), + }, }); - return params.canonicalDeliveryOwned ? delivered === true : true; + + try { + await recordTaskRunLifecycleEvent(db, { + runId: params.run.id, + taskId: params.run.taskId, + eventType: 'decision', + message: `Queued pull request feedback for ${pullRequest.repository ?? 'unknown'}#${pullRequest.number ?? 'unknown'} for the Fast parent orchestrator.`, + details: { + reason: 'fast_agent_parent_pr_feedback_event', + fastAgentSessionId: parent.sessionId, + provider: pullRequest.provider, + repository: pullRequest.repository, + prNumber: pullRequest.number, + prUrl: pullRequest.url, + }, + }); + } catch (error) { + console.error( + `[notifyFastAgentParentOnPrFeedback] Failed to record queue admission for run ${params.run.id}: ${error instanceof Error ? error.message : String(error)}`, + ); + } + + return true; } diff --git a/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-conflict.ts b/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-conflict.ts index 5445cf975..0775d01ff 100644 --- a/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-conflict.ts +++ b/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-conflict.ts @@ -1,5 +1,3 @@ -import { createHash } from 'node:crypto'; - import { getTaskUrl } from '@roomote/cloud-agents/server'; import { type TaskRun, @@ -13,23 +11,9 @@ import { } from '@roomote/types'; import { type FastAgentPullRequestContext } from '../fast-agent-parent-event'; -import { enqueueFastAgentParentEventAndWait } from '../fast-agent-parent-event-queue'; -import { deliverFastAgentParentPrEvent } from './deliver-fast-agent-parent-pr-event'; +import { enqueueFastAgentParentEvent } from '../fast-agent-parent-event-queue'; import { buildPullRequestConflictMessage } from './pull-request-mergeability-check'; -const PR_CONFLICT_DELIVERY_WAIT_TIMEOUT_MS = 30_000; - -function buildNotifiedResultKey(params: { - prUrl: string; - conflictDetectedAt: Date; -}): string { - const digest = createHash('sha256') - .update(`${params.prUrl}:${params.conflictDetectedAt.toISOString()}`) - .digest('hex') - .slice(0, 24); - return `fastAgentParentPrConflict:${digest}`; -} - /** Pass a durable PR conflict transition to the Fast conversation that delegated it. */ export async function notifyFastAgentParentOnPullRequestConflict(params: { run: Pick; @@ -53,10 +37,6 @@ export async function notifyFastAgentParentOnPullRequestConflict(params: { return false; } - const deliveryKey = buildNotifiedResultKey({ - prUrl: params.pullRequest.url, - conflictDetectedAt: params.conflictDetectedAt, - }); const pullRequest: FastAgentPullRequestContext = { provider: params.pullRequest.provider, host: params.pullRequest.host ?? null, @@ -67,52 +47,49 @@ export async function notifyFastAgentParentOnPullRequestConflict(params: { status: 'open', }; - await deliverFastAgentParentPrEvent({ - run: params.run, - deliveryKey, - logPrefix: 'notifyFastAgentParentOnPullRequestConflict', - deliver: () => - enqueueFastAgentParentEventAndWait( - { - parent, - event: { - type: 'pull_request_conflict_detected', - taskId: params.run.taskId, - runId: params.run.id, - taskUrl: getTaskUrl({ - taskId: params.run.taskId, - utm: { - source: parent.conversation.surface, - campaign: 'fast-delegation-pr-conflict', - }, - }), - pullRequest, - conflictDetectedAt: params.conflictDetectedAt.toISOString(), - message: buildPullRequestConflictMessage({ - title: params.pullRequest.title, - url: params.pullRequest.url, - }), - }, - }, - { timeoutMs: PR_CONFLICT_DELIVERY_WAIT_TIMEOUT_MS }, - ), - recordLifecycle: () => - recordTaskRunLifecycleEvent(db, { - runId: params.run.id, + await enqueueFastAgentParentEvent({ + parent, + event: { + type: 'pull_request_conflict_detected', + taskId: params.run.taskId, + runId: params.run.id, + taskUrl: getTaskUrl({ taskId: params.run.taskId, - eventType: 'decision', - message: `Passed merge conflicts on pull request ${pullRequest.repository ?? 'unknown'}#${pullRequest.number ?? 'unknown'} to the Fast parent orchestrator.`, - details: { - reason: 'fast_agent_parent_pr_conflict_event', - fastAgentSessionId: parent.sessionId, - provider: pullRequest.provider, - repository: pullRequest.repository, - prNumber: pullRequest.number, - prUrl: pullRequest.url, - conflictDetectedAt: params.conflictDetectedAt.toISOString(), + utm: { + source: parent.conversation.surface, + campaign: 'fast-delegation-pr-conflict', }, }), + pullRequest, + conflictDetectedAt: params.conflictDetectedAt.toISOString(), + message: buildPullRequestConflictMessage({ + title: params.pullRequest.title, + url: params.pullRequest.url, + }), + }, }); + try { + await recordTaskRunLifecycleEvent(db, { + runId: params.run.id, + taskId: params.run.taskId, + eventType: 'decision', + message: `Queued merge conflicts on pull request ${pullRequest.repository ?? 'unknown'}#${pullRequest.number ?? 'unknown'} for the Fast parent orchestrator.`, + details: { + reason: 'fast_agent_parent_pr_conflict_event', + fastAgentSessionId: parent.sessionId, + provider: pullRequest.provider, + repository: pullRequest.repository, + prNumber: pullRequest.number, + prUrl: pullRequest.url, + conflictDetectedAt: params.conflictDetectedAt.toISOString(), + }, + }); + } catch (error) { + console.error( + `[notifyFastAgentParentOnPullRequestConflict] Failed to record queue admission for run ${params.run.id}: ${error instanceof Error ? error.message : String(error)}`, + ); + } + return true; } diff --git a/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-status-changed.ts b/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-status-changed.ts index 67bf7fe8b..524220ca7 100644 --- a/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-status-changed.ts +++ b/packages/sdk/src/server/lib/task-runs/notify-fast-agent-parent-on-pull-request-status-changed.ts @@ -1,5 +1,3 @@ -import { createHash } from 'node:crypto'; - import { getTaskUrl } from '@roomote/cloud-agents/server'; import { type TaskRun, @@ -14,18 +12,6 @@ import { import type { FastAgentPullRequestContext } from '../fast-agent-parent-event'; import { enqueueFastAgentParentEvent } from '../fast-agent-parent-event-queue'; -import { deliverFastAgentParentPrEvent } from './deliver-fast-agent-parent-pr-event'; - -function buildNotifiedResultKey(params: { - prUrl: string; - status: 'merged' | 'closed'; -}): string { - const digest = createHash('sha256') - .update(`${params.prUrl}:${params.status}`) - .digest('hex') - .slice(0, 24); - return `fastAgentParentPrStatus:${digest}`; -} /** Pass a terminal task PR status to the Fast conversation that delegated it. */ export async function notifyFastAgentParentOnPullRequestStatusChanged(params: { @@ -54,10 +40,6 @@ export async function notifyFastAgentParentOnPullRequestStatusChanged(params: { return; } - const notifiedResultKey = buildNotifiedResultKey({ - prUrl: params.pullRequest.url, - status: params.pullRequest.status, - }); const pullRequest: FastAgentPullRequestContext = { provider: params.pullRequest.provider, host: params.pullRequest.host ?? null, @@ -69,48 +51,46 @@ export async function notifyFastAgentParentOnPullRequestStatusChanged(params: { status: params.pullRequest.status, }; - await deliverFastAgentParentPrEvent({ - run: params.run, - deliveryKey: notifiedResultKey, - logPrefix: 'notifyFastAgentParentOnPullRequestStatusChanged', - deliver: async () => { - await enqueueFastAgentParentEvent({ - parent, - event: { - type: 'pull_request_status_changed', - taskId: params.run.taskId, - runId: params.run.id, - taskUrl: getTaskUrl({ - taskId: params.run.taskId, - utm: { - source: parent.conversation.surface, - campaign: 'fast-delegation-pr-status', - }, - }), - pullRequest, - status: params.pullRequest.status, - actorLogin: params.actorLogin, - }, - }); - return 'delivered'; - }, - recordLifecycle: () => - recordTaskRunLifecycleEvent(db, { - runId: params.run.id, + await enqueueFastAgentParentEvent({ + parent, + event: { + type: 'pull_request_status_changed', + taskId: params.run.taskId, + runId: params.run.id, + taskUrl: getTaskUrl({ taskId: params.run.taskId, - eventType: 'decision', - message: `Queued ${params.pullRequest.status} pull request ${pullRequest.repository ?? 'unknown'}#${pullRequest.number ?? 'unknown'} for the Fast parent orchestrator.`, - details: { - reason: 'fast_agent_parent_pr_status_event', - fastAgentSessionId: parent.sessionId, - provider: pullRequest.provider, - repository: pullRequest.repository, - prNumber: pullRequest.number, - prUrl: pullRequest.url, - targetBranch: pullRequest.targetBranch, - status: params.pullRequest.status, - actorLogin: params.actorLogin, + utm: { + source: parent.conversation.surface, + campaign: 'fast-delegation-pr-status', }, }), + pullRequest, + status: params.pullRequest.status, + actorLogin: params.actorLogin, + }, }); + + try { + await recordTaskRunLifecycleEvent(db, { + runId: params.run.id, + taskId: params.run.taskId, + eventType: 'decision', + message: `Queued ${params.pullRequest.status} pull request ${pullRequest.repository ?? 'unknown'}#${pullRequest.number ?? 'unknown'} for the Fast parent orchestrator.`, + details: { + reason: 'fast_agent_parent_pr_status_event', + fastAgentSessionId: parent.sessionId, + provider: pullRequest.provider, + repository: pullRequest.repository, + prNumber: pullRequest.number, + prUrl: pullRequest.url, + targetBranch: pullRequest.targetBranch, + status: params.pullRequest.status, + actorLogin: params.actorLogin, + }, + }); + } catch (error) { + console.error( + `[notifyFastAgentParentOnPullRequestStatusChanged] Failed to record queue admission for run ${params.run.id}: ${error instanceof Error ? error.message : String(error)}`, + ); + } } From a9df5a6dbc0b1654957d254a3fcb8439430b13ea Mon Sep 17 00:00:00 2001 From: "@daniel-lxs" <57051444+daniel-lxs@users.noreply.github.com> Date: Thu, 3 Sep 2026 19:09:35 +0000 Subject: [PATCH 3/6] chore: remove obsolete delivery marker check --- .../sdk/src/server/lib/task-runs/fast-agent-delivery-claim.ts | 4 ---- 1 file changed, 4 deletions(-) diff --git a/packages/sdk/src/server/lib/task-runs/fast-agent-delivery-claim.ts b/packages/sdk/src/server/lib/task-runs/fast-agent-delivery-claim.ts index dbda8ebad..2f55dcb36 100644 --- a/packages/sdk/src/server/lib/task-runs/fast-agent-delivery-claim.ts +++ b/packages/sdk/src/server/lib/task-runs/fast-agent-delivery-claim.ts @@ -9,10 +9,6 @@ export function buildFastAgentDeliveringMarker(): string { return `delivering:${Date.now()}`; } -export function isFastAgentDeliveringMarker(value: unknown): value is string { - return typeof value === 'string' && value.startsWith('delivering:'); -} - /** * Claim predicate for a jsonb delivery key on task_runs.result: the key is * unclaimed, or holds a 'delivering:' lease older than the lease From 256605372f4ebba285ff9f7e63d71e44cb2d4db1 Mon Sep 17 00:00:00 2001 From: "@daniel-lxs" <57051444+daniel-lxs@users.noreply.github.com> Date: Thu, 3 Sep 2026 19:20:37 +0000 Subject: [PATCH 4/6] fix: preserve legacy artifact delivery markers --- .../__tests__/upload-complete.test.ts | 8 ++++ .../src/handlers/artifacts/upload-complete.ts | 6 +++ .../notify-fast-agent-parent.test.ts | 48 +++++++++++++++++++ .../lib/artifacts/notify-fast-agent-parent.ts | 31 +++++++++++- 4 files changed, 92 insertions(+), 1 deletion(-) diff --git a/apps/api/src/handlers/artifacts/__tests__/upload-complete.test.ts b/apps/api/src/handlers/artifacts/__tests__/upload-complete.test.ts index 825708792..75042434c 100644 --- a/apps/api/src/handlers/artifacts/__tests__/upload-complete.test.ts +++ b/apps/api/src/handlers/artifacts/__tests__/upload-complete.test.ts @@ -86,4 +86,12 @@ describe('markArtifactUploadComplete', () => { expect(response.status).toBe(503); }); + + it('retries while a previous-release artifact delivery is in progress', async () => { + mocks.notifyParent.mockResolvedValueOnce('in_progress'); + + const response = await markArtifactUploadComplete(context()); + + expect(response.status).toBe(503); + }); }); diff --git a/apps/api/src/handlers/artifacts/upload-complete.ts b/apps/api/src/handlers/artifacts/upload-complete.ts index 14ca9e3dd..f3aee7532 100644 --- a/apps/api/src/handlers/artifacts/upload-complete.ts +++ b/apps/api/src/handlers/artifacts/upload-complete.ts @@ -68,5 +68,11 @@ export async function markArtifactUploadComplete( 503, ); } + if (notification === 'in_progress') { + return c.json( + { error: 'Artifact published; parent notification is in progress' }, + 503, + ); + } return new Response(null, { status: 200 }); } diff --git a/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts b/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts index f7c4014c2..9aff0ab2d 100644 --- a/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts +++ b/packages/sdk/src/server/lib/artifacts/__tests__/notify-fast-agent-parent.test.ts @@ -114,6 +114,54 @@ describe('notifyFastAgentParentOnArtifact', () => { expect(mocks.recordLifecycle).not.toHaveBeenCalled(); }); + it('does not enqueue an artifact already delivered by the previous release', async () => { + mocks.findRun.mockResolvedValueOnce({ + id: 200, + taskId: 'child-task', + payload: { fastAgentParent: fastParent }, + result: { + 'fastAgentArtifact:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa': 'delivered', + }, + }); + + await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( + 'queued', + ); + expect(mocks.enqueueParentEvent).not.toHaveBeenCalled(); + }); + + it('keeps retrying while a previous-release delivery claim is live', async () => { + mocks.findRun.mockResolvedValueOnce({ + id: 200, + taskId: 'child-task', + payload: { fastAgentParent: fastParent }, + result: { + 'fastAgentArtifact:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa': `delivering:${Date.now()}`, + }, + }); + + await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( + 'in_progress', + ); + expect(mocks.enqueueParentEvent).not.toHaveBeenCalled(); + }); + + it('admits an artifact after a previous-release delivery claim expires', async () => { + mocks.findRun.mockResolvedValueOnce({ + id: 200, + taskId: 'child-task', + payload: { fastAgentParent: fastParent }, + result: { + 'fastAgentArtifact:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa': `delivering:${Date.now() - 16 * 60 * 1000}`, + }, + }); + + await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( + 'queued', + ); + expect(mocks.enqueueParentEvent).toHaveBeenCalledOnce(); + }); + it('uses inherited Fast parent metadata on resumed runs', async () => { mocks.findRun.mockResolvedValueOnce({ id: 200, diff --git a/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts b/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts index 1fe112243..0e5f0dfea 100644 --- a/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts +++ b/packages/sdk/src/server/lib/artifacts/notify-fast-agent-parent.ts @@ -12,9 +12,27 @@ import { enqueueFastAgentParentEvent } from '../fast-agent-parent-event-queue'; export type FastArtifactNotificationResult = | 'not_applicable' + | 'in_progress' | 'queued' | 'failed'; +const LEGACY_ARTIFACT_DELIVERY_LEASE_MS = 15 * 60 * 1000; + +function getLegacyArtifactDeliveryState( + marker: unknown, +): 'in_progress' | 'settled' | null { + if (marker === null || marker === undefined) return null; + if (typeof marker !== 'string' || !marker.startsWith('delivering:')) { + return 'settled'; + } + + const claimedAt = Number(marker.slice('delivering:'.length)); + return Number.isFinite(claimedAt) && + claimedAt >= Date.now() - LEGACY_ARTIFACT_DELIVERY_LEASE_MS + ? 'in_progress' + : null; +} + function buildArtifactViewUrl(input: { taskId: string; path: string; @@ -44,13 +62,24 @@ export async function notifyFastAgentParentOnArtifact(input: { const run = await db.query.taskRuns.findFirst({ where: and(eq(taskRuns.id, input.runId), eq(taskRuns.taskId, input.taskId)), - columns: { id: true, taskId: true, payload: true }, + columns: { id: true, taskId: true, payload: true, result: true }, }); const parent = getFastAgentParentFromPayload(run?.payload); if (!run || !parent) { return 'not_applicable'; } + // N-1 compatibility: a previous release recorded presentation claims on the + // run. Honor settled and live markers without creating new dual claims; stale + // claims fall through to the durable queue's event-key idempotency. + const legacyDeliveryState = getLegacyArtifactDeliveryState( + (run.result as Record | null)?.[ + `fastAgentArtifact:${input.id}` + ], + ); + if (legacyDeliveryState === 'settled') return 'queued'; + if (legacyDeliveryState === 'in_progress') return 'in_progress'; + try { await enqueueFastAgentParentEvent({ parent, From 3cc8cbde6f3dfcdf5f0fd9482c1d3bfd55e7f47e Mon Sep 17 00:00:00 2001 From: "@daniel-lxs" <57051444+daniel-lxs@users.noreply.github.com> Date: Thu, 3 Sep 2026 19:28:44 +0000 Subject: [PATCH 5/6] test: isolate review notification payload checks --- .../notify-fast-agent-parent-on-pr-feedback.test.ts | 9 +++++++++ ...t-agent-parent-on-pull-request-status-changed.test.ts | 9 +++++++++ 2 files changed, 18 insertions(+) diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts index 471d315c1..23e6e9619 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts @@ -22,6 +22,15 @@ vi.mock('@roomote/cloud-agents/server', () => ({ getTaskUrl: mocks.getTaskUrl, })); +vi.mock('@roomote/types', async (importOriginal) => ({ + ...(await importOriginal()), + getFastAgentParentFromPayload: (payload: Record) => + payload.fastAgentParent, + isPrReviewPayload: (payload: { type?: string }) => + payload.type === 'github_pr_review' || + payload.type === 'github_pr_review_sync', +})); + vi.mock('../../fast-agent-parent-event-queue', () => ({ enqueueFastAgentParentEvent: mocks.enqueueParentEvent, })); diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts index e9dcfef43..849fb9de2 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts @@ -15,6 +15,15 @@ vi.mock('@roomote/cloud-agents/server', () => ({ getTaskUrl: mocks.getTaskUrl, })); +vi.mock('@roomote/types', async (importOriginal) => ({ + ...(await importOriginal()), + getFastAgentParentFromPayload: (payload: Record) => + payload.fastAgentParent, + isPrReviewPayload: (payload: { type?: string }) => + payload.type === 'github_pr_review' || + payload.type === 'github_pr_review_sync', +})); + vi.mock('../../fast-agent-parent-event-queue', () => ({ enqueueFastAgentParentEvent: mocks.enqueueParentEvent, })); From b91a511d30591a20b6f59981319511c67838b606 Mon Sep 17 00:00:00 2001 From: "@daniel-lxs" <57051444+daniel-lxs@users.noreply.github.com> Date: Thu, 3 Sep 2026 19:35:46 +0000 Subject: [PATCH 6/6] test: model PR review payload kind in notifications --- .../notify-fast-agent-parent-on-pr-feedback.test.ts | 10 +--------- ...agent-parent-on-pull-request-status-changed.test.ts | 10 +--------- 2 files changed, 2 insertions(+), 18 deletions(-) diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts index 23e6e9619..9e1b0af76 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pr-feedback.test.ts @@ -22,15 +22,6 @@ vi.mock('@roomote/cloud-agents/server', () => ({ getTaskUrl: mocks.getTaskUrl, })); -vi.mock('@roomote/types', async (importOriginal) => ({ - ...(await importOriginal()), - getFastAgentParentFromPayload: (payload: Record) => - payload.fastAgentParent, - isPrReviewPayload: (payload: { type?: string }) => - payload.type === 'github_pr_review' || - payload.type === 'github_pr_review_sync', -})); - vi.mock('../../fast-agent-parent-event-queue', () => ({ enqueueFastAgentParentEvent: mocks.enqueueParentEvent, })); @@ -55,6 +46,7 @@ function makeRun( id: 200, taskId: 'child-task', payload, + payloadKind: typeof payload.type === 'string' ? payload.type : 'standard', result: null, error: null, ...overrides, diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts index 849fb9de2..c3d3ac921 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/notify-fast-agent-parent-on-pull-request-status-changed.test.ts @@ -15,15 +15,6 @@ vi.mock('@roomote/cloud-agents/server', () => ({ getTaskUrl: mocks.getTaskUrl, })); -vi.mock('@roomote/types', async (importOriginal) => ({ - ...(await importOriginal()), - getFastAgentParentFromPayload: (payload: Record) => - payload.fastAgentParent, - isPrReviewPayload: (payload: { type?: string }) => - payload.type === 'github_pr_review' || - payload.type === 'github_pr_review_sync', -})); - vi.mock('../../fast-agent-parent-event-queue', () => ({ enqueueFastAgentParentEvent: mocks.enqueueParentEvent, })); @@ -46,6 +37,7 @@ function makeRun(payload: Record): TaskRun { id: 200, taskId: 'child-task', payload, + payloadKind: typeof payload.type === 'string' ? payload.type : 'standard', result: null, error: null, } as TaskRun;