fix(sftp): preserve recovery ownership through deferred writes

application/app/dedicatedResumeProgress.ts

@@ -1,10 +1,12 @@
+import type { TransferOwnerObservation } from "../state/sftp/waitForTransferOwner";
 import type { TransferStatus, TransferTask } from "../../domain/models";
 
 export const DEDICATED_RESUME_LARGE_HISTORY_THRESHOLD = 4_096;
 export const DEDICATED_RESUME_CHILD_UPDATE_BATCH_SIZE = 512;
 
 export interface DedicatedResumeChildUpdateBatcher {
-  push(task: TransferTask): void;
+  push(task: TransferTask, observation?: TransferOwnerObservation): void;
+  discard(taskId: string): void;
   flush(): void;
 }
 
@@ -24,25 +26,39 @@ export function createDedicatedResumeChildUpdateBatcher(deps: {
   hasTask: (taskId: string) => boolean;
   upsertTasks: (tasks: readonly TransferTask[]) => void;
 }): DedicatedResumeChildUpdateBatcher {
-  const pending = new Map<string, TransferTask>();
+  const pending = new Map<string, { task: TransferTask; observation?: TransferOwnerObservation }>();
   const flush = () => {
     if (pending.size === 0) return;
     const batch = [...pending.values()];
     pending.clear();
-    deps.upsertTasks(batch);
+    try {
+      const updates = batch.filter(({ observation }) => !observation?.hasIdentityConflict()).map(({ task }) => task);
+      if (updates.length) deps.upsertTasks(updates);
+    } finally {
+      for (const { observation } of batch) observation?.dispose();
+    }
   };
   return {
-    push(task) {
+    push(task, observation) {
+      if (observation?.hasIdentityConflict()) {
+        pending.get(task.id)?.observation?.dispose();
+        pending.delete(task.id);
+        observation.dispose();
+        return;
+      }
       const shouldBatch = !!task.parentTaskId
         && deps.getTaskCount() >= DEDICATED_RESUME_LARGE_HISTORY_THRESHOLD
         && deps.hasTask(task.id);
       if (!shouldBatch) {
-        deps.upsertTasks([task]);
+        try { deps.upsertTasks([task]); } finally { observation?.dispose(); }
         return;
       }
-      pending.set(task.id, task);
+      const previous = pending.get(task.id);
+      if (previous?.observation && previous.observation !== observation) previous.observation.dispose();
+      pending.set(task.id, { task, observation });
       if (pending.size >= DEDICATED_RESUME_CHILD_UPDATE_BATCH_SIZE) flush();
     },
+    discard: (taskId) => { pending.get(taskId)?.observation?.dispose(); pending.delete(taskId); },
     flush,
   };
 }

application/app/useAppStartupEffects.ts

@@ -181,11 +181,14 @@ export function useAppStartupEffects(ctx: StartupEffectsContext) {
           },
           {
             children,
-            onChildUpdate: (child) => {
+            onChildUpdate: (child, observation) => {
               if (acceptsResumeCallbacks) {
-                childUpdateBatcher.push({ ...child, ownerId: "dedicated-resume" });
+                childUpdateBatcher.push({ ...child, ownerId: "dedicated-resume" }, observation);
+                return true;
               }
             },
+            onChildSuperseded: (taskId) => childUpdateBatcher.discard(taskId),
+            flushChildUpdates: () => childUpdateBatcher.flush(),
             onDirectoryCheckpointUpdate: (checkpoint) => {
               if (acceptsResumeCallbacks) {
                 sftpTransferCenterStore.patchTask(task.id, {

application/state/sftp/dedicatedTransferResume.test.ts

@@ -41,8 +41,12 @@ const host = (id: string, label: string, hostname = label): Host => ({
 
 for (const retainedStatus of [undefined, "interrupted", "failed", "paused"] as const) {
 for (const newerPause of retainedStatus === "paused" ? [false, true] : [false]) {
+for (const ownerChange of retainedStatus === "interrupted" || retainedStatus === undefined ? ["same", "active", "completed"] : ["same"]) {
+for (const completedBeforeReplacement of ownerChange === "active" ? [false, true] : [false]) {
+for (const flushBeforeReply of ownerChange === "same" || retainedStatus === undefined ? [false] : [false, true]) {
+for (const replyRejects of ownerChange !== "same" && !completedBeforeReplacement && !flushBeforeReply ? [false, true] : [false]) {
 const batchExistingIdentity = retainedStatus !== undefined;
-test(`superseded folder child settles when its completion was compacted into the parent: retained=${retainedStatus ?? "none"}, newerPause=${newerPause}`, async (t) => {
+test(`superseded folder child settles when its completion was compacted into the parent: retained=${retainedStatus ?? "none"}, newerPause=${newerPause}, owner=${ownerChange}, flush=${flushBeforeReply}, priorComplete=${completedBeforeReplacement}, reject=${replyRejects}`, async (t) => {
   const { sftpTransferCenterStore } = await import("../sftpTransferCenterStore");
   const previousLocalStorage = Object.getOwnPropertyDescriptor(globalThis, "localStorage");
   Object.defineProperty(globalThis, "localStorage", {
@@ -75,14 +79,19 @@ test(`superseded folder child settles when its completion was compacted into the
     ...parent, id: "retained-child", parentTaskId: parent.id, isDirectory: false,
     sourcePath: "/local/folder/a.bin", targetPath: "/remote/folder/a.bin",
     totalBytes: 10, sourceLastModified: 1, status: retainedStatus ?? "interrupted",
-    directoryEntryIndex: 0, directoryEntryIdentity: "a".repeat(64),
+    directoryEntryIndex: 0,
+    directoryEntryIdentity: newerPause ? createDirectoryEntryIdentity({
+      sourcePath: "/local/folder/a.bin", targetPath: "/remote/folder/a.bin", size: 10, lastModified: 1,
+    }) : "a".repeat(64),
   };
   const { createDedicatedResumeChildUpdateBatcher } = await import("../../app/dedicatedResumeProgress");
   // Exercise the real large-history batching branch with one relevant retained row.
   const childBatcher = createDedicatedResumeChildUpdateBatcher({
     getTaskCount: () => 4096,
-    hasTask: (id) => id === persisted.id,
-    upsertTasks: (updates) => sftpTransferCenterStore.upsertTasks(updates),
+    hasTask: (id) => id === persisted.id || id.startsWith("flush-peer-"),
+    // Peer rows only drive the real 512-entry flush; keep the relevant store
+    // fixture small so unrelated history maintenance cannot mask ownership.
+    upsertTasks: (updates) => sftpTransferCenterStore.upsertTasks(updates.filter((task) => !task.id.startsWith("flush-peer-"))),
   });
   sftpTransferCenterStore.publishOwner("compacted-owner", [parent, ...(batchExistingIdentity ? [persisted] : [])]);
   (netcattyBridge as { get: () => unknown }).get = () => ({
@@ -106,20 +115,50 @@ test(`superseded folder child settles when its completion was compacted into the
         directoryEntryIndex: options.directoryEntryIndex,
         directoryEntryIdentity: options.directoryEntryIdentity,
       });
-      // A newer same-id owner completes before this older invocation rejoins.
-      sftpTransferCenterStore.ingestBackgroundEvent({
-        type: "completed", transferId: childId, transferred: 10, totalBytes: 10, lifecycleEpoch: 0,
-      });
-      return { superseded: true };
+      if (ownerChange !== "same") {
+        const current = sftpTransferCenterStore.getTask(childId)!;
+        if (completedBeforeReplacement) {
+          sftpTransferCenterStore.ingestBackgroundEvent({
+            type: "completed", transferId: childId, transferred: 10, totalBytes: 10, lifecycleEpoch: 0,
+          });
+        }
+        sftpTransferCenterStore.upsertTasks([{
+          ...current, status: "transferring", directoryEntryIdentity: "e".repeat(64), lifecycleEpoch: 1,
+        }]);
+      }
+      if (ownerChange !== "active") {
+        // A newer same-id owner completes before this older invocation rejoins.
+        sftpTransferCenterStore.ingestBackgroundEvent({
+          type: "completed", transferId: childId, transferred: 10, totalBytes: 10, lifecycleEpoch: ownerChange === "same" ? 0 : 1,
+        });
+      }
+      if (flushBeforeReply) {
+        for (let index = 0; index < 512; index += 1) {
+          childBatcher.push({ ...persisted, id: `flush-peer-${index}`, status: "transferring" });
+        }
+        if (ownerChange === "active") {
+          assert.equal(sftpTransferCenterStore.getTask(childId)?.directoryEntryIdentity, "e".repeat(64),
+            "automatic flush before the displaced reply must preserve the new owner");
+        } else assert.equal(sftpTransferCenterStore.getTask(childId), undefined,
+          "automatic flush must not recreate the compacted new owner");
+      }
+      if (replyRejects) throw new Error("old invocation transport failed");
+      return completedBeforeReplacement ? {} : { superseded: true };
     },
   });
   running = resumeTransferWithDedicatedSession(parent, {
     hosts: [host("h1", "box", "1.2.3.4")], keys: [], identities: [],
   }, undefined, {
     children: batchExistingIdentity ? [persisted] : [], shouldAbort: () => abort,
-    onChildUpdate: (child) => batchExistingIdentity
-      ? childBatcher.push(child)
-      : sftpTransferCenterStore.publishOwner("compacted-owner", [parent, child]),
+    onChildSuperseded: (id) => childBatcher.discard(id),
+    flushChildUpdates: () => childBatcher.flush(),
+    onChildUpdate: (child, observation) => {
+      if (batchExistingIdentity) {
+        childBatcher.push(child, observation);
+        return true;
+      }
+      sftpTransferCenterStore.publishOwner("compacted-owner", [parent, child]);
+    },
   });
   const result = await Promise.race([
     running,
@@ -133,6 +172,15 @@ test(`superseded folder child settles when its completion was compacted into the
   }
   assert.notEqual(result, "still-waiting", "completed compacted child must not leave its folder waiting forever");
   assert.ok(childId);
+  if (ownerChange !== "same") {
+    childBatcher.flush();
+    assert.equal((result as { success: boolean }).success, false, "the displaced walk must not report success");
+    if (ownerChange === "active") {
+      assert.equal(sftpTransferCenterStore.getTask(childId)?.status, "transferring");
+      assert.equal(sftpTransferCenterStore.getTask(childId)?.directoryEntryIdentity, "e".repeat(64));
+    } else assert.equal(sftpTransferCenterStore.getTask(childId), undefined, "old waiter must not recreate a failed row");
+    return;
+  }
   assert.equal(sftpTransferCenterStore.getTask(childId), undefined, "completed row was compacted");
   assert.equal(sftpTransferCenterStore.getTask(parent.id)?.directoryResumeCheckpoint?.completedEntries, 1);
   assert.notEqual(result, "still-waiting", "completed compacted child must not leave its folder waiting forever");
@@ -140,6 +188,10 @@ test(`superseded folder child settles when its completion was compacted into the
 });
 }
 }
+}
+}
+}
+}
 
 test("resolveDirectoryResumeTargetRoot prefers staged replace path", () => {
   assert.equal(
@@ -2131,3 +2183,202 @@ for (const scenario of ["deleted-id", "ambiguous-legacy-label"] as const) {
     assert.equal(uploads, 0);
   });
 }
+
+for (const completeDuringSetup of [false, true]) {
+test(`large-history replace recovery retransfers retained completed children into its empty stage: late=${completeDuringSetup}`, async (t) => {
+  const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+  const { createDedicatedResumeChildUpdateBatcher } = await import("../../app/dedicatedResumeProgress");
+  const originalGet = netcattyBridge.get;
+  const previousLocalStorage = Object.getOwnPropertyDescriptor(globalThis, "localStorage");
+  Object.defineProperty(globalThis, "localStorage", {
+    configurable: true, value: { getItem: () => "2", setItem: () => {}, removeItem: () => {} },
+  });
+  const { createRequire } = await import("node:module");
+  const tempDirBridge = createRequire(import.meta.url)("../../../electron/bridges/tempDirBridge.cjs");
+  const root = await fs.promises.mkdtemp(`${tempDirBridge.getTempFilePath("folder-rebuild-test")}-`);
+  const finalPath = path.join(root, "final");
+  const stage = `${finalPath}.netcatty-retained-replace.part`;
+  await fs.promises.mkdir(stage);
+  await fs.promises.writeFile(path.join(stage, "b.txt"), "old contents");
+  const parent: TransferTask = {
+    id: "retained-replace", fileName: "folder", sourcePath: "/remote/folder", targetPath: finalPath,
+    stagedTargetPath: stage, replaceExistingTarget: true, sourceConnectionId: "old-sftp", targetConnectionId: "local",
+    sourceHostId: "h1", direction: "download", status: "transferring", totalBytes: 2,
+    transferredBytes: 1, speed: 0, startTime: 1, isDirectory: true, progressMode: "files", reconnectRequired: true,
+  };
+  const retained: TransferTask = {
+    ...parent, id: "retained-replace-b", parentTaskId: parent.id, isDirectory: false,
+    sourcePath: "/remote/folder/b.txt", targetPath: `${stage}/b.txt`, fileName: "b.txt",
+    status: completeDuringSetup ? "interrupted" : "completed", totalBytes: 10, transferredBytes: 10, checkpointBytes: 10,
+    sourceLastModified: 2, directoryEntryIndex: 1,
+    directoryEntryIdentity: createDirectoryEntryIdentity({
+      sourcePath: "/remote/folder/b.txt", targetPath: `${stage}/b.txt`, size: 10, lastModified: 2,
+    }),
+  };
+  store.upsertTasks([parent, retained]);
+  assert.ok(store.getTask(retained.id), "missing earlier manifest index retains the completed row");
+  const batcher = createDedicatedResumeChildUpdateBatcher({
+    getTaskCount: () => 4096,
+    hasTask: (id) => !!store.getTask(id),
+    upsertTasks: (updates) => store.upsertTasks(updates),
+  });
+  let releaseFirst!: () => void;
+  const secondDispatched = new Promise<void>((resolve) => { releaseFirst = resolve; });
+  t.after(async () => {
+    await fs.promises.rm(root, { recursive: true, force: true });
+    netcattyBridge.get = originalGet;
+    store.patchTask(parent.id, { status: "completed" });
+    store.dismiss(parent.id);
+    if (previousLocalStorage) Object.defineProperty(globalThis, "localStorage", previousLocalStorage);
+    else Reflect.deleteProperty(globalThis, "localStorage");
+    resetDedicatedSessionOpenGateForTests();
+  });
+  (netcattyBridge as { get: () => unknown }).get = () => ({
+    openSftp: async () => "dedicated-sftp", closeSftp: async () => {},
+    listSftp: async (_id: string, remotePath: string) => {
+      if (completeDuringSetup) store.patchTask(retained.id, { status: "completed", ownerId: "dedicated-resume" });
+      return remotePath === parent.sourcePath
+        ? ["a.txt", "b.txt"].map((name) => ({ name, type: "file", size: 10, lastModified: 2 })) : [];
+    },
+    mkdirLocal: async (target: string) => { await fs.promises.mkdir(target, { recursive: true }); },
+    statLocal: async (target: string) => {
+      try {
+        const stat = await fs.promises.stat(target);
+        return { type: stat.isDirectory() ? "directory" : "file", size: stat.size, lastModified: stat.mtimeMs };
+      } catch (error) {
+        if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
+        throw error;
+      }
+    },
+    statSftp: async (_id: string, source: string) => {
+      if (source.endsWith("a.txt")) await secondDispatched;
+      return { type: "file", size: 10, lastModified: 2 };
+    },
+    deleteLocalFile: async (target: string) => { await fs.promises.rm(target, { recursive: true }); },
+    renameLocalFile: async (source: string, target: string) => { await fs.promises.rename(source, target); },
+    startStreamTransfer: async (options: { targetPath: string; checkpointBytes: number }) => {
+      assert.equal(options.checkpointBytes, 0, "an empty replacement stage cannot reuse byte checkpoints");
+      await fs.promises.writeFile(options.targetPath, `new ${path.basename(options.targetPath)}`);
+      return {};
+    },
+  });
+  const result = await resumeTransferWithDedicatedSession(parent, {
+    hosts: [host("h1", "box")], keys: [], identities: [],
+  }, undefined, {
+    children: [retained],
+    flushChildUpdates: () => batcher.flush(),
+    onChildUpdate: (task, observation) => {
+      batcher.push(task, observation);
+      if (task.id === retained.id && task.status === "completed") releaseFirst();
+      return true;
+    },
+    onDirectoryCheckpointUpdate: (value) => store.patchTask(parent.id, { directoryResumeCheckpoint: value }),
+  });
+  batcher.flush();
+  assert.equal(result.success, true, result.error);
+  assert.deepEqual((await fs.promises.readdir(finalPath)).sort(), ["a.txt", "b.txt"],
+    "published replacement must include the previously completed file");
+  assert.equal(await fs.promises.readFile(path.join(finalPath, "b.txt"), "utf8"), "new b.txt");
+});
+}
+
+for (const boundary of ["before-admission", "source-missing", "source-modified", "after-completion", "after-commit"] as const) {
+  test(`dedicated recovery preserves replacement ownership ${boundary}`, async (t) => {
+    const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+    const { createDedicatedResumeChildUpdateBatcher } = await import("../../app/dedicatedResumeProgress");
+    const originalGet = netcattyBridge.get;
+    const originalStorage = Object.getOwnPropertyDescriptor(globalThis, "localStorage");
+    Object.defineProperty(globalThis, "localStorage", { configurable: true, value: {
+      getItem: () => "2", setItem: () => {}, removeItem: () => {},
+    } });
+    const parent: TransferTask = {
+      id: `ownership-${boundary}`, fileName: "folder", sourcePath: "/local/folder", targetPath: "/remote/folder",
+      sourceConnectionId: "local", targetConnectionId: "old", targetHostId: "h1", direction: "upload",
+      status: "transferring", totalBytes: 2, transferredBytes: 0, speed: 0, startTime: 1,
+      isDirectory: true, progressMode: "files", reconnectRequired: true,
+    };
+    const child: TransferTask = {
+      ...parent, id: `${parent.id}-a`, parentTaskId: parent.id, isDirectory: false,
+      sourcePath: "/local/folder/a", targetPath: "/remote/folder/a", fileName: "a", totalBytes: 1,
+      status: "interrupted", directoryEntryIndex: 0, directoryEntryIdentity: "a".repeat(64), sourceLastModified: 1,
+    };
+    store.upsertTasks([parent, boundary === "before-admission"
+      ? { ...child, status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }
+      : child]);
+    const batch = createDedicatedResumeChildUpdateBatcher({
+      getTaskCount: () => 4096, hasTask: (id) => id === child.id || id.startsWith("peer-"),
+      upsertTasks: (rows) => store.upsertTasks(rows.filter((row) => !row.id.startsWith("peer-"))),
+    });
+    if (boundary === "before-admission") {
+      for (let n = 0; n < 511; n++) batch.push({ ...child, id: `peer-${n}` });
+    }
+    let releaseB!: () => void;
+    const bGate = new Promise<void>((resolve) => { releaseB = resolve; });
+    let timedOut = false;
+    const watchdog = setTimeout(() => { timedOut = true; releaseB(); }, 1000);
+    t.after(() => {
+      clearTimeout(watchdog);
+      releaseB();
+      batch.flush();
+      store.patchTask(parent.id, { status: "completed" });
+      store.dismiss(parent.id);
+      netcattyBridge.get = originalGet;
+      if (originalStorage) Object.defineProperty(globalThis, "localStorage", originalStorage);
+      else Reflect.deleteProperty(globalThis, "localStorage");
+      resetDedicatedSessionOpenGateForTests();
+    });
+    let starts = 0;
+    let displaced = 0;
+    let replaced = false;
+    (netcattyBridge as { get: () => unknown }).get = () => ({
+      openSftp: async () => "dedicated", closeSftp: async () => {
+        if (boundary === "after-commit") {
+          assert.equal(store.getTask(child.id), undefined, "completion must be committed before async cleanup");
+          store.upsertTasks([{ ...child, status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
+        }
+      },
+      listLocalTree: async () => (boundary === "after-completion" ? ["a", "b"] : ["a"]).map((name) => ({
+        localPath: `/local/folder/${name}`, relativePath: name, type: "file", size: 1, lastModified: 1,
+      })),
+      mkdirSftp: async () => {}, statLocal: async () => {
+        if (boundary === "source-missing" || boundary === "source-modified") {
+          store.upsertTasks([{ ...child, status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
+          if (boundary === "source-missing") return null;
+          return { type: "file", size: 1, lastModified: 2 };
+        }
+        return { type: "file", size: 1, lastModified: 1 };
+      },
+      startStreamTransfer: async (options: { sourcePath: string }) => {
+        starts++;
+        if (options.sourcePath.endsWith("/b")) await bGate;
+        return {};
+      },
+    });
+    const result = await resumeTransferWithDedicatedSession(parent, {
+      hosts: [host("h1", "box")], keys: [], identities: [],
+    }, undefined, {
+      children: [child],
+      onChildSuperseded: (id) => { displaced++; batch.discard(id); },
+      flushChildUpdates: () => batch.flush(),
+      onChildUpdate: (task, observation) => {
+        batch.push(task, observation);
+        if (boundary === "after-completion" && task.id === child.id && task.status === "completed" && !replaced) {
+          replaced = true;
+          queueMicrotask(() => {
+            try {
+              store.upsertTasks([{ ...task, status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
+            } finally { releaseB(); }
+          });
+        }
+        return true;
+      },
+    });
+    batch.flush();
+    assert.equal(timedOut, false);
+    assert.equal(result.success, boundary === "after-commit", "only a committed success may precede a legitimate replacement");
+    assert.equal(store.getTask(child.id)?.directoryEntryIdentity, "b".repeat(64));
+    assert.equal(store.getTask(child.id)?.status, "transferring");
+    if (boundary !== "after-completion" && boundary !== "after-commit") assert.equal(starts, 0, "source validation and a full batch must not bypass ownership");
+    else assert.equal(displaced, boundary === "after-commit" ? 0 : 1);
+  });
+}

application/state/sftp/dedicatedTransferResume.ts

@@ -1,4 +1,4 @@
-import { runTransferAndWaitForOwner } from "./waitForTransferOwner";
+import { runTransferAndWaitForOwner, TransferOwnerChangedError, type TransferOwnerObservation } from "./waitForTransferOwner";
 import type { Host, Identity, KnownHost, SSHKey, TerminalSettings, TransferTask } from "../../../domain/models";
 import { validateTransferResumeSource } from "../../../domain/sftpTransferCenter";
 import { STORAGE_KEY_SFTP_TRANSFER_CONCURRENCY } from "../../../infrastructure/config/storageKeys";
@@ -62,7 +62,10 @@ export type DedicatedResumeResult = {
 
 export type DedicatedResumeOptions = {
   children?: readonly TransferTask[];
-  onChildUpdate?: (child: TransferTask) => void;
+  /** Return true only when the consumer takes responsibility for disposing the observation. */
+  onChildUpdate?: (child: TransferTask, observation?: TransferOwnerObservation) => unknown;
+  onChildSuperseded?: (taskId: string) => void;
+  flushChildUpdates?: () => void;
   onDirectoryCheckpointUpdate?: (checkpoint: TransferTask["directoryResumeCheckpoint"]) => void;
   shouldAbort?: () => boolean;
 };
@@ -1090,6 +1093,12 @@ async function resumeDirectoryWithDedicatedSession(
           // the source since the interrupted attempt, so rebuild it from empty.
           await resetDirectoryReplaceStage(parent, endpoints, targetSftpId);
         }
+        // Session setup and traversal may finish more children. Capture only
+        // after resetting the stage/checkpoint: all these completions refer to
+        // the discarded destination, while later completions remain valid.
+        const completedAtRestart = new Map(sftpTransferCenterStore.getSnapshot().tasks
+          .filter((child) => child.parentTaskId === parent.id && child.status === "completed")
+          .map((child) => [child.id, child]));
         const destRoot = resolveDirectoryResumeTargetRoot(parent);
         if (endpoints.isDownload) await ensureLocalDir(destRoot);
         if (targetSftpId) await ensureRemoteDir(targetSftpId, destRoot);
@@ -1188,8 +1197,25 @@ async function resumeDirectoryWithDedicatedSession(
             }
             if (targetSftpId) await ensureRemoteDir(targetSftpId, getParentPath(file.targetPath));
 
+            let countedCompleted = false;
+            const onOwnerChanged = () => {
+              if (countedCompleted) {
+                countedCompleted = false;
+                completedCount -= 1;
+                failedCount += 1;
+              }
+              options?.onChildSuperseded?.(childId);
+            };
+            let childObservation: TransferOwnerObservation | undefined = sftpTransferCenterStore.observeTaskSettlement(
+              childBase, resetPersistedCheckpoint ? completedAtRestart.get(childId) : undefined, onOwnerChanged,
+            );
+            const publishChild = (child: TransferTask) => {
+              if (childObservation?.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed before child update");
+              if (options?.onChildUpdate?.(child, childObservation) === true) childObservation = undefined;
+            };
             try {
               if (options?.shouldAbort?.()) throw new Error("Transfer cancelled");
+              if (childObservation.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed before source validation");
 
               const sourceType = endpoints.isUpload ? "local" as const : "sftp" as const;
               const targetType = endpoints.isDownload ? "local" as const : "sftp" as const;
@@ -1217,8 +1243,7 @@ async function resumeDirectoryWithDedicatedSession(
                   sourceLastModified: sourceStat.lastModified,
                 };
               } else if (classified.kind === "modified") {
-                attentionCount += 1;
-                options?.onChildUpdate?.({
+                publishChild({
                   ...childBase,
                   status: "attention",
                   error: classified.message || validationError || "Source was modified",
@@ -1227,6 +1252,7 @@ async function resumeDirectoryWithDedicatedSession(
                   phase: undefined,
                   retryable: true,
                 });
+                attentionCount += 1;
                 return;
               } else if (classified.kind === "fatal") {
                 throw new Error(classified.message || validationError || "Resume validation failed");
@@ -1251,41 +1277,49 @@ async function resumeDirectoryWithDedicatedSession(
 
               // Re-check abort after async stat before inserting a transferring child.
               if (options?.shouldAbort?.()) throw new Error("Transfer cancelled");
-              options?.onChildUpdate?.(childBase);
+              const streamResult = await runTransferAndWaitForOwner(childBase, () => {
+                // Publish only after admission, while identity changes are observed.
+                options?.onChildUpdate?.(childBase);
+                return bridge.startStreamTransfer!({
+                  transferId: childId,
+                  // Lifecycle events must carry the current identity even while
+                  // large-history renderer child updates remain batched.
+                  parentTaskId: childBase.parentTaskId,
+                  directoryEntryIndex: childBase.directoryEntryIndex,
+                  directoryEntryIdentity: childBase.directoryEntryIdentity,
+                  sourcePath: file.sourcePath,
+                  targetPath: file.targetPath,
+                  sourceType,
+                  targetType,
+                  sourceSftpId,
+                  targetSftpId,
+                  sourceHostId: endpoints.sourceHost?.id,
+                  targetHostId: endpoints.targetHost?.id,
+                  totalBytes: Number.isFinite(childBase.totalBytes)
+                    ? childBase.totalBytes
+                    : (file.size || undefined),
+                  resumable: parent.resumable !== false,
+                  checkpointBytes: childBase.checkpointBytes ?? 0,
+                  resumeStage: childBase.resumeStage,
+                  downloadCheckpointBytes: childBase.downloadCheckpointBytes,
+                  uploadCheckpointBytes: childBase.uploadCheckpointBytes,
+                  sourceFingerprint: childBase.sourceFingerprint,
+                  skipAdmission: true,
+                });
+              }, () => options?.shouldAbort?.() === true, pausedAtResume.get(childId),
+                resetPersistedCheckpoint ? completedAtRestart.get(childId) : undefined,
+                // Drop queued UI updates at the ownership change itself: waiting
+                // for this invocation's reply may allow an intervening batch flush.
+                onOwnerChanged,
+                (observation) => { childObservation = observation; return true; }, childObservation);
 
-              const streamResult = await runTransferAndWaitForOwner(childBase, () => bridge.startStreamTransfer!({
-                transferId: childId,
-                // Lifecycle events must carry the current identity even while
-                // large-history renderer child updates remain batched.
-                parentTaskId: childBase.parentTaskId,
-                directoryEntryIndex: childBase.directoryEntryIndex,
-                directoryEntryIdentity: childBase.directoryEntryIdentity,
-                sourcePath: file.sourcePath,
-                targetPath: file.targetPath,
-                sourceType,
-                targetType,
-                sourceSftpId,
-                targetSftpId,
-                sourceHostId: endpoints.sourceHost?.id,
-                targetHostId: endpoints.targetHost?.id,
-                totalBytes: Number.isFinite(childBase.totalBytes)
-                  ? childBase.totalBytes
-                  : (file.size || undefined),
-                resumable: parent.resumable !== false,
-                checkpointBytes: childBase.checkpointBytes ?? 0,
-                resumeStage: childBase.resumeStage,
-                downloadCheckpointBytes: childBase.downloadCheckpointBytes,
-                uploadCheckpointBytes: childBase.uploadCheckpointBytes,
-                sourceFingerprint: childBase.sourceFingerprint,
-                skipAdmission: true,
-              }), () => options?.shouldAbort?.() === true, pausedAtResume.get(childId));
+              if (childObservation?.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed before child update");
 
               if (streamResult?.error || streamResult?.cancelled) {
                 throw new Error(streamResult.error || "Transfer cancelled");
               }
 
-              completedCount += 1;
-              options?.onChildUpdate?.({
+              publishChild({
                 ...childBase,
                 status: "completed",
                 transferredBytes: Number.isFinite(childBase.totalBytes)
@@ -1297,13 +1331,21 @@ async function resumeDirectoryWithDedicatedSession(
                 reconnectRequired: false,
                 phase: undefined,
               });
+              countedCompleted = true;
+              completedCount += 1;
               bumpParentProgress(0);
             } catch (error) {
               if (options?.shouldAbort?.() || /cancelled|canceled/i.test(error instanceof Error ? error.message : String(error))) {
                 throw error instanceof Error ? error : new Error(String(error));
               }
               failedCount += 1;
-              options?.onChildUpdate?.({
+              // Another invocation owns this ID now. Report this walk's failure
+              // without replacing the winner or recreating its compacted row.
+              if (error instanceof TransferOwnerChangedError || childObservation?.hasIdentityConflict()) {
+                options?.onChildSuperseded?.(childId);
+                return;
+              }
+              publishChild({
                 ...childBase,
                 status: "failed",
                 error: error instanceof Error ? error.message : String(error),
@@ -1312,6 +1354,8 @@ async function resumeDirectoryWithDedicatedSession(
                 reconnectRequired: false,
                 phase: undefined,
               });
+            } finally {
+              childObservation?.dispose();
             }
           },
         );
@@ -1322,6 +1366,10 @@ async function resumeDirectoryWithDedicatedSession(
 
     if (result?.error) throw new Error(result.error);
 
+    // Commit deferred child outcomes before deciding success or promoting a stage.
+    // Async connection cleanup must not outlive provisional completion evidence.
+    options?.flushChildUpdates?.();
+
     if (attentionCount > 0 && failedCount === 0 && completedCount + attentionCount >= totalFiles) {
       return {
         success: false,

application/state/sftp/transferDirectoryOps.discovery.test.tsx

@@ -334,7 +334,8 @@ test("live directory download rejects a Windows backslash traversal entry", asyn
   }
 });
 
-test("live folder settles a superseded child whose completed row was compacted", async () => {
+for (const ownerChange of ["same", "changed-active", "changed-completed"] as const) {
+test(`live folder settles superseded ownership without clobbering its replacement: ${ownerChange}`, async () => {
   const { sftpTransferCenterStore } = await import("../sftpTransferCenterStore");
   const previousWindow = (globalThis as { window?: unknown }).window;
   const previousLocalStorage = Object.getOwnPropertyDescriptor(globalThis, "localStorage");
@@ -355,10 +356,18 @@ test("live folder settles a superseded child whose completed row was compacted",
     startStreamTransfer: async (options: { transferId: string }) => {
       childId = options.transferId;
       sftpTransferCenterStore.publishOwner("live-compacted-owner", tasks);
-      // The winning invocation has already completed; history compaction drops its row.
-      sftpTransferCenterStore.ingestBackgroundEvent({
-        type: "completed", transferId: childId, transferred: 1, totalBytes: 1, lifecycleEpoch: 0,
-      });
+      if (ownerChange !== "same") {
+        const current = sftpTransferCenterStore.getTask(childId)!;
+        sftpTransferCenterStore.admitTaskRun({ ...current, directoryEntryIdentity: "b".repeat(64) });
+      }
+      if (ownerChange !== "changed-active") {
+        // The winning invocation completed; history compaction drops its row.
+        sftpTransferCenterStore.ingestBackgroundEvent({
+          type: "completed", transferId: childId, transferred: 1, totalBytes: 1, lifecycleEpoch: 0,
+        });
+      }
+      tasks = [...sftpTransferCenterStore.getSnapshot().tasks];
+      transfersRef.current = tasks;
       return { superseded: true };
     },
   } };
@@ -387,10 +396,15 @@ test("live folder settles a superseded child whose completed row was compacted",
       new Promise<"still-waiting">((resolve) => setTimeout(() => resolve("still-waiting"), 450)),
     ]);
     assert.ok(childId);
-    assert.equal(sftpTransferCenterStore.getTask(childId), undefined);
-    assert.equal(sftpTransferCenterStore.getTask(root.id)?.directoryResumeCheckpoint?.completedEntries, 1);
+    if (ownerChange === "changed-active") {
+      assert.equal(tasks.find((task) => task.id === childId)?.status, "transferring");
+      assert.equal(tasks.find((task) => task.id === childId)?.directoryEntryIdentity, "b".repeat(64));
+    } else {
+      assert.equal(tasks.find((task) => task.id === childId), undefined, "old waiter must not reinsert a compacted child");
+      assert.equal(sftpTransferCenterStore.getTask(root.id)?.directoryResumeCheckpoint?.completedEntries, 1);
+    }
     assert.notEqual(result, "still-waiting", "compacted completion must settle the live folder");
-    assert.equal(result, 0);
+    assert.equal(result, ownerChange === "same" ? 0 : 1);
   } finally {
     cancelledTasksRef.current.add(root.id);
     await running?.catch(() => {});
@@ -404,3 +418,4 @@ test("live folder settles a superseded child whose completed row was compacted",
   }
 });
 
+}

application/state/sftp/transferDirectoryOps.ts

@@ -1,5 +1,5 @@
 import { reconcileSupersededControls } from "./globalSftpTransferControl";
-import { runTransferAndWaitForOwner } from "./waitForTransferOwner";
+import { runTransferAndWaitForOwner, TransferOwnerChangedError } from "./waitForTransferOwner";
 import { useCallback, type Dispatch, type MutableRefObject, type SetStateAction } from "react";
 import type { Host, SftpFileEntry, SftpFilenameEncoding, TransferStatus, TransferTask } from "../../../domain/models";
 import {
@@ -978,6 +978,10 @@ export function useSftpDirectoryTransferOps({
               errors.push(err instanceof Error ? err : new Error(message));
               return;
             }
+            if (err instanceof TransferOwnerChangedError) {
+              errors.push(err);
+              return;
+            }
             // Mark child as failed
             setTransfers((prev) =>
               prev.map((t) =>

application/state/sftp/transferSettlementObservation.test.ts

@@ -169,3 +169,284 @@ test("fresh directory recovery authorizes only an unchanged retained pause under
   store.patchTask("root", { status: "paused", lifecycleEpoch: 6 });
   assert.equal(store.admitTaskRun(task, currentPause), "paused", "parent pause still wins");
 });
+
+for (const compactNewOwner of [false, true]) {
+  test(`superseded waiter rejects a changed identity even after new owner compaction: ${compactNewOwner}`, async (t) => {
+    const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+    const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+    const task = { ...child(), id: `changed-owner-${compactNewOwner}`, parentTaskId: `changed-root-${compactNewOwner}` };
+    store.upsertTasks([{ ...task, id: task.parentTaskId, parentTaskId: undefined, isDirectory: true }, task]);
+    let abort = false;
+    const running = runTransferAndWaitForOwner(task, async () => {
+      const changed = { ...task, directoryEntryIdentity: "b".repeat(64) };
+      assert.equal(store.admitTaskRun(changed), "ready");
+      if (compactNewOwner) {
+        store.patchTask(changed.id, { status: "completed" });
+        assert.equal(store.getTask(changed.id), undefined);
+      }
+      return { superseded: true };
+    }, () => abort);
+    const settled = running.then(() => "completed", (error: Error) => error.message);
+    t.after(async () => {
+      abort = true;
+      await settled;
+      store.patchTask(task.parentTaskId, { status: "completed" });
+      store.dismiss(task.parentTaskId);
+    });
+    assert.match(String(await Promise.race([
+      settled, new Promise((resolve) => setTimeout(() => resolve("still-waiting"), 450)),
+    ])), /identity changed/i);
+  });
+}
+
+test("destination rebuild restarts only the captured completion and preserves newer controls", async (t) => {
+  const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
+  const store = createSftpTransferCenterStore();
+  const task = { ...child(), parentTaskId: undefined, status: "completed" as const, checkpointBytes: 10 };
+  store.upsertTasks([task]);
+  const captured = store.getTask(task.id)!;
+  const restart = { ...task, status: "transferring" as const, checkpointBytes: 0, transferredBytes: 0 };
+  latchTransferPause(task.id);
+  t.after(resetTransferPauseLatchesForTests);
+  assert.equal(store.admitTaskRun(restart, undefined, captured), "paused");
+  resetTransferPauseLatchesForTests();
+  assert.equal(store.admitTaskRun(restart, undefined, captured), "ready");
+  assert.equal(store.getTask(task.id)?.checkpointBytes, 0);
+  store.patchTask(task.id, { status: "completed" });
+  assert.equal(store.admitTaskRun(restart, undefined, captured), "completed", "new completion must not be restarted");
+  store.patchTask(task.id, { status: "cancelled" });
+  assert.equal(store.admitTaskRun(restart, undefined, captured), "cancelled");
+});
+
+test("destination rebuild waiting on pause reuses a newer compacted completion", async (t) => {
+  const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+  const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+  const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
+  const task = { ...child(), id: "restart-wait-child", parentTaskId: "restart-wait-root", directoryEntryIndex: 1 };
+  const root = { ...task, id: task.parentTaskId, parentTaskId: undefined, isDirectory: true };
+  store.upsertTasks([root, { ...task, status: "completed" }]);
+  const captured = store.getTask(task.id)!;
+  latchTransferPause(task.id);
+  let starts = 0;
+  let abort = false;
+  const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort, undefined, captured);
+  const settled = running.then(() => "completed", (error: Error) => error.message);
+  t.after(async () => {
+    abort = true;
+    resetTransferPauseLatchesForTests();
+    await settled;
+    store.patchTask(root.id, { status: "completed" });
+    store.dismiss(root.id);
+  });
+  // An unrelated lifecycle emit must not pin the old completion in the observation.
+  store.patchTask(root.id, { reconnectRequired: true });
+  assert.equal(await Promise.race([
+    settled, new Promise((resolve) => setTimeout(() => resolve("still-waiting"), 230)),
+  ]), "still-waiting", "old completion cannot finish recovery while a new transfer is still needed");
+  assert.equal(starts, 0);
+  resetTransferPauseLatchesForTests();
+  store.patchTask(task.id, { status: "transferring", lifecycleEpoch: 1 });
+  store.upsertTasks([{ ...task, id: "restart-wait-first", directoryEntryIndex: 0, status: "completed" }]);
+  store.patchTask(task.id, { status: "completed", lifecycleEpoch: 2 });
+  assert.equal(store.getTask(task.id), undefined);
+  assert.equal(await settled, "completed");
+  assert.equal(starts, 0, "new completion after stage rebuild must be reused even when its row was compacted");
+});
+
+test("paused admission cannot reclaim an identity replaced while it waited", async (t) => {
+  const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+  const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+  const task = { ...child(), id: "paused-displaced", parentTaskId: undefined };
+  store.upsertTasks([{ ...task, status: "paused" }]);
+  let starts = 0;
+  let abort = false;
+  const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort);
+  const settled = running.then(() => "completed", (error: Error) => error.message);
+  t.after(async () => {
+    abort = true;
+    await settled;
+    store.patchTask(task.id, { status: "completed" });
+    store.dismiss(task.id);
+  });
+  store.upsertTasks([{ ...task, status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
+  assert.match(await settled, /identity changed/i);
+  assert.equal(starts, 0);
+  assert.equal(store.getTask(task.id)?.directoryEntryIdentity, "b".repeat(64));
+});
+
+for (const identityChangesDuringWait of [false, true]) {
+  test(`paused recovery distinguishes initial stale identity from later ownership: changed=${identityChangesDuringWait}`, async (t) => {
+    const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+    const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+    const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
+    const task = { ...child(), id: `paused-origin-${identityChangesDuringWait}`, parentTaskId: `paused-origin-root-${identityChangesDuringWait}` };
+    const root = { ...task, id: task.parentTaskId, parentTaskId: undefined, isDirectory: true };
+    store.upsertTasks([root, {
+      ...task, status: identityChangesDuringWait ? "failed" : "paused",
+      directoryEntryIdentity: identityChangesDuringWait ? task.directoryEntryIdentity : "b".repeat(64),
+    }]);
+    const initialPause = store.getTask(task.id)!;
+    latchTransferPause(task.id);
+    let starts = 0;
+    let abort = false;
+    let conflicts = 0;
+    const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort,
+      identityChangesDuringWait ? undefined : initialPause, undefined, () => { conflicts += 1; });
+    const settled = running.then(() => "completed", (error: Error) => error.message);
+    t.after(async () => {
+      abort = true;
+      resetTransferPauseLatchesForTests();
+      await settled;
+      store.patchTask(root.id, { status: "completed" });
+      store.dismiss(root.id);
+    });
+    // This captures the prior failed result, or republishes the unchanged
+    // initial old identity. Neither is evidence of a replacement by itself.
+    store.patchTask(root.id, { reconnectRequired: true });
+    resetTransferPauseLatchesForTests();
+    if (identityChangesDuringWait) {
+      store.patchTask(task.id, { status: "transferring", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 });
+    }
+    const outcome = await settled;
+    if (identityChangesDuringWait) {
+      assert.match(outcome, /identity changed/i);
+      assert.equal(starts, 0);
+      assert.equal(conflicts, 1);
+      assert.equal(store.getTask(task.id)?.directoryEntryIdentity, "b".repeat(64));
+    } else {
+      assert.equal(outcome, "completed");
+      assert.equal(starts, 1);
+      assert.equal(conflicts, 0);
+    }
+  });
+}
+
+test("paused retry observes newer same-identity completion after an old failure", async (t) => {
+  const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+  const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+  const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
+  const task = { ...child(), id: "failed-then-completed", parentTaskId: "failed-then-completed-root" };
+  const root = { ...task, id: task.parentTaskId, parentTaskId: undefined, isDirectory: true };
+  store.upsertTasks([root, { ...task, status: "failed" }]);
+  latchTransferPause(task.id);
+  let starts = 0;
+  let abort = false;
+  const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort);
+  const settled = running.then(() => "completed", (error: Error) => error.message);
+  t.after(async () => {
+    abort = true;
+    resetTransferPauseLatchesForTests();
+    await settled;
+    store.patchTask(root.id, { status: "completed" });
+    store.dismiss(root.id);
+  });
+  store.patchTask(root.id, { reconnectRequired: true });
+  resetTransferPauseLatchesForTests();
+  store.patchTask(task.id, { status: "transferring", lifecycleEpoch: 1 });
+  store.patchTask(task.id, { status: "completed", lifecycleEpoch: 2 });
+  assert.equal(store.getTask(task.id), undefined);
+  assert.equal(await settled, "completed");
+  assert.equal(starts, 0, "a newer completed attempt must replace the captured old failure");
+});
+
+test("an unchanged owner retains its original transport rejection", async (t) => {
+  const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+  const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+  const task = { ...child(), id: "unchanged-rejected-owner", parentTaskId: undefined };
+  store.upsertTasks([task]);
+  t.after(() => { store.patchTask(task.id, { status: "completed" }); store.dismiss(task.id); });
+  const error = new Error("original transport failure");
+  await assert.rejects(runTransferAndWaitForOwner(task, async () => { throw error; }, () => false),
+    (actual) => actual === error);
+});
+
+for (const activation of ["patch", "progress", "owner"] as const) {
+test(`initial stale identity exemption ends when another owner activates that row: ${activation}`, async (t) => {
+  const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+  const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+  const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
+  const task = { ...child(), id: "activated-initial-identity", parentTaskId: undefined };
+  store.upsertTasks([{ ...task, status: "paused", directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
+  const initial = store.getTask(task.id)!;
+  latchTransferPause(task.id);
+  let starts = 0;
+  let abort = false;
+  const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort, initial);
+  const settled = running.then(() => "completed", (error: Error) => error.message);
+  t.after(async () => {
+    abort = true;
+    resetTransferPauseLatchesForTests();
+    await settled;
+    store.patchTask(task.id, { status: "completed" });
+    store.dismiss(task.id);
+  });
+  resetTransferPauseLatchesForTests();
+  if (activation === "progress") {
+    store.ingestBackgroundEvent({ type: "progress", transferId: task.id,
+      transferred: 1, lifecycleEpoch: 2, lifecycleState: "transferring" });
+  } else if (activation === "owner") store.patchTask(task.id, { ownerId: "replacement-owner" });
+  else store.patchTask(task.id, { status: "transferring", lifecycleEpoch: 2 });
+  assert.match(String(await Promise.race([
+    settled, new Promise((resolve) => setTimeout(() => resolve("still-waiting"), 450)),
+  ])), /identity changed/i);
+  assert.equal(starts, 0);
+  assert.equal(store.getTask(task.id)?.directoryEntryIdentity, "b".repeat(64));
+});
+
+}
+
+test("already active different identity is rejected before first admission", async (t) => {
+  const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+  const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+  const task = { ...child(), id: "already-active-identity", parentTaskId: undefined };
+  store.upsertTasks([{ ...task, directoryEntryIdentity: "b".repeat(64), lifecycleEpoch: 1 }]);
+  t.after(() => { store.patchTask(task.id, { status: "completed" }); store.dismiss(task.id); });
+  let starts = 0;
+  let conflicts = 0;
+  await assert.rejects(runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => false,
+    undefined, undefined, () => { conflicts += 1; }), /identity changed/i);
+  assert.equal(starts, 0);
+  assert.equal(conflicts, 1);
+  assert.equal(store.getTask(task.id)?.directoryEntryIdentity, "b".repeat(64));
+});
+
+for (const replacementStatus of ["transferring", "completed"] as const) {
+  for (const oldStatus of ["completed", "failed"] as const) {
+    test(`deferred ${oldStatus} cannot overwrite a ${replacementStatus} replacement after waiter exit`, async () => {
+      const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+      const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+      const { createDedicatedResumeChildUpdateBatcher } = await import("../../app/dedicatedResumeProgress");
+      const task = child();
+      const root = { ...task, id: task.parentTaskId!, parentTaskId: undefined, isDirectory: true };
+      const batcher = createDedicatedResumeChildUpdateBatcher({
+        getTaskCount: () => 4096, hasTask: () => true,
+        upsertTasks: (updates) => store.upsertTasks(updates),
+      });
+      store.upsertTasks([root, task]);
+      let observation: ReturnType<typeof store.observeTaskSettlement> | undefined;
+      try {
+        await runTransferAndWaitForOwner(task, async () => {
+          store.patchTask(task.id, { status: oldStatus });
+          return {};
+        }, () => false, undefined, undefined, () => batcher.discard(task.id), (value) => {
+          observation = value;
+          return true;
+        });
+        assert.ok(observation);
+        batcher.push({ ...task, status: oldStatus }, observation);
+        store.upsertTasks([{ ...task, directoryEntryIdentity: "b".repeat(64), status: replacementStatus }]);
+        const replacement = store.getTask(task.id);
+        batcher.flush();
+        assert.equal(store.getTask(task.id), replacement);
+        if (replacementStatus === "transferring") assert.equal(replacement?.directoryEntryIdentity, "b".repeat(64));
+        else assert.equal(replacement, undefined, "the replacement remains compacted");
+        assert.equal(observation.read(), undefined, "the retained observation is disposed");
+      } finally {
+        batcher.flush();
+        observation?.dispose();
+        store.patchTask(root.id, { status: "completed" });
+        store.dismiss(root.id);
+      }
+    });
+  }
+}

application/state/sftp/waitForTransferOwner.ts

@@ -1,6 +1,11 @@
 import type { TransferTask } from "../../../domain/models";
 import { sftpTransferCenterStore } from "../sftpTransferCenterStore";
 
+/** The displaced invocation may fail its walk, but no longer owns the child row. */
+export class TransferOwnerChangedError extends Error {}
+
+export type TransferOwnerObservation = ReturnType<typeof sftpTransferCenterStore.observeTaskSettlement>;
+
 type StreamResult = { error?: string; cancelled?: boolean; superseded?: boolean } | undefined;
 
 /** A live waiter owns bounded settlement evidence; persisted history stays compact. */
@@ -9,33 +14,47 @@ export async function runTransferAndWaitForOwner(
   start: () => Promise<StreamResult>,
   shouldAbort: () => boolean,
   pausedAtResume?: TransferTask,
+  completedAtRestart?: TransferTask,
+  onOwnerChanged?: () => void,
+  retainObservation?: (observation: TransferOwnerObservation) => boolean,
+  existingObservation?: TransferOwnerObservation,
 ): Promise<StreamResult> {
   // Register before admission/start: an owner may finish while dispatch waits for resume.
-  let observation = sftpTransferCenterStore.observeTaskSettlement(task);
+  let observation = existingObservation ?? sftpTransferCenterStore.observeTaskSettlement(task, completedAtRestart, onOwnerChanged);
   try {
     for (;;) {
       if (shouldAbort()) throw new Error("Transfer cancelled");
-      const admission = sftpTransferCenterStore.admitTaskRun(task, pausedAtResume);
+      if (observation.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed before dispatch");
+      const admission = sftpTransferCenterStore.admitTaskRun(task, pausedAtResume, completedAtRestart);
       if (admission === "cancelled") throw new Error("Transfer cancelled");
       if (admission === "completed" || observation.read()?.status === "completed") return {};
-      if (admission === "conflict") throw new Error("Transfer identity changed before dispatch");
+      if (admission === "conflict") throw new TransferOwnerChangedError("Transfer identity changed before dispatch");
       if (admission === "ready") break;
       await new Promise((resolve) => setTimeout(resolve, 200));
     }
     // Admission begins a new attempt; discard any previous failed settlement.
     observation.dispose();
-    observation = sftpTransferCenterStore.observeTaskSettlement(task);
+    observation = sftpTransferCenterStore.observeTaskSettlement(task, undefined, onOwnerChanged);
     const result = await start();
+    if (observation.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed while waiting for its owner");
     if (!result?.superseded) return result;
     for (;;) {
       if (shouldAbort()) throw new Error("Transfer cancelled");
+      if (observation.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed while waiting for its owner");
       const latest = observation.read();
       if (latest?.status === "completed") return { ...result, superseded: false };
       if (latest?.status === "failed") throw new Error(latest.error || "Transfer failed");
       if (latest?.status === "cancelled") throw new Error("Transfer cancelled");
       await new Promise((resolve) => setTimeout(resolve, 200));
     }
+  } catch (error) {
+    // A displaced invocation can reject instead of returning superseded. Its
+    // transport error must not regain permission to write the winning row.
+    if (observation.hasIdentityConflict()) throw new TransferOwnerChangedError("Transfer identity changed while waiting for its owner");
+    throw error;
   } finally {
-    observation.dispose();
+    // A deferred child update may outlive the invocation. Its caller must either
+    // retain this evidence through the write or release it after handling exit.
+    if (!retainObservation?.(observation)) observation.dispose();
   }
 }

application/state/sftpTransferCenterStore.ts

@@ -42,6 +42,14 @@ import { restoreSftpTransferHistoryCooperatively } from "./sftp/transferHistoryR
 import { cancelExternalUploadRuntime } from "./sftp/externalUploadRuntime";
 
 type Listener = () => void;
+type TaskSettlementObserver = {
+  expected: TransferTask;
+  initialInactiveRow?: TransferTask;
+  settled?: TransferTask;
+  conflicted?: boolean;
+  ignoredCompletion?: TransferTask;
+  onIdentityConflict?: () => void;
+};
 
 // Ordinary bounded history restores synchronously so existing callers receive
 // it immediately. Legacy directory snapshots can contain tens of thousands of
@@ -98,9 +106,9 @@ export interface SftpTransferCenterStore {
   /** Single task row — same object identity until that row is patched. */
   getTask(taskId: string): TransferTask | undefined;
   /** Observe exact task settlement before completed child rows are compacted. */
-  observeTaskSettlement(task: TransferTask): { read(): TransferTask | undefined; dispose(): void };
+  observeTaskSettlement(task: TransferTask, ignoredCompletion?: TransferTask, onIdentityConflict?: () => void): { read(): TransferTask | undefined; hasIdentityConflict(): boolean; dispose(): void };
   /** Publish an explicit dispatch identity without flushing large child-history batches. */
-  admitTaskRun(task: TransferTask, pausedAtResume?: TransferTask): "ready" | "paused" | "completed" | "cancelled" | "conflict";
+  admitTaskRun(task: TransferTask, pausedAtResume?: TransferTask, completedAtRestart?: TransferTask): "ready" | "paused" | "completed" | "cancelled" | "conflict";
   getOwnerTasks(ownerId: string): TransferTask[];
   publishOwner(ownerId: string, tasks: readonly TransferTask[]): void;
   registerOwner(ownerId: string, controls: SftpTransferOwnerControls): () => void;
@@ -405,7 +413,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
   let snapshotDirty = false;
   const listeners = new Set<Listener>();
   const progressListeners = new Set<Listener>();
-  const settlementObservers = new Map<string, Set<{ expected: TransferTask; settled?: TransferTask }>>();
+  const settlementObservers = new Map<string, Set<TaskSettlementObserver>>();
   const matchesObservedTask = (expected: TransferTask, candidate: TransferTask) => (
     expected.id === candidate.id
     && expected.sourcePath === candidate.sourcePath
@@ -414,6 +422,29 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
     && expected.directoryEntryIndex === candidate.directoryEntryIndex
     && expected.directoryEntryIdentity === candidate.directoryEntryIdentity
   );
+  const captureObservedTask = (task: TransferTask) => {
+    for (const observer of settlementObservers.get(task.id) ?? []) {
+      if (observer.conflicted || task === observer.ignoredCompletion) continue;
+      const matchesExpected = matchesObservedTask(observer.expected, task);
+      // A retry may plan a new source identity while its initial stale row is
+      // still paused. Only the unchanged initial attempt is exempt: activating
+      // it or replacing its owner/epoch must not let the stale plan reclaim it.
+      if (!matchesExpected && observer.initialInactiveRow
+        && matchesObservedTask(observer.initialInactiveRow, task)
+        && observer.initialInactiveRow.status === task.status
+        && observer.initialInactiveRow.lifecycleEpoch === task.lifecycleEpoch
+        && observer.initialInactiveRow.ownerId === task.ownerId) continue;
+      observer.initialInactiveRow = undefined;
+      if (!matchesExpected) {
+        observer.conflicted = true;
+        observer.onIdentityConflict?.();
+      } else if (observer.settled?.status !== "completed") {
+        // Failed/cancelled evidence belongs to the latest observed attempt. A
+        // resumed owner can supersede it; exact successful completion stays final.
+        observer.settled = TERMINAL_OWNER_STATUSES.has(task.status) ? task : undefined;
+      }
+    }
+  };
   const refreshBadgeSnapshot = () => {
     const next = buildBadgeSnapshot(tasks);
     if (
@@ -633,12 +664,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
     // settles; everything else can release its whole tree here.
     settleFinishedTransferControlState(tasks, tasks);
     if (settlementObservers.size > 0) {
-      for (const task of tasks) {
-        if (!TERMINAL_OWNER_STATUSES.has(task.status)) continue;
-        for (const observer of settlementObservers.get(task.id) ?? []) {
-          if (!observer.settled && matchesObservedTask(observer.expected, task)) observer.settled = task;
-        }
-      }
+      for (const task of tasks) captureObservedTask(task);
     }
     const beforePrune = tasks;
     tasks = pruneSftpTransferHistory(tasks);
@@ -1338,7 +1364,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
     },
     getSnapshot: () => ensureSnapshot(),
     getBadgeSnapshot: () => badgeSnapshot,
-    admitTaskRun(incoming, pausedAtResume) {
+    admitTaskRun(incoming, pausedAtResume, completedAtRestart) {
       const rootId = incoming.parentTaskId ?? incoming.id;
       if (isTransferOrRootCancelled(rootId, incoming.id)) return "cancelled";
       const parent = incoming.parentTaskId ? tasks.find((task) => task.id === rootId) : undefined;
@@ -1348,7 +1374,10 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
       if (existing && (existing.sourcePath !== incoming.sourcePath || existing.targetPath !== incoming.targetPath
         || existing.parentTaskId !== incoming.parentTaskId)) return "conflict";
       if (existing?.status === "cancelled") return "cancelled";
-      if (existing?.status === "completed") {
+      // Only the exact completion captured at the destination reset is
+      // stale. A newer completion or user control must keep its authority.
+      const restartsCompletion = existing?.status === "completed" && existing === completedAtRestart;
+      if (existing?.status === "completed" && !restartsCompletion) {
         return matchesObservedTask(incoming, existing) ? "completed" : "conflict";
       }
       const resumesUnchangedPause = existing === pausedAtResume
@@ -1369,6 +1398,14 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
         ...existing,
         directoryEntryIndex: incoming.directoryEntryIndex,
         directoryEntryIdentity: incoming.directoryEntryIdentity,
+        ...(restartsCompletion ? {
+          checkpointBytes: incoming.checkpointBytes,
+          transferredBytes: incoming.transferredBytes,
+          resumeStage: incoming.resumeStage,
+          downloadCheckpointBytes: incoming.downloadCheckpointBytes,
+          uploadCheckpointBytes: incoming.uploadCheckpointBytes,
+          sourceFingerprint: incoming.sourceFingerprint,
+        } : {}),
         ...(existing.status === "transferring" ? {} : {
           status: "transferring" as const,
           lifecycleEpoch: undefined,
@@ -1378,25 +1415,36 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
         }),
       };
       tasks = next;
+      captureObservedTask(next[index]);
       emitProgress(false);
       return "ready";
     },
-    observeTaskSettlement(expected) {
-      const observer: { expected: TransferTask; settled?: TransferTask } = { expected: { ...expected } };
+    observeTaskSettlement(expected, ignoredCompletion, onIdentityConflict) {
+      const initial = tasks.find((task) => task.id === expected.id);
+      const observer: TaskSettlementObserver = {
+        expected: { ...expected }, ignoredCompletion, onIdentityConflict,
+        initialInactiveRow: initial && ["paused", "failed", "interrupted", "attention"].includes(initial.status)
+          && !matchesObservedTask(expected, initial) ? { ...initial } : undefined,
+      };
       const observers = settlementObservers.get(expected.id) ?? new Set();
       observers.add(observer);
       settlementObservers.set(expected.id, observers);
+      // A different active owner may predate registration. Detect it now,
+      // before admission can publish the stale caller's identity over it.
+      if (initial) captureObservedTask(initial);
       let disposed = false;
       return {
         read() {
           if (disposed) return undefined;
           if (observer.settled) return observer.settled;
           const current = tasks.find((task) => task.id === expected.id);
-          return current && matchesObservedTask(observer.expected, current) ? current : undefined;
+          return current && current !== ignoredCompletion && matchesObservedTask(observer.expected, current) ? current : undefined;
         },
+        hasIdentityConflict: () => !disposed && observer.conflicted === true,
         dispose() {
           disposed = true;
           observer.settled = undefined;
+          observer.conflicted = undefined;
           observers.delete(observer);
           if (observers.size === 0) settlementObservers.delete(expected.id);
         },
@@ -1662,6 +1710,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
       const next = tasks.slice();
       next[index] = nextTask;
       tasks = next;
+      captureObservedTask(nextTask);
       if (changed) {
         const lifecycleChanged = (
           (updates.status !== undefined && updates.status !== task.status)
@@ -2023,7 +2072,8 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
         backgroundEventsDuringRestore.set(event.transferId, pending);
         return;
       }
-      let existing = tasks.find((task) => task.id === event.transferId);
+      let existingIndex = tasks.findIndex((task) => task.id === event.transferId);
+      let existing = tasks[existingIndex];
       const eventParent = event.parentTaskId
         ? tasks.find((task) => task.id === event.parentTaskId)
         : undefined;
@@ -2039,7 +2089,8 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
           directoryEntryIdentity: event.directoryEntryIdentity ?? task.directoryEntryIdentity,
           background: false,
         } : task);
-        existing = tasks.find((task) => task.id === event.transferId);
+        existing = tasks[existingIndex];
+        captureObservedTask(existing);
       }
       // A delayed child progress event can arrive after the renderer compacted
       // that completed child into its parent checkpoint. Never resurrect it as
@@ -2119,6 +2170,7 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
       if ((event.type === "queued" || event.type === "started" || event.type === "progress") && !existing) {
         const sourcePath = event.sourcePath ?? "";
         const targetPath = event.targetPath ?? "";
+        existingIndex = tasks.length;
         tasks.push({
           id: event.transferId,
           ownerId: eventParent?.ownerId
@@ -2352,6 +2404,9 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
           phase: undefined,
         } : task);
       }
+      // Ownership changes must be observed synchronously even when a progress
+      // event takes the lightweight UI path. Reuse the located row, not a scan.
+      if (existingIndex >= 0) captureObservedTask(tasks[existingIndex]);
       // Progress-only ticks skip prune + coalesce listener paints; lifecycle
       // events still take the full emit path so UI never lags on pause/cancel.
       if (event.type === "progress" && !persistImmediately) {

docs/research/sftp-folder-settlement.md

@@ -22,3 +22,121 @@ This addresses a separately reproduced folder-never-settles condition relevant
 to #2568 and #3155. It does not establish the original reporters' precise cause.
 
 Large-history recovery follow-up: stream lifecycle events carry the current child hierarchy identity. Before dispatch, the store admits the explicit retry into the current row without a full history compaction, so a batched old failed row cannot reject the new completion. Admission distinguishes pause waiting, cancellation, identity conflict and exact prior completion; active lifecycle epochs and newer pause/cancel intent remain protected.
+
+## Follow-up: retained completions and changed owners (#3155)
+
+Two remaining #3287 review cases reproduce independently of the original
+Windows report:
+
+- With the 4096-row batching path enabled, a retained completed child at manifest
+  index 1 can reach dispatch before index 0 is published. Rebuilding a replacement
+  stage deletes that child's old output, but completion admission used to skip
+  its new transfer. The actual recovery entrypoint returned success while the
+  promoted directory omitted the file. A filesystem regression now checks both
+  filenames and the retransferred contents; transport and source listing are
+  controlled fixtures, not a real remote server.
+- A superseded invocation waited forever after a new same-ID owner changed file
+  identity, including when the new owner completed and its row was compacted.
+  Both bounded regressions returned `still-waiting` before the fix.
+
+Recovery now authorizes reuse of a completed row for a fresh transfer only when
+it is the exact row captured at the destination/checkpoint reset, after stage
+deletion has finished. This includes files completed during session setup or
+traversal. Newer
+completion, pause and cancellation still win. Reset byte checkpoints and source
+fingerprints are published with that admission. Ordinary recovery continues to
+reuse valid completed work.
+
+Settlement observations retain identity-conflict evidence before compaction,
+including explicit admission changes. Before each admission attempt, an observed identity conflict stops the old
+invocation from reclaiming an active row after a pause. A displaced invocation fails explicitly
+instead of waiting for completion evidence belonging to another file. Both live
+and dedicated walks report that failure without overwriting the new owner's
+child row; dedicated recovery discards its deferred update synchronously when
+the observation detects displacement. This also prevents an automatic 512-entry
+batch flush from publishing the old snapshot while its invoke reply is delayed.
+Exact completion evidence is retained independently of later ownership changes:
+an old success still cannot overwrite a newer owner. Initial retained identities
+are distinguished from changes observed while waiting, and newer same-file
+attempts supersede old failed evidence. Missing rows alone still do not prove
+success. Observations remain scoped to their waiter or its pending child update and are disposed after the final write or discard.
+
+These cases explain specific large-history recovery failures. They do not prove
+that the original Windows v1.1.82 freeze was caused by either case; its file count,
+transfer direction, server and logs remain unavailable.
+
+## Ownership rules for the follow-up
+
+The expected identity is the tuple of task ID, source path, target path, parent
+ID, manifest index and manifest identity. Successful completion evidence and
+permission to update the current row are separate: an old completion never grants
+permission to overwrite a different identity. Identity conflict is sticky for a
+waiter, immediately discards its deferred child update, and ends the old walk
+without writing that child or reporting the walk as successful.
+
+Only two initial-state allowances exist, both bounded to one waiter:
+
+1. An initial **inactive** row (`paused`, `failed`, `interrupted`, `attention`)
+   can precede a fresh recovery plan. It is tolerated only while identity,
+   status, lifecycle epoch and owner still match that initial snapshot. It
+   does not authorize dispatch: the ordinary pause/cancel/admission gates remain.
+2. An exact completed row captured **after** destination/checkpoint reset is
+   obsolete completion evidence for the deleted destination. Only that row
+   object can be restarted. A newer completion cannot inherit this permission.
+
+| Boundary | Same expected identity | Different identity | Missing row |
+| --- | --- | --- | --- |
+| Registration | Capture current state immediately; completion may be reused unless it is the exact reset-invalidated row | Tolerate only the unchanged initial inactive snapshot or exact reset-invalidated completion; already-active differences conflict immediately | No evidence of completion; continue to normal admission |
+| Waiting for admission | A new active/paused state replaces old failed/cancelled evidence; a later success is retained. Pause/cancel gates still apply | Status, epoch or owner changes end the initial inactive allowance; any non-allowed mismatch records conflict and discards pending updates immediately | Retained exact success can settle the wait; absence alone cannot |
+| Before every admission | Check cancellation and recorded conflict first. Reuse exact completion, or restart only the exact reset-invalidated row. An unchanged captured pause requires an active parent and released pause latches | Recorded conflict stops before `admitTaskRun` can overwrite the row; structural mismatch is also rejected by admission | Dispatch only if normal admission allows it and no retained completion/conflict settles the waiter |
+| Invocation resolves | Check conflict before interpreting success or superseded. A superseded call follows observed same-file state | Fail with an ownership-change error; do not write old success/failure into the new row | A superseded call needs retained exact settlement evidence; ordinary success retains existing invocation semantics |
+| Invocation rejects | Without conflict, preserve the original transport error | Convert to ownership-change error so caller does not overwrite the winner | Do not infer success; preserve original error unless conflict was already observed |
+| Exit | Live waiters dispose on every path; dedicated child handlers keep the same observation until their final update is synchronously applied or handed to the batcher | Live/dedicated walks count failure, skip old child writes; dedicated recovery also discards deferred updates | Never recreate a compacted winning row from an obsolete invocation |
+
+For the same expected file identity, an epoch change alone is not an identity
+conflict: the bridge still owns invocation deduplication and lifecycle ordering.
+Epoch/status/owner changes matter when deciding whether a *different* initial
+inactive identity is still the unchanged pre-recovery record. This follow-up does
+not replace the existing same-file control/epoch machinery or introduce a new
+persistent attempt protocol.
+
+Evidence maps directly to these rules: `transferSettlementObservation.test.ts`
+covers registration, waiting, pre-admission, initial inactive/active states,
+old failure versus new success, compacted completion, newer epochs and original
+error propagation. `dedicatedTransferResume.test.ts` exercises actual recovery
+with immediate/batched updates, automatic flush before delayed replies, new
+active/completed owners, old success/rejection, and real filesystem stage reset
+including completion during setup. `transferDirectoryOps.discovery.test.tsx`
+checks the actual live hook does not overwrite replacement owners. Existing
+pause/cancel, directory/store and batching tests remain in the targeted run.
+
+Observation updates follow state writes, not UI notification type. Explicit
+admission, direct row patches and background event ingestion synchronously
+capture the affected row, including progress-only epoch/status/owner changes.
+Lifecycle publication still captures settlement before history compaction.
+The background path reuses its existing row index; progress does not gain a
+full-history scan, serialization or synchronous persistence.
+
+Deferred child writes retain the original observation, without another store
+lookup or observer registration. Dedicated recovery owns it while handling the
+invocation result; the update callback returns `true` only when its consumer
+accepts responsibility for disposal. The batcher stores the update and observation
+together, checks conflict immediately before writing, and disposes on write or
+discard. A callback that does not accept ownership leaves disposal to recovery's
+`finally`. This also covers a new identity appearing after one file finishes
+while other directory files are still running. Evidence includes both old
+success/failure and active/compacted replacement rows after waiter exit.
+
+The write/discard is the ownership deadline for a deferred update, not the end
+of the entire directory. A completed child already written successfully may
+legitimately be replaced later; that does not retroactively fail the old success.
+A completed child still waiting to be written is provisional: if replaced before
+its write, its pending completion count is withdrawn and the old directory fails.
+Observation starts before source validation and is passed into the waiter, so
+validation errors and attention outcomes obey the same write-ownership rule.
+
+Recovery flushes child updates before deciding the directory result or promoting
+a replacement stage. Its later asynchronous connection cleanup cannot outlive
+provisional success. The outer cleanup retains an idempotent flush as an error
+fallback. A regression replaces a child during connection close and verifies its
+old completion was already committed and the legitimate new owner survives.