diff --git a/packages/cli/src/aggregate-stream.ts b/packages/cli/src/aggregate-stream.ts
new file mode 100644
index 0000000..257a1f6
--- /dev/null
+++ b/packages/cli/src/aggregate-stream.ts
@@ -0,0 +1,164 @@
+/**
+ * Stream an indexed aggregate's full byte stream to a provider pull.
+ *
+ * The response is the exact byte layout the aggregate root commits to (see
+ * `ipfs2foc-core/indexed-aggregate`): each sub-piece's canonical CAR at its
+ * unpadded offset with zero fill, then the embedded data segment index. The
+ * provider recomputes the piece commitment over these bytes on pull, so every
+ * sub-piece is hash-verified here as it passes through — a drifted gateway
+ * response aborts the stream instead of feeding the provider bytes that can
+ * only fail its commP check after the whole download.
+ */
+
+import { once } from 'node:events'
+import { createReadStream } from 'node:fs'
+import type { ServerResponse } from 'node:http'
+import * as Hasher from '@web3-storage/data-segment/multihash'
+import { buildIndexedAggregate, type IndexedAggregate } from 'ipfs2foc-core/indexed-aggregate'
+import type { CID } from 'multiformats/cid'
+import * as Raw from 'multiformats/codecs/raw'
+import * as Link from 'multiformats/link'
+import type { MigrationDB } from './db.ts'
+import { fetchCanonicalCar } from './gateway-blocks.ts'
+import { log } from './util.ts'
+
+const ZERO_CHUNK = new Uint8Array(64 * 1024)
+
+async function writeChunk(res: ServerResponse, chunk: Uint8Array): Promise<void> {
+  if (!res.write(chunk)) {
+    await once(res, 'drain')
+  }
+}
+
+async function writeZeros(res: ServerResponse, length: number): Promise<void> {
+  let remaining = length
+  while (remaining > 0) {
+    const n = Math.min(remaining, ZERO_CHUNK.length)
+    await writeChunk(res, n === ZERO_CHUNK.length ? ZERO_CHUNK : ZERO_CHUNK.subarray(0, n))
+    remaining -= n
+  }
+}
+
+/**
+ * Recompute the indexed aggregate layout for a planned aggregate row. The
+ * layout is deterministic from the member set (the builder re-sorts and picks
+ * the smallest fitting deal size), so nothing beyond the members needs to be
+ * persisted; the recomputed root is checked against the stored one before any
+ * byte is served.
+ */
+export function layoutForAggregate(db: MigrationDB, idx: number, expectedRoot: string): IndexedAggregate {
+  const members = db.aggregateManifest(idx)
+  const layout = buildIndexedAggregate(members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize })))
+  if (layout.rootPieceCid !== expectedRoot) {
+    throw new Error(
+      `recomputed indexed aggregate root ${layout.rootPieceCid} does not match stored root ${expectedRoot}`
+    )
+  }
+  return layout
+}
+
+/**
+ * Pipe one sub-piece's canonical CAR into the response while recomputing its
+ * piece commitment. Bytes come from the assembled CAR on disk when the
+ * sub-piece has one, otherwise from the block-verified canonical gateway
+ * stream (`fetchCanonicalCar`) — the same bytes the plan-time commitment was
+ * computed over. Returns the byte count written.
+ */
+async function pipeSubPiece(
+  db: MigrationDB,
+  res: ServerResponse,
+  pieceCid: string,
+  expectedLength: number
+): Promise<number> {
+  const subPiece = db.subPieceByCid(pieceCid)
+  if (subPiece == null || subPiece.status !== 'built') {
+    throw new Error(`sub-piece ${pieceCid} is ${subPiece == null ? 'unknown' : 'not built'}`)
+  }
+
+  const hasher = Hasher.create()
+  let written = 0
+
+  const consume = async (chunks: AsyncIterable<Uint8Array>): Promise<void> => {
+    for await (const chunk of chunks) {
+      hasher.write(chunk)
+      written += chunk.length
+      if (written > expectedLength) {
+        throw new Error(`sub-piece ${pieceCid} produced more than the expected ${expectedLength} bytes`)
+      }
+      await writeChunk(res, chunk)
+    }
+  }
+
+  if (subPiece.carPath != null) {
+    await consume(createReadStream(subPiece.carPath))
+  } else if (subPiece.url != null && subPiece.url !== '') {
+    const carUrl = new URL(subPiece.url)
+    const sourceCid = carUrl.pathname.replace(/^\/ipfs\//, '')
+    const { body } = await fetchCanonicalCar(carUrl.origin, sourceCid)
+    await consume(body as unknown as AsyncIterable<Uint8Array>)
+  } else {
+    throw new Error(`sub-piece ${pieceCid} has neither a CAR file nor a gateway URL`)
+  }
+
+  if (written !== expectedLength) {
+    throw new Error(`sub-piece ${pieceCid} produced ${written} bytes, expected ${expectedLength}`)
+  }
+  const recomputed = (Link.create(Raw.code, hasher.digest()) as CID).toString()
+  if (recomputed !== pieceCid) {
+    throw new Error(`sub-piece bytes recomputed to ${recomputed}, expected ${pieceCid}`)
+  }
+  return written
+}
+
+/**
+ * Answer a pull for an indexed aggregate root: `Content-Length` up front, then
+ * the regions in stream order. A mid-stream failure destroys the socket — the
+ * provider sees a transport abort, never a clean-looking short body.
+ */
+export async function serveIndexedAggregate(
+  db: MigrationDB,
+  idx: number,
+  expectedRoot: string,
+  res: ServerResponse,
+  head: boolean
+): Promise<void> {
+  let layout: IndexedAggregate
+  try {
+    layout = layoutForAggregate(db, idx, expectedRoot)
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err)
+    log(`aggregate ${idx}: ${message}`)
+    res.writeHead(500, { 'content-type': 'text/plain' })
+    res.end('aggregate layout unavailable')
+    return
+  }
+
+  res.writeHead(200, {
+    'content-type': 'application/octet-stream',
+    'content-length': String(layout.streamLength),
+    'cache-control': 'no-store',
+  })
+  if (head) {
+    res.end()
+    return
+  }
+
+  try {
+    for (const region of layout.regions) {
+      if (region.kind === 'zero') {
+        await writeZeros(res, region.length)
+      } else if (region.kind === 'index') {
+        await writeChunk(res, layout.indexBytes)
+      } else {
+        const member = layout.members[region.memberIndex]
+        const written = await pipeSubPiece(db, res, member.pieceCid, region.payloadLength)
+        await writeZeros(res, region.length - written)
+      }
+    }
+    res.end()
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err)
+    log(`aggregate ${idx}: stream aborted — ${message}`)
+    res.destroy()
+  }
+}
diff --git a/packages/cli/src/aggregate.ts b/packages/cli/src/aggregate.ts
index ab0cefa..9d39eff 100644
--- a/packages/cli/src/aggregate.ts
+++ b/packages/cli/src/aggregate.ts
@@ -9,17 +9,22 @@
  * re-derives on add.
  */
 
