diff --git a/README.md b/README.md
index 3833371..40480be 100644
--- a/README.md
+++ b/README.md
@@ -123,11 +123,6 @@ you target (default `mainnet`; pass `--network calibration` for the testnet).
 Complete **Prerequisites** above first. Default network is **mainnet**; pass
 `--network calibration` for the testnet.
 
-Step 3 starts a long-running process and step 4 runs against it, so the two
-overlap: start step 3, wait for it to log its public URL, and leave it up for
-the whole of step 4. Two shells, a background job, or a terminal multiplexer
-all work.
-
 > **First time?** Rehearse the whole flow on the testnet with the
 > [calibration tutorial](docs/tutorial-first-migration.md) before spending real
 > funds on mainnet.
@@ -138,57 +133,50 @@ export PRIVATE_KEY=0x...
 # One-time payer setup: deposit USDFC and approve FWSS as a payments operator.
 npx filecoin-pin@latest payments setup --auto
 
-# 0. (Once per provider) Provision a data set with withIPFSIndexing. Note the
-#    printed `dataSetId`; reuse it in steps 4 and 5.
-ipfs2foc create-data-set --provider-id <provider-id>
-
 # 1. Confirm a trustless gateway returns a deterministic CAR for one of your CIDs.
 ipfs2foc probe <sample-cid> --gateway https://trustless-gateway.link
 
-# 2. Compute piece commitments and pack aggregates (one source CID per sub-piece).
+# 2. Migrate: download the CIDs, pack ~1 GiB multi-root CARs under --car-store,
+#    stream each straight to two providers, and batch the on-chain adds before
+#    the provider's parked-piece GC window closes. Resumable; re-run to continue.
+#    Data sets are provisioned on first commit (withIPFSIndexing set); their ids
+#    are in the printed summary.
 printf '%s\n' <cid> > cids.txt
-ipfs2foc plan --cids cids.txt --db migrate.db
-
-# 3. Serve sub-pieces over public HTTPS. Leave this running for step 4.
-#    `--ingress cloudflared` spawns a no-signup Cloudflare tunnel, then logs:
-#      cloudflared ingress: ready at https://<words>.trycloudflare.com
-#      Pass it to pdp-submit: --source-base https://<words>.trycloudflare.com
-#    Wait for those lines; the second one is step 4's --source-base verbatim.
-#    (`ipfs2foc serve --ingress cloudflared` answers the same /piece route from
-#     the console daemon, so one process carries both and step 3 is not needed.)
-ipfs2foc redirect-serve --db migrate.db --port 4322 --ingress cloudflared
-
-# 4. Pull, park, and add each aggregate onto the provider's data set, while
-#    step 3 keeps running. `--source-base` is the public HTTPS origin only
-#    (scheme and host, no path) that step 3 logged.
-ipfs2foc pdp-submit --db migrate.db --data-set-id <data-set-id> \
-  --source-base https://<public-host>
+ipfs2foc upload --cids cids.txt --db migrate.db --car-store ./cars
 
-# 5. Confirm every CID landed: reconcile local state against the on-chain pieces.
+# 3. Confirm every CID landed: reconcile local state against the on-chain pieces.
 ipfs2foc report --db migrate.db --data-set-id <data-set-id>
 ```
 
 `cids.txt`: one CID per line; blank lines and `#` comments are ignored.
 
-`plan` is **INSERT-only**: re-running it after appending CIDs adds new
-sub-pieces and aggregates without rewriting prior planning state. Existing
-`submitted`/`parked`/`committed` aggregates are never touched.
+`upload` needs no public origin, tunnel, or relay: the client uploads once to
+the primary provider and each secondary pulls its copy from the primary.
+Staged CARs are deleted as soon as every copy is committed, so the disk
+footprint stays near `--pack-target-size`, not the size of the migration.
 
-### Single-asset vs multi-asset
+### Legacy provider-pull path (`--legacy-pull`)
 
-The quickstart runs the **single-asset** path: each source CID becomes one
-passthrough sub-piece pulled straight from the gateway, with no staging disk.
-Use the **multi-asset** path when source CIDs are smaller than the provider's
-`Min Piece Size` or you want fewer on-chain pieces per source CID. It replaces
-step 2 with a plan that defers packing, then assembles multi-root CARs on disk:
+Before `upload`, submission worked the other way around: the client served
+pieces over public HTTPS and the provider pulled them (`plan` → `redirect-serve`
+or `serve --ingress` → `pdp-submit`). That path requires a publicly reachable
+origin, which most consumer networks cannot provide. It remains available for
+self-hosting operators — every ingress command now requires the explicit
+`--legacy-pull` flag:
 
 ```bash
-ipfs2foc plan --cids cids.txt --db migrate.db --no-auto-pack
-ipfs2foc pack-cars --db migrate.db --car-store /var/foc-cars --pack-target-size 512MiB
+ipfs2foc plan --cids cids.txt --db migrate.db
+ipfs2foc redirect-serve --db migrate.db --port 4322 --ingress cloudflared --legacy-pull
+ipfs2foc pdp-submit --db migrate.db --data-set-id <data-set-id> \
+  --source-base https://<public-host> --legacy-pull
 ```
 
+`plan` is **INSERT-only**: re-running it after appending CIDs adds new
+sub-pieces and aggregates without rewriting prior planning state. Existing
+`submitted`/`parked`/`committed` aggregates are never touched.
+
 [`docs/personas.md`](docs/personas.md) maps disk, bandwidth, and time budgets to
-concrete knob settings for both paths.
+concrete knob settings.
 
 ## Commands
 
