From 590eca5ccdfe5289a3ebfd01cef883cc0627f2c1 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Mon, 6 Jul 2026 21:51:52 +0800 Subject: [PATCH] feat(server-v2): wire server-v2 into CLI and close v1 parity gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- apps/kimi-code/package.json | 1 + .../kimi-code/scripts/native/check-bundle.mjs | 2 +- apps/kimi-code/src/cli/sub/server/run.ts | 126 ++++-- apps/kimi-code/test/cli/server/server.test.ts | 23 ++ apps/kimi-code/tsdown.config.ts | 3 +- apps/kimi-code/tsdown.native.config.ts | 3 +- build/hash-imports-plugin.mjs | 84 ++++ .../src/agent/plan/planService.ts | 2 +- .../src/app/sessionIndex/sessionIndex.ts | 9 + .../app/sessionIndex/sessionIndexService.ts | 21 + .../src/app/sessionLegacy/sessionLegacy.ts | 8 +- .../app/sessionLegacy/sessionLegacyService.ts | 141 ++++++- .../sessionMetadata/sessionMetadata.ts | 9 + .../sessionMetadata/sessionMetadataService.ts | 24 +- .../test/sessionIndex/sessionIndex.test.ts | 13 + packages/server-v2/src/main.ts | 10 +- packages/server-v2/src/routes/meta.ts | 6 + .../src/routes/registerApiV1Routes.ts | 10 + packages/server-v2/src/routes/sessions.ts | 310 +++++++++----- packages/server-v2/src/routes/skills.ts | 147 ++++++- packages/server-v2/src/routes/snapshot.ts | 207 ++++++---- packages/server-v2/src/routes/webAssets.ts | 131 ++++++ .../server-v2/src/services/snapshot/index.ts | 11 + .../src/services/snapshot/snapshot.ts | 38 ++ .../src/services/snapshot/snapshotConfig.ts | 37 ++ .../src/services/snapshot/snapshotReader.ts | 379 ++++++++++++++++++ packages/server-v2/src/start.ts | 37 ++ .../ws/v1/sessionEventBroadcaster.ts | 95 ++++- .../server-v2/test/disableAuth.e2e.test.ts | 126 ++++++ packages/server-v2/test/sessions.test.ts | 240 +++++++++++ packages/server-v2/test/skills.test.ts | 87 +++- packages/server-v2/test/snapshot.test.ts | 160 +++++++- .../test/snapshotReader.unit.test.ts | 350 ++++++++++++++++ packages/server/src/routes/files.ts | 8 +- packages/server/src/start.ts | 2 +- pnpm-lock.yaml | 3 + 36 files changed, 2595 insertions(+), 268 deletions(-) create mode 100644 build/hash-imports-plugin.mjs create mode 100644 packages/server-v2/src/routes/webAssets.ts create mode 100644 packages/server-v2/src/services/snapshot/index.ts create mode 100644 packages/server-v2/src/services/snapshot/snapshot.ts create mode 100644 packages/server-v2/src/services/snapshot/snapshotConfig.ts create mode 100644 packages/server-v2/src/services/snapshot/snapshotReader.ts create mode 100644 packages/server-v2/test/disableAuth.e2e.test.ts create mode 100644 packages/server-v2/test/snapshotReader.unit.test.ts diff --git a/apps/kimi-code/package.json b/apps/kimi-code/package.json index 70accd50a..805d85e8f 100644 --- a/apps/kimi-code/package.json +++ b/apps/kimi-code/package.json @@ -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", diff --git a/apps/kimi-code/scripts/native/check-bundle.mjs b/apps/kimi-code/scripts/native/check-bundle.mjs index 39fd70b45..3cd10c278 100644 --- a/apps/kimi-code/scripts/native/check-bundle.mjs +++ b/apps/kimi-code/scripts/native/check-bundle.mjs @@ -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(/(?> = 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; +} + 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 }); diff --git a/apps/kimi-code/test/cli/server/server.test.ts b/apps/kimi-code/test/cli/server/server.test.ts index b6ee71f95..e0962ac52 100644 --- a/apps/kimi-code/test/cli/server/server.test.ts +++ b/apps/kimi-code/test/cli/server/server.test.ts @@ -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'); diff --git a/apps/kimi-code/tsdown.config.ts b/apps/kimi-code/tsdown.config.ts index 858aeeb48..af6a58927 100644 --- a/apps/kimi-code/tsdown.config.ts +++ b/apps/kimi-code/tsdown.config.ts @@ -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'), }, diff --git a/apps/kimi-code/tsdown.native.config.ts b/apps/kimi-code/tsdown.native.config.ts index c1008cb61..622dd5fd9 100644 --- a/apps/kimi-code/tsdown.native.config.ts +++ b/apps/kimi-code/tsdown.native.config.ts @@ -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'), }, diff --git a/build/hash-imports-plugin.mjs b/build/hash-imports-plugin.mjs new file mode 100644 index 000000000..94ddf6897 --- /dev/null +++ b/build/hash-imports-plugin.mjs @@ -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//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; + }, + }; +} diff --git a/packages/agent-core-v2/src/agent/plan/planService.ts b/packages/agent-core-v2/src/agent/plan/planService.ts index 5d9f2f86d..5fb35fe64 100644 --- a/packages/agent-core-v2/src/agent/plan/planService.ts +++ b/packages/agent-core-v2/src/agent/plan/planService.ts @@ -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; } diff --git a/packages/agent-core-v2/src/app/sessionIndex/sessionIndex.ts b/packages/agent-core-v2/src/app/sessionIndex/sessionIndex.ts index 39be692d9..f05d55d41 100644 --- a/packages/agent-core-v2/src/app/sessionIndex/sessionIndex.ts +++ b/packages/agent-core-v2/src/app/sessionIndex/sessionIndex.ts @@ -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; diff --git a/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts b/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts index 05f01db70..6d51e816b 100644 --- a/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts +++ b/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts @@ -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 | 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)['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']), diff --git a/packages/agent-core-v2/src/app/sessionLegacy/sessionLegacy.ts b/packages/agent-core-v2/src/app/sessionLegacy/sessionLegacy.ts index ce9a6b36d..53fa750f2 100644 --- a/packages/agent-core-v2/src/app/sessionLegacy/sessionLegacy.ts +++ b/packages/agent-core-v2/src/app/sessionLegacy/sessionLegacy.ts @@ -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; fork(sessionId: string, body: ForkSessionRequest): Promise; createChild(sessionId: string, body: CreateSessionChildRequest): Promise; listChildren(sessionId: string, query: SessionChildrenQuery): Promise; diff --git a/packages/agent-core-v2/src/app/sessionLegacy/sessionLegacyService.ts b/packages/agent-core-v2/src/app/sessionLegacy/sessionLegacyService.ts index 1a09e7c77..4e8e9b6ab 100644 --- a/packages/agent-core-v2/src/app/sessionLegacy/sessionLegacyService.ts +++ b/packages/agent-core-v2/src/app/sessionLegacy/sessionLegacyService.ts @@ -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 { + 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) } }); + } + + 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 { const handle = await this.lifecycle.fork({ sourceSessionId: sessionId, @@ -78,12 +128,11 @@ export class SessionLegacyService implements ISessionLegacyService { metadata: body.metadata as Record | 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 { - 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, + ): Promise { + 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 { const session = this.lifecycle.get(sessionId); if (session === undefined) { diff --git a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadata.ts b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadata.ts index 5c0fbfd6a..6952198b7 100644 --- a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadata.ts +++ b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadata.ts @@ -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>; diff --git a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts index efb79dc99..fdda33059 100644 --- a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts +++ b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts @@ -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); diff --git a/packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts b/packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts index 1fc277141..0f97ec37e 100644 --- a/packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts +++ b/packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts @@ -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 }); diff --git a/packages/server-v2/src/main.ts b/packages/server-v2/src/main.ts index 60d07b088..a04eef8ac 100644 --- a/packages/server-v2/src/main.ts +++ b/packages/server-v2/src/main.ts @@ -3,7 +3,8 @@ * DI engine) in the foreground and blocks until SIGINT/SIGTERM. * * Flags: `--port ` (default 58627), `--host ` (default 127.0.0.1), - * `--log-level ` (default info). + * `--log-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 { @@ -59,6 +64,7 @@ async function main(): Promise { host: opts.host, port: opts.port, logLevel: opts.logLevel ?? 'info', + disableAuth: opts.disableAuth, }); const origin = `http://${server.host}:${server.port}`; diff --git a/packages/server-v2/src/routes/meta.ts b/packages/server-v2/src/routes/meta.ts index 7bcccec1f..98fa8cb73 100644 --- a/packages/server-v2/src/routes/meta.ts +++ b/packages/server-v2/src/routes/meta.ts @@ -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( diff --git a/packages/server-v2/src/routes/registerApiV1Routes.ts b/packages/server-v2/src/routes/registerApiV1Routes.ts index 750ff7250..4320cb4c6 100644 --- a/packages/server-v2/src/routes/registerApiV1Routes.ts +++ b/packages/server-v2/src/routes/registerApiV1Routes.ts @@ -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[0], core); @@ -134,6 +143,7 @@ export async function registerApiV1Routes( registerSnapshotRoutes(apiV1 as unknown as Parameters[0], { core, broadcaster: opts.broadcaster, + reader: opts.snapshotReader, }); if (opts.enableShutdown !== false) { registerShutdownRoutes(apiV1 as unknown as Parameters[0], { diff --git a/packages/server-v2/src/routes/sessions.ts b/packages/server-v2/src/routes/sessions.ts index b0e5123ac..518108e9a 100644 --- a/packages/server-v2/src/routes/sessions.ts +++ b/packages/server-v2/src/routes/sessions.ts @@ -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; } -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)); diff --git a/packages/server-v2/src/routes/skills.ts b/packages/server-v2/src/routes/skills.ts index 8ccd9eb86..720c2c506 100644 --- a/packages/server-v2/src/routes/skills.ts +++ b/packages/server-v2/src/routes/skills.ts @@ -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[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[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 { + 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). // --------------------------------------------------------------------------- diff --git a/packages/server-v2/src/routes/snapshot.ts b/packages/server-v2/src/routes/snapshot.ts index 6ea09db33..a6cac5803 100644 --- a/packages/server-v2/src/routes/snapshot.ts +++ b/packages/server-v2/src/routes/snapshot.ts @@ -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[2]); } +async function readViaReader( + reader: ISnapshotReader, + sid: string, + timeoutMs: number, +): Promise { + let timer: NodeJS.Timeout | undefined; + const timeoutPromise = new Promise((_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 { + // 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 { diff --git a/packages/server-v2/src/routes/webAssets.ts b/packages/server-v2/src/routes/webAssets.ts new file mode 100644 index 000000000..7b7ad0176 --- /dev/null +++ b/packages/server-v2/src/routes/webAssets.ts @@ -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; +} + +export async function registerWebAssetRoutes( + app: WebAssetRouteHost, + assetsDir: string, +): Promise { + 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 { + 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 { + 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 { + 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'; + } +} diff --git a/packages/server-v2/src/services/snapshot/index.ts b/packages/server-v2/src/services/snapshot/index.ts new file mode 100644 index 000000000..9f25a48a2 --- /dev/null +++ b/packages/server-v2/src/services/snapshot/index.ts @@ -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'; diff --git a/packages/server-v2/src/services/snapshot/snapshot.ts b/packages/server-v2/src/services/snapshot/snapshot.ts new file mode 100644 index 000000000..b37eb1561 --- /dev/null +++ b/packages/server-v2/src/services/snapshot/snapshot.ts @@ -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; +} + +/** 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; + } +} diff --git a/packages/server-v2/src/services/snapshot/snapshotConfig.ts b/packages/server-v2/src/services/snapshot/snapshotConfig.ts new file mode 100644 index 000000000..20a56ca14 --- /dev/null +++ b/packages/server-v2/src/services/snapshot/snapshotConfig.ts @@ -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), + }; +} diff --git a/packages/server-v2/src/services/snapshot/snapshotReader.ts b/packages/server-v2/src/services/snapshot/snapshotReader.ts new file mode 100644 index 000000000..25c52b406 --- /dev/null +++ b/packages/server-v2/src/services/snapshot/snapshotReader.ts @@ -0,0 +1,379 @@ +/** + * `SnapshotReader` — server-layer disk reader for `GET /sessions/{sid}/snapshot` + * (`KIMI_SNAPSHOT_READER=auto`, the default). + * + * Reads `/sessions///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, 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(); + + constructor(private readonly deps: SnapshotReaderDeps) {} + + async read(sid: string): Promise { + 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 { + 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 { + 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, + ): SessionStatus { + if (live === undefined) return 'idle'; + return live.accessor.get(ISessionActivity).status(); + } + + private attachCurrentPromptId( + sid: string, + live: ReturnType, + 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, + ): { approvals: ReturnType[]; questions: ReturnType[] } { + 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:;` media URLs from `/blobs/`. Mirrors v1; unresolvable refs become `[media missing]`. */ + private async rehydrateBlobRefs(messages: readonly ContextMessage[], blobsDir: string): Promise { + const cache = new Map(); + for (const message of messages) { + for (const part of message.content) { + for (const value of Object.values(part as unknown as Record)) { + 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): 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 { + 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, +): Promise { + 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; +} \ No newline at end of file diff --git a/packages/server-v2/src/start.ts b/packages/server-v2/src/start.ts index 6aa393376..7d2df9e8a 100644 --- a/packages/server-v2/src/start.ts +++ b/packages/server-v2/src/start.ts @@ -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 { @@ -285,6 +312,8 @@ export async function startServer(opts: ServerStartOptions = {}): Promise { 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 { 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 { 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 { @@ -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 | undefined { + if (typeof payload !== 'object' || payload === null) return undefined; + const candidate = payload as Partial; + 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 }; +} diff --git a/packages/server-v2/test/disableAuth.e2e.test.ts b/packages/server-v2/test/disableAuth.e2e.test.ts new file mode 100644 index 000000000..cbd4e7d89 --- /dev/null +++ b/packages/server-v2/test/disableAuth.e2e.test.ts @@ -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); + }); +}); diff --git a/packages/server-v2/test/sessions.test.ts b/packages/server-v2/test/sessions.test.ts index 7681b4722..4350b6179 100644 --- a/packages/server-v2/test/sessions.test.ts +++ b/packages/server-v2/test/sessions.test.ts @@ -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 }; } + async function deleteJson(path: string): Promise<{ status: number; body: Envelope }> { + 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 }; + } + it('creates a session from metadata.cwd', async () => { const cwd = home as string; const { status, body } = await postJson('/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('/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('/api/v1/sessions', { metadata: { cwd } }); + const b = await postJson('/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('/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('/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('/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( + '/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('/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('/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('/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('/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('/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('/api/v1/sessions', { metadata: { cwd } }); + const parentId = parent.body.data.id; + const child = await postJson(`/api/v1/sessions/${parentId}/children`, {}); + const childId = child.body.data.id; + expect(child.body.data.status).toBe('idle'); + + const idle = await getJson(`/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(`/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('/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('/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(`/api/v1/sessions/${id}`); + expect(got.body.code).toBe(0); + expect(got.body.data.metadata.cwd).toBe(cwd); + + const profile = await getJson(`/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('/api/v1/sessions', { metadata: { cwd } }); + const id = created.body.data.id; + + const first = await postJson(`/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(`/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('/api/v1/sessions', { metadata: { cwd } }); + const id = created.body.data.id; + + await postJson(`/api/v1/sessions/${id}/profile`, { metadata: { foo: 'bar' } }); + const second = await postJson(`/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('/api/v1/sessions', { metadata: { cwd } }); + const id = created.body.data.id; + + const first = await postJson(`/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(`/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('/api/v1/sessions', { metadata: { cwd } }); + const id = created.body.data.id; + + const first = await postJson(`/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(`/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('/api/v1/sessions', { metadata: { cwd } }); + const id = created.body.data.id; + + const first = await postJson(`/api/v1/sessions/${id}/profile`, { + agent_config: { goal_objective: 'ship the feature' }, + }); + expect(first.body.code).toBe(0); + + const dup = await postJson(`/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('/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(`/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('/api/v1/sessions/sess_missing_profile/profile', { + title: 'nope', + }); + expect(body.code).toBe(40401); + }); }); diff --git a/packages/server-v2/test/skills.test.ts b/packages/server-v2/test/skills.test.ts index 442353a14..a8cda5459 100644 --- a/packages/server-v2/test/skills.test.ts +++ b/packages/server-v2/test/skills.test.ts @@ -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 }; } - async function createSession(): Promise { + async function createSession(cwd: string = home as string): Promise { 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 { + 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 { + 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 `/.kimi-code/skills//SKILL.md`. */ + async function seedProjectSkill(root: string, name: string): Promise { + 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('/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( + '/api/v1/workspaces/wd_does-not-exist_000000000000/skills', + ); + expect(body.code).toBe(40410); + }); + }); }); diff --git a/packages/server-v2/test/snapshot.test.ts b/packages/server-v2/test/snapshot.test.ts index 342b29cd4..1b690db14 100644 --- a/packages/server-v2/test/snapshot.test.ts +++ b/packages/server-v2/test/snapshot.test.ts @@ -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) | 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, + ) { + const previous = new Map(); + 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) + | 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(() => {}), // 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 diff --git a/packages/server-v2/test/snapshotReader.unit.test.ts b/packages/server-v2/test/snapshotReader.unit.test.ts new file mode 100644 index 000000000..8970b7130 --- /dev/null +++ b/packages/server-v2/test/snapshotReader.unit.test.ts @@ -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) { + const services = new Map(entries); + return { + get(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; + reader: SnapshotReader; + broadcaster: { seq: number; epoch: string; inFlightTurn: unknown }; +} + +const tmpDirs: string[] = []; + +async function makeFixtureAsync(opts?: { cacheLimit?: number }): Promise { + const homeDir = await mkdtemp(join(tmpdir(), 'kimi-snapshot-reader-')); + tmpDirs.push(homeDir); + const workspaceId = 'wd_unittest_012345abcdef'; + const index = new Map(); + 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 }, +): Promise { + 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): Promise { + 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); + }); +}); diff --git a/packages/server/src/routes/files.ts b/packages/server/src/routes/files.ts index fd64d8099..7cd7cc9fd 100644 --- a/packages/server/src/routes/files.ts +++ b/packages/server/src/routes/files.ts @@ -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) { diff --git a/packages/server/src/start.ts b/packages/server/src/start.ts index f3db7845a..f3cc94d90 100644 --- a/packages/server/src/start.ts +++ b/packages/server/src/start.ts @@ -563,7 +563,7 @@ export async function startServer(opts: ServerStartOptions): Promise