+import { Index } from '@web3-storage/data-segment'
 import * as Piece from '@web3-storage/data-segment/piece'
+import { buildIndexedAggregate } from 'ipfs2foc-core/indexed-aggregate'
 import { pieceAggregateCommP } from 'ipfs2foc-core/piece-aggregate'
 import type { PieceResult } from './piece.ts'
 
 const NODE_SIZE = 32n
+const INDEX_ENTRY_SIZE = 64n
 
 export interface AggregatePlan {
   index: number
   /** Aggregate root PieceCID v2 (aggregate piece commitment). */
   rootPieceCid: string
   members: PieceResult[]
+  /** True when the root is an indexed (data segment) aggregate. */
+  indexed: boolean
 }
 
 export interface PackResult {
@@ -51,7 +56,7 @@ export function packAggregates(pieces: PieceResult[], aggregateSizeBytes: bigint
       return
     }
     const root = pieceAggregateCommP(members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize }))).rootPieceCid
-    aggregates.push({ index: aggregates.length, rootPieceCid: root, members })
+    aggregates.push({ index: aggregates.length, rootPieceCid: root, members, indexed: false })
     members = []
     used = 0n
   }
@@ -72,3 +77,55 @@ export function packAggregates(pieces: PieceResult[], aggregateSizeBytes: bigint
   flush()
   return { aggregates, oversized }
 }
