feat(agent-core-v2): read and write the v1 session index file

- add a legacySessionIndex module defining the v1 session_index.jsonl
  byte format: tolerant line parsing and v1's entry validation (absolute
  sessionDir inside sessionsDir, basename matches sessionId), shared by
  the v2 readers and writer
- append a v1-compatible index line on session create/fork (never on
  resume) through the append-log store, so a v1 CLI sharing the homeDir
  discovers v2 sessions
- locate sessions through the legacy index in FileSessionIndex.get (and
  read-model cold misses) with a directory-scan fallback for stale lines
- reuse the shared parser in the workspace registry's one-shot rebuild
This commit is contained in:
haozhe.yang 2026-07-13 20:46:45 +08:00
parent 098623ed9f
commit f1274fc22a
7 changed files with 359 additions and 37 deletions

View file

@ -0,0 +1,5 @@
---
"@moonshot-ai/agent-core-v2": patch
---
Write v1-compatible session index entries when the v2 engine creates or forks sessions, so both engine generations discover each other's sessions on the same data directory.

View file

@ -0,0 +1,74 @@
/**
* `sessionIndex` domain (L2) v1 legacy session index file format.
*
* Single definition of the byte format of `<homeDir>/session_index.jsonl`,
* shared by the v2 writer (`sessionLifecycle` appends one line per
* create/fork) and the v2 readers (`FileSessionIndex` point lookups,
* `workspaceRegistry` one-shot rebuild). The file is a v1 interop artifact:
* v2's own source of truth is the `<sessionsDir>/<workspaceId>/<sessionId>/`
* directory tree; this file exists so a v1 CLI sharing the same homeDir can
* discover v2-created sessions, and so v2 can locate sessions the v1 way.
*
* Format (identical to v1 `packages/agent-core/src/session/store/session-index.ts`):
* append-only JSONL, one `{sessionId, sessionDir, workDir}` object per line;
* later lines override earlier ones for the same sessionId; `workDir` is
* informational only and never authoritative on read.
*/
import { basename, dirname, isAbsolute, relative, resolve } from 'pathe';
/** Scope of the legacy index file: the homeDir root (join skips empty segments). */
export const LEGACY_SESSION_INDEX_SCOPE = '';
export const LEGACY_SESSION_INDEX_KEY = 'session_index.jsonl';
export interface LegacySessionIndexEntry {
readonly sessionId: string;
readonly sessionDir: string;
readonly workDir: string;
}
/**
* Tolerantly parse one index line. Returns `undefined` for blank-ish garbage,
* non-JSON text, and entries with non-string fields same acceptance as v1's
* `parseIndexLine`, so a corrupt line never breaks a whole read.
*/
export function parseLegacySessionIndexLine(line: string): LegacySessionIndexEntry | undefined {
try {
const parsed = JSON.parse(line) as unknown;
if (typeof parsed !== 'object' || parsed === null) return undefined;
const entry = parsed as Partial<LegacySessionIndexEntry>;
if (
typeof entry.sessionId !== 'string' ||
typeof entry.sessionDir !== 'string' ||
typeof entry.workDir !== 'string'
) {
return undefined;
}
return {
sessionId: entry.sessionId,
sessionDir: entry.sessionDir,
workDir: entry.workDir,
};
} catch {
return undefined;
}
}
/**
* Validate a parsed entry against v1's read-side rules and derive the
* workspaceId from the sessionDir layout. Returns `undefined` for entries v1
* would skip: a non-absolute `sessionDir`, one outside `sessionsDir`, or one
* whose basename does not match `sessionId`. The workspaceId is the name of
* the bucket directory the session lives in (`<sessionsDir>/<workspaceId>/<sessionId>`).
*/
export function validateLegacySessionIndexEntry(
entry: LegacySessionIndexEntry,
sessionsDir: string,
): { sessionDir: string; workspaceId: string } | undefined {
if (!isAbsolute(entry.sessionDir)) return undefined;
const sessionDir = resolve(entry.sessionDir);
const rel = relative(resolve(sessionsDir), sessionDir);
if (rel === '' || rel.startsWith('..') || isAbsolute(rel)) return undefined;
if (basename(sessionDir) !== entry.sessionId) return undefined;
return { sessionDir, workspaceId: basename(dirname(sessionDir)) };
}

View file

