refactor(agents): centralize typed activity event emission (#122188)

src/agents/embedded-agent-runner/run/progress-controller.ts

@@ -2,7 +2,7 @@ import {
   FAST_MODE_AUTO_PROGRESS_KIND,
   type ReplyPayload,
 } from "../../../auto-reply/reply-payload.js";
-import { emitAgentItemEvent } from "../../../infra/agent-activity-events.js";
+import { emitAgentActivityEvent } from "../../../infra/agent-activity-events.js";
 import { formatErrorMessage } from "../../../infra/errors.js";
 import { resolveFastModeModelAutoOnSeconds } from "../../../shared/fast-mode.js";
 import {
@@ -54,9 +54,10 @@ export function createEmbeddedRunProgressController(params: {
   }) => {
     const summary = formatFastModeAutoProgressText(payload);
     try {
-      emitAgentItemEvent({
+      emitAgentActivityEvent({
         runId: params.attempt.runId,
         ...(params.attempt.sessionKey ? { sessionKey: params.attempt.sessionKey } : {}),
+        stream: "item",
         data: {
           itemId: `fast-mode-auto:${payload.enabled ? "on" : "off"}`,
           kind: "status",

src/agents/embedded-agent-subscribe.handlers.tools.completion.ts

@@ -4,15 +4,11 @@ import {
   normalizeHeartbeatToolResponse,
 } from "../auto-reply/heartbeat-tool-response.js";
 import { parseSessionThreadInfoFast } from "../config/sessions/thread-info.js";
-import type {
-  AgentCommandOutputEventData,
-  AgentItemEventData,
-  AgentPatchSummaryEventData,
-} from "../infra/agent-activity-events.js";
 import {
-  emitAgentApprovalEvent,
-  emitAgentCommandOutputEvent,
-  emitAgentPatchSummaryEvent,
+  emitAgentActivityEvent,
+  type AgentCommandOutputEventData,
+  type AgentItemEventData,
+  type AgentPatchSummaryEventData,
 } from "../infra/agent-activity-events.js";
 import { emitAgentEvent, type AgentApprovalEventData } from "../infra/agent-events.js";
 import type { PluginHookAfterToolCallEvent } from "../plugins/types.js";
@@ -452,9 +448,10 @@ export async function handleToolExecutionEnd(
         ...(execDetails.status === "approval-unavailable" ? { reason: execDetails.reason } : {}),
         message: execDetails.warningText,
       };
-      emitAgentApprovalEvent({
+      emitAgentActivityEvent({
         runId: ctx.params.runId,
         ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
+        stream: "approval",
         data: approvalData,
       });
       emitAgentEventCallbackBestEffort(ctx, {
@@ -519,9 +516,10 @@ export async function handleToolExecutionEnd(
           ? { cwd: execDetails.cwd }
           : {}),
       };
-      emitAgentCommandOutputEvent({
+      emitAgentActivityEvent({
         runId: ctx.params.runId,
         ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
+        stream: "command_output",
         data: outputData,
       });
       emitAgentEventCallbackBestEffort(ctx, {
@@ -545,9 +543,10 @@ export async function handleToolExecutionEnd(
             toolCallId,
             message: parsedApprovalResult.body || parsedApprovalResult.raw,
           };
-          emitAgentApprovalEvent({
+          emitAgentActivityEvent({
             runId: ctx.params.runId,
             ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
+            stream: "approval",
             data: approvalData,
           });
           emitAgentEventCallbackBestEffort(ctx, {
@@ -591,9 +590,10 @@ export async function handleToolExecutionEnd(
         deleted: patchSummary.deleted,
         summary: summaryText ?? buildPatchSummaryText(patchSummary),
       };
-      emitAgentPatchSummaryEvent({
+      emitAgentActivityEvent({
         runId: ctx.params.runId,
         ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
+        stream: "patch",
         data: patchData,
       });
       emitAgentEventCallbackBestEffort(ctx, {

src/agents/embedded-agent-subscribe.handlers.tools.progress.ts

@@ -3,11 +3,11 @@ import {
   asOptionalRecord as readRecordField,
 } from "@openclaw/normalization-core/record-coerce";
 import { readStringValue } from "@openclaw/normalization-core/string-coerce";
-import type {
-  AgentCommandOutputEventData,
-  AgentItemEventData,
+import {
+  emitAgentActivityEvent,
+  type AgentCommandOutputEventData,
+  type AgentItemEventData,
 } from "../infra/agent-activity-events.js";
-import { emitAgentCommandOutputEvent } from "../infra/agent-activity-events.js";
 import { emitAgentEvent } from "../infra/agent-events.js";
 import { extractLiveExecOutput } from "./embedded-agent-subscribe.handlers.tools.results.js";
 import {
@@ -146,9 +146,10 @@ export function handleToolExecutionUpdate(
         output,
         status: "running",
       };
-      emitAgentCommandOutputEvent({
+      emitAgentActivityEvent({
         runId: ctx.params.runId,
         ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
+        stream: "command_output",
         data: outputData,
       });
       emitAgentEventCallbackBestEffort(ctx, {

src/agents/embedded-agent-subscribe.handlers.tools.start.ts

@@ -5,8 +5,7 @@ import {
 } from "@openclaw/normalization-core/string-coerce";
 import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
 import { parseSessionThreadInfoFast } from "../config/sessions/thread-info.js";
-import type { AgentItemEventData } from "../infra/agent-activity-events.js";
-import { emitAgentItemEvent } from "../infra/agent-activity-events.js";
+import { emitAgentActivityEvent, type AgentItemEventData } from "../infra/agent-activity-events.js";
 import { emitAgentEvent } from "../infra/agent-events.js";
 import { REQUIRED_PARAM_GROUPS, type RequiredParamGroup } from "./agent-tools.params.js";
 import { sanitizeForConsole } from "./console-sanitize.js";
@@ -255,9 +254,10 @@ export function emitTrackedItemEvent(ctx: ToolHandlerContext, itemData: AgentIte
     ctx.state.itemActiveIds.delete(itemData.itemId);
     ctx.state.itemCompletedCount += 1;
   }
-  emitAgentItemEvent({
+  emitAgentActivityEvent({
     runId: ctx.params.runId,
     ...(ctx.params.sessionKey ? { sessionKey: ctx.params.sessionKey } : {}),
+    stream: "item",
     data: itemData,
   });
   emitAgentEventCallbackBestEffort(ctx, {

src/infra/agent-activity-events.test.ts

@@ -0,0 +1,105 @@
+import { beforeEach, describe, expect, expectTypeOf, test } from "vitest";
+import {
+  emitAgentActivityEvent,
+  type AgentCommandOutputEventData,
+  type AgentItemEventData,
+  type AgentPatchSummaryEventData,
+} from "./agent-activity-events.js";
+import {
+  type AgentApprovalEventData,
+  type AgentEventPayload,
+  onAgentEvent,
+  resetAgentEventsForTest,
+} from "./agent-events.js";
+
+describe("agent activity events", () => {
+  beforeEach(() => {
+    resetAgentEventsForTest();
+  });
+
+  test("emits every activity stream with shared sequencing and context", () => {
+    const itemData: AgentItemEventData = {
+      itemId: "item-1",
+      phase: "start",
+      kind: "tool",
+      title: "Read",
+      status: "running",
+    };
+    const approvalData: AgentApprovalEventData = {
+      phase: "requested",
+      kind: "exec",
+      status: "pending",
+      title: "Approve",
+    };
+    const commandData: AgentCommandOutputEventData = {
+      itemId: "command-1",
+      phase: "delta",
+      title: "Command",
+      toolCallId: "tool-1",
+      output: "working",
+    };
+    const patchData: AgentPatchSummaryEventData = {
+      itemId: "patch-1",
+      phase: "end",
+      title: "Patch",
+      toolCallId: "tool-2",
+      added: ["new.ts"],
+      modified: [],
+      deleted: [],
+      summary: "Added new.ts",
+    };
+    const events: AgentEventPayload[] = [];
+    const unsubscribe = onAgentEvent((event) => events.push(event));
+
+    emitAgentActivityEvent({
+      runId: "run-1",
+      sessionKey: "session-1",
+      stream: "item",
+      data: itemData,
+    });
+    emitAgentActivityEvent({
+      runId: "run-1",
+      sessionKey: "session-1",
+      stream: "approval",
+      data: approvalData,
+    });
+    emitAgentActivityEvent({
+      runId: "run-1",
+      sessionKey: "session-1",
+      stream: "command_output",
+      data: commandData,
+    });
+    emitAgentActivityEvent({
+      runId: "run-1",
+      sessionKey: "",
+      stream: "patch",
+      data: patchData,
+    });
+
+    expect(
+      events.map(({ runId, seq, stream, sessionKey }) => ({ runId, seq, stream, sessionKey })),
+    ).toEqual([
+      { runId: "run-1", seq: 1, stream: "item", sessionKey: "session-1" },
+      { runId: "run-1", seq: 2, stream: "approval", sessionKey: "session-1" },
+      { runId: "run-1", seq: 3, stream: "command_output", sessionKey: "session-1" },
+      { runId: "run-1", seq: 4, stream: "patch", sessionKey: undefined },
+    ]);
+    expect(events.map((event) => event.data)).toEqual([
+      itemData,
+      approvalData,
+      commandData,
+      patchData,
+    ]);
+    expect(events[0]?.data).toBe(itemData);
+    unsubscribe();
+  });
+
+  test("rejects mismatched stream and payload pairs", () => {
+    type ItemData = Extract<
+      Parameters<typeof emitAgentActivityEvent>[0],
+      { stream: "item" }
+    >["data"];
+
+    expectTypeOf<AgentApprovalEventData>().not.toMatchTypeOf<ItemData>();
+  });
+});

src/infra/agent-activity-events.ts

@@ -58,57 +58,27 @@ export type AgentPatchSummaryEventData = {
   summary: string;
 };
 
-/** Emits an item activity event on the shared agent event bus. */
-export function emitAgentItemEvent(params: {
-  runId: string;
-  data: AgentItemEventData;
-  sessionKey?: string;
-}) {
-  emitAgentEvent({
-    runId: params.runId,
-    stream: "item",
-    data: params.data as unknown as Record<string, unknown>,
-    ...(params.sessionKey ? { sessionKey: params.sessionKey } : {}),
-  });
-}
+type AgentActivityEventDataByStream = {
+  item: AgentItemEventData;
+  approval: AgentApprovalEventData;
+  command_output: AgentCommandOutputEventData;
+  patch: AgentPatchSummaryEventData;
+};
 
-/** Emits an approval event on the shared agent event bus. */
-export function emitAgentApprovalEvent(params: {
-  runId: string;
-  data: AgentApprovalEventData;
-  sessionKey?: string;
-}) {
-  emitAgentEvent({
-    runId: params.runId,
-    stream: "approval",
-    data: params.data as unknown as Record<string, unknown>,
-    ...(params.sessionKey ? { sessionKey: params.sessionKey } : {}),
-  });
-}
+type AgentActivityEventParams = {
+  [Stream in keyof AgentActivityEventDataByStream]: {
+    runId: string;
+    sessionKey?: string;
+    stream: Stream;
+    data: AgentActivityEventDataByStream[Stream];
+  };
+}[keyof AgentActivityEventDataByStream];
 
-/** Emits command output for a running or completed item/tool call. */
-export function emitAgentCommandOutputEvent(params: {
-  runId: string;
-  data: AgentCommandOutputEventData;
-  sessionKey?: string;
-}) {
+/** Emits a typed activity event on the shared agent event bus. */
+export function emitAgentActivityEvent(params: AgentActivityEventParams): void {
   emitAgentEvent({
     runId: params.runId,
-    stream: "command_output",
-    data: params.data as unknown as Record<string, unknown>,
-    ...(params.sessionKey ? { sessionKey: params.sessionKey } : {}),
-  });
-}
-
-/** Emits a patch summary for a completed file-editing item/tool call. */
-export function emitAgentPatchSummaryEvent(params: {
-  runId: string;
-  data: AgentPatchSummaryEventData;
-  sessionKey?: string;
-}) {
-  emitAgentEvent({
-    runId: params.runId,
-    stream: "patch",
+    stream: params.stream,
     data: params.data as unknown as Record<string, unknown>,
     ...(params.sessionKey ? { sessionKey: params.sessionKey } : {}),
   });