fix(protocol): 同线程固定 Claude Code thread betas

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## 未发布
 
 - 面板凭据：选择 Claude Code 官方登录时直接打开槽终端，已有 OAuth 授权链接不再挡住 CC 登录路径。（#622）
+- 协议：同一线程后续 hop 固定首条 user 消息上的 Claude Code beta 列表（含 mid-conversation output config），避免多轮间 beta 漂移。（#624）
 - Responses WebSocket：入站 `response.interrupt`（Codex steer）本地中止在飞 hop 并回 `response.interrupt.accepted`；取消走既有 handleProtocol 终态记账，不再报 `unsupported Responses WebSocket message type` 或网关 `incomplete_response`。上游 interrupt 转发对齐 CPR 后置。（#616）
 - Fable `7d_oi`：reset 已过（含秒级时间戳）不再永久判 rejected；`fableWindowResetAt` / weekly split 用 `parseResetMs`；过期清理纳入 `unified['7d_oi']`，且 `elapsedExtraResetMs` 纳入 `7d_oi` 后会在窗口过期时触发 usage probe；无 reset 时仍按 rejected 处理。（#601）
 - OpenAI 槽：从 OAuth `id_token` 读取 `chatgpt_subscription_active_until`（含 epoch 秒/毫秒归一化为 ISO），刷新时新 token 带 claim 则覆盖到期日、不带则保留旧值；账号页只读展示「订阅到期（以凭证为准）」，缺失显示「—」。`plan_type` 缺失时用 id_token 的 `chatgpt_plan_type` 兜底展示；wham 写入的套餐类型优先。（#609）

test/unit/go-worker-client.test.mjs

@@ -17,6 +17,7 @@ import {
 import { prepareCliHopBody } from '../../src/lib/protocol/outbound-attempt.mjs'
 import { callerAutoCompactWindow, defaultSeedPolicy, sanitizeInboundBody } from '../../src/lib/protocol/seed-policy.mjs'
 import { extractOpenaiUsage } from '../../src/lib/protocol/openai-usage.mjs'
+import { rememberThreadTurn, resetThreadBetasForTest } from '../../src/lib/protocol/thread-betas.mjs'
 import { callRustKernel, rustKernelHealth } from '../../src/lib/transport/rust-kernel-client.mjs'
 import { forwardApi, readApiJson } from '../../src/lib/transport/api-kernel-client.mjs'
 
@@ -166,6 +167,33 @@ test('cli-hop envelope keeps role=system turns that the VM betas do not declare'
   assert.deepEqual(out.body.system, body.system)
 })
 
