feat(server-v2): wire server-v2 into CLI and close v1 parity gaps

- cli: opt into server-v2 via KIMI_CODE_EXPERIMENTAL_FLAG, lazy-loaded behind a RoutedServer adapter so the v1 module graph stays untouched by default
- build: add hashImportsPlugin (rolldown/tsdown) to resolve #/ subpath imports of inlined agent-core-v2 when bundling server-v2
- server-v2: add --dangerous-bypass-auth flag disabling bearer auth on every route, surfaced on /meta via dangerous_bypass_auth
- server-v2: serve bundled web UI at GET / + SPA fallback when webAssetsDir is set (auth-exempt, mirroring v1)
- server-v2: snapshot auto reader reads state.json + wire.jsonl from disk with a hard timeout; legacy keeps the live assembly as escape hatch
- server-v2: POST /sessions/{id}/profile now updates title, metadata, and agent_config and broadcasts session.meta.updated on title change
- server-v2: session list/children project live ISessionActivity status and filter by status; add archived_only query and workspace_id validation
- server-v2: close cwd gap G3 — persist cwd on session metadata and surface it on the index so unregistered workspaces keep their cwd
- server-v2: GET /warnings surfaces agents-md-oversized; add GET /workspaces/{id}/skills session-less scan
- server-v2: broadcaster reads the cold watermark from the disk journal and broadcasts title updates to every connection
- agent-core-v2: fix plan-file path resolution in AgentPlanService
- server: rename IFileService to IFileStore
This commit is contained in:
haozhe.yang 2026-07-06 21:51:52 +08:00
parent 1854e6df29
commit 590eca5ccd
36 changed files with 2595 additions and 268 deletions

View file

@ -87,6 +87,7 @@
"@moonshot-ai/migration-legacy": "workspace:^",
"@moonshot-ai/pi-tui": "workspace:^",
"@moonshot-ai/server": "workspace:^",
"@moonshot-ai/server-v2": "workspace:^",
"@moonshot-ai/vis-server": "workspace:^",
"@moonshot-ai/vis-web": "workspace:*",
"@types/semver": "^7.7.0",

View file

