Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
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
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- Fixed OpenAI-compatible Chat Completions replay dropping opaque `reasoning_details` between turns.
84 changes: 79 additions & 5 deletions packages/ai/src/providers/openai-completions.ts
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,31 @@ function isImageContentBlock(block: { type: string }): block is ImageContent {
return block.type === "image";
}

const REASONING_DETAILS_SIGNATURE_TYPE = "openai-completions.reasoning_details.v1";

interface ReasoningDetailsSignature {
type: typeof REASONING_DETAILS_SIGNATURE_TYPE;
details: Record<string, unknown>[];
}

function encodeReasoningDetails(details: Record<string, unknown>[]): string {
return JSON.stringify({ type: REASONING_DETAILS_SIGNATURE_TYPE, details } satisfies ReasoningDetailsSignature);
}

function decodeReasoningDetails(signature?: string): Record<string, unknown>[] | undefined {
if (!signature?.startsWith("{")) return undefined;
try {
const parsed = JSON.parse(signature) as Partial<ReasoningDetailsSignature>;
if (parsed.type !== REASONING_DETAILS_SIGNATURE_TYPE || !Array.isArray(parsed.details)) return undefined;
if (parsed.details.some((detail) => !detail || typeof detail !== "object" || Array.isArray(detail))) {
return undefined;
}
return parsed.details as Record<string, unknown>[];
} catch {
return undefined;
}
}

