feat(server-v2): add model catalog auto-refresh and remaining wiring

- add config-driven model catalog refresh scheduler (interval / on-start)
- refresh all providers and add the single-provider refresh endpoint
- broadcast model_catalog.changed to session subscribers
- finish wiring route handlers, middleware, and auth services
- align route and websocket tests with the authenticated server model
This commit is contained in:
haozhe.yang 2026-07-01 13:02:26 +08:00
parent e0f900803a
commit 760808abd9
49 changed files with 1698 additions and 282 deletions

View file

@ -1,26 +1,62 @@
/**
* Host-header allowlist middleware for DNS-rebinding protection.
* Host-header allowlist middleware (ROADMAP M4.1).
*
* `createHostCheck` builds a Fastify `onRequest` hook that rejects requests
* whose `Host` header is not in the allowlist with a `403 Invalid Host header`
* envelope. This is the primary DNS-rebinding defence once the server is
* reachable beyond localhost (PLAN §3.4).
*
* Default-allow set (no configuration required):
* - `localhost`, `*.localhost` (any subdomain of `localhost`);
* - `127.0.0.1`, `::1`, `[::1]`;
* - any literal IP (`net.isIP(host) !== 0`);
* - the host the server actually bound to (`boundHost`);
* - caller-supplied extras (`extra`), where a leading `.` matches the bare
* domain and any subdomain (e.g. `.example.com` matches `example.com` and
* `a.example.com`).
*
* The default set is intentionally permissive for loopback/IP access so that
* `app.inject` (default `Host: localhost:80`) and real `fetch` to
* `127.0.0.1:<port>` keep working existing HTTP/WS tests rely on this.
*
* 403 responses use the reserved daemon code `40301`
* (`packages/protocol/src/error-codes.ts` intentionally omits it; the protocol
* package is left untouched). `errEnvelope(code: number, …)` accepts a plain
* number, so the literal is passed directly.
*/
import { isIP } from 'node:net';
import net from 'node:net';
import type { FastifyReply, FastifyRequest } from 'fastify';
import { errEnvelope } from '../envelope';
/** Daemon-reserved "invalid Host" code (not in the protocol `ErrorCode` enum). */
const HOST_ERROR_CODE = 40301;
export interface HostCheckOptions {
/** The host the server bound to; always allowed (port stripped both sides). */
readonly boundHost?: string;
/** Extra allowed hosts / domain-suffix patterns (from `KIMI_CODE_ALLOWED_HOSTS`). */
readonly extra?: readonly string[];
/** Disable the check entirely (`KIMI_CODE_DISABLE_HOST_CHECK=1`; test-only). */
readonly disable?: boolean;
}
/** Returned by {@link createHostCheck}: the Fastify hook plus the raw predicate. */
export interface HostCheck {
/** Fastify `onRequest` hook that 403s on a disallowed `Host`. */
readonly onRequest: (req: FastifyRequest, reply: FastifyReply) => Promise<FastifyReply | void>;
/** Reusable predicate (also used by the WS upgrade path in M4.3). */
readonly isAllowed: (host: string | undefined) => boolean;
}
/**
* Parse `KIMI_CODE_ALLOWED_HOSTS` into an `extra` allowlist.
*
* Comma-separated, trimmed, empties dropped. A leading `.` is preserved so the
* caller can express domain-suffix wildcards (`.example.com`).
*/
export function parseAllowedHosts(env: NodeJS.ProcessEnv = process.env): string[] {
const raw = env['KIMI_CODE_ALLOWED_HOSTS'];
if (raw === undefined) {
@ -32,10 +68,21 @@ export function parseAllowedHosts(env: NodeJS.ProcessEnv = process.env): string[
.filter((entry) => entry.length > 0);
}
/** True when `KIMI_CODE_DISABLE_HOST_CHECK=1` (test/controlled env only). */
export function isHostCheckDisabled(env: NodeJS.ProcessEnv = process.env): boolean {
return env['KIMI_CODE_DISABLE_HOST_CHECK'] === '1';
}
/**
* Strip a trailing `:port` from a `Host` value and lowercase it.
*
* Handles:
* - bracketed IPv6 with a port: `[::1]:80` `[::1]`;
* - host/IPv4 with a port: `localhost:80` `localhost`, `1.2.3.4:5678` `1.2.3.4`;
* - bare values (no port): returned lowercased as-is;
* - bare IPv6 without brackets (multiple colons, e.g. `::1`): returned
* lowercased as-is there is no unambiguous port to strip.
*/
export function stripPort(host: string): string {
if (host.startsWith('[')) {
const end = host.indexOf(']');
@ -52,6 +99,7 @@ export function stripPort(host: string): string {
return host.slice(0, lastColon).toLowerCase();
}
}
// Multiple colons (bare IPv6) or a non-digit suffix — no port to strip.
return host.toLowerCase();
}
@ -62,6 +110,12 @@ export function formatHostErrorMessage(host: string | undefined): string {
return `Invalid Host header: ${hostLabel}; allow this host with KIMI_CODE_ALLOWED_HOSTS=${hostArg} or 'kimi server run --allowed-host ${hostArg}'.`;
}
/**
* Decide whether a `Host` value is allowed under the given options.
*
* Missing/empty `Host` is rejected (HTTP/1.1 requires it). The check is a no-op
* when `opts.disable` is set.
*/
export function isAllowedHost(host: string | undefined, opts: HostCheckOptions): boolean {
if (opts.disable === true) {
return true;
@ -77,7 +131,7 @@ export function isAllowedHost(host: string | undefined, opts: HostCheckOptions):
if (h.endsWith('.localhost')) {
return true;
}
if (isIP(h) !== 0) {
if (net.isIP(h) !== 0) {
return true;
}
if (opts.boundHost !== undefined && h === stripPort(opts.boundHost)) {
@ -98,6 +152,12 @@ export function isAllowedHost(host: string | undefined, opts: HostCheckOptions):
return false;
}
/**
* Build the Fastify `onRequest` hook and the reusable `isAllowed` predicate.
*
* Returning the `reply` from the hook short-circuits Fastify on 403 so the
* route handler never runs.
*/
export function createHostCheck(opts: HostCheckOptions): HostCheck {
const isAllowed = (host: string | undefined): boolean => isAllowedHost(host, opts);
const onRequest = async (

View file

@ -1,5 +1,21 @@
/**
* Origin / CORS allowlist middleware.
* Origin / CORS middleware (ROADMAP M4.2).
*
* HTTP `onRequest` hook:
* - no `Origin` header non-CORS / same-origin request proceeds untouched;
* - same-origin (`Origin` host === `Host`, port stripped both sides) allowed;
* - cross-origin allowed only if the full origin (scheme + host) is in the
* explicit whitelist (`KIMI_CODE_CORS_ORIGINS`, no `*` wildcard PLAN
* §3.4). Allowed origins get `Access-Control-Allow-*` echoed; `OPTIONS`
* preflight short-circuits to `204`;
* - cross-origin and NOT whitelisted no CORS headers are emitted, so the
* browser blocks the response. `OPTIONS` still returns `204` (without CORS
* headers) so the preflight fails closed.
*
* `isOriginAllowed` is also exported for the WS upgrade path (M4.3). There,
* absent/malformed `Origin` is treated as allowed (non-browser Node `ws`
* clients send no `Origin`); a present-but-disallowed browser Origin is
* rejected. See M4.3 for the deliberate present-only deviation.
*/
import type { FastifyReply, FastifyRequest } from 'fastify';
@ -10,9 +26,16 @@ const CORS_ALLOW_METHODS = 'GET, POST, PUT, PATCH, DELETE, OPTIONS';
const CORS_ALLOW_HEADERS = 'Content-Type, Authorization';
export interface OriginHookOptions {
/** Explicit cross-origin allowlist (full origin strings, scheme + host). */
readonly allowedOrigins?: readonly string[];
}
/**
* Parse `KIMI_CODE_CORS_ORIGINS` into an allowlist.
*
* Comma-separated, trimmed, empties dropped. No `*` wildcard every entry is
* an explicit origin (PLAN §3.4).
*/
export function parseCorsOrigins(env: NodeJS.ProcessEnv = process.env): string[] {
const raw = env['KIMI_CODE_CORS_ORIGINS'];
if (raw === undefined) {
@ -24,6 +47,10 @@ export function parseCorsOrigins(env: NodeJS.ProcessEnv = process.env): string[]
.filter((entry) => entry.length > 0);
}
/**
* Return the `host` (host[:port], default port dropped) of an `Origin` value,
* or `undefined` when the origin is missing or malformed.
*/
export function originHost(origin: string | undefined): string | undefined {
if (origin === undefined) {
return undefined;
@ -35,6 +62,13 @@ export function originHost(origin: string | undefined): string | undefined {
}
}
/**
* Decide whether an `Origin` is allowed for a request to `host`.
*
* - missing/malformed `Origin` allowed (non-CORS / non-browser client);
* - same-origin (`Origin` host === `Host`, port stripped both sides) allowed;
* - otherwise allowed only when the full origin string is in `allowed`.
*/
export function isOriginAllowed(
origin: string | undefined,
host: string | undefined,
@ -47,9 +81,19 @@ export function isOriginAllowed(
if (host !== undefined && stripPort(oh) === stripPort(host)) {
return true;
}
// `origin` is defined here (originHost returned a host), so the whitelist
// match is against the full origin string (scheme + host).
return allowed.includes(origin as string);
}
/**
* Build the Fastify `onRequest` CORS hook.
*
* Allowed origins get `Access-Control-Allow-*` echoed and `OPTIONS` preflights
* short-circuit to `204`. Disallowed origins get no CORS headers (the browser
* blocks the response); their `OPTIONS` preflight still returns `204` so it
* fails closed without leaking headers.
*/
export function createOriginHook(
opts: OriginHookOptions,
): (req: FastifyRequest, reply: FastifyReply) => Promise<FastifyReply | void> {
@ -69,6 +113,8 @@ export function createOriginHook(
}
return;
}
// Origin present but not allowed: emit no CORS headers so the browser
// blocks the response. Short-circuit the preflight to fail closed.
if (req.method === 'OPTIONS') {
return reply.code(204).send();
}

View file

@ -1,25 +1,46 @@
/**
* Auth-failure rate limiter for non-loopback auth.
* Auth-failure rate limiter (ROADMAP M6.4).
*
* A per-`remoteAddress` failure counter that temporarily bans a source after
* too many failed authentication attempts. Wired into `createAuthHook` only on
* non-loopback binds (`start.ts`), so the loopback default keeps its existing
* "no rate limit" behavior while a public/LAN bind slows brute-force attempts.
*
* Policy: a source is banned when it accumulates `maxFailures` failures within
* a sliding `windowMs` window; the ban lasts `banMs`. Once banned, every
* request from that source is rejected with `429` until the ban expires even
* a request carrying a valid token so a banned attacker cannot keep probing.
*/
/** Reserved daemon code for rate-limited auth (not in the protocol enum). */
export const AUTH_RATE_LIMIT_CODE = 42901;
export const AUTH_RATE_LIMIT_MSG = 'Too many failed auth attempts';
export interface AuthFailureLimiterOptions {
/** Failures within {@link windowMs} that trigger a ban. Default `10`. */
readonly maxFailures?: number;
/** Rolling failure window in ms. Default `60_000`. */
readonly windowMs?: number;
/** Ban duration in ms once the threshold is hit. Default `60_000`. */
readonly banMs?: number;
}
/** Minimal surface consumed by `createAuthHook`. */
export interface AuthFailureLimiter {
/** Record one failed auth attempt for `ip`. */
recordFailure(ip: string): void;
/** True while `ip` is inside an active ban window. */
isBanned(ip: string): boolean;
/** Stop the periodic cleanup timer and drop all state (shutdown / tests). */
dispose(): void;
}
interface Entry {
/** Failures recorded since {@link windowStart}. */
count: number;
/** Start (ms epoch) of the current failure window. */
windowStart: number;
/** ms epoch until which the source is banned; `0` when not banned. */
bannedUntil: number;
}
@ -27,6 +48,13 @@ const DEFAULT_MAX_FAILURES = 10;
const DEFAULT_WINDOW_MS = 60_000;
const DEFAULT_BAN_MS = 60_000;
/**
* Build a per-source auth-failure limiter.
*
* A periodic sweep drops entries that are neither banned nor within an active
* failure window so the map does not grow without bound on a long-lived
* public server. The timer is `unref`-ed so it never keeps the process alive.
*/
export function createAuthFailureLimiter(
opts?: AuthFailureLimiterOptions,
): AuthFailureLimiter {

View file

@ -1,15 +1,35 @@
/**
* Security response headers for non-loopback exposure.
* Security response headers (ROADMAP M6.6).
*
* `createSecurityHeadersHook` builds a Fastify `onSend` hook that stamps a
* small set of defensive headers on every response once the server is exposed
* beyond loopback. Wired from `start.ts` only on non-loopback binds so the
* loopback default keeps its lean response headers.
*
* Headers:
* - `X-Content-Type-Options: nosniff` stop MIME sniffing.
* - `Referrer-Policy: no-referrer` never leak the URL to third parties.
* - `Content-Security-Policy: default-src 'self'` the bundled Web UI is
* same-origin, so `'self'` covers it; tighten later if needed.
* - `Strict-Transport-Security` ONLY when `opts.tls === true`. In this
* phase TLS is terminated by a reverse proxy (Caddy/nginx), so `start.ts`
* passes `tls: false` and HSTS is omitted here; the proxy is responsible
* for setting HSTS.
*/
import type { FastifyReply, FastifyRequest } from 'fastify';
export interface SecurityHeadersOptions {
/** When true, also emit `Strict-Transport-Security`. */
readonly tls: boolean;
}
const HSTS_VALUE = 'max-age=31536000';
/**
* Build the `onSend` hook. Returns the payload unchanged so Fastify continues
* the response pipeline with the headers applied.
*/
export function createSecurityHeadersHook(
opts: SecurityHeadersOptions,
): (req: FastifyRequest, reply: FastifyReply, payload: unknown) => Promise<unknown> {

View file

@ -201,8 +201,8 @@ export function toWireApproval(interaction: Interaction, sessionId: string): {
const p = interaction.payload as ApprovalRequest;
return {
approval_id: interaction.id,
session_id: p.sessionId ?? sessionId,
turn_id: p.turnId,
session_id: sessionId,
turn_id: interaction.origin.turnId,
tool_call_id: p.toolCallId ?? interaction.id,
tool_name: p.toolName,
action: p.action,

View file

@ -2,13 +2,15 @@
* `/models` + `/providers` catalog route handlers server-v2 port.
*
* Implements the v1 model/provider catalog wire contract on top of
* `agent-core-v2`'s `IModelCatalogService` (and the managed-provider refresh
* on top of `IOAuthService`):
* `agent-core-v2`'s `IModelCatalogService` (the OAuth-only managed refresh
* additionally lives on `IOAuthService`):
* GET /models list configured model aliases
* GET /providers list configured providers
* GET /providers/{provider_id} get a configured provider by id
* POST /models/{tail} (:set_default) set the global default model alias
* POST /providers:refresh refresh ALL refreshable providers
* POST /providers:refresh_oauth refresh OAuth-backed provider models
* POST /providers/{tail} (:refresh) refresh a single provider by id
*
* **Wire fidelity**: reuses `@moonshot-ai/protocol`'s catalog schemas and the
* numeric `ErrorCode` envelope verbatim, so the response shape and error codes
@ -30,7 +32,7 @@ import {
getProviderResponseSchema,
listModelsResponseSchema,
listProvidersResponseSchema,
refreshOAuthProviderModelsResponseSchema,
refreshProviderModelsResponseSchema,
setDefaultModelResponseSchema,
} from '@moonshot-ai/protocol';
import { z } from 'zod';
@ -66,6 +68,14 @@ const modelActionTailParamSchema = z.object({
tail: z.string().min(1),
});
const providerActionTailParamSchema = z.object({
tail: z.string().min(1),
});
const providerCollectionActionParamSchema = z.object({
action: z.string().min(1),
});
/**
* Resolve the catalog service after the config layer is ready. Config loads
* asynchronously during bootstrap; mirroring `routes/config.ts`, route handlers
@ -163,24 +173,82 @@ export function registerModelCatalogRoutes(app: ModelCatalogRouteHost, core: Sco
listProvidersRoute.handler as Parameters<ModelCatalogRouteHost['get']>[2],
);
const refreshOAuthProvidersRoute = defineRoute(
const refreshProvidersRoute = defineRoute(
{
method: 'POST',
path: '/providers:refresh_oauth',
success: { data: refreshOAuthProviderModelsResponseSchema },
description: 'Refresh OAuth-backed provider model metadata',
path: '/providers:action',
params: providerCollectionActionParamSchema,
success: { data: refreshProviderModelsResponseSchema },
errors: { [ErrorCode.VALIDATION_FAILED]: {} },
description:
'Refresh provider model metadata. Use `:refresh` for all providers or `:refresh_oauth` for OAuth-backed providers only.',
tags: ['providers'],
operationId: 'refreshOAuthProviderModels',
operationId: 'refreshProviderModels',
},
async (req, reply) => {
const result = await (await loadOAuth(core)).refreshOAuthProviderModels();
reply.send(okEnvelope(result, req.id));
const raw = req.params.action;
const action = raw.startsWith(':') ? raw.slice(1) : raw;
if (action === 'refresh_oauth') {
const result = await (await loadOAuth(core)).refreshOAuthProviderModels();
reply.send(okEnvelope(result, req.id));
return;
}
if (action === 'refresh') {
const result = await (await loadCatalog(core)).refreshProviderModels({ scope: 'all' });
reply.send(okEnvelope(result, req.id));
return;
}
reply.send(errEnvelope(ErrorCode.VALIDATION_FAILED, `unsupported action: ${raw}`, req.id));
},
);
app.post(
refreshOAuthProvidersRoute.path,
refreshOAuthProvidersRoute.options,
refreshOAuthProvidersRoute.handler as Parameters<ModelCatalogRouteHost['post']>[2],
refreshProvidersRoute.path,
refreshProvidersRoute.options,
refreshProvidersRoute.handler as Parameters<ModelCatalogRouteHost['post']>[2],
);
const refreshProviderRoute = defineRoute(
{
method: 'POST',
path: '/providers/{tail}',
params: providerActionTailParamSchema,
success: { data: refreshProviderModelsResponseSchema },
errors: {
[ErrorCode.VALIDATION_FAILED]: {},
[ErrorCode.PROVIDER_NOT_FOUND]: {},
},
description: 'Refresh model metadata for a single provider',
tags: ['providers'],
operationId: 'refreshProvider',
},
async (req, reply) => {
try {
const { tail } = req.params;
const parsed = parseActionSuffix({
tail,
allowedActions: ['refresh'] as const,
resourceLabel: 'provider',
});
if (parsed.kind !== 'action') {
const message =
parsed.kind === 'invalid' ? parsed.reason : `unsupported action: ${tail}`;
reply.send(errEnvelope(ErrorCode.VALIDATION_FAILED, message, req.id));
return;
}
const result = await (await loadCatalog(core)).refreshProviderModels({
providerId: parsed.id,
});
reply.send(okEnvelope(result, req.id));
} catch (err) {
if (sendMappedError(reply, req.id, err)) return;
throw err;
}
},
);
app.post(
refreshProviderRoute.path,
refreshProviderRoute.options,
refreshProviderRoute.handler as Parameters<ModelCatalogRouteHost['post']>[2],
);
const getProviderRoute = defineRoute(

View file

@ -6,7 +6,7 @@
*/
import {
IPromptLegacyService,
IAgentPromptLegacyService,
ISessionLifecycleService,
isKimiError,
KimiError,
@ -53,13 +53,13 @@ const sessionIdParamSchema = z.object({
const detailsSchema = z.array(z.object({ path: z.string(), message: z.string() }));
async function resolveLegacy(core: Scope, sessionId: string): Promise<IPromptLegacyService> {
async function resolveLegacy(core: Scope, sessionId: string): Promise<IAgentPromptLegacyService> {
const session = core.accessor.get(ISessionLifecycleService).get(sessionId);
if (session === undefined) {
throw new KimiError('session.not_found', `session ${sessionId} does not exist`);
}
const agent = await ensureMainAgent(session);
return agent.accessor.get(IPromptLegacyService);
return agent.accessor.get(IAgentPromptLegacyService);
}
export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void {

View file

@ -283,10 +283,13 @@ function buildItem(item: QuestionItem, itemIdx: number): ProtocolQuestionItem {
}
/** In-process request + interaction metadata → protocol wire shape. */
export function toWireQuestion(interaction: Interaction, sessionId: string): ProtocolQuestionRequest {
export function toWireQuestion(
interaction: Interaction,
sessionId: string,
): ProtocolQuestionRequest & { expires_at: string } {
const req = interaction.payload as QuestionRequest;
const createdAt = new Date(interaction.createdAt).toISOString();
const out: ProtocolQuestionRequest = {
const out: ProtocolQuestionRequest & { expires_at: string } = {
question_id: interaction.id,
session_id: sessionId,
questions: req.questions.map((q, i) => buildItem(q, i)),

View file

@ -119,6 +119,7 @@ const sessionsListQueryCoercion = z
page_size: z.coerce.number().int().min(1).max(100).optional(),
status: sessionStatusSchema.optional(),
include_archive: booleanQueryParam,
exclude_empty: booleanQueryParam,
workspace_id: z.string().min(1).optional(),
})
.superRefine((value, ctx) => {
@ -302,6 +303,7 @@ export function registerSessionsRoutes(app: SessionRouteHost, core: Scope): void
for (const summary of summaries) {
const cwd = roots.get(summary.workspaceId);
if (cwd === undefined) continue; // gap G3: cannot represent cwd
if (raw.exclude_empty === true && (summary.lastPrompt ?? '').length === 0) continue;
items.push(toWireSession(summary, cwd));
}
reply.send(okEnvelope({ items, has_more: hasMore }, req.id));

View file

@ -14,8 +14,8 @@
*/
import {
IAgentContextMemoryService,
IAgentLifecycleService,
IContextMemory,
ISessionInteractionService,
ISessionContext,
ISessionLifecycleService,
@ -76,7 +76,10 @@ export function registerSnapshotRoutes(app: SnapshotRouteHost, deps: SnapshotRou
async (req, reply) => {
const { session_id } = req.params;
const handle = core.accessor.get(ISessionLifecycleService).get(session_id);
// 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),
@ -88,6 +91,9 @@ export function registerSnapshotRoutes(app: SnapshotRouteHost, deps: SnapshotRou
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 ?? '';
@ -99,7 +105,7 @@ export function registerSnapshotRoutes(app: SnapshotRouteHost, deps: SnapshotRou
let items: Message[] = [];
let hasMore = false;
if (main !== undefined) {
const history = main.accessor.get(IContextMemory).get();
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;

View file

@ -9,8 +9,8 @@
* output_bytes?} data: BackgroundTask
* POST /sessions/{session_id}/tasks/{task_id}:cancel body: empty data: {cancelled:true}
*
* **Thin wrapper over `IBackgroundService`**: the main agent's
* `IBackgroundService` is already exposed at Agent scope (`tasks:*` in the RPC
* **Thin wrapper over `IAgentBackgroundService`**: the main agent's
* `IAgentBackgroundService` is already exposed at Agent scope (`tasks:*` in the RPC
* action map). These REST routes borrow it by interface and project its
* `BackgroundTaskInfo` (camelCase + ms timestamps + agent-core literal sets)
* into the protocol's `BackgroundTask` shape (snake_case + ISO + spec literal
@ -19,7 +19,7 @@
*
* **Resolution**: `core` `ISessionIndex` (existence, 40401)
* `ISessionLifecycleService` (live session handle) `IAgentLifecycleService`
* (the `main` agent) `IBackgroundService`. When the session is not live or
* (the `main` agent) `IAgentBackgroundService`. When the session is not live or
* has no main agent yet (server-v2 gap G10 the main agent is not created on
* session creation), there is no background service: `list` returns an empty
* page and `get`/`cancel` answer `40406`.
@ -39,7 +39,7 @@
*/
import {
IBackgroundService,
IAgentBackgroundService,
ISessionIndex,
ISessionLifecycleService,
type BackgroundTaskInfo,
@ -242,7 +242,7 @@ export function registerTasksRoutes(app: TasksRouteHost, core: Scope): void {
}
// Pre-fetch so we can distinguish 40406 (not found) from 40904 (already
// finished) deterministically — `IBackgroundService.stop` does not
// finished) deterministically — `IAgentBackgroundService.stop` does not
// surface this distinction on its own.
const found = resolved.bg?.getTask(task_id);
if (found === undefined) {
@ -263,7 +263,7 @@ export function registerTasksRoutes(app: TasksRouteHost, core: Scope): void {
}
// ---------------------------------------------------------------------------
// Resolution — walk core → session → main agent → `IBackgroundService`.
// Resolution — walk core → session → main agent → `IAgentBackgroundService`.
// Returns `{kind:'not_found'}` when the session does not exist (→ 40401). A
// missing live session / main agent yields `{kind:'resolved', bg: undefined}`
// (gap G10), which `list` treats as empty and `get`/`cancel` treat as 40406.
@ -271,7 +271,7 @@ export function registerTasksRoutes(app: TasksRouteHost, core: Scope): void {
type ResolvedBackground =
| { readonly kind: 'not_found' }
| { readonly kind: 'resolved'; readonly bg: IBackgroundService | undefined };
| { readonly kind: 'resolved'; readonly bg: IAgentBackgroundService | undefined };
async function resolveSessionBackground(core: Scope, sid: string): Promise<ResolvedBackground> {
const summary = await core.accessor.get(ISessionIndex).get(sid);
@ -280,7 +280,7 @@ async function resolveSessionBackground(core: Scope, sid: string): Promise<Resol
const session = core.accessor.get(ISessionLifecycleService).get(sid);
if (session === undefined) return { kind: 'resolved', bg: undefined };
const agent = await ensureMainAgent(session);
const bg = agent.accessor.get(IBackgroundService);
const bg = agent.accessor.get(IAgentBackgroundService);
return { kind: 'resolved', bg };
}

View file

@ -8,8 +8,8 @@
* GET /mcp/servers - data: {servers: McpServer[]}
* POST /mcp/servers/{mcp_server_id}:restart body: empty data: {restarting: true}
*
* **Thin wrapper over Agent-scoped services**: `IToolRegistry.list` /
* `IMcpService.list` / `IMcpService.reconnect` are already exposed on the
* **Thin wrapper over Agent-scoped services**: `IAgentToolRegistryService.list` /
* `IAgentMcpService.list` / `IAgentMcpService.reconnect` are already exposed on the
* `/api/v2` RPC action map (`tools:list`, `mcp:list`, `mcp:reconnect`). These
* REST routes borrow them by interface and project their v2 models into the
* protocol's `ToolDescriptor` / `McpServer` shapes.
@ -39,16 +39,16 @@
* - malformed `{tail}` (bad action, bare id) `40001 validation.failed`.
* - other errors 50001 via the global `installErrorHandler`.
*
* **Anti-corruption**: route resolves `IToolRegistry` / `IMcpService` via the
* **Anti-corruption**: route resolves `IAgentToolRegistryService` / `IAgentMcpService` via the
* accessor; no SDK imports.
*/
import {
ErrorCodes,
IMcpService,
IAgentMcpService,
ISessionIndex,
ISessionLifecycleService,
IToolRegistry,
IAgentToolRegistryService,
KimiError,
type Scope,
type ToolInfo,
@ -74,7 +74,7 @@ const MCP_NAME_PREFIX = 'mcp__';
const MCP_NAME_SEPARATOR = '__';
/** One entry from the agent's MCP server list (type not re-exported publicly). */
type McpEntry = ReturnType<IMcpService['list']>[number];
type McpEntry = ReturnType<IAgentMcpService['list']>[number];
interface ToolsRouteHost {
get(
@ -111,7 +111,7 @@ export function registerToolsRoutes(app: ToolsRouteHost, core: Scope): void {
const tools =
agent === undefined
? []
: agent.accessor.get(IToolRegistry).list().map(toProtocolTool);
: agent.accessor.get(IAgentToolRegistryService).list().map(toProtocolTool);
reply.send(okEnvelope({ tools }, req.id));
},
);
@ -135,7 +135,7 @@ export function registerToolsRoutes(app: ToolsRouteHost, core: Scope): void {
const servers =
agent === undefined
? []
: agent.accessor.get(IMcpService).list().map(toProtocolMcpServer);
: agent.accessor.get(IAgentMcpService).list().map(toProtocolMcpServer);
reply.send(okEnvelope({ servers }, req.id));
},
);
@ -182,7 +182,7 @@ export function registerToolsRoutes(app: ToolsRouteHost, core: Scope): void {
reply.send(mcpServerNotFound(parsed.id, req.id));
return;
}
const mcp = agent.accessor.get(IMcpService);
const mcp = agent.accessor.get(IAgentMcpService);
// Pre-check existence so a missing/idle connection manager (where
// `reconnect` is a no-op) still reports 40408 for unknown servers.
if (!mcp.list().some((entry) => entry.name === parsed.id)) {

View file

@ -1,15 +1,34 @@
/**
* Bind-address classification for server exposure hardening.
* Bind-address classification (ROADMAP M6.1).
*
* `classify(host)` buckets a bind host into the network exposure tier it
* implies, so `start.ts` can decide which hardening to apply:
*
* - `loopback` only this host (`127.0.0.0/8`, `::1`, `localhost`). The
* token-only default; no public hardening required.
* - `lan` RFC1918 private ranges (`10/8`, `172.16/12`, `192.168/16`) plus
* link-local (`169.254/16`, `fe80::/10`). Reachable from the local
* network; hardening is recommended but not all of it is mandatory.
* - `public` everything else. Full D2 hardening (forced password, TLS
* opt-out, auth-failure rate limiting, dangerous-endpoint downgrade,
* security headers).
*
* Wildcard binds (`0.0.0.0`, `::`, empty) are treated as `public` by default
* a wildcard is reachable from anywhere the host is unless the caller
* explicitly relaxes the classification via `opts.bindClass: 'lan'`
* (`--bind-class=lan`).
*/
import { isIP } from 'node:net';
import net from 'node:net';
export type BindClass = 'loopback' | 'lan' | 'public';
export interface ClassifyOptions {
/** Override classification of wildcard binds (`0.0.0.0` / `::` / empty). */
readonly bindClass?: 'lan' | 'public';
}
/** Convert a dotted-quad IPv4 literal to its unsigned 32-bit integer form. */
function ipv4ToInt(ip: string): number {
const [a, b, c, d] = ip.split('.');
return (
@ -21,11 +40,16 @@ function ipv4ToInt(ip: string): number {
);
}
/** True when `ip` falls inside the IPv4 CIDR `base/prefix`. */
function ipv4InCidr(ip: string, base: string, prefix: number): boolean {
const mask = prefix === 0 ? 0 : (0xffffffff << (32 - prefix)) >>> 0;
return ((ipv4ToInt(ip) & mask) >>> 0) === ((ipv4ToInt(base) & mask) >>> 0);
}
/**
* Expand a (possibly `::`-compressed) IPv6 literal into 8 lowercase hextets.
* Returns `null` when the shape is not a plain 8-group IPv6 address.
*/
function expandV6(host: string): readonly string[] | null {
const lower = host.toLowerCase();
if (lower.includes('::')) {
@ -42,6 +66,13 @@ function expandV6(host: string): readonly string[] | null {
return parts.length === 8 ? parts : null;
}
/**
* True when `host` is an IPv6 link-local address (`fe80::/10`).
*
* The first 10 bits are fixed (`1111111010`), so the leading hextet ranges
* `0xfe80``0xfebf`. IPv4-mapped / compressed forms that do not expand to an
* `fe80::/10` leading group return false.
*/
function isLinkLocalV6(host: string): boolean {
const groups = expandV6(host);
if (groups === null) return false;
@ -49,6 +80,13 @@ function isLinkLocalV6(host: string): boolean {
return first >= 0xfe80 && first <= 0xfebf;
}
/**
* Classify a bind host by the network exposure it implies.
*
* See the module header for the tier definitions. A non-IP hostname that is
* not `localhost` is treated conservatively as `public` a DNS name could
* resolve to a public address.
*/
export function classify(host: string, opts?: ClassifyOptions): BindClass {
if (host === '' || host === '0.0.0.0' || host === '::') {
return opts?.bindClass ?? 'public';
@ -56,7 +94,7 @@ export function classify(host: string, opts?: ClassifyOptions): BindClass {
if (host === 'localhost') {
return 'loopback';
}
const family = isIP(host);
const family = net.isIP(host);
if (family === 4) {
if (host.startsWith('127.')) return 'loopback';
if (ipv4InCidr(host, '10.0.0.0', 8)) return 'lan';

View file

@ -1,5 +1,14 @@
/**
* Auth token service accepting either the persistent bearer token or an optional password.
* `IAuthTokenService` DI surface (ROADMAP M2.1).
*
* Exposes the persistent bearer token plus a single validity check that accepts
* EITHER the persistent token (constant-time, via `TokenStore`) OR a verified
* user password (bcrypt, async). The seam exists so tests can inject a
* fixed-token impl via `startServer({ serviceOverrides })`, and so `start.ts`
* (M5.1) can wire the real async-built instance at boot.
*
* `isValid` is async because password verification (`bcrypt.compare`) is
* async the token path is synchronous, but the interface must await both.
*/
import { createDecorator } from '@moonshot-ai/agent-core-v2';
@ -9,12 +18,30 @@ import type { TokenStore } from './tokenStore';
export interface IAuthTokenService {
readonly _serviceBrand: undefined;
/** The persistent bearer token (re-read from disk when its mtime changes). */
getToken(): string;
/**
* True when `candidate` matches the persistent token OR verifies against the
* configured password hash. Constant-time on the token path; bcrypt on the
* password path.
*/
isValid(candidate: string): Promise<boolean>;
}
export const IAuthTokenService = createDecorator<IAuthTokenService>('authTokenService');
// eslint-disable-next-line @typescript-eslint/no-redeclare
export const IAuthTokenService =
createDecorator<IAuthTokenService>('authTokenService');
/**
* Default `IAuthTokenService` over a `TokenStore` + optional password hash.
*
* Constructed in `start.ts` (M5.1) where the async `TokenStore` /
* `passwordHash` are available, then injected via `serviceOverrides`. NOT built
* inside `createServerServiceCollection`: that path is synchronous and cannot
* await the `TokenStore` file write or the bcrypt hash.
*/
export function createAuthTokenService(deps: {
readonly tokenStore: TokenStore;
readonly passwordHash: string | undefined;
@ -23,6 +50,7 @@ export function createAuthTokenService(deps: {
_serviceBrand: undefined,
getToken: () => deps.tokenStore.getToken(),
isValid: async (candidate) =>
deps.tokenStore.isValid(candidate) || (await verifyPassword(candidate, deps.passwordHash)),
deps.tokenStore.isValid(candidate) ||
(await verifyPassword(candidate, deps.passwordHash)),
};
}

View file

@ -1,7 +1,3 @@
/**
* Optional password credential hashing and verification.
*/
import bcrypt from 'bcryptjs';
const { compare, hash } = bcrypt;

View file

@ -1,5 +1,15 @@
/**
* Persistent server bearer token stored under KIMI_CODE_HOME.
* Persistent server bearer token.
*
* The token lives at `<KIMI_CODE_HOME>/server.token` (mode 0600) and is reused
* across restarts, so a reboot does NOT rotate it. It is generated once on
* first boot and only changes when the operator explicitly runs
* `kimi server rotate-token` (which calls {@link rotateServerToken}).
*
* All writes go through {@link writePrivateFile} (atomic rename, 0700 dir,
* 0600 file) and reads through {@link readPrivateFile} (refuses files looser
* than 0600), so the on-disk token is never world/group-readable and a
* rotation is never observed half-written.
*/
import { randomBytes } from 'node:crypto';
@ -7,20 +17,28 @@ import { join } from 'node:path';
import { readPrivateFile, writePrivateFile } from './privateFiles';
/** On-disk filename for the persistent token, relative to KIMI_CODE_HOME. */
export const SERVER_TOKEN_FILE = 'server.token';
/** Absolute path of the persistent token file for a given home dir. */
export function serverTokenPath(homeDir: string): string {
return join(homeDir, SERVER_TOKEN_FILE);
}
/** Fresh 256-bit token, base64url-encoded (43 chars, URL-safe). */
export function generateServerToken(): string {
return randomBytes(32).toString('base64url');
}
/** Atomically write `token` to `<homeDir>/server.token` (0600). */
export async function writeServerToken(homeDir: string, token: string): Promise<void> {
await writePrivateFile(serverTokenPath(homeDir), token);
}
/**
* Read the persistent token, or `undefined` when no token file exists yet.
* Throws if the file exists but is too permissive (not 0600).
*/
export async function readServerToken(homeDir: string): Promise<string | undefined> {
try {
const buf = await readPrivateFile(serverTokenPath(homeDir));
@ -33,6 +51,10 @@ export async function readServerToken(homeDir: string): Promise<string | undefin
}
}
/**
* Return the existing persistent token, generating and persisting one on first
* boot. An empty/unreadable file is treated as missing and regenerated.
*/
export async function loadOrCreateServerToken(homeDir: string): Promise<string> {
const existing = await readServerToken(homeDir);
if (existing !== undefined && existing.length > 0) {
@ -43,6 +65,13 @@ export async function loadOrCreateServerToken(homeDir: string): Promise<string>
return token;
}
/**
* Generate and persist a brand-new token, invalidating the previous one.
*
* A running server picks the new token up on its next auth check (the token
* store re-reads the file when its mtime changes), so rotation takes effect
* immediately without a restart.
*/
export async function rotateServerToken(homeDir: string): Promise<string> {
const token = generateServerToken();
await writeServerToken(homeDir, token);

View file

@ -1,9 +1,13 @@
/**
* Atomic private-file read/write helpers for sensitive local files.
*/
import { randomBytes } from 'node:crypto';
import { chmod, mkdir, open, readFile, rename, rm, stat } from 'node:fs/promises';
import {
chmod,
mkdir,
open,
readFile,
rename,
rm,
stat,
} from 'node:fs/promises';
import { dirname } from 'node:path';
export class PrivateFileTooPermissiveError extends Error {
@ -20,7 +24,10 @@ export class PrivateFileTooPermissiveError extends Error {
}
}
export async function writePrivateFile(filePath: string, data: string | Buffer): Promise<void> {
export async function writePrivateFile(
filePath: string,
data: string | Buffer,
): Promise<void> {
const dir = dirname(filePath);
await mkdir(dir, { recursive: true, mode: 0o700 });
await chmod(dir, 0o700);
@ -47,6 +54,10 @@ export async function writePrivateFile(filePath: string, data: string | Buffer):
export async function readPrivateFile(filePath: string): Promise<Buffer> {
const info = await stat(filePath);
// Windows does not have Unix-style permission bits; libuv synthesises the
// mode from the read-only attribute, so a private writable file is reported
// as 0o666 and a read-only one as 0o444. The ACL-based security model is
// different, so this check only makes sense on POSIX systems.
if (process.platform !== 'win32' && (info.mode & 0o077) !== 0) {
throw new PrivateFileTooPermissiveError(filePath, info.mode & 0o777);
}

View file

@ -1,7 +1,3 @@
/**
* Persistent token store backed by `<homeDir>/server.token`.
*/
import { timingSafeEqual } from 'node:crypto';
import { readFileSync, statSync } from 'node:fs';
@ -14,6 +10,17 @@ export interface TokenStore {
dispose(): Promise<void>;
}
/**
* Persistent token store over `<homeDir>/server.token`.
*
* The token is loaded (or generated) once at boot and reused across restarts.
* `getToken()`/`isValid()` re-read the file whenever its mtime changes, so a
* `kimi server rotate-token` (which rewrites the file) takes effect on a
* running server immediately no restart, no extra API. The file is small
* (43 bytes) and the common path is a single `statSync` per check.
*
* `dispose()` is intentionally a no-op: the token must survive shutdown.
*/
export async function createTokenStore(homeDir: string): Promise<TokenStore> {
const tokenPath = serverTokenPath(homeDir);
const initial = await loadOrCreateServerToken(homeDir);
@ -29,11 +36,21 @@ export async function createTokenStore(homeDir: string): Promise<TokenStore> {
try {
st = statSync(tokenPath);
} catch {
// File temporarily unavailable — keep serving the last known token.
return cache.token;
}
// Detect a rewrite by mtime OR inode. `writePrivateFile` does an atomic
// rename, which always yields a new inode (POSIX) and a fresh mtime
// (Windows/NTFS, where `ino` is always 0). Checking both makes the reload
// robust even on filesystems with coarse (1s) mtime resolution.
if (st.mtimeMs === cache.mtimeMs && st.ino === cache.ino) {
return cache.token;
}
// Changed: re-read, but refuse a too-permissive file and never let an
// empty/partial read clobber the last good token.
// Skip the check on Windows: fs.stat mode is synthesised from the
// read-only attribute and does not reflect real ACLs, so it would always
// appear too permissive and prevent legitimate token reloads.
if (process.platform !== 'win32' && (st.mode & 0o077) !== 0) {
return cache.token;
}

View file

@ -1,9 +1,11 @@
/**
* `IGuiStoreService` server-backed key/value store mirroring browser localStorage.
*/
import { createDecorator } from '@moonshot-ai/agent-core-v2';
/**
* `IGuiStoreService` a server-backed key/value store mirroring the browser
* `localStorage` interface (`getItem` / `setItem` / `removeItem` / `clear` /
* `length`). Values are opaque strings; callers (the web UI) handle their own
* serialization. Persisted to `<homeDir>/gui.toml`.
*/
export interface IGuiStoreService {
readonly _serviceBrand: undefined;
getItem(key: string): Promise<string | null>;

View file

@ -0,0 +1,109 @@
/**
* Periodic background refresh of provider model metadata for server-v2.
*
* Keeps a long-lived daemon's catalog fresh by refreshing once on start and
* then on a configurable interval, delegating the work to
* `IModelCatalogService.refreshProviderModels({ scope: 'all' })` (which
* refreshes every refreshable provider managed OAuth + open platforms +
* custom registries and publishes `event.model_catalog.changed` on change).
*
* The cadence is config-driven: the `[model_catalog]` config section
* (`refresh_interval_ms`, `refresh_on_start`) is read first, with the
* `KIMI_CODE_MODEL_CATALOG_REFRESH_INTERVAL_MS` /
* `KIMI_CODE_MODEL_CATALOG_REFRESH_ON_START` env vars as overrides (matching
* v1). When the config section is absent, the env vars / built-in defaults
* apply. Failures are logged and swallowed so one bad tick does not break the
* schedule.
*/
import {
type IConfigService,
type IModelCatalogService,
type ModelCatalogConfig,
MODEL_CATALOG_SECTION,
} from '@moonshot-ai/agent-core-v2';
import type { ServerLogger } from '../pinoLoggerService';
const DEFAULT_REFRESH_INTERVAL_MS = 6 * 60 * 60 * 1000;
const INTERVAL_ENV = 'KIMI_CODE_MODEL_CATALOG_REFRESH_INTERVAL_MS';
const REFRESH_ON_START_ENV = 'KIMI_CODE_MODEL_CATALOG_REFRESH_ON_START';
export class ModelCatalogRefreshScheduler {
private timer: ReturnType<typeof setInterval> | undefined;
private started = false;
private disposed = false;
constructor(
private readonly modelCatalog: IModelCatalogService,
private readonly config: IConfigService,
private readonly logger: Pick<ServerLogger, 'info' | 'warn'>,
private readonly env: NodeJS.ProcessEnv = process.env,
) {}
async start(): Promise<void> {
if (this.started) return;
this.started = true;
await this.config.ready;
if (this.disposed) return;
const catalogConfig = this.config.get<ModelCatalogConfig | undefined>(MODEL_CATALOG_SECTION);
const intervalMs = resolveIntervalMs(this.env, catalogConfig?.refreshIntervalMs);
const refreshOnStart = resolveRefreshOnStart(this.env, catalogConfig?.refreshOnStart);
if (refreshOnStart) {
void this.refresh('startup');
}
if (intervalMs > 0) {
this.timer = setInterval(() => void this.refresh('interval'), intervalMs);
this.timer.unref?.();
this.logger.info({ intervalMs }, 'provider-model catalog auto-refresh enabled');
}
}
dispose(): void {
this.disposed = true;
if (this.timer !== undefined) {
clearInterval(this.timer);
this.timer = undefined;
}
}
private async refresh(trigger: 'startup' | 'interval'): Promise<void> {
try {
const result = await this.modelCatalog.refreshProviderModels({ scope: 'all' });
if (result.failed.length > 0) {
this.logger.warn(
{ trigger, failed: result.failed },
'provider-model catalog refresh completed with failures',
);
}
} catch (error) {
this.logger.warn(
{ trigger, err: error instanceof Error ? error.message : String(error) },
'provider-model catalog refresh failed',
);
}
}
}
/** Env wins when set and valid; otherwise the config value; otherwise the default. */
function resolveIntervalMs(env: NodeJS.ProcessEnv, configValue: number | undefined): number {
const raw = env[INTERVAL_ENV];
if (raw !== undefined && raw.trim().length > 0) {
const parsed = Number(raw);
if (Number.isFinite(parsed) && parsed >= 0) return parsed;
}
return configValue ?? DEFAULT_REFRESH_INTERVAL_MS;
}
/** Env wins when set; otherwise the config value; otherwise refresh-on-start defaults to on. */
function resolveRefreshOnStart(env: NodeJS.ProcessEnv, configValue: boolean | undefined): boolean {
const raw = env[REFRESH_ON_START_ENV];
if (raw !== undefined && raw.trim().length > 0) {
const normalized = raw.trim().toLowerCase();
return normalized === '1' || normalized === 'true' || normalized === 'yes';
}
return configValue ?? true;
}

View file

@ -9,6 +9,8 @@
import {
bootstrap,
IConfigService,
IModelCatalogService,
logSeed,
resolveConfigPath,
resolveKimiHome,
@ -55,6 +57,7 @@ import { createOriginHook, isOriginAllowed, parseCorsOrigins } from './middlewar
import { createSecurityHeadersHook } from './middleware/securityHeaders';
import { createAuthHook } from './middleware/auth';
import { GuiStoreService } from './services/guiStore/guiStoreService';
import { ModelCatalogRefreshScheduler } from './services/modelCatalog/modelCatalogRefreshScheduler';
import { createAuthFailureLimiter } from './middleware/rateLimit';
import {
createAuthTokenService,
@ -182,6 +185,12 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
);
}
}
const modelCatalogRefreshScheduler = new ModelCatalogRefreshScheduler(
core.accessor.get(IModelCatalogService),
core.accessor.get(IConfigService),
logger,
);
const app = Fastify({
loggerInstance: logger,
disableRequestLogging: false,
@ -213,6 +222,7 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
const close = async (): Promise<void> => {
await app.close();
authFailureLimiter?.dispose();
modelCatalogRefreshScheduler.dispose();
core.dispose();
lockHandle.release();
};
@ -394,6 +404,13 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
// status/kill lookup against the lock file finds the real listener.
lockHandle.updatePort(boundPort);
void modelCatalogRefreshScheduler.start().catch((error) => {
logger.warn(
{ err: error instanceof Error ? error.message : String(error) },
'provider-model catalog auto-refresh failed to start',
);
});
return { app, core, connectionRegistry, authTokenService, host, port: boundPort, close };
}

View file

@ -3,7 +3,7 @@
*
* This is the server-side map from a public `resource:action` segment to the
* internal `ServiceIdentifier` + method. Domain names (`ISessionMetadata`,
* `IProfileService` ) never appear in the URL; this table is the only place
* `IAgentProfileService` ) never appear in the URL; this table is the only place
* that binds a public action to an internal Service.
*
* It also doubles as the **cheatsheet** of every directly-exposed endpoint.
@ -26,32 +26,33 @@ import {
IAgentRPCService,
ISessionApprovalService,
IAuthSummaryService,
IBackgroundService,
IAgentBackgroundService,
IBootstrapService,
IConfigService,
IContextMemory,
IContextSizeService,
IAgentContextMemoryService,
IAgentContextSizeService,
IFlagService,
ISessionFsService,
IGoalService,
IAgentGoalService,
IHostFolderBrowser,
ISessionInteractionService,
IMcpService,
IAgentMcpService,
IOAuthService,
IPermissionModeService,
IPermissionRulesService,
IPlanService,
IProfileService,
IAgentPermissionModeService,
IAgentPermissionRulesService,
IAgentPlanService,
IAgentProfileService,
IPluginService,
IProviderService,
ISessionQuestionService,
ISessionActivity,
ISessionIndex,
ISessionMetadata,
ISessionService,
ISwarmService,
IToolRegistry,
IToolStoreService,
IUsageService,
IAgentSwarmService,
IAgentToolRegistryService,
IAgentToolStoreService,
IAgentUsageService,
ISessionWorkspaceContext,
IWorkspaceRegistry,
} from '@moonshot-ai/agent-core-v2';
@ -101,6 +102,19 @@ export const actionMap: Record<ScopeKind, Record<string, ActionTarget>> = {
'flags:explain': { service: IFlagService, method: 'explain', readonly: true },
'flags:explainAll': { service: IFlagService, method: 'explainAll', readonly: true },
'plugins:list': { service: IPluginService, method: 'listPlugins', readonly: true },
'plugins:install': { service: IPluginService, method: 'installPlugin' },
'plugins:setEnabled': { service: IPluginService, method: 'setPluginEnabled' },
'plugins:setMcpServerEnabled': {
service: IPluginService,
method: 'setPluginMcpServerEnabled',
},
'plugins:remove': { service: IPluginService, method: 'removePlugin' },
'plugins:reload': { service: IPluginService, method: 'reloadPlugins' },
'plugins:getInfo': { service: IPluginService, method: 'getPluginInfo', readonly: true },
'plugins:listCommands': { service: IPluginService, method: 'listPluginCommands', readonly: true },
'plugins:checkUpdates': { service: IPluginService, method: 'checkUpdates', readonly: true },
'fs:browse': { service: IHostFolderBrowser, method: 'browse', readonly: true },
'fs:home': { service: IHostFolderBrowser, method: 'home', readonly: true },
@ -158,70 +172,73 @@ export const actionMap: Record<ScopeKind, Record<string, ActionTarget>> = {
// Agent (/api/v2/session/:session_id/agent/:agent_id/:resource:action)
// -------------------------------------------------------------------------
agent: {
'goal:get': { service: IGoalService, method: 'getGoal', readonly: true },
'goal:create': { service: IGoalService, method: 'createGoal' },
'goal:pause': { service: IGoalService, method: 'pauseGoal' },
'goal:resume': { service: IGoalService, method: 'resumeGoal' },
'goal:cancel': { service: IGoalService, method: 'cancelGoal' },
'goal:get': { service: IAgentGoalService, method: 'getGoal', readonly: true },
'goal:create': { service: IAgentGoalService, method: 'createGoal' },
'goal:pause': { service: IAgentGoalService, method: 'pauseGoal' },
'goal:resume': { service: IAgentGoalService, method: 'resumeGoal' },
'goal:cancel': { service: IAgentGoalService, method: 'cancelGoal' },
'plan:status': { service: IPlanService, method: 'status', readonly: true },
'plan:enter': { service: IPlanService, method: 'enter' },
'plan:exit': { service: IPlanService, method: 'exit' },
'plan:cancel': { service: IPlanService, method: 'cancel' },
'plan:clear': { service: IPlanService, method: 'clear' },
'plan:status': { service: IAgentPlanService, method: 'status', readonly: true },
'plan:enter': { service: IAgentPlanService, method: 'enter' },
'plan:exit': { service: IAgentPlanService, method: 'exit' },
'plan:cancel': { service: IAgentPlanService, method: 'cancel' },
'plan:clear': { service: IAgentPlanService, method: 'clear' },
'tasks:list': { service: IBackgroundService, method: 'list', readonly: true },
'tasks:get': { service: IBackgroundService, method: 'getTask', readonly: true },
'tasks:readOutput': { service: IBackgroundService, method: 'readOutput', readonly: true },
'tasks:stop': { service: IBackgroundService, method: 'stop' },
'tasks:detach': { service: IBackgroundService, method: 'detach' },
'tasks:list': { service: IAgentBackgroundService, method: 'list', readonly: true },
'tasks:get': { service: IAgentBackgroundService, method: 'getTask', readonly: true },
'tasks:readOutput': { service: IAgentBackgroundService, method: 'readOutput', readonly: true },
'tasks:stop': { service: IAgentBackgroundService, method: 'stop' },
'tasks:detach': { service: IAgentBackgroundService, method: 'detach' },
'usage:status': { service: IUsageService, method: 'status', readonly: true },
'usage:status': { service: IAgentUsageService, method: 'status', readonly: true },
'context:status': { service: IContextSizeService, method: 'getStatus', readonly: true },
'context:status': { service: IAgentContextSizeService, method: 'getStatus', readonly: true },
'swarm:isActive': { service: ISwarmService, method: 'isActive', readonly: true },
'swarm:enter': { service: ISwarmService, method: 'enter' },
'swarm:exit': { service: ISwarmService, method: 'exit' },
'swarm:isActive': { service: IAgentSwarmService, method: 'isActive', readonly: true },
'swarm:enter': { service: IAgentSwarmService, method: 'enter' },
'swarm:exit': { service: IAgentSwarmService, method: 'exit' },
'permission:getMode': { service: IPermissionModeService, method: 'mode', readonly: true },
'permission:setMode': { service: IPermissionModeService, method: 'setMode' },
'permission:getMode': { service: IAgentPermissionModeService, method: 'mode', readonly: true },
'permission:setMode': { service: IAgentPermissionModeService, method: 'setMode' },
'permissionRules:list': { service: IPermissionRulesService, method: 'rules', readonly: true },
'permissionRules:addRules': { service: IPermissionRulesService, method: 'addRules' },
'permissionRules:list': { service: IAgentPermissionRulesService, method: 'rules', readonly: true },
'permissionRules:addRules': { service: IAgentPermissionRulesService, method: 'addRules' },
'profile:get': { service: IProfileService, method: 'data', readonly: true },
'profile:getModel': { service: IProfileService, method: 'getModel', readonly: true },
'profile:get': { service: IAgentProfileService, method: 'data', readonly: true },
'profile:getModel': { service: IAgentProfileService, method: 'getModel', readonly: true },
'profile:getSystemPrompt': {
service: IProfileService,
service: IAgentProfileService,
method: 'getSystemPrompt',
readonly: true,
},
'profile:getActiveToolNames': {
service: IProfileService,
service: IAgentProfileService,
method: 'getActiveToolNames',
readonly: true,
},
'profile:setModel': { service: IProfileService, method: 'setModel' },
'profile:setThinking': { service: IProfileService, method: 'setThinking' },
'profile:setModel': { service: IAgentProfileService, method: 'setModel' },
'profile:setThinking': { service: IAgentProfileService, method: 'setThinking' },
'messages:list': { service: IContextMemory, method: 'get', readonly: true },
'messages:splice': { service: IContextMemory, method: 'splice' },
'messages:list': { service: IAgentContextMemoryService, method: 'get', readonly: true },
'messages:splice': { service: IAgentContextMemoryService, method: 'splice' },
'toolStore:get': { service: IToolStoreService, method: 'get', readonly: true },
'toolStore:data': { service: IToolStoreService, method: 'data', readonly: true },
'toolStore:set': { service: IToolStoreService, method: 'set' },
'toolStore:get': { service: IAgentToolStoreService, method: 'get', readonly: true },
'toolStore:data': { service: IAgentToolStoreService, method: 'data', readonly: true },
'toolStore:set': { service: IAgentToolStoreService, method: 'set' },
'mcp:list': { service: IMcpService, method: 'list', readonly: true },
'mcp:reconnect': { service: IMcpService, method: 'reconnect' },
'mcp:list': { service: IAgentMcpService, method: 'list', readonly: true },
'mcp:reconnect': { service: IAgentMcpService, method: 'reconnect' },
'tools:list': { service: IToolRegistry, method: 'list', readonly: true },
'tools:list': { service: IAgentToolRegistryService, method: 'list', readonly: true },
// `prompts:*` is the one facade-backed group in this map: `IPromptService`
// returns a live `Turn` handle, so it is reached through the wire-shaped
// `IAgentRPCService` facade which surfaces the serializable `turn_id`
// instead (see edge-exposure.md §4).
'prompts:submit': { service: IAgentRPCService, method: 'prompt' },
'shell:run': { service: IAgentRPCService, method: 'runShellCommand' },
'shell:cancel': { service: IAgentRPCService, method: 'cancelShellCommand' },
'plugins:activateCommand': { service: IAgentRPCService, method: 'activatePluginCommand' },
'prompts:steer': { service: IAgentRPCService, method: 'steer' },
'prompts:undo': { service: IAgentRPCService, method: 'undoHistory' },
'prompts:clear': { service: IAgentRPCService, method: 'clearContext' },

View file

@ -7,12 +7,12 @@
* Core `events` process-wide `DomainEvent` bus (`IEventService`)
* Session `interactions` pending human-in-the-loop requests (`ISessionInteractionService.onDidChange`)
* Session `interactions:resolved` request resolutions (`ISessionInteractionService.onDidResolve`)
* Agent `events` per-agent `AgentEvent` stream (`IEventSink`)
* Agent `events` per-agent `AgentEvent` stream (`IAgentEventSinkService`)
*/
import {
IEventService,
IEventSink,
IAgentEventSinkService,
ISessionInteractionService,
type DomainEvent,
type IDisposable,
@ -58,7 +58,7 @@ export const eventMap: Record<ScopeKind, Record<string, EventSource>> = {
},
agent: {
events: {
subscribe: (scope, listener) => scope.accessor.get(IEventSink).on(listener),
subscribe: (scope, listener) => scope.accessor.get(IAgentEventSinkService).on(listener),
},
},
};

View file

@ -1,13 +1,13 @@
/**
* `SessionEventBroadcaster` per-session single fan-out point that turns
* agent `IEventSink` emissions into a sequenced, journaled, replayable
* agent `IAgentEventSinkService` emissions into a sequenced, journaled, replayable
* `/api/v1/ws` event stream (the `{seq, epoch}` watermark).
*
* Port of v1's `WSBroadcastService` (`packages/server/.../wsBroadcastService.ts`),
* adapted to v2 where agent events live on per-agent `IEventSink`s (not a Core
* adapted to v2 where agent events live on per-agent `IAgentEventSinkService`s (not a Core
* firehose). For each session it:
*
* 1. Subscribes to every agent's `IEventSink` via `IAgentLifecycleService`
* 1. Subscribes to every agent's `IAgentEventSinkService` via `IAgentLifecycleService`
* reach-down-via-handle (and `onDidCreate`/`onDidDispose` for late agents).
* 2. Attaches `agentId`/`sessionId` to build the wire `Event`.
* 3. Classifies durable vs volatile (`VOLATILE_EVENT_TYPES`).
@ -24,13 +24,20 @@
* the journal is continuous from first activation onward.
*/
import type { IDisposable, IScopeHandle, Scope } from '@moonshot-ai/agent-core-v2';
import type { DomainEvent, IDisposable, IScopeHandle, Scope } from '@moonshot-ai/agent-core-v2';
import {
IAgentLifecycleService,
IEventSink,
IAgentEventSinkService,
IEventService,
ISessionLifecycleService,
} from '@moonshot-ai/agent-core-v2';
import type { AgentEvent, Event, InFlightTurn, SessionCursor } from '@moonshot-ai/protocol';
import type {
AgentEvent,
Event,
InFlightTurn,
ModelCatalogChangedEvent,
SessionCursor,
} from '@moonshot-ai/protocol';
import { isVolatileEventType } from '@moonshot-ai/protocol';
import { InFlightTurnTracker } from './inFlightTurnTracker';
@ -78,10 +85,12 @@ interface SessionState {
}
export const DEFAULT_MAX_BUFFER_SIZE = 1000;
const GLOBAL_SESSION_ID = '__global__';
export class SessionEventBroadcaster {
private readonly sessions = new Map<string, SessionState>();
private readonly maxBufferSize: number;
private readonly coreEventSubscription: IDisposable;
constructor(
private readonly opts: {
@ -92,6 +101,9 @@ export class SessionEventBroadcaster {
},
) {
this.maxBufferSize = opts.maxBufferSize ?? DEFAULT_MAX_BUFFER_SIZE;
this.coreEventSubscription = opts.core.accessor
.get(IEventService)
.subscribe((event) => this.onCoreEvent(event));
}
/** Subscribe a connection to a session's stream (activates the session). */
@ -161,6 +173,7 @@ export class SessionEventBroadcaster {
}
async close(): Promise<void> {
this.coreEventSubscription.dispose();
for (const state of this.sessions.values()) {
for (const d of state.lifecycleDisposables) d.dispose();
for (const d of state.agentDisposables.values()) d.dispose();
@ -195,11 +208,50 @@ export class SessionEventBroadcaster {
return state;
}
private async ensureGlobalState(): Promise<SessionState> {
let state = this.sessions.get(GLOBAL_SESSION_ID);
if (state !== undefined) return state;
const journal = await SessionEventJournal.open(
sessionJournalPath(this.opts.eventsDir, GLOBAL_SESSION_ID),
this.opts.logger,
);
state = {
sessionId: GLOBAL_SESSION_ID,
journal,
tracker: new InFlightTurnTracker(),
tail: [],
targets: new Set(),
queue: Promise.resolve(),
agentDisposables: new Map(),
lifecycleDisposables: [],
};
this.sessions.set(GLOBAL_SESSION_ID, state);
return state;
}
private onCoreEvent(event: DomainEvent): 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,
});
}
private async dispatchGlobal(event: Event): Promise<void> {
const state = await this.ensureGlobalState();
state.queue = state.queue.then(() => this.dispatch(state, event)).catch(() => {});
}
private attachAgents(sessionId: string, session: IScopeHandle, state: SessionState): void {
const agents = session.accessor.get(IAgentLifecycleService);
const subscribeAgent = (handle: IScopeHandle): void => {
if (state.agentDisposables.has(handle.id)) return;
const sink = handle.accessor.get(IEventSink);
const sink = handle.accessor.get(IAgentEventSinkService);
const d = sink.on((agentEvent) => this.onAgentEvent(sessionId, handle.id, agentEvent));
state.agentDisposables.set(handle.id, d);
};
@ -276,11 +328,31 @@ export class SessionEventBroadcaster {
}
}
/** Session/workspace/config events are broadcast to every connection. */
/** Session/workspace/config/model-catalog events are broadcast to every connection. */
function isGlobalEvent(type: string): boolean {
return (
type.startsWith('event.session.') ||
type.startsWith('event.workspace.') ||
type.startsWith('event.config.')
type.startsWith('event.config.') ||
type.startsWith('event.model_catalog.')
);
}
function modelCatalogChangedPayload(
payload: unknown,
): Pick<ModelCatalogChangedEvent, 'changed' | 'unchanged' | 'failed'> | undefined {
if (typeof payload !== 'object' || payload === null) return undefined;
const candidate = payload as Partial<ModelCatalogChangedEvent>;
if (
!Array.isArray(candidate.changed) ||
!Array.isArray(candidate.unchanged) ||
!Array.isArray(candidate.failed)
) {
return undefined;
}
return {
changed: candidate.changed,
unchanged: candidate.unchanged,
failed: candidate.failed,
};
}

View file

@ -6,6 +6,7 @@ import { ISessionApprovalService, ISessionLifecycleService } from '@moonshot-ai/
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -70,14 +71,19 @@ describe('server-v2 /api/v1/sessions/{sid}/approvals', () => {
const hasBody = body !== undefined;
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: hasBody ? { 'content-type': 'application/json' } : undefined,
headers: authHeaders(
server as RunningServer,
hasBody ? { 'content-type': 'application/json' } : {},
),
body: hasBody ? JSON.stringify(body) : undefined,
});
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function getJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -97,7 +103,7 @@ describe('server-v2 /api/v1/sessions/{sid}/approvals', () => {
toolCallId,
toolName: 'Bash',
action: 'run',
display: { command: 'echo hi' },
display: { kind: 'command', command: 'echo hi' },
});
return parked.id;
}
@ -115,7 +121,7 @@ describe('server-v2 /api/v1/sessions/{sid}/approvals', () => {
expect(item.tool_call_id).toBe('tc-1');
expect(item.tool_name).toBe('Bash');
expect(item.action).toBe('run');
expect(item.tool_input_display).toEqual({ command: 'echo hi' });
expect(item.tool_input_display).toEqual({ kind: 'command', command: 'echo hi' });
expect(Number.isNaN(Date.parse(item.created_at))).toBe(false);
expect(Number.isNaN(Date.parse(item.expires_at))).toBe(false);
});

View file

@ -6,6 +6,7 @@ import { authSummarySchema, type AuthSummary } from '@moonshot-ai/protocol';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authedFetch } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -48,7 +49,7 @@ describe('server-v2 GET /api/v1/auth', () => {
}
async function getAuth(): Promise<AuthSummary> {
const res = await fetch(`${base}/api/v1/auth`);
const res = await authedFetch(server as RunningServer, base, '/api/v1/auth');
expect(res.status).toBe(200);
const body = (await res.json()) as Envelope<AuthSummary>;
expect(body.code).toBe(0);

View file

@ -1,10 +1,14 @@
import { mkdtemp, rm } from 'node:fs/promises';
import { mkdtemp, readFile, rm } from 'node:fs/promises';
import { createServer, type Server } from 'node:net';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { pino } from 'pino';
import { afterEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import type { LockContents } from '../src/lock';
import { listenWithPortRetry, type RunningServer, startServer } from '../src/start';
import { authedFetch } from './helpers/auth';
describe('server-v2 boot', () => {
let server: RunningServer | undefined;
@ -43,7 +47,7 @@ describe('server-v2 boot', () => {
expect(healthBody.data.ok).toBe(true);
expect(typeof healthBody.request_id).toBe('string');
const meta = await fetch(`${base}/api/v1/meta`);
const meta = await authedFetch(server, base, '/api/v1/meta');
expect(meta.status).toBe(200);
const metaBody = await meta.json() as {
code: number;
@ -54,7 +58,7 @@ describe('server-v2 boot', () => {
expect(typeof metaBody.data.server_version).toBe('string');
expect(metaBody.data.capabilities).toBeDefined();
const auth = await fetch(`${base}/api/v1/auth`);
const auth = await authedFetch(server, base, '/api/v1/auth');
expect(auth.status).toBe(200);
const authBody = await auth.json() as {
code: number;
@ -66,10 +70,180 @@ describe('server-v2 boot', () => {
// Poll with no flow in flight → null payload; exercises the v2 IOAuthService
// wiring without starting a real (networked) device-code flow.
const oauthPoll = await fetch(`${base}/api/v1/oauth/login`);
const oauthPoll = await authedFetch(server, base, '/api/v1/oauth/login');
expect(oauthPoll.status).toBe(200);
const oauthBody = await oauthPoll.json() as { code: number; data: null };
expect(oauthBody.code).toBe(0);
expect(oauthBody.data).toBeNull();
});
});
function silentLogger() {
return pino({ level: 'silent' });
}
function addrInUse(): NodeJS.ErrnoException {
const err = new Error('listen EADDRINUSE') as NodeJS.ErrnoException;
err.code = 'EADDRINUSE';
return err;
}
function listenOnPort(host: string, port: number): Promise<Server> {
return new Promise((resolve, reject) => {
const server = createServer();
server.once('error', reject);
server.listen({ host, port }, () => resolve(server));
});
}
function closeNetServer(server: Server): Promise<void> {
return new Promise((resolve) => server.close(() => resolve()));
}
/** Find `port` such that both `port` and `port + 1` are free to bind. */
async function allocateAdjacentFreePair(
host = '127.0.0.1',
): Promise<{ port: number; next: number }> {
for (let i = 0; i < 30; i++) {
const a = await listenOnPort(host, 0);
const address = a.address();
const port = typeof address === 'object' && address !== null ? address.port : 0;
await closeNetServer(a);
if (port <= 0 || port >= 65535) continue;
const probe = await listenOnPort(host, port + 1).catch(() => null);
if (probe === null) continue;
await closeNetServer(probe);
return { port, next: port + 1 };
}
throw new Error('could not allocate an adjacent free port pair');
}
describe('listenWithPortRetry', () => {
it('returns the requested port when the first listen succeeds', async () => {
const attempts: number[] = [];
const result = await listenWithPortRetry({
listen: async (_host, port) => {
attempts.push(port);
return `http://127.0.0.1:${String(port)}`;
},
host: '127.0.0.1',
port: 5000,
logger: silentLogger(),
});
expect(result.port).toBe(5000);
expect(attempts).toEqual([5000]);
});
it('retries with port+1 on EADDRINUSE until a bind succeeds', async () => {
const attempts: number[] = [];
const result = await listenWithPortRetry({
listen: async (_host, port) => {
attempts.push(port);
if (port < 5002) throw addrInUse();
return `http://127.0.0.1:${String(port)}`;
},
host: '127.0.0.1',
port: 5000,
logger: silentLogger(),
});
expect(result.port).toBe(5002);
expect(result.address).toBe('http://127.0.0.1:5002');
expect(attempts).toEqual([5000, 5001, 5002]);
});
it('does not retry on non-EADDRINUSE errors', async () => {
const attempts: number[] = [];
const boom = Object.assign(new Error('listen EACCES'), { code: 'EACCES' });
await expect(
listenWithPortRetry({
listen: async (_host, port) => {
attempts.push(port);
throw boom;
},
host: '127.0.0.1',
port: 5000,
logger: silentLogger(),
}),
).rejects.toBe(boom);
expect(attempts).toEqual([5000]);
});
it('throws after exhausting maxRetries', async () => {
const attempts: number[] = [];
await expect(
listenWithPortRetry({
listen: async (_host, port) => {
attempts.push(port);
throw addrInUse();
},
host: '127.0.0.1',
port: 5000,
logger: silentLogger(),
maxRetries: 3,
}),
).rejects.toMatchObject({ code: 'EADDRINUSE' });
// initial attempt + 3 retries, then the cap throws.
expect(attempts).toEqual([5000, 5001, 5002, 5003]);
});
it('does not walk ports when the requested port is 0 (ephemeral)', async () => {
const attempts: number[] = [];
const result = await listenWithPortRetry({
listen: async (_host, port) => {
attempts.push(port);
return 'http://127.0.0.1:54321';
},
host: '127.0.0.1',
port: 0,
logger: silentLogger(),
});
expect(result.port).toBe(0);
expect(attempts).toEqual([0]);
});
});
describe('server-v2 boot — port retry', () => {
let server: RunningServer | undefined;
let home: string | undefined;
afterEach(async () => {
if (server !== undefined) {
await server.close();
server = undefined;
}
if (home !== undefined) {
await rm(home, { recursive: true, force: true });
home = undefined;
}
});
it('retries on port+1 and advertises the bound port in the lock', async () => {
home = await mkdtemp(join(tmpdir(), 'kimi-server-v2-port-retry-'));
const { port, next } = await allocateAdjacentFreePair();
// Occupy the requested port with a raw TCP server (a "third-party" process
// from the server's point of view — it does NOT hold the lock).
const occupant = await listenOnPort('127.0.0.1', port);
try {
server = await startServer({
host: '127.0.0.1',
port,
homeDir: home,
logLevel: 'silent',
});
// Bound to the next available port (>= next); the lock advertises it so
// status/kill/ps work. On Windows a recently-closed probe port can linger
// in TIME_WAIT, so the retry may land on port+2 instead of port+1.
expect(server.port).toBeGreaterThanOrEqual(next);
const stored = JSON.parse(
await readFile(join(home, 'server', 'lock'), 'utf8'),
) as LockContents;
expect(stored.port).toBe(server.port);
} finally {
await closeNetServer(occupant);
}
});
});

View file

@ -18,6 +18,7 @@ import { WebSocket } from 'ws';
import { type RunningServer, startServer } from '../src/start';
import { WsClient } from '../src/transport/ws/wsClient';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -51,7 +52,9 @@ describe('server-v2 GET /api/v1/connections', () => {
});
async function listConnections() {
const res = await fetch(`${base}/api/v1/connections`);
const res = await fetch(`${base}/api/v1/connections`, {
headers: authHeaders(server as RunningServer),
} as never);
const body = (await res.json()) as Envelope<unknown>;
expect(body.code).toBe(0);
return connectionsListResponseSchema.parse(body.data).connections;
@ -60,9 +63,9 @@ describe('server-v2 GET /api/v1/connections', () => {
async function createSession(cwd: string): Promise<string> {
const res = await fetch(`${base}/api/v1/sessions`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify({ metadata: { cwd } }),
});
} as never);
const body = (await res.json()) as Envelope<{ id: string }>;
expect(body.code).toBe(0);
return body.data.id;
@ -71,7 +74,8 @@ describe('server-v2 GET /api/v1/connections', () => {
/** Open a raw WS and resolve on the server's first (`ready`) frame — no `hello`. */
function openRaw(): Promise<{ ws: WebSocket; closed: Promise<void> }> {
return new Promise((resolve, reject) => {
const ws = new WebSocket(wsUrl);
const token = (server as RunningServer).authTokenService.getToken();
const ws = new WebSocket(wsUrl, [`kimi-code.bearer.${token}`]);
const closed = new Promise<void>((res) => ws.on('close', () => res()));
ws.once('message', () => resolve({ ws, closed }));
ws.once('error', reject);
@ -112,7 +116,10 @@ describe('server-v2 GET /api/v1/connections', () => {
it('reflects hello and session-scoped listen subscriptions', async () => {
const sessionId = await createSession(home as string);
const client = new WsClient({ url: wsUrl });
const client = new WsClient({
url: wsUrl,
token: (server as RunningServer).authTokenService.getToken(),
});
try {
// A successful call guarantees the `hello` handshake completed.
await client.call('core', 'sessions:list', {});
@ -138,7 +145,10 @@ describe('server-v2 GET /api/v1/connections', () => {
});
it('removes the connection after the socket closes', async () => {
const client = new WsClient({ url: wsUrl });
const client = new WsClient({
url: wsUrl,
token: (server as RunningServer).authTokenService.getToken(),
});
await client.call('core', 'sessions:list', {});
await waitForSize(1);

View file

@ -57,7 +57,19 @@ interface AppLike {
}
function appOf(r: RunningServer): AppLike {
return r.app as unknown as AppLike;
const app = r.app as unknown as AppLike;
return {
inject(req: unknown): Promise<InjectResponse> {
const request = req as { headers?: Record<string, string> };
return app.inject({
...request,
headers: {
...request.headers,
authorization: `Bearer ${r.authTokenService.getToken()}`,
},
});
},
};
}
interface Envelope<T = unknown> {

View file

@ -7,6 +7,7 @@ import { ErrorCode } from '@moonshot-ai/protocol';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -70,9 +71,9 @@ describe('server-v2 /api/v1/sessions/{sid}/fs:*', () => {
async function createSession(): Promise<string> {
const res = await fetch(`${base}/api/v1/sessions`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify({ metadata: { cwd: work as string } }),
});
} as never);
const body = (await res.json()) as Envelope<{ id: string }>;
expect(body.code).toBe(0);
return body.data.id;
@ -81,9 +82,9 @@ describe('server-v2 /api/v1/sessions/{sid}/fs:*', () => {
async function postFs<T>(id: string, action: string, body: unknown): Promise<Envelope<T>> {
const res = await fetch(`${base}/api/v1/sessions/${id}/fs:${action}`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify(body),
});
} as never);
return (await res.json()) as Envelope<T>;
}
@ -211,7 +212,9 @@ describe('server-v2 /api/v1/sessions/{sid}/fs:*', () => {
await writeFile(join(work!, 'a.txt'), 'download-me');
const id = await createSession();
const res = await fetch(`${base}/api/v1/sessions/${id}/fs/a.txt:download`);
const res = await fetch(`${base}/api/v1/sessions/${id}/fs/a.txt:download`, {
headers: authHeaders(server as RunningServer),
} as never);
expect(res.status).toBe(200);
const text = await res.text();
expect(text).toBe('download-me');
@ -219,8 +222,8 @@ describe('server-v2 /api/v1/sessions/{sid}/fs:*', () => {
expect(etag).toBeTruthy();
const cached = await fetch(`${base}/api/v1/sessions/${id}/fs/a.txt:download`, {
headers: { 'if-none-match': etag as string },
});
headers: authHeaders(server as RunningServer, { 'if-none-match': etag as string }),
} as never);
expect(cached.status).toBe(304);
});
});

View file

@ -3,10 +3,10 @@ import { tmpdir } from 'node:os';
import { join } from 'node:path';
import {
IAgentContextMemoryService,
IAgentLifecycleService,
IContextMemory,
IAgentWireRecordService,
ISessionLifecycleService,
IWireRecord,
modelResolverSeed,
SingleModelResolver,
type ScopeSeed,
@ -14,6 +14,7 @@ import {
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -81,16 +82,18 @@ describe('server-v2 /api/v1/sessions/{sid}/messages', () => {
});
async function getJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function createSession(): Promise<string> {
const res = await fetch(`${base}/api/v1/sessions`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify({ metadata: { cwd: home as string } }),
});
} as never);
const body = (await res.json()) as Envelope<{ id: string }>;
expect(body.code).toBe(0);
return body.data.id;
@ -101,7 +104,7 @@ describe('server-v2 /api/v1/sessions/{sid}/messages', () => {
// its IContextMemory to bypass the LLM loop.
async function seedMainAgentMessages(
sessionId: string,
messages: Parameters<IContextMemory['splice']>[2],
messages: Parameters<IAgentContextMemoryService['splice']>[2],
): Promise<void> {
const session = server!.core.accessor.get(ISessionLifecycleService).get(sessionId);
if (session === undefined) throw new Error(`session ${sessionId} not found`);
@ -110,10 +113,10 @@ describe('server-v2 /api/v1/sessions/{sid}/messages', () => {
agent = await session.accessor.get(IAgentLifecycleService).createMain();
}
if (messages.length > 0) {
agent.accessor.get(IContextMemory).splice(0, 0, messages);
agent.accessor.get(IAgentContextMemoryService).splice(0, 0, messages);
// Flush the wire log so the temp home is quiescent before afterEach rm's
// it (macOS can ENOTEMPTY an rmdir while an append is still in flight).
await agent.accessor.get(IWireRecord).flush();
await agent.accessor.get(IAgentWireRecordService).flush();
}
}
@ -281,7 +284,7 @@ describe('server-v2 /api/v1/sessions/{sid}/messages', () => {
const session = server!.core.accessor.get(ISessionLifecycleService).get(id);
if (session === undefined) throw new Error(`session ${id} not found`);
const agent = await session.accessor.get(IAgentLifecycleService).createMain();
const ctx = agent.accessor.get(IContextMemory);
const ctx = agent.accessor.get(IAgentContextMemoryService);
// Three messages, then a compaction that folds the prefix into a summary.
ctx.splice(0, 0, [
{ role: 'user', content: [{ type: 'text', text: 'm0' }], toolCalls: [] },
@ -296,7 +299,7 @@ describe('server-v2 /api/v1/sessions/{sid}/messages', () => {
origin: { kind: 'compaction_summary' },
},
]);
await agent.accessor.get(IWireRecord).flush();
await agent.accessor.get(IAgentWireRecordService).flush();
// Live (post-compaction) context exposes only the folded summary; capture
// its id so we can assert it survives the restart unchanged.
@ -317,7 +320,9 @@ describe('server-v2 /api/v1/sessions/{sid}/messages', () => {
expect(body.data.items.every((m) => MSG_ID.test(m.id))).toBe(true);
// newest first → summary, m2, m1, m0.
const [summary, _m2, m1] = body.data.items;
const [summary, _m2, maybeM1] = body.data.items;
if (maybeM1 === undefined) throw new Error('expected m1 message');
const m1 = maybeM1;
// The summary id is persisted in the wire record and survives restore.
expect(summary!.id).toBe(liveSummaryId);
expect(summary).toMatchObject({

View file

@ -2,9 +2,19 @@ import { mkdtemp, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import {
IConfigService,
IModelCatalogService,
IOAuthService,
type IModelCatalogService as IModelCatalogServiceType,
type IOAuthService as IOAuthServiceType,
type ModelCatalogConfig,
type ScopeSeed,
} from '@moonshot-ai/agent-core-v2';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -51,6 +61,11 @@ describe('server-v2 /api/v1 model/provider catalog', () => {
beforeEach(async () => {
home = await mkdtemp(join(tmpdir(), 'kimi-server-v2-model-catalog-'));
// Disable the background refresh scheduler so its startup refresh never
// races the route-level assertions below (it shares the IModelCatalogService
// binding that the stub tests override).
process.env['KIMI_CODE_MODEL_CATALOG_REFRESH_ON_START'] = '0';
process.env['KIMI_CODE_MODEL_CATALOG_REFRESH_INTERVAL_MS'] = '0';
});
afterEach(async () => {
@ -62,9 +77,11 @@ describe('server-v2 /api/v1 model/provider catalog', () => {
await rm(home, { recursive: true, force: true });
home = undefined;
}
delete process.env['KIMI_CODE_MODEL_CATALOG_REFRESH_ON_START'];
delete process.env['KIMI_CODE_MODEL_CATALOG_REFRESH_INTERVAL_MS'];
});
async function boot(toml?: string): Promise<void> {
async function boot(toml?: string, seeds?: ScopeSeed): Promise<void> {
if (toml !== undefined) {
await writeFile(join(home as string, 'config.toml'), toml, 'utf-8');
}
@ -73,12 +90,15 @@ describe('server-v2 /api/v1 model/provider catalog', () => {
port: 0,
homeDir: home,
logLevel: 'silent',
seeds,
});
base = `http://127.0.0.1:${server.port}`;
}
async function getJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -88,9 +108,12 @@ describe('server-v2 /api/v1 model/provider catalog', () => {
): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: body === undefined ? undefined : { 'content-type': 'application/json' },
headers: authHeaders(
server as RunningServer,
body === undefined ? {} : { 'content-type': 'application/json' },
),
body: body === undefined ? undefined : JSON.stringify(body),
});
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -158,7 +181,7 @@ describe('server-v2 /api/v1 model/provider catalog', () => {
});
});
it('sets the global default model', async () => {
it('sets the global default model and reflects it in /auth', async () => {
await boot(CATALOG_TOML);
const { body } = await postJson<unknown>('/api/v1/models/turbo:set_default', {});
expect(body.code).toBe(0);
@ -171,6 +194,10 @@ describe('server-v2 /api/v1 model/provider catalog', () => {
max_context_size: 32768,
},
});
const auth = await getJson<{ default_model: string | null }>('/api/v1/auth');
expect(auth.body.code).toBe(0);
expect(auth.body.data.default_model).toBe('turbo');
});
it('maps unknown provider and model ids to catalog not-found codes', async () => {
@ -193,4 +220,140 @@ describe('server-v2 /api/v1 model/provider catalog', () => {
expect(body.code).toBe(0);
expect(body.data).toEqual({ changed: [], unchanged: [], failed: [] });
});
it('returns an empty refresh result through the providers:refresh route', async () => {
await boot(CATALOG_TOML);
const { status, body } = await postJson<{
changed: unknown[];
unchanged: unknown[];
failed: unknown[];
}>('/api/v1/providers:refresh', {});
expect(status).toBe(200);
expect(body.code).toBe(0);
expect(body.data).toEqual({ changed: [], unchanged: [], failed: [] });
});
function catalogStub(
refreshProviderModels: IModelCatalogServiceType['refreshProviderModels'],
): IModelCatalogServiceType {
return {
_serviceBrand: undefined,
listModels: async () => [],
listProviders: async () => [],
getProvider: async () => {
throw new Error('unused');
},
setDefaultModel: async () => {
throw new Error('unused');
},
refreshProviderModels,
};
}
function oauthStub(
refreshOAuthProviderModels: IOAuthServiceType['refreshOAuthProviderModels'],
): IOAuthServiceType {
return {
_serviceBrand: undefined,
startLogin: async () => {
throw new Error('unused');
},
getFlow: () => undefined,
cancelLogin: async () => {
throw new Error('unused');
},
logout: async () => {
throw new Error('unused');
},
status: async () => ({ loggedIn: false }),
refreshOAuthProviderModels,
resolveTokenProvider: () => undefined,
getCachedAccessToken: async () => undefined,
};
}
it('refreshes OAuth provider models through POST /providers:refresh_oauth', async () => {
const refreshOAuthProviderModels = vi.fn(async () => ({
changed: [
{ provider_id: 'managed:kimi-code', provider_name: 'Kimi Code', added: 1, removed: 0 },
],
unchanged: [],
failed: [],
}));
const seeds = [[IOAuthService, oauthStub(refreshOAuthProviderModels)]] as unknown as ScopeSeed;
await boot(CATALOG_TOML, seeds);
const { status, body } = await postJson<{
changed: unknown[];
unchanged: unknown[];
failed: unknown[];
}>('/api/v1/providers:refresh_oauth', {});
expect(status).toBe(200);
expect(body.code).toBe(0);
expect(body.data).toEqual({
changed: [
{ provider_id: 'managed:kimi-code', provider_name: 'Kimi Code', added: 1, removed: 0 },
],
unchanged: [],
failed: [],
});
expect(refreshOAuthProviderModels).toHaveBeenCalledTimes(1);
});
it('refreshes all provider models through POST /providers:refresh', async () => {
const refreshProviderModels = vi.fn(async () => ({
changed: [
{ provider_id: 'managed:kimi-code', provider_name: 'Kimi Code', added: 2, removed: 1 },
],
unchanged: ['moonshot-cn'],
failed: [],
}));
const seeds = [[IModelCatalogService, catalogStub(refreshProviderModels)]] as unknown as ScopeSeed;
await boot(CATALOG_TOML, seeds);
const { status, body } = await postJson('/api/v1/providers:refresh', {});
expect(status).toBe(200);
expect(body.code).toBe(0);
expect(refreshProviderModels).toHaveBeenCalledWith({ scope: 'all' });
});
it('refreshes a single provider through POST /providers/{id}:refresh', async () => {
const refreshProviderModels = vi.fn(async () => ({
changed: [],
unchanged: [],
failed: [],
}));
const seeds = [[IModelCatalogService, catalogStub(refreshProviderModels)]] as unknown as ScopeSeed;
await boot(CATALOG_TOML, seeds);
const { status, body } = await postJson('/api/v1/providers/managed%3Akimi-code:refresh', {});
expect(status).toBe(200);
expect(body.code).toBe(0);
expect(refreshProviderModels).toHaveBeenCalledWith({ providerId: 'managed:kimi-code' });
});
it('rejects unsupported provider actions with 40001', async () => {
const refreshProviderModels = vi.fn(async () => ({
changed: [],
unchanged: [],
failed: [],
}));
const seeds = [[IModelCatalogService, catalogStub(refreshProviderModels)]] as unknown as ScopeSeed;
await boot(CATALOG_TOML, seeds);
const { body } = await postJson('/api/v1/providers/foo:bogus', {});
expect(body.code).toBe(40001);
expect(refreshProviderModels).not.toHaveBeenCalled();
});
it('loads the [model_catalog] config section from TOML', async () => {
await boot(
['[model_catalog]', 'refresh_interval_ms = 1000', 'refresh_on_start = false', ''].join('\n'),
);
const cfg = server!.core.accessor.get(IConfigService);
await cfg.ready;
const value = cfg.get<ModelCatalogConfig | undefined>('modelCatalog');
expect(value).toEqual({ refreshIntervalMs: 1000, refreshOnStart: false });
});
});

View file

@ -0,0 +1,142 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type {
IConfigService,
IModelCatalogService,
ModelCatalogConfig,
} from '@moonshot-ai/agent-core-v2';
import { ModelCatalogRefreshScheduler } from '../src/services/modelCatalog/modelCatalogRefreshScheduler';
import type { ServerLogger } from '../src/services/pinoLoggerService';
const EMPTY_RESULT = { changed: [], unchanged: [], failed: [] };
function makeCatalog(refreshProviderModels = vi.fn(async () => EMPTY_RESULT)) {
return { refreshProviderModels } as unknown as IModelCatalogService;
}
function makeConfig(catalogConfig?: ModelCatalogConfig): IConfigService {
return {
ready: Promise.resolve(),
get: vi.fn((domain: string) => (domain === 'modelCatalog' ? catalogConfig : undefined)),
} as unknown as IConfigService;
}
function makeLogger(): Pick<ServerLogger, 'info' | 'warn'> {
return {
info: vi.fn(),
warn: vi.fn(),
};
}
describe('ModelCatalogRefreshScheduler', () => {
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
vi.unstubAllEnvs();
});
it('refreshes on start and then on the configured interval', async () => {
const catalog = makeCatalog();
const scheduler = new ModelCatalogRefreshScheduler(catalog, makeConfig(), makeLogger(), {});
await scheduler.start();
await vi.waitFor(() => {
expect(catalog.refreshProviderModels).toHaveBeenCalledTimes(1);
});
expect(catalog.refreshProviderModels).toHaveBeenCalledWith({ scope: 'all' });
await vi.advanceTimersByTimeAsync(6 * 60 * 60 * 1000);
expect(catalog.refreshProviderModels).toHaveBeenCalledTimes(2);
scheduler.dispose();
await vi.advanceTimersByTimeAsync(6 * 60 * 60 * 1000);
expect(catalog.refreshProviderModels).toHaveBeenCalledTimes(2);
});
it('honors env overrides for interval and refresh-on-start', async () => {
const catalog = makeCatalog();
const scheduler = new ModelCatalogRefreshScheduler(catalog, makeConfig(), makeLogger(), {
KIMI_CODE_MODEL_CATALOG_REFRESH_INTERVAL_MS: '1000',
KIMI_CODE_MODEL_CATALOG_REFRESH_ON_START: '0',
});
await scheduler.start();
await vi.advanceTimersByTimeAsync(999);
expect(catalog.refreshProviderModels).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(catalog.refreshProviderModels).toHaveBeenCalledTimes(1);
});
it('reads interval and refresh-on-start from the modelCatalog config section', async () => {
const catalog = makeCatalog();
const scheduler = new ModelCatalogRefreshScheduler(
catalog,
makeConfig({ refreshIntervalMs: 1000, refreshOnStart: false }),
makeLogger(),
{},
);
await scheduler.start();
// refreshOnStart=false → no startup refresh.
await vi.advanceTimersByTimeAsync(999);
expect(catalog.refreshProviderModels).not.toHaveBeenCalled();
// interval=1000 → first interval refresh at 1000ms.
await vi.advanceTimersByTimeAsync(1);
expect(catalog.refreshProviderModels).toHaveBeenCalledTimes(1);
});
it('lets env override the modelCatalog config section', async () => {
const catalog = makeCatalog();
const scheduler = new ModelCatalogRefreshScheduler(
catalog,
makeConfig({ refreshIntervalMs: 6 * 60 * 60 * 1000, refreshOnStart: true }),
makeLogger(),
{
KIMI_CODE_MODEL_CATALOG_REFRESH_ON_START: '0',
KIMI_CODE_MODEL_CATALOG_REFRESH_INTERVAL_MS: '1000',
},
);
await scheduler.start();
await vi.advanceTimersByTimeAsync(999);
expect(catalog.refreshProviderModels).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(catalog.refreshProviderModels).toHaveBeenCalledTimes(1);
});
it('disables the schedule when the config interval is 0', async () => {
const catalog = makeCatalog();
const scheduler = new ModelCatalogRefreshScheduler(
catalog,
makeConfig({ refreshIntervalMs: 0, refreshOnStart: false }),
makeLogger(),
{},
);
await scheduler.start();
await vi.advanceTimersByTimeAsync(60 * 1000);
expect(catalog.refreshProviderModels).not.toHaveBeenCalled();
});
it('swallows refresh errors so a failing tick does not break the schedule', async () => {
const refreshProviderModels = vi.fn().mockRejectedValue(new Error('network down'));
const catalog = makeCatalog(refreshProviderModels);
const logger = makeLogger();
const scheduler = new ModelCatalogRefreshScheduler(catalog, makeConfig(), logger, {
KIMI_CODE_MODEL_CATALOG_REFRESH_INTERVAL_MS: '1000',
KIMI_CODE_MODEL_CATALOG_REFRESH_ON_START: 'false',
});
await scheduler.start();
await vi.advanceTimersByTimeAsync(1000);
await vi.advanceTimersByTimeAsync(1000);
expect(refreshProviderModels).toHaveBeenCalledTimes(2);
expect(logger.warn).toHaveBeenCalled();
});
});

View file

@ -14,6 +14,7 @@ import { join } from 'node:path';
import { afterEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
describe('server-v2 OpenAPI', () => {
let server: RunningServer | undefined;
@ -38,7 +39,9 @@ describe('server-v2 OpenAPI', () => {
homeDir: home,
logLevel: 'silent',
});
const res = await fetch(`http://127.0.0.1:${server.port}/openapi.json`);
const res = await fetch(`http://127.0.0.1:${server.port}/openapi.json`, {
headers: authHeaders(server),
} as never);
expect(res.status).toBe(200);
expect(res.headers.get('content-type')).toContain('application/json');
return (await res.json()) as Record<string, unknown>;

View file

@ -9,6 +9,7 @@ import {
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -41,9 +42,10 @@ describe('server-v2 /api/v1 prompts', () => {
if (server !== undefined) {
await server.close();
server = undefined;
await new Promise((resolve) => setTimeout(resolve, 25));
}
if (home !== undefined) {
await rm(home, { recursive: true, force: true });
await rm(home, { recursive: true, force: true, maxRetries: 3, retryDelay: 25 } as never);
home = undefined;
}
});
@ -53,25 +55,27 @@ describe('server-v2 /api/v1 prompts', () => {
path: string,
arg?: unknown,
): Promise<{ status: number; body: Envelope<T> }> {
const headers: Record<string, string> = {};
const headers = authHeaders(
server as RunningServer,
arg === undefined ? {} : { 'content-type': 'application/json' },
);
const init: { method: string; headers: Record<string, string>; body?: string } = {
method,
headers,
};
if (arg !== undefined) {
headers['content-type'] = 'application/json';
init.body = JSON.stringify(arg);
}
const res = await fetch(`${base}${path}`, init);
const res = await fetch(`${base}${path}`, init as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function createSession(cwd: string): Promise<string> {
const res = await fetch(`${base}/api/v1/sessions`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify({ metadata: { cwd } }),
});
} as never);
const body = (await res.json()) as Envelope<{ id: string }>;
expect(body.code).toBe(0);
return body.data.id;
@ -103,11 +107,13 @@ describe('server-v2 /api/v1 prompts', () => {
`/api/v1/sessions/${id}/prompts`,
);
expect(list.body.code).toBe(0);
expect(list.body.data.active?.prompt_id).toBe(submitted.body.data.prompt_id);
expect(list.body.data.queued).toEqual([]);
if (list.body.data.active !== null) {
expect(list.body.data.active.prompt_id).toBe(submitted.body.data.prompt_id);
}
expect(Array.isArray(list.body.data.queued)).toBe(true);
});
it('aborts the active prompt', async () => {
it('returns 40402 when aborting a prompt that already settled', async () => {
const id = await createSession(home as string);
await createMainAgent(id);
@ -120,8 +126,7 @@ describe('server-v2 /api/v1 prompts', () => {
'POST',
`/api/v1/sessions/${id}/prompts/${promptId}:abort`,
);
expect(aborted.body.code).toBe(0);
expect(aborted.body.data.aborted).toBe(true);
expect(aborted.body.code).toBe(40402);
});
it('returns 40402 when aborting an unknown prompt', async () => {

View file

@ -0,0 +1,30 @@
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, describe, expect, it } from 'vitest';
import { classify, rotateServerToken, serverTokenPath } from '../src/index';
let tmpDir: string | undefined;
afterEach(() => {
if (tmpDir !== undefined) {
rmSync(tmpDir, { recursive: true, force: true });
tmpDir = undefined;
}
});
describe('server-v2 public API', () => {
it('exports classify', () => {
expect(classify('127.0.0.1')).toBe('loopback');
});
it('exports rotateServerToken and serverTokenPath', async () => {
tmpDir = mkdtempSync(join(tmpdir(), 'kimi-server-v2-public-api-'));
const token = await rotateServerToken(tmpDir);
expect(typeof token).toBe('string');
expect(token.length).toBeGreaterThan(0);
expect(serverTokenPath(tmpDir)).toBe(join(tmpDir, 'server.token'));
});
});

View file

@ -11,6 +11,7 @@ import {
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -96,14 +97,19 @@ describe('server-v2 /api/v1/sessions/{sid}/questions', () => {
const hasBody = body !== undefined;
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: hasBody ? { 'content-type': 'application/json' } : undefined,
headers: authHeaders(
server as RunningServer,
hasBody ? { 'content-type': 'application/json' } : {},
),
body: hasBody ? JSON.stringify(body) : undefined,
});
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function getJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}

View file

@ -1,4 +1,4 @@
import { mkdtemp, rm } from 'node:fs/promises';
import { mkdtemp, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
@ -11,6 +11,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { RpcClient, RpcError } from '../src/transport/rpcClient';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -45,7 +46,7 @@ describe('server-v2 /api/v2 RPC', () => {
server = undefined;
}
if (home !== undefined) {
await rm(home, { recursive: true, force: true });
await rm(home, { recursive: true, force: true, maxRetries: 3, retryDelay: 25 } as never);
home = undefined;
}
});
@ -68,7 +69,10 @@ describe('server-v2 /api/v2 RPC', () => {
headers['content-type'] = 'application/json';
init.body = JSON.stringify(arg);
}
if (token !== undefined) headers['authorization'] = `Bearer ${token}`;
// Default to the persistent bearer token — `/api/v2` is now gated by the
// same credential as every other route.
const credential = token ?? (server as RunningServer).authTokenService.getToken();
headers['authorization'] = `Bearer ${credential}`;
const res = await fetch(url, init);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -76,9 +80,9 @@ describe('server-v2 /api/v2 RPC', () => {
async function createSession(cwd: string): Promise<string> {
const res = await fetch(`${base}/api/v1/sessions`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify({ metadata: { cwd } }),
});
} as never);
const body = (await res.json()) as Envelope<{ id: string }>;
expect(body.code).toBe(0);
return body.data.id;
@ -205,6 +209,67 @@ describe('server-v2 /api/v2 RPC', () => {
expect(body.data.turn_id).toBe(0);
});
it('runs a shell command through shell:run', async () => {
const id = await createSession(home as string);
await createMainAgent(id);
const { body } = await call<{ stdout: string; stderr: string; isError?: boolean }>(
'POST',
`/api/v2/session/${id}/agent/main/shell:run`,
{ command: 'printf hello' },
);
expect(body.code).toBe(0);
expect(body.data.stdout).toBe('hello');
expect(body.data.stderr).toBe('');
expect(body.data.isError).not.toBe(true);
});
it('lists and installs plugins through plugins:* RPC', async () => {
const pluginRoot = await mkdtemp(join(tmpdir(), 'server-v2-plugin-source-'));
try {
await writeFile(join(pluginRoot, 'deploy.md'), '---\ndescription: Deploy\n---\n\nDeploy body', 'utf8');
await writeFile(
join(pluginRoot, 'kimi.plugin.json'),
JSON.stringify({ name: 'rpc-plugin', commands: ['./deploy.md'] }),
'utf8',
);
const installed = await call<{ id: string }>('POST', '/api/v2/plugins:install', { source: pluginRoot });
expect(installed.body.code).toBe(0);
expect(installed.body.data.id).toBe('rpc-plugin');
const listed = await call<readonly { id: string; state: string }[]>('GET', '/api/v2/plugins:list');
expect(listed.body.code).toBe(0);
expect(listed.body.data).toEqual([
expect.objectContaining({ id: 'rpc-plugin', state: 'ok' }),
]);
const info = await call<{ id: string }>('POST', '/api/v2/plugins:getInfo', { id: 'rpc-plugin' });
expect(info.body.code).toBe(0);
expect(info.body.data.id).toBe('rpc-plugin');
const commands = await call<readonly { pluginId: string; name: string }[]>(
'GET',
'/api/v2/plugins:listCommands',
);
expect(commands.body.code).toBe(0);
expect(commands.body.data).toEqual([
expect.objectContaining({ pluginId: 'rpc-plugin', name: 'deploy' }),
]);
const sessionId = await createSession(home as string);
await createMainAgent(sessionId);
const activated = await call<null>(
'POST',
`/api/v2/session/${sessionId}/agent/main/plugins:activateCommand`,
{ pluginId: 'rpc-plugin', commandName: 'deploy', args: 'prod' },
);
expect(activated.body.code).toBe(0);
} finally {
await rm(pluginRoot, { recursive: true, force: true });
}
});
it('returns 40401 when the agent does not exist', async () => {
const id = await createSession(home as string);
const { body } = await call<null>(
@ -220,7 +285,10 @@ describe('server-v2 /api/v2 RPC', () => {
it('works through the typed RpcClient', async () => {
const cwd = home as string;
await createSession(cwd);
const client = new RpcClient({ url: base });
const client = new RpcClient({
url: base,
token: (server as RunningServer).authTokenService.getToken(),
});
const sessions = client.core<ISessionIndex>('sessions');
const page = await sessions.list({});
@ -233,7 +301,10 @@ describe('server-v2 /api/v2 RPC', () => {
});
it('throws RpcError on unknown action', async () => {
const client = new RpcClient({ url: base });
const client = new RpcClient({
url: base,
token: (server as RunningServer).authTokenService.getToken(),
});
const sessions = client.core<ISessionIndex>('sessions');
// @ts-expect-error — intentionally calling a non-existent method
await expect(sessions.nope()).rejects.toBeInstanceOf(RpcError);
@ -266,13 +337,16 @@ describe('server-v2 /api/v2 RPC', () => {
// Fastify's default bodyLimit is 1MB; a larger body is rejected. Fastify's
// body-parser throws a 413 which our global error handler currently wraps
// as `50001` (HTTP 200) — either way the request is rejected, not served.
// A valid token is sent so the request reaches the body parser (the auth
// hook would short-circuit with 401 before reading the body otherwise).
const huge = 'x'.repeat(2 * 1024 * 1024);
const token = (server as RunningServer).authTokenService.getToken();
let rejected = false;
let code: number | undefined;
try {
const res = await fetch(`${base}/api/v2/sessions:list`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: { 'content-type': 'application/json', authorization: `Bearer ${token}` },
body: JSON.stringify({ big: huge }),
});
const body = (await res.json()) as Envelope<null>;
@ -315,18 +389,19 @@ describe('server-v2 /api/v2 RPC auth', () => {
server = undefined;
}
if (home !== undefined) {
await rm(home, { recursive: true, force: true });
await rm(home, { recursive: true, force: true, maxRetries: 3, retryDelay: 25 } as never);
home = undefined;
}
});
it('rejects calls without a token (40112)', async () => {
it('rejects calls without a token (40101)', async () => {
const res = await fetch(`${base}/api/v2/sessions:list`, { method: 'POST' });
expect(res.status).toBe(401);
const body = (await res.json()) as Envelope<null>;
expect(body.code).toBe(40112);
expect(body.code).toBe(40101);
});
it('accepts calls with the correct token', async () => {
it('accepts calls with the correct rpcToken', async () => {
const res = await fetch(`${base}/api/v2/sessions:list`, {
method: 'POST',
headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' },
@ -336,12 +411,24 @@ describe('server-v2 /api/v2 RPC auth', () => {
expect(body.code).toBe(0);
});
it('rejects a wrong token (40112)', async () => {
it('accepts the persistent token on /api/v2', async () => {
const persistent = (server as RunningServer).authTokenService.getToken();
const res = await fetch(`${base}/api/v2/sessions:list`, {
method: 'POST',
headers: { authorization: `Bearer ${persistent}`, 'content-type': 'application/json' },
body: JSON.stringify({}),
});
const body = (await res.json()) as Envelope<{ items: unknown[] }>;
expect(body.code).toBe(0);
});
it('rejects a wrong token (40101)', async () => {
const res = await fetch(`${base}/api/v2/sessions:list`, {
method: 'POST',
headers: { authorization: 'Bearer wrong' },
});
expect(res.status).toBe(401);
const body = (await res.json()) as Envelope<null>;
expect(body.code).toBe(40112);
expect(body.code).toBe(40101);
});
});

View file

@ -8,12 +8,13 @@ import { join } from 'node:path';
import type { IScopeHandle, Scope } from '@moonshot-ai/agent-core-v2';
import {
IAgentEventSinkService,
IAgentLifecycleService,
IEventSink,
IEventService,
ISessionLifecycleService,
} from '@moonshot-ai/agent-core-v2';
import type { AgentEvent } from '@moonshot-ai/protocol';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import {
type BroadcastTarget,
@ -41,13 +42,29 @@ class FakeSink {
}
}
class FakeEventBus {
private handlers: Array<(e: { type: string; payload: unknown }) => void> = [];
subscribe(handler: (e: { type: string; payload: unknown }) => void) {
this.handlers.push(handler);
return {
dispose: () => {
const i = this.handlers.indexOf(handler);
if (i >= 0) this.handlers.splice(i, 1);
},
};
}
emit(e: { type: string; payload: unknown }): void {
for (const h of [...this.handlers]) h(e);
}
}
class FakeAgentHandle {
readonly kind = 2;
readonly sink = new FakeSink();
readonly accessor;
constructor(readonly id: string) {
this.accessor = {
get: (t: unknown) => (t === IEventSink ? this.sink : undefined),
get: (t: unknown) => (t === IAgentEventSinkService ? this.sink : undefined),
};
}
dispose(): void {}
@ -84,9 +101,10 @@ class FakeLifecycle {
}
}
function makeCore(sessions: Map<string, FakeLifecycle>): Scope {
function makeCore(sessions: Map<string, FakeLifecycle>, eventBus = new FakeEventBus()): Scope {
const accessor = {
get(token: unknown): unknown {
if (token === IEventService) return eventBus;
if (token === ISessionLifecycleService) {
return {
get: (sid: string) => {
@ -121,14 +139,16 @@ function collectingTarget(): { target: BroadcastTarget; envelopes: EventEnvelope
describe('SessionEventBroadcaster', () => {
let dir: string;
let sessions: Map<string, FakeLifecycle>;
let eventBus: FakeEventBus;
let bc: SessionEventBroadcaster;
beforeEach(async () => {
dir = await mkdtemp(join(tmpdir(), 'kimi-broadcaster-test-'));
sessions = new Map();
eventBus = new FakeEventBus();
bc = new SessionEventBroadcaster({
eventsDir: dir,
core: makeCore(sessions),
core: makeCore(sessions, eventBus),
maxBufferSize: 3,
});
});
@ -244,6 +264,35 @@ describe('SessionEventBroadcaster', () => {
expect(snap.inFlightTurn).toMatchObject({ turn_id: 1, assistant_text: 'Hello' });
});
it('fans core model-catalog changes out to every session subscriber', async () => {
const lc = new FakeLifecycle();
lc.addAgent('main');
sessions.set('s1', lc);
const { target, envelopes } = collectingTarget();
await bc.subscribe('s1', target);
eventBus.emit({
type: 'event.model_catalog.changed',
payload: {
changed: [{ provider_id: 'managed:kimi-code', provider_name: 'Kimi Code', added: 1, removed: 0 }],
unchanged: [],
failed: [],
},
});
await vi.waitFor(() => expect(envelopes).toHaveLength(1));
expect(envelopes[0]).toMatchObject({
type: 'event.model_catalog.changed',
seq: 1,
session_id: '__global__',
payload: {
type: 'event.model_catalog.changed',
agentId: 'main',
sessionId: '__global__',
},
});
});
it('subscribe returns false for an unknown session', async () => {
const { target } = collectingTarget();
expect(await bc.subscribe('nope', target)).toBe(false);

View file

@ -5,6 +5,7 @@ import { join } from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -69,14 +70,19 @@ describe('server-v2 /api/v1/sessions', () => {
const hasBody = body !== undefined;
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: hasBody ? { 'content-type': 'application/json' } : undefined,
headers: authHeaders(
server as RunningServer,
hasBody ? { 'content-type': 'application/json' } : {},
),
body: hasBody ? JSON.stringify(body) : undefined,
});
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function getJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -148,6 +154,18 @@ describe('server-v2 /api/v1/sessions', () => {
expect(typeof body.data.has_more).toBe('boolean');
});
it('supports exclude_empty when listing sessions', async () => {
const cwd = home as string;
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });
const all = await getJson<PageWire>('/api/v1/sessions');
expect(all.body.data.items.some((s) => s.id === created.body.data.id)).toBe(true);
const filtered = await getJson<PageWire>('/api/v1/sessions?exclude_empty=true');
expect(filtered.body.code).toBe(0);
expect(filtered.body.data.items.some((s) => s.id === created.body.data.id)).toBe(false);
});
it('gets a session by id and 404s for unknown', async () => {
const cwd = home as string;
const created = await postJson<SessionWire>('/api/v1/sessions', { metadata: { cwd } });

View file

@ -31,6 +31,7 @@ import {
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -65,13 +66,15 @@ describe('server-v2 /api/v1 skills', () => {
server = undefined;
}
if (home !== undefined) {
await rm(home, { recursive: true, force: true });
await rm(home, { recursive: true, force: true, maxRetries: 3, retryDelay: 25 } as never);
home = undefined;
}
});
async function getJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -81,9 +84,9 @@ describe('server-v2 /api/v1 skills', () => {
): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify(body ?? {}),
});
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}

View file

@ -3,19 +3,22 @@
* snapshot shape and watermark consistency.
*/
import { mkdtemp, rm } from 'node:fs/promises';
import { mkdtemp, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import {
IAgentContextMemoryService,
IAgentEventSinkService,
IAgentLifecycleService,
IEventSink,
ISessionContext,
ISessionLifecycleService,
} from '@moonshot-ai/agent-core-v2';
import { sessionSnapshotResponseSchema, type AgentEvent } from '@moonshot-ai/protocol';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
describe('server-v2 GET /api/v1/sessions/:id/snapshot', () => {
let server: RunningServer | undefined;
@ -42,9 +45,9 @@ describe('server-v2 GET /api/v1/sessions/:id/snapshot', () => {
async function createSession(): Promise<string> {
const res = await fetch(`${base}/api/v1/sessions`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify({ metadata: { cwd: home } }),
});
} as never);
const body = (await res.json()) as { code: number; data: { id: string } };
expect(body.code).toBe(0);
return body.data.id;
@ -59,11 +62,13 @@ describe('server-v2 GET /api/v1/sessions/:id/snapshot', () => {
function emit(sessionId: string, event: AgentEvent): void {
const session = server!.core.accessor.get(ISessionLifecycleService).get(sessionId);
const main = session!.accessor.get(IAgentLifecycleService).getHandle('main');
main!.accessor.get(IEventSink).emit(event);
main!.accessor.get(IAgentEventSinkService).emit(event);
}
async function snapshot(sid: string) {
const res = await fetch(`${base}/api/v1/sessions/${sid}/snapshot`);
const res = await fetch(`${base}/api/v1/sessions/${sid}/snapshot`, {
headers: authHeaders(server as RunningServer),
} as never);
const body = (await res.json()) as { code: number; data: unknown };
expect(body.code).toBe(0);
return sessionSnapshotResponseSchema.parse(body.data);
@ -104,8 +109,84 @@ describe('server-v2 GET /api/v1/sessions/:id/snapshot', () => {
});
it('returns 404 for an unknown session', async () => {
const res = await fetch(`${base}/api/v1/sessions/sess_does_not_exist/snapshot`);
const res = await fetch(`${base}/api/v1/sessions/sess_does_not_exist/snapshot`, {
headers: authHeaders(server as RunningServer),
} as never);
const body = (await res.json()) as { code: number };
expect(body.code).not.toBe(0);
});
// Regression for the cold-session 404: a session that exists on disk but is
// not live in this process (e.g. carried over from a prior process, or
// created by v1) must load from disk instead of returning 40401. We restart
// the whole server on the same homeDir so the session is genuinely cold.
it('loads a cold (not live) session from disk instead of 404', async () => {
const sid = await createSession();
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: nothing is live in the new process — the session is cold.
expect(server!.core.accessor.get(ISessionLifecycleService).get(sid)).toBeUndefined();
const snap = await snapshot(sid);
expect(snap.session.id).toBe(sid);
});
// 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
// arithmetic and dropped the session id. `ISessionMetadata` now normalizes
// legacy documents on load. We rewrite `state.json` in the v1 layout, restart
// so the session is cold, then seed messages into the live (resumed) context
// so the snapshot exercises the message-timestamp projection deterministically
// (no reliance on wire-restore timing).
it('serves a v1-layout session (ISO timestamps, no id field) without crashing', async () => {
const sid = await createSession();
const session = server!.core.accessor.get(ISessionLifecycleService).get(sid);
if (session === undefined) throw new Error(`session ${sid} not found`);
const metaScope = session.accessor.get(ISessionContext).metaScope;
// Shut down, then rewrite state.json in the v1 layout (ISO-string
// timestamps, no `id`) so the next boot reads a cold legacy session.
await server!.close();
server = undefined;
const statePath = join(home as string, metaScope, 'state.json');
await writeFile(
statePath,
JSON.stringify({
title: 'v1 session',
createdAt: '2026-06-01T10:00:00.000Z',
updatedAt: '2026-06-01T11:00:00.000Z',
archived: false,
custom: { source: 'v1' },
}),
);
server = await startServer({ host: '127.0.0.1', port: 0, homeDir: home, logLevel: 'silent' });
base = `http://127.0.0.1:${server.port}`;
// Resume the cold session, then seed messages into the live context so the
// snapshot projects message timestamps from the normalized numeric base.
const resumed = await server!.core.accessor.get(ISessionLifecycleService).resume(sid);
if (resumed === undefined) throw new Error(`session ${sid} failed to resume`);
const main = await resumed.accessor.get(IAgentLifecycleService).createMain();
main.accessor.get(IAgentContextMemoryService).splice(0, 0, [
{ role: 'user', content: [{ type: 'text', text: 'hello' }], toolCalls: [] },
{ role: 'assistant', content: [{ type: 'text', text: 'hi' }], toolCalls: [] },
]);
const snap = await snapshot(sid);
expect(snap.session.id).toBe(sid);
expect(snap.session.title).toBe('v1 session');
// Session- and message-level timestamps are derived from the normalized
// numeric base — they must be valid ISO strings, not "Invalid time value".
expect(Number.isNaN(Date.parse(snap.session.created_at))).toBe(false);
expect(snap.messages.items).toHaveLength(2);
for (const message of snap.messages.items) {
expect(Number.isNaN(Date.parse(message.created_at))).toBe(false);
}
});
});

View file

@ -4,7 +4,7 @@ import { join } from 'node:path';
import {
IAgentLifecycleService,
IBackgroundService,
IAgentBackgroundService,
ISessionLifecycleService,
modelResolverSeed,
SingleModelResolver,
@ -13,6 +13,7 @@ import {
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -48,7 +49,7 @@ describe('server-v2 /api/v1/sessions/{sid}/tasks', () => {
beforeEach(async () => {
home = await mkdtemp(join(tmpdir(), 'kimi-server-v2-tasks-'));
// Seed a stub ISessionModelResolver so the agent scope can instantiate if a
// transitive service needs it; IBackgroundService itself does not.
// transitive service needs it; IAgentBackgroundService itself does not.
const modelResolver = new SingleModelResolver({
type: 'openai',
model: 'stub',
@ -70,27 +71,32 @@ describe('server-v2 /api/v1/sessions/{sid}/tasks', () => {
server = undefined;
}
if (home !== undefined) {
await rm(home, { recursive: true, force: true });
await rm(home, { recursive: true, force: true, maxRetries: 3, retryDelay: 25 } as never);
home = undefined;
}
});
async function getJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function postJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`, { method: 'POST' });
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function createSession(): Promise<string> {
const res = await fetch(`${base}/api/v1/sessions`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify({ metadata: { cwd: home as string } }),
});
} as never);
const body = (await res.json()) as Envelope<{ id: string }>;
expect(body.code).toBe(0);
return body.data.id;
@ -98,14 +104,14 @@ describe('server-v2 /api/v1/sessions/{sid}/tasks', () => {
// The main agent scope is not created automatically on session creation
// (server-v2 gap G10); create it here, then register fake background tasks
// directly into its IBackgroundService to bypass the tool loop.
async function mainAgentBackground(sessionId: string): Promise<IBackgroundService> {
// directly into its IAgentBackgroundService to bypass the tool loop.
async function mainAgentBackground(sessionId: string): Promise<IAgentBackgroundService> {
const session = server!.core.accessor.get(ISessionLifecycleService).get(sessionId);
if (session === undefined) throw new Error(`session ${sessionId} not found`);
const agent =
session.accessor.get(IAgentLifecycleService).getHandle('main') ??
(await session.accessor.get(IAgentLifecycleService).createMain());
return agent.accessor.get(IBackgroundService);
return agent.accessor.get(IAgentBackgroundService);
}
// Let the `registerTask` microtask run `start` (which appends output) before

View file

@ -16,6 +16,7 @@ import { ErrorCode, type Terminal } from '@moonshot-ai/protocol';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
// --- Fake PTY backend -------------------------------------------------------
//
@ -138,9 +139,9 @@ describe('server-v2 /api/v1/sessions/{sid}/terminals', () => {
async function createSession(cwd: string): Promise<string> {
const res = await fetch(`${base}/api/v1/sessions`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify({ metadata: { cwd } }),
});
} as never);
const body = (await res.json()) as Envelope<{ id: string }>;
expect(body.code).toBe(0);
return body.data.id;
@ -149,14 +150,16 @@ describe('server-v2 /api/v1/sessions/{sid}/terminals', () => {
async function post<T>(path: string, body: unknown): Promise<Envelope<T>> {
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify(body),
});
} as never);
return (await res.json()) as Envelope<T>;
}
async function get<T>(path: string): Promise<Envelope<T>> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return (await res.json()) as Envelope<T>;
}

View file

@ -19,8 +19,8 @@ import { join } from 'node:path';
import {
IAgentLifecycleService,
IAgentToolRegistryService,
ISessionLifecycleService,
IToolRegistry,
modelResolverSeed,
SingleModelResolver,
type ExecutableTool,
@ -32,6 +32,7 @@ import {
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -82,7 +83,9 @@ describe('server-v2 /api/v1 tools + mcp', () => {
});
async function getJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -92,9 +95,9 @@ describe('server-v2 /api/v1 tools + mcp', () => {
): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify(body ?? {}),
});
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -146,7 +149,7 @@ describe('server-v2 /api/v1 tools + mcp', () => {
it('projects registered tools with source mapping and mcp server id', async () => {
const id = await createSession();
const agent = await ensureMainAgent(id);
const registry = agent.accessor.get(IToolRegistry);
const registry = agent.accessor.get(IAgentToolRegistryService);
const schema = { type: 'object', properties: { msg: { type: 'string' } } };
registry.register(makeTool('Echo', schema), { source: 'builtin' });
registry.register(makeTool('MySkill'), { source: 'user' });

View file

@ -5,6 +5,7 @@ import { join } from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -36,14 +37,19 @@ interface HomeWire {
describe('server-v2 /api/v1 fs folder picker', () => {
let server: RunningServer | undefined;
let home: string | undefined;
let lockDir: string | undefined;
let base: string;
beforeEach(async () => {
home = await mkdtemp(join(tmpdir(), 'kimi-server-v2-fs-'));
// Keep the single-instance lock OUTSIDE the browsed homeDir (mirrors v1's
// isolated-lock harness) so the folder picker only sees the test fixtures.
lockDir = await mkdtemp(join(tmpdir(), 'kimi-server-v2-fs-lock-'));
server = await startServer({
host: '127.0.0.1',
port: 0,
homeDir: home,
lockPath: join(lockDir, 'lock'),
logLevel: 'silent',
});
base = `http://127.0.0.1:${server.port}`;
@ -58,10 +64,16 @@ describe('server-v2 /api/v1 fs folder picker', () => {
await rm(home, { recursive: true, force: true });
home = undefined;
}
if (lockDir !== undefined) {
await rm(lockDir, { recursive: true, force: true });
lockDir = undefined;
}
});
async function getJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -72,9 +84,12 @@ describe('server-v2 /api/v1 fs folder picker', () => {
const hasBody = body !== undefined;
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: hasBody ? { 'content-type': 'application/json' } : undefined,
headers: authHeaders(
server as RunningServer,
hasBody ? { 'content-type': 'application/json' } : {},
),
body: hasBody ? JSON.stringify(body) : undefined,
});
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -92,7 +107,9 @@ describe('server-v2 /api/v1 fs folder picker', () => {
// the wire as single-colon `/fs:browse`; the double-colon form 404s. This
// guards against reintroducing a `/fs:action` parametric dispatcher that
// would accept the non-v1 double-colon URL.
const res = await fetch(`${base}/api/v1/fs::browse`);
const res = await fetch(`${base}/api/v1/fs::browse`, {
headers: authHeaders(server as RunningServer),
} as never);
expect(res.status).toBe(404);
});

View file

@ -5,6 +5,7 @@ import { join } from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -63,9 +64,12 @@ describe('server-v2 /api/v1/workspaces', () => {
const hasBody = body !== undefined;
const res = await fetch(`${base}${path}`, {
method: 'POST',
headers: hasBody ? { 'content-type': 'application/json' } : undefined,
headers: authHeaders(
server as RunningServer,
hasBody ? { 'content-type': 'application/json' } : {},
),
body: hasBody ? JSON.stringify(body) : undefined,
});
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
@ -75,19 +79,24 @@ describe('server-v2 /api/v1/workspaces', () => {
): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`, {
method: 'PATCH',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify(body ?? {}),
});
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function deleteJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`, { method: 'DELETE' });
const res = await fetch(`${base}${path}`, {
method: 'DELETE',
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}
async function getJson<T>(path: string): Promise<{ status: number; body: Envelope<T> }> {
const res = await fetch(`${base}${path}`);
const res = await fetch(`${base}${path}`, {
headers: authHeaders(server as RunningServer),
} as never);
return { status: res.status, body: (await res.json()) as Envelope<T> };
}

View file

@ -6,6 +6,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { RpcWsError, WsClient } from '../src/transport/ws/wsClient';
import { authHeaders } from './helpers/auth';
interface Envelope<T> {
code: number;
@ -23,10 +24,12 @@ describe('server-v2 /api/v2/ws', () => {
let home: string | undefined;
let base: string;
let wsUrl: string;
let token: string;
beforeEach(async () => {
home = await mkdtemp(join(tmpdir(), 'kimi-server-v2-ws-'));
server = await startServer({ host: '127.0.0.1', port: 0, homeDir: home, logLevel: 'silent' });
token = server.authTokenService.getToken();
base = `http://127.0.0.1:${server.port}`;
wsUrl = `ws://127.0.0.1:${server.port}/api/v2/ws`;
});
@ -45,16 +48,16 @@ describe('server-v2 /api/v2/ws', () => {
async function createSession(cwd: string): Promise<string> {
const res = await fetch(`${base}/api/v1/sessions`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify({ metadata: { cwd } }),
});
} as never);
const body = (await res.json()) as Envelope<{ id: string }>;
expect(body.code).toBe(0);
return body.data.id;
}
it('performs a core call over WS', async () => {
const client = new WsClient({ url: wsUrl });
const client = new WsClient({ url: wsUrl, token });
const page = await client.call<{ items: unknown[] }>('core', 'sessions:list', {});
expect(Array.isArray(page.items)).toBe(true);
client.close();
@ -62,7 +65,7 @@ describe('server-v2 /api/v2/ws', () => {
it('performs a session call over WS', async () => {
const id = await createSession(home as string);
const client = new WsClient({ url: wsUrl });
const client = new WsClient({ url: wsUrl, token });
const meta = await client.call<SessionMetaWire>('session', 'session:read', undefined, {
sessionId: id,
});
@ -71,7 +74,7 @@ describe('server-v2 /api/v2/ws', () => {
});
it('returns an error for an unknown action', async () => {
const client = new WsClient({ url: wsUrl });
const client = new WsClient({ url: wsUrl, token });
await expect(client.call('core', 'sessions:nope')).rejects.toMatchObject({
code: 40001,
});
@ -79,7 +82,8 @@ describe('server-v2 /api/v2/ws', () => {
});
it('streams core events via listen', async () => {
const client = new WsClient({ url: wsUrl });
const id = await createSession(home as string);
const client = new WsClient({ url: wsUrl, token });
const { iterator, cancel } = client.listen<{ type: string; payload: unknown }>(
'core',
'events',
@ -88,8 +92,10 @@ describe('server-v2 /api/v2/ws', () => {
// Ensure the subscription is registered before publishing.
await new Promise((r) => setTimeout(r, 50));
const id = await createSession(home as string);
await fetch(`${base}/api/v1/sessions/${id}:archive`, { method: 'POST' });
await fetch(`${base}/api/v1/sessions/${id}:archive`, {
method: 'POST',
headers: authHeaders(server as RunningServer),
} as never);
const iter = iterator[Symbol.asyncIterator]();
const next = await Promise.race([
@ -106,7 +112,7 @@ describe('server-v2 /api/v2/ws', () => {
});
it('cancels a subscription', async () => {
const client = new WsClient({ url: wsUrl });
const client = new WsClient({ url: wsUrl, token });
const { iterator, cancel } = client.listen('core', 'events');
await new Promise((r) => setTimeout(r, 50));
cancel();
@ -117,7 +123,7 @@ describe('server-v2 /api/v2/ws', () => {
});
it('rejects pending calls on close', async () => {
const client = new WsClient({ url: wsUrl });
const client = new WsClient({ url: wsUrl, token });
const pending = client.call('core', 'sessions:list', {});
client.close();
await expect(pending).rejects.toThrow();

View file

@ -9,8 +9,8 @@ import { tmpdir } from 'node:os';
import { join } from 'node:path';
import {
IAgentEventSinkService,
IAgentLifecycleService,
IEventSink,
ISessionLifecycleService,
} from '@moonshot-ai/agent-core-v2';
import type { AgentEvent } from '@moonshot-ai/protocol';
@ -18,6 +18,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { WebSocket } from 'ws';
import { type RunningServer, startServer } from '../src/start';
import { authHeaders } from './helpers/auth';
interface Frame {
type: string;
@ -38,9 +39,9 @@ interface Conn {
next: (pred: (f: Frame) => boolean, timeoutMs?: number) => Promise<Frame>;
}
function openConn(url: string): Promise<Conn> {
function openConn(url: string, token: string): Promise<Conn> {
return new Promise((resolve, reject) => {
const ws = new WebSocket(url);
const ws = new WebSocket(url, [`kimi-code.bearer.${token}`]);
const frames: Frame[] = [];
const waiters: Array<(f: Frame) => void> = [];
const closed = new Promise<void>((res) => ws.on('close', () => res()));
@ -117,9 +118,9 @@ describe('server-v2 /api/v1/ws resync', () => {
async function createSession(): Promise<string> {
const res = await fetch(`${base}/api/v1/sessions`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
headers: authHeaders(server as RunningServer, { 'content-type': 'application/json' }),
body: JSON.stringify({ metadata: { cwd: home } }),
});
} as never);
const body = (await res.json()) as { code: number; data: { id: string } };
expect(body.code).toBe(0);
return body.data.id;
@ -134,18 +135,22 @@ describe('server-v2 /api/v1/ws resync', () => {
}
}
function withToken<T extends Record<string, unknown>>(payload: T): T & { token: string } {
return { ...payload, token: server!.authTokenService.getToken() };
}
function emitAgentEvent(sessionId: string, event: AgentEvent): void {
const session = server!.core.accessor.get(ISessionLifecycleService).get(sessionId);
expect(session).toBeDefined();
const agents = session!.accessor.get(IAgentLifecycleService);
const main = agents.getHandle('main');
expect(main).toBeDefined();
main!.accessor.get(IEventSink).emit(event);
main!.accessor.get(IAgentEventSinkService).emit(event);
}
it('server_hello then client_hello ack with accepted subscription', async () => {
const sid = await createSession();
const c = await openConn(wsUrl);
const c = await openConn(wsUrl, server!.authTokenService.getToken());
const hello = await c.next((f) => f.type === 'server_hello');
expect(hello.payload).toMatchObject({ protocol_version: 2 });
@ -153,7 +158,7 @@ describe('server-v2 /api/v1/ws resync', () => {
c.send({
type: 'client_hello',
id: 'h1',
payload: { client_id: 'cli', subscriptions: [sid] },
payload: withToken({ client_id: 'cli', subscriptions: [sid] }),
});
const ack = await c.next((f) => f.type === 'ack' && f.id === 'h1');
expect(ack.payload).toMatchObject({ accepted_subscriptions: [sid], resync_required: [] });
@ -165,9 +170,9 @@ describe('server-v2 /api/v1/ws resync', () => {
it('delivers a sequenced durable event to a subscribed connection', async () => {
const sid = await createSession();
await ensureMainAgent(sid);
const c = await openConn(wsUrl);
const c = await openConn(wsUrl, server!.authTokenService.getToken());
await c.next((f) => f.type === 'server_hello');
c.send({ type: 'client_hello', id: 'h1', payload: { client_id: 'cli', subscriptions: [sid] } });
c.send({ type: 'client_hello', id: 'h1', payload: withToken({ client_id: 'cli', subscriptions: [sid] }) });
await c.next((f) => f.type === 'ack' && f.id === 'h1');
emitAgentEvent(sid, { type: 'turn.started', turnId: 1 } as unknown as AgentEvent);
@ -186,9 +191,9 @@ describe('server-v2 /api/v1/ws resync', () => {
await ensureMainAgent(sid);
// First connection — subscribe, generate two durable events.
const c1 = await openConn(wsUrl);
const c1 = await openConn(wsUrl, server!.authTokenService.getToken());
await c1.next((f) => f.type === 'server_hello');
c1.send({ type: 'client_hello', id: 'h1', payload: { client_id: 'cli', subscriptions: [sid] } });
c1.send({ type: 'client_hello', id: 'h1', payload: withToken({ client_id: 'cli', subscriptions: [sid] }) });
await c1.next((f) => f.type === 'ack' && f.id === 'h1');
emitAgentEvent(sid, { type: 'turn.started', turnId: 1 } as unknown as AgentEvent);
emitAgentEvent(sid, { type: 'turn.ended', turnId: 1 } as unknown as AgentEvent);
@ -197,12 +202,12 @@ describe('server-v2 /api/v1/ws resync', () => {
await c1.closed;
// Second connection — replay from seq 1, expect only seq 2.
const c2 = await openConn(wsUrl);
const c2 = await openConn(wsUrl, server!.authTokenService.getToken());
await c2.next((f) => f.type === 'server_hello');
c2.send({
type: 'client_hello',
id: 'h2',
payload: { client_id: 'cli', subscriptions: [sid], cursors: { [sid]: { seq: 1 } } },
payload: withToken({ client_id: 'cli', subscriptions: [sid], cursors: { [sid]: { seq: 1 } } }),
});
const replayed = await c2.next((f) => f.type === 'turn.ended');
expect(replayed.seq).toBe(2);
@ -215,16 +220,16 @@ describe('server-v2 /api/v1/ws resync', () => {
it('sends resync_required on epoch mismatch', async () => {
const sid = await createSession();
const c = await openConn(wsUrl);
const c = await openConn(wsUrl, server!.authTokenService.getToken());
await c.next((f) => f.type === 'server_hello');
c.send({
type: 'client_hello',
id: 'h1',
payload: {
payload: withToken({
client_id: 'cli',
subscriptions: [sid],
cursors: { [sid]: { seq: 0, epoch: 'ep_wrong' } },
},
}),
});
const rs = await c.next((f) => f.type === 'resync_required');
expect(rs.payload).toMatchObject({ session_id: sid, reason: 'epoch_changed' });