import type { Event as OpencodeEvent, Part } from "@opencode-ai/sdk/v2"; import { writeLlmRequestAuditLog } from "../audit/llmRequestAudit.js"; import { type SupportedModel } from "../chat/models.js"; import { logger } from "../logger.js"; import { type PermissionReply, type OpencodeRuntimeAdapter, } from "../runtime/opencode.js"; import { collectTextContent, extractSkillAuditInfo, getErrorMessage, getAssistantMessagePhase, getUnknownErrorMessage, hasToolParams, isPermissionAskedEvent, isPermissionRepliedEvent, isPermissionV2AskedEvent, isPermissionV2RepliedEvent, isQuestionAskedEvent, isObjectRecord, isQuestionRejectedEvent, isQuestionRepliedEvent, isQuestionV2AskedEvent, isQuestionV2RejectedEvent, isQuestionV2RepliedEvent, isSessionEvent, isSkillEvent, initialActivity, logDevelopmentDebug, normalizeQuestionAnswers, normalizeQuestionPayload, normalizeQuestionToolPayload, normalizeTodoPriority, normalizeTodoStatus, normalizeToolParams, type ActivityPayload, type ActivityUpdatePayload, type PermissionRequestPayload, type QuestionRequestPayload, type AssistantMessagePhase, type TodoItemPayload, type TodoUpdatePayload, } from "./chatStreamEvents.js"; import { createActivityTracker } from "./chatActivityTracker.js"; import { resolvePermissionApproval, type ApprovalMode, } from "./chatPermissionPolicy.js"; export { collectTextContent, initialActivity, type PermissionRequestPayload, type QuestionRequestPayload, type TodoItemPayload, type TodoUpdatePayload, type ActivityPayload, type ActivityUpdatePayload, } from "./chatStreamEvents.js"; export type { ApprovalMode } from "./chatPermissionPolicy.js"; type StreamPromptOptions = { runtime: OpencodeRuntimeAdapter; sessionId: string; clientSessionId: string; message: string; model?: SupportedModel; approvalMode?: ApprovalMode; traceId?: string; projectId?: string; signal?: AbortSignal; workspaceRoot?: string; write: (event: string, data: Record) => void; }; const getPermissionTarget = (metadata: unknown) => { if (!isObjectRecord(metadata)) { return undefined; } for (const key of ["command", "path", "file", "filepath", "directory"]) { const value = metadata[key]; if (typeof value === "string" && value.trim()) { return value.trim(); } } return undefined; }; const toRuntimeModel = (model?: SupportedModel) => { if (!model) { return undefined; } const [providerID, modelID] = model.split("/"); if (!providerID || !modelID) { return undefined; } return { providerID, modelID, }; }; const FINAL_ANSWER_TOOL_NAME = "final_answer"; const ACTIVITY_UPDATE_TOOL_NAME = "activity_update"; const extractFinalAnswer = (value: unknown) => isObjectRecord(value) && typeof value.answer === "string" ? value.answer.trim() : ""; const resolveFinalMessage = async ( runtime: OpencodeRuntimeAdapter, sessionId: string, currentAssistantMessageIds: Set, assistantTextParts: Map>, assistantTextPartPhases: Map, ) => { let text = [...currentAssistantMessageIds] .reverse() .map((messageId) => { const parts = assistantTextParts.get(messageId); if (!parts) { return ""; } const candidates = [...parts.entries()]; const finalText = candidates .filter(([partId]) => assistantTextPartPhases.get(partId) === "final_answer") .map(([, content]) => content) .join(""); if (finalText) { return finalText; } return candidates .filter(([partId]) => assistantTextPartPhases.get(partId) !== "commentary") .map(([, content]) => content) .join(""); }) .find((content) => content.length > 0) ?? ""; if (!text) { const messages = await runtime.messages(sessionId); const assistantMessage = [...messages] .reverse() .find( (message) => message.info.role === "assistant" && (currentAssistantMessageIds.size === 0 || currentAssistantMessageIds.has(message.info.id)), ); const assistantParts = assistantMessage?.parts ?? []; text = collectTextContent( assistantParts.filter( (part) => part.type !== "text" || getAssistantMessagePhase(part.metadata) !== "commentary", ), ); if (!text) { const finalAnswerPart = [...assistantParts] .reverse() .find( (part) => part.type === "tool" && part.tool === FINAL_ANSWER_TOOL_NAME, ); if (finalAnswerPart?.type === "tool") { text = extractFinalAnswer(finalAnswerPart.state.input); } } } return text.trim(); }; const normalizeActivityTodos = (value: unknown): TodoItemPayload[] | undefined => { if (!Array.isArray(value)) return undefined; const now = Date.now(); return value .filter(isObjectRecord) .map((todo, index) => { const content = typeof todo.content === "string" ? todo.content.trim() : ""; return { id: typeof todo.id === "string" && todo.id.trim() ? todo.id.trim() : `todo-${index}-${content.slice(0, 24)}`, content, status: normalizeTodoStatus( typeof todo.status === "string" ? todo.status : "pending", ), priority: normalizeTodoPriority( typeof todo.priority === "string" ? todo.priority : "", ), updated_at: now, }; }) .filter((todo) => todo.content.length > 0); }; export const streamPromptResponse = async ({ runtime, sessionId, clientSessionId, message, model, approvalMode = "request", traceId, projectId, signal, workspaceRoot, write, }: StreamPromptOptions): Promise<{ aborted: boolean; failed: boolean; toolCallCount: number; }> => { const eventStream = await runtime.subscribeEvents(workspaceRoot); const iterator = eventStream[Symbol.asyncIterator](); const requestStartedAt = Date.now(); const promptStartedAt = Date.now(); const emittedToolParts = new Set(); const emittedActivityParts = new Set(); const emittedQuestionToolParts = new Set(); const emittedQuestionRequestIds = new Set(); const currentAssistantMessageIds = new Set(); const assistantTextParts = new Map>(); const assistantTextPartPhases = new Map(); const partTypes = new Map(); const pendingTextDeltas = new Map(); const reasoningStatuses = new Map(); const toolStatuses = new Map(); let firstSessionEventLogged = false; let firstNonStatusEventLogged = false; let firstToolEventLogged = false; let lastSessionStatus: string | null = null; let lastSessionStatusMessage: string | null = null; let sawResponseActivity = false; let finalAnswerText = ""; let toolCallCount = 0; let done = false; let promptSettled = false; let aborted = signal?.aborted ?? false; let failed = false; const debugContext = { sessionId, clientSessionId, traceId, projectId, model: model ?? null, }; logDevelopmentDebug("chat stream started", { ...debugContext, messageChars: message.length, }); const abortPromise = signal ? new Promise<{ type: "abort" }>((resolve) => { if (signal.aborted) { resolve({ type: "abort" }); return; } signal.addEventListener("abort", () => resolve({ type: "abort" }), { once: true, }); }) : null; const activityTracker = createActivityTracker({ clientSessionId, write }); const captureFinalText = (text: string) => { const answer = text.trim(); if (answer) finalAnswerText = answer; }; activityTracker.start( initialActivity.id, initialActivity.title, initialActivity.reason, ); const promptPromise = runtime .prompt( sessionId, message, toRuntimeModel(model), ) .then(() => { promptSettled = true; logDevelopmentDebug("runtime.prompt resolved", { ...debugContext, elapsedMs: Math.max(0, Date.now() - promptStartedAt), }); }) .catch((error: unknown) => { promptSettled = true; logDevelopmentDebug("runtime.prompt failed", { ...debugContext, elapsedMs: Math.max(0, Date.now() - promptStartedAt), error: getUnknownErrorMessage(error), }); throw error; }); logDevelopmentDebug("runtime.prompt dispatched", { ...debugContext, }); let pendingIteratorNext: ReturnType | undefined; try { while (!done) { if (signal?.aborted) { aborted = true; logDevelopmentDebug("chat stream noticed abort signal", { ...debugContext, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); break; } pendingIteratorNext ??= iterator.next(); const nextEvent = pendingIteratorNext .then((result) => ({ type: "event" as const, result })); const nextPrompt = promptSettled ? null : promptPromise.then( () => ({ type: "prompt" as const }), (error: unknown) => ({ type: "prompt-error" as const, error }), ); const next = await Promise.race( [ ...(nextPrompt ? [nextEvent, nextPrompt] : [nextEvent]), ...(abortPromise ? [abortPromise] : []), ], ); if (next.type === "abort") { aborted = true; break; } if (next.type === "prompt-error") { throw next.error; } if (next.type === "prompt") { continue; } pendingIteratorNext = undefined; if (next.result.done) { break; } const event = next.result.value as OpencodeEvent; if (!isSessionEvent(event, sessionId)) { continue; } if (!firstSessionEventLogged) { firstSessionEventLogged = true; logDevelopmentDebug("first session event received", { ...debugContext, eventType: event.type, elapsedMs: Math.max(0, Date.now() - requestStartedAt), sincePromptDispatchMs: Math.max(0, Date.now() - promptStartedAt), }); } if (event.type === "session.status") { const nextStatus = event.properties.status.type; const nextStatusMessage = "message" in event.properties.status && typeof event.properties.status.message === "string" ? event.properties.status.message : null; if ( nextStatus !== lastSessionStatus || nextStatusMessage !== lastSessionStatusMessage ) { lastSessionStatus = nextStatus; lastSessionStatusMessage = nextStatusMessage; logDevelopmentDebug("session status updated", { ...debugContext, status: nextStatus, statusMessage: nextStatusMessage, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); } continue; } if (!firstNonStatusEventLogged) { firstNonStatusEventLogged = true; logDevelopmentDebug("first non-status session event received", { ...debugContext, eventType: event.type, elapsedMs: Math.max(0, Date.now() - requestStartedAt), sincePromptDispatchMs: Math.max(0, Date.now() - promptStartedAt), }); } if (isPermissionAskedEvent(event)) { sawResponseActivity = true; const permissionApproval = resolvePermissionApproval( approvalMode, event.properties.permission, { metadata: event.properties.metadata, patterns: event.properties.patterns, workspaceRoot, }, ); logDevelopmentDebug("permission request received", { ...debugContext, requestId: event.properties.id, permission: event.properties.permission, patterns: event.properties.patterns, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); if (permissionApproval.autoApprove || permissionApproval.autoReject) { const reply = permissionApproval.autoReject ? "reject" : "once"; await runtime.replyPermission({ requestId: event.properties.id, sessionId, directory: workspaceRoot, reply, }); write("permission_response", { session_id: clientSessionId, request_id: event.properties.id, reply: reply satisfies PermissionReply, }); continue; } const activity = activityTracker.getCurrentContext(); write("permission_request", { session_id: clientSessionId, request_id: event.properties.id, permission: event.properties.permission, patterns: event.properties.patterns, target: getPermissionTarget(event.properties.metadata), activity_id: activity?.id, reason: activity?.reason, always: event.properties.always, tool: event.properties.tool, created_at: Date.now(), } satisfies PermissionRequestPayload); continue; } if (isPermissionV2AskedEvent(event)) { sawResponseActivity = true; const permissionApproval = resolvePermissionApproval( approvalMode, event.properties.action, { metadata: event.properties.metadata, patterns: event.properties.resources, workspaceRoot, }, ); logDevelopmentDebug("permission v2 request received", { ...debugContext, requestId: event.properties.id, action: event.properties.action, resources: event.properties.resources, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); if (permissionApproval.autoApprove || permissionApproval.autoReject) { const reply = permissionApproval.autoReject ? "reject" : "once"; await runtime.replyPermission({ requestId: event.properties.id, sessionId, directory: workspaceRoot, reply, }); write("permission_response", { session_id: clientSessionId, request_id: event.properties.id, reply: reply satisfies PermissionReply, }); continue; } const activity = activityTracker.getCurrentContext(); write("permission_request", { session_id: clientSessionId, request_id: event.properties.id, permission: event.properties.action, patterns: event.properties.resources, target: getPermissionTarget(event.properties.metadata), activity_id: activity?.id, reason: activity?.reason, always: event.properties.save ?? [], tool: undefined, created_at: Date.now(), } satisfies PermissionRequestPayload); continue; } if (isPermissionRepliedEvent(event)) { sawResponseActivity = true; logDevelopmentDebug("permission request replied", { ...debugContext, requestId: event.properties.requestID, reply: event.properties.reply, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); write("permission_response", { session_id: clientSessionId, request_id: event.properties.requestID, reply: event.properties.reply satisfies PermissionReply, }); continue; } if (isPermissionV2RepliedEvent(event)) { sawResponseActivity = true; logDevelopmentDebug("permission v2 request replied", { ...debugContext, requestId: event.properties.requestID, reply: event.properties.reply, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); write("permission_response", { session_id: clientSessionId, request_id: event.properties.requestID, reply: event.properties.reply satisfies PermissionReply, }); continue; } if (isQuestionAskedEvent(event) || isQuestionV2AskedEvent(event)) { sawResponseActivity = true; logDevelopmentDebug("question request received", { ...debugContext, requestId: event.properties.id, questionCount: event.properties.questions.length, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); const payload = normalizeQuestionPayload(event, clientSessionId); emittedQuestionRequestIds.add(payload.request_id); write("question_request", payload); continue; } if (isQuestionRepliedEvent(event) || isQuestionV2RepliedEvent(event)) { sawResponseActivity = true; logDevelopmentDebug("question request replied", { ...debugContext, requestId: event.properties.requestID, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); write("question_response", { session_id: clientSessionId, request_id: event.properties.requestID, answers: normalizeQuestionAnswers(event.properties.answers), }); continue; } if (isQuestionRejectedEvent(event) || isQuestionV2RejectedEvent(event)) { sawResponseActivity = true; logDevelopmentDebug("question request rejected", { ...debugContext, requestId: event.properties.requestID, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); write("question_response", { session_id: clientSessionId, request_id: event.properties.requestID, rejected: true, }); continue; } if (isSkillEvent(event)) { sawResponseActivity = true; const { name, payload } = extractSkillAuditInfo(event); const activity = activityTracker.getCurrentContext(); logDevelopmentDebug("skill event received", { ...debugContext, skill: name, payloadKeys: Object.keys(payload).slice(0, 8), elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); void writeLlmRequestAuditLog({ kind: "skill", sessionId: sessionId, clientSessionId, traceId, projectId, target: name, activityId: activity?.id, activityTitle: activity?.title, activityReason: activity?.reason, payload, }).catch((error) => { logger.warn({ err: error }, "failed to write skill audit log"); }); } if (event.type === "message.updated") { if (event.properties.info.role === "assistant") { sawResponseActivity = true; currentAssistantMessageIds.add(event.properties.info.id); } continue; } if (event.type === "message.part.delta" && event.properties.field === "text") { sawResponseActivity = true; currentAssistantMessageIds.add(event.properties.messageID); const partType = partTypes.get(event.properties.partID); if (partType === "text") { const messageParts = assistantTextParts.get(event.properties.messageID) ?? new Map(); const text = `${messageParts.get(event.properties.partID) ?? ""}${event.properties.delta}`; messageParts.set(event.properties.partID, text); assistantTextParts.set(event.properties.messageID, messageParts); const phase = assistantTextPartPhases.get(event.properties.partID) ?? "unknown"; if (phase === "final_answer") { captureFinalText(text); } } else if (!partType) { const pending = pendingTextDeltas.get(event.properties.partID) ?? []; pending.push(event.properties.delta); pendingTextDeltas.set(event.properties.partID, pending); } continue; } if (event.type === "message.part.updated") { sawResponseActivity = true; const part = event.properties.part; partTypes.set(part.id, part.type); if (part.type === "text" || part.type === "reasoning" || part.type === "tool") { currentAssistantMessageIds.add(part.messageID); } if (part.type === "text") { const phase = getAssistantMessagePhase(part.metadata); assistantTextPartPhases.set(part.id, phase); const pendingText = (pendingTextDeltas.get(part.id) ?? []).join(""); pendingTextDeltas.delete(part.id); const messageParts = assistantTextParts.get(part.messageID) ?? new Map(); const text = part.text || pendingText; messageParts.set(part.id, text); assistantTextParts.set(part.messageID, messageParts); if (phase === "final_answer") { captureFinalText(text); } } else { pendingTextDeltas.delete(part.id); } if (part.type === "reasoning") { const reasoningStatus = part.time.end ? "completed" : "running"; if (reasoningStatuses.get(part.id) !== reasoningStatus) { reasoningStatuses.set(part.id, reasoningStatus); logDevelopmentDebug("reasoning part status changed", { ...debugContext, partId: part.id, status: reasoningStatus, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); } } if (part.type === "tool") { if (!firstToolEventLogged) { firstToolEventLogged = true; logDevelopmentDebug("first tool event received", { ...debugContext, partId: part.id, tool: part.tool, status: part.state.status, elapsedMs: Math.max(0, Date.now() - requestStartedAt), sincePromptDispatchMs: Math.max(0, Date.now() - promptStartedAt), }); } const toolParams = normalizeToolParams(part.state.input); const isToolFinalState = part.state.status === "completed" || part.state.status === "error"; const nextToolStatus = String(part.state.status); if (toolStatuses.get(part.id) !== nextToolStatus) { toolStatuses.set(part.id, nextToolStatus); logDevelopmentDebug("tool part status changed", { ...debugContext, partId: part.id, tool: part.tool, status: nextToolStatus, inputKeys: Object.keys(toolParams).slice(0, 8), error: part.state.status === "error" ? (part.state.error ?? "unknown") : null, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); } if (part.tool === FINAL_ANSWER_TOOL_NAME) { if (part.state.status === "error") { logger.warn( { sessionId, clientSessionId, partId: part.id, error: part.state.error, }, "final answer tool failed", ); continue; } if ( (part.state.status !== "running" && part.state.status !== "completed") ) { continue; } const answer = extractFinalAnswer(toolParams); if (!answer) { logger.warn( { sessionId, clientSessionId, partId: part.id }, "final answer tool received without an answer", ); continue; } captureFinalText(answer); logDevelopmentDebug("final answer submitted through tool", { ...debugContext, partId: part.id, tool: part.tool, answerChars: answer.length, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); continue; } if (part.tool === ACTIVITY_UPDATE_TOOL_NAME) { const title = typeof toolParams.title === "string" ? toolParams.title.trim() : ""; const reason = typeof toolParams.reason === "string" ? toolParams.reason.trim() : ""; const todos = normalizeActivityTodos(toolParams.todos); if ( title && reason && !emittedActivityParts.has(part.id) && (hasToolParams(toolParams) || isToolFinalState) ) { emittedActivityParts.add(part.id); const activity = activityTracker.start(part.id, title, reason, todos); void writeLlmRequestAuditLog({ kind: "activity", sessionId, clientSessionId, traceId, projectId, target: ACTIVITY_UPDATE_TOOL_NAME, activityId: activity.id, activityTitle: activity.title, activityReason: activity.reason, payload: toolParams, }).catch((error) => { logger.warn({ err: error }, "failed to write activity audit log"); }); } continue; } const questionToolPayload = normalizeQuestionToolPayload( part, toolParams, clientSessionId, ); if (questionToolPayload) { if (!emittedQuestionToolParts.has(part.id)) { emittedQuestionToolParts.add(part.id); emittedQuestionRequestIds.add(questionToolPayload.request_id); logDevelopmentDebug("question tool request received", { ...debugContext, requestId: questionToolPayload.request_id, tool: part.tool, questionCount: questionToolPayload.questions.length, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); write("question_request", questionToolPayload); } continue; } if (part.tool === "todowrite" || part.tool === "todo") { continue; } activityTracker.upsertAction(part, toolParams); if ( !emittedToolParts.has(part.id) && (hasToolParams(toolParams) || isToolFinalState) ) { emittedToolParts.add(part.id); toolCallCount += 1; const activity = activityTracker.getActionContext(part.id); void writeLlmRequestAuditLog({ kind: "tool", sessionId: sessionId, clientSessionId, traceId, projectId, target: part.tool, activityId: activity?.id, activityTitle: activity?.title, activityReason: activity?.reason, payload: toolParams, }).catch((error) => { logger.warn({ err: error }, "failed to write tool audit log"); }); write("tool_call", { session_id: clientSessionId, tool: part.tool, params: toolParams, }); } } continue; } if (event.type === "todo.updated") { sawResponseActivity = true; const todos = event.properties.todos as Array<{ content: string; status: string; priority: string; }>; const normalizedTodos = todos.map((todo, index) => ({ id: `todo-${index}-${todo.content.slice(0, 24)}`, content: todo.content, status: normalizeTodoStatus(todo.status), priority: normalizeTodoPriority(todo.priority), updated_at: Date.now(), })); write("todo_update", { session_id: clientSessionId, todos: normalizedTodos, created_at: Date.now(), } satisfies TodoUpdatePayload); continue; } if (event.type === "session.error") { sawResponseActivity = true; logDevelopmentDebug("session error received", { ...debugContext, elapsedMs: Math.max(0, Date.now() - requestStartedAt), error: event.properties.error ? getErrorMessage(event.properties.error) : "opencode session error", }); activityTracker.finalize("error"); write("error", { session_id: clientSessionId, message: event.properties.error ? getErrorMessage(event.properties.error) : "opencode session error", detail: event.properties.error?.name, total_duration_ms: Math.max(0, Date.now() - requestStartedAt), }); failed = true; done = true; continue; } if (event.type === "session.idle") { if (!sawResponseActivity) { logDevelopmentDebug("ignoring session idle before response activity", { ...debugContext, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); continue; } logDevelopmentDebug("session idle received", { ...debugContext, hasFinalAnswer: Boolean(finalAnswerText), toolCallCount, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); done = true; } } if (aborted) { logDevelopmentDebug("chat stream aborting session", { ...debugContext, elapsedMs: Math.max(0, Date.now() - requestStartedAt), }); activityTracker.finalize("cancelled"); await runtime.abortSession(sessionId).catch((error) => { logger.warn({ sessionId: sessionId, err: error }, "failed to abort opencode session"); }); await runtime.waitForSessionIdle(sessionId).catch((error) => { logger.warn( { sessionId: sessionId, err: error }, "failed while waiting for aborted opencode session to become idle", ); }); return { aborted: true, failed: false, toolCallCount }; } if (failed) { return { aborted: false, failed: true, toolCallCount }; } await promptPromise; if (!finalAnswerText) { finalAnswerText = await resolveFinalMessage( runtime, sessionId, currentAssistantMessageIds, assistantTextParts, assistantTextPartPhases, ); } activityTracker.finalize("completed"); if (finalAnswerText) { // Keep one compatibility cycle for deployed frontends that only consume token. write("token", { session_id: clientSessionId, content: finalAnswerText, }); write("final_answer", { session_id: clientSessionId, content: finalAnswerText, }); } write("done", { session_id: clientSessionId, total_duration_ms: Math.max(0, Date.now() - requestStartedAt), }); logDevelopmentDebug("chat stream completed", { ...debugContext, hasFinalAnswer: Boolean(finalAnswerText), toolCallCount, totalDurationMs: Math.max(0, Date.now() - requestStartedAt), }); return { aborted: false, failed: false, toolCallCount }; } finally { await iterator.return?.(undefined); if (!promptSettled && !aborted) { await promptPromise.catch(() => undefined); } else if (!promptSettled) { void promptPromise.catch(() => undefined); } logDevelopmentDebug("chat stream cleanup finished", { ...debugContext, promptSettled, totalDurationMs: Math.max(0, Date.now() - requestStartedAt), }); } };