mirror of
https://github.com/MoonshotAI/kimi-code.git
synced 2026-08-20 14:16:22 +00:00
feat(kap-server): add page mode, updated_before, and batch archive/restore to v2 sessions (#2983)
* feat(kap-server): add page-number mode and total to GET /api/v2/sessions
The v2 session list gains a stateless 1-based `page` parameter beside the
opaque page_token cursor for admin-style lists that jump arbitrarily:
each request stays a full independent snapshot, no token is minted, and
`page` + `page_token` together fail 40001. Every response now carries
`total` (the filtered/sorted set size) in both pagination modes.
* feat(kap-server): add meta.updated_before filter to GET /api/v2/sessions
Symmetric with meta.updated_after (inclusive boundary, Unix ms), applied
at the edge over the drained set and bound into the page_token query
fingerprint like every other condition.
* 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().
* docs(server-api): document v2 sessions page mode, total, updated_before, and batch archive/restore
* fix(kap-server): deep-import workspace lifecycle symbols in the v2 sessions route
CI's tsgo/rolldown (Linux) fail to bind liveHandlerForSession and
IWorkspaceLifecycleService through the agent-core-v2 package-root
barrel even though it re-exports them; the same files use the
established deep-import pattern already used for the git domain.
* fix(kap-server): inline the live-handler lookup in the batch route
The previous deep imports still fail to resolve on CI's Linux toolchain
(tsgo TS2307, rolldown MISSING_EXPORT) while every other module path
from the same package binds fine. Keep the route self-contained: the
hot-path lookup is a five-line loop over IWorkspaceLifecycleService's
handlers (mirrors agent-core-v2's liveHandlerForSession), and the tests
assert non-materialization behaviorally via the live map instead of
importing the same two symbols for spies.
* fix(kap-server): drive the batch hot path through getLiveSessionById
The phantom only hits the workspaceLifecycle-group symbols in these two
files on CI's Linux toolchain; getLiveSessionById is observed to bind
fine there. It returns the session's live scope directly (no resume),
which is exactly what the batch hot path needs.
* refactor(kap-server): move the batch live/cold split into agent-core-v2
setSessionArchivedBatch owns the split next to the cold patch: live
sessions go through the full lifecycle chain via the workspace handler
accessor (the v1-proven resolution path), cold sessions through the
direct write. The route becomes a thin wire-code adapter, and the batch
tests assert the live chain behaviorally (disposal, events, index)
instead of spying through scope accessors.
* fix(agent-core-v2): import sessionLookup relatively from coldSessionArchive
The '#/app/workspaceLifecycle/*' specifier resolves from src/ and
src/app/* files on CI's Linux toolchain but not from
src/workspace/sessionLifecycle/ (tsgo TS2307, rolldown follows); a
relative import bypasses the package-imports mapping.
* fix(agent-core-v2): migrate the batch hot path to ISessionManager
Main's workspace/session DI refactor removed the workspaceLifecycle
lookup modules; the live branch now goes through the App-level
ISessionManager (the same entry the v1 action route uses post-refactor)
with getLiveSessionById from the new sessionManager lookup.
* feat(kap-server): add the id,archived item projection to GET /api/v2/sessions
fields=id,archived trims each item to { id, archived } for
select-all-matching flows (the session admin page's Gmail-style
select-all). Only that projection gets the relaxed page_size ceiling
(10000); unknown fields, non-pair subsets, and include=git combinations
are 40001, and the projection binds into the page_token fingerprint so
shapes never flip mid-pagination.
* fix(agent-core-v2): serialize the batch cold write against in-flight resumes
Codex review on #2983: while a resume is in flight the live registry
hides the handle, so the batch route could classify the session as cold
and its direct write would race the materializing metadata service (its
stale in-memory document wins the next write, silently un-archiving the
session after the endpoint reported success).
The batch now settles the resume first: SessionManager registers the
whole resume promise synchronously at the App level (controllerForSession
is async, so the controller's own resuming map learns about it a few
microtasks late) and whenResumeSettled awaits it before classification —
a settled resume lands the item on the live chain, a failed one falls
back to the cold path. Also folds the module header down to the
package's external-role comment convention.
* fix(agent-core-v2): publish SessionArchived as an Event2 class in cold archive
* fix(agent-core-v2): serialize batch archive/restore with session lifecycle transitions
* fix(agent-core-v2): serialize session delete with the lifecycle chain
* fix(agent-core-v2): mirror the persisted metadata on cold archive, not the index summary
* docs(agent-core-v2): bring sessionManager comments and new tests to package conventions
* fix(agent-core-v2): normalize legacy session metadata before the cold archive write
* fix(kap-server): serialize the v1 single-session archive with the lifecycle chain
* chore: drop changesets for internal-only protocol work
* fix(agent-core-v2): encode cold-archived metadata for v1 readers
* fix(agent-core-v2): serialize fork and createChild with the source session's chain
* refactor(agent-core-v2): chain every session lifecycle method and hand batch sections unguarded ops
* fix(agent-core-v2): propagate failed resumes to the next settle
* fix(agent-core-v2): roll back the unannounced handle when a resume fails mid-materialization
* fix(agent-core-v2): read and migrate the legacy session-meta location on cold archive
* fix(agent-core-v2): serialize explicit-id session creation with the lifecycle chain
create() with a caller-supplied sessionId bypassed the per-session chain,
so a concurrent batch archive could classify the half-created session as
cold and write archived state that the live metadata service later
overwrites. Creation now queues on the target id's chain whenever an
explicit id is present.
Also type the resume-failure maps as Error and normalize at the catch
site, satisfying only-throw-error.
* style(kap-server): strip comments from the session routes per the no-comments convention
* fix(agent-core-v2): serialize explicit fork and child target ids on the lifecycle chain
fork() and createChild() with a newSessionId locked only the source id, so
a batch archive of the target could slip into the creation window: the
index already knows the half-created session, the batch writes archived
state to its document, and the fork's in-memory metadata later overwrites
it. Both operations now acquire the deduped, sorted key set so multi-key
sections always take locks in one deterministic order.
This commit is contained in:
liruifengv
GitHub
parent
5ae82cd5bc
commit
eaa3969dd3
15 changed files with 1519 additions and 47 deletions
|
|
@ -77,7 +77,7 @@ Error codes are grouped by band:
|
|||
List endpoints come in two styles:
|
||||
|
||||
- **Cursor style**: `before_id` / `after_id` (mutually exclusive) plus `page_size` (1–100), responding with `{ items, has_more }`. Used by the session list, message list, transcript, and others.
|
||||
- **`page_token`**: an opaque token (bound to a fingerprint of the query conditions), used by `POST /api/v1/search` and `GET /api/v2/sessions`. Changing any query condition mid-pagination invalidates the token: v2 returns `40922`, search returns `40001`.
|
||||
- **`page_token`**: an opaque token (bound to a fingerprint of the query conditions), used by `POST /api/v1/search` and `GET /api/v2/sessions`. Changing any query condition mid-pagination invalidates the token: v2 returns `40922`, search returns `40001`. `GET /api/v2/sessions` also offers a stateless `page` page-number mode as an alternative.
|
||||
|
||||
## REST endpoints
|
||||
|
||||
|
|
@ -245,6 +245,8 @@ In-session file operations go through `POST /api/v1/sessions/{session_id}/fs:{ac
|
|||
| `POST /api/v1/search` | Cross-session full-text search; `mode` is `terms` (default) or `literal` (exact substring); `page_token` pagination |
|
||||
| `GET /api/v1/connections` | List live WebSocket connections |
|
||||
| `GET /api/v2/sessions` | Next-generation session list, see below |
|
||||
| `POST /api/v2/sessions:archive` | Batch-archive sessions, see below |
|
||||
| `POST /api/v2/sessions:restore` | Batch-restore archived sessions, see below |
|
||||
| `/api/v1/debug/*` | Reflection debug RPC; mounted only with `--debug-endpoints` on loopback, not a stable protocol |
|
||||
|
||||
### `GET /api/v2/sessions`
|
||||
|
|
@ -256,13 +258,38 @@ A next-generation session query for list views — filtering, sorting, and field
|
|||
| `workspace.id` | Filter by workspace; repeatable |
|
||||
| `activity.status` | Filter by activity status: `running` / `approval` / `question` / `failed` / `idle`; repeatable |
|
||||
| `meta.updated_after` | Only sessions updated after this time (epoch milliseconds) |
|
||||
| `meta.updated_before` | Only sessions updated before this time (epoch milliseconds) |
|
||||
| `meta.archived` | `true` / `false` (default) / `all` |
|
||||
| `sort` | `meta.updated_at_desc` (default) / `meta.updated_at_asc` / `meta.created_at_desc` |
|
||||
| `include` | Comma-separated extra field groups; currently only `git` (branch and PR info, deduplicated per directory and cached for 60 seconds) |
|
||||
| `page_size` | 1–100, default 50 |
|
||||
| `fields` | Comma-separated item projection; currently only `id,archived`, trimming each item to `{ id, archived }` (select-all-matching flows). Not combinable with `include=git` (`40001`) |
|
||||
| `page_size` | 1–100, default 50; up to 10000 with the `id,archived` projection |
|
||||
| `page_token` | Pagination token from the previous page |
|
||||
| `page` | Stateless 1-based page number; mutually exclusive with `page_token` (`40001` when combined) |
|
||||
|
||||
Every response item carries the `workspace`, `meta`, and `activity` groups, plus `git` when `include=git`. The page token binds the first page's query conditions; changing them mid-pagination returns `40922`.
|
||||
Every response item carries the `workspace`, `meta`, and `activity` groups, plus `git` when `include=git` — or just `{ id, archived }` under `fields=id,archived`. Every page additionally carries `total`, the size of the filtered set. The page token binds the first page's query conditions (including the projection); changing them mid-pagination returns `40922`. `page` mode is a stateless alternative for jumping to arbitrary pages: every request is an independent snapshot, no token is minted, and `next_page_token` is always `null`.
|
||||
|
||||
### `POST /api/v2/sessions:archive` and `POST /api/v2/sessions:restore`
|
||||
|
||||
Batch archive/restore for session-management views. The body is `{ "ids": ["session_..."] }` — non-empty, at most 5000 unique ids (duplicates collapse). Live sessions go through the full lifecycle; cold sessions are patched on disk without being loaded.
|
||||
|
||||
Only a body validation failure fails the whole request (`40001`). Otherwise the response is per-item: `data.results` keeps the input order with `{ id, ok }` or `{ id, ok: false, error }` (an unknown id reports `40401` in its own item), plus `succeeded` / `failed` counts.
|
||||
|
||||
```json
|
||||
{
|
||||
"code": 0,
|
||||
"msg": "success",
|
||||
"data": {
|
||||
"results": [
|
||||
{ "id": "session_a", "ok": true },
|
||||
{ "id": "session_b", "ok": false, "error": { "code": 40401, "message": "session session_b does not exist" } }
|
||||
],
|
||||
"succeeded": 1,
|
||||
"failed": 1
|
||||
},
|
||||
"request_id": "req_..."
|
||||
}
|
||||
```
|
||||
|
||||
## WebSocket protocol
|
||||
|
||||
|
|
|
|||
|
|
@ -77,7 +77,7 @@ HTTP 状态码几乎总是 200，业务结果以 `code` 为准。例外情况：
|
|||
列表端点有两种分页风格：
|
||||
|
||||
- **游标式**：`before_id` / `after_id`（互斥）加 `page_size`（1–100），响应为 `{ items, has_more }`。用于会话列表、消息列表、转录等。
|
||||
- **`page_token`**：不透明令牌（内部绑定了查询条件指纹），用于 `POST /api/v1/search` 与 `GET /api/v2/sessions`。翻页途中改变任何查询条件会使令牌失效：v2 返回 `40922`，search 返回 `40001`。
|
||||
- **`page_token`**：不透明令牌（内部绑定了查询条件指纹），用于 `POST /api/v1/search` 与 `GET /api/v2/sessions`。翻页途中改变任何查询条件会使令牌失效：v2 返回 `40922`，search 返回 `40001`。`GET /api/v2/sessions` 另提供无状态的 `page` 页码模式作为替代。
|
||||
|
||||
## REST 端点
|
||||
|
||||
|
|
@ -245,6 +245,8 @@ PTY 终端接口，仅 loopback 绑定时挂载。
|
|||
| `POST /api/v1/search` | 跨会话全文搜索，`mode` 为 `terms`（默认）或 `literal`（精确子串），`page_token` 分页 |
|
||||
| `GET /api/v1/connections` | 列出当前在线的 WebSocket 连接 |
|
||||
| `GET /api/v2/sessions` | 新一代会话列表，见下节 |
|
||||
| `POST /api/v2/sessions:archive` | 批量归档会话，见下节 |
|
||||
| `POST /api/v2/sessions:restore` | 批量恢复已归档会话，见下节 |
|
||||
| `/api/v1/debug/*` | 反射式调试 RPC，仅 `--debug-endpoints` 且 loopback 时挂载，不属于稳定协议 |
|
||||
|
||||
### `GET /api/v2/sessions`
|
||||
|
|
@ -256,13 +258,38 @@ PTY 终端接口，仅 loopback 绑定时挂载。
|
|||
| `workspace.id` | 按工作区过滤，可重复 |
|
||||
| `activity.status` | 按活动状态过滤：`running` / `approval` / `question` / `failed` / `idle`，可重复 |
|
||||
| `meta.updated_after` | 只看该时间（epoch 毫秒）之后更新过的会话 |
|
||||
| `meta.updated_before` | 只看该时间（epoch 毫秒）之前更新过的会话 |
|
||||
| `meta.archived` | `true` / `false`（默认）/ `all` |
|
||||
| `sort` | `meta.updated_at_desc`（默认）/ `meta.updated_at_asc` / `meta.created_at_desc` |
|
||||
| `include` | 逗号分隔的附加字段组；目前支持 `git`（分支与 PR 信息，按目录去重并缓存 60 秒） |
|
||||
| `page_size` | 1–100，默认 50 |
|
||||
| `fields` | 逗号分隔的字段投影；目前仅支持 `id,archived`，每项裁剪为 `{ id, archived }`（用于全选匹配场景）。不可与 `include=git` 同传（`40001`） |
|
||||
| `page_size` | 1–100，默认 50；使用 `id,archived` 投影时上限放宽至 10000 |
|
||||
| `page_token` | 上一页返回的翻页令牌 |
|
||||
| `page` | 无状态的 1 起始页码；与 `page_token` 互斥（同传返回 `40001`） |
|
||||
|
||||
响应每项固定包含 `workspace`、`meta`、`activity` 三组，`include=git` 时附加 `git` 组。翻页令牌绑定首页查询条件，中途改条件返回 `40922`。
|
||||
响应每项固定包含 `workspace`、`meta`、`activity` 三组，`include=git` 时附加 `git` 组；`fields=id,archived` 时仅返回 `{ id, archived }`。每页额外携带 `total`，即过滤后的集合大小。翻页令牌绑定首页查询条件（含投影），中途改条件返回 `40922`。`page` 模式是跳页用的无状态替代：每次请求都是独立快照，不签发令牌，`next_page_token` 恒为 `null`。
|
||||
|
||||
### `POST /api/v2/sessions:archive` 与 `POST /api/v2/sessions:restore`
|
||||
|
||||
面向会话管理页的批量归档/恢复。请求体为 `{ "ids": ["session_..."] }`——非空、去重后不超过 5000 条。仍在线的会话走完整生命周期；未加载的冷会话直接改写磁盘上的元数据，不会被加载。
|
||||
|
||||
只有请求体校验失败才会让整个请求失败（`40001`）；其余情况按条返回：`data.results` 保持输入顺序，每项为 `{ id, ok }` 或 `{ id, ok: false, error }`（不存在的 id 在自身条目里报 `40401`），并附 `succeeded` / `failed` 计数。
|
||||
|
||||
```json
|
||||
{
|
||||
"code": 0,
|
||||
"msg": "success",
|
||||
"data": {
|
||||
"results": [
|
||||
{ "id": "session_a", "ok": true },
|
||||
{ "id": "session_b", "ok": false, "error": { "code": 40401, "message": "session session_b does not exist" } }
|
||||
],
|
||||
"succeeded": 1,
|
||||
"failed": 1
|
||||
},
|
||||
"request_id": "req_..."
|
||||
}
|
||||
```
|
||||
|
||||
## WebSocket 协议
|
||||
|
||||
|
|
|
|||
|
|
@ -18,6 +18,11 @@ export interface CreateManagedSessionOptions extends CreateSessionOptions {
|
|||
readonly workspaceId?: string;
|
||||
}
|
||||
|
||||
export interface UnguardedSessionLifecycle {
|
||||
archive(): Promise<void>;
|
||||
restore(): Promise<ISessionScopeHandle | undefined>;
|
||||
}
|
||||
|
||||
export interface ISessionManager {
|
||||
readonly _serviceBrand: undefined;
|
||||
readonly onWillCreateSession?: Event<SessionWillCreateEvent>;
|
||||
|
|
@ -29,6 +34,11 @@ export interface ISessionManager {
|
|||
create(options: CreateManagedSessionOptions): Promise<ISessionScopeHandle>;
|
||||
resume(sessionId: string, options?: ResumeSessionOptions): Promise<ISessionScopeHandle | undefined>;
|
||||
get(sessionId: string): ISessionScopeHandle | undefined;
|
||||
whenResumeSettled(sessionId: string): Promise<void>;
|
||||
withLifecycleSerialization<T>(
|
||||
sessionId: string,
|
||||
work: (unguarded: UnguardedSessionLifecycle) => Promise<T>,
|
||||
): Promise<T>;
|
||||
list(): readonly ISessionScopeHandle[];
|
||||
close(sessionId: string): Promise<void>;
|
||||
archive(sessionId: string): Promise<void>;
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
|
||||
import { DisposableStore } from '#/_base/di/lifecycle';
|
||||
import { Emitter, type Event, type IWaitUntil } from '#/_base/event';
|
||||
import { ScopeActivation, registerScopedService, type ISessionScopeHandle } from '#/_base/di/scope';
|
||||
|
|
@ -18,7 +19,11 @@ import {
|
|||
import type { SessionLifecycleService } from '#/workspace/sessionLifecycle/sessionLifecycleService';
|
||||
import { IWorkspaceInstanceManager } from '#/workspace/workspaceInstance/workspaceInstanceManager';
|
||||
|
||||
import { ISessionManager, type CreateManagedSessionOptions } from './sessionManager';
|
||||
import {
|
||||
ISessionManager,
|
||||
type CreateManagedSessionOptions,
|
||||
type UnguardedSessionLifecycle,
|
||||
} from './sessionManager';
|
||||
|
||||
interface SessionControllerEntry {
|
||||
readonly generation: string;
|
||||
|
|
@ -31,6 +36,9 @@ export class SessionManager implements ISessionManager {
|
|||
declare readonly _serviceBrand: undefined;
|
||||
private readonly sessions = new Map<string, ISessionScopeHandle>();
|
||||
private readonly owners = new Map<string, SessionLifecycleService>();
|
||||
private readonly pendingResumes = new Map<string, Promise<ISessionScopeHandle | undefined>>();
|
||||
private readonly resumeFailures = new Map<string, Error>();
|
||||
private readonly lifecycleChains = new Map<string, Promise<void>>();
|
||||
private readonly controllers = new Map<string, SessionControllerEntry>();
|
||||
private readonly controllerEntries = new Set<SessionControllerEntry>();
|
||||
private readonly willCreateEmitter = new Emitter<SessionWillCreateEvent>();
|
||||
|
|
@ -57,61 +65,139 @@ export class SessionManager implements ISessionManager {
|
|||
? { root: options.workDir }
|
||||
: { workspaceId: options.workspaceId, root: options.workDir },
|
||||
);
|
||||
return this.controllerForWorkspace(workspace.id).create(options);
|
||||
const controller = this.controllerForWorkspace(workspace.id);
|
||||
if (options.sessionId === undefined) return controller.create(options);
|
||||
return this.serializeLifecycle(options.sessionId, () => controller.create(options));
|
||||
}
|
||||
|
||||
async resume(sessionId: string, options?: ResumeSessionOptions): Promise<ISessionScopeHandle | undefined> {
|
||||
return (await this.controllerForSession(sessionId))?.resume(sessionId, options);
|
||||
const inflight = this.pendingResumes.get(sessionId);
|
||||
if (inflight !== undefined) return inflight;
|
||||
this.resumeFailures.delete(sessionId);
|
||||
const promise = this.serializeLifecycle(sessionId, async () =>
|
||||
(await this.controllerForSession(sessionId))?.resume(sessionId, options),
|
||||
).finally(() => this.pendingResumes.delete(sessionId));
|
||||
this.pendingResumes.set(sessionId, promise);
|
||||
void promise.catch((error: unknown) => {
|
||||
this.resumeFailures.set(sessionId, error instanceof Error ? error : new Error('session resume failed'));
|
||||
});
|
||||
return promise;
|
||||
}
|
||||
|
||||
get(sessionId: string): ISessionScopeHandle | undefined {
|
||||
return this.sessions.get(sessionId);
|
||||
}
|
||||
|
||||
async whenResumeSettled(sessionId: string): Promise<void> {
|
||||
await this.pendingResumes.get(sessionId);
|
||||
const failure = this.resumeFailures.get(sessionId);
|
||||
if (failure !== undefined) throw failure;
|
||||
await this.owners.get(sessionId)?.whenResumeSettled(sessionId);
|
||||
}
|
||||
|
||||
private serializeLifecycle<T>(sessionId: string, work: () => Promise<T>): Promise<T> {
|
||||
const prev = this.lifecycleChains.get(sessionId) ?? Promise.resolve();
|
||||
const run = prev.then(work, work);
|
||||
const next = run.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
this.lifecycleChains.set(sessionId, next);
|
||||
void next.finally(() => {
|
||||
if (this.lifecycleChains.get(sessionId) === next) this.lifecycleChains.delete(sessionId);
|
||||
});
|
||||
return run;
|
||||
}
|
||||
|
||||
private serializeLifecycleForKeys<T>(keys: readonly string[], work: () => Promise<T>): Promise<T> {
|
||||
const [first, ...rest] = keys;
|
||||
if (first === undefined) return work();
|
||||
return this.serializeLifecycle(first, () => this.serializeLifecycleForKeys(rest, work));
|
||||
}
|
||||
|
||||
private lifecycleKeys(...ids: (string | undefined)[]): string[] {
|
||||
return [...new Set(ids.filter((id): id is string => id !== undefined))].sort();
|
||||
}
|
||||
|
||||
withLifecycleSerialization<T>(
|
||||
sessionId: string,
|
||||
work: (unguarded: UnguardedSessionLifecycle) => Promise<T>,
|
||||
): Promise<T> {
|
||||
return this.serializeLifecycle(sessionId, () =>
|
||||
work({
|
||||
archive: () => this.archiveInner(sessionId),
|
||||
restore: () => this.restoreInner(sessionId),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
list(): readonly ISessionScopeHandle[] {
|
||||
return [...this.sessions.values()];
|
||||
}
|
||||
|
||||
async close(sessionId: string): Promise<void> {
|
||||
await this.owners.get(sessionId)?.close(sessionId);
|
||||
await this.serializeLifecycle(sessionId, async () => this.owners.get(sessionId)?.close(sessionId));
|
||||
}
|
||||
|
||||
async archive(sessionId: string): Promise<void> {
|
||||
private async archiveInner(sessionId: string): Promise<void> {
|
||||
await (await this.controllerForSession(sessionId))?.archive(sessionId);
|
||||
}
|
||||
|
||||
async restore(sessionId: string, options?: ResumeSessionOptions): Promise<ISessionScopeHandle | undefined> {
|
||||
async archive(sessionId: string): Promise<void> {
|
||||
await this.serializeLifecycle(sessionId, () => this.archiveInner(sessionId));
|
||||
}
|
||||
|
||||
private async restoreInner(
|
||||
sessionId: string,
|
||||
options?: ResumeSessionOptions,
|
||||
): Promise<ISessionScopeHandle | undefined> {
|
||||
return (await this.controllerForSession(sessionId))?.restore(sessionId, options);
|
||||
}
|
||||
|
||||
async restore(sessionId: string, options?: ResumeSessionOptions): Promise<ISessionScopeHandle | undefined> {
|
||||
return this.serializeLifecycle(sessionId, () => this.restoreInner(sessionId, options));
|
||||
}
|
||||
|
||||
async delete(sessionId: string): Promise<void> {
|
||||
const controller = await this.controllerForSession(sessionId);
|
||||
if (controller === undefined) {
|
||||
throw new Error2(ErrorCodes.SESSION_NOT_FOUND, `session ${sessionId} does not exist`);
|
||||
}
|
||||
await controller.delete(sessionId);
|
||||
await this.serializeLifecycle(sessionId, async () => {
|
||||
const controller = await this.controllerForSession(sessionId);
|
||||
if (controller === undefined) {
|
||||
throw new Error2(ErrorCodes.SESSION_NOT_FOUND, `session ${sessionId} does not exist`);
|
||||
}
|
||||
await controller.delete(sessionId);
|
||||
});
|
||||
}
|
||||
|
||||
async fork(options: ForkSessionOptions): Promise<ISessionScopeHandle> {
|
||||
const controller = await this.controllerForSession(options.sourceSessionId);
|
||||
if (controller === undefined) {
|
||||
throw new Error2(
|
||||
ErrorCodes.SESSION_NOT_FOUND,
|
||||
`session ${options.sourceSessionId} does not exist`,
|
||||
);
|
||||
}
|
||||
return controller.fork(options);
|
||||
return this.serializeLifecycleForKeys(
|
||||
this.lifecycleKeys(options.sourceSessionId, options.newSessionId),
|
||||
async () => {
|
||||
const controller = await this.controllerForSession(options.sourceSessionId);
|
||||
if (controller === undefined) {
|
||||
throw new Error2(
|
||||
ErrorCodes.SESSION_NOT_FOUND,
|
||||
`session ${options.sourceSessionId} does not exist`,
|
||||
);
|
||||
}
|
||||
return controller.fork(options);
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
async createChild(options: CreateChildSessionOptions): Promise<ISessionScopeHandle> {
|
||||
const controller = await this.controllerForSession(options.sourceSessionId);
|
||||
if (controller === undefined) {
|
||||
throw new Error2(
|
||||
ErrorCodes.SESSION_NOT_FOUND,
|
||||
`session ${options.sourceSessionId} does not exist`,
|
||||
);
|
||||
}
|
||||
return controller.createChild(options);
|
||||
return this.serializeLifecycleForKeys(
|
||||
this.lifecycleKeys(options.sourceSessionId, options.newSessionId),
|
||||
async () => {
|
||||
const controller = await this.controllerForSession(options.sourceSessionId);
|
||||
if (controller === undefined) {
|
||||
throw new Error2(
|
||||
ErrorCodes.SESSION_NOT_FOUND,
|
||||
`session ${options.sourceSessionId} does not exist`,
|
||||
);
|
||||
}
|
||||
return controller.createChild(options);
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
|
|
|
|||
|
|
@ -451,6 +451,7 @@ export * from '#/workspace/workspaceContext/workspaceContext';
|
|||
export * from '#/workspace/sessionLifecycle/sessionLifecycle';
|
||||
export * from '#/workspace/sessionLifecycle/sessionLifecycleEvents';
|
||||
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';
|
||||
|
|
|
|||
|
|
@ -292,7 +292,7 @@ function isSessionTitleKind(value: unknown): value is SessionTitleKind {
|
|||
|
||||
type PersistedSessionMeta = SessionMeta & { readonly isCustomTitle: boolean };
|
||||
|
||||
function encodeSessionMeta(meta: SessionMeta): PersistedSessionMeta {
|
||||
export function encodeSessionMeta(meta: SessionMeta): PersistedSessionMeta {
|
||||
return { ...meta, isCustomTitle: meta.titleKind === 'custom' };
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,112 @@
|
|||
|
||||
import type { ServicesAccessor } from '#/_base/di/instantiation';
|
||||
import { IBootstrapService } from '#/app/bootstrap/bootstrap';
|
||||
import { IEventService } from '#/app/event/event';
|
||||
import { ISessionManager } from '#/app/sessionManager/sessionManager';
|
||||
import { getLiveSessionById } from '#/app/sessionManager/sessionLookup';
|
||||
import { ISessionIndex, ISessionIndexMirror } from '#/app/sessionIndex/sessionIndex';
|
||||
import { buildSessionSummary } from '#/app/sessionIndex/sessionIndexSource';
|
||||
import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
|
||||
import type { SessionMeta } from '#/session/sessionMetadata/sessionMetadata';
|
||||
import { normalizeSessionMeta, encodeSessionMeta } from '#/session/sessionMetadata/sessionMetadataService';
|
||||
|
||||
import { sessionScopeOf, legacySessionMetaScopeOf, workspacePersistenceScope } from './internal/addressing';
|
||||
import { SessionArchived } from './sessionLifecycleEvents';
|
||||
|
||||
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 raw = await docs.get<SessionMeta>(metaScope, 'state.json');
|
||||
let legacyMetaScope: string | undefined;
|
||||
if (raw === undefined) {
|
||||
legacyMetaScope = legacySessionMetaScopeOf(metaScope);
|
||||
raw = await docs.get<SessionMeta>(legacyMetaScope, 'state.json');
|
||||
}
|
||||
if (raw === undefined) return 'not_found';
|
||||
const persisted = normalizeSessionMeta(raw, sessionId);
|
||||
const archivedAt = archived ? Date.now() : undefined;
|
||||
const nextMeta: SessionMeta = { ...persisted, archived, archivedAt };
|
||||
await docs.set(metaScope, 'state.json', encodeSessionMeta(nextMeta));
|
||||
if (legacyMetaScope !== undefined) await docs.delete(legacyMetaScope, 'state.json');
|
||||
accessor.get(ISessionIndexMirror).record(
|
||||
buildSessionSummary({
|
||||
id: sessionId,
|
||||
workspaceId: summary.workspaceId,
|
||||
cwd: nextMeta.cwd ?? summary.cwd,
|
||||
title: nextMeta.title,
|
||||
lastPrompt: nextMeta.lastPrompt,
|
||||
createdAt: nextMeta.createdAt,
|
||||
updatedAt: nextMeta.updatedAt,
|
||||
archived,
|
||||
archivedAt,
|
||||
custom: nextMeta.custom,
|
||||
lastTurnReason: nextMeta.lastTurnReason,
|
||||
}),
|
||||
);
|
||||
if (archived) {
|
||||
accessor.get(IEventService).publish(new SessionArchived({ payload: { sessionId } }));
|
||||
}
|
||||
return 'updated';
|
||||
}
|
||||
|
||||
export type SessionArchiveBatchItemOutcome =
|
||||
| { id: string; ok: true }
|
||||
| { id: string; ok: false; reason: 'not_found' | 'error'; message: string };
|
||||
|
||||
export async function setSessionArchivedBatch(
|
||||
accessor: ServicesAccessor,
|
||||
ids: readonly string[],
|
||||
archived: boolean,
|
||||
): Promise<SessionArchiveBatchItemOutcome[]> {
|
||||
const outcomes: (SessionArchiveBatchItemOutcome | undefined)[] = ids.map(() => undefined);
|
||||
const applyOne = async (id: string): Promise<SessionArchiveBatchItemOutcome> => {
|
||||
try {
|
||||
const manager = accessor.get(ISessionManager);
|
||||
return await manager.withLifecycleSerialization(id, async (unguarded) => {
|
||||
await manager.whenResumeSettled(id);
|
||||
const live = getLiveSessionById(accessor, id);
|
||||
if (live !== undefined) {
|
||||
if (archived) await unguarded.archive();
|
||||
else await unguarded.restore();
|
||||
return { id, ok: true };
|
||||
}
|
||||
const outcome = await setColdSessionArchived(accessor, id, archived);
|
||||
return outcome === 'updated'
|
||||
? { id, ok: true }
|
||||
: { id, ok: false, reason: 'not_found', message: `session ${id} does not exist` };
|
||||
});
|
||||
} catch (error) {
|
||||
return {
|
||||
id,
|
||||
ok: false,
|
||||
reason: 'error',
|
||||
message: error instanceof Error ? error.message : String(error),
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
const BATCH_CONCURRENCY = 8;
|
||||
let next = 0;
|
||||
const workers = Array.from({ length: Math.min(BATCH_CONCURRENCY, ids.length) }, async () => {
|
||||
while (next < ids.length) {
|
||||
const index = next++;
|
||||
outcomes[index] = await applyOne(ids[index] as string);
|
||||
}
|
||||
});
|
||||
await Promise.all(workers);
|
||||
return outcomes as SessionArchiveBatchItemOutcome[];
|
||||
}
|
||||
|
|
@ -15,3 +15,7 @@ export function sessionDirOf(homeDir: string, handlerScope: string, sessionId: s
|
|||
export function agentScopeOf(sessionScope: string, agentId: string): string {
|
||||
return `${sessionScope}/agents/${agentId}`;
|
||||
}
|
||||
|
||||
export function legacySessionMetaScopeOf(sessionScope: string): string {
|
||||
return `${sessionScope}/session-meta`;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -133,6 +133,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
|
|||
private readonly _onDidForkSession = this._register(new Emitter<SessionForkedEvent>());
|
||||
readonly onDidForkSession: Event<SessionForkedEvent> = this._onDidForkSession.event;
|
||||
private readonly resuming = new Map<string, Promise<ISessionScopeHandle | undefined>>();
|
||||
private readonly resumeFailures = new Map<string, Error>();
|
||||
|
||||
constructor(
|
||||
private readonly instantiation: IInstantiationService,
|
||||
|
|
@ -311,6 +312,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
|
|||
if (inflight !== undefined) return inflight;
|
||||
const live = this.sessions.get(sessionId);
|
||||
if (live !== undefined) return Promise.resolve(live);
|
||||
this.resumeFailures.delete(sessionId);
|
||||
const promise = this.doResume(sessionId, opts)
|
||||
.catch((error: unknown) => {
|
||||
this.telemetry
|
||||
|
|
@ -318,6 +320,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
|
|||
.track2('session_load_failed', {
|
||||
reason: isError2(error) ? error.code : error instanceof Error ? error.name : 'unknown',
|
||||
});
|
||||
this.resumeFailures.set(sessionId, error instanceof Error ? error : new Error('session resume failed'));
|
||||
throw error;
|
||||
})
|
||||
.finally(() => this.resuming.delete(sessionId));
|
||||
|
|
@ -325,6 +328,12 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
|
|||
return promise;
|
||||
}
|
||||
|
||||
async whenResumeSettled(sessionId: string): Promise<void> {
|
||||
await this.resuming.get(sessionId);
|
||||
const failure = this.resumeFailures.get(sessionId);
|
||||
if (failure !== undefined) throw failure;
|
||||
}
|
||||
|
||||
private async doResume(
|
||||
sessionId: string,
|
||||
opts?: ResumeSessionOptions,
|
||||
|
|
@ -342,11 +351,17 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
|
|||
additionalDirs: opts?.additionalDirs,
|
||||
mcpServers: opts?.mcpServers,
|
||||
});
|
||||
const agents = handle.accessor.get(IAgentLifecycleService);
|
||||
if (agents.get(MAIN_AGENT_ID) === undefined) {
|
||||
await agents.create({ agentId: MAIN_AGENT_ID });
|
||||
try {
|
||||
const agents = handle.accessor.get(IAgentLifecycleService);
|
||||
if (agents.get(MAIN_AGENT_ID) === undefined) {
|
||||
await agents.create({ agentId: MAIN_AGENT_ID });
|
||||
}
|
||||
await this.announceCreated({ sessionId, handle, source: 'resume' });
|
||||
} catch (error) {
|
||||
this.sessions.delete(sessionId);
|
||||
handle.dispose();
|
||||
throw error;
|
||||
}
|
||||
await this.announceCreated({ sessionId, handle, source: 'resume' });
|
||||
return handle;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ import {
|
|||
} from '#/app/sessionExport/sessionExportService';
|
||||
import { writeExportZip } from '#/app/sessionExport/zip';
|
||||
import { ISessionIndex, type SessionSummary } from '#/app/sessionIndex/sessionIndex';
|
||||
import { ISessionManager } from '#/app/sessionManager/sessionManager';
|
||||
import { ISessionManager, type UnguardedSessionLifecycle } from '#/app/sessionManager/sessionManager';
|
||||
import { ISessionLifecycleService } from '#/workspace/sessionLifecycle/sessionLifecycle';
|
||||
import { IWorkspaceService } from '#/app/workspace/workspace';
|
||||
import { Error2 } from '#/errors';
|
||||
|
|
@ -890,6 +890,11 @@ function registerSessionExportServices(
|
|||
},
|
||||
resume: async () => options.lifecycleHandle,
|
||||
get: () => options.lifecycleHandle,
|
||||
whenResumeSettled: async () => {},
|
||||
withLifecycleSerialization: async <T>(
|
||||
_sessionId: string,
|
||||
work: (unguarded: UnguardedSessionLifecycle) => Promise<T>,
|
||||
): Promise<T> => work({ archive: async () => {}, restore: async () => undefined }),
|
||||
list: () => (options.lifecycleHandle === undefined ? [] : [options.lifecycleHandle]),
|
||||
close: async () => {},
|
||||
archive: async () => {},
|
||||
|
|
|
|||
|
|
@ -51,7 +51,366 @@ function controller(sessionId = 'session-1'): {
|
|||
return { service, handle };
|
||||
}
|
||||
|
||||
async function drainMicrotasks(ticks = 50): Promise<void> {
|
||||
for (let i = 0; i < ticks; i++) await Promise.resolve();
|
||||
}
|
||||
|
||||
describe('SessionManager', () => {
|
||||
it('serializes resume, close, and lifecycle critical sections per session', async () => {
|
||||
const didCreate = new Emitter<SessionCreatedEvent>();
|
||||
const didClose = new Emitter<SessionClosedEvent>();
|
||||
const handle = { id: 'session-1' } as unknown as ISessionScopeHandle;
|
||||
let releaseResume!: () => void;
|
||||
const resumeGate = new Promise<void>((resolve) => {
|
||||
releaseResume = resolve;
|
||||
});
|
||||
const order: string[] = [];
|
||||
const service = {
|
||||
onWillCreateSession: Event.None,
|
||||
onDidCreateSession: didCreate.event,
|
||||
onWillCloseSession: Event.None,
|
||||
onDidCloseSession: didClose.event,
|
||||
onDidArchiveSession: Event.None,
|
||||
onDidForkSession: Event.None,
|
||||
create: async () => handle,
|
||||
get: () => undefined,
|
||||
list: () => [],
|
||||
resume: async () => {
|
||||
order.push('resume:start');
|
||||
await resumeGate;
|
||||
didCreate.fire({ sessionId: 'session-1', handle, source: 'startup' });
|
||||
order.push('resume:end');
|
||||
return handle;
|
||||
},
|
||||
close: async () => {
|
||||
order.push('close');
|
||||
didClose.fire({ sessionId: 'session-1' });
|
||||
},
|
||||
archive: async () => {},
|
||||
restore: async () => handle,
|
||||
delete: async () => {},
|
||||
fork: async () => handle,
|
||||
createChild: async () => handle,
|
||||
dispose: () => {},
|
||||
} as unknown as SessionLifecycleService;
|
||||
const workspace = {
|
||||
id: 'workspace-1',
|
||||
program: { sessionControllerGeneration: 'generation-1', createSessionController: () => service },
|
||||
} as unknown as WorkspaceInstance;
|
||||
const workspaces = {
|
||||
getOrCreate: async () => workspace,
|
||||
get: () => workspace,
|
||||
} as unknown as IWorkspaceInstanceManager;
|
||||
const index = {
|
||||
get: async () => ({ workspaceId: 'workspace-1', cwd: '/workspace' }),
|
||||
} as unknown as ISessionIndex;
|
||||
const manager = new SessionManager(workspaces, index);
|
||||
|
||||
const resumePromise = manager.resume('session-1');
|
||||
const section = manager.withLifecycleSerialization('session-1', async () => {
|
||||
order.push('section');
|
||||
});
|
||||
const closePromise = manager.close('session-1');
|
||||
await drainMicrotasks();
|
||||
expect(order).toEqual(['resume:start']);
|
||||
releaseResume();
|
||||
await Promise.all([resumePromise, section, closePromise]);
|
||||
expect(order).toEqual(['resume:start', 'resume:end', 'section', 'close']);
|
||||
manager.dispose();
|
||||
});
|
||||
|
||||
it('holds a resume started during a lifecycle critical section', async () => {
|
||||
const fake = controller();
|
||||
const workspace = {
|
||||
id: 'workspace-1',
|
||||
program: { sessionControllerGeneration: 'generation-1', createSessionController: () => fake.service },
|
||||
} as unknown as WorkspaceInstance;
|
||||
const workspaces = {
|
||||
getOrCreate: async () => workspace,
|
||||
get: () => workspace,
|
||||
} as unknown as IWorkspaceInstanceManager;
|
||||
const index = {
|
||||
get: async () => ({ workspaceId: 'workspace-1', cwd: '/workspace' }),
|
||||
} as unknown as ISessionIndex;
|
||||
const manager = new SessionManager(workspaces, index);
|
||||
|
||||
let releaseSection!: () => void;
|
||||
const sectionGate = new Promise<void>((resolve) => {
|
||||
releaseSection = resolve;
|
||||
});
|
||||
const order: string[] = [];
|
||||
const section = manager.withLifecycleSerialization('session-1', async () => {
|
||||
order.push('section:start');
|
||||
await sectionGate;
|
||||
order.push('section:end');
|
||||
});
|
||||
const resumePromise = manager.resume('session-1').then((handle) => {
|
||||
order.push('resume');
|
||||
return handle;
|
||||
});
|
||||
await drainMicrotasks();
|
||||
expect(order).toEqual(['section:start']);
|
||||
releaseSection();
|
||||
await Promise.all([section, resumePromise]);
|
||||
expect(order).toEqual(['section:start', 'section:end', 'resume']);
|
||||
manager.dispose();
|
||||
});
|
||||
|
||||
it('serializes delete with the per-session lifecycle chain', async () => {
|
||||
const order: string[] = [];
|
||||
const fake = controller();
|
||||
(fake.service as unknown as { delete: () => Promise<void> }).delete = async () => {
|
||||
order.push('delete');
|
||||
};
|
||||
const workspace = {
|
||||
id: 'workspace-1',
|
||||
program: { sessionControllerGeneration: 'generation-1', createSessionController: () => fake.service },
|
||||
} as unknown as WorkspaceInstance;
|
||||
const workspaces = {
|
||||
getOrCreate: async () => workspace,
|
||||
get: () => workspace,
|
||||
} as unknown as IWorkspaceInstanceManager;
|
||||
const index = {
|
||||
get: async () => ({ workspaceId: 'workspace-1', cwd: '/workspace' }),
|
||||
} as unknown as ISessionIndex;
|
||||
const manager = new SessionManager(workspaces, index);
|
||||
|
||||
let releaseSection!: () => void;
|
||||
const sectionGate = new Promise<void>((resolve) => {
|
||||
releaseSection = resolve;
|
||||
});
|
||||
const section = manager.withLifecycleSerialization('session-1', async () => {
|
||||
order.push('section:start');
|
||||
await sectionGate;
|
||||
order.push('section:end');
|
||||
});
|
||||
const deletePromise = manager.delete('session-1');
|
||||
await drainMicrotasks();
|
||||
expect(order).toEqual(['section:start']);
|
||||
releaseSection();
|
||||
await Promise.all([section, deletePromise]);
|
||||
expect(order).toEqual(['section:start', 'section:end', 'delete']);
|
||||
manager.dispose();
|
||||
});
|
||||
|
||||
it('serializes fork of the source session with the lifecycle chain', async () => {
|
||||
const order: string[] = [];
|
||||
const fake = controller();
|
||||
(fake.service as unknown as { fork: () => Promise<unknown> }).fork = async () => {
|
||||
order.push('fork');
|
||||
return fake.handle;
|
||||
};
|
||||
const workspace = {
|
||||
id: 'workspace-1',
|
||||
program: { sessionControllerGeneration: 'generation-1', createSessionController: () => fake.service },
|
||||
} as unknown as WorkspaceInstance;
|
||||
const workspaces = {
|
||||
getOrCreate: async () => workspace,
|
||||
get: () => workspace,
|
||||
} as unknown as IWorkspaceInstanceManager;
|
||||
const index = {
|
||||
get: async () => ({ workspaceId: 'workspace-1', cwd: '/workspace' }),
|
||||
} as unknown as ISessionIndex;
|
||||
const manager = new SessionManager(workspaces, index);
|
||||
|
||||
let releaseSection!: () => void;
|
||||
const sectionGate = new Promise<void>((resolve) => {
|
||||
releaseSection = resolve;
|
||||
});
|
||||
const section = manager.withLifecycleSerialization('session-1', async () => {
|
||||
order.push('section:start');
|
||||
await sectionGate;
|
||||
order.push('section:end');
|
||||
});
|
||||
const forkPromise = manager.fork({ sourceSessionId: 'session-1' } as never);
|
||||
await drainMicrotasks();
|
||||
expect(order).toEqual(['section:start']);
|
||||
releaseSection();
|
||||
await Promise.all([section, forkPromise]);
|
||||
expect(order).toEqual(['section:start', 'section:end', 'fork']);
|
||||
manager.dispose();
|
||||
});
|
||||
|
||||
it('serializes fork of an explicit target id with the lifecycle chain', async () => {
|
||||
const order: string[] = [];
|
||||
const fake = controller();
|
||||
(fake.service as unknown as { fork: () => Promise<unknown> }).fork = async () => {
|
||||
order.push('fork');
|
||||
return fake.handle;
|
||||
};
|
||||
const workspace = {
|
||||
id: 'workspace-1',
|
||||
program: { sessionControllerGeneration: 'generation-1', createSessionController: () => fake.service },
|
||||
} as unknown as WorkspaceInstance;
|
||||
const workspaces = {
|
||||
getOrCreate: async () => workspace,
|
||||
get: () => workspace,
|
||||
} as unknown as IWorkspaceInstanceManager;
|
||||
const index = {
|
||||
get: async () => ({ workspaceId: 'workspace-1', cwd: '/workspace' }),
|
||||
} as unknown as ISessionIndex;
|
||||
const manager = new SessionManager(workspaces, index);
|
||||
|
||||
let releaseSection!: () => void;
|
||||
const sectionGate = new Promise<void>((resolve) => {
|
||||
releaseSection = resolve;
|
||||
});
|
||||
const section = manager.withLifecycleSerialization('session-2', async () => {
|
||||
order.push('section:start');
|
||||
await sectionGate;
|
||||
order.push('section:end');
|
||||
});
|
||||
const forkPromise = manager.fork({ sourceSessionId: 'session-1', newSessionId: 'session-2' } as never);
|
||||
await drainMicrotasks();
|
||||
expect(order).toEqual(['section:start']);
|
||||
releaseSection();
|
||||
await Promise.all([section, forkPromise]);
|
||||
expect(order).toEqual(['section:start', 'section:end', 'fork']);
|
||||
manager.dispose();
|
||||
});
|
||||
|
||||
it('serializes createChild of an explicit target id with the lifecycle chain', async () => {
|
||||
const order: string[] = [];
|
||||
const fake = controller();
|
||||
(fake.service as unknown as { createChild: () => Promise<unknown> }).createChild = async () => {
|
||||
order.push('createChild');
|
||||
return fake.handle;
|
||||
};
|
||||
const workspace = {
|
||||
id: 'workspace-1',
|
||||
program: { sessionControllerGeneration: 'generation-1', createSessionController: () => fake.service },
|
||||
} as unknown as WorkspaceInstance;
|
||||
const workspaces = {
|
||||
getOrCreate: async () => workspace,
|
||||
get: () => workspace,
|
||||
} as unknown as IWorkspaceInstanceManager;
|
||||
const index = {
|
||||
get: async () => ({ workspaceId: 'workspace-1', cwd: '/workspace' }),
|
||||
} as unknown as ISessionIndex;
|
||||
const manager = new SessionManager(workspaces, index);
|
||||
|
||||
let releaseSection!: () => void;
|
||||
const sectionGate = new Promise<void>((resolve) => {
|
||||
releaseSection = resolve;
|
||||
});
|
||||
const section = manager.withLifecycleSerialization('session-2', async () => {
|
||||
order.push('section:start');
|
||||
await sectionGate;
|
||||
order.push('section:end');
|
||||
});
|
||||
const childPromise = manager.createChild({ sourceSessionId: 'session-1', newSessionId: 'session-2' } as never);
|
||||
await drainMicrotasks();
|
||||
expect(order).toEqual(['section:start']);
|
||||
releaseSection();
|
||||
await Promise.all([section, childPromise]);
|
||||
expect(order).toEqual(['section:start', 'section:end', 'createChild']);
|
||||
manager.dispose();
|
||||
});
|
||||
|
||||
it('serializes create with an explicit session id with the lifecycle chain', async () => {
|
||||
const order: string[] = [];
|
||||
const fake = controller();
|
||||
(fake.service as unknown as { create: () => Promise<unknown> }).create = async () => {
|
||||
order.push('create');
|
||||
return fake.handle;
|
||||
};
|
||||
const workspace = {
|
||||
id: 'workspace-1',
|
||||
program: { sessionControllerGeneration: 'generation-1', createSessionController: () => fake.service },
|
||||
} as unknown as WorkspaceInstance;
|
||||
const workspaces = {
|
||||
getOrCreate: async () => workspace,
|
||||
get: () => workspace,
|
||||
} as unknown as IWorkspaceInstanceManager;
|
||||
const index = {
|
||||
get: async () => ({ workspaceId: 'workspace-1', cwd: '/workspace' }),
|
||||
} as unknown as ISessionIndex;
|
||||
const manager = new SessionManager(workspaces, index);
|
||||
|
||||
let releaseSection!: () => void;
|
||||
const sectionGate = new Promise<void>((resolve) => {
|
||||
releaseSection = resolve;
|
||||
});
|
||||
const section = manager.withLifecycleSerialization('session-1', async () => {
|
||||
order.push('section:start');
|
||||
await sectionGate;
|
||||
order.push('section:end');
|
||||
});
|
||||
const createPromise = manager.create({ sessionId: 'session-1', workDir: '/workspace' } as never);
|
||||
await drainMicrotasks();
|
||||
expect(order).toEqual(['section:start']);
|
||||
releaseSection();
|
||||
await Promise.all([section, createPromise]);
|
||||
expect(order).toEqual(['section:start', 'section:end', 'create']);
|
||||
manager.dispose();
|
||||
});
|
||||
|
||||
it('serializes archive with the per-session lifecycle chain', async () => {
|
||||
const order: string[] = [];
|
||||
const fake = controller();
|
||||
(fake.service as unknown as { archive: () => Promise<void> }).archive = async () => {
|
||||
order.push('archive');
|
||||
};
|
||||
const workspace = {
|
||||
id: 'workspace-1',
|
||||
program: { sessionControllerGeneration: 'generation-1', createSessionController: () => fake.service },
|
||||
} as unknown as WorkspaceInstance;
|
||||
const workspaces = {
|
||||
getOrCreate: async () => workspace,
|
||||
get: () => workspace,
|
||||
} as unknown as IWorkspaceInstanceManager;
|
||||
const index = {
|
||||
get: async () => ({ workspaceId: 'workspace-1', cwd: '/workspace' }),
|
||||
} as unknown as ISessionIndex;
|
||||
const manager = new SessionManager(workspaces, index);
|
||||
|
||||
let releaseSection!: () => void;
|
||||
const sectionGate = new Promise<void>((resolve) => {
|
||||
releaseSection = resolve;
|
||||
});
|
||||
const section = manager.withLifecycleSerialization('session-1', async () => {
|
||||
order.push('section:start');
|
||||
await sectionGate;
|
||||
order.push('section:end');
|
||||
});
|
||||
const archivePromise = manager.archive('session-1');
|
||||
await drainMicrotasks();
|
||||
expect(order).toEqual(['section:start']);
|
||||
releaseSection();
|
||||
await Promise.all([section, archivePromise]);
|
||||
expect(order).toEqual(['section:start', 'section:end', 'archive']);
|
||||
manager.dispose();
|
||||
});
|
||||
|
||||
it('propagates a failed resume to the next settle until a fresh attempt supersedes', async () => {
|
||||
let fail = true;
|
||||
const fake = controller();
|
||||
(fake.service as unknown as { resume: () => Promise<unknown> }).resume = async () => {
|
||||
if (fail) throw new Error('boom');
|
||||
return fake.handle;
|
||||
};
|
||||
const workspace = {
|
||||
id: 'workspace-1',
|
||||
program: { sessionControllerGeneration: 'generation-1', createSessionController: () => fake.service },
|
||||
} as unknown as WorkspaceInstance;
|
||||
const workspaces = {
|
||||
getOrCreate: async () => workspace,
|
||||
get: () => workspace,
|
||||
} as unknown as IWorkspaceInstanceManager;
|
||||
const index = {
|
||||
get: async () => ({ workspaceId: 'workspace-1', cwd: '/workspace' }),
|
||||
} as unknown as ISessionIndex;
|
||||
const manager = new SessionManager(workspaces, index);
|
||||
|
||||
await expect(manager.resume('session-1')).rejects.toThrow('boom');
|
||||
await expect(manager.whenResumeSettled('session-1')).rejects.toThrow('boom');
|
||||
|
||||
fail = false;
|
||||
await manager.resume('session-1');
|
||||
await expect(manager.whenResumeSettled('session-1')).resolves.toBeUndefined();
|
||||
manager.dispose();
|
||||
});
|
||||
|
||||
it('owns one global live-session registry across workspace controllers', async () => {
|
||||
const fake = controller();
|
||||
const workspace = {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,209 @@
|
|||
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import type { ServiceIdentifier, ServicesAccessor } from '#/_base/di/instantiation';
|
||||
import { IBootstrapService } from '#/app/bootstrap/bootstrap';
|
||||
import { IEventService } from '#/app/event/event';
|
||||
import { ISessionManager, type UnguardedSessionLifecycle } from '#/app/sessionManager/sessionManager';
|
||||
import {
|
||||
ISessionIndex,
|
||||
ISessionIndexMirror,
|
||||
type SessionSummary,
|
||||
} from '#/app/sessionIndex/sessionIndex';
|
||||
import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
|
||||
import type { SessionMeta } from '#/session/sessionMetadata/sessionMetadata';
|
||||
import {
|
||||
setSessionArchivedBatch,
|
||||
} from '#/workspace/sessionLifecycle/coldSessionArchive';
|
||||
|
||||
function accessor(
|
||||
entries: ReadonlyArray<readonly [ServiceIdentifier<unknown>, unknown]>,
|
||||
): ServicesAccessor {
|
||||
return {
|
||||
get<T>(id: ServiceIdentifier<T>): T {
|
||||
for (const [key, value] of entries) {
|
||||
if (key === id) return value as T;
|
||||
}
|
||||
throw new Error(`Unexpected service request: ${String(id)}`);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const summary: SessionSummary = {
|
||||
id: 's1',
|
||||
workspaceId: 'wd',
|
||||
cwd: '/workspace',
|
||||
createdAt: 1,
|
||||
updatedAt: 1,
|
||||
archived: false,
|
||||
};
|
||||
|
||||
interface ColdPathOptions {
|
||||
readonly storeGet: (scope: string, key: string) => Promise<SessionMeta | undefined>;
|
||||
readonly indexSummary?: SessionSummary;
|
||||
readonly onMirrorRecord?: (recorded: SessionSummary) => void;
|
||||
readonly onStoreSet?: (scope: string, key: string, value: unknown) => void;
|
||||
readonly onStoreDelete?: (scope: string, key: string) => void;
|
||||
readonly resumeError?: Error;
|
||||
}
|
||||
|
||||
function coldPathAccessor(options: ColdPathOptions): ServicesAccessor {
|
||||
return accessor([
|
||||
[
|
||||
ISessionManager,
|
||||
{
|
||||
withLifecycleSerialization: <T>(
|
||||
_id: string,
|
||||
work: (unguarded: UnguardedSessionLifecycle) => Promise<T>,
|
||||
): Promise<T> => work({ archive: async () => {}, restore: async () => undefined }),
|
||||
whenResumeSettled: async () => {
|
||||
if (options.resumeError !== undefined) throw options.resumeError;
|
||||
},
|
||||
get: () => undefined,
|
||||
},
|
||||
],
|
||||
[ISessionIndex, { get: async () => options.indexSummary ?? summary }],
|
||||
[IBootstrapService, { scope: () => 'sessions' }],
|
||||
[
|
||||
IAtomicDocumentStore,
|
||||
{
|
||||
get: (scope: string, key: string) => options.storeGet(scope, key),
|
||||
set: async (scope: string, key: string, value: unknown) => {
|
||||
options.onStoreSet?.(scope, key, value);
|
||||
},
|
||||
delete: async (scope: string, key: string) => {
|
||||
options.onStoreDelete?.(scope, key);
|
||||
},
|
||||
},
|
||||
],
|
||||
[
|
||||
ISessionIndexMirror,
|
||||
{ record: (recorded: SessionSummary) => options.onMirrorRecord?.(recorded) },
|
||||
],
|
||||
[IEventService, { publish: () => {} }],
|
||||
]);
|
||||
}
|
||||
|
||||
describe('setSessionArchivedBatch', () => {
|
||||
it('maps a metadata read failure to a per-item internal error, not not_found', async () => {
|
||||
const outcomes = await setSessionArchivedBatch(
|
||||
coldPathAccessor({
|
||||
storeGet: async () => {
|
||||
throw new Error('disk on fire');
|
||||
},
|
||||
}),
|
||||
['s1'],
|
||||
true,
|
||||
);
|
||||
expect(outcomes).toEqual([{ id: 's1', ok: false, reason: 'error', message: 'disk on fire' }]);
|
||||
});
|
||||
|
||||
it('maps a missing metadata document to not_found', async () => {
|
||||
const outcomes = await setSessionArchivedBatch(
|
||||
coldPathAccessor({ storeGet: async () => undefined }),
|
||||
['s1'],
|
||||
true,
|
||||
);
|
||||
expect(outcomes).toEqual([
|
||||
{ id: 's1', ok: false, reason: 'not_found', message: 'session s1 does not exist' },
|
||||
]);
|
||||
});
|
||||
|
||||
it('mirrors the persisted metadata, not a stale index summary', async () => {
|
||||
const recorded: SessionSummary[] = [];
|
||||
const outcomes = await setSessionArchivedBatch(
|
||||
coldPathAccessor({
|
||||
indexSummary: { ...summary, title: 'stale', lastPrompt: 'stale-p', updatedAt: 1 },
|
||||
storeGet: async () => ({
|
||||
id: 's1',
|
||||
title: 'fresh',
|
||||
lastPrompt: 'fresh-p',
|
||||
createdAt: 1,
|
||||
updatedAt: 9,
|
||||
archived: false,
|
||||
}),
|
||||
onMirrorRecord: (r) => recorded.push(r),
|
||||
}),
|
||||
['s1'],
|
||||
true,
|
||||
);
|
||||
expect(outcomes).toEqual([{ id: 's1', ok: true }]);
|
||||
expect(recorded).toHaveLength(1);
|
||||
expect(recorded[0]).toMatchObject({
|
||||
workspaceId: 'wd',
|
||||
title: 'fresh',
|
||||
lastPrompt: 'fresh-p',
|
||||
updatedAt: 9,
|
||||
archived: true,
|
||||
});
|
||||
expect(typeof recorded[0]?.archivedAt).toBe('number');
|
||||
});
|
||||
|
||||
it('normalizes legacy v1 metadata before persisting and mirroring', async () => {
|
||||
const recorded: SessionSummary[] = [];
|
||||
const written: unknown[] = [];
|
||||
const legacy = {
|
||||
workDir: '/workspace',
|
||||
customTitle: 'legacy title',
|
||||
createdAt: '2026-07-21T19:40:00.000Z',
|
||||
updatedAt: '2026-07-22T02:00:00.000Z',
|
||||
archived: false,
|
||||
} as unknown as SessionMeta;
|
||||
const outcomes = await setSessionArchivedBatch(
|
||||
coldPathAccessor({
|
||||
storeGet: async () => legacy,
|
||||
onMirrorRecord: (r) => recorded.push(r),
|
||||
onStoreSet: (_scope, _key, value) => written.push(value),
|
||||
}),
|
||||
['s1'],
|
||||
true,
|
||||
);
|
||||
expect(outcomes).toEqual([{ id: 's1', ok: true }]);
|
||||
|
||||
const rec = recorded[0];
|
||||
expect(rec?.title).toBe('legacy title');
|
||||
expect(rec?.updatedAt).toBe(Date.parse('2026-07-22T02:00:00.000Z'));
|
||||
|
||||
const persisted = written[0] as Record<string, unknown>;
|
||||
expect(persisted['version']).toBe(2);
|
||||
expect(typeof persisted['updatedAt']).toBe('number');
|
||||
expect(persisted['customTitle']).toBeUndefined();
|
||||
expect(persisted['isCustomTitle']).toBe(true);
|
||||
});
|
||||
|
||||
it('fails the item when a concurrent resume failed instead of cold-classifying', async () => {
|
||||
const outcomes = await setSessionArchivedBatch(
|
||||
coldPathAccessor({
|
||||
storeGet: async () => {
|
||||
throw new Error('unreachable — the settle throws first');
|
||||
},
|
||||
resumeError: new Error('resume boom'),
|
||||
}),
|
||||
['s1'],
|
||||
true,
|
||||
);
|
||||
expect(outcomes).toEqual([
|
||||
{ id: 's1', ok: false, reason: 'error', message: 'resume boom' },
|
||||
]);
|
||||
});
|
||||
|
||||
it('reads and migrates the legacy session-meta location before answering not_found', async () => {
|
||||
const written: Array<{ scope: string; value: unknown }> = [];
|
||||
const deleted: string[] = [];
|
||||
const meta: SessionMeta = { id: 's1', createdAt: 1, updatedAt: 2, archived: false };
|
||||
const outcomes = await setSessionArchivedBatch(
|
||||
coldPathAccessor({
|
||||
storeGet: async (scope) => (scope.endsWith('/session-meta') ? meta : undefined),
|
||||
onStoreSet: (scope, _key, value) => written.push({ scope, value }),
|
||||
onStoreDelete: (scope) => deleted.push(scope),
|
||||
}),
|
||||
['s1'],
|
||||
true,
|
||||
);
|
||||
expect(outcomes).toEqual([{ id: 's1', ok: true }]);
|
||||
expect(written).toHaveLength(1);
|
||||
expect(written[0]?.scope.endsWith('/session-meta')).toBe(false);
|
||||
expect(deleted).toHaveLength(1);
|
||||
expect(deleted[0]?.endsWith('/session-meta')).toBe(true);
|
||||
});
|
||||
});
|
||||
|
|
@ -1,9 +1,12 @@
|
|||
|
||||
import { createHash } from 'node:crypto';
|
||||
|
||||
import {
|
||||
ISessionIndex,
|
||||
ISessionIndexMirror,
|
||||
IWorkspaceAliases,
|
||||
IWorkspaceService,
|
||||
setSessionArchivedBatch,
|
||||
type Scope,
|
||||
type SessionSummary,
|
||||
} from '@moonshot-ai/agent-core-v2';
|
||||
|
|
@ -24,6 +27,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;
|
||||
}
|
||||
|
||||
export const v2ActivityStatusSchema = z.enum([
|
||||
|
|
@ -56,18 +67,48 @@ function includeDomains(include: string | undefined): string[] {
|
|||
.filter((value) => value.length > 0);
|
||||
}
|
||||
|
||||
const KNOWN_FIELDS = new Set(['id', 'archived']);
|
||||
const IDS_PROJECTION_PAGE_SIZE_MAX = 10000;
|
||||
const FULL_PAGE_SIZE_MAX = 100;
|
||||
|
||||
function parseFields(raw: string | undefined): string[] {
|
||||
return [
|
||||
...new Set(
|
||||
(raw ?? '')
|
||||
.split(',')
|
||||
.map((value) => value.trim())
|
||||
.filter((value) => value.length > 0),
|
||||
),
|
||||
];
|
||||
}
|
||||
|
||||
function isIdsProjection(fields: readonly string[]): boolean {
|
||||
return fields.length === 2 && fields.every((field) => KNOWN_FIELDS.has(field));
|
||||
}
|
||||
|
||||
const v2SessionsListQuerySchema = z
|
||||
.object({
|
||||
'workspace.id': repeatedParam(z.string().min(1)),
|
||||
'activity.status': repeatedParam(v2ActivityStatusSchema),
|
||||
'meta.updated_after': z.coerce.number().int().nonnegative().optional(),
|
||||
'meta.updated_before': z.coerce.number().int().nonnegative().optional(),
|
||||
'meta.archived': z.enum(['true', 'false', 'all']).optional(),
|
||||
sort: v2SortSchema.optional(),
|
||||
include: z.string().optional(),
|
||||
page_size: z.coerce.number().int().min(1).max(100).optional(),
|
||||
fields: z.string().optional(),
|
||||
page_size: z.coerce.number().int().min(1).max(IDS_PROJECTION_PAGE_SIZE_MAX).optional(),
|
||||
page: z.coerce.number().int().min(1).optional(),
|
||||
page_token: z.string().min(1).optional(),
|
||||
})
|
||||
.superRefine((value, ctx) => {
|
||||
if (value.page !== undefined && value.page_token !== undefined) {
|
||||
ctx.addIssue({
|
||||
code: 'custom',
|
||||
message: 'page and page_token are mutually exclusive',
|
||||
path: ['page'],
|
||||
params: { code: ErrorCode.VALIDATION_FAILED },
|
||||
});
|
||||
}
|
||||
for (const domain of includeDomains(value.include)) {
|
||||
if (!KNOWN_INCLUDE_DOMAINS.has(domain)) {
|
||||
ctx.addIssue({
|
||||
|
|
@ -78,6 +119,45 @@ const v2SessionsListQuerySchema = z
|
|||
});
|
||||
}
|
||||
}
|
||||
const fields = parseFields(value.fields);
|
||||
for (const field of fields) {
|
||||
if (!KNOWN_FIELDS.has(field)) {
|
||||
ctx.addIssue({
|
||||
code: 'custom',
|
||||
message: `unknown field '${field}'`,
|
||||
path: ['fields'],
|
||||
params: { code: ErrorCode.VALIDATION_FAILED },
|
||||
});
|
||||
}
|
||||
}
|
||||
const projection = fields.length > 0 && fields.every((field) => KNOWN_FIELDS.has(field));
|
||||
if (projection && !isIdsProjection(fields)) {
|
||||
ctx.addIssue({
|
||||
code: 'custom',
|
||||
message: "unsupported fields projection; the only supported value is 'id,archived'",
|
||||
path: ['fields'],
|
||||
params: { code: ErrorCode.VALIDATION_FAILED },
|
||||
});
|
||||
}
|
||||
if (projection && includeDomains(value.include).includes('git')) {
|
||||
ctx.addIssue({
|
||||
code: 'custom',
|
||||
message: 'include=git is not available with the ids projection',
|
||||
path: ['include'],
|
||||
params: { code: ErrorCode.VALIDATION_FAILED },
|
||||
});
|
||||
}
|
||||
const pageSizeMax = projection ? IDS_PROJECTION_PAGE_SIZE_MAX : FULL_PAGE_SIZE_MAX;
|
||||
if (value.page_size !== undefined && value.page_size > pageSizeMax) {
|
||||
ctx.addIssue({
|
||||
code: 'custom',
|
||||
message: projection
|
||||
? `page_size must be at most ${IDS_PROJECTION_PAGE_SIZE_MAX}`
|
||||
: `page_size must be at most ${FULL_PAGE_SIZE_MAX} without the ids projection`,
|
||||
path: ['page_size'],
|
||||
params: { code: ErrorCode.VALIDATION_FAILED },
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
function asArray<T>(value: T | T[] | undefined): T[] | undefined {
|
||||
|
|
@ -89,10 +169,12 @@ interface NormalizedQuery {
|
|||
readonly workspaceFilter?: readonly string[];
|
||||
readonly statuses?: readonly V2ActivityStatus[];
|
||||
readonly updatedAfter?: number;
|
||||
readonly updatedBefore?: number;
|
||||
readonly archived: 'true' | 'false' | 'all';
|
||||
readonly sort: V2Sort;
|
||||
readonly includeGit: boolean;
|
||||
readonly pageSize: number;
|
||||
readonly projection: boolean;
|
||||
}
|
||||
|
||||
const v2GitDomainSchema = z.object({
|
||||
|
|
@ -121,16 +203,53 @@ const v2SessionSchema = z.object({
|
|||
git: v2GitDomainSchema.optional(),
|
||||
});
|
||||
|
||||
const v2SessionIdProjectionSchema = z.object({
|
||||
id: z.string(),
|
||||
archived: z.boolean(),
|
||||
});
|
||||
|
||||
const v2SessionPageSchema = z.object({
|
||||
items: z.array(v2SessionSchema),
|
||||
items: z.array(z.union([v2SessionSchema, v2SessionIdProjectionSchema])),
|
||||
total: z.number().int(),
|
||||
has_more: z.boolean(),
|
||||
next_page_token: z.string().nullable(),
|
||||
});
|
||||
|
||||
const detailsSchema = z.array(z.object({ path: z.string(), message: z.string() }));
|
||||
|
||||
|
||||
const BATCH_IDS_MAX = 5000;
|
||||
|
||||
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>;
|
||||
type V2SessionIdProjection = z.infer<typeof v2SessionIdProjectionSchema>;
|
||||
|
||||
class PageTokenMismatchError extends Error {}
|
||||
|
||||
|
|
@ -177,10 +296,12 @@ function queryFingerprint(query: NormalizedQuery): string {
|
|||
query.workspaceFilter === undefined ? null : [...query.workspaceFilter].toSorted(),
|
||||
query.statuses === undefined ? null : [...query.statuses].toSorted(),
|
||||
query.updatedAfter ?? null,
|
||||
query.updatedBefore ?? null,
|
||||
query.archived,
|
||||
query.sort,
|
||||
query.includeGit,
|
||||
query.pageSize,
|
||||
query.projection,
|
||||
];
|
||||
return createHash('sha256').update(JSON.stringify(canonical)).digest('base64url').slice(0, 16);
|
||||
}
|
||||
|
|
@ -271,6 +392,34 @@ class GitDomainResolver {
|
|||
}
|
||||
}
|
||||
|
||||
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 outcomes = await setSessionArchivedBatch(core.accessor, ids, archived);
|
||||
const results: V2BatchItemResult[] = outcomes.map((outcome) =>
|
||||
outcome.ok
|
||||
? { id: outcome.id, ok: true }
|
||||
: {
|
||||
id: outcome.id,
|
||||
ok: false,
|
||||
error:
|
||||
outcome.reason === 'not_found'
|
||||
? { code: ErrorCode.SESSION_NOT_FOUND, message: outcome.message }
|
||||
: { code: ErrorCode.INTERNAL_ERROR, message: outcome.message },
|
||||
},
|
||||
);
|
||||
await core.accessor.get(ISessionIndexMirror).drain();
|
||||
const succeeded = results.filter((result) => result.ok).length;
|
||||
reply.send(
|
||||
okEnvelope({ results, succeeded, failed: results.length - succeeded }, requestId),
|
||||
);
|
||||
}
|
||||
export function registerV2SessionsRoutes(app: V2SessionsRouteHost, core: Scope): void {
|
||||
const gitResolver = new GitDomainResolver(core);
|
||||
|
||||
|
|
@ -285,7 +434,7 @@ export function registerV2SessionsRoutes(app: V2SessionsRouteHost, core: Scope):
|
|||
[ErrorCode.PAGE_TOKEN_MISMATCH]: {},
|
||||
},
|
||||
description:
|
||||
'List sessions with domain-grouped metadata (workspace / meta / activity; git via include=git). Opaque-cursor pagination: page_token binds the first page’s query conditions.',
|
||||
"List sessions with domain-grouped metadata (workspace / meta / activity; git via include=git). Paginate with the opaque page_token (binds the first page’s query conditions) or with the stateless 1-based page parameter; every page carries total. fields=id,archived trims each item to the lightweight ids projection (select-all-matching flows; page_size ceiling relaxed to 10000).",
|
||||
tags: ['v2-sessions'],
|
||||
},
|
||||
async (req, reply) => {
|
||||
|
|
@ -295,10 +444,12 @@ export function registerV2SessionsRoutes(app: V2SessionsRouteHost, core: Scope):
|
|||
workspaceFilter: asArray(raw['workspace.id']),
|
||||
statuses: asArray(raw['activity.status']),
|
||||
updatedAfter: raw['meta.updated_after'],
|
||||
updatedBefore: raw['meta.updated_before'],
|
||||
archived: raw['meta.archived'] ?? 'false',
|
||||
sort: raw.sort ?? 'meta.updated_at_desc',
|
||||
includeGit: includeDomains(raw.include).includes('git'),
|
||||
pageSize: raw.page_size ?? DEFAULT_PAGE_SIZE,
|
||||
projection: parseFields(raw.fields).length > 0,
|
||||
};
|
||||
|
||||
const fingerprint = queryFingerprint(query);
|
||||
|
|
@ -344,6 +495,9 @@ export function registerV2SessionsRoutes(app: V2SessionsRouteHost, core: Scope):
|
|||
if (query.updatedAfter !== undefined && summary.updatedAt < query.updatedAfter) {
|
||||
return false;
|
||||
}
|
||||
if (query.updatedBefore !== undefined && summary.updatedAt > query.updatedBefore) {
|
||||
return false;
|
||||
}
|
||||
if (
|
||||
query.statuses !== undefined &&
|
||||
!query.statuses.includes(mapActivityStatus(factsOf(summary.id), summary.lastTurnReason))
|
||||
|
|
@ -357,7 +511,9 @@ export function registerV2SessionsRoutes(app: V2SessionsRouteHost, core: Scope):
|
|||
const sorted = filtered.toSorted(comparator);
|
||||
|
||||
let start = 0;
|
||||
if (cursor !== undefined) {
|
||||
if (raw.page !== undefined) {
|
||||
start = (raw.page - 1) * query.pageSize;
|
||||
} else if (cursor !== undefined) {
|
||||
const [cursorKey, cursorId] = cursor;
|
||||
const cursorItem = {
|
||||
id: cursorId,
|
||||
|
|
@ -372,10 +528,28 @@ export function registerV2SessionsRoutes(app: V2SessionsRouteHost, core: Scope):
|
|||
const hasMore = start + query.pageSize < sorted.length;
|
||||
const lastServed = window.at(-1);
|
||||
const nextPageToken =
|
||||
hasMore && lastServed !== undefined
|
||||
raw.page === undefined && hasMore && lastServed !== undefined
|
||||
? encodePageToken(fingerprint, sortKeyOf(query.sort)(lastServed), lastServed.id)
|
||||
: null;
|
||||
|
||||
if (query.projection) {
|
||||
const projected: V2SessionIdProjection[] = window.map((summary) => ({
|
||||
id: summary.id,
|
||||
archived: summary.archived,
|
||||
}));
|
||||
reply.send(
|
||||
okEnvelope(
|
||||
{
|
||||
items: projected,
|
||||
total: sorted.length,
|
||||
has_more: hasMore,
|
||||
next_page_token: nextPageToken,
|
||||
},
|
||||
req.id,
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
const roots = new Map(
|
||||
(await core.accessor.get(IWorkspaceService).list()).map(
|
||||
(workspace) => [workspace.id, workspace.root] as const,
|
||||
|
|
@ -415,7 +589,12 @@ export function registerV2SessionsRoutes(app: V2SessionsRouteHost, core: Scope):
|
|||
};
|
||||
});
|
||||
|
||||
reply.send(okEnvelope({ items, has_more: hasMore, next_page_token: nextPageToken }, req.id));
|
||||
reply.send(
|
||||
okEnvelope(
|
||||
{ items, total: sorted.length, has_more: hasMore, next_page_token: nextPageToken },
|
||||
req.id,
|
||||
),
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
|
|
@ -424,4 +603,28 @@ 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',
|
||||
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],
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -468,6 +468,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}",
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
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';
|
||||
|
||||
|
|
@ -6,6 +6,12 @@ import {
|
|||
Error2,
|
||||
ErrorCodes,
|
||||
ISessionIndex,
|
||||
IEventService,
|
||||
closeSessionById,
|
||||
getLiveSessionById,
|
||||
resumeSessionById,
|
||||
sessionDirOf,
|
||||
type Event2,
|
||||
type SessionSummary,
|
||||
} from '@moonshot-ai/agent-core-v2';
|
||||
import {
|
||||
|
|
@ -13,7 +19,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';
|
||||
|
|
@ -39,6 +45,7 @@ interface SessionWireV2 {
|
|||
|
||||
interface PageWireV2 {
|
||||
items: SessionWireV2[];
|
||||
total: number;
|
||||
has_more: boolean;
|
||||
next_page_token: string | null;
|
||||
}
|
||||
|
|
@ -256,6 +263,32 @@ describe('server /api/v2/sessions', () => {
|
|||
expect(page.items.map((item) => item.id)).toEqual(['s1', 's2']);
|
||||
});
|
||||
|
||||
it('filters by meta.updated_before (inclusive), combined into a range', async () => {
|
||||
const before = await getData('?meta.updated_before=4000');
|
||||
expect(before.items.map((item) => item.id)).toEqual(['s2', 's3']);
|
||||
expect(before.total).toBe(2);
|
||||
|
||||
const range = await getData('?meta.updated_after=3000&meta.updated_before=4000');
|
||||
expect(range.items.map((item) => item.id)).toEqual(['s2', 's3']);
|
||||
|
||||
const bogus = await getError('?meta.updated_before=-1');
|
||||
expect(bogus.code).toBe(40001);
|
||||
});
|
||||
|
||||
it('binds meta.updated_before into the page_token fingerprint', async () => {
|
||||
const page1 = await getData('?page_size=1&meta.updated_before=4500');
|
||||
expect(page1.items.map((item) => item.id)).toEqual(['s2']);
|
||||
expect(page1.has_more).toBe(true);
|
||||
|
||||
const page2 = await getData(
|
||||
`?page_size=1&meta.updated_before=4500&page_token=${page1.next_page_token}`,
|
||||
);
|
||||
expect(page2.items.map((item) => item.id)).toEqual(['s3']);
|
||||
|
||||
const drifted = await getError(`?page_size=1&page_token=${page1.next_page_token}`);
|
||||
expect(drifted.code).toBe(40922);
|
||||
});
|
||||
|
||||
it('filters by meta.archived (default false / true / all)', async () => {
|
||||
const only = await getData('?meta.archived=true');
|
||||
expect(only.items.map((item) => item.id)).toEqual(['s4']);
|
||||
|
|
@ -285,6 +318,63 @@ describe('server /api/v2/sessions', () => {
|
|||
}
|
||||
});
|
||||
|
||||
it('projects items to {id, archived} with fields=id,archived (relaxed page_size ceiling)', async () => {
|
||||
const page = await getData('?fields=id,archived&page_size=10000');
|
||||
expect(page.total).toBe(3);
|
||||
expect(page.items).toEqual([
|
||||
{ id: 's1', archived: false },
|
||||
{ id: 's2', archived: false },
|
||||
{ id: 's3', archived: false },
|
||||
]);
|
||||
|
||||
const all = await getData('?fields=id,archived&meta.archived=all&sort=meta.updated_at_asc');
|
||||
expect(all.items).toEqual([
|
||||
{ id: 's4', archived: true },
|
||||
{ id: 's3', archived: false },
|
||||
{ id: 's2', archived: false },
|
||||
{ id: 's1', archived: false },
|
||||
]);
|
||||
|
||||
const full = await getError('?page_size=101');
|
||||
expect(full.code).toBe(40001);
|
||||
const tooBig = await getError('?fields=id,archived&page_size=10001');
|
||||
expect(tooBig.code).toBe(40001);
|
||||
});
|
||||
|
||||
it('rejects malformed fields projections (40001)', async () => {
|
||||
expect((await getError('?fields=id,foo')).code).toBe(40001);
|
||||
expect((await getError('?fields=id')).code).toBe(40001);
|
||||
expect((await getError('?fields=archived')).code).toBe(40001);
|
||||
expect((await getError('?fields=id,archived&include=git')).code).toBe(40001);
|
||||
});
|
||||
|
||||
it('paginates the ids projection with an opaque cursor', async () => {
|
||||
const page1 = await getData('?fields=id,archived&page_size=2');
|
||||
expect(page1.items).toEqual([
|
||||
{ id: 's1', archived: false },
|
||||
{ id: 's2', archived: false },
|
||||
]);
|
||||
expect(page1.has_more).toBe(true);
|
||||
|
||||
const page2 = await getData(
|
||||
`?fields=id,archived&page_size=2&page_token=${page1.next_page_token}`,
|
||||
);
|
||||
expect(page2.items).toEqual([{ id: 's3', archived: false }]);
|
||||
expect(page2.has_more).toBe(false);
|
||||
});
|
||||
|
||||
it('binds the projection into the page_token fingerprint', async () => {
|
||||
const full = await getData('?page_size=2');
|
||||
expect(
|
||||
(await getError(`?fields=id,archived&page_size=2&page_token=${full.next_page_token}`)).code,
|
||||
).toBe(40922);
|
||||
|
||||
const projected = await getData('?fields=id,archived&page_size=2');
|
||||
expect((await getError(`?page_size=2&page_token=${projected.next_page_token}`)).code).toBe(
|
||||
40922,
|
||||
);
|
||||
});
|
||||
|
||||
it('paginates with an opaque cursor across pages', async () => {
|
||||
const page1 = await getData('?page_size=2');
|
||||
expect(page1.items.map((item) => item.id)).toEqual(['s1', 's2']);
|
||||
|
|
@ -330,6 +420,53 @@ describe('server /api/v2/sessions', () => {
|
|||
expect(resorted.code).toBe(40922);
|
||||
});
|
||||
|
||||
it('carries total (filtered set size) in every page mode', async () => {
|
||||
const all = await getData();
|
||||
expect(all.total).toBe(3);
|
||||
|
||||
const filtered = await getData(`?workspace.id=${WS_A}`);
|
||||
expect(filtered.total).toBe(2);
|
||||
|
||||
const page1 = await getData('?page_size=2');
|
||||
expect(page1.total).toBe(3);
|
||||
const page2 = await getData(`?page_size=2&page_token=${page1.next_page_token}`);
|
||||
expect(page2.total).toBe(3);
|
||||
});
|
||||
|
||||
it('paginates by 1-based page without minting tokens', async () => {
|
||||
const page1 = await getData('?page=1&page_size=2');
|
||||
expect(page1.items.map((item) => item.id)).toEqual(['s1', 's2']);
|
||||
expect(page1.total).toBe(3);
|
||||
expect(page1.has_more).toBe(true);
|
||||
expect(page1.next_page_token).toBeNull();
|
||||
|
||||
const page2 = await getData('?page=2&page_size=2');
|
||||
expect(page2.items.map((item) => item.id)).toEqual(['s3']);
|
||||
expect(page2.total).toBe(3);
|
||||
expect(page2.has_more).toBe(false);
|
||||
|
||||
const beyond = await getData('?page=7&page_size=2');
|
||||
expect(beyond.items).toEqual([]);
|
||||
expect(beyond.total).toBe(3);
|
||||
expect(beyond.has_more).toBe(false);
|
||||
});
|
||||
|
||||
it('honors filters and sort in page mode', async () => {
|
||||
const page = await getData(`?workspace.id=${WS_A}&sort=meta.updated_at_asc&page=2&page_size=1`);
|
||||
expect(page.items.map((item) => item.id)).toEqual(['s1']);
|
||||
expect(page.total).toBe(2);
|
||||
expect(page.has_more).toBe(false);
|
||||
});
|
||||
|
||||
it('rejects page combined with page_token (40001), and page=0', async () => {
|
||||
const first = await getData('?page_size=2');
|
||||
const both = await getError(`?page=2&page_token=${first.next_page_token}`);
|
||||
expect(both.code).toBe(40001);
|
||||
|
||||
const zero = await getError('?page=0');
|
||||
expect(zero.code).toBe(40001);
|
||||
});
|
||||
|
||||
it('rejects a corrupted page_token (40922)', async () => {
|
||||
const body = await getError('?page_token=!!!not-a-token');
|
||||
expect(body.code).toBe(40922);
|
||||
|
|
@ -388,6 +525,275 @@ 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;
|
||||
}
|
||||
|
||||
function collectEvents(): { events: Event2[]; dispose(): void } {
|
||||
const events: Event2[] = [];
|
||||
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();
|
||||
|
||||
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(getLiveSessionById(core(), created.id)).toBeUndefined();
|
||||
|
||||
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']);
|
||||
|
||||
expect(await indexArchived(created.id)).toBe(true);
|
||||
expect(await listedIds('?meta.archived=true')).toEqual([created.id]);
|
||||
expect(await listedIds()).toEqual([]);
|
||||
|
||||
expect(
|
||||
events
|
||||
.filter((event) => event.type === 'event.session.archived')
|
||||
.map((event) => ({
|
||||
type: event.type,
|
||||
payload: (event as { readonly payload?: unknown }).payload,
|
||||
})),
|
||||
).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();
|
||||
expect(getLiveSessionById(core(), created.id)).toBeDefined();
|
||||
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 }]);
|
||||
|
||||
expect(getLiveSessionById(core(), created.id)).toBeUndefined();
|
||||
expect(
|
||||
events.some(
|
||||
(event) =>
|
||||
event.type === 'event.session.archived' &&
|
||||
((event as { readonly payload?: unknown }).payload as { sessionId: string })
|
||||
.sessionId === created.id,
|
||||
),
|
||||
).toBe(true);
|
||||
expect(await indexArchived(created.id)).toBe(true);
|
||||
dispose();
|
||||
});
|
||||
|
||||
it('settles an in-flight resume before classifying (no cold-write race)', async () => {
|
||||
const created = await createSession();
|
||||
await closeSessionById(core(), created.id);
|
||||
|
||||
const resumePromise = resumeSessionById(core(), created.id);
|
||||
const batchPromise = postBatch('/api/v2/sessions:archive', { ids: [created.id] });
|
||||
|
||||
const handle = await resumePromise;
|
||||
expect(handle).toBeDefined();
|
||||
const body = await batchPromise;
|
||||
expect(body.data?.results).toEqual([{ id: created.id, ok: true }]);
|
||||
|
||||
expect(getLiveSessionById(core(), created.id)).toBeUndefined();
|
||||
expect(await indexArchived(created.id)).toBe(true);
|
||||
expect((await readStateJson(created.workspace_id, created.id))['archived']).toBe(true);
|
||||
});
|
||||
|
||||
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 { 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(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]);
|
||||
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] });
|
||||
expect(await resumeSessionById(core(), created.id)).toBeDefined();
|
||||
|
||||
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(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);
|
||||
|
||||
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 };
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue