diff --git a/.changeset/align-print-background-policy.md b/.changeset/align-print-background-policy.md new file mode 100644 index 0000000000..a9bbd568e0 --- /dev/null +++ b/.changeset/align-print-background-policy.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Align the print-mode run lifecycle across engines: `print_background_mode` and `print_max_turns` now take effect for `kimi -p` on the experimental engine, with the same exit / drain / steer semantics and defaults as the default engine, and `kimi -p "/goal ..."` now stays alive until the goal reaches a terminal state instead of exiting after the first turn. diff --git a/.changeset/align-subagent-timeout.md b/.changeset/align-subagent-timeout.md new file mode 100644 index 0000000000..41920aa5a3 --- /dev/null +++ b/.changeset/align-subagent-timeout.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Align the subagent timeout across engines: a fixed 2-hour default, overridable with `[subagent] timeout_ms` in config.toml or the KIMI_SUBAGENT_TIMEOUT_MS environment variable. diff --git a/.changeset/fix-session-format-backward-compat.md b/.changeset/fix-session-format-backward-compat.md new file mode 100644 index 0000000000..274ebd21f9 --- /dev/null +++ b/.changeset/fix-session-format-backward-compat.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Fix sessions created by newer builds failing to open in older CLI builds on the same machine; new sessions are written in a compatible layout, and existing sessions are healed on first open. diff --git a/.changeset/server-version-from-cli.md b/.changeset/server-version-from-cli.md new file mode 100644 index 0000000000..17e7f6ff50 --- /dev/null +++ b/.changeset/server-version-from-cli.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Fix `kimi server` reporting the internal server package version instead of the CLI version in its metadata; the web UI settings now show the CLI version. diff --git a/.changeset/web-message-history-real-times.md b/.changeset/web-message-history-real-times.md new file mode 100644 index 0000000000..5c7b78cb67 --- /dev/null +++ b/.changeset/web-message-history-real-times.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +web: Show each message's actual send time in chat history after reloading a session, instead of the session creation time. diff --git a/apps/kimi-code/src/cli/sub/server/run.ts b/apps/kimi-code/src/cli/sub/server/run.ts index 6a903123dd..b7a9ffb033 100644 --- a/apps/kimi-code/src/cli/sub/server/run.ts +++ b/apps/kimi-code/src/cli/sub/server/run.ts @@ -401,6 +401,9 @@ async function runServerInProcess( const v2 = await startServer({ host: options.host, port: options.port, + // Report the CLI's product version as `server_version` (/meta, web UI) + // rather than kap-server's private package version. + version, logLevel: options.logLevel, logger, debugEndpoints: options.debugEndpoints, diff --git a/apps/kimi-code/src/cli/v2/run-v2-print.ts b/apps/kimi-code/src/cli/v2/run-v2-print.ts index dd8e6c7a9e..ffdbe0668d 100644 --- a/apps/kimi-code/src/cli/v2/run-v2-print.ts +++ b/apps/kimi-code/src/cli/v2/run-v2-print.ts @@ -10,7 +10,8 @@ * native `DomainEvent` stream (payloads are already v1-protocol-shaped), * - drives a turn through `IAgentPromptService.enqueue()` and awaits * `Turn.result` for authoritative completion, - * - drains background tasks (config-driven) before exiting. + * - applies the print-mode background policy (config-driven, v1-aligned: + * `exit` / `drain` / `steer`) before exiting. * * Selected by `runPrompt` when `KIMI_CODE_EXPERIMENTAL_FLAG` is set. */ @@ -34,13 +35,16 @@ import { ensureMainAgent, hostRequestHeadersSeed, logSeed, + resolveAgentTaskConfig, resolveKimiHome, resolveLoggingConfig, + resolvePrintBackgroundMode, skillCatalogRuntimeOptionsSeed, type DomainEvent, type IAgentScopeHandle, type ISessionScopeHandle, type LoopRunResult, + type PrintBackgroundMode, type Scope, } from '@moonshot-ai/agent-core-v2'; import { createKimiDefaultHeaders, createKimiDeviceId } from '@moonshot-ai/kimi-code-oauth'; @@ -81,12 +85,7 @@ import { const PROMPT_UI_MODE = 'print'; const DEFAULT_PRINT_WAIT_CEILING_S = 3600; -const TASK_CONFIG_SECTION = 'task'; -const LEGACY_BACKGROUND_CONFIG_SECTION = 'background'; - -interface TaskPrintWaitConfig { - readonly printWaitCeilingS?: number; -} +const DEFAULT_PRINT_MAX_TURNS = 50; export async function runV2Print( opts: CLIOptions, @@ -336,8 +335,13 @@ async function runNativeTurn( await agent.accessor.get(IAuthSummaryService).ensureReady(); + const turnEndings = createPrintTurnEndings(); const subscription = agent.accessor.get(IEventBus).subscribe((event: DomainEvent) => { dispatchNativeEvent(writer, event, stderr); + // Arm the turn-endings collector before `turn.result` settles so a + // background-task completion that steers a new turn right after the main + // turn ends cannot have its `turn.ended` slip past the policy loop. + if (event.type === 'turn.ended') turnEndings.push(event); }); try { const handle = await agent.accessor.get(IAgentPromptService).enqueue({ @@ -361,16 +365,41 @@ async function runNativeTurn( } const result = await turn.result; - // Turn settled, but `-p` is not done until any background work the turn - // spawned has drained (config-bounded). Flush the buffered assistant - // message first so a long drain does not withhold the final message. + // Turn settled, but `-p` is not done until the print-mode background + // policy says so (config-driven: exit / drain / steer). Flush the buffered + // assistant message first so a long drain/steer wait does not withhold the + // final message. writer.flushAssistant(); if (result.type === 'completed') { + const configService = app.accessor.get(IConfigService); + const taskConfig = resolveAgentTaskConfig(configService); + const goalService = agent.accessor.get(IAgentGoalService); try { - await drainBackgroundTasks(app, session); - } catch { - // Draining is best-effort; a wedged background task must not fail the - // (already completed) turn. Swallow and proceed to finish. + await applyPrintBackgroundPolicy({ + mode: resolvePrintBackgroundMode(configService), + ceilingS: taskConfig?.printWaitCeilingS ?? DEFAULT_PRINT_WAIT_CEILING_S, + maxTurns: taskConfig?.printMaxTurns ?? DEFAULT_PRINT_MAX_TURNS, + countPending: () => countPendingBackgroundTasks(session), + drain: () => drainBackgroundTasks(session, taskConfig?.printWaitCeilingS), + turnEndings, + skipTurnId: turn.id, + warn: (message) => stderr.write(`Warning: ${message}\n`), + now: () => Date.now(), + goalActive: () => goalService.getGoal().goal?.status === 'active', + }); + } catch (error) { + // A steered turn that fails fails the run (v1 parity). Anything else + // is best-effort: a wedged background task must not fail the (already + // completed) main turn. + if (error instanceof PrintSteeredTurnFailedError) { + writer.finish(); + throw error; + } + stderr.write( + `Warning: print background policy failed: ${ + error instanceof Error ? error.message : String(error) + }\n`, + ); } writer.finish(); return; @@ -466,12 +495,182 @@ function dispatchNativeEvent( } } -async function drainBackgroundTasks(app: Scope, session: ISessionScopeHandle): Promise { - const config = app.accessor.get(IConfigService); - const section = - config.get(TASK_CONFIG_SECTION) ?? - config.get(LEGACY_BACKGROUND_CONFIG_SECTION); - const ceilingS = section?.printWaitCeilingS; +export type PrintTurnEnding = Extract; + +/** + * Source of `turn.ended` events for the print steer loop. `next` resolves with + * the next ending (skipping `skipTurnId`, the main turn's own buffered + * ending), or `null` when `remainingMs` elapses first. + */ +export interface PrintTurnEndings { + next(remainingMs: number, skipTurnId: number): Promise; +} + +/** + * Buffered `turn.ended` collector fed from the agent event bus. Events that + * arrive while no one is waiting are queued, so endings that fire between the + * main turn settling and the policy loop starting are not missed. + */ +export function createPrintTurnEndings(): PrintTurnEndings & { + push: (event: PrintTurnEnding) => void; +} { + const buffer: PrintTurnEnding[] = []; + let waiter: ((ending: PrintTurnEnding | null) => void) | undefined; + return { + push: (event) => { + const resolve = waiter; + if (resolve !== undefined) { + waiter = undefined; + resolve(event); + return; + } + buffer.push(event); + }, + next: async (remainingMs, skipTurnId) => { + const deadlineAt = Date.now() + remainingMs; + const waitOnce = (ms: number): Promise => + new Promise((resolve) => { + let settled = false; + const settle = (value: PrintTurnEnding | null): void => { + if (settled) return; + settled = true; + clearTimeout(timer); + waiter = undefined; + // oxlint-disable-next-line promise/no-multiple-resolved -- `settled` guards the single resolve; the rule cannot see it + resolve(value); + }; + const timer = Number.isFinite(ms) + ? setTimeout(() => { + settle(null); + }, ms) + : undefined; + waiter = settle; + }); + for (;;) { + while (buffer.length > 0) { + const ending = buffer.shift()!; + if (ending.turnId !== skipTurnId) return ending; + } + const ms = deadlineAt - Date.now(); + if (ms <= 0) return null; + const ending = await waitOnce(ms); + if (ending === null) return null; + if (ending.turnId !== skipTurnId) return ending; + // The skipped turn's own ending: keep waiting within the same budget. + } + }, + }; +} + +/** A background-task completion steered a new main turn that did not complete. */ +export class PrintSteeredTurnFailedError extends Error {} + +export interface PrintBackgroundPolicyInput { + readonly mode: PrintBackgroundMode; + readonly ceilingS: number; + readonly maxTurns: number; + readonly countPending: () => number; + readonly drain: () => Promise; + readonly turnEndings: PrintTurnEndings; + readonly skipTurnId: number; + readonly warn: (message: string) => void; + readonly now: () => number; + /** + * Reports whether an agent goal is still `active`. v2 drives goal + * continuation as new turns (v1 keeps a single turn alive), so a `-p` goal + * run must stay alive until the goal leaves `active`, independent of the + * background policy. + */ + readonly goalActive?: () => boolean; +} + +/** + * Apply the print-mode (`kimi -p`) background-task policy after the main turn + * completes. Mirrors v1's `Session.handlePrintMainTurnCompleted`: + * - goal : while a goal is `active`, keep waiting for its continuation + * turns (bounded by `ceilingS` as a safety net), regardless of + * the background mode; the goal summary drives the exit code. + * - 'exit' : return immediately (default). + * - 'drain' : suppress + drain background tasks, then return. + * - 'steer' : while background tasks are still pending, stay alive so task + * completions steer new main turns; return once quiescent, or + * when the wall-clock ceiling (`ceilingS`) or the turn cap + * (`maxTurns`) is reached. A steered turn that does not complete + * fails the run. + */ +export async function applyPrintBackgroundPolicy( + input: PrintBackgroundPolicyInput, +): Promise { + if (input.goalActive !== undefined) { + const goalDeadline = input.now() + input.ceilingS * 1000; + while (input.goalActive()) { + const ended = await input.turnEndings.next( + goalDeadline - input.now(), + input.skipTurnId, + ); + if (ended === null) { + input.warn(`print goal wait ceiling reached (${input.ceilingS}s), finishing`); + return; + } + // A continuation turn that does not complete pauses/blocks the goal, so + // the loop condition exits on the next check. + } + } + if (input.mode === 'exit') return; + if (input.mode === 'drain') { + await input.drain(); + return; + } + + // 'steer' + const deadline = input.now() + input.ceilingS * 1000; + let turns = 0; + for (;;) { + turns += 1; + if (input.now() >= deadline) { + input.warn(`print steer ceiling reached (${input.ceilingS}s), finishing`); + return; + } + if (turns > input.maxTurns) { + input.warn(`print steer max turns reached (${input.maxTurns}), finishing`); + return; + } + if (input.countPending() === 0) return; + const ended = await input.turnEndings.next(deadline - input.now(), input.skipTurnId); + if (ended === null) { + // The wait itself ran out: no further turn ended before the ceiling. + input.warn(`print steer ceiling reached (${input.ceilingS}s), finishing`); + return; + } + if (ended.reason !== 'completed') { + throw new PrintSteeredTurnFailedError(formatTurnEndingFailure(ended)); + } + } +} + +function formatTurnEndingFailure(ending: PrintTurnEnding): string { + if (ending.error?.code === 'provider.filtered') { + return 'Provider safety policy blocked the response.'; + } + if (ending.error !== undefined) return `${ending.error.code}: ${ending.error.message}`; + if (ending.reason === 'blocked') { + return 'Prompt hook blocked the request.'; + } + return `Prompt turn ended with reason: ${ending.reason}`; +} + +function countPendingBackgroundTasks(session: ISessionScopeHandle): number { + let count = 0; + for (const handle of session.accessor.get(IAgentLifecycleService).list()) { + count += handle.accessor.get(IAgentTaskService).list(true).length; + } + return count; +} + +async function drainBackgroundTasks( + session: ISessionScopeHandle, + ceilingS: number | undefined, +): Promise { const ceilingMs = typeof ceilingS === 'number' && Number.isFinite(ceilingS) && ceilingS > 0 ? ceilingS * 1000 diff --git a/apps/kimi-code/test/cli/run-v2-print.test.ts b/apps/kimi-code/test/cli/run-v2-print.test.ts new file mode 100644 index 0000000000..f2535ec02c --- /dev/null +++ b/apps/kimi-code/test/cli/run-v2-print.test.ts @@ -0,0 +1,258 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { + applyPrintBackgroundPolicy, + createPrintTurnEndings, + PrintSteeredTurnFailedError, + type PrintTurnEnding, + type PrintTurnEndings, +} from '#/cli/v2/run-v2-print'; + +function ending( + turnId: number, + reason: PrintTurnEnding['reason'] = 'completed', +): PrintTurnEnding { + return { type: 'turn.ended', turnId, reason }; +} + +interface ScriptedEntry { + readonly event: PrintTurnEnding; + /** Side effect applied when this entry is consumed (e.g. mutate pending). */ + readonly apply?: () => void; +} + +/** + * Scripted `PrintTurnEndings`: replays queued endings (honouring `skipTurnId`), + * then resolves `null` once the script is exhausted (the wait "timed out"). + */ +function scriptedTurnEndings(entries: ScriptedEntry[]): PrintTurnEndings { + const queue = [...entries]; + return { + next: async (_remainingMs: number, skipTurnId: number) => { + while (queue.length > 0) { + const entry = queue.shift()!; + if (entry.event.turnId === skipTurnId) continue; + entry.apply?.(); + return entry.event; + } + return null; + }, + }; +} + +describe('applyPrintBackgroundPolicy', () => { + it('exit returns immediately without draining or waiting', async () => { + const drain = vi.fn(async () => {}); + const countPending = vi.fn(() => 1); + await applyPrintBackgroundPolicy({ + mode: 'exit', + ceilingS: 60, + maxTurns: 50, + countPending, + drain, + turnEndings: scriptedTurnEndings([]), + skipTurnId: 1, + warn: () => {}, + now: () => Date.now(), + }); + expect(drain).not.toHaveBeenCalled(); + expect(countPending).not.toHaveBeenCalled(); + }); + + it('drain drains once and returns', async () => { + const drain = vi.fn(async () => {}); + await applyPrintBackgroundPolicy({ + mode: 'drain', + ceilingS: 60, + maxTurns: 50, + countPending: () => 1, + drain, + turnEndings: scriptedTurnEndings([]), + skipTurnId: 1, + warn: () => {}, + now: () => Date.now(), + }); + expect(drain).toHaveBeenCalledTimes(1); + }); + + it('steer returns once background tasks are quiescent', async () => { + let pending = 1; + const warn = vi.fn(); + await applyPrintBackgroundPolicy({ + mode: 'steer', + ceilingS: 60, + maxTurns: 50, + countPending: () => pending, + drain: async () => {}, + turnEndings: scriptedTurnEndings([ + // The main turn's own buffered ending is skipped. + { event: ending(1) }, + // A background task completed and steered a new turn; it finished and + // no tasks remain. + { event: ending(2), apply: () => { pending = 0; } }, + ]), + skipTurnId: 1, + warn, + now: () => Date.now(), + }); + expect(warn).not.toHaveBeenCalled(); + }); + + it('steer finishes with a warning when max turns is reached', async () => { + const warn = vi.fn(); + await applyPrintBackgroundPolicy({ + mode: 'steer', + ceilingS: 60, + maxTurns: 2, + countPending: () => 1, + drain: async () => {}, + turnEndings: scriptedTurnEndings([{ event: ending(2) }, { event: ending(3) }]), + skipTurnId: 1, + warn, + now: () => Date.now(), + }); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0]?.[0]).toContain('max turns'); + }); + + it('steer finishes with a warning when the ceiling is reached', async () => { + let now = 0; + const warn = vi.fn(); + await applyPrintBackgroundPolicy({ + mode: 'steer', + ceilingS: 10, + maxTurns: 50, + countPending: () => 1, + drain: async () => {}, + turnEndings: scriptedTurnEndings([ + { event: ending(2), apply: () => { now = 10_001; } }, + ]), + skipTurnId: 1, + warn, + now: () => now, + }); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0]?.[0]).toContain('ceiling'); + }); + + it('steer warns and returns when the wait times out with tasks still pending', async () => { + const warn = vi.fn(); + await applyPrintBackgroundPolicy({ + mode: 'steer', + ceilingS: 60, + maxTurns: 50, + countPending: () => 1, + drain: async () => {}, + // Empty script: no further turn ends before the deadline. + turnEndings: scriptedTurnEndings([]), + skipTurnId: 1, + warn, + now: () => Date.now(), + }); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0]?.[0]).toContain('ceiling'); + }); + + it('steer throws when a steered turn does not complete', async () => { + await expect( + applyPrintBackgroundPolicy({ + mode: 'steer', + ceilingS: 60, + maxTurns: 50, + countPending: () => 1, + drain: async () => {}, + turnEndings: scriptedTurnEndings([ + { + event: { + type: 'turn.ended', + turnId: 2, + reason: 'failed', + error: { code: 'provider.overloaded', message: 'try later' }, + } as PrintTurnEnding, + }, + ]), + skipTurnId: 1, + warn: () => {}, + now: () => Date.now(), + }), + ).rejects.toThrow(PrintSteeredTurnFailedError); + }); + + it('waits for goal continuation turns before applying the mode', async () => { + let active = true; + let consumed = 0; + const drain = vi.fn(async () => {}); + await applyPrintBackgroundPolicy({ + mode: 'drain', + ceilingS: 60, + maxTurns: 50, + countPending: () => 0, + drain, + turnEndings: scriptedTurnEndings([ + { event: ending(2), apply: () => { consumed += 1; } }, + { + event: ending(3), + apply: () => { + consumed += 1; + active = false; + }, + }, + ]), + skipTurnId: 1, + warn: () => {}, + now: () => Date.now(), + goalActive: () => active, + }); + // Both continuation turns ended before the mode ('drain') ran. + expect(consumed).toBe(2); + expect(drain).toHaveBeenCalledTimes(1); + }); + + it('warns and returns when the goal wait hits the ceiling', async () => { + const warn = vi.fn(); + await applyPrintBackgroundPolicy({ + mode: 'exit', + ceilingS: 10, + maxTurns: 50, + countPending: () => 0, + drain: async () => {}, + // Empty script: no continuation turn ever ends. + turnEndings: scriptedTurnEndings([]), + skipTurnId: 1, + warn, + now: () => Date.now(), + goalActive: () => true, + }); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0]?.[0]).toContain('goal wait ceiling'); + }); +}); + +describe('createPrintTurnEndings', () => { + it('buffers events pushed before next() and skips the given turn id', async () => { + const endings = createPrintTurnEndings(); + endings.push(ending(1)); + endings.push(ending(2)); + await expect(endings.next(1000, 1)).resolves.toMatchObject({ turnId: 2 }); + }); + + it('delivers a pushed event to a pending next()', async () => { + const endings = createPrintTurnEndings(); + const pending = endings.next(1000, 1); + endings.push(ending(3)); + await expect(pending).resolves.toMatchObject({ turnId: 3 }); + }); + + it('resolves null when the remaining time elapses', async () => { + const endings = createPrintTurnEndings(); + await expect(endings.next(5, 1)).resolves.toBeNull(); + }); + + it('keeps waiting when only the skipped turn ends', async () => { + const endings = createPrintTurnEndings(); + const pending = endings.next(1000, 1); + endings.push(ending(1)); + endings.push(ending(4)); + await expect(pending).resolves.toMatchObject({ turnId: 4 }); + }); +}); diff --git a/packages/agent-core-v2/src/agent/swarm/tools/agent-swarm.ts b/packages/agent-core-v2/src/agent/swarm/tools/agent-swarm.ts index c1475c06d5..09db93c303 100644 --- a/packages/agent-core-v2/src/agent/swarm/tools/agent-swarm.ts +++ b/packages/agent-core-v2/src/agent/swarm/tools/agent-swarm.ts @@ -19,13 +19,14 @@ import { } from '#/tool/toolContract'; import { registerTool } from '#/agent/toolRegistry/toolContribution'; import { toInputJsonSchema } from '#/tool/input-schema'; +import { IConfigService } from '#/app/config/config'; import { ISessionSwarmService, type SessionSwarmTask } from '#/session/swarm/sessionSwarm'; import { IAgentScopeContext } from '#/agent/scopeContext/scopeContext'; import { IAgentSwarmService } from '#/agent/swarm/swarm'; +import { resolveSubagentTimeoutMs } from '#/session/subagent/configSection'; import AGENT_SWARM_DESCRIPTION from './agent-swarm.md?raw'; const DEFAULT_SUBAGENT_TYPE = 'coder'; -const DEFAULT_SUBAGENT_TIMEOUT_MS = 30 * 60 * 1000; const PROMPT_TEMPLATE_PLACEHOLDER = '{{item}}'; const MAX_AGENT_SWARM_SUBAGENTS = 128; @@ -107,6 +108,7 @@ export class AgentSwarmTool implements BuiltinTool { @ISessionSwarmService private readonly swarmService: ISessionSwarmService, @IAgentScopeContext scopeContext: IAgentScopeContext, @IAgentSwarmService private readonly swarmMode: IAgentSwarmService, + @IConfigService private readonly config: IConfigService, ) { this.callerAgentId = scopeContext.agentId; } @@ -150,6 +152,7 @@ export class AgentSwarmTool implements BuiltinTool { toolCallId: string, ): Promise { const profileName = normalizeOptionalString(args.subagent_type) ?? DEFAULT_SUBAGENT_TYPE; + const timeoutMs = resolveSubagentTimeoutMs(this.config); const specs = await createAgentSwarmSpecs(args, (agentId) => this.swarmService.getSwarmItem({ callerAgentId: this.callerAgentId, agentId }), ); @@ -165,7 +168,7 @@ export class AgentSwarmTool implements BuiltinTool { runInBackground: false, swarmItem: spec.item, signal, - timeout: DEFAULT_SUBAGENT_TIMEOUT_MS, + timeout: timeoutMs, }; if (spec.kind === 'resume') { return { diff --git a/packages/agent-core-v2/src/agent/task/configSection.ts b/packages/agent-core-v2/src/agent/task/configSection.ts index c8f7d5e312..405f9c0f4b 100644 --- a/packages/agent-core-v2/src/agent/task/configSection.ts +++ b/packages/agent-core-v2/src/agent/task/configSection.ts @@ -7,7 +7,10 @@ * fields as the base and let `[task]` override matching fields. * `keepAliveOnExit` also * accepts the v1 env override `KIMI_CODE_BACKGROUND_KEEP_ALIVE_ON_EXIT` - * (applied live by the config env overlay, never persisted). Self-registered + * (applied live by the config env overlay, never persisted). Also owns the + * `kimi -p` print-mode background policy (`printBackgroundMode` / + * `printWaitCeilingS` / `printMaxTurns`), resolved with v1 semantics by + * `resolvePrintBackgroundMode`. Self-registered * at module load via `registerConfigSection`, so the `config` domain never * imports this domain's types. */ @@ -21,12 +24,18 @@ import { registerConfigSection } from '#/app/config/configSectionContributions'; export const TASK_SECTION = 'task'; export const LEGACY_BACKGROUND_SECTION = 'background'; +export const PrintBackgroundModeSchema = z.enum(['exit', 'drain', 'steer']); + +export type PrintBackgroundMode = z.infer; + export const AgentTaskConfigSchema = z.object({ maxRunningTasks: z.number().int().min(1).optional(), keepAliveOnExit: z.boolean().optional(), bashAutoBackgroundOnTimeout: z.boolean().optional(), killGracePeriodMs: z.number().int().min(0).optional(), printWaitCeilingS: z.number().int().min(1).optional(), + printBackgroundMode: PrintBackgroundModeSchema.optional(), + printMaxTurns: z.number().int().min(1).optional(), }); export type AgentTaskConfig = z.infer; @@ -39,6 +48,20 @@ export function resolveAgentTaskConfig(config: IConfigService): AgentTaskConfig return { ...legacy, ...current }; } +/** + * Resolve the effective print-mode (`kimi -p`) background-task policy, mirroring + * v1's `Session.resolvePrintBackgroundMode`: `printBackgroundMode` is + * authoritative when set; otherwise fall back to the legacy `keepAliveOnExit` + * mapping (`true` ⇒ `'drain'`, otherwise `'exit'`). The + * `KIMI_CODE_BACKGROUND_KEEP_ALIVE_ON_EXIT` env override is applied by the + * config env overlay (see `taskEnvBindings`), so it is covered here. + */ +export function resolvePrintBackgroundMode(config: IConfigService): PrintBackgroundMode { + const section = resolveAgentTaskConfig(config); + if (section?.printBackgroundMode !== undefined) return section.printBackgroundMode; + return section?.keepAliveOnExit === true ? 'drain' : 'exit'; +} + export const KEEP_ALIVE_ON_EXIT_ENV = 'KIMI_CODE_BACKGROUND_KEEP_ALIVE_ON_EXIT'; export const taskEnvBindings: EnvBindings = envBindings(AgentTaskConfigSchema, { diff --git a/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts b/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts index d10522584b..e7b9b531ff 100644 --- a/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts +++ b/packages/agent-core-v2/src/agent/wireRecord/wireRecord.ts @@ -1,3 +1,15 @@ +/** + * `wireRecord` contract (L6) — the persisted wire journal's public surface. + * + * Defines the on-disk record vocabulary (the `metadata` envelope and the + * migration records) and `IAgentWireRecordService`. `seal` starts a fresh log + * with the `metadata` envelope at agent creation (a no-op once any record + * exists) so released v1 builds — whose replay hard-rejects a non-empty log + * lacking the envelope — can read sessions on a shared `KIMI_CODE_HOME`; + * legacy envelope-less logs are healed by `restore`, never by `seal`. Bound + * at Agent scope. + */ + import { createDecorator } from '#/_base/di/instantiation'; import type { WireMigrationRecord } from '#/agent/wireRecord/migration/migration'; @@ -24,6 +36,7 @@ export interface WireRecordRestoreResult { export interface IAgentWireRecordService { readonly _serviceBrand: undefined; + seal(): Promise; getRecords(): readonly PersistedWireRecord[]; restore( records?: readonly PersistedWireRecord[], diff --git a/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts b/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts index 7eb92eb52d..12d8ba8377 100644 --- a/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts +++ b/packages/agent-core-v2/src/agent/wireRecord/wireRecordService.ts @@ -3,9 +3,13 @@ * * Restores and retains the owning agent's wire journal, applies protocol * migrations, rejects non-empty unversioned logs, and awaits durable atomic - * rewrites before restore completes. Tracks live records through `wire`, uses - * `agent/scopeContext` for storage addressing, and persists through the - * `appendLog` access-pattern store. Bound at Agent scope. + * rewrites before restore completes. Seals fresh logs with the `metadata` + * envelope at creation (`seal`) so released v1 builds — whose replay + * hard-rejects envelope-less logs — can read sessions on a shared + * `KIMI_CODE_HOME`; legacy envelope-less logs are healed on `restore`. + * Tracks live records through `wire`, uses `agent/scopeContext` for storage + * addressing, and persists through the `appendLog` access-pattern store. + * Bound at Agent scope. */ import { relative } from 'pathe'; @@ -13,6 +17,7 @@ import { relative } from 'pathe'; import { InstantiationType } from '#/_base/di/extensions'; import { Disposable } from '#/_base/di/lifecycle'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { onUnexpectedError } from '#/_base/errors/unexpectedError'; import { IAgentScopeContext } from '#/agent/scopeContext/scopeContext'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IAgentWireService } from '#/wire/tokens'; @@ -64,6 +69,14 @@ export class AgentWireRecordService extends Disposable implements IAgentWireReco return [...this.records]; } + async seal(): Promise { + if (this.log === undefined) return; + if (await hasAnyRecord(this.log, this.wireScope, WIRE_RECORD_FILENAME)) return; + this.log.append(this.wireScope, WIRE_RECORD_FILENAME, metadataRecord(), { + onError: onUnexpectedError, + }); + } + async restore( records?: readonly PersistedWireRecord[], options: WireRecordRestoreOptions = {}, @@ -161,6 +174,14 @@ function isWireRecordMetadata(record: PersistedWireRecord): record is WireRecord return record.type === 'metadata' && typeof record['protocol_version'] === 'string'; } +async function hasAnyRecord(log: IAppendLogStore, scope: string, key: string): Promise { + for await (const record of log.read(scope, key)) { + void record; + return true; + } + return false; +} + export const WIRE_RECORD_FILENAME = 'wire.jsonl'; export function missingWireMetadataError(): Error { diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index d928d3e933..f7bdfd0c14 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -204,6 +204,12 @@ export * from '#/agent/toolSelect/toolSelectAnnouncements'; export * from '#/agent/toolSelect/toolSelectAnnouncementsService'; import '#/agent/task/configSection'; +export { + resolveAgentTaskConfig, + resolvePrintBackgroundMode, + type AgentTaskConfig, + type PrintBackgroundMode, +} from '#/agent/task/configSection'; import '#/agent/task/tools/task-list'; import '#/agent/task/tools/task-output'; import '#/agent/task/tools/task-stop'; @@ -233,6 +239,7 @@ export * from '#/session/subagent/subagentService'; export * from '#/session/subagent/tools/subagent-task'; export { AGENT_RUN_PROMPT_ORIGIN } from '#/session/subagent/runAgentTurn'; export * from '#/session/subagent/mirrorAgentRun'; +import '#/session/subagent/configSection'; import '#/session/subagent/tools/agent'; export * from '#/app/sessionLifecycle/sessionLifecycle'; export * from '#/app/sessionLifecycle/sessionLifecycleService'; diff --git a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts index 65485aa4c4..dc4917e6c6 100644 --- a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts +++ b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts @@ -55,6 +55,7 @@ import { IImageConfigBridge } from '#/agent/media/imageConfigBridge'; import { IAgentMcpService } from '#/agent/mcp/mcp'; import { IAgentExternalHooksService } from '#/agent/externalHooks/externalHooks'; import { IAgentPluginService } from '#/agent/plugin/agentPlugin'; +import { IAgentWireRecordService } from '#/agent/wireRecord/wireRecord'; import { ISessionInteractionService } from '#/session/interaction/interaction'; import { type AgentListFilter, @@ -168,6 +169,7 @@ export class AgentLifecycleService extends Disposable implements IAgentLifecycle ) as IAgentScopeHandle; this.handles.set(agentId, handle); try { + await handle.accessor.get(IAgentWireRecordService).seal(); await this.sessionMetadata.registerAgent(agentId, { homedir: agentHomedir, type: agentId === 'main' ? 'main' : 'sub', @@ -182,10 +184,7 @@ export class AgentLifecycleService extends Disposable implements IAgentLifecycle // Bootstrap (profile binding and the force-instantiated observer // services) is complete: drive the activity kernel `initializing → idle` // so the agent can admit turns. Until this point `begin` rejects with - // `activity.initializing`. The wire log's metadata envelope is NOT - // seeded here — `wireRecord.restore()` heals envelope-less logs on - // resume (prepend + rewrite), so creation stays free of log-format - // concerns. + // `activity.initializing`. handle.accessor.get(IAgentActivityService).markReady(); return handle; } catch (error) { diff --git a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts index e05901c177..c3bd1a2516 100644 --- a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts +++ b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts @@ -4,8 +4,13 @@ * Persists the session metadata document (`state.json`) through the `storage` * access-pattern store (`IAtomicDocumentStore`), rooted at the `metaScope` * namespace from `sessionContext`. Loads the existing document on - * construction (creating it on first run), and logs through `log`. Bound at - * Session scope. + * construction (creating it on first run), and logs through `log`. The + * document always carries the `agents` / `custom` maps that v1's + * `Session.resume()` reads unconditionally — seeded at creation, backfilled + * and persisted on load for documents written before the seeding existed + * (without touching `updatedAt`, so a format heal never reorders session + * listings) — keeping sessions on a shared `KIMI_CODE_HOME` resumable by + * released v1 builds. Bound at Session scope. * * Read-model mirroring (flag `persistence_minidb_readmodel`): after a metadata * update is persisted, the fresh summary is mirrored into the `IQueryStore` @@ -131,6 +136,14 @@ export class SessionMetadata extends Disposable implements ISessionMetadata { const existing = await this.store.get(this.scope, META_KEY); if (existing !== undefined) { this.data = normalizeSessionMeta(existing, this.ctx.sessionId); + if (this.data.agents === undefined || this.data.custom === undefined) { + this.data = { + ...this.data, + agents: this.data.agents ?? {}, + custom: this.data.custom ?? {}, + }; + await this.store.set(this.scope, META_KEY, this.data); + } return; } const now = Date.now(); @@ -141,6 +154,8 @@ export class SessionMetadata extends Disposable implements ISessionMetadata { createdAt: now, updatedAt: now, archived: false, + agents: {}, + custom: {}, }; await this.store.set(this.scope, META_KEY, this.data); this.log.debug('session metadata created', { sessionId: this.ctx.sessionId }); diff --git a/packages/agent-core-v2/src/session/subagent/configSection.ts b/packages/agent-core-v2/src/session/subagent/configSection.ts new file mode 100644 index 0000000000..57aa44d287 --- /dev/null +++ b/packages/agent-core-v2/src/session/subagent/configSection.ts @@ -0,0 +1,79 @@ +/** + * `subagent` domain (L6) — subagent config-section schema, env binding, and + * timeout resolution. + * + * Owns the `[subagent]` configuration section (`timeout_ms` on disk) together + * with the `KIMI_SUBAGENT_TIMEOUT_MS` env override, mirroring v1's + * `resolveSubagentTimeoutMs` precedence (env > config.toml > 2h default). Both + * collaboration tools — `Agent` in this domain and `AgentSwarm` in the `swarm` + * domain — resolve their per-run timeout through `resolveSubagentTimeoutMs`, + * and render the timeout message with `formatSubagentTimeoutDescription`. + * Self-registered at module load via `registerConfigSection`, so the `config` + * domain never imports this domain's types. + */ + +import { z } from 'zod'; + +import { type EnvBindings, envBindings, type IConfigService } from '#/app/config/config'; +import { registerConfigSection } from '#/app/config/configSectionContributions'; + +export const SUBAGENT_SECTION = 'subagent'; + +export const SubagentConfigSchema = z.object({ + /** Per-run subagent timeout in milliseconds; set a large value to effectively disable the cap. */ + timeoutMs: z.number().int().min(1).optional(), +}); + +export type SubagentConfig = z.infer; + +/** Default per-run subagent timeout: 2 hours, same as v1. */ +export const DEFAULT_SUBAGENT_TIMEOUT_MS = 2 * 60 * 60 * 1000; + +export const SUBAGENT_TIMEOUT_ENV = 'KIMI_SUBAGENT_TIMEOUT_MS'; + +/** Parse the env override; anything but a positive integer is ignored (v1 semantics). */ +function parseTimeoutMsEnv(raw: string): number | undefined { + const parsed = Number(raw); + return Number.isInteger(parsed) && parsed >= 1 ? parsed : undefined; +} + +export const subagentEnvBindings: EnvBindings = envBindings( + SubagentConfigSchema, + { + timeoutMs: { env: SUBAGENT_TIMEOUT_ENV, parse: parseTimeoutMsEnv }, + }, +); + +registerConfigSection(SUBAGENT_SECTION, SubagentConfigSchema, { + defaultValue: { timeoutMs: DEFAULT_SUBAGENT_TIMEOUT_MS }, + env: subagentEnvBindings, +}); + +/** + * Resolve the effective per-run subagent timeout. Governs foreground and + * background subagents (and AgentSwarm) through the task manager's per-task + * timeout. + */ +export function resolveSubagentTimeoutMs(config: IConfigService): number { + return ( + config.get(SUBAGENT_SECTION)?.timeoutMs ?? + DEFAULT_SUBAGENT_TIMEOUT_MS + ); +} + +/** Human-readable duration for the subagent timeout message. */ +export function formatSubagentTimeoutDescription(ms: number): string { + if (ms % (60 * 60 * 1000) === 0) { + const h = ms / (60 * 60 * 1000); + return `${h} hour${h === 1 ? '' : 's'}`; + } + if (ms % (60 * 1000) === 0) { + const m = ms / (60 * 1000); + return `${m} minute${m === 1 ? '' : 's'}`; + } + if (ms % 1000 === 0) { + const s = ms / 1000; + return `${s} second${s === 1 ? '' : 's'}`; + } + return `${ms} ms`; +} diff --git a/packages/agent-core-v2/src/session/subagent/tools/agent.md b/packages/agent-core-v2/src/session/subagent/tools/agent.md index ec0533e7ef..d8b65d7c0d 100644 --- a/packages/agent-core-v2/src/session/subagent/tools/agent.md +++ b/packages/agent-core-v2/src/session/subagent/tools/agent.md @@ -9,7 +9,7 @@ Writing the prompt: Usage notes: - When the task continues earlier work a subagent already did, prefer resuming that agent (pass its `resume` id) over spawning a fresh instance — the resumed agent keeps its prior context. - A subagent's result is only visible to you, not to the user. When the user needs to see what a subagent produced, summarize the relevant parts yourself in your own reply. -- Subagents use a fixed 30-minute timeout. If one times out, resume the same agent instead of starting over. +- Subagents use a fixed 2-hour timeout. If one times out, resume the same agent instead of starting over. When NOT to use Agent: skip delegation for trivial work you can do directly — reading a file whose path you already know, searching a small known set of files, or any task that takes only a step or two. Delegation has a context-handoff cost; it pays off only when the task is substantial enough to outweigh it. diff --git a/packages/agent-core-v2/src/session/subagent/tools/agent.ts b/packages/agent-core-v2/src/session/subagent/tools/agent.ts index 6193b0a069..00d94b15a8 100644 --- a/packages/agent-core-v2/src/session/subagent/tools/agent.ts +++ b/packages/agent-core-v2/src/session/subagent/tools/agent.ts @@ -41,6 +41,7 @@ import { registerTool } from '#/agent/toolRegistry/toolContribution'; import { IAgentProfileCatalogService, type AgentProfile } from '#/app/agentProfileCatalog/agentProfileCatalog'; import { applyProfilePromptPrefix } from '#/app/agentProfileCatalog/promptPrefix'; import { ILogService } from '#/_base/log/log'; +import { IConfigService } from '#/app/config/config'; import { IAgentLifecycleService } from '#/session/agentLifecycle/agentLifecycle'; import { isSubagentMeta, subagentLabels, subagentParentAgentId } from '#/session/agentLifecycle/subagentMetadata'; import { ISessionProcessRunner } from '#/session/process/processRunner'; @@ -49,6 +50,10 @@ import { ISessionWorkspaceContext } from '#/session/workspaceContext/workspaceCo import { emitAgentRunSpawned, mirrorAgentRun } from '../mirrorAgentRun'; import { ISessionSubagentService } from '../subagent'; +import { + formatSubagentTimeoutDescription, + resolveSubagentTimeoutMs, +} from '../configSection'; import { SubagentTask, type SubagentHandle } from './subagent-task'; import AGENT_BACKGROUND_DISABLED_DESCRIPTION from './agent-background-disabled.md?raw'; @@ -57,8 +62,6 @@ import AGENT_DESCRIPTION_BASE from './agent.md?raw'; const DEFAULT_PROFILE_NAME = 'coder'; const RESUMED_LABEL = 'subagent'; -export const DEFAULT_SUBAGENT_TIMEOUT_MS = 30 * 60 * 1000; -export const DEFAULT_SUBAGENT_TIMEOUT_DESCRIPTION = '30 minutes'; export const AgentToolInputSchema = z.preprocess( (input) => { @@ -146,6 +149,7 @@ export class AgentTool implements BuiltinTool { @ISessionMetadata private readonly sessionMetadata: ISessionMetadata, @ILogService private readonly log: ILogService, @IAgentPermissionModeService private readonly permissionMode: IAgentPermissionModeService, + @IConfigService private readonly config: IConfigService, ) { this.callerAgentId = scopeContext.agentId; this.canRunInBackground = () => @@ -334,6 +338,7 @@ export class AgentTool implements BuiltinTool { if (runInBackground && !allowBackground) { return { output: BACKGROUND_AGENT_UNAVAILABLE, isError: true }; } + const timeoutMs = resolveSubagentTimeoutMs(this.config); const controller = new AbortController(); const abortBeforeRegister = (): void => { @@ -363,7 +368,7 @@ export class AgentTool implements BuiltinTool { try { const registerOptions: RegisterAgentTaskOptions = { detached: runInBackground, - timeoutMs: DEFAULT_SUBAGENT_TIMEOUT_MS, + timeoutMs, signal: runInBackground ? undefined : signal, }; taskId = this.tasks.registerTask( @@ -403,7 +408,7 @@ export class AgentTool implements BuiltinTool { output: formatBackgroundAgentResult(taskId, handle, args.description, allowBackground), }; } - return await this.formatForegroundResult(taskId, handle); + return await this.formatForegroundResult(taskId, handle, timeoutMs); } catch (error) { return { output: `subagent error: ${launchErrorMessage(error, signal)}`, isError: true }; } @@ -412,6 +417,7 @@ export class AgentTool implements BuiltinTool { private async formatForegroundResult( taskId: string, handle: SubagentHandle, + timeoutMs: number, ): Promise { const info = this.tasks.getTask(taskId); if (info?.status === 'completed') { @@ -421,7 +427,7 @@ export class AgentTool implements BuiltinTool { } const timedOut = info?.status === 'timed_out'; const message = timedOut - ? `Agent timed out after ${DEFAULT_SUBAGENT_TIMEOUT_DESCRIPTION}.` + ? `Agent timed out after ${formatSubagentTimeoutDescription(timeoutMs)}.` : info?.stopReason === 'Interrupted by user' ? USER_INTERRUPTED_SUBAGENT_MESSAGE : info?.stopReason !== undefined diff --git a/packages/agent-core-v2/test/agent/contextMemory/stubs.ts b/packages/agent-core-v2/test/agent/contextMemory/stubs.ts index 178b2f24e5..fec907db1b 100644 --- a/packages/agent-core-v2/test/agent/contextMemory/stubs.ts +++ b/packages/agent-core-v2/test/agent/contextMemory/stubs.ts @@ -24,6 +24,7 @@ import { IAgentWireRecordService } from '#/agent/wireRecord/wireRecord'; export function stubWireRecord(): IAgentWireRecordService { return { _serviceBrand: undefined, + seal: () => Promise.resolve(), restore: () => Promise.resolve({}), flush: () => Promise.resolve(), close: () => Promise.resolve(), diff --git a/packages/agent-core-v2/test/agent/loop/loop.test.ts b/packages/agent-core-v2/test/agent/loop/loop.test.ts index 61a3b5ecb3..2f78b4708b 100644 --- a/packages/agent-core-v2/test/agent/loop/loop.test.ts +++ b/packages/agent-core-v2/test/agent/loop/loop.test.ts @@ -96,8 +96,8 @@ describe('Agent loop', () => { [emit] context.spliced { "start": 0, "deleteCount": 0, "messages": [ { "role": "user", "content": [ { "type": "text", "text": "Hello" } ], "toolCalls": [], "origin": { "kind": "user" }, "id": "" } ] } [emit] turn.step.started { "turnId": 0, "step": 1, "stepId": "" } [wire] context.append_loop_event { "event": { "type": "step.begin", "uuid": "", "turnId": "0", "step": 1 }, "time": "