diff --git a/changelog.d/late-module-events.fixed.md b/changelog.d/late-module-events.fixed.md new file mode 100644 index 000000000..f807814fd --- /dev/null +++ b/changelog.d/late-module-events.fixed.md @@ -0,0 +1 @@ +- A module's event that arrives after `stop()` began is dropped with a log line naming the module, through `ModuleContext.pushEvent()` or the deprecated `queue` handle alike, instead of throwing `Queue is closed`. `stop()` closes the event queue before it stops modules, so such an event comes from a listener that fires while `stop()` runs, where the throw can go unhandled, or from the module's own `stop()`, where it made the framework's `stop()` fail. A direct caller of `AgentFramework.pushEvent()` still gets the throw for anything but a tool result, and a module's tool result keeps that method's drop line, which names the call, the tool and its outcome. The deprecated `ModuleContext.queue` is now an object with `push` alone, as its `ProcessQueue` type always said; it was the framework's own queue object. diff --git a/src/framework.ts b/src/framework.ts index dbb2e4033..05d878513 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1630,14 +1630,16 @@ export class AgentFramework { }); // Initialize module registry with callbacks - this.moduleRegistry = new ModuleRegistry(store, this.queue, { + // A module's events, through pushEvent() or its deprecated queue handle, + // go through pushModuleEvent(), which names the module. + this.moduleRegistry = new ModuleRegistry(store, { push: (event, moduleName) => this.pushModuleEvent(event, moduleName, false) }, { getAgents: () => Array.from(this.agents.values()), addMessage: (p, c, m) => this.addMessage(p, c, m), editMessage: (id, c) => this.editMessage(id, c), removeMessage: (id) => this.removeMessage(id), getMessage: (id) => this.getMessage(id), queryMessages: (filter) => this.queryMessages(filter), - pushEvent: (event) => this.pushEvent(event), + pushEvent: (event, moduleName) => this.pushModuleEvent(event, moduleName, true), onTrace: (listener) => this.onTrace(listener), // The ONLY entry that confers module origin — a private closure the // framework hands to its own module registry. The public @@ -2667,6 +2669,29 @@ export class AgentFramework { this.emitTrace({ type: 'process:received', processEvent: event }); } + /** + * A module's event. stop() closes the queue before it stops modules, so a + * module's event can arrive after that: from a listener that fires while + * stop() runs, or from the module's own stop(). Nothing will process it, + * and a throw there can go unhandled or fail stop(), so it's dropped with a + * log line naming the module. Otherwise it's queued as pushEvent() queues + * it, or, through the deprecated queue handle, without the trace. + */ + private pushModuleEvent(event: ProcessEvent, moduleName: string | undefined, traced: boolean): void { + if (this.queue.isClosed) { + // A tool result takes pushEvent()'s own drop, whose line names the + // call, the tool and its outcome. + if (event.type === 'tool-result') { + this.pushEvent(event); + return; + } + console.error(`[framework] a '${event.type}' event from module ${moduleName ?? '(unnamed)'} arrived after stop() began — dropped`); + return; + } + if (traced) this.pushEvent(event); + else this.queue.push(event); + } + /** * Add a trace event listener for observability. */ diff --git a/src/module-registry.ts b/src/module-registry.ts index 061107ed0..42060860f 100644 --- a/src/module-registry.ts +++ b/src/module-registry.ts @@ -26,6 +26,12 @@ import type { import type { Agent } from './agent.js'; import { DEFAULT_SHUTDOWN_TIMEOUT_MS, validateShutdownTimeout, waitForShutdown } from './shutdown.js'; +/** How the registry queues a module's event: its push names the module. A + * `ProcessQueue` serves too, ignoring the name. */ +export interface ModuleQueue { + push(event: ProcessEvent, moduleName?: string): void; +} + const MODULE_STATE_PREFIX = 'modules/'; /** @@ -78,7 +84,7 @@ export class ModuleRegistry { private moduleContexts: Map = new Map(); private speechHandlers: SpeechHandler[] = []; private store: JsStore; - private queue: ProcessQueue; + private queue: ModuleQueue; private getAgents: () => Agent[]; private addMessageFn: (participant: string, content: ContentBlock[], metadata?: MessageMetadata) => MessageId; private editMessageFn: (id: MessageId, content: ContentBlock[]) => void; @@ -86,7 +92,7 @@ export class ModuleRegistry { private getMessageFn: (id: MessageId) => StoredMessage | null; private queryMessagesFn: (filter: MessageQuery) => MessageQueryResult; - private pushEventFn: (event: ProcessEvent) => void; + private pushEventFn: (event: ProcessEvent, moduleName?: string) => void; private onTraceFn: (listener: TraceEventListener) => () => void; private callToolFn: (call: ToolCall) => Promise; /** Framework ops-alert channel, surfaced to modules via @@ -101,7 +107,7 @@ export class ModuleRegistry { constructor( store: JsStore, - queue: ProcessQueue, + queue: ModuleQueue, options: { getAgents: () => Agent[]; addMessage: (participant: string, content: ContentBlock[], metadata?: MessageMetadata) => MessageId; @@ -109,7 +115,8 @@ export class ModuleRegistry { removeMessage: (id: MessageId) => void; getMessage: (id: MessageId) => StoredMessage | null; queryMessages: (filter: MessageQuery) => MessageQueryResult; - pushEvent: (event: ProcessEvent) => void; + /** Called with the pushing module's name. */ + pushEvent: (event: ProcessEvent, moduleName?: string) => void; onTrace: (listener: TraceEventListener) => () => void; callTool: (call: ToolCall) => Promise; notifyOps?: (kind: string, agentName: string, message: string, data?: Record) => void; @@ -528,7 +535,7 @@ class ModuleContextImpl implements ModuleContext { private removeMessageFn: (id: MessageId) => void; private getMessageFn: (id: MessageId) => StoredMessage | null; private queryMessagesFn: (filter: MessageQuery) => MessageQueryResult; - private pushEventFn: (event: ProcessEvent) => void; + private pushEventFn: (event: ProcessEvent, moduleName?: string) => void; private onTraceFn: (listener: TraceEventListener) => () => void; private callToolFn: (call: ToolCall) => Promise; @@ -539,7 +546,7 @@ class ModuleContextImpl implements ModuleContext { moduleName: string, store: JsStore, stateId: string, - queue: ProcessQueue, + queue: ModuleQueue, registry: ModuleRegistry, isRestart: boolean, getAgents: () => Agent[], @@ -548,14 +555,15 @@ class ModuleContextImpl implements ModuleContext { removeMessage: (id: MessageId) => void, getMessage: (id: MessageId) => StoredMessage | null, queryMessages: (filter: MessageQuery) => MessageQueryResult, - pushEvent: (event: ProcessEvent) => void, + pushEvent: (event: ProcessEvent, moduleName?: string) => void, onTrace: (listener: TraceEventListener) => () => void, callTool: (call: ToolCall) => Promise ) { this.moduleName = moduleName; this.store = store; this.stateId = stateId; - this.queue = queue; + // The deprecated handle: a push names this module, as pushEvent() does. + this.queue = { push: (event) => queue.push(event, moduleName) }; this.registry = registry; this.isRestart = isRestart; this.getAgentsFn = getAgents; @@ -738,7 +746,7 @@ class ModuleContextImpl implements ModuleContext { } pushEvent(event: ProcessEvent): void { - this.pushEventFn(event); + this.pushEventFn(event, this.moduleName); } private persistExternalIdMap(): void { diff --git a/src/types/module.ts b/src/types/module.ts index 0fbec386a..afb2638da 100644 --- a/src/types/module.ts +++ b/src/types/module.ts @@ -217,14 +217,19 @@ export interface ModuleContext { getLogLength(name: string): number; /** - * Process queue for pushing events from external listeners. + * Process queue for pushing events from external listeners. Once stop() + * has begun, a push is dropped with a log line naming this module, as + * pushEvent()'s is. * @deprecated Use pushEvent() instead — it also emits process:received traces. */ readonly queue: ProcessQueue; /** * Push a process event to the queue and emit a process:received trace. - * Preferred over direct queue.push() for proper observability. + * Preferred over direct queue.push() for proper observability. Once stop() + * has begun, nothing will process the event: it's dropped with a log line + * naming this module, untraced, rather than thrown into the listener or + * stop() that pushed it. */ pushEvent(event: ProcessEvent): void; @@ -370,14 +375,17 @@ export interface ProcessState { getActiveTools(): ToolDefinition[]; /** - * Queue for emitting follow-up events. + * Queue for emitting follow-up events. Once stop() has begun, a push is + * dropped with a log line naming the module, as pushEvent()'s is. * @deprecated Use pushEvent() instead — it also emits process:received traces. */ readonly queue: ProcessQueue; /** * Push a process event to the queue and emit a process:received trace. - * Preferred over direct queue.push() for proper observability. + * Preferred over direct queue.push() for proper observability. Once stop() + * has begun, the event is dropped with a log line naming the module, + * untraced. */ pushEvent(event: ProcessEvent): void; } diff --git a/test/late-events-after-stop.test.ts b/test/late-events-after-stop.test.ts new file mode 100644 index 000000000..548b499a9 --- /dev/null +++ b/test/late-events-after-stop.test.ts @@ -0,0 +1,134 @@ +/** + * stop() closes the event queue before it stops modules, so a module's event + * can arrive after the queue closed: from its own stop(), or from a listener + * that fires while stop() runs. Nothing will process it, and the module's + * callback has no one to catch a throw, so the event is dropped with a log + * line. A direct caller of pushEvent() still learns the framework is stopped + * by its throw. + */ +import { test, type TestContext } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { AgentFramework } from '../src/index.js'; +import type { Module, ModuleContext, ProcessEvent } from '../src/index.js'; +import { MockMembrane } from './helpers/mock-membrane.js'; + +/** A module that records the probe events it processes and keeps its context. */ +function recorder(onStop?: (ctx: ModuleContext) => void, name = 'recorder'): { module: Module; seen: string[]; context: () => ModuleContext } { + let context: ModuleContext | undefined; + const seen: string[] = []; + return { + seen, + context: () => context!, + module: { + name, + async start(ctx) { context = ctx; }, + async stop() { onStop?.(context!); }, + async onProcess(event) { if (event.type.startsWith('probe:')) seen.push(event.type); return {}; }, + getTools: () => [], + async handleToolCall() { return { success: false, error: 'no tools', isError: true }; }, + }, + }; +} + +async function makeFramework(t: TestContext, modules: Module[]): Promise { + const dir = mkdtempSync(join(tmpdir(), 'late-events-')); + t.after(() => rmSync(dir, { recursive: true, force: true })); + return AgentFramework.create({ + storePath: join(dir, 'store'), + membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'trunk', model: 'test-model', systemPrompt: 'test' }], + modules, + }); +} + +/** What the test logs to console.error. */ +function captureErrors(t: TestContext): string[] { + const logged: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }); + return logged; +} + +/** Fails, rather than hangs, when what the test waits on never comes. */ +async function within(promise: Promise, what: string, ms = 5000): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`timed out waiting for ${what}`)), ms); }), + ]); + } finally { + clearTimeout(timer); + } +} + +const dropped = (type: string, module: string) => + `[framework] a '${type}' event from module ${module} arrived after stop() began — dropped`; + +test("a module's events are queued through pushEvent() and its deprecated queue handle while the framework runs", async (t) => { + const probe = recorder(); + const framework = await makeFramework(t, [probe.module]); + t.after(() => framework.stop().catch(() => {})); + const traced: string[] = []; + framework.onTrace((event) => { if (event.type === 'process:received') traced.push(event.processEvent.type); }); + probe.context().pushEvent({ type: 'probe:through-push-event' } as ProcessEvent); + probe.context().queue.push({ type: 'probe:through-queue' } as ProcessEvent); + await within(framework.runUntilIdle(), 'the events processed'); + assert.deepEqual(probe.seen, ['probe:through-push-event', 'probe:through-queue']); + assert.deepEqual(traced.filter((type) => type.startsWith('probe:')), ['probe:through-push-event'], 'only pushEvent() traces, as before'); +}); + +test("a module's event from its own stop() is dropped with a log line naming that module, and stop() completes", async (t) => { + const pushes = (ctx: ModuleContext) => { + ctx.pushEvent({ type: 'probe:from-stop' } as ProcessEvent); + ctx.queue.push({ type: 'probe:from-stop-queue' } as ProcessEvent); + }; + const first = recorder(pushes, 'first'); + const second = recorder(pushes, 'second'); + const framework = await makeFramework(t, [first.module, second.module]); + const logged = captureErrors(t); + await within(framework.stop(), 'stop()'); + for (const name of ['first', 'second']) { + for (const type of ['probe:from-stop', 'probe:from-stop-queue']) { + assert.ok(logged.some((line) => line.includes(dropped(type, name))), `${name}'s ${type} drop is logged:\n${logged.join('\n')}`); + } + } + assert.equal(logged.some((line) => /Queue is closed/.test(line)), false, logged.join('\n')); +}); + +test("a module's tool result after stop() began keeps the drop line that names the call, the tool and its outcome", async (t) => { + const probe = recorder(); + const framework = await makeFramework(t, [probe.module]); + const logged = captureErrors(t); + await within(framework.stop(), 'stop()'); + const result = { type: 'tool-result', callId: 'c7', toolName: 'fetch', agentName: 'trunk', moduleName: 'recorder', result: { success: true } }; + probe.context().pushEvent(result as ProcessEvent); + probe.context().queue.push({ ...result, callId: 'c8' } as ProcessEvent); + for (const callId of ['c7', 'c8']) { + assert.ok( + logged.some((line) => line.includes(`[framework] trunk: result of tool call ${callId} (fetch, recorder) arrived after stop — dropped; the tool succeeded`)), + logged.join('\n'), + ); + } +}); + +test("a module's event after stop() finished is dropped with a log line, not thrown into its callback", async (t) => { + const probe = recorder(); + const framework = await makeFramework(t, [probe.module]); + const logged = captureErrors(t); + await within(framework.stop(), 'stop()'); + assert.doesNotThrow(() => probe.context().pushEvent({ type: 'probe:late-push-event' } as ProcessEvent)); + assert.doesNotThrow(() => probe.context().queue.push({ type: 'probe:late-queue' } as ProcessEvent)); + for (const type of ['probe:late-push-event', 'probe:late-queue']) { + assert.ok(logged.some((line) => line.includes(dropped(type, 'recorder'))), `${type}'s drop is logged:\n${logged.join('\n')}`); + } +}); + +test('a direct caller of pushEvent() still learns the framework is stopped', async (t) => { + const framework = await makeFramework(t, []); + await within(framework.stop(), 'stop()'); + assert.throws(() => framework.pushEvent({ type: 'probe:direct' } as ProcessEvent), /Queue is closed/); +});