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

refactor: drop multi-server flag and dedupe session ownership classification

Browse source
This commit is contained in:
7Sageer 2026-07-21 12:25:52 +08:00
parent 3d52ba575c
commit 2762bfc909
49 changed files with 579 additions and 914 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.changeset/cross-process-lock-unification.md
View file

@ -2,4 +2,4 @@
"@moonshot-ai/kimi-code": patch
---
Fix cross-process state corruption by replacing ad-hoc lockfiles with kernel-backed locks for session, server, and database coordination, with automatic release when a process exits.
Fix cross-process state corruption by replacing ad-hoc lockfiles with kernel-backed locks for server and database coordination, with automatic release when a process exits.

2
.changeset/multi-instance-session-sync.md
View file

@ -2,4 +2,4 @@
"@moonshot-ai/kimi-code": patch
---
web: When several server instances share one home, opening a session held by another instance now redirects to that instance, and the session list refreshes automatically when a peer adds or removes sessions. CLI/SDK clients follow the same redirect transparently.
web: When several server instances share one home, opening a session held by another instance now redirects to that instance, and the session list refreshes automatically when a peer adds or removes sessions.

2
.changeset/stale-write-fencing.md
View file

@ -2,4 +2,4 @@
"@moonshot-ai/kimi-code": patch
---
Track files read and written per session to detect conflicting edits across server instances: with multi-server mode enabled, stale or never-read file writes are rejected, otherwise they are flagged in the tool result.
Track files read and written per session to detect conflicting edits: a Write/Edit over a file that changed on disk since it was last read (or was never read in this session) is now rejected with a read-first reason, so conflicting edits across server instances cannot be applied silently.

2
apps/kimi-web/src/api/daemon/sessionOwnership.ts
View file

@ -9,7 +9,7 @@
// Wire semantics (from the protocol schema):
// - creating lease file observed mid-creation; retry shortly
// - routable holder is live and registered an address; redirect
// - holder-unresponsive holder pid alive but heartbeat stale; retry later
// - holder-unresponsive legacy heartbeat-based server response; retry later
// - held-by-local-instance holder has no address (local/embedded); terminal
// - unregistered-writer session dir written by an unregistered process

3
apps/kimi-web/src/api/daemon/wire.ts
View file

