diff --git a/README.md b/README.md
index 964b7db0..297e1d55 100644
--- a/README.md
+++ b/README.md
@@ -94,9 +94,64 @@ See [README_TABLES.md](README_TABLES.md) for detailed documentation.
 Tweaking concurrency can have an impact on performance, especially if latency to the server is tested. 
 Most benchmarks will also use different prefixes for each "thread" running.
 
-By default all benchmarks save all request details to a file named `warp-operation-yyyy-mm-dd[hhmmss]-xxxx.csv.zst`. 
-A custom file name can be specified using the `--benchdata` parameter. 
-The raw data is [zstandard](https://facebook.github.io/zstd/) compressed CSV data.
+By default benchmarks save an aggregated summary to a file named
+`warp-operation-yyyy-mm-dd[hhmmss]-xxxx.json.zst`. A custom base name can be specified
+using the `--benchdata` parameter.
+
+## Full Per-Transaction Logging (`--full`)
+
+Adding `--full` to any benchmark command enables full per-transaction logging. With this
+flag warp writes **both** output files:
+
+| File | Contents |
+|------|----------|
+| `<benchdata>.csv.zst` | Every individual request — one row per operation, [zstandard](https://facebook.github.io/zstd/)-compressed CSV/TSV |
+| `<benchdata>.json.zst` | Aggregated summary (same as default) |
+
+```bash
+warp put --host=... --full
+```
+
+> **Note:** `--full` buffers every individual operation in memory during the benchmark run.
+> Memory overhead scales with operation count and grows with concurrency — measured on a
+> real MinIO target at 64 KiB objects:
+>
+> | Duration | Concurrency | Extra memory | Bytes per op |
+> |----------|-------------|--------------|--------------|
+> | 30 s     | 8           | +9 MB        | ~310 B/op    |
+> | 120 s    | 8           | +84 MB       | ~740 B/op    |
+> | 60 s     | 32          | +124 MB      | ~850 B/op    |
+>
+> At high concurrency the per-op overhead grows due to Go's GC arena behavior. A 1-hour
+> run at 32 concurrent workers could accumulate several GB of additional memory.
+>
+> Analysis time also differs: a `.csv.zst` file takes ~5× longer to analyze with `--full`
+> and ~13× longer to re-aggregate without it, compared to a `.json.zst` aggregate (which
+> analyzes in under 0.2 s regardless of run length). For most benchmarks the overhead is
+> acceptable; avoid `--full` for very long or very high-concurrency runs.
+
+### Analyzing Results
+
+To get accurate per-operation statistics from a `.csv.zst` file, pass `--full` to `warp analyze`:
+
+```bash
+warp analyze --full warp-put-2026-04-07[162451]-xxxx.csv.zst
+```
+
+This reads every individual operation and computes exact latency percentiles and throughput
+split across the full time range.
+
+Running `warp analyze` on a `.csv.zst` **without `--full`** re-aggregates operations into
+1-second buckets (the same representation stored in `.json.zst`). This loses per-operation
+granularity — latency percentiles and throughput can differ slightly, and per-request
+filtering flags (`--analyze.op`, `--analyze.host`, `--analyze.skip`, `--analyze.limit`) have
+no additional effect over the `.json.zst` path. Warp will print a warning in this case.
+
+> **Note:** Passing `--full` when analyzing a `.json.zst` file is silently ignored —
+> aggregate files do not contain individual operations.
+
+Without `--full` only the `.json.zst` aggregate is written and `warp analyze` must be
+pointed at that file instead.
 
 ## Multiple Hosts
 
diff --git a/cli/addcollector_test.go b/cli/addcollector_test.go
new file mode 100644
index 00000000..00c79389
--- /dev/null
+++ b/cli/addcollector_test.go
@@ -0,0 +1,341 @@
+package cli
+
+// Tests for addCollector behavior.
+//
+// The invariants we care about:
+//
+//  1. WITHOUT --full: the live aggregating collector is always active (updates ≠ nil,
+//     live display and autoterm work); per-transaction ops are NOT stored (retrieveOps
+//     returns empty), so no csv.zst is written.
+//
+//  2. WITH --full:    the live aggregating collector is ALSO active (updates ≠ nil,
+//     live display and autoterm still work); per-transaction ops ARE stored
+//     (retrieveOps returns all sent ops), so csv.zst IS written. --full is additive.
+//
+//  3. DiscardOutput=true: both retrieveOps and updates are nil/empty regardless of --full.
+
+import (
+	"context"
+	"flag"
+	"testing"
+	"time"
+
+	mcli "github.com/minio/cli"
+	"github.com/minio/warp/pkg/aggregate"
+	"github.com/minio/warp/pkg/bench"
+)
+
+// ---------------------------------------------------------------------------
+// helpers
+// ---------------------------------------------------------------------------
+
+// stubBench is the minimal bench.Benchmark stub needed by addCollector.
+// It only needs GetCommon(); the other methods are never called in tests.
+type stubBench struct {
+	bench.Common
+}
+
+func (s *stubBench) Prepare(_ context.Context) error                { return nil }
+func (s *stubBench) Start(_ context.Context, _ chan struct{}) error { return nil }
+func (s *stubBench) Cleanup(_ context.Context)                      {}
+
+// makeCtx builds a *cli.Context with only the flags addCollector reads.
+func makeCtx(full bool) *mcli.Context {
+	app := mcli.NewApp()
+	set := flag.NewFlagSet("test", flag.ContinueOnError)
+	set.Bool("full", full, "")
+	return mcli.NewContext(app, set, nil)
+}
+
+// sendOp sends a single representative operation to the collector and returns it.
+func sendOp(t *testing.T, c bench.Collector) bench.Operation {
+	t.Helper()
+	now := time.Now()
+	op := bench.Operation{
+		OpType: "GET",
+		Start:  now,
+		End:    now.Add(time.Millisecond),
+		Size:   1024,
+		Thread: 0,
+	}
+	c.Receiver() <- op
+	return op
+}
+
+// requestFinal asks the live collector for its final aggregate (blocking).
+func requestFinal(t *testing.T, updates chan<- aggregate.UpdateReq) *aggregate.Realtime {
+	t.Helper()
+	ch := make(chan *aggregate.Realtime, 1)
+	updates <- aggregate.UpdateReq{Final: true, C: ch}
+	select {
+	case v := <-ch:
+		return v
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for final aggregate from updates channel")
+		return nil
+	}
+}
+
+// ---------------------------------------------------------------------------
+// Test 1: default mode (no --full)
+// ---------------------------------------------------------------------------
+
+// Without --full:
+//   - updates must be non-nil  (live display / autoterm work)
+//   - retrieveOps must return empty  (no per-transaction recording → no csv.zst)
+func TestAddCollector_DefaultMode_NoOpsStored(t *testing.T) {
+	ctx := makeCtx(false)
+	b := &stubBench{}
+
+	retrieveOps, updates := addCollector(ctx, b)
+
+	// updates must be non-nil so live display and autoterm remain functional.
+	if updates == nil {
+		t.Fatal("expected non-nil updates channel in default (no --full) mode")
+	}
+
+	// Send an operation to the collector and close it.
+	sendOp(t, b.Collector)
+	b.Collector.Close()
+
+	// Without --full, retrieveOps is EmptyOpsCollector → must return empty.
+	ops := retrieveOps()
+	if len(ops) != 0 {
+		t.Errorf("expected 0 ops from retrieveOps in default mode, got %d", len(ops))
+	}
+}
+
+// ---------------------------------------------------------------------------
+// Test 2: --full mode — ops are stored for csv.zst
+// ---------------------------------------------------------------------------
+
+// With --full:
+//   - updates must be non-nil  (live display / autoterm still work)
+//   - retrieveOps must return all ops that were sent  (enables csv.zst write)
+func TestAddCollector_FullMode_OpsAreCollected(t *testing.T) {
+	ctx := makeCtx(true)
+	b := &stubBench{}
+
+	retrieveOps, updates := addCollector(ctx, b)
+
+	// updates must be non-nil even in --full mode.
+	if updates == nil {
+		t.Fatal("expected non-nil updates channel in --full mode")
+	}
+
+	const numOps = 5
+	for i := 0; i < numOps; i++ {
+		sendOp(t, b.Collector)
+	}
+	b.Collector.Close()
+
+	ops := retrieveOps()
+	if len(ops) != numOps {
+		t.Errorf("expected %d ops from retrieveOps in --full mode, got %d", numOps, len(ops))
+	}
+}
+
+// ---------------------------------------------------------------------------
+// Test 3: --full mode — live collector ALSO receives every op (additive)
+// ---------------------------------------------------------------------------
+
+// With --full the live collector is wired in via the extra-channel fan-out.
+// After closing the collector the live aggregate must reflect the ops that
+// were sent (TotalRequests > 0), proving the live path is active alongside
+// the per-transaction path.
+func TestAddCollector_FullMode_LiveCollectorAlsoReceivesOps(t *testing.T) {
+	ctx := makeCtx(true)
+	b := &stubBench{}
+
+	retrieveOps, updates := addCollector(ctx, b)
+
+	const numOps = 3
+	for i := 0; i < numOps; i++ {
+		sendOp(t, b.Collector)
+	}
+	// Close flushes bench.OpsCollector and, via the extra channel, also
+	// signals the live collector to finish computing its aggregate.
+	b.Collector.Close()
+
+	// Sanity-check: per-transaction ops are present.
+	ops := retrieveOps()
+	if len(ops) != numOps {
+		t.Fatalf("expected %d ops, got %d", numOps, len(ops))
+	}
+
+	// Live collector must also have received the ops.
+	final := requestFinal(t, updates)
+	if final == nil {
+		t.Fatal("final aggregate is nil; live collector did not receive ops")
+	}
+	if final.Total.TotalRequests == 0 {
+		t.Errorf("live collector TotalRequests == 0; ops were not forwarded to the live collector")
+	}
+}
+
+// ---------------------------------------------------------------------------
+// Test 4: default mode — updates channel is writable (live collector active)
+// ---------------------------------------------------------------------------
+
+// Without --full the live collector is the sole collector.
+// We verify the updates channel is non-nil and functional. We cannot safely
+// call requestFinal in this case because aggregate.collector.Close() sets its
+// internal channel field to nil — if the LiveCollector goroutine reads after
+// that, it would block on a nil channel range. In production this is fine
+// because Close() is called minutes into a benchmark; in tests we avoid the
+// race by only checking structural invariants here.
+func TestAddCollector_DefaultMode_UpdatesChannelFunctional(t *testing.T) {
+	ctx := makeCtx(false)
+	b := &stubBench{}
+
+	_, updates := addCollector(ctx, b)
+
+	if updates == nil {
+		t.Fatal("expected non-nil updates channel in default mode")
+	}
+
+	// Send ops then close.
+	const numOps = 4
+	for i := 0; i < numOps; i++ {
+		sendOp(t, b.Collector)
+	}
+	b.Collector.Close()
+
+	// The updates channel is buffered (capacity 1000); a write must not block.
+	// A Reset request is a no-op for the live collector and safe to use here.
+	select {
+	case updates <- aggregate.UpdateReq{Reset: true}:
+		// Channel is live and accepting writes.
+	default:
+		t.Fatal("updates channel blocked unexpectedly")
+	}
+}
+
+// ---------------------------------------------------------------------------
+// Test 5: DiscardOutput=true always suppresses collection
+// ---------------------------------------------------------------------------
+
+// When DiscardOutput is true, both retrieveOps and updates must be nil/empty.
+// This is the "silent / terse run" mode used in distributed client mode.
+func TestAddCollector_DiscardOutput_NullCollector(t *testing.T) {
+	// DiscardOutput with --full=false
+	t.Run("full=false", func(t *testing.T) {
+		ctx := makeCtx(false)
+		b := &stubBench{}
+		b.DiscardOutput = true
+
+		retrieveOps, updates := addCollector(ctx, b)
+
+		if updates != nil {
+			t.Error("expected nil updates when DiscardOutput=true and full=false")
+		}
+		ops := retrieveOps()
+		if len(ops) != 0 {
+			t.Errorf("expected 0 ops when DiscardOutput=true and full=false, got %d", len(ops))
+		}
+	})
+
+	// DiscardOutput with --full=true (--full must not override DiscardOutput)
+	t.Run("full=true", func(t *testing.T) {
+		ctx := makeCtx(true)
+		b := &stubBench{}
+		b.DiscardOutput = true
+
+		retrieveOps, updates := addCollector(ctx, b)
+
+		if updates != nil {
+			t.Error("expected nil updates when DiscardOutput=true even with --full")
+		}
+		ops := retrieveOps()
+		if len(ops) != 0 {
+			t.Errorf("expected 0 ops when DiscardOutput=true even with --full, got %d", len(ops))
+		}
+	})
+}
+
+// ---------------------------------------------------------------------------
+// Test 6: --full is ADDITIVE — both file outputs can coexist in one run
+// ---------------------------------------------------------------------------
+
+// This test validates the core contract: --full must not disable any existing
+// behavior, only add the per-transaction csv.zst path on top.
+// We verify this by confirming both the ops slice AND the live aggregate are
+// non-empty after the same set of operations.
+func TestAddCollector_FullMode_IsAdditive(t *testing.T) {
+	ctx := makeCtx(true)
+	b := &stubBench{}
+
+	retrieveOps, updates := addCollector(ctx, b)
+
+	// Both must be non-nil immediately after addCollector returns.
+	if updates == nil {
+		t.Fatal("updates must be non-nil with --full (live display required)")
+	}
+
+	// Send ops.
+	const numOps = 6
+	for i := 0; i < numOps; i++ {
+		sendOp(t, b.Collector)
+	}
+	b.Collector.Close()
+
+	// Per-transaction store must be full.
+	ops := retrieveOps()
+	if len(ops) != numOps {
+		t.Errorf("per-transaction store: expected %d ops, got %d", numOps, len(ops))
+	}
+
+	// Live aggregate must also be non-empty (additive, not replacing).
+	final := requestFinal(t, updates)
+	if final == nil {
+		t.Fatal("live aggregate is nil; --full must not disable the live collector")
+	}
+	if final.Total.TotalRequests == 0 {
+		t.Errorf("live aggregate TotalRequests == 0; --full must not disable the live collector")
+	}
+}
+
+// ---------------------------------------------------------------------------
+// Test 7: ops correctness — values are preserved through the collector
+// ---------------------------------------------------------------------------
+
+// With --full, the ops returned by retrieveOps must be the exact operations
+// that were sent — no data loss or corruption through the fan-out path.
+func TestAddCollector_FullMode_OpValuesPreserved(t *testing.T) {
+	ctx := makeCtx(true)
+	b := &stubBench{}
+
+	retrieveOps, _ := addCollector(ctx, b)
+
+	now := time.Now().Truncate(time.Millisecond) // avoid sub-ms rounding
+	op := bench.Operation{
+		OpType: "PUT",
+		Start:  now,
+		End:    now.Add(42 * time.Millisecond),
+		Size:   8192,
+		Thread: 3,
+	}
+	b.Collector.Receiver() <- op
+	b.Collector.Close()
+
+	ops := retrieveOps()
+	if len(ops) != 1 {
+		t.Fatalf("expected 1 op, got %d", len(ops))
+	}
+	got := ops[0]
+	if got.OpType != op.OpType {
+		t.Errorf("OpType: got %q want %q", got.OpType, op.OpType)
+	}
+	if !got.Start.Equal(op.Start) {
+		t.Errorf("Start: got %v want %v", got.Start, op.Start)
+	}
+	if !got.End.Equal(op.End) {
+		t.Errorf("End: got %v want %v", got.End, op.End)
+	}
+	if got.Size != op.Size {
+		t.Errorf("Size: got %d want %d", got.Size, op.Size)
+	}
+	if got.Thread != op.Thread {
+		t.Errorf("Thread: got %d want %d", got.Thread, op.Thread)
+	}
+}
diff --git a/cli/analyze.go b/cli/analyze.go
index 923c9e7c..c739ae57 100644
--- a/cli/analyze.go
+++ b/cli/analyze.go
@@ -147,6 +147,12 @@ func mainAnalyze(ctx *cli.Context) error {
 					log("Loading %q", arg)
 				}
 			} else {
+				if !globalQuiet && !globalJSON {
+					console.SetColor("Print", color.New(color.FgHiYellow))
+					console.Println("Note: Analyzing .csv.zst without --full re-aggregates into 1-second buckets (same as .json.zst).")
+					console.Println("      Use 'warp analyze --full <file>' for exact per-operation latency and throughput.")
+					console.SetColor("Print", color.New(color.FgWhite))
+				}
 				opCh := make(chan bench.Operation, 10000)
 				go func() {
 					err := bench.StreamOperationsFromCSV(rc, false, ctx.Int("analyze.offset"), ctx.Int("analyze.limit"), log, opCh)
diff --git a/cli/benchmark.go b/cli/benchmark.go
index 03faee30..4fbcbaea 100644
--- a/cli/benchmark.go
+++ b/cli/benchmark.go
@@ -256,6 +256,37 @@ func runBench(ctx *cli.Context, b bench.Benchmark) error {
 				}()
 			}
 		}
+		// --full is additive: also write the json.zst aggregate alongside csv.zst.
+		if updates != nil {
+			finalCh := make(chan *aggregate.Realtime, 1)
+			updates <- aggregate.UpdateReq{Final: true, C: finalCh}
+			final := <-finalCh
+			final.Commandline = commandLine(ctx)
+			final.WarpVersion = GlobalVersion
+			final.WarpDate = GlobalDate
+			final.WarpCommit = GlobalCommit
+			f, err := os.Create(fileName + ".json.zst")
+			if err != nil {
+				monitor.Errorln("Unable to write benchmark data:", err)
+			} else {
+				func() {
+					defer f.Close()
+					enc, err := zstd.NewWriter(f, zstd.WithEncoderLevel(zstd.SpeedBetterCompression))
+					if err != nil {
+						monitor.Errorln("Unable to compress benchmark data:", err)
+						return
+					}
+					defer enc.Close()
+					js := json.NewEncoder(enc)
+					js.SetIndent("", "  ")
+					err = js.Encode(final)
+					if err != nil {
+						monitor.Errorln("Unable to write benchmark data:", err)
+					}
+					monitor.InfoLn(fmt.Sprintf("\nBenchmark data written to %q\n\n", fileName+".json.zst"))
+				}()
+			}
+		}
 		monitor.OperationsReady(ops, fileName, commandLine(ctx))
 		var buf bytes.Buffer
 		printAnalysis(ctx, &buf, ops)
@@ -567,22 +598,27 @@ func runClientBenchmark(ctx *cli.Context, b bench.Benchmark, cb *clientBenchmark
 }
 
 func addCollector(ctx *cli.Context, b bench.Benchmark) (bench.OpsCollector, chan<- aggregate.UpdateReq) {
-	// Add collectors
 	common := b.GetCommon()
-
-	if !ctx.Bool("full") {
-		updates := make(chan aggregate.UpdateReq, 1000)
-		c := aggregate.LiveCollector(context.Background(), updates, pRandASCII(4), common.ExtraOut)
-		common.Collector = c
-		return bench.EmptyOpsCollector, updates
-	}
 	if common.DiscardOutput {
 		common.Collector = bench.NewNullCollector(common.ExtraOut...)
 		return bench.EmptyOpsCollector, nil
 	}
-	var retrieveOps bench.OpsCollector
-	common.Collector, retrieveOps = bench.NewOpsCollector(common.ExtraOut...)
-	return retrieveOps, nil
+	// Always create the live aggregating collector for real-time display and autoterm.
+	updates := make(chan aggregate.UpdateReq, 1000)
+	if ctx.Bool("full") {
+		// --full additionally collects every individual operation so that a
+		// per-transaction csv.zst file is written after the benchmark.
+		// Each op is forwarded to the live collector via the extra channel,
+		// so live display and autoterm continue to work normally.
+		liveC := aggregate.LiveCollector(context.Background(), updates, pRandASCII(4), nil)
+		var retrieveOps bench.OpsCollector
+		common.Collector, retrieveOps = bench.NewOpsCollector(append(common.ExtraOut, liveC.Receiver())...)
+		return retrieveOps, updates
+	}
+	// Default: live aggregating collector only; no per-transaction file.
+	c := aggregate.LiveCollector(context.Background(), updates, pRandASCII(4), common.ExtraOut)
+	common.Collector = c
+	return bench.EmptyOpsCollector, updates
 }
 
 type runningProfiles struct {
diff --git a/cli/benchserver.go b/cli/benchserver.go
index 4f0e6f18..7d8dc3db 100644
--- a/cli/benchserver.go
+++ b/cli/benchserver.go
@@ -103,10 +103,6 @@ func runServerBenchmark(ctx *cli.Context, b bench.Benchmark) (bool, error) {
 		return false, nil
 	}
 
-	if ctx.Bool("autoterm") && ctx.Bool("full") {
-		return true, errors.New("use of -autoterm cannot be used with --full on remote benchmarks")
-	}
-
 	var ui ui
 	if !globalQuiet && !globalJSON {
 		go ui.Run()
@@ -217,7 +213,7 @@ func runServerBenchmark(ctx *cli.Context, b bench.Benchmark) (bool, error) {
 	var updates chan aggregate.UpdateReq
 	srv := wui.New(nil)
 	showAddress := ""
-	if !ctx.Bool("full") {
+	{
 		updates = make(chan aggregate.UpdateReq, 10)
 		monitor.SetUpdate(updates)
 		if ctx.Bool("web") {
@@ -245,9 +241,6 @@ func runServerBenchmark(ctx *cli.Context, b bench.Benchmark) (bool, error) {
 	ui.SetSubText("Press 'q' to stop benchmark. " + showAddress)
 
 	if ctx.Bool("autoterm") {
-		if ctx.Bool("full") {
-			return true, errors.New("use of -autoterm cannot be combined with -full on remote benchmarks")
-		}
 		common.AutoTermDur = ctx.Duration("autoterm.dur")
 		common.AutoTermScale = ctx.Float64("autoterm.pct") / 100
 		if common.AutoTermDur > 0 {
@@ -267,7 +260,7 @@ func runServerBenchmark(ctx *cli.Context, b bench.Benchmark) (bool, error) {
 	prof.stop(context.Background(), ctx, fileName+".profiles.zip")
 
 	ui.SetPhase("Downloading Operations")
-	if updates == nil {
+	if ctx.Bool("full") {
 		downloaded := conns.downloadOps()
 		switch len(downloaded) {
 		case 0:
@@ -300,6 +293,34 @@ func runServerBenchmark(ctx *cli.Context, b bench.Benchmark) (bool, error) {
 				}()
 			}
 		}
+		// --full is additive: also download and write the json.zst aggregate.
+		{
+			final := conns.downloadAggr()
+			final.Commandline = commandLine(ctx)
+			final.WarpVersion = GlobalVersion
+			final.WarpDate = GlobalDate
+			final.WarpCommit = GlobalCommit
+			if final.Total.TotalRequests == 0 || len(allOps) == 0 {
+				fatalIf(probe.NewError(errors.New("no operations received")), "No benchmark data received")
+			}
+			f, err := os.Create(fileName + ".json.zst")
+			if err != nil {
+				monitor.Errorln("Unable to write benchmark data:", err)
+			} else {
+				func() {
+					defer f.Close()
+					enc, err := zstd.NewWriter(f, zstd.WithEncoderLevel(zstd.SpeedBetterCompression))
+					fatalIf(probe.NewError(err), "Unable to compress benchmark output")
+					defer enc.Close()
+					js := json.NewEncoder(enc)
+					js.SetIndent("", "  ")
+					err = js.Encode(final)
+					fatalIf(probe.NewError(err), "Unable to write benchmark output")
+					monitor.InfoLn(fmt.Sprintf("Benchmark data written to %q\n", fileName+".json.zst"))
+				}()
+			}
+			monitor.UpdateAggregate(&final, fileName)
+		}
 		monitor.OperationsReady(allOps, fileName, commandLine(ctx))
 		ui.Update(tea.Quit())
 		ui.Wait()
diff --git a/pkg/aggregate/collector.go b/pkg/aggregate/collector.go
index 48d96876..203216d8 100644
--- a/pkg/aggregate/collector.go
+++ b/pkg/aggregate/collector.go
@@ -38,8 +38,9 @@ func LiveCollector(ctx context.Context, updates chan UpdateReq, clientID string,
 		updates = make(chan UpdateReq, 1000)
 	}
 	c.updates = updates
+	rcv := c.rcv
 	go func() {
-		final := Live(c.rcv, updates, clientID, extra)
+		final := Live(rcv, updates, clientID, extra)
 		for {
 			select {
 			case <-ctx.Done():