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..75042434c 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); @@ -90,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 725d63595..f3aee7532 100644 --- a/apps/api/src/handlers/artifacts/upload-complete.ts +++ b/apps/api/src/handlers/artifacts/upload-complete.ts @@ -69,14 +69,10 @@ export async function markArtifactUploadComplete( ); } 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 bc51f5d45..a5f4e52fe 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 36ed296b1..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 @@ -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(), - deliverParentEvent: 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,9 +24,8 @@ vi.mock('@roomote/env', () => ({ Env: { R_APP_URL: 'https://roomote.example' }, })); -vi.mock('../../fast-agent-parent-event', () => ({ - deliverFastAgentParentEvent: mocks.deliverParentEvent, - FastAgentParentEventDeliveryError: mocks.FastAgentParentEventDeliveryError, +vi.mock('../../fast-agent-parent-event-queue', () => ({ + enqueueFastAgentParentEvent: mocks.enqueueParentEvent, })); import { notifyFastAgentParentOnArtifact } from '../notify-fast-agent-parent'; @@ -96,34 +66,33 @@ describe('notifyFastAgentParentOnArtifact', () => { payload: { fastAgentParent: fastParent }, result: {}, }); - mocks.claimReturning.mockResolvedValue([{ id: 200 }]); - mocks.deliverParentEvent.mockResolvedValue(undefined); + 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.deliverParentEvent).toHaveBeenCalledWith( - expect.objectContaining({ - parent: fastParent, - lockWaitMs: expect.any(Number), - 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', }), }), - ); + }); expect(mocks.recordLifecycle).toHaveBeenCalledWith( expect.anything(), expect.objectContaining({ @@ -134,81 +103,63 @@ describe('notifyFastAgentParentOnArtifact', () => { ); }); - it('deduplicates an event already claimed by another delivery', async () => { - mocks.claimReturning.mockResolvedValueOnce([]); - - await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( - 'already_delivered', + it('reports a durable enqueue failure for retry', async () => { + mocks.enqueueParentEvent.mockRejectedValueOnce( + new Error('database offline'), ); - expect(mocks.deliverParentEvent).not.toHaveBeenCalled(); - }); - - it('releases a failed orchestrator delivery for retry', async () => { - mocks.deliverParentEvent.mockRejectedValueOnce(new Error('model 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); + expect(mocks.recordLifecycle).not.toHaveBeenCalled(); }); - it('reports an in-flight delivery as in_progress instead of delivered', async () => { - mocks.claimReturning.mockResolvedValueOnce([]); - mocks.findRun.mockResolvedValue({ + 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': `delivering:${Date.now()}`, + 'fastAgentArtifact:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa': 'delivered', }, }); await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( - 'in_progress', + 'queued', ); - expect(mocks.deliverParentEvent).not.toHaveBeenCalled(); + expect(mocks.enqueueParentEvent).not.toHaveBeenCalled(); }); - it('keeps the claim when the failure happened after the Slack post', async () => { - mocks.deliverParentEvent.mockRejectedValueOnce( - new mocks.FastAgentParentEventDeliveryError('lifecycle write failed', { - replyPosted: true, - }), - ); + 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( - 'delivered', + 'in_progress', ); - expect( - mocks.updateSet.mock.calls.some(([values]) => { - const result = (values as { result?: { strings?: string[] } }).result; - return result?.strings?.join('').includes(' - ') === true; - }), - ).toBe(false); + expect(mocks.enqueueParentEvent).not.toHaveBeenCalled(); }); - it('settles the claim as skipped when no retry can ever succeed', async () => { - mocks.deliverParentEvent.mockRejectedValueOnce( - new mocks.FastAgentParentEventDeliveryError('parent session gone', { - replyPosted: false, - permanent: true, - }), - ); + 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( - 'skipped', + 'queued', ); - 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.enqueueParentEvent).toHaveBeenCalledOnce(); }); it('uses inherited Fast parent metadata on resumed runs', async () => { @@ -224,9 +175,9 @@ describe('notifyFastAgentParentOnArtifact', () => { }); await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( - 'delivered', + 'queued', ); - expect(mocks.deliverParentEvent).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEvent).toHaveBeenCalledOnce(); }); it('does nothing for standalone artifacts', async () => { @@ -240,6 +191,6 @@ describe('notifyFastAgentParentOnArtifact', () => { await expect(notifyFastAgentParentOnArtifact(artifact())).resolves.toBe( 'not_applicable', ); - expect(mocks.deliverParentEvent).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 5020d1e52..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 @@ -4,32 +4,34 @@ import { db, eq, recordTaskRunLifecycleEvent, - sql, taskRuns, } from '@roomote/db/server'; import { Env } from '@roomote/env'; -import { - FastAgentParentEventDeliveryError, - deliverFastAgentParentEvent, -} from '../fast-agent-parent-event'; -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'; -/** 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; +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; @@ -67,47 +69,19 @@ export async function notifyFastAgentParentOnArtifact(input: { 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; + // 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 deliverFastAgentParentEvent({ + await enqueueFastAgentParentEvent({ parent, event: { type: 'artifact_published', @@ -121,60 +95,35 @@ export async function notifyFastAgentParentOnArtifact(input: { viewUrl: buildArtifactViewUrl(input), }, }, - lockWaitMs: ARTIFACT_DELIVERY_LOCK_WAIT_MS, }); - 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, - }, - }); + 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 'delivered'; + 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/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 c6cc8dbe3..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 @@ -1,80 +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(), - deliverParentEvent: 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', () => ({ - deliverFastAgentParentEvent: mocks.deliverParentEvent, - FastAgentParentEventDeliveryError: mocks.FastAgentParentEventDeliveryError, +vi.mock('../../fast-agent-parent-event-queue', () => ({ + enqueueFastAgentParentEvent: mocks.enqueueParentEvent, })); import { notifyFastAgentParentOnPrFeedback } from '../notify-fast-agent-parent-on-pr-feedback'; @@ -97,6 +46,7 @@ function makeRun( id: 200, taskId: 'child-task', payload, + payloadKind: typeof payload.type === 'string' ? payload.type : 'standard', result: null, error: null, ...overrides, @@ -118,37 +68,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.deliverParentEvent.mockResolvedValue('delivered'); + mocks.enqueueParentEvent.mockResolvedValue({ + eventKey: 'feedback-event', + queued: true, + }); mocks.recordLifecycle.mockResolvedValue(undefined); }); @@ -160,9 +97,8 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }), ).resolves.toBe(true); - expect(mocks.deliverParentEvent).toHaveBeenCalledWith({ + expect(mocks.enqueueParentEvent).toHaveBeenCalledWith({ parent: fastParent, - lockWaitMs: 30_000, event: { type: 'pull_request_feedback', feedbackId: expect.stringMatching(/^[a-f0-9]{24}$/), @@ -194,7 +130,50 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }), ).resolves.toBe(false); - expect(mocks.deliverParentEvent).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 () => { @@ -210,8 +189,8 @@ describe('notifyFastAgentParentOnPrFeedback', () => { feedbackSourceIds: ['delivery-1', 'delivery-2'], }); - expect(claimedFeedbackIds()[1]).toBe(claimedFeedbackIds()[0]); - expect(mocks.deliverParentEvent).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 () => { @@ -229,11 +208,11 @@ describe('notifyFastAgentParentOnPrFeedback', () => { feedbackSourceIds: ['github-review:123'], }); - expect(claimedFeedbackIds()[1]).toBe(claimedFeedbackIds()[0]); - expect(mocks.deliverParentEvent).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' }, @@ -250,8 +229,8 @@ describe('notifyFastAgentParentOnPrFeedback', () => { await notifyFastAgentParentOnPrFeedback({ run: olderRun, ...input }); await notifyFastAgentParentOnPrFeedback({ run: newerRun, ...input }); - expect(mocks.deliverParentEvent).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 () => { @@ -272,8 +251,8 @@ describe('notifyFastAgentParentOnPrFeedback', () => { }), ).resolves.toBe(true); - expect(mocks.deliverParentEvent).toHaveBeenCalledOnce(); - expect(mocks.deliverParentEvent).toHaveBeenCalledWith( + expect(mocks.enqueueParentEvent).toHaveBeenCalledOnce(); + expect(mocks.enqueueParentEvent).toHaveBeenCalledWith( expect.objectContaining({ event: expect.objectContaining({ taskId: 'newer-task', @@ -284,20 +263,6 @@ describe('notifyFastAgentParentOnPrFeedback', () => { ); }); - 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.deliverParentEvent).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({ @@ -315,8 +280,8 @@ describe('notifyFastAgentParentOnPrFeedback', () => { reviewHeadSha: 'abc123', }); - expect(claimedFeedbackIds()[1]).toBe(claimedFeedbackIds()[0]); - expect(mocks.deliverParentEvent).toHaveBeenCalledOnce(); + expect(enqueuedFeedbackIds()[1]).toBe(enqueuedFeedbackIds()[0]); + expect(mocks.enqueueParentEvent).toHaveBeenCalledTimes(2); }); it('preserves structured terminal review metadata in the Fast event', async () => { @@ -334,7 +299,7 @@ describe('notifyFastAgentParentOnPrFeedback', () => { reviewResult, }); - expect(mocks.deliverParentEvent).toHaveBeenCalledWith( + expect(mocks.enqueueParentEvent).toHaveBeenCalledWith( expect.objectContaining({ event: expect.objectContaining({ reviewResult }), }), @@ -344,6 +309,6 @@ describe('notifyFastAgentParentOnPrFeedback', () => { it('does nothing for a task without a Fast parent', async () => { await notifyFastAgentParentOnPrFeedback({ run: makeRun({}), ...input }); - expect(mocks.deliverParentEvent).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 8e3091e03..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 @@ -1,72 +1,20 @@ 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(), - 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, })); @@ -89,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; @@ -108,9 +57,6 @@ 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, @@ -168,16 +114,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.deliverParentEvent.mockResolvedValue('delivered'); + mocks.enqueueParentEvent.mockResolvedValue({ + eventKey: 'pr-conflict-event', + queued: true, + }); mocks.recordLifecycle.mockResolvedValue(undefined); }); @@ -197,9 +144,8 @@ describe('notifyFastAgentParentOnPullRequestConflict', () => { }); expect(delivered).toBe(true); - expect(mocks.deliverParentEvent).toHaveBeenCalledWith({ + expect(mocks.enqueueParentEvent).toHaveBeenCalledWith({ parent: fastParent, - lockWaitMs: 30_000, event: expect.objectContaining({ type: 'pull_request_conflict_detected', taskId: 'child-task', @@ -214,4 +160,48 @@ describe('notifyFastAgentParentOnPullRequestConflict', () => { }), }); }); + + 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/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 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 1704e2363..329f2b77e 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 @@ -14,13 +14,8 @@ import { isPrReviewRun, } from '@roomote/types'; -import { - deliverFastAgentParentEvent, - type FastAgentPullRequestContext, -} from '../fast-agent-parent-event'; -import { deliverFastAgentParentPrEvent } from './deliver-fast-agent-parent-pr-event'; - -const PR_FEEDBACK_DELIVERY_LOCK_WAIT_MS = 30_000; +import { type FastAgentPullRequestContext } from '../fast-agent-parent-event'; +import { enqueueFastAgentParentEvent } from '../fast-agent-parent-event-queue'; function buildFeedbackId(params: { conversation: { @@ -99,7 +94,6 @@ export async function notifyFastAgentParentOnPrFeedback(params: { feedbackSourceIds?: string[]; suggestedActionQuestion?: string; suggestedActionPrompt?: string; - canonicalDeliveryOwned?: boolean; reviewActionDeliveryId?: string; reviewResult?: { reviewKind: 'initial' | 'sync' | null; @@ -149,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, @@ -160,60 +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: () => - deliverFastAgentParentEvent({ - 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 } - : {}), - }, - lockWaitMs: PR_FEEDBACK_DELIVERY_LOCK_WAIT_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 dd3d7de99..d09fc974f 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, @@ -12,26 +10,10 @@ import { isPrReviewRun, } from '@roomote/types'; -import { - deliverFastAgentParentEvent, - type FastAgentPullRequestContext, -} from '../fast-agent-parent-event'; -import { deliverFastAgentParentPrEvent } from './deliver-fast-agent-parent-pr-event'; +import { type FastAgentPullRequestContext } from '../fast-agent-parent-event'; +import { enqueueFastAgentParentEvent } from '../fast-agent-parent-event-queue'; import { buildPullRequestConflictMessage } from './pull-request-mergeability-check'; -const PR_CONFLICT_DELIVERY_LOCK_WAIT_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; @@ -55,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, @@ -69,50 +47,49 @@ export async function notifyFastAgentParentOnPullRequestConflict(params: { status: 'open', }; - await deliverFastAgentParentPrEvent({ - run: params.run, - deliveryKey, - logPrefix: 'notifyFastAgentParentOnPullRequestConflict', - deliver: () => - deliverFastAgentParentEvent({ - 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, - }), - }, - lockWaitMs: PR_CONFLICT_DELIVERY_LOCK_WAIT_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 19e97a19a..dbeaf4bf0 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)}`, + ); + } }