fix(core): Preserve ownership during session cleanup

docs/developers/qwen-serve-protocol.md

@@ -2481,7 +2481,7 @@ Response:
 }
 ```
 
-`resolveConflicts` is optional and defaults to `false`. By default, active and archived files with the same id are reported in `errors`, and neither copy is moved, removed, or overwritten. Archiving a live session still performs the strict close described above before classifying the conflict, so that close may flush queued records to the active transcript. With `resolveConflicts: true`, archive keeps the archived copy, removes the active copy, and reports the id in both `archived` and `resolvedConflicts`. `errors` entries have `{ "sessionId": "<uuid>", "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": "<uuid>", "error": "message" }`.
 
 Lifecycle conflicts are batch item outcomes: the workspace-less and workspace-qualified routes return HTTP `200` with the conflict in `errors`. This replaces the earlier workspace-qualified HTTP `409 session_conflict` envelope; clients that called that route must inspect the batch response. Internal-runtime REST batches preserve the safe conflict message while continuing to redact other per-session failure details.
 
@@ -2507,7 +2507,7 @@ Response:
 }
 ```
 
-`resolveConflicts` is optional and defaults to `false`. By default, simultaneous active and archived JSONL files produce a conflict in `errors`, and neither copy is moved, removed, or overwritten; an active-only session is returned in `alreadyActive`. With `resolveConflicts: true`, unarchive keeps the active copy, removes the archived copy, and reports the id in both `unarchived` and `resolvedConflicts`. Archive or unarchive in flight for the same id returns `409 session_archiving` before starting the batch.
+`resolveConflicts` is optional and defaults to `false`. By default, simultaneous active and archived JSONL files produce a conflict in `errors`, and neither copy is moved, removed, or overwritten; an active-only session is returned in `alreadyActive`. With `resolveConflicts: true`, unarchive repairs the conflict only when both copies are regular transcript files that the selected workspace may maintain, including owned empty or damaged transcripts. It keeps the active copy, removes the archived copy, and reports the id in both `unarchived` and `resolvedConflicts`. The option does not bypass ownership checks; mixed local/foreign or otherwise ambiguous ownership is reported in `errors`, and neither copy is moved. Archive or unarchive in flight for the same id returns `409 session_archiving` before starting the batch.
 
 ACP-over-HTTP uses the same request and response bodies through vendor methods `_qwen/sessions/archive` and `_qwen/sessions/unarchive`. The REST route table maps `POST /sessions/archive` and `POST /sessions/unarchive` to those methods for ACP transports.
 

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';
@@ -330,12 +331,14 @@ function mockArchivedStandalone(storageSessionId = sessionId): void {
 
 function mockWriterLease(): {
   assertOwnedAndUnchanged: ReturnType<typeof vi.fn>;
+  assertCleanupOwned: ReturnType<typeof vi.fn>;
   release: ReturnType<typeof vi.fn>;
   isReleased: boolean;
   isReleaseDurabilityPending: boolean;
 } {
   const lease = {
     assertOwnedAndUnchanged: vi.fn(async () => undefined),
+    assertCleanupOwned: vi.fn(),
     release: vi.fn(async () => undefined),
     isReleased: false,
     isReleaseDurabilityPending: false,
@@ -628,33 +631,46 @@ 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],
       alreadyArchived: [],
       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 +680,15 @@ describe('StandaloneSessionService', () => {
       notFound: [],
       errors: [],
     });
+    const unarchiveOptions = unarchive.mock.calls[0]?.[1];
+    expect(unarchiveOptions).toEqual(
+      expect.objectContaining({
+        assertCanMutate: expect.any(Function),
+        assertCleanupOwned: expect.any(Function),
+      }),
+    );
+    unarchiveOptions?.assertCleanupOwned?.();
+    expect(unarchiveLease.assertCleanupOwned).toHaveBeenCalledOnce();
     expect(unarchiveLease.release).toHaveBeenCalledOnce();
   });
 
@@ -694,10 +719,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 +745,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();
   });
 
@@ -845,6 +876,61 @@ describe('StandaloneSessionService', () => {
     expect(harness.deletionJournal.clear).toHaveBeenCalledOnce();
   });
 
