feat(web): stream agent runs straight from the gateway

Modified web/scripts/lint-i18n.mjs +1 −0 View web/scripts/lint-i18n.mjs

@@ -34,6 +34,7 @@

3434/** Files whose CJK text is authored content, not translatable copy. */

3535const CJK_EXEMPT = [

3636 'src/shared/api/', // generated client; descriptions come from the OpenAPI doc

37 'src/shared/sagent/', // vendored upstream agent SDK (agent/sdk/ts-sdk), kept byte-identical

3738 'src/shared/i18n/messages/', // the dictionaries themselves

3839 'src/shared/i18n/locale.ts', // `localeLabel` is an endonym table: 简体中文 is

3940 // the correct label for Chinese even while the UI is in English

Modified web/src/features/agent-console/ui/agent-detail.tsx +122 −11 View web/src/features/agent-console/ui/agent-detail.tsx

@@ -7,17 +7,19 @@

77 * the activity feed — is a read-only mirror of the gateway session.

88 */

99import { useState } from 'react'

10import { ArrowLeft, Bot, CirclePlay, Pencil, Square, Trash2, Undo2, User } from 'lucide-react'

10import { ArrowLeft, Bot, CirclePlay, Pencil, Send, Square, Trash2, Undo2, User } from 'lucide-react'

1111import { Alert, Button, ErrorPanel, Markdown } from '@/shared/ui'

1212import { useSave } from '@/shared/ui'

13import { useT } from '@/shared/i18n'

13import { useT, type MessageKey } from '@/shared/i18n'

1414import { fmtRelative, fmtTime } from '@/shared/lib'

15import { mergeAgentMessages, useAgentStream, type AgentStreamStatus } from '@/shared/sagent'

1516import { peekCache, writeCache } from '@/shared/storage'

1617import {

1718 cancelTask,

1819 closeTask,

1920 eventLabel,

2021 describeEventContent,

22 getTaskToken,

2123 isActiveState,

2224 priorityOf,

2325 reopenTask,

@@ -27,11 +29,47 @@

2729 stateOf,

2830 WATCH_LABELS,

2931 WATCH_MODES,

32 type AgentMessage,

3033 type AgentRef,

3134 type WatchMode,

3235} from '@/entities/agent'

3336import type { AgentConsoleData } from '../model/use-agent-console-data'

3437

38/** A transcript row, whichever side of the wire it came from. */

39interface Turn {

40 seq: number

41 role: keyof typeof ROLE_LABELS

42 content: string

43 at: string

44}

45

46const ARCHIVE_TURNS = (messages: readonly AgentMessage[]): Turn[] =>

47 messages.map(message => ({

48 seq: message.seq,

49 role: message.role,

50 content: message.content,

51 at: message.created_at,

52 }))

53

54/** Live frames use the gateway's vocabulary: `assistant` is our `agent`. */

55const LIVE_TURNS = (messages: readonly { seq: number; role: string; content: string; created_at_millis: number }[]): Turn[] =>

56 messages.map(message => ({

57 seq: message.seq,

58 role: message.role === 'assistant' ? 'agent' : 'user',

59 content: message.content,

60 at: new Date(message.created_at_millis).toISOString(),

61 }))

62

63const STREAM_LABELS = {

64 idle: 'agent.console.stream.idle',

65 connecting: 'agent.console.stream.connecting',

66 live: 'agent.console.stream.live',

67 reconnecting: 'agent.console.stream.reconnecting',

68 expired: 'agent.console.stream.expired',

69 error: 'agent.console.stream.error',

70 stopped: 'agent.console.stream.stopped',

71} as const satisfies Record<AgentStreamStatus, MessageKey>

72

3573export interface AgentDetailProps {

3674 agentRef: AgentRef

3775 data: AgentConsoleData

@@ -46,6 +84,28 @@

4684 const save = useSave()

4785 const { task, messages, events } = data

4886

87 // The transcript is a read-only mirror in the API host; live frames come

88 // straight from the gateway session so a running task does not need polling.

89 const live = useAgentStream({

90 sessionId: task?.session_id ?? null,

91 fetchToken: async () => {

92 const target = data.task

93 if (!target) throw new Error('agent task is not loaded')

94 const token = await getTaskToken(agentRef, target.id)

95 return {

96 sessionId: token.session_id,

97 token: token.token,

98 baseUrl: token.base_url,

99 expiresAt: token.expires_at,

100 }

101 },

102 })

103 const [draft, setDraft] = useState('')

104 const [sending, setSending] = useState(false)

105 const [sendError, setSendError] = useState<string | null>(null)

106 const transcript = mergeAgentMessages(ARCHIVE_TURNS(messages), LIVE_TURNS(live.messages))

107 const streamAlarm = live.status === 'error' || live.status === 'expired' || live.status === 'reconnecting'

108

49109 // There is no read endpoint for the watch row, so the panel remembers the

50110 // last mode this browser set and shows that; the server stays authoritative.

51111 const watchCacheKey = `agent-watch:${agentRef.space}/${agentRef.collection}/${task?.id ?? ''}`

@@ -95,6 +155,23 @@

95155 await save.run(() => setTaskWatch(agentRef, task.id, mode))

96156 }

97157

158 // `send` goes straight to the gateway session; the API host has no send route

159 // on purpose, so quota and moderation hooks do not exist for follow-ups.

160 const sendDraft = async () => {

161 const text = draft.trim()

162 if (text === '') return

163 setSending(true)

164 setSendError(null)

165 try {

166 await live.send(text)

167 setDraft('')

168 } catch (cause) {

169 setSendError(cause instanceof Error ? cause.message : String(cause))

170 } finally {

171 setSending(false)

172 }

173 }

174

98175 return (

99176 <article className="flex flex-col gap-4 rounded-lg border-2 border-shell-hairline bg-shell-surface p-4">

100177 <header className="flex items-start gap-2">

@@ -220,32 +297,66 @@

220297 </section>

221298

222299 <section className="flex flex-col gap-2">

223 <h3 className="text-[12px] font-semibold uppercase tracking-[0.4px] text-shell-ink-muted">

300 <h3 className="flex items-center gap-2 text-[12px] font-semibold uppercase tracking-[0.4px] text-shell-ink-muted">

224301 {t('agent.console.task.conversation')}

302 <span

303 className={`text-[11px] font-normal normal-case tracking-normal ${streamAlarm ? 'text-error' : ''}`}

304 title={live.error ?? undefined}

305 >

306 {t(STREAM_LABELS[live.status])}

307 </span>

225308 </h3>

226 {messages.length === 0 ? (

309 {transcript.length === 0 ? (

227310 <p className="text-[12.5px] text-shell-ink-muted">{t('agent.console.task.noMessages')}</p>

228311 ) : (

229312 <ol className="flex flex-col gap-2">

230 {messages.map(message => (

231 <li key={message.id} className="flex gap-2">

313 {transcript.map(turn => (

314 <li key={turn.seq} className="flex gap-2">

232315 <span className="mt-0.5 flex h-5 w-5 shrink-0 items-center justify-center rounded-full bg-shell-field text-shell-ink-muted">

233 {message.role === 'user' ? <User className="h-3 w-3" /> : <Bot className="h-3 w-3" />}

316 {turn.role === 'user' ? <User className="h-3 w-3" /> : <Bot className="h-3 w-3" />}

234317 </span>

235318 <div className="flex min-w-0 flex-1 flex-col gap-0.5">

236319 <span className="flex items-center gap-2 text-[11.5px] text-shell-ink-muted">

237 <span className="font-semibold">{ROLE_LABELS[message.role] ?? message.role}</span>

238 <span>{fmtRelative(message.created_at)}</span>

239 <span className="font-mono">{`#${message.seq}`}</span>

320 <span className="font-semibold">{ROLE_LABELS[turn.role] ?? turn.role}</span>

321 <span>{fmtRelative(turn.at)}</span>

322 <span className="font-mono">{`#${turn.seq}`}</span>

240323 </span>

241324 <div className="text-[12.5px] leading-[1.6]">

242 <Markdown>{message.content}</Markdown>

325 <Markdown>{turn.content}</Markdown>

243326 </div>

244327 </div>

245328 </li>

246329 ))}

247330 </ol>

248331 )}

332 {task.session_id ? (

333 <div className="flex flex-col gap-1.5">

334 <div className="flex items-end gap-2">

335 <textarea

336 value={draft}

337 rows={2}

338 placeholder={t('agent.console.task.composerPlaceholder')}

339 onChange={event => setDraft(event.target.value)}

340 onKeyDown={event => {

341 if (event.key === 'Enter' && !event.shiftKey) {

342 event.preventDefault()

343 void sendDraft()

344 }

345 }}

346 className="min-h-16 flex-1 resize-y rounded-md border-2 border-shell-hairline bg-shell-field px-3 py-2 text-[12.5px] text-shell-ink outline-none focus:border-shell-ink"

347 />

348 <Button

349 className="h-8 px-3 text-[12.5px]"

350 disabled={sending || draft.trim() === ''}

351 onClick={() => void sendDraft()}

352 >

353 <Send className="h-3.5 w-3.5" />

354 {t('agent.console.task.send')}

355 </Button>

356 </div>

357 {sendError ? <Alert>{`${t('agent.console.task.sendFailed')}: ${sendError}`}</Alert> : null}

358 </div>

359 ) : null}

249360 </section>

250361

251362 <section className="flex flex-col gap-2">

Modified web/src/shared/i18n/messages/en/agent.ts +8 −0 View web/src/shared/i18n/messages/en/agent.ts

@@ -46,6 +46,14 @@

4646 'agent.console.task.noBody': 'No task brief.',

4747 'agent.console.task.conversation': 'Conversation',

4848 'agent.console.task.noMessages': 'No messages yet.',

49 'agent.console.task.sendFailed': 'Could not send',

50 'agent.console.stream.idle': 'Not connected',

51 'agent.console.stream.connecting': 'Connecting',

52 'agent.console.stream.live': 'Live',

53 'agent.console.stream.reconnecting': 'Reconnecting',

54 'agent.console.stream.expired': 'Token expired',

55 'agent.console.stream.error': 'Disconnected',

56 'agent.console.stream.stopped': 'Closed',

4957 'agent.console.task.activity': 'Activity',

5058 'agent.console.task.noEvents': 'No activity yet.',

5159 'agent.console.task.composerPlaceholder': 'Send the agent a message…',

Modified web/src/shared/i18n/messages/zh-CN/agent.ts +8 −0 View web/src/shared/i18n/messages/zh-CN/agent.ts

@@ -46,6 +46,14 @@

4646 'agent.console.task.noBody': '没有填写任务描述。',

4747 'agent.console.task.conversation': '对话',

4848 'agent.console.task.noMessages': '还没有消息。',

49 'agent.console.task.sendFailed': '发送失败',

50 'agent.console.stream.idle': '未连接',

51 'agent.console.stream.connecting': '连接中',

52 'agent.console.stream.live': '实时',

53 'agent.console.stream.reconnecting': '重连中',

54 'agent.console.stream.expired': '令牌过期',

55 'agent.console.stream.error': '连接失败',

56 'agent.console.stream.stopped': '已断开',

4957 'agent.console.task.activity': '活动记录',

5058 'agent.console.task.noEvents': '还没有活动记录。',

5159 'agent.console.task.composerPlaceholder': '给智能体发一条消息…',

Added web/src/shared/sagent/client.ts +208 −0 View web/src/shared/sagent/client.ts

@@ -0,0 +1,208 @@

1import { AgentSdkError, errorFromResponse } from './errors'

2import { SseParser, toSseEvent, watermarkOf } from './sse'

3import type { MessageAccepted, MessageDto, Page, SessionDto, SseEvent, SseFrame } from './types'

4

5export interface AgentClientOptions {

6 baseUrl: string

7 token: string

8 sessionId: string

9 fetch?: typeof fetch

10}

11

12export interface StreamEventsOptions {

13 onEvent?: (event: SseEvent) => void

14 onMessage?: (message: MessageDto) => void

15 onError?: (error: AgentSdkError) => boolean | void

16 signal?: AbortSignal

17 lastEventId?: number

18 maxRetries?: number

19}

20

21const DEFAULT_MAX_RETRIES = 5

22const BASE_BACKOFF_MS = 500

23const MAX_BACKOFF_MS = 8_000

24

25export class AgentClient {

26 readonly sessionId: string

27

28 private readonly baseUrl: string

29 private readonly token: string

30 private readonly fetchImpl: typeof fetch

31

32 constructor(options: AgentClientOptions) {

33 const base = options.baseUrl.trim().replace(/\/+$/, '')

34 if (base === '') throw new AgentSdkError('baseUrl 不能为空')

35 if (options.token.trim() === '') throw new AgentSdkError('token 不能为空')

36 if (options.sessionId.trim() === '') throw new AgentSdkError('sessionId 不能为空')

37 this.baseUrl = base

38 this.token = options.token

39 this.sessionId = options.sessionId

40 const injected = options.fetch

41 if (injected !== undefined) {

42 this.fetchImpl = injected

43 } else if (typeof globalThis.fetch === 'function') {

44 this.fetchImpl = globalThis.fetch.bind(globalThis)

45 } else {

46 throw new AgentSdkError('当前环境没有 fetch；请在 Node 18+ 或浏览器中使用')

47 }

48 }

49

50 private url(path: string): string {

51 return `${this.baseUrl}${path}`

52 }

53

54 private headers(extra?: Record<string, string>): Record<string, string> {

55 return { authorization: `Bearer ${this.token}`, ...extra }

56 }

57

58 private async requestJson<T>(path: string, init?: RequestInit): Promise<T> {

59 let response: Response

60 try {

61 response = await this.fetchImpl(this.url(path), {

62 ...init,

63 headers: this.headers({

64 'content-type': 'application/json',

65 ...(init?.headers as Record<string, string> | undefined),

66 }),

67 })

68 } catch (cause) {

69 throw new AgentSdkError(`请求失败: ${String(cause)}`, { cause })

70 }

71 if (!response.ok) throw await errorFromResponse(response)

72 const text = await response.text()

73 try {

74 return JSON.parse(text) as T

75 } catch (cause) {

76 throw new AgentSdkError(`响应不是合法 JSON: ${text.slice(0, 200)}`, { cause })

77 }

78 }

79

80 async getSession(): Promise<SessionDto> {

81 return this.requestJson<SessionDto>(`/api/v1/sessions/${this.sessionId}`)

82 }

83

84 async listMessages(options: { limit?: number; cursor?: number } = {}): Promise<Page<MessageDto>> {

85 const query = new URLSearchParams()

86 if (options.limit !== undefined) query.set('limit', String(options.limit))

87 if (options.cursor !== undefined) query.set('cursor', String(options.cursor))

88 const suffix = query.toString() === '' ? '' : `?${query.toString()}`

89 return this.requestJson<Page<MessageDto>>(`/api/v1/sessions/${this.sessionId}/messages${suffix}`)

90 }

91

92 async sendMessage(text: string, idempotencyKey?: string): Promise<MessageAccepted> {

93 if (text.trim() === '') throw new AgentSdkError('text 不能为空')

94 const body: Record<string, unknown> = { text }

95 if (idempotencyKey !== undefined) body.idempotency_key = idempotencyKey

96 return this.requestJson<MessageAccepted>(`/api/v1/sessions/${this.sessionId}/messages`, {

97 method: 'POST',

98 body: JSON.stringify(body),

99 })

100 }

101

102 async streamEvents(options: StreamEventsOptions = {}): Promise<void> {

103 const maxRetries = options.maxRetries ?? DEFAULT_MAX_RETRIES

104 let watermark = options.lastEventId ?? 0

105 let attempt = 0

106

107 while (!options.signal?.aborted) {

108 try {

109 const stopped = await this.streamOnce(

110 options,

111 () => watermark,

112 seq => {

113 watermark = seq

114 },

115 )

116 if (stopped) return

117 } catch (cause) {

118 const error =

119 cause instanceof AgentSdkError

120 ? cause

121 : new AgentSdkError(`SSE 中断: ${String(cause)}`, { cause })

122 // A `false` verdict means "do not retry this stream" — the caller either

123 // refreshed the token (and will reopen it) or gave up.

124 if (options.onError?.(error) === false) return

125 }

126 if (attempt >= maxRetries) return

127 attempt += 1

128 const backoff = Math.min(BASE_BACKOFF_MS * 2 ** (attempt - 1), MAX_BACKOFF_MS)

129 await sleep(backoff, options.signal)

130 }

131 }

132

133 private async streamOnce(

134 options: StreamEventsOptions,

135 watermark: () => number,

136 update: (seq: number) => void,

137 ): Promise<boolean> {

138 const headers = this.headers({ accept: 'text/event-stream' })

139 // The gateway resumes strictly after the watermark, so a reconnect during a

140 // long run replays only what this client has not seen.

141 if (watermark() > 0) headers['last-event-id'] = `m${watermark()}`

142

143 let response: Response

144 try {

145 response = await this.fetchImpl(this.url(`/api/v1/sessions/${this.sessionId}/events`), {

146 headers,

147 signal: options.signal,

148 })

149 } catch (cause) {

150 if (options.signal?.aborted) return true

151 throw new AgentSdkError(`SSE 连接失败: ${String(cause)}`, { cause })

152 }

153 if (!response.ok) throw await errorFromResponse(response)

154 if (response.body === null) throw new AgentSdkError('SSE 响应没有 body')

155

156 const reader = response.body.getReader()

157 const decoder = new TextDecoder()

158 const parser = new SseParser()

159 try {

160 while (true) {

161 const { done, value } = await reader.read()

162 if (done) break

163 parser.push(decoder.decode(value, { stream: true }))

164 for (let frame = parser.popFrame(); frame !== null; frame = parser.popFrame()) {

165 this.dispatch(frame, options, update)

166 }

167 }

168 const tail = parser.flush()

169 if (tail !== null) this.dispatch(tail, options, update)

170 return false

171 } catch (cause) {

172 if (options.signal?.aborted) return true

173 throw new AgentSdkError(`SSE 读取失败: ${String(cause)}`, { cause })

174 } finally {

175 try {

176 await reader.cancel()

177 } catch {

178 // The body is already unusable; nothing to release.

179 }

180 }

181 }

182

183 private dispatch(

184 frame: SseFrame,

185 options: StreamEventsOptions,

186 update: (seq: number) => void,

187 ): void {

188 const seq = watermarkOf(frame)

189 if (seq !== null && seq > 0) update(seq)

190 const event = toSseEvent(frame)

191 options.onEvent?.(event)

192 if (event.type === 'message_added') options.onMessage?.(event.message)

193 }

194}

195

196function sleep(ms: number, signal?: AbortSignal): Promise<void> {

197 return new Promise(resolve => {

198 const timer = setTimeout(resolve, ms)

199 signal?.addEventListener(

200 'abort',

201 () => {

202 clearTimeout(timer)

203 resolve()

204 },

205 { once: true },

206 )

207 })

208}

No newline at end of file

Added web/src/shared/sagent/errors.ts +61 −0 View web/src/shared/sagent/errors.ts

@@ -0,0 +1,61 @@

1import type { ApiErrorBody } from './types'

2

3export class AgentSdkError extends Error {

4 readonly status: number | undefined

5 readonly code: string | undefined

6 readonly retryAfterMs: number | undefined

7 readonly body: string | undefined

8

9 constructor(

10 message: string,

11 options: {

12 status?: number

13 code?: string

14 retryAfterMs?: number

15 body?: string

16 cause?: unknown

17 } = {},

18 ) {

19 super(message, options.cause === undefined ? undefined : { cause: options.cause })

20 this.name = 'AgentSdkError'

21 this.status = options.status

22 this.code = options.code

23 this.retryAfterMs = options.retryAfterMs

24 this.body = options.body

25 }

26

27 get isUnauthorized(): boolean {

28 return this.status === 401

29 }

30

31 get isForbidden(): boolean {

32 return this.status === 403

33 }

34

35 get isNotFound(): boolean {

36 return this.status === 404

37 }

38

39 get isRetryable(): boolean {

40 if (this.status === undefined) return true

41 if (this.status === 429) return true

42 return this.status >= 500

43 }

44}

45

46export async function errorFromResponse(response: Response): Promise<AgentSdkError> {

47 const body = await response.text().catch(() => '')

48 let parsed: ApiErrorBody | undefined

49 try {

50 parsed = JSON.parse(body) as ApiErrorBody

51 } catch {

52 parsed = undefined

53 }

54 const message = parsed?.message ?? body.trim() ?? response.statusText

55 return new AgentSdkError(message, {

56 status: response.status,

57 code: parsed?.code,

58 retryAfterMs: parsed?.retry_after_ms,

59 body,

60 })

61}

No newline at end of file

Added web/src/shared/sagent/index.ts +43 −0 View web/src/shared/sagent/index.ts

@@ -0,0 +1,43 @@

1// Public API of the shared/sagent segment — the browser-direct connection to

2// the sandbox agent gateway at `agent.gitdata.ai`.

3//

4// The gateway is a different origin from this app and the backend deliberately

5// does not proxy it: `apihost` mints a short-lived session JWT

6// (`GET /api/v1/agents/{space}/collections/{c}/tasks/{t}/token`) and the browser

7// talks to `/api/v1/sessions/{session}/**` itself. That is why the client lives

8// in `shared` — nothing about it is specific to the 智能体 console.

9//

10// Typical usage:

11// const stream = useAgentStream({

12// sessionId: task.session_id,

13// fetchToken: async () => {

14// const { data } = await Apis.sagent.get_task_token({ pathParams: {...} })

15// return { sessionId: data.session_id, token: data.token, baseUrl: data.base_url, expiresAt: data.expires_at }

16// },

17// })

18// stream.messages // live frames, seq-ascending

19// stream.status // idle | connecting | live | reconnecting | expired | error | stopped

20// await stream.send('跑一下测试') // POST straight to the gateway

21export { AgentClient, type AgentClientOptions, type StreamEventsOptions } from './client'

22export { AgentSdkError, errorFromResponse } from './errors'

23export { SseParser, toSseEvent, watermarkOf } from './sse'

24export {

25 mergeAgentMessages,

26 useAgentStream,

27 type AgentStreamState,

28 type AgentStreamStatus,

29 type AgentStreamToken,

30 type UseAgentStreamOptions,

31 type UseAgentStreamResult,

32} from './react'

33export type {

34 ApiErrorBody,

35 MessageAccepted,

36 MessageDto,

37 MessageRole,

38 Page,

39 SessionDto,

40 SseEvent,

41 SseEventName,

42 SseFrame,

43} from './types'

No newline at end of file

Added web/src/shared/sagent/react.ts +243 −0 View web/src/shared/sagent/react.ts

@@ -0,0 +1,243 @@

1/**

2 * React binding for the agent gateway's session stream. Browser-direct by

3 * design: the backend only mints the session JWT (`GET …/tasks/{id}/token`),

4 * and this hook opens `GET /api/v1/sessions/{session}/events` itself.

5 *

6 * Three things the raw `AgentClient` does not do:

7 *

8 * - **Token lifetime.** The JWT lives 600s, which is shorter than a long run.

9 * The hook silently re-mints shortly before `expiresAt` and reopens the

10 * stream at the same watermark, so a subscriber never sees the gap.

11 * - **Resume across reconnects.** `Last-Event-ID` is derived from the frames

12 * the caller has already been told about, and it survives `fetchToken`

13 * rotation.

14 * - **State for React.** Messages are deduplicated by `seq` and kept ordered,

15 * because a resumed stream legitimately replays frames.

16 */

17import { useCallback, useEffect, useRef, useState } from 'react'

18import { AgentClient } from './client'

19import { AgentSdkError } from './errors'

20import type { MessageDto, SseEvent } from './types'

21

22/** What the token endpoint hands the browser. */

23export interface AgentStreamToken {

24 sessionId: string

25 token: string

26 baseUrl: string

27 /** ISO-8601; the hook refreshes 60s before this. */

28 expiresAt: string

29}

30

31export type AgentStreamStatus =

32 | 'idle'

33 | 'connecting'

34 | 'live'

35 | 'reconnecting'

36 | 'expired'

37 | 'error'

38 | 'stopped'

39

40export interface UseAgentStreamOptions {

41 /** Gateway session behind the task; `null` keeps the hook idle. */

42 sessionId?: string | null

43 /** Callers can hold the stream closed (e.g. the panel is collapsed). */

44 enabled?: boolean

45 fetchToken: () => Promise<AgentStreamToken>

46 onMessage?: (message: MessageDto) => void

47 onEvent?: (event: SseEvent) => void

48 onRunState?: (state: string) => void

49}

50

51export interface AgentStreamState {

52 status: AgentStreamStatus

53 /** Live frames only, `seq`-ascending. Merge with the archive for display. */

54 messages: MessageDto[]

55 runState?: string

56 error?: string

57 /** Highest `message_added` seq seen — the resume cursor. */

58 watermark: number

59}

60

61export interface UseAgentStreamResult extends AgentStreamState {

62 /** POST the text straight to the gateway (the API proxy has no send route). */

63 send: (text: string) => Promise<void>

64 /** Drop the token and reopen the stream at the current watermark. */

65 reconnect: () => void

66}

67

68const REFRESH_LEAD_MS = 60_000

69

70/**

71 * Keep one session stream open for as long as the caller is mounted. The

72 * returned `messages` are the frames this client actually saw; the caller owns

73 * the historical ones.

74 */

75export function useAgentStream(options: UseAgentStreamOptions): UseAgentStreamResult {

76 const { sessionId, enabled = true, fetchToken } = options

77

78 const [state, setState] = useState<AgentStreamState & { sessionId: string | null }>({

79 status: 'idle',

80 messages: [],

81 watermark: 0,

82 sessionId: null,

83 })

84 const [generation, setGeneration] = useState(0)

85

86 const handlers = useRef(options)

87 const fetchTokenRef = useRef(fetchToken)

88 const tokenRef = useRef<AgentStreamToken | null>(null)

89 const clientRef = useRef<AgentClient | null>(null)

90 const watermarkRef = useRef(0)

91 const watermarkSessionRef = useRef<string | null>(null)

92 const sessionRef = useRef<string | null>(sessionId ?? null)

93

94 // The latest callbacks and inputs live in refs so the stream is opened once per

95 // generation: a re-render (a keystroke in the composer) must not reconnect.

96 useEffect(() => {

97 handlers.current = options

98 fetchTokenRef.current = fetchToken

99 sessionRef.current = sessionId ?? null

100 })

101

102 const ensureClient = useCallback(async (): Promise<AgentClient> => {

103 const session = sessionRef.current

104 if (!session) throw new AgentSdkError('会话尚未创建，暂时无法连接智能体网关')

105 const current = clientRef.current

106 if (current !== null && current.sessionId === session) return current

107 // Minting a JWT is a round trip to the API host, so the panel says

108 // "connecting" from here rather than from the effect that opened the stream.

109 setState(prev => ({ ...prev, status: 'connecting', error: undefined }))

110 const minted = tokenRef.current ?? (await fetchTokenRef.current())

111 tokenRef.current = minted

112 const client = new AgentClient({

113 baseUrl: minted.baseUrl,

114 token: minted.token,

115 sessionId: minted.sessionId || session,

116 })

117 clientRef.current = client

118 return client

119 }, [])

120

121 const reconnect = useCallback(() => {

122 tokenRef.current = null

123 clientRef.current = null

124 setGeneration(n => n + 1)

125 }, [])

126

127 useEffect(() => {

128 if (!enabled || !sessionId) return

129 const controller = new AbortController()

130 let timer: ReturnType<typeof setTimeout> | undefined

131

132 void (async () => {

133 try {

134 const client = await ensureClient()

135 if (controller.signal.aborted) return

136 // The resume cursor belongs to one gateway session: reusing it across

137 // tasks would ask the new session for frames it never had.

138 if (watermarkSessionRef.current !== sessionId) {

139 watermarkRef.current = 0

140 watermarkSessionRef.current = sessionId

141 }

142 setState(prev => ({

143 ...prev,

144 status: 'live',

145 error: undefined,

146 sessionId,

147 watermark: watermarkRef.current,

148 messages: prev.sessionId === sessionId ? prev.messages : [],

149 }))

150

151 const minted = tokenRef.current

152 if (minted !== null) {

153 // Silent rotation: the SDK's own retry loop would replay a dead JWT.

154 const until = Date.parse(minted.expiresAt) - Date.now() - REFRESH_LEAD_MS

155 timer = setTimeout(

156 () => {

157 tokenRef.current = null

158 setGeneration(n => n + 1)

159 },

160 Math.max(1_000, Number.isFinite(until) ? until : 0),

161 )

162 }

163

164 await client.streamEvents({

165 signal: controller.signal,

166 lastEventId: watermarkRef.current,

167 onMessage: message => {

168 watermarkRef.current = Math.max(watermarkRef.current, message.seq)

169 setState(prev => ({

170 ...prev,

171 watermark: watermarkRef.current,

172 messages: prev.messages.some(existing => existing.seq === message.seq)

173 ? prev.messages

174 : [...prev.messages, message].sort((a, b) => a.seq - b.seq),

175 }))

176 handlers.current.onMessage?.(message)

177 },

178 onEvent: event => handlers.current.onEvent?.(event),

179 onError: error => {

180 if (error.isUnauthorized || error.isForbidden) {

181 tokenRef.current = null

182 clientRef.current = null

183 setState(prev => ({ ...prev, status: 'expired', error: error.message }))

184 setGeneration(n => n + 1)

185 return false

186 }

187 setState(prev => ({ ...prev, status: 'reconnecting', error: error.message }))

188 return undefined

189 },

190 })

191 if (!controller.signal.aborted) setState(prev => ({ ...prev, status: 'stopped' }))

192 } catch (cause) {

193 if (controller.signal.aborted) return

194 setState(prev => ({

195 ...prev,

196 status: 'error',

197 error: cause instanceof Error ? cause.message : String(cause),

198 }))

199 }

200 })()

201

202 return () => {

203 controller.abort()

204 if (timer !== undefined) clearTimeout(timer)

205 clientRef.current = null

206 }

207 }, [enabled, sessionId, generation, ensureClient])

208

209 const send = useCallback(

210 async (text: string) => {

211 const client = await ensureClient()

212 await client.sendMessage(text)

213 },

214 [ensureClient],

215 )

216

217 // A stream that is not being held open for this session reports `idle` and no

218 // frames, whatever the previous session left behind.

219 const attached = enabled && Boolean(sessionId) && state.sessionId === sessionId

220 return {

221 ...state,

222 status: attached ? state.status : 'idle',

223 messages: attached ? state.messages : [],

224 watermark: attached ? state.watermark : 0,

225 send,

226 reconnect,

227 }

228}

229

230/**

231 * Merge the archive (rows the backend's subscriber persisted) with the live

232 * frames, newest per `seq`, oldest first.

233 */

234export function mergeAgentMessages<T extends { seq: number }>(

235 archive: readonly T[],

236 live: readonly T[],

237): T[] {

238 if (live.length === 0) return [...archive]

239 const bySeq = new Map<number, T>()

240 for (const message of archive) bySeq.set(message.seq, message)

241 for (const message of live) bySeq.set(message.seq, message)

242 return [...bySeq.values()].sort((a, b) => a.seq - b.seq)

243}

No newline at end of file

Added web/src/shared/sagent/sse.ts +79 −0 View web/src/shared/sagent/sse.ts

@@ -0,0 +1,79 @@

1import type { MessageDto, SseEvent, SseFrame } from './types'

2

3/**

4 * 增量 SSE 解析器：跨 chunk 保持状态。

5 *

6 * 浏览器的 `EventSource` 无法带 `Authorization` 头，因此 SDK 用 `fetch` +

7 * `ReadableStream` 手工解析。这里按空行切帧，`data:` 可跨多行。

8 */

9export class SseParser {

10 private buffer = ''

11

12 push(chunk: string): void {

13 this.buffer += chunk

14 this.buffer = this.buffer.replace(/\r\n/g, '\n')

15 }

16

17 popFrame(): SseFrame | null {

18 const sep = this.buffer.indexOf('\n\n')

19 if (sep < 0) return null

20 const block = this.buffer.slice(0, sep)

21 this.buffer = this.buffer.slice(sep + 2)

22 return parseBlock(block)

23 }

24

25 flush(): SseFrame | null {

26 if (this.buffer.trim() === '') {

27 this.buffer = ''

28 return null

29 }

30 const block = this.buffer

31 this.buffer = ''

32 return parseBlock(block)

33 }

34}

35

36function parseBlock(block: string): SseFrame | null {

37 let event: string | undefined

38 let id: string | undefined

39 const dataLines: string[] = []

40 for (const line of block.split('\n')) {

41 if (line.startsWith('event:')) event = line.slice(6).trim()

42 else if (line.startsWith('id:')) id = line.slice(3).trim()

43 else if (line.startsWith('data:')) {

44 const value = line.slice(5)

45 dataLines.push(value.startsWith(' ') ? value.slice(1) : value)

46 }

47 }

48 if (event === undefined) return null

49 const frame: SseFrame = { event, data: dataLines.join('\n') }

50 if (id !== undefined && id !== '') frame.id = id

51 return frame

52}

53

54export function toSseEvent(frame: SseFrame): SseEvent {

55 try {

56 if (frame.event === 'message_added') {

57 return { type: 'message_added', message: JSON.parse(frame.data) as MessageDto }

58 }

59 if (frame.event === 'run_event') {

60 const p = JSON.parse(frame.data) as { run_id: string; seq: number; event: unknown }

61 return { type: 'run_event', run_id: p.run_id, seq: p.seq, event: p.event }

62 }

63 if (frame.event === 'run_state') {

64 const p = JSON.parse(frame.data) as { run_id: string; state: string }

65 return { type: 'run_state', run_id: p.run_id, state: p.state }

66 }

67 } catch {

68 // 未识别/半截的帧按 `other` 透传，交给上层决定是否忽略。

69 }

70 return { type: 'other', event: frame.event, data: frame.data }

71}

72

73export function watermarkOf(frame: SseFrame): number | null {

74 if (frame.event !== 'message_added') return null

75 const raw = frame.id

76 if (raw === undefined || !raw.startsWith('m')) return null

77 const seq = Number.parseInt(raw.slice(1), 10)

78 return Number.isFinite(seq) ? seq : null

79}

No newline at end of file

Added web/src/shared/sagent/types.ts +67 −0 View web/src/shared/sagent/types.ts

@@ -0,0 +1,67 @@

1/**

2 * Vendored from `agent/sdk/ts-sdk` (`@gitdataai/agent-sdk` 0.1.0, `private: true`,

3 * never published) — the browser-side client for the sandbox agent gateway at

4 * `agent.gitdata.ai`.

5 *

6 * It lives in `shared` because the connection belongs to a *session* on a

7 * different origin than this app, and only `useAgentStream` knows how to hold

8 * one open; nothing about it is specific to the 智能体 console.

9 *

10 * Kept byte-identical to upstream except for extensionless relative imports

11 * (see `client.ts`/`sse.ts`/`errors.ts`), so a `diff -r` against

12 * `C:/code/GitDataAi/agent/sdk/ts-sdk/src` stays readable.

13 */

14export type MessageRole = 'user' | 'assistant'

15

16export interface SessionDto {

17 session_id: string

18 title: string

19 metadata: Record<string, unknown>

20 created_at_millis: number

21 updated_at_millis: number

22 deleted_at_millis?: number

23}

24

25export interface MessageDto {

26 message_id: string

27 session_id: string

28 seq: number

29 role: MessageRole

30 content: string

31 run_id?: string

32 created_at_millis: number

33}

34

35export interface Page<T> {

36 items: T[]

37 next_cursor?: string

38}

39

40export interface MessageAccepted {

41 message_id: string

42 session_id: string

43 seq: number

44 run_id: string

45 state: string

46 events_url: string

47}

48

49export interface ApiErrorBody {

50 code: string

51 message: string

52 retry_after_ms?: number

53}

54

55export type SseEventName = 'message_added' | 'run_event' | 'run_state' | 'run_finished' | string

56

57export interface SseFrame {

58 event: SseEventName

59 id?: string

60 data: string

61}

62

63export type SseEvent =

64 | { type: 'message_added'; message: MessageDto }

65 | { type: 'run_event'; run_id: string; seq: number; event: unknown }

66 | { type: 'run_state'; run_id: string; state: string }

67 | { type: 'other'; event: string; data: string }

No newline at end of file