vrr/kimi-code
2
0
Fork 0
mirror of https://github.com/MoonshotAI/kimi-code.git synced 2026-08-26 17:14:33 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refactor(agent-core-v2): converge session release

Browse source
This commit is contained in:
7Sageer 2026-07-22 14:10:43 +08:00
parent 95d92dc37c
commit 735bbdc21c
43 changed files with 580 additions and 753 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.changeset/session-lease-write-fencing.md
View file

@ -2,4 +2,4 @@
"@moonshot-ai/kimi-code": patch
---
Opening the same session from a second instance now fails with a clear ownership error instead of silently interleaving writes.
Opening the same session from a second instance now fails with a clear ownership error, while shutdown blocks late writes and releases ambiguous closes through a dirty fallback.

4
packages/agent-core-v2/src/agent/task/persist.ts
View file

@ -22,7 +22,6 @@ import { join } from 'pathe';
import type { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
import type { IFileSystemStorageService } from '#/persistence/interface/storage';
import { assertScopeWritable, IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import type { AgentTaskInfo, AgentTaskStatus } from './types';
@ -75,7 +74,6 @@ export class AgentTaskPersistence {
private readonly docs: IAtomicDocumentStore,
private readonly bytes: IFileSystemStorageService,
private readonly fallbackRoot: AgentTaskPersistenceRoot | undefined,
private readonly authorityRegistry: IWriteAuthorityRegistry,
) {}
private primaryRoot(): AgentTaskPersistenceRoot {
@ -105,7 +103,6 @@ export class AgentTaskPersistence {
async writeTask(task: PersistedTask): Promise<void> {
validateTaskId(task.taskId);
assertScopeWritable(this.agentScope, this.authorityRegistry);
await this.docs.set(this.tasksScope(), `${task.taskId}${JSON_SUFFIX}`, task);
}
@ -126,7 +123,6 @@ export class AgentTaskPersistence {
async appendTaskOutput(taskId: string, chunk: string): Promise<void> {
if (chunk.length === 0) return;
validateTaskId(taskId);
assertScopeWritable(this.agentScope, this.authorityRegistry);
await this.bytes.append(this.taskOutputScope(taskId), OUTPUT_LOG_KEY, textEncoder.encode(chunk));
}

3
packages/agent-core-v2/src/agent/task/taskService.ts
View file

@ -63,7 +63,6 @@ import { IConfigService } from '#/app/config/config';
import { ISessionContext } from '#/session/sessionContext/sessionContext';
import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { ITelemetryService } from '#/app/telemetry/telemetry';
import { defineModel } from '#/wire/model';
import { IWireService } from '#/wire/wire';
@ -232,7 +231,6 @@ export class AgentTaskService extends Disposable implements IAgentTaskService {
@IConfigService private readonly config: IConfigService,
@IAtomicDocumentStore atomicDocs: IAtomicDocumentStore,
@IFileSystemStorageService byteStore: IFileSystemStorageService,
@IWriteAuthorityRegistry authorityRegistry: IWriteAuthorityRegistry,
@ISessionContext session: ISessionContext,
@IAgentScopeContext scopeContext: IAgentScopeContext,
@ITaskService private readonly taskService: ITaskService,
@ -252,7 +250,6 @@ export class AgentTaskService extends Disposable implements IAgentTaskService {
atomicDocs,
byteStore,
fallbackRoot,
authorityRegistry,
);
this._register(
this.wire.hooks.onDidRestore.register('task', async (_ctx, next) => {

9
packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycle.ts
View file

@ -5,10 +5,10 @@
* `ForkSessionOptions`, `CreateChildSessionOptions`, and the
* `ISessionLifecycleService` used to create sessions (`create`), look up the
* live ones (`get` / `list`), close them (`close`), archive/restore them,
* fork them (`fork`), and fork-then-tag them as direct children (`createChild`). Announces
* an ambiguous durability failure through explicit `forceAbort`, which records a
* dirty-abort marker before releasing the session lease.
* lifecycle transitions through ordered hook slots plus
* fork them (`fork`), and fork-then-tag them as direct children (`createChild`).
* Close and archive always release their lease; a release failure is
* dirty-marked and abandoned internally. Lifecycle transitions run through
* ordered hook slots plus
* `onDidCreateSession` / `onDidCloseSession` / `onDidArchiveSession` /
* `onDidForkSession`. App-scoped — a single
* process-wide instance owns the live session scope tree. Persisted
@ -103,7 +103,6 @@ export interface ISessionLifecycleService {
resume(sessionId: string): Promise<ISessionScopeHandle | undefined>;
close(sessionId: string): Promise<void>;
closeAll(): Promise<void>;
forceAbort(sessionId: string): Promise<void>;
archive(sessionId: string): Promise<void>;
restore(sessionId: string): Promise<ISessionScopeHandle | undefined>;
fork(opts: ForkSessionOptions): Promise<ISessionScopeHandle>;

415
packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts
View file

@ -34,15 +34,12 @@
* cron) are force-instantiated at the same point.
*
* Every materialization (create/resume/fork-target) first takes the session's
* cross-process write lease under `session-leases/` and registers its
* `ISessionWriteAuthority` with the `writeAuthorityRegistry`, so the
* journal/state fencing gates have exactly one authority per live session.
* A preparing scope is private until metadata, MCP, caller-specific setup and
* lifecycle hooks finish. Close/archive move the entry through a draining
* phase, dispose producers, durably flush only that session's append-log tail,
* and release authority only after the barrier succeeds. A failed barrier
* keeps the lease registered for a safe retry; lease loss takes the explicit
* dirty-abort path without claiming a clean handoff.
* cross-process write lease under `session-leases/` and registers that lease
* as the session's write gate. 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
* operation. Lease loss follows the same fail-closed release path.
*/
import { randomUUID } from 'node:crypto';
@ -84,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 { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { IWriteGateRegistry } from '#/persistence/interface/writeGate';
import {
type CrossProcessLockInspection,
ICrossProcessLockService,
@ -138,20 +135,29 @@ type MaterializeSessionOptions = Omit<CreateSessionOptions, 'sessionId'> & {
readonly workspaceId?: string;
};
type SessionEntryPhase = 'preparing' | 'active' | 'draining' | 'flush-failed';
type SessionCloseKind = 'close' | 'archive';
type SessionEntryState = 'opening' | 'active' | 'closing';
type SessionReleaseKind = 'close' | 'archive';
type SessionReleaseStage =
| 'set-archived'
| 'drain-agents'
| 'publish-archive'
| 'will-close'
| 'dispose-scope'
| 'will-release'
| 'flush'
| 'seal'
| 'drain-writes'
| 'lease-lost'
| 'shutdown';
interface SessionEntry {
phase: SessionEntryPhase;
state: SessionEntryState;
readonly handle: ISessionScopeHandle;
readonly lease: SessionLease;
readonly registration: IDisposable;
readonly scope: string;
disposed: boolean;
closeKind?: SessionCloseKind;
closeStep: number;
closePromise?: Promise<void>;
dirtyAbortPromise?: Promise<void>;
releasePromise?: Promise<void>;
}
export class SessionLifecycleService extends Disposable implements ISessionLifecycleService {
@ -191,7 +197,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
@ITelemetryService private readonly telemetry: ITelemetryService,
@ILogService private readonly log: ILogService,
@ICrossProcessLockService private readonly locks: ICrossProcessLockService,
@IWriteAuthorityRegistry private readonly authorityRegistry: IWriteAuthorityRegistry,
@IWriteGateRegistry private readonly writeGates: IWriteGateRegistry,
@ISessionLeaseContactProvider
private readonly leaseContact: ISessionLeaseContactProvider,
) {
@ -270,7 +276,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
const lease = await this.acquireSessionLease(opts.sessionId);
let registration: IDisposable;
try {
registration = this.authorityRegistry.register(lease);
registration = this.writeGates.register(sessionScope, lease);
} catch (error) {
lease.release();
throw error;
@ -289,13 +295,12 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
handle.accessor.get(ISessionWorkspaceContext).setAdditionalDirs(additionalDirs);
}
entry = {
phase: 'preparing',
state: 'opening',
handle,
lease,
registration,
scope: sessionScope,
disposed: false,
closeStep: 0,
};
this.entries.set(opts.sessionId, entry);
handle.accessor.get(ISessionExternalHooksService);
@ -347,7 +352,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
get(sessionId: string): ISessionScopeHandle | undefined {
const entry = this.entries.get(sessionId);
return entry?.phase === 'active' ? entry.handle : undefined;
return entry?.state === 'active' ? entry.handle : undefined;
}
resume(sessionId: string): Promise<ISessionScopeHandle | undefined> {
@ -403,73 +408,22 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
list(): readonly ISessionScopeHandle[] {
const ready: ISessionScopeHandle[] = [];
for (const entry of this.entries.values()) {
if (entry.phase === 'active') ready.push(entry.handle);
if (entry.state === 'active') ready.push(entry.handle);
}
return ready;
}
async close(sessionId: string): Promise<void> {
await this.closeSession(sessionId, 'close');
await this.releaseSession(sessionId, 'close');
}
async closeAll(): Promise<void> {
await this.beginClose();
const failures: unknown[] = [];
for (const [sessionId, entry] of this.entries) {
try {
if (entry.phase === 'flush-failed') {
await this.forceAbort(sessionId);
continue;
}
try {
await this.close(sessionId);
} catch (error) {
if (this.entries.get(sessionId)?.phase !== 'flush-failed') throw error;
await this.forceAbort(sessionId);
}
} catch (error) {
failures.push(error);
}
}
if (failures.length === 1) throw failures[0];
if (failures.length > 1) throw new AggregateError(failures, 'failed to close all sessions');
}
async forceAbort(sessionId: string): Promise<void> {
const entry = this.entries.get(sessionId);
if (entry === undefined) return;
if (entry.phase !== 'flush-failed') {
throw new Error2(
ErrorCodes.SESSION_DURABILITY_FAILED,
`session ${sessionId} can only be force-aborted after a durability barrier failure`,
{ details: { sessionId, phase: entry.phase } },
);
}
entry.lease.assertWritable();
await this.docs.update<SessionMeta>(entry.scope, 'state.json', (current) => {
if (current === undefined) {
throw new Error2(
ErrorCodes.SESSION_DURABILITY_FAILED,
`session ${sessionId} metadata is missing during force-abort`,
{ details: { sessionId } },
);
}
return {
...current,
custom: {
...current.custom,
dirtyAbort: { reason: 'flush-failed', at: Date.now() },
},
};
});
this.telemetry.track2('session_dirty_abort', { session_id: sessionId, reason: 'flush-failed' });
this.log.warn('force-aborting session after an ambiguous durability failure', { sessionId });
await this.dirtyAbortSession(entry);
await Promise.allSettled([...this.entries].map(([sessionId]) => this.close(sessionId)));
}
async archive(sessionId: string): Promise<void> {
await this.closeSession(sessionId, 'archive');
await this.releaseSession(sessionId, 'archive');
}
async restore(sessionId: string): Promise<ISessionScopeHandle | undefined> {
@ -483,15 +437,15 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
await this.hooks.onWillCloseSession.run(event);
}
private async announceWillRelease(
event: SessionWillReleaseEvent,
entry: SessionEntry,
): Promise<void> {
private async announceWillRelease(event: SessionWillReleaseEvent): Promise<void> {
try {
await this.hooks.onWillReleaseSession.run(event);
} catch (error) {
entry.phase = 'flush-failed';
throw error;
this.log.warn('session release hook failed', {
sessionId: event.sessionId,
reason: event.reason,
error: String(error),
});
}
}
@ -502,53 +456,72 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
}
}
private async closeSession(sessionId: string, kind: SessionCloseKind): Promise<void> {
private releaseSession(
sessionId: string,
kind: SessionReleaseKind,
): Promise<void> {
const entry = this.entries.get(sessionId);
if (entry === undefined) return;
if (entry.phase === 'preparing') {
if (entry === undefined) return Promise.resolve();
if (entry.state === 'opening') {
this.rollbackSession(entry);
return;
}
if (entry.phase === 'active') {
entry.phase = 'draining';
entry.closeKind = kind;
entry.closeStep = 0;
}
if (entry.closePromise !== undefined) {
await entry.closePromise;
return;
}
const closePromise = this.runSessionClose(sessionId, entry);
entry.closePromise = closePromise;
try {
await closePromise;
} finally {
if (this.entries.get(sessionId) === entry && entry.closePromise === closePromise) {
entry.closePromise = undefined;
}
return Promise.resolve();
}
if (entry.releasePromise !== undefined) return entry.releasePromise;
entry.state = 'closing';
const releasePromise = this.runSessionRelease(sessionId, entry, kind);
entry.releasePromise = releasePromise;
return releasePromise;
}
private async runSessionClose(sessionId: string, entry: SessionEntry): Promise<void> {
const kind = entry.closeKind ?? 'close';
const stepCount = kind === 'close' ? 4 : 6;
while (entry.closeStep < stepCount) {
if (kind === 'close') {
await this.runCloseStep(sessionId, entry);
} else {
await this.runArchiveStep(sessionId, entry);
}
entry.closeStep++;
}
private async runSessionRelease(
sessionId: string,
entry: SessionEntry,
kind: SessionReleaseKind,
): Promise<void> {
let stage: SessionReleaseStage = kind === 'archive' ? 'set-archived' : 'will-close';
try {
if (kind === 'archive') {
await entry.handle.accessor.get(ISessionMetadata).setArchived(true);
stage = 'drain-agents';
await this.drainAgents(entry.handle);
stage = 'publish-archive';
this.event.publish({
type: 'event.session.archived',
payload: { sessionId },
});
} else {
await this.announceWillClose({ sessionId, handle: entry.handle, reason: 'exit' });
stage = 'drain-agents';
await this.drainAgents(entry.handle);
}
stage = 'will-close';
if (kind === 'archive') {
await this.announceWillClose({ sessionId, handle: entry.handle, reason: 'exit' });
}
stage = 'dispose-scope';
this.disposeSessionHandle(entry);
stage = 'will-release';
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();
} catch (error) {
entry.phase = 'flush-failed';
throw error;
await this.abandonSession(entry, stage, error);
throw new Error2(
ErrorCodes.SESSION_DURABILITY_FAILED,
`session ${sessionId} was abandoned while releasing at ${stage}`,
{
details: { sessionId, stage },
cause: error,
},
);
}
entry.registration.dispose();
entry.lease.release();
if (this.entries.get(sessionId) === entry) this.entries.delete(sessionId);
this.finishSessionRelease(entry);
if (kind === 'archive') {
this._onDidArchiveSession.fire({ sessionId });
} else {
@ -556,51 +529,102 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
}
}
private async runCloseStep(sessionId: string, entry: SessionEntry): Promise<void> {
switch (entry.closeStep) {
case 0:
await this.announceWillClose({ sessionId, handle: entry.handle, reason: 'exit' });
return;
case 1:
await this.drainAgents(entry.handle);
return;
case 2:
this.disposeSessionHandle(entry);
return;
case 3:
await this.announceWillRelease({ sessionId, reason: 'close' }, entry);
return;
private async abandonSession(
entry: SessionEntry,
stage: SessionReleaseStage,
cause: unknown,
): Promise<void> {
const sessionId = entry.handle.id;
const reason = stage === 'flush' ? 'flush-failed' : 'release-failed';
this.log.warn('abandoning session after release failed', {
sessionId,
stage,
error: String(cause),
});
const taskServices = this.collectTaskServices(entry);
try {
this.disposeSessionHandle(entry);
} catch {
}
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();
this.finishSessionRelease(entry);
if (reason === 'flush-failed') {
this.telemetry.track2('session_dirty_abort', {
session_id: sessionId,
reason,
});
}
}
private async runArchiveStep(sessionId: string, entry: SessionEntry): Promise<void> {
switch (entry.closeStep) {
case 0:
await entry.handle.accessor.get(ISessionMetadata).setArchived(true);
return;
case 1:
await this.drainAgents(entry.handle);
return;
case 2:
this.event.publish({
type: 'event.session.archived',
payload: { sessionId },
});
return;
case 3:
await this.announceWillClose({ sessionId, handle: entry.handle, reason: 'exit' });
return;
case 4:
this.disposeSessionHandle(entry);
return;
case 5:
await this.announceWillRelease({ sessionId, reason: 'archive' }, entry);
return;
private async writeDirtyMarker(
entry: SessionEntry,
reason: 'flush-failed' | 'release-failed',
stage: SessionReleaseStage,
): Promise<void> {
try {
await this.docs.update<SessionMeta>(entry.scope, 'state.json', (current) => {
if (current === undefined) {
throw new Error2(
ErrorCodes.SESSION_DURABILITY_FAILED,
`session ${entry.handle.id} metadata is missing while writing dirty marker`,
{ details: { sessionId: entry.handle.id } },
);
}
return {
...current,
custom: {
...current.custom,
dirtyAbort: { reason, stage, at: Date.now() },
},
};
});
} catch (error) {
this.log.warn('failed to persist session dirty marker', {
sessionId: entry.handle.id,
error: String(error),
});
}
}
private collectTaskServices(entry: SessionEntry): IAgentTaskService[] {
try {
return entry.handle
.accessor.get(IAgentLifecycleService)
.list()
.map((agent) => agent.accessor.get(IAgentTaskService));
} catch {
return [];
}
}
private finishSessionRelease(entry: SessionEntry): void {
try {
entry.registration.dispose();
} catch (error) {
this.log.warn('failed to unregister session write gate', {
sessionId: entry.handle.id,
error: String(error),
});
}
try {
entry.lease.release();
} catch (error) {
this.log.warn('failed to release session lease', {
sessionId: entry.handle.id,
error: String(error),
});
}
if (this.entries.get(entry.handle.id) === entry) this.entries.delete(entry.handle.id);
}
private activateSession(entry: SessionEntry): void {
if (this.entries.get(entry.handle.id) !== entry || entry.phase !== 'preparing') {
if (this.entries.get(entry.handle.id) !== entry || entry.state !== 'opening') {
throw new Error2(
ErrorCodes.SESSION_LEASE_LOST,
`session ${entry.handle.id} was torn down before activation`,
@ -608,7 +632,7 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
);
}
entry.lease.assertWritable();
entry.phase = 'active';
entry.state = 'active';
}
private rollbackSession(entry: SessionEntry): void {
@ -877,7 +901,9 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
override dispose(): void {
this.closing = true;
for (const entry of this.entries.values()) void this.dirtyAbortSession(entry);
for (const entry of this.entries.values()) {
this.startAbandon(entry, 'shutdown', new Error('session lifecycle disposed'));
}
super.dispose();
}
@ -901,7 +927,32 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
private onLeaseLost(sessionId: string): void {
this.log.error('session lease lost; tearing the session down', { sessionId });
const entry = this.entries.get(sessionId);
if (entry !== undefined) void this.dirtyAbortSession(entry);
if (entry !== undefined) {
this.startAbandon(
entry,
'lease-lost',
new Error2(ErrorCodes.SESSION_LEASE_LOST, `session ${sessionId} lost its write lease`, {
details: { sessionId },
}),
);
}
}
private startAbandon(
entry: SessionEntry,
stage: SessionReleaseStage,
cause: unknown,
): void {
if (entry.releasePromise !== undefined) return;
entry.state = 'closing';
const releasePromise = this.abandonSession(entry, stage, cause);
entry.releasePromise = releasePromise;
void releasePromise.catch((error) => {
this.log.error('unexpected failure while abandoning session', {
sessionId: entry.handle.id,
error: String(error),
});
});
}
private async acquireSessionLease(sessionId: string): Promise<SessionLease> {
@ -960,46 +1011,6 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
}
}
private dirtyAbortSession(entry: SessionEntry): Promise<void> {
if (entry.dirtyAbortPromise !== undefined) return entry.dirtyAbortPromise;
if (this.entries.get(entry.handle.id) === entry) this.entries.delete(entry.handle.id);
let taskServices: IAgentTaskService[] = [];
try {
taskServices = entry.handle
.accessor.get(IAgentLifecycleService)
.list()
.map((agent) => agent.accessor.get(IAgentTaskService));
} catch {
}
try {
this.disposeSessionHandle(entry);
} catch {
}
const dirtyAbortPromise = (async (): Promise<void> => {
try {
await this.hooks.onWillReleaseSession.run({
sessionId: entry.handle.id,
reason: 'dirty-abort',
});
} catch (error) {
this.log.warn('session release hook failed during dirty abort', {
sessionId: entry.handle.id,
error: String(error),
});
}
await Promise.allSettled(taskServices.map((tasks) => tasks.flushPersistence()));
try {
entry.registration.dispose();
} catch {
}
entry.lease.release();
})();
entry.dirtyAbortPromise = dirtyAbortPromise;
void dirtyAbortPromise.catch(() => {});
return dirtyAbortPromise;
}
private async readMetaFromDisk(
workspaceId: string,
sessionId: string,

4
packages/agent-core-v2/src/index.ts
View file

@ -338,10 +338,10 @@ export * from '#/persistence/interface/appendLogStore';
export * from '#/persistence/interface/atomicDocumentStore';
export * from '#/persistence/interface/queryStore';
export * from '#/persistence/interface/blobStore';
export * from '#/persistence/interface/writeAuthority';
export * from '#/persistence/interface/writeGate';
export * from '#/persistence/backends/node-fs/fileStorageService';
export * from '#/persistence/backends/node-fs/appendLogStore';
export * from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
export * from '#/persistence/backends/node-fs/writeGateRegistryService';
export * from '#/persistence/backends/node-fs/atomicDocumentStore';
export * from '#/persistence/backends/node-fs/blobStoreService';
export * from '#/persistence/backends/node-fs/projectLocalConfigService';

24
packages/agent-core-v2/src/persistence/backends/node-fs/appendLogStore.ts
View file

@ -12,14 +12,8 @@
* post-rewrite drain is durable, supports scoped durability barriers, waits
* every selected key before a flush reports an error, and preserves failed
* retired buffers so their in-memory tail remains observable while replacement
* owners wait for the prior generation. Session-scoped writes (journal
* bytes under `sessions/<wsId>/<sessionId>`) are fenced: `drain` and `rewrite`
* re-verify the session's registered `ISessionWriteAuthority` through
* `IWriteAuthorityRegistry` immediately before bytes hit storage, a session
* scope with no registered authority fails closed, and a fencing failure
* sticks the buffer like any ambiguous storage failure (the session teardown
* follows). The root scope and scopes outside the sessions tree carry no
* authority and pass untouched. Bound at App scope.
* owners wait for the prior generation. Physical writes are admitted and
* tracked by the underlying byte-storage backend. Bound at App scope.
*/
import { InstantiationType } from '#/_base/di/extensions';
@ -27,10 +21,6 @@ import { toDisposable, type IDisposable } from '#/_base/di/lifecycle';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
import {
assertScopeWritable,
IWriteAuthorityRegistry,
} from '#/persistence/interface/writeAuthority';
import {
AppendLogCorruptedError,
IAppendLogStore,
@ -57,10 +47,7 @@ export class AppendLogStore implements IAppendLogStore {
private readonly logs = new Map<string, LogState>();
constructor(
@IFileSystemStorageService private readonly storage: IFileSystemStorageService,
@IWriteAuthorityRegistry private readonly authorityRegistry: IWriteAuthorityRegistry,
) {}
constructor(@IFileSystemStorageService private readonly storage: IFileSystemStorageService) {}
append<R>(scope: string, key: string, record: R, options?: AppendLogOptions): void {
const state = this.state(scope, key);
@ -124,7 +111,6 @@ export class AppendLogStore implements IAppendLogStore {
);
const rewrite = priorSettled.then(async () => {
try {
this.assertScopeWritable(scope);
await this.storage.write(scope, key, encoded, { atomic: true });
state.storageFailure = undefined;
} catch (error) {
@ -265,7 +251,6 @@ export class AppendLogStore implements IAppendLogStore {
while (state.pending.length > 0) {
const batch = state.pending.slice();
try {
this.assertScopeWritable(scope);
await this.storage.append(scope, key, encodeBatch(batch), { durable: true });
} catch (error) {
const failure = (state.storageFailure ??= { error });
@ -276,9 +261,6 @@ export class AppendLogStore implements IAppendLogStore {
}
}
private assertScopeWritable(scope: string): void {
assertScopeWritable(scope, this.authorityRegistry);
}
}
function logId(scope: string, key: string): string {

96
packages/agent-core-v2/src/persistence/backends/node-fs/fileStorageService.ts
View file

@ -20,9 +20,9 @@
* control over append offsets, fsync, atomic rename and streaming, which the
* 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 are fenced
* through the App-scoped write-authority registry immediately before storage
* I/O, so every file-backed Store shares the same fail-closed boundary.
* 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.
*/
import { createReadStream, mkdirSync, statSync } from 'node:fs';
@ -48,10 +48,7 @@ import type {
StorageWriteOptions,
} from '#/persistence/interface/storage';
import { StorageError, StorageErrors, toStorageIoError } from '#/persistence/interface/storage';
import {
assertScopeWritable,
IWriteAuthorityRegistry,
} from '#/persistence/interface/writeAuthority';
import { IWriteGateRegistry } from '#/persistence/interface/writeGate';
const WATCH_DEBOUNCE_MS = 150;
const STORAGE_LOCK_WAIT_TIMEOUT_MS = 10_000;
@ -81,8 +78,8 @@ export class FileStorageService implements IFileSystemStorageService {
private readonly dirMode?: number,
private readonly fileMode?: number,
@optional(ICrossProcessLockService) private readonly locks?: ICrossProcessLockService,
@optional(IWriteAuthorityRegistry)
private readonly authorityRegistry?: IWriteAuthorityRegistry,
@optional(IWriteGateRegistry)
private readonly writeGates?: IWriteGateRegistry,
) {}
async read(scope: string, key: string): Promise<Uint8Array | undefined> {
@ -122,19 +119,15 @@ export class FileStorageService implements IFileSystemStorageService {
_options: StorageWriteOptions = {},
): Promise<void> {
const filePath = this.path(scope, key);
this.assertScopeWritable(scope);
try {
await mkdir(dirname(filePath), { recursive: true, mode: this.dirMode });
} catch (error) {
throw toStorageIoError(error, { path: filePath, op: 'write' });
}
this.assertScopeWritable(scope);
try {
await atomicWrite(filePath, data, undefined, this.fileMode);
await this.syncDirOnce(dirname(filePath));
} catch (error) {
throw toStorageIoError(error, { path: filePath, op: 'write' });
}
await this.runWrite(scope, async () => {
try {
await mkdir(dirname(filePath), { recursive: true, mode: this.dirMode });
await atomicWrite(filePath, data, undefined, this.fileMode);
await this.syncDirOnce(dirname(filePath));
} catch (error) {
throw toStorageIoError(error, { path: filePath, op: 'write' });
}
});
}
async append(
@ -145,29 +138,25 @@ export class FileStorageService implements IFileSystemStorageService {
): Promise<void> {
const filePath = this.path(scope, key);
const dir = dirname(filePath);
this.assertScopeWritable(scope);
try {
await mkdir(dir, { recursive: true, mode: this.dirMode });
} catch (error) {
throw toStorageIoError(error, { path: filePath, op: 'append' });
}
this.assertScopeWritable(scope);
try {
const fh = await open(filePath, 'a', this.fileMode);
await this.runWrite(scope, async () => {
try {
if (data.byteLength > 0) {
await fh.writeFile(data);
await mkdir(dir, { recursive: true, mode: this.dirMode });
const fh = await open(filePath, 'a', this.fileMode);
try {
if (data.byteLength > 0) {
await fh.writeFile(data);
}
if (options.durable !== false) {
await fh.sync();
}
} finally {
await fh.close();
}
if (options.durable !== false) {
await fh.sync();
}
} finally {
await fh.close();
await this.syncDirOnce(dir);
} catch (error) {
throw toStorageIoError(error, { path: filePath, op: 'append' });
}
await this.syncDirOnce(dir);
} catch (error) {
throw toStorageIoError(error, { path: filePath, op: 'append' });
}
});
}
async list(scope: string, prefix?: string): Promise<readonly string[]> {
@ -183,13 +172,14 @@ export class FileStorageService implements IFileSystemStorageService {
async delete(scope: string, key: string): Promise<void> {
const filePath = this.path(scope, key);
this.assertScopeWritable(scope);
try {
await unlink(filePath);
} catch (error) {
if (isEnoent(error)) return;
throw toStorageIoError(error, { path: filePath, op: 'delete' });
}
await this.runWrite(scope, async () => {
try {
await unlink(filePath);
} catch (error) {
if (isEnoent(error)) return;
throw toStorageIoError(error, { path: filePath, op: 'delete' });
}
});
}
watch(scope: string, key: string): Event<void> {
@ -267,7 +257,7 @@ export class FileStorageService implements IFileSystemStorageService {
async runExclusive<T>(scope: string, key: string, op: () => Promise<T>): Promise<T> {
const filePath = this.path(scope, key);
const lockPath = `${filePath}.lock`;
this.assertScopeWritable(scope);
await this.runWrite(scope, async () => {});
if (this.locks === undefined) {
throw new StorageError(
StorageErrors.codes.STORAGE_IO_FAILED,
@ -280,7 +270,7 @@ export class FileStorageService implements IFileSystemStorageService {
lockPath,
{ wait: { timeoutMs: STORAGE_LOCK_WAIT_TIMEOUT_MS } },
async () => {
this.assertScopeWritable(scope);
await this.runWrite(scope, async () => {});
return op();
},
);
@ -312,8 +302,8 @@ export class FileStorageService implements IFileSystemStorageService {
return join(this.baseDir, scope);
}
private assertScopeWritable(scope: string): void {
assertScopeWritable(scope, this.authorityRegistry);
private runWrite<T>(scope: string, write: () => Promise<T>): Promise<T> {
return this.writeGates?.run(scope, write) ?? write();
}
private async syncDirOnce(dir: string): Promise<void> {

52
packages/agent-core-v2/src/persistence/backends/node-fs/writeAuthorityRegistryService.ts
View file

@ -1,52 +0,0 @@
/**
* `storage` domain (L1) — `IWriteAuthorityRegistry` implementation.
*
* A plain `sessionId → ISessionWriteAuthority` map with no storage or
* filesystem dependencies: the session lifecycle registers an authority when
* the session's lease is acquired and disposes the registration when the
* lease is released, and the `AppendLogStore` resolves authorities at drain /
* rewrite time. Double registration for the same session is a bug (two live
* writers for one session must never coexist), so it throws a
* `BugIndicatingError` instead of replacing. Bound at App scope.
*/
import { InstantiationType } from '#/_base/di/extensions';
import { toDisposable, type IDisposable } from '#/_base/di/lifecycle';
import { LifecycleScope, registerScopedService } from '#/_base/di/scope';
import { BugIndicatingError } from '#/_base/errors/errors';
import {
type ISessionWriteAuthority,
IWriteAuthorityRegistry,
} from '#/persistence/interface/writeAuthority';
export class WriteAuthorityRegistryService implements IWriteAuthorityRegistry {
declare readonly _serviceBrand: undefined;
private readonly authorities = new Map<string, ISessionWriteAuthority>();
register(authority: ISessionWriteAuthority): IDisposable {
if (this.authorities.get(authority.sessionId) !== undefined) {
throw new BugIndicatingError(
`write authority already registered for session ${authority.sessionId}`,
);
}
this.authorities.set(authority.sessionId, authority);
return toDisposable(() => {
if (this.authorities.get(authority.sessionId) === authority) {
this.authorities.delete(authority.sessionId);
}
});
}
resolve(sessionId: string): ISessionWriteAuthority | undefined {
return this.authorities.get(sessionId);
}
}
registerScopedService(
LifecycleScope.App,
IWriteAuthorityRegistry,
WriteAuthorityRegistryService,
InstantiationType.Eager,
'storage',
);

69
packages/agent-core-v2/src/persistence/backends/node-fs/writeGateRegistryService.ts Normal file
View file

@ -0,0 +1,69 @@
/**
* `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<string, ISessionWriteGate>();
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<T>(scope: string, write: () => Promise<T>): Promise<T> {
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',
);

75
packages/agent-core-v2/src/persistence/interface/writeAuthority.ts
View file

@ -1,75 +0,0 @@
/**
* `persistence/interface` — session-scoped write fencing contract.
*
* Defines `ISessionWriteAuthority`, the per-session lease proof that a Store
* write must re-verify immediately before its bytes hit storage (the
* pre-commit kernel-handle check is the hard gate and must fail closed), and the App-scoped
* `IWriteAuthorityRegistry` the `AppendLogStore` resolves authorities through.
* The registry never creates semantics of its own: it only maps `sessionId`
* to the authority the session lifecycle registered, so a write for a
* session with no registered authority is a bypass attempt and must be
* rejected. `sessionIdFromScope` keeps the filesystem-layout knowledge
* (`sessions/<wsId>/<sessionId>[/agents/<agentId>]`) in exactly one place and
* `assertScopeWritable` applies the fail-closed gate for every backend;
* the root scope (`''`, e.g. `session_index.jsonl`) and any scope outside
* the sessions tree deliberately carry no authority and pass untouched.
* The concrete registry lives in
* `persistence/backends/node-fs/writeAuthorityRegistryService.ts`.
*/
import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation';
import type { IDisposable } from '#/_base/di/lifecycle';
import { Error2, ErrorCodes } from '#/errors';
export interface ISessionWriteAuthority {
readonly sessionId: string;
/** Checks the held kernel-lock handle. Throws
`Error2(session.lease_lost)` when this instance no longer holds the
lease; must be called immediately before any durable write. */
assertWritable(): void;
}
export const IWriteAuthorityRegistry: ServiceIdentifier<IWriteAuthorityRegistry> =
createDecorator<IWriteAuthorityRegistry>('writeAuthorityRegistry');
export interface IWriteAuthorityRegistry {
readonly _serviceBrand: undefined;
/** Registers the session's authority. Throws when one is already
registered for the sessionId — double registration is a bug, never a
takeover. Dispose the returned handle to unregister. */
register(authority: ISessionWriteAuthority): IDisposable;
resolve(sessionId: string): ISessionWriteAuthority | undefined;
}
export function sessionIdFromScope(scope: string): string | undefined {
if (scope === '') return undefined;
const parts = scope.split('/');
if (parts.length < 3 || parts[0] !== 'sessions') return undefined;
const sessionId = parts[2];
return parts[1] === '' || sessionId === undefined || sessionId === '' ? undefined : sessionId;
}
/**
* The pre-write fencing gate every Store backend applies immediately before
* bytes hit storage: resolve the scope's session authority through the
* registry and re-verify it. The root scope and scopes outside the sessions
* tree carry no authority and pass untouched, as does a missing registry
* (a consumer whose DI binding is `@optional`); a session scope with no
* registered authority is a bypass attempt and fails closed with
* `Error2(session.lease_lost)`.
*/
export function assertScopeWritable(
scope: string,
authorityRegistry: IWriteAuthorityRegistry | undefined,
): void {
const sessionId = sessionIdFromScope(scope);
if (sessionId === undefined || authorityRegistry === undefined) return;
const authority = authorityRegistry.resolve(sessionId);
if (authority === undefined) {
throw new Error2(ErrorCodes.SESSION_LEASE_LOST, 'session has no registered write authority', {
details: { sessionId },
});
}
authority.assertWritable();
}

30
packages/agent-core-v2/src/persistence/interface/writeGate.ts Normal file
View file

@ -0,0 +1,30 @@
/**
* `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<T>(write: () => Promise<T>): Promise<T>;
seal(): void;
drained(): Promise<void>;
}
export const ISessionWriteGate: ServiceIdentifier<ISessionWriteGate> =
createDecorator<ISessionWriteGate>('sessionWriteGate');
export const IWriteGateRegistry: ServiceIdentifier<IWriteGateRegistry> =
createDecorator<IWriteGateRegistry>('writeGateRegistry');
export interface IWriteGateRegistry {
readonly _serviceBrand: undefined;
register(sessionScope: string, gate: ISessionWriteGate): IDisposable;
run<T>(scope: string, write: () => Promise<T>): Promise<T>;
}

65
packages/agent-core-v2/src/session/sessionLease/sessionLease.ts
View file

@ -1,22 +1,21 @@
/**
* `sessionLease` domain (L1) — the per-session write lease.
*
* Defines `ISessionLeaseService`, the Session-scope seeded capability that
* state writers use to verify they still own the session's durable state,
* and the `SessionLease` object that satisfies it: an App-owned wrapper
* 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
* (`SessionLifecycleService` builds it; it is deliberately not a DI service)
* around the cross-process lock handle at
* `<homeDir>/session-leases/<sessionId>.lock`. `assertWritable` is the hard
* gate: it checks the live kernel-lock handle — a released or replaced
* sentinel fails closed with
* `session.lease_lost`, marks the lease lost, and fires the loss callback
* exactly once so the owning session tears itself down. Release order is the
* lifecycle's business; `release()` only forwards to the idempotent kernel
* lock release.
* `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.
*
* No default is registered for `ISessionLeaseService`: every production
* No default is registered for either Session-scoped view: every production
* session scope is seeded by `sessionLifecycle` via {@link sessionLeaseSeed};
* resolving it unseeded (a session that bypassed materialization) is a bug
* resolving one unseeded (a session that bypassed materialization) is a bug
* and must fail loudly rather than silently disable the fencing gate.
*/
@ -29,7 +28,7 @@ import type {
CrossProcessLockInspection,
ICrossProcessLockHandle,
} from '#/os/interface/crossProcessLock';
import type { ISessionWriteAuthority } from '#/persistence/interface/writeAuthority';
import { ISessionWriteGate } from '#/persistence/interface/writeGate';
export const LEASE_CREATING_RETRY_AFTER_MS = 1000;
@ -101,13 +100,16 @@ export interface ISessionLeaseService {
export const ISessionLeaseService: ServiceIdentifier<ISessionLeaseService> =
createDecorator<ISessionLeaseService>('sessionLeaseService');
export class SessionLease implements ISessionWriteAuthority, ISessionLeaseService {
export class SessionLease implements ISessionWriteGate, ISessionLeaseService {
declare readonly _serviceBrand: undefined;
readonly lockId: string;
private _released = false;
private _lost = false;
private _lossFired = false;
private _sealed = false;
private inFlightWrites = 0;
private readonly drainWaiters = new Set<() => void>();
constructor(
readonly sessionId: string,
@ -143,8 +145,43 @@ export class SessionLease implements ISessionWriteAuthority, ISessionLeaseServic
}
}
async run<T>(write: () => Promise<T>): Promise<T> {
if (this._sealed) throw this.writeGateClosedError();
this.assertWritable();
this.inFlightWrites++;
try {
return await write();
} finally {
this.inFlightWrites--;
if (this.inFlightWrites === 0) {
for (const resolve of this.drainWaiters) resolve();
this.drainWaiters.clear();
}
}
}
seal(): void {
this._sealed = true;
}
drained(): Promise<void> {
if (this.inFlightWrites === 0) return Promise.resolve();
return new Promise<void>((resolve) => {
this.drainWaiters.add(resolve);
});
}
private writeGateClosedError(): Error2 {
return new Error2(
ErrorCodes.SESSION_LEASE_LOST,
`session ${this.sessionId} write gate is sealed`,
{ details: { sessionId: this.sessionId } },
);
}
private markLost(): void {
this._lost = true;
this.seal();
if (this._lossFired) return;
this._lossFired = true;
this.onLeaseLost(this.sessionId);
@ -152,6 +189,7 @@ export class SessionLease implements ISessionWriteAuthority, ISessionLeaseServic
release(): void {
if (this._released) return;
this.seal();
this._released = true;
this.handle.release();
}
@ -162,5 +200,8 @@ export function sessionLeasePath(homeDir: string, sessionId: string): string {
}
export function sessionLeaseSeed(lease: SessionLease): ScopeSeed {
return [[ISessionLeaseService as ServiceIdentifier<unknown>, lease]];
return [
[ISessionLeaseService as ServiceIdentifier<unknown>, lease],
[ISessionWriteGate as ServiceIdentifier<unknown>, lease],
];
}

14
packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts
View file

@ -13,11 +13,10 @@
* released v1 builds. Re-registering an agent whose metadata is unchanged is
* 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 passes the
* `sessionLease` hard gate first (`ISessionLeaseService.assertWritable`,
* checking the held kernel-lock handle), so an instance that lost the
* session lease fails closed instead of overwriting a live peer's state.
* Bound at Session scope.
* never reorders session listings. Every durable write is fenced by the
* storage backend's per-session write gate, so an instance that lost or
* released the session lease fails closed instead of overwriting a live
* peer's state. Bound at Session scope.
*
* Read-model mirroring (flag `persistence_minidb_readmodel`): after a metadata
* update is persisted, the fresh summary is mirrored into the `IQueryStore`
@ -37,7 +36,6 @@ import { IFlagService } from '#/app/flag/flag';
import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
import { IQueryStore } from '#/persistence/interface/queryStore';
import { ISessionContext } from '#/session/sessionContext/sessionContext';
import { ISessionLeaseService } from '#/session/sessionLease/sessionLease';
import {
ISessionMetadata,
@ -70,7 +68,6 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
@ILogService private readonly log: ILogService,
@IQueryStore private readonly queryStore: IQueryStore,
@IFlagService private readonly flags: IFlagService,
@ISessionLeaseService private readonly lease: ISessionLeaseService,
) {
super();
this.scope = ctx.metaScope;
@ -90,7 +87,6 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
private async applyUpdate(patch: SessionMetaPatch): Promise<void> {
await this.ready;
this.data = { ...this.data, ...patch, updatedAt: Date.now() };
this.lease.assertWritable();
await this.store.set(this.scope, META_KEY, this.data);
await this.mirrorToReadModel();
this._onDidChangeMetadata.fire({
@ -154,7 +150,6 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
agents: this.data.agents ?? {},
custom: this.data.custom ?? {},
};
this.lease.assertWritable();
await this.store.set(this.scope, META_KEY, this.data);
}
return;
@ -170,7 +165,6 @@ export class SessionMetadata extends Disposable implements ISessionMetadata {
agents: {},
custom: {},
};
this.lease.assertWritable();
await this.store.set(this.scope, META_KEY, this.data);
this.log.debug('session metadata created', { sessionId: this.ctx.sessionId });
}

3
packages/agent-core-v2/test/agent/contextMemory/splice-replay.test.ts
View file

@ -29,8 +29,6 @@ import { IEventBus } from '#/app/event/eventBus';
import { EventBusService } from '#/app/event/eventBusService';
import type { ContentPart } from '#/kosong/contract/message';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -155,7 +153,6 @@ function buildHost(key: string): Host {
const ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.stub(IAgentBlobService, blob);
ix.set(IEventBus, new SyncDescriptor(EventBusService));
ix.set(IAgentContextMemoryService, new SyncDescriptor(AgentContextMemoryService));

3
packages/agent-core-v2/test/agent/fullCompaction/compactionOps.test.ts
View file

@ -12,8 +12,6 @@ import {
fullCompactionComplete,
} from '#/agent/fullCompaction/compactionOps';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -33,7 +31,6 @@ function buildHost(key: string): { wire: IWireService; log: IAppendLogStore; eve
const ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.set(IEventBus, new SyncDescriptor(EventBusService));
const wire = registerTestAgentWire(ix, testWireScope(SCOPE, key), {
log: ix.get(IAppendLogStore),

3
packages/agent-core-v2/test/agent/goal/goalOps.test.ts
View file

@ -28,8 +28,6 @@ import { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor';
import { IAgentUsageService } from '#/agent/usage/usage';
import { ITelemetryService } from '#/app/telemetry/telemetry';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -117,7 +115,6 @@ function buildHost(key: string): {
const ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.set(IEventBus, new SyncDescriptor(EventBusService));
ix.stub(IAgentLoopService, createLoopStub());
ix.stub(IAgentUsageService, {

4
packages/agent-core-v2/test/agent/permissionMode/permissionMode.test.ts
View file

@ -13,8 +13,6 @@ import { AgentPermissionModeService } from '#/agent/permissionMode/permissionMod
import { PermissionModeModel } from '#/agent/permissionMode/permissionModeOps';
import type { PermissionMode } from '#/agent/permissionPolicy/types';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -59,7 +57,6 @@ beforeEach(() => {
ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.stub(IAgentContextInjectorService, injectorStub);
ix.set(IAgentPermissionModeService, new SyncDescriptor(AgentPermissionModeService));
log = ix.get(IAppendLogStore);
@ -202,7 +199,6 @@ describe('AgentPermissionModeService (wire-backed)', () => {
const ix2 = disposables.add(new TestInstantiationService());
ix2.stub(IFileSystemStorageService, new InMemoryStorageService());
ix2.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix2.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
const log2 = ix2.get(IAppendLogStore);
const fresh = registerTestAgentWire(ix2, testWireScope(SCOPE, 'permission-mode-replay'), {
log: log2,

4
packages/agent-core-v2/test/agent/permissionRules/permissionRules.test.ts
View file

@ -7,8 +7,6 @@ import { IAgentPermissionRulesService, type PermissionApprovalResultRecord, type
import { AgentPermissionRulesService } from '#/agent/permissionRules/permissionRulesService';
import { PermissionRulesModel } from '#/agent/permissionRules/permissionRulesOps';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -44,7 +42,6 @@ beforeEach(() => {
ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.set(IAgentPermissionRulesService, new SyncDescriptor(AgentPermissionRulesService));
log = ix.get(IAppendLogStore);
registerTestAgentWire(ix, testWireScope(SCOPE, KEY), { log });
@ -125,7 +122,6 @@ describe('AgentPermissionRulesService (wire-backed)', () => {
const ix2 = disposables.add(new TestInstantiationService());
ix2.stub(IFileSystemStorageService, new InMemoryStorageService());
ix2.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix2.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
const log2 = ix2.get(IAppendLogStore);
const fresh = registerTestAgentWire(ix2, testWireScope(SCOPE, 'permission-rules-replay'), {
log: log2,

3
packages/agent-core-v2/test/agent/plan/planOps.test.ts
View file

@ -13,8 +13,6 @@ import {
planRevision,
} from '#/agent/plan/planOps';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -34,7 +32,6 @@ function buildHost(key: string): { wire: IWireService; log: IAppendLogStore; eve
const ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.set(IEventBus, new SyncDescriptor(EventBusService));
const wire = registerTestAgentWire(ix, testWireScope(SCOPE, key), {
log: ix.get(IAppendLogStore),

3
packages/agent-core-v2/test/agent/profile/profileOps.test.ts
View file

@ -19,8 +19,6 @@ import { IAgentScopeContext, makeAgentScopeContext } from '#/agent/scopeContext/
import { IHostEnvironment } from '#/os/interface/hostEnvironment';
import { IHostFileSystem } from '#/os/interface/hostFileSystem';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -199,7 +197,6 @@ function buildHost(key: string): {
const host = disposables.add(new TestInstantiationService());
host.stub(IFileSystemStorageService, new InMemoryStorageService());
host.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
host.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
host.stub(ITelemetryService, createTelemetryStub());
host.stub(IAgentScopeContext, makeAgentScopeContext({ agentId: 'main', agentScope: '' }));
host.stub(

4
packages/agent-core-v2/test/agent/swarm/swarm.test.ts
View file

@ -30,8 +30,6 @@ import type { AgentProfile } from '#/app/agentProfileCatalog/agentProfileCatalog
import { ISessionAgentProfileCatalog } from '#/session/sessionAgentProfileCatalog/sessionAgentProfileCatalog';
import { IAgentProfileService } from '#/agent/profile/profile';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -134,7 +132,6 @@ describe('AgentSwarmService', () => {
ix.stub(IAgentContextMemoryService, stubContextMemory());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.set(IEventBus, new SyncDescriptor(EventBusService));
ix.stub(IAgentLoopService, stubLoopWithHooks());
ix.set(IAgentToolRegistryService, new SyncDescriptor(AgentToolRegistryService));
@ -208,7 +205,6 @@ describe('AgentSwarmService', () => {
const ix2 = disposables.add(new TestInstantiationService());
ix2.stub(IFileSystemStorageService, new InMemoryStorageService());
ix2.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix2.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
const fresh = registerTestAgentWire(ix2, testWireScope('wire', 'swarm-replay'), {
log: ix2.get(IAppendLogStore),
});

6
packages/agent-core-v2/test/agent/task/persist.test.ts
View file

@ -22,10 +22,8 @@ import {
} from '#/agent/task/task';
import { JsonAtomicDocumentStore } from '#/persistence/backends/node-fs/atomicDocumentStore';
import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
import type { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
const SESSION_SCOPE = 'session';
const AGENT_SCOPE = `${SESSION_SCOPE}/agents/main`;
@ -34,7 +32,6 @@ let disposables: DisposableStore;
let sessionDir: string;
let docs: IAtomicDocumentStore;
let bytes: IFileSystemStorageService;
let authorityRegistry: IWriteAuthorityRegistry;
let persistence: AgentTaskPersistence;
function sample(overrides: Partial<Extract<AgentTaskInfo, { kind: 'process' }>> = {}): Extract<AgentTaskInfo, { kind: 'process' }> {
@ -67,14 +64,12 @@ beforeEach(async () => {
ix.set(IAtomicDocumentStore, new SyncDescriptor(JsonAtomicDocumentStore));
docs = ix.get(IAtomicDocumentStore);
bytes = ix.get(IFileSystemStorageService);
authorityRegistry = new WriteAuthorityRegistryService();
persistence = new AgentTaskPersistence(
sessionDir,
SESSION_SCOPE,
docs,
bytes,
undefined,
authorityRegistry,
);
});
@ -94,7 +89,6 @@ describe('AgentTaskPersistence', () => {
docs,
bytes,
fallbackRoot,
authorityRegistry,
);
}

17
packages/agent-core-v2/test/agent/task/stubs.ts
View file

@ -14,8 +14,6 @@ import {
} from '#/agent/task/task';
import { JsonAtomicDocumentStore } from '#/persistence/backends/node-fs/atomicDocumentStore';
import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import type { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
export type TaskServiceTestManager = IAgentTaskService & {
loadFromDisk(): Promise<void>;
@ -26,20 +24,6 @@ export const TASK_TEST_SESSION_SCOPE = 'sessions/test-workspace/test-session';
export const TASK_TEST_AGENT_SCOPE = `${TASK_TEST_SESSION_SCOPE}/agents/main`;
/**
* A real write-authority registry pre-registered with a lenient authority for
* the test session, so persistence writes pass the fencing check without a
* kernel lease.
*/
export function stubWriteAuthorityRegistry(sessionId = 'test-session'): IWriteAuthorityRegistry {
const registry = new WriteAuthorityRegistryService();
registry.register({
sessionId,
assertWritable: () => {},
});
return registry;
}
export function createAgentTaskPersistence(homedir: string): AgentTaskPersistence {
const storage = new FileStorageService(homedir);
return new AgentTaskPersistence(
@ -48,6 +32,5 @@ export function createAgentTaskPersistence(homedir: string): AgentTaskPersistenc
new JsonAtomicDocumentStore(storage),
storage,
undefined,
stubWriteAuthorityRegistry(),
);
}

3
packages/agent-core-v2/test/agent/task/taskOps.test.ts
View file

@ -8,8 +8,6 @@ import { EventBusService } from '#/app/event/eventBusService';
import type { AgentTaskInfo } from '#/agent/task/task';
import { TaskModel, taskStarted, taskTerminated } from '#/agent/task/taskOps';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -29,7 +27,6 @@ function buildHost(key: string): { wire: IWireService; log: IAppendLogStore; eve
const ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.set(IEventBus, new SyncDescriptor(EventBusService));
const wire = registerTestAgentWire(ix, testWireScope(SCOPE, key), {
log: ix.get(IAppendLogStore),

5
packages/agent-core-v2/test/agent/task/taskService.test.ts
View file

@ -36,7 +36,6 @@ import { IAgentScopeContext, makeAgentScopeContext } from '#/agent/scopeContext/
import { ISessionContext, makeSessionContext } from '#/session/sessionContext/sessionContext';
import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { ITelemetryService } from '#/app/telemetry/telemetry';
import { IAgentToolRegistryService } from '#/agent/toolRegistry/toolRegistry';
import { createHooks } from '#/hooks';
@ -48,7 +47,7 @@ import { InMemoryStorageService } from '#/persistence/backends/memory/inMemorySt
import { stubContextMemory } from '../contextMemory/stubs';
import { stubLoopWithHooks } from '../loop/stubs';
import { stubWriteAuthorityRegistry, type TaskServiceTestManager } from './stubs';
import type { TaskServiceTestManager } from './stubs';
function fakeProcessTask(): AgentTask {
return {
@ -149,7 +148,6 @@ describe('AgentTaskService', () => {
flush: async () => {},
close: async () => {},
});
ix.stub(IWriteAuthorityRegistry, stubWriteAuthorityRegistry());
ix.set(IAgentTaskService, new SyncDescriptor(AgentTaskService));
});
afterEach(() => disposables.dispose());
@ -506,7 +504,6 @@ describe('AgentTaskService', () => {
);
ix.stub(IAtomicDocumentStore, docs);
ix.stub(IFileSystemStorageService, bytes);
ix.stub(IWriteAuthorityRegistry, stubWriteAuthorityRegistry());
ix.set(IAgentTaskService, new SyncDescriptor(AgentTaskService));
return ix;
}

4
packages/agent-core-v2/test/agent/usage/usage.test.ts
View file

@ -11,8 +11,6 @@ import {
import { AgentUsageService } from '#/agent/usage/usageService';
import { UsageModel } from '#/agent/usage/usageOps';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -36,7 +34,6 @@ beforeEach(() => {
ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.set(IEventBus, new SyncDescriptor(EventBusService));
ix.set(IAgentUsageService, new SyncDescriptor(AgentUsageService));
log = ix.get(IAppendLogStore);
@ -62,7 +59,6 @@ function createFreshWire(logKey: string): { readonly fresh: IWireService; readon
const freshIx = disposables.add(new TestInstantiationService());
freshIx.stub(IFileSystemStorageService, new InMemoryStorageService());
freshIx.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
freshIx.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
const freshLog = freshIx.get(IAppendLogStore);
const fresh = registerTestAgentWire(freshIx, testWireScope(SCOPE, logKey), {
log: freshLog,

5
packages/agent-core-v2/test/agent/userTool/userTool.test.ts
View file

@ -10,8 +10,6 @@ import { IAgentUserToolService, type UserToolRegistration } from '#/agent/userTo
import { AgentUserToolService } from '#/agent/userTool/userToolService';
import { UserToolModel } from '#/agent/userTool/userToolOps';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -77,7 +75,6 @@ beforeEach(() => {
ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.set(IAgentToolRegistryService, new SyncDescriptor(AgentToolRegistryService));
profile = createProfileStub();
ix.stub(IAgentProfileService, profile);
@ -142,7 +139,6 @@ describe('AgentUserToolService (wire-backed)', () => {
const ixChild = disposables.add(new TestInstantiationService());
ixChild.stub(IFileSystemStorageService, new InMemoryStorageService());
ixChild.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ixChild.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ixChild.set(IAgentToolRegistryService, new SyncDescriptor(AgentToolRegistryService));
const childProfile = createProfileStub();
ixChild.stub(IAgentProfileService, childProfile);
@ -189,7 +185,6 @@ describe('AgentUserToolService (wire-backed)', () => {
const ix2 = disposables.add(new TestInstantiationService());
ix2.stub(IFileSystemStorageService, new InMemoryStorageService());
ix2.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix2.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix2.set(IAgentToolRegistryService, new SyncDescriptor(AgentToolRegistryService));
const profile2 = createProfileStub();
ix2.stub(IAgentProfileService, profile2);

1
packages/agent-core-v2/test/app/externalHooksRunner/integration.test.ts
View file

@ -202,7 +202,6 @@ function stubSessionLifecycle(): ISessionLifecycleService {
resume: async () => undefined,
close: async () => {},
closeAll: async () => {},
forceAbort: async () => {},
archive: async () => {},
restore: async () => undefined,
fork: async () => {

1
packages/agent-core-v2/test/app/sessionExport/sessionExport.test.ts
View file

@ -866,7 +866,6 @@ function registerSessionExportServices(
resume: async () => options.lifecycleHandle,
close: async () => {},
closeAll: async () => {},
forceAbort: async () => {},
archive: async () => {},
restore: async () => options.lifecycleHandle,
fork: async () => {

146
packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts
View file

@ -46,10 +46,10 @@ 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 { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { ISessionWriteGate, IWriteGateRegistry } from '#/persistence/interface/writeGate';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { WriteGateRegistryService } from '#/persistence/backends/node-fs/writeGateRegistryService';
import { ISessionWorkspaceContext } from '#/session/workspaceContext/workspaceContext';
import { SessionWorkspaceContextService } from '#/session/workspaceContext/workspaceContextService';
import { IWorkspaceService, type Workspace } from '#/app/workspace/workspace';
@ -513,7 +513,7 @@ describe('SessionLifecycleService', () => {
stubPair(ILogService, stubLog()),
stubPair(IFlagService, stubFlag(false)),
stubPair(ICrossProcessLockService, stubCrossProcessLock()),
stubPair(IWriteAuthorityRegistry, new WriteAuthorityRegistryService()),
stubPair(IWriteGateRegistry, new WriteGateRegistryService()),
stubPair(ISessionLeaseContactProvider, new SessionLeaseContactProvider()),
...extra,
]);
@ -1015,6 +1015,21 @@ describe('SessionLifecycleService', () => {
expect(closed).toEqual(['s1']);
});
it('logs and continues when a release hook fails', async () => {
const svc = build();
const closed: string[] = [];
svc.onDidCloseSession((e) => closed.push(e.sessionId));
svc.hooks.onWillReleaseSession.register('failing-hook', async () => {
throw new Error('hook failed');
});
await svc.create({ sessionId: 's1', workDir: '/tmp/proj' });
await expect(svc.close('s1')).resolves.toBeUndefined();
expect(closed).toEqual(['s1']);
expect(svc.get('s1')).toBeUndefined();
});
it('fires onDidArchiveSession when a session is archived', async () => {
const svc = build([
stubPair(IAgentLifecycleService, {
@ -1355,7 +1370,7 @@ describe('SessionLifecycleService', () => {
return [
stubPair(IBootstrapService, tmpBootstrapStub(root)),
stubPair(ICrossProcessLockService, new CrossProcessLockService()),
stubPair(IWriteAuthorityRegistry, new WriteAuthorityRegistryService()),
stubPair(IWriteGateRegistry, new WriteGateRegistryService()),
...over,
];
}
@ -1363,20 +1378,20 @@ describe('SessionLifecycleService', () => {
function realAlsSeeds(root: string): {
seeds: ReturnType<typeof stubPair>[];
appendLog: AppendLogStore;
registry: WriteAuthorityRegistryService;
registry: WriteGateRegistryService;
storage: FileStorageService;
docs: JsonAtomicDocumentStore;
} {
const registry = new WriteAuthorityRegistryService();
const registry = new WriteGateRegistryService();
const locks = new CrossProcessLockService();
const storage = new FileStorageService(root, undefined, undefined, locks);
const appendLog = new AppendLogStore(storage, registry);
const storage = new FileStorageService(root, undefined, undefined, locks, registry);
const appendLog = new AppendLogStore(storage);
const docs = new JsonAtomicDocumentStore(storage);
return {
seeds: [
stubPair(IBootstrapService, tmpBootstrapStub(root)),
stubPair(ICrossProcessLockService, locks),
stubPair(IWriteAuthorityRegistry, registry),
stubPair(IWriteGateRegistry, registry),
stubPair(IAppendLogStore, appendLog),
stubPair(IAtomicDocumentStore, docs),
],
@ -1555,27 +1570,49 @@ describe('SessionLifecycleService', () => {
await expectLeaseFree(root, 's1');
});
it('keeps authority and lease when the session durability barrier fails', async () => {
it('seals new writes and waits for an admitted write before releasing the lease', async () => {
const root = await makeTmpRoot();
const { seeds, appendLog, registry, storage } = realAlsSeeds(root);
const svc = build(seeds);
await svc.create({ sessionId: 's1', workDir: '/tmp/proj' });
const failure = new Error('durable append failed');
poisonSessionAppend(storage, 's1', failure);
appendLog.append('sessions/wd_stub/s1/agents/main', 'wire.jsonl', { tail: true });
const svc = build(realInstanceSeeds(root));
const handle = await svc.create({ sessionId: 's1', workDir: '/tmp/proj' });
const writeGate = handle.accessor.get(ISessionWriteGate);
let finishWrite!: () => void;
const writeBlocked = new Promise<void>((resolve) => {
finishWrite = resolve;
});
let enterWrite!: () => void;
const writeEntered = new Promise<void>((resolve) => {
enterWrite = resolve;
});
const write = writeGate.run(async () => {
enterWrite();
await writeBlocked;
});
await writeEntered;
await expect(svc.close('s1')).rejects.toBe(failure);
expect(svc.get('s1')).toBeUndefined();
expect(registry.resolve('s1')).toBeDefined();
await expect(stat(leaseFile(root, 's1'))).resolves.toBeDefined();
await expect(
new CrossProcessLockService().acquire(leaseFile(root, 's1')),
).rejects.toMatchObject({ code: CrossProcessLockErrorCode.Held });
await expect(svc.close('s1')).rejects.toBe(failure);
expect(registry.resolve('s1')).toBeDefined();
let closeSettled = false;
const closing = svc.close('s1').finally(() => {
closeSettled = true;
});
let sealed = false;
for (let attempt = 0; attempt < 10 && !sealed; attempt++) {
try {
await writeGate.run(async () => {});
} catch (error) {
expect(error).toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST });
sealed = true;
}
if (!sealed) await tick();
}
expect(sealed).toBe(true);
expect(closeSettled).toBe(false);
finishWrite();
await write;
await closing;
await expectLeaseFree(root, 's1');
});
it('requires an explicit dirty abort before releasing a failed durability lease', async () => {
it('abandons and releases the lease when the session durability barrier fails', async () => {
const root = await makeTmpRoot();
const { seeds, appendLog, registry, storage, docs } = realAlsSeeds(root);
const svc = build(seeds);
@ -1585,10 +1622,38 @@ describe('SessionLifecycleService', () => {
poisonSessionAppend(storage, 's1', failure);
appendLog.append('sessions/wd_stub/s1/agents/main', 'wire.jsonl', { tail: true });
await expect(svc.close('s1')).rejects.toBe(failure);
await svc.forceAbort('s1');
await expect(svc.close('s1')).rejects.toMatchObject({
code: ErrorCodes.SESSION_DURABILITY_FAILED,
cause: failure,
details: { sessionId: 's1', stage: 'flush' },
});
expect(svc.get('s1')).toBeUndefined();
await expect(
registry.run('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',
'state.json',
)).toMatchObject({ custom: { dirtyAbort: { reason: 'flush-failed' } } });
await expectLeaseReleased(root, 's1');
await expectLeaseFree(root, 's1');
await expect(svc.close('s1')).resolves.toBeUndefined();
});
it('reports an automatic dirty abort after a durability failure', async () => {
const root = await makeTmpRoot();
const { seeds, appendLog, storage } = realAlsSeeds(root);
const svc = build(seeds);
await svc.create({ sessionId: 's1', workDir: '/tmp/proj' });
const failure = new Error('durable append failed');
poisonSessionAppend(storage, 's1', failure);
appendLog.append('sessions/wd_stub/s1/agents/main', 'wire.jsonl', { tail: true });
await expect(svc.close('s1')).rejects.toMatchObject({
code: ErrorCodes.SESSION_DURABILITY_FAILED,
cause: failure,
});
expect(registry.resolve('s1')).toBeUndefined();
await expectLeaseReleased(root, 's1');
await expectLeaseFree(root, 's1');
expect(telemetryRecords).toContainEqual({
@ -1608,7 +1673,9 @@ describe('SessionLifecycleService', () => {
await svc.closeAll();
expect(registry.resolve('s1')).toBeUndefined();
await expect(
registry.run('sessions/wd_stub/s1', async () => {}),
).rejects.toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST });
expect(await docs.get<{ custom?: { dirtyAbort?: { reason?: string } } }>(
'sessions/wd_stub/s1',
'state.json',
@ -1616,24 +1683,17 @@ describe('SessionLifecycleService', () => {
await expectLeaseFree(root, 's1');
});
it('includes an already flush-failed session when closeAll drains materialized entries', async () => {
it('shares one release when close and closeAll overlap', async () => {
const root = await makeTmpRoot();
const { seeds, appendLog, registry, storage, docs } = realAlsSeeds(root);
const { seeds, registry } = realAlsSeeds(root);
const svc = build(seeds);
await svc.create({ sessionId: 's1', workDir: '/tmp/proj' });
await writeStateDoc(docs, 's1');
const failure = new Error('durable append failed');
poisonSessionAppend(storage, 's1', failure);
appendLog.append('sessions/wd_stub/s1/agents/main', 'wire.jsonl', { tail: true });
await expect(svc.close('s1')).rejects.toBe(failure);
await svc.closeAll();
await Promise.all([svc.close('s1'), svc.closeAll()]);
expect(registry.resolve('s1')).toBeUndefined();
expect(await docs.get<{ custom?: { dirtyAbort?: { reason?: string } } }>(
'sessions/wd_stub/s1',
'state.json',
)).toMatchObject({ custom: { dirtyAbort: { reason: 'flush-failed' } } });
await expect(
registry.run('sessions/wd_stub/s1', async () => {}),
).rejects.toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST });
await expectLeaseFree(root, 's1');
});

3
packages/agent-core-v2/test/harness/agent.ts
View file

@ -165,12 +165,10 @@ import { ISessionQuestionService, type QuestionResult } from '#/session/question
import { ISessionSkillCatalog } from '#/session/sessionSkillCatalog/skillCatalog';
import { ISessionSwarmService } from '#/session/swarm/sessionSwarm';
import type { PathAccessOperation } from '#/session/workspaceContext/workspaceContext';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { IHostFsWatchService } from '#/os/interface/hostFsWatch';
import { recordAgentEvents, type RecordedEventEntry } from '../snapshot/events';
import { createFakeHostFs, createFakeProcessRunner } from '../tools/fixtures/fake-exec';
import { stubWriteAuthorityRegistry } from '../agent/task/stubs';
import { fakeHostFsWatch } from '../session/sessionFs/stubs';
import { stubSessionLeaseService } from '../session/sessionLease/stubs';
import { createScriptedGenerate } from './scripted-generate';
@ -997,7 +995,6 @@ export class AgentTestContext {
})) {
reg.defineInstance(id, value);
}
reg.defineInstance(IWriteAuthorityRegistry, stubWriteAuthorityRegistry(sessionId));
reg.defineInstance(IHostFsWatchService, fakeHostFsWatch().service);
const memoryStorage = (): SyncDescriptor<IFileSystemStorageService> =>
new SyncDescriptor(InMemoryStorageService, [], true);

4
packages/agent-core-v2/test/index.test.ts
View file

@ -18,8 +18,6 @@ import { SyncDescriptor } from '#/_base/di/descriptors';
import { DisposableStore } from '#/_base/di/lifecycle';
import { TestInstantiationService } from '#/_base/di/test';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -101,7 +99,6 @@ describe('v1 wire vocabulary', () => {
const ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
log = ix.get(IAppendLogStore);
wire = registerTestAgentWire(ix, SCOPE, { log });
});
@ -160,7 +157,6 @@ describe('v1 wire vocabulary', () => {
const ix2 = store.add(new TestInstantiationService());
ix2.stub(IFileSystemStorageService, new InMemoryStorageService());
ix2.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix2.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
const log2 = ix2.get(IAppendLogStore);
const fresh = registerTestAgentWire(ix2, SCOPE, { log: log2 });

135
packages/agent-core-v2/test/persistence/backends/node-fs/appendLogStore.test.ts
View file

@ -7,8 +7,6 @@
* test/persistence/backends/node-fs/appendLogStore.test.ts`.
*/
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
@ -16,18 +14,10 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { SyncDescriptor } from '#/_base/di/descriptors';
import { DisposableStore } from '#/_base/di/lifecycle';
import { TestInstantiationService } from '#/_base/di/test';
import { ErrorCodes } from '#/errors';
import { AppendLogCorruptedError, IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import {
sessionIdFromScope,
IWriteAuthorityRegistry,
} from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { CrossProcessLockService } from '#/os/backends/node-local/crossProcessLockService';
import { SessionLease, sessionLeasePath } from '#/session/sessionLease/sessionLease';
const enc = new TextEncoder();
@ -66,7 +56,6 @@ describe('AppendLogStore', () => {
storage = new InMemoryStorageService();
ix.stub(IFileSystemStorageService, storage);
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
record = ix.get(IAppendLogStore);
});
@ -595,7 +584,6 @@ describe('AppendLogStore', () => {
const localIx = disposables.add(new TestInstantiationService());
localIx.stub(IFileSystemStorageService, chunkedStorage(chunks));
localIx.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
localIx.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
const log = localIx.get(IAppendLogStore);
const out: Rec[] = [];
@ -612,7 +600,6 @@ describe('AppendLogStore', () => {
const localIx = disposables.add(new TestInstantiationService());
localIx.stub(IFileSystemStorageService, chunkedStorage(chunks));
localIx.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
localIx.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
const log = localIx.get(IAppendLogStore);
const first: Array<{ type: string }> = [];
@ -637,7 +624,6 @@ describe('AppendLogStore', () => {
const localIx = disposables.add(new TestInstantiationService());
localIx.stub(IFileSystemStorageService, chunkedStorage(chunks));
localIx.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
localIx.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
const log = localIx.get(IAppendLogStore);
const readAll = async (): Promise<Array<{ s: string }>> => {
@ -657,7 +643,6 @@ describe('AppendLogStore', () => {
const localIx = disposables.add(new TestInstantiationService());
localIx.stub(IFileSystemStorageService, chunkedStorage(chunks));
localIx.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
localIx.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
const log = localIx.get(IAppendLogStore);
const out: Array<Rec & { s?: string }> = [];
@ -668,124 +653,4 @@ describe('AppendLogStore', () => {
]);
});
describe('write fencing', () => {
const SESSION_SCOPE = 'sessions/wd_test/s1';
let tmpDir: string;
let locks: CrossProcessLockService;
let registry: WriteAuthorityRegistryService;
beforeEach(() => {
tmpDir = mkdtempSync(join(tmpdir(), 'kimi-als-fence-'));
locks = new CrossProcessLockService();
registry = new WriteAuthorityRegistryService();
});
afterEach(() => {
rmSync(tmpDir, { recursive: true, force: true });
});
function makeStore(): { store: IAppendLogStore; storage: InMemoryStorageService } {
const storage = new InMemoryStorageService();
const localIx = disposables.add(new TestInstantiationService());
localIx.stub(IFileSystemStorageService, storage);
localIx.stub(IWriteAuthorityRegistry, registry);
localIx.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
return { store: localIx.get(IAppendLogStore), storage };
}
async function leaseFor(sessionId: string): Promise<SessionLease> {
return new SessionLease(
sessionId,
await locks.acquire(sessionLeasePath(tmpDir, sessionId)),
() => {},
);
}
it('parses the session id out of session and agent scopes only', () => {
expect(sessionIdFromScope('')).toBeUndefined();
expect(sessionIdFromScope('sessions')).toBeUndefined();
expect(sessionIdFromScope('sessions/wd_test')).toBeUndefined();
expect(sessionIdFromScope('sessions//s1')).toBeUndefined();
expect(sessionIdFromScope('wire')).toBeUndefined();
expect(sessionIdFromScope('agents/main')).toBeUndefined();
expect(sessionIdFromScope('credentials/x')).toBeUndefined();
expect(sessionIdFromScope('sessions/wd_test/s1')).toBe('s1');
expect(sessionIdFromScope('sessions/wd_test/s1/agents/main')).toBe('s1');
});
it('flush rejects with session.lease_lost when the lease is gone and writes no bytes', async () => {
const lease = await leaseFor('s1');
registry.register(lease);
const { store, storage } = makeStore();
let appendAttempts = 0;
const originalAppend = storage.append.bind(storage);
storage.append = async (...args) => {
appendAttempts++;
return originalAppend(...args);
};
lease.release();
store.append(SESSION_SCOPE, KEY, { n: 1 });
await expect(store.flush()).rejects.toMatchObject({
code: ErrorCodes.SESSION_LEASE_LOST,
});
// Sticky like any ambiguous storage failure: the buffer does not retry.
await expect(store.flush()).rejects.toMatchObject({
code: ErrorCodes.SESSION_LEASE_LOST,
});
expect(appendAttempts).toBe(0);
expect(await storage.read(SESSION_SCOPE, KEY)).toBeUndefined();
});
it('rewrite is fenced by the same hard gate', async () => {
const lease = await leaseFor('s1');
registry.register(lease);
const { store, storage } = makeStore();
let writeAttempts = 0;
const originalWrite = storage.write.bind(storage);
storage.write = async (...args) => {
writeAttempts++;
return originalWrite(...args);
};
lease.release();
await expect(store.rewrite(SESSION_SCOPE, KEY, [{ n: 1 }])).rejects.toMatchObject({
code: ErrorCodes.SESSION_LEASE_LOST,
});
expect(writeAttempts).toBe(0);
expect(await storage.read(SESSION_SCOPE, KEY)).toBeUndefined();
});
it('session-scoped writes without a registered authority fail closed', async () => {
const { store, storage } = makeStore();
store.append(SESSION_SCOPE, KEY, { n: 1 });
await expect(store.flush()).rejects.toMatchObject({
code: ErrorCodes.SESSION_LEASE_LOST,
message: 'session has no registered write authority',
details: { sessionId: 's1' },
});
expect(await storage.read(SESSION_SCOPE, KEY)).toBeUndefined();
});
it('flush awaits the retired buffer final flush before returning', async () => {
const lease = await leaseFor('s1');
registry.register(lease);
const { store, storage } = makeStore();
const handle = store.acquire(SESSION_SCOPE, KEY);
store.append(SESSION_SCOPE, KEY, { n: 1 });
handle.dispose();
await store.flush();
const bytes = await storage.read(SESSION_SCOPE, KEY);
expect(new TextDecoder().decode(bytes)).toBe('{"n":1}\n');
});
it('double registration for a session is rejected and dispose unregisters', () => {
const registration = registry.register({ sessionId: 's1', assertWritable: () => {} });
expect(() => registry.register({ sessionId: 's1', assertWritable: () => {} })).toThrow(
/already registered/,
);
registration.dispose();
expect(registry.resolve('s1')).toBeUndefined();
});
});
});

28
packages/agent-core-v2/test/persistence/backends/node-fs/fileStorageService.test.ts
View file

@ -6,7 +6,8 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { Error2, ErrorCodes } from '#/errors';
import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { WriteGateRegistryService } from '#/persistence/backends/node-fs/writeGateRegistryService';
import type { ISessionWriteGate } from '#/persistence/interface/writeGate';
const isWin = process.platform === 'win32';
const encoder = new TextEncoder();
@ -101,17 +102,23 @@ describe('FileStorageService — session write fencing', () => {
await rm(dir, { recursive: true, force: true });
});
it('revalidates the session authority for write, append, and delete', async () => {
const registry = new WriteAuthorityRegistryService();
it('runs write, append, and delete through the session gate', async () => {
const registry = new WriteGateRegistryService();
let writable = true;
const registration = registry.register({
sessionId: 'session',
assertWritable: () => {
const gate: ISessionWriteGate = {
run: async (write) => {
if (!writable) {
throw new Error2(ErrorCodes.SESSION_LEASE_LOST, 'session lease lost');
}
return write();
},
});
seal: () => {},
drained: async () => {},
};
const registration = registry.register('sessions/workspace/session', gate);
expect(() => registry.register('sessions/workspace/session', gate)).toThrow(
/already registered/,
);
const svc = new FileStorageService(dir, undefined, undefined, undefined, registry);
const scope = 'sessions/workspace/session/agents/main/tool-results';
@ -131,10 +138,13 @@ describe('FileStorageService — session write fencing', () => {
});
expect(await readFile(join(dir, scope, 'result.txt'), 'utf8')).toBe('ab');
registration.dispose();
await expect(svc.append(scope, 'result.txt', encoder.encode('c'))).rejects.toMatchObject({
code: ErrorCodes.SESSION_LEASE_LOST,
});
});
it('fails closed without a session authority and leaves non-session scopes untouched', async () => {
const registry = new WriteAuthorityRegistryService();
it('fails closed without a session gate and leaves non-session scopes untouched', async () => {
const registry = new WriteGateRegistryService();
const svc = new FileStorageService(dir, undefined, undefined, undefined, registry);
await expect(

35
packages/agent-core-v2/test/session/sessionLease/sessionLease.test.ts
View file

@ -2,7 +2,7 @@
* `sessionLease` domain — unit tests for the per-session write lease.
*
* Runs against the real node-local kernel-lock service rooted at a mkdtemp
* home, asserting the once-only loss notification and idempotent release.
* home, asserting loss notification, write admission/draining, and release.
*/
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs';
@ -70,4 +70,37 @@ describe('SessionLease', () => {
expect(lease.info).toBeUndefined();
expect(thrownError(() => lease.assertWritable()).code).toBe(ErrorCodes.SESSION_LEASE_LOST);
});
it('seal rejects new writes while drained waits for an admitted write', async () => {
const lease = await acquire();
let enterWrite!: () => void;
const writeEntered = new Promise<void>((resolve) => {
enterWrite = resolve;
});
let finishWrite!: () => void;
const writeGate = new Promise<void>((resolve) => {
finishWrite = resolve;
});
const write = lease.run(async () => {
enterWrite();
await writeGate;
});
await writeEntered;
lease.seal();
let drained = false;
const drain = lease.drained().then(() => {
drained = true;
});
await expect(lease.run(async () => {})).rejects.toMatchObject({
code: ErrorCodes.SESSION_LEASE_LOST,
});
expect(drained).toBe(false);
finishWrite();
await write;
await drain;
expect(drained).toBe(true);
lease.release();
});
});

31
packages/agent-core-v2/test/session/sessionMetadata/sessionMetadata.test.ts
View file

@ -5,9 +5,7 @@ import { DisposableStore } from '#/_base/di/lifecycle';
import { TestInstantiationService } from '#/_base/di/test';
import { IFlagService } from '#/app/flag/flag';
import { ILogService } from '#/_base/log/log';
import { Error2, ErrorCodes } from '#/errors';
import { ISessionContext, makeSessionContext } from '#/session/sessionContext/sessionContext';
import { ISessionLeaseService } from '#/session/sessionLease/sessionLease';
import { ISessionMetadata } from '#/session/sessionMetadata/sessionMetadata';
import { SessionMetadata } from '#/session/sessionMetadata/sessionMetadataService';
import { JsonAtomicDocumentStore } from '#/persistence/backends/node-fs/atomicDocumentStore';
@ -19,7 +17,6 @@ import { IQueryStore } from '#/persistence/interface/queryStore';
import { stubFlag } from '../../app/flag/stubs';
import { stubLog } from '../../_base/log/stubs';
import { stubQueryStore } from '../../persistence/interface/stubs';
import { stubSessionLeaseService } from '../sessionLease/stubs';
const META_SCOPE = 'sessions/wd_test/s1/session-meta';
@ -45,7 +42,6 @@ describe('SessionMetadata', () => {
ix.stub(ISessionContext, makeContext());
ix.stub(IQueryStore, stubQueryStore());
ix.stub(IFlagService, stubFlag(false));
ix.stub(ISessionLeaseService, stubSessionLeaseService());
ix.set(IFileSystemStorageService, new SyncDescriptor(InMemoryStorageService));
ix.set(IAtomicDocumentStore, new SyncDescriptor(JsonAtomicDocumentStore));
ix.set(ISessionMetadata, new SyncDescriptor(SessionMetadata));
@ -252,31 +248,4 @@ describe('SessionMetadata', () => {
expect(next.updatedAt).toBeGreaterThan(before);
});
it('gates updates behind the lease and leaves state.json untouched when it fails', async () => {
let leaseLost = false;
ix.stub(
ISessionLeaseService,
stubSessionLeaseService({
assertWritable: () => {
if (leaseLost) {
throw new Error2(ErrorCodes.SESSION_LEASE_LOST, 'lease lost', {
details: { sessionId: 's1' },
});
}
},
}),
);
const meta = ix.get(ISessionMetadata);
await meta.ready;
leaseLost = true;
await expect(meta.update({ title: 'x' })).rejects.toMatchObject({
code: ErrorCodes.SESSION_LEASE_LOST,
});
const raw = await ix
.get(IAtomicDocumentStore)
.get<{ title?: string }>(META_SCOPE, 'state.json');
expect(raw?.title).toBeUndefined();
});
});

3
packages/agent-core-v2/test/wire/persistence.test.ts
View file

@ -22,8 +22,6 @@ import {
WIRE_PROTOCOL_VERSION,
IFileSystemStorageService,
IAppendLogStore,
IWriteAuthorityRegistry,
WriteAuthorityRegistryService,
type WireRecord,
} from '#/index';
import { IWireService } from '#/wire/wire';
@ -65,7 +63,6 @@ function createAppendLogHarness(storage: IFileSystemStorageService): IAppendLogS
const ix = disposable.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, storage);
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
return ix.get(IAppendLogStore);
}

3
packages/agent-core-v2/test/wire/store-event.test.ts
View file

@ -7,8 +7,6 @@ import { TestInstantiationService } from '#/_base/di/test';
import { type DomainEvent, IEventBus } from '#/app/event/eventBus';
import { EventBusService } from '#/app/event/eventBusService';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -65,7 +63,6 @@ function setup(logKey: string): {
const ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.set(IEventBus, new SyncDescriptor(EventBusService));
const log = ix.get(IAppendLogStore);
const eventBus = ix.get(IEventBus);

4
packages/agent-core-v2/test/wire/wire-compat.test.ts
View file

@ -11,8 +11,6 @@ import { DisposableStore } from '#/_base/di/lifecycle';
import { TestInstantiationService } from '#/_base/di/test';
import { resetUnexpectedErrorHandler, setUnexpectedErrorHandler } from '#/_base/errors/unexpectedError';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -60,7 +58,6 @@ function makeContainer(storage: IFileSystemStorageService, logKey: string) {
const ix = store.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, storage);
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
const log = ix.get(IAppendLogStore);
const wire = registerTestAgentWire(ix, testWireScope(SCOPE, logKey), { log });
return { ix, wire, log };
@ -72,7 +69,6 @@ function makeReader(storage: IFileSystemStorageService): IAppendLogStore {
const ix = store.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, storage);
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
return ix.get(IAppendLogStore);
}

4
packages/agent-core-v2/test/wire/wireService.test.ts
View file

@ -8,8 +8,6 @@ import { resetUnexpectedErrorHandler, setUnexpectedErrorHandler } from '#/_base/
import { IEventBus } from '#/app/event/eventBus';
import { EventBusService } from '#/app/event/eventBusService';
import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore';
import { WriteAuthorityRegistryService } from '#/persistence/backends/node-fs/writeAuthorityRegistryService';
import { IWriteAuthorityRegistry } from '#/persistence/interface/writeAuthority';
import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
import { IAppendLogStore } from '#/persistence/interface/appendLogStore';
import { IFileSystemStorageService } from '#/persistence/interface/storage';
@ -89,7 +87,6 @@ beforeEach(() => {
ix = disposables.add(new TestInstantiationService());
ix.stub(IFileSystemStorageService, new InMemoryStorageService());
ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix.set(IEventBus, new SyncDescriptor(EventBusService));
log = ix.get(IAppendLogStore);
eventBus = ix.get(IEventBus);
@ -135,7 +132,6 @@ describe('WireService', () => {
const ix2 = disposables.add(new TestInstantiationService());
ix2.stub(IFileSystemStorageService, new InMemoryStorageService());
ix2.set(IAppendLogStore, new SyncDescriptor(AppendLogStore));
ix2.stub(IWriteAuthorityRegistry, new WriteAuthorityRegistryService());
ix2.set(IEventBus, new SyncDescriptor(EventBusService));
const log2 = ix2.get(IAppendLogStore);
const replayEventBus = ix2.get(IEventBus);

1
packages/klient/src/contract/session/lifecycle.ts
View file

@ -79,7 +79,6 @@ export const sessionLifecycleContract = {
create: { input: z.tuple([createSessionOptionsSchema]), output: handleWireSchema },
resume: { input: z.tuple([z.string()]), output: maybe(handleWireSchema) },
close: { input: z.tuple([z.string()]), output: noResult },
forceAbort: { input: z.tuple([z.string()]), output: noResult },
archive: { input: z.tuple([z.string()]), output: noResult },
restore: { input: z.tuple([z.string()]), output: maybe(handleWireSchema) },
fork: { input: z.tuple([forkSessionOptionsSchema]), output: handleWireSchema },

5
packages/klient/src/core/facade/session.ts
View file

@ -66,9 +66,8 @@ export interface SessionFacade {
update(patch: SessionMetaPatch): Promise<void>;
setArchived(archived: boolean): Promise<void>;
status(): Promise<SessionStatus>;
/** Release the session; failures abandon internally before rejecting. */
close(): Promise<void>;
/** Explicitly release a session lease after an ambiguous close flush failure. */
forceAbort(): Promise<void>;
archive(): Promise<void>;
/** Re-materialize a closed session; `false` when it no longer exists. */
restore(): Promise<boolean>;
@ -130,8 +129,6 @@ export function createSessionFacade(call: ScopedCaller, sessionId: string): Sess
return 'idle';
},
close: () => call({}, 'sessionLifecycleService', 'close', [sessionId]) as Promise<void>,
forceAbort: () =>
call({}, 'sessionLifecycleService', 'forceAbort', [sessionId]) as Promise<void>,
archive: () => call({}, 'sessionLifecycleService', 'archive', [sessionId]) as Promise<void>,
restore: async () => {
const handle = (await call({}, 'sessionLifecycleService', 'restore', [