From 22e9b38f70b7d80f17617da90c07daac81a2b0ac Mon Sep 17 00:00:00 2001 From: weishu Date: Sat, 18 Jul 2026 12:17:52 +0800 Subject: [PATCH] fix(kimi): sync local sessions to web and adapt to new kimi-code architecture hapi kimi local mode spawned the kimi TUI with no transcript sync, so terminal conversations never reached the hub and the web UI stayed empty. After the kimi-code rewrite (data moved from ~/.kimi to ~/.kimi-code), model resolution also broke: hapi read the gone ~/.kimi/config.toml and fell back to the invalid hardcoded default kimi-k2, and the KIMI_MODEL / KIMI_PROJECT_DIR env vars it set no longer exist upstream. Local sync (mirrors the codex transcript scanner): - kimiWireLocator: derive the kimi-code workspace id (wd__, ported verbatim from upstream workdir-slug), poll for the session dir created by the just-spawned process, and watch its agents/main/wire.jsonl. Pre-existing sessions are snapshotted and excluded (awaited before spawn) so a retry cannot bind to a stale session; multiple fresh candidates are refused as ambiguous. - kimiWireScanner: incrementally read wire.jsonl and convert events into hapi messages (user prompts/steers, assistant text/thinking, tool call/result incl. is_error, step.end usage with cached input summed into inputTokens). - kimiLocalLauncher: attach locator+scanner, report kimiSessionId on discovery (enables web resume and local<->remote handoff). Model handling: - config.ts: read /config.toml (legacy ~/.kimi fallback); drop the hardcoded kimi-k2 default and the dead KIMI_MODEL env source - when nothing is configured, omit --model so kimi-code uses its own default_model. - kimiBackend/kimiLocal: stop setting KIMI_MODEL and KIMI_PROJECT_DIR (both unused by new kimi-code). - kimiRemoteLauncher: apply the resolved model over ACP after session creation (session/set_model, falling back to the advertised model config option), and display the agent-reported current model instead of the env guess. Verified against live kimi-code 0.26.0: ACP initialize/session-new/ prompt probes, locator discovery of a running session, and converter robustness over a real 800-line wire.jsonl. --- cli/src/kimi/kimiLocal.ts | 3 +- cli/src/kimi/kimiLocalLauncher.ts | 81 ++++- cli/src/kimi/kimiRemoteLauncher.ts | 69 ++++- cli/src/kimi/utils/config.test.ts | 52 ++++ cli/src/kimi/utils/config.ts | 84 +++-- cli/src/kimi/utils/kimiBackend.ts | 22 +- cli/src/kimi/utils/kimiWireLocator.test.ts | 155 ++++++++++ cli/src/kimi/utils/kimiWireLocator.ts | 249 +++++++++++++++ cli/src/kimi/utils/kimiWireScanner.test.ts | 166 ++++++++++ cli/src/kimi/utils/kimiWireScanner.ts | 341 +++++++++++++++++++++ 10 files changed, 1151 insertions(+), 71 deletions(-) create mode 100644 cli/src/kimi/utils/config.test.ts create mode 100644 cli/src/kimi/utils/kimiWireLocator.test.ts create mode 100644 cli/src/kimi/utils/kimiWireLocator.ts create mode 100644 cli/src/kimi/utils/kimiWireScanner.test.ts create mode 100644 cli/src/kimi/utils/kimiWireScanner.ts diff --git a/cli/src/kimi/kimiLocal.ts b/cli/src/kimi/kimiLocal.ts index 1e732bc6..306e23f2 100644 --- a/cli/src/kimi/kimiLocal.ts +++ b/cli/src/kimi/kimiLocal.ts @@ -25,8 +25,7 @@ export async function kimiLocal(opts: { } const env: NodeJS.ProcessEnv = { - ...process.env, - KIMI_PROJECT_DIR: opts.path + ...process.env }; logger.debug(`[KimiLocal] Spawning kimi with args: ${JSON.stringify(args)}`); diff --git a/cli/src/kimi/kimiLocalLauncher.ts b/cli/src/kimi/kimiLocalLauncher.ts index c70ab775..a7dff3b7 100644 --- a/cli/src/kimi/kimiLocalLauncher.ts +++ b/cli/src/kimi/kimiLocalLauncher.ts @@ -1,7 +1,10 @@ +import { logger } from '@/ui/logger'; import { kimiLocal } from './kimiLocal'; import { KimiSession } from './session'; import type { PermissionMode } from './types'; import { BaseLocalLauncher } from '@/modules/common/launcher/BaseLocalLauncher'; +import { createKimiWireLocator, type KimiWireLocator } from './utils/kimiWireLocator'; +import { convertKimiWireEvent, createKimiWireScanner, type KimiWireScanner } from './utils/kimiWireScanner'; function mapApprovalMode(mode: PermissionMode | undefined): { yolo: boolean; plan: boolean } { if (!mode || mode === 'default' || mode === 'read-only') { @@ -19,6 +22,65 @@ export async function kimiLocalLauncher( model?: string; } ): Promise<'switch' | 'exit'> { + // Local mode spawns the kimi TUI directly, so the only way to mirror the + // terminal conversation to hub/web is to watch the wire.jsonl journal the + // kimi-code process writes (same role as the codex transcript scanner). + const startupTimestampMs = Date.now(); + let shuttingDown = false; + let scanner: KimiWireScanner | null = null; + let pendingScannerSetup: Promise | null = null; + + const attachWireScanner = (wirePath: string): Promise => { + const setup = (async () => { + const created = await createKimiWireScanner({ + wirePath, + onEvent: (event) => { + if (shuttingDown) { + return; + } + const converted = convertKimiWireEvent(event); + if (!converted) { + return; + } + if (converted.userMessage) { + session.sendUserMessage(converted.userMessage); + } + if (converted.message) { + session.sendAgentMessage(converted.message); + } + } + }); + if (shuttingDown) { + await created.cleanup(); + return; + } + scanner = created; + logger.debug(`[kimi-local]: Attached wire scanner to ${wirePath}`); + })(); + pendingScannerSetup = setup.catch((error) => { + logger.warn(`[kimi-local]: Wire scanner setup failed for ${wirePath}`, error); + }).finally(() => { + pendingScannerSetup = null; + }); + return pendingScannerSetup; + }; + + const locator: KimiWireLocator = createKimiWireLocator({ + cwd: session.path, + startupTimestampMs, + resumeSessionId: session.sessionId, + onLocated: ({ sessionId, wirePath }) => { + if (shuttingDown) { + return; + } + session.onSessionFound(sessionId); + void attachWireScanner(wirePath); + }, + onAmbiguous: (sessionIds) => { + logger.warn(`[kimi-local]: Multiple fresh kimi sessions found (${sessionIds.join(', ')}); transcript sync disabled for this launch`); + } + }); + const launcher = new BaseLocalLauncher({ label: 'kimi-local', failureLabel: 'Local Kimi process failed', @@ -45,5 +107,22 @@ export async function kimiLocalLauncher( } }); - return await launcher.run(); + try { + // Ensure the pre-existing-session snapshot completed before kimi + // spawns; otherwise the session dir created by this launch could be + // captured in the snapshot and permanently excluded from sync. + await locator.ready; + return await launcher.run(); + } finally { + shuttingDown = true; + await locator.cleanup(); + if (pendingScannerSetup) { + await pendingScannerSetup; + } + const activeScanner = scanner as KimiWireScanner | null; + if (activeScanner) { + await activeScanner.cleanup(); + scanner = null; + } + } } diff --git a/cli/src/kimi/kimiRemoteLauncher.ts b/cli/src/kimi/kimiRemoteLauncher.ts index dc4c79bc..12a39a42 100644 --- a/cli/src/kimi/kimiRemoteLauncher.ts +++ b/cli/src/kimi/kimiRemoteLauncher.ts @@ -53,14 +53,8 @@ class KimiRemoteLauncher extends RemoteLauncherBase { this.happyServer = happyServer; const runtimeConfig = resolveKimiRuntimeConfig({ model: this.model }); - this.displayModel = runtimeConfig.model; - messageBuffer.addMessage(`[MODEL:${runtimeConfig.model}]`, 'system'); - const backend = createKimiBackend({ - model: runtimeConfig.model, - cwd: session.path, - permissionMode: session.getPermissionMode() as string | undefined - }); + const backend = createKimiBackend(); this.backend = backend; backend.onStderrError((error) => { @@ -105,8 +99,25 @@ class KimiRemoteLauncher extends RemoteLauncherBase { backend, () => session.getPermissionMode() as PermissionMode | undefined ); - this.currentBackendModel = runtimeConfig.model; - this.applyDisplayMode(session.getPermissionMode() as PermissionMode, this.currentBackendModel); + // Model selection goes over ACP: new kimi-code ignores the KIMI_MODEL + // env var (only the KIMI_MODEL_NAME provider-synthesis family exists), + // so the resolved model is applied explicitly here. Without one, adopt + // the agent-reported current model so the UI shows the truth. + let effectiveModel: string | null = null; + if (runtimeConfig.model) { + effectiveModel = await this.applyInitialModel(backend, acpSessionId, runtimeConfig.model); + } + if (!effectiveModel) { + effectiveModel = backend.getConfigOptionByCategory(acpSessionId, 'model')?.currentValue + ?? backend.getSessionModelsMetadata(acpSessionId)?.currentModelId + ?? null; + } + this.currentBackendModel = effectiveModel; + if (effectiveModel) { + this.displayModel = effectiveModel; + messageBuffer.addMessage(`[MODEL:${effectiveModel}]`, 'system'); + } + this.applyDisplayMode(session.getPermissionMode() as PermissionMode, effectiveModel ?? undefined); this.setupAbortHandlers(session.client.rpcHandlerManager, { onAbort: () => this.handleAbort(), @@ -128,7 +139,7 @@ class KimiRemoteLauncher extends RemoteLauncherBase { if (batch.mode.model && batch.mode.model !== this.currentBackendModel) { if (!backend.setModel || this.setModelSupported === false) { - batch.mode.model = this.currentBackendModel!; + batch.mode.model = this.currentBackendModel ?? undefined; } else { logger.debug(`[kimi-remote] Switching model inline: ${this.currentBackendModel} -> ${batch.mode.model}`); try { @@ -152,7 +163,7 @@ class KimiRemoteLauncher extends RemoteLauncherBase { message: `Failed to switch model to ${batch.mode.model}. Continuing with ${this.currentBackendModel}.` }); } - batch.mode.model = this.currentBackendModel!; + batch.mode.model = this.currentBackendModel ?? undefined; } } } @@ -256,6 +267,42 @@ class KimiRemoteLauncher extends RemoteLauncherBase { } } + private async applyInitialModel( + backend: ReturnType, + sessionId: string, + model: string + ): Promise { + try { + await backend.setModel(sessionId, model); + this.setModelSupported = true; + return model; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (/method not found/i.test(message)) { + this.setModelSupported = false; + } + logger.debug('[kimi-remote] session/set_model failed, trying model config option', error); + } + + const option = backend.getConfigOptionByCategory(sessionId, 'model'); + if (!option) { + logger.warn(`[kimi-remote] Cannot apply model ${model}: agent exposes no model config option`); + return null; + } + try { + await backend.setConfigOption(sessionId, option.id, model); + return model; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + logger.warn(`[kimi-remote] Failed to apply model ${model}`, error); + this.session.sendSessionEvent({ + type: 'message', + message: `Failed to switch model to ${model}: ${message}. Using the agent default.` + }); + return null; + } + } + private applyDisplayMode(permissionMode: PermissionMode | undefined, model?: string): void { if (permissionMode && permissionMode !== this.displayPermissionMode) { this.displayPermissionMode = permissionMode; diff --git a/cli/src/kimi/utils/config.test.ts b/cli/src/kimi/utils/config.test.ts new file mode 100644 index 00000000..1e639fd7 --- /dev/null +++ b/cli/src/kimi/utils/config.test.ts @@ -0,0 +1,52 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { mkdir, rm, writeFile } from 'node:fs/promises'; +import { existsSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { resolveKimiRuntimeConfig } from './config'; + +describe('resolveKimiRuntimeConfig', () => { + let homeDir: string; + let previousHome: string | undefined; + + beforeEach(async () => { + previousHome = process.env.KIMI_CODE_HOME; + homeDir = join(tmpdir(), `kimi-cfg-${Date.now()}-${Math.random().toString(36).slice(2)}`); + await mkdir(homeDir, { recursive: true }); + process.env.KIMI_CODE_HOME = homeDir; + }); + + afterEach(async () => { + if (previousHome === undefined) { + delete process.env.KIMI_CODE_HOME; + } else { + process.env.KIMI_CODE_HOME = previousHome; + } + if (existsSync(homeDir)) { + await rm(homeDir, { recursive: true, force: true }); + } + }); + + it('prefers the explicit model', async () => { + await writeFile(join(homeDir, 'config.toml'), 'default_model = "kimi-code/k3"\n'); + expect(resolveKimiRuntimeConfig({ model: 'explicit-model' })).toEqual({ + model: 'explicit-model', + modelSource: 'explicit' + }); + }); + + it('reads default_model from the new kimi-code home', async () => { + await writeFile(join(homeDir, 'config.toml'), 'default_model = "kimi-code/k3"\n'); + expect(resolveKimiRuntimeConfig()).toEqual({ + model: 'kimi-code/k3', + modelSource: 'local' + }); + }); + + it('returns no model when nothing is configured (no hardcoded fallback)', () => { + expect(resolveKimiRuntimeConfig()).toEqual({ + model: undefined, + modelSource: 'default' + }); + }); +}); diff --git a/cli/src/kimi/utils/config.ts b/cli/src/kimi/utils/config.ts index e3ae7ebd..4f20ae0b 100644 --- a/cli/src/kimi/utils/config.ts +++ b/cli/src/kimi/utils/config.ts @@ -3,16 +3,28 @@ import { homedir } from 'node:os'; import { join } from 'node:path'; import { logger } from '@/ui/logger'; -export const KIMI_MODEL_ENV = 'KIMI_MODEL'; - export type KimiLocalConfig = { model?: string; }; -export type KimiModelSource = 'explicit' | 'env' | 'local' | 'default'; +export type KimiModelSource = 'explicit' | 'local' | 'default'; -const KIMI_DIR = join(homedir(), '.kimi'); -const CONFIG_PATH = join(KIMI_DIR, 'config.toml'); +const LEGACY_KIMI_DIR = join(homedir(), '.kimi'); + +/** + * kimi-code data root (sessions, config, logs). Overridable via KIMI_CODE_HOME; + * defaults to ~/.kimi-code. See kimi-code docs `configuration/data-locations`. + */ +export function getKimiCodeHome(): string { + return process.env.KIMI_CODE_HOME || join(homedir(), '.kimi-code'); +} + +function getConfigCandidates(): string[] { + return [ + join(getKimiCodeHome(), 'config.toml'), + join(LEGACY_KIMI_DIR, 'config.toml') + ]; +} function readTomlFile(path: string): Record | null { if (!existsSync(path)) { @@ -54,50 +66,34 @@ function extractModel(config: Record): string | undefined { } export function readKimiLocalConfig(): KimiLocalConfig { - const configFile = readTomlFile(CONFIG_PATH); - - return { - model: configFile ? extractModel(configFile) : undefined - }; + for (const candidate of getConfigCandidates()) { + const configFile = readTomlFile(candidate); + const model = configFile ? extractModel(configFile) : undefined; + if (model) { + return { model }; + } + } + return {}; } +/** + * Resolves which model alias hapi should ask kimi-code to use. Returns + * `model: undefined` when nothing is configured — callers must then omit + * `--model` entirely so kimi-code falls back to its own `default_model` + * (there is no valid built-in alias hapi could hardcode; model aliases are + * user-defined `[models.""]` entries in kimi-code's config.toml). + */ export function resolveKimiRuntimeConfig(opts: { model?: string; -} = {}): { model: string; modelSource: KimiModelSource } { +} = {}): { model: string | undefined; modelSource: KimiModelSource } { + if (opts.model) { + return { model: opts.model, modelSource: 'explicit' }; + } + const local = readKimiLocalConfig(); - - let modelSource: KimiModelSource = 'default'; - let model: string = 'kimi-k2'; - - if (opts.model) { - model = opts.model; - modelSource = 'explicit'; - } else if (process.env[KIMI_MODEL_ENV]) { - model = process.env[KIMI_MODEL_ENV]!; - modelSource = 'env'; - } else if (local.model) { - model = local.model; - modelSource = 'local'; + if (local.model) { + return { model: local.model, modelSource: 'local' }; } - return { model, modelSource }; -} - -export function buildKimiEnv(opts: { - model?: string; - cwd?: string; -}): NodeJS.ProcessEnv { - const env: NodeJS.ProcessEnv = { - ...process.env - }; - - if (opts.model) { - env[KIMI_MODEL_ENV] = opts.model; - } - - if (opts.cwd) { - env.KIMI_PROJECT_DIR = opts.cwd; - } - - return env; + return { model: undefined, modelSource: 'default' }; } diff --git a/cli/src/kimi/utils/kimiBackend.ts b/cli/src/kimi/utils/kimiBackend.ts index e99563b0..ed4ce569 100644 --- a/cli/src/kimi/utils/kimiBackend.ts +++ b/cli/src/kimi/utils/kimiBackend.ts @@ -1,5 +1,4 @@ import { AcpSdkBackend } from '@/agent/backends/acp'; -import { buildKimiEnv } from './config'; function filterEnv(env: NodeJS.ProcessEnv): Record { const result: Record = {}; @@ -11,20 +10,17 @@ function filterEnv(env: NodeJS.ProcessEnv): Record { return result; } -export function createKimiBackend(opts: { - model?: string; - resumeSessionId?: string | null; - cwd?: string; - permissionMode?: string; -}): AcpSdkBackend { - const env = filterEnv(buildKimiEnv({ - model: opts.model, - cwd: opts.cwd - })); - +/** + * Creates the ACP backend for `kimi acp`. Model selection is intentionally + * NOT passed via environment: new kimi-code ignores a plain KIMI_MODEL var + * (only the KIMI_MODEL_NAME provider-synthesis family exists), so the model + * is applied over ACP (`session/set_model` / `session/set_config_option`) + * after session creation — see kimiRemoteLauncher. + */ +export function createKimiBackend(): AcpSdkBackend { return new AcpSdkBackend({ command: 'kimi', args: ['acp'], - env + env: filterEnv(process.env) }); } diff --git a/cli/src/kimi/utils/kimiWireLocator.test.ts b/cli/src/kimi/utils/kimiWireLocator.test.ts new file mode 100644 index 00000000..719af9b2 --- /dev/null +++ b/cli/src/kimi/utils/kimiWireLocator.test.ts @@ -0,0 +1,155 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { mkdir, rm, writeFile } from 'node:fs/promises'; +import { existsSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { + createKimiWireLocator, + encodeKimiWorkDirKey, + getKimiWirePath, + type LocatedKimiWire +} from './kimiWireLocator'; + +const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +describe('encodeKimiWorkDirKey', () => { + it('matches the kimi-code workspace id scheme', () => { + // wd__ + expect(encodeKimiWorkDirKey('/Users/weishu/dev/github/hapi')).toBe('wd_hapi_dd2c162dd303'); + expect(encodeKimiWorkDirKey('/tmp')).toMatch(/^wd_tmp_[0-9a-f]{12}$/); + expect(encodeKimiWorkDirKey('/home/user/My Project!')).toMatch(/^wd_my-project_[0-9a-f]{12}$/); + }); +}); + +describe('kimiWireLocator', () => { + let homeDir: string; + let workDir: string; + let previousHome: string | undefined; + + beforeEach(async () => { + previousHome = process.env.KIMI_CODE_HOME; + homeDir = join(tmpdir(), `kimi-home-${Date.now()}-${Math.random().toString(36).slice(2)}`); + workDir = join(tmpdir(), `kimi-wd-${Date.now()}-${Math.random().toString(36).slice(2)}`); + await mkdir(homeDir, { recursive: true }); + await mkdir(workDir, { recursive: true }); + process.env.KIMI_CODE_HOME = homeDir; + }); + + afterEach(async () => { + if (previousHome === undefined) { + delete process.env.KIMI_CODE_HOME; + } else { + process.env.KIMI_CODE_HOME = previousHome; + } + for (const dir of [homeDir, workDir]) { + if (existsSync(dir)) { + await rm(dir, { recursive: true, force: true }); + } + } + }); + + async function seedSession(sessionId: string): Promise { + const sessionDir = join(homeDir, 'sessions', encodeKimiWorkDirKey(workDir), sessionId); + await mkdir(join(sessionDir, 'agents', 'main'), { recursive: true }); + await writeFile(join(sessionDir, 'state.json'), JSON.stringify({ workDir })); + await writeFile(getKimiWirePath(sessionDir), JSON.stringify({ type: 'metadata', protocol_version: '1.4', created_at: Date.now() }) + '\n'); + return sessionDir; + } + + it('locates a freshly created session and resolves its wire path', async () => { + const located: LocatedKimiWire[] = []; + const locator = createKimiWireLocator({ + cwd: workDir, + startupTimestampMs: Date.now(), + intervalMs: 50, + onLocated: (result) => located.push(result) + }); + + try { + // Session dir appears slightly after hapi's launch timestamp. + await wait(100); + await seedSession('session_aaa-bbb'); + + await wait(400); + expect(located).toHaveLength(1); + expect(located[0]?.sessionId).toBe('session_aaa-bbb'); + expect(located[0]?.wirePath).toBe(getKimiWirePath(join(homeDir, 'sessions', encodeKimiWorkDirKey(workDir), 'session_aaa-bbb'))); + } finally { + await locator.cleanup(); + } + }); + + it('targets the resumed session id directly, ignoring creation time', async () => { + await seedSession('session_old-resume'); + + const located: LocatedKimiWire[] = []; + const locator = createKimiWireLocator({ + cwd: workDir, + startupTimestampMs: Date.now() + 60_000, // dir is "older" than launch + resumeSessionId: 'session_old-resume', + intervalMs: 50, + onLocated: (result) => located.push(result) + }); + + try { + await wait(300); + expect(located).toHaveLength(1); + expect(located[0]?.sessionId).toBe('session_old-resume'); + } finally { + await locator.cleanup(); + } + }); + + it('excludes sessions that already existed when the locator started (retry race)', async () => { + // Simulate an immediate `hapi kimi` retry: the previous launch's + // session dir already exists and is within the birth-time grace + // window, but it must not be adopted — the new process has not + // created its session yet. + await seedSession('session_previous-launch'); + + const located: LocatedKimiWire[] = []; + const locator = createKimiWireLocator({ + cwd: workDir, + startupTimestampMs: Date.now() - 1000, // dir birth time is inside the grace window + intervalMs: 50, + onLocated: (result) => located.push(result) + }); + + try { + await wait(300); + expect(located).toHaveLength(0); + + // The session created by the new process is still picked up. + await seedSession('session_new-launch'); + await wait(300); + expect(located).toHaveLength(1); + expect(located[0]?.sessionId).toBe('session_new-launch'); + } finally { + await locator.cleanup(); + } + }); + + it('refuses to attach when multiple fresh sessions appear', async () => { + const ambiguous: string[][] = []; + const located: LocatedKimiWire[] = []; + const locator = createKimiWireLocator({ + cwd: workDir, + startupTimestampMs: Date.now(), + intervalMs: 50, + onLocated: (result) => located.push(result), + onAmbiguous: (ids) => ambiguous.push(ids) + }); + + try { + await seedSession('session_one'); + await seedSession('session_two'); + await wait(300); + + expect(located).toHaveLength(0); + expect(ambiguous).toHaveLength(1); + expect(ambiguous[0]?.sort()).toEqual(['session_one', 'session_two']); + } finally { + await locator.cleanup(); + } + }); +}); diff --git a/cli/src/kimi/utils/kimiWireLocator.ts b/cli/src/kimi/utils/kimiWireLocator.ts new file mode 100644 index 00000000..91e2e59f --- /dev/null +++ b/cli/src/kimi/utils/kimiWireLocator.ts @@ -0,0 +1,249 @@ +import { createHash } from 'node:crypto'; +import { readdir, readFile, stat } from 'node:fs/promises'; +import { join, resolve } from 'node:path'; +import { logger } from '@/ui/logger'; +import { getKimiCodeHome } from './config'; + +export type LocatedKimiWire = { + sessionId: string; + wirePath: string; +}; + +export type KimiWireLocator = { + ready: Promise; + cleanup: () => Promise; +}; + +type KimiWireLocatorOptions = { + cwd: string; + startupTimestampMs: number; + resumeSessionId?: string | null; + intervalMs?: number; + onLocated: (located: LocatedKimiWire) => void; + onAmbiguous?: (sessionIds: string[]) => void; +}; + +const DEFAULT_LOCATOR_INTERVAL_MS = 500; +// Grace for filesystem timestamp skew between hapi recording its launch time +// and kimi-code creating the session directory right after. +const STARTUP_GRACE_MS = 2000; + +const WORKDIR_KEY_PREFIX = 'wd_'; +const WORKDIR_HASH_LENGTH = 12; +const MAX_WORKDIR_SLUG_LENGTH = 40; + +function slugifyWorkDirName(name: string): string { + const slug = name + .toLowerCase() + .replaceAll(/[^a-z0-9._-]+/g, '-') + .replaceAll(/^-+|-+$/g, '') + .slice(0, MAX_WORKDIR_SLUG_LENGTH) + .replaceAll(/^-+|-+$/g, ''); + return slug === '' || slug === '.' || slug === '..' ? 'workspace' : slug; +} + +/** + * Mirrors kimi-code's workspace identity (`packages/agent-core-v2/src/_base/utils/workdir-slug.ts`): + * `wd__`. Sessions for a working + * directory live under `/sessions//session_/`. + */ +export function encodeKimiWorkDirKey(workDir: string): string { + const normalized = workDir.replace(/\\/g, '/').replace(/\/+$/, ''); + const base = normalized.split('/').pop() ?? normalized; + const slug = slugifyWorkDirName(base); + const hash = createHash('sha256').update(normalized).digest('hex').slice(0, WORKDIR_HASH_LENGTH); + return `${WORKDIR_KEY_PREFIX}${slug}_${hash}`; +} + +export function getKimiWorkspaceDir(workDir: string): string { + return join(getKimiCodeHome(), 'sessions', encodeKimiWorkDirKey(workDir)); +} + +export function getKimiWirePath(sessionDir: string): string { + return join(sessionDir, 'agents', 'main', 'wire.jsonl'); +} + +/** + * Polls the kimi-code session storage for the session the locally spawned + * `kimi` process just created in this working directory, and resolves with + * its wire transcript path. Mirrors the codex transcript locator: sessions + * created before hapi's launch are ignored; multiple fresh candidates are + * treated as ambiguous rather than attaching to the wrong session. + */ +export function createKimiWireLocator(options: KimiWireLocatorOptions): KimiWireLocator { + const locator = new KimiWireLocatorImpl(options); + const ready = locator.start().catch((error) => { + logger.debug('[kimi-wire-locator] Failed to initialize', error); + }); + return { + ready, + cleanup: async () => { + await locator.cleanup(); + await ready; + } + }; +} + +class KimiWireLocatorImpl { + private readonly workspaceDir: string; + private readonly targetCwd: string; + private readonly startupTimestampMs: number; + private readonly resumeSessionId: string | null; + private readonly intervalMs: number; + private readonly onLocated: KimiWireLocatorOptions['onLocated']; + private readonly onAmbiguous?: KimiWireLocatorOptions['onAmbiguous']; + private readonly initialSessionIds = new Set(); + private interval: ReturnType | null = null; + private scanPromise: Promise | null = null; + private stopped = false; + + constructor(options: KimiWireLocatorOptions) { + this.workspaceDir = getKimiWorkspaceDir(options.cwd); + this.targetCwd = normalizePath(options.cwd); + this.startupTimestampMs = options.startupTimestampMs; + this.resumeSessionId = options.resumeSessionId ?? null; + this.intervalMs = options.intervalMs ?? DEFAULT_LOCATOR_INTERVAL_MS; + this.onLocated = options.onLocated; + this.onAmbiguous = options.onAmbiguous; + } + + async start(): Promise { + if (!this.resumeSessionId) { + // Snapshot pre-existing sessions: only directories that appear + // AFTER this locator starts may belong to the process hapi just + // spawned. Without this, an immediate retry could bind to the + // previous launch's session (created inside the birth-time grace + // window) before the new process has written anything. + try { + const entries = await readdir(this.workspaceDir, { withFileTypes: true }); + for (const entry of entries) { + if (entry.isDirectory() && entry.name.startsWith('session_')) { + this.initialSessionIds.add(entry.name); + } + } + } catch { + // Workspace dir does not exist yet — nothing to exclude. + } + } + if (this.stopped) return; + + void this.scan(); + this.interval = setInterval(() => void this.scan(), this.intervalMs); + this.interval.unref?.(); + } + + async cleanup(): Promise { + this.stopped = true; + if (this.interval) { + clearInterval(this.interval); + this.interval = null; + } + await this.scanPromise?.catch(() => {}); + } + + private stopPolling(): void { + this.stopped = true; + if (this.interval) { + clearInterval(this.interval); + this.interval = null; + } + } + + private async scan(): Promise { + if (this.stopped || this.scanPromise) { + return this.scanPromise ?? Promise.resolve(); + } + this.scanPromise = this.runScan(); + try { + await this.scanPromise; + } finally { + this.scanPromise = null; + } + } + + private async runScan(): Promise { + const candidates = await this.listCandidates(); + if (this.stopped || candidates.length === 0) { + return; + } + + if (candidates.length > 1) { + logger.warn( + `[kimi-wire-locator] Ambiguous kimi sessions (${candidates.length} fresh candidates); refusing attachment`, + candidates.map((candidate) => candidate.sessionId) + ); + this.stopPolling(); + this.onAmbiguous?.(candidates.map((candidate) => candidate.sessionId)); + return; + } + + const [located] = candidates; + logger.debug(`[kimi-wire-locator] Located ${located.sessionId} at ${located.wirePath}`); + this.stopPolling(); + this.onLocated(located); + } + + private async listCandidates(): Promise { + let entries; + try { + entries = await readdir(this.workspaceDir, { withFileTypes: true }); + } catch { + return []; + } + + const candidates: LocatedKimiWire[] = []; + for (const entry of entries) { + if (!entry.isDirectory() || !entry.name.startsWith('session_')) { + continue; + } + if (this.resumeSessionId && entry.name !== this.resumeSessionId) { + continue; + } + + const sessionDir = join(this.workspaceDir, entry.name); + const wirePath = getKimiWirePath(sessionDir); + const wireStats = await stat(wirePath).catch(() => null); + if (!wireStats || !wireStats.isFile()) { + continue; + } + + if (!this.resumeSessionId) { + if (this.initialSessionIds.has(entry.name)) { + continue; + } + const dirStats = await stat(sessionDir).catch(() => null); + const createdMs = dirStats?.birthtimeMs && dirStats.birthtimeMs > 0 + ? dirStats.birthtimeMs + : dirStats?.mtimeMs ?? 0; + if (createdMs < this.startupTimestampMs - STARTUP_GRACE_MS) { + continue; + } + if (!(await this.matchesWorkDir(sessionDir))) { + continue; + } + } + + candidates.push({ sessionId: entry.name, wirePath }); + } + return candidates; + } + + private async matchesWorkDir(sessionDir: string): Promise { + try { + const raw = await readFile(join(sessionDir, 'state.json'), 'utf8'); + const parsed = JSON.parse(raw) as { workDir?: unknown }; + if (typeof parsed.workDir !== 'string' || parsed.workDir.length === 0) { + return true; + } + return normalizePath(parsed.workDir) === this.targetCwd; + } catch { + // state.json may not be written yet — do not exclude the candidate. + return true; + } + } +} + +function normalizePath(value: string): string { + const normalized = resolve(value); + return process.platform === 'win32' ? normalized.toLowerCase() : normalized; +} diff --git a/cli/src/kimi/utils/kimiWireScanner.test.ts b/cli/src/kimi/utils/kimiWireScanner.test.ts new file mode 100644 index 00000000..660255aa --- /dev/null +++ b/cli/src/kimi/utils/kimiWireScanner.test.ts @@ -0,0 +1,166 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { appendFile, mkdir, rm, writeFile } from 'node:fs/promises'; +import { existsSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { convertKimiWireEvent, createKimiWireScanner, type KimiWireScanner } from './kimiWireScanner'; + +const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +describe('convertKimiWireEvent', () => { + it('converts user prompts and steers', () => { + expect(convertKimiWireEvent({ + type: 'turn.prompt', + input: [{ type: 'text', text: 'hello' }], + origin: { kind: 'user' }, + time: 1 + })).toEqual({ userMessage: 'hello' }); + + expect(convertKimiWireEvent({ + type: 'turn.steer', + input: [{ type: 'text', text: 'focus now' }], + origin: { kind: 'user' } + })).toEqual({ userMessage: 'focus now' }); + }); + + it('ignores non-user prompts and unrelated records', () => { + expect(convertKimiWireEvent({ type: 'turn.prompt', input: [{ type: 'text', text: 'x' }], origin: { kind: 'system' } })).toBeNull(); + expect(convertKimiWireEvent({ type: 'metadata', protocol_version: '1.4', created_at: 1 })).toBeNull(); + expect(convertKimiWireEvent({ type: 'config.update', systemPrompt: '...' })).toBeNull(); + expect(convertKimiWireEvent({ type: 'llm.request', kind: 'loop' })).toBeNull(); + expect(convertKimiWireEvent({ type: 'usage.record', usage: {} })).toBeNull(); + expect(convertKimiWireEvent({ type: 'context.append_message', message: { role: 'user', content: [] } })).toBeNull(); + }); + + it('converts assistant text and thinking parts', () => { + expect(convertKimiWireEvent({ + type: 'context.append_loop_event', + event: { type: 'content.part', uuid: 'u1', part: { type: 'text', text: 'answer' } } + })).toEqual({ message: { type: 'message', message: 'answer' } }); + + expect(convertKimiWireEvent({ + type: 'context.append_loop_event', + event: { type: 'content.part', uuid: 'u2', part: { type: 'think', think: 'hmm' } } + })).toEqual({ message: { type: 'reasoning', message: 'hmm', id: 'u2' } }); + }); + + it('converts tool calls and results', () => { + expect(convertKimiWireEvent({ + type: 'context.append_loop_event', + event: { type: 'tool.call', uuid: 'tool_1', toolCallId: 'tool_1', name: 'Grep', args: { pattern: 'kimi' } } + })).toEqual({ + message: { type: 'tool-call', name: 'Grep', callId: 'tool_1', input: { pattern: 'kimi' } } + }); + + expect(convertKimiWireEvent({ + type: 'context.append_loop_event', + event: { type: 'tool.result', parentUuid: 'tool_1', toolCallId: 'tool_1', result: { output: 'matches' } } + })).toEqual({ + message: { type: 'tool-call-result', callId: 'tool_1', output: 'matches' } + }); + }); + + it('forwards tool failure status as is_error', () => { + expect(convertKimiWireEvent({ + type: 'context.append_loop_event', + event: { type: 'tool.result', parentUuid: 'tool_2', toolCallId: 'tool_2', result: { output: 'boom', isError: true } } + })).toEqual({ + message: { type: 'tool-call-result', callId: 'tool_2', output: 'boom', is_error: true } + }); + }); + + it('converts step.end usage into token_count with cached input included', () => { + expect(convertKimiWireEvent({ + type: 'context.append_loop_event', + event: { + type: 'step.end', + uuid: 's1', + usage: { inputOther: 100, output: 20, inputCacheRead: 50, inputCacheCreation: 10 } + } + })).toEqual({ + message: { + type: 'token_count', + info: { total: { inputTokens: 160, outputTokens: 20, cachedInputTokens: 50 } } + } + }); + }); + + it('ignores step.begin and unknown loop events', () => { + expect(convertKimiWireEvent({ + type: 'context.append_loop_event', + event: { type: 'step.begin', uuid: 's0', step: 1 } + })).toBeNull(); + }); +}); + +describe('kimiWireScanner', () => { + let testDir: string; + let wirePath: string; + let scanner: KimiWireScanner | null = null; + let events: { type: string }[] = []; + + beforeEach(async () => { + testDir = join(tmpdir(), `kimi-wire-${Date.now()}-${Math.random().toString(36).slice(2)}`); + await mkdir(testDir, { recursive: true }); + wirePath = join(testDir, 'wire.jsonl'); + events = []; + }); + + afterEach(async () => { + if (scanner) { + await scanner.cleanup(); + scanner = null; + } + if (existsSync(testDir)) { + await rm(testDir, { recursive: true, force: true }); + } + }); + + it('primes to EOF on attach and only emits new events', async () => { + await writeFile( + wirePath, + [ + JSON.stringify({ type: 'metadata', protocol_version: '1.4', created_at: 1 }), + JSON.stringify({ type: 'turn.prompt', input: [{ type: 'text', text: 'old' }], origin: { kind: 'user' } }) + ].join('\n') + '\n' + ); + + scanner = await createKimiWireScanner({ + wirePath, + onEvent: (event) => events.push(event) + }); + + await wait(300); + expect(events).toHaveLength(0); + + await appendFile( + wirePath, + JSON.stringify({ + type: 'context.append_loop_event', + event: { type: 'content.part', uuid: 'u1', part: { type: 'text', text: 'new answer' } } + }) + '\n' + ); + + await wait(700); + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe('context.append_loop_event'); + }); + + it('handles lines split across reads', async () => { + await writeFile(wirePath, ''); + scanner = await createKimiWireScanner({ + wirePath, + onEvent: (event) => events.push(event) + }); + + const line = JSON.stringify({ type: 'turn.prompt', input: [{ type: 'text', text: 'hi' }], origin: { kind: 'user' } }); + await appendFile(wirePath, line.slice(0, 30)); + await wait(300); + expect(events).toHaveLength(0); + + await appendFile(wirePath, line.slice(30) + '\n'); + await wait(700); + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe('turn.prompt'); + }); +}); diff --git a/cli/src/kimi/utils/kimiWireScanner.ts b/cli/src/kimi/utils/kimiWireScanner.ts new file mode 100644 index 00000000..24e3a2d8 --- /dev/null +++ b/cli/src/kimi/utils/kimiWireScanner.ts @@ -0,0 +1,341 @@ +import { randomUUID } from 'node:crypto'; +import { open, stat } from 'node:fs/promises'; +import { BaseSessionScanner, type SessionFileScanResult, type SessionFileScanStats } from '@/modules/common/session/BaseSessionScanner'; +import { logger } from '@/ui/logger'; +import type { CodexMessage } from '@/agent/messageConverter'; + +/** + * One line of a kimi-code `wire.jsonl` journal: `{ type, ...payload, time }`. + * See `packages/agent-core-v2/src/wire/record.ts` (`opToWireRecord`) in + * MoonshotAI/kimi-code — payload fields are flattened onto the record. + */ +export type KimiWireEvent = { + type: string; + time?: number; + [key: string]: unknown; +}; + +export type KimiWireConversion = { + userMessage?: string; + message?: CodexMessage; +}; + +function asRecord(value: unknown): Record | null { + return value && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : null; +} + +function asString(value: unknown): string | null { + return typeof value === 'string' && value.length > 0 ? value : null; +} + +function asFiniteNumber(value: unknown): number | null { + return typeof value === 'number' && Number.isFinite(value) ? value : null; +} + +function extractInputText(input: unknown): string | null { + if (!Array.isArray(input)) { + return null; + } + const parts: string[] = []; + for (const block of input) { + const record = asRecord(block); + if (record?.type === 'text') { + const text = asString(record.text); + if (text) { + parts.push(text); + } + } + } + return parts.length > 0 ? parts.join('\n') : null; +} + +/** + * Converts a kimi-code wire journal event into hapi's codex-family message + * shapes (same wire contract the hub and web already render for ACP agents). + * + * Mapped events (observed on wire protocol 1.4, kimi-code 0.26): + * turn.prompt / turn.steer (origin.kind = 'user') → user message + * context.append_loop_event/content.part (text) → assistant message + * context.append_loop_event/content.part (think) → reasoning + * context.append_loop_event/tool.call → tool-call + * context.append_loop_event/tool.result → tool-call-result + * context.append_loop_event/step.end → token_count + * Everything else (metadata, config.update, llm.request, usage.record, + * step.begin, plan_mode.*, …) is ignored. + */ +export function convertKimiWireEvent(event: KimiWireEvent): KimiWireConversion | null { + if (event.type === 'turn.prompt' || event.type === 'turn.steer') { + const origin = asRecord(event.origin); + if (asString(origin?.kind) !== 'user') { + return null; + } + const text = extractInputText(event.input); + return text ? { userMessage: text } : null; + } + + if (event.type !== 'context.append_loop_event') { + return null; + } + + const loopEvent = asRecord(event.event); + const loopType = asString(loopEvent?.type); + if (!loopEvent || !loopType) { + return null; + } + + if (loopType === 'content.part') { + const part = asRecord(loopEvent.part); + const partType = asString(part?.type); + if (partType === 'text') { + const text = asString(part?.text); + return text + ? { message: { type: 'message', message: text } } + : null; + } + if (partType === 'think') { + const think = asString(part?.think); + return think + ? { message: { type: 'reasoning', message: think, id: asString(loopEvent.uuid) ?? randomUUID() } } + : null; + } + return null; + } + + if (loopType === 'tool.call') { + const name = asString(loopEvent.name); + const callId = asString(loopEvent.toolCallId) ?? asString(loopEvent.uuid); + if (!name || !callId) { + return null; + } + return { + message: { + type: 'tool-call', + name, + callId, + input: loopEvent.args ?? null + } + }; + } + + if (loopType === 'tool.result') { + const callId = asString(loopEvent.toolCallId) ?? asString(loopEvent.parentUuid); + if (!callId) { + return null; + } + const result = asRecord(loopEvent.result); + return { + message: { + type: 'tool-call-result', + callId, + output: result ? (result.output ?? null) : null, + // kimi-code wire: result.isError === true marks a failed tool + ...(result?.isError === true ? { is_error: true } : {}) + } + }; + } + + if (loopType === 'step.end') { + const usage = asRecord(loopEvent.usage); + if (!usage) { + return null; + } + // kimi-code splits input into uncached (`inputOther`) and cached + // portions; hapi's inputTokens contract expects the full input total. + const inputOther = asFiniteNumber(usage.inputOther) ?? 0; + const cacheRead = asFiniteNumber(usage.inputCacheRead) ?? 0; + const cacheCreation = asFiniteNumber(usage.inputCacheCreation) ?? 0; + return { + message: { + type: 'token_count', + info: { + total: { + inputTokens: inputOther + cacheRead + cacheCreation, + outputTokens: asFiniteNumber(usage.output) ?? 0, + cachedInputTokens: cacheRead + } + } + } + }; + } + + return null; +} + +interface KimiWireScannerOptions { + wirePath: string; + onEvent: (event: KimiWireEvent) => void; +} + +export interface KimiWireScanner { + cleanup: () => Promise; +} + +export async function createKimiWireScanner(opts: KimiWireScannerOptions): Promise { + const scanner = new KimiWireScannerImpl(opts); + await scanner.start(); + return { + cleanup: async () => { + await scanner.cleanup(); + } + }; +} + +class KimiWireScannerImpl extends BaseSessionScanner { + private readonly wirePath: string; + private readonly onEvent: (event: KimiWireEvent) => void; + private fileEpoch = 0; + private fileState: { + device: number; + inode: number; + partialLine: Buffer; + nextLineIndex: number; + } | null = null; + + constructor(opts: KimiWireScannerOptions) { + super({ intervalMs: 2000 }); + this.wirePath = opts.wirePath; + this.onEvent = opts.onEvent; + } + + protected async initialize(): Promise { + // Prime to EOF: only events written after hapi attaches are forwarded, + // so reopening an existing session does not replay its whole history. + const { events, nextCursor } = await this.readWire(0); + const keys = events.map((entry) => this.generateEventKey(entry.event, { + filePath: this.wirePath, + lineIndex: entry.lineIndex + })); + this.seedProcessedKeys(keys); + this.setCursor(this.wirePath, nextCursor); + } + + protected async findSessionFiles(): Promise { + return [this.wirePath]; + } + + protected shouldWatchFile(filePath: string): boolean { + return filePath === this.wirePath; + } + + protected async parseSessionFile(_filePath: string, cursor: number): Promise> { + return this.readWire(cursor); + } + + protected generateEventKey(_event: KimiWireEvent, context: { filePath: string; lineIndex?: number }): string { + return `${context.filePath}:${this.fileEpoch}:${context.lineIndex ?? -1}`; + } + + protected async handleFileScan(stats: SessionFileScanStats): Promise { + for (const event of stats.events) { + this.onEvent(event); + } + if (stats.newCount > 0) { + logger.debug(`[kimi-wire-scanner] ${stats.newCount} new events from ${stats.filePath}`); + } + } + + private async readWire(startOffset: number): Promise> { + let fileStats; + try { + fileStats = await stat(this.wirePath); + } catch (error) { + logger.debug(`[kimi-wire-scanner] Failed to stat wire file ${this.wirePath}: ${error}`); + return { events: [], nextCursor: startOffset }; + } + + const previous = this.fileState; + const identityChanged = Boolean( + previous + && (previous.device !== fileStats.dev || previous.inode !== fileStats.ino) + ); + let effectiveStartOffset = startOffset; + let partialLine = previous?.partialLine ?? Buffer.alloc(0); + let nextLineIndex = previous?.nextLineIndex ?? 0; + + if (identityChanged || fileStats.size < effectiveStartOffset) { + effectiveStartOffset = 0; + partialLine = Buffer.alloc(0); + nextLineIndex = 0; + this.fileEpoch += 1; + } + + const bytesToRead = fileStats.size - effectiveStartOffset; + let appended: Buffer = Buffer.alloc(0); + if (bytesToRead > 0) { + try { + appended = await readWireRange(this.wirePath, effectiveStartOffset, bytesToRead); + } catch (error) { + logger.debug(`[kimi-wire-scanner] Failed to read wire file ${this.wirePath}: ${error}`); + return { events: [], nextCursor: startOffset }; + } + } + + const content = partialLine.length > 0 + ? Buffer.concat([partialLine, appended]) + : appended; + const events: { event: KimiWireEvent; lineIndex: number }[] = []; + + const parseLine = (lineBuffer: Buffer, lineIndex: number, allowIncomplete: boolean): boolean => { + const line = lineBuffer.toString('utf-8'); + if (!line || line.trim().length === 0) return true; + try { + const parsed = JSON.parse(line) as KimiWireEvent; + if (typeof parsed?.type === 'string' && parsed.type.length > 0) { + events.push({ event: parsed, lineIndex }); + } + return true; + } catch (error) { + if (!allowIncomplete) { + logger.debug(`[kimi-wire-scanner] Failed to parse wire line ${this.wirePath}:${lineIndex + 1}: ${error}`); + } + return false; + } + }; + + let lineStart = 0; + for (let index = 0; index < content.length; index += 1) { + if (content[index] !== 0x0a) continue; + parseLine(content.subarray(lineStart, index), nextLineIndex, false); + nextLineIndex += 1; + lineStart = index + 1; + } + + const trailing = content.subarray(lineStart); + if (trailing.length > 0 && parseLine(trailing, nextLineIndex, true)) { + partialLine = Buffer.alloc(0); + nextLineIndex += 1; + } else { + partialLine = Buffer.from(trailing); + } + + this.fileState = { + device: fileStats.dev, + inode: fileStats.ino, + partialLine, + nextLineIndex + }; + + return { + events, + nextCursor: effectiveStartOffset + appended.length + }; + } +} + +async function readWireRange(filePath: string, startOffset: number, length: number): Promise { + const content = Buffer.allocUnsafe(length); + let bytesRead = 0; + const handle = await open(filePath, 'r'); + try { + while (bytesRead < length) { + const result = await handle.read(content, bytesRead, length - bytesRead, startOffset + bytesRead); + if (result.bytesRead === 0) break; + bytesRead += result.bytesRead; + } + } finally { + await handle.close(); + } + return bytesRead === content.length ? content : content.subarray(0, bytesRead); +}