Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,11 @@
- 远端读文件在 BSD 工具链(macOS 节点)上改为先拒绝符号链接与非普通文件再读取;GNU 节点仍用 `dd iflag=nofollow`。
- 面板 Claude 用量:`utilization_*` 超过 1(如 header 1.04)按 ratio×100 显示,不再误成约 1%;Codex `used_percent` 路径不变。(#625,#631)
- 自动/手动备份流水线改到 worker 线程执行(VACUUM INTO + tar),避免主线程长时间阻塞网关导致 Cloudflare 524。(#638,#639)
- cli-hop 出站收成官方两段 system(identity + 调用方正文,ephemeral 无 TTL)、缓存标记打最新 user、去掉 eager_input_streaming/工具 cache_control/temperature/context_management,thinking display 默认 summarized 等。(#641)
- 设置 → 遥测新增日志总开关和保存时间(默认 24 小时,1–168)。这是各槽上传日志的唯一启动源:关闭不写、不显示;打开后槽位运维页直接显示该槽保留窗口内已有记录。sidecar 按同一窗口裁 `telemetry-upload.log`。
- 槽内 cli-node 进入 native loop 前丢掉 `settings.json` 里安全名单以外的 env(如 `LD_PRELOAD`)。
- 授权页默认 OAuth。顺序为 Claude Code、Cookie、授权链接;Setup Token 为授权链接、Cookie。去掉 Console 导入。
- 入站 device 迁移改为 `041_usage_logs_inbound_device`,避开 macOS 节点占用的 040。

## 1.3.202 — 2026-10-20

Expand Down
1 change: 1 addition & 0 deletions src/lib/admin/request-log.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -565,6 +565,7 @@ export class RequestLogStore {
session_id: sessionIdForLog(extra.session_id),
outbound_session_id: sessionIdForLog(extra.outbound_session_id),
reasoning_effort: REASONING_EFFORTS.has(extra.reasoning_effort) ? extra.reasoning_effort : null,
inbound_device_id: sessionIdForLog(extra.inbound_device_id),
// Gate verdict JSON from intercept-gate; the risk audit groups on it.
intercept: extra.intercept ? String(extra.intercept).slice(0, 400) : null,
// sub2api ownership + billing columns
Expand Down
6 changes: 6 additions & 0 deletions src/lib/db/migrations/041_usage_logs_inbound_device.sql
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
-- 041_usage_logs_inbound_device — client device id captured before outbound rewrite.
-- Source: metadata.user_id.device_id, body device_id, or x-kin-device-id.
-- Absent on the inbound request stays NULL. Historical rows stay NULL.

ALTER TABLE usage_logs ADD COLUMN inbound_device_id TEXT;
CREATE INDEX IF NOT EXISTS idx_usage_logs_inbound_device ON usage_logs(inbound_device_id, created_at);
1 change: 1 addition & 0 deletions src/lib/db/repos/usage-logs-repo.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,7 @@ const SUMMARY_COLUMNS = [
'session_id',
'outbound_session_id',
'reasoning_effort',
'inbound_device_id',
'intercept',
'vm_owner_user_id',
]
Expand Down
1 change: 1 addition & 0 deletions src/lib/db/repos/usage-logs-view.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,7 @@ export function toUsageLogRow(row, names, chain = []) {
speed: row.speed ?? null,
context1mApplied: row.long_context == null ? null : row.long_context === 1,
reasoningEffort: row.reasoning_effort ?? null,
inboundDeviceId: row.inbound_device_id ?? null,
errorClass: row.error_class ?? null,
errorLabel: row.error_label ?? null,
errorOwner: row.error_owner ?? null,
Expand Down
4 changes: 2 additions & 2 deletions src/lib/protocol/anthropic-messages.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ import {
} from './sanitize.mjs'
import { sanitizeAnthropicTools } from './web-search.mjs'

export function officialMessagesBody(body = {}, { stream = undefined } = {}) {
export function officialMessagesBody(body = {}, { stream = undefined, preserveContentShape = false } = {}) {
const out = copyOfficialAnthropicFields(body)
applyStructuredOutput(out, body)
normalizeAnthropicMessages(out)
Expand All @@ -26,7 +26,7 @@ export function officialMessagesBody(body = {}, { stream = undefined } = {}) {
out.stop_sequences = Array.isArray(out.stop) ? out.stop : [out.stop]
delete out.stop
}
return canonicalizeClaudeMessagesShape(out)
return canonicalizeClaudeMessagesShape(out, { preserveContentShape })
}

const DISABLED = {
Expand Down
7 changes: 7 additions & 0 deletions src/lib/protocol/cache-prefix.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -210,3 +210,10 @@ export function confirmCachePrefix(sessionKey, now = Date.now()) {
export function clearCachePrefixSessions() {
sessions.clear()
}

/** Human-readable location of a cache-prefix break for logs. */
export function formatCacheBreak(prefix) {
const br = prefix?.break
if (!br?.section) return ''
return br.section === 'messages' ? `messages[${br.index}]` : br.section
}
5 changes: 3 additions & 2 deletions src/lib/protocol/cache-ttl.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -601,10 +601,11 @@ function dropMessageBreakpoints(messages) {
function stampMessageTail(messages, idx, ttl) {
const message = messages[idx]
if (!message || typeof message !== 'object') return messages
const marker = ttl == null ? { type: 'ephemeral' } : { type: 'ephemeral', ttl }
const content = message.content
if (typeof content === 'string') {
const next = messages.slice()
next[idx] = { ...message, content: [{ type: 'text', text: content, cache_control: { type: 'ephemeral', ttl } }] }
next[idx] = { ...message, content: [{ type: 'text', text: content, cache_control: marker }] }
return next
}
if (!Array.isArray(content) || content.length === 0) return messages
Expand All @@ -615,7 +616,7 @@ function stampMessageTail(messages, idx, ttl) {
break
}
if (target < 0) return messages
const stamped = stampNode(content[target], ttl)
const stamped = ttl == null ? { ...content[target], cache_control: marker } : stampNode(content[target], ttl)
if (stamped === content[target]) return messages
const nextContent = content.slice()
nextContent[target] = stamped
Expand Down
5 changes: 4 additions & 1 deletion src/lib/protocol/convert.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -290,7 +290,10 @@ export function normalizeOpenAIToolsOnMessages(body) {
export function toClaudeMessages(protocol, body, opts = { rewrite: false, model_map: true }) {
if (protocol === 'anthropic.messages' && !opts.rewrite) {
const normalized = normalizeOpenAIToolsOnMessages(body)
const out = sanitizeAnthropicBody(normalized, { strictPassthrough: opts.strict_passthrough })
const out = sanitizeAnthropicBody(normalized, {
strictPassthrough: opts.strict_passthrough,
preserveContentShape: opts.preserve_content_shape === true,
})
if (!out.max_tokens) out.max_tokens = defaultMaxTokensForModel(out.model)
return { claude: withUpstreamStream(out), mode: inboundHasOpenAIToolShape(body) ? 'convert' : 'passthrough' }
}
Expand Down
11 changes: 4 additions & 7 deletions src/lib/protocol/eager-tool-streaming.mjs
Original file line number Diff line number Diff line change
@@ -1,16 +1,13 @@
/**
* routing failover.eager_tool_streaming (default on): mark every custom tool
* eager_input_streaming, the shape official Claude Code sends when its
* fine-grained tool streaming is on. Without it upstream buffers a whole Write
* content and stays silent for minutes, which trips the stream idle limit.
* Official Claude Code only enables it on a first-party base URL, so relayed
* clients never send it. Explicit false disables; a caller's own boolean wins;
* routing failover.eager_tool_streaming (default off): mark custom tools
* eager_input_streaming only when the panel turns it on. Official Claude Code
* omits the field. Explicit false stays off; a caller's own boolean wins;
* server tools are left alone.
*/
import { isAnthropicServerTool, isWebSearchTool } from './web-search.mjs'

export function eagerToolStreamingEnabled(routing) {
return routing?.failover?.eager_tool_streaming !== false
return routing?.failover?.eager_tool_streaming === true
}

export function callerEagerInputStreaming(tool) {
Expand Down
24 changes: 12 additions & 12 deletions src/lib/protocol/handle-protocol.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ import {
fillMissingCacheTtl,
validateCacheBreakpoints,
} from './cache-ttl.mjs'
import { confirmCachePrefix, describeCacheContinuity, trackCachePrefix } from './cache-prefix.mjs'
import { confirmCachePrefix, describeCacheContinuity, formatCacheBreak, trackCachePrefix } from './cache-prefix.mjs'
import { ensureClaudeWebSearch, shouldInjectClaudeWebSearch } from './web-search.mjs'
import { dispatchStreamInference } from '../transport/kernel-router.mjs'
import { syncClaudeKernelConfigsFromFile } from '../transport/rust-kernel-supervisor.mjs'
Expand Down Expand Up @@ -829,10 +829,14 @@ export function createHandleProtocol(deps) {

const hdrRewrite = String(req.headers['x-kin-rewrite'] || '') === '1'
const rewriteEnabled = cfg.rewrite.enabled || hdrRewrite
const officialInbound =
isOfficialClaudeCodeTraffic(req.headers, inbound) ||
(detectProxiedOfficialCcFromRoutingFile(routingConfigPath) && isProxiedOfficialClaudeCode(inbound, req.headers))
const converted = toClaudeMessages(protocol, ctx.body, {
rewrite: rewriteEnabled,
model_map: false,
strict_passthrough: String(req.headers['x-kin-strict-passthrough'] || '') === '1',
preserve_content_shape: officialInbound,
})
stats.requests++
stats.by_route[protocol] = (stats.by_route[protocol] || 0) + 1
Expand All @@ -841,9 +845,7 @@ export function createHandleProtocol(deps) {
else stats.convert++

ctx = applyIntercept(cfg.intercept.rules, 'before_upstream', { ...ctx, body: converted.claude })
const officialTraffic =
isOfficialClaudeCodeTraffic(req.headers, inbound) ||
(detectProxiedOfficialCcFromRoutingFile(routingConfigPath) && isProxiedOfficialClaudeCode(inbound, req.headers))
const officialTraffic = officialInbound
const requestContext = classifyClaudeRequestPurpose(inbound, { officialTraffic })
if (requestContext) {
logBag.classifier = classifierRequestSummary(inbound, ctx.body, requestContext)
Expand Down Expand Up @@ -896,6 +898,7 @@ export function createHandleProtocol(deps) {
// read before outbound cleaning. API key never scopes it.
const inboundIdentity = resolveInboundIdentity({ inbound, body: ctx.body, headers: req.headers })
const stickyDeviceId = inboundIdentity.deviceId
logBag.inbound_device_id = sessionIdForLog(stickyDeviceId)
if (childDeclaredWithoutParent(inbound, req.headers)) {
stats.errors++
logBag.error_code = 'family_relation_required'
Expand Down Expand Up @@ -1201,7 +1204,7 @@ export function createHandleProtocol(deps) {
logBag.cache_prefix = prefix
logBag.cache_prefix_key = key
if (!prefix?.break) return
const where = prefix.break.section === 'messages' ? `messages[${prefix.break.index}]` : prefix.break.section
const where = formatCacheBreak(prefix)
console.warn(`[cache-prefix] request ${logCtx.request_id} turn ${prefix.turn} broke at ${where}`)
}
let result
Expand Down Expand Up @@ -1275,9 +1278,7 @@ export function createHandleProtocol(deps) {
boundAccountId: keptAccount,
boundVmId: keptVm,
vmId: selected.vmId,
epoch: extra.freshSlot
? `slot:${selected.slotIndex ?? 'x'}:${attemptStartedAt}`
: `${attemptStartedAt}:${selected.vmId || extra.attemptNo || ''}`,
epoch: extra.freshSlot ? `slot:${selected.slotIndex ?? 'x'}:${attemptStartedAt}` : 'pending',
})
if (identity && attemptSessionId) identity.callerSessionId = attemptSessionId
const credMode = credentialModeFromOauth(selected.vm?.claude || {})
Expand All @@ -1301,18 +1302,17 @@ export function createHandleProtocol(deps) {
)
let hopBody = body
if (cliHop) {
// Classifier keeps its own system. Official family keeps its inbound
// system only when the protocol tab turns passthrough on.
const panelRewrite = !requestContext && !officialPassthrough
// The front rewrite already produces the official two-block shape.
const panelRewrite = false
hopBody = prepareCliHopBody(repaired ? body : hopBody, {
stream: upstreamStream,
repaired,
cacheTtl: requestedCacheTtl,
requestContext,
preserveSystem: officialPassthrough,
preserveThinkingBlocks: preserveCliHopThinkingBlocks,
preserveCallerEnvironment,
autoCompactWindow: ctx.autoCompactWindow,
routing: routingNow,
})
hopBody = await materializeRemoteImageSources(hopBody)
if (panelRewrite) {
Expand Down
147 changes: 117 additions & 30 deletions src/lib/protocol/outbound-attempt.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -167,18 +167,107 @@ function stabilizeMessageBudgets(body) {
})
return changed ? { ...body, messages } : body
}
const OFFICIAL_CACHE = Object.freeze({ type: 'ephemeral' })

function textOf(block) {
if (typeof block === 'string') return block
return typeof block?.text === 'string' ? block.text : ''
}

/**
* Every cli-hop request leaves as the captured Claude Code shape: the identity
* sentence, then the caller's own prompt. Both markers omit ttl. A request that
* already has that identity keeps its second block; anything else is joined
* behind a fresh identity.
*/
export function officialSystemTemplate(system) {
const blocks = Array.isArray(system) ? system : system == null || system === '' ? [] : [system]
const texts = blocks.map(textOf).filter((text) => text.trim())
const identityAt = texts.findIndex((text) => text.trim() === CRS_OFFICIAL_CLI_SYSTEM)
const body = identityAt >= 0 ? texts.slice(identityAt + 1).join('\n\n') : texts.join('\n\n')
const out = [{ type: 'text', text: CRS_OFFICIAL_CLI_SYSTEM, cache_control: { ...OFFICIAL_CACHE } }]
if (body.trim()) out.push({ type: 'text', text: body, cache_control: { ...OFFICIAL_CACHE } })
return out
}

/** One marker, on the last text of the current tail, ttl omitted. */
function officialMessageMarker(messages) {
if (!Array.isArray(messages) || messages.length === 0) return messages
const next = messages.map((message) => {
if (!message || typeof message !== 'object') return message
if (typeof message.content === 'string') {
if (!message.cache_control) return message
const copy = { ...message }
delete copy.cache_control
return copy
}
if (!Array.isArray(message.content)) return message
let changed = false
const content = message.content.map((block) => {
if (!block?.cache_control) return block
changed = true
const copy = { ...block }
delete copy.cache_control
return copy
})
return changed ? { ...message, content } : message
})
let tail = -1
for (let i = next.length - 1; i >= 0; i--) {
if (next[i]?.role !== 'user') continue
const content = next[i].content
if (typeof content === 'string' ? content : Array.isArray(content) && content.length) {
tail = i
break
}
}
if (tail < 0) return next
const message = next[tail]
if (typeof message.content === 'string') {
next[tail] = {
...message,
content: [{ type: 'text', text: message.content, cache_control: { ...OFFICIAL_CACHE } }],
}
return next
}
const content = message.content.slice()
for (let i = content.length - 1; i >= 0; i--) {
if (content[i]?.type === 'thinking' || content[i]?.type === 'redacted_thinking') continue
content[i] = { ...content[i], cache_control: { ...OFFICIAL_CACHE } }
next[tail] = { ...message, content }
return next
}
return next
}
/** Official Claude Code omits eager streaming and tool cache markers. A caller TTL here makes CLI fill the omitted system and message markers. */
function stripEagerInputStreaming(tools) {
if (!Array.isArray(tools)) return tools
let changed = false
const next = tools.map((tool) => {
if (!tool || typeof tool !== 'object') return tool
const hasOwn = Object.prototype.hasOwnProperty.call(tool, 'eager_input_streaming')
const hasCache = Object.prototype.hasOwnProperty.call(tool, 'cache_control')
const fn = tool.function
const fnHas = fn && typeof fn === 'object' && Object.prototype.hasOwnProperty.call(fn, 'eager_input_streaming')
if (!hasOwn && !hasCache && !fnHas) return tool
changed = true
const copy = { ...tool }
delete copy.eager_input_streaming
delete copy.cache_control
if (fnHas) {
copy.function = { ...fn }
delete copy.function.eager_input_streaming
}
return copy
})
return changed ? next : tools
}

/**
* Caller fields plus Node-owned message breakpoints.
* CLI owns persona layout and system/tools markers; kernel only forwards.
*
* role=system turns stay where the caller put them, including a trailing one:
* Claude Code ends most turns with a reminder (SessionStart context, then
* `<total_tokens>`). Moving it into system[] puts a different text in the
* cached system block whenever the reminder changes, so the whole prefix
* misses. In place, it is history on the next turn and the prefix only grows.
* This must not depend on client classification: relays strip the billing
* block and rewrite the UA, so relayed Claude Code looks third-party.
* Every non-classifier body leaves as the official two-block template.
* Node owns that shape: identity, caller body, omitted TTL, one marker on the
* latest user turn. A trailing role=system stays a string so the cached prefix
* does not move when the reminder changes.
*/
export function prepareCliHopBody(
canonicalBody,
Expand All @@ -187,48 +276,46 @@ export function prepareCliHopBody(
repaired = false,
cacheTtl = DEFAULT_CACHE_TTL,
requestContext = null,
preserveSystem = false,
preserveThinkingBlocks = false,
preserveCallerEnvironment = false,
autoCompactWindow = null,
routing = null,
} = {},
) {
if (requestContext?.purpose === 'auto_mode_classifier') return prepareClassifierBody(canonicalBody, { stream })
if (preserveSystem) validateCacheBreakpoints(canonicalBody)
let body = officialMessagesBody(canonicalBody, { stream })
let body = officialMessagesBody(canonicalBody, { stream, preserveContentShape: true })
const windowTokens = Number(autoCompactWindow)
if (Number.isInteger(windowTokens) && windowTokens >= 100_000 && windowTokens <= 1_000_000) {
body.auto_compact_window = windowTokens
}
delete body.metadata
if (!preserveSystem) {
const leftover = stripCliOwnedSystem(body.system, { preserveCallerEnvironment })
if (leftover == null) delete body.system
else body.system = leftover
}
if (!preserveSystem) {
body = stabilizeSystemBudget(body)
body = stabilizeMessageBudgets(body)
}
const leftover = stripCliOwnedSystem(body.system, { preserveCallerEnvironment })
if (leftover == null) delete body.system
else body.system = leftover
body = stabilizeSystemBudget(body)
body = stabilizeMessageBudgets(body)
// cli-node sends mid-conversation-system. Only models that reject the role need the lift.
if (!modelSupportsMidConversationSystem(body.model)) body = liftMidConversationSystemMessages(body)
if (!preserveSystem) body = collapseSystemBudgetReminders(body)
body = collapseSystemBudgetReminders(body)

if (!repaired) {
body = ensureUnofficialAdaptiveThinking(body)
normalizeThinkingForModel(body)
body = pinHaikuCliThinking(body)
body = ensureUnofficialEffortHigh(body)
body = ensureClearThinkingContextManagement(body)
}
body = applyThinkingDisplay(body, thinkingDisplayMode(routing))
body = stripInvalidThinkingBlocks(body, { preserveThinkingBlocks })
body = applyModelRequestRules(body)
body = ensureOutputConfigSchema(body)
body = alignSamplingWithThinking(body)
if (!preserveSystem) {
body = stripIllegalCacheControlFields(body)
body = removeCacheControlFields(body)
body = applyMessageBreakpoints(body, normalizeCacheTtl(cacheTtl), 'rewrite')
if (Object.prototype.hasOwnProperty.call(body, 'temperature')) delete body.temperature
if (Object.prototype.hasOwnProperty.call(body, 'top_p')) delete body.top_p
if (Object.prototype.hasOwnProperty.call(body, 'top_k')) delete body.top_k
if (Object.prototype.hasOwnProperty.call(body, 'context_management')) delete body.context_management
if (Object.prototype.hasOwnProperty.call(body, 'cache_control')) delete body.cache_control
if (!requestContext) {
body = { ...body, system: officialSystemTemplate(body.system) }
body.messages = officialMessageMarker(body.messages)
if (Array.isArray(body.tools)) body = { ...body, tools: stripEagerInputStreaming(body.tools) }
}
return body
}
Expand Down
Loading
Loading