From 744948856cbd1f5246c8724ee98bf9fde2cb3ec6 Mon Sep 17 00:00:00 2001 From: Saqeb Akhter Date: Mon, 9 Nov 2026 06:15:00 +0400 Subject: [PATCH] fix(discord): isolate durable Notifications delivery safety (#218) --- docs/discord-bridge.md | 20 ++ src/discord-bridge.ts | 245 +++++++++++++++++--- src/discord-delivery.ts | 291 ++++++++++++++++++++++++ test/discord-bridge.test.ts | 14 +- test/discord-delivery.test.ts | 414 ++++++++++++++++++++++++++++++++++ 5 files changed, 949 insertions(+), 35 deletions(-) create mode 100644 src/discord-delivery.ts create mode 100644 test/discord-delivery.test.ts diff --git a/docs/discord-bridge.md b/docs/discord-bridge.md index 86210d3ea..b36cf9f27 100644 --- a/docs/discord-bridge.md +++ b/docs/discord-bridge.md @@ -135,3 +135,23 @@ The text explains `codeman notify "message" --urgent`. `codeman` is on the PATH | Posts arrive but replies do nothing | Message Content Intent is off, or your user ID is not in the allowlist | | Thread is not deleted | The bot lacks Manage Threads on the parent channel | | Nothing arrives on the phone | Check the Discord app's notification setting for the server | + +## Delivery retries and restarts + +Authorized inbound message IDs are durably claimed before attachment or terminal +writes. Transport acceptance is recorded as input accepted, not task completion. +An interrupted or unconfirmed input is never replayed automatically. A corrupt, +unsafe or full delivery journal pauses new input rather than forgetting old IDs. + +Versioned full-text responses record each outbound part and its payload +fingerprint separately. A retry skips confirmed parts; an ambiguous part pauses +that response rather than reporting it delivered or resending it blindly. REST +retries use one enforced nonce and respect Discord's actual retry delay. Thread +creation is not retried after an ambiguous network failure. An identical automatic +notification without a source version also pauses after uncertain delivery; a +separate, different notification can still proceed. Explicit deliberate notices +remain distinct user requests. + +`discord-delivery.json` is bounded, mode 0600 metadata with seven-day retention. +It contains IDs, hashes and fixed state/reason codes, not prompts, response text, +attachment bytes or credentials. diff --git a/src/discord-bridge.ts b/src/discord-bridge.ts index a87033710..152df91c6 100644 --- a/src/discord-bridge.ts +++ b/src/discord-bridge.ts @@ -1,3 +1,5 @@ +import { createHash, randomUUID } from 'node:crypto'; +import { DiscordDeliveryJournal } from './discord-delivery.js'; /** * @fileoverview Two-way Discord bridge: one thread per tab, alerts out, replies typed back in. * @@ -22,7 +24,6 @@ import { existsSync, mkdirSync } from 'node:fs'; import fs from 'node:fs/promises'; -import { randomUUID } from 'node:crypto'; import { join } from 'node:path'; import { probePathKind } from './utils/bounded-path-probe.js'; import { DISCORD_RECEIVED_DIR } from './web/paste-image-gc.js'; @@ -354,6 +355,8 @@ export function isDiscordCdnUrl(raw: string): boolean { } export interface DiscordPost { + nonce?: string; + enforce_nonce?: boolean; content: string; allowed_mentions: { parse: string[]; users: string[] }; } @@ -420,6 +423,7 @@ function withTimeout(p: Promise, ms: number): Promise { /** A failure with a reason worth showing as it is (unlike a network error, whose text is not). */ class BridgeProblem extends Error {} +class UncertainDiscordDelivery extends BridgeProblem {} export async function readDiscordConfig(configDir: string): Promise { try { @@ -490,7 +494,8 @@ export class DiscordHttpError extends Error { constructor( readonly status: number, /** Discord's own error code (10003 unknown channel, 50001 missing access ...). */ - readonly code?: number + readonly code?: number, + readonly retryAfterMs?: number ) { super(`Discord HTTP ${status}${code ? ` (code ${code})` : ''}`); } @@ -504,24 +509,54 @@ export class DiscordApi { private readonly sleep: (ms: number) => Promise = (ms) => new Promise((r) => setTimeout(r, ms)) ) {} - async request(method: string, path: string, body?: unknown, retried = false): Promise { + async request( + method: string, + path: string, + body?: unknown, + attempt = 0, + current?: () => boolean, + deferRateLimit = false + ): Promise { + if (current && !current()) return null; const isForm = typeof FormData !== 'undefined' && body instanceof FormData; - const res = await this.fetchImpl(new URL(`${API}${path}`), { - method, - headers: { - Authorization: `Bot ${this.token()}`, - // A form body sets its own multipart Content-Type (with the boundary). - ...(isForm ? {} : { 'Content-Type': 'application/json' }), - 'User-Agent': 'DiscordBot (https://github.com/Ark0N/Codeman, 1)', - }, - body: body === undefined ? undefined : isForm ? (body as FormData) : JSON.stringify(body), - redirect: 'manual', - signal: AbortSignal.timeout(SEND_TIMEOUT_MS), - }); - if (res.status === 429 && !retried) { - const j = (await res.json().catch(() => ({}))) as { retry_after?: number }; - await this.sleep(Math.min(Math.max((j.retry_after ?? 1) * 1000, 100), 10_000)); - return this.request(method, path, body, true); + const payload = isForm ? JSON.parse(String((body as FormData).get('payload_json') ?? '{}')) : body; + const retrySafe = method !== 'POST' || !!(payload as DiscordPost | undefined)?.enforce_nonce; + let res: Response; + try { + res = await this.fetchImpl(new URL(`${API}${path}`), { + method, + headers: { + Authorization: `Bot ${this.token()}`, + // A form body sets its own multipart Content-Type (with the boundary). + ...(isForm ? {} : { 'Content-Type': 'application/json' }), + 'User-Agent': 'DiscordBot (https://github.com/Ark0N/Codeman, 1)', + }, + body: body === undefined ? undefined : isForm ? (body as FormData) : JSON.stringify(body), + redirect: 'manual', + signal: AbortSignal.timeout(SEND_TIMEOUT_MS), + }); + } catch (err) { + // Creating a thread/channel has no idempotency key: a lost response must not create a duplicate. + if (!retrySafe || attempt >= 2) throw err; + await this.sleep(500 * 2 ** attempt); + return this.request(method, path, body, attempt + 1, current, deferRateLimit); + } + if (res.status === 429) { + const j = (await res.json().catch(() => ({}))) as { retry_after?: number; code?: number }; + const headerSeconds = Number(res.headers.get('retry-after')); + const seconds = Math.max( + typeof j.retry_after === 'number' && Number.isFinite(j.retry_after) && j.retry_after > 0 ? j.retry_after : 0, + Number.isFinite(headerSeconds) && headerSeconds > 0 ? headerSeconds : 0 + ); + const delay = seconds > 0 ? Math.max(seconds * 1000, 100) : 1000; + // Never retry earlier than Discord permits. A longer wait is surfaced instead of blocking a session chain. + if (deferRateLimit || attempt >= 2 || delay > 30_000) throw new DiscordHttpError(429, j.code, delay); + await this.sleep(delay); + return this.request(method, path, body, attempt + 1, current, deferRateLimit); + } + if (res.status >= 500 && retrySafe && attempt < 2) { + await this.sleep(500 * 2 ** attempt); + return this.request(method, path, body, attempt + 1, current, deferRateLimit); } if (!res.ok) { const j = (await res.json().catch(() => ({}))) as { code?: number }; @@ -535,14 +570,28 @@ export class DiscordApi { // type 11 = public thread. A week of inactivity before Discord archives it. return this.request('POST', `/channels/${parentId}/threads`, { name, type: 11, auto_archive_duration: 10080 }); } - post(channelId: string, p: DiscordPost): Promise<{ id: string } | null> { - return this.request('POST', `/channels/${channelId}/messages`, p); + post(channelId: string, p: DiscordPost, current?: () => boolean): Promise<{ id: string } | null> { + return this.request( + 'POST', + `/channels/${channelId}/messages`, + { ...p, nonce: p.nonce ?? randomUUID().replaceAll('-', '').slice(0, 24), enforce_nonce: true }, + 0, + current + ); } - postWithFile(channelId: string, p: DiscordPost, file: DiscordFile): Promise<{ id: string } | null> { + postWithFile( + channelId: string, + p: DiscordPost, + file: DiscordFile, + current?: () => boolean + ): Promise<{ id: string } | null> { const form = new FormData(); - form.append('payload_json', JSON.stringify(p)); + form.append( + 'payload_json', + JSON.stringify({ ...p, nonce: p.nonce ?? randomUUID().replaceAll('-', '').slice(0, 24), enforce_nonce: true }) + ); form.append('files[0]', new Blob([Buffer.from(file.data)]), safeAttachmentName(file.name)); - return this.request('POST', `/channels/${channelId}/messages`, form); + return this.request('POST', `/channels/${channelId}/messages`, form, 0, current); } rename(threadId: string, name: string): Promise { return this.request('PATCH', `/channels/${threadId}`, { name }); @@ -854,12 +903,19 @@ export class DiscordBridge { private last: DiscordResult | null = null; private readonly api: DiscordApi; private readonly now: () => number; + private readonly deliveries: DiscordDeliveryJournal; + private stopped = false; constructor(private readonly deps: DiscordBridgeDeps) { this.now = deps.now ?? Date.now; + this.deliveries = new DiscordDeliveryJournal(deps.configDir, this.now); this.api = new DiscordApi(() => this.cfg.token, deps.fetchImpl, deps.sleep); } + recentDeliveries() { + return this.deliveries.recent(); + } + get lastResult(): DiscordResult | null { return this.last; } @@ -958,6 +1014,7 @@ export class DiscordBridge { /** (Re)read the config and start or stop the gateway to match. Call at boot and after a save. */ async reload(): Promise { + this.stopped = false; this.cfg = await readDiscordConfig(this.deps.configDir); // The thread map is read once, by whoever gets here first; everyone else waits for that read. A second // read finishing later would replace the map and drop threads made in between. @@ -997,6 +1054,7 @@ export class DiscordBridge { } stop(): void { + this.stopped = true; clearInterval(this.sweepTimer); this.sweepTimer = undefined; this.gateway?.stop(); @@ -1096,14 +1154,16 @@ export class DiscordBridge { for (const [k, v] of this.lastSent) if (t - v > DEDUPE_WINDOW_MS) this.lastSent.delete(k); } let chunks: string[] = []; - // Remembered only once the first post is out: a failed post must leave the message to be sent again. + // Confirm every durable response part independently; ambiguous output is not replayed. let posted: string | undefined; + let fullDeliveryVersion: string | undefined; if (msg.fullText) { try { const got = await withTimeout(msg.fullText(), FULL_TEXT_TIMEOUT_MS); const text = got?.text.trim(); const sid = msg.sessionId; if (text && sid) { + fullDeliveryVersion = got?.version; // Keyed on the text and the turn it came from, so a later turn that ends on the same words is posted. const key = `${got?.version ?? ''}\u0000${text}`; msg = { ...msg, tail: undefined }; @@ -1117,6 +1177,12 @@ export class DiscordBridge { } } const post = buildPost(msg, this.cfg, explicitUrgent); + const deliveryKey = + opts.deliberate || !posted + ? randomUUID() + : createHash('sha256') + .update(JSON.stringify([msg.event, msg.sessionId, fullDeliveryVersion ?? posted])) + .digest('hex'); // One post when the head and the first part fit together, the rest follow without a ping. if (chunks.length > 0 && post.content.length + 1 + chunks[0].length <= MAX_CONTENT) { post.content += `\n${chunks.shift()}`; @@ -1131,7 +1197,22 @@ export class DiscordBridge { if (typeof thread !== 'string') return (this.last = thread); target = thread; } - const send = (to: string) => (msg.file ? this.api.postWithFile(to, post, msg.file) : this.api.post(to, post)); + const scope = JSON.stringify(this.cfg); + const current = (to: string) => + !this.stopped && + this.cfg.enabled && + scope === JSON.stringify(this.cfg) && + (!msg.sessionId || this.threads[msg.sessionId] === to); + const send = (to: string) => + this.deliverPost( + to, + post, + msg.file, + msg.sessionId, + deliveryKey + ':0', + () => current(to), + !posted && !opts.deliberate + ); if (posted && msg.sessionId) { claimed = { sid: msg.sessionId, before: this.lastFull.get(msg.sessionId) }; this.lastFull.set(msg.sessionId, posted); @@ -1149,10 +1230,20 @@ export class DiscordBridge { throw err; } } - // Parts 2 to 8 are best effort on purpose: the first post, which carries the title, is what is reported. - for (const part of chunks) await this.say(this.threads[msg.sessionId ?? ''] ?? target, part); + // Every part is confirmed separately; a trailing failure must not report full delivery. + for (const [i, part] of chunks.entries()) + await this.deliverPost( + this.threads[msg.sessionId ?? ''] ?? target, + { content: part, allowed_mentions: { parse: [], users: [] } }, + undefined, + msg.sessionId, + deliveryKey + ':' + (i + 1), + () => current(this.threads[msg.sessionId ?? ''] ?? target) + ); return (this.last = { ok: true, at: this.now() }); } catch (err) { + if (prev === undefined) this.lastSent.delete(key); + else this.lastSent.set(key, prev); if (claimed) { if (claimed.before === undefined) this.lastFull.delete(claimed.sid); else this.lastFull.set(claimed.sid, claimed.before); @@ -1233,6 +1324,72 @@ export class DiscordBridge { else if (this.cfg.onSessionDelete === 'archive') await this.api.archive(thread).catch(() => undefined); } + private async deliverPost( + channelId: string, + post: DiscordPost, + file: DiscordFile | undefined, + sessionId: string | undefined, + key: string, + current?: () => boolean, + checkUnversioned = false + ): Promise { + const id = createHash('sha256').update(`${this.cfg.guildId}:${channelId}:${key}`).digest('hex'); + const hash = createHash('sha256').update( + JSON.stringify({ + post, + file: Array.isArray(file) ? file.map((f) => [f.name, f.data.byteLength]) : file?.name, + }) + ); + if (Array.isArray(file)) for (const f of file) hash.update(f.data); + else if (file) hash.update(file.data); + const payloadHash = hash.digest('hex'); + if ( + checkUnversioned && + (await this.deliveries.uncertainPayload(payloadHash, { sessionId, channelId, guildId: this.cfg.guildId })) + ) + throw new UncertainDiscordDelivery( + 'An identical notification has uncertain delivery; review it before resending' + ); + const claim = await this.deliveries.claim({ + id, + direction: 'outbound', + sessionId, + channelId, + guildId: this.cfg.guildId, + payloadHash, + }); + if (!claim.fresh && claim.record.payloadHash !== payloadHash) + throw new BridgeProblem('Response formatting changed since the previous delivery; review it before resending'); + if (!claim.fresh && claim.record.state === 'delivered') return { id: claim.record.messageId }; + if (!claim.fresh && ['uncertain', 'submitting'].includes(claim.record.state)) { + throw new BridgeProblem('Previous Discord delivery is uncertain; it will not be replayed automatically'); + } + await this.deliveries.update(id, 'submitting'); + if (current && !current()) { + await this.deliveries.update(id, 'failed', 'scope_changed'); + throw new BridgeProblem('Discord destination changed'); + } + const payload = { ...post, nonce: id.slice(0, 24), enforce_nonce: true }; + try { + const result = file + ? await this.api.postWithFile(channelId, payload, file, current) + : await this.api.post(channelId, payload, current); + if (!result?.id) throw new BridgeProblem('Discord did not confirm delivery'); + await this.deliveries.annotate(id, { messageId: result.id }); + await this.deliveries.update(id, 'delivered'); + return result; + } catch (err) { + const refused = err instanceof DiscordHttpError && err.status < 500; + await this.deliveries.update( + id, + refused ? 'failed' : 'uncertain', + refused ? 'discord_refused' : 'delivery_uncertain' + ); + if (!refused) throw new UncertainDiscordDelivery('Discord delivery was not confirmed'); + throw err; + } + } + private say(channelId: string, content: string): Promise { return this.api.post(channelId, { content, allowed_mentions: { parse: [], users: [] } }).catch(() => undefined); } @@ -1302,8 +1459,26 @@ export class DiscordBridge { const sessionId = Object.keys(this.threads).find((k) => this.threads[k] === m.channel_id); if (!sessionId) return; const t = this.now(); + const scope = JSON.stringify(this.cfg); + const current = () => + !this.stopped && + this.cfg.enabled && + scope === JSON.stringify(this.cfg) && + this.threads[sessionId] === m.channel_id; + const deliveryId = createHash('sha256') + .update(JSON.stringify([m.guild_id, m.channel_id, m.id])) + .digest('hex'); + const claim = await this.deliveries.claim({ + id: deliveryId, + direction: 'inbound', + sessionId, + channelId: m.channel_id, + guildId: m.guild_id!, + }); + if (!claim.fresh) return; const prev = this.lastReply.get(sessionId); if (prev !== undefined && t - prev < REPLY_MIN_INTERVAL_MS) { + await this.deliveries.update(deliveryId, 'failed', 'rate_limited'); // Too soon after the last reply: say it was not typed, instead of dropping it silently. await this.api.react(m.channel_id, m.id, '⏳').catch(() => undefined); return; @@ -1313,6 +1488,7 @@ export class DiscordBridge { return; } this.lastReply.set(sessionId, t); + if (!current()) return; const saved = await this.saveAttachments(sessionId, m); const text = sanitiseReply(m.content ?? '', saved.paths); if (!text) { @@ -1322,11 +1498,22 @@ export class DiscordBridge { if (saved.refused) await this.say(m.channel_id, saved.refused); // A lone digit answers a dialog by itself: an Enter after it would land on whatever comes next. const keys = /^\d\r$/.test(text) && this.deps.awaitingChoice?.(sessionId) ? text.slice(0, -1) : text; + if (!current() || this.deps.agentExited?.(sessionId)) { + await this.deliveries.update(deliveryId, 'failed', 'scope_changed'); + return; + } + await this.deliveries.update(deliveryId, 'submitting'); + if (!current() || this.deps.agentExited?.(sessionId)) { + await this.deliveries.update(deliveryId, 'failed', 'scope_changed'); + return; + } const ok = await this.deps.sendInput(sessionId, keys).catch(() => false); + await this.deliveries.update(deliveryId, ok ? 'input_accepted' : 'uncertain', ok ? undefined : 'input_uncertain'); const reply = ok ? this.api.react(m.channel_id, m.id, '✅') : this.api.post(m.channel_id, { - content: 'That tab is gone, so nothing was typed.', + content: + 'Input was not confirmed. Check the terminal before retrying; it will not be replayed automatically.', allowed_mentions: { parse: [], users: [] }, }); await reply.catch(() => undefined); diff --git a/src/discord-delivery.ts b/src/discord-delivery.ts new file mode 100644 index 000000000..88685deda --- /dev/null +++ b/src/discord-delivery.ts @@ -0,0 +1,291 @@ +/** @fileoverview + * Private, bounded Discord delivery metadata. No prompt, file bytes, token or message body is stored. + * A durable claim precedes input; interrupted input is uncertain and is never replayed automatically. + */ +import { randomUUID } from 'node:crypto'; +import fs from 'node:fs/promises'; +import { join } from 'node:path'; + +export const DISCORD_DELIVERY_LIMIT = 2048; +const RETENTION_MS = 7 * 24 * 60 * 60 * 1000; +const MAX_BYTES = 2 * 1024 * 1024; +export class DiscordDeliveryError extends Error {} +const STATES = [ + 'received', + 'awaiting_choice', + 'queued', + 'cancelled', + 'submitted', + 'files_transferred', + 'submitting', + 'input_accepted', + 'running', + 'completed', + 'interrupted', + 'execution_error', + 'delivered', + 'failed', + 'uncertain', +] as const; +export type DiscordDeliveryState = (typeof STATES)[number]; +export interface DiscordDelivery { + id: string; + direction: 'inbound' | 'outbound'; + sessionId?: string; + channelId: string; + guildId: string; + state: DiscordDeliveryState; + /** Outbound formatting fingerprint; guards multipart retries after a formatter upgrade. */ + payloadHash?: string; + messageId?: string; + createdAt: number; + updatedAt: number; + /** Fixed codes only. Never store external exception messages or prompt text here. */ + reason?: + | 'restart_before_submit' + | 'restart_during_submit' + | 'session_unavailable' + | 'input_uncertain' + | 'empty' + | 'transfer_failed' + | 'scope_changed' + | 'discord_refused' + | 'delivery_uncertain' + | 'rate_limited' + | 'input_too_large' + | 'invalid_command'; +} + +export class DiscordDeliveryJournal { + private records = new Map(); + private loadRun?: Promise; + private chain: Promise = Promise.resolve(); + + constructor( + private readonly dir: string, + private readonly now: () => number = Date.now + ) {} + + private serial(fn: () => Promise): Promise { + const run = this.chain.then(fn); + this.chain = run.catch(() => undefined); + return run; + } + + load(): Promise { + this.loadRun ??= this.read(); + return this.loadRun; + } + + private async read(): Promise { + const path = join(this.dir, 'discord-delivery.json'); + let raw: unknown; + try { + const info = await fs.lstat(path); + if (!info.isFile() || info.isSymbolicLink() || info.size > MAX_BYTES) + throw new DiscordDeliveryError('Unsafe Discord delivery journal'); + raw = JSON.parse(await fs.readFile(path, 'utf8')); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') return; + // Losing a deduplication journal cannot silently turn a replay into another terminal write. + throw new DiscordDeliveryError('Discord delivery journal cannot be read; input is paused'); + } + if (!Array.isArray(raw) || raw.length > DISCORD_DELIVERY_LIMIT) + throw new DiscordDeliveryError('Invalid Discord delivery journal'); + const cutoff = this.now() - RETENTION_MS; + for (const r of raw) { + if ( + !r || + typeof r.id !== 'string' || + r.id.length > 200 || + !['inbound', 'outbound'].includes(r.direction) || + typeof r.channelId !== 'string' || + !/^\d{15,25}$/.test(r.channelId) || + typeof r.guildId !== 'string' || + !/^\d{15,25}$/.test(r.guildId) || + !STATES.includes(r.state) || + !Number.isFinite(r.createdAt) || + !Number.isFinite(r.updatedAt) || + (r.sessionId !== undefined && (typeof r.sessionId !== 'string' || r.sessionId.length > 100)) + ) + throw new DiscordDeliveryError('Invalid Discord delivery journal'); + if (r.createdAt < cutoff) continue; + // Copy only recognized metadata. Extra data from disk must never leak through the status API. + const record: DiscordDelivery = { + id: r.id, + direction: r.direction, + sessionId: r.sessionId, + channelId: r.channelId, + guildId: r.guildId, + state: r.state, + createdAt: r.createdAt, + updatedAt: r.updatedAt, + }; + // Retain only this fixed refusal proof; never restore arbitrary external reason text. + if (r.reason === 'discord_refused') record.reason = 'discord_refused'; + if (r.payloadHash !== undefined) { + if (typeof r.payloadHash !== 'string' || !/^[a-f0-9]{64}$/.test(r.payloadHash)) + throw new DiscordDeliveryError('Invalid Discord delivery fingerprint'); + record.payloadHash = r.payloadHash; + } + if (r.messageId !== undefined) { + if (typeof r.messageId !== 'string' || r.messageId.length > 100) + throw new DiscordDeliveryError('Invalid Discord message id'); + record.messageId = r.messageId; + } + if (['received', 'files_transferred', 'awaiting_choice', 'submitting'].includes(record.state)) { + record.reason = record.state === 'submitting' ? 'restart_during_submit' : 'restart_before_submit'; + record.state = record.reason === 'restart_during_submit' ? 'uncertain' : 'failed'; + record.updatedAt = this.now(); + } + this.records.set(record.id, record); + } + } + + private async persist(): Promise { + await fs.mkdir(this.dir, { recursive: true }); + const destination = join(this.dir, 'discord-delivery.json'); + const tmp = `${destination}.${randomUUID()}.tmp`; + let file; + try { + file = await fs.open(tmp, 'wx', 0o600); + await file.writeFile(JSON.stringify([...this.records.values()])); + await file.sync(); + await file.close(); + file = undefined; + await fs.rename(tmp, destination); + } finally { + await file?.close().catch(() => undefined); + await fs.unlink(tmp).catch(() => undefined); + } + } + + claim( + record: Omit + ): Promise<{ fresh: boolean; record: DiscordDelivery }> { + return this.serial(async () => { + await this.load(); + const existing = this.records.get(record.id); + if (existing) { + // A definitive refusal created no message. A corrected formatter may retry + // that same destination/key, but never rewrite accepted or uncertain output. + if ( + existing.direction === 'outbound' && + record.direction === 'outbound' && + existing.state === 'failed' && + existing.reason === 'discord_refused' && + !existing.messageId && + existing.sessionId === record.sessionId && + existing.guildId === record.guildId && + existing.channelId === record.channelId && + record.payloadHash && + existing.payloadHash !== record.payloadHash + ) { + const revised = { ...existing, payloadHash: record.payloadHash, updatedAt: this.now() }; + this.records.set(record.id, revised); + try { + await this.persist(); + } catch (err) { + this.records.set(record.id, existing); + throw err; + } + return { fresh: false, record: { ...revised } }; + } + return { fresh: false, record: { ...existing } }; + } + const before = new Map(this.records); + const now = this.now(); + for (const [id, r] of this.records) if (r.createdAt < now - RETENTION_MS) this.records.delete(id); + const completedPosts = [...this.records.values()].filter( + (r) => r.direction === 'outbound' && ['delivered', 'failed'].includes(r.state) + ); + // Routine notifications must not crowd out inbound deduplication records. + for (const r of completedPosts.slice(0, Math.max(0, completedPosts.length - 255))) this.records.delete(r.id); + // Do not evict recent inbound IDs to make room: saturation pauses input instead of weakening dedupe. + if (this.records.size >= DISCORD_DELIVERY_LIMIT) + throw new DiscordDeliveryError('Discord delivery journal is full'); + const next: DiscordDelivery = { ...record, state: 'received', createdAt: now, updatedAt: now }; + this.records.set(next.id, next); + try { + await this.persist(); + } catch (err) { + this.records = before; + throw err; + } + return { fresh: true, record: { ...next } }; + }); + } + + update(id: string, state: DiscordDeliveryState, reason?: DiscordDelivery['reason']): Promise { + return this.serial(async () => { + await this.load(); + const before = this.records.get(id); + if (!before) throw new DiscordDeliveryError('Unknown Discord delivery'); + this.records.set(id, { ...before, state, reason, updatedAt: this.now() }); + try { + await this.persist(); + } catch (err) { + this.records.set(id, before); + throw err; + } + }); + } + + annotate(id: string, metadata: Pick): Promise { + return this.serial(async () => { + await this.load(); + const before = this.records.get(id); + if (!before) throw new DiscordDeliveryError('Unknown delivery'); + this.records.set(id, { ...before, ...metadata, updatedAt: this.now() }); + try { + await this.persist(); + } catch (err) { + this.records.set(id, before); + throw err; + } + }); + } + /** An unversioned automatic notice cannot distinguish a new send from a retry. Fail closed on an identical uncertain payload. */ + async uncertainPayload( + payloadHash: string, + scope: { sessionId?: string; channelId: string; guildId: string } + ): Promise { + await this.load(); + await this.chain; + return [...this.records.values()].some( + (r) => + r.direction === 'outbound' && + r.payloadHash === payloadHash && + r.sessionId === scope.sessionId && + r.channelId === scope.channelId && + r.guildId === scope.guildId && + r.createdAt > this.now() - RETENTION_MS && + (r.state === 'uncertain' || r.state === 'submitting') + ); + } + async deliveredMessage( + id: string, + scope: { sessionId: string; guildId: string; channelId: string } + ): Promise { + await this.load(); + await this.chain; + const record = this.records.get(id); + return record?.direction === 'outbound' && + record.state === 'delivered' && + record.sessionId === scope.sessionId && + record.guildId === scope.guildId && + record.channelId === scope.channelId && + record.createdAt > this.now() - RETENTION_MS && + /^\d{15,25}$/.test(record.messageId ?? '') + ? record.messageId + : undefined; + } + async recent(): Promise { + await this.load(); + await this.chain; + return [...this.records.values()] + .sort((a, b) => b.updatedAt - a.updatedAt) + .slice(0, 50) + .map((r) => ({ ...r })); + } +} diff --git a/test/discord-bridge.test.ts b/test/discord-bridge.test.ts index 32db32d7e..e0a598cad 100644 --- a/test/discord-bridge.test.ts +++ b/test/discord-bridge.test.ts @@ -1,3 +1,4 @@ +let deliveryTestSequence = 900000000000000000n; // @vitest-environment node /** * @fileoverview Discord bridge: pure rules, the 0600 store, the REST client, the gateway state @@ -432,7 +433,7 @@ describe('DiscordBridge', () => { describe('replies', () => { const reply = (over: Record = {}) => ({ - id: 'm1', + id: String(deliveryTestSequence++), channel_id: THREAD, guild_id: GUILD, content: 'press y', @@ -482,12 +483,12 @@ describe('DiscordBridge', () => { b.stop(); }); - it('says so in the thread when the tab is gone', async () => { + it('reports unconfirmed input without claiming nothing was typed', async () => { sendInput.mockResolvedValueOnce(false); const b = await ready(); await b.onGatewayMessage(reply()); expect(posts[0].method).toBe('POST'); - expect(String(posts[0].body?.content)).toMatch(/gone/); + expect(String(posts[0].body?.content)).toMatch(/not confirmed/); b.stop(); }); }); @@ -1028,7 +1029,7 @@ describe('files', () => { describe('DiscordBridge files and startup hint', () => { const reply = (over: Record = {}) => ({ - id: '555555555555555555', + id: String(deliveryTestSequence++), channel_id: THREAD, guild_id: GUILD, content: '', @@ -1088,8 +1089,9 @@ describe('DiscordBridge files and startup hint', () => { it('saves an attachment under .codeman-discord and types its path, even with no text', async () => { const b = await make(); - await b.onGatewayMessage(reply()); - const dest = join(work, '.codeman-discord', '555555555555555555-shot_one.png'); + const message = reply(); + await b.onGatewayMessage(message); + const dest = join(work, '.codeman-discord', `${message.id}-shot_one.png`); expect(readFileSync(dest)).toEqual(Buffer.from([7, 8, 9])); expect(statSync(dest).mode & 0o777).toBe(0o600); expect(sent).toEqual([['s1', `[attached: ${dest}]\r`]]); diff --git a/test/discord-delivery.test.ts b/test/discord-delivery.test.ts new file mode 100644 index 000000000..42f552f91 --- /dev/null +++ b/test/discord-delivery.test.ts @@ -0,0 +1,414 @@ +/** Fault injection for persistent deduplication and retry safety. No real Discord or terminal input. */ +import { mkdtemp, readFile, rm, stat, symlink, writeFile } from 'node:fs/promises'; +import fs from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { DiscordDeliveryJournal, DISCORD_DELIVERY_LIMIT } from '../src/discord-delivery.js'; +import { + DiscordApi, + DiscordBridge, + DEFAULT_DISCORD_CONFIG, + writeDiscordConfig, + type DiscordFetch, + type GatewayMessage, +} from '../src/discord-bridge.js'; + +const GUILD = '111111111111111111'; +const PARENT = '222222222222222222'; +const USER = '333333333333333333'; +const THREAD = '444444444444444444'; +const MESSAGE = '555555555555555555'; +const CFG = { + ...DEFAULT_DISCORD_CONFIG, + enabled: true, + token: 'fake-secret-token', + guildId: GUILD, + parentChannelId: PARENT, + allowedUserIds: [USER], +}; +const MESSAGE_EVENT: GatewayMessage = { + id: MESSAGE, + guild_id: GUILD, + channel_id: THREAD, + author: { id: USER }, + content: 'perform this task once', +}; +const meta = { id: 'in:one', direction: 'inbound' as const, channelId: THREAD, guildId: GUILD, sessionId: 's1' }; +let dir: string; +let now: number; +let bridges: DiscordBridge[]; +beforeEach(async () => { + dir = await mkdtemp(join(tmpdir(), 'discord-delivery-')); + now = Date.now(); + bridges = []; +}); +afterEach(async () => { + vi.restoreAllMocks(); + for (const b of bridges) b.stop(); + await rm(dir, { recursive: true, force: true }); +}); + +it('durably claims an ID once under concurrency, with private permissions', async () => { + const journal = new DiscordDeliveryJournal(dir, () => now); + const claims = await Promise.all(Array.from({ length: 10 }, () => journal.claim(meta))); + expect(claims.filter((c) => c.fresh)).toHaveLength(1); + expect((await stat(join(dir, 'discord-delivery.json'))).mode & 0o777).toBe(0o600); + expect((await new DiscordDeliveryJournal(dir, () => now).claim(meta)).fresh).toBe(false); +}); + +it.each(['received', 'files_transferred', 'submitting'] as const)( + 'reconciles interrupted %s without replaying input', + async (state) => { + const first = new DiscordDeliveryJournal(dir, () => now); + await first.claim(meta); + await first.update(meta.id, state); + const restored = new DiscordDeliveryJournal(dir, () => now); + const duplicate = await restored.claim(meta); + expect(duplicate.fresh).toBe(false); + expect(duplicate.record.state).toBe(state === 'submitting' ? 'uncertain' : 'failed'); + expect(duplicate.record.reason).toBe(state === 'submitting' ? 'restart_during_submit' : 'restart_before_submit'); + } +); + +it('retains accepted input across restart and returns copies of bounded metadata', async () => { + const first = new DiscordDeliveryJournal(dir, () => now); + await first.claim(meta); + await first.update(meta.id, 'input_accepted'); + const restored = new DiscordDeliveryJournal(dir, () => now); + const records = await restored.recent(); + expect(records[0].state).toBe('input_accepted'); + records[0].state = 'failed'; + expect((await restored.recent())[0].state).toBe('input_accepted'); +}); + +it('fails closed on corruption, oversized metadata and symlinks', async () => { + const path = join(dir, 'discord-delivery.json'); + for (const contents of ['not JSON', '{}', 'x'.repeat(2 * 1024 * 1024 + 1)]) { + await writeFile(path, contents); + await expect(new DiscordDeliveryJournal(dir).claim(meta)).rejects.toThrow(/journal/i); + } + await rm(path); + const outside = join(dir, 'outside'); + await writeFile(outside, '[]'); + await symlink(outside, path); + await expect(new DiscordDeliveryJournal(dir).claim(meta)).rejects.toThrow(/journal/i); +}); + +it('never evicts recent inbound IDs when saturated, but expires records after seven days', async () => { + const records = Array.from({ length: DISCORD_DELIVERY_LIMIT }, (_, i) => ({ + ...meta, + id: String(i), + state: 'input_accepted', + createdAt: now, + updatedAt: now, + })); + await writeFile(join(dir, 'discord-delivery.json'), JSON.stringify(records)); + const journal = new DiscordDeliveryJournal(dir, () => now); + await expect(journal.claim(meta)).rejects.toThrow(/full/); + now += 8 * 24 * 60 * 60 * 1000; + expect((await journal.claim(meta)).fresh).toBe(true); +}); + +describe('Discord retry rules', () => { + it('uses one nonce after an accepted request loses its response', async () => { + const seen = new Set(); + const bodies: any[] = []; + const api = new DiscordApi( + () => CFG.token, + async (_url, init) => { + const body = JSON.parse(String(init.body)); + bodies.push(body); + if (!seen.has(body.nonce)) { + seen.add(body.nonce); + throw new TypeError('lost response'); + } + return Response.json({ id: MESSAGE }); + }, + async () => undefined + ); + expect(await api.post(THREAD, { content: 'hello', allowed_mentions: { parse: [], users: [] } })).toEqual({ + id: MESSAGE, + }); + expect(bodies).toHaveLength(2); + expect(seen.size).toBe(1); + expect(bodies[0]).toMatchObject({ enforce_nonce: true }); + expect(bodies[0].nonce.length).toBeLessThanOrEqual(25); + }); + + it('retries server errors a bounded number of times, but not permission failures', async () => { + const fetch = vi.fn(async () => Response.json({}, { status: 503 })); + const api = new DiscordApi( + () => CFG.token, + fetch, + async () => undefined + ); + await expect(api.post(THREAD, { content: 'hello', allowed_mentions: { parse: [], users: [] } })).rejects.toThrow(); + expect(fetch).toHaveBeenCalledTimes(3); + fetch.mockClear().mockImplementation(async () => Response.json({}, { status: 403 })); + await expect(api.post(THREAD, { content: 'hello', allowed_mentions: { parse: [], users: [] } })).rejects.toThrow(); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it('does not retry ambiguous non-idempotent thread creation', async () => { + const fetch = vi.fn(async () => { + throw new TypeError('lost response'); + }); + const api = new DiscordApi( + () => CFG.token, + fetch, + async () => undefined + ); + await expect(api.createThread(PARENT, 'test')).rejects.toThrow(); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it('honors retry_after and refuses to retry early when the wait exceeds its bound', async () => { + const sleep = vi.fn(async () => undefined); + const fetch = vi.fn(async () => Response.json({ retry_after: 45 }, { status: 429 })); + const api = new DiscordApi(() => CFG.token, fetch, sleep); + await expect(api.post(THREAD, { content: 'hello', allowed_mentions: { parse: [], users: [] } })).rejects.toThrow( + /429/ + ); + expect(sleep).not.toHaveBeenCalled(); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it('keeps multipart file nonce stable through retries', async () => { + const nonces: string[] = []; + const api = new DiscordApi( + () => CFG.token, + async (_url, init) => { + nonces.push(JSON.parse(String((init.body as FormData).get('payload_json'))).nonce); + return nonces.length === 1 ? Response.json({}, { status: 502 }) : Response.json({ id: MESSAGE }); + }, + async () => undefined + ); + await api.postWithFile( + THREAD, + { content: 'file', allowed_mentions: { parse: [], users: [] } }, + { name: 'a.txt', data: new Uint8Array([1]) } + ); + expect(nonces).toHaveLength(2); + expect(nonces[0]).toBe(nonces[1]); + }); +}); + +async function ready(sendInput = vi.fn(async () => true), fetchOverride?: DiscordFetch) { + await writeDiscordConfig(dir, CFG); + const b = new DiscordBridge({ + configDir: dir, + now: () => now, + sendInput, + sleep: async () => undefined, + fetchImpl: + fetchOverride ?? (async (url) => Response.json({ id: url.pathname.endsWith('/threads') ? THREAD : MESSAGE })), + socketFactory: () => ({ send() {}, close() {}, on() {} }), + }); + bridges.push(b); + await b.notify({ event: 'test', title: 'test', body: 'test', urgency: 'critical', sessionId: 's1' }); + return { b, sendInput }; +} + +it('submits once for concurrent duplicate events and once across a restart', async () => { + const { b, sendInput } = await ready(); + await Promise.all([b.onGatewayMessage(MESSAGE_EVENT), b.onGatewayMessage(MESSAGE_EVENT)]); + b.stop(); + now += 10_000; + const restored = await ready(sendInput); + await restored.b.onGatewayMessage(MESSAGE_EVENT); + expect(sendInput).toHaveBeenCalledTimes(1); +}); + +it('persists the submitting boundary before calling input and never stores prompt/token contents', async () => { + const input = vi.fn(async () => { + const raw = await readFile(join(dir, 'discord-delivery.json'), 'utf8'); + expect(JSON.parse(raw).find((r: any) => r.direction === 'inbound').state).toBe('submitting'); + expect(raw).not.toContain(MESSAGE_EVENT.content); + expect(raw).not.toContain(CFG.token); + return true; + }); + const { b } = await ready(input); + await b.onGatewayMessage(MESSAGE_EVENT); + expect(input).toHaveBeenCalledTimes(1); + expect((await b.recentDeliveries()).find((r) => r.direction === 'inbound')?.state).toBe('input_accepted'); +}); + +it('does not replay a partial/throwing input or confuse it with verified execution', async () => { + const input = vi.fn(async () => { + throw new Error('partial write'); + }); + const { b } = await ready(input); + await b.onGatewayMessage(MESSAGE_EVENT); + now += 10_000; + await b.onGatewayMessage(MESSAGE_EVENT); + expect(input).toHaveBeenCalledTimes(1); + expect((await b.recentDeliveries()).find((r) => r.direction === 'inbound')?.state).toBe('uncertain'); +}); + +it('does not claim successful multi-part output when a trailing part is refused', async () => { + let parts = 0; + const { b } = await ready(undefined, async (url) => { + if (url.pathname.endsWith('/threads')) return Response.json({ id: THREAD }); + return ++parts >= 3 ? Response.json({}, { status: 403 }) : Response.json({ id: MESSAGE }); + }); + now += 10_000; + const result = await b.notify({ + event: 'done', + title: 'done', + body: 'done', + sessionId: 's1', + urgency: 'critical', + fullText: async () => ({ text: 'a long output paragraph\n\n'.repeat(300), version: 'turn1' }), + }); + expect(result?.ok).toBe(false); + expect((await b.recentDeliveries()).some((r) => r.direction === 'outbound' && r.state === 'failed')).toBe(true); +}); + +it('retries a refused trailing part without repeating the already delivered start', async () => { + let refuse = false; + let index = 0; + const contents: string[] = []; + const { b } = await ready(undefined, async (url, init) => { + if (url.pathname.endsWith('/threads')) return Response.json({ id: THREAD }); + const body = JSON.parse(String(init.body)); + if (refuse && ++index === 2) return Response.json({}, { status: 403 }); + contents.push(body.content); + return Response.json({ id: MESSAGE }); + }); + const msg = { + event: 'done', + title: 'done', + body: 'done', + sessionId: 's1', + urgency: 'critical' as const, + fullText: async () => ({ + text: Array.from({ length: 150 }, (_, i) => `Paragraph ${i}: ${'x'.repeat(50)}`).join('\n\n'), + version: 'turn1', + }), + }; + now += 10_000; + refuse = true; + expect((await b.notify(msg))?.ok).toBe(false); + refuse = false; + expect((await b.notify(msg))?.ok).toBe(true); + expect(contents.filter((c) => c.includes('Paragraph 0:'))).toHaveLength(1); +}); + +it('does not blindly replay output after retry exhaustion leaves delivery uncertain', async () => { + let fail = false; + let requests = 0; + const { b } = await ready(undefined, async (url) => { + if (url.pathname.endsWith('/threads')) return Response.json({ id: THREAD }); + requests++; + if (fail) throw new TypeError('lost response'); + return Response.json({ id: MESSAGE }); + }); + const msg = { + event: 'done', + title: 'done', + body: 'done', + sessionId: 's1', + urgency: 'critical' as const, + fullText: async () => ({ text: 'one exact native response', version: 'turn1' }), + }; + now += 10_000; + fail = true; + expect((await b.notify(msg))?.ok).toBe(false); + const before = requests; + fail = false; + expect((await b.notify(msg))?.ok).toBe(false); + expect(requests).toBe(before); +}); + +it('refuses a changed rendering plan after part of a response has been delivered', async () => { + let count = 0; + let responsePosts = 0; + let refuse = false; + const { b } = await ready(undefined, async (url, init) => { + if (url.pathname.endsWith('/threads')) return Response.json({ id: THREAD }); + count++; + if ( + init.method === 'POST' && + url.pathname.endsWith('/messages') && + String(init.body).includes('a long native response paragraph') + ) { + responsePosts++; + if (refuse && responsePosts === 2) return Response.json({}, { status: 403 }); + } + return Response.json({ id: MESSAGE }); + }); + const msg = { + event: 'done', + title: 'original title', + body: 'done', + sessionId: 's1', + urgency: 'critical' as const, + fullText: async () => ({ text: 'a long native response paragraph\n\n'.repeat(150), version: 'turn1' }), + }; + now += 10_000; + refuse = true; + expect((await b.notify(msg))?.ok).toBe(false); + const before = count; + refuse = false; + const changed = await b.notify({ ...msg, title: 'different rendering header' }); + expect(changed?.ok).toBe(false); + expect(changed?.error).toMatch(/formatting changed/); + expect(count).toBe(before); +}); + +it('cancels a queued REST retry when authorization changes during the backoff', async () => { + let current = true, + requests = 0; + const api = new DiscordApi( + () => CFG.token, + async () => { + requests++; + return Response.json({ retry_after: 0.1 }, { status: 429 }); + }, + async () => { + current = false; + } + ); + expect( + await api.post(THREAD, { content: 'test', allowed_mentions: { parse: [], users: [] } }, () => current) + ).toBeNull(); + expect(requests).toBe(1); +}); +it('never submits when a durable inbound claim cannot be persisted', async () => { + const { b, sendInput } = await ready(); + const spy = vi.spyOn(fs, 'rename').mockRejectedValueOnce(new Error('read-only store')); + await b.onGatewayMessage(MESSAGE_EVENT); + expect(sendInput).not.toHaveBeenCalled(); + spy.mockRestore(); +}); + +it('does not blindly repeat an identical unversioned notification after restart', async () => { + let fail = false, + requests = 0; + const fetch: DiscordFetch = async (url) => { + if (url.pathname.endsWith('/threads')) return Response.json({ id: THREAD }); + requests++; + if (fail) throw new Error('lost acknowledgement'); + return Response.json({ id: MESSAGE }); + }; + const { b } = await ready(undefined, fetch); + const msg = { + event: 'alert', + title: 'Needs attention', + body: 'Unversioned alert', + urgency: 'critical' as const, + sessionId: 's1', + }; + now += 10000; + fail = true; + expect((await b.notify(msg))?.ok).toBe(false); + b.stop(); + fail = false; + const restarted = (await ready(undefined, fetch)).b; + now += 10000; + const before = requests; + expect((await restarted.notify(msg))?.ok).toBe(false); + expect(requests).toBe(before); + expect((await restarted.notify({ ...msg, body: 'A distinct new alert' }))?.ok).toBe(true); +});