fix(sftp): resume retained pauses without bypassing newer controls

application/state/sftp/dedicatedTransferResume.test.ts

@@ -39,6 +39,108 @@ const host = (id: string, label: string, hostname = label): Host => ({
   protocol: "ssh",
 } as Host);
 
+for (const retainedStatus of [undefined, "interrupted", "failed", "paused"] as const) {
+for (const newerPause of retainedStatus === "paused" ? [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) => {
+  const { sftpTransferCenterStore } = await import("../sftpTransferCenterStore");
+  const previousLocalStorage = Object.getOwnPropertyDescriptor(globalThis, "localStorage");
+  Object.defineProperty(globalThis, "localStorage", {
+    configurable: true,
+    value: { getItem: () => null, setItem: () => {}, removeItem: () => {} },
+  });
+  const originalGet = netcattyBridge.get;
+  let abort = false;
+  let running: Promise<unknown> | undefined;
+  let childId: string | undefined;
+  const parent: TransferTask = {
+    id: "superseded-compacted-folder", fileName: "folder",
+    sourcePath: "/local/folder", targetPath: "/remote/folder",
+    sourceConnectionId: "local", targetConnectionId: "expired-sftp",
+    targetHostId: "h1", direction: "upload", status: "transferring",
+    totalBytes: 1, transferredBytes: 0, speed: 0, startTime: Date.now(),
+    isDirectory: true, progressMode: "files", reconnectRequired: true,
+  };
+  t.after(async () => {
+    abort = true;
+    await running;
+    netcattyBridge.get = originalGet;
+    sftpTransferCenterStore.patchTask(parent.id, { status: "completed" });
+    sftpTransferCenterStore.dismiss(parent.id);
+    if (previousLocalStorage) Object.defineProperty(globalThis, "localStorage", previousLocalStorage);
+    else Reflect.deleteProperty(globalThis, "localStorage");
+    resetDedicatedSessionOpenGateForTests();
+  });
+  const persisted: TransferTask = {
+    ...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),
+  };
+  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),
+  });
+  sftpTransferCenterStore.publishOwner("compacted-owner", [parent, ...(batchExistingIdentity ? [persisted] : [])]);
+  (netcattyBridge as { get: () => unknown }).get = () => ({
+    openSftp: async () => "dedicated-sftp", closeSftp: async () => {},
+    listLocalTree: async () => [{
+      localPath: "/local/folder/a.bin", relativePath: "a.bin", type: "file", size: 10, lastModified: 1,
+    }],
+    mkdirSftp: async () => {},
+    statLocal: async () => {
+      if (newerPause) sftpTransferCenterStore.patchTask(persisted.id, { status: "paused", lifecycleEpoch: 9 });
+      return { size: 10, lastModified: 1 };
+    },
+    startStreamTransfer: async (options: {
+      transferId: string; parentTaskId?: string; directoryEntryIndex?: number; directoryEntryIdentity?: string;
+    }) => {
+      childId = options.transferId;
+      // Main-process lifecycle events carry the stream's metadata, not queued UI updates.
+      sftpTransferCenterStore.ingestBackgroundEvent({
+        type: "started", transferId: childId, lifecycleEpoch: 0,
+        parentTaskId: options.parentTaskId,
+        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 };
+    },
+  });
+  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]),
+  });
+  const result = await Promise.race([
+    running,
+    new Promise<"still-waiting">((resolve) => setTimeout(() => resolve("still-waiting"), 450)),
+  ]);
+  if (newerPause) {
+    assert.equal(result, "still-waiting", "a later pause must still hold fresh recovery");
+    assert.equal(childId, undefined, "newer paused child must not start");
+    assert.equal(sftpTransferCenterStore.getTask(persisted.id)?.lifecycleEpoch, 9);
+    return;
+  }
+  assert.notEqual(result, "still-waiting", "completed compacted child must not leave its folder waiting forever");
+  assert.ok(childId);
+  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");
+  assert.equal((result as { success: boolean }).success, true);
+});
+}
+}
+
 test("resolveDirectoryResumeTargetRoot prefers staged replace path", () => {
   assert.equal(
     resolveDirectoryResumeTargetRoot({

application/state/sftp/dedicatedTransferResume.ts

@@ -1,3 +1,4 @@
+import { runTransferAndWaitForOwner } 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";
@@ -1000,6 +1001,11 @@ async function resumeDirectoryWithDedicatedSession(
   onProgress?: (progress: DedicatedResumeProgress) => void,
   options?: DedicatedResumeOptions,
 ): Promise<DedicatedResumeResult> {
+  // Fresh recovery authorizes only pauses already present at its entry. Keep
+  // row references so a newer cross-window pause invalidates that permission.
+  const pausedAtResume = new Map(sftpTransferCenterStore.getSnapshot().tasks
+    .filter((child) => child.parentTaskId === parent.id && child.status === "paused")
+    .map((child) => [child.id, child]));
   const bridge = netcattyBridge.get();
   if (!bridge?.startStreamTransfer) {
     return { success: false, error: "Transfer bridge unavailable" };
@@ -1247,8 +1253,13 @@ async function resumeDirectoryWithDedicatedSession(
               if (options?.shouldAbort?.()) throw new Error("Transfer cancelled");
               options?.onChildUpdate?.(childBase);
 
-              const streamResult = await bridge.startStreamTransfer!({
+              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,
@@ -1267,19 +1278,9 @@ async function resumeDirectoryWithDedicatedSession(
                 uploadCheckpointBytes: childBase.uploadCheckpointBytes,
                 sourceFingerprint: childBase.sourceFingerprint,
                 skipAdmission: true,
-              });
+              }), () => options?.shouldAbort?.() === true, pausedAtResume.get(childId));
 
-              if (streamResult?.superseded === true) {
-                for (;;) {
-                  if (options?.shouldAbort?.()) throw new Error("Transfer cancelled");
-                  const latest = sftpTransferCenterStore.getTask(childBase.id);
-                  const status = latest?.status;
-                  if (status === "completed") break;
-                  if (status === "failed") throw new Error(latest?.error || "Transfer failed");
-                  if (status === "cancelled") throw new Error("Transfer cancelled");
-                  await new Promise((resolve) => setTimeout(resolve, 200));
-                }
-              } else if (streamResult?.error || streamResult?.cancelled) {
+              if (streamResult?.error || streamResult?.cancelled) {
                 throw new Error(streamResult.error || "Transfer cancelled");
               }
 

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

@@ -233,3 +233,74 @@ test("live directory download rejects a Windows backslash traversal entry", asyn
     (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = previousActEnvironment;
   }
 });
+
+test("live folder settles a superseded child whose completed row was compacted", async () => {
+  const { sftpTransferCenterStore } = await import("../sftpTransferCenterStore");
+  const previousWindow = (globalThis as { window?: unknown }).window;
+  const previousLocalStorage = Object.getOwnPropertyDescriptor(globalThis, "localStorage");
+  const previousActEnvironment = (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT;
+  (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true;
+  Object.defineProperty(globalThis, "localStorage", {
+    configurable: true,
+    value: { getItem: () => null, setItem: () => {}, removeItem: () => {} },
+  });
+  const root = { ...rootTask(), id: "live-compacted-root", startTime: Date.now() };
+  let tasks: TransferTask[] = [root];
+  const transfersRef = { current: tasks };
+  const cancelledTasksRef = { current: new Set<string>() };
+  let childId: string | undefined;
+  (globalThis as { window?: unknown }).window = { netcatty: {
+    mkdirLocal: async () => undefined,
+    statLocal: async () => ({ type: "directory" }),
+    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,
+      });
+      return { superseded: true };
+    },
+  } };
+  let operations: ReturnType<typeof useSftpDirectoryTransferOps> | undefined;
+  let renderer: ReactTestRenderer | null = null;
+  let running: Promise<number> | undefined;
+  const Probe = () => {
+    operations = useSftpDirectoryTransferOps({
+      ownerId: "live-compacted-owner", cancelledTasksRef,
+      pausedTasksRef: { current: new Set() }, waitUntilTransferResumed: async () => undefined,
+      activeChildIdsRef: { current: new Map() }, transfersRef,
+      setTransfers: (update) => {
+        tasks = typeof update === "function" ? update(tasks) : update;
+        transfersRef.current = tasks;
+      },
+      listLocalFiles: async () => [], listRemoteFiles: async () => [fileEntry("file.txt")],
+    });
+    return null;
+  };
+  try {
+    await act(async () => { renderer = create(React.createElement(Probe)); });
+    assert.ok(operations);
+    running = operations.transferDirectory(root, "source-sftp", null, false, true, "auto", "auto", root.id);
+    const result = await Promise.race([
+      running,
+      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);
+    assert.notEqual(result, "still-waiting", "compacted completion must settle the live folder");
+    assert.equal(result, 0);
+  } finally {
+    cancelledTasksRef.current.add(root.id);
+    await running?.catch(() => {});
+    await act(async () => { renderer?.unmount(); });
+    sftpTransferCenterStore.patchTask(root.id, { status: "completed" });
+    sftpTransferCenterStore.dismiss(root.id);
+    (globalThis as { window?: unknown }).window = previousWindow;
+    if (previousLocalStorage) Object.defineProperty(globalThis, "localStorage", previousLocalStorage);
+    else Reflect.deleteProperty(globalThis, "localStorage");
+    (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = previousActEnvironment;
+  }
+});
+

application/state/sftp/transferDirectoryOps.ts

@@ -1,3 +1,4 @@
+import { runTransferAndWaitForOwner } from "./waitForTransferOwner";
 import { useCallback, type Dispatch, type MutableRefObject, type SetStateAction } from "react";
 import type { Host, SftpFileEntry, SftpFilenameEncoding, TransferStatus, TransferTask } from "../../../domain/models";
 import {
@@ -412,7 +413,11 @@ export function useSftpDirectoryTransferOps({
                 }
                 // Await the invoke result — cancel resolves with { error } and may
                 // not fire onComplete/onError after preload clears listeners.
-                const transferPromise = netcattyBridge.require().startStreamTransfer!(options);
+                const transferPromise = runTransferAndWaitForOwner(
+                  task,
+                  () => netcattyBridge.require().startStreamTransfer!(options),
+                  () => isCancelledLocalOrGlobal(cancelledTasksRef, rootTaskId, task.id),
+                );
                 // Streams that arm after the parent pause round never receive the
                 // initial pauseTransfer. Keep pausing while the folder is latched.
                 // Capture epoch per attempt so Resume (epoch bump) undoes a late pause.
@@ -450,33 +455,6 @@ export function useSftpDirectoryTransferOps({
                   watchPaused = false;
                   await pauseWatch.catch(() => {});
                 }
-                // Same-id retry stole ownership while OPEN was pending. The live
-                // owner still drives progress/completion events; this invoke must
-                // not mark the child completed or failed (Codex P2 on b17f64e9).
-                if (result?.superseded === true) {
-                  // Wait for the live same-id owner only — no fixed deadline so
-                  // long transfers are not failed while still running (Codex P2).
-                  for (;;) {
-                    if (
-                      cancelledTasksRef.current.has(task.id)
-                      || cancelledTasksRef.current.has(rootTaskId)
-                    ) {
-                      throw new Error("Transfer cancelled");
-                    }
-                    const latest = sftpTransferCenterStore.getTask(task.id)
-                      ?? transfersRef.current.find((candidate) => candidate.id === task.id);
-                    const status = latest?.status;
-                    if (status === "completed") break;
-                    if (status === "failed") {
-                      throw new Error(latest?.error || "Transfer failed");
-                    }
-                    if (status === "cancelled") {
-                      throw new Error("Transfer cancelled");
-                    }
-                    await new Promise((resolve) => setTimeout(resolve, 200));
-                  }
-                  break;
-                }
                 if (result?.error || result?.cancelled) {
                   throw new Error(result.error || "Transfer cancelled");
                 }

application/state/sftp/transferSettlementObservation.test.ts

@@ -0,0 +1,171 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import type { TransferTask } from "../../../domain/models";
+import { createSftpTransferCenterStore } from "../sftpTransferCenterStore";
+
+function child(): TransferTask {
+  return {
+    id: "child", parentTaskId: "root", directoryEntryIndex: 0, directoryEntryIdentity: "a".repeat(64),
+    sourcePath: "/source/a", targetPath: "/target/a", fileName: "a",
+    sourceConnectionId: "local", targetConnectionId: "remote", direction: "upload",
+    totalBytes: 1, transferredBytes: 0, speed: 0, startTime: 0, isDirectory: false, status: "transferring",
+  };
+}
+
+for (const status of ["completed", "failed", "cancelled"] as const) {
+  test(`settlement observation retains exact ${status} before history pruning`, () => {
+    const store = createSftpTransferCenterStore();
+    const task = child();
+    store.upsertTasks([{ ...task, id: "root", parentTaskId: undefined, isDirectory: true }, task]);
+    const observation = store.observeTaskSettlement(task);
+    store.patchTask(task.id, { status, error: status === "failed" ? "write failed" : undefined });
+    assert.equal(observation.read()?.status, status);
+    if (status === "completed") assert.equal(store.getTask(task.id), undefined);
+    observation.dispose();
+    assert.equal(observation.read(), undefined);
+  });
+}
+
+test("another file reusing an id and index is not completion evidence", () => {
+  const store = createSftpTransferCenterStore();
+  const task = child();
+  store.upsertTasks([task]);
+  const observation = store.observeTaskSettlement(task);
+  store.upsertTasks([{ ...task, directoryEntryIdentity: "b".repeat(64), status: "completed" }]);
+  assert.equal(observation.read(), undefined);
+  observation.dispose();
+});
+
+test("a disposed observer receives no later completion", () => {
+  const store = createSftpTransferCenterStore();
+  const task = child();
+  store.upsertTasks([task]);
+  const observation = store.observeTaskSettlement(task);
+  observation.dispose();
+  store.patchTask(task.id, { status: "completed" });
+  assert.equal(observation.read(), undefined);
+});
+
+test("explicit dispatch refreshes failed identity without overwriting checkpoints", () => {
+  const store = createSftpTransferCenterStore();
+  const task = { ...child(), status: "failed" as const, checkpointBytes: 7 };
+  store.upsertTasks([task]);
+  const next = { ...task, directoryEntryIdentity: "b".repeat(64), checkpointBytes: 0 };
+  assert.equal(store.admitTaskRun(next), "ready");
+  assert.equal(store.getTask(task.id)?.status, "transferring");
+  assert.equal(store.getTask(task.id)?.directoryEntryIdentity, next.directoryEntryIdentity);
+  assert.equal(store.getTask(task.id)?.checkpointBytes, 7);
+});
+
+for (const status of ["cancelled", "completed", "paused", "pausing"] as const) {
+  test(`dispatch cannot revive a retained ${status} row`, () => {
+    const store = createSftpTransferCenterStore();
+    const task = child();
+    store.upsertTasks([
+      { ...task, id: "root", parentTaskId: undefined },
+      { ...task, status },
+    ]);
+    assert.equal(store.admitTaskRun(task), status === "pausing" ? "paused" : status);
+    assert.equal(store.getTask(task.id)?.status, status);
+  });
+}
+
+test("dispatch preserves active lifecycle guards and avoids replacing unchanged rows", () => {
+  const store = createSftpTransferCenterStore();
+  const task = { ...child(), lifecycleEpoch: 9 };
+  store.upsertTasks([task]);
+  const before = store.getTask(task.id);
+  assert.equal(store.admitTaskRun(task), "ready");
+  assert.equal(store.getTask(task.id), before);
+  assert.equal(store.admitTaskRun({ ...task, directoryEntryIdentity: "b".repeat(64) }), "ready");
+  assert.equal(store.getTask(task.id)?.lifecycleEpoch, 9);
+});
+
+test("dispatch rejects a later pause or cancellation before changing identity", async (t) => {
+  const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
+  const { markTransferCancelledTree, settleTransferCancelTree } = await import("./transferCancelLatch");
+  const store = createSftpTransferCenterStore();
+  const task = { ...child(), status: "failed" as const };
+  store.upsertTasks([task]);
+  t.after(() => { resetTransferPauseLatchesForTests(); settleTransferCancelTree("root", [task.id]); });
+  latchTransferPause("root");
+  assert.equal(store.admitTaskRun(task), "paused");
+  resetTransferPauseLatchesForTests();
+  markTransferCancelledTree("root", [task.id]);
+  assert.equal(store.admitTaskRun(task), "cancelled");
+  assert.equal(store.getTask(task.id)?.status, "failed");
+});
+
+for (const nextStatus of ["transferring", "completed", "cancelled"] as const) {
+  test(`dispatch waits through a later pause until ${nextStatus}`, async (t) => {
+    const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+    const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+    const task = { ...child(), id: `paused-admission-${nextStatus}`, parentTaskId: undefined };
+    store.upsertTasks([{ ...task, status: "paused" }]);
+    let starts = 0;
+    let abort = false;
+    const running = runTransferAndWaitForOwner(task, async () => { starts += 1; return {}; }, () => abort);
+    // Attach rejection handling before the cancellation event is delivered.
+    const settled = running.then(() => "completed", (error: Error) => error.message);
+    t.after(async () => { abort = true; await settled; store.dismiss(task.id); });
+    await new Promise((resolve) => setTimeout(resolve, 30));
+    assert.equal(starts, 0);
+    assert.equal(store.getTask(task.id)?.status, "paused");
+    if (nextStatus !== "cancelled") store.patchTask(task.id, { status: "transferring", lifecycleEpoch: 1 });
+    if (nextStatus === "cancelled") {
+      const { markTransferCancelledTree, settleTransferCancelTree } = await import("./transferCancelLatch");
+      markTransferCancelledTree(task.id, []);
+      t.after(() => settleTransferCancelTree(task.id, []));
+    } else store.patchTask(task.id, { status: nextStatus });
+    const outcome = await Promise.race([settled, new Promise((resolve) => setTimeout(() => resolve("still-waiting"), 1000))]);
+    assert.equal(outcome, nextStatus === "cancelled" ? "Transfer cancelled" : "completed");
+    assert.equal(starts, nextStatus === "transferring" ? 1 : 0);
+  });
+}
+
+test("completed dispatch consumes only exact identity and never restarts", async (t) => {
+  const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+  const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+  const task = { ...child(), id: "completed-admission", parentTaskId: undefined };
+  store.upsertTasks([{ ...task, status: "completed" }]);
+  t.after(() => store.dismiss(task.id));
+  const start = async () => { assert.fail("completed transfer must not restart"); };
+  await runTransferAndWaitForOwner(task, start, () => false);
+  await assert.rejects(runTransferAndWaitForOwner({ ...task, directoryEntryIdentity: "b".repeat(64) }, start, () => false), /identity changed/);
+  assert.equal(store.getTask(task.id)?.status, "completed");
+});
+
+test("a resumed retry does not inherit failed settlement captured while admission was paused", async (t) => {
+  const { sftpTransferCenterStore: store } = await import("../sftpTransferCenterStore");
+  const { runTransferAndWaitForOwner } = await import("./waitForTransferOwner");
+  const { latchTransferPause, resetTransferPauseLatchesForTests } = await import("./transferPauseLatch");
+  const task = { ...child(), id: "failed-paused-admission", parentTaskId: undefined };
+  store.upsertTasks([{ ...task, status: "failed", error: "previous attempt failed" }]);
+  latchTransferPause(task.id);
+  let abort = false;
+  const running = runTransferAndWaitForOwner(task, async () => {
+    store.patchTask(task.id, { status: "completed" });
+    return { superseded: true };
+  }, () => abort);
+  const settled = running.then(() => "completed", (error: Error) => error.message);
+  t.after(async () => { abort = true; resetTransferPauseLatchesForTests(); await settled; store.dismiss(task.id); });
+  // An unrelated lifecycle publication captures the previous failed row while waiting.
+  store.upsertTasks([{ ...child(), id: "unrelated-admission", parentTaskId: undefined }]);
+  t.after(() => store.dismiss("unrelated-admission"));
+  resetTransferPauseLatchesForTests();
+  assert.equal(await Promise.race([settled, new Promise((resolve) => setTimeout(() => resolve("still-waiting"), 1000))]), "completed");
+});
+
+test("fresh directory recovery authorizes only an unchanged retained pause under an active parent", () => {
+  const store = createSftpTransferCenterStore();
+  const task = child();
+  store.upsertTasks([{ ...task, id: "root", parentTaskId: undefined, isDirectory: true, status: "pending" }, { ...task, status: "paused", lifecycleEpoch: 4 }]);
+  const paused = store.getTask(task.id)!;
+  assert.equal(store.admitTaskRun(task), "paused", "ordinary live dispatch must respect cross-window pause");
+  assert.equal(store.admitTaskRun(task, paused), "ready");
+  store.patchTask(task.id, { status: "paused", lifecycleEpoch: 5 });
+  assert.equal(store.admitTaskRun(task, paused), "paused", "newer child pause invalidates fresh recovery permission");
+  const currentPause = store.getTask(task.id)!;
+  store.patchTask("root", { status: "paused", lifecycleEpoch: 6 });
+  assert.equal(store.admitTaskRun(task, currentPause), "paused", "parent pause still wins");
+});

application/state/sftp/waitForTransferOwner.ts

@@ -0,0 +1,41 @@
+import type { TransferTask } from "../../../domain/models";
+import { sftpTransferCenterStore } from "../sftpTransferCenterStore";
+
+type StreamResult = { error?: string; cancelled?: boolean; superseded?: boolean } | undefined;
+
+/** A live waiter owns bounded settlement evidence; persisted history stays compact. */
+export async function runTransferAndWaitForOwner(
+  task: TransferTask,
+  start: () => Promise<StreamResult>,
+  shouldAbort: () => boolean,
+  pausedAtResume?: TransferTask,
+): Promise<StreamResult> {
+  // Register before admission/start: an owner may finish while dispatch waits for resume.
+  let observation = sftpTransferCenterStore.observeTaskSettlement(task);
+  try {
+    for (;;) {
+      if (shouldAbort()) throw new Error("Transfer cancelled");
+      const admission = sftpTransferCenterStore.admitTaskRun(task, pausedAtResume);
+      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 === "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);
+    const result = await start();
+    if (!result?.superseded) return result;
+    for (;;) {
+      if (shouldAbort()) throw new Error("Transfer cancelled");
+      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));
+    }
+  } finally {
+    observation.dispose();
+  }
+}

application/state/sftpTransferCenterStore.ts

@@ -28,6 +28,7 @@ import {
 } from "./sftp/transferControlEpoch";
 import { isTransferWalkInFlight } from "./sftp/transferWalkRegistry";
 import {
+  isTransferOrRootCancelled,
   markTransferCancelledTree,
   settleTransferCancelTree,
 } from "./sftp/transferCancelLatch";
@@ -96,6 +97,10 @@ export interface SftpTransferCenterStore {
   getBadgeSnapshot(): { count: number; hasAttention: boolean };
   /** 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 };
+  /** Publish an explicit dispatch identity without flushing large child-history batches. */
+  admitTaskRun(task: TransferTask, pausedAtResume?: TransferTask): "ready" | "paused" | "completed" | "cancelled" | "conflict";
   getOwnerTasks(ownerId: string): TransferTask[];
   publishOwner(ownerId: string, tasks: readonly TransferTask[]): void;
   registerOwner(ownerId: string, controls: SftpTransferOwnerControls): () => void;
@@ -398,6 +403,15 @@ 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 matchesObservedTask = (expected: TransferTask, candidate: TransferTask) => (
+    expected.id === candidate.id
+    && expected.sourcePath === candidate.sourcePath
+    && expected.targetPath === candidate.targetPath
+    && expected.parentTaskId === candidate.parentTaskId
+    && expected.directoryEntryIndex === candidate.directoryEntryIndex
+    && expected.directoryEntryIdentity === candidate.directoryEntryIdentity
+  );
   const refreshBadgeSnapshot = () => {
     const next = buildBadgeSnapshot(tasks);
     if (
@@ -609,6 +623,14 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
     // live renderer walk keeps its controls until TransferRuntime.runWalk
     // 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;
+        }
+      }
+    }
     const beforePrune = tasks;
     tasks = pruneSftpTransferHistory(tasks);
     const retainedIds = new Set(tasks.map((task) => task.id));
@@ -1307,6 +1329,70 @@ export function createSftpTransferCenterStore(persistence?: StorePersistence): S
     },
     getSnapshot: () => ensureSnapshot(),
     getBadgeSnapshot: () => badgeSnapshot,
+    admitTaskRun(incoming, pausedAtResume) {
+      const rootId = incoming.parentTaskId ?? incoming.id;
+      if (isTransferOrRootCancelled(rootId, incoming.id)) return "cancelled";
+      const parent = incoming.parentTaskId ? tasks.find((task) => task.id === rootId) : undefined;
+      if (parent?.status === "cancelled") return "cancelled";
+      const index = tasks.findIndex((task) => task.id === incoming.id);
+      const existing = index < 0 ? undefined : tasks[index];
+      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") {
+        return matchesObservedTask(incoming, existing) ? "completed" : "conflict";
+      }
+      const resumesUnchangedPause = existing === pausedAtResume
+        && existing?.status === "paused"
+        && (parent?.status === "pending" || parent?.status === "transferring");
+      if (isTransferOrRootPauseLatched(rootId, incoming.id)
+        || parent?.status === "paused" || parent?.status === "pausing"
+        || (existing?.status === "paused" && !resumesUnchangedPause) || existing?.status === "pausing") return "paused";
+      if (parent?.status === "completed") return "conflict";
+      if (!existing) return "ready";
+      if (existing.status === "transferring"
+        && existing.directoryEntryIndex === incoming.directoryEntryIndex
+        && existing.directoryEntryIdentity === incoming.directoryEntryIdentity) return "ready";
+      // Explicit retry may leave failed/interrupted and replace its manifest identity.
+      // Do not overwrite byte checkpoints with a delayed renderer snapshot.
+      const next = tasks.slice();
+      next[index] = {
+        ...existing,
+        directoryEntryIndex: incoming.directoryEntryIndex,
+        directoryEntryIdentity: incoming.directoryEntryIdentity,
+        ...(existing.status === "transferring" ? {} : {
+          status: "transferring" as const,
+          lifecycleEpoch: undefined,
+          error: undefined,
+          endTime: undefined,
+          reconnectRequired: false,
+        }),
+      };
+      tasks = next;
+      emitProgress(false);
+      return "ready";
+    },
+    observeTaskSettlement(expected) {
+      const observer: { expected: TransferTask; settled?: TransferTask } = { expected: { ...expected } };
+      const observers = settlementObservers.get(expected.id) ?? new Set();
+      observers.add(observer);
+      settlementObservers.set(expected.id, observers);
+      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;
+        },
+        dispose() {
+          disposed = true;
+          observer.settled = undefined;
+          observers.delete(observer);
+          if (observers.size === 0) settlementObservers.delete(expected.id);
+        },
+      };
+    },
     getTask(taskId) {
       return tasks.find((task) => task.id === taskId);
     },

docs/research/sftp-folder-settlement.md

@@ -0,0 +1,24 @@
+# Folder completion after same-ID ownership handoff
+
+Both normal directory transfer and dedicated directory recovery can receive a
+superseded stream result: another invocation now owns the same child transfer.
+The old caller must wait for that owner rather than report premature completion.
+Previously it polled the visible child row. Completed rows are compacted into
+parent checkpoints, so a completion arriving before the superseded reply removes
+the row first. Polling then waits forever; a stale panel row can mask this further.
+
+Two regressions use the actual React directory hook / dedicated recovery entrypoint
+and actual store compaction. Each records one completed file in the parent while
+the transfer operation remains pending on the baseline.
+
+The fix registers a bounded settlement observation before starting the stream.
+The store captures terminal state for exact observed file identities before
+history compaction. One shared helper waits for the actual owner in both paths,
+then releases its observation on success, failure or cancellation. There is no
+persistent per-file tombstone list and no inference that a missing row means
+success. Reused IDs with different indexed file identities are not evidence.
+
+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.