diff --git a/apps/api/src/handlers/ado/handlePullRequest.ts b/apps/api/src/handlers/ado/handlePullRequest.ts index cc588b4f5..48d1a413e 100644 --- a/apps/api/src/handlers/ado/handlePullRequest.ts +++ b/apps/api/src/handlers/ado/handlePullRequest.ts @@ -22,6 +22,8 @@ import { import { enqueueTask } from '@roomote/cloud-agents/server'; import { recordPrStatusChangeInTaskHistory, + PrStatusFastDeliveryError, + PrStatusHistoryRecordingError, updateTaskPrStatus, } from '@roomote/sdk/server'; @@ -79,6 +81,8 @@ async function notifyTerminalPullRequestThreads( payload: AdoPullRequestWebhook, repoFullName: string, status: 'merged' | 'closed', + includeFastParentTargets: boolean, + includeFastParentTaskIds: string[], ): Promise { const prUrl = getAdoPullRequestUrl({ resourceContainers: payload.resourceContainers, @@ -113,6 +117,8 @@ async function notifyTerminalPullRequestThreads( actorLogin: getAdoIdentityName(payload.resource.closedBy) ?? 'someone in Azure DevOps', + ...(includeFastParentTargets ? { includeFastParentTargets: true } : {}), + ...(includeFastParentTaskIds.length ? { includeFastParentTaskIds } : {}), }, `PR #${payload.resource.pullRequestId}`, ); @@ -251,8 +257,10 @@ export async function handleAdoPullRequest( scheduleAdoPullRequestFactSync(payload, repoFullName, 'closed'); - await Promise.resolve( - recordPrStatusChangeInTaskHistory({ + let includeFastParentTargets = false; + let includeFastParentTaskIds: string[] = []; + try { + await recordPrStatusChangeInTaskHistory({ sourceControlProvider: 'ado', repository: repoFullName, prNumber: pullRequest.pullRequestId, @@ -266,16 +274,29 @@ export async function handleAdoPullRequest( actorLogin: getAdoIdentityName(payload.resource.closedBy) ?? 'someone in Azure DevOps', - }), - ).catch((error) => { + }); + } catch (error) { + if (error instanceof PrStatusFastDeliveryError) { + includeFastParentTaskIds = error.taskIds; + } else { + includeFastParentTargets = !( + error instanceof PrStatusHistoryRecordingError + ); + } console.warn( `[handleAdoPullRequest] Failed to record PR status in task history for ${repoFullName}#${pullRequest.pullRequestId}: ${ error instanceof Error ? error.message : String(error) }`, ); - }); + } - await notifyTerminalPullRequestThreads(payload, repoFullName, 'closed'); + await notifyTerminalPullRequestThreads( + payload, + repoFullName, + 'closed', + includeFastParentTargets, + includeFastParentTaskIds, + ); return { status: 'ok' }; } @@ -297,8 +318,10 @@ export async function handleAdoPullRequest( scheduleAdoPullRequestFactSync(payload, repoFullName, 'merged'); - await Promise.resolve( - recordPrStatusChangeInTaskHistory({ + let includeFastParentTargets = false; + let includeFastParentTaskIds: string[] = []; + try { + await recordPrStatusChangeInTaskHistory({ sourceControlProvider: 'ado', repository: repoFullName, prNumber: pullRequest.pullRequestId, @@ -312,16 +335,29 @@ export async function handleAdoPullRequest( actorLogin: getAdoIdentityName(payload.resource.closedBy) ?? 'someone in Azure DevOps', - }), - ).catch((error) => { + }); + } catch (error) { + if (error instanceof PrStatusFastDeliveryError) { + includeFastParentTaskIds = error.taskIds; + } else { + includeFastParentTargets = !( + error instanceof PrStatusHistoryRecordingError + ); + } console.warn( `[handleAdoPullRequest] Failed to record PR status in task history for ${repoFullName}#${pullRequest.pullRequestId}: ${ error instanceof Error ? error.message : String(error) }`, ); - }); + } - await notifyTerminalPullRequestThreads(payload, repoFullName, 'merged'); + await notifyTerminalPullRequestThreads( + payload, + repoFullName, + 'merged', + includeFastParentTargets, + includeFastParentTaskIds, + ); return { status: 'ok' }; } diff --git a/apps/api/src/handlers/bitbucket/handlePullRequest.ts b/apps/api/src/handlers/bitbucket/handlePullRequest.ts index 4121d8987..3b37029f4 100644 --- a/apps/api/src/handlers/bitbucket/handlePullRequest.ts +++ b/apps/api/src/handlers/bitbucket/handlePullRequest.ts @@ -16,6 +16,8 @@ import { import { enqueueTask } from '@roomote/cloud-agents/server'; import { recordPrStatusChangeInTaskHistory, + PrStatusFastDeliveryError, + PrStatusHistoryRecordingError, updateTaskPrStatus, } from '@roomote/sdk/server'; @@ -61,6 +63,8 @@ async function notifyTerminalPullRequestThreads( payload: BitbucketPullRequestWebhook, repoFullName: string, status: 'merged' | 'closed', + includeFastParentTargets: boolean, + includeFastParentTaskIds: string[], ): Promise { const prUrl = getBitbucketPullRequestUrl(payload); const webhookHost = toHostFromUrl(prUrl); @@ -91,6 +95,8 @@ async function notifyTerminalPullRequestThreads( prUrl, status, actorLogin: getBitbucketUsername(payload.actor) ?? 'someone on Bitbucket', + ...(includeFastParentTargets ? { includeFastParentTargets: true } : {}), + ...(includeFastParentTaskIds.length ? { includeFastParentTaskIds } : {}), }, `PR #${prNumber}`, ); @@ -131,8 +137,10 @@ export async function handleBitbucketPullRequest( }, }); - await Promise.resolve( - recordPrStatusChangeInTaskHistory({ + let includeFastParentTargets = false; + let includeFastParentTaskIds: string[] = []; + try { + await recordPrStatusChangeInTaskHistory({ sourceControlProvider: 'bitbucket', repository: repoFullName, prNumber, @@ -141,16 +149,29 @@ export async function handleBitbucketPullRequest( status, actorLogin: getBitbucketUsername(payload.actor) ?? 'someone on Bitbucket', - }), - ).catch((error) => { + }); + } catch (error) { + if (error instanceof PrStatusFastDeliveryError) { + includeFastParentTaskIds = error.taskIds; + } else { + includeFastParentTargets = !( + error instanceof PrStatusHistoryRecordingError + ); + } console.warn( `[handleBitbucketPullRequest] Failed to record PR status in task history for ${repoFullName}#${prNumber}: ${ error instanceof Error ? error.message : String(error) }`, ); - }); + } - await notifyTerminalPullRequestThreads(payload, repoFullName, status); + await notifyTerminalPullRequestThreads( + payload, + repoFullName, + status, + includeFastParentTargets, + includeFastParentTaskIds, + ); return { status: 'ok' }; } diff --git a/apps/api/src/handlers/gitea/handlePullRequest.ts b/apps/api/src/handlers/gitea/handlePullRequest.ts index 15aa6425d..bca5b27b4 100644 --- a/apps/api/src/handlers/gitea/handlePullRequest.ts +++ b/apps/api/src/handlers/gitea/handlePullRequest.ts @@ -16,6 +16,8 @@ import { import { enqueueTask } from '@roomote/cloud-agents/server'; import { recordPrStatusChangeInTaskHistory, + PrStatusFastDeliveryError, + PrStatusHistoryRecordingError, updateTaskPrStatus, } from '@roomote/sdk/server'; @@ -64,6 +66,8 @@ async function notifyTerminalPullRequestThreads( payload: GiteaPullRequestWebhook, repoFullName: string, status: 'merged' | 'closed', + includeFastParentTargets: boolean, + includeFastParentTaskIds: string[], ): Promise { const prUrl = getPullRequestUrl(payload); const webhookHost = toHostFromUrl(prUrl); @@ -92,6 +96,8 @@ async function notifyTerminalPullRequestThreads( prUrl, status, actorLogin: getGiteaUsername(payload.sender) ?? 'someone on Gitea', + ...(includeFastParentTargets ? { includeFastParentTargets: true } : {}), + ...(includeFastParentTaskIds.length ? { includeFastParentTaskIds } : {}), }, `PR #${payload.number}`, ); @@ -127,8 +133,10 @@ export async function handleGiteaPullRequest( }, }); - await Promise.resolve( - recordPrStatusChangeInTaskHistory({ + let includeFastParentTargets = false; + let includeFastParentTaskIds: string[] = []; + try { + await recordPrStatusChangeInTaskHistory({ sourceControlProvider: 'gitea', repository: repoFullName, prNumber: payload.number, @@ -136,16 +144,29 @@ export async function handleGiteaPullRequest( prUrl: getPullRequestUrl(payload), status, actorLogin: getGiteaUsername(payload.sender) ?? 'someone on Gitea', - }), - ).catch((error) => { + }); + } catch (error) { + if (error instanceof PrStatusFastDeliveryError) { + includeFastParentTaskIds = error.taskIds; + } else { + includeFastParentTargets = !( + error instanceof PrStatusHistoryRecordingError + ); + } console.warn( `[handleGiteaPullRequest] Failed to record PR status in task history for ${repoFullName}#${payload.number}: ${ error instanceof Error ? error.message : String(error) }`, ); - }); + } - await notifyTerminalPullRequestThreads(payload, repoFullName, status); + await notifyTerminalPullRequestThreads( + payload, + repoFullName, + status, + includeFastParentTargets, + includeFastParentTaskIds, + ); return { status: 'ok' }; } diff --git a/apps/api/src/handlers/github/__tests__/handlePrMerge.test.ts b/apps/api/src/handlers/github/__tests__/handlePrMerge.test.ts index 0427cd0f1..07cf02da4 100644 --- a/apps/api/src/handlers/github/__tests__/handlePrMerge.test.ts +++ b/apps/api/src/handlers/github/__tests__/handlePrMerge.test.ts @@ -106,4 +106,26 @@ describe('handlePrMerge', () => { expect(result.status).toBe('ok'); expect(mockedScheduleNotify).not.toHaveBeenCalled(); }); + + it('includes Fast parent targets when status recording failed', async () => { + const payload = makePayload(); + + await handlePrMerge(payload, { includeFastParentTargets: true }); + + expect(mockedScheduleNotify).toHaveBeenCalledWith( + expect.objectContaining({ includeFastParentTargets: true }), + 'PR #42', + ); + }); + + it('includes only failed Fast task targets for per-task fallback', async () => { + const payload = makePayload(); + + await handlePrMerge(payload, { includeFastParentTaskIds: ['task-2'] }); + + expect(mockedScheduleNotify).toHaveBeenCalledWith( + expect.objectContaining({ includeFastParentTaskIds: ['task-2'] }), + 'PR #42', + ); + }); }); diff --git a/apps/api/src/handlers/github/__tests__/index.test.ts b/apps/api/src/handlers/github/__tests__/index.test.ts index dbac48847..61662aa16 100644 --- a/apps/api/src/handlers/github/__tests__/index.test.ts +++ b/apps/api/src/handlers/github/__tests__/index.test.ts @@ -25,6 +25,8 @@ const { mockUpdateTaskPrStatus, mockUpsertGitHubPullRequestFactFromWebhook, mockRecordPrStatusChangeInTaskHistory, + MockPrStatusFastDeliveryError, + MockPrStatusHistoryRecordingError, mockIsFromKnownInstallation, mockVerify, mockVerifyAndReceive, @@ -61,6 +63,15 @@ const { mockUpdateTaskPrStatus: vi.fn(), mockUpsertGitHubPullRequestFactFromWebhook: vi.fn(), mockRecordPrStatusChangeInTaskHistory: vi.fn(), + MockPrStatusFastDeliveryError: class extends Error { + readonly taskIds: string[]; + + constructor(message: string, taskIds: string[]) { + super(message); + this.taskIds = taskIds; + } + }, + MockPrStatusHistoryRecordingError: class extends Error {}, mockIsFromKnownInstallation: vi.fn(), mockVerify: vi.fn(), mockVerifyAndReceive: vi.fn(), @@ -120,6 +131,8 @@ vi.mock('@roomote/sdk/server', () => ({ upsertGitHubPullRequestFactFromWebhook: mockUpsertGitHubPullRequestFactFromWebhook, recordPrStatusChangeInTaskHistory: mockRecordPrStatusChangeInTaskHistory, + PrStatusFastDeliveryError: MockPrStatusFastDeliveryError, + PrStatusHistoryRecordingError: MockPrStatusHistoryRecordingError, })); vi.mock('../../logging', () => ({ @@ -362,7 +375,85 @@ describe('github webhook router', () => { const response = await responsePromise; expect(response.status).toBe(200); - expect(mockHandlePrMerge).toHaveBeenCalledWith(payload); + expect(mockHandlePrMerge).toHaveBeenCalledWith(payload, { + includeFastParentTargets: false, + }); + }); + + it('restores direct Fast targets when status handling fails before delivery', async () => { + mockUpdateTaskPrStatus.mockResolvedValue(undefined); + mockUpsertGitHubPullRequestFactFromWebhook.mockResolvedValue(undefined); + mockRecordPrStatusChangeInTaskHistory.mockRejectedValue( + new Error('redis unavailable'), + ); + mockHandlePrMerge.mockResolvedValue({ status: 'ok' }); + const payload = makePullRequestPayload('closed'); + + const response = await app.request('http://localhost/api/webhooks/github', { + method: 'POST', + headers: { + 'x-github-delivery': 'delivery-status-history-failure', + 'x-github-event': 'pull_request', + 'x-hub-signature-256': 'sha256=test', + }, + body: JSON.stringify(payload), + }); + + expect(response.status).toBe(200); + expect(mockHandlePrMerge).toHaveBeenCalledWith(payload, { + includeFastParentTargets: true, + }); + }); + + it('keeps Fast targets suppressed when only task-history persistence fails', async () => { + mockUpdateTaskPrStatus.mockResolvedValue(undefined); + mockUpsertGitHubPullRequestFactFromWebhook.mockResolvedValue(undefined); + mockRecordPrStatusChangeInTaskHistory.mockRejectedValue( + new MockPrStatusHistoryRecordingError('redis unavailable'), + ); + mockHandlePrMerge.mockResolvedValue({ status: 'ok' }); + const payload = makePullRequestPayload('closed'); + + const response = await app.request('http://localhost/api/webhooks/github', { + method: 'POST', + headers: { + 'x-github-delivery': 'delivery-task-history-failure', + 'x-github-event': 'pull_request', + 'x-hub-signature-256': 'sha256=test', + }, + body: JSON.stringify(payload), + }); + + expect(response.status).toBe(200); + expect(mockHandlePrMerge).toHaveBeenCalledWith(payload, { + includeFastParentTargets: false, + }); + }); + + it('restores direct fallback only for Fast tasks whose relay failed', async () => { + mockUpdateTaskPrStatus.mockResolvedValue(undefined); + mockUpsertGitHubPullRequestFactFromWebhook.mockResolvedValue(undefined); + mockRecordPrStatusChangeInTaskHistory.mockRejectedValue( + new MockPrStatusFastDeliveryError('relay unavailable', ['task-2']), + ); + mockHandlePrMerge.mockResolvedValue({ status: 'ok' }); + const payload = makePullRequestPayload('closed'); + + const response = await app.request('http://localhost/api/webhooks/github', { + method: 'POST', + headers: { + 'x-github-delivery': 'delivery-fast-relay-failure', + 'x-github-event': 'pull_request', + 'x-hub-signature-256': 'sha256=test', + }, + body: JSON.stringify(payload), + }); + + expect(response.status).toBe(200); + expect(mockHandlePrMerge).toHaveBeenCalledWith(payload, { + includeFastParentTargets: false, + includeFastParentTaskIds: ['task-2'], + }); }); it('does not notify linked tasks when terminal status persistence fails', async () => { @@ -1033,7 +1124,9 @@ describe('github webhook router', () => { status, }), ); - expect(mockHandlePrMerge).toHaveBeenCalledWith(payload); + expect(mockHandlePrMerge).toHaveBeenCalledWith(payload, { + includeFastParentTargets: false, + }); }, ); diff --git a/apps/api/src/handlers/github/__tests__/notifyPullRequestTerminalStatus.test.ts b/apps/api/src/handlers/github/__tests__/notifyPullRequestTerminalStatus.test.ts index f083292aa..3b342492a 100644 --- a/apps/api/src/handlers/github/__tests__/notifyPullRequestTerminalStatus.test.ts +++ b/apps/api/src/handlers/github/__tests__/notifyPullRequestTerminalStatus.test.ts @@ -85,6 +85,7 @@ vi.mock('@roomote/db/server', () => ({ slackInstallations: {}, githubInstallations: {}, taskPullRequests: {}, + desc: vi.fn((column: unknown) => ({ desc: column })), eq: vi.fn((...args: unknown[]) => ({ eq: args })), and: vi.fn((...args: unknown[]) => ({ and: args })), inArray: vi.fn((...args: unknown[]) => ({ inArray: args })), @@ -132,14 +133,22 @@ const discordPayload = { }; function fastParentSlackPayload(channelId: string, threadId: string) { + return fastParentPayload('slack', channelId, threadId); +} + +function fastParentPayload( + surface: 'slack' | 'teams' | 'telegram' | 'discord', + channelId: string, + threadId?: string, +) { return { fastAgentParent: { sessionId: '00000000-0000-4000-8000-000000000001', conversation: { - surface: 'slack', + surface, workspaceId: 'T123', - conversationId: threadId, - replyTarget: { channelId, threadId }, + conversationId: threadId ?? channelId, + replyTarget: { channelId, ...(threadId ? { threadId } : {}) }, }, }, }; @@ -270,60 +279,189 @@ describe('notifyPullRequestTerminalStatus', () => { expect(SLACK_PR_CLOSED_REACTION_EMOJI).toBe('-1'); }); - it.each([ - { - label: 'direct task-run', - payload: { - communicationProvider: 'slack', - communicationChannelId: 'CSHARED', - communicationThreadId: 'shared-thread-ts', + it('normalizes a direct task-run Slack binding through the delivery path', async () => { + mockedGithubFind.mockResolvedValue({ id: 1 } as any); + mockedTaskPullRequestsFind.mockResolvedValue([{ taskId: 'task-1' }] as any); + mockedTaskRunsFind.mockResolvedValue([ + { + taskId: 'task-1', + payload: { + communicationProvider: 'slack', + communicationChannelId: 'CSHARED', + communicationThreadId: 'shared-thread-ts', + }, }, - }, - { - label: 'Fast-parent', - payload: fastParentSlackPayload('CSHARED', 'shared-thread-ts'), - }, - ])( - 'normalizes $label Slack bindings through one delivery path', - async ({ payload }) => { - mockedGithubFind.mockResolvedValue({ id: 1 } as any); - mockedTaskPullRequestsFind.mockResolvedValue([ - { taskId: 'task-1' }, - ] as any); - mockedTaskRunsFind.mockResolvedValue([ - { taskId: 'task-1', payload }, - ] as any); - mockedSlackFind.mockResolvedValue({ - botAccessToken: 'xoxb-token', - } as any); - - await notifyPullRequestTerminalStatus({ - ...baseParams, - status: 'closed', - actorLogin: 'closer', - }); - - expect(mockStickyFooterPost).toHaveBeenCalledWith( - expect.objectContaining({ - channel: 'CSHARED', - threadTs: 'shared-thread-ts', - taskId: 'task-1', - }), - ); - expect(mockAddReaction).toHaveBeenCalledWith({ + ] as any); + mockedSlackFind.mockResolvedValue({ botAccessToken: 'xoxb-token' } as any); + + await notifyPullRequestTerminalStatus({ + ...baseParams, + status: 'closed', + actorLogin: 'closer', + }); + + expect(mockStickyFooterPost).toHaveBeenCalledWith( + expect.objectContaining({ channel: 'CSHARED', - timestamp: 'shared-thread-ts', - name: SLACK_PR_CLOSED_REACTION_EMOJI, - }); - expect(mockRemoveReaction).toHaveBeenCalledWith({ + threadTs: 'shared-thread-ts', + taskId: 'task-1', + }), + ); + }); + + it('does not post the canned status directly to a Fast parent conversation', async () => { + mockedGithubFind.mockResolvedValue({ id: 1 } as any); + mockedTaskPullRequestsFind.mockResolvedValue([{ taskId: 'task-1' }] as any); + mockedTaskRunsFind.mockResolvedValue([ + { + taskId: 'task-1', + payload: { + communicationProvider: 'slack', + communicationChannelId: 'CSHARED', + communicationThreadId: 'shared-thread-ts', + ...fastParentSlackPayload('CSHARED', 'shared-thread-ts'), + }, + }, + ] as any); + + await notifyPullRequestTerminalStatus(baseParams); + + expect(mockStickyFooterPost).not.toHaveBeenCalled(); + expect(mockAddReaction).not.toHaveBeenCalled(); + expect(mockRemoveReaction).not.toHaveBeenCalled(); + }); + + it('posts directly to the Fast parent when status recording failed', async () => { + mockedGithubFind.mockResolvedValue({ id: 1 } as any); + mockedTaskPullRequestsFind.mockResolvedValue([{ taskId: 'task-1' }] as any); + mockedTaskRunsFind.mockResolvedValue([ + { + taskId: 'task-1', + payload: fastParentSlackPayload('CSHARED', 'shared-thread-ts'), + }, + ] as any); + mockedSlackFind.mockResolvedValue({ botAccessToken: 'xoxb-token' } as any); + + await notifyPullRequestTerminalStatus({ + ...baseParams, + includeFastParentTargets: true, + }); + + expect(mockStickyFooterPost).toHaveBeenCalledWith( + expect.objectContaining({ channel: 'CSHARED', - timestamp: 'shared-thread-ts', - name: 'eyes', - }); - }, - ); + threadTs: 'shared-thread-ts', + text: 'Test PR was merged by merger', + }), + ); + expect(mockAddReaction).toHaveBeenCalledWith({ + channel: 'CSHARED', + timestamp: 'shared-thread-ts', + name: 'white_check_mark', + }); + }); - it('deduplicates an overlapping Fast-parent binding when cleanup rejects', async () => { + it('posts direct fallback only for the Fast task whose relay failed', async () => { + mockedGithubFind.mockResolvedValue({ id: 1 } as any); + mockedTaskPullRequestsFind.mockResolvedValue([ + { taskId: 'task-1' }, + { taskId: 'task-2' }, + ] as any); + mockedTaskRunsFind.mockResolvedValue([ + { + taskId: 'task-1', + payload: fastParentSlackPayload('C1', 'thread-1'), + }, + { + taskId: 'task-2', + payload: fastParentSlackPayload('C2', 'thread-2'), + }, + ] as any); + mockedSlackFind.mockResolvedValue({ botAccessToken: 'xoxb-token' } as any); + + await notifyPullRequestTerminalStatus({ + ...baseParams, + includeFastParentTaskIds: ['task-2'], + }); + + expect(mockStickyFooterPost).toHaveBeenCalledTimes(1); + expect(mockStickyFooterPost).toHaveBeenCalledWith( + expect.objectContaining({ channel: 'C2', threadTs: 'thread-2' }), + ); + expect(mockStickyFooterPost).not.toHaveBeenCalledWith( + expect.objectContaining({ channel: 'C1', threadTs: 'thread-1' }), + ); + }); + + it('does not fall back when a successful task shares the same Fast parent', async () => { + mockedGithubFind.mockResolvedValue({ id: 1 } as any); + mockedTaskPullRequestsFind.mockResolvedValue([ + { taskId: 'task-1' }, + { taskId: 'task-2' }, + ] as any); + mockedTaskRunsFind.mockResolvedValue([ + { + taskId: 'task-1', + payload: fastParentSlackPayload('C1', 'shared-thread'), + }, + { + taskId: 'task-2', + payload: fastParentSlackPayload('C1', 'shared-thread'), + }, + ] as any); + + await notifyPullRequestTerminalStatus({ + ...baseParams, + includeFastParentTaskIds: ['task-2'], + }); + + expect(mockStickyFooterPost).not.toHaveBeenCalled(); + expect(mockAddReaction).not.toHaveBeenCalled(); + }); + + it('ignores an older Fast payload when the latest run is no longer Fast', async () => { + mockedGithubFind.mockResolvedValue({ id: 1 } as any); + mockedTaskPullRequestsFind.mockResolvedValue([ + { taskId: 'task-1' }, + { taskId: 'task-2' }, + ] as any); + mockedTaskRunsFind.mockResolvedValue([ + { + id: 3, + taskId: 'task-1', + payload: { + communicationProvider: 'slack', + communicationChannelId: 'C-current', + communicationThreadId: 'current-thread', + }, + }, + { + id: 2, + taskId: 'task-2', + payload: fastParentSlackPayload('C-shared', 'shared-thread'), + }, + { + id: 1, + taskId: 'task-1', + payload: fastParentSlackPayload('C-shared', 'shared-thread'), + }, + ] as any); + mockedSlackFind.mockResolvedValue({ botAccessToken: 'xoxb-token' } as any); + + await notifyPullRequestTerminalStatus({ + ...baseParams, + includeFastParentTaskIds: ['task-2'], + }); + + expect(mockStickyFooterPost).toHaveBeenCalledWith( + expect.objectContaining({ + channel: 'C-shared', + threadTs: 'shared-thread', + }), + ); + }); + + it('suppresses a task-row Slack binding that matches the Fast parent', async () => { mockedGithubFind.mockResolvedValue({ id: 1 } as any); mockedTaskPullRequestsFind.mockResolvedValue([{ taskId: 'task-1' }] as any); mockedTasksFind.mockResolvedValue([ @@ -340,14 +478,38 @@ describe('notifyPullRequestTerminalStatus', () => { payload: fastParentSlackPayload('C123', 'thread-ts-1'), }, ] as any); + + await notifyPullRequestTerminalStatus(baseParams); + + expect(mockStickyFooterPost).not.toHaveBeenCalled(); + expect(mockAddReaction).not.toHaveBeenCalled(); + expect(mockRemoveReaction).not.toHaveBeenCalled(); + }); + + it('keeps a child-task Slack thread distinct from the Fast parent conversation', async () => { + mockedGithubFind.mockResolvedValue({ id: 1 } as any); + mockedTaskPullRequestsFind.mockResolvedValue([{ taskId: 'task-1' }] as any); + mockedTaskRunsFind.mockResolvedValue([ + { + taskId: 'task-1', + payload: { + communicationProvider: 'slack', + communicationChannelId: 'C123', + communicationThreadId: 'child-thread', + ...fastParentSlackPayload('C123', 'parent-thread'), + }, + }, + ] as any); mockedSlackFind.mockResolvedValue({ botAccessToken: 'xoxb-token' } as any); - mockRemoveReaction.mockRejectedValueOnce(new Error('Slack unavailable')); await notifyPullRequestTerminalStatus(baseParams); - expect(mockStickyFooterPost).toHaveBeenCalledTimes(1); - expect(mockAddReaction).toHaveBeenCalledTimes(1); - expect(mockRemoveReaction).toHaveBeenCalledTimes(1); + expect(mockStickyFooterPost).toHaveBeenCalledWith( + expect.objectContaining({ + channel: 'C123', + threadTs: 'child-thread', + }), + ); }); it('reports a rejected terminal reaction without failing the status post', async () => { @@ -453,6 +615,60 @@ describe('notifyPullRequestTerminalStatus', () => { }); }); + it('does not post provider-canned statuses to Fast parent conversations', async () => { + mockedGithubFind.mockResolvedValue({ id: 1 } as any); + mockedTaskPullRequestsFind.mockResolvedValue([{ taskId: 'task-1' }] as any); + mockedTaskRunsFind.mockResolvedValue([ + { + payload: { + ...teamsPayload, + ...fastParentPayload('teams', 'conversation-1', 'thread-1'), + }, + }, + { + payload: { + ...telegramPayload, + ...fastParentPayload('telegram', 'chat-1', 'thread-1'), + }, + }, + { + payload: { + ...discordPayload, + ...fastParentPayload('discord', 'channel-1', 'discord-thread-1'), + }, + }, + ] as any); + + await notifyPullRequestTerminalStatus(baseParams); + + expect(mockGetCommunicationProviderAdapter).not.toHaveBeenCalled(); + expect(mockPostMessage).not.toHaveBeenCalled(); + expect(mockAddReaction).not.toHaveBeenCalled(); + expect(mockRemoveReaction).not.toHaveBeenCalled(); + }); + + it('keeps a Discord child-task thread distinct from its Fast parent', async () => { + mockedGithubFind.mockResolvedValue({ id: 1 } as any); + mockedTaskPullRequestsFind.mockResolvedValue([{ taskId: 'task-1' }] as any); + mockedTaskRunsFind.mockResolvedValue([ + { + payload: { + ...discordPayload, + ...fastParentPayload('discord', 'channel-1', 'parent-thread'), + }, + }, + ] as any); + + await notifyPullRequestTerminalStatus(baseParams); + + expect(mockPostMessage).toHaveBeenCalledWith( + expect.objectContaining({ + channelId: 'channel-1', + threadId: 'discord-thread-1', + }), + ); + }); + it('keeps a bracketed PR tag outside the Discord link label', async () => { mockedGithubFind.mockResolvedValue({ id: 1 } as any); mockedTaskPullRequestsFind.mockResolvedValue([{ taskId: 'task-1' }] as any); diff --git a/apps/api/src/handlers/github/handlePrMerge.ts b/apps/api/src/handlers/github/handlePrMerge.ts index 16bb8c60f..8e89cca87 100644 --- a/apps/api/src/handlers/github/handlePrMerge.ts +++ b/apps/api/src/handlers/github/handlePrMerge.ts @@ -4,12 +4,13 @@ import type { WebhookPullRequestClosed } from './types'; import { scheduleNotifyPullRequestTerminalStatus } from './notifyPullRequestTerminalStatus'; import { toHostFromUrl } from '../utils'; -export const handlePrMerge = async ({ - installation, - repository, - pull_request, - sender, -}: WebhookPullRequestClosed): Promise => { +export const handlePrMerge = async ( + { installation, repository, pull_request, sender }: WebhookPullRequestClosed, + options: { + includeFastParentTargets?: boolean; + includeFastParentTaskIds?: string[]; + } = {}, +): Promise => { const status = pull_request.merged ? ('merged' as const) : ('closed' as const); @@ -30,6 +31,12 @@ export const handlePrMerge = async ({ actorLogin: (pull_request.merged ? pull_request.merged_by?.login : null) || sender.login, + ...(options.includeFastParentTargets + ? { includeFastParentTargets: true } + : {}), + ...(options.includeFastParentTaskIds?.length + ? { includeFastParentTaskIds: options.includeFastParentTaskIds } + : {}), }, `PR #${pull_request.number}`, ); diff --git a/apps/api/src/handlers/github/index.ts b/apps/api/src/handlers/github/index.ts index a3f61c976..bae6ea6c5 100644 --- a/apps/api/src/handlers/github/index.ts +++ b/apps/api/src/handlers/github/index.ts @@ -9,6 +9,8 @@ import { } from '@roomote/github'; import { recordPrStatusChangeInTaskHistory, + PrStatusFastDeliveryError, + PrStatusHistoryRecordingError, updateTaskPrStatus, upsertGitHubPullRequestFactFromWebhook, } from '@roomote/sdk/server'; @@ -638,8 +640,10 @@ github.post('/', async (c) => { // Persist merged/closed status into linked task history so agents get // the same out-of-band context path as PR review-feedback notifications. - void Promise.resolve( - recordPrStatusChangeInTaskHistory({ + let includeFastParentTargets = false; + let includeFastParentTaskIds: string[] = []; + try { + await recordPrStatusChangeInTaskHistory({ sourceControlProvider: 'github', repository: payload.repository.full_name, prNumber: payload.pull_request.number, @@ -650,18 +654,30 @@ github.post('/', async (c) => { (payload.pull_request.merged ? payload.pull_request.merged_by?.login : null) || payload.sender.login, - }), - ).catch((error) => { + }); + } catch (error) { + if (error instanceof PrStatusFastDeliveryError) { + includeFastParentTaskIds = error.taskIds; + } else { + includeFastParentTargets = !( + error instanceof PrStatusHistoryRecordingError + ); + } console.warn( `[pull_request.closed] Failed to record PR status in task history for ${payload.repository.full_name}#${payload.pull_request.number}: ${ error instanceof Error ? error.message : String(error) }`, ); - }); + } // Skipped repositories suppress automated review work, not lifecycle // notifications for tasks that already track this pull request. - return handlePrMerge(payload); + return handlePrMerge(payload, { + includeFastParentTargets, + ...(includeFastParentTaskIds.length + ? { includeFastParentTaskIds } + : {}), + }); }), ); diff --git a/apps/api/src/handlers/github/notifyPullRequestTerminalStatus.ts b/apps/api/src/handlers/github/notifyPullRequestTerminalStatus.ts index 18e762503..7de2157de 100644 --- a/apps/api/src/handlers/github/notifyPullRequestTerminalStatus.ts +++ b/apps/api/src/handlers/github/notifyPullRequestTerminalStatus.ts @@ -5,6 +5,7 @@ import { taskPullRequests, taskRuns, tasks, + desc, eq, and, inArray, @@ -114,6 +115,11 @@ interface NotifyPullRequestTerminalStatusParams { * not been updated yet. */ mergedBy?: string; + /** Preserve the direct Fast-conversation fallback when status recording or + * Fast event delivery failed before the terminal notifier was scheduled. */ + includeFastParentTargets?: boolean; + /** Restore direct fallback only for Fast tasks whose relay failed. */ + includeFastParentTaskIds?: string[]; } type SlackTarget = { @@ -127,28 +133,91 @@ type SlackReplyTarget = { threadId: string; }; -function resolveSlackReplyTarget(payload: unknown): SlackReplyTarget | null { - const directReplyTarget = - getCommunicationProviderFromTaskPayload(payload) === 'slack' - ? { - channelId: getCommunicationChannelFromTaskPayload(payload), - threadId: getCommunicationThreadIdFromTaskPayload(payload), - } - : null; - const fastConversation = getFastAgentParentFromPayload(payload)?.conversation; - const fastReplyTarget = - fastConversation?.surface === 'slack' ? fastConversation.replyTarget : null; +type DirectCommunicationProvider = 'slack' | 'teams' | 'telegram' | 'discord'; + +function isFastParentConversationTarget(params: { + payload: unknown; + provider: DirectCommunicationProvider; + channelId: string; + threadId?: string; +}): boolean { + const conversation = getFastAgentParentFromPayload( + params.payload, + )?.conversation; + if (!conversation || conversation.surface !== params.provider) { + return false; + } return ( - [directReplyTarget, fastReplyTarget].find( - (target): target is SlackReplyTarget => - Boolean(target?.channelId && target.threadId), - ) ?? null + conversation.replyTarget.channelId === params.channelId && + conversation.replyTarget.threadId === params.threadId ); } -function getSlackTarget(taskId: string, payload: unknown): SlackTarget | null { - const replyTarget = resolveSlackReplyTarget(payload); +function getFastParentConversationTargetKey(payload: unknown): string | null { + const conversation = getFastAgentParentFromPayload(payload)?.conversation; + if ( + !conversation || + conversation.surface === 'web' || + conversation.surface === 'automation' + ) { + return null; + } + + return [ + conversation.surface, + conversation.replyTarget.channelId, + conversation.replyTarget.threadId ?? '', + ].join('\0'); +} + +function resolveSlackReplyTarget( + payload: unknown, + includeFastParentTargets: boolean, +): SlackReplyTarget | null { + const isDirectSlackTarget = + getCommunicationProviderFromTaskPayload(payload) === 'slack'; + const channelId = isDirectSlackTarget + ? getCommunicationChannelFromTaskPayload(payload) + : undefined; + const threadId = isDirectSlackTarget + ? getCommunicationThreadIdFromTaskPayload(payload) + : undefined; + if (channelId && threadId) { + if ( + !includeFastParentTargets && + isFastParentConversationTarget({ + payload, + provider: 'slack', + channelId, + threadId, + }) + ) { + return null; + } + return { channelId, threadId }; + } + + const conversation = getFastAgentParentFromPayload(payload)?.conversation; + return includeFastParentTargets && + conversation?.surface === 'slack' && + conversation.replyTarget.threadId + ? { + channelId: conversation.replyTarget.channelId, + threadId: conversation.replyTarget.threadId, + } + : null; +} + +function getSlackTarget( + taskId: string, + payload: unknown, + includeFastParentTargets: boolean, +): SlackTarget | null { + const replyTarget = resolveSlackReplyTarget( + payload, + includeFastParentTargets, + ); if (!replyTarget) return null; return { @@ -180,7 +249,10 @@ type DiscordTarget = { }; }; -function getTeamsTarget(payload: unknown): TeamsTarget | null { +function getTeamsTarget( + payload: unknown, + includeFastParentTargets: boolean, +): TeamsTarget | null { if ( !payload || typeof payload !== 'object' || @@ -197,6 +269,17 @@ function getTeamsTarget(payload: unknown): TeamsTarget | null { } const threadId = getCommunicationThreadIdFromTaskPayload(payload); + if ( + !includeFastParentTargets && + isFastParentConversationTarget({ + payload, + provider: 'teams', + channelId, + ...(threadId ? { threadId } : {}), + }) + ) { + return null; + } return { channelId, @@ -205,7 +288,10 @@ function getTeamsTarget(payload: unknown): TeamsTarget | null { }; } -function getTelegramTarget(payload: unknown): TelegramTarget | null { +function getTelegramTarget( + payload: unknown, + includeFastParentTargets: boolean, +): TelegramTarget | null { if ( !payload || typeof payload !== 'object' || @@ -222,6 +308,17 @@ function getTelegramTarget(payload: unknown): TelegramTarget | null { const threadId = getCommunicationThreadIdFromTaskPayload(payload); const replyToMessageId = getCommunicationMessageIdFromTaskPayload(payload); + if ( + !includeFastParentTargets && + isFastParentConversationTarget({ + payload, + provider: 'telegram', + channelId: chatId, + ...(threadId ? { threadId } : {}), + }) + ) { + return null; + } return { chatId, @@ -230,7 +327,10 @@ function getTelegramTarget(payload: unknown): TelegramTarget | null { }; } -function getDiscordTarget(payload: unknown): DiscordTarget | null { +function getDiscordTarget( + payload: unknown, + includeFastParentTargets: boolean, +): DiscordTarget | null { if ( !payload || typeof payload !== 'object' || @@ -247,6 +347,17 @@ function getDiscordTarget(payload: unknown): DiscordTarget | null { const threadId = getCommunicationThreadIdFromTaskPayload(payload); const reactionTarget = getDiscordReactionTargetFromTaskPayload(payload); + if ( + !includeFastParentTargets && + isFastParentConversationTarget({ + payload, + provider: 'discord', + channelId, + ...(threadId ? { threadId } : {}), + }) + ) { + return null; + } return { channelId, @@ -749,6 +860,8 @@ export async function notifyPullRequestTerminalStatus({ status = 'merged', actorLogin, mergedBy, + includeFastParentTargets = false, + includeFastParentTaskIds = [], }: NotifyPullRequestTerminalStatusParams): Promise { const resolvedActorLogin = actorLogin || mergedBy || 'someone'; @@ -826,17 +939,76 @@ export async function notifyPullRequestTerminalStatus({ db.query.taskRuns.findMany({ where: inArray(taskRuns.taskId, taskIds), columns: { + id: true, taskId: true, payload: true, }, + orderBy: [desc(taskRuns.createdAt), desc(taskRuns.id)], }), ]); const slackTargets: SlackTarget[] = []; const linearSessionIds: string[] = []; + const fastFallbackTaskIds = new Set(includeFastParentTaskIds); + const latestRunByTaskId = new Map(); + for (const run of linkedRuns) { + if (!latestRunByTaskId.has(run.taskId)) { + latestRunByTaskId.set(run.taskId, run); + } + } + const latestRuns = [...latestRunByTaskId.values()]; + const latestRunIds = new Set(latestRuns.map((run) => run.id)); + const successfulFastConversationTargets = new Set( + includeFastParentTargets + ? [] + : latestRuns.flatMap((run) => { + if (fastFallbackTaskIds.has(run.taskId)) { + return []; + } + const targetKey = getFastParentConversationTargetKey(run.payload); + return targetKey ? [targetKey] : []; + }), + ); + const includeFastFallbackForRun = (run: (typeof linkedRuns)[number]) => { + if (includeFastParentTargets) { + return true; + } + if (!fastFallbackTaskIds.has(run.taskId)) { + return false; + } + if (!latestRunIds.has(run.id)) { + return false; + } + const targetKey = getFastParentConversationTargetKey(run.payload); + return !targetKey || !successfulFastConversationTargets.has(targetKey); + }; + const fastSlackConversationTargets = includeFastParentTargets + ? new Set() + : new Set( + latestRuns.flatMap((run) => { + if (fastFallbackTaskIds.has(run.taskId)) { + return []; + } + const conversation = getFastAgentParentFromPayload( + run.payload, + )?.conversation; + return conversation?.surface === 'slack' && + conversation.replyTarget.threadId + ? [ + `${conversation.replyTarget.channelId}\0${conversation.replyTarget.threadId}`, + ] + : []; + }), + ); for (const task of linkedTasks) { - if (task.slackThreadTs && task.slackChannelId) { + if ( + task.slackThreadTs && + task.slackChannelId && + !fastSlackConversationTargets.has( + `${task.slackChannelId}\0${task.slackThreadTs}`, + ) + ) { slackTargets.push({ taskId: task.id, slackThreadTs: task.slackThreadTs, @@ -851,20 +1023,30 @@ export async function notifyPullRequestTerminalStatus({ slackTargets.push( ...linkedRuns - .map((run) => getSlackTarget(run.taskId, run.payload)) + .map((run) => + getSlackTarget( + run.taskId, + run.payload, + includeFastFallbackForRun(run), + ), + ) .filter((target): target is SlackTarget => target !== null), ); const teamsTargets = linkedRuns - .map((run) => getTeamsTarget(run.payload)) + .map((run) => getTeamsTarget(run.payload, includeFastFallbackForRun(run))) .filter((target): target is TeamsTarget => target !== null); const telegramTargets = linkedRuns - .map((run) => getTelegramTarget(run.payload)) + .map((run) => + getTelegramTarget(run.payload, includeFastFallbackForRun(run)), + ) .filter((target): target is TelegramTarget => target !== null); const discordTargets = linkedRuns - .map((run) => getDiscordTarget(run.payload)) + .map((run) => + getDiscordTarget(run.payload, includeFastFallbackForRun(run)), + ) .filter((target): target is DiscordTarget => target !== null); if ( diff --git a/apps/api/src/handlers/gitlab/handleMergeRequest.ts b/apps/api/src/handlers/gitlab/handleMergeRequest.ts index 04d1bdddf..ba7d7a608 100644 --- a/apps/api/src/handlers/gitlab/handleMergeRequest.ts +++ b/apps/api/src/handlers/gitlab/handleMergeRequest.ts @@ -16,6 +16,8 @@ import { import { enqueueTask } from '@roomote/cloud-agents/server'; import { recordPrStatusChangeInTaskHistory, + PrStatusFastDeliveryError, + PrStatusHistoryRecordingError, updateTaskPrStatus, } from '@roomote/sdk/server'; @@ -59,6 +61,8 @@ async function notifyTerminalMergeRequestThreads( payload: GitLabMergeRequestWebhook, repoFullName: string, status: 'merged' | 'closed', + includeFastParentTargets: boolean, + includeFastParentTaskIds: string[], ): Promise { const webhookHost = toHostFromUrl(payload.object_attributes.url); const repositoryRows = await db.query.repositories.findMany({ @@ -87,6 +91,8 @@ async function notifyTerminalMergeRequestThreads( status, actorLogin: payload.user?.username ?? payload.user?.name ?? 'someone on GitLab', + ...(includeFastParentTargets ? { includeFastParentTargets: true } : {}), + ...(includeFastParentTaskIds.length ? { includeFastParentTaskIds } : {}), }, `MR !${payload.object_attributes.iid}`, ); @@ -128,8 +134,10 @@ export async function handleGitLabMergeRequest( }, }); - await Promise.resolve( - recordPrStatusChangeInTaskHistory({ + let includeFastParentTargets = false; + let includeFastParentTaskIds: string[] = []; + try { + await recordPrStatusChangeInTaskHistory({ sourceControlProvider: 'gitlab', repository: repoFullName, prNumber: mergeRequest.iid, @@ -138,17 +146,30 @@ export async function handleGitLabMergeRequest( status, actorLogin: payload.user?.username ?? payload.user?.name ?? 'someone on GitLab', - }), - ).catch((error) => { + }); + } catch (error) { + if (error instanceof PrStatusFastDeliveryError) { + includeFastParentTaskIds = error.taskIds; + } else { + includeFastParentTargets = !( + error instanceof PrStatusHistoryRecordingError + ); + } console.warn( `[handleGitLabMergeRequest] Failed to record PR status in task history for ${repoFullName}!${mergeRequest.iid}: ${ error instanceof Error ? error.message : String(error) }`, ); - }); + } if (mergeRequest.action === 'merge' || mergeRequest.action === 'close') { - await notifyTerminalMergeRequestThreads(payload, repoFullName, status); + await notifyTerminalMergeRequestThreads( + payload, + repoFullName, + status, + includeFastParentTargets, + includeFastParentTaskIds, + ); } return { status: 'ok' }; diff --git a/packages/sdk/src/server/index.ts b/packages/sdk/src/server/index.ts index 863996826..ef931be65 100644 --- a/packages/sdk/src/server/index.ts +++ b/packages/sdk/src/server/index.ts @@ -294,6 +294,8 @@ export * from './lib/task-runs/notify-fast-agent-parent-on-pull-request-conflict export { formatPrStatusChangeTaskHistoryText, formatPullRequestReference, + PrStatusFastDeliveryError, + PrStatusHistoryRecordingError, recordPrStatusChangeInTaskHistory, recordPrStatusChangeInTaskHistoryInputSchema, type RecordPrStatusChangeInTaskHistoryInput, diff --git a/packages/sdk/src/server/lib/fast-agent-parent-event.test.ts b/packages/sdk/src/server/lib/fast-agent-parent-event.test.ts index 2628a0fb1..09e0e06e3 100644 --- a/packages/sdk/src/server/lib/fast-agent-parent-event.test.ts +++ b/packages/sdk/src/server/lib/fast-agent-parent-event.test.ts @@ -168,7 +168,10 @@ vi.mock('./fast-automation-suggestions', () => ({ postFastAutomationSuggestionsToDiscord: mocks.postDiscordSuggestions, })); -import { deliverFastAgentParentEvent } from './fast-agent-parent-event'; +import { + deliverFastAgentParentEvent, + resolveFastAgentPlatformEventPolicy, +} from './fast-agent-parent-event'; const parent = { sessionId: '11111111-1111-4111-8111-111111111111', @@ -194,6 +197,28 @@ const event = { }, }; +describe('resolveFastAgentPlatformEventPolicy', () => { + it.each([ + ['pull_request_opened', 'slack', 'default', 'optional'], + ['pull_request_feedback', 'slack', 'present_only', 'required'], + ['pull_request_conflict_detected', 'discord', 'present_only', 'required'], + ['pull_request_status_changed', 'slack', 'default', 'optional'], + ['pull_request_status_changed', 'teams', 'default', 'optional'], + ['pull_request_status_changed', 'telegram', 'default', 'optional'], + ['pull_request_status_changed', 'discord', 'default', 'optional'], + ['pull_request_status_changed', 'web', 'default', 'optional'], + ['pull_request_status_changed', 'automation', 'default', 'optional'], + ['automation_triggered', 'automation', 'default', 'required'], + ] as const)( + 'routes %s on %s as %s/%s', + (eventType, surface, handling, visibility) => { + expect( + resolveFastAgentPlatformEventPolicy({ eventType, surface }), + ).toEqual({ handling, visibility }); + }, + ); +}); + describe('deliverFastAgentParentEvent', () => { beforeEach(() => { vi.clearAllMocks(); @@ -1509,6 +1534,8 @@ describe('deliverFastAgentParentEvent', () => { '"type":"pull_request_status_changed"', ), turnSource: 'platform_event', + platformEventHandling: 'default', + platformEventVisibility: 'optional', }), ); expect(firstClientMessageId).toEqual(expect.any(String)); diff --git a/packages/sdk/src/server/lib/fast-agent-parent-event.ts b/packages/sdk/src/server/lib/fast-agent-parent-event.ts index fc4384a9b..960d60310 100644 --- a/packages/sdk/src/server/lib/fast-agent-parent-event.ts +++ b/packages/sdk/src/server/lib/fast-agent-parent-event.ts @@ -8,6 +8,8 @@ import { createFastAgentWebTaskLauncher, fastAgentConversationRepository, resolveApiBaseUrl, + type FastAgentPlatformEventHandling, + type FastAgentPlatformEventVisibility, type FastAgentTurnAdapter, type LaunchFastAgentTask, } from '@roomote/cloud-agents/server'; @@ -201,6 +203,27 @@ export type FastAgentParentEvent = message: string; }; +export function resolveFastAgentPlatformEventPolicy(params: { + eventType: FastAgentParentEvent['type']; + surface: FastAgentConversation['surface']; +}): { + handling: FastAgentPlatformEventHandling; + visibility: FastAgentPlatformEventVisibility; +} { + if ( + params.eventType === 'pull_request_feedback' || + params.eventType === 'pull_request_conflict_detected' + ) { + return { handling: 'present_only', visibility: 'required' }; + } + + return { + handling: 'default', + visibility: + params.eventType === 'automation_triggered' ? 'required' : 'optional', + }; +} + export async function listFastAgentPullRequestContexts( taskId: string, ): Promise { @@ -1254,6 +1277,10 @@ export async function deliverFastAgentParentEvent(params: { // origin matches its own apiBaseUrl, so a mismatched pair silently drops // every deployment MCP server from parent-event turns. const apiBaseUrl = resolveApiBaseUrl() ?? undefined; + const eventPolicy = resolveFastAgentPlatformEventPolicy({ + eventType: params.event.type, + surface: parentTurn.conversation.surface, + }); await answerFastAgentQuestion({ question: `${JSON.stringify(params.event)}`, userId: parentTurn.userId, @@ -1262,17 +1289,8 @@ export async function deliverFastAgentParentEvent(params: { apiBaseUrl, signal: releaseTurnLock.signal, turnSource: 'platform_event', - platformEventHandling: - params.event.type === 'pull_request_feedback' || - params.event.type === 'pull_request_conflict_detected' - ? 'present_only' - : 'default', - platformEventVisibility: - params.event.type === 'pull_request_feedback' || - params.event.type === 'pull_request_conflict_detected' || - params.event.type === 'automation_triggered' - ? 'required' - : 'optional', + platformEventHandling: eventPolicy.handling, + platformEventVisibility: eventPolicy.visibility, platformEventKind: params.event.type === 'automation_triggered' ? 'automation' diff --git a/packages/sdk/src/server/lib/task-runs/__tests__/record-pr-status-change.test.ts b/packages/sdk/src/server/lib/task-runs/__tests__/record-pr-status-change.test.ts index 7d8c0f75c..e9e8dab21 100644 --- a/packages/sdk/src/server/lib/task-runs/__tests__/record-pr-status-change.test.ts +++ b/packages/sdk/src/server/lib/task-runs/__tests__/record-pr-status-change.test.ts @@ -53,6 +53,8 @@ import { import { formatPrStatusChangeTaskHistoryText, formatPullRequestReference, + PrStatusFastDeliveryError, + PrStatusHistoryRecordingError, recordPrStatusChangeInTaskHistory, } from '../record-pr-status-change'; @@ -268,14 +270,29 @@ describe('recordPrStatusChangeInTaskHistory', () => { ); }); - it('releases the claim when no linked task has a run', async () => { + it('skips the history claim when no linked task has a run', async () => { mockFindFirstTaskRun.mockResolvedValue(null); await expect(recordPrStatusChangeInTaskHistory(baseInput)).resolves.toEqual( { recordedTaskCount: 0, reason: 'no_task_runs' }, ); - expect(mockRedisDel).toHaveBeenCalled(); + expect(mockRedisSet).not.toHaveBeenCalled(); + expect(mockRedisDel).not.toHaveBeenCalled(); + expect(mockRecordTaskMessageEnvelope).not.toHaveBeenCalled(); + }); + + it('delivers to Fast before claiming independent task history', async () => { + mockRedisSet.mockRejectedValue(new Error('redis down')); + + await expect(recordPrStatusChangeInTaskHistory(baseInput)).rejects.toThrow( + PrStatusHistoryRecordingError, + ); + + expect(mockNotifyFastAgentParent).toHaveBeenCalledOnce(); + expect(mockNotifyFastAgentParent.mock.invocationCallOrder[0]).toBeLessThan( + mockRedisSet.mock.invocationCallOrder[0]!, + ); expect(mockRecordTaskMessageEnvelope).not.toHaveBeenCalled(); }); @@ -283,9 +300,55 @@ describe('recordPrStatusChangeInTaskHistory', () => { mockRecordTaskMessageEnvelope.mockRejectedValue(new Error('db down')); await expect(recordPrStatusChangeInTaskHistory(baseInput)).rejects.toThrow( - 'db down', + PrStatusHistoryRecordingError, ); expect(mockRedisDel).toHaveBeenCalled(); }); + + it('continues Fast delivery for later tasks after a history failure', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([ + { taskId: 'task-1' }, + { taskId: 'task-2' }, + ]); + mockFindFirstTaskRun + .mockResolvedValueOnce({ id: 11, taskId: 'task-1', payload: {} }) + .mockResolvedValueOnce({ id: 22, taskId: 'task-2', payload: {} }); + mockRecordTaskMessageEnvelope.mockRejectedValueOnce(new Error('db down')); + + await expect(recordPrStatusChangeInTaskHistory(baseInput)).rejects.toThrow( + PrStatusHistoryRecordingError, + ); + + expect(mockNotifyFastAgentParent).toHaveBeenCalledTimes(2); + expect(mockNotifyFastAgentParent).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + run: { id: 22, taskId: 'task-2', payload: {} }, + }), + ); + expect(mockRecordTaskMessageEnvelope).toHaveBeenCalledTimes(2); + }); + + it('reports only the later task whose Fast relay failed', async () => { + mockFindManyTaskPullRequests.mockResolvedValue([ + { taskId: 'task-1' }, + { taskId: 'task-2' }, + ]); + mockFindFirstTaskRun + .mockResolvedValueOnce({ id: 11, taskId: 'task-1', payload: {} }) + .mockResolvedValueOnce({ id: 22, taskId: 'task-2', payload: {} }); + mockNotifyFastAgentParent + .mockResolvedValueOnce(undefined) + .mockRejectedValueOnce(new Error('relay unavailable')); + mockRecordTaskMessageEnvelope.mockRejectedValueOnce(new Error('db down')); + + const error = await recordPrStatusChangeInTaskHistory(baseInput).catch( + (caught) => caught, + ); + + expect(error).toBeInstanceOf(PrStatusFastDeliveryError); + expect(error.taskIds).toEqual(['task-2']); + expect(mockNotifyFastAgentParent).toHaveBeenCalledTimes(2); + }); }); diff --git a/packages/sdk/src/server/lib/task-runs/record-pr-status-change.ts b/packages/sdk/src/server/lib/task-runs/record-pr-status-change.ts index 7929c0e4e..2fa7384c0 100644 --- a/packages/sdk/src/server/lib/task-runs/record-pr-status-change.ts +++ b/packages/sdk/src/server/lib/task-runs/record-pr-status-change.ts @@ -41,6 +41,22 @@ type RecordPrStatusChangeInTaskHistoryResult = { reason?: string; }; +/** Fast delivery completed, but the independent task-history persistence did + * not. Webhook callers should retry history without restoring direct Fast + * conversation delivery, which would duplicate the already-posted event. */ +export class PrStatusHistoryRecordingError extends Error {} + +/** Fast delivery failed for specific linked tasks. Webhook callers can restore + * direct delivery only for these targets without duplicating earlier tasks. */ +export class PrStatusFastDeliveryError extends Error { + readonly taskIds: string[]; + + constructor(message: string, taskIds: string[], options?: ErrorOptions) { + super(message, options); + this.taskIds = taskIds; + } +} + /** * Provider-native shorthand for a pull/merge request number. * GitHub/Gitea/Bitbucket use `#n`, GitLab uses `!n`, Azure DevOps has no @@ -156,44 +172,25 @@ export async function recordPrStatusChangeInTaskHistory( let recordedTaskCount = 0; let claimedTaskCount = 0; let skippedAlreadyRecorded = 0; + let historyError: PrStatusHistoryRecordingError | null = null; + let fastDeliveryCause: unknown; + const fastFallbackTaskIds: string[] = []; for (const taskId of taskIds) { - // Claim once per task so webhook redeliveries and mid-loop failures do not - // rewrite history for tasks that already succeeded. - const claimKey = buildStatusRecordedKey({ - sourceControlProvider, - repository: parsedInput.repository, - prNumber: parsedInput.prNumber, - status: parsedInput.status, - taskId, + const latestRun = await db.query.taskRuns.findFirst({ + where: eq(taskRuns.taskId, taskId), + orderBy: [desc(taskRuns.createdAt), desc(taskRuns.id)], + columns: { id: true, taskId: true, payload: true }, }); - const claim = await redis.set( - claimKey, - '1', - 'EX', - STATUS_RECORDED_TTL_SECONDS, - 'NX', - ); - if (claim !== 'OK') { - skippedAlreadyRecorded += 1; + if (!latestRun) { continue; } - claimedTaskCount += 1; - + // Fast delivery owns the user-visible event for Fast conversations and has + // its own durable delivery claim. Complete it before history's Redis claim + // so a Redis outage cannot suppress both the Fast event and direct fallback. try { - const latestRun = await db.query.taskRuns.findFirst({ - where: eq(taskRuns.taskId, taskId), - orderBy: [desc(taskRuns.createdAt)], - columns: { id: true, taskId: true, payload: true }, - }); - - if (!latestRun) { - await redis.del(claimKey).catch(() => undefined); - continue; - } - await notifyFastAgentParentOnPullRequestStatusChanged({ run: latestRun, pullRequest: { @@ -206,7 +203,45 @@ export async function recordPrStatusChangeInTaskHistory( }, actorLogin: parsedInput.actorLogin, }); + } catch (error) { + fastDeliveryCause ??= error; + fastFallbackTaskIds.push(taskId); + } + // Claim once per task so webhook redeliveries and mid-loop failures do not + // rewrite history for tasks that already succeeded. + const claimKey = buildStatusRecordedKey({ + sourceControlProvider, + repository: parsedInput.repository, + prNumber: parsedInput.prNumber, + status: parsedInput.status, + taskId, + }); + let claim: Awaited>; + try { + claim = await redis.set( + claimKey, + '1', + 'EX', + STATUS_RECORDED_TTL_SECONDS, + 'NX', + ); + } catch (error) { + historyError ??= new PrStatusHistoryRecordingError( + error instanceof Error ? error.message : String(error), + { cause: error }, + ); + continue; + } + + if (claim !== 'OK') { + skippedAlreadyRecorded += 1; + continue; + } + + claimedTaskCount += 1; + + try { await recordTaskMessageEnvelope({ runId: latestRun.id, taskId, @@ -240,10 +275,28 @@ export async function recordPrStatusChangeInTaskHistory( // Release so a later webhook delivery can retry this task instead of // staying silent for the full TTL after a transient write failure. await redis.del(claimKey).catch(() => undefined); - throw error; + historyError ??= new PrStatusHistoryRecordingError( + error instanceof Error ? error.message : String(error), + { cause: error }, + ); + continue; } } + if (fastFallbackTaskIds.length > 0) { + throw new PrStatusFastDeliveryError( + fastDeliveryCause instanceof Error + ? fastDeliveryCause.message + : String(fastDeliveryCause), + fastFallbackTaskIds, + { cause: fastDeliveryCause }, + ); + } + + if (historyError) { + throw historyError; + } + if (recordedTaskCount === 0) { if (skippedAlreadyRecorded > 0 && claimedTaskCount === 0) { return { recordedTaskCount: 0, reason: 'already_recorded' };