admin-valentin/AIClient-2-API
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: 修复 Codex→Claude 流式转换中 content_block_start 被跳过的问题

Browse source 
CodexConverter 是单例，streamParams 和 lastClaudeStreamResponseId 在并发请求间共享。
当多个 Codex→Claude 流同时活跃时，delta 事件（不携带 response.id）会通过共享的
lastClaudeStreamResponseId 解析到错误的流状态，导致 content_block_start 被跳过。

修复：利用 Codex SSE 事件中的 item_id，在 response.output_item.added 时建立
item_id→resId 映射，使 delta 事件能精确关联到自己所属的 response state。
This commit is contained in:
Cishoon 2026-03-08 14:31:31 +08:00
parent ea86844be2
commit 042d9c2004
1 changed files with 29 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
src/converters/strategies/CodexConverter.js
View file

@ -1127,10 +1127,19 @@ export class CodexConverter extends BaseConverter {
toClaudeStreamChunk(chunk, model) {
const type = chunk.type;
// Codex 的多数增量事件不带 response.id，需要将其归并到最近活跃的流状态。
// 初始化 item_id → resId 映射表（用于并发流隔离）
if (!this._claudeItemToResId) {
this._claudeItemToResId = new Map();
}
// Codex 的多数增量事件不带 response.id，需要通过 item_id 映射或兜底逻辑归并到正确的流状态。
let resId = chunk.response?.id;
if (!resId) {
if (this.lastClaudeStreamResponseId && this.streamParams.has(this.lastClaudeStreamResponseId)) {
// 优先通过 item_id 精确匹配到对应的 response（并发安全）
const itemId = chunk.item_id || chunk.item?.id;
if (itemId && this._claudeItemToResId.has(itemId)) {
resId = this._claudeItemToResId.get(itemId);
} else if (this.lastClaudeStreamResponseId && this.streamParams.has(this.lastClaudeStreamResponseId)) {
resId = this.lastClaudeStreamResponseId;
} else if (this.streamParams.size === 1) {
resId = this.streamParams.keys().next().value;
@ -1163,6 +1172,16 @@ export class CodexConverter extends BaseConverter {
const state = this.streamParams.get(resId);
state.lastUpdatedAt = Date.now();
// 捕获 response.output_item.added 事件中的 item_id → resId 映射，
// 使后续 delta 事件能通过 item_id 精确关联到正确的流（并发安全）。
if (type === 'response.output_item.added') {
const itemId = chunk.item?.id;
if (itemId && resId) {
this._claudeItemToResId.set(itemId, resId);
}
return null; // 此事件不产生 Claude 输出
}
if (type === 'response.created') {
state.responseID = chunk.response.id;
this.lastClaudeStreamResponseId = state.responseID;
@ -1285,6 +1304,14 @@ export class CodexConverter extends BaseConverter {
},
{ type: "message_stop" }
);
// 清理 item_id → resId 映射，避免内存泄漏
if (this._claudeItemToResId) {
for (const [itemId, mappedResId] of this._claudeItemToResId.entries()) {
if (mappedResId === resId) {
this._claudeItemToResId.delete(itemId);
}
}
}
this.streamParams.delete(resId);
if (this.lastClaudeStreamResponseId === resId) {
this.lastClaudeStreamResponseId = null;