fix(framework): a module's event after stop() began is dropped with a log line, not thrown

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.

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.
    */

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,15 +84,15 @@ export class ModuleRegistry {
   private moduleContexts: Map<string, ModuleContextImpl> = 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;
   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<ToolResult>;
   /** Framework ops-alert channel, surfaced to modules via
@@ -101,15 +107,16 @@ export class ModuleRegistry {
 
   constructor(
     store: JsStore,
-    queue: ProcessQueue,
+    queue: ModuleQueue,
     options: {
       getAgents: () => Agent[];
       addMessage: (participant: string, content: ContentBlock[], metadata?: MessageMetadata) => MessageId;
       editMessage: (id: MessageId, content: ContentBlock[]) => void;
       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<ToolResult>;
       notifyOps?: (kind: string, agentName: string, message: string, data?: Record<string, unknown>) => 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<ToolResult>;
 
@@ -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<ToolResult>
   ) {
     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 {

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;
 }

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<AgentFramework> {
+  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<T>(promise: Promise<T>, what: string, ms = 5000): Promise<T> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  try {
+    return await Promise.race([
+      promise,
+      new Promise<never>((_, 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/);
+});