diff --git a/src/transport/conversation.ts b/src/transport/conversation.ts index 2259046..c942562 100644 --- a/src/transport/conversation.ts +++ b/src/transport/conversation.ts @@ -31,6 +31,10 @@ export interface ConversationDeps { /** R2-F3: note embedder — enables ranked note injection (falls back to recency) */ embedder?: EmbeddingProvider; budget?: { maxHops?: number; maxTokens?: number; wallMs?: number }; + /** #7: opt-in — abort the active turn when the requesting client disconnects + * mid-flight (tab close, network drop). Off by default: proxied long-poll + * deployments where socket close is routine must not trip it. */ + abortOnDisconnect?: boolean; /** consumer's event sink (WS bus in the demo; anything callable) */ emit(event: TurnEvent & { sessionId: string }): void; /** streaming seam: per-session delta handler factory */ @@ -171,6 +175,17 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver const outcome: TurnOutcome = { sessionId: id, ok: false, tokens: { in: 0, out: 0 }, hops: 0 }; const ac = new AbortController(); activeTurns.set(id, ac); + // #7: client disconnect mid-flight aborts the turn. The seam is the + // RESPONSE stream: req.raw 'close' fires when the request body completes + // (not socket death); reply.raw 'close' with !writableEnded = the client + // vanished before the answer was flushed. The map-identity guard makes + // the normal post-response close (which fires after activeTurns.delete) + // a no-op. + if (deps.abortOnDisconnect) { + reply.raw.on('close', () => { + if (!reply.raw.writableEnded && activeTurns.get(id) === ac) ac.abort(); + }); + } try { // core-side note injection (review F3): active notes ride the system // prompt, provenance-marked — knowledge crosses sessions diff --git a/test/abort.test.ts b/test/abort.test.ts index 124166a..c0c19dd 100644 --- a/test/abort.test.ts +++ b/test/abort.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from 'vitest'; +import net from 'node:net'; import Fastify from 'fastify'; import { SqliteStorage } from '../src/memory/sqlite.ts'; import { MockProvider } from '../src/provider/mock.ts'; @@ -47,3 +48,74 @@ describe('abort mid-flight', () => { storage.close(); }); }); + + +// ── #7: client disconnect mid-flight aborts the turn (opt-in) ── +describe('abort on client disconnect (#7)', () => { + it('socket destroyed mid-turn → turn aborts (opt-in flag on)', async () => { + const storage = new SqliteStorage(':memory:'); + const registry = new ToolRegistry().register({ + name: 'noop', description: 'n', permission: 'read', parameters: { type: 'object', properties: {} }, + execute: async () => 'done', + }); + const app = Fastify(); + await registerConversationApi(app, { + storage, provider: new MockProvider([{ kind: 'hang', ms: 20_000 }]), registry, + policy: { name: 'draft', allows: 'draft' }, systemPrompt: 'sys', + budget: { wallMs: 60_000 }, emit: () => undefined, + abortOnDisconnect: true, + }); + await app.listen({ port: 0, host: '127.0.0.1' }); + const port = (app.server.address() as { port: number }).port; + + const created = await (await fetch(`http://127.0.0.1:${port}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const sid = created.session.id; + + // open a raw socket, send the message, then DESTROY the connection mid-turn + const sock = net.connect({ port, host: '127.0.0.1' }); + await new Promise((r) => sock.once('connect', () => r())); + sock.write(`POST /api/agent/sessions/${sid}/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: ${JSON.stringify({ content: 'hang' }).length}\r\nConnection: close\r\n\r\n${JSON.stringify({ content: 'hang' })}`); + await new Promise((r) => setTimeout(r, 300)); // turn is now hanging in-flight + sock.destroy(); + + // the abort lands in the transport's catch → session status flips to error + let status: string | undefined; + for (let i = 0; i < 40 && status !== 'error'; i++) { + await new Promise((r) => setTimeout(r, 100)); + status = (await storage.getSession(sid))?.status; + } + expect(status).toBe('error'); + await app.close(); + }); + + it('flag OFF (default) → disconnect does NOT abort; the turn outlives the socket', async () => { + const storage = new SqliteStorage(':memory:'); + const registry = new ToolRegistry().register({ + name: 'noop', description: 'n', permission: 'read', parameters: { type: 'object', properties: {} }, + execute: async () => 'done', + }); + const app = Fastify(); + await registerConversationApi(app, { + storage, provider: new MockProvider([{ kind: 'hang', ms: 3_000 }]), registry, + policy: { name: 'draft', allows: 'draft' }, systemPrompt: 'sys', + budget: { wallMs: 60_000 }, emit: () => undefined, + // no abortOnDisconnect — default off + }); + await app.listen({ port: 0, host: '127.0.0.1' }); + const port = (app.server.address() as { port: number }).port; + const created = await (await fetch(`http://127.0.0.1:${port}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const sid = created.session.id; + const sock = net.connect({ port, host: '127.0.0.1' }); + await new Promise((r) => sock.once('connect', () => r())); + sock.write(`POST /api/agent/sessions/${sid}/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: ${JSON.stringify({ content: 'hang' }).length}\r\nConnection: close\r\n\r\n${JSON.stringify({ content: 'hang' })}`); + await new Promise((r) => setTimeout(r, 300)); + sock.destroy(); + // within the hang window the session has NOT been failed by the disconnect + await new Promise((r) => setTimeout(r, 700)); + const st = (await storage.getSession(sid))?.status; + expect(st).not.toBe('error'); + // cleanup: abort via the route so the suite doesn't wait the full hang + await fetch(`http://127.0.0.1:${port}/api/agent/sessions/${sid}/abort`, { method: 'POST' }); + await app.close(); + }); +});