fix(sftp): wait for pending write OPEN before fastPut fallback

Expose pendingWriteOpenPathGate on the transfer so non-resumable concurrent
isolated failures wait for their own truncating OPEN to settle before
fastPut rewrites the same stage.

application/app/useAppStartupEffects.ts

@@ -409,6 +409,9 @@ export function useAppStartupEffects(ctx: StartupEffectsContext) {
               targetType: task.direction === "download" ? "local" : "sftp",
               sourceSftpId: task.direction === "download" ? sftpId : undefined,
               targetSftpId: task.direction === "upload" ? sftpId : undefined,
+              // Keep host-scoped path gates across session reopen (Codex P1).
+              sourceHostId: task.sourceHostId,
+              targetHostId: task.targetHostId,
               totalBytes: task.totalBytes,
               resumable: task.resumable !== false,
               checkpointBytes,
@@ -420,7 +423,30 @@ export function useAppStartupEffects(ctx: StartupEffectsContext) {
             });
           },
         );
-        if (result?.cancelled || result?.error === "Transfer cancelled") {
+        // Same-id retry stole ownership; wait for the live owner's terminal
+        // status instead of treating this invoke as completed (Codex P2).
+        if (result?.superseded === true) {
+          // Wait for live owner terminal status only (no fixed deadline).
+          for (;;) {
+            const latest = sftpTransferCenterStore.getSnapshot().tasks.find((candidate) => candidate.id === task.id);
+            const status = latest?.status;
+            if (status === "completed" || status === "cancelled" || status === "failed") {
+              if (status === "failed") {
+                throw new Error(latest?.error || "Transfer failed");
+              }
+              if (status === "cancelled") {
+                sftpTransferCenterStore.ingestBackgroundEvent({
+                  type: "cancelled",
+                  transferId: task.id,
+                  endedAt: Date.now(),
+                });
+              }
+              // completed: events already applied; cancelled handled above.
+              break;
+            }
+            await new Promise((resolve) => setTimeout(resolve, 200));
+          }
+        } else if (result?.cancelled || result?.error === "Transfer cancelled") {
           sftpTransferCenterStore.ingestBackgroundEvent({ type: "cancelled", transferId: task.id, endedAt: Date.now() });
         } else if (result?.error) {
           throw new Error(result.error);

application/state/sftp/dedicatedTransferResume.ts

@@ -29,6 +29,7 @@ import {
   isMissingDirectoryReplacePathError,
   promoteDirectoryReplaceStage as promoteDirectoryReplacePaths,
 } from "./directoryReplacePromotion";
+import { sftpTransferCenterStore } from "../sftpTransferCenterStore";
 
 export interface DedicatedResumeDeps {
   hosts: readonly Host[];
@@ -708,8 +709,19 @@ async function resumeSingleFileWithDedicatedSession(
             skipAdmission: true,
           });
 
-          if (streamResult?.error) {
-            throw new Error(streamResult.error);
+          if (streamResult?.superseded === true) {
+            // Live same-id owner still running; wait for terminal events only.
+            for (;;) {
+              if (shouldAbort?.()) throw new Error("Transfer cancelled");
+              const latest = sftpTransferCenterStore.getTask(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));
+            }
+          } else if (streamResult?.error || streamResult?.cancelled) {
+            throw new Error(streamResult.error || "Transfer cancelled");
           }
         }, { retries: 1, delayMs: 600 });
         return { transferId: task.id };
@@ -1258,7 +1270,17 @@ async function resumeDirectoryWithDedicatedSession(
                 skipAdmission: true,
               });
 
-              if (streamResult?.error || streamResult?.cancelled) {
+              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) {
                 throw new Error(streamResult.error || "Transfer cancelled");
               }
 

application/state/sftp/transferDirectoryOps.ts

@@ -357,13 +357,40 @@ export function useSftpDirectoryTransferOps({
                     await new Promise((resolve) => setTimeout(resolve, 80));
                   }
                 })();
-                let result: { error?: string; cancelled?: boolean } | undefined;
+                let result: { error?: string; cancelled?: boolean; superseded?: boolean } | undefined;
                 try {
                   result = await transferPromise;
                 } finally {
                   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");
                 }