mirror of
https://github.com/wu736139669/hapi.git
synced 2026-08-05 06:24:37 +00:00
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_<slug>_<sha256(cwd).12>, 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 <KIMI_CODE_HOME|~/.kimi-code>/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.
This commit is contained in:
weishu
@@ -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)}`);
|
||||
|
||||
@@ -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<void> | null = null;
|
||||
|
||||
const attachWireScanner = (wirePath: string): Promise<void> => {
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<typeof createKimiBackend>,
|
||||
sessionId: string,
|
||||
model: string
|
||||
): Promise<string | null> {
|
||||
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;
|
||||
|
||||
@@ -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'
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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<string, unknown> | null {
|
||||
if (!existsSync(path)) {
|
||||
@@ -54,50 +66,34 @@ function extractModel(config: Record<string, unknown>): 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."<alias>"]` 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' };
|
||||
}
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import { AcpSdkBackend } from '@/agent/backends/acp';
|
||||
import { buildKimiEnv } from './config';
|
||||
|
||||
function filterEnv(env: NodeJS.ProcessEnv): Record<string, string> {
|
||||
const result: Record<string, string> = {};
|
||||
@@ -11,20 +10,17 @@ function filterEnv(env: NodeJS.ProcessEnv): Record<string, string> {
|
||||
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)
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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_<slug>_<sha256(normalizedWorkDir).slice(0,12)>
|
||||
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<string> {
|
||||
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();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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<void>;
|
||||
cleanup: () => Promise<void>;
|
||||
};
|
||||
|
||||
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_<slug>_<sha256(normalizedWorkDir).slice(0,12)>`. Sessions for a working
|
||||
* directory live under `<KIMI_CODE_HOME>/sessions/<workspaceId>/session_<id>/`.
|
||||
*/
|
||||
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<string>();
|
||||
private interval: ReturnType<typeof setInterval> | null = null;
|
||||
private scanPromise: Promise<void> | 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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<LocatedKimiWire[]> {
|
||||
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<boolean> {
|
||||
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;
|
||||
}
|
||||
@@ -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');
|
||||
});
|
||||
});
|
||||
@@ -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<string, unknown> | null {
|
||||
return value && typeof value === 'object' && !Array.isArray(value)
|
||||
? value as Record<string, unknown>
|
||||
: 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<void>;
|
||||
}
|
||||
|
||||
export async function createKimiWireScanner(opts: KimiWireScannerOptions): Promise<KimiWireScanner> {
|
||||
const scanner = new KimiWireScannerImpl(opts);
|
||||
await scanner.start();
|
||||
return {
|
||||
cleanup: async () => {
|
||||
await scanner.cleanup();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
class KimiWireScannerImpl extends BaseSessionScanner<KimiWireEvent> {
|
||||
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<void> {
|
||||
// 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<string[]> {
|
||||
return [this.wirePath];
|
||||
}
|
||||
|
||||
protected shouldWatchFile(filePath: string): boolean {
|
||||
return filePath === this.wirePath;
|
||||
}
|
||||
|
||||
protected async parseSessionFile(_filePath: string, cursor: number): Promise<SessionFileScanResult<KimiWireEvent>> {
|
||||
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<KimiWireEvent>): Promise<void> {
|
||||
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<SessionFileScanResult<KimiWireEvent>> {
|
||||
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<Buffer> {
|
||||
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);
|
||||
}
|
||||
Reference in New Issue
Block a user