@ -583,7 +583,8 @@ export interface WireInFlightTurn {
/** `GET /sessions/{sid}/snapshot` — atomic rebuild state at a watermark. */
export interface WireSessionSnapshot {
as_of_seq: number;
epoch: string;
/** Absent until the journal's first durable event: "no baseline" ≠ "baseline changed". */
epoch?: string;
session: WireSession;
messages: { items: WireMessage[]; has_more: boolean };
in_flight_turn: WireInFlightTurn | null;

9
apps/kimi-web/src/api/daemon/ws.ts
View file

@ -390,19 +390,18 @@ export class DaemonEventSocket {
// Multi-instance hint (volatile, no payload): some instance sharing this
// home created/archived/deleted a session. Consumed by name here because
// classifyFrame would route the unknown unprefixed type to the agent
// projector, which no-ops on it. Emitted with or without the "event."
// prefix — accept both.
// projector, which no-ops on it. The server (sessionEventBroadcaster)
// only ever emits the bare form.
case 'session.list_changed':
case 'event.session.list_changed':
this.handlers.onSessionListChanged?.();
break;
// Volatile per-session hint: the daemon's skill catalog for this session
// changed (e.g. a skill file edited on disk). Consumed by name here for
// the same reason as session.list_changed above; the frame carries its
// own per-connection seq, never the durable watermark.
// own per-connection seq, never the durable watermark. The server
// (skillCatalogBridge) only ever emits the bare form.
case 'skill_catalog.changed':
case 'event.skill_catalog.changed':
this.handlers.onSkillCatalogChanged?.(frame.session_id as string);
break;

16
apps/kimi-web/test/ws-lifecycle.test.ts
View file

@ -159,7 +159,7 @@ describe('DaemonEventSocket frame dispatch (multi-instance surface)', () => {
vi.useRealTimers();
});
it('consumes session.list_changed (bare and event.-prefixed) via onSessionListChanged only', () => {
it('consumes bare session.list_changed via onSessionListChanged; the event.-prefixed form is not special-cased', () => {
let wireEvents = 0;
let listChanged = 0;
const handlers: DaemonEventSocketHandlers = {
@ -179,13 +179,15 @@ describe('DaemonEventSocket frame dispatch (multi-instance surface)', () => {
ws.emitMessage(SERVER_HELLO);
ws.emitMessage({ type: 'session.list_changed', payload: {} });
// The server never emits the prefixed form; it must NOT reach
// onSessionListChanged (it falls through to the generic protocol path).
ws.emitMessage({ type: 'event.session.list_changed', payload: {} });
expect(listChanged).toBe(2);
expect(wireEvents).toBe(0);
expect(listChanged).toBe(1);
expect(wireEvents).toBe(1);
});
it('consumes skill_catalog.changed (bare and event.-prefixed) via onSkillCatalogChanged only', () => {
it('consumes bare skill_catalog.changed via onSkillCatalogChanged; the event.-prefixed form is not special-cased', () => {
let wireEvents = 0;
let rawAgentEvents = 0;
const changed: string[] = [];
@ -216,6 +218,8 @@ describe('DaemonEventSocket frame dispatch (multi-instance surface)', () => {
volatile: true,
payload: { type: 'skill_catalog.changed', sourceId: 'workspace-file' },
});
// The server never emits the prefixed form; it must NOT reach
// onSkillCatalogChanged (it falls through to the generic protocol path).
ws.emitMessage({
type: 'event.skill_catalog.changed',
seq: 2,
@ -225,8 +229,8 @@ describe('DaemonEventSocket frame dispatch (multi-instance surface)', () => {
payload: { type: 'skill_catalog.changed', sourceId: 'plugin' },
});
expect(changed).toEqual(['sess_1', 'sess_2']);
expect(wireEvents).toBe(0);
expect(changed).toEqual(['sess_1']);
expect(wireEvents).toBe(1);
expect(rawAgentEvents).toBe(0);
});

5
packages/agent-core-v2/scripts/check-domain-layers.mjs
View file

@ -145,7 +145,6 @@ const DOMAIN_LAYER = new Map([
['permissionGate', 3],
['toolApproval', 3],
['flag', 3],
['multiServer', 3],
['toolExecutor', 3],
['toolResultTruncation', 3],
['toolRegistry', 3],
@ -175,8 +174,8 @@ const DOMAIN_LAYER = new Map([
['runtime', 4],
['toolDedupe', 4],
// `fileFencing` is the Agent-scope optimistic-concurrency gate: a
// tool-executor hook participant (L3) over the `sessionFileLedger` (L2)
// and flag state (L3), so it sits at L4 beside `toolDedupe`.
// tool-executor hook participant (L3) over the `sessionFileLedger` (L2),
// so it sits at L4 beside `toolDedupe`.
['fileFencing', 4],
['toolSelect', 4],
['toolPolicy', 4],

49
packages/agent-core-v2/src/agent/fileFencing/fileFencingService.ts
View file

@ -11,16 +11,13 @@
* — the exact canonical path the tool itself computed — so the ledger and
* the watcher key it identically. The before-hook records the target keyed
* by `toolCall.id` (cleared in the did-hook and swept on turn change) and,
* for `Write`/`Edit`, computes the ledger verdict: with the `multi_server`
* flag on, `stale` blocks with an outside-modification conflict and
* `no-baseline` blocks with a read-first reason (Edit-over-existing, or
* Write over an already existing file); with the flag off nothing ever
* blocks and the verdict is marked for the did-hook. The did-hook records the
* revision captured by the successful fenced call (ranged Reads excepted —
* per the ledger contract they never count as full reads) and,
* for a flag-off stale mark, composes a `<system>` advisory onto the result
* note; direct creation of a new file is verdict-`clean`, so it is never
* advisory'd. Watcher echos of the session's own writes are absorbed by the
* for `Write`/`Edit`, computes the ledger verdict: `stale` blocks with an
* outside-modification conflict and `no-baseline` blocks with a read-first
* reason (Edit-over-existing, or Write over an already existing file). The
* did-hook records the revision captured by the successful fenced call
* (ranged Reads excepted — per the ledger contract they never count as full
* reads); direct creation of a new file is verdict-`clean`, so it never
* blocks. Watcher echos of the session's own writes are absorbed by the
* ledger's stat punch, so consecutive Edits stay clean. Checked after
* `permission` (ignition order is set by `agentLifecycle`). Bound at Agent
* scope.
@ -35,8 +32,6 @@ import type {
ToolDidExecuteContext,
ToolExecutionHookContext,
} from '#/agent/toolExecutor/toolHooks';
import { IFlagService } from '#/app/flag/flag';
import { MULTI_SERVER_FLAG_ID } from '#/app/multiServer/flag';
import {
ISessionFileLedger,
type FileLedgerVerdict,
@ -92,20 +87,6 @@ function blockReason(toolName: string, path: string, verdict: FileLedgerVerdict)
);
}
function advisoryNote(target: FencingTarget, verdict: FileLedgerVerdict): string {
const body =
verdict === 'no-baseline'
? `"${target.path}" already existed on disk and had not been read in this session; your change overwrote it anyway.`
: `"${target.path}" changed on disk since it was last read in this session; your change was applied anyway.`;
return `<system>Warning: ${body} Read the file to verify the current content.</system>`;
}
function composeNote(existing: string | undefined, advisory: string): string {
return existing === undefined || existing.length === 0
? advisory
: `${existing}\n${advisory}`;
}
function revisionForTarget(
revision: ToolFileRevision | undefined,
targetPath: string,
@ -117,12 +98,10 @@ export class AgentFileFencingService extends Disposable implements IAgentFileFen
declare readonly _serviceBrand: undefined;
private readonly targets = new Map<string, FencingTarget>();
private readonly staleMarks = new Map<string, FileLedgerVerdict>();
private markerTurnId: number | undefined;
constructor(
@ISessionFileLedger private readonly ledger: ISessionFileLedger,
@IFlagService private readonly flags: IFlagService,
@IAgentToolExecutorService toolExecutor: IAgentToolExecutorService,
) {
super();
@ -144,7 +123,6 @@ export class AgentFileFencingService extends Disposable implements IAgentFileFen
if (this.markerTurnId !== ctx.turnId) {
this.markerTurnId = ctx.turnId;
this.targets.clear();
this.staleMarks.clear();
}
this.targets.set(ctx.toolCall.id, { toolName: ctx.toolCall.name, path });
if (!WRITE_TOOLS.has(ctx.toolCall.name)) return;
@ -154,12 +132,8 @@ export class AgentFileFencingService extends Disposable implements IAgentFileFen
execute: async (executeCtx) => {
const verdict = await this.ledger.compare(path);
if (verdict !== 'clean') {
if (this.flags.enabled(MULTI_SERVER_FLAG_ID)) {
const reason = blockReason(ctx.toolCall.name, path, verdict);
ctx.decision = { ...ctx.decision, block: true, reason };
return { output: reason, isError: true };
}
this.staleMarks.set(ctx.toolCall.id, verdict);
const reason = blockReason(ctx.toolCall.name, path, verdict);
return { output: reason, isError: true };
}
return execute(executeCtx);
},
@ -170,8 +144,6 @@ export class AgentFileFencingService extends Disposable implements IAgentFileFen
if (!isFenced(ctx)) return;
const target = this.targets.get(ctx.toolCall.id);
this.targets.delete(ctx.toolCall.id);
const mark = this.staleMarks.get(ctx.toolCall.id);
this.staleMarks.delete(ctx.toolCall.id);
if (target === undefined || ctx.result.isError === true) return;
if (target.toolName === READ_TOOL && isRangedRead(ctx.args)) return;
const revision = revisionForTarget(ctx.result[toolFileRevision], target.path);
@ -183,9 +155,6 @@ export class AgentFileFencingService extends Disposable implements IAgentFileFen
size: revision.size,
});
}
if (mark !== undefined) {
ctx.result = { ...ctx.result, note: composeNote(ctx.result.note, advisoryNote(target, mark)) };
}
}
}

10
packages/agent-core-v2/src/app/cron/clock.ts
View file

@ -10,11 +10,11 @@
*
* 2. monotonic ms — a strictly non-decreasing counter that never
* jumps backwards across NTP adjustments, suspend/resume, or
* simulated-clock injection. Used for the poll cadence and the
* lock heartbeat — anything where "did 5 seconds elapse since we
* last looked" must hold even when the wall clock is frozen.
* simulated-clock injection. Used for the poll cadence — anything
* where "did 5 seconds elapse since we last looked" must hold
* even when the wall clock is frozen.
*
* Mixing the two pollutes test reproducibility: a heartbeat tied to
* Mixing the two pollutes test reproducibility: a poll cadence tied to
* `wallNow()` will appear stuck when the test clock is frozen; a cron
* fire tied to `monoNowMs()` will not advance when the bench rewinds
* the simulated day. Every component in the cron domain MUST take a
@ -22,7 +22,7 @@
*
* `monoNowMs` is ALWAYS `process.hrtime.bigint()` (converted to ms).
* It is not overridable — accepting an external monotonic clock would
* defeat the safety net the lock heartbeat depends on.
* let a frozen test clock silently stall the poll cadence.
*
* `wallNow` resolution is driven by the `KIMI_CRON_CLOCK` env var; see
* `resolveClockSources` below. Defaults to `Date.now()`.

29
packages/agent-core-v2/src/app/multiServer/flag.ts
View file

@ -1,29 +0,0 @@
/**
* `multi_server` experimental flag — gates shared-home coordination among
* cooperating lease-aware server versions.
*
* When enabled, a kap-server instance registers itself under
* `<home>/server/instances/<serverId>.json` instead of taking the legacy
* single-instance `<home>/server/lock`, so multiple servers can share one home
* directory. Off by default; enable via `KIMI_CODE_EXPERIMENTAL_MULTI_SERVER`,
* the master `KIMI_CODE_EXPERIMENTAL_FLAG`, or the `[experimental]` config
* section. Imported for its side effect (registers the definition) from the
* package barrel.
*/
import { type FlagDefinitionInput, registerFlagDefinition } from '#/app/flag/flagRegistry';
export const MULTI_SERVER_FLAG_ID = 'multi_server';
export const MULTI_SERVER_FLAG_ENV = 'KIMI_CODE_EXPERIMENTAL_MULTI_SERVER';
export const multiServerFlag: FlagDefinitionInput = {
id: MULTI_SERVER_FLAG_ID,
title: 'multi-server shared home',
description:
'Allow cooperating lease-aware kap-server instances to share one home directory by registering each instance under server/instances/ instead of taking a single homedir lock.',
env: MULTI_SERVER_FLAG_ENV,
default: false,
surface: 'core',
};
registerFlagDefinition(multiServerFlag);

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

@ -69,8 +69,6 @@ import { CRON_SESSION_TAG, type CronTask } from '#/app/cron/cronTask';
import { ICronTaskPersistence } from '#/app/cron/cronTaskPersistence';
import { IConfigService } from '#/app/config/config';
import { IEventService } from '#/app/event/event';
import { IFlagService } from '#/app/flag/flag';
import { MULTI_SERVER_FLAG_ID } from '#/app/multiServer/flag';
import {
CHILD_SESSION_KIND,
CHILD_SESSION_KIND_KEY,
@ -102,6 +100,7 @@ import { ISessionContext, sessionContextSeed } from '#/session/sessionContext/se
import { ISessionCronService } from '#/session/cron/sessionCronService';
import {
type HeldByPeerDetails,
heldByPeerDetailsFromInspection,
LEASE_CREATING_RETRY_AFTER_MS,
SessionLease,
sessionLeasePath,
@ -189,7 +188,6 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
@IEventService private readonly event: IEventService,
@ITelemetryService private readonly telemetry: ITelemetryService,
@ILogService private readonly log: ILogService,
@IFlagService private readonly flags: IFlagService,
@ICrossProcessLockService private readonly locks: ICrossProcessLockService,
@IWriteAuthorityRegistry private readonly authorityRegistry: IWriteAuthorityRegistry,
@ISessionLeaseContactProvider
@ -925,16 +923,15 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
}
private heldByPeerDetails(inspection: CrossProcessLockInspection): HeldByPeerDetails {
if (inspection.state === 'held' && inspection.payload !== undefined) {
const payload = inspection.payload;
if (payload.address !== undefined && this.flags.enabled(MULTI_SERVER_FLAG_ID)) {
return { kind: 'held-by-peer', phase: 'routable', address: payload.address };
// A free lease here means the holder vanished between the failed acquire
// and this probe; that race converges by retrying, same as 'creating'.
return (
heldByPeerDetailsFromInspection(inspection) ?? {
kind: 'held-by-peer',
phase: 'creating',
retry_after_ms: LEASE_CREATING_RETRY_AFTER_MS,
}
return { kind: 'held-by-peer', phase: 'held-by-local-instance' };
}
// 'creating' mid-creation, and races where the holder vanished between the
// failed acquire and this probe, both converge by retrying shortly.
return { kind: 'held-by-peer', phase: 'creating', retry_after_ms: LEASE_CREATING_RETRY_AFTER_MS };
);
}
private async flushSessionTail(sessionId: string, scope: string): Promise<void> {

7
packages/agent-core-v2/src/os/interface/crossProcessLock.ts
View file

@ -78,7 +78,6 @@ export const OsLockErrors = {
codes: {
OS_LOCK_HELD: 'os.lock.held',
OS_LOCK_WAIT_TIMEOUT: 'os.lock.wait_timeout',
OS_LOCK_LOST: 'os.lock.lost',
OS_LOCK_IO: 'os.lock.io',
},
info: {
@ -92,11 +91,6 @@ export const OsLockErrors = {
retryable: true,
public: true,
},
'os.lock.lost': {
title: 'Lock ownership was lost',
retryable: false,
public: true,
},
'os.lock.io': {
title: 'Lock file I/O failed',
retryable: true,
@ -110,7 +104,6 @@ registerErrorDomain(OsLockErrors);
export const CrossProcessLockErrorCode = {
Held: OsLockErrors.codes.OS_LOCK_HELD,
WaitTimeout: OsLockErrors.codes.OS_LOCK_WAIT_TIMEOUT,
Lost: OsLockErrors.codes.OS_LOCK_LOST,
Io: OsLockErrors.codes.OS_LOCK_IO,
} as const;

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

@ -20,7 +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.
* 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.
*/
import { createReadStream, mkdirSync, statSync } from 'node:fs';
@ -33,6 +35,7 @@ import { optional } from '#/_base/di/instantiation';
import { Emitter, type Event } from '#/_base/event';
import { onUnexpectedError } from '#/_base/errors/unexpectedError';
import { atomicWrite, syncDir } from '#/_base/utils/fs';
import { Error2, ErrorCodes } from '#/errors';
import {
CrossProcessLockError,
CrossProcessLockErrorCode,
@ -46,6 +49,10 @@ import type {
StorageWriteOptions,
} from '#/persistence/interface/storage';
import { StorageError, StorageErrors, toStorageIoError } from '#/persistence/interface/storage';
import {
sessionIdFromScope,
IWriteAuthorityRegistry,
} from '#/persistence/interface/writeAuthority';
const WATCH_DEBOUNCE_MS = 150;
const STORAGE_LOCK_WAIT_TIMEOUT_MS = 10_000;
@ -83,6 +90,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,
) {}
async read(scope: string, key: string): Promise<Uint8Array | undefined> {
@ -122,8 +131,14 @@ 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) {
@ -139,9 +154,14 @@ 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);
try {
if (data.byteLength > 0) {
@ -172,6 +192,7 @@ 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) {
@ -256,6 +277,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);
if (this.locks === undefined) {
throw new StorageError(
StorageErrors.codes.STORAGE_IO_FAILED,
@ -267,7 +289,10 @@ export class FileStorageService implements IFileSystemStorageService {
return await this.locks.withLock(
lockPath,
{ wait: { timeoutMs: STORAGE_LOCK_WAIT_TIMEOUT_MS } },
op,
async () => {
this.assertScopeWritable(scope);
return op();
},
);
} catch (error) {
if (!(error instanceof CrossProcessLockError)) throw error;
@ -297,6 +322,18 @@ export class FileStorageService implements IFileSystemStorageService {
return join(this.baseDir, scope);
}
private assertScopeWritable(scope: string): void {
const sessionId = sessionIdFromScope(scope);
if (sessionId === undefined || this.authorityRegistry === undefined) return;
const authority = this.authorityRegistry.resolve(sessionId);
if (authority === undefined) {
throw new Error2(ErrorCodes.SESSION_LEASE_LOST, 'session has no registered write authority', {
details: { sessionId },
});
}
authority.assertWritable();
}
private async syncDirOnce(dir: string): Promise<void> {
if (this.syncedDirs.has(dir)) return;
try {

2
packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts
View file

@ -346,7 +346,7 @@ export class AgentLifecycleService extends Disposable implements IAgentLifecycle
this.handles.delete(agentId);
const tasks = handle.accessor.get(IAgentTaskService);
await tasks.stopAllOnExit('Session closed');
await tasks.flushPersistence?.();
await tasks.flushPersistence();
const loop = handle.accessor.get(IAgentLoopService);
const compaction = handle.accessor.get(IAgentFullCompactionService).compacting;
const compactionSettled = compaction?.promise.catch(() => undefined) ?? Promise.resolve();

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

@ -25,7 +25,10 @@ import { join } from 'pathe';
import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation';
import type { ScopeSeed } from '#/_base/di/scope';
import { Error2, ErrorCodes } from '#/errors';
import type { ICrossProcessLockHandle } from '#/os/interface/crossProcessLock';
import type {
CrossProcessLockInspection,
ICrossProcessLockHandle,
} from '#/os/interface/crossProcessLock';
import type { ISessionWriteAuthority } from '#/persistence/interface/writeAuthority';
export const LEASE_CREATING_RETRY_AFTER_MS = 1000;
@ -49,6 +52,33 @@ export type HeldByPeerDetails = {
export type SessionOwnershipDetails = HeldByPeerDetails | { readonly kind: 'unregistered-writer' };
/**
* Classify a lease inspection into `held-by-peer` details. Shared by every
* surface that reports session ownership — the lifecycle's
* post-acquire-failure probe and kap-server's read-only probes — so all of
* them classify the same lease the same way. Returns `undefined` when the
* lease is free; a caller on a failed-acquire path should map that to
* `'creating'`, since a holder that vanished mid-race converges by retrying.
*/
export function heldByPeerDetailsFromInspection(
inspection: CrossProcessLockInspection,
): HeldByPeerDetails | undefined {
if (inspection.state === 'held' && inspection.payload !== undefined) {
const { address } = inspection.payload;
return address !== undefined
? { kind: 'held-by-peer', phase: 'routable', address }
: { kind: 'held-by-peer', phase: 'held-by-local-instance' };
}
if (inspection.state === 'creating') {
return {
kind: 'held-by-peer',
phase: 'creating',
retry_after_ms: LEASE_CREATING_RETRY_AFTER_MS,
};
}
return undefined;
}
export interface ISessionLeaseInfo {
readonly sessionId: string;
readonly lockId: string;
@ -83,15 +113,11 @@ export class SessionLease implements ISessionWriteAuthority, ISessionLeaseServic
this.lockId = handle.lockId;
}
get released(): boolean {
return this._released;
}
get info(): ISessionLeaseInfo | undefined {
return this._released ? undefined : { sessionId: this.sessionId, lockId: this.lockId };
}
checkHeld(): boolean {
private checkHeld(): boolean {
return !this._released && this.handle.checkHeld();
}
@ -103,7 +129,7 @@ export class SessionLease implements ISessionWriteAuthority, ISessionLeaseServic
{ details: { sessionId: this.sessionId } },
);
}
if (this._lost || !this.handle.checkHeld()) {
if (this._lost || !this.checkHeld()) {
this.markLost();
throw new Error2(
ErrorCodes.SESSION_LEASE_LOST,
@ -113,7 +139,7 @@ export class SessionLease implements ISessionWriteAuthority, ISessionLeaseServic
}
}
markLost(): void {
private markLost(): void {
this._lost = true;
if (this._lossFired) return;
this._lossFired = true;

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

@ -15,7 +15,7 @@
* 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`,
* synchronously re-reading the lease payload), so an instance that lost the
* 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.
*

620
packages/agent-core-v2/test/agent/fileFencing/fileFencing.test.ts
View file

@ -2,8 +2,8 @@
* `fileFencing` domain (L4) — verifies the write/read-first gate end to end
* through the real DI scope tree: a real tmpdir, the real `HostFileSystem`,
* the real watch service folding fake os-watcher events, and the registered
* `writeFencing` hook participant over a real `OrderedHookSlot`. Covers both
* flag postures (`multi_server` on = hard block, off = advisory note), the
* `writeFencing` hook participant over a real `OrderedHookSlot`. Covers the
* hard-block verdicts (stale outside change / never-read existing file), the
* own-write echo / truncated-window stale checks, out-of-root stat fallback,
* watched-root ensuring for additional dirs, and ledger isolation between
* two Session scopes sharing one workspace (two-instance conflict).
@ -46,6 +46,7 @@ import {
import { AgentToolExecutorService } from '#/agent/toolExecutor/toolExecutorService';
import { stubToolExecutor } from '../loop/stubs';
import { stubFlag } from '../../app/flag/stubs';
import { fakeHostFsWatch, type FakeWatch } from '../../session/sessionFs/stubs';
void AgentFileFencingService;
@ -54,8 +55,6 @@ void SessionFsWatchService;
void SessionWorkspaceContextService;
void AgentToolExecutorService;
let multiServer = false;
function countingHostFs(): { fs: IHostFileSystem; statCalls: () => number } {
const real = new HostFileSystem();
let count = 0;
@ -90,10 +89,7 @@ function makeEnv(): Env {
const host = createScopedTestHost([
stubPair(IHostFileSystem, fs),
stubPair(IHostFsWatchService, fake.service),
stubPair(IFlagService, {
_serviceBrand: undefined,
enabled: () => multiServer,
} as unknown as IFlagService),
stubPair(IFlagService, stubFlag(false)),
]);
hosts.push(host);
return { host, fake, workDir, outsideDir, statCalls };
@ -190,9 +186,11 @@ async function runBefore(
return ctx;
}
async function runPrepared(ctx: ToolBeforeExecuteContext): Promise<void> {
if (ctx.decision?.execute === undefined) return;
await ctx.decision.execute({
async function runPrepared(
ctx: ToolBeforeExecuteContext,
): Promise<ExecutableToolResult | undefined> {
if (ctx.decision?.execute === undefined) return undefined;
return ctx.decision.execute({
turnId: ctx.turnId,
toolCallId: ctx.toolCall.id,
trace: ctx.trace,
@ -242,8 +240,10 @@ async function runOk(
path: string,
opts: { id?: string; turnId?: number; args?: Record<string, unknown> } = {},
): Promise<ToolDidExecuteContext> {
const ctx = await runBefore(world, beforeCtx(toolName, path, opts));
expect(ctx.decision?.block).not.toBe(true);
const ctx = beforeCtx(toolName, path, opts);
await world.executor.hooks.onBeforeExecuteTool.run(ctx);
const prepared = await runPrepared(ctx);
expect(prepared?.isError).not.toBe(true);
if (toolName === 'Write') {
const args = ctx.args as { readonly content: string; readonly mode?: 'overwrite' | 'append' };
writeFileSync(path, args.content, { flag: args.mode === 'append' ? 'a' : 'w' });
@ -251,6 +251,20 @@ async function runOk(
return runDid(world, ctx);
}
async function runBlocked(
world: AgentWorld,
toolName: string,
path: string,
): Promise<ExecutableToolResult> {
const ctx = beforeCtx(toolName, path);
await world.executor.hooks.onBeforeExecuteTool.run(ctx);
const result = await runPrepared(ctx);
if (result?.isError !== true) {
throw new Error(`expected ${toolName} on ${path} to be blocked, got: ${JSON.stringify(result)}`);
}
return result;
}
function foldChange(world: AgentWorld, rel: string, action: 'created' | 'modified' | 'deleted'): void {
world.env.fake.fire(rel, action);
vi.advanceTimersByTime(200);
@ -266,7 +280,6 @@ const cleanupPaths: string[] = [];
describe('AgentFileFencingService', () => {
beforeEach(() => {
multiServer = false;
vi.useFakeTimers();
});
afterEach(() => {
@ -275,360 +288,277 @@ describe('AgentFileFencingService', () => {
vi.useRealTimers();
});
describe('with the multi_server flag on', () => {
it('blocks Edit on an existing file that was never read, read-first', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
it('blocks Edit on an existing file that was never read, read-first', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
const ctx = await runBefore(world, beforeCtx('Edit', file));
expect(ctx.decision?.block).toBe(true);
expect(ctx.decision?.reason).toContain('has not been read in this session');
expect(ctx.decision?.reason).toContain('Read it first');
});
it('blocks Edit when the file changed on disk since the last read, and unblocks after re-read', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
writeFileSync(file, 'hello world');
foldChange(world, 'a.txt', 'modified');
const ctx = await runBefore(world, beforeCtx('Edit', file));
expect(ctx.decision?.block).toBe(true);
expect(ctx.decision?.reason).toContain('changed on disk since');
await runOk(world, 'Read', file);
const retry = await runBefore(world, beforeCtx('Edit', file));
expect(retry.decision?.block).not.toBe(true);
});
it('checks the file at execution time rather than during preflight', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
const ctx = beforeCtx('Edit', file);
await world.executor.hooks.onBeforeExecuteTool.run(ctx);
writeFileSync(file, 'changed while queued');
await runPrepared(ctx);
expect(ctx.decision?.block).toBe(true);
expect(ctx.decision?.reason).toContain('changed on disk since');
});
it('keeps the revision captured by Read when the file changes before the did-hook', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
const ctx = await runBefore(world, beforeCtx('Read', file));
const readRevision = makeToolFileRevision(file, statSync(file));
writeFileSync(file, 'changed after read');
await runDid(world, ctx, {
output: 'hello',
[toolFileRevision]: readRevision,
});
const edit = await runBefore(world, beforeCtx('Edit', file));
expect(edit.decision?.block).toBe(true);
expect(edit.decision?.reason).toContain('changed on disk since');
});
it('wraps an execution override installed by an earlier hook', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
let overrideCalls = 0;
const ctx = beforeCtx('Edit', file);
ctx.decision = {
execute: async () => {
overrideCalls++;
return { output: 'overridden' };
},
};
await world.executor.hooks.onBeforeExecuteTool.run(ctx);
await runPrepared(ctx);
expect(overrideCalls).toBe(1);
});
it('blocks Write over an existing file that was never read', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
const ctx = await runBefore(world, beforeCtx('Write', file));
expect(ctx.decision?.block).toBe(true);
expect(ctx.decision?.reason).toContain('already exists');
expect(ctx.decision?.reason).toContain('has not been read in this session');
});
it('allows Write creating a new file and baselines it', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'new.txt');
await runOk(world, 'Write', file);
const ctx = await runBefore(world, beforeCtx('Edit', file));
expect(ctx.decision?.block).not.toBe(true);
});
it('allows Edit right after a full Read', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
const ctx = await runBefore(world, beforeCtx('Edit', file));
expect(ctx.decision?.block).not.toBe(true);
});
it('allows consecutive Edits without watcher events', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
await runOk(world, 'Edit', file);
const ctx = await runBefore(world, beforeCtx('Edit', file));
expect(ctx.decision?.block).not.toBe(true);
});
it('keeps consecutive Edits clean through the own-write watcher echo and re-baselines', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
expect(world.env.statCalls()).toBe(0);
foldChange(world, 'a.txt', 'modified');
const ctx = await runBefore(world, beforeCtx('Edit', file));
expect(ctx.decision?.block).not.toBe(true);
expect(world.env.statCalls()).toBe(1);
const again = await runBefore(world, beforeCtx('Edit', file));
expect(again.decision?.block).not.toBe(true);
expect(world.env.statCalls()).toBe(2);
});
it('resolves a truncated window by stat punch: unchanged passes, changed blocks', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
foldJunk(world.env);
const unchanged = await runBefore(world, beforeCtx('Edit', file));
expect(unchanged.decision?.block).not.toBe(true);
writeFileSync(file, 'hello world');
foldJunk(world.env);
const changed = await runBefore(world, beforeCtx('Edit', file));
expect(changed.decision?.block).toBe(true);
expect(changed.decision?.reason).toContain('changed on disk since');
});
it('blocks ranged-Read followed by Edit because ranged reads never baseline', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, '1\n2\n3\n4\n5\n6\n');
await runOk(world, 'Read', file, { args: { path: file, line_offset: 5 } });
const ctx = await runBefore(world, beforeCtx('Edit', file));
expect(ctx.decision?.block).toBe(true);
expect(ctx.decision?.reason).toContain('has not been read in this session');
});
it('blocks out-of-root writes through the stat-only fallback and allows them after Read', async () => {
multiServer = true;
const world = setup();
const file = join(world.env.outsideDir, 'b.txt');
writeFileSync(file, 'hello');
const first = await runBefore(world, beforeCtx('Edit', file));
expect(first.decision?.block).toBe(true);
expect(first.decision?.reason).toContain('has not been read in this session');
await runOk(world, 'Read', file);
const afterRead = await runBefore(world, beforeCtx('Edit', file));
expect(afterRead.decision?.block).not.toBe(true);
writeFileSync(file, 'hello world');
const changed = await runBefore(world, beforeCtx('Edit', file));
expect(changed.decision?.block).toBe(true);
expect(changed.decision?.reason).toContain('changed on disk since');
});
it('ensures an additional dir becomes watched when a write target falls under it', async () => {
multiServer = true;
const world = setup();
world.workspace.addAdditionalDir(world.env.outsideDir);
const file = join(world.env.outsideDir, 'new.txt');
await runOk(world, 'Write', file);
expect(world.watch.watchedRoots).toContain(world.env.outsideDir);
expect(world.env.fake.watchCalls).toContain(world.env.outsideDir);
writeFileSync(file, 'changed outside');
world.env.fake.handles
.find((h) => h.root === world.env.outsideDir)
?.fire('new.txt', 'modified');
vi.advanceTimersByTime(200);
const ctx = await runBefore(world, beforeCtx('Write', file));
expect(ctx.decision?.block).toBe(true);
expect(ctx.decision?.reason).toContain('changed on disk since');
});
it('keeps ledgers on two session scopes sharing one workspace independent and flags the peer change', async () => {
multiServer = true;
const env = makeEnv();
const worldA = makeAgent(env, makeSession(env, 'sA', env.workDir));
const worldB = makeAgent(env, makeSession(env, 'sB', env.workDir));
const file = join(env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(worldA, 'Read', file);
const neverRead = await runBefore(worldB, beforeCtx('Edit', file));
expect(neverRead.decision?.block).toBe(true);
expect(neverRead.decision?.reason).toContain('has not been read in this session');
writeFileSync(file, 'hello world');
env.fake.handles
.findLast((h) => h.root === env.workDir)
?.fire('a.txt', 'modified');
vi.advanceTimersByTime(200);
const conflict = await runBefore(worldB, beforeCtx('Edit', file));
expect(conflict.decision?.block).toBe(true);
expect(conflict.decision?.reason).toContain('changed on disk since');
});
const blocked = await runBlocked(world, 'Edit', file);
expect(blocked.output).toContain('has not been read in this session');
expect(blocked.output).toContain('Read it first');
});
describe('with the multi_server flag off', () => {
it('admits Edit on an unread existing file with an advisory, then re-baselines', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
it('blocks Edit when the file changed on disk since the last read, and unblocks after re-read', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
const did = await runOk(world, 'Edit', file);
expect(did.result.note).toContain('<system>Warning:');
expect(did.result.note).toContain('had not been read in this session');
writeFileSync(file, 'hello world');
foldChange(world, 'a.txt', 'modified');
const second = await runOk(world, 'Edit', file);
expect(second.result.note).toBeUndefined();
const blocked = await runBlocked(world, 'Edit', file);
expect(blocked.output).toContain('changed on disk since');
await runOk(world, 'Read', file);
await runOk(world, 'Edit', file);
});
it('checks the file at execution time rather than during preflight', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
const ctx = beforeCtx('Edit', file);
await world.executor.hooks.onBeforeExecuteTool.run(ctx);
writeFileSync(file, 'changed while queued');
const blocked = await runPrepared(ctx);
expect(blocked?.isError).toBe(true);
expect(blocked?.output).toContain('changed on disk since');
});
it('keeps the revision captured by Read when the file changes before the did-hook', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
const ctx = await runBefore(world, beforeCtx('Read', file));
const readRevision = makeToolFileRevision(file, statSync(file));
writeFileSync(file, 'changed after read');
await runDid(world, ctx, {
output: 'hello',
[toolFileRevision]: readRevision,
});
it('admits Write over an unread existing file with an advisory', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
const blocked = await runBlocked(world, 'Edit', file);
expect(blocked.output).toContain('changed on disk since');
});
const did = await runOk(world, 'Write', file);
expect(did.result.note).toContain('already existed on disk');
});
it('wraps an execution override installed by an earlier hook', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
let overrideCalls = 0;
const ctx = beforeCtx('Edit', file);
ctx.decision = {
execute: async () => {
overrideCalls++;
return { output: 'overridden' };
},
};
it('admits Edit after an outside change with the applied-anyway advisory', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
await world.executor.hooks.onBeforeExecuteTool.run(ctx);
await runPrepared(ctx);
writeFileSync(file, 'hello world');
foldChange(world, 'a.txt', 'modified');
expect(overrideCalls).toBe(1);
});
const did = await runOk(world, 'Edit', file);
expect(did.result.note).toContain('changed on disk since it was last read in this session');
expect(did.result.note).toContain('your change was applied anyway');
it('blocks Write over an existing file that was never read', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
const second = await runOk(world, 'Edit', file);
expect(second.result.note).toBeUndefined();
});
const blocked = await runBlocked(world, 'Write', file);
expect(blocked.output).toContain('already exists');
expect(blocked.output).toContain('has not been read in this session');
});
it('composes the advisory with an existing result note', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
it('allows Write creating a new file and baselines it', async () => {
const world = setup();
const file = join(world.env.workDir, 'new.txt');
const ctx = await runBefore(world, beforeCtx('Edit', file));
const did = await runDid(world, ctx, { output: 'done', note: '<system>existing</system>' });
expect(did.result.note).toBe(
'<system>existing</system>\n' +
`<system>Warning: "${file}" already existed on disk and had not been read in this session; ` +
'your change overwrote it anyway. Read the file to verify the current content.</system>',
);
});
await runOk(world, 'Write', file);
await runOk(world, 'Edit', file);
});
it('never advisories direct creation of a new file', async () => {
const world = setup();
const did = await runOk(world, 'Write', join(world.env.workDir, 'new.txt'));
expect(did.result.note).toBeUndefined();
});
it('allows Edit right after a full Read', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
it('adds no advisory and no baseline when the tool result is an error', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
await runOk(world, 'Read', file);
await runOk(world, 'Edit', file);
});
writeFileSync(file, 'hello world');
foldChange(world, 'a.txt', 'modified');
it('allows consecutive Edits without watcher events', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
const failed = await runBefore(world, beforeCtx('Edit', file));
const failedDid = await runDid(world, failed, { output: 'boom', isError: true });
expect(failedDid.result.note).toBeUndefined();
await runOk(world, 'Read', file);
await runOk(world, 'Edit', file);
await runOk(world, 'Edit', file);
});
const retry = await runOk(world, 'Edit', file);
expect(retry.result.note).toContain('changed on disk since it was last read in this session');
});
it('keeps consecutive Edits clean through the own-write watcher echo and re-baselines', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
expect(world.env.statCalls()).toBe(0);
it('does not leak a stale mark across turns', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
foldChange(world, 'a.txt', 'modified');
await runBefore(world, beforeCtx('Edit', file, { id: 'call-abandoned', turnId: 1 }));
await runBefore(
world,
beforeCtx('Edit', join(world.env.workDir, 'other.txt'), { turnId: 2 }),
);
await runOk(world, 'Edit', file);
expect(world.env.statCalls()).toBe(1);
const abandonedCtx = beforeCtx('Edit', file, { id: 'call-abandoned', turnId: 1 });
const did = await runDid(world, abandonedCtx);
expect(did.result.note).toBeUndefined();
});
await runOk(world, 'Edit', file);
expect(world.env.statCalls()).toBe(2);
});
it('ignores tools other than Read/Write/Edit entirely', async () => {
const world = setup();
const ctx = await runBefore(
world,
beforeCtx('Bash', join(world.env.workDir, 'a.txt'), { args: { command: 'ls' } }),
);
expect(ctx.decision).toBeUndefined();
it('resolves a truncated window by stat punch: unchanged passes, changed blocks', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
const did = await runDid(world, ctx);
expect(did.result.note).toBeUndefined();
expect(world.env.statCalls()).toBe(0);
});
foldJunk(world.env);
await runOk(world, 'Edit', file);
writeFileSync(file, 'hello world');
foldJunk(world.env);
const blocked = await runBlocked(world, 'Edit', file);
expect(blocked.output).toContain('changed on disk since');
});
it('blocks ranged-Read followed by Edit because ranged reads never baseline', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, '1\n2\n3\n4\n5\n6\n');
await runOk(world, 'Read', file, { args: { path: file, line_offset: 5 } });
const blocked = await runBlocked(world, 'Edit', file);
expect(blocked.output).toContain('has not been read in this session');
});
it('blocks out-of-root writes through the stat-only fallback and allows them after Read', async () => {
const world = setup();
const file = join(world.env.outsideDir, 'b.txt');
writeFileSync(file, 'hello');
const first = await runBlocked(world, 'Edit', file);
expect(first.output).toContain('has not been read in this session');
await runOk(world, 'Read', file);
await runOk(world, 'Edit', file);
writeFileSync(file, 'hello world');
const changed = await runBlocked(world, 'Edit', file);
expect(changed.output).toContain('changed on disk since');
});
it('ensures an additional dir becomes watched when a write target falls under it', async () => {
const world = setup();
world.workspace.addAdditionalDir(world.env.outsideDir);
const file = join(world.env.outsideDir, 'new.txt');
await runOk(world, 'Write', file);
expect(world.watch.watchedRoots).toContain(world.env.outsideDir);
expect(world.env.fake.watchCalls).toContain(world.env.outsideDir);
writeFileSync(file, 'changed outside');
world.env.fake.handles
.find((h) => h.root === world.env.outsideDir)
?.fire('new.txt', 'modified');
vi.advanceTimersByTime(200);
const blocked = await runBlocked(world, 'Write', file);
expect(blocked.output).toContain('changed on disk since');
});
it('keeps ledgers on two session scopes sharing one workspace independent and flags the peer change', async () => {
const env = makeEnv();
const worldA = makeAgent(env, makeSession(env, 'sA', env.workDir));
const worldB = makeAgent(env, makeSession(env, 'sB', env.workDir));
const file = join(env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(worldA, 'Read', file);
const neverRead = await runBlocked(worldB, 'Edit', file);
expect(neverRead.output).toContain('has not been read in this session');
writeFileSync(file, 'hello world');
env.fake.handles
.findLast((h) => h.root === env.workDir)
?.fire('a.txt', 'modified');
vi.advanceTimersByTime(200);
const conflict = await runBlocked(worldB, 'Edit', file);
expect(conflict.output).toContain('changed on disk since');
});
it('leaves direct creation of a new file without a result note', async () => {
const world = setup();
const did = await runOk(world, 'Write', join(world.env.workDir, 'new.txt'));
expect(did.result.note).toBeUndefined();
});
it('records no baseline and stays blocked when the fenced call fails', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runOk(world, 'Read', file);
writeFileSync(file, 'hello world');
foldChange(world, 'a.txt', 'modified');
// The stale verdict blocks the Edit; the wrapper's error result must not
// be baselined by the did-hook, so the file stays stale until re-read.
const failed = beforeCtx('Edit', file);
await world.executor.hooks.onBeforeExecuteTool.run(failed);
const failedResult = await runPrepared(failed);
expect(failedResult?.isError).toBe(true);
await runDid(world, failed, { output: 'boom', isError: true });
const retry = await runBlocked(world, 'Edit', file);
expect(retry.output).toContain('changed on disk since');
});
it('does not leak a target across turns', async () => {
const world = setup();
const file = join(world.env.workDir, 'a.txt');
writeFileSync(file, 'hello');
await runBefore(world, beforeCtx('Edit', file, { id: 'call-abandoned', turnId: 1 }));
await runBefore(
world,
beforeCtx('Edit', join(world.env.workDir, 'other.txt'), { turnId: 2 }),
);
// A late did-hook for the abandoned turn-1 call finds no swept target and
// records nothing: a.txt stays never-read, so a real Edit still blocks.
const abandonedCtx = beforeCtx('Edit', file, { id: 'call-abandoned', turnId: 1 });
await runDid(world, abandonedCtx);
const retry = await runBlocked(world, 'Edit', file);
expect(retry.output).toContain('has not been read in this session');
});
it('ignores tools other than Read/Write/Edit entirely', async () => {
const world = setup();
const ctx = await runBefore(
world,
beforeCtx('Bash', join(world.env.workDir, 'a.txt'), { args: { command: 'ls' } }),
);
expect(ctx.decision).toBeUndefined();
const did = await runDid(world, ctx);
expect(did.result.note).toBeUndefined();
expect(world.env.statCalls()).toBe(0);
});
});

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

@ -1493,39 +1493,7 @@ describe('SessionLifecycleService', () => {
}
});
it('reports the routable phase with the holder address when multi_server is on', async () => {
const root = await makeTmpRoot();
const first = build(
realInstanceSeeds(root, [
stubPair(IFlagService, stubFlag(true)),
stubPair(
ISessionLeaseContactProvider,
new SessionLeaseContactProvider(() => ({
type: 'address',
address: 'http://127.0.0.1:5555',
})),
),
]),
);
const firstHost = host!;
await first.create({ sessionId: 's1', workDir: '/tmp/proj' });
try {
const second = build(
realInstanceSeeds(root, [stubPair(IFlagService, stubFlag(true))]),
);
const error = await createError(second, 's1');
expect(error.code).toBe(ErrorCodes.SESSION_HELD_BY_PEER);
expect(error.details).toEqual({
kind: 'held-by-peer',
phase: 'routable',
address: 'http://127.0.0.1:5555',
});
} finally {
firstHost.dispose();
}
});
it('never emits the holder address when multi_server is off', async () => {
it('reports the routable phase with the holder address', async () => {
const root = await makeTmpRoot();
const first = build(
realInstanceSeeds(root, [
@ -1544,7 +1512,11 @@ describe('SessionLifecycleService', () => {
const second = build(realInstanceSeeds(root));
const error = await createError(second, 's1');
expect(error.code).toBe(ErrorCodes.SESSION_HELD_BY_PEER);
expect(error.details).toEqual({ kind: 'held-by-peer', phase: 'held-by-local-instance' });
expect(error.details).toEqual({
kind: 'held-by-peer',
phase: 'routable',
address: 'http://127.0.0.1:5555',
});
} finally {
firstHost.dispose();
}

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

@ -1,10 +1,12 @@
import { mkdtemp, mkdir, rm, stat, writeFile } from 'node:fs/promises';
import { mkdtemp, mkdir, readFile, rm, stat, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'pathe';
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';
const isWin = process.platform === 'win32';
const encoder = new TextEncoder();
@ -87,3 +89,57 @@ describe('FileStorageService — error translation', () => {
});
});
});
describe('FileStorageService — session write fencing', () => {
let dir: string;
beforeEach(async () => {
dir = await mkdtemp(join(tmpdir(), 'fss-fence-'));
});
afterEach(async () => {
await rm(dir, { recursive: true, force: true });
});
it('revalidates the session authority for write, append, and delete', async () => {
const registry = new WriteAuthorityRegistryService();
let writable = true;
const registration = registry.register({
sessionId: 'session',
assertWritable: () => {
if (!writable) {
throw new Error2(ErrorCodes.SESSION_LEASE_LOST, 'session lease lost');
}
},
});
const svc = new FileStorageService(dir, undefined, undefined, undefined, registry);
const scope = 'sessions/workspace/session/agents/main/tool-results';
await svc.write(scope, 'result.txt', encoder.encode('a'));
await svc.append(scope, 'result.txt', encoder.encode('b'));
expect(await readFile(join(dir, scope, 'result.txt'), 'utf8')).toBe('ab');
writable = false;
await expect(svc.write(scope, 'result.txt', encoder.encode('c'))).rejects.toMatchObject({
code: ErrorCodes.SESSION_LEASE_LOST,
});
await expect(svc.append(scope, 'result.txt', encoder.encode('c'))).rejects.toMatchObject({
code: ErrorCodes.SESSION_LEASE_LOST,
});
await expect(svc.delete(scope, 'result.txt')).rejects.toMatchObject({
code: ErrorCodes.SESSION_LEASE_LOST,
});
expect(await readFile(join(dir, scope, 'result.txt'), 'utf8')).toBe('ab');
registration.dispose();
});
it('fails closed without a session authority and leaves non-session scopes untouched', async () => {
const registry = new WriteAuthorityRegistryService();
const svc = new FileStorageService(dir, undefined, undefined, undefined, registry);
await expect(
svc.write('sessions/workspace/session/agents/main/blobs', 'blob', encoder.encode('x')),
).rejects.toMatchObject({ code: ErrorCodes.SESSION_LEASE_LOST });
await expect(svc.write('cron/workspace', 'task.json', encoder.encode('{}'))).resolves.toBeUndefined();
});
});

1
packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts
View file

@ -347,6 +347,7 @@ describe('AgentLifecycleService', () => {
ix.stub(IAgentTaskService, {
_serviceBrand: undefined,
stopAllOnExit,
flushPersistence: async () => {},
} as unknown as IAgentTaskService);
ix.stub(IAgentFullCompactionService, {
_serviceBrand: undefined,

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

@ -6,7 +6,7 @@
* notification, idempotent release, and contact-provider seed semantics.
*/
import { existsSync, mkdtempSync, rmSync } from 'node:fs';
import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
@ -68,7 +68,6 @@ function hostWith(seeds: Parameters<typeof createScopedTestHost>[0] = []): Scope
describe('SessionLease', () => {
it('reports its identity through info and passes the hard gate while held', async () => {
const lease = await acquire();
expect(lease.checkHeld()).toBe(true);
expect(lease.info).toEqual({ sessionId: 's1', lockId: lease.lockId });
expect(() => lease.assertWritable()).not.toThrow();
lease.release();
@ -77,12 +76,13 @@ describe('SessionLease', () => {
it('fires the loss notification once and then fails closed', async () => {
const onLost = vi.fn();
const lease = await acquire('s1', onLost);
lease.markLost();
// Replacing the sentinel fails the kernel handle's dev/ino identity
// check, driving the loss path through the real gate.
rmSync(sessionLeasePath(tmpDir, 's1'));
writeFileSync(sessionLeasePath(tmpDir, 's1'), '');
expect(thrownError(() => lease.assertWritable()).code).toBe(ErrorCodes.SESSION_LEASE_LOST);
expect(onLost).toHaveBeenCalledTimes(1);
expect(onLost).toHaveBeenCalledWith('s1');
lease.markLost();
expect(onLost).toHaveBeenCalledTimes(1);
expect(thrownError(() => lease.assertWritable()).code).toBe(ErrorCodes.SESSION_LEASE_LOST);
expect(onLost).toHaveBeenCalledTimes(1);
});
@ -92,7 +92,6 @@ describe('SessionLease', () => {
lease.release();
lease.release();
expect(lease.released).toBe(true);
expect(lease.info).toBeUndefined();
expect(existsSync(sessionLeasePath(tmpDir, 's1'))).toBe(true);
expect(existsSync(`${sessionLeasePath(tmpDir, 's1')}.owner.json`)).toBe(false);

7
packages/kap-server/src/envelope.ts
View file

@ -4,4 +4,9 @@
* Keep this file as a re-export shim so downstream `from './envelope'`
* imports inside the server stay stable and don't all need to be touched.
*/
export { okEnvelope, errEnvelope, type Envelope } from './protocol/envelope';
export {
okEnvelope,
errEnvelope,
ownershipRedirectEnvelope,
type Envelope,
} from './protocol/envelope';

12
packages/kap-server/src/error-handler.ts
View file

@ -23,7 +23,7 @@
*/
import { ErrorCodes, isError2 } from '@moonshot-ai/agent-core-v2';
import { errEnvelope } from './envelope';
import { errEnvelope, ownershipRedirectEnvelope } from './envelope';
import { ErrorCode } from './protocol/error-codes';
import type { FastifyError } from 'fastify';
@ -50,15 +50,7 @@ export function installErrorHandler(app: ErrorHandlerHost): void {
// server failure: surface 40921 with the structured details (phase /
// redirect address) and keep the stack in the log only.
if (isError2(err) && err.code === ErrorCodes.SESSION_HELD_BY_PEER) {
reply.status(200).send(
errEnvelope(
ErrorCode.SESSION_HELD_BY_PEER,
err.message,
requestId,
undefined,
err.details,
),
);
reply.status(200).send(ownershipRedirectEnvelope(err, requestId));
return;
}
req.log.error({ err, request_id: requestId }, 'unhandled error');

22
packages/kap-server/src/protocol/envelope.ts
View file

@ -6,6 +6,8 @@
import { z } from 'zod';
import { ErrorCode } from './error-codes';
export const envelopeSchema = <T extends z.ZodTypeAny>(data: T) =>
z.object({
code: z.number().int(),
@ -48,3 +50,23 @@ export function errEnvelope(
): Envelope<null> {
return { code, msg, data: null, request_id: requestId, stack, details };
}
/**
* Build the `40921 session.held_by_peer` ownership-redirect envelope. The
* structured `details` payload (`held-by-peer` phase / redirect address) is
* the actionable part, so it rides the envelope while the stack stays
* server-side. Accepts the `Error2` shape structurally so this module stays
* free of the engine dependency.
*/
export function ownershipRedirectEnvelope(
err: { readonly message: string; readonly details?: unknown },
requestId: string,
): Envelope<null> {
return errEnvelope(
ErrorCode.SESSION_HELD_BY_PEER,
err.message,
requestId,
undefined,
err.details,
);
}

15
packages/kap-server/src/routes/fs.ts
View file

@ -33,7 +33,7 @@ import {
} from '@moonshot-ai/agent-core-v2/session/sessionFs/fs';
import { z } from 'zod';
import { errEnvelope, okEnvelope } from '../envelope';
import { errEnvelope, okEnvelope, ownershipRedirectEnvelope } from '../envelope';
import {
launchDetached,
openFileCommandFor,
@ -532,18 +532,7 @@ function sendMappedError(reply: Reply, req: { id: string }, err: unknown): void
reply.send(errEnvelope(ErrorCode.SESSION_NOT_FOUND, err.message, requestId, err.stack));
return;
case ErrorCodes.SESSION_HELD_BY_PEER:
// Ownership redirect: the details payload (`held-by-peer` phase /
// address) is the actionable part, so it rides the envelope and the
// stack stays server-side.
reply.send(
errEnvelope(
ErrorCode.SESSION_HELD_BY_PEER,
err.message,
requestId,
undefined,
err.details,
),
);
reply.send(ownershipRedirectEnvelope(err, requestId));
return;
// hostFs errors that escaped the sessionFs layer keep their `os.fs.*`
// code; map them onto the closest v1 wire code (ENOTDIR collapses into

17
packages/kap-server/src/routes/prompts.ts
View file

@ -64,7 +64,7 @@ import {
} from '../protocol/rest-prompt';
import { z } from 'zod';
import { errEnvelope, okEnvelope } from '../envelope';
import { errEnvelope, okEnvelope, ownershipRedirectEnvelope } from '../envelope';
import { requestLog } from '../lib/requestLog';
import { defineRoute } from '../middleware/defineRoute';
import { ensureMainAgent, MAIN_AGENT_ID } from '../transport/mainAgent';
@ -819,19 +819,8 @@ function sendMappedError(
case 'session.busy':
reply.send(errEnvelope(ErrorCode.SESSION_BUSY, err.message, requestId, err.stack));
return;
case 'session.held_by_peer':
// Ownership redirect: the details payload (`held-by-peer` phase /
// address) is the actionable part, so it rides the envelope and the
// stack stays server-side.
reply.send(
errEnvelope(
ErrorCode.SESSION_HELD_BY_PEER,
err.message,
requestId,
undefined,
err.details,
),
);
case ErrorCodes.SESSION_HELD_BY_PEER:
reply.send(ownershipRedirectEnvelope(err, requestId));
return;
case 'prompt.already_completed':
reply.send({

28
packages/kap-server/src/routes/sessions.ts
View file

@ -138,7 +138,7 @@ import {
import { workspaceIdSchema } from '../protocol/workspace';
import { z } from 'zod';
import { errEnvelope, okEnvelope } from '../envelope';
import { errEnvelope, okEnvelope, ownershipRedirectEnvelope } from '../envelope';
import { requestLog } from '../lib/requestLog';
import { defineRoute } from '../middleware/defineRoute';
import { ensureMainAgent, MAIN_AGENT_ID } from '../transport/mainAgent';
@ -1147,12 +1147,13 @@ function resolvePendingInteraction(
*
* - 'self': this instance materialized the session and still holds its write
* lease (`ISessionLeaseService.info` survives only while held).
* - 'peer': a lease is on disk and held (or mid-creation) by someone else;
* `address` rides along when the holder advertised one — the redirect
* target. An unresponsive holder still counts as 'peer' (it is never
* auto-taken over).
* - 'none': no lease on disk, or a stale husk a dead holder left behind —
* the session is materialized nowhere and any instance may acquire it.
* - 'peer': the lease file is held (or mid-creation) under a live kernel
* lock by someone else; `address` rides along when the holder advertised
* one — the redirect target.
* - 'none': no live kernel lock on the lease file — the session is
* materialized nowhere and any instance may acquire it. A dead holder's
* lock is released by the kernel, so leftover owner metadata alone does
* not count as held.
*
* Advisory only: the sync `inspect` read can observe a holder that died the
* next millisecond; the lease's own acquisition protocol stays the authority.
@ -1296,18 +1297,7 @@ function sendMappedError(
});
return;
case ErrorCodes.SESSION_HELD_BY_PEER:
// Ownership redirect: the details payload (`held-by-peer` phase /
// address) is the actionable part, so it rides the envelope and the
// stack stays server-side.
reply.send(
errEnvelope(
ErrorCode.SESSION_HELD_BY_PEER,
err.message,
requestId,
undefined,
err.details,
),
);
reply.send(ownershipRedirectEnvelope(err, requestId));
return;
case ErrorCodes.GOAL_ALREADY_EXISTS:
reply.send(errEnvelope(ErrorCode.GOAL_ALREADY_EXISTS, err.message, requestId, err.stack));

15
packages/kap-server/src/routes/terminals.ts
View file

@ -32,7 +32,7 @@ import {
import { createTerminalRequestSchema } from '@moonshot-ai/agent-core-v2/os/interface/terminal';
import { z } from 'zod';
import { errEnvelope, okEnvelope } from '../envelope';
import { errEnvelope, okEnvelope, ownershipRedirectEnvelope } from '../envelope';
import { requestLog } from '../lib/requestLog';
import { defineRoute } from '../middleware/defineRoute';
import { ErrorCode } from '../protocol/error-codes';
@ -242,18 +242,7 @@ function sendMappedError(
reply.send(errEnvelope(ErrorCode.TERMINAL_NOT_FOUND, err.message, requestId, err.stack));
return;
case ErrorCodes.SESSION_HELD_BY_PEER:
// Ownership redirect: the details payload (`held-by-peer` phase /
// address) is the actionable part, so it rides the envelope and the
// stack stays server-side.
reply.send(
errEnvelope(
ErrorCode.SESSION_HELD_BY_PEER,
err.message,
requestId,
undefined,
err.details,
),
);
reply.send(ownershipRedirectEnvelope(err, requestId));
return;
}
}

28
packages/kap-server/src/services/sessionListWatch/sessionListWatchService.ts
View file

@ -2,18 +2,18 @@
* `SessionListWatchService` — the event plane of multi-instance session-list
* sync.
*
* Several kap-server instances can share one home directory (the
* `multi_server` experimental flag). The session list itself needs no
* synchronization: `ISessionIndex.list()` re-enumerates the shared
* `<home>/sessions` tree on every request, so a peer's sessions are visible on
* the next pull. What a peer can never produce is the *event* — core events
* are process-local. This service closes that gap locally: it watches the
* shared sessions tree and, on any workspace/session directory appearing or
* disappearing, publishes ONE debounced `session.list_changed` hint on this
* instance's core `IEventService`, which the `SessionEventBroadcaster` fans
* out live (volatile, never journaled) to every connected WS client. Clients
* then re-pull the list — the directory scan stays the single authority, the
* hint is pure "go refetch" advice and deliberately carries no payload.
* Several kap-server instances can share one home directory. The session list
* itself needs no synchronization: `ISessionIndex.list()` re-enumerates the
* shared `<home>/sessions` tree on every request, so a peer's sessions are
* visible on the next pull. What a peer can never produce is the *event* —
* core events are process-local. This service closes that gap locally: it
* watches the shared sessions tree and, on any workspace/session directory
* appearing or disappearing, publishes ONE debounced `session.list_changed`
* hint on this instance's core `IEventService`, which the
* `SessionEventBroadcaster` fans out live (volatile, never journaled) to
* every connected WS client. Clients then re-pull the list — the directory
* scan stays the single authority, the hint is pure "go refetch" advice and
* deliberately carries no payload.
*
* Two-layer topology (a root recursive watch was rejected as an event flood):
* - root `<home>/sessions` at depth 0: workspace directories appearing /
@ -24,8 +24,8 @@
* `ignoreInitial` so boot produces no hint flood.
*
* This is transport state (like `FsWatchBridge` / `SessionEventBroadcaster`):
* constructed in `start.ts` when `multi_server` is on — never DI-registered —
* and disposed during server close before the core scope goes down.
* constructed in `start.ts` on every boot — never DI-registered — and
* disposed during server close before the core scope goes down.
*/
import { mkdirSync } from 'node:fs';

36
packages/kap-server/src/start.ts
View file

@ -146,22 +146,6 @@ export interface RunningServer {
const DEFAULT_HOST = '127.0.0.1';
const DEFAULT_PORT = 58627;
/**
* Env gate for the multi-server session-list sync below
* (`KIMI_CODE_EXPERIMENTAL_MULTI_SERVER`). Resolved directly from the
* environment *before* bootstrap, and deliberately NOT via the flag service /
* master `KIMI_CODE_EXPERIMENTAL_FLAG`: that switch already enables the v2
* engine itself, and coupling multi-instance behavior to it would change the
* watch surface of every v2 server before the feature is ready. Keeping the
* gate specific makes multi-server strictly opt-in.
*/
const MULTI_SERVER_FLAG_ENV = 'KIMI_CODE_EXPERIMENTAL_MULTI_SERVER';
function isMultiServerEnabled(env: NodeJS.ProcessEnv): boolean {
const raw = (env[MULTI_SERVER_FLAG_ENV] ?? '').trim().toLowerCase();
return raw === '1' || raw === 'true' || raw === 'yes' || raw === 'on';
}
export async function startServer(opts: ServerStartOptions = {}): Promise<RunningServer> {
const host = opts.host ?? DEFAULT_HOST;
const port = opts.port ?? DEFAULT_PORT;
@ -333,7 +317,7 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
const lifecycleDrain = lifecycle.beginClose();
// Stop the sessions-tree watcher first: a debounced hint firing mid-close
// would publish into an event bus whose subscribers are already unwinding.
sessionListWatch?.dispose();
sessionListWatch.dispose();
// Release-order contract: sessions close FIRST (each runs the
// onWillCloseSession hooks and drains agents while the transport is still
// alive, so teardown events still fan out and land in the journal), the
@ -370,17 +354,13 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
// Multi-instance session-list sync, event plane (design §3.8): watches the
// shared sessions tree for peer-created/removed workspace & session dirs
// and hints `session.list_changed` so this instance's clients re-pull.
// Gated by the dedicated `multi_server` env flag (see isMultiServerEnabled)
// — flag off is the single-instance shape where peers cannot exist.
const sessionListWatch = isMultiServerEnabled(process.env)
? new SessionListWatchService({
sessionsDir: join(homeDir, 'sessions'),
fsWatch: core.accessor.get(IHostFsWatchService),
events: core.accessor.get(IEventService),
logger,
})
: undefined;
sessionListWatch?.start().catch((error: unknown) =>
const sessionListWatch = new SessionListWatchService({
sessionsDir: join(homeDir, 'sessions'),
fsWatch: core.accessor.get(IHostFsWatchService),
events: core.accessor.get(IEventService),
logger,
});
sessionListWatch.start().catch((error: unknown) =>
logger.error({ err: error }, 'session list watch failed to start'),
);

72
packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts
View file

@ -39,8 +39,6 @@
* connections without a transcript spec are unaffected.
*/
import { dirname } from 'node:path';
import type {
AgentActivityState,
ApprovalResponse,
@ -64,7 +62,7 @@ import {
ISessionIndex,
ISessionLifecycleService,
MAIN_AGENT_ID,
LEASE_CREATING_RETRY_AFTER_MS,
heldByPeerDetailsFromInspection,
sessionLeasePath,
} from '@moonshot-ai/agent-core-v2';
import type { TurnEndReason } from '@moonshot-ai/agent-core-v2/agent/loop/turnEvents';
@ -245,7 +243,7 @@ export class SessionEventBroadcaster {
constructor(
private readonly opts: {
readonly eventsDir: string;
readonly homeDir?: string;
readonly homeDir: string;
readonly core: Scope;
readonly logger?: JournalLogger;
readonly maxBufferSize?: number;
@ -370,23 +368,10 @@ export class SessionEventBroadcaster {
const summary = await this.opts.core.accessor.get(ISessionIndex).get(sessionId);
if (summary === undefined) return undefined;
const homeDir = this.opts.homeDir ?? dirname(dirname(this.opts.eventsDir));
const inspection = this.opts.core
.accessor.get(ICrossProcessLockService)
.inspect(sessionLeasePath(homeDir, sessionId));
if (inspection.state === 'creating') {
return {
kind: 'held-by-peer',
phase: 'creating',
retry_after_ms: LEASE_CREATING_RETRY_AFTER_MS,
};
}
if (inspection.state !== 'held' || inspection.payload === undefined) return undefined;
if (inspection.payload.address !== undefined) {
return { kind: 'held-by-peer', phase: 'routable', address: inspection.payload.address };
}
return { kind: 'held-by-peer', phase: 'held-by-local-instance' };
.inspect(sessionLeasePath(this.opts.homeDir, sessionId));
return heldByPeerDetailsFromInspection(inspection);
}
unsubscribe(sessionId: string, target: BroadcastTarget): void {
@ -681,8 +666,10 @@ export class SessionEventBroadcaster {
// While the journal has unrecovered write failures, never serve from the
// tail: those events are not durable, and replaying them as if they were
// would resurrect the "fake durable" hole after a restart. `readSince`
// retries the pending flush and throws the sticky JournalStorageError
// instead — the replay edge maps that to a client-visible resync.
// flushes first (which retries a transient write failure); once the
// journal is sticky the flush is a no-op and `readSince` throws the
// sticky JournalStorageError instead — the replay edge maps that to a
// client-visible resync.
if (!journal.writeFailure) {
const tailStart = tail[0]?.seq;
if (tailStart !== undefined && tailStart <= cursor.seq + 1) {
@ -844,49 +831,6 @@ export class SessionEventBroadcaster {
return state;
}
private ensureGlobalState(): Promise<SessionState> {
const existing = this.sessions.get(GLOBAL_SESSION_ID);
if (existing !== undefined) return Promise.resolve(existing);
let pending = this.pendingStates.get(GLOBAL_SESSION_ID);
if (pending === undefined) {
pending = this.createGlobalState().finally(() => {
if (this.pendingStates.get(GLOBAL_SESSION_ID) === pending) {
this.pendingStates.delete(GLOBAL_SESSION_ID);
}
});
this.pendingStates.set(GLOBAL_SESSION_ID, pending);
}
return pending as Promise<SessionState>;
}
private async createGlobalState(): Promise<SessionState> {
const journal = await SessionEventJournal.open(
sessionJournalPath(this.opts.eventsDir, GLOBAL_SESSION_ID),
this.opts.logger,
);
const state: SessionState = {
sessionId: GLOBAL_SESSION_ID,
journal,
tracker: new InFlightTurnTracker(),
roster: new SubagentRosterTracker(),
activityByAgent: new Map(),
emittedBusy: false,
emittedMainTurnActive: false,
emittedPendingInteraction: 'none',
pendingInteraction: 'none',
tail: [],
targets: new Map(),
queue: Promise.resolve(),
agentDisposables: new Map(),
lifecycleDisposables: [],
knownInteractions: new Map(),
transcriptSeeded: new Set(),
deferredTranscriptSeeds: new Map(),
};
this.sessions.set(GLOBAL_SESSION_ID, state);
return state;
}
private onCoreEvent(event: GlobalEvent): void {
if (event.type === 'session.list_changed') {
// Published by `SessionListWatchService` when a workspace/session

19
packages/kap-server/src/transport/ws/v1/sessionEventJournal.ts
View file

@ -27,12 +27,15 @@
* for fan-out); bytes are flushed on a microtask-scheduled async batch. Each
* batch uses a single `open(path, 'a')` → write → fsync → close cycle. Pending
* lines are dequeued only AFTER the batch is durable; a failed round keeps the
* whole batch (and the pending header) for the retry. After
* whole batch (and the pending header) for a retry driven by the next
* append-scheduled or read-triggered flush. After
* {@link STICKY_FAILURE_THRESHOLD} consecutive failures the journal goes
* sticky: `nextSeq()`/`append()` fail fast (pending can never grow unbounded)
* and `readSince()` throws a {@link JournalStorageError} instead of silently
* serving fewer events — "not served" must stay distinguishable from "nothing
* to serve". `readSince()` flushes first so replay never misses queued lines.
* sticky: `flush()` turns into a no-op (the kept pending lines are never
* retried), `nextSeq()`/`append()` fail fast (pending can never grow
* unbounded), and `readSince()` throws a {@link JournalStorageError} instead
* of silently serving fewer events — "not served" must stay distinguishable
* from "nothing to serve". `readSince()` flushes first so replay never misses
* queued lines.
* A torn trailing line from a crash is tolerated and ignored on open, and a
* pure cold-read open → close writes zero bytes.
*/
@ -320,8 +323,8 @@ export class SessionEventJournal {
}
await this.flushPromise;
// Give up once sticky instead of hot-spinning on a persistently failing
// disk; the kept pending lines are retried by the next append-scheduled
// or read-triggered round.
// disk: appends now fail fast and `readSince` throws, so the kept
// pending lines are never retried.
if (this.stickyError !== undefined) return;
}
}
@ -379,7 +382,7 @@ export class SessionEventJournal {
} catch (error) {
const committed = await countCommittedPrefix(this.filePath, lines);
if (committed > 0) {
if (headerLine !== undefined && committed > 0) this.headerPending = false;
if (headerLine !== undefined) this.headerPending = false;
this.pendingLines.splice(0, Math.max(0, committed - (headerLine === undefined ? 0 : 1)));
this.stickyError ??= new JournalStorageError(this.filePath, error);
return true;

4
packages/kap-server/src/transport/ws/v1/wsConnectionV1.ts
View file

@ -269,7 +269,7 @@ export class WsConnectionV1 implements BroadcastTarget {
transcriptSince: transcriptSince?.[sid],
});
if (!ok) {
const ownership = await this.broadcaster.getSubscriptionFailure?.(sid);
const ownership = await this.broadcaster.getSubscriptionFailure(sid);
if (ownership !== undefined) ownershipDetails[sid] = ownership;
else notFound.push(sid);
continue;
@ -363,7 +363,7 @@ export class WsConnectionV1 implements BroadcastTarget {
transcriptSince,
});
if (!ok) {
const ownership = await this.broadcaster.getSubscriptionFailure?.(sid);
const ownership = await this.broadcaster.getSubscriptionFailure(sid);
if (ownership !== undefined) ownershipDetails[sid] = ownership;
else notFound.push(sid);
return;

25
packages/kap-server/test/session-ownership.e2e.test.ts
View file

@ -1,10 +1,10 @@
/**
* Multi-server session ownership — two kap-servers on ONE kimi home (the
* `multi_server` experimental flag on). Instance A creates the session and
* holds its write lease; instance B's materializing routes for the same
* session are answered with `40921 session.held_by_peer` carrying the
* structured ownership details (phase `routable` + A's address), so clients
* can redirect to the holder. Closing A releases the lease and B takes over.
* Multi-server session ownership — two kap-servers on ONE kimi home. Instance
* A creates the session and holds its write lease; instance B's materializing
* routes for the same session are answered with
* `40921 session.held_by_peer` carrying the structured ownership details
* (phase `routable` + A's address), so clients can redirect to the holder.
* Closing A releases the lease and B takes over.
* Run: `pnpm --filter @moonshot-ai/kap-server exec vitest run test/session-ownership.e2e.test.ts`.
*/
import { mkdtemp, readFile, rm } from 'node:fs/promises';
@ -13,13 +13,10 @@ import { join } from 'node:path';
import { sessionLeasePath } from '@moonshot-ai/agent-core-v2';
import { ErrorCode } from '../src/protocol/error-codes';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { afterEach, describe, expect, it } from 'vitest';
import { type RunningServer, startServer } from '../src/start';
/** Same env gate start.ts checks locally (`KIMI_CODE_EXPERIMENTAL_MULTI_SERVER`). */
const MULTI_SERVER_FLAG_ENV = 'KIMI_CODE_EXPERIMENTAL_MULTI_SERVER';
interface Envelope<T> {
code: number;
msg: string;
@ -37,16 +34,8 @@ describe('multi-server session ownership (session.held_by_peer → 40921)', () =
let home: string | undefined;
let serverA: RunningServer | undefined;
let serverB: RunningServer | undefined;
let previousFlag: string | undefined;
beforeEach(() => {
previousFlag = process.env[MULTI_SERVER_FLAG_ENV];
process.env[MULTI_SERVER_FLAG_ENV] = '1';
});
afterEach(async () => {
if (previousFlag === undefined) delete process.env[MULTI_SERVER_FLAG_ENV];
else process.env[MULTI_SERVER_FLAG_ENV] = previousFlag;
if (serverB !== undefined) {
await serverB.close();
serverB = undefined;

2
packages/kap-server/test/sessionEventBroadcaster.test.ts
View file

@ -279,6 +279,7 @@ describe('SessionEventBroadcaster', () => {
eventBus = new FakeEventBus();
bc = new SessionEventBroadcaster({
eventsDir: dir,
homeDir: dir,
core: makeCore(sessions, eventBus),
maxBufferSize: 3,
});
@ -1468,6 +1469,7 @@ describe('SessionEventBroadcaster', () => {
const dir2 = await mkdtemp(join(tmpdir(), 'kimi-broadcaster-test-'));
const bc2 = new SessionEventBroadcaster({
eventsDir: dir2,
homeDir: dir2,
core: makeCore(sessions, eventBus),
maxBufferSize: 20,
});

1
packages/kap-server/test/wsConnectionV1.test.ts
View file

@ -62,6 +62,7 @@ function makeBroadcaster(): SessionEventBroadcaster {
return {
subscribe: async () => true,
unsubscribe: () => {},
getSubscriptionFailure: async () => undefined,
getCursor: async () => ({ seq: 0, epoch: '' }),
getBufferedSince: async () => ({
events: [],

56
packages/kernel-file-lock/src/index.ts
View file

@ -9,19 +9,12 @@ export interface KernelFileLockBinding {
export type KernelFileLockBindingLoader = () => KernelFileLockBinding | undefined;
export interface KernelFileLockAcquireOptions {
readonly timeoutMs: number;
readonly retryIntervalMs?: number;
}
export interface KernelFileLockHandle {
readonly path: string;
readonly held: boolean;
checkHeld(): boolean;
release(): void;
}
const DEFAULT_RETRY_INTERVAL_MS = 50;
const bindingLoaderKey = Symbol.for('@moonshot-ai/kernel-file-lock/binding-loader');
const nodeRequire = createRequire(import.meta.url);
@ -42,10 +35,6 @@ function getBinding(): KernelFileLockBinding {
return binding;
}
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
class KernelFileLockHandleImpl implements KernelFileLockHandle {
private released = false;
@ -55,10 +44,6 @@ class KernelFileLockHandleImpl implements KernelFileLockHandle {
private readonly binding: KernelFileLockBinding,
) {}
get held(): boolean {
return this.checkHeld();
}
checkHeld(): boolean {
if (this.released) return false;
try {
@ -108,44 +93,3 @@ export function tryAcquireKernelFileLock(path: string): KernelFileLockHandle | u
throw error;
}
}
export async function acquireKernelFileLock(
path: string,
options: KernelFileLockAcquireOptions,
): Promise<KernelFileLockHandle> {
const deadline = Date.now() + options.timeoutMs;
const retryIntervalMs = options.retryIntervalMs ?? DEFAULT_RETRY_INTERVAL_MS;
let firstAttempt = true;
for (;;) {
const isFirstAttempt = firstAttempt;
if (!isFirstAttempt && Date.now() >= deadline) {
throw new KernelFileLockTimeoutError(path, options.timeoutMs);
}
firstAttempt = false;
const handle = tryAcquireKernelFileLock(path);
if (handle !== undefined) {
if (!isFirstAttempt && Date.now() >= deadline) {
handle.release();
throw new KernelFileLockTimeoutError(path, options.timeoutMs);
}
return handle;
}
const remainingMs = deadline - Date.now();
if (remainingMs <= 0) {
throw new KernelFileLockTimeoutError(path, options.timeoutMs);
}
await sleep(Math.min(retryIntervalMs, remainingMs));
}
}
export class KernelFileLockTimeoutError extends Error {
readonly code = 'ELOCKTIMEOUT';
constructor(
readonly path: string,
readonly timeoutMs: number,
) {
super(`timed out waiting ${timeoutMs}ms for kernel file lock: ${path}`);
this.name = 'KernelFileLockTimeoutError';
}
}

72
packages/kernel-file-lock/test/kernel-file-lock.test.ts
View file

@ -8,11 +8,8 @@ import { fileURLToPath } from 'node:url';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import {
acquireKernelFileLock,
KernelFileLockTimeoutError,
setKernelFileLockBindingLoader,
tryAcquireKernelFileLock,
type KernelFileLockBinding,
type KernelFileLockHandle,
} from '../src/index.js';
@ -58,75 +55,6 @@ describe('kernel-file-lock', () => {
handles.push(second!);
});
it('waits and times out without taking ownership', async () => {
const first = tryAcquireKernelFileLock(lockPath)!;
handles.push(first);
await expect(
acquireKernelFileLock(lockPath, { timeoutMs: 15, retryIntervalMs: 5 }),
).rejects.toBeInstanceOf(KernelFileLockTimeoutError);
});
it('does not acquire a lock released after the deadline', async () => {
const first = tryAcquireKernelFileLock(lockPath)!;
handles.push(first);
const releaseTimer = setTimeout(() => first.release(), 20);
const result = await acquireKernelFileLock(lockPath, {
timeoutMs: 10,
retryIntervalMs: 100,
}).then(
(handle) => ({ status: 'acquired' as const, handle }),
(error: unknown) => ({ status: 'rejected' as const, error }),
);
clearTimeout(releaseTimer);
if (result.status === 'acquired') handles.push(result.handle);
expect(result.status).toBe('rejected');
if (result.status === 'rejected') {
expect(result.error).toBeInstanceOf(KernelFileLockTimeoutError);
}
});
it('releases a lock acquired as a retry crosses the deadline', async () => {
let lockCalls = 0;
let unlockCalls = 0;
const times = [0, 0, 9, 10];
const nativeBinding: KernelFileLockBinding = {
tryLock: () => {
lockCalls++;
return lockCalls > 1;
},
unlock: () => {
unlockCalls++;
},
};
setKernelFileLockBindingLoader(() => nativeBinding);
vi.spyOn(Date, 'now').mockImplementation(() => times.shift() ?? 10);
const result = await acquireKernelFileLock(lockPath, {
timeoutMs: 10,
retryIntervalMs: 0,
}).then(
(handle) => ({ status: 'acquired' as const, handle }),
(error: unknown) => ({ status: 'rejected' as const, error }),
);
if (result.status === 'acquired') handles.push(result.handle);
expect(result.status).toBe('rejected');
if (result.status === 'rejected') {
expect(result.error).toBeInstanceOf(KernelFileLockTimeoutError);
}
expect(unlockCalls).toBe(1);
});
it('allows the immediate first attempt with a zero timeout', async () => {
const handle = await acquireKernelFileLock(lockPath, { timeoutMs: 0 });
handles.push(handle);
expect(handle.checkHeld()).toBe(true);
});
it('coordinates with a separate process', async () => {
const holderPath = fileURLToPath(new URL('./holder.ts', import.meta.url));
const child = spawn(process.execPath, ['--import', 'tsx', holderPath, lockPath], {

4
packages/klient/AGENTS.md
View file

@ -79,10 +79,6 @@ Pick the helper by what the case needs:
Hard rules:
- Same-home dual boot requires `KIMI_CODE_EXPERIMENTAL_MULTI_SERVER=1` at
boot time or the second boot fails with `ServerLockedError`. The helpers
set/restore it themselves (child env for spawned pairs) — do not export it
globally.
- Always `port: 0`. A fixed busy port silently walks to `port + 1`, which
breaks registry/port assertions and cross-test isolation.
- One pair per test file/worker; never share a `RunningServer` or

28
packages/klient/test/e2e/harness/testing/serverPair.ts
View file

@ -6,13 +6,9 @@
* `spawnServerProcess` instead only when the case is signal-sensitive
* (SIGSTOP / SIGKILL need real, distinct pids).
*
* Two hard requirements this helper encapsulates:
* - `KIMI_CODE_EXPERIMENTAL_MULTI_SERVER=1` must be set when each instance
* boots (read by `startServer` at call time only), or the second boot on
* the same home fails with `ServerLockedError`. The previous env value is
* saved and restored after boot / on `dispose()`.
* - Both instances must bind `port: 0` (OS-assigned) — a fixed busy port
* silently walks to `port + 1`, which breaks assertions on the registry.
* One hard requirement this helper encapsulates: both instances must bind
* `port: 0` (OS-assigned) — a fixed busy port silently walks to `port + 1`,
* which breaks assertions on the registry.
*
* `@moonshot-ai/kap-server` is imported lazily *inside* the function: its
* module graph contains `*.md?raw` imports that plain `tsx` (running without
@ -28,13 +24,6 @@ import type { RunningServer } from '@moonshot-ai/kap-server';
import { HttpClient } from '../http.js';
/**
* Literal copy of agent-core-v2's `MULTI_SERVER_FLAG_ENV`. Duplicated on
* purpose: importing the constant would pull the kap-server / agent-core-v2
* module graph into every consumer of this barrel (see file header).
*/
export const MULTI_SERVER_FLAG_ENV = 'KIMI_CODE_EXPERIMENTAL_MULTI_SERVER' as const;
// `recursive` rm can hit ENOTEMPTY on macOS while the closing server is still
// flushing/unlinking its own files — retry briefly (same trick as the v2
// smoke test's home cleanup).
@ -86,9 +75,7 @@ export async function startServerPair(options: ServerPairOptions = {}): Promise<
const home = options.home ?? (await mkdtemp(join(tmpdir(), 'kimi-e2e-pair-')));
const ownsHome = options.home === undefined;
const disableAuth = options.disableAuth ?? true;
// The multi-server gate wins over `options.env`: without it the second boot
// below cannot succeed on a shared home.
const envPatch: Record<string, string> = { ...options.env, [MULTI_SERVER_FLAG_ENV]: '1' };
const envPatch: Record<string, string> = { ...options.env };
const savedEnv = saveEnv(envPatch);
let envRestored = false;
const restoreEnv = (): void => {
@ -116,13 +103,6 @@ export async function startServerPair(options: ServerPairOptions = {}): Promise<
await a.close();
throw error;
}
// The flag is also read at request time — `heldByPeerDetails` phase
// classification and the unregistered-writer check consult it on every
// ownership rejection — not just inside `startServer`. Keep the env
// patched for the pair's whole lifetime; dispose() restores it.
// (Restoring it here made request-time reads fall back to the registry
// default `false`, turning routable 40921s into held-by-local-instance
// on any environment without the master flag.)
const baseUrl = (server: RunningServer): string => `http://${server.host}:${server.port}`;
let disposed = false;

10
packages/klient/test/e2e/harness/testing/serverProcess.ts
View file

@ -22,8 +22,6 @@
* - `build/register-raw-text-loader.mjs` makes `*.md?raw` prompt-template
* imports (kap-server → agent-core-v2) resolvable outside a bundler;
* plain `node` fails on those imports without it.
* - registers/lock state: same-home coexistence still requires
* `KIMI_CODE_EXPERIMENTAL_MULTI_SERVER=1`, passed through the child env.
*
* Readiness is the child's `{type:'ready'}` stdout line (printed after
* `startServer` resolved, i.e. the port is already listening). When driving
@ -38,7 +36,6 @@ import { dirname, join, resolve } from 'node:path';
import { createInterface } from 'node:readline';
import { fileURLToPath, pathToFileURL } from 'node:url';
import { MULTI_SERVER_FLAG_ENV } from './serverPair.js';
import {
SPAWN_SERVER_HOME_ENV,
type SpawnServerMessage,
@ -189,9 +186,9 @@ export async function spawnServerProcess(
}
/**
* Pair of spawned children sharing one home; the multi-server flag is pushed
* into both child envs — process-level patching like `startServerPair` does
* would not reach them.
* Pair of spawned children sharing one home; `options.env` is pushed into both
* child envs — process-level patching like `startServerPair` does would not
* reach them.
*/
export async function spawnServerProcessPair(
options: SpawnServerProcessOptions = {},
@ -201,7 +198,6 @@ export async function spawnServerProcessPair(
const childOptions: SpawnServerProcessOptions = {
...options,
home,
env: { ...options.env, [MULTI_SERVER_FLAG_ENV]: '1' },
};
try {
const a = await spawnServerProcess(childOptions);

27
packages/klient/test/e2e/legacy/dual-instance.test.ts
View file

@ -11,19 +11,15 @@
* + `listLiveServerInstances` agree).
* 2. Closing instance `a` leaves instance `b` serving (healthz 200, registry
* down to one live entry) while `a`'s port refuses connections.
* 3. `dispose()` restores `KIMI_CODE_EXPERIMENTAL_MULTI_SERVER` to its
* pre-boot value and removes the helper-created home directory.
* (Upstream removed the single-instance server lock — multi-server is
* the always-on model now, so the former "no flag → ServerLockedError"
* case no longer exists.)
* 3. `dispose()` removes the helper-created home directory.
*
* Subprocess (`spawnServerProcess`):
* 4. A child boots on a real distinct pid, answers healthz, and serves
* token-gated routes WITHOUT a token (`disableAuth`); `stop()` (SIGTERM)
* exits the child and removes the helper-created home.
* 5. A spawned pair shares one home (flag reaches the children's env); a
* SIGKILLed child's pid actually dies and its registry entry is swept as
* stale on the next `listLiveServerInstances` read.
* 5. A spawned pair shares one home; a SIGKILLed child's pid actually dies
* and its registry entry is swept as stale on the next
* `listLiveServerInstances` read.
*
* KNOWN BRANCH GAP (refactor-fs-watch WIP): kap-server's `close()` currently
* throws `appendLogStore depends on writeAuthorityRegistry which is NOT
@ -44,7 +40,6 @@ import { describe, expect, it } from 'vitest';
import { HttpClient } from '../harness/http.js';
import {
MULTI_SERVER_FLAG_ENV,
spawnServerProcess,
spawnServerProcessPair,
startServerPair,
@ -135,25 +130,15 @@ describe('dual-instance helpers', () => {
);
it(
'dispose() restores the multi-server env flag and removes the created home',
'dispose() removes the created home',
{ timeout: 30_000 },
async () => {
const log = createCaseLogger('dual-instance/dispose-cleanup');
const ambientFlag = process.env[MULTI_SERVER_FLAG_ENV];
const pair = await startServerPair();
// The flag stays patched for the pair's whole lifetime: request-time
// readers (40921 phase classification, unregistered-writer checks)
// consult it on every ownership rejection, not just at boot.
expect(process.env[MULTI_SERVER_FLAG_ENV]).toBe('1');
expect(existsSync(pair.home)).toBe(true);
await pair.dispose();
log('after dispose()', {
restoredFlag: process.env[MULTI_SERVER_FLAG_ENV] ?? null,
ambientFlag: ambientFlag ?? null,
homeExists: existsSync(pair.home),
});
expect(process.env[MULTI_SERVER_FLAG_ENV]).toBe(ambientFlag);
log('after dispose()', { homeExists: existsSync(pair.home) });
expect(existsSync(pair.home)).toBe(false);
},
);

7
packages/minidb/src/index.ts
View file

@ -1608,13 +1608,6 @@ export class MiniDb<V = unknown> {
// ---- maintenance --------------------------------------------------------
/** Verify that this process still holds the database's kernel lock. */
async renewLock(): Promise<void> {
if (this.lock === null) return;
await this.lock.renew();
this.ensureWritable();
}
/** Advanced/internal (read-replica owners such as the cluster shard pool):
* incrementally apply WAL frames appended to db.wal after `offset` — the
* same frames open-time recovery would replay, interpreted identically

4
packages/minidb/src/lockfile.ts
View file

@ -48,10 +48,6 @@ export class LockFile {
if (!this.checkHeld()) throw new LockError(`database write lock was lost: ${this.path}`);
}
async renew(): Promise<void> {
this.assertHeld();
}
async release(): Promise<void> {
this.releaseSync();
}

2
packages/minidb/test/cluster/lock.test.ts
View file

@ -91,7 +91,7 @@ test('read-only instance coexists with a live writer and sees its commits', asyn
}
});
test('a cached writer keeps a stable sentinel without renewal writes', async () => {
test('a cached writer keeps a stable sentinel', async () => {
const dir = await tmpDir('minidb-cluster-');
try {
const db = await ClusterDb.open({ dir, shardCount: 4, valueCodec: 'json', lockHoldMs: 0 });

2
packages/minidb/test/lock.test.ts
View file

@ -105,7 +105,7 @@ test('rewriting sentinel contents cannot transfer a live lock', async () => {
assert.equal(await first.acquire(), true);
await fs.writeFile(lockPath, 'operator note');
await assert.doesNotReject(() => first.renew());
assert.doesNotThrow(() => first.assertHeld());
assert.equal(await second.acquire(), false);
await first.release();

1
packages/minidb/test/review-fixes.test.ts
View file

@ -91,7 +91,6 @@ test('editing the sentinel payload cannot create a successor generation', async
await oldWriter.set('generation', 'old');
await fs.writeFile(path.join(dir, 'db.lock'), 'successor-generation');
await assert.doesNotReject(() => oldWriter.renewLock());
await assert.rejects(
() => MiniDb.open({ dir, valueCodec: 'string', autoCompact: false }),
/locked/,

1
packages/protocol/src/events.ts
View file

@ -316,7 +316,6 @@ export type KimiErrorCode =
| 'os.process.kill_failed'
| 'os.lock.held'
| 'os.lock.wait_timeout'
| 'os.lock.lost'
| 'os.lock.io'
| 'storage.not_found'
| 'storage.decode_failed'