+  it('does not park a lost lease before exact deletion recovery', async () => {
+    mockActiveStandalone();
+    const harness = createHarness();
+    const lostLease = mockWriterLease();
+    lostLease.release.mockRejectedValue(new SessionWriterLostError());
+    const nextLease = {
+      assertOwnedAndUnchanged: vi.fn(async () => undefined),
+      assertCleanupOwned: vi.fn(),
+      release: vi.fn(async () => undefined),
+      isReleased: false,
+      isReleaseDurabilityPending: false,
+    };
+    vi.mocked(SessionService.prototype.acquireSessionWriterLease)
+      .mockResolvedValueOnce(lostLease as never)
+      .mockResolvedValueOnce(nextLease as never);
+    vi.spyOn(
+      SessionService.prototype,
+      'removeSessionTranscriptForLifecycle',
+    ).mockResolvedValue(true);
+    vi.spyOn(
+      SessionService.prototype,
+      'cleanupRemovedSessionStateForLifecycle',
+    ).mockResolvedValue();
+
+    await expect(harness.service.delete([sessionId])).resolves.toEqual({
+      removed: [sessionId],
+      notFound: [],
+      errors: [],
+      fileCleanupPending: [sessionId],
+    });
+
+    vi.mocked(
+      SessionService.prototype.findSessionIdIgnoringCase,
+    ).mockResolvedValue(undefined);
+    harness.deletionJournal.read.mockResolvedValueOnce(
+      deletionEntry() as never,
+    );
+    harness.inspectStandaloneDeletionPaths.mockResolvedValueOnce({
+      status: 'absent',
+    });
+
+    await expect(harness.service.delete([sessionId])).resolves.toEqual({
+      removed: [sessionId],
+      notFound: [],
+      errors: [],
+      fileCleanupPending: [],
+    });
+
+    expect(
+      SessionService.prototype.acquireSessionWriterLease,
+    ).toHaveBeenCalledTimes(2);
+    expect(nextLease.release).toHaveBeenCalledOnce();
+    expect(harness.deletionJournal.clear).toHaveBeenCalledOnce();
+  });
+
   it('rolls back a staged directory when recovery finds the transcript intact', async () => {
     mockActiveStandalone();
     const harness = createHarness();
@@ -1138,12 +1224,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 +1254,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 () => {

packages/cli/src/serve/conversations/standalone-session-service.ts

@@ -27,6 +27,7 @@ import {
   SessionStorageEntryError,
   SessionTranscriptDurabilityError,
   SessionTranscriptChangedError,
+  SessionWriterLostError,
   SessionWriterUnavailableError,
   type ApprovalMode,
   type SessionArchiveState,
@@ -1253,6 +1254,10 @@ export class StandaloneSessionService {
           record.storageSessionId,
           {
             assertCanMutate: () => this.options.assertRuntimeCurrent(runtime),
+            assertCleanupOwned: () => {
+              this.options.assertRuntimeCurrent(runtime);
+              lease.assertCleanupOwned();
+            },
           },
         );
       } catch {
@@ -1375,7 +1380,13 @@ export class StandaloneSessionService {
           this.pendingLifecycleLeaseReleases.delete(pendingKey);
         }
         return true;
-      } catch {
+      } catch (error) {
+        if (error instanceof SessionWriterLostError) {
+          if (this.pendingLifecycleLeaseReleases.get(pendingKey) === lease) {
+            this.pendingLifecycleLeaseReleases.delete(pendingKey);
+          }
+          return false;
+        }
         // The lease clears retryable terminal failures itself.
       }
     }
@@ -1555,6 +1566,10 @@ export class StandaloneSessionService {
                           lease.assertOwnedAndUnchanged(),
                         assertCanMutate: () =>
                           this.options.assertRuntimeCurrent(runtime),
+                        assertCleanupOwned: () => {
+                          this.options.assertRuntimeCurrent(runtime);
+                          lease.assertCleanupOwned();
+                        },
                       },
                     );
                     if (archived.errors[0]) throw archived.errors[0].error;
@@ -1655,6 +1670,10 @@ export class StandaloneSessionService {
                           lease.assertOwnedAndUnchanged(),
                         assertCanMutate: () =>
                           this.options.assertRuntimeCurrent(runtime),
+                        assertCleanupOwned: () => {
+                          this.options.assertRuntimeCurrent(runtime);
+                          lease.assertCleanupOwned();
+                        },
                       },
                     );
                     if (unarchived.errors[0]) throw unarchived.errors[0].error;
@@ -1959,6 +1978,10 @@ export class StandaloneSessionService {
             locked.storageSessionId,
             {
               assertCanMutate: () => this.options.assertRuntimeCurrent(runtime),
+              assertCleanupOwned: () => {
+                this.options.assertRuntimeCurrent(runtime);
+                lease.assertCleanupOwned();
+              },
             },
           );
         } catch {

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);
 

packages/cli/src/serve/server/session-archive.ts

@@ -225,6 +225,7 @@ async function runWithDaemonWriterLease<T>(params: {
   service: SessionService;
   mutate: (
     assertOwnedAndUnchanged: () => Promise<void>,
+    assertCleanupOwned: () => void,
   ) => Promise<{ value: T; mutationApplied: boolean }>;
   mutationAppliedAfterError: () => Promise<boolean>;
   afterMutationApplied: () => Promise<void>;
@@ -266,7 +267,10 @@ async function runWithDaemonWriterLease<T>(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) {