diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 7b3cbfb09d9..37eeea3a668 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -2455,7 +2455,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. @@ -2481,7 +2481,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/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 684dffaf681..3e7d272eb3b 100644 --- a/packages/core/src/services/session-writer-lease.test.ts +++ b/packages/core/src/services/session-writer-lease.test.ts @@ -930,6 +930,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, ); @@ -940,6 +941,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, ); @@ -951,6 +953,30 @@ describe('SessionWriterLease', () => { ); }); + 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 0ab985ee227..0352693d7ed 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'; @@ -1966,6 +1967,62 @@ export class SessionWriterLease { return record; } + /** Verify ownership after the transcript snapshot intentionally changes. */ + assertCleanupOwned(): void { + if (this.released) throw new SessionWriterLostError(); + 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(); + 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(); + } + } catch (error) { + if (error instanceof SessionWriterLostError) throw error; + throw new SessionWriterUnavailableError(); + } finally { + nodeFs.closeSync(descriptor); + } + } + 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 f1fb3886c70..b546cab18ab 100644 --- a/packages/core/src/services/sessionService.test.ts +++ b/packages/core/src/services/sessionService.test.ts @@ -45,6 +45,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({ @@ -1817,6 +1818,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(5); + 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'), { @@ -2165,7 +2216,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(); @@ -2183,17 +2234,57 @@ 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 () => { @@ -2229,7 +2320,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'); @@ -2239,12 +2330,40 @@ describe('SessionService', () => { .mockImplementation(() => { throw generationChanged; }); + const assertCleanupOwned = vi.fn(); const result = await sessionService.archiveSessions([sessionIdA], { assertCanMutate, + assertCleanupOwned, + }); + + 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(generationChanged); + expect(result.errors[0]?.error).toBe(ownershipLost); expect(renameSyncSpy).toHaveBeenCalledWith( expect.stringContaining(`/chats/${sessionIdA}.jsonl`), expect.stringContaining(`/chats/archive/${sessionIdA}.jsonl`), @@ -2449,6 +2568,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', () => { @@ -2607,6 +2751,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(); @@ -2693,7 +2862,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'); @@ -2703,12 +2872,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 cd17d816cd8..56c2e7d92f6 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, @@ -290,6 +291,7 @@ export interface RemoveSessionsResult { export interface RemoveSessionOptions { assertStorageUnchanged?: () => Promise; assertCanMutate?: () => void; + assertCleanupOwned?: () => void; } export interface ArchiveSessionsResult { @@ -305,6 +307,7 @@ export interface ArchiveSessionsOptions { resolveConflicts?: boolean; assertStorageUnchanged?: () => Promise; assertCanMutate?: () => void; + assertCleanupOwned?: () => void; } export interface UnarchiveSessionsResult { @@ -320,6 +323,7 @@ export interface UnarchiveSessionsOptions { resolveConflicts?: boolean; assertStorageUnchanged?: () => Promise; assertCanMutate?: () => void; + assertCleanupOwned?: () => void; } export interface SessionServiceOptions { @@ -1390,7 +1394,7 @@ export class SessionService { private async removeSessionOrganization( sessionId: string, - assertCanMutate?: () => void, + assertCleanupOwned?: () => void, ): Promise { try { const service = new SessionOrganizationService( @@ -1399,15 +1403,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?.(); this.warn( `removeSession: failed to clear session organization for ${sessionId}: ${error}`, ); @@ -2376,7 +2381,10 @@ export class SessionService { ): Promise { const removed = await this.removeSessionFiles(sessionId, options); if (removed) { - await this.removeSessionOrganization(sessionId, options.assertCanMutate); + await this.removeSessionOrganization( + sessionId, + options.assertCleanupOwned, + ); } return removed; } @@ -2446,13 +2454,13 @@ export class SessionService { preparedUsage.get(identity.filePath) ?? null, ); } - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); this.removeWorktreeSidecars(sessionId); - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); this.removePrSidecars(sessionId); - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); this.removePromptLedgers(sessionId); - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); this.removeFileHistoryBackups(sessionId); return true; } catch (error) { @@ -2508,23 +2516,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'), ); @@ -2560,7 +2571,7 @@ export class SessionService { } catch (error) { throw this.sessionFileMoveError('archive', error); } - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); try { this.moveOptionalFile(activeSidecar, archivedSidecar); } catch (sidecarError) { @@ -2572,19 +2583,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}`, ); @@ -2649,23 +2666,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'), ); @@ -2693,7 +2713,7 @@ export class SessionService { } catch (error) { throw this.sessionFileMoveError('unarchive', error); } - options.assertCanMutate?.(); + options.assertCleanupOwned?.(); try { this.moveOptionalFile(archivedSidecar, activeSidecar); } catch (sidecarError) { @@ -2705,10 +2725,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}`, ); @@ -2722,10 +2745,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}`, );