fix: guard tool event callbacks (AI-assisted) (#81696)

enjoylife1243 · web-flow · commit 6368c1173ccf · 2026-06-19T05:13:35.000Z
Summary: - This PR wraps embedded-agent tool-handler onExecutionPhase and per-run onAgentEvent emissions in best-effort warning guards and adds regression tests for throwing and rejecting callbacks. - PR surface: Source +31, Tests +44. Total +75 across 2 files. - Reproducibility: yes. Current main directly invokes the relevant callbacks in the tool-start and tool-event ... sync observer can leak unless guarded; I did not run a failing current-main repro in this read-only review. Automerge notes: - No ClawSweeper repair was needed after automerge opt-in. Validation: - ClawSweeper review passed for head 65de17d. - Required merge gates passed before the squash merge. Prepared head SHA: 65de17d Review: #81696 (comment) Co-authored-by: xuyi1243 <maginaxwhz@gmail.com>
diff --git a/src/agents/embedded-agent-subscribe.handlers.tools.test.ts b/src/agents/embedded-agent-subscribe.handlers.tools.test.ts
@@ -334,6 +334,50 @@ describe("handleToolExecutionStart read path checks", () => {
     expect(ctx.state.itemActiveIds.has("tool:tool-await-flush")).toBe(true);
     expect(ctx.state.itemActiveIds.has("command:tool-await-flush")).toBe(true);
   });
+
+  it("keeps processing tool start when progress callbacks throw", async () => {
+    const { ctx, warn, onExecutionPhase, onAgentEvent } = createTestContext();
+    onExecutionPhase.mockImplementation(() => {
+      throw new Error("phase exploded");
+    });
+    onAgentEvent.mockImplementation(() => {
+      throw new Error("event exploded");
+    });
+
+    const evt: ToolExecutionStartEvent = {
+      type: "tool_execution_start",
+      toolName: "exec",
+      toolCallId: "tool-callback-throws",
+      args: { command: "echo hi" },
+    };
+
+    await handleToolExecutionStart(ctx, evt);
+
+    expect(ctx.state.toolMetaById.has("tool-callback-throws")).toBe(true);
+    expect(ctx.state.itemStartedCount).toBe(2);
+    expect(warn).toHaveBeenCalledWith(
+      expect.stringContaining("tool execution phase callback failed"),
+    );
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining("tool agent event callback failed"));
+  });
+
+  it("does not leak unhandled rejections when tool start progress rejects", async () => {
+    const { ctx, warn, onAgentEvent } = createTestContext();
+    onAgentEvent.mockRejectedValue(new Error("progress failed"));
+
+    const evt: ToolExecutionStartEvent = {
+      type: "tool_execution_start",
+      toolName: "exec",
+      toolCallId: "tool-callback-rejects",
+      args: { command: "echo hi" },
+    };
+
+    await handleToolExecutionStart(ctx, evt);
+    await Promise.resolve();
+
+    expect(ctx.state.toolMetaById.has("tool-callback-rejects")).toBe(true);
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining("tool agent event callback failed"));
+  });
 });
 
 describe("handleToolExecutionEnd cron mutation tracking", () => {
diff --git a/src/agents/embedded-agent-subscribe.handlers.tools.ts b/src/agents/embedded-agent-subscribe.handlers.tools.ts
@@ -299,12 +299,43 @@ function emitTrackedItemEvent(ctx: ToolHandlerContext, itemData: AgentItemEventD
     ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
     data: itemData,
   });
-  void ctx.params.onAgentEvent?.({
+  emitAgentEventCallbackBestEffort(ctx, {
     stream: "item",
     data: itemData,
   });
 }
 
+function warnBestEffortEventFailure(ctx: ToolHandlerContext, label: string, error: unknown): void {
+  ctx.log.warn(`${label} callback failed: ${String(error)}`);
+}
+
+function emitExecutionPhaseBestEffort(
+  ctx: ToolHandlerContext,
+  info: Parameters<NonNullable<ToolHandlerContext["params"]["onExecutionPhase"]>>[0],
+): void {
+  try {
+    ctx.params.onExecutionPhase?.(info);
+  } catch (error) {
+    warnBestEffortEventFailure(ctx, "tool execution phase", error);
+  }
+}
+
+function emitAgentEventCallbackBestEffort(
+  ctx: ToolHandlerContext,
+  event: Parameters<NonNullable<ToolHandlerContext["params"]["onAgentEvent"]>>[0],
+): void {
+  try {
+    const result = ctx.params.onAgentEvent?.(event);
+    if (isPromiseLike<void>(result)) {
+      void Promise.resolve(result).catch((error: unknown) => {
+        warnBestEffortEventFailure(ctx, "tool agent event", error);
+      });
+    }
+  } catch (error) {
+    warnBestEffortEventFailure(ctx, "tool agent event", error);
+  }
+}
+
 function readToolResultDetailsRecord(result: unknown): Record<string, unknown> | undefined {
   return readRecordField(asOptionalObjectRecord(result)?.details);
 }
@@ -780,7 +811,7 @@ export function handleToolExecutionStart(
     const args = evt.args;
     const runId = ctx.params.runId;
     ctx.state.toolExecutionSinceLastBlockReply = true;
-    ctx.params.onExecutionPhase?.({
+    emitExecutionPhaseBestEffort(ctx, {
       phase: "tool_execution_started",
       tool: toolName,
       toolCallId,
@@ -898,7 +929,7 @@ export function handleToolExecutionStart(
     };
     emitTrackedItemEvent(ctx, itemData);
     // Best-effort typing signal; do not block tool summaries on slow emitters.
-    void ctx.params.onAgentEvent?.({
+    emitAgentEventCallbackBestEffort(ctx, {
       stream: "tool",
       data: {
         phase: "start",
@@ -1037,7 +1068,7 @@ export function handleToolExecutionUpdate(
   };
   emitTrackedItemEvent(ctx, itemData);
   if (!toolProgress) {
-    void ctx.params.onAgentEvent?.({
+    emitAgentEventCallbackBestEffort(ctx, {
       stream: "tool",
       data: {
         phase: "update",
@@ -1075,7 +1106,7 @@ export function handleToolExecutionUpdate(
         ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
         data: outputData,
       });
-      void ctx.params.onAgentEvent?.({
+      emitAgentEventCallbackBestEffort(ctx, {
         stream: "command_output",
         data: outputData,
       });
@@ -1322,7 +1353,7 @@ export async function handleToolExecutionEnd(
       : {}),
   };
   emitTrackedItemEvent(ctx, itemData);
-  void ctx.params.onAgentEvent?.({
+  emitAgentEventCallbackBestEffort(ctx, {
     stream: "tool",
     data: {
       phase: "result",
@@ -1368,7 +1399,7 @@ export async function handleToolExecutionEnd(
         ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
         data: approvalData,
       });
-      void ctx.params.onAgentEvent?.({
+      emitAgentEventCallbackBestEffort(ctx, {
         stream: "approval",
         data: approvalData,
       });
@@ -1435,7 +1466,7 @@ export async function handleToolExecutionEnd(
         ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
         data: outputData,
       });
-      void ctx.params.onAgentEvent?.({
+      emitAgentEventCallbackBestEffort(ctx, {
         stream: "command_output",
         data: outputData,
       });
@@ -1461,7 +1492,7 @@ export async function handleToolExecutionEnd(
             ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
             data: approvalData,
           });
-          void ctx.params.onAgentEvent?.({
+          emitAgentEventCallbackBestEffort(ctx, {
             stream: "approval",
             data: approvalData,
           });
@@ -1507,7 +1538,7 @@ export async function handleToolExecutionEnd(
         ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
         data: patchData,
       });
-      void ctx.params.onAgentEvent?.({
+      emitAgentEventCallbackBestEffort(ctx, {
         stream: "patch",
         data: patchData,
       });