@ -7,6 +7,12 @@
* session ids are enumerated via `IFileSystemStorageService.list`, and each session's
* metadata document is read via `IAtomicDocumentStore` to build its summary.
*
* Point lookups (`get`) first consult v1's legacy `<homeDir>/session_index.jsonl`
* (format defined in `./legacySessionIndex`) to locate a session's workspace in
* O(1), falling back to the directory scan when the file has no line for the id
* or the line is stale. Listings never trust the file: the tree stays
* authoritative.
*
* The session metadata document lives at `<sessionDir>/state.json`, a layout
* shared by v1 and v2; the `version` field distinguishes them (`2` = v2,
* epoch-ms timestamps; absent = v1, ISO-string timestamps). The reader also
@ -48,12 +54,20 @@ import {
type SessionListQuery,
type SessionSummary,
} from './sessionIndex';
import {
LEGACY_SESSION_INDEX_KEY,
LEGACY_SESSION_INDEX_SCOPE,
parseLegacySessionIndexLine,
validateLegacySessionIndexEntry,
} from './legacySessionIndex';
const META_SCOPE = 'session-meta';
const META_KEY = 'state.json';
const SESSION_COLLECTION = 'session';
const READ_MODEL_FLAG = 'persistence_minidb_readmodel';
const textDecoder = new TextDecoder();
/** Accept both v2 (epoch ms number) and v1 (ISO string) timestamps. */
function parseTime(value: unknown): number {
if (typeof value === 'number' && Number.isFinite(value)) return value;
@ -194,7 +208,14 @@ export class FileSessionIndex implements ISessionIndex {
private async getFromReadModel(id: string): Promise<SessionSummary | undefined> {
const cached = await this.queryStore.get<SessionSummary>(SESSION_COLLECTION, id);
if (cached !== undefined) return cached;
// Cold miss: locate the session on disk, then read + backfill.
// Cold miss: locate the session on disk, then read + backfill. Prefer the
// legacy index (O(1)) over the per-workspace scan; a stale line falls
// through to the scan.
const located = (await this.readLegacyIndex()).get(id);
if (located !== undefined) {
const summary = await this.getCachedSummary(located, id);
if (summary !== undefined) return summary;
}
for (const workspaceId of await this.listWorkspaceIds()) {
if (!(await this.hasSession(workspaceId, id))) continue;
return this.getCachedSummary(workspaceId, id);
@ -275,6 +296,13 @@ export class FileSessionIndex implements ISessionIndex {
}
private async getLegacy(id: string): Promise<SessionSummary | undefined> {
// Prefer the legacy index (O(1)) over the per-workspace scan; a stale line
// (deleted dir / unreadable metadata) falls through to the scan.
const located = (await this.readLegacyIndex()).get(id);
if (located !== undefined) {
const summary = await this.readSummary(located, id);
if (summary !== undefined) return summary;
}
for (const workspaceId of await this.listWorkspaceIds()) {
if (!(await this.hasSession(workspaceId, id))) continue;
const summary = await this.readSummary(workspaceId, id);
@ -283,6 +311,29 @@ export class FileSessionIndex implements ISessionIndex {
return undefined;
}
/**
* Read v1's global session index (`<homeDir>/session_index.jsonl`) as a
* sessionId workspaceId locator. Entries are validated with v1's rules
* (absolute sessionDir inside `sessionsDir`, basename === sessionId) and
* later lines override earlier ones. Returns an empty map when the file is
* missing or unreadable.
*/
private async readLegacyIndex(): Promise<Map<string, string>> {
const result = new Map<string, string>();
const bytes = await this.storage.read(LEGACY_SESSION_INDEX_SCOPE, LEGACY_SESSION_INDEX_KEY);
if (bytes === undefined) return result;
for (const line of textDecoder.decode(bytes).split(/\r?\n/)) {
const trimmed = line.trim();
if (trimmed === '') continue;
const entry = parseLegacySessionIndexLine(trimmed);
if (entry === undefined) continue;
const located = validateLegacySessionIndexEntry(entry, this.bootstrap.sessionsDir);
if (located === undefined) continue;
result.set(entry.sessionId, located.workspaceId);
}
return result;
}
private async countActiveLegacy(workspaceId: string): Promise<number> {
let count = 0;
for (const sessionId of await this.listSessionIds(workspaceId)) {

View file

@ -47,6 +47,10 @@ import {
ISessionIndex,
PARENT_SESSION_ID_KEY,
} from '#/app/sessionIndex/sessionIndex';
import {
LEGACY_SESSION_INDEX_KEY,
LEGACY_SESSION_INDEX_SCOPE,
} from '#/app/sessionIndex/legacySessionIndex';
import { IWorkspaceLocalConfigService } from '#/app/workspaceLocalConfig/workspaceLocalConfig';
import { IWorkspaceRegistry } from '#/app/workspaceRegistry/workspaceRegistry';
import { ITelemetryService } from '#/app/telemetry/telemetry';
@ -82,6 +86,12 @@ import {
type MaterializeSessionOptions = Omit<CreateSessionOptions, 'sessionId'> & {
readonly sessionId: string;
readonly workspaceId?: string;
/**
* Append a v1-compatible `session_index.jsonl` line once the session is
* materialized. Set for brand-new sessions (create / fork target); resume
* must not re-append (v1 only writes on create/fork).
*/
readonly legacyIndex?: boolean;
};
export class SessionLifecycleService extends Disposable implements ISessionLifecycleService {
@ -126,7 +136,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
async create(opts: CreateSessionOptions): Promise<ISessionScopeHandle> {
const sessionId = opts.sessionId ?? createSessionId();
const handle = await this.materializeSession({ ...opts, sessionId });
const handle = await this.materializeSession({ ...opts, sessionId, legacyIndex: true });
await this.appendSessionIndexEntry(sessionId, opts.workDir);
if (this.config.get<boolean>(DEFAULT_PLAN_MODE_SECTION) === true) {
const main = await ensureMainAgent(handle);
@ -193,6 +203,21 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
void handle.accessor.get(ISessionSkillCatalog).ready;
await handle.accessor.get(IAgentLifecycleService).ensureMcpReady();
handle.accessor.get(ISessionExternalHooksService);
if (opts.legacyIndex === true) {
// Project the new session into v1's global index so a v1 CLI sharing
// this homeDir can discover it. `AppendLogStore` framing is
// byte-identical to v1's `JSON.stringify(entry) + '\n'`; the explicit
// flush makes the line durable before create/fork returns. Placed at
// the materialization point (not fork completion) because v2 does not
// roll back a failed fork's directory — the tree is the index, so the
// file line matches on-disk reality.
this.appendLogStore.append(LEGACY_SESSION_INDEX_SCOPE, LEGACY_SESSION_INDEX_KEY, {
sessionId: opts.sessionId,
sessionDir,
workDir: opts.workDir,
});
await this.appendLogStore.flush();
}
return handle;
}
@ -389,6 +414,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
const target = await this.materializeSession({
sessionId: targetId,
workDir: workspace.root,
legacyIndex: true,
});
const targetCtx = target.accessor.get(ISessionContext);
const targetMeta = target.accessor.get(ISessionMetadata);

View file

@ -22,6 +22,11 @@ import { basename, isAbsolute } from 'pathe';
import { InstantiationType } from '#/_base/di/extensions';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
import { encodeWorkDirKey } from '#/_base/utils/workdir-slug';
import {
LEGACY_SESSION_INDEX_KEY,
LEGACY_SESSION_INDEX_SCOPE,
parseLegacySessionIndexLine,
} from '#/app/sessionIndex/legacySessionIndex';
import { ErrorCodes, Error2, unwrapErrorCause } from '#/errors';
import { IHostFileSystem } from '#/os/interface/hostFileSystem';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -29,19 +34,8 @@ import { IFileSystemStorageService } from '#/persistence/interface/storage';
import { IWorkspaceRegistry, type Workspace, type WorkspaceUpdate } from './workspaceRegistry';
import { IWorkspacePersistence } from './workspacePersistence';
// Legacy v1 session index, read only for the one-shot rebuild. Empty scope
// resolves to `<homeDir>/<key>` (join skips empty segments).
const SESSION_INDEX_SCOPE = '';
const SESSION_INDEX_KEY = 'session_index.jsonl';
const textDecoder = new TextDecoder();
interface SessionIndexLine {
readonly sessionId: string;
readonly sessionDir: string;
readonly workDir: string;
}
export class WorkspaceRegistryService implements IWorkspaceRegistry {
declare readonly _serviceBrand: undefined;
@ -146,13 +140,13 @@ export class WorkspaceRegistryService implements IWorkspaceRegistry {
private async rebuildFromSessionIndex(): Promise<Map<string, Workspace>> {
const result = new Map<string, Workspace>();
const bytes = await this.storage.read(SESSION_INDEX_SCOPE, SESSION_INDEX_KEY);
const bytes = await this.storage.read(LEGACY_SESSION_INDEX_SCOPE, LEGACY_SESSION_INDEX_KEY);
if (bytes === undefined) return result;
const now = Date.now();
for (const line of textDecoder.decode(bytes).split(/\r?\n/)) {
const trimmed = line.trim();
if (trimmed === '') continue;
const entry = parseSessionIndexLine(trimmed);
const entry = parseLegacySessionIndexLine(trimmed);
if (entry === undefined) continue;
if (!isAbsolute(entry.workDir)) continue;
const id = encodeWorkDirKey(entry.workDir);
@ -178,28 +172,6 @@ export class WorkspaceRegistryService implements IWorkspaceRegistry {
}
}
function parseSessionIndexLine(line: string): SessionIndexLine | undefined {
try {
const parsed = JSON.parse(line) as unknown;
if (typeof parsed !== 'object' || parsed === null) return undefined;
const entry = parsed as Partial<SessionIndexLine>;
if (
typeof entry.sessionId !== 'string' ||
typeof entry.sessionDir !== 'string' ||
typeof entry.workDir !== 'string'
) {
return undefined;
}
return {
sessionId: entry.sessionId,
sessionDir: entry.sessionDir,
workDir: entry.workDir,
};
} catch {
return undefined;
}
}
/**
* Collapse registered workspaces that share a `root`. The persisted catalog
* (v1-compatible `workspaces.json`) can hold legacy entries whose id was

View file

@ -90,6 +90,13 @@ describe('FileSessionIndex (legacy)', () => {
await fsp.mkdir(join(sessionsDir, wsId, sessionId), { recursive: true });
}
async function seedIndexLine(
entry: { sessionId: string; sessionDir: string; workDir: string } | string,
): Promise<void> {
const line = typeof entry === 'string' ? entry : JSON.stringify(entry);
await fsp.appendFile(join(homeDir, 'session_index.jsonl'), `${line}\n`);
}
it('list returns non-archived sessions by default', async () => {
await seedSession('active', { createdAt: 1, updatedAt: 2 });
await seedSession('archived', { archived: true });
@ -189,6 +196,83 @@ describe('FileSessionIndex (legacy)', () => {
expect(await store.countActive(workspaceId)).toBe(2);
expect(await store.countActive('wd_unknown')).toBe(0);
});
it('get locates a session through the legacy index file', async () => {
await seedSession('indexed', { title: 'via index' });
await seedIndexLine({
sessionId: 'indexed',
sessionDir: join(sessionsDir, workspaceId, 'indexed'),
workDir: WORK_DIR,
});
const store = build();
const summary = await store.get('indexed');
expect(summary?.id).toBe('indexed');
expect(summary?.workspaceId).toBe(workspaceId);
expect(summary?.title).toBe('via index');
});
it('get ignores index lines that fail v1 validation', async () => {
await seedSession('valid', { title: 'ok' });
// Non-absolute sessionDir.
await seedIndexLine({ sessionId: 'rel', sessionDir: 'relative/path/rel', workDir: WORK_DIR });
// Outside sessionsDir.
await seedIndexLine({
sessionId: 'outside',
sessionDir: join(homeDir, 'elsewhere', 'outside'),
workDir: WORK_DIR,
});
// Basename does not match sessionId.
await seedIndexLine({
sessionId: 'mismatch',
sessionDir: join(sessionsDir, workspaceId, 'other-name'),
workDir: WORK_DIR,
});
// Corrupt line.
await seedIndexLine('{ not json');
const store = build();
expect(await store.get('rel')).toBeUndefined();
expect(await store.get('outside')).toBeUndefined();
expect(await store.get('mismatch')).toBeUndefined();
// A corrupt file never breaks unrelated lookups.
expect((await store.get('valid'))?.title).toBe('ok');
});
it('later index lines override earlier ones for the same session', async () => {
const otherWs = 'wd_other';
await seedSession('moved', { title: 'old location' });
await seedSession('moved', { title: 'new location' }, otherWs);
await seedIndexLine({
sessionId: 'moved',
sessionDir: join(sessionsDir, workspaceId, 'moved'),
workDir: WORK_DIR,
});
await seedIndexLine({
sessionId: 'moved',
sessionDir: join(sessionsDir, otherWs, 'moved'),
workDir: '/other',
});
const store = build();
const summary = await store.get('moved');
expect(summary?.workspaceId).toBe(otherWs);
expect(summary?.title).toBe('new location');
});
it('get falls back to the directory scan when the index line is stale', async () => {
await seedSession('real', { title: 'on disk' });
// Valid-looking line, but the directory was never created.
await seedIndexLine({
sessionId: 'ghost',
sessionDir: join(sessionsDir, workspaceId, 'ghost'),
workDir: WORK_DIR,
});
const store = build();
expect(await store.get('ghost')).toBeUndefined();
expect((await store.get('real'))?.title).toBe('on disk');
});
});
describe('FileSessionIndex (read model)', () => {
@ -363,4 +447,22 @@ describe('FileSessionIndex (read model)', () => {
await lockHolder.close();
}
});
it('cold get backfills through the legacy index locator', async () => {
await seedSession('indexed', { title: 'cold', createdAt: 1, updatedAt: 2 });
await fsp.appendFile(
join(homeDir, 'session_index.jsonl'),
`${JSON.stringify({
sessionId: 'indexed',
sessionDir: join(sessionsDir, workspaceId, 'indexed'),
workDir: WORK_DIR,
})}\n`,
);
const store = build();
const got = await store.get('indexed');
expect(got?.title).toBe('cold');
// The cold read backfilled the read model.
expect(await queryStore.get(SESSION_COLLECTION, 'indexed')).toMatchObject({ title: 'cold' });
});
});

View file

@ -34,6 +34,7 @@ import { ISessionActivity } from '#/session/sessionActivity/sessionActivity';
import { ISessionMetadata } from '#/session/sessionMetadata/sessionMetadata';
import { ISessionSkillCatalog } from '#/session/sessionSkillCatalog/skillCatalog';
import { ISessionIndex, type SessionSummary } from '#/app/sessionIndex/sessionIndex';
import { parseLegacySessionIndexLine } from '#/app/sessionIndex/legacySessionIndex';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
import { IWorkspaceLocalConfigService } from '#/app/workspaceLocalConfig/workspaceLocalConfig';
@ -219,6 +220,20 @@ function appendLogStoreStub(): IAppendLogStore {
};
}
function recordingAppendLogStore(): {
store: IAppendLogStore;
appends: { scope: string; key: string; record: unknown }[];
} {
const appends: { scope: string; key: string; record: unknown }[] = [];
const store: IAppendLogStore = {
...appendLogStoreStub(),
append: (scope, key, record) => {
appends.push({ scope, key, record });
},
};
return { store, appends };
}
function atomicDocumentStoreStub(): IAtomicDocumentStore {
return {
_serviceBrand: undefined,
@ -974,4 +989,81 @@ describe('SessionLifecycleService', () => {
expect(enter).not.toHaveBeenCalled();
});
});
describe('legacy session index projection', () => {
it('create appends one v1-compatible line to session_index.jsonl', async () => {
const { store, appends } = recordingAppendLogStore();
const svc = build([stubPair(IAppendLogStore, store)]);
await svc.create({ sessionId: 's1', workDir: '/tmp/proj' });
expect(appends).toEqual([
{
scope: '',
key: 'session_index.jsonl',
record: {
sessionId: 's1',
sessionDir: '/tmp/sessions/wd_stub/s1',
workDir: '/tmp/proj',
},
},
]);
// The appended record round-trips through v1's line format.
const record = appends[0]!.record;
expect(parseLegacySessionIndexLine(JSON.stringify(record))).toEqual(record);
});
it('resume does not append to the legacy index', async () => {
const { store, appends } = recordingAppendLogStore();
const svc = build([
stubPair(IAppendLogStore, store),
stubPair(ISessionIndex, sessionIndexWithSummary('s1', '/tmp/proj')),
stubPair(IAgentLifecycleService, agentLifecycleWithMainStub()),
]);
await svc.resume('s1');
expect(appends).toEqual([]);
});
it('fork appends a line for the target session', async () => {
const { store, appends } = recordingAppendLogStore();
const svc = build([
stubPair(IAppendLogStore, store),
stubPair(ISessionActivity, {
_serviceBrand: undefined,
status: () => 'idle' as const,
isIdle: () => true,
}),
stubPair(IWorkspaceRegistry, {
...workspaceRegistryStub(),
get: () =>
Promise.resolve({
id: 'wd_stub',
root: '/tmp/proj',
name: 'stub',
createdAt: 0,
lastOpenedAt: 0,
}),
}),
]);
await svc.create({ sessionId: 'src', workDir: '/tmp/proj' });
await svc.fork({ sourceSessionId: 'src', newSessionId: 'dst' });
expect(appends.map((a) => (a.record as { sessionId: string }).sessionId)).toEqual([
'src',
'dst',
]);
expect(appends[1]).toMatchObject({
scope: '',
key: 'session_index.jsonl',
record: {
sessionId: 'dst',
sessionDir: '/tmp/sessions/wd_stub/dst',
workDir: '/tmp/proj',
},
});
});
});
});