diff --git a/AGENTS.md b/AGENTS.md
index d44dac6..1a9757b 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -66,6 +66,7 @@ Notes on the tests and on configuration:
 | `internal/mountfs`, `internal/treefs`, `internal/kefkash` | The hook sandbox filesystem and shell wiring. |
 | `internal/metrics`            | Every Prometheus vector, plus thin helpers.                            |
 | `internal/slog.go`            | JSON handler init.                                                     |
+| `cmd/membench/`               | Push memory benchmark harness. Not shipped; see `docs/usage/memory-benchmark.md`. |
 
 ## Architecture
 
@@ -85,7 +86,8 @@ Two more directories carry detail:
 
 - `docs/reference/` — the `.cue` binary layout, the failure modes, and the
   build order.
-- `docs/usage/` — how to use a feature, such as a hook script.
+- `docs/usage/` — how to use a feature, such as a hook script, or the push
+  memory benchmark.
 
 ## Conventions
 
diff --git a/cmd/membench/main.go b/cmd/membench/main.go
new file mode 100644
index 0000000..316edea
--- /dev/null
+++ b/cmd/membench/main.go
@@ -0,0 +1,513 @@
+// Command membench measures how much memory objgitd uses while it takes
+// pushes. It starts a daemon it owns, pushes one real repository into many
+// fresh ones, and samples the daemon's resident set and Go heap the whole time,
+// grabbing a pprof heap profile every time memory sets a new high-water mark.
+//
+// The run has three phases. A baseline establishes what the daemon costs
+// sitting idle. A sequential phase pushes one repository at a time with idle
+// gaps, which isolates the per-push cost and shows whether memory comes back
+// down afterwards. A concurrency sweep then runs K pushes at once for a few
+// values of K, which gives the slope you actually size a machine with.
+//
+// Every repository is created fresh under a UUID, so no push is ever measured
+// against a repository that already holds its objects. The harness never
+// deletes them; it writes repos.txt and leaves cleanup to you.
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/exec"
+	"os/signal"
+	"path/filepath"
+	"runtime"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/facebookgo/flagenv"
+	"github.com/tigrisdata/objgit/internal"
+	"golang.org/x/sync/errgroup"
+)
+
+// Phase names, as they appear in the phase column of samples.csv.
+const (
+	phaseBaseline = "baseline"
+	phaseIdle     = "idle"
+	phaseSeq      = "seq"
+	phaseConc     = "conc"
+)
+
+var (
+	repoPath = flag.String("repo", filepath.Join(os.Getenv("HOME"), "Code/Xe/x"), "repository to push; it is mirror-cloned once and never written to")
+	outBase  = flag.String("out", "", "parent directory for run output; empty uses the OS temp directory")
+	org      = flag.String("org", "benchtest", "org segment every benchmark repository is created under")
+
+	seqPushes = flag.Int("seq-pushes", 5, "number of sequential pushes, each to a fresh repository")
+	concSteps = flag.String("conc-steps", "1,2,4,8", "comma-separated concurrency levels to sweep; empty skips the sweep")
+
+	sampleEvery  = flag.Duration("sample-interval", 250*time.Millisecond, "how often to sample /proc and /metrics")
+	idleGap      = flag.Duration("idle-gap", 5*time.Second, "idle time between pushes, so memory has a chance to settle")
+	baselineFor  = flag.Duration("baseline", 10*time.Second, "how long to sample the idle daemon before pushing anything")
+	windowSlack  = flag.Duration("window-slack", 0, "widen each push's measurement window by this much on both sides, so work finishing after git exits still counts against it; zero uses -idle-gap")
+	peakGrowth   = flag.Float64("peak-growth", 0.05, "fractional rise in resident set that triggers a heap profile capture")
+	peakCooldown = flag.Duration("peak-cooldown", 2*time.Second, "minimum time between two peak captures")
+
+	daemonBinary     = flag.String("daemon-binary", "", "prebuilt objgitd to run; empty builds ./cmd/objgitd")
+	daemonHTTPBind   = flag.String("daemon-http-bind", "127.0.0.1:8080", "address the daemon under test serves smart HTTP on")
+	daemonMetrics    = flag.String("daemon-metrics-bind", "127.0.0.1:9090", "address the daemon under test serves /metrics and /debug/pprof on")
+	daemonAllowHooks = flag.Bool("daemon-allow-hooks", false, "run push hooks in the daemon under test; off by default so hook cost is not mistaken for push cost")
+	daemonReadyWait  = flag.Duration("daemon-ready-wait", 60*time.Second, "how long to wait for the daemon to answer /metrics before giving up")
+
+	slogLevel = flag.String("slog-level", "INFO", "log level (DEBUG, INFO, WARN, ERROR)")
+)
+
+func main() {
+	flagenv.Parse()
+	flag.Parse()
+
+	logger, err := internal.InitSlog(*slogLevel)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "bad -slog-level: %v\n", err)
+		os.Exit(1)
+	}
+	slog.SetDefault(logger)
+
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer cancel()
+
+	if err := run(ctx); err != nil {
+		slog.Error("benchmark failed", "err", err)
+		os.Exit(1)
+	}
+}
+
+func run(ctx context.Context) error {
+	steps, err := parseSteps(*concSteps)
+	if err != nil {
+		return err
+	}
+
+	root, err := moduleRoot()
+	if err != nil {
+		return err
+	}
+
+	startedAt := time.Now()
+	runDir := filepath.Join(outDir(), startedAt.Format("20060102-150405"))
+	if err := os.MkdirAll(runDir, 0o755); err != nil {
+		return fmt.Errorf("membench: can't create %s: %w", runDir, err)
+	}
+	slog.Info("run directory", "path", runDir)
+
+	mirror := filepath.Join(runDir, "source.git")
+	if err := mirrorClone(ctx, *repoPath, mirror); err != nil {
+		return err
+	}
+
+	packed, err := packBytes(mirror)
+	if err != nil {
+		return err
+	}
+	slog.Info("mirrored source repository", "path", mirror, "pack_bytes", packed)
+
+	bin := *daemonBinary
+	if bin == "" {
+		bin = filepath.Join(runDir, "objgitd")
+		if err := buildDaemon(ctx, root, bin); err != nil {
+			return err
+		}
+	}
+
+	// -pack-cache-dir names the parent the daemon makes its own cache directory
+	// under, so it has to exist before the daemon starts.
+	packCache := filepath.Join(runDir, "packcache")
+	if err := os.MkdirAll(packCache, 0o755); err != nil {
+		return fmt.Errorf("membench: can't create %s: %w", packCache, err)
+	}
+
+	args := []string{
+		"-http-bind", *daemonHTTPBind,
+		"-metrics-bind", *daemonMetrics,
+		"-ssh-bind=",
+		"-allow-push",
+		fmt.Sprintf("-allow-hooks=%t", *daemonAllowHooks),
+		"-pack-cache-dir", packCache,
+	}
+
+	daemon, err := startDaemon(root, bin, filepath.Join(runDir, "daemon.log"), args)
+	if err != nil {
+		return err
+	}
+	defer stopDaemon(daemon)
+
+	if err := waitReady(ctx, *daemonMetrics, *daemonReadyWait); err != nil {
+		return err
+	}
+	slog.Info("daemon ready", "pid", daemon.Process.Pid, "http", *daemonHTTPBind, "metrics", *daemonMetrics)
+
+	s := newSampler(daemon.Process.Pid, *daemonMetrics, runDir, *sampleEvery, *peakCooldown, *peakGrowth)
+	sampleCtx, stopSampling := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		s.run(sampleCtx)
+	}()
+
+	pushes, runErr := drive(ctx, s, mirror, steps)
+
+	stopSampling()
+	<-done
+
+	samples, profiles := s.snapshot()
+	meta := runMeta{
+		StartedAt:   startedAt,
+		FinishedAt:  time.Now(),
+		Host:        hostname(),
+		GoVersion:   runtime.Version(),
+		GOGC:        os.Getenv("GOGC"),
+		GOMEMLIMIT:  os.Getenv("GOMEMLIMIT"),
+		NumCPU:      runtime.NumCPU(),
+		SourceRepo:  *repoPath,
+		PackBytes:   packed,
+		Org:         *org,
+		DaemonArgs:  args,
+		SampleEvery: *sampleEvery,
+		WindowSlack: slack(),
+		RunDir:      runDir,
+	}
+
+	if err := writeCSV(filepath.Join(runDir, "samples.csv"), samples); err != nil {
+		return err
+	}
+	if err := writeRepoList(filepath.Join(runDir, "repos.txt"), *org, pushes); err != nil {
+		return err
+	}
+	if err := writeReport(filepath.Join(runDir, "report.md"), meta, samples, profiles, pushes); err != nil {
+		return err
+	}
+
+	slog.Info("wrote results",
+		"report", filepath.Join(runDir, "report.md"),
+		"samples", len(samples),
+		"profiles", len(profiles),
+		"repos", len(pushes),
+	)
+	fmt.Printf("\n%s\n\nRepositories left in the bucket are listed in %s.\n",
+		filepath.Join(runDir, "report.md"), filepath.Join(runDir, "repos.txt"))
+
+	return runErr
+}
+
+// drive walks the three phases. It returns whatever it managed to record even
+// when a push fails, because a partial run with a heap profile in it is still
+// worth reading.
+func drive(ctx context.Context, s *sampler, mirror string, steps []int) ([]pushResult, error) {
+	var pushes []pushResult
+
+	s.label(phaseBaseline, "")
+	slog.Info("sampling idle baseline", "for", *baselineFor)
+	if err := idle(ctx, *baselineFor); err != nil {
+		return pushes, err
+	}
+	if _, err := s.captureProfile(ctx, "heap", true, "heap-baseline.pb.gz", "idle baseline before any push", lastRSS(s)); err != nil {
+		slog.Warn("can't capture baseline heap profile", "err", err)
+	}
+
+	for i := range *seqPushes {
+		if err := ctx.Err(); err != nil {
+			return pushes, err
+		}
+
+		name, err := newUUID()
+		if err != nil {
+			return pushes, err
+		}
+
+		s.label(phaseSeq, name)
+		slog.Info("sequential push", "n", i+1, "of", *seqPushes, "repo", *org+"/"+name)
+		pushes = append(pushes, doPush(ctx, mirror, phaseSeq, 1, name))
+
+		s.label(phaseIdle, "")
+		if err := idle(ctx, *idleGap); err != nil {
+			return pushes, err
+		}
+	}
+
+	for _, k := range steps {
+		if err := ctx.Err(); err != nil {
+			return pushes, err
+		}
+
+		names := make([]string, k)
+		for i := range names {
+			name, err := newUUID()
+			if err != nil {
+				return pushes, err
+			}
+			names[i] = name
+		}
+
+		s.label(phaseConc, fmt.Sprintf("k=%d", k))
+		slog.Info("concurrent push step", "k", k)
+
+		results := make([]pushResult, k)
+		g, gCtx := errgroup.WithContext(ctx)
+		for i, name := range names {
+			g.Go(func() error {
+				results[i] = doPush(gCtx, mirror, phaseConc, k, name)
+				return nil
+			})
+		}
+		_ = g.Wait()
+		pushes = append(pushes, results...)
+
+		s.label(phaseIdle, "")
+		if err := idle(ctx, *idleGap); err != nil {
+			return pushes, err
+		}
+	}
+
+	s.label(phaseIdle, "")
+	if err := idle(ctx, *idleGap); err != nil {
+		return pushes, err
+	}
+	if _, err := s.captureProfile(ctx, "heap", true, "heap-final.pb.gz", "settled heap after every push", lastRSS(s)); err != nil {
+		slog.Warn("can't capture final heap profile", "err", err)
+	}
+	if _, err := s.captureProfile(ctx, "allocs", false, "allocs-final.pb.gz", "total allocation over the whole run", lastRSS(s)); err != nil {
+		slog.Warn("can't capture allocation profile", "err", err)
+	}
+
+	return pushes, nil
+}
+
+func doPush(ctx context.Context, mirror, phase string, concurrency int, name string) pushResult {
+	url := fmt.Sprintf("http://%s/%s/%s.git", *daemonHTTPBind, *org, name)
+
+	start := time.Now()
+	err := push(ctx, mirror, url)
+	end := time.Now()
+
+	if err != nil {
+		slog.Error("push failed", "repo", *org+"/"+name, "err", err)
+	} else {
+		slog.Info("push done", "repo", *org+"/"+name, "took", end.Sub(start).Round(time.Millisecond))
+	}
+
+	return pushResult{Phase: phase, Concurrency: concurrency, Repo: name, Start: start, End: end, Err: err}
+}
+
+// push sends every branch and tag to a repository that does not exist yet.
+// The refspecs are forced because the harness does not care about history on
+// the receiving side; it only cares what the daemon does with the pack.
+func push(ctx context.Context, mirror, url string) error {
+	cmd := exec.CommandContext(ctx, "git", "-C", mirror, "push", "--quiet", url,
+		"+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*")
+
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("git push to %s: %w: %s", url, err, strings.TrimSpace(string(out)))
+	}
+
+	return nil
+}
+
+// mirrorClone copies the source repository once, so the benchmark never writes
+// to the repository it was pointed at and concurrent pushes all read from the
+// same immutable copy.
+func mirrorClone(ctx context.Context, src, dst string) error {
+	cmd := exec.CommandContext(ctx, "git", "clone", "--mirror", src, dst)
+
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("membench: can't mirror %s: %w: %s", src, err, strings.TrimSpace(string(out)))
+	}
+
+	return nil
+}
+
+func buildDaemon(ctx context.Context, root, out string) error {
+	cmd := exec.CommandContext(ctx, "go", "build", "-o", out, "./cmd/objgitd")
+	cmd.Dir = root
+
+	res, err := cmd.CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("membench: can't build objgitd: %w: %s", err, strings.TrimSpace(string(res)))
+	}
+
+	slog.Info("built daemon under test", "path", out)
+	return nil
+}
+
+// startDaemon runs objgitd with its working directory set to the module root,
+// because objgitd loads its bucket and credentials from the .env file there.
+// The context is deliberately not attached: shutdown goes through stopDaemon so
+// the daemon gets the same SIGINT it would in production.
+func startDaemon(root, bin, logPath string, args []string) (*exec.Cmd, error) {
+	log, err := os.Create(logPath)
+	if err != nil {
+		return nil, fmt.Errorf("membench: can't create %s: %w", logPath, err)
+	}
+
+	cmd := exec.Command(bin, args...)
+	cmd.Dir = root
+	cmd.Stdout = log
+	cmd.Stderr = log
+
+	if err := cmd.Start(); err != nil {
+		log.Close()
+		return nil, fmt.Errorf("membench: can't start %s: %w", bin, err)
+	}
+
+	return cmd, nil
+}
+
+func stopDaemon(cmd *exec.Cmd) {
+	if cmd.Process == nil {
+		return
+	}
+
+	if err := cmd.Process.Signal(os.Interrupt); err != nil {
+		slog.Warn("can't interrupt daemon", "err", err)
+	}
+
+	done := make(chan error, 1)
+	go func() { done <- cmd.Wait() }()
+
+	select {
+	case <-done:
+	case <-time.After(15 * time.Second):
+		slog.Warn("daemon did not exit on SIGINT, killing it")
+		_ = cmd.Process.Kill()
+		<-done
+	}
+}
+
+func waitReady(ctx context.Context, metricsAddr string, limit time.Duration) error {
+	url := "http://" + metricsAddr + "/metrics"
+	client := &http.Client{Timeout: 2 * time.Second}
+	deadline := time.Now().Add(limit)
+
+	for time.Now().Before(deadline) {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+		if err != nil {
+			return fmt.Errorf("membench: can't build readiness request: %w", err)
+		}
+
+		resp, err := client.Do(req)
+		if err == nil {
+			resp.Body.Close()
+			if resp.StatusCode == http.StatusOK {
+				return nil
+			}
+		}
+
+		if err := idle(ctx, 250*time.Millisecond); err != nil {
+			return err
+		}
+	}
+
+	return fmt.Errorf("membench: daemon did not answer %s within %s", url, limit)
+}
+
+// idle sleeps, but gives up as soon as the run is cancelled.
+func idle(ctx context.Context, d time.Duration) error {
+	t := time.NewTimer(d)
+	defer t.Stop()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-t.C:
+		return nil
+	}
+}
+
+func lastRSS(s *sampler) uint64 {
+	samples, _ := s.snapshot()
+	if len(samples) == 0 {
+		return 0
+	}
+	return samples[len(samples)-1].Proc.VmRSS
+}
+
+// parseSteps turns "1,2,4,8" into the concurrency levels to sweep.
+func parseSteps(raw string) ([]int, error) {
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return nil, nil
+	}
+
+	var out []int
+	for _, part := range strings.Split(raw, ",") {
+		part = strings.TrimSpace(part)
+		if part == "" {
+			continue
+		}
+		n, err := strconv.Atoi(part)
+		if err != nil {
+			return nil, fmt.Errorf("membench: %q in -conc-steps is not a number: %w", part, err)
+		}
+		if n < 1 {
+			return nil, fmt.Errorf("membench: -conc-steps must be positive, got %d", n)
+		}
+		out = append(out, n)
+	}
+
+	return out, nil
+}
+
+// slack is how far outside a push's wall clock the harness still attributes
+// memory to it. The daemon keeps uploading after git has exited, so the default
+// is the whole idle gap that follows.
+func slack() time.Duration {
+	if *windowSlack > 0 {
+		return *windowSlack
+	}
+	return *idleGap
+}
+
+func outDir() string {
+	if *outBase != "" {
+		return *outBase
+	}
+	return filepath.Join(os.TempDir(), "membench")
+}
+
+// moduleRoot walks up from the working directory to the directory holding
+// go.mod. That is where objgitd's .env lives, and where `go build` has to run.
+func moduleRoot() (string, error) {
+	dir, err := os.Getwd()
+	if err != nil {
+		return "", fmt.Errorf("membench: can't read working directory: %w", err)
+	}
+
+	for {
+		if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil {
+			return dir, nil
+		}
+		parent := filepath.Dir(dir)
+		if parent == dir {
+			return "", fmt.Errorf("membench: no go.mod above the working directory; run this from the objgit checkout")
+		}
+		dir = parent
+	}
+}
+
+func hostname() string {
+	h, err := os.Hostname()
+	if err != nil {
+		return "unknown"
+	}
+	return h
+}
diff --git a/cmd/membench/metrics.go b/cmd/membench/metrics.go
new file mode 100644
index 0000000..dbbf986
--- /dev/null
+++ b/cmd/membench/metrics.go
@@ -0,0 +1,80 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"strconv"
+	"strings"
+)
+
+// goMetrics is the subset of objgitd's /metrics this harness records. The Go
+// runtime numbers say what the heap is doing; ProcessRSS is the same figure
+// /proc reports, kept as a cross-check that the sampler is watching the pid it
+// thinks it is.
+type goMetrics struct {
+	HeapInuse  uint64
+	HeapSys    uint64
+	NextGC     uint64
+	Goroutines uint64
+	ProcessRSS uint64
+}
+
+// field returns where a metric name lands, or nil for a metric this harness
+// does not record.
+func (m *goMetrics) field(name string) *uint64 {
+	switch name {
+	case "go_memstats_heap_inuse_bytes":
+		return &m.HeapInuse
+	case "go_memstats_heap_sys_bytes":
+		return &m.HeapSys
+	case "go_memstats_next_gc_bytes":
+		return &m.NextGC
+	case "go_goroutines":
+		return &m.Goroutines
+	case "process_resident_memory_bytes":
+		return &m.ProcessRSS
+	}
+	return nil
+}
+
+// parseMetrics reads the Prometheus text exposition format far enough to pull
+// out the handful of unlabelled gauges above. It deliberately skips any sample
+// carrying labels: every metric this harness wants is a single series, so a
+// name with a "{" is one of objgitd's own vectors and not a match.
+func parseMetrics(r io.Reader) (goMetrics, error) {
+	var m goMetrics
+
+	sc := bufio.NewScanner(r)
+	sc.Buffer(make([]byte, 0, 64*1024), 1<<20)
+	for sc.Scan() {
+		line := sc.Text()
+		if line == "" || line[0] == '#' {
+			continue
+		}
+
+		name, value, ok := strings.Cut(line, " ")
+		if !ok || strings.ContainsRune(name, '{') {
+			continue
+		}
+
+		dst := m.field(name)
+		if dst == nil {
+			continue
+		}
+
+		f, err := strconv.ParseFloat(value, 64)
+		if err != nil {
+			return m, fmt.Errorf("membench: metric %s has non-numeric value %q: %w", name, value, err)
+		}
+		if f < 0 {
+			f = 0
+		}
+		*dst = uint64(f)
+	}
+	if err := sc.Err(); err != nil {
+		return m, fmt.Errorf("membench: reading metrics: %w", err)
+	}
+
+	return m, nil
+}
diff --git a/cmd/membench/metrics_test.go b/cmd/membench/metrics_test.go
new file mode 100644
index 0000000..bba6d75
--- /dev/null
+++ b/cmd/membench/metrics_test.go
@@ -0,0 +1,77 @@
+package main
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestParseMetrics(t *testing.T) {
+	t.Parallel()
+
+	for _, tt := range []struct {
+		name    string
+		input   string
+		want    goMetrics
+		wantErr bool
+	}{
+		{
+			name: "the metrics the harness records",
+			input: "# HELP go_goroutines Number of goroutines that currently exist.\n" +
+				"# TYPE go_goroutines gauge\n" +
+				"go_goroutines 42\n" +
+				"go_memstats_heap_inuse_bytes 1.2345678e+07\n" +
+				"go_memstats_heap_sys_bytes 3.3554432e+07\n" +
+				"go_memstats_next_gc_bytes 4.194304e+06\n" +
+				"process_resident_memory_bytes 2.8835840e+07\n",
+			want: goMetrics{
+				HeapInuse:  12345678,
+				HeapSys:    33554432,
+				NextGC:     4194304,
+				Goroutines: 42,
+				ProcessRSS: 28835840,
+			},
+		},
+		{
+			name: "labelled series are skipped",
+			input: "objgit_pushes_total{repo=\"a/b\"} 99\n" +
+				"go_goroutines 7\n",
+			want: goMetrics{Goroutines: 7},
+		},
+		{
+			name:  "a metric name that happens to prefix a wanted one",
+			input: "go_goroutines_extra 5\n",
+			want:  goMetrics{},
+		},
+		{
+			name:    "non-numeric value",
+			input:   "go_goroutines NaNsense\n",
+			wantErr: true,
+		},
+		{
+			name:  "comments and blanks only",
+			input: "# HELP nothing\n\n",
+			want:  goMetrics{},
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := parseMetrics(strings.NewReader(tt.input))
+			if tt.wantErr {
+				if err == nil {
+					t.Error("wanted an error, got none")
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+
+			if got != tt.want {
+				t.Logf("want: %+v", tt.want)
+				t.Logf("got:  %+v", got)
+				t.Error("parsed the wrong metrics")
+			}
+		})
+	}
+}
diff --git a/cmd/membench/proc.go b/cmd/membench/proc.go
new file mode 100644
index 0000000..1b982e4
--- /dev/null
+++ b/cmd/membench/proc.go
@@ -0,0 +1,83 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+	"strconv"
+	"strings"
+)
+
+// procStatus is the subset of /proc/<pid>/status and /proc/<pid>/smaps_rollup
+// this harness records, in bytes. The kernel reports kB; parseProcKV converts.
+//
+// VmHWM is the number that matters for sizing: it is the high-water mark the
+// kernel itself keeps, so it cannot be missed between two samples the way a
+// spike in VmRSS can be.
+type procStatus struct {
+	VmRSS        uint64
+	VmHWM        uint64
+	Pss          uint64
+	PrivateDirty uint64
+}
+
+// parseProcKV reads the "Key:\t   1234 kB" lines that both /proc/<pid>/status
+// and /proc/<pid>/smaps_rollup use, and returns them in bytes. Lines that do
+// not carry a kB count (Name, State, the smaps_rollup header) are skipped,
+// because no caller wants them.
+func parseProcKV(r io.Reader) (map[string]uint64, error) {
+	out := map[string]uint64{}
+	sc := bufio.NewScanner(r)
+	for sc.Scan() {
+		key, rest, ok := strings.Cut(sc.Text(), ":")
+		if !ok {
+			continue
+		}
+		fields := strings.Fields(rest)
+		if len(fields) != 2 || fields[1] != "kB" {
+			continue
+		}
+		n, err := strconv.ParseUint(fields[0], 10, 64)
+		if err != nil {
+			return nil, fmt.Errorf("membench: %q is not a kB count: %w", sc.Text(), err)
+		}
+		out[key] = n * 1024
+	}
+	if err := sc.Err(); err != nil {
+		return nil, fmt.Errorf("membench: reading proc file: %w", err)
+	}
+	return out, nil
+}
+
+// readProcStatus samples one process. /proc/<pid>/status is required;
+// smaps_rollup is best effort, because it needs a 4.14 or newer kernel and
+// permission to read another process's mappings. Losing Pss costs detail, not
+// the measurement.
+func readProcStatus(pid int) (procStatus, error) {
+	var ps procStatus
+
+	kv, err := readProcFile(fmt.Sprintf("/proc/%d/status", pid))
+	if err != nil {
+		return ps, err
+	}
+	ps.VmRSS = kv["VmRSS"]
+	ps.VmHWM = kv["VmHWM"]
+
+	if rollup, err := readProcFile(fmt.Sprintf("/proc/%d/smaps_rollup", pid)); err == nil {
+		ps.Pss = rollup["Pss"]
+		ps.PrivateDirty = rollup["Private_Dirty"]
+	}
+
+	return ps, nil
+}
+
+func readProcFile(name string) (map[string]uint64, error) {
+	f, err := os.Open(name)
+	if err != nil {
+		return nil, fmt.Errorf("membench: can't open %s: %w", name, err)
+	}
+	defer f.Close()
+
+	return parseProcKV(f)
+}
diff --git a/cmd/membench/proc_test.go b/cmd/membench/proc_test.go
new file mode 100644
index 0000000..2a89917
--- /dev/null
+++ b/cmd/membench/proc_test.go
@@ -0,0 +1,77 @@
+package main
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestParseProcKV(t *testing.T) {
+	t.Parallel()
+
+	for _, tt := range []struct {
+		name    string
+		input   string
+		want    map[string]uint64
+		wantErr bool
+	}{
+		{
+			name: "status excerpt",
+			input: "Name:\tobjgitd\n" +
+				"State:\tS (sleeping)\n" +
+				"VmRSS:\t   28160 kB\n" +
+				"VmHWM:\t   31744 kB\n",
+			want: map[string]uint64{"VmRSS": 28160 * 1024, "VmHWM": 31744 * 1024},
+		},
+		{
+			name: "smaps_rollup excerpt",
+			input: "55d0c0000000-7ffd0c0f1000 ---p 00000000 00:00 0                          [rollup]\n" +
+				"Rss:                7168 kB\n" +
+				"Pss:                4096 kB\n" +
+				"Private_Dirty:      2048 kB\n",
+			want: map[string]uint64{"Rss": 7168 * 1024, "Pss": 4096 * 1024, "Private_Dirty": 2048 * 1024},
+		},
+		{
+			name:  "no kB lines",
+			input: "Name:\tobjgitd\nThreads:\t12\n",
+			want:  map[string]uint64{},
+		},
+		{
+			name:    "unparseable count",
+			input:   "VmRSS:\t   twelve kB\n",
+			wantErr: true,
+		},
+		{
+			name:  "empty",
+			input: "",
+			want:  map[string]uint64{},
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := parseProcKV(strings.NewReader(tt.input))
+			if tt.wantErr {
+				if err == nil {
+					t.Error("wanted an error, got none")
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+
+			if len(got) != len(tt.want) {
+				t.Logf("want: %v", tt.want)
+				t.Logf("got:  %v", got)
+				t.Fatalf("got %d keys, want %d", len(got), len(tt.want))
+			}
+			for k, want := range tt.want {
+				if got[k] != want {
+					t.Logf("want: %d", want)
+					t.Logf("got:  %d", got[k])
+					t.Errorf("%s is wrong", k)
+				}
+			}
+		})
+	}
+}
diff --git a/cmd/membench/report.go b/cmd/membench/report.go
new file mode 100644
index 0000000..3a6b5a6
--- /dev/null
+++ b/cmd/membench/report.go
@@ -0,0 +1,374 @@
+package main
+
+import (
+	"encoding/csv"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// pushResult is one git push the harness drove, with the window the sampler
+// should be searched over to find what that push cost.
+type pushResult struct {
+	Phase       string
+	Concurrency int
+	Repo        string
+	Start       time.Time
+	End         time.Time
+	Err         error
+}
+
+func (p pushResult) duration() time.Duration { return p.End.Sub(p.Start) }
+
+// runMeta is everything about the run that is not a measurement, but that the
+// numbers are meaningless without. Go's resident set is a function of GOGC and
+// GOMEMLIMIT, so a peak-RSS figure recorded without them cannot be compared to
+// anything.
+type runMeta struct {
+	StartedAt   time.Time
+	FinishedAt  time.Time
+	Host        string
+	GoVersion   string
+	GOGC        string
+	GOMEMLIMIT  string
+	NumCPU      int
+	SourceRepo  string
+	PackBytes   int64
+	Org         string
+	DaemonArgs  []string
+	SampleEvery time.Duration
+	WindowSlack time.Duration
+	RunDir      string
+}
+
+// windowPeak returns the highest resident set and heap-in-use seen between two
+// instants. The window is widened by one sample interval on each side because
+// a push's cost does not land inside its own wall clock exactly: the daemon is
+// still flushing to the bucket when git has already exited.
+func windowPeak(samples []sample, start, end time.Time, slack time.Duration) (rss, heap uint64) {
+	from, to := start.Add(-slack), end.Add(slack)
+	for _, s := range samples {
+		if s.At.Before(from) || s.At.After(to) {
+			continue
+		}
+		if s.Proc.VmRSS > rss {
+			rss = s.Proc.VmRSS
+		}
+		if s.GoOK && s.Go.HeapInuse > heap {
+			heap = s.Go.HeapInuse
+		}
+	}
+	return rss, heap
+}
+
+// rssAt returns the resident set at the most recent sample at or before t, and
+// zero when the sampler had not started yet. It is what makes retention legible:
+// the reading going into a push, compared with the reading coming out of it.
+func rssAt(samples []sample, t time.Time) uint64 {
+	var out uint64
+	for _, s := range samples {
+		if s.At.After(t) {
+			break
+		}
+		out = s.Proc.VmRSS
+	}
+	return out
+}
+
+// phaseTail returns the last sample recorded for a phase, which is the settled
+// reading for it: the idle value after everything has drained.
+//
+// Samples whose metrics scrape failed are skipped. The last tick before the
+// daemon exits routinely fails, and taking it would report a settled heap of
+// zero rather than the figure that was actually reached.
+func phaseTail(samples []sample, phase string) (sample, bool) {
+	for i := len(samples) - 1; i >= 0; i-- {
+		if samples[i].Phase == phase && samples[i].GoOK {
+			return samples[i], true
+		}
+	}
+	return sample{}, false
+}
+
+// goCell renders a runtime metric, or an empty cell when that tick's scrape
+// failed. Empty is deliberate: a zero here would plot as a cliff to zero.
+func goCell(ok bool, v uint64) string {
+	if !ok {
+		return ""
+	}
+	return strconv.FormatUint(v, 10)
+}
+
+func writeCSV(path string, samples []sample) error {
+	f, err := os.Create(path)
+	if err != nil {
+		return fmt.Errorf("membench: can't create %s: %w", path, err)
+	}
+	defer f.Close()
+
+	w := csv.NewWriter(f)
+	defer w.Flush()
+
+	header := []string{
+		"t_ms", "iso8601", "phase", "repo",
+		"vm_rss", "vm_hwm", "pss", "private_dirty",
+		"heap_inuse", "heap_sys", "next_gc", "goroutines", "process_rss",
+	}
+	if err := w.Write(header); err != nil {
+		return fmt.Errorf("membench: can't write CSV header: %w", err)
+	}
+
+	if len(samples) == 0 {
+		return nil
+	}
+
+	origin := samples[0].At
+	for _, s := range samples {
+		row := []string{
+			strconv.FormatInt(s.At.Sub(origin).Milliseconds(), 10),
+			s.At.Format(time.RFC3339Nano),
+			s.Phase,
+			s.Repo,
+			strconv.FormatUint(s.Proc.VmRSS, 10),
+			strconv.FormatUint(s.Proc.VmHWM, 10),
+			strconv.FormatUint(s.Proc.Pss, 10),
+			strconv.FormatUint(s.Proc.PrivateDirty, 10),
+			goCell(s.GoOK, s.Go.HeapInuse),
+			goCell(s.GoOK, s.Go.HeapSys),
+			goCell(s.GoOK, s.Go.NextGC),
+			goCell(s.GoOK, s.Go.Goroutines),
+			goCell(s.GoOK, s.Go.ProcessRSS),
+		}
+		if err := w.Write(row); err != nil {
+			return fmt.Errorf("membench: can't write CSV row: %w", err)
+		}
+	}
+
+	return w.Error()
+}
+
+// mib formats bytes as MiB, which is the only unit anyone reads these numbers
+// in.
+func mib(b uint64) string { return fmt.Sprintf("%.1f MiB", float64(b)/(1<<20)) }
+
+// deltaMiB formats a signed difference in MiB.
+func deltaMiB(now, base uint64) string {
+	d := (float64(now) - float64(base)) / (1 << 20)
+	return fmt.Sprintf("%+.1f MiB", d)
+}
+
+func writeReport(path string, meta runMeta, samples []sample, profiles []profileCapture, pushes []pushResult) error {
+	var b strings.Builder
+
+	baseRSS, baseHeap := uint64(0), uint64(0)
+	if s, ok := phaseTail(samples, phaseBaseline); ok {
+		baseRSS, baseHeap = s.Proc.VmRSS, s.Go.HeapInuse
+	}
+
+	var hwm uint64
+	for _, s := range samples {
+		if s.Proc.VmHWM > hwm {
+			hwm = s.Proc.VmHWM
+		}
+	}
+
+	fmt.Fprintf(&b, "# objgitd push memory benchmark\n\n")
+	fmt.Fprintf(&b, "Run started %s, finished %s (%s).\n\n",
+		meta.StartedAt.Format(time.RFC3339),
+		meta.FinishedAt.Format(time.RFC3339),
+		meta.FinishedAt.Sub(meta.StartedAt).Round(time.Second))
+
+	fmt.Fprintf(&b, "## Headline\n\n")
+	fmt.Fprintf(&b, "- Peak resident set for the whole run (VmHWM): **%s**\n", mib(hwm))
+	fmt.Fprintf(&b, "- Idle baseline before any push: %s resident, %s heap in use\n", mib(baseRSS), mib(baseHeap))
+	fmt.Fprintf(&b, "- Source pack pushed each time: %s\n", mib(uint64(meta.PackBytes)))
+	if s, ok := phaseTail(samples, phaseIdle); ok {
+		fmt.Fprintf(&b, "- Settled after the last push, once idle: %s resident, %s heap in use (%s of heap over baseline)\n",
+			mib(s.Proc.VmRSS), mib(s.Go.HeapInuse), deltaMiB(s.Go.HeapInuse, baseHeap))
+	}
+	fmt.Fprintf(&b, "- Repositories created: %d under `%s/`\n\n", len(pushes), meta.Org)
+
+	fmt.Fprintf(&b, "## Run conditions\n\n")
+	fmt.Fprintf(&b, "| Setting | Value |\n| --- | --- |\n")
+	fmt.Fprintf(&b, "| Host | %s |\n", meta.Host)
+	fmt.Fprintf(&b, "| Go | %s |\n", meta.GoVersion)
+	fmt.Fprintf(&b, "| GOGC | %s |\n", orDefault(meta.GOGC, "unset (100)"))
+	fmt.Fprintf(&b, "| GOMEMLIMIT | %s |\n", orDefault(meta.GOMEMLIMIT, "unset (no limit)"))
+	fmt.Fprintf(&b, "| CPUs | %d |\n", meta.NumCPU)
+	fmt.Fprintf(&b, "| Source repo | %s |\n", meta.SourceRepo)
+	fmt.Fprintf(&b, "| Sample interval | %s |\n", meta.SampleEvery)
+	fmt.Fprintf(&b, "| Daemon flags | `%s` |\n", strings.Join(meta.DaemonArgs, " "))
+	fmt.Fprintf(&b, "| Run directory | %s |\n\n", meta.RunDir)
+
+	slack := meta.WindowSlack
+
+	fmt.Fprintf(&b, "## Sequential pushes\n\n")
+	fmt.Fprintf(&b, "One push at a time, each to a fresh repository, with an idle gap between them. ")
+	fmt.Fprintf(&b, "Peaks are taken over the push plus %s of slack, so work the daemon finishes after git exits still counts against the push that caused it.\n\n", slack)
+	fmt.Fprintf(&b, "Read the before and after columns together. After is the settled reading once the idle gap has passed: if it tracks upwards push after push instead of falling back to before, the daemon is keeping memory it no longer needs.\n\n")
+	fmt.Fprintf(&b, "| # | Repo | Wall clock | RSS before | Peak RSS | Peak heap | RSS after | Kept | Result |\n")
+	fmt.Fprintf(&b, "| --- | --- | --- | --- | --- | --- | --- | --- | --- |\n")
+	n := 0
+	for _, p := range pushes {
+		if p.Phase != phaseSeq {
+			continue
+		}
+		n++
+		rss, heap := windowPeak(samples, p.Start, p.End, slack)
+		before := rssAt(samples, p.Start)
+		after := rssAt(samples, p.End.Add(slack))
+		fmt.Fprintf(&b, "| %d | `%s` | %s | %s | %s | %s | %s | %s | %s |\n",
+			n, p.Repo, p.duration().Round(time.Millisecond),
+			mib(before), mib(rss), mib(heap), mib(after), deltaMiB(after, before), result(p.Err))
+	}
+
+	fmt.Fprintf(&b, "\n## Concurrency sweep\n\n")
+	fmt.Fprintf(&b, "K simultaneous pushes to K fresh repositories. The rise is measured from the reading going into the step, not from the original idle baseline: ")
+	fmt.Fprintf(&b, "if earlier pushes left memory behind, measuring from the baseline would charge that to this step as well.\n\n")
+	fmt.Fprintf(&b, "Per concurrent push is the rise divided by K. That is the slope to size a machine with, and it is only meaningful if it stays flat as K grows.\n\n")
+	fmt.Fprintf(&b, "| K | Wall clock | RSS before | Peak RSS | Peak heap | Rise | Per concurrent push | Failures |\n")
+	fmt.Fprintf(&b, "| --- | --- | --- | --- | --- | --- | --- | --- |\n")
+	for _, step := range concurrencySteps(pushes) {
+		group := pushesAt(pushes, step)
+		start, end := groupWindow(group)
+		rss, heap := windowPeak(samples, start, end, slack)
+		before := rssAt(samples, start)
+		rise := float64(rss) - float64(before)
+		fails := 0
+		for _, p := range group {
+			if p.Err != nil {
+				fails++
+			}
+		}
+		fmt.Fprintf(&b, "| %d | %s | %s | %s | %s | %s | %.1f MiB | %d |\n",
+			step, end.Sub(start).Round(time.Millisecond), mib(before), mib(rss), mib(heap),
+			deltaMiB(rss, before), rise/float64(step)/(1<<20), fails)
+	}
+
+	fmt.Fprintf(&b, "\n## Captured profiles\n\n")
+	if len(profiles) == 0 {
+		fmt.Fprintf(&b, "None.\n")
+	} else {
+		fmt.Fprintf(&b, "| File | Taken at | RSS then | Why |\n| --- | --- | --- | --- |\n")
+		for _, p := range profiles {
+			fmt.Fprintf(&b, "| `%s` | %s | %s | %s |\n",
+				p.Name, p.At.Format("15:04:05.000"), mib(p.RSS), p.Reason)
+		}
+	}
+
+	fmt.Fprintf(&b, "\n## Reading the profiles\n\n")
+	fmt.Fprintf(&b, "```sh\n")
+	fmt.Fprintf(&b, "# What was on the heap at the worst moment.\n")
+	fmt.Fprintf(&b, "go tool pprof -http=: %s/heap-peak-*.pb.gz\n\n", meta.RunDir)
+	fmt.Fprintf(&b, "# What the pushes left behind: the settled heap minus the idle baseline.\n")
+	fmt.Fprintf(&b, "go tool pprof -http=: -base %s/heap-baseline.pb.gz %s/heap-final.pb.gz\n\n", meta.RunDir, meta.RunDir)
+	fmt.Fprintf(&b, "# Total allocation over the whole run, which finds the churn GC is working to keep up with.\n")
+	fmt.Fprintf(&b, "go tool pprof -sample_index=alloc_space -http=: %s/allocs-final.pb.gz\n", meta.RunDir)
+	fmt.Fprintf(&b, "```\n\n")
+
+	fmt.Fprintf(&b, "## Caveats\n\n")
+	fmt.Fprintf(&b, "- Peak heap profiles are taken with `gc=0`. Forcing a collection would change the number that triggered the capture, so the profile shows the heap as it was, sampling error included.\n")
+	fmt.Fprintf(&b, "- Resident set lags the heap. Go returns freed pages to the kernel lazily, so RSS staying high after a push is not by itself a leak; the baseline-diffed heap profile is what settles that.\n")
+	fmt.Fprintf(&b, "- Pushes go to a real Tigris bucket, so wall clock includes network time and varies run to run. Memory does not depend on it, but throughput comparisons across runs do.\n")
+	fmt.Fprintf(&b, "- The pack cache is on local disk (`-pack-cache-dir` under the run directory) and is cold at the start of every run. Disk pressure there is not memory pressure.\n")
+
+	if err := os.WriteFile(path, []byte(b.String()), 0o644); err != nil {
+		return fmt.Errorf("membench: can't write %s: %w", path, err)
+	}
+
+	return nil
+}
+
+func orDefault(s, fallback string) string {
+	if s == "" {
+		return fallback
+	}
+	return s
+}
+
+func result(err error) string {
+	if err != nil {
+		return "FAILED: " + err.Error()
+	}
+	return "ok"
+}
+
+// concurrencySteps returns the distinct K values in the concurrency phase, in
+// ascending order.
+func concurrencySteps(pushes []pushResult) []int {
+	seen := map[int]bool{}
+	var out []int
+	for _, p := range pushes {
+		if p.Phase != phaseConc || seen[p.Concurrency] {
+			continue
+		}
+		seen[p.Concurrency] = true
+		out = append(out, p.Concurrency)
+	}
+	sort.Ints(out)
+	return out
+}
+
+func pushesAt(pushes []pushResult, k int) []pushResult {
+	var out []pushResult
+	for _, p := range pushes {
+		if p.Phase == phaseConc && p.Concurrency == k {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
+// groupWindow is the span from the first push starting to the last one
+// finishing, which is the window the whole concurrent step occupied.
+func groupWindow(group []pushResult) (start, end time.Time) {
+	for i, p := range group {
+		if i == 0 || p.Start.Before(start) {
+			start = p.Start
+		}
+		if i == 0 || p.End.After(end) {
+			end = p.End
+		}
+	}
+	return start, end
+}
+
+// writeRepoList records every repository the run created, so the bucket can be
+// cleaned up afterwards. The harness never deletes anything itself.
+func writeRepoList(path string, org string, pushes []pushResult) error {
+	var b strings.Builder
+	for _, p := range pushes {
+		fmt.Fprintf(&b, "%s/%s\n", org, p.Repo)
+	}
+	if err := os.WriteFile(path, []byte(b.String()), 0o644); err != nil {
+		return fmt.Errorf("membench: can't write %s: %w", path, err)
+	}
+	return nil
+}
+
+// packBytes totals the pack files in a bare repository, which is how much data
+// each push actually moves.
+func packBytes(gitDir string) (int64, error) {
+	entries, err := os.ReadDir(filepath.Join(gitDir, "objects", "pack"))
+	if err != nil {
+		return 0, fmt.Errorf("membench: can't read pack directory: %w", err)
+	}
+
+	var total int64
+	for _, e := range entries {
+		if !strings.HasSuffix(e.Name(), ".pack") {
+			continue
+		}
+		info, err := e.Info()
+		if err != nil {
+			return 0, fmt.Errorf("membench: can't stat %s: %w", e.Name(), err)
+		}
+		total += info.Size()
+	}
+
+	return total, nil
+}
diff --git a/cmd/membench/report_test.go b/cmd/membench/report_test.go
new file mode 100644
index 0000000..1c6ac37
--- /dev/null
+++ b/cmd/membench/report_test.go
@@ -0,0 +1,313 @@
+package main
+
+import (
+	"regexp"
+	"testing"
+	"time"
+)
+
+// at builds a sample n seconds after a fixed origin, so window arithmetic in
+// these tests reads as plainly as it can.
+func at(origin time.Time, sec int, phase string, rss, heap uint64) sample {
+	return sample{
+		At:    origin.Add(time.Duration(sec) * time.Second),
+		Phase: phase,
+		Proc:  procStatus{VmRSS: rss, VmHWM: rss},
+		Go:    goMetrics{HeapInuse: heap},
+		GoOK:  true,
+	}
+}
+
+// atFailedScrape is a tick where /proc was read but the metrics endpoint did
+// not answer, which is what the last sample before shutdown looks like.
+func atFailedScrape(origin time.Time, sec int, phase string, rss uint64) sample {
+	s := at(origin, sec, phase, rss, 0)
+	s.Go = goMetrics{}
+	s.GoOK = false
+	return s
+}
+
+func TestPhaseTailSkipsFailedScrapes(t *testing.T) {
+	t.Parallel()
+
+	origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC)
+	samples := []sample{
+		at(origin, 0, phaseIdle, 700, 500),
+		atFailedScrape(origin, 1, phaseIdle, 690),
+	}
+
+	got, ok := phaseTail(samples, phaseIdle)
+	if !ok {
+		t.Fatal("wanted the last good sample, got none")
+	}
+	if got.Go.HeapInuse != 500 {
+		t.Logf("want: 500")
+		t.Logf("got:  %d", got.Go.HeapInuse)
+		t.Error("a failed scrape was reported as a settled heap")
+	}
+}
+
+func TestWindowPeakIgnoresFailedScrapeHeap(t *testing.T) {
+	t.Parallel()
+
+	origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC)
+	samples := []sample{
+		at(origin, 1, phaseSeq, 100, 60),
+		atFailedScrape(origin, 2, phaseSeq, 900),
+	}
+
+	rss, heap := windowPeak(samples, origin.Add(time.Second), origin.Add(2*time.Second), 0)
+	if rss != 900 {
+		t.Logf("want rss: 900")
+		t.Logf("got rss:  %d", rss)
+		t.Error("proc numbers should survive a failed scrape")
+	}
+	if heap != 60 {
+		t.Logf("want heap: 60")
+		t.Logf("got heap:  %d", heap)
+		t.Error("heap from a failed scrape leaked into the peak")
+	}
+}
+
+func TestWindowPeak(t *testing.T) {
+	t.Parallel()
+
+	origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC)
+	samples := []sample{
+		at(origin, 0, phaseBaseline, 10, 5),
+		at(origin, 1, phaseSeq, 40, 20),
+		at(origin, 2, phaseSeq, 90, 60),
+		at(origin, 3, phaseIdle, 30, 10),
+		at(origin, 9, phaseSeq, 999, 999),
+	}
+
+	for _, tt := range []struct {
+		name       string
+		start, end int
+		slack      time.Duration
+		wantRSS    uint64
+		wantHeap   uint64
+	}{
+		{name: "covers the push window", start: 1, end: 2, wantRSS: 90, wantHeap: 60},
+		{name: "slack pulls in the settling sample", start: 1, end: 2, slack: time.Second, wantRSS: 90, wantHeap: 60},
+		{name: "excludes a later unrelated spike", start: 0, end: 3, wantRSS: 90, wantHeap: 60},
+		{name: "empty window", start: 5, end: 6, wantRSS: 0, wantHeap: 0},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			start := origin.Add(time.Duration(tt.start) * time.Second)
+			end := origin.Add(time.Duration(tt.end) * time.Second)
+
+			rss, heap := windowPeak(samples, start, end, tt.slack)
+			if rss != tt.wantRSS || heap != tt.wantHeap {
+				t.Logf("want: rss=%d heap=%d", tt.wantRSS, tt.wantHeap)
+				t.Logf("got:  rss=%d heap=%d", rss, heap)
+				t.Error("wrong peak")
+			}
+		})
+	}
+}
+
+func TestRSSAt(t *testing.T) {
+	t.Parallel()
+
+	origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC)
+	samples := []sample{
+		at(origin, 1, phaseBaseline, 10, 5),
+		at(origin, 2, phaseSeq, 90, 60),
+		at(origin, 3, phaseIdle, 70, 40),
+	}
+
+	for _, tt := range []struct {
+		name string
+		sec  int
+		want uint64
+	}{
+		{name: "before the first sample", sec: 0, want: 0},
+		{name: "exactly on a sample", sec: 2, want: 90},
+		{name: "between samples takes the earlier one", sec: 2, want: 90},
+		{name: "after the last sample", sec: 99, want: 70},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got := rssAt(samples, origin.Add(time.Duration(tt.sec)*time.Second))
+			if got != tt.want {
+				t.Logf("want: %d", tt.want)
+				t.Logf("got:  %d", got)
+				t.Error("wrong reading")
+			}
+		})
+	}
+}
+
+func TestPhaseTail(t *testing.T) {
+	t.Parallel()
+
+	origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC)
+	samples := []sample{
+		at(origin, 0, phaseBaseline, 10, 5),
+		at(origin, 1, phaseBaseline, 12, 6),
+		at(origin, 2, phaseSeq, 90, 60),
+	}
+
+	for _, tt := range []struct {
+		name    string
+		phase   string
+		wantRSS uint64
+		wantOK  bool
+	}{
+		{name: "last baseline sample wins", phase: phaseBaseline, wantRSS: 12, wantOK: true},
+		{name: "only sample in the phase", phase: phaseSeq, wantRSS: 90, wantOK: true},
+		{name: "phase never happened", phase: phaseConc, wantOK: false},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, ok := phaseTail(samples, tt.phase)
+			if ok != tt.wantOK {
+				t.Logf("want ok: %v", tt.wantOK)
+				t.Logf("got ok:  %v", ok)
+				t.Fatal("wrong presence")
+			}
+			if ok && got.Proc.VmRSS != tt.wantRSS {
+				t.Logf("want: %d", tt.wantRSS)
+				t.Logf("got:  %d", got.Proc.VmRSS)
+				t.Error("wrong sample")
+			}
+		})
+	}
+}
+
+func TestConcurrencySteps(t *testing.T) {
+	t.Parallel()
+
+	for _, tt := range []struct {
+		name   string
+		pushes []pushResult
+		want   []int
+	}{
+		{
+			name: "distinct levels, ascending, sequential pushes ignored",
+			pushes: []pushResult{
+				{Phase: phaseSeq, Concurrency: 1},
+				{Phase: phaseConc, Concurrency: 4},
+				{Phase: phaseConc, Concurrency: 4},
+				{Phase: phaseConc, Concurrency: 1},
+				{Phase: phaseConc, Concurrency: 2},
+			},
+			want: []int{1, 2, 4},
+		},
+		{name: "no concurrency phase", pushes: []pushResult{{Phase: phaseSeq, Concurrency: 1}}},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got := concurrencySteps(tt.pushes)
+			if len(got) != len(tt.want) {
+				t.Logf("want: %v", tt.want)
+				t.Logf("got:  %v", got)
+				t.Fatal("wrong number of steps")
+			}
+			for i := range got {
+				if got[i] != tt.want[i] {
+					t.Logf("want: %v", tt.want)
+					t.Logf("got:  %v", got)
+					t.Fatal("wrong steps")
+				}
+			}
+		})
+	}
+}
+
+func TestGroupWindow(t *testing.T) {
+	t.Parallel()
+
+	origin := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC)
+	group := []pushResult{
+		{Start: origin.Add(2 * time.Second), End: origin.Add(5 * time.Second)},
+		{Start: origin, End: origin.Add(3 * time.Second)},
+		{Start: origin.Add(time.Second), End: origin.Add(9 * time.Second)},
+	}
+
+	start, end := groupWindow(group)
+	if !start.Equal(origin) {
+		t.Logf("want: %v", origin)
+		t.Logf("got:  %v", start)
+		t.Error("wrong window start")
+	}
+	if want := origin.Add(9 * time.Second); !end.Equal(want) {
+		t.Logf("want: %v", want)
+		t.Logf("got:  %v", end)
+		t.Error("wrong window end")
+	}
+}
+
+func TestParseSteps(t *testing.T) {
+	t.Parallel()
+
+	for _, tt := range []struct {
+		name    string
+		input   string
+		want    []int
+		wantErr bool
+	}{
+		{name: "the default sweep", input: "1,2,4,8", want: []int{1, 2, 4, 8}},
+		{name: "spaces and a trailing comma", input: " 1, 3 ,", want: []int{1, 3}},
+		{name: "empty disables the sweep", input: "  "},
+		{name: "not a number", input: "1,two", wantErr: true},
+		{name: "zero is not a concurrency", input: "0", wantErr: true},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := parseSteps(tt.input)
+			if tt.wantErr {
+				if err == nil {
+					t.Error("wanted an error, got none")
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+
+			if len(got) != len(tt.want) {
+				t.Logf("want: %v", tt.want)
+				t.Logf("got:  %v", got)
+				t.Fatal("wrong number of steps")
+			}
+			for i := range got {
+				if got[i] != tt.want[i] {
+					t.Logf("want: %v", tt.want)
+					t.Logf("got:  %v", got)
+					t.Fatal("wrong steps")
+				}
+			}
+		})
+	}
+}
+
+func TestNewUUID(t *testing.T) {
+	t.Parallel()
+
+	shape := regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$`)
+
+	seen := map[string]bool{}
+	for range 100 {
+		got, err := newUUID()
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if !shape.MatchString(got) {
+			t.Logf("want: a version 4 UUID")
+			t.Logf("got:  %s", got)
+			t.Fatal("wrong shape")
+		}
+		if seen[got] {
+			t.Fatalf("%s came back twice", got)
+		}
+		seen[got] = true
+	}
+}
diff --git a/cmd/membench/sampler.go b/cmd/membench/sampler.go
new file mode 100644
index 0000000..f93cde7
--- /dev/null
+++ b/cmd/membench/sampler.go
@@ -0,0 +1,224 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+)
+
+// sample is one row of samples.csv: everything the harness knew about the
+// daemon at one instant, tagged with what the daemon was being asked to do.
+type sample struct {
+	At    time.Time
+	Phase string
+	Repo  string
+	Proc  procStatus
+	Go    goMetrics
+	// GoOK is false when the /metrics scrape failed for this tick, which
+	// happens routinely while the daemon is shutting down. The proc numbers are
+	// still good, so the row is kept, but every Go figure in it is a zero value
+	// rather than a measurement and must not be read as one.
+	GoOK bool
+}
+
+// profileCapture records one pprof profile written to disk, so the report can
+// say which resident-memory reading each profile belongs to.
+type profileCapture struct {
+	Name   string
+	Path   string
+	At     time.Time
+	RSS    uint64
+	Reason string
+}
+
+// sampler polls the daemon's /proc entries and /metrics on a fixed tick, and
+// grabs a heap profile whenever resident memory sets a new high-water mark.
+//
+// The phase and repo labels are set by the driver as it walks the run, so every
+// row says what the daemon was doing when it was taken. That is what turns the
+// CSV from a wall of numbers into something you can attribute.
+type sampler struct {
+	pid        int
+	metricsURL string
+	pprofURL   string
+	outDir     string
+	interval   time.Duration
+	growth     float64
+	cooldown   time.Duration
+	client     *http.Client
+
+	mu       sync.Mutex
+	phase    string
+	repo     string
+	samples  []sample
+	profiles []profileCapture
+
+	peakRSS     uint64
+	lastCapture time.Time
+}
+
+func newSampler(pid int, metricsAddr, outDir string, interval, cooldown time.Duration, growth float64) *sampler {
+	base := "http://" + metricsAddr
+	return &sampler{
+		pid:        pid,
+		metricsURL: base + "/metrics",
+		pprofURL:   base + "/debug/pprof",
+		outDir:     outDir,
+		interval:   interval,
+		growth:     growth,
+		cooldown:   cooldown,
+		client:     &http.Client{Timeout: 30 * time.Second},
+	}
+}
+
+// label tells the sampler what the driver is about to do. Every sample taken
+// until the next call carries these two values.
+func (s *sampler) label(phase, repo string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.phase, s.repo = phase, repo
+}
+
+// run samples until the context is cancelled.
+func (s *sampler) run(ctx context.Context) {
+	t := time.NewTicker(s.interval)
+	defer t.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			s.collect(ctx)
+		}
+	}
+}
+
+// collect takes one sample. A failed read is logged and dropped rather than
+// aborting the run: a single missed tick costs one row, and the benchmark is
+// several minutes of pushes we do not want to throw away.
+func (s *sampler) collect(ctx context.Context) {
+	ps, err := readProcStatus(s.pid)
+	if err != nil {
+		slog.Debug("can't read proc status", "pid", s.pid, "err", err)
+		return
+	}
+
+	gm, err := s.fetchMetrics(ctx)
+	goOK := err == nil
+	if err != nil {
+		slog.Debug("can't read metrics", "url", s.metricsURL, "err", err)
+	}
+
+	s.mu.Lock()
+	now := time.Now()
+	s.samples = append(s.samples, sample{At: now, Phase: s.phase, Repo: s.repo, Proc: ps, Go: gm, GoOK: goOK})
+
+	capture := false
+	if float64(ps.VmRSS) > float64(s.peakRSS)*(1+s.growth) && now.Sub(s.lastCapture) > s.cooldown {
+		capture = true
+		s.lastCapture = now
+	}
+	if ps.VmRSS > s.peakRSS {
+		s.peakRSS = ps.VmRSS
+	}
+	phase, repo := s.phase, s.repo
+	s.mu.Unlock()
+
+	if !capture {
+		return
+	}
+
+	// gc=0 on purpose. Forcing a collection here would change the very number
+	// that triggered the capture; we want the heap as it actually was at the
+	// peak, sampling error and all.
+	name := fmt.Sprintf("heap-peak-%s-%d.pb.gz", now.Format("150405.000"), ps.VmRSS)
+	reason := fmt.Sprintf("new RSS high-water mark during %s %s", phase, repo)
+	if _, err := s.captureProfile(ctx, "heap", false, name, reason, ps.VmRSS); err != nil {
+		slog.Warn("can't capture peak heap profile", "err", err)
+	}
+}
+
+func (s *sampler) fetchMetrics(ctx context.Context) (goMetrics, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.metricsURL, nil)
+	if err != nil {
+		return goMetrics{}, fmt.Errorf("membench: can't build metrics request: %w", err)
+	}
+
+	resp, err := s.client.Do(req)
+	if err != nil {
+		return goMetrics{}, fmt.Errorf("membench: metrics request failed: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return goMetrics{}, fmt.Errorf("membench: %s returned %s", s.metricsURL, resp.Status)
+	}
+
+	return parseMetrics(resp.Body)
+}
+
+// captureProfile fetches one pprof profile and writes it into the run
+// directory. gc forces a collection before the heap is dumped: use it for the
+// baseline and the final capture, where a settled heap is what you want, and
+// never mid-push.
+func (s *sampler) captureProfile(ctx context.Context, kind string, gc bool, name, reason string, rss uint64) (profileCapture, error) {
+	url := fmt.Sprintf("%s/%s?gc=0", s.pprofURL, kind)
+	if gc {
+		url = fmt.Sprintf("%s/%s?gc=1", s.pprofURL, kind)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return profileCapture{}, fmt.Errorf("membench: can't build pprof request: %w", err)
+	}
+
+	resp, err := s.client.Do(req)
+	if err != nil {
+		return profileCapture{}, fmt.Errorf("membench: pprof request failed: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return profileCapture{}, fmt.Errorf("membench: %s returned %s", url, resp.Status)
+	}
+
+	path := filepath.Join(s.outDir, name)
+	f, err := os.Create(path)
+	if err != nil {
+		return profileCapture{}, fmt.Errorf("membench: can't create %s: %w", path, err)
+	}
+	defer f.Close()
+
+	if _, err := io.Copy(f, resp.Body); err != nil {
+		return profileCapture{}, fmt.Errorf("membench: can't write %s: %w", path, err)
+	}
+
+	pc := profileCapture{Name: name, Path: path, At: time.Now(), RSS: rss, Reason: reason}
+	s.mu.Lock()
+	s.profiles = append(s.profiles, pc)
+	s.mu.Unlock()
+
+	slog.Info("captured profile", "name", name, "reason", reason, "rss_bytes", rss)
+	return pc, nil
+}
+
+// snapshot returns copies of everything collected so far, so the report writer
+// never touches the sampler's state while it is still running.
+func (s *sampler) snapshot() ([]sample, []profileCapture) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	samples := make([]sample, len(s.samples))
+	copy(samples, s.samples)
+	profiles := make([]profileCapture, len(s.profiles))
+	copy(profiles, s.profiles)
+
+	return samples, profiles
+}
diff --git a/cmd/membench/uuid.go b/cmd/membench/uuid.go
new file mode 100644
index 0000000..aaa7df4
--- /dev/null
+++ b/cmd/membench/uuid.go
@@ -0,0 +1,21 @@
+package main
+
+import (
+	"crypto/rand"
+	"fmt"
+)
+
+// newUUID returns a random version 4 UUID. The harness only needs repository
+// names that never collide with an earlier run, so crypto/rand covers it and no
+// dependency is needed.
+func newUUID() (string, error) {
+	var b [16]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return "", fmt.Errorf("membench: can't read random bytes: %w", err)
+	}
+
+	b[6] = (b[6] & 0x0f) | 0x40 // version 4
+	b[8] = (b[8] & 0x3f) | 0x80 // variant 10
+
+	return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]), nil
+}
diff --git a/docs/plans/streaming-container-sink.md b/docs/plans/streaming-container-sink.md
new file mode 100644
index 0000000..1fd66cb
--- /dev/null
+++ b/docs/plans/streaming-container-sink.md
@@ -0,0 +1,258 @@
+# Replace the scratch go-git storage with a container sink
+
+**Status:** stage 1 of 4 implemented (`internal/storage/tigris/segmentstore.go`,
+9 tests, full suite green). Stages 2 to 4 not started.
+
+Landed separately while this was being written, both from the earlier specs:
+`b130832` caps the zstd encoder concurrency, `877cd16` bounds concurrent pushes.
+Those change the memory baseline this plan should be measured against, so
+re-baseline with `cmd/membench` before evaluating stage 3.
+
+Supersedes `var/no-materializing-deltas.md`, whose subject
+(`planObjects`) this deletes outright.
+
+## Context
+
+A push currently materializes objects three times over and round-trips the
+whole pack through a second on-disk git repository:
+
+```
+wire -> Scanner(tee) -> dotgit.PackWriter -> temp .pack
+                                          -> buildIndex -> Parser.Parse
+                                               (resolve every delta, write into scratch)
+packWriter.Close -> planObjects        (resolve every delta AGAIN, for hash/type/size)
+                 -> resolveDeltaBases  (8 more filesystem.Storage handles over scratch)
+                 -> payloadFor -> deltaForm -> Scanner.WriteObject
+                                          (materialize a THIRD time, to pick delta form)
+                               -> .bin/.cue
+```
+
+### Live heap
+
+Site 1 (`Parser.Parse` -> `processDelta` -> `ensureContent`) was 67 to 79% of
+live heap early in a push. Site 2 (`planObjects` -> `getMemoryObject` ->
+`ApplyDelta`) was 63 to 82% of live heap at the concurrent peak. Every push
+pays roughly 476 MiB of working set regardless of how warm the process is.
+
+### Allocation churn, which is the larger story
+
+Live heap understates site 2 badly, because the objects it materializes are
+freed the moment their metadata has been read and so barely register in
+`inuse_space`. Total allocation tells a different story. Measured over two
+runs (13.07 GB across 3 pushes, 89.33 GB across 20), a single push of a 48 MiB
+pack allocates:
+
+**~4.4 GB per push, about 93x the size of the pack.**
+
+By allocation site:
+
+| Site | Share | Per push |
+| --- | --- | --- |
+| `bytes.growSlice` | 47.9% | 2.09 GB |
+| `plumbing.(*MemoryObject).Write` | 31.3% | 1.36 GB |
+| everything else | 20.8% | ~0.9 GB |
+
+By objgit entry point, cumulative:
+
+```
+main.writePack                    11323 MB   84.6%
+  packWriter.Close                10637 MB   79.5%
+    packWriter.planObjects         8880 MB   66.4%
+```
+
+**`planObjects` alone is 66% of all allocation per push**, roughly 2.9 GB. It
+is the metadata walk that wants only hash, type, and size.
+
+The `MemoryObject.Write` traces also name a second contributor not visible in
+the live-heap profiles: `payloadFor` -> `deltaForm` (`packwriter.go:333`) ->
+`deltaFormOn` (`packwriter.go:339`) -> `Scanner.WriteObject`, which
+re-materializes objects a third time while deciding whether to store each one
+as a delta.
+
+**This inverts the earlier ranking.** Judged by live heap, site 1 looked like
+the bigger term and this plan looked like it addressed the smaller half. Judged
+by allocation, site 2 and the delta-form probe dominate, and this plan targets
+roughly two thirds of the churn. Churn is also what the GC is working against,
+which is why `GOGC=50` bought 24% of peak and 67% of settled heap for under 3%
+of wall clock.
+
+The scratch storage exists to provide three things: delta resolution, random
+access by hash, and after-the-fact delta-form lookup. Only the first is
+irreducible.
+
+## The load-bearing constraint
+
+**The temp `.pack` file cannot be removed.** `parser.go:106` reads:
+
+```go
+if p.scanner.seeker == nil {
+    p.lowMemoryMode = false
+}
+```
+
+and with low memory mode off, `scanner.go:526` retains `oh.content` for every
+object rather than re-inflating from disk on demand. Driving the parser
+directly off the network reader would therefore *increase* memory, not reduce
+it.
+
+So this plan keeps a seekable temp `.pack` and removes the `filesystem.Storage`
+wrapped around it. What goes away is the `.idx` build, the loose/packed object
+store, `planObjects`, `resolveDeltaBases`, and the eight worker storages.
+
+## Design
+
+`packSegment` is already the container this needs. `newPackSegment`
+(`packwriter.go:402`) opens an `os.CreateTemp` staging file, `add` appends
+payloads, `seg.recs` holds one `cueRecord` per object (hash, type, codec,
+offset, stored, raw, base), and `seal` checksums and uploads. That is an
+append-only blob plus a hash-to-offset side index, in the format the bucket
+already stores. Nothing needs inventing, and notably a tar file would be
+strictly worse: 512-byte header plus block padding per entry is over 11 MB of
+overhead on this repository's 21637 objects, and would still need transcoding
+to `.bin` before upload.
+
+The new flow:
+
+```
+wire -> Scanner(tee) -> temp .pack                       (unchanged, seekable)
+     -> pass 1: header-only scan of the temp .pack       (delta topology)
+     -> pass 2: Parser.Parse(storage: segmentStore)      (resolve once, stream out)
+                  -> RawObjectWriter(typ, sz) -> packSegment -> .bin/.cue
+```
+
+### `segmentStore`
+
+A new type implementing `storer.EncodedObjectStorer`, backed by the
+`packSegment` staging files and an in-memory index.
+
+- `RawObjectWriter(typ, sz)` returns a writer that appends through the existing
+  `writePayload` band logic, hashes as it streams (exactly as `stageWriter`
+  already does at `writer.go:33`), and on `Close` appends a `cueRecord` and
+  records `hash -> (segment, offset, stored, raw, codec)` in the index. Seals
+  and opens a new segment when the byte cap is reached.
+- `EncodedObject(typ, hash)` serves the parser's REF-delta base lookups
+  (`parser.go:342`) from the index plus `ReadAt` on the staging file, running
+  `decodeBody` for the codec.
+- `HasEncodedObject`, `EncodedObjectSize` read the index.
+- `IterEncodedObjects` walks the index.
+- `NewEncodedObject`, `SetEncodedObject` delegate to the existing `Storer`
+  implementations.
+- `AddAlternate` returns the same unsupported error the `Storer` already does.
+- `LowMemoryMode() bool` returns true, so the parser keeps its low-memory path.
+
+### Delta preservation
+
+This is the part that actually needs care, and it is why the header-only pass
+exists. The parser hands out fully resolved content only, and `RawObjectWriter`
+rejects delta types outright (`writer.go:35`), so a naive sink would store every
+object whole and inflate the bucket. `TestPackfileWriterKeepsDeltas` and
+`TestPackfileWriterDemotesDeltaAcrossSplit` encode the current contract and must
+keep passing.
+
+Because the temp `.pack` is seekable, a header-only pass over it is cheap: seek
+to each offset, parse the `ObjectHeader`, read `Type`, `OffsetReference`, and
+`Reference`, and never inflate content. That yields `offset -> base offset` for
+OFS deltas and `offset -> base hash` for REF deltas.
+
+To turn offsets into hashes, register a `packfile.Observer` via
+`WithScannerObservers`. `parser.go:146` and `:150` call
+`OnInflatedObjectHeader(oh.Type, oh.Size, oh.Offset)` and
+`OnInflatedObjectContent(oh.Hash, oh.Offset, oh.Crc32, nil)` for every object,
+where `oh.Type` and `oh.Size` are the *resolved* values and `oh.Offset` is the
+pack offset. Joining on offset gives `hash -> base hash`, which is precisely
+what `plannedObject.base` holds today, obtained without a single delta
+application.
+
+Note the observer's `content` argument is always `nil` on this path, so
+observers cannot carry object bytes. The bytes must come through
+`RawObjectWriter`; this is why both mechanisms are needed rather than either
+alone.
+
+### Ordering and the containment rule
+
+`Close` today plans the whole object set before writing, which lets `emit`
+place bases before deltas and lets `inSeg` guarantee a delta's base is in the
+same container. Streaming gives that up.
+
+Decision: **keep the two-pass shape, but make pass 1 cheap.** The header-only
+pass already walks every object; have it build the same `order` slice
+`planObjects` produces today, from headers alone. Pass 2 then drives the parser
+and writes in that order. This preserves `emit`, `inSeg`, and the split-demotion
+behaviour exactly, at the cost of the parser resolving in its own order rather
+than ours.
+
+If that ordering mismatch proves awkward, the fallback is to accept pack order
+and demote any delta whose base did not land in the same segment, which is
+already the documented behaviour when a split strands a delta.
+
+### Base read-back across a sealed segment
+
+If a late REF delta references an object in a segment that already sealed and
+uploaded, `EncodedObject` must still find it. Decision: **keep every staging
+file until the push completes**, then delete. Disk, not memory, and it matches
+the current lifetime of the scratch directory.
+
+## What this does not fix
+
+Site 1 remains. `ensureContent` (`parser.go:236`) inflates each delta into a
+pooled buffer and patches it into another, whatever the sink is. It is
+inherent: a git object's hash is over its resolved content, so every delta must
+be reconstructed at least once. The floor is one resolved object plus its base
+at a time.
+
+Expected recovery is site 2 in full, the delta-form probe in full, the `.idx`
+build, and the eight worker storages. Against the churn measurement that is
+about two thirds of the 4.4 GB a push allocates today. Against live heap it is
+the larger of the two terms at the concurrent peak but a less dramatic figure.
+
+**Gate the result on `alloc_space`, not `inuse_space`.** Site 2's objects are
+short-lived by construction, so an `inuse_space` comparison will understate
+this change by a wide margin. `cmd/membench` writes `allocs-final.pb.gz` for
+exactly this; divide its total by the number of pushes in the run.
+
+## Staging
+
+Each stage is independently testable and leaves the tree working.
+
+1. **DONE. `segmentStore` in isolation** (`internal/storage/tigris/segmentstore.go`).
+   The type, its index, and its `EncodedObjectStorer` methods, driven directly
+   by tests. The live write path still uses scratch, so nothing user-visible
+   changed. Notes from building it:
+   - `RawObjectWriter` stages each object to its own temp file while hashing,
+     then hands it to the existing `packSegment.add` as a `stagedObject`. The
+     staging hop is what keeps the object out of memory in one piece; reusing
+     `add` is what keeps the container format decisions in one place.
+   - Staging files stay open for the life of the store, so a base written into
+     a segment that filled early is still readable when its delta arrives.
+   - An index miss falls through to the backing `Storer`, which is the thin-pack
+     path; a miss in both surfaces as `plumbing.ErrObjectNotFound`.
+   - Duplicate objects within one pack are stored once.
+2. **The header-only pass.** Topology extraction and `order` construction from
+   the temp `.pack`, tested against the same fixtures `planObjects` is tested
+   with, asserting identical output.
+3. **Wire it up.** `packWriter.Close` drives `Parser.Parse` with `segmentStore`
+   and the observer instead of the scratch storage. This is the stage where the
+   existing pack tests are the gate.
+4. **Delete the dead code.** `planObjects`, `resolveDeltaBases`, `payloadFor`,
+   `deltaScanWorkers`, and the scratch `filesystem.Storage`.
+
+## Risks
+
+- **Format correctness is not caught by a memory benchmark.** A wrong `raw` or
+  `typ` in a cue record writes a repository that pushes fine and reads corrupt
+  later. Stage 2 must assert byte-identical `cueRecord` output against the
+  current implementation before stage 3 switches anything over.
+- **Thin packs.** `parser.go:342` falls back to `storage.EncodedObject` for a
+  REF delta base not in the pack. Against `segmentStore` that is an index miss;
+  it must then fall through to the real `Storer` and hence the bucket. The
+  current scratch storage has no alternates, so this path may not behave
+  identically today, and the difference needs establishing rather than assuming.
+- **Failure atomicity.** A failed push today discards a scratch directory and
+  the bucket is untouched. Streaming into segments means partial containers can
+  exist. `packSegment.discard()` exists; the invariant that nothing seals until
+  the parse completes must be explicit.
+- **SHA-256.** `writePack` threads `packfile.WithSHA256()` from the config
+  extension. Both new passes need the matching object ID size or they will
+  mis-parse every header.
+- **go-git v6 is alpha.** This couples objgit to `Parser`, `Scanner`,
+  `ObjectHeader`, and `Observer`. All are exported, none are stable.
diff --git a/docs/usage/memory-benchmark.md b/docs/usage/memory-benchmark.md
new file mode 100644
index 0000000..b725271
--- /dev/null
+++ b/docs/usage/memory-benchmark.md
@@ -0,0 +1,124 @@
+# Measuring push memory
+
+`cmd/membench` answers one question: how much memory does `objgitd` need while
+it is taking pushes, and does that memory come back afterwards.
+
+It starts a daemon it owns, pushes one real repository into many fresh ones,
+samples the daemon's resident set and Go heap throughout, and captures a pprof
+heap profile every time memory sets a new high-water mark. Nothing in `objgitd`
+had to change for this: `cmd/objgitd/main.go` already serves `net/http/pprof`
+and the default Prometheus registry (which carries `go_memstats_*` and
+`process_resident_memory_bytes`) on the metrics listener.
+
+## Running it
+
+```text
+go run ./cmd/membench -out ./bench-out
+```
+
+Run it from the objgit checkout. The harness finds `go.mod`, builds
+`./cmd/objgitd`, and starts the daemon with that directory as its working
+directory, so the daemon picks up `BUCKET` and the AWS credentials from `.env`
+exactly as it would normally.
+
+Pushes go to a real Tigris bucket. There is no local S3 fake in this repo, and a
+fake would hide the allocation behaviour of the Tigris storer, which is the
+thing being measured.
+
+## What a run does
+
+1. **Mirror.** `git clone --mirror` of the source repository into the run
+   directory, once. Every push reads from that copy, so the repository you
+   pointed at is never written to and concurrent pushes share an immutable
+   source.
+2. **Baseline.** Sample the idle daemon for `-baseline`, then capture
+   `heap-baseline.pb.gz` with a forced GC. Every later number is read against
+   this.
+3. **Sequential.** `-seq-pushes` pushes, one at a time, each to a fresh
+   repository, with `-idle-gap` between them. This isolates the per-push cost
+   and shows whether memory returns after a push.
+4. **Concurrency sweep.** For each K in `-conc-steps`, K pushes at once to K
+   fresh repositories, with an idle gap between steps. The rise per concurrent
+   push is the slope you size a machine with.
+5. **Settle.** One more idle gap, then `heap-final.pb.gz` (forced GC) and
+   `allocs-final.pb.gz`.
+
+Every repository is created fresh under a UUID, so no push is ever measured
+against a repository that already holds its objects. With the defaults that is
+20 repositories.
+
+## Repository names
+
+Repositories are `{-org}/{uuidv4}`, defaulting to `benchtest/<uuid>`. That is
+the exact `{orgID}/{repoName}` shape `internal/repofs.Parse` requires, and it
+means every benchmark repository shares one key prefix in the bucket.
+
+The harness never deletes anything. It writes `repos.txt` listing every
+repository it created, and cleanup is yours to run.
+
+## Output
+
+Each run gets its own timestamped directory under `-out`:
+
+| File                     | What it holds                                                        |
+| ------------------------ | -------------------------------------------------------------------- |
+| `report.md`              | The tables. Start here.                                              |
+| `samples.csv`            | One row per sample, tagged with phase and repository. Plot-ready.    |
+| `heap-baseline.pb.gz`    | Settled heap before any push.                                        |
+| `heap-peak-*.pb.gz`      | Heap at each new resident-set high-water mark.                       |
+| `heap-final.pb.gz`       | Settled heap after every push.                                       |
+| `allocs-final.pb.gz`     | Total allocation over the run.                                       |
+| `daemon.log`             | The daemon's own JSON log.                                           |
+| `repos.txt`              | Every repository created, for cleanup.                               |
+| `source.git`, `objgitd`  | The mirror and the binary under test.                                |
+
+The two commands that get the most out of a run:
+
+```sh
+# What the pushes left behind.
+go tool pprof -http=: -base <run>/heap-baseline.pb.gz <run>/heap-final.pb.gz
+
+# What was on the heap at the worst moment.
+go tool pprof -http=: <run>/heap-peak-<largest>.pb.gz
+```
+
+## Reading the numbers
+
+- **`VmHWM` is the peak**, not `VmRSS`. It is the kernel's own high-water mark,
+  so it cannot be missed between two samples the way a spike in `VmRSS` can.
+- **Resident set is not the heap.** Go returns freed pages to the kernel lazily,
+  so `VmRSS` staying high after a push is not by itself a leak. The column that
+  settles it is `heap_inuse` in the CSV, and the baseline-diffed heap profile.
+  If `heap_inuse` also stays high while the daemon is idle, that is retention.
+- **Peak profiles are taken with `gc=0`.** Forcing a collection would change the
+  number that triggered the capture.
+- **Every figure is relative to `GOGC` and `GOMEMLIMIT`.** The report records
+  both, because a peak-RSS number without them cannot be compared to anything.
+- **Wall clock includes network time** to the bucket and varies between runs.
+  Memory does not depend on it; throughput comparisons across runs do.
+
+## Flags worth knowing
+
+| Flag                    | Default              | Meaning                                                                  |
+| ----------------------- | -------------------- | ------------------------------------------------------------------------ |
+| `-repo`                 | `$HOME/Code/Xe/x`    | Repository to push. Mirror-cloned once, never written to.                |
+| `-org`                  | `benchtest`          | Org segment every benchmark repository is created under.                 |
+| `-seq-pushes`           | `5`                  | Sequential pushes, each to a fresh repository.                           |
+| `-conc-steps`           | `1,2,4,8`            | Concurrency levels to sweep. Empty skips the sweep.                      |
+| `-sample-interval`      | `250ms`              | How often `/proc` and `/metrics` are read.                               |
+| `-idle-gap`             | `5s`                 | Idle time between pushes, so memory can settle.                          |
+| `-window-slack`         | `0` (uses idle gap)  | How far outside a push's wall clock memory is still attributed to it.    |
+| `-peak-growth`          | `0.05`               | Fractional rise in resident set that triggers a heap capture.            |
+| `-peak-cooldown`        | `2s`                 | Minimum time between two peak captures.                                  |
+| `-daemon-binary`        | build it             | Prebuilt `objgitd` to test instead of building `./cmd/objgitd`.          |
+| `-daemon-allow-hooks`   | `false`              | Off so hook cost is not mistaken for push cost.                          |
+
+All flags take an environment fallback through `flagenv`, in UPPER_SNAKE. The
+daemon-facing flags are prefixed `-daemon-` so they do not collide with
+`objgitd`'s own `HTTP_BIND` and `METRICS_BIND`.
+
+A cheap smoke run, three repositories and well under a minute:
+
+```text
+go run ./cmd/membench -seq-pushes 1 -conc-steps 2 -baseline 3s -idle-gap 3s
+```
diff --git a/internal/storage/tigris/segmentstore.go b/internal/storage/tigris/segmentstore.go
new file mode 100644
index 0000000..f0c724e
--- /dev/null
+++ b/internal/storage/tigris/segmentstore.go
@@ -0,0 +1,368 @@
+package tigris
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+
+	"github.com/go-git/go-git/v6/plumbing"
+	"github.com/go-git/go-git/v6/plumbing/storer"
+)
+
+// segmentStore is a storer.EncodedObjectStorer that lands objects straight into
+// pack containers.
+//
+// It exists to replace the scratch go-git repository that PackfileWriter
+// currently decodes a push into. go-git's packfile.Parser writes every object
+// it resolves through RawObjectWriter (non-deltas from the scanner, resolved
+// deltas from storeOrCache), so handing the parser one of these means each
+// object is resolved exactly once and appended to a .bin as it arrives, instead
+// of being written to a scratch repository and read back out.
+//
+// Staging files are deliberately kept open for the whole push rather than
+// sealed as they fill: the parser reads bases back out through EncodedObject
+// (packfile/parser.go's REF-delta path), and a base can live in a segment that
+// filled long before the delta that needs it arrives.
+var _ storer.EncodedObjectStorer = (*segmentStore)(nil)
+
+type segmentStore struct {
+	s         *Storer
+	byteLimit int64
+
+	mu     sync.Mutex
+	open   *packSegment
+	filled []*packSegment
+	index  map[plumbing.Hash]stagedRef
+}
+
+// stagedRef locates one object inside one still-open staging file.
+type stagedRef struct {
+	seg *packSegment
+	rec cueRecord
+}
+
+func newSegmentStore(s *Storer) *segmentStore {
+	limit := s.maxPackBytes
+	if limit <= 0 {
+		limit = maxPackBytes
+	}
+	return &segmentStore{
+		s:         s,
+		byteLimit: limit,
+		index:     map[plumbing.Hash]stagedRef{},
+	}
+}
+
+// LowMemoryMode reports that the parser may drop object contents and re-inflate
+// them from the pack on demand. Answering true here is what keeps
+// packfile.Parser off the path that retains every object's bytes in memory;
+// see the seeker check in packfile/parser.go.
+func (ss *segmentStore) LowMemoryMode() bool { return true }
+
+// segments returns every staging file written so far, the open one last. The
+// caller seals and uploads them; this type never touches the bucket.
+func (ss *segmentStore) segments() []*packSegment {
+	ss.mu.Lock()
+	defer ss.mu.Unlock()
+
+	out := make([]*packSegment, 0, len(ss.filled)+1)
+	out = append(out, ss.filled...)
+	if ss.open != nil && len(ss.open.recs) > 0 {
+		out = append(out, ss.open)
+	}
+	return out
+}
+
+// discard throws away every staging file. Idempotent, and safe to defer for the
+// error path: sealing removes segments from this store's ownership first.
+func (ss *segmentStore) discard() {
+	ss.mu.Lock()
+	defer ss.mu.Unlock()
+
+	for _, seg := range ss.filled {
+		seg.discard()
+	}
+	if ss.open != nil {
+		ss.open.discard()
+	}
+	ss.filled, ss.open = nil, nil
+	ss.index = map[plumbing.Hash]stagedRef{}
+}
+
+// segmentFor returns a segment with room for an object of size sz, rotating the
+// open one out when it would overflow the byte cap.
+//
+// The cap is checked before the add and only when the segment already holds
+// something, which is the same rule packWriter.Close applies: an object larger
+// than the whole cap gets a container to itself, because it has to live
+// somewhere.
+func (ss *segmentStore) segmentFor(sz int64) (*packSegment, error) {
+	if ss.open != nil && len(ss.open.recs) > 0 && ss.open.offset+sz > ss.byteLimit {
+		ss.filled = append(ss.filled, ss.open)
+		ss.open = nil
+	}
+	if ss.open == nil {
+		seg, err := newPackSegment(ss.s)
+		if err != nil {
+			return nil, err
+		}
+		ss.open = seg
+	}
+	return ss.open, nil
+}
+
+// RawObjectWriter stages one object to its own temp file, hashing as it goes,
+// and appends it to a container on Close. The staging hop buys streaming: the
+// object never exists in memory in one piece, which is the whole point of
+// taking the parser's output this way rather than through SetEncodedObject.
+func (ss *segmentStore) RawObjectWriter(typ plumbing.ObjectType, sz int64) (io.WriteCloser, error) {
+	if typ == plumbing.OFSDeltaObject || typ == plumbing.REFDeltaObject {
+		return nil, plumbing.ErrInvalidType
+	}
+	if sz < 0 {
+		return nil, fmt.Errorf("tigris: negative object size %d", sz)
+	}
+
+	f, err := os.CreateTemp("", "objgit-tigris-seg-*")
+	if err != nil {
+		return nil, fmt.Errorf("tigris: create staging file: %w", err)
+	}
+
+	return &segmentWriter{
+		ss:     ss,
+		f:      f,
+		typ:    typ,
+		size:   sz,
+		hasher: plumbing.NewHasher(ss.s.of, typ, sz),
+	}, nil
+}
+
+type segmentWriter struct {
+	ss     *segmentStore
+	f      *os.File
+	typ    plumbing.ObjectType
+	size   int64
+	wrote  int64
+	hasher plumbing.Hasher
+	done   bool
+}
+
+func (w *segmentWriter) Write(p []byte) (int, error) {
+	if w.done {
+		return 0, errors.New("tigris: write on discarded segment writer")
+	}
+
+	n, err := io.MultiWriter(w.f, w.hasher).Write(p)
+	w.wrote += int64(n)
+	if err != nil {
+		// File and hasher may disagree from here on; the stream is poison.
+		w.discard()
+		return n, fmt.Errorf("tigris: stage write: %w", err)
+	}
+	if w.wrote > w.size {
+		w.discard()
+		return n, fmt.Errorf("tigris: wrote %d bytes beyond declared size %d", w.wrote-w.size, w.size)
+	}
+	return n, nil
+}
+
+func (w *segmentWriter) discard() {
+	if w.done {
+		return
+	}
+	w.done = true
+	w.f.Close()
+	os.Remove(w.f.Name())
+}
+
+func (w *segmentWriter) Close() error {
+	if w.done {
+		return nil
+	}
+	w.done = true
+	defer os.Remove(w.f.Name())
+
+	if w.wrote != w.size {
+		w.f.Close()
+		return fmt.Errorf("tigris: staged %d bytes but declared %d", w.wrote, w.size)
+	}
+	if err := w.f.Close(); err != nil {
+		return fmt.Errorf("tigris: close staging file: %w", err)
+	}
+
+	h := w.hasher.Sum()
+	obj := &stagedObject{path: w.f.Name(), typ: w.typ, size: w.size, hash: h}
+
+	w.ss.mu.Lock()
+	defer w.ss.mu.Unlock()
+
+	// An object already staged is not written twice. A pack may legitimately
+	// carry the same object more than once, and the second copy would only add
+	// bytes to the container under a hash the index already resolves.
+	if _, ok := w.ss.index[h]; ok {
+		return nil
+	}
+
+	seg, err := w.ss.segmentFor(w.size)
+	if err != nil {
+		return err
+	}
+
+	before := len(seg.recs)
+	if err := seg.add(storedObject{payload: obj, hash: h, typ: w.typ, raw: w.size}); err != nil {
+		return err
+	}
+	if len(seg.recs) != before+1 {
+		return fmt.Errorf("tigris: segment recorded %d records for one object", len(seg.recs)-before)
+	}
+
+	w.ss.index[h] = stagedRef{seg: seg, rec: seg.recs[before]}
+	return nil
+}
+
+// stagedObject is one object held in its own temp file, presented to
+// packSegment.add as an EncodedObject. Reader opens the file fresh each call so
+// writePayload can stream it, and rewind it for the probe path, without holding
+// the bytes.
+type stagedObject struct {
+	path string
+	typ  plumbing.ObjectType
+	size int64
+	hash plumbing.Hash
+}
+
+func (o *stagedObject) Hash() plumbing.Hash            { return o.hash }
+func (o *stagedObject) Type() plumbing.ObjectType      { return o.typ }
+func (o *stagedObject) SetType(t plumbing.ObjectType)  { o.typ = t }
+func (o *stagedObject) Size() int64                    { return o.size }
+func (o *stagedObject) SetSize(n int64)                { o.size = n }
+func (o *stagedObject) Reader() (io.ReadCloser, error) { return os.Open(o.path) }
+
+func (o *stagedObject) Writer() (io.WriteCloser, error) {
+	return nil, errors.New("tigris: staged object is read-only")
+}
+
+// EncodedObject serves the parser's base lookups out of the staging files. A
+// miss falls through to the backing Storer, which is what lets a thin pack
+// resolve a delta against an object the repository already holds.
+func (ss *segmentStore) EncodedObject(t plumbing.ObjectType, h plumbing.Hash) (plumbing.EncodedObject, error) {
+	ss.mu.Lock()
+	ref, ok := ss.index[h]
+	ss.mu.Unlock()
+
+	if !ok {
+		return ss.s.EncodedObject(t, h)
+	}
+	if t != plumbing.AnyObject && ref.rec.typ != t {
+		return nil, plumbing.ErrObjectNotFound
+	}
+
+	stored := make([]byte, ref.rec.stored)
+	if _, err := ref.seg.file.ReadAt(stored, ref.rec.offset); err != nil {
+		return nil, fmt.Errorf("tigris: read %s from staging: %w", h.String(), err)
+	}
+
+	plain, err := ss.s.decodePackedPayload(h, packEntry{
+		typ:    ref.rec.typ,
+		codec:  ref.rec.codec,
+		offset: ref.rec.offset,
+		stored: ref.rec.stored,
+		raw:    ref.rec.raw,
+	}, bytes.NewReader(stored))
+	if err != nil {
+		return nil, err
+	}
+
+	obj := plumbing.NewMemoryObject(ss.s.oh)
+	obj.SetType(ref.rec.typ)
+	obj.SetSize(ref.rec.raw)
+	if _, err := obj.Write(plain); err != nil {
+		return nil, fmt.Errorf("tigris: rebuild %s: %w", h.String(), err)
+	}
+	return obj, nil
+}
+
+func (ss *segmentStore) HasEncodedObject(h plumbing.Hash) error {
+	ss.mu.Lock()
+	_, ok := ss.index[h]
+	ss.mu.Unlock()
+
+	if ok {
+		return nil
+	}
+	return ss.s.HasEncodedObject(h)
+}
+
+func (ss *segmentStore) EncodedObjectSize(h plumbing.Hash) (int64, error) {
+	ss.mu.Lock()
+	ref, ok := ss.index[h]
+	ss.mu.Unlock()
+
+	if ok {
+		return ref.rec.raw, nil
+	}
+	return ss.s.EncodedObjectSize(h)
+}
+
+// NewEncodedObject and SetEncodedObject delegate: the parser reaches for
+// RawObjectWriter, so these exist to satisfy the interface and to keep any
+// caller that does use them behaving exactly as the Storer does.
+func (ss *segmentStore) NewEncodedObject() plumbing.EncodedObject { return ss.s.NewEncodedObject() }
+
+func (ss *segmentStore) SetEncodedObject(obj plumbing.EncodedObject) (plumbing.Hash, error) {
+	return ss.s.SetEncodedObject(obj)
+}
+
+func (ss *segmentStore) IterEncodedObjects(t plumbing.ObjectType) (storer.EncodedObjectIter, error) {
+	ss.mu.Lock()
+	hashes := make([]plumbing.Hash, 0, len(ss.index))
+	for h, ref := range ss.index {
+		if t == plumbing.AnyObject || ref.rec.typ == t {
+			hashes = append(hashes, h)
+		}
+	}
+	ss.mu.Unlock()
+
+	return &segmentIter{ss: ss, typ: t, hashes: hashes}, nil
+}
+
+func (ss *segmentStore) AddAlternate(remote string) error { return ss.s.AddAlternate(remote) }
+
+type segmentIter struct {
+	ss     *segmentStore
+	typ    plumbing.ObjectType
+	hashes []plumbing.Hash
+	i      int
+}
+
+func (it *segmentIter) Next() (plumbing.EncodedObject, error) {
+	if it.i >= len(it.hashes) {
+		return nil, io.EOF
+	}
+	h := it.hashes[it.i]
+	it.i++
+	return it.ss.EncodedObject(it.typ, h)
+}
+
+func (it *segmentIter) ForEach(fn func(plumbing.EncodedObject) error) error {
+	for {
+		obj, err := it.Next()
+		if errors.Is(err, io.EOF) {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		if err := fn(obj); err != nil {
+			if errors.Is(err, storer.ErrStop) {
+				return nil
+			}
+			return err
+		}
+	}
+}
+
+func (it *segmentIter) Close() { it.i = len(it.hashes) }
diff --git a/internal/storage/tigris/segmentstore_test.go b/internal/storage/tigris/segmentstore_test.go
new file mode 100644
index 0000000..61f8141
--- /dev/null
+++ b/internal/storage/tigris/segmentstore_test.go
@@ -0,0 +1,384 @@
+package tigris
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"testing"
+
+	"github.com/go-git/go-git/v6/plumbing"
+)
+
+// putRaw writes one object through the segment store the way packfile.Parser
+// does, and returns the hash it landed under.
+func putRaw(t *testing.T, ss *segmentStore, typ plumbing.ObjectType, body []byte) plumbing.Hash {
+	t.Helper()
+
+	w, err := ss.RawObjectWriter(typ, int64(len(body)))
+	if err != nil {
+		t.Fatalf("RawObjectWriter: %v", err)
+	}
+	if _, err := w.Write(body); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+
+	h := plumbing.NewHasher(ss.s.of, typ, int64(len(body)))
+	if _, err := h.Write(body); err != nil {
+		t.Fatalf("hash: %v", err)
+	}
+	return h.Sum()
+}
+
+func readBack(t *testing.T, ss *segmentStore, h plumbing.Hash) (plumbing.ObjectType, []byte) {
+	t.Helper()
+
+	obj, err := ss.EncodedObject(plumbing.AnyObject, h)
+	if err != nil {
+		t.Fatalf("EncodedObject(%s): %v", h, err)
+	}
+	rd, err := obj.Reader()
+	if err != nil {
+		t.Fatalf("reader: %v", err)
+	}
+	defer rd.Close()
+
+	got, err := io.ReadAll(rd)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	return obj.Type(), got
+}
+
+func TestSegmentStoreRoundTrip(t *testing.T) {
+	t.Parallel()
+
+	for _, tt := range []struct {
+		name string
+		typ  plumbing.ObjectType
+		body []byte
+	}{
+		{name: "tiny blob below the compression floor", typ: plumbing.BlobObject, body: []byte("hello")},
+		{name: "empty blob", typ: plumbing.BlobObject, body: []byte{}},
+		{name: "commit", typ: plumbing.CommitObject, body: []byte("tree deadbeef\n\nmessage\n")},
+		{name: "compressible blob above the floor", typ: plumbing.BlobObject, body: bytes.Repeat([]byte("abcd"), 4096)},
+		{name: "incompressible blob above the floor", typ: plumbing.BlobObject, body: pseudoRandom(8192)},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			ss := newSegmentStore(newTestStorer(t, newFakeS3(t)))
+			defer ss.discard()
+
+			h := putRaw(t, ss, tt.typ, tt.body)
+
+			gotTyp, gotBody := readBack(t, ss, h)
+			if gotTyp != tt.typ {
+				t.Logf("want: %v", tt.typ)
+				t.Logf("got:  %v", gotTyp)
+				t.Error("wrong type")
+			}
+			if !bytes.Equal(gotBody, tt.body) {
+				t.Logf("want %d bytes", len(tt.body))
+				t.Logf("got  %d bytes", len(gotBody))
+				t.Error("payload did not round-trip")
+			}
+
+			size, err := ss.EncodedObjectSize(h)
+			if err != nil {
+				t.Fatalf("EncodedObjectSize: %v", err)
+			}
+			if size != int64(len(tt.body)) {
+				t.Logf("want: %d", len(tt.body))
+				t.Logf("got:  %d", size)
+				t.Error("wrong recorded size")
+			}
+			if err := ss.HasEncodedObject(h); err != nil {
+				t.Errorf("HasEncodedObject: %v", err)
+			}
+		})
+	}
+}
+
+func TestSegmentStoreRejections(t *testing.T) {
+	t.Parallel()
+
+	for _, tt := range []struct {
+		name string
+		typ  plumbing.ObjectType
+		size int64
+		err  error
+	}{
+		{name: "ref delta", typ: plumbing.REFDeltaObject, size: 4, err: plumbing.ErrInvalidType},
+		{name: "ofs delta", typ: plumbing.OFSDeltaObject, size: 4, err: plumbing.ErrInvalidType},
+		{name: "negative size", typ: plumbing.BlobObject, size: -1},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			ss := newSegmentStore(newTestStorer(t, newFakeS3(t)))
+			defer ss.discard()
+
+			_, err := ss.RawObjectWriter(tt.typ, tt.size)
+			if err == nil {
+				t.Fatal("wanted an error, got none")
+			}
+			if tt.err != nil && !errors.Is(err, tt.err) {
+				t.Logf("want: %v", tt.err)
+				t.Logf("got:  %v", err)
+				t.Error("wrong error")
+			}
+		})
+	}
+}
+
+func TestSegmentStoreShortAndOverlongWrites(t *testing.T) {
+	t.Parallel()
+
+	for _, tt := range []struct {
+		name    string
+		declare int64
+		body    []byte
+		failsOn string
+	}{
+		{name: "fewer bytes than declared", declare: 10, body: []byte("abc"), failsOn: "close"},
+		{name: "more bytes than declared", declare: 3, body: []byte("abcdefgh"), failsOn: "write"},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			ss := newSegmentStore(newTestStorer(t, newFakeS3(t)))
+			defer ss.discard()
+
+			w, err := ss.RawObjectWriter(plumbing.BlobObject, tt.declare)
+			if err != nil {
+				t.Fatalf("RawObjectWriter: %v", err)
+			}
+
+			_, writeErr := w.Write(tt.body)
+			closeErr := w.Close()
+
+			switch tt.failsOn {
+			case "write":
+				if writeErr == nil {
+					t.Error("wanted the write to fail, got no error")
+				}
+			case "close":
+				if writeErr != nil {
+					t.Fatalf("unexpected write error: %v", writeErr)
+				}
+				if closeErr == nil {
+					t.Error("wanted the close to fail, got no error")
+				}
+			}
+
+			if got := len(ss.segments()); got != 0 {
+				t.Logf("want: 0 segments")
+				t.Logf("got:  %d", got)
+				t.Error("a rejected object still opened a container")
+			}
+		})
+	}
+}
+
+func TestSegmentStoreSplitsAtByteLimit(t *testing.T) {
+	t.Parallel()
+
+	// Four objects of 1 KiB against a 2 KiB cap: the third has to open a second
+	// container, and the cap is checked before the add, so no container ever
+	// exceeds it.
+	ss := newSegmentStore(newTestStorer(t, newFakeS3(t), withMaxPackBytes(2<<10)))
+	defer ss.discard()
+
+	var hashes []plumbing.Hash
+	for i := range 4 {
+		hashes = append(hashes, putRaw(t, ss, plumbing.BlobObject, pseudoRandomSeeded(1<<10, i)))
+	}
+
+	segs := ss.segments()
+	if len(segs) < 2 {
+		t.Logf("want: at least 2 segments")
+		t.Logf("got:  %d", len(segs))
+		t.Fatal("byte cap did not split the container")
+	}
+
+	total := 0
+	for _, seg := range segs {
+		total += len(seg.recs)
+		if seg.offset > 2<<10 && len(seg.recs) > 1 {
+			t.Errorf("segment holds %d bytes over a %d cap with %d records", seg.offset, 2<<10, len(seg.recs))
+		}
+	}
+	if total != 4 {
+		t.Logf("want: 4 records across all segments")
+		t.Logf("got:  %d", total)
+		t.Error("objects went missing across the split")
+	}
+
+	// Every object stays readable after its container filled, which is what
+	// lets the parser resolve a base written long before its delta arrives.
+	for i, h := range hashes {
+		if _, body := readBack(t, ss, h); len(body) != 1<<10 {
+			t.Errorf("object %d read back %d bytes, want %d", i, len(body), 1<<10)
+		}
+	}
+}
+
+func TestSegmentStoreDeduplicates(t *testing.T) {
+	t.Parallel()
+
+	ss := newSegmentStore(newTestStorer(t, newFakeS3(t)))
+	defer ss.discard()
+
+	body := []byte("the same object twice")
+	first := putRaw(t, ss, plumbing.BlobObject, body)
+	second := putRaw(t, ss, plumbing.BlobObject, body)
+
+	if first != second {
+		t.Fatal("the same bytes hashed differently")
+	}
+
+	segs := ss.segments()
+	if len(segs) != 1 {
+		t.Fatalf("want 1 segment, got %d", len(segs))
+	}
+	if got := len(segs[0].recs); got != 1 {
+		t.Logf("want: 1 record")
+		t.Logf("got:  %d", got)
+		t.Error("a duplicate object was stored twice")
+	}
+}
+
+func TestSegmentStoreMissFallsThroughToStorer(t *testing.T) {
+	t.Parallel()
+
+	ss := newSegmentStore(newTestStorer(t, newFakeS3(t)))
+	defer ss.discard()
+
+	// Nothing staged, so this can only be answered by the backing Storer. The
+	// object is not there either, so the fall-through must surface as a
+	// not-found rather than as a staging error.
+	absent := plumbing.NewHasher(ss.s.of, plumbing.BlobObject, 3).Sum()
+
+	_, err := ss.EncodedObject(plumbing.AnyObject, absent)
+	if err == nil {
+		t.Fatal("wanted an error for an object in neither place, got none")
+	}
+	if !errors.Is(err, plumbing.ErrObjectNotFound) {
+		t.Logf("want: %v", plumbing.ErrObjectNotFound)
+		t.Logf("got:  %v", err)
+		t.Error("wrong error from the fall-through path")
+	}
+}
+
+func TestSegmentStoreTypeFilter(t *testing.T) {
+	t.Parallel()
+
+	ss := newSegmentStore(newTestStorer(t, newFakeS3(t)))
+	defer ss.discard()
+
+	h := putRaw(t, ss, plumbing.BlobObject, []byte("a blob"))
+
+	if _, err := ss.EncodedObject(plumbing.BlobObject, h); err != nil {
+		t.Errorf("matching type should resolve: %v", err)
+	}
+	_, err := ss.EncodedObject(plumbing.CommitObject, h)
+	if !errors.Is(err, plumbing.ErrObjectNotFound) {
+		t.Logf("want: %v", plumbing.ErrObjectNotFound)
+		t.Logf("got:  %v", err)
+		t.Error("a type mismatch should read as not found")
+	}
+}
+
+func TestSegmentStoreIter(t *testing.T) {
+	t.Parallel()
+
+	ss := newSegmentStore(newTestStorer(t, newFakeS3(t)))
+	defer ss.discard()
+
+	want := map[plumbing.Hash]bool{}
+	for i := range 3 {
+		want[putRaw(t, ss, plumbing.BlobObject, []byte(fmt.Sprintf("blob %d", i)))] = true
+	}
+	commit := putRaw(t, ss, plumbing.CommitObject, []byte("a commit"))
+
+	iter, err := ss.IterEncodedObjects(plumbing.BlobObject)
+	if err != nil {
+		t.Fatalf("IterEncodedObjects: %v", err)
+	}
+	got := map[plumbing.Hash]bool{}
+	if err := iter.ForEach(func(o plumbing.EncodedObject) error {
+		got[o.Hash()] = true
+		return nil
+	}); err != nil {
+		t.Fatalf("ForEach: %v", err)
+	}
+
+	if len(got) != len(want) {
+		t.Logf("want: %d blobs", len(want))
+		t.Logf("got:  %d", len(got))
+		t.Error("wrong number of blobs")
+	}
+	for h := range want {
+		if !got[h] {
+			t.Errorf("blob %s missing from the iteration", h)
+		}
+	}
+	if got[commit] {
+		t.Error("the commit leaked into a blob-only iteration")
+	}
+}
+
+func TestSegmentStoreDiscardLeavesNothing(t *testing.T) {
+	t.Parallel()
+
+	ss := newSegmentStore(newTestStorer(t, newFakeS3(t), withMaxPackBytes(2<<10)))
+	for i := range 4 {
+		putRaw(t, ss, plumbing.BlobObject, pseudoRandomSeeded(1<<10, i))
+	}
+
+	paths := []string{}
+	for _, seg := range ss.segments() {
+		paths = append(paths, seg.path)
+	}
+	if len(paths) < 2 {
+		t.Fatalf("want at least 2 staging files, got %d", len(paths))
+	}
+
+	ss.discard()
+
+	for _, p := range paths {
+		if fileExists(p) {
+			t.Errorf("staging file %s survived discard", p)
+		}
+	}
+	if got := len(ss.segments()); got != 0 {
+		t.Errorf("want 0 segments after discard, got %d", got)
+	}
+}
+
+// pseudoRandom returns n bytes that zstd cannot shrink, so a test can reach the
+// raw-codec branch deliberately.
+func pseudoRandom(n int) []byte { return pseudoRandomSeeded(n, 0) }
+
+// fileExists reports whether a staging path is still on disk.
+func fileExists(path string) bool {
+	_, err := os.Stat(path)
+	return err == nil
+}
+
+func pseudoRandomSeeded(n, seed int) []byte {
+	out := make([]byte, n)
+	x := uint64(seed)*2862933555777941757 + 3037000493
+	for i := range out {
+		x ^= x << 13
+		x ^= x >> 7
+		x ^= x << 17
+		out[i] = byte(x)
+	}
+	return out
+}