export interface OpenAICompletionsOptions extends StreamOptions {
toolChoice?: "auto" | "none" | "required" | { type: "function"; function: { name: string } };
reasoningEffort?: "minimal" | "low" | "medium" | "high" | "xhigh" | "max";
Expand Down Expand Up @@ -175,6 +200,9 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions", OpenA
let thinkingBlock: ThinkingContent | null = null;
const toolCallBlocksByIndex = new Map<number, StreamingToolCallBlock>();
const toolCallBlocksById = new Map<string, StreamingToolCallBlock>();
const reasoningDetailsByIndex = new Map<number, Record<string, unknown>>();
let nextReasoningDetailsIndex = 0;
let reasoningDetailsBlock: ThinkingContent | null = null;
const blocks = output.content as StreamingBlock[];
const getContentIndex = (block: StreamingBlock) => blocks.indexOf(block);
const finishBlock = (block: StreamingBlock) => {
Expand Down Expand Up @@ -372,15 +400,50 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions", OpenA
const reasoningDetails = (choice.delta as any).reasoning_details;
if (reasoningDetails && Array.isArray(reasoningDetails)) {
for (const detail of reasoningDetails) {
if (detail.type === "reasoning.encrypted" && detail.id && detail.data) {
if (!detail || typeof detail !== "object" || Array.isArray(detail)) continue;
const detailRecord = detail as Record<string, unknown>;
const explicitIndex = typeof detailRecord.index === "number" ? detailRecord.index : undefined;
const index = explicitIndex ?? nextReasoningDetailsIndex;
nextReasoningDetailsIndex = Math.max(nextReasoningDetailsIndex, index + 1);
const previousDetail = reasoningDetailsByIndex.get(index);
const mergedDetail = { ...previousDetail, ...detailRecord };
for (const field of ["text", "summary"] as const) {
const previousFragment = previousDetail?.[field];
const fragment = detailRecord[field];
if (typeof previousFragment === "string" && typeof fragment === "string") {
mergedDetail[field] = previousFragment + fragment;
}
}
reasoningDetailsByIndex.set(index, mergedDetail);
if (
detailRecord.type === "reasoning.encrypted" &&
typeof detailRecord.id === "string" &&
detailRecord.data
) {
const matchingToolCall = output.content.find(
(b) => b.type === "toolCall" && b.id === detail.id,
(b) => b.type === "toolCall" && b.id === detailRecord.id,
) as ToolCall | undefined;
if (matchingToolCall) {
matchingToolCall.thoughtSignature = JSON.stringify(detail);
matchingToolCall.thoughtSignature = JSON.stringify(detailRecord);
}
}
}
if (reasoningDetailsByIndex.size > 0) {
if (!reasoningDetailsBlock) {
reasoningDetailsBlock = { type: "thinking", thinking: "", redacted: true };
blocks.push(reasoningDetailsBlock);
stream.push({
type: "thinking_start",
contentIndex: getContentIndex(reasoningDetailsBlock),
partial: output,
});
}
reasoningDetailsBlock.thinkingSignature = encodeReasoningDetails(
[...reasoningDetailsByIndex.entries()]
.sort(([left], [right]) => left - right)
.map(([, detail]) => detail),
);
}
}
}
}
Expand Down Expand Up @@ -826,8 +889,16 @@ export function convertMessages(
);
const assistantText = assistantTextParts.map((part) => part.text).join("");

const replayReasoningDetails = msg.content
.filter(isThinkingContentBlock)
.flatMap((block) => decodeReasoningDetails(block.thinkingSignature) ?? []);
if (replayReasoningDetails.length > 0) {
(assistantMsg as any).reasoning_details = replayReasoningDetails;
}

const nonEmptyThinkingBlocks = msg.content
.filter(isThinkingContentBlock)
.filter((block) => decodeReasoningDetails(block.thinkingSignature) === undefined)
.filter((block) => block.thinking.trim().length > 0);
if (nonEmptyThinkingBlocks.length > 0) {
if (compat.requiresThinkingAsText) {
Expand Down Expand Up @@ -894,7 +965,7 @@ export function convertMessages(
}
})
.filter(Boolean);
if (reasoningDetails.length > 0) {
if (reasoningDetails.length > 0 && replayReasoningDetails.length === 0) {
(assistantMsg as any).reasoning_details = reasoningDetails;
}
}
Expand All @@ -905,6 +976,9 @@ export function convertMessages(
) {
(assistantMsg as { reasoning_content?: string }).reasoning_content = "";
}
if (replayReasoningDetails.length > 0 && assistantMsg.content === null && !assistantMsg.tool_calls) {
assistantMsg.content = "";
}
// Skip assistant messages that have no content and no tool calls.
// Some providers require "either content or tool_calls, but not none".
// Other providers also don't accept empty assistant messages.
Expand All @@ -914,7 +988,7 @@ export function convertMessages(
content !== null &&
content !== undefined &&
(typeof content === "string" ? content.length > 0 : content.length > 0);
if (!hasContent && !assistantMsg.tool_calls) {
if (!hasContent && !assistantMsg.tool_calls && replayReasoningDetails.length === 0) {
continue;
}
params.push(assistantMsg);
Expand Down
173 changes: 173 additions & 0 deletions packages/ai/test/openai-completions-tool-choice.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -760,6 +760,179 @@ describe("openai-completions tool_choice", () => {
expect(writeCall).not.toHaveProperty("partialArgs");
});

it("round-trips opaque reasoning_details without a matching tool call id", async () => {
const details = [
{
type: "reasoning.summary",
index: 0,
format: "unknown",
summary: "brief plan",
},
{
type: "reasoning.encrypted",
index: 1,
format: "unknown",
id: "rs_not_a_tool_call",
data: "opaque-continuation",
},
];
mockState.chunks = [
{
id: "chatcmpl-reasoning-details",
choices: [{ delta: { reasoning_details: details }, finish_reason: "stop" }],
},
];

const { compat: _compat, ...baseModel } = getModel("openai", "gpt-4o-mini")!;
const model = { ...baseModel, api: "openai-completions" } as const;
const first = await streamSimple(
model,
{
messages: [{ role: "user", content: "Think privately.", timestamp: 1 }],
},
{ apiKey: "test" },
).result();
const opaque = first.content.find((block) => block.type === "thinking" && block.redacted);
expect(opaque).toBeDefined();

mockState.chunks = [
{
id: "chatcmpl-after-replay",
choices: [{ delta: { content: "done" }, finish_reason: "stop" }],
},
];
await streamSimple(
model,
{
messages: [
{ role: "user", content: "Think privately.", timestamp: 1 },
first,
{ role: "user", content: "Continue.", timestamp: 2 },
],
},
{ apiKey: "test" },
).result();

const params = mockState.lastParams as { messages: Array<Record<string, unknown>> };
expect(params.messages[1]?.reasoning_details).toEqual(details);
expect(params.messages[1]?.content).toBe("");
});

it("keeps index-less reasoning details after explicitly indexed details", async () => {
const explicitDetail = {
type: "reasoning.summary",
index: 0,
format: "unknown",
summary: "brief plan",
};
const indexlessDetail = {
type: "reasoning.encrypted",
format: "unknown",
id: "rs_indexless",
data: "opaque-continuation",
};
mockState.chunks = [
{
id: "chatcmpl-reasoning-index-order",
choices: [{ delta: { reasoning_details: [explicitDetail, indexlessDetail] }, finish_reason: "stop" }],
},
];

const { compat: _compat, ...baseModel } = getModel("openai", "gpt-4o-mini")!;
const model = { ...baseModel, api: "openai-completions" } as const;
const first = await streamSimple(
model,
{ messages: [{ role: "user", content: "Think privately.", timestamp: 1 }] },
{ apiKey: "test" },
).result();

mockState.chunks = [
{
id: "chatcmpl-after-index-replay",
choices: [{ delta: { content: "done" }, finish_reason: "stop" }],
},
];
await streamSimple(
model,
{
messages: [
{ role: "user", content: "Think privately.", timestamp: 1 },
first,
{ role: "user", content: "Continue.", timestamp: 2 },
],
},
{ apiKey: "test" },
).result();

const params = mockState.lastParams as { messages: Array<Record<string, unknown>> };
expect(params.messages[1]?.reasoning_details).toEqual([explicitDetail, indexlessDetail]);
});

it("concatenates same-index reasoning detail fragments", async () => {
mockState.chunks = [
{
id: "chatcmpl-reasoning-fragments",
choices: [
{
delta: {
reasoning_details: [
{ type: "reasoning.text", index: 0, format: "unknown", text: "first " },
{ type: "reasoning.summary", index: 1, format: "unknown", summary: "brief " },
],
},
finish_reason: null,
},
],
},
{
id: "chatcmpl-reasoning-fragments",
choices: [
{
delta: {
reasoning_details: [
{ type: "reasoning.text", index: 0, text: "second" },
{ type: "reasoning.summary", index: 1, summary: "plan" },
],
},
finish_reason: "stop",
},
],
},
];

const { compat: _compat, ...baseModel } = getModel("openai", "gpt-4o-mini")!;
const model = { ...baseModel, api: "openai-completions" } as const;
const first = await streamSimple(
model,
{ messages: [{ role: "user", content: "Think privately.", timestamp: 1 }] },
{ apiKey: "test" },
).result();

mockState.chunks = [
{
id: "chatcmpl-after-fragment-replay",
choices: [{ delta: { content: "done" }, finish_reason: "stop" }],
},
];
await streamSimple(
model,
{
messages: [
{ role: "user", content: "Think privately.", timestamp: 1 },
first,
{ role: "user", content: "Continue.", timestamp: 2 },
],
},
{ apiKey: "test" },
).result();

const params = mockState.lastParams as { messages: Array<Record<string, unknown>> };
expect(params.messages[1]?.reasoning_details).toEqual([
{ type: "reasoning.text", index: 0, format: "unknown", text: "first second" },
{ type: "reasoning.summary", index: 1, format: "unknown", summary: "brief plan" },
]);
});

it("does not double-count reasoning tokens in completion usage", async () => {
mockState.chunks = [
{
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- Changed ACP prompt requests to resolve only after all causally admitted subagent and parent work has settled.
19 changes: 17 additions & 2 deletions packages/coding-agent/src/modes/acp/acp-mode.ts
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,8 @@ interface AcpPendingTerminal {
boundary: TurnBoundary;
outcome: "result" | "error";
abort: AbortController;
status?: AgentAutonomousStatus;
turnFailure?: string;
failure?: string;
task?: Promise<void>;
}
Expand Down Expand Up @@ -641,6 +643,8 @@ export async function runAcpModeWithConnection(
if (pending.abort.signal.aborted || session !== entry || entry.pendingTerminal !== pending) return;
const terminalQuiescence = quiescenceMeta(status, liveChildren);
if (terminalQuiescence.outstandingSubagents !== 0) continue;
pending.status = status;
pending.turnFailure = finalFailure;
Comment thread
cursor[bot] marked this conversation as resolved.
Resolved

entry.producer.sealTerminal(pending.promptTurnId);
const autonomous = autonomousMeta(status);
Expand Down Expand Up @@ -869,6 +873,7 @@ export async function runAcpModeWithConnection(
// delivered. This prevents late producer events becoming the next turn.
const promptTurnId = entry.producer.beginPrompt();
let responseBoundaryEmitted = false;
let terminalSettlementCancelled = false;
try {
const { text, images } = promptContent(params.prompt);
const priorMessages = turnBoundary(await connection.getMessages());
Expand Down Expand Up @@ -907,6 +912,7 @@ export async function runAcpModeWithConnection(
return { stopReason: "cancelled" satisfies AcpStopReason };
}
const outcome = failure ? "error" : "result";
let terminalStatus = status;
const observedQuiescence = quiescenceMeta(status, liveChildren);
// The roster is telemetry at the response cut, not proof of terminality:
// a child can publish a terminal status before its result reaches the parent.
Expand Down Expand Up @@ -937,12 +943,21 @@ export async function runAcpModeWithConnection(
const pending: AcpPendingTerminal = { promptTurnId, boundary: priorMessages, outcome, abort };
entry.pendingTerminal = pending;
finalizePendingTerminal(entry, pending);
await pending.task;
terminalSettlementCancelled = abort.signal.aborted;
if (pending.failure) {
throw new Error(`ACP lifecycle reconciliation failed: ${pending.failure}`);
}
if (pending.turnFailure) {
throw new Error(`prime-agent turn failed: ${pending.turnFailure}`);
}
terminalStatus = pending.status ?? status;
}
if (failure) throw new Error(`prime-agent turn failed: ${failure}`);
return {
stopReason: acpStopReason({
cancelled: abort.signal.aborted && !entry.producer.isResponseCommitted(promptTurnId),
autonomous: status,
cancelled: terminalSettlementCancelled,
autonomous: terminalStatus,
}),
};
} catch (error) {
Expand Down
Loading
Loading