+test('cli-hop continue reuses the beta list sent by its create turn', () => {
+  resetThreadBetasForTest()
+  const exec = { homeDir: '', vm: { claude: { mode: 'setup-token', scope: 'user:inference' } } }
+  const send = (body) => finalizeWorkerPayload({ body, reqHeaders: {}, exec, identity: null, cliHop: true })
+  const created = send({
+    model: 'claude-opus-5-5',
+    thread: { type: 'create' },
+    messages: [
+      { role: 'user', content: 'go' },
+      { role: 'system', content: 'reminder' },
+    ],
+  })
+  assert.ok(created.body.betas.includes('mid-conversation-system-2026-04-07'))
+  assert.equal(created.thread.type, 'create')
+  rememberThreadTurn('msg_create', created.thread)
+  const continued = send({
+    model: 'claude-opus-5-5',
+    thread: { type: 'continue', previous_message_id: 'msg_create' },
+    messages: [{ role: 'user', content: 'next' }],
+  })
+  assert.deepEqual(continued.body.betas, created.body.betas)
+  assert.equal(continued.headers['anthropic-beta'], created.headers['anthropic-beta'])
+  assert.equal(continued.thread.pin, 'pinned')
+  assert.deepEqual(continued.thread.added, ['mid-conversation-system-2026-04-07'])
+  assert.equal(send({ model: 'claude-opus-5-5', messages: [{ role: 'user', content: 'hi' }] }).thread, null)
+})
+
 test('cli-hop keeps the caller compact window outside settings', () => {
   const inbound = {
     model: 'claude-opus-5-5',
@@ -392,6 +420,70 @@ unixTest('callGoWorker marks null TTL as client-owned cache breakpoints', async
   }
 })
 
+unixTest('callGoWorker records thread betas only after a successful JSON response', async () => {
+  resetThreadBetasForTest()
+  const sent = []
+  const fx = await fixture(async (req, res) => {
+    const chunks = []
+    for await (const chunk of req) chunks.push(chunk)
+    const payload = JSON.parse(Buffer.concat(chunks).toString('utf8'))
+    sent.push(payload.body.betas)
+    res.setHeader('content-type', 'application/json')
+    if (sent.length === 1) {
+      res.statusCode = 400
+      res.end(JSON.stringify({ type: 'error', id: 'msg_failed', error: { message: 'rejected' } }))
+      return
+    }
+    res.end(
+      JSON.stringify({
+        type: 'message',
+        id: `msg_turn${sent.length}`,
+        role: 'assistant',
+        content: [{ type: 'text', text: 'ok' }],
+      }),
+    )
+  })
+  try {
+    const exec = { ...fx.exec, vm: { ...fx.exec.vm, claude: { mode: 'setup-token', scope: 'user:inference' } } }
+    const create = {
+      model: 'claude-opus-5-5',
+      thread: { type: 'create' },
+      messages: [
+        { role: 'user', content: 'go' },
+        { role: 'system', content: 'reminder' },
+      ],
+    }
+    const failed = await callGoWorker({ exec, cliHop: true, body: create })
+    assert.equal(failed.ok, false)
+    const afterFailure = finalizeWorkerPayload({
+      body: {
+        model: 'claude-opus-5-5',
+        thread: { type: 'continue', previous_message_id: 'msg_failed' },
+        messages: [{ role: 'user', content: 'next' }],
+      },
+      exec,
+      cliHop: true,
+    })
+    assert.equal(afterFailure.thread.pin, 'miss')
+    const created = await callGoWorker({ exec, cliHop: true, body: create })
+    assert.equal(created.ok, true)
+    const resumed = await callGoWorker({
+      exec,
+      cliHop: true,
+      body: {
+        model: 'claude-opus-5-5',
+        thread: { type: 'continue', previous_message_id: 'msg_turn2' },
+        messages: [{ role: 'user', content: 'next' }],
+      },
+    })
+    assert.equal(resumed.ok, true)
+    assert.equal(resumed.threadBetas.pin, 'pinned')
+    assert.deepEqual(sent[2], sent[1])
+  } finally {
+    await fx.close()
+  }
+})
+
 unixTest('streamGoWorker accepts message_stop before delayed EOF', async () => {
   const fx = await fixture((req, res) => {
     res.setHeader('content-type', 'text/event-stream')
@@ -417,6 +509,55 @@ unixTest('streamGoWorker accepts message_stop before delayed EOF', async () => {
   }
 })
 
+unixTest('streamGoWorker pins continue betas using the streamed response message id', async () => {
+  resetThreadBetasForTest()
+  const sent = []
+  const fx = await fixture(async (req, res) => {
+    const chunks = []
+    for await (const chunk of req) chunks.push(chunk)
+    sent.push(JSON.parse(Buffer.concat(chunks).toString('utf8')).body.betas)
+    res.setHeader('content-type', 'text/event-stream')
+    writeData(res, {
+      type: 'message_start',
+      message: {
+        type: 'message',
+        id: `msg_turn${sent.length}`,
+        role: 'assistant',
+        model: 'claude-opus-5-5',
+        content: [],
+      },
+    })
+    writeData(res, { type: 'content_block_start', index: 0, content_block: { type: 'text', text: 'ok' } })
+    writeData(res, { type: 'message_delta', delta: { stop_reason: 'end_turn' }, usage: { output_tokens: 1 } })
+    writeData(res, { type: 'message_stop' })
+    res.end()
+  })
+  try {
+    const exec = { ...fx.exec, vm: { ...fx.exec.vm, claude: { mode: 'setup-token', scope: 'user:inference' } } }
+    const hop = (thread, messages) =>
+      streamGoWorker({
+        exec,
+        cliHop: true,
+        body: { model: 'claude-opus-5-5', stream: true, thread, messages },
+      })
+    const created = await hop({ type: 'create' }, [
+      { role: 'user', content: 'go' },
+      { role: 'system', content: 'reminder' },
+    ])
+    assert.equal(created.ok, true)
+    assert.equal(created.threadBetas.type, 'create')
+    const resumed = await hop({ type: 'continue', previous_message_id: 'msg_turn1' }, [
+      { role: 'user', content: 'next' },
+    ])
+    assert.equal(resumed.ok, true)
+    assert.equal(resumed.threadBetas.pin, 'pinned')
+    assert.deepEqual(resumed.threadBetas.added, ['mid-conversation-system-2026-04-07'])
+    assert.deepEqual(sent[1], sent[0])
+  } finally {
+    await fx.close()
+  }
+})
+
 unixTest('streamGoWorker names the aborted hop to the kernel cancel route', async () => {
   let hopId = null
   let cancelled = null