diff --git a/packages/ai/src/providers/openai-completions.test.ts b/packages/ai/src/providers/openai-completions.test.ts index 168e6b2695a4..2a3a7543c6d2 100644 --- a/packages/ai/src/providers/openai-completions.test.ts +++ b/packages/ai/src/providers/openai-completions.test.ts @@ -3,11 +3,14 @@ import type { ChatCompletionChunk } from "openai/resources/chat/completions.js"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { configureAiTransportHost } from "../host.js"; import type { Context, Model, SimpleStreamOptions, TextContent } from "../types.js"; +import { onLlmRequestActivity } from "../utils/llm-request-activity.js"; import { SYSTEM_PROMPT_CACHE_BOUNDARY } from "../utils/system-prompt-cache-boundary.js"; type DeepPartial = { [P in keyof T]?: DeepPartial }; type OpenAICompatibleDelta = DeepPartial & { reasoning_content?: string; + reasoning?: string; + reasoning_text?: string; }; type OpenAICompatibleChoice = Omit< DeepPartial, @@ -1430,6 +1433,40 @@ describe("openai-completions stop-reason tool-call guard", () => { expect(result.content.some((block) => block.type === "thinking")).toBe(false); }); + it.each(["reasoning_content", "reasoning", "reasoning_text"] as const)( + "reports hidden %s chunks as request activity", + async (reasoningField) => { + mockChunksRef.chunks = [ + { + id: "chatcmpl-test", + choices: [{ index: 0, delta: { [reasoningField]: "private reasoning" } }], + }, + { + id: "chatcmpl-test", + choices: [{ index: 0, delta: { [reasoningField]: "still private" } }], + }, + makeTextChunk("visible answer"), + makeFinishChunk("stop"), + ]; + const abortController = new AbortController(); + const onActivity = vi.fn(); + const unsubscribe = onLlmRequestActivity(abortController.signal, onActivity); + + try { + const result = await streamOpenAICompletions(model, context, { + apiKey: "sk-test", + signal: abortController.signal, + }).result(); + + expect(onActivity).toHaveBeenCalledTimes(mockChunksRef.chunks.length); + expect(result.content).toContainEqual({ type: "text", text: "visible answer" }); + expect(result.content.some((block) => block.type === "thinking")).toBe(false); + } finally { + unsubscribe(); + } + }, + ); + it("seals the native reasoning block before the answer text begins", async () => { // deepseek streams reasoning_content, then switches to content with no // boundary event; thinking_end must precede the answer so channels do not diff --git a/packages/ai/src/providers/openai-completions.ts b/packages/ai/src/providers/openai-completions.ts index 69f45566815c..eb07a040301c 100644 --- a/packages/ai/src/providers/openai-completions.ts +++ b/packages/ai/src/providers/openai-completions.ts @@ -45,6 +45,7 @@ import { import { AssistantMessageEventStream } from "../utils/event-stream.js"; import { headersToRecord } from "../utils/headers.js"; import { parseStreamingJson } from "../utils/json-parse.js"; +import { notifyLlmRequestActivity } from "../utils/llm-request-activity.js"; import { formatProviderError } from "../utils/provider-error.js"; import { createReasoningTagTextPartitioner } from "../utils/reasoning-tag-text-partitioner.js"; import { @@ -393,6 +394,9 @@ export const streamOpenAICompletions: StreamFunction< continue; } + // Hidden reasoning is still provider progress; keep the idle watchdog alive without exposing it. + notifyLlmRequestActivity(options?.signal); + // OpenAI documents ChatCompletionChunk.id as the unique chat completion identifier, // and each chunk in a streamed completion carries the same id. output.responseId ||= chunk.id; diff --git a/packages/ai/src/transports/openai-completions-transport.ts b/packages/ai/src/transports/openai-completions-transport.ts index 85dea5dd2475..413b2eedb5fd 100644 --- a/packages/ai/src/transports/openai-completions-transport.ts +++ b/packages/ai/src/transports/openai-completions-transport.ts @@ -33,6 +33,7 @@ import { } from "../utils/assistant-text-phase.js"; import { createAssistantMessageEventStream } from "../utils/event-stream.js"; import { parseStreamingJson } from "../utils/json-parse.js"; +import { notifyLlmRequestActivity } from "../utils/llm-request-activity.js"; import { createReasoningTagTextPartitioner } from "../utils/reasoning-tag-text-partitioner.js"; import { createFirstStreamEventAbortController, @@ -631,6 +632,8 @@ async function processOpenAICompletionsStream( await cooperativeScheduler.afterEvent(); continue; } + // Hidden reasoning is still provider progress; keep the idle watchdog alive without exposing it. + notifyLlmRequestActivity(options?.signal); const chunk = rawChunk as ChatCompletionChunk; output.responseId ||= chunk.id; let hasReasoningUsageActivity = false; diff --git a/src/agents/openai-transport-stream.streaming.test.ts b/src/agents/openai-transport-stream.streaming.test.ts index d52886abbd25..3bca5045885d 100644 --- a/src/agents/openai-transport-stream.streaming.test.ts +++ b/src/agents/openai-transport-stream.streaming.test.ts @@ -6,6 +6,7 @@ import { classifyAssistantFailoverReason, formatUserFacingAssistantErrorText, } from "./embedded-agent-helpers.js"; +import { streamWithIdleTimeout } from "./embedded-agent-runner/run/llm-idle-timeout.js"; import { parseTransportChunkUsage, type CapturedStreamEvent, @@ -181,6 +182,111 @@ describe("openai transport stream", () => { } }); + it.each(["reasoning_content", "reasoning"] as const)( + "keeps hidden local %s streams alive beyond the model idle timeout", + async (reasoningField) => { + const reasoningChunkCount = 5; + const reasoningChunkDelayMs = 35; + const idleTimeoutMs = 100; + const server = createServer((req, res) => { + req.resume(); + req.on("end", () => { + res.writeHead(200, { + "content-type": "text/event-stream; charset=utf-8", + "cache-control": "no-cache", + connection: "keep-alive", + }); + + let reasoningChunksSent = 0; + const writeNextChunk = () => { + if (res.destroyed) { + return; + } + if (reasoningChunksSent < reasoningChunkCount) { + reasoningChunksSent += 1; + const reasoningChunk = { + id: "chatcmpl-local-reasoning", + object: "chat.completion.chunk", + created: 1, + model: "nemotron-local", + choices: [ + { + index: 0, + delta: { [reasoningField]: "private reasoning" }, + finish_reason: null, + }, + ], + }; + res.write(`data: ${JSON.stringify(reasoningChunk)}\n\n`); + setTimeout(writeNextChunk, reasoningChunkDelayMs); + return; + } + + res.write( + `data: ${JSON.stringify(makeCompletionsChunk({ role: "assistant", content: "OK" }))}\n\n`, + ); + res.write(`data: ${JSON.stringify(makeCompletionsChunk({}, "stop"))}\n\n`); + res.write("data: [DONE]\n\n"); + res.end(); + }; + + writeNextChunk(); + }); + }); + + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + try { + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("Missing loopback server address"); + } + const model = makeCompletionsModel({ + id: "nemotron-local", + name: "Local Nemotron", + provider: "inference", + baseUrl: `http://127.0.0.1:${address.port}/v1`, + reasoning: false, + }); + const onIdleTimeout = vi.fn(); + const streamFn = streamWithIdleTimeout( + createOpenAICompletionsTransportStreamFn(), + idleTimeoutMs, + onIdleTimeout, + ); + const stream = streamFn( + model, + { + systemPrompt: "system", + messages: [{ role: "user", content: "Reply OK", timestamp: Date.now() }], + tools: [], + } as never, + { apiKey: "test-key" } as never, + ); + + let text = ""; + let thinking = ""; + for await (const event of stream as AsyncIterable<{ type: string; delta?: string }>) { + if (event.type === "text_delta") { + text += event.delta ?? ""; + } + if (event.type === "thinking_delta") { + thinking += event.delta ?? ""; + } + } + + expect(text).toBe("OK"); + expect(thinking).toBe(""); + expect(onIdleTimeout).not.toHaveBeenCalled(); + } finally { + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + } + }, + ); + it("refuses ModelStudio chat streams with no user or assistant payload turns", async () => { const model = makeCompletionsModel({ id: "qwen-coder-plus",