diff --git a/packages/plugin-sdk/src/backend.ts b/packages/plugin-sdk/src/backend.ts index 95fdb9e9f..1f14e9288 100644 --- a/packages/plugin-sdk/src/backend.ts +++ b/packages/plugin-sdk/src/backend.ts @@ -1635,7 +1635,11 @@ export interface PluginFetchInit { method?: string; headers?: Record; body?: string; - /** Aborts the request after this long. Defaults to 30 seconds. */ + /** + * Aborts the request when no response headers arrive within this long. + * Defaults to 30 seconds. A streamed body may also sit idle between chunks + * for this long, never less than 5 minutes. + */ timeoutMs?: number; /** * Exact private or loopback hosts this request may reach. Everything else diff --git a/plugins/ai/src/backend/engine.ts b/plugins/ai/src/backend/engine.ts index c8a735ea3..1cf4e9096 100644 --- a/plugins/ai/src/backend/engine.ts +++ b/plugins/ai/src/backend/engine.ts @@ -1,4 +1,5 @@ import { getErrorMessage } from "./errors.js"; +import { describeProviderError } from "./providers/http.js"; import { getAdapter } from "./providers/registry.js"; import type { ChatMessage, @@ -27,7 +28,8 @@ export type EngineEvent = | { type: "tool_call"; name: string; arguments: Record } | { type: "tool_result"; name: string; result: unknown } | { type: "proposal"; draft: ProposalDraft } - | { type: "assistant_message"; content: string; toolCalls: ToolCall[] } + /** A message added to the conversation, for the caller to store. */ + | { type: "message"; message: ChatMessage } | { type: "done" } | { type: "error"; message: string }; @@ -74,23 +76,34 @@ export async function* runAgent( } } } catch (error) { - const message = getErrorMessage(error, "The provider request failed"); + // The user stopped it; there is no one left to show an error to. + if (options.signal?.aborted) return; + const message = describeProviderError( + error, + "The provider request failed", + ); yield { type: "error", message }; return; } if (failed) return; - yield { type: "assistant_message", content: text, toolCalls: calls }; - if (!calls.length) { + yield { type: "message", message: { role: "assistant", content: text } }; yield { type: "done" }; return; } - messages.push({ role: "assistant", content: text, toolCalls: calls }); + const assistant: ChatMessage = { + role: "assistant", + content: text, + toolCalls: calls, + }; + messages.push(assistant); + yield { type: "message", message: assistant }; for (const call of calls) { + if (options.signal?.aborted) return; yield { type: "tool_call", name: call.name, arguments: call.arguments }; const result = await runTool(byName, call, options.context); @@ -107,7 +120,7 @@ export async function* runAgent( yield { type: "proposal", draft: result }; // The model is told the proposal is awaiting the user rather than done, // so it does not go on to describe the change as applied. - messages.push({ + const pending: ChatMessage = { role: "tool", content: JSON.stringify({ status: "awaiting_user_approval", @@ -115,17 +128,21 @@ export async function* runAgent( }), toolCallId: call.id, toolName: call.name, - }); + }; + messages.push(pending); + yield { type: "message", message: pending }; continue; } yield { type: "tool_result", name: call.name, result: redact(result) }; - messages.push({ + const answer: ChatMessage = { role: "tool", content: redactToJson(result), toolCallId: call.id, toolName: call.name, - }); + }; + messages.push(answer); + yield { type: "message", message: answer }; } } diff --git a/plugins/ai/src/backend/history.ts b/plugins/ai/src/backend/history.ts new file mode 100644 index 000000000..c5be09cff --- /dev/null +++ b/plugins/ai/src/backend/history.ts @@ -0,0 +1,98 @@ +import type { ChatMessage, ToolCall } from "./providers/types.js"; + +export interface StoredMessage { + role: string; + content: string; + toolCalls: string | null; +} + +const INTERRUPTED_RESULT = JSON.stringify({ + error: "No result: the reply was interrupted before this tool finished.", +}); + +function parseCalls(raw: string | null): ToolCall[] { + if (!raw) return []; + try { + const parsed = JSON.parse(raw); + return Array.isArray(parsed) ? parsed : []; + } catch { + return []; + } +} + +/** A tool row keeps the call it answers in toolCalls, as a one-item list. */ +export function toStoredMessage(message: ChatMessage): StoredMessage { + if (message.role === "tool") { + return { + role: "tool", + content: message.content, + toolCalls: JSON.stringify([ + { id: message.toolCallId, name: message.toolName }, + ]), + }; + } + return { + role: message.role, + content: message.content, + toolCalls: message.toolCalls?.length + ? JSON.stringify(message.toolCalls) + : null, + }; +} + +/** + * Rebuilds what the provider sees from the stored rows. + * + * Every provider rejects an assistant turn whose tool calls have no results, + * which is what a run cut off mid-way leaves behind, so any missing result is + * filled in as interrupted. Tool rows that answer nothing are dropped. + */ +export function toChatHistory(rows: StoredMessage[]): ChatMessage[] { + const history: ChatMessage[] = []; + let open: ToolCall[] = []; + + const closeOpen = () => { + for (const call of open) { + history.push({ + role: "tool", + content: INTERRUPTED_RESULT, + toolCallId: call.id, + toolName: call.name, + }); + } + open = []; + }; + + for (const row of rows) { + if (row.role === "tool") { + const answered = parseCalls(row.toolCalls)[0]; + const index = open.findIndex((call) => call.id === answered?.id); + if (index === -1) continue; + const [call] = open.splice(index, 1); + history.push({ + role: "tool", + content: row.content, + toolCallId: call.id, + toolName: call.name, + }); + continue; + } + + closeOpen(); + + if (row.role === "assistant") { + const calls = parseCalls(row.toolCalls); + history.push({ + role: "assistant", + content: row.content, + ...(calls.length ? { toolCalls: calls } : {}), + }); + open = [...calls]; + } else if (row.role === "user") { + history.push({ role: "user", content: row.content }); + } + } + + closeOpen(); + return history; +} diff --git a/plugins/ai/src/backend/providers/http.ts b/plugins/ai/src/backend/providers/http.ts index b37c71ca0..067e64dce 100644 --- a/plugins/ai/src/backend/providers/http.ts +++ b/plugins/ai/src/backend/providers/http.ts @@ -60,6 +60,59 @@ export function createProviderFetch( }; } +const NETWORK_MESSAGES: Record = { + UND_ERR_BODY_TIMEOUT: + "The provider stopped sending data for too long and the reply was cut off. A slow or overloaded model can cause this.", + UND_ERR_SOCKET: + "The provider closed the connection before the reply finished.", + ECONNRESET: "The provider closed the connection before the reply finished.", + UND_ERR_HEADERS_TIMEOUT: "The provider did not answer in time.", + UND_ERR_CONNECT_TIMEOUT: + "Timed out connecting to the provider. Check the base URL.", + ETIMEDOUT: "Timed out connecting to the provider. Check the base URL.", + ECONNREFUSED: + "The provider refused the connection. Check the base URL and that the provider is running.", + ENOTFOUND: "The provider's host name could not be found. Check the base URL.", + EAI_AGAIN: "The provider's host name could not be found. Check the base URL.", + EHOSTUNREACH: "The provider's host is unreachable from this server.", + ENETUNREACH: "The provider's host is unreachable from this server.", +}; + +/** + * undici reports network failures as a bare "fetch failed" or "terminated" + * with the real reason on `cause`, which is useless to show as is. + */ +export function describeProviderError( + error: unknown, + fallback: string, +): string { + if (error instanceof AiProviderError) return error.message; + if (!(error instanceof Error)) { + return typeof error === "string" && error ? error : fallback; + } + + const cause = (error as { cause?: unknown }).cause as + (Error & { code?: string }) | undefined; + const code = cause?.code ?? (error as { code?: string }).code; + if (code && NETWORK_MESSAGES[code]) return NETWORK_MESSAGES[code]; + + const causeMessage = cause instanceof Error ? cause.message : ""; + if (/private destinations/i.test(causeMessage)) { + return PRIVATE_DESTINATION_MESSAGE; + } + if (code && /CERT|SELF_SIGNED|UNABLE_TO_VERIFY/i.test(code)) { + return `The provider's TLS certificate is not trusted (${code}).`; + } + + if (/^(fetch failed|terminated)$/i.test(error.message)) { + if (causeMessage) return `The provider request failed: ${causeMessage}`; + return error.message === "terminated" + ? NETWORK_MESSAGES.UND_ERR_SOCKET + : fallback; + } + return error.message || fallback; +} + export function joinUrl(base: string, path: string): string { return `${base.replace(/\/+$/, "")}/${path.replace(/^\/+/, "")}`; } diff --git a/plugins/ai/src/backend/routes.ts b/plugins/ai/src/backend/routes.ts index 80806097f..a38067f9e 100644 --- a/plugins/ai/src/backend/routes.ts +++ b/plugins/ai/src/backend/routes.ts @@ -9,18 +9,18 @@ import { readPrivateAllowlist, resolveAiAccess, } from "./gating.js"; -import { createProviderFetch } from "./providers/http.js"; +import { toChatHistory, toStoredMessage } from "./history.js"; +import { + createProviderFetch, + describeProviderError, +} from "./providers/http.js"; import { FALLBACK_MODELS, getAdapter, REQUIRES_API_KEY, REQUIRES_BASE_URL, } from "./providers/registry.js"; -import type { - AiProviderType, - ChatMessage, - ProviderConfig, -} from "./providers/types.js"; +import type { AiProviderType, ProviderConfig } from "./providers/types.js"; import { isAiProviderType } from "./providers/types.js"; import type { AiRepository } from "./repository.js"; import { serviceAvailable } from "./services.js"; @@ -464,13 +464,18 @@ export function registerAiRoutes( res.json({ models, source: "live" }); } catch (err) { + const warning = describeProviderError( + err, + "Could not reach the provider", + ); + ctx.log.warn(`Model detection failed: ${warning}`); // A provider that cannot be reached yet still gets a usable list, so // the form is never a blank text box the user has to guess into. const fallback = FALLBACK_MODELS[providerType as AiProviderType] ?? []; res.json({ models: fallback, source: fallback.length ? "fallback" : "none", - warning: err instanceof Error ? err.message : undefined, + warning, }); } }, @@ -515,7 +520,10 @@ export function registerAiRoutes( res.json({ models }); } catch (err) { // The message can carry the allowlist hint, which the user needs. - const message = getErrorMessage(err, "Could not reach the provider"); + const message = describeProviderError( + err, + "Could not reach the provider", + ); ctx.log.warn(`Failed to list provider models: ${message}`); res.status(502).json({ error: message }); } @@ -752,30 +760,24 @@ export function registerAiRoutes( } }, 30000); + // req's own "close" already fired once express read the body, so the + // response is what tells us the client went away. const abort = new AbortController(); - req.on("close", () => { + res.on("close", () => { clearInterval(heartbeat); - abort.abort(); + if (!res.writableFinished) abort.abort(); }); const send = (event: unknown) => { + if (res.writableEnded || res.destroyed) return; res.write(`data: ${JSON.stringify(event)}\n\n`); }; send({ type: "conversation", conversationId: conversation.id }); - const chatHistory: ChatMessage[] = history.map((entry) => ({ - role: entry.role as ChatMessage["role"], - content: entry.content, - ...(entry.toolCalls - ? { toolCalls: JSON.parse(entry.toolCalls) } - : {}), - })); + const chatHistory = toChatHistory(history); chatHistory.push({ role: "user", content: message.trim() }); - let assistantText = ""; - let assistantToolCalls: unknown[] = []; - try { for await (const event of runAgent({ config, @@ -795,11 +797,14 @@ export function registerAiRoutes( tools, signal: abort.signal, })) { - if (event.type === "assistant_message") { - assistantText = event.content; - // Kept so the next message replays them verbatim. Gemini rejects - // a turn whose functionCall parts lost their thoughtSignature. - assistantToolCalls = event.toolCalls; + if (event.type === "message") { + // Every step is kept, tool results included, so the next message + // replays the run verbatim. Gemini rejects a turn whose + // functionCall parts lost their thoughtSignature. + await repository.appendMessage({ + conversationId: conversation.id, + ...toStoredMessage(event.message), + }); continue; } @@ -829,23 +834,13 @@ export function registerAiRoutes( } finally { clearInterval(heartbeat); } - - if (assistantText || assistantToolCalls.length) { - await repository.appendMessage({ - conversationId: conversation.id, - role: "assistant", - content: assistantText, - toolCalls: assistantToolCalls.length - ? JSON.stringify(assistantToolCalls) - : null, - }); - } await repository.touchConversation(conversation.id); send({ type: "done" }); res.end(); } catch (err) { logError("AI chat stream failed", err); + if (res.writableEnded || res.destroyed) return; if (res.headersSent) { res.write( `data: ${JSON.stringify({ type: "error", message: "The assistant stopped unexpectedly" })}\n\n`, diff --git a/plugins/ai/src/frontend/AiPanel.tsx b/plugins/ai/src/frontend/AiPanel.tsx index b7eb78749..fd0b9ff94 100644 --- a/plugins/ai/src/frontend/AiPanel.tsx +++ b/plugins/ai/src/frontend/AiPanel.tsx @@ -11,7 +11,6 @@ import { Textarea, } from "@termix/plugin-sdk/ui"; import { - getAiConversation, getAiProviders, getAiStatus, setAiOptIn, @@ -22,6 +21,12 @@ import { AiMessage } from "./AiMessage"; import { AiProviderSettings } from "./AiProviderSettings"; import { AiToolCall } from "./AiToolCall"; import { ProposalCard } from "./ProposalCard"; +import { + buildTimeline, + finishedRunEntries, + userEntry, + type HistoryEntry, +} from "./transcript"; import { useAiStream, type ToolActivity } from "./use-ai-stream"; import { activeMentionQuery, @@ -30,22 +35,6 @@ import { } from "./useMentions"; import { mentionLabel } from "./labels"; -interface HistoryEntry { - role: "user" | "assistant"; - content: string; -} - -/** One entry in the rendered conversation, in the order it happened. */ -type TimelineItem = - | { - kind: "message"; - key: string; - role: "user" | "assistant"; - content: string; - } - | { kind: "tool"; key: string; tool: ToolActivity } - | { kind: "proposal"; key: string; proposal: AiProposal }; - export function AiPanel({ activeTab }: { activeTab?: string | null }) { const { t } = useTranslation(); const { state, send, stop } = useAiStream(); @@ -59,6 +48,7 @@ export function AiPanel({ activeTab }: { activeTab?: string | null }) { const [resolvedProposals, setResolvedProposals] = useState< Record >({}); + const runCountRef = useRef(0); const scrollRef = useRef(null); const textareaRef = useRef(null); @@ -157,33 +147,15 @@ export function AiPanel({ activeTab }: { activeTab?: string | null }) { }); }, [state.assistantText, state.tools.length, history.length]); - const reloadHistory = useCallback( - async (conversationId: number | null, reply: string) => { - // Reload from the server so the transcript is the stored one rather - // than a client-side reconstruction. - if (conversationId) { - try { - const loaded = await getAiConversation(conversationId); - setHistory( - loaded.messages - .filter((entry) => entry.role !== "tool") - .map((entry) => ({ - role: entry.role as "user" | "assistant", - content: entry.content, - })), - ); - return; - } catch { - // Fall through to keeping the streamed copy. - } - } - - // No stored copy to fall back on, so the streamed reply is appended - // locally. Without this it would vanish when the next turn resets the - // stream state. - if (reply) { - setHistory((prev) => [...prev, { role: "assistant", content: reply }]); - } + // The run's steps stay in the transcript, so the next message does not wipe + // what the assistant already looked at. + const keepRun = useCallback( + (_conversationId: number | null, reply: string, tools: ToolActivity[]) => { + const runId = runCountRef.current++; + setHistory((prev) => [ + ...prev, + ...finishedRunEntries(runId, tools, reply), + ]); }, [], ); @@ -193,14 +165,14 @@ export function AiPanel({ activeTab }: { activeTab?: string | null }) { if (!message || !providerId || state.streaming) return; setInput(""); - setHistory((prev) => [...prev, { role: "user", content: message }]); + setHistory((prev) => [...prev, userEntry(message)]); await send({ message, providerId, conversationId: state.conversationId, activeTab, - onComplete: reloadHistory, + onComplete: keepRun, }); } @@ -233,14 +205,14 @@ export function AiPanel({ activeTab }: { activeTab?: string | null }) { : `I approved "${label}". It completed.` : `I rejected "${label}". Do not retry it unless I ask.`; - setHistory((prev) => [...prev, { role: "user", content: followUp }]); + setHistory((prev) => [...prev, userEntry(followUp)]); void send({ message: followUp, providerId, conversationId: state.conversationId, activeTab, - onComplete: reloadHistory, + onComplete: keepRun, }); } @@ -254,70 +226,7 @@ export function AiPanel({ activeTab }: { activeTab?: string | null }) { }; }); - /** - * The reply exists in two places once a turn finishes: the streamed text in - * `state`, and the stored copy that `onComplete` reloads into `history`. - * Whichever arrives is shown exactly once, and the stored copy wins because - * it sits in the right place in the timeline. - */ - const last = history[history.length - 1]; - const historyEndsWithThisReply = - state.assistantText.length > 0 && - last?.role === "assistant" && - last.content === state.assistantText; - - const streamingReply = historyEndsWithThisReply ? "" : state.assistantText; - - /** - * One ordered list rather than three stacked blocks. - * - * Tool calls happen before the reply that describes them, but once a turn - * ends its reply moves into `history`, which used to render above a separate - * tools block. The tool call appeared to jump below the answer it produced. - * Splicing the current turn's activity in ahead of its own reply keeps the - * order stable across that handover. - */ - const timeline: TimelineItem[] = []; - - const trailingReply = historyEndsWithThisReply ? history.length - 1 : -1; - - history.forEach((entry, index) => { - if (index === trailingReply) return; - timeline.push({ - kind: "message", - key: `history-${index}`, - role: entry.role, - content: entry.content, - }); - }); - - for (const tool of state.tools) { - timeline.push({ kind: "tool", key: tool.id, tool }); - } - - if (trailingReply >= 0) { - timeline.push({ - kind: "message", - key: `history-${trailingReply}`, - role: "assistant", - content: history[trailingReply].content, - }); - } else if (streamingReply) { - timeline.push({ - kind: "message", - key: "streaming", - role: "assistant", - content: streamingReply, - }); - } - - for (const proposal of proposals) { - timeline.push({ - kind: "proposal", - key: `proposal-${proposal.id}`, - proposal, - }); - } + const timeline = buildTimeline(history, state, proposals); if (loading) { return ( @@ -405,7 +314,7 @@ export function AiPanel({ activeTab }: { activeTab?: string | null }) { ); } diff --git a/plugins/ai/src/frontend/terminal/TerminalAiPanel.tsx b/plugins/ai/src/frontend/terminal/TerminalAiPanel.tsx index da83cff05..7fd3e65a8 100644 --- a/plugins/ai/src/frontend/terminal/TerminalAiPanel.tsx +++ b/plugins/ai/src/frontend/terminal/TerminalAiPanel.tsx @@ -11,6 +11,12 @@ import { Button } from "@termix/plugin-sdk/ui"; import { getAiProviders, type AiProposal, type AiProvider } from "../ai-api"; import { AiMessage } from "../AiMessage"; import { AiToolCall } from "../AiToolCall"; +import { + buildTimeline, + finishedRunEntries, + userEntry, + type HistoryEntry, +} from "../transcript"; import { useAiStream } from "../use-ai-stream"; import { TerminalProposalCard } from "./TerminalProposalCard"; import { @@ -20,25 +26,6 @@ import { type ToolbarPosition, } from "./panel-geometry"; -interface HistoryEntry { - role: "user" | "assistant"; - content: string; -} - -type TimelineItem = - | { - kind: "message"; - key: string; - role: "user" | "assistant"; - content: string; - } - | { - kind: "tool"; - key: string; - tool: import("../use-ai-stream").ToolActivity; - } - | { kind: "proposal"; key: string; proposal: AiProposal }; - interface TerminalAiPanelProps { hostLabel: string; hostId: number; @@ -75,6 +62,7 @@ export function TerminalAiPanel({ Record >({}); const conversationIdRef = useRef(null); + const runCountRef = useRef(0); const scrollRef = useRef(null); // Prefilled rather than sent: the user sees exactly what the terminal is @@ -202,7 +190,7 @@ export function TerminalAiPanel({ const message = input.trim(); if (!message || !providerId || state.streaming) return; - setHistory((current) => [...current, { role: "user", content: message }]); + setHistory((current) => [...current, userEntry(message)]); setInput(""); void send({ @@ -210,14 +198,13 @@ export function TerminalAiPanel({ providerId, conversationId: conversationIdRef.current, activeTab: activeTab ?? `terminal:${hostLabel}`, - onComplete: (conversationId, reply) => { + onComplete: (conversationId, reply, tools) => { conversationIdRef.current = conversationId; - if (reply) { - setHistory((current) => [ - ...current, - { role: "assistant", content: reply }, - ]); - } + const runId = runCountRef.current++; + setHistory((current) => [ + ...current, + ...finishedRunEntries(runId, tools, reply), + ]); }, }); }, [activeTab, hostLabel, input, providerId, send, state.streaming]); @@ -232,58 +219,7 @@ export function TerminalAiPanel({ }; }); - // The reply exists in two places once a turn finishes: the streamed text in - // `state`, and the copy `onComplete` appended to `history`. Show it once, - // and keep tool calls spliced in ahead of the reply they produced instead - // of in a separate trailing block, so the timeline reads in the order - // things actually happened rather than text-then-tools-then-proposals. - const last = history[history.length - 1]; - const historyEndsWithThisReply = - state.assistantText.length > 0 && - last?.role === "assistant" && - last.content === state.assistantText; - const streamingReply = historyEndsWithThisReply ? "" : state.assistantText; - const trailingReply = historyEndsWithThisReply ? history.length - 1 : -1; - - const timeline: TimelineItem[] = []; - - history.forEach((entry, index) => { - if (index === trailingReply) return; - timeline.push({ - kind: "message", - key: `history-${index}`, - role: entry.role, - content: entry.content, - }); - }); - - for (const tool of state.tools) { - timeline.push({ kind: "tool", key: tool.id, tool }); - } - - if (trailingReply >= 0) { - timeline.push({ - kind: "message", - key: `history-${trailingReply}`, - role: "assistant", - content: history[trailingReply].content, - }); - } else if (streamingReply) { - timeline.push({ - kind: "message", - key: "streaming", - role: "assistant", - content: streamingReply, - }); - } - - for (const proposal of proposals) { - timeline.push({ - kind: "proposal", - key: `proposal-${proposal.id}`, - proposal, - }); - } + const timeline = buildTimeline(history, state, proposals); return (
); } diff --git a/plugins/ai/src/frontend/transcript.ts b/plugins/ai/src/frontend/transcript.ts new file mode 100644 index 000000000..8bc4c21ed --- /dev/null +++ b/plugins/ai/src/frontend/transcript.ts @@ -0,0 +1,82 @@ +import type { AiProposal } from "./ai-api"; +import type { StreamState, ToolActivity } from "./use-ai-stream"; + +/** A finished part of the conversation, kept after the stream resets. */ +export type HistoryEntry = + | { kind: "message"; role: "user" | "assistant"; content: string } + | { kind: "tool"; tool: ToolActivity }; + +/** One entry in the rendered conversation, in the order it happened. */ +export type TimelineItem = + | { + kind: "message"; + key: string; + role: "user" | "assistant"; + content: string; + } + /** live is false for a past run's step, which can never still be running. */ + | { kind: "tool"; key: string; tool: ToolActivity; live: boolean } + | { kind: "proposal"; key: string; proposal: AiProposal }; + +export function userEntry(content: string): HistoryEntry { + return { kind: "message", role: "user", content }; +} + +/** + * What a finished run leaves in the transcript: its tool steps, then its + * reply. Tool ids are only unique within one run, so they get a run prefix. + */ +export function finishedRunEntries( + runId: number, + tools: ToolActivity[], + reply: string, +): HistoryEntry[] { + const entries: HistoryEntry[] = tools.map((tool) => ({ + kind: "tool", + tool: { ...tool, id: `run-${runId}-${tool.id}` }, + })); + if (reply) + entries.push({ kind: "message", role: "assistant", content: reply }); + return entries; +} + +/** Past runs, then the run in progress, then its proposals. */ +export function buildTimeline( + history: HistoryEntry[], + state: Pick, + proposals: AiProposal[], +): TimelineItem[] { + const timeline: TimelineItem[] = history.map((entry, index) => + entry.kind === "tool" + ? { kind: "tool", key: entry.tool.id, tool: entry.tool, live: false } + : { + kind: "message", + key: `history-${index}`, + role: entry.role, + content: entry.content, + }, + ); + + for (const tool of state.tools) { + timeline.push({ kind: "tool", key: tool.id, tool, live: true }); + } + + if (state.assistantText) { + timeline.push({ + kind: "message", + key: "streaming", + role: "assistant", + content: state.assistantText, + }); + } + + for (const proposal of proposals) { + timeline.push({ + kind: "proposal", + key: `proposal-${proposal.id}`, + proposal, + }); + } + + return timeline; +} diff --git a/plugins/ai/src/frontend/use-ai-stream.ts b/plugins/ai/src/frontend/use-ai-stream.ts index cfeed6c4c..1edf59a5f 100644 --- a/plugins/ai/src/frontend/use-ai-stream.ts +++ b/plugins/ai/src/frontend/use-ai-stream.ts @@ -43,9 +43,9 @@ export function useAiStream() { setState(INITIAL); }, []); + // The run's own cleanup clears abortRef, after it has handed its steps over. const stop = useCallback(() => { abortRef.current?.abort(); - abortRef.current = null; setState((prev) => ({ ...prev, streaming: false })); }, []); @@ -56,7 +56,15 @@ export function useAiStream() { model?: string; conversationId?: number | null; activeTab?: string | null; - onComplete?: (conversationId: number | null, reply: string) => void; + /** + * Called however the run ends, with its steps, which are then cleared + * from the live state. Keeping them is up to the caller. + */ + onComplete?: ( + conversationId: number | null, + reply: string, + tools: ToolActivity[], + ) => void; }) => { abortRef.current?.abort(); const controller = new AbortController(); @@ -74,6 +82,20 @@ export function useAiStream() { let conversationId = input.conversationId ?? null; let replyText = ""; let toolSequence = 0; + const runTools: ToolActivity[] = []; + + const finish = (error?: string | null) => { + // A newer send took over; this run's state is already gone. + if (abortRef.current !== controller) return; + input.onComplete?.(conversationId, replyText, runTools); + setState((prev) => ({ + ...prev, + streaming: false, + assistantText: "", + tools: [], + ...(error ? { error } : {}), + })); + }; try { const response = await aiApp().fetch("chat/stream", { @@ -152,6 +174,11 @@ export function useAiStream() { // prev.tools.length appended a duplicate entry that no result // could ever match, leaving it stuck on "running". const activityId = `tool-${toolSequence++}`; + runTools.push({ + id: activityId, + name: event.name, + arguments: event.arguments ?? {}, + }); setState((prev) => ({ ...prev, tools: [ @@ -164,6 +191,15 @@ export function useAiStream() { ], })); } else if (event.type === "tool_result") { + for (let i = runTools.length - 1; i >= 0; i -= 1) { + if ( + runTools[i].name === event.name && + !("result" in runTools[i]) + ) { + runTools[i] = { ...runTools[i], result: event.result }; + break; + } + } setState((prev) => { const tools = [...prev.tools]; // Attach to the most recent call of this tool awaiting a result. @@ -186,20 +222,15 @@ export function useAiStream() { } } - setState((prev) => ({ ...prev, streaming: false })); - input.onComplete?.(conversationId, replyText); + finish(); } catch (error) { - if (controller.signal.aborted) { - setState((prev) => ({ ...prev, streaming: false })); - return; - } - setState((prev) => ({ - ...prev, - streaming: false, - error: getErrorMessage(error, "The assistant stopped unexpectedly"), - })); + finish( + controller.signal.aborted + ? null + : getErrorMessage(error, "The assistant stopped unexpectedly"), + ); } finally { - abortRef.current = null; + if (abortRef.current === controller) abortRef.current = null; } }, [], diff --git a/plugins/ai/tests/backend/engine.test.ts b/plugins/ai/tests/backend/engine.test.ts index 77beaefad..a86d42049 100644 --- a/plugins/ai/tests/backend/engine.test.ts +++ b/plugins/ai/tests/backend/engine.test.ts @@ -200,4 +200,79 @@ describe("runAgent", () => { expect(events.at(-1)).toMatchObject({ type: "error" }); expect(streamChat.mock.calls.length).toBeLessThanOrEqual(8); }); + + it("emits every message it adds so the caller can store the run", async () => { + handler.mockResolvedValue({ hosts: [] }); + streamChat + .mockReturnValueOnce( + chunks( + { type: "text", text: "Looking" }, + { + type: "tool_call", + call: { id: "c1", name: "list_hosts", arguments: {} }, + }, + { type: "done" }, + ), + ) + .mockReturnValueOnce( + chunks({ type: "text", text: "none" }, { type: "done" }), + ); + + const stored = (await collect()) + .filter((e) => e.type === "message") + .map((e) => e.message); + + expect(stored).toEqual([ + { + role: "assistant", + content: "Looking", + toolCalls: [{ id: "c1", name: "list_hosts", arguments: {} }], + }, + { + role: "tool", + content: JSON.stringify({ hosts: [] }), + toolCallId: "c1", + toolName: "list_hosts", + }, + { role: "assistant", content: "none" }, + ]); + }); + + it("explains a stream the provider cut off", async () => { + streamChat.mockReturnValueOnce( + (async function* () { + yield { type: "text", text: "par" } as ChatChunk; + throw Object.assign(new TypeError("terminated"), { + cause: Object.assign(new Error("Body Timeout Error"), { + code: "UND_ERR_BODY_TIMEOUT", + }), + }); + })(), + ); + + const events = await collect(); + expect(events.at(-1).type).toBe("error"); + expect(events.at(-1).message).toMatch(/stopped sending data/); + }); + + it("stays quiet when the user stopped it", async () => { + const controller = new AbortController(); + streamChat.mockReturnValueOnce( + (async function* () { + yield { type: "text", text: "" } as ChatChunk; + controller.abort(); + throw new Error("This operation was aborted"); + })(), + ); + + const events: any[] = []; + for await (const event of runAgent({ + ...BASE, + history: [], + signal: controller.signal, + })) { + events.push(event); + } + expect(events.some((e) => e.type === "error")).toBe(false); + }); }); diff --git a/plugins/ai/tests/backend/history.test.ts b/plugins/ai/tests/backend/history.test.ts new file mode 100644 index 000000000..7daacfb6a --- /dev/null +++ b/plugins/ai/tests/backend/history.test.ts @@ -0,0 +1,116 @@ +import { describe, expect, it } from "vitest"; +import { toChatHistory, toStoredMessage } from "../../src/backend/history.js"; + +const calls = [ + { id: "c1", name: "list_hosts", arguments: {} }, + { id: "c2", name: "get_host", arguments: { hostId: 1 } }, +]; + +describe("toStoredMessage", () => { + it("keeps the answered call on a tool row", () => { + expect( + toStoredMessage({ + role: "tool", + content: "{}", + toolCallId: "c1", + toolName: "list_hosts", + }), + ).toEqual({ + role: "tool", + content: "{}", + toolCalls: JSON.stringify([{ id: "c1", name: "list_hosts" }]), + }); + }); + + it("stores no tool calls for a plain reply", () => { + expect(toStoredMessage({ role: "assistant", content: "hi" })).toEqual({ + role: "assistant", + content: "hi", + toolCalls: null, + }); + }); +}); + +describe("toChatHistory", () => { + it("round-trips a full run", () => { + const rows = [ + { role: "user", content: "check", toolCalls: null }, + toStoredMessage({ role: "assistant", content: "", toolCalls: calls }), + toStoredMessage({ + role: "tool", + content: '{"hosts":[]}', + toolCallId: "c1", + toolName: "list_hosts", + }), + toStoredMessage({ + role: "tool", + content: '{"host":{}}', + toolCallId: "c2", + toolName: "get_host", + }), + { role: "assistant", content: "done", toolCalls: null }, + ]; + + expect(toChatHistory(rows)).toEqual([ + { role: "user", content: "check" }, + { role: "assistant", content: "", toolCalls: calls }, + { + role: "tool", + content: '{"hosts":[]}', + toolCallId: "c1", + toolName: "list_hosts", + }, + { + role: "tool", + content: '{"host":{}}', + toolCallId: "c2", + toolName: "get_host", + }, + { role: "assistant", content: "done" }, + ]); + }); + + it("fills in results for calls a cut off run never answered", () => { + const rows = [ + { role: "user", content: "check", toolCalls: null }, + toStoredMessage({ role: "assistant", content: "", toolCalls: calls }), + toStoredMessage({ + role: "tool", + content: "{}", + toolCallId: "c1", + toolName: "list_hosts", + }), + { role: "user", content: "continue", toolCalls: null }, + ]; + + const history = toChatHistory(rows); + expect(history.map((m) => m.role)).toEqual([ + "user", + "assistant", + "tool", + "tool", + "user", + ]); + expect(history[3]).toMatchObject({ + toolCallId: "c2", + toolName: "get_host", + }); + expect(JSON.parse(history[3].content).error).toMatch(/interrupted/); + }); + + it("repairs a legacy row saved with tool calls and no results", () => { + const history = toChatHistory([ + { role: "assistant", content: "", toolCalls: JSON.stringify(calls) }, + ]); + expect(history.filter((m) => m.role === "tool")).toHaveLength(2); + }); + + it("drops tool rows that answer nothing", () => { + expect( + toChatHistory([ + { role: "tool", content: "{}", toolCalls: null }, + { role: "user", content: "hi", toolCalls: null }, + ]), + ).toEqual([{ role: "user", content: "hi" }]); + }); +}); diff --git a/plugins/ai/tests/backend/providers/provider-http.test.ts b/plugins/ai/tests/backend/providers/provider-http.test.ts index 4cf7ba642..e40706f4a 100644 --- a/plugins/ai/tests/backend/providers/provider-http.test.ts +++ b/plugins/ai/tests/backend/providers/provider-http.test.ts @@ -1,5 +1,9 @@ import { describe, expect, it, vi } from "vitest"; -import { createProviderFetch } from "../../../src/backend/providers/http.js"; +import { + createProviderFetch, + describeProviderError, +} from "../../../src/backend/providers/http.js"; +import { AiProviderError } from "../../../src/backend/providers/types.js"; import { PRIVATE_DESTINATION_MESSAGE } from "../../../src/backend/egress.js"; describe("createProviderFetch", () => { @@ -62,3 +66,53 @@ describe("createProviderFetch", () => { ).rejects.toThrow("socket hang up"); }); }); + +describe("describeProviderError", () => { + const undiciError = (message: string, cause?: unknown) => + Object.assign(new TypeError(message), { cause }); + + it("explains a body that went quiet instead of a bare terminated", () => { + const error = undiciError( + "terminated", + Object.assign(new Error("Body Timeout Error"), { + code: "UND_ERR_BODY_TIMEOUT", + }), + ); + expect(describeProviderError(error, "fallback")).toMatch( + /stopped sending data/, + ); + }); + + it("explains a refused connection", () => { + const error = undiciError( + "fetch failed", + Object.assign(new Error("connect ECONNREFUSED"), { + code: "ECONNREFUSED", + }), + ); + expect(describeProviderError(error, "fallback")).toMatch(/refused/); + }); + + it("maps a private address refused after DNS to the allowlist hint", () => { + const error = undiciError( + "fetch failed", + new Error("Private destinations are not allowed"), + ); + expect(describeProviderError(error, "fallback")).toBe( + PRIVATE_DESTINATION_MESSAGE, + ); + }); + + it("names the cause of an unfamiliar fetch failure", () => { + const error = undiciError("fetch failed", new Error("something odd")); + expect(describeProviderError(error, "fallback")).toBe( + "The provider request failed: something odd", + ); + }); + + it("keeps a provider error's own message", () => { + expect( + describeProviderError(new AiProviderError("bad key", 401), "fallback"), + ).toBe("bad key"); + }); +}); diff --git a/plugins/ai/tests/frontend/transcript.test.ts b/plugins/ai/tests/frontend/transcript.test.ts new file mode 100644 index 000000000..2c396628f --- /dev/null +++ b/plugins/ai/tests/frontend/transcript.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it } from "vitest"; +import { + buildTimeline, + finishedRunEntries, + userEntry, +} from "../../src/frontend/transcript"; + +describe("finishedRunEntries", () => { + it("keeps the run's tool steps ahead of its reply", () => { + const entries = finishedRunEntries( + 2, + [{ id: "tool-0", name: "list_hosts", arguments: {}, result: [] }], + "done", + ); + expect(entries).toEqual([ + { + kind: "tool", + tool: { + id: "run-2-tool-0", + name: "list_hosts", + arguments: {}, + result: [], + }, + }, + { kind: "message", role: "assistant", content: "done" }, + ]); + }); + + it("adds no empty reply", () => { + expect(finishedRunEntries(0, [], "")).toEqual([]); + }); +}); + +describe("buildTimeline", () => { + it("keeps past steps when a new run starts", () => { + const history = [ + userEntry("check"), + ...finishedRunEntries( + 0, + [{ id: "tool-0", name: "list_hosts", arguments: {} }], + "", + ), + userEntry("continue"), + ]; + const timeline = buildTimeline( + history, + { + tools: [{ id: "tool-0", name: "get_host", arguments: {} }], + assistantText: "hi", + }, + [], + ); + + expect(timeline.map((item) => item.kind)).toEqual([ + "message", + "tool", + "message", + "tool", + "message", + ]); + const tools = timeline.filter((item) => item.kind === "tool"); + expect(tools.map((item) => item.live)).toEqual([false, true]); + expect(new Set(timeline.map((item) => item.key)).size).toBe( + timeline.length, + ); + }); +}); diff --git a/src/backend/plugins/ctx-notify.ts b/src/backend/plugins/ctx-notify.ts index f40d3d8af..fbe4aebe6 100644 --- a/src/backend/plugins/ctx-notify.ts +++ b/src/backend/plugins/ctx-notify.ts @@ -185,6 +185,7 @@ export function createPluginFetch(deps: Deps): PluginFetch { ? { rejectUnauthorized: false } : {}), }, + init.timeoutMs, ); } finally { clearTimeout(timer); diff --git a/src/backend/tests/plugins/ctx-notify.test.ts b/src/backend/tests/plugins/ctx-notify.test.ts index a7ea183df..8a8a00a6a 100644 --- a/src/backend/tests/plugins/ctx-notify.test.ts +++ b/src/backend/tests/plugins/ctx-notify.test.ts @@ -25,6 +25,7 @@ const state = vi.hoisted(() => ({ ]), fetchCalls: [] as Array<{ url: string; allowlist: readonly string[] }>, lastTls: null as Record | null, + lastIdleTimeout: undefined as number | undefined, lastSignal: null as AbortSignal | null, lastRedirect: undefined as RequestRedirect | undefined, })); @@ -65,8 +66,10 @@ vi.mock("../../utils/safe-outbound-fetch.js", () => ({ init: RequestInit, allowlist: readonly string[], tls: Record = {}, + idleTimeoutMs?: number, ) => { state.lastTls = tls; + state.lastIdleTimeout = idleTimeoutMs; state.lastRedirect = init.redirect; state.lastSignal = init.signal ?? null; state.fetchCalls.push({ url, allowlist }); @@ -302,6 +305,12 @@ describe("ctx.fetch", () => { expect(state.lastTls).toEqual({}); }); + it("hands the caller's timeout to the guard for the body idle timeout", async () => { + const ctx = contextFor(["network:outbound"]); + await ctx.fetch("https://example.com/stream", { timeoutMs: 600_000 }); + expect(state.lastIdleTimeout).toBe(600_000); + }); + it("aborts the request, and a streamed body, when the caller's signal fires", async () => { const ctx = contextFor(["network:outbound"]); const controller = new AbortController(); diff --git a/src/backend/tests/utils/safe-outbound-fetch.test.ts b/src/backend/tests/utils/safe-outbound-fetch.test.ts index cd24b1db7..131d48987 100644 --- a/src/backend/tests/utils/safe-outbound-fetch.test.ts +++ b/src/backend/tests/utils/safe-outbound-fetch.test.ts @@ -5,6 +5,7 @@ import type { LookupAddress, LookupOptions } from "dns"; import { createDnsLookupHook, isBlockedAddress, + outboundIdleTimeout, readResponseTextLimited, safeOutboundFetch, } from "../../utils/safe-outbound-fetch.js"; @@ -282,6 +283,17 @@ describe("safeOutboundFetch", () => { }, 10_000); }); +describe("outboundIdleTimeout", () => { + it("keeps undici's 5 minute default as the floor", () => { + expect(outboundIdleTimeout()).toBe(300_000); + expect(outboundIdleTimeout(30_000)).toBe(300_000); + }); + + it("raises the idle timeout for a longer request timeout", () => { + expect(outboundIdleTimeout(600_000)).toBe(600_000); + }); +}); + describe("redirect handling", () => { it.each([undefined, "error", "follow", "manual"] as const)( "never follows redirects with mode %s", diff --git a/src/backend/utils/safe-outbound-fetch.ts b/src/backend/utils/safe-outbound-fetch.ts index 72eb85a1f..f328c35a7 100644 --- a/src/backend/utils/safe-outbound-fetch.ts +++ b/src/backend/utils/safe-outbound-fetch.ts @@ -188,11 +188,19 @@ export async function readResponseTextLimited( return Buffer.concat(chunks, total).toString("utf8"); } +/** undici's own default for both, kept as the floor. */ +const DEFAULT_IDLE_TIMEOUT_MS = 300_000; + +export function outboundIdleTimeout(timeoutMs?: number): number { + return Math.max(timeoutMs ?? 0, DEFAULT_IDLE_TIMEOUT_MS); +} + export async function safeOutboundFetch( rawUrl: string, options: RequestInit, allowedPrivateHosts: readonly string[] = [], tls: OutboundTlsOptions = {}, + idleTimeoutMs?: number, ): Promise { const url = new URL(rawUrl); if ( @@ -226,7 +234,12 @@ export async function safeOutboundFetch( } as never)) as unknown as Response; } + // A slow model can go quiet for minutes mid-stream. Past bodyTimeout undici + // kills the body with a bare "terminated". + const idle = outboundIdleTimeout(idleTimeoutMs); const dispatcher = new Agent({ + headersTimeout: idle, + bodyTimeout: idle, connect: { lookup: createDnsLookupHook(lookup, allowPrivate), ...(tls.ca ? { ca: tls.ca } : {}),