From ec54ea44733fb6ff8c152b4562448fd095ddb677 Mon Sep 17 00:00:00 2001
From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com>
Date: Mon, 6 Jul 2026 17:17:47 +0800
Subject: [PATCH] feat(web): AcpPanel streaming over WebSocket

Replace the synchronous useAcpPrompt mutation with a streaming WS
hook. The agent's reply now arrives chunk-by-chunk and the same
agent bubble grows in real time.

- web/src/lib/api/acp.ts:
  - Add AcpStreamEvent / AcpStreamStatus types.
  - Add useAcpStream(workspaceId) hook: per-prompt WS lifecycle
    (idle -> connecting -> open -> closed / error), open(content)
    / close() / onEvent(cb) API. Reuses the reconnection / 1000-1001
    / pinger pattern from useProcessWebSocket.
  - Keep useAcpPrompt and useAcpCancel exported for back-compat.
- web/src/components/acp/AcpPanel.tsx:
  - Switch to useAcpStream. handleSend pushes an optimistic user
    message, opens the stream, sets an inflight agent bubble with
    empty text.
  - chunk events append to inflight.text; complete clears inflight;
    error clears inflight and shows a local error toast.
  - renderMessages = coalesced + localUserMessage + inflight.
  - Local user message is auto-cleared when the next history poll
    surfaces the same text (avoids duplicate render).
  - Auto-scroll effect depends on inflight.text so it fires as
    chunks stream.
  - Cancel button closes the WS (server's read loop triggers
    session/cancel).
  - Remove the standalone ThinkingIndicator; the in-flight bubble
    serves as the streaming affordance.

Conversation: 7B62CB8E-ACC8-4333-BC64-B927C8FDC397
---
 web/src/components/acp/AcpPanel.tsx | 180 ++++++++++++++++++-----
 web/src/lib/api/acp.ts              | 213 ++++++++++++++++++++++++++++
 2 files changed, 355 insertions(+), 38 deletions(-)

diff --git a/web/src/components/acp/AcpPanel.tsx b/web/src/components/acp/AcpPanel.tsx
index 4f97c4d..b364e41 100644
--- a/web/src/components/acp/AcpPanel.tsx
+++ b/web/src/components/acp/AcpPanel.tsx
@@ -1,18 +1,12 @@
 import { Ban, Send } from "lucide-react";
-import { useEffect, useMemo, useRef, useState } from "react";
+import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 
 import { Button } from "@/components/ui/button";
 import { ScrollArea } from "@/components/ui/scroll-area";
 import { MessageBubble } from "./MessageBubble";
-import { ThinkingIndicator } from "./ThinkingIndicator";
-import { groupMessages } from "./groupMessages";
+import { groupMessages, type CoalescedMessage } from "./groupMessages";
 
