fix(core): Preserve ownership during session cleanup

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": "<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.
 
@@ -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.
 

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

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 () => {

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<void> {
     return this.runExclusive(() => this.assertOwnedAndUnchangedOnce());
   }

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