diff --git a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts index 69b3aec21..dec2ade2b 100644 --- a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts +++ b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts @@ -35,7 +35,7 @@ * * Every materialization (create/resume/fork-target) first takes the session's * cross-process write lease under `session-leases/` and registers that lease - * as the session's write gate. Close/archive stop producers, flush the + * as the session's write admission. Close/archive stop producers, flush the * session's append-log tail, seal new write admission, await already-admitted * I/O, and then release the lease. Any release failure converges internally to * a dirty-marked abandoned release; callers never need a second teardown @@ -81,7 +81,7 @@ import { IHostEnvironment } from '#/os/interface/hostEnvironment'; import { IHostFileSystem, type HostDirEntry } from '#/os/interface/hostFileSystem'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; -import { IWriteGateRegistry } from '#/persistence/interface/writeGate'; +import { IStorageWriteAdmission } from '#/persistence/interface/storageWriteAdmission'; import { type CrossProcessLockInspection, ICrossProcessLockService, @@ -145,7 +145,6 @@ type SessionReleaseStage = | 'dispose-scope' | 'will-release' | 'flush' - | 'seal' | 'drain-writes' | 'lease-lost' | 'shutdown'; @@ -197,7 +196,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec @ITelemetryService private readonly telemetry: ITelemetryService, @ILogService private readonly log: ILogService, @ICrossProcessLockService private readonly locks: ICrossProcessLockService, - @IWriteGateRegistry private readonly writeGates: IWriteGateRegistry, + @IStorageWriteAdmission private readonly writeAdmission: IStorageWriteAdmission, @ISessionLeaseContactProvider private readonly leaseContact: ISessionLeaseContactProvider, ) { @@ -276,7 +275,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec const lease = await this.acquireSessionLease(opts.sessionId); let registration: IDisposable; try { - registration = this.writeGates.register(sessionScope, lease); + registration = this.writeAdmission.registerSession(sessionScope, lease); } catch (error) { lease.release(); throw error; @@ -505,10 +504,8 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec await this.announceWillRelease({ sessionId, reason: kind }); stage = 'flush'; await this.flushSessionTail(sessionId, entry.scope); - stage = 'seal'; - entry.lease.seal(); stage = 'drain-writes'; - await entry.lease.drained(); + await entry.lease.sealAndDrain(); } catch (error) { await this.abandonSession(entry, stage, error); throw new Error2( @@ -551,8 +548,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec await this.writeDirtyMarker(entry, reason, stage); await this.announceWillRelease({ sessionId, reason: 'dirty-abort' }); await Promise.allSettled(taskServices.map((tasks) => tasks.flushPersistence())); - entry.lease.seal(); - await entry.lease.drained(); + await entry.lease.sealAndDrain(); this.finishSessionRelease(entry); if (reason === 'flush-failed') { this.telemetry.track2('session_dirty_abort', { @@ -607,7 +603,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec try { entry.registration.dispose(); } catch (error) { - this.log.warn('failed to unregister session write gate', { + this.log.warn('failed to unregister session write admission', { sessionId: entry.handle.id, error: String(error), }); diff --git a/packages/agent-core-v2/src/app/workspace/fileWorkspacePersistence.ts b/packages/agent-core-v2/src/app/workspace/fileWorkspacePersistence.ts index d41b6f63e..b174439de 100644 --- a/packages/agent-core-v2/src/app/workspace/fileWorkspacePersistence.ts +++ b/packages/agent-core-v2/src/app/workspace/fileWorkspacePersistence.ts @@ -33,7 +33,7 @@ export class FileWorkspacePersistence implements IWorkspacePersistence { constructor(@IAtomicDocumentStore private readonly docs: IAtomicDocumentStore) {} runExclusive(op: () => Promise): Promise { - return this.docs.runExclusive(WORKSPACE_CATALOG_SCOPE, WORKSPACE_CATALOG_KEY, op); + return this.docs.withExclusiveKeyMutation(WORKSPACE_CATALOG_SCOPE, WORKSPACE_CATALOG_KEY, op); } async load(): Promise { diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index ef8364fa1..73437e86a 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -338,10 +338,11 @@ export * from '#/persistence/interface/appendLogStore'; export * from '#/persistence/interface/atomicDocumentStore'; export * from '#/persistence/interface/queryStore'; export * from '#/persistence/interface/blobStore'; -export * from '#/persistence/interface/writeGate'; +export * from '#/persistence/interface/sessionWriteAdmission'; +export * from '#/persistence/interface/storageWriteAdmission'; export * from '#/persistence/backends/node-fs/fileStorageService'; export * from '#/persistence/backends/node-fs/appendLogStore'; -export * from '#/persistence/backends/node-fs/writeGateRegistryService'; +export * from '#/persistence/backends/node-fs/storageWriteAdmissionService'; export * from '#/persistence/backends/node-fs/atomicDocumentStore'; export * from '#/persistence/backends/node-fs/blobStoreService'; export * from '#/persistence/backends/node-fs/projectLocalConfigService'; diff --git a/packages/agent-core-v2/src/persistence/backends/memory/inMemoryStorageService.ts b/packages/agent-core-v2/src/persistence/backends/memory/inMemoryStorageService.ts index f53acfdb7..9e80581d8 100644 --- a/packages/agent-core-v2/src/persistence/backends/memory/inMemoryStorageService.ts +++ b/packages/agent-core-v2/src/persistence/backends/memory/inMemoryStorageService.ts @@ -133,8 +133,12 @@ export class InMemoryStorageService implements IFileSystemStorageService { }; } - runExclusive(scope: string, key: string, op: () => Promise): Promise { - return enqueueKeyedOperation(this.operationQueues, this.watchKey(scope, key), op); + withExclusiveKeyMutation( + scope: string, + key: string, + mutation: () => Promise, + ): Promise { + return enqueueKeyedOperation(this.operationQueues, this.watchKey(scope, key), mutation); } private notifyWatchers(scope: string, key: string): void { diff --git a/packages/agent-core-v2/src/persistence/backends/node-fs/atomicDocumentStore.ts b/packages/agent-core-v2/src/persistence/backends/node-fs/atomicDocumentStore.ts index 527d66403..e56a4552d 100644 --- a/packages/agent-core-v2/src/persistence/backends/node-fs/atomicDocumentStore.ts +++ b/packages/agent-core-v2/src/persistence/backends/node-fs/atomicDocumentStore.ts @@ -82,18 +82,22 @@ class AtomicDocumentStoreBase implements IAtomicDocumentStore { key: string, mutate: (current: T | undefined) => T | Promise, ): Promise { - return this.runExclusive(scope, key, async () => { + return this.withExclusiveKeyMutation(scope, key, async () => { const next = await mutate(await this.get(scope, key)); await this.set(scope, key, next); return next; }); } - runExclusive(scope: string, key: string, op: () => Promise): Promise { - if (this.storage.runExclusive !== undefined) { - return this.storage.runExclusive(scope, key, op); + withExclusiveKeyMutation( + scope: string, + key: string, + mutation: () => Promise, + ): Promise { + if (this.storage.withExclusiveKeyMutation !== undefined) { + return this.storage.withExclusiveKeyMutation(scope, key, mutation); } - return enqueueKeyedOperation(this.operationQueues, `${scope}\0${key}`, op); + return enqueueKeyedOperation(this.operationQueues, `${scope}\0${key}`, mutation); } async delete(scope: string, key: string): Promise { diff --git a/packages/agent-core-v2/src/persistence/backends/node-fs/fileStorageService.ts b/packages/agent-core-v2/src/persistence/backends/node-fs/fileStorageService.ts index 02564700d..cf9f7fd42 100644 --- a/packages/agent-core-v2/src/persistence/backends/node-fs/fileStorageService.ts +++ b/packages/agent-core-v2/src/persistence/backends/node-fs/fileStorageService.ts @@ -10,8 +10,9 @@ * fsync, so the replacement is both atomic and durable. * - `append` → `open('a')` + write + `fh.sync()` (when `durable`), plus a * one-time directory fsync per scope. - * - `runExclusive` → the shared cross-process lock protocol on - * `.lock`, bounding lock waits to 10 seconds. + * - `withExclusiveKeyMutation` → the shared cross-process lock protocol + * on `.lock`, bounding waits to + * 10 seconds. * - `watch` → chokidar on the parent directory, filtered to the exact key and * debounced, so it survives atomic-replace renames and observes a * file that does not exist yet at subscription time. @@ -21,8 +22,8 @@ * agent-execution-environment abstraction does not expose. Higher-level code * (wire journal, blob store) goes through the Store / Storage interfaces above * this backend, never `node:fs` directly. Session-rooted mutations run through - * the App-scoped write-gate registry, which rejects writes after sealing and - * tracks admitted I/O until it settles. + * the App-scoped write admission, which rejects writes after sealing and + * tracks admitted physical I/O until it settles. */ import { createReadStream, mkdirSync, statSync } from 'node:fs'; @@ -48,7 +49,7 @@ import type { StorageWriteOptions, } from '#/persistence/interface/storage'; import { StorageError, StorageErrors, toStorageIoError } from '#/persistence/interface/storage'; -import { IWriteGateRegistry } from '#/persistence/interface/writeGate'; +import { IStorageWriteAdmission } from '#/persistence/interface/storageWriteAdmission'; const WATCH_DEBOUNCE_MS = 150; const STORAGE_LOCK_WAIT_TIMEOUT_MS = 10_000; @@ -78,8 +79,8 @@ export class FileStorageService implements IFileSystemStorageService { private readonly dirMode?: number, private readonly fileMode?: number, @optional(ICrossProcessLockService) private readonly locks?: ICrossProcessLockService, - @optional(IWriteGateRegistry) - private readonly writeGates?: IWriteGateRegistry, + @optional(IStorageWriteAdmission) + private readonly writeAdmission?: IStorageWriteAdmission, ) {} async read(scope: string, key: string): Promise { @@ -119,7 +120,7 @@ export class FileStorageService implements IFileSystemStorageService { _options: StorageWriteOptions = {}, ): Promise { const filePath = this.path(scope, key); - await this.runWrite(scope, async () => { + await this.withPhysicalWrite(scope, async () => { try { await mkdir(dirname(filePath), { recursive: true, mode: this.dirMode }); await atomicWrite(filePath, data, undefined, this.fileMode); @@ -138,7 +139,7 @@ export class FileStorageService implements IFileSystemStorageService { ): Promise { const filePath = this.path(scope, key); const dir = dirname(filePath); - await this.runWrite(scope, async () => { + await this.withPhysicalWrite(scope, async () => { try { await mkdir(dir, { recursive: true, mode: this.dirMode }); const fh = await open(filePath, 'a', this.fileMode); @@ -172,7 +173,7 @@ export class FileStorageService implements IFileSystemStorageService { async delete(scope: string, key: string): Promise { const filePath = this.path(scope, key); - await this.runWrite(scope, async () => { + await this.withPhysicalWrite(scope, async () => { try { await unlink(filePath); } catch (error) { @@ -254,10 +255,14 @@ export class FileStorageService implements IFileSystemStorageService { }; } - async runExclusive(scope: string, key: string, op: () => Promise): Promise { + async withExclusiveKeyMutation( + scope: string, + key: string, + mutation: () => Promise, + ): Promise { const filePath = this.path(scope, key); const lockPath = `${filePath}.lock`; - await this.runWrite(scope, async () => {}); + this.assertCanWriteNow(scope); if (this.locks === undefined) { throw new StorageError( StorageErrors.codes.STORAGE_IO_FAILED, @@ -270,8 +275,8 @@ export class FileStorageService implements IFileSystemStorageService { lockPath, { wait: { timeoutMs: STORAGE_LOCK_WAIT_TIMEOUT_MS } }, async () => { - await this.runWrite(scope, async () => {}); - return op(); + this.assertCanWriteNow(scope); + return mutation(); }, ); } catch (error) { @@ -302,8 +307,12 @@ export class FileStorageService implements IFileSystemStorageService { return join(this.baseDir, scope); } - private runWrite(scope: string, write: () => Promise): Promise { - return this.writeGates?.run(scope, write) ?? write(); + private assertCanWriteNow(scope: string): void { + this.writeAdmission?.assertCanWriteNow(scope); + } + + private withPhysicalWrite(scope: string, io: () => Promise): Promise { + return this.writeAdmission?.withPhysicalWrite(scope, io) ?? io(); } private async syncDirOnce(dir: string): Promise { diff --git a/packages/agent-core-v2/src/persistence/backends/node-fs/storageWriteAdmissionService.ts b/packages/agent-core-v2/src/persistence/backends/node-fs/storageWriteAdmissionService.ts new file mode 100644 index 000000000..2e0b997f9 --- /dev/null +++ b/packages/agent-core-v2/src/persistence/backends/node-fs/storageWriteAdmissionService.ts @@ -0,0 +1,76 @@ +/** + * `storage` domain (L1) — `IStorageWriteAdmission` implementation. + * + * Routes session-rooted storage scopes to admissions registered by session + * lifecycle. Double registration is a bug, missing session admissions fail + * closed, and root or non-session storage scopes execute directly. Bound at + * App scope. + */ + +import { InstantiationType } from '#/_base/di/extensions'; +import { BugIndicatingError } from '#/_base/errors/errors'; +import { toDisposable, type IDisposable } from '#/_base/di/lifecycle'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { Error2, ErrorCodes } from '#/errors'; +import type { ISessionWriteAdmission } from '#/persistence/interface/sessionWriteAdmission'; +import { IStorageWriteAdmission } from '#/persistence/interface/storageWriteAdmission'; + +export class StorageWriteAdmissionService implements IStorageWriteAdmission { + declare readonly _serviceBrand: undefined; + + private readonly admissions = new Map(); + + registerSession(sessionScope: string, admission: ISessionWriteAdmission): IDisposable { + if (this.admissions.has(sessionScope)) { + throw new BugIndicatingError(`write admission already registered for ${sessionScope}`); + } + this.admissions.set(sessionScope, admission); + return toDisposable(() => { + if (this.admissions.get(sessionScope) === admission) this.admissions.delete(sessionScope); + }); + } + + assertCanWriteNow(scope: string): void { + this.admissionFor(scope)?.assertCanWriteNow(); + } + + async withPhysicalWrite(scope: string, io: () => Promise): Promise { + const admission = this.admissionFor(scope); + return admission === undefined ? io() : admission.withPhysicalWrite(io); + } + + private admissionFor(scope: string): ISessionWriteAdmission | undefined { + const sessionScope = sessionScopeFromStorageScope(scope); + if (sessionScope === undefined) return undefined; + const admission = this.admissions.get(sessionScope); + if (admission === undefined) { + throw new Error2(ErrorCodes.SESSION_LEASE_LOST, 'session has no registered write admission', { + details: { sessionId: sessionScope.slice(sessionScope.lastIndexOf('/') + 1) }, + }); + } + return admission; + } +} + +function sessionScopeFromStorageScope(scope: string): string | undefined { + if (scope === '') return undefined; + const parts = scope.split('/'); + if ( + parts.length < 3 || + parts[0] !== 'sessions' || + parts[1] === '' || + parts[2] === undefined || + parts[2] === '' + ) { + return undefined; + } + return parts.slice(0, 3).join('/'); +} + +registerScopedService( + LifecycleScope.App, + IStorageWriteAdmission, + StorageWriteAdmissionService, + InstantiationType.Eager, + 'storage', +); diff --git a/packages/agent-core-v2/src/persistence/backends/node-fs/writeGateRegistryService.ts b/packages/agent-core-v2/src/persistence/backends/node-fs/writeGateRegistryService.ts deleted file mode 100644 index 53e3eb262..000000000 --- a/packages/agent-core-v2/src/persistence/backends/node-fs/writeGateRegistryService.ts +++ /dev/null @@ -1,69 +0,0 @@ -/** - * `storage` domain (L1) — `IWriteGateRegistry` implementation. - * - * Routes session-rooted storage scopes to the write gate registered by the - * session lifecycle. Double registration is a bug, missing session gates fail - * closed, and root or non-session storage scopes execute without a gate. - * Bound at App scope. - */ - -import { BugIndicatingError } from '#/_base/errors/errors'; -import { InstantiationType } from '#/_base/di/extensions'; -import { toDisposable, type IDisposable } from '#/_base/di/lifecycle'; -import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { Error2, ErrorCodes } from '#/errors'; -import { - type ISessionWriteGate, - IWriteGateRegistry, -} from '#/persistence/interface/writeGate'; - -export class WriteGateRegistryService implements IWriteGateRegistry { - declare readonly _serviceBrand: undefined; - - private readonly gates = new Map(); - - register(sessionScope: string, gate: ISessionWriteGate): IDisposable { - if (this.gates.has(sessionScope)) { - throw new BugIndicatingError(`write gate already registered for ${sessionScope}`); - } - this.gates.set(sessionScope, gate); - return toDisposable(() => { - if (this.gates.get(sessionScope) === gate) this.gates.delete(sessionScope); - }); - } - - async run(scope: string, write: () => Promise): Promise { - const sessionScope = sessionScopeFromStorageScope(scope); - if (sessionScope === undefined) return write(); - const gate = this.gates.get(sessionScope); - if (gate === undefined) { - throw new Error2(ErrorCodes.SESSION_LEASE_LOST, 'session has no registered write gate', { - details: { sessionId: sessionScope.slice(sessionScope.lastIndexOf('/') + 1) }, - }); - } - return gate.run(write); - } -} - -function sessionScopeFromStorageScope(scope: string): string | undefined { - if (scope === '') return undefined; - const parts = scope.split('/'); - if ( - parts.length < 3 || - parts[0] !== 'sessions' || - parts[1] === '' || - parts[2] === undefined || - parts[2] === '' - ) { - return undefined; - } - return parts.slice(0, 3).join('/'); -} - -registerScopedService( - LifecycleScope.App, - IWriteGateRegistry, - WriteGateRegistryService, - InstantiationType.Eager, - 'storage', -); diff --git a/packages/agent-core-v2/src/persistence/interface/atomicDocumentStore.ts b/packages/agent-core-v2/src/persistence/interface/atomicDocumentStore.ts index 2a413f13a..88792a252 100644 --- a/packages/agent-core-v2/src/persistence/interface/atomicDocumentStore.ts +++ b/packages/agent-core-v2/src/persistence/interface/atomicDocumentStore.ts @@ -29,7 +29,11 @@ export interface IAtomicDocumentStore { key: string, mutate: (current: T | undefined) => T | Promise, ): Promise; - runExclusive(scope: string, key: string, op: () => Promise): Promise; + withExclusiveKeyMutation( + scope: string, + key: string, + mutation: () => Promise, + ): Promise; delete(scope: string, key: string): Promise; list(scope: string, prefix?: string): Promise; watch(scope: string, key: string): Event; diff --git a/packages/agent-core-v2/src/persistence/interface/sessionWriteAdmission.ts b/packages/agent-core-v2/src/persistence/interface/sessionWriteAdmission.ts new file mode 100644 index 000000000..77a7ea0f5 --- /dev/null +++ b/packages/agent-core-v2/src/persistence/interface/sessionWriteAdmission.ts @@ -0,0 +1,19 @@ +/** + * `storage` domain (L1) — per-session physical-write admission contract. + * + * Defines the Session-scoped admission capability used to reject new writes + * after sealing, track admitted physical I/O, and await its drain. + */ + +import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; + +export interface ISessionWriteAdmission { + readonly _serviceBrand: undefined; + + assertCanWriteNow(): void; + withPhysicalWrite(io: () => Promise): Promise; + sealAndDrain(): Promise; +} + +export const ISessionWriteAdmission: ServiceIdentifier = + createDecorator('sessionWriteAdmission'); diff --git a/packages/agent-core-v2/src/persistence/interface/storage.ts b/packages/agent-core-v2/src/persistence/interface/storage.ts index 5bbe5a849..c3b7392fc 100644 --- a/packages/agent-core-v2/src/persistence/interface/storage.ts +++ b/packages/agent-core-v2/src/persistence/interface/storage.ts @@ -7,8 +7,9 @@ * * - `write` — atomic whole-value replacement (the `Config` access pattern). * - `append` — ordered, durable byte extension (the `Record` access pattern). - * - `runExclusive` — keyed read-modify-write exclusion when the backend can - * coordinate independent processes. + * - `withExclusiveKeyMutation` — keyed read-modify-write exclusion when the + * backend can coordinate independent + * processes. * * They are not interchangeable: building `append` on top of `write` is O(n) * per append, and building `write` on top of `append` yields awkward "read @@ -130,7 +131,11 @@ export interface IFileSystemStorageService { list(scope: string, prefix?: string): Promise; delete(scope: string, key: string): Promise; watch?(scope: string, key: string): Event; - runExclusive?(scope: string, key: string, op: () => Promise): Promise; + withExclusiveKeyMutation?( + scope: string, + key: string, + mutation: () => Promise, + ): Promise; flush(): Promise; close(): Promise; } diff --git a/packages/agent-core-v2/src/persistence/interface/storageWriteAdmission.ts b/packages/agent-core-v2/src/persistence/interface/storageWriteAdmission.ts new file mode 100644 index 000000000..64e9e29c5 --- /dev/null +++ b/packages/agent-core-v2/src/persistence/interface/storageWriteAdmission.ts @@ -0,0 +1,22 @@ +/** + * `storage` domain (L1) — App-scoped storage write-admission contract. + * + * Routes session-rooted storage scopes to their per-session admission while + * leaving root and non-session scopes unrestricted. Missing session + * registrations fail closed. + */ + +import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; +import type { IDisposable } from '#/_base/di/lifecycle'; +import type { ISessionWriteAdmission } from '#/persistence/interface/sessionWriteAdmission'; + +export interface IStorageWriteAdmission { + readonly _serviceBrand: undefined; + + registerSession(sessionScope: string, admission: ISessionWriteAdmission): IDisposable; + assertCanWriteNow(scope: string): void; + withPhysicalWrite(scope: string, io: () => Promise): Promise; +} + +export const IStorageWriteAdmission: ServiceIdentifier = + createDecorator('storageWriteAdmission'); diff --git a/packages/agent-core-v2/src/persistence/interface/writeGate.ts b/packages/agent-core-v2/src/persistence/interface/writeGate.ts deleted file mode 100644 index 6e871e056..000000000 --- a/packages/agent-core-v2/src/persistence/interface/writeGate.ts +++ /dev/null @@ -1,30 +0,0 @@ -/** - * `persistence/interface` — session write-admission gate contract. - * - * Defines the per-session `ISessionWriteGate` that fences and tracks physical - * writes, plus the App-scoped `IWriteGateRegistry` used by storage backends to - * route a storage scope to its owning session gate. Session scopes without a - * registered gate fail closed; root and non-session scopes are not gated. - */ - -import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; -import type { IDisposable } from '#/_base/di/lifecycle'; - -export interface ISessionWriteGate { - run(write: () => Promise): Promise; - seal(): void; - drained(): Promise; -} - -export const ISessionWriteGate: ServiceIdentifier = - createDecorator('sessionWriteGate'); - -export const IWriteGateRegistry: ServiceIdentifier = - createDecorator('writeGateRegistry'); - -export interface IWriteGateRegistry { - readonly _serviceBrand: undefined; - - register(sessionScope: string, gate: ISessionWriteGate): IDisposable; - run(scope: string, write: () => Promise): Promise; -} diff --git a/packages/agent-core-v2/src/session/sessionLease/sessionLease.ts b/packages/agent-core-v2/src/session/sessionLease/sessionLease.ts index 16b04d8fc..1cc1a9c15 100644 --- a/packages/agent-core-v2/src/session/sessionLease/sessionLease.ts +++ b/packages/agent-core-v2/src/session/sessionLease/sessionLease.ts @@ -3,7 +3,7 @@ * * Defines `ISessionLeaseService`, the Session-scope seeded ownership view, * and the `SessionLease` object that satisfies it together with the - * `ISessionWriteGate` used by storage: an App-owned wrapper + * `ISessionWriteAdmission` used by storage: an App-owned wrapper * (`SessionLifecycleService` builds it; it is deliberately not a DI service) * around the cross-process lock handle at * `/session-leases/.lock`. `assertWritable` is the hard @@ -11,7 +11,8 @@ * sentinel fails closed with * `session.lease_lost`, marks the lease lost, seals write admission, and fires * the loss callback exactly once so the owning session tears itself down. - * The gate tracks admitted writes so lifecycle release can await their drain. + * The admission tracks physical writes so lifecycle release can await their + * drain. * * No default is registered for either Session-scoped view: every production * session scope is seeded by `sessionLifecycle` via {@link sessionLeaseSeed}; @@ -28,7 +29,7 @@ import type { CrossProcessLockInspection, ICrossProcessLockHandle, } from '#/os/interface/crossProcessLock'; -import { ISessionWriteGate } from '#/persistence/interface/writeGate'; +import { ISessionWriteAdmission } from '#/persistence/interface/sessionWriteAdmission'; export const LEASE_CREATING_RETRY_AFTER_MS = 1000; @@ -100,7 +101,7 @@ export interface ISessionLeaseService { export const ISessionLeaseService: ServiceIdentifier = createDecorator('sessionLeaseService'); -export class SessionLease implements ISessionWriteGate, ISessionLeaseService { +export class SessionLease implements ISessionWriteAdmission, ISessionLeaseService { declare readonly _serviceBrand: undefined; readonly lockId: string; @@ -145,12 +146,16 @@ export class SessionLease implements ISessionWriteGate, ISessionLeaseService { } } - async run(write: () => Promise): Promise { - if (this._sealed) throw this.writeGateClosedError(); + assertCanWriteNow(): void { + if (this._sealed) throw this.writeAdmissionClosedError(); this.assertWritable(); + } + + async withPhysicalWrite(io: () => Promise): Promise { + this.assertCanWriteNow(); this.inFlightWrites++; try { - return await write(); + return await io(); } finally { this.inFlightWrites--; if (this.inFlightWrites === 0) { @@ -160,28 +165,33 @@ export class SessionLease implements ISessionWriteGate, ISessionLeaseService { } } - seal(): void { + sealAndDrain(): Promise { + this.sealWrites(); + return this.whenDrained(); + } + + private sealWrites(): void { this._sealed = true; } - drained(): Promise { + private whenDrained(): Promise { if (this.inFlightWrites === 0) return Promise.resolve(); return new Promise((resolve) => { this.drainWaiters.add(resolve); }); } - private writeGateClosedError(): Error2 { + private writeAdmissionClosedError(): Error2 { return new Error2( ErrorCodes.SESSION_LEASE_LOST, - `session ${this.sessionId} write gate is sealed`, + `session ${this.sessionId} write admission is sealed`, { details: { sessionId: this.sessionId } }, ); } private markLost(): void { this._lost = true; - this.seal(); + this.sealWrites(); if (this._lossFired) return; this._lossFired = true; this.onLeaseLost(this.sessionId); @@ -189,7 +199,7 @@ export class SessionLease implements ISessionWriteGate, ISessionLeaseService { release(): void { if (this._released) return; - this.seal(); + this.sealWrites(); this._released = true; this.handle.release(); } @@ -202,6 +212,6 @@ export function sessionLeasePath(homeDir: string, sessionId: string): string { export function sessionLeaseSeed(lease: SessionLease): ScopeSeed { return [ [ISessionLeaseService as ServiceIdentifier, lease], - [ISessionWriteGate as ServiceIdentifier, lease], + [ISessionWriteAdmission as ServiceIdentifier, lease], ]; } diff --git a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts index 9c2c4eb68..f1f3a21b3 100644 --- a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts +++ b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts @@ -14,7 +14,7 @@ * a no-op (no write, no mirror, no event), so resuming a session — which * re-registers its agents as they materialize — never bumps `updatedAt` and * never reorders session listings. Every durable write is fenced by the - * storage backend's per-session write gate, so an instance that lost or + * storage backend's per-session write admission, so an instance that lost or * released the session lease fails closed instead of overwriting a live * peer's state. Bound at Session scope. * diff --git a/packages/agent-core-v2/test/agent/profile/binding.test.ts b/packages/agent-core-v2/test/agent/profile/binding.test.ts index 8e2f4545c..2bf212f1e 100644 --- a/packages/agent-core-v2/test/agent/profile/binding.test.ts +++ b/packages/agent-core-v2/test/agent/profile/binding.test.ts @@ -52,6 +52,16 @@ function createAtomicDocumentStore(): AtomicDocumentStore { set: async (scope: string, key: string, value: T) => { documents.set(documentKey(scope, key), structuredClone(value)); }, + update: async (scope: string, key: string, mutate: (current: T | undefined) => T | Promise) => { + const next = await mutate(documents.get(documentKey(scope, key)) as T | undefined); + documents.set(documentKey(scope, key), structuredClone(next)); + return next; + }, + withExclusiveKeyMutation: async ( + _scope: string, + _key: string, + mutation: () => Promise, + ) => mutation(), delete: async (scope: string, key: string) => { documents.delete(documentKey(scope, key)); }, diff --git a/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts b/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts index 75030dbc9..aff67f27d 100644 --- a/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts +++ b/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts @@ -46,10 +46,11 @@ import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; import { IProjectLocalConfigService } from '#/app/projectLocalConfig/projectLocalConfig'; import { JsonAtomicDocumentStore } from '#/persistence/backends/node-fs/atomicDocumentStore'; -import { ISessionWriteGate, IWriteGateRegistry } from '#/persistence/interface/writeGate'; +import { ISessionWriteAdmission } from '#/persistence/interface/sessionWriteAdmission'; +import { IStorageWriteAdmission } from '#/persistence/interface/storageWriteAdmission'; import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore'; import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService'; -import { WriteGateRegistryService } from '#/persistence/backends/node-fs/writeGateRegistryService'; +import { StorageWriteAdmissionService } from '#/persistence/backends/node-fs/storageWriteAdmissionService'; import { ISessionWorkspaceContext } from '#/session/workspaceContext/workspaceContext'; import { SessionWorkspaceContextService } from '#/session/workspaceContext/workspaceContextService'; import { IWorkspaceService, type Workspace } from '#/app/workspace/workspace'; @@ -300,7 +301,7 @@ function atomicDocumentStoreStub(): IAtomicDocumentStore { get: () => Promise.resolve(undefined), set: () => Promise.resolve(), update: (_scope, _key, mutate) => Promise.resolve(mutate(undefined)), - runExclusive: (_scope, _key, op) => op(), + withExclusiveKeyMutation: (_scope, _key, mutation) => mutation(), delete: () => Promise.resolve(), list: () => Promise.resolve([]), watch: () => (_listener) => ({ dispose: () => {} }), @@ -513,7 +514,7 @@ describe('SessionLifecycleService', () => { stubPair(ILogService, stubLog()), stubPair(IFlagService, stubFlag(false)), stubPair(ICrossProcessLockService, stubCrossProcessLock()), - stubPair(IWriteGateRegistry, new WriteGateRegistryService()), + stubPair(IStorageWriteAdmission, new StorageWriteAdmissionService()), stubPair(ISessionLeaseContactProvider, new SessionLeaseContactProvider()), ...extra, ]); @@ -1370,7 +1371,7 @@ describe('SessionLifecycleService', () => { return [ stubPair(IBootstrapService, tmpBootstrapStub(root)), stubPair(ICrossProcessLockService, new CrossProcessLockService()), - stubPair(IWriteGateRegistry, new WriteGateRegistryService()), + stubPair(IStorageWriteAdmission, new StorageWriteAdmissionService()), ...over, ]; } @@ -1378,11 +1379,11 @@ describe('SessionLifecycleService', () => { function realAlsSeeds(root: string): { seeds: ReturnType[]; appendLog: AppendLogStore; - registry: WriteGateRegistryService; + registry: StorageWriteAdmissionService; storage: FileStorageService; docs: JsonAtomicDocumentStore; } { - const registry = new WriteGateRegistryService(); + const registry = new StorageWriteAdmissionService(); const locks = new CrossProcessLockService(); const storage = new FileStorageService(root, undefined, undefined, locks, registry); const appendLog = new AppendLogStore(storage); @@ -1391,7 +1392,7 @@ describe('SessionLifecycleService', () => { seeds: [ stubPair(IBootstrapService, tmpBootstrapStub(root)), stubPair(ICrossProcessLockService, locks), - stubPair(IWriteGateRegistry, registry), + stubPair(IStorageWriteAdmission, registry), stubPair(IAppendLogStore, appendLog), stubPair(IAtomicDocumentStore, docs), ], @@ -1574,7 +1575,7 @@ describe('SessionLifecycleService', () => { const root = await makeTmpRoot(); const svc = build(realInstanceSeeds(root)); const handle = await svc.create({ sessionId: 's1', workDir: '/tmp/proj' }); - const writeGate = handle.accessor.get(ISessionWriteGate); + const writeAdmission = handle.accessor.get(ISessionWriteAdmission); let finishWrite!: () => void; const writeBlocked = new Promise((resolve) => { finishWrite = resolve; @@ -1583,7 +1584,7 @@ describe('SessionLifecycleService', () => { const writeEntered = new Promise((resolve) => { enterWrite = resolve; }); - const write = writeGate.run(async () => { + const write = writeAdmission.withPhysicalWrite(async () => { enterWrite(); await writeBlocked; }); @@ -1596,7 +1597,7 @@ describe('SessionLifecycleService', () => { let sealed = false; for (let attempt = 0; attempt < 10 && !sealed; attempt++) { try { - await writeGate.run(async () => {}); + writeAdmission.assertCanWriteNow(); } catch (error) { expect(error).toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST }); sealed = true; @@ -1629,7 +1630,7 @@ describe('SessionLifecycleService', () => { }); expect(svc.get('s1')).toBeUndefined(); await expect( - registry.run('sessions/wd_stub/s1/agents/main', async () => {}), + registry.withPhysicalWrite('sessions/wd_stub/s1/agents/main', async () => {}), ).rejects.toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST }); expect(await docs.get<{ custom?: { dirtyAbort?: { reason?: string } } }>( 'sessions/wd_stub/s1', @@ -1674,7 +1675,7 @@ describe('SessionLifecycleService', () => { await svc.closeAll(); await expect( - registry.run('sessions/wd_stub/s1', async () => {}), + registry.withPhysicalWrite('sessions/wd_stub/s1', async () => {}), ).rejects.toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST }); expect(await docs.get<{ custom?: { dirtyAbort?: { reason?: string } } }>( 'sessions/wd_stub/s1', @@ -1692,7 +1693,7 @@ describe('SessionLifecycleService', () => { await Promise.all([svc.close('s1'), svc.closeAll()]); await expect( - registry.run('sessions/wd_stub/s1', async () => {}), + registry.withPhysicalWrite('sessions/wd_stub/s1', async () => {}), ).rejects.toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST }); await expectLeaseFree(root, 's1'); }); diff --git a/packages/agent-core-v2/test/harness/agent.ts b/packages/agent-core-v2/test/harness/agent.ts index 274c70a19..99eadc118 100644 --- a/packages/agent-core-v2/test/harness/agent.ts +++ b/packages/agent-core-v2/test/harness/agent.ts @@ -111,6 +111,7 @@ import { ISessionBtwService, ISessionContext, ISessionLeaseService, + ISessionWriteAdmission, ISessionProcessRunner, IAgentScopeContext, IAgentShellCommandService, @@ -124,6 +125,7 @@ import { IAgentBuiltinToolsRegistrar, IAgentUserToolService, IAgentUsageService, + IStorageWriteAdmission, ISessionWorkspaceContext, AgentLLMRequesterService, LifecycleScope, @@ -532,7 +534,7 @@ export function homeDirServices(homeDir: string | undefined): TestAgentServiceOv reg.defineInstance(id, value); } const file = (): SyncDescriptor => - new SyncDescriptor(FileStorageService, [homeDir], true); + new SyncDescriptor(FileStorageService, [homeDir, undefined, undefined], true); reg.defineDescriptor(IFileSystemStorageService, file()); reg.define(IBlobStore, BlobStoreService); } @@ -956,6 +958,7 @@ export class AgentTestContext { private readonly root: Scope; private readonly session: Scope; private readonly agent: Scope; + private readonly sessionWriteAdmissionRegistration: IDisposable; private readonly disposables: IDisposable[] = []; private suppressWireSnapshot = false; private pluginSessionStartRegistered = false; @@ -1072,6 +1075,10 @@ export class AgentTestContext { .get(ITelemetryService) .withContext({ agent_id: agentId }); const sessionScope = bootstrap.sessionScope(workspaceId, sessionId); + const sessionLease = stubSessionLeaseService(); + this.sessionWriteAdmissionRegistration = this.root.accessor + .get(IStorageWriteAdmission) + .registerSession(sessionScope, sessionLease); this.session = this.root.createChild(LifecycleScope.Session, sessionId, { extra: collectScopeSeed( [ @@ -1086,7 +1093,8 @@ export class AgentTestContext { scope: (subKey?: string): string => subKey === undefined || subKey === '' ? sessionScope : `${sessionScope}/${subKey}`, }); - reg.defineInstance(ISessionLeaseService, stubSessionLeaseService()); + reg.defineInstance(ISessionLeaseService, sessionLease); + reg.defineInstance(ISessionWriteAdmission, sessionLease); reg.defineInstance(ISessionInteractionService, this.createInteractionService()); reg.defineInstance(ISessionApprovalService, this.createApprovalService()); reg.defineInstance(ISessionQuestionService, this.createQuestionService()); @@ -1701,6 +1709,7 @@ export class AgentTestContext { disposable.dispose(); } await this.closeWire(); + this.sessionWriteAdmissionRegistration.dispose(); this.root.dispose(); } diff --git a/packages/agent-core-v2/test/persistence/backends/node-fs/fileStorageService.test.ts b/packages/agent-core-v2/test/persistence/backends/node-fs/fileStorageService.test.ts index 0ab05893d..d53f1e6ec 100644 --- a/packages/agent-core-v2/test/persistence/backends/node-fs/fileStorageService.test.ts +++ b/packages/agent-core-v2/test/persistence/backends/node-fs/fileStorageService.test.ts @@ -5,9 +5,10 @@ import { join } from 'pathe'; import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { Error2, ErrorCodes } from '#/errors'; +import { CrossProcessLockService } from '#/os/backends/node-local/crossProcessLockService'; import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService'; -import { WriteGateRegistryService } from '#/persistence/backends/node-fs/writeGateRegistryService'; -import type { ISessionWriteGate } from '#/persistence/interface/writeGate'; +import { StorageWriteAdmissionService } from '#/persistence/backends/node-fs/storageWriteAdmissionService'; +import type { ISessionWriteAdmission } from '#/persistence/interface/sessionWriteAdmission'; const isWin = process.platform === 'win32'; const encoder = new TextEncoder(); @@ -91,7 +92,7 @@ describe('FileStorageService — error translation', () => { }); }); -describe('FileStorageService — session write fencing', () => { +describe('FileStorageService — session write admission', () => { let dir: string; beforeEach(async () => { @@ -102,21 +103,25 @@ describe('FileStorageService — session write fencing', () => { await rm(dir, { recursive: true, force: true }); }); - it('runs write, append, and delete through the session gate', async () => { - const registry = new WriteGateRegistryService(); + it('runs write, append, and delete through session write admission', async () => { + const registry = new StorageWriteAdmissionService(); let writable = true; - const gate: ISessionWriteGate = { - run: async (write) => { - if (!writable) { - throw new Error2(ErrorCodes.SESSION_LEASE_LOST, 'session lease lost'); - } - return write(); - }, - seal: () => {}, - drained: async () => {}, + const assertCanWriteNow = (): void => { + if (!writable) { + throw new Error2(ErrorCodes.SESSION_LEASE_LOST, 'session lease lost'); + } }; - const registration = registry.register('sessions/workspace/session', gate); - expect(() => registry.register('sessions/workspace/session', gate)).toThrow( + const admission: ISessionWriteAdmission = { + _serviceBrand: undefined, + assertCanWriteNow, + withPhysicalWrite: async (io) => { + assertCanWriteNow(); + return io(); + }, + sealAndDrain: async () => {}, + }; + const registration = registry.registerSession('sessions/workspace/session', admission); + expect(() => registry.registerSession('sessions/workspace/session', admission)).toThrow( /already registered/, ); const svc = new FileStorageService(dir, undefined, undefined, undefined, registry); @@ -143,8 +148,8 @@ describe('FileStorageService — session write fencing', () => { }); }); - it('fails closed without a session gate and leaves non-session scopes untouched', async () => { - const registry = new WriteGateRegistryService(); + it('fails closed without session admission and leaves non-session scopes unrestricted', async () => { + const registry = new StorageWriteAdmissionService(); const svc = new FileStorageService(dir, undefined, undefined, undefined, registry); await expect( @@ -152,4 +157,43 @@ describe('FileStorageService — session write fencing', () => { ).rejects.toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST }); await expect(svc.write('cron/workspace', 'task.json', encoder.encode('{}'))).resolves.toBeUndefined(); }); + + it('revalidates session admission after acquiring the exclusive key mutation', async () => { + const registry = new StorageWriteAdmissionService(); + let checks = 0; + const admission: ISessionWriteAdmission = { + _serviceBrand: undefined, + assertCanWriteNow: () => { + checks++; + if (checks > 1) { + throw new Error2( + ErrorCodes.SESSION_LEASE_LOST, + 'session admission sealed while waiting', + ); + } + }, + withPhysicalWrite: (io) => io(), + sealAndDrain: async () => {}, + }; + registry.registerSession('sessions/workspace/session', admission); + const svc = new FileStorageService( + dir, + undefined, + undefined, + new CrossProcessLockService(), + registry, + ); + let mutationRan = false; + + await expect( + svc.withExclusiveKeyMutation( + 'sessions/workspace/session', + 'state.json', + async () => { + mutationRan = true; + }, + ), + ).rejects.toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST }); + expect(mutationRan).toBe(false); + }); }); 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 da0f93ec7..a82b3c78d 100644 --- a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts +++ b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts @@ -221,8 +221,11 @@ describe('AgentLifecycleService', () => { atomicDocs.set(`${scope}/${key}`, next); return next; }, - runExclusive: async (_scope: string, _key: string, op: () => Promise): Promise => - op(), + withExclusiveKeyMutation: async ( + _scope: string, + _key: string, + mutation: () => Promise, + ): Promise => mutation(), delete: async (scope: string, key: string): Promise => { atomicDocs.delete(`${scope}/${key}`); }, diff --git a/packages/agent-core-v2/test/session/sessionLease/sessionLease.test.ts b/packages/agent-core-v2/test/session/sessionLease/sessionLease.test.ts index 1cf4617db..df3f10695 100644 --- a/packages/agent-core-v2/test/session/sessionLease/sessionLease.test.ts +++ b/packages/agent-core-v2/test/session/sessionLease/sessionLease.test.ts @@ -71,28 +71,27 @@ describe('SessionLease', () => { expect(thrownError(() => lease.assertWritable()).code).toBe(ErrorCodes.SESSION_LEASE_LOST); }); - it('seal rejects new writes while drained waits for an admitted write', async () => { + it('sealAndDrain rejects new writes while waiting for an admitted physical write', async () => { const lease = await acquire(); let enterWrite!: () => void; const writeEntered = new Promise((resolve) => { enterWrite = resolve; }); let finishWrite!: () => void; - const writeGate = new Promise((resolve) => { + const physicalWriteBlocked = new Promise((resolve) => { finishWrite = resolve; }); - const write = lease.run(async () => { + const write = lease.withPhysicalWrite(async () => { enterWrite(); - await writeGate; + await physicalWriteBlocked; }); await writeEntered; - lease.seal(); let drained = false; - const drain = lease.drained().then(() => { + const drain = lease.sealAndDrain().then(() => { drained = true; }); - await expect(lease.run(async () => {})).rejects.toMatchObject({ + await expect(lease.withPhysicalWrite(async () => {})).rejects.toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST, }); expect(drained).toBe(false); diff --git a/packages/agent-core-v2/test/session/sessionLease/stubs.ts b/packages/agent-core-v2/test/session/sessionLease/stubs.ts index c03a39bf6..da90cfe4c 100644 --- a/packages/agent-core-v2/test/session/sessionLease/stubs.ts +++ b/packages/agent-core-v2/test/session/sessionLease/stubs.ts @@ -1,20 +1,28 @@ /** - * `sessionLease` test stubs — no-op `ISessionLeaseService` for unit tests. + * `sessionLease` test stubs — no-op session lease and write admission. * * Lives under `test/` (not `src/`). The default stub reports no lease info - * and passes every `assertWritable` gate; tests that exercise fencing pass an - * `assertWritable` override that throws. Import from a relative path. + * and admits every write; tests that exercise fencing override the relevant + * gate. Import from a relative path. */ +import type { ISessionWriteAdmission } from '#/persistence/interface/sessionWriteAdmission'; import { ISessionLeaseService } from '#/session/sessionLease/sessionLease'; export function stubSessionLeaseService( - overrides: Partial = {}, -): ISessionLeaseService { - return { + overrides: Partial = {}, +): ISessionLeaseService & ISessionWriteAdmission { + const lease: ISessionLeaseService & ISessionWriteAdmission = { _serviceBrand: undefined, info: undefined, assertWritable: () => {}, + assertCanWriteNow: () => lease.assertWritable(), + withPhysicalWrite: async (io) => { + lease.assertCanWriteNow(); + return io(); + }, + sealAndDrain: async () => {}, ...overrides, }; + return lease; }