From fae0d51c99114dc4fed5874465cf5ca2d3dde5a3 Mon Sep 17 00:00:00 2001 From: ted Date: Sun, 30 Aug 2026 04:06:35 +0800 Subject: [PATCH 1/4] feat(kin): stateless native_messages slot host (protocol v2) CLI-side implementation of native_slot's stateless native_messages mode (.trellis/tasks/08-30-native-slot-stateless). The CLI holds no tools, agents, or cross-job state; Rust owns tool execution, continuation, and cancellation, and each job is a single queryKinMessagesWithStreaming() call that routes caller-supplied messages/system/tools/thinking/sampling straight through the real queryModel pipeline (tools: [] + extraToolSchemas). - add queryKinMessagesWithStreaming() thin wrapper in services/api/claude.ts - add nativeMessagesRunner.ts: idle -> running -> cancelling slot state machine, 7-step cancel protocol, wired into print.ts runHeadless() - stdioProtocol v2: drop kin_hello/kin_tool_result/kin_job_parked, KinStdin is now kin_job_start | kin_cancel only - main.tsx: skip stdin -p peek and MCP connect under CLAUDE_CODE_KIN_NATIVE_SLOTS (native pipe never EOFs) - remove obsolete nativeSlotRunner.ts (protocol v1 / QueryEngine design) Verified: bun run typecheck clean, bun run check (biome) clean, bun test shows only pre-existing unrelated failures. --- src/cli/print.ts | 12 ++ src/kin/nativeMessagesRunner.ts | 330 ++++++++++++++++++++++++++++++++ src/kin/stdioProtocol.ts | 88 +++++++++ src/kin/systemLayout.ts | 82 ++++++++ src/main.tsx | 157 +++++++-------- src/services/api/claude.ts | 107 +++++++++-- 6 files changed, 690 insertions(+), 86 deletions(-) create mode 100644 src/kin/nativeMessagesRunner.ts create mode 100644 src/kin/stdioProtocol.ts create mode 100644 src/kin/systemLayout.ts diff --git a/src/cli/print.ts b/src/cli/print.ts index 84b62787f5..6e943772f0 100644 --- a/src/cli/print.ts +++ b/src/cli/print.ts @@ -494,6 +494,18 @@ export async function runHeadless( setSDKStatus?: (status: SDKStatus) => void }, ): Promise { + // Native messages slot host: exclusive stdin/stdout kin_* protocol. Enter + // before Grove / structuredIO so a live kernel pipe is never treated as a + // -p prompt. CLI holds no tools/agents/canUseTool — the caller executes + // every tool_use itself (see .trellis/tasks/08-30-native-slot-stateless). + if (process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS) { + const { runNativeMessagesLoop } = await import( + '../kin/nativeMessagesRunner.js' + ) + await runNativeMessagesLoop({ options }) + return + } + if ( process.env.USER_TYPE === 'ant' && isEnvTruthy(process.env.CLAUDE_CODE_EXIT_AFTER_FIRST_RENDER) diff --git a/src/kin/nativeMessagesRunner.ts b/src/kin/nativeMessagesRunner.ts new file mode 100644 index 0000000000..aace80e7dd --- /dev/null +++ b/src/kin/nativeMessagesRunner.ts @@ -0,0 +1,330 @@ +/** + * NativeMessagesRunner — stateless native_messages slot host. + * + * The CLI holds no tools/agents/canUseTool and no cross-job state. Each slot + * is a plain { id, phase, jobId?, abort?, task? } record; a job is exactly + * one queryKinMessagesWithStreaming() call whose caller-supplied + * messages/system/tools/tool_choice/thinking/sampling flow straight to the + * real API pipeline. Tool execution, continuation, and cancellation + * bookkeeping all live in Rust (.trellis/tasks/08-30-native-slot-stateless). + */ +import { createInterface } from 'node:readline' +import type { + BetaToolChoiceAuto, + BetaToolChoiceTool, + BetaToolUnion, +} from '@anthropic-ai/sdk/resources/beta/messages/messages.mjs' +import type { ContentBlockParam } from '@anthropic-ai/sdk/resources/index.mjs' +import { queryKinMessagesWithStreaming } from '../services/api/claude.js' +import type { Message } from '../types/message.js' +import { createAssistantMessage, createUserMessage } from '../utils/messages.js' +import type { ThinkingConfig } from '../utils/thinking.js' +import { getKinTimezone, getSystemLayout } from './systemLayout.js' +import { + KIN_CAPABILITIES, + KIN_PROTOCOL_VERSION, + parseStdinLine, + slotId, + writeStdout, +} from './stdioProtocol.js' + +type Phase = 'idle' | 'running' | 'cancelling' + +type SlotState = { + id: string + phase: Phase + jobId?: string + abort?: AbortController + task?: Promise +} + +export function nativeSlotCount(): number { + const raw = process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS + const n = raw ? Number(raw) : 0 + if (!Number.isFinite(n) || n <= 0) return 0 + return Math.min(20, Math.max(1, Math.floor(n))) +} + +export async function runNativeMessagesLoop(_ctx: { + options: { + userSpecifiedModel?: string + fallbackModel?: string + thinkingConfig?: ThinkingConfig + } +}): Promise { + const n = nativeSlotCount() + if (n <= 0) return + process.env.CLAUDE_CODE_DISABLE_NONSTREAMING_FALLBACK = '1' + + process.stderr.write( + `[kin] native_messages loop n=${n} protocol=${KIN_PROTOCOL_VERSION}\n`, + ) + const slots = new Map() + for (let i = 0; i < n; i++) { + const id = slotId(i) + slots.set(id, { id, phase: 'idle' }) + } + await writeStdout({ + type: 'kin_host_ready', + protocol_version: KIN_PROTOCOL_VERSION, + slots: n, + system_layout: getSystemLayout(), + timezone: getKinTimezone(), + capabilities: [...KIN_CAPABILITIES], + }) + for (const id of slots.keys()) { + await writeStdout({ type: 'kin_slot_ready', slot_id: id }) + } + + const rl = createInterface({ input: process.stdin, crlfDelay: Infinity }) + for await (const line of rl) { + const msg = parseStdinLine(line) + if (!msg) continue + switch (msg.type) { + case 'kin_job_start': + startJob(slots, msg.slot_id, msg.job_id, msg.request, _ctx.options) + break + case 'kin_cancel': + await cancelJob(slots, msg.slot_id, msg.job_id) + break + } + } +} + +function startJob( + slots: Map, + slotIdArg: string, + jobId: string, + request: Record, + hostOptions: { + userSpecifiedModel?: string + fallbackModel?: string + thinkingConfig?: ThinkingConfig + }, +): void { + const slot = slots.get(slotIdArg) + if (!slot) { + void writeStdout({ + type: 'kin_job_error', + job_id: jobId, + slot_id: slotIdArg, + error: `unknown slot ${slotIdArg}`, + }) + return + } + if (slot.phase !== 'idle') { + void writeStdout({ + type: 'kin_job_error', + job_id: jobId, + slot_id: slot.id, + error: `slot ${slot.id} busy phase=${slot.phase}`, + }) + return + } + + const abort = new AbortController() + slot.phase = 'running' + slot.jobId = jobId + slot.abort = abort + slot.task = runJob(slot, jobId, request, hostOptions, abort).finally(() => { + if (slot.jobId === jobId) { + slot.phase = 'idle' + slot.jobId = undefined + slot.abort = undefined + slot.task = undefined + } + }) +} + +async function cancelJob( + slots: Map, + slotIdArg: string | undefined, + jobId: string, +): Promise { + const slot = slotIdArg ? slots.get(slotIdArg) : undefined + if (!slot || slot.jobId !== jobId) return + if (slot.phase !== 'running') return + slot.phase = 'cancelling' + slot.abort?.abort() + await slot.task + slot.phase = 'idle' + slot.jobId = undefined + slot.abort = undefined + slot.task = undefined + await writeStdout({ type: 'kin_cancel_ack', job_id: jobId, slot_id: slot.id }) +} + +async function runJob( + slot: SlotState, + jobId: string, + request: Record, + hostOptions: { + userSpecifiedModel?: string + fallbackModel?: string + thinkingConfig?: ThinkingConfig + }, + abort: AbortController, +): Promise { + const model = + typeof request.model === 'string' && request.model + ? request.model + : hostOptions.userSpecifiedModel || '' + const messages = messagesFromRequest(request) + const system = systemFromRequest(request) + const toolSchemas = Array.isArray(request.tools) + ? (request.tools as BetaToolUnion[]) + : [] + const toolChoice = request.tool_choice as + | BetaToolChoiceTool + | BetaToolChoiceAuto + | undefined + const thinking = thinkingFromRequest(request, hostOptions.thinkingConfig) + const maxTokens = + typeof request.max_tokens === 'number' ? request.max_tokens : undefined + const temperature = + typeof request.temperature === 'number' ? request.temperature : undefined + const topP = typeof request.top_p === 'number' ? request.top_p : undefined + const topK = typeof request.top_k === 'number' ? request.top_k : undefined + const stopSequences = Array.isArray(request.stop_sequences) + ? (request.stop_sequences as string[]) + : undefined + + let stopReason = 'end_turn' + let usage: unknown = {} + try { + const stream = queryKinMessagesWithStreaming({ + messages, + system, + toolSchemas, + toolChoice, + thinking, + maxTokens, + temperature, + topP, + topK, + stopSequences, + model, + signal: abort.signal, + }) + for await (const ev of stream) { + if (ev.type === 'stream_event') { + const rec = ev as { event?: unknown } + await writeStdout({ + type: 'kin_stream_event', + job_id: jobId, + slot_id: slot.id, + event: rec.event, + }) + continue + } + if (ev.type === 'assistant') { + const rec = ev as { + message?: { stop_reason?: string; usage?: unknown } + } + stopReason = rec.message?.stop_reason || stopReason + usage = rec.message?.usage || usage + continue + } + if (ev.type === 'system') { + const text = extractErrorText(ev as Record) + await writeStdout({ + type: 'kin_job_error', + job_id: jobId, + slot_id: slot.id, + error: text, + }) + return + } + } + } catch (err) { + if (!abort.signal.aborted) { + await writeStdout({ + type: 'kin_job_error', + job_id: jobId, + slot_id: slot.id, + error: err instanceof Error ? err.message : String(err), + }) + } + return + } + + if (abort.signal.aborted) return + + await writeStdout({ + type: 'kin_job_done', + job_id: jobId, + slot_id: slot.id, + stop_reason: stopReason, + usage, + }) +} + +function extractErrorText(ev: Record): string { + const message = ev.message as { content?: unknown } | undefined + const content = message?.content + if (typeof content === 'string') return content + if (Array.isArray(content)) { + const first = content[0] as { text?: unknown } | undefined + if (first && typeof first.text === 'string') return first.text + } + return 'system error' +} + +function systemFromRequest(request: Record): string[] { + const system = request.system + if (typeof system === 'string') return [system] + if (!Array.isArray(system)) return [] + return system.map(block => { + if (typeof block === 'string') return block + if (block && typeof block === 'object' && 'text' in block) { + return String((block as { text?: unknown }).text || '') + } + return '' + }) +} + +function messagesFromRequest(request: Record): Message[] { + const messages = Array.isArray(request.messages) ? request.messages : [] + return messages.map(toEngineMessage).filter((m): m is Message => m !== null) +} + +function toEngineMessage(raw: unknown): Message | null { + if (!raw || typeof raw !== 'object') return null + const msg = raw as { role?: string; content?: unknown } + const content = msg.content + const asContent: string | ContentBlockParam[] = + typeof content === 'string' || Array.isArray(content) + ? (content as string | ContentBlockParam[]) + : '' + if (msg.role === 'assistant') { + return createAssistantMessage({ content: asContent as never }) + } + if (msg.role === 'user' || msg.role === 'tool') { + return createUserMessage({ content: asContent }) + } + return null +} + +function thinkingFromRequest( + request: Record, + fallback: ThinkingConfig | undefined, +): ThinkingConfig { + const raw = request.thinking + if (raw && typeof raw === 'object') { + const rec = raw as { + type?: string + budget_tokens?: number + budgetTokens?: number + } + if (rec.type === 'enabled') { + return { + type: 'enabled', + budgetTokens: Number(rec.budget_tokens ?? rec.budgetTokens ?? 0), + } + } + if (rec.type === 'adaptive') return { type: 'adaptive' } + if (rec.type === 'disabled' || rec.type === 'none') + return { type: 'disabled' } + } + return fallback ?? { type: 'disabled' } +} diff --git a/src/kin/stdioProtocol.ts b/src/kin/stdioProtocol.ts new file mode 100644 index 0000000000..feaf2d3433 --- /dev/null +++ b/src/kin/stdioProtocol.ts @@ -0,0 +1,88 @@ +export const KIN_PROTOCOL_VERSION = 2 +export const KIN_CAPABILITIES = [ + 'multi_slot', + 'native_sse', + 'stateless', +] as const + +export type KinStdin = + | { + type: 'kin_job_start' + job_id: string + slot_id: string + request: Record + } + | { type: 'kin_cancel'; job_id: string; slot_id?: string } + +export type KinStdout = + | { + type: 'kin_host_ready' + protocol_version: number + slots: number + system_layout: string + timezone: string + capabilities: string[] + config_hash?: string + } + | { type: 'kin_slot_ready'; slot_id: string } + | { + type: 'kin_stream_event' + job_id: string + slot_id: string + event: unknown + } + | { + type: 'kin_job_done' + job_id: string + slot_id: string + stop_reason: string + usage?: unknown + } + | { type: 'kin_job_error'; job_id: string; slot_id?: string; error: string } + | { type: 'kin_cancel_ack'; job_id: string; slot_id: string } + +export function parseStdinLine(line: string): KinStdin | null { + const trimmed = line.trim() + if (!trimmed) return null + try { + const parsed = JSON.parse(trimmed) as KinStdin + if (!parsed || typeof parsed.type !== 'string') return null + if (!parsed.type.startsWith('kin_')) return null + return parsed + } catch { + return null + } +} + +let writeChain: Promise = Promise.resolve() + +export function writeStdout(frame: KinStdout): Promise { + const line = JSON.stringify(frame) + '\n' + const next = writeChain.then( + () => writeLine(line), + () => writeLine(line), + ) + writeChain = next.then( + () => undefined, + () => undefined, + ) + return next +} + +function writeLine(line: string): Promise { + return new Promise((resolve, reject) => { + let settled = false + const done = (err?: Error | null) => { + if (settled) return + settled = true + if (err) reject(err) + else resolve() + } + const ok = process.stdout.write(line, err => done(err)) + if (!ok) process.stdout.once('drain', () => done()) + }) +} + +export function slotId(index: number): string { + return `s${String(index).padStart(2, '0')}` +} diff --git a/src/kin/systemLayout.ts b/src/kin/systemLayout.ts new file mode 100644 index 0000000000..b24ac659af --- /dev/null +++ b/src/kin/systemLayout.ts @@ -0,0 +1,82 @@ +/** + * P1: CLI-owned system layout. Rust must not construct billing. + * + * zero: attribution + ` prompt_version=official sentence;` in one block + * identity: attribution block, then official sentence block + * Always append `# Environment\nTime zone: ` and optional leftover. + * + * cch is whatever getAttributionHeader() already emitted. Node builds omit it. + */ +export const IDENTITY = + "You are a Claude agent, built on Anthropic's Claude Agent SDK." + +export type SystemLayout = 'stock' | 'zero' | 'identity' + +export function getSystemLayout(): SystemLayout { + const raw = ( + process.env.CLAUDE_CODE_SYSTEM_LAYOUT || + process.env.KIN_SYSTEM_MODE || + '' + ) + .trim() + .toLowerCase() + if (raw === 'zero') return 'zero' + if (raw === 'identity' || raw === 'id' || raw === 'block') return 'identity' + if (process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS) return 'zero' + return 'stock' +} + +export function getKinTimezone(): string { + const tz = ( + process.env.CLAUDE_CODE_TIMEZONE || + process.env.KIN_SLOT_TZ || + process.env.TZ || + 'America/New_York' + ).trim() + return tz || 'America/New_York' +} + +export function environmentBlock(timezone = getKinTimezone()): string { + return `# Environment\nTime zone: ${timezone}` +} + +export function layoutSystemBlocks(opts: { + attribution: string + leftover?: string +}): string[] { + const layout = getSystemLayout() + const attribution = (opts.attribution || '').trim() + const leftover = opts.leftover?.trim() + const env = environmentBlock() + const blocks: string[] = [] + + if (layout === 'zero') { + const billing = attribution + ? attribution.replace(/;?\s*$/, '') + `; prompt_version=${IDENTITY};` + : `x-anthropic-billing-header: prompt_version=${IDENTITY};` + blocks.push(billing) + } else { + if (attribution) blocks.push(attribution) + blocks.push(IDENTITY) + } + blocks.push(env) + if (leftover) blocks.push(leftover) + return blocks +} + +export function leftoverFromSystemPrompt( + systemPrompt: readonly string[], +): string | undefined { + const parts: string[] = [] + for (const block of systemPrompt) { + const text = (block || '').trim() + if (!text) continue + if (text.startsWith('x-anthropic-billing-header')) continue + if (text === IDENTITY) continue + if (text.startsWith('# Environment')) continue + if (text.includes('mcp__kin_runtime__') || text.includes('persistent Kin')) + continue + parts.push(text) + } + return parts.length ? parts.join('\n\n') : undefined +} diff --git a/src/main.tsx b/src/main.tsx index 3317f99c9c..2b211b9be6 100644 --- a/src/main.tsx +++ b/src/main.tsx @@ -1031,6 +1031,11 @@ async function getInputPrompt( prompt: string, inputFormat: 'text' | 'stream-json', ): Promise> { + // Native slot host owns stdin for kin_* frames. `-p` peek waits for EOF + // after the first byte; a live kernel pipe never ends, so we must not read. + if (process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS) { + return prompt; + } if ( !process.stdin.isTTY && // Input hijacking breaks MCP. @@ -3271,85 +3276,87 @@ async function run(): Promise { // fetch was kicked off early (line ~2558) so only residual time blocks // here. --bare skips claude.ai entirely for perf-sensitive scripts. profileCheckpoint('before_connectMcp'); - await connectMcpBatch(regularMcpConfigs, 'regular'); - profileCheckpoint('after_connectMcp'); - // Dedup: suppress plugin MCP servers that duplicate a claude.ai - // connector (connector wins), then connect claude.ai servers. - // Bounded wait — #23725 made this blocking so single-turn -p sees - // connectors, but with 40+ slow connectors tengu_startup_perf p99 - // climbed to 76s. If fetch+connect doesn't finish in time, proceed; - // the promise keeps running and updates headlessStore in the - // background so turn 2+ still sees connectors. - const CLAUDE_AI_MCP_TIMEOUT_MS = 5_000; - const claudeaiConnect = claudeaiConfigPromise.then(claudeaiConfigs => { - if (Object.keys(claudeaiConfigs).length > 0) { - const claudeaiSigs = new Set(); - for (const config of Object.values(claudeaiConfigs)) { - const sig = getMcpServerSignature(config); - if (sig) claudeaiSigs.add(sig); - } - const suppressed = new Set(); - for (const [name, config] of Object.entries(regularMcpConfigs)) { - if (!name.startsWith('plugin:')) continue; - const sig = getMcpServerSignature(config); - if (sig && claudeaiSigs.has(sig)) suppressed.add(name); - } - if (suppressed.size > 0) { - logForDebugging( - `[MCP] Lazy dedup: suppressing ${suppressed.size} plugin server(s) that duplicate claude.ai connectors: ${[...suppressed].join(', ')}`, - ); - // Disconnect before filtering from state. Only connected - // servers need cleanup — clearServerCache on a never-connected - // server triggers a real connect just to kill it (memoize - // cache-miss path, see useManageMCPConnections.ts:870). - for (const c of headlessStore.getState().mcp.clients) { - if (!suppressed.has(c.name) || c.type !== 'connected') continue; - c.client.onclose = undefined; - void clearServerCache(c.name, c.config).catch(() => {}); + if (!process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS) { + await connectMcpBatch(regularMcpConfigs, 'regular'); + profileCheckpoint('after_connectMcp'); + // Dedup: suppress plugin MCP servers that duplicate a claude.ai + // connector (connector wins), then connect claude.ai servers. + // Bounded wait — #23725 made this blocking so single-turn -p sees + // connectors, but with 40+ slow connectors tengu_startup_perf p99 + // climbed to 76s. If fetch+connect doesn't finish in time, proceed; + // the promise keeps running and updates headlessStore in the + // background so turn 2+ still sees connectors. + const CLAUDE_AI_MCP_TIMEOUT_MS = 5_000; + const claudeaiConnect = claudeaiConfigPromise.then(claudeaiConfigs => { + if (Object.keys(claudeaiConfigs).length > 0) { + const claudeaiSigs = new Set(); + for (const config of Object.values(claudeaiConfigs)) { + const sig = getMcpServerSignature(config); + if (sig) claudeaiSigs.add(sig); } - headlessStore.setState(prev => { - let { clients, tools, commands, resources } = prev.mcp; - clients = clients.filter(c => !suppressed.has(c.name)); - tools = tools.filter(t => !t.mcpInfo || !suppressed.has(t.mcpInfo.serverName)); - for (const name of suppressed) { - commands = excludeCommandsByServer(commands, name); - resources = excludeResourcesByServer(resources, name); + const suppressed = new Set(); + for (const [name, config] of Object.entries(regularMcpConfigs)) { + if (!name.startsWith('plugin:')) continue; + const sig = getMcpServerSignature(config); + if (sig && claudeaiSigs.has(sig)) suppressed.add(name); + } + if (suppressed.size > 0) { + logForDebugging( + `[MCP] Lazy dedup: suppressing ${suppressed.size} plugin server(s) that duplicate claude.ai connectors: ${[...suppressed].join(', ')}`, + ); + // Disconnect before filtering from state. Only connected + // servers need cleanup — clearServerCache on a never-connected + // server triggers a real connect just to kill it (memoize + // cache-miss path, see useManageMCPConnections.ts:870). + for (const c of headlessStore.getState().mcp.clients) { + if (!suppressed.has(c.name) || c.type !== 'connected') continue; + c.client.onclose = undefined; + void clearServerCache(c.name, c.config).catch(() => {}); } - return { - ...prev, - mcp: { - ...prev.mcp, - clients, - tools, - commands, - resources, - }, - }; - }); + headlessStore.setState(prev => { + let { clients, tools, commands, resources } = prev.mcp; + clients = clients.filter(c => !suppressed.has(c.name)); + tools = tools.filter(t => !t.mcpInfo || !suppressed.has(t.mcpInfo.serverName)); + for (const name of suppressed) { + commands = excludeCommandsByServer(commands, name); + resources = excludeResourcesByServer(resources, name); + } + return { + ...prev, + mcp: { + ...prev.mcp, + clients, + tools, + commands, + resources, + }, + }; + }); + } } + // Suppress claude.ai connectors that duplicate an enabled + // manual server (URL-signature match). Plugin dedup above only + // handles `plugin:*` keys; this catches manual `.mcp.json` entries. + // plugin:* must be excluded here — step 1 already suppressed + // those (claude.ai wins); leaving them in suppresses the + // connector too, and neither survives (gh-39974). + const nonPluginConfigs = pickBy(regularMcpConfigs, (_, n) => !n.startsWith('plugin:')); + const { servers: dedupedClaudeAi } = dedupClaudeAiMcpServers(claudeaiConfigs, nonPluginConfigs); + return connectMcpBatch(dedupedClaudeAi, 'claudeai'); + }); + let claudeaiTimer: ReturnType | undefined; + const claudeaiTimedOut = await Promise.race([ + claudeaiConnect.then(() => false), + new Promise(resolve => { + claudeaiTimer = setTimeout(r => r(true), CLAUDE_AI_MCP_TIMEOUT_MS, resolve); + }), + ]); + if (claudeaiTimer) clearTimeout(claudeaiTimer); + if (claudeaiTimedOut) { + logForDebugging( + `[MCP] claude.ai connectors not ready after ${CLAUDE_AI_MCP_TIMEOUT_MS}ms — proceeding; background connection continues`, + ); } - // Suppress claude.ai connectors that duplicate an enabled - // manual server (URL-signature match). Plugin dedup above only - // handles `plugin:*` keys; this catches manual `.mcp.json` entries. - // plugin:* must be excluded here — step 1 already suppressed - // those (claude.ai wins); leaving them in suppresses the - // connector too, and neither survives (gh-39974). - const nonPluginConfigs = pickBy(regularMcpConfigs, (_, n) => !n.startsWith('plugin:')); - const { servers: dedupedClaudeAi } = dedupClaudeAiMcpServers(claudeaiConfigs, nonPluginConfigs); - return connectMcpBatch(dedupedClaudeAi, 'claudeai'); - }); - let claudeaiTimer: ReturnType | undefined; - const claudeaiTimedOut = await Promise.race([ - claudeaiConnect.then(() => false), - new Promise(resolve => { - claudeaiTimer = setTimeout(r => r(true), CLAUDE_AI_MCP_TIMEOUT_MS, resolve); - }), - ]); - if (claudeaiTimer) clearTimeout(claudeaiTimer); - if (claudeaiTimedOut) { - logForDebugging( - `[MCP] claude.ai connectors not ready after ${CLAUDE_AI_MCP_TIMEOUT_MS}ms — proceeding; background connection continues`, - ); } profileCheckpoint('after_connectMcp_claudeai'); diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index 5c75e22221..023419a8aa 100644 --- a/src/services/api/claude.ts +++ b/src/services/api/claude.ts @@ -29,6 +29,11 @@ import { getAttributionHeader, getCLISyspromptPrefix, } from '../../constants/system.js' +import { + getSystemLayout, + layoutSystemBlocks, + leftoverFromSystemPrompt, +} from '../../kin/systemLayout.js' import { getEmptyToolPermissionContext, type QueryChainTracking, @@ -710,6 +715,9 @@ export type Options = { enablePromptCaching?: boolean skipCacheWrite?: boolean temperatureOverride?: number + topP?: number + topK?: number + stopSequences?: string[] effortValue?: EffortValue mcpTools: Tools hasPendingMcpServers?: boolean @@ -801,6 +809,68 @@ export async function* queryModelWithStreaming({ }) } +/** + * Kin native_messages thin wrapper (.trellis/tasks/08-30-native-slot-stateless). + * Caller (Rust) owns messages/system/tools/tool_choice/thinking/sampling — + * this only routes them through the real queryModel pipeline with tools: [] + * so the CLI never holds a local tool executor; extraToolSchemas carries the + * caller's tool declarations through untouched. + */ +export async function* queryKinMessagesWithStreaming({ + messages, + system, + toolSchemas, + toolChoice, + thinking, + maxTokens, + temperature, + topP, + topK, + stopSequences, + model, + signal, +}: { + messages: Message[] + system: string[] + toolSchemas: BetaToolUnion[] + toolChoice?: BetaToolChoiceTool | BetaToolChoiceAuto + thinking: ThinkingConfig + maxTokens?: number + temperature?: number + topP?: number + topK?: number + stopSequences?: string[] + model: string + signal: AbortSignal +}): AsyncGenerator< + StreamEvent | AssistantMessage | SystemAPIErrorMessage, + void +> { + yield* queryModelWithStreaming({ + messages, + systemPrompt: asSystemPrompt(system), + thinkingConfig: thinking, + tools: [], + signal, + options: { + model, + toolChoice, + extraToolSchemas: toolSchemas, + maxOutputTokensOverride: maxTokens, + temperatureOverride: temperature, + topP, + topK, + stopSequences, + getToolPermissionContext: async () => getEmptyToolPermissionContext(), + isNonInteractiveSession: true, + hasAppendSystemPrompt: false, + querySource: 'kin_native_messages', + agents: [], + mcpTools: [], + }, + }) +} + /** * Determines if an LSP tool should be deferred (tool appears with defer_loading: true) * because LSP initialization is not yet complete. @@ -1430,18 +1500,28 @@ async function* queryModel( useSearchExtraTools && hasChromeTools && !isMcpInstructionsDeltaEnabled() // filter(Boolean) works by converting each element to a boolean - empty strings become false and are filtered out. - systemPrompt = asSystemPrompt( - [ - getAttributionHeader(), - getCLISyspromptPrefix({ - isNonInteractive: options.isNonInteractiveSession, - hasAppendSystemPrompt: options.hasAppendSystemPrompt, + const kinSystemLayout = getSystemLayout() + if (kinSystemLayout !== 'stock') { + systemPrompt = asSystemPrompt( + layoutSystemBlocks({ + attribution: getAttributionHeader(), + leftover: leftoverFromSystemPrompt(systemPrompt), }), - ...systemPrompt, - ...(advisorModel ? [ADVISOR_TOOL_INSTRUCTIONS] : []), - ...(injectChromeHere ? [CHROME_SEARCH_EXTRA_TOOLS_INSTRUCTIONS] : []), - ].filter(Boolean), - ) + ) + } else { + systemPrompt = asSystemPrompt( + [ + getAttributionHeader(), + getCLISyspromptPrefix({ + isNonInteractive: options.isNonInteractiveSession, + hasAppendSystemPrompt: options.hasAppendSystemPrompt, + }), + ...systemPrompt, + ...(advisorModel ? [ADVISOR_TOOL_INSTRUCTIONS] : []), + ...(injectChromeHere ? [CHROME_SEARCH_EXTRA_TOOLS_INSTRUCTIONS] : []), + ].filter(Boolean), + ) + } // ── Break-cache integration ── // If a one-time break-cache marker exists, or always-mode is on, append a @@ -1804,6 +1884,11 @@ async function* queryModel( max_tokens: maxOutputTokens, thinking, ...(temperature !== undefined && { temperature }), + ...(options.topP !== undefined && { top_p: options.topP }), + ...(options.topK !== undefined && { top_k: options.topK }), + ...(options.stopSequences?.length && { + stop_sequences: options.stopSequences, + }), ...(contextManagement && useBetas && betasParams.includes(CONTEXT_MANAGEMENT_BETA_HEADER) && { From 09cca33edfafbb4bf1a3dc428461e94a2112010d Mon Sep 17 00:00:00 2001 From: ted Date: Wed, 23 Sep 2026 09:03:25 +0800 Subject: [PATCH 2/4] fix(kin): stabilize native prompt cache prefix --- src/kin/__tests__/runtimeConfig.test.ts | 153 +++++++++++++++++ src/kin/nativeMessagesRunner.ts | 51 +++++- src/kin/runtimeConfig.ts | 64 +++++++ src/kin/systemLayout.ts | 62 +++++-- .../claude-cache-breakpoints.test.ts | 54 ++++++ src/services/api/claude.ts | 159 ++++++++++-------- 6 files changed, 446 insertions(+), 97 deletions(-) create mode 100644 src/kin/__tests__/runtimeConfig.test.ts create mode 100644 src/kin/runtimeConfig.ts create mode 100644 src/services/api/__tests__/claude-cache-breakpoints.test.ts diff --git a/src/kin/__tests__/runtimeConfig.test.ts b/src/kin/__tests__/runtimeConfig.test.ts new file mode 100644 index 0000000000..8b770d0a41 --- /dev/null +++ b/src/kin/__tests__/runtimeConfig.test.ts @@ -0,0 +1,153 @@ +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, test } from 'bun:test' +import { readKinRuntimeConfig } from '../runtimeConfig' +import { + COMPACT_IDENTITY, + IDENTITY, + getSystemLayout, + layoutSystemBlocks, +} from '../systemLayout' + +const originalEnv = { + path: process.env.KIN_KERNEL_CONFIG_PATH, + native: process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS, + layout: process.env.CLAUDE_CODE_SYSTEM_LAYOUT, + mode: process.env.KIN_SYSTEM_MODE, +} + +function restoreEnv(): void { + for (const [key, value] of [ + ['KIN_KERNEL_CONFIG_PATH', originalEnv.path], + ['CLAUDE_CODE_KIN_NATIVE_SLOTS', originalEnv.native], + ['CLAUDE_CODE_SYSTEM_LAYOUT', originalEnv.layout], + ['KIN_SYSTEM_MODE', originalEnv.mode], + ] as const) { + if (value === undefined) delete process.env[key] + else process.env[key] = value + } +} + +afterEach(restoreEnv) + +describe('Kin runtime configuration', () => { + test('reads updated kernel config on every native job', () => { + const root = mkdtempSync(join(tmpdir(), 'kin-runtime-config-')) + const path = join(root, 'kernel.json') + process.env.KIN_KERNEL_CONFIG_PATH = path + process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS = '1' + try { + writeFileSync( + path, + JSON.stringify({ + default_cache_ttl: '5m', + persona_preset: 'zero', + system_layout: 'zero', + timezone: 'UTC', + }), + ) + expect(readKinRuntimeConfig()).toMatchObject({ + default_cache_ttl: '5m', + persona_preset: 'zero', + system_layout: 'zero', + timezone: 'UTC', + }) + writeFileSync( + path, + JSON.stringify({ + default_cache_ttl: '1h', + persona_preset: 'official_full', + system_layout: 'identity', + timezone: 'America/Los_Angeles', + }), + ) + expect(readKinRuntimeConfig()).toMatchObject({ + default_cache_ttl: '1h', + persona_preset: 'official_full', + system_layout: 'identity', + timezone: 'America/Los_Angeles', + }) + } finally { + rmSync(root, { recursive: true, force: true }) + } + }) + + test('native file config overrides fixed standalone layout environment', () => { + const root = mkdtempSync(join(tmpdir(), 'kin-runtime-layout-')) + const path = join(root, 'kernel.json') + process.env.KIN_KERNEL_CONFIG_PATH = path + process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS = '1' + process.env.CLAUDE_CODE_SYSTEM_LAYOUT = 'zero' + process.env.KIN_SYSTEM_MODE = 'zero' + try { + writeFileSync( + path, + JSON.stringify({ + default_cache_ttl: '1h', + persona_preset: 'official', + system_layout: 'identity', + }), + ) + expect(getSystemLayout(readKinRuntimeConfig())).toBe('identity') + } finally { + rmSync(root, { recursive: true, force: true }) + } + }) + + test('layoutSystemBlocks keeps the three persona contracts distinct', () => { + const common = { + attribution: 'x-anthropic-billing-header: cc_version=test;', + } + const zero = layoutSystemBlocks({ + ...common, + runtime: { + default_cache_ttl: '5m', + persona_preset: 'zero', + system_layout: 'zero', + }, + }) + expect(zero[0]).toContain(`prompt_version=${COMPACT_IDENTITY}`) + expect(zero).not.toContain(IDENTITY) + + const official = layoutSystemBlocks({ + ...common, + runtime: { + default_cache_ttl: '1h', + persona_preset: 'official', + system_layout: 'identity', + }, + }) + expect(official).toContain(IDENTITY) + expect(official).not.toContain('# Environment\nTime zone:') + + const full = layoutSystemBlocks({ + ...common, + officialFullPrompt: [ + 'official static prompt', + '# Environment\nTime zone: UTC', + ], + runtime: { + default_cache_ttl: '1h', + persona_preset: 'official_full', + system_layout: 'identity', + }, + }) + expect(full).toContain( + 'official static prompt\n\n# Environment\nTime zone: UTC', + ) + }) + + test('removes turn-scoped attribution fields from the native cache prefix', () => { + const blocks = layoutSystemBlocks({ + attribution: + 'x-anthropic-billing-header: cc_version=test; cch=abcdef; cc_workload=cron;', + runtime: { + default_cache_ttl: '1h', + persona_preset: 'official', + system_layout: 'identity', + }, + }) + expect(blocks[0]).toBe('x-anthropic-billing-header: cc_version=test;') + }) +}) diff --git a/src/kin/nativeMessagesRunner.ts b/src/kin/nativeMessagesRunner.ts index aace80e7dd..56e798a1f4 100644 --- a/src/kin/nativeMessagesRunner.ts +++ b/src/kin/nativeMessagesRunner.ts @@ -19,6 +19,7 @@ import { queryKinMessagesWithStreaming } from '../services/api/claude.js' import type { Message } from '../types/message.js' import { createAssistantMessage, createUserMessage } from '../utils/messages.js' import type { ThinkingConfig } from '../utils/thinking.js' +import { readKinRuntimeConfig } from './runtimeConfig.js' import { getKinTimezone, getSystemLayout } from './systemLayout.js' import { KIN_CAPABILITIES, @@ -172,13 +173,18 @@ async function runJob( const messages = messagesFromRequest(request) const system = systemFromRequest(request) const toolSchemas = Array.isArray(request.tools) - ? (request.tools as BetaToolUnion[]) + ? request.tools + .filter((tool): tool is BetaToolUnion => + Boolean(tool && typeof tool === 'object'), + ) + .map(stripToolCacheControl) : [] const toolChoice = request.tool_choice as | BetaToolChoiceTool | BetaToolChoiceAuto | undefined const thinking = thinkingFromRequest(request, hostOptions.thinkingConfig) + const runtime = readKinRuntimeConfig() const maxTokens = typeof request.max_tokens === 'number' ? request.max_tokens : undefined const temperature = @@ -198,6 +204,9 @@ async function runJob( toolSchemas, toolChoice, thinking, + cacheTtl: runtime.default_cache_ttl, + personaPreset: runtime.persona_preset, + runtimeConfig: runtime, maxTokens, temperature, topP, @@ -270,22 +279,46 @@ function extractErrorText(ev: Record): string { return 'system error' } +function stripToolCacheControl(tool: BetaToolUnion): BetaToolUnion { + const next = { ...tool } + // BetaToolUnion omits the runtime-only marker field from its public type. + const runtimeTool = next as BetaToolUnion & { cache_control?: unknown } + delete runtimeTool.cache_control + return next +} + function systemFromRequest(request: Record): string[] { const system = request.system if (typeof system === 'string') return [system] if (!Array.isArray(system)) return [] - return system.map(block => { - if (typeof block === 'string') return block - if (block && typeof block === 'object' && 'text' in block) { - return String((block as { text?: unknown }).text || '') - } - return '' + return system.flatMap(block => { + if (typeof block === 'string') return [block] + if (!block || typeof block !== 'object' || !('text' in block)) return [] + const text = block.text + return typeof text === 'string' ? [text] : [] }) } - function messagesFromRequest(request: Record): Message[] { const messages = Array.isArray(request.messages) ? request.messages : [] - return messages.map(toEngineMessage).filter((m): m is Message => m !== null) + return messages + .map(stripMessageCacheControl) + .map(toEngineMessage) + .filter((m): m is Message => m !== null) +} + +function stripMessageCacheControl(message: unknown): unknown { + if (!message || typeof message !== 'object') return message + const next = { ...(message as Record) } + delete next.cache_control + if (Array.isArray(next.content)) { + next.content = next.content.map(block => { + if (!block || typeof block !== 'object') return block + const clean = { ...(block as Record) } + delete clean.cache_control + return clean + }) + } + return next } function toEngineMessage(raw: unknown): Message | null { diff --git a/src/kin/runtimeConfig.ts b/src/kin/runtimeConfig.ts new file mode 100644 index 0000000000..6eb126c3de --- /dev/null +++ b/src/kin/runtimeConfig.ts @@ -0,0 +1,64 @@ +import { existsSync, readFileSync } from 'node:fs' + +export type KinCacheTtl = '5m' | '1h' +export type KinPersonaPreset = 'zero' | 'official' | 'official_full' + +export type KinRuntimeConfig = { + default_cache_ttl: KinCacheTtl + persona_preset: KinPersonaPreset + system_layout: 'zero' | 'identity' + timezone?: string +} + +const DEFAULT_CONFIG_PATH = '/run/kin/kernel.json' + +function normalizeCacheTtl(value: unknown): KinCacheTtl { + return String(value ?? '') + .trim() + .toLowerCase() === '5m' + ? '5m' + : '1h' +} + +function normalizePersonaPreset(value: unknown): KinPersonaPreset { + const raw = String(value ?? '') + .trim() + .toLowerCase() + if (raw === 'zero' || raw === 'zero_inject' || raw === '0inject') + return 'zero' + if (raw === 'official_full' || raw === 'full') return 'official_full' + return 'official' +} + +function configPaths(): string[] { + const configured = String( + process.env.KIN_KERNEL_CONFIG_PATH || process.env.KIN_KERNEL_CONFIG || '', + ).trim() + return configured ? [configured, DEFAULT_CONFIG_PATH] : [DEFAULT_CONFIG_PATH] +} + +/** Read kernel.json for every native job; panel writes are therefore hot-reloaded. */ +export function readKinRuntimeConfig(): KinRuntimeConfig { + for (const filePath of configPaths()) { + if (!existsSync(filePath)) continue + try { + const raw = JSON.parse(readFileSync(filePath, 'utf8')) as Record< + string, + unknown + > + return { + default_cache_ttl: normalizeCacheTtl(raw.default_cache_ttl), + persona_preset: normalizePersonaPreset(raw.persona_preset), + system_layout: raw.system_layout === 'zero' ? 'zero' : 'identity', + ...(typeof raw.timezone === 'string' && raw.timezone.trim() + ? { timezone: raw.timezone.trim() } + : {}), + } + } catch {} + } + return { + default_cache_ttl: '1h', + persona_preset: 'official', + system_layout: 'identity', + } +} diff --git a/src/kin/systemLayout.ts b/src/kin/systemLayout.ts index b24ac659af..f32293ea2f 100644 --- a/src/kin/systemLayout.ts +++ b/src/kin/systemLayout.ts @@ -1,18 +1,22 @@ /** * P1: CLI-owned system layout. Rust must not construct billing. * - * zero: attribution + ` prompt_version=official sentence;` in one block - * identity: attribution block, then official sentence block - * Always append `# Environment\nTime zone: ` and optional leftover. - * - * cch is whatever getAttributionHeader() already emitted. Node builds omit it. + * The kernel JSON is the live authority for native slots. Environment variables + * remain fallback values for standalone CLI runs without a Kin kernel. */ +import { readKinRuntimeConfig, type KinRuntimeConfig } from './runtimeConfig.js' + export const IDENTITY = "You are a Claude agent, built on Anthropic's Claude Agent SDK." +export const COMPACT_IDENTITY = 'You are Anthropic Claude Agent SDK.' export type SystemLayout = 'stock' | 'zero' | 'identity' -export function getSystemLayout(): SystemLayout { +export function getSystemLayout(runtime?: KinRuntimeConfig): SystemLayout { + if (runtime) return runtime.system_layout + if (process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS) { + return readKinRuntimeConfig().system_layout + } const raw = ( process.env.CLAUDE_CODE_SYSTEM_LAYOUT || process.env.KIN_SYSTEM_MODE || @@ -22,11 +26,14 @@ export function getSystemLayout(): SystemLayout { .toLowerCase() if (raw === 'zero') return 'zero' if (raw === 'identity' || raw === 'id' || raw === 'block') return 'identity' - if (process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS) return 'zero' return 'stock' } -export function getKinTimezone(): string { +export function getKinTimezone(runtime?: KinRuntimeConfig): string { + if (runtime?.timezone) return runtime.timezone + if (process.env.CLAUDE_CODE_KIN_NATIVE_SLOTS) { + return readKinRuntimeConfig().timezone || 'America/New_York' + } const tz = ( process.env.CLAUDE_CODE_TIMEZONE || process.env.KIN_SLOT_TZ || @@ -40,26 +47,45 @@ export function environmentBlock(timezone = getKinTimezone()): string { return `# Environment\nTime zone: ${timezone}` } +function stableAttributionHeader(attribution: string): string { + // cch is replaced by Bun's request attestation on every request, and + // cc_workload is turn-scoped. Neither belongs before a message breakpoint: + // either value invalidates the whole native conversation cache prefix. + return attribution + .replace(/\s+cch=[^;\s]+;/gi, '') + .replace(/\s+cc_workload=[^;\s]+;/gi, '') +} + export function layoutSystemBlocks(opts: { attribution: string leftover?: string + officialFullPrompt?: readonly string[] + runtime?: KinRuntimeConfig }): string[] { - const layout = getSystemLayout() - const attribution = (opts.attribution || '').trim() + const layout = getSystemLayout(opts.runtime) + const persona = + opts.runtime?.persona_preset ?? (layout === 'zero' ? 'zero' : 'official') + const attribution = stableAttributionHeader((opts.attribution || '').trim()) const leftover = opts.leftover?.trim() - const env = environmentBlock() const blocks: string[] = [] - if (layout === 'zero') { + if (persona === 'zero' || layout === 'zero') { const billing = attribution - ? attribution.replace(/;?\s*$/, '') + `; prompt_version=${IDENTITY};` - : `x-anthropic-billing-header: prompt_version=${IDENTITY};` + ? attribution.replace(/;?\s*$/, '') + + `; prompt_version=${COMPACT_IDENTITY};` + : `x-anthropic-billing-header: prompt_version=${COMPACT_IDENTITY};` blocks.push(billing) - } else { - if (attribution) blocks.push(attribution) - blocks.push(IDENTITY) + if (leftover) blocks.push(leftover) + return blocks + } + + if (attribution) blocks.push(attribution) + blocks.push(IDENTITY) + if (persona === 'official_full') { + const full = opts.officialFullPrompt?.filter(Boolean).join('\n\n').trim() + if (full) blocks.push(full) + else blocks.push(environmentBlock(getKinTimezone(opts.runtime))) } - blocks.push(env) if (leftover) blocks.push(leftover) return blocks } diff --git a/src/services/api/__tests__/claude-cache-breakpoints.test.ts b/src/services/api/__tests__/claude-cache-breakpoints.test.ts new file mode 100644 index 0000000000..3bcfe9d032 --- /dev/null +++ b/src/services/api/__tests__/claude-cache-breakpoints.test.ts @@ -0,0 +1,54 @@ +import { describe, expect, test } from 'bun:test' +import { addCacheBreakpoints, buildSystemPromptBlocks } from '../claude' +import { asSystemPrompt } from '../../../utils/systemPromptType' + +function message(type: 'user' | 'assistant', text: string) { + return { type, message: { role: type, content: text } } as never +} + +describe('native Messages cache breakpoints', () => { + test('marks only the current tail for native Messages', () => { + const result = addCacheBreakpoints( + [ + message('user', 'u1'), + message('assistant', 'a1'), + message('user', 'u2'), + message('assistant', 'a2'), + message('user', 'u3'), + ], + true, + 'kin_native_messages', + false, + null, + [], + false, + '1h', + ) + + const marked = result.flatMap((entry, index) => { + const content = Array.isArray(entry.content) ? entry.content : [] + return content.some( + block => + typeof block === 'object' && + block !== null && + 'cache_control' in block, + ) + ? [index] + : [] + }) + + expect(marked).toEqual([4]) + }) + test('does not add a system marker for native Messages', () => { + const result = buildSystemPromptBlocks( + asSystemPrompt(['stable system prompt']), + true, + { + querySource: 'kin_native_messages', + cacheTtl: '1h', + }, + ) + + expect(result.some(block => 'cache_control' in block)).toBe(false) + }) +}) diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index 023419a8aa..82981d1c8e 100644 --- a/src/services/api/claude.ts +++ b/src/services/api/claude.ts @@ -29,6 +29,7 @@ import { getAttributionHeader, getCLISyspromptPrefix, } from '../../constants/system.js' +import { getSystemPrompt } from '../../constants/prompts.js' import { getSystemLayout, layoutSystemBlocks, @@ -357,17 +358,20 @@ export function getPromptCachingEnabled(model: string): boolean { export function getCacheControl({ scope, querySource, + cacheTtl, }: { scope?: CacheScope querySource?: QuerySource + cacheTtl?: '5m' | '1h' } = {}): { type: 'ephemeral' - ttl?: '1h' + ttl?: '5m' | '1h' scope?: CacheScope } { + const ttl = cacheTtl ?? (should1hCacheTTL(querySource) ? '1h' : undefined) return { type: 'ephemeral', - ...(should1hCacheTTL(querySource) && { ttl: '1h' }), + ...(ttl !== undefined && { ttl }), ...(scope === 'global' && { scope }), } } @@ -588,6 +592,7 @@ export function userMessageToMessageParam( addCache = false, enablePromptCaching: boolean, querySource?: QuerySource, + cacheTtl?: '5m' | '1h', ): MessageParam { if (addCache) { if (typeof message.message!.content === 'string') { @@ -598,28 +603,22 @@ export function userMessageToMessageParam( type: 'text', text: message.message!.content, ...(enablePromptCaching && { - cache_control: getCacheControl({ querySource }), + cache_control: getCacheControl({ querySource, cacheTtl }), }), }, ], } - } else { - return { - role: 'user', - content: message.message!.content!.map((_, i) => ({ - ..._, - ...(i === message.message!.content!.length - 1 - ? enablePromptCaching - ? { cache_control: getCacheControl({ querySource }) } - : {} - : {}), - })), - } + } + return { + role: 'user', + content: message.message!.content!.map((block, i) => ({ + ...block, + ...(i === message.message!.content!.length - 1 && enablePromptCaching + ? { cache_control: getCacheControl({ querySource, cacheTtl }) } + : {}), + })), } } - // Clone array content to prevent in-place mutations (e.g., insertCacheEditsBlock's - // splice) from contaminating the original message. Without cloning, multiple calls - // to addCacheBreakpoints share the same array and each splices in duplicate cache_edits. return { role: 'user', content: (Array.isArray(message.message!.content) @@ -634,6 +633,7 @@ export function assistantMessageToMessageParam( addCache = false, enablePromptCaching: boolean, querySource?: QuerySource, + cacheTtl?: '5m' | '1h', ): MessageParam { if (addCache) { if (typeof message.message!.content === 'string') { @@ -644,31 +644,30 @@ export function assistantMessageToMessageParam( type: 'text', text: message.message!.content, ...(enablePromptCaching && { - cache_control: getCacheControl({ querySource }), + cache_control: getCacheControl({ querySource, cacheTtl }), }), }, ], } - } else { - return { - role: 'assistant', - content: message.message!.content!.map((_, i) => { - const contentBlock = stripGeminiProviderMetadata(_) - return { - ...contentBlock, - ...(i === message.message!.content!.length - 1 && - contentBlock.type !== 'thinking' && - contentBlock.type !== 'redacted_thinking' && - (feature('CONNECTOR_TEXT') - ? !isConnectorTextBlock(contentBlock) - : true) - ? enablePromptCaching - ? { cache_control: getCacheControl({ querySource }) } - : {} - : {}), - } - }), - } + } + return { + role: 'assistant', + content: message.message!.content!.map((block, i) => { + const contentBlock = stripGeminiProviderMetadata(block) + const cacheable = + i === message.message!.content!.length - 1 && + contentBlock.type !== 'thinking' && + contentBlock.type !== 'redacted_thinking' && + (feature('CONNECTOR_TEXT') + ? !isConnectorTextBlock(contentBlock) + : true) + return { + ...contentBlock, + ...(cacheable && enablePromptCaching + ? { cache_control: getCacheControl({ querySource, cacheTtl }) } + : {}), + } + }), } } return { @@ -718,21 +717,19 @@ export type Options = { topP?: number topK?: number stopSequences?: string[] + cacheTtl?: '5m' | '1h' + personaPreset?: 'zero' | 'official' | 'official_full' + runtimeConfig?: import('../../kin/runtimeConfig.js').KinRuntimeConfig effortValue?: EffortValue mcpTools: Tools hasPendingMcpServers?: boolean queryTracking?: QueryChainTracking - agentId?: AgentId // Only set for subagents + agentId?: AgentId outputFormat?: BetaJSONOutputFormat fastMode?: boolean advisorModel?: string addNotification?: (notif: Notification) => void - // API-side task budget (output_config.task_budget). Distinct from the - // tokenBudget.ts +500k auto-continue feature — this one is sent to the API - // so the model can pace itself. `remaining` is computed by the caller - // (query.ts decrements across the agentic loop). taskBudget?: { total: number; remaining?: number } - /** Langfuse root trace span for observability. No-op if null/undefined. */ langfuseTrace?: LangfuseSpan | null } @@ -822,6 +819,9 @@ export async function* queryKinMessagesWithStreaming({ toolSchemas, toolChoice, thinking, + cacheTtl, + personaPreset, + runtimeConfig, maxTokens, temperature, topP, @@ -835,6 +835,9 @@ export async function* queryKinMessagesWithStreaming({ toolSchemas: BetaToolUnion[] toolChoice?: BetaToolChoiceTool | BetaToolChoiceAuto thinking: ThinkingConfig + cacheTtl?: '5m' | '1h' + personaPreset?: 'zero' | 'official' | 'official_full' + runtimeConfig?: import('../../kin/runtimeConfig.js').KinRuntimeConfig maxTokens?: number temperature?: number topP?: number @@ -865,12 +868,14 @@ export async function* queryKinMessagesWithStreaming({ isNonInteractiveSession: true, hasAppendSystemPrompt: false, querySource: 'kin_native_messages', + cacheTtl, + personaPreset, + runtimeConfig, agents: [], mcpTools: [], }, }) } - /** * Determines if an LSP tool should be deferred (tool appears with defer_loading: true) * because LSP initialization is not yet complete. @@ -1498,14 +1503,26 @@ async function* queryModel( ) const injectChromeHere = useSearchExtraTools && hasChromeTools && !isMcpInstructionsDeltaEnabled() - - // filter(Boolean) works by converting each element to a boolean - empty strings become false and are filtered out. - const kinSystemLayout = getSystemLayout() + const kinSystemLayout = getSystemLayout(options.runtimeConfig) + const officialFullPrompt = + options.personaPreset === 'official_full' + ? await getSystemPrompt([], options.model, [], []) + : undefined if (kinSystemLayout !== 'stock') { systemPrompt = asSystemPrompt( layoutSystemBlocks({ attribution: getAttributionHeader(), leftover: leftoverFromSystemPrompt(systemPrompt), + officialFullPrompt, + runtime: + options.runtimeConfig ?? + (options.personaPreset + ? { + default_cache_ttl: options.cacheTtl ?? '1h', + persona_preset: options.personaPreset, + system_layout: kinSystemLayout, + } + : undefined), }), ) } else { @@ -1553,20 +1570,21 @@ async function* queryModel( const enablePromptCaching = options.enablePromptCaching ?? getPromptCachingEnabled(options.model) - const system = buildSystemPromptBlocks(systemPrompt, enablePromptCaching, { - skipGlobalCacheForSystemPrompt: needsToolBasedCacheMarker, - querySource: options.querySource, - }) + const systemBlocks = buildSystemPromptBlocks( + systemPrompt, + enablePromptCaching, + { + skipGlobalCacheForSystemPrompt: needsToolBasedCacheMarker, + querySource: options.querySource, + cacheTtl: options.cacheTtl, + }, + ) + const system = systemBlocks const useBetas = betas.length > 0 // Build minimal context for detailed tracing (when beta tracing is enabled) - // Note: The actual new_context message extraction is done in sessionTracing.ts using - // hash-based tracking per querySource (agent) from the messagesForAPI array const extraToolSchemas = [...(options.extraToolSchemas ?? [])] if (advisorModel) { - // Server tools must be in the tools array by API contract. Appended after - // toolSchemas (which carries the cache_control marker) so toggling /advisor - // only churns the small suffix, not the cached prefix. extraToolSchemas.push({ type: 'advisor_20260301', name: 'advisor', @@ -1875,6 +1893,7 @@ async function* queryModel( consumedCacheEdits as any, consumedPinnedEdits as any, options.skipCacheWrite, + options.cacheTtl, ), system, tools: allTools, @@ -3307,6 +3326,7 @@ export function addCacheBreakpoints( newCacheEdits?: CachedMCEditsBlock | null, pinnedEdits?: CachedMCPinnedEdits[], skipCacheWrite = false, + cacheTtl?: '5m' | '1h', ): MessageParam[] { logEvent('tengu_api_cache_breakpoints', { totalMessageCount: messages.length, @@ -3314,17 +3334,9 @@ export function addCacheBreakpoints( skipCacheWrite, }) - // Exactly one message-level cache_control marker per request. Mycro's - // turn-to-turn eviction (page_manager/index.rs: Index::insert) frees - // local-attention KV pages at any cached prefix position NOT in - // cache_store_int_token_boundaries. With two markers the second-to-last - // position is protected and its locals survive an extra turn even though - // nothing will ever resume from there — with one marker they're freed - // immediately. For fire-and-forget forks (skipCacheWrite) we shift the - // marker to the second-to-last message: that's the last shared-prefix - // point, so the write is a no-op merge on mycro (entry already exists) - // and the fork doesn't leave its own tail in the KVCC. Dense pages are - // refcounted and survive via the new hash either way. + // Keep exactly one message-level marker. The native runner already owns + // the stable system/tools prefix; adding extra message anchors makes the + // upstream choose the shorter shared prefix and freezes cache_read_tokens. const markerIndex = skipCacheWrite ? messages.length - 2 : messages.length - 1 const result = messages.map((msg, index) => { const addCache = index === markerIndex @@ -3334,6 +3346,7 @@ export function addCacheBreakpoints( addCache, enablePromptCaching, querySource, + cacheTtl, ) } return assistantMessageToMessageParam( @@ -3341,6 +3354,7 @@ export function addCacheBreakpoints( addCache, enablePromptCaching, querySource, + cacheTtl, ) }) @@ -3456,9 +3470,12 @@ export function buildSystemPromptBlocks( options?: { skipGlobalCacheForSystemPrompt?: boolean querySource?: QuerySource + cacheTtl?: '5m' | '1h' }, ): TextBlockParam[] { - // IMPORTANT: Do not add any more blocks for caching or you will get a 400 + // Native Messages keeps only the message-tail breakpoint. A system + // breakpoint would win as the shorter shared prefix and freeze cache reads. + const cacheSystemPrompt = options?.querySource !== 'kin_native_messages' return splitSysPromptPrefix(systemPrompt, { skipGlobalCacheForSystemPrompt: options?.skipGlobalCacheForSystemPrompt, }).map(block => { @@ -3466,10 +3483,12 @@ export function buildSystemPromptBlocks( type: 'text' as const, text: block.text, ...(enablePromptCaching && + cacheSystemPrompt && block.cacheScope !== null && { cache_control: getCacheControl({ scope: block.cacheScope, querySource: options?.querySource, + cacheTtl: options?.cacheTtl, }), }), } From a3b892917b19c107aae7b23ab1e0230b0ce1dd20 Mon Sep 17 00:00:00 2001 From: ted Date: Wed, 23 Sep 2026 09:46:47 +0800 Subject: [PATCH 3/4] fix(cache): preserve native prompt cache reads --- .../claude-cache-breakpoints.test.ts | 4 ++-- src/services/api/claude.ts | 21 ++++++++++++++----- 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/src/services/api/__tests__/claude-cache-breakpoints.test.ts b/src/services/api/__tests__/claude-cache-breakpoints.test.ts index 3bcfe9d032..03631fae55 100644 --- a/src/services/api/__tests__/claude-cache-breakpoints.test.ts +++ b/src/services/api/__tests__/claude-cache-breakpoints.test.ts @@ -7,7 +7,7 @@ function message(type: 'user' | 'assistant', text: string) { } describe('native Messages cache breakpoints', () => { - test('marks only the current tail for native Messages', () => { + test('marks the current tail and previous user turn', () => { const result = addCacheBreakpoints( [ message('user', 'u1'), @@ -37,7 +37,7 @@ describe('native Messages cache breakpoints', () => { : [] }) - expect(marked).toEqual([4]) + expect(marked).toEqual([2, 4]) }) test('does not add a system marker for native Messages', () => { const result = buildSystemPromptBlocks( diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index 82981d1c8e..da1f736052 100644 --- a/src/services/api/claude.ts +++ b/src/services/api/claude.ts @@ -3334,12 +3334,23 @@ export function addCacheBreakpoints( skipCacheWrite, }) - // Keep exactly one message-level marker. The native runner already owns - // the stable system/tools prefix; adding extra message anchors makes the - // upstream choose the shorter shared prefix and freezes cache_read_tokens. - const markerIndex = skipCacheWrite ? messages.length - 2 : messages.length - 1 + // Keep the current tail and the previous user turn marked. The current tail + // creates the next cache entry; the previous user marker lets the next turn + // read the prefix that already existed before its new user message. + const markerIndices = new Set() + const tailIndex = skipCacheWrite ? messages.length - 2 : messages.length - 1 + if (tailIndex >= 0) markerIndices.add(tailIndex) + if (!skipCacheWrite && messages.length >= 4) { + const userIndices = messages + .map((message, index) => (message.type === 'user' ? index : -1)) + .filter(index => index >= 0) + const penultimateUserIndex = userIndices[userIndices.length - 2] + if (penultimateUserIndex !== undefined) { + markerIndices.add(penultimateUserIndex) + } + } const result = messages.map((msg, index) => { - const addCache = index === markerIndex + const addCache = markerIndices.has(index) if (msg.type === 'user') { return userMessageToMessageParam( msg, From 69b7b804b03245a1dca75eac4328ad0f8c903db9 Mon Sep 17 00:00:00 2001 From: ted Date: Wed, 23 Sep 2026 10:15:46 +0800 Subject: [PATCH 4/4] chore(release): prepare v2.8.5 --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 79239d07e1..93f69ba005 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "claude-code-best", - "version": "2.8.4", + "version": "2.8.5", "description": "Reverse-engineered Anthropic Claude Code CLI — interactive AI coding assistant in the terminal", "type": "module", "author": "claude-code-best ",