@@ -199,6 +187,15 @@ ipfs2foc probe <cid> [--gateway https://gateway.pinata.cloud]...
 # Compute one PieceCID v2
 ipfs2foc commp <cid> [--gateway URL]...
 
+# Direct upload (PRIVATE_KEY env): download, pack multi-root CARs, stream each
+# straight to the providers, and batch addPieces before the provider's
+# parked-piece GC window closes. No public origin, relay, or ingress required.
+ipfs2foc upload [--cids cids.txt] --car-store <dir> [--db migrate.db] [--gateway URL]... \
+  [--network mainnet|calibration] [--rpc-url URL] [--copies 2] \
+  [--provider-id <id>]... [--data-set-id <id>]... \
+  [--pack-target-size 1000MiB] [--concurrency 8] [--fetch-concurrency 4] \
+  [--assumed-window-minutes 60]
+
 # Full pipeline: commitments + aggregate packing into a SQLite DB.
 # Default auto-wraps each source CID as a passthrough sub-piece. Pass
 # --no-auto-pack to defer sub-piece assembly to `pack-cars` (multi-asset).
@@ -228,7 +225,7 @@ ipfs2foc analyze [--cids cids.txt] [--db migrate.db] [--car-store <dir>] [--gate
 # Background daemon + browser console (start/pause/resume, add CIDs, add gateways)
 ipfs2foc serve [--db migrate.db] [--cids cids.txt] [--gateway URL]... \
   [--port 4321] [--network mainnet|calibration] [--rpc-url URL] [--max-base-fee N] \
-  [--app-dir <dir>] [--ingress cloudflared | --public-base https://<host>]
+  [--app-dir <dir>] [--ingress cloudflared | --public-base https://<host>] [--legacy-pull]
 
 # Current network base fee and whether to pause submission
 ipfs2foc gas [--network mainnet|calibration] [--rpc-url URL] [--max-base-fee 1000000]
@@ -236,7 +233,7 @@ ipfs2foc gas [--network mainnet|calibration] [--rpc-url URL] [--max-base-fee 100
 # Sub-piece server: GET /piece/{pcidv2} -> 302 to the gateway CAR for a
 # passthrough sub-piece, or byte-serves the assembled CAR file for a
 # multi-asset sub-piece.
-ipfs2foc redirect-serve [--db migrate.db] [--port 4322] [--ingress funnel|cloudflared]
+ipfs2foc redirect-serve [--db migrate.db] [--port 4322] [--ingress funnel|cloudflared] --legacy-pull
 
 # Provision a new FWSS data set with withIPFSIndexing (PRIVATE_KEY env)
 ipfs2foc create-data-set --provider-id <id> \
@@ -248,7 +245,8 @@ ipfs2foc create-data-set --provider-id <id> \
 ipfs2foc pdp-submit --db migrate.db --data-set-id <id> \
   (--source-base https://<public-host> | --source-relay https://<relay-base>) \
   [--network mainnet|calibration] [--rpc-url URL] \
-  [--max-in-flight 4] [--max-base-fee 1000000] [--pull-batch 32] [--poll-seconds 15]
+  [--max-in-flight 4] [--max-base-fee 1000000] [--pull-batch 32] [--poll-seconds 15] \
+  --legacy-pull
 
 # Verification report: reconcile a run against the data set's on-chain pieces
 ipfs2foc report --db migrate.db --data-set-id <id> \
diff --git a/packages/cli/src/db.ts b/packages/cli/src/db.ts
index 2b41c72..48a1327 100644
--- a/packages/cli/src/db.ts
+++ b/packages/cli/src/db.ts
@@ -121,6 +121,39 @@ export interface AggregateRow {
   error: string | null
 }
 
+/**
+ * Direct-upload lifecycle, per (sub-piece, provider) pair — the `upload`
+ * command stores the same CAR on every provider copy independently.
+ *
+ *   parked           the provider holds the bytes; nothing on-chain. Curio
+ *                    garbage-collects parked pieces after an undiscoverable
+ *                    window (~2h default), so parked_at drives the flush timer
+ *   add_unconfirmed  an addPieces batch containing this piece was attempted;
+ *                    outcome unknown. Never auto-reset into a blind re-add
+ *                    (same invariant as AggregateStatus)
+ *   committed        addPieces landed; data_set_id/piece_id/tx_hash are final
+ *   collected        the provider garbage-collected the parked piece before
+ *                    commit; the CAR must be stored again
+ *   failed           store or commit failed for a non-GC reason
+ */
+export type UploadStatus = 'parked' | 'add_unconfirmed' | 'committed' | 'collected' | 'failed'
+
+export type UploadRole = 'primary' | 'secondary'
+
+export interface UploadRow {
+  subPieceCid: string
+  providerId: string
+  role: UploadRole
+  dataSetId: string | null
+  status: UploadStatus
+  /** ISO timestamp of store() completion — the moment the GC clock starts. */
+  parkedAt: string
+  txHash: string | null
+  pieceId: string | null
+  committedAt: string | null
+  error: string | null
+}
+
 export class MigrationDB {
   #db: DatabaseSync
   /** The sqlite file path this instance opened, so callers can echo it back. */
@@ -226,6 +259,26 @@ export class MigrationDB {
         created_at      TEXT NOT NULL,
         updated_at      TEXT NOT NULL
       );
+      CREATE TABLE IF NOT EXISTS uploads (
+        sub_piece_cid TEXT NOT NULL,
+        provider_id   TEXT NOT NULL,
+        role          TEXT NOT NULL,
+        data_set_id   TEXT,
+        status        TEXT NOT NULL DEFAULT 'parked',
+        parked_at     TEXT NOT NULL,
+        tx_hash       TEXT,
+        piece_id      TEXT,
+        committed_at  TEXT,
+        error         TEXT,
+        updated_at    TEXT NOT NULL,
+        PRIMARY KEY (sub_piece_cid, provider_id),
+        FOREIGN KEY (sub_piece_cid) REFERENCES sub_pieces(sub_piece_cid)
+      );
+      CREATE TABLE IF NOT EXISTS provider_windows (
+        provider_id       TEXT PRIMARY KEY,
+        assumed_window_ms INTEGER NOT NULL,
+        updated_at        TEXT NOT NULL
+      );
     `)
   }
 
@@ -917,15 +970,7 @@ export class MigrationDB {
       )
       .get(subPieceCid) as Record<string, unknown> | undefined
     if (row == null) return null
-    return {
-      subPieceCid: String(row.sub_piece_cid),
-      assembledCarLength: Number(row.assembled_car_length),
-      assembledSha256: row.assembled_sha256 == null ? null : String(row.assembled_sha256),
-      targetSizeBytes: Number(row.target_size_bytes),
-      carPath: row.car_path == null ? null : String(row.car_path),
-      url: row.url == null ? null : String(row.url),
-      status: row.status as SubPieceStatus,
-    }
+    return toSubPieceRow(row)
   }
 
   /** Sub-pieces in the given status. */
@@ -936,18 +981,7 @@ export class MigrationDB {
                 car_path, url, status FROM sub_pieces WHERE status = ? ORDER BY sub_piece_cid`
       )
       .all(status)
-    return rows.map((r) => {
-      const row = r as Record<string, unknown>
-      return {
-        subPieceCid: String(row.sub_piece_cid),
-        assembledCarLength: Number(row.assembled_car_length),
-        assembledSha256: row.assembled_sha256 == null ? null : String(row.assembled_sha256),
-        targetSizeBytes: Number(row.target_size_bytes),
-        carPath: row.car_path == null ? null : String(row.car_path),
-        url: row.url == null ? null : String(row.url),
-        status: row.status as SubPieceStatus,
-      }
-    })
+    return rows.map(toSubPieceRow)
   }
 
   /** Sub-piece member CIDs locked into a planned sub-piece (cannot be re-packed). */
@@ -1022,6 +1056,181 @@ export class MigrationDB {
       .filter((p): p is string => p != null && p !== '')
   }
 
+  // ---- direct-upload state (uploads / provider_windows) ----
+
+  /** Built, file-backed sub-pieces with no live upload row for this provider. */
+  subPiecesNeedingUpload(providerId: string): SubPieceRow[] {
+    const rows = this.#db
+      .prepare(
+        `SELECT sp.sub_piece_cid, sp.assembled_car_length, sp.assembled_sha256,
+                sp.target_size_bytes, sp.car_path, sp.url, sp.status
+         FROM sub_pieces sp
+         WHERE sp.status = 'built' AND sp.car_path IS NOT NULL
+           AND NOT EXISTS (
+             SELECT 1 FROM uploads u
+             WHERE u.sub_piece_cid = sp.sub_piece_cid
+               AND u.provider_id = ?
+               AND u.status IN ('parked', 'add_unconfirmed', 'committed')
+           )
+         ORDER BY sp.created_at`
+      )
+      .all(providerId)
+    return rows.map(toSubPieceRow)
+  }
+
+  /** Record a successful store(): the provider holds the bytes, GC clock starts. */
+  recordUploadParked(subPieceCid: string, providerId: string, role: UploadRole, dataSetId: string | null): void {
+    const now = new Date().toISOString()
+    this.#db
+      .prepare(
+        `INSERT INTO uploads (sub_piece_cid, provider_id, role, data_set_id, status, parked_at, updated_at)
+         VALUES (?, ?, ?, ?, 'parked', ?, ?)
+         ON CONFLICT (sub_piece_cid, provider_id) DO UPDATE SET
+           role = excluded.role, data_set_id = excluded.data_set_id,
+           status = 'parked', parked_at = excluded.parked_at,
+           tx_hash = NULL, piece_id = NULL, committed_at = NULL, error = NULL,
+           updated_at = excluded.updated_at`
+      )
+      .run(subPieceCid, providerId, role, dataSetId, now, now)
+  }
+
+  /** Parked uploads for one provider, oldest parked first (the flush batch). */
+  parkedUploads(providerId: string): UploadRow[] {
+    return this.uploadsByStatus(providerId, 'parked')
+  }
+
+  /**
+   * One transactional UPDATE applied to each (sub_piece_cid, provider_id) in
+   * the batch. `set` is a compile-time constant fragment, never caller input.
+   */
+  #updateUploadsBatch(subPieceCids: string[], providerId: string, set: string, params: unknown[]): void {
+    const now = new Date().toISOString()
+    const stmt = this.#db.prepare(
+      `UPDATE uploads SET ${set}, updated_at = ? WHERE sub_piece_cid = ? AND provider_id = ?`
+    )
+    this.#db.exec('BEGIN')
+    try {
+      for (const cid of subPieceCids) stmt.run(...(params as string[]), now, cid, providerId)
+      this.#db.exec('COMMIT')
+    } catch (err) {
+      this.#db.exec('ROLLBACK')
+      throw err
+    }
+  }
+
+  /**
+   * Durable breadcrumb set immediately before the addPieces attempt, so a
+   * crash mid-commit is never auto-resolved into a blind re-add.
+   */
+  markUploadsAddUnconfirmed(subPieceCids: string[], providerId: string): void {
+    this.#updateUploadsBatch(subPieceCids, providerId, `status = 'add_unconfirmed'`, [])
+  }
+
+  markUploadTxSubmitted(subPieceCids: string[], providerId: string, txHash: string): void {
+    this.#updateUploadsBatch(subPieceCids, providerId, 'tx_hash = ?', [txHash])
+  }
+
+  markUploadCommitted(
+    subPieceCid: string,
+    providerId: string,
+    info: { dataSetId: string; pieceId: string; txHash: string }
+  ): void {
+    const now = new Date().toISOString()
+    this.#db
+      .prepare(
+        `UPDATE uploads SET status = 'committed', data_set_id = ?, piece_id = ?, tx_hash = ?,
+           committed_at = ?, error = NULL, updated_at = ?
+         WHERE sub_piece_cid = ? AND provider_id = ?`
+      )
+      .run(info.dataSetId, info.pieceId, info.txHash, now, now, subPieceCid, providerId)
+  }
+
+  /**
+   * Return add_unconfirmed rows to parked after a re-verify confirmed the
+   * provider still holds the bytes. Deliberately leaves parked_at untouched:
+   * the GC clock started at store() time and a failed commit does not reset it.
+   */
+  revertUploadsToParked(subPieceCids: string[], providerId: string): void {
+    this.#updateUploadsBatch(subPieceCids, providerId, `status = 'parked'`, [])
+  }
+
+  /** The provider garbage-collected the parked piece; the CAR must be stored again. */
+  markUploadCollected(subPieceCid: string, providerId: string): void {
+    this.#db
+      .prepare(
+        `UPDATE uploads SET status = 'collected', updated_at = ?
+         WHERE sub_piece_cid = ? AND provider_id = ?`
+      )
+      .run(new Date().toISOString(), subPieceCid, providerId)
+  }
+
+  /**
+   * Upsert, not update: a secondary whose pull fails has no uploads row yet —
+   * the failure record is the first thing written for that (piece, provider).
+   */
+  markUploadFailed(subPieceCid: string, providerId: string, role: UploadRole, error: string): void {
+    const now = new Date().toISOString()
+    this.#db
+      .prepare(
+        `INSERT INTO uploads (sub_piece_cid, provider_id, role, status, parked_at, error, updated_at)
+         VALUES (?, ?, ?, 'failed', ?, ?, ?)
+         ON CONFLICT (sub_piece_cid, provider_id) DO UPDATE SET
+           status = 'failed', error = excluded.error, updated_at = excluded.updated_at`
+      )
+      .run(subPieceCid, providerId, role, now, error, now)
+  }
+
+  uploadsByStatus(providerId: string, status: UploadStatus): UploadRow[] {
+    const rows = this.#db
+      .prepare(
+        `SELECT sub_piece_cid, provider_id, role, data_set_id, status, parked_at,
+                tx_hash, piece_id, committed_at, error
+         FROM uploads WHERE provider_id = ? AND status = ? ORDER BY parked_at`
+      )
+      .all(providerId, status)
+    return rows.map(toUploadRow)
+  }
+
+  /** Sub-piece CARs safe to evict: committed on every provider that holds them. */
+  carPathsFullyCommitted(): string[] {
+    const rows = this.#db
+      .prepare(
+        `SELECT sp.car_path FROM sub_pieces sp
+         WHERE sp.car_path IS NOT NULL
+           AND EXISTS (SELECT 1 FROM uploads u WHERE u.sub_piece_cid = sp.sub_piece_cid)
+           AND NOT EXISTS (
+             SELECT 1 FROM uploads u WHERE u.sub_piece_cid = sp.sub_piece_cid
+               AND u.status != 'committed'
+           )`
+      )
+      .all()
+    return rows.map((r) => (r as { car_path: string }).car_path)
+  }
+
+  /**
+   * Per-provider GC-window estimate. Reads fall back to the caller's default;
+   * writes only ever lower the stored value (a successful run proves nothing
+   * about the window being longer).
+   */
+  providerWindowMs(providerId: string, defaultMs: number): number {
+    const row = this.#db
+      .prepare(`SELECT assumed_window_ms FROM provider_windows WHERE provider_id = ?`)
+      .get(providerId) as { assumed_window_ms: number } | undefined
+    return row == null ? defaultMs : Math.min(row.assumed_window_ms, defaultMs)
+  }
+
+  lowerProviderWindow(providerId: string, windowMs: number): void {
+    this.#db
+      .prepare(
+        `INSERT INTO provider_windows (provider_id, assumed_window_ms, updated_at)
+         VALUES (?, ?, ?)
+         ON CONFLICT (provider_id) DO UPDATE SET
+           assumed_window_ms = MIN(provider_windows.assumed_window_ms, excluded.assumed_window_ms),
+           updated_at = excluded.updated_at`
+      )
+      .run(providerId, windowMs, new Date().toISOString())
+  }
+
   /**
    * Persist the browser-granted session key (single active session per DB).
    * The key signs only FWSS CreateDataSet/AddPieces typed-data, is time-boxed
@@ -1091,6 +1300,35 @@ export interface SessionKeyRow {
   expiresAt: number
 }
 
+function toSubPieceRow(r: unknown): SubPieceRow {
+  const row = r as Record<string, unknown>
+  return {
+    subPieceCid: String(row.sub_piece_cid),
+    assembledCarLength: Number(row.assembled_car_length),
+    assembledSha256: row.assembled_sha256 == null ? null : String(row.assembled_sha256),
+    targetSizeBytes: Number(row.target_size_bytes),
+    carPath: row.car_path == null ? null : String(row.car_path),
+    url: row.url == null ? null : String(row.url),
+    status: row.status as SubPieceStatus,
+  }
+}
+
+function toUploadRow(r: unknown): UploadRow {
+  const row = r as Record<string, unknown>
+  return {
+    subPieceCid: String(row.sub_piece_cid),
+    providerId: String(row.provider_id),
+    role: row.role as UploadRole,
+    dataSetId: row.data_set_id == null ? null : String(row.data_set_id),
+    status: row.status as UploadStatus,
+    parkedAt: String(row.parked_at),
+    txHash: row.tx_hash == null ? null : String(row.tx_hash),
+    pieceId: row.piece_id == null ? null : String(row.piece_id),
+    committedAt: row.committed_at == null ? null : String(row.committed_at),
+    error: row.error == null ? null : String(row.error),
+  }
+}
+
 function toPieceRow(r: unknown): PieceRow {
   const row = r as Record<string, unknown>
   return {
diff --git a/packages/cli/src/direct-upload.ts b/packages/cli/src/direct-upload.ts
new file mode 100644
index 0000000..a97d195
--- /dev/null
+++ b/packages/cli/src/direct-upload.ts
@@ -0,0 +1,429 @@
+/**
+ * Direct upload: stream locally packed CARs straight to storage providers and
+ * batch the on-chain adds (issue #70). This replaces provider-pull ingress for
+ * users with no public origin: 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 unlike pdp-submit 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 { findPiece } from '@filoz/synapse-core/sp'
+import { calibration, mainnet, Synapse } from '@filoz/synapse-sdk'
+import { CID } from 'multiformats/cid'
+import { type Hex, http } from 'viem'
+import { privateKeyToAccount } from 'viem/accounts'
+import type { MigrationDB } from './db.ts'
+import { resolveRpcUrl } from './gas.ts'
+import {
+  collectedCidFromError,
+  DEFAULT_ASSUMED_WINDOW_MS,
+  lowerWindowOnGc,
+  MAX_ADD_PIECES_BATCH,
+  marginFromConfirmations,
+  shouldFlush,
+} from './gc-window.ts'
+import { formatBytes, formatDuration, Timer } from './metrics.ts'
+import { log } from './util.ts'
+
+export interface DirectUploadOptions {
+  network: 'calibration' | 'mainnet'
+  rpcUrl?: string
+  privateKey?: Hex
+  /** Number of provider copies (contexts). Default 2: primary + one secondary. */
+  copies?: number
+  /** Pin specific providers instead of SDK selection. */
+  providerIds?: bigint[]
+  /** Reuse existing data sets instead of creating new ones. */
+  dataSetIds?: bigint[]
+  /** Starting GC-window guess; persisted per-provider lowering still applies. */
+  assumedWindowMs?: number
+  /** Data-set metadata. Defaults to requesting IPFS indexing/IPNI announce. */
+  dataSetMetadata?: Record<string, string>
+}
+
+/** 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 }>; 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, rpcUrl: string): 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>
+}
+
+export const defaultDirectUploadDeps: DirectUploadDeps = {
+  async setup(opts, rpcUrl) {
+    if (opts.privateKey == null) {
+      throw new Error('direct upload requires PRIVATE_KEY')
+    }
+    const chain = opts.network === 'mainnet' ? mainnet : calibration
+    const account = privateKeyToAccount(opts.privateKey)
+    const synapse = await Synapse.create({ account, transport: http(rpcUrl), chain, source: null })
+    const contexts = await synapse.storage.createContexts({
+      copies: opts.copies ?? 2,
+      ...(opts.providerIds == null ? {} : { providerIds: opts.providerIds }),
+      ...(opts.dataSetIds == null ? {} : { dataSetIds: opts.dataSetIds }),
+      metadata: opts.dataSetMetadata ?? { withIPFSIndexing: '' },
+    })
+    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, retry: false })
+              return true
+            } catch {
+              return false
+            }
+          },
+        }
+      }),
+    }
+  },
+  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(`warn: failed to evict cached CAR ${path}: ${(err as Error).message}`)
+      }
+    }
+  },
+}
+
+export interface DirectUploadSummary {
+  network: string
+  providers: Array<{
+    providerId: string
+    role: 'primary' | 'secondary'
+    dataSetId: string | null
+    committed: number
+    collected: number
+    failed: number
+    flushes: number
+    assumedWindowMs: number
+  }>
+  storedBytes: number
+  evictedCars: number
+}
+
+export async function runDirectUpload(
+  db: MigrationDB,
+  opts: DirectUploadOptions,
+  deps: DirectUploadDeps = defaultDirectUploadDeps
+): Promise<DirectUploadSummary> {
+  const rpcUrl = resolveRpcUrl({ rpcUrl: opts.rpcUrl, network: opts.network })
+  const { contexts } = await deps.setup(opts, rpcUrl)
+  const [primary, ...secondaries] = contexts
+  if (primary == null) throw new Error('no primary storage context')
+
+  log(
+    `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(`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 (same invariant as pdp-submit).
+    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,
+        })
+      })
+      log(`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(`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(`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(
+          `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(`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.carPathsFullyCommitted()) {
+      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 reason = shouldFlush({
+          batchSize: parked.length,
+          oldestParkedAtMs: parked.length === 0 ? null : Date.parse(parked[0].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, ctx)
+  }
+
+  // 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.
+  for (;;) {
+    const pending = db.subPiecesNeedingUpload(primary.providerId)
+    const next = pending[0]
+    if (next == null) break
+    if (next.carPath == null) {
+      // subPiecesNeedingUpload selects car_path IS NOT NULL; reaching this is a query bug.
+      throw new Error(`sub-piece ${next.subPieceCid} has no local CAR path`)
+    }
+
+    const storeTimer = new Timer()
+    const stored = await storeCar(primary, deps, next.carPath, next.subPieceCid)
+    storedBytes += stored.size
+    db.recordUploadParked(next.subPieceCid, primary.providerId, 'primary', primary.dataSetId)
+    log(
+      `parked ${next.subPieceCid} (${formatBytes(stored.size)}) on primary ${primary.providerId} ` +
+        `in ${formatDuration(storeTimer.stop())}`
+    )
+
+    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++) {
+    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) break
+    log(`retrying ${needsRetry.length} piece(s) that did not land (attempt ${attempt + 1})`)
+    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(`error: collected ${u.subPieceCid} has no local CAR; cannot re-store`)
+        continue
+      }
+      const stored = await storeCar(ctx, deps, sub.carPath, sub.subPieceCid)
+      storedBytes += stored.size
+      db.recordUploadParked(sub.subPieceCid, ctx.providerId, u.role, ctx.dataSetId)
+    }
+    await maybeFlush(true)
+  }
+
+  await evictCommitted()
+
+  const summary: DirectUploadSummary = {
+    network: opts.network,
+    providers: contexts.map((ctx, i) => ({
+      providerId: ctx.providerId,
+      role: i === 0 ? 'primary' : 'secondary',
+      dataSetId: latestDataSetId(db, ctx),
+      committed: db.uploadsByStatus(ctx.providerId, 'committed').length,
+      collected: db.uploadsByStatus(ctx.providerId, 'collected').length,
+      failed: db.uploadsByStatus(ctx.providerId, 'failed').length,
+      flushes: flushCounts.get(ctx.providerId) ?? 0,
+      assumedWindowMs: windowFor(ctx),
+    })),
+    storedBytes,
+    evictedCars: evictedPaths.size,
+  }
+  log(`direct upload finished in ${formatDuration(runTimer.stop())}: ${formatBytes(storedBytes)} stored`)
+  return summary
+}
+
+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) })
+  return { size: result.size }
+}
+
+/** Resolve every add_unconfirmed row by probing the provider for the bytes. */
+async function reconcileUnconfirmed(db: MigrationDB, ctx: UploadContextLike): Promise<void> {
+  for (const u of db.uploadsByStatus(ctx.providerId, 'add_unconfirmed')) {
+    if (await ctx.hasPiece(CID.parse(u.subPieceCid))) {
+      db.revertUploadsToParked([u.subPieceCid], ctx.providerId)
+      log(`resume: ${u.subPieceCid} still parked on provider ${ctx.providerId}; re-queued for commit`)
+    } else {
+      db.markUploadCollected(u.subPieceCid, ctx.providerId)
+      log(`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(`parked ${subPieceCid} on secondary ${secondary.providerId} (pulled from primary)`)
+    } else {
+      db.markUploadFailed(subPieceCid, secondary.providerId, 'secondary', 'secondary pull failed')
+      log(`warn: secondary ${secondary.providerId} failed to pull ${subPieceCid}`)
+    }
+  } catch (err) {
+    db.markUploadFailed(subPieceCid, secondary.providerId, 'secondary', (err as Error).message)
+    log(`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')
+  return committed.length > 0 ? committed[committed.length - 1].dataSetId : ctx.dataSetId
+}
diff --git a/packages/cli/src/gc-window.ts b/packages/cli/src/gc-window.ts
new file mode 100644
index 0000000..0c94e58
--- /dev/null
+++ b/packages/cli/src/gc-window.ts
@@ -0,0 +1,110 @@
+/**
+ * Flush scheduling and GC-window estimation for the direct-upload flow
+ * (issue #70).
+ *
+ * Curio garbage-collects parked pieces that have not been added on chain.
+ * The window is roughly 2h, SP-configurable, and not discoverable, so the
+ * client flushes before its own best guess expires. The guess starts
+ * conservative and only ever moves down: a successful run proves you stayed
+ * under the window, not that the window is longer, while a detected GC proves
+ * the window is at most the collected piece's parked age.
+ *
+ * The costs are asymmetric — an early flush costs one extra transaction, a
+ * collected piece costs a full re-upload of up to MAX_UPLOAD_SIZE bytes — so
+ * every rounding here rounds toward flushing sooner.
+ *
+ * Everything in this module is pure; the orchestrator in direct-upload.ts
+ * owns the clock and the DB.
+ */
+
+/**
+ * FWSS `addPieces` batch cap. synapse-core >= 1.x exports this as
+ * `SIZE_CONSTANTS.MAX_ADD_PIECES_BATCH_SIZE`; the pinned 0.5.x does not, so
+ * it lives here until the SDK is bumped.
+ */
+export const MAX_ADD_PIECES_BATCH = 40
+
+/**
+ * Starting guess for a provider's GC window. Half of Curio's ~2h default:
+ * under-guessing costs early flushes, over-guessing costs re-uploads.
+ */
+export const DEFAULT_ASSUMED_WINDOW_MS = 60 * 60_000
+
+/** Never trust a window guess below this — flushing degenerates to per-piece adds. */
+export const MIN_WINDOW_MS = 5 * 60_000
+
+/** Floor for the flush margin, covering commit build + confirmation. */
+export const MIN_MARGIN_MS = 10 * 60_000
+
+export type FlushReason = 'batch-full' | 'window' | 'drained'
+
+export interface FlushInput {
+  /** Parked pieces currently awaiting commit on this provider. */
+  batchSize: number
+  /** Epoch ms of the oldest parked piece, or null when the batch is empty. */
+  oldestParkedAtMs: number | null
+  nowMs: number
+  assumedWindowMs: number
+  marginMs: number
+  /** True when no further pieces will be parked (source drained). */
+  drained: boolean
+}
+
+/** Decide whether the parked batch must be committed now, and why. */
+export function shouldFlush(input: FlushInput): FlushReason | null {
+  if (input.batchSize <= 0) return null
+  if (input.batchSize >= MAX_ADD_PIECES_BATCH) return 'batch-full'
+  if (
+    input.oldestParkedAtMs != null &&
+    input.nowMs >= input.oldestParkedAtMs + input.assumedWindowMs - input.marginMs
+  ) {
+    return 'window'
+  }
+  if (input.drained) return 'drained'
+  return null
+}
+
+/**
+ * Margin derived from confirmations observed earlier in the same run, per
+ * issue #70 — a constant cannot cover a congested chain. Doubling the worst
+ * observed commit duration leaves room for one full retry; the floor covers
+ * the first flush of a run, before any observation exists.
+ */
+export function marginFromConfirmations(observedCommitMs: number[], floorMs: number = MIN_MARGIN_MS): number {
+  if (observedCommitMs.length === 0) return floorMs
+  return Math.max(floorMs, 2 * Math.max(...observedCommitMs))
+}
+
+/**
+ * Lower the assumed window after a detected GC. The collected piece survived
+ * less than `collectedParkedAgeMs`, so the true window is at most that; 3/4
+ * of it keeps the next guess strictly inside. Never raises the estimate.
+ */
+export function lowerWindowOnGc(assumedWindowMs: number, collectedParkedAgeMs: number): number {
+  const fromEvidence = Math.floor((collectedParkedAgeMs * 3) / 4)
+  return Math.max(MIN_WINDOW_MS, Math.min(assumedWindowMs, fromEvidence))
+}
+
+/**
+ * The upload rate below which one max-size piece cannot be parked and
+ * confirmed inside the window — the flow cannot work at all under this and
+ * the caller should retain locally and upload in a burst instead (rvagg:
+ * "don't dribble this in").
+ */
+export function bandwidthFloorBytesPerSec(pieceSizeBytes: number, assumedWindowMs: number, marginMs: number): number {
+  const usableMs = Math.max(1, assumedWindowMs - marginMs)
+  return Math.ceil((pieceSizeBytes * 1000) / usableMs)
+}
+
+/**
+ * Parse Curio's GC rejection out of a failed addPieces/commit error. Returns
+ * the named sub-piece CID or null when the failure is something else. The
+ * message conflates "not parked" with "wrong service", so the caller should
+ * treat the result as evidence, not proof, and re-verify the whole batch
+ * (Curio returns on the FIRST miss; several collected pieces surface one at
+ * a time).
+ */
+export function collectedCidFromError(message: string): string | null {
+  const match = message.match(/subPiece CID ([A-Za-z0-9]+) not found or does not belong to service/)
+  return match ? match[1] : null
+}
diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts
index 3b728de..eae1676 100644
--- a/packages/cli/src/index.ts
+++ b/packages/cli/src/index.ts
@@ -13,6 +13,7 @@
  *   gas     [--network N] [opts]            Current network base fee and whether to pause.
  *   redirect-serve [--db FILE] [--port N] [--ingress funnel|cloudflared]   GET /piece/{pcidv2} -> 302 gateway CAR.
  *   create-data-set --provider-id ID [opts] Provision a new FWSS data set with withIPFSIndexing (PRIVATE_KEY env).
+ *   upload  --cids FILE --car-store DIR [opts]  Download, pack, and upload direct to providers (PRIVATE_KEY env).
  *   pdp-submit --data-set-id ID [opts]      Pull, park, and add aggregates over PDP (PRIVATE_KEY env).
  *   report  --data-set-id ID [opts]         Reconcile a run against on-chain pieces; emit explorer links.
  *   pack-cars --car-store DIR [opts]        Assemble multi-root CARs for the multi-asset path.
@@ -29,6 +30,7 @@ import { parseArgs } from 'node:util'
 import { DEFAULT_PROBE_CONCURRENCY, DEFAULT_SAMPLE, formatAnalyzeText, runAnalyze } from './analyze.ts'
 import { runCreateDataSet } from './create-data-set.ts'
 import { MigrationDB } from './db.ts'
+import { runDirectUpload } from './direct-upload.ts'
 import { buildExportManifest } from './export-manifest.ts'
 import { classifyBaseFee, DEFAULT_MAX_BASE_FEE, getBaseFee, resolveRpcUrl } from './gas.ts'
 import { DEFAULT_GATEWAYS, probeGateway } from './gateway.ts'
@@ -49,6 +51,23 @@ import { log, parseCidList, parsePositiveInt, parseSize } from './util.ts'
 
 const DEFAULT_DB = './migrate.db'
 
+/**
+ * The provider-pull ingress paths (redirect-serve, serve --ingress,
+ * pdp-submit --source-*) are gated: they require a public HTTPS origin, which
+ * most consumer networks cannot provide, and the direct `upload` command
+ * replaces them without any ingress. They stay available for self-hosting
+ * operators who already run a reachable origin.
+ */
+function requireLegacyPull(values: { 'legacy-pull'?: boolean }, what: string): void {
+  if (values['legacy-pull'] !== true) {
+    throw new Error(
+      `${what} uses the legacy provider-pull path, which needs a public HTTPS origin. ` +
+        `The 'upload' command uploads directly and needs no ingress. ` +
+        `Pass --legacy-pull if you run a reachable origin and want provider pull anyway.`
+    )
+  }
+}
+
 const USAGE = `ipfs2foc — migrate pinned IPFS CIDs to FOC without re-chunking
 
 Usage:
@@ -63,16 +82,26 @@ Usage:
   ipfs2foc serve  [--db <file>] [--cids <file>] [--gateway URL]... [--piece-size 32GiB]
                      [--concurrency 8] [--port 4321] [--network mainnet|calibration] [--max-base-fee N]
                      [--app-dir <dir>]  (or IPFS2FOC_APP_DIR; defaults to the bundled console)
-                     [--ingress cloudflared | --public-base <https-url>]
-                     (serve also answers GET/HEAD /piece/{pcidv2}; with ingress it is the pull source)
+                     [--ingress cloudflared | --public-base <https-url>] [--legacy-pull]
+                     (serve also answers GET/HEAD /piece/{pcidv2}; with ingress it is the pull source,
+                      which requires --legacy-pull — prefer 'upload', which needs no ingress)
   ipfs2foc gas    [--network mainnet|calibration] [--rpc-url URL] [--max-base-fee N]
-  ipfs2foc redirect-serve [--db <file>] [--port 4322] [--ingress funnel|cloudflared]
+  ipfs2foc redirect-serve [--db <file>] [--port 4322] [--ingress funnel|cloudflared] [--legacy-pull]
+  ipfs2foc upload [--cids <file>] --car-store <dir> [--db <file>] [--gateway URL]...
+                     [--network mainnet|calibration] [--rpc-url URL] [--copies 2]
+                     [--provider-id <id>]... [--data-set-id <id>]...
+                     [--pack-target-size 1000MiB] [--concurrency 8] [--fetch-concurrency 4]
+                     [--assumed-window-minutes 60]  (uses PRIVATE_KEY env)
+                     (download, pack multi-root CARs, stream each straight to the providers, and
+                      batch addPieces before the provider's parked-piece GC window closes; no
+                      public origin, relay, or ingress required)
   ipfs2foc create-data-set --provider-id <id> [--network mainnet|calibration] [--cdn]
                      (uses PRIVATE_KEY env)
   ipfs2foc pdp-submit --data-set-id <id> (--source-base <https-url> | --source-relay <https-url>) [--db <file>]
                      [--network mainnet|calibration] [--max-in-flight 4] [--max-base-fee N] [--pull-batch 32]
                      [--strict-piece-size]  (refuse pieces below the provider's advertised minimum;
                      default warns and proceeds — the floor is advisory in practice)
+                     [--legacy-pull]  (required: provider pull needs a public origin; prefer 'upload')
                      (--source-base: your own redirect-serve; --source-relay: a shared stateless relay, passthrough only)
                      (uses PRIVATE_KEY env)
   ipfs2foc report --data-set-id <id> [--db <file>] [--network mainnet|calibration] [--json]
@@ -94,19 +123,24 @@ Defaults:
   network     mainnet (serve base-fee monitor off unless --network or --rpc-url given)
 
 Examples:
+  # Migrate a CID list end to end: download, pack ~1 GiB multi-root CARs, and
+  # upload straight to two providers. No public origin or tunnel needed.
+  ipfs2foc upload --cids cids.txt --car-store ./cars --network calibration
+
   # Pre-flight a gateway, then plan a CID list
   ipfs2foc probe <cid> --gateway https://trustless-gateway.link
   ipfs2foc plan --cids cids.txt
 
+  # Legacy provider-pull path (self-hosted public origin required).
   # One process carries the console and the pull source. Start it, wait for it
   # to log "ingress: ready at https://<host>", then submit against that host.
-  ipfs2foc serve --ingress cloudflared
-  ipfs2foc pdp-submit --data-set-id 42 --source-base https://<tunnel-host>
+  ipfs2foc serve --ingress cloudflared --legacy-pull
+  ipfs2foc pdp-submit --data-set-id 42 --source-base https://<tunnel-host> --legacy-pull
 
   # Or run the pull source on its own: start it first, leave it running, and
   # submit from a second shell against the host it logs.
-  ipfs2foc redirect-serve --ingress cloudflared --port 4322
-  ipfs2foc pdp-submit --data-set-id 42 --source-base https://<public-host>
+  ipfs2foc redirect-serve --ingress cloudflared --port 4322 --legacy-pull
+  ipfs2foc pdp-submit --data-set-id 42 --source-base https://<public-host> --legacy-pull
 
   # Confirm everything landed on chain
   ipfs2foc report --data-set-id 42
@@ -133,6 +167,7 @@ const KNOWN_COMMANDS = [
   'gas',
   'redirect-serve',
   'create-data-set',
+  'upload',
   'pdp-submit',
   'report',
   'pack-cars',
@@ -573,8 +608,10 @@ async function cmdPdpSubmit(argv: string[]): Promise<void> {
       'poll-seconds': { type: 'string', default: '15' },
       'pull-batch': { type: 'string', default: '32' },
       'strict-piece-size': { type: 'boolean', default: false },
+      'legacy-pull': { type: 'boolean', default: false },
     },
   })
+  requireLegacyPull(values, 'pdp-submit')
   if (values['data-set-id'] == null) {
     throw new Error('pdp-submit requires --data-set-id <id>')
   }
@@ -661,8 +698,10 @@ async function cmdRedirectServe(argv: string[]): Promise<void> {
       db: { type: 'string', default: DEFAULT_DB },
       port: { type: 'string', default: '4322' },
       ingress: { type: 'string', default: 'funnel' },
+      'legacy-pull': { type: 'boolean', default: false },
     },
   })
+  requireLegacyPull(values, 'redirect-serve')
   const db = new MigrationDB(values.db as string)
   const port = parsePositiveInt(values.port as string, '--port')
   const ingress = values.ingress as string
@@ -700,6 +739,81 @@ async function cmdGas(argv: string[]): Promise<void> {
   )
 }
 
+async function cmdUpload(argv: string[]): Promise<void> {
+  const { values } = parseArgs({
+    args: argv,
+    options: {
+      cids: { type: 'string' },
+      db: { type: 'string', default: DEFAULT_DB },
+      gateway: { type: 'string', multiple: true },
+      'car-store': { type: 'string' },
+      network: { type: 'string', default: 'mainnet' },
+      'rpc-url': { type: 'string' },
+      copies: { type: 'string', default: '2' },
+      'provider-id': { type: 'string', multiple: true },
+      'data-set-id': { type: 'string', multiple: true },
+      // 1016 MiB is the SDK's per-piece cap; 1000 MiB leaves headroom for the
+      // multi-root CAR header and framing the bin-packing weight ignores.
+      'pack-target-size': { type: 'string', default: '1000MiB' },
+      concurrency: { type: 'string', default: '8' },
+      'fetch-concurrency': { type: 'string', default: '4' },
+      'assumed-window-minutes': { type: 'string', default: '60' },
+    },
+  })
+  if (values['car-store'] == null) {
+    throw new Error('upload requires --car-store <dir> (packed CARs are staged here until committed)')
+  }
+  const network = values.network as string
+  if (network !== 'mainnet' && network !== 'calibration') {
+    throw new Error(`unknown --network ${network} (expected mainnet|calibration)`)
+  }
+  const key = process.env.PRIVATE_KEY
+  if (key == null || !/^0x[0-9a-fA-F]{64}$/.test(key)) {
+    throw new Error('set PRIVATE_KEY (0x + 64 hex) in the environment (e.g. `source .env`)')
+  }
+  const gateways = gatewaysFrom(values)
+  const packTargetBytes = Number(parseSize(values['pack-target-size'] as string))
+  const db = new MigrationDB(values.db as string)
+  try {
+    if (values.cids != null) {
+      const cids = parseCidList(await readFile(values.cids as string, 'utf8'))
+      db.addCids(cids)
+      log(`registered ${cids.length} CID(s) from ${values.cids}`)
+    }
+    // Download + commP without auto-pack: pieces stay free for the multi-root
+    // binning below rather than being wrapped as single-piece sub-pieces.
+    await runPlan(db, {
+      gateways,
+      aggregateSizeBytes: BigInt(packTargetBytes),
+      concurrency: parsePositiveInt(values.concurrency as string, '--concurrency'),
+      autoPack: false,
+    })
+    const packSummary = await runPackCars(db, {
+      gateways,
+      targetSizeBytes: packTargetBytes,
+      carStore: values['car-store'] as string,
+      fetchConcurrency: parsePositiveInt(values['fetch-concurrency'] as string, '--fetch-concurrency'),
+      skipAggregatePlanning: true,
+      buildOversizedAsSingles: true,
+    })
+    log(`packed ${packSummary.built} multi-root CAR(s) under ${values['car-store']}`)
+
+    const summary = await runDirectUpload(db, {
+      network,
+      rpcUrl: values['rpc-url'],
+      privateKey: key as `0x${string}`,
+      copies: parsePositiveInt(values.copies as string, '--copies'),
+      providerIds: (values['provider-id'] as string[] | undefined)?.map((v) => BigInt(v)),
+      dataSetIds: (values['data-set-id'] as string[] | undefined)?.map((v) => BigInt(v)),
+      assumedWindowMs:
+        parsePositiveInt(values['assumed-window-minutes'] as string, '--assumed-window-minutes') * 60_000,
+    })
+    console.log(JSON.stringify(summary, bigintJsonReplacer, 2))
+  } finally {
+    db.close()
+  }
+}
+
 async function cmdPackCars(argv: string[]): Promise<void> {
   const { values } = parseArgs({
     args: argv,
@@ -833,8 +947,13 @@ async function cmdServe(argv: string[]): Promise<void> {
       'ipfs-fallback': { type: 'boolean', default: false },
       'ipfs-fallback-mode': { type: 'string' },
       'ipfs-fallback-timeout-seconds': { type: 'string' },
+      'legacy-pull': { type: 'boolean', default: false },
     },
   })
+  // The console itself is not legacy — only serving as a provider-pull origin is.
+  if (values.ingress != null || values['public-base'] != null) {
+    requireLegacyPull(values, 'serve with --ingress/--public-base')
+  }
 
   // Network feeds /api/capabilities (and the console's command hints). The gas
   // monitor below keys off the RAW flag on purpose: it stays opt-in, so the
@@ -990,6 +1109,9 @@ async function main(): Promise<void> {
     case 'create-data-set':
       await cmdCreateDataSet(rest)
       break
+    case 'upload':
+      await cmdUpload(rest)
+      break
     case 'pdp-submit':
       await cmdPdpSubmit(rest)
       break
diff --git a/packages/cli/src/pack-cars.ts b/packages/cli/src/pack-cars.ts
index 49dbb21..e70f252 100644
--- a/packages/cli/src/pack-cars.ts
+++ b/packages/cli/src/pack-cars.ts
@@ -55,6 +55,9 @@ const DEFAULT_FETCH_CONCURRENCY = 4
 /** Maximum CAR raw size accepted by Curio's PDP pull (`PieceSizeLimit`, see Q4). */
 export const PIECE_RAW_SIZE_LIMIT = 1_069_547_520
 
+/** synapse-sdk's per-piece upload cap (`SIZE_CONSTANTS.MAX_UPLOAD_SIZE`, 1016 MiB). */
+export const SDK_MAX_UPLOAD_BYTES = 1_065_353_216
+
 export interface PackPlanInput {
   /** Source CID. */
   cid: string
@@ -309,6 +312,19 @@ export interface PackCarsOptions {
   carStore: string
   /** Bounded fan-out per assembly. Default 4. */
   fetchConcurrency?: number
+  /**
+   * When true, skip the trailing `appendAggregatesFromFreeSubPieces` pass.
+   * The direct-upload flow commits each built sub-piece as its own on-chain
+   * piece; the aggregates table only serves the provider-pull path.
+   */
+  skipAggregatePlanning?: boolean
+  /**
+   * When true, a piece too large to share a bin is built as its own
+   * single-member CAR file instead of a URL-backed passthrough sub-piece.
+   * Direct upload streams from local CARs only — a passthrough sub-piece has
+   * no `car_path` and would silently fall out of the upload work list.
+   */
+  buildOversizedAsSingles?: boolean
 }
 
 export interface PackCarsSummary {
@@ -364,6 +380,22 @@ export async function runPackCars(
   }))
   const { bins, oversizedForPacking } = planBins(inputsForBuild, target)
 
+  // Direct-upload mode: an oversized-for-bin piece still ships as a CAR file,
+  // alone in its own bin, as long as it fits one uploadable piece. Anything
+  // over the per-piece cap genuinely cannot migrate on this path.
+  const passthroughCandidates: PackPlanInput[] = []
+  if (opts.buildOversizedAsSingles === true) {
+    for (const p of oversizedForPacking) {
+      if (p.rawSize > SDK_MAX_UPLOAD_BYTES) {
+        log(`! ${p.cid} (${p.rawSize} bytes) exceeds the ${SDK_MAX_UPLOAD_BYTES}-byte upload cap; not migrated`)
+        continue
+      }
+      bins.push({ memberCids: [p.cid], totalRawSize: p.rawSize })
+    }
+  } else {
+    passthroughCandidates.push(...oversizedForPacking)
+  }
+
   const summary: PackCarsSummary = { bins: bins.length, built: 0, failed: 0, skipped: 0, failedMemberCids: [] }
   const fetchConcurrency = opts.fetchConcurrency ?? DEFAULT_FETCH_CONCURRENCY
   for (const bin of bins) {
@@ -407,7 +439,7 @@ export async function runPackCars(
   // with no sub-piece and no aggregate, silently absent from the migration.
   let passthroughAdded = 0
   const noGatewayUrl: string[] = []
-  for (const p of oversizedForPacking) {
+  for (const p of passthroughCandidates) {
     const row = piecesByCid.get(p.cid)
     if (row?.pieceCid == null || row.rawSize == null || row.url == null || row.url === '') {
       // No gateway URL (IPFS-fallback only): cannot be served by the HTTP pull
@@ -437,7 +469,7 @@ export async function runPackCars(
   // composition `plan` wrote and asks the SP to pull individual files (most
   // of which are below the provider's minimum piece size). Frozen aggregates
   // (submitted/parked/committed) are left untouched.
-  if (summary.built > 0 || passthroughAdded > 0) {
+  if ((summary.built > 0 || passthroughAdded > 0) && opts.skipAggregatePlanning !== true) {
     // Append new aggregates over the freshly built multi-asset sub-pieces.
     // No DELETE of existing aggregates — `plan` already added passthrough
     // aggregates over the source pieces, and those stay as the alternative
diff --git a/packages/cli/test/direct-upload-flow.test.ts b/packages/cli/test/direct-upload-flow.test.ts
new file mode 100644
index 0000000..0fde6a5
--- /dev/null
+++ b/packages/cli/test/direct-upload-flow.test.ts
@@ -0,0 +1,195 @@
+import assert from 'node:assert/strict'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { test } from 'node:test'
+import { MigrationDB } from '../src/db.ts'
+import {
+  type DirectUploadDeps,
+  type DirectUploadOptions,
+  runDirectUpload,
+  type UploadContextLike,
+} from '../src/direct-upload.ts'
+
+// Drives the real runDirectUpload control flow with fake providers, to lock in
+// the issue-70 guarantees: store-then-batch-commit, the add_unconfirmed
+// breadcrumb, GC detection lowering the window and re-storing only what is
+// actually gone, and CAR eviction only after every copy is committed.
+
+// Real PieceCIDs so CID.parse succeeds.
+const P1 = 'bafkzcibf3ck4uais4fgennh4hbfx5z3i6hue4xgq2cdeamtus4hjbsrjs5lf2azbxmsa'
+const P2 = 'bafkzcibewpkqwewyhz3yxutlxbpt2nkb6si5qilg4qqtzzij32uw7ammsc73a4wkgi'
+
+const OPTS: DirectUploadOptions = { network: 'calibration', copies: 2 }
+
+async function dbAt(name: string) {
+  const dir = await mkdtemp(join(tmpdir(), `foc-${name}-`))
+  return { dir, db: new MigrationDB(join(dir, 'migrate.db')) }
+}
+
+function seedBuilt(db: MigrationDB, subPieceCid: string, carPath: string) {
+  const src = `src-${subPieceCid.slice(-8)}`
+  db.addCids([src])
+  db.recordPieceSuccess(src, subPieceCid, 1024, 'g', `https://gw/ipfs/${src}?format=car`, null)
+  db.recordBuiltSubPiece({
+    subPieceCid,
+    assembledCarLength: 1024,
+    targetSizeBytes: 1024,
+    carPath,
+    assembledSha256: 'sha',
+    members: [{ cid: src, sha256: null, 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
+}
+
+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: null,
+    async store(_data, options) {
+      calls.store.push(`${providerId}:${String(options.pieceCid)}`)
+      return { pieceCid: options.pieceCid, size: 1024 }
+    },
+    async presignForCommit() {
+      return '0xfake'
+    },
+    async pull(options) {
+      calls.pull++
+      const status = b.pullFails ? 'failed' : 'complete'
+      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) {
+        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(),
+    openCar: () => new Uint8Array(8),
+    evictCar: async (path) => {
+      evicted.push(path)
+    },
+  }
+  return { deps, calls, evicted }
+}
+
+test('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)
+
+    assert.deepEqual(calls.store, [`p1:${P1}`, `p1:${P2}`])
+    assert.equal(calls.pull, 2)
+    // One drained flush per provider, both pieces in one batch.
+    assert.equal(calls.commit.get('p1'), 1)
+    assert.equal(calls.commit.get('p2'), 1)
+    for (const provider of ['p1', 'p2']) {
+      const committed = db.uploadsByStatus(provider, 'committed')
+      assert.deepEqual(committed.map((u) => u.subPieceCid).sort(), [P2, P1].sort())
+      for (const u of committed) assert.equal(u.dataSetId, '7')
+    }
+    assert.equal(evicted.length, 2)
+    assert.equal(summary.providers[0].committed, 2)
+    assert.equal(summary.providers[0].role, 'primary')
+  } finally {
+    db.close()
+    await rm(dir, { recursive: true, force: true })
+  }
+})
+
+test('secondary pull failure leaves the primary committed and the secondary failed', async () => {
+  const { dir, db } = await dbAt('du-pullfail')
+  try {
+    seedBuilt(db, P1, join(dir, 'a.car'))
+    const { deps, evicted } = fakeDeps({ pullFails: true })
+    await runDirectUpload(db, OPTS, deps)
+
+    assert.equal(db.uploadsByStatus('p1', 'committed').length, 1)
+    assert.equal(db.uploadsByStatus('p2', 'failed').length, 1)
+    // The CAR must survive: the secondary copy never landed.
+    assert.equal(evicted.length, 0)
+  } finally {
+    db.close()
+    await rm(dir, { recursive: true, force: true })
+  }
+})
+
+test('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.
+    const p1Stores = calls.store.filter((s) => s === `p1:${P1}`).length
+    assert.equal(p1Stores, 2)
+    assert.equal(calls.store.filter((s) => s === `p1:${P2}`).length, 1)
+    // Both pieces end up committed via the retry flush.
+    assert.equal(db.uploadsByStatus('p1', 'committed').length, 2)
+    // The window guess dropped below the default for the flaky provider only.
+    const defaultMs = 60 * 60_000
+    assert.ok(db.providerWindowMs('p1', defaultMs) < defaultMs)
+    assert.equal(db.providerWindowMs('p2', defaultMs), defaultMs)
+  } finally {
+    db.close()
+    await rm(dir, { recursive: true, force: true })
+  }
+})
+
+test('non-GC commit failure leaves the batch add_unconfirmed, and a later run reconciles it', 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)
+    assert.equal(db.uploadsByStatus('p1', 'add_unconfirmed').length, 1)
+
+    // Second run: the piece is still parked on the provider, so the resume
+    // reconciliation re-queues it and the commit lands — without re-storing.
+    const second = fakeDeps()
+    await runDirectUpload(db, OPTS, second.deps)
+    assert.equal(db.uploadsByStatus('p1', 'committed').length, 1)
+    assert.equal(second.calls.store.filter((s) => s.startsWith('p1:')).length, 0)
+  } finally {
+    db.close()
+    await rm(dir, { recursive: true, force: true })
+  }
+})
diff --git a/packages/cli/test/gc-window.test.ts b/packages/cli/test/gc-window.test.ts
new file mode 100644
index 0000000..4fe973d
--- /dev/null
+++ b/packages/cli/test/gc-window.test.ts
@@ -0,0 +1,73 @@
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import {
+  bandwidthFloorBytesPerSec,
+  collectedCidFromError,
+  DEFAULT_ASSUMED_WINDOW_MS,
+  lowerWindowOnGc,
+  MAX_ADD_PIECES_BATCH,
+  MIN_MARGIN_MS,
+  MIN_WINDOW_MS,
+  marginFromConfirmations,
+  shouldFlush,
+} from '../src/gc-window.ts'
+
+// Locks in the flush-scheduling rules from issue #70: every tie breaks toward
+// flushing sooner, and the window estimate only ever moves down.
+
+const base = {
+  batchSize: 1,
+  oldestParkedAtMs: 0,
+  nowMs: 0,
+  assumedWindowMs: DEFAULT_ASSUMED_WINDOW_MS,
+  marginMs: MIN_MARGIN_MS,
+  drained: false,
+}
+
+test('shouldFlush: empty batch never flushes, even drained', () => {
+  assert.equal(shouldFlush({ ...base, batchSize: 0, drained: true }), null)
+})
+
+test('shouldFlush: full batch flushes regardless of timers', () => {
+  assert.equal(shouldFlush({ ...base, batchSize: MAX_ADD_PIECES_BATCH }), 'batch-full')
+})
+
+test('shouldFlush: window expiry includes the margin', () => {
+  const edge = DEFAULT_ASSUMED_WINDOW_MS - MIN_MARGIN_MS
+  assert.equal(shouldFlush({ ...base, nowMs: edge - 1 }), null)
+  assert.equal(shouldFlush({ ...base, nowMs: edge }), 'window')
+})
+
+test('shouldFlush: drained flushes a partial batch', () => {
+  assert.equal(shouldFlush({ ...base, drained: true }), 'drained')
+})
+
+test('marginFromConfirmations: floor without observations, 2x worst with', () => {
+  assert.equal(marginFromConfirmations([]), MIN_MARGIN_MS)
+  assert.equal(marginFromConfirmations([1_000, 2_000]), MIN_MARGIN_MS)
+  const slow = 20 * 60_000
+  assert.equal(marginFromConfirmations([1_000, slow]), 2 * slow)
+})
+
+test('lowerWindowOnGc: lowers from evidence, never raises, floors', () => {
+  const hourMs = 60 * 60_000
+  assert.equal(lowerWindowOnGc(hourMs, 40 * 60_000), 30 * 60_000)
+  // Evidence above the current guess must not raise it.
+  assert.equal(lowerWindowOnGc(hourMs, 10 * hourMs), hourMs)
+  assert.equal(lowerWindowOnGc(hourMs, 0), MIN_WINDOW_MS)
+})
+
+test('bandwidthFloorBytesPerSec: matches the issue-70 example order of magnitude', () => {
+  // ~1016 MiB over 2h minus 10 min margin ≈ 1.3 Mbit/s ≈ 162 KB/s.
+  const floor = bandwidthFloorBytesPerSec(1_065_353_216, 2 * 60 * 60_000, 10 * 60_000)
+  assert.ok(floor > 100_000 && floor < 250_000, `unexpected floor ${floor}`)
+})
+
+test('collectedCidFromError: parses the Curio GC rejection and nothing else', () => {
+  const cid = 'bafkzcibf3ck4uais4fgennh4hbfx5z3i6hue4xgq2cdeamtus4hjbsrjs5lf2azbxmsa'
+  assert.equal(
+    collectedCidFromError(`Failed to process request: subPiece CID ${cid} not found or does not belong to service foo`),
+    cid
+  )
+  assert.equal(collectedCidFromError('insufficient funds'), null)
+})