+
+/**
+ * Pack pieces into indexed (data segment) aggregates. The packing budget
+ * reserves the index area up front — `maxIndexEntriesInDeal(dealSize) * 64`
+ * padded bytes at the deal's tail — so a full group still fits the deal size
+ * with its index. Groups of one skip the wrapper entirely: the piece's own
+ * CID is the aggregate root and it is pulled/added as itself (an index over
+ * a single entry is rejected by the provider's aggregate parsing).
+ */
+export function packIndexedAggregates(pieces: PieceResult[], aggregateSizeBytes: bigint): PackResult {
+  const maxEntries = BigInt(Index.maxIndexEntriesInDeal(aggregateSizeBytes))
+  const budget = aggregateSizeBytes - maxEntries * INDEX_ENTRY_SIZE
+
+  const aggregates: AggregatePlan[] = []
+  const oversized: PieceResult[] = []
+
+  let members: PieceResult[] = []
+  let used = 0n
+
+  const flush = (): void => {
+    if (members.length === 0) {
+      return
+    }
+    if (members.length === 1) {
+      aggregates.push({ index: aggregates.length, rootPieceCid: members[0].pieceCid, members, indexed: false })
+    } else {
+      const root = buildIndexedAggregate(
+        members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize })),
+        { maxDealSize: aggregateSizeBytes }
+      ).rootPieceCid
+      aggregates.push({ index: aggregates.length, rootPieceCid: root, members, indexed: true })
+    }
+    members = []
+    used = 0n
+  }
+
+  for (const piece of pieces) {
+    const size = paddedSize(piece.pieceCid)
+    if (size > budget) {
+      oversized.push(piece)
+      continue
+    }
+    if (used + size > budget || BigInt(members.length) >= maxEntries) {
+      flush()
+    }
+    members.push(piece)
+    used += size
+  }
+
+  flush()
+  return { aggregates, oversized }
+}
diff --git a/packages/cli/src/db.ts b/packages/cli/src/db.ts
index a82de4f..88d01ff 100644
--- a/packages/cli/src/db.ts
+++ b/packages/cli/src/db.ts
@@ -103,6 +103,12 @@ export interface AggregateRow {
   /** Aggregate root PieceCID v2 — the parent CID added on-chain for the aggregate. */
   rootPieceCid: string
   pieceSizeBytes: string
+  /**
+   * True when the aggregate embeds a data segment index (single-piece add;
+   * the provider pulls one assembled stream). False is the bare layout whose
+   * sub-pieces are pulled individually.
+   */
+  indexed: boolean
   status: AggregateStatus
   /** Synthetic per-aggregate pull marker, set when submission begins. */
   pullId: string | null
@@ -162,6 +168,7 @@ export class MigrationDB {
         idx              INTEGER PRIMARY KEY,
         root_piece_cid   TEXT NOT NULL,
         piece_size_bytes TEXT NOT NULL,
+        indexed          INTEGER NOT NULL DEFAULT 0,
         status           TEXT NOT NULL DEFAULT 'planned',
         pull_id          TEXT,
         data_set_id      TEXT,
@@ -366,10 +373,10 @@ export class MigrationDB {
    * a sub-piece — single-asset source CIDs become 1-member passthrough
    * sub-pieces at plan time, so the pull/add path has one canonical shape.
    */
-  saveAggregate(idx: number, rootPieceCid: string, pieceSizeBytes: bigint, members: string[]): void {
+  saveAggregate(idx: number, rootPieceCid: string, pieceSizeBytes: bigint, members: string[], indexed = false): void {
     const aggregateStmt = this.#db.prepare(
-      `INSERT INTO aggregates (idx, root_piece_cid, piece_size_bytes, status, created_at)
-       VALUES (?, ?, ?, 'planned', ?)`
+      `INSERT INTO aggregates (idx, root_piece_cid, piece_size_bytes, indexed, status, created_at)
+       VALUES (?, ?, ?, ?, 'planned', ?)`
     )
     const memberStmt = this.#db.prepare(
       `INSERT INTO aggregate_members (aggregate_idx, segment_index, sub_piece_cid)
@@ -381,7 +388,7 @@ export class MigrationDB {
     // recordBuiltSubPiece / recordPassthroughSubPiece writers.
     this.#db.exec('BEGIN')
     try {
-      aggregateStmt.run(idx, rootPieceCid, pieceSizeBytes.toString(), new Date().toISOString())
+      aggregateStmt.run(idx, rootPieceCid, pieceSizeBytes.toString(), indexed ? 1 : 0, new Date().toISOString())
       members.forEach((subPieceCid, segmentIndex) => {
         memberStmt.run(idx, segmentIndex, subPieceCid)
       })
@@ -392,13 +399,25 @@ export class MigrationDB {
     }
   }
 
+  /**
+   * The indexed aggregate whose root is `rootPieceCid`, if any. Serves the
+   * `/piece/{pcidv2}` route's aggregate dispatch: an indexed root answers
+   * with the assembled stream, not a sub-piece redirect.
+   */
+  indexedAggregateByRoot(rootPieceCid: string): { idx: number } | null {
+    const row = this.#db
+      .prepare(`SELECT idx FROM aggregates WHERE root_piece_cid = ? AND indexed = 1`)
+      .get(rootPieceCid) as { idx: number | bigint } | undefined
+    return row == null ? null : { idx: Number(row.idx) }
+  }
+
   aggregates(): AggregateRow[] {
     // `member_count` is the *source-CID* count, expanding packed sub-pieces
     // through `sub_piece_members`. A 48-CID packed aggregate reports 48, not
     // 1, so operator-facing counters (report, status) reflect input shape.
     const rows = this.#db
       .prepare(
-        `SELECT a.idx, a.root_piece_cid, a.piece_size_bytes, a.status, a.pull_id,
+        `SELECT a.idx, a.root_piece_cid, a.piece_size_bytes, a.indexed, a.status, a.pull_id,
                 a.data_set_id, a.piece_id, a.tx_hash, a.committed_block, a.error,
                 a.submitted_at, a.parked_at, a.committed_at,
                 (
@@ -417,6 +436,7 @@ export class MigrationDB {
         idx: Number(row.idx),
         rootPieceCid: String(row.root_piece_cid),
         pieceSizeBytes: String(row.piece_size_bytes),
+        indexed: Number(row.indexed) === 1,
         status: row.status as AggregateStatus,
         pullId: str(row.pull_id),
         dataSetId: str(row.data_set_id),
diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts
index 9b5810b..048cbb6 100644
--- a/packages/cli/src/index.ts
+++ b/packages/cli/src/index.ts
@@ -56,6 +56,9 @@ Usage:
   ipfs2foc commp  <cid> [--gateway URL]...
   ipfs2foc plan   --cids <file> [--db <file>] [--gateway URL]... [--piece-size 32GiB]
                      [--concurrency 8]
+                     [--indexed-aggregates]  (experimental: embed a data segment index so each
+                     aggregate is pulled as one stream and added as one self-describing piece;
+                     requires a provider whose indexing understands embedded indexes)
   ipfs2foc import-manifest <manifest.json> [--db <file>] [--network mainnet|calibration]
   ipfs2foc export [--db <file>] [--out <file>] [--network mainnet|calibration] [--source-relay <https-url>]
                      [--piece-size 32GiB] [--no-auto-pack]
@@ -262,6 +265,7 @@ async function cmdPlan(argv: string[]): Promise<void> {
       'ipfs-fallback-mode': { type: 'string' },
       'ipfs-fallback-timeout-seconds': { type: 'string' },
       'no-auto-pack': { type: 'boolean', default: false },
+      'indexed-aggregates': { type: 'boolean', default: false },
     },
   })
   if (values.cids == null) {
@@ -283,6 +287,7 @@ async function cmdPlan(argv: string[]): Promise<void> {
       ipfsFallback: fallback.ipfsFallback,
       fallbackTimeoutMs: fallback.fallbackTimeoutMs,
       autoPack: values['no-auto-pack'] !== true,
+      indexedAggregates: values['indexed-aggregates'] === true,
     })
 
     log('')
diff --git a/packages/cli/src/migrate.ts b/packages/cli/src/migrate.ts
index aff062f..0986dc3 100644
--- a/packages/cli/src/migrate.ts
+++ b/packages/cli/src/migrate.ts
@@ -10,7 +10,7 @@
  * aggregates without touching prior state.
  */
 
-import { packAggregates } from './aggregate.ts'
+import { packAggregates, packIndexedAggregates } from './aggregate.ts'
 import type { MigrationDB } from './db.ts'
 import { formatStageSummary, StageStats, Timer } from './metrics.ts'
 import { categoryOf, fetchAndComputePiece, recordPieceOutcome } from './piece.ts'
@@ -29,6 +29,13 @@ export interface PlanOptions {
    * sub-piece is intended.
    */
   autoPack?: boolean
+  /**
+   * Pack indexed (data segment) aggregates instead of the bare layout: one
+   * pulled stream and one added piece per aggregate, self-describing so the
+   * provider can index every sub-piece from the bytes alone. Requires a
+   * provider whose indexing understands the embedded index.
+   */
+  indexedAggregates?: boolean
 }
 
 export interface PlanSummary {
@@ -79,7 +86,7 @@ export async function runPlan(
   let oversized: string[] = []
   if (autoPack) {
     wrapDonePiecesAsPassthroughSubPieces(db)
-    oversized = appendAggregatesFromFreeSubPieces(db, opts.aggregateSizeBytes)
+    oversized = appendAggregatesFromFreeSubPieces(db, opts.aggregateSizeBytes, opts.indexedAggregates === true)
   }
 
   const counts = db.counts()
@@ -134,7 +141,11 @@ export function wrapDonePiecesAsPassthroughSubPieces(db: MigrationDB): void {
  * part of an aggregate. Existing aggregates are never deleted — composition
  * is set at INSERT and frozen for the row's lifetime.
  */
-export function appendAggregatesFromFreeSubPieces(db: MigrationDB, aggregateSizeBytes: bigint): string[] {
+export function appendAggregatesFromFreeSubPieces(
+  db: MigrationDB,
+  aggregateSizeBytes: bigint,
+  indexed = false
+): string[] {
   const aggregated = db.subPieceCidsAlreadyAggregated()
   const subPieces = db.subPiecesByStatus('built').filter((sp) => !aggregated.has(sp.subPieceCid))
   if (subPieces.length === 0) return []
@@ -147,7 +158,8 @@ export function appendAggregatesFromFreeSubPieces(db: MigrationDB, aggregateSize
     url: sp.url ?? '',
   }))
 
-  const { aggregates, oversized } = packAggregates(units, aggregateSizeBytes)
+  const pack = indexed ? packIndexedAggregates : packAggregates
+  const { aggregates, oversized } = pack(units, aggregateSizeBytes)
 
   const base = db.nextAggregateIndex()
   aggregates.forEach((agg, i) => {
@@ -155,7 +167,8 @@ export function appendAggregatesFromFreeSubPieces(db: MigrationDB, aggregateSize
       base + i,
       agg.rootPieceCid,
       aggregateSizeBytes,
-      agg.members.map((m) => m.cid)
+      agg.members.map((m) => m.cid),
+      agg.indexed
     )
   })
 
diff --git a/packages/cli/src/redirect-server.ts b/packages/cli/src/redirect-server.ts
index a6a3cdf..ada648a 100644
--- a/packages/cli/src/redirect-server.ts
+++ b/packages/cli/src/redirect-server.ts
@@ -19,6 +19,7 @@
 import { createReadStream } from 'node:fs'
 import { stat } from 'node:fs/promises'
 import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'
+import { serveIndexedAggregate } from './aggregate-stream.ts'
 import type { MigrationDB } from './db.ts'
 import { CAR_ACCEPT } from './gateway.ts'
 import { log } from './util.ts'
@@ -48,6 +49,16 @@ export async function handlePieceRequest(
   req: IncomingMessage,
   res: ServerResponse
 ): Promise<void> {
+  // An indexed aggregate root answers with the assembled aggregate stream:
+  // sub-piece CARs at their layout offsets plus the embedded data segment
+  // index. Sub-piece lookups below never match an indexed root (the root is
+  // not a sub-piece row), so this dispatch is unambiguous.
+  const indexedAggregate = db.indexedAggregateByRoot(pieceCid)
+  if (indexedAggregate != null) {
+    await serveIndexedAggregate(db, indexedAggregate.idx, pieceCid, res, req.method === 'HEAD')
+    return
+  }
+
   // One lookup. Every piece commitment in the schema is a sub-piece —
   // passthrough sub-pieces carry the gateway URL, assembled sub-pieces
   // carry the CAR file path. The branch picks the response shape.
diff --git a/packages/cli/src/submit-pdp.ts b/packages/cli/src/submit-pdp.ts
index 5696074..6bbd22a 100644
--- a/packages/cli/src/submit-pdp.ts
+++ b/packages/cli/src/submit-pdp.ts
@@ -19,12 +19,13 @@
 import { unlink } from 'node:fs/promises'
 import { calibration, mainnet, Synapse } from '@filoz/synapse-sdk'
 import { canonicalCid, relayPullUrl } from 'ipfs2foc-core'
+import { buildIndexedAggregate } from 'ipfs2foc-core/indexed-aggregate'
 import { checkMinPieceSize } from 'ipfs2foc-core/min-piece-guard'
 import { pieceAggregateCommP } from 'ipfs2foc-core/piece-aggregate'
 import { CID } from 'multiformats/cid'
 import { type Hex, http } from 'viem'
 import { privateKeyToAccount } from 'viem/accounts'
-import type { MigrationDB } from './db.ts'
+import type { AggregateRow, MigrationDB } from './db.ts'
 import { classifyBaseFee, getBaseFee, resolveRpcUrl } from './gas.ts'
 import { formatBytes, formatDuration, formatRate, Timer } from './metrics.ts'
 import { type AddStatusResult, PdpClient, PullBackpressure, type PullResponse } from './pdp.ts'
@@ -92,6 +93,39 @@ export interface SubmitPdpOptions {
 
 const sleep = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms))
 
+/**
+ * What one aggregate row pulls and adds, in either layout.
+ *
+ * Bare: every sub-piece is pulled from its own source URL and the add lists
+ * them under the recomputed aggregate root.
+ *
+ * Indexed: the provider pulls ONE stream — the assembled aggregate served by
+ * redirect-serve at `/piece/{root}` — and the add lists the root as its own
+ * (only) sub-piece, the 1:1 shape Curio's v0 add requires
+ * (`pdp/handlers_add.go transformAddPiecesRequest` rejects zero subPieces).
+ * The embedded data segment index carries the sub-piece structure instead of
+ * the request body.
+ */
+function aggregateShapeFor(
+  agg: Pick<AggregateRow, 'indexed'>,
+  members: Array<{ pieceCid: string; url: string; rawSize: number }>
+): {
+  rootPieceCid: string
+  subPieceCids: string[]
+  pullUnits: Array<{ pieceCid: string; url: string; rawSize: number }>
+} {
+  if (agg.indexed) {
+    const layout = buildIndexedAggregate(members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize })))
+    return {
+      rootPieceCid: layout.rootPieceCid,
+      subPieceCids: [layout.rootPieceCid],
+      pullUnits: [{ pieceCid: layout.rootPieceCid, url: '', rawSize: layout.streamLength }],
+    }
+  }
+  const aggregate = pieceAggregateCommP(members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize })))
+  return { rootPieceCid: aggregate.rootPieceCid, subPieceCids: aggregate.orderedSubPieceCids, pullUnits: members }
+}
+
 /**
  * The pull `sourceUrl` the provider is handed for one sub-piece. Either the
  * operator's redirect server (`{sourceBase}/piece/{pcid}`) or the shared
@@ -253,6 +287,16 @@ export async function runSubmitPdp(
 
     const members = db.aggregateManifest(agg.idx)
     const aggBytesPlanned = members.reduce((sum, m) => sum + m.rawSize, 0)
+    const shape = aggregateShapeFor(agg, members)
+    // An indexed aggregate's one pull URL is the operator's redirect server
+    // streaming the assembled bytes; the stateless relay has no way to build
+    // that stream.
+    if (agg.indexed && opts.sourceRelay != null && opts.sourceRelay !== '') {
+      log(
+        `aggregate ${agg.idx}: indexed aggregates need --source-base (redirect-serve); --source-relay cannot serve them`
+      )
+      break
+    }
 
     // The provider's advertised min piece size is advisory in practice: pull
     // admission accepted and the chain committed sub-floor pieces in live
@@ -287,7 +331,7 @@ export async function runSubmitPdp(
     // receipt parse landed. Skip pull + add and jump straight to receipt
     // validation using the persisted tx_hash.
     if (agg.txHash != null) {
-      const aggregate = pieceAggregateCommP(members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize })))
+      const aggregate = shape
       log(`aggregate ${agg.idx}: resuming receipt validation for tx ${agg.txHash} (root ${aggregate.rootPieceCid})`)
       const onChain = await deps.activePieceCids(rpcUrl, opts.network, opts.dataSetId)
       if (onChain.has(aggregate.rootPieceCid)) {
@@ -329,7 +373,7 @@ export async function runSubmitPdp(
     // landed, otherwise leave the row for the operator to verify and reset
     // (`--retry-unconfirmed`). Never blindly re-add.
     if (agg.status === 'add_unconfirmed') {
-      const aggregate = pieceAggregateCommP(members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize })))
+      const aggregate = shape
       const onChain = await deps.activePieceCids(rpcUrl, opts.network, opts.dataSetId)
       if (onChain.has(aggregate.rootPieceCid)) {
         db.markCommitted(agg.idx, { dataSetId: String(opts.dataSetId) })
@@ -356,9 +400,10 @@ export async function runSubmitPdp(
     // exceeded the max size"). Each batch carries its own FWSS extraData
     // (presigned over that batch). The on-chain aggregate-add below stays a
     // single top-level piece regardless, so its event is unaffected.
-    const aggBytes = members.reduce((sum, m) => sum + m.rawSize, 0)
+    const pullUnits = shape.pullUnits
+    const aggBytes = pullUnits.reduce((sum, m) => sum + m.rawSize, 0)
     log(
-      `aggregate ${agg.idx}: pulling ${members.length} sub-piece(s), ${formatBytes(aggBytes)} ` +
+      `aggregate ${agg.idx}: pulling ${pullUnits.length} piece(s), ${formatBytes(aggBytes)} ` +
         `in batches of ${opts.pullBatch}`
     )
     db.markSubmitted(agg.idx, `pull-${agg.idx}`)
@@ -369,8 +414,8 @@ export async function runSubmitPdp(
     // have no on-chain effect and are idempotent, so a failed aggregate is safe
     // to reset and re-pull.
     let pullErrored: string | null = null
-    for (let start = 0; start < members.length && pullErrored == null; start += opts.pullBatch) {
-      const batch = members.slice(start, start + opts.pullBatch)
+    for (let start = 0; start < pullUnits.length && pullErrored == null; start += opts.pullBatch) {
+      const batch = pullUnits.slice(start, start + opts.pullBatch)
       const batchCids = batch.map((m) => m.pieceCid)
       const attemptId = db.recordPullBatchStart(agg.idx, batchCids)
       const pullExtra = (await ctx.presignForCommit(
@@ -424,14 +469,15 @@ export async function runSubmitPdp(
     db.markParked(agg.idx)
     totalPullMs += pullMs
     log(
-      `aggregate ${agg.idx}: parked ${members.length} sub-piece(s) in ${formatDuration(pullMs)} ` +
+      `aggregate ${agg.idx}: parked ${pullUnits.length} piece(s) in ${formatDuration(pullMs)} ` +
         `(provider pull ${formatRate(aggBytes, pullMs)})`
     )
 
-    // 2. Add the aggregate over the parked sub-pieces. The provider computes the
-    // aggregate piece commitment (commputils.PieceAggregateCommP), so compute it
-    // the same way and order sub-pieces largest-padded-first.
-    const aggregate = pieceAggregateCommP(members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize })))
+    // 2. Add the aggregate over the parked pieces. Bare: the provider
+    // recomputes the aggregate piece commitment over the listed sub-pieces.
+    // Indexed: the parked piece IS the aggregate (already verified byte-for-
+    // byte on pull) and it is added 1:1 as its own sub-piece.
+    const aggregate = shape
 
     // The provider can land the on-chain AddPieces and then fail a later
     // bookkeeping step, returning an error without a tx hash. Reconcile against
@@ -455,12 +501,7 @@ export async function runSubmitPdp(
     db.markAggregateAddUnconfirmed(agg.idx, 'AddPieces submitted; awaiting on-chain confirmation')
     let txHash: string
     try {
-      ;({ txHash } = await pdp.addAggregate(
-        opts.dataSetId,
-        aggregate.rootPieceCid,
-        aggregate.orderedSubPieceCids,
-        addExtra
-      ))
+      ;({ txHash } = await pdp.addAggregate(opts.dataSetId, aggregate.rootPieceCid, aggregate.subPieceCids, addExtra))
       // Persist the tx hash as soon as we have it so resume can poll the receipt
       // (`aggregatesAwaitingReceipt`) rather than re-adding.
       db.markAggregateTxSubmitted(agg.idx, txHash)
diff --git a/packages/cli/test/aggregate-serve.test.ts b/packages/cli/test/aggregate-serve.test.ts
new file mode 100644
index 0000000..de11cd7
--- /dev/null
+++ b/packages/cli/test/aggregate-serve.test.ts
@@ -0,0 +1,81 @@
+/**
+ * End-to-end serve check for indexed aggregates: an HTTP GET of
+ * `/piece/{root}` on the redirect handler must stream bytes whose recomputed
+ * piece commitment equals the aggregate root — the exact verification the
+ * provider's pull runs before parking the piece.
+ */
+
+import assert from 'node:assert/strict'
+import { mkdtemp, rm, writeFile } from 'node:fs/promises'
+import { createServer } from 'node:http'
+import type { AddressInfo } from 'node:net'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { test } from 'node:test'
+import { Piece } from '@web3-storage/data-segment'
+import { buildIndexedAggregate } from 'ipfs2foc-core/indexed-aggregate'
+import { MigrationDB } from '../src/db.ts'
+import { makeRedirectHandler } from '../src/redirect-server.ts'
+
+function payloadOf(byteLength: number, seed: number): Uint8Array {
+  const bytes = new Uint8Array(byteLength)
+  for (let i = 0; i < byteLength; i++) {
+    bytes[i] = (i * 13 + seed * 101 + ((i >> 7) & 0xff)) & 0xff
+  }
+  return bytes
+}
+
+test('GET /piece/{root} streams the assembled indexed aggregate byte-exactly', async () => {
+  const dir = await mkdtemp(join(tmpdir(), 'foc-aggserve-'))
+  const db = new MigrationDB(join(dir, 'migrate.db'))
+  const server = createServer(makeRedirectHandler(db))
+  try {
+    // Two assembled sub-pieces on disk (the payload bytes are what the piece
+    // commitment was computed over; the streamer re-verifies them on the way
+    // out).
+    const subs = [payloadOf(150_000, 1), payloadOf(90_000, 2)].map((payload, i) => {
+      const piece = Piece.fromPayload(payload)
+      return { payload, pieceCid: piece.link.toString(), carPath: join(dir, `sub-${i}.car`) }
+    })
+    db.addCids(subs.map((s) => `src-${s.pieceCid}`))
+    for (const s of subs) {
+      await writeFile(s.carPath, s.payload)
+      db.recordBuiltSubPiece({
+        subPieceCid: s.pieceCid,
+        assembledCarLength: s.payload.length,
+        targetSizeBytes: 1 << 20,
+        carPath: s.carPath,
+        assembledSha256: 'unused',
+        members: [{ cid: `src-${s.pieceCid}`, rawSize: s.payload.length, sha256: null }],
+      })
+    }
+    const layout = buildIndexedAggregate(subs.map((s) => ({ pieceCid: s.pieceCid, rawSize: s.payload.length })))
+    db.saveAggregate(
+      0,
+      layout.rootPieceCid,
+      1n << 30n,
+      subs.map((s) => s.pieceCid),
+      true
+    )
+
+    await new Promise<void>((resolve) => server.listen(0, resolve))
+    const port = (server.address() as AddressInfo).port
+    const res = await fetch(`http://127.0.0.1:${port}/piece/${layout.rootPieceCid}`)
+    assert.equal(res.status, 200)
+    assert.equal(res.headers.get('content-length'), String(layout.streamLength))
+
+    const body = new Uint8Array(await res.arrayBuffer())
+    assert.equal(body.length, layout.streamLength)
+    // The provider-side check: commP over the pulled bytes equals the root.
+    assert.equal(Piece.fromPayload(body).link.toString(), layout.rootPieceCid)
+
+    // HEAD answers the same length without a body.
+    const head = await fetch(`http://127.0.0.1:${port}/piece/${layout.rootPieceCid}`, { method: 'HEAD' })
+    assert.equal(head.status, 200)
+    assert.equal(head.headers.get('content-length'), String(layout.streamLength))
+  } finally {
+    server.close()
+    db.close()
+    await rm(dir, { recursive: true, force: true })
+  }
+})
diff --git a/packages/cli/test/indexed-aggregate.test.ts b/packages/cli/test/indexed-aggregate.test.ts
new file mode 100644
index 0000000..ba4ef83
--- /dev/null
+++ b/packages/cli/test/indexed-aggregate.test.ts
@@ -0,0 +1,119 @@
+/**
+ * Indexed aggregate layout parity.
+ *
+ * The decisive check mirrors what the provider does on pull: recompute the
+ * piece commitment over the assembled byte stream and require it to equal the
+ * aggregate root CID. That only holds if every sub-piece region, zero gap,
+ * and the fr32-unpadded index tail sit at exactly the offsets go-data-segment's
+ * `AggregateObjectReader` would emit.
+ */
+
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { Fr32, Piece, Segment } from '@web3-storage/data-segment'
+import { buildIndexedAggregate } from 'ipfs2foc-core/indexed-aggregate'
+
+const ENTRY_SIZE = 64
+
+function payloadOf(byteLength: number, seed: number): Uint8Array {
+  const bytes = new Uint8Array(byteLength)
+  for (let i = 0; i < byteLength; i++) {
+    bytes[i] = (i * 31 + seed * 7 + ((i >> 8) & 0xff)) & 0xff
+  }
+  return bytes
+}
+
+function subPieceOf(byteLength: number, seed: number): { pieceCid: string; rawSize: number; payload: Uint8Array } {
+  const payload = payloadOf(byteLength, seed)
+  const piece = Piece.fromPayload(payload)
+  return { pieceCid: piece.link.toString(), rawSize: byteLength, payload }
+}
+
+function assembleStream(
+  aggregate: ReturnType<typeof buildIndexedAggregate>,
+  payloads: Map<string, Uint8Array>
+): Uint8Array {
+  const stream = new Uint8Array(aggregate.streamLength)
+  for (const region of aggregate.regions) {
+    if (region.kind === 'piece') {
+      const member = aggregate.members[region.memberIndex]
+      const payload = payloads.get(member.pieceCid)
+      assert.ok(payload, `payload for ${member.pieceCid}`)
+      assert.equal(payload.length, region.payloadLength)
+      stream.set(payload, region.start)
+    } else if (region.kind === 'index') {
+      stream.set(aggregate.indexBytes, region.start)
+    }
+  }
+  return stream
+}
+
+test('assembled stream recomputes to the aggregate root CID', () => {
+  const subs = [subPieceOf(200_000, 1), subPieceOf(65_000, 2), subPieceOf(130_000, 3)]
+  const aggregate = buildIndexedAggregate(subs.map(({ pieceCid, rawSize }) => ({ pieceCid, rawSize })))
+
+  // Regions must cover [0, streamLength) contiguously.
+  let cursor = 0
+  for (const region of aggregate.regions) {
+    assert.equal(region.start, cursor)
+    cursor += region.length
+  }
+  assert.equal(cursor, aggregate.streamLength)
+
+  const stream = assembleStream(aggregate, new Map(subs.map((s) => [s.pieceCid, s.payload])))
+
+  // The provider's pull-side verification: commP over the exact bytes.
+  const recomputed = Piece.fromPayload(stream)
+  assert.equal(recomputed.link.toString(), aggregate.rootPieceCid)
+})
+
+test('index entries round-trip to member offsets and pass checksum', () => {
+  const subs = [subPieceOf(500_000, 4), subPieceOf(300_000, 5), subPieceOf(70_000, 6), subPieceOf(66_000, 7)]
+  const aggregate = buildIndexedAggregate(subs.map(({ pieceCid, rawSize }) => ({ pieceCid, rawSize })))
+
+  // Members are sorted largest padded size first and sit at non-overlapping,
+  // 127-multiple (fr32 quantum) offsets.
+  for (const [i, member] of aggregate.members.entries()) {
+    assert.equal(member.offset % 127, 0)
+    if (i > 0) {
+      assert.ok(member.offset >= aggregate.members[i - 1].offset + aggregate.members[i - 1].length)
+    }
+  }
+
+  // Re-pad the index area and decode the 64-byte descriptors: root must match
+  // each member's piece root, offsets must map back via unpadded = x - x/128,
+  // and the trailing checksum must be exactly what Segment.toBytes computes.
+  const entryArea = aggregate.indexBytes.subarray(
+    0,
+    Math.ceil((aggregate.members.length * ENTRY_SIZE * 127) / 128 / 127) * 127
+  )
+  const padded = Fr32.pad(entryArea)
+  for (const [i, member] of aggregate.members.entries()) {
+    const entry = padded.subarray(i * ENTRY_SIZE, (i + 1) * ENTRY_SIZE)
+    const root = entry.subarray(0, 32)
+    const view = new DataView(entry.buffer, entry.byteOffset)
+    const paddedOffset = view.getBigUint64(32, true)
+    const paddedSize = view.getBigUint64(40, true)
+    assert.deepEqual(root, Piece.fromString(member.pieceCid).root)
+    assert.equal(Number(paddedOffset - paddedOffset / 128n), member.offset)
+    assert.equal(Number(paddedSize - paddedSize / 128n), member.length)
+    const expected = Segment.toBytes({ root, offset: paddedOffset, size: paddedSize })
+    assert.deepEqual(entry, expected)
+  }
+})
+
+test('rejects a single sub-piece and oversized payloads', () => {
+  const only = subPieceOf(10_000, 8)
+  assert.throws(() => buildIndexedAggregate([{ pieceCid: only.pieceCid, rawSize: only.rawSize }]), RangeError)
+
+  const a = subPieceOf(10_000, 9)
+  const b = subPieceOf(10_000, 10)
+  assert.throws(
+    () =>
+      buildIndexedAggregate([
+        { pieceCid: a.pieceCid, rawSize: 10_000_000 },
+        { pieceCid: b.pieceCid, rawSize: b.rawSize },
+      ]),
+    RangeError
+  )
+})
diff --git a/packages/cli/test/submit-pdp-flow.test.ts b/packages/cli/test/submit-pdp-flow.test.ts
index a007f4e..16280b6 100644
--- a/packages/cli/test/submit-pdp-flow.test.ts
+++ b/packages/cli/test/submit-pdp-flow.test.ts
@@ -3,6 +3,7 @@ import { mkdtemp, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { test } from 'node:test'
+import { buildIndexedAggregate } from 'ipfs2foc-core/indexed-aggregate'
 import { pieceAggregateCommP } from 'ipfs2foc-core/piece-aggregate'
 import { MigrationDB } from '../src/db.ts'
 import type { AddStatusResult, PullResponse } from '../src/pdp.ts'
@@ -258,3 +259,59 @@ test('MED-1: a pull that never progresses trips the stall watchdog and fails the
     assert.equal(calls.addAggregate, 0)
   })
 })
