From 2762bfc9098975fb2b68da029fa02e6a11e548ca Mon Sep 17 00:00:00 2001 From: 7Sageer <7sageer@djwcb.cn> Date: Tue, 21 Jul 2026 12:25:52 +0800 Subject: [PATCH] refactor: drop multi-server flag and dedupe session ownership classification --- .changeset/cross-process-lock-unification.md | 2 +- .changeset/multi-instance-session-sync.md | 2 +- .changeset/stale-write-fencing.md | 2 +- .../src/api/daemon/sessionOwnership.ts | 2 +- apps/kimi-web/src/api/daemon/wire.ts | 3 +- apps/kimi-web/src/api/daemon/ws.ts | 9 +- apps/kimi-web/test/ws-lifecycle.test.ts | 16 +- .../scripts/check-domain-layers.mjs | 5 +- .../agent/fileFencing/fileFencingService.ts | 49 +- packages/agent-core-v2/src/app/cron/clock.ts | 10 +- .../agent-core-v2/src/app/multiServer/flag.ts | 29 - .../sessionLifecycleService.ts | 21 +- .../src/os/interface/crossProcessLock.ts | 7 - .../backends/node-fs/fileStorageService.ts | 43 +- .../agentLifecycle/agentLifecycleService.ts | 2 +- .../src/session/sessionLease/sessionLease.ts | 42 +- .../sessionMetadata/sessionMetadataService.ts | 2 +- .../agent/fileFencing/fileFencing.test.ts | 620 ++++++++---------- .../sessionLifecycle/sessionLifecycle.test.ts | 40 +- .../node-fs/fileStorageService.test.ts | 58 +- .../agentLifecycle/agentLifecycle.test.ts | 1 + .../session/sessionLease/sessionLease.test.ts | 11 +- packages/kap-server/src/envelope.ts | 7 +- packages/kap-server/src/error-handler.ts | 12 +- packages/kap-server/src/protocol/envelope.ts | 22 + packages/kap-server/src/routes/fs.ts | 15 +- packages/kap-server/src/routes/prompts.ts | 17 +- packages/kap-server/src/routes/sessions.ts | 28 +- packages/kap-server/src/routes/terminals.ts | 15 +- .../sessionListWatchService.ts | 28 +- packages/kap-server/src/start.ts | 36 +- .../ws/v1/sessionEventBroadcaster.ts | 72 +- .../transport/ws/v1/sessionEventJournal.ts | 19 +- .../src/transport/ws/v1/wsConnectionV1.ts | 4 +- .../test/session-ownership.e2e.test.ts | 25 +- .../test/sessionEventBroadcaster.test.ts | 2 + .../kap-server/test/wsConnectionV1.test.ts | 1 + packages/kernel-file-lock/src/index.ts | 56 -- .../test/kernel-file-lock.test.ts | 72 -- packages/klient/AGENTS.md | 4 - .../test/e2e/harness/testing/serverPair.ts | 28 +- .../test/e2e/harness/testing/serverProcess.ts | 10 +- .../test/e2e/legacy/dual-instance.test.ts | 27 +- packages/minidb/src/index.ts | 7 - packages/minidb/src/lockfile.ts | 4 - packages/minidb/test/cluster/lock.test.ts | 2 +- packages/minidb/test/lock.test.ts | 2 +- packages/minidb/test/review-fixes.test.ts | 1 - packages/protocol/src/events.ts | 1 - 49 files changed, 579 insertions(+), 914 deletions(-) delete mode 100644 packages/agent-core-v2/src/app/multiServer/flag.ts diff --git a/.changeset/cross-process-lock-unification.md b/.changeset/cross-process-lock-unification.md index 2e0c62aa5..a9409cd48 100644 --- a/.changeset/cross-process-lock-unification.md +++ b/.changeset/cross-process-lock-unification.md @@ -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. diff --git a/.changeset/multi-instance-session-sync.md b/.changeset/multi-instance-session-sync.md index 4c4ecfb41..a546025b9 100644 --- a/.changeset/multi-instance-session-sync.md +++ b/.changeset/multi-instance-session-sync.md @@ -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. diff --git a/.changeset/stale-write-fencing.md b/.changeset/stale-write-fencing.md index 638b58ed6..b287b4773 100644 --- a/.changeset/stale-write-fencing.md +++ b/.changeset/stale-write-fencing.md @@ -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. diff --git a/apps/kimi-web/src/api/daemon/sessionOwnership.ts b/apps/kimi-web/src/api/daemon/sessionOwnership.ts index 4e429bc0f..b99b253a8 100644 --- a/apps/kimi-web/src/api/daemon/sessionOwnership.ts +++ b/apps/kimi-web/src/api/daemon/sessionOwnership.ts @@ -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 diff --git a/apps/kimi-web/src/api/daemon/wire.ts b/apps/kimi-web/src/api/daemon/wire.ts index bb87f3e3c..5971b2cb4 100644 --- a/apps/kimi-web/src/api/daemon/wire.ts +++ b/apps/kimi-web/src/api/daemon/wire.ts @@ -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; diff --git a/apps/kimi-web/src/api/daemon/ws.ts b/apps/kimi-web/src/api/daemon/ws.ts index eff16ce61..a2c1bc879 100644 --- a/apps/kimi-web/src/api/daemon/ws.ts +++ b/apps/kimi-web/src/api/daemon/ws.ts @@ -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; diff --git a/apps/kimi-web/test/ws-lifecycle.test.ts b/apps/kimi-web/test/ws-lifecycle.test.ts index 91d5edaa9..e50cab5fd 100644 --- a/apps/kimi-web/test/ws-lifecycle.test.ts +++ b/apps/kimi-web/test/ws-lifecycle.test.ts @@ -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); }); diff --git a/packages/agent-core-v2/scripts/check-domain-layers.mjs b/packages/agent-core-v2/scripts/check-domain-layers.mjs index fa98f97d6..f633b8b83 100644 --- a/packages/agent-core-v2/scripts/check-domain-layers.mjs +++ b/packages/agent-core-v2/scripts/check-domain-layers.mjs @@ -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], diff --git a/packages/agent-core-v2/src/agent/fileFencing/fileFencingService.ts b/packages/agent-core-v2/src/agent/fileFencing/fileFencingService.ts index aa7d5cb3a..261d6699e 100644 --- a/packages/agent-core-v2/src/agent/fileFencing/fileFencingService.ts +++ b/packages/agent-core-v2/src/agent/fileFencing/fileFencingService.ts @@ -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 `` 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 `Warning: ${body} Read the file to verify the current content.`; -} - -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(); - private readonly staleMarks = new Map(); 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)) }; - } } } diff --git a/packages/agent-core-v2/src/app/cron/clock.ts b/packages/agent-core-v2/src/app/cron/clock.ts index 43105a75e..2db4b26ed 100644 --- a/packages/agent-core-v2/src/app/cron/clock.ts +++ b/packages/agent-core-v2/src/app/cron/clock.ts @@ -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()`. diff --git a/packages/agent-core-v2/src/app/multiServer/flag.ts b/packages/agent-core-v2/src/app/multiServer/flag.ts deleted file mode 100644 index 22c7c1e6a..000000000 --- a/packages/agent-core-v2/src/app/multiServer/flag.ts +++ /dev/null @@ -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 - * `/server/instances/.json` instead of taking the legacy - * single-instance `/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); diff --git a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts index ba348bf9f..ea278bcb5 100644 --- a/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts +++ b/packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts @@ -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 { diff --git a/packages/agent-core-v2/src/os/interface/crossProcessLock.ts b/packages/agent-core-v2/src/os/interface/crossProcessLock.ts index b31c07bba..03c8210ad 100644 --- a/packages/agent-core-v2/src/os/interface/crossProcessLock.ts +++ b/packages/agent-core-v2/src/os/interface/crossProcessLock.ts @@ -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; diff --git a/packages/agent-core-v2/src/persistence/backends/node-fs/fileStorageService.ts b/packages/agent-core-v2/src/persistence/backends/node-fs/fileStorageService.ts index 6919bf3d6..028e66c97 100644 --- a/packages/agent-core-v2/src/persistence/backends/node-fs/fileStorageService.ts +++ b/packages/agent-core-v2/src/persistence/backends/node-fs/fileStorageService.ts @@ -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 { @@ -122,8 +131,14 @@ export class FileStorageService implements IFileSystemStorageService { _options: StorageWriteOptions = {}, ): Promise { 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 { 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 { 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(scope: string, key: string, op: () => Promise): Promise { 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 { if (this.syncedDirs.has(dir)) return; try { diff --git a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts index b03935dac..b8bc9e756 100644 --- a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts +++ b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts @@ -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(); diff --git a/packages/agent-core-v2/src/session/sessionLease/sessionLease.ts b/packages/agent-core-v2/src/session/sessionLease/sessionLease.ts index c2b5a2dad..4e5bcb367 100644 --- a/packages/agent-core-v2/src/session/sessionLease/sessionLease.ts +++ b/packages/agent-core-v2/src/session/sessionLease/sessionLease.ts @@ -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; diff --git a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts index d483ea9db..7324b8e83 100644 --- a/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts +++ b/packages/agent-core-v2/src/session/sessionMetadata/sessionMetadataService.ts @@ -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. * diff --git a/packages/agent-core-v2/test/agent/fileFencing/fileFencing.test.ts b/packages/agent-core-v2/test/agent/fileFencing/fileFencing.test.ts index 508f78a23..79aad012c 100644 --- a/packages/agent-core-v2/test/agent/fileFencing/fileFencing.test.ts +++ b/packages/agent-core-v2/test/agent/fileFencing/fileFencing.test.ts @@ -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 { - if (ctx.decision?.execute === undefined) return; - await ctx.decision.execute({ +async function runPrepared( + ctx: ToolBeforeExecuteContext, +): Promise { + 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 } = {}, ): Promise { - 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 { + 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('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: 'existing' }); - expect(did.result.note).toBe( - 'existing\n' + - `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.', - ); - }); + 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); }); }); diff --git a/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts b/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts index 4e3e93550..965b526cc 100644 --- a/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts +++ b/packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts @@ -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(); } diff --git a/packages/agent-core-v2/test/persistence/backends/node-fs/fileStorageService.test.ts b/packages/agent-core-v2/test/persistence/backends/node-fs/fileStorageService.test.ts index c17316477..63c8e6755 100644 --- a/packages/agent-core-v2/test/persistence/backends/node-fs/fileStorageService.test.ts +++ b/packages/agent-core-v2/test/persistence/backends/node-fs/fileStorageService.test.ts @@ -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(); + }); +}); diff --git a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts index 2e4b1270a..8fda9b5bc 100644 --- a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts +++ b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts @@ -347,6 +347,7 @@ describe('AgentLifecycleService', () => { ix.stub(IAgentTaskService, { _serviceBrand: undefined, stopAllOnExit, + flushPersistence: async () => {}, } as unknown as IAgentTaskService); ix.stub(IAgentFullCompactionService, { _serviceBrand: undefined, diff --git a/packages/agent-core-v2/test/session/sessionLease/sessionLease.test.ts b/packages/agent-core-v2/test/session/sessionLease/sessionLease.test.ts index 37bc759da..fe44c2c36 100644 --- a/packages/agent-core-v2/test/session/sessionLease/sessionLease.test.ts +++ b/packages/agent-core-v2/test/session/sessionLease/sessionLease.test.ts @@ -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[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); diff --git a/packages/kap-server/src/envelope.ts b/packages/kap-server/src/envelope.ts index 88f626cca..2611cb672 100644 --- a/packages/kap-server/src/envelope.ts +++ b/packages/kap-server/src/envelope.ts @@ -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'; diff --git a/packages/kap-server/src/error-handler.ts b/packages/kap-server/src/error-handler.ts index 2fdb5af3a..5225d4067 100644 --- a/packages/kap-server/src/error-handler.ts +++ b/packages/kap-server/src/error-handler.ts @@ -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'); diff --git a/packages/kap-server/src/protocol/envelope.ts b/packages/kap-server/src/protocol/envelope.ts index b28d2b09f..96361eec3 100644 --- a/packages/kap-server/src/protocol/envelope.ts +++ b/packages/kap-server/src/protocol/envelope.ts @@ -6,6 +6,8 @@ import { z } from 'zod'; +import { ErrorCode } from './error-codes'; + export const envelopeSchema = (data: T) => z.object({ code: z.number().int(), @@ -48,3 +50,23 @@ export function errEnvelope( ): Envelope { 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 { + return errEnvelope( + ErrorCode.SESSION_HELD_BY_PEER, + err.message, + requestId, + undefined, + err.details, + ); +} diff --git a/packages/kap-server/src/routes/fs.ts b/packages/kap-server/src/routes/fs.ts index 29aec21f0..6bf577fef 100644 --- a/packages/kap-server/src/routes/fs.ts +++ b/packages/kap-server/src/routes/fs.ts @@ -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 diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index ed757ff0c..e6f44fac7 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -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({ diff --git a/packages/kap-server/src/routes/sessions.ts b/packages/kap-server/src/routes/sessions.ts index 734b5b2e9..977e9ee2c 100644 --- a/packages/kap-server/src/routes/sessions.ts +++ b/packages/kap-server/src/routes/sessions.ts @@ -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)); diff --git a/packages/kap-server/src/routes/terminals.ts b/packages/kap-server/src/routes/terminals.ts index f07b3f389..4bc551c8e 100644 --- a/packages/kap-server/src/routes/terminals.ts +++ b/packages/kap-server/src/routes/terminals.ts @@ -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; } } diff --git a/packages/kap-server/src/services/sessionListWatch/sessionListWatchService.ts b/packages/kap-server/src/services/sessionListWatch/sessionListWatchService.ts index ca753dcf0..f68c86b59 100644 --- a/packages/kap-server/src/services/sessionListWatch/sessionListWatchService.ts +++ b/packages/kap-server/src/services/sessionListWatch/sessionListWatchService.ts @@ -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 - * `/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 `/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 `/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'; diff --git a/packages/kap-server/src/start.ts b/packages/kap-server/src/start.ts index e338c7bf8..89816dffe 100644 --- a/packages/kap-server/src/start.ts +++ b/packages/kap-server/src/start.ts @@ -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 { const host = opts.host ?? DEFAULT_HOST; const port = opts.port ?? DEFAULT_PORT; @@ -333,7 +317,7 @@ export async function startServer(opts: ServerStartOptions = {}): Promise + 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'), ); diff --git a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts index dceca5fcd..3e84f1ce5 100644 --- a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts +++ b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts @@ -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 { - 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; - } - - private async createGlobalState(): Promise { - 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 diff --git a/packages/kap-server/src/transport/ws/v1/sessionEventJournal.ts b/packages/kap-server/src/transport/ws/v1/sessionEventJournal.ts index 8a2c6a0da..91c21853a 100644 --- a/packages/kap-server/src/transport/ws/v1/sessionEventJournal.ts +++ b/packages/kap-server/src/transport/ws/v1/sessionEventJournal.ts @@ -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; diff --git a/packages/kap-server/src/transport/ws/v1/wsConnectionV1.ts b/packages/kap-server/src/transport/ws/v1/wsConnectionV1.ts index cf5c42fd1..3d383b906 100644 --- a/packages/kap-server/src/transport/ws/v1/wsConnectionV1.ts +++ b/packages/kap-server/src/transport/ws/v1/wsConnectionV1.ts @@ -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; diff --git a/packages/kap-server/test/session-ownership.e2e.test.ts b/packages/kap-server/test/session-ownership.e2e.test.ts index 83156280a..e50db2405 100644 --- a/packages/kap-server/test/session-ownership.e2e.test.ts +++ b/packages/kap-server/test/session-ownership.e2e.test.ts @@ -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 { 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; diff --git a/packages/kap-server/test/sessionEventBroadcaster.test.ts b/packages/kap-server/test/sessionEventBroadcaster.test.ts index 09feacdd0..4acd1b68b 100644 --- a/packages/kap-server/test/sessionEventBroadcaster.test.ts +++ b/packages/kap-server/test/sessionEventBroadcaster.test.ts @@ -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, }); diff --git a/packages/kap-server/test/wsConnectionV1.test.ts b/packages/kap-server/test/wsConnectionV1.test.ts index f14f966d3..c4f3971e0 100644 --- a/packages/kap-server/test/wsConnectionV1.test.ts +++ b/packages/kap-server/test/wsConnectionV1.test.ts @@ -62,6 +62,7 @@ function makeBroadcaster(): SessionEventBroadcaster { return { subscribe: async () => true, unsubscribe: () => {}, + getSubscriptionFailure: async () => undefined, getCursor: async () => ({ seq: 0, epoch: '' }), getBufferedSince: async () => ({ events: [], diff --git a/packages/kernel-file-lock/src/index.ts b/packages/kernel-file-lock/src/index.ts index c984d92bd..7810a5cc6 100644 --- a/packages/kernel-file-lock/src/index.ts +++ b/packages/kernel-file-lock/src/index.ts @@ -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 { - 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 { - 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'; - } -} diff --git a/packages/kernel-file-lock/test/kernel-file-lock.test.ts b/packages/kernel-file-lock/test/kernel-file-lock.test.ts index 2e286232d..d558532cc 100644 --- a/packages/kernel-file-lock/test/kernel-file-lock.test.ts +++ b/packages/kernel-file-lock/test/kernel-file-lock.test.ts @@ -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], { diff --git a/packages/klient/AGENTS.md b/packages/klient/AGENTS.md index 68f4712f0..bb2d805de 100644 --- a/packages/klient/AGENTS.md +++ b/packages/klient/AGENTS.md @@ -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 diff --git a/packages/klient/test/e2e/harness/testing/serverPair.ts b/packages/klient/test/e2e/harness/testing/serverPair.ts index 98ff9ae84..b0ace0ed3 100644 --- a/packages/klient/test/e2e/harness/testing/serverPair.ts +++ b/packages/klient/test/e2e/harness/testing/serverPair.ts @@ -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 = { ...options.env, [MULTI_SERVER_FLAG_ENV]: '1' }; + const envPatch: Record = { ...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; diff --git a/packages/klient/test/e2e/harness/testing/serverProcess.ts b/packages/klient/test/e2e/harness/testing/serverProcess.ts index cc887df18..2d6110ec0 100644 --- a/packages/klient/test/e2e/harness/testing/serverProcess.ts +++ b/packages/klient/test/e2e/harness/testing/serverProcess.ts @@ -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); diff --git a/packages/klient/test/e2e/legacy/dual-instance.test.ts b/packages/klient/test/e2e/legacy/dual-instance.test.ts index 3c6d1a93d..6014989ff 100644 --- a/packages/klient/test/e2e/legacy/dual-instance.test.ts +++ b/packages/klient/test/e2e/legacy/dual-instance.test.ts @@ -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); }, ); diff --git a/packages/minidb/src/index.ts b/packages/minidb/src/index.ts index 4e058e7e6..c18eebe9f 100644 --- a/packages/minidb/src/index.ts +++ b/packages/minidb/src/index.ts @@ -1608,13 +1608,6 @@ export class MiniDb { // ---- maintenance -------------------------------------------------------- - /** Verify that this process still holds the database's kernel lock. */ - async renewLock(): Promise { - 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 diff --git a/packages/minidb/src/lockfile.ts b/packages/minidb/src/lockfile.ts index b3046ca37..e9c97bf28 100644 --- a/packages/minidb/src/lockfile.ts +++ b/packages/minidb/src/lockfile.ts @@ -48,10 +48,6 @@ export class LockFile { if (!this.checkHeld()) throw new LockError(`database write lock was lost: ${this.path}`); } - async renew(): Promise { - this.assertHeld(); - } - async release(): Promise { this.releaseSync(); } diff --git a/packages/minidb/test/cluster/lock.test.ts b/packages/minidb/test/cluster/lock.test.ts index 57569cc17..3d6d4a477 100644 --- a/packages/minidb/test/cluster/lock.test.ts +++ b/packages/minidb/test/cluster/lock.test.ts @@ -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 }); diff --git a/packages/minidb/test/lock.test.ts b/packages/minidb/test/lock.test.ts index c597814c4..84207da29 100644 --- a/packages/minidb/test/lock.test.ts +++ b/packages/minidb/test/lock.test.ts @@ -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(); diff --git a/packages/minidb/test/review-fixes.test.ts b/packages/minidb/test/review-fixes.test.ts index 5a4746c22..313726d03 100644 --- a/packages/minidb/test/review-fixes.test.ts +++ b/packages/minidb/test/review-fixes.test.ts @@ -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/, diff --git a/packages/protocol/src/events.ts b/packages/protocol/src/events.ts index 77d23af97..888c873d7 100644 --- a/packages/protocol/src/events.ts +++ b/packages/protocol/src/events.ts @@ -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'