fix(discord): say when the agent has exited, keep replies in order, settle a stalled response, prune dead tabs

- a reply to a tab whose agent has exited is answered in the thread, with no tick
- inbound messages run on one chain per thread, so a reply with a file cannot be overtaken
- postJson settles when a response stalls after its headers
- the thread map is pruned against the live sessions once after boot
This commit is contained in:
Devvyn committed 2026-10-30 10:33:34 +08:00
1 parent 72a59de96c
commit b68792f988
6 files changed
+108 -4

No files matched your search

+10 -1
View File
@@ -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<string, string | number> = {
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;
+30 -1
View File
@@ -655,6 +655,8 @@ export interface DiscordBridgeDeps {
* Codeman types itself.
*/
sendInput: (sessionId: string, text: string, fromUser: boolean) => Promise<boolean>;
/** 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<void> = Promise.resolve();
/** Per-thread chain of inbound messages (see onGatewayMessage). */
private inbound = new Map<string, Promise<void>>();
private mapLoad?: Promise<void>;
/** The first full load (config, thread map, gateway), shared by every caller that arrives before it ends. */
private firstLoad?: Promise<void>;
@@ -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<string>): Promise<void> {
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<void> {
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<void> {
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);
+5 -1
View File
@@ -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
+15
View File
@@ -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<void>((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);
+34 -1
View File
@@ -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<DiscordConfig> = {},
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);
+14
View File
@@ -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();