491f98e
/**
* React binding for the agent gateway's session stream. Browser-direct by
* design: the backend only mints the session JWT (`GET …/tasks/{id}/token`),
* and this hook opens `GET /api/v1/sessions/{session}/events` itself.
*
* Three things the raw `AgentClient` does not do:
*
* - **Token lifetime.** The JWT lives 600s, which is shorter than a long run.
* The hook silently re-mints shortly before `expiresAt` and reopens the
* stream at the same watermark, so a subscriber never sees the gap.
* - **Resume across reconnects.** `Last-Event-ID` is derived from the frames
* the caller has already been told about, and it survives `fetchToken`
* rotation.
* - **State for React.** Messages are deduplicated by `seq` and kept ordered,
* because a resumed stream legitimately replays frames.
*/
import { useCallback, useEffect, useRef, useState } from 'react'
import { AgentClient } from './client'
import { AgentSdkError } from './errors'
import type { MessageDto, SseEvent } from './types'
/** What the token endpoint hands the browser. */
export interface AgentStreamToken {
sessionId: string
token: string
baseUrl: string
/** ISO-8601; the hook refreshes 60s before this. */
expiresAt: string
}
export type AgentStreamStatus =
| 'idle'
| 'connecting'
| 'live'
| 'reconnecting'
| 'expired'
| 'error'
| 'stopped'
export interface UseAgentStreamOptions {
/** Gateway session behind the task; `null` keeps the hook idle. */
sessionId?: string | null
/** Callers can hold the stream closed (e.g. the panel is collapsed). */
enabled?: boolean
fetchToken: () => Promise<AgentStreamToken>
onMessage?: (message: MessageDto) => void
onEvent?: (event: SseEvent) => void
onRunState?: (state: string) => void
}
export interface AgentStreamState {
status: AgentStreamStatus
/** Live frames only, `seq`-ascending. Merge with the archive for display. */
messages: MessageDto[]
runState?: string
error?: string
/** Highest `message_added` seq seen — the resume cursor. */
watermark: number
}
export interface UseAgentStreamResult extends AgentStreamState {
/** POST the text straight to the gateway (the API proxy has no send route). */
send: (text: string) => Promise<void>
/** Drop the token and reopen the stream at the current watermark. */
reconnect: () => void
}
const REFRESH_LEAD_MS = 60_000
/**
* Keep one session stream open for as long as the caller is mounted. The
* returned `messages` are the frames this client actually saw; the caller owns
* the historical ones.
*/
export function useAgentStream(options: UseAgentStreamOptions): UseAgentStreamResult {
const { sessionId, enabled = true, fetchToken } = options
const [state, setState] = useState<AgentStreamState & { sessionId: string | null }>({
status: 'idle',
messages: [],
watermark: 0,
sessionId: null,
})
const [generation, setGeneration] = useState(0)
const handlers = useRef(options)
const fetchTokenRef = useRef(fetchToken)
const tokenRef = useRef<AgentStreamToken | null>(null)
const clientRef = useRef<AgentClient | null>(null)
const watermarkRef = useRef(0)
const watermarkSessionRef = useRef<string | null>(null)
const sessionRef = useRef<string | null>(sessionId ?? null)
// The latest callbacks and inputs live in refs so the stream is opened once per
// generation: a re-render (a keystroke in the composer) must not reconnect.
useEffect(() => {
handlers.current = options
fetchTokenRef.current = fetchToken
sessionRef.current = sessionId ?? null
})
const ensureClient = useCallback(async (): Promise<AgentClient> => {
const session = sessionRef.current
if (!session) throw new AgentSdkError('会话尚未创建，暂时无法连接智能体网关')
const current = clientRef.current
if (current !== null && current.sessionId === session) return current
// Minting a JWT is a round trip to the API host, so the panel says
// "connecting" from here rather than from the effect that opened the stream.
setState(prev => ({ ...prev, status: 'connecting', error: undefined }))
const minted = tokenRef.current ?? (await fetchTokenRef.current())
tokenRef.current = minted
const client = new AgentClient({
baseUrl: minted.baseUrl,
token: minted.token,
sessionId: minted.sessionId || session,
})
clientRef.current = client
return client
}, [])
const reconnect = useCallback(() => {
tokenRef.current = null
clientRef.current = null
setGeneration(n => n + 1)
}, [])
useEffect(() => {
if (!enabled || !sessionId) return
const controller = new AbortController()
let timer: ReturnType<typeof setTimeout> | undefined
void (async () => {
try {
const client = await ensureClient()
if (controller.signal.aborted) return
// The resume cursor belongs to one gateway session: reusing it across
// tasks would ask the new session for frames it never had.
if (watermarkSessionRef.current !== sessionId) {
watermarkRef.current = 0
watermarkSessionRef.current = sessionId
}
setState(prev => ({
...prev,
status: 'live',
error: undefined,
sessionId,
watermark: watermarkRef.current,
messages: prev.sessionId === sessionId ? prev.messages : [],
}))
const minted = tokenRef.current
if (minted !== null) {
// Silent rotation: the SDK's own retry loop would replay a dead JWT.
const until = Date.parse(minted.expiresAt) - Date.now() - REFRESH_LEAD_MS
timer = setTimeout(
() => {
tokenRef.current = null
setGeneration(n => n + 1)
},
Math.max(1_000, Number.isFinite(until) ? until : 0),
)
}
await client.streamEvents({
signal: controller.signal,
lastEventId: watermarkRef.current,
onMessage: message => {
watermarkRef.current = Math.max(watermarkRef.current, message.seq)
setState(prev => ({
...prev,
watermark: watermarkRef.current,
messages: prev.messages.some(existing => existing.seq === message.seq)
? prev.messages
: [...prev.messages, message].sort((a, b) => a.seq - b.seq),
}))
handlers.current.onMessage?.(message)
},
onEvent: event => handlers.current.onEvent?.(event),
onError: error => {
if (error.isUnauthorized || error.isForbidden) {
tokenRef.current = null
clientRef.current = null
setState(prev => ({ ...prev, status: 'expired', error: error.message }))
setGeneration(n => n + 1)
return false
}
setState(prev => ({ ...prev, status: 'reconnecting', error: error.message }))
return undefined
},
})
if (!controller.signal.aborted) setState(prev => ({ ...prev, status: 'stopped' }))
} catch (cause) {
if (controller.signal.aborted) return
setState(prev => ({
...prev,
status: 'error',
error: cause instanceof Error ? cause.message : String(cause),
}))
}
})()
return () => {
controller.abort()
if (timer !== undefined) clearTimeout(timer)
clientRef.current = null
}
}, [enabled, sessionId, generation, ensureClient])
const send = useCallback(
async (text: string) => {
const client = await ensureClient()
await client.sendMessage(text)
},
[ensureClient],
)
// A stream that is not being held open for this session reports `idle` and no
// frames, whatever the previous session left behind.
const attached = enabled && Boolean(sessionId) && state.sessionId === sessionId
return {
...state,
status: attached ? state.status : 'idle',
messages: attached ? state.messages : [],
watermark: attached ? state.watermark : 0,
send,
reconnect,
}
}
/**
* Merge the archive (rows the backend's subscriber persisted) with the live
* frames, newest per `seq`, oldest first.
*/
export function mergeAgentMessages<T extends { seq: number }>(
archive: readonly T[],
live: readonly T[],
): T[] {
if (live.length === 0) return [...archive]
const bySeq = new Map<number, T>()
for (const message of archive) bySeq.set(message.seq, message)
for (const message of live) bySeq.set(message.seq, message)
return [...bySeq.values()].sort((a, b) => a.seq - b.seq)
}