Feature/stream agent replies

apps/web/app/api/daemon/client.integration.test.ts

@@ -171,6 +171,37 @@ describe("remote daemon client integration", () => {
         ],
       });
 
+      await client.reportMessages(claimed.task.id, {
+        messages: [
+          {
+            type: "text",
+            content: "我先",
+          },
+        ],
+      });
+      await client.reportMessages(claimed.task.id, {
+        messages: [
+          {
+            type: "text",
+            content: "给你整理。",
+          },
+        ],
+      });
+
+      const streamingState = readWorkspaceStateSync();
+      const streamingDirectChannel = streamingState.channels.find(
+        (channel) => channel.kind === "direct" && channel.employeeNames.some((name) => name === "Atlas"),
+      );
+      const streamingPending = streamingState.messages.find(
+        (message) =>
+          message.channel === streamingDirectChannel?.name &&
+          message.role === "agent" &&
+          message.status === "pending" &&
+          message.speaker === "Atlas",
+      );
+      expect(streamingPending?.summary).toBe("我先给你整理。");
+      expect(streamingPending?.data?.stream_started).toBe("true");
+
       await client.uploadOutputBundle(claimed.task.id, {
         version: 1,
         format: "json-inline-v1",

apps/web/app/api/daemon/tasks/[taskId]/messages/route.ts

@@ -5,6 +5,8 @@ import {
   type QueuedTaskRecord,
 } from "@agent-space/db";
 import type { DaemonProvider, DaemonTaskMessageInput, ReportTaskMessagesRequest } from "@agent-space/domain";
+import { parseTaskPayload } from "agent-space-daemon";
+import { streamAgentChannelReplyDeltaSync } from "@agent-space/services";
 import { readTaskForWorkspace, requireDaemonAuth } from "../../../_lib/auth";
 
 export const runtime = "nodejs";
@@ -30,14 +32,20 @@ export async function POST(
     return Response.json({ error: "messages[] is required." }, { status: 400 });
   }
 
-  const appended = body.messages.map((message) => appendSingleMessage(task, message));
+  const payload = parseTaskPayload(task);
+  const appended = body.messages.map((message) => appendSingleMessage(task, payload, message));
   return Response.json({ messages: appended });
 }
 
-function appendSingleMessage(task: QueuedTaskRecord, message: DaemonTaskMessageInput) {
+function appendSingleMessage(
+  task: QueuedTaskRecord,
+  payload: ReturnType<typeof parseTaskPayload>,
+  message: DaemonTaskMessageInput,
+) {
   if (message.type === "provider_session_invalid") {
     handleProviderSessionInvalid(task, message);
   }
+  streamChannelReplyDeltaBestEffort(task, payload, message);
   return appendTaskMessageSync({
     taskId: task.id,
     type: message.type === "provider_session_invalid" ? "status" : message.type,
@@ -48,6 +56,34 @@ function appendSingleMessage(task: QueuedTaskRecord, message: DaemonTaskMessageI
   });
 }
 
+function streamChannelReplyDeltaBestEffort(
+  task: QueuedTaskRecord,
+  payload: ReturnType<typeof parseTaskPayload>,
+  message: DaemonTaskMessageInput,
+): void {
+  if (message.type !== "text" || !message.content) {
+    return;
+  }
+  const channelName = payload.channelName ?? payload.channel;
+  const pendingSpeaker = payload.assignee ?? payload.contactId ?? task.agentId;
+  if (!channelName || !pendingSpeaker) {
+    return;
+  }
+
+  try {
+    streamAgentChannelReplyDeltaSync({
+      channel: channelName,
+      pendingSpeaker,
+      delta: message.content,
+      sourceTaskQueueId: task.id,
+    }, task.workspaceId);
+  } catch (error) {
+    console.error(
+      `Failed to stream channel reply delta for ${task.id}: ${error instanceof Error ? error.message : String(error)}`,
+    );
+  }
+}
+
 function handleProviderSessionInvalid(task: QueuedTaskRecord, message: DaemonTaskMessageInput): void {
   if (!task.routerSessionId) {
     return;

apps/web/app/api/workspaces/[workspaceId]/channels/[channelName]/events/route.test.ts

@@ -57,12 +57,13 @@ describe("channel realtime events route", () => {
 
   it("streams matching channel events without leaking other channels", async () => {
     let listener: ((event: {
-      type: "channel.message.created";
+      type: "channel.message.created" | "channel.message.updated";
       workspaceId: string;
       channelName: string;
       messageId: string;
       sequence: number;
-      createdAt: string;
+      createdAt?: string;
+      updatedAt?: string;
     }) => void) | null = null;
     mockSubscribeWorkspaceRealtimeEvents.mockImplementation((_workspaceId, nextListener) => {
       listener = nextListener;
@@ -104,6 +105,21 @@ describe("channel realtime events route", () => {
     expect(eventText).toContain("event: channel.message.created");
     expect(eventText).toContain("message-visible");
     expect(eventText).not.toContain("message-hidden");
+
+    emit({
+      type: "channel.message.updated",
+      workspaceId: "workspace-1",
+      channelName: "general",
+      messageId: "message-visible",
+      sequence: 3,
+      updatedAt: "2026-05-01T00:00:02.000Z",
+    });
+
+    const updateChunk = await reader.read();
+    const updateText = decoder.decode(updateChunk.value);
+    expect(updateText).toContain("event: channel.message.updated");
+    expect(updateText).toContain("message-visible");
+    expect(updateText).toContain("2026-05-01T00:00:02.000Z");
     await reader.cancel();
   });
 });

apps/web/app/api/workspaces/[workspaceId]/channels/[channelName]/events/route.ts

@@ -74,6 +74,14 @@ export async function GET(
                 sequence: event.sequence,
                 createdAt: event.createdAt,
               }
+            : event.type === "channel.message.updated"
+              ? {
+                  type: event.type,
+                  channelName: event.channelName,
+                  messageId: event.messageId,
+                  sequence: event.sequence,
+                  updatedAt: event.updatedAt,
+                }
             : event.type === "channel.thread.changed"
               ? {
                   type: event.type,

apps/web/features/channels/channels-page-client.test.tsx

@@ -402,6 +402,7 @@ describe("ChannelsPageClient", () => {
 
     const composer = screen.getByPlaceholderText("发送到 tour visit");
     fireEvent.change(composer, { target: { value: "草稿", selectionStart: 2 } });
+    composer.focus();
     expect(eventSources[0]?.url).toBe("/api/workspaces/workspace-1/channels/tour%20visit/events");
 
     const activeSource = eventSources.at(-1);
@@ -410,10 +411,15 @@ describe("ChannelsPageClient", () => {
       messageId: "message-2",
       sequence: 1,
     });
+    activeSource?.emit("channel.message.updated", {
+      channelName: "tour visit",
+      messageId: "message-2",
+      sequence: 2,
+    });
     activeSource?.emit("channel.message.created", {
       channelName: "tour visit",
       messageId: "message-3",
-      sequence: 2,
+      sequence: 3,
     });
 
     expect(routerRefreshMock).not.toHaveBeenCalled();

apps/web/features/channels/channels-page-client.tsx

@@ -420,10 +420,12 @@ function useChannelRealtimeRefresh({
     };
 
     source.addEventListener("channel.message.created", scheduleRefresh as EventListener);
+    source.addEventListener("channel.message.updated", scheduleRefresh as EventListener);
     source.addEventListener("channel.thread.changed", scheduleRefresh as EventListener);
 
     return () => {
       source.removeEventListener("channel.message.created", scheduleRefresh as EventListener);
+      source.removeEventListener("channel.message.updated", scheduleRefresh as EventListener);
       source.removeEventListener("channel.thread.changed", scheduleRefresh as EventListener);
       source.close();
       if (refreshTimerRef.current !== null) {
@@ -822,11 +824,11 @@ export function ChannelsPageClient({
 
   const selectedChannel = selectedChannelId ? visibleChannelById.get(selectedChannelId) ?? null : null;
   const selectedConversationChannelName = resolveSelectedChannelName(selectedChannel);
-  const refreshChannelData = useCallback(() => {
+  const refreshChannelData = useCallback((options?: { allowWhileInputActive?: boolean }) => {
     if (typeof document !== "undefined" && document.visibilityState === "hidden") {
       return;
     }
-    if (isDocumentInputActive()) {
+    if (!options?.allowWhileInputActive && isDocumentInputActive()) {
       return;
     }
     if (refreshInFlightRef.current || transitionPendingRef.current) {
@@ -943,7 +945,7 @@ export function ChannelsPageClient({
       Boolean(selectedConversationChannelName) &&
       !selectedChannelRequiresAccess,
     onInvalidation,
-    refresh: refreshChannelData,
+    refresh: () => refreshChannelData({ allowWhileInputActive: true }),
   });
 
   useEffect(() => {

apps/web/features/chat/chat-primitives.test.tsx

@@ -1,6 +1,6 @@
-import { fireEvent, render, screen } from "@testing-library/react";
+import { act, fireEvent, render, screen } from "@testing-library/react";
 import userEvent from "@testing-library/user-event";
-import { beforeEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { ChatAttachmentRow, ConversationMessageBubble } from "@/features/chat/chat-primitives";
 import { LanguageProvider } from "@/features/i18n/language-provider";
 import type { MessageAttachment } from "@/shared/types/workspace";
@@ -53,6 +53,10 @@ describe("ConversationMessageBubble", () => {
     window.localStorage.clear();
   });
 
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
   it("translates the system speaker label in English", () => {
     render(
       <LanguageProvider initialLanguage="en">
@@ -116,7 +120,7 @@ describe("ConversationMessageBubble", () => {
     const onReviewApproval = vi.fn(async () => {});
 
     render(
-      <LanguageProvider>
+      <LanguageProvider initialLanguage="zh">
         <ConversationMessageBubble
           message={{
             id: "message-approval",
@@ -147,4 +151,76 @@ describe("ConversationMessageBubble", () => {
 
     expect(onReviewApproval).toHaveBeenCalledWith("approval-1", "approved");
   });
+
+  it("reveals streamed pending agent reply content instead of hiding it behind dots", () => {
+    vi.useFakeTimers();
+
+    const { container } = render(
+      <LanguageProvider>
+        <ConversationMessageBubble
+          message={{
+            id: "message-stream-pending",
+            speaker: "Atlas",
+            role: "agent",
+            content: "我正在整理第一版。",
+            code: "agent.pending",
+            data: { stream_started: "true", source_task_queue_id: "queue-1" },
+            timestamp: "10:00",
+            status: "pending",
+          }}
+        />
+      </LanguageProvider>,
+    );
+
+    expect(screen.queryByText("我正在整理第一版。")).not.toBeInTheDocument();
+
+    act(() => {
+      vi.advanceTimersByTime(16);
+    });
+    expect(screen.getByText("我")).toBeInTheDocument();
+    expect(container.querySelector(".contacts-pending-dots")).toBeInTheDocument();
+
+    act(() => {
+      vi.advanceTimersByTime(16 * 8);
+    });
+    expect(screen.getByText("我正在整理第一版。")).toBeInTheDocument();
+    expect(container.querySelector(".contacts-pending-dots")).not.toBeInTheDocument();
+  });
+
+  it("reveals completed streamed agent replies progressively", () => {
+    vi.useFakeTimers();
+
+    render(
+      <LanguageProvider>
+        <ConversationMessageBubble
+          message={{
+            id: "message-stream-complete",
+            speaker: "Atlas",
+            role: "agent",
+            content: "abcdef",
+            data: { stream_started: "true", source_task_queue_id: "queue-1" },
+            timestamp: "10:00",
+            status: "completed",
+          }}
+        />
+      </LanguageProvider>,
+    );
+
+    expect(screen.queryByText("abcdef")).not.toBeInTheDocument();
+
+    act(() => {
+      vi.advanceTimersByTime(16);
+    });
+    expect(screen.getByText("a")).toBeInTheDocument();
+
+    act(() => {
+      vi.advanceTimersByTime(16 * 2);
+    });
+    expect(screen.getByText("abc")).toBeInTheDocument();
+
+    act(() => {
+      vi.advanceTimersByTime(16 * 3);
+    });
+    expect(screen.getByText("abcdef")).toBeInTheDocument();
+  });
 });