diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 9bc270e85f2..69659bd4580 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -2498,7 +2498,11 @@ 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" }`. + +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. @@ -2524,7 +2528,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 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` 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. 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. 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..02c3ec76c45 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'; @@ -26,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, @@ -330,12 +332,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 +632,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 +648,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 +681,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(); }); @@ -686,6 +712,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(); @@ -694,10 +786,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 +812,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(); }); @@ -767,6 +865,61 @@ 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('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(); @@ -845,6 +998,124 @@ 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('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(); @@ -1107,6 +1378,69 @@ 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('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(); @@ -1138,12 +1472,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 +1502,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..736389d5812 100644 --- a/packages/cli/src/serve/conversations/standalone-session-service.ts +++ b/packages/cli/src/serve/conversations/standalone-session-service.ts @@ -27,6 +27,8 @@ import { SessionStorageEntryError, SessionTranscriptDurabilityError, SessionTranscriptChangedError, + SessionWriterError, + SessionWriterLostError, SessionWriterUnavailableError, type ApprovalMode, type SessionArchiveState, @@ -62,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, @@ -1248,22 +1251,32 @@ export class StandaloneSessionService { cleanupPending = true; } } + let cleanupOwnershipLost = false; try { await service.cleanupRemovedSessionStateForLifecycle( record.storageSessionId, { assertCanMutate: () => this.options.assertRuntimeCurrent(runtime), + assertCleanupOwned: () => { + this.options.assertRuntimeCurrent(runtime); + lease.assertCleanupOwned(); + }, }, ); - } catch { + } catch (error) { cleanupPending = true; + cleanupOwnershipLost = + error instanceof SessionWriterError || + error instanceof ConversationRuntimeOwnershipError; } - 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; @@ -1333,7 +1346,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.', @@ -1375,7 +1389,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. } } @@ -1501,6 +1521,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 { @@ -1555,6 +1595,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; @@ -1586,6 +1630,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' @@ -1655,6 +1704,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; @@ -1686,6 +1739,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' @@ -1954,32 +2012,42 @@ export class StandaloneSessionService { } } + let cleanupOwnershipLost = false; try { await service.cleanupRemovedSessionStateForLifecycle( locked.storageSessionId, { assertCanMutate: () => this.options.assertRuntimeCurrent(runtime), + assertCleanupOwned: () => { + this.options.assertRuntimeCurrent(runtime); + lease.assertCleanupOwned(); + }, }, ); - } catch { - cleanupPending = true; - } - try { - await runtime.bridge.deleteSessionAttachments(sessionId, { - assertCanCommit: () => this.options.assertRuntimeCurrent(runtime), - }); - } catch { + } catch (error) { cleanupPending = true; + cleanupOwnershipLost = + error instanceof SessionWriterError || + error instanceof ConversationRuntimeOwnershipError; } - 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)) diff --git a/packages/cli/src/serve/server/session-archive.test.ts b/packages/cli/src/serve/server/session-archive.test.ts index 343e8d26cef..abeb1c0fc08 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,31 @@ 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'); + fs.writeFileSync(sessionPath(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'); @@ -926,6 +949,7 @@ describe('archiveDaemonSessions', () => { }); vi.spyOn(service, 'acquireSessionWriterLease').mockResolvedValue({ assertOwnedAndUnchanged: vi.fn().mockResolvedValue(undefined), + assertCleanupOwned: vi.fn(), release, } as unknown as SessionWriterLease); @@ -1078,7 +1102,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'; @@ -1098,7 +1122,7 @@ describe('unarchiveDaemonSessions', () => { notFound: [missingId], errors: [], }); - expect(acquire).toHaveBeenCalledTimes(1); + expect(acquire).toHaveBeenCalledTimes(2); expect(fs.existsSync(sessionPath(workspaceDir, archivedId, 'active'))).toBe( true, ); @@ -1107,6 +1131,33 @@ describe('unarchiveDaemonSessions', () => { ).toBe(false); }); + 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, + 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'); @@ -1849,3 +1900,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 1040d1d443e..4b49fd3a205 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), @@ -858,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, @@ -895,7 +875,7 @@ export async function archiveDaemonSessions(params: { action: 'archive', sessionId, service, - mutate: async (assertOwnedAndUnchanged) => { + mutate: async (assertOwnedAndUnchanged, assertCleanupOwned) => { const lockedLocation = await classifySessionLocation( service, sessionId, @@ -906,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); } @@ -919,6 +893,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) { @@ -958,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 @@ -1055,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, @@ -1092,7 +1062,7 @@ export async function unarchiveDaemonSessions(params: { action: 'unarchive', sessionId, service, - mutate: async (assertOwnedAndUnchanged) => { + mutate: async (assertOwnedAndUnchanged, assertCleanupOwned) => { const lockedLocation = await classifySessionLocation( service, sessionId, @@ -1103,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); } @@ -1116,6 +1080,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) { @@ -1154,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/config/config.test.ts b/packages/core/src/config/config.test.ts index c7c8ed41147..9eef25d8842 100644 --- a/packages/core/src/config/config.test.ts +++ b/packages/core/src/config/config.test.ts @@ -3737,6 +3737,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(); diff --git a/packages/core/src/services/session-writer-lease.test.ts b/packages/core/src/services/session-writer-lease.test.ts index ef4ed662b3f..3517b024fe7 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, @@ -112,11 +113,85 @@ const readFileFault = vi.hoisted(() => ({ afterRead: undefined as (() => Promise | void) | undefined, })); +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[]) => { + 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; + 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) { @@ -126,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], @@ -459,6 +552,9 @@ afterEach(async () => { readFileFault.triggerCall = 0; readFileFault.calls = 0; readFileFault.afterRead = undefined; + descriptorReadHook.afterRead = undefined; + lockIdentityPrecisionFault.path = undefined; + lockIdentityPrecisionFault.replaced = false; setDebugLogSession(null); resetDebugLoggingState(); Storage.setRuntimeBaseDir(null); @@ -969,6 +1065,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 +1076,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 +1088,136 @@ 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 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 () => { + 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 () => { + 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 () => { @@ -2239,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 dae0a72d920..e3ea59cb75a 100644 --- a/packages/core/src/services/session-writer-lease.ts +++ b/packages/core/src/services/session-writer-lease.ts @@ -7,7 +7,8 @@ import { execFile } from 'node:child_process'; import * as nodeConstants from 'node:constants'; import { createHash, randomUUID, type Hash } from 'node:crypto'; -import type { Stats } from 'node:fs'; +import * as nodeFs 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'; @@ -1582,6 +1583,7 @@ export class SessionWriterLease { private terminalPromise: Promise | undefined; private operationTail: Promise = Promise.resolve(); private readonly lockRecordRaw: string; + private lockFileIdentity: { dev: bigint; ino: bigint } | 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, @@ -2039,16 +2042,22 @@ 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(); } 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; @@ -2069,9 +2078,99 @@ 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: bigint; ino: bigint } { + 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, { bigint: true }); + if (!stat.isFile()) throw new SessionWriterLostError(); + if (!hasVerifiableInode(stat.ino)) { + throw new SessionWriterUnavailableError(); + } + const assertPathMatchesDescriptor = (): void => { + let pathStat: BigIntStats; + try { + pathStat = nodeFs.lstatSync(this.lockPath, { bigint: true }); + } 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(); + } + }; + assertPathMatchesDescriptor(); + 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(); + } + assertPathMatchesDescriptor(); + 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..057b2c4c0aa 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, @@ -997,7 +1005,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,17 +1025,129 @@ 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); }, ); + 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( + action === 'archive' ? '' : '{"uuid":"torn-head"', + 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.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..5e3eaae15d2 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}`, @@ -1656,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, @@ -2600,7 +2661,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 +2803,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 +2817,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); } @@ -2794,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; } @@ -2809,23 +2880,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'), ); @@ -2837,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); @@ -2861,35 +2919,11 @@ export class SessionService { } catch (error) { throw this.sessionFileMoveError('archive', error); } - options.assertCanMutate?.(); - 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.assertCanMutate, - ); - } catch (sidecarError) { - options.assertCanMutate?.(); - this.warn( - `archiveSessions: failed to move pr sidecar for ${sessionId}: ${sidecarError}`, - ); - } - try { - options.assertCanMutate?.(); - this.moveLedgerSidecar(activeLedger, archivedLedger); - } catch (ledgerError) { - options.assertCanMutate?.(); - 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 ( @@ -2935,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; } @@ -2950,23 +2990,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'), ); @@ -2977,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?.(); @@ -2994,43 +3029,11 @@ export class SessionService { } catch (error) { throw this.sessionFileMoveError('unarchive', error); } - options.assertCanMutate?.(); - 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.assertCanMutate, - ); - } catch (sidecarError) { - options.assertCanMutate?.(); - this.warn( - `unarchiveSessions: failed to move pr sidecar for ${sessionId}: ${sidecarError}`, - ); - } - const archivedLedger = this.getPromptLedgerPathForState( - sessionId, - 'archived', - ); - const activeLedger = this.getPromptLedgerPathForState( + await this.moveArchiveSidecars( sessionId, - 'active', + 'unarchive', + options.assertCleanupOwned, ); - try { - options.assertCanMutate?.(); - this.moveLedgerSidecar(archivedLedger, activeLedger); - } catch (ledgerError) { - options.assertCanMutate?.(); - this.warn( - `unarchiveSessions: failed to move prompt ledger for ${sessionId} from ${archivedLedger} to ${activeLedger}: ${ledgerError}`, - ); - } unarchived.push(sessionId); } catch (error) { if (