fix(session): isolate event listener failures in agent fan-out

Both `AgentSession.#emit` (session/agent-session.ts) and `Agent.#emit`
(packages/agent/src/agent.ts) iterated listeners with no error isolation.
A synchronous throw in any subscriber aborted the for-loop, so later
subscribers (TUI rendering, ACP bridge, task executor progress,
hindsight) silently missed events. Many listeners — see
`modes/controllers/event-controller.ts:141` and
`modes/controllers/input-controller.ts:576` — are registered as
`async (event) => { await this.handleEvent(event); }`; the returned
Promise was dropped, so any rejection became an unhandled rejection.

Wrap each listener invocation in try/catch and attach a `.catch` to any
returned thenable. Errors are logged via `logger.warn` (already imported
in agent-session.ts) and `console.error` (agent.ts has no logger
dependency, keep it that way).

Test: new `test/session/emit-listener-isolation.test.ts` registers two
listeners on both classes; first listener throws (or returns a rejecting
Promise); asserts the second listener still receives the event AND no
`unhandledRejection` fires. 4 cases (sync+async × Agent+AgentSession).
All fail on current main; all pass with the fix.
This commit is contained in:
oldschoola
2026-05-26 23:26:15 -07:00
parent 159f2d8fb1
commit 08bccfe0ec
3 changed files with 171 additions and 3 deletions
+10 -1
View File
@@ -1074,7 +1074,16 @@ export class Agent {
#emit(e: AgentEvent) {
for (const listener of this.#listeners) {
listener(e);
try {
const result = listener(e) as unknown;
if (result && typeof (result as Promise<unknown>).then === "function") {
(result as Promise<unknown>).catch(err => {
console.error("Agent listener rejected:", err instanceof Error ? err.message : err);
});
}
} catch (err) {
console.error("Agent listener threw:", err instanceof Error ? err.message : err);
}
}
}
@@ -1307,10 +1307,25 @@ export class AgentSession {
/** Emit an event to all listeners */
#emit(event: AgentSessionEvent): void {
// Copy array before iteration to avoid mutation during iteration
// Copy array before iteration to avoid mutation during iteration.
const listeners = [...this.#eventListeners];
for (const l of listeners) {
l(event);
try {
const result = l(event);
// Listener may be an async function whose returned Promise we don't await;
// attach a catch so a rejection does not become an unhandled rejection.
if (result && typeof (result as Promise<unknown>).then === "function") {
(result as Promise<unknown>).catch(err => {
logger.warn("AgentSession listener rejected", {
error: err instanceof Error ? err.message : String(err),
});
});
}
} catch (err) {
logger.warn("AgentSession listener threw", {
error: err instanceof Error ? err.message : String(err),
});
}
}
}
@@ -0,0 +1,144 @@
// Tests #emit listener isolation in agent.ts (Agent) and agent-session.ts (AgentSession):
// a throwing or rejecting listener must not prevent later listeners from receiving the event,
// and an async listener's rejection must not become an unhandled rejection.
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import * as fs from "node:fs";
import * as os from "node:os";
import * as path from "node:path";
import { Agent, type AgentEvent } from "@oh-my-pi/pi-agent-core";
import { getBundledModel } from "@oh-my-pi/pi-ai";
import { createMockModel } from "@oh-my-pi/pi-ai/providers/mock";
import { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry";
import { Settings } from "@oh-my-pi/pi-coding-agent/config/settings";
import { AgentSession } from "@oh-my-pi/pi-coding-agent/session/agent-session";
import { AuthStorage } from "@oh-my-pi/pi-coding-agent/session/auth-storage";
import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manager";
import { Snowflake } from "@oh-my-pi/pi-utils";
function makeEvent(): AgentEvent {
return { type: "tool_execution_start", toolCallId: "probe-1", toolName: "probe" };
}
describe("#emit listener isolation", () => {
describe("Agent.#emit (packages/agent/src/agent.ts)", () => {
it("continues to deliver to later listeners when an earlier listener throws synchronously", () => {
const agent = new Agent();
const calls: string[] = [];
agent.subscribe(() => {
calls.push("A");
throw new Error("listener A boom");
});
agent.subscribe(() => {
calls.push("B");
});
agent.emitExternalEvent(makeEvent());
expect(calls).toEqual(["A", "B"]);
});
it("does not surface an unhandled rejection when an async listener rejects", async () => {
const agent = new Agent();
const unhandled: unknown[] = [];
const onUnhandled = (reason: unknown) => unhandled.push(reason);
process.on("unhandledRejection", onUnhandled);
const calls: string[] = [];
try {
agent.subscribe(async () => {
calls.push("A");
throw new Error("async A boom");
});
agent.subscribe(() => {
calls.push("B");
});
agent.emitExternalEvent(makeEvent());
await Bun.sleep(10);
expect(calls).toEqual(["A", "B"]);
expect(unhandled).toEqual([]);
} finally {
process.off("unhandledRejection", onUnhandled);
}
});
});
describe("AgentSession.#emit (packages/coding-agent/src/session/agent-session.ts)", () => {
let session: AgentSession;
let tempDir: string;
let authStorage: AuthStorage | undefined;
beforeEach(async () => {
tempDir = path.join(os.tmpdir(), `pi-emit-iso-${Snowflake.next()}`);
fs.mkdirSync(tempDir, { recursive: true });
const model = getBundledModel("anthropic", "claude-sonnet-4-5");
if (!model) throw new Error("Test model not found");
authStorage = await AuthStorage.create(path.join(tempDir, "auth.db"));
authStorage.setRuntimeApiKey("anthropic", "test-key");
const modelRegistry = new ModelRegistry(authStorage, path.join(tempDir, "models.yml"));
const mock = createMockModel({ responses: [{ content: ["ok"] }] });
const agent = new Agent({
initialState: { model, systemPrompt: ["t"], tools: [], messages: [] },
streamFn: mock.stream,
});
session = new AgentSession({
agent,
sessionManager: SessionManager.inMemory(),
settings: Settings.isolated(),
modelRegistry,
});
});
afterEach(async () => {
await session.dispose();
authStorage?.close();
authStorage = undefined;
if (fs.existsSync(tempDir)) {
try {
fs.rmSync(tempDir, { recursive: true, force: true });
} catch {
// Windows may hold sqlite handles briefly after close; best-effort cleanup.
}
}
});
it("continues to deliver to later listeners when an earlier listener throws synchronously", () => {
const calls: string[] = [];
session.subscribe(() => {
calls.push("A");
throw new Error("session listener A boom");
});
session.subscribe(() => {
calls.push("B");
});
session.emitNotice("info", "probe", "test");
expect(calls).toEqual(["A", "B"]);
});
it("does not surface an unhandled rejection when an async listener rejects", async () => {
const unhandled: unknown[] = [];
const onUnhandled = (reason: unknown) => unhandled.push(reason);
process.on("unhandledRejection", onUnhandled);
const calls: string[] = [];
try {
session.subscribe(async () => {
calls.push("A");
throw new Error("async session A boom");
});
session.subscribe(() => {
calls.push("B");
});
session.emitNotice("info", "probe", "test");
await Bun.sleep(10);
expect(calls).toEqual(["A", "B"]);
expect(unhandled).toEqual([]);
} finally {
process.off("unhandledRejection", onUnhandled);
}
});
});
});