diff --git a/CHANGELOG.md b/CHANGELOG.md index 152b80e40..30d8900c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## 未发布 +- OpenAI Responses 上游：生命周期元数据（`response.created` / `queued` / `in_progress` / `metadata` / `keepalive`）之后若失败仍可 failover；首条真实输出前不 commit，避免仅 metadata 即锁定 attempt。（#645） + ## 1.3.213 — 2026-10-22 - 官方透传（`compatibility.official_cc_passthrough`、`codex.official_passthrough`）：system/instructions、messages/input、tools、cache_control、采样字段等 body 仍完全透传；outbound session 与 Codex `prompt_cache_key` 恢复 attempt/VM rebuild/mint（不再沿用入站 session 或调用方 cache key）；`metadata.user_id` 继续重绑槽位 device/account。默认开关仍关。（#676） diff --git a/src/lib/pool/api-protocol.mjs b/src/lib/pool/api-protocol.mjs index 1f4ca9a8a..c54f1b43d 100644 --- a/src/lib/pool/api-protocol.mjs +++ b/src/lib/pool/api-protocol.mjs @@ -51,6 +51,89 @@ async function readLines(stream, onLine) { if (buf) await onLine(buf) } +function openAIResponsesSseEvent(line) { + const raw = String(line || '').trim() + if (!raw.startsWith('data:')) return null + const payload = raw.slice(5).trim() + if (!payload || payload === '[DONE]') return null + try { + return JSON.parse(payload) + } catch { + return null + } +} + +function openAIResponsesItemHasOutput(item) { + if (!item || typeof item !== 'object') return false + if (item.arguments || item.input || item.result || item.encrypted_content) return true + for (const value of [item.content, item.summary]) { + if (!Array.isArray(value)) continue + if ( + value.some( + (part) => + part && + typeof part === 'object' && + (part.text || part.transcript || part.arguments || part.input || part.result || part.encrypted_content), + ) + ) + return true + } + return false +} + +/** Lifecycle events are transport metadata, not downstream model output. */ +export function isOpenAIResponsesCommitEvent(event) { + if (!event || typeof event !== 'object') return false + const type = String(event.type || '').trim() + if ( + !type || + type === 'error' || + type === 'response.failed' || + type === 'response.created' || + type === 'response.queued' || + type === 'response.in_progress' || + type === 'response.metadata' || + type === 'keepalive' + ) + return false + if (type === 'response.output_item.added' || type === 'response.output_item.done') + return openAIResponsesItemHasOutput(event.item) + if ( + type === 'response.content_part.added' || + type === 'response.content_part.done' || + type === 'response.reasoning_summary_part.added' || + type === 'response.reasoning_summary_part.done' + ) + return !!(event.part?.text || event.part?.transcript) + if (type === 'response.function_call_arguments.done') return !!event.arguments + if (type === 'response.custom_tool_call_input.done') return !!event.input + if (type === 'response.image_generation_call.partial_image') return !!event.partial_image_b64 + return type.startsWith('response.output_') || (type.startsWith('response.') && /\.(delta|done)$/.test(type)) +} + +function openAIResponsesFailure(event) { + if (!event || typeof event !== 'object') return null + if (event.type === 'error') return event + if (event.type !== 'response.failed') return null + return { + type: 'error', + error: event.response?.error || event.error || { type: 'server_error', message: 'Response failed' }, + } +} + +function mappedOpenAIResponsesFailure(resultBase, event, committed, extras = {}) { + const declared = Number(event?.error?.status || event?.status) + const mapped = mapUpstreamError(declared >= 400 && declared < 600 ? declared : 500, event) + return { + ...resultBase, + ...mapped, + ok: false, + terminalState: committed ? 'incomplete' : 'rejected', + committed, + ...extras, + } +} + export async function runApiInference({ req, res, @@ -386,23 +469,14 @@ async function runOpenAIResponsesUpstream({ let sawTerminal = false let ttftMs = null let streamedUsage = null + let terminalFailure = null + const pendingLines = [] const started = Date.now() - await readLines(upstream, async (line) => { - if (conversionError) return - const raw = String(line || '') - const seen = usageFromSseLine(raw) - if (seen) streamedUsage = seen - if (raw.startsWith('data:') && ttftMs == null) ttftMs = Date.now() - started - if (!committed && raw.startsWith('data:')) committed = true - if (!clientStream) { - chunks.push(raw) - if (/response\.(completed|done)/.test(raw) || raw.trim() === 'data: [DONE]') sawTerminal = true - return - } + const writeStreamLine = (raw) => { if (!res.headersSent) converters.writeSSEHeaders(res) if (protocol === 'openai.responses') { - res.write(raw.endsWith('\n') ? raw : `${raw}\n`) + res.write(raw.endsWith('\n') ? raw : raw + '\n') if (/response\.(completed|done)/.test(raw)) completed = true return } @@ -418,15 +492,45 @@ async function runOpenAIResponsesUpstream({ } catch (error) { if (error.code !== 'tool_arguments_mismatch') throw error conversionError = { type: 'upstream_protocol_error', code: error.code, message: error.message } + const payload = JSON.stringify({ error: conversionError }) res.write( protocol === 'anthropic.messages' - ? `event: error\ndata: ${JSON.stringify({ type: 'error', error: conversionError })}\n\n` - : `data: ${JSON.stringify({ error: conversionError })}\n\ndata: [DONE]\n\n`, + ? 'event: error\ndata: ' + JSON.stringify({ type: 'error', error: conversionError }) + '\n\n' + : 'data: ' + payload + '\n\ndata: [DONE]\n\n', ) } + } + + await readLines(upstream, async (line) => { + if (conversionError) return + const raw = String(line || '') + const event = openAIResponsesSseEvent(raw) + const failure = openAIResponsesFailure(event) + if (failure) terminalFailure = failure + const seen = usageFromSseLine(raw) + if (seen) streamedUsage = seen + const startsOutput = isOpenAIResponsesCommitEvent(event) + const completes = event?.type === 'response.completed' || event?.type === 'response.done' + const wasCommitted = committed + if ((startsOutput || completes) && ttftMs == null) ttftMs = Date.now() - started + if (!committed && (startsOutput || completes)) committed = true + if (!clientStream) { + chunks.push(raw) + if (/response\.(completed|done)/.test(raw) || raw.trim() === 'data: [DONE]') sawTerminal = true + return + } + if (!wasCommitted) { + pendingLines.push(raw) + if (!committed) return + for (const pending of pendingLines.splice(0)) writeStreamLine(pending) + return + } + writeStreamLine(raw) }) if (!clientStream) { + if (terminalFailure) + return mappedOpenAIResponsesFailure(resultBase, terminalFailure, committed, { usage: streamedUsage, ttftMs }) let assembled let outBody try { @@ -443,7 +547,7 @@ async function runOpenAIResponsesUpstream({ status: 502, body: { error: { type: 'upstream_protocol_error', code: error.code, message: error.message } }, terminalState: 'incomplete', - committed: true, + committed, usage: streamedUsage, } } @@ -455,9 +559,11 @@ async function runOpenAIResponsesUpstream({ body: outBody, usage: assembled?.usage || streamedUsage || outBody?.usage || null, terminalState: complete ? 'verified' : hasBody ? 'incomplete' : 'rejected', - committed: true, + committed, } } + if (terminalFailure) + return mappedOpenAIResponsesFailure(resultBase, terminalFailure, committed, { usage: streamedUsage, ttftMs }) if (conversionError) return { ...resultBase, diff --git a/test/unit/api-backend.test.mjs b/test/unit/api-backend.test.mjs index 7c3ed1452..b47f1a790 100644 --- a/test/unit/api-backend.test.mjs +++ b/test/unit/api-backend.test.mjs @@ -10,9 +10,16 @@ import { FailoverRunner } from '../../src/lib/pool/failover-runner.mjs' import { CRS_OFFICIAL_AGENT_PROMPT } from '../../src/lib/identity/crs-persona.mjs' import { sessionIdFromOutboundBody } from '../../src/lib/identity/identity-rewrite.mjs' import { DEFAULT_AGENT_STANDING } from '../../src/lib/identity/persona-template.mjs' -import { resolveInferenceBackend, messagesUrl, runApiInference } from '../../src/lib/pool/api-protocol.mjs' +import { + isOpenAIResponsesCommitEvent, + resolveInferenceBackend, + messagesUrl, + runApiInference, +} from '../../src/lib/pool/api-protocol.mjs' import { createClaudeMessageAssembler, applyClaudeSSELineToMessage } from '../../src/lib/protocol/convert.mjs' +const unixTest = process.platform === 'win32' ? test.skip : test + function fakeResponse() { return { headersSent: false, on() {}, once() {}, off() {}, write() {}, end() {} } } @@ -1419,3 +1426,105 @@ test('Responses API backend returns 502 for conflicting tool arguments when the fs.rmSync(root, { recursive: true, force: true }) } }) +test('OpenAI Responses lifecycle metadata does not commit downstream output', () => { + for (const type of [ + 'response.created', + 'response.queued', + 'response.in_progress', + 'response.metadata', + 'keepalive', + ]) { + assert.equal(isOpenAIResponsesCommitEvent({ type }), false, type) + } + assert.equal(isOpenAIResponsesCommitEvent({ type: 'error', error: { code: 'server_is_overloaded' } }), false) + assert.equal(isOpenAIResponsesCommitEvent({ type: 'response.failed' }), false) + assert.equal( + isOpenAIResponsesCommitEvent({ type: 'response.output_item.added', item: { type: 'reasoning', summary: [] } }), + false, + ) + assert.equal(isOpenAIResponsesCommitEvent({ type: 'response.output_text.delta', delta: 'hello' }), true) +}) + +async function runOpenAIResponsesFixture(events) { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'kin-openai-metadata-')) + const socket = path.join(root, 'run', 'api-kernel.sock') + fs.mkdirSync(path.dirname(socket), { recursive: true }) + const kernel = http.createServer((req, response) => { + req.resume() + req.on('end', () => { + response.writeHead(200, { 'content-type': 'text/event-stream' }) + response.end(events.map((event) => 'data: ' + JSON.stringify(event) + '\n\n').join('') + 'data: [DONE]\n\n') + }) + }) + await new Promise((resolve) => kernel.listen(socket, resolve)) + let wire = '' + const res = { + headersSent: false, + write(chunk) { + wire += chunk + }, + } + try { + const result = await runApiInference({ + cfg: { paths: { data: root } }, + res, + protocol: 'openai.responses', + clientStream: true, + deliveryMode: 'verified', + convertedBody: { model: 'gpt-5.4', messages: [{ role: 'user', content: 'hello' }] }, + inbound: {}, + converters: { + writeSSEHeaders: () => { + res.headersSent = true + }, + }, + scheduler: { + pick: () => ({ + ok: true, + endpoint: { id: 'ep', kind: 'openai', protocol: 'openai', base_url: 'https://fixture.invalid' }, + key: { id: 'key', api_key: 'fixture' }, + upstream_model: 'gpt-5.4', + }), + }, + }) + return { result, wire, headersSent: res.headersSent } + } finally { + kernel.closeAllConnections() + await new Promise((resolve) => kernel.close(resolve)) + fs.rmSync(root, { recursive: true, force: true }) + } +} + +unixTest('API OpenAI metadata followed by overload remains uncommitted for failover', async () => { + const run = await runOpenAIResponsesFixture([ + { type: 'response.created', response: { id: 'resp_meta' } }, + { type: 'response.queued', response: { id: 'resp_meta' } }, + { type: 'response.metadata', response: { id: 'resp_meta' } }, + { type: 'error', error: { type: 'server_error', code: 'server_is_overloaded', status: 503, message: 'busy' } }, + { + type: 'response.failed', + response: { error: { type: 'server_error', code: 'server_is_overloaded', status: 503, message: 'busy' } }, + }, + ]) + assert.equal(run.result.ok, false) + assert.equal(run.result.status, 503) + assert.equal(run.result.committed, false) + assert.equal(run.headersSent, false) + assert.equal(run.wire, '') +}) + +unixTest('API OpenAI flushes buffered metadata with the first real output', async () => { + const run = await runOpenAIResponsesFixture([ + { type: 'response.created', response: { id: 'resp_ok' } }, + { type: 'response.queued', response: { id: 'resp_ok' } }, + { type: 'response.output_text.delta', delta: 'hello' }, + { type: 'response.completed', response: { id: 'resp_ok', output: [] } }, + ]) + assert.equal(run.result.ok, true) + assert.equal(run.result.committed, true) + assert.equal(run.result.terminalState, 'verified') + assert.match(run.wire, /response\.created/) + assert.match(run.wire, /response\.queued/) + assert.match(run.wire, /response\.output_text\.delta/) + assert.match(run.wire, /response\.completed/) +})