Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 26 additions & 1 deletion src/adapters/openai-chat.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import { isCyberPolicyCode } from "../lib/errors";
import { redactSecretString } from "../lib/redact";
import { contentPartsToText } from "./image";
import { identifyRoutedModel } from "./identity";
import { peekReasoningForCall } from "../responses/reasoning-replay-cache";
import { buildNonOpenAIToolCatalogNudgeForTools, shouldInjectNonOpenAIToolCatalogNudge } from "./tool-catalog-nudge";
import { openRouterProviderPayload, resolveOpenRouterRouting } from "../providers/openrouter-routing";
import {
Expand Down Expand Up @@ -324,7 +325,26 @@ function messagesToChatFormat(parsed: OcxParsedRequest, provider: OcxProviderCon
if (textParts.length > 0) {
chatMsg.content = textParts.map(p => p.text).join("");
}
const reasoningContent = thinkingParts.map(p => p.thinking).join("");
let reasoningContent = thinkingParts.map(p => p.thinking).join("");
// History transformations (compaction, lost assistant turn, resumed
// threads) can strip the reasoning item while the tool round survives.
// Re-attach the reasoning the bridge recorded for these call ids so
// preserveReasoningContentModels providers (DeepSeek thinking mode)
// never receive a bare tool-call continuation (issue #950).
if (
reasoningContent.length === 0
&& toolCalls.length > 0
&& modelInList(provider.preserveReasoningContentModels, parsed.modelId)
) {
const cached = toolCalls
.map(tc => (tc.id ? peekReasoningForCall(tc.id) : undefined))
.filter((text): text is string => typeof text === "string" && text.length > 0);
// Parallel calls share one preceding reasoning block, which is
// recorded under every call id — join unique texts only.
if (cached.length > 0) {
reasoningContent = [...new Set(cached)].join("\n");
}
}
if (reasoningContent.length > 0 && modelInList(provider.preserveReasoningContentModels, parsed.modelId)) {
chatMsg.reasoning_content = reasoningContent;
}
Expand Down Expand Up @@ -382,9 +402,14 @@ function messagesToChatFormat(parsed: OcxParsedRequest, provider: OcxProviderCon
// role:"tool" message unless an assistant tool_call with the same id immediately precedes it.
flushPendingToolCalls();
const name = safeToolName(msg.toolName);
// The orphan repair synthesizes an assistant tool call for a result
// whose assistant turn was lost; carry the recorded reasoning so the
// replayed round stays valid for thinking-mode providers (#950).
const cachedReasoning = toolCallId ? peekReasoningForCall(toolCallId) : undefined;
out.push({
role: "assistant",
content: emptyAssistantContent(provider),
...(cachedReasoning ? { reasoning_content: cachedReasoning } : {}),
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
Comment thread
Wibias marked this conversation as resolved.
Resolved
tool_calls: [{
id: toolCallId,
type: "function",
Expand Down
26 changes: 26 additions & 0 deletions src/bridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import type { AdapterEvent, OcxMessagePhase, OcxProviderContinuationState, OcxUs
import { adapterFailureFromMessage, classifyError, CYBER_POLICY_ERROR_CODE, isCyberPolicyCode, type OcxErrorPayload } from "./lib/errors";
import { encodeCompactionSummary } from "./responses/compaction";
import { encodeReasoningEnvelope, type ReasoningEnvelope } from "./responses/reasoning-envelope";
import { rememberReasoningForCall } from "./responses/reasoning-replay-cache";
import { resolveStallTimeoutSec } from "./stall-timeout";
import { usageDisplayTotalTokens } from "./usage/totals";
import {
Expand Down Expand Up @@ -426,8 +427,15 @@ export function bridgeToResponsesSSE(
// encodeReasoningEnvelope: takeReasoningEnvelope's sig/red guard would drop txt-only.
let hiddenRawReasoningText = "";
let hiddenRawReasoningBytes = 0;
// Raw reasoning text flushed most recently, waiting for the tool call it
// preceded. Recorded into the replay cache on tool_call_start so a later
// continuation can re-attach it when history lost the reasoning item
// (issue #950). Kept until new reasoning/text arrives: parallel tool
// calls share the same preceding reasoning block.
let rawReasoningForNextToolCall = "";
const flushHiddenRawReasoning = () => {
if (!hiddenRawReasoningText) return;
rawReasoningForNextToolCall = hiddenRawReasoningText;
const previousBytes = hiddenRawReasoningBytes;
const encrypted = encodeReasoningEnvelope({ txt: hiddenRawReasoningText });
const reservation = budget?.reserveTransient(bytesOf(encrypted), { kind: "reasoning" });
Expand Down Expand Up @@ -519,6 +527,7 @@ export function bridgeToResponsesSSE(

const closeCurrentRawReasoning = () => {
if (!currentRawReasoning) return;
rawReasoningForNextToolCall = currentRawReasoning.text;
const item = {
type: "reasoning", id: currentRawReasoning.itemId, summary: [],
content: [{ type: "reasoning_text", text: currentRawReasoning.text }],
Expand Down Expand Up @@ -779,6 +788,7 @@ export function bridgeToResponsesSSE(
if (currentReasoning) closeCurrentReasoning();
if (currentRawReasoning) closeCurrentRawReasoning();
flushHiddenRawReasoning();
rawReasoningForNextToolCall = "";
if (currentToolCall) closeCurrentToolCall();
flushHiddenReasoningEnvelope();
break;
Expand All @@ -787,6 +797,8 @@ export function bridgeToResponsesSSE(
if (currentReasoning) closeCurrentReasoning();
if (currentRawReasoning) closeCurrentRawReasoning();
flushHiddenRawReasoning();
// Reasoning consumed by a text turn, not a tool call: no cache target.
rawReasoningForNextToolCall = "";
if (currentToolCall) closeCurrentToolCall();
// Only flush on an explicit phase change. A later delta that omits `phase` must
// keep appending to the current message rather than wiping the earlier phase.
Expand Down Expand Up @@ -832,6 +844,7 @@ export function bridgeToResponsesSSE(
if (currentMsg) closeCurrentMessage("commentary");
if (currentRawReasoning) closeCurrentRawReasoning();
flushHiddenRawReasoning();
rawReasoningForNextToolCall = "";
if (currentToolCall) closeCurrentToolCall();
if (!currentReasoning) {
const itemId = `rs_${uuid()}`;
Expand Down Expand Up @@ -905,6 +918,9 @@ export function bridgeToResponsesSSE(
if (currentReasoning) closeCurrentReasoning();
if (currentRawReasoning) closeCurrentRawReasoning();
flushHiddenRawReasoning();
if (rawReasoningForNextToolCall) {
rememberReasoningForCall(event.id, rawReasoningForNextToolCall);
}
if (currentToolCall) closeCurrentToolCall();
const mapped = toolNsMap?.get(event.name);
const realName = mapped?.name ?? event.name;
Expand Down Expand Up @@ -1356,6 +1372,9 @@ function buildResponseJSONWithBudget(
let currentSummaryReasoningBytes = 0;
let currentRawReasoning = "";
let currentRawReasoningBytes = 0;
// Same replay-cache handoff as the streaming path (issue #950): the most
// recently flushed raw reasoning waits for the tool call it preceded.
let rawReasoningForNextToolCall = "";
// Anthropic extended-thinking round-trip (batch): see bridgeToResponsesSSE counterpart.
let batchSignature: string | undefined;
let batchSignatureBytes = 0;
Expand Down Expand Up @@ -1423,6 +1442,7 @@ function buildResponseJSONWithBudget(
};
const flushRawReasoning = () => {
if (!currentRawReasoning) return;
rawReasoningForNextToolCall = currentRawReasoning;
if (options?.hideThinkingSummary === true) {
// Same contract as the streaming path: no visible reasoning, txt-only envelope round-trip.
pushOutput({
Expand Down Expand Up @@ -1480,6 +1500,7 @@ function buildResponseJSONWithBudget(
flushText("commentary");
flushSummaryReasoning();
flushRawReasoning();
rawReasoningForNextToolCall = "";
flushToolCall();
break;
case "text_delta":
Expand All @@ -1488,6 +1509,7 @@ function buildResponseJSONWithBudget(
if (currentText && e.phase !== undefined && currentTextPhase !== e.phase) flushText("commentary");
if (currentSummaryReasoning) flushSummaryReasoning();
if (currentRawReasoning) flushRawReasoning();
rawReasoningForNextToolCall = "";
if (currentToolCallId) flushToolCall();
// Compaction turns keep the summary out of normal message output (replay dedup — see
// bridgeToResponsesSSE); it ships only inside the synthetic compaction item below.
Expand All @@ -1506,6 +1528,7 @@ function buildResponseJSONWithBudget(
case "thinking_delta":
if (currentText) flushText("commentary");
if (currentRawReasoning) flushRawReasoning();
rawReasoningForNextToolCall = "";
if (currentToolCallId) flushToolCall();
{
({ value: currentSummaryReasoning, bytes: currentSummaryReasoningBytes } = appendBatchString(
Expand Down Expand Up @@ -1542,6 +1565,9 @@ function buildResponseJSONWithBudget(
if (currentText) flushText("commentary");
if (currentSummaryReasoning) flushSummaryReasoning();
if (currentRawReasoning) flushRawReasoning();
if (rawReasoningForNextToolCall) {
rememberReasoningForCall(e.id, rawReasoningForNextToolCall);
}
flushToolCall();
currentToolCallId = e.id;
budget?.openCall(e.id);
Expand Down
15 changes: 15 additions & 0 deletions src/images/loop.ts
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,7 @@ function extractIterationThinking(events: AdapterEvent[]): OcxThinkingContent[]
const parts: OcxThinkingContent[] = [];
let thinking = "";
let signature: string | undefined;
let rawReasoning = "";

const flushVisible = () => {
if (!thinking && !signature) return;
Expand All @@ -170,19 +171,33 @@ function extractIterationThinking(events: AdapterEvent[]): OcxThinkingContent[]
thinking = "";
signature = undefined;
};
const flushRaw = () => {
if (!rawReasoning) return;
parts.push({ type: "thinking", thinking: rawReasoning });
rawReasoning = "";
};

for (const e of events) {
if (e.type === "thinking_delta") {
flushRaw();
thinking += e.thinking;
} else if (e.type === "reasoning_raw_delta") {
// OpenAI-compatible providers emit raw reasoning instead of signed
// thinking; DeepSeek thinking mode requires it back alongside replayed
// tool_calls (mirrors src/web-search/loop.ts, issue #950).
flushVisible();
rawReasoning += e.text;
} else if (e.type === "thinking_signature") {
signature = e.signature;
flushVisible();
} else if (e.type === "redacted_thinking") {
flushVisible();
flushRaw();
parts.push({ type: "thinking", thinking: "", redacted: [e.data] });
}
}
flushVisible();
flushRaw();
return parts;
}

Expand Down
34 changes: 32 additions & 2 deletions src/responses/parser.ts
Original file line number Diff line number Diff line change
Expand Up @@ -263,6 +263,34 @@ function findToolById(messages: OcxMessage[], callId: string): { name: string; n
return { name: "" };
}

/**
* Attach pending reasoning to the assistant turn that owns the given call id.
* Reconstructed histories (resume/retry/synthetic) can order a `reasoning`
* item AFTER the `function_call` it belongs to; without this, the pending
* buffer is cleared at the tool output and the turn serializes without
* `reasoning_content`, which DeepSeek thinking mode rejects with HTTP 400
* (issue #950).
*/
function attachPendingReasoningToCallOwner(
messages: OcxMessage[],
callId: string,
pendingReasoning: Array<{ part: OcxThinkingContent; envelopeSigned: boolean }>,
): void {
if (pendingReasoning.length === 0 || !callId) return;
for (let i = messages.length - 1; i >= 0; i--) {
const m = messages[i];
if (m.role !== "assistant") continue;
for (const part of m.content) {
if (part.type === "toolCall" && part.id === callId) {
// Prepend so thinking still precedes tool_use for adapters that require
// that ordering (Anthropic-style replay).
m.content = [...pendingReasoning.map(entry => entry.part), ...m.content];
return;
}
}
}
}

const REASONING_EFFORTS = new Set(["none", "minimal", "low", "medium", "high", "xhigh", "max"]);

export function parseRequest(body: unknown): OcxParsedRequest {
Expand Down Expand Up @@ -545,8 +573,9 @@ export function parseRequest(body: unknown): OcxParsedRequest {
}

if (effectiveType === "function_call_output") {
pendingReasoning.length = 0;
const output = item as { call_id: string; output?: string | unknown[] };
attachPendingReasoningToCallOwner(messages, output.call_id, pendingReasoning);
pendingReasoning.length = 0;
const toolInfo = findToolById(messages, output.call_id);
messages.push({
role: "toolResult", toolCallId: output.call_id,
Expand All @@ -558,8 +587,9 @@ export function parseRequest(body: unknown): OcxParsedRequest {
}

if (effectiveType === "custom_tool_call_output") {
pendingReasoning.length = 0;
const output = item as { call_id: string; output: string | unknown[] };
attachPendingReasoningToCallOwner(messages, output.call_id, pendingReasoning);
pendingReasoning.length = 0;
const toolInfo = findToolById(messages, output.call_id);
messages.push({
role: "toolResult", toolCallId: output.call_id,
Expand Down
83 changes: 83 additions & 0 deletions src/responses/reasoning-replay-cache.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
/**
* In-process fallback store pairing raw reasoning text with the tool call it
* preceded (issue #950).
*
* DeepSeek thinking mode requires the assistant's original `reasoning_content`
* to be replayed on every continuation of a tool-call turn. The bridge records
* the raw reasoning here when it closes a reasoning block and a tool call
* follows; the openai-chat adapter re-attaches it when a `tool_calls`
* assistant message is about to serialize without thinking parts (compacted
* history, lost assistant turn, orphan-repaired tool results).
*
* Privacy: entries hold reasoning text in memory only — never logged,
* serialized, or exported. Bounded by entry count, total bytes, and TTL, so a
* long-lived proxy cannot grow without limit.
*/

const MAX_ENTRIES = 64;
const MAX_TOTAL_BYTES = 256 * 1024;
const TTL_MS = 60 * 60 * 1000;

interface CacheEntry {
text: string;
bytes: number;
at: number;
}

const entries = new Map<string, CacheEntry>();
Comment thread
Wibias marked this conversation as resolved.
Resolved
let totalBytes = 0;
let clockForTests: (() => number) | null = null;

const now = (): number => clockForTests?.() ?? Date.now();

/** Record the raw reasoning text that preceded the given tool call. */
export function rememberReasoningForCall(callId: string, text: string): void {
if (!callId || typeof text !== "string" || text.length === 0) return;
const bytes = Buffer.byteLength(text, "utf8");
// A single entry larger than the whole budget would immediately evict itself.
if (bytes > MAX_TOTAL_BYTES) return;
const at = now();
const previous = entries.get(callId);
if (previous) totalBytes -= previous.bytes;
entries.set(callId, { text, bytes, at });
totalBytes += bytes;
// Evict oldest-first until both caps hold (never evict the entry just written
// while it is the only one — the loop guards on size > 1).
while ((totalBytes > MAX_TOTAL_BYTES || entries.size > MAX_ENTRIES) && entries.size > 1) {
let oldestKey: string | undefined;
let oldestAt = Infinity;
for (const [key, entry] of entries) {
if (entry.at < oldestAt) {
oldestAt = entry.at;
oldestKey = key;
}
}
if (oldestKey === undefined) break;
const evicted = entries.get(oldestKey)!;
totalBytes -= evicted.bytes;
entries.delete(oldestKey);
}
}

/**
* Read the recorded reasoning for a call id without removing it: retries after
* a failed continuation reuse the same fallback.
*/
export function peekReasoningForCall(callId: string): string | undefined {
if (!callId) return undefined;
const entry = entries.get(callId);
if (!entry) return undefined;
if (now() - entry.at > TTL_MS) {
entries.delete(callId);
totalBytes -= entry.bytes;
return undefined;
}
return entry.text;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
}

/** Test-only: reset the cache and optionally pin the clock. */
export function clearReasoningReplayCacheForTests(clock?: (() => number) | null): void {
entries.clear();
totalBytes = 0;
clockForTests = clock ?? null;
}
Loading
Loading