tao.chen/codespace
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(web): AcpPanel streaming over WebSocket

Browse Source 
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
This commit is contained in:
tao.chen
2026-07-06 17:17:47 +08:00
parent 031451fe3e
commit ec54ea4473
2 changed files with 355 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/src/components/acp/AcpPanel.tsx
+142 -38
View File
@@ -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>
web/src/lib/api/acp.ts
+213
View File
@@ -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],
);
}