diff --git a/src/cli-notify.ts b/src/cli-notify.ts index af925bcd..5cdae57a 100644 --- a/src/cli-notify.ts +++ b/src/cli-notify.ts @@ -43,7 +43,14 @@ export const postJson: JsonPost = (apiUrl, apiPath, payload, timeoutMs = REQUEST const url = new URL(apiPath, apiUrl); const body = JSON.stringify(payload); const transport = url.protocol === 'https:' ? https : http; - return new Promise((resolve) => { + return new Promise((resolveOnce) => { + // Settle once: a response that stalls after its headers ends in 'close' without 'end'. + let settled = false; + const resolve: typeof resolveOnce = (v) => { + if (settled) return; + settled = true; + resolveOnce(v); + }; const headers: Record = { Accept: 'application/json', 'Content-Type': 'application/json', @@ -64,6 +71,8 @@ export const postJson: JsonPost = (apiUrl, apiPath, payload, timeoutMs = REQUEST (res) => { let text = ''; res.on('data', (c) => (text += c)); + res.on('error', () => resolve({ status: 0 })); + res.on('close', () => resolve({ status: 0 })); res.on('end', () => { let error: string | undefined; let delivery: { ok: boolean; error?: string } | undefined; diff --git a/src/discord-bridge.ts b/src/discord-bridge.ts index dd39f2a5..85f3a442 100644 --- a/src/discord-bridge.ts +++ b/src/discord-bridge.ts @@ -655,6 +655,8 @@ export interface DiscordBridgeDeps { * Codeman types itself. */ sendInput: (sessionId: string, text: string, fromUser: boolean) => Promise; + /** True once the tab's agent has exited: tmux still accepts keys for a dead pane, so a tick would lie. */ + agentExited?: (sessionId: string) => boolean; /** True while the session shows a permission or question dialog, which acts on a bare keypress. */ awaitingChoice?: (sessionId: string) => boolean; /** A session's working folder, where received files are saved. Undefined when it is gone or remote. */ @@ -683,6 +685,8 @@ export class DiscordBridge { private held = false; /** Thread-map writes run one after another: concurrent creations must not interleave on the tmp file. */ private persistChain: Promise = Promise.resolve(); + /** Per-thread chain of inbound messages (see onGatewayMessage). */ + private inbound = new Map>(); private mapLoad?: Promise; /** The first full load (config, thread map, gateway), shared by every caller that arrives before it ends. */ private firstLoad?: Promise; @@ -711,6 +715,17 @@ export class DiscordBridge { await this.firstLoad; } + /** Forget thread entries for tabs that no longer exist (dropped without a delete, for example). Once, at boot. */ + async prune(liveIds: Iterable): Promise { + await this.ensureLoaded(); + const live = new Set(liveIds); + if (live.size === 0) return; // nothing restored: more likely a failed restore than an empty server + const stale = Object.keys(this.threads).filter((id) => !live.has(id)); + if (stale.length === 0) return; + for (const id of stale) delete this.threads[id]; + await this.persistThreads(); + } + /** Hold the gateway back (boot): a reply must not arrive before the restored tabs exist. */ hold(): void { this.held = true; @@ -765,6 +780,7 @@ export class DiscordBridge { this.lastReply.clear(); this.names.clear(); this.fresh.clear(); + this.inbound.clear(); } private fail(err: unknown): DiscordResult { @@ -991,8 +1007,17 @@ export class DiscordBridge { }; } - /** Inbound reply. Every gate here drops silently: a stranger learns nothing. */ + /** Inbound reply. One chain per thread, so a reply that waits on a file download cannot be overtaken by a later one. */ async onGatewayMessage(m: GatewayMessage): Promise { + const key = String(m.channel_id); + const run = (this.inbound.get(key) ?? Promise.resolve()).then(() => this.handleMessage(m)).catch(() => undefined); + this.inbound.set(key, run); + await run; + if (this.inbound.get(key) === run) this.inbound.delete(key); + } + + /** Every gate here drops silently: a stranger learns nothing. */ + private async handleMessage(m: GatewayMessage): Promise { if (!this.cfg.enabled || m.author?.bot) return; if (m.guild_id !== this.cfg.guildId) return; if (!this.cfg.allowedUserIds.includes(m.author?.id)) return; @@ -1005,6 +1030,10 @@ export class DiscordBridge { await this.api.react(m.channel_id, m.id, '⏳').catch(() => undefined); return; } + if (this.deps.agentExited?.(sessionId)) { + await this.say(m.channel_id, 'The agent in that tab has exited, so nothing was typed.'); + return; + } this.lastReply.set(sessionId, t); const saved = await this.saveAttachments(sessionId, m); const text = sanitiseReply(m.content ?? '', saved.paths); diff --git a/src/web/server.ts b/src/web/server.ts index 527f1d02..734c8597 100644 --- a/src/web/server.ts +++ b/src/web/server.ts @@ -520,6 +520,7 @@ export class WebServer extends EventEmitter { if (await session.writeViaMux(text, { fromUser }).catch(() => false)) return true; return session.write(text, { fromUser }); }, + agentExited: (id) => !!this.sessions.get(id)?.paneExit, awaitingChoice: (id) => approvalInbox.listPending().some((item) => item.sessionId === id && item.kind !== 'idle'), // Remote (ssh) sessions keep their folder on another host, so received files have nowhere local to go. sessionDir: (sessionId) => { @@ -3258,7 +3259,10 @@ export class WebServer extends EventEmitter { if (restoredMail > 0) console.log(`[AgentInbox] restored ${restoredMail} pending message(s)`); const restored = await this.restoreMuxSessions(); await this.finalizeRestoredState(restored); - void this.discordBridge.release().catch(() => undefined); + void this.discordBridge + .release() + .then(() => this.discordBridge.prune(this.sessions.keys())) + .catch(() => undefined); // Instance-scoped reaper: after restore, `docker rm -f` managed containers of // THIS instance whose case is gone from docker-cases.json (best-effort, never diff --git a/test/cli-notify.test.ts b/test/cli-notify.test.ts index a1e6ad3d..5127cf54 100644 --- a/test/cli-notify.test.ts +++ b/test/cli-notify.test.ts @@ -71,6 +71,21 @@ describe('runNotify', () => { } }); + it('postJson settles when the server sends headers and then stalls the body', async () => { + const server = createServer((_req, res) => { + res.writeHead(200, { 'Content-Type': 'application/json', 'Content-Length': '500' }); + res.write('{"success":'); + }); + await new Promise((r) => server.listen(0, '127.0.0.1', r)); + try { + const { port } = server.address() as AddressInfo; + expect((await postJson(`http://127.0.0.1:${port}`, '/x', {}, 100)).status).toBe(0); + } finally { + server.closeAllConnections(); + server.close(); + } + }); + it('treats a disabled bridge (409) as success that says so', async () => { const r = await runNotify({ message: 'x' }, ENV, '/work', fake(409)); expect(r.code).toBe(0); diff --git a/test/discord-bridge.test.ts b/test/discord-bridge.test.ts index 376f244c..0ec6e008 100644 --- a/test/discord-bridge.test.ts +++ b/test/discord-bridge.test.ts @@ -551,6 +551,17 @@ describe('DiscordBridge', () => { b.stop(); }); + it('prune drops thread entries for tabs that no longer exist, and keeps the live ones', async () => { + const b = bridge(); + await b.notify(MSG); + await b.notify({ ...MSG, sessionId: 's2', sessionName: 'w2' }); + await b.prune(['s2']); + expect(Object.keys(await readThreadMap(dir))).toEqual(['s2']); + await b.prune([]); // nothing restored: leave the map alone + expect(Object.keys(await readThreadMap(dir))).toEqual(['s2']); + b.stop(); + }); + describe('sendTest', () => { it('posts with the config it is given and leaves the live config alone', async () => { const auth: string[] = []; @@ -734,7 +745,8 @@ describe('DiscordBridge files and startup hint', () => { function make( cfg: Partial = {}, dirFor: (id: string) => string | undefined = () => work, - awaitingChoice?: (id: string) => boolean + awaitingChoice?: (id: string) => boolean, + agentExited?: (id: string) => boolean ) { const b = new DiscordBridge({ configDir: dir, @@ -751,6 +763,7 @@ describe('DiscordBridge files and startup hint', () => { sendInput: async (id, text, fromUser) => (sent.push([id, text, fromUser]), true), sessionDir: dirFor, awaitingChoice, + agentExited, socketFactory: () => new FakeSocket(), now: () => t, sleep: async () => undefined, @@ -834,6 +847,26 @@ describe('DiscordBridge files and startup hint', () => { b.stop(); }); + it('says the agent has exited and types nothing when the tab reports it', async () => { + const b = await make({}, undefined, undefined, () => true); + await b.onGatewayMessage(reply({ content: 'hello', attachments: [] })); + expect(sent).toHaveLength(0); + expect(String(posted.at(-1)?.body?.content)).toMatch(/exited/); + expect(posted.some((p) => p.path.includes('reactions'))).toBe(false); + b.stop(); + }); + + it('types a reply that waits on a download before a later text reply', async () => { + const b = await make(); + const slow = b.onGatewayMessage(reply({ content: 'first' })); + await new Promise((r) => setTimeout(r, 0)); // the first reply is under way (gate passed) before the clock moves + t += 5_000; + const fast = b.onGatewayMessage(reply({ id: '555555555555555556', content: 'second', attachments: [] })); + await Promise.all([slow, fast]); + expect(sent.map((s) => s[1].split(' ')[0])).toEqual(['first', 'second\r']); + b.stop(); + }); + it('types a lone digit without Enter while a dialog is waiting, and with Enter otherwise', async () => { let waiting = true; const b = await make({}, undefined, () => waiting); diff --git a/test/discord-push-glue.test.ts b/test/discord-push-glue.test.ts index 96d1e9c9..e062e4d9 100644 --- a/test/discord-push-glue.test.ts +++ b/test/discord-push-glue.test.ts @@ -43,6 +43,20 @@ function server(): Internals { beforeEach(() => sendNotification.mockClear()); +describe('bridge deps', () => { + it('agentExited is true only for a session whose pane exit was observed', () => { + const s = new WebServer(0, false, true, '127.0.0.1', 'box') as unknown as Internals & { + discordBridge: { deps: { agentExited: (id: string) => boolean } }; + }; + s.sessions.set('dead', { paneExit: { observedAt: 1 } }); + s.sessions.set('alive', {}); + const { agentExited } = s.discordBridge.deps; + expect(agentExited('dead')).toBe(true); + expect(agentExited('alive')).toBe(false); + expect(agentExited('missing')).toBe(false); + }); +}); + describe('sendPushNotifications -> Discord', () => { it('posts a permission prompt with zero push subscriptions, with the rendered screen as the tail', async () => { const s = server();