feat(kap-server): add POST /api/v2/sessions:archive and :restore batch endpoints

Batch archive/restore for session-management views: { ids } (non-empty,
≤5000 unique after dedup) answers per-item results in input order with
succeeded/failed counts — only a body validation failure fails the whole
request, and an unknown id folds into its own item as 40401.

The live/cold split keeps the batch cheap: a session with a live handle
goes through the full ISessionLifecycleService chain (agents drain,
scope teardown, mirror drain), while a cold session is never
materialized — the new setColdSessionArchived helper in agent-core-v2
patches the persisted state.json (archived/archivedAt, updatedAt
preserved, mirroring setArchived's touchUpdatedAt: false semantics),
mirrors the flipped summary into the read-model queue, and republishes
the same event.session.archived bus event the live lifecycle emits
(:restore publishes nothing, matching the live restore). Hot items run
with bounded concurrency and the batch ends with one shared
ISessionIndexMirror.drain().
This commit is contained in:
liruifengv 2026-08-15 13:04:53 +08:00
parent bd51811d43
commit 2855ee59d3
6 changed files with 536 additions and 6 deletions

View file

@ -0,0 +1,5 @@
---
"@moonshot-ai/agent-core-v2": patch
---
Add a cold-session archive/restore path that patches the persisted metadata document, mirrors the flipped summary into the session-index read model, and republishes the archived bus event without materializing the session, backing the new `POST /api/v2/sessions:archive` / `:restore` batch endpoints (per-item results; live sessions still run the full lifecycle).

View file

@ -416,6 +416,7 @@ export * from '#/app/workspaceLifecycle/sessionLookup';
export * from '#/workspace/workspaceContext/workspaceContext';
export * from '#/workspace/sessionLifecycle/sessionLifecycle';
export * from '#/workspace/sessionLifecycle/sessionLifecycleService';
export * from '#/workspace/sessionLifecycle/coldSessionArchive';
export * from '#/workspace/sessionLifecycle/internal/addressing';
export * from '#/session/externalHooks/externalHooks';
export * from '#/session/externalHooks/externalHooksService';

View file

@ -0,0 +1,67 @@
/**
* `sessionLifecycle` domain cold-session archive/restore without session
* materialization.
*
* Writes the archived flag straight into the persisted metadata document
* (`state.json` under the handler-chain scope derived through
* `internal/addressing` from the `bootstrap` sessions scope) via the
* `storage` access-pattern store, mirrors the flipped summary into the
* `sessionIndex` mirror queue (drained by the caller, never per item), and
* publishes the same `event.session.archived` bus event the live
* `ISessionLifecycleService.archive` emits through `event` restore
* publishes nothing, matching the live `restore` (which only flips the
* flag through `ISessionMetadata`). `updatedAt` is preserved verbatim,
* mirroring `setArchived`'s `touchUpdatedAt: false` semantics, and every
* other persisted field survives the read-modify-write untouched. Call
* only for a session with no live handle in any workspace handler a
* live session must go through the full lifecycle so its agents drain
* and its scope tears down; the direct write deliberately races a
* concurrent resume unsynchronized (the read model heals by
* reconciliation). Existence reads from `ISessionIndex`: an unknown id
* and an index entry whose document is unreadable both report
* `not_found`.
*/
import type { ServicesAccessor } from '#/_base/di/instantiation';
import { IBootstrapService } from '#/app/bootstrap/bootstrap';
import { IEventService } from '#/app/event/event';
import { ISessionIndex, ISessionIndexMirror } from '#/app/sessionIndex/sessionIndex';
import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
import type { SessionMeta } from '#/session/sessionMetadata/sessionMetadata';
import { sessionScopeOf, workspacePersistenceScope } from './internal/addressing';
export type ColdSessionArchiveOutcome = 'updated' | 'not_found';
export async function setColdSessionArchived(
accessor: ServicesAccessor,
sessionId: string,
archived: boolean,
): Promise<ColdSessionArchiveOutcome> {
const summary = await accessor.get(ISessionIndex).get(sessionId);
if (summary === undefined) return 'not_found';
const docs = accessor.get(IAtomicDocumentStore);
const metaScope = sessionScopeOf(
workspacePersistenceScope(
accessor.get(IBootstrapService).scope('sessions'),
summary.workspaceId,
),
sessionId,
);
let persisted: SessionMeta | undefined;
try {
persisted = await docs.get<SessionMeta>(metaScope, 'state.json');
} catch {
persisted = undefined;
}
if (persisted === undefined) return 'not_found';
const archivedAt = archived ? Date.now() : undefined;
await docs.set(metaScope, 'state.json', { ...persisted, archived, archivedAt });
accessor.get(ISessionIndexMirror).record({ ...summary, archived, archivedAt });
if (archived) {
accessor
.get(IEventService)
.publish({ type: 'event.session.archived', payload: { sessionId } });
}
return 'updated';
}

View file

@ -36,14 +36,35 @@
* same edge pattern as v1's unpaged `GET /api/v1/sessions`. All sorts
* share one comparator + one cursor encoding, so every sort paginates
* identically.
*
* Batch actions: `POST /sessions:archive` / `POST /sessions:restore`
* (registered as `/sessions::{action}` find-my-way splits a segment at
* its first `:`, so the wire path carries a single colon, same as the v1
* `/fs::browse` precedent) take `{ ids }` (non-empty, 5000 unique) and
* answer per-item results `data.results[]` in input order with
* `ok` / `error`, plus `succeeded` / `failed` counts; only a body
* validation failure fails the whole request. A live session goes
* through the full `ISessionLifecycleService` chain (agents drain, scope
* teardown, mirror drain); a cold session is never materialized its
* archived flag is patched straight into the persisted metadata
* document, mirrored into the read model, and (`:archive` only)
* announced through the same `event.session.archived` bus event the live
* lifecycle publishes, while `:restore` publishes nothing, matching the
* live restore. An unknown id folds into its own item as 40401. The
* batch ends with one shared `ISessionIndexMirror.drain()`, never one
* per item.
*/
import { createHash } from 'node:crypto';
import {
ISessionIndex,
ISessionIndexMirror,
ISessionLifecycleService,
IWorkspaceAliases,
IWorkspaceService,
liveHandlerForSession,
setColdSessionArchived,
type Scope,
type SessionSummary,
} from '@moonshot-ai/agent-core-v2';
@ -64,6 +85,14 @@ interface V2SessionsRouteHost {
reply: { send(payload: unknown): unknown },
) => Promise<void> | void,
): unknown;
post(
path: string,
options: { preHandler: unknown[]; schema?: Record<string, unknown> },
handler: (
req: { id: string; body: unknown; params: unknown; headers: Record<string, unknown> },
reply: { send(payload: unknown): unknown },
) => Promise<void> | void,
): unknown;
}
// ---------------------------------------------------------------------------
@ -199,6 +228,43 @@ const v2SessionPageSchema = z.object({
/** `40001 validation.failed` carries the offending fields (REST.md §1.4). */
const detailsSchema = z.array(z.object({ path: z.string(), message: z.string() }));
// ---------------------------------------------------------------------------
// Batch archive / restore contract
// ---------------------------------------------------------------------------
/** Cap on unique ids per batch, keeping one request's edge work bounded. */
const BATCH_IDS_MAX = 5000;
/** Hot-path lifecycle calls run with this many in flight at most. */
const BATCH_CONCURRENCY = 8;
const v2SessionsBatchBodySchema = z
.object({ ids: z.array(z.string().min(1)).min(1) })
.superRefine((value, ctx) => {
if (new Set(value.ids).size > BATCH_IDS_MAX) {
ctx.addIssue({
code: 'custom',
message: `ids must contain at most ${BATCH_IDS_MAX} unique entries`,
path: ['ids'],
params: { code: ErrorCode.VALIDATION_FAILED },
});
}
});
const v2SessionsBatchResultSchema = z.object({
results: z.array(
z.object({
id: z.string(),
ok: z.boolean(),
error: z.object({ code: z.number().int(), message: z.string() }).optional(),
}),
),
succeeded: z.number().int(),
failed: z.number().int(),
});
type V2BatchItemResult = z.infer<typeof v2SessionsBatchResultSchema>['results'][number];
type V2GitDomain = z.infer<typeof v2GitDomainSchema>;
type V2SessionWire = z.infer<typeof v2SessionSchema>;
@ -376,6 +442,77 @@ class GitDomainResolver {
// Route
// ---------------------------------------------------------------------------
/**
* Run one `:archive` / `:restore` batch: live sessions through the full
* `ISessionLifecycleService` chain, cold sessions through the direct cold
* patch (no materialization); per-item failures fold into the result list
* in input order. Ends with a single shared mirror drain.
*/
async function runBatchArchive(
core: Scope,
action: 'archive' | 'restore',
rawIds: readonly string[],
requestId: string,
reply: { send(payload: unknown): unknown },
): Promise<void> {
const archived = action === 'archive';
const ids = [...new Set(rawIds)];
const results: (V2BatchItemResult | undefined)[] = ids.map(() => undefined);
// Per-item work never throws: a failure folds into its own result and
// the rest of the batch still runs.
const applyOne = async (id: string): Promise<V2BatchItemResult> => {
try {
const liveHandler = liveHandlerForSession(core.accessor, id);
if (liveHandler !== undefined) {
const lifecycle = liveHandler.accessor.get(ISessionLifecycleService);
if (archived) await lifecycle.archive(id);
else await lifecycle.restore(id);
return { id, ok: true };
}
const outcome = await setColdSessionArchived(core.accessor, id, archived);
return outcome === 'updated'
? { id, ok: true }
: {
id,
ok: false,
error: {
code: ErrorCode.SESSION_NOT_FOUND,
message: `session ${id} does not exist`,
},
};
} catch (error) {
return {
id,
ok: false,
error: {
code: ErrorCode.INTERNAL_ERROR,
message: error instanceof Error ? error.message : String(error),
},
};
}
};
let next = 0;
const workers = Array.from({ length: Math.min(BATCH_CONCURRENCY, ids.length) }, async () => {
while (next < ids.length) {
const index = next++;
results[index] = await applyOne(ids[index] as string);
}
});
await Promise.all(workers);
// One drain for the whole batch — cold records queue in the mirror, and
// the hot path already drained itself per call.
await core.accessor.get(ISessionIndexMirror).drain();
// Every slot was assigned by the workers — no undefined entries remain.
const settled = results as V2BatchItemResult[];
const succeeded = settled.filter((result) => result.ok).length;
reply.send(
okEnvelope({ results: settled, succeeded, failed: settled.length - succeeded }, requestId),
);
}
export function registerV2SessionsRoutes(app: V2SessionsRouteHost, core: Scope): void {
const gitResolver = new GitDomainResolver(core);
@ -551,4 +688,30 @@ export function registerV2SessionsRoutes(app: V2SessionsRouteHost, core: Scope):
listRoute.options,
listRoute.handler as Parameters<V2SessionsRouteHost['get']>[2],
);
for (const action of ['archive', 'restore'] as const) {
const batchRoute = defineRoute(
{
method: 'POST',
// `/sessions::${action}` in find-my-way serves the wire path
// `/sessions:archive` / `/sessions:restore` (single colon).
path: `/sessions::${action}`,
body: v2SessionsBatchBodySchema,
success: { data: v2SessionsBatchResultSchema },
errors: {
[ErrorCode.VALIDATION_FAILED]: { detailsSchema },
},
description: `Batch-${action} sessions by id ({ ids }, ≤5000 unique). Per-item results — a missing session folds into its own item; cold sessions are patched without materialization.`,
tags: ['v2-sessions'],
},
async (req, reply) => {
await runBatchArchive(core, action, req.body.ids, req.id, reply);
},
);
app.post(
batchRoute.path,
batchRoute.options,
batchRoute.handler as Parameters<V2SessionsRouteHost['post']>[2],
);
}
}

View file

@ -444,6 +444,14 @@ exports[`API surface snapshot > matches the documented v2 route table and meta e
"POST",
"/api/v1/workspaces/{workspace_id}/untrust",
],
[
"POST",
"/api/v2/sessions:archive",
],
[
"POST",
"/api/v2/sessions:restore",
],
[
"PUT",
"/api/v1/providers/{provider_id}",

View file

@ -1,13 +1,16 @@
/**
* Scenario: `/api/v2/sessions` domain-grouped session list query.
* Scenario: `/api/v2/sessions` domain-grouped session list query + batch actions.
* Responsibilities: envelope wire shape (business outcome in `code`: 40001
* invalid params / 40922 page_token mismatch), filters, sort orders, opaque
* page tokens, git domain dedup/cache/degradation, v2 auth error shape, and
* the activity-status mapper.
* Wiring: real kap-server; `ISessionIndex` / `IGitService` stubbed via DI seeds.
* page tokens, page-number mode + total, git domain dedup/cache/degradation,
* v2 auth error shape, the activity-status mapper, and the
* `POST /sessions:archive` / `:restore` batch endpoints (per-item results,
* live/cold split, cold path never materializes).
* Wiring: real kap-server; the list tests stub `ISessionIndex` / `IGitService`
* via DI seeds, the batch tests run real sessions in a temp home.
* Run: `pnpm --filter @moonshot-ai/kap-server exec vitest run test/v2Sessions.test.ts`.
*/
import { mkdtemp, rm } from 'node:fs/promises';
import { mkdtemp, readFile, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
@ -15,6 +18,15 @@ import {
Error2,
ErrorCodes,
ISessionIndex,
ISessionLifecycleService,
IEventService,
IWorkspaceLifecycleService,
closeSessionById,
getLiveSessionById,
liveHandlerForSession,
resumeSessionById,
sessionDirOf,
type GlobalEvent,
type SessionSummary,
} from '@moonshot-ai/agent-core-v2';
import {
@ -22,7 +34,7 @@ import {
type FsPullRequest,
IGitService,
} from '@moonshot-ai/agent-core-v2/app/git/git';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
import { mapActivityStatus } from '../src/routes/v2/sessions';
@ -492,6 +504,280 @@ describe('server /api/v2/sessions', () => {
});
});
describe('server /api/v2/sessions batch archive/restore', () => {
interface BatchItemWire {
id: string;
ok: boolean;
error?: { code: number; message: string };
}
interface BatchWire {
results: BatchItemWire[];
succeeded: number;
failed: number;
}
interface BatchEnvelopeWire {
code: number;
msg: string;
data: BatchWire | null;
request_id: string;
details?: { path: string; message: string }[];
}
let server: RunningServer | undefined;
let home: string | undefined;
let base: string;
beforeEach(async () => {
home = await mkdtemp(join(tmpdir(), 'kimi-server-v2-sessions-batch-'));
server = await startServer({
hostIdentity: TEST_HOST_IDENTITY,
host: '127.0.0.1',
port: 0,
homeDir: home,
logLevel: 'silent',
});
base = `http://127.0.0.1:${server.port}`;
});
afterEach(async () => {
vi.restoreAllMocks();
if (server !== undefined) {
await server.close();
server = undefined;
}
if (home !== undefined) {
await new Promise((resolve) => setTimeout(resolve, 25));
await rm(home, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 } as never);
home = undefined;
}
});
function core(): RunningServer['core']['accessor'] {
return (server as RunningServer).core.accessor;
}
/** Subscribe a bus-event collector; caller disposes the returned sub. */
function collectEvents(): { events: GlobalEvent[]; dispose(): void } {
const events: GlobalEvent[] = [];
const sub = core().get(IEventService).subscribe((event) => events.push(event));
return {
events,
dispose: () => {
sub.dispose();
},
};
}
async function createSession(): Promise<{ id: string; workspace_id: string }> {
const res = await authedFetch(server as RunningServer, base, '/api/v1/sessions', {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ metadata: { cwd: home } }),
});
const body = (await res.json()) as {
code: number;
data: { id: string; workspace_id: string };
};
expect(body.code).toBe(0);
return body.data;
}
async function postBatch(path: string, body?: unknown): Promise<BatchEnvelopeWire> {
const res = await authedFetch(server as RunningServer, base, path, {
method: 'POST',
headers: body !== undefined ? { 'content-type': 'application/json' } : {},
body: body !== undefined ? JSON.stringify(body) : undefined,
});
expect(res.status).toBe(200);
return (await res.json()) as BatchEnvelopeWire;
}
async function readStateJson(workspaceId: string, id: string): Promise<Record<string, unknown>> {
const dir = sessionDirOf(home as string, `sessions/${workspaceId}`, id);
return JSON.parse(await readFile(join(dir, 'state.json'), 'utf-8')) as Record<string, unknown>;
}
async function indexArchived(id: string): Promise<boolean | undefined> {
return (await core().get(ISessionIndex).get(id))?.archived;
}
async function listedIds(query = ''): Promise<string[]> {
const res = await authedFetch(server as RunningServer, base, `/api/v2/sessions${query}`);
const body = (await res.json()) as { code: number; data: { items: { id: string }[] } };
expect(body.code).toBe(0);
return body.data.items.map((item) => item.id);
}
it('archives a cold session without materializing it or touching a workspace handler', async () => {
const created = await createSession();
await closeSessionById(core(), created.id);
expect(getLiveSessionById(core(), created.id)).toBeUndefined();
// Any materialization (resume, or the v1 single-archive route) must go
// through handlerFor; the cold path never touches it.
const handlerForSpy = vi.spyOn(core().get(IWorkspaceLifecycleService), 'handlerFor');
const { events, dispose } = collectEvents();
const before = await readStateJson(created.workspace_id, created.id);
const body = await postBatch('/api/v2/sessions:archive', { ids: [created.id] });
expect(body.code).toBe(0);
expect(body.data).toMatchObject({
succeeded: 1,
failed: 0,
results: [{ id: created.id, ok: true }],
});
expect(handlerForSpy).not.toHaveBeenCalled();
expect(getLiveSessionById(core(), created.id)).toBeUndefined();
// The persisted metadata flips exactly like setArchived(true): archived
// (+ archivedAt), updatedAt and every other field preserved.
const after = await readStateJson(created.workspace_id, created.id);
expect(after['archived']).toBe(true);
expect(typeof after['archivedAt']).toBe('number');
expect(after['updatedAt']).toBe(before['updatedAt']);
expect(after['createdAt']).toBe(before['createdAt']);
expect(after['agents']).toEqual(before['agents']);
// The route drained the mirror once: the read model already answers
// archived, and the v2 list serves the session under meta.archived=true.
expect(await indexArchived(created.id)).toBe(true);
expect(await listedIds('?meta.archived=true')).toEqual([created.id]);
expect(await listedIds()).toEqual([]);
// Same bus event the live lifecycle publishes.
expect(events.filter((event) => event.type === 'event.session.archived')).toEqual([
{ type: 'event.session.archived', payload: { sessionId: created.id } },
]);
dispose();
});
it('archives a live session through the full lifecycle chain', async () => {
const created = await createSession();
const liveHandler = liveHandlerForSession(core(), created.id);
expect(liveHandler).toBeDefined();
const lifecycle = liveHandler?.accessor.get(ISessionLifecycleService);
const archiveSpy = vi.spyOn(lifecycle as ISessionLifecycleService, 'archive');
const resumeSpy = vi.spyOn(lifecycle as ISessionLifecycleService, 'resume');
const { events, dispose } = collectEvents();
const body = await postBatch('/api/v2/sessions:archive', { ids: [created.id] });
expect(body.code).toBe(0);
expect(body.data?.results).toEqual([{ id: created.id, ok: true }]);
// The full chain ran: archive() (no resume needed for a live session)
// closed and disposed the session and published the event itself.
expect(archiveSpy).toHaveBeenCalledWith(created.id);
expect(resumeSpy).not.toHaveBeenCalled();
expect(getLiveSessionById(core(), created.id)).toBeUndefined();
expect(
events.some(
(event) =>
event.type === 'event.session.archived' &&
(event.payload as { sessionId: string }).sessionId === created.id,
),
).toBe(true);
expect(await indexArchived(created.id)).toBe(true);
dispose();
});
it('reports per-item results in input order for a live/cold/missing mixed batch', async () => {
const live = await createSession();
const cold = await createSession();
await closeSessionById(core(), cold.id);
const body = await postBatch('/api/v2/sessions:archive', {
ids: [live.id, cold.id, 'sess_missing'],
});
expect(body.code).toBe(0);
expect(body.data?.results).toEqual([
{ id: live.id, ok: true },
{ id: cold.id, ok: true },
{
id: 'sess_missing',
ok: false,
error: { code: 40401, message: 'session sess_missing does not exist' },
},
]);
expect(body.data?.succeeded).toBe(2);
expect(body.data?.failed).toBe(1);
expect(await indexArchived(live.id)).toBe(true);
expect(await indexArchived(cold.id)).toBe(true);
});
it('restores a cold session without materializing it and publishes no archived event', async () => {
const created = await createSession();
await closeSessionById(core(), created.id);
await postBatch('/api/v2/sessions:archive', { ids: [created.id] });
expect(await indexArchived(created.id)).toBe(true);
const handlerForSpy = vi.spyOn(core().get(IWorkspaceLifecycleService), 'handlerFor');
const { events, dispose } = collectEvents();
const before = await readStateJson(created.workspace_id, created.id);
const body = await postBatch('/api/v2/sessions:restore', { ids: [created.id] });
expect(body.code).toBe(0);
expect(body.data?.results).toEqual([{ id: created.id, ok: true }]);
expect(handlerForSpy).not.toHaveBeenCalled();
expect(getLiveSessionById(core(), created.id)).toBeUndefined();
const after = await readStateJson(created.workspace_id, created.id);
expect(after['archived']).toBe(false);
expect('archivedAt' in after).toBe(false);
expect(after['updatedAt']).toBe(before['updatedAt']);
expect(await indexArchived(created.id)).toBe(false);
expect(await listedIds()).toEqual([created.id]);
// The live restore publishes nothing either — no event at all.
expect(events.filter((event) => event.type === 'event.session.archived')).toEqual([]);
dispose();
});
it('restores a live session through the lifecycle chain and keeps it live', async () => {
const created = await createSession();
await postBatch('/api/v2/sessions:archive', { ids: [created.id] });
// Back to live-but-archived: resume materializes regardless of the flag.
expect(await resumeSessionById(core(), created.id)).toBeDefined();
const lifecycle = liveHandlerForSession(core(), created.id)?.accessor.get(
ISessionLifecycleService,
);
const restoreSpy = vi.spyOn(lifecycle as ISessionLifecycleService, 'restore');
const body = await postBatch('/api/v2/sessions:restore', { ids: [created.id] });
expect(body.code).toBe(0);
expect(body.data?.results).toEqual([{ id: created.id, ok: true }]);
expect(restoreSpy).toHaveBeenCalledWith(created.id);
expect(getLiveSessionById(core(), created.id)).toBeDefined();
expect(await indexArchived(created.id)).toBe(false);
});
it('validates the batch body: empty, missing, over the unique cap, duplicates', async () => {
for (const body of [{ ids: [] }, {}]) {
const rejected = await postBatch('/api/v2/sessions:archive', body);
expect(rejected.code).toBe(40001);
expect(rejected.data).toBeNull();
}
const tooMany = await postBatch('/api/v2/sessions:archive', {
ids: Array.from({ length: 5001 }, (_, i) => `sess_${i}`),
});
expect(tooMany.code).toBe(40001);
// Duplicates collapse before the cap; a repeated id runs once.
const deduped = await postBatch('/api/v2/sessions:archive', {
ids: Array.from({ length: 5001 }, () => 'sess_dup'),
});
expect(deduped.code).toBe(0);
expect(deduped.data?.results).toHaveLength(1);
expect(deduped.data?.results[0]?.ok).toBe(false);
expect(deduped.data?.results[0]?.error?.code).toBe(40401);
});
});
describe('mapActivityStatus', () => {
it('maps a cold persisted failure to failed, live outcomes still win', () => {
const coldIdle = { busy: false, mainTurnActive: false, pendingInteraction: 'none' as const, live: false as const };