diff --git a/packages/sdk/src/actions/index.ts b/packages/sdk/src/actions/index.ts index 2f6ab2e3..d9fb2b1b 100644 --- a/packages/sdk/src/actions/index.ts +++ b/packages/sdk/src/actions/index.ts @@ -27,4 +27,11 @@ export { type UpdateGatewayResult, } from "./rpc"; -export { StreamAction, type StreamState, type StreamPayload } from "./stream"; +export { + StreamAction, + type StreamPayload, + type StreamEvent, + type StreamMessageEvent, + type StreamToolEvent, + extractTextFromEvent, +} from "./stream"; diff --git a/packages/sdk/src/actions/stream.ts b/packages/sdk/src/actions/stream.ts index da8dae31..51329040 100644 --- a/packages/sdk/src/actions/stream.ts +++ b/packages/sdk/src/actions/stream.ts @@ -2,19 +2,48 @@ export const StreamAction = "stream" as const; -/** 流消息状态 */ -export type StreamState = "delta" | "final" | "error"; +/** + * AgentEvent types forwarded by the Hub to frontend clients. + * These mirror the subset of AgentEvent from @mariozechner/pi-agent-core + * that the Hub forwards (filtered at the Hub layer). + */ +export interface StreamMessageEvent { + type: "message_start" | "message_update" | "message_end"; + message: { + id?: string; + role: string; + content?: Array<{ type: string; text?: string }>; + }; + assistantMessageEvent?: unknown; +} -/** 流消息 payload */ +export interface StreamToolEvent { + type: "tool_execution_start" | "tool_execution_end"; + toolCallId: string; + toolName: string; + args?: unknown; + result?: unknown; + isError?: boolean; +} + +export type StreamEvent = StreamMessageEvent | StreamToolEvent; + +/** 流消息 payload — wraps a raw AgentEvent with stream/agent identifiers */ export interface StreamPayload { - /** 流 ID(即 messageId),关联同一个流的所有消息 */ + /** 流 ID,关联同一个流的所有消息 */ streamId: string; /** 所属 agent ID */ agentId: string; - /** 流状态 */ - state: StreamState; - /** 累计文本内容(delta/final 时) */ - content?: string; - /** 错误信息(error 时) */ - error?: string; + /** Raw agent event from the engine */ + event: StreamEvent; +} + +/** Extract plain text from an AgentMessage content array */ +export function extractTextFromEvent(event: StreamMessageEvent): string { + const content = event.message?.content; + if (!Array.isArray(content)) return ""; + return content + .filter((c) => c.type === "text") + .map((c) => c.text ?? "") + .join(""); } diff --git a/packages/store/src/gateway.ts b/packages/store/src/gateway.ts index 5682b70a..ed1d4814 100644 --- a/packages/store/src/gateway.ts +++ b/packages/store/src/gateway.ts @@ -1,5 +1,5 @@ import { create } from "zustand" -import { GatewayClient, StreamAction, type ConnectionState, type DeviceInfo, type SendErrorResponse, type StreamPayload } from "@multica/sdk" +import { GatewayClient, StreamAction, extractTextFromEvent, type ConnectionState, type DeviceInfo, type SendErrorResponse, type StreamPayload, type StreamMessageEvent } from "@multica/sdk" import { useMessagesStore } from "./messages" const DEFAULT_GATEWAY_URL = "http://localhost:3000" @@ -45,26 +45,32 @@ export const useGatewayStore = create()((set, get) => ({ }) .onStateChange((connectionState) => set({ connectionState })) .onMessage((msg) => { - // Handle streaming messages + // Handle streaming messages (new protocol: payload.event is a raw AgentEvent) if (msg.action === StreamAction) { const payload = msg.payload as StreamPayload const store = useMessagesStore.getState() - switch (payload.state) { - case "delta": { - const exists = store.messages.some((m) => m.id === payload.streamId) - if (!exists) { - store.startStream(payload.streamId, payload.agentId) - } - if (payload.content) { - store.appendStream(payload.streamId, payload.content) - } + const { event } = payload + + switch (event.type) { + case "message_start": { + store.startStream(payload.streamId, payload.agentId) + const text = extractTextFromEvent(event as StreamMessageEvent) + if (text) store.appendStream(payload.streamId, text) break } - case "final": - store.endStream(payload.streamId, payload.content ?? "") + case "message_update": { + const text = extractTextFromEvent(event as StreamMessageEvent) + store.appendStream(payload.streamId, text) break - case "error": - store.endStream(payload.streamId, `[error] ${payload.error}`) + } + case "message_end": { + const text = extractTextFromEvent(event as StreamMessageEvent) + store.endStream(payload.streamId, text) + break + } + case "tool_execution_start": + case "tool_execution_end": + // TODO: surface tool execution status in UI break } return