initialHEADmain

22 files changed, 2947 insertions, 0 deletions

diff --git a/internal/cli/cmd.go b/internal/cli/cmd.go
new file mode 100644
index 0000000..698cf7c
--- /dev/null
+++ b/internal/cli/cmd.go
@@ -0,0 +1,73 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+
+	"github.com/spf13/pflag"
+
+	"tsne.dev/hopper/internal/log"
+)
+
+// Run executes the CLI and returns the process exit code.
+//
+// Note: Help paths will call exit directly.
+func Run(ctx context.Context, args []string) int {
+	main := newMainCommand()
+	err := runCommand(ctx, main, args)
+	if err != nil {
+		var exit exitError
+		if errors.As(err, &exit) {
+			return exit.code
+		}
+
+		var ev errorVerboser
+		if errors.As(err, &ev) {
+			log.Error("%s\n  %s", err.Error(), ev.ErrorVerbose())
+		} else {
+			log.Error(err.Error())
+		}
+		return 1
+	}
+	return 0
+}
+
+type command interface {
+	Usage() string
+	Run(ctx context.Context, args []string) error
+}
+
+func parseFlags(cmd command, flags *pflag.FlagSet, args []string) {
+	err := flags.Parse(args)
+	if err != nil {
+		if errors.Is(err, pflag.ErrHelp) {
+			log.Print(cmd.Usage())
+			os.Exit(0)
+		} else {
+			log.Error("%s\n\n%s", err.Error(), cmd.Usage())
+			os.Exit(2)
+		}
+	}
+}
+
+func runCommand(ctx context.Context, cmd command, args []string) error {
+	err := cmd.Run(ctx, args)
+	if err != nil {
+		var usageErr usageError
+		if errors.As(err, &usageErr) {
+			log.Error("%s\n\n%s", usageErr.msg, cmd.Usage())
+			os.Exit(2)
+		}
+		return err
+	}
+	return nil
+}
+
+func newFlagset(name string) *pflag.FlagSet {
+	fs := pflag.NewFlagSet(name, pflag.ContinueOnError)
+	fs.SetOutput(io.Discard)
+	fs.Usage = func() {}
+	return fs
+}
diff --git a/internal/cli/cmd_main.go b/internal/cli/cmd_main.go
new file mode 100644
index 0000000..3d4c3db
--- /dev/null
+++ b/internal/cli/cmd_main.go
@@ -0,0 +1,130 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/spf13/pflag"
+
+	"tsne.dev/hopper/internal/log"
+)
+
+var _ command = (*mainCommand)(nil)
+
+type mainCommand struct {
+	flags *pflag.FlagSet
+
+	zone          string
+	endpoint      string
+	accessKeyFile string
+	accessKey     string
+	verbose       bool
+	quiet         bool
+}
+
+func newMainCommand() *mainCommand {
+	cmd := &mainCommand{}
+	cmd.flags = newFlagset("hopper")
+	cmd.flags.SetInterspersed(false)
+	cmd.flags.StringVar(&cmd.zone,
+		"zone",
+		lookupEnv("BUNNY_ZONE", ""),
+		"Bunny storage zone name (required) (env: BUNNY_ZONE)",
+	)
+	cmd.flags.StringVar(&cmd.endpoint,
+		"endpoint",
+		lookupEnv("BUNNY_ENDPOINT", "storage.bunnycdn.com"),
+		"Bunny storage endpoint hostname (env: BUNNY_ENDPOINT)",
+	)
+	cmd.flags.StringVar(&cmd.accessKeyFile,
+		"access-key-file",
+		lookupEnv("BUNNY_ACCESS_KEY_FILE", ""),
+		"path to a file containing the storage access key (env: BUNNY_ACCESS_KEY_FILE)",
+	)
+	cmd.flags.BoolVar(
+		&cmd.verbose,
+		"verbose",
+		false,
+		"print more details",
+	)
+	cmd.flags.BoolVar(
+		&cmd.quiet,
+		"quiet",
+		false,
+		"print only the final summary and errors",
+	)
+	return cmd
+}
+
+func (cmd *mainCommand) Usage() string {
+	return "Usage:" +
+		"\n    hopper [global flags] ⟨command⟩ [command flags]" +
+		"\n" +
+		"\nCommands:" +
+		"\n    push    upload a release to Bunny Storage" +
+		"\n    prune   remove orphaned files from Bunny Storage" +
+		"\n" +
+		"\nGlobal flags:" +
+		"\n" + cmd.flags.FlagUsages()
+}
+
+func (cmd *mainCommand) Run(ctx context.Context, args []string) error {
+	parseFlags(cmd, cmd.flags, args)
+
+	log.SetVerbose(cmd.verbose)
+	log.SetQuiet(cmd.quiet)
+
+	if cmd.accessKeyFile == "" {
+		cmd.accessKey = os.Getenv("BUNNY_ACCESS_KEY")
+	}
+
+	switch {
+	case cmd.zone == "":
+		return usageError{"missing zone (provide --zone or BUNNY_ZONE)"}
+	case cmd.endpoint == "":
+		// --endpoint has a default, so an empty value here is always deliberate.
+		return usageError{"the endpoint must not be empty"}
+	case cmd.accessKey == "" && cmd.accessKeyFile == "":
+		return usageError{"missing access key (provide --access-key-file, BUNNY_ACCESS_KEY_FILE, or BUNNY_ACCESS_KEY)"}
+	}
+
+	subargs := cmd.flags.Args()
+	if len(subargs) == 0 || subargs[0] == "" {
+		return usageError{"missing subcommand"}
+	}
+
+	var sub command
+	switch subargs[0] {
+	case "push":
+		sub = newPushCommand(cmd)
+	case "prune":
+		sub = newPruneCommand(cmd)
+	default:
+		return usageError{fmt.Sprintf("unknown subcommand %q", subargs[0])}
+	}
+
+	return runCommand(ctx, sub, subargs[1:])
+}
+
+func (cmd *mainCommand) lookupAccessKey() error {
+	if cmd.accessKey == "" {
+		accessKey, err := os.ReadFile(cmd.accessKeyFile)
+		if err != nil {
+			return fmt.Errorf("read access key file: %w", err)
+		}
+		cmd.accessKey = strings.TrimSpace(string(accessKey))
+	}
+	if cmd.accessKey == "" {
+		return usageError{"empty access key"}
+	}
+	return nil
+}
+
+func lookupEnv(env string, defaultValue string) string {
+	if val := os.Getenv(env); val != "" {
+		return val
+	}
+	return defaultValue
+}
diff --git a/internal/cli/cmd_main_test.go b/internal/cli/cmd_main_test.go
new file mode 100644
index 0000000..d383970
--- /dev/null
+++ b/internal/cli/cmd_main_test.go
@@ -0,0 +1,126 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// clearGlobalEnv neutralizes the BUNNY_ env fallbacks so a test controls
+// resolution entirely through flags.
+func clearGlobalEnv(t *testing.T) {
+	t.Helper()
+	for _, k := range []string{"BUNNY_ZONE", "BUNNY_ENDPOINT", "BUNNY_ACCESS_KEY_FILE", "BUNNY_ACCESS_KEY"} {
+		t.Setenv(k, "")
+	}
+}
+
+func writeKeyFile(t *testing.T) string {
+	t.Helper()
+	p := filepath.Join(t.TempDir(), "key")
+	if err := os.WriteFile(p, []byte("secret\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return p
+}
+
+func runMain(args []string) error {
+	return newMainCommand().Run(context.Background(), args)
+}
+
+func TestMainMissingSubcommand(t *testing.T) {
+	clearGlobalEnv(t)
+	captureLogs(t)
+
+	var ue usageError
+	if err := runMain(nil); !errors.As(err, &ue) {
+		t.Fatalf("expected usageError, got %v", err)
+	}
+}
+
+func TestMainMissingZone(t *testing.T) {
+	clearGlobalEnv(t)
+	captureLogs(t)
+
+	var ue usageError
+	err := runMain([]string{"push"})
+	if !errors.As(err, &ue) || !strings.Contains(err.Error(), "zone") {
+		t.Fatalf("expected missing-zone usageError, got %v", err)
+	}
+}
+
+func TestMainMissingAccessKey(t *testing.T) {
+	clearGlobalEnv(t)
+	captureLogs(t)
+
+	var ue usageError
+	err := runMain([]string{"--zone", "z", "push"})
+	if !errors.As(err, &ue) || !strings.Contains(err.Error(), "access key") {
+		t.Fatalf("expected missing-access-key usageError, got %v", err)
+	}
+}
+
+func TestMainUnknownSubcommand(t *testing.T) {
+	clearGlobalEnv(t)
+	captureLogs(t)
+	kf := writeKeyFile(t)
+
+	var ue usageError
+	err := runMain([]string{"--zone", "z", "--access-key-file", kf, "frob"})
+	if !errors.As(err, &ue) || !strings.Contains(err.Error(), "unknown subcommand") {
+		t.Fatalf("expected unknown-subcommand usageError, got %v", err)
+	}
+}
+
+func TestMainDispatchesToPrune(t *testing.T) {
+	clearGlobalEnv(t)
+	captureLogs(t)
+	kf := writeKeyFile(t)
+
+	// A fully-resolved global set reaches dispatch; the missing archive surfaces a
+	// plain (non-usage) error, proving parse + validation + dispatch worked.
+	err := runMain([]string{"--zone", "z", "--access-key-file", kf, "prune", "no-such.tgz"})
+	if err == nil || !strings.Contains(err.Error(), "open archive") {
+		t.Fatalf("expected prune dispatch error, got %v", err)
+	}
+}
+
+func TestMainAccessKeyFromEnv(t *testing.T) {
+	clearGlobalEnv(t)
+	captureLogs(t)
+	t.Setenv("BUNNY_ACCESS_KEY", "literalkey")
+
+	// With no --access-key-file, the literal BUNNY_ACCESS_KEY resolves the key
+	// and resolution reaches prune.
+	err := runMain([]string{"--zone", "z", "prune", "no-such.tgz"})
+	if err == nil || !strings.Contains(err.Error(), "open archive") {
+		t.Fatalf("expected prune dispatch error (env access key resolved), got %v", err)
+	}
+}
+
+func TestMainZoneFromEnv(t *testing.T) {
+	clearGlobalEnv(t)
+	captureLogs(t)
+	t.Setenv("BUNNY_ZONE", "envzone")
+	kf := writeKeyFile(t)
+
+	// With zone supplied via env, resolution succeeds and reaches prune.
+	err := runMain([]string{"--access-key-file", kf, "prune", "no-such.tgz"})
+	if err == nil || !strings.Contains(err.Error(), "open archive") {
+		t.Fatalf("expected prune dispatch error (env zone resolved), got %v", err)
+	}
+}
+
+func TestLookupEnv(t *testing.T) {
+	t.Setenv("BUNNY_TEST_LOOKUP", "value")
+	if got := lookupEnv("BUNNY_TEST_LOOKUP", "fallback"); got != "value" {
+		t.Fatalf("set env: got %q", got)
+	}
+	t.Setenv("BUNNY_TEST_LOOKUP", "")
+	if got := lookupEnv("BUNNY_TEST_LOOKUP", "fallback"); got != "fallback" {
+		t.Fatalf("empty env should fall back: got %q", got)
+	}
+}
diff --git a/internal/cli/cmd_prune.go b/internal/cli/cmd_prune.go
new file mode 100644
index 0000000..14c3f0f
--- /dev/null
+++ b/internal/cli/cmd_prune.go
@@ -0,0 +1,186 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sync/atomic"
+	"time"
+
+	"github.com/spf13/pflag"
+
+	"tsne.dev/hopper/internal/log"
+	"tsne.dev/hopper/internal/source"
+)
+
+var _ command = (*pruneCommand)(nil)
+
+type pruneCommand struct {
+	flags *pflag.FlagSet
+	main  *mainCommand
+
+	concurrency uint
+	prefix      string
+	root        string
+	olderThan   duration
+}
+
+func newPruneCommand(main *mainCommand) *pruneCommand {
+	cmd := &pruneCommand{
+		flags: newFlagset("prune"),
+		main:  main,
+	}
+	cmd.flags.UintVar(&cmd.concurrency, "concurrency", 0, "number of concurrent transfers (0 = auto-detect based on CPUs)")
+	registerPathPrefixFlag(cmd.flags, &cmd.prefix, "prefix", "", "path prefix within the zone to target")
+	registerPathPrefixFlag(cmd.flags, &cmd.root, "root", "", "subtree of the source to consider live")
+	registerDurationFlag(cmd.flags, &cmd.olderThan, "older-than", "30d", "grace window, only orphans older than this are deleted")
+	registerDryRunFlag(cmd.flags)
+	return cmd
+}
+
+func (cmd *pruneCommand) Usage() string {
+	return "Usage:" +
+		"\n  hopper [global flags] prune [flags] ⟨source⟩" +
+		"\n" +
+		"\nFlags:" +
+		"\n" + cmd.flags.FlagUsages()
+}
+
+func (cmd *pruneCommand) Run(ctx context.Context, args []string) error {
+	parseFlags(cmd, cmd.flags, args)
+
+	if cmd.olderThan.isZero() {
+		return usageError{"grace window must not be zero"}
+	}
+
+	positionals := cmd.flags.Args()
+	if len(positionals) == 0 || positionals[0] == "" {
+		return usageError{"missing source"}
+	}
+	if len(positionals) > 1 {
+		return usageError{"only a single source is supported"}
+	}
+
+	if err := cmd.main.lookupAccessKey(); err != nil {
+		return err
+	}
+
+	api := newStorageAPI(cmd.main.endpoint, cmd.main.zone, cmd.main.accessKey)
+	return cmd.prune(ctx, api, positionals[0], time.Now())
+}
+
+func (cmd *pruneCommand) prune(ctx context.Context, api storageAPI, sourceArg string, now time.Time) error {
+	start := time.Now()
+
+	live, err := cmd.liveSet(sourceArg)
+	if err != nil {
+		return err
+	}
+	if len(live) == 0 {
+		return errors.New("empty source (refusing to prune). check --root, --prefix, or the archive contents")
+	}
+
+	entries, err := listRemoteEntries(ctx, api, cmd.prefix)
+	if err != nil {
+		return err
+	}
+
+	cutoff := cmd.olderThan.cutoff(now)
+	candidates := make([]*remoteEntry, 0, len(entries)/4)
+	for i := range entries {
+		e := &entries[i]
+		if !e.isDir && !live[e.path] {
+			if e.lastChanged.IsZero() {
+				log.Warn("could not determine last modified date for %s (skipping)", e.path)
+			} else if e.lastChanged.Before(cutoff) {
+				candidates = append(candidates, e)
+			}
+		}
+	}
+
+	// Each delete is independent, so the pool never fails fast. A successful
+	// delete tombstones its entry.
+	pool := newWorkerPool(ctx, cmd.concurrency, false)
+	freed := int64(0)
+	for _, e := range candidates {
+		action := fmt.Sprintf("deleting %s (%s)", e.path, humanSize(e.size))
+		pool.submit(action, func(ctx context.Context) error {
+			if err := api.Delete(ctx, e.path); err != nil {
+				return fileError{path: e.path, err: err}
+			}
+			e.deleted = true
+			atomic.AddInt64(&freed, int64(e.size))
+			return nil
+		})
+	}
+
+	deleted, errs := pool.wait()
+
+	var cmdErr error
+	if reportFailures(ctx, "failed to delete files", errs) {
+		cmdErr = exitError{1}
+	}
+
+	cmd.removeEmptyDirs(ctx, api, entries)
+
+	failed := len(errs)
+	logSummary("deleted: %d"+
+		"\nfailed:  %d"+
+		"\nfreed:   %s"+
+		"\nelapsed: %s",
+		deleted, failed, humanSize64(freed), time.Since(start).Round(time.Millisecond),
+	)
+
+	return cmdErr
+}
+
+// removeEmptyDirs deletes directories left with no surviving children after the
+// file deletes and returns the number successfully removed. It runs as a
+// best-effort second pass: failures are reported but never propagated to the
+// exit code.
+func (cmd *pruneCommand) removeEmptyDirs(ctx context.Context, api storageAPI, entries []remoteEntry) {
+	dirs := emptyDirs(entries)
+	if len(dirs) == 0 {
+		return
+	}
+
+	pool := newWorkerPool(ctx, cmd.concurrency, false)
+	for _, dir := range dirs {
+		action := fmt.Sprintf("removing empty directory %s", dir.path)
+		pool.submit(action, func(ctx context.Context) error {
+			err := api.DeleteDir(ctx, dir.path)
+			if err != nil {
+				return fileError{path: dir.path, err: err}
+			}
+			return nil
+		})
+	}
+
+	_, errs := pool.wait()
+	if len(errs) > 0 {
+		_ = reportFailures(ctx, "failed to remove empty directories", errs)
+	}
+}
+
+// liveSet walks the source once and returns the set of prefixed storage keys in
+// the current release. It iterates entries only and never reads content.
+func (cmd *pruneCommand) liveSet(sourceArg string) (map[string]bool, error) {
+	src, err := source.NewTarGzSource(sourceArg, cmd.root)
+	if err != nil {
+		return nil, err
+	}
+	defer src.Close()
+
+	live := map[string]bool{}
+	for {
+		entry, err := src.Next()
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				return live, nil
+			}
+			return nil, err
+		}
+		live[prefixedPath(cmd.prefix, entry.Path)] = true
+	}
+}
diff --git a/internal/cli/cmd_prune_test.go b/internal/cli/cmd_prune_test.go
new file mode 100644
index 0000000..e3b9398
--- /dev/null
+++ b/internal/cli/cmd_prune_test.go
@@ -0,0 +1,244 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"sort"
+	"strings"
+	"testing"
+	"time"
+
+	"tsne.dev/hopper/internal/bunny"
+)
+
+func runPrune(source string, api storageAPI, older duration, now time.Time) error {
+	cmd := &pruneCommand{concurrency: 8, olderThan: older}
+	return cmd.prune(context.Background(), api, source, now)
+}
+
+func days(n int) duration { return duration{days: n} }
+
+func sortedDeletes(f *fakeAPI) []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	out := append([]string(nil), f.deletes...)
+	sort.Strings(out)
+	return out
+}
+
+// TestPruneSelectsCandidates covers the core §10 correctness properties: an
+// orphan older than the cutoff is deleted; a file still in the release is kept
+// even with an old LastChanged; an orphan newer than the cutoff is kept.
+func TestPruneSelectsCandidates(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+		{name: "assets/live.js", body: "x"},
+	})
+	now := time.Date(2024, time.June, 1, 0, 0, 0, 0, time.UTC)
+	old := now.AddDate(0, 0, -100)
+	recent := now.AddDate(0, 0, -5)
+
+	fc := newFakeAPI()
+	fc.lists[""] = []bunny.Object{
+		{Path: "index.html", LastChanged: old}, // live, old -> kept
+		{Path: "assets", IsDirectory: true},
+		{Path: "orphan-old.js", Size: 3, LastChanged: old}, // orphan, old -> deleted
+		{Path: "orphan-new.js", LastChanged: recent},       // orphan, new -> kept
+	}
+	fc.lists["assets"] = []bunny.Object{
+		{Path: "assets/live.js", LastChanged: old},          // live asset, old -> kept
+		{Path: "assets/gone.js", Size: 4, LastChanged: old}, // orphan, old -> deleted
+	}
+	captureLogs(t)
+
+	if err := runPrune(arc, fc, days(30), now); err != nil {
+		t.Fatalf("prune: %v", err)
+	}
+
+	got := sortedDeletes(fc)
+	want := []string{"assets/gone.js", "orphan-old.js"}
+	if strings.Join(got, ",") != strings.Join(want, ",") {
+		t.Fatalf("deletes = %v, want %v", got, want)
+	}
+}
+
+func TestPruneAppliesPrefix(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{{name: "app.js", body: "x"}})
+	now := time.Date(2024, time.June, 1, 0, 0, 0, 0, time.UTC)
+	old := now.AddDate(0, 0, -100)
+
+	fc := newFakeAPI()
+	fc.lists["foo"] = []bunny.Object{
+		{Path: "foo/app.js", LastChanged: old},   // live (prefixed) -> kept
+		{Path: "foo/stale.js", LastChanged: old}, // orphan -> deleted
+	}
+	captureLogs(t)
+
+	cmd := &pruneCommand{concurrency: 8, prefix: "foo", olderThan: days(30)}
+	if err := cmd.prune(context.Background(), fc, arc, now); err != nil {
+		t.Fatalf("prune: %v", err)
+	}
+	if got := sortedDeletes(fc); len(got) != 1 || got[0] != "foo/stale.js" {
+		t.Fatalf("deletes = %v, want [foo/stale.js]", got)
+	}
+}
+
+func TestPruneEmptyLiveSetIsFatal(t *testing.T) {
+	arc := makeArchive(t, nil)
+	fc := newFakeAPI()
+	// A List failure would be reported if reached; the empty-guard must fire
+	// first, so no Bunny call happens.
+	fc.listErr = errors.New("should not be called")
+	captureLogs(t)
+
+	err := runPrune(arc, fc, days(30), time.Now())
+	if err == nil || !strings.Contains(err.Error(), "empty source") {
+		t.Fatalf("expected empty-live-set error, got %v", err)
+	}
+	if len(fc.deletes) != 0 {
+		t.Fatalf("expected no deletes, got %v", fc.deletes)
+	}
+}
+
+func TestPruneListFailureIsFatal(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{{name: "a.js", body: "x"}})
+	fc := newFakeAPI()
+	fc.listErr = errors.New("boom")
+	captureLogs(t)
+
+	if err := runPrune(arc, fc, days(30), time.Now()); err == nil {
+		t.Fatal("expected error")
+	}
+	if len(fc.deletes) != 0 {
+		t.Fatalf("expected no deletes, got %v", fc.deletes)
+	}
+}
+
+func TestPruneDrainAndReport(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{{name: "keep.js", body: "x"}})
+	now := time.Date(2024, time.June, 1, 0, 0, 0, 0, time.UTC)
+	old := now.AddDate(0, 0, -100)
+
+	fc := newFakeAPI()
+	fc.lists[""] = []bunny.Object{
+		{Path: "keep.js", LastChanged: old},
+		{Path: "a.js", LastChanged: old},
+		{Path: "b.js", LastChanged: old},
+		{Path: "c.js", LastChanged: old},
+	}
+	fc.failErr = errors.New("boom")
+	fc.failWhen = func(path string) bool { return path == "b.js" }
+	_, errb := captureLogs(t)
+
+	err := runPrune(arc, fc, days(30), now)
+	var exit exitError
+	if !errors.As(err, &exit) {
+		t.Fatalf("expected exitError (non-zero exit) on delete failure, got %v", err)
+	}
+	// One failed delete must not stop the others: all three orphans are attempted.
+	if got := sortedDeletes(fc); strings.Join(got, ",") != "a.js,b.js,c.js" {
+		t.Fatalf("all candidates must be attempted, got %v", got)
+	}
+	if !strings.Contains(errb.String(), "b.js") {
+		t.Fatalf("failure report must name b.js, got %v", errb.String())
+	}
+}
+
+// TestPruneRemovesEmptyDirsBottomUp covers §10 step 5: a directory whose only
+// file is deleted is removed, and its now-empty parent is removed too.
+func TestPruneRemovesEmptyDirsBottomUp(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{{name: "index.html", body: "<html>"}})
+	now := time.Date(2024, time.June, 1, 0, 0, 0, 0, time.UTC)
+	old := now.AddDate(0, 0, -100)
+
+	fc := newFakeAPI()
+	fc.lists[""] = []bunny.Object{
+		{Path: "index.html", LastChanged: old},
+		{Path: "a", IsDirectory: true},
+	}
+	fc.lists["a"] = []bunny.Object{{Path: "a/b", IsDirectory: true}}
+	fc.lists["a/b"] = []bunny.Object{{Path: "a/b/gone.js", Size: 3, LastChanged: old}}
+	captureLogs(t)
+
+	if err := runPrune(arc, fc, days(30), now); err != nil {
+		t.Fatalf("prune: %v", err)
+	}
+
+	got := sortedDeletes(fc)
+	want := []string{"a", "a/b/gone.js"}
+	if strings.Join(got, ",") != strings.Join(want, ",") {
+		t.Fatalf("deletes = %v, want %v", got, want)
+	}
+}
+
+// TestPruneKeepsDirWithFailedChild covers §10 step 5: a directory whose file
+// delete FAILED must not be removed, because the node survives.
+func TestPruneKeepsDirWithFailedChild(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{{name: "index.html", body: "<html>"}})
+	now := time.Date(2024, time.June, 1, 0, 0, 0, 0, time.UTC)
+	old := now.AddDate(0, 0, -100)
+
+	fc := newFakeAPI()
+	fc.lists[""] = []bunny.Object{
+		{Path: "index.html", LastChanged: old},
+		{Path: "a", IsDirectory: true},
+	}
+	fc.lists["a"] = []bunny.Object{{Path: "a/gone.js", Size: 3, LastChanged: old}}
+	fc.failWhen = func(path string) bool { return path == "a/gone.js" }
+	fc.failErr = errors.New("boom")
+	captureLogs(t)
+
+	if err := runPrune(arc, fc, days(30), now); err == nil {
+		t.Fatal("expected non-zero exit on file-delete failure")
+	}
+	for _, d := range fc.deletes {
+		if d == "a" {
+			t.Fatalf("dir with failed child must not be removed, got %v", fc.deletes)
+		}
+	}
+}
+
+// TestPruneEmptyDirFailureNonFatal covers §10 step 6: an empty-dir removal
+// failure is warned and counted but never flips the exit code.
+func TestPruneEmptyDirFailureNonFatal(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{{name: "index.html", body: "<html>"}})
+	now := time.Date(2024, time.June, 1, 0, 0, 0, 0, time.UTC)
+	old := now.AddDate(0, 0, -100)
+
+	fc := newFakeAPI()
+	fc.lists[""] = []bunny.Object{
+		{Path: "index.html", LastChanged: old},
+		{Path: "a", IsDirectory: true},
+	}
+	fc.lists["a"] = []bunny.Object{{Path: "a/gone.js", Size: 3, LastChanged: old}}
+	fc.failWhen = func(path string) bool { return path == "a" }
+	fc.failErr = errors.New("boom")
+	_, errb := captureLogs(t)
+
+	if err := runPrune(arc, fc, days(30), now); err != nil {
+		t.Fatalf("empty-dir failure must be non-fatal, got %v", err)
+	}
+	if !strings.Contains(errb.String(), "Error:") {
+		t.Fatalf("expected error on empty-dir failure, got %q", errb.String())
+	}
+}
+
+func TestPruneDryRunDeletesNothing(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{{name: "keep.js", body: "x"}})
+	now := time.Date(2024, time.June, 1, 0, 0, 0, 0, time.UTC)
+	old := now.AddDate(0, 0, -100)
+
+	fc := newFakeAPI()
+	fc.lists[""] = []bunny.Object{
+		{Path: "keep.js", LastChanged: old},
+		{Path: "stale.js", LastChanged: old},
+	}
+	dryRun = true
+
+	if err := runPrune(arc, dryRunStorageAPI{fc}, days(30), now); err != nil {
+		t.Fatalf("prune: %v", err)
+	}
+	if len(fc.deletes) != 0 {
+		t.Fatalf("dry-run must delete nothing, got %v", fc.deletes)
+	}
+}
diff --git a/internal/cli/cmd_push.go b/internal/cli/cmd_push.go
new file mode 100644
index 0000000..c58f75c
--- /dev/null
+++ b/internal/cli/cmd_push.go
@@ -0,0 +1,266 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/spf13/pflag"
+
+	"tsne.dev/hopper/internal/bunny"
+	"tsne.dev/hopper/internal/log"
+	"tsne.dev/hopper/internal/source"
+)
+
+// pruneHintFloor and pruneHintMultiple gate the post-push prune hint: the zone
+// must hold at least floor files and more than multiple times the current
+// release before stale assets are worth flagging. multiple covers the current
+// release plus five retained rollbacks (six releases' worth of hashed assets).
+const (
+	pruneHintFloor    = 1000
+	pruneHintMultiple = 6
+)
+
+var _ command = (*pushCommand)(nil)
+
+type pushCommand struct {
+	flags *pflag.FlagSet
+	main  *mainCommand
+
+	concurrency uint
+	prefix      string
+	root        string
+}
+
+func newPushCommand(main *mainCommand) *pushCommand {
+	cmd := &pushCommand{
+		flags: newFlagset("push"),
+		main:  main,
+	}
+	cmd.flags.UintVar(&cmd.concurrency, "concurrency", 0, "number of concurrent transfers (0 = auto-detect based on CPUs)")
+	registerPathPrefixFlag(cmd.flags, &cmd.prefix, "prefix", "", "path prefix within the zone to target")
+	registerPathPrefixFlag(cmd.flags, &cmd.root, "root", "", "subtree of the source to upload")
+	registerDryRunFlag(cmd.flags)
+	return cmd
+}
+
+func (cmd *pushCommand) Usage() string {
+	return "Usage:" +
+		"\n  hopper [global flags] push [flags] ⟨source⟩" +
+		"\n" +
+		"\nFlags:" +
+		"\n" + cmd.flags.FlagUsages()
+}
+
+func (cmd *pushCommand) Run(ctx context.Context, args []string) error {
+	parseFlags(cmd, cmd.flags, args)
+
+	positionals := cmd.flags.Args()
+	if len(positionals) == 0 || positionals[0] == "" {
+		return usageError{"missing source"}
+	}
+	if len(positionals) > 1 {
+		return usageError{"only a single source is supported"}
+	}
+	source := positionals[0]
+
+	if err := cmd.main.lookupAccessKey(); err != nil {
+		return err
+	}
+
+	api := newStorageAPI(cmd.main.endpoint, cmd.main.zone, cmd.main.accessKey)
+	return cmd.push(ctx, api, source)
+}
+
+func (cmd *pushCommand) push(ctx context.Context, api storageAPI, sourceArg string) error {
+	src, err := source.NewTarGzSource(sourceArg, cmd.root)
+	if err != nil {
+		return err
+	}
+	defer src.Close()
+
+	// Drain the complete remote listing before any writes. A List failure is
+	// fatal here, so push never proceeds on a partial picture.
+	// Index the remote listing by storage key so each source entry's skip /
+	// conflict decision is an O(1) lookup rather than a linear scan per file
+	// (the walk itself is O(remote)). A file and a directory can never share an
+	// exact key in object storage, so keying by path is unambiguous.
+	start := time.Now()
+
+	remoteByPath := make(map[string]bunny.Object, 128)
+	remoteFiles := 0
+	it := api.List(ctx, cmd.prefix)
+	for it.Next() {
+		o := it.Object()
+		if !o.IsDirectory {
+			remoteFiles++
+		}
+		remoteByPath[o.Path] = o
+	}
+	if err := it.Err(); err != nil {
+		return err
+	}
+
+	skipped := 0
+
+	// stashedMutables holds mutable entries (e.g. *.html files) read during the
+	// asset phase but not uploaded until all immutable assets are successfully
+	// uploaded. The stash is in-memory. A release normally has only a handful of
+	// mutable files only.
+	stashedMutables := make([]mutableEntry, 0, 8)
+
+	// Upload every immutable asset via the pool and stash mutables. The single
+	// reader below buffers each entry's bytes and hands them to a fail-fast worker
+	// pool. The first asset that fails to upload cancels the pool.
+	assetsPool := newWorkerPool(ctx, cmd.concurrency, true)
+	var readErr error
+	for !assetsPool.cancelled() {
+		entry, err := src.Next()
+		if err != nil {
+			if !errors.Is(err, io.EOF) {
+				readErr = err
+			}
+			break
+		}
+
+		// Decide against the remote listing: a directory at this exact path is a
+		// hard conflict; a matching-size file is skipped; any other case (absent,
+		// size mismatch, 0/missing) falls through to re-upload.
+		key := prefixedPath(cmd.prefix, entry.Path)
+		remoteObj, remoteExists := remoteByPath[key]
+		if remoteExists && remoteObj.IsDirectory {
+			readErr = fmt.Errorf("remote path %q is a directory, release has a file there", key)
+			break
+		}
+
+		if entry.IsMutable {
+			data, err := entry.ReadContent(ctx)
+			if err != nil {
+				readErr = fmt.Errorf("read %q: %w", entry.Path, err)
+				break
+			}
+			stashedMutables = append(stashedMutables, mutableEntry{path: key, data: data})
+			continue
+		}
+
+		if remoteExists && remoteObj.Size == entry.Size {
+			logAction("skipping %s (exists)", key)
+			skipped++
+			continue
+		}
+
+		data, err := entry.ReadContent(ctx)
+		if err != nil {
+			readErr = fmt.Errorf("read %q: %w", entry.Path, err)
+			break
+		}
+
+		action := fmt.Sprintf("uploading %s (%s)", key, humanSize(len(data)))
+		if !assetsPool.submit(action, uploadTask(api, key, data)) {
+			break
+		}
+	}
+
+	// A reader error is fatal and abandons all in-flight uploads.
+	if readErr != nil {
+		assetsPool.cancel(errors.New("upload cancelled"))
+	}
+
+	assetsUploaded, errs := assetsPool.wait()
+	reported := reportFailures(ctx, "failed to upload files", errs)
+	if readErr != nil {
+		return readErr
+	} else if reported {
+		return exitError{1}
+	}
+
+	// Upload all stashed mutable files. This batch is neither atomic nor ordered.
+	// A partial mutable-phase failure can leave an entry point live while a mutable
+	// file it references is not yet uploaded. Mitigation: We report an error and
+	// a re-run is idempotent.
+	mutablesPool := newWorkerPool(ctx, cmd.concurrency, false)
+	for _, m := range stashedMutables {
+		action := fmt.Sprintf("uploading %s (%s)", m.path, humanSize(len(m.data)))
+		if !mutablesPool.submit(action, uploadTask(api, m.path, m.data)) {
+			break // the parent context was cancelled
+		}
+	}
+
+	// Commit phase (HTML) is drain-and-report: every stashed mutable is attempted
+	// and all failures are aggregated. Unlike the pre-commit asset phase, uploads
+	// here genuinely happened, so we print a summary with counts even on partial
+	// failure — but the failure list goes out first, keeping the summary last.
+	mutablesUploaded, mutErrs := mutablesPool.wait()
+	mutablesFailed := len(mutErrs)
+
+	sourceFiles := assetsUploaded + mutablesUploaded + skipped + mutablesFailed
+	uploaded := assetsUploaded + mutablesUploaded
+
+	if sourceFiles == 0 {
+		log.Warn("source contains no files to upload (pushed nothing). check --root, --prefix, or the archive contents")
+	}
+
+	var cmdErr error
+	if reportFailures(ctx, "failed to upload files", mutErrs) {
+		cmdErr = exitError{1}
+	}
+
+	summary := "uploaded: %d" +
+		"\nskipped:  %d" +
+		"\nremote:   %d" +
+		"\nfailed:   %d" +
+		"\nelapsed:  %s"
+	if shouldHintPrune(sourceFiles, remoteFiles) {
+		summary += "\n\nHint: There are many stale files on the zone. Consider running hopper prune."
+	}
+	logSummary(
+		summary,
+		uploaded, skipped, remoteFiles, mutablesFailed, time.Since(start).Round(time.Millisecond),
+	)
+
+	return cmdErr
+}
+
+func shouldHintPrune(sourceFiles, remoteFiles int) bool {
+	return sourceFiles > 0 &&
+		remoteFiles >= pruneHintFloor &&
+		remoteFiles > pruneHintMultiple*sourceFiles
+}
+
+// uploadTask builds a pool task that PUTs data and, on failure, tags the error
+// with the path so the report can name the file.
+func uploadTask(api storageAPI, path string, data []byte) func(context.Context) error {
+	return func(ctx context.Context) error {
+		if err := api.Upload(ctx, path, data); err != nil {
+			return fileError{path: path, err: err}
+		}
+		return nil
+	}
+}
+
+// mutableEntry is a mutable file (current rule: *.html) read during the asset
+// phase and held in memory until the commit point. data is owned by this
+// struct.
+type mutableEntry struct {
+	path string
+	data []byte
+}
+
+func humanSize(n int) string {
+	return humanSize64(int64(n))
+}
+
+func humanSize64(n int64) string {
+	const unit = 1024
+	if n < unit {
+		return fmt.Sprintf("%d B", n)
+	}
+	div, exp := int64(unit), 0
+	for x := n / unit; x >= unit; x /= unit {
+		div *= unit
+		exp++
+	}
+	return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp])
+}
diff --git a/internal/cli/cmd_push_test.go b/internal/cli/cmd_push_test.go
new file mode 100644
index 0000000..9a199b2
--- /dev/null
+++ b/internal/cli/cmd_push_test.go
@@ -0,0 +1,633 @@
+package cli
+
+import (
+	"archive/tar"
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+
+	"tsne.dev/hopper/internal/bunny"
+	"tsne.dev/hopper/internal/log"
+)
+
+// runPush drives the push flow against a fake storage API for the given archive,
+// using the default concurrency.
+func runPush(source string, api storageAPI) error {
+	return runPushN(source, api, 8)
+}
+
+func runPushN(source string, api storageAPI, concurrency uint) error {
+	cmd := &pushCommand{concurrency: concurrency}
+	return cmd.push(context.Background(), api, source)
+}
+
+func TestPushAppliesPrefix(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+		{name: "assets/app.abc.js", body: "console.log(1)"},
+		{name: "assets/skip.js", body: "same"},
+	})
+	fc := newFakeAPI()
+	// The remote walk starts under the prefix and returns already-prefixed keys.
+	fc.lists["foo"] = []bunny.Object{{Path: "foo/assets", IsDirectory: true}}
+	fc.lists["foo/assets"] = []bunny.Object{
+		{Path: "foo/assets/skip.js", Size: int(len("same"))},
+	}
+	out, _ := captureLogs(t)
+
+	cmd := &pushCommand{concurrency: 8, prefix: "foo"}
+	if err := cmd.push(context.Background(), fc, arc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	got := fc.paths()
+	want := []string{"foo/assets/app.abc.js", "foo/index.html"}
+	if strings.Join(got, ",") != strings.Join(want, ",") {
+		t.Fatalf("paths = %v, want %v", got, want)
+	}
+	if !strings.Contains(out.String(), "skipping foo/assets/skip.js (exists)") {
+		t.Fatalf("skip-check must compare prefixed key: %q", out.String())
+	}
+}
+
+func TestPushSelectsRootAndStrips(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "README.md", body: "ignored"},
+		{name: "app/index.html", body: "<html>"},
+		{name: "app/assets/x.js", body: "x"},
+		{name: "other/y.js", body: "ignored"},
+	})
+	fc := newFakeAPI()
+	captureLogs(t)
+
+	cmd := &pushCommand{concurrency: 8, root: "app"}
+	if err := cmd.push(context.Background(), fc, arc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	got := fc.paths()
+	want := []string{"assets/x.js", "index.html"}
+	if strings.Join(got, ",") != strings.Join(want, ",") {
+		t.Fatalf("paths = %v, want %v", got, want)
+	}
+}
+
+func TestPushSelectsNestedRoot(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "dist/public/index.html", body: "<html>"},
+		{name: "dist/public/assets/x.js", body: "x"},
+		{name: "dist/private/secret", body: "ignored"},
+	})
+	fc := newFakeAPI()
+	captureLogs(t)
+
+	cmd := &pushCommand{concurrency: 8, root: "dist/public"}
+	if err := cmd.push(context.Background(), fc, arc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	got := fc.paths()
+	want := []string{"assets/x.js", "index.html"}
+	if strings.Join(got, ",") != strings.Join(want, ",") {
+		t.Fatalf("paths = %v, want %v", got, want)
+	}
+}
+
+func TestPushRootWithPrefix(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "app/assets/x.js", body: "x"},
+	})
+	fc := newFakeAPI()
+	captureLogs(t)
+
+	cmd := &pushCommand{concurrency: 8, root: "app", prefix: "foo"}
+	if err := cmd.push(context.Background(), fc, arc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	got := fc.paths()
+	want := []string{"foo/assets/x.js"}
+	if strings.Join(got, ",") != strings.Join(want, ",") {
+		t.Fatalf("paths = %v, want %v", got, want)
+	}
+}
+
+func TestPushRootMatchesNothing(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "app/index.html", body: "<html>"},
+	})
+	fc := newFakeAPI()
+	out, _ := captureLogs(t)
+
+	cmd := &pushCommand{concurrency: 8, root: "nope"}
+	if err := cmd.push(context.Background(), fc, arc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	if got := fc.paths(); len(got) != 0 {
+		t.Fatalf("paths = %v, want none", got)
+	}
+	if !strings.Contains(out.String(), "uploaded: 0") {
+		t.Fatalf("summary missing: %q", out.String())
+	}
+}
+
+func TestPushUploadsRegularFiles(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+		{name: "assets/app.abc.js", body: "console.log(1)"},
+		{name: "dir/", typeflag: tar.TypeDir},
+	})
+	fc := newFakeAPI()
+	out, _ := captureLogs(t)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	got := fc.paths()
+	want := []string{"assets/app.abc.js", "index.html"}
+	if strings.Join(got, ",") != strings.Join(want, ",") {
+		t.Fatalf("paths = %v, want %v", got, want)
+	}
+	if string(fc.puts["index.html"]) != "<html>" {
+		t.Fatalf("bad body for index.html: %q", fc.puts["index.html"])
+	}
+	if string(fc.puts["assets/app.abc.js"]) != "console.log(1)" {
+		t.Fatalf("bad body for app.abc.js: %q", fc.puts["assets/app.abc.js"])
+	}
+	if !strings.Contains(out.String(), "uploaded: 2") {
+		t.Fatalf("summary missing: %q", out.String())
+	}
+}
+
+func TestPushUploadsHTMLAfterAllAssets(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+		{name: "a.js", body: "x"},
+		{name: "nested/page.HTML", body: "<p>"}, // case-insensitive .html
+		{name: "b.css", body: "y"},
+	})
+	fc := newFakeAPI()
+	captureLogs(t)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	// Every HTML Put must happen after the last asset Put.
+	lastAsset := -1
+	firstHTML := len(fc.order)
+	for i, p := range fc.order {
+		if strings.HasSuffix(strings.ToLower(p), ".html") {
+			if i < firstHTML {
+				firstHTML = i
+			}
+		} else if i > lastAsset {
+			lastAsset = i
+		}
+	}
+	if firstHTML <= lastAsset {
+		t.Fatalf("HTML uploaded before an asset; order = %v", fc.order)
+	}
+}
+
+func TestPushDoesNotUploadHTMLWhenAssetFails(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+		{name: "a.js", body: "x"},
+	})
+	fc := newFakeAPI()
+	fc.failOn = "a.js"
+	captureLogs(t)
+
+	if err := runPush(arc, fc); err == nil {
+		t.Fatal("expected error")
+	}
+	for _, p := range fc.order {
+		if strings.HasSuffix(strings.ToLower(p), ".html") {
+			t.Fatalf("HTML must not be uploaded when an asset fails; order = %v", fc.order)
+		}
+	}
+}
+
+func TestPushSkipsSymlinkWithWarning(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "a.js", body: "x"},
+		{name: "link", typeflag: tar.TypeSymlink, linkname: "a.js"},
+	})
+	fc := newFakeAPI()
+	_, errb := captureLogs(t)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	if got := fc.paths(); len(got) != 1 || got[0] != "a.js" {
+		t.Fatalf("paths = %v", got)
+	}
+	if !strings.Contains(errb.String(), "Warning") {
+		t.Fatalf("expected warning on stderr, got %q", errb.String())
+	}
+}
+
+func TestPushFailsOnUploadError(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "a.js", body: "x"},
+		{name: "b.js", body: "y"},
+	})
+	fc := newFakeAPI()
+	fc.failOn = "a.js"
+	captureLogs(t)
+
+	if err := runPush(arc, fc); err == nil {
+		t.Fatal("expected error")
+	}
+}
+
+func TestPushSkipsUnchanged(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+		{name: "assets/app.abc.js", body: "console.log(1)"},
+	})
+	fc := newFakeAPI()
+	fc.lists[""] = []bunny.Object{
+		{Path: "index.html", Size: int(len("<html>"))},
+		{Path: "assets", IsDirectory: true},
+	}
+	fc.lists["assets"] = []bunny.Object{
+		{Path: "assets/app.abc.js", Size: int(len("console.log(1)"))},
+	}
+	out, _ := captureLogs(t)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	// The asset is skipped (matching size); the *.html is the commit point and is
+	// always uploaded even when an identically-sized copy exists remotely.
+	if got := fc.paths(); len(got) != 1 || got[0] != "index.html" {
+		t.Fatalf("expected only index.html uploaded, got %v", got)
+	}
+	s := out.String()
+	if !strings.Contains(s, "skipping assets/app.abc.js (exists)") {
+		t.Fatalf("missing skip line: %q", s)
+	}
+	if strings.Contains(s, "skipping index.html") {
+		t.Fatalf("index.html must never be skipped: %q", s)
+	}
+	if !strings.Contains(s, "uploaded: 1") || !strings.Contains(s, "skipped:  1") {
+		t.Fatalf("summary missing: %q", s)
+	}
+}
+
+func TestPushReuploadsOnSizeMismatch(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "a.js", body: "hello"}, // 5 bytes
+		{name: "b.js", body: "world"}, // present, matching size -> skip
+	})
+	fc := newFakeAPI()
+	fc.lists[""] = []bunny.Object{
+		{Path: "a.js", Size: 1}, // size mismatch -> upload
+		{Path: "b.js", Size: int(len("world"))},
+	}
+	captureLogs(t)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	if got := fc.paths(); len(got) != 1 || got[0] != "a.js" {
+		t.Fatalf("paths = %v, want [a.js]", got)
+	}
+}
+
+func TestPushErrorsOnDirectoryFileCollision(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "data", body: ""}, // 0-byte asset colliding with a remote dir
+	})
+	fc := newFakeAPI()
+	// Remote has a directory named "data"; a file at the same path is a hard
+	// conflict (and a 0-size dir must never silently shadow a 0-byte file).
+	fc.lists[""] = []bunny.Object{{Path: "data", IsDirectory: true}}
+	captureLogs(t)
+
+	if err := runPush(arc, fc); err == nil {
+		t.Fatal("expected error on directory/file collision")
+	}
+	if got := fc.paths(); len(got) != 0 {
+		t.Fatalf("expected no uploads, got %v", got)
+	}
+}
+
+func TestPushAbortsOnListFailure(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{{name: "a.js", body: "x"}})
+	fc := newFakeAPI()
+	fc.listErr = errors.New("boom")
+	captureLogs(t)
+
+	if err := runPush(arc, fc); err == nil {
+		t.Fatal("expected error")
+	}
+	if got := fc.paths(); len(got) != 0 {
+		t.Fatalf("expected no uploads before abort, got %v", got)
+	}
+}
+
+func TestPushRunRejectsBadSourceArgs(t *testing.T) {
+	captureLogs(t)
+	cmd := newPushCommand(&mainCommand{})
+
+	var ue usageError
+	if err := cmd.Run(context.Background(), nil); !errors.As(err, &ue) {
+		t.Fatalf("missing source: expected usageError, got %v", err)
+	}
+	if err := cmd.Run(context.Background(), []string{"a.tgz", "b.tgz"}); !errors.As(err, &ue) {
+		t.Fatalf("multiple sources: expected usageError, got %v", err)
+	}
+}
+
+func TestPushRejectsNegativeConcurrencyAtParse(t *testing.T) {
+	captureLogs(t)
+	// A negative value is not a valid uint, so pflag rejects it at parse time.
+	cmd := newPushCommand(&mainCommand{})
+	if err := cmd.flags.Parse([]string{"--concurrency", "-1", "a.tgz"}); err == nil {
+		t.Fatal("--concurrency -1: expected a parse error, got nil")
+	}
+}
+
+func TestPushRejectsEscapingPrefix(t *testing.T) {
+	captureLogs(t)
+	cmd := newPushCommand(&mainCommand{})
+	var ue usageError
+	if err := cmd.flags.Parse([]string{"--prefix", "../x", "a.tgz"}); !errors.As(err, &ue) {
+		t.Fatalf("--prefix ../x: expected usageError, got %v", err)
+	}
+}
+
+func TestPushBoundsInFlightUploads(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "a.js", body: "1"},
+		{name: "b.js", body: "2"},
+		{name: "c.js", body: "3"},
+		{name: "d.js", body: "4"},
+		{name: "e.js", body: "5"},
+	})
+	const limit = 2
+	gate := &gatingAPI{fakeAPI: newFakeAPI(), release: make(chan struct{})}
+	captureLogs(t)
+
+	done := make(chan error, 1)
+	go func() { done <- runPushN(arc, gate, limit) }()
+
+	// The reader blocks on the semaphore once the pool is saturated, so no more
+	// uploads can begin until one drains; release them so the run finishes.
+	for i := 0; i < 5; i++ {
+		gate.release <- struct{}{}
+	}
+	if err := <-done; err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	if peak := gate.peak(); peak > limit {
+		t.Fatalf("peak in-flight = %d, want <= %d", peak, limit)
+	}
+}
+
+func TestPushAssetFailureReportsOneFileAndSkipsHTML(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+		{name: "a.js", body: "1"},
+		{name: "b.js", body: "2"},
+	})
+	fc := newFakeAPI()
+	// Every asset fails: fail-fast cancels the pool after the first genuine
+	// failure. All genuine failures that landed before cancellation are
+	// aggregated (§8.3); cancellation collateral (context.Canceled) is filtered
+	// out (see TestPushFailFastFiltersCancellationCollateral). HTML is never
+	// written.
+	fc.failAll = true
+	captureLogs(t)
+
+	err := runPushN(arc, fc, 8)
+	var ee exitError
+	if !errors.As(err, &ee) {
+		t.Fatalf("expected exitError, got %T", err)
+	}
+	for _, p := range fc.order {
+		if strings.HasSuffix(strings.ToLower(p), ".html") {
+			t.Fatalf("HTML must not be written when an asset fails; order = %v", fc.order)
+		}
+	}
+}
+
+// TestPushFailFastFiltersCancellationCollateral verifies that the fail-fast
+// report names ONLY the genuine upload failure, not the in-flight uploads that
+// were abandoned by the cancellation. A context.Canceled is a consequence of the
+// stop decision, not an independent file failure, and is swallowed uniformly so
+// push (fail-fast) and a user SIGINT behave identically (§8.3). The collateral
+// errors WRAP context.Canceled.
+func TestPushFailFastFiltersCancellationCollateral(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "a.js", body: "1"},
+		{name: "b.js", body: "2"},
+		{name: "c.js", body: "3"},
+		{name: "index.html", body: "<html>"},
+	})
+	api := &ctxCancelAPI{
+		fakeAPI:     newFakeAPI(),
+		triggerPath: "a.js",
+		triggerErr:  fileError{path: "a.js", err: errors.New("boom")},
+	}
+	_, errb := captureLogs(t)
+
+	// Concurrency covers all three assets so b.js/c.js are genuinely in-flight
+	// (blocked) when a.js triggers the cancellation.
+	err := runPushN(arc, api, 3)
+	var ee exitError
+	if !errors.As(err, &ee) {
+		t.Fatalf("expected exitError, got %T (%v)", err, err)
+	}
+	report := errb.String()
+	if !strings.Contains(report, "a.js") || !strings.Contains(report, "boom") {
+		t.Fatalf("report must name the genuine failure a.js: %q", report)
+	}
+	if strings.Contains(report, "b.js") || strings.Contains(report, "c.js") {
+		t.Fatalf("report must NOT name cancellation collateral b.js/c.js: %q", report)
+	}
+	for _, p := range api.order {
+		if strings.HasSuffix(strings.ToLower(p), ".html") {
+			t.Fatalf("HTML must not be written when an asset fails; order = %v", api.order)
+		}
+	}
+}
+
+func TestPushHTMLFailuresAreAllReported(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+		{name: "about.html", body: "<html>"},
+		{name: "a.js", body: "1"},
+	})
+	fc := newFakeAPI()
+	// Assets succeed; both HTML commit uploads fail. The drain phase reports
+	// every failure under the friendly headline.
+	fc.failErr = errors.New("boom")
+	fc.failWhen = func(path string) bool { return strings.HasSuffix(path, ".html") }
+	out, errb := captureLogs(t)
+
+	err := runPushN(arc, fc, 8)
+	var exit exitError
+	if !errors.As(err, &exit) {
+		t.Fatalf("expected exitError, got %v", err)
+	}
+	// The drain report lists every HTML failure on stderr...
+	if !strings.Contains(errb.String(), "index.html") || !strings.Contains(errb.String(), "about.html") {
+		t.Fatalf("drain report must list every HTML failure: %q", errb.String())
+	}
+	// ...and the summary is still printed (last) on stdout.
+	if !strings.Contains(out.String(), "failed:") {
+		t.Fatalf("summary must be printed after a commit-phase failure: %q", out.String())
+	}
+}
+
+func TestShouldHintPrune(t *testing.T) {
+	cases := []struct {
+		name        string
+		sourceFiles int
+		remoteFiles int
+		want        bool
+	}{
+		{"below floor", 1, 999, false},
+		{"at floor within ratio", 1000, 6000, false},
+		{"at exact multiple", 100, 600, false},
+		{"one over multiple but below floor", 10, 61, false},
+		{"one over multiple and at floor", 200, 1201, true},
+		{"empty source", 0, 100000, false},
+	}
+	for _, tc := range cases {
+		if got := shouldHintPrune(tc.sourceFiles, tc.remoteFiles); got != tc.want {
+			t.Errorf("%s: shouldHintPrune(%d, %d) = %v, want %v",
+				tc.name, tc.sourceFiles, tc.remoteFiles, got, tc.want)
+		}
+	}
+}
+
+// stalePushSetup builds a one-file source (source_files == 1) and a fake API
+// whose remote listing holds remoteFiles file objects and remoteDirs directory
+// objects, so tests can drive the hint decision at chosen boundaries.
+func stalePushSetup(t *testing.T, remoteFiles, remoteDirs int) (string, *fakeAPI) {
+	t.Helper()
+	arc := makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+	})
+	fc := newFakeAPI()
+	objs := make([]bunny.Object, 0, remoteFiles+remoteDirs)
+	for i := 0; i < remoteFiles; i++ {
+		objs = append(objs, bunny.Object{Path: fmt.Sprintf("stale/%d.js", i), Size: 1})
+	}
+	for i := 0; i < remoteDirs; i++ {
+		objs = append(objs, bunny.Object{Path: fmt.Sprintf("dir%d", i), IsDirectory: true})
+	}
+	fc.lists[""] = objs
+	return arc, fc
+}
+
+func TestPushHintFiresWhenStale(t *testing.T) {
+	arc, fc := stalePushSetup(t, 1001, 0)
+	out, _ := captureLogs(t)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	s := out.String()
+	if !strings.Contains(s, "remote:   1001") {
+		t.Fatalf("summary missing remote count: %q", s)
+	}
+	if !strings.Contains(s, "\n\nHint:") {
+		t.Fatalf("expected prune hint: %q", s)
+	}
+	// The hint rides on stdout with the summary, not the stderr warning path.
+	if strings.Contains(s, "\n\nHint:") && strings.Index(s, "remote:") > strings.Index(s, "\n\nHint:") {
+		t.Fatalf("hint must follow the summary: %q", s)
+	}
+}
+
+func TestPushNoHintWithinRetentionWindow(t *testing.T) {
+	arc, fc := stalePushSetup(t, 6, 0)
+	out, _ := captureLogs(t)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	if strings.Contains(out.String(), "hint:") {
+		t.Fatalf("unexpected hint: %q", out.String())
+	}
+}
+
+func TestPushNoHintBelowFloor(t *testing.T) {
+	arc, fc := stalePushSetup(t, 999, 0)
+	out, _ := captureLogs(t)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	if strings.Contains(out.String(), "hint:") {
+		t.Fatalf("unexpected hint below floor: %q", out.String())
+	}
+}
+
+func TestPushHintIgnoresRemoteDirectories(t *testing.T) {
+	// Many directory nodes, few files: directories must not inflate remote_files.
+	arc, fc := stalePushSetup(t, 5, 2000)
+	out, _ := captureLogs(t)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	s := out.String()
+	if !strings.Contains(s, "remote:   5") {
+		t.Fatalf("directories must not count: %q", s)
+	}
+	if strings.Contains(s, "hint:") {
+		t.Fatalf("unexpected hint from directory nodes: %q", s)
+	}
+}
+
+func TestPushHintFiresUnderQuiet(t *testing.T) {
+	arc, fc := stalePushSetup(t, 1001, 0)
+	out, _ := captureLogs(t)
+	log.SetQuiet(true)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	if !strings.Contains(out.String(), "\n\nHint:") {
+		t.Fatalf("hint must appear under --quiet: %q", out.String())
+	}
+}
+
+func TestPushHintFiresUnderDryRun(t *testing.T) {
+	arc, fc := stalePushSetup(t, 1001, 0)
+	out, _ := captureLogs(t)
+	dryRun = true
+
+	if err := runPush(arc, dryRunStorageAPI{fc}); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+	if !strings.Contains(out.String(), "\n\nHint:") {
+		t.Fatalf("hint must appear under dry-run: %q", out.String())
+	}
+}
+
+func TestHumanSize(t *testing.T) {
+	cases := map[int]string{
+		0:           "0 B",
+		512:         "512 B",
+		1024:        "1.0 KiB",
+		1536:        "1.5 KiB",
+		1024 * 1024: "1.0 MiB",
+	}
+	for in, want := range cases {
+		if got := humanSize(in); got != want {
+			t.Errorf("humanSize(%d) = %q, want %q", in, got, want)
+		}
+	}
+}
diff --git a/internal/cli/dryrun.go b/internal/cli/dryrun.go
new file mode 100644
index 0000000..ed65f48
--- /dev/null
+++ b/internal/cli/dryrun.go
@@ -0,0 +1,11 @@
+package cli
+
+import (
+	"github.com/spf13/pflag"
+)
+
+var dryRun = false
+
+func registerDryRunFlag(flags *pflag.FlagSet) {
+	flags.BoolVar(&dryRun, "dryrun", false, "run the command without executing any mutating operations")
+}
diff --git a/internal/cli/dryrun_test.go b/internal/cli/dryrun_test.go
new file mode 100644
index 0000000..c2538d6
--- /dev/null
+++ b/internal/cli/dryrun_test.go
@@ -0,0 +1,78 @@
+package cli
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"tsne.dev/hopper/internal/bunny"
+)
+
+// TestDryRunPushStubsUploads asserts that under dry-run the push flow performs
+// NO uploads against the wrapped API, yet prints the normal per-file action
+// line and summary (logging is owned by the command, not the decorator). The
+// recording fake is the wrapped (real-stand-in) API, so a zero put-count
+// provably shows the write side was never invoked.
+func TestDryRunPushStubsUploads(t *testing.T) {
+	arc := makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+		{name: "assets/app.abc.js", body: "console.log(1)"},
+	})
+
+	spy := newFakeAPI()
+	out, _ := captureLogs(t)
+	dryRun = true
+
+	if err := runPush(arc, dryRunStorageAPI{spy}); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	if got := spy.paths(); len(got) != 0 {
+		t.Fatalf("expected no real uploads, got %v", got)
+	}
+
+	s := out.String()
+	for _, want := range []string{
+		"[dryrun] uploading index.html",
+		"[dryrun] uploading assets/app.abc.js",
+		"uploaded: 2",
+	} {
+		if !strings.Contains(s, want) {
+			t.Fatalf("output missing %q; got:\n%s", want, s)
+		}
+	}
+}
+
+// TestDryRunDeleteStubbed asserts Delete never reaches the wrapped API
+// (reused by prune in issue 008, which will own the delete log line).
+func TestDryRunDeleteStubbed(t *testing.T) {
+	spy := newFakeAPI()
+	api := dryRunStorageAPI{spy}
+
+	if err := api.Delete(context.Background(), "old/file.js"); err != nil {
+		t.Fatalf("Delete: %v", err)
+	}
+	if len(spy.deletes) != 0 {
+		t.Fatalf("expected no real deletes, got %v", spy.deletes)
+	}
+}
+
+// TestDryRunListDelegates asserts the read side still reaches the wrapped API
+// (real Bunny in production), unaffected by the write stubbing.
+func TestDryRunListDelegates(t *testing.T) {
+	spy := newFakeAPI()
+	spy.lists[""] = []bunny.Object{{Path: "a.js", Size: 3}}
+	api := dryRunStorageAPI{spy}
+
+	it := api.List(context.Background(), "")
+	var got []bunny.Object
+	for it.Next() {
+		got = append(got, it.Object())
+	}
+	if err := it.Err(); err != nil {
+		t.Fatalf("List: %v", err)
+	}
+	if len(got) != 1 || got[0].Path != "a.js" {
+		t.Fatalf("List delegate = %v", got)
+	}
+}
diff --git a/internal/cli/duration.go b/internal/cli/duration.go
new file mode 100644
index 0000000..ee40562
--- /dev/null
+++ b/internal/cli/duration.go
@@ -0,0 +1,136 @@
+package cli
+
+import (
+	"fmt"
+	"math"
+	"strings"
+	"time"
+
+	"github.com/spf13/pflag"
+)
+
+// Per-component upper bounds. A cache-safety grace window is measured in days
+// to a few years; anything past these is a typo or an attempt to overflow the
+// cutoff arithmetic. They are chosen so cutoff = now.AddDate(-y,-m,-d) can never
+// overflow the int year addition or the int64 range of time.Time (representable
+// to ~year 292 billion), so the destructive cutoff can never silently wrap.
+const (
+	maxDurationYears  = 100000
+	maxDurationMonths = 100000 * 12
+	maxDurationDays   = 100000 * 366
+)
+
+var _ pflag.Value = (*duration)(nil)
+
+// duration is a calendar-aware grace window . It holds a tuple
+// (years, months, days) rather than a time.Duration.
+type duration struct {
+	years  int
+	months int
+	days   int
+}
+
+func registerDurationFlag(flags *pflag.FlagSet, v *duration, name string, def string, usage string) {
+	if err := v.Set(def); err != nil {
+		panic(fmt.Sprintf("invalid default for --%s: %v", name, err))
+	}
+	flags.Var(v, name, usage)
+}
+
+// Type implements the `pflags.Value` interface.
+func (d *duration) Type() string {
+	return "duration"
+}
+
+// String implements the `pflags.Value` interface.
+func (d *duration) String() string {
+	sb := &strings.Builder{}
+	if d.years != 0 {
+		fmt.Fprintf(sb, "%dy", d.years)
+	}
+	if d.months != 0 {
+		fmt.Fprintf(sb, "%dm", d.months)
+	}
+	if d.days != 0 {
+		fmt.Fprintf(sb, "%dd", d.days)
+	}
+
+	if sb.Len() == 0 {
+		return "0d"
+	}
+	return sb.String()
+}
+
+// Set parses a combinable value like "1y2m10d", "7d" or "30d". Each component is
+// a non-negative integer followed by one of the units y, m (month), d, and each
+// unit may appear at most once. An empty value or a value with no components is
+// rejected.
+//
+// Set implements the `pflags.Value` interface.
+func (d *duration) Set(raw string) error {
+	parsed := duration{}
+	seen := map[byte]bool{}
+	components := 0
+
+	for i := 0; i < len(raw); {
+		c := raw[i]
+		if c < '0' || c > '9' {
+			return usageError{fmt.Sprintf("invalid duration %q", raw)}
+		}
+
+		n := 0
+		for i < len(raw) && raw[i] >= '0' && raw[i] <= '9' {
+			digit := int(raw[i] - '0')
+			if n > (math.MaxInt-digit)/10 {
+				return usageError{fmt.Sprintf("invalid duration %q (out of range)", raw)}
+			}
+			n = n*10 + digit
+			i++
+		}
+		if i >= len(raw) {
+			return usageError{fmt.Sprintf("invalid duration %q (number %d has no unit)", raw, n)}
+		}
+
+		unit := raw[i]
+		if seen[unit] {
+			return usageError{fmt.Sprintf("invalid duration %q (unit %q repeated)", raw, string(unit))}
+		}
+		switch unit {
+		case 'y':
+			if n > maxDurationYears {
+				return usageError{fmt.Sprintf("invalid duration %q (years exceeds %d)", raw, maxDurationYears)}
+			}
+			parsed.years = n
+		case 'm':
+			if n > maxDurationMonths {
+				return usageError{fmt.Sprintf("invalid duration %q (months exceeds %d)", raw, maxDurationMonths)}
+			}
+			parsed.months = n
+		case 'd':
+			if n > maxDurationDays {
+				return usageError{fmt.Sprintf("invalid duration %q (days exceeds %d)", raw, maxDurationDays)}
+			}
+			parsed.days = n
+		default:
+			return usageError{fmt.Sprintf("invalid duration %q (unknown unit %q)", raw, string(unit))}
+		}
+		seen[unit] = true
+		components++
+		i++
+	}
+
+	if components == 0 {
+		return usageError{fmt.Sprintf("invalid duration %q", raw)}
+	}
+
+	*d = parsed
+	return nil
+}
+
+func (d duration) cutoff(tm time.Time) time.Time {
+	return tm.AddDate(-d.years, -d.months, -d.days)
+}
+
+func (d duration) isZero() bool {
+	return d.years == 0 && d.months == 0 && d.days == 0
+}
diff --git a/internal/cli/duration_test.go b/internal/cli/duration_test.go
new file mode 100644
index 0000000..60d446a
--- /dev/null
+++ b/internal/cli/duration_test.go
@@ -0,0 +1,51 @@
+package cli
+
+import (
+	"testing"
+	"time"
+)
+
+func TestDurationValueParses(t *testing.T) {
+	cases := map[string]duration{
+		"30d":     {days: 30},
+		"7d":      {days: 7},
+		"1y2m10d": {years: 1, months: 2, days: 10},
+		"2m":      {months: 2},
+		"1y":      {years: 1},
+		"0d":      {days: 0},
+	}
+	for in, want := range cases {
+		var v duration
+		if err := v.Set(in); err != nil {
+			t.Fatalf("Set(%q): %v", in, err)
+		}
+		if v != want {
+			t.Fatalf("Set(%q) = %+v, want %+v", in, v, want)
+		}
+	}
+}
+
+func TestDurationValueRejectsBadInput(t *testing.T) {
+	for _, in := range []string{"", "d", "5", "5x", "1h", "1m2m", "-3d", "y", "5 d", "99999999999999999999d"} {
+		var v duration
+		if err := v.Set(in); err == nil {
+			t.Fatalf("Set(%q): expected error", in)
+		}
+	}
+}
+
+func TestDurationValueMonthIsCalendarMonth(t *testing.T) {
+	var v duration
+	if err := v.Set("1m"); err != nil {
+		t.Fatal(err)
+	}
+	now := time.Date(2024, time.March, 31, 0, 0, 0, 0, time.UTC)
+	// A calendar-aware month back from March 31 lands in February, not "31 days".
+	got := v.cutoff(now)
+	if got.Month() != time.February && got.Month() != time.March {
+		t.Fatalf("cutoff month = %v, want calendar-aware", got.Month())
+	}
+	if got.Equal(now.AddDate(0, 0, -30)) {
+		t.Fatalf("cutoff must not use a fixed 30-day month")
+	}
+}
diff --git a/internal/cli/errors.go b/internal/cli/errors.go
new file mode 100644
index 0000000..aeeede1
--- /dev/null
+++ b/internal/cli/errors.go
@@ -0,0 +1,89 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"tsne.dev/hopper/internal/log"
+)
+
+// errorVerboser is implemented by error types that can contribute extra
+// technical detail to failure reports under --verbose (e.g. *bunny.apiError
+// exposing verb/path/status/cause). reportFailures prints it indented below the
+// friendly message when verbose is enabled.
+type errorVerboser interface {
+	ErrorVerbose() string
+}
+
+// usageError marks a parse or validation failure that should print a short
+// usage line and exit 2, distinct from a runtime failure (exit 1).
+type usageError struct {
+	msg string
+}
+
+func (e usageError) Error() string { return e.msg }
+
+// exitError signals that the command has already emitted all its user-facing
+// output (e.g. a failure list followed by the summary) and the process should
+// simply exit with the given code, printing nothing further. This keeps the
+// summary as the last line.
+type exitError struct {
+	code int
+}
+
+func (e exitError) Error() string { return fmt.Sprintf("exit status %d", e.code) }
+
+// reportFailures prints a batch of per-file failures to stderr under a headline,
+// using the same layout everywhere or the context's cancellation cause.
+// It returns true if it reported an error, false otherwise.
+func reportFailures(ctx context.Context, msg string, errs []error) bool {
+	if len(errs) > 0 {
+		sb := strings.Builder{}
+		sb.WriteString(msg)
+		for _, err := range errs {
+			sb.WriteString("\n  ")
+			sb.WriteString(err.Error())
+			if log.Verbose() {
+				var v errorVerboser
+				if errors.As(err, &v) {
+					sb.WriteString("\n    ")
+					sb.WriteString(v.ErrorVerbose())
+				}
+			}
+		}
+		log.Error(sb.String())
+		return true
+	} else if ctxErr := context.Cause(ctx); ctxErr != nil {
+		log.Error(ctxErr.Error())
+		return true
+	} else {
+		return false
+	}
+}
+
+// fileError attaches the storage path to an operation failure so the report can
+// name the offending file. It preserves the wrapped cause (e.g. the bunny API
+// error) in the chain so both friendly rendering (Error) and verbose detail
+// (ErrorVerbose) still work via errors.As.
+type fileError struct {
+	path string
+	err  error
+}
+
+func (e fileError) Error() string { return e.path + ": " + e.err.Error() }
+func (e fileError) Unwrap() error { return e.err }
+
+type cancellationError struct {
+	msg string
+}
+
+// CancellationError returns an error that wraps `context.Canceled` but
+// with a custom error message.
+func CancellationError(msg string) error {
+	return cancellationError{msg}
+}
+
+func (e cancellationError) Error() string { return e.msg }
+func (e cancellationError) Unwrap() error { return context.Canceled }
diff --git a/internal/cli/helpers_test.go b/internal/cli/helpers_test.go
new file mode 100644
index 0000000..8aa533f
--- /dev/null
+++ b/internal/cli/helpers_test.go
@@ -0,0 +1,231 @@
+package cli
+
+import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"sync"
+	"testing"
+
+	"tsne.dev/hopper/internal/bunny"
+	"tsne.dev/hopper/internal/log"
+)
+
+// captureLogs redirects the global logger to fresh buffers for the duration of
+// a test and restores the defaults (and the dryRun global) afterward. Returns
+// the stdout and stderr buffers.
+func captureLogs(t *testing.T) (*bytes.Buffer, *bytes.Buffer) {
+	t.Helper()
+	var out, errb bytes.Buffer
+	log.Init(&out, &errb)
+	t.Cleanup(func() {
+		log.Init(os.Stdout, os.Stderr)
+		log.SetVerbose(false)
+		log.SetQuiet(false)
+		dryRun = false
+	})
+	return &out, &errb
+}
+
+// fakeAPI is an in-memory recording storageAPI: the reusable test seam for all
+// command tests. It records every attempted Upload/Delete and serves scripted
+// List results. It satisfies storageAPI, so it can be substituted for the real
+// *bunny.API anywhere (including wrapped by dryRunStorageAPI).
+type fakeAPI struct {
+	mu       sync.Mutex
+	puts     map[string][]byte
+	order    []string
+	deletes  []string
+	lists    map[string][]bunny.Object
+	listErr  error
+	failOn   string
+	failAll  bool
+	failWhen func(path string) bool
+	failErr  error
+}
+
+func newFakeAPI() *fakeAPI {
+	return &fakeAPI{puts: map[string][]byte{}, lists: map[string][]bunny.Object{}}
+}
+
+func (f *fakeAPI) Upload(ctx context.Context, path string, body []byte) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.order = append(f.order, path)
+	if f.failAll || path == f.failOn || (f.failWhen != nil && f.failWhen(path)) {
+		if f.failErr != nil {
+			return f.failErr
+		}
+		return errors.New("forced failure")
+	}
+	b := make([]byte, len(body))
+	copy(b, body)
+	f.puts[path] = b
+	return nil
+}
+
+func (f *fakeAPI) Delete(ctx context.Context, path string) error {
+	return f.recordDelete(path)
+}
+
+// DeleteDir records the logical directory path (no trailing slash) so tests can
+// assert on it uniformly with file deletes; the real trailing-slash quirk lives
+// in the bunny package and is exercised by bunny's own tests.
+func (f *fakeAPI) DeleteDir(ctx context.Context, path string) error {
+	return f.recordDelete(path)
+}
+
+func (f *fakeAPI) recordDelete(path string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.deletes = append(f.deletes, path)
+	if f.failAll || path == f.failOn || (f.failWhen != nil && f.failWhen(path)) {
+		if f.failErr != nil {
+			return f.failErr
+		}
+		return errors.New("forced failure")
+	}
+	return nil
+}
+
+func (f *fakeAPI) List(ctx context.Context, dir string) *bunny.Iterator {
+	return bunny.NewIterator(ctx, dir, f.listDir)
+}
+
+func (f *fakeAPI) listDir(ctx context.Context, dir string) ([]bunny.Object, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.listErr != nil {
+		return nil, f.listErr
+	}
+	return f.lists[dir], nil
+}
+
+func (f *fakeAPI) paths() []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	var ps []string
+	for p := range f.puts {
+		ps = append(ps, p)
+	}
+	sort.Strings(ps)
+	return ps
+}
+
+// ctxCancelAPI simulates the real fail-fast scenario: the upload of triggerPath
+// fails immediately with a genuine (non-context) error, while every other
+// upload blocks until the pool cancels its context and then returns an error
+// that WRAPS context.Canceled (mirroring a real *bunny.apiError built from an
+// aborted in-flight request). It lets a test assert that cancellation collateral
+// is filtered out of the failure report while the genuine trigger survives.
+type ctxCancelAPI struct {
+	*fakeAPI
+	triggerPath string
+	triggerErr  error
+}
+
+func (c *ctxCancelAPI) Upload(ctx context.Context, path string, body []byte) error {
+	c.fakeAPI.mu.Lock()
+	c.fakeAPI.order = append(c.fakeAPI.order, path)
+	c.fakeAPI.mu.Unlock()
+
+	if path == c.triggerPath {
+		return c.triggerErr
+	}
+	// Collateral: wait for the fail-fast cancellation, then report it as a
+	// context-wrapped error, exactly like a real aborted upload.
+	<-ctx.Done()
+	return fmt.Errorf("upload %s aborted: %w", path, ctx.Err())
+}
+
+// gatingAPI wraps fakeAPI to make each Upload block on a release channel while
+// recording the peak number of concurrent in-flight uploads, so tests can
+// assert the pool never exceeds --concurrency.
+type gatingAPI struct {
+	*fakeAPI
+	release chan struct{}
+
+	gmu     sync.Mutex
+	cur     int
+	maxSeen int
+}
+
+func (g *gatingAPI) Upload(ctx context.Context, path string, body []byte) error {
+	g.gmu.Lock()
+	g.cur++
+	if g.cur > g.maxSeen {
+		g.maxSeen = g.cur
+	}
+	g.gmu.Unlock()
+
+	<-g.release
+
+	g.gmu.Lock()
+	g.cur--
+	g.gmu.Unlock()
+	return g.fakeAPI.Upload(ctx, path, body)
+}
+
+func (g *gatingAPI) peak() int {
+	g.gmu.Lock()
+	defer g.gmu.Unlock()
+	return g.maxSeen
+}
+
+// tarEntry describes a single file/header to write into a test archive.
+type tarEntry struct {
+	name     string
+	body     string
+	typeflag byte
+	linkname string
+}
+
+func makeArchive(t *testing.T, entries []tarEntry) string {
+	t.Helper()
+	dir := t.TempDir()
+	p := filepath.Join(dir, "a.tar.gz")
+	f, err := os.Create(p)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer f.Close()
+	gz := gzip.NewWriter(f)
+	tw := tar.NewWriter(gz)
+	for _, e := range entries {
+		tf := e.typeflag
+		if tf == 0 {
+			tf = tar.TypeReg
+		}
+		hdr := &tar.Header{
+			Name:     e.name,
+			Mode:     0o644,
+			Size:     int64(len(e.body)),
+			Typeflag: tf,
+			Linkname: e.linkname,
+		}
+		if tf != tar.TypeReg {
+			hdr.Size = 0
+		}
+		if err := tw.WriteHeader(hdr); err != nil {
+			t.Fatal(err)
+		}
+		if tf == tar.TypeReg {
+			if _, err := tw.Write([]byte(e.body)); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+	if err := tw.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if err := gz.Close(); err != nil {
+		t.Fatal(err)
+	}
+	return p
+}
diff --git a/internal/cli/log.go b/internal/cli/log.go
new file mode 100644
index 0000000..c837649
--- /dev/null
+++ b/internal/cli/log.go
@@ -0,0 +1,22 @@
+package cli
+
+import (
+	"tsne.dev/hopper/internal/log"
+)
+
+// logAction prints a per-action line.
+// In dry-run mode the line is prefixed with "[dryrun] ".
+func logAction(format string, args ...any) {
+	if dryRun {
+		format = "[dryrun] " + format
+	}
+	log.Print(format, args...)
+}
+
+// logSummary prints the final summary line.
+func logSummary(format string, args ...any) {
+	if dryRun {
+		format += "\n\nNote: This was a dry run. Nothing on the zone was modified."
+	}
+	log.PrintImportant(format, args...)
+}
diff --git a/internal/cli/path_prefix.go b/internal/cli/path_prefix.go
new file mode 100644
index 0000000..355349e
--- /dev/null
+++ b/internal/cli/path_prefix.go
@@ -0,0 +1,51 @@
+package cli
+
+import (
+	"fmt"
+	"path"
+	"strings"
+
+	"github.com/spf13/pflag"
+)
+
+var _ pflag.Value = (*pathPrefixValue)(nil)
+
+type pathPrefixValue struct {
+	p    *string
+	name string
+}
+
+func registerPathPrefixFlag(flags *pflag.FlagSet, p *string, name string, val string, usage string) {
+	*p = val
+	flags.Var(&pathPrefixValue{p, name}, name, usage)
+}
+
+func (v *pathPrefixValue) Type() string   { return "string" }
+func (v *pathPrefixValue) String() string { return *v.p }
+
+func (v *pathPrefixValue) Set(raw string) error {
+	trimmed := strings.Trim(raw, "/")
+	if trimmed == "" {
+		*v.p = ""
+		return nil
+	}
+
+	cleaned := path.Clean(trimmed)
+	if cleaned == "." {
+		*v.p = ""
+		return nil
+	}
+	if cleaned == ".." || strings.HasPrefix(cleaned, "../") {
+		return usageError{fmt.Sprintf("%s %q escapes", v.name, raw)}
+	}
+
+	*v.p = cleaned
+	return nil
+}
+
+func prefixedPath(prefix string, filePath string) string {
+	if prefix == "" {
+		return filePath
+	}
+	return path.Join(prefix, filePath)
+}
diff --git a/internal/cli/path_prefix_test.go b/internal/cli/path_prefix_test.go
new file mode 100644
index 0000000..bdb4e42
--- /dev/null
+++ b/internal/cli/path_prefix_test.go
@@ -0,0 +1,52 @@
+package cli
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestPathPrefixValueSet(t *testing.T) {
+	cases := map[string]string{
+		"":           "",
+		"/":          "",
+		"foo":        "foo",
+		"/foo":       "foo",
+		"foo/":       "foo",
+		"/foo/":      "foo",
+		"foo//bar":   "foo/bar",
+		"foo/./bar":  "foo/bar",
+		"foo/bar/..": "foo",
+		"a/b/../c":   "a/c",
+	}
+	for in, want := range cases {
+		var p string
+		v := &pathPrefixValue{p: &p}
+		if err := v.Set(in); err != nil {
+			t.Errorf("Set(%q): unexpected error %v", in, err)
+			continue
+		}
+		if p != want {
+			t.Errorf("Set(%q) = %q, want %q", in, p, want)
+		}
+	}
+}
+
+func TestPathPrefixValueSetEscapes(t *testing.T) {
+	for _, in := range []string{"..", "../x", "/../x", "foo/../../x"} {
+		var p string
+		v := &pathPrefixValue{p: &p}
+		var ue usageError
+		if err := v.Set(in); !errors.As(err, &ue) {
+			t.Errorf("Set(%q): expected usageError, got %v", in, err)
+		}
+	}
+}
+
+func TestPrefixedPath(t *testing.T) {
+	if got := prefixedPath("", "assets/x.js"); got != "assets/x.js" {
+		t.Errorf("empty prefix: got %q", got)
+	}
+	if got := prefixedPath("foo", "assets/x.js"); got != "foo/assets/x.js" {
+		t.Errorf("prefixed: got %q", got)
+	}
+}
diff --git a/internal/cli/remote.go b/internal/cli/remote.go
new file mode 100644
index 0000000..43fe93e
--- /dev/null
+++ b/internal/cli/remote.go
@@ -0,0 +1,94 @@
+package cli
+
+import (
+	"context"
+	"sort"
+	"strings"
+	"time"
+)
+
+type remoteEntry struct {
+	path        string
+	size        int
+	lastChanged time.Time
+	isDir       bool
+	deleted     bool
+
+	// orderKey is path with '/' replaced by 0x00, precomputed so the sort
+	// compares native strings. '/' and 0x00 are the only two bytes that can
+	// never appear in a path, and 0x00 sorts below every other byte, so this
+	// orders a directory's descendants ("a/...") as a contiguous run immediately
+	// after the directory ("a") and before any sibling that shares its base name
+	// ("a.js", "ab"). emptyDirs relies on that contiguity.
+	orderKey string
+}
+
+// listRemoteEntries fetches all remote entries from the api. The returned list
+// is sorted by orderKey (see remoteEntry.orderKey), which is the ordering
+// emptyDirs requires. Callers must not re-sort it by path.
+func listRemoteEntries(ctx context.Context, api storageAPI, prefix string) ([]remoteEntry, error) {
+	entries := make([]remoteEntry, 0, 64)
+	it := api.List(ctx, prefix)
+	for it.Next() {
+		o := it.Object()
+		entries = append(entries, remoteEntry{
+			path:        o.Path,
+			size:        o.Size,
+			isDir:       o.IsDirectory,
+			lastChanged: o.LastChanged,
+			orderKey:    pathOrderKey(o.Path),
+		})
+	}
+	if err := it.Err(); err != nil {
+		return nil, err
+	}
+	sort.Slice(entries, func(i, j int) bool {
+		return entries[i].orderKey < entries[j].orderKey
+	})
+	return entries, nil
+}
+
+// emptyDirs returns the directory entries whose subtree holds no surviving
+// file. entries must be sorted by orderKey (see remoteEntry.orderKey), so a
+// directory's subtree is the contiguous run of following entries sharing its
+// path prefix. A directory exists only to hold files, so it survives iff some
+// file beneath it survives. This makes the bottom-up cascade (a parent emptied
+// by removing its last child directory) fall out without extra bookkeeping.
+// A file survives unless it was deleted.
+func emptyDirs(entries []remoteEntry) []*remoteEntry {
+	n := len(entries)
+	empty := make([]*remoteEntry, 0, n/4)
+	for i := 0; i < n; {
+		dir := &entries[i]
+		i++
+		if !dir.isDir {
+			continue
+		}
+
+		existingFiles := 0
+		c := i
+		for prefix := dir.path + "/"; c < n; c++ {
+			child := &entries[c]
+			if !strings.HasPrefix(child.path, prefix) {
+				break
+			}
+			if !child.isDir && !child.deleted {
+				existingFiles++
+			}
+		}
+
+		if existingFiles == 0 {
+			// Emit only the top of the empty subtree and skip its descendants.
+			// Deleting a Bunny directory removes the subtree recursively, so
+			// deleting `dir` also removes any empty sub-dirs beneath it.
+			empty = append(empty, dir)
+			i = c
+		}
+	}
+
+	return empty
+}
+
+func pathOrderKey(path string) string {
+	return strings.ReplaceAll(path, "/", "\x00")
+}
diff --git a/internal/cli/remote_test.go b/internal/cli/remote_test.go
new file mode 100644
index 0000000..b4e20d9
--- /dev/null
+++ b/internal/cli/remote_test.go
@@ -0,0 +1,141 @@
+package cli
+
+import (
+	"sort"
+	"strings"
+	"testing"
+)
+
+func dir(path string) remoteEntry {
+	return remoteEntry{path: path, isDir: true, orderKey: pathOrderKey(path)}
+}
+func file(path string) remoteEntry {
+	return remoteEntry{path: path, orderKey: pathOrderKey(path)}
+}
+
+func deleted(e remoteEntry) remoteEntry {
+	e.deleted = true
+	return e
+}
+
+func emptyDirPaths(entries []remoteEntry) []string {
+	// Reproduce the production ordering (see listRemoteEntries) so the sort rule
+	// and the emptyDirs scan rule are always exercised together, rather than
+	// trusting hand-ordered literals.
+	sort.Slice(entries, func(i, j int) bool {
+		return entries[i].orderKey < entries[j].orderKey
+	})
+	got := emptyDirs(entries)
+	paths := make([]string, len(got))
+	for i, e := range got {
+		paths[i] = e.path
+	}
+	sort.Strings(paths)
+	return paths
+}
+
+func TestEmptyDirs(t *testing.T) {
+	tests := []struct {
+		name    string
+		entries []remoteEntry
+		want    []string
+	}{
+		{
+			name: "dir with only a deleted file is empty",
+			entries: []remoteEntry{
+				dir("a"),
+				deleted(file("a/gone.js")),
+			},
+			want: []string{"a"},
+		},
+		{
+			name: "dir with a surviving file is kept",
+			entries: []remoteEntry{
+				dir("a"),
+				file("a/live.js"),
+			},
+			want: []string{},
+		},
+		{
+			name: "bottom-up cascade removes parent and child",
+			entries: []remoteEntry{
+				dir("a"),
+				dir("a/b"),
+				deleted(file("a/b/gone.js")),
+			},
+			want: []string{"a"},
+		},
+		{
+			name: "deep file deleted but sibling file alive keeps parent",
+			entries: []remoteEntry{
+				dir("a"),
+				dir("a/b"),
+				deleted(file("a/b/gone.js")),
+				file("a/c.js"),
+			},
+			want: []string{"a/b"},
+		},
+		{
+			name: "sibling with shared name prefix is not swallowed",
+			entries: []remoteEntry{
+				dir("a"),
+				deleted(file("a/gone.js")),
+				file("ab/live.js"),
+			},
+			want: []string{"a"},
+		},
+		{
+			name: "sibling colliding on base name does not hide live children",
+			entries: []remoteEntry{
+				dir("a"),
+				file("a.js"),
+				file("a/live.js"),
+			},
+			want: []string{},
+		},
+		{
+			name: "empty dir with no children is removed",
+			entries: []remoteEntry{
+				dir("a"),
+			},
+			want: []string{"a"},
+		},
+		{
+			name: "nested dirs with no files at all are all removed",
+			entries: []remoteEntry{
+				dir("a"),
+				dir("a/b"),
+				dir("a/c"),
+			},
+			want: []string{"a"},
+		},
+		{
+			name: "empty dir nested under a dir holding a live file keeps outer, removes inner",
+			entries: []remoteEntry{
+				dir("a"),
+				dir("a/b"),
+				dir("a/b/c"),
+				file("a/live.js"),
+			},
+			want: []string{"a/b"},
+		},
+		{
+			name: "live file under a nested dir keeps the whole chain",
+			entries: []remoteEntry{
+				dir("a"),
+				dir("a/b"),
+				file("a/b/live.js"),
+			},
+			want: []string{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := emptyDirPaths(tt.entries)
+			if strings.Join(got, ",") != strings.Join(tt.want, ",") {
+				t.Fatalf("emptyDirs = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/internal/cli/storage.go b/internal/cli/storage.go
new file mode 100644
index 0000000..654ed85
--- /dev/null
+++ b/internal/cli/storage.go
@@ -0,0 +1,46 @@
+package cli
+
+import (
+	"context"
+
+	"tsne.dev/hopper/internal/bunny"
+)
+
+// storageAPI is the full storage surface the dry-run decorator delegates to.
+// Its read side (List) must stay real even under --dry-run; its
+// write side (Upload/Delete) is what the decorator stubs. The real
+// *bunny.Client satisfies this, as does the recording fake used in tests.
+type storageAPI interface {
+	Upload(ctx context.Context, path string, body []byte) error
+	List(ctx context.Context, dir string) *bunny.Iterator
+	Delete(ctx context.Context, path string) error
+	DeleteDir(ctx context.Context, path string) error
+}
+
+func newStorageAPI(endpoint, zone, accessKey string) storageAPI {
+	var api storageAPI = bunny.NewAPI(endpoint, zone, accessKey)
+	if dryRun {
+		api = dryRunStorageAPI{api}
+	}
+	return api
+}
+
+type dryRunStorageAPI struct {
+	upstream storageAPI
+}
+
+func (api dryRunStorageAPI) List(ctx context.Context, dir string) *bunny.Iterator {
+	return api.upstream.List(ctx, dir)
+}
+
+func (api dryRunStorageAPI) Upload(ctx context.Context, path string, body []byte) error {
+	return nil
+}
+
+func (api dryRunStorageAPI) Delete(ctx context.Context, path string) error {
+	return nil
+}
+
+func (api dryRunStorageAPI) DeleteDir(ctx context.Context, path string) error {
+	return nil
+}
diff --git a/internal/cli/verbosity_test.go b/internal/cli/verbosity_test.go
new file mode 100644
index 0000000..68bda1d
--- /dev/null
+++ b/internal/cli/verbosity_test.go
@@ -0,0 +1,107 @@
+package cli
+
+import (
+	"archive/tar"
+	"strings"
+	"testing"
+
+	"tsne.dev/hopper/internal/log"
+)
+
+// verbosityArchive is a small release with one immutable asset and one mutable
+// entry point, enough to exercise action lines and the summary.
+func verbosityArchive(t *testing.T) string {
+	return makeArchive(t, []tarEntry{
+		{name: "index.html", body: "<html>"},
+		{name: "assets/app.abc.js", body: "console.log(1)"},
+		// A symlink triggers an always-on warning regardless of mode.
+		{name: "assets/link.js", typeflag: tar.TypeSymlink, linkname: "app.abc.js"},
+	})
+}
+
+func TestPushDefaultOutput(t *testing.T) {
+	arc := verbosityArchive(t)
+	fc := newFakeAPI()
+	out, errb := captureLogs(t)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	stdout := out.String()
+	if !strings.Contains(stdout, "uploading assets/app.abc.js") {
+		t.Fatalf("default output missing action line: %q", stdout)
+	}
+	if !strings.Contains(stdout, "uploaded: 2") {
+		t.Fatalf("default output missing summary: %q", stdout)
+	}
+	if !strings.Contains(errb.String(), "Warning:") {
+		t.Fatalf("warning missing from stderr: %q", errb.String())
+	}
+}
+
+func TestPushQuietOnlySummaryAndWarnings(t *testing.T) {
+	arc := verbosityArchive(t)
+	fc := newFakeAPI()
+	out, errb := captureLogs(t)
+	log.SetQuiet(true)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	stdout := out.String()
+	if strings.Contains(stdout, "uploading") {
+		t.Fatalf("quiet must suppress action lines: %q", stdout)
+	}
+	if !strings.Contains(stdout, "uploaded: 2") {
+		t.Fatalf("quiet must keep summary: %q", stdout)
+	}
+	if !strings.Contains(errb.String(), "Warning:") {
+		t.Fatalf("quiet must keep warnings on stderr: %q", errb.String())
+	}
+}
+
+func TestPushQuietWinsOverVerbose(t *testing.T) {
+	arc := verbosityArchive(t)
+	fc := newFakeAPI()
+	out, errb := captureLogs(t)
+	log.SetVerbose(true)
+	log.SetQuiet(true)
+
+	if err := runPush(arc, fc); err != nil {
+		t.Fatalf("push: %v", err)
+	}
+
+	stdout := out.String()
+	if strings.Contains(stdout, "uploading") {
+		t.Fatalf("quiet+verbose must suppress action lines: %q", stdout)
+	}
+	if !strings.Contains(stdout, "uploaded: 2") {
+		t.Fatalf("quiet+verbose must keep summary: %q", stdout)
+	}
+	if !strings.Contains(errb.String(), "Warning:") {
+		t.Fatalf("quiet+verbose must keep warnings: %q", errb.String())
+	}
+}
+
+func TestLogStreamRouting(t *testing.T) {
+	out, errb := captureLogs(t)
+
+	log.Print("normal")
+	log.PrintImportant("summary")
+	log.Warn("careful")
+	log.Error("boom")
+
+	stdout := out.String()
+	stderr := errb.String()
+	if !strings.Contains(stdout, "normal") || !strings.Contains(stdout, "summary") {
+		t.Fatalf("stdout missing normal output: %q", stdout)
+	}
+	if strings.Contains(stdout, "careful") || strings.Contains(stdout, "boom") {
+		t.Fatalf("warnings/errors leaked to stdout: %q", stdout)
+	}
+	if !strings.Contains(stderr, "Warning: careful") || !strings.Contains(stderr, "Error: boom") {
+		t.Fatalf("stderr missing warning/error: %q", stderr)
+	}
+}
diff --git a/internal/cli/worker_pool.go b/internal/cli/worker_pool.go
new file mode 100644
index 0000000..fe44644
--- /dev/null
+++ b/internal/cli/worker_pool.go
@@ -0,0 +1,110 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"runtime"
+	"sync"
+	"sync/atomic"
+)
+
+// pool is a bounded worker pool for concurrent, independent tasks (§14). A
+// single caller feeds work via submit; up to concurrency tasks run in flight at
+// once, bounded by the sem token held from submit until the task returns.
+//
+// The pool is deliberately upload-agnostic: a task is any func(ctx) error, so
+// the same machinery serves push uploads today and prune deletes later. Task
+// wording, error identity, and any success side-effect live in the closure.
+//
+// submit is the only place that logs a per-action line, and it is only ever
+// called from one goroutine (the caller feeding work), so output needs no
+// locking; only the shared counters are guarded. Tasks run concurrently, so any
+// state a task closure touches beyond its own arguments is the caller's
+// responsibility to synchronize.
+type workerPool struct {
+	ctx    context.Context
+	cancel context.CancelCauseFunc
+	sem    chan struct{}
+	wg     sync.WaitGroup
+
+	// failFast cancels the pool on the first failure so in-flight tasks are
+	// abandoned and queued ones never start (the push-asset policy, §8.3). When
+	// false, every submitted task runs and all failures are collected (the
+	// commit/prune-delete policy).
+	failFast bool
+
+	mtx       sync.Mutex
+	errs      []error
+	succeeded int64
+}
+
+// newWorkerPool creates a new worker pool to process tasks concurrently. If `concurrency`
+// is zero, the number of parallel jobs is aut-detected based on the number of CPUs.
+func newWorkerPool(ctx context.Context, concurrency uint, failFast bool) *workerPool {
+	if concurrency == 0 {
+		concurrency = 2 * uint(runtime.NumCPU())
+	}
+	ctx, cancel := context.WithCancelCause(ctx)
+	return &workerPool{
+		ctx:      ctx,
+		cancel:   cancel,
+		sem:      make(chan struct{}, concurrency),
+		failFast: failFast,
+	}
+}
+
+// submit blocks until a worker slot is free, then runs task in the background.
+// It reports false when the pool has been cancelled (fail-fast or the parent
+// context), signalling the caller to stop feeding work. action is the text that
+// will be logged before the task is executed.
+func (p *workerPool) submit(action string, task func(ctx context.Context) error) bool {
+	select {
+	case p.sem <- struct{}{}:
+	case <-p.ctx.Done():
+		return false
+	}
+
+	logAction(action)
+
+	p.wg.Add(1)
+	go func() {
+		defer p.wg.Done()
+		defer func() { <-p.sem }()
+
+		if err := task(p.ctx); err != nil {
+			if !errors.Is(err, context.Canceled) {
+				p.mtx.Lock()
+				p.errs = append(p.errs, err)
+				p.mtx.Unlock()
+			}
+			if p.failFast {
+				p.cancel(CancellationError("operation aborted"))
+			}
+			return
+		}
+		atomic.AddInt64(&p.succeeded, 1)
+	}()
+	return true
+}
+
+// cancelled reports whether the pool's context is already done (fail-fast
+// triggered or the parent context cancelled). A feeder loop can poll this to
+// stop producing/buffering work promptly, instead of only discovering the
+// cancellation on its next submit.
+func (p *workerPool) cancelled() bool {
+	select {
+	case <-p.ctx.Done():
+		return true
+	default:
+		return false
+	}
+}
+
+// wait blocks until all in-flight tasks finish, then releases the pool's
+// context. It returns the number of succeeded tasks and the errors the
+// workers collected.
+func (p *workerPool) wait() (int, []error) {
+	p.wg.Wait()
+	p.cancel(nil)
+	return int(p.succeeded), p.errs
+}
diff --git a/internal/cli/worker_pool_test.go b/internal/cli/worker_pool_test.go
new file mode 100644
index 0000000..db3edf0
--- /dev/null
+++ b/internal/cli/worker_pool_test.go
@@ -0,0 +1,70 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+// TestWorkerPoolFailFastSwallowsCancellation locks in the §8.3 invariant that a
+// fail-fast pool reports only GENUINE failures: the first real error cancels the
+// pool, and the in-flight siblings that consequently return context.Canceled are
+// swallowed (cancellation collateral is never an independent per-file failure).
+func TestWorkerPoolFailFastSwallowsCancellation(t *testing.T) {
+	pool := newWorkerPool(context.Background(), 8, true)
+
+	// Occupy several slots with tasks that block until the pool is cancelled and
+	// then return context.Canceled (the collateral that must be swallowed).
+	const blockers = 5
+	for i := 0; i < blockers; i++ {
+		pool.submit("block", func(ctx context.Context) error {
+			<-ctx.Done()
+			return ctx.Err() // context.Canceled
+		})
+	}
+
+	// One genuine failure triggers fail-fast cancellation of the blockers above.
+	pool.submit("fail", func(ctx context.Context) error {
+		return fileError{path: "a.js", err: errors.New("boom")}
+	})
+
+	succeeded, errs := pool.wait()
+	if succeeded != 0 {
+		t.Fatalf("succeeded = %d, want 0", succeeded)
+	}
+	if len(errs) != 1 {
+		t.Fatalf("errs = %d (%v), want exactly 1 genuine failure", len(errs), errs)
+	}
+	if errors.Is(errs[0], context.Canceled) {
+		t.Fatalf("reported error is cancellation collateral, want the genuine failure: %v", errs[0])
+	}
+	var fe fileError
+	if !errors.As(errs[0], &fe) || fe.path != "a.js" {
+		t.Fatalf("reported error = %v, want fileError for a.js", errs[0])
+	}
+}
+
+// TestWorkerPoolDrainReportsAll confirms the non-fail-fast (drain-and-report)
+// mode collects every failure and still runs all tasks.
+func TestWorkerPoolDrainReportsAll(t *testing.T) {
+	pool := newWorkerPool(context.Background(), 4, false)
+
+	const tasks = 6
+	for i := 0; i < tasks; i++ {
+		odd := i%2 == 1
+		pool.submit("t", func(ctx context.Context) error {
+			if odd {
+				return errors.New("fail")
+			}
+			return nil
+		})
+	}
+
+	succeeded, errs := pool.wait()
+	if succeeded != 3 {
+		t.Fatalf("succeeded = %d, want 3", succeeded)
+	}
+	if len(errs) != 3 {
+		t.Fatalf("errs = %d, want 3", len(errs))
+	}
+}