From 58eb07117b4d0286ccac63516886ffc0d1fe0979 Mon Sep 17 00:00:00 2001 From: jinye Date: Thu, 30 Jul 2026 07:36:58 +0800 Subject: [PATCH] fix(serve): Add certified session writer handoff (#7976) * fix(core): add certified session writer handoff Seal managed writer ownership after a durable recorder drain and allow trusted replacements to take over only when the transcript proof still matches. Co-authored-by: Qwen-Coder * fix(core): retain writer claim after failed rollback Keep the fixed transition claim unless the exact predecessor primary is restored, and avoid rollback after claim ownership changes or unlink completes with a reported error. Co-authored-by: Qwen-Coder * fix(core): close session handoff claim races Wait for claim-aware primary candidates to back out of transition gaps, and release losing claims when no primary transition started. Co-authored-by: Qwen-Coder * fix(core): bound session handoff candidate waits Fail closed when a claim-aware primary candidate is abandoned during a handoff transition, preserving the transition fence for authoritative recovery. Co-authored-by: Qwen-Coder * fix(core): reconcile session handoff proof edges Reject dangling transcript paths and reconcile ambiguous lock-record link outcomes by exact bytes before continuing a certified handoff. Co-authored-by: Qwen-Coder * fix(core): preserve inspect error cause and document handoff fail-closed paths (#7976) --------- Co-authored-by: Qwen-Coder Co-authored-by: qwen-code-dev-bot Co-authored-by: Shaojin Wen --- .../certified-session-writer-handoff.md | 220 ++++ .../cli/src/acp-integration/acpAgent.test.ts | 7 + packages/cli/src/acp-integration/acpAgent.ts | 3 +- packages/core/src/config/config.test.ts | 23 + packages/core/src/config/config.ts | 23 +- .../src/services/chatRecordingService.test.ts | 25 + .../core/src/services/chatRecordingService.ts | 29 +- .../src/services/session-writer-lease.test.ts | 1146 +++++++++++++++++ .../core/src/services/session-writer-lease.ts | 1081 +++++++++++++++- 9 files changed, 2513 insertions(+), 44 deletions(-) create mode 100644 docs/design/certified-session-writer-handoff.md diff --git a/docs/design/certified-session-writer-handoff.md b/docs/design/certified-session-writer-handoff.md new file mode 100644 index 00000000000..45ee8db2010 --- /dev/null +++ b/docs/design/certified-session-writer-handoff.md @@ -0,0 +1,220 @@ +# Certified session writer handoff + +## Problem + +Cooperative managed shutdown currently releases each session writer lock before +the ACP child exits. That fixes the ordinary replacement path, but it cannot +distinguish a writer that deliberately stopped recording from a foreign-host +writer that disappeared without releasing its active lock. Treating hostname, +PID visibility, lock age, or transcript inactivity as proof of death would +allow two live Pods to append to the same transcript. + +## Scope + +This change adds an integrity-protected handoff state for managed ACP writers. +After closing admission and durably draining accepted recorder work, a trusted +managed child can replace its active lock with a sealed record. A trusted +managed replacement can take ownership only after validating that record +against the exact transcript requested by the new Config. + +Transcript paths must be absent or resolve to the same regular file opened for +the proof. A dangling symlink is not treated as an absent transcript. + +The protocol remains gated by `experimental.sessionWriterLease`, which is +disabled by default and snapshotted at ACP process startup. Standalone ACP, +interactive, and headless recorders do not gain certified takeover. Normal +per-session close still releases its lock instead of leaving a sealed record. + +This change does not reclaim an active lock left by SIGKILL, an event-loop +stall, an unhandled crash, or a storage failure before sealing completes. It +does not add TTLs, heartbeats, hostname stealing, Kubernetes API lookups, +operator force-steal endpoints, maintenance leases, or mixed-version support. +Those active-lock cases still require an authoritative external writer fence +and explicit recovery. + +## Lock records + +New owners write schema v2 records. An active record retains the existing +immutable owner diagnostics and adds: + +```json +{ + "schema_version": 2, + "state": "active" +} +``` + +A sealed record retains the previous owner's diagnostics and adds: + +```json +{ + "schema_version": 2, + "state": "sealed", + "sealed_at": "2026-07-28T00:00:00.000Z", + "transcript": { + "relative_path": "", + "exists": true, + "byte_length": 1234, + "sha256": "" + } +} +``` + +The relative key must resolve to the transcript path already supplied by the +new Config. It is never used to select an arbitrary filesystem path. Schema v1 +records remain valid active records for compatibility during rollback, but +they can never be interpreted as sealed. + +## Fixed claim + +The fixed claim path is: + +```text +.claim +``` + +It serializes the two transitions that temporarily remove the primary path: +active-to-sealed and sealed-to-active. The claim is created with the existing +write-sync-and-hard-link primitive. Ambiguous link errors are reconciled +against the exact claim bytes before the transition continues or cleans up. A +claim is never reclaimed by PID, hostname, or age. Any pre-existing claim +returns `session_writer_unavailable`; manual cleanup is allowed only after an +authoritative external writer fence. + +Ordinary acquisition checks the claim before every attempt to install a +missing primary lock. It checks again after installation and removes only its +own exact candidate if a concurrent transition acquired the claim. This keeps +the current active-lock fast path and local stale-owner recovery while making +both paths respect a handoff transition. Mixed-version writers remain +unsupported because an older writer does not know about the fixed claim. + +An acquirer can pass its first claim check immediately before a transition +creates the claim, then install its active candidate during the transition's +primary-path gap. The transition recognizes that same-session schema-v2 active +record as a claim-aware candidate, preserves the retired predecessor, and +waits for the candidate's mandatory second check to remove it before retrying +the hard link. This wait is bounded; if the candidate stalls or exits before +its second check, the transition fails unavailable while retaining its claim +and retired predecessor. Unknown, malformed, or cross-session successors are +never removed or overwritten. + +## Sealing + +Managed shutdown synchronously stops session and recorder admission, then +starts all writer terminals in parallel. A writer terminal: + +1. drains every recorder operation accepted before the cutoff; +2. opens the expected transcript without following a symlink, verifies the + existing owner and transcript snapshot, and hashes the bytes through that + kept-open file descriptor; +3. writes and syncs an owner-unique sealed candidate; +4. acquires the fixed claim and revalidates the active owner plus the kept-open + transcript identity and metadata; +5. renames the exact active primary to an owner-unique retired path; +6. hard-links the sealed candidate into the now-missing primary path without + replacement; and +7. removes only its exact retired, candidate, and claim records. + +The primary-path transition is logically atomic for cooperating writers +because every installation respects the fixed claim, and the final hard link +cannot overwrite a lock created by another process. An error-after-effect is +reconciled from exact record bytes. The old owner never deletes or overwrites +an unknown primary. + +A managed flush or proof failure retains the active lock. Normal per-session +close preserves the existing release behavior, including exact-owner cleanup. +If a normal release races managed shutdown and commits first, the missing +primary is already a safe handoff and the replacement performs ordinary +acquisition. + +Failure cleanup removes the fixed claim only after proving that the exact +pre-transition primary was restored. If rollback cannot restore or verify that +record, the claim remains even when another primary appears, because that path +may be a losing ordinary-acquisition candidate that will remove itself after +observing the claim. Rollback itself is attempted only while the fixed claim +still contains this transition's exact record; a missing or replaced claim +means the current primary must not be changed. Recovery then requires the same +authoritative external writer fence as any other residual claim. + +Failures before the primary transition starts are different: the claimant has +not created a primary-path gap or moved the predecessor, so it releases only +its own exact claim even if another certified contender already replaced the +observed sealed primary. This prevents a delayed losing contender from +stranding a claim after the winner becomes active. + +## Certified takeover + +Only a Config created under a trusted managed parent enables certified +takeover. When acquisition observes a sealed record, it: + +1. verifies that the record's relative key matches the Config's expected + transcript; +2. opens and hashes that transcript outside the fixed claim, retaining the + file descriptor and its identity; +3. acquires the fixed claim; +4. re-reads the exact sealed primary and revalidates the kept-open descriptor, + path identity, metadata, byte length, and digest; +5. renames the sealed primary to a candidate-unique retired path; +6. hard-links the synced active candidate from the claim into the primary + path without replacement; and +7. removes only the exact retired record and its own claim. + +The lease then performs the existing authoritative session reload and +transcript fence before recorder activation. Two replacements racing the same +sealed record can produce at most one active owner. A loser receives a +conflict or unavailable result depending on whether it observes the winner's +active primary or an in-progress/residual claim. + +## Failure contract + +| Condition | Result | +| ---------------------------------------------------------------------------------- | ---------------------------------- | +| Valid active owner, including a foreign-host or managed dead-PID record | `session_writer_conflict` / 409 | +| Sealed proof does not match the expected transcript | `session_transcript_changed` / 409 | +| Malformed record, non-regular path, residual claim, or uncertain filesystem result | `session_writer_unavailable` / 503 | +| Current writer no longer owns its exact active record | `session_writer_lost` / 409 | + +Public errors remain sanitized. Successful seal and takeover logs include the +session ID, previous hostname/PID, and sealing time, but never the owner token +or transcript path. + +## Compatibility and rollout + +The feature gate must remain disabled during a mixed-version rollout. Enabling +or disabling it requires draining old ACP processes. A schema v2 reader still +accepts schema v1 active records, but an older reader does not understand +schema v2. Rollback therefore requires draining all new writers and confirming +that no active, sealed, or claim record from this protocol remains. + +A writer that does not acquire a lease — for example a plain `qwen --resume` +session, because standalone, interactive, and headless recorders run outside +this protocol — can still append to a transcript a managed writer has sealed. +That append invalidates the sealed proof, so a later certified takeover of the +same session fails closed with `session_transcript_changed` and the daemon +stays fenced until an authoritative external writer fence clears the residual +record. Rollout must therefore keep non-lease writers away from any transcript +that participates in certified handoff. + +Hashing is intentionally performed at seal and takeover instead of adding an +incremental digest to every append. This keeps the first implementation small +and makes the proof independent of process memory. A very large transcript may +cause managed sealing to exceed the parent deadline; that fails closed with +the active lock retained and is observable as an unclean shutdown. + +## Verification + +Unit and multiprocess coverage must prove: + +- schema v1 and v2 active locks keep their existing live/stale behavior; +- managed acquisition never reclaims an active dead-PID or foreign-host lock; +- successful sealing records the exact relative key, existence, byte length, + and SHA-256 digest; +- a certified replacement reloads the sealed transcript and becomes active; +- two replacements racing one sealed record elect exactly one owner; +- transcript modification, replacement, truncation, proof corruption, + malformed sealed records, and residual claims fail closed; +- failures before and after each primary-path transition never overwrite or + delete an unknown successor; +- managed flush failure retains the active primary; +- normal recorder close releases instead of sealing; and +- the default-off and standalone ACP paths remain unchanged. diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index c41e20a5a17..c371e3a6d1e 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -2075,6 +2075,12 @@ describe('QwenAgent MCP SSE/HTTP support', () => { expect(innerConfig.setSessionWriterReclaimPolicy).toHaveBeenCalledWith( 'never', ); + expect(innerConfig.setSessionWriterTakeoverPolicy).toHaveBeenCalledWith( + 'certified', + ); + expect(innerConfig.closeSessionWriter).toHaveBeenCalledWith({ + handoff: true, + }); expect(order).toEqual(['writer', 'resources']); expect(mockRunExitCleanup).toHaveBeenCalledOnce(); }); @@ -3016,6 +3022,7 @@ describe('QwenAgent MCP SSE/HTTP support', () => { shutdown: vi.fn().mockResolvedValue(undefined), closeSessionWriter: vi.fn().mockResolvedValue(undefined), setSessionWriterReclaimPolicy: vi.fn(), + setSessionWriterTakeoverPolicy: vi.fn(), waitForMcpReady: vi.fn().mockResolvedValue(undefined), getModelsConfig: vi.fn().mockReturnValue({ getCurrentAuthType: vi.fn().mockReturnValue('api-key'), diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index 75dc22caced..1448c27abb4 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -3460,7 +3460,7 @@ class QwenAgent implements Agent { const writerTerminals: Array> = []; for (const config of configList) { try { - writerTerminals.push(config.closeSessionWriter()); + writerTerminals.push(config.closeSessionWriter({ handoff: true })); } catch (error) { writerTerminals.push(Promise.reject(error)); } @@ -10795,6 +10795,7 @@ class QwenAgent implements Agent { this.assertManagedSessionAdmission(); if (this.isTrustedManagedParent()) { config.setSessionWriterReclaimPolicy('never'); + config.setSessionWriterTakeoverPolicy('certified'); } } catch (error) { return this.cleanupAfterRequestFailure(error, () => diff --git a/packages/core/src/config/config.test.ts b/packages/core/src/config/config.test.ts index 20351214950..5c049d26a7b 100644 --- a/packages/core/src/config/config.test.ts +++ b/packages/core/src/config/config.test.ts @@ -2772,6 +2772,29 @@ describe('Server Config (config.ts)', () => { }); describe('initialize', () => { + it('accepts managed handoff only after certified takeover is configured', async () => { + const standalone = new Config(baseParams); + await standalone.closeSessionWriter({ handoff: true }); + expect( + ( + standalone as unknown as { + sessionWriterHandoffRequested: boolean; + } + ).sessionWriterHandoffRequested, + ).toBe(false); + + const managed = new Config(baseParams); + managed.setSessionWriterTakeoverPolicy('certified'); + await managed.closeSessionWriter({ handoff: true }); + expect( + ( + managed as unknown as { + sessionWriterHandoffRequested: boolean; + } + ).sessionWriterHandoffRequested, + ).toBe(true); + }); + it.each([ [ 'an ACP session without an opt-in', diff --git a/packages/core/src/config/config.ts b/packages/core/src/config/config.ts index 798a01d0dec..68c013ca4b6 100644 --- a/packages/core/src/config/config.ts +++ b/packages/core/src/config/config.ts @@ -1668,7 +1668,9 @@ export class Config { private sessionProjectDirRegistered = false; private pendingSessionWriterLease?: SessionWriterLease; private sessionWriterReclaimPolicy: 'local' | 'never' = 'local'; + private sessionWriterTakeoverPolicy: 'never' | 'certified' = 'never'; private sessionWriterShutdownRequested = false; + private sessionWriterHandoffRequested = false; private sessionWriterActivationPromise: Promise | undefined; private sessionWriterClosePromise: Promise | undefined; /** @@ -3013,6 +3015,7 @@ export class Config { processKind: 'acp', qwenVersion: this.cliVersion ?? null, reclaimPolicy: this.sessionWriterReclaimPolicy, + takeoverPolicy: this.sessionWriterTakeoverPolicy, onOwnershipAcquired: (acquiredLease) => { lease = acquiredLease; this.pendingSessionWriterLease = acquiredLease; @@ -7074,9 +7077,21 @@ export class Config { this.sessionWriterReclaimPolicy = policy; } - closeSessionWriter(): Promise { + setSessionWriterTakeoverPolicy(policy: 'never' | 'certified'): void { + if (this.initialized) { + throw new SessionWriterUnavailableError(); + } + this.sessionWriterTakeoverPolicy = policy; + } + + closeSessionWriter(options?: { handoff?: boolean }): Promise { + if (options?.handoff && this.sessionWriterTakeoverPolicy === 'certified') { + this.sessionWriterHandoffRequested = true; + } this.sessionWriterShutdownRequested = true; - this.chatRecordingService?.beginClose(); + this.chatRecordingService?.beginClose({ + handoff: this.sessionWriterHandoffRequested, + }); this.sessionWriterClosePromise ??= this.closeSessionWriterOnce(); return this.sessionWriterClosePromise; } @@ -7091,7 +7106,9 @@ export class Config { } } try { - await this.chatRecordingService?.close(); + await this.chatRecordingService?.close({ + handoff: this.sessionWriterHandoffRequested, + }); } catch (error) { failures.push(error); } diff --git a/packages/core/src/services/chatRecordingService.test.ts b/packages/core/src/services/chatRecordingService.test.ts index b9d57209330..4f93ba4dda1 100644 --- a/packages/core/src/services/chatRecordingService.test.ts +++ b/packages/core/src/services/chatRecordingService.test.ts @@ -111,6 +111,7 @@ describe('ChatRecordingService', () => { ), assertOwnedAndUnchanged: vi.fn().mockResolvedValue(undefined), release: vi.fn().mockResolvedValue(undefined), + sealForHandoff: vi.fn().mockResolvedValue(undefined), } as unknown as SessionWriterLease; chatRecordingService = activateRecording( new ChatRecordingService(mockConfig), @@ -1880,6 +1881,30 @@ describe('ChatRecordingService', () => { }); describe('close', () => { + it('seals instead of releasing after a successful handoff drain', async () => { + chatRecordingService.recordUserMessage([{ text: 'durable' }]); + + await expect( + chatRecordingService.close({ handoff: true }), + ).resolves.toBeUndefined(); + expect(mockLease.sealForHandoff).toHaveBeenCalledOnce(); + expect(mockLease.release).not.toHaveBeenCalled(); + expect(chatRecordingService.hasWriteOwnership()).toBe(false); + }); + + it('retains ownership when a handoff drain fails', async () => { + const failure = new SessionTranscriptChangedError(); + vi.mocked(mockLease.appendJsonLine).mockRejectedValueOnce(failure); + chatRecordingService.recordUserMessage([{ text: 'not durable' }]); + + await expect(chatRecordingService.close({ handoff: true })).rejects.toBe( + failure, + ); + expect(mockLease.sealForHandoff).not.toHaveBeenCalled(); + expect(mockLease.release).not.toHaveBeenCalled(); + expect(chatRecordingService.hasWriteOwnership()).toBe(true); + }); + it('releases the writer lease before reporting a flush failure', async () => { const failure = new SessionTranscriptChangedError(); vi.mocked(mockLease.appendJsonLine).mockRejectedValueOnce(failure); diff --git a/packages/core/src/services/chatRecordingService.ts b/packages/core/src/services/chatRecordingService.ts index a02aafcb08f..706fbe8d626 100644 --- a/packages/core/src/services/chatRecordingService.ts +++ b/packages/core/src/services/chatRecordingService.ts @@ -611,6 +611,7 @@ export class ChatRecordingService { private operationTail: Promise = Promise.resolve(); private acceptingWrites = false; private closePromise: Promise | undefined; + private handoffRequested = false; /** First async JSONL write failure; permanently degrades this recorder. */ private writeFailure: Error | undefined; private integrityFailure: Error | undefined; @@ -1152,15 +1153,21 @@ export class ChatRecordingService { } } - close(): Promise { + close(options?: { handoff?: boolean }): Promise { + if (options?.handoff) { + this.handoffRequested = true; + } if (this.closePromise) return this.closePromise; if (this.state === 'closed') return Promise.resolve(); - this.beginClose(); + this.beginClose(options); this.closePromise = this.closeOnce(); return this.closePromise; } - beginClose(): void { + beginClose(options?: { handoff?: boolean }): void { + if (options?.handoff) { + this.handoffRequested = true; + } this.autoTitleController?.abort(); this.acceptingWrites = false; if (this.state === 'active') { @@ -1175,9 +1182,23 @@ export class ChatRecordingService { } catch (error) { flushFailure = error; } + if (this.handoffRequested && flushFailure !== undefined) { + // Fail closed: a handoff whose final flush did not durably land must + // neither seal (the proof would be incomplete) nor release (a successor + // could take over records that were never persisted). Unlike the + // release-then-throw normal-close path below, it retains the active lock + // so write ownership stays unambiguous until an external writer fence + // recovers it. + this.state = 'integrity_failed'; + throw flushFailure; + } const lease = this.binding?.lease; try { - await lease?.release(); + if (this.handoffRequested) { + await lease?.sealForHandoff(); + } else { + await lease?.release(); + } this.binding = undefined; this.state = 'closed'; } catch (error) { diff --git a/packages/core/src/services/session-writer-lease.test.ts b/packages/core/src/services/session-writer-lease.test.ts index 49fc5c2d1f3..ee3cc57385d 100644 --- a/packages/core/src/services/session-writer-lease.test.ts +++ b/packages/core/src/services/session-writer-lease.test.ts @@ -5,6 +5,7 @@ */ import { fork, type ChildProcess } from 'node:child_process'; +import { createHash } from 'node:crypto'; import { chmodSync, unlinkSync } from 'node:fs'; import * as fs from 'node:fs/promises'; import * as os from 'node:os'; @@ -42,6 +43,45 @@ const lstatFault = vi.hoisted(() => ({ calls: 0, })); +const transitionFault = vi.hoisted(() => ({ + renameFrom: undefined as string | undefined, + renameTo: undefined as string | undefined, + afterRename: undefined as (() => Promise) | undefined, + linkFrom: undefined as string | undefined, + linkTo: undefined as string | undefined, + afterLink: undefined as (() => Promise | void) | undefined, + throwAfterLink: false, +})); + +const restoreLinkFault = vi.hoisted(() => ({ + linkTo: undefined as string | undefined, + remainingFailures: 0, +})); + +const unlinkFault = vi.hoisted(() => ({ + path: undefined as string | undefined, + afterUnlink: undefined as (() => Promise | void) | undefined, + throwAfterUnlink: false, +})); + +const writeFault = vi.hoisted(() => ({ + contains: undefined as string | undefined, + onEntered: undefined as (() => void) | undefined, + wait: undefined as Promise | undefined, +})); + +const claimInstallFault = vi.hoisted(() => ({ + path: undefined as string | undefined, + afterInstall: undefined as (() => Promise | void) | undefined, +})); + +const readFileFault = vi.hoisted(() => ({ + path: undefined as string | undefined, + triggerCall: 0, + calls: 0, + afterRead: undefined as (() => Promise | void) | undefined, +})); + vi.mock('node:fs/promises', async (importOriginal) => { const actual = await importOriginal(); return { @@ -58,6 +98,89 @@ vi.mock('node:fs/promises', async (importOriginal) => { } return actual.lstat(filePath); }, + rename: async ( + oldPath: Parameters[0], + newPath: Parameters[1], + ) => { + await actual.rename(oldPath, newPath); + if ( + oldPath === transitionFault.renameFrom && + (transitionFault.renameTo === undefined || + newPath === transitionFault.renameTo) + ) { + await transitionFault.afterRename?.(); + } + }, + link: async ( + existingPath: Parameters[0], + newPath: Parameters[1], + ) => { + if ( + newPath === restoreLinkFault.linkTo && + restoreLinkFault.remainingFailures > 0 + ) { + restoreLinkFault.remainingFailures--; + throw Object.assign(new Error('injected restore link failure'), { + code: 'EIO', + }); + } + await actual.link(existingPath, newPath); + if (newPath === claimInstallFault.path) { + await claimInstallFault.afterInstall?.(); + } + if ( + existingPath === transitionFault.linkFrom && + newPath === transitionFault.linkTo + ) { + await transitionFault.afterLink?.(); + } + if ( + transitionFault.throwAfterLink && + existingPath === transitionFault.linkFrom && + newPath === transitionFault.linkTo + ) { + throw Object.assign(new Error('injected error after link'), { + code: 'EIO', + }); + } + }, + unlink: async (filePath: Parameters[0]) => { + await actual.unlink(filePath); + if (filePath === unlinkFault.path) { + await unlinkFault.afterUnlink?.(); + } + if (unlinkFault.throwAfterUnlink && filePath === unlinkFault.path) { + throw Object.assign(new Error('injected error after unlink'), { + code: 'EIO', + }); + } + }, + readFile: async (...args: Parameters) => { + const result = await actual.readFile(...args); + if (args[0] === readFileFault.path) { + readFileFault.calls++; + if (readFileFault.calls === readFileFault.triggerCall) { + await readFileFault.afterRead?.(); + } + } + return result; + }, + open: async (...args: Parameters) => { + const handle = await actual.open(...args); + const writeFile = handle.writeFile.bind(handle); + handle.writeFile = async (data, options) => { + if ( + writeFault.contains && + Buffer.isBuffer(data) && + data.toString('utf8').includes(writeFault.contains) + ) { + writeFault.onEntered?.(); + await writeFault.wait; + } + return writeFile(data, options); + }; + return handle; + }, }; }); @@ -162,6 +285,27 @@ afterEach(async () => { lstatFault.path = undefined; lstatFault.remainingFailures = 0; lstatFault.calls = 0; + transitionFault.renameFrom = undefined; + transitionFault.renameTo = undefined; + transitionFault.afterRename = undefined; + transitionFault.linkFrom = undefined; + transitionFault.linkTo = undefined; + transitionFault.afterLink = undefined; + transitionFault.throwAfterLink = false; + restoreLinkFault.linkTo = undefined; + restoreLinkFault.remainingFailures = 0; + unlinkFault.path = undefined; + unlinkFault.afterUnlink = undefined; + unlinkFault.throwAfterUnlink = false; + writeFault.contains = undefined; + writeFault.onEntered = undefined; + writeFault.wait = undefined; + claimInstallFault.path = undefined; + claimInstallFault.afterInstall = undefined; + readFileFault.path = undefined; + readFileFault.triggerCall = 0; + readFileFault.calls = 0; + readFileFault.afterRead = undefined; setDebugLogSession(null); resetDebugLoggingState(); Storage.setRuntimeBaseDir(null); @@ -280,6 +424,71 @@ describe('SessionWriterLease', () => { ).rejects.toMatchObject({ code: 'ENOENT' }); }); + it('hands a real managed ACP Config to a certified replacement', async () => { + const fixture = await createFixture('managed-config-handoff-session'); + const createConfig = () => + Storage.runWithRuntimeBaseDir( + fixture.runtimeBaseDir, + fixture.projectRoot, + () => + new Config({ + sessionId: fixture.options.sessionId, + cwd: fixture.projectRoot, + targetDir: fixture.projectRoot, + debugMode: false, + model: 'test-model', + chatRecording: true, + experimentalZedIntegration: true, + sessionWriterLeaseEnabled: true, + bareMode: true, + telemetry: { enabled: false }, + usageStatisticsEnabled: false, + }), + ); + const initialize = (config: Config) => + config.initialize({ + skipGeminiInitialization: true, + skipHooks: true, + skipMcpDiscovery: true, + skipSkillManager: true, + skipFileCheckpointing: true, + lenientToolWarmup: true, + }); + + const first = createConfig(); + first.setSessionWriterReclaimPolicy('never'); + first.setSessionWriterTakeoverPolicy('certified'); + await initialize(first); + first.getChatRecordingService()?.recordUserMessage('handoff tail'); + await first.closeSessionWriter({ handoff: true }); + expect( + JSON.parse( + await fs.readFile( + getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ), + 'utf8', + ), + ), + ).toMatchObject({ schema_version: 2, state: 'sealed' }); + + const replacement = createConfig(); + replacement.setSessionWriterReclaimPolicy('never'); + replacement.setSessionWriterTakeoverPolicy('certified'); + await initialize(replacement); + expect( + replacement.getResumedSessionData()?.conversation.messages.at(-1)?.message + ?.parts, + ).toEqual([{ text: 'handoff tail' }]); + + await first.shutdown({ + shutdownTelemetry: false, + skipSessionWriter: true, + }); + await replacement.shutdown({ shutdownTelemetry: false }); + }); + it('restores and re-anchors a persisted title outside the active UUID chain', async () => { const fixture = await createFixture('11111111-1111-4111-8111-111111111111'); const firstUser = record( @@ -705,6 +914,57 @@ describe('SessionWriterLease', () => { ).rejects.toMatchObject({ code: 'ENOENT' }); }); + it('rejects a dangling transcript symlink introduced before sealing', async () => { + const fixture = await createFixture('dangling-seal-session'); + const lease = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const activeRaw = await fs.readFile(lockPath, 'utf8'); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.symlink( + `${fixture.transcriptPath}.missing`, + fixture.transcriptPath, + ); + + await expect(lease.sealForHandoff()).rejects.toBeInstanceOf( + SessionWriterUnavailableError, + ); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(activeRaw); + await expect(fs.lstat(`${lockPath}.claim`)).rejects.toMatchObject({ + code: 'ENOENT', + }); + }); + + it('rejects a dangling transcript symlink before certified takeover', async () => { + const fixture = await createFixture('dangling-takeover-session'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const sealedRaw = await fs.readFile(lockPath, 'utf8'); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.symlink( + `${fixture.transcriptPath}.missing`, + fixture.transcriptPath, + ); + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionWriterUnavailableError); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(sealedRaw); + await expect(fs.lstat(`${lockPath}.claim`)).rejects.toMatchObject({ + code: 'ENOENT', + }); + }); + it('detects an equal-length atomic transcript replacement', async () => { const fixture = await createFixture(); await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); @@ -814,6 +1074,892 @@ describe('SessionWriterLease', () => { ).rejects.toBeInstanceOf(SessionWriterConflictError); }); + it('never treats a foreign-host active record as a certified handoff', async () => { + const fixture = await createFixture('foreign-active-session'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const active = JSON.parse(await fs.readFile(lockPath, 'utf8')) as Record< + string, + unknown + >; + await first.release(); + await fs.writeFile( + lockPath, + JSON.stringify({ + ...active, + hostname: 'retired-foreign-host', + pid: 2_147_483_647, + }), + ); + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionWriterConflictError); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + state: 'active', + hostname: 'retired-foreign-host', + }); + }); + + it('keeps schema v1 records on the active-owner path', async () => { + const fixture = await createFixture('legacy-active-session'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const active = JSON.parse(await fs.readFile(lockPath, 'utf8')) as Record< + string, + unknown + >; + await first.release(); + delete active['state']; + active['schema_version'] = 1; + await fs.writeFile(lockPath, JSON.stringify(active)); + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionWriterConflictError); + + await fs.writeFile( + lockPath, + JSON.stringify({ + ...active, + pid: 2_147_483_647, + }), + ); + const replacement = await SessionWriterLease.acquire(fixture.options); + await replacement.release(); + }); + + it('seals a transcript proof and permits only certified takeover', async () => { + const fixture = await createFixture('sealed-session'); + const initial = `${JSON.stringify({ record: 'initial' })}\n`; + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, initial); + const first = await SessionWriterLease.acquire(fixture.options); + await first.appendJsonLine({ record: 'final' }); + const expectedTranscript = await fs.readFile(fixture.transcriptPath); + + await first.sealForHandoff(); + + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const sealed = JSON.parse(await fs.readFile(lockPath, 'utf8')) as { + schema_version: number; + state: string; + transcript: { + relative_path: string; + exists: boolean; + byte_length: number; + sha256: string; + }; + }; + expect(sealed).toMatchObject({ + schema_version: 2, + state: 'sealed', + transcript: { + relative_path: path + .relative(fixture.runtimeBaseDir, fixture.transcriptPath) + .split(path.sep) + .join('/'), + exists: true, + byte_length: expectedTranscript.byteLength, + sha256: createHash('sha256').update(expectedTranscript).digest('hex'), + }, + }); + await expect( + first.appendJsonLine({ record: 'too-late' }), + ).rejects.toBeInstanceOf(SessionWriterLostError); + await expect(fs.readFile(fixture.transcriptPath)).resolves.toEqual( + expectedTranscript, + ); + await expect( + SessionWriterLease.acquire(fixture.options), + ).rejects.toBeInstanceOf(SessionWriterConflictError); + + const replacement = await SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }); + expect(replacement.ownerId).not.toBe(first.ownerId); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + schema_version: 2, + state: 'active', + owner_id: replacement.ownerId, + }); + await replacement.assertOwnedAndUnchanged(); + await replacement.release(); + }); + + it('waits for an accepted append before sealing the transcript', async () => { + const fixture = await createFixture('sealed-append-race-session'); + const lease = await SessionWriterLease.acquire(fixture.options); + let resumeWrite: (() => void) | undefined; + writeFault.contains = '"late":true'; + writeFault.wait = new Promise((resolve) => { + resumeWrite = resolve; + }); + const writeEntered = new Promise((resolve) => { + writeFault.onEntered = resolve; + }); + + const append = lease.appendJsonLine({ late: true }); + await writeEntered; + const seal = lease.sealForHandoff(); + await expect( + Promise.race([ + seal.then( + () => 'settled', + () => 'settled', + ), + new Promise<'pending'>((resolve) => + setTimeout(() => resolve('pending'), 50), + ), + ]), + ).resolves.toBe('pending'); + + resumeWrite?.(); + await append; + await seal; + const transcript = await fs.readFile(fixture.transcriptPath); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const sealed = JSON.parse(await fs.readFile(lockPath, 'utf8')) as { + transcript: { byte_length: number; sha256: string }; + }; + expect(transcript.toString('utf8')).toBe('{"late":true}\n'); + expect(sealed.transcript).toMatchObject({ + byte_length: transcript.byteLength, + sha256: createHash('sha256').update(transcript).digest('hex'), + }); + }); + + it('reconciles a sealing error reported after the sealed primary is installed', async () => { + const fixture = await createFixture('sealed-after-effect-session'); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, '{"record":"tail"}\n'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + transitionFault.linkFrom = `${lockPath}.sealed-candidate.${encodeURIComponent( + first.ownerId, + )}`; + transitionFault.linkTo = lockPath; + transitionFault.throwAfterLink = true; + + await expect(first.sealForHandoff()).resolves.toBeUndefined(); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + schema_version: 2, + state: 'sealed', + }); + }); + + it('reconciles a sealing claim link error after effect', async () => { + const fixture = await createFixture('sealed-claim-link-session'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + claimInstallFault.path = `${lockPath}.claim`; + claimInstallFault.afterInstall = () => { + throw Object.assign(new Error('injected error after claim link'), { + code: 'EIO', + }); + }; + + await expect(first.sealForHandoff()).resolves.toBeUndefined(); + await expect(fs.lstat(`${lockPath}.claim`)).rejects.toMatchObject({ + code: 'ENOENT', + }); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + state: 'sealed', + }); + }); + + it('reconciles a sealing claim unlink error after effect', async () => { + const fixture = await createFixture('sealed-claim-unlink-session'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + unlinkFault.path = `${lockPath}.claim`; + unlinkFault.throwAfterUnlink = true; + + await expect(first.sealForHandoff()).resolves.toBeUndefined(); + await expect(fs.lstat(`${lockPath}.claim`)).rejects.toMatchObject({ + code: 'ENOENT', + }); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + state: 'sealed', + }); + }); + + it('does not roll back after the released claim is replaced', async () => { + const fixture = await createFixture('sealed-replaced-claim-session'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const successorClaim = '{"successorClaim":true}'; + unlinkFault.path = `${lockPath}.claim`; + unlinkFault.afterUnlink = () => + fs.writeFile(`${lockPath}.claim`, successorClaim, 'utf8'); + unlinkFault.throwAfterUnlink = true; + + await expect(first.sealForHandoff()).resolves.toBeUndefined(); + await expect(fs.readFile(`${lockPath}.claim`, 'utf8')).resolves.toBe( + successorClaim, + ); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + state: 'sealed', + }); + }); + + it('does not roll back sealing after claim ownership changes', async () => { + const fixture = await createFixture('sealed-changed-claim-session'); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, '{"record":"tail"}\n'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const successorClaim = '{"successorClaim":true}'; + transitionFault.linkFrom = `${lockPath}.sealed-candidate.${encodeURIComponent( + first.ownerId, + )}`; + transitionFault.linkTo = lockPath; + transitionFault.afterLink = async () => { + await fs.unlink(`${lockPath}.claim`); + await fs.writeFile(`${lockPath}.claim`, successorClaim, 'utf8'); + lstatFault.path = fixture.transcriptPath; + lstatFault.remainingFailures = 1; + }; + + await expect(first.sealForHandoff()).rejects.toBeInstanceOf( + SessionWriterUnavailableError, + ); + await expect(fs.readFile(`${lockPath}.claim`, 'utf8')).resolves.toBe( + successorClaim, + ); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + state: 'sealed', + }); + }); + + it('retains the claim when sealing rollback cannot restore the primary', async () => { + const fixture = await createFixture('sealed-rollback-failure-session'); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, '{"record":"tail"}\n'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const activeRaw = await fs.readFile(lockPath, 'utf8'); + transitionFault.linkFrom = `${lockPath}.sealed-candidate.${encodeURIComponent( + first.ownerId, + )}`; + transitionFault.linkTo = lockPath; + transitionFault.afterLink = () => { + lstatFault.path = fixture.transcriptPath; + lstatFault.remainingFailures = 1; + restoreLinkFault.linkTo = lockPath; + restoreLinkFault.remainingFailures = 1; + }; + + await expect(first.sealForHandoff()).rejects.toBeInstanceOf( + SessionWriterUnavailableError, + ); + await expect(fs.lstat(lockPath)).rejects.toMatchObject({ code: 'ENOENT' }); + await expect(fs.readFile(`${lockPath}.claim`, 'utf8')).resolves.toBe( + activeRaw, + ); + await expect( + fs.readFile( + `${lockPath}.handoff.${encodeURIComponent(first.ownerId)}`, + 'utf8', + ), + ).resolves.toBe(activeRaw); + await fs.appendFile(fixture.transcriptPath, '{"external":true}\n'); + await expect( + SessionWriterLease.acquire(fixture.options), + ).rejects.toBeInstanceOf(SessionWriterUnavailableError); + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionWriterUnavailableError); + }); + + it('removes the claim after sealing rollback restores the primary', async () => { + const fixture = await createFixture('sealed-rollback-success-session'); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, '{"record":"tail"}\n'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const activeRaw = await fs.readFile(lockPath, 'utf8'); + transitionFault.linkFrom = `${lockPath}.sealed-candidate.${encodeURIComponent( + first.ownerId, + )}`; + transitionFault.linkTo = lockPath; + transitionFault.afterLink = () => { + lstatFault.path = fixture.transcriptPath; + lstatFault.remainingFailures = 1; + }; + + await expect(first.sealForHandoff()).rejects.toBeInstanceOf( + SessionWriterUnavailableError, + ); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(activeRaw); + await expect(fs.lstat(`${lockPath}.claim`)).rejects.toMatchObject({ + code: 'ENOENT', + }); + }); + + it('waits for a claim-aware primary candidate before completing sealing', async () => { + const fixture = await createFixture('sealed-primary-candidate-session'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const active = JSON.parse(await fs.readFile(lockPath, 'utf8')) as Record< + string, + unknown + >; + const candidateRaw = JSON.stringify({ + ...active, + owner_id: 'claim-aware-candidate', + }); + transitionFault.renameFrom = lockPath; + transitionFault.renameTo = `${lockPath}.handoff.${encodeURIComponent( + first.ownerId, + )}`; + transitionFault.afterRename = async () => { + await fs.writeFile(lockPath, candidateRaw, 'utf8'); + readFileFault.path = lockPath; + readFileFault.triggerCall = 2; + readFileFault.afterRead = () => fs.unlink(lockPath); + }; + + await expect(first.sealForHandoff()).resolves.toBeUndefined(); + expect(readFileFault.calls).toBeGreaterThanOrEqual(2); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + state: 'sealed', + owner_id: first.ownerId, + }); + await expect(fs.lstat(`${lockPath}.claim`)).rejects.toMatchObject({ + code: 'ENOENT', + }); + }); + + it('fails closed when a primary candidate is abandoned during sealing', async () => { + const fixture = await createFixture('sealed-abandoned-candidate-session'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const activeRaw = await fs.readFile(lockPath, 'utf8'); + const candidateRaw = JSON.stringify({ + ...(JSON.parse(activeRaw) as Record), + owner_id: 'abandoned-candidate', + }); + const retiredPath = `${lockPath}.handoff.${encodeURIComponent( + first.ownerId, + )}`; + transitionFault.renameFrom = lockPath; + transitionFault.renameTo = retiredPath; + transitionFault.afterRename = () => + fs.writeFile(lockPath, candidateRaw, 'utf8'); + + await expect(first.sealForHandoff()).rejects.toBeInstanceOf( + SessionWriterUnavailableError, + ); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(candidateRaw); + await expect(fs.readFile(`${lockPath}.claim`, 'utf8')).resolves.toBe( + activeRaw, + ); + await expect(fs.readFile(retiredPath, 'utf8')).resolves.toBe(activeRaw); + }); + + it('waits for a claim-aware primary candidate while rolling back sealing', async () => { + const fixture = await createFixture('sealed-rollback-candidate-session'); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, '{"record":"tail"}\n'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const activeRaw = await fs.readFile(lockPath, 'utf8'); + const active = JSON.parse(activeRaw) as Record; + const candidateRaw = JSON.stringify({ + ...active, + owner_id: 'rollback-candidate', + }); + transitionFault.linkFrom = `${lockPath}.sealed-candidate.${encodeURIComponent( + first.ownerId, + )}`; + transitionFault.linkTo = lockPath; + transitionFault.afterLink = () => { + lstatFault.path = fixture.transcriptPath; + lstatFault.remainingFailures = 1; + unlinkFault.path = lockPath; + unlinkFault.afterUnlink = async () => { + unlinkFault.path = undefined; + await fs.writeFile(lockPath, candidateRaw, 'utf8'); + readFileFault.path = lockPath; + readFileFault.triggerCall = 2; + readFileFault.afterRead = () => fs.unlink(lockPath); + }; + }; + + await expect(first.sealForHandoff()).rejects.toBeInstanceOf( + SessionWriterUnavailableError, + ); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(activeRaw); + await expect(fs.lstat(`${lockPath}.claim`)).rejects.toMatchObject({ + code: 'ENOENT', + }); + await expect( + fs.lstat(`${lockPath}.handoff.${encodeURIComponent(first.ownerId)}`), + ).rejects.toMatchObject({ code: 'ENOENT' }); + }); + + it('fails closed when a primary candidate is abandoned during rollback', async () => { + const fixture = await createFixture( + 'sealed-rollback-abandoned-candidate-session', + ); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, '{"record":"tail"}\n'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const activeRaw = await fs.readFile(lockPath, 'utf8'); + const candidateRaw = JSON.stringify({ + ...(JSON.parse(activeRaw) as Record), + owner_id: 'abandoned-rollback-candidate', + }); + const retiredPath = `${lockPath}.handoff.${encodeURIComponent( + first.ownerId, + )}`; + transitionFault.linkFrom = `${lockPath}.sealed-candidate.${encodeURIComponent( + first.ownerId, + )}`; + transitionFault.linkTo = lockPath; + transitionFault.afterLink = () => { + lstatFault.path = fixture.transcriptPath; + lstatFault.remainingFailures = 1; + unlinkFault.path = lockPath; + unlinkFault.throwAfterUnlink = true; + unlinkFault.afterUnlink = async () => { + unlinkFault.path = undefined; + await fs.writeFile(lockPath, candidateRaw, 'utf8'); + }; + }; + + await expect(first.sealForHandoff()).rejects.toBeInstanceOf( + SessionWriterUnavailableError, + ); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(candidateRaw); + await expect(fs.readFile(`${lockPath}.claim`, 'utf8')).resolves.toBe( + activeRaw, + ); + await expect(fs.readFile(retiredPath, 'utf8')).resolves.toBe(activeRaw); + }); + + it('never overwrites a primary installed during the sealing transition', async () => { + const fixture = await createFixture('sealed-successor-session'); + const first = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const successorRaw = '{"successor":true}'; + transitionFault.renameFrom = lockPath; + transitionFault.renameTo = `${lockPath}.handoff.${encodeURIComponent( + first.ownerId, + )}`; + transitionFault.afterRename = () => + fs.writeFile(lockPath, successorRaw, 'utf8'); + + await expect(first.sealForHandoff()).rejects.toBeInstanceOf( + SessionWriterUnavailableError, + ); + expect(first.isReleased).toBe(true); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(successorRaw); + await expect(fs.lstat(`${lockPath}.claim`)).resolves.toBeDefined(); + }); + + it('elects exactly one certified replacement for a sealed session', async () => { + const fixture = await createFixture('sealed-race-session'); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, '{"record":"tail"}\n'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + + const contenders = [startLeaseProcess(), startLeaseProcess()]; + const options = { + ...fixture.options, + reclaimPolicy: 'never' as const, + takeoverPolicy: 'certified' as const, + }; + const results = await Promise.all( + contenders.map((child) => + requestChild(child, { type: 'acquire', options }), + ), + ); + expect(results.filter((result) => result.ok)).toHaveLength(1); + const winner = contenders[results.findIndex((result) => result.ok)]!; + expect(await requestChild(winner, { type: 'release' })).toMatchObject({ + ok: true, + }); + }); + + it('releases a losing takeover claim before its transition starts', async () => { + const fixture = await createFixture( + 'takeover-pre-transition-loser-session', + ); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + let winnerRaw = ''; + claimInstallFault.path = `${lockPath}.claim`; + claimInstallFault.afterInstall = async () => { + const contender = JSON.parse( + await fs.readFile(`${lockPath}.claim`, 'utf8'), + ) as Record; + winnerRaw = JSON.stringify({ + ...contender, + owner_id: 'certified-winner', + }); + await fs.unlink(lockPath); + await fs.writeFile(lockPath, winnerRaw, 'utf8'); + }; + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionWriterConflictError); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(winnerRaw); + await expect(fs.lstat(`${lockPath}.claim`)).rejects.toMatchObject({ + code: 'ENOENT', + }); + }); + + it('reconciles a takeover error reported after the active primary is installed', async () => { + const fixture = await createFixture('takeover-after-effect-session'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + transitionFault.linkFrom = `${lockPath}.claim`; + transitionFault.linkTo = lockPath; + transitionFault.throwAfterLink = true; + + const replacement = await SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + schema_version: 2, + state: 'active', + owner_id: replacement.ownerId, + }); + await replacement.release(); + }); + + it('reconciles a takeover claim link error after effect', async () => { + const fixture = await createFixture('takeover-claim-link-session'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + claimInstallFault.path = `${lockPath}.claim`; + claimInstallFault.afterInstall = () => { + throw Object.assign(new Error('injected error after claim link'), { + code: 'EIO', + }); + }; + + const replacement = await SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }); + await expect(fs.lstat(`${lockPath}.claim`)).rejects.toMatchObject({ + code: 'ENOENT', + }); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + state: 'active', + owner_id: replacement.ownerId, + }); + await replacement.release(); + }); + + it('reconciles a takeover claim unlink error after effect', async () => { + const fixture = await createFixture('takeover-claim-unlink-session'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + unlinkFault.path = `${lockPath}.claim`; + unlinkFault.throwAfterUnlink = true; + + const replacement = await SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }); + await expect(fs.lstat(`${lockPath}.claim`)).rejects.toMatchObject({ + code: 'ENOENT', + }); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + state: 'active', + owner_id: replacement.ownerId, + }); + await replacement.release(); + }); + + it('does not roll back takeover after claim ownership changes', async () => { + const fixture = await createFixture('takeover-changed-claim-session'); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, '{"record":"tail"}\n'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const successorClaim = '{"successorClaim":true}'; + transitionFault.linkFrom = `${lockPath}.claim`; + transitionFault.linkTo = lockPath; + transitionFault.afterLink = async () => { + await fs.unlink(`${lockPath}.claim`); + await fs.writeFile(`${lockPath}.claim`, successorClaim, 'utf8'); + lstatFault.path = fixture.transcriptPath; + lstatFault.remainingFailures = 1; + }; + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionWriterUnavailableError); + await expect(fs.readFile(`${lockPath}.claim`, 'utf8')).resolves.toBe( + successorClaim, + ); + expect(JSON.parse(await fs.readFile(lockPath, 'utf8'))).toMatchObject({ + state: 'active', + }); + }); + + it('retains the claim when takeover rollback cannot restore the primary', async () => { + const fixture = await createFixture('takeover-rollback-failure-session'); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, '{"record":"tail"}\n'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const sealedRaw = await fs.readFile(lockPath, 'utf8'); + transitionFault.linkFrom = `${lockPath}.claim`; + transitionFault.linkTo = lockPath; + transitionFault.afterLink = () => { + lstatFault.path = fixture.transcriptPath; + lstatFault.remainingFailures = 1; + restoreLinkFault.linkTo = lockPath; + restoreLinkFault.remainingFailures = 1; + }; + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionWriterUnavailableError); + await expect(fs.lstat(lockPath)).rejects.toMatchObject({ code: 'ENOENT' }); + const claimRaw = await fs.readFile(`${lockPath}.claim`, 'utf8'); + const claim = JSON.parse(claimRaw) as { owner_id: string; state: string }; + expect(claim.state).toBe('active'); + await expect( + fs.readFile( + `${lockPath}.sealed.${encodeURIComponent( + first.ownerId, + )}.${encodeURIComponent(claim.owner_id)}`, + 'utf8', + ), + ).resolves.toBe(sealedRaw); + await fs.appendFile(fixture.transcriptPath, '{"external":true}\n'); + await expect( + SessionWriterLease.acquire(fixture.options), + ).rejects.toBeInstanceOf(SessionWriterUnavailableError); + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionWriterUnavailableError); + }); + + it('never overwrites a primary installed during the takeover transition', async () => { + const fixture = await createFixture('takeover-successor-session'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const successorRaw = '{"successor":true}'; + transitionFault.renameFrom = lockPath; + transitionFault.afterRename = () => + fs.writeFile(lockPath, successorRaw, 'utf8'); + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionWriterUnavailableError); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(successorRaw); + await expect(fs.lstat(`${lockPath}.claim`)).resolves.toBeDefined(); + }); + + it.each(['append', 'truncate', 'replace'] as const)( + 'retains a sealed lock when the transcript proof changes by %s', + async (mutation) => { + const fixture = await createFixture(`sealed-${mutation}-session`); + await fs.mkdir(path.dirname(fixture.transcriptPath), { recursive: true }); + await fs.writeFile(fixture.transcriptPath, '{"record":"tail"}\n'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const sealedRaw = await fs.readFile(lockPath, 'utf8'); + if (mutation === 'append') { + await fs.appendFile(fixture.transcriptPath, '{"external":true}\n'); + } else if (mutation === 'truncate') { + await fs.truncate(fixture.transcriptPath, 0); + } else { + const replacementPath = `${fixture.transcriptPath}.replacement`; + await fs.writeFile(replacementPath, '{"record":"evil"}\n'); + await fs.rename(replacementPath, fixture.transcriptPath); + } + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionTranscriptChangedError); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(sealedRaw); + }, + ); + + it.each([ + ['valid but mismatched', '0'.repeat(64), SessionTranscriptChangedError], + ['malformed', 'invalid', SessionWriterUnavailableError], + ])( + 'retains a sealed primary with a %s transcript digest', + async (_description, sha256, ErrorType) => { + const fixture = await createFixture('sealed-proof-session'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const sealed = JSON.parse(await fs.readFile(lockPath, 'utf8')) as { + transcript: { sha256: string }; + }; + sealed.transcript.sha256 = sha256; + const sealedRaw = JSON.stringify(sealed); + await fs.writeFile(lockPath, sealedRaw); + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(ErrorType); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(sealedRaw); + }, + ); + + it('fails closed without changing a sealed primary when a claim remains', async () => { + const fixture = await createFixture('sealed-claim-session'); + const first = await SessionWriterLease.acquire(fixture.options); + await first.sealForHandoff(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const sealedRaw = await fs.readFile(lockPath, 'utf8'); + await fs.writeFile(`${lockPath}.claim`, '{"residual":true}'); + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + reclaimPolicy: 'never', + takeoverPolicy: 'certified', + }), + ).rejects.toBeInstanceOf(SessionWriterUnavailableError); + await expect(fs.readFile(lockPath, 'utf8')).resolves.toBe(sealedRaw); + }); + it('cannot remove a successor lock after release commits', async () => { const fixture = await createFixture(); const first = await SessionWriterLease.acquire(fixture.options); diff --git a/packages/core/src/services/session-writer-lease.ts b/packages/core/src/services/session-writer-lease.ts index 96bd736d04c..2a43dad890b 100644 --- a/packages/core/src/services/session-writer-lease.ts +++ b/packages/core/src/services/session-writer-lease.ts @@ -5,16 +5,18 @@ */ import { execFile } from 'node:child_process'; -import { randomUUID } from 'node:crypto'; +import { createHash, randomUUID } from 'node:crypto'; import type { Stats } from 'node:fs'; import * as fs from 'node:fs/promises'; import * as os from 'node:os'; import * as path from 'node:path'; import { createDebugLogger } from '../utils/debugLogger.js'; -const LOCK_SCHEMA_VERSION = 1; +const LEGACY_LOCK_SCHEMA_VERSION = 1; +const LOCK_SCHEMA_VERSION = 2; const MALFORMED_RETRY_COUNT = 3; const MALFORMED_RETRY_DELAY_MS = 50; +const CLAIMED_PRIMARY_WAIT_ATTEMPTS = 20; const RELEASE_PRECHECK_ATTEMPTS = 3; const RELEASE_PRECHECK_RETRY_DELAY_MS = 50; const ACQUIRE_ATTEMPTS = 8; @@ -102,8 +104,7 @@ export class SessionWriterUnavailableError extends SessionWriterError { } } -interface SessionWriterLockRecord { - schema_version: number; +interface SessionWriterOwnerRecord { session_id: string; owner_id: string; pid: number; @@ -114,6 +115,38 @@ interface SessionWriterLockRecord { qwen_version: string | null; } +interface LegacySessionWriterLockRecord extends SessionWriterOwnerRecord { + schema_version: typeof LEGACY_LOCK_SCHEMA_VERSION; +} + +interface ActiveSessionWriterLockRecord extends SessionWriterOwnerRecord { + schema_version: typeof LOCK_SCHEMA_VERSION; + state: 'active'; +} + +interface SealedTranscriptProof { + relative_path: string; + exists: boolean; + byte_length: number; + sha256: string; +} + +interface SealedSessionWriterLockRecord extends SessionWriterOwnerRecord { + schema_version: typeof LOCK_SCHEMA_VERSION; + state: 'sealed'; + sealed_at: string; + transcript: SealedTranscriptProof; +} + +type SessionWriterLockRecord = + | LegacySessionWriterLockRecord + | ActiveSessionWriterLockRecord + | SealedSessionWriterLockRecord; + +type ActiveLockRecord = + | LegacySessionWriterLockRecord + | ActiveSessionWriterLockRecord; + export interface AcquireSessionWriterLeaseOptions { runtimeBaseDir: string; sessionId: string; @@ -121,13 +154,15 @@ export interface AcquireSessionWriterLeaseOptions { processKind?: SessionWriterProcessKind; qwenVersion?: string | null; reclaimPolicy?: 'local' | 'never'; + takeoverPolicy?: 'never' | 'certified'; onOwnershipAcquired?: (lease: SessionWriterLease) => void; } type ExistingLockState = | { kind: 'missing' } - | { kind: 'live' } - | { kind: 'stale'; record: SessionWriterLockRecord } + | { kind: 'live'; record: ActiveLockRecord; raw: string } + | { kind: 'stale'; record: ActiveLockRecord; raw: string } + | { kind: 'sealed'; record: SealedSessionWriterLockRecord; raw: string } | { kind: 'malformed' }; interface TranscriptFingerprint { @@ -146,6 +181,12 @@ type TranscriptState = fingerprint: TranscriptFingerprint; }; +interface OpenTranscriptProof { + readonly state: TranscriptState; + readonly sha256: string; + readonly handle?: fs.FileHandle; +} + function isProcessAlive(pid: number): boolean { try { process.kill(pid, 0); @@ -226,12 +267,11 @@ async function readProcessStartIdentity(pid: number): Promise { return null; } -function isLockRecord(value: unknown): value is SessionWriterLockRecord { - if (!value || typeof value !== 'object' || Array.isArray(value)) return false; - const record = value as Record; +function hasValidOwnerFields( + record: Record, +): record is Record & SessionWriterOwnerRecord { const processKind = record['process_kind']; return ( - record['schema_version'] === LOCK_SCHEMA_VERSION && typeof record['session_id'] === 'string' && record['session_id'].length > 0 && typeof record['owner_id'] === 'string' && @@ -252,6 +292,40 @@ function isLockRecord(value: unknown): value is SessionWriterLockRecord { ); } +function isSealedTranscriptProof( + value: unknown, +): value is SealedTranscriptProof { + if (!value || typeof value !== 'object' || Array.isArray(value)) return false; + const proof = value as Record; + return ( + typeof proof['relative_path'] === 'string' && + proof['relative_path'].length > 0 && + typeof proof['exists'] === 'boolean' && + Number.isSafeInteger(proof['byte_length']) && + (proof['byte_length'] as number) >= 0 && + typeof proof['sha256'] === 'string' && + /^[0-9a-f]{64}$/.test(proof['sha256']) && + (proof['exists'] || proof['byte_length'] === 0) + ); +} + +function isLockRecord(value: unknown): value is SessionWriterLockRecord { + if (!value || typeof value !== 'object' || Array.isArray(value)) return false; + const record = value as Record; + if (!hasValidOwnerFields(record)) return false; + if (record['schema_version'] === LEGACY_LOCK_SCHEMA_VERSION) { + return record['state'] === undefined; + } + if (record['schema_version'] !== LOCK_SCHEMA_VERSION) return false; + if (record['state'] === 'active') return true; + return ( + record['state'] === 'sealed' && + typeof record['sealed_at'] === 'string' && + Number.isFinite(Date.parse(record['sealed_at'])) && + isSealedTranscriptProof(record['transcript']) + ); +} + function parseLockRecord(raw: string): SessionWriterLockRecord | null { try { const parsed: unknown = JSON.parse(raw); @@ -261,17 +335,27 @@ function parseLockRecord(raw: string): SessionWriterLockRecord | null { } } -async function lockStateForRecord( +function isActiveLockRecord( record: SessionWriterLockRecord, +): record is ActiveLockRecord { + return ( + record.schema_version === LEGACY_LOCK_SCHEMA_VERSION || + record.state === 'active' + ); +} + +async function lockStateForRecord( + record: ActiveLockRecord, + raw: string, ): Promise { - if (record.hostname !== os.hostname()) return { kind: 'live' }; - if (!isProcessAlive(record.pid)) return { kind: 'stale', record }; - if (!record.process_start_identity) return { kind: 'live' }; + if (record.hostname !== os.hostname()) return { kind: 'live', record, raw }; + if (!isProcessAlive(record.pid)) return { kind: 'stale', record, raw }; + if (!record.process_start_identity) return { kind: 'live', record, raw }; const currentStartIdentity = await readProcessStartIdentity(record.pid); return currentStartIdentity !== null && currentStartIdentity !== record.process_start_identity - ? { kind: 'stale', record } - : { kind: 'live' }; + ? { kind: 'stale', record, raw } + : { kind: 'live', record, raw }; } async function delay(ms: number): Promise { @@ -313,6 +397,18 @@ function sameTranscriptState( ); } +async function assertTranscriptPathMissing(filePath: string): Promise { + try { + await fs.lstat(filePath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; + throw error; + } + throw new SessionWriterUnavailableError({ + cause: new Error('Session transcript path is not a regular file'), + }); +} + async function getTranscriptState(filePath: string): Promise { let handle: fs.FileHandle | undefined; try { @@ -320,6 +416,7 @@ async function getTranscriptState(filePath: string): Promise { handle = await fs.open(filePath, 'r'); } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + await assertTranscriptPathMissing(filePath); return { exists: false, byteLength: 0 }; } throw error; @@ -367,6 +464,199 @@ async function getTranscriptState(filePath: string): Promise { } } +function getTranscriptRelativePath( + runtimeBaseDir: string, + transcriptPath: string, +): string { + const relativePath = path.relative(runtimeBaseDir, transcriptPath); + if ( + relativePath.length === 0 || + path.isAbsolute(relativePath) || + relativePath === '..' || + relativePath.startsWith(`..${path.sep}`) + ) { + throw new SessionWriterUnavailableError({ + cause: new Error('Session transcript is outside the runtime base'), + }); + } + return relativePath.split(path.sep).join('/'); +} + +async function openTranscriptProof( + filePath: string, +): Promise { + let handle: fs.FileHandle | undefined; + try { + try { + handle = await fs.open(filePath, 'r'); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + await assertTranscriptPathMissing(filePath); + return { + state: { exists: false, byteLength: 0 }, + sha256: createHash('sha256').digest('hex'), + }; + } + throw error; + } + + const [beforeStat, pathStat] = await Promise.all([ + handle.stat(), + fs.lstat(filePath), + ]); + if ( + !beforeStat.isFile() || + !pathStat.isFile() || + pathStat.isSymbolicLink() + ) { + throw new SessionWriterUnavailableError(); + } + const beforeState: TranscriptState = { + exists: true, + byteLength: beforeStat.size, + fingerprint: transcriptFingerprint(beforeStat), + }; + if ( + !sameFileIdentity( + beforeState.fingerprint, + transcriptFingerprint(pathStat), + ) + ) { + throw new SessionTranscriptChangedError(); + } + if (beforeStat.size > 0) { + const lastByte = Buffer.allocUnsafe(1); + const { bytesRead } = await handle.read( + lastByte, + 0, + 1, + beforeStat.size - 1, + ); + if (bytesRead !== 1 || lastByte[0] !== 0x0a) { + throw new SessionTranscriptChangedError(); + } + } + + const hash = createHash('sha256'); + const buffer = Buffer.allocUnsafe(64 * 1024); + let position = 0; + while (position < beforeStat.size) { + const { bytesRead } = await handle.read( + buffer, + 0, + Math.min(buffer.byteLength, beforeStat.size - position), + position, + ); + if (bytesRead === 0) throw new SessionTranscriptChangedError(); + hash.update(buffer.subarray(0, bytesRead)); + position += bytesRead; + } + + const [afterStat, afterPathStat] = await Promise.all([ + handle.stat(), + fs.lstat(filePath), + ]); + const afterState: TranscriptState = { + exists: true, + byteLength: afterStat.size, + fingerprint: transcriptFingerprint(afterStat), + }; + if ( + !sameTranscriptState(beforeState, afterState) || + !afterPathStat.isFile() || + afterPathStat.isSymbolicLink() || + !sameFileIdentity( + afterState.fingerprint, + transcriptFingerprint(afterPathStat), + ) + ) { + throw new SessionTranscriptChangedError(); + } + + const proof = { + state: beforeState, + sha256: hash.digest('hex'), + handle, + }; + handle = undefined; + return proof; + } catch (error) { + if (error instanceof SessionWriterError) throw error; + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } finally { + await handle?.close().catch(() => {}); + } +} + +async function validateOpenTranscriptProof( + filePath: string, + proof: OpenTranscriptProof, +): Promise { + if (!proof.state.exists) { + const current = await getTranscriptState(filePath); + if (!sameTranscriptState(current, proof.state)) { + throw new SessionTranscriptChangedError(); + } + return; + } + const handle = proof.handle; + if (!handle) throw new SessionWriterUnavailableError(); + try { + const [handleStat, pathStat] = await Promise.all([ + handle.stat(), + fs.lstat(filePath), + ]); + if ( + !handleStat.isFile() || + !pathStat.isFile() || + pathStat.isSymbolicLink() + ) { + throw new SessionWriterUnavailableError(); + } + const current: TranscriptState = { + exists: true, + byteLength: handleStat.size, + fingerprint: transcriptFingerprint(handleStat), + }; + if ( + !sameTranscriptState(current, proof.state) || + !sameFileIdentity(current.fingerprint, transcriptFingerprint(pathStat)) + ) { + throw new SessionTranscriptChangedError(); + } + } catch (error) { + if (error instanceof SessionWriterError) throw error; + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + throw new SessionTranscriptChangedError(); + } + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } +} + +async function closeTranscriptProof(proof: OpenTranscriptProof): Promise { + await proof.handle?.close().catch(() => {}); +} + +function assertSealedProofMatches( + sealed: SealedSessionWriterLockRecord, + relativePath: string, + proof: OpenTranscriptProof, +): void { + const transcript = sealed.transcript; + if ( + transcript.relative_path !== relativePath || + transcript.exists !== proof.state.exists || + transcript.byte_length !== proof.state.byteLength || + transcript.sha256 !== proof.sha256 + ) { + throw new SessionTranscriptChangedError(); + } +} + async function restoreMovedLock( movedPath: string, lockPath: string, @@ -392,8 +682,9 @@ async function installLockRecord( const temporaryPath = `${lockPath}.${record.owner_id}.tmp`; let handle: fs.FileHandle | undefined; try { + const recordRaw = JSON.stringify(record); handle = await fs.open(temporaryPath, 'wx', 0o600); - await handle.writeFile(JSON.stringify(record), 'utf8'); + await handle.writeFile(recordRaw, 'utf8'); await handle.sync(); await handle.close(); handle = undefined; @@ -401,7 +692,14 @@ async function installLockRecord( await fs.link(temporaryPath, lockPath); return true; } catch (error) { - if ((error as NodeJS.ErrnoException).code === 'EEXIST') return false; + const { state } = await inspectExactRecord(lockPath, recordRaw); + if (state === 'exact') return true; + if ( + state === 'other' || + (error as NodeJS.ErrnoException).code === 'EEXIST' + ) { + return false; + } throw error; } } catch (error) { @@ -418,7 +716,7 @@ async function installLockRecord( async function acquireReclaimGuard( lockPath: string, staleOwnerId: string, - record: SessionWriterLockRecord, + record: ActiveLockRecord, inspect: ( lockPath: string, expectedSessionId: string, @@ -430,8 +728,11 @@ async function acquireReclaimGuard( if (await installLockRecord(guardPath, record)) return guardPath; const state = await inspect(guardPath, record.session_id); if (state.kind === 'missing') continue; - if (state.kind === 'live') throw new SessionWriterUnavailableError(); - if (state.kind === 'malformed') { + if ( + state.kind === 'live' || + state.kind === 'sealed' || + state.kind === 'malformed' + ) { throw new SessionWriterUnavailableError(); } guardPath = `${basePath}.${encodeURIComponent(state.record.owner_id)}`; @@ -444,12 +745,366 @@ async function removeOwnedLock( ownerId: string, ): Promise { const record = parseLockRecord(await fs.readFile(lockPath, 'utf8')); - if (!record || record.owner_id !== ownerId) { + if (!record || !isActiveLockRecord(record) || record.owner_id !== ownerId) { throw new SessionWriterLostError(); } await fs.unlink(lockPath); } +async function assertPathMissing(candidatePath: string): Promise { + try { + await fs.lstat(candidatePath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } + throw new SessionWriterUnavailableError({ + cause: new Error('Session writer transition claim already exists'), + }); +} + +async function removeExactRecord( + candidatePath: string, + expectedRaw: string, +): Promise { + let raw: string; + try { + raw = await fs.readFile(candidatePath, 'utf8'); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } + if (raw !== expectedRaw) throw new SessionWriterLostError(); + try { + await fs.unlink(candidatePath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } +} + +type ExactRecordState = 'exact' | 'missing' | 'other' | 'unknown'; + +interface RecordInspection { + state: State; + cause?: Error; +} + +async function inspectExactRecord( + candidatePath: string, + expectedRaw: string, +): Promise> { + try { + const stat = await fs.lstat(candidatePath); + if (!stat.isFile() || stat.isSymbolicLink()) return { state: 'other' }; + const raw = await fs.readFile(candidatePath, 'utf8'); + return { state: raw === expectedRaw ? 'exact' : 'other' }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + return { state: 'missing' }; + } + return { + state: 'unknown', + cause: error instanceof Error ? error : undefined, + }; + } +} + +async function removeTransitionClaimAfterFailure( + lockPath: string, + expectedPrimaryRaw: string, + claimPath: string, + claimRaw: string, +): Promise { + const { state: primaryState } = await inspectExactRecord( + lockPath, + expectedPrimaryRaw, + ); + if (primaryState !== 'exact') { + throw new SessionWriterUnavailableError({ + cause: new Error( + 'Session writer primary was not restored; transition claim retained', + ), + }); + } + await removeExactRecord(claimPath, claimRaw); +} + +async function releaseTransitionClaim( + claimPath: string, + claimRaw: string, +): Promise { + let raw: string; + try { + raw = await fs.readFile(claimPath, 'utf8'); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } + if (raw !== claimRaw) return; + try { + await fs.unlink(claimPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; + const { state } = await inspectExactRecord(claimPath, claimRaw); + if (state === 'missing' || state === 'other') return; + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } +} + +type ClaimedPrimaryState = + | 'source' + | 'candidate' + | 'missing' + | 'other' + | 'unknown'; + +async function inspectClaimedPrimary( + lockPath: string, + sourceRaw: string, + sessionId: string, +): Promise> { + try { + const stat = await fs.lstat(lockPath); + if (!stat.isFile() || stat.isSymbolicLink()) return { state: 'other' }; + const raw = await fs.readFile(lockPath, 'utf8'); + if (raw === sourceRaw) return { state: 'source' }; + const record = parseLockRecord(raw); + return { + state: + record?.schema_version === LOCK_SCHEMA_VERSION && + record.state === 'active' && + record.session_id === sessionId + ? 'candidate' + : 'other', + }; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + return { state: 'missing' }; + } + return { + state: 'unknown', + cause: error instanceof Error ? error : undefined, + }; + } +} + +async function assertExactTransitionClaim( + claimPath: string, + claimRaw: string, +): Promise { + const { state: claimState } = await inspectExactRecord(claimPath, claimRaw); + if (claimState !== 'exact') { + throw new SessionWriterUnavailableError({ + cause: new Error('Session writer transition claim ownership was lost'), + }); + } +} + +async function waitForClaimedPrimaryCandidate(attempt: number): Promise { + if (attempt >= CLAIMED_PRIMARY_WAIT_ATTEMPTS) { + throw new SessionWriterUnavailableError({ + cause: new Error( + 'Session writer primary candidate did not release the claimed path', + ), + }); + } + await delay(MALFORMED_RETRY_DELAY_MS); +} + +async function linkClaimedPrimary( + lockPath: string, + sourcePath: string, + sourceRaw: string, + sessionId: string, + claimPath: string, + claimRaw: string, +): Promise { + let candidateWaitAttempts = 0; + for (;;) { + await assertExactTransitionClaim(claimPath, claimRaw); + try { + await fs.link(sourcePath, lockPath); + return; + } catch (error) { + const { state } = await inspectClaimedPrimary( + lockPath, + sourceRaw, + sessionId, + ); + if (state === 'source') return; + if (state === 'candidate') { + // A schema-v2 acquirer can pass its first claim check immediately + // before this transition creates the claim. It must remove its primary + // candidate after the second check, so keep the predecessor and wait. + await waitForClaimedPrimaryCandidate(++candidateWaitAttempts); + continue; + } + if (state === 'other') throw new SessionWriterLostError(); + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } + } +} + +async function removeClaimedPrimary( + lockPath: string, + sourceRaw: string, + sessionId: string, + claimPath: string, + claimRaw: string, +): Promise { + let candidateWaitAttempts = 0; + for (;;) { + await assertExactTransitionClaim(claimPath, claimRaw); + const { state, cause } = await inspectClaimedPrimary( + lockPath, + sourceRaw, + sessionId, + ); + if (state === 'missing') return; + if (state === 'candidate') { + await waitForClaimedPrimaryCandidate(++candidateWaitAttempts); + continue; + } + if (state === 'other') throw new SessionWriterLostError(); + if (state === 'unknown') { + throw new SessionWriterUnavailableError({ cause }); + } + try { + await fs.unlink(lockPath); + return; + } catch (error) { + const { state: afterState } = await inspectClaimedPrimary( + lockPath, + sourceRaw, + sessionId, + ); + if (afterState === 'missing') return; + if (afterState === 'candidate') { + await waitForClaimedPrimaryCandidate(++candidateWaitAttempts); + continue; + } + if (afterState === 'other') throw new SessionWriterLostError(); + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } + } +} + +async function transitionExactPrimary( + lockPath: string, + expectedRaw: string, + replacementPath: string, + replacementRaw: string, + retiredPath: string, + sessionId: string, + claimPath: string, + claimRaw: string, +): Promise { + try { + await fs.rename(lockPath, retiredPath); + } catch (error) { + const [primaryExpected, retired] = await Promise.all([ + inspectExactRecord(lockPath, expectedRaw), + inspectExactRecord(retiredPath, expectedRaw), + ]); + if (primaryExpected.state === 'exact' && retired.state === 'missing') { + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } + if (retired.state !== 'exact') { + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } + } + try { + await linkClaimedPrimary( + lockPath, + replacementPath, + replacementRaw, + sessionId, + claimPath, + claimRaw, + ); + } catch (error) { + try { + await linkClaimedPrimary( + lockPath, + retiredPath, + expectedRaw, + sessionId, + claimPath, + claimRaw, + ); + await removeExactRecord(retiredPath, expectedRaw); + } catch (restoreError) { + throw new SessionWriterUnavailableError({ + cause: new AggregateError( + [error, restoreError], + 'Session writer primary transition could not be restored', + ), + }); + } + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } +} + +async function rollbackExactTransition( + lockPath: string, + replacementRaw: string, + retiredPath: string, + expectedRaw: string, + sessionId: string, + claimPath: string, + claimRaw: string, +): Promise { + await removeClaimedPrimary( + lockPath, + replacementRaw, + sessionId, + claimPath, + claimRaw, + ); + const retired = await inspectExactRecord(retiredPath, expectedRaw); + if (retired.state !== 'exact') { + throw retired.state === 'other' + ? new SessionWriterLostError() + : new SessionWriterUnavailableError({ cause: retired.cause }); + } + try { + await linkClaimedPrimary( + lockPath, + retiredPath, + expectedRaw, + sessionId, + claimPath, + claimRaw, + ); + await removeExactRecord(retiredPath, expectedRaw); + } catch (error) { + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } +} + export function getSessionWriterLockPath( runtimeBaseDir: string, sessionId: string, @@ -469,13 +1124,15 @@ export class SessionWriterLease { readonly transcriptPath: string; private expectedTranscriptState: TranscriptState | undefined; private released = false; - private releasePromise: Promise | undefined; + private terminalPromise: Promise | undefined; + private operationTail: Promise = Promise.resolve(); private readonly lockRecordRaw: string; private readonly retiredPath: string; + private readonly claimPath: string; private constructor( private readonly lockPath: string, - lockRecord: SessionWriterLockRecord, + lockRecord: ActiveLockRecord, options: AcquireSessionWriterLeaseOptions, ) { this.ownerId = lockRecord.owner_id; @@ -484,6 +1141,7 @@ export class SessionWriterLease { this.transcriptPath = options.transcriptPath; this.lockRecordRaw = JSON.stringify(lockRecord); this.retiredPath = `${lockPath}.released.${encodeURIComponent(this.ownerId)}`; + this.claimPath = `${lockPath}.claim`; } get transcriptExistedAtAcquire(): boolean { @@ -546,8 +1204,9 @@ export class SessionWriterLease { } const processStartIdentity = await readProcessStartIdentity(process.pid); - const lockRecord: SessionWriterLockRecord = { + const lockRecord: ActiveSessionWriterLockRecord = { schema_version: LOCK_SCHEMA_VERSION, + state: 'active', session_id: normalizedOptions.sessionId, owner_id: randomUUID(), pid: process.pid, @@ -559,9 +1218,17 @@ export class SessionWriterLease { acquired_at: new Date().toISOString(), qwen_version: normalizedOptions.qwenVersion ?? null, }; + const claimPath = `${lockPath}.claim`; for (let attempt = 0; attempt < ACQUIRE_ATTEMPTS; attempt++) { + await assertPathMissing(claimPath); if (await installLockRecord(lockPath, lockRecord)) { + try { + await assertPathMissing(claimPath); + } catch (error) { + await removeOwnedLock(lockPath, lockRecord.owner_id).catch(() => {}); + throw error; + } return SessionWriterLease.finishAcquisition( lockPath, lockRecord, @@ -580,6 +1247,17 @@ export class SessionWriterLease { cause: new Error('Existing session writer lock is malformed'), }); } + if (state.kind === 'sealed') { + if (normalizedOptions.takeoverPolicy !== 'certified') { + throw new SessionWriterConflictError(); + } + return SessionWriterLease.takeOverSealed( + lockPath, + state, + lockRecord, + normalizedOptions, + ); + } if (normalizedOptions.reclaimPolicy === 'never') { throw new SessionWriterConflictError(); } @@ -596,6 +1274,7 @@ export class SessionWriterLease { let staleMoved = false; const stalePath = `${lockPath}.stale.${process.pid}.${randomUUID()}`; try { + await assertPathMissing(claimPath); const currentState = await SessionWriterLease.inspectExistingLock( lockPath, normalizedOptions.sessionId, @@ -630,6 +1309,7 @@ export class SessionWriterLease { throw new SessionWriterUnavailableError(); } primaryInstalled = true; + await assertPathMissing(claimPath); const lease = await SessionWriterLease.finishAcquisition( lockPath, lockRecord, @@ -655,10 +1335,145 @@ export class SessionWriterLease { throw new SessionWriterUnavailableError(); } + private static async takeOverSealed( + lockPath: string, + observed: Extract, + lockRecord: ActiveSessionWriterLockRecord, + options: AcquireSessionWriterLeaseOptions, + ): Promise { + const relativePath = getTranscriptRelativePath( + options.runtimeBaseDir, + options.transcriptPath, + ); + const proof = await openTranscriptProof(options.transcriptPath); + const claimPath = `${lockPath}.claim`; + const claimRaw = JSON.stringify(lockRecord); + const retiredPath = `${lockPath}.sealed.${encodeURIComponent( + observed.record.owner_id, + )}.${encodeURIComponent(lockRecord.owner_id)}`; + let claimAcquired = false; + let transitionStarted = false; + let transitionCommitted = false; + try { + assertSealedProofMatches(observed.record, relativePath, proof); + await assertPathMissing(retiredPath); + if (!(await installLockRecord(claimPath, lockRecord))) { + throw new SessionWriterUnavailableError({ + cause: new Error('Session writer transition claim already exists'), + }); + } + claimAcquired = true; + const current = await SessionWriterLease.inspectExistingLock( + lockPath, + options.sessionId, + ); + if (current.kind !== 'sealed' || current.raw !== observed.raw) { + throw current.kind === 'live' + ? new SessionWriterConflictError() + : new SessionWriterUnavailableError(); + } + await validateOpenTranscriptProof(options.transcriptPath, proof); + assertSealedProofMatches(current.record, relativePath, proof); + transitionStarted = true; + await transitionExactPrimary( + lockPath, + observed.raw, + claimPath, + claimRaw, + retiredPath, + options.sessionId, + claimPath, + claimRaw, + ); + transitionCommitted = true; + await validateOpenTranscriptProof(options.transcriptPath, proof); + const lease = await SessionWriterLease.finishAcquisition( + lockPath, + lockRecord, + options, + proof.state, + ); + await releaseTransitionClaim(claimPath, claimRaw); + claimAcquired = false; + await removeExactRecord(retiredPath, observed.raw).catch((error) => { + debugLogger.debug( + `Session writer sealed lock cleanup failed path=${JSON.stringify(retiredPath)} ` + + `error=${describeDiagnosticError(error)}`, + ); + }); + debugLogger.info( + `Certified session writer handoff accepted sessionId=${JSON.stringify(options.sessionId)} ` + + `previousHostname=${JSON.stringify(observed.record.hostname)} ` + + `previousPid=${observed.record.pid} sealedAt=${observed.record.sealed_at}`, + ); + return lease; + } catch (error) { + const cleanupFailures: unknown[] = []; + const claimState = claimAcquired + ? (await inspectExactRecord(claimPath, claimRaw)).state + : 'missing'; + if (claimState === 'unknown') { + cleanupFailures.push( + new SessionWriterUnavailableError({ + cause: new Error( + 'Session writer transition claim ownership is unreadable', + ), + }), + ); + } + if (transitionCommitted && claimState === 'exact') { + try { + await rollbackExactTransition( + lockPath, + claimRaw, + retiredPath, + observed.raw, + options.sessionId, + claimPath, + claimRaw, + ); + } catch (cleanupError) { + cleanupFailures.push(cleanupError); + } + } + if (claimState === 'exact') { + try { + if (transitionStarted) { + await removeTransitionClaimAfterFailure( + lockPath, + observed.raw, + claimPath, + claimRaw, + ); + } else { + await releaseTransitionClaim(claimPath, claimRaw); + } + } catch (cleanupError) { + cleanupFailures.push(cleanupError); + } + } + if (cleanupFailures.length > 0) { + throw new SessionWriterUnavailableError({ + cause: new AggregateError( + [error, ...cleanupFailures], + 'Certified session writer takeover cleanup failed', + ), + }); + } + if (error instanceof SessionWriterError) throw error; + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } finally { + await closeTranscriptProof(proof); + } + } + private static async finishAcquisition( lockPath: string, - lockRecord: SessionWriterLockRecord, + lockRecord: ActiveLockRecord, options: AcquireSessionWriterLeaseOptions, + requiredTranscriptState?: TranscriptState, ): Promise { const lease = new SessionWriterLease(lockPath, lockRecord, options); try { @@ -666,6 +1481,15 @@ export class SessionWriterLease { lease.expectedTranscriptState = await getTranscriptState( options.transcriptPath, ); + if ( + requiredTranscriptState && + !sameTranscriptState( + lease.expectedTranscriptState, + requiredTranscriptState, + ) + ) { + throw new SessionTranscriptChangedError(); + } return lease; } catch (error) { try { @@ -717,7 +1541,10 @@ export class SessionWriterLease { cause: new Error('Session writer lock belongs to another session'), }); } - return lockStateForRecord(record); + if (!isActiveLockRecord(record)) { + return { kind: 'sealed', record, raw }; + } + return lockStateForRecord(record, raw); } if (attempt + 1 < MALFORMED_RETRY_COUNT) { await delay(MALFORMED_RETRY_DELAY_MS); @@ -726,7 +1553,7 @@ export class SessionWriterLease { return { kind: 'malformed' }; } - private async readOwnedLock(): Promise { + private async readOwnedLock(): Promise { if (this.released) throw new SessionWriterLostError(); let stat: Awaited>; try { @@ -752,6 +1579,7 @@ export class SessionWriterLease { const record = parseLockRecord(raw); if ( !record || + !isActiveLockRecord(record) || record.owner_id !== this.ownerId || raw !== this.lockRecordRaw ) { @@ -760,7 +1588,11 @@ export class SessionWriterLease { return record; } - async assertOwnedAndUnchanged(): Promise { + assertOwnedAndUnchanged(): Promise { + return this.runExclusive(() => this.assertOwnedAndUnchangedOnce()); + } + + private async assertOwnedAndUnchangedOnce(): Promise { await this.readOwnedLock(); if (!this.expectedTranscriptState) { throw new SessionWriterUnavailableError(); @@ -771,7 +1603,11 @@ export class SessionWriterLease { } } - async appendJsonLine(value: unknown): Promise { + appendJsonLine(value: unknown): Promise { + return this.runExclusive(() => this.appendJsonLineOnce(value)); + } + + private async appendJsonLineOnce(value: unknown): Promise { let serialized: string | undefined; try { serialized = JSON.stringify(value); @@ -782,7 +1618,7 @@ export class SessionWriterLease { } if (serialized === undefined) throw new SessionWriterUnavailableError(); const bytes = Buffer.from(`${serialized}\n`, 'utf8'); - await this.assertOwnedAndUnchanged(); + await this.assertOwnedAndUnchangedOnce(); const expectedBefore = this.expectedTranscriptState; if (!expectedBefore) throw new SessionWriterUnavailableError(); const nextByteLength = expectedBefore.byteLength + bytes.byteLength; @@ -845,14 +1681,28 @@ export class SessionWriterLease { } release(): Promise { - this.releasePromise ??= this.releaseOnce(); - return this.releasePromise; + this.terminalPromise ??= this.runExclusive(() => this.releaseOnce()); + return this.terminalPromise; + } + + sealForHandoff(): Promise { + this.terminalPromise ??= this.runExclusive(() => this.sealForHandoffOnce()); + return this.terminalPromise; } get isReleased(): boolean { return this.released; } + private runExclusive(operation: () => Promise): Promise { + const pending = this.operationTail.then(operation, operation); + this.operationTail = pending.then( + () => undefined, + () => undefined, + ); + return pending; + } + private async releaseOnce(): Promise { if (this.released) return; await this.readOwnedLockForRelease(); @@ -887,11 +1737,169 @@ export class SessionWriterLease { } } - private async readOwnedLockForRelease(): Promise { + private async sealForHandoffOnce(): Promise { + if (this.released) return; + const ownedRecord = await this.readOwnedLockForRelease(); + if (!this.expectedTranscriptState) { + throw new SessionWriterUnavailableError(); + } + const relativePath = getTranscriptRelativePath( + this.runtimeBaseDir, + this.transcriptPath, + ); + const proof = await openTranscriptProof(this.transcriptPath); + const sealedRecord: SealedSessionWriterLockRecord = { + ...ownedRecord, + schema_version: LOCK_SCHEMA_VERSION, + state: 'sealed', + sealed_at: new Date().toISOString(), + transcript: { + relative_path: relativePath, + exists: proof.state.exists, + byte_length: proof.state.byteLength, + sha256: proof.sha256, + }, + }; + const sealedRaw = JSON.stringify(sealedRecord); + const sealedCandidatePath = `${this.lockPath}.sealed-candidate.${encodeURIComponent( + this.ownerId, + )}`; + const handoffRetiredPath = `${this.lockPath}.handoff.${encodeURIComponent( + this.ownerId, + )}`; + let claimAcquired = false; + let transitionStarted = false; + let transitionCommitted = false; + try { + if (!sameTranscriptState(proof.state, this.expectedTranscriptState)) { + throw new SessionTranscriptChangedError(); + } + if (!(await installLockRecord(sealedCandidatePath, sealedRecord))) { + throw new SessionWriterUnavailableError({ + cause: new Error('Session writer sealed candidate already exists'), + }); + } + await assertPathMissing(handoffRetiredPath); + if (!(await installLockRecord(this.claimPath, ownedRecord))) { + throw new SessionWriterUnavailableError({ + cause: new Error('Session writer transition claim already exists'), + }); + } + claimAcquired = true; + await this.readOwnedLock(); + await validateOpenTranscriptProof(this.transcriptPath, proof); + transitionStarted = true; + await transitionExactPrimary( + this.lockPath, + this.lockRecordRaw, + sealedCandidatePath, + sealedRaw, + handoffRetiredPath, + this.sessionId, + this.claimPath, + this.lockRecordRaw, + ); + transitionCommitted = true; + await validateOpenTranscriptProof(this.transcriptPath, proof); + await releaseTransitionClaim(this.claimPath, this.lockRecordRaw); + claimAcquired = false; + this.released = true; + await removeExactRecord(handoffRetiredPath, this.lockRecordRaw).catch( + (error) => { + debugLogger.debug( + `Session writer handoff retired cleanup failed path=${JSON.stringify(handoffRetiredPath)} ` + + `error=${describeDiagnosticError(error)}`, + ); + }, + ); + await removeExactRecord(sealedCandidatePath, sealedRaw).catch((error) => { + debugLogger.debug( + `Session writer handoff candidate cleanup failed path=${JSON.stringify(sealedCandidatePath)} ` + + `error=${describeDiagnosticError(error)}`, + ); + }); + debugLogger.info( + `Session writer handoff sealed sessionId=${JSON.stringify(this.sessionId)} ` + + `hostname=${JSON.stringify(ownedRecord.hostname)} pid=${ownedRecord.pid} ` + + `sealedAt=${sealedRecord.sealed_at}`, + ); + } catch (error) { + const cleanupFailures: unknown[] = []; + const claimState = claimAcquired + ? (await inspectExactRecord(this.claimPath, this.lockRecordRaw)).state + : 'missing'; + if (claimState === 'unknown') { + cleanupFailures.push( + new SessionWriterUnavailableError({ + cause: new Error( + 'Session writer transition claim ownership is unreadable', + ), + }), + ); + } + if (transitionCommitted && claimState === 'exact') { + try { + await rollbackExactTransition( + this.lockPath, + sealedRaw, + handoffRetiredPath, + this.lockRecordRaw, + this.sessionId, + this.claimPath, + this.lockRecordRaw, + ); + } catch (cleanupError) { + cleanupFailures.push(cleanupError); + } + } + if (claimState === 'exact') { + try { + if (transitionStarted) { + await removeTransitionClaimAfterFailure( + this.lockPath, + this.lockRecordRaw, + this.claimPath, + this.lockRecordRaw, + ); + } else { + await releaseTransitionClaim(this.claimPath, this.lockRecordRaw); + } + } catch (cleanupError) { + cleanupFailures.push(cleanupError); + } + } + try { + await removeExactRecord(sealedCandidatePath, sealedRaw); + } catch (cleanupError) { + cleanupFailures.push(cleanupError); + } + if ( + (await inspectExactRecord(this.lockPath, this.lockRecordRaw)).state !== + 'exact' + ) { + this.released = true; + } + if (cleanupFailures.length > 0) { + throw new SessionWriterUnavailableError({ + cause: new AggregateError( + [error, ...cleanupFailures], + 'Session writer handoff cleanup failed', + ), + }); + } + if (error instanceof SessionWriterError) throw error; + throw new SessionWriterUnavailableError({ + cause: error instanceof Error ? error : undefined, + }); + } finally { + await closeTranscriptProof(proof); + } + } + + private async readOwnedLockForRelease(): Promise { for (let attempt = 0; attempt < RELEASE_PRECHECK_ATTEMPTS; attempt++) { try { - await this.readOwnedLock(); - return; + return await this.readOwnedLock(); } catch (error) { if ( !(error instanceof SessionWriterUnavailableError) || @@ -902,6 +1910,7 @@ export class SessionWriterLease { } await delay(RELEASE_PRECHECK_RETRY_DELAY_MS); } + throw new SessionWriterUnavailableError(); } private async inspectReleasePath(