Verify existing migration files before conditional uploads

packages/app/src/agents/mutation-coordinator.ts

@@ -16,6 +16,7 @@ import {
 import { createAgentHistoryPorter, createProjectHistoryLifecycle } from "../lib/agent-porter";
 import {
   createSiteStudioBoundaryContext,
+  emitDiagnostic,
   type SiteStudioLoggingContextData,
 } from "../lib/logging";
 import {
@@ -103,13 +104,40 @@ export class MutationCoordinator extends DurableObject<Env> {
     subject: string,
   ): Promise<MigrationResult> {
     return this.mutations.run(async () => {
-      await new OwnerMutationService(
-        this.env.SITE_STUDIO_BUCKET,
-        this.ctx.storage,
-        undefined,
-        createProjectHistoryLifecycle(this.env),
-      ).recover(anonUserId);
-      const { manifest } = await loadRecoveryManifest(this.env.SITE_STUDIO_BUCKET, recoveryId);
+      const logging = this.env.CAIL_LOG_ENV
+        ? createSiteStudioBoundaryContext(this.env)
+        : undefined;
+      try {
+        await new OwnerMutationService(
+          this.env.SITE_STUDIO_BUCKET,
+          this.ctx.storage,
+          undefined,
+          createProjectHistoryLifecycle(this.env),
+        ).recover(anonUserId);
+      } catch (error) {
+        emitDiagnostic(
+          "error",
+          "legacy_recovery_source_journal_failed",
+          {},
+          logging,
+        );
+        throw error;
+      }
+      let manifest: Awaited<ReturnType<typeof loadRecoveryManifest>>["manifest"];
+      try {
+        ({ manifest } = await loadRecoveryManifest(
+          this.env.SITE_STUDIO_BUCKET,
+          recoveryId,
+        ));
+      } catch (error) {
+        emitDiagnostic(
+          "error",
+          "legacy_recovery_manifest_load_failed",
+          {},
+          logging,
+        );
+        throw error;
+      }
       if (manifest.source.owner !== anonUserId) throw new LegacyRecoveryError("conflict");
       // The first source-queue entry must revalidate immediately before copy.
       // A later entry with a migration claim is a resume: source deletion may
@@ -119,14 +147,25 @@ export class MutationCoordinator extends DurableObject<Env> {
         migrationClaimKey(anonUserId),
         "json",
       );
-      await validateStagedRecovery(this.env.SITE_STUDIO_BUCKET, manifest, {
-        allowMissing: Boolean(migrationClaim),
-      });
+      try {
+        await validateStagedRecovery(this.env.SITE_STUDIO_BUCKET, manifest, {
+          allowMissing: Boolean(migrationClaim),
+        });
+      } catch (error) {
+        emitDiagnostic(
+          "error",
+          "legacy_recovery_source_validation_failed",
+          {},
+          logging,
+        );
+        throw error;
+      }
       return migrateAnonymousData({
         bucket: this.env.SITE_STUDIO_BUCKET,
         kv: this.env.SESSION_KV,
         anonUserId,
         subject,
+        logging,
       });
     });
   }

packages/app/src/lib/migration.test.ts

@@ -289,6 +289,48 @@ describe("migrateAnonymousData", () => {
     expect(kv.store.has(migrationPendingKey(SUBJECT))).toBe(false);
   });
 
+  it("validates an already copied file without attempting another upload", async () => {
+    const content = "<h1>large preserved source</h1>";
+    seedAnonProject(bucket, "portfolio", {}, content);
+    // SAFETY: seedAnonProject serializes this value from metadataFor's ProjectMetadata fixture.
+    const sourceMetadata = JSON.parse(
+      bucket.store.get(`projects/${ANON}/portfolio/.metadata.json`)!.data,
+    ) as ProjectMetadata;
+    bucket.store.set(`projects/${SUBJECT}/portfolio/.metadata.json`, {
+      data: JSON.stringify({
+        ...sourceMetadata,
+        importedFrom: ANON,
+        importedOriginalId: "portfolio",
+      }),
+    });
+    const destination = `projects/${SUBJECT}/portfolio/index.html`;
+    bucket.store.set(destination, { data: content });
+    kv.store.set(
+      migrationClaimKey(ANON),
+      JSON.stringify({
+        subject: SUBJECT,
+        status: "pending",
+        startedAt: "2026-01-01T00:00:00.000Z",
+      } satisfies MigrationClaim),
+    );
+
+    const originalPut = bucket.put;
+    const destinationPuts: string[] = [];
+    bucket.put = vi.fn(
+      async (key: string, data: string, options?: R2PutOptions) => {
+        if (key === destination) destinationPuts.push(key);
+        return originalPut(key, data, options);
+      },
+    );
+
+    await expect(run()).resolves.toMatchObject({
+      status: "migrated",
+      projects: { portfolio: "portfolio" },
+    });
+    expect(destinationPuts).toEqual([]);
+    expect(textOf(bucket.store.get(destination))).toBe(content);
+  });
+
   it("is idempotent: a second run is a no-op that changes nothing", async () => {
     seedAnonProject(bucket, "portfolio");
     await run();

packages/app/src/lib/migration.ts

@@ -166,6 +166,22 @@ async function copyIfAbsent(
   const object = await bucket.get(fromKey);
   if (!object) return; // source vanished (concurrent run finished it) — fine
   const sourceBytes = await object.arrayBuffer();
+
+  // A retry or the defensive second sweep usually finds the destination
+  // already present. Compare that object before attempting a conditional PUT:
+  // the R2 transport sends the entire request body before it handles a failed
+  // precondition, so needlessly re-uploading a large existing object can fail
+  // before the binding returns the documented null result.
+  const existing = await bucket.get(toKey);
+  if (existing) {
+    if (!bytesEqual(sourceBytes, await existing.arrayBuffer())) {
+      throw new Error(
+        "Anonymous-data migration stopped because the destination changed concurrently.",
+      );
+    }
+    return;
+  }
+
   const wrote = await bucket.put(toKey, sourceBytes, {
     httpMetadata: object.httpMetadata,
     onlyIf: { etagDoesNotMatch: "*" },
@@ -183,6 +199,30 @@ async function copyIfAbsent(
   }
 }
 
+const migrationFailureType = {
+  inventory: "migration_inventory_failed",
+  firstCopy: "migration_first_copy_failed",
+  secondCopy: "migration_second_copy_failed",
+  handle: "migration_handle_failed",
+  sourceRetirement: "migration_source_retirement_failed",
+  completion: "migration_completion_failed",
+} as const;
+
+async function runMigrationStage<T>(
+  stage: keyof typeof migrationFailureType,
+  logging: SiteStudioLoggingContext | undefined,
+  operation: () => Promise<T>,
+): Promise<T> {
+  try {
+    return await operation();
+  } catch (error) {
+    // Static stage categories only: never project an exception message, key,
+    // owner, subject, or project identifier into operational logs.
+    emitDiagnostic("error", migrationFailureType[stage], {}, logging);
+    throw error;
+  }
+}
+
 async function putJsonIfAbsentOrEqual(
   bucket: R2Bucket,
   key: string,
@@ -611,64 +651,84 @@ export async function migrateAnonymousData(options: {
   };
 
   // ---- Inventory ----
-  const anonProjectIds = await listProjectIds(bucket, anonUserId);
-  const uploadKeys = await listKeys(bucket, uploadsPrefix(anonUserId));
+  const { anonProjectIds, uploadKeys } = await runMigrationStage(
+    "inventory",
+    logging,
+    async () => ({
+      anonProjectIds: await listProjectIds(bucket, anonUserId),
+      uploadKeys: await listKeys(bucket, uploadsPrefix(anonUserId)),
+    }),
+  );
 
   if (anonProjectIds.length === 0 && uploadKeys.length === 0) {
     // There are no project chats to port, so the planned handle can be
     // committed before retiring the completed anonymous session as well.
-    await migrateHandle({ bucket, anonUserId, subject, now });
-    return finish("nothing-to-migrate", {});
+    await runMigrationStage("handle", logging, () =>
+      migrateHandle({ bucket, anonUserId, subject, now })
+    );
+    return runMigrationStage("completion", logging, () =>
+      finish("nothing-to-migrate", {})
+    );
   }
 
   // ---- First copy sweep ----
-  const firstSweep = await copyAnonymousNamespace({
-    bucket,
-    anonUserId,
-    subject,
-    porter,
-    knownProjects: {},
-    knownSlugs: {},
-    logging,
-  });
+  const firstSweep = await runMigrationStage("firstCopy", logging, () =>
+    copyAnonymousNamespace({
+      bucket,
+      anonUserId,
+      subject,
+      porter,
+      knownProjects: {},
+      knownSlugs: {},
+      logging,
+    })
+  );
 
   // ---- Second copy sweep, immediately before delete ----
   // The owner-scoped MutationCoordinator prevents adopted anonymous writes
   // during this run. Re-list anyway so an older deployment or out-of-band
   // bucket writer cannot be deleted merely because it appeared after the first
   // inventory. Conditional copy-or-equal keeps this sweep idempotent.
-  const { projectMap } = await copyAnonymousNamespace({
-    bucket,
-    anonUserId,
-    subject,
-    porter,
-    knownProjects: firstSweep.projectMap,
-    knownSlugs: firstSweep.slugMap,
-    logging,
-  });
+  const { projectMap } = await runMigrationStage("secondCopy", logging, () =>
+    copyAnonymousNamespace({
+      bucket,
+      anonUserId,
+      subject,
+      porter,
+      knownProjects: firstSweep.projectMap,
+      knownSlugs: firstSweep.slugMap,
+      logging,
+    })
+  );
 
   // All project chat ports have now completed. Re-home the handle only after
   // that last success and before deleting
   // the anonymous source. A handle failure therefore also leaves the source
   // and pending claim available for a later retry.
-  await migrateHandle({ bucket, anonUserId, subject, now });
+  await runMigrationStage("handle", logging, () =>
+    migrateHandle({ bucket, anonUserId, subject, now })
+  );
 
   // ---- Delete originals ----
   // Retire secondary artifacts first. If a snapshot/upload delete fails, the
   // project inventory remains discoverable as the retry anchor instead of
   // leaving an orphaned secondary namespace that a later completed marker can
   // no longer recover.
-  for (const key of await listKeys(bucket, snapshotUserPrefix(anonUserId))) {
-    await bucket.delete(key);
-  }
-  for (const key of await listKeys(bucket, uploadsPrefix(anonUserId))) {
-    await bucket.delete(key);
-  }
-  for (const key of await listKeys(bucket, projectPrefix(anonUserId))) {
-    await bucket.delete(key);
-  }
+  await runMigrationStage("sourceRetirement", logging, async () => {
+    for (const key of await listKeys(bucket, snapshotUserPrefix(anonUserId))) {
+      await bucket.delete(key);
+    }
+    for (const key of await listKeys(bucket, uploadsPrefix(anonUserId))) {
+      await bucket.delete(key);
+    }
+    for (const key of await listKeys(bucket, projectPrefix(anonUserId))) {
+      await bucket.delete(key);
+    }
+  });
 
-  return finish("migrated", projectMap);
+  return runMigrationStage("completion", logging, () =>
+    finish("migrated", projectMap)
+  );
 }
 
 /**