fix: restore incremental Responses reasoning

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

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.

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<string, CommandCodeApi>()
   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<typeof streamNativeProvider>[2] =
+        resolveStreamOptions(options)
+      const nativeOptions =
+        wire === "openai-responses"
+          ? {
+              ...resolvedOptions,
+              fetch: async (...args: Parameters<typeof fetch>) =>
+                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", () => {

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}'",

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<string, unknown> {
+  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<Uint8Array>): void {
+    const text = this.decoder.decode(chunk, { stream: true })
+    this.push(text, (frame) => controller.enqueue(this.encoder.encode(frame)), false)
+  }
+
+  flush(controller: TransformStreamDefaultController<Uint8Array>): 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<Uint8Array, Uint8Array>({
+      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,
+  })
+}

src/runtime.ts

@@ -31,6 +31,7 @@ export interface CommandCodeRuntimeOptions<TProviderConfig> {
   loadCachedModels: () => Promise<readonly CommandCodeModel[]>
   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<TProviderConfig, TContext extends CommandCodeCom
     return {
       ...this.status,
       transport: this.options.getTransport?.() ?? "unknown",
+      streamWarning: this.options.getStreamWarning?.(),
     }
   }
 
@@ -304,7 +310,10 @@ export class CommandCodeRuntime<TProviderConfig, TContext extends CommandCodeCom
       description: "Show redacted Command Code provider diagnostics",
       handler: async (_args, ctx) => {
         const status = this.getStatus()
-        ctx.ui.notify(formatCommandCodeStatus(status), status.warning ? "warning" : "info")
+        ctx.ui.notify(
+          formatCommandCodeStatus(status),
+          status.warning || status.streamWarning ? "warning" : "info",
+        )
       },
     })
   }