diff --git a/docs/deploy.md b/docs/deploy.md
index d5633bf0..7191780f 100644
--- a/docs/deploy.md
+++ b/docs/deploy.md
@@ -35,6 +35,13 @@ This release does not convert VM or snapshot state from older releases. Drain ol
 claims and use fresh `data_dir` and `checkpoint_dir` locations when upgrading;
 older checkpoints and promoted templates must not be reused.
 
+The S3 `checkpoint_store` changed its object layout (chunk manifests instead of
+whole export files). Records written by an earlier release are not readable and
+there is no fallback reader: before starting the new binary, wake or release
+every archived claim and delete the checkpoints, then re-promote templates, or
+point `prefix` at an empty location. Objects of the old layout under a reused
+prefix stay in the bucket until you delete them.
+
 The scalar egress-attachment keys are retired: rename `"bridge": "br0"` to
 `"bridges": ["br0"]` and `"network": "cni"` to `"networks": ["cni"]` before
 starting the new binary — config loading rejects the old spellings loudly
@@ -128,7 +135,7 @@ gets the intersection of the two.
 | `preview_secret` | — | cluster-shared HMAC secret signing preview tokens (all nodes share one) |
 | `preview_advertise` | = `preview_listen` | the browser-facing preview base URL, minted into every preview URL, so it must name a routable host (a wildcard `preview_listen` needs it set explicitly); nodes behind one TLS proxy may share it, while signed tokens route internally through each owner's `advertise_addr` |
 | `checkpoint_dir` | `<data_dir>/checkpoints` | where checkpoints and promoted templates live. Point it at a shared FUSE mount (JuiceFS over object storage, NFS) and every node sharing the mount can branch every checkpoint — records are generation-addressed with `meta.json` as the atomic commit pointer, so no cross-node locking is required of the filesystem. A re-publish retains its superseded export generation for at least ~1h and until a following hourly sweep so an in-flight clone that resolved the old metadata can finish; budget the current generation plus every generation retained across that grace-and-sweep window. An explicit delete can make a concurrent clone fail visibly. One contract on any shared root (mount or bucket): a template key has a single writer — promotes go to the sandbox's owner node, and operators must not race promotes of one name from different nodes (checkpoint ids are node-generated and never collide) |
-| `checkpoint_store` | dir | checkpoint AND promoted-template backend (both live in one store root, id-namespaced ck_/tp_): `{"kind": "s3", "s3": {"bucket": "…", "prefix": "ck/", "endpoint": "…", "region": "…", "force_path_style": true}}` stores checkpoints in object storage (any node claims any checkpoint, no shared mount needed). Credentials come from the standard AWS chain (env/IAM role), never this file. Re-publish retains prior export generations until Delete so an in-flight fetch that selected old metadata can finish; budget storage for those generations. When caching a new generation, each node prunes other cached generations of that record unused for at least an hour. An explicit S3 Delete can still make a concurrent fetch that has not finished materializing fail visibly. A crash between upload and the meta.json commit marker leaves orphan objects invisible to listings — add an S3 lifecycle rule to reclaim them. Absent = the dir backend at `checkpoint_dir` |
+| `checkpoint_store` | dir | checkpoint AND promoted-template backend (both live in one store root, id-namespaced ck_/tp_): `{"kind": "s3", "s3": {"bucket": "…", "prefix": "ck/", "endpoint": "…", "region": "…", "force_path_style": true}}` stores checkpoints in object storage (any node claims any checkpoint, no shared mount needed). Credentials come from the standard AWS chain (env/IAM role), never this file. An export file is stored as 16 MiB chunks keyed by SHA-256 under `<prefix>chunks/<lineage>/`, with one `<prefix>manifests/<id>/<generation>.json` per published generation naming each file's size and present chunks, and `<prefix><id>/meta.json` uploaded last as the commit marker (the record prefix exists only while the record is live). Holes and all-zero chunks are never uploaded, and a fetch recreates the file at its full size and writes only the present chunks, so a mostly-empty disk costs its data, not its size, on the wire and on the fetching node. A lineage is the dedup scope: it defaults to the record id, and every checkpoint and archive of one claim shares the claim's sandbox id, so consecutive captures upload only the chunks that changed (the first capture after a wake uploads everything again when no other live record of that claim remains, because the wake deletes the record it restored from); a publish into a shared lineage writes a zero-byte marker `<prefix>lineages/<lineage>/<id>` after its manifest and before `meta.json`, and reads only the records those markers list; a crash between manifest and marker only costs a missed dedup, and Delete removes the record's marker right away. Re-publish retains the prior generation's manifest so an in-flight fetch that selected old metadata can finish. sandboxd never deletes a chunk or a manifest from the bucket: Delete removes `meta.json`, the labels and the lineage markers and re-stamps the manifests, so a fetch that already selected the record still completes, and reclaiming storage is the operator's job, see [Reclaiming S3 storage](#reclaiming-s3-storage). A publish or fetch is aborted after 30 minutes. Do not put an expiry lifecycle rule on the prefix: it would delete chunks a live record still names. Budget storage for every generation and chunk until your reclaim runs. When caching a new generation, each node prunes other cached generations of that record unused for at least an hour. Absent = the dir backend at `checkpoint_dir` |
 | `checkpoint_ttl_hours` | 0 (keep forever) | ages out checkpoints older than this; the sweep runs hourly and at startup. Explicit deletes never wait for it. Must be nonzero and match fleet-wide when `checkpoint_peer_heal` is on — it is the expiry eligibility point for a healed replica a delete broadcast missed, after which its next successful hourly sweep removes it; persistent sweep failure extends retention until one succeeds, so it is not a hard ceiling |
 | `checkpoint_peer_heal` | false | on a cluster, lets a node pull a checkpoint it lacks from a peer — found via a live probe, not gossip — rather than failing the branch; see [placement lifecycle](cluster.md#checkpoints-on-a-cluster). Three requirements, all enforced at config load: a nonempty `api_token` (the blob transfer between peers authenticates with it; without one the raw record stream would be open), `mesh.cluster_key` set (the pull presents the fleet `api_token` to an address learned from the peer probe, so the gossip layer carrying that address must itself be authenticated), and `checkpoint_ttl_hours` nonzero (a replica a delete broadcast missed becomes eligible for expiry after it, and its next successful hourly sweep removes it — so it is the finite eligibility point, not an exact ceiling). A shared checkpoint store (`checkpoint_store` kind `s3`) ignores this setting — every node already resolves every checkpoint directly, so there is nothing to heal |
 | `warm_max` (pool entry) | 0 (static) | turns on the demand-adaptive watermark for that pool: the warm target rises from `warm` toward `warm_max` while claims arrive faster than the measured provision lead covers, and decays back over ~a minute of silence. The pool follows the target down: warm VMs above it are destroyed once the count has stayed above it for a full decay period, so a burst does not leave the pool parked at its high-water mark |
@@ -142,7 +149,7 @@ gets the intersection of the two.
 | `disk_reserve` | unset | `{"free": "100G", "dirs": ["/srv/sbx/run"]}`: while the filesystem of any listed directory (default `data_dir`) has less than `free` available, checked every 2 s, the node refuses new claims, forks, checkpoint claims, wakes of hibernated or archived sandboxes, hibernates, checkpoint captures and promotes with 429, and parks refill; the idle sweep and a lease-end archive leave their sandboxes running and retry on the next tick; `/v1/info` reports it as at capacity and the mesh sees no warm VMs on it. Running sandboxes keep running. On an XFS project quota the free space is the project's. List cocoon's `run_dir` when it sits on another filesystem, since the copy-on-write disks grow there. Size `free` to at least two sandboxes' worst-case growth (copy-on-write disk plus memory file each) and keep the refusal above your disk alert's warning level, so the warning comes first: on a 640G project with 40G disks and 8G memory files, `100G` refuses at 84% used |
 | `idle_hibernate_seconds` | 0 (off) | node-wide idle policy for unpooled claims (template/checkpoint claims): a none-lane claim is hibernated once it has had no open data-plane connection (relay, buffered exec, preview) and no egress request in flight for this long; the clock restarts when the last connection closes, so a long command is never cut short. The next call that reaches the guest wakes it transparently. Per-pool `idle_hibernate_seconds` does the same for that pool's claims; pooled keys ignore the node-wide value, and egress pools reject it because they cannot resume safely. Opt in deliberately: a wake costs latency and the snapshot, so callers with their own idle logic must not pay twice |
 | `archive_after_seconds` | 0 (off) | tier below hibernation: a hibernated claim idle this long is checkpointed to the store and its local VM dropped, freeing the node entirely; the next call that reaches the guest restores it transparently (a checkpoint restore's latency) with a fresh lease of the length the claim was granted (the server default of 5m when it asked for none); a lease moves only when a client calls [`/renew`](sandboxd-api.md#post-v1sandboxesidrenew). Keep `idle_hibernate_seconds` under the leases your claims use, or a woken sandbox that idles again is destroyed by the reap before it can hibernate, its archive already consumed by the wake; one still in use when its lease ends is destroyed the same way, unless it claimed `on_expire: archive`, which archives at the lease end whatever this setting. Requires `idle_hibernate_seconds > 0` and must exceed it. Node-wide for unpooled keys; per-pool overrides for that pool |
-| `archive_delete_after_seconds` | 0 (keep) | purge an archived claim's store checkpoint this long after it was archived, reclaiming storage; the claim is then gone for good. On archive this retention window replaces the live claim deadline; 0 clears the deadline so the archive is kept forever. Same node-wide/per-pool split |
+| `archive_delete_after_seconds` | 0 (keep) | purge an archived claim's store checkpoint this long after it was archived; the claim is then gone for good. On S3, chunks and manifests remain until [offline reclamation](#reclaiming-s3-storage). On archive this retention window replaces the live claim deadline; 0 clears the deadline so the archive is kept forever. Same node-wide/per-pool split |
 | `archive_failover` | false | let another node adopt this node's archived claims when it leaves the mesh: each archive also writes a row (token SHA-256, owner, epoch, checkpoint id, retention deadline) to the shared PostgreSQL, and a `Lookup` that reaches any node takes the claim over there (see [archived claims across node loss](cluster.md#archived-claims-across-node-loss)). Requires `meta_store` with a `cell` (rows are scoped by it, so meshes sharing one database never adopt from each other), a `checkpoint_store` of kind `s3` and a `mesh`, all checked at config load. With the meta store down, archives, wakes and releases of archived claims fail and retry; running claims never touch it. Restart only |
 | `mesh` | unset | join a cluster ([Clusters](cluster.md)); unset = single node |
 | `pools[]` | — | warm pools, keyed by `(template, net, size)`. `warm` defaults to 4 only when both `warm` and `warm_max` are omitted; an explicit `warm: 0` is kept, so `warm: 0` under a `warm_max` starts the pool empty and grows it on demand, and `warm: 0` alone registers the key with no warm VMs (`PUT /v1/pools` never defaults `warm`); `net` is `none` or `egress`; `size` is a tier, below. Retune online without a restart via [`PUT /v1/pools`](sandboxd-api.md#put-v1pools) — omitted pools drain, and their goldens are deleted once they are idle (`warm: 0` keeps the golden). This is the **first-boot seed**: once a node takes a `PUT /v1/pools`, the applied set persists to `<data_dir>/pools.json` and overrides this section on every later boot (a startup log notes it); delete `pools.json` to return to config-owned pools. With a `meta_store` `cell`, the cell's set overrides it instead (see [shared pool set](#shared-pool-set)). Egress stays config-owned either way. A template re-pulled or re-imported under the same ref is picked up within a minute: the pool rebuilds its golden and replaces its warm VMs. See [state ownership](cluster.md#state-ownership) |
@@ -305,6 +312,55 @@ admitted and fails at mount time (500) instead of being refused early (409).
 Whoever hands out attach-only `rw` access owns that trade; see
 [sandboxd-api](sandboxd-api.md#attach-only-volumes).
 
+### Reclaiming S3 storage
+
+sandboxd never garbage-collects the S3 `checkpoint_store`. It only adds
+objects and, on a delete, removes the record's `meta.json`, labels and lineage
+markers; chunks, manifests and everything a superseded record wrote stay in the
+bucket until an operator job removes them. The `dir` backend is different: it
+keeps its own local generation sweep.
+
+**Key layout** (`<prefix>` is the configured prefix, ending in `/`):
+
+| Key | Content |
+|---|---|
+| `<prefix><id>/meta.json` | The live record, uploaded last as the commit marker. Absent once the record is deleted. Its current generation is `export-` plus the first 16 hex characters of the SHA-256 of the object body. |
+| `<prefix><id>/labels.json` | Optional labels, removed with the record. |
+| `<prefix>manifests/<id>/<gen>.json` | One JSON manifest per published generation: `lineage` and, per file, `path`, `size` and the present `chunks` (`index`, `sha256`). `<gen>` is `export-<16 hex>`. |
+| `<prefix>chunks/<lineage>/<sha256>` | A 16 MiB chunk. A chunk is shared by every record of its lineage that names the same digest, and only by those. |
+| `<prefix>lineages/<lineage>/<id>` | Zero-byte marker: record `<id>` published into `<lineage>`. It only drives dedup, never chunk liveness. |
+
+**What sandboxd never deletes:** chunks, manifests of deleted records,
+superseded generations of a live record, and stale markers.
+
+**Offline reclamation only.** Stop every sandboxd node using the prefix and
+wait for all in-flight publishes, deletes and fetches to finish. Keep the
+prefix offline throughout the inventory-and-delete pass. Online reclamation
+is not supported: a publish can replace an old chunk after it was listed but
+before it is deleted. Object age and the transfer timeout do not close that
+window.
+
+From that stable inventory:
+
+1. Read every live `meta.json` and keep the manifest of the generation it
+   selects, for both checkpoints and templates.
+2. Read all kept manifests and collect their chunk keys, including the
+   lineage. If the inventory or a required read fails, stop before deleting.
+3. Delete manifests outside that set and chunks no kept manifest references.
+   A lineage marker or `labels.json` may be deleted when its record has no
+   `meta.json`.
+
+Deletion is idempotent under this offline precondition; a later pass must
+build a fresh inventory. Do not use a bucket lifecycle expiry instead: it
+would remove chunks a live record still names.
+
+**Cost of not running it.** Storage grows without bound. Every archive wake,
+user checkpoint delete and, with replication, every replica round leaves the
+superseded record's unique chunks behind. Successive captures of a running
+claim share few memory chunks, because most of them change between captures, so
+each superseded capture leaves most of its memory image behind; an archive that
+took 10,752 MiB as whole files stored about 912 MiB as chunks.
+
 ### A fuller config
 
 The block above is the minimum. A production node with guarded
diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md
index 9d67c7c1..beb32aaf 100644
--- a/docs/sandboxd-api.md
+++ b/docs/sandboxd-api.md
@@ -750,7 +750,9 @@ concurrent-heal cap is already full — retryable, and the response carries a
 The peer-transfer route behind the probe and heal above — internal, not
 part of the public API; an SDK caller has no reason to call it directly.
 
-- `GET` streams the whole record — guest memory, disk, and meta — as a tar.
+- `GET` streams the whole record — guest memory, disk, and meta — as a tar
+  in which every export file travels as its size plus its non-zero 16 MiB chunks,
+  each with its SHA-256, so holes cost nothing on the wire.
   Operator-token only: the stream carries no tenant scoping, and its only
   real caller is a peer's heal pull, which presents the fleet `api_token`.
   401 missing or unrecognized token, 403 a valid tenant token (authenticated
diff --git a/sandboxd/go.mod b/sandboxd/go.mod
index 0011f4a4..cb2e5056 100644
--- a/sandboxd/go.mod
+++ b/sandboxd/go.mod
@@ -5,7 +5,6 @@ go 1.27.1
 require (
 	github.com/aws/aws-sdk-go-v2 v1.47.3
 	github.com/aws/aws-sdk-go-v2/config v1.33.9
-	github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.4.13
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.114.0
 	github.com/aws/smithy-go v1.28.5
 	github.com/cocoonstack/sandbox/protocol/wire v0.0.0-00010101000000-000000000000
diff --git a/sandboxd/go.sum b/sandboxd/go.sum
index cd59e1d3..80cd1378 100644
--- a/sandboxd/go.sum
+++ b/sandboxd/go.sum
@@ -8,8 +8,6 @@ github.com/aws/aws-sdk-go-v2/credentials v1.20.9 h1:SNO55PnWhaDafsOcv16lCA0kdoFM
 github.com/aws/aws-sdk-go-v2/credentials v1.20.9/go.mod h1:+2reOTRtdD3VSBqpRDNwrZIYy8yMCYa7oK3WKTJ/I1I=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.3 h1:gr3b87OLB29jymWPzp/CAyXwQRJVE+sC4Wm0qBIKHsk=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.3/go.mod h1:BRjf4L/Yx2hY54o+ZibPbLJyMMeSwqYQXwcJfqW66a8=
-github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.4.13 h1:wO7TVbywHwdpHLUiX6DnmP2RDYOACVeJCb6zMfSFViU=
-github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.4.13/go.mod h1:Zc9r0r7wMid/NkbsLrkGxe5vZufWyP0CiC2dDXZ8ldk=
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.6 h1:CqL8NN3p4I0JTwwV5MEoatIl4Y2amZ2X9dRD+0hbRL0=
 github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.6/go.mod h1:exdJ32hMqXl0Wyo6NwqEC5yIhtua3+PDaKE8P29m4qE=
 github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.6 h1:FpBq24TG6WVSObTNhQOv7Q0088rounUUlwn3zlNItkM=
diff --git a/sandboxd/pool/archive_test.go b/sandboxd/pool/archive_test.go
index 3460ceb9..9bacdd2a 100644
--- a/sandboxd/pool/archive_test.go
+++ b/sandboxd/pool/archive_test.go
@@ -51,6 +51,23 @@ func TestArchivePublishWindowPinsCheckpoint(t *testing.T) {
 	}
 }
 
+func TestArchivePublishesUnderTheSandboxLineage(t *testing.T) {
+	eng := newFakeEngine()
+	m := newTestManager(t, eng, archivePool(3600))
+	sb := mustClaim(t, m, testKey)
+	if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil {
+		t.Fatalf("hibernate: %v", err)
+	}
+	rec := &lineageStore{Store: m.ckpts}
+	m.ckpts = rec
+	if err := m.archive(t.Context(), sb); err != nil {
+		t.Fatalf("archive: %v", err)
+	}
+	if rec.lineage != sb.ID {
+		t.Errorf("archive lineage = %q, want the sandbox id %q", rec.lineage, sb.ID)
+	}
+}
+
 func TestArchiveWakeRehibernateWindowAborts(t *testing.T) {
 	eng := newFakeEngine()
 	m := newTestManager(t, eng, archivePool(3600))
@@ -155,7 +172,7 @@ func TestReconcileReclaimsOrphanArchiveCk(t *testing.T) {
 		if err := os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(meta), 0o600); err != nil {
 			t.Fatalf("meta: %v", err)
 		}
-		if err := m.ckpts.Publish(t.Context(), staging, id); err != nil {
+		if err := m.ckpts.Publish(t.Context(), staging, id, ""); err != nil {
 			t.Fatalf("publish: %v", err)
 		}
 	}
@@ -920,8 +937,8 @@ type stallingStore struct {
 	release   chan struct{}
 }
 
-func (s *stallingStore) Publish(ctx context.Context, staging, id string) error {
-	if err := s.Store.Publish(ctx, staging, id); err != nil {
+func (s *stallingStore) Publish(ctx context.Context, staging, id, lineage string) error {
+	if err := s.Store.Publish(ctx, staging, id, lineage); err != nil {
 		return err
 	}
 	s.published <- id
@@ -929,6 +946,16 @@ func (s *stallingStore) Publish(ctx context.Context, staging, id string) error {
 	return nil
 }
 
+type lineageStore struct {
+	store.Store
+	lineage string
+}
+
+func (s *lineageStore) Publish(ctx context.Context, staging, id, lineage string) error {
+	s.lineage = lineage
+	return s.Store.Publish(ctx, staging, id, lineage)
+}
+
 type failingDeleteStore struct {
 	store.Store
 	attempts chan string
diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go
index f74e9559..9234f71e 100644
--- a/sandboxd/pool/checkpoint.go
+++ b/sandboxd/pool/checkpoint.go
@@ -189,7 +189,7 @@ func (m *Manager) publishCheckpoint(ctx context.Context, sb *types.Sandbox, ckID
 	if err := os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); err != nil {
 		return types.Checkpoint{}, "", fmt.Errorf("write checkpoint meta: %w", err)
 	}
-	if err := m.ckpts.Publish(ctx, staging, ckpt.ID); err != nil {
+	if err := m.ckpts.Publish(ctx, staging, ckpt.ID, sb.ID); err != nil {
 		return types.Checkpoint{}, "", fmt.Errorf("commit checkpoint: %w", err)
 	}
 	return ckpt, srcSnap, nil
@@ -278,7 +278,7 @@ func (m *Manager) runHeal(ctx context.Context, ckptID string) (types.Checkpoint,
 	if ckpt, err := m.loadCheckpoint(ctx, ckptID); err == nil {
 		return ckpt, nil // published by another path while this one staged
 	}
-	if err := m.ckpts.Publish(ctx, staging, ckptID); err != nil {
+	if err := m.ckpts.Publish(ctx, staging, ckptID, ""); err != nil {
 		return types.Checkpoint{}, fmt.Errorf("publish healed checkpoint: %w", err)
 	}
 	return m.loadCheckpoint(ctx, ckptID)
diff --git a/sandboxd/pool/checkpoint_test.go b/sandboxd/pool/checkpoint_test.go
index 3346bac9..bb38cd61 100644
--- a/sandboxd/pool/checkpoint_test.go
+++ b/sandboxd/pool/checkpoint_test.go
@@ -7,6 +7,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -363,6 +364,29 @@ func TestRejectedCheckpointOpsLeakNoRecLock(t *testing.T) {
 	}
 }
 
+func TestSharedStoreSweepsOncePerTick(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		shared    bool
+		wantCkpts int32
+		wantTpls  int32
+	}{
+		{"per-node store sweeps both views", false, 1, 1},
+		{"shared store sweeps its one prefix once", true, 1, 0},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			m := newTestManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey})
+			ckpts := &sweepCountingStore{Store: m.ckpts}
+			tpls := &sweepCountingStore{Store: m.tpls}
+			m.ckpts, m.tpls, m.ckptsShared = ckpts, tpls, tc.shared
+			m.sweepStoreGenerations(t.Context())
+			if ckpts.sweeps.Load() != tc.wantCkpts || tpls.sweeps.Load() != tc.wantTpls {
+				t.Errorf("sweeps = checkpoints %d, templates %d, want %d, %d", ckpts.sweeps.Load(), tpls.sweeps.Load(), tc.wantCkpts, tc.wantTpls)
+			}
+		})
+	}
+}
+
 type stallingMetasStore struct {
 	store.Store
 	entered chan struct{}
@@ -378,17 +402,27 @@ func (s *stallingMetasStore) Metas(ctx context.Context) ([]store.Record, error)
 	return s.Store.Metas(ctx)
 }
 
+type sweepCountingStore struct {
+	store.Store
+	sweeps atomic.Int32
+}
+
+func (s *sweepCountingStore) SweepGenerations(ctx context.Context) error {
+	s.sweeps.Add(1)
+	return s.Store.SweepGenerations(ctx)
+}
+
 type stallingSweepStore struct {
 	store.Store
 	entered chan struct{}
 	release chan struct{}
 }
 
-func (s *stallingSweepStore) SweepGenerations() error {
+func (s *stallingSweepStore) SweepGenerations(ctx context.Context) error {
 	select {
 	case s.entered <- struct{}{}:
 	default:
 	}
 	<-s.release
-	return s.Store.SweepGenerations()
+	return s.Store.SweepGenerations(ctx)
 }
diff --git a/sandboxd/pool/expiry_test.go b/sandboxd/pool/expiry_test.go
index cf5848ff..7e2e0ec9 100644
--- a/sandboxd/pool/expiry_test.go
+++ b/sandboxd/pool/expiry_test.go
@@ -312,10 +312,10 @@ type failingPublishStore struct {
 	publishes atomic.Int32
 }
 
-func (s *failingPublishStore) Publish(ctx context.Context, staging, id string) error {
+func (s *failingPublishStore) Publish(ctx context.Context, staging, id, lineage string) error {
 	s.publishes.Add(1)
 	if s.fail.Load() {
 		return errors.New("store unavailable")
 	}
-	return s.Store.Publish(ctx, staging, id)
+	return s.Store.Publish(ctx, staging, id, lineage)
 }
diff --git a/sandboxd/pool/failover_test.go b/sandboxd/pool/failover_test.go
index 45185463..7c263a8e 100644
--- a/sandboxd/pool/failover_test.go
+++ b/sandboxd/pool/failover_test.go
@@ -1914,7 +1914,7 @@ func seedArchiveCk(t *testing.T, m *Manager, id, sandboxID string) {
 	if err := os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(meta), 0o600); err != nil {
 		t.Fatalf("meta: %v", err)
 	}
-	if err := m.ckpts.Publish(t.Context(), staging, id); err != nil {
+	if err := m.ckpts.Publish(t.Context(), staging, id, ""); err != nil {
 		t.Fatalf("publish: %v", err)
 	}
 }
@@ -1982,9 +1982,9 @@ type publishCountingStore struct {
 	n atomic.Int32
 }
 
-func (s *publishCountingStore) Publish(ctx context.Context, staging, id string) error {
+func (s *publishCountingStore) Publish(ctx context.Context, staging, id, lineage string) error {
 	s.n.Add(1)
-	return s.Store.Publish(ctx, staging, id)
+	return s.Store.Publish(ctx, staging, id, lineage)
 }
 
 type unreadableStore struct {
diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go
index 73fd5cb3..ca47f286 100644
--- a/sandboxd/pool/pool.go
+++ b/sandboxd/pool/pool.go
@@ -676,10 +676,13 @@ func (m *Manager) sweepStoreGenerations(ctx context.Context) {
 	}
 	defer m.genSweeping.Store(false)
 	logger := log.WithFunc("pool.sweepStoreGenerations")
-	if err := m.ckpts.SweepGenerations(); err != nil {
+	if err := m.ckpts.SweepGenerations(ctx); err != nil {
 		logger.Error(ctx, err, "sweep checkpoint generations")
 	}
-	if err := m.tpls.SweepGenerations(); err != nil {
+	if m.ckptsShared {
+		return
+	}
+	if err := m.tpls.SweepGenerations(ctx); err != nil {
 		logger.Error(ctx, err, "sweep template generations")
 	}
 }
diff --git a/sandboxd/pool/promote_test.go b/sandboxd/pool/promote_test.go
index 43304cc6..c33f895a 100644
--- a/sandboxd/pool/promote_test.go
+++ b/sandboxd/pool/promote_test.go
@@ -441,7 +441,7 @@ func TestTemplatesListWhatTheNodeHoldsAcrossARestart(t *testing.T) {
 	if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+oldID+`","created_at":"2026-01-01T00:00:00Z"}`), 0o600); err != nil {
 		t.Fatalf("write keyless meta: %v", err)
 	}
-	if _, err = m.tpls.PublishDigested(t.Context(), staging, oldID); err != nil {
+	if _, err = m.tpls.PublishDigested(t.Context(), staging, oldID, ""); err != nil {
 		t.Fatalf("publish keyless record: %v", err)
 	}
 
diff --git a/sandboxd/pool/template.go b/sandboxd/pool/template.go
index 67186ac8..b2cc7d25 100644
--- a/sandboxd/pool/template.go
+++ b/sandboxd/pool/template.go
@@ -303,7 +303,7 @@ func (m *Manager) commitTemplate(ctx context.Context, staging string, rec templa
 	if err = os.WriteFile(filepath.Join(staging, store.MetaFile), meta, 0o600); err != nil {
 		return "", err
 	}
-	digest, err := m.tpls.PublishDigested(ctx, staging, id)
+	digest, err := m.tpls.PublishDigested(ctx, staging, id, "")
 	if err != nil {
 		return "", fmt.Errorf("publish template: %w", err)
 	}
diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go
index a30b5bf4..1dc40dc5 100644
--- a/sandboxd/store/dir/dir.go
+++ b/sandboxd/store/dir/dir.go
@@ -56,11 +56,11 @@ func (d *Store) Stage(id string) (string, error) {
 	return os.MkdirTemp(d.root, id+"-*.tmp")
 }
 
-func (d *Store) Publish(ctx context.Context, staging, id string) error {
+func (d *Store) Publish(ctx context.Context, staging, id, _ string) error {
 	return d.publish(ctx, staging, id, "")
 }
 
-func (d *Store) PublishDigested(ctx context.Context, staging, id string) (string, error) {
+func (d *Store) PublishDigested(ctx context.Context, staging, id, _ string) (string, error) {
 	digest, err := hashExport(ctx, filepath.Join(staging, store.ExportDir))
 	if err != nil {
 		return "", fmt.Errorf("digest export: %w", err)
@@ -155,10 +155,14 @@ func (d *Store) SweepStaging() error {
 	if err := utils.RemoveDirEntries(d.root, func(name string) bool { return strings.HasSuffix(name, ".tmp") }); err != nil {
 		return err
 	}
-	return d.SweepGenerations()
+	return d.sweepRecords()
 }
 
-func (d *Store) SweepGenerations() error {
+func (d *Store) SweepGenerations(context.Context) error {
+	return d.sweepRecords()
+}
+
+func (d *Store) sweepRecords() error {
 	entries, err := os.ReadDir(d.root)
 	if err != nil {
 		return err
diff --git a/sandboxd/store/dir/dir_test.go b/sandboxd/store/dir/dir_test.go
index 16aa9680..144ed61c 100644
--- a/sandboxd/store/dir/dir_test.go
+++ b/sandboxd/store/dir/dir_test.go
@@ -134,7 +134,7 @@ func TestPublishDigestedChunkBoundaries(t *testing.T) {
 			if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(metaJSON(tt.name)), 0o600); err != nil {
 				t.Fatalf("write meta: %v", err)
 			}
-			digest, err := st.PublishDigested(t.Context(), staging, id)
+			digest, err := st.PublishDigested(t.Context(), staging, id, "")
 			if err != nil {
 				t.Fatalf("PublishDigested: %v", err)
 			}
@@ -216,7 +216,7 @@ func TestPublishDigestedFailureAndFreshRetry(t *testing.T) {
 	if err = os.Mkdir(blocker, 0o750); err != nil {
 		t.Fatalf("block digest commit: %v", err)
 	}
-	if digest, publishErr := st.PublishDigested(t.Context(), failed, id); publishErr == nil || digest != "" {
+	if digest, publishErr := st.PublishDigested(t.Context(), failed, id, ""); publishErr == nil || digest != "" {
 		t.Fatalf("PublishDigested with blocked sidecar = %q, %v, want empty/error", digest, publishErr)
 	}
 	dir, meta, digest, err := st.Fetch(t.Context(), id)
@@ -274,7 +274,7 @@ func TestSweepGenerationsPairsCurrentAndSupersededDigests(t *testing.T) {
 		backdate(t, path)
 	}
 
-	if err = st.SweepGenerations(); err != nil {
+	if err = st.SweepGenerations(t.Context()); err != nil {
 		t.Fatalf("SweepGenerations: %v", err)
 	}
 	for _, path := range []string{firstGen, firstSidecar} {
@@ -311,7 +311,7 @@ func TestPublishRetriesAfterExpiredInstall(t *testing.T) {
 		t.Fatalf("stage retry: %v", err)
 	}
 	seedRecord(t, retry, "same")
-	if publishErr := st.Publish(t.Context(), retry, id); publishErr == nil {
+	if publishErr := st.Publish(t.Context(), retry, id, ""); publishErr == nil {
 		t.Fatal("expired installed generation was revived")
 	}
 	if removeErr := os.RemoveAll(retry); removeErr != nil {
@@ -320,7 +320,7 @@ func TestPublishRetriesAfterExpiredInstall(t *testing.T) {
 	if _, readErr := st.ReadMeta(t.Context(), id); !errors.Is(readErr, store.ErrNotFound) {
 		t.Fatalf("expired retry committed meta: %v", readErr)
 	}
-	if sweepErr := st.SweepGenerations(); sweepErr != nil {
+	if sweepErr := st.SweepGenerations(t.Context()); sweepErr != nil {
 		t.Fatalf("sweep expired generation: %v", sweepErr)
 	}
 	if _, statErr := os.Stat(gen); !errors.Is(statErr, fs.ErrNotExist) {
@@ -454,7 +454,7 @@ func TestSweepReclaimsOnlyAgedUncommittedGenerations(t *testing.T) {
 	if mkdirErr := os.MkdirAll(commitBlocker, 0o750); mkdirErr != nil {
 		t.Fatalf("block meta commit: %v", mkdirErr)
 	}
-	if publishErr := writer.Publish(t.Context(), staging, publishID); publishErr == nil {
+	if publishErr := writer.Publish(t.Context(), staging, publishID, ""); publishErr == nil {
 		t.Fatal("publish with blocked meta commit succeeded")
 	}
 	fresh := filepath.Join(root, publishID, store.ExportGen(meta))
@@ -513,7 +513,7 @@ func TestSweepGenerationsToleratesConcurrentDelete(t *testing.T) {
 			if err := os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(metaJSON(id)), 0o600); err != nil {
 				return err
 			}
-			if err := st.Publish(ctx, staging, id); err != nil {
+			if err := st.Publish(ctx, staging, id, ""); err != nil {
 				return fmt.Errorf("publish: %w", err)
 			}
 			if err := st.Delete(ctx, id); err != nil {
@@ -529,7 +529,7 @@ func TestSweepGenerationsToleratesConcurrentDelete(t *testing.T) {
 				return nil
 			default:
 			}
-			if err := st.SweepGenerations(); err != nil {
+			if err := st.SweepGenerations(t.Context()); err != nil {
 				return fmt.Errorf("sweep: %w", err)
 			}
 		}
@@ -646,7 +646,7 @@ func mustPublish(t testing.TB, st *Store, id, metaID string) {
 		t.Fatalf("stage %s: %v", metaID, err)
 	}
 	seedRecord(t, staging, metaID)
-	if err := st.Publish(t.Context(), staging, id); err != nil {
+	if err := st.Publish(t.Context(), staging, id, ""); err != nil {
 		t.Fatalf("publish %s: %v", metaID, err)
 	}
 }
@@ -658,7 +658,7 @@ func mustPublishDigested(t *testing.T, st *Store, id, metaID string) string {
 		t.Fatalf("stage %s: %v", metaID, err)
 	}
 	seedRecord(t, staging, metaID)
-	digest, err := st.PublishDigested(t.Context(), staging, id)
+	digest, err := st.PublishDigested(t.Context(), staging, id, "")
 	if err != nil {
 		t.Fatalf("publish digested %s: %v", metaID, err)
 	}
diff --git a/sandboxd/store/peer/transport.go b/sandboxd/store/peer/transport.go
index 059fdb91..262dc929 100644
--- a/sandboxd/store/peer/transport.go
+++ b/sandboxd/store/peer/transport.go
@@ -15,7 +15,9 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/cocoonstack/sandbox/sandboxd/store"
@@ -31,6 +33,10 @@ const (
 
 	// recordTrailer is written last, so EOF without it means the transfer stopped early.
 	recordTrailer = ".record-complete"
+
+	// paxFileSize marks the entry that declares a sparse file; paxChunk marks one present chunk as "<index>:<sha256>".
+	paxFileSize = "SANDBOX.size"
+	paxChunk    = "SANDBOX.chunk"
 )
 
 // ErrNotFound reports that a peer does not hold the requested record.
@@ -109,6 +115,7 @@ func TarRecord(exportDir string, meta []byte, w io.Writer) error {
 // Untar writes a tar stream into dst, validating every entry name against traversal.
 func Untar(r io.Reader, dst string) error {
 	tr := tar.NewReader(r)
+	sizes := map[string]int64{}
 	var written int64
 	var complete bool
 	for {
@@ -143,13 +150,59 @@ func Untar(r io.Reader, dst string) error {
 			if written > maxRecordBytes {
 				return fmt.Errorf("untar: record exceeds %d bytes", int64(maxRecordBytes))
 			}
-			if err := writeFile(target, tr, os.FileMode(hdr.Mode).Perm()); err != nil { //nolint:gosec // Perm masks to 0777
+			if err := untarRegular(tr, hdr, target, sizes); err != nil {
 				return err
 			}
 		}
 	}
 }
 
+func untarRegular(r io.Reader, hdr *tar.Header, target string, sizes map[string]int64) error {
+	mode := os.FileMode(hdr.Mode).Perm() //nolint:gosec // Perm masks to 0777
+	if raw, ok := hdr.PAXRecords[paxFileSize]; ok {
+		return createSparse(target, raw, mode, sizes)
+	}
+	if raw, ok := hdr.PAXRecords[paxChunk]; ok {
+		return writeChunk(target, raw, r, sizes)
+	}
+	return writeFile(target, r, mode)
+}
+
+func createSparse(target, rawSize string, mode os.FileMode, sizes map[string]int64) error {
+	size, err := strconv.ParseInt(rawSize, 10, 64)
+	if err != nil || size < 0 || size > maxRecordBytes {
+		return fmt.Errorf("untar: bad size %q for %s", rawSize, target)
+	}
+	f, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, cmp.Or(mode, os.FileMode(0o600))) //nolint:gosec // target resolved by safeJoin
+	if err != nil {
+		return fmt.Errorf("untar create %s: %w", target, err)
+	}
+	if err = f.Truncate(size); err != nil {
+		_ = f.Close()
+		return fmt.Errorf("untar size %s: %w", target, err)
+	}
+	sizes[target] = size
+	return f.Close()
+}
+
+func writeChunk(target, raw string, r io.Reader, sizes map[string]int64) (err error) {
+	size, ok := sizes[target]
+	rawIndex, sum, found := strings.Cut(raw, ":")
+	index, parseErr := strconv.ParseInt(rawIndex, 10, 64)
+	if !ok || !found || parseErr != nil {
+		return fmt.Errorf("untar: bad chunk entry %q for %s", raw, target)
+	}
+	f, err := os.OpenFile(target, os.O_WRONLY, 0) //nolint:gosec // target resolved by safeJoin
+	if err != nil {
+		return fmt.Errorf("untar open %s: %w", target, err)
+	}
+	defer func() { err = errors.Join(err, f.Close()) }()
+	if err = store.WriteChunk(f, size, index, sum, r); err != nil {
+		return fmt.Errorf("untar %s: %w", target, err)
+	}
+	return nil
+}
+
 func writeFile(target string, r io.Reader, mode os.FileMode) error {
 	mode = cmp.Or(mode, os.FileMode(0o600))
 	f, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, mode) //nolint:gosec // target resolved by safeJoin
@@ -187,8 +240,9 @@ func safeJoin(root, name string) (string, error) {
 	return target, nil
 }
 
-// tarInto emits only regular files and directories: a symlink would steer the reader outside dst.
+// tarInto emits only regular files and directories (a symlink would steer the reader outside dst); a file is a size entry plus its non-zero chunks.
 func tarInto(src, prefix string, tw *tar.Writer) error {
+	scratch := sync.OnceValue(func() []byte { return make([]byte, store.DigestChunkSize) })
 	err := filepath.WalkDir(src, func(path string, entry fs.DirEntry, err error) error {
 		if err != nil {
 			return err
@@ -216,23 +270,39 @@ func tarInto(src, prefix string, tw *tar.Writer) error {
 			})
 		}
 		if err = tw.WriteHeader(&tar.Header{
-			Name:     name,
-			Mode:     int64(fi.Mode().Perm()),
-			Size:     fi.Size(),
-			Typeflag: tar.TypeReg,
+			Name:       name,
+			Mode:       int64(fi.Mode().Perm()),
+			Typeflag:   tar.TypeReg,
+			PAXRecords: map[string]string{paxFileSize: strconv.FormatInt(fi.Size(), 10)},
 		}); err != nil {
 			return err
 		}
-		f, err := os.Open(path) //nolint:gosec // path comes from WalkDir over src
-		if err != nil {
-			return err
-		}
-		defer func() { _ = f.Close() }()
-		_, err = io.Copy(tw, f)
-		return err
+		return tarChunks(path, name, fi.Size(), tw, scratch)
 	})
 	if err != nil {
 		return fmt.Errorf("tar %s: %w", src, err)
 	}
 	return nil
 }
+
+func tarChunks(path, name string, size int64, tw *tar.Writer, scratch func() []byte) error {
+	f, openErr := os.Open(path) //nolint:gosec // path comes from WalkDir over src
+	if openErr != nil {
+		return openErr
+	}
+	defer func() { _ = f.Close() }()
+	_, err := store.ScanFile(f, name, size, scratch, func(c store.Chunk) error {
+		if err := tw.WriteHeader(&tar.Header{
+			Name:       name,
+			Mode:       0o600,
+			Size:       int64(len(c.Data)),
+			Typeflag:   tar.TypeReg,
+			PAXRecords: map[string]string{paxChunk: strconv.FormatInt(c.Index, 10) + ":" + c.SHA256},
+		}); err != nil {
+			return err
+		}
+		_, err := tw.Write(c.Data)
+		return err
+	})
+	return err
+}
diff --git a/sandboxd/store/peer/transport_test.go b/sandboxd/store/peer/transport_test.go
index 6ac3862b..4cb26170 100644
--- a/sandboxd/store/peer/transport_test.go
+++ b/sandboxd/store/peer/transport_test.go
@@ -3,6 +3,7 @@ package peer
 import (
 	"archive/tar"
 	"bytes"
+	"crypto/sha256"
 	"errors"
 	"io"
 	"net/http"
@@ -10,8 +11,11 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"syscall"
 	"testing"
 	"time"
+
+	"github.com/cocoonstack/sandbox/sandboxd/store"
 )
 
 func TestTarUntarRoundTrip(t *testing.T) {
@@ -296,6 +300,75 @@ func TestTarRecordUntarRoundTripComplete(t *testing.T) {
 	}
 }
 
+func TestTarRecordKeepsSparseFilesSparse(t *testing.T) {
+	size := 10*store.DigestChunkSize + 4096
+	exportDir := t.TempDir()
+	src := filepath.Join(exportDir, "cow.raw")
+	f, err := os.Create(src)
+	if err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	if err = f.Truncate(size); err != nil {
+		t.Fatalf("truncate: %v", err)
+	}
+	for off, data := range map[int64]string{0: "sparse-head", 4*store.DigestChunkSize + 7: "sparse-mid", 8*store.DigestChunkSize + 5: "tail"} {
+		if _, err = f.WriteAt([]byte(data), off); err != nil {
+			t.Fatalf("write at %d: %v", off, err)
+		}
+	}
+	if err = f.Close(); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+	if err = os.WriteFile(filepath.Join(exportDir, "snapshot.json"), []byte("{}"), 0o600); err != nil {
+		t.Fatalf("write snapshot: %v", err)
+	}
+	want := sha256OfFile(t, src)
+
+	var buf bytes.Buffer
+	if err = TarRecord(exportDir, []byte(`{"id":"ck_0000000000000001"}`), &buf); err != nil {
+		t.Fatalf("TarRecord: %v", err)
+	}
+	sent := int64(buf.Len())
+	if limit := 3*store.DigestChunkSize + 1<<20; sent > limit {
+		t.Errorf("stream carries %d bytes for a %d-byte file with 3 data chunks, want at most %d", sent, size, limit)
+	}
+	dst := t.TempDir()
+	if err = Untar(&buf, dst); err != nil {
+		t.Fatalf("Untar: %v", err)
+	}
+	got := filepath.Join(dst, "export", "cow.raw")
+	info, err := os.Stat(got)
+	if err != nil || info.Size() != size {
+		t.Fatalf("pulled size = %v, %v, want %d", info, err, size)
+	}
+	if sha256OfFile(t, got) != want {
+		t.Error("pulled content differs from the source")
+	}
+	if allocated := info.Sys().(*syscall.Stat_t).Blocks * 512; allocated > sent {
+		t.Errorf("pulled file allocates %d bytes, more than the %d sent", allocated, sent)
+	}
+	if body, err := os.ReadFile(filepath.Join(dst, "export", "snapshot.json")); err != nil || string(body) != "{}" {
+		t.Errorf("small file = %q, %v", body, err)
+	}
+}
+
+func TestUntarRejectsDamagedChunk(t *testing.T) {
+	exportDir := t.TempDir()
+	if err := os.WriteFile(filepath.Join(exportDir, "mem"), []byte("guest-pages-payload"), 0o600); err != nil {
+		t.Fatalf("setup: %v", err)
+	}
+	var buf bytes.Buffer
+	if err := TarRecord(exportDir, []byte(`{"id":"ck_1"}`), &buf); err != nil {
+		t.Fatalf("TarRecord: %v", err)
+	}
+	raw := buf.Bytes()
+	raw[bytes.Index(raw, []byte("guest-pages-payload"))] ^= 0xff
+	err := Untar(bytes.NewReader(raw), t.TempDir())
+	if err == nil || !strings.Contains(err.Error(), "checksum") {
+		t.Errorf("Untar of a corrupted chunk = %v, want a checksum error", err)
+	}
+}
+
 func tarDir(src string, w io.Writer) error {
 	tw := tar.NewWriter(w)
 	defer func() { _ = tw.Close() }()
@@ -307,3 +380,17 @@ func tarDir(src string, w io.Writer) error {
 	}
 	return tw.Close()
 }
+
+func sha256OfFile(t *testing.T, path string) [sha256.Size]byte {
+	t.Helper()
+	f, err := os.Open(path)
+	if err != nil {
+		t.Fatalf("open %s: %v", path, err)
+	}
+	defer func() { _ = f.Close() }()
+	h := sha256.New()
+	if _, err = io.Copy(h, f); err != nil {
+		t.Fatalf("hash %s: %v", path, err)
+	}
+	return [sha256.Size]byte(h.Sum(nil))
+}
diff --git a/sandboxd/store/s3/digest.go b/sandboxd/store/s3/digest.go
deleted file mode 100644
index 5d6f7566..00000000
--- a/sandboxd/store/s3/digest.go
+++ /dev/null
@@ -1,62 +0,0 @@
-package s3
-
-import (
-	"crypto/sha256"
-	"fmt"
-	"hash"
-	"io"
-
-	"github.com/cocoonstack/sandbox/sandboxd/store"
-)
-
-type digestReader struct {
-	body       io.Reader
-	chunkHash  hash.Hash
-	path       string
-	size       int64
-	read       int64
-	chunkBytes int64
-	chunks     [][sha256.Size]byte
-	finished   bool
-}
-
-func newDigestReader(body io.Reader, path string, size int64) *digestReader {
-	return &digestReader{body: body, chunkHash: sha256.New(), path: path, size: size}
-}
-
-func (r *digestReader) Read(p []byte) (int, error) {
-	n, err := r.body.Read(p)
-	r.read += int64(n)
-	for offset := 0; offset < n; {
-		remaining := int(store.DigestChunkSize - r.chunkBytes)
-		written := min(n-offset, remaining)
-		_, _ = r.chunkHash.Write(p[offset : offset+written])
-		r.chunkBytes += int64(written)
-		offset += written
-		if r.chunkBytes == store.DigestChunkSize {
-			r.finishChunk()
-		}
-	}
-	return n, err
-}
-
-func (r *digestReader) Digest() (store.DigestFile, error) {
-	if !r.finished {
-		if r.chunkBytes > 0 {
-			r.finishChunk()
-		}
-		r.finished = true
-	}
-	if r.read != r.size {
-		return store.DigestFile{}, fmt.Errorf("read %s: got %d bytes, want %d", r.path, r.read, r.size)
-	}
-	return store.DigestFile{Path: r.path, Size: r.size, Chunks: r.chunks}, nil
-}
-
-func (r *digestReader) finishChunk() {
-	var sum [sha256.Size]byte
-	r.chunkHash.Sum(sum[:0])
-	r.chunks = append(r.chunks, sum)
-	r.chunkHash.Reset()
-	r.chunkBytes = 0
-}
diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go
index 45b216f0..0cb1848c 100644
--- a/sandboxd/store/s3/s3.go
+++ b/sandboxd/store/s3/s3.go
@@ -1,26 +1,31 @@
 // Package s3 is the object-store record backend for nodes without a
-// shared POSIX namespace: <prefix><id>/{export-<gen>/...,meta.json}
-// objects, meta.json uploaded last as the commit marker (S3 has no atomic
-// multi-object rename). The aws dependency is scoped to this package.
+// shared POSIX namespace: <prefix><id>/meta.json (live records only) plus
+// <prefix>manifests/<id>/<gen>.json and the content-addressed chunks the
+// manifests name, meta.json uploaded last as the commit marker (S3 has no
+// atomic multi-object rename). Exports are 16 MiB chunks keyed by SHA-256;
+// holes and all-zero chunks are absent. The aws dependency stays in this package.
 package s3
 
 import (
 	"bytes"
+	"cmp"
 	"context"
+	"encoding/json/v2"
 	"errors"
 	"fmt"
 	"io"
 	"io/fs"
 	"os"
+	"path"
 	"path/filepath"
 	"regexp"
 	"slices"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	awsconfig "github.com/aws/aws-sdk-go-v2/config"
-	"github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager"
 	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
 	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/aws/smithy-go"
@@ -33,24 +38,42 @@ import (
 
 const (
 	digestMetadataKey   = "content-digest"
-	uploadPartSizeBytes = 16 << 20
-	uploadConcurrency   = 8 // multipart parts in flight per transfer
+	manifestsDir        = "manifests"
+	manifestExt         = ".json"
+	chunksDir           = "chunks"
+	lineagesDir         = "lineages"
+	uploadConcurrency   = 8 // chunk PUTs in flight per publish
 	metaReadConcurrency = 8
-	publishConcurrency  = 4 // files in parallel; each already multiparts internally
-	fetchConcurrency    = 4
-	fetchBudget         = 30 * time.Minute
+	publishConcurrency  = 4 // files scanned in parallel
+	fetchConcurrency    = 8 // chunk GETs in flight per fetch
+	transferBudget      = 30 * time.Minute
 	deleteBatch         = 1000 // DeleteObjects caps one request at 1000 keys
 )
 
+var chunkBufs = sync.Pool{New: func() any {
+	buf := make([]byte, store.DigestChunkSize)
+	return &buf
+}}
+
 type publishFile struct {
-	path       string
-	key        string
-	digestPath string
+	path string
+	rel  string
+}
+
+type manifest struct {
+	Lineage string         `json:"lineage"`
+	Files   []manifestFile `json:"files"`
 }
 
-type transferManager interface {
-	UploadObject(context.Context, *transfermanager.UploadObjectInput, ...func(*transfermanager.Options)) (*transfermanager.UploadObjectOutput, error)
-	DownloadObject(context.Context, *transfermanager.DownloadObjectInput, ...func(*transfermanager.Options)) (*transfermanager.DownloadObjectOutput, error)
+type manifestFile struct {
+	Path   string          `json:"path"`
+	Size   int64           `json:"size"`
+	Chunks []manifestChunk `json:"chunks"`
+}
+
+type manifestChunk struct {
+	Index  int64  `json:"index"`
+	SHA256 string `json:"sha256"`
 }
 
 var _ store.Store = (*Store)(nil)
@@ -58,7 +81,6 @@ var _ store.Store = (*Store)(nil)
 // Store stages locally and publishes to the bucket; idRe names the instance's id namespace.
 type Store struct {
 	client  *awss3.Client
-	tm      transferManager
 	bucket  string
 	prefix  string
 	staging string
@@ -88,25 +110,20 @@ func New(ctx context.Context, cfg store.S3Config, stagingRoot string, idRe *rege
 		}
 		o.UsePathStyle = cfg.ForcePathStyle
 	})
-	// snapshot exports are hundreds of MB: multipart concurrency keeps transfers bandwidth-bound.
-	tm := transfermanager.New(client, func(o *transfermanager.Options) {
-		o.PartSizeBytes = uploadPartSizeBytes
-		o.Concurrency = uploadConcurrency
-	})
-	return &Store{client: client, tm: tm, bucket: cfg.Bucket, prefix: cfg.Prefix, staging: stagingRoot, idRe: idRe}, nil
+	return &Store{client: client, bucket: cfg.Bucket, prefix: cfg.Prefix, staging: stagingRoot, idRe: idRe}, nil
 }
 
 func (s *Store) Stage(id string) (string, error) {
 	return os.MkdirTemp(s.staging, id+"-*.tmp")
 }
 
-func (s *Store) Publish(ctx context.Context, staging, id string) error {
-	_, err := s.publish(ctx, staging, id, false)
+func (s *Store) Publish(ctx context.Context, staging, id, lineage string) error {
+	_, err := s.publish(ctx, staging, id, lineage, false)
 	return err
 }
 
-func (s *Store) PublishDigested(ctx context.Context, staging, id string) (string, error) {
-	return s.publish(ctx, staging, id, true)
+func (s *Store) PublishDigested(ctx context.Context, staging, id, lineage string) (string, error) {
+	return s.publish(ctx, staging, id, lineage, true)
 }
 
 func (s *Store) Fetch(ctx context.Context, id string) (string, []byte, string, error) {
@@ -122,7 +139,7 @@ func (s *Store) Fetch(ctx context.Context, id string) (string, []byte, string, e
 		return export, meta, digest, nil
 	}
 	flight := s.fetches.DoChan(gen, func() (any, error) {
-		fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), fetchBudget)
+		fctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), transferBudget)
 		defer cancel()
 		return nil, s.populate(fctx, id, meta, gen)
 	})
@@ -143,22 +160,9 @@ func (s *Store) ReadMeta(ctx context.Context, id string) ([]byte, error) {
 }
 
 func (s *Store) Metas(ctx context.Context) ([]store.Record, error) {
-	// delimiter listing yields one CommonPrefix per record instead of every export object.
-	var ids []string
-	p := awss3.NewListObjectsV2Paginator(s.client, &awss3.ListObjectsV2Input{
-		Bucket: &s.bucket, Prefix: &s.prefix, Delimiter: new("/"),
-	})
-	for p.HasMorePages() {
-		page, err := p.NextPage(ctx)
-		if err != nil {
-			return nil, fmt.Errorf("list %s: %w", s.prefix, err)
-		}
-		for _, cp := range page.CommonPrefixes {
-			id := strings.TrimSuffix(strings.TrimPrefix(*cp.Prefix, s.prefix), "/")
-			if s.idRe.MatchString(id) {
-				ids = append(ids, id)
-			}
-		}
+	ids, err := s.recordIDs(ctx)
+	if err != nil {
+		return nil, err
 	}
 	recs := make([]store.Record, len(ids))
 	g, gctx := errgroup.WithContext(ctx)
@@ -191,83 +195,141 @@ func (s *Store) SetLabels(ctx context.Context, id string, labels []byte) error {
 	if labels == nil {
 		return s.deleteKeys(ctx, []string{key})
 	}
-	return s.uploadReader(ctx, key, bytes.NewReader(labels), int64(len(labels)), nil)
+	return s.putObject(ctx, key, labels, nil)
 }
 
 func (s *Store) Delete(ctx context.Context, id string) error {
 	_ = os.RemoveAll(filepath.Join(s.staging, "cache", id))
-	metaKey := s.key(id, store.MetaFile)
-	keys, err := s.list(ctx, s.key(id, "")+"/")
+	objs, err := s.list(ctx, s.manifestPrefix(id))
 	if err != nil {
 		return err
 	}
-	// keep the commit marker so a failed export cleanup remains discoverable on retry.
-	keys = slices.DeleteFunc(keys, func(key string) bool { return key == metaKey })
-	if err := s.deleteKeys(ctx, keys); err != nil {
+	drop := []string{s.key(id, store.LabelsFile)}
+	for _, key := range objectKeys(objs) {
+		man, refreshErr := s.refreshManifest(ctx, key)
+		if refreshErr != nil {
+			return refreshErr
+		}
+		lineage := cmp.Or(man.Lineage, id)
+		if marker := s.lineageKey(lineage, id); lineage != id && !slices.Contains(drop, marker) {
+			drop = append(drop, marker)
+		}
+	}
+	if err = s.deleteKeys(ctx, drop); err != nil {
 		return err
 	}
-	return s.deleteKeys(ctx, []string{metaKey})
+	return s.deleteKeys(ctx, []string{s.key(id, store.MetaFile)})
 }
 
+func (s *Store) SweepGenerations(context.Context) error { return nil }
+
 // SweepStaging clears local staging residue at startup.
 func (s *Store) SweepStaging() error {
-	return utils.RemoveDirEntries(s.staging, nil) // objects orphaned before meta.json fall to the bucket's lifecycle rule
+	return utils.RemoveDirEntries(s.staging, nil)
 }
 
-func (s *Store) SweepGenerations() error { return nil }
-
 func (s *Store) readMeta(ctx context.Context, id string) ([]byte, string, error) {
-	meta, metadata, err := s.getObject(ctx, s.key(id, store.MetaFile))
+	meta, out, err := s.getObject(ctx, s.key(id, store.MetaFile))
 	if err != nil {
 		return nil, "", err
 	}
-	return meta, metadata[digestMetadataKey], nil
+	return meta, out.Metadata[digestMetadataKey], nil
 }
 
-func (s *Store) getObject(ctx context.Context, key string) ([]byte, map[string]string, error) {
-	out, err := s.client.GetObject(ctx, &awss3.GetObjectInput{Bucket: &s.bucket, Key: &key})
+func (s *Store) currentManifest(ctx context.Context, id string) (manifest, error) {
+	meta, _, err := s.getObject(ctx, s.key(id, store.MetaFile))
 	if err != nil {
-		apiErr, ok := errors.AsType[smithy.APIError](err)
-		if ok && (apiErr.ErrorCode() == "NoSuchKey" || apiErr.ErrorCode() == "NotFound") {
-			return nil, nil, store.ErrNotFound
-		}
-		return nil, nil, fmt.Errorf("object %s: %w", key, err)
+		return manifest{}, err
+	}
+	return s.readManifest(ctx, id, store.ExportGen(meta))
+}
+
+func (s *Store) readManifest(ctx context.Context, id, gen string) (manifest, error) {
+	raw, _, err := s.getObject(ctx, s.manifestKey(id, gen))
+	if err != nil {
+		return manifest{}, err
+	}
+	var man manifest
+	if err := json.Unmarshal(raw, &man); err != nil {
+		return manifest{}, fmt.Errorf("decode manifest of %s/%s: %w", id, gen, err)
+	}
+	return man, nil
+}
+
+func (s *Store) getObject(ctx context.Context, key string) ([]byte, *awss3.GetObjectOutput, error) {
+	out, err := s.get(ctx, key)
+	if err != nil {
+		return nil, nil, err
 	}
 	defer func() { _ = out.Body.Close() }()
 	body, err := io.ReadAll(out.Body)
 	if err != nil {
 		return nil, nil, err
 	}
-	return body, out.Metadata, nil
+	return body, out, nil
 }
 
-func (s *Store) publish(ctx context.Context, staging, id string, digested bool) (string, error) {
+func (s *Store) get(ctx context.Context, key string) (*awss3.GetObjectOutput, error) {
+	out, err := s.client.GetObject(ctx, &awss3.GetObjectInput{Bucket: &s.bucket, Key: &key})
+	if err != nil {
+		apiErr, ok := errors.AsType[smithy.APIError](err)
+		if ok && (apiErr.ErrorCode() == "NoSuchKey" || apiErr.ErrorCode() == "NotFound") {
+			return nil, store.ErrNotFound
+		}
+		return nil, fmt.Errorf("object %s: %w", key, err)
+	}
+	return out, nil
+}
+
+func (s *Store) publish(ctx context.Context, staging, id, lineage string, digested bool) (string, error) {
+	ctx, cancel := context.WithTimeout(ctx, transferBudget)
+	defer cancel()
+	lineage = cmp.Or(lineage, id)
 	metaRaw, err := os.ReadFile(filepath.Join(staging, store.MetaFile)) //nolint:gosec // our own staging dir
 	if err != nil {
 		return "", fmt.Errorf("staging has no %s: %w", store.MetaFile, err)
 	}
-	files, err := s.publishFiles(staging, id, store.ExportGen(metaRaw))
+	gen := store.ExportGen(metaRaw)
+	files, err := publishFiles(staging)
+	if err != nil {
+		return "", err
+	}
+	have, err := s.referencedChunks(ctx, id, lineage)
 	if err != nil {
 		return "", err
 	}
+	man := manifest{Lineage: lineage, Files: make([]manifestFile, len(files))}
 	digestFiles := make([]store.DigestFile, len(files))
-	g, gctx := errgroup.WithContext(ctx)
-	g.SetLimit(publishConcurrency)
+	chunkGroup, chunkCtx := errgroup.WithContext(ctx)
+	chunkGroup.SetLimit(uploadConcurrency)
+	up := &chunkUploader{store: s, group: chunkGroup, ctx: chunkCtx, lineage: lineage, have: have}
+	fileGroup, fileCtx := errgroup.WithContext(chunkCtx)
+	fileGroup.SetLimit(publishConcurrency)
 	for i, file := range files {
-		g.Go(func() error {
-			if !digested {
-				return s.upload(gctx, file.key, file.path)
-			}
-			fileDigest, digestErr := s.uploadDigested(gctx, file.key, file.path, file.digestPath)
-			if digestErr == nil {
-				digestFiles[i] = fileDigest
-			}
-			return digestErr
+		fileGroup.Go(func() (err error) {
+			digestFiles[i], man.Files[i], err = up.scan(fileCtx, file)
+			return err
 		})
 	}
-	if err = g.Wait(); err != nil {
+	fileErr := fileGroup.Wait()
+	if err = chunkGroup.Wait(); err == nil {
+		err = fileErr
+	}
+	if err != nil {
+		return "", err
+	}
+	manifestRaw, err := json.Marshal(man)
+	if err != nil {
+		return "", fmt.Errorf("encode manifest: %w", err)
+	}
+	if err = s.putObject(ctx, s.manifestKey(id, gen), manifestRaw, nil); err != nil {
 		return "", err
 	}
+	if lineage != id {
+		if err = s.putObject(ctx, s.lineageKey(lineage, id), nil, nil); err != nil {
+			return "", err
+		}
+	}
 	digest := ""
 	var metadata map[string]string
 	if digested {
@@ -277,30 +339,16 @@ func (s *Store) publish(ctx context.Context, staging, id string, digested bool)
 		}
 		metadata = map[string]string{digestMetadataKey: digest}
 	}
-	if err = s.uploadReader(ctx, s.key(id, store.MetaFile), bytes.NewReader(metaRaw), int64(len(metaRaw)), metadata); err != nil {
+	if err = s.putObject(ctx, s.key(id, store.MetaFile), metaRaw, metadata); err != nil {
 		return "", err
 	}
-	// keep old generations: another node may have selected the previous meta; Delete reclaims them.
+	// keep old generations: another node may have selected the previous meta.
 	if err := os.RemoveAll(staging); err != nil {
 		return "", err
 	}
 	return digest, nil
 }
 
-func (s *Store) publishFiles(staging, id, gen string) ([]publishFile, error) {
-	root := filepath.Join(staging, store.ExportDir)
-	var files []publishFile
-	err := store.WalkExport(root, func(rel, path string, _ fs.FileInfo) error {
-		files = append(files, publishFile{
-			path:       path,
-			key:        s.key(id, gen+"/"+rel),
-			digestPath: rel,
-		})
-		return nil
-	})
-	return files, err
-}
-
 // populate downloads one cache generation and installs it atomically.
 func (s *Store) populate(ctx context.Context, id string, meta []byte, gen string) error {
 	if _, err := os.Stat(gen); err == nil {
@@ -311,22 +359,12 @@ func (s *Store) populate(ctx context.Context, id string, meta []byte, gen string
 		return err
 	}
 	defer func() { _ = os.RemoveAll(local) }()
-	exportPrefix := s.key(id, store.ExportGen(meta)) + "/"
-	keys, err := s.list(ctx, exportPrefix)
+	exportGen := store.ExportGen(meta)
+	man, err := s.readManifest(ctx, id, exportGen)
 	if err != nil {
 		return err
 	}
-	if len(keys) == 0 {
-		return store.ErrNotFound
-	}
-	g, gctx := errgroup.WithContext(ctx)
-	g.SetLimit(fetchConcurrency)
-	for _, key := range keys {
-		g.Go(func() error {
-			return s.download(gctx, key, filepath.Join(local, store.ExportDir, strings.TrimPrefix(key, exportPrefix)))
-		})
-	}
-	if err := g.Wait(); err != nil {
+	if err := s.materialize(ctx, man, filepath.Join(local, store.ExportDir)); err != nil {
 		return err
 	}
 	if err := os.WriteFile(filepath.Join(local, store.MetaFile), meta, 0o600); err != nil {
@@ -342,6 +380,55 @@ func (s *Store) populate(ctx context.Context, id string, meta []byte, gen string
 	return nil
 }
 
+// materialize sizes each file to its logical length and writes only its present chunks, so holes stay holes.
+func (s *Store) materialize(ctx context.Context, man manifest, root string) (err error) {
+	handles := make([]*os.File, len(man.Files))
+	defer func() {
+		for _, f := range handles {
+			if f != nil {
+				err = errors.Join(err, f.Close())
+			}
+		}
+	}()
+	for i, file := range man.Files {
+		target := filepath.Join(root, filepath.FromSlash(file.Path))
+		if err = os.MkdirAll(filepath.Dir(target), 0o750); err != nil {
+			return err
+		}
+		if handles[i], err = os.Create(target); err != nil { //nolint:gosec // path derives from our own temp dir
+			return err
+		}
+		if err = handles[i].Truncate(file.Size); err != nil {
+			return err
+		}
+	}
+	g, gctx := errgroup.WithContext(ctx)
+	g.SetLimit(fetchConcurrency)
+	for i, file := range man.Files {
+		for _, chunk := range file.Chunks {
+			g.Go(func() error {
+				return s.fetchChunk(gctx, s.chunkKey(man.Lineage, chunk.SHA256), handles[i], file.Size, chunk)
+			})
+		}
+	}
+	return g.Wait()
+}
+
+func (s *Store) fetchChunk(ctx context.Context, key string, f *os.File, size int64, chunk manifestChunk) error {
+	out, err := s.get(ctx, key)
+	if errors.Is(err, store.ErrNotFound) {
+		return fmt.Errorf("chunk %s is missing", key) // not ErrNotFound: the record exists but is damaged
+	}
+	if err != nil {
+		return err
+	}
+	defer func() { _ = out.Body.Close() }()
+	if err := store.WriteChunk(f, size, chunk.Index, chunk.SHA256, out.Body); err != nil {
+		return fmt.Errorf("chunk %s: %w", key, err)
+	}
+	return nil
+}
+
 func (s *Store) key(id, rest string) string {
 	if rest == "" {
 		return s.prefix + id
@@ -349,65 +436,106 @@ func (s *Store) key(id, rest string) string {
 	return s.prefix + id + "/" + filepath.ToSlash(rest)
 }
 
-func (s *Store) upload(ctx context.Context, key, path string) error {
-	f, err := os.Open(path) //nolint:gosec // path walked from our own staging dir
-	if err != nil {
-		return err
-	}
-	defer func() { _ = f.Close() }()
-	info, err := f.Stat()
-	if err != nil {
-		return err
+func (s *Store) manifestPrefix(id string) string {
+	return s.prefix + manifestsDir + "/" + id + "/"
+}
+
+func (s *Store) manifestKey(id, gen string) string {
+	return s.manifestPrefix(id) + gen + manifestExt
+}
+
+func (s *Store) lineageKey(lineage, id string) string {
+	return s.prefix + lineagesDir + "/" + lineage + "/" + id
+}
+
+func (s *Store) chunkKey(lineage, sha string) string {
+	return s.prefix + chunksDir + "/" + lineage + "/" + sha
+}
+
+func (s *Store) recordIDs(ctx context.Context) ([]string, error) {
+	// delimiter listing yields one CommonPrefix per record instead of every export object.
+	var ids []string
+	p := awss3.NewListObjectsV2Paginator(s.client, &awss3.ListObjectsV2Input{
+		Bucket: &s.bucket, Prefix: &s.prefix, Delimiter: new("/"),
+	})
+	for p.HasMorePages() {
+		page, err := p.NextPage(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("list %s: %w", s.prefix, err)
+		}
+		for _, cp := range page.CommonPrefixes {
+			id := strings.TrimSuffix(strings.TrimPrefix(*cp.Prefix, s.prefix), "/")
+			if s.idRe.MatchString(id) {
+				ids = append(ids, id)
+			}
+		}
 	}
-	return s.uploadReader(ctx, key, f, info.Size(), nil)
+	return ids, nil
 }
 
-func (s *Store) uploadDigested(ctx context.Context, key, path, digestPath string) (store.DigestFile, error) {
-	f, err := os.Open(path) //nolint:gosec // path walked from our own staging dir
-	if err != nil {
-		return store.DigestFile{}, err
+func (s *Store) referencedChunks(ctx context.Context, id, lineage string) (map[string]struct{}, error) {
+	ids := []string{id}
+	if lineage != id {
+		markers, err := s.list(ctx, s.lineageKey(lineage, ""))
+		if err != nil {
+			return nil, err
+		}
+		ids = ids[:0]
+		for _, marker := range markers {
+			ids = append(ids, path.Base(*marker.Key))
+		}
 	}
-	defer func() { _ = f.Close() }()
-	info, err := f.Stat()
-	if err != nil {
-		return store.DigestFile{}, err
+	refs := &chunkRefs{live: map[string]map[string]struct{}{}}
+	g, gctx := errgroup.WithContext(ctx)
+	g.SetLimit(metaReadConcurrency)
+	for _, recordID := range ids {
+		g.Go(func() error {
+			man, err := s.currentManifest(gctx, recordID)
+			if errors.Is(err, store.ErrNotFound) {
+				return nil
+			}
+			if err != nil {
+				return err
+			}
+			refs.add(man)
+			return nil
+		})
 	}
-	if err := store.RequireRegular(digestPath, info.Mode()); err != nil {
-		return store.DigestFile{}, err
+	if err := g.Wait(); err != nil {
+		return nil, err
 	}
-	reader := newDigestReader(f, digestPath, info.Size())
-	if err := s.uploadReader(ctx, key, reader, info.Size(), nil); err != nil {
-		return store.DigestFile{}, err
+	have := refs.live[lineage]
+	if have == nil {
+		have = map[string]struct{}{}
 	}
-	return reader.Digest()
+	return have, nil
 }
 
-func (s *Store) uploadReader(ctx context.Context, key string, body io.Reader, size int64, metadata map[string]string) error {
-	input := &transfermanager.UploadObjectInput{
-		Bucket:        &s.bucket,
-		Key:           &key,
-		Body:          body,
-		ContentLength: new(size),
-		Metadata:      metadata,
+func (s *Store) refreshManifest(ctx context.Context, key string) (manifest, error) {
+	body, _, err := s.getObject(ctx, key)
+	if errors.Is(err, store.ErrNotFound) {
+		return manifest{}, nil
 	}
-	if _, err := s.tm.UploadObject(ctx, input); err != nil {
-		return fmt.Errorf("upload %s: %w", key, err)
+	if err != nil {
+		return manifest{}, err
 	}
-	return nil
+	var man manifest
+	if err = json.Unmarshal(body, &man); err != nil {
+		return manifest{}, fmt.Errorf("decode manifest %s: %w", key, err)
+	}
+	return man, s.putObject(ctx, key, body, nil)
 }
 
-func (s *Store) download(ctx context.Context, key, path string) error {
-	if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
-		return err
-	}
-	f, err := os.Create(path) //nolint:gosec // path derives from our own temp dir
+func (s *Store) putObject(ctx context.Context, key string, body []byte, metadata map[string]string) error {
+	_, err := s.client.PutObject(ctx, &awss3.PutObjectInput{
+		Bucket:        &s.bucket,
+		Key:           &key,
+		Body:          bytes.NewReader(body),
+		ContentLength: new(int64(len(body))),
+		Metadata:      metadata,
+	})
 	if err != nil {
-		return err
-	}
-	defer func() { _ = f.Close() }()
-	// DownloadObject, not GetObject: the WriterAt form downloads parts in parallel.
-	if _, err := s.tm.DownloadObject(ctx, &transfermanager.DownloadObjectInput{Bucket: &s.bucket, Key: &key, WriterAt: f}); err != nil {
-		return fmt.Errorf("download %s: %w", key, err)
+		return fmt.Errorf("upload %s: %w", key, err)
 	}
 	return nil
 }
@@ -436,19 +564,111 @@ func (s *Store) deleteKeys(ctx context.Context, keys []string) error {
 	return nil
 }
 
-func (s *Store) list(ctx context.Context, prefix string) ([]string, error) {
-	var keys []string
+func (s *Store) list(ctx context.Context, prefix string) ([]s3types.Object, error) {
+	var objs []s3types.Object
 	p := awss3.NewListObjectsV2Paginator(s.client, &awss3.ListObjectsV2Input{Bucket: &s.bucket, Prefix: &prefix})
 	for p.HasMorePages() {
 		page, err := p.NextPage(ctx)
 		if err != nil {
 			return nil, fmt.Errorf("list %s: %w", prefix, err)
 		}
-		for _, obj := range page.Contents {
-			keys = append(keys, *obj.Key)
+		objs = append(objs, page.Contents...)
+	}
+	return objs, nil
+}
+
+type chunkUploader struct {
+	store *Store
+	group *errgroup.Group
+	ctx   context.Context // the group's, so a finished scan cannot cancel uploads in flight
+
+	lineage string
+
+	mu   sync.Mutex
+	have map[string]struct{}
+}
+
+func (u *chunkUploader) scan(ctx context.Context, file publishFile) (store.DigestFile, manifestFile, error) {
+	f, err := os.Open(file.path)
+	if err != nil {
+		return store.DigestFile{}, manifestFile{}, err
+	}
+	defer func() { _ = f.Close() }()
+	info, err := f.Stat()
+	if err != nil {
+		return store.DigestFile{}, manifestFile{}, err
+	}
+	if err = store.RequireRegular(file.rel, info.Mode()); err != nil {
+		return store.DigestFile{}, manifestFile{}, err
+	}
+	entry := manifestFile{Path: file.rel, Size: info.Size()}
+	digest, err := store.ScanFile(f, file.rel, info.Size(), getChunkBuf, func(c store.Chunk) error {
+		entry.Chunks = append(entry.Chunks, manifestChunk{Index: c.Index, SHA256: c.SHA256})
+		return u.upload(ctx, c)
+	})
+	return digest, entry, err
+}
+
+func (u *chunkUploader) upload(ctx context.Context, c store.Chunk) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	if !u.claim(c.SHA256) {
+		putChunkBuf(c.Data)
+		return nil
+	}
+	u.group.Go(func() error {
+		defer putChunkBuf(c.Data)
+		return u.store.putObject(u.ctx, u.store.chunkKey(u.lineage, c.SHA256), c.Data, nil)
+	})
+	return nil
+}
+
+func (u *chunkUploader) claim(sha string) bool {
+	u.mu.Lock()
+	defer u.mu.Unlock()
+	if _, ok := u.have[sha]; ok {
+		return false
+	}
+	u.have[sha] = struct{}{}
+	return true
+}
+
+type chunkRefs struct {
+	mu   sync.Mutex
+	live map[string]map[string]struct{} // lineage -> chunk sha
+}
+
+func (r *chunkRefs) add(man manifest) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	set := r.live[man.Lineage]
+	if set == nil {
+		set = map[string]struct{}{}
+		r.live[man.Lineage] = set
+	}
+	for _, file := range man.Files {
+		for _, chunk := range file.Chunks {
+			set[chunk.SHA256] = struct{}{}
 		}
 	}
-	return keys, nil
+}
+
+func publishFiles(staging string) ([]publishFile, error) {
+	var files []publishFile
+	err := store.WalkExport(filepath.Join(staging, store.ExportDir), func(rel, abs string, _ fs.FileInfo) error {
+		files = append(files, publishFile{path: abs, rel: rel})
+		return nil
+	})
+	return files, err
+}
+
+func objectKeys(objs []s3types.Object) []string {
+	keys := make([]string, len(objs))
+	for i, obj := range objs {
+		keys[i] = *obj.Key
+	}
+	return keys
 }
 
 // pruneIdleGenerations drops cached generations of one record that no Fetch used within the grace; a hit refreshes the mtime.
@@ -458,9 +678,16 @@ func pruneIdleGenerations(dir, keep string) {
 		return
 	}
 	for _, entry := range entries {
-		path := filepath.Join(dir, entry.Name())
-		if info, err := entry.Info(); err == nil && path != keep && time.Since(info.ModTime()) >= store.GenerationGrace {
-			_ = os.RemoveAll(path)
+		entryPath := filepath.Join(dir, entry.Name())
+		if info, err := entry.Info(); err == nil && entryPath != keep && time.Since(info.ModTime()) >= store.GenerationGrace {
+			_ = os.RemoveAll(entryPath)
 		}
 	}
 }
+
+func getChunkBuf() []byte { return *chunkBufs.Get().(*[]byte) }
+
+func putChunkBuf(buf []byte) {
+	buf = buf[:cap(buf)]
+	chunkBufs.Put(&buf)
+}
diff --git a/sandboxd/store/s3/s3_test.go b/sandboxd/store/s3/s3_test.go
index fe37acc7..7f7bb7af 100644
--- a/sandboxd/store/s3/s3_test.go
+++ b/sandboxd/store/s3/s3_test.go
@@ -5,6 +5,7 @@ import (
 	"cmp"
 	"context"
 	"crypto/sha256"
+	"encoding/hex"
 	"encoding/xml"
 	"errors"
 	"fmt"
@@ -18,11 +19,10 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"syscall"
 	"testing"
 	"time"
 
-	"github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager"
-
 	"github.com/cocoonstack/sandbox/sandboxd/store"
 	"github.com/cocoonstack/sandbox/sandboxd/store/storetest"
 )
@@ -36,8 +36,8 @@ func TestS3BackendContract(t *testing.T) {
 func TestDeleteRetryConverges(t *testing.T) {
 	const id = "ck_00000000000000bb"
 	fake := &fakeS3{objects: map[string][]byte{
-		"ck/" + id + "/" + store.MetaFile:                []byte(`{"id":"` + id + `"}`),
-		"ck/" + id + "/" + store.ExportDir + "/disk.img": []byte("bytes"),
+		"ck/" + id + "/" + store.MetaFile:                            []byte(`{"id":"` + id + `"}`),
+		"ck/manifests/" + id + "/" + store.ExportDir + "-first.json": []byte(`{"lineage":"` + id + `","files":[]}`),
 	}}
 	st := newTestStore(t, fake)
 	if err := st.Delete(t.Context(), id); err != nil {
@@ -46,17 +46,17 @@ func TestDeleteRetryConverges(t *testing.T) {
 	if err := st.Delete(t.Context(), id); err != nil {
 		t.Fatalf("delete retry after keys already gone: %v", err)
 	}
-	if len(fake.objects) != 0 {
-		t.Errorf("objects left behind: %v", fake.objects)
+	if keys := fake.keys(); len(keys) != 1 || !strings.HasPrefix(keys[0], "ck/manifests/") {
+		t.Errorf("objects after Delete = %v, want only the manifest", keys)
 	}
 }
 
-func TestDeleteRetainsMetaUntilExportsAreGone(t *testing.T) {
+func TestDeleteRetainsMetaUntilManifestsAreRefreshed(t *testing.T) {
 	const id = "ck_00000000000000bb"
 	metaKey := "ck/" + id + "/" + store.MetaFile
-	exportKeys := []string{
-		"ck/" + id + "/export-first/disk.img",
-		"ck/" + id + "/export-second/disk.img",
+	manifestKeys := []string{
+		"ck/manifests/" + id + "/export-first.json",
+		"ck/manifests/" + id + "/export-second.json",
 	}
 	for _, tc := range []struct {
 		name       string
@@ -69,9 +69,9 @@ func TestDeleteRetainsMetaUntilExportsAreGone(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			fake := &fakeS3{
 				objects: map[string][]byte{
-					metaKey:       []byte(`{"id":"` + id + `"}`),
-					exportKeys[0]: []byte("first"),
-					exportKeys[1]: []byte("second"),
+					metaKey:         []byte(`{"id":"` + id + `"}`),
+					manifestKeys[0]: []byte(`{"lineage":"` + id + `","files":[]}`),
+					manifestKeys[1]: []byte(`{"lineage":"` + id + `","files":[]}`),
 				},
 				failList:   tc.failList,
 				failDelete: tc.failDelete,
@@ -84,12 +84,12 @@ func TestDeleteRetainsMetaUntilExportsAreGone(t *testing.T) {
 			fake.mu.Lock()
 			if _, ok := fake.objects[metaKey]; !ok {
 				fake.mu.Unlock()
-				t.Fatal("first Delete removed meta before export cleanup succeeded")
+				t.Fatal("first Delete removed meta before the cleanup succeeded")
 			}
 			for _, batch := range fake.deleteBatches {
 				if slices.Contains(batch, metaKey) {
 					fake.mu.Unlock()
-					t.Fatalf("first Delete included meta in export batch %v", batch)
+					t.Fatalf("first Delete included meta in batch %v", batch)
 				}
 			}
 			fake.failList = false
@@ -102,11 +102,16 @@ func TestDeleteRetainsMetaUntilExportsAreGone(t *testing.T) {
 			}
 			fake.mu.Lock()
 			defer fake.mu.Unlock()
-			if len(fake.objects) != 0 {
-				t.Errorf("objects left after retry: %v", fake.objects)
+			if len(fake.objects) != len(manifestKeys) {
+				t.Errorf("objects after retry = %d, want only the manifests", len(fake.objects))
+			}
+			for _, key := range manifestKeys {
+				if time.Since(fake.modified[key]) > time.Minute {
+					t.Errorf("manifest %s was not refreshed by Delete", key)
+				}
 			}
 			if len(fake.deleteBatches) < 2 {
-				t.Fatalf("delete batches = %v, want exports then meta", fake.deleteBatches)
+				t.Fatalf("delete batches = %v, want labels then meta", fake.deleteBatches)
 			}
 			last := len(fake.deleteBatches) - 1
 			for i, batch := range fake.deleteBatches {
@@ -173,76 +178,22 @@ func TestFetchOutlivesTheCallerThatStartedIt(t *testing.T) {
 	}
 }
 
-func TestDigestReaderIgnoresReadBoundaries(t *testing.T) {
-	data := bytes.Repeat([]byte{'x'}, int(store.DigestChunkSize))
-	data = append(data, 'y')
-	source := &patternReader{data: data, limits: []int{1, 31, 64 << 10, 7, 79 << 10}}
-	reader := newDigestReader(source, "boundary.bin", int64(len(data)))
-	if _, ok := any(reader).(io.Seeker); ok {
-		t.Fatal("digest reader exposes io.Seeker")
-	}
-	if _, err := io.CopyBuffer(io.Discard, reader, make([]byte, 113<<10)); err != nil {
-		t.Fatalf("read digest stream: %v", err)
-	}
-	file, err := reader.Digest()
-	if err != nil {
-		t.Fatalf("finish digest stream: %v", err)
-	}
-	digest, err := store.AssembleDigest([]store.DigestFile{file})
-	if err != nil {
-		t.Fatalf("assemble digest: %v", err)
-	}
-	const want = "sha256:da8dd289e7d66408bd7d202642f180f2c22bc508cf0fe1d6e9f0e0b7967f6a5e"
-	if digest != want {
-		t.Errorf("digest = %q, want %q", digest, want)
-	}
-}
-
-func TestUploadDigestedSetsContentLength(t *testing.T) {
-	data := bytes.Repeat([]byte{'x'}, int(store.DigestChunkSize))
-	data = append(data, 'y')
-	path := filepath.Join(t.TempDir(), "boundary.bin")
-	if err := os.WriteFile(path, data, 0o600); err != nil {
-		t.Fatalf("write export: %v", err)
-	}
-	tm := &recordingTransferManager{}
-	st := &Store{tm: tm}
-	file, err := st.uploadDigested(t.Context(), "key", path, "boundary.bin")
-	if err != nil {
-		t.Fatalf("upload digested: %v", err)
-	}
-	if tm.contentLength == nil || *tm.contentLength != int64(len(data)) {
-		t.Fatalf("ContentLength = %v, want %d", tm.contentLength, len(data))
-	}
-	if tm.read != int64(len(data)) {
-		t.Fatalf("uploaded bytes = %d, want %d", tm.read, len(data))
-	}
-	digest, err := store.AssembleDigest([]store.DigestFile{file})
-	if err != nil {
-		t.Fatalf("assemble digest: %v", err)
-	}
-	const want = "sha256:da8dd289e7d66408bd7d202642f180f2c22bc508cf0fe1d6e9f0e0b7967f6a5e"
-	if digest != want {
-		t.Errorf("digest = %q, want %q", digest, want)
-	}
-}
-
 func TestPublishDigestedRetryAndMetaRequestAccounting(t *testing.T) {
 	const id = "ck_00000000000000dd"
 	fake := &fakeS3{objects: map[string][]byte{}}
 	st := newTestStore(t, fake)
 	meta := []byte(`{"id":"` + id + `","gen":1}`)
 	content := "retry-body"
-	exportKey := "ck/" + id + "/" + store.ExportGen(meta) + "/disk.img"
+	chunk := sha256.Sum256([]byte(content))
+	exportKey := "ck/chunks/" + id + "/" + hex.EncodeToString(chunk[:])
 	metaKey := "ck/" + id + "/" + store.MetaFile
 	fake.setPutFailures(exportKey, 1)
 
 	staging := stageRecord(t, st, id, meta, content)
-	digest, err := st.PublishDigested(t.Context(), staging, id)
+	digest, err := st.PublishDigested(t.Context(), staging, id, "")
 	if err != nil {
 		t.Fatalf("PublishDigested: %v", err)
 	}
-	chunk := sha256.Sum256([]byte(content))
 	want, err := store.AssembleDigest([]store.DigestFile{{
 		Path: "disk.img", Size: int64(len(content)), Chunks: [][sha256.Size]byte{chunk},
 	}})
@@ -301,18 +252,19 @@ func TestPublishDigestedFailurePreservesCommittedGeneration(t *testing.T) {
 	st := newTestStore(t, fake)
 	oldMeta := []byte(`{"id":"` + id + `","gen":1}`)
 	oldStaging := stageRecord(t, st, id, oldMeta, "old")
-	oldDigest, err := st.PublishDigested(t.Context(), oldStaging, id)
+	oldDigest, err := st.PublishDigested(t.Context(), oldStaging, id, "")
 	if err != nil {
 		t.Fatalf("publish old generation: %v", err)
 	}
 
 	newMeta := []byte(`{"id":"` + id + `","gen":2}`)
 	newStaging := stageRecord(t, st, id, newMeta, "replacement")
-	newExportKey := "ck/" + id + "/" + store.ExportGen(newMeta) + "/disk.img"
+	replacement := sha256.Sum256([]byte("replacement"))
+	newExportKey := "ck/chunks/" + id + "/" + hex.EncodeToString(replacement[:])
 	metaKey := "ck/" + id + "/" + store.MetaFile
 	markerPuts := fake.requestCount(http.MethodPut, metaKey)
 	fake.setPutFailures(newExportKey, 100)
-	digest, err := st.PublishDigested(t.Context(), newStaging, id)
+	digest, err := st.PublishDigested(t.Context(), newStaging, id, "")
 	if err == nil {
 		t.Fatal("PublishDigested succeeded with injected export failure")
 	}
@@ -336,7 +288,7 @@ func TestPublishDigestedFailurePreservesCommittedGeneration(t *testing.T) {
 	}
 
 	fake.setPutFailures(newExportKey, 0)
-	newDigest, err := st.PublishDigested(t.Context(), newStaging, id)
+	newDigest, err := st.PublishDigested(t.Context(), newStaging, id, "")
 	if err != nil {
 		t.Fatalf("retry PublishDigested: %v", err)
 	}
@@ -365,22 +317,20 @@ func TestRepublishRetainsGenerationSelectedByAnotherStore(t *testing.T) {
 	}
 	publishRecord(t, writer, id, []byte(`{"id":"`+id+`","gen":2}`), "second")
 
+	if err = writer.Delete(t.Context(), id); err != nil {
+		t.Fatalf("delete the record: %v", err)
+	}
+	if _, ok := fake.get("ck/" + id + "/" + store.MetaFile); ok {
+		t.Error("meta survived Delete")
+	}
 	gen := filepath.Join(reader.staging, "selected-first")
 	if err = reader.populate(t.Context(), id, selected, gen); err != nil {
-		t.Fatalf("fetch selected first generation after re-publish: %v", err)
+		t.Fatalf("fetch of the selected generation after the record was deleted: %v", err)
 	}
 	got, err := os.ReadFile(filepath.Join(gen, store.ExportDir, "disk.img"))
 	if err != nil || string(got) != "first" {
 		t.Fatalf("selected generation bytes: %q, %v, want first", got, err)
 	}
-	if err = writer.Delete(t.Context(), id); err != nil {
-		t.Fatalf("delete all generations: %v", err)
-	}
-	fake.mu.Lock()
-	defer fake.mu.Unlock()
-	if len(fake.objects) != 0 {
-		t.Errorf("objects left after Delete: %v", fake.objects)
-	}
 }
 
 func TestFetchPrunesSupersededGenerationsIdlePastTheGrace(t *testing.T) {
@@ -452,10 +402,326 @@ func TestS3BackendContractRealEndpoint(t *testing.T) {
 	storetest.RunContract(t, st)
 }
 
+func TestSparseExportStaysSparseThroughS3(t *testing.T) {
+	const id = "ck_00000000000000a1"
+	fake := &fakeS3{objects: map[string][]byte{}}
+	st := newTestStore(t, fake)
+	size := 12*store.DigestChunkSize + 4096
+	meta := []byte(`{"id":"` + id + `"}`)
+	staging, src := stageSparseRecord(t, st, id, meta, size, map[int64]string{0: "head", 3*store.DigestChunkSize + 100: "mid", 11*store.DigestChunkSize + 10: "tail"})
+	srcSum := fileSum(t, src)
+	if err := st.Publish(t.Context(), staging, id, ""); err != nil {
+		t.Fatalf("Publish: %v", err)
+	}
+	count, uploaded := fake.chunkPuts("ck/chunks/" + id + "/")
+	if want := 3 * store.DigestChunkSize; count != 3 || uploaded != want {
+		t.Fatalf("chunk PUTs = %d carrying %d bytes, want 3 carrying %d", count, uploaded, want)
+	}
+
+	dir, _, _, err := st.Fetch(t.Context(), id)
+	if err != nil {
+		t.Fatalf("Fetch: %v", err)
+	}
+	got := filepath.Join(dir, "disk.img")
+	info, err := os.Stat(got)
+	if err != nil || info.Size() != size {
+		t.Fatalf("fetched size = %v, %v, want %d", info, err, size)
+	}
+	if fileSum(t, got) != srcSum {
+		t.Errorf("fetched content differs from the source")
+	}
+	if blocks := allocatedBytes(t, got); blocks > uploaded+(1<<20) {
+		t.Errorf("fetched file allocates %d bytes, want at most the %d uploaded", blocks, uploaded)
+	}
+}
+
+func TestAllZeroDataChunkIsNotUploaded(t *testing.T) {
+	const id = "ck_00000000000000a2"
+	fake := &fakeS3{objects: map[string][]byte{}}
+	st := newTestStore(t, fake)
+	meta := []byte(`{"id":"` + id + `"}`)
+	staging := stageRecord(t, st, id, meta, "")
+	src := filepath.Join(staging, store.ExportDir, "disk.img")
+	dense := make([]byte, 2*store.DigestChunkSize+10)
+	for i := range int(store.DigestChunkSize) {
+		dense[int(store.DigestChunkSize)+i] = 'x'
+	}
+	if err := os.WriteFile(src, dense, 0o600); err != nil {
+		t.Fatalf("write dense export: %v", err)
+	}
+	if err := st.Publish(t.Context(), staging, id, ""); err != nil {
+		t.Fatalf("Publish: %v", err)
+	}
+	if count, _ := fake.chunkPuts("ck/chunks/" + id + "/"); count != 1 {
+		t.Errorf("chunk PUTs = %d, want only the non-zero chunk", count)
+	}
+	dir, _, _, err := st.Fetch(t.Context(), id)
+	if err != nil {
+		t.Fatalf("Fetch: %v", err)
+	}
+	if got, err := os.ReadFile(filepath.Join(dir, "disk.img")); err != nil || !bytes.Equal(got, dense) {
+		t.Errorf("fetched content differs from the source: %v", err)
+	}
+}
+
+func TestFetchFailsOnDamagedChunks(t *testing.T) {
+	const id = "ck_00000000000000a3"
+	meta := []byte(`{"id":"` + id + `"}`)
+	sum := sha256.Sum256([]byte("payload"))
+	chunkKey := "ck/chunks/" + id + "/" + hex.EncodeToString(sum[:])
+	for _, tc := range []struct {
+		name   string
+		damage func(*fakeS3)
+		want   string
+	}{
+		{"flipped byte", func(f *fakeS3) { f.objects[chunkKey][0] ^= 0xff }, "checksum"},
+		{"missing chunk", func(f *fakeS3) { delete(f.objects, chunkKey) }, "is missing"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			fake := &fakeS3{objects: map[string][]byte{}}
+			st := newTestStore(t, fake)
+			publishRecord(t, st, id, meta, "payload")
+			tc.damage(fake)
+			_, _, _, err := st.Fetch(t.Context(), id)
+			if err == nil || !strings.Contains(err.Error(), tc.want) {
+				t.Fatalf("Fetch = %v, want an error containing %q", err, tc.want)
+			}
+			if errors.Is(err, store.ErrNotFound) {
+				t.Errorf("a damaged record must not read as absent: %v", err)
+			}
+		})
+	}
+}
+
+func TestPublishDigestedHashesHolesLikeZeros(t *testing.T) {
+	const id = "ck_00000000000000a4"
+	st := newTestStore(t, &fakeS3{objects: map[string][]byte{}})
+	size := 3*store.DigestChunkSize + 7
+	staging, src := stageSparseRecord(t, st, id, []byte(`{"id":"`+id+`"}`), size, map[int64]string{store.DigestChunkSize + 5: "data", size - 1: "z"})
+	sums := denseChunkSums(t, src)
+	digest, err := st.PublishDigested(t.Context(), staging, id, "")
+	if err != nil {
+		t.Fatalf("PublishDigested: %v", err)
+	}
+	want, err := store.AssembleDigest([]store.DigestFile{{Path: "disk.img", Size: size, Chunks: sums}})
+	if err != nil {
+		t.Fatalf("assemble reference digest: %v", err)
+	}
+	if digest != want {
+		t.Errorf("digest = %q, want the dense-read digest %q", digest, want)
+	}
+}
+
+func TestPublishDigestedSplitsChunksAtTheFixedBoundary(t *testing.T) {
+	const id = "ck_00000000000000a5"
+	st := newTestStore(t, &fakeS3{objects: map[string][]byte{}})
+	content := string(bytes.Repeat([]byte{'x'}, int(store.DigestChunkSize))) + "y"
+	staging := stageRecord(t, st, id, []byte(`{"id":"`+id+`"}`), content)
+	digest, err := st.PublishDigested(t.Context(), staging, id, "")
+	if err != nil {
+		t.Fatalf("PublishDigested: %v", err)
+	}
+	want, err := store.AssembleDigest([]store.DigestFile{{
+		Path: "disk.img", Size: store.DigestChunkSize + 1,
+		Chunks: [][sha256.Size]byte{sha256.Sum256([]byte(content[:store.DigestChunkSize])), sha256.Sum256([]byte("y"))},
+	}})
+	if err != nil {
+		t.Fatalf("assemble reference digest: %v", err)
+	}
+	if digest != want {
+		t.Errorf("digest = %q, want %q", digest, want)
+	}
+}
+
+func TestLineageDedupUploadsOnlyChangedChunks(t *testing.T) {
+	const lineage = "sbx_00000000000000a1"
+	fake := &fakeS3{objects: map[string][]byte{}}
+	st := newTestStore(t, fake)
+	cs := store.DigestChunkSize
+	base := map[int64]string{0: "c0", 2 * cs: "c2", 4 * cs: "c4", 6 * cs: "c6"}
+	changed := maps.Clone(base)
+	changed[4*cs] = "changed"
+	publish := func(id string, gen int, writes map[int64]string, scope string) (puts int, srcSum [sha256.Size]byte) {
+		t.Helper()
+		before, _ := fake.chunkPuts("ck/chunks/")
+		staging, src := stageSparseRecord(t, st, id, fmt.Appendf(nil, `{"id":"%s","gen":%d}`, id, gen), 8*cs, writes)
+		srcSum = fileSum(t, src)
+		if err := st.Publish(t.Context(), staging, id, scope); err != nil {
+			t.Fatalf("Publish %s: %v", id, err)
+		}
+		after, _ := fake.chunkPuts("ck/chunks/")
+		return after - before, srcSum
+	}
+
+	if puts, _ := publish("ck_00000000000000b1", 1, base, lineage); puts != 4 {
+		t.Fatalf("first publish uploaded %d chunks, want 4", puts)
+	}
+	puts, srcSum := publish("ck_00000000000000b2", 1, changed, lineage)
+	if puts != 1 {
+		t.Errorf("second record of the lineage uploaded %d chunks, want only the changed one", puts)
+	}
+	if puts, _ = publish("ck_00000000000000b1", 2, base, lineage); puts != 0 {
+		t.Errorf("re-publish of identical content uploaded %d chunks, want 0", puts)
+	}
+	if puts, _ = publish("ck_00000000000000b3", 1, base, ""); puts != 4 {
+		t.Errorf("a record outside the lineage uploaded %d chunks, want all 4", puts)
+	}
+
+	dir, _, _, err := st.Fetch(t.Context(), "ck_00000000000000b2")
+	if err != nil {
+		t.Fatalf("Fetch: %v", err)
+	}
+	if fileSum(t, filepath.Join(dir, "disk.img")) != srcSum {
+		t.Errorf("fetched content differs from the source")
+	}
+}
+
+func TestPublishReadsOnlyItsLineage(t *testing.T) {
+	const lineage = "sbx_00000000000000a4"
+	fake := &fakeS3{objects: map[string][]byte{}}
+	st := newTestStore(t, fake)
+	for _, id := range []string{"ck_00000000000000d1", "ck_00000000000000d2"} {
+		publishRecord(t, st, id, []byte(`{"id":"`+id+`"}`), "unrelated")
+	}
+	publishIn := func(id string) {
+		t.Helper()
+		staging := stageRecord(t, st, id, []byte(`{"id":"`+id+`"}`), "claim")
+		if err := st.Publish(t.Context(), staging, id, lineage); err != nil {
+			t.Fatalf("Publish %s: %v", id, err)
+		}
+	}
+	publishIn("ck_00000000000000e1")
+	publishIn("ck_00000000000000e2")
+	fake.resetRequests()
+	publishIn("ck_00000000000000e3")
+
+	meta := func(id string) int { return fake.requestCount(http.MethodGet, "ck/"+id+"/"+store.MetaFile) }
+	if got := meta("ck_00000000000000d1") + meta("ck_00000000000000d2"); got != 0 {
+		t.Errorf("a publish read %d metas of records outside its lineage", got)
+	}
+	if got := meta("ck_00000000000000e1") + meta("ck_00000000000000e2"); got != 2 {
+		t.Errorf("a publish read %d metas of its lineage's records, want 2", got)
+	}
+}
+
+func TestPublishUploadsAnUnreferencedChunkAgain(t *testing.T) {
+	const lineage = "sbx_00000000000000a3"
+	fake := &fakeS3{objects: map[string][]byte{}}
+	st := newTestStore(t, fake)
+	publishIn := func(id, content, scope string) int {
+		t.Helper()
+		before, _ := fake.chunkPuts("ck/chunks/")
+		staging := stageRecord(t, st, id, []byte(`{"id":"`+id+`"}`), content)
+		if err := st.Publish(t.Context(), staging, id, scope); err != nil {
+			t.Fatalf("Publish %s: %v", id, err)
+		}
+		after, _ := fake.chunkPuts("ck/chunks/")
+		return after - before
+	}
+	publish := func(id string) int { return publishIn(id, "orphaned", lineage) }
+
+	if puts := publish("ck_00000000000000c1"); puts != 1 {
+		t.Fatalf("first publish uploaded %d chunks, want 1", puts)
+	}
+	if err := st.Delete(t.Context(), "ck_00000000000000c1"); err != nil {
+		t.Fatalf("Delete: %v", err)
+	}
+	if puts := publish("ck_00000000000000c2"); puts != 1 {
+		t.Errorf("a listed chunk no manifest references was reused: %d uploads, want 1", puts)
+	}
+	if puts := publish("ck_00000000000000c3"); puts != 0 {
+		t.Errorf("a chunk a committed manifest references was uploaded again: %d uploads, want 0", puts)
+	}
+	publishIn("ck_00000000000000c4", "elsewhere", "")
+	if puts := publishIn("ck_00000000000000c5", "elsewhere", lineage); puts != 1 {
+		t.Errorf("a chunk another lineage's manifest references was reused: %d uploads, want 1", puts)
+	}
+}
+
+func TestDeletedRecordLeavesNoPrefixToList(t *testing.T) {
+	const (
+		gone = "ck_00000000000000b1"
+		live = "ck_00000000000000b2"
+	)
+	fake := &fakeS3{objects: map[string][]byte{}}
+	st := newTestStore(t, fake)
+	publishRecord(t, st, gone, []byte(`{"id":"`+gone+`"}`), "gone-payload")
+	publishRecord(t, st, live, []byte(`{"id":"`+live+`"}`), "live-payload")
+	if err := st.Delete(t.Context(), gone); err != nil {
+		t.Fatalf("Delete: %v", err)
+	}
+	for _, key := range fake.keys() {
+		if strings.HasPrefix(key, "ck/"+gone+"/") {
+			t.Errorf("Delete left %s under the record prefix", key)
+		}
+	}
+
+	fake.resetRequests()
+	recs, err := st.Metas(t.Context())
+	if err != nil || len(recs) != 1 {
+		t.Fatalf("Metas = %d records, %v, want only the live one", len(recs), err)
+	}
+	if got := fake.requestCount(http.MethodGet, "ck/"+gone+"/"+store.MetaFile); got != 0 {
+		t.Errorf("Metas read the meta of the deleted record %d times, want 0", got)
+	}
+
+	sum := sha256.Sum256([]byte("gone-payload"))
+	if _, ok := fake.get("ck/chunks/" + gone + "/" + hex.EncodeToString(sum[:])); !ok {
+		t.Error("Delete freed the chunk of the deleted record")
+	}
+}
+
+func TestSweepGenerationsLeavesTheBucketAlone(t *testing.T) {
+	const id = "ck_00000000000000a8"
+	fake := &fakeS3{objects: map[string][]byte{}}
+	st := newTestStore(t, fake)
+	publishRecord(t, st, id, []byte(`{"id":"`+id+`","gen":1}`), "first")
+	publishRecord(t, st, id, []byte(`{"id":"`+id+`","gen":2}`), "second")
+	if err := st.Delete(t.Context(), id); err != nil {
+		t.Fatalf("Delete: %v", err)
+	}
+	fake.age("", 2*store.GenerationGrace)
+	want := fake.keys()
+	if err := st.SweepGenerations(t.Context()); err != nil {
+		t.Fatalf("SweepGenerations: %v", err)
+	}
+	if keys := fake.keys(); !slices.Equal(keys, want) {
+		t.Errorf("objects after SweepGenerations = %v, want %v", keys, want)
+	}
+}
+
+func TestDeleteRemovesTheLineageMarker(t *testing.T) {
+	const lineage = "sbx_00000000000000a5"
+	fake := &fakeS3{objects: map[string][]byte{}}
+	st := newTestStore(t, fake)
+	for _, id := range []string{"ck_00000000000000f1", "ck_00000000000000f2"} {
+		staging := stageRecord(t, st, id, []byte(`{"id":"`+id+`"}`), "claim")
+		if err := st.Publish(t.Context(), staging, id, lineage); err != nil {
+			t.Fatalf("Publish %s: %v", id, err)
+		}
+	}
+	if err := st.Delete(t.Context(), "ck_00000000000000f1"); err != nil {
+		t.Fatalf("Delete: %v", err)
+	}
+	if _, ok := fake.get("ck/lineages/" + lineage + "/ck_00000000000000f1"); ok {
+		t.Error("Delete left the lineage marker of the deleted record")
+	}
+	if _, ok := fake.get("ck/lineages/" + lineage + "/ck_00000000000000f2"); !ok {
+		t.Error("Delete removed the marker of another record")
+	}
+}
+
+func TestTransferBudgetStaysBelowTheReuseWindow(t *testing.T) {
+	budget := transferBudget
+	if budget >= store.GenerationGrace {
+		t.Errorf("transfer budget %v must stay below the generation grace %v", budget, store.GenerationGrace)
+	}
+}
+
 func publishRecord(t *testing.T, st *Store, id string, meta []byte, content string) {
 	t.Helper()
 	staging := stageRecord(t, st, id, meta, content)
-	if err := st.Publish(t.Context(), staging, id); err != nil {
+	if err := st.Publish(t.Context(), staging, id, ""); err != nil {
 		t.Fatalf("publish record: %v", err)
 	}
 }
@@ -479,6 +745,69 @@ func stageRecord(t *testing.T, st *Store, id string, meta []byte, content string
 	return staging
 }
 
+func stageSparseRecord(t *testing.T, st *Store, id string, meta []byte, size int64, writes map[int64]string) (staging, path string) {
+	t.Helper()
+	staging = stageRecord(t, st, id, meta, "")
+	path = filepath.Join(staging, store.ExportDir, "disk.img")
+	f, err := os.OpenFile(path, os.O_WRONLY, 0)
+	if err != nil {
+		t.Fatalf("open export: %v", err)
+	}
+	defer func() { _ = f.Close() }()
+	if err = f.Truncate(size); err != nil {
+		t.Fatalf("truncate export: %v", err)
+	}
+	for off, data := range writes {
+		if _, err = f.WriteAt([]byte(data), off); err != nil {
+			t.Fatalf("write export at %d: %v", off, err)
+		}
+	}
+	return staging, path
+}
+
+func fileSum(t *testing.T, path string) [sha256.Size]byte {
+	t.Helper()
+	f, err := os.Open(path)
+	if err != nil {
+		t.Fatalf("open %s: %v", path, err)
+	}
+	defer func() { _ = f.Close() }()
+	h := sha256.New()
+	if _, err = io.Copy(h, f); err != nil {
+		t.Fatalf("hash %s: %v", path, err)
+	}
+	return [sha256.Size]byte(h.Sum(nil))
+}
+
+func allocatedBytes(t *testing.T, path string) int64 {
+	t.Helper()
+	info, err := os.Stat(path)
+	if err != nil {
+		t.Fatalf("stat %s: %v", path, err)
+	}
+	return info.Sys().(*syscall.Stat_t).Blocks * 512
+}
+
+func denseChunkSums(t *testing.T, path string) [][sha256.Size]byte {
+	t.Helper()
+	f, err := os.Open(path)
+	if err != nil {
+		t.Fatalf("open %s: %v", path, err)
+	}
+	defer func() { _ = f.Close() }()
+	var sums [][sha256.Size]byte
+	buf := make([]byte, store.DigestChunkSize)
+	for {
+		n, err := io.ReadFull(f, buf)
+		if n > 0 {
+			sums = append(sums, sha256.Sum256(buf[:n]))
+		}
+		if err != nil {
+			return sums
+		}
+	}
+}
+
 func newTestStore(t *testing.T, fake *fakeS3) *Store {
 	t.Helper()
 	ts := httptest.NewServer(fake)
@@ -508,51 +837,6 @@ func assertSingleMetaGet(t *testing.T, fake *fakeS3, metaKey string) {
 	}
 }
 
-type patternReader struct {
-	data   []byte
-	limits []int
-	offset int
-	read   int
-}
-
-func (r *patternReader) Read(p []byte) (int, error) {
-	if r.offset == len(r.data) {
-		return 0, io.EOF
-	}
-	limit := r.limits[r.read%len(r.limits)]
-	r.read++
-	n := min(len(p), limit, len(r.data)-r.offset)
-	copy(p, r.data[r.offset:r.offset+n])
-	r.offset += n
-	return n, nil
-}
-
-type recordingTransferManager struct {
-	contentLength *int64
-	read          int64
-}
-
-func (m *recordingTransferManager) UploadObject(
-	_ context.Context,
-	input *transfermanager.UploadObjectInput,
-	_ ...func(*transfermanager.Options),
-) (*transfermanager.UploadObjectOutput, error) {
-	if input.ContentLength != nil {
-		m.contentLength = new(*input.ContentLength)
-	}
-	read, err := io.CopyBuffer(io.Discard, input.Body, make([]byte, 113<<10))
-	m.read = read
-	return &transfermanager.UploadObjectOutput{}, err
-}
-
-func (*recordingTransferManager) DownloadObject(
-	context.Context,
-	*transfermanager.DownloadObjectInput,
-	...func(*transfermanager.Options),
-) (*transfermanager.DownloadObjectOutput, error) {
-	return nil, fmt.Errorf("unexpected download")
-}
-
 type fakeRequest struct {
 	method string
 	key    string
@@ -565,6 +849,7 @@ type fakeS3 struct {
 	failPuts      map[string]int
 	requests      map[fakeRequest]int
 	putLengths    map[string][]int64
+	modified      map[string]time.Time
 	deleteBatches [][]string
 	failList      bool
 	failDelete    bool
@@ -633,6 +918,43 @@ func (f *fakeS3) objectMetadata(key string) map[string]string {
 	return maps.Clone(f.metadata[key])
 }
 
+func (f *fakeS3) chunkPuts(prefix string) (count int, bytes int64) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for key, lengths := range f.putLengths {
+		if strings.HasPrefix(key, prefix) {
+			count += len(lengths)
+			for _, n := range lengths {
+				bytes += n
+			}
+		}
+	}
+	return count, bytes
+}
+
+func (f *fakeS3) get(key string) ([]byte, bool) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	body, ok := f.objects[key]
+	return body, ok
+}
+
+func (f *fakeS3) keys() []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return slices.Sorted(maps.Keys(f.objects))
+}
+
+func (f *fakeS3) age(prefix string, by time.Duration) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for key := range f.objects {
+		if strings.HasPrefix(key, prefix) {
+			f.modified[key] = time.Now().Add(-by)
+		}
+	}
+}
+
 func (f *fakeS3) resetRequests() {
 	f.mu.Lock()
 	defer f.mu.Unlock()
@@ -661,6 +983,10 @@ func (f *fakeS3) handlePutObject(w http.ResponseWriter, r *http.Request, key str
 		f.metadata = map[string]map[string]string{}
 	}
 	f.objects[key] = bytes.Clone(body)
+	if f.modified == nil {
+		f.modified = map[string]time.Time{}
+	}
+	f.modified[key] = time.Now()
 	metadata := requestMetadata(r.Header)
 	if len(metadata) == 0 {
 		delete(f.metadata, key)
@@ -678,8 +1004,9 @@ func (f *fakeS3) handleListObjects(w http.ResponseWriter, r *http.Request) {
 	prefix := r.URL.Query().Get("prefix")
 	delim := r.URL.Query().Get("delimiter")
 	type object struct {
-		Key  string `xml:"Key"`
-		Size int    `xml:"Size"`
+		Key          string `xml:"Key"`
+		Size         int    `xml:"Size"`
+		LastModified string `xml:"LastModified"`
 	}
 	type commonPrefix struct {
 		Prefix string `xml:"Prefix"`
@@ -705,7 +1032,9 @@ func (f *fakeS3) handleListObjects(w http.ResponseWriter, r *http.Request) {
 				continue
 			}
 		}
-		result.Contents = append(result.Contents, object{Key: key, Size: len(body)})
+		result.Contents = append(result.Contents, object{
+			Key: key, Size: len(body), LastModified: f.modified[key].UTC().Format("2006-01-02T15:04:05.000Z"),
+		})
 	}
 	slices.SortFunc(result.Contents, func(a, b object) int { return cmp.Compare(a.Key, b.Key) })
 	slices.SortFunc(result.CommonPrefixes, func(a, b commonPrefix) int { return cmp.Compare(a.Prefix, b.Prefix) })
@@ -731,6 +1060,9 @@ func (f *fakeS3) handleGetObject(w http.ResponseWriter, r *http.Request, key str
 		return
 	}
 	writeMetadata(w.Header(), f.metadata[key])
+	if mod := f.modified[key]; !mod.IsZero() {
+		w.Header().Set("Last-Modified", mod.UTC().Format(http.TimeFormat))
+	}
 	if rng := r.Header.Get("Range"); rng != "" {
 		var start, end int
 		if _, err := fmt.Sscanf(rng, "bytes=%d-%d", &start, &end); err == nil && start < len(body) {
diff --git a/sandboxd/store/sparse.go b/sandboxd/store/sparse.go
new file mode 100644
index 00000000..0f632653
--- /dev/null
+++ b/sandboxd/store/sparse.go
@@ -0,0 +1,129 @@
+package store
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+
+	"golang.org/x/sys/unix"
+)
+
+const zeroBlockSize = 1 << 20
+
+var zeroChunkSum = sync.OnceValue(func() [sha256.Size]byte { return zeroSum(DigestChunkSize) })
+
+// Chunk is one non-zero DigestChunkSize slice of a file; Data is owned by the emit callee.
+type Chunk struct {
+	Index  int64
+	SHA256 string // hex of the chunk bytes
+	Data   []byte
+}
+
+// ScanFile emits every non-zero chunk of f, reading only its data segments, and returns the file's digest entry; newBuf must return a DigestChunkSize buffer.
+func ScanFile(f *os.File, rel string, size int64, newBuf func() []byte, emit func(Chunk) error) (DigestFile, error) {
+	digest := DigestFile{Path: rel, Size: size, Chunks: make([][sha256.Size]byte, ChunkCount(size))}
+	visited := make([]bool, len(digest.Chunks))
+	var buf []byte
+	var next int64
+	for pos := int64(0); pos < size; {
+		start, end, err := dataExtent(f, pos)
+		if err != nil {
+			return DigestFile{}, fmt.Errorf("scan %s: %w", rel, err)
+		}
+		if end <= start {
+			break
+		}
+		for index := max(next, start/DigestChunkSize); index*DigestChunkSize < min(end, size); index++ {
+			if buf == nil {
+				buf = newBuf()
+			}
+			data := buf[:chunkLen(size, index)]
+			if _, err = f.ReadAt(data, index*DigestChunkSize); err != nil {
+				return DigestFile{}, fmt.Errorf("read %s chunk %d: %w", rel, index, err)
+			}
+			next = index + 1
+			visited[index] = true
+			if isZero(data) {
+				digest.Chunks[index] = chunkZeroSum(int64(len(data)))
+				continue
+			}
+			sum := sha256.Sum256(data)
+			digest.Chunks[index] = sum
+			if err = emit(Chunk{Index: index, SHA256: hex.EncodeToString(sum[:]), Data: data}); err != nil {
+				return DigestFile{}, err
+			}
+			buf = nil
+		}
+		pos = end
+	}
+	for i, seen := range visited {
+		if !seen {
+			digest.Chunks[i] = chunkZeroSum(chunkLen(size, int64(i)))
+		}
+	}
+	return digest, nil
+}
+
+// WriteChunk streams one chunk from r into f at its offset and fails unless it is chunkLen bytes hashing to wantSHA256.
+func WriteChunk(f *os.File, size, index int64, wantSHA256 string, r io.Reader) error {
+	if index < 0 || index >= ChunkCount(size) {
+		return fmt.Errorf("chunk %d outside a %d-byte file", index, size)
+	}
+	want := chunkLen(size, index)
+	h := sha256.New()
+	n, err := io.Copy(io.MultiWriter(h, io.NewOffsetWriter(f, index*DigestChunkSize)), io.LimitReader(r, want+1))
+	if err != nil {
+		return fmt.Errorf("write chunk %d: %w", index, err)
+	}
+	if n != want {
+		return fmt.Errorf("chunk %d is %d bytes, want %d", index, n, want)
+	}
+	if got := hex.EncodeToString(h.Sum(nil)); got != wantSHA256 {
+		return fmt.Errorf("chunk %d checksum %s, want %s", index, got, wantSHA256)
+	}
+	return nil
+}
+
+func dataExtent(f *os.File, pos int64) (start, end int64, err error) {
+	start, err = f.Seek(pos, unix.SEEK_DATA)
+	if errors.Is(err, unix.ENXIO) {
+		return 0, 0, nil
+	}
+	if err != nil {
+		return 0, 0, err
+	}
+	end, err = f.Seek(start, unix.SEEK_HOLE)
+	return start, end, err
+}
+
+func chunkLen(size, index int64) int64 {
+	return min(DigestChunkSize, size-index*DigestChunkSize)
+}
+
+func chunkZeroSum(n int64) [sha256.Size]byte {
+	if n == DigestChunkSize {
+		return zeroChunkSum()
+	}
+	return zeroSum(n)
+}
+
+func zeroSum(n int64) [sha256.Size]byte {
+	h := sha256.New()
+	block := make([]byte, min(n, zeroBlockSize))
+	for ; n > 0; n -= int64(len(block)) {
+		_, _ = h.Write(block[:min(n, int64(len(block)))])
+	}
+	var sum [sha256.Size]byte
+	h.Sum(sum[:0])
+	return sum
+}
+
+// isZero compares data with itself shifted by one byte, which holds only for a constant run.
+func isZero(data []byte) bool {
+	return data[0] == 0 && bytes.Equal(data[1:], data[:len(data)-1])
+}
diff --git a/sandboxd/store/sparse_test.go b/sandboxd/store/sparse_test.go
new file mode 100644
index 00000000..2a56e1af
--- /dev/null
+++ b/sandboxd/store/sparse_test.go
@@ -0,0 +1,120 @@
+package store
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"io"
+	"os"
+	"path/filepath"
+	"slices"
+	"strings"
+	"testing"
+)
+
+func TestScanFileEmitsEachNonZeroChunkOnce(t *testing.T) {
+	size := 4*DigestChunkSize + 5
+	path := filepath.Join(t.TempDir(), "disk.img")
+	f, err := os.Create(path)
+	if err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	defer func() { _ = f.Close() }()
+	if err = f.Truncate(size); err != nil {
+		t.Fatalf("truncate: %v", err)
+	}
+	for off, data := range map[int64]string{DigestChunkSize - 2: "abcd", DigestChunkSize + 10: "zz", DigestChunkSize + 1<<20: "yy", 4*DigestChunkSize + 1: "t"} {
+		if _, err = f.WriteAt([]byte(data), off); err != nil {
+			t.Fatalf("write at %d: %v", off, err)
+		}
+	}
+
+	var emitted []int64
+	digest, err := ScanFile(f, "disk.img", size, func() []byte { return make([]byte, DigestChunkSize) }, func(c Chunk) error {
+		emitted = append(emitted, c.Index)
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("ScanFile: %v", err)
+	}
+	if want := []int64{0, 1, 4}; !slices.Equal(emitted, want) {
+		t.Errorf("emitted chunks %v, want %v", emitted, want)
+	}
+	want := make([][sha256.Size]byte, 0, 5)
+	buf := make([]byte, DigestChunkSize)
+	for off := int64(0); off < size; off += DigestChunkSize {
+		n, readErr := f.ReadAt(buf, off)
+		if readErr != nil && !errors.Is(readErr, io.EOF) {
+			t.Fatalf("read at %d: %v", off, readErr)
+		}
+		want = append(want, sha256.Sum256(buf[:n]))
+	}
+	if !slices.Equal(digest.Chunks, want) {
+		t.Errorf("chunk sums differ from a dense read")
+	}
+}
+
+func TestScanFileDoesNotReadHoles(t *testing.T) {
+	size := 6 * DigestChunkSize
+	f, err := os.Create(filepath.Join(t.TempDir(), "holes.img"))
+	if err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	defer func() { _ = f.Close() }()
+	if err = f.Truncate(size); err != nil {
+		t.Fatalf("truncate: %v", err)
+	}
+	reads := 0
+	digest, err := ScanFile(f, "holes.img", size, func() []byte {
+		reads++
+		return make([]byte, DigestChunkSize)
+	}, func(Chunk) error { return nil })
+	if err != nil {
+		t.Fatalf("ScanFile: %v", err)
+	}
+	if reads != 0 {
+		t.Errorf("a hole-only file asked for %d read buffers, want 0", reads)
+	}
+	if len(digest.Chunks) != 6 || digest.Chunks[0] != zeroSum(DigestChunkSize) {
+		t.Errorf("hole chunks do not carry the zero-chunk sum: %d chunks", len(digest.Chunks))
+	}
+}
+
+func TestWriteChunkRejectsWrongLengthAndChecksum(t *testing.T) {
+	data := "payload"
+	sum := sha256.Sum256([]byte(data))
+	badSum := "0000000000000000000000000000000000000000000000000000000000000000"
+	for _, tc := range []struct {
+		name  string
+		size  int64
+		index int64
+		sha   string
+		body  string
+		want  string
+	}{
+		{"short body", int64(len(data)) + 1, 0, hex.EncodeToString(sum[:]), data, "want 8"},
+		{"long body", int64(len(data)) - 1, 0, hex.EncodeToString(sum[:]), data, "want 6"},
+		{"wrong sum", int64(len(data)), 0, badSum, data, "checksum"},
+		{"index outside file", int64(len(data)), 1, hex.EncodeToString(sum[:]), data, "outside"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f, err := os.Create(filepath.Join(t.TempDir(), "chunk"))
+			if err != nil {
+				t.Fatalf("create: %v", err)
+			}
+			defer func() { _ = f.Close() }()
+			err = WriteChunk(f, tc.size, tc.index, tc.sha, strings.NewReader(tc.body))
+			if err == nil || !strings.Contains(err.Error(), tc.want) {
+				t.Errorf("WriteChunk = %v, want an error containing %q", err, tc.want)
+			}
+		})
+	}
+	f, err := os.Create(filepath.Join(t.TempDir(), "ok"))
+	if err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	defer func() { _ = f.Close() }()
+	if err = WriteChunk(f, int64(len(data)), 0, hex.EncodeToString(sum[:]), strings.NewReader(data)); err != nil {
+		t.Errorf("WriteChunk of a matching chunk: %v", err)
+	}
+}
diff --git a/sandboxd/store/store.go b/sandboxd/store/store.go
index bb0d386c..55227273 100644
--- a/sandboxd/store/store.go
+++ b/sandboxd/store/store.go
@@ -59,10 +59,10 @@ type S3Config struct {
 type Store interface {
 	// Stage returns a writable staging directory whose Publish is atomic.
 	Stage(id string) (string, error)
-	// Publish atomically replaces the record's previous generation with the staged directory.
-	Publish(ctx context.Context, staging, id string) error
+	// Publish atomically replaces the record's previous generation with the staged directory; lineage scopes chunk dedup in a content-addressed backend and "" means the record id.
+	Publish(ctx context.Context, staging, id, lineage string) error
 	// PublishDigested applies Publish semantics and returns the export digest.
-	PublishDigested(ctx context.Context, staging, id string) (string, error)
+	PublishDigested(ctx context.Context, staging, id, lineage string) (string, error)
 	// Fetch materializes a record's export locally; the caller holds the record lock until the clone ends.
 	Fetch(ctx context.Context, id string) (dir string, meta []byte, digest string, err error)
 	// ReadMeta returns a record's metadata, or an error when the record does not exist.
@@ -76,7 +76,7 @@ type Store interface {
 	// SweepStaging removes abandoned staging left by a crash mid-publish.
 	SweepStaging() error
 	// SweepGenerations runs generation retention without disturbing staging or fetch caches.
-	SweepGenerations() error
+	SweepGenerations(ctx context.Context) error
 }
 
 // CheckpointID, TemplateID, and TemplateHash are the one home for the ck_/tp_ id scheme.
diff --git a/sandboxd/store/storetest/storetest.go b/sandboxd/store/storetest/storetest.go
index ef4adbb9..cb7d63a1 100644
--- a/sandboxd/store/storetest/storetest.go
+++ b/sandboxd/store/storetest/storetest.go
@@ -106,7 +106,7 @@ func runDigestContract(t *testing.T, st store.Store) {
 	if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+id+`"}`), 0o600); err != nil {
 		t.Fatalf("write digested meta: %v", err)
 	}
-	digest, err := st.PublishDigested(ctx, staging, id)
+	digest, err := st.PublishDigested(ctx, staging, id, "")
 	if err != nil {
 		t.Fatalf("PublishDigested: %v", err)
 	}
@@ -199,7 +199,7 @@ func rejectNonRegularReplacement(t *testing.T, st store.Store, id, wantDigest st
 	if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(`{"id":"`+id+`","gen":2}`), 0o600); err != nil {
 		t.Fatalf("write non-regular meta: %v", err)
 	}
-	if _, err = st.PublishDigested(t.Context(), staging, id); err == nil {
+	if _, err = st.PublishDigested(t.Context(), staging, id, ""); err == nil {
 		t.Fatal("PublishDigested accepted a non-regular export entry")
 	}
 	dir, meta, digest, err := st.Fetch(t.Context(), id)
@@ -225,7 +225,7 @@ func publishRecord(t *testing.T, st store.Store, id, meta, file, content string)
 	if err = os.WriteFile(filepath.Join(staging, store.MetaFile), []byte(meta), 0o600); err != nil {
 		t.Fatalf("write meta: %v", err)
 	}
-	if err = st.Publish(t.Context(), staging, id); err != nil {
+	if err = st.Publish(t.Context(), staging, id, ""); err != nil {
 		t.Fatalf("Publish %s: %v", id, err)
 	}
 }