diff --git a/src/migrate/direct-upload.ts b/src/migrate/direct-upload.ts
new file mode 100644
index 00000000..6f88617e
--- /dev/null
+++ b/src/migrate/direct-upload.ts
@@ -0,0 +1,641 @@
+/**
+ * Direct upload: stream locally packed CARs straight to storage providers and
+ * batch the on-chain adds. Nothing here requires inbound connectivity.
+ *
+ * Per built sub-piece CAR:
+ *   1. store() the bytes on the primary provider: the piece is now "parked"
+ *      and Curio's GC clock starts.
+ *   2. Each secondary pulls the piece from the primary's retrieval URL
+ *      (provider-to-provider; the client uploads once).
+ *   3. Parked pieces accumulate per provider and are flushed through one
+ *      commit() (addPieces) when the batch fills, the GC-window guess nears
+ *      expiry, or the source drains: see gc-window.ts for the scheduling
+ *      rules and why every tie breaks toward flushing early.
+ *
+ * Gas is the provider's cost (the provider submits addPieces with the
+ * client's EIP-712 authorisation), so this loop has no base-fee gate:
+ * pausing would save the provider gas while running the client's parked
+ * pieces into GC.
+ */
+
+import { createReadStream } from 'node:fs'
+import { unlink } from 'node:fs/promises'
+import { Readable } from 'node:stream'
+import { FindPieceError } from '@filoz/synapse-core/errors'
+import { findPiece } from '@filoz/synapse-core/sp'
+import type { Synapse } from '@filoz/synapse-sdk'
+import { CID } from 'multiformats/cid'
+import { formatFileSize } from '../utils/cli-helpers.js'
+import type { MigrationDB } from './db.js'
+import {
+  collectedCidFromError,
+  DEFAULT_ASSUMED_WINDOW_MS,
+  lowerWindowOnGc,
+  MAX_ADD_PIECES_BATCH,
+  marginFromConfirmations,
+  shouldFlush,
+} from './gc-window.js'
+import { formatDuration, Timer } from './metrics.js'
+import { type AddPiecesEvent, dataSetPieceId, fetchAddPiecesEvent, txLanded } from './pdp-verifier.js'
+import { log } from '../utils/cli-logger.js'
+
+export interface DirectUploadOptions {
+  /** Initialized Synapse instance (auth, chain, and transport already resolved). */
+  synapse: Synapse
+  /** Number of provider copies (contexts). Default 2: primary + one secondary. */
+  copies?: number | undefined
+  /** Pin specific providers instead of SDK selection. */
+  providerIds?: bigint[] | undefined
+  /** Reuse existing data sets instead of creating new ones. */
+  dataSetIds?: bigint[] | undefined
+  /** Starting GC-window guess; persisted per-provider lowering still applies. */
+  assumedWindowMs?: number | undefined
+  /** Data-set metadata applied when creating or matching contexts. */
+  dataSetMetadata?: Record<string, string> | undefined
+  /** Route retrieval through the FilBeam egress CDN. */
+  withCDN?: boolean | undefined
+  /**
+   * Pipeline hook: called when no sub-piece is currently pending. Resolves
+   * true when new sub-pieces may have appeared (re-poll), false when the
+   * source is drained. Absent means the source was fully staged up-front.
+   */
+  waitForMore?: (() => Promise<boolean>) | undefined
+  /**
+   * Pipeline hook: when it returns true while the upload queue is empty,
+   * parked pieces are flushed immediately instead of waiting for a full
+   * batch or the GC-window timer. The disk-budget gate uses this so blocked
+   * downloads are never left waiting on a commit the batcher sees no reason
+   * to hurry.
+   */
+  forceFlush?: (() => boolean) | undefined
+}
+
+/** The storage-context surface the loop drives; narrowed for fakes in tests. */
+export interface UploadContextLike {
+  providerId: string
+  serviceURL: string
+  dataSetId: string | null
+  store(
+    data: ReadableStream | Uint8Array,
+    options: { pieceCid?: unknown; onProgress?: (bytes: number) => void }
+  ): Promise<{ pieceCid: unknown; size: number }>
+  /** EIP-712 authorization for pulls/commits of these pieces on this provider. */
+  presignForCommit(pieces: Array<{ pieceCid: unknown }>): Promise<unknown>
+  pull(options: {
+    pieces: unknown[]
+    /**
+     * Pull source. MUST be the per-piece URL function form: the SDK treats a
+     * string as a service-URL base and appends its own path, which mangles an
+     * already-complete piece URL into a source the provider cannot fetch.
+     */
+    from: (pieceCid: unknown) => string
+    extraData?: unknown
+  }): Promise<{
+    status: 'complete' | 'failed'
+    pieces: Array<{ pieceCid: unknown; status: 'complete' | 'failed' }>
+  }>
+  commit(options: {
+    pieces: Array<{ pieceCid: unknown; pieceMetadata?: Record<string, string> }>
+    onSubmitted?: (txHash: string) => void
+  }): Promise<{
+    txHash: string
+    pieceIds: bigint[]
+    dataSetId: bigint
+  }>
+  getPieceUrl(pieceCid: unknown): string
+  /** Probe whether a parked piece is still present (post-GC re-verify). */
+  hasPiece(pieceCid: unknown): Promise<boolean>
+}
+
+export interface DirectUploadDeps {
+  setup(opts: DirectUploadOptions): Promise<{ contexts: UploadContextLike[] }>
+  /** Injectable clock so tests can drive the window timer. */
+  now(): number
+  /** Open a built CAR for streaming. Injectable so tests skip the filesystem. */
+  openCar(path: string): ReadableStream | Uint8Array
+  evictCar(path: string): Promise<void>
+  /** Whether a transaction landed successfully on chain (receipt status 1). */
+  txLanded(synapse: Synapse, txHash: string): Promise<boolean>
+  /** Canonical on-chain witness for a landed addPieces (see pdp-verifier). */
+  fetchAddPiecesEvent(synapse: Synapse, txHash: string, dataSetId: string | null): Promise<AddPiecesEvent | null>
+  /** On-chain piece id lookup for a data set, or null when absent (see pdp-verifier). */
+  dataSetPieceId(synapse: Synapse, dataSetId: string, pieceCid: string): Promise<string | null>
+}
+
+export const defaultDirectUploadDeps: DirectUploadDeps = {
+  async setup(opts) {
+    const contexts = await opts.synapse.storage.createContexts({
+      copies: opts.copies ?? 2,
+      ...(opts.providerIds == null ? {} : { providerIds: opts.providerIds }),
+      ...(opts.dataSetIds == null ? {} : { dataSetIds: opts.dataSetIds }),
+      // When targeting existing data sets by ID, metadata is not used for
+      // matching: pass it only for creation-path selection so an existing
+      // set with different metadata stays reachable.
+      ...(opts.dataSetIds == null && opts.dataSetMetadata != null ? { metadata: opts.dataSetMetadata } : {}),
+      ...(opts.withCDN === true ? { withCDN: true } : {}),
+    })
+    if (contexts.length === 0) throw new Error('no storage contexts resolved')
+    return {
+      contexts: contexts.map((ctx): UploadContextLike => {
+        const serviceURL = ctx.provider.pdp.serviceURL
+        return {
+          providerId: String(ctx.provider.id),
+          serviceURL,
+          dataSetId: ctx.dataSetId == null ? null : String(ctx.dataSetId),
+          store: (data, options) => ctx.store(data as never, options as never),
+          presignForCommit: (pieces) => ctx.presignForCommit(pieces as never),
+          pull: (options) => ctx.pull(options as never),
+          commit: (options) => ctx.commit(options as never),
+          getPieceUrl: (pieceCid) => ctx.getPieceUrl(pieceCid as never),
+          hasPiece: async (pieceCid) => {
+            try {
+              await findPiece({ serviceURL, pieceCid: pieceCid as never, retryCount: 0 })
+              return true
+            } catch (err) {
+              // False feeds "collected" transitions that trigger a full
+              // re-store, so only the SP's own not-found answer counts.
+              // Timeouts and transport errors propagate and abort the pass.
+              if (err instanceof FindPieceError && !err.message.includes('Timeout')) return false
+              throw err
+            }
+          },
+        }
+      }),
+    }
+  },
+  now: () => Date.now(),
+  openCar: (path) => Readable.toWeb(createReadStream(path)) as ReadableStream,
+  evictCar: async (path) => {
+    try {
+      await unlink(path)
+    } catch (err) {
+      // A resumed run may find the CAR already evicted by a prior run.
+      if ((err as NodeJS.ErrnoException).code !== 'ENOENT') {
+        log.message(`warn: failed to evict cached CAR ${path}: ${(err as Error).message}`)
+      }
+    }
+  },
+  txLanded,
+  fetchAddPiecesEvent,
+  dataSetPieceId,
+}
+
+export interface DirectUploadSummary {
+  network: string
+  providers: Array<{
+    providerId: string
+    role: 'primary' | 'secondary'
+    dataSetId: string | null
+    committed: number
+    collected: number
+    failed: number
+    /**
+     * Pieces whose addPieces outcome is unknown (the attempt was made but
+     * never confirmed either way). Anything non-zero means the run is
+     * incomplete: resolve on chain before retrying.
+     */
+    addUnconfirmed: number
+    flushes: number
+    assumedWindowMs: number
+    /** Distinct addPieces transaction hashes behind the committed pieces. */
+    txHashes: string[]
+  }>
+  storedBytes: number
+  evictedCars: number
+}
+
+export async function runDirectUpload(
+  db: MigrationDB,
+  opts: DirectUploadOptions,
+  deps: DirectUploadDeps = defaultDirectUploadDeps
+): Promise<DirectUploadSummary> {
+  const { synapse } = opts
+  const { contexts } = await deps.setup(opts)
+  const [primary, ...secondaries] = contexts
+  if (primary == null) throw new Error('no primary storage context')
+
+  log.message(
+    `direct upload to ${contexts.length} provider(s): ` +
+      contexts.map((c, i) => `${i === 0 ? 'primary' : 'secondary'} ${c.providerId} (${c.serviceURL})`).join(', ')
+  )
+
+  const observedCommitMs: number[] = []
+  const flushCounts = new Map<string, number>()
+  const runTimer = new Timer()
+  let storedBytes = 0
+
+  const windowFor = (ctx: UploadContextLike): number =>
+    db.providerWindowMs(ctx.providerId, opts.assumedWindowMs ?? DEFAULT_ASSUMED_WINDOW_MS)
+
+  const flush = async (ctx: UploadContextLike, reason: string): Promise<void> => {
+    const batch = db.parkedUploads(ctx.providerId).slice(0, MAX_ADD_PIECES_BATCH)
+    if (batch.length === 0) return
+    flushCounts.set(ctx.providerId, (flushCounts.get(ctx.providerId) ?? 0) + 1)
+    const cids = batch.map((b) => b.subPieceCid)
+    log.message(`flush [${reason}] provider ${ctx.providerId}: committing ${batch.length} piece(s)`)
+    // Durable breadcrumb before the attempt: a crash mid-commit must never be
+    // auto-resolved into a blind re-add.
+    db.markUploadsAddUnconfirmed(cids, ctx.providerId)
+    const commitTimer = new Timer()
+    try {
+      const result = await ctx.commit({
+        pieces: cids.map((cid) => ({ pieceCid: CID.parse(cid) })),
+        onSubmitted: (txHash) => db.markUploadTxSubmitted(cids, ctx.providerId, txHash),
+      })
+      observedCommitMs.push(commitTimer.stop())
+      batch.forEach((b, i) => {
+        db.markUploadCommitted(b.subPieceCid, ctx.providerId, {
+          dataSetId: String(result.dataSetId),
+          pieceId: String(result.pieceIds[i] ?? ''),
+          txHash: result.txHash,
+        })
+      })
+      // A context created without a data set learns its id at the first
+      // commit; recording it here means later parks and resumes carry the
+      // set id on the row instead of depending on context re-resolution.
+      ctx.dataSetId = String(result.dataSetId)
+      log.message(`committed ${batch.length} piece(s) on provider ${ctx.providerId} (data set ${result.dataSetId})`)
+    } catch (err) {
+      const message = (err as Error).message ?? String(err)
+      const gcCid = collectedCidFromError(message)
+      if (gcCid == null) {
+        // Not a GC rejection: leave the batch in add_unconfirmed for the
+        // resume reconciliation: a blind retry could double-add on chain.
+        log.message(`error: commit failed on provider ${ctx.providerId} (batch left add_unconfirmed): ${message}`)
+        return
+      }
+      // Curio rejected the batch because a parked piece is gone. The batch is
+      // atomic and pre-chain, so nothing landed. Lower the window from the
+      // collected piece's parked age, then re-verify every batch member:
+      // Curio reports only the FIRST miss.
+      const collected = batch.find((b) => b.subPieceCid === gcCid)
+      if (collected == null) {
+        log.message(`warn: provider ${ctx.providerId} rejected unknown sub-piece ${gcCid}; re-verifying batch`)
+      } else {
+        const age = deps.now() - Date.parse(collected.parkedAt)
+        const lowered = lowerWindowOnGc(windowFor(ctx), age)
+        db.lowerProviderWindow(ctx.providerId, lowered)
+        log.message(
+          `GC detected on provider ${ctx.providerId}: ${gcCid} collected after ${formatDuration(age)} parked; ` +
+            `window lowered to ${formatDuration(lowered)}`
+        )
+      }
+      for (const b of batch) {
+        const present = b.subPieceCid !== gcCid && (await ctx.hasPiece(CID.parse(b.subPieceCid)))
+        if (present) {
+          db.revertUploadsToParked([b.subPieceCid], ctx.providerId)
+        } else {
+          db.markUploadCollected(b.subPieceCid, ctx.providerId)
+          log.message(`collected: ${b.subPieceCid} on provider ${ctx.providerId} (will re-store)`)
+        }
+      }
+    }
+  }
+
+  // Evict staged CARs whose every copy is committed. Runs after every flush,
+  // not just at run end: the disk high-water mark must track the uncommitted
+  // window, not the whole migration. The DB keeps car_path after eviction (the
+  // row is the piece's provenance), so track what this run already unlinked.
+  const evictedPaths = new Set<string>()
+  const evictCommitted = async (): Promise<void> => {
+    for (const path of db.carPathsEvictable()) {
+      if (evictedPaths.has(path)) continue
+      await deps.evictCar(path)
+      evictedPaths.add(path)
+    }
+  }
+
+  const maybeFlush = async (drained: boolean): Promise<void> => {
+    for (const ctx of contexts) {
+      // Loop: a full batch may leave more parked pieces behind it.
+      for (;;) {
+        const parked = db.parkedUploads(ctx.providerId)
+        const oldest = parked[0]
+        const reason = shouldFlush({
+          batchSize: parked.length,
+          oldestParkedAtMs: oldest == null ? null : Date.parse(oldest.parkedAt),
+          nowMs: deps.now(),
+          assumedWindowMs: windowFor(ctx),
+          marginMs: marginFromConfirmations(observedCommitMs),
+          drained,
+        })
+        if (reason == null) break
+        await flush(ctx, reason)
+        if (db.parkedUploads(ctx.providerId).length === parked.length) break // no progress; avoid spinning
+      }
+    }
+    await evictCommitted()
+  }
+
+  // Reconcile add_unconfirmed leftovers from a previous run before uploading
+  // anything new: their outcome is unknown and a blind re-add would duplicate.
+  for (const ctx of contexts) {
+    await reconcileUnconfirmed(db, synapse, ctx, deps)
+  }
+
+  // Main loop: store on the primary, fan out to secondaries, flush as batches
+  // and window timers demand. Sequential per piece: the upstream bandwidth is
+  // the bottleneck, and one in-flight store keeps the disk footprint bounded.
+  // In the pipeline an empty pending list means "wait for the packer", not
+  // "done": `waitForMore` resolves false only when the source is drained.
+  // A piece the provider disagreed with (commP mismatch) is recorded as a
+  // failed upload and drops out of this query, so it cannot re-store in a
+  // loop.
+  for (;;) {
+    const pending = db.subPiecesNeedingUpload(primary.providerId)
+    const next = pending[0]
+    if (next == null) {
+      if (opts.forceFlush?.() === true) {
+        // Downloads are blocked on the disk budget; commit whatever is
+        // parked now so eviction can free space, rather than holding the
+        // batch for the GC-window timer.
+        await maybeFlush(true)
+      }
+      if (opts.waitForMore != null && (await opts.waitForMore())) {
+        continue
+      }
+      break
+    }
+    if (next.carPath == null) {
+      // subPiecesNeedingUpload selects built rows, which always carry a CAR
+      // path; reaching this is a query bug.
+      throw new Error(`sub-piece ${next.subPieceCid} has no local CAR path`)
+    }
+
+    const storeTimer = new Timer()
+    let stored: { size: number }
+    try {
+      stored = await storeCar(primary, deps, next.carPath, next.subPieceCid)
+    } catch (err) {
+      if (err instanceof CommPMismatchError) {
+        db.markUploadFailed(next.subPieceCid, primary.providerId, 'primary', err.message)
+        log.message(`error: ${err.message}`)
+        continue
+      }
+      throw err
+    }
+    storedBytes += stored.size
+    db.recordUploadParked(next.subPieceCid, primary.providerId, 'primary', primary.dataSetId)
+    log.message(
+      `parked ${next.subPieceCid} (${formatFileSize(stored.size)}) on primary ${primary.providerId} ` +
+        `in ${formatDuration(storeTimer.stop())}`
+    )
+
+    // Check the flush timers before the secondary fanout as well as after: a
+    // slow provider-to-provider pull must not age already-parked pieces past
+    // the GC window.
+    await maybeFlush(false)
+    for (const secondary of secondaries) {
+      await pullToSecondary(db, primary, secondary, next.subPieceCid)
+    }
+
+    await maybeFlush(false)
+  }
+
+  // Source drained. Flush whatever is parked, then retry what didn't land:
+  // collected pieces (GC'd before commit) and failed secondary pulls. A
+  // primary copy re-uploads from the staged CAR; a secondary copy re-pulls
+  // from the primary, which still holds the bytes.
+  await maybeFlush(true)
+  for (let attempt = 0; attempt < 3; attempt++) {
+    // A crash between the primary store and the secondary pull leaves a
+    // sub-piece with a live primary row and no row at all for the secondary;
+    // repair those alongside the recorded retry states, or the secondary
+    // copy would silently never exist.
+    const missingSecondaries = secondaries.flatMap((ctx) =>
+      db.subPiecesMissingSecondary(primary.providerId, ctx.providerId).map((subPieceCid) => ({ ctx, subPieceCid }))
+    )
+    const needsRetry = contexts.flatMap((ctx, i) =>
+      ['collected' as const, ...(i > 0 ? ['failed' as const] : [])]
+        .flatMap((status) => db.uploadsByStatus(ctx.providerId, status))
+        .map((u) => ({ ctx, u }))
+    )
+    if (needsRetry.length === 0 && missingSecondaries.length === 0) break
+    log.message(`retrying ${needsRetry.length + missingSecondaries.length} piece(s) that did not land (attempt ${attempt + 1})`)
+    for (const { ctx, subPieceCid } of missingSecondaries) {
+      await pullToSecondary(db, primary, ctx, subPieceCid)
+    }
+    for (const { ctx, u } of needsRetry) {
+      if (u.role === 'secondary') {
+        await pullToSecondary(db, primary, ctx, u.subPieceCid)
+        continue
+      }
+      const sub = db.subPieceByCid(u.subPieceCid)
+      if (sub?.carPath == null) {
+        log.message(`error: collected ${u.subPieceCid} has no local CAR; cannot re-store`)
+        continue
+      }
+      try {
+        const stored = await storeCar(ctx, deps, sub.carPath, sub.subPieceCid)
+        storedBytes += stored.size
+        db.recordUploadParked(sub.subPieceCid, ctx.providerId, u.role, ctx.dataSetId)
+      } catch (err) {
+        if (err instanceof CommPMismatchError) {
+          db.markUploadFailed(sub.subPieceCid, ctx.providerId, u.role, err.message)
+          log.message(`error: ${err.message}`)
+          continue
+        }
+        throw err
+      }
+    }
+    await maybeFlush(true)
+  }
+
+  await evictCommitted()
+
+  const summary: DirectUploadSummary = {
+    network: synapse.chain.name,
+    providers: contexts.map((ctx, i) => {
+      const committed = db.uploadsByStatus(ctx.providerId, 'committed')
+      return {
+        providerId: ctx.providerId,
+        role: i === 0 ? ('primary' as const) : ('secondary' as const),
+        dataSetId: latestDataSetId(db, ctx),
+        committed: committed.length,
+        collected: db.uploadsByStatus(ctx.providerId, 'collected').length,
+        failed: db.uploadsByStatus(ctx.providerId, 'failed').length,
+        addUnconfirmed: db.uploadsByStatus(ctx.providerId, 'add_unconfirmed').length,
+        flushes: flushCounts.get(ctx.providerId) ?? 0,
+        assumedWindowMs: windowFor(ctx),
+        txHashes: [...new Set(committed.map((u) => u.txHash).filter((h): h is string => h != null))],
+      }
+    }),
+    storedBytes,
+    evictedCars: evictedPaths.size,
+  }
+  log.message(`direct upload finished in ${formatDuration(runTimer.stop())}: ${formatFileSize(storedBytes)} stored`)
+  return summary
+}
+
+/**
+ * How old an add_unconfirmed breadcrumb must be before its piece re-enters
+ * the flow: for a hashless row, before an absent-on-chain verdict is
+ * trusted; for a row whose tx has no receipt, before the tx is presumed
+ * dead. Sized to outlast any realistic confirmation window, because the
+ * failure mode on both paths is a duplicate on-chain add.
+ */
+export const UNCONFIRMED_REQUEUE_AFTER_MS = 60 * 60_000
+
+/** Thrown when the provider's commitment over the uploaded bytes disagrees with ours. */
+export class CommPMismatchError extends Error {
+  constructor(subPieceCid: string, providerId: string, got: string) {
+    super(`commP mismatch on provider ${providerId}: expected ${subPieceCid}, provider computed ${got}`)
+    this.name = 'CommPMismatchError'
+  }
+}
+
+async function storeCar(
+  ctx: UploadContextLike,
+  deps: DirectUploadDeps,
+  carPath: string,
+  subPieceCid: string
+): Promise<{ size: number }> {
+  const result = await ctx.store(deps.openCar(carPath), { pieceCid: CID.parse(subPieceCid) })
+  // The committed CID must be the commitment over the bytes the provider
+  // actually holds; trusting the SDK to throw on divergence is not enough.
+  const got = String(result.pieceCid)
+  if (got !== subPieceCid) {
+    throw new CommPMismatchError(subPieceCid, ctx.providerId, got)
+  }
+  return { size: result.size }
+}
+
+/**
+ * Resolve every add_unconfirmed row. The transaction receipt is checked first:
+ * piece presence alone cannot distinguish "commit never landed" from "commit
+ * landed but confirmation was missed": the provider keeps the bytes either
+ * way, and re-queueing a landed commit would add the piece twice.
+ */
+async function reconcileUnconfirmed(
+  db: MigrationDB,
+  synapse: Synapse,
+  ctx: UploadContextLike,
+  deps: DirectUploadDeps
+): Promise<void> {
+  for (const u of db.uploadsByStatus(ctx.providerId, 'add_unconfirmed')) {
+    if (u.txHash == null) {
+      // The crash landed between the transaction broadcast and the hash
+      // callback, so no receipt can be checked. The data set itself is the
+      // witness: a piece present on chain is committed; an absent one did
+      // not land. Without a known data set neither can be told apart, so the
+      // row stays unresolved and the run exits incomplete.
+      const dataSetId = u.dataSetId ?? ctx.dataSetId
+      if (dataSetId == null) {
+        log.message(
+          `resume: ${u.subPieceCid} has an unconfirmed addPieces with no transaction hash and no known data set ` +
+            `on provider ${ctx.providerId}; left add_unconfirmed for manual resolution`
+        )
+        continue
+      }
+      const pieceId = await deps.dataSetPieceId(synapse, dataSetId, u.subPieceCid)
+      if (pieceId != null) {
+        db.markUploadCommitted(u.subPieceCid, ctx.providerId, { dataSetId, pieceId, txHash: null })
+        log.message(`resume: ${u.subPieceCid} found on chain in data set ${dataSetId} (piece ${pieceId}); marked committed`)
+        continue
+      }
+      // Absent on chain. A transaction the provider broadcast just before
+      // the crash could still land, and re-queueing while it can would add
+      // the piece twice, so the row only re-enters the flow once the
+      // breadcrumb is older than any realistic confirmation window. Younger
+      // rows stay unresolved and the run exits incomplete; a re-run later
+      // resolves them one way or the other.
+      const ageMs = deps.now() - Date.parse(u.updatedAt)
+      if (ageMs < UNCONFIRMED_REQUEUE_AFTER_MS) {
+        log.message(
+          `resume: ${u.subPieceCid} has an unconfirmed addPieces with no transaction hash and is absent from ` +
+            `data set ${dataSetId}; too recent to rule out an in-flight transaction, left add_unconfirmed ` +
+            `(re-run after ${formatDuration(UNCONFIRMED_REQUEUE_AFTER_MS - ageMs)})`
+        )
+        continue
+      }
+      // Old enough that an in-flight transaction would have landed or died.
+      // Fall through to the presence check below so the piece re-parks or
+      // re-stores like any other unconfirmed attempt.
+    }
+    if (u.txHash != null) {
+      if (await deps.txLanded(synapse, u.txHash)) {
+        // The commit landed; only local confirmation was missed. Resolve it
+        // from the canonical witness (the PiecesAdded event) rather than
+        // trusting any side channel. The row's own data set takes precedence
+        // when known; a first-commit crash on a new data set has none, and
+        // then the event's own setId is the recovery path.
+        const event = await deps.fetchAddPiecesEvent(synapse, u.txHash, u.dataSetId ?? ctx.dataSetId)
+        const pieceIndex = event == null ? -1 : event.pieceCids.indexOf(u.subPieceCid)
+        if (event != null && pieceIndex >= 0) {
+          const dataSetId = String(event.dataSetId)
+          db.markUploadCommitted(u.subPieceCid, ctx.providerId, {
+            dataSetId,
+            pieceId: String(event.pieceIds[pieceIndex] ?? ''),
+            txHash: u.txHash,
+          })
+          log.message(
+            `resume: ${u.subPieceCid} confirmed on chain via PiecesAdded (tx ${u.txHash}, ` +
+              `data set ${dataSetId}); marked committed`
+          )
+          continue
+        }
+        log.message(
+          `resume: ${u.subPieceCid} has a LANDED addPieces tx ${u.txHash} on provider ${ctx.providerId} ` +
+            `but its PiecesAdded event could not be verified; leaving add_unconfirmed: check the data set ` +
+            `on the explorer before any manual retry (a blind re-add would duplicate the piece)`
+        )
+        continue
+      }
+      // Not landed is not dead: the tx can still be in the mempool, and
+      // re-parking now would let the next flush double-add once it lands.
+      // Same age gate as the hashless path: only a breadcrumb older than any
+      // realistic confirmation window re-enters the flow.
+      const ageMs = deps.now() - Date.parse(u.updatedAt)
+      if (ageMs < UNCONFIRMED_REQUEUE_AFTER_MS) {
+        log.message(
+          `resume: ${u.subPieceCid} has an unlanded addPieces tx ${u.txHash} on provider ${ctx.providerId}; ` +
+            `too recent to rule out an in-flight transaction, left add_unconfirmed ` +
+            `(re-run after ${formatDuration(UNCONFIRMED_REQUEUE_AFTER_MS - ageMs)})`
+        )
+        continue
+      }
+    }
+    if (await ctx.hasPiece(CID.parse(u.subPieceCid))) {
+      db.revertUploadsToParked([u.subPieceCid], ctx.providerId)
+      log.message(`resume: ${u.subPieceCid} still parked on provider ${ctx.providerId}; re-queued for commit`)
+    } else {
+      db.markUploadCollected(u.subPieceCid, ctx.providerId)
+      log.message(`resume: ${u.subPieceCid} gone from provider ${ctx.providerId}; will re-store`)
+    }
+  }
+}
+
+/** Have one secondary pull a freshly parked piece from the primary. */
+async function pullToSecondary(
+  db: MigrationDB,
+  primary: UploadContextLike,
+  secondary: UploadContextLike,
+  subPieceCid: string
+): Promise<void> {
+  try {
+    // Curio authenticates the pull with the same EIP-712 authorization used
+    // for commit: a pull without it is rejected.
+    const extraData = await secondary.presignForCommit([{ pieceCid: CID.parse(subPieceCid) }])
+    const pulled = await secondary.pull({
+      pieces: [CID.parse(subPieceCid)],
+      from: (pieceCid) => primary.getPieceUrl(pieceCid),
+      extraData,
+    })
+    if (pulled.status === 'complete') {
+      db.recordUploadParked(subPieceCid, secondary.providerId, 'secondary', secondary.dataSetId)
+      log.message(`parked ${subPieceCid} on secondary ${secondary.providerId} (pulled from primary)`)
+    } else {
+      db.markUploadFailed(subPieceCid, secondary.providerId, 'secondary', 'secondary pull failed')
+      log.message(`warn: secondary ${secondary.providerId} failed to pull ${subPieceCid}`)
+    }
+  } catch (err) {
+    db.markUploadFailed(subPieceCid, secondary.providerId, 'secondary', (err as Error).message)
+    log.message(`warn: secondary ${secondary.providerId} pull error for ${subPieceCid}: ${(err as Error).message}`)
+  }
+}
+
+function latestDataSetId(db: MigrationDB, ctx: UploadContextLike): string | null {
+  const committed = db.uploadsByStatus(ctx.providerId, 'committed')
+  const last = committed[committed.length - 1]
+  return last != null ? last.dataSetId : ctx.dataSetId
+}
diff --git a/src/migrate/pdp-verifier.ts b/src/migrate/pdp-verifier.ts
new file mode 100644
index 00000000..716c6a13
--- /dev/null
+++ b/src/migrate/pdp-verifier.ts
@@ -0,0 +1,94 @@
+/**
+ * Read-only PDPVerifier access for reconciling a migrate run's on-chain
+ * state. The ABI and contract address come from the connected chain via
+ * `@filoz/synapse-core`.
+ */
+
+import { pdp as PDP_ABI } from '@filoz/synapse-core/abis'
+import { findPieceIdsByCid } from '@filoz/synapse-core/pdp-verifier'
+import { from as pieceCidFrom } from '@filoz/synapse-core/piece'
+import type { Synapse } from '@filoz/synapse-sdk'
+import { type Hash, type Hex, parseEventLogs, TransactionReceiptNotFoundError } from 'viem'
+
+/**
+ * The on-chain PiecesAdded event parsed from an AddPieces tx receipt.
+ * PDPVerifier emits one event per AddPieces call carrying parallel arrays of
+ * pieceIds + pieceCids (event `PiecesAdded(uint256 indexed setId, uint256[]
+ * pieceIds, struct Cids.Cid[] pieceCids)`); `dataSetId` is the event's own
+ * setId, so a caller that lost track of its data set can recover it from
+ * the receipt.
+ */
+export interface AddPiecesEvent {
+  dataSetId: bigint
+  blockNumber: bigint
+  pieceIds: bigint[]
+  pieceCids: string[]
+}
+
+/**
+ * The on-chain piece id of `pieceCid` in `dataSetId`, or null when the data
+ * set does not contain it. One contract read; used to resolve an
+ * add_unconfirmed row that never captured its transaction hash. `dataSetId`
+ * stays a string end to end: on-chain set ids are uint256 and `Number`
+ * would silently truncate.
+ */
+export async function dataSetPieceId(synapse: Synapse, dataSetId: string, pieceCid: string): Promise<string | null> {
+  const ids = await findPieceIdsByCid(synapse.client as never, {
+    dataSetId: BigInt(dataSetId),
+    pieceCid: pieceCidFrom(pieceCid),
+  })
+  const first = ids[0]
+  return first == null ? null : String(first)
+}
+
+/**
+ * Whether a transaction landed successfully on chain (receipt status
+ * success). Only a definitive missing receipt maps to false; any other
+ * error (RPC 5xx, rate limit, timeout) propagates, because callers make
+ * re-add decisions on a false and a transient blip must not look like a
+ * dead transaction.
+ */
+export async function txLanded(synapse: Synapse, txHash: string): Promise<boolean> {
+  try {
+    const receipt = await synapse.client.getTransactionReceipt({ hash: txHash as Hash })
+    return receipt.status === 'success'
+  } catch (err) {
+    if (err instanceof TransactionReceiptNotFoundError) return false
+    throw err
+  }
+}
+
+/**
+ * Fetch and parse the PiecesAdded event out of an AddPieces tx receipt.
+ * When `dataSetId` is known it must match the event's setId; when null (a
+ * crash before the first commit of a new data set recorded one) any
+ * PDPVerifier-emitted event in the receipt matches and its setId is
+ * returned. Returns null when the receipt carries no matching event (a
+ * reverted inner call leaves no PiecesAdded log even when the tx itself
+ * succeeded).
+ */
+export async function fetchAddPiecesEvent(
+  synapse: Synapse,
+  txHash: string,
+  dataSetId: string | null
+): Promise<AddPiecesEvent | null> {
+  const pdpAddress = synapse.chain.contracts.pdp.address
+  const receipt = await synapse.client.waitForTransactionReceipt({ hash: txHash as Hash })
+  const events = parseEventLogs({
+    abi: PDP_ABI,
+    eventName: 'PiecesAdded',
+    logs: receipt.logs,
+  })
+  const target = dataSetId == null ? null : BigInt(dataSetId)
+  const match = events.find(
+    (ev) => ev.address.toLowerCase() === pdpAddress.toLowerCase() && (target == null || ev.args.setId === target)
+  )
+  if (match == null) return null
+  const pieceCids = match.args.pieceCids.map((p: { data: Hex }) => pieceCidFrom(p.data).toString())
+  return {
+    dataSetId: match.args.setId,
+    blockNumber: receipt.blockNumber,
+    pieceIds: [...match.args.pieceIds],
+    pieceCids,
+  }
+}
diff --git a/src/test/unit/migrate-direct-upload-flow.test.ts b/src/test/unit/migrate-direct-upload-flow.test.ts
new file mode 100644
index 00000000..cb5f9bf5
--- /dev/null
+++ b/src/test/unit/migrate-direct-upload-flow.test.ts
@@ -0,0 +1,522 @@
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import type { Synapse } from '@filoz/synapse-sdk'
+import { describe, expect, it } from 'vitest'
+import { MigrationDB } from '../../migrate/db.js'
+import {
+  type DirectUploadDeps,
+  type DirectUploadOptions,
+  runDirectUpload,
+  type UploadContextLike,
+} from '../../migrate/direct-upload.js'
+
+// Drives the real runDirectUpload control flow with fake providers, to lock in
+// the direct-upload guarantees: store-then-batch-commit, the add_unconfirmed
+// breadcrumb, GC detection lowering the window and re-storing only what is
+// actually gone, commP verification of what the provider received, and CAR
+// eviction once the primary copy is committed.
+
+// Real PieceCIDs so CID.parse succeeds.
+const P1 = 'bafkzcibf3ck4uais4fgennh4hbfx5z3i6hue4xgq2cdeamtus4hjbsrjs5lf2azbxmsa'
+const P2 = 'bafkzcibewpkqwewyhz3yxutlxbpt2nkb6si5qilg4qqtzzij32uw7ammsc73a4wkgi'
+
+const fakeSynapse = { chain: { name: 'calibration' } } as unknown as Synapse
+const OPTS: DirectUploadOptions = { synapse: fakeSynapse, copies: 2 }
+const SCOPE = 'calibration:0xabc'
+
+async function dbAt(name: string) {
+  const dir = await mkdtemp(join(tmpdir(), `fp-${name}-`))
+  return { dir, db: new MigrationDB(join(dir, 'migrate.db'), SCOPE) }
+}
+
+function seedBuilt(db: MigrationDB, subPieceCid: string, carPath: string) {
+  const src = `src-${subPieceCid.slice(-8)}`
+  db.addCids([src])
+  db.recordPieceSuccess(src, {
+    pieceCid: subPieceCid,
+    rawSize: 1024,
+    gateway: 'g',
+    url: `https://gw/ipfs/${src}?format=car`,
+    memberCarPath: `${carPath}.member`,
+    memberSha256: 'sha',
+  })
+  db.recordBuiltSubPiece({
+    subPieceCid,
+    assembledCarLength: 1024,
+    targetSizeBytes: 1024,
+    carPath,
+    assembledSha256: 'sha',
+    members: [{ cid: src, sha256: 'sha', rawSize: 1024 }],
+  })
+}
+
+interface FakeBehavior {
+  /** Throw on the numbered commit call (1-based) on the given provider. */
+  failCommit?: { providerId: string; call: number; message: string }
+  /** Per-CID presence answers for hasPiece during re-verify. */
+  present?: (cid: string) => boolean
+  pullFails?: boolean
+  /** Store answers a different commitment than requested (commP mismatch). */
+  storeReturns?: (pieceCid: string) => string
+  /** Receipt answers for add_unconfirmed reconciliation. Default: not landed. */
+  txLanded?: (txHash: string) => boolean
+  /** PiecesAdded event answers for landed-tx resolution. Default: none found. */
+  addPiecesEvent?: (
+    dataSetId: string | null,
+    txHash: string
+  ) => { dataSetId: bigint; pieceIds: bigint[]; pieceCids: string[] } | null
+  /** On-chain piece-id answers for hashless reconciliation. Default: absent. */
+  dataSetPieceId?: (dataSetId: string, pieceCid: string) => string | null
+  /** Pre-resolved data set id for every context. Default null (created lazily). */
+  ctxDataSetId?: string
+  /** Offset added to the fake clock, for aging add_unconfirmed breadcrumbs. */
+  nowOffsetMs?: number
+}
+
+function fakeDeps(b: FakeBehavior = {}) {
+  const calls = { store: [] as string[], pull: 0, commit: new Map<string, number>() }
+  const evicted: string[] = []
+  const mkCtx = (providerId: string): UploadContextLike => ({
+    providerId,
+    serviceURL: `fake://${providerId}`,
+    dataSetId: b.ctxDataSetId ?? null,
+    async store(_data, options) {
+      const requested = String(options.pieceCid)
+      calls.store.push(`${providerId}:${requested}`)
+      const answered = b.storeReturns ? b.storeReturns(requested) : requested
+      return { pieceCid: answered, size: 1024 }
+    },
+    async presignForCommit() {
+      return '0xfake'
+    },
+    async pull(options) {
+      calls.pull++
+      const status = b.pullFails ? ('failed' as const) : ('complete' as const)
+      return { status, pieces: options.pieces.map((p) => ({ pieceCid: p, status })) }
+    },
+    async commit(options) {
+      const n = (calls.commit.get(providerId) ?? 0) + 1
+      calls.commit.set(providerId, n)
+      const f = b.failCommit
+      if (f != null && f.providerId === providerId && f.call === n) {
+        // Simulate a commit that was submitted (tx hash recorded) but then
+        // failed at confirmation: the shape of the add_unconfirmed hazard.
+        options.onSubmitted?.(`0xtx-${providerId}-${n}`)
+        throw new Error(f.message)
+      }
+      return {
+        txHash: `0xtx-${providerId}-${n}`,
+        pieceIds: options.pieces.map((_, i) => BigInt(i)),
+        dataSetId: 7n,
+      }
+    },
+    getPieceUrl: (pieceCid) => `fake://${providerId}/piece/${String(pieceCid)}`,
+    hasPiece: async (pieceCid) => (b.present ? b.present(String(pieceCid)) : true),
+  })
+  const deps: DirectUploadDeps = {
+    async setup() {
+      return { contexts: [mkCtx('p1'), mkCtx('p2')] }
+    },
+    now: () => Date.now() + (b.nowOffsetMs ?? 0),
+    openCar: () => new Uint8Array(8),
+    evictCar: async (path) => {
+      evicted.push(path)
+    },
+    txLanded: async (_synapse, txHash) => (b.txLanded ? b.txLanded(txHash) : false),
+    fetchAddPiecesEvent: async (_synapse, txHash, dataSetId) => {
+      const event = b.addPiecesEvent ? b.addPiecesEvent(dataSetId, txHash) : null
+      return event == null ? null : { ...event, blockNumber: 1n }
+    },
+    dataSetPieceId: async (_synapse, dataSetId, pieceCid) =>
+      b.dataSetPieceId ? b.dataSetPieceId(dataSetId, pieceCid) : null,
+  }
+  return { deps, calls, evicted }
+}
+
+describe('runDirectUpload', () => {
+  it('happy path: stores on primary, pulls to secondary, drained flush commits both, evicts CARs', async () => {
+    const { dir, db } = await dbAt('du-happy')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      seedBuilt(db, P2, join(dir, 'b.car'))
+      const { deps, calls, evicted } = fakeDeps()
+      const summary = await runDirectUpload(db, OPTS, deps)
+
+      expect(calls.store).toEqual([`p1:${P1}`, `p1:${P2}`])
+      expect(calls.pull).toBe(2)
+      // One drained flush per provider, both pieces in one batch.
+      expect(calls.commit.get('p1')).toBe(1)
+      expect(calls.commit.get('p2')).toBe(1)
+      for (const provider of ['p1', 'p2']) {
+        const committed = db.uploadsByStatus(provider, 'committed')
+        expect(committed.map((u) => u.subPieceCid).sort()).toEqual([P2, P1].sort())
+        for (const u of committed) expect(u.dataSetId).toBe('7')
+      }
+      expect(evicted).toHaveLength(2)
+      expect(summary.providers[0]?.committed).toBe(2)
+      expect(summary.providers[0]?.role).toBe('primary')
+      expect(summary.providers[0]?.addUnconfirmed).toBe(0)
+      // The summary must surface the addPieces transactions behind the commits.
+      expect(summary.providers[0]?.txHashes).toEqual(['0xtx-p1-1'])
+      expect(summary.providers[1]?.txHashes).toEqual(['0xtx-p2-1'])
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('secondary pull failure leaves the secondary failed but frees the CAR once the primary commits', async () => {
+    const { dir, db } = await dbAt('du-pullfail')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      const { deps, evicted } = fakeDeps({ pullFails: true })
+      const summary = await runDirectUpload(db, OPTS, deps)
+
+      expect(db.uploadsByStatus('p1', 'committed')).toHaveLength(1)
+      expect(db.uploadsByStatus('p2', 'failed')).toHaveLength(1)
+      // A secondary retry pulls provider-to-provider from the committed
+      // primary; it never needs the local file, so the CAR is evicted.
+      expect(evicted).toHaveLength(1)
+      // The failed copy still marks the run incomplete for the exit code.
+      expect(summary.providers[1]?.failed).toBe(1)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('a commP mismatch from store() fails the piece instead of committing the wrong bytes', async () => {
+    const { dir, db } = await dbAt('du-commp')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      const { deps, calls, evicted } = fakeDeps({ storeReturns: () => P2 })
+      await runDirectUpload(db, OPTS, deps)
+
+      const failed = db.uploadsByStatus('p1', 'failed')
+      expect(failed).toHaveLength(1)
+      expect(failed[0]?.error).toMatch(/commP mismatch/)
+      expect(calls.commit.get('p1') ?? 0).toBe(0)
+      expect(evicted).toHaveLength(0)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('repairs a missing secondary copy left by a crash between store and pull', async () => {
+    const { dir, db } = await dbAt('du-repair')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      // The crash shape: the primary parked but the secondary pull never
+      // started, so the secondary has no row at all.
+      db.recordUploadParked(P1, 'p1', 'primary', null)
+      const { deps, calls } = fakeDeps()
+      await runDirectUpload(db, OPTS, deps)
+
+      // No re-store on the primary; the secondary copy was pulled and both
+      // copies committed.
+      expect(calls.store).toHaveLength(0)
+      expect(calls.pull).toBe(1)
+      expect(db.uploadsByStatus('p1', 'committed')).toHaveLength(1)
+      expect(db.uploadsByStatus('p2', 'committed')).toHaveLength(1)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('GC rejection: lowers the provider window, re-stores only the collected piece, keeps the rest parked', async () => {
+    const { dir, db } = await dbAt('du-gc')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      seedBuilt(db, P2, join(dir, 'b.car'))
+      const { deps, calls } = fakeDeps({
+        failCommit: {
+          providerId: 'p1',
+          call: 1,
+          message: `Failed to process request: subPiece CID ${P1} not found or does not belong to service svc`,
+        },
+        present: (cid) => cid !== P1,
+      })
+      await runDirectUpload(db, OPTS, deps)
+
+      // P1 was re-stored on the primary after being collected; P2 was not.
+      expect(calls.store.filter((s) => s === `p1:${P1}`)).toHaveLength(2)
+      expect(calls.store.filter((s) => s === `p1:${P2}`)).toHaveLength(1)
+      // Both pieces end up committed via the retry flush.
+      expect(db.uploadsByStatus('p1', 'committed')).toHaveLength(2)
+      // The window guess dropped below the default for the flaky provider only.
+      const defaultMs = 60 * 60_000
+      expect(db.providerWindowMs('p1', defaultMs)).toBeLessThan(defaultMs)
+      expect(db.providerWindowMs('p2', defaultMs)).toBe(defaultMs)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('a landed-but-unconfirmed tx is never re-queued: no blind re-add', async () => {
+    const { dir, db } = await dbAt('du-landed')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      const first = fakeDeps({
+        failCommit: { providerId: 'p1', call: 1, message: 'confirmation poll timed out' },
+      })
+      await runDirectUpload(db, OPTS, first.deps)
+      expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1)
+
+      // Second run: the tx actually landed on chain even though confirmation
+      // was missed. The reconciliation must leave the row alone: no store,
+      // no commit: and the summary must count it.
+      const second = fakeDeps({ txLanded: () => true })
+      const summary = await runDirectUpload(db, OPTS, second.deps)
+      expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1)
+      expect(second.calls.store.filter((s) => s.startsWith('p1:'))).toHaveLength(0)
+      expect(second.calls.commit.get('p1') ?? 0).toBe(0)
+      expect(summary.providers[0]?.addUnconfirmed).toBe(1)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('a landed tx resolves from the PiecesAdded event using the data set recorded on the row', async () => {
+    const { dir, db } = await dbAt('du-landed-event')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      const first = fakeDeps({
+        ctxDataSetId: '7',
+        failCommit: { providerId: 'p1', call: 1, message: 'confirmation poll timed out' },
+      })
+      await runDirectUpload(db, OPTS, first.deps)
+      expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1)
+
+      // Second run opens fresh contexts with no resolved data set; the row's
+      // own data_set_id (recorded when the piece parked) must drive the
+      // event lookup.
+      const second = fakeDeps({
+        txLanded: () => true,
+        addPiecesEvent: (dataSetId) =>
+          dataSetId === '7' ? { dataSetId: 7n, pieceIds: [42n], pieceCids: [P1] } : null,
+      })
+      await runDirectUpload(db, OPTS, second.deps)
+      const committed = db.uploadsByStatus('p1', 'committed')
+      expect(committed).toHaveLength(1)
+      expect(committed[0]?.pieceId).toBe('42')
+      expect(committed[0]?.dataSetId).toBe('7')
+      // Resolved from the chain, not re-executed.
+      expect(second.calls.commit.get('p1') ?? 0).toBe(0)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('non-GC commit failure leaves the batch add_unconfirmed with no stale tx hash after re-park', async () => {
+    const { dir, db } = await dbAt('du-unconfirmed')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      const first = fakeDeps({
+        failCommit: { providerId: 'p1', call: 1, message: 'insufficient funds' },
+      })
+      await runDirectUpload(db, OPTS, first.deps)
+      expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1)
+
+      // Second run, past the requeue window (a younger row with a captured
+      // tx hash is held: the tx could still land and re-adding would
+      // double-add). The piece is still parked on the provider, so the
+      // resume reconciliation re-queues it and the commit lands, without
+      // re-storing and with the failed attempt's tx hash replaced by the
+      // real one.
+      const second = fakeDeps({ nowOffsetMs: 61 * 60_000 })
+      await runDirectUpload(db, OPTS, second.deps)
+      const committed = db.uploadsByStatus('p1', 'committed')
+      expect(committed).toHaveLength(1)
+      expect(committed[0]?.txHash).toBe('0xtx-p1-1')
+      expect(second.calls.store.filter((s) => s.startsWith('p1:'))).toHaveLength(0)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('resolves a hashless add_unconfirmed row from the data set itself when the piece is on chain', async () => {
+    const { dir, db } = await dbAt('du-hashless-committed')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      // The crash shape: the breadcrumb was written and the commit may have
+      // broadcast, but the process died before the tx hash callback.
+      db.recordUploadParked(P1, 'p1', 'primary', '7')
+      db.markUploadsAddUnconfirmed([P1], 'p1')
+
+      const { deps, calls } = fakeDeps({ dataSetPieceId: (dataSetId) => (dataSetId === '7' ? '42' : null) })
+      await runDirectUpload(db, OPTS, deps)
+
+      const committed = db.uploadsByStatus('p1', 'committed')
+      expect(committed).toHaveLength(1)
+      expect(committed[0]?.pieceId).toBe('42')
+      // Resolved from the chain: no re-store, no second commit for it.
+      expect(calls.store.filter((s) => s.startsWith('p1:'))).toHaveLength(0)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('holds a fresh add_unconfirmed row whose tx has no receipt yet', async () => {
+    const { dir, db } = await dbAt('du-pending-tx')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      db.recordUploadParked(P1, 'p1', 'primary', '7')
+      db.markUploadsAddUnconfirmed([P1], 'p1')
+      db.markUploadTxSubmitted([P1], 'p1', '0xpending')
+
+      // txLanded false = no receipt. The tx can still be in the mempool, so
+      // the row must not re-park (a second commit would double-add).
+      const { deps, calls } = fakeDeps({ txLanded: () => false })
+      await runDirectUpload(db, OPTS, deps)
+
+      expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1)
+      expect(calls.commit.get('p1') ?? 0).toBe(0)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('recovers the data set id from the PiecesAdded event when none was recorded', async () => {
+    const { dir, db } = await dbAt('du-event-setid')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      // First-commit crash shape on a new data set: parked before any set id
+      // existed, tx hash captured, no data_set_id anywhere.
+      db.recordUploadParked(P1, 'p1', 'primary', null)
+      db.markUploadsAddUnconfirmed([P1], 'p1')
+      db.markUploadTxSubmitted([P1], 'p1', '0xfirst')
+
+      const { deps } = fakeDeps({
+        txLanded: () => true,
+        addPiecesEvent: (dataSetId) =>
+          dataSetId == null ? { dataSetId: 9n, pieceIds: [5n], pieceCids: [P1] } : null,
+      })
+      await runDirectUpload(db, OPTS, deps)
+
+      const committed = db.uploadsByStatus('p1', 'committed')
+      expect(committed).toHaveLength(1)
+      expect(committed[0]?.dataSetId).toBe('9')
+      expect(committed[0]?.pieceId).toBe('5')
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('holds a fresh hashless add_unconfirmed row even when absent from the data set', async () => {
+    const { dir, db } = await dbAt('du-hashless-fresh')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      db.recordUploadParked(P1, 'p1', 'primary', '7')
+      db.markUploadsAddUnconfirmed([P1], 'p1')
+
+      // A just-written breadcrumb cannot rule out a transaction still in
+      // flight; the row must stay unresolved this run.
+      const { deps, calls } = fakeDeps()
+      const summary = await runDirectUpload(db, OPTS, deps)
+
+      expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1)
+      expect(calls.commit.get('p1') ?? 0).toBe(0)
+      expect(summary.providers[0]?.addUnconfirmed).toBe(1)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('re-parks a hashless add_unconfirmed row absent from the data set once it has aged out', async () => {
+    const { dir, db } = await dbAt('du-hashless-aged')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      db.recordUploadParked(P1, 'p1', 'primary', '7')
+      db.markUploadsAddUnconfirmed([P1], 'p1')
+
+      // Two hours later, an in-flight transaction would have landed or died;
+      // absent on chain and still parked on the provider means re-queue, and
+      // the commit lands without re-storing.
+      const { deps, calls } = fakeDeps({ nowOffsetMs: 2 * 60 * 60_000 })
+      await runDirectUpload(db, OPTS, deps)
+
+      expect(db.uploadsByStatus('p1', 'committed')).toHaveLength(1)
+      expect(calls.store.filter((s) => s.startsWith('p1:'))).toHaveLength(0)
+      expect(calls.commit.get('p1')).toBe(1)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('leaves a hashless add_unconfirmed row alone when no data set is known', async () => {
+    const { dir, db } = await dbAt('du-hashless-unknown')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      db.recordUploadParked(P1, 'p1', 'primary', null)
+      db.markUploadsAddUnconfirmed([P1], 'p1')
+
+      const { deps, calls } = fakeDeps()
+      const summary = await runDirectUpload(db, OPTS, deps)
+
+      expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1)
+      expect(calls.commit.get('p1') ?? 0).toBe(0)
+      expect(summary.providers[0]?.addUnconfirmed).toBe(1)
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('refuses to rebuild a staged piece while it has live upload state', async () => {
+    const { dir, db } = await dbAt('du-rebuild-guard')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      db.recordUploadParked(P1, 'p1', 'primary', '7')
+      db.markUploadsAddUnconfirmed([P1], 'p1')
+
+      // An unresolved breadcrumb must survive any rebuild attempt: deleting
+      // it and re-adding the source CIDs could duplicate the piece on chain.
+      expect(() => db.deleteSubPieceForRebuild(P1)).toThrow(/live upload/)
+      expect(db.uploadsByStatus('p1', 'add_unconfirmed')).toHaveLength(1)
+
+      // Once reconciliation resolves the row to a terminal retry state, the
+      // rebuild goes through and frees the members.
+      db.markUploadCollected(P1, 'p1')
+      const members = db.deleteSubPieceForRebuild(P1)
+      expect(members).toHaveLength(1)
+      expect(db.subPieceByCid(P1)).toBeNull()
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('scopes state: another network/owner scope sees none of the rows', async () => {
+    const { dir, db } = await dbAt('du-scope')
+    try {
+      seedBuilt(db, P1, join(dir, 'a.car'))
+      const { deps } = fakeDeps()
+      await runDirectUpload(db, OPTS, deps)
+      expect(db.uploadsByStatus('p1', 'committed')).toHaveLength(1)
+
+      const other = new MigrationDB(db.path, 'mainnet:0xdef')
+      try {
+        expect(other.subPiecesNeedingUpload('p1')).toHaveLength(0)
+        expect(other.uploadsByStatus('p1', 'committed')).toHaveLength(0)
+        expect(other.counts().total).toBe(0)
+      } finally {
+        other.close()
+      }
+    } finally {
+      db.close()
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+})