diff --git a/src/agents/embedded-agent-runner/run/progress-controller.ts b/src/agents/embedded-agent-runner/run/progress-controller.ts index 9029da5a7189..e65c74a26fbf 100644 --- a/src/agents/embedded-agent-runner/run/progress-controller.ts +++ b/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", diff --git a/src/agents/embedded-agent-subscribe.handlers.tools.completion.ts b/src/agents/embedded-agent-subscribe.handlers.tools.completion.ts index 6fbcd93368e4..89e1d61c021a 100644 --- a/src/agents/embedded-agent-subscribe.handlers.tools.completion.ts +++ b/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, { diff --git a/src/agents/embedded-agent-subscribe.handlers.tools.progress.ts b/src/agents/embedded-agent-subscribe.handlers.tools.progress.ts index 3282c289ed4e..b473aec2b2bb 100644 --- a/src/agents/embedded-agent-subscribe.handlers.tools.progress.ts +++ b/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, { diff --git a/src/agents/embedded-agent-subscribe.handlers.tools.start.ts b/src/agents/embedded-agent-subscribe.handlers.tools.start.ts index 39a73758f7c1..cdfa34867928 100644 --- a/src/agents/embedded-agent-subscribe.handlers.tools.start.ts +++ b/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, { diff --git a/src/infra/agent-activity-events.test.ts b/src/infra/agent-activity-events.test.ts new file mode 100644 index 000000000000..16b6260b19d8 --- /dev/null +++ b/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[0], + { stream: "item" } + >["data"]; + + expectTypeOf().not.toMatchTypeOf(); + }); +}); diff --git a/src/infra/agent-activity-events.ts b/src/infra/agent-activity-events.ts index 1d906807e5e9..ce506a42975b 100644 --- a/src/infra/agent-activity-events.ts +++ b/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, - ...(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, - ...(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, - ...(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, ...(params.sessionKey ? { sessionKey: params.sessionKey } : {}), });