-import {
-  useAcpCancel,
-  useAcpHistory,
-  useAcpPrompt,
-  useAcpStatus,
-} from "@/lib/api/acp";
+import { useAcpHistory, useAcpStatus, useAcpStream } from "@/lib/api/acp";
 import { cn } from "@/lib/utils";
 
 interface AcpPanelProps {
@@ -71,13 +65,22 @@ function StatusPill({ status }: { status: ReturnType<typeof useAcpStatus> }) {
 export function AcpPanel({ workspaceId }: AcpPanelProps) {
   const acpStatus = useAcpStatus(workspaceId);
   const acpHistory = useAcpHistory(workspaceId);
-  const prompt = useAcpPrompt();
-  const cancel = useAcpCancel();
+  const stream = useAcpStream(workspaceId);
+  const { onEvent, close } = stream;
 
   const [input, setInput] = useState("");
   const [localErrors, setLocalErrors] = useState<LocalError[]>([]);
+  const [localUserMessage, setLocalUserMessage] =
+    useState<CoalescedMessage | null>(null);
+  const [inflight, setInflight] = useState<{
+    messageId: string;
+    text: string;
+  } | null>(null);
+
   const scrollRootRef = useRef<HTMLDivElement | null>(null);
   const sentinelRef = useRef<HTMLDivElement | null>(null);
+  const localTimeoutRef = useRef<number | null>(null);
+  const errorTimeoutsRef = useRef<Set<number>>(new Set());
 
   const messages = useMemo(
     () => acpHistory.data?.messages ?? [],
@@ -85,19 +88,81 @@ export function AcpPanel({ workspaceId }: AcpPanelProps) {
   );
   const coalesced = useMemo(() => groupMessages(messages), [messages]);
 
-  useEffect(() => {
-    if (prompt.error) {
-      const id = Date.now();
-      setLocalErrors((prev) => [
-        ...prev,
-        { id, message: prompt.error?.message ?? "Failed to send prompt" },
-      ]);
-      const timeout = window.setTimeout(() => {
-        setLocalErrors((prev) => prev.filter((e) => e.id !== id));
-      }, 5000);
-      return () => window.clearTimeout(timeout);
+  const pushError = useCallback((message: string) => {
+    const id = Date.now();
+    setLocalErrors((prev) => [...prev, { id, message }]);
+    const timeout = window.setTimeout(() => {
+      errorTimeoutsRef.current.delete(timeout);
+      setLocalErrors((prev) => prev.filter((e) => e.id !== id));
+    }, 5000);
+    errorTimeoutsRef.current.add(timeout);
+  }, []);
+
+  const scheduleLocalClear = useCallback(() => {
+    if (localTimeoutRef.current !== null) {
+      window.clearTimeout(localTimeoutRef.current);
     }
-  }, [prompt.error]);
+    localTimeoutRef.current = window.setTimeout(() => {
+      localTimeoutRef.current = null;
+      setLocalUserMessage(null);
+    }, 6000);
+  }, []);
+
+  useEffect(() => {
+    return onEvent((event) => {
+      switch (event.type) {
+        case "chunk":
+          setInflight((cur) =>
+            cur
+              ? {
+                  messageId: event.messageId || cur.messageId,
+                  text: cur.text + event.text,
+                }
+              : cur,
+          );
+          break;
+        case "complete":
+          setInflight(null);
+          close();
+          break;
+        case "error":
+          setInflight(null);
+          pushError(event.error || "stream error");
+          close();
+          break;
+      }
+    });
+  }, [onEvent, close, pushError]);
+
+  useEffect(() => {
+    if (stream.status === "error") {
+      setInflight(null);
+      pushError("stream connection failed");
+    }
+  }, [stream.status, pushError]);
+
+  useEffect(() => {
+    if (!localUserMessage) return;
+    const lastUser = [...coalesced].reverse().find((m) => m.role === "user");
+    if (lastUser && lastUser.text === localUserMessage.text) {
+      setLocalUserMessage(null);
+      if (localTimeoutRef.current !== null) {
+        window.clearTimeout(localTimeoutRef.current);
+        localTimeoutRef.current = null;
+      }
+    }
+  }, [coalesced, localUserMessage]);
+
+  useEffect(() => {
+    const timeoutsSet = errorTimeoutsRef.current;
+    return () => {
+      if (localTimeoutRef.current !== null) {
+        window.clearTimeout(localTimeoutRef.current);
+      }
+      const timeouts = Array.from(timeoutsSet);
+      timeouts.forEach((timeout) => window.clearTimeout(timeout));
+    };
+  }, []);
 
   useEffect(() => {
     const root = scrollRootRef.current;
@@ -114,13 +179,26 @@ export function AcpPanel({ workspaceId }: AcpPanelProps) {
     if (distanceFromBottom <= 80) {
       sentinel.scrollIntoView({ behavior: "smooth", block: "end" });
     }
-  }, [coalesced, prompt.isPending]);
+  }, [coalesced, inflight?.text, localUserMessage]);
 
   const handleSend = () => {
     const content = input.trim();
-    if (!content || !workspaceId || prompt.isPending) return;
-    prompt.mutate({ workspaceId, content });
+    if (!content || !workspaceId || inflight) return;
+    setLocalUserMessage({
+      role: "user",
+      text: content,
+      time: new Date().toISOString(),
+    });
+    scheduleLocalClear();
     setInput("");
+    setInflight({ messageId: "", text: "" });
+    stream.open(content);
+  };
+
+  const handleCancel = () => {
+    if (!inflight) return;
+    stream.close();
+    setInflight(null);
   };
 
   const handleKeyDown = (e: React.KeyboardEvent<HTMLTextAreaElement>) => {
@@ -137,6 +215,21 @@ export function AcpPanel({ workspaceId }: AcpPanelProps) {
       ? `${sessionId.slice(0, 12)}…`
       : sessionId;
 
+  const renderMessages = useMemo(() => {
+    const items: CoalescedMessage[] = [...coalesced];
+    if (localUserMessage) {
+      items.push(localUserMessage);
+    }
+    if (inflight) {
+      items.push({
+        role: "agent",
+        text: inflight.text,
+        time: new Date().toISOString(),
+      });
+    }
+    return items;
+  }, [coalesced, localUserMessage, inflight]);
+
   if (!workspaceId) {
     return (
       <section className="flex h-full w-full items-center justify-center bg-background text-sm text-muted-foreground">
@@ -164,15 +257,26 @@ export function AcpPanel({ workspaceId }: AcpPanelProps) {
 
       <ScrollArea ref={scrollRootRef} className="flex-1">
         <div className="flex flex-col gap-3 p-3">
-          {messages.length === 0 && isReady && (
-            <div className="flex flex-1 items-center justify-center py-8 text-sm text-muted-foreground">
-              Send a message to start the conversation.
-            </div>
-          )}
-          {coalesced.map((message) => (
-            <MessageBubble key={message.time} message={message} />
+          {messages.length === 0 &&
+            !localUserMessage &&
+            !inflight &&
+            isReady && (
+              <div className="flex flex-1 items-center justify-center py-8 text-sm text-muted-foreground">
+                Send a message to start the conversation.
+              </div>
+            )}
+          {renderMessages.map((message, index) => (
+            <MessageBubble
+              key={
+                localUserMessage && message === localUserMessage
+                  ? "local-user"
+                  : inflight && index === renderMessages.length - 1
+                    ? "inflight-agent"
+                    : message.time
+              }
+              message={message}
+            />
           ))}
-          {prompt.isPending && <ThinkingIndicator />}
           {localErrors.map((err) => (
             <div
               key={err.id}
@@ -196,12 +300,12 @@ export function AcpPanel({ workspaceId }: AcpPanelProps) {
           disabled={!workspaceId}
           className="max-h-32 min-h-8 flex-1 resize-none rounded-md border border-input bg-transparent px-3 py-2 text-sm shadow-sm placeholder:text-muted-foreground focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-ring disabled:cursor-not-allowed disabled:opacity-50"
         />
-        {prompt.isPending ? (
+        {inflight ? (
           <Button
             variant="outline"
             size="sm"
-            onClick={() => workspaceId && cancel.mutate(workspaceId)}
-            disabled={cancel.isPending}
+            onClick={handleCancel}
+            disabled={stream.status !== "open"}
           >
             <Ban className="size-4" aria-hidden="true" />
             <span className="sr-only">Cancel</span>
@@ -210,7 +314,7 @@ export function AcpPanel({ workspaceId }: AcpPanelProps) {
           <Button
             size="sm"
             onClick={handleSend}
-            disabled={!input.trim() || prompt.isPending}
+            disabled={!input.trim() || !!inflight}
           >
             <Send className="size-4" aria-hidden="true" />
             <span className="sr-only">Send</span>
diff --git a/web/src/lib/api/acp.ts b/web/src/lib/api/acp.ts
index 30e12b8..ac8beb9 100644
--- a/web/src/lib/api/acp.ts
+++ b/web/src/lib/api/acp.ts
@@ -7,6 +7,13 @@ import {
 } from "@tanstack/react-query";
 
 import { apiClient } from "@/lib/api/client";
+import {
+  useCallback,
+  useEffect,
+  useMemo,
+  useRef,
+  useState,
+} from "react";
 
 export interface AcpStatus {
   workspaceId: string;
@@ -34,6 +41,25 @@ export interface AcpPromptResponse {
   text: string;
 }
 
+export type AcpStreamEvent =
+  | { type: "chunk"; messageId?: string; text: string }
+  | { type: "complete"; stopReason: string }
+  | { type: "error"; error: string };
+
+export type AcpStreamStatus =
+  | "idle"
+  | "connecting"
+  | "open"
+  | "closed"
+  | "error";
+
+export interface AcpStream {
+  status: AcpStreamStatus;
+  open: (content: string) => void;
+  close: () => void;
+  onEvent: (cb: (event: AcpStreamEvent) => void) => () => void;
+}
+
 function acpStatusQueryKey(workspaceId: string) {
   return ["workspaces", workspaceId, "acp", "status"] as const;
 }
@@ -147,3 +173,190 @@ export function useAcpCancel(): UseMutationResult<void, Error, string> {
     },
   });
 }
+
+function isAcpStreamEvent(data: unknown): data is AcpStreamEvent {
+  if (!data || typeof data !== "object") {
+    return false;
+  }
+  const ev = data as Record<string, unknown>;
+  if (ev.type === "chunk") {
+    return typeof ev.text === "string";
+  }
+  if (ev.type === "complete") {
+    return typeof ev.stopReason === "string";
+  }
+  if (ev.type === "error") {
+    return typeof ev.error === "string";
+  }
+  return false;
+}
+
+export function useAcpStream(workspaceId: string | null): AcpStream {
+  const [status, setStatus] = useState<AcpStreamStatus>("idle");
+  const statusRef = useRef(status);
+  useEffect(() => {
+    statusRef.current = status;
+  }, [status]);
+
+  const socketRef = useRef<WebSocket | null>(null);
+  const reconnectAttemptsRef = useRef(0);
+  const reconnectTimeoutRef = useRef<number | null>(null);
+  const intentionalCloseRef = useRef(false);
+  const serverCleanCloseRef = useRef(false);
+  const terminalEventRef = useRef(false);
+  const pendingContentRef = useRef<string | null>(null);
+  const callbacksRef = useRef(new Set<(event: AcpStreamEvent) => void>());
+
+  const clearReconnectTimeout = useCallback(() => {
+    if (reconnectTimeoutRef.current !== null) {
+      window.clearTimeout(reconnectTimeoutRef.current);
+      reconnectTimeoutRef.current = null;
+    }
+  }, []);
+
+  const closeSocket = useCallback(() => {
+    intentionalCloseRef.current = true;
+    clearReconnectTimeout();
+    const socket = socketRef.current;
+    if (socket) {
+      socket.close();
+      socketRef.current = null;
+    }
+    setStatus("closed");
+  }, [clearReconnectTimeout]);
+
+  const open = useCallback(
+    (content: string) => {
+      if (!workspaceId) {
+        return;
+      }
+
+      // Close any existing socket before starting a fresh prompt stream.
+      if (socketRef.current) {
+        intentionalCloseRef.current = true;
+        const socket = socketRef.current;
+        socket.onclose = null;
+        socket.close();
+        socketRef.current = null;
+        intentionalCloseRef.current = false;
+      }
+
+      clearReconnectTimeout();
+      reconnectAttemptsRef.current = 0;
+      serverCleanCloseRef.current = false;
+      terminalEventRef.current = false;
+      pendingContentRef.current = content;
+
+      const openSocket = () => {
+        const wsUrl =
+          (window.location.protocol === "https:" ? "wss:" : "ws:") +
+          "//" +
+          window.location.host +
+          "/api/workspaces/" +
+          encodeURIComponent(workspaceId) +
+          "/acp/stream";
+
+        const ws = new WebSocket(wsUrl);
+        socketRef.current = ws;
+        setStatus("connecting");
+
+        ws.onopen = () => {
+          reconnectAttemptsRef.current = 0;
+          serverCleanCloseRef.current = false;
+          terminalEventRef.current = false;
+          setStatus("open");
+          const contentToSend = pendingContentRef.current;
+          if (contentToSend !== null) {
+            ws.send(JSON.stringify({ type: "prompt", content: contentToSend }));
+          }
+        };
+
+        ws.onmessage = (event) => {
+          let data: unknown;
+          try {
+            data = JSON.parse(
+              typeof event.data === "string" ? event.data : "",
+            );
+          } catch {
+            return;
+          }
+          if (!isAcpStreamEvent(data)) {
+            return;
+          }
+          callbacksRef.current.forEach((cb) => cb(data));
+          if (data.type === "complete" || data.type === "error") {
+            terminalEventRef.current = true;
+            ws.close();
+          }
+        };
+
+        ws.onclose = (event) => {
+          socketRef.current = null;
+          if (
+            (event.code === 1000 || event.code === 1001) &&
+            !intentionalCloseRef.current
+          ) {
+            serverCleanCloseRef.current = true;
+          }
+          if (
+            intentionalCloseRef.current ||
+            terminalEventRef.current ||
+            serverCleanCloseRef.current
+          ) {
+            setStatus("closed");
+            return;
+          }
+          const attempts = reconnectAttemptsRef.current;
+          if (attempts < 3) {
+            setStatus("connecting");
+            reconnectTimeoutRef.current = window.setTimeout(() => {
+              reconnectTimeoutRef.current = null;
+              openSocket();
+            }, Math.pow(2, attempts) * 1000);
+            reconnectAttemptsRef.current = attempts + 1;
+          } else {
+            setStatus("error");
+          }
+        };
+
+        ws.onerror = () => {
+          // The browser fires onclose after an error; handle retries there.
+        };
+      };
+
+      openSocket();
+    },
+    [workspaceId, clearReconnectTimeout],
+  );
+
+  useEffect(() => {
+    return () => {
+      clearReconnectTimeout();
+      const socket = socketRef.current;
+      if (socket) {
+        socket.onclose = null;
+        socket.close();
+        socketRef.current = null;
+      }
+    };
+  }, [workspaceId, clearReconnectTimeout]);
+
+  const onEvent = useCallback(
+    (cb: (event: AcpStreamEvent) => void) => {
+      callbacksRef.current.add(cb);
+      return () => {
+        callbacksRef.current.delete(cb);
+      };
+    },
+    [],
+  );
+
+  const close = useCallback(() => {
+    closeSocket();
+  }, [closeSocket]);
+
+  return useMemo(
+    () => ({ status, open, close, onEvent }),
+    [status, open, close, onEvent],
+  );
+}