diff --git a/packages/agent-core-v2/src/app/workspace/fileWorkspacePersistence.ts b/packages/agent-core-v2/src/app/workspace/fileWorkspacePersistence.ts index 920412fe897..dd0e13bc672 100644 --- a/packages/agent-core-v2/src/app/workspace/fileWorkspacePersistence.ts +++ b/packages/agent-core-v2/src/app/workspace/fileWorkspacePersistence.ts @@ -1,6 +1,8 @@ import { LifecycleScope } from '#/app/scopes'; +import { Disposable } from '#/_base/di/lifecycle'; import { ScopeActivation, registerScopedService } from '#/_base/di/scope'; +import { Emitter, type Event } from '#/_base/event'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; import type { Workspace } from './workspace'; @@ -15,10 +17,20 @@ const WORKSPACE_CATALOG_VERSION = 1; const WORKSPACE_CATALOG_SCOPE = ''; const WORKSPACE_CATALOG_KEY = 'workspaces.json'; -export class FileWorkspacePersistence implements IWorkspacePersistence { +export class FileWorkspacePersistence extends Disposable implements IWorkspacePersistence { declare readonly _serviceBrand: undefined; - constructor(@IAtomicDocumentStore private readonly docs: IAtomicDocumentStore) {} + private readonly changeEmitter = this._register(new Emitter()); + readonly onDidChange: Event = this.changeEmitter.event; + + constructor(@IAtomicDocumentStore private readonly docs: IAtomicDocumentStore) { + super(); + this._register( + this.docs.watch(WORKSPACE_CATALOG_SCOPE, WORKSPACE_CATALOG_KEY)(() => { + this.changeEmitter.fire(); + }), + ); + } async load(): Promise { const file = await this.docs.get( @@ -70,6 +82,7 @@ export class FileWorkspacePersistence implements IWorkspacePersistence { deleted_workspace_ids: [...catalog.deletedIds], }; await this.docs.set(WORKSPACE_CATALOG_SCOPE, WORKSPACE_CATALOG_KEY, file); + this.changeEmitter.fire(); } } diff --git a/packages/agent-core-v2/src/app/workspace/workspacePersistence.ts b/packages/agent-core-v2/src/app/workspace/workspacePersistence.ts index 4c96e4d6a3e..f3b941e6231 100644 --- a/packages/agent-core-v2/src/app/workspace/workspacePersistence.ts +++ b/packages/agent-core-v2/src/app/workspace/workspacePersistence.ts @@ -1,4 +1,5 @@ import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; +import type { Event } from '#/_base/event'; import type { Workspace } from './workspace'; @@ -23,6 +24,8 @@ export interface WorkspaceCatalog { export interface IWorkspacePersistence { readonly _serviceBrand: undefined; + readonly onDidChange: Event; + load(): Promise; save(catalog: WorkspaceCatalog): Promise; } diff --git a/packages/agent-core-v2/src/app/workspaceAliases/workspaceAliasesService.ts b/packages/agent-core-v2/src/app/workspaceAliases/workspaceAliasesService.ts index 8fc84bcccd9..d9ceb425fb9 100644 --- a/packages/agent-core-v2/src/app/workspaceAliases/workspaceAliasesService.ts +++ b/packages/agent-core-v2/src/app/workspaceAliases/workspaceAliasesService.ts @@ -1,34 +1,171 @@ import { LifecycleScope } from '#/app/scopes'; +import { Disposable } from '#/_base/di/lifecycle'; import { ScopeActivation, registerScopedService } from '#/_base/di/scope'; -import { IWorkspaceService } from '#/app/workspace/workspace'; +import { encodeWorkDirKey, workspaceRootKey } from '#/_base/utils/workdir-slug'; +import { IWorkspaceService, type Workspace } from '#/app/workspace/workspace'; import { - collectAliasIds, readSessionIndexEntries, + SESSION_INDEX_KEY, + SESSION_INDEX_SCOPE, } from '#/app/workspace/workspaceAlias'; import { IWorkspacePersistence } from '#/app/workspace/workspacePersistence'; +import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { IWorkspaceAliases } from './workspaceAliases'; -export class WorkspaceAliasesService implements IWorkspaceAliases { +interface CatalogSnapshot { + readonly byId: ReadonlyMap; + readonly idsByRootKey: ReadonlyMap; +} + +interface SessionIndexSnapshot { + readonly idsByRootKey: ReadonlyMap; +} + +function rootKeyIndex( + items: readonly T[], + rootOf: (item: T) => string, + idOf: (item: T) => string, +): Map { + const map = new Map(); + for (const item of items) { + const key = workspaceRootKey(rootOf(item)); + const id = idOf(item); + const bucket = map.get(key); + if (bucket === undefined) { + map.set(key, [id]); + } else if (!bucket.includes(id)) { + bucket.push(id); + } + } + return map; +} + +export class WorkspaceAliasesService extends Disposable implements IWorkspaceAliases { declare readonly _serviceBrand: undefined; + private catalogCache: CatalogSnapshot | undefined; + private sessionIndexCache: { snapshot: SessionIndexSnapshot; size: number | undefined } | undefined; + private catalogPromise: + | Promise<{ snapshot: CatalogSnapshot; generation: number }> + | undefined; + private sessionIndexPromise: + | Promise<{ snapshot: SessionIndexSnapshot; generation: number }> + | undefined; + private invalidationGeneration = 0; + private catalogMergePrimed = false; + constructor( @IWorkspaceService private readonly workspaces: IWorkspaceService, @IWorkspacePersistence private readonly store: IWorkspacePersistence, @IFileSystemStorageService private readonly storage: IFileSystemStorageService, - ) {} + @IAppendLogStore private readonly appendLogs: IAppendLogStore, + ) { + super(); + this._register( + this.store.onDidChange(() => { + this.invalidationGeneration += 1; + this.catalogCache = undefined; + }), + ); + this._register( + this.appendLogs.onDidWrite((write) => { + if (write.scope === SESSION_INDEX_SCOPE && write.key === SESSION_INDEX_KEY) { + this.invalidationGeneration += 1; + this.sessionIndexCache = undefined; + } + }), + ); + } async resolveAliasIds(id: string): Promise { - const entry = await this.workspaces.get(id); - if (entry === undefined) return [id]; - const catalog = (await this.store.load()) ?? { workspaces: [], deletedIds: [] }; - return collectAliasIds( - catalog.workspaces, - await readSessionIndexEntries(this.storage), - entry.root, - ); + for (;;) { + const generation = this.invalidationGeneration; + const [catalog, index] = await Promise.all([this.catalog(), this.sessionIndex()]); + if (generation !== this.invalidationGeneration) continue; + const entry = catalog.byId.get(id); + if (entry === undefined) return [id]; + const rootKey = workspaceRootKey(entry.root); + const fromCatalog = catalog.idsByRootKey.get(rootKey); + const fromIndex = index.idsByRootKey.get(rootKey); + if (fromCatalog === undefined) return fromIndex ?? [id]; + if (fromIndex === undefined) return fromCatalog; + const merged = [...fromCatalog]; + for (const alias of fromIndex) { + if (!merged.includes(alias)) merged.push(alias); + } + return merged; + } + } + + private async catalog(): Promise { + if (this.catalogCache !== undefined) return this.catalogCache; + this.catalogPromise ??= this.loadCatalog(); + const { snapshot, generation } = await this.catalogPromise; + if (generation !== this.invalidationGeneration) return this.catalog(); + return snapshot; + } + + private async loadCatalog(): Promise<{ snapshot: CatalogSnapshot; generation: number }> { + try { + if (!this.catalogMergePrimed) { + await this.workspaces.list(); + this.catalogMergePrimed = true; + } + const generation = this.invalidationGeneration; + const workspaces = (await this.store.load())?.workspaces ?? []; + const snapshot: CatalogSnapshot = { + byId: new Map(workspaces.map((ws) => [ws.id, ws] as const)), + idsByRootKey: rootKeyIndex( + workspaces, + (ws) => ws.root, + (ws) => ws.id, + ), + }; + if (generation === this.invalidationGeneration) { + this.catalogCache = snapshot; + } + return { snapshot, generation }; + } finally { + this.catalogPromise = undefined; + } + } + + private async sessionIndex(): Promise { + const cache = this.sessionIndexCache; + if ( + cache !== undefined && + (await this.storage.size(SESSION_INDEX_SCOPE, SESSION_INDEX_KEY)) === cache.size + ) { + return cache.snapshot; + } + this.sessionIndexPromise ??= this.loadSessionIndex(); + const { snapshot, generation } = await this.sessionIndexPromise; + if (generation !== this.invalidationGeneration) return this.sessionIndex(); + return snapshot; + } + + private async loadSessionIndex(): Promise<{ snapshot: SessionIndexSnapshot; generation: number }> { + try { + const generation = this.invalidationGeneration; + const entries = await readSessionIndexEntries(this.storage); + const snapshot: SessionIndexSnapshot = { + idsByRootKey: rootKeyIndex(entries, (entry) => entry.workDir, (entry) => + encodeWorkDirKey(entry.workDir), + ), + }; + if (generation === this.invalidationGeneration) { + this.sessionIndexCache = { + snapshot, + size: await this.storage.size(SESSION_INDEX_SCOPE, SESSION_INDEX_KEY), + }; + } + return { snapshot, generation }; + } finally { + this.sessionIndexPromise = undefined; + } } } diff --git a/packages/agent-core-v2/src/persistence/backends/node-fs/appendLogStore.ts b/packages/agent-core-v2/src/persistence/backends/node-fs/appendLogStore.ts index b4eee62e9af..e38c7aeec58 100644 --- a/packages/agent-core-v2/src/persistence/backends/node-fs/appendLogStore.ts +++ b/packages/agent-core-v2/src/persistence/backends/node-fs/appendLogStore.ts @@ -1,6 +1,7 @@ -import { toDisposable, type IDisposable } from '#/_base/di/lifecycle'; +import { Disposable, toDisposable, type IDisposable } from '#/_base/di/lifecycle'; import { LifecycleScope } from '#/app/scopes'; import { ScopeActivation, registerScopedService } from '#/_base/di/scope'; +import { Emitter, type Event } from '#/_base/event'; import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { @@ -8,6 +9,7 @@ import { IAppendLogStore, type AppendLogOptions, type AppendLogReadOptions, + type AppendLogWrite, } from '#/persistence/interface/appendLogStore'; const textEncoder = new TextEncoder(); @@ -33,12 +35,16 @@ interface LogState { onError?: (error: unknown) => void; } -export class AppendLogStore implements IAppendLogStore { +export class AppendLogStore extends Disposable implements IAppendLogStore { declare readonly _serviceBrand: undefined; private readonly logs = new Map(); + private readonly writeEmitter = this._register(new Emitter()); + readonly onDidWrite: Event = this.writeEmitter.event; - constructor(@IFileSystemStorageService private readonly storage: IFileSystemStorageService) {} + constructor(@IFileSystemStorageService private readonly storage: IFileSystemStorageService) { + super(); + } append(scope: string, key: string, record: R, options?: AppendLogOptions): void { const state = this.state(scope, key); @@ -117,12 +123,13 @@ export class AppendLogStore implements IAppendLogStore { try { await this.storage.write(scope, key, encoded, { atomic: true }); state.storageFailure = undefined; + return true; } catch (error) { state.storageFailure = { error }; throw error; } }); - await this.ownFlush(scope, key, state, rewrite); + await this.ownFlush(scope, key, state, rewrite, { value: false }); } async flush(): Promise { @@ -190,7 +197,8 @@ export class AppendLogStore implements IAppendLogStore { private flushState(scope: string, key: string, state: LogState): Promise { if (state.flushPromise !== undefined) return state.flushPromise; if (state.storageFailure !== undefined) return Promise.reject(state.storageFailure.error); - return this.ownFlush(scope, key, state, this.drain(scope, key, state)); + const wroteBox = { value: false }; + return this.ownFlush(scope, key, state, this.drain(scope, key, state, wroteBox), wroteBox); } private release(scope: string, key: string, state: LogState): void { @@ -216,10 +224,11 @@ export class AppendLogStore implements IAppendLogStore { scope: string, key: string, state: LogState, - operation: Promise, + operation: Promise, + wroteBox: { value: boolean }, ): Promise { let owned!: Promise; - owned = this.finishOwnedFlush(scope, key, state, operation, () => owned); + owned = this.finishOwnedFlush(scope, key, state, operation, wroteBox, () => owned); state.flushPromise = owned; return owned; } @@ -228,47 +237,55 @@ export class AppendLogStore implements IAppendLogStore { scope: string, key: string, state: LogState, - operation: Promise, + operation: Promise, + wroteBox: { value: boolean }, owner: () => Promise, ): Promise { let failure: { readonly error: unknown } | undefined; try { - await operation; - } catch (error) { - failure = { error }; - } - const owned = owner(); - if (state.flushPromise === owned) { - try { - if (failure === undefined) { - while (state.flushPromise === owned && state.pending.length > 0) { - await this.drain(scope, key, state); - } - } - } finally { - if (state.flushPromise === owned) { - state.flushPromise = undefined; + if (await operation) wroteBox.value = true; + const owned = owner(); + if (state.flushPromise === owned) { + while (state.flushPromise === owned && state.pending.length > 0) { + await this.drain(scope, key, state, wroteBox); } } + } catch (error) { + failure ??= { error }; + } finally { + const owned = owner(); + if (state.flushPromise === owned) { + state.flushPromise = undefined; + } } + if (wroteBox.value) this.writeEmitter.fire({ scope, key }); if (failure !== undefined) throw failure.error; } - private async drain(scope: string, key: string, state: LogState): Promise { + private async drain( + scope: string, + key: string, + state: LogState, + wroteBox?: { value: boolean }, + ): Promise { const cutoverEpoch = state.cutoverEpoch; await state.ready; - if (state.cutoverEpoch !== cutoverEpoch) return; + if (state.cutoverEpoch !== cutoverEpoch) return false; + let wrote = false; while (state.pending.length > 0) { const batch = state.pending.slice(); try { await this.storage.append(scope, key, encodeBatch(batch), { durable: true }); + wrote = true; + if (wroteBox !== undefined) wroteBox.value = true; } catch (error) { const failure = (state.storageFailure ??= { error }); throw failure.error; } - if (state.cutoverEpoch !== cutoverEpoch) return; + if (state.cutoverEpoch !== cutoverEpoch) return wrote; state.pending.splice(0, batch.length); } + return wrote; } } diff --git a/packages/agent-core-v2/src/persistence/interface/appendLogStore.ts b/packages/agent-core-v2/src/persistence/interface/appendLogStore.ts index a1d78ccb91d..973c0afd6e4 100644 --- a/packages/agent-core-v2/src/persistence/interface/appendLogStore.ts +++ b/packages/agent-core-v2/src/persistence/interface/appendLogStore.ts @@ -1,5 +1,6 @@ import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; import { type IDisposable } from '#/_base/di/lifecycle'; +import type { Event } from '#/_base/event'; import { StorageError, StorageErrors } from '#/persistence/interface/storage'; @@ -31,9 +32,16 @@ export interface AppendLogReadOptions { readonly onTruncate?: (truncation: AppendLogTruncation) => void; } +export interface AppendLogWrite { + readonly scope: string; + readonly key: string; +} + export interface IAppendLogStore { readonly _serviceBrand: undefined; + readonly onDidWrite: Event; + append(scope: string, key: string, record: R, options?: AppendLogOptions): void; read(scope: string, key: string, options?: AppendLogReadOptions): AsyncIterable; rewrite(scope: string, key: string, records: readonly R[]): Promise; diff --git a/packages/agent-core-v2/test/app/workspaceAliases/workspaceAliasesService.test.ts b/packages/agent-core-v2/test/app/workspaceAliases/workspaceAliasesService.test.ts index edec7f532c1..08233f3930b 100644 --- a/packages/agent-core-v2/test/app/workspaceAliases/workspaceAliasesService.test.ts +++ b/packages/agent-core-v2/test/app/workspaceAliases/workspaceAliasesService.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { promises as fsp } from 'node:fs'; import os from 'node:os'; @@ -13,17 +13,20 @@ import { createScopedTestHost, stubPair } from '#/_base/di/test'; import { encodeWorkDirKey } from '#/_base/utils/workdir-slug'; import { HostFileSystem } from '#/os/backends/node-local/hostFsService'; import { IHostFileSystem } from '#/os/interface/hostFileSystem'; +import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore'; import { JsonAtomicDocumentStore } from '#/persistence/backends/node-fs/atomicDocumentStore'; import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService'; +import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; import { IFileSystemStorageService } from '#/persistence/interface/storage'; import { IEventService } from '#/app/event/event'; -import { IWorkspaceService } from '#/app/workspace/workspace'; +import { IWorkspaceService, type Workspace } from '#/app/workspace/workspace'; import { WorkspaceService } from '#/app/workspace/workspaceService'; import { FileWorkspacePersistence } from '#/app/workspace/fileWorkspacePersistence'; import { IWorkspacePersistence, type PersistedWorkspaceEntry, + type WorkspaceCatalog, } from '#/app/workspace/workspacePersistence'; import { IWorkspaceAliases } from '#/app/workspaceAliases/workspaceAliases'; import { WorkspaceAliasesService } from '#/app/workspaceAliases/workspaceAliasesService'; @@ -70,11 +73,24 @@ describe('WorkspaceAliasesService (file-backed)', () => { await fsp.rm(homeDir, { recursive: true, force: true }); }); - function build(hostFs: IHostFileSystem = new HostFileSystem()): IWorkspaceAliases { - const fileStorage = new FileStorageService(homeDir); + class CountingStorage extends FileStorageService { + reads = 0; + override async read(scope: string, key: string): Promise { + this.reads += 1; + return super.read(scope, key); + } + } + + function build( + hostFs: IHostFileSystem = new HostFileSystem(), + fileStorage: FileStorageService = new FileStorageService(homeDir), + persistence?: IWorkspacePersistence, + ): IWorkspaceAliases { const host = createScopedTestHost([ stubPair(IFileSystemStorageService, fileStorage), stubPair(IAtomicDocumentStore, new JsonAtomicDocumentStore(fileStorage)), + stubPair(IAppendLogStore, new AppendLogStore(fileStorage)), + ...(persistence !== undefined ? [stubPair(IWorkspacePersistence, persistence)] : []), stubPair(IHostFileSystem, hostFs), stubPair(IEventService, { publish: () => {}, @@ -168,4 +184,270 @@ describe('WorkspaceAliasesService (file-backed)', () => { ]); expect(await aliases.resolveAliasIds(id)).toEqual([id]); }); + + it('resolveAliasIds reuses the loaded catalog and session index across calls', async () => { + const root = join(homeDir, 'proj'); + const id = encodeWorkDirKey(root); + await writeWorkspacesJson({ + [id]: { + root, + name: 'proj', + created_at: '2026-01-01T00:00:00.000Z', + last_opened_at: '2026-01-01T00:00:00.000Z', + }, + }); + await seedSessionIndex([{ sessionId: 's1', sessionDir: 'sessions/a/s1', workDir: root }]); + const storage = new CountingStorage(homeDir); + const aliases = build(undefined, storage); + + await aliases.resolveAliasIds(id); + const readsAfterWarm = storage.reads; + await aliases.resolveAliasIds(id); + await aliases.resolveAliasIds('wd_missing_000000000000'); + expect(storage.reads).toBe(readsAfterWarm); + }); + + it('resolveAliasIds coalesces concurrent cold loads', async () => { + const root = join(homeDir, 'proj'); + const id = encodeWorkDirKey(root); + await writeWorkspacesJson({ + [id]: { + root, + name: 'proj', + created_at: '2026-01-01T00:00:00.000Z', + last_opened_at: '2026-01-01T00:00:00.000Z', + }, + }); + await seedSessionIndex([{ sessionId: 's1', sessionDir: 'sessions/a/s1', workDir: root }]); + const storage = new CountingStorage(homeDir); + const aliases = build(undefined, storage); + + await Promise.all([ + aliases.resolveAliasIds(id), + aliases.resolveAliasIds(id), + aliases.resolveAliasIds('wd_missing_000000000000'), + aliases.resolveAliasIds(encodeWorkDirKey(join(homeDir, 'nowhere'))), + ]); + expect(storage.reads).toBeLessThanOrEqual(6); + }); + + it('resolveAliasIds follows in-process catalog writes synchronously', async () => { + const entry = (root: string): PersistedWorkspaceEntry => ({ + root, + name: 'proj', + created_at: '2026-01-01T00:00:00.000Z', + last_opened_at: '2026-01-01T00:00:00.000Z', + }); + const typedRoot = 'C:\\Users\\Foo\\Proj'; + const typedId = encodeWorkDirKey(typedRoot); + const legacyId = 'wd_proj_deadbeef0002'; + await writeWorkspacesJson({ [typedId]: entry(typedRoot) }); + const aliases = build(); + expect(await aliases.resolveAliasIds(typedId)).toEqual([typedId]); + + const persistence = currentHost!.app.accessor.get(IWorkspacePersistence); + await persistence.save({ + workspaces: [ + { id: typedId, root: typedRoot, name: 'proj', createdAt: 0, lastOpenedAt: 0 }, + { id: legacyId, root: 'c:\\users\\foo\\proj', name: 'proj', createdAt: 0, lastOpenedAt: 0 }, + ], + deletedIds: [], + }); + expect((await aliases.resolveAliasIds(typedId)).toSorted()).toEqual( + [legacyId, typedId].toSorted(), + ); + }); + + it('resolveAliasIds retries a shared load that spanned a write', async () => { + class GatedPersistence implements IWorkspacePersistence { + declare readonly _serviceBrand: undefined; + loads = 0; + gate: Promise | undefined; + constructor(private readonly inner: IWorkspacePersistence) {} + get onDidChange(): IWorkspacePersistence['onDidChange'] { + return this.inner.onDidChange; + } + async load(): ReturnType { + this.loads += 1; + const catalog = await this.inner.load(); + if (this.gate !== undefined) await this.gate; + return catalog; + } + save(catalog: WorkspaceCatalog): Promise { + return this.inner.save(catalog); + } + } + const entry = (root: string): PersistedWorkspaceEntry => ({ + root, + name: 'proj', + created_at: '2026-01-01T00:00:00.000Z', + last_opened_at: '2026-01-01T00:00:00.000Z', + }); + const typedRoot = 'C:\\Users\\Foo\\Proj'; + const typedId = encodeWorkDirKey(typedRoot); + const legacyId = 'wd_proj_deadbeef0002'; + await writeWorkspacesJson({ [typedId]: entry(typedRoot) }); + const storage = new FileStorageService(homeDir); + const persistence = new GatedPersistence( + new FileWorkspacePersistence(new JsonAtomicDocumentStore(storage)), + ); + const aliases = build(undefined, storage, persistence); + const ws = (id: string, root: string): Workspace => ({ + id, + root, + name: 'proj', + createdAt: 0, + lastOpenedAt: 0, + }); + + await aliases.resolveAliasIds(typedId); + await persistence.save({ workspaces: [ws(typedId, typedRoot)], deletedIds: [] }); + + let release: (() => void) | undefined; + persistence.gate = new Promise((resolve) => { + release = resolve; + }); + const baseline = persistence.loads; + const p1 = aliases.resolveAliasIds(typedId); + await vi.waitFor(() => { + expect(persistence.loads).toBe(baseline + 1); + }); + await persistence.save({ + workspaces: [ws(typedId, typedRoot), ws(legacyId, 'c:\\users\\foo\\proj')], + deletedIds: [], + }); + const p2 = aliases.resolveAliasIds(legacyId); + release!(); + + const [r1, r2] = await Promise.all([p1, p2]); + expect(r1.toSorted()).toEqual([legacyId, typedId].toSorted()); + expect(r2.toSorted()).toEqual([legacyId, typedId].toSorted()); + }); + + it('resolveAliasIds does not mix snapshots across a mid-resolution write', async () => { + class GatedStorage extends FileStorageService { + indexReads = 0; + gate: Promise | undefined; + override async read(scope: string, key: string): Promise { + if (key === 'session_index.jsonl') { + this.indexReads += 1; + if (this.gate !== undefined) await this.gate; + } + return super.read(scope, key); + } + } + const entry = (root: string): PersistedWorkspaceEntry => ({ + root, + name: 'proj', + created_at: '2026-01-01T00:00:00.000Z', + last_opened_at: '2026-01-01T00:00:00.000Z', + }); + const typedRoot = 'C:\\Users\\Foo\\Proj'; + const typedId = encodeWorkDirKey(typedRoot); + const legacyId = 'wd_proj_deadbeef0002'; + await writeWorkspacesJson({ [typedId]: entry(typedRoot) }); + const storage = new GatedStorage(homeDir); + const aliases = build(undefined, storage); + const persistence = currentHost!.app.accessor.get(IWorkspacePersistence); + const appendLogs = currentHost!.app.accessor.get(IAppendLogStore); + const ws = (id: string, root: string): Workspace => ({ + id, + root, + name: 'proj', + createdAt: 0, + lastOpenedAt: 0, + }); + + await aliases.resolveAliasIds(typedId); + appendLogs.append('', 'session_index.jsonl', { + sessionId: 's9', + sessionDir: 'sessions/s/s9', + workDir: join(homeDir, 'unrelated'), + }); + await appendLogs.flush(); + + let release: (() => void) | undefined; + storage.gate = new Promise((resolve) => { + release = resolve; + }); + const baseline = storage.indexReads; + const p = aliases.resolveAliasIds(typedId); + await vi.waitFor(() => { + expect(storage.indexReads).toBe(baseline + 1); + }); + await persistence.save({ + workspaces: [ws(typedId, typedRoot), ws(legacyId, 'c:\\users\\foo\\proj')], + deletedIds: [], + }); + release!(); + + expect((await p).toSorted()).toEqual([legacyId, typedId].toSorted()); + }); + + it('resolveAliasIds follows in-process session-index writes synchronously', async () => { + const entry = (root: string): PersistedWorkspaceEntry => ({ + root, + name: 'proj', + created_at: '2026-01-01T00:00:00.000Z', + last_opened_at: '2026-01-01T00:00:00.000Z', + }); + const typedRoot = 'C:\\Users\\Foo\\Proj'; + const typedId = encodeWorkDirKey(typedRoot); + await writeWorkspacesJson({ [typedId]: entry(typedRoot) }); + const aliases = build(); + expect(await aliases.resolveAliasIds(typedId)).toEqual([typedId]); + + const appendLogs = currentHost!.app.accessor.get(IAppendLogStore); + appendLogs.append('', 'session_index.jsonl', { + sessionId: 's9', + sessionDir: 'sessions/s/s9', + workDir: 'c:\\Users\\Foo\\Proj', + }); + await appendLogs.flush(); + const indexOnlyId = encodeWorkDirKey('c:\\Users\\Foo\\Proj'); + expect((await aliases.resolveAliasIds(typedId)).toSorted()).toEqual( + [indexOnlyId, typedId].toSorted(), + ); + }); + + it('resolveAliasIds picks up catalog and session index changes', async () => { + const entry = (root: string): PersistedWorkspaceEntry => ({ + root, + name: 'proj', + created_at: '2026-01-01T00:00:00.000Z', + last_opened_at: '2026-01-01T00:00:00.000Z', + }); + const typedRoot = 'C:\\Users\\Foo\\Proj'; + const typedId = encodeWorkDirKey(typedRoot); + const legacyId = 'wd_proj_deadbeef0002'; + await writeWorkspacesJson({ [typedId]: entry(typedRoot) }); + const aliases = build(); + expect(await aliases.resolveAliasIds(typedId)).toEqual([typedId]); + + await writeWorkspacesJson({ + [typedId]: entry(typedRoot), + [legacyId]: entry('c:\\users\\foo\\proj'), + }); + await vi.waitFor( + async () => { + expect((await aliases.resolveAliasIds(typedId)).toSorted()).toEqual( + [legacyId, typedId].toSorted(), + ); + }, + { timeout: 5000 }, + ); + + const indexOnlyId = encodeWorkDirKey('c:\\Users\\Foo\\Proj'); + await seedSessionIndex([ + { sessionId: 's1', sessionDir: 'sessions/a/s1', workDir: 'c:\\Users\\Foo\\Proj' }, + ]); + await vi.waitFor( + async () => { + expect((await aliases.resolveAliasIds(typedId)).toSorted()).toEqual( + [indexOnlyId, legacyId, typedId].toSorted(), + ); + }, + { timeout: 5000 }, + ); + }); }); diff --git a/packages/agent-core-v2/test/harness/agent.ts b/packages/agent-core-v2/test/harness/agent.ts index 27079579646..7f5ce5a65fb 100644 --- a/packages/agent-core-v2/test/harness/agent.ts +++ b/packages/agent-core-v2/test/harness/agent.ts @@ -976,6 +976,7 @@ function reassertServiceOverrides( class PersistenceAppendLogStore implements IAppendLogStore { declare readonly _serviceBrand: undefined; + readonly onDidWrite: IAppendLogStore['onDidWrite'] = Event.None as IAppendLogStore['onDidWrite']; private readonly history: WireRecord[] = []; private readSeeded = false; diff --git a/packages/agent-core-v2/test/persistence/backends/node-fs/appendLogStore.test.ts b/packages/agent-core-v2/test/persistence/backends/node-fs/appendLogStore.test.ts index 6652fde8997..10be8db29fc 100644 --- a/packages/agent-core-v2/test/persistence/backends/node-fs/appendLogStore.test.ts +++ b/packages/agent-core-v2/test/persistence/backends/node-fs/appendLogStore.test.ts @@ -759,4 +759,35 @@ describe('AppendLogStore', () => { { n: 2, s: '日本語' }, ]); }); + + it('does not emit onDidWrite when a flush persists nothing', async () => { + const events: string[] = []; + record.onDidWrite((write) => events.push(`${write.scope}/${write.key}`)); + + record.append(SCOPE, KEY, { n: 1 }); + await record.flush(); + expect(events).toEqual([`${SCOPE}/${KEY}`]); + + await record.flush(); + await record.flush(); + expect(events).toEqual([`${SCOPE}/${KEY}`]); + }); + + it('emits onDidWrite for batches persisted before a later drain failure', async () => { + const events: string[] = []; + record.onDidWrite((write) => events.push(`${write.scope}/${write.key}`)); + const original = storage.append.bind(storage); + let calls = 0; + storage.append = async (scope, key, data, options) => { + calls += 1; + if (calls > 1) throw new Error('disk full'); + const result = await original(scope, key, data, options); + record.append(scope, key, { n: 2 }); + return result; + }; + + record.append(SCOPE, KEY, { n: 1 }); + await expect(record.flush()).rejects.toThrow('disk full'); + expect(events).toEqual([`${SCOPE}/${KEY}`]); + }); }); diff --git a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts index dce5a5c8fd2..034fc1b3e00 100644 --- a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts +++ b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts @@ -152,6 +152,7 @@ function recordingAppendLog(initial: readonly WireRecord[] = []): { const state: { rewritten?: readonly WireRecord[] } = {}; const store: IAppendLogStore = { _serviceBrand: undefined, + onDidWrite: Event.None as IAppendLogStore['onDidWrite'], append: (_scope: string, _key: string, record: R) => { const persisted = record as unknown as WireRecord; records.push(persisted); diff --git a/packages/agent-core-v2/test/session/subagent/forkParity.test.ts b/packages/agent-core-v2/test/session/subagent/forkParity.test.ts index 28e0fd7015c..2cb036a1fe3 100644 --- a/packages/agent-core-v2/test/session/subagent/forkParity.test.ts +++ b/packages/agent-core-v2/test/session/subagent/forkParity.test.ts @@ -42,6 +42,7 @@ import { stubFlag } from '../../app/flag/stubs'; class ScopedAppendLogStore implements IAppendLogStore { declare readonly _serviceBrand: undefined; private readonly logs = new Map(); + readonly onDidWrite: IAppendLogStore['onDidWrite'] = Event.None as IAppendLogStore['onDidWrite']; recordsFor(scope: string, key: string): WireRecord[] { return structuredClone(this.logs.get(`${scope}/${key}`) ?? []); diff --git a/packages/agent-core-v2/test/wire/stubs.ts b/packages/agent-core-v2/test/wire/stubs.ts index b081c3b05f2..c2c69cb1e3d 100644 --- a/packages/agent-core-v2/test/wire/stubs.ts +++ b/packages/agent-core-v2/test/wire/stubs.ts @@ -1,6 +1,7 @@ import { SyncDescriptor } from '#/_base/di/descriptors'; import { toDisposable } from '#/_base/di/lifecycle'; import type { ServiceRegistration, TestInstantiationService } from '#/_base/di/test'; +import { Event } from '#/_base/event'; import { ILogService } from '#/_base/log/log'; import { IAgentBlobService } from '#/agent/blob/agentBlobService'; import { AgentRuntimeSet } from '#/agent/runtime/agentRuntimeSet'; @@ -36,6 +37,7 @@ interface TestAgentWireDependencies { const noopLog: IAppendLogStore = { _serviceBrand: undefined, + onDidWrite: Event.None as IAppendLogStore['onDidWrite'], append: () => {}, read: async function* () {}, rewrite: async () => {}, @@ -234,6 +236,7 @@ export function recordingWireLog( ): IAppendLogStore { return { _serviceBrand: undefined, + onDidWrite: Event.None as IAppendLogStore['onDidWrite'], append: (_scope, _key, record) => { records.push(record as WireRecord); onAppend?.(record as WireRecord);