feat(transport): opt-in abort of the active turn on client disconnect (#7)

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

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<void>((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<void>((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();
+  });
+});