import { createInteractionRequestApi, type InteractionRequestApi } from "./interaction-requests"; import { env } from "@/shared/config/env"; import type { AccessTokenProvider } from "@/shared/auth/keycloak-auth"; import { agentMessagePageSchema, type AgentMessagePage } from "./agent-messages"; import { AnswerViewStateApiError, readAnswerViewStateResponse, type AnswerViewStateRecord, type AnswerViewStateWrite } from "./answer-view-state"; export type AgentRunStatus = "running" | "completed" | "error" | "aborted" | "waiting_input"; export type AgentSessionPurpose = "ordinary_chat" | "task_execution" | "task_collaboration"; export type AgentEventTriggerContext = { schema_version: "agent-trigger@1"; kind: "event"; trigger_id: string; occurred_at: string; source: "peripheral_sample"; subject_type: "work_order"; subject_id: string; workflow: "water_quality_complaint"; payload: { title: string; summary: string; location?: string | null; priority: "low" | "normal" | "high" | "urgent"; scenario_code?: string | null; origin: "system_data"; }; }; export type AgentPermissionReply = "once" | "always" | "reject"; export type AgentModelOption = { id: string; label: string; description: string; icon: "bolt" | "sparkle"; }; export type AgentModelsResponse = { defaultModel: string; models: AgentModelOption[]; }; export type AgentInputDelivery = "steer" | "queue"; export type AgentResponseInput = { id: string; text: string; delivery: AgentInputDelivery; status: "pending" | "delivered" | "cancelled" }; export class AgentResponseFinishedError extends Error {} export class AgentSessionStreamError extends Error { constructor(message: string, readonly status: number) { super(message); } } export type AgentTopic = { id: string; name: string; createdAt: string; updatedAt: string }; export type AgentChatSession = { id?: string; session_id: string; title?: string; topic_id?: string | null; status?: string; run_status?: AgentRunStatus; is_streaming?: boolean; messages?: unknown[]; created_at?: number | string; updated_at?: number | string; is_title_manually_edited?: boolean; parent_session_id?: string; created?: boolean; event_duplicate?: boolean; trigger_id?: string; purpose?: AgentSessionPurpose; task_id?: string; }; export type AgentChatSessionSummary = { trigger?: { subjectId: string; workflow: "water_quality_complaint"; occurredAt: string }; id: string; title: string; topicId?: string | null; topicName?: string; searchSnippet?: string; createdAt: number; updatedAt: number; status?: string; parentSessionId?: string; isStreaming?: boolean; runStatus?: AgentRunStatus; needsInput?: boolean; purpose: AgentSessionPurpose; taskId?: string; }; export type AgentRecentActivity = { id: string; kind: "alert" | "task" | "analysis" | "session"; title: string; occurred_at: string; summary?: string; status?: string; origin: "backend" | "agent" | "demo"; }; export type AgentPromptSuggestion = { question: string; activityId: string | null; }; export type AgentMessagePagination = { hasMore: boolean; beforeCursor: string | null }; export type AgentSessionPage = { sessions: AgentChatSessionSummary[]; hasMore: boolean; nextCursor: string | null }; export type AgentHistoryMessagePage = { messages: unknown[]; messagePage: AgentMessagePagination }; export type AgentLoadedChatSession = AgentChatSessionSummary & { sessionId: string; isTitleManuallyEdited: boolean; messages: unknown[]; messagePage?: AgentMessagePagination; }; export type AgentSessionStreamDataEventType = | "execution_plan" | "final_answer" | "token" | "progress" | "todo_update" | "permission_request" | "permission_response" | "run_status" | "interaction_request" | "question_request" | "question_response" | "tool_call" | "frontend_action" | "frontend_action_result" | "answer_ui" | "ui_envelope" | "session_title" | "done" | "aborted" | "error" | "auth_required" | "credential_refresh_required" | "credential_refreshed" | "credential_refresh_failed" | "agent_message" | "user_input"; export type AgentSessionStreamEvent = | { type: "state"; executionPlans?: unknown[]; messagePage?: AgentMessagePagination; sessionId: string; messages: unknown[]; isStreaming: boolean; runStatus?: AgentRunStatus; } | { type: AgentSessionStreamDataEventType; sessionId?: string; data: Record; }; export type AgentApiClient = InteractionRequestApi & { getHealth: (signal: AbortSignal) => Promise; getAnswerViewState: (sessionId: string, messageId: string, blockId: string) => Promise; putAnswerViewState: (sessionId: string, state: AnswerViewStateWrite) => Promise; downloadFile: (path: string) => Promise; submitResponseInput: (sessionId: string, text: string, delivery: AgentInputDelivery, id: string) => Promise; listResponseInputs: (sessionId: string) => Promise; updateResponseInput: (sessionId: string, id: string, delivery: AgentInputDelivery) => Promise; cancelResponseInput: (sessionId: string, id: string) => Promise; listAgentMessages: (sessionId: string, options?: { cursor?: string; signal?: AbortSignal }) => Promise; streamAgentMessages: (sessionId: string, options: { signal?: AbortSignal; onMessage: () => void }) => Promise; generatePromptSuggestions: (activities: AgentRecentActivity[], model: string, signal?: AbortSignal) => Promise; createSession: (options?: { triggerContext?: AgentEventTriggerContext; topicId?: string }) => Promise; listSessions: (options?: { purpose?: AgentSessionPurpose; taskId?: string }) => Promise; listSessionPage: (options?: { q?: string; cursor?: string; limit?: number; purpose?: AgentSessionPurpose; topicId?: string; includeCollaboration?: boolean; signal?: AbortSignal }) => Promise; listTopics: () => Promise; createTopic: (name: string) => Promise; renameTopic: (id: string, name: string) => Promise; deleteTopic: (id: string) => Promise; moveSessionToTopic: (sessionId: string, topicId: string | null) => Promise; loadOlderMessages: (sessionId: string, before: string, signal?: AbortSignal) => Promise; createTaskCollaborationSession: (taskId: string) => Promise; createTaskExecutionDebugSession: (taskId: string, runId: string, sessionId: string) => Promise; loadTaskExecutionSession: (taskId: string, runId: string, sessionId: string, signal?: AbortSignal) => Promise; loadSession: (sessionId: string, options?: { messageLimit?: number; signal?: AbortSignal }) => Promise; streamSession: ( sessionId: string, options: { signal?: AbortSignal; messageLimit?: number; onEvent: (event: AgentSessionStreamEvent) => void; } ) => Promise; updateSessionTitle: ( sessionId: string, title: string, options?: { isTitleManuallyEdited?: boolean } ) => Promise; deleteSession: (sessionId: string) => Promise; getModels: () => Promise; getUiRegistry: () => Promise; getFrontendActionRegistry: () => Promise; submitFrontendActionResult: (sessionId: string, actionId: string, result: unknown) => Promise; resolveRenderRef: (renderRef: string, sessionId: string) => Promise; replyPermission: ( requestId: string, options: { sessionId: string; reply: AgentPermissionReply; message?: string } ) => Promise; getQuestionStatus: (sessionId: string, requestId: string) => Promise<{ status: "pending" | "answered" | "rejected" | "invalidated" }>; replyQuestion: ( requestId: string, options: { sessionId: string; answers: string[][] } ) => Promise; rejectQuestion: (requestId: string, options: { sessionId: string }) => Promise; refreshCredential: (sessionId: string, requestId: string) => Promise; abort: (sessionId: string) => Promise; }; export type AgentApiClientOptions = { getAccessToken?: AccessTokenProvider; getProjectId?: () => Promise; }; const API_PATH = "/api/v1"; export function createAgentApiClient( baseUrl: string = env.TJWATER_AGENT_URL, options: AgentApiClientOptions = {} ): AgentApiClient { const normalizedBaseUrl = baseUrl.replace(/\/$/, ""); const request = (path: string, init?: RequestInit) => fetchAgent( normalizedBaseUrl, path, init, options.getAccessToken, options.getProjectId ); const requestJson = async (path: string, init?: RequestInit) => { const response = await request(path, init); const text = await response.text(); const data = text ? JSON.parse(text) : null; if (!response.ok) { throw new Error(getResponseErrorMessage(data, response.status)); } return data as T; }; let answerStateActor: string | undefined; const checkedAnswerStateToken = options.getAccessToken ? async () => { const token = await options.getAccessToken!(); const actor = answerStateTokenActor(token); if (answerStateActor !== undefined && actor !== answerStateActor) { throw new AnswerViewStateApiError("登录身份已变化,请刷新后继续。", 403, "ANSWER_VIEW_IDENTITY_CHANGED"); } answerStateActor = actor; return token; } : undefined; const answerStateRequest = async (path: string, init?: RequestInit) => { const response = await fetchAgent(normalizedBaseUrl, path, init, checkedAnswerStateToken, options.getProjectId); // An identity change while awaiting the network also invalidates the result. await checkedAnswerStateToken?.(); return response; }; return { ...createInteractionRequestApi(answerStateRequest), async getHealth(signal) { const response = await fetchAgent(normalizedBaseUrl, "/health", { signal, cache: "no-store" }, undefined, undefined, ""); if (!response.ok) throw new Error("Agent 服务暂不可用"); const data: unknown = await response.json(); if (!isRecord(data) || data.ok !== true || data.ready !== true) throw new Error("Agent 服务尚未就绪"); }, async getAnswerViewState(sessionId, messageId, blockId) { const query = new URLSearchParams({ message_id: messageId, block_id: blockId }); return readAnswerViewStateResponse(await answerStateRequest(`/session/${encodeURIComponent(sessionId)}/ui-state?${query}`)); }, async putAnswerViewState(sessionId, state) { return readAnswerViewStateResponse(await answerStateRequest(`/session/${encodeURIComponent(sessionId)}/ui-state`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify(state) })); }, async downloadFile(path) { const match = /^\/api\/v1(\/session\/[^/?#]+\/files\/mcp-[a-f0-9-]{36}\.(?:zip|xlsx))$/.exec(path); if (!match) throw new Error("无效的会话文件地址"); const response = await request(match[1], { redirect: "error" }); if (!response.ok) throw new Error("文件不可用或当前账号无权下载"); const blob = await response.blob(); const url = URL.createObjectURL(blob); const anchor = document.createElement("a"); anchor.href = url; anchor.download = path.split("/").at(-1) ?? "export.zip"; anchor.click(); setTimeout(() => URL.revokeObjectURL(url), 1000); }, async submitResponseInput(sessionId, text, delivery, id) { const response = await request("/responses", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ input: text, conversation: sessionId, stream: false, tjwater: { delivery, input_id: id } }) }); const data = await response.json(); if (response.status === 409 && data?.code === "RESPONSE_FINISHED") throw new AgentResponseFinishedError("当前回复已结束"); if (!response.ok) throw new Error(getResponseErrorMessage(data, response.status)); return data as AgentResponseInput; }, async listResponseInputs(sessionId) { const result = await requestJson<{ items: AgentResponseInput[] }>(`/session/${encodeURIComponent(sessionId)}/inputs`); return result.items; }, async updateResponseInput(sessionId, id, delivery) { return requestJson(`/session/${encodeURIComponent(sessionId)}/inputs/${encodeURIComponent(id)}`, { method: "PATCH", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ delivery }) }); }, async cancelResponseInput(sessionId, id) { return requestJson(`/session/${encodeURIComponent(sessionId)}/inputs/${encodeURIComponent(id)}`, { method: "DELETE" }); }, async generatePromptSuggestions(activities, model, signal) { const result = await requestJson("/suggestions", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ activities, model }), signal, }); if (!isRecord(result) || !Array.isArray(result.suggestions) || result.suggestions.length < 1 || result.suggestions.length > 3) { throw new Error("Invalid Agent suggestions"); } const ids = new Set(activities.map((item) => item.id)); const questions = new Set(); return result.suggestions.map((item): AgentPromptSuggestion => { if (!isRecord(item) || typeof item.question !== "string" || !item.question.trim() || item.question.length > 180 || (item.activity_id !== null && (typeof item.activity_id !== "string" || !ids.has(item.activity_id)))) { throw new Error("Invalid Agent suggestion"); } const question = item.question.trim(); if (questions.has(question)) throw new Error("Duplicate Agent suggestion"); questions.add(question); return { question, activityId: item.activity_id as string | null }; }); }, async createSession(options) { return requestJson("/session", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ trigger_context: options?.triggerContext, topic_id: options?.topicId }) }); }, async listSessions(listOptions) { const params = new URLSearchParams(); if (listOptions?.purpose) params.set("purpose", listOptions.purpose); if (listOptions?.taskId) params.set("task_id", listOptions.taskId); const payload = await requestJson<{ sessions?: unknown[] }>( `/sessions${params.size ? `?${params.toString()}` : ""}` ); return (payload.sessions ?? []).map(toSessionSummary).filter(isPresent).sort(compareSessionSummaries); }, async listSessionPage(options) { const params = new URLSearchParams({ purpose: options?.purpose ?? "ordinary_chat", limit: String(options?.limit ?? 30) }); if (options?.q) params.set("q", options.q); if (options?.topicId) params.set("topic_id", options.topicId); if (options?.includeCollaboration) params.set("include_collaboration", "true"); if (options?.cursor) params.set("cursor", options.cursor); const payload = await requestJson<{ sessions?: unknown[]; has_more?: boolean; next_cursor?: string | null }>( `/sessions?${params}`, { signal: options?.signal }); return { sessions: (payload.sessions ?? []).map(toSessionSummary).filter(isPresent), hasMore: payload.has_more === true, nextCursor: typeof payload.next_cursor === "string" ? payload.next_cursor : null }; }, async listTopics() { return (await requestJson<{ topics: AgentTopic[] }>("/topics")).topics; }, async createTopic(name) { return requestJson("/topics", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ name }) }); }, async renameTopic(id, name) { return requestJson(`/topics/${encodeURIComponent(id)}`, { method: "PATCH", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ name }) }); }, async deleteTopic(id) { await requestJson(`/topics/${encodeURIComponent(id)}`, { method: "DELETE" }); }, async moveSessionToTopic(sessionId, topicId) { await requestJson(`/session/${encodeURIComponent(sessionId)}/topic`, { method: "PATCH", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ topic_id: topicId }), }); }, async loadOlderMessages(sessionId, before, signal) { const query = new URLSearchParams({ limit: "40", before }); const payload = await requestJson<{ messages?: unknown[]; message_page?: unknown }>( `/session/${encodeURIComponent(sessionId)}/messages?${query}`, { signal }); return { messages: payload.messages ?? [], messagePage: readMessagePagination(payload.message_page) ?? { hasMore: false, beforeCursor: null } }; }, async createTaskCollaborationSession(taskId) { return requestJson( `/task/${encodeURIComponent(taskId)}/collaboration/session`, { method: "POST" } ); }, async createTaskExecutionDebugSession(taskId, runId, sessionId) { return requestJson( `/task/${encodeURIComponent(taskId)}/execution-session/debug`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ run_id: runId, session_id: sessionId }) } ); }, async loadTaskExecutionSession(taskId, runId, sessionId, signal) { const query = new URLSearchParams({ run_id: runId, session_id: sessionId }); const response = await request( `/task/${encodeURIComponent(taskId)}/execution-session?${query}`, { signal } ); const text = await response.text(); const data = text ? JSON.parse(text) : null; if (response.status === 404) return null; if (!response.ok) throw new Error(getResponseErrorMessage(data, response.status)); return toLoadedSession(data); }, async getFrontendActionRegistry() { return requestJson("/frontend-action-registry"); }, async submitFrontendActionResult(sessionId, actionId, result) { await requestJson(`/frontend-actions/${encodeURIComponent(actionId)}/result`, { method: "POST", headers: { "Content-Type": "application/json", "x-agent-session-id": sessionId }, body: JSON.stringify(result) }); }, async loadSession(sessionId, options) { const query = options?.messageLimit ? `?message_limit=${options.messageLimit}` : ""; const response = await request(`/session/${encodeURIComponent(sessionId)}${query}`, { signal: options?.signal }); const text = await response.text(); const data = text ? JSON.parse(text) : null; if (response.status === 404) { return null; } if (!response.ok) { throw new Error(getResponseErrorMessage(data, response.status)); } return toLoadedSession(data); }, async streamSession(sessionId, options) { const response = await request( `/session/${encodeURIComponent(sessionId)}/stream${options.messageLimit ? `?message_limit=${options.messageLimit}` : ""}`, { signal: options.signal } ); if (!response.ok) { const text = await response.text(); const data = text ? JSON.parse(text) : null; throw new AgentSessionStreamError(getResponseErrorMessage(data, response.status), response.status); } await readSessionSseStream(response, options.onEvent); }, async listAgentMessages(sessionId, options) { const query = new URLSearchParams({ limit: "50" }); if (options?.cursor) query.set("cursor", options.cursor); return agentMessagePageSchema.parse(await requestJson( `/session/${encodeURIComponent(sessionId)}/agent-messages?${query}`, { signal: options?.signal } )); }, async streamAgentMessages(sessionId, options) { const response = await request(`/session/${encodeURIComponent(sessionId)}/agent-messages/stream`, { signal: options.signal }); if (!response.ok) throw new Error(getResponseErrorMessage(await response.json(), response.status)); await readSessionSseStream(response, (event) => { if (event.type === "agent_message" && event.sessionId === sessionId) options.onMessage(); }); }, async updateSessionTitle(sessionId, title, options) { const normalizedTitle = title.trim(); if (!normalizedTitle) { return; } await requestJson( `/session/${encodeURIComponent(sessionId)}/title`, { method: "PATCH", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ title: normalizedTitle, is_title_manually_edited: options?.isTitleManuallyEdited }) } ); }, async deleteSession(sessionId) { await requestJson( `/session/${encodeURIComponent(sessionId)}`, { method: "DELETE" } ); }, async getModels() { const payload = await requestJson("/models"); return toModelsResponse(payload); }, async getUiRegistry() { return requestJson("/ui-registry"); }, async resolveRenderRef(renderRef, sessionId) { const params = new URLSearchParams({ session_id: sessionId }); return requestJson( `/render-ref/${encodeURIComponent(renderRef)}?${params.toString()}` ); }, async replyPermission(requestId, options) { await requestJson(`/permission/${encodeURIComponent(requestId)}/reply`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ session_id: options.sessionId, reply: options.reply, message: options.message }) }); }, async getQuestionStatus(sessionId, requestId) { const value = await requestJson<{ status: "pending" | "answered" | "rejected" | "invalidated" }>(`/session/${encodeURIComponent(sessionId)}/questions/${encodeURIComponent(requestId)}`, { cache: "no-store" }); if (!["pending", "answered", "rejected", "invalidated"].includes(value.status)) throw new Error("问题状态核对失败"); return value; }, async replyQuestion(requestId, options) { await requestJson(`/question/${encodeURIComponent(requestId)}/reply`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ session_id: options.sessionId, answers: options.answers }) }); }, async rejectQuestion(requestId, options) { await requestJson(`/question/${encodeURIComponent(requestId)}/reject`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ session_id: options.sessionId }) }); }, async refreshCredential(sessionId, requestId) { await options.getAccessToken?.(true); await requestJson(`/session/${encodeURIComponent(sessionId)}/credentials`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ request_id: requestId }) }); }, async abort(sessionId) { await requestJson("/abort", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ session_id: sessionId }) }); } }; } async function fetchAgent( baseUrl: string, path: string, init?: RequestInit, getAccessToken?: AccessTokenProvider, getProjectId?: () => Promise, prefix = API_PATH ) { const accessToken = await getAccessToken?.(); const projectId = await getProjectId?.(); const hasContextHeaders = Boolean(accessToken || projectId); const headers = hasContextHeaders ? new Headers(init?.headers) : undefined; if (accessToken) headers?.set("Authorization", `Bearer ${accessToken}`); if (projectId) headers?.set("X-Project-Id", projectId); const requestInit = hasContextHeaders ? { ...init, headers } : init; return fetch(`${baseUrl}${prefix}${path}`, requestInit); } function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null; } function isPresent(value: T | null | undefined): value is T { return value !== null && value !== undefined; } function toModelsResponse(value: unknown): AgentModelsResponse { if (!isRecord(value)) { return { defaultModel: "", models: [] }; } const models = Array.isArray(value.models) ? value.models.map(toModelOption).filter(isPresent) : []; const defaultModel = typeof value.default_model === "string" ? value.default_model : models[0]?.id ?? ""; return { defaultModel, models }; } function toModelOption(value: unknown): AgentModelOption | null { if (!isRecord(value) || typeof value.id !== "string") { return null; } const label = typeof value.label === "string" && value.label.trim() ? value.label : value.id; const description = typeof value.description === "string" && value.description.trim() ? value.description : label; const icon = value.icon === "bolt" || value.icon === "sparkle" ? value.icon : "sparkle"; return { id: value.id, label, description, icon }; } function getResponseErrorMessage(data: unknown, status: number) { return isRecord(data) && typeof data.message === "string" ? data.message : `Agent API failed with HTTP ${status}`; } function toMillis(value: unknown) { if (typeof value === "number" && Number.isFinite(value)) { return value; } if (typeof value === "string" && value.trim()) { const parsed = new Date(value).getTime(); return Number.isFinite(parsed) ? parsed : Date.now(); } return Date.now(); } function normalizeTitle(value: unknown) { return typeof value === "string" && value.trim() ? value.trim() : "新对话"; } function readRunStatus(value: unknown): AgentRunStatus | undefined { return value === "waiting_input" || value === "running" || value === "completed" || value === "error" || value === "aborted" ? value : undefined; } function readSessionPurpose(value: unknown): AgentSessionPurpose { return value === "task_execution" || value === "task_collaboration" ? value : "ordinary_chat"; } function toSessionSummary(value: unknown): AgentChatSessionSummary | null { if (!isRecord(value)) { return null; } const id = typeof value.id === "string" ? value.id : typeof value.session_id === "string" ? value.session_id : ""; if (!id) { return null; } return { id, title: normalizeTitle(value.title), topicId: typeof value.topic_id === "string" ? value.topic_id : null, topicName: typeof value.topic_name === "string" ? value.topic_name : undefined, searchSnippet: typeof value.search_snippet === "string" ? value.search_snippet : undefined, createdAt: toMillis(value.created_at), updatedAt: toMillis(value.updated_at), status: typeof value.status === "string" ? value.status : undefined, parentSessionId: typeof value.parent_session_id === "string" ? value.parent_session_id : undefined, isStreaming: typeof value.is_streaming === "boolean" ? value.is_streaming : undefined, runStatus: readRunStatus(value.run_status), purpose: readSessionPurpose(value.purpose), taskId: typeof value.task_id === "string" ? value.task_id : undefined, trigger: isRecord(value.trigger) && value.trigger.workflow === "water_quality_complaint" && typeof value.trigger.subject_id === "string" && typeof value.trigger.occurred_at === "string" && Number.isFinite(Date.parse(value.trigger.occurred_at)) ? { subjectId: value.trigger.subject_id, workflow: "water_quality_complaint", occurredAt: value.trigger.occurred_at } : undefined }; } function toLoadedSession(value: unknown): AgentLoadedChatSession { const summary = toSessionSummary(value); if (!summary || !isRecord(value)) { throw new Error("Invalid agent session payload"); } return { ...summary, sessionId: typeof value.session_id === "string" ? value.session_id : summary.id, isTitleManuallyEdited: typeof value.is_title_manually_edited === "boolean" ? value.is_title_manually_edited : false, messages: Array.isArray(value.messages) ? value.messages : [], messagePage: readMessagePagination(value.message_page) }; } async function readSessionSseStream( response: Response, onEvent: (event: AgentSessionStreamEvent) => void ) { if (!response.body) { return; } const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ""; while (true) { const { done, value } = await reader.read(); buffer += decoder.decode(value, { stream: !done }); let delimiterIndex = buffer.indexOf("\n\n"); while (delimiterIndex !== -1) { const rawEvent = buffer.slice(0, delimiterIndex); buffer = buffer.slice(delimiterIndex + 2); emitSessionSseEvent(rawEvent, onEvent); delimiterIndex = buffer.indexOf("\n\n"); } if (done) { break; } } if (buffer.trim()) { emitSessionSseEvent(buffer, onEvent); } } function emitSessionSseEvent( rawEvent: string, onEvent: (event: AgentSessionStreamEvent) => void ) { const lines = rawEvent.split(/\r?\n/); const eventType = lines .find((line) => line.startsWith("event:")) ?.slice("event:".length) .trim(); const dataText = lines .filter((line) => line.startsWith("data:")) .map((line) => line.slice("data:".length).trimStart()) .join("\n"); if (!eventType || !dataText) { return; } const data = JSON.parse(dataText) as unknown; if (!isRecord(data)) { return; } if (eventType === "state") { const sessionId = typeof data.session_id === "string" ? data.session_id : ""; onEvent({ type: "state", executionPlans: Array.isArray(data.execution_plans) ? data.execution_plans : undefined, messagePage: readMessagePagination(data.message_page), sessionId, messages: Array.isArray(data.messages) ? data.messages : [], isStreaming: data.is_streaming === true, runStatus: readRunStatus(data.run_status) }); return; } if (!isSessionStreamDataEventType(eventType)) { return; } onEvent({ type: eventType, sessionId: typeof data.session_id === "string" ? data.session_id : undefined, data }); } function isSessionStreamDataEventType(value: string): value is AgentSessionStreamDataEventType { return ( value === "final_answer" || value === "token" || value === "progress" || value === "todo_update" || value === "execution_plan" || value === "permission_request" || value === "permission_response" || value === "run_status" || value === "interaction_request" || value === "question_request" || value === "question_response" || value === "tool_call" || value === "frontend_action" || value === "frontend_action_result" || value === "answer_ui" || value === "ui_envelope" || value === "session_title" || value === "done" || value === "aborted" || value === "error" || value === "credential_refresh_required" || value === "credential_refreshed" || value === "credential_refresh_failed" || value === "auth_required" || value === "agent_message" || value === "user_input" ); } function compareSessionSummaries(left: AgentChatSessionSummary, right: AgentChatSessionSummary) { const createdAtDiff = right.createdAt - left.createdAt; if (createdAtDiff !== 0) { return createdAtDiff; } const updatedAtDiff = right.updatedAt - left.updatedAt; if (updatedAtDiff !== 0) { return updatedAtDiff; } return right.id.localeCompare(left.id); } function readMessagePagination(value: unknown): AgentMessagePagination | undefined { if (!isRecord(value)) return undefined; return { hasMore: value.has_more === true, beforeCursor: typeof value.before_cursor === "string" ? value.before_cursor : null }; } function answerStateTokenActor(token: string | null): string { if (!token) return "development-context"; try { const payload: unknown = JSON.parse(atob(token.split(".")[1].replaceAll("-", "+").replaceAll("_", "/"))); if (isRecord(payload) && typeof payload.sub === "string") return JSON.stringify([payload.iss, payload.sub]); } catch { /* Non-JWT development credentials use the enclosing workbench owner. */ } return "development-context"; }