From eea0a2b3b203b8fb54ea13be6c962d1eae52c40d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=98=93=E8=89=AF?= <1204183885@qq.com> Date: Sun, 2 Aug 2026 15:27:44 +0800 Subject: [PATCH] fix(github-channel): recover interrupted inbound tasks (#8306) * fix(github-channel): recover interrupted inbound tasks * fix(github-channel): make inbound recovery bounded * test(github-channel): cover delivery-failure lifecycle and audit-hit recovery Add a direct test for the onTaskLifecycle failed/delivery -> reply_pending transition and a recovery test for the publication-audit match-and-remove path. Restore the blank line between the constructor and createInitialCursor. * fix(github-channel): preserve cancelled inbound tasks and fail closed on bookkeeping (#8306) * fix(github-channel): harden inbound task lifecycle against partial persistence failures (#8306) * fix(github-channel): close crash-window duplicates and make tests load-bearing (#8306) * test(github-channel): add recovery test for suppressed audit outcome (#8306) * fix(github-channel): persist errorCommentPosted after post, add review test coverage (#8306) --------- Co-authored-by: Shaojin Wen Co-authored-by: qwen-code-ci-bot Co-authored-by: qwen-code-dev-bot --- docs/users/features/channels/github.md | 20 +- .../channels/github/src/GithubAdapter.test.ts | 904 +++++++++++++++++- packages/channels/github/src/GithubAdapter.ts | 582 ++++++++++- 3 files changed, 1448 insertions(+), 58 deletions(-) diff --git a/docs/users/features/channels/github.md b/docs/users/features/channels/github.md index 1cd97744bc..73523f7156 100644 --- a/docs/users/features/channels/github.md +++ b/docs/users/features/channels/github.md @@ -82,7 +82,7 @@ Use `reasonFilter` to drop noisy notification classes such as `ci_activity` or ` Valid `reasonFilter` values are `mention`, `review_requested`, `assign`, `author`, `comment`, `ci_activity`, `manual`, `state_change`, `subscribed`, `team_mention`, `security_alert`, `approval_requested`, `invitation`, `member_feature_requested`, and `security_advisory_credit`. -Filtered notifications are still marked read before they are skipped. Removing the filter later will not replay notifications the channel already skipped. +Filtered notifications are marked read only after all accepted work in the poll window completes. Removing the filter later will not replay notifications the channel already skipped. ## ⚠️ Security @@ -101,12 +101,15 @@ The adapter detects mentions by scanning comment text and first-contact issue or The adapter uses GitHub's Notifications API as a wake-up signal: 1. **Poll** `GET /notifications` for unread threads -2. **Mark read** via `markNotificationsAsRead` (best-effort cleanup, before processing) -3. **Enumerate** comments via `listComments` within a cursor-based time window +2. **Enumerate** comments via `listComments` within a cursor-based time window +3. **Persist accepted work** before dispatch, including the source envelope and deduplication keys 4. **Dispatch** by notification reason: strict mention matching, pull request review, issue triage, followed-thread comment aggregation, or per-comment fallback -5. **First-contact fallback**: a brand-new unread issue/PR body can be processed when no comment was dispatched; mention notifications still require an actual body mention +5. **Commit the poll window** only after accepted work completes: mark notifications read and advance the cursor +6. **First-contact fallback**: a brand-new unread issue/PR body can be processed when no comment was dispatched; mention notifications still require an actual body mention -The comment window is `(previousCursor, currentMaxUpdatedAt]` — comments already eligible in a previous poll cycle are excluded by the cursor, preventing duplicate replies even when the async mark-read has not taken effect. If the process crashes mid-processing, the user can re-mention the bot to retry. +The comment window is `(previousCursor, currentMaxUpdatedAt]`. Accepted, running, and failed tasks are stored under `~/.qwen/channels//` with private file permissions. On restart, the channel recovers those tasks before polling GitHub again. Failed tasks are attempted up to three times, then become terminal; cancelled tasks are terminal and are not rerun. A task whose final reply was already posted, suppressed, or queued for definite no-write retry is not rerun. + +The notification cursor does not advance while recoverable tasks remain, or when inbound task state cannot be read or written. This prevents a crash or agent failure from losing an accepted comment and preserves the deduplication keys needed to avoid a second dispatch from the notification feed. Non-comment activity (push, label changes) bumps the notification's `updated_at` but produces zero new comments in the window, so re-fetched threads are skipped without triggering the agent. @@ -127,9 +130,10 @@ The GitHub channel always forces final-only delivery. The adapter sets `blockStr If GitHub returns a definite no-write delivery failure, such as a rate-limit response, the channel stores the final reply in `~/.qwen/channels//--github-pending-deliveries.json` -with private file permissions and retries it on the next channel start. -Ambiguous delivery failures are not retried automatically because GitHub may -have created the comment. +with private file permissions and retries it on the next channel start. The +corresponding inbound task remains in `reply_pending` state until that delivery +succeeds or reaches a definite terminal failure. Ambiguous delivery failures are +not retried automatically because GitHub may have created the comment. ## Known Limitations diff --git a/packages/channels/github/src/GithubAdapter.test.ts b/packages/channels/github/src/GithubAdapter.test.ts index 9a063a10bc..6fd8990142 100644 --- a/packages/channels/github/src/GithubAdapter.test.ts +++ b/packages/channels/github/src/GithubAdapter.test.ts @@ -171,6 +171,66 @@ function makeIssueEvent(overrides: Record = {}) { }; } +function inboundTaskPath( + cwd = '/tmp/test', + channelName = 'test-github', +): string { + const nameHash = createHash('sha256') + .update(channelName) + .digest('hex') + .slice(0, 16); + return join( + process.env.QWEN_HOME!, + 'channels', + getWorkspaceScopeDirName(cwd), + `${channelName}-${nameHash}-github-inbound-tasks.json`, + ); +} + +function makeInboundTaskRecord(overrides: Record = {}) { + const envelope = { + channelName: 'test-github', + senderId: 'alice', + senderName: 'alice', + chatId: 'owner/repo', + threadId: 'issue:42', + messageId: '1001', + text: 'please fix this', + isGroup: true, + isMentioned: true, + isReplyToBot: false, + metadata: 'Trigger: mention.', + }; + return { + version: 1, + id: 'inbound-task-1', + createdAt: '2026-07-02T10:00:00.000Z', + updatedAt: '2026-07-02T10:00:00.000Z', + state: 'accepted', + issueNumber: 42, + source: { + chatId: envelope.chatId, + threadId: envelope.threadId, + messageId: envelope.messageId, + }, + envelope, + dedupe: { dispatchedComments: ['C_1001'] }, + ...overrides, + }; +} + +function writeInboundTasks(records: unknown[]): void { + const path = inboundTaskPath(); + mkdirSync(join(path, '..'), { recursive: true }); + writeFileSync(path, `${JSON.stringify(records)}\n`, 'utf-8'); +} + +function readInboundTasks(): Array> { + return JSON.parse(readFileSync(inboundTaskPath(), 'utf-8')) as Array< + Record + >; +} + /** Subclass that captures envelopes instead of running the full ChannelBase pipeline. */ class TestableGithubChannel extends GithubChannel { inboundEnvelopes: Envelope[] = []; @@ -179,6 +239,7 @@ class TestableGithubChannel extends GithubChannel { sourceMessageId: string | undefined; sourceSenderId: string | undefined; sourceMetadata: string | undefined; + handleInboundHook: ((envelope: Envelope) => void | Promise) | undefined; protected getResponseMessageId(_sessionId: string): string | undefined { return this.sourceMessageId; @@ -193,11 +254,16 @@ class TestableGithubChannel extends GithubChannel { } override async handleInbound(envelope: Envelope): Promise { + await this.handleInboundHook?.(envelope); if (this.handleInboundError) throw this.handleInboundError; if (this.usePreflight && !(await this.preflightInbound(envelope))) return; this.inboundEnvelopes.push(envelope); } + triggerTaskLifecycleForTest(event: unknown): void { + this.onTaskLifecycle(event as never); + } + async testSendThreadMessage( chatId: string, threadId: string, @@ -271,6 +337,7 @@ describe('GithubChannel', () => { await channel.connect(); channel.disconnect(); channel.cursor = { lastProcessedAt: '2026-07-01T00:00:00.000Z' }; + vi.clearAllMocks(); } async function pollOnce() { @@ -509,7 +576,7 @@ describe('GithubChannel', () => { expect(channel.inboundEnvelopes[0]!.chatId).toBe('owner/repo'); }); - it('marks notifications as read before processing (best-effort)', async () => { + it('marks notifications as read after accepted work completes', async () => { const notification = makeNotification({ updated_at: '2026-07-02T10:00:00.000Z', }); @@ -517,7 +584,20 @@ describe('GithubChannel', () => { mockOctokit.paginate .mockResolvedValueOnce([notification]) .mockResolvedValueOnce([makeComment()]); - await pollOnce(); + let releaseInbound!: () => void; + channel.handleInboundHook = () => + new Promise((resolve) => { + releaseInbound = resolve; + }); + const poll = pollOnce(); + await vi.waitFor(() => + expect(mockOctokit.paginate).toHaveBeenCalledTimes(2), + ); + expect( + mockOctokit.rest.activity.markNotificationsAsRead, + ).not.toHaveBeenCalled(); + releaseInbound(); + await poll; expect( mockOctokit.rest.activity.markNotificationsAsRead, @@ -526,10 +606,12 @@ describe('GithubChannel', () => { read: true, }); const markOrder = - mockOctokit.rest.activity.markNotificationsAsRead.mock - .invocationCallOrder[0]!; - const commentOrder = mockOctokit.paginate.mock.invocationCallOrder[2]!; - expect(markOrder).toBeLessThan(commentOrder); + mockOctokit.rest.activity.markNotificationsAsRead.mock.invocationCallOrder.at( + -1, + )!; + const commentOrder = + mockOctokit.paginate.mock.invocationCallOrder.at(-1)!; + expect(markOrder).toBeGreaterThan(commentOrder); }); it('marks all fetched notifications read even on failure', async () => { @@ -646,10 +728,10 @@ describe('GithubChannel', () => { .mockResolvedValueOnce([makeComment()]); await pollOnce(); - // Call 1: initWithoutLoop's poll; call 2: listNotifications; - // call 3: listComments — the comment enumeration window. + // initWithoutLoop clears call history; call 1 lists notifications and call 2 + // enumerates comments using the durable cursor lower bound. expect(mockOctokit.paginate).toHaveBeenNthCalledWith( - 3, + 2, expect.anything(), expect.objectContaining({ since: '2026-07-01T00:00:00.000Z' }), ); @@ -1433,7 +1515,10 @@ describe('GithubChannel', () => { createHash('sha256').update(response).digest('hex'), ); expect(audit).not.toContain(response); - expect(JSON.parse(audit)).toMatchObject({ + const auditLines = audit.trim().split('\n'); + expect(JSON.parse(auditLines[0]!)).toMatchObject({ outcome: 'posting' }); + const lastAuditLine = auditLines.pop()!; + expect(JSON.parse(lastAuditLine)).toMatchObject({ outcome: 'posted', repository: 'owner/repo', number: 42, @@ -1494,7 +1579,8 @@ describe('GithubChannel', () => { expect(mockOctokit.rest.issues.createComment).toHaveBeenCalledTimes(1); expect(existsSync(pendingPath())).toBe(false); const audit = readFileSync(auditPath(), 'utf-8'); - expect(JSON.parse(audit)).toMatchObject({ + const lastAuditLine = audit.trim().split('\n').pop()!; + expect(JSON.parse(lastAuditLine)).toMatchObject({ outcome: 'failed', failurePhase: 'delivery', failureError: 'ambiguous transport failure', @@ -1761,7 +1847,20 @@ describe('GithubChannel', () => { }); it('drops and audits an ambiguous pending final retry failure', async () => { - writePending([pendingRecord({ triggerKind: 'mention' })]); + writePending([ + pendingRecord({ triggerKind: 'mention', sourceMessageId: '1001' }), + ]); + writeInboundTasks([ + makeInboundTaskRecord({ + state: 'reply_pending', + envelope: undefined, + source: { + chatId: 'owner/repo', + threadId: 'issue:42', + messageId: '1001', + }, + }), + ]); mockOctokit.rest.issues.createComment.mockRejectedValue( new Error('ambiguous'), ); @@ -1770,6 +1869,7 @@ describe('GithubChannel', () => { expect(mockOctokit.rest.issues.createComment).toHaveBeenCalledTimes(1); expect(existsSync(pendingPath())).toBe(false); + expect(existsSync(inboundTaskPath())).toBe(false); expect(JSON.parse(readFileSync(auditPath(), 'utf-8'))).toMatchObject({ outcome: 'failed', triggerKind: 'mention', @@ -2096,6 +2196,11 @@ describe('GithubChannel', () => { mockOctokit.rest.issues.createComment.mockRejectedValue( new Error('ambiguous transport failure'), ); + ( + channel as unknown as { + abortableSleep: (ms: number) => Promise; + } + ).abortableSleep = vi.fn().mockResolvedValue(undefined); const publish = ( channel as unknown as { publishFinalResponse: ( @@ -2489,7 +2594,7 @@ describe('GithubChannel', () => { ); }); - it('still marks thread as read after handleInbound failure', async () => { + it('keeps a failed inbound task recoverable and leaves the notification unread', async () => { channel.handleInboundError = new Error('agent down'); await initWithoutLoop(); mockOctokit.paginate @@ -2500,7 +2605,766 @@ describe('GithubChannel', () => { expect( mockOctokit.rest.activity.markNotificationsAsRead, - ).toHaveBeenCalledWith(expect.objectContaining({ read: true })); + ).not.toHaveBeenCalled(); + expect(channel.cursor.lastProcessedAt).toBe('2026-07-01T00:00:00.000Z'); + expect(channel.cursor.dispatchedComments).toEqual(['C_1001']); + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ + state: 'failed', + issueNumber: 42, + envelope: expect.objectContaining({ messageId: '1001' }), + dedupe: { dispatchedComments: ['C_1001'] }, + error: 'agent down', + }), + ]); + }); + + it('bounds failed task recovery and avoids duplicate error comments', async () => { + channel.handleInboundError = new Error('agent down'); + await initWithoutLoop(); + mockOctokit.paginate + .mockResolvedValueOnce([makeNotification()]) + .mockResolvedValueOnce([makeComment()]); + + await pollOnce(); + mockOctokit.paginate.mockResolvedValue([]); + await pollOnce(); + await pollOnce(); + + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ + state: 'failed', + attempts: 3, + errorCommentPosted: true, + }), + ]); + expect( + mockOctokit.rest.issues.createComment.mock.calls.filter((call) => + String(call[0]?.body).includes('Failed to process'), + ), + ).toHaveLength(1); + expect(channel.cursor.lastProcessedAt).toBe('2026-07-02T10:00:00.000Z'); + expect( + mockOctokit.rest.activity.markNotificationsAsRead, + ).toHaveBeenCalledWith({ + last_read_at: '2026-07-02T10:00:00.000Z', + read: true, + }); + }); + + it('retries the error comment when the first post fails', async () => { + channel.handleInboundError = new Error('agent down'); + await initWithoutLoop(); + ( + channel as unknown as { + abortableSleep: (ms: number) => Promise; + } + ).abortableSleep = vi.fn().mockResolvedValue(undefined); + mockOctokit.paginate + .mockResolvedValueOnce([makeNotification()]) + .mockResolvedValueOnce([makeComment()]); + mockOctokit.rest.issues.createComment.mockRejectedValue( + new Error('comment creation outage'), + ); + + await pollOnce(); + + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ + state: 'failed', + attempts: 1, + errorCommentPosted: false, + }), + ]); + + mockOctokit.rest.issues.createComment.mockResolvedValue({ data: {} }); + mockOctokit.paginate.mockResolvedValue([]); + await pollOnce(); + + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ + state: 'failed', + attempts: 2, + errorCommentPosted: true, + }), + ]); + expect( + mockOctokit.rest.issues.createComment.mock.calls.filter((call) => + String(call[0]?.body).includes('Failed to process'), + ), + ).toHaveLength(4); + }); + + it('reuses an existing inbound task record instead of creating a duplicate', async () => { + await initWithoutLoop(); + writeInboundTasks([ + makeInboundTaskRecord({ + state: 'failed', + attempts: 2, + errorCommentPosted: true, + }), + ]); + channel.handleInboundError = new Error('agent down'); + const privateChannel = channel as unknown as { + dispatchEnvelope: ( + envelope: Record, + issueNumber: number, + dedupe: Record, + ) => Promise; + }; + + await privateChannel.dispatchEnvelope( + { + channelName: 'test-github', + senderId: 'alice', + senderName: 'alice', + chatId: 'owner/repo', + threadId: 'issue:42', + messageId: '1001', + text: 'please fix this', + isGroup: true, + isMentioned: true, + isReplyToBot: false, + metadata: 'Trigger: mention.', + }, + 42, + { dispatchedComments: ['C_1001'] }, + ); + + const tasks = readInboundTasks(); + expect(tasks).toHaveLength(1); + expect(tasks[0]).toMatchObject({ + id: 'inbound-task-1', + state: 'failed', + attempts: 3, + errorCommentPosted: true, + }); + expect( + mockOctokit.rest.issues.createComment.mock.calls.filter((call) => + String(call[0]?.body).includes('Failed to process'), + ), + ).toHaveLength(0); + }); + + it('blocks cursor commit when inbound task state is invalid', async () => { + await initWithoutLoop(); + writeInboundTasks([ + makeInboundTaskRecord({ + dedupe: { dispatchedComments: 'C_1001' }, + }), + ]); + const privateChannel = channel as unknown as { + inboundRecoveryPending: boolean; + }; + privateChannel.inboundRecoveryPending = true; + mockOctokit.paginate + .mockResolvedValueOnce([ + makeNotification({ + updated_at: '2026-07-02T10:00:00.000Z', + }), + ]) + .mockResolvedValueOnce([]); + + await pollOnce(); + + expect( + mockOctokit.rest.activity.markNotificationsAsRead, + ).not.toHaveBeenCalled(); + expect(channel.cursor.lastProcessedAt).toBe('2026-07-01T00:00:00.000Z'); + }); + + it('persists cancellation as a terminal task state', async () => { + await initWithoutLoop(); + writeInboundTasks([makeInboundTaskRecord({ state: 'running' })]); + const privateChannel = channel as unknown as { + activeInboundTaskIdsByMessage: Map; + }; + privateChannel.activeInboundTaskIdsByMessage.set( + 'owner/repo|1001', + 'inbound-task-1', + ); + + channel.triggerTaskLifecycleForTest({ + type: 'cancelled', + chatId: 'owner/repo', + messageId: '1001', + }); + + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ state: 'cancelled' }), + ]); + }); + + it('does not turn cancellation into a retryable failure', async () => { + await initWithoutLoop(); + const task = makeInboundTaskRecord(); + writeInboundTasks([task]); + channel.handleInboundHook = async () => { + channel.triggerTaskLifecycleForTest({ + type: 'cancelled', + chatId: 'owner/repo', + messageId: '1001', + }); + throw new Error('cancelled'); + }; + const privateChannel = channel as unknown as { + runInboundTask: (task: typeof task) => Promise; + }; + + await privateChannel.runInboundTask(task); + + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ state: 'cancelled' }), + ]); + expect(mockOctokit.rest.issues.createComment).not.toHaveBeenCalledWith( + expect.objectContaining({ body: expect.stringContaining('Failed') }), + ); + }); + + it('does not transition a cancelled task to reply_pending on FinalPublicationError', async () => { + await initWithoutLoop(); + const task = makeInboundTaskRecord(); + writeInboundTasks([task]); + const rateLimitError = Object.assign(new Error('rate limited'), { + status: 429, + response: { headers: { 'x-ratelimit-remaining': '0' } }, + }); + mockOctokit.rest.issues.createComment.mockRejectedValue(rateLimitError); + channel.handleInboundHook = async (envelope) => { + channel.triggerTaskLifecycleForTest({ + type: 'cancelled', + chatId: 'owner/repo', + messageId: '1001', + }); + await ( + channel as unknown as { + publishFinalResponse: ( + chatId: string, + threadId: string, + text: string, + sessionId: string, + ) => Promise; + } + ).publishFinalResponse( + envelope.chatId, + envelope.threadId!, + 'cancelled response', + 'session-1', + ); + }; + channel.sourceMessageId = '1001'; + channel.sourceSenderId = 'alice'; + channel.sourceMetadata = 'Trigger: mention.'; + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + abortableSleep: (ms: number) => Promise; + runInboundTask: (task: Record) => Promise; + }; + privateChannel.octokit = mockOctokit as never; + privateChannel.abortableSleep = vi.fn().mockResolvedValue(undefined); + + await privateChannel.runInboundTask(task); + + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ state: 'cancelled' }), + ]); + }); + + it('continues polling after inbound recovery failure', async () => { + await initWithoutLoop(); + writeInboundTasks([makeInboundTaskRecord({ state: 'running' })]); + const pendingPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-pending-deliveries.json', + ); + mkdirSync(pendingPath, { recursive: true }); + mockOctokit.paginate + .mockResolvedValueOnce([makeNotification()]) + .mockResolvedValueOnce([makeComment()]); + + await pollOnce(); + + expect(channel.inboundEnvelopes.map((env) => env.messageId)).toEqual([ + '1001', + ]); + }); + + it('keeps the cancelled record when cancellation resolves normally', async () => { + await initWithoutLoop(); + const task = makeInboundTaskRecord(); + writeInboundTasks([task]); + channel.handleInboundHook = async () => { + channel.triggerTaskLifecycleForTest({ + type: 'cancelled', + chatId: 'owner/repo', + messageId: '1001', + }); + }; + const privateChannel = channel as unknown as { + runInboundTask: (task: typeof task) => Promise; + }; + + await privateChannel.runInboundTask(task); + + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ state: 'cancelled' }), + ]); + }); + + it('fails closed when post-success bookkeeping cannot read state', async () => { + await initWithoutLoop(); + const task = makeInboundTaskRecord(); + writeInboundTasks([task]); + const pendingPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-pending-deliveries.json', + ); + writeFileSync(pendingPath, '{not valid json', 'utf-8'); + const privateChannel = channel as unknown as { + runInboundTask: (task: typeof task) => Promise; + }; + + await expect(privateChannel.runInboundTask(task)).rejects.toThrow(); + + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ state: 'running' }), + ]); + expect(mockOctokit.rest.issues.createComment).not.toHaveBeenCalledWith( + expect.objectContaining({ body: expect.stringContaining('Failed') }), + ); + }); + + it('recovers an accepted task before polling and removes it after success', async () => { + writeInboundTasks([makeInboundTaskRecord()]); + channel.handleInboundHook = async () => { + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ state: 'running' }), + ]); + }; + mockOctokit.paginate.mockResolvedValue([]); + + await channel.connect(); + await vi.waitFor(() => { + expect(channel.inboundEnvelopes.map((item) => item.messageId)).toEqual([ + '1001', + ]); + }); + channel.disconnect(); + + expect(existsSync(inboundTaskPath())).toBe(false); + }); + + it('commits the recovered notification window after restart', async () => { + writeInboundTasks([makeInboundTaskRecord()]); + channel.cursor = { lastProcessedAt: '2026-07-01T00:00:00.000Z' }; + mockOctokit.paginate.mockResolvedValueOnce([ + makeNotification({ + last_read_at: '2026-07-02T09:00:00.000Z', + updated_at: '2026-07-02T10:00:00.000Z', + }), + ]); + + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + pollOnce: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + await privateChannel.pollOnce(); + + expect(existsSync(inboundTaskPath())).toBe(false); + expect(channel.cursor.lastProcessedAt).toBe('2026-07-02T10:00:00.000Z'); + expect( + mockOctokit.rest.activity.markNotificationsAsRead, + ).toHaveBeenCalledWith({ + last_read_at: '2026-07-02T10:00:00.000Z', + read: true, + }); + }); + + it('does not re-run a task whose reply was posted before a crash', async () => { + writeInboundTasks([makeInboundTaskRecord({ state: 'running' })]); + const auditFilePath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-audit.jsonl', + ); + mkdirSync(join(auditFilePath, '..'), { recursive: true }); + writeFileSync( + auditFilePath, + `${JSON.stringify({ + at: '2026-07-02T10:00:00.000Z', + type: 'github_publication', + outcome: 'posting', + channel: 'test-github', + repository: 'owner/repo', + number: 42, + sessionId: 'session-1', + threadId: 'issue:42', + sourceMessageId: '1001', + bodySha256: 'abc', + bodyChars: 5, + })}\n`, + ); + channel.cursor = { lastProcessedAt: '2026-07-01T00:00:00.000Z' }; + mockOctokit.paginate.mockResolvedValueOnce([]); + + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + pollOnce: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + await privateChannel.pollOnce(); + + expect(existsSync(inboundTaskPath())).toBe(false); + expect(channel.inboundEnvelopes).toHaveLength(0); + }); + + it('restores persisted dedupe before polling after recovery', async () => { + writeInboundTasks([makeInboundTaskRecord()]); + channel.cursor = { lastProcessedAt: '2026-07-01T00:00:00.000Z' }; + (channel as unknown as { botUsername: string }).botUsername = 'test-bot'; + mockOctokit.paginate + .mockResolvedValueOnce([ + makeNotification({ + last_read_at: '2026-07-02T09:00:00.000Z', + updated_at: '2026-07-02T10:00:00.000Z', + }), + ]) + .mockResolvedValueOnce([makeComment()]); + + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + pollOnce: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + await privateChannel.pollOnce(); + + expect(channel.inboundEnvelopes.map((item) => item.messageId)).toEqual([ + '1001', + ]); + }); + + it('keeps the inbound envelope recoverable when pending delivery persistence fails', async () => { + writeInboundTasks([makeInboundTaskRecord({ state: 'running' })]); + const pendingPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-pending-deliveries.json', + ); + mkdirSync(pendingPath, { recursive: true }); + const error = Object.assign(new Error('rate limited'), { + status: 429, + response: { headers: { 'x-ratelimit-remaining': '0' } }, + }); + mockOctokit.rest.issues.createComment.mockRejectedValue(error); + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + abortableSleep: (ms: number) => Promise; + runInboundTask: (task: Record) => Promise; + }; + privateChannel.octokit = mockOctokit as never; + channel.sourceMessageId = '1001'; + channel.sourceSenderId = 'alice'; + channel.sourceMetadata = 'Trigger: mention.'; + + vi.spyOn( + channel as unknown as { + postErrorComment: ( + chatId: string, + issueNumber: number, + ) => Promise; + }, + 'postErrorComment', + ).mockResolvedValue(true); + privateChannel.abortableSleep = vi.fn().mockResolvedValue(undefined); + channel.handleInboundHook = async (envelope) => { + await ( + channel as unknown as { + publishFinalResponse: ( + chatId: string, + threadId: string, + text: string, + sessionId: string, + ) => Promise; + } + ).publishFinalResponse( + envelope.chatId, + envelope.threadId!, + 'completed response', + 'session-1', + ); + }; + + await privateChannel.runInboundTask( + makeInboundTaskRecord({ state: 'running' }), + ); + + const [persistedTask] = readInboundTasks(); + expect(persistedTask).toMatchObject({ + state: 'failed', + envelope: { messageId: '1001' }, + }); + expect(String(persistedTask?.error)).toContain( + 'failed to persist pending GitHub delivery', + ); + }); + + it('does not rerun recovered work when delivery evidence is unreadable', async () => { + writeInboundTasks([makeInboundTaskRecord({ state: 'running' })]); + const pendingPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-pending-deliveries.json', + ); + mkdirSync(pendingPath, { recursive: true }); + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + pollOnce: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + + mockOctokit.paginate.mockResolvedValueOnce([]); + await privateChannel.pollOnce(); + + expect(channel.inboundEnvelopes).toHaveLength(0); + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ state: 'running' }), + ]); + }); + + it('does not rerun recovered work when publication audit is unreadable', async () => { + writeInboundTasks([makeInboundTaskRecord({ state: 'running' })]); + const auditPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-audit.jsonl', + ); + mkdirSync(auditPath, { recursive: true }); + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + pollOnce: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + + mockOctokit.paginate.mockResolvedValueOnce([]); + await privateChannel.pollOnce(); + + expect(channel.inboundEnvelopes).toHaveLength(0); + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ state: 'running' }), + ]); + }); + + it('removes a recovered task whose reply already has a posted audit record', async () => { + writeInboundTasks([makeInboundTaskRecord({ state: 'running' })]); + const auditPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-audit.jsonl', + ); + mkdirSync(join(auditPath, '..'), { recursive: true }); + writeFileSync( + auditPath, + `${JSON.stringify({ + outcome: 'posted', + repository: 'owner/repo', + threadId: 'issue:42', + sourceMessageId: '1001', + })}\n`, + 'utf-8', + ); + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + pollOnce: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + mockOctokit.paginate.mockResolvedValueOnce([]); + + await privateChannel.pollOnce(); + + expect(channel.inboundEnvelopes).toHaveLength(0); + expect(existsSync(inboundTaskPath())).toBe(false); + }); + + it('removes a recovered task whose reply has a suppressed audit record', async () => { + writeInboundTasks([makeInboundTaskRecord({ state: 'running' })]); + const auditPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-audit.jsonl', + ); + mkdirSync(join(auditPath, '..'), { recursive: true }); + writeFileSync( + auditPath, + `${JSON.stringify({ + outcome: 'suppressed', + repository: 'owner/repo', + threadId: 'issue:42', + sourceMessageId: '1001', + })}\n`, + 'utf-8', + ); + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + pollOnce: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + mockOctokit.paginate.mockResolvedValueOnce([]); + + await privateChannel.pollOnce(); + + expect(channel.inboundEnvelopes).toHaveLength(0); + expect(existsSync(inboundTaskPath())).toBe(false); + }); + + it('re-runs a task whose audit record is failed, not delivered', async () => { + writeInboundTasks([makeInboundTaskRecord({ state: 'running' })]); + const auditPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-audit.jsonl', + ); + mkdirSync(join(auditPath, '..'), { recursive: true }); + writeFileSync( + auditPath, + `${JSON.stringify({ + outcome: 'failed', + repository: 'owner/repo', + threadId: 'issue:42', + sourceMessageId: '1001', + })}\n`, + 'utf-8', + ); + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + pollOnce: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + mockOctokit.paginate.mockResolvedValueOnce([]); + + await privateChannel.pollOnce(); + + expect(channel.inboundEnvelopes.map((env) => env.messageId)).toEqual([ + '1001', + ]); + }); + + it('re-runs a task when the audit sourceMessageId does not match', async () => { + writeInboundTasks([makeInboundTaskRecord({ state: 'running' })]); + const auditPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-audit.jsonl', + ); + mkdirSync(join(auditPath, '..'), { recursive: true }); + writeFileSync( + auditPath, + `${JSON.stringify({ + outcome: 'posted', + repository: 'owner/repo', + threadId: 'issue:42', + sourceMessageId: '9999', + })}\n`, + 'utf-8', + ); + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + pollOnce: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + mockOctokit.paginate.mockResolvedValueOnce([]); + + await privateChannel.pollOnce(); + + expect(channel.inboundEnvelopes.map((env) => env.messageId)).toEqual([ + '1001', + ]); + }); + + it('removes a reply-pending task when a posted pending delivery is reconciled', async () => { + writeInboundTasks([ + makeInboundTaskRecord({ + state: 'reply_pending', + envelope: undefined, + }), + ]); + const pendingPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-pending-deliveries.json', + ); + const auditPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-audit.jsonl', + ); + const pending = { + id: 'pending-1', + createdAt: '2026-07-02T10:01:00.000Z', + chatId: 'owner/repo', + threadId: 'issue:42', + fullText: 'completed response', + sessionId: 'session-1', + sourceMessageId: '1001', + }; + writeFileSync(pendingPath, `${JSON.stringify([pending])}\n`, 'utf-8'); + writeFileSync( + auditPath, + `${JSON.stringify({ outcome: 'posted', pendingId: pending.id })}\n`, + 'utf-8', + ); + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + retryPendingFinalDeliveries: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + + await privateChannel.retryPendingFinalDeliveries(); + + expect(existsSync(pendingPath)).toBe(false); + expect(existsSync(inboundTaskPath())).toBe(false); + expect(mockOctokit.rest.issues.createComment).not.toHaveBeenCalled(); + }); + + it('does not rerun a task whose final reply is already pending delivery', async () => { + writeInboundTasks([ + makeInboundTaskRecord({ + state: 'running', + sessionId: 'session-1', + runId: 'run-1', + }), + ]); + const pendingPath = inboundTaskPath().replace( + 'github-inbound-tasks.json', + 'github-pending-deliveries.json', + ); + writeFileSync( + pendingPath, + `${JSON.stringify([ + { + id: 'pending-1', + createdAt: '2026-07-02T10:01:00.000Z', + chatId: 'owner/repo', + threadId: 'issue:42', + fullText: 'completed response', + sessionId: 'session-1', + sourceMessageId: '1001', + actor: 'alice', + triggerKind: 'mention', + }, + ])}\n`, + 'utf-8', + ); + const privateChannel = channel as unknown as { + octokit: typeof mockOctokit; + retryPendingFinalDeliveries: () => Promise; + pollOnce: () => Promise; + }; + privateChannel.octokit = mockOctokit as never; + mockOctokit.paginate.mockResolvedValueOnce([]); + await privateChannel.pollOnce(); + expect(channel.inboundEnvelopes).toHaveLength(0); + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ state: 'reply_pending' }), + ]); + + await privateChannel.retryPendingFinalDeliveries(); + + expect(mockOctokit.rest.issues.createComment).toHaveBeenCalledWith( + expect.objectContaining({ body: 'completed response' }), + ); + + expect(existsSync(inboundTaskPath())).toBe(false); }); it('posts only one error comment when dispatch fails on a new thread', async () => { @@ -2557,7 +3421,17 @@ describe('GithubChannel', () => { '1001', '1003', ]); - expect(channel.cursor.dispatchedComments).toEqual(['C_1001', 'C_1003']); + expect(channel.cursor.dispatchedComments).toEqual([ + 'C_1001', + 'C_1002', + 'C_1003', + ]); + expect(readInboundTasks()).toEqual([ + expect.objectContaining({ + state: 'failed', + envelope: expect.objectContaining({ messageId: '1002' }), + }), + ]); }); }); diff --git a/packages/channels/github/src/GithubAdapter.ts b/packages/channels/github/src/GithubAdapter.ts index f6b0559471..890e8dea5d 100644 --- a/packages/channels/github/src/GithubAdapter.ts +++ b/packages/channels/github/src/GithubAdapter.ts @@ -17,6 +17,7 @@ import type { ChannelAgentBridge, ChannelBaseOptions, ChannelConfig, + ChannelTaskLifecycleEvent, Envelope, } from '@qwen-code/channel-base'; import { @@ -158,7 +159,7 @@ interface PostedGithubComment { interface PublicationAuditRecord { at: string; type: 'github_publication'; - outcome: 'posted' | 'suppressed' | 'failed'; + outcome: 'posted' | 'suppressed' | 'failed' | 'posting'; channel: string; triggerKind?: string; repository: string; @@ -199,6 +200,114 @@ interface PendingFinalDelivery { triggerKind?: string; } +type InboundTaskState = + | 'accepted' + | 'running' + | 'reply_pending' + | 'failed' + | 'cancelled'; + +const MAX_INBOUND_TASK_ATTEMPTS = 3; + +interface InboundTaskDedupe { + dispatchedBodies?: string[]; + dispatchedComments?: string[]; + dispatchedEvents?: string[]; +} + +interface InboundTaskRecord { + version: 1; + id: string; + createdAt: string; + updatedAt: string; + state: InboundTaskState; + issueNumber: number; + source: { + chatId: string; + threadId?: string; + messageId?: string; + }; + envelope?: Envelope; + dedupe: InboundTaskDedupe; + attempts?: number; + errorCommentPosted?: boolean; + error?: string; +} + +function isOptionalStringArray(value: unknown): value is string[] | undefined { + return ( + value === undefined || + (Array.isArray(value) && value.every((item) => typeof item === 'string')) + ); +} + +function isInboundEnvelope(value: unknown): value is Envelope | undefined { + if (value === undefined) return true; + if (value === null || typeof value !== 'object') return false; + const envelope = value as Envelope; + return ( + typeof envelope.channelName === 'string' && + typeof envelope.senderId === 'string' && + typeof envelope.senderName === 'string' && + typeof envelope.chatId === 'string' && + typeof envelope.text === 'string' && + typeof envelope.isGroup === 'boolean' && + typeof envelope.isMentioned === 'boolean' && + typeof envelope.isReplyToBot === 'boolean' && + (envelope.chatName === undefined || + typeof envelope.chatName === 'string') && + (envelope.threadId === undefined || + typeof envelope.threadId === 'string') && + (envelope.messageId === undefined || + typeof envelope.messageId === 'string') && + (envelope.referencedText === undefined || + typeof envelope.referencedText === 'string') && + (envelope.imageBase64 === undefined || + typeof envelope.imageBase64 === 'string') && + (envelope.imageMimeType === undefined || + typeof envelope.imageMimeType === 'string') && + (envelope.attachments === undefined || + Array.isArray(envelope.attachments)) && + (envelope.metadata === undefined || + typeof envelope.metadata === 'string') && + (envelope.alreadyPrefixed === undefined || + envelope.alreadyPrefixed === true) + ); +} + +function isInboundTaskRecord(value: unknown): value is InboundTaskRecord { + if (value === null || typeof value !== 'object') return false; + const record = value as InboundTaskRecord; + return ( + record.version === 1 && + typeof record.id === 'string' && + typeof record.createdAt === 'string' && + typeof record.updatedAt === 'string' && + ['accepted', 'running', 'reply_pending', 'failed', 'cancelled'].includes( + record.state, + ) && + Number.isSafeInteger(record.issueNumber) && + record.source !== null && + typeof record.source === 'object' && + typeof record.source.chatId === 'string' && + (record.source.threadId === undefined || + typeof record.source.threadId === 'string') && + (record.source.messageId === undefined || + typeof record.source.messageId === 'string') && + isInboundEnvelope(record.envelope) && + record.dedupe !== null && + typeof record.dedupe === 'object' && + isOptionalStringArray(record.dedupe.dispatchedBodies) && + isOptionalStringArray(record.dedupe.dispatchedComments) && + isOptionalStringArray(record.dedupe.dispatchedEvents) && + (record.attempts === undefined || + (Number.isSafeInteger(record.attempts) && record.attempts >= 0)) && + (record.errorCommentPosted === undefined || + typeof record.errorCommentPosted === 'boolean') && + (record.error === undefined || typeof record.error === 'string') + ); +} + class FinalPublicationError extends Error {} const NO_REPLY_SENTINEL = ''; @@ -268,6 +377,12 @@ export class GithubChannel extends PollingChannelBase { private pendingFinalDeliveryRequestsActive = 0; private pendingFinalDeliveryRetryStopRequested = false; private reasonFilter: Set | null = null; + private inboundRecoveryPending = true; + private recoverableInboundTasks = 0; + private activeInboundTaskIdsByMessage = new Map(); + private cancelledInboundTaskIds = new Set(); + private pendingCursorUpdatedAt: string | undefined; + private inboundPersistenceBlocked = false; constructor( name: string, @@ -358,6 +473,15 @@ export class GithubChannel extends PollingChannelBase { } this.gate.replaceAllowedUsers(allowed); this.migrateLegacyPublicationState(); + this.inboundPersistenceBlocked = false; + this.inboundRecoveryPending = true; + try { + this.recoverableInboundTasks = this.readInboundTasks().filter((record) => + this.isRecoverableInboundTask(record), + ).length; + } catch { + this.recoverableInboundTasks = 0; + } this.pendingFinalDeliveryRetryStopRequested = false; this.startPollLoop(); if (this.pendingFinalDeliveryRetry) { @@ -457,6 +581,12 @@ export class GithubChannel extends PollingChannelBase { ); } + this.recordPublicationAudit({ + ...auditBase, + at: new Date().toISOString(), + type: 'github_publication', + outcome: 'posting', + }); try { const comment = await this.createIssueComment( chatId, @@ -486,12 +616,24 @@ export class GithubChannel extends PollingChannelBase { ), }); if (threadId && isDefiniteNoWriteGithubError(err)) { - this.enqueuePendingFinalDelivery({ - ...auditBase, - chatId, - threadId, - fullText, - }); + try { + this.enqueuePendingFinalDelivery({ + ...auditBase, + chatId, + threadId, + fullText, + }); + } catch (persistError) { + throw new Error( + `[Channel:${this.name}] failed to persist pending GitHub delivery: ${sanitizeLogText( + persistError instanceof Error + ? persistError.message + : String(persistError), + 200, + )}`, + { cause: persistError }, + ); + } } throw new FinalPublicationError( err instanceof Error ? err.message : String(err), @@ -552,19 +694,10 @@ export class GithubChannel extends PollingChannelBase { actor: input.actor, triggerKind: input.triggerKind, }; - try { - const pending = this.readPendingFinalDeliveries().filter( - (item) => item.id !== record.id, - ); - this.writePendingFinalDeliveries([...pending, record]); - } catch (err) { - process.stderr.write( - `[Channel:${this.name}] failed to persist pending GitHub delivery: ${sanitizeLogText( - err instanceof Error ? err.message : String(err), - 200, - )}\n`, - ); - } + const pending = this.readPendingFinalDeliveries(true).filter( + (item) => item.id !== record.id, + ); + this.writePendingFinalDeliveries([...pending, record]); } private async retryPendingFinalDeliveries( @@ -591,9 +724,13 @@ export class GithubChannel extends PollingChannelBase { 80, )}\n`, ); - this.updatePendingFinalDeliveries((current) => - current.filter((item) => item.id !== record.id), - ); + if ( + this.updatePendingFinalDeliveries((current) => + current.filter((item) => item.id !== record.id), + ) + ) { + this.removeReplyPendingInboundTask(record); + } continue; } try { @@ -624,6 +761,7 @@ export class GithubChannel extends PollingChannelBase { ) { continue; } + this.removeReplyPendingInboundTask(record); } catch (err) { if (signal?.aborted) return; if (isDefiniteNoWriteGithubError(err)) { @@ -647,6 +785,7 @@ export class GithubChannel extends PollingChannelBase { ) { continue; } + this.removeReplyPendingInboundTask(record); } } } @@ -689,6 +828,29 @@ export class GithubChannel extends PollingChannelBase { } } + private removeReplyPendingInboundTask(delivery: PendingFinalDelivery): void { + try { + const tasks = this.readInboundTasks(); + const matching = tasks.filter( + (task) => + task.state === 'reply_pending' && + task.source.chatId === delivery.chatId && + task.source.threadId === delivery.threadId && + task.source.messageId === delivery.sourceMessageId, + ); + for (const task of matching) { + this.removeInboundTask(task.id); + } + } catch (cleanupErr) { + process.stderr.write( + `[Channel:${this.name}] failed to clean up inbound task: ${sanitizeLogText( + cleanupErr instanceof Error ? cleanupErr.message : String(cleanupErr), + 200, + )}\n`, + ); + } + } + private pendingFinalDeliveriesPath(): string { return this.channelFilePath('github-pending-deliveries.json'); } @@ -874,6 +1036,19 @@ export class GithubChannel extends PollingChannelBase { } protected async pollOnce(): Promise { + this.inboundPersistenceBlocked = false; + if (this.inboundRecoveryPending) { + try { + await this.recoverInboundTasks(); + } catch (err) { + process.stderr.write( + `[Channel:${this.name}] inbound task recovery failed, will retry next poll: ${err}\n`, + ); + } finally { + this.inboundRecoveryPending = false; + } + } + this.cursor.metaFloor ??= this.cursor.lastProcessedAt; const since = new Date( new Date(this.cursor.lastProcessedAt).getTime() - 1000, @@ -901,12 +1076,13 @@ export class GithubChannel extends PollingChannelBase { // PUT /notifications' async mark fails to mark the thread read. const windowSince = this.cursor.lastProcessedAt; - await this.markNotificationsAsRead(maxUpdatedAt); - if (maxUpdatedAt > this.cursor.lastProcessedAt) { - this.cursor.lastProcessedAt = maxUpdatedAt; + this.pendingCursorUpdatedAt = + !this.pendingCursorUpdatedAt || + maxUpdatedAt > this.pendingCursorUpdatedAt + ? maxUpdatedAt + : this.pendingCursorUpdatedAt; } - for (const notification of notifications) { if (!notification.subject.url) continue; const extracted = this.extractFromSubjectUrl(notification.subject.url); @@ -972,6 +1148,21 @@ export class GithubChannel extends PollingChannelBase { continue; } } + if (this.hasRecoverableInboundTasks()) { + this.inboundRecoveryPending = true; + } + if ( + !this.inboundPersistenceBlocked && + !this.hasRecoverableInboundTasks() && + this.pendingCursorUpdatedAt + ) { + const committedAt = this.pendingCursorUpdatedAt; + await this.markNotificationsAsRead(committedAt); + if (committedAt > this.cursor.lastProcessedAt) { + this.cursor.lastProcessedAt = committedAt; + } + this.pendingCursorUpdatedAt = undefined; + } } private async processCommentLane( @@ -1008,7 +1199,11 @@ export class GithubChannel extends PollingChannelBase { metadata: this.buildRouteMetadata(ctx), }; - if (!(await this.dispatchEnvelope(envelope, ctx.issueNumber))) { + if ( + !(await this.dispatchEnvelope(envelope, ctx.issueNumber, { + dispatchedComments: [key], + })) + ) { dispatched = true; continue; } @@ -1057,7 +1252,9 @@ export class GithubChannel extends PollingChannelBase { isReplyToBot: false, metadata: `${this.buildMetadata(ctx.chatId, ctx.threadId, title)}\n${GITHUB_PUBLICATION_INSTRUCTIONS}\nTrigger: ${reason}.\n${buildTriggerGuidance(reason)}\n${details}`, }; - await this.dispatchEnvelope(envelope, ctx.issueNumber); + await this.dispatchEnvelope(envelope, ctx.issueNumber, { + dispatchedEvents: [trigger.key], + }); this.recordDispatched('dispatchedEvents', trigger.key); } @@ -1102,7 +1299,11 @@ export class GithubChannel extends PollingChannelBase { metadata: this.buildRouteMetadata(ctx), }; - await this.dispatchEnvelope(envelope, ctx.issueNumber); + await this.dispatchEnvelope(envelope, ctx.issueNumber, { + dispatchedComments: allComments.map( + (comment) => comment.node_id || String(comment.id), + ), + }); } private async findDirectTrigger( @@ -1252,7 +1453,9 @@ export class GithubChannel extends PollingChannelBase { metadata: this.buildRouteMetadata(ctx), }; - await this.dispatchEnvelope(envelope, issueNumber); + await this.dispatchEnvelope(envelope, issueNumber, { + dispatchedBodies: [bodyKey], + }); this.recordDispatchedBody(bodyKey); } catch (err) { process.stderr.write( @@ -1281,18 +1484,325 @@ export class GithubChannel extends PollingChannelBase { private async dispatchEnvelope( envelope: Envelope, issueNumber: number, + dedupe: InboundTaskDedupe = {}, ): Promise { + const task = this.claimInboundTask(envelope, issueNumber, dedupe); + return this.runInboundTask(task); + } + + private claimInboundTask( + envelope: Envelope, + issueNumber: number, + dedupe: InboundTaskDedupe, + ): InboundTaskRecord { + const existing = this.readInboundTasks().find( + (record) => + record.source.chatId === envelope.chatId && + record.source.threadId === envelope.threadId && + record.source.messageId === envelope.messageId, + ); + if (existing) { + this.applyTaskDedupe(existing); + return existing; + } + const now = new Date().toISOString(); + const task: InboundTaskRecord = { + version: 1, + id: randomUUID(), + createdAt: now, + updatedAt: now, + state: 'accepted', + issueNumber, + source: { + chatId: envelope.chatId, + threadId: envelope.threadId, + messageId: envelope.messageId, + }, + envelope, + dedupe, + attempts: 0, + }; + this.updateInboundTasks((records) => [...records, task]); + this.applyTaskDedupe(task); + return task; + } + + private async runInboundTask(task: InboundTaskRecord): Promise { + if (!this.isRecoverableInboundTask(task)) return true; + const envelope = task.envelope; + if (!envelope) return true; + const attempts = (task.attempts ?? 0) + 1; + this.activeInboundTaskIdsByMessage.set( + this.inboundMessageKey(envelope.chatId, envelope.messageId), + task.id, + ); + this.transitionInboundTask(task.id, 'running', { attempts }); + let cancelled = false; try { await this.handleInbound(envelope); - return true; + cancelled = this.cancelledInboundTaskIds.has(task.id); } catch (err) { + const error = sanitizeLogText( + err instanceof Error ? err.message : String(err), + 200, + ); process.stderr.write( `[Channel:${this.name}] handleInbound failed for ${envelope.messageId}: ${err}\n`, ); - if (!(err instanceof FinalPublicationError)) { - await this.postErrorComment(envelope.chatId, issueNumber); + if (err instanceof FinalPublicationError) { + if (this.cancelledInboundTaskIds.has(task.id)) { + // already persisted as 'cancelled' by onTaskLifecycle + } else if (this.hasPendingFinalDeliveryForTask(task)) { + this.transitionInboundTask(task.id, 'reply_pending', { + envelope: undefined, + error, + }); + } else { + this.removeInboundTask(task.id); + } + } else if (!this.cancelledInboundTaskIds.has(task.id)) { + let posted = task.errorCommentPosted === true; + if (!posted) { + posted = await this.postErrorComment( + envelope.chatId, + task.issueNumber, + ); + } + this.transitionInboundTask(task.id, 'failed', { + error, + attempts, + errorCommentPosted: posted, + }); } return false; + } finally { + this.activeInboundTaskIdsByMessage.delete( + this.inboundMessageKey(envelope.chatId, envelope.messageId), + ); + this.cancelledInboundTaskIds.delete(task.id); + } + // A base-class cancellation resolves handleInbound normally (the cancel is + // absorbed internally), so honour the terminal cancelled state captured + // above instead of removing the persisted record. Bookkeeping runs outside + // the try so a state read/write failure fails closed rather than being + // misclassified as a task failure that recovery would re-run. + if (cancelled) return true; + if (this.hasPendingFinalDeliveryForTask(task)) { + this.transitionInboundTask(task.id, 'reply_pending', { + envelope: undefined, + }); + } else { + this.removeInboundTask(task.id); + } + return true; + } + + private async recoverInboundTasks(): Promise { + const tasks = this.readInboundTasks().filter((task) => + this.isRecoverableInboundTask(task), + ); + const pendingDeliveries = tasks.length + ? this.readPendingFinalDeliveries(true) + : []; + const publicationAuditKeys = tasks.length + ? this.readPublicationAuditKeys(true) + : new Set(); + + for (const task of tasks) { + if (!task.envelope) continue; + this.applyTaskDedupe(task); + if ( + pendingDeliveries.some( + (record) => + record.chatId === task.source.chatId && + record.threadId === task.source.threadId && + record.sourceMessageId === task.source.messageId, + ) + ) { + this.transitionInboundTask(task.id, 'reply_pending', { + envelope: undefined, + }); + continue; + } + if (publicationAuditKeys.has(this.inboundTaskSourceKey(task))) { + this.removeInboundTask(task.id); + continue; + } + await this.runInboundTask(task); + } + } + + private isRecoverableInboundTask(task: InboundTaskRecord): boolean { + return ( + task.state === 'accepted' || + task.state === 'running' || + (task.state === 'failed' && + (task.attempts ?? 0) < MAX_INBOUND_TASK_ATTEMPTS) + ); + } + + private hasRecoverableInboundTasks(): boolean { + return this.recoverableInboundTasks > 0; + } + + private hasPendingFinalDeliveryForTask(task: InboundTaskRecord): boolean { + return this.readPendingFinalDeliveries(true).some( + (record) => + record.chatId === task.source.chatId && + record.threadId === task.source.threadId && + record.sourceMessageId === task.source.messageId, + ); + } + + private inboundTaskSourceKey(task: InboundTaskRecord): string { + return `${task.source.chatId}|${task.source.threadId ?? ''}|${task.source.messageId ?? ''}`; + } + + private readPublicationAuditKeys(strict = false): Set { + try { + const keys = new Set(); + for (const line of readFileSync( + this.channelFilePath('github-audit.jsonl'), + 'utf-8', + ).split('\n')) { + if (!line) continue; + try { + const record = JSON.parse(line) as PublicationAuditRecord; + if ( + record.outcome === 'posted' || + record.outcome === 'suppressed' || + record.outcome === 'posting' + ) { + keys.add( + `${record.repository}|${record.threadId ?? ''}|${record.sourceMessageId ?? ''}`, + ); + } + } catch { + continue; + } + } + return keys; + } catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') return new Set(); + if (strict) throw err; + return new Set(); + } + } + + private applyTaskDedupe(task: InboundTaskRecord): void { + for (const [field, keys] of Object.entries(task.dedupe) as Array< + [keyof InboundTaskDedupe, string[] | undefined] + >) { + for (const key of keys ?? []) { + this.recordDispatched(field, key); + } + } + } + + private transitionInboundTask( + taskId: string, + state: InboundTaskState, + updates: Partial< + Pick< + InboundTaskRecord, + 'envelope' | 'error' | 'attempts' | 'errorCommentPosted' + > + > = {}, + ): void { + this.updateInboundTasks((records) => + records.map((record) => + record.id === taskId + ? { + ...record, + ...updates, + state, + updatedAt: new Date().toISOString(), + } + : record, + ), + ); + } + + private removeInboundTask(taskId: string): void { + this.updateInboundTasks((records) => + records.filter((record) => record.id !== taskId), + ); + } + + private updateInboundTasks( + update: (records: InboundTaskRecord[]) => InboundTaskRecord[], + ): void { + try { + const records = update(this.readInboundTasks()); + this.writeInboundTasks(records); + this.recoverableInboundTasks = records.filter((record) => + this.isRecoverableInboundTask(record), + ).length; + } catch (err) { + this.inboundPersistenceBlocked = true; + throw err; + } + } + + private inboundTasksPath(): string { + return this.channelFilePath('github-inbound-tasks.json'); + } + + private readInboundTasks(): InboundTaskRecord[] { + try { + const parsed = JSON.parse(readFileSync(this.inboundTasksPath(), 'utf-8')); + if (!Array.isArray(parsed) || !parsed.every(isInboundTaskRecord)) { + throw new Error('invalid GitHub inbound task state'); + } + return parsed; + } catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') return []; + this.inboundPersistenceBlocked = true; + process.stderr.write( + `[Channel:${this.name}] failed to read GitHub inbound tasks: ${sanitizeLogText( + err instanceof Error ? err.message : String(err), + 200, + )}\n`, + ); + throw err; + } + } + + private writeInboundTasks(records: InboundTaskRecord[]): void { + const path = this.inboundTasksPath(); + if (records.length === 0) { + try { + unlinkSync(path); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== 'ENOENT') throw err; + } + return; + } + const dir = dirname(path); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + chmodSync(dir, 0o700); + const tmpPath = `${path}.${process.pid}.${randomUUID()}.tmp`; + writeFileSync(tmpPath, `${JSON.stringify(records)}\n`, { + encoding: 'utf-8', + mode: 0o600, + }); + chmodSync(tmpPath, 0o600); + renameSync(tmpPath, path); + chmodSync(path, 0o600); + } + + private inboundMessageKey(chatId: string, messageId?: string): string { + return `${chatId}|${messageId ?? ''}`; + } + + protected override onTaskLifecycle(event: ChannelTaskLifecycleEvent): void { + const taskId = this.activeInboundTaskIdsByMessage.get( + this.inboundMessageKey(event.chatId, event.messageId), + ); + if (!taskId) return; + if (event.type === 'cancelled') { + this.cancelledInboundTaskIds.add(taskId); + this.transitionInboundTask(taskId, 'cancelled'); } } @@ -1475,7 +1985,7 @@ export class GithubChannel extends PollingChannelBase { private async postErrorComment( chatId: string, issueNumber: number, - ): Promise { + ): Promise { try { await this.githubApi( () => @@ -1487,10 +1997,12 @@ export class GithubChannel extends PollingChannelBase { }), `postErrorComment(${chatId}#${issueNumber})`, ); + return true; } catch (err) { process.stderr.write( `[Channel:${this.name}] postErrorComment also failed for ${chatId}#${issueNumber}, user must re-mention manually: ${err}\n`, ); + return false; } } }