From 3171d115ba157efde5133bfa2489de75ca5e7ccc Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Thu, 27 Aug 2026 20:10:06 +0800 Subject: [PATCH 01/12] fix(core): preserve ownership during session cleanup Co-authored-by: Qwen-Coder --- docs/developers/qwen-serve-protocol.md | 4 +- .../standalone-session-service.test.ts | 83 +++++-- .../standalone-session-service.ts | 16 ++ .../src/serve/server/session-archive.test.ts | 1 + .../cli/src/serve/server/session-archive.ts | 15 +- .../src/services/session-writer-lease.test.ts | 71 ++++++ .../core/src/services/session-writer-lease.ts | 80 ++++++- .../sessionService.corruption.test.ts | 39 +++- .../core/src/services/sessionService.test.ts | 209 +++++++++++++++++- packages/core/src/services/sessionService.ts | 89 +++++--- 10 files changed, 538 insertions(+), 69 deletions(-) diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 4147504603f..37f79d9bbaa 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -2481,7 +2481,7 @@ Response: } ``` -`resolveConflicts` is optional and defaults to `false`. By default, active and archived files with the same id are reported in `errors`, and neither copy is moved, removed, or overwritten. Archiving a live session still performs the strict close described above before classifying the conflict, so that close may flush queued records to the active transcript. With `resolveConflicts: true`, archive keeps the archived copy, removes the active copy, and reports the id in both `archived` and `resolvedConflicts`. `errors` entries have `{ "sessionId": "", "error": "message" }`. +`resolveConflicts` is optional and defaults to `false`. By default, active and archived files with the same id are reported in `errors`, and neither copy is moved, removed, or overwritten. Archiving a live session still performs the strict close described above before classifying the conflict, so that close may flush queued records to the active transcript. With `resolveConflicts: true`, archive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the archived copy, removes the active copy, and reports the id in both `archived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. `errors` entries have `{ "sessionId": "", "error": "message" }`. Lifecycle conflicts are batch item outcomes: the workspace-less and workspace-qualified routes return HTTP `200` with the conflict in `errors`. This replaces the earlier workspace-qualified HTTP `409 session_conflict` envelope; clients that called that route must inspect the batch response. Internal-runtime REST batches preserve the safe conflict message while continuing to redact other per-session failure details. @@ -2507,7 +2507,7 @@ Response: } ``` -`resolveConflicts` is optional and defaults to `false`. By default, simultaneous active and archived JSONL files produce a conflict in `errors`, and neither copy is moved, removed, or overwritten; an active-only session is returned in `alreadyActive`. With `resolveConflicts: true`, unarchive keeps the active copy, removes the archived copy, and reports the id in both `unarchived` and `resolvedConflicts`. Archive or unarchive in flight for the same id returns `409 session_archiving` before starting the batch. +`resolveConflicts` is optional and defaults to `false`. By default, simultaneous active and archived JSONL files produce a conflict in `errors`, and neither copy is moved, removed, or overwritten; an active-only session is returned in `alreadyActive`. With `resolveConflicts: true`, unarchive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the active copy, removes the archived copy, and reports the id in both `unarchived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. Archive or unarchive in flight for the same id returns `409 session_archiving` before starting the batch. ACP-over-HTTP uses the same request and response bodies through vendor methods `_qwen/sessions/archive` and `_qwen/sessions/unarchive`. The REST route table maps `POST /sessions/archive` and `POST /sessions/unarchive` to those methods for ACP transports. diff --git a/packages/cli/src/serve/conversations/standalone-session-service.test.ts b/packages/cli/src/serve/conversations/standalone-session-service.test.ts index b4f76c36c60..e286aad2cf5 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.test.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.test.ts @@ -330,12 +330,14 @@ function mockArchivedStandalone(storageSessionId = sessionId): void { function mockWriterLease(): { assertOwnedAndUnchanged: ReturnType; + assertCleanupOwned: ReturnType; release: ReturnType; isReleased: boolean; isReleaseDurabilityPending: boolean; } { const lease = { assertOwnedAndUnchanged: vi.fn(async () => undefined), + assertCleanupOwned: vi.fn(), release: vi.fn(async () => undefined), isReleased: false, isReleaseDurabilityPending: false, @@ -628,13 +630,15 @@ describe('StandaloneSessionService', () => { const archiveHarness = createHarness(); mockActiveStandalone(); const archiveLease = mockWriterLease(); - vi.spyOn(SessionService.prototype, 'archiveSessions').mockResolvedValue({ - archived: [sessionId], - alreadyArchived: [], - resolvedConflicts: [], - notFound: [], - errors: [], - }); + const archive = vi + .spyOn(SessionService.prototype, 'archiveSessions') + .mockResolvedValue({ + archived: [sessionId], + alreadyArchived: [], + resolvedConflicts: [], + notFound: [], + errors: [], + }); await expect(archiveHarness.service.archive([sessionId])).resolves.toEqual({ archived: [sessionId], @@ -642,19 +646,30 @@ describe('StandaloneSessionService', () => { notFound: [], errors: [], }); + const archiveOptions = archive.mock.calls[0]?.[1]; + expect(archiveOptions).toEqual( + expect.objectContaining({ + assertCanMutate: expect.any(Function), + assertCleanupOwned: expect.any(Function), + }), + ); + archiveOptions?.assertCleanupOwned?.(); + expect(archiveLease.assertCleanupOwned).toHaveBeenCalledOnce(); expect(archiveLease.release).toHaveBeenCalledOnce(); vi.restoreAllMocks(); const unarchiveHarness = createHarness(); mockArchivedStandalone(); const unarchiveLease = mockWriterLease(); - vi.spyOn(SessionService.prototype, 'unarchiveSessions').mockResolvedValue({ - unarchived: [sessionId], - alreadyActive: [], - resolvedConflicts: [], - notFound: [], - errors: [], - }); + const unarchive = vi + .spyOn(SessionService.prototype, 'unarchiveSessions') + .mockResolvedValue({ + unarchived: [sessionId], + alreadyActive: [], + resolvedConflicts: [], + notFound: [], + errors: [], + }); await expect( unarchiveHarness.service.unarchive([sessionId]), @@ -664,6 +679,15 @@ describe('StandaloneSessionService', () => { notFound: [], errors: [], }); + const unarchiveOptions = unarchive.mock.calls[0]?.[1]; + expect(unarchiveOptions).toEqual( + expect.objectContaining({ + assertCanMutate: expect.any(Function), + assertCleanupOwned: expect.any(Function), + }), + ); + unarchiveOptions?.assertCleanupOwned?.(); + expect(unarchiveLease.assertCleanupOwned).toHaveBeenCalledOnce(); expect(unarchiveLease.release).toHaveBeenCalledOnce(); }); @@ -694,10 +718,9 @@ describe('StandaloneSessionService', () => { SessionService.prototype, 'removeSessionTranscriptForLifecycle', ).mockResolvedValue(true); - vi.spyOn( - SessionService.prototype, - 'cleanupRemovedSessionStateForLifecycle', - ).mockResolvedValue(); + const cleanupRemovedState = vi + .spyOn(SessionService.prototype, 'cleanupRemovedSessionStateForLifecycle') + .mockResolvedValue(); await expect(harness.service.delete([sessionId])).resolves.toEqual({ removed: [sessionId], @@ -721,6 +744,13 @@ describe('StandaloneSessionService', () => { { assertCanCommit: expect.any(Function) }, ); expect(harness.deletionJournal.clear).toHaveBeenCalledOnce(); + const cleanupOptions = cleanupRemovedState.mock.calls[0]?.[1]; + expect(cleanupOptions).toEqual({ + assertCanMutate: expect.any(Function), + assertCleanupOwned: expect.any(Function), + }); + cleanupOptions?.assertCleanupOwned?.(); + expect(lease.assertCleanupOwned).toHaveBeenCalledOnce(); expect(lease.release).toHaveBeenCalledOnce(); }); @@ -1138,12 +1168,11 @@ describe('StandaloneSessionService', () => { SessionService.prototype, 'findSessionIdIgnoringCase', ).mockResolvedValue(undefined); - vi.spyOn( - SessionService.prototype, - 'cleanupRemovedSessionStateForLifecycle', - ).mockResolvedValue(); + const cleanupRemovedState = vi + .spyOn(SessionService.prototype, 'cleanupRemovedSessionStateForLifecycle') + .mockResolvedValue(); const harness = createHarness(); - mockWriterLease(); + const lease = mockWriterLease(); harness.deletionJournal.read .mockResolvedValueOnce(deletionEntry() as never) .mockResolvedValueOnce(deletionEntry() as never); @@ -1169,6 +1198,14 @@ describe('StandaloneSessionService', () => { fileCleanupPending: [], }); expect(harness.deletionJournal.clear).toHaveBeenCalledWith(sessionId, root); + const cleanupCalls = cleanupRemovedState.mock.calls; + const cleanupOptions = cleanupCalls[cleanupCalls.length - 1]?.[1]; + expect(cleanupOptions).toEqual({ + assertCanMutate: expect.any(Function), + assertCleanupOwned: expect.any(Function), + }); + cleanupOptions?.assertCleanupOwned?.(); + expect(lease.assertCleanupOwned).toHaveBeenCalledOnce(); }); it('keeps deletion outcome unknown when transcript directory sync fails after unlink', async () => { diff --git a/packages/cli/src/serve/conversations/standalone-session-service.ts b/packages/cli/src/serve/conversations/standalone-session-service.ts index 05f71e807c5..f5c61d02968 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.ts @@ -1253,6 +1253,10 @@ export class StandaloneSessionService { record.storageSessionId, { assertCanMutate: () => this.options.assertRuntimeCurrent(runtime), + assertCleanupOwned: () => { + this.options.assertRuntimeCurrent(runtime); + lease.assertCleanupOwned(); + }, }, ); } catch { @@ -1555,6 +1559,10 @@ export class StandaloneSessionService { lease.assertOwnedAndUnchanged(), assertCanMutate: () => this.options.assertRuntimeCurrent(runtime), + assertCleanupOwned: () => { + this.options.assertRuntimeCurrent(runtime); + lease.assertCleanupOwned(); + }, }, ); if (archived.errors[0]) throw archived.errors[0].error; @@ -1655,6 +1663,10 @@ export class StandaloneSessionService { lease.assertOwnedAndUnchanged(), assertCanMutate: () => this.options.assertRuntimeCurrent(runtime), + assertCleanupOwned: () => { + this.options.assertRuntimeCurrent(runtime); + lease.assertCleanupOwned(); + }, }, ); if (unarchived.errors[0]) throw unarchived.errors[0].error; @@ -1959,6 +1971,10 @@ export class StandaloneSessionService { locked.storageSessionId, { assertCanMutate: () => this.options.assertRuntimeCurrent(runtime), + assertCleanupOwned: () => { + this.options.assertRuntimeCurrent(runtime); + lease.assertCleanupOwned(); + }, }, ); } catch { diff --git a/packages/cli/src/serve/server/session-archive.test.ts b/packages/cli/src/serve/server/session-archive.test.ts index 343e8d26cef..6bdbba87f6e 100644 --- a/packages/cli/src/serve/server/session-archive.test.ts +++ b/packages/cli/src/serve/server/session-archive.test.ts @@ -926,6 +926,7 @@ describe('archiveDaemonSessions', () => { }); vi.spyOn(service, 'acquireSessionWriterLease').mockResolvedValue({ assertOwnedAndUnchanged: vi.fn().mockResolvedValue(undefined), + assertCleanupOwned: vi.fn(), release, } as unknown as SessionWriterLease); diff --git a/packages/cli/src/serve/server/session-archive.ts b/packages/cli/src/serve/server/session-archive.ts index 1040d1d443e..ecc70687d4d 100644 --- a/packages/cli/src/serve/server/session-archive.ts +++ b/packages/cli/src/serve/server/session-archive.ts @@ -225,6 +225,7 @@ async function runWithDaemonWriterLease(params: { service: SessionService; mutate: ( assertOwnedAndUnchanged: () => Promise, + assertCleanupOwned: () => void, ) => Promise<{ value: T; mutationApplied: boolean }>; mutationAppliedAfterError: () => Promise; afterMutationApplied: () => Promise; @@ -266,7 +267,10 @@ async function runWithDaemonWriterLease(params: { let mutationApplied = false; let mutationError: unknown; try { - const mutation = await mutate(() => lease.assertOwnedAndUnchanged()); + const mutation = await mutate( + () => lease.assertOwnedAndUnchanged(), + () => lease.assertCleanupOwned(), + ); value = mutation.value; mutationApplied = mutation.mutationApplied; } catch (error) { @@ -438,7 +442,7 @@ async function deletePersistedSessionWithLease( action: 'delete', sessionId, service, - mutate: async (assertOwnedAndUnchanged) => { + mutate: async (assertOwnedAndUnchanged, assertCleanupOwned) => { const lockedLocation = await classifySessionLocation(service, sessionId); if (lockedLocation === undefined) { return { @@ -449,6 +453,7 @@ async function deletePersistedSessionWithLease( const removed = await service.removeSession(sessionId, { assertStorageUnchanged: assertOwnedAndUnchanged, assertCanMutate, + assertCleanupOwned, }); return { value: removed ? ('removed' as const) : ('notFound' as const), @@ -895,7 +900,7 @@ export async function archiveDaemonSessions(params: { action: 'archive', sessionId, service, - mutate: async (assertOwnedAndUnchanged) => { + mutate: async (assertOwnedAndUnchanged, assertCleanupOwned) => { const lockedLocation = await classifySessionLocation( service, sessionId, @@ -919,6 +924,7 @@ export async function archiveDaemonSessions(params: { resolveConflicts, assertStorageUnchanged: assertOwnedAndUnchanged, assertCanMutate, + assertCleanupOwned, }); if (result.errors[0]) throw result.errors[0].error; if (result.archived.length > 0) { @@ -1092,7 +1098,7 @@ export async function unarchiveDaemonSessions(params: { action: 'unarchive', sessionId, service, - mutate: async (assertOwnedAndUnchanged) => { + mutate: async (assertOwnedAndUnchanged, assertCleanupOwned) => { const lockedLocation = await classifySessionLocation( service, sessionId, @@ -1116,6 +1122,7 @@ export async function unarchiveDaemonSessions(params: { resolveConflicts, assertStorageUnchanged: assertOwnedAndUnchanged, assertCanMutate, + assertCleanupOwned, }); if (result.errors[0]) throw result.errors[0].error; if (result.unarchived.length > 0) { diff --git a/packages/core/src/services/session-writer-lease.test.ts b/packages/core/src/services/session-writer-lease.test.ts index 6fc09c3821b..ded60f8d822 100644 --- a/packages/core/src/services/session-writer-lease.test.ts +++ b/packages/core/src/services/session-writer-lease.test.ts @@ -11,6 +11,7 @@ import { constants as fsConstants, mkdirSync, readFileSync, + renameSync, statSync, unlinkSync, utimesSync, @@ -969,6 +970,7 @@ describe('SessionWriterLease', () => { const lease = await SessionWriterLease.acquire(fixture.options); await fs.appendFile(fixture.transcriptPath, '{"external":true}\n'); + expect(() => lease.assertCleanupOwned()).not.toThrow(); await expect(lease.assertOwnedAndUnchanged()).rejects.toBeInstanceOf( SessionTranscriptChangedError, ); @@ -979,6 +981,7 @@ describe('SessionWriterLease', () => { ); await fs.unlink(lockPath); await fs.writeFile(lockPath, '{"replacement":true}'); + expect(() => lease.assertCleanupOwned()).toThrow(SessionWriterLostError); await expect(lease.assertOwnedAndUnchanged()).rejects.toBeInstanceOf( SessionWriterLostError, ); @@ -990,6 +993,74 @@ describe('SessionWriterLease', () => { ); }); + it.runIf(process.platform !== 'win32')( + 'rejects a byte-identical atomic replacement during cleanup', + async () => { + const fixture = await createFixture(); + const lease = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const replacementPath = `${lockPath}.replacement`; + const lockRecord = await fs.readFile(lockPath, 'utf8'); + await fs.writeFile(replacementPath, lockRecord); + await fs.rename(replacementPath, lockPath); + + expect(() => lease.assertCleanupOwned()).toThrow(SessionWriterLostError); + await expect(lease.release()).rejects.toBeInstanceOf( + SessionWriterLostError, + ); + }, + ); + + it.runIf(process.platform !== 'win32')( + 'rejects a byte-identical atomic replacement during acquisition', + async () => { + const fixture = await createFixture(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const replacementPath = `${lockPath}.replacement`; + + await expect( + SessionWriterLease.acquire({ + ...fixture.options, + onOwnershipAcquired: () => { + writeFileSync(replacementPath, readFileSync(lockPath)); + renameSync(replacementPath, lockPath); + }, + }), + ).rejects.toBeInstanceOf(SessionWriterUnavailableError); + await fs.unlink(lockPath); + }, + ); + + it.runIf(process.platform !== 'win32')( + 'rejects a symlinked cleanup lock', + async () => { + const fixture = await createFixture(); + const lease = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const targetPath = `${lockPath}.replacement`; + const lockRecord = await fs.readFile(lockPath, 'utf8'); + await fs.writeFile(targetPath, lockRecord); + await fs.unlink(lockPath); + await fs.symlink(targetPath, lockPath); + + expect(() => lease.assertCleanupOwned()).toThrow(SessionWriterLostError); + await expect(lease.release()).rejects.toBeInstanceOf( + SessionWriterLostError, + ); + await fs.unlink(lockPath); + await fs.unlink(targetPath); + }, + ); + it.runIf(process.platform !== 'win32')( 'classifies an unreadable owned lock as unavailable', async () => { diff --git a/packages/core/src/services/session-writer-lease.ts b/packages/core/src/services/session-writer-lease.ts index dae0a72d920..0f4b8a6c941 100644 --- a/packages/core/src/services/session-writer-lease.ts +++ b/packages/core/src/services/session-writer-lease.ts @@ -7,6 +7,7 @@ import { execFile } from 'node:child_process'; import * as nodeConstants from 'node:constants'; import { createHash, randomUUID, type Hash } from 'node:crypto'; +import * as nodeFs from 'node:fs'; import type { Stats } from 'node:fs'; import * as fs from 'node:fs/promises'; import * as os from 'node:os'; @@ -1582,6 +1583,7 @@ export class SessionWriterLease { private terminalPromise: Promise | undefined; private operationTail: Promise = Promise.resolve(); private readonly lockRecordRaw: string; + private lockFileIdentity: { dev: number; ino: number } | undefined; private readonly retiredPath: string; private readonly claimPath: string; @@ -1935,6 +1937,7 @@ export class SessionWriterLease { ): Promise { const lease = new SessionWriterLease(lockPath, lockRecord, options); try { + lease.lockFileIdentity = lease.readVerifiedLockIdentity(); options.onOwnershipAcquired?.(lease); const snapshot = await captureTranscriptSnapshot( options.transcriptPath, @@ -2048,7 +2051,13 @@ export class SessionWriterLease { } throw new SessionWriterUnavailableError(); } - if (!stat.isFile() || stat.isSymbolicLink()) { + if ( + !stat.isFile() || + stat.isSymbolicLink() || + (this.lockFileIdentity !== undefined && + (stat.dev !== this.lockFileIdentity.dev || + stat.ino !== this.lockFileIdentity.ino)) + ) { throw new SessionWriterLostError(); } let raw: string; @@ -2072,6 +2081,75 @@ export class SessionWriterLease { return record; } + private readVerifiedLockIdentity(): { dev: number; ino: number } { + let descriptor: number; + try { + descriptor = nodeFs.openSync( + this.lockPath, + nodeConstants.O_RDONLY | + (nodeConstants.O_NOFOLLOW ?? 0) | + (nodeConstants.O_NONBLOCK ?? 0), + ); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === 'ENOENT' || code === 'ELOOP') { + throw new SessionWriterLostError(); + } + throw new SessionWriterUnavailableError(); + } + try { + const stat = nodeFs.fstatSync(descriptor); + if (!stat.isFile()) throw new SessionWriterLostError(); + if (!hasVerifiableInode(stat.ino)) { + throw new SessionWriterUnavailableError(); + } + let pathStat: Stats; + try { + pathStat = nodeFs.lstatSync(this.lockPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + throw new SessionWriterLostError(); + } + throw new SessionWriterUnavailableError(); + } + if ( + !pathStat.isFile() || + pathStat.isSymbolicLink() || + pathStat.dev !== stat.dev || + pathStat.ino !== stat.ino + ) { + throw new SessionWriterLostError(); + } + const raw = nodeFs.readFileSync(descriptor, 'utf8'); + const record = parseLockRecord(raw); + if ( + !record || + !isActiveLockRecord(record) || + record.owner_id !== this.ownerId || + raw !== this.lockRecordRaw + ) { + throw new SessionWriterLostError(); + } + return { dev: stat.dev, ino: stat.ino }; + } catch (error) { + if (error instanceof SessionWriterError) throw error; + throw new SessionWriterUnavailableError(); + } finally { + nodeFs.closeSync(descriptor); + } + } + + /** Verify ownership after the transcript snapshot intentionally changes. */ + assertCleanupOwned(): void { + if (this.released) throw new SessionWriterLostError(); + const expected = this.lockFileIdentity; + if (expected === undefined) throw new SessionWriterUnavailableError(); + const current = this.readVerifiedLockIdentity(); + if (current.dev !== expected.dev || current.ino !== expected.ino) { + throw new SessionWriterLostError(); + } + } + assertOwnedAndUnchanged(): Promise { return this.runExclusive(() => this.assertOwnedAndUnchangedOnce()); } diff --git a/packages/core/src/services/sessionService.corruption.test.ts b/packages/core/src/services/sessionService.corruption.test.ts index 6c982bfd310..3fb11161a78 100644 --- a/packages/core/src/services/sessionService.corruption.test.ts +++ b/packages/core/src/services/sessionService.corruption.test.ts @@ -997,7 +997,7 @@ describe('SessionService lifecycle maintenance', () => { }); it.each(['archive', 'unarchive'] as const)( - 'does not swallow a generation rejection at the %s ledger fence', + 'finishes the %s ledger move after the generation closes', async (action) => { const state = action === 'archive' ? 'active' : 'archived'; const { service, sessionId, paths } = createHarness('transcript', state); @@ -1017,12 +1017,47 @@ describe('SessionService lifecycle maintenance', () => { .mockImplementation(() => { throw generationChanged; }); + const assertCleanupOwned = vi.fn(); const result = await service[`${action}Sessions`]([sessionId], { assertCanMutate, + assertCleanupOwned, }); - expect(result.errors[0]?.error).toBe(generationChanged); + expect(result.errors).toEqual([]); + expect(assertCanMutate).toHaveBeenCalledOnce(); + expect(assertCleanupOwned).toHaveBeenCalled(); + expect(fs.existsSync(sourceLedger)).toBe(false); + expect(fs.existsSync(destinationLedger)).toBe(true); + }, + ); + + it.each(['archive', 'unarchive'] as const)( + 'stops the %s ledger move after cleanup ownership is lost', + async (action) => { + const state = action === 'archive' ? 'active' : 'archived'; + const { service, sessionId, paths } = createHarness('transcript', state); + const sourcePath = paths[state]; + const destinationPath = + action === 'archive' ? paths.archived : paths.active; + const sourceLedger = sourcePath.replace(/\.jsonl$/, '.ledger.jsonl'); + const destinationLedger = destinationPath.replace( + /\.jsonl$/, + '.ledger.jsonl', + ); + fs.writeFileSync(sourceLedger, '{"promptId":"p1"}\n'); + const ownershipLost = new Error('writer ownership lost'); + + const result = await service[`${action}Sessions`]([sessionId], { + assertCanMutate: vi.fn(), + assertCleanupOwned: () => { + throw ownershipLost; + }, + }); + + expect(result.errors[0]?.error).toBe(ownershipLost); + expect(fs.existsSync(sourcePath)).toBe(false); + expect(fs.existsSync(destinationPath)).toBe(true); expect(fs.existsSync(sourceLedger)).toBe(true); expect(fs.existsSync(destinationLedger)).toBe(false); }, diff --git a/packages/core/src/services/sessionService.test.ts b/packages/core/src/services/sessionService.test.ts index 5c902a1d95d..12f96bb22d4 100644 --- a/packages/core/src/services/sessionService.test.ts +++ b/packages/core/src/services/sessionService.test.ts @@ -47,6 +47,7 @@ import { CompressionStatus } from '../core/turn.js'; import type { ChatRecord } from './chatRecordingService.js'; import * as jsonl from '../utils/jsonl-utils.js'; import { readSessionPrs, writeSessionPrs } from './session-pr-service.js'; +import { SessionWriterLostError } from './session-writer-lease.js'; vi.mock('./usageHistoryService.js', () => ({ prepareUsageBeforeTranscriptDeletion: vi.fn().mockResolvedValue({ @@ -1825,6 +1826,56 @@ describe('SessionService', () => { expect(unlinkSyncSpy).not.toHaveBeenCalled(); }); + it('finishes committed deletion cleanup after the generation closes', async () => { + vi.mocked(jsonl.readLines).mockResolvedValue([recordA1]); + const assertCanMutate = vi + .fn() + .mockImplementationOnce(() => undefined) + .mockImplementation(() => { + throw new Error('generation changed'); + }); + const assertCleanupOwned = vi.fn(); + const removeOrganizationSpy = vi + .spyOn(SessionOrganizationService.prototype, 'removeSession') + .mockImplementation(async (_sessionId, options) => { + options?.assertCanCommit?.(); + }); + + await expect( + sessionService.removeSession(sessionIdA, { + assertCanMutate, + assertCleanupOwned, + }), + ).resolves.toBe(true); + + expect(assertCanMutate).toHaveBeenCalledOnce(); + expect(assertCleanupOwned).toHaveBeenCalledTimes(6); + expect(removeOrganizationSpy).toHaveBeenCalledWith(sessionIdA, { + assertCanCommit: assertCleanupOwned, + }); + expect(rmSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`file-history/${sessionIdA}`), + { recursive: true, force: true }, + ); + }); + + it('stops committed deletion cleanup after writer ownership is lost', async () => { + vi.mocked(jsonl.readLines).mockResolvedValue([recordA1]); + const ownershipLost = new Error('writer ownership lost'); + + await expect( + sessionService.removeSession(sessionIdA, { + assertCanMutate: vi.fn(), + assertCleanupOwned: () => { + throw ownershipLost; + }, + }), + ).rejects.toBe(ownershipLost); + + expect(unlinkSyncSpy).toHaveBeenCalledTimes(2); + expect(rmSyncSpy).not.toHaveBeenCalled(); + }); + it('does not commit usage when transcript deletion fails', async () => { vi.mocked(jsonl.readLines).mockResolvedValue([recordA1]); const unlinkError = Object.assign(new Error('permission denied'), { @@ -2550,7 +2601,7 @@ describe('SessionService', () => { ); }); - it('passes the generation fence to an asynchronous pr-sidecar commit', async () => { + it('passes cleanup ownership to an asynchronous pr-sidecar commit', async () => { mockActiveSessionOnly(); existsSyncSpy.mockImplementation((filePath) => { const value = filePath.toString(); @@ -2568,19 +2619,59 @@ describe('SessionService', () => { .mockResolvedValueOnce([entry]) .mockResolvedValueOnce([entry]); const assertCanMutate = vi.fn(); + const assertCleanupOwned = vi.fn(); const result = await sessionService.archiveSessions([sessionIdA], { assertCanMutate, + assertCleanupOwned, }); expect(result.errors).toEqual([]); + expect(assertCanMutate).toHaveBeenCalledOnce(); expect(writeSessionPrs).toHaveBeenCalledWith( expect.stringContaining(`/chats/archive/${sessionIdA}.pr.json`), [entry], - { assertCanCommit: assertCanMutate }, + { assertCanCommit: assertCleanupOwned }, ); }); + it('does not swallow writer ownership loss during a pr-sidecar commit', async () => { + mockActiveSessionOnly(); + existsSyncSpy.mockImplementation((filePath) => { + const value = filePath.toString(); + return ( + value.endsWith(`/chats/${sessionIdA}.pr.json`) || + value.endsWith(`/chats/archive/${sessionIdA}.pr.json`) + ); + }); + const entry = { + number: 100, + url: 'https://github.com/o/r/pull/100', + createdAt: '2026-08-20T00:00:00.000Z', + }; + vi.mocked(readSessionPrs).mockResolvedValue([entry]); + vi.mocked(writeSessionPrs).mockImplementation( + async (_filePath, _entries, options) => { + options?.assertCanCommit?.(); + }, + ); + const ownershipLost = new SessionWriterLostError(); + const assertCleanupOwned = vi + .fn() + .mockImplementationOnce(() => undefined) + .mockImplementation(() => { + throw ownershipLost; + }); + + const result = await sessionService.archiveSessions([sessionIdA], { + assertCanMutate: vi.fn(), + assertCleanupOwned, + }); + + expect(result.errors[0]?.error).toBe(ownershipLost); + expect(assertCleanupOwned).toHaveBeenCalledTimes(2); + }); + it('should archive JSONL and warn when archiving worktree sidecar fails', async () => { mockActiveSessionOnly(); mockActiveWorktreeSidecarOnly(); @@ -2614,7 +2705,7 @@ describe('SessionService', () => { ); }); - it('rechecks the generation before moving the active worktree sidecar', async () => { + it('finishes moving active sidecars after the generation closes', async () => { mockActiveSessionOnly(); mockActiveWorktreeSidecarOnly(); const generationChanged = new Error('generation changed'); @@ -2624,12 +2715,40 @@ describe('SessionService', () => { .mockImplementation(() => { throw generationChanged; }); + const assertCleanupOwned = vi.fn(); const result = await sessionService.archiveSessions([sessionIdA], { assertCanMutate, + assertCleanupOwned, }); - expect(result.errors[0]?.error).toBe(generationChanged); + expect(result.archived).toEqual([sessionIdA]); + expect(result.errors).toEqual([]); + expect(assertCanMutate).toHaveBeenCalledOnce(); + expect(assertCleanupOwned).toHaveBeenCalled(); + expect(renameSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`/chats/${sessionIdA}.jsonl`), + expect.stringContaining(`/chats/archive/${sessionIdA}.jsonl`), + ); + expect(renameSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`/chats/${sessionIdA}.worktree.json`), + expect.stringContaining(`/chats/archive/${sessionIdA}.worktree.json`), + ); + }); + + it('stops archive sidecar cleanup after writer ownership is lost', async () => { + mockActiveSessionOnly(); + mockActiveWorktreeSidecarOnly(); + const ownershipLost = new Error('writer ownership lost'); + + const result = await sessionService.archiveSessions([sessionIdA], { + assertCanMutate: vi.fn(), + assertCleanupOwned: () => { + throw ownershipLost; + }, + }); + + expect(result.errors[0]?.error).toBe(ownershipLost); expect(renameSyncSpy).toHaveBeenCalledWith( expect.stringContaining(`/chats/${sessionIdA}.jsonl`), expect.stringContaining(`/chats/archive/${sessionIdA}.jsonl`), @@ -2834,6 +2953,31 @@ describe('SessionService', () => { expect(prepareUsageBeforeTranscriptDeletion).not.toHaveBeenCalled(); expect(commitUsageBeforeTranscriptDeletion).not.toHaveBeenCalled(); }); + + it('finishes archive conflict cleanup after the generation closes', async () => { + vi.mocked(jsonl.readLines).mockResolvedValue([recordA1]); + const assertCanMutate = vi + .fn() + .mockImplementationOnce(() => undefined) + .mockImplementation(() => { + throw new Error('generation changed'); + }); + const assertCleanupOwned = vi.fn(); + + const result = await sessionService.archiveSessions([sessionIdA], { + resolveConflicts: true, + assertCanMutate, + assertCleanupOwned, + }); + + expect(result).toMatchObject({ + archived: [sessionIdA], + resolvedConflicts: [sessionIdA], + errors: [], + }); + expect(assertCanMutate).toHaveBeenCalledOnce(); + expect(assertCleanupOwned).toHaveBeenCalledTimes(3); + }); }); describe('unarchiveSessions', () => { @@ -2992,6 +3136,31 @@ describe('SessionService', () => { expect(commitUsageBeforeTranscriptDeletion).not.toHaveBeenCalled(); }); + it('finishes unarchive conflict cleanup after the generation closes', async () => { + vi.mocked(jsonl.readLines).mockResolvedValue([recordA1]); + const assertCanMutate = vi + .fn() + .mockImplementationOnce(() => undefined) + .mockImplementation(() => { + throw new Error('generation changed'); + }); + const assertCleanupOwned = vi.fn(); + + const result = await sessionService.unarchiveSessions([sessionIdA], { + resolveConflicts: true, + assertCanMutate, + assertCleanupOwned, + }); + + expect(result).toMatchObject({ + unarchived: [sessionIdA], + resolvedConflicts: [sessionIdA], + errors: [], + }); + expect(assertCanMutate).toHaveBeenCalledOnce(); + expect(assertCleanupOwned).toHaveBeenCalledTimes(3); + }); + it('should recreate active chats directory before moving archived sessions', async () => { mockArchivedSessionOnly(); @@ -3078,7 +3247,7 @@ describe('SessionService', () => { ); }); - it('rechecks the generation before moving the archived worktree sidecar', async () => { + it('finishes moving archived sidecars after the generation closes', async () => { mockArchivedSessionOnly(); mockArchivedWorktreeSidecarOnly(); const generationChanged = new Error('generation changed'); @@ -3088,12 +3257,40 @@ describe('SessionService', () => { .mockImplementation(() => { throw generationChanged; }); + const assertCleanupOwned = vi.fn(); const result = await sessionService.unarchiveSessions([sessionIdA], { assertCanMutate, + assertCleanupOwned, + }); + + expect(result.unarchived).toEqual([sessionIdA]); + expect(result.errors).toEqual([]); + expect(assertCanMutate).toHaveBeenCalledOnce(); + expect(assertCleanupOwned).toHaveBeenCalled(); + expect(renameSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`/chats/archive/${sessionIdA}.jsonl`), + expect.stringContaining(`/chats/${sessionIdA}.jsonl`), + ); + expect(renameSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`/chats/archive/${sessionIdA}.worktree.json`), + expect.stringContaining(`/chats/${sessionIdA}.worktree.json`), + ); + }); + + it('stops unarchive sidecar cleanup after writer ownership is lost', async () => { + mockArchivedSessionOnly(); + mockArchivedWorktreeSidecarOnly(); + const ownershipLost = new Error('writer ownership lost'); + + const result = await sessionService.unarchiveSessions([sessionIdA], { + assertCanMutate: vi.fn(), + assertCleanupOwned: () => { + throw ownershipLost; + }, }); - expect(result.errors[0]?.error).toBe(generationChanged); + expect(result.errors[0]?.error).toBe(ownershipLost); expect(renameSyncSpy).toHaveBeenCalledWith( expect.stringContaining(`/chats/archive/${sessionIdA}.jsonl`), expect.stringContaining(`/chats/${sessionIdA}.jsonl`), diff --git a/packages/core/src/services/sessionService.ts b/packages/core/src/services/sessionService.ts index 49e90d99520..34149976448 100644 --- a/packages/core/src/services/sessionService.ts +++ b/packages/core/src/services/sessionService.ts @@ -63,6 +63,7 @@ import { type SessionRestoreProjection, } from './session-transcript-reader.js'; import { + SessionWriterError, SessionWriterLease, SessionTranscriptChangedError, SessionTranscriptIdentityUnavailableError, @@ -306,6 +307,7 @@ export interface RemoveSessionsResult { export interface RemoveSessionOptions { assertStorageUnchanged?: () => Promise; assertCanMutate?: () => void; + assertCleanupOwned?: () => void; } export interface ArchiveSessionsResult { @@ -321,6 +323,7 @@ export interface ArchiveSessionsOptions { resolveConflicts?: boolean; assertStorageUnchanged?: () => Promise; assertCanMutate?: () => void; + assertCleanupOwned?: () => void; } export interface UnarchiveSessionsResult { @@ -336,6 +339,7 @@ export interface UnarchiveSessionsOptions { resolveConflicts?: boolean; assertStorageUnchanged?: () => Promise; assertCanMutate?: () => void; + assertCleanupOwned?: () => void; } export interface SessionServiceOptions { @@ -1529,7 +1533,7 @@ export class SessionService { private async removeSessionOrganization( sessionId: string, - assertCanMutate?: () => void, + assertCleanupOwned?: () => void, propagateFailure = false, ): Promise { try { @@ -1539,15 +1543,16 @@ export class SessionService { this.warn(message); }, ); - if (assertCanMutate) { + if (assertCleanupOwned) { await service.removeSession(sessionId, { - assertCanCommit: assertCanMutate, + assertCanCommit: assertCleanupOwned, }); } else { await service.removeSession(sessionId); } } catch (error) { - assertCanMutate?.(); + if (error instanceof SessionWriterError) throw error; + assertCleanupOwned?.(); if (propagateFailure) throw error; this.warn( `removeSession: failed to clear session organization for ${sessionId}: ${error}`, @@ -2600,7 +2605,7 @@ export class SessionService { await assertDurableDirectoryHandle(parent); } await this.cleanupRemovedSessionStateInternal(sessionId, options, true); - options.assertCanMutate?.(); + (options.assertCleanupOwned ?? options.assertCanMutate)?.(); for (const parent of parents) { await syncDurableDirectory(parent); } @@ -2742,10 +2747,12 @@ export class SessionService { propagateOrganizationFailure: boolean, ): Promise { this.cleanupRemovedSessionFiles(sessionId, options); - options.assertCanMutate?.(); + const assertCleanupOwned = + options.assertCleanupOwned ?? options.assertCanMutate; + assertCleanupOwned?.(); await this.removeSessionOrganization( sessionId, - options.assertCanMutate, + assertCleanupOwned, propagateOrganizationFailure, ); } @@ -2754,13 +2761,15 @@ export class SessionService { sessionId: string, options: RemoveSessionOptions, ): void { - options.assertCanMutate?.(); + const assertCleanupOwned = + options.assertCleanupOwned ?? options.assertCanMutate; + assertCleanupOwned?.(); this.removeWorktreeSidecars(sessionId); - options.assertCanMutate?.(); + assertCleanupOwned?.(); this.removePrSidecars(sessionId); - options.assertCanMutate?.(); + assertCleanupOwned?.(); this.removePromptLedgers(sessionId); - options.assertCanMutate?.(); + assertCleanupOwned?.(); this.removeFileHistoryBackups(sessionId); } @@ -2809,23 +2818,26 @@ export class SessionService { this.assertMaintainableSessionUnchanged(sessionId, snapshot); this.removeFileIfExists(active.filePath); try { - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); await this.movePrSidecar( this.getPrSessionPathForState(sessionId, 'active'), this.getPrSessionPathForState(sessionId, 'archived'), - options.assertCanMutate, + options.assertCleanupOwned, ); } catch (sidecarError) { - options.assertCanMutate?.(); + if (sidecarError instanceof SessionWriterError) { + throw sidecarError; + } + options.assertCleanupOwned?.(); this.warn( `archiveSessions: failed to merge active pr sidecar for ${sessionId}: ${sidecarError}`, ); } - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); this.removeFileIfExists( this.getWorktreeSessionPathForState(sessionId, 'active'), ); - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); this.removeFileIfExists( this.getPromptLedgerPathForState(sessionId, 'active'), ); @@ -2861,7 +2873,7 @@ export class SessionService { } catch (error) { throw this.sessionFileMoveError('archive', error); } - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); try { this.moveOptionalFile(activeSidecar, archivedSidecar); } catch (sidecarError) { @@ -2873,19 +2885,25 @@ export class SessionService { await this.movePrSidecar( this.getPrSessionPathForState(sessionId, 'active'), this.getPrSessionPathForState(sessionId, 'archived'), - options.assertCanMutate, + options.assertCleanupOwned, ); } catch (sidecarError) { - options.assertCanMutate?.(); + if (sidecarError instanceof SessionWriterError) { + throw sidecarError; + } + options.assertCleanupOwned?.(); this.warn( `archiveSessions: failed to move pr sidecar for ${sessionId}: ${sidecarError}`, ); } try { - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); this.moveLedgerSidecar(activeLedger, archivedLedger); } catch (ledgerError) { - options.assertCanMutate?.(); + if (ledgerError instanceof SessionWriterError) { + throw ledgerError; + } + options.assertCleanupOwned?.(); this.warn( `archiveSessions: failed to move prompt ledger for ${sessionId} from ${activeLedger} to ${archivedLedger}: ${ledgerError}`, ); @@ -2950,23 +2968,26 @@ export class SessionService { this.assertMaintainableSessionUnchanged(sessionId, snapshot); this.removeFileIfExists(archived.filePath); try { - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); await this.movePrSidecar( this.getPrSessionPathForState(sessionId, 'archived'), this.getPrSessionPathForState(sessionId, 'active'), - options.assertCanMutate, + options.assertCleanupOwned, ); } catch (sidecarError) { - options.assertCanMutate?.(); + if (sidecarError instanceof SessionWriterError) { + throw sidecarError; + } + options.assertCleanupOwned?.(); this.warn( `unarchiveSessions: failed to merge archived pr sidecar for ${sessionId}: ${sidecarError}`, ); } - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); this.removeFileIfExists( this.getWorktreeSessionPathForState(sessionId, 'archived'), ); - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); this.removeFileIfExists( this.getPromptLedgerPathForState(sessionId, 'archived'), ); @@ -2994,7 +3015,7 @@ export class SessionService { } catch (error) { throw this.sessionFileMoveError('unarchive', error); } - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); try { this.moveOptionalFile(archivedSidecar, activeSidecar); } catch (sidecarError) { @@ -3006,10 +3027,13 @@ export class SessionService { await this.movePrSidecar( this.getPrSessionPathForState(sessionId, 'archived'), this.getPrSessionPathForState(sessionId, 'active'), - options.assertCanMutate, + options.assertCleanupOwned, ); } catch (sidecarError) { - options.assertCanMutate?.(); + if (sidecarError instanceof SessionWriterError) { + throw sidecarError; + } + options.assertCleanupOwned?.(); this.warn( `unarchiveSessions: failed to move pr sidecar for ${sessionId}: ${sidecarError}`, ); @@ -3023,10 +3047,13 @@ export class SessionService { 'active', ); try { - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); this.moveLedgerSidecar(archivedLedger, activeLedger); } catch (ledgerError) { - options.assertCanMutate?.(); + if (ledgerError instanceof SessionWriterError) { + throw ledgerError; + } + options.assertCleanupOwned?.(); this.warn( `unarchiveSessions: failed to move prompt ledger for ${sessionId} from ${archivedLedger} to ${activeLedger}: ${ledgerError}`, ); From 429da3ca863b04479143db8e71dbb1921c7c952d Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Thu, 27 Aug 2026 22:33:42 +0800 Subject: [PATCH 02/12] codex: fix CI failure on PR #10300 Co-authored-by: Qwen-Coder --- packages/core/src/config/config.test.ts | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/packages/core/src/config/config.test.ts b/packages/core/src/config/config.test.ts index 26fc348be68..19ec48be93d 100644 --- a/packages/core/src/config/config.test.ts +++ b/packages/core/src/config/config.test.ts @@ -3723,6 +3723,14 @@ describe('Server Config (config.ts)', () => { } return result; }); + const actualFs = + await vi.importActual('node:fs'); + (fs.readFileSync as Mock).mockImplementation( + (pathOrDescriptor: unknown) => + typeof pathOrDescriptor === 'number' + ? actualFs.readFileSync(pathOrDescriptor, 'utf8') + : undefined, + ); try { const initialize = config.initialize(); From e773f6cc567c2d41ef450346f2d6886613049289 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Fri, 28 Aug 2026 01:33:51 +0800 Subject: [PATCH 03/12] codex: address PR review feedback (#10300) Co-authored-by: Qwen-Coder --- .../standalone-session-service.test.ts | 56 +++++++++++++++++++ .../standalone-session-service.ts | 9 ++- 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/packages/cli/src/serve/conversations/standalone-session-service.test.ts b/packages/cli/src/serve/conversations/standalone-session-service.test.ts index e286aad2cf5..ceb6de6e56c 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.test.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.test.ts @@ -18,6 +18,7 @@ import { SessionService, SessionStorageEntryError, SessionTranscriptDurabilityError, + SessionWriterLostError, writeSessionPrs, } from '@qwen-code/qwen-code-core'; import { promises as fs } from 'node:fs'; @@ -875,6 +876,61 @@ describe('StandaloneSessionService', () => { expect(harness.deletionJournal.clear).toHaveBeenCalledOnce(); }); + it('does not park a lost lease before exact deletion recovery', async () => { + mockActiveStandalone(); + const harness = createHarness(); + const lostLease = mockWriterLease(); + lostLease.release.mockRejectedValue(new SessionWriterLostError()); + const nextLease = { + assertOwnedAndUnchanged: vi.fn(async () => undefined), + assertCleanupOwned: vi.fn(), + release: vi.fn(async () => undefined), + isReleased: false, + isReleaseDurabilityPending: false, + }; + vi.mocked(SessionService.prototype.acquireSessionWriterLease) + .mockResolvedValueOnce(lostLease as never) + .mockResolvedValueOnce(nextLease as never); + vi.spyOn( + SessionService.prototype, + 'removeSessionTranscriptForLifecycle', + ).mockResolvedValue(true); + vi.spyOn( + SessionService.prototype, + 'cleanupRemovedSessionStateForLifecycle', + ).mockResolvedValue(); + + await expect(harness.service.delete([sessionId])).resolves.toEqual({ + removed: [sessionId], + notFound: [], + errors: [], + fileCleanupPending: [sessionId], + }); + + vi.mocked( + SessionService.prototype.findSessionIdIgnoringCase, + ).mockResolvedValue(undefined); + harness.deletionJournal.read.mockResolvedValueOnce( + deletionEntry() as never, + ); + harness.inspectStandaloneDeletionPaths.mockResolvedValueOnce({ + status: 'absent', + }); + + await expect(harness.service.delete([sessionId])).resolves.toEqual({ + removed: [sessionId], + notFound: [], + errors: [], + fileCleanupPending: [], + }); + + expect( + SessionService.prototype.acquireSessionWriterLease, + ).toHaveBeenCalledTimes(2); + expect(nextLease.release).toHaveBeenCalledOnce(); + expect(harness.deletionJournal.clear).toHaveBeenCalledOnce(); + }); + it('rolls back a staged directory when recovery finds the transcript intact', async () => { mockActiveStandalone(); const harness = createHarness(); diff --git a/packages/cli/src/serve/conversations/standalone-session-service.ts b/packages/cli/src/serve/conversations/standalone-session-service.ts index f5c61d02968..d848b570475 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.ts @@ -27,6 +27,7 @@ import { SessionStorageEntryError, SessionTranscriptDurabilityError, SessionTranscriptChangedError, + SessionWriterLostError, SessionWriterUnavailableError, type ApprovalMode, type SessionArchiveState, @@ -1379,7 +1380,13 @@ export class StandaloneSessionService { this.pendingLifecycleLeaseReleases.delete(pendingKey); } return true; - } catch { + } catch (error) { + if (error instanceof SessionWriterLostError) { + if (this.pendingLifecycleLeaseReleases.get(pendingKey) === lease) { + this.pendingLifecycleLeaseReleases.delete(pendingKey); + } + return false; + } // The lease clears retryable terminal failures itself. } } From 79f5c29fadfe46d6e1bad5d2e45033cfeced789a Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Fri, 28 Aug 2026 10:13:54 +0800 Subject: [PATCH 04/12] codex: address PR review feedback (#10300) Co-authored-by: Qwen-Coder --- .../standalone-session-service.test.ts | 129 ++++++++++++++++++ .../standalone-session-service.ts | 33 ++++- .../src/services/session-writer-lease.test.ts | 43 ++++++ .../core/src/services/session-writer-lease.ts | 34 +++-- 4 files changed, 223 insertions(+), 16 deletions(-) diff --git a/packages/cli/src/serve/conversations/standalone-session-service.test.ts b/packages/cli/src/serve/conversations/standalone-session-service.test.ts index ceb6de6e56c..7a482f5725f 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.test.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.test.ts @@ -711,6 +711,72 @@ describe('StandaloneSessionService', () => { }); }); + it('invalidates the catalog when archive cleanup fails after the move', async () => { + mockActiveStandalone(); + const harness = createHarness(); + mockWriterLease(); + vi.mocked(SessionService.prototype.getSessionLocation) + .mockResolvedValueOnce('active') + .mockResolvedValueOnce('active') + .mockResolvedValue('archived'); + vi.spyOn(SessionService.prototype, 'archiveSessions').mockResolvedValue({ + archived: [], + alreadyArchived: [], + resolvedConflicts: [], + notFound: [], + errors: [{ sessionId, error: new SessionWriterLostError() }], + }); + + await expect(harness.service.archive([sessionId])).resolves.toMatchObject({ + archived: [], + errors: [ + { + sessionId, + code: 'standalone_session_operation_failed', + }, + ], + }); + + expect(harness.bridge.markSessionCatalogChanged).toHaveBeenCalledOnce(); + expect(harness.invalidateSessionListCache).toHaveBeenCalledWith( + harness.runtime, + ); + }); + + it('invalidates the catalog when unarchive cleanup fails after the move', async () => { + mockArchivedStandalone(); + const harness = createHarness(); + mockWriterLease(); + vi.mocked(SessionService.prototype.getSessionLocation) + .mockResolvedValueOnce('archived') + .mockResolvedValueOnce('archived') + .mockResolvedValue('active'); + vi.spyOn(SessionService.prototype, 'unarchiveSessions').mockResolvedValue({ + unarchived: [], + alreadyActive: [], + resolvedConflicts: [], + notFound: [], + errors: [{ sessionId, error: new SessionWriterLostError() }], + }); + + await expect(harness.service.unarchive([sessionId])).resolves.toMatchObject( + { + unarchived: [], + errors: [ + { + sessionId, + code: 'standalone_session_operation_failed', + }, + ], + }, + ); + + expect(harness.bridge.markSessionCatalogChanged).toHaveBeenCalledOnce(); + expect(harness.invalidateSessionListCache).toHaveBeenCalledWith( + harness.runtime, + ); + }); + it('journals, stages, commits, and cleans a standalone deletion', async () => { mockActiveStandalone(); const harness = createHarness(); @@ -931,6 +997,69 @@ describe('StandaloneSessionService', () => { expect(harness.deletionJournal.clear).toHaveBeenCalledOnce(); }); + it('evicts a parked lost lease before exact deletion recovery', async () => { + mockActiveStandalone(); + const harness = createHarness(); + const lostLease = mockWriterLease(); + lostLease.release + .mockImplementationOnce(async () => { + lostLease.isReleased = true; + lostLease.isReleaseDurabilityPending = true; + throw new Error('release I/O failed'); + }) + .mockRejectedValueOnce(new Error('release I/O failed')) + .mockRejectedValueOnce(new SessionWriterLostError()); + const nextLease = { + assertOwnedAndUnchanged: vi.fn(async () => undefined), + assertCleanupOwned: vi.fn(), + release: vi.fn(async () => undefined), + isReleased: false, + isReleaseDurabilityPending: false, + }; + vi.mocked(SessionService.prototype.acquireSessionWriterLease) + .mockResolvedValueOnce(lostLease as never) + .mockResolvedValueOnce(nextLease as never); + vi.spyOn( + SessionService.prototype, + 'removeSessionTranscriptForLifecycle', + ).mockResolvedValue(true); + vi.spyOn( + SessionService.prototype, + 'cleanupRemovedSessionStateForLifecycle', + ).mockResolvedValue(); + + await expect(harness.service.delete([sessionId])).resolves.toEqual({ + removed: [sessionId], + notFound: [], + errors: [], + fileCleanupPending: [sessionId], + }); + + vi.mocked( + SessionService.prototype.findSessionIdIgnoringCase, + ).mockResolvedValue(undefined); + harness.deletionJournal.read.mockResolvedValueOnce( + deletionEntry() as never, + ); + harness.inspectStandaloneDeletionPaths.mockResolvedValueOnce({ + status: 'absent', + }); + + await expect(harness.service.delete([sessionId])).resolves.toEqual({ + removed: [sessionId], + notFound: [], + errors: [], + fileCleanupPending: [], + }); + + expect(lostLease.release).toHaveBeenCalledTimes(3); + expect( + SessionService.prototype.acquireSessionWriterLease, + ).toHaveBeenCalledTimes(2); + expect(nextLease.release).toHaveBeenCalledOnce(); + expect(harness.deletionJournal.clear).toHaveBeenCalledOnce(); + }); + it('rolls back a staged directory when recovery finds the transcript intact', async () => { mockActiveStandalone(); const harness = createHarness(); diff --git a/packages/cli/src/serve/conversations/standalone-session-service.ts b/packages/cli/src/serve/conversations/standalone-session-service.ts index d848b570475..725d43a2407 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.ts @@ -1338,7 +1338,8 @@ export class StandaloneSessionService { const pending = this.pendingLifecycleLeaseReleases.get(pendingKey); if ( pending && - !(await this.releaseLifecycleLease(pending, storageSessionId)) + !(await this.releaseLifecycleLease(pending, storageSessionId)) && + this.pendingLifecycleLeaseReleases.get(pendingKey) === pending ) { throw new SessionWriterUnavailableError({ message: 'A previous session writer lease is still being released.', @@ -1512,6 +1513,26 @@ export class StandaloneSessionService { return { sessionId, code: mapped.code, message: mapped.message }; } + private async reconcileCatalogAfterLifecycleError( + runtime: WorkspaceRuntime, + sessionId: string, + expectedLocation: 'active' | 'archived', + ): Promise { + try { + const durable = await this.inspectStoredStandalone(runtime, sessionId); + if ( + durable.kind !== 'standalone' || + durable.location !== expectedLocation + ) { + return; + } + runtime.bridge.markSessionCatalogChanged(); + this.options.invalidateSessionListCache(runtime); + } catch { + return; + } + } + private async archiveMany( sessionIds: string[], ): Promise { @@ -1601,6 +1622,11 @@ export class StandaloneSessionService { runtime.bridge.markSessionCatalogChanged(); this.options.invalidateSessionListCache(runtime); } catch (error) { + await this.reconcileCatalogAfterLifecycleError( + runtime, + sessionId, + 'archived', + ); if ( error instanceof StandaloneSessionServiceError && error.code === 'standalone_session_not_found' @@ -1705,6 +1731,11 @@ export class StandaloneSessionService { runtime.bridge.markSessionCatalogChanged(); this.options.invalidateSessionListCache(runtime); } catch (error) { + await this.reconcileCatalogAfterLifecycleError( + runtime, + sessionId, + 'active', + ); if ( error instanceof StandaloneSessionServiceError && error.code === 'standalone_session_not_found' diff --git a/packages/core/src/services/session-writer-lease.test.ts b/packages/core/src/services/session-writer-lease.test.ts index ded60f8d822..aadd9908c77 100644 --- a/packages/core/src/services/session-writer-lease.test.ts +++ b/packages/core/src/services/session-writer-lease.test.ts @@ -113,6 +113,26 @@ const readFileFault = vi.hoisted(() => ({ afterRead: undefined as (() => Promise | void) | undefined, })); +const descriptorReadHook = vi.hoisted(() => ({ + afterRead: undefined as (() => void) | undefined, +})); + +vi.mock('node:fs', async (importOriginal) => { + const actual = await importOriginal(); + const readFileSyncWithHook = ((...args: unknown[]) => { + const result = (actual.readFileSync as (...readArgs: unknown[]) => unknown)( + ...args, + ); + if (typeof args[0] === 'number') { + const afterRead = descriptorReadHook.afterRead; + descriptorReadHook.afterRead = undefined; + afterRead?.(); + } + return result; + }) as typeof actual.readFileSync; + return { ...actual, readFileSync: readFileSyncWithHook }; +}); + vi.mock('node:fs/promises', async (importOriginal) => { const actual = await importOriginal(); return { @@ -460,6 +480,7 @@ afterEach(async () => { readFileFault.triggerCall = 0; readFileFault.calls = 0; readFileFault.afterRead = undefined; + descriptorReadHook.afterRead = undefined; setDebugLogSession(null); resetDebugLoggingState(); Storage.setRuntimeBaseDir(null); @@ -1014,6 +1035,28 @@ describe('SessionWriterLease', () => { }, ); + it.runIf(process.platform !== 'win32')( + 'rejects a lock replaced while cleanup ownership is being verified', + async () => { + const fixture = await createFixture(); + const lease = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const replacementPath = `${lockPath}.replacement`; + writeFileSync(replacementPath, readFileSync(lockPath)); + descriptorReadHook.afterRead = () => { + renameSync(replacementPath, lockPath); + }; + + expect(() => lease.assertCleanupOwned()).toThrow(SessionWriterLostError); + await expect(lease.release()).rejects.toBeInstanceOf( + SessionWriterLostError, + ); + }, + ); + it.runIf(process.platform !== 'win32')( 'rejects a byte-identical atomic replacement during acquisition', async () => { diff --git a/packages/core/src/services/session-writer-lease.ts b/packages/core/src/services/session-writer-lease.ts index 0f4b8a6c941..bf3d4ce7050 100644 --- a/packages/core/src/services/session-writer-lease.ts +++ b/packages/core/src/services/session-writer-lease.ts @@ -2103,23 +2103,26 @@ export class SessionWriterLease { if (!hasVerifiableInode(stat.ino)) { throw new SessionWriterUnavailableError(); } - let pathStat: Stats; - try { - pathStat = nodeFs.lstatSync(this.lockPath); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + const assertPathMatchesDescriptor = (): void => { + let pathStat: Stats; + try { + pathStat = nodeFs.lstatSync(this.lockPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + throw new SessionWriterLostError(); + } + throw new SessionWriterUnavailableError(); + } + if ( + !pathStat.isFile() || + pathStat.isSymbolicLink() || + pathStat.dev !== stat.dev || + pathStat.ino !== stat.ino + ) { throw new SessionWriterLostError(); } - throw new SessionWriterUnavailableError(); - } - if ( - !pathStat.isFile() || - pathStat.isSymbolicLink() || - pathStat.dev !== stat.dev || - pathStat.ino !== stat.ino - ) { - throw new SessionWriterLostError(); - } + }; + assertPathMatchesDescriptor(); const raw = nodeFs.readFileSync(descriptor, 'utf8'); const record = parseLockRecord(raw); if ( @@ -2130,6 +2133,7 @@ export class SessionWriterLease { ) { throw new SessionWriterLostError(); } + assertPathMatchesDescriptor(); return { dev: stat.dev, ino: stat.ino }; } catch (error) { if (error instanceof SessionWriterError) throw error; From 02bfccaf551acda9c4bbce5ce9eeff1a801bb3d5 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Fri, 28 Aug 2026 13:09:04 +0800 Subject: [PATCH 05/12] codex: address PR review feedback (#10300) Co-authored-by: Qwen-Coder --- docs/developers/qwen-serve-protocol.md | 4 ++ .../standalone-session-service.test.ts | 54 +++++++++++++++++++ .../standalone-session-service.ts | 49 ++++++++++------- 3 files changed, 87 insertions(+), 20 deletions(-) diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 37f79d9bbaa..ffbcd8551c2 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -2483,6 +2483,8 @@ Response: `resolveConflicts` is optional and defaults to `false`. By default, active and archived files with the same id are reported in `errors`, and neither copy is moved, removed, or overwritten. Archiving a live session still performs the strict close described above before classifying the conflict, so that close may flush queued records to the active transcript. With `resolveConflicts: true`, archive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the archived copy, removes the active copy, and reports the id in both `archived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. `errors` entries have `{ "sessionId": "", "error": "message" }`. +The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `archived` and `resolvedConflicts`. Before treating such an error as proof that the active copy or conflict remains, re-read the session list for the same workspace and use its authoritative archive state. + Lifecycle conflicts are batch item outcomes: the workspace-less and workspace-qualified routes return HTTP `200` with the conflict in `errors`. This replaces the earlier workspace-qualified HTTP `409 session_conflict` envelope; clients that called that route must inspect the batch response. Internal-runtime REST batches preserve the safe conflict message while continuing to redact other per-session failure details. ### `POST /sessions/unarchive` @@ -2509,6 +2511,8 @@ Response: `resolveConflicts` is optional and defaults to `false`. By default, simultaneous active and archived JSONL files produce a conflict in `errors`, and neither copy is moved, removed, or overwritten; an active-only session is returned in `alreadyActive`. With `resolveConflicts: true`, unarchive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the active copy, removes the archived copy, and reports the id in both `unarchived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. Archive or unarchive in flight for the same id returns `409 session_archiving` before starting the batch. +The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `unarchived` and `resolvedConflicts`. Before treating such an error as proof that the archived copy or conflict remains, re-read the session list for the same workspace and use its authoritative archive state. + ACP-over-HTTP uses the same request and response bodies through vendor methods `_qwen/sessions/archive` and `_qwen/sessions/unarchive`. The REST route table maps `POST /sessions/archive` and `POST /sessions/unarchive` to those methods for ACP transports. ### Multi-workspace live-session routing diff --git a/packages/cli/src/serve/conversations/standalone-session-service.test.ts b/packages/cli/src/serve/conversations/standalone-session-service.test.ts index 7a482f5725f..caf45d41804 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.test.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.test.ts @@ -864,6 +864,31 @@ describe('StandaloneSessionService', () => { expect(harness.removeStagedStandaloneDirectory).toHaveBeenCalledOnce(); }); + it('stops destructive cleanup when deletion loses writer ownership', async () => { + mockActiveStandalone(); + const harness = createHarness(); + mockWriterLease(); + vi.spyOn( + SessionService.prototype, + 'removeSessionTranscriptForLifecycle', + ).mockResolvedValue(true); + vi.spyOn( + SessionService.prototype, + 'cleanupRemovedSessionStateForLifecycle', + ).mockRejectedValue(new SessionWriterLostError()); + + await expect(harness.service.delete([sessionId])).resolves.toEqual({ + removed: [sessionId], + notFound: [], + errors: [], + fileCleanupPending: [sessionId], + }); + + expect(harness.bridge.deleteSessionAttachments).not.toHaveBeenCalled(); + expect(harness.removeStagedStandaloneDirectory).not.toHaveBeenCalled(); + expect(harness.deletionJournal.clear).not.toHaveBeenCalled(); + }); + it('retains deletion evidence when attachment cleanup fails', async () => { mockActiveStandalone(); const harness = createHarness(); @@ -1322,6 +1347,35 @@ describe('StandaloneSessionService', () => { expect(harness.deletionJournal.clear).not.toHaveBeenCalled(); }); + it('stops attachment cleanup when recovery loses writer ownership', async () => { + vi.spyOn( + SessionService.prototype, + 'findSessionIdIgnoringCase', + ).mockResolvedValue(undefined); + vi.spyOn( + SessionService.prototype, + 'cleanupRemovedSessionStateForLifecycle', + ).mockRejectedValue(new SessionWriterLostError()); + const harness = createHarness(); + mockWriterLease(); + harness.deletionJournal.read.mockResolvedValueOnce( + deletionEntry() as never, + ); + harness.inspectStandaloneDeletionPaths.mockResolvedValueOnce({ + status: 'absent', + }); + + await expect(harness.service.delete([sessionId])).resolves.toEqual({ + removed: [sessionId], + notFound: [], + errors: [], + fileCleanupPending: [sessionId], + }); + + expect(harness.bridge.deleteSessionAttachments).not.toHaveBeenCalled(); + expect(harness.deletionJournal.clear).not.toHaveBeenCalled(); + }); + it('reconfirms a restored normal directory before clearing recovery evidence', async () => { mockActiveStandalone(); const harness = createHarness(); diff --git a/packages/cli/src/serve/conversations/standalone-session-service.ts b/packages/cli/src/serve/conversations/standalone-session-service.ts index 725d43a2407..005e4e7ea13 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.ts @@ -27,6 +27,7 @@ import { SessionStorageEntryError, SessionTranscriptDurabilityError, SessionTranscriptChangedError, + SessionWriterError, SessionWriterLostError, SessionWriterUnavailableError, type ApprovalMode, @@ -1249,6 +1250,7 @@ export class StandaloneSessionService { cleanupPending = true; } } + let cleanupOwnershipLost = false; try { await service.cleanupRemovedSessionStateForLifecycle( record.storageSessionId, @@ -1260,15 +1262,18 @@ export class StandaloneSessionService { }, }, ); - } catch { + } catch (error) { cleanupPending = true; + cleanupOwnershipLost = error instanceof SessionWriterError; } - try { - await runtime.bridge.deleteSessionAttachments(sessionId, { - assertCanCommit: () => this.options.assertRuntimeCurrent(runtime), - }); - } catch { - cleanupPending = true; + if (!cleanupOwnershipLost) { + try { + await runtime.bridge.deleteSessionAttachments(sessionId, { + assertCanCommit: () => this.options.assertRuntimeCurrent(runtime), + }); + } catch { + cleanupPending = true; + } } if (!(await this.releaseLifecycleLease(lease, record.storageSessionId))) { cleanupPending = true; @@ -2004,6 +2009,7 @@ export class StandaloneSessionService { } } + let cleanupOwnershipLost = false; try { await service.cleanupRemovedSessionStateForLifecycle( locked.storageSessionId, @@ -2015,25 +2021,28 @@ export class StandaloneSessionService { }, }, ); - } catch { - cleanupPending = true; - } - try { - await runtime.bridge.deleteSessionAttachments(sessionId, { - assertCanCommit: () => this.options.assertRuntimeCurrent(runtime), - }); - } catch { + } catch (error) { cleanupPending = true; + cleanupOwnershipLost = error instanceof SessionWriterError; } - if (directoryWasStaged && paths.status === 'normal') { + if (!cleanupOwnershipLost) { try { - await this.options.workspace.removeStagedStandaloneDirectory( - sessionId, - paths.identity, - ); + await runtime.bridge.deleteSessionAttachments(sessionId, { + assertCanCommit: () => this.options.assertRuntimeCurrent(runtime), + }); } catch { cleanupPending = true; } + if (directoryWasStaged && paths.status === 'normal') { + try { + await this.options.workspace.removeStagedStandaloneDirectory( + sessionId, + paths.identity, + ); + } catch { + cleanupPending = true; + } + } } if ( !(await this.releaseLifecycleLease(lease, durable.storageSessionId)) From aca83d7ca36c714e282e202307e1f349eb5c8327 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Sat, 29 Aug 2026 03:51:55 +0800 Subject: [PATCH 06/12] codex: address PR review feedback (#10300) Co-authored-by: Qwen-Coder --- docs/developers/qwen-serve-protocol.md | 4 +- .../standalone-session-service.test.ts | 65 +++++++ .../standalone-session-service.ts | 9 +- .../src/serve/server/session-archive.test.ts | 126 ++++++++++++- .../cli/src/serve/server/session-archive.ts | 106 +++++------ .../src/services/session-writer-lease.test.ts | 122 +++++++++++- .../core/src/services/session-writer-lease.ts | 33 +++- .../sessionService.corruption.test.ts | 82 ++++++++ packages/core/src/services/sessionService.ts | 176 ++++++++---------- 9 files changed, 539 insertions(+), 184 deletions(-) diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 6578180236a..e4c287fb6d2 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -2500,7 +2500,7 @@ Response: `resolveConflicts` is optional and defaults to `false`. By default, active and archived files with the same id are reported in `errors`, and neither copy is moved, removed, or overwritten. Archiving a live session still performs the strict close described above before classifying the conflict, so that close may flush queued records to the active transcript. With `resolveConflicts: true`, archive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the archived copy, removes the active copy, and reports the id in both `archived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. `errors` entries have `{ "sessionId": "", "error": "message" }`. -The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `archived` and `resolvedConflicts`. Before treating such an error as proof that the active copy or conflict remains, re-read the session list for the same workspace and use its authoritative archive state. +The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `archived` and `resolvedConflicts`. Retry the same lifecycle request for the same workspace before treating the error as proof that the active copy or conflict remains. The retry reports the authoritative `alreadyArchived` state even for empty or damaged transcripts that session listing omits, and resumes pending sidecar cleanup under a fresh writer lease. Lifecycle conflicts are batch item outcomes: the workspace-less and workspace-qualified routes return HTTP `200` with the conflict in `errors`. This replaces the earlier workspace-qualified HTTP `409 session_conflict` envelope; clients that called that route must inspect the batch response. Internal-runtime REST batches preserve the safe conflict message while continuing to redact other per-session failure details. @@ -2528,7 +2528,7 @@ Response: `resolveConflicts` is optional and defaults to `false`. By default, simultaneous active and archived JSONL files produce a conflict in `errors`, and neither copy is moved, removed, or overwritten; an active-only session is returned in `alreadyActive`. With `resolveConflicts: true`, unarchive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the active copy, removes the archived copy, and reports the id in both `unarchived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. Archive or unarchive in flight for the same id returns `409 session_archiving` before starting the batch. -The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `unarchived` and `resolvedConflicts`. Before treating such an error as proof that the archived copy or conflict remains, re-read the session list for the same workspace and use its authoritative archive state. +The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `unarchived` and `resolvedConflicts`. Retry the same lifecycle request for the same workspace before treating the error as proof that the archived copy or conflict remains. The retry reports the authoritative `alreadyActive` state even for empty or damaged transcripts that session listing omits, and resumes pending sidecar cleanup under a fresh writer lease. ACP-over-HTTP uses the same request and response bodies through vendor methods `_qwen/sessions/archive` and `_qwen/sessions/unarchive`. The REST route table maps `POST /sessions/archive` and `POST /sessions/unarchive` to those methods for ACP transports. diff --git a/packages/cli/src/serve/conversations/standalone-session-service.test.ts b/packages/cli/src/serve/conversations/standalone-session-service.test.ts index caf45d41804..02c3ec76c45 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.test.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.test.ts @@ -27,6 +27,7 @@ import * as path from 'node:path'; import { afterEach, describe, expect, it, vi } from 'vitest'; import type { WorkspaceRuntime } from '../workspace-registry.js'; import { SessionArchiveCoordinator } from '../server/session-archive.js'; +import { ConversationRuntimeOwnershipError } from './conversation-runtime-errors.js'; import { StandaloneSessionService, type StandaloneSessionServiceOptions, @@ -889,6 +890,36 @@ describe('StandaloneSessionService', () => { expect(harness.deletionJournal.clear).not.toHaveBeenCalled(); }); + it('stops destructive cleanup when deletion loses runtime ownership', async () => { + mockActiveStandalone(); + const harness = createHarness(); + mockWriterLease(); + vi.spyOn( + SessionService.prototype, + 'removeSessionTranscriptForLifecycle', + ).mockResolvedValue(true); + vi.spyOn( + SessionService.prototype, + 'cleanupRemovedSessionStateForLifecycle', + ).mockRejectedValue( + new ConversationRuntimeOwnershipError( + 'conversation_runtime_unavailable', + true, + ), + ); + + await expect(harness.service.delete([sessionId])).resolves.toEqual({ + removed: [sessionId], + notFound: [], + errors: [], + fileCleanupPending: [sessionId], + }); + + expect(harness.bridge.deleteSessionAttachments).not.toHaveBeenCalled(); + expect(harness.removeStagedStandaloneDirectory).not.toHaveBeenCalled(); + expect(harness.deletionJournal.clear).not.toHaveBeenCalled(); + }); + it('retains deletion evidence when attachment cleanup fails', async () => { mockActiveStandalone(); const harness = createHarness(); @@ -1376,6 +1407,40 @@ describe('StandaloneSessionService', () => { expect(harness.deletionJournal.clear).not.toHaveBeenCalled(); }); + it('stops attachment cleanup when recovery loses runtime ownership', async () => { + vi.spyOn( + SessionService.prototype, + 'findSessionIdIgnoringCase', + ).mockResolvedValue(undefined); + vi.spyOn( + SessionService.prototype, + 'cleanupRemovedSessionStateForLifecycle', + ).mockRejectedValue( + new ConversationRuntimeOwnershipError( + 'conversation_runtime_unavailable', + true, + ), + ); + const harness = createHarness(); + mockWriterLease(); + harness.deletionJournal.read.mockResolvedValueOnce( + deletionEntry() as never, + ); + harness.inspectStandaloneDeletionPaths.mockResolvedValueOnce({ + status: 'absent', + }); + + await expect(harness.service.delete([sessionId])).resolves.toEqual({ + removed: [sessionId], + notFound: [], + errors: [], + fileCleanupPending: [sessionId], + }); + + expect(harness.bridge.deleteSessionAttachments).not.toHaveBeenCalled(); + expect(harness.deletionJournal.clear).not.toHaveBeenCalled(); + }); + it('reconfirms a restored normal directory before clearing recovery evidence', async () => { mockActiveStandalone(); const harness = createHarness(); diff --git a/packages/cli/src/serve/conversations/standalone-session-service.ts b/packages/cli/src/serve/conversations/standalone-session-service.ts index 005e4e7ea13..736389d5812 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.ts @@ -64,6 +64,7 @@ import { } from '../workspace-runtime-storage.js'; import type { WorkspaceRuntime } from '../workspace-registry.js'; import type { ConversationWorkspace } from './conversation-workspace.js'; +import { ConversationRuntimeOwnershipError } from './conversation-runtime-errors.js'; import { StandaloneDeletionJournalError, type StandaloneDeletionJournal, @@ -1264,7 +1265,9 @@ export class StandaloneSessionService { ); } catch (error) { cleanupPending = true; - cleanupOwnershipLost = error instanceof SessionWriterError; + cleanupOwnershipLost = + error instanceof SessionWriterError || + error instanceof ConversationRuntimeOwnershipError; } if (!cleanupOwnershipLost) { try { @@ -2023,7 +2026,9 @@ export class StandaloneSessionService { ); } catch (error) { cleanupPending = true; - cleanupOwnershipLost = error instanceof SessionWriterError; + cleanupOwnershipLost = + error instanceof SessionWriterError || + error instanceof ConversationRuntimeOwnershipError; } if (!cleanupOwnershipLost) { try { diff --git a/packages/cli/src/serve/server/session-archive.test.ts b/packages/cli/src/serve/server/session-archive.test.ts index 6bdbba87f6e..598d14644d3 100644 --- a/packages/cli/src/serve/server/session-archive.test.ts +++ b/packages/cli/src/serve/server/session-archive.test.ts @@ -18,8 +18,10 @@ import { type SessionWriterLease, Storage, getCronFilePath, + readSessionPrs, readCronTasks, updateCronTasks, + writeSessionPrs, } from '@qwen-code/qwen-code-core'; import { SessionArchivedError, @@ -581,7 +583,7 @@ describe('archiveDaemonSessions', () => { expect(result.errors).toEqual([{ sessionId, error: expect.any(Error) }]); }); - it('does not acquire writer leases for ids already archived or missing', async () => { + it('acquires a writer lease for already archived ids but not missing ids', async () => { const archivedId = '550e8400-e29b-41d4-a716-446655440003'; const missingId = '550e8400-e29b-41d4-a716-446655440004'; writeSessionFile(workspaceDir, archivedId, 'archived'); @@ -603,10 +605,30 @@ describe('archiveDaemonSessions', () => { notFound: [missingId], errors: [], }); - expect(acquire).not.toHaveBeenCalled(); + expect(acquire).toHaveBeenCalledTimes(1); expect(closeSession).toHaveBeenCalledTimes(2); }); + it('reconciles stranded sidecars before returning already archived', async () => { + const sessionId = '550e8400-e29b-41d4-a716-446655440103'; + writeSessionFile(workspaceDir, sessionId, 'archived'); + const service = new SessionService(workspaceDir); + const sidecars = await writeLifecycleSidecars(service, sessionId, 'active'); + + const result = await archiveDaemonSessions({ + sessionIds: [sessionId], + service, + bridge: { closeSession: vi.fn().mockResolvedValue(undefined) }, + coordinator: new SessionArchiveCoordinator(), + }); + + expect(result).toMatchObject({ + alreadyArchived: [sessionId], + errors: [], + }); + await expectLifecycleSidecarsMoved(sidecars, 'archived'); + }); + it('does not archive while another writer holds the lease', async () => { const sessionId = '550e8400-e29b-41d4-a716-446655440005'; writeSessionFile(workspaceDir, sessionId, 'active'); @@ -1079,7 +1101,7 @@ describe('unarchiveDaemonSessions', () => { vi.restoreAllMocks(); }); - it('deduplicates ids and does not lock already active or missing ids', async () => { + it('deduplicates ids and locks already active ids for reconciliation', async () => { const archivedId = '550e8400-e29b-41d4-a716-446655440011'; const activeId = '550e8400-e29b-41d4-a716-446655440012'; const missingId = '550e8400-e29b-41d4-a716-446655440013'; @@ -1099,7 +1121,7 @@ describe('unarchiveDaemonSessions', () => { notFound: [missingId], errors: [], }); - expect(acquire).toHaveBeenCalledTimes(1); + expect(acquire).toHaveBeenCalledTimes(2); expect(fs.existsSync(sessionPath(workspaceDir, archivedId, 'active'))).toBe( true, ); @@ -1108,6 +1130,29 @@ describe('unarchiveDaemonSessions', () => { ).toBe(false); }); + it('reconciles stranded sidecars before returning already active', async () => { + const sessionId = '550e8400-e29b-41d4-a716-446655440113'; + writeSessionFile(workspaceDir, sessionId, 'active'); + const service = new SessionService(workspaceDir); + const sidecars = await writeLifecycleSidecars( + service, + sessionId, + 'archived', + ); + + const result = await unarchiveDaemonSessions({ + sessionIds: [sessionId], + service, + coordinator: new SessionArchiveCoordinator(), + }); + + expect(result).toMatchObject({ + alreadyActive: [sessionId], + errors: [], + }); + await expectLifecycleSidecarsMoved(sidecars, 'active'); + }); + it('collapses case-variant spellings in one batch to a single unarchive', async () => { const sessionId = '550e8400-e29b-41d4-a716-446655440111'; writeSessionFile(workspaceDir, sessionId, 'archived'); @@ -1850,3 +1895,76 @@ function sessionPath( `${sessionId}.jsonl`, ); } + +async function writeLifecycleSidecars( + service: SessionService, + sessionId: string, + sourceState: 'active' | 'archived', +): Promise<{ + sessionId: string; + service: SessionService; + sourceState: 'active' | 'archived'; + pr: { number: number; url: string; createdAt: string }; +}> { + const worktreePath = service.getWorktreeSessionPathForArchiveState( + sessionId, + sourceState, + ); + const prPath = service.getPrSessionPathForArchiveState( + sessionId, + sourceState, + ); + const ledgerPath = path.join( + path.dirname(prPath), + `${sessionId}.ledger.jsonl`, + ); + const pr = { + number: 10300, + url: 'https://github.com/QwenLM/qwen-code/pull/10300', + createdAt: '2026-08-28T00:00:00.000Z', + }; + fs.mkdirSync(path.dirname(worktreePath), { recursive: true }); + fs.writeFileSync(worktreePath, '{}'); + await writeSessionPrs(prPath, [pr]); + fs.writeFileSync(ledgerPath, '{"promptId":"p1"}\n'); + return { sessionId, service, sourceState, pr }; +} + +async function expectLifecycleSidecarsMoved( + fixture: Awaited>, + destinationState: 'active' | 'archived', +): Promise { + const { sessionId, service, sourceState, pr } = fixture; + const sourceWorktree = service.getWorktreeSessionPathForArchiveState( + sessionId, + sourceState, + ); + const destinationWorktree = service.getWorktreeSessionPathForArchiveState( + sessionId, + destinationState, + ); + const sourcePr = service.getPrSessionPathForArchiveState( + sessionId, + sourceState, + ); + const destinationPr = service.getPrSessionPathForArchiveState( + sessionId, + destinationState, + ); + const sourceLedger = path.join( + path.dirname(sourcePr), + `${sessionId}.ledger.jsonl`, + ); + const destinationLedger = path.join( + path.dirname(destinationPr), + `${sessionId}.ledger.jsonl`, + ); + expect(fs.existsSync(sourceWorktree)).toBe(false); + expect(fs.existsSync(destinationWorktree)).toBe(true); + expect(fs.existsSync(sourcePr)).toBe(false); + await expect(readSessionPrs(destinationPr)).resolves.toEqual([pr]); + expect(fs.existsSync(sourceLedger)).toBe(false); + expect(fs.readFileSync(destinationLedger, 'utf8')).toContain( + '"promptId":"p1"', + ); +} diff --git a/packages/cli/src/serve/server/session-archive.ts b/packages/cli/src/serve/server/session-archive.ts index ecc70687d4d..4b49fd3a205 100644 --- a/packages/cli/src/serve/server/session-archive.ts +++ b/packages/cli/src/serve/server/session-archive.ts @@ -863,31 +863,6 @@ export async function archiveDaemonSessions(params: { if (initialLocation === undefined) { return { kind: 'notFound' as const, mutationApplied: false }; } - if (initialLocation === 'archived') { - let maintenanceError: unknown; - try { - await updateScheduledTaskForMaintenance( - service, - sessionId, - 'archive', - assertCanMutate, - ); - } catch (error) { - maintenanceError = error; - logSessionArchiveWarning( - `scheduled task lifecycle update failed action=archive workspace=${safeLogValue( - service.getProjectRoot(), - )} session=${safeLogValue(sessionId)} error=${safeLogValue( - errorMessage(error), - )}`, - ); - } - return { - kind: 'alreadyArchived' as const, - mutationApplied: false, - maintenanceError, - }; - } if (initialLocation === 'conflict' && !resolveConflicts) { return { kind: 'error' as const, @@ -911,12 +886,6 @@ export async function archiveDaemonSessions(params: { mutationApplied: false, }; } - if (lockedLocation === 'archived') { - return { - value: 'alreadyArchived' as const, - mutationApplied: false, - }; - } if (lockedLocation === 'conflict' && !resolveConflicts) { throw sessionLocationError(sessionId); } @@ -964,10 +933,30 @@ export async function archiveDaemonSessions(params: { maintenanceError: mutation.maintenanceError, }; } + let maintenanceError = mutation.maintenanceError; + if (mutation.value === 'alreadyArchived') { + try { + await updateScheduledTaskForMaintenance( + service, + sessionId, + 'archive', + assertCanMutate, + ); + } catch (error) { + maintenanceError = error; + logSessionArchiveWarning( + `scheduled task lifecycle update failed action=archive workspace=${safeLogValue( + service.getProjectRoot(), + )} session=${safeLogValue(sessionId)} error=${safeLogValue( + errorMessage(error), + )}`, + ); + } + } return { kind: mutation.value ?? 'notFound', mutationApplied: mutation.mutationApplied, - maintenanceError: mutation.maintenanceError, + maintenanceError, }; }; return await (coordinatorLockHeld @@ -1061,31 +1050,6 @@ export async function unarchiveDaemonSessions(params: { if (initialLocation === undefined) { return { kind: 'notFound' as const, mutationApplied: false }; } - if (initialLocation === 'active') { - let maintenanceError: unknown; - try { - await updateScheduledTaskForMaintenance( - service, - sessionId, - 'unarchive', - assertCanMutate, - ); - } catch (error) { - maintenanceError = error; - logSessionArchiveWarning( - `scheduled task lifecycle update failed action=unarchive workspace=${safeLogValue( - service.getProjectRoot(), - )} session=${safeLogValue(sessionId)} error=${safeLogValue( - errorMessage(error), - )}`, - ); - } - return { - kind: 'alreadyActive' as const, - mutationApplied: false, - maintenanceError, - }; - } if (initialLocation === 'conflict' && !resolveConflicts) { return { kind: 'error' as const, @@ -1109,12 +1073,6 @@ export async function unarchiveDaemonSessions(params: { mutationApplied: false, }; } - if (lockedLocation === 'active') { - return { - value: 'alreadyActive' as const, - mutationApplied: false, - }; - } if (lockedLocation === 'conflict' && !resolveConflicts) { throw sessionLocationError(sessionId); } @@ -1161,10 +1119,30 @@ export async function unarchiveDaemonSessions(params: { maintenanceError: mutation.maintenanceError, }; } + let maintenanceError = mutation.maintenanceError; + if (mutation.value === 'alreadyActive') { + try { + await updateScheduledTaskForMaintenance( + service, + sessionId, + 'unarchive', + assertCanMutate, + ); + } catch (error) { + maintenanceError = error; + logSessionArchiveWarning( + `scheduled task lifecycle update failed action=unarchive workspace=${safeLogValue( + service.getProjectRoot(), + )} session=${safeLogValue(sessionId)} error=${safeLogValue( + errorMessage(error), + )}`, + ); + } + } return { kind: mutation.value ?? 'notFound', mutationApplied: mutation.mutationApplied, - maintenanceError: mutation.maintenanceError, + maintenanceError, }; }; return await (coordinatorLockHeld diff --git a/packages/core/src/services/session-writer-lease.test.ts b/packages/core/src/services/session-writer-lease.test.ts index 5c5a531c352..3517b024fe7 100644 --- a/packages/core/src/services/session-writer-lease.test.ts +++ b/packages/core/src/services/session-writer-lease.test.ts @@ -117,6 +117,11 @@ const descriptorReadHook = vi.hoisted(() => ({ afterRead: undefined as (() => void) | undefined, })); +const lockIdentityPrecisionFault = vi.hoisted(() => ({ + path: undefined as string | undefined, + replaced: false, +})); + vi.mock('node:fs', async (importOriginal) => { const actual = await importOriginal(); const readFileSyncWithHook = ((...args: unknown[]) => { @@ -130,14 +135,63 @@ vi.mock('node:fs', async (importOriginal) => { } return result; }) as typeof actual.readFileSync; - return { ...actual, readFileSync: readFileSyncWithHook }; + const applyLockIdentityFault = ( + result: unknown, + bigint: boolean, + replaced: boolean, + ): unknown => { + if (typeof result !== 'object' || result === null) return result; + const base = 9_007_199_254_740_992n; + Object.defineProperty(result, 'dev', { + value: bigint ? 1n : 1, + }); + Object.defineProperty(result, 'ino', { + value: bigint + ? base + (replaced ? 1n : 0n) + : Number(base + (replaced ? 1n : 0n)), + }); + return result; + }; + const fstatSyncWithHook = ((...args: unknown[]) => { + const result = (actual.fstatSync as (...callArgs: unknown[]) => unknown)( + ...args, + ); + if (lockIdentityPrecisionFault.path === undefined) return result; + const bigint = + typeof args[1] === 'object' && + args[1] !== null && + (args[1] as { bigint?: boolean }).bigint === true; + return applyLockIdentityFault(result, bigint, false); + }) as typeof actual.fstatSync; + const lstatSyncWithHook = ((...args: unknown[]) => { + const result = (actual.lstatSync as (...callArgs: unknown[]) => unknown)( + ...args, + ); + if (args[0] !== lockIdentityPrecisionFault.path) return result; + const bigint = + typeof args[1] === 'object' && + args[1] !== null && + (args[1] as { bigint?: boolean }).bigint === true; + return applyLockIdentityFault( + result, + bigint, + lockIdentityPrecisionFault.replaced, + ); + }) as typeof actual.lstatSync; + return { + ...actual, + fstatSync: fstatSyncWithHook, + lstatSync: lstatSyncWithHook, + readFileSync: readFileSyncWithHook, + }; }); vi.mock('node:fs/promises', async (importOriginal) => { const actual = await importOriginal(); return { ...actual, - lstat: async (filePath: Parameters[0]) => { + lstat: async (...args: unknown[]) => { + const filePath = args[0] as Parameters[0]; if (filePath === lstatFault.path) { lstatFault.calls++; if (lstatFault.remainingFailures > 0) { @@ -147,7 +201,25 @@ vi.mock('node:fs/promises', async (importOriginal) => { }); } } - return actual.lstat(filePath); + const result = await ( + actual.lstat as (...callArgs: unknown[]) => Promise + )(...args); + if (filePath !== lockIdentityPrecisionFault.path) return result; + const bigint = + typeof args[1] === 'object' && + args[1] !== null && + (args[1] as { bigint?: boolean }).bigint === true; + if (typeof result !== 'object' || result === null) return result; + const base = 9_007_199_254_740_992n; + Object.defineProperty(result, 'dev', { + value: bigint ? 1n : 1, + }); + Object.defineProperty(result, 'ino', { + value: bigint + ? base + (lockIdentityPrecisionFault.replaced ? 1n : 0n) + : Number(base + (lockIdentityPrecisionFault.replaced ? 1n : 0n)), + }); + return result; }, stat: async ( filePath: Parameters[0], @@ -481,6 +553,8 @@ afterEach(async () => { readFileFault.calls = 0; readFileFault.afterRead = undefined; descriptorReadHook.afterRead = undefined; + lockIdentityPrecisionFault.path = undefined; + lockIdentityPrecisionFault.replaced = false; setDebugLogSession(null); resetDebugLoggingState(); Storage.setRuntimeBaseDir(null); @@ -1035,6 +1109,46 @@ describe('SessionWriterLease', () => { }, ); + it.runIf(process.platform !== 'win32')( + 'rejects a byte-identical replacement during an asynchronous ownership read', + async () => { + const fixture = await createFixture(); + const lease = await SessionWriterLease.acquire(fixture.options); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + const replacementPath = `${lockPath}.replacement`; + await fs.writeFile(replacementPath, await fs.readFile(lockPath, 'utf8')); + readFileFault.path = lockPath; + readFileFault.triggerCall = 1; + readFileFault.afterRead = () => fs.rename(replacementPath, lockPath); + + await expect(lease.assertOwnedAndUnchanged()).rejects.toBeInstanceOf( + SessionWriterLostError, + ); + await expect(lease.release()).rejects.toBeInstanceOf( + SessionWriterLostError, + ); + }, + ); + + it('compares lock identities without losing large inode precision', async () => { + const fixture = await createFixture(); + const lockPath = getSessionWriterLockPath( + fixture.runtimeBaseDir, + fixture.options.sessionId, + ); + lockIdentityPrecisionFault.path = lockPath; + const lease = await SessionWriterLease.acquire(fixture.options); + lockIdentityPrecisionFault.replaced = true; + + expect(() => lease.assertCleanupOwned()).toThrow(SessionWriterLostError); + await expect(lease.release()).rejects.toBeInstanceOf( + SessionWriterLostError, + ); + }); + it.runIf(process.platform !== 'win32')( 'rejects a lock replaced while cleanup ownership is being verified', async () => { @@ -2353,7 +2467,7 @@ describe('SessionWriterLease', () => { lstatFault.remainingFailures = 1; await expect(lease.release()).resolves.toBeUndefined(); - expect(lstatFault.calls).toBe(2); + expect(lstatFault.calls).toBe(3); expect(lease.isReleased).toBe(true); lstatFault.path = undefined; await expect(fs.lstat(lockPath)).rejects.toMatchObject({ code: 'ENOENT' }); diff --git a/packages/core/src/services/session-writer-lease.ts b/packages/core/src/services/session-writer-lease.ts index bf3d4ce7050..e3ea59cb75a 100644 --- a/packages/core/src/services/session-writer-lease.ts +++ b/packages/core/src/services/session-writer-lease.ts @@ -8,7 +8,7 @@ import { execFile } from 'node:child_process'; import * as nodeConstants from 'node:constants'; import { createHash, randomUUID, type Hash } from 'node:crypto'; import * as nodeFs from 'node:fs'; -import type { Stats } from 'node:fs'; +import type { BigIntStats, Stats } from 'node:fs'; import * as fs from 'node:fs/promises'; import * as os from 'node:os'; import * as path from 'node:path'; @@ -1583,7 +1583,7 @@ export class SessionWriterLease { private terminalPromise: Promise | undefined; private operationTail: Promise = Promise.resolve(); private readonly lockRecordRaw: string; - private lockFileIdentity: { dev: number; ino: number } | undefined; + private lockFileIdentity: { dev: bigint; ino: bigint } | undefined; private readonly retiredPath: string; private readonly claimPath: string; @@ -2042,9 +2042,9 @@ export class SessionWriterLease { private async readOwnedLock(): Promise { if (this.released) throw new SessionWriterLostError(); - let stat: Awaited>; + let stat: BigIntStats; try { - stat = await fs.lstat(this.lockPath); + stat = await fs.lstat(this.lockPath, { bigint: true }); } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') { throw new SessionWriterLostError(); @@ -2078,10 +2078,27 @@ export class SessionWriterLease { ) { throw new SessionWriterLostError(); } + let current: BigIntStats; + try { + current = await fs.lstat(this.lockPath, { bigint: true }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + throw new SessionWriterLostError(); + } + throw new SessionWriterUnavailableError(); + } + if ( + !current.isFile() || + current.isSymbolicLink() || + current.dev !== stat.dev || + current.ino !== stat.ino + ) { + throw new SessionWriterLostError(); + } return record; } - private readVerifiedLockIdentity(): { dev: number; ino: number } { + private readVerifiedLockIdentity(): { dev: bigint; ino: bigint } { let descriptor: number; try { descriptor = nodeFs.openSync( @@ -2098,15 +2115,15 @@ export class SessionWriterLease { throw new SessionWriterUnavailableError(); } try { - const stat = nodeFs.fstatSync(descriptor); + const stat = nodeFs.fstatSync(descriptor, { bigint: true }); if (!stat.isFile()) throw new SessionWriterLostError(); if (!hasVerifiableInode(stat.ino)) { throw new SessionWriterUnavailableError(); } const assertPathMatchesDescriptor = (): void => { - let pathStat: Stats; + let pathStat: BigIntStats; try { - pathStat = nodeFs.lstatSync(this.lockPath); + pathStat = nodeFs.lstatSync(this.lockPath, { bigint: true }); } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') { throw new SessionWriterLostError(); diff --git a/packages/core/src/services/sessionService.corruption.test.ts b/packages/core/src/services/sessionService.corruption.test.ts index 3fb11161a78..e3d2178f28f 100644 --- a/packages/core/src/services/sessionService.corruption.test.ts +++ b/packages/core/src/services/sessionService.corruption.test.ts @@ -338,6 +338,14 @@ describe('SessionService lifecycle maintenance', () => { id: string, state: 'active' | 'archived', ) => string; + getPromptLedgerPathForState: ( + id: string, + state: 'active' | 'archived', + ) => string; + getWorktreeSessionPathForState: ( + id: string, + state: 'active' | 'archived', + ) => string; sessionBelongsToCurrentProject: ( sessionId: string, cwd: string, @@ -1063,6 +1071,80 @@ describe('SessionService lifecycle maintenance', () => { }, ); + it.each(['archive', 'unarchive'] as const)( + 'reconciles stranded %s sidecars on an exact retry', + async (action) => { + const sourceState = action === 'archive' ? 'active' : 'archived'; + const destinationState = action === 'archive' ? 'archived' : 'active'; + const { service, sessionId } = createHarness('transcript', sourceState); + const internals = service as unknown as Privates; + const sourceWorktree = internals.getWorktreeSessionPathForState( + sessionId, + sourceState, + ); + const destinationWorktree = internals.getWorktreeSessionPathForState( + sessionId, + destinationState, + ); + const sourcePr = internals.getPrSessionPathForState( + sessionId, + sourceState, + ); + const destinationPr = internals.getPrSessionPathForState( + sessionId, + destinationState, + ); + const sourceLedger = internals.getPromptLedgerPathForState( + sessionId, + sourceState, + ); + const destinationLedger = internals.getPromptLedgerPathForState( + sessionId, + destinationState, + ); + fs.writeFileSync(sourceWorktree, '{}'); + const pr = { + number: 123, + url: 'https://github.com/QwenLM/qwen-code/pull/123', + createdAt: '2026-08-28T00:00:00.000Z', + }; + await writeSessionPrs(sourcePr, [pr]); + fs.writeFileSync(sourceLedger, '{"promptId":"p1"}\n'); + const ownershipLost = new Error('writer ownership lost'); + + const first = await service[`${action}Sessions`]([sessionId], { + assertCleanupOwned: () => { + throw ownershipLost; + }, + }); + expect(first.errors[0]?.error).toBe(ownershipLost); + + const assertCanMutate = vi.fn(); + const assertCleanupOwned = vi.fn(); + const retry = await service[`${action}Sessions`]([sessionId], { + assertCanMutate, + assertCleanupOwned, + }); + + expect(retry).toMatchObject({ + [action === 'archive' ? 'alreadyArchived' : 'alreadyActive']: [ + sessionId, + ], + errors: [], + }); + expect(fs.existsSync(sourceWorktree)).toBe(false); + expect(fs.existsSync(destinationWorktree)).toBe(true); + expect(fs.existsSync(sourcePr)).toBe(false); + await expect(readSessionPrs(destinationPr)).resolves.toEqual([pr]); + expect(fs.existsSync(sourceLedger)).toBe(false); + expect(fs.readFileSync(destinationLedger, 'utf8')).toContain( + '"promptId":"p1"', + ); + expect(assertCanMutate).toHaveBeenCalled(); + expect(assertCleanupOwned).toHaveBeenCalled(); + }, + ); + it('rejects an in-place rewrite during ownership classification', async () => { const { service, sessionId, paths, cwd } = createHarness('', 'active'); fs.writeFileSync( diff --git a/packages/core/src/services/sessionService.ts b/packages/core/src/services/sessionService.ts index 34149976448..5e3eaae15d2 100644 --- a/packages/core/src/services/sessionService.ts +++ b/packages/core/src/services/sessionService.ts @@ -1661,6 +1661,62 @@ export class SessionService { fs.unlinkSync(sourcePath); } + private async moveArchiveSidecars( + sessionId: string, + action: 'archive' | 'unarchive', + assertCleanupOwned?: () => void, + ): Promise { + const sourceState = action === 'archive' ? 'active' : 'archived'; + const destinationState = action === 'archive' ? 'archived' : 'active'; + const sourceWorktree = this.getWorktreeSessionPathForState( + sessionId, + sourceState, + ); + const destinationWorktree = this.getWorktreeSessionPathForState( + sessionId, + destinationState, + ); + assertCleanupOwned?.(); + try { + this.moveOptionalFile(sourceWorktree, destinationWorktree); + } catch (error) { + this.warn( + `${action}Sessions: failed to move worktree sidecar for ${sessionId} from ${sourceWorktree} to ${destinationWorktree}: ${error}`, + ); + } + try { + await this.movePrSidecar( + this.getPrSessionPathForState(sessionId, sourceState), + this.getPrSessionPathForState(sessionId, destinationState), + assertCleanupOwned, + ); + } catch (error) { + if (error instanceof SessionWriterError) throw error; + assertCleanupOwned?.(); + this.warn( + `${action}Sessions: failed to move pr sidecar for ${sessionId}: ${error}`, + ); + } + const sourceLedger = this.getPromptLedgerPathForState( + sessionId, + sourceState, + ); + const destinationLedger = this.getPromptLedgerPathForState( + sessionId, + destinationState, + ); + try { + assertCleanupOwned?.(); + this.moveLedgerSidecar(sourceLedger, destinationLedger); + } catch (error) { + if (error instanceof SessionWriterError) throw error; + assertCleanupOwned?.(); + this.warn( + `${action}Sessions: failed to move prompt ledger for ${sessionId} from ${sourceLedger} to ${destinationLedger}: ${error}`, + ); + } + } + private sessionFileMoveError( action: 'archive' | 'unarchive', error: unknown, @@ -2803,6 +2859,12 @@ export class SessionService { continue; } if (location === 'archived') { + if (options.assertCleanupOwned) { + await this.moveArchiveSidecars(sessionId, 'archive', () => { + options.assertCanMutate?.(); + options.assertCleanupOwned?.(); + }); + } alreadyArchived.push(sessionId); continue; } @@ -2849,22 +2911,6 @@ export class SessionService { const sourcePath = this.getSessionFilePath(sessionId, 'active'); const targetPath = this.getSessionFilePath(sessionId, 'archived'); fs.mkdirSync(this.getArchiveChatsDir(), { recursive: true }); - const activeSidecar = this.getWorktreeSessionPathForState( - sessionId, - 'active', - ); - const archivedSidecar = this.getWorktreeSessionPathForState( - sessionId, - 'archived', - ); - const activeLedger = this.getPromptLedgerPathForState( - sessionId, - 'active', - ); - const archivedLedger = this.getPromptLedgerPathForState( - sessionId, - 'archived', - ); await options.assertStorageUnchanged?.(); options.assertCanMutate?.(); this.assertMaintainableSessionUnchanged(sessionId, snapshot); @@ -2873,41 +2919,11 @@ export class SessionService { } catch (error) { throw this.sessionFileMoveError('archive', error); } - options.assertCleanupOwned?.(); - try { - this.moveOptionalFile(activeSidecar, archivedSidecar); - } catch (sidecarError) { - this.warn( - `archiveSessions: failed to move worktree sidecar for ${sessionId} from ${activeSidecar} to ${archivedSidecar}: ${sidecarError}`, - ); - } - try { - await this.movePrSidecar( - this.getPrSessionPathForState(sessionId, 'active'), - this.getPrSessionPathForState(sessionId, 'archived'), - options.assertCleanupOwned, - ); - } catch (sidecarError) { - if (sidecarError instanceof SessionWriterError) { - throw sidecarError; - } - options.assertCleanupOwned?.(); - this.warn( - `archiveSessions: failed to move pr sidecar for ${sessionId}: ${sidecarError}`, - ); - } - try { - options.assertCleanupOwned?.(); - this.moveLedgerSidecar(activeLedger, archivedLedger); - } catch (ledgerError) { - if (ledgerError instanceof SessionWriterError) { - throw ledgerError; - } - options.assertCleanupOwned?.(); - this.warn( - `archiveSessions: failed to move prompt ledger for ${sessionId} from ${activeLedger} to ${archivedLedger}: ${ledgerError}`, - ); - } + await this.moveArchiveSidecars( + sessionId, + 'archive', + options.assertCleanupOwned, + ); archived.push(sessionId); } catch (error) { if ( @@ -2953,6 +2969,12 @@ export class SessionService { continue; } if (location === 'active') { + if (options.assertCleanupOwned) { + await this.moveArchiveSidecars(sessionId, 'unarchive', () => { + options.assertCanMutate?.(); + options.assertCleanupOwned?.(); + }); + } alreadyActive.push(sessionId); continue; } @@ -2998,14 +3020,6 @@ export class SessionService { const sourcePath = this.getSessionFilePath(sessionId, 'archived'); const targetPath = this.getSessionFilePath(sessionId, 'active'); - const archivedSidecar = this.getWorktreeSessionPathForState( - sessionId, - 'archived', - ); - const activeSidecar = this.getWorktreeSessionPathForState( - sessionId, - 'active', - ); fs.mkdirSync(path.dirname(targetPath), { recursive: true }); await options.assertStorageUnchanged?.(); options.assertCanMutate?.(); @@ -3015,49 +3029,11 @@ export class SessionService { } catch (error) { throw this.sessionFileMoveError('unarchive', error); } - options.assertCleanupOwned?.(); - try { - this.moveOptionalFile(archivedSidecar, activeSidecar); - } catch (sidecarError) { - this.warn( - `unarchiveSessions: failed to move worktree sidecar for ${sessionId} from ${archivedSidecar} to ${activeSidecar}: ${sidecarError}`, - ); - } - try { - await this.movePrSidecar( - this.getPrSessionPathForState(sessionId, 'archived'), - this.getPrSessionPathForState(sessionId, 'active'), - options.assertCleanupOwned, - ); - } catch (sidecarError) { - if (sidecarError instanceof SessionWriterError) { - throw sidecarError; - } - options.assertCleanupOwned?.(); - this.warn( - `unarchiveSessions: failed to move pr sidecar for ${sessionId}: ${sidecarError}`, - ); - } - const archivedLedger = this.getPromptLedgerPathForState( + await this.moveArchiveSidecars( sessionId, - 'archived', + 'unarchive', + options.assertCleanupOwned, ); - const activeLedger = this.getPromptLedgerPathForState( - sessionId, - 'active', - ); - try { - options.assertCleanupOwned?.(); - this.moveLedgerSidecar(archivedLedger, activeLedger); - } catch (ledgerError) { - if (ledgerError instanceof SessionWriterError) { - throw ledgerError; - } - options.assertCleanupOwned?.(); - this.warn( - `unarchiveSessions: failed to move prompt ledger for ${sessionId} from ${archivedLedger} to ${activeLedger}: ${ledgerError}`, - ); - } unarchived.push(sessionId); } catch (error) { if ( From 4a42ea1082548857559932ebe1ee507a0594fe45 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Sat, 29 Aug 2026 03:58:17 +0800 Subject: [PATCH 07/12] codex: address PR review feedback (#10300) Co-authored-by: Qwen-Coder --- packages/cli/src/serve/server/session-archive.test.ts | 5 +++++ packages/core/src/services/sessionService.corruption.test.ts | 5 ++++- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/packages/cli/src/serve/server/session-archive.test.ts b/packages/cli/src/serve/server/session-archive.test.ts index 598d14644d3..abeb1c0fc08 100644 --- a/packages/cli/src/serve/server/session-archive.test.ts +++ b/packages/cli/src/serve/server/session-archive.test.ts @@ -612,6 +612,7 @@ describe('archiveDaemonSessions', () => { it('reconciles stranded sidecars before returning already archived', async () => { const sessionId = '550e8400-e29b-41d4-a716-446655440103'; writeSessionFile(workspaceDir, sessionId, 'archived'); + fs.writeFileSync(sessionPath(workspaceDir, sessionId, 'archived'), ''); const service = new SessionService(workspaceDir); const sidecars = await writeLifecycleSidecars(service, sessionId, 'active'); @@ -1133,6 +1134,10 @@ describe('unarchiveDaemonSessions', () => { it('reconciles stranded sidecars before returning already active', async () => { const sessionId = '550e8400-e29b-41d4-a716-446655440113'; writeSessionFile(workspaceDir, sessionId, 'active'); + fs.writeFileSync( + sessionPath(workspaceDir, sessionId, 'active'), + '{"uuid":"torn-head"', + ); const service = new SessionService(workspaceDir); const sidecars = await writeLifecycleSidecars( service, diff --git a/packages/core/src/services/sessionService.corruption.test.ts b/packages/core/src/services/sessionService.corruption.test.ts index e3d2178f28f..057b2c4c0aa 100644 --- a/packages/core/src/services/sessionService.corruption.test.ts +++ b/packages/core/src/services/sessionService.corruption.test.ts @@ -1076,7 +1076,10 @@ describe('SessionService lifecycle maintenance', () => { async (action) => { const sourceState = action === 'archive' ? 'active' : 'archived'; const destinationState = action === 'archive' ? 'archived' : 'active'; - const { service, sessionId } = createHarness('transcript', sourceState); + const { service, sessionId } = createHarness( + action === 'archive' ? '' : '{"uuid":"torn-head"', + sourceState, + ); const internals = service as unknown as Privates; const sourceWorktree = internals.getWorktreeSessionPathForState( sessionId, From 0f084fd43a9bce6878e551b78f8c13b077522c56 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Sat, 29 Aug 2026 08:19:13 +0800 Subject: [PATCH 08/12] codex: address PR review feedback (#10300) Co-authored-by: Qwen-Coder --- docs/developers/qwen-serve-protocol.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index e4c287fb6d2..365b84afd81 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -2500,7 +2500,7 @@ Response: `resolveConflicts` is optional and defaults to `false`. By default, active and archived files with the same id are reported in `errors`, and neither copy is moved, removed, or overwritten. Archiving a live session still performs the strict close described above before classifying the conflict, so that close may flush queued records to the active transcript. With `resolveConflicts: true`, archive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the archived copy, removes the active copy, and reports the id in both `archived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. `errors` entries have `{ "sessionId": "", "error": "message" }`. -The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `archived` and `resolvedConflicts`. Retry the same lifecycle request for the same workspace before treating the error as proof that the active copy or conflict remains. The retry reports the authoritative `alreadyArchived` state even for empty or damaged transcripts that session listing omits, and resumes pending sidecar cleanup under a fresh writer lease. +The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `archived` and `resolvedConflicts`. Retry the same lifecycle request for the same workspace before treating the error as proof that the active copy or conflict remains. When the service can verify the stored transcript identity and acquire the required writer or maintenance lease, the retry reports the authoritative `alreadyArchived` state even for empty or damaged transcripts that session listing omits, and resumes pending sidecar cleanup. Transcripts whose stored identity cannot be verified continue to report `errors` on retry and require manual inspection; retries that cannot acquire the required lease remain in `errors` until it becomes available. Lifecycle conflicts are batch item outcomes: the workspace-less and workspace-qualified routes return HTTP `200` with the conflict in `errors`. This replaces the earlier workspace-qualified HTTP `409 session_conflict` envelope; clients that called that route must inspect the batch response. Internal-runtime REST batches preserve the safe conflict message while continuing to redact other per-session failure details. @@ -2526,9 +2526,9 @@ Response: } ``` -`resolveConflicts` is optional and defaults to `false`. By default, simultaneous active and archived JSONL files produce a conflict in `errors`, and neither copy is moved, removed, or overwritten; an active-only session is returned in `alreadyActive`. With `resolveConflicts: true`, unarchive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the active copy, removes the archived copy, and reports the id in both `unarchived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. Archive or unarchive in flight for the same id returns `409 session_archiving` before starting the batch. +`resolveConflicts` is optional and defaults to `false`. By default, simultaneous active and archived JSONL files produce a conflict in `errors`, and neither copy is moved, removed, or overwritten; an active-only session is returned in `alreadyActive` after the daemon acquires a writer or maintenance lease to reconcile pending sidecar cleanup. If a live session still holds the writer lease, the active-only id is reported in `errors` until that session closes. With `resolveConflicts: true`, unarchive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the active copy, removes the archived copy, and reports the id in both `unarchived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. Archive or unarchive in flight for the same id returns `409 session_archiving` before starting the batch. -The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `unarchived` and `resolvedConflicts`. Retry the same lifecycle request for the same workspace before treating the error as proof that the archived copy or conflict remains. The retry reports the authoritative `alreadyActive` state even for empty or damaged transcripts that session listing omits, and resumes pending sidecar cleanup under a fresh writer lease. +The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `unarchived` and `resolvedConflicts`. Retry the same lifecycle request for the same workspace before treating the error as proof that the archived copy or conflict remains. When the service can verify the stored transcript identity and acquire the required writer or maintenance lease, the retry reports the authoritative `alreadyActive` state even for empty or damaged transcripts that session listing omits, and resumes pending sidecar cleanup. Transcripts whose stored identity cannot be verified continue to report `errors` on retry and require manual inspection; retries that cannot acquire the required lease, including because an active session still holds it, remain in `errors` until the lease becomes available. ACP-over-HTTP uses the same request and response bodies through vendor methods `_qwen/sessions/archive` and `_qwen/sessions/unarchive`. The REST route table maps `POST /sessions/archive` and `POST /sessions/unarchive` to those methods for ACP transports. From dc6898484b9dcbb564d3fc273204d9bc37e7914e Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Sat, 29 Aug 2026 18:32:46 +0800 Subject: [PATCH 09/12] codex: address PR review feedback (#10300) Co-authored-by: Qwen-Coder --- docs/developers/qwen-serve-protocol.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 365b84afd81..69659bd4580 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -2500,6 +2500,8 @@ Response: `resolveConflicts` is optional and defaults to `false`. By default, active and archived files with the same id are reported in `errors`, and neither copy is moved, removed, or overwritten. Archiving a live session still performs the strict close described above before classifying the conflict, so that close may flush queued records to the active transcript. With `resolveConflicts: true`, archive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the archived copy, removes the active copy, and reports the id in both `archived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. `errors` entries have `{ "sessionId": "", "error": "message" }`. +An archived-only session is returned in `alreadyArchived` after the daemon acquires a writer or maintenance lease to reconcile pending sidecar cleanup. If another process still holds the writer lease, the archived-only id is reported in `errors` until the lease becomes available. + The transcript move or conflict repair is not rolled back if a later cleanup ownership check fails. In that case the id may appear only in `errors`, even though the archive state already changed, and may be omitted from `archived` and `resolvedConflicts`. Retry the same lifecycle request for the same workspace before treating the error as proof that the active copy or conflict remains. When the service can verify the stored transcript identity and acquire the required writer or maintenance lease, the retry reports the authoritative `alreadyArchived` state even for empty or damaged transcripts that session listing omits, and resumes pending sidecar cleanup. Transcripts whose stored identity cannot be verified continue to report `errors` on retry and require manual inspection; retries that cannot acquire the required lease remain in `errors` until it becomes available. Lifecycle conflicts are batch item outcomes: the workspace-less and workspace-qualified routes return HTTP `200` with the conflict in `errors`. This replaces the earlier workspace-qualified HTTP `409 session_conflict` envelope; clients that called that route must inspect the batch response. Internal-runtime REST batches preserve the safe conflict message while continuing to redact other per-session failure details. From 4fa1b3079bd3d33be353fab7e6b7e0b12dd7ffe6 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Sat, 29 Aug 2026 22:30:05 +0800 Subject: [PATCH 10/12] codex: address PR review feedback (#10300) Co-authored-by: Qwen-Coder --- .../src/serve/server/session-archive.test.ts | 44 +++++++++++++++++++ packages/core/src/services/sessionService.ts | 27 +++++++++--- 2 files changed, 64 insertions(+), 7 deletions(-) diff --git a/packages/cli/src/serve/server/session-archive.test.ts b/packages/cli/src/serve/server/session-archive.test.ts index abeb1c0fc08..b84fa8aed8d 100644 --- a/packages/cli/src/serve/server/session-archive.test.ts +++ b/packages/cli/src/serve/server/session-archive.test.ts @@ -23,6 +23,10 @@ import { updateCronTasks, writeSessionPrs, } from '@qwen-code/qwen-code-core'; +import { + danglingInFlightPromptIds, + readPromptLedgerRecords, +} from '@qwen-code/acp-bridge/promptLedger'; import { SessionArchivedError, SessionArchivingError, @@ -1158,6 +1162,46 @@ describe('unarchiveDaemonSessions', () => { await expectLifecycleSidecarsMoved(sidecars, 'active'); }); + it('keeps archived ledger records before newer active records during reconciliation', async () => { + const sessionId = '550e8400-e29b-41d4-a716-446655440114'; + writeSessionFile(workspaceDir, sessionId, 'active'); + const service = new SessionService(workspaceDir); + const activeLedger = service.getPromptLedgerPath(sessionId); + const archivedPr = service.getPrSessionPathForArchiveState( + sessionId, + 'archived', + ); + const archivedLedger = path.join( + path.dirname(archivedPr), + `${sessionId}.ledger.jsonl`, + ); + fs.mkdirSync(path.dirname(activeLedger), { recursive: true }); + fs.mkdirSync(path.dirname(archivedLedger), { recursive: true }); + fs.writeFileSync( + archivedLedger, + '{"v":1,"promptId":"p1","state":"in_flight","at":1}\n', + ); + fs.writeFileSync( + activeLedger, + '{"v":1,"promptId":"p1","terminal":"completed","at":2}\n', + ); + + const result = await unarchiveDaemonSessions({ + sessionIds: [sessionId], + service, + coordinator: new SessionArchiveCoordinator(), + }); + + expect(result).toMatchObject({ + alreadyActive: [sessionId], + errors: [], + }); + const records = readPromptLedgerRecords(activeLedger); + expect(records.map((record) => record.at)).toEqual([1, 2]); + expect(danglingInFlightPromptIds(records)).toEqual([]); + expect(fs.existsSync(archivedLedger)).toBe(false); + }); + it('collapses case-variant spellings in one batch to a single unarchive', async () => { const sessionId = '550e8400-e29b-41d4-a716-446655440111'; writeSessionFile(workspaceDir, sessionId, 'archived'); diff --git a/packages/core/src/services/sessionService.ts b/packages/core/src/services/sessionService.ts index 5e3eaae15d2..a51cd7d7165 100644 --- a/packages/core/src/services/sessionService.ts +++ b/packages/core/src/services/sessionService.ts @@ -1592,12 +1592,16 @@ export class SessionService { /** * Move a prompt terminal ledger sidecar across archive states. Unlike a * bare rename, an existing destination does not wedge the pair forever: - * the ledger is append-only JSONL, so the source records are concatenated - * onto the destination (preserving write order) and the source is - * unlinked. Throws propagate to the caller, which owns the warn-only - * policy — a ledger problem must never block the transcript move. + * the ledger is append-only JSONL, so the older half is concatenated before + * the newer half (preserving write order) and the source is unlinked. Throws + * propagate to the caller, which owns the warn-only policy — a ledger + * problem must never block the transcript move. */ - private moveLedgerSidecar(sourcePath: string, destinationPath: string): void { + private moveLedgerSidecar( + sourcePath: string, + destinationPath: string, + action: 'archive' | 'unarchive', + ): void { if (!fs.existsSync(sourcePath)) { return; } @@ -1616,7 +1620,16 @@ export class SessionService { const payload = sourceContents.endsWith('\n') ? sourceContents : `${sourceContents}\n`; - fs.appendFileSync(destinationPath, `\n${payload}`, 'utf8'); + if (action === 'archive') { + fs.appendFileSync(destinationPath, `\n${payload}`, 'utf8'); + } else { + const destinationContents = fs.readFileSync(destinationPath, 'utf8'); + fs.writeFileSync( + destinationPath, + `${payload}${destinationContents.length > 0 ? '\n' : ''}${destinationContents}`, + 'utf8', + ); + } } fs.unlinkSync(sourcePath); } @@ -1707,7 +1720,7 @@ export class SessionService { ); try { assertCleanupOwned?.(); - this.moveLedgerSidecar(sourceLedger, destinationLedger); + this.moveLedgerSidecar(sourceLedger, destinationLedger, action); } catch (error) { if (error instanceof SessionWriterError) throw error; assertCleanupOwned?.(); From f2496e4e152e4694df99568ae040d9d2d06fdade Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Sun, 30 Aug 2026 02:23:48 +0800 Subject: [PATCH 11/12] codex: address PR review feedback (#10300) Co-authored-by: Qwen-Coder --- .../src/serve/server/session-archive.test.ts | 67 +++++++++++++++++++ packages/core/src/services/sessionService.ts | 14 +++- 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/packages/cli/src/serve/server/session-archive.test.ts b/packages/cli/src/serve/server/session-archive.test.ts index b84fa8aed8d..5d836c75a26 100644 --- a/packages/cli/src/serve/server/session-archive.test.ts +++ b/packages/cli/src/serve/server/session-archive.test.ts @@ -8,6 +8,7 @@ import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; import { execFileSync } from 'node:child_process'; +import { syncBuiltinESMExports } from 'node:module'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { SessionIdCaseConflictError, @@ -1202,6 +1203,72 @@ describe('unarchiveDaemonSessions', () => { expect(fs.existsSync(archivedLedger)).toBe(false); }); + it('preserves both ledger halves when reconciliation cannot commit the merge', async () => { + const sessionId = '550e8400-e29b-41d4-a716-446655440115'; + writeSessionFile(workspaceDir, sessionId, 'active'); + const warnings: string[] = []; + const service = new SessionService(workspaceDir, { + onWarning: (message) => warnings.push(message), + }); + const activeLedger = service.getPromptLedgerPath(sessionId); + const archivedPr = service.getPrSessionPathForArchiveState( + sessionId, + 'archived', + ); + const archivedLedger = path.join( + path.dirname(archivedPr), + `${sessionId}.ledger.jsonl`, + ); + const activeContents = + '{"v":1,"promptId":"p1","terminal":"completed","at":2}\n'; + fs.mkdirSync(path.dirname(activeLedger), { recursive: true }); + fs.mkdirSync(path.dirname(archivedLedger), { recursive: true }); + fs.writeFileSync( + archivedLedger, + '{"v":1,"promptId":"p1","state":"in_flight","at":1}\n', + ); + fs.writeFileSync(activeLedger, activeContents, { mode: 0o600 }); + + const writeFileSync = fs.writeFileSync.bind(fs); + const writeSpy = vi + .spyOn(fs, 'writeFileSync') + .mockImplementation((file, data, options) => { + const filePath = file.toString(); + if ( + filePath === activeLedger || + (filePath.startsWith(`${activeLedger}.`) && filePath.endsWith('.tmp')) + ) { + writeFileSync(file, String(data).slice(0, 32), options); + const error = new Error('ENOSPC: injected ledger write failure'); + (error as NodeJS.ErrnoException).code = 'ENOSPC'; + throw error; + } + return writeFileSync(file, data, options); + }); + syncBuiltinESMExports(); + + let result: Awaited>; + try { + result = await unarchiveDaemonSessions({ + sessionIds: [sessionId], + service, + coordinator: new SessionArchiveCoordinator(), + }); + } finally { + writeSpy.mockRestore(); + syncBuiltinESMExports(); + } + + expect(result).toMatchObject({ + alreadyActive: [sessionId], + errors: [], + }); + expect(fs.readFileSync(activeLedger, 'utf8')).toBe(activeContents); + expect(fs.existsSync(archivedLedger)).toBe(true); + expect(warnings).toHaveLength(1); + expect(warnings[0]).toContain('failed to move prompt ledger'); + }); + it('collapses case-variant spellings in one batch to a single unarchive', async () => { const sessionId = '550e8400-e29b-41d4-a716-446655440111'; writeSessionFile(workspaceDir, sessionId, 'archived'); diff --git a/packages/core/src/services/sessionService.ts b/packages/core/src/services/sessionService.ts index a51cd7d7165..2d031c65ada 100644 --- a/packages/core/src/services/sessionService.ts +++ b/packages/core/src/services/sessionService.ts @@ -11,6 +11,7 @@ import { type PreparedUsageBeforeTranscriptDeletion, } from './usageHistoryService.js'; import { getProjectHash } from '../utils/paths.js'; +import { atomicWriteFileSync } from '../utils/atomicFileWrite.js'; import path from 'node:path'; import fs from 'node:fs'; import { randomUUID } from 'node:crypto'; @@ -1601,6 +1602,7 @@ export class SessionService { sourcePath: string, destinationPath: string, action: 'archive' | 'unarchive', + assertCanCommit?: () => void, ): void { if (!fs.existsSync(sourcePath)) { return; @@ -1624,13 +1626,14 @@ export class SessionService { fs.appendFileSync(destinationPath, `\n${payload}`, 'utf8'); } else { const destinationContents = fs.readFileSync(destinationPath, 'utf8'); - fs.writeFileSync( + atomicWriteFileSync( destinationPath, `${payload}${destinationContents.length > 0 ? '\n' : ''}${destinationContents}`, - 'utf8', + { encoding: 'utf8', assertCanCommit }, ); } } + assertCanCommit?.(); fs.unlinkSync(sourcePath); } @@ -1720,7 +1723,12 @@ export class SessionService { ); try { assertCleanupOwned?.(); - this.moveLedgerSidecar(sourceLedger, destinationLedger, action); + this.moveLedgerSidecar( + sourceLedger, + destinationLedger, + action, + assertCleanupOwned, + ); } catch (error) { if (error instanceof SessionWriterError) throw error; assertCleanupOwned?.(); From ff01d1e075d02652380807f0a0e22cdb6c7011ee Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Sun, 30 Aug 2026 04:50:59 +0800 Subject: [PATCH 12/12] codex: address PR review feedback (#10300) Co-authored-by: Qwen-Coder --- .../core/src/services/sessionService.test.ts | 42 +++++++++++++++++++ packages/core/src/services/sessionService.ts | 1 + 2 files changed, 43 insertions(+) diff --git a/packages/core/src/services/sessionService.test.ts b/packages/core/src/services/sessionService.test.ts index 12f96bb22d4..4174d2369d7 100644 --- a/packages/core/src/services/sessionService.test.ts +++ b/packages/core/src/services/sessionService.test.ts @@ -2847,6 +2847,48 @@ describe('SessionService', () => { ); }); + it('does not append a prompt ledger after writer ownership is lost', async () => { + mockActiveSessionOnly(); + const sourceLedger = + '{"v":1,"promptId":"p1","state":"in_flight","at":1}\n'; + const originalDestination = + '{"v":1,"promptId":"p0","state":"committed","at":0}\n'; + let destinationLedger = originalDestination; + let sourceExists = true; + vi.spyOn(fs, 'readFileSync').mockReturnValue(sourceLedger); + vi.spyOn(fs, 'appendFileSync').mockImplementation( + (_filePath, contents) => { + destinationLedger += contents.toString(); + }, + ); + unlinkSyncSpy.mockImplementation((filePath) => { + if (filePath.toString().endsWith(`/chats/${sessionIdA}.ledger.jsonl`)) { + sourceExists = false; + } + }); + existsSyncSpy.mockImplementation((filePath) => + filePath.toString().endsWith(`${sessionIdA}.ledger.jsonl`), + ); + const ownershipLost = new SessionWriterLostError(); + const assertCleanupOwned = vi + .fn() + .mockImplementationOnce(() => undefined) + .mockImplementationOnce(() => undefined) + .mockImplementation(() => { + throw ownershipLost; + }); + + const result = await sessionService.archiveSessions([sessionIdA], { + assertCanMutate: vi.fn(), + assertCleanupOwned, + }); + + expect(result.errors[0]?.error).toBe(ownershipLost); + expect(destinationLedger).toBe(originalDestination); + expect(sourceExists).toBe(true); + expect(assertCleanupOwned).toHaveBeenCalledTimes(3); + }); + it('should not move worktree sidecar when archiving JSONL fails', async () => { mockActiveSessionOnly(); mockActiveWorktreeSidecarOnly(); diff --git a/packages/core/src/services/sessionService.ts b/packages/core/src/services/sessionService.ts index 2d031c65ada..65fa9b6f072 100644 --- a/packages/core/src/services/sessionService.ts +++ b/packages/core/src/services/sessionService.ts @@ -1623,6 +1623,7 @@ export class SessionService { ? sourceContents : `${sourceContents}\n`; if (action === 'archive') { + assertCanCommit?.(); fs.appendFileSync(destinationPath, `\n${payload}`, 'utf8'); } else { const destinationContents = fs.readFileSync(destinationPath, 'utf8');