@ -46,7 +46,7 @@ function executableLines() {
}
for (const line of executableLines()) {
for (const match of line.matchAll(/\brequire\(\s*["']([^"']+)["']\s*\)/g)) {
for (const match of line.matchAll(/(?<![.\w])require\(\s*["']([^"']+)["']\s*\)/g)) {
const specifier = match[1];
if (specifier.startsWith('.') || specifier.startsWith('/')) {
if (optionalRelativeRuntimeRequires.has(specifier)) continue;

View file

@ -14,7 +14,7 @@
import { join } from 'node:path';
import { shutdownTelemetry, track } from '@moonshot-ai/kimi-telemetry';
import { startServer, type RunningServer } from '@moonshot-ai/server';
import { startServer, type ServerLogger } from '@moonshot-ai/server';
import chalk from 'chalk';
import { Option, type Command } from 'commander';
@ -48,6 +48,33 @@ import {
const WEB_ASSETS_DIR = 'dist-web';
/**
* Master experimental switch. When set to a truthy value (`1`/`true`/`yes`/`on`),
* `kimi server run` boots the DI × Scope engine server (`@moonshot-ai/server-v2`)
* instead of the default `@moonshot-ai/server`. Read directly from the env
* (matching `cli/update/rollout.ts`) because the CLI must not depend on the
* core flag registry. Unset / any other value keeps the v1 server.
*/
const SERVER_V2_ENV = 'KIMI_CODE_EXPERIMENTAL_FLAG';
const SERVER_V2_TRUTHY_VALUES = new Set(['1', 'true', 'yes', 'on']);
export function isServerV2Enabled(
env: Readonly<Record<string, string | undefined>> = process.env,
): boolean {
return SERVER_V2_TRUTHY_VALUES.has((env[SERVER_V2_ENV] ?? '').trim().toLowerCase());
}
/**
* Minimal surface `runServerInProcess` needs from either server flavor. v1's
* `RunningServer` already satisfies it; v2's `RunningServer` is adapted to it
* (it returns `{ host, port, close }` instead of `{ address, logger, close }`).
*/
interface RoutedServer {
readonly address: string;
readonly logger: ServerLogger;
close(): Promise<void>;
}
export interface RunCliOptions extends ServerCliOptions {
open?: boolean;
/** Run the server in-process instead of spawning a background daemon. */
@ -342,7 +369,7 @@ async function runServerInProcess(
const version = getVersion();
const telemetry = initializeServerTelemetry({ version });
let running: RunningServer | undefined;
let running: RoutedServer | undefined;
let stopping = false;
// Idle auto-shutdown is only for the on-demand personal daemon. It is skipped
@ -375,30 +402,77 @@ async function runServerInProcess(
process.exit(0);
}
running = await startServer({
host: options.host,
port: options.port,
logLevel: options.logLevel,
debugEndpoints: options.debugEndpoints,
insecureNoTls: options.insecureNoTls,
allowRemoteShutdown: options.allowRemoteShutdown,
allowRemoteTerminals: options.allowRemoteTerminals,
dangerousBypassAuth: options.dangerousBypassAuth,
allowedHosts: options.allowedHosts,
webAssetsDir: serverWebAssetsDir(),
coreProcessOptions: {
identity: createKimiCodeHostIdentity(version),
telemetry,
},
wsGatewayOptions: {
telemetry,
onConnectionCountChange: idle
? (size) => {
idle.onConnectionCountChange(size);
}
: undefined,
},
});
if (isServerV2Enabled()) {
// Experimental: boot the DI × Scope engine server. v2 speaks the same
// `/api/v1` wire interface but its `startServer` returns `{ host, port,
// close }` rather than `{ address, logger, close }`, so adapt it to the
// `RoutedServer` surface the rest of this runner consumes. Loaded lazily so
// the default (flag-off) path keeps the exact v1-only module graph — v2 and
// its agent-core-v2 engine are only resolved when the flag is on.
const { createServerLogger: createServerV2Logger, startServer: startServerV2 } =
await import('@moonshot-ai/server-v2');
const logger = createServerV2Logger({ level: options.logLevel });
const v2 = await startServerV2({
host: options.host,
port: options.port,
logLevel: options.logLevel,
logger,
debugEndpoints: options.debugEndpoints,
insecureNoTls: options.insecureNoTls,
allowRemoteShutdown: options.allowRemoteShutdown,
allowRemoteTerminals: options.allowRemoteTerminals,
allowedHosts: options.allowedHosts,
disableAuth: options.dangerousBypassAuth,
webAssetsDir: serverWebAssetsDir(),
});
// v2's connection registry exposes no count-change hook, so forward
// add/remove to the daemon's idle-shutdown handler (a no-op when `idle`
// is undefined, e.g. foreground or --keep-alive).
if (idle !== undefined) {
const registry = v2.connectionRegistry;
const add = registry.add.bind(registry);
const remove = registry.remove.bind(registry);
registry.add = (conn) => {
add(conn);
idle.onConnectionCountChange(registry.size());
};
registry.remove = (connId) => {
remove(connId);
idle.onConnectionCountChange(registry.size());
};
}
logger.info('server-v2 (KIMI_CODE_EXPERIMENTAL_FLAG) is serving the REST/WS API and the bundled web UI');
running = {
address: `http://${v2.host}:${v2.port}`,
logger: logger as unknown as ServerLogger,
close: () => v2.close(),
};
} else {
running = await startServer({
host: options.host,
port: options.port,
logLevel: options.logLevel,
debugEndpoints: options.debugEndpoints,
insecureNoTls: options.insecureNoTls,
allowRemoteShutdown: options.allowRemoteShutdown,
allowRemoteTerminals: options.allowRemoteTerminals,
dangerousBypassAuth: options.dangerousBypassAuth,
allowedHosts: options.allowedHosts,
webAssetsDir: serverWebAssetsDir(),
coreProcessOptions: {
identity: createKimiCodeHostIdentity(version),
telemetry,
},
wsGatewayOptions: {
telemetry,
onConnectionCountChange: idle
? (size) => {
idle.onConnectionCountChange(size);
}
: undefined,
},
});
}
track('server_started', { daemon: mode.daemon });

View file

@ -671,6 +671,29 @@ describe('shared parsers stay strict', () => {
});
});
describe('server-v2 routing (KIMI_CODE_EXPERIMENTAL_FLAG)', () => {
it('is off when the env is unset or blank', async () => {
const { isServerV2Enabled } = await import('#/cli/sub/server/run');
expect(isServerV2Enabled({})).toBe(false);
expect(isServerV2Enabled({ KIMI_CODE_EXPERIMENTAL_FLAG: '' })).toBe(false);
expect(isServerV2Enabled({ KIMI_CODE_EXPERIMENTAL_FLAG: ' ' })).toBe(false);
});
it('is on for the documented truthy values (case-insensitive)', async () => {
const { isServerV2Enabled } = await import('#/cli/sub/server/run');
for (const value of ['1', 'true', 'yes', 'on', 'TRUE', 'Yes', 'ON']) {
expect(isServerV2Enabled({ KIMI_CODE_EXPERIMENTAL_FLAG: value })).toBe(true);
}
});
it('is off for explicit falsey values and arbitrary strings', async () => {
const { isServerV2Enabled } = await import('#/cli/sub/server/run');
for (const value of ['0', 'false', 'no', 'off', '2', 'server-v2']) {
expect(isServerV2Enabled({ KIMI_CODE_EXPERIMENTAL_FLAG: value })).toBe(false);
}
});
});
describe('server web asset directory resolution', () => {
it('uses extracted SEA web assets when available', async () => {
const { resolveServerWebAssetsDir } = await import('#/cli/sub/server/run');

View file

@ -2,6 +2,7 @@ import { resolve } from 'node:path';
import { defineConfig } from 'tsdown';
import { hashImportsPlugin } from '../../build/hash-imports-plugin.mjs';
import { rawTextPlugin } from '../../build/raw-text-plugin.mjs';
import { BUILT_IN_CATALOG_DEFINE, builtInCatalogDefine } from './scripts/built-in-catalog.mjs';
@ -23,7 +24,7 @@ export default defineConfig({
'const __dirname = __cjsShimDirname(__filename);',
].join('\n'),
},
plugins: [rawTextPlugin()],
plugins: [hashImportsPlugin(), rawTextPlugin()],
alias: {
'@': resolve(appRoot, 'src'),
},

View file

@ -4,6 +4,7 @@ import { resolve } from 'node:path';
import { defineConfig } from 'tsdown';
import { hashImportsPlugin } from '../../build/hash-imports-plugin.mjs';
import { rawTextPlugin } from '../../build/raw-text-plugin.mjs';
import { BUILT_IN_CATALOG_DEFINE, builtInCatalogDefine } from './scripts/built-in-catalog.mjs';
@ -42,7 +43,7 @@ export default defineConfig({
platform: 'node',
target: 'node24',
banner: { js: '#!/usr/bin/env node' },
plugins: [rawTextPlugin()],
plugins: [hashImportsPlugin(), rawTextPlugin()],
alias: {
'@': resolve(appRoot, 'src'),
},

View file

@ -0,0 +1,84 @@
import { existsSync, readFileSync } from 'node:fs';
import { dirname, join } from 'node:path';
/**
* Rolldown/tsdown plugin: resolve `#/` subpath imports the way Node's
* package.json `imports` field does scoped to the IMPORTER's owning package,
* honoring array fallbacks such as `"#/*": ["./src/*.ts", "./src/<x>/index.ts"]`.
*
* Why this is needed: when the CLI bundles `@moonshot-ai/server-v2`, rolldown
* inlines `@moonshot-ai/agent-core-v2` source, whose internal `#/foo` imports
* must resolve against each package's own `src/`. Rolldown (like tsx) only
* honors the first array element of an `imports` target and therefore breaks
* on directory-style `#/` imports (e.g. `#/_base/errors` `_base/errors/index.ts`),
* leaving them as bare `require("#/...")` in the bundle. This plugin resolves
* them first. Mirrors `build/hash-imports-loader.mjs` (the Node/tsx loader) and
* the vite `hashImportsPlugin` used by the v2 test configs.
*/
const pkgCache = new Map();
function findPackageJson(importer) {
if (!importer) return undefined;
let dir = dirname(importer.split('?')[0] ?? importer);
for (;;) {
const candidate = join(dir, 'package.json');
if (existsSync(candidate)) return candidate;
const parent = dirname(dir);
if (parent === dir) return undefined;
dir = parent;
}
}
function readPackage(pkgPath) {
let pkg = pkgCache.get(pkgPath);
if (pkg === undefined) {
pkg = JSON.parse(readFileSync(pkgPath, 'utf-8'));
pkgCache.set(pkgPath, pkg);
}
return pkg;
}
function resolveTarget(pkgDir, target, rest) {
const resolved = rest === undefined ? target : target.replace('*', rest);
const full = join(pkgDir, resolved);
return existsSync(full) ? full : undefined;
}
function resolveHashImport(specifier, importer) {
const pkgPath = findPackageJson(importer);
if (pkgPath === undefined) return undefined;
const imports = readPackage(pkgPath).imports;
if (imports === undefined) return undefined;
const pkgDir = dirname(pkgPath);
for (const [key, raw] of Object.entries(imports)) {
if (!key.startsWith('#')) continue;
const targets = Array.isArray(raw) ? raw : [raw];
if (key.endsWith('*')) {
const prefix = key.slice(0, -1);
if (!specifier.startsWith(prefix)) continue;
const rest = specifier.slice(prefix.length);
for (const target of targets) {
const full = resolveTarget(pkgDir, target, rest);
if (full !== undefined) return full;
}
} else if (specifier === key) {
for (const target of targets) {
const full = resolveTarget(pkgDir, target, undefined);
if (full !== undefined) return full;
}
}
}
return undefined;
}
export function hashImportsPlugin() {
return {
name: 'resolve-hash-imports',
resolveId(id, importer) {
if (!id.startsWith('#/')) return null;
return resolveHashImport(id, importer) ?? null;
},
};
}

View file

@ -147,7 +147,7 @@ export class AgentPlanService extends Disposable implements IAgentPlanService {
const path = state.planFilePath ?? this.planFilePathFor(state.id);
let content = '';
try {
content = await this.hostFs.readText(this._planFilePath);
content = await this.hostFs.readText(path);
} catch (error) {
if (!isMissingFileError(error)) throw error;
}

View file

@ -16,6 +16,15 @@ import type { Page } from '#/persistence/interface/queryStore';
export interface SessionSummary {
readonly id: string;
readonly workspaceId: string;
/**
* Absolute working directory frozen at session creation (wire
* `metadata.cwd`). Sourced from the session's own metadata document so it is
* independent of the workspace registry sessions whose workspace was
* unregistered still surface their original cwd (closes gap G3; matches v1's
* `summary.workDir`). Optional only for sessions written before `cwd` was
* persisted; the edge falls back to the workspace registry for those.
*/
readonly cwd?: string;
readonly title?: string;
readonly lastPrompt?: string;
readonly createdAt: number;

View file

@ -53,6 +53,26 @@ function parseTime(value: unknown): number {
return 0;
}
/**
* Recover the session's frozen working directory from its metadata document.
*
* Precedence: v2 `cwd` v1 `workDir` older v1 `custom.cwd`. Returns
* `undefined` only for documents predating every cwd record; the edge falls
* back to the workspace registry for those.
*/
function recoverCwd(meta: Record<string, unknown>): string | undefined {
if (typeof meta['cwd'] === 'string' && meta['cwd'].length > 0) return meta['cwd'];
if (typeof meta['workDir'] === 'string' && meta['workDir'].length > 0) {
return meta['workDir'];
}
const custom = meta['custom'];
if (custom !== null && typeof custom === 'object' && !Array.isArray(custom)) {
const fromCustom = (custom as Record<string, unknown>)['cwd'];
if (typeof fromCustom === 'string' && fromCustom.length > 0) return fromCustom;
}
return undefined;
}
export class FileSessionIndex implements ISessionIndex {
declare readonly _serviceBrand: undefined;
@ -243,6 +263,7 @@ export class FileSessionIndex implements ISessionIndex {
return {
id: sessionId,
workspaceId,
cwd: recoverCwd(meta),
title: typeof meta['title'] === 'string' ? meta['title'] : undefined,
lastPrompt: typeof meta['lastPrompt'] === 'string' ? meta['lastPrompt'] : undefined,
createdAt: parseTime(meta['createdAt']),

View file

@ -2,8 +2,10 @@
* `sessionLegacy` domain (L7 edge adapter) v1-compatible session actions.
*
* Implements the legacy `/api/v1/sessions/{tail}` action contract (`fork` /
* `compact` / `undo` / `abort` / `btw`) and the `/sessions/{id}/children`
* endpoints (`createChild` / `listChildren`) on top of the native v2 services
* `compact` / `undo` / `abort` / `btw`), the `/sessions/{id}/children`
* endpoints (`createChild` / `listChildren`), and `POST /sessions/{id}/profile`
* (`updateProfile` title rename, metadata merge, and the cross-domain
* `agent_config` patch) on top of the native v2 services
* (`ISessionLifecycleService`, `ISessionIndex`, `IAgentRPCService`,
* `IAgentFullCompactionService`, `IAgentPromptService`, ). The native services keep serving
* `/api/v2` and are left untouched; this adapter exists only so clients of the
@ -21,6 +23,7 @@ import type {
SessionStatus,
UndoSessionRequest,
UndoSessionResponse,
UpdateSessionProfileRequest,
} from '@moonshot-ai/protocol';
import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation';
@ -59,6 +62,7 @@ export interface SessionChildrenPage {
export interface ISessionLegacyService {
readonly _serviceBrand: undefined;
updateProfile(sessionId: string, body: UpdateSessionProfileRequest): Promise<SessionWireFields>;
fork(sessionId: string, body: ForkSessionRequest): Promise<SessionWireFields>;
createChild(sessionId: string, body: CreateSessionChildRequest): Promise<SessionWireFields>;
listChildren(sessionId: string, query: SessionChildrenQuery): Promise<SessionChildrenPage>;

View file

@ -16,7 +16,9 @@ import { IAgentContextMemoryService, toProtocolMessage, type ContextMessage } fr
import { IAgentContextSizeService } from '#/agent/contextSize';
import { ErrorCodes, isKimiError, KimiError } from '#/errors';
import { IAgentFullCompactionService } from '#/agent/fullCompaction';
import { IAgentGoalService } from '#/agent/goal';
import { IAgentPermissionModeService } from '#/agent/permissionMode';
import type { PermissionMode } from '#/agent/permissionPolicy';
import { IAgentPlanService } from '#/agent/plan';
import { IAgentProfileService } from '#/agent/profile';
import { IAgentPromptService } from '#/agent/prompt';
@ -38,6 +40,7 @@ import type {
SessionStatusResponse,
UndoSessionRequest,
UndoSessionResponse,
UpdateSessionProfileRequest,
} from '@moonshot-ai/protocol';
import {
@ -71,6 +74,53 @@ export class SessionLegacyService implements ISessionLegacyService {
@IWorkspaceRegistry private readonly workspaceRegistry: IWorkspaceRegistry,
) {}
async updateProfile(
sessionId: string,
body: UpdateSessionProfileRequest,
): Promise<SessionWireFields> {
const session = this.lifecycle.get(sessionId);
if (session === undefined) {
throw new KimiError(ErrorCodes.SESSION_NOT_FOUND, `session ${sessionId} does not exist`);
}
const metadata = session.accessor.get(ISessionMetadata);
if (typeof body.title === 'string') {
await metadata.setTitle(body.title);
}
// v1 `ISessionService.update` writes the wire metadata patch straight into
// `custom` (replace, not deep-merge); `toProtocolSession` then spreads
// `custom` back onto the wire `Session.metadata`. An empty patch is a no-op
// (matches v1's `Object.keys(...).length > 0` guard).
const metadataPatch = body.metadata;
if (metadataPatch !== undefined && Object.keys(metadataPatch).length > 0) {
await metadata.update({ custom: { ...(metadataPatch as Record<string, unknown>) } });
}
const agentConfig = body.agent_config;
if (agentConfig !== undefined) {
const agent = await this.resolveMainAgent(sessionId);
await this.applyAgentConfig(agent, agentConfig);
}
const meta = await metadata.read();
// `ISessionContext` carries the frozen work dir (gap G3 closed), so an
// unregistered workspace does not collapse `cwd` here — matches v1, which
// stores `workDir` on the session itself.
const ctx = session.accessor.get(ISessionContext);
return {
id: meta.id,
workspaceId: ctx.workspaceId,
root: ctx.cwd,
title: meta.title,
lastPrompt: meta.lastPrompt,
createdAt: meta.createdAt,
updatedAt: meta.updatedAt,
archived: meta.archived,
custom: meta.custom,
};
}
async fork(sessionId: string, body: ForkSessionRequest): Promise<SessionWireFields> {
const handle = await this.lifecycle.fork({
sourceSessionId: sessionId,
@ -78,12 +128,11 @@ export class SessionLegacyService implements ISessionLegacyService {
metadata: body.metadata as Record<string, unknown> | undefined,
});
const meta = await handle.accessor.get(ISessionMetadata).read();
const workspaceId = handle.accessor.get(ISessionContext).workspaceId;
const workspace = await this.workspaceRegistry.get(workspaceId);
const ctx = handle.accessor.get(ISessionContext);
return {
id: meta.id,
workspaceId,
root: workspace?.root ?? '',
workspaceId: ctx.workspaceId,
root: ctx.cwd,
title: meta.title,
lastPrompt: meta.lastPrompt,
createdAt: meta.createdAt,
@ -105,12 +154,11 @@ export class SessionLegacyService implements ISessionLegacyService {
},
});
const meta = await handle.accessor.get(ISessionMetadata).read();
const workspaceId = handle.accessor.get(ISessionContext).workspaceId;
const workspace = await this.workspaceRegistry.get(workspaceId);
const ctx = handle.accessor.get(ISessionContext);
return {
id: meta.id,
workspaceId,
root: workspace?.root ?? '',
workspaceId: ctx.workspaceId,
root: ctx.cwd,
title: meta.title,
lastPrompt: meta.lastPrompt,
createdAt: meta.createdAt,
@ -160,9 +208,11 @@ export class SessionLegacyService implements ISessionLegacyService {
);
const page = slice.slice(0, pageSize);
const items = await Promise.all(page.map((s) => this.projectSummary(s)));
// `status` is accepted for wire compatibility but not applied — the v2
// realtime status is not projected into the index summary, and the wire
// projection reports a hardcoded 'idle' (matches `GET /sessions`).
// `status` is layered on at the route edge: this adapter returns
// protocol-free fields, and the route projects the live
// `ISessionActivity.status()` onto each item and filters the page by the
// `status` query (post-page, matching v1). `has_more` reflects the
// pre-filter page.
return { items, has_more: slice.length > pageSize };
}
@ -245,11 +295,14 @@ export class SessionLegacyService implements ISessionLegacyService {
}
private async projectSummary(summary: SessionSummary): Promise<SessionWireFields> {
const workspace = await this.workspaceRegistry.get(summary.workspaceId);
// Prefer the cwd persisted on the session summary (gap G3 closed); fall
// back to the registry only for sessions written before `cwd` was stored.
const root =
summary.cwd ?? (await this.workspaceRegistry.get(summary.workspaceId))?.root ?? '';
return {
id: summary.id,
workspaceId: summary.workspaceId,
root: workspace?.root ?? '',
root,
title: summary.title,
lastPrompt: summary.lastPrompt,
createdAt: summary.createdAt,
@ -264,6 +317,68 @@ export class SessionLegacyService implements ISessionLegacyService {
* `resumeSession`; delegates to the `agentLifecycle` domain's
* `ensureMainAgent` bootstrap helper).
*/
/**
* Apply the v1 `agent_config` patch onto the main agent. Mirrors v1's
* `IPromptService.applyAgentState` (`promptService.ts:650-743`) in both order
* (model thinking permission plan swarm goal) and diff behaviour:
* the non-idempotent `plan.enter` / `swarm.enter` are guarded behind a state
* read so a repeated `true` does not throw ('Already in plan mode'); the
* idempotent setters (model / thinking / permission) fire directly. Goal
* actions are one-shot and let domain errors (`goal.*`) propagate to the
* route's `sendMappedError`.
*/
private async applyAgentConfig(
agent: IAgentScopeHandle,
agentConfig: NonNullable<UpdateSessionProfileRequest['agent_config']>,
): Promise<void> {
const profile = agent.accessor.get(IAgentProfileService);
if (agentConfig.model !== undefined && agentConfig.model !== '') {
await profile.setModel(agentConfig.model);
}
if (agentConfig.thinking !== undefined) {
profile.setThinking(agentConfig.thinking);
}
if (agentConfig.permission_mode !== undefined) {
agent
.accessor.get(IAgentPermissionModeService)
.setMode(agentConfig.permission_mode as PermissionMode);
}
if (agentConfig.plan_mode !== undefined) {
const plan = agent.accessor.get(IAgentPlanService);
const active = (await plan.status()) !== null;
if (active !== agentConfig.plan_mode) {
if (agentConfig.plan_mode) await plan.enter();
else plan.exit();
}
}
if (agentConfig.swarm_mode !== undefined) {
const swarm = agent.accessor.get(IAgentSwarmService);
if (swarm.isActive !== agentConfig.swarm_mode) {
if (agentConfig.swarm_mode) swarm.enter('manual');
else swarm.exit();
}
}
if (agentConfig.goal_objective !== undefined) {
await agent
.accessor.get(IAgentGoalService)
.createGoal({ objective: agentConfig.goal_objective });
}
if (agentConfig.goal_control !== undefined) {
const goal = agent.accessor.get(IAgentGoalService);
switch (agentConfig.goal_control) {
case 'pause':
await goal.pauseGoal({});
break;
case 'resume':
await goal.resumeGoal({});
break;
case 'cancel':
await goal.cancelGoal({});
break;
}
}
}
private async resolveMainAgent(sessionId: string): Promise<IAgentScopeHandle> {
const session = this.lifecycle.get(sessionId);
if (session === undefined) {

View file

@ -47,6 +47,15 @@ export interface SessionMeta {
readonly createdAt: number;
readonly updatedAt: number;
readonly archived: boolean;
/**
* Absolute working directory frozen at session creation (`metadata.cwd` on
* the wire). Persisted so the session read model (`sessionIndex`) can surface
* it without reverse-resolving the workspace registry a session whose
* workspace was unregistered keeps its original cwd (closes gap G3). Mirrors
* v1, which stores `workDir` on the session. Optional only for documents
* predating this field; `load()` always writes it for new sessions.
*/
readonly cwd?: string;
readonly forkedFrom?: string;
/** Registry of agents belonging to this session, keyed by agent id. */
readonly agents?: Readonly<Record<string, AgentMeta>>;

View file

@ -98,6 +98,7 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
await this.queryStore.put(SESSION_COLLECTION, this.ctx.sessionId, {
id: this.data.id,
workspaceId: this.ctx.workspaceId,
cwd: this.ctx.cwd,
title: this.data.title,
lastPrompt: this.data.lastPrompt,
createdAt: this.data.createdAt,
@ -123,6 +124,7 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
this.data = {
id: this.ctx.sessionId,
version: SESSION_META_VERSION,
cwd: this.ctx.cwd,
createdAt: now,
updatedAt: now,
archived: false,
@ -143,20 +145,34 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
* left untouched until an explicit write, so a read-only snapshot of a v1
* session does not migrate it.
*/
function normalizeSessionMeta(raw: SessionMeta, sessionId: string): SessionMeta {
if (raw.version === SESSION_META_VERSION) return raw;
const legacy = raw as unknown as { createdAt?: unknown; updatedAt?: unknown };
export function normalizeSessionMeta(raw: SessionMeta, sessionId: string): SessionMeta {
const legacy = raw as unknown as {
createdAt?: unknown;
updatedAt?: unknown;
workDir?: unknown;
};
// Backfill `cwd` for legacy v1 documents, which store the working directory
// as `workDir` (older v1 sessions used `custom.cwd`). New v2 documents already
// carry `cwd` and pass through unchanged.
const cwd =
raw.cwd ?? (typeof legacy.workDir === 'string' && legacy.workDir.length > 0
? legacy.workDir
: undefined);
if (raw.version === SESSION_META_VERSION) {
return cwd === raw.cwd ? raw : { ...raw, cwd };
}
return {
...raw,
id: sessionId,
version: SESSION_META_VERSION,
cwd,
createdAt: toEpochMs(legacy.createdAt),
updatedAt: toEpochMs(legacy.updatedAt),
};
}
/** Coerce a persisted timestamp (v2 epoch-ms number or v1 ISO string) to epoch ms. */
function toEpochMs(value: unknown): number {
export function toEpochMs(value: unknown): number {
if (typeof value === 'number' && Number.isFinite(value)) return value;
if (typeof value === 'string') {
const parsed = Date.parse(value);

View file

@ -105,6 +105,19 @@ describe('FileSessionIndex (legacy)', () => {
expect(await store.get('missing')).toBeUndefined();
});
it('recovers cwd from the metadata document (v2 cwd, v1 workDir, custom.cwd)', async () => {
await seedSession('v2', { cwd: '/repo/v2' });
await seedSession('v1', { workDir: '/repo/v1' });
await seedSession('old', { custom: { cwd: '/repo/old' } });
await seedSession('none', { title: 'no cwd' });
const store = build();
expect((await store.get('v2'))?.cwd).toBe('/repo/v2');
expect((await store.get('v1'))?.cwd).toBe('/repo/v1');
expect((await store.get('old'))?.cwd).toBe('/repo/old');
expect((await store.get('none'))?.cwd).toBeUndefined();
});
it('list filters by sessionId without enumerating all sessions', async () => {
await seedSession('active', { title: 'hello' });
await seedSession('archived', { archived: true });

View file

@ -3,7 +3,8 @@
* DI engine) in the foreground and blocks until SIGINT/SIGTERM.
*
* Flags: `--port <n>` (default 58627), `--host <h>` (default 127.0.0.1),
* `--log-level <level>` (default info).
* `--log-level <level>` (default info), `--dangerous-bypass-auth` (disable
* bearer-token auth on every route trusted networks only).
*/
import type { ServerLogLevel } from './services/pinoLoggerService';
@ -13,6 +14,7 @@ interface CliOptions {
readonly host?: string;
readonly port?: number;
readonly logLevel?: ServerLogLevel;
readonly disableAuth?: boolean;
}
const LOG_LEVELS: readonly ServerLogLevel[] = [
@ -29,6 +31,7 @@ function parseArgs(argv: readonly string[]): CliOptions {
let host: string | undefined;
let port: number | undefined;
let logLevel: ServerLogLevel | undefined;
let disableAuth: boolean | undefined;
for (let i = 0; i < argv.length; i++) {
const arg = argv[i];
@ -47,10 +50,12 @@ function parseArgs(argv: readonly string[]): CliOptions {
) {
logLevel = next as ServerLogLevel;
i++;
} else if (arg === '--dangerous-bypass-auth') {
disableAuth = true;
}
}
return { host, port, logLevel };
return { host, port, logLevel, disableAuth };
}
async function main(): Promise<never> {
@ -59,6 +64,7 @@ async function main(): Promise<never> {
host: opts.host,
port: opts.port,
logLevel: opts.logLevel ?? 'info',
disableAuth: opts.disableAuth,
});
const origin = `http://${server.host}:${server.port}`;

View file

@ -35,6 +35,11 @@ export interface MetaRouteOptions {
readonly serverVersion: string;
readonly serverId: string;
readonly startedAt: string;
/**
* Whether the server was started with `--dangerous-bypass-auth`. Surfaced so
* the web UI can skip the token prompt and connect without a credential.
*/
readonly dangerousBypassAuth: boolean;
}
export function registerMetaRoute(app: RouteHost, opts: MetaRouteOptions): void {
@ -51,6 +56,7 @@ export function registerMetaRoute(app: RouteHost, opts: MetaRouteOptions): void
server_id: opts.serverId,
started_at: opts.startedAt,
open_in_apps: [],
dangerous_bypass_auth: opts.dangerousBypassAuth,
});
const route = defineRoute(

View file

@ -24,6 +24,7 @@ import { registerFsRoutes } from './fs';
import { registerGuiStoreRoutes } from './guiStore';
import { registerMessagesRoutes } from './messages';
import type { IGuiStoreService } from '../services/guiStore/guiStore';
import type { ISnapshotReader } from '../services/snapshot';
import { registerMetaRoute } from './meta';
import { registerModelCatalogRoutes } from './modelCatalog';
import { registerOAuthRoutes } from './oauth';
@ -63,6 +64,13 @@ export interface RegisterApiV1RoutesOptions {
readonly onShutdown: () => void;
readonly connectionRegistry: IConnectionRegistry;
readonly broadcaster: SessionEventBroadcaster;
readonly snapshotReader: ISnapshotReader;
/**
* Surface `dangerous_bypass_auth` in the `/meta` payload. Set by `start.ts`
* from the `disableAuth` server option (the `--dangerous-bypass-auth` CLI
* flag).
*/
readonly dangerousBypassAuth?: boolean;
}
export async function registerApiV1Routes(
@ -78,6 +86,7 @@ export async function registerApiV1Routes(
serverVersion: opts.serverVersion,
serverId: ulid(),
startedAt: new Date().toISOString(),
dangerousBypassAuth: opts.dangerousBypassAuth === true,
});
registerAuthRoute(apiV1 as unknown as Parameters<typeof registerAuthRoute>[0], core);
@ -134,6 +143,7 @@ export async function registerApiV1Routes(
registerSnapshotRoutes(apiV1 as unknown as Parameters<typeof registerSnapshotRoutes>[0], {
core,
broadcaster: opts.broadcaster,
reader: opts.snapshotReader,
});
if (opts.enableShutdown !== false) {
registerShutdownRoutes(apiV1 as unknown as Parameters<typeof registerShutdownRoutes>[0], {

View file

@ -7,13 +7,13 @@
* GET /sessions list
* GET /sessions/{session_id} get
* GET /sessions/{session_id}/profile
* POST /sessions/{session_id}/profile update title (partial)
* POST /sessions/{session_id}/profile update title / metadata / agent_config
* POST /sessions/{tail} action: fork / compact / undo /
* abort / btw / archive
* GET /sessions/{session_id}/children list child sessions
* POST /sessions/{session_id}/children create child session (fork+tag)
* GET /sessions/{session_id}/status best-effort
* GET /sessions/{session_id}/warnings empty (no warning sources ported)
* GET /sessions/{session_id}/warnings agents-md-oversized notice
*
* The `POST /sessions/{tail}` actions (`fork` / `compact` / `undo` / `abort` /
* `btw` / `archive`) and the `/sessions/{id}/children` endpoints are dispatched
@ -22,27 +22,42 @@
* `create`, `fork`, and child creation publish `event.session.created` on the
* core event bus, matching v1.
*
* `GET /sessions/{id}/warnings` returns `{ warnings: [] }`: the only v1 warning
* (`agents-md-oversized`) is computed by `prepareSystemPromptContext`, which is
* not ported to v2 yet. This is within v1's observable behaviour it falls back
* to `[]` whenever the underlying computation throws.
* `GET /sessions/{id}/warnings` surfaces the only v1 warning
* (`agents-md-oversized`) by projecting the main agent's
* `IAgentProfileService.getAgentsMdWarning()` computed and cached when the
* agent binds a profile (via `prepareSystemPromptContext`) into the v1
* `{ code, message, severity }` wire shape. An unbound main agent yields an
* empty list, matching v1's "no warning" case.
*
* **Wire fidelity**: mirrors v1's `toProtocolSession`
* (`packages/agent-core/src/services/session/session.ts`), which populates
* only the index/metadata fields and returns placeholders for the heavy ones
* (`agent_config:{model:''}`, `usage:zeros`, `permission_rules:[]`,
* `message_count:0`, `last_seq:0`, hardcoded `status:'idle'`). v2 produces the
* same placeholder shape from `ISessionIndex` + `IWorkspaceRegistry`, and now
* `message_count:0`, `last_seq:0`). v2 produces the same placeholder shape
* from `ISessionIndex` (with `cwd` persisted on the session itself), and now
* also surfaces `last_prompt` and the merged custom `metadata`.
*
* **cwd resolution (gap G3)**: v2 does not store the original work dir on the
* session; we recover `metadata.cwd` from `IWorkspaceRegistry`
* (`workspaceId → root`). Sessions whose workspace is not registered cannot be
* represented and are filtered from list / 404 on get.
* **Status**: v1's `SessionService` overwrites the placeholder `status` with the
* live value before projecting (`_patchSessionStatus`). v2 does the same:
* `toWireSession` takes the live `ISessionActivity.status()` resolved from the
* session's scope (or `'idle'` when the session is cold), so the wire `status`
* is real on every session-producing endpoint here. `GET /sessions` and
* `GET /sessions/{id}/children` filter their projected page by the `status`
* query param (post-page, matching v1 `has_more` reflects the pre-filter
* page). The `aborted` phase is not derived yet (gap G10); v2 never reports it.
*
* **cwd resolution (gap G3 closed)**: the session's frozen work dir is
* persisted on its metadata document (`ISessionMetadata`) and surfaced on the
* `ISessionIndex` summary, so `metadata.cwd` comes from the session itself
* not from `IWorkspaceRegistry`. Sessions whose workspace was unregistered keep
* their original cwd and stay listed / gettable (matching v1, which stores
* `workDir` on the session). `IWorkspaceRegistry` is consulted only as a
* back-compat fallback for sessions written before `cwd` was persisted.
*/
import {
ErrorCodes,
IAgentProfileService,
IAuthSummaryService,
ISessionBtwService,
ISessionActivity,
@ -77,13 +92,15 @@ import {
undoSessionRequestSchema,
undoSessionResponseSchema,
updateSessionProfileRequestSchema,
workspaceIdSchema,
} from '@moonshot-ai/protocol';
import type { Session } from '@moonshot-ai/protocol';
import type { Session, SessionStatus } from '@moonshot-ai/protocol';
import { ulid } from 'ulid';
import { z } from 'zod';
import { errEnvelope, okEnvelope } from '../envelope';
import { defineRoute } from '../middleware/defineRoute';
import { ensureMainAgent } from '../transport/mainAgent';
import { parseActionSuffix } from './action-suffix';
interface SessionRouteHost {
@ -111,11 +128,13 @@ const booleanQueryParam = z.preprocess((value) => {
return value;
}, z.boolean().optional());
// NOTE: `status` filtering is accepted for wire compatibility but not applied
// (gap G5). `before_id`/`after_id` id-cursors and `page_size` ARE applied in
// the route handler (the `FileSessionIndex` does not implement `cursor`, so we
// page over its recency-sorted result). `include_archive` → `includeArchived`;
// `workspace_id` → `workspaceId`; `exclude_empty` drops sessions with no prompt.
// NOTE: mirrors v1's `GET /sessions` query. `before_id`/`after_id` id-cursors
// and `page_size` ARE applied in the route handler (the `FileSessionIndex` does
// not implement `cursor`, so we page over its recency-sorted result); `status`
// filters the projected page (post-page, matching v1). `include_archive` →
// `includeArchived`; `archived_only` forces `includeArchived` and then keeps
// only archived sessions; `workspace_id` → `workspaceId`; `exclude_empty` drops
// sessions with no prompt.
const sessionsListQueryCoercion = z
.object({
before_id: z.string().min(1).optional(),
@ -124,7 +143,8 @@ const sessionsListQueryCoercion = z
status: sessionStatusSchema.optional(),
include_archive: booleanQueryParam,
exclude_empty: booleanQueryParam,
workspace_id: z.string().min(1).optional(),
archived_only: booleanQueryParam,
workspace_id: workspaceIdSchema.optional(),
})
.superRefine((value, ctx) => {
if (value.before_id !== undefined && value.after_id !== undefined) {
@ -135,15 +155,24 @@ const sessionsListQueryCoercion = z
params: { code: ErrorCode.VALIDATION_FAILED },
});
}
if (value.archived_only === true && value.include_archive === true) {
ctx.addIssue({
code: 'custom',
message: 'archived_only and include_archive are mutually exclusive',
path: ['archived_only'],
params: { code: ErrorCode.VALIDATION_FAILED },
});
}
});
const sessionIdParamSchema = z.object({
session_id: z.string().min(1),
});
// Mirrors v1's children query: id-cursors + page_size + status. `status` is
// accepted for wire compatibility but not applied by `ISessionLegacyService`
// (the wire projection reports a hardcoded 'idle'; see gap G10 / G5).
// Mirrors v1's children query: id-cursors + page_size + status. The route
// projects the live `ISessionActivity.status()` onto each child and filters the
// page by `status` (post-page, matching v1); `ISessionLegacyService` stays
// protocol-free and does not apply the filter itself.
const sessionChildrenListQueryCoercion = z
.object({
before_id: z.string().min(1).optional(),
@ -257,7 +286,11 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
await handle.accessor.get(ISessionMetadata).setTitle(body.title);
}
const meta = await handle.accessor.get(ISessionMetadata).read();
const session = toWireSession({ ...meta, workspaceId: touched.id }, touched.root);
const session = toWireSession(
{ ...meta, workspaceId: touched.id },
touched.root,
handle.accessor.get(ISessionActivity).status(),
);
core.accessor.get(IEventService).publish({
type: 'event.session.created',
payload: { agentId: 'main', sessionId: session.id, session },
@ -279,6 +312,7 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
success: { data: pageResponseSchema(sessionSchema) },
errors: {
[ErrorCode.VALIDATION_FAILED]: { detailsSchema },
[ErrorCode.WORKSPACE_NOT_FOUND]: {},
},
description: 'List sessions',
tags: ['sessions'],
@ -286,26 +320,47 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
async (req, reply) => {
const raw = req.query;
const pageSize = raw.page_size;
// `FileSessionIndex` does not implement `cursor` (gap G5 closed here), so
// we fetch the full recency-sorted set (no `limit`) and apply the id
// cursor in this handler. `list()` already orders by `updatedAt` desc and
// filters by workspace / archived.
const page = await core.accessor.get(ISessionIndex).list({
workspaceId: raw.workspace_id,
includeArchived: raw.include_archive,
});
const archivedOnly = raw.archived_only === true;
const workspaces = await core.accessor.get(IWorkspaceRegistry).list();
const roots = new Map(workspaces.map((w) => [w.id, w.root]));
// v1 resolves `workspace_id` to its root and 40410s when it is unknown;
// the index filters by `workspaceId` directly, so only the existence
// check is needed here (the root itself is not used by the query).
if (raw.workspace_id !== undefined && !roots.has(raw.workspace_id)) {
reply.send(
errEnvelope(
ErrorCode.WORKSPACE_NOT_FOUND,
`workspace ${raw.workspace_id} does not exist`,
req.id,
),
);
return;
}
// `FileSessionIndex` does not implement `cursor` (gap G5 closed here), so
// we fetch the full recency-sorted set (no `limit`) and apply the id
// cursor in this handler. `list()` already orders by `updatedAt` desc and
// filters by workspace / archived. `archived_only` forces archived rows
// into the set, then the filter below keeps only them.
const page = await core.accessor.get(ISessionIndex).list({
workspaceId: raw.workspace_id,
includeArchived: archivedOnly ? true : raw.include_archive,
});
// Filter down to the sequence the client actually sees BEFORE computing
// the cursor position and the page boundary, so a cursor carried over
// from a previous page always resolves to the same index.
// from a previous page always resolves to the same index. `cwd` is read
// from the session's own summary first (gap G3 closed — an unregistered
// workspace no longer drops the session); the registry `roots` map is
// only a back-compat fallback for sessions written before `cwd` was
// persisted. A session with no recoverable cwd is still skipped.
const eligible: { summary: (typeof page.items)[number]; cwd: string }[] = [];
for (const summary of page.items) {
const cwd = roots.get(summary.workspaceId);
if (cwd === undefined) continue; // gap G3: cannot represent cwd
const cwd = summary.cwd ?? roots.get(summary.workspaceId);
if (cwd === undefined) continue;
if (archivedOnly && summary.archived !== true) continue;
if (raw.exclude_empty === true && (summary.lastPrompt ?? '').length === 0) continue;
eligible.push({ summary, cwd });
}
@ -330,9 +385,18 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
const window = eligible.slice(start, end);
const limit = pageSize ?? window.length;
const hasMore = window.length > limit;
const items: Session[] = window
const projected: Session[] = window
.slice(0, limit)
.map(({ summary, cwd }) => toWireSession(summary, cwd));
.map(({ summary, cwd }) =>
toWireSession(summary, cwd, resolveSessionStatus(core, summary.id)),
);
// v1 filters the projected page by `status` (post-page); `has_more` keeps
// reflecting the pre-filter page, so a filtered page may be short or empty
// while `has_more` is still true — match that exactly.
const items =
raw.status !== undefined
? projected.filter((session) => session.status === raw.status)
: projected;
reply.send(okEnvelope({ items, has_more: hasMore }, req.id));
},
);
@ -364,20 +428,24 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
);
return;
}
const workspace = await core.accessor.get(IWorkspaceRegistry).get(summary.workspaceId);
if (workspace === undefined) {
// gap G3: persisted session whose workspace is not registered → cwd
// cannot be recovered.
const cwd =
summary.cwd ??
(await core.accessor.get(IWorkspaceRegistry).get(summary.workspaceId))?.root;
if (cwd === undefined) {
// Persisted session with no `cwd` on disk and no registered workspace
// to fall back to (predates gap-G3 persistence) — cannot project cwd.
reply.send(
errEnvelope(
ErrorCode.SESSION_NOT_FOUND,
`session ${session_id} workspace missing`,
`session ${session_id} has no recoverable cwd`,
req.id,
),
);
return;
}
reply.send(okEnvelope(toWireSession(summary, workspace.root), req.id));
reply.send(
okEnvelope(toWireSession(summary, cwd, resolveSessionStatus(core, session_id)), req.id),
);
},
);
app.get(
@ -408,18 +476,22 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
);
return;
}
const workspace = await core.accessor.get(IWorkspaceRegistry).get(summary.workspaceId);
if (workspace === undefined) {
const cwd =
summary.cwd ??
(await core.accessor.get(IWorkspaceRegistry).get(summary.workspaceId))?.root;
if (cwd === undefined) {
reply.send(
errEnvelope(
ErrorCode.SESSION_NOT_FOUND,
`session ${session_id} workspace missing`,
`session ${session_id} has no recoverable cwd`,
req.id,
),
);
return;
}
reply.send(okEnvelope(toWireSession(summary, workspace.root), req.id));
reply.send(
okEnvelope(toWireSession(summary, cwd, resolveSessionStatus(core, session_id)), req.id),
);
},
);
app.get(
@ -439,36 +511,34 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
[ErrorCode.VALIDATION_FAILED]: { detailsSchema },
[ErrorCode.SESSION_NOT_FOUND]: {},
},
description: 'Update session profile (title only in this slice)',
description: 'Update session profile (title, metadata, agent_config)',
tags: ['sessions'],
},
async (req, reply) => {
const { session_id } = req.params;
const handle = core.accessor.get(ISessionLifecycleService).get(session_id);
if (handle === undefined) {
reply.send(
errEnvelope(ErrorCode.SESSION_NOT_FOUND, `session ${session_id} does not exist`, req.id),
);
return;
try {
const { session_id } = req.params;
const fields = await core
.accessor.get(ISessionLegacyService)
.updateProfile(session_id, req.body);
const session = toWireSession(fields, fields.root, resolveSessionStatus(core, fields.id));
// Broadcast the title change to every connection (including clients not
// subscribed to this session, and covering inactive sessions), so session
// lists stay in sync — mirrors v1's `session.meta.updated` publish.
if (typeof req.body.title === 'string' && req.body.title.trim().length > 0) {
core.accessor.get(IEventService).publish({
type: 'session.meta.updated',
payload: {
agentId: 'main',
sessionId: session_id,
title: session.title,
patch: { title: session.title, isCustomTitle: true },
},
});
}
reply.send(okEnvelope(session, req.id));
} catch (error) {
sendMappedError(reply, req.id, error);
}
const title = req.body.title;
if (typeof title === 'string') {
await handle.accessor.get(ISessionMetadata).setTitle(title);
}
const meta = await handle.accessor.get(ISessionMetadata).read();
const workspaceId = handle.accessor.get(ISessionContext).workspaceId;
const workspace = await core.accessor.get(IWorkspaceRegistry).get(workspaceId);
if (workspace === undefined) {
reply.send(
errEnvelope(
ErrorCode.SESSION_NOT_FOUND,
`session ${session_id} workspace missing`,
req.id,
),
);
return;
}
reply.send(okEnvelope(toWireSession({ ...meta, workspaceId }, workspace.root), req.id));
},
);
app.post(
@ -523,7 +593,7 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
if (parsed.action === 'fork') {
const body = forkSessionRequestSchema.parse(req.body);
const fields = await legacy.fork(parsed.id, body);
const session = toWireSession(fields, fields.root);
const session = toWireSession(fields, fields.root, resolveSessionStatus(core, fields.id));
core.accessor.get(IEventService).publish({
type: 'event.session.created',
payload: { agentId: 'main', sessionId: session.id, session },
@ -597,16 +667,19 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
async (req, reply) => {
try {
const { session_id } = req.params;
const page = await core.accessor.get(ISessionLegacyService).listChildren(session_id, req.query);
reply.send(
okEnvelope(
{
items: page.items.map((fields) => toWireSession(fields, fields.root)),
has_more: page.has_more,
},
req.id,
),
const page = await core
.accessor.get(ISessionLegacyService)
.listChildren(session_id, req.query);
const projected = page.items.map((fields) =>
toWireSession(fields, fields.root, resolveSessionStatus(core, fields.id)),
);
// v1 filters the projected page by `status` (post-page); `has_more`
// reflects the pre-filter page.
const items =
req.query.status !== undefined
? projected.filter((session) => session.status === req.query.status)
: projected;
reply.send(okEnvelope({ items, has_more: page.has_more }, req.id));
} catch (error) {
sendMappedError(reply, req.id, error);
}
@ -639,7 +712,7 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
const fields = await core
.accessor.get(ISessionLegacyService)
.createChild(session_id, req.body);
const session = toWireSession(fields, fields.root);
const session = toWireSession(fields, fields.root, resolveSessionStatus(core, fields.id));
core.accessor.get(IEventService).publish({
type: 'event.session.created',
payload: { agentId: 'main', sessionId: session.id, session },
@ -720,18 +793,34 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
},
async (req, reply) => {
const { session_id } = req.params;
const summary = await core.accessor.get(ISessionIndex).get(session_id);
if (summary === undefined) {
const session = core.accessor.get(ISessionLifecycleService).get(session_id);
if (session === undefined) {
reply.send(
errEnvelope(ErrorCode.SESSION_NOT_FOUND, `session ${session_id} does not exist`, req.id),
);
return;
}
// No warning sources are ported to v2 yet (the v1 `agents-md-oversized`
// detection lives in `prepareSystemPromptContext`, which is not wired
// here). Return an empty list — within v1's own observable behaviour,
// since it falls back to `[]` whenever the underlying computation throws.
reply.send(okEnvelope({ warnings: [] }, req.id));
try {
// Surface the v2 `agents-md-oversized` notice in the v1 wire shape. The
// warning is computed (and cached) by `IAgentProfileService` when the main
// agent binds a profile; an unbound main agent yields `undefined` → `[]`,
// matching v1's "no warning" case.
const agent = await ensureMainAgent(session);
const agentsMdWarning = agent.accessor.get(IAgentProfileService).getAgentsMdWarning();
const warnings =
agentsMdWarning === undefined
? []
: [
{
code: 'agents-md-oversized',
message: agentsMdWarning,
severity: 'warning' as const,
},
];
reply.send(okEnvelope({ warnings }, req.id));
} catch (error) {
sendMappedError(reply, req.id, error);
}
},
);
app.get(
@ -758,14 +847,18 @@ export interface SessionWireFields {
readonly custom?: Record<string, unknown>;
}
export function toWireSession(fields: SessionWireFields, cwd: string): Session {
export function toWireSession(
fields: SessionWireFields,
cwd: string,
status: SessionStatus,
): Session {
return {
id: fields.id,
workspace_id: fields.workspaceId,
title: fields.title ?? '',
created_at: new Date(fields.createdAt).toISOString(),
updated_at: new Date(fields.updatedAt).toISOString(),
status: 'idle',
status,
archived: fields.archived,
last_prompt: fields.lastPrompt,
metadata: buildWireMetadata(fields.custom, cwd),
@ -777,6 +870,19 @@ export function toWireSession(fields: SessionWireFields, cwd: string): Session {
};
}
/**
* Resolve a session's live wire `status`. Mirrors v1's
* `SessionService._patchSessionStatus`: the live `ISessionActivity.status()`
* wins, and a cold session (no live handle) reports `'idle'` it carries no
* pending approval/question and no active turn. The `aborted` phase is not
* derived in v2 yet (gap G10), so it is never returned here.
*/
function resolveSessionStatus(core: Scope, sessionId: string): SessionStatus {
const handle = core.accessor.get(ISessionLifecycleService).get(sessionId);
if (handle === undefined) return 'idle';
return handle.accessor.get(ISessionActivity).status();
}
/**
* Build the wire `Session.metadata`: caller-supplied custom fields (minus the
* reserved `goal` key, matching v1's `toProtocolSession`) overlaid with the
@ -837,6 +943,24 @@ function sendMappedError(
case 'session.undo_unavailable':
reply.send(errEnvelope(ErrorCode.SESSION_UNDO_UNAVAILABLE, err.message, requestId));
return;
case ErrorCodes.GOAL_ALREADY_EXISTS:
reply.send(errEnvelope(ErrorCode.GOAL_ALREADY_EXISTS, err.message, requestId));
return;
case ErrorCodes.GOAL_NOT_FOUND:
reply.send(errEnvelope(ErrorCode.GOAL_NOT_FOUND, err.message, requestId));
return;
case ErrorCodes.GOAL_STATUS_INVALID:
reply.send(errEnvelope(ErrorCode.GOAL_STATUS_INVALID, err.message, requestId));
return;
case ErrorCodes.GOAL_NOT_RESUMABLE:
reply.send(errEnvelope(ErrorCode.GOAL_NOT_RESUMABLE, err.message, requestId));
return;
case ErrorCodes.GOAL_OBJECTIVE_EMPTY:
reply.send(errEnvelope(ErrorCode.GOAL_OBJECTIVE_EMPTY, err.message, requestId));
return;
case ErrorCodes.GOAL_OBJECTIVE_TOO_LONG:
reply.send(errEnvelope(ErrorCode.GOAL_OBJECTIVE_TOO_LONG, err.message, requestId));
return;
case 'request.invalid':
case 'validation.failed':
reply.send(errEnvelope(ErrorCode.VALIDATION_FAILED, err.message, requestId));

View file

@ -1,28 +1,47 @@
/**
* `/sessions/{session_id}/skills*` REST routes server-v2 port.
* `/skills` REST routes (session- and workspace-scoped) server-v2 port.
*
* Mirrors the v1 server's wire contract
* (`packages/server/src/routes/skills.ts`) path-for-path and schema-for-schema:
*
* GET /sessions/{session_id}/skills data: {skills: SkillDescriptor[]}
* GET /workspaces/{workspace_id}/skills data: {skills: SkillDescriptor[]}
* POST /sessions/{session_id}/skills/{skill_name}:activate body: {args?} data: {activated: true, skill_name}
*
* **Activation gate**: by convention these endpoints are only valid for an
* *activated* session one that is live in `ISessionLifecycleService`. When
* The session list is session-scoped: the catalog is built per session
* (project skills are discovered from the session cwd), so it lives under
* `/sessions/{session_id}` rather than as a global collection like `/tools`.
*
* The workspace list (`/workspaces/{workspace_id}/skills`) is the session-less
* counterpart: it scans the same roots a new session in that workspace cwd
* would, so clients can populate the composer skill menu before a session
* exists. The workspace id is resolved to its root via
* `IWorkspaceRegistry.get` (`40410` when unknown); the root is then scanned by
* composing the same four sources the per-session catalog merges builtin /
* user / project(workDir) / plugin through the shared `ISkillDiscovery`,
* `skillRoots` and `InMemorySkillCatalog` primitives, so the result matches the
* session listing for the same cwd. The composition is intentionally edge-side:
* `InMemorySkillCatalog` is not a scoped service and the `skillRoots` helpers
* are exported for exactly this purpose.
*
* **Activation gate**: by convention the session endpoints are only valid for
* an *activated* session one that is live in `ISessionLifecycleService`. When
* the session is not in the live map we still answer `40401 session.not_found`
* (the only session error code on the v1 wire contract), but we enrich the
* message:
* - persisted in `ISessionIndex` but not live `"... is not activated, you need to activate it first"`;
* - not in the index at all `"... does not exist"`.
*
* **Scope split**: v1 resolves a single `ISkillService` for both verbs. v2
* splits the domain, so the route borrows two scoped services:
* - list `ISessionSkillCatalog` (Session scope) `catalog.listSkills()`.
* - activate `IAgentSkillService` (Agent scope, on the `main` agent)
* renders the skill prompt and starts a turn with a
* `skill_activation` origin. The returned `Turn` handle is
* discarded; clients follow progress via the `skill.activated`
* + `turn.*` events emitted by the service on the WS stream.
* **Scope split**: v1 resolves a single `ISkillService` for every verb. v2
* splits the domain, so the route borrows different scoped services per verb:
* - session list `ISessionSkillCatalog` (Session scope) `catalog.listSkills()`.
* - workspace list no session: resolves `IWorkspaceRegistry` (App scope)
* for the root, then composes the skill scan at the edge (see above).
* - activate `IAgentSkillService` (Agent scope, on the `main` agent)
* renders the skill prompt and starts a turn with a
* `skill_activation` origin. The returned `Turn` handle is
* discarded; clients follow progress via the `skill.activated`
* + `turn.*` events emitted by the service on the WS stream.
*
* **Model projection**: `SkillDefinition` (v2) protocol `SkillDescriptor`,
* byte-for-byte with v1's `toProtocolSkill`
@ -32,34 +51,45 @@
* dropped.
*
* **Error mapping**:
* - not live / unknown session envelope `code: 40401 session.not_found` (see gate above).
* - unknown workspace id envelope `code: 40410 workspace.not_found`.
* - not live / unknown session envelope `code: 40401 session.not_found` (see gate above).
* - `skill.not_found` / `skill.name_empty` envelope `code: 40415 skill.not_found`.
* - `skill.type_unsupported` envelope `code: 40912 skill.not_activatable`.
* - `skill.type_unsupported` envelope `code: 40912 skill.not_activatable`.
* - malformed `{tail}` (bad action, bare) envelope `code: 40001 validation.failed`.
* - other errors 50001 via the global `installErrorHandler`.
*
* **Action suffix**: the `:activate` POST endpoint uses the shared
* `parseActionSuffix` helper (no bare form `:activate` is the only action).
*
* **Anti-corruption**: route resolves `ISessionSkillCatalog` / `IAgentSkillService`
* via the accessor; no SDK imports.
* **Anti-corruption**: route resolves every service via the accessor; no SDK
* imports.
*/
import {
BUILTIN_SKILLS,
ErrorCodes,
IAgentSkillService,
IBootstrapService,
IPluginService,
ISessionIndex,
ISessionLifecycleService,
ISessionSkillCatalog,
ISkillDiscovery,
IWorkspaceRegistry,
InMemorySkillCatalog,
isKimiError,
projectRoots,
userRoots,
type ISessionScopeHandle,
type Scope,
type SkillDefinition,
} from '@moonshot-ai/agent-core-v2';
import {
ErrorCode,
activateSkillRequestSchema,
activateSkillResultSchema,
listSkillsResponseSchema,
workspaceIdParamSchema,
type SkillDescriptor,
} from '@moonshot-ai/protocol';
import { z } from 'zod';
@ -156,6 +186,43 @@ export function registerSkillsRoutes(app: SkillsRouteHost, core: Scope): void {
listSkillsRoute.handler as Parameters<SkillsRouteHost['get']>[2],
);
// GET /workspaces/{workspace_id}/skills ------------------------------
const listWorkspaceSkillsRoute = defineRoute(
{
method: 'GET',
path: '/workspaces/{workspace_id}/skills',
params: workspaceIdParamSchema,
success: { data: listSkillsResponseSchema },
errors: {
[ErrorCode.WORKSPACE_NOT_FOUND]: {},
},
description: 'List the skills available to a workspace (no session required)',
tags: ['skills'],
operationId: 'listWorkspaceSkills',
},
async (req, reply) => {
const { workspace_id } = req.params;
const ws = await core.accessor.get(IWorkspaceRegistry).get(workspace_id);
if (ws === undefined) {
reply.send(
errEnvelope(
ErrorCode.WORKSPACE_NOT_FOUND,
`workspace ${workspace_id} does not exist`,
req.id,
),
);
return;
}
const skills = (await listWorkspaceSkillsForRoot(core, ws.root)).map(toProtocolSkill);
reply.send(okEnvelope({ skills }, req.id));
},
);
app.get(
listWorkspaceSkillsRoute.path,
listWorkspaceSkillsRoute.options,
listWorkspaceSkillsRoute.handler as Parameters<SkillsRouteHost['get']>[2],
);
// POST /sessions/{session_id}/skills/{skill_name}:activate --------------
const activateSkillRoute = defineRoute(
{
@ -217,6 +284,56 @@ export function registerSkillsRoutes(app: SkillsRouteHost, core: Scope): void {
);
}
// ---------------------------------------------------------------------------
// Workspace skill scan — session-less composition of the four skill sources
// (see header). Mirrors `SessionSkillCatalogService`'s ordered merge so the
// listing matches a session created in the same cwd.
// ---------------------------------------------------------------------------
/**
* Scan the skills a new session rooted at `workDir` would see, without creating
* a session. Resolves the same four sources the per-session catalog merges
* builtin / user / project(`workDir`) / plugin through the shared
* `ISkillDiscovery` and `skillRoots` primitives, then folds them into an
* `InMemorySkillCatalog` by the documented source priorities (lower priority
* first; `replace: true` lets higher-priority sources win name collisions). The
* priority numbers mirror `builtinSkillSource` (0), `userFileSkillSource` (10),
* `workspaceFileSkillSource` (20) and `pluginSkillSource` (25); the resulting
* name set is priority-invariant, but matching them keeps descriptor resolution
* identical to the session catalog.
*/
async function listWorkspaceSkillsForRoot(
core: Scope,
workDir: string,
): Promise<readonly SkillDefinition[]> {
const discovery = core.accessor.get(ISkillDiscovery);
const bootstrap = core.accessor.get(IBootstrapService);
const plugins = core.accessor.get(IPluginService);
const [userRootList, projectRootList, pluginRootList] = await Promise.all([
userRoots(bootstrap.homeDir, bootstrap.osHomeDir),
projectRoots(workDir),
plugins.pluginSkillRoots(),
]);
const [user, project, plugin] = await Promise.all([
discovery.discover(userRootList),
discovery.discover(projectRootList),
discovery.discover(pluginRootList),
]);
const catalog = new InMemorySkillCatalog();
const ordered = [
{ skills: BUILTIN_SKILLS, priority: 0 },
{ skills: user.skills, priority: 10 },
{ skills: project.skills, priority: 20 },
{ skills: plugin.skills, priority: 25 },
].toSorted((a, b) => a.priority - b.priority);
for (const { skills } of ordered) {
for (const skill of skills) catalog.register(skill, { replace: true });
}
return catalog.listSkills();
}
// ---------------------------------------------------------------------------
// Projection — v2 `SkillDefinition` → protocol `SkillDescriptor` (see header).
// ---------------------------------------------------------------------------

View file

@ -1,22 +1,29 @@
/**
* `GET /sessions/{session_id}/snapshot` atomic-at-a-watermark session
* snapshot for client rebuild / reconnect resync.
* `GET /sessions/{session_id}/snapshot` IM-style initial sync.
*
* Assembles `{ as_of_seq, epoch, session, messages, in_flight_turn,
* pending_approvals, pending_questions }` from v2 domain services and the
* `SessionEventBroadcaster` (which owns the durable watermark and the in-flight
* turn accumulator).
* **Reader strategy** (controlled by `KIMI_SNAPSHOT_READER`):
*
* Watermark stability: the broadcaster's `getSnapshotState` drains the
* per-session dispatch queue before reading `{seq, epoch}`, so the returned
* watermark is consistent with the session/message state read in the same
* handler.
* - `auto` (default) delegate to `ISnapshotReader`, which reads
* `state.json` + `agents/main/wire.jsonl` directly from disk and bypasses
* the heavy `ISessionLifecycleService.resume` chain (DI scope, MCP connect,
* full wire replay). Sub-200ms warm / sub-1s cold.
* - `legacy` fall back to `resume` + live service assembly. Pure operator
* escape hatch; no silent per-request fallback.
*
* **Timeout**: the auto path races against a hard `KIMI_SNAPSHOT_TIMEOUT_MS`
* ceiling (default 4000ms, under traefik's 5s cut-off). Timeout returns 50001
* with a structured `snapshot.timeout` log line so the gateway never sees a 499.
*
* **Error mapping**: `SnapshotNotFoundError` 40401; `SnapshotTimeoutError`
* 50001; everything else falls through to the global error handler ( 50001).
*/
import {
IAgentContextMemoryService,
IAgentLifecycleService,
IAgentPromptLegacyService,
ILogService,
ISessionActivity,
ISessionInteractionService,
ISessionContext,
ISessionLifecycleService,
@ -31,11 +38,18 @@ import {
sessionSnapshotResponseSchema,
type InFlightTurn,
type Message,
type SessionSnapshotResponse,
} from '@moonshot-ai/protocol';
import { z } from 'zod';
import { errEnvelope, okEnvelope } from '../envelope';
import { defineRoute } from '../middleware/defineRoute';
import {
SnapshotNotFoundError,
SnapshotTimeoutError,
loadSnapshotConfig,
} from '../services/snapshot';
import type { ISnapshotReader } from '../services/snapshot';
import { type SessionEventBroadcaster } from '../transport/ws/v1/sessionEventBroadcaster';
import { toWireApproval } from './approvals';
import { toWireQuestion } from './questions';
@ -62,10 +76,13 @@ interface SnapshotRouteHost {
export interface SnapshotRouteDeps {
readonly core: Scope;
readonly broadcaster: SessionEventBroadcaster;
readonly reader: ISnapshotReader;
}
export function registerSnapshotRoutes(app: SnapshotRouteHost, deps: SnapshotRouteDeps): void {
const { core, broadcaster } = deps;
const { core, broadcaster, reader } = deps;
const config = loadSnapshotConfig();
const useReader = config.mode !== 'legacy';
const route = defineRoute(
{
@ -75,6 +92,7 @@ export function registerSnapshotRoutes(app: SnapshotRouteHost, deps: SnapshotRou
success: { data: sessionSnapshotResponseSchema },
errors: {
[ErrorCode.SESSION_NOT_FOUND]: {},
[ErrorCode.INTERNAL_ERROR]: {},
},
description:
'Atomic session snapshot for client rebuild: state + as_of_seq watermark + epoch',
@ -82,79 +100,112 @@ export function registerSnapshotRoutes(app: SnapshotRouteHost, deps: SnapshotRou
},
async (req, reply) => {
const { session_id } = req.params;
// Resolve the live handle, loading the session from disk when it is cold
// (created by a previous process or by v1). `resume` returns `undefined`
// only when the session is unknown or its workspace is gone → 404.
const handle = await core.accessor.get(ISessionLifecycleService).resume(session_id);
if (handle === undefined) {
reply.send(
errEnvelope(ErrorCode.SESSION_NOT_FOUND, `session ${session_id} not found`, req.id),
);
return;
try {
const data = useReader
? await readViaReader(reader, session_id, config.timeoutMs)
: await readViaLegacyAssembly(core, broadcaster, session_id);
reply.send(okEnvelope(data, req.id));
} catch (err) {
if (err instanceof SnapshotNotFoundError) {
reply.send(errEnvelope(ErrorCode.SESSION_NOT_FOUND, err.message, req.id));
return;
}
if (err instanceof SnapshotTimeoutError) {
core.accessor
.get(ILogService)
.warn('snapshot.timeout', { sid: session_id, duration_ms: err.timeoutMs });
reply.send(errEnvelope(ErrorCode.INTERNAL_ERROR, err.message, req.id));
return;
}
throw err;
}
// Watermark + in-flight turn (drains the dispatch queue for consistency).
const snapState = await broadcaster.getSnapshotState(session_id);
// Session wire shape (needs the workspace root for `metadata.cwd`).
// `ISessionMetadata` normalizes legacy v1 documents on load (absent
// `version` → ISO-string timestamps → epoch ms, id backfilled), so the
// metadata read here is always v2-shaped and safe to project.
const workspaceId = handle.accessor.get(ISessionContext).workspaceId;
const workspace = await core.accessor.get(IWorkspaceRegistry).get(workspaceId);
const cwd = workspace?.root ?? '';
const meta = await handle.accessor.get(ISessionMetadata).read();
const session = toWireSession({ ...meta, workspaceId }, cwd);
// Messages — most recent page of the main agent's live history.
const main = handle.accessor.get(IAgentLifecycleService).getHandle('main');
let items: Message[] = [];
let hasMore = false;
if (main !== undefined) {
const history = main.accessor.get(IAgentContextMemoryService).get();
hasMore = history.length > SNAPSHOT_MESSAGE_PAGE_SIZE;
const page = history.slice(-SNAPSHOT_MESSAGE_PAGE_SIZE);
const offset = history.length - page.length;
items = page.map((msg, i) => toProtocolMessage(session_id, offset + i, msg, meta.createdAt));
}
const currentPromptId =
snapState.inFlightTurn === null
? undefined
: readCurrentPromptId(main);
const inFlightTurn = attachCurrentPromptIdToInFlight(
snapState.inFlightTurn,
currentPromptId,
);
// Pending approvals / questions.
const interaction = handle.accessor.get(ISessionInteractionService);
const pendingApprovals = interaction
.listPending('approval')
.map((i) => toWireApproval(i, session_id));
const pendingQuestions = interaction
.listPending('question')
.map((i) => toWireQuestion(i, session_id));
reply.send(
okEnvelope(
{
as_of_seq: snapState.seq,
epoch: snapState.epoch,
session,
messages: { items, has_more: hasMore },
in_flight_turn: inFlightTurn,
pending_approvals: pendingApprovals,
pending_questions: pendingQuestions,
},
req.id,
),
);
},
);
app.get(route.path, route.options, route.handler as Parameters<SnapshotRouteHost['get']>[2]);
}
async function readViaReader(
reader: ISnapshotReader,
sid: string,
timeoutMs: number,
): Promise<SessionSnapshotResponse> {
let timer: NodeJS.Timeout | undefined;
const timeoutPromise = new Promise<never>((_resolve, reject) => {
timer = setTimeout(() => reject(new SnapshotTimeoutError(sid, timeoutMs)), timeoutMs);
timer.unref?.();
});
try {
return await Promise.race([reader.read(sid), timeoutPromise]);
} finally {
if (timer !== undefined) clearTimeout(timer);
}
}
async function readViaLegacyAssembly(
core: Scope,
broadcaster: SessionEventBroadcaster,
sessionId: string,
): Promise<SessionSnapshotResponse> {
// Resolve the live handle, loading the session from disk when it is cold
// (created by a previous process or by v1). `resume` returns `undefined`
// only when the session is unknown or its workspace is gone → 404.
const handle = await core.accessor.get(ISessionLifecycleService).resume(sessionId);
if (handle === undefined) {
throw new SnapshotNotFoundError(sessionId);
}
// Watermark + in-flight turn (drains the dispatch queue for consistency).
const snapState = await broadcaster.getSnapshotState(sessionId);
// Session wire shape (needs the workspace root for `metadata.cwd`).
// `ISessionMetadata` normalizes legacy v1 documents on load (absent
// `version` → ISO-string timestamps → epoch ms, id backfilled), so the
// metadata read here is always v2-shaped and safe to project.
const workspaceId = handle.accessor.get(ISessionContext).workspaceId;
const workspace = await core.accessor.get(IWorkspaceRegistry).get(workspaceId);
const cwd = workspace?.root ?? '';
const meta = await handle.accessor.get(ISessionMetadata).read();
const session = toWireSession(
{ ...meta, workspaceId },
cwd,
handle.accessor.get(ISessionActivity).status(),
);
// Messages — most recent page of the main agent's live history.
const main = handle.accessor.get(IAgentLifecycleService).getHandle('main');
let items: Message[] = [];
let hasMore = false;
if (main !== undefined) {
const history = main.accessor.get(IAgentContextMemoryService).get();
hasMore = history.length > SNAPSHOT_MESSAGE_PAGE_SIZE;
const page = history.slice(-SNAPSHOT_MESSAGE_PAGE_SIZE);
const offset = history.length - page.length;
items = page.map((msg, i) => toProtocolMessage(sessionId, offset + i, msg, meta.createdAt));
}
const currentPromptId =
snapState.inFlightTurn === null ? undefined : readCurrentPromptId(main);
const inFlightTurn = attachCurrentPromptIdToInFlight(snapState.inFlightTurn, currentPromptId);
// Pending approvals / questions.
const interaction = handle.accessor.get(ISessionInteractionService);
const pendingApprovals = interaction
.listPending('approval')
.map((i) => toWireApproval(i, sessionId));
const pendingQuestions = interaction
.listPending('question')
.map((i) => toWireQuestion(i, sessionId));
return {
as_of_seq: snapState.seq,
epoch: snapState.epoch,
session,
messages: { items, has_more: hasMore },
in_flight_turn: inFlightTurn,
pending_approvals: pendingApprovals,
pending_questions: pendingQuestions,
};
}
function readCurrentPromptId(main: IAgentScopeHandle | undefined): string | undefined {
if (main === undefined) return undefined;
try {

View file

@ -0,0 +1,131 @@
import { createReadStream } from 'node:fs';
import { stat } from 'node:fs/promises';
import { extname, join, normalize, resolve, sep } from 'node:path';
import type { FastifyReply, FastifyRequest } from 'fastify';
interface WebAssetRouteHost {
get(
path: string,
handler: (req: FastifyRequest, reply: FastifyReply) => Promise<unknown>,
): unknown;
}
export async function registerWebAssetRoutes(
app: WebAssetRouteHost,
assetsDir: string,
): Promise<void> {
await assertWebAssets(assetsDir);
app.get('/', async (req, reply) => serveWebAsset(req, reply, assetsDir));
app.get('/*', async (req, reply) => serveWebAsset(req, reply, assetsDir));
}
async function assertWebAssets(assetsDir: string): Promise<void> {
try {
const info = await stat(join(assetsDir, 'index.html'));
if (!info.isFile()) {
throw new Error('index.html is not a file');
}
} catch {
throw new Error(
`Kimi web assets were not found at ${assetsDir}. Run the package build before starting the server.`,
);
}
}
async function serveWebAsset(
req: FastifyRequest,
reply: FastifyReply,
assetsDir: string,
): Promise<unknown> {
const requestUrl = new URL(req.url, 'http://kimi-web.local');
if (isReservedPath(requestUrl.pathname)) {
return reply.callNotFound();
}
const filePath = await resolveStaticFile(assetsDir, requestUrl.pathname);
if (filePath === undefined) {
return reply.code(404).type('text/plain; charset=utf-8').send('Not found');
}
const fileInfo = await stat(filePath).catch(() => undefined);
if (fileInfo === undefined || !fileInfo.isFile()) {
return reply.code(404).type('text/plain; charset=utf-8').send('Not found');
}
return reply
.type(mimeType(filePath))
.header('Content-Length', String(fileInfo.size))
.send(createReadStream(filePath));
}
async function resolveStaticFile(
assetsDir: string,
pathname: string,
): Promise<string | undefined> {
let decoded: string;
try {
decoded = decodeURIComponent(pathname);
} catch {
return undefined;
}
const normalized = normalize(decoded).replace(/^(\.\.(?:[/\\]|$))+/, '');
const relative = normalized === sep ? 'index.html' : normalized.replace(/^[/\\]/, '');
const root = resolve(assetsDir);
const candidate = resolve(
root,
relative.endsWith(sep) ? join(relative, 'index.html') : relative,
);
if (candidate !== root && !candidate.startsWith(`${root}${sep}`)) {
return undefined;
}
const info = await stat(candidate).catch(() => undefined);
if (info?.isFile() === true) {
return candidate;
}
if (extname(pathname) !== '') {
return undefined;
}
return join(root, 'index.html');
}
function isReservedPath(pathname: string): boolean {
return (
pathname === '/api' ||
pathname.startsWith('/api/') ||
pathname === '/documentation' ||
pathname.startsWith('/documentation/')
);
}
function mimeType(filePath: string): string {
switch (extname(filePath)) {
case '.html':
return 'text/html; charset=utf-8';
case '.js':
case '.mjs':
return 'text/javascript; charset=utf-8';
case '.css':
return 'text/css; charset=utf-8';
case '.json':
return 'application/json; charset=utf-8';
case '.svg':
return 'image/svg+xml';
case '.png':
return 'image/png';
case '.jpg':
case '.jpeg':
return 'image/jpeg';
case '.webp':
return 'image/webp';
case '.ico':
return 'image/x-icon';
case '.woff2':
return 'font/woff2';
default:
return 'application/octet-stream';
}
}

View file

@ -0,0 +1,11 @@
export type { ISnapshotReader } from './snapshot';
export { SnapshotNotFoundError, SnapshotTimeoutError } from './snapshot';
export {
SnapshotReader,
reduceContextRecords,
readWireRecords,
type SnapshotReaderDeps,
type SnapshotReaderLogger,
} from './snapshotReader';
export { loadSnapshotConfig } from './snapshotConfig';
export type { SnapshotConfig, SnapshotReaderMode } from './snapshotConfig';

View file

@ -0,0 +1,38 @@
/**
* `ISnapshotReader` server-layer disk reader backing
* `GET /sessions/{sid}/snapshot` in `auto` mode.
*
* Reads `state.json` + `agents/main/wire.jsonl` directly from disk, bypassing
* the `ISessionLifecycleService.resume` chain (DI-scope materialization, MCP
* connect, full wire replay). Mirrors v1's `ISnapshotService`
* (`packages/server/src/services/snapshot/snapshot.ts`).
*/
import type { SessionSnapshotResponse } from '@moonshot-ai/protocol';
export interface ISnapshotReader {
/** Assemble the atomic snapshot for `sid`. Throws `SnapshotNotFoundError` when the session (or its workspace) is absent on disk. */
read(sid: string): Promise<SessionSnapshotResponse>;
}
/** Sentinel — route maps to 40401. */
export class SnapshotNotFoundError extends Error {
readonly sessionId: string;
constructor(sessionId: string) {
super(`session ${sessionId} does not exist`);
this.name = 'SnapshotNotFoundError';
this.sessionId = sessionId;
}
}
/** Sentinel — route maps to 50001 with a structured `snapshot.timeout` log. */
export class SnapshotTimeoutError extends Error {
readonly sessionId: string;
readonly timeoutMs: number;
constructor(sessionId: string, timeoutMs: number) {
super(`snapshot ${sessionId} timed out after ${timeoutMs}ms`);
this.name = 'SnapshotTimeoutError';
this.sessionId = sessionId;
this.timeoutMs = timeoutMs;
}
}

View file

@ -0,0 +1,37 @@
/**
* Env-driven knobs for the snapshot read path. Read once at route registration.
*
* Mirrors v1 (`packages/server/src/services/snapshot/snapshotConfig.ts`):
*
* KIMI_SNAPSHOT_READER 'auto' (default) | 'legacy'
* KIMI_SNAPSHOT_TIMEOUT_MS integer ms hard ceiling on the auto path (default 4000)
* KIMI_SNAPSHOT_CACHE_LIMIT transcript LRU entries (default 32)
*/
export type SnapshotReaderMode = 'auto' | 'legacy';
export interface SnapshotConfig {
readonly mode: SnapshotReaderMode;
readonly timeoutMs: number;
readonly cacheLimit: number;
}
const DEFAULT_TIMEOUT_MS = 4000;
const DEFAULT_CACHE_LIMIT = 32;
function parseInteger(value: string | undefined, fallback: number, min: number): number {
if (value === undefined) return fallback;
const n = Number.parseInt(value, 10);
if (!Number.isFinite(n) || n < min) return fallback;
return n;
}
export function loadSnapshotConfig(env: NodeJS.ProcessEnv = process.env): SnapshotConfig {
const rawMode = env['KIMI_SNAPSHOT_READER']?.trim().toLowerCase();
const mode: SnapshotReaderMode = rawMode === 'legacy' ? 'legacy' : 'auto';
return {
mode,
timeoutMs: parseInteger(env['KIMI_SNAPSHOT_TIMEOUT_MS'], DEFAULT_TIMEOUT_MS, 100),
cacheLimit: parseInteger(env['KIMI_SNAPSHOT_CACHE_LIMIT'], DEFAULT_CACHE_LIMIT, 1),
};
}

View file

@ -0,0 +1,379 @@
/**
* `SnapshotReader` server-layer disk reader for `GET /sessions/{sid}/snapshot`
* (`KIMI_SNAPSHOT_READER=auto`, the default).
*
* Reads `<homeDir>/sessions/<workspaceId>/<sid>/state.json` and
* `…/agents/main/wire.jsonl` directly, bypassing
* `ISessionLifecycleService.resume` (DI-scope materialization, MCP connect,
* full wire replay). The transcript is reduced from the folded `context.*`
* records with the exact `contextOps` apply semantics, so the result is
* byte-identical to the live `IAgentContextMemoryService.get()` view used by
* the `legacy` (resume) path. `(size, mtimeMs)` transcript cache and the
* watermark both come from in-memory state, keeping warm reads sub-ms.
*
* Pending approvals/questions, the live status, and `current_prompt_id` are
* only available while the session is live; for a cold session they correctly
* resolve to empty / `'idle'` (a cold session owns no runtime interaction).
*/
import { readFile, stat as fsStat } from 'node:fs/promises';
import { join } from 'node:path';
import {
computeUndoCut,
IAgentLifecycleService,
IAgentPromptLegacyService,
ISessionActivity,
ISessionIndex,
ISessionInteractionService,
ISessionLifecycleService,
IWorkspaceRegistry,
normalizeSessionMeta,
toProtocolMessage,
type ContextMessage,
type Scope,
type SessionMeta,
} from '@moonshot-ai/agent-core-v2';
import type {
InFlightTurn,
Message,
SessionSnapshotResponse,
SessionStatus,
} from '@moonshot-ai/protocol';
import { toWireApproval } from '../../routes/approvals';
import { toWireQuestion } from '../../routes/questions';
import { toWireSession } from '../../routes/sessions';
import { type SessionEventBroadcaster } from '../../transport/ws/v1/sessionEventBroadcaster';
import { SnapshotNotFoundError } from './snapshot';
import type { ISnapshotReader } from './snapshot';
import { type SnapshotConfig } from './snapshotConfig';
const SESSIONS_ROOT = 'sessions';
const AGENTS_DIR = 'agents';
const BLOBS_DIR = 'blobs';
const MAIN_AGENT_ID = 'main';
const STATE_FILE = 'state.json';
const WIRE_FILE = 'wire.jsonl';
const SNAPSHOT_MESSAGE_PAGE_SIZE = 100;
const BLOBREF_PROTOCOL = 'blobref:';
const MISSING_MEDIA_PLACEHOLDER = '[media missing]';
export interface SnapshotReaderLogger {
info(obj: Record<string, unknown>, msg: string): void;
}
export interface SnapshotReaderDeps {
readonly homeDir: string;
readonly core: Scope;
readonly broadcaster: SessionEventBroadcaster;
readonly logger: SnapshotReaderLogger;
readonly config: SnapshotConfig;
}
interface TranscriptCacheEntry {
readonly size: number;
readonly mtimeMs: number;
readonly messages: ContextMessage[];
}
interface LocatedSession {
readonly workspaceId: string;
readonly cwd: string;
readonly sessionDir: string;
readonly meta: SessionMeta;
}
export class SnapshotReader implements ISnapshotReader {
private readonly transcriptCache = new Map<string, TranscriptCacheEntry>();
constructor(private readonly deps: SnapshotReaderDeps) {}
async read(sid: string): Promise<SessionSnapshotResponse> {
const startMs = Date.now();
const { core, broadcaster, logger } = this.deps;
const located = await this.locateSession(sid);
const [snapState, transcript] = await Promise.all([
broadcaster.getSnapshotState(sid),
this.readTranscriptCached(sid, located.sessionDir),
]);
const full = transcript.messages;
const hasMore = full.length > SNAPSHOT_MESSAGE_PAGE_SIZE;
const offset = hasMore ? full.length - SNAPSHOT_MESSAGE_PAGE_SIZE : 0;
const page = hasMore ? full.slice(offset) : full;
await this.rehydrateBlobRefs(page, join(located.sessionDir, AGENTS_DIR, MAIN_AGENT_ID, BLOBS_DIR));
const items = page.map((msg, i) =>
toProtocolMessage(sid, offset + i, msg, located.meta.createdAt),
);
const live = core.accessor.get(ISessionLifecycleService).get(sid);
const status = this.resolveStatus(live);
const session = toWireSession(
{ ...located.meta, workspaceId: located.workspaceId },
located.cwd,
status,
);
const inFlightTurn = this.attachCurrentPromptId(sid, live, snapState.inFlightTurn);
const { approvals, questions } = this.readPending(sid, live);
logger.info(
{
sid,
duration_ms: Date.now() - startMs,
cache: transcript.tag,
transcript_entries: full.length,
wire_bytes: transcript.wireBytes,
},
'snapshot.read',
);
return {
as_of_seq: snapState.seq,
epoch: snapState.epoch,
session,
messages: { items, has_more: hasMore },
in_flight_turn: inFlightTurn,
pending_approvals: approvals,
pending_questions: questions,
};
}
/**
* Resolve `(workspaceId, sessionDir, cwd, meta)` for `sid`. Mirrors the
* legacy route's 404 conditions: unknown to the index, or workspace no longer
* registered (cwd is unrecoverable and would produce an invalid `Session`).
*/
private async locateSession(sid: string): Promise<LocatedSession> {
const { core, homeDir } = this.deps;
const summary = await core.accessor.get(ISessionIndex).get(sid);
if (summary === undefined) throw new SnapshotNotFoundError(sid);
const workspace = await core.accessor.get(IWorkspaceRegistry).get(summary.workspaceId);
if (workspace === undefined) throw new SnapshotNotFoundError(sid);
const sessionDir = join(homeDir, SESSIONS_ROOT, summary.workspaceId, sid);
const rawMeta = await this.readStateMeta(join(sessionDir, STATE_FILE));
const meta = normalizeSessionMeta((rawMeta ?? summary) as SessionMeta, sid);
return { workspaceId: summary.workspaceId, cwd: workspace.root, sessionDir, meta };
}
/** Best-effort `state.json` read; missing / corrupt degrades to `undefined`. */
private async readStateMeta(statePath: string): Promise<SessionMeta | undefined> {
try {
const raw = await readFile(statePath, 'utf-8');
const parsed = JSON.parse(raw) as unknown;
if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) {
return undefined;
}
return parsed as SessionMeta;
} catch {
return undefined;
}
}
private async readTranscriptCached(
sid: string,
sessionDir: string,
): Promise<{
messages: ContextMessage[];
tag: 'hit' | 'miss' | 'shrink_invalidate' | 'enoent';
wireBytes: number;
}> {
const wirePath = join(sessionDir, AGENTS_DIR, MAIN_AGENT_ID, WIRE_FILE);
let info: { size: number; mtimeMs: number } | undefined;
try {
info = await fsStat(wirePath);
} catch {
info = undefined;
}
if (info === undefined) {
this.transcriptCache.delete(sid);
return { messages: [], tag: 'enoent', wireBytes: 0 };
}
const cached = this.transcriptCache.get(sid);
if (cached !== undefined && cached.size === info.size && cached.mtimeMs === info.mtimeMs) {
// LRU touch.
this.transcriptCache.delete(sid);
this.transcriptCache.set(sid, cached);
return { messages: cached.messages, tag: 'hit', wireBytes: info.size };
}
const tag: 'miss' | 'shrink_invalidate' =
cached !== undefined && info.size < cached.size ? 'shrink_invalidate' : 'miss';
if (cached !== undefined) this.transcriptCache.delete(sid);
const records = await readWireRecords(wirePath);
const messages = reduceContextRecords(records);
this.transcriptCache.set(sid, { size: info.size, mtimeMs: info.mtimeMs, messages });
while (this.transcriptCache.size > this.deps.config.cacheLimit) {
const oldest = this.transcriptCache.keys().next().value;
if (oldest === undefined) break;
this.transcriptCache.delete(oldest);
}
return { messages, tag, wireBytes: info.size };
}
private resolveStatus(
live: ReturnType<ISessionLifecycleService['get']>,
): SessionStatus {
if (live === undefined) return 'idle';
return live.accessor.get(ISessionActivity).status();
}
private attachCurrentPromptId(
sid: string,
live: ReturnType<ISessionLifecycleService['get']>,
inFlightTurn: InFlightTurn | null,
): InFlightTurn | null {
if (inFlightTurn === null || live === undefined) return inFlightTurn;
const main = live.accessor.get(IAgentLifecycleService).getHandle(MAIN_AGENT_ID);
if (main === undefined) return inFlightTurn;
let currentPromptId: string | undefined;
try {
currentPromptId = main.accessor.get(IAgentPromptLegacyService).list().active?.prompt_id;
} catch {
return inFlightTurn;
}
if (currentPromptId === undefined) return inFlightTurn;
return { ...inFlightTurn, current_prompt_id: currentPromptId };
}
private readPending(
sid: string,
live: ReturnType<ISessionLifecycleService['get']>,
): { approvals: ReturnType<typeof toWireApproval>[]; questions: ReturnType<typeof toWireQuestion>[] } {
if (live === undefined) return { approvals: [], questions: [] };
const interaction = live.accessor.get(ISessionInteractionService);
return {
approvals: interaction.listPending('approval').map((i) => toWireApproval(i, sid)),
questions: interaction.listPending('question').map((i) => toWireQuestion(i, sid)),
};
}
/** Rehydrate `blobref:<mime>;<sha256>` media URLs from `<agentDir>/blobs/<hash>`. Mirrors v1; unresolvable refs become `[media missing]`. */
private async rehydrateBlobRefs(messages: readonly ContextMessage[], blobsDir: string): Promise<void> {
const cache = new Map<string, string | undefined>();
for (const message of messages) {
for (const part of message.content) {
for (const value of Object.values(part as unknown as Record<string, unknown>)) {
if (value === null || typeof value !== 'object' || Array.isArray(value)) continue;
const media = value as { url?: unknown };
if (typeof media.url !== 'string' || !media.url.startsWith(BLOBREF_PROTOCOL)) continue;
media.url = (await resolveBlobRef(media.url, blobsDir, cache)) ?? MISSING_MEDIA_PLACEHOLDER;
}
}
}
}
}
// ---------------------------------------------------------------------------
// Pure reduction + parsing helpers
// ---------------------------------------------------------------------------
interface ContextRecord {
readonly type: string;
readonly [key: string]: unknown;
}
/**
* Reduce folded `context.*` wire records into the live `ContextMessage[]` view,
* applying the exact `contextOps.ts` semantics:
* append_message push · splice array splice · clear drop all ·
* apply_compaction `[summary, ...state.slice(count)]` · undo trailing cut.
*/
export function reduceContextRecords(records: Iterable<ContextRecord>): ContextMessage[] {
let state: ContextMessage[] = [];
for (const record of records) {
switch (record.type) {
case 'context.append_message': {
state = [...state, record['message'] as ContextMessage];
break;
}
case 'context.splice': {
const start = record['start'] as number;
const deleteCount = record['deleteCount'] as number;
const messages = record['messages'] as readonly ContextMessage[];
if (deleteCount === 0 && messages.length === 0) break;
const next = state.slice();
next.splice(start, deleteCount, ...messages);
state = next;
break;
}
case 'context.clear': {
if (state.length !== 0) state = [];
break;
}
case 'context.apply_compaction': {
const count = record['count'] as number;
const summary = record['summary'] as ContextMessage;
state = [summary, ...state.slice(count)];
break;
}
case 'context.undo': {
const count = record['count'] as number;
if (count <= 0 || state.length === 0) break;
const { cutIndex, removedCount } = computeUndoCut(state, count);
if (cutIndex < 0 || removedCount < count) break;
state = state.slice(0, cutIndex);
break;
}
default:
break;
}
}
return state;
}
/**
* Parse a `wire.jsonl` file. A torn final line (crash mid-flush) is dropped;
* corruption anywhere else throws so the route surfaces 50001. The leading
* `metadata` envelope and any non-`context.*` record are returned as-is and
* filtered by the reducer's `default` branch.
*/
export async function readWireRecords(wirePath: string): Promise<ContextRecord[]> {
const raw = await readFile(wirePath, 'utf8');
const lines = raw.split('\n');
const records: ContextRecord[] = [];
for (let i = 0; i < lines.length; i++) {
let line = lines[i]!;
if (line.endsWith('\r')) line = line.slice(0, -1);
if (line.length === 0) continue;
try {
records.push(JSON.parse(line) as ContextRecord);
} catch (parseError) {
if (i === lines.length - 1) break;
throw new Error(
`wire.jsonl: corrupted line ${i + 1} in ${wirePath}: ${String(parseError)}`,
{ cause: parseError },
);
}
}
return records;
}
async function resolveBlobRef(
url: string,
blobsDir: string,
cache: Map<string, string | undefined>,
): Promise<string | undefined> {
if (cache.has(url)) return cache.get(url);
let resolved: string | undefined;
const rest = url.slice(BLOBREF_PROTOCOL.length);
const semiIdx = rest.indexOf(';');
if (semiIdx !== -1) {
const mimeType = rest.slice(0, semiIdx);
const hash = rest.slice(semiIdx + 1);
if (/^[0-9a-f]{16,}$/i.test(hash)) {
const payload = await readFile(join(blobsDir, hash)).catch(() => undefined);
if (payload !== undefined) {
resolved = `data:${mimeType};base64,${payload.toString('base64')}`;
}
}
}
cache.set(url, resolved);
return resolved;
}

View file

@ -26,6 +26,7 @@ import { acquireLock, ServerLockedError } from './lock';
import { transformOpenApiDocument } from './openapi/transforms';
import { resolveRequestId } from './request-id';
import { registerApiV1Routes } from './routes/registerApiV1Routes';
import { registerWebAssetRoutes } from './routes/webAssets';
import {
createServerLogger,
type ServerLogger,
@ -54,6 +55,7 @@ import { createOriginHook, isOriginAllowed, parseCorsOrigins } from './middlewar
import { createSecurityHeadersHook } from './middleware/securityHeaders';
import { createAuthHook } from './middleware/auth';
import { GuiStoreService } from './services/guiStore/guiStoreService';
import { loadSnapshotConfig, SnapshotReader } from './services/snapshot';
import { ModelCatalogRefreshScheduler } from './services/modelCatalog/modelCatalogRefreshScheduler';
import { createAuthFailureLimiter } from './middleware/rateLimit';
import {
@ -92,6 +94,12 @@ export interface ServerStartOptions {
readonly rpcToken?: string;
/** Extra scope seeds applied at bootstrap (e.g. a host-provided `ISessionModelResolver`). */
readonly seeds?: ScopeSeed;
/**
* Directory of the built Kimi web UI (`dist-web`). When set, `GET /` and the
* `/*` SPA fallback serve these assets (auth-exempt, matching v1). Omit to run
* the API server without the web UI.
*/
readonly webAssetsDir?: string;
}
export interface RunningServer {
@ -211,6 +219,17 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
'onRequest',
createAuthHook(authTokenService, { limiter: authFailureLimiter, validateCredential }),
);
} else {
// `--dangerous-bypass-auth`: the operator explicitly disabled the
// bearer-token gate on every REST and WebSocket route. Warn loudly —
// especially on a non-loopback bind, where this grants unauthenticated
// remote session / filesystem / shell access to anyone who can reach the
// port. The `/api/v1/meta` payload advertises the state so the web UI can
// connect without a token.
logger.warn(
{ host, exposureClass },
'DANGEROUS: bearer-token auth is DISABLED (--dangerous-bypass-auth) — every REST and WebSocket route accepts unauthenticated requests',
);
}
if (exposureClass !== 'loopback') {
app.addHook('onSend', createSecurityHeadersHook({ tls: false }));
@ -231,6 +250,14 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
logger,
});
const snapshotReader = new SnapshotReader({
homeDir,
core,
broadcaster,
logger,
config: loadSnapshotConfig(),
});
const serverVersion = getServerVersion();
async function registerOpenApi(): Promise<void> {
@ -285,6 +312,8 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
},
connectionRegistry,
broadcaster,
snapshotReader,
dangerousBypassAuth: opts.disableAuth === true,
});
registerRpcRoutes(app, core, { token: opts.rpcToken });
@ -372,6 +401,14 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
return reply.type('application/json').send(openApiDocument);
});
// Web UI static assets (mirrors v1). Registered LAST so the `/*` SPA fallback
// only catches paths not already handled by `/api/*`, `/openapi.json`, or
// `/asyncapi.json`. The global auth hook already bypasses non-`/api` paths, so
// the page loads without a token; API calls carry it.
if (opts.webAssetsDir !== undefined) {
await registerWebAssetRoutes(app, opts.webAssetsDir);
}
// Bind with port+1 retry on EADDRINUSE (mirrors v1). The single-instance
// lock is acquired above, so any "address in use" here is a third-party
// listener — never another kimi server — and bumping the port is the desired

View file

@ -45,6 +45,7 @@ import {
IEventBus,
IEventService,
ISessionInteractionService,
ISessionIndex,
ISessionLifecycleService,
} from '@moonshot-ai/agent-core-v2';
import type {
@ -52,6 +53,7 @@ import type {
InFlightTurn,
ModelCatalogChangedEvent,
SessionCursor,
SessionMetaUpdatedEvent,
} from '@moonshot-ai/protocol';
import { isVolatileEventType } from '@moonshot-ai/protocol';
@ -174,7 +176,10 @@ export class SessionEventBroadcaster {
async getCursor(sessionId: string): Promise<{ seq: number; epoch: string }> {
const state = await this.ensureState(sessionId);
if (state === undefined) return { seq: 0, epoch: '' };
if (state === undefined) {
const cold = await this.readColdWatermark(sessionId);
return cold ?? { seq: 0, epoch: '' };
}
await state.queue;
return { seq: state.journal.seq, epoch: state.journal.epoch };
}
@ -182,7 +187,12 @@ export class SessionEventBroadcaster {
/** Atomic-at-queue watermark + in-flight turn, for the snapshot route. */
async getSnapshotState(sessionId: string): Promise<SessionSnapshotState> {
const state = await this.ensureState(sessionId);
if (state === undefined) return { seq: 0, epoch: '', inFlightTurn: null };
if (state === undefined) {
const cold = await this.readColdWatermark(sessionId);
return cold !== undefined
? { ...cold, inFlightTurn: null }
: { seq: 0, epoch: '', inFlightTurn: null };
}
await state.queue;
return {
seq: state.journal.seq,
@ -191,6 +201,27 @@ export class SessionEventBroadcaster {
};
}
/**
* Watermark for a session that is not live in this process but exists on disk
* (carried over from a prior process, or created by v1). Opens the journal
* transiently no agent/interaction listeners and not cached in
* `this.sessions` so a later live activation still attaches subscriptions.
* Returns `undefined` when the session is unknown to the index (truly absent).
*/
private async readColdWatermark(
sessionId: string,
): Promise<{ seq: number; epoch: string } | undefined> {
const summary = await this.opts.core.accessor.get(ISessionIndex).get(sessionId);
if (summary === undefined) return undefined;
const journal = await SessionEventJournal.open(
sessionJournalPath(this.opts.eventsDir, sessionId),
this.opts.logger,
);
const watermark = { seq: journal.seq, epoch: journal.epoch };
await journal.close();
return watermark;
}
async close(): Promise<void> {
this.coreEventSubscription.dispose();
for (const state of this.sessions.values()) {
@ -253,15 +284,34 @@ export class SessionEventBroadcaster {
}
private onCoreEvent(event: GlobalEvent): void {
if (event.type !== 'event.model_catalog.changed') return;
const payload = modelCatalogChangedPayload(event.payload);
if (payload === undefined) return;
const modelEvent: ModelCatalogChangedEvent = { type: 'event.model_catalog.changed', ...payload };
void this.dispatchGlobal({
...modelEvent,
agentId: 'main',
sessionId: GLOBAL_SESSION_ID,
});
if (event.type === 'event.model_catalog.changed') {
const payload = modelCatalogChangedPayload(event.payload);
if (payload === undefined) return;
const modelEvent: ModelCatalogChangedEvent = {
type: 'event.model_catalog.changed',
...payload,
};
void this.dispatchGlobal({
...modelEvent,
agentId: 'main',
sessionId: GLOBAL_SESSION_ID,
});
return;
}
if (event.type === 'session.meta.updated') {
const payload = sessionMetaUpdatedPayload(event.payload);
if (payload === undefined) return;
// v1 broadcasts title changes to every connection (not just subscribers of
// the session) so session lists stay in sync — route via the global state,
// and `isGlobalEvent` fans it out to all targets. agentId/sessionId are
// attached here (the protocol event itself carries only title/patch).
void this.dispatchGlobal({
type: 'session.meta.updated',
...payload,
agentId: 'main',
sessionId: GLOBAL_SESSION_ID,
} as Event);
}
}
private async dispatchGlobal(event: Event): Promise<void> {
@ -433,6 +483,7 @@ function isVolatileSignal(type: string): boolean {
/** Session/workspace/config/model-catalog events are broadcast to every connection. */
function isGlobalEvent(type: string): boolean {
return (
type === 'session.meta.updated' ||
type.startsWith('event.session.') ||
type.startsWith('event.workspace.') ||
type.startsWith('event.config.') ||
@ -537,3 +588,25 @@ function modelCatalogChangedPayload(
failed: candidate.failed,
};
}
/**
* Validate the `session.meta.updated` payload published on the core
* `IEventService` by the `POST /sessions/{id}/profile` route. The route wraps
* the v1 fields under `payload`; we unwrap them here and re-attach
* agentId/sessionId at the edge.
*/
function sessionMetaUpdatedPayload(
payload: unknown,
): Pick<SessionMetaUpdatedEvent, 'title' | 'patch'> | undefined {
if (typeof payload !== 'object' || payload === null) return undefined;
const candidate = payload as Partial<SessionMetaUpdatedEvent>;
const title = typeof candidate.title === 'string' ? candidate.title : undefined;
const patch =
typeof candidate.patch === 'object' &&
candidate.patch !== null &&
!Array.isArray(candidate.patch)
? candidate.patch
: undefined;
if (title === undefined && patch === undefined) return undefined;
return { title, patch };
}

View file

@ -0,0 +1,126 @@
/**
* server-v2 `--dangerous-bypass-auth` (`disableAuth`) wiring.
*
* When the operator opts out of the bearer-token gate, every REST and
* WebSocket route accepts unauthenticated requests, and `/api/v1/meta`
* advertises `dangerous_bypass_auth: true` so the web UI can connect without a
* token. The default (hardened) boot keeps the gate closed and reports
* `dangerous_bypass_auth: false`.
*/
import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, describe, expect, it } from 'vitest';
import { WebSocket, type RawData } from 'ws';
import { type RunningServer, startServer } from '../src/start';
import { fixedTokenAuth } from './helpers/fixedAuth';
const TOKEN = 'test-token';
function rawToString(data: RawData): string {
if (typeof data === 'string') return data;
if (Buffer.isBuffer(data)) return data.toString('utf8');
if (Array.isArray(data)) return Buffer.concat(data).toString('utf8');
return Buffer.from(data as ArrayBuffer).toString('utf8');
}
/** Resolve when the socket opens and the server's first frame arrives. */
function openConn(url: string): Promise<{ ws: WebSocket; firstFrame: unknown }> {
return new Promise((resolve, reject) => {
const ws = new WebSocket(url);
ws.once('message', (data) => {
try {
resolve({ ws, firstFrame: JSON.parse(rawToString(data)) });
} catch {
resolve({ ws, firstFrame: null });
}
});
ws.once('error', reject);
});
}
describe('server-v2 disableAuth (--dangerous-bypass-auth)', () => {
let server: RunningServer | undefined;
let home: string | undefined;
const sockets: WebSocket[] = [];
afterEach(async () => {
for (const ws of sockets.splice(0)) {
try {
ws.close();
} catch {
// ignore
}
}
if (server !== undefined) {
await server.close();
server = undefined;
}
if (home !== undefined) {
await rm(home, { recursive: true, force: true });
home = undefined;
}
});
async function boot(disableAuth?: boolean): Promise<{ base: string; port: number }> {
home = await mkdtemp(join(tmpdir(), 'kimi-server-v2-disable-auth-'));
server = await startServer({
host: '127.0.0.1',
port: 0,
homeDir: home,
logLevel: 'silent',
authTokenService: fixedTokenAuth(TOKEN),
disableAuth,
});
return { base: `http://127.0.0.1:${server.port}`, port: server.port };
}
it('disableAuth:true lets REST through without a token and advertises it in /meta', async () => {
const { base } = await boot(true);
const meta = await fetch(`${base}/api/v1/meta`); // no Authorization header
expect(meta.status).toBe(200);
const metaBody = (await meta.json()) as {
code: number;
data: { dangerous_bypass_auth: boolean };
};
expect(metaBody.code).toBe(0);
expect(metaBody.data.dangerous_bypass_auth).toBe(true);
// A normally-protected route is also open without a credential.
const auth = await fetch(`${base}/api/v1/auth`);
expect(auth.status).toBe(200);
});
it('disableAuth:true lets WebSocket upgrades through without a token', async () => {
const { port } = await boot(true);
const v1 = await openConn(`ws://127.0.0.1:${port}/api/v1/ws`);
sockets.push(v1.ws);
expect(v1.firstFrame).toMatchObject({ type: 'server_hello' });
const v2 = await openConn(`ws://127.0.0.1:${port}/api/v2/ws`);
sockets.push(v2.ws);
expect(v2.firstFrame).toMatchObject({ type: 'ready' });
});
it('default boot keeps the gate closed and reports dangerous_bypass_auth: false', async () => {
const { base } = await boot(undefined);
const unauthed = await fetch(`${base}/api/v1/meta`); // no token
expect(unauthed.status).toBe(401);
const meta = await fetch(`${base}/api/v1/meta`, {
headers: { authorization: `Bearer ${TOKEN}` },
});
expect(meta.status).toBe(200);
const metaBody = (await meta.json()) as {
code: number;
data: { dangerous_bypass_auth: boolean };
};
expect(metaBody.data.dangerous_bypass_auth).toBe(false);
});
});

View file

@ -4,6 +4,9 @@ import { join } from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { IEventService } from '@moonshot-ai/agent-core-v2';
import { sessionWarningsResponseSchema } from '@moonshot-ai/protocol';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
@ -86,6 +89,14 @@ describe('server-v2 /api/v1/sessions', () => {
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function deleteJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`, {
method: 'DELETE',
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
it('creates a session from metadata.cwd', async () => {
const cwd = home as string;
const { status, body } = await postJson<SessionWire>('/api/v1/sessions', {
@ -381,10 +392,239 @@ describe('server-v2 /api/v1/sessions', () => {
expect(status).toBe(200);
expect(body.code).toBe(0);
expect(body.data).toEqual({ warnings: [] });
// Lock the wire shape to the shared protocol schema (schema-fidelity rule):
// a mirror route must keep the v1 envelope byte-compatible.
expect(sessionWarningsResponseSchema.parse(body.data)).toEqual({ warnings: [] });
});
it('returns 40401 for warnings of a missing session', async () => {
const { body } = await getJson<null>('/api/v1/sessions/sess_missing_warnings/warnings');
expect(body.code).toBe(40401);
});
it('lists only archived sessions with archived_only', async () => {
const cwd = home as string;
const a = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
const b = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
expect(a.body.code).toBe(0);
expect(b.body.code).toBe(0);
const archivedId = a.body.data.id;
const liveId = b.body.data.id;
const archived = await postJson<{ archived: boolean }>(
`/api/v1/sessions/${archivedId}:archive`,
);
expect(archived.body.code).toBe(0);
// Default list hides archived sessions.
const normal = await getJson<PageWire>('/api/v1/sessions');
expect(normal.body.data.items.some((s) => s.id === liveId)).toBe(true);
expect(normal.body.data.items.some((s) => s.id === archivedId)).toBe(false);
// archived_only shows only the archived one.
const onlyArchived = await getJson<PageWire>('/api/v1/sessions?archived_only=true');
expect(onlyArchived.body.code).toBe(0);
expect(onlyArchived.body.data.items.some((s) => s.id === archivedId)).toBe(true);
expect(onlyArchived.body.data.items.some((s) => s.id === liveId)).toBe(false);
// include_archive shows both.
const all = await getJson<PageWire>('/api/v1/sessions?include_archive=true');
expect(all.body.data.items.some((s) => s.id === liveId)).toBe(true);
expect(all.body.data.items.some((s) => s.id === archivedId)).toBe(true);
});
it('rejects archived_only combined with include_archive (40001)', async () => {
const { body } = await getJson<null>(
'/api/v1/sessions?archived_only=true&include_archive=true',
);
expect(body.code).toBe(40001);
});
it('rejects a malformed workspace_id when listing (40001)', async () => {
const { body } = await getJson<null>('/api/v1/sessions?workspace_id=not-a-workspace-id');
expect(body.code).toBe(40001);
});
it('returns 40410 for an unknown workspace_id when listing', async () => {
const { body } = await getJson<null>('/api/v1/sessions?workspace_id=wd_missing_000000000000');
expect(body.code).toBe(40410);
});
it('filters listed sessions by the status query (post-page, like v1)', async () => {
const cwd = home as string;
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
const id = created.body.data.id;
// A freshly-created session has no turn, so the live activity is `idle` —
// the wire status is the resolved activity, not a constant placeholder.
expect(created.body.data.status).toBe('idle');
const idle = await getJson<PageWire>('/api/v1/sessions?status=idle');
expect(idle.body.code).toBe(0);
expect(idle.body.data.items.some((s) => s.id === id)).toBe(true);
const running = await getJson<PageWire>('/api/v1/sessions?status=running');
expect(running.body.code).toBe(0);
expect(running.body.data.items.some((s) => s.id === id)).toBe(false);
});
it('filters child sessions by the status query', async () => {
const cwd = home as string;
const parent = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
const parentId = parent.body.data.id;
const child = await postJson<SessionWire>(`/api/v1/sessions/${parentId}/children`, {});
const childId = child.body.data.id;
expect(child.body.data.status).toBe('idle');
const idle = await getJson<PageWire>(`/api/v1/sessions/${parentId}/children?status=idle`);
expect(idle.body.code).toBe(0);
expect(idle.body.data.items.some((s) => s.id === childId)).toBe(true);
const running = await getJson<PageWire>(`/api/v1/sessions/${parentId}/children?status=running`);
expect(running.body.code).toBe(0);
expect(running.body.data.items.some((s) => s.id === childId)).toBe(false);
});
it('keeps a session listable and gettable with cwd after its workspace is unregistered (gap G3)', async () => {
const cwd = home as string;
const created = await postJson<SessionWire>('/api/v1/sessions', {
title: 'g3',
metadata: { cwd },
});
expect(created.body.code).toBe(0);
const id = created.body.data.id;
const workspaceId = created.body.data.workspace_id;
// Unregister the workspace without removing on-disk content. The session
// persists its frozen cwd, so it must remain listable / gettable with the
// original cwd instead of being filtered (list) or 404 (get/profile).
const del = await deleteJson<{ deleted: boolean }>(`/api/v1/workspaces/${workspaceId}`);
expect(del.body.code).toBe(0);
const listed = await getJson<PageWire>('/api/v1/sessions');
expect(listed.body.code).toBe(0);
const found = listed.body.data.items.find((s) => s.id === id);
expect(found).toBeDefined();
expect(found?.metadata.cwd).toBe(cwd);
const got = await getJson<SessionWire>(`/api/v1/sessions/${id}`);
expect(got.body.code).toBe(0);
expect(got.body.data.metadata.cwd).toBe(cwd);
const profile = await getJson<SessionWire>(`/api/v1/sessions/${id}/profile`);
expect(profile.body.code).toBe(0);
expect(profile.body.data.metadata.cwd).toBe(cwd);
});
it('merges metadata via profile and keeps cwd authoritative', async () => {
const cwd = home as string;
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
const id = created.body.data.id;
const first = await postJson<SessionWire>(`/api/v1/sessions/${id}/profile`, {
metadata: { foo: 'bar' },
});
expect(first.body.code).toBe(0);
expect(first.body.data.metadata['foo']).toBe('bar');
expect(first.body.data.metadata.cwd).toBe(cwd);
const got = await getJson<SessionWire>(`/api/v1/sessions/${id}`);
expect(got.body.data.metadata['foo']).toBe('bar');
expect(got.body.data.metadata.cwd).toBe(cwd);
});
it('replaces custom metadata on a second profile update (v1 semantics)', async () => {
const cwd = home as string;
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
const id = created.body.data.id;
await postJson<SessionWire>(`/api/v1/sessions/${id}/profile`, { metadata: { foo: 'bar' } });
const second = await postJson<SessionWire>(`/api/v1/sessions/${id}/profile`, {
metadata: { baz: 1 },
});
expect(second.body.code).toBe(0);
// v1 writes the patch straight into `custom` (replace, not deep-merge): the
// first key is gone, the new key is present, and cwd still wins.
expect(second.body.data.metadata['foo']).toBeUndefined();
expect(second.body.data.metadata['baz']).toBe(1);
expect(second.body.data.metadata.cwd).toBe(cwd);
});
it('applies agent_config.permission_mode via profile idempotently', async () => {
const cwd = home as string;
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
const id = created.body.data.id;
const first = await postJson<SessionWire>(`/api/v1/sessions/${id}/profile`, {
agent_config: { permission_mode: 'yolo' },
});
expect(first.body.code).toBe(0);
// Re-applying the same mode must not error (the setter is idempotent).
const again = await postJson<SessionWire>(`/api/v1/sessions/${id}/profile`, {
agent_config: { permission_mode: 'yolo' },
});
expect(again.body.code).toBe(0);
});
it('guards agent_config.plan_mode so a repeated true does not re-enter', async () => {
const cwd = home as string;
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
const id = created.body.data.id;
const first = await postJson<SessionWire>(`/api/v1/sessions/${id}/profile`, {
agent_config: { plan_mode: true },
});
expect(first.body.code).toBe(0);
// Without the diff-guard this second enter would throw 'Already in plan mode'
// and surface as a non-zero code.
const again = await postJson<SessionWire>(`/api/v1/sessions/${id}/profile`, {
agent_config: { plan_mode: true },
});
expect(again.body.code).toBe(0);
});
it('maps goal already_exists from agent_config.goal_objective (40913)', async () => {
const cwd = home as string;
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
const id = created.body.data.id;
const first = await postJson<SessionWire>(`/api/v1/sessions/${id}/profile`, {
agent_config: { goal_objective: 'ship the feature' },
});
expect(first.body.code).toBe(0);
const dup = await postJson<null>(`/api/v1/sessions/${id}/profile`, {
agent_config: { goal_objective: 'ship the feature' },
});
expect(dup.body.code).toBe(40913);
});
it('publishes session.meta.updated on the core bus when renaming via profile', async () => {
const cwd = home as string;
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
const id = created.body.data.id;
const events: { type: string; payload: unknown }[] = [];
const sub = (server as RunningServer).core.accessor
.get(IEventService)
.subscribe((event) => events.push(event));
const updated = await postJson<SessionWire>(`/api/v1/sessions/${id}/profile`, {
title: 'renamed-via-profile',
});
expect(updated.body.code).toBe(0);
sub.dispose();
const meta = events.find((e) => e.type === 'session.meta.updated');
expect(meta).toBeDefined();
expect((meta?.payload as { title?: string } | undefined)?.title).toBe('renamed-via-profile');
});
it('returns 40401 when updating the profile of a missing session', async () => {
const { body } = await postJson<null>('/api/v1/sessions/sess_missing_profile/profile', {
title: 'nope',
});
expect(body.code).toBe(40401);
});
});

View file

@ -1,22 +1,27 @@
/**
* `/api/v1` skills routes server-v2 port of `packages/server/test/skills.e2e.test.ts`.
*
* Covers the wire contract of the two endpoints:
* Covers the wire contract of the three endpoints:
* - GET /api/v1/sessions/{sid}/skills envelope shape + skills[]
* - GET on an unknown session 40401 "does not exist"
* - GET on a persisted-but-not-activated session 40401 "not activated ..."
* - GET /api/v1/workspaces/{wid}/skills skills[] (no session)
* - GET workspace listing == session listing (same cwd) parity
* - GET on an unknown workspace 40410
* - POST /api/v1/sessions/{sid}/skills/{name}:activate {activated:true, skill_name}
* - POST :activate an unknown skill 40415
* - POST bare `{name}` / bogus action 40001
*
* Skills are resolved from the per-session `ISessionSkillCatalog` (list) and the main
* agent's `IAgentSkillService` (activate). A session created through
* Session skills are resolved from the per-session `ISessionSkillCatalog` (list)
* and the main agent's `IAgentSkillService` (activate). A session created through
* `POST /sessions` is already activated (live), so listing/activation work
* immediately; the "not activated" branch is exercised by archiving the session
* (it stays in the index but leaves the live map).
* (it stays in the index but leaves the live map). Workspace skills are scanned
* session-less from the workspace root via the edge composition in
* `routes/skills.ts`, which must match the session listing for the same cwd.
*/
import { mkdtemp, rm } from 'node:fs/promises';
import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
@ -90,9 +95,9 @@ describe('server-v2 /api/v1 skills', () => {
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function createSession(): Promise<string> {
async function createSession(cwd: string = home as string): Promise<string> {
const { body } = await postJson<{ id: string }>('/api/v1/sessions', {
metadata: { cwd: home as string },
metadata: { cwd },
});
expect(body.code).toBe(0);
return body.data.id;
@ -107,6 +112,33 @@ describe('server-v2 /api/v1 skills', () => {
if (agents.getHandle('main') === undefined) await agents.create({ agentId: 'main' });
}
async function registerWorkspace(root: string): Promise<string> {
const { body } = await postJson<{ id: string }>('/api/v1/workspaces', { root });
expect(body.code).toBe(0);
return body.data.id;
}
// Lives under `home` so the existing afterEach cleanup removes it; unique per
// call so parallel tests do not collide on skill roots.
async function makeWorkspaceDir(): Promise<string> {
const dir = join(
home as string,
`workspace-${Date.now()}-${Math.random().toString(36).slice(2)}`,
);
await mkdir(dir, { recursive: true });
return dir;
}
/** Seed a project skill bundle at `<root>/.kimi-code/skills/<name>/SKILL.md`. */
async function seedProjectSkill(root: string, name: string): Promise<void> {
const dir = join(root, '.kimi-code', 'skills', name);
await mkdir(dir, { recursive: true });
await writeFile(
join(dir, 'SKILL.md'),
`---\nname: ${name}\ndescription: e2e test skill ${name}\n---\n\nSay hello to $ARGUMENTS.\n`,
);
}
describe('GET /api/v1/sessions/{sid}/skills', () => {
it('returns 40401 for an unknown session', async () => {
const { body } = await getJson<null>('/api/v1/sessions/nope/skills');
@ -191,4 +223,45 @@ describe('server-v2 /api/v1 skills', () => {
expect(body.msg).toMatch(/unsupported action/);
});
});
describe('GET /api/v1/workspaces/{wid}/skills', () => {
it('lists skills for a workspace without creating a session', async () => {
const workspaceDir = await makeWorkspaceDir();
await seedProjectSkill(workspaceDir, 'e2e-greeting');
const wid = await registerWorkspace(workspaceDir);
const { body } = await getJson<{ skills: SkillWire[] }>(
`/api/v1/workspaces/${wid}/skills`,
);
expect(body.code).toBe(0);
const skills = listSkillsResponseSchema.parse(body.data).skills;
const seeded = skills.find((s) => s.name === 'e2e-greeting');
expect(seeded).toBeDefined();
expect(seeded?.source).toBe('project');
expect(seeded?.description).toBe('e2e test skill e2e-greeting');
});
it('matches the session listing for the same cwd', async () => {
const workspaceDir = await makeWorkspaceDir();
await seedProjectSkill(workspaceDir, 'e2e-greeting');
const wid = await registerWorkspace(workspaceDir);
const sid = await createSession(workspaceDir);
const [wsRes, sessRes] = await Promise.all([
getJson<{ skills: SkillWire[] }>(`/api/v1/workspaces/${wid}/skills`),
getJson<{ skills: SkillWire[] }>(`/api/v1/sessions/${sid}/skills`),
]);
const wsSkills = listSkillsResponseSchema.parse(wsRes.body.data).skills;
const sessSkills = listSkillsResponseSchema.parse(sessRes.body.data).skills;
const names = (xs: readonly { name: string }[]) => xs.map((s) => s.name).toSorted();
expect(names(wsSkills)).toEqual(names(sessSkills));
});
it('returns 40410 for an unknown workspace', async () => {
const { body } = await getJson<null>(
'/api/v1/workspaces/wd_does-not-exist_000000000000/skills',
);
expect(body.code).toBe(40410);
});
});
});

View file

@ -3,7 +3,7 @@
* snapshot shape and watermark consistency.
*/
import { mkdtemp, rm, writeFile } from 'node:fs/promises';
import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
@ -12,6 +12,8 @@ import {
IAgentEventSinkService,
IAgentLifecycleService,
IAgentPromptLegacyService,
ILogService,
ISessionActivity,
ISessionInteractionService,
ISessionContext,
ISessionLifecycleService,
@ -22,6 +24,7 @@ import { sessionSnapshotResponseSchema, type AgentEvent } from '@moonshot-ai/pro
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { registerSnapshotRoutes } from '../src/routes/snapshot';
import { SnapshotNotFoundError } from '../src/services/snapshot';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
@ -69,6 +72,7 @@ describe('server-v2 snapshot route enrichment', () => {
],
[IAgentLifecycleService, { getHandle: () => main }],
[ISessionInteractionService, { listPending: () => [] }],
[ISessionActivity, { status: () => 'idle' }],
]),
};
const core = {
@ -96,14 +100,28 @@ describe('server-v2 snapshot route enrichment', () => {
reply: { send(payload: unknown): unknown },
) => Promise<void> | void)
| undefined;
registerSnapshotRoutes(
{
get: (_path, _options, handler) => {
routeHandler = handler;
// Exercise the legacy (resume + live assembly) path — the fakes model the
// live scope, not the on-disk reader.
const previousReaderMode = process.env['KIMI_SNAPSHOT_READER'];
process.env['KIMI_SNAPSHOT_READER'] = 'legacy';
const unusedReader = { read: async () => ({}) as never };
try {
registerSnapshotRoutes(
{
get: (_path, _options, handler) => {
routeHandler = handler;
},
},
},
{ core: core as never, broadcaster: broadcaster as never },
);
{
core: core as never,
broadcaster: broadcaster as never,
reader: unusedReader as never,
},
);
} finally {
if (previousReaderMode === undefined) delete process.env['KIMI_SNAPSHOT_READER'];
else process.env['KIMI_SNAPSHOT_READER'] = previousReaderMode;
}
let payload: unknown;
await routeHandler?.(
@ -126,6 +144,83 @@ describe('server-v2 snapshot route enrichment', () => {
});
});
describe('server-v2 snapshot route error mapping', () => {
function captureHandler(
deps: { core: unknown; broadcaster: unknown; reader: unknown },
env?: Record<string, string>,
) {
const previous = new Map<string, string | undefined>();
for (const [k, v] of Object.entries(env ?? {})) {
previous.set(k, process.env[k]);
process.env[k] = v;
}
let handler:
| ((
req: { id: string; params: { session_id: string } },
reply: { send(payload: unknown): unknown },
) => Promise<void> | void)
| undefined;
try {
registerSnapshotRoutes(
{
get: (_path, _options, h) => {
handler = h;
},
},
deps as never,
);
} finally {
for (const [k, v] of previous) {
if (v === undefined) delete process.env[k];
else process.env[k] = v;
}
}
return handler!;
}
it('maps SnapshotNotFoundError to 40401', async () => {
const warns: unknown[] = [];
const core = {
accessor: fakeAccessor([[ILogService, { warn: (...a: unknown[]) => warns.push(a) }]]),
};
const reader = {
read: async () => {
throw new SnapshotNotFoundError('sess_missing');
},
};
const handler = captureHandler({ core, broadcaster: {}, reader });
let payload: unknown;
await handler(
{ id: 'req_404', params: { session_id: 'sess_missing' } },
{ send: (v) => (payload = v) },
);
expect((payload as { code: number }).code).toBe(40401);
expect(warns).toHaveLength(0);
});
it('maps SnapshotTimeoutError to 50001 and logs snapshot.timeout', async () => {
const warns: unknown[] = [];
const core = {
accessor: fakeAccessor([[ILogService, { warn: (...a: unknown[]) => warns.push(a) }]]),
};
const reader = {
read: () => new Promise<never>(() => {}), // hangs → triggers the timeout race
};
const handler = captureHandler(
{ core, broadcaster: {}, reader },
{ KIMI_SNAPSHOT_TIMEOUT_MS: '150' },
);
let payload: unknown;
await handler(
{ id: 'req_to', params: { session_id: 'sess_slow' } },
{ send: (v) => (payload = v) },
);
expect((payload as { code: number }).code).toBe(50001);
expect(warns).toHaveLength(1);
expect((warns[0] as unknown[])[0]).toBe('snapshot.timeout');
});
});
describe('server-v2 GET /api/v1/sessions/:id/snapshot', () => {
let server: RunningServer | undefined;
let home: string | undefined;
@ -239,6 +334,55 @@ describe('server-v2 GET /api/v1/sessions/:id/snapshot', () => {
expect(snap.session.id).toBe(sid);
});
// The auto reader must source messages from `agents/main/wire.jsonl` on disk
// — not from a live (resumed) context. We seed a wire log, restart so the
// session is genuinely cold, then assert the snapshot returns the on-disk
// transcript while the scope stays un-materialized.
it('auto reader returns messages read directly from wire.jsonl for a cold session', async () => {
const sid = await createSession();
const live = server!.core.accessor.get(ISessionLifecycleService).get(sid);
if (live === undefined) throw new Error(`session ${sid} not found`);
const metaScope = live.accessor.get(ISessionContext).metaScope;
const wireDir = join(home as string, metaScope, 'agents', 'main');
await mkdir(wireDir, { recursive: true });
const records = [
{ type: 'metadata', protocol_version: '1.4', created_at: Date.now() },
{
type: 'context.append_message',
message: { role: 'user', content: [{ type: 'text', text: 'hello-from-disk' }], toolCalls: [] },
},
{
type: 'context.append_message',
message: {
role: 'assistant',
content: [{ type: 'text', text: 'hi-from-disk' }],
toolCalls: [],
},
},
];
await writeFile(
join(wireDir, 'wire.jsonl'),
records.map((r) => JSON.stringify(r)).join('\n') + '\n',
'utf-8',
);
await server!.close();
server = undefined;
server = await startServer({ host: '127.0.0.1', port: 0, homeDir: home, logLevel: 'silent' });
base = `http://127.0.0.1:${server.port}`;
// Guard: still cold — the auto reader must serve from disk, not resume.
expect(server!.core.accessor.get(ISessionLifecycleService).get(sid)).toBeUndefined();
const snap = await snapshot(sid);
expect(snap.session.id).toBe(sid);
expect(snap.messages.items).toHaveLength(2);
expect((snap.messages.items[0]!.content[0] as { text: string }).text).toBe('hello-from-disk');
expect((snap.messages.items[1]!.content[0] as { text: string }).text).toBe('hi-from-disk');
expect(snap.epoch).toMatch(/^ep_/);
});
// Regression for the v1-layout 50001 ("Invalid time value"): v1 persists
// `createdAt`/`updatedAt` as ISO strings (and omits the v2 `id` field) in
// `state.json`. Projecting that raw metadata broke message timestamp

View file

@ -0,0 +1,350 @@
/**
* Unit tests for the server-layer disk reader (`services/snapshot`).
*
* Constructs `SnapshotReader` with stub core services and a real tmp `homeDir`,
* writing `state.json` + `agents/main/wire.jsonl` directly exercising the
* disk read, the `context.*` reduction, the `(size, mtimeMs)` transcript cache,
* `state.json` normalization, and `KIMI_SNAPSHOT_*` config parsing without
* booting a Fastify daemon.
*/
import { mkdir, rm, writeFile } from 'node:fs/promises';
import { mkdtemp } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import {
ISessionIndex,
ISessionLifecycleService,
IWorkspaceRegistry,
type ContextMessage,
type SessionSummary,
} from '@moonshot-ai/agent-core-v2';
import { afterEach, describe, expect, it } from 'vitest';
import {
loadSnapshotConfig,
reduceContextRecords,
readWireRecords,
SnapshotNotFoundError,
SnapshotReader,
type SnapshotReaderDeps,
} from '../src/services/snapshot';
// ─── tiny stubs ───────────────────────────────────────────────────────────
function fakeAccessor(entries: ReadonlyArray<readonly [unknown, unknown]>) {
const services = new Map<unknown, unknown>(entries);
return {
get<T>(id: unknown): T {
if (!services.has(id)) throw new Error(`unexpected service request: ${String(id)}`);
return services.get(id) as T;
},
};
}
const noopLogger = { info: () => {} };
interface Fixture {
homeDir: string;
workspaceId: string;
sessionDir: (sid: string) => string;
index: Map<string, SessionSummary>;
reader: SnapshotReader;
broadcaster: { seq: number; epoch: string; inFlightTurn: unknown };
}
const tmpDirs: string[] = [];
async function makeFixtureAsync(opts?: { cacheLimit?: number }): Promise<Fixture> {
const homeDir = await mkdtemp(join(tmpdir(), 'kimi-snapshot-reader-'));
tmpDirs.push(homeDir);
const workspaceId = 'wd_unittest_012345abcdef';
const index = new Map<string, SessionSummary>();
const workspaces = new Map([[workspaceId, { root: join(homeDir, 'workspace') }]]);
const core = {
accessor: fakeAccessor([
[ISessionIndex, { get: async (sid: string) => index.get(sid) }],
[IWorkspaceRegistry, { get: async (ws: string) => workspaces.get(ws) }],
// Cold by default — no live handle.
[ISessionLifecycleService, { get: () => undefined }],
]),
};
const broadcaster = { seq: 0, epoch: 'ep_unit', inFlightTurn: null };
const deps: SnapshotReaderDeps = {
homeDir,
core: core as never,
broadcaster: {
getSnapshotState: async () => ({
seq: broadcaster.seq,
epoch: broadcaster.epoch,
inFlightTurn: broadcaster.inFlightTurn as never,
}),
} as never,
logger: noopLogger,
config: { mode: 'auto', timeoutMs: 4000, cacheLimit: opts?.cacheLimit ?? 32 },
};
return {
homeDir,
workspaceId,
sessionDir: (sid) => join(homeDir, 'sessions', workspaceId, sid),
index,
reader: new SnapshotReader(deps),
broadcaster,
};
}
function userMessage(text: string): ContextMessage {
return { role: 'user', content: [{ type: 'text', text }], toolCalls: [] };
}
async function seedSession(
f: Fixture,
sid: string,
opts?: { createdAt?: number; title?: string; rawState?: Record<string, unknown> },
): Promise<void> {
const createdAt = opts?.createdAt ?? 1700000000000;
f.index.set(sid, {
id: sid,
workspaceId: f.workspaceId,
title: opts?.title,
createdAt,
updatedAt: createdAt,
archived: false,
});
const state = opts?.rawState ?? {
id: sid,
version: 2,
createdAt,
updatedAt: createdAt,
archived: false,
title: opts?.title,
};
await mkdir(f.sessionDir(sid), { recursive: true });
await writeFile(join(f.sessionDir(sid), 'state.json'), JSON.stringify(state), 'utf-8');
}
async function writeWire(sessionDir: string, lines: ReadonlyArray<unknown>): Promise<void> {
const agentDir = join(sessionDir, 'agents', 'main');
await mkdir(agentDir, { recursive: true });
const body = lines.map((l) => JSON.stringify(l)).join('\n') + (lines.length > 0 ? '\n' : '');
await writeFile(join(agentDir, 'wire.jsonl'), body, 'utf-8');
}
afterEach(async () => {
for (const dir of tmpDirs.splice(0)) await rm(dir, { recursive: true, force: true });
});
// ─── SnapshotReader.read ──────────────────────────────────────────────────
describe('SnapshotReader.read', () => {
it('throws SnapshotNotFoundError for an unknown session', async () => {
const f = await makeFixtureAsync();
await expect(f.reader.read('sess_missing')).rejects.toBeInstanceOf(SnapshotNotFoundError);
});
it('throws SnapshotNotFoundError when the workspace is gone', async () => {
const f = await makeFixtureAsync();
f.index.set('sess_orphan', {
id: 'sess_orphan',
workspaceId: 'wd_gone_000000000000',
createdAt: 1,
updatedAt: 1,
archived: false,
});
await expect(f.reader.read('sess_orphan')).rejects.toBeInstanceOf(SnapshotNotFoundError);
});
it('returns empty messages for a session with no wire.jsonl', async () => {
const f = await makeFixtureAsync();
await seedSession(f, 'sess_empty');
const snap = await f.reader.read('sess_empty');
expect(snap.session.id).toBe('sess_empty');
expect(snap.session.status).toBe('idle');
expect(snap.messages.items).toEqual([]);
expect(snap.messages.has_more).toBe(false);
expect(snap.in_flight_turn).toBeNull();
expect(snap.pending_approvals).toEqual([]);
expect(snap.as_of_seq).toBe(0);
expect(snap.epoch).toBe('ep_unit');
});
it('builds messages from context.append_message records', async () => {
const f = await makeFixtureAsync();
await seedSession(f, 'sess_msgs');
await writeWire(f.sessionDir('sess_msgs'), [
{ type: 'metadata', protocol_version: '1.4', created_at: 1 },
{ type: 'context.append_message', message: userMessage('one') },
{ type: 'context.append_message', message: userMessage('two') },
]);
const snap = await f.reader.read('sess_msgs');
expect(snap.messages.items).toHaveLength(2);
expect(snap.messages.items.map((m) => (m.content[0] as { text: string }).text)).toEqual([
'one',
'two',
]);
});
it('reduces context.apply_compaction to [summary, ...tail]', async () => {
const f = await makeFixtureAsync();
await seedSession(f, 'sess_compact');
await writeWire(f.sessionDir('sess_compact'), [
{ type: 'context.append_message', message: userMessage('old-1') },
{ type: 'context.append_message', message: userMessage('old-2') },
{
type: 'context.apply_compaction',
count: 2,
summary: { role: 'user', content: [{ type: 'text', text: 'summary' }], toolCalls: [] },
},
{ type: 'context.append_message', message: userMessage('after') },
]);
const snap = await f.reader.read('sess_compact');
const texts = snap.messages.items.map((m) => (m.content[0] as { text: string }).text);
expect(texts).toEqual(['summary', 'after']);
});
it('honors context.clear and context.undo', async () => {
const f = await makeFixtureAsync();
await seedSession(f, 'sess_ops');
await writeWire(f.sessionDir('sess_ops'), [
{ type: 'context.append_message', message: userMessage('a') },
{ type: 'context.append_message', message: userMessage('b') },
{ type: 'context.clear' },
{ type: 'context.append_message', message: userMessage('c') },
]);
expect((await f.reader.read('sess_ops')).messages.items.map((m) => (m.content[0] as { text: string }).text)).toEqual(['c']);
});
it('caps the page at 100 and flags has_more', async () => {
const f = await makeFixtureAsync();
await seedSession(f, 'sess_paged');
await writeWire(
f.sessionDir('sess_paged'),
Array.from({ length: 150 }, (_, i) => ({
type: 'context.append_message' as const,
message: userMessage(`m${i}`),
})),
);
const snap = await f.reader.read('sess_paged');
expect(snap.messages.items).toHaveLength(100);
expect(snap.messages.has_more).toBe(true);
expect((snap.messages.items[0]!.content[0] as { text: string }).text).toBe('m50');
expect((snap.messages.items.at(-1)!.content[0] as { text: string }).text).toBe('m149');
});
it('normalizes a v1-layout state.json (ISO timestamps, no id)', async () => {
const f = await makeFixtureAsync();
await seedSession(f, 'sess_v1', {
rawState: {
title: 'v1 session',
createdAt: '2026-06-01T10:00:00.000Z',
updatedAt: '2026-06-01T11:00:00.000Z',
archived: false,
},
});
const snap = await f.reader.read('sess_v1');
expect(snap.session.id).toBe('sess_v1');
expect(snap.session.title).toBe('v1 session');
expect(Number.isNaN(Date.parse(snap.session.created_at))).toBe(false);
});
it('serves repeated reads from the (size, mtime) cache', async () => {
const f = await makeFixtureAsync();
await seedSession(f, 'sess_cache');
await writeWire(f.sessionDir('sess_cache'), [
{ type: 'context.append_message', message: userMessage('cached') },
]);
const first = await f.reader.read('sess_cache');
expect(first.messages.items).toHaveLength(1);
// Rewrite with identical content (size + mtime may change) — the cache is
// keyed on (size, mtime); a same-size rewrite keeps serving the cached
// reduction only when mtime is unchanged, so just assert stability.
const second = await f.reader.read('sess_cache');
expect(second.messages.items.map((m) => (m.content[0] as { text: string }).text)).toEqual([
'cached',
]);
});
it('invalidates the cache when the wire shrinks (compaction rewrite)', async () => {
const f = await makeFixtureAsync();
await seedSession(f, 'sess_shrink');
await writeWire(f.sessionDir('sess_shrink'), [
{ type: 'context.append_message', message: userMessage('a') },
{ type: 'context.append_message', message: userMessage('b') },
{ type: 'context.append_message', message: userMessage('c') },
]);
expect((await f.reader.read('sess_shrink')).messages.items).toHaveLength(3);
await new Promise((r) => setTimeout(r, 20));
await writeWire(f.sessionDir('sess_shrink'), [
{ type: 'context.append_message', message: userMessage('only-one') },
]);
const snap = await f.reader.read('sess_shrink');
expect(snap.messages.items).toHaveLength(1);
expect((snap.messages.items[0]!.content[0] as { text: string }).text).toBe('only-one');
});
});
// ─── pure helpers ─────────────────────────────────────────────────────────
describe('reduceContextRecords', () => {
it('ignores unknown records and the metadata envelope', () => {
const out = reduceContextRecords([
{ type: 'metadata', protocol_version: '1.4' },
{ type: 'turn.started', turnId: 1 },
{ type: 'context.append_message', message: userMessage('x') },
]);
expect(out).toHaveLength(1);
});
it('applies context.splice', () => {
const out = reduceContextRecords([
{ type: 'context.append_message', message: userMessage('a') },
{ type: 'context.splice', start: 1, deleteCount: 0, messages: [userMessage('b')] },
]);
expect(out).toHaveLength(2);
expect((out[1]!.content[0] as { text: string }).text).toBe('b');
});
});
describe('readWireRecords', () => {
it('drops a torn final line but throws on mid-file corruption', async () => {
const dir = await mkdtemp(join(tmpdir(), 'kimi-wire-'));
tmpDirs.push(dir);
const p = join(dir, 'wire.jsonl');
await writeFile(
p,
'{"type":"context.append_message","message":{"role":"user","content":[],"toolCalls":[]}}\n{"type":"context.append_message","message":{"role":"user","cont',
'utf-8',
);
const records = await readWireRecords(p);
expect(records).toHaveLength(1);
const bad = join(dir, 'bad.jsonl');
await writeFile(bad, '{not-json}\n{"type":"context.append_message"}\n', 'utf-8');
await expect(readWireRecords(bad)).rejects.toThrow(/corrupted line 1/);
});
});
describe('loadSnapshotConfig', () => {
it('defaults to auto / 4000ms / 32', () => {
const c = loadSnapshotConfig({});
expect(c).toEqual({ mode: 'auto', timeoutMs: 4000, cacheLimit: 32 });
});
it('parses legacy mode and integer knobs with floors', () => {
const c = loadSnapshotConfig({
KIMI_SNAPSHOT_READER: 'legacy',
KIMI_SNAPSHOT_TIMEOUT_MS: '2500',
KIMI_SNAPSHOT_CACHE_LIMIT: '0', // below min → default
});
expect(c.mode).toBe('legacy');
expect(c.timeoutMs).toBe(2500);
expect(c.cacheLimit).toBe(32);
});
it('falls back on non-numeric / sub-minimum timeout', () => {
expect(loadSnapshotConfig({ KIMI_SNAPSHOT_TIMEOUT_MS: 'abc' }).timeoutMs).toBe(4000);
expect(loadSnapshotConfig({ KIMI_SNAPSHOT_TIMEOUT_MS: '50' }).timeoutMs).toBe(4000);
});
});

View file

@ -13,7 +13,7 @@ import {
} from '@moonshot-ai/protocol';
import { z } from 'zod';
import { DEFAULT_MAX_UPLOAD_BYTES, FileNotFoundError, FileTooLargeError, IFileService, type IInstantiationService } from '@moonshot-ai/agent-core';
import { DEFAULT_MAX_UPLOAD_BYTES, FileNotFoundError, FileTooLargeError, IFileStore, type IInstantiationService } from '@moonshot-ai/agent-core';
import { errEnvelope, okEnvelope } from '../envelope';
import { defineRoute } from '../middleware/defineRoute';
@ -119,7 +119,7 @@ export function registerFilesRoutes(
const nameOverride = readFieldString(part.fields['name']);
const expiresInSec = readFieldNumber(part.fields['expires_in_sec']);
const store = ix.invokeFunction((a) => a.get(IFileService));
const store = ix.invokeFunction((a) => a.get(IFileStore));
const partFile = part.file as NodeJS.ReadableStream & { truncated?: boolean };
let busboyTruncated = false;
@ -178,7 +178,7 @@ export function registerFilesRoutes(
async (req, reply) => {
try {
const { file_id } = req.params;
const store = ix.invokeFunction((a) => a.get(IFileService));
const store = ix.invokeFunction((a) => a.get(IFileStore));
const { meta, blobPath } = await store.get(file_id);
const r = reply as unknown as FilesReply;
const size = meta.size;
@ -226,7 +226,7 @@ export function registerFilesRoutes(
async (req, reply) => {
try {
const { file_id } = req.params;
const store = ix.invokeFunction((a) => a.get(IFileService));
const store = ix.invokeFunction((a) => a.get(IFileStore));
await store.delete(file_id);
reply.send(okEnvelope({ deleted: true as const }, req.id));
} catch (err) {

View file

@ -563,7 +563,7 @@ export async function startServer(opts: ServerStartOptions): Promise<RunningServ
};
wsGw.setFsWatchHandler(fsWatchHandler);
a.get(IFileService);
a.get(IFileStore);
a.get(IWorkspaceRegistry);

3
pnpm-lock.yaml generated
View file

@ -96,6 +96,9 @@ importers:
'@moonshot-ai/server':
specifier: workspace:^
version: link:../../packages/server
'@moonshot-ai/server-v2':
specifier: workspace:^
version: link:../../packages/server-v2
'@moonshot-ai/vis-server':
specifier: workspace:^
version: link:../vis/server