diff --git a/CHANGELOG.md b/CHANGELOG.md index 54ee935..5d5336e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Restore incremental reasoning on the Provider API OpenAI Responses wire by translating Command Code's non-standard `response.reasoning.delta` events into the canonical `response.reasoning_text.delta` before they reach pi-ai, including valid trailing frames without a final blank line. Reasoning now streams while the reasoning item is still open instead of appearing only at `response.output_item.done`. Unsupported reasoning event shapes remain unchanged and produce a content-free warning in `/commandcode-status`. Reproduced with `deepseek/deepseek-v4.1-flash-fast`; standard Responses events, Chat Completions, Anthropic Messages, and the generate fallback are unchanged. - Add reviewed display pricing for the October catalog additions — `claude-sonnet-5-5`, `gpt-6.1-sol` (with its 272K long-context tier), `deepseek/deepseek-v4.1-flash-fast`, and the free `inclusionai/ling-3.1-flash:free`. Apply the DeepSeek V4 weekday peak-pricing window to `deepseek/deepseek-v4.1-flash-fast`. Models absent from `MODEL_COSTS` silently fall back to a zero display cost, so the snapshot now covers all 85 advertised models. ## 0.7.5 - 2026-10-06 diff --git a/README.md b/README.md index cfb0264..d64010e 100644 --- a/README.md +++ b/README.md @@ -133,7 +133,7 @@ When a valid cache exists, the provider registers the cached catalog immediately While pi is running, use these provider commands without restarting: - `/commandcode-refresh` fetches and re-registers the current model catalog. Overlapping refreshes are coalesced, and a failed refresh keeps the last valid catalog active. -- `/commandcode-status` shows redacted discovery diagnostics, including the source, model count, timestamps, cache path, endpoint, and warning. +- `/commandcode-status` shows redacted discovery diagnostics, including the source, model count, timestamps, cache path, endpoint, and warning. If a Responses reasoning event has an unsupported shape, a separate stream warning remains visible until the extension is reloaded or the host restarts. The event is preserved unchanged; no reasoning text is included in the warning. - `/commandcode-quota` shows your Command Code account usage and quota in a dashboard-style layout: credits remaining and used with a percentage, monthly/purchased/free sources, the current plan, available usage totals, the API key name, and the 5-hour and weekly usage windows. The `commandcode-quota` command reads from the Command Code alpha usage endpoints (the same ones the `cmd` CLI `/usage` command uses): `whoami`, `billing/credits`, `billing/subscriptions`, and `usage/summary`. It authenticates with the same API key the provider already uses. If the command cannot reach those endpoints or an endpoint schema changes, unavailable sections are reported explicitly instead of being displayed as zero usage. Output is plain text (via `ui.notify`) so it works across pi and compatible hosts such as OMP. diff --git a/index.ts b/index.ts index 8e39886..da485c6 100644 --- a/index.ts +++ b/index.ts @@ -38,6 +38,7 @@ import { getApiKey as getOAuthApiKey, login, refreshToken } from "./src/oauth.ts import { normalizeCommandCodeMessage } from "./src/overflow.ts" import { MODEL_COSTS, ZERO_MODEL_COST } from "./src/pricing.ts" import { registerCommandCodeQuota } from "./src/quota-command.ts" +import { normalizeCommandCodeResponsesResponse } from "./src/responses-stream.ts" import { createCommandCodeRuntime } from "./src/runtime.ts" import { createCommandCodeUsageProvider, type UsageProvider } from "./src/usage.ts" import { transcriptReadersFrom, withTranscriptPromptAndTools } from "./src/transcript.ts" @@ -220,19 +221,37 @@ export default async function (pi: ExtensionAPI) { const modelApis = new Map() const resolveModelApi = (modelId: string): CommandCodeApi => modelApis.get(modelId) ?? apiForModelId(modelId) + let streamWarning: string | undefined const transport = createCommandCodeTransportRouter({ createStream: () => new AssistantMessageEventStream(), - streamProvider: (model, context, options) => - streamNativeProvider( + streamProvider: (model, context, options) => { + const wire = resolveModelApi(model.id) + const resolvedOptions: Parameters[2] = + resolveStreamOptions(options) + const nativeOptions = + wire === "openai-responses" + ? { + ...resolvedOptions, + fetch: async (...args: Parameters) => + normalizeCommandCodeResponsesResponse( + await (resolvedOptions?.fetch ?? fetch)(...args), + (warning) => { + streamWarning = warning + }, + ), + } + : resolvedOptions + return streamNativeProvider( { ...model, - api: resolveModelApi(model.id), + api: wire, cost: commandCodeCostRatesAt(model.id, model.cost), compat: model.compatConfig ?? model.compat, }, context, - resolveStreamOptions(options), - ), + nativeOptions, + ) + }, streamGenerate: (model, context, options) => streamGenerate( model, @@ -279,6 +298,7 @@ export default async function (pi: ExtensionAPI) { return createProviderConfig(models, apiBase, transport.stream) }, getTransport: transport.getTransport, + getStreamWarning: () => streamWarning, }) pi.on("session_shutdown", () => { diff --git a/package.json b/package.json index 7aa00cf..3b0b9f6 100644 --- a/package.json +++ b/package.json @@ -29,7 +29,7 @@ "LICENSE" ], "scripts": { - "test": "npm run typecheck && node --test tests/test-release.mjs && tsx tests/test-package-manifest.ts && tsx tests/test-api-key.ts && tsx tests/test-pure-functions.ts && tsx tests/test-models.ts && tsx tests/test-model-metadata-check.ts && tsx tests/test-runtime.ts && tsx tests/test-pricing.ts && tsx tests/test-cost.ts && tsx tests/test-oauth.ts && tsx tests/test-abort.ts && tsx tests/test-overflow.ts && tsx tests/test-stream.ts && tsx tests/test-quota.ts && tsx tests/test-quota-command.ts && tsx tests/test-usage.ts && tsx tests/test-retry.ts && tsx tests/test-transport.ts && node tests/test-pi-isolated.mjs && node tests/test-pi-authenticated.mjs && node tests/test-pi-local.mjs && node tests/test-omp-compat.mjs", + "test": "npm run typecheck && node --test tests/test-release.mjs tests/test-pi-child-cleanup.mjs && tsx tests/test-package-manifest.ts && tsx tests/test-api-key.ts && tsx tests/test-pure-functions.ts && tsx tests/test-models.ts && tsx tests/test-model-metadata-check.ts && tsx tests/test-runtime.ts && tsx tests/test-pricing.ts && tsx tests/test-cost.ts && tsx tests/test-oauth.ts && tsx tests/test-abort.ts && tsx tests/test-overflow.ts && tsx tests/test-stream.ts && tsx tests/test-quota.ts && tsx tests/test-quota-command.ts && tsx tests/test-usage.ts && tsx tests/test-retry.ts && tsx tests/test-transport.ts && node tests/test-pi-isolated.mjs && node tests/test-pi-authenticated.mjs && node tests/test-pi-local.mjs && node tests/test-omp-compat.mjs", "typecheck": "tsc --noEmit", "format:check": "prettier --check '**/*.{ts,mjs,json,md}'", "format": "prettier --write '**/*.{ts,mjs,json,md}'", diff --git a/src/responses-stream.ts b/src/responses-stream.ts new file mode 100644 index 0000000..1c7a2e9 --- /dev/null +++ b/src/responses-stream.ts @@ -0,0 +1,249 @@ +/** + * Normalizes Command Code's non-standard Responses reasoning deltas. + * + * Command Code emits `response.reasoning.delta` while a reasoning item is still + * streaming, but pi-ai 1.0.3 only understands the canonical + * `response.reasoning_text.delta`. Without translation the deltas are dropped + * and the reasoning text only appears at `response.output_item.done`. Rewriting + * the frame in flight restores incremental thinking without touching any other + * Responses event. + */ + +const SSE_MIME_TYPE = "text/event-stream" +const REASONING_DELTA_EVENT = "response.reasoning.delta" +const REASONING_TEXT_DELTA_EVENT = "response.reasoning_text.delta" + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value) +} + +interface SseLine { + content: string + ending: string +} + +/** Splits a complete SSE frame into lines, preserving each original ending. */ +function splitSseLines(frame: string): SseLine[] { + const lines: SseLine[] = [] + let start = 0 + let index = 0 + while (index < frame.length) { + const code = frame.charCodeAt(index) + if (code === 0x0a) { + lines.push({ content: frame.slice(start, index), ending: "\n" }) + index += 1 + start = index + } else if (code === 0x0d) { + if (index + 1 < frame.length && frame.charCodeAt(index + 1) === 0x0a) { + lines.push({ content: frame.slice(start, index), ending: "\r\n" }) + index += 2 + } else { + lines.push({ content: frame.slice(start, index), ending: "\r" }) + index += 1 + } + start = index + } else { + index += 1 + } + } + if (start < frame.length) { + lines.push({ content: frame.slice(start), ending: "" }) + } + return lines +} + +interface SseField { + field: string + value: string +} + +/** Parses one SSE line. Returns undefined for comments and blank lines. */ +function parseSseField(content: string): SseField | undefined { + if (content.startsWith(":")) return undefined + const colon = content.indexOf(":") + if (colon === -1) return { field: content, value: "" } + const field = content.slice(0, colon) + let value = content.slice(colon + 1) + if (value.startsWith(" ")) value = value.slice(1) + return { field, value } +} + +/** + * Rewrites `response.reasoning.delta` frames as `response.reasoning_text.delta` + * frames. Everything else, including malformed JSON, is returned unchanged. + */ +function normalizeResponsesSseFrame(frame: string, onSchemaMismatch: () => void): string { + const lines = splitSseLines(frame) + const dataValues: string[] = [] + for (let index = 0; index < lines.length; index += 1) { + const content = index === 0 ? stripBom(lines[index].content) : lines[index].content + const parsed = parseSseField(content) + if (parsed?.field === "data") dataValues.push(parsed.value) + } + if (dataValues.length === 0) return frame + + let payload: unknown + try { + payload = JSON.parse(dataValues.join("\n")) + } catch { + return frame + } + if (!isRecord(payload)) return frame + if (payload.type !== REASONING_DELTA_EVENT) return frame + if ( + typeof payload.delta !== "string" || + !Number.isInteger(payload.output_index) || + (payload.output_index as number) < 0 + ) { + onSchemaMismatch() + return frame + } + + const replacement = { ...payload, type: REASONING_TEXT_DELTA_EVENT } + const replacementLine = `data: ${JSON.stringify(replacement)}` + const output: string[] = [] + let replacedData = false + for (let index = 0; index < lines.length; index += 1) { + const line = lines[index] + const content = index === 0 ? stripBom(line.content) : line.content + const bom = index === 0 && line.content.startsWith("\uFEFF") ? "\uFEFF" : "" + const parsed = parseSseField(content) + if (parsed?.field === "data") { + if (!replacedData) { + output.push(`${bom}${replacementLine}${line.ending}`) + replacedData = true + } + continue + } + if (parsed?.field === "event" && parsed.value === REASONING_DELTA_EVENT) { + output.push(`${bom}event: ${REASONING_TEXT_DELTA_EVENT}${line.ending}`) + continue + } + output.push(`${line.content}${line.ending}`) + } + return output.join("") +} + +/** Drops a leading UTF-8 BOM for field recognition only, never from output. */ +function stripBom(content: string): string { + return content.startsWith("\uFEFF") ? content.slice(1) : content +} + +/** Stateful splitter that emits one normalized frame per SSE event. */ +class ResponsesSseNormalizer { + private readonly decoder = new TextDecoder("utf-8", { ignoreBOM: true }) + private readonly encoder = new TextEncoder() + private buffer = "" + private scanIndex = 0 + private lineStart = 0 + private warned = false + + constructor(private readonly onWarning?: (message: string) => void) {} + + private normalize(frame: string): string { + return normalizeResponsesSseFrame(frame, () => { + if (this.warned) return + this.warned = true + try { + this.onWarning?.( + "Unsupported Command Code response.reasoning.delta shape; incremental reasoning may be unavailable. Original events were preserved.", + ) + } catch { + // Diagnostics must not interrupt a usable response stream. + } + }) + } + + push(text: string, emit: (frame: string) => void, flush: boolean): void { + this.buffer += text + this.process(emit, flush) + } + + finish(emit: (frame: string) => void): void { + if (this.buffer.length === 0) return + emit(this.normalize(this.buffer)) + this.buffer = "" + this.scanIndex = 0 + this.lineStart = 0 + } + + private process(emit: (frame: string) => void, flush: boolean): void { + while (this.scanIndex < this.buffer.length) { + const code = this.buffer.charCodeAt(this.scanIndex) + if (code === 0x0a) { + this.consumeLine(this.scanIndex + 1, this.scanIndex, emit) + } else if (code === 0x0d) { + const next = this.scanIndex + 1 + if (next >= this.buffer.length) { + if (!flush) return + this.consumeLine(next, this.scanIndex, emit) + } else if (this.buffer.charCodeAt(next) === 0x0a) { + this.consumeLine(next + 1, this.scanIndex, emit) + } else { + this.consumeLine(next, this.scanIndex, emit) + } + } else { + this.scanIndex += 1 + } + } + } + + private consumeLine(end: number, lineEnd: number, emit: (frame: string) => void): void { + if (lineEnd === this.lineStart) { + const frame = this.buffer.slice(0, end) + emit(this.normalize(frame)) + this.buffer = this.buffer.slice(end) + this.scanIndex = 0 + this.lineStart = 0 + return + } + this.lineStart = end + this.scanIndex = end + } + + transform(chunk: Uint8Array, controller: TransformStreamDefaultController): void { + const text = this.decoder.decode(chunk, { stream: true }) + this.push(text, (frame) => controller.enqueue(this.encoder.encode(frame)), false) + } + + flush(controller: TransformStreamDefaultController): void { + this.push( + this.decoder.decode(), + (frame) => controller.enqueue(this.encoder.encode(frame)), + true, + ) + this.finish((frame) => controller.enqueue(this.encoder.encode(frame))) + } +} + +/** + * Wraps a Command Code Responses HTTP response so streaming reasoning deltas + * reach pi-ai. Returns the original response when it cannot be an SSE stream. + */ +export function normalizeCommandCodeResponsesResponse( + response: Response, + onWarning?: (message: string) => void, +): Response { + if (!response.ok) return response + if (response.body === null) return response + const mimeType = response.headers.get("content-type")?.split(";")[0]?.trim().toLowerCase() + if (mimeType !== SSE_MIME_TYPE) return response + + const normalizer = new ResponsesSseNormalizer(onWarning) + const body = response.body.pipeThrough( + new TransformStream({ + transform: (chunk, controller) => normalizer.transform(chunk, controller), + flush: (controller) => normalizer.flush(controller), + }), + ) + + const headers = new Headers(response.headers) + headers.delete("content-length") + headers.delete("content-encoding") + + return new Response(body, { + status: response.status, + statusText: response.statusText, + headers, + }) +} diff --git a/src/runtime.ts b/src/runtime.ts index 549fdd4..afb0ba2 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -31,6 +31,7 @@ export interface CommandCodeRuntimeOptions { loadCachedModels: () => Promise createProviderConfig: (models: readonly CommandCodeModel[]) => TProviderConfig getTransport?: () => "unknown" | "provider" | "generate" + getStreamWarning?: () => string | undefined now?: () => number logWarning?: (message: string) => void } @@ -44,6 +45,7 @@ export interface CommandCodeRuntimeStatus { cachePath: string endpoint: string warning?: string + streamWarning?: string refreshing: boolean } @@ -101,6 +103,9 @@ export function formatCommandCodeStatus(status: CommandCodeRuntimeStatus): strin ] lines.push(`warning: ${status.warning ? redactDiagnosticText(status.warning) : "none"}`) + if (status.streamWarning) { + lines.push(`stream warning: ${redactDiagnosticText(status.streamWarning)}`) + } return lines.join("\n") } @@ -134,6 +139,7 @@ export class CommandCodeRuntime { const status = this.getStatus() - ctx.ui.notify(formatCommandCodeStatus(status), status.warning ? "warning" : "info") + ctx.ui.notify( + formatCommandCodeStatus(status), + status.warning || status.streamWarning ? "warning" : "info", + ) }, }) } diff --git a/tests/test-pi-child-cleanup.mjs b/tests/test-pi-child-cleanup.mjs new file mode 100644 index 0000000..d6db241 --- /dev/null +++ b/tests/test-pi-child-cleanup.mjs @@ -0,0 +1,48 @@ +import assert from "node:assert/strict" +import { EventEmitter } from "node:events" +import { readFileSync } from "node:fs" +import { test } from "node:test" +import { runInNewContext } from "node:vm" + +const source = readFileSync(new URL("./test-pi-local.mjs", import.meta.url), "utf8") +const helpers = source.slice( + source.indexOf("function runPi("), + source.indexOf("async function runRpcQuery("), +) + +test("runPi timeout waits for child close and preserves the timeout result", async () => { + const child = new EventEmitter() + child.stdout = new EventEmitter() + child.stderr = new EventEmitter() + let notifyKilled + const killed = new Promise((resolve) => { + notifyKilled = resolve + }) + child.kill = () => notifyKilled() + const runPi = runInNewContext(`${helpers}\nrunPi`, { + spawn: () => child, + PI_BIN: "mock-pi", + PROJECT_DIR: ".", + env: {}, + setTimeout, + clearTimeout, + }) + + let settled = false + const result = runPi([], 1).then((value) => { + settled = true + return value + }) + try { + await killed + await new Promise((resolve) => setImmediate(resolve)) + assert.equal(settled, false, "timeout must not return before the child closes") + child.stdout.emit("data", Buffer.from("final output")) + } finally { + child.emit("close", 0) + } + const outcome = await result + assert.equal(outcome.code, -1) + assert.equal(outcome.stdout, "final output") + assert.match(outcome.stderr, /TIMEOUT after 1ms/) +}) diff --git a/tests/test-pi-local.mjs b/tests/test-pi-local.mjs index a7096eb..4fcf19a 100644 --- a/tests/test-pi-local.mjs +++ b/tests/test-pi-local.mjs @@ -19,7 +19,7 @@ const EXT_PATH = resolve(PROJECT_DIR, "index.ts") const COMPAT_CALLER_EXT_PATH = resolve(__dirname, "fixtures", "compat-caller-extension.ts") const TEST_MODEL = "gpt-5.4" const CLAUDE_TEST_MODEL = "claude-sonnet-4-6" -const RESPONSES_TEST_MODEL = "gpt-5.6-sol" +const RESPONSES_TEST_MODEL = "deepseek/deepseek-v4.1-flash-fast" function findPiBinary() { if (process.env.PI_BIN) return process.env.PI_BIN @@ -72,6 +72,9 @@ let overflowRequestCount = 0 let modelsDelayMs = 0 let includeRefreshedModel = false let includeResponsesModel = false +// When set, the Responses mock holds the reasoning item open until the test +// resolves this promise, proving deltas arrive before the item completes. +let responsesReasoningGate function modelCatalog() { const data = [ @@ -111,7 +114,7 @@ function modelCatalog() { object: "model", created: 1779824324, owned_by: "command-code", - name: "GPT 5.6 Sol", + name: "DeepSeek V4.1 Flash Fast", context_length: 1_000_000, supported_endpoints: ["/chat/completions", "/responses"], }) @@ -130,7 +133,7 @@ function modelCatalog() { } /** Minimal OpenAI Responses stream: one assistant text message, then a terminal event. */ -function writeResponsesStream(res, text) { +async function writeResponsesStream(res, text) { const item = { type: "message", id: "msg_mock", @@ -138,14 +141,58 @@ function writeResponsesStream(res, text) { status: "completed", content: [{ type: "output_text", text, annotations: [] }], } - const send = (event) => res.write(`data: ${JSON.stringify(event)}\n\n`) + const send = (event, name) => + res.write(`${name ? `event: ${name}\n` : ""}data: ${JSON.stringify(event)}\n\n`) send({ type: "response.created", response: { id: "resp_mock", object: "response", status: "in_progress", output: [] }, }) + const reasoningItem = { + type: "reasoning", + id: "rs_mock", + summary: [], + content: [{ type: "reasoning_text", text: "We think" }], + } + const textIndex = responsesReasoningGate ? 1 : 0 + if (responsesReasoningGate) { + send({ + type: "response.output_item.added", + output_index: 0, + item: { type: "reasoning", id: "rs_mock", summary: [], content: [] }, + }) + send( + { + type: "response.reasoning.delta", + output_index: 0, + content_index: 0, + item_id: "rs_mock", + delta: "We", + }, + "response.reasoning.delta", + ) + send( + { + type: "response.reasoning.delta", + output_index: 0, + content_index: 0, + item_id: "rs_mock", + delta: " think", + }, + "response.reasoning.delta", + ) + await responsesReasoningGate + send({ + type: "response.reasoning.done", + output_index: 0, + content_index: 0, + item_id: "rs_mock", + text: "We think", + }) + send({ type: "response.output_item.done", output_index: 0, item: reasoningItem }) + } send({ type: "response.output_item.added", - output_index: 0, + output_index: textIndex, item: { type: "message", id: "msg_mock", @@ -156,26 +203,26 @@ function writeResponsesStream(res, text) { }) send({ type: "response.output_text.delta", - output_index: 0, + output_index: textIndex, content_index: 0, item_id: "msg_mock", delta: text, }) send({ type: "response.output_text.done", - output_index: 0, + output_index: textIndex, content_index: 0, item_id: "msg_mock", text, }) - send({ type: "response.output_item.done", output_index: 0, item }) + send({ type: "response.output_item.done", output_index: textIndex, item }) send({ type: "response.completed", response: { id: "resp_mock", object: "response", status: "completed", - output: [item], + output: responsesReasoningGate ? [reasoningItem, item] : [item], usage: { input_tokens: 1, output_tokens: 1, @@ -278,7 +325,7 @@ const server = createServer((req, res) => { : "overflow-recovered" : "mock-pi-ok" if (isResponsesRequest) { - writeResponsesStream(res, text) + writeResponsesStream(res, text).catch((error) => res.destroy(error)) return } if (isAnthropicRequest) { @@ -353,10 +400,14 @@ function runPi(args, timeoutOrOptions = 30_000) { env: childEnv, stdio: ["ignore", "pipe", "pipe"], }) + const waitForClose = rpcChildCloseWaiter(child) let stdout = "" let stderr = "" - const timer = setTimeout(() => { + let timedOut = false + const timer = setTimeout(async () => { + timedOut = true child.kill() + await waitForClose() resolve({ code: -1, stdout, @@ -371,16 +422,30 @@ function runPi(args, timeoutOrOptions = 30_000) { }) child.on("close", (code) => { clearTimeout(timer) - resolve({ code, stdout, stderr }) + if (!timedOut) resolve({ code, stdout, stderr }) }) }) } +/** Observe close before shutdown so cleanup never races session-file writes. */ +function rpcChildCloseWaiter(child) { + const closed = new Promise((resolve) => child.once("close", resolve)) + return async () => { + const timer = setTimeout(() => child.kill("SIGKILL"), 5_000) + try { + await closed + } finally { + clearTimeout(timer) + } + } +} + async function runRpcQuery( timeoutMs = 30_000, promptMessage = "say mock token", extraArgs = [], promptFields = {}, + options = {}, ) { const child = spawn( PI_BIN, @@ -393,7 +458,7 @@ async function runRpcQuery( "--provider", "commandcode", "--model", - TEST_MODEL, + options.model ?? TEST_MODEL, ...extraArgs, ], { @@ -403,6 +468,7 @@ async function runRpcQuery( }, ) + const waitForClose = rpcChildCloseWaiter(child) let stdout = "" let stderr = "" let buffer = "" @@ -449,6 +515,7 @@ async function runRpcQuery( try { const event = JSON.parse(trimmed) events.push(event) + options.onEvent?.(event) if (event.type === "response" && event.id === "prompt-1" && event.success === true) { sawPromptAccepted = true } @@ -476,6 +543,7 @@ async function runRpcQuery( }) const ok = await done + await waitForClose() return { ok, stdout, @@ -508,6 +576,7 @@ async function runRpcExtensionCommands(timeoutMs = 30_000) { }, ) + const waitForClose = rpcChildCloseWaiter(child) let buffer = "" let stderr = "" const events = [] @@ -619,6 +688,7 @@ async function runRpcExtensionCommands(timeoutMs = 30_000) { } } finally { child.kill() + await waitForClose() } } @@ -645,6 +715,7 @@ async function runRpcCompatCall(timeoutMs = 30_000) { }, ) + const waitForClose = rpcChildCloseWaiter(child) let buffer = "" let stderr = "" @@ -688,6 +759,7 @@ async function runRpcCompatCall(timeoutMs = 30_000) { return { message: await notification, stderr } } finally { child.kill() + await waitForClose() } } @@ -712,6 +784,7 @@ async function runRpcOverflowRecovery(timeoutMs = 60_000) { }, ) + const waitForClose = rpcChildCloseWaiter(child) let buffer = "" let stderr = "" const events = [] @@ -768,6 +841,7 @@ async function runRpcOverflowRecovery(timeoutMs = 60_000) { }) const outcome = await result + await waitForClose() return { ...outcome, requests: overflowRequestCount, @@ -1092,6 +1166,77 @@ try { assert.equal(lastRequestBody?.model, RESPONSES_TEST_MODEL) assert.equal(lastRequestBody?.stream, true) assert.ok(Array.isArray(lastRequestBody?.input), "Responses request should use the input array") + + console.log("[pi-local] Responses reasoning deltas stream before the item completes") + requestCount = 0 + let releaseResponsesReasoningGate + responsesReasoningGate = new Promise((resolve) => { + releaseResponsesReasoningGate = resolve + }) + const thinkingDeltas = [] + let sawThinkingEndBeforeGate = false + let thinkingEndCount = 0 + try { + const reasoningRpc = await runRpcQuery( + 30_000, + "say mock token", + ["--thinking", "high"], + {}, + { + model: RESPONSES_TEST_MODEL, + onEvent: (event) => { + if (event.type !== "message_update") return + const assistantEvent = event.assistantMessageEvent + if (assistantEvent?.type === "thinking_delta") { + thinkingDeltas.push(assistantEvent.delta) + if (thinkingDeltas.length === 2) releaseResponsesReasoningGate() + } else if (assistantEvent?.type === "thinking_end") { + thinkingEndCount += 1 + if (thinkingDeltas.length < 2) sawThinkingEndBeforeGate = true + } + }, + }, + ) + assert.equal( + reasoningRpc.ok, + true, + JSON.stringify( + { + stderr: reasoningRpc.stderr, + stdout: reasoningRpc.stdout, + events: reasoningRpc.events.slice(-10), + }, + null, + 2, + ), + ) + assert.equal(sawThinkingEndBeforeGate, false) + assert.deepEqual(thinkingDeltas, ["We", " think"]) + assert.equal(thinkingEndCount, 1) + const thinkingEnd = reasoningRpc.events.find( + (event) => + event.type === "message_update" && event.assistantMessageEvent?.type === "thinking_end", + ) + assert.equal(thinkingEnd?.assistantMessageEvent?.content, "We think") + const finalMessage = [...reasoningRpc.events] + .reverse() + .find((event) => event.type === "message_end" && event.message?.role === "assistant") + const thinkingBlock = finalMessage?.message?.content?.find((block) => block.type === "thinking") + assert.equal(thinkingBlock?.thinking, "We think") + assert.equal( + finalMessage?.message?.content?.some( + (block) => block.type === "text" && block.text.includes("mock-pi-ok"), + ), + true, + ) + assert.equal(lastRequestPath, "/provider/v1/responses") + assert.equal(requestCount, 1) + assert.notEqual(finalMessage?.message?.stopReason, "error") + } finally { + releaseResponsesReasoningGate() + responsesReasoningGate = undefined + } + includeResponsesModel = false console.log("[pi-local] Claude request through Anthropic Messages endpoint") @@ -1196,14 +1341,10 @@ try { assert.equal(overflowRpc.sawCompactionRetry, true, JSON.stringify(overflowRpc)) assert.equal(overflowRpc.stderrHasSecrets, false) overflowMode = false - - console.log("[pi-local] PASS") } finally { await new Promise((resolve) => server.close(resolve)) - // RPC children can still flush session files while exiting after SIGTERM, - // so a single rmSync can hit transient ENOTEMPTY even with maxRetries - // (retry behavior on ENOTEMPTY varies across Node versions). Poll with - // fresh attempts instead, but keep a persistent cleanup failure fatal. + // Children have closed; retain fresh removal attempts as a filesystem + // backstop because maxRetries behavior on ENOTEMPTY varies across Node versions. let lastCleanupError for (let attempt = 0; attempt < 30; attempt += 1) { try { @@ -1217,3 +1358,4 @@ try { } if (lastCleanupError) throw lastCleanupError } +console.log("[pi-local] PASS") diff --git a/tests/test-runtime.ts b/tests/test-runtime.ts index 91b411b..8d2d441 100644 --- a/tests/test-runtime.ts +++ b/tests/test-runtime.ts @@ -114,6 +114,37 @@ describe("Command Code runtime", () => { assert.equal(warn.mock.callCount(), 0) }) + it("keeps stream warnings visible across catalog refreshes without replacing catalog warnings", async () => { + const pi = new ExtensionAPITestDouble() + const context = new CommandContext() + let streamWarning: string | undefined + let catalogWarning: string | undefined = "Catalog warning" + const runtime = createCommandCodeRuntime(pi, { + endpoint: "https://api.commandcode.ai/provider/v1/models", + cachePath: "/tmp/commandcode-models.json", + loadModels: async () => loaded([FIRST_MODEL], "live", catalogWarning), + loadCachedModels: async () => [], + createProviderConfig: (models) => ({ models }), + getStreamWarning: () => streamWarning, + }) + await runtime.initialize() + streamWarning = "Unsupported reasoning shape token=user_secret_value" + await runtime.refresh() + assert.equal(runtime.getStatus().warning, "Catalog warning") + await pi.commands.get("commandcode-status")!("", context) + const notification = context.notifications.at(-1)! + assert.equal(notification.type, "warning") + assert.match(notification.message, /stream warning: Unsupported reasoning shape/) + assert.doesNotMatch(notification.message, /user_secret_value/) + assert.match(notification.message, /warning: Catalog warning/) + catalogWarning = undefined + await runtime.refresh() + await pi.commands.get("commandcode-status")!("", context) + assert.equal(context.notifications.at(-1)?.type, "warning") + assert.match(context.notifications.at(-1)?.message ?? "", /stream warning:/) + assert.equal(runtime.getStatus().warning, undefined) + }) + it("registers refresh and status commands and exposes redacted state", async () => { const pi = new ExtensionAPITestDouble() const context = new CommandContext() diff --git a/tests/test-transport.ts b/tests/test-transport.ts index 82e9771..a08ce2b 100644 --- a/tests/test-transport.ts +++ b/tests/test-transport.ts @@ -1,6 +1,7 @@ import assert from "node:assert/strict" import { describe, it } from "node:test" +import { normalizeCommandCodeResponsesResponse } from "../src/responses-stream.ts" import { createCommandCodeTransportRouter } from "../src/transport.ts" import type { AssistantMessageEvent, @@ -245,3 +246,345 @@ describe("Command Code transport router", () => { assert.equal(generateCalls, 0) }) }) + +const encoder = new TextEncoder() + +function sseHeaders(extra: Record = {}): Record { + return { "content-type": "text/event-stream; charset=utf-8", ...extra } +} + +async function readAll(response: Response): Promise { + const reader = response.body!.getReader() + const decoder = new TextDecoder("utf-8", { ignoreBOM: true }) + let output = "" + for (;;) { + const { done, value } = await reader.read() + if (done) break + output += decoder.decode(value, { stream: true }) + } + output += decoder.decode() + return output +} + +async function readWithTimeout( + reader: ReadableStreamDefaultReader, + timeoutMs = 2_000, +): Promise> { + return await Promise.race([ + reader.read(), + new Promise((_, reject) => + setTimeout(() => reject(new Error(`read timed out after ${timeoutMs}ms`)), timeoutMs), + ), + ]) +} + +function findBytes(haystack: Uint8Array, needle: readonly number[]): number { + outer: for (let i = 0; i + needle.length <= haystack.length; i += 1) { + for (let j = 0; j < needle.length; j += 1) { + if (haystack[i + j] !== needle[j]) continue outer + } + return i + } + return -1 +} + +describe("Command Code Responses stream normalizer", () => { + it("emits the canonical event incrementally across split chunks", async () => { + const event = { + type: "response.reasoning.delta", + output_index: 0, + content_index: 0, + item_id: "rs_1", + delta: "наб", + } + const frame = `event: response.reasoning.delta\ndata: ${JSON.stringify(event)}\n\n` + const bytes = encoder.encode(frame) + const typeSplit = findBytes(bytes, [0x72, 0x65, 0x61, 0x73, 0x6f, 0x6e]) + 4 + // Land inside the first Cyrillic character so the decoder must buffer bytes. + const charSplit = findBytes(bytes, [0xd0, 0xbd]) + 1 + + let sourceController!: ReadableStreamDefaultController + const source = new ReadableStream({ + start(controller) { + sourceController = controller + }, + }) + const response = new Response(source, { status: 200, headers: sseHeaders() }) + const normalized = normalizeCommandCodeResponsesResponse(response) + assert.notEqual(normalized, response) + const reader = normalized.body!.getReader() + + sourceController.enqueue(bytes.slice(0, typeSplit)) + sourceController.enqueue(bytes.slice(typeSplit, charSplit)) + sourceController.enqueue(bytes.slice(charSplit)) + + const first = await readWithTimeout(reader) + assert.equal(first.done, false) + const text = new TextDecoder().decode(first.value) + assert.match(text, /^event: response\.reasoning_text\.delta\n/) + const dataLine = text.split("\n").find((line) => line.startsWith("data: ")) + assert.ok(dataLine, `missing data line in ${JSON.stringify(text)}`) + const parsed = JSON.parse(dataLine.slice("data: ".length)) + assert.equal(parsed.type, "response.reasoning_text.delta") + assert.equal(parsed.delta, "наб") + assert.equal(parsed.item_id, "rs_1") + assert.equal(parsed.output_index, 0) + assert.equal(parsed.content_index, 0) + + // The upstream response is still open, yet the delta was already emitted. + sourceController.close() + const end = await readWithTimeout(reader) + assert.equal(end.done, true) + }) + + it("preserves a leading BOM when rewriting event or data lines", async () => { + const alias = { type: "response.reasoning.delta", output_index: 0, delta: "hi" } + const canonical = { ...alias, type: "response.reasoning_text.delta" } + for (const includeEvent of [false, true]) { + const input = + "\uFEFF" + + (includeEvent ? "event: response.reasoning.delta\n" : "") + + `data: ${JSON.stringify(alias)}\n\n` + const expected = + "\uFEFF" + + (includeEvent ? "event: response.reasoning_text.delta\n" : "") + + `data: ${JSON.stringify(canonical)}\n\n` + const response = new Response(input, { headers: sseHeaders() }) + assert.equal(await readAll(normalizeCommandCodeResponsesResponse(response)), expected) + } + }) + + it("recognizes LF, CRLF, and CR boundaries and merges split data", async () => { + const split = (json: string): [string, string] => [json.slice(0, 1), json.slice(1)] + for (const ending of ["\n", "\r\n", "\r"]) { + const alias = { type: "response.reasoning.delta", output_index: 0, delta: "hi" } + const canonical = { + type: "response.reasoning_text.delta", + output_index: 1, + delta: "yo", + item_id: "rs_2", + } + const [head, tail] = split(JSON.stringify(alias)) + const frame = + `event: response.reasoning.delta${ending}` + + `data: ${head}${ending}` + + `data: ${tail}${ending}${ending}` + + `data: ${JSON.stringify(canonical)}${ending}${ending}` + + // Split the first CRLF pair across two chunks so the scanner must wait. + const firstCrlf = frame.indexOf("\r\n") + const chunks = + ending === "\r\n" && firstCrlf !== -1 + ? [frame.slice(0, firstCrlf + 1), frame.slice(firstCrlf + 1)] + : [frame] + + const source = new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)) + controller.close() + }, + }) + const response = new Response(source, { status: 200, headers: sseHeaders() }) + const output = await readAll(normalizeCommandCodeResponsesResponse(response)) + + assert.equal(output.split("response.reasoning.delta").length - 1, 0, output) + assert.equal(output.split("response.reasoning_text.delta").length - 1, 3, output) + assert.ok(output.includes('"delta":"hi"'), output) + assert.ok(output.includes(JSON.stringify(canonical)), output) + } + }) + + it("normalizes trailing reasoning frames without adding a separator", async () => { + const alias = { type: "response.reasoning.delta", output_index: 0, delta: "hi" } + const canonical = { ...alias, type: "response.reasoning_text.delta" } + for (const ending of ["", "\n", "\r\n", "\r"]) { + const input = `event: response.reasoning.delta\ndata: ${JSON.stringify(alias)}${ending}` + const expected = `event: response.reasoning_text.delta\ndata: ${JSON.stringify(canonical)}${ending}` + assert.equal( + await readAll( + normalizeCommandCodeResponsesResponse(new Response(input, { headers: sseHeaders() })), + ), + expected, + ) + } + const truncated = 'data: {"type":"response.reasoning.delta"' + assert.equal( + await readAll( + normalizeCommandCodeResponsesResponse(new Response(truncated, { headers: sseHeaders() })), + ), + truncated, + ) + }) + + it("warns once per response about unsupported reasoning shapes without exposing data", async () => { + const warnings: string[] = [] + const invalid = [ + { type: "response.reasoning.delta", output_index: "0", delta: "private reasoning" }, + { type: "response.reasoning.delta", output_index: -1, delta: "private reasoning" }, + { type: "response.reasoning.delta", output_index: 0, delta: { secret: "private reasoning" } }, + { type: "response.reasoning.delta", output_index: 0 }, + ] + const input = invalid.map((event) => `data: ${JSON.stringify(event)}\n\n`).join("") + const response = new Response(input, { headers: sseHeaders() }) + assert.equal( + await readAll( + normalizeCommandCodeResponsesResponse(response, (warning) => warnings.push(warning)), + ), + input, + ) + assert.equal(warnings.length, 1) + assert.match(warnings[0], /incremental reasoning/) + assert.doesNotMatch(warnings[0], /private reasoning/) + + const trailing = `data: ${JSON.stringify(invalid[0])}` + assert.equal( + await readAll( + normalizeCommandCodeResponsesResponse( + new Response(trailing, { headers: sseHeaders() }), + () => { + throw new Error("diagnostic failure") + }, + ), + ), + trailing, + ) + }) + + it("does not warn for canonical reasoning or unrelated malformed frames", async () => { + const input = + 'data: {"type":"response.reasoning_text.delta","output_index":0,"delta":"hi"}\n\ndata: {not json\n\n' + const warnings: string[] = [] + assert.equal( + await readAll( + normalizeCommandCodeResponsesResponse( + new Response(input, { headers: sseHeaders() }), + (warning) => warnings.push(warning), + ), + ), + input, + ) + assert.deepEqual(warnings, []) + }) + + it("leaves standard and malformed frames untouched", async () => { + const aliasInDelta = { + type: "response.output_text.delta", + output_index: 0, + delta: "response.reasoning.delta", + } + const input = + `data: ${JSON.stringify({ type: "response.output_text.delta", output_index: 0, delta: "x" })}\n\n` + + `data: ${JSON.stringify({ type: "response.reasoning.done", output_index: 0, text: "t" })}\n\n` + + `data: ${JSON.stringify({ type: "response.output_item.done", output_index: 0, item: { type: "reasoning", id: "rs", content: [] } })}\n\n` + + `data: ${JSON.stringify(aliasInDelta)}\n\n` + + `: keep this comment\n\n` + + `data: [DONE]\n\n` + + `data: {not json\n\n` + + `data: ${JSON.stringify({ type: "response.reasoning.delta", output_index: 0 })}\n\n` + + `data: ${JSON.stringify({ type: "response.reasoning.delta", output_index: -1, delta: "x" })}\n\n` + + const source = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(input)) + controller.close() + }, + }) + const response = new Response(source, { status: 200, headers: sseHeaders() }) + const output = await readAll(normalizeCommandCodeResponsesResponse(response)) + assert.equal(output, input) + }) + + it("returns non-SSE responses unchanged without reading the body", async () => { + const errorResponse = new Response("nope", { status: 500 }) + assert.equal(normalizeCommandCodeResponsesResponse(errorResponse), errorResponse) + + const noBody = new Response(null, { status: 200 }) + assert.equal(noBody.body, null) + assert.equal(normalizeCommandCodeResponsesResponse(noBody), noBody) + + let pulled = 0 + const body = new ReadableStream({ + pull(controller) { + pulled += 1 + controller.enqueue(encoder.encode("{}")) + controller.close() + }, + }) + const jsonResponse = new Response(body, { + status: 200, + headers: { "content-type": "application/json" }, + }) + assert.equal(normalizeCommandCodeResponsesResponse(jsonResponse), jsonResponse) + assert.equal(pulled, 0) + }) + + it("preserves status and request metadata while dropping body-encoding headers", async () => { + const source = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode("data: [DONE]\n\n")) + controller.close() + }, + }) + const response = new Response(source, { + status: 200, + headers: sseHeaders({ + "request-id": "req_123", + "content-length": "999", + "content-encoding": "gzip", + }), + }) + const normalized = normalizeCommandCodeResponsesResponse(response) + assert.equal(normalized.status, 200) + assert.equal(normalized.statusText, response.statusText) + assert.equal(normalized.headers.get("request-id"), "req_123") + assert.equal(normalized.headers.get("content-length"), null) + assert.equal(normalized.headers.get("content-encoding"), null) + }) + + it("propagates cancellation to the source stream", async () => { + let resolveCancel!: (reason: unknown) => void + const cancelled = new Promise((resolve) => { + resolveCancel = resolve + }) + const source = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode("data: {}\n\n")) + }, + cancel(reason) { + resolveCancel(reason) + }, + }) + const response = new Response(source, { status: 200, headers: sseHeaders() }) + const reader = normalizeCommandCodeResponsesResponse(response).body!.getReader() + await readWithTimeout(reader) + await reader.cancel("stop") + + const reason = await Promise.race([ + cancelled, + new Promise((_, reject) => + setTimeout(() => reject(new Error("source was not cancelled")), 2_000), + ), + ]) + assert.equal(reason, "stop") + }) + + it("surfaces an input stream error instead of a clean end", async () => { + const failure = new Error("boom") + let first = true + const source = new ReadableStream({ + pull(controller) { + if (first) { + first = false + controller.enqueue(encoder.encode("data: {}\n\n")) + return + } + controller.error(failure) + }, + }) + const response = new Response(source, { status: 200, headers: sseHeaders() }) + const reader = normalizeCommandCodeResponsesResponse(response).body!.getReader() + const chunk = await readWithTimeout(reader) + assert.equal(chunk.done, false) + await assert.rejects(readWithTimeout(reader), /boom/) + }) +})