+
+test('indexed aggregate: one pull for the assembled stream, add lists the root 1:1', async () => {
+  await withDb('flow-indexed', async (db) => {
+    db.addCids(['bafSrcA', 'bafSrcB'])
+    db.recordPieceSuccess('bafSrcA', M1.pcid, M1.raw, 'g', 'https://gw/ipfs/bafSrcA?format=car', 'sha-a')
+    db.recordPieceSuccess('bafSrcB', M2.pcid, M2.raw, 'g', 'https://gw/ipfs/bafSrcB?format=car', 'sha-b')
+    db.recordPassthroughSubPiece({
+      subPieceCid: M1.pcid,
+      sourceCid: 'bafSrcA',
+      url: 'https://gw/ipfs/bafSrcA?format=car',
+      rawSize: M1.raw,
+      memberSha256: null,
+    })
+    db.recordPassthroughSubPiece({
+      subPieceCid: M2.pcid,
+      sourceCid: 'bafSrcB',
+      url: 'https://gw/ipfs/bafSrcB?format=car',
+      rawSize: M2.raw,
+      memberSha256: null,
+    })
+    const layout = buildIndexedAggregate([
+      { pieceCid: M1.pcid, rawSize: M1.raw },
+      { pieceCid: M2.pcid, rawSize: M2.raw },
+    ])
+    db.saveAggregate(0, layout.rootPieceCid, 32n * 1024n * 1024n * 1024n, [M1.pcid, M2.pcid], true)
+
+    const pulled: Array<{ pieceCid: string; sourceUrl: string }> = []
+    const added: Array<{ root: string; subPieceCids: string[] }> = []
+    const { deps } = fakeDeps({
+      event: { blockNumber: 100n, pieceIds: [7n], pieceCids: [layout.rootPieceCid] },
+    })
+    const setup = deps.setup
+    deps.setup = async (...args: Parameters<typeof setup>) => {
+      const s = await setup(...args)
+      const pull = s.pdp.pull.bind(s.pdp)
+      s.pdp.pull = async (body) => {
+        pulled.push(...(body.pieces as Array<{ pieceCid: string; sourceUrl: string }>))
+        return pull(body)
+      }
+      const addAggregate = s.pdp.addAggregate.bind(s.pdp)
+      s.pdp.addAggregate = async (dataSetId, root, subPieceCids, extraData) => {
+        added.push({ root, subPieceCids })
+        return addAggregate(dataSetId, root, subPieceCids, extraData)
+      }
+      return s
+    }
+
+    await runSubmitPdp(db, baseOpts, deps)
+
+    assert.deepEqual(pulled, [
+      { pieceCid: layout.rootPieceCid, sourceUrl: `http://redirect.local/piece/${layout.rootPieceCid}` },
+    ])
+    assert.deepEqual(added, [{ root: layout.rootPieceCid, subPieceCids: [layout.rootPieceCid] }])
+    assert.equal(db.aggregates()[0]?.status, 'committed')
+  })
+})
diff --git a/packages/core/package.json b/packages/core/package.json
index ac47a06..1012288 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -14,7 +14,8 @@
     "./manifest": "./src/manifest.ts",
     "./min-piece-guard": "./src/min-piece-guard.ts",
     "./pdp-verifier": "./src/pdp-verifier.ts",
-    "./piece-aggregate": "./src/piece-aggregate.ts"
+    "./piece-aggregate": "./src/piece-aggregate.ts",
+    "./indexed-aggregate": "./src/indexed-aggregate.ts"
   },
   "scripts": {
     "typecheck": "tsc --noEmit"
diff --git a/packages/core/src/indexed-aggregate.ts b/packages/core/src/indexed-aggregate.ts
new file mode 100644
index 0000000..51770f6
--- /dev/null
+++ b/packages/core/src/indexed-aggregate.ts
@@ -0,0 +1,204 @@
+/**
+ * Indexed aggregate (FRC-0058 data segment) construction, in pure JS.
+ *
+ * Unlike the bare aggregate commitment in `piece-aggregate` (sub-piece trees
+ * combined with zero padding, no self-description), an indexed aggregate
+ * embeds a data segment index in the piece's tail. The provider can recover
+ * every sub-piece from the aggregate bytes alone: it seeks to the index start
+ * offset, parses the entries, and reads each sub-piece at its recorded
+ * offset. This is the layout Curio's aggregate indexing consumes
+ * (`tasks/indexing/index_helpers.go IndexAggregate` reads the index via
+ * `datasegment.DataSegmentIndexStartOffset` + `parseDataSegmentIndex`).
+ *
+ * The byte stream this module describes mirrors go-data-segment
+ * `Aggregate.AggregateObjectReader` (verified: filecoin-project/go-data-segment
+ * datasegment/creation.go AggregateObjectReader):
+ *
+ *   [sub-piece 0 bytes][zero fill]...[sub-piece n bytes][zero fill]
+ *   [zero gap][fr32-unpadded index entries][zero fill to end]
+ *
+ * All offsets and lengths are in the unpadded (raw byte) domain:
+ *   unpadded(x) = x - x/128   (verified: datasegment/index.go
+ *   SegmentDesc.UnpaddedOffest / UnpaddedLength)
+ *
+ * The index start is `unpadded(dealSize) - unpadded(maxEntries * 64)`
+ * (verified: datasegment/parse_index.go DataSegmentIndexStartOffset), and each
+ * serialized entry is 64 bytes: root(32) || offset u64 LE || size u64 LE ||
+ * checksum(16) (`@web3-storage/data-segment` `Segment.toBytes`, which cites
+ * go's index.go serialization). The entry bytes land in the stream fr32
+ * -unpadded (verified: datasegment/creation.go IndexReader).
+ *
+ * The aggregate root includes the index nodes, so the piece commitment binds
+ * the index: `Aggregate.build` places each entry's two nodes into the tree the
+ * same way go-data-segment's `NewAggregate` does.
+ *
+ * Pure module - no `node:` imports - so the browser app can build the same
+ * aggregates the CLI submits.
+ */
+
+import { Aggregate, type API, Fr32, Index, Piece, Segment } from '@web3-storage/data-segment'
+
+/**
+ * `AggregateView` omits the `index` field the builder's `Aggregate` class
+ * carries at runtime (upstream type gap; see `.research/upstream-gaps.md`).
+ */
+type BuiltAggregate = API.AggregateView & { index: API.SegmentInfo[] }
+
+const ENTRY_SIZE = 64
+const FR32_QUANTUM = 128n
+
+/** unpadded(x): bytes of payload that fr32-expand to `x` padded bytes. */
+function unpadded(size: bigint): bigint {
+  return size - size / FR32_QUANTUM
+}
+
+export interface IndexedAggregateSubPiece {
+  /** Sub-piece PieceCID v2. */
+  pieceCid: string
+  /** Actual byte length of the sub-piece payload (the CAR). */
+  rawSize: number
+}
+
+export interface IndexedAggregateMember {
+  pieceCid: string
+  rawSize: number
+  /** Start of this sub-piece's region in the aggregate byte stream. */
+  offset: number
+  /** Region length; `rawSize` payload bytes then zero fill. */
+  length: number
+}
+
+export type IndexedAggregateRegion =
+  | { kind: 'piece'; start: number; length: number; memberIndex: number; payloadLength: number }
+  | { kind: 'zero'; start: number; length: number }
+  | { kind: 'index'; start: number; length: number }
+
+export interface IndexedAggregate {
+  /** Aggregate PieceCID v2; rawSize is the full stream length. */
+  rootPieceCid: string
+  /** Members in stream order (largest padded size first). */
+  members: IndexedAggregateMember[]
+  /** Padded (on-chain) aggregate size in bytes; a power of two. */
+  dealSize: bigint
+  /** Total unpadded byte-stream length: unpadded(dealSize). */
+  streamLength: number
+  /** Unpadded offset where the serialized index begins. */
+  indexStartOffset: number
+  /** Serialized index: fr32-unpadded entries, zero-filled to the index area. */
+  indexBytes: Uint8Array
+  /** Contiguous cover of [0, streamLength): what to emit, in order. */
+  regions: IndexedAggregateRegion[]
+}
+
+/**
+ * Build the indexed aggregate over sub-pieces. Sub-pieces are laid out
+ * largest-padded-first (matching `piece-aggregate` ordering, which minimizes
+ * alignment gaps). Throws RangeError when the pieces cannot fit any deal size
+ * up to `maxDealSize` (default 64 GiB, the provider ceiling).
+ */
+export function buildIndexedAggregate(
+  subPieces: IndexedAggregateSubPiece[],
+  options: { maxDealSize?: bigint } = {}
+): IndexedAggregate {
+  if (subPieces.length < 2) {
+    // Curio's IndexAggregate requires at least 2 entries; a single piece
+    // should be submitted as itself, not wrapped.
+    throw new RangeError(`indexed aggregate needs at least 2 sub-pieces, got ${subPieces.length}`)
+  }
+  const maxDealSize = options.maxDealSize ?? 2n ** 36n
+
+  const entries = subPieces
+    .map((sp) => {
+      const piece = Piece.fromString(sp.pieceCid)
+      return { pieceCid: sp.pieceCid, rawSize: sp.rawSize, piece }
+    })
+    .sort((a, b) => b.piece.height - a.piece.height)
+
+  // Every sub-piece payload must fit its own padded tree, or the zero fill
+  // would truncate real bytes.
+  for (const e of entries) {
+    const capacity = unpadded(2n ** BigInt(e.piece.height) * 32n)
+    if (BigInt(e.rawSize) > capacity) {
+      throw new RangeError(`sub-piece ${e.pieceCid} rawSize ${e.rawSize} exceeds its padded capacity ${capacity}`)
+    }
+  }
+
+  // Find the smallest power-of-two deal size the builder accepts: alignment
+  // gaps and the index reservation mean the sum of padded sizes is only a
+  // lower bound.
+  const sumPadded = entries.reduce((acc, e) => acc + 2n ** BigInt(e.piece.height) * 32n, 0n)
+  let dealSize = 2n ** BigInt(sumPadded.toString(2).length - ((sumPadded & (sumPadded - 1n)) === 0n ? 1 : 0))
+  let aggregate: BuiltAggregate | undefined
+  for (; dealSize <= maxDealSize; dealSize *= 2n) {
+    try {
+      aggregate = Aggregate.build({
+        pieces: entries.map((e) => e.piece),
+        size: Aggregate.Size.from(dealSize),
+      }) as BuiltAggregate
+      break
+    } catch (err) {
+      if (err instanceof RangeError) {
+        continue
+      }
+      throw err
+    }
+  }
+  if (aggregate === undefined) {
+    throw new RangeError(`sub-pieces do not fit an aggregate of at most ${maxDealSize} padded bytes`)
+  }
+
+  const streamLength = Number(unpadded(dealSize))
+
+  // Index area geometry (verified: go-data-segment parse_index.go
+  // DataSegmentIndexStartOffset; @web3-storage/data-segment index.js
+  // maxIndexEntriesInDeal matches go's MaxIndexEntriesInDeal).
+  const maxEntries = Index.maxIndexEntriesInDeal(dealSize)
+  const indexAreaLength = Number(unpadded(BigInt(maxEntries * ENTRY_SIZE)))
+  const indexStartOffset = streamLength - indexAreaLength
+
+  // Serialize the entries exactly as go's IndexReader does: concatenated
+  // 64-byte descriptors, zero-padded to a 128-byte multiple, fr32-unpadded,
+  // then zero-filled to the index area.
+  const segmentBytes = new Uint8Array(Math.ceil((aggregate.index.length * ENTRY_SIZE) / 128) * 128)
+  for (const [i, segment] of aggregate.index.entries()) {
+    segmentBytes.set(Segment.toBytes(segment), i * ENTRY_SIZE)
+  }
+  const unpaddedEntries = Fr32.unpad(segmentBytes)
+  const indexBytes = new Uint8Array(indexAreaLength)
+  indexBytes.set(unpaddedEntries, 0)
+
+  // Members and regions in stream order. `aggregate.index` preserves write
+  // order (largest-first), and each entry's padded offset/size convert to the
+  // unpadded domain by the go formulas above.
+  const members: IndexedAggregateMember[] = []
+  const regions: IndexedAggregateRegion[] = []
+  let cursor = 0
+  for (const [i, segment] of aggregate.index.entries()) {
+    const start = Number(unpadded(segment.offset))
+    const length = Number(unpadded(segment.size))
+    if (start < cursor) {
+      throw new Error(`aggregate layout error: segment ${i} starts at ${start}, before cursor ${cursor}`)
+    }
+    if (start > cursor) {
+      regions.push({ kind: 'zero', start: cursor, length: start - cursor })
+    }
+    const member = entries[i]
+    members.push({ pieceCid: member.pieceCid, rawSize: member.rawSize, offset: start, length })
+    regions.push({ kind: 'piece', start, length, memberIndex: i, payloadLength: member.rawSize })
+    cursor = start + length
+  }
+  if (indexStartOffset < cursor) {
+    throw new Error(`aggregate layout error: index starts at ${indexStartOffset}, before cursor ${cursor}`)
+  }
+  if (indexStartOffset > cursor) {
+    regions.push({ kind: 'zero', start: cursor, length: indexStartOffset - cursor })
+  }
+  regions.push({ kind: 'index', start: indexStartOffset, length: indexAreaLength })
+
+  // The v2 envelope carries the full stream length as payload: padding 0 and
+  // the deal tree's own height, so the provider's recomputed commP over
+  // exactly `streamLength` pulled bytes matches this CID.
+  const rootPieceCid = Piece.toLink({ root: aggregate.root, height: aggregate.height, padding: 0n }).toString()
+
+  return { rootPieceCid, members, dealSize, streamLength, indexStartOffset, indexBytes, regions }
+}