From 0b903168c85a05e5a2469aee5dcecbc98def9271 Mon Sep 17 00:00:00 2001
From: Jerry Yan <792602257@qq.com>
Date: Thu, 7 May 2026 17:59:07 +0800
Subject: [PATCH] feat(web): useAcpStream composable (auto-reconnect + ?since=
 + prompt/cancel)

---
 web/src/composables/useAcpStream.ts | 160 ++++++++++++++++++++++++++++
 1 file changed, 160 insertions(+)
 create mode 100644 web/src/composables/useAcpStream.ts

diff --git a/web/src/composables/useAcpStream.ts b/web/src/composables/useAcpStream.ts
new file mode 100644
index 0000000..642e512
--- /dev/null
+++ b/web/src/composables/useAcpStream.ts
@@ -0,0 +1,160 @@
+// useAcpStream wraps the ACP WebSocket per spec §10.3:
+//   - connect + auto-reconnect once with exponential backoff (1s/3s, then give up)
+//   - ?since=lastEventID resumption
+//   - send session/prompt / session/cancel (generates client-<uuid> ID)
+//   - accumulate events into a ref array
+//   - expose sessionStatus + connection-state status
+import { ref, onScopeDispose, type Ref } from 'vue'
+import { openSessionWS, type AcpEvent } from '@/api/acp'
+import { useAuthStore } from '@/stores/auth'
+
+export type StreamStatus = 'idle' | 'connecting' | 'streaming' | 'closed' | 'error'
+export type SessionStatus = 'starting' | 'running' | 'crashed' | 'exited'
+
+export interface UseAcpStreamReturn {
+  status: Ref<StreamStatus>
+  sessionStatus: Ref<SessionStatus>
+  events: Ref<AcpEvent[]>
+  prompt: (text: string, agentSessionID: string) => void
+  cancel: (agentSessionID: string) => void
+  reconnect: () => void
+  close: () => void
+}
+
+interface ServerStateEvent {
+  kind: 'state'
+  status: SessionStatus
+}
+
+interface SessionTerminatedPayload {
+  status?: 'crashed' | 'exited'
+}
+
+export function useAcpStream(sessionId: string): UseAcpStreamReturn {
+  const status = ref<StreamStatus>('idle')
+  const sessionStatus = ref<SessionStatus>('starting')
+  const events = ref<AcpEvent[]>([])
+  const lastEventID = ref(0)
+
+  let ws: WebSocket | null = null
+  let retried = 0
+  let manualClose = false
+  let reconnectTimer: ReturnType<typeof setTimeout> | null = null
+
+  const auth = useAuthStore()
+
+  function connect() {
+    if (ws || manualClose) return
+    status.value = 'connecting'
+    const token = auth.token ?? ''
+    const sock = openSessionWS(sessionId, lastEventID.value, token)
+    ws = sock
+
+    sock.onopen = () => {
+      status.value = 'streaming'
+      retried = 0
+    }
+
+    sock.onmessage = (ev: MessageEvent) => {
+      let parsed: unknown
+      try {
+        parsed = JSON.parse(ev.data as string)
+      } catch {
+        return
+      }
+      if (!parsed || typeof parsed !== 'object') return
+      const obj = parsed as Record<string, unknown>
+
+      // Initial state event from server
+      if (obj.kind === 'state') {
+        const s = (parsed as ServerStateEvent).status
+        if (s) sessionStatus.value = s
+        return
+      }
+
+      // Final session_terminated event
+      if (obj.kind === 'session_terminated') {
+        let payloadObj: SessionTerminatedPayload = {}
+        const rawPayload = obj.payload
+        if (typeof rawPayload === 'string') {
+          try {
+            payloadObj = JSON.parse(rawPayload) as SessionTerminatedPayload
+          } catch {
+            payloadObj = {}
+          }
+        } else if (rawPayload && typeof rawPayload === 'object') {
+          payloadObj = rawPayload as SessionTerminatedPayload
+        }
+        sessionStatus.value = payloadObj.status === 'crashed' ? 'crashed' : 'exited'
+        events.value.push(parsed as AcpEvent)
+        return
+      }
+
+      // Regular event envelope
+      const ev2 = parsed as AcpEvent
+      if (typeof ev2.id === 'number' && ev2.id > lastEventID.value) {
+        lastEventID.value = ev2.id
+      }
+      events.value.push(ev2)
+    }
+
+    sock.onerror = () => {
+      status.value = 'error'
+    }
+
+    sock.onclose = () => {
+      ws = null
+      if (manualClose) {
+        status.value = 'closed'
+        return
+      }
+      if (retried < 2) {
+        const delay = retried === 0 ? 1000 : 3000
+        retried += 1
+        reconnectTimer = setTimeout(connect, delay)
+      } else {
+        status.value = 'closed'
+      }
+    }
+  }
+
+  function send(method: string, params: Record<string, unknown>) {
+    if (!ws || ws.readyState !== WebSocket.OPEN) return
+    const id = 'client-' + crypto.randomUUID()
+    const msg = { jsonrpc: '2.0', id, method, params }
+    ws.send(JSON.stringify(msg))
+  }
+
+  function prompt(text: string, agentSessionID: string) {
+    send('session/prompt', {
+      sessionId: agentSessionID,
+      prompt: [{ type: 'text', text }],
+    })
+  }
+
+  function cancel(agentSessionID: string) {
+    send('session/cancel', { sessionId: agentSessionID })
+  }
+
+  function reconnect() {
+    if (ws) ws.close()
+    retried = 0
+    manualClose = false
+    connect()
+  }
+
+  function close() {
+    manualClose = true
+    if (reconnectTimer) {
+      clearTimeout(reconnectTimer)
+      reconnectTimer = null
+    }
+    if (ws) ws.close()
+    status.value = 'closed'
+  }
+
+  connect()
+  onScopeDispose(close)
+
+  return { status, sessionStatus, events, prompt, cancel, reconnect, close }
+}