import { toSessionScope } from "../sessions/session-scope.js"; import type { TaskCalculationWaiter } from "../mcp/task-calculation-waiter.js"; import { createResponsesProjection, responseRequestSchema, responsePrompt } from "../chat/responses-protocol.js"; import { ResponseInputTracker, WaitingResponseInputQueue } from "../chat/response-inputs.js"; import { reconcileInteractionDelivery } from "../sessions/interaction-delivery-reconciliation.js"; import type { InteractionRequestStore } from "../sessions/interaction-request-store.js"; import { renderInteractionAnswer, type InteractionRequest } from "../chat/interaction-request-contract.js"; import { createInteractionReplyMessage } from "../chat/interaction-message-presentation.js"; import { createChatRunAccumulator, publishRunEvent } from "./chat-run-accumulator.js"; import type { AgentMessageCoordinator } from "../chat/agent-message-coordinator.js"; import { credentialRefreshCoordinator } from "../product/credential-refresh.js"; import { type Router } from "express"; import { randomUUID } from "node:crypto"; import { type UIMessage, type UIMessageChunk, } from "ai"; import { z } from "zod"; import { createExecutionPlanSnapshot, executionPlanNeedsTaskPublish, readExecutionPlans } from "./executionPlan.js"; import { executionPlanSchema } from "./executionPlan.js"; import { isSupportedModel } from "../chat/models.js"; import { config } from "../config.js"; import { type ChatSessionBridge } from "../chat/sessionBridge.js"; import { getLocalAgentContext, getTrustedProductCredential, } from "../context/localContext.js"; import { type FrontendActionCoordinator } from "../frontendAction/coordinator.js"; import { type LearningOrchestrator } from "../learning/orchestrator.js"; import { logger } from "../logger.js"; import { type MemoryStore } from "../memory/store.js"; import { McpPreparationError, type OpencodeRuntimeAdapter } from "../runtime/opencode.js"; import { getRuntimeSessionContext, setRuntimeSessionContext, } from "../runtime/sessionContext.js"; import { type SessionMetadataStore, type SessionRecord, readPurpose, } from "../sessions/metadataStore.js"; import { bindOperationalTaskCredential, clearOperationalTaskLease, operationalTaskClient, releaseOperationalTaskCredential, createScadaPresentationContinuation, taskRunInterruptionMessage, } from "../product/operational-task-client.js"; import { type SessionTranscriptStore } from "../sessions/transcriptStore.js"; import { type SessionUiStateStore } from "../sessions/uiStateStore.js"; import { toActorKey, toProjectKey } from "../utils/fileStore.js"; import { buildPromptWithLearningContext, extractLatestFrontendTurn, generateSessionTitle, resolveRequestedStreamSessionId, shouldGenerateSessionTitle, shouldRestoreConversationForRuntime, } from "./chatSession.js"; import { collectTextContent, streamPromptResponse, } from "./chatStream.js"; import { type ChatRunState } from "./chatRunState.js"; import { type ActiveRun, createInitialStreamingMessages, isObjectRecord, updateLastAssistantMessage, } from "./chatUiState.js"; import { agentTriggerContextSchema, buildAgentTriggerPrompt, } from "./agentTrigger.js"; const payloadSchema = z.object({ id: z.string().max(128).optional(), messages: z.array(z.unknown()).max(200).optional(), message: z.string().min(1).max(10000).optional(), session_id: z.string().max(128).optional(), model: z.string().refine(isSupportedModel, { message: "unsupported model", }).optional(), approval_mode: z.enum(["request", "always"]).optional().default("request"), capabilities: z.array(z.string().max(64)).max(20).optional().default([]), trigger_context: agentTriggerContextSchema.optional(), task_context: z.object({ report_revision: z.number().int().positive().optional(), evidence_id: z.string().min(1).max(128).optional(), gap_id: z.string().min(1).max(128).optional(), }).strict().optional(), }); const runtimeHasConversation = async ( runtime: OpencodeRuntimeAdapter, sessionId: string, ) => { const messages = await runtime.messages(sessionId, 1); return messages.some( (message) => message.info.role === "user" || message.info.role === "assistant", ); }; const isUiMessage = (value: unknown): value is UIMessage => isObjectRecord(value) && typeof value.id === "string" && (value.role === "system" || value.role === "user" || value.role === "assistant") && Array.isArray(value.parts); const readUiMessageText = (message: UIMessage | undefined) => { const finalPart = message?.parts.slice().reverse().find((part) => isObjectRecord(part) && part.type === "data-final_answer" && isObjectRecord(part.data) && typeof part.data.content === "string"); if (finalPart && "data" in finalPart && isObjectRecord(finalPart.data)) return String(finalPart.data.content).trim(); return message?.parts .flatMap((part) => isObjectRecord(part) && part.type === "text" && typeof part.text === "string" ? [part.text] : [], ) .join("") .trim() ?? ""; }; const readLatestUserText = (messages: UIMessage[]) => { for (let index = messages.length - 1; index >= 0; index -= 1) { const message = messages[index]; if (message.role !== "user") { continue; } const text = readUiMessageText(message); if (text) { return text; } } return ""; }; const toUiMessages = (messages: unknown[] | undefined) => Array.isArray(messages) ? messages.filter(isUiMessage) : []; export const toCollaborationMessages = (messages: UIMessage[]) => messages.flatMap((message) => { if (message.role !== "user" && message.role !== "assistant") return []; const content = readUiMessageText(message); const reference = message.parts.filter((part) => part.type === "data-task_context" || part.type === "data-interaction_reply"); const rich = message.parts.some((part) => part.type === "data-answer_ui" || part.type === "data-final_answer"); const parts = rich ? message.parts.filter((part) => ["text", "data-answer_ui", "data-final_answer", "data-task_context", "data-interaction_reply"].includes(part.type)) : [{ type: "text" as const, text: content }, ...reference]; return content || rich ? [{ ...message, parts } as UIMessage] : []; }); const toLegacyMessages = (messages: UIMessage[]) => messages.flatMap((message) => { const content = readUiMessageText(message); const rich = message.parts.some((part) => part.type === "data-answer_ui" || part.type === "data-final_answer"); if ((!content && !rich) || (message.role !== "user" && message.role !== "assistant")) { return []; } return [ { id: message.id, role: message.role, content, ...(rich ? { parts: message.parts } : {}), }, ]; }); const createFallbackUserUiMessage = (content: string, id?: string): UIMessage => ({ id: id ?? `user-${randomUUID()}`, role: "user", parts: [{ type: "text", text: content }], }); export const buildFailedTaskExecutionPlan = (history: Record[]) => { const previous = history.at(-1); const parsed = previous ? executionPlanSchema.safeParse(previous) : null; if (!parsed?.success) { return { title: "工况任务执行失败", reason: "Agent 模型或工具执行失败，已记录失败并停止本次执行。", nodes: [{ id: "AGENT_RUN_FAILURE", title: "Agent 执行失败", status: "failed" as const, owner: "agent" as const, detail: "模型或工具执行失败，等待人工检查或重新发起。", evidence: [], action_refs: [], }], edges: [], }; } let changed = false; const nodes = parsed.data.nodes.map((node) => { if (node.status === "running") { changed = true; return { ...node, status: "failed" as const, detail: "模型或工具执行失败，当前节点未完成，等待人工检查或重新发起。", }; } if (node.status === "pending" || node.status === "waiting") { return { ...node, status: "cancelled" as const, detail: `${node.detail ? `${node.detail}；` : ""}本次 Agent 执行失败，后续工作未启动。`, }; } return node; }); if (!changed && !nodes.some((node) => node.status === "failed")) { const known = new Set(nodes.map((node) => node.id)); let index = 1; let id = "AGENT_RUN_FAILURE"; while (known.has(id)) id = `AGENT_RUN_FAILURE_${index++}`; nodes.push({ id, title: "Agent 执行失败", status: "failed", owner: "agent", detail: "模型或工具执行失败，等待人工检查或重新发起。", evidence: [], action_refs: [], }); } return { ...parsed.data, reason: "Agent 模型或工具执行失败，已记录失败并停止本次执行。", nodes, }; }; type RegisterChatStreamRouteOptions = { waitTaskCalculations?: TaskCalculationWaiter; interactionRequestStore?: InteractionRequestStore; agentMessages?: AgentMessageCoordinator; frontendActionCoordinator: FrontendActionCoordinator; learningOrchestrator: LearningOrchestrator; memoryStore: MemoryStore; runState: ChatRunState; runtime: OpencodeRuntimeAdapter; sessionBridge: ChatSessionBridge; sessionMetadataStore: SessionMetadataStore; sessionTranscriptStore: SessionTranscriptStore; sessionUiStateStore: SessionUiStateStore; }; export const registerChatStreamRoute = ( chatRouter: Router, { interactionRequestStore, waitTaskCalculations, frontendActionCoordinator, learningOrchestrator, memoryStore, runState, runtime, sessionBridge, sessionMetadataStore, sessionTranscriptStore, sessionUiStateStore, agentMessages, }: RegisterChatStreamRouteOptions, ) => { const { activeRuns, lastRunStatuses } = runState; const waitingInputs = new WaitingResponseInputQueue(runtime); const startingSessions = new Set(); chatRouter.post("/responses", async (req, res) => { const request = responseRequestSchema.safeParse(req.body); if (!request.success) { res.status(400).json({ error: { type: "invalid_request_error", code: "invalid_request", message: "Unsupported or invalid Responses request", param: request.error.issues[0]?.path.join(".") ?? null } }); return; } const input = request.data; const parsed = payloadSchema.safeParse({ message: responsePrompt(input.input), model: input.model, session_id: typeof input.conversation === "string" ? input.conversation : input.conversation?.id, ...input.tjwater, }); if (!parsed.success) { res.status(400).json({ message: "invalid request payload", detail: parsed.error.flatten(), }); return; } let wire: ReturnType | undefined; let boundOperationalSessionId: string | undefined; let startingSessionId: string | undefined; let continuation: InteractionRequest | null = null; let continuationScope: ReturnType | undefined; let waitingInteraction: InteractionRequest | null = null; try { const authContext = getLocalAgentContext(req); const projectId = authContext.projectId; const traceId = req.header("x-trace-id") ?? undefined; const userId = authContext.userId; const actorKey = authContext.actorKey; const projectKey = authContext.projectKey; const trigger = parsed.data.trigger_context; const requestUiMessages = toUiMessages(parsed.data.messages); const requestPromptText = parsed.data.message?.trim() || readLatestUserText(requestUiMessages); let promptText = trigger ? buildAgentTriggerPrompt(trigger) : requestPromptText; if (!promptText) { res.status(400).json({ message: "message or messages with latest user text is required", }); return; } if (trigger && !parsed.data.session_id?.trim()) { res.status(400).json({ message: "event trigger requires a reserved session_id", }); return; } if (input.tjwater.topic_id && !await sessionMetadataStore.topics.get(authContext, input.tjwater.topic_id)) { res.status(404).json({ message: "主题不存在" }); return; } const requestedSessionId = resolveRequestedStreamSessionId({ sessionId: parsed.data.session_id, promptText, createClientSessionId: () => sessionBridge.createClientSessionId(), }); const existingSessionRecord = requestedSessionId ? await sessionMetadataStore.get( { actorKey, projectId, projectKey, userId }, requestedSessionId, ) : null; if (requestedSessionId && !existingSessionRecord) { res.status(404).json({ message: "session not found" }); return; } if (trigger && existingSessionRecord?.triggerId !== trigger.trigger_id) { res.status(409).json({ message: "trigger does not match session" }); return; } if (input.tjwater.interaction_id && (!existingSessionRecord || !interactionRequestStore || trigger || input.tjwater.delivery || input.tjwater.input_id)) { res.status(400).json({ message: "Interaction continuation requires an existing conversation" }); return; } if (existingSessionRecord && interactionRequestStore) { continuationScope = toSessionScope(existingSessionRecord); if (input.tjwater.interaction_id) { continuation = await interactionRequestStore.get(continuationScope, input.tjwater.interaction_id); if (!continuation?.delivery || !["answered", "skipped"].includes(continuation.status) || continuation.delivery.status === "cancelled") { res.status(409).json({ code: "INTERACTION_NOT_READY", message: "Question has no answer waiting to continue" }); return; } promptText = renderInteractionAnswer(continuation); } const blocking = await interactionRequestStore.getBlocking(continuationScope); waitingInteraction = blocking ? await reconcileInteractionDelivery(continuationScope, blocking, interactionRequestStore, runtime, activeRuns.get(existingSessionRecord.sessionId)?.status === "running") : null; if (continuation && waitingInteraction?.interaction_id === continuation.interaction_id) continuation = waitingInteraction; if (waitingInteraction?.delivery?.status === "delivered") waitingInteraction = null; if (waitingInteraction && waitingInteraction.interaction_id !== continuation?.interaction_id && !(input.tjwater.delivery && input.tjwater.input_id && !input.stream && !trigger && readPurpose(existingSessionRecord) !== "task_execution")) { res.status(409).json({ code: "INTERACTION_PENDING", message: "请先回答或取消当前提问。" }); return; } if (continuation?.delivery?.status === "delivered") { // An acknowledged retry must not append or reorder persisted conversation messages. const text = await runtime.readInteractionResult(existingSessionRecord.sessionId, continuation.delivery.native_message_id); if (input.stream) { res.setHeader("Content-Type", "text/event-stream; charset=utf-8"); res.setHeader("Cache-Control", "private, no-store"); } const replay = createResponsesProjection({ model: parsed.data.model ?? config.OPENCODE_MODEL, conversationId: existingSessionRecord.sessionId, emit: (event) => { if (input.stream) res.write(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`); }, }); replay.write({ type: "start", messageId: `answer_${continuation.delivery.native_message_id}` }); replay.write({ type: "text-start", id: "recovered-answer" }); replay.write({ type: "text-delta", id: "recovered-answer", delta: text }); replay.write({ type: "text-end", id: "recovered-answer" }); replay.finish(); if (input.stream) res.end(); else res.json(replay.response); return; } } const running = existingSessionRecord ? activeRuns.get(existingSessionRecord.sessionId) : undefined; if (running?.status === "running") { if (continuation) { res.status(409).json({ code: "INTERACTION_TURN_BUSY", message: "答案已保存，请等待当前轮次结束后继续。" }); return; } if (!input.tjwater.delivery || !input.tjwater.input_id || input.stream || trigger || readPurpose(existingSessionRecord!) === "task_execution") { res.status(409).json({ message: "Running conversations require a non-streaming steer/queue input with input_id" }); return; } if (!running.inputs) { res.status(409).json({ message: "Session is starting; retry after it is ready" }); return; } const receipt = await running.inputs.submit({ id: input.tjwater.input_id, text: promptText, delivery: input.tjwater.delivery, status: "pending" }); res.status(202).json({ object: "tjwater.response_input", session_id: existingSessionRecord!.sessionId, ...receipt }); return; } if (waitingInteraction && !continuation && existingSessionRecord && input.tjwater.delivery && input.tjwater.input_id && !input.stream && !trigger && readPurpose(existingSessionRecord) !== "task_execution") { const receipt = await waitingInputs.submit(existingSessionRecord.sessionId, { id: input.tjwater.input_id, text: promptText, delivery: input.tjwater.delivery, status: "pending" }); res.status(202).json({ object: "tjwater.response_input", session_id: existingSessionRecord.sessionId, ...receipt }); return; } if (input.tjwater.delivery || input.tjwater.input_id) { res.status(409).json({ code: "RESPONSE_FINISHED", message: "The response has finished; send a new response instead" }); return; } if ( existingSessionRecord && readPurpose(existingSessionRecord) === "task_execution" && existingSessionRecord.modelId && parsed.data.model !== existingSessionRecord.modelId ) { res.status(409).json({ message: "task execution model does not match the reserved session" }); return; } const hadExistingRuntimeSession = Boolean(existingSessionRecord); if (existingSessionRecord) { if (startingSessions.has(existingSessionRecord.sessionId)) { res.status(409).json({ message: "Session is starting; retry after it is ready" }); return; } startingSessions.add(existingSessionRecord.sessionId); startingSessionId = existingSessionRecord.sessionId; } const { binding, requestContext, created } = await sessionBridge.resolve({ preserveExecution: Boolean(continuation), sessionId: requestedSessionId, network: authContext.network, projectId, traceId, userId, actorKey, projectKey, }); setRuntimeSessionContext({ ...(getRuntimeSessionContext(binding.sessionId)!), capabilities: parsed.data.capabilities, purpose: existingSessionRecord ? readPurpose(existingSessionRecord) : "ordinary_chat", taskId: existingSessionRecord?.taskId, }); const { record: ensuredSessionRecord, created: sessionCreated } = await sessionMetadataStore.ensure({ actorKey, projectId, projectKey, sessionId: binding.sessionId, topicId: existingSessionRecord?.topicId ?? input.tjwater.topic_id, userId, }); const activeSessionRecord = await sessionMetadataStore.touch(ensuredSessionRecord); const sessionPurpose = readPurpose(activeSessionRecord); if ( activeRuns.get(activeSessionRecord.sessionId)?.status === "running" ) { res.status(409).json({ message: "session is already streaming", session_id: activeSessionRecord.sessionId, }); return; } startingSessions.add(activeSessionRecord.sessionId); startingSessionId = activeSessionRecord.sessionId; bindOperationalTaskCredential( binding.sessionId, getTrustedProductCredential(req), ); boundOperationalSessionId = binding.sessionId; const resolvedTaskContext = activeSessionRecord.taskId && (sessionPurpose === "task_collaboration" || sessionPurpose === "task_execution") ? await operationalTaskClient.getAgentContext(binding.sessionId, activeSessionRecord.taskId, sessionPurpose === "task_collaboration" ? parsed.data.task_context?.report_revision : undefined) : null; const taskContext = sessionPurpose === "task_collaboration" ? resolvedTaskContext : null; const taskSource = z.object({ replay_context: z.record(z.unknown()).nullish() }).passthrough() .safeParse(resolvedTaskContext?.task.source); setRuntimeSessionContext({ ...getRuntimeSessionContext(binding.sessionId)!, taskWorkflow: resolvedTaskContext?.task.workflow, taskReplay: taskSource.success && !!taskSource.data.replay_context }); if (taskContext) { const report = taskContext.selected_report ?? taskContext.task.latest_report; const reference = z.object({ revision: z.number().int().positive() }).passthrough().safeParse(report); setRuntimeSessionContext({ ...getRuntimeSessionContext(binding.sessionId)!, reportRevision: reference.success ? reference.data.revision : undefined }); } const hasRuntimeConversation = hadExistingRuntimeSession ? await runtimeHasConversation(runtime, binding.sessionId) : false; const shouldRestoreConversation = shouldRestoreConversationForRuntime({ hadExistingSessionRecord: hadExistingRuntimeSession, runtimeHasConversation: hasRuntimeConversation, }); const historyContext = { actorKey: requestContext.actorKey, clientSessionId: requestContext.clientSessionId, projectKey: requestContext.projectKey, sessionId: requestContext.clientSessionId, }; const initialSessionState = await sessionUiStateStore.read( toSessionScope(activeSessionRecord), ); const persistedUiMessages = toUiMessages(initialSessionState?.messages); if (!continuation && input.tjwater.client_message_id && persistedUiMessages.some((message) => message.id === input.tjwater.client_message_id)) { res.status(409).json({ message: "client_message_id already exists in this conversation" }); return; } let executionPlanHistory = initialSessionState?.executionPlans ?? readExecutionPlans(initialSessionState?.messages ?? []); const visiblePersistedMessages = sessionPurpose === "task_collaboration" ? toCollaborationMessages(persistedUiMessages) : persistedUiMessages; const visibleRequestMessages = sessionPurpose === "task_collaboration" ? toCollaborationMessages(requestUiMessages) : requestUiMessages; const continuationAssistantId = continuation ? `answer_${continuation.delivery!.native_message_id}` : undefined; const continuationUserExists = continuation && visiblePersistedMessages.some((message) => message.id === continuation!.delivery!.native_message_id); const continuationReply = continuation ? createInteractionReplyMessage(continuation) : undefined; const requestMessages = continuationReply ? [...visiblePersistedMessages.filter((message) => message.id !== continuationAssistantId) .map((message) => message.id === continuationReply.id ? { ...message, ...continuationReply } : message), ...(continuationUserExists ? [] : [continuationReply])] : visibleRequestMessages.length > 0 ? visibleRequestMessages : [ ...visiblePersistedMessages, createFallbackUserUiMessage(promptText, input.tjwater.client_message_id), ]; const baseMessages = toLegacyMessages(requestMessages); const recentTurns = await sessionTranscriptStore.getRecentTurns(historyContext, 8); logger.info( { clientSessionId: requestContext.clientSessionId, sessionId: binding.sessionId, created: created || sessionCreated, model: parsed.data.model, approvalMode: parsed.data.approval_mode, traceId: requestContext.traceId, projectId: requestContext.projectId, }, "processing chat request", ); res.status(200); if (input.stream) { res.setHeader("Content-Type", "text/event-stream; charset=utf-8"); res.setHeader("Cache-Control", "no-cache, no-transform"); res.setHeader("X-Accel-Buffering", "no"); res.flushHeaders?.(); } const clientSessionId = requestContext.clientSessionId; wire = createResponsesProjection({ model: parsed.data.model ?? config.OPENCODE_MODEL, conversationId: clientSessionId, metadata: input.metadata, emit: (event) => { if (input.stream && !res.writableEnded && !res.destroyed) res.write(`event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`); }, }); wire.begin(); const abortController = new AbortController(); sessionBridge.registerAbortController(clientSessionId, abortController); const initialMessages = createInitialStreamingMessages( baseMessages, promptText, ); const activeRun: ActiveRun = { clientSessionId, controller: abortController, messages: updateLastAssistantMessage(initialMessages, (message) => ({ ...message, executionPlans: executionPlanHistory, })), pendingPermissions: new Map(), pendingQuestions: new Map(), status: "running", subscribers: new Set(), }; activeRuns.set(clientSessionId, activeRun); startingSessions.delete(activeSessionRecord.sessionId); startingSessionId = undefined; lastRunStatuses.set(clientSessionId, "running"); const sessionUiStateContext = toSessionScope(activeSessionRecord); const assistantUiMessageId = continuationAssistantId ?? `msg_${randomUUID().replaceAll("-", "")}`; setRuntimeSessionContext({ ...getRuntimeSessionContext(binding.sessionId)!, hostTurnId: wire.response.id, hostMessageId: assistantUiMessageId }); const accumulator = createChatRunAccumulator({ run: activeRun, messages: requestMessages, messageId: assistantUiMessageId, sessionId: clientSessionId, collaboration: sessionPurpose === "task_collaboration", write: (chunk) => wire!.write(chunk), onStatus: (status) => lastRunStatuses.set(clientSessionId, status), }); const writeUiChunk = (chunk: UIMessageChunk) => wire!.write(chunk); writeUiChunk({ type: "start", messageId: assistantUiMessageId }); let persistQueue = sessionUiStateStore.write(sessionUiStateContext, { executionPlans: executionPlanHistory, sessionId: activeSessionRecord.sessionId, isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false, messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(accumulator.uiMessages()) : accumulator.uiMessages(), }); const queueSessionUiStatePersist = () => { const snapshot = { executionPlans: executionPlanHistory, sessionId: activeSessionRecord.sessionId, isTitleManuallyEdited: initialSessionState?.isTitleManuallyEdited ?? false, messages: sessionPurpose === "task_collaboration" ? toCollaborationMessages(accumulator.uiMessages()) : accumulator.uiMessages(), }; persistQueue = persistQueue .catch((error) => { logger.warn( { err: error, sessionId: clientSessionId }, "failed to persist previous chat stream state", ); }) .then(() => sessionUiStateStore.write(sessionUiStateContext, snapshot)); return persistQueue; }; let taskArtifactPublishQueue: Promise = Promise.resolve(); const publish = (event: string, data: Record) => { let subscriberData = event === "token" ? { ...data, message_id: typeof data.message_id === "string" ? data.message_id : assistantUiMessageId, } : data; if (event === "execution_plan") { try { const snapshot = createExecutionPlanSnapshot(data, [{ executionPlans: executionPlanHistory }], clientSessionId); executionPlanHistory = [...executionPlanHistory, snapshot]; subscriberData = snapshot; if (activeSessionRecord.taskId && executionPlanNeedsTaskPublish(data)) { taskArtifactPublishQueue = taskArtifactPublishQueue .catch(() => undefined) .then(() => operationalTaskClient.publishArtifacts(binding.sessionId, activeSessionRecord.taskId!, { requestId: randomUUID(), sourceSessionId: clientSessionId, executionPlan: snapshot, })) .catch((error) => { logger.warn({ err: error, sessionId: clientSessionId, taskId: activeSessionRecord.taskId }, "failed to publish execution plan to task record"); }); } } catch (error) { logger.warn({ err: error, sessionId: clientSessionId }, "rejected invalid execution plan"); publish("progress", { id: "execution-plan-rejected", phase: "execution_plan", status: "error", title: "计划更新未通过校验，保留上一版", detail: error instanceof Error ? error.message : "计划结构无效", }); return; } } const hiddenCollaborationEvent = sessionPurpose === "task_collaboration" && [ "progress", "todo_update", "tool_call", "ui_envelope", "execution_plan", ].includes(event); publishRunEvent(activeRun, event, subscriberData, !hiddenCollaborationEvent); void queueSessionUiStatePersist().catch((error) => { logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); }); }; const unregisterFrontendActionSink = parsed.data.capabilities.includes("frontend-action@2") ? frontendActionCoordinator.registerSink(clientSessionId, (request) => publish("frontend_action", request as unknown as Record)) : undefined; let credentialFailure: Error | undefined; const unsubscribeCredentialRefresh = credentialRefreshCoordinator.subscribe(binding.sessionId, (event) => { publish(event.type, { ...event, session_id: clientSessionId, request_id: event.requestId }); if (event.type === "credential_refresh_failed") { credentialFailure = new Error(event.message); abortController.abort(); } }); if (sessionPurpose === "task_execution") credentialRefreshCoordinator.enableBackground(binding.sessionId); const cancelCredentialRefresh = () => credentialRefreshCoordinator.cancelSession(binding.sessionId); abortController.signal.addEventListener("abort", cancelCredentialRefresh, { once: true }); let taskHeartbeat: ReturnType | undefined; let taskHeartbeatActive = true; let taskCompleting = false; let taskHeartbeatPromise: Promise | undefined; let taskLeaseFailure: Error | undefined; let taskRunFinalized = false; const continueScadaPresentation = createScadaPresentationContinuation(); const stopTaskHeartbeat = async () => { taskHeartbeatActive = false; if (taskHeartbeat) clearInterval(taskHeartbeat); await taskHeartbeatPromise?.catch(() => undefined); }; if (sessionPurpose === "task_execution" && activeSessionRecord.taskId) { taskHeartbeat = setInterval(() => { if (!taskHeartbeatActive || taskHeartbeatPromise || taskCompleting) return; const heartbeat = operationalTaskClient.heartbeatTask( binding.sessionId, activeSessionRecord.taskId!, activeSessionRecord.runId!, clientSessionId, ).then(() => undefined).catch((error) => { if (!taskHeartbeatActive || abortController.signal.aborted) return; taskLeaseFailure = error instanceof Error ? error : new Error("task lease heartbeat failed"); taskHeartbeatActive = false; if (taskHeartbeat) clearInterval(taskHeartbeat); logger.warn( { err: error, sessionId: clientSessionId, taskId: activeSessionRecord.taskId }, "task lease heartbeat failed; stopping stale execution", ); abortController.abort(); }).finally(() => { if (taskHeartbeatPromise === heartbeat) taskHeartbeatPromise = undefined; }); taskHeartbeatPromise = heartbeat; }, 20_000); } taskHeartbeat?.unref(); try { if (taskContext) { const raw = taskContext.selected_report ?? taskContext.task.latest_report; const reportRef = z.object({ revision: z.number().int().positive(), execution_generation: z.number().int().positive().nullish() }).passthrough().safeParse(raw); const generation = taskContext.task.execution_generation; const reference = { task_id: activeSessionRecord.taskId!, execution_generation: reportRef.success ? reportRef.data.execution_generation ?? null : typeof generation === "number" && generation > 0 ? generation : null, report_revision: reportRef.success ? reportRef.data.revision : null, }; accumulator.apply("task_context", reference); } const preparedMessage = taskContext ? [ "你正在与用户协作理解一个工况任务。只根据下面由后端授权返回的任务、报告和证据解释判断；selected_report 存在时用户正在阅读该轮次与修订号，优先据此回答并明确引用。不要展示后台执行会话、工具轨迹或内部推理。普通提问不改变任务，只有用户明确补充的事实可用结构化工具记录并重新分析；假设和查询要求不能当作现场事实。queued 表示等待领取，不能声称已经开始分析。", `任务快照：${JSON.stringify(taskContext)}`, `当前讨论对象：${JSON.stringify(parsed.data.task_context ?? {})}`, `用户问题：${promptText}`, ].join("\n\n") : await buildPromptWithLearningContext( memoryStore, requestContext.actorKey, requestContext.projectKey, { recentTurns, persistedMessages: baseMessages, message: promptText, restoreConversation: shouldRestoreConversation, }, ); const messaging = getTrustedProductCredential(req) ? agentMessages : undefined; messaging?.start(activeSessionRecord); const inputTracker = new ResponseInputTracker(runtime, binding.sessionId, (receipt) => { if (receipt.status === "delivered") { accumulator.appendInput({ id: receipt.id, role: "user", parts: [{ type: "text", text: receipt.text }] } as UIMessage, receipt.text); } publish("user_input", { session_id: clientSessionId, ...receipt }); }); activeRun.inputs = inputTracker; if (continuation && interactionRequestStore && continuationScope && continuation.delivery?.status !== "delivered") { continuation = await interactionRequestStore.claimDelivery(continuationScope, continuation.interaction_id, { expected_revision: continuation.delivery!.revision, response_id: wire.response.id, }); } const streamResult = await streamPromptResponse({ interactionMessageId: continuation?.delivery?.native_message_id, onHostQuestion: interactionRequestStore ? async (id) => { const record = await interactionRequestStore.get(sessionUiStateContext, id); if (!record || record.turn_id !== wire!.response.id || record.message_id !== assistantUiMessageId) throw new Error("Question does not belong to this turn"); return { ...record }; } : undefined, runtime, sessionId: binding.sessionId, clientSessionId, message: preparedMessage, model: parsed.data.model, approvalMode: parsed.data.approval_mode, backgroundSandboxPermissions: sessionPurpose === "task_execution", traceId: requestContext.traceId, projectId: requestContext.projectId, signal: abortController.signal, suppressAnswerUi: sessionPurpose === "task_execution", beforePrompt: async () => { await inputTracker.restore(); await messaging?.flush(binding.sessionId); }, onPromptDispatched: () => inputTracker.markReady(), onInboxEvent: (type, data) => inputTracker.observe(type, data), waitForUserInput: async () => !await inputTracker.canClose(), beforeAnswerClose: async () => { // A consumed input may still be awaiting its model step. Let the // native runner drain it instead of interrupting on an older answer. if (inputTracker.hasAdditionalInput) return false; if (messaging && (messaging.hasInputThisTurn(binding.sessionId) || !await messaging.closeInteractive(activeSessionRecord))) return false; return inputTracker.canClose(); }, beforeIdle: messaging || sessionPurpose === "task_execution" ? async () => { if (sessionPurpose !== "task_execution") return messaging ? messaging.closeInteractive(activeSessionRecord) : true; if (await messaging?.flush(binding.sessionId)) return false; const calculationResume = await waitTaskCalculations?.(binding.sessionId, abortController.signal); if (calculationResume) return calculationResume; await taskArtifactPublishQueue; if (taskLeaseFailure || !activeSessionRecord.taskId || !activeSessionRecord.runId) return true; taskCompleting = true; try { // Finish any earlier lease renewal before making the run terminal. await taskHeartbeatPromise; if (taskLeaseFailure) return true; const task = await operationalTaskClient.completeTaskRun(binding.sessionId, activeSessionRecord.taskId, activeSessionRecord.runId, clientSessionId); taskRunFinalized = true; messaging?.stop(binding.sessionId); await stopTaskHeartbeat(); if (task.lifecycle_status === "failed") throw new Error("Server rejected incomplete investigation artifacts"); return true; } catch (error) { const continuation = continueScadaPresentation(error); if (continuation) return continuation; if (error && typeof error === "object" && "code" in error && error.code === "PENDING_AGENT_MESSAGES") { await messaging?.flush(binding.sessionId); return false; } throw error; } finally { taskCompleting = false; } } : undefined, write: (event, data) => { publish(event, data); }, }); if (credentialFailure) { publish("error", { session_id: clientSessionId, message: "会话凭据续期失败，本次执行已停止。" }); } else if (taskLeaseFailure) { publish("error", { session_id: clientSessionId, message: taskRunInterruptionMessage(taskLeaseFailure), detail: taskLeaseFailure.message, }); } else if (streamResult.failed && sessionPurpose === "task_execution" && activeSessionRecord.taskId) { publish("execution_plan", buildFailedTaskExecutionPlan(executionPlanHistory)); } await taskArtifactPublishQueue; if ( sessionPurpose === "task_execution" && activeSessionRecord.taskId && activeSessionRecord.runId && !taskLeaseFailure && !taskRunFinalized ) { if (streamResult.failed || streamResult.aborted) { await operationalTaskClient.failTaskRun( binding.sessionId, activeSessionRecord.taskId, activeSessionRecord.runId, clientSessionId, credentialFailure ? "CREDENTIAL_REFRESH_FAILED" : streamResult.aborted ? "AGENT_ABORTED" : "AGENT_STREAM_FAILED", streamResult.aborted ? "Agent execution was aborted" : "Agent stream failed", ); taskRunFinalized = true; } else { try { if (await waitTaskCalculations?.(binding.sessionId, abortController.signal)) throw Object.assign(new Error("Calculation results require investigation continuation"), { code: "CALCULATION_CONTINUATION_REQUIRED" }); const completedTask = await operationalTaskClient.completeTaskRun( binding.sessionId, activeSessionRecord.taskId, activeSessionRecord.runId, clientSessionId, ); taskRunFinalized = true; if (completedTask.lifecycle_status === "failed") { throw new Error("Server rejected the run result because required artifacts were incomplete"); } } catch (error) { if (!taskRunFinalized) { const code = error && typeof error === "object" && "code" in error && typeof error.code === "string" ? error.code : "AGENT_COMPLETION_FAILED"; await operationalTaskClient.failTaskRun( binding.sessionId, activeSessionRecord.taskId, activeSessionRecord.runId, clientSessionId, code, error instanceof Error ? error.message : "Agent run completion failed", ); taskRunFinalized = true; } throw error; } } } await stopTaskHeartbeat(); await persistQueue.catch((error) => { logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); }); if (continuation && interactionRequestStore && continuationScope) continuation = await interactionRequestStore.get(continuationScope, continuation.interaction_id); if (continuation && interactionRequestStore && continuationScope && continuation.delivery && !["delivered", "cancelled"].includes(continuation.delivery.status)) { const nativeState = await runtime.agentMessageState(binding.sessionId, continuation.delivery!.native_message_id); continuation = await interactionRequestStore.updateDelivery(continuationScope, continuation.interaction_id, { expected_revision: continuation.delivery!.revision, status: nativeState === "delivered" ? "delivered" : "blocked", response_id: wire.response.id, ...(nativeState !== "delivered" ? { error_code: "CONTINUATION_NOT_CONSUMED" } : {}), }); publish("interaction_request", { ...continuation, session_id: clientSessionId }); } if (!streamResult.aborted && !streamResult.failed && !streamResult.waitingInput) { const messages = await runtime.messages(binding.sessionId, 60); const assistantMessage = [...messages] .reverse() .find((message) => message.info.role === "assistant"); const assistantText = readUiMessageText(accumulator.uiMessages().at(-1)) || collectTextContent(assistantMessage?.parts ?? []); const latestSessionRecord = (await sessionMetadataStore.get( { actorKey, projectId, projectKey, userId }, activeSessionRecord.sessionId, )) ?? activeSessionRecord; const latestSessionState = await sessionUiStateStore.read( toSessionScope(latestSessionRecord), ); const existingSessionTitle = latestSessionRecord.title; let sessionTitle = existingSessionTitle; const shouldGenerateTitle = shouldGenerateSessionTitle({ recentTurnCount: recentTurns.length, isTitleManuallyEdited: latestSessionState?.isTitleManuallyEdited ?? false, }); if (shouldGenerateTitle) { sessionTitle = await generateSessionTitle(runtime, { sessionId: binding.sessionId, latestAssistantMessage: assistantText, latestUserMessage: promptText, fallbackTitle: existingSessionTitle, taskTitle: taskContext?.task.title, }); } await sessionMetadataStore.touch(latestSessionRecord, { ...(sessionTitle && sessionTitle !== existingSessionTitle ? { title: sessionTitle } : {}), }); if ( shouldGenerateTitle && sessionTitle && sessionTitle !== existingSessionTitle ) { publish("session_title", { session_id: clientSessionId, title: sessionTitle, }); await stopTaskHeartbeat(); await persistQueue.catch((error) => { logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); }); } const latestTurn = extractLatestFrontendTurn(activeRun.messages); if (latestTurn && sessionPurpose === "ordinary_chat") { void learningOrchestrator.onTurnCompleted({ ...latestTurn, requestContext, sessionId: clientSessionId, }).catch((error) => { logger.warn( { err: error, sessionId: clientSessionId }, "stream-completed learning failed", ); }); } } } catch (error) { const preparationFailure = error instanceof McpPreparationError ? error : null; publish("error", { session_id: clientSessionId, message: abortController.signal.aborted ? "请求已中断" : preparationFailure?.message ?? "Agent response failed", ...(preparationFailure && !abortController.signal.aborted ? { code: preparationFailure.code } : {}) }); await stopTaskHeartbeat(); if ( sessionPurpose === "task_execution" && activeSessionRecord.taskId ) { publish("execution_plan", buildFailedTaskExecutionPlan(executionPlanHistory)); await taskArtifactPublishQueue; if (activeSessionRecord.runId && !taskLeaseFailure && !taskRunFinalized) { await operationalTaskClient.failTaskRun( binding.sessionId, activeSessionRecord.taskId, activeSessionRecord.runId, clientSessionId, abortController.signal.aborted ? "AGENT_ABORTED" : preparationFailure?.code ?? "AGENT_EXECUTION_ERROR", abortController.signal.aborted ? "Agent execution was aborted" : error instanceof Error ? error.message : "Agent execution failed", ).then(() => { taskRunFinalized = true; }).catch((finishError) => { logger.warn({ err: finishError, sessionId: clientSessionId }, "failed to record task run failure"); }); } } throw error; } finally { activeRun.inputs?.close(); if (abortController.signal.aborted || activeRun.status === "error") { await runtime.cancelPendingUserInputs?.(binding.sessionId).catch((error) => { logger.warn({ err: error, sessionId: clientSessionId }, "failed to cancel pending user inputs"); }); if (activeRun.inputs?.hasAdditionalInput && activeRun.status === "error") await runtime.abortSession(binding.sessionId).catch((error) => { logger.warn({ err: error, sessionId: clientSessionId }, "failed to stop pending input continuation"); }); } unsubscribeCredentialRefresh(); abortController.signal.removeEventListener("abort", cancelCredentialRefresh); await stopTaskHeartbeat(); await taskArtifactPublishQueue.catch(() => undefined); unregisterFrontendActionSink?.(); if (abortController.signal.aborted && !taskLeaseFailure) { publishRunEvent(activeRun, "aborted", { session_id: clientSessionId, message: "请求已中断" }); void queueSessionUiStatePersist().catch((error) => { logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist aborted chat stream state"); }); } if (!accumulator.finished) { publishRunEvent(activeRun, activeRun.status === "error" ? "error" : "done", { session_id: clientSessionId }); } await queueSessionUiStatePersist().catch((error) => { logger.warn({ err: error, sessionId: clientSessionId }, "failed to persist chat stream state"); }); sessionBridge.finalizeRequest(clientSessionId); agentMessages?.stop(binding.sessionId); releaseOperationalTaskCredential(binding.sessionId); activeRun.status = (taskLeaseFailure || credentialFailure) ? "error" : abortController.signal.aborted ? "aborted" : activeRun.status === "running" ? "completed" : activeRun.status; lastRunStatuses.set(clientSessionId, activeRun.status); wire.finish(activeRun.status === "aborted" ? "cancelled" : activeRun.status === "error" ? "failed" : "completed"); if (!input.stream && !res.writableEnded && !res.destroyed) res.json(wire.response); for (const subscriber of activeRun.subscribers) { subscriber.close(); } activeRun.subscribers.clear(); if ( activeRun.pendingPermissions.size === 0 && activeRun.pendingQuestions.size === 0 ) { activeRuns.delete(clientSessionId); } } if (!res.writableEnded && !res.destroyed) { res.end(); } } catch (error) { if (startingSessionId) { startingSessions.delete(startingSessionId); } if (boundOperationalSessionId) { releaseOperationalTaskCredential(boundOperationalSessionId); } if (continuation && continuationScope && interactionRequestStore) { try { const latest = await interactionRequestStore.get(continuationScope, continuation.interaction_id); if (latest?.delivery && !["delivered", "cancelled"].includes(latest.delivery.status)) await interactionRequestStore.updateDelivery(continuationScope, latest.interaction_id, { expected_revision: latest.delivery.revision, status: "blocked", error_code: "CONTINUATION_UNCONFIRMED", response_id: latest.delivery.response_id, }); } catch { /* Keep the durable input identity even when state storage is unavailable. */ } } logger.error({ err: error }, "chat stream failed"); if (wire) { wire.write({ type: "error", errorText: "Agent response failed" }); wire.finish("failed"); if (!res.writableEnded && !res.destroyed) { if (!input.stream) res.json(wire.response); else res.end(); } return; } const inputStatus = error && typeof error === "object" && "status" in error ? error.status : undefined; if (inputStatus === 409 || inputStatus === 429) { res.status(inputStatus).json({ message: inputStatus === 429 ? "Too many pending user inputs" : "Response input is no longer available; refresh the conversation" }); return; } res.status(500).json({ error: { type: "server_error", code: "agent_response_failed", message: "Agent response failed" } }); } }); };