Merge pull request #69 from multica-ai/fix/streaming-protocol-sync

fix: align streaming protocol and add generation indicator

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";

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("");
 }

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<GatewayStore>()((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

packages/ui/src/components/markdown/StreamingMarkdown.tsx

@@ -1,5 +1,6 @@
 import * as React from 'react'
 import { Markdown, type RenderMode } from './Markdown'
+import { Spinner } from '@multica/ui/components/spinner'
 
 export interface StreamingMarkdownProps {
   content: string
@@ -162,6 +163,17 @@ export function StreamingMarkdown({
     )
   }
 
+  const indicator = (
+    <div className="flex items-center gap-2 py-1 text-muted-foreground">
+      <Spinner className="text-xs" />
+      <span className="text-xs">Generating...</span>
+    </div>
+  )
+
+  if (blocks.length === 0) {
+    return indicator
+  }
+
   return (
     <>
       {blocks.map((block, i) => {
@@ -169,7 +181,7 @@ export function StreamingMarkdown({
 
         // Complete blocks use content hash as key -> stable identity -> memoized
         // Last block uses "active" prefix -> always re-renders on content change
-        const key = isLastBlock ? `active-${i}` : `block-${simpleHash(block.content)}`
+        const key = isLastBlock ? `active-${i}` : `block-${i}-${simpleHash(block.content)}`
 
         return (
           <MemoizedBlock
@@ -181,6 +193,7 @@ export function StreamingMarkdown({
           />
         )
       })}
+      {indicator}
     </>
   )
 }