fix: pin thread betas and verify live docker memory

docs/DEPLOY.md

@@ -28,7 +28,7 @@ VM2API_DB_SECRET='再一串'
 
 不是一个父容器里多个子进程。
 
-槽位默认内存上限为 `512m`，在面板「规格」或 `routing.json` 的 `vm.memory` 配置（`256m` / `512m` / `1g` / `2g` / `4g` / `8g` / `16g`）。创建时可按槽覆盖，结果写在该槽 `vm.machine.memory`。已不再读取 `KIN_VM_MEMORY`。低于 1G 时，常驻 native host 与临时官方 CLI 同时运行可能 OOM；额度 API 探测本身不应额外启动推理 CLI。已有容器不会因控制面升级自动改内存，可在核对宿主余量后使用 `docker update --memory 1g --memory-swap 1g kin-<槽>`，不重建容器。
+槽位默认内存上限为 `512m`，在面板「规格」或 `routing.json` 的 `vm.memory` 配置（`256m` / `512m` / `1g` / `2g` / `4g` / `8g` / `16g`）。创建时可按槽覆盖，结果写在该槽 `vm.machine.memory`。已不再读取 `KIN_VM_MEMORY`。低于 1G 时，常驻 native host 与临时官方 CLI 同时运行可能 OOM；额度 API 探测本身不应额外启动推理 CLI。已有容器不会因控制面升级自动改内存，可在核对宿主余量后使用面板在线扩容，不重建容器。面板成功响应以 Docker 的内存和 swap 上限、容器内 `memory.max` 三者均匹配为准；内核每次准入按当前 `memory.max` 动态计算阈值，并按请求实际序列化大小限制在途负载。单次请求的 CLI 内存突增仍可能超过 512 MiB，准入不保证任意请求都能在此上限内运行。
 
 KVM 槽需要宿主 `/dev/kvm` 可读写（控制面在容器内时，用一次 `--device /dev/kvm` 的探测容器确认）。runner 降权给槽位用户跑 QEMU 时会带上 `/dev/kvm` 的设备组，属组与槽位主 GID 不同也能用。已有 KVM 槽要更新 `kin-kvm-*` 镜像并重建一次。没有 KVM 时不要在生产打开 `vm.allow_tcg`：那是 TCG 软件模拟，仅供测试。
 

src/lib/protocol/claude-code-betas.mjs

@@ -49,6 +49,14 @@ const FAST_MODE_MODELS = {
   'claude-opus-5': true,
   'claude-opus-5-5': true,
 }
+const MID_OUTPUT_CONFIG_MODELS = {
+  'claude-haiku-5-5': true,
+  'claude-sonnet-5-5': true,
+  'claude-opus-5': true,
+  'claude-opus-5-5': true,
+  'claude-fable-5-1': true,
+  'claude-mythos-5-1': true,
+}
 
 function hasHourCache(node) {
   if (Array.isArray(node)) return node.some(hasHourCache)
@@ -69,6 +77,7 @@ export function withRequestProtocolBetas(tokens = [], body = {}) {
     .replace(/-\d{8}$/, '')
   if (Object.hasOwn(PER_TURN_MODELS, model)) add('per-turn-control-2026-07-01')
   if (Object.hasOwn(MID_TOOL_CHANGE_MODELS, model)) add(BETA_MID_CONVERSATION_TOOL_CHANGES)
+  if (body.thread != null && Object.hasOwn(MID_OUTPUT_CONFIG_MODELS, model)) add(BETA_MID_CONVERSATION_OUTPUT_CONFIG)
   if (body.output_config?.effort != null) add(BETA_EFFORT)
   if (body.thinking?.display === 'updates') add(BETA_THINKING_DISPLAY_UPDATES)
   if (Array.isArray(body.safeguards) && body.safeguards.some((item) => item?.type === 'dangerous_tool_use'))

src/lib/protocol/handle-protocol.mjs

@@ -1558,6 +1558,7 @@ export function createHandleProtocol(deps) {
     logBag.vm_id = result?.vmId || null
     logBag.account_id = result?.accountId || null
     logBag.final_account_id = result?.accountId || null
+    if (result?.threadBetas) logBag.hop_meta = { ...logBag.hop_meta, thread_betas: result.threadBetas }
     // Zero executions is a real 0, not unknown; the VM is the last one that ran.
     logBag.attempt_count = result?.attemptCount ?? 0
     logBag.final_state = result?.finalState || result?.terminalState || null

src/lib/protocol/thread-betas.mjs

@@ -0,0 +1,38 @@
+const TTL_MS = 24 * 60 * 60 * 1000
+const MAX_ENTRIES = 10_000
+
+const pins = new Map()
+
+function sweep(now) {
+  for (const [id, pin] of pins) {
+    if (pins.size <= MAX_ENTRIES && pin.expiresAt > now) break
+    pins.delete(id)
+  }
+}
+
+export function pinThreadBetas(body, betas, { now = Date.now() } = {}) {
+  const thread = body?.thread
+  if (!thread || typeof thread !== 'object') return { betas, turn: null }
+  const turn = { type: String(thread.type || ''), model: String(body.model || ''), betas }
+  if (turn.type !== 'continue') return { betas, turn }
+  const from = String(thread.previous_message_id || '')
+  const pin = from ? pins.get(from) : null
+  if (!pin || pin.expiresAt <= now) return { betas, turn: { ...turn, pin: 'miss', from } }
+  if (pin.model !== turn.model) return { betas, turn: { ...turn, pin: 'model_changed', from } }
+  const added = pin.betas.filter((token) => !betas.includes(token))
+  const extra = betas.filter((token) => !pin.betas.includes(token))
+  const pinned = [...pin.betas, ...extra]
+  return { betas: pinned, turn: { ...turn, betas: pinned, pin: 'pinned', from, added, extra } }
+}
+
+export function rememberThreadTurn(messageId, turn, { now = Date.now() } = {}) {
+  const id = String(messageId || '')
+  if (!id || !turn) return
+  pins.delete(id)
+  pins.set(id, { betas: [...turn.betas], model: turn.model, expiresAt: now + TTL_MS })
+  sweep(now)
+}
+
+export function resetThreadBetasForTest() {
+  pins.clear()
+}

src/lib/transport/go-worker-client.mjs

@@ -5,6 +5,7 @@ import path from 'node:path'
 import { prepareOutboundHeaders } from '../protocol/outbound-attempt.mjs'
 import { sanitizeAnthropicBodyForBetaTokens } from '../protocol/anthropic-policy.mjs'
 import { withRequestProtocolBetas } from '../protocol/claude-code-betas.mjs'
+import { pinThreadBetas, rememberThreadTurn } from '../protocol/thread-betas.mjs'
 import { applyThinkingDisplay, thinkingDisplayMode } from '../protocol/thinking.mjs'
 import { sealClaudeCodeCch } from '../identity/cch.mjs'
 import { credentialModeFromOauth } from '../oauth/credential-mode.mjs'
@@ -448,7 +449,11 @@ export function finalizeWorkerPayload({
   })
   const displayed =
     requestContext?.purpose === 'auto_mode_classifier' ? body : applyThinkingDisplay(body, thinkingDisplayMode(routing))
-  const betas = withRequestProtocolBetas(String(headers?.['anthropic-beta'] || '').split(','), displayed)
+  const thread = pinThreadBetas(
+    displayed,
+    withRequestProtocolBetas(String(headers?.['anthropic-beta'] || '').split(','), displayed),
+  )
+  const betas = thread.betas
   headers['anthropic-beta'] = betas.join(',')
   // Envelope headers are discarded; the existing MessageRequest.betas channel
   // carries request gates to the native CLI, which builds the actual HTTP wire.
@@ -458,6 +463,7 @@ export function finalizeWorkerPayload({
   return {
     headers,
     body: sealClaudeCodeCch(gated, { enabled: routing?.compatibility?.cch_enabled !== false, seal: !cliHop }),
+    thread: thread.turn,
   }
 }
 
@@ -495,7 +501,7 @@ function workerEnvelope({
     cache_ttl: cacheTtl == null ? null : String(cacheTtl),
   }
   dumpSessionEnvelope(envelope)
-  return envelope
+  return { envelope, thread: finalized.thread }
 }
 
 function mockScenario(exec) {
@@ -548,34 +554,35 @@ export async function callGoWorker({
   const requestId = randomUUID()
   const stopCancel = cancelOnAbort(exec, signal, requestId)
   try {
+    const prepared = envelope
+      ? { envelope, thread: null }
+      : workerEnvelope({
+          body,
+          reqHeaders,
+          exec,
+          identity,
+          routing,
+          stream: false,
+          want1m,
+          cacheTtl,
+          preserveCacheBreakpoints,
+          cliHop,
+          requestContext,
+        })
     const response = await workerRequest(exec, {
       method: 'POST',
       requestPath,
-      body: {
-        ...(envelope ||
-          workerEnvelope({
-            body,
-            reqHeaders,
-            exec,
-            identity,
-            routing,
-            stream: false,
-            want1m,
-            cacheTtl,
-            preserveCacheBreakpoints,
-            cliHop,
-            requestContext,
-          })),
-        request_id: requestId,
-      },
+      body: { ...prepared.envelope, request_id: requestId },
       signal,
       timeoutMs,
     })
     const data = await readAll(response)
     const parsed = parseJson(data)
     const headers = mergeRateLimitHeaders(publicHeaders(response.headers))
+    const ok = response.statusCode >= 200 && response.statusCode < 300 && parsed?.type !== 'error'
+    if (ok) rememberThreadTurn(parsed?.id, prepared.thread)
     return restoreKernelErrorStatus({
-      ok: response.statusCode >= 200 && response.statusCode < 300 && parsed?.type !== 'error',
+      ok,
       status: response.statusCode || 0,
       via: 'go-worker',
       body: parsed,
@@ -585,6 +592,7 @@ export async function callGoWorker({
       stopReason: parsed?.stop_reason || null,
       terminalState: headers['x-terminal-state'] || null,
       transportError: false,
+      threadBetas: prepared.thread,
     })
   } catch (error) {
     return {
@@ -715,27 +723,26 @@ export async function streamGoWorker({
   const requestId = randomUUID()
   const stopCancel = cancelOnAbort(exec, signal, requestId)
   try {
+    const prepared = envelope
+      ? { envelope, thread: null }
+      : workerEnvelope({
+          body,
+          reqHeaders,
+          exec,
+          identity,
+          routing,
+          stream: true,
+          deliveryMode,
+          want1m,
+          cacheTtl,
+          preserveCacheBreakpoints,
+          cliHop,
+          requestContext,
+        })
     const response = await workerRequest(exec, {
       method: 'POST',
       requestPath,
-      body: {
-        ...(envelope ||
-          workerEnvelope({
-            body,
-            reqHeaders,
-            exec,
-            identity,
-            routing,
-            stream: true,
-            deliveryMode,
-            want1m,
-            cacheTtl,
-            preserveCacheBreakpoints,
-            cliHop,
-            requestContext,
-          })),
-        request_id: requestId,
-      },
+      body: { ...prepared.envelope, request_id: requestId },
       signal,
       timeoutMs,
       timeoutMode: 'first-byte',
@@ -753,6 +760,7 @@ export async function streamGoWorker({
         committed: false,
         terminalState: headers['x-terminal-state'] || 'error',
         transportError: false,
+        threadBetas: prepared.thread,
       })
     }
     let buffer = ''
@@ -797,6 +805,7 @@ export async function streamGoWorker({
       }
       if (event.type === 'error') lastError = event
       if (event.type === 'message_stop') sawMessageStop = true
+      if (event.type === 'message_start') rememberThreadTurn(event.message?.id, prepared.thread)
       const evUsage = usageFromSseEvent(event)
       if (evUsage) sseUsage = mergeUsage(sseUsage, evUsage)
       if (event.message?.model) sseModel = event.message.model
@@ -891,6 +900,7 @@ export async function streamGoWorker({
         committed,
         terminalState,
         transportError: false,
+        threadBetas: prepared.thread,
       })
     } finally {
       if (idleTimer) clearInterval(idleTimer)

src/lib/vm/docker-slot-memory.mjs

@@ -57,19 +57,34 @@ function dockerExec(vm, argv, execImpl, timeout) {
 
 export function inspectDockerContainerMemory(vm, execImpl = defaultExec) {
   const name = containerName(vm?.id)
-  const r = dockerExec(vm, ['docker', 'inspect', '-f', '{{.HostConfig.Memory}}', name], execImpl, 10_000)
+  const r = dockerExec(
+    vm,
+    ['docker', 'inspect', '-f', '{{.HostConfig.Memory}} {{.HostConfig.MemorySwap}} {{.State.Running}}', name],
+    execImpl,
+    10_000,
+  )
   if (!r.ok) {
     return {
       ok: false,
       code: 'container_inspect_failed',
       message: (r.stderr || r.stdout || 'docker inspect failed').trim(),
     }
   }
-  const formatted = formatDockerMemoryBytes(String(r.stdout || '').trim())
-  if (!formatted) {
+  const [memoryValue, swapValue, runningValue] = String(r.stdout || '')
+    .trim()
+    .split(/\s+/)
+  const formatted = formatDockerMemoryBytes(memoryValue)
+  const swapBytes = Number(swapValue)
+  if (
+    !formatted ||
+    !Number.isSafeInteger(Number(memoryValue)) ||
+    !Number.isSafeInteger(swapBytes) ||
+    swapBytes < -1 ||
+    !['true', 'false'].includes(runningValue)
+  ) {
     return { ok: false, code: 'container_memory_unreadable', message: '无法读取容器内存上限' }
   }
-  return { ok: true, memory: formatted, bytes: Number(String(r.stdout || '').trim()) }
+  return { ok: true, memory: formatted, bytes: Number(memoryValue), swapBytes, running: runningValue === 'true' }
 }
 
 export function validateDockerMemoryRaise(vm, requestedMemory, routing) {
@@ -106,6 +121,11 @@ export function raiseDockerSlotMemory(projectRoot, vm, requestedMemory, { routin
 
   const mem = validated.memory
   const name = containerName(vm.id)
+  const before = inspectDockerContainerMemory(vm, execImpl)
+  if (!before.ok) return before
+  if (!before.running) {
+    return { ok: false, code: 'container_not_running', message: '容器未运行，无法验证在线扩容' }
+  }
   const update = dockerExec(vm, ['docker', 'update', '--memory', mem, '--memory-swap', mem, name], execImpl, 15_000)
   if (!update.ok) {
     const inspected = inspectDockerContainerMemory(vm, execImpl)
@@ -127,11 +147,22 @@ export function raiseDockerSlotMemory(projectRoot, vm, requestedMemory, { routin
       requested: mem,
     }
   }
-  if (memoryMiB(inspected.memory) < memoryMiB(mem)) {
+  const requestedBytes = memoryMiB(mem) * 1024 ** 2
+  if (inspected.bytes !== requestedBytes || inspected.swapBytes !== requestedBytes) {
     return {
       ok: false,
       code: 'memory_diverged',
-      message: `容器内存为 ${inspected.memory}，与请求的 ${mem} 不一致，未写入配置`,
+      message: `容器内存或 swap 上限与请求的 ${mem} 不一致，未写入配置`,
+      requested: mem,
+      container_memory: inspected.memory,
+    }
+  }
+  const cgroup = dockerExec(vm, ['docker', 'exec', name, 'cat', '/sys/fs/cgroup/memory.max'], execImpl, 10_000)
+  if (!cgroup.ok || Number(String(cgroup.stdout || '').trim()) !== requestedBytes) {
+    return {
+      ok: false,
+      code: 'cgroup_memory_diverged',
+      message: '容器实际 cgroup 内存上限与请求不一致，未写入配置',
       requested: mem,
       container_memory: inspected.memory,
     }