Merge PR #7281: fix(ai): preserve Codex WebSocket chaining for normalized IDs (@roboomp)

This commit is contained in:
can1357
2026-08-01 20:12:44 +02:00
3 changed files with 155 additions and 11 deletions
+4
View File
@@ -2,6 +2,10 @@
## [Unreleased]
### Fixed
- Fixed Codex WebSocket tool-result turns replaying full history when the preceding tool-call ID required Responses API normalization ([#7279](https://github.com/can1357/oh-my-pi/issues/7279)).
## [17.2.3] - 2026-08-01
### Added
@@ -2279,12 +2279,15 @@ class CodexStreamProcessor {
resetCodexWebSocketAppendState(state);
} else {
state.lastRequest = structuredCloneJSON(runtime.requestBodyForState);
if (responseId) {
const replayableResponseItems = sanitizeOpenAIResponsesAssistantHistoryItemsForReplay(
structuredCloneJSON(runtime.nativeOutputItems),
);
if (responseId && replayableResponseItems) {
state.lastResponseId = responseId;
state.lastResponseItems = stripInputItemIds(structuredCloneJSON(runtime.nativeOutputItems));
state.lastResponseItems = replayableResponseItems;
state.canAppend = rawEvent.type === "response.done" || rawEvent.type === "response.completed";
} else {
// Without a response id the append baseline cannot be trusted.
// Without both a response id and replayable output, the append baseline cannot be trusted.
state.canAppend = false;
}
}
@@ -3005,14 +3008,6 @@ export function getOpenAICodexTransportDetails(
};
}
function stripInputItemIds(items: Array<Record<string, unknown>>): InputItem[] {
return items.map(item => {
if (item.id == null) return item as InputItem;
const { id: _id, ...rest } = item;
return rest as InputItem;
});
}
const codexDiagnosticsTextEncoder = new TextEncoder();
function jsonByteLength(value: unknown): number {
@@ -3375,6 +3375,151 @@ describe("openai-codex streaming", () => {
});
});
it("chains websocket tool output after normalizing an oversized call id", async () => {
const tempDir = TempDir.createSync("@pi-codex-stream-");
setAgentDir(tempDir.path());
const sentRequests: Array<Record<string, unknown>> = [];
const longCallId = `call_${"x".repeat(80)}`;
class NormalizedCallIdWebSocket extends MockWebSocket {
constructor(url: string, options?: WsOptions) {
super(url, options);
this.scheduleOpen();
}
send(data: string): void {
sentRequests.push(JSON.parse(data) as Record<string, unknown>);
if (sentRequests.length === 1) {
this.sendJson({
type: "response.output_item.added",
item: {
type: "function_call",
id: "fc_oversized",
call_id: longCallId,
name: "read_file",
arguments: "",
},
});
this.sendJson({
type: "response.output_item.done",
item: {
type: "function_call",
id: "fc_oversized",
call_id: longCallId,
name: "read_file",
arguments: '{"path":"README.md"}',
},
});
this.sendJson({
type: "response.completed",
response: { id: "resp_tool", status: "completed", usage: DEFAULT_USAGE },
});
return;
}
this.emitCodexResponse({
messageId: "msg_done",
responseId: "resp_done",
text: "Done",
terminalType: "response.completed",
});
}
}
global.WebSocket = NormalizedCallIdWebSocket as unknown as typeof WebSocket;
const model = createCodexTestModel("https://chatgpt.com/backend-api");
const providerSessionState = new Map<string, ProviderSessionState>();
const firstUser = { role: "user" as const, content: "Read the file", timestamp: Date.now() };
const options = {
apiKey: createCodexTestToken(),
fetch: vi.fn(async () => {
throw new Error("SSE fallback should not be called");
}) as FetchImpl,
providerSessionState,
sessionId: "ws-normalized-call-id",
};
const firstResponse = await streamOpenAICodexResponses(
model,
{ systemPrompt: ["You are a helpful assistant."], messages: [firstUser] },
options,
).result();
const toolCall = firstResponse.content.find(
(block): block is Extract<(typeof firstResponse.content)[number], { type: "toolCall" }> =>
block.type === "toolCall",
);
if (!toolCall) throw new Error("expected a tool call");
const toolResult = {
role: "toolResult" as const,
toolCallId: toolCall.id,
toolName: toolCall.name,
content: [{ type: "text" as const, text: "file contents" }],
isError: false,
timestamp: Date.now(),
};
await streamOpenAICodexResponses(
model,
{
systemPrompt: ["You are a helpful assistant."],
messages: [firstUser, firstResponse, toolResult],
},
options,
).result();
expect(sentRequests).toHaveLength(2);
expect(sentRequests[1]?.previous_response_id).toBe("resp_tool");
expect(sentRequests[1]?.input).toEqual([
expect.objectContaining({
type: "function_call_output",
output: "file contents",
}),
]);
});
it("does not enable websocket append state for a non-replayable response", async () => {
const tempDir = TempDir.createSync("@pi-codex-stream-");
setAgentDir(tempDir.path());
class NonReplayableResponseWebSocket extends MockWebSocket {
constructor(url: string, options?: WsOptions) {
super(url, options);
this.scheduleOpen();
}
send(): void {
this.sendJson({
type: "response.completed",
response: {
id: "resp_empty",
status: "incomplete",
incomplete_details: { reason: "max_output_tokens" },
usage: DEFAULT_USAGE,
},
});
}
}
global.WebSocket = NonReplayableResponseWebSocket as unknown as typeof WebSocket;
const model = createCodexTestModel("https://chatgpt.com/backend-api");
const providerSessionState = new Map<string, ProviderSessionState>();
const result = await streamOpenAICodexResponses(model, createCodexTestContext(), {
apiKey: createCodexTestToken(),
fetch: vi.fn(async () => {
throw new Error("SSE fallback should not be called");
}) as FetchImpl,
providerSessionState,
sessionId: "ws-non-replayable-response",
}).result();
expect(result.stopReason).toBe("length");
expect(
getOpenAICodexTransportDetails(model, {
providerSessionState,
sessionId: "ws-non-replayable-response",
}).canAppend,
).toBe(false);
});
it("drops a stale terminal frame from the prior response leaking onto a reused websocket", async () => {
const tempDir = TempDir.createSync("@pi-codex-stale-frame-");
setAgentDir(tempDir.path());