diff --git a/.changeset/wire-request-trace.md b/.changeset/wire-request-trace.md new file mode 100644 index 000000000..12e5d8bdb --- /dev/null +++ b/.changeset/wire-request-trace.md @@ -0,0 +1,7 @@ +--- +"@moonshot-ai/agent-core": minor +"@moonshot-ai/kimi-code": minor +"@moonshot-ai/kosong": minor +--- + +Record a request trace in each agent's `wire.jsonl`: content-addressed snapshots of the tool schemas sent to the model, one record per model request (including retries, strict resends, and compaction rounds) with the effective request parameters, and the raw MCP tool listing per server, so sessions carry enough data to reconstruct every model request for debugging. Chat providers now expose the effective completion-token cap they send on the wire, so the trace records the provider-clamped value rather than the requested budget. diff --git a/apps/vis/server/src/lib/context-projector.ts b/apps/vis/server/src/lib/context-projector.ts index 4cb2468ab..df9271b3a 100644 --- a/apps/vis/server/src/lib/context-projector.ts +++ b/apps/vis/server/src/lib/context-projector.ts @@ -495,7 +495,9 @@ export function projectContext( case 'swarm_mode.exit': swarm = { active: false }; break; - // Kinds that don't affect the projected timeline / derived state: + // Kinds that don't affect the projected timeline / derived state, + // including the observability records (request trace — `llm.*`, + // `mcp.tools_discovered`), which are never part of context state: case 'metadata': case 'forked': case 'turn.prompt': @@ -509,6 +511,9 @@ export function projectContext( case 'tools.unregister_user_tool': case 'tools.set_active_tools': case 'tools.update_store': + case 'llm.tools_snapshot': + case 'llm.request': + case 'mcp.tools_discovered': break; default: { const _exhaustive: never = rec; diff --git a/apps/vis/web/src/components/wire/renderers.tsx b/apps/vis/web/src/components/wire/renderers.tsx index 85d6160a8..6ec3eb67f 100644 --- a/apps/vis/web/src/components/wire/renderers.tsx +++ b/apps/vis/web/src/components/wire/renderers.tsx @@ -592,6 +592,153 @@ export const WIRE_RENDERERS: RendererMap = { label: 'goal×', headline: () => ({ main: goal cleared }), }, + + // Observability records — the request trace (see agent-core records/types.ts). + + 'llm.tools_snapshot': { + tone: 'tools', + label: 'req·tools', + headline: (r) => ({ + main: ( + + + {r.tools.length} tool{r.tools.length === 1 ? '' : 's'} + + + {r.tools + .slice(0, 4) + .map((tool) => tool.name) + .join(', ')} + {r.tools.length > 4 ? ` +${r.tools.length - 4} more` : ''} + + + ), + right: #{r.hash.slice(0, 8)}, + }), + }, + + 'llm.request': { + tone: 'meta', + label: 'llm→', + headline: (r) => { + const parts: string[] = []; + if (r.turnStep !== undefined) parts.push(`step ${r.turnStep}`); + if (r.attempt !== undefined) parts.push(`attempt ${r.attempt}`); + parts.push(`${r.messageCount} msgs`); + if (r.maxTokens !== undefined) parts.push(`max ${r.maxTokens} tok`); + return { + main: ( + + + {r.kind} + + {r.model} + {parts.join(' · ')} + + ), + right: + r.projection !== undefined ? ( + + {r.projection} + + ) : undefined, + }; + }, + detail: (r) => ( +
+ + {r.provider} + + + {r.model} + + {r.modelAlias !== undefined ? ( + + {r.modelAlias} + + ) : null} + {r.thinkingEffort !== undefined ? ( + + {r.thinkingEffort} + + ) : null} + {r.thinkingKeep !== undefined ? ( + + {r.thinkingKeep} + + ) : null} + {r.temperature !== undefined ? ( + + {r.temperature} + + ) : null} + {r.topP !== undefined ? ( + + {r.topP} + + ) : null} + {r.maxTokens !== undefined ? ( + + {r.maxTokens} + + ) : null} + {r.betaApi !== undefined ? ( + + {String(r.betaApi)} + + ) : null} + + {String(r.toolSelect)} + + + {r.toolsHash} + + + {r.systemPromptHash} + + {r.systemPrompt !== undefined ? ( + + + 
{r.systemPrompt}
+ + + ) : null} + {r.droppedCount !== undefined ? ( + + {r.droppedCount} + + ) : null} +
+ ), + }, + + 'mcp.tools_discovered': { + tone: 'tools', + label: 'mcp·list', + headline: (r) => ({ + main: ( + + {r.serverName} + + {r.tools.length} tool{r.tools.length === 1 ? '' : 's'} · {r.enabledNames.length}{' '} + enabled + + + ), + right: + r.collisions !== undefined && r.collisions.length > 0 ? ( + + {r.collisions.length} collision{r.collisions.length === 1 ? '' : 's'} + + ) : ( + #{r.hash.slice(0, 8)} + ), + }), + }, }; /** Look up a renderer by a runtime `type` string. Returns `undefined` for kinds diff --git a/docs/en/guides/sessions.md b/docs/en/guides/sessions.md index 15c56fcca..2d605b153 100644 --- a/docs/en/guides/sessions.md +++ b/docs/en/guides/sessions.md @@ -22,7 +22,7 @@ All sessions are saved under `$KIMI_CODE_HOME/sessions/` (default: `~/.kimi-code ``` - `state.json`: session metadata such as title and creation time. -- `agents/*/wire.jsonl`: the agent event stream, used for session recovery and replay. +- `agents/*/wire.jsonl`: the agent event stream, used for session recovery and replay. It also carries a request trace — the tool schemas, request parameters, and MCP tool listings sent to the model — for debugging. ::: warning Do not manually edit files inside the `sessions/` directory — doing so may prevent sessions from being restored correctly. diff --git a/docs/zh/guides/sessions.md b/docs/zh/guides/sessions.md index 444fb4489..a9c781981 100644 --- a/docs/zh/guides/sessions.md +++ b/docs/zh/guides/sessions.md @@ -22,7 +22,7 @@ Kimi Code CLI 把每次对话持久化为一个「会话」，保留消息历史 ``` - `state.json`：会话标题、创建时间等元数据。 -- `agents/*/wire.jsonl`：Agent 事件流，用于会话恢复和回放。 +- `agents/*/wire.jsonl`：Agent 事件流，用于会话恢复和回放；同时记录发给模型的请求轨迹（工具 schema、请求参数、MCP 工具清单），便于调试。 ::: warning 注意 `sessions/` 目录下的文件请勿手动编辑，否则可能导致会话无法正常恢复。 diff --git a/packages/agent-core/src/agent/compaction/full.ts b/packages/agent-core/src/agent/compaction/full.ts index a87851c9e..0bbc12577 100644 --- a/packages/agent-core/src/agent/compaction/full.ts +++ b/packages/agent-core/src/agent/compaction/full.ts @@ -18,6 +18,7 @@ import { } from '@moonshot-ai/kosong'; import type { Agent } from '..'; +import type { GenerateOptionsWithRequestLogFields } from '../llm-request-logger'; import type { ContextMessage } from '../context/types'; import { collectLoadedDynamicToolNames, @@ -43,8 +44,7 @@ import { import { applyCompletionBudget, resolveCompletionBudget, -} from '../../utils/completion-budget'; -import { renderPrompt } from '../../utils/render-prompt'; +} from '../../utils/completion-budget';import { renderPrompt } from '../../utils/render-prompt'; import compactionInstructionTemplate from './compaction-instruction.md?raw'; import type { CompactionBeginData, CompactionResult } from './types'; import { @@ -514,13 +514,17 @@ export class FullCompaction { ]; const estimatedCompactionRequestTokens = this.estimateRequestTokens(messages); try { + const generateOptions: GenerateOptionsWithRequestLogFields = { + signal, + requestLogFields: { kind: 'compaction', droppedCount }, + }; const response = await this.agent.generate( provider, this.agent.config.systemPrompt, [...this.agent.tools.loopTools], messages, undefined, - { signal }, + generateOptions, ); if (response.finishReason === 'truncated') { throw new CompactionTruncatedError(); diff --git a/packages/agent-core/src/agent/index.ts b/packages/agent-core/src/agent/index.ts index e3463613b..6b9d7cf19 100644 --- a/packages/agent-core/src/agent/index.ts +++ b/packages/agent-core/src/agent/index.ts @@ -55,6 +55,7 @@ import { TurnFlow } from './turn'; import { KosongLLM } from './turn/kosong-llm'; import { UsageRecorder } from './usage'; import { LlmRequestLogger, splitGenerateOptions } from './llm-request-logger'; +import { LlmRequestRecorder } from './llm-request-recorder'; import { resolveCompletionBudget } from '../utils/completion-budget'; import type { Kaos } from '@moonshot-ai/kaos'; import type { ToolServices } from '../tools/support/services'; @@ -125,6 +126,7 @@ export class Agent { readonly experimentalFlags: ExperimentalFlagResolver; readonly llmRequestLogger: LlmRequestLogger; + readonly llmRequestRecorder: LlmRequestRecorder; readonly blobStore: BlobStore | undefined; readonly records: AgentRecords; readonly fullCompaction: FullCompaction; @@ -180,6 +182,7 @@ export class Agent { this.systemPromptContextProvider = options.systemPromptContextProvider; this.llmRequestLogger = new LlmRequestLogger(this.log); + this.llmRequestRecorder = new LlmRequestRecorder(this); this.blobStore = options.homedir ? new BlobStore({ blobsDir: join(options.homedir, 'blobs') }) : undefined; @@ -255,14 +258,27 @@ export class Agent { const { requestLogFields, generateOptions } = splitGenerateOptions(options); const modelAlias = this.config.modelAlias; const run = (requestOptions: Parameters[5]) => { - this.llmRequestLogger.logRequest({ - provider, - modelAlias, - systemPrompt, - tools, - messages: history, - fields: requestLogFields, - }); + // Mirror kosong generate()'s pre-flight abort check: a call whose + // signal is already aborted never reaches the wire (generate throws + // before dispatching), so it must not leave a request trace or a + // diagnostic log line claiming a request was sent. + if (requestOptions?.signal?.aborted !== true) { + this.llmRequestLogger.logRequest({ + provider, + modelAlias, + systemPrompt, + tools, + messages: history, + fields: requestLogFields, + }); + this.llmRequestRecorder.record({ + provider, + systemPrompt, + tools, + messages: history, + fields: requestLogFields, + }); + } return this.rawGenerate(provider, systemPrompt, tools, history, callbacks, requestOptions); }; if (generateOptions?.auth !== undefined) { diff --git a/packages/agent-core/src/agent/llm-request-logger.ts b/packages/agent-core/src/agent/llm-request-logger.ts index 1474546c0..d743b15c8 100644 --- a/packages/agent-core/src/agent/llm-request-logger.ts +++ b/packages/agent-core/src/agent/llm-request-logger.ts @@ -68,10 +68,10 @@ export function splitGenerateOptions(options: GenerateOptionsWithRequestLogField return { requestLogFields, generateOptions }; } -function toolSignature(tools: readonly Tool[]) { +export function toolSignature(tools: readonly Tool[]) { return tools.map(({ name, description, parameters }) => ({ name, description, parameters })); } -function fingerprint(content: string): string { +export function fingerprint(content: string): string { return createHash('sha256').update(content).digest('hex'); } diff --git a/packages/agent-core/src/agent/llm-request-recorder.ts b/packages/agent-core/src/agent/llm-request-recorder.ts new file mode 100644 index 000000000..4bb47e0ac --- /dev/null +++ b/packages/agent-core/src/agent/llm-request-recorder.ts @@ -0,0 +1,140 @@ +/** + * Durable request-trace recorder: writes the observability records + * (`llm.tools_snapshot`, `llm.request`) that make every outbound model + * request reconstructable from the wire log. Called from the single + * `Agent.generate` choke point, so loop steps, retry attempts, strict + * resends, and compaction rounds all leave a trace. + * + * Sibling of `LlmRequestLogger` (diagnostic log lines, hashes only); this + * class owns the wire-record side. See the observability-records note in + * `records/types.ts` for the persistence contract. + */ + +import { KimiChatProvider, type ChatProvider, type Message, type Tool } from '@moonshot-ai/kosong'; + +import { parseFloatEnv } from '#/config/resolve'; +import { resolveThinkingKeep } from '#/config/kimi-env-params'; + +import type { Agent } from '.'; +import type { LLMRequestLogFields } from '../loop'; +import { fingerprint, toolSignature } from './llm-request-logger'; + +export class LlmRequestRecorder { + /** Hashes of tool tables already durable in this wire log. */ + private readonly seenToolsHashes = new Set(); + /** + * Identity cache over the last wire tool table. Tool instances are treated + * as immutable and are stable across steps (rebuilt only by + * `initializeBuiltinTools` / MCP re-registration), so element-wise identity + * implies content equality — the common per-step path costs no hashing. + */ + private lastWireTools: readonly Tool[] | undefined; + private lastToolsHash: string | undefined; + private lastSystemPrompt: string | undefined; + private lastSystemPromptHash: string | undefined; + + constructor(private readonly agent: Agent) {} + + /** Replay: a snapshot with this hash is already durable; never re-log it. */ + restoreToolsSnapshot(hash: string): void { + this.seenToolsHashes.add(hash); + } + + record(input: { + readonly provider: ChatProvider; + readonly systemPrompt: string; + readonly tools: readonly Tool[]; + readonly messages: readonly Message[]; + readonly fields: LLMRequestLogFields | undefined; + }): void { + const { provider, systemPrompt, messages } = input; + const fields = input.fields ?? {}; + // Deferred tools are stripped by kosong generate() before the provider + // sees them; snapshot what actually goes on the wire. In disclosure mode + // this keeps the snapshot byte-stable across select_tools loads. + const wireTools = input.tools.filter((tool) => tool.deferred !== true); + const toolsHash = this.toolsHashFor(wireTools); + if (!this.seenToolsHashes.has(toolsHash)) { + this.seenToolsHashes.add(toolsHash); + this.agent.records.logRecord({ + type: 'llm.tools_snapshot', + hash: toolsHash, + tools: toolSignature(wireTools), + }); + } + + const modelAlias = this.agent.config.modelAlias; + // Mirror the ConfigState.provider pipeline for Kimi-only request params: + // env sampling overrides and the preserved-thinking keep passthrough + // reach the wire only for Kimi providers, resolved by the same exported + // helpers used at construction. thinkingEffort needs no mirroring — the + // Kimi provider derives it from the request body's thinking payload, so + // env effort overrides are already reflected in the read value. + const isKimiProvider = provider instanceof KimiChatProvider; + this.agent.records.logRecord({ + type: 'llm.request', + kind: fields.kind ?? 'loop', + provider: provider.name, + model: provider.modelName, + modelAlias, + thinkingEffort: provider.thinkingEffort ?? undefined, + thinkingKeep: isKimiProvider + ? resolveThinkingKeep( + process.env, + this.agent.kimiConfig?.thinking?.keep, + provider.thinkingEffort ?? 'off', + ) + : undefined, + temperature: isKimiProvider + ? parseFloatEnv(process.env['KIMI_MODEL_TEMPERATURE'], 'KIMI_MODEL_TEMPERATURE') + : undefined, + topP: isKimiProvider + ? parseFloatEnv(process.env['KIMI_MODEL_TOP_P'], 'KIMI_MODEL_TOP_P') + : undefined, + maxTokens: provider.maxCompletionTokens, + betaApi: + modelAlias === undefined + ? undefined + : this.agent.kimiConfig?.models?.[modelAlias]?.betaApi, + toolSelect: this.agent.toolSelectEnabled, + systemPromptHash: this.systemPromptHashFor(systemPrompt), + systemPrompt: + systemPrompt === this.agent.config.systemPrompt ? undefined : systemPrompt, + toolsHash, + messageCount: messages.length, + turnStep: fields.turnStep, + attempt: fields.attempt, + projection: fields.projection, + droppedCount: fields.droppedCount, + }); + } + + private toolsHashFor(wireTools: readonly Tool[]): string { + if (this.lastToolsHash !== undefined && sameToolInstances(this.lastWireTools, wireTools)) { + return this.lastToolsHash; + } + const hash = fingerprint(JSON.stringify(toolSignature(wireTools))); + this.lastWireTools = wireTools; + this.lastToolsHash = hash; + return hash; + } + + private systemPromptHashFor(systemPrompt: string): string { + if (this.lastSystemPromptHash === undefined || systemPrompt !== this.lastSystemPrompt) { + this.lastSystemPrompt = systemPrompt; + this.lastSystemPromptHash = fingerprint(systemPrompt); + } + return this.lastSystemPromptHash; + } +} + +function sameToolInstances( + previous: readonly Tool[] | undefined, + current: readonly Tool[], +): boolean { + if (previous === undefined || previous.length !== current.length) return false; + for (let i = 0; i < current.length; i++) { + if (previous[i] !== current[i]) return false; + } + return true; +} diff --git a/packages/agent-core/src/agent/records/index.ts b/packages/agent-core/src/agent/records/index.ts index 1e7cd035c..6873c5277 100644 --- a/packages/agent-core/src/agent/records/index.ts +++ b/packages/agent-core/src/agent/records/index.ts @@ -129,6 +129,17 @@ function restoreAgentRecord(agent: Agent, input: AgentRecord): void { case 'goal.clear': agent.goal.restoreClear(input); return; + // Observability records: no state to rebuild; only restore the + // write-dedup cursors so a resumed session does not re-log snapshots + // that are already durable in this wire log. + case 'llm.tools_snapshot': + agent.llmRequestRecorder.restoreToolsSnapshot(input.hash); + return; + case 'llm.request': + return; + case 'mcp.tools_discovered': + agent.tools.restoreMcpDiscovery(input.serverName, input.hash); + return; } } @@ -143,6 +154,17 @@ export interface AgentRecordsReplayOptions { export class AgentRecords { private _restoring: RestoringContext | null = null; private metadataInitialized = false; + private _replaying = false; + /** + * One-shot latch: the durable log is "open" once replay has completed (the + * write-dedup cursors of observability records are restored) or, for agents + * that never resume, once the first record has been logged live. Producers + * of observability records (MCP discovery) park their writes until then — + * logging earlier would both duplicate records that replay is about to + * dedupe and append a stray metadata record ahead of replay. + */ + private _opened = false; + private readonly onOpenedCallbacks: Array<() => void> = []; constructor( private readonly agent: Agent, @@ -153,6 +175,38 @@ export class AgentRecords { return this._restoring; } + /** + * Whether observability records may be written directly. False before the + * log is opened (see `_opened`); producers should park and re-attempt from + * an `onOpened` callback. Always true without persistence — there is no + * durable log to protect. + */ + get observabilityReady(): boolean { + return this.persistence === undefined || this._opened; + } + + /** + * Register a callback fired once, when the log opens. Not fired for a + * range-limited (frozen) replay — those agents are transient previews and + * must not append new records. + */ + onOpened(callback: () => void): void { + if (this._opened) { + callback(); + return; + } + this.onOpenedCallbacks.push(callback); + } + + private markOpened(): void { + if (this._opened) return; + this._opened = true; + const callbacks = this.onOpenedCallbacks.splice(0); + for (const callback of callbacks) { + callback(); + } + } + logRecord(record: AgentRecord): void { if (this._restoring !== null) return; const stamped: AgentRecord = @@ -173,6 +227,13 @@ export class AgentRecords { this.metadataInitialized = true; } this.persistence?.append(stamped); + // A live record was durably logged, so this agent is not waiting on a + // replay: open the log for observability producers. Guarded against the + // (currently hypothetical) mid-replay logRecord — opening there would + // let observability writes race the dedup-cursor restore. + if (!this._replaying) { + this.markOpened(); + } } restore(record: AgentRecord): boolean { @@ -194,46 +255,57 @@ export class AgentRecords { let warning: string | undefined; const replayedRecords: AgentRecord[] | undefined = rewriteMigratedRecords ? [] : undefined; let completed = true; - for await (const record of this.persistence.read()) { - if (!hasMetadata) { - if (record.type !== 'metadata') { - throw new Error('AgentRecords replay expected metadata as the first record'); + this._replaying = true; + try { + for await (const record of this.persistence.read()) { + if (!hasMetadata) { + if (record.type !== 'metadata') { + throw new Error('AgentRecords replay expected metadata as the first record'); + } + hasMetadata = true; + this.metadataInitialized = true; + const readVersion = record.protocol_version; + if (isNewerWireVersion(readVersion)) { + warning = `Session wire protocol version ${readVersion} is newer than the current version ${AGENT_WIRE_PROTOCOL_VERSION}. Records will be replayed without migration.`; + shouldRewrite = false; + } else { + migrations = resolveWireMigrations(readVersion); + shouldRewrite = readVersion !== AGENT_WIRE_PROTOCOL_VERSION; + } } - hasMetadata = true; - this.metadataInitialized = true; - const readVersion = record.protocol_version; - if (isNewerWireVersion(readVersion)) { - warning = `Session wire protocol version ${readVersion} is newer than the current version ${AGENT_WIRE_PROTOCOL_VERSION}. Records will be replayed without migration.`; - shouldRewrite = false; - } else { - migrations = resolveWireMigrations(readVersion); - shouldRewrite = readVersion !== AGENT_WIRE_PROTOCOL_VERSION; + let migratedRecord = migrateWireRecord( + record as WireMigrationRecord, + migrations, + ) as AgentRecord; + if (migratedRecord.type === 'metadata') { + migratedRecord = { + ...migratedRecord, + protocol_version: AGENT_WIRE_PROTOCOL_VERSION, + }; + } + replayedRecords?.push(migratedRecord); + if (this.restore(migratedRecord)) { + completed = false; + break; } } - let migratedRecord = migrateWireRecord( - record as WireMigrationRecord, - migrations, - ) as AgentRecord; - if (migratedRecord.type === 'metadata') { - migratedRecord = { - ...migratedRecord, - protocol_version: AGENT_WIRE_PROTOCOL_VERSION, - }; + if (completed && shouldRewrite && replayedRecords !== undefined) { + this.persistence.rewrite(replayedRecords); + await this.persistence.flush(); } - replayedRecords?.push(migratedRecord); - if (this.restore(migratedRecord)) { - completed = false; - break; + if (completed && this.agent.blobStore !== undefined) { + for (const msg of this.agent.context.history) { + await this.agent.blobStore.rehydrateParts(msg.content); + } } + } finally { + this._replaying = false; } - if (completed && shouldRewrite && replayedRecords !== undefined) { - this.persistence.rewrite(replayedRecords); - await this.persistence.flush(); - } - if (completed && this.agent.blobStore !== undefined) { - for (const msg of this.agent.context.history) { - await this.agent.blobStore.rehydrateParts(msg.content); - } + // Open only AFTER the migration rewrite has flushed — records appended by + // onOpened callbacks before the rewrite would be wiped by it. A frozen + // (range-limited) replay never opens: see onOpened. + if (completed) { + this.markOpened(); } return { warning }; } diff --git a/packages/agent-core/src/agent/records/types.ts b/packages/agent-core/src/agent/records/types.ts index eb07d10a0..8fffe6abf 100644 --- a/packages/agent-core/src/agent/records/types.ts +++ b/packages/agent-core/src/agent/records/types.ts @@ -1,20 +1,38 @@ -import type { ContentPart, TokenUsage } from '@moonshot-ai/kosong'; +import type { ContentPart, ThinkingEffort, TokenUsage } from '@moonshot-ai/kosong'; import type { LoopRecordedEvent } from '../../loop'; import type { GoalActor, GoalBudgetLimits, GoalStatus } from '../goal'; +import type { MCPToolDefinition } from '../../mcp/types'; import type { ToolStoreUpdate } from '../../tools/store'; import type { CompactionBeginData, CompactionResult } from '../compaction'; import type { AgentConfigUpdateData } from '../config'; import type { ContextMessage, PromptOrigin } from '../context'; import type { PermissionApprovalResultRecord, PermissionMode } from '../permission'; -import type { UserToolRegistration } from '../tool'; +import type { McpToolCollision, UserToolRegistration } from '../tool'; import type { UsageRecordScope } from '../usage'; import type { SwarmModeTrigger } from '../swarm'; +/** One entry of a tools table as sent in a request's top-level `tools[]`. */ +export interface LlmRequestToolSchema { + name: string; + description: string; + parameters: Record; +} + // Agent records are the ordered event log used to rebuild agent state on resume. // Use records, not state.json, when correctness depends on the order in which -// state transitions happened. Each persisted record type must have explicit -// resume semantics in restoreAgentRecord; a write-only record is not persistence. +// state transitions happened. +// +// Two record classes exist, and being persisted is not the same as being +// replayed: +// - State records (the default): each type must have explicit state-rebuild +// semantics in restoreAgentRecord; a write-only state record is not +// persistence. +// - Observability records (`llm.tools_snapshot`, `llm.request`, +// `mcp.tools_discovered`): a durable trace of the data sent to the model, +// for debugging and trajectory replay. They never feed state rebuild; +// their only resume semantics is restoring the write-dedup cursors so a +// resumed session does not re-log snapshots it already persisted. export interface AgentRecordEvents { metadata: { protocol_version: string; @@ -98,6 +116,85 @@ export interface AgentRecordEvents { actor?: GoalActor; }; 'goal.clear': {}; + + // Observability records (see the header note): request-trace data, not + // state. Resume only restores the write-dedup cursors. + + /** + * Content-addressed snapshot of a request's top-level `tools[]` (after the + * `deferred` strip — exactly what the provider receives). Written once per + * unique table; `llm.request.toolsHash` points here. + */ + 'llm.tools_snapshot': { + hash: string; + tools: readonly LlmRequestToolSchema[]; + }; + + /** + * One record per outbound model request (every retry attempt, strict + * resend, and compaction round included). Together with `config.update` + * (system prompt full text), context records (messages), and + * `llm.tools_snapshot` (tool schemas), this makes each request + * reconstructable from the wire log at the logical-request level. + */ + 'llm.request': { + kind: 'loop' | 'compaction'; + provider: string; + model: string; + modelAlias?: string; + /** + * Provider-effective thinking effort — for Kimi providers this is derived + * from the request body's thinking payload, so env overrides + * (`KIMI_MODEL_THINKING_EFFORT`) are already reflected. + */ + thinkingEffort?: ThinkingEffort; + /** + * Kimi preserved-thinking passthrough (`thinking.keep`) in effect for + * this request — resolved from env, config, and the default, none of + * which are otherwise recorded. + */ + thinkingKeep?: string; + /** Effective env-driven sampling overrides (Kimi provider only). */ + temperature?: number; + topP?: number; + /** + * Effective completion-token cap the provider sends on the wire — read + * from the effective provider, so provider-side clamping (remaining + * context window, transport ceilings) and provider-level defaults (e.g. + * Anthropic's required `max_tokens`) are included. + */ + maxTokens?: number; + betaApi?: boolean; + /** Progressive tool disclosure in effect (env flag × model capability). */ + toolSelect: boolean; + systemPromptHash: string; + /** + * Inlined only when the request's system prompt differs from the current + * `config.update` value (no such caller today; defensive for future ones). + */ + systemPrompt?: string; + toolsHash: string; + messageCount: number; + turnStep?: string; + attempt?: string; + /** Set when this request is the strict wire-compliant rebuild resend. */ + projection?: 'strict'; + /** Compaction only: messages dropped so far by overflow/empty shrinking. */ + droppedCount?: number; + }; + + /** + * Raw MCP `tools/list` result as advertised by the server, plus how this + * agent gated it (allow-list, name collisions). Written on registration, + * deduplicated per server by content hash. + */ + 'mcp.tools_discovered': { + serverName: string; + hash: string; + tools: readonly MCPToolDefinition[]; + enabledNames: readonly string[]; + collisions?: readonly McpToolCollision[]; + }; } export type AgentRecord = { diff --git a/packages/agent-core/src/agent/tool/index.ts b/packages/agent-core/src/agent/tool/index.ts index 9219713d2..d0a6d6b26 100644 --- a/packages/agent-core/src/agent/tool/index.ts +++ b/packages/agent-core/src/agent/tool/index.ts @@ -12,9 +12,10 @@ import { createMcpAuthTool } from '../../mcp/auth-tool'; import type { McpConnectionManager, McpServerEntry } from '../../mcp'; import { mcpResultToExecutableOutput } from '../../mcp/output'; import { isMcpToolName, qualifyMcpToolName } from '../../mcp/tool-naming'; -import type { MCPClient } from '../../mcp/types'; +import type { MCPClient, MCPToolDefinition } from '../../mcp/types'; import { DEFAULT_AGENT_PROFILES } from '../../profile'; import { extendWorkspaceWithSkillRoots } from '../../skill'; +import { fingerprint } from '../llm-request-logger'; import * as b from '../../tools/builtin'; import type { ToolStore, ToolStoreData, ToolStoreKey } from '../../tools/store'; import type { @@ -35,6 +36,13 @@ interface McpToolEntry { readonly serverName: string; } +interface PendingMcpDiscovery { + readonly serverName: string; + readonly rawTools: readonly MCPToolDefinition[]; + readonly enabledNames: readonly string[]; + readonly collisions: readonly McpToolCollision[]; +} + export class ToolManager { protected builtinTools: Map = new Map(); protected readonly userTools: Map = new Map(); @@ -55,6 +63,20 @@ export class ToolManager { private readonly pendingLoadedDynamicTools = new Set(); protected readonly store: Partial = {}; private mcpToolStatusUnsubscribe: (() => void) | undefined; + /** + * `serverName\nhash` keys of `mcp.tools_discovered` records already durable + * in this wire log. Restored on replay; reconnects with an unchanged raw + * tool list, allow-list, and collision outcome do not re-log. + */ + private readonly seenMcpDiscoveries = new Set(); + /** + * Discoveries observed before the record log opened (constructor-time + * attach can run before `agent.resume()` replays the wire — see + * `AgentRecords.observabilityReady`). The dedup decision must be re-made at + * drain time, after replay has restored `seenMcpDiscoveries`. + */ + private readonly pendingMcpDiscoveries: PendingMcpDiscovery[] = []; + private mcpDiscoveryDrainSubscribed = false; /** Abort controllers for in-flight `!` shell commands, keyed by commandId so * the TUI can cancel (Esc / Ctrl+C) a running command. */ @@ -386,6 +408,12 @@ export class ToolManager { resolved.tools, resolved.enabledNames, ); + this.recordMcpToolsDiscovered( + entry.name, + resolved.rawTools, + resolved.enabledNames, + result.collisions, + ); this.emitMcpToolCollisions(entry.name, result.collisions); this.agent.emitEvent({ type: 'tool.list.updated', @@ -394,6 +422,71 @@ export class ToolManager { }); } + /** Replay: a discovery with this hash is already durable; never re-log it. */ + restoreMcpDiscovery(serverName: string, hash: string): void { + this.seenMcpDiscoveries.add(`${serverName}\n${hash}`); + } + + /** + * Observability record: the server's verbatim `tools/list` result plus how + * this agent gated it (allow-list, collisions). See `records/types.ts`. + * Parked while the record log has not opened yet (pre-replay window). + */ + private recordMcpToolsDiscovered( + serverName: string, + rawTools: readonly MCPToolDefinition[], + enabledNames: ReadonlySet, + collisions: readonly McpToolCollision[], + ): void { + const discovery: PendingMcpDiscovery = { + serverName, + rawTools, + enabledNames: [...enabledNames].toSorted((a, b) => a.localeCompare(b)), + collisions, + }; + if (!this.agent.records.observabilityReady) { + this.pendingMcpDiscoveries.push(discovery); + // Lazy one-shot subscription: only agents that actually parked need + // the drain callback, and at park time the log is guaranteed unopened. + if (!this.mcpDiscoveryDrainSubscribed) { + this.mcpDiscoveryDrainSubscribed = true; + this.agent.records.onOpened(() => { + this.drainPendingMcpDiscoveries(); + }); + } + return; + } + this.writeMcpDiscovery(discovery); + } + + private drainPendingMcpDiscoveries(): void { + const pending = this.pendingMcpDiscoveries.splice(0); + for (const discovery of pending) { + this.writeMcpDiscovery(discovery); + } + } + + private writeMcpDiscovery(discovery: PendingMcpDiscovery): void { + const { serverName, rawTools, enabledNames, collisions } = discovery; + // The hash covers everything the record captures — the raw list, the + // allow-list, AND the collision outcome. Collisions depend on which + // other servers hold a qualified name at registration time, so the same + // server can re-register with identical tools but a different outcome; + // that change must produce a new record. + const hash = fingerprint(JSON.stringify({ tools: rawTools, enabledNames, collisions })); + const key = `${serverName}\n${hash}`; + if (this.seenMcpDiscoveries.has(key)) return; + this.seenMcpDiscoveries.add(key); + this.agent.records.logRecord({ + type: 'mcp.tools_discovered', + serverName, + hash, + tools: rawTools, + enabledNames, + collisions: collisions.length > 0 ? collisions : undefined, + }); + } + private emitMcpToolCollisions(serverName: string, collisions: readonly McpToolCollision[]): void { if (collisions.length === 0) return; const summary = collisions diff --git a/packages/agent-core/src/loop/llm.ts b/packages/agent-core/src/loop/llm.ts index 9d58dce7c..accac0e4a 100644 --- a/packages/agent-core/src/loop/llm.ts +++ b/packages/agent-core/src/loop/llm.ts @@ -23,9 +23,20 @@ export interface ToolCallDelta { readonly argumentsPart?: string | undefined; } +/** + * Request-scoped side channel from the host layers (loop, LLM adapter, + * compaction) down to the `Agent.generate` choke point, consumed there by the + * diagnostic logger and the wire-record request trace. + */ export interface LLMRequestLogFields { - readonly turnStep: string; + readonly turnStep?: string; readonly attempt?: string; + /** Request purpose; absent means a regular loop step. */ + readonly kind?: 'loop' | 'compaction'; + /** Set when the messages are the strict wire-compliant rebuild resend. */ + readonly projection?: 'strict'; + /** Compaction only: messages dropped so far by overflow/empty shrinking. */ + readonly droppedCount?: number; } export interface LLMStreamTiming { diff --git a/packages/agent-core/src/loop/retry.ts b/packages/agent-core/src/loop/retry.ts index 199b409e3..0d26781d4 100644 --- a/packages/agent-core/src/loop/retry.ts +++ b/packages/agent-core/src/loop/retry.ts @@ -88,12 +88,18 @@ function paramsForAttempt( maxAttempts: number, ): LLMChatParams { const turnStep = `${input.turnId}.${String(input.currentStep)}`; + // Preserve caller-set fields (e.g. the strict-resend projection marker); + // only the per-attempt turnStep/attempt pair is owned here. return { ...input.params, requestLogFields: attempt === 1 - ? { turnStep } - : { turnStep, attempt: `${String(attempt)}/${String(maxAttempts)}` }, + ? { ...input.params.requestLogFields, turnStep } + : { + ...input.params.requestLogFields, + turnStep, + attempt: `${String(attempt)}/${String(maxAttempts)}`, + }, }; } diff --git a/packages/agent-core/src/loop/turn-step.ts b/packages/agent-core/src/loop/turn-step.ts index dc81ca53a..74a3899b8 100644 --- a/packages/agent-core/src/loop/turn-step.ts +++ b/packages/agent-core/src/loop/turn-step.ts @@ -162,7 +162,11 @@ export async function executeLoopStep(deps: ExecuteLoopStepDeps): Promise<{ try { response = await chatWithRetry({ ...retryInput, - params: { ...chatParams, messages: strictMessages }, + params: { + ...chatParams, + messages: strictMessages, + requestLogFields: { projection: 'strict' }, + }, }); } catch (strictError) { // The strictly-sanitized rebuild was still rejected — our wire-compliance diff --git a/packages/agent-core/src/mcp/connection-manager.ts b/packages/agent-core/src/mcp/connection-manager.ts index f5c011fae..a9bc5e447 100644 --- a/packages/agent-core/src/mcp/connection-manager.ts +++ b/packages/agent-core/src/mcp/connection-manager.ts @@ -11,7 +11,7 @@ import { SseMcpClient } from './client-sse'; import type { UnexpectedCloseReason } from './client-shared'; import { StdioMcpClient } from './client-stdio'; import type { McpOAuthService } from './oauth'; -import { assertMcpInputSchema, type MCPClient } from './types'; +import { assertMcpInputSchema, type MCPClient, type MCPToolDefinition } from './types'; export type McpServerStatus = 'pending' | 'connected' | 'failed' | 'disabled' | 'needs-auth'; @@ -29,6 +29,8 @@ interface InternalEntry { attemptId: number; status: McpServerStatus; tools?: readonly Tool[]; + /** Verbatim `tools/list` result the converted {@link tools} came from. */ + rawTools?: readonly MCPToolDefinition[]; enabledNames?: ReadonlySet; error?: string; client?: RuntimeMcpClient; @@ -136,12 +138,18 @@ export class McpConnectionManager { resolved( name: string, ): - | { client: MCPClient; tools: readonly Tool[]; enabledNames: ReadonlySet } + | { + client: MCPClient; + tools: readonly Tool[]; + rawTools: readonly MCPToolDefinition[]; + enabledNames: ReadonlySet; + } | undefined { const entry = this.entries.get(name); if ( entry?.status !== 'connected' || entry.tools === undefined || + entry.rawTools === undefined || entry.client === undefined ) { return undefined; @@ -149,6 +157,7 @@ export class McpConnectionManager { return { client: entry.client, tools: entry.tools, + rawTools: entry.rawTools, enabledNames: entry.enabledNames ?? new Set(entry.tools.map((t) => t.name)), }; } @@ -191,6 +200,7 @@ export class McpConnectionManager { await this.closeClient(entry); entry.status = 'disabled'; entry.tools = undefined; + entry.rawTools = undefined; entry.enabledNames = undefined; entry.error = undefined; this.emit(entry); @@ -242,6 +252,7 @@ export class McpConnectionManager { if (!this.isCurrent(entry, attemptId)) return; entry.status = 'pending'; entry.tools = undefined; + entry.rawTools = undefined; entry.enabledNames = undefined; entry.error = undefined; this.emit(entry); @@ -263,7 +274,7 @@ export class McpConnectionManager { const startupClient = this.createClient(entry.config, entry.name); client = startupClient; entry.client = startupClient; - const tools = await withTimeout( + const discovered = await withTimeout( this.connectAndDiscoverTools(startupClient), timeoutMs, () => { @@ -275,8 +286,9 @@ export class McpConnectionManager { await this.closeRuntimeClient(startupClient); return; } - entry.tools = tools; - entry.enabledNames = computeEnabledNames(entry.config, tools); + entry.tools = discovered.tools; + entry.rawTools = discovered.rawTools; + entry.enabledNames = computeEnabledNames(entry.config, discovered.tools); entry.status = 'connected'; this.watchForUnexpectedClose(entry, startupClient, attemptId); } catch (error) { @@ -294,6 +306,7 @@ export class McpConnectionManager { entry.error = formatStartupError(error, client); } entry.tools = undefined; + entry.rawTools = undefined; entry.enabledNames = undefined; // Drop the client reference so a later reconnect builds a fresh one. await this.closeClient(entry); @@ -315,6 +328,7 @@ export class McpConnectionManager { entry.status = 'failed'; entry.error = formatUnexpectedCloseError(entry.name, reason); entry.tools = undefined; + entry.rawTools = undefined; entry.enabledNames = undefined; entry.client = undefined; // Best-effort close; the transport is already gone, but this lets the @@ -376,14 +390,19 @@ export class McpConnectionManager { return isUnauthorizedLikeError(error); } - private async connectAndDiscoverTools(client: RuntimeMcpClient): Promise { + private async connectAndDiscoverTools( + client: RuntimeMcpClient, + ): Promise<{ tools: Tool[]; rawTools: MCPToolDefinition[] }> { await client.connect(); const mcpTools = await client.listTools(); - return mcpTools.map((mcpTool) => ({ - name: mcpTool.name, - description: mcpTool.description, - parameters: assertMcpInputSchema(mcpTool.name, mcpTool.inputSchema), - })); + return { + rawTools: mcpTools, + tools: mcpTools.map((mcpTool) => ({ + name: mcpTool.name, + description: mcpTool.description, + parameters: assertMcpInputSchema(mcpTool.name, mcpTool.inputSchema), + })), + }; } private async closeClient(entry: InternalEntry): Promise { diff --git a/packages/agent-core/src/session/provider-manager.ts b/packages/agent-core/src/session/provider-manager.ts index 4c93fbe3f..b67aef318 100644 --- a/packages/agent-core/src/session/provider-manager.ts +++ b/packages/agent-core/src/session/provider-manager.ts @@ -39,8 +39,6 @@ interface ProviderManagerOptions { readonly kimiRequestHeaders?: Record; readonly resolveOAuthTokenProvider?: OAuthTokenProviderResolver; readonly promptCacheKey?: string; - // remove before commit - readonly adaptiveThinkingOverride?: () => boolean | undefined; } type AuthorizedRequest = ( @@ -121,9 +119,6 @@ export class ProviderManager implements ModelProvider { ); } - // remove before commit - const adaptiveThinkingOverride = this.options.adaptiveThinkingOverride?.(); - const effectiveAdaptiveThinking = adaptiveThinkingOverride ?? effectiveAlias.adaptiveThinking; const provider = toKosongProviderConfig( providerConfig, alias.model, @@ -132,7 +127,7 @@ export class ProviderManager implements ModelProvider { effectiveAlias.maxOutputSize, effectiveAlias.reasoningKey, this.options.promptCacheKey, - effectiveAdaptiveThinking, + effectiveAlias.adaptiveThinking, alias.betaApi, effectiveAlias.supportEfforts, ); diff --git a/packages/agent-core/test/agent/basic.test.ts b/packages/agent-core/test/agent/basic.test.ts index f7b73da4f..11b318cea 100644 --- a/packages/agent-core/test/agent/basic.test.ts +++ b/packages/agent-core/test/agent/basic.test.ts @@ -27,6 +27,8 @@ it('runs a text-only agent turn from prompt to completion', async () => { [wire] context.append_message { "message": { "role": "user", "content": [ { "type": "text", "text": "Hello" } ], "toolCalls": [], "origin": { "kind": "user" } }, "time": "