fix(openai): 生命周期元数据之后保留 Responses 故障转移

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）

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,

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/)
+})