From 6eb6e465b57c6945a7424ec4ed83c45f93a4b294 Mon Sep 17 00:00:00 2001 From: Nell-1783 Date: Wed, 7 Oct 2026 23:21:05 -0400 Subject: [PATCH 1/7] fix(lifecycle): late tool results and MCPL arrivals at stop() settle instead of rejecting unhandled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit stop() closes the process queue first and the MCPL connections only after awaiting stream settlement, maintenance and the loop. Anything arriving in between reached ProcessQueueImpl.push, which throws "Queue is closed", from async callbacks with nothing to catch it. A tool still running at stop() completed later: dispatchToolCallEvent's successful result's pushEvent threw, its .catch misreported that as the module's failure (markDispatchFailure, tool:failed), and its own pushEvent threw again, unhandled. On main this fails test files that stop a framework while a tool runs: conversation-routing failed 3 of 3 at file level without --test-force-exit (Petra-1012, room-203 #57742). Two outcomes, one boundary: - pushEvent drops a `tool-result` once the queue is closed, with a log line: the stream that asked is gone and the tool has already run, so there is no one to refuse it to. Every other event type still throws, so fresh input to a stopped framework is refused visibly to its caller (an API request, a module) instead of silently accepted. tune-out's two tool-result pushes go through pushEvent like every other tool's (one sat in a .then with no catch). - The MCPL `push-event` and `channels-incoming` listeners check the stopped state after the awareness barrier and before the handler, so a fresh post-stop item is refused whole (-32603, "the host is stopping") and never recorded as accepted on its way to a queue error. They also catch whatever the handler throws (the host stopping while it runs, or a fault), answered -32603 too ("the host stopped while handling this request", or the handler's message), logged as a failed request that may have admitted part of itself first (an earlier message of a batch), never as a per-message rejection. A notification is logged. The two are told apart by an explicit outcome ({kind: 'refused'} or {kind: 'failed', error}): a handler that throws `undefined` or rejects with nothing is a failed request on a live host ("the handler failed"), not a stopping refusal. Thrown out of the listener, it was an unhandled rejection, and the server's request waited for its timeout. - Queue checks use optional chaining for prototype-built harnesses. Tests: stop-with-work-in-flight (new) — a tool that succeeds after stop() (dropped and logged, no tool:failed), one whose module throws (its error result dropped), a dispatch that rejects (reported failed, its failure result dropped), fresh input still throwing, a push and an incoming message after stop() answered -32603 and never handed to the handler, a push held at the awareness barrier refused once it lifts, a throwing handler answered with its error, and a handler rejecting with nothing answered "the handler failed" and logged as failed, never refused. The same change is 4180ccd in agent-framework #256's chain, where #256's aborted-stream test exposed the race and Mira-1782 reviewed the fix (room-203). This is that commit cherry-picked onto main without #256's present-while-acting test, which doesn't exist here; the changelog's first paragraph names conversation-routing instead of #256's work. Whichever of this and #256 merges second drops the duplicate. Co-Authored-By: Claude Opus 5.5 --- changelog.d/stop-with-work-in-flight.fixed.md | 20 ++ src/framework.ts | 86 +++++- test/stop-with-work-in-flight.test.ts | 282 ++++++++++++++++++ 3 files changed, 379 insertions(+), 9 deletions(-) create mode 100644 changelog.d/stop-with-work-in-flight.fixed.md create mode 100644 test/stop-with-work-in-flight.test.ts diff --git a/changelog.d/stop-with-work-in-flight.fixed.md b/changelog.d/stop-with-work-in-flight.fixed.md new file mode 100644 index 000000000..186e58b30 --- /dev/null +++ b/changelog.d/stop-with-work-in-flight.fixed.md @@ -0,0 +1,20 @@ +- **Late tool results, and MCPL pushes or incoming messages, at `stop()` no + longer end in an unhandled rejection.** `stop()` closes the process queue + before it closes MCPL connections, and nothing caught the "Queue is + closed" that this meant for late arrivals. Test files that stop a + framework while a tool still runs, such as `conversation-routing`, failed + at file level with it. + - A tool's result that arrives after `stop()` is logged and dropped. The + stream that asked for it is gone, and the tool has already run. Before, + its push threw, the throw was reported as the tool's own failure, and + that failure's push threw again, unhandled. tune-out's results now go + through the same `pushEvent` as every other tool's. + - A server's `push/event` or `channels/incoming` that arrives once the + host is stopping is refused before its handler runs, with a JSON-RPC + error (`-32603`, "the host is stopping"). It is never answered + "accepted". + - A handler that throws, because the host stopped while it ran or because + of a fault, is answered with an error too. Part of the request may + already have been admitted. The error is never thrown out of the + connection's listener. A notification is logged. + - Direct input to `pushEvent` after `stop()` still throws to its caller. diff --git a/src/framework.ts b/src/framework.ts index 349bdd693..d691b5e12 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -2356,10 +2356,58 @@ export class AgentFramework { : undefined; } + /** + * A server's push or incoming message the host couldn't handle, answered + * with a JSON-RPC error when it is a request (a notification has no one to + * answer, and is logged), rather than thrown out of the connection's + * listener, where it was an unhandled rejection and the server's request + * waited for its timeout. + * - `refused`: the listener turned it away before its handler ran; the + * host is stopping, and none of it was taken. + * - `failed`: the handler threw (the host stopped while it ran, or a + * fault; whatever it threw, `undefined` included). The request failed, + * but the handler may have admitted part of it first (an earlier message + * of a batch), so nothing here claims that none of it was taken. + */ + private refuseServerItem( + method: string, + serverId: string, + responder: { respondError?: (code: number, message: string) => void } | undefined, + outcome: { kind: 'refused' } | { kind: 'failed'; error: unknown }, + ): void { + // Optional chaining: prototype-built harnesses leave the queue unset. + const reason = outcome.kind === 'refused' + ? 'the host is stopping' + : this.queue?.isClosed + ? 'the host stopped while handling this request' + : outcome.error instanceof Error + ? outcome.error.message + : `the handler failed${outcome.error === undefined ? '' : `: ${String(outcome.error)}`}`; + console.error(outcome.kind === 'refused' + ? `[mcpl] ${method} from ${serverId} refused: ${reason}` + : `[mcpl] ${method} from ${serverId} failed: ${reason} (part of it may already have been admitted)`); + try { + responder?.respondError?.(-32603, reason); + } catch { + // The connection is going too; there is no one left to tell. + } + } + /** * Push a process event to the queue. + * + * After stop() the queue is closed and this throws, so a direct caller + * learns the framework is stopped — except for a tool's result. A call + * still running at stop() completes later, into a callback with nothing + * to catch the throw (an unhandled rejection), and nothing could take its + * result anyway: the stream that asked is gone, and the tool has already + * run. That result is dropped, and the drop is logged. */ pushEvent(event: ProcessEvent): void { + if (event.type === 'tool-result' && this.queue?.isClosed) { + console.error(`[framework] ${event.agentName}: result of tool call ${event.callId} (${event.moduleName}) arrived after stop — dropped`); + return; + } this.queue.push(event); this.emitTrace({ type: 'process:received', processEvent: event }); } @@ -12327,7 +12375,9 @@ export class AgentFramework { void this.tuneOutCoordinator .handleSubconsciousTool(enrichedCall.name, enrichedCall.input as Record) .then((result) => { - this.queue.push({ + // Through pushEvent, like every other tool's result, so one that + // arrives after stop() is dropped rather than rejected unhandled. + this.pushEvent({ type: 'tool-result', callId: enrichedCall.id, agentName, @@ -14280,9 +14330,19 @@ export class AgentFramework { params: PushEventParams, responder?: { respond: (result: unknown) => void; respondError: (code: number, message: string) => void }, ) => { - const barrier = this.discordAwarenessBarrier; - if (barrier) await barrier.promise; - await this.pushHandler?.handlePushEvent(connection.id, params, responder as never); + try { + const barrier = this.discordAwarenessBarrier; + if (barrier) await barrier.promise; + // Checked here, after the barrier and before anything records the + // push as accepted: a stopped host refuses it. + if (this.queue?.isClosed) { + this.refuseServerItem('push/event', connection.id, responder, { kind: 'refused' }); + return; + } + await this.pushHandler?.handlePushEvent(connection.id, params, responder as never); + } catch (error) { + this.refuseServerItem('push/event', connection.id, responder, { kind: 'failed', error }); + } }); // Handle server-initiated inference requests (Step 6) @@ -14381,11 +14441,19 @@ export class AgentFramework { // Handle incoming channel messages (Step 7) connection.on('channels-incoming', async ( params: ChannelsIncomingParams, - responder?: { respond: (result: unknown) => void }, + responder?: { respond: (result: unknown) => void; respondError?: (code: number, message: string) => void }, ) => { - const barrier = this.discordAwarenessBarrier; - if (barrier) await barrier.promise; - await this.channelRegistry?.handleIncoming(connection.id, params, responder as never); + try { + const barrier = this.discordAwarenessBarrier; + if (barrier) await barrier.promise; + if (this.queue?.isClosed) { + this.refuseServerItem('channels/incoming', connection.id, responder, { kind: 'refused' }); + return; + } + await this.channelRegistry?.handleIncoming(connection.id, params, responder as never); + } catch (error) { + this.refuseServerItem('channels/incoming', connection.id, responder, { kind: 'failed', error }); + } }); // Handle host-level admin commands from a surface (e.g. Discord /undo) @@ -15273,7 +15341,7 @@ export class AgentFramework { } : { success: false, error: r.error, isError: false }; } - this.queue.push({ + this.pushEvent({ type: 'tool-result', callId: call.id, agentName, diff --git a/test/stop-with-work-in-flight.test.ts b/test/stop-with-work-in-flight.test.ts new file mode 100644 index 000000000..b346cada2 --- /dev/null +++ b/test/stop-with-work-in-flight.test.ts @@ -0,0 +1,282 @@ +/** + * stop() with work still in flight. + * + * stop() closes the process queue first and closes MCPL connections later, + * so for a while work started before stop() keeps completing, and servers + * keep sending. Two outcomes, one boundary (found through #256's aborted- + * stream test, where a robot--move outlived the framework; the race itself + * is older than #256): + * - a tool's result that arrives after stop() is logged and dropped: the + * stream that asked is gone, and the tool has already run. Before, its + * push threw "Queue is closed", the throw was misreported as the tool's + * failure, and that failure's push threw again, unhandled; + * - a server's push or incoming message the host can't take any more is + * answered with a JSON-RPC error, never "accepted", and never thrown + * out of the connection's listener. + * Fresh input to a stopped framework still throws to its caller. + */ + +import { describe, it, beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type { ContentBlock } from '@animalabs/membrane'; +import type { + Module, + ModuleContext, + ProcessEvent, + ProcessState, + EventResponse, + ToolCall, + ToolDefinition, + ToolResult, +} from '../src/index.js'; +import { AgentFramework } from '../src/index.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; + +/** A tool that settles only when the test says so, so the test owns the work it started. */ +class SlowModule implements Module { + readonly name = 'slow'; + private settle: { resolve: (r: ToolResult) => void; reject: (e: Error) => void } | null = null; + readonly started: Promise; + private markStarted!: () => void; + + constructor() { + this.started = new Promise((r) => { this.markStarted = r; }); + } + + async start(_ctx: ModuleContext): Promise {} + async stop(): Promise {} + + getTools(): ToolDefinition[] { + return [{ name: 'work', description: 'Slow work', inputSchema: { type: 'object', properties: {} } }]; + } + + handleToolCall(_call: ToolCall): Promise { + return new Promise((resolve, reject) => { + this.settle = { resolve, reject }; + this.markStarted(); + }); + } + + /** Settle the running call; resolves once its completion callbacks have run. */ + async finish(outcome: 'success' | 'failure'): Promise { + if (outcome === 'success') this.settle!.resolve({ success: true, data: { done: true } }); + else this.settle!.reject(new Error('the work failed')); + await new Promise((r) => setTimeout(r, 20)); + } + + async onProcess(event: ProcessEvent, _state: ProcessState): Promise { + if (event.type === 'external-message') { + return { + addMessages: [{ participant: 'User', content: [{ type: 'text', text: String((event as { content?: unknown }).content) }] }], + requestInference: true, + }; + } + return {}; + } +} + +describe('stop() with work still in flight', () => { + let tempDir: string; + let rejections: unknown[]; + let logged: string[]; + const onRejection = (reason: unknown): void => { rejections.push(reason); }; + const consoleError = console.error; + + beforeEach(() => { + tempDir = mkdtempSync(join(tmpdir(), 'af-stop-in-flight-')); + rejections = []; + logged = []; + process.on('unhandledRejection', onRejection); + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + }); + + afterEach(() => { + console.error = consoleError; + process.off('unhandledRejection', onRejection); + rmSync(tempDir, { recursive: true, force: true }); + }); + + const dropped = (): string[] => logged.filter((line) => line.includes('arrived after stop')); + + async function frameworkWithRunningTool( + opts: { dispatch?: 'rejects' } = {}, + ): Promise<{ framework: AgentFramework; module: SlowModule; failed: string[] }> { + const membrane = new MockMembrane(); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Working on it.' }, + { type: 'tool_use', id: 'c1', name: 'slow--work', input: {} }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Never written.' }] as ContentBlock[])); + const module = new SlowModule(); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), + membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'test' }], + modules: [module], + }); + const failed: string[] = []; + framework.onTrace((e) => { + if (e.type === 'tool:failed') failed.push(String((e as { tool?: unknown }).tool)); + }); + if (opts.dispatch === 'rejects') { + // The module registry turns a module's throw into an error result; this + // is the dispatch itself rejecting (the host got no result at all). + const registry = (framework as unknown as { moduleRegistry: { handleToolCall(call: ToolCall): Promise } }).moduleRegistry; + const handle = registry.handleToolCall.bind(registry); + registry.handleToolCall = (call) => handle(call).then(() => { throw new Error('the dispatch was lost'); }); + } + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'go', metadata: {} } as unknown as ProcessEvent); + await module.started; + return { framework, module, failed }; + } + + it('a tool that succeeds after stop() has its result dropped, not thrown or misreported as a failure', async () => { + const { framework, module, failed } = await frameworkWithRunningTool(); + await framework.stop(); + await module.finish('success'); + + assert.deepEqual(rejections, []); + assert.deepEqual(failed, [], 'the tool succeeded; nothing reports it as a failure'); + assert.equal(dropped().length, 1, logged.join('\n')); + }); + + it('a tool that fails after stop() has its error result dropped, not thrown', async () => { + const { framework, module, failed } = await frameworkWithRunningTool(); + await framework.stop(); + await module.finish('failure'); + + assert.deepEqual(rejections, []); + assert.deepEqual(failed, [], 'the registry answers a module throw with an error result'); + assert.equal(dropped().length, 1, logged.join('\n')); + }); + + it('a dispatch that rejects after stop() is reported as failed, and its failure result dropped, not thrown', async () => { + const { framework, module, failed } = await frameworkWithRunningTool({ dispatch: 'rejects' }); + await framework.stop(); + await module.finish('success'); + + assert.deepEqual(rejections, []); + assert.equal(failed.length, 1, 'the host got no result: a dispatch failure'); + assert.equal(dropped().length, 1, logged.join('\n')); + }); + + it('fresh input to a stopped framework still throws to its caller', async () => { + const { framework, module } = await frameworkWithRunningTool(); + await framework.stop(); + assert.throws( + () => framework.pushEvent({ type: 'external-message', source: 'test', content: 'too late', metadata: {} } as unknown as ProcessEvent), + /Queue is closed/, + ); + await module.finish('success'); + assert.deepEqual(rejections, []); + }); + + describe('a server sending while the host stops', () => { + async function wiredFramework() { + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), + membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'test' }], + modules: [], + }); + const connection = Object.assign(new EventEmitter(), { id: 'srv' }); + (framework as unknown as { wireMcplEvents(c: unknown): void }).wireMcplEvents(connection); + return { framework, connection }; + } + + function responder() { + const answers: Array<{ result?: unknown; error?: { code: number; message: string } }> = []; + return { + answers, + respond: (result: unknown) => { answers.push({ result }); }, + respondError: (code: number, message: string) => { answers.push({ error: { code, message } }); }, + }; + } + + const settle = (): Promise => new Promise((r) => setTimeout(r, 10)); + + it('a push or incoming message after stop() is answered with an error, never accepted', async () => { + const { framework, connection } = await wiredFramework(); + let handled = 0; + (framework as unknown as { pushHandler: unknown }).pushHandler = { handlePushEvent: async () => { handled++; } }; + await framework.stop(); + + const push = responder(); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e1', payload: { content: [{ type: 'text', text: 'late' }] } }, push); + const incoming = responder(); + connection.emit('channels-incoming', { messages: [{ channelId: 'c', messageId: 'm', content: [] }] }, incoming); + await settle(); + + assert.equal(handled, 0, 'never handed to the push handler, which would record it as accepted'); + assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'the host is stopping' } }]); + assert.deepEqual(incoming.answers, [{ error: { code: -32603, message: 'the host is stopping' } }]); + assert.deepEqual(rejections, []); + }); + + it('a push held at the awareness barrier while the host stops is refused once the barrier lifts', async () => { + const { framework, connection } = await wiredFramework(); + let lift!: () => void; + (framework as unknown as { discordAwarenessBarrier: unknown }).discordAwarenessBarrier = { + promise: new Promise((r) => { lift = r; }), + }; + let handled = 0; + (framework as unknown as { pushHandler: unknown }).pushHandler = { handlePushEvent: async () => { handled++; } }; + + const push = responder(); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e2', payload: { content: [{ type: 'text', text: 'waiting' }] } }, push); + await framework.stop(); + lift(); + await settle(); + + assert.equal(handled, 0); + assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'the host is stopping' } }]); + assert.deepEqual(rejections, []); + }); + + it('a handler that throws is answered with its error, not rejected into the listener; a notification is only logged', async () => { + const { framework, connection } = await wiredFramework(); + (framework as unknown as { pushHandler: unknown }).pushHandler = { + handlePushEvent: async () => { throw new Error('the handler broke'); }, + }; + + const push = responder(); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e3', payload: { content: [{ type: 'text', text: 'x' }] } }, push); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e4', payload: { content: [{ type: 'text', text: 'y' }] } }); + await settle(); + + assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'the handler broke' } }]); + assert.deepEqual(rejections, []); + await framework.stop(); + }); + + it('a handler that rejects with nothing is a failed request on a live host, not a stopping refusal', async () => { + const { framework, connection } = await wiredFramework(); + (framework as unknown as { pushHandler: unknown }).pushHandler = { + // eslint-disable-next-line prefer-promise-reject-errors + handlePushEvent: () => Promise.reject(), + }; + const fw = framework as unknown as { channelRegistry: unknown }; + const registry = fw.channelRegistry; + fw.channelRegistry = { handleIncoming: async () => { throw undefined; } }; + + const push = responder(); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e5', payload: { content: [{ type: 'text', text: 'x' }] } }, push); + const incoming = responder(); + connection.emit('channels-incoming', { messages: [{ channelId: 'c', messageId: 'm', content: [] }] }, incoming); + await settle(); + + assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'the handler failed' } }]); + assert.deepEqual(incoming.answers, [{ error: { code: -32603, message: 'the handler failed' } }]); + assert.ok(logged.some((line) => line.includes('push/event from srv failed: the handler failed')), logged.join('\n')); + assert.ok(!logged.some((line) => line.includes('refused')), 'a live host refuses nothing'); + assert.deepEqual(rejections, []); + fw.channelRegistry = registry; + await framework.stop(); + }); + }); +}); From aaeb2a31cd1dda7b43deeaa343fce92292953252 Mon Sep 17 00:00:00 2001 From: Nell-1783 Date: Fri, 9 Oct 2026 17:42:56 -0400 Subject: [PATCH 2/7] fix(lifecycle): a result dropped at stop() is logged with its tool and outcome; CI runs the late-arrival files unforced MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answers slimepriestess's two non-blocking notes on #263 (review at 6eb6e46). 1. The drop's log line names the tool and whether it succeeded. A result that arrives after stop() is dropped: the stream that asked is gone, and after a restart the history holds that tool_use with a placeholder result, so the resident may run the tool again (a duplicate post, for a posting tool). The line is what an operator reconciles from, and it named only the call id and module. It now reads, for example: [framework] assistant: result of tool call c1 (slow--work, slow) arrived after stop — dropped; the tool succeeded or "…; the tool failed: ". Success means result.success and not result.isError. - The tool's name wasn't on the result: ToolResultEvent carried callId, agentName and moduleName only. It gains an optional `toolName` ("as its call named it"; for a module tool that's the prefixed name, slow--work), and every tool-result the framework produces sets it: all 24 literals in framework.ts, each from the ToolCall it already holds (call.name or enrichedCall.name), checked by script to pair with its callId's variable. Basil-1007's review caught the 24th, tool presentation's compact one-line literal, which my scripted edit had missed. Optional, so a host or module constructing these events doesn't break; no producer outside framework.ts exists in the anima repositories (connectome-host, agent-framework's modules). - Tests: the three drop cases now assert the line. Success names "(slow--work, slow)" and ends "the tool succeeded"; a module throw reads "the tool failed: …"; a rejected dispatch "the tool failed: … the dispatch was lost". 2. CI runs stop-with-work-in-flight and conversation-routing unforced, after npm test. npm test passes --test-force-exit, which hides async work a test leaves running: this race showed only unforced, in a sibling PR's qualification. The step is ~a second; both files pass unforced at this head (23/23, and 3/3 runs each on the VPS). Changelog fragment: a bullet for the drop's log line and the new field. Evidence (VPS via on-vps, node v24.21.0), at this commit: tsc clean; npm test 1385/1385 (main's 1377 plus #263's 8) with no async-activity report; the CI step's command 23/23. An earlier full run, before the 24th literal got its field, counted 1373: --test-force-exit had cut history-module-realstore short, and that file alone, unforced, passes 28/28. Reviewed independently by Basil-1007 (room-369 #64428). Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 7 ++++ changelog.d/stop-with-work-in-flight.fixed.md | 4 ++ src/framework.ts | 38 +++++++++++++++---- src/types/events.ts | 4 ++ test/stop-with-work-in-flight.test.ts | 6 +++ 5 files changed, 51 insertions(+), 8 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f28642031..630e417e9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -50,3 +50,10 @@ jobs: # are asynchronous on macOS), and ubuntu-only runs have missed that. - name: Test run: npm test + + # npm test runs with --test-force-exit, which hides async work a test + # leaves running, such as a late arrival after stop() ("Queue is + # closed"). These files run unforced so that class fails here, not in a + # later PR's qualification. + - name: Test unforced (work in flight at stop) + run: node --test dist/test/stop-with-work-in-flight.test.js dist/test/conversation-routing.test.js diff --git a/changelog.d/stop-with-work-in-flight.fixed.md b/changelog.d/stop-with-work-in-flight.fixed.md index 186e58b30..3ac51bb5b 100644 --- a/changelog.d/stop-with-work-in-flight.fixed.md +++ b/changelog.d/stop-with-work-in-flight.fixed.md @@ -9,6 +9,10 @@ its push threw, the throw was reported as the tool's own failure, and that failure's push threw again, unhandled. tune-out's results now go through the same `pushEvent` as every other tool's. + - The drop's log line names the tool and whether it succeeded, which is + what decides whether anything needs reconciling (the resident may run + the tool again). For this, the framework's tool results carry the tool's + name (`ToolResultEvent.toolName`, optional). - A server's `push/event` or `channels/incoming` that arrives once the host is stopping is refused before its handler runs, with a JSON-RPC error (`-32603`, "the host is stopping"). It is never answered diff --git a/src/framework.ts b/src/framework.ts index d691b5e12..81970b204 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -2405,7 +2405,12 @@ export class AgentFramework { */ pushEvent(event: ProcessEvent): void { if (event.type === 'tool-result' && this.queue?.isClosed) { - console.error(`[framework] ${event.agentName}: result of tool call ${event.callId} (${event.moduleName}) arrived after stop — dropped`); + // What an operator reconciling a dropped result needs (the resident + // may run the tool again): which tool, and whether it succeeded. + const outcome = event.result.success && !event.result.isError + ? 'succeeded' + : `failed${event.result.error ? `: ${event.result.error}` : ''}`; + console.error(`[framework] ${event.agentName}: result of tool call ${event.callId} (${event.toolName ? `${event.toolName}, ` : ''}${event.moduleName}) arrived after stop — dropped; the tool ${outcome}`); return; } this.queue.push(event); @@ -11306,6 +11311,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: 'code_execution', result, @@ -12312,7 +12318,7 @@ export class AgentFramework { const enrichedCall: ToolCall = { ...call, callerAgentName: agentName }; if (isPresentationTool(call.name)) { const result = this.editToolPresentation(agentName, enrichedCall); - this.pushEvent({type:'tool-result',callId:call.id,agentName,moduleName:'tool-presentation',result}); + this.pushEvent({type:'tool-result',callId:call.id,toolName:call.name,agentName,moduleName:'tool-presentation',result}); return; } @@ -12380,6 +12386,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: enrichedCall.id, + toolName: enrichedCall.name, agentName, moduleName: 'tune-out', result, @@ -12450,6 +12457,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: enrichedCall.id, + toolName: enrichedCall.name, agentName, moduleName: 'framework', result: { success: true, data: PROSE_ROUTING_HELP }, @@ -12499,6 +12507,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName, result, @@ -12520,6 +12529,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName, result: { @@ -14847,6 +14857,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: `mcpl:${serverId}`, result: { success: false, error: `MCPL server not found: ${serverId}`, isError: true }, @@ -14861,6 +14872,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: `mcpl:${serverId}`, result: { @@ -15020,6 +15032,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: `mcpl:${serverId}`, result: { @@ -15039,6 +15052,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: `mcpl:${serverId}`, result: { success: false, error: err.message, isError: true }, @@ -15070,6 +15084,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: 'channels', result: { success: false, error, isError: true }, @@ -15138,6 +15153,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: 'channels', result, @@ -15150,6 +15166,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: 'channels', result: { success: false, error: err.message, isError: true }, @@ -15171,7 +15188,7 @@ export class AgentFramework { module: 'gate', tool: call.name, callId: call.id, durationMs: 0, ...(result.isError ? { error: result.error } : {}), }); - this.pushEvent({ type: 'tool-result', callId: call.id, agentName, moduleName: 'gate', result }); + this.pushEvent({ type: 'tool-result', callId: call.id, toolName: call.name, agentName, moduleName: 'gate', result }); }; try { @@ -15344,6 +15361,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: 'tune-out', result, @@ -15357,7 +15375,7 @@ export class AgentFramework { const finish = (result: ToolResult) => { this.emitTrace({ type: 'tool:completed', module: 'gate', tool: call.name, callId: call.id, durationMs: 0 }); - this.pushEvent({ type: 'tool-result', callId: call.id, agentName, moduleName: 'gate', result }); + this.pushEvent({ type: 'tool-result', callId: call.id, toolName: call.name, agentName, moduleName: 'gate', result }); }; if (call.name === 'wake') { @@ -15460,7 +15478,7 @@ export class AgentFramework { ...(result.isError ? { error: result.error } : {}), }); this.pushEvent({ - type: 'tool-result', callId: call.id, agentName, moduleName: 'workspace', + type: 'tool-result', callId: call.id, toolName: call.name, agentName, moduleName: 'workspace', result: { ...result, isError: result.isError ?? false }, }); })(); @@ -15477,7 +15495,7 @@ export class AgentFramework { durationMs: 0, ...(result.isError ? { error: result.error } : {}), }); - this.pushEvent({ type: 'tool-result', callId: call.id, agentName, moduleName: 'workspace', result }); + this.pushEvent({ type: 'tool-result', callId: call.id, toolName: call.name, agentName, moduleName: 'workspace', result }); }; void (async (): Promise => { try { @@ -15616,7 +15634,7 @@ export class AgentFramework { durationMs: 0, ...(result.isError ? { error: result.error } : {}), }); - this.pushEvent({ type: 'tool-result', callId: call.id, agentName, moduleName: 'workspace', result }); + this.pushEvent({ type: 'tool-result', callId: call.id, toolName: call.name, agentName, moduleName: 'workspace', result }); }; type Candidate = | { kind: 'attachment'; data: string | null; mediaType: string; messagesBack: number } @@ -16020,6 +16038,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: 'framework', result, @@ -16193,6 +16212,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: 'agent', result, @@ -16233,7 +16253,7 @@ export class AgentFramework { this.toolLifecycleEmitter?.markDispatchFailure(agentName, call.id); result = { success: false, error: err.message, isError: true }; } - this.pushEvent({ type: 'tool-result', callId: call.id, agentName, moduleName: 'gate', result }); + this.pushEvent({ type: 'tool-result', callId: call.id, toolName: call.name, agentName, moduleName: 'gate', result }); } private dispatchGateToolCall(agentName: string, call: ToolCall): void { @@ -16247,6 +16267,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: 'gate', result, @@ -16259,6 +16280,7 @@ export class AgentFramework { this.pushEvent({ type: 'tool-result', callId: call.id, + toolName: call.name, agentName, moduleName: 'gate', result: { success: false, error: err.message, isError: true }, diff --git a/src/types/events.ts b/src/types/events.ts index b5e8afb8e..f78e5ee99 100644 --- a/src/types/events.ts +++ b/src/types/events.ts @@ -92,6 +92,10 @@ export interface ToolResultEvent { agentName: string; /** Module that handled the call */ moduleName: string; + /** The tool's name, as its call named it. The framework's own + * producers set it, so a result nothing can take any more (one arriving + * after stop()) is still logged by tool. */ + toolName?: string; /** Result of the tool call */ result: ToolResult; } diff --git a/test/stop-with-work-in-flight.test.ts b/test/stop-with-work-in-flight.test.ts index b346cada2..9fe03c06d 100644 --- a/test/stop-with-work-in-flight.test.ts +++ b/test/stop-with-work-in-flight.test.ts @@ -143,6 +143,10 @@ describe('stop() with work still in flight', () => { assert.deepEqual(rejections, []); assert.deepEqual(failed, [], 'the tool succeeded; nothing reports it as a failure'); assert.equal(dropped().length, 1, logged.join('\n')); + // The drop names the tool and that it succeeded: what an operator needs + // to reconcile a result nothing could take. + assert.match(dropped()[0]!, /tool call c1 \(slow--work, slow\)/, dropped()[0]); + assert.match(dropped()[0]!, /the tool succeeded$/, dropped()[0]); }); it('a tool that fails after stop() has its error result dropped, not thrown', async () => { @@ -153,6 +157,7 @@ describe('stop() with work still in flight', () => { assert.deepEqual(rejections, []); assert.deepEqual(failed, [], 'the registry answers a module throw with an error result'); assert.equal(dropped().length, 1, logged.join('\n')); + assert.match(dropped()[0]!, /the tool failed: /, dropped()[0]); }); it('a dispatch that rejects after stop() is reported as failed, and its failure result dropped, not thrown', async () => { @@ -163,6 +168,7 @@ describe('stop() with work still in flight', () => { assert.deepEqual(rejections, []); assert.equal(failed.length, 1, 'the host got no result: a dispatch failure'); assert.equal(dropped().length, 1, logged.join('\n')); + assert.match(dropped()[0]!, /the tool failed: .*the dispatch was lost/, dropped()[0]); }); it('fresh input to a stopped framework still throws to its caller', async () => { From 1d77b44d73a2199461c397641ffc9dd0abd0f1b9 Mon Sep 17 00:00:00 2001 From: Nell-1783 Date: Fri, 9 Oct 2026 18:50:40 -0400 Subject: [PATCH 3/7] test(lifecycle): a dropped failure result names its tool too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Checking aaeb2a3 as a change of its own (skill:preparing-upstream-pull- requests: each push after review gets the baseline and its pieces removed), one piece had no test: the toolName on dispatchToolCallEvent's failure-path result. Removing it left stop-with-work-in-flight at 8/8, since the rejected-dispatch test asserted the outcome ("the tool failed: … the dispatch was lost") but not the tool. It now asserts "(slow--work, slow)" as the success test does; with that removal it fails 7/8. The other four pieces of aaeb2a3 each fail a test when removed: the success path's toolName (the success test), the outcome's success test (both failure tests), the line's tool name (the success test) and the line's outcome (all three). The remaining 22 producers' toolName has no test of its own; a script pairs every `type: 'tool-result'` literal's toolName with its callId's variable (24 of 24). Evidence (VPS via on-vps, node v24.21.0): red-first, aaeb2a3's tests on 6eb6e46's source fail 3 of 8 (the three drop tests); at this commit tsc is clean, npm test 1385/1385, the CI step's command 23/23. aaeb2a3's own CI passed on all four cells, the unforced step 23/23 on each. Co-Authored-By: Claude Opus 5.5 --- test/stop-with-work-in-flight.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/test/stop-with-work-in-flight.test.ts b/test/stop-with-work-in-flight.test.ts index 9fe03c06d..facc30cdd 100644 --- a/test/stop-with-work-in-flight.test.ts +++ b/test/stop-with-work-in-flight.test.ts @@ -168,6 +168,7 @@ describe('stop() with work still in flight', () => { assert.deepEqual(rejections, []); assert.equal(failed.length, 1, 'the host got no result: a dispatch failure'); assert.equal(dropped().length, 1, logged.join('\n')); + assert.match(dropped()[0]!, /tool call c1 \(slow--work, slow\)/, dropped()[0]); assert.match(dropped()[0]!, /the tool failed: .*the dispatch was lost/, dropped()[0]); }); From 97178208eb761bcfc0fe29fdc4315be9d9dd55c5 Mon Sep 17 00:00:00 2001 From: Nell-1783 Date: Sun, 11 Oct 2026 09:25:42 -0400 Subject: [PATCH 4/7] fix(lifecycle): every MCPL request settles at stop(), a stopping host answers -32000, and tune-out's rejection is answered From the house's haiku review of this PR (30 issues; the author's verdicts are recorded with each one's reason). Code: - tune-out's subconscious tool chain had no rejection handler, older than this PR but its own class: a rejection of handleSubconsciousTool is now answered with a failed result through pushEvent, as dispatchUtilsToolCall answers a dispatch failure, instead of an unhandled rejection with the call never answered. - inference/request, channels/register and channels/changed get the same boundary as push/event and channels/incoming: refused before the handler once the host is stopping, and a handler's throw answered (or, for a notification, logged), never rejected out of the listener. A stopping host starts no inference for a server. - One code per meaning: the push coalescer already refuses a stopping host with -32000 "host is stopping" (push-coalescer.ts), so the listeners' stop refusals are -32000 too ("the host is stopping"; a handler failing while the host stops, "the host stopped while handling this request ()"). -32603 stays for a handler's fault on a live host, with its message. - refuseServerItem keeps the handler's own message when the host stops, reads any thrown value safely (a value with no string form, such as Object.create(null), made String() throw outside the catch), bounds the text at 500 characters (also in the drop line's tool error), and doesn't answer a request twice: the listeners wrap the responder (answerOnce), and a handler that answered and then threw is logged as failed after it was answered. - pushEvent's doc no longer says the asking stream is gone at drop time: the queue that would carry the result to its stream is closed, and stop() is ending the streams. Tests (test/stop-with-work-in-flight.test.ts, 17 cases): waits on the event each needs (a log line, an answer) instead of 10 and 20 ms sleeps, then one more turn so a late rejection is seen; stop() in finally where a test stops at its end. New: the drop's isError and no-error-text clauses; tune-out's late result dropped with its tool name, and its rejection answered with a failed result; a static check that every tool-result literal in src names its tool (24 found); all five listeners refusing after stop() with -32000 and handling nothing; channels/incoming held at the awareness barrier, as push/event was; a handler awaiting across stop() and then failing (-32000 with its own message, and its log line); the notification's log line; nothing, a string and an unprintable value thrown; a 5,000-character error bounded; an answered request not answered again; a throwing responder logged, not rejected. Each code fix removed in the build fails one of them (the tune-out catch, the -32000 code, the answered check, the safe read, the inference stop check). Changelog: the title covers MCPL requests and notifications; tune-out's results emit process:received like every other tool's; the five methods and -32000; the handler-failure answers; for operators, that a handler fault no longer ends the process (a supervisor would have restarted it); and "any other event" still throws, since a tool's result doesn't. Checked: tsc clean; the two unforced files 32/32; the whole suite 1393 of 1394, the one failure being mcpl-coalescing's timed "busy channel never hits a subject cap" (43 s for 300 messages at load average ~9), which passes alone (28.5 s) and touches none of this. --- changelog.d/stop-with-work-in-flight.fixed.md | 41 +- src/framework.ts | 159 +++++-- test/stop-with-work-in-flight.test.ts | 405 ++++++++++++++---- 3 files changed, 484 insertions(+), 121 deletions(-) diff --git a/changelog.d/stop-with-work-in-flight.fixed.md b/changelog.d/stop-with-work-in-flight.fixed.md index 3ac51bb5b..f662952fb 100644 --- a/changelog.d/stop-with-work-in-flight.fixed.md +++ b/changelog.d/stop-with-work-in-flight.fixed.md @@ -1,24 +1,35 @@ -- **Late tool results, and MCPL pushes or incoming messages, at `stop()` no +- **Late tool results, and MCPL requests and notifications, at `stop()` no longer end in an unhandled rejection.** `stop()` closes the process queue before it closes MCPL connections, and nothing caught the "Queue is closed" that this meant for late arrivals. Test files that stop a framework while a tool still runs, such as `conversation-routing`, failed at file level with it. - - A tool's result that arrives after `stop()` is logged and dropped. The - stream that asked for it is gone, and the tool has already run. Before, - its push threw, the throw was reported as the tool's own failure, and - that failure's push threw again, unhandled. tune-out's results now go - through the same `pushEvent` as every other tool's. + - A tool's result that arrives after `stop()` is logged and dropped: the + queue that would carry it to its stream is closed, and the tool has + already run. Before, its push threw, the throw was reported as the + tool's own failure, and that failure's push threw again, unhandled. + - tune-out's results now go through the same `pushEvent` as every other + tool's, so they also emit `process:received`, and a rejection of its + handler is answered with a failed result, as other tools' dispatch + failures are, instead of being left unhandled. - The drop's log line names the tool and whether it succeeded, which is what decides whether anything needs reconciling (the resident may run the tool again). For this, the framework's tool results carry the tool's name (`ToolResultEvent.toolName`, optional). - - A server's `push/event` or `channels/incoming` that arrives once the - host is stopping is refused before its handler runs, with a JSON-RPC - error (`-32603`, "the host is stopping"). It is never answered - "accepted". - - A handler that throws, because the host stopped while it ran or because - of a fault, is answered with an error too. Part of the request may - already have been admitted. The error is never thrown out of the - connection's listener. A notification is logged. - - Direct input to `pushEvent` after `stop()` still throws to its caller. + - A server's `push/event`, `channels/incoming`, `channels/register`, + `channels/changed` or `inference/request` that arrives once the host is + stopping is refused before its handler runs, with a JSON-RPC error + (`-32000`, "the host is stopping", the code the push coalescer already + gives a stopping host). It is never answered "accepted", and no + inference starts. + - A handler that throws is answered with an error too: while the host + stops, `-32000` with the handler's own message after "the host stopped + while handling this request"; on a live host, `-32603` with its message. + The message is bounded, any thrown value is printed safely, and a + request the handler already answered isn't answered twice. Part of the + request may already have been admitted. A notification is logged. + - **For operators:** such a handler fault used to be an unhandled + rejection, which ends a Node process by default, so a supervisor would + have restarted the host. Now the host stays up and logs the failure. + - Any other event pushed to a stopped framework still throws to its + caller. diff --git a/src/framework.ts b/src/framework.ts index 81970b204..1caee2ef3 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -952,6 +952,50 @@ function truncateReason(reason: string, max = 160): string { return reason.length <= max ? reason : reason.slice(0, max) + '…'; } +/** The most of a thrown value's or a tool error's text that reaches a log line or a server's error answer. */ +const THROWN_TEXT_LIMIT = 500; + +/** + * What a thrown value says, safely and briefly: an Error's message, or any + * other value as a string. Reading it can't throw (a value with no string + * form, a hostile getter or a revoked Proxy reads as unreadable), and the + * text is bounded, since a provider or handler error can carry a whole + * request body. + */ +function thrownText(error: unknown): string { + let text: string; + try { + text = error instanceof Error ? error.message : String(error); + } catch { + text = '(a value that could not be read)'; + } + return text.length <= THROWN_TEXT_LIMIT ? text : `${text.slice(0, THROWN_TEXT_LIMIT)}… (${text.length} characters)`; +} + +/** A server's responder, and whether the request it belongs to has been answered yet. */ +type AnswerOnce = { + responder: { respond?: (result: unknown) => void; respondError?: (code: number, message: string, data?: unknown) => void } | undefined; + answered: () => boolean; +}; + +/** + * Wrap a server request's responder so a later failure can tell whether the + * handler already answered: a request is answered once, and a handler that + * answered and then threw is logged, not answered a second time. + */ +function answerOnce void; respondError?: (code: number, message: string, data?: unknown) => void }>( + responder: R | undefined, +): AnswerOnce & { responder: R | undefined } { + if (!responder) return { responder: undefined, answered: () => false }; + let done = false; + const wrapped = { + ...responder, + respond: (result: unknown) => { done = true; responder.respond?.(result); }, + respondError: (code: number, message: string, data?: unknown) => { done = true; responder.respondError?.(code, message, data); }, + } as R; + return { responder: wrapped, answered: () => done }; +} + export class AgentFramework { private toolPresentations = new Map(); private presentationPreviews = new WeakMap(); @@ -2357,37 +2401,51 @@ export class AgentFramework { } /** - * A server's push or incoming message the host couldn't handle, answered + * A server's request or notification the host couldn't handle, answered * with a JSON-RPC error when it is a request (a notification has no one to * answer, and is logged), rather than thrown out of the connection's * listener, where it was an unhandled rejection and the server's request * waited for its timeout. * - `refused`: the listener turned it away before its handler ran; the - * host is stopping, and none of it was taken. - * - `failed`: the handler threw (the host stopped while it ran, or a - * fault; whatever it threw, `undefined` included). The request failed, - * but the handler may have admitted part of it first (an earlier message - * of a batch), so nothing here claims that none of it was taken. + * host is stopping, and none of it was taken. Answered -32000, "the host + * is stopping", the code the push coalescer already gives a stopping + * host: the server can send it again to the host's next run. + * - `failed`: the handler threw, whatever it threw, `undefined` included. + * While the host stops (its queue closed), that's answered -32000 too, + * with the handler's own message after the stop wording; on a live host + * it's a fault, answered -32603 with the handler's message. The handler + * may have admitted part of the request first (an earlier message of a + * batch), so nothing here claims that none of it was taken. + * A request the handler already answered before it threw isn't answered + * again: the failure is logged. */ private refuseServerItem( method: string, serverId: string, - responder: { respondError?: (code: number, message: string) => void } | undefined, + answer: AnswerOnce, outcome: { kind: 'refused' } | { kind: 'failed'; error: unknown }, ): void { // Optional chaining: prototype-built harnesses leave the queue unset. + const stopping = outcome.kind === 'refused' || !!this.queue?.isClosed; + const what = outcome.kind === 'refused' + ? '' + : outcome.error instanceof Error + ? thrownText(outcome.error) + : `the handler failed${outcome.error === undefined ? '' : `: ${thrownText(outcome.error)}`}`; const reason = outcome.kind === 'refused' ? 'the host is stopping' - : this.queue?.isClosed - ? 'the host stopped while handling this request' - : outcome.error instanceof Error - ? outcome.error.message - : `the handler failed${outcome.error === undefined ? '' : `: ${String(outcome.error)}`}`; + : stopping + ? `the host stopped while handling this request (${what})` + : what; + if (outcome.kind === 'failed' && answer.answered()) { + console.error(`[mcpl] ${method} from ${serverId} failed after it was answered: ${reason}`); + return; + } console.error(outcome.kind === 'refused' ? `[mcpl] ${method} from ${serverId} refused: ${reason}` : `[mcpl] ${method} from ${serverId} failed: ${reason} (part of it may already have been admitted)`); try { - responder?.respondError?.(-32603, reason); + answer.responder?.respondError?.(stopping ? -32000 : -32603, reason); } catch { // The connection is going too; there is no one left to tell. } @@ -2400,8 +2458,9 @@ export class AgentFramework { * learns the framework is stopped — except for a tool's result. A call * still running at stop() completes later, into a callback with nothing * to catch the throw (an unhandled rejection), and nothing could take its - * result anyway: the stream that asked is gone, and the tool has already - * run. That result is dropped, and the drop is logged. + * result anyway: the queue that would carry it to its stream is closed, + * stop() is ending the streams, and the tool has already run. That + * result is dropped, and the drop is logged. */ pushEvent(event: ProcessEvent): void { if (event.type === 'tool-result' && this.queue?.isClosed) { @@ -2409,7 +2468,7 @@ export class AgentFramework { // may run the tool again): which tool, and whether it succeeded. const outcome = event.result.success && !event.result.isError ? 'succeeded' - : `failed${event.result.error ? `: ${event.result.error}` : ''}`; + : `failed${event.result.error ? `: ${thrownText(event.result.error)}` : ''}`; console.error(`[framework] ${event.agentName}: result of tool call ${event.callId} (${event.toolName ? `${event.toolName}, ` : ''}${event.moduleName}) arrived after stop — dropped; the tool ${outcome}`); return; } @@ -12380,6 +12439,9 @@ export class AgentFramework { ) { void this.tuneOutCoordinator .handleSubconsciousTool(enrichedCall.name, enrichedCall.input as Record) + // A rejection is answered with a failed result, as a framework tool's + // dispatch failure is (dispatchUtilsToolCall), never left unhandled. + .catch((err): ToolResult => ({ success: false, error: thrownText(err), isError: true })) .then((result) => { // Through pushEvent, like every other tool's result, so one that // arrives after stop() is dropped rather than rejected unhandled. @@ -14340,18 +14402,19 @@ export class AgentFramework { params: PushEventParams, responder?: { respond: (result: unknown) => void; respondError: (code: number, message: string) => void }, ) => { + const answer = answerOnce(responder); try { const barrier = this.discordAwarenessBarrier; if (barrier) await barrier.promise; // Checked here, after the barrier and before anything records the // push as accepted: a stopped host refuses it. if (this.queue?.isClosed) { - this.refuseServerItem('push/event', connection.id, responder, { kind: 'refused' }); + this.refuseServerItem('push/event', connection.id, answer, { kind: 'refused' }); return; } - await this.pushHandler?.handlePushEvent(connection.id, params, responder as never); + await this.pushHandler?.handlePushEvent(connection.id, params, answer.responder as never); } catch (error) { - this.refuseServerItem('push/event', connection.id, responder, { kind: 'failed', error }); + this.refuseServerItem('push/event', connection.id, answer, { kind: 'failed', error }); } }); @@ -14366,22 +14429,42 @@ export class AgentFramework { return; } if (this.inferenceRouter && responder) { - const barrier = this.discordAwarenessBarrier; - if (barrier) await barrier.promise; - await this.inferenceRouter.handleInferenceRequest(connection.id, params, { - respond: responder.respond, - respondError: responder.respondError, - requestId: responder.id, - }); + const answer = answerOnce(responder); + try { + const barrier = this.discordAwarenessBarrier; + if (barrier) await barrier.promise; + // A stopping host starts no inference for a server: refused, like a + // push, before the router runs. + if (this.queue?.isClosed) { + this.refuseServerItem('inference/request', connection.id, answer, { kind: 'refused' }); + return; + } + await this.inferenceRouter.handleInferenceRequest(connection.id, params, { + respond: answer.responder!.respond, + respondError: answer.responder!.respondError, + requestId: responder.id, + }); + } catch (error) { + this.refuseServerItem('inference/request', connection.id, answer, { kind: 'failed', error }); + } } }); // Handle channel registration (Step 7) connection.on('channels-register', async ( params: ChannelsRegisterParams, - responder?: { respond: (result: unknown) => void }, + responder?: { respond: (result: unknown) => void; respondError?: (code: number, message: string) => void }, ) => { - await this.channelRegistry?.handleRegister(connection.id, params, responder as never); + const answer = answerOnce(responder); + try { + if (this.queue?.isClosed) { + this.refuseServerItem('channels/register', connection.id, answer, { kind: 'refused' }); + return; + } + await this.channelRegistry?.handleRegister(connection.id, params, answer.responder as never); + } catch (error) { + this.refuseServerItem('channels/register', connection.id, answer, { kind: 'failed', error }); + } }); // Handle channel changes (Step 7) — §14.5 dual-mode: the responder is @@ -14390,9 +14473,18 @@ export class AgentFramework { // filtering plus a diagnostic trace. connection.on('channels-changed', async ( params: ChannelsChangedParams, - responder?: { respond: (result: unknown) => void }, + responder?: { respond: (result: unknown) => void; respondError?: (code: number, message: string) => void }, ) => { - await this.channelRegistry?.handleChanged(connection.id, params, responder as never); + const answer = answerOnce(responder); + try { + if (this.queue?.isClosed) { + this.refuseServerItem('channels/changed', connection.id, answer, { kind: 'refused' }); + return; + } + await this.channelRegistry?.handleChanged(connection.id, params, answer.responder as never); + } catch (error) { + this.refuseServerItem('channels/changed', connection.id, answer, { kind: 'failed', error }); + } }); // §6.6/§12: requests that previously had no handler and hung forever @@ -14453,16 +14545,17 @@ export class AgentFramework { params: ChannelsIncomingParams, responder?: { respond: (result: unknown) => void; respondError?: (code: number, message: string) => void }, ) => { + const answer = answerOnce(responder); try { const barrier = this.discordAwarenessBarrier; if (barrier) await barrier.promise; if (this.queue?.isClosed) { - this.refuseServerItem('channels/incoming', connection.id, responder, { kind: 'refused' }); + this.refuseServerItem('channels/incoming', connection.id, answer, { kind: 'refused' }); return; } - await this.channelRegistry?.handleIncoming(connection.id, params, responder as never); + await this.channelRegistry?.handleIncoming(connection.id, params, answer.responder as never); } catch (error) { - this.refuseServerItem('channels/incoming', connection.id, responder, { kind: 'failed', error }); + this.refuseServerItem('channels/incoming', connection.id, answer, { kind: 'failed', error }); } }); diff --git a/test/stop-with-work-in-flight.test.ts b/test/stop-with-work-in-flight.test.ts index facc30cdd..b4dcc3837 100644 --- a/test/stop-with-work-in-flight.test.ts +++ b/test/stop-with-work-in-flight.test.ts @@ -10,16 +10,19 @@ * stream that asked is gone, and the tool has already run. Before, its * push threw "Queue is closed", the throw was misreported as the tool's * failure, and that failure's push threw again, unhandled; - * - a server's push or incoming message the host can't take any more is - * answered with a JSON-RPC error, never "accepted", and never thrown - * out of the connection's listener. - * Fresh input to a stopped framework still throws to its caller. + * - a server's push, incoming message or other request the host can't + * take any more is answered with a JSON-RPC error (-32000, "the host is + * stopping"), never "accepted", and never thrown out of the + * connection's listener; a handler's own fault is answered -32603. + * Any other event pushed to a stopped framework still throws to its caller. + * Each test waits on the event it needs (a log line, an answer), not on a + * fixed delay, and then one more turn, so a late rejection would be seen. */ import { describe, it, beforeEach, afterEach } from 'node:test'; import assert from 'node:assert/strict'; import { EventEmitter } from 'node:events'; -import { mkdtempSync, rmSync } from 'node:fs'; +import { mkdtempSync, readFileSync, readdirSync, rmSync, statSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import type { ContentBlock } from '@animalabs/membrane'; @@ -36,6 +39,18 @@ import type { import { AgentFramework } from '../src/index.js'; import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; +/** Resolve once `cond` holds, polling each turn; fail, naming `what`, after `ms`. */ +async function until(cond: () => boolean, what: string, ms = 5000): Promise { + const deadline = Date.now() + ms; + while (!cond()) { + if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`); + await new Promise((r) => setImmediate(r)); + } +} + +/** One more turn of the event loop: an unhandled rejection is reported by then. */ +const turn = (): Promise => new Promise((r) => setImmediate(() => setImmediate(r))); + /** A tool that settles only when the test says so, so the test owns the work it started. */ class SlowModule implements Module { readonly name = 'slow'; @@ -61,11 +76,19 @@ class SlowModule implements Module { }); } - /** Settle the running call; resolves once its completion callbacks have run. */ - async finish(outcome: 'success' | 'failure'): Promise { + /** + * Settle the running call. The caller waits for what it expects to follow + * (the drop's log line); this returns at once. + * - success: { success: true } + * - failure: the module throws (the registry answers an error result) + * - isError: { success: true, isError: true }, a result that reports an error + * - bare-failure: { success: false } with no error text + */ + finish(outcome: 'success' | 'failure' | 'isError' | 'bare-failure'): void { if (outcome === 'success') this.settle!.resolve({ success: true, data: { done: true } }); + else if (outcome === 'isError') this.settle!.resolve({ success: true, isError: true, data: { done: false } }); + else if (outcome === 'bare-failure') this.settle!.resolve({ success: false }); else this.settle!.reject(new Error('the work failed')); - await new Promise((r) => setTimeout(r, 20)); } async onProcess(event: ProcessEvent, _state: ProcessState): Promise { @@ -135,54 +158,174 @@ describe('stop() with work still in flight', () => { return { framework, module, failed }; } + /** Wait for the drop's log line, then a turn, so a late rejection would be seen. */ + async function droppedLine(): Promise { + await until(() => dropped().length > 0, 'the drop log line'); + await turn(); + assert.equal(dropped().length, 1, logged.join('\n')); + return dropped()[0]!; + } + it('a tool that succeeds after stop() has its result dropped, not thrown or misreported as a failure', async () => { const { framework, module, failed } = await frameworkWithRunningTool(); await framework.stop(); - await module.finish('success'); + module.finish('success'); + const line = await droppedLine(); assert.deepEqual(rejections, []); assert.deepEqual(failed, [], 'the tool succeeded; nothing reports it as a failure'); - assert.equal(dropped().length, 1, logged.join('\n')); // The drop names the tool and that it succeeded: what an operator needs // to reconcile a result nothing could take. - assert.match(dropped()[0]!, /tool call c1 \(slow--work, slow\)/, dropped()[0]); - assert.match(dropped()[0]!, /the tool succeeded$/, dropped()[0]); + assert.match(line, /tool call c1 \(slow--work, slow\)/, line); + assert.match(line, /the tool succeeded$/, line); }); it('a tool that fails after stop() has its error result dropped, not thrown', async () => { const { framework, module, failed } = await frameworkWithRunningTool(); await framework.stop(); - await module.finish('failure'); + module.finish('failure'); + const line = await droppedLine(); assert.deepEqual(rejections, []); assert.deepEqual(failed, [], 'the registry answers a module throw with an error result'); - assert.equal(dropped().length, 1, logged.join('\n')); - assert.match(dropped()[0]!, /the tool failed: /, dropped()[0]); + assert.match(line, /the tool failed: /, line); + }); + + it('a result that reports an error is dropped as failed, and a failure with no text says only that', async () => { + // { success: true, isError: true } is a failure: the drop must not call it a success. + { + const { framework, module } = await frameworkWithRunningTool(); + await framework.stop(); + module.finish('isError'); + const line = await droppedLine(); + assert.match(line, /the tool failed$/, line); + logged.length = 0; + } + { + const { framework, module } = await frameworkWithRunningTool(); + await framework.stop(); + module.finish('bare-failure'); + const line = await droppedLine(); + assert.match(line, /the tool failed$/, line); + } + assert.deepEqual(rejections, []); }); it('a dispatch that rejects after stop() is reported as failed, and its failure result dropped, not thrown', async () => { const { framework, module, failed } = await frameworkWithRunningTool({ dispatch: 'rejects' }); await framework.stop(); - await module.finish('success'); + module.finish('success'); + const line = await droppedLine(); assert.deepEqual(rejections, []); assert.equal(failed.length, 1, 'the host got no result: a dispatch failure'); - assert.equal(dropped().length, 1, logged.join('\n')); - assert.match(dropped()[0]!, /tool call c1 \(slow--work, slow\)/, dropped()[0]); - assert.match(dropped()[0]!, /the tool failed: .*the dispatch was lost/, dropped()[0]); + assert.match(line, /tool call c1 \(slow--work, slow\)/, line); + assert.match(line, /the tool failed: .*the dispatch was lost/, line); }); - it('fresh input to a stopped framework still throws to its caller', async () => { + it('any other event pushed to a stopped framework still throws to its caller', async () => { const { framework, module } = await frameworkWithRunningTool(); await framework.stop(); assert.throws( () => framework.pushEvent({ type: 'external-message', source: 'test', content: 'too late', metadata: {} } as unknown as ProcessEvent), /Queue is closed/, ); - await module.finish('success'); + module.finish('success'); + await droppedLine(); assert.deepEqual(rejections, []); }); + describe("tune-out's subconscious tools", () => { + /** A framework whose tune-out coordinator is a stand-in that settles when the test says. */ + async function withTuneOut() { + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'store'), + membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'test' }], + modules: [], + }); + let settle!: { resolve: (r: ToolResult) => void; reject: (e: unknown) => void }; + const fw = framework as unknown as { + tuneOutCoordinator: unknown; subconsciousAgentName: string | null; + dispatchToolCall(agentName: string, call: ToolCall): void; + }; + fw.tuneOutCoordinator = { + handleSubconsciousTool: () => new Promise((resolve, reject) => { settle = { resolve, reject }; }), + stop: () => {}, + }; + fw.subconsciousAgentName = 'assistant'; + const pushed: ProcessEvent[] = []; + const push = framework.pushEvent.bind(framework); + framework.pushEvent = (event: ProcessEvent) => { pushed.push(event); push(event); }; + fw.dispatchToolCall('assistant', { id: 't1', name: 'deliver_summary', input: {} } as ToolCall); + return { framework, settle: () => settle, pushed }; + } + + it('a result that arrives after stop() is dropped and logged with its tool, not thrown', async () => { + const { framework, settle } = await withTuneOut(); + await framework.stop(); + settle().resolve({ success: true, data: { delivered: true } }); + const line = await droppedLine(); + assert.match(line, /tool call t1 \(deliver_summary, tune-out\)/, line); + assert.match(line, /the tool succeeded$/, line); + assert.deepEqual(rejections, []); + }); + + it('a handler that rejects is answered with a failed result, never left unhandled', async () => { + const { framework, settle, pushed } = await withTuneOut(); + try { + settle().reject(new Error('the summary failed')); + await until(() => pushed.length > 0, 'the failed result'); + await turn(); + const result = (pushed[0] as unknown as { result: ToolResult }).result; + assert.deepEqual(result, { success: false, error: 'the summary failed', isError: true }); + assert.deepEqual(rejections, []); + } finally { + await framework.stop(); + } + }); + }); + + it('every tool result the framework produces names its tool', () => { + // The drop's log line names the tool from the event's toolName, so a + // producer that leaves it out logs a drop nobody can reconcile. Every + // object literal in src with type: 'tool-result' carries toolName. + const src = new URL('../../src/', import.meta.url).pathname; + const files: string[] = []; + const walk = (dir: string): void => { + for (const name of readdirSync(dir)) { + const full = join(dir, name); + if (statSync(full).isDirectory()) walk(full); + else if (name.endsWith('.ts')) files.push(full); + } + }; + walk(src); + const missing: string[] = []; + let literals = 0; + for (const file of files) { + const text = readFileSync(file, 'utf8'); + for (const m of text.matchAll(/type:\s*'tool-result',/g)) { + literals++; + let start = m.index!; + for (let depth = 0; start > 0; start--) { + const c = text[start - 1]; + if (c === '}') depth++; + else if (c === '{') { if (depth === 0) { start--; break; } depth--; } + } + let end = start; + for (let depth = 0; end < text.length; end++) { + if (text[end] === '{') depth++; + else if (text[end] === '}' && --depth === 0) break; + } + if (!/\btoolName\b/.test(text.slice(start, end + 1))) { + missing.push(`${file.slice(src.length)}:${text.slice(0, m.index).split('\n').length}`); + } + } + } + assert.ok(literals > 20, `found ${literals} tool-result literals; the scan has lost them`); + assert.deepEqual(missing, [], 'tool-result events without toolName'); + }); + describe('a server sending while the host stops', () => { async function wiredFramework() { const framework = await AgentFramework.create({ @@ -196,94 +339,210 @@ describe('stop() with work still in flight', () => { return { framework, connection }; } - function responder() { + function responder(opts: { throws?: boolean } = {}) { const answers: Array<{ result?: unknown; error?: { code: number; message: string } }> = []; return { + id: 7, answers, respond: (result: unknown) => { answers.push({ result }); }, - respondError: (code: number, message: string) => { answers.push({ error: { code, message } }); }, + respondError: (code: number, message: string) => { + answers.push({ error: { code, message } }); + if (opts.throws) throw new Error('the connection is gone'); + }, }; } - const settle = (): Promise => new Promise((r) => setTimeout(r, 10)); + /** Wait until each responder has its answer, then a turn. */ + async function answered(...rs: Array<{ answers: unknown[] }>): Promise { + await until(() => rs.every((r) => r.answers.length > 0), 'every answer'); + await turn(); + } - it('a push or incoming message after stop() is answered with an error, never accepted', async () => { + const stopping = { error: { code: -32000, message: 'the host is stopping' } }; + + it('a push, incoming message or other request after stop() is answered -32000, never handled or accepted', async () => { const { framework, connection } = await wiredFramework(); - let handled = 0; - (framework as unknown as { pushHandler: unknown }).pushHandler = { handlePushEvent: async () => { handled++; } }; await framework.stop(); + // Stand-in handlers, put in after stop() (which stops the real registry): every arrival below comes later. + let handled = 0; + const fw = framework as unknown as { pushHandler: unknown; channelRegistry: unknown; inferenceRouter: unknown }; + fw.pushHandler = { handlePushEvent: async () => { handled++; } }; + fw.channelRegistry = { + handleIncoming: async () => { handled++; }, + handleRegister: async () => { handled++; }, + handleChanged: async () => { handled++; }, + }; + fw.inferenceRouter = { handleInferenceRequest: async () => { handled++; } }; const push = responder(); connection.emit('push-event', { featureSet: 'chat', eventId: 'e1', payload: { content: [{ type: 'text', text: 'late' }] } }, push); const incoming = responder(); connection.emit('channels-incoming', { messages: [{ channelId: 'c', messageId: 'm', content: [] }] }, incoming); - await settle(); - - assert.equal(handled, 0, 'never handed to the push handler, which would record it as accepted'); - assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'the host is stopping' } }]); - assert.deepEqual(incoming.answers, [{ error: { code: -32603, message: 'the host is stopping' } }]); + const register = responder(); + connection.emit('channels-register', { channels: [] }, register); + const changed = responder(); + connection.emit('channels-changed', { added: [] }, changed); + const inference = responder(); + connection.emit('inference-request', { messages: [] }, inference); + await answered(push, incoming, register, changed, inference); + + assert.equal(handled, 0, 'never handed to a handler, which would record it as accepted'); + for (const r of [push, incoming, register, changed, inference]) assert.deepEqual(r.answers, [stopping]); assert.deepEqual(rejections, []); }); - it('a push held at the awareness barrier while the host stops is refused once the barrier lifts', async () => { + for (const [event, method, params] of [ + ['push-event', 'push/event', { featureSet: 'chat', eventId: 'e2', payload: { content: [{ type: 'text', text: 'waiting' }] } }], + ['channels-incoming', 'channels/incoming', { messages: [{ channelId: 'c', messageId: 'm2', content: [] }] }], + ] as const) { + it(`a ${method} held at the awareness barrier while the host stops is refused once the barrier lifts`, async () => { + const { framework, connection } = await wiredFramework(); + let lift!: () => void; + (framework as unknown as { discordAwarenessBarrier: unknown }).discordAwarenessBarrier = { + promise: new Promise((r) => { lift = r; }), + }; + let handled = 0; + const fw = framework as unknown as { pushHandler: unknown; channelRegistry: unknown }; + fw.pushHandler = { handlePushEvent: async () => { handled++; } }; + fw.channelRegistry = { handleIncoming: async () => { handled++; }, stopAll: () => {} }; + + const r = responder(); + connection.emit(event, params, r); + await framework.stop(); + lift(); + await answered(r); + + assert.equal(handled, 0); + assert.deepEqual(r.answers, [stopping]); + assert.ok(logged.some((line) => line.includes(`${method} from srv refused: the host is stopping`)), logged.join('\n')); + assert.deepEqual(rejections, []); + }); + } + + it('a handler running when the host stops, and then failing, is answered -32000 with its own message', async () => { const { framework, connection } = await wiredFramework(); - let lift!: () => void; - (framework as unknown as { discordAwarenessBarrier: unknown }).discordAwarenessBarrier = { - promise: new Promise((r) => { lift = r; }), + let release!: () => void; + const gate = new Promise((r) => { release = r; }); + (framework as unknown as { pushHandler: unknown }).pushHandler = { + handlePushEvent: async () => { await gate; throw new Error('Queue is closed'); }, }; - let handled = 0; - (framework as unknown as { pushHandler: unknown }).pushHandler = { handlePushEvent: async () => { handled++; } }; const push = responder(); - connection.emit('push-event', { featureSet: 'chat', eventId: 'e2', payload: { content: [{ type: 'text', text: 'waiting' }] } }, push); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e6', payload: { content: [{ type: 'text', text: 'x' }] } }, push); await framework.stop(); - lift(); - await settle(); + release(); + await answered(push); - assert.equal(handled, 0); - assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'the host is stopping' } }]); + const reason = 'the host stopped while handling this request (Queue is closed)'; + assert.deepEqual(push.answers, [{ error: { code: -32000, message: reason } }]); + assert.ok(logged.some((line) => line.includes(`push/event from srv failed: ${reason} (part of it may already have been admitted)`)), logged.join('\n')); assert.deepEqual(rejections, []); }); - it('a handler that throws is answered with its error, not rejected into the listener; a notification is only logged', async () => { + it('a handler that throws is answered -32603 with its error, not rejected into the listener; a notification is only logged', async () => { const { framework, connection } = await wiredFramework(); - (framework as unknown as { pushHandler: unknown }).pushHandler = { - handlePushEvent: async () => { throw new Error('the handler broke'); }, - }; + try { + (framework as unknown as { pushHandler: unknown }).pushHandler = { + handlePushEvent: async (_s: string, params: { eventId: string }) => { throw new Error(`the handler broke on ${params.eventId}`); }, + }; + + const push = responder(); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e3', payload: { content: [{ type: 'text', text: 'x' }] } }, push); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e4', payload: { content: [{ type: 'text', text: 'y' }] } }); + await answered(push); + await until(() => logged.some((line) => line.includes('the handler broke on e4')), "the notification's log line"); + + assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'the handler broke on e3' } }]); + assert.ok(logged.some((line) => line.includes('push/event from srv failed: the handler broke on e4 (part of it may already have been admitted)')), logged.join('\n')); + assert.deepEqual(rejections, []); + } finally { + await framework.stop(); + } + }); - const push = responder(); - connection.emit('push-event', { featureSet: 'chat', eventId: 'e3', payload: { content: [{ type: 'text', text: 'x' }] } }, push); - connection.emit('push-event', { featureSet: 'chat', eventId: 'e4', payload: { content: [{ type: 'text', text: 'y' }] } }); - await settle(); + it('a handler that rejects with nothing, a string or an unprintable value is a failed request, not a stopping refusal', async () => { + const { framework, connection } = await wiredFramework(); + try { + const fw = framework as unknown as { pushHandler: unknown; channelRegistry: unknown }; + const thrown: Record = { e5: undefined, e7: 'a plain string', e8: Object.create(null) }; + fw.pushHandler = { + // eslint-disable-next-line prefer-promise-reject-errors + handlePushEvent: (_s: string, params: { eventId: string }) => Promise.reject(thrown[params.eventId]), + }; + fw.channelRegistry = { handleIncoming: async () => { throw undefined; }, stopAll: () => {} }; + + const answersFor = (eventId: string) => { + const r = responder(); + connection.emit('push-event', { featureSet: 'chat', eventId, payload: { content: [{ type: 'text', text: 'x' }] } }, r); + return r; + }; + const e5 = answersFor('e5'); + const e7 = answersFor('e7'); + const e8 = answersFor('e8'); + const incoming = responder(); + connection.emit('channels-incoming', { messages: [{ channelId: 'c', messageId: 'm', content: [] }] }, incoming); + await answered(e5, e7, e8, incoming); + + assert.deepEqual(e5.answers, [{ error: { code: -32603, message: 'the handler failed' } }]); + assert.deepEqual(e7.answers, [{ error: { code: -32603, message: 'the handler failed: a plain string' } }]); + assert.deepEqual(e8.answers, [{ error: { code: -32603, message: 'the handler failed: (a value that could not be read)' } }]); + assert.deepEqual(incoming.answers, [{ error: { code: -32603, message: 'the handler failed' } }]); + assert.ok(!logged.some((line) => line.includes('refused')), 'a live host refuses nothing'); + assert.deepEqual(rejections, []); + } finally { + await framework.stop(); + } + }); - assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'the handler broke' } }]); - assert.deepEqual(rejections, []); - await framework.stop(); + it('a long error is answered and logged bounded', async () => { + const { framework, connection } = await wiredFramework(); + try { + (framework as unknown as { pushHandler: unknown }).pushHandler = { + handlePushEvent: async () => { throw new Error('x'.repeat(5000)); }, + }; + const push = responder(); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e9', payload: { content: [{ type: 'text', text: 'x' }] } }, push); + await answered(push); + const message = push.answers[0]!.error!.message; + assert.ok(message.length < 600, `answered ${message.length} characters`); + assert.match(message, /… \(5000 characters\)$/); + } finally { + await framework.stop(); + } }); - it('a handler that rejects with nothing is a failed request on a live host, not a stopping refusal', async () => { + it('a request the handler already answered is not answered again when the handler then throws', async () => { const { framework, connection } = await wiredFramework(); - (framework as unknown as { pushHandler: unknown }).pushHandler = { - // eslint-disable-next-line prefer-promise-reject-errors - handlePushEvent: () => Promise.reject(), - }; - const fw = framework as unknown as { channelRegistry: unknown }; - const registry = fw.channelRegistry; - fw.channelRegistry = { handleIncoming: async () => { throw undefined; } }; + try { + (framework as unknown as { pushHandler: unknown }).pushHandler = { + handlePushEvent: async (_s: string, _p: unknown, r: { respond(x: unknown): void }) => { + r.respond({ accepted: true }); + throw new Error('the trace emit broke'); + }, + }; + const push = responder(); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e10', payload: { content: [{ type: 'text', text: 'x' }] } }, push); + await answered(push); + await until(() => logged.some((line) => line.includes('failed after it was answered')), 'the log line'); + + assert.deepEqual(push.answers, [{ result: { accepted: true } }]); + assert.ok(logged.some((line) => line.includes('push/event from srv failed after it was answered: the trace emit broke')), logged.join('\n')); + assert.deepEqual(rejections, []); + } finally { + await framework.stop(); + } + }); - const push = responder(); - connection.emit('push-event', { featureSet: 'chat', eventId: 'e5', payload: { content: [{ type: 'text', text: 'x' }] } }, push); - const incoming = responder(); - connection.emit('channels-incoming', { messages: [{ channelId: 'c', messageId: 'm', content: [] }] }, incoming); - await settle(); + it('a responder that throws while answering is logged, never rejected into the listener', async () => { + const { framework, connection } = await wiredFramework(); + await framework.stop(); + const push = responder({ throws: true }); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e11', payload: { content: [{ type: 'text', text: 'x' }] } }, push); + await answered(push); - assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'the handler failed' } }]); - assert.deepEqual(incoming.answers, [{ error: { code: -32603, message: 'the handler failed' } }]); - assert.ok(logged.some((line) => line.includes('push/event from srv failed: the handler failed')), logged.join('\n')); - assert.ok(!logged.some((line) => line.includes('refused')), 'a live host refuses nothing'); + assert.deepEqual(push.answers, [stopping]); + assert.ok(logged.some((line) => line.includes('push/event from srv refused: the host is stopping')), logged.join('\n')); assert.deepEqual(rejections, []); - fw.channelRegistry = registry; - await framework.stop(); }); }); }); From f281ba28dd667abbf5a892abcfa80cc5d746239d Mon Sep 17 00:00:00 2001 From: Nell-1783 Date: Sun, 11 Oct 2026 09:37:11 -0400 Subject: [PATCH 5/7] fix(lifecycle): every inspection of a thrown value is guarded, and the render-time inference refusal can't reject Carmen-2214's review of 9717820 (room-369 #70680) reproduced two exception boundaries on the compiled code (her probe, ~/.local/state/carmen-2214/af263-boundary-probe.mjs): - refuseServerItem classified the thrown value with `instanceof Error` outside any try, so a revoked Proxy threw there; and thrownText's try covered reading an Error's message but not its use, so an Error whose message is 42 reached `text.slice` and threw. Both left the request unanswered and the listener rejecting. thrownText now does every inspection and coercion inside its try (String() of the message or the value), and the new handlerFailureText guards the Error test itself, so refuseServerItem only ever handles strings. - inference-request's early -32600 refusal during a render answered outside any try, so a throwing respondError rejected the listener. It keeps its protocol-specific refusal, now with the same answer-attempt guard as the other exits. Tests: the thrown-values case adds a revoked Proxy ("the handler failed: (a value that could not be read)") and an Error whose message is 42 ("42"); a new case refuses an inference request during a render with -32600 through a throwing responder, with no rejection. Each new guard removed in the build fails one of them. Carmen's probe on this build: all three cases settle, none escapes. --- src/framework.ts | 38 ++++++++++++++++++++------- test/stop-with-work-in-flight.test.ts | 33 +++++++++++++++++++++-- 2 files changed, 59 insertions(+), 12 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index 1caee2ef3..aa09e0fa4 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -957,21 +957,39 @@ const THROWN_TEXT_LIMIT = 500; /** * What a thrown value says, safely and briefly: an Error's message, or any - * other value as a string. Reading it can't throw (a value with no string - * form, a hostile getter or a revoked Proxy reads as unreadable), and the - * text is bounded, since a provider or handler error can carry a whole + * other value, as a string. Every inspection and coercion of the value is + * inside the try, so reading it can't throw: a revoked Proxy (whose + * instanceof throws), a hostile getter, an Error whose message isn't a + * string, or a value with no string form reads as text or as unreadable. + * The text is bounded, since a provider or handler error can carry a whole * request body. */ function thrownText(error: unknown): string { let text: string; try { - text = error instanceof Error ? error.message : String(error); + text = String(error instanceof Error ? error.message : error); } catch { text = '(a value that could not be read)'; } return text.length <= THROWN_TEXT_LIMIT ? text : `${text.slice(0, THROWN_TEXT_LIMIT)}… (${text.length} characters)`; } +/** + * A failed handler's reason, never throwing: an Error's message, else "the + * handler failed" and what it threw (nothing, for `undefined`). Telling an + * Error apart is itself guarded, since instanceof throws on a revoked Proxy. + */ +function handlerFailureText(error: unknown): string { + let isError = false; + try { + isError = error instanceof Error; + } catch { + // Not an Error that can be read; thrownText says so. + } + if (isError) return thrownText(error); + return error === undefined ? 'the handler failed' : `the handler failed: ${thrownText(error)}`; +} + /** A server's responder, and whether the request it belongs to has been answered yet. */ type AnswerOnce = { responder: { respond?: (result: unknown) => void; respondError?: (code: number, message: string, data?: unknown) => void } | undefined; @@ -2427,11 +2445,7 @@ export class AgentFramework { ): void { // Optional chaining: prototype-built harnesses leave the queue unset. const stopping = outcome.kind === 'refused' || !!this.queue?.isClosed; - const what = outcome.kind === 'refused' - ? '' - : outcome.error instanceof Error - ? thrownText(outcome.error) - : `the handler failed${outcome.error === undefined ? '' : `: ${thrownText(outcome.error)}`}`; + const what = outcome.kind === 'refused' ? '' : handlerFailureText(outcome.error); const reason = outcome.kind === 'refused' ? 'the host is stopping' : stopping @@ -14425,7 +14439,11 @@ export class AgentFramework { ) => { // RFC-006 §5.3 / SPEC §10.7: no inference from inside a render. if (this.pushCoalescer?.isRendering(connection.id)) { - responder?.respondError(-32600, 'Cannot request inference while push/render is in progress'); + try { + responder?.respondError(-32600, 'Cannot request inference while push/render is in progress'); + } catch { + // The connection is going; there is no one left to tell. + } return; } if (this.inferenceRouter && responder) { diff --git a/test/stop-with-work-in-flight.test.ts b/test/stop-with-work-in-flight.test.ts index b4dcc3837..a3c469e34 100644 --- a/test/stop-with-work-in-flight.test.ts +++ b/test/stop-with-work-in-flight.test.ts @@ -464,7 +464,14 @@ describe('stop() with work still in flight', () => { const { framework, connection } = await wiredFramework(); try { const fw = framework as unknown as { pushHandler: unknown; channelRegistry: unknown }; - const thrown: Record = { e5: undefined, e7: 'a plain string', e8: Object.create(null) }; + // A revoked Proxy throws on instanceof itself; an Error whose message is a number has no slice. + const revoked = Proxy.revocable({}, {}); + revoked.revoke(); + const numbered = new Error('placeholder'); + Object.defineProperty(numbered, 'message', { value: 42 }); + const thrown: Record = { + e5: undefined, e7: 'a plain string', e8: Object.create(null), e12: revoked.proxy, e13: numbered, + }; fw.pushHandler = { // eslint-disable-next-line prefer-promise-reject-errors handlePushEvent: (_s: string, params: { eventId: string }) => Promise.reject(thrown[params.eventId]), @@ -479,13 +486,17 @@ describe('stop() with work still in flight', () => { const e5 = answersFor('e5'); const e7 = answersFor('e7'); const e8 = answersFor('e8'); + const e12 = answersFor('e12'); + const e13 = answersFor('e13'); const incoming = responder(); connection.emit('channels-incoming', { messages: [{ channelId: 'c', messageId: 'm', content: [] }] }, incoming); - await answered(e5, e7, e8, incoming); + await answered(e5, e7, e8, e12, e13, incoming); assert.deepEqual(e5.answers, [{ error: { code: -32603, message: 'the handler failed' } }]); assert.deepEqual(e7.answers, [{ error: { code: -32603, message: 'the handler failed: a plain string' } }]); assert.deepEqual(e8.answers, [{ error: { code: -32603, message: 'the handler failed: (a value that could not be read)' } }]); + assert.deepEqual(e12.answers, [{ error: { code: -32603, message: 'the handler failed: (a value that could not be read)' } }]); + assert.deepEqual(e13.answers, [{ error: { code: -32603, message: '42' } }]); assert.deepEqual(incoming.answers, [{ error: { code: -32603, message: 'the handler failed' } }]); assert.ok(!logged.some((line) => line.includes('refused')), 'a live host refuses nothing'); assert.deepEqual(rejections, []); @@ -533,6 +544,24 @@ describe('stop() with work still in flight', () => { } }); + it('an inference request during a render is refused -32600, and a throwing responder there is not rejected into the listener', async () => { + const { framework, connection } = await wiredFramework(); + try { + const fw = framework as unknown as { pushCoalescer: unknown }; + const coalescer = fw.pushCoalescer; + fw.pushCoalescer = { isRendering: () => true, suspend: () => {}, flush: () => {} }; + const inference = responder({ throws: true }); + connection.emit('inference-request', { messages: [] }, inference); + await answered(inference); + + assert.deepEqual(inference.answers, [{ error: { code: -32600, message: 'Cannot request inference while push/render is in progress' } }]); + assert.deepEqual(rejections, []); + fw.pushCoalescer = coalescer; + } finally { + await framework.stop(); + } + }); + it('a responder that throws while answering is logged, never rejected into the listener', async () => { const { framework, connection } = await wiredFramework(); await framework.stop(); From 47c5f778d987de6580b20e131100f56451c2c779 Mon Sep 17 00:00:00 2001 From: Nell-1783 Date: Sun, 11 Oct 2026 09:41:01 -0400 Subject: [PATCH 6/7] test(lifecycle): the toolName check reads the syntax tree; the test file's drop wording follows I30 Carmen-2214's review (room-369 #70710): the static check that every tool-result producer names its tool matched only the text `type: 'tool-result',`, so a producer spelled with double quotes and no toolName passed it (her probe, ~/.local/state/carmen-2214/af263-static-probe.py: typecheck clean, test green), and the >20 count let the miss through. The check now parses each src file with TypeScript's compiler API and visits every object literal whose `type` property is the string 'tool-result' in any quoting (through as/satisfies/parentheses), requiring a toolName property (assigned or shorthand). Her probe against it: the double-quoted producer without toolName now fails the test, naming framework.ts:12394, as the single-quoted one does. The test file's introduction still said the asking stream is gone at drop time; it now says what 9717820 says in pushEvent's doc: the queue that would carry the result to its stream is closed, stop() is ending the streams, and the tool has already run. --- test/stop-with-work-in-flight.test.ts | 44 +++++++++++++++------------ 1 file changed, 25 insertions(+), 19 deletions(-) diff --git a/test/stop-with-work-in-flight.test.ts b/test/stop-with-work-in-flight.test.ts index a3c469e34..e6680f234 100644 --- a/test/stop-with-work-in-flight.test.ts +++ b/test/stop-with-work-in-flight.test.ts @@ -7,7 +7,8 @@ * stream test, where a robot--move outlived the framework; the race itself * is older than #256): * - a tool's result that arrives after stop() is logged and dropped: the - * stream that asked is gone, and the tool has already run. Before, its + * queue that would carry it to its stream is closed, stop() is ending + * the streams, and the tool has already run. Before, its * push threw "Queue is closed", the throw was misreported as the tool's * failure, and that failure's push threw again, unhandled; * - a server's push, incoming message or other request the host can't @@ -25,6 +26,7 @@ import { EventEmitter } from 'node:events'; import { mkdtempSync, readFileSync, readdirSync, rmSync, statSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import ts from 'typescript'; import type { ContentBlock } from '@animalabs/membrane'; import type { Module, @@ -289,7 +291,9 @@ describe('stop() with work still in flight', () => { it('every tool result the framework produces names its tool', () => { // The drop's log line names the tool from the event's toolName, so a // producer that leaves it out logs a drop nobody can reconcile. Every - // object literal in src with type: 'tool-result' carries toolName. + // object literal in src whose `type` is the string 'tool-result' (in any + // quoting) carries toolName. Read as TypeScript's syntax tree, not as + // text, so no spelling of the literal escapes the check. const src = new URL('../../src/', import.meta.url).pathname; const files: string[] = []; const walk = (dir: string): void => { @@ -300,27 +304,29 @@ describe('stop() with work still in flight', () => { } }; walk(src); + const nameOf = (n: ts.PropertyName): string | undefined => + ts.isIdentifier(n) || ts.isStringLiteral(n) || ts.isNoSubstitutionTemplateLiteral(n) ? n.text : undefined; + const isToolResult = (e: ts.Expression): boolean => { + while (ts.isAsExpression(e) || ts.isSatisfiesExpression(e) || ts.isParenthesizedExpression(e)) e = e.expression; + return (ts.isStringLiteral(e) || ts.isNoSubstitutionTemplateLiteral(e)) && e.text === 'tool-result'; + }; const missing: string[] = []; let literals = 0; for (const file of files) { - const text = readFileSync(file, 'utf8'); - for (const m of text.matchAll(/type:\s*'tool-result',/g)) { - literals++; - let start = m.index!; - for (let depth = 0; start > 0; start--) { - const c = text[start - 1]; - if (c === '}') depth++; - else if (c === '{') { if (depth === 0) { start--; break; } depth--; } - } - let end = start; - for (let depth = 0; end < text.length; end++) { - if (text[end] === '{') depth++; - else if (text[end] === '}' && --depth === 0) break; - } - if (!/\btoolName\b/.test(text.slice(start, end + 1))) { - missing.push(`${file.slice(src.length)}:${text.slice(0, m.index).split('\n').length}`); + const sf = ts.createSourceFile(file, readFileSync(file, 'utf8'), ts.ScriptTarget.Latest, true); + const visit = (node: ts.Node): void => { + if (ts.isObjectLiteralExpression(node)) { + const type = node.properties.find((p) => ts.isPropertyAssignment(p) && nameOf(p.name) === 'type'); + if (type && ts.isPropertyAssignment(type) && isToolResult(type.initializer)) { + literals++; + const named = node.properties.some((p) => + (ts.isPropertyAssignment(p) || ts.isShorthandPropertyAssignment(p)) && nameOf(p.name) === 'toolName'); + if (!named) missing.push(`${file.slice(src.length)}:${sf.getLineAndCharacterOfPosition(node.getStart()).line + 1}`); + } } - } + ts.forEachChild(node, visit); + }; + visit(sf); } assert.ok(literals > 20, `found ${literals} tool-result literals; the scan has lost them`); assert.deepEqual(missing, [], 'tool-result events without toolName'); From bb99420a720a0a708f160d2eec55e24b3f783b2d Mon Sep 17 00:00:00 2001 From: Nell-1783 Date: Mon, 12 Oct 2026 23:55:03 -0400 Subject: [PATCH 7/7] fix(lifecycle): answer each server request at most once, and refuse host/command at stop() too answerOnce marked a request answered before the responder ran, so a respond that threw, such as a result the connection can't serialize, counted as an answer: the listener's failure path then only logged it, and the server waited for its timeout. It also forwarded every later call, so a handler that answered after the listener had answered its failure sent a second response for the same id. And it gave every responder a respondError, so a handler that checks for one never took its fallback. - answerOnce counts an answer once the responder returns, refuses and logs a second one, and wraps only the methods the responder has. - host/command gets the boundary that push/event, channels/incoming, channels/register, channels/changed and inference/request already have: refused with -32000 "the host is stopping" once the queue is closed (after the awareness barrier), answered at most once, and a failure answered through refuseServerItem, so reading what it threw can't throw. Tests (stop-with-work-in-flight, now 25): - a respond that throws on a BigInt result: the request gets -32603; - a handler that answers after its failure was answered: one response, and the second is logged; - a responder without respondError reaches the handler without one; - host/command after stop(), and held at the awareness barrier: refused and never run; an inference/request held at the barrier, likewise; - each of those six listeners answers its handler's throw with -32603 and logs it; - a late tool result's 5,000-character error is bounded in the drop line; - an Error whose message getter throws is answered as unreadable; - the syntax-tree check also pairs each producer's toolName with the call its callId reads, and finds src with fileURLToPath. Changelog: host/command joins the list, and a request is answered at most once. --- changelog.d/stop-with-work-in-flight.fixed.md | 19 +- src/framework.ts | 60 +++-- test/stop-with-work-in-flight.test.ts | 205 +++++++++++++++--- 3 files changed, 232 insertions(+), 52 deletions(-) diff --git a/changelog.d/stop-with-work-in-flight.fixed.md b/changelog.d/stop-with-work-in-flight.fixed.md index f662952fb..46ba4387e 100644 --- a/changelog.d/stop-with-work-in-flight.fixed.md +++ b/changelog.d/stop-with-work-in-flight.fixed.md @@ -17,19 +17,22 @@ the tool again). For this, the framework's tool results carry the tool's name (`ToolResultEvent.toolName`, optional). - A server's `push/event`, `channels/incoming`, `channels/register`, - `channels/changed` or `inference/request` that arrives once the host is - stopping is refused before its handler runs, with a JSON-RPC error - (`-32000`, "the host is stopping", the code the push coalescer already - gives a stopping host). It is never answered "accepted", and no - inference starts. + `channels/changed`, `inference/request` or `host/command` that arrives + once the host is stopping is refused before its handler runs, with a + JSON-RPC error (`-32000`, "the host is stopping", the code the push + coalescer already gives a stopping host). It is never answered + "accepted", no inference starts, and no host command runs. - A handler that throws is answered with an error too: while the host stops, `-32000` with the handler's own message after "the host stopped while handling this request"; on a live host, `-32603` with its message. - The message is bounded, any thrown value is printed safely, and a - request the handler already answered isn't answered twice. Part of the - request may already have been admitted. A notification is logged. + The message is bounded, and any thrown value is printed safely. Part of + the request may already have been admitted. A notification is logged. - **For operators:** such a handler fault used to be an unhandled rejection, which ends a Node process by default, so a supervisor would have restarted the host. Now the host stays up and logs the failure. + - A request is answered at most once. A second answer, such as a + handler's late one after its failure was answered, is refused and + logged. An answer that throws, such as a result the connection can't + serialize, doesn't count, so the request still gets the error answer. - Any other event pushed to a stopped framework still throws to its caller. diff --git a/src/framework.ts b/src/framework.ts index aa09e0fa4..73566252e 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -997,21 +997,41 @@ type AnswerOnce = { }; /** - * Wrap a server request's responder so a later failure can tell whether the - * handler already answered: a request is answered once, and a handler that - * answered and then threw is logged, not answered a second time. + * Wrap a server request's responder so the request is answered at most once, + * and a later failure can tell whether the handler already answered. + * - An answer counts once the responder returns. One that throws, such as a + * result the transport can't serialize, leaves the request unanswered, so + * the listener's failure answer still goes out. This assumes a responder + * that throws has sent nothing, which holds for the connection's own: they + * serialize before they write, and their transports' writes don't throw + * after writing. + * - A second answer is refused and logged: a handler that answered and then + * threw isn't answered again, and a handler that answers after the + * listener answered its failure doesn't send a second response. + * - Only the methods the responder has are wrapped, so a handler that checks + * for `respondError` still finds a responder without one. */ function answerOnce void; respondError?: (code: number, message: string, data?: unknown) => void }>( responder: R | undefined, + method: string, + serverId: string, ): AnswerOnce & { responder: R | undefined } { if (!responder) return { responder: undefined, answered: () => false }; let done = false; - const wrapped = { - ...responder, - respond: (result: unknown) => { done = true; responder.respond?.(result); }, - respondError: (code: number, message: string, data?: unknown) => { done = true; responder.respondError?.(code, message, data); }, - } as R; - return { responder: wrapped, answered: () => done }; + const once = (kind: string, send: (...args: A) => void) => (...args: A): void => { + if (done) { + console.error(`[mcpl] ${method} from ${serverId}: a second answer (${kind}) was refused; the request was already answered`); + return; + } + send(...args); + done = true; + }; + const wrapped: { respond?: (result: unknown) => void; respondError?: (code: number, message: string, data?: unknown) => void } = { ...responder }; + if (responder.respond) wrapped.respond = once('respond', (result: unknown) => responder.respond!(result)); + if (responder.respondError) { + wrapped.respondError = once('respondError', (code: number, message: string, data?: unknown) => responder.respondError!(code, message, data)); + } + return { responder: wrapped as R, answered: () => done }; } export class AgentFramework { @@ -14416,7 +14436,7 @@ export class AgentFramework { params: PushEventParams, responder?: { respond: (result: unknown) => void; respondError: (code: number, message: string) => void }, ) => { - const answer = answerOnce(responder); + const answer = answerOnce(responder, 'push/event', connection.id); try { const barrier = this.discordAwarenessBarrier; if (barrier) await barrier.promise; @@ -14447,7 +14467,7 @@ export class AgentFramework { return; } if (this.inferenceRouter && responder) { - const answer = answerOnce(responder); + const answer = answerOnce(responder, 'inference/request', connection.id); try { const barrier = this.discordAwarenessBarrier; if (barrier) await barrier.promise; @@ -14473,7 +14493,7 @@ export class AgentFramework { params: ChannelsRegisterParams, responder?: { respond: (result: unknown) => void; respondError?: (code: number, message: string) => void }, ) => { - const answer = answerOnce(responder); + const answer = answerOnce(responder, 'channels/register', connection.id); try { if (this.queue?.isClosed) { this.refuseServerItem('channels/register', connection.id, answer, { kind: 'refused' }); @@ -14493,7 +14513,7 @@ export class AgentFramework { params: ChannelsChangedParams, responder?: { respond: (result: unknown) => void; respondError?: (code: number, message: string) => void }, ) => { - const answer = answerOnce(responder); + const answer = answerOnce(responder, 'channels/changed', connection.id); try { if (this.queue?.isClosed) { this.refuseServerItem('channels/changed', connection.id, answer, { kind: 'refused' }); @@ -14563,7 +14583,7 @@ export class AgentFramework { params: ChannelsIncomingParams, responder?: { respond: (result: unknown) => void; respondError?: (code: number, message: string) => void }, ) => { - const answer = answerOnce(responder); + const answer = answerOnce(responder, 'channels/incoming', connection.id); try { const barrier = this.discordAwarenessBarrier; if (barrier) await barrier.promise; @@ -14583,14 +14603,20 @@ export class AgentFramework { responder?: { respond: (result: unknown) => void; respondError: (code: number, message: string) => void }, ) => { if (!responder) return; + const answer = answerOnce(responder, 'host/command', connection.id); try { const barrier = this.discordAwarenessBarrier; if (barrier) await barrier.promise; + // A stopping host runs no host command: refused like the requests + // above, before its handler runs. + if (this.queue?.isClosed) { + this.refuseServerItem('host/command', connection.id, answer, { kind: 'refused' }); + return; + } const result = await this.handleHostCommand(connection.id, params ?? {}); - responder.respond(result); + answer.responder!.respond(result); } catch (error) { - const err = error instanceof Error ? error : new Error(String(error)); - responder.respondError(-32603, err.message); + this.refuseServerItem('host/command', connection.id, answer, { kind: 'failed', error }); } }); diff --git a/test/stop-with-work-in-flight.test.ts b/test/stop-with-work-in-flight.test.ts index e6680f234..401842805 100644 --- a/test/stop-with-work-in-flight.test.ts +++ b/test/stop-with-work-in-flight.test.ts @@ -26,6 +26,7 @@ import { EventEmitter } from 'node:events'; import { mkdtempSync, readFileSync, readdirSync, rmSync, statSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; import ts from 'typescript'; import type { ContentBlock } from '@animalabs/membrane'; import type { @@ -85,11 +86,13 @@ class SlowModule implements Module { * - failure: the module throws (the registry answers an error result) * - isError: { success: true, isError: true }, a result that reports an error * - bare-failure: { success: false } with no error text + * - long-failure: the module throws a 5,000-character error */ - finish(outcome: 'success' | 'failure' | 'isError' | 'bare-failure'): void { + finish(outcome: 'success' | 'failure' | 'isError' | 'bare-failure' | 'long-failure'): void { if (outcome === 'success') this.settle!.resolve({ success: true, data: { done: true } }); else if (outcome === 'isError') this.settle!.resolve({ success: true, isError: true, data: { done: false } }); else if (outcome === 'bare-failure') this.settle!.resolve({ success: false }); + else if (outcome === 'long-failure') this.settle!.reject(new Error('x'.repeat(5000))); else this.settle!.reject(new Error('the work failed')); } @@ -193,6 +196,17 @@ describe('stop() with work still in flight', () => { assert.match(line, /the tool failed: /, line); }); + it("a tool's long error is bounded in the drop's log line", async () => { + const { framework, module } = await frameworkWithRunningTool(); + await framework.stop(); + module.finish('long-failure'); + const line = await droppedLine(); + + assert.ok(line.length < 800, `logged ${line.length} characters`); + assert.match(line, /the tool failed: x+… \(5000 characters\)$/, line); + assert.deepEqual(rejections, []); + }); + it('a result that reports an error is dropped as failed, and a failure with no text says only that', async () => { // { success: true, isError: true } is a failure: the drop must not call it a success. { @@ -288,13 +302,15 @@ describe('stop() with work still in flight', () => { }); }); - it('every tool result the framework produces names its tool', () => { + it('every tool result the framework produces names its tool, and the call it answers', () => { // The drop's log line names the tool from the event's toolName, so a - // producer that leaves it out logs a drop nobody can reconcile. Every + // producer that leaves it out logs a drop nobody can reconcile, and one + // that names another call's tool misleads whoever reconciles it. Every // object literal in src whose `type` is the string 'tool-result' (in any - // quoting) carries toolName. Read as TypeScript's syntax tree, not as - // text, so no spelling of the literal escapes the check. - const src = new URL('../../src/', import.meta.url).pathname; + // quoting) carries toolName, and where its callId and toolName both read + // a call (`X.id`, `Y.name`), they read the same one. Read as TypeScript's + // syntax tree, not as text, so no spelling of the literal escapes the check. + const src = fileURLToPath(new URL('../../src/', import.meta.url)); const files: string[] = []; const walk = (dir: string): void => { for (const name of readdirSync(dir)) { @@ -310,8 +326,16 @@ describe('stop() with work still in flight', () => { while (ts.isAsExpression(e) || ts.isSatisfiesExpression(e) || ts.isParenthesizedExpression(e)) e = e.expression; return (ts.isStringLiteral(e) || ts.isNoSubstitutionTemplateLiteral(e)) && e.text === 'tool-result'; }; + // The call an `X.id` or `X.name` reads, as source text (`call`, `tc`). + const callRead = (p: ts.ObjectLiteralElementLike | undefined, field: 'id' | 'name'): string | undefined => { + if (!p || !ts.isPropertyAssignment(p)) return undefined; + const e = p.initializer; + return ts.isPropertyAccessExpression(e) && e.name.text === field ? e.expression.getText() : undefined; + }; const missing: string[] = []; + const mismatched: string[] = []; let literals = 0; + let paired = 0; for (const file of files) { const sf = ts.createSourceFile(file, readFileSync(file, 'utf8'), ts.ScriptTarget.Latest, true); const visit = (node: ts.Node): void => { @@ -319,9 +343,17 @@ describe('stop() with work still in flight', () => { const type = node.properties.find((p) => ts.isPropertyAssignment(p) && nameOf(p.name) === 'type'); if (type && ts.isPropertyAssignment(type) && isToolResult(type.initializer)) { literals++; - const named = node.properties.some((p) => - (ts.isPropertyAssignment(p) || ts.isShorthandPropertyAssignment(p)) && nameOf(p.name) === 'toolName'); - if (!named) missing.push(`${file.slice(src.length)}:${sf.getLineAndCharacterOfPosition(node.getStart()).line + 1}`); + const where = `${file.slice(src.length)}:${sf.getLineAndCharacterOfPosition(node.getStart()).line + 1}`; + const property = (name: string) => node.properties.find((p) => + (ts.isPropertyAssignment(p) || ts.isShorthandPropertyAssignment(p)) && nameOf(p.name) === name); + const toolName = property('toolName'); + if (!toolName) missing.push(where); + const idOf = callRead(property('callId'), 'id'); + const nameFrom = callRead(toolName, 'name'); + if (idOf !== undefined && nameFrom !== undefined) { + paired++; + if (idOf !== nameFrom) mismatched.push(`${where}: callId reads ${idOf}, toolName reads ${nameFrom}`); + } } } ts.forEachChild(node, visit); @@ -330,6 +362,8 @@ describe('stop() with work still in flight', () => { } assert.ok(literals > 20, `found ${literals} tool-result literals; the scan has lost them`); assert.deepEqual(missing, [], 'tool-result events without toolName'); + assert.ok(paired > 20, `paired ${paired} callId/toolName reads; the pairing has lost them`); + assert.deepEqual(mismatched, [], "tool-result events naming another call's tool"); }); describe('a server sending while the host stops', () => { @@ -345,12 +379,46 @@ describe('stop() with work still in flight', () => { return { framework, connection }; } - function responder(opts: { throws?: boolean } = {}) { + /** + * Put a stand-in in for every request listener's handler, each calling + * `handler` with what it was given: the push handler, the channel + * registry's three, the inference router and the host-command handler. + */ + function standIns(framework: AgentFramework, handler: (...args: never[]) => Promise): void { + const fw = framework as unknown as { + pushHandler: unknown; channelRegistry: unknown; inferenceRouter: unknown; handleHostCommand: unknown; + }; + fw.pushHandler = { handlePushEvent: handler }; + fw.channelRegistry = { handleIncoming: handler, handleRegister: handler, handleChanged: handler, stopAll: () => {} }; + fw.inferenceRouter = { handleInferenceRequest: handler }; + fw.handleHostCommand = handler; + } + + /** Every request listener: its event, its method, and parameters it takes. */ + const requests = [ + ['push-event', 'push/event', { featureSet: 'chat', eventId: 'e20', payload: { content: [{ type: 'text', text: 'x' }] } }], + ['channels-incoming', 'channels/incoming', { messages: [{ channelId: 'c', messageId: 'm20', content: [] }] }], + ['channels-register', 'channels/register', { channels: [] }], + ['channels-changed', 'channels/changed', { added: [] }], + ['inference-request', 'inference/request', { messages: [] }], + ['host-command', 'host/command', { command: 'undo' }], + ] as const; + + /** + * A server's responder, recording each answer. + * - throws: respondError records its answer, then throws (the connection is going). + * - serializes: respond serializes the result before it records it, as the + * connection's sendResponse does, so a result JSON can't hold throws. + */ + function responder(opts: { throws?: boolean; serializes?: boolean } = {}) { const answers: Array<{ result?: unknown; error?: { code: number; message: string } }> = []; return { id: 7, answers, - respond: (result: unknown) => { answers.push({ result }); }, + respond: (result: unknown) => { + if (opts.serializes) JSON.stringify(result); + answers.push({ result }); + }, respondError: (code: number, message: string) => { answers.push({ error: { code, message } }); if (opts.throws) throw new Error('the connection is gone'); @@ -371,14 +439,7 @@ describe('stop() with work still in flight', () => { await framework.stop(); // Stand-in handlers, put in after stop() (which stops the real registry): every arrival below comes later. let handled = 0; - const fw = framework as unknown as { pushHandler: unknown; channelRegistry: unknown; inferenceRouter: unknown }; - fw.pushHandler = { handlePushEvent: async () => { handled++; } }; - fw.channelRegistry = { - handleIncoming: async () => { handled++; }, - handleRegister: async () => { handled++; }, - handleChanged: async () => { handled++; }, - }; - fw.inferenceRouter = { handleInferenceRequest: async () => { handled++; } }; + standIns(framework, async () => { handled++; return {}; }); const push = responder(); connection.emit('push-event', { featureSet: 'chat', eventId: 'e1', payload: { content: [{ type: 'text', text: 'late' }] } }, push); @@ -390,16 +451,20 @@ describe('stop() with work still in flight', () => { connection.emit('channels-changed', { added: [] }, changed); const inference = responder(); connection.emit('inference-request', { messages: [] }, inference); - await answered(push, incoming, register, changed, inference); + const command = responder(); + connection.emit('host-command', { command: 'undo' }, command); + await answered(push, incoming, register, changed, inference, command); - assert.equal(handled, 0, 'never handed to a handler, which would record it as accepted'); - for (const r of [push, incoming, register, changed, inference]) assert.deepEqual(r.answers, [stopping]); + assert.equal(handled, 0, 'never handed to a handler, which would record it as accepted or act on it'); + for (const r of [push, incoming, register, changed, inference, command]) assert.deepEqual(r.answers, [stopping]); assert.deepEqual(rejections, []); }); for (const [event, method, params] of [ ['push-event', 'push/event', { featureSet: 'chat', eventId: 'e2', payload: { content: [{ type: 'text', text: 'waiting' }] } }], ['channels-incoming', 'channels/incoming', { messages: [{ channelId: 'c', messageId: 'm2', content: [] }] }], + ['inference-request', 'inference/request', { messages: [] }], + ['host-command', 'host/command', { command: 'undo' }], ] as const) { it(`a ${method} held at the awareness barrier while the host stops is refused once the barrier lifts`, async () => { const { framework, connection } = await wiredFramework(); @@ -408,9 +473,7 @@ describe('stop() with work still in flight', () => { promise: new Promise((r) => { lift = r; }), }; let handled = 0; - const fw = framework as unknown as { pushHandler: unknown; channelRegistry: unknown }; - fw.pushHandler = { handlePushEvent: async () => { handled++; } }; - fw.channelRegistry = { handleIncoming: async () => { handled++; }, stopAll: () => {} }; + standIns(framework, async () => { handled++; return {}; }); const r = responder(); connection.emit(event, params, r); @@ -475,8 +538,11 @@ describe('stop() with work still in flight', () => { revoked.revoke(); const numbered = new Error('placeholder'); Object.defineProperty(numbered, 'message', { value: 42 }); + // An Error whose message getter throws: reading it is itself guarded. + const unreadable = new Error('placeholder'); + Object.defineProperty(unreadable, 'message', { get() { throw new Error('the message getter broke'); } }); const thrown: Record = { - e5: undefined, e7: 'a plain string', e8: Object.create(null), e12: revoked.proxy, e13: numbered, + e5: undefined, e7: 'a plain string', e8: Object.create(null), e12: revoked.proxy, e13: numbered, e14: unreadable, }; fw.pushHandler = { // eslint-disable-next-line prefer-promise-reject-errors @@ -494,15 +560,17 @@ describe('stop() with work still in flight', () => { const e8 = answersFor('e8'); const e12 = answersFor('e12'); const e13 = answersFor('e13'); + const e14 = answersFor('e14'); const incoming = responder(); connection.emit('channels-incoming', { messages: [{ channelId: 'c', messageId: 'm', content: [] }] }, incoming); - await answered(e5, e7, e8, e12, e13, incoming); + await answered(e5, e7, e8, e12, e13, e14, incoming); assert.deepEqual(e5.answers, [{ error: { code: -32603, message: 'the handler failed' } }]); assert.deepEqual(e7.answers, [{ error: { code: -32603, message: 'the handler failed: a plain string' } }]); assert.deepEqual(e8.answers, [{ error: { code: -32603, message: 'the handler failed: (a value that could not be read)' } }]); assert.deepEqual(e12.answers, [{ error: { code: -32603, message: 'the handler failed: (a value that could not be read)' } }]); assert.deepEqual(e13.answers, [{ error: { code: -32603, message: '42' } }]); + assert.deepEqual(e14.answers, [{ error: { code: -32603, message: '(a value that could not be read)' } }]); assert.deepEqual(incoming.answers, [{ error: { code: -32603, message: 'the handler failed' } }]); assert.ok(!logged.some((line) => line.includes('refused')), 'a live host refuses nothing'); assert.deepEqual(rejections, []); @@ -550,6 +618,89 @@ describe('stop() with work still in flight', () => { } }); + it("a handler's answer that throws, such as a result JSON can't hold, isn't counted, so the request still gets the listener's answer", async () => { + const { framework, connection } = await wiredFramework(); + try { + standIns(framework, async (_s: string, _p: unknown, r: { respond(x: unknown): void }) => { + r.respond({ accepted: true, sequence: 1n }); + }); + const push = responder({ serializes: true }); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e21', payload: { content: [{ type: 'text', text: 'x' }] } }, push); + await answered(push); + + assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'Do not know how to serialize a BigInt' } }]); + assert.ok(!logged.some((line) => line.includes('failed after it was answered')), logged.join('\n')); + assert.deepEqual(rejections, []); + } finally { + await framework.stop(); + } + }); + + it('a handler that answers after its failure was answered is refused, so the server gets one response', async () => { + const { framework, connection } = await wiredFramework(); + try { + standIns(framework, async (_s: string, _p: unknown, r: { respond(x: unknown): void }) => { + setImmediate(() => r.respond({ accepted: true })); + throw new Error('the handler broke first'); + }); + const push = responder(); + connection.emit('push-event', { featureSet: 'chat', eventId: 'e22', payload: { content: [{ type: 'text', text: 'x' }] } }, push); + await answered(push); + await until(() => logged.some((line) => line.includes('a second answer (respond) was refused')), 'the refused second answer'); + await turn(); + + assert.deepEqual(push.answers, [{ error: { code: -32603, message: 'the handler broke first' } }]); + assert.ok(logged.some((line) => line.includes('push/event from srv: a second answer (respond) was refused; the request was already answered')), logged.join('\n')); + assert.deepEqual(rejections, []); + } finally { + await framework.stop(); + } + }); + + it('a responder without respondError reaches the handler without one', async () => { + const { framework, connection } = await wiredFramework(); + try { + let seen: string | undefined; + standIns(framework, async (_s: string, _p: unknown, r: { respond(x: unknown): void; respondError?: unknown }) => { + seen = typeof r.respondError; + r.respond({ results: [] }); + }); + const answers: unknown[] = []; + connection.emit('channels-incoming', { messages: [{ channelId: 'c', messageId: 'm23', content: [] }] }, { + id: 7, respond: (result: unknown) => { answers.push(result); }, + }); + await until(() => answers.length > 0, 'the answer'); + await turn(); + + assert.equal(seen, 'undefined', "a handler checking for respondError takes its fallback"); + assert.deepEqual(answers, [{ results: [] }]); + assert.deepEqual(rejections, []); + } finally { + await framework.stop(); + } + }); + + it("every request listener answers its handler's throw -32603 with its error, never rejecting into the listener", async () => { + const { framework, connection } = await wiredFramework(); + try { + standIns(framework, async () => { throw new Error('the handler broke'); }); + const answers = requests.map(([event, , params]) => { + const r = responder(); + connection.emit(event, params, r); + return r; + }); + await answered(...answers); + + for (const [i, [, method]] of requests.entries()) { + assert.deepEqual(answers[i]!.answers, [{ error: { code: -32603, message: 'the handler broke' } }], method); + assert.ok(logged.some((line) => line.includes(`${method} from srv failed: the handler broke`)), `${method}:\n${logged.join('\n')}`); + } + assert.deepEqual(rejections, []); + } finally { + await framework.stop(); + } + }); + it('an inference request during a render is refused -32600, and a throwing responder there is not rejected into the listener', async () => { const { framework, connection } = await wiredFramework(); try {