mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-08-15 03:34:59 +00:00
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 <shaojin.wensj@alibaba-inc.com> Co-authored-by: qwen-code-ci-bot <qwen-code-ci-bot@users.noreply.github.com> Co-authored-by: qwen-code-dev-bot <qwen-code-dev@service.alibaba.com>
This commit is contained in:
易良
GitHub
parent
e967cc9037
commit
eea0a2b3b2
3 changed files with 1448 additions and 58 deletions
|
|
@ -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/<workspace-scope>/` 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/<workspace-scope>/<channel>-<name-hash>-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
|
||||
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -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 = '<no-reply/>';
|
||||
|
|
@ -268,6 +377,12 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
|||
private pendingFinalDeliveryRequestsActive = 0;
|
||||
private pendingFinalDeliveryRetryStopRequested = false;
|
||||
private reasonFilter: Set<string> | null = null;
|
||||
private inboundRecoveryPending = true;
|
||||
private recoverableInboundTasks = 0;
|
||||
private activeInboundTaskIdsByMessage = new Map<string, string>();
|
||||
private cancelledInboundTaskIds = new Set<string>();
|
||||
private pendingCursorUpdatedAt: string | undefined;
|
||||
private inboundPersistenceBlocked = false;
|
||||
|
||||
constructor(
|
||||
name: string,
|
||||
|
|
@ -358,6 +473,15 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
|||
}
|
||||
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<GithubCursor> {
|
|||
);
|
||||
}
|
||||
|
||||
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<GithubCursor> {
|
|||
),
|
||||
});
|
||||
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<GithubCursor> {
|
|||
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<GithubCursor> {
|
|||
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<GithubCursor> {
|
|||
) {
|
||||
continue;
|
||||
}
|
||||
this.removeReplyPendingInboundTask(record);
|
||||
} catch (err) {
|
||||
if (signal?.aborted) return;
|
||||
if (isDefiniteNoWriteGithubError(err)) {
|
||||
|
|
@ -647,6 +785,7 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
|||
) {
|
||||
continue;
|
||||
}
|
||||
this.removeReplyPendingInboundTask(record);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -689,6 +828,29 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
|||
}
|
||||
}
|
||||
|
||||
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<GithubCursor> {
|
|||
}
|
||||
|
||||
protected async pollOnce(): Promise<void> {
|
||||
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<GithubCursor> {
|
|||
// 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<GithubCursor> {
|
|||
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<GithubCursor> {
|
|||
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<GithubCursor> {
|
|||
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<GithubCursor> {
|
|||
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<GithubCursor> {
|
|||
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<GithubCursor> {
|
|||
private async dispatchEnvelope(
|
||||
envelope: Envelope,
|
||||
issueNumber: number,
|
||||
dedupe: InboundTaskDedupe = {},
|
||||
): Promise<boolean> {
|
||||
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<boolean> {
|
||||
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<void> {
|
||||
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<string>();
|
||||
|
||||
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<string> {
|
||||
try {
|
||||
const keys = new Set<string>();
|
||||
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<GithubCursor> {
|
|||
private async postErrorComment(
|
||||
chatId: string,
|
||||
issueNumber: number,
|
||||
): Promise<void> {
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
await this.githubApi(
|
||||
() =>
|
||||
|
|
@ -1487,10 +1997,12 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
|||
}),
|
||||
`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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue