diff --git a/Makefile b/Makefile
index 3778351..924afa0 100644
--- a/Makefile
+++ b/Makefile
@@ -7,7 +7,7 @@ COVERAGE_FILE=coverage.out
 # Linker flags to strip debug information
 LDFLAGS=-ldflags="-s -w"
 
-.PHONY: all build clean test coverage release help it-clean fixtures
+.PHONY: all build clean fmt vet test coverage release help it-clean fixtures
 
 # Name of the shared, reused integration-test broker container.
 IT_BROKER=artemisctl-it-broker
@@ -22,12 +22,20 @@ build:
 	go build -o $(BUILD_DIR)/$(BINARY_NAME) $(MAIN_PATH)
 	@echo "==> Done. Binary is in $(BUILD_DIR)/$(BINARY_NAME)"
 
-## test: Run all tests (including our Testcontainers broker tests)
-test:
-	@echo "==> Running linting..."
+## fmt: Format every Go file in the module (gofmt -s, in place)
+fmt:
+	@echo "==> Formatting..."
 	gofmt -s -w .
+	@echo "==> Done."
+
+## vet: Run go vet over the whole module
+vet:
 	@echo "==> Running go vet..."
 	go vet ./...
+	@echo "==> Done."
+
+## test: Run all tests (including our Testcontainers broker tests)
+test: fmt vet
 	@echo "==> Running tests..."
 	go test -v -p 1 ./internal/...
 
@@ -86,7 +94,9 @@ help:
 	@echo ""
 	@echo "Targets:"
 	@echo "  build    - Compile the CLI for your current operating system"
-	@echo "  test     - Run all tests"
+	@echo "  fmt      - Format every Go file (gofmt -s -w)"
+	@echo "  vet      - Run go vet over the whole module"
+	@echo "  test     - Format, vet, then run all tests"
 	@echo "  coverage - Run tests with coverage and generate an HTML report"
 	@echo "  release  - Cross-compile the CLI for Linux, macOS, and Windows"
 	@echo "  fixtures - Regenerate internal/journal/testdata from a live 2.42 container"
diff --git a/internal/broker/drain.go b/internal/broker/drain.go
index f97fc41..cb0c8b4 100644
--- a/internal/broker/drain.go
+++ b/internal/broker/drain.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/Azure/go-amqp"
@@ -17,20 +18,184 @@ type RecordSink interface {
 	Sync() error
 }
 
-// DrainQueue consumes every message currently on queue, persisting each
-// batch to sink (Append + Sync/fsync) before acking the broker, so a crash
-// mid-drain never loses a message: it is either still on the broker
-// (unacked) or durably on disk (or both, which is safe to re-drain/replay).
+// PartialDrainError reports that a drain finished with messages still on the
+// broker. It carries the broker's own counters at the moment the drain gave
+// up, so the caller can see WHY the rest did not come: scheduled for later,
+// in flight to another consumer, or a paused queue.
 //
-// Idle detection: each Receive uses a per-call context with timeout idle;
-// a context.DeadlineExceeded from Receive is treated as "queue is empty",
-// not an error, and ends the drain — but only when the outer ctx is still
-// live. If the caller's own ctx has expired/been canceled, that deadline
-// propagates to the child context too, so DeadlineExceeded alone can't
-// distinguish "queue idle" from "caller ran out of time"; we disambiguate
-// against ctx.Err() below so a caller-timeout is reported as an error
-// instead of a false "fully drained".
+// This exists because an idle receive is not proof of an empty queue, and an
+// export that quietly returns a partial file is worse than one that fails.
+type PartialDrainError struct {
+	Queue   string
+	Drained int
+	Stat    QueueStat
+
+	// Scanned reports whether a countMessages scan ran; Counted is what it
+	// found. Stat.MessageCount is only a counter, so when a scan finds 0 while
+	// the counter reports a backlog, the counter has drifted and the
+	// "remaining" messages do not exist -- no drain can ever retrieve them, and
+	// saying so is the difference between a broker bug and a tool bug.
+	//
+	// Scanned gates Counted so that the zero value means "no scan ran" rather
+	// than the far more alarming "a scan found nothing".
+	Scanned bool
+	Counted int64
+}
+
+// outcome is what the broker's counters say about a finished drain pass.
+type outcome int
+
+const (
+	// drainComplete: the broker holds nothing more.
+	drainComplete outcome = iota
+	// drainRetry: messages remain and the broker would hand them over, so the
+	// quiet gap was a stall (GC, depaging, slow network) rather than an end.
+	drainRetry
+	// drainStuck: messages remain that the broker will not deliver to us now.
+	drainStuck
+)
+
+// maxFruitlessPasses bounds how many times in a row DrainQueue will re-attempt
+// a queue that the broker says still holds deliverable messages but that hands
+// over nothing. One fruitless pass is expected and harmless (the broker can
+// still be settling the previous pass's acks, or briefly stalled), but a queue
+// that keeps promising messages it never delivers must fail rather than spin.
+// Each pass already costs a full idle timeout, so this stays cheap.
+const maxFruitlessPasses = 3
+
+func (e *PartialDrainError) Error() string {
+	reasons := make([]string, 0, 3)
+	if e.Stat.ScheduledCount > 0 {
+		reasons = append(reasons, fmt.Sprintf("%d scheduled for later delivery", e.Stat.ScheduledCount))
+	}
+	if e.Stat.DeliveringCount > 0 && e.Stat.ConsumerCount > 0 {
+		reasons = append(reasons, fmt.Sprintf("%d in flight to %d other consumer(s)",
+			e.Stat.DeliveringCount, e.Stat.ConsumerCount))
+	}
+	if e.Stat.Paused {
+		reasons = append(reasons, "queue is paused")
+	}
+	// With no counter to explain the remainder, the broker is telling us these
+	// messages ARE deliverable yet handing over none of them -- a broker-side
+	// stall, not a queue holding messages back. Say exactly that: "the broker
+	// stopped delivering them" reads like a tool giving up, and leaves the
+	// operator with nowhere to look.
+	why := fmt.Sprintf("all %d are deliverable now, but the broker handed over none across %d retries"+
+		"; this is a broker-side stall, not a queue holding them back -- check the broker log and its disk",
+		e.Stat.DeliverableNow(), maxFruitlessPasses)
+	if e.Scanned && e.Counted == 0 && e.Stat.MessageCount > 0 {
+		// The counter advertises a backlog that a scan of the queue cannot
+		// find. Nothing is recoverable here and no retry will help: say so
+		// plainly rather than let it read as an export that failed.
+		why = fmt.Sprintf("messageCount reports %d but a countMessages scan finds 0"+
+			": the queue's message counter has drifted and those messages do not exist"+
+			" -- nothing is left to export; restart the broker to rebuild the counter from its journal",
+			e.Stat.MessageCount)
+	} else if e.Scanned && e.Counted > 0 && len(reasons) == 0 {
+		why = fmt.Sprintf("a scan confirms %d real message(s) and all are deliverable now,"+
+			" yet the broker handed over none across %d retries"+
+			": this is a broker-side stall -- check the broker log and its disk",
+			e.Counted, maxFruitlessPasses)
+	}
+	if len(reasons) > 0 {
+		why = strings.Join(reasons, ", ")
+	}
+	return fmt.Sprintf("incomplete drain of %s: took %d message(s), but %d remain on the broker (%s)",
+		e.Queue, e.Drained, e.Stat.MessageCount, why)
+}
+
+// DrainQueue consumes every message on queue, persisting each batch to sink
+// (Append + Sync/fsync) before acking the broker, so a crash mid-drain never
+// loses a message: it is either still on the broker (unacked) or durably on
+// disk (or both, which is safe to re-drain/replay).
+//
+// Completion is decided by the broker, not by silence. A gap in delivery only
+// ends a drain *pass*; DrainQueue then asks the broker for the queue's real
+// depth and acts on it:
+//
+//   - depth 0: the queue is genuinely empty, the drain is complete.
+//   - messages remain and are deliverable: the gap was a stall (broker GC,
+//     depaging, a slow network), so it drains again.
+//   - messages remain but the broker is holding them back (scheduled, in
+//     flight to another consumer, paused queue): no amount of waiting will
+//     produce them, so it returns *PartialDrainError with the counters.
+//
+// The earlier behaviour — treat any idle gap as "queue drained" and return nil
+// — silently under-drained: a queue reporting 210K messages could export ~50K
+// and still exit 0, because most of the depth was never deliverable to us.
 func (c *Client) DrainQueue(ctx context.Context, queue string, sink RecordSink, idle time.Duration, batch int) (int, error) {
+	total := 0
+	fruitless := 0
+	for {
+		n, err := c.drainPass(ctx, queue, sink, idle, batch)
+		total += n
+		if err != nil {
+			return total, err
+		}
+		if n > 0 {
+			fruitless = 0
+		} else {
+			fruitless++
+		}
+
+		stat, err := c.QueueStatByName(ctx, queue)
+		if err != nil {
+			if errors.Is(err, ErrQueueNotFound) {
+				// Auto-delete removed the queue once we emptied it; nothing
+				// is left behind, so the drain is complete.
+				return total, nil
+			}
+			return total, fmt.Errorf("verify drain of %s: %w", queue, err)
+		}
+		switch drainOutcome(stat, fruitless) {
+		case drainComplete:
+			return total, nil
+		case drainRetry:
+			continue
+		default:
+			// Only now, on the failure path, pay for a scan: countMessages
+			// walks the queue, so it is far more expensive than the counter
+			// read above and is worth it exactly once, to say WHY.
+			pde := &PartialDrainError{Queue: queue, Drained: total, Stat: stat}
+			if n, cErr := c.CountMessages(ctx, queue); cErr == nil {
+				pde.Scanned, pde.Counted = true, n
+			}
+			return total, pde
+		}
+	}
+}
+
+// drainOutcome decides what a pass means, given the broker's counters
+// afterwards and how many consecutive passes have now come back empty
+// (fruitless == 0 means the last pass took messages). It is the whole of the
+// "is this drain actually finished?" judgement, kept pure so every case is
+// testable without a broker.
+func drainOutcome(stat QueueStat, fruitless int) outcome {
+	if stat.MessageCount == 0 {
+		return drainComplete
+	}
+	// Retrying is only worth it if the broker would actually hand the
+	// remainder over -- otherwise we would spin forever on messages we cannot
+	// have -- and only while passes are still producing something.
+	if stat.DeliverableNow() > 0 && fruitless < maxFruitlessPasses {
+		return drainRetry
+	}
+	return drainStuck
+}
+
+// drainPass consumes messages from queue until the broker goes quiet for idle,
+// returning how many it took. A quiet broker ends the pass; it does NOT mean
+// the queue is empty -- only DrainQueue's check against the broker's counters
+// can establish that.
+//
+// Idle detection: each Receive uses a per-call context with timeout idle;
+// a context.DeadlineExceeded from Receive ends the pass, but only when the
+// outer ctx is still live. If the caller's own ctx has expired/been canceled,
+// that deadline propagates to the child context too, so DeadlineExceeded alone
+// can't distinguish "broker quiet" from "caller ran out of time"; we
+// disambiguate against ctx.Err() below so a caller-timeout is reported as an
+// error instead of a false "fully drained".
+func (c *Client) drainPass(ctx context.Context, queue string, sink RecordSink, idle time.Duration, batch int) (int, error) {
 	if batch <= 0 {
 		// Credit: int32(batch) with batch<=0 grants zero link credit, which
 		// silently receives nothing (or reports (0,nil) on a non-empty
@@ -81,7 +246,7 @@ func (c *Client) DrainQueue(ctx context.Context, queue string, sink RecordSink,
 					// surface it as an error instead.
 					return total, ctx.Err()
 				}
-				break // queue idle => drained
+				break // broker quiet => end of pass (NOT proof of an empty queue)
 			}
 			// Messages already Appended-but-not-yet-Acked when this error
 			// aborts the drain stay safely on the broker (no loss), but may
@@ -141,21 +306,33 @@ func (c *Client) DrainQueue(ctx context.Context, queue string, sink RecordSink,
 // DrainAll enumerates every queue via ListQueues and drains each one in
 // turn, invoking onQueue (if non-nil) after each queue completes with the
 // queue's name and the number of messages drained from it.
+//
+// A queue that cannot be fully drained does not abort the run: whatever the
+// other queues hold is still worth exporting, so DrainAll records the
+// *PartialDrainError, moves on, and returns every one of them joined together
+// at the end. The messages it did take are already persisted, and the returned
+// count reflects them, but the error means the export is NOT complete.
+// Any other error (a broken connection, a failing sink) aborts immediately.
 func (c *Client) DrainAll(ctx context.Context, sink RecordSink, idle time.Duration, batch int, onQueue func(name string, n int)) (int, error) {
 	queues, err := c.ListQueues(ctx)
 	if err != nil {
 		return 0, err
 	}
 	total := 0
+	var partial []error
 	for _, q := range queues {
 		n, err := c.DrainQueue(ctx, q.Name, sink, idle, batch)
+		total += n
 		if err != nil {
-			return total, err
+			var pde *PartialDrainError
+			if !errors.As(err, &pde) {
+				return total, err
+			}
+			partial = append(partial, err)
 		}
 		if onQueue != nil {
 			onQueue(q.Name, n)
 		}
-		total += n
 	}
-	return total, nil
+	return total, errors.Join(partial...)
 }
diff --git a/internal/broker/drain_verify_integration_test.go b/internal/broker/drain_verify_integration_test.go
new file mode 100644
index 0000000..79d7682
--- /dev/null
+++ b/internal/broker/drain_verify_integration_test.go
@@ -0,0 +1,302 @@
+package broker
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/Azure/go-amqp"
+)
+
+// scheduleAt marks every message for delivery at t. Artemis honours the AMQP
+// x-opt-delivery-time annotation for scheduled delivery.
+func scheduleAt(msgs []*amqp.Message, at time.Time) {
+	for _, m := range msgs {
+		m.Annotations = amqp.Annotations{"x-opt-delivery-time": at.UnixMilli()}
+	}
+}
+
+// TestDrainQueueScheduledRemainderIsReported pins the reported bug: a queue
+// whose depth is mostly messages the broker will not deliver yet must NOT look
+// like a completed drain. Before the fix, DrainQueue took the deliverable
+// messages, saw the broker go quiet, and returned nil -- so a DLQ reporting
+// 210K exported ~50K and still exited 0.
+func TestDrainQueueScheduledRemainderIsReported(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("sched-remainder")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	if _, err := c.Produce(ctx, queue, GenerateMessages(400, 256, nil), 0, 4, nil); err != nil {
+		t.Fatalf("produce deliverable: %v", err)
+	}
+	later := GenerateMessages(600, 256, nil)
+	scheduleAt(later, time.Now().Add(time.Hour))
+	if _, err := c.Produce(ctx, queue, later, 0, 4, nil); err != nil {
+		t.Fatalf("produce scheduled: %v", err)
+	}
+
+	sink := &sliceSink{}
+	n, err := c.DrainQueue(ctx, queue, sink, 3*time.Second, 100)
+
+	var pde *PartialDrainError
+	if !errors.As(err, &pde) {
+		t.Fatalf("drain returned n=%d err=%v, want *PartialDrainError", n, err)
+	}
+	if n != 400 || len(sink.recs) != 400 {
+		t.Fatalf("drained %d (sink %d), want the 400 deliverable messages", n, len(sink.recs))
+	}
+	if pde.Stat.ScheduledCount != 600 {
+		t.Fatalf("reported ScheduledCount = %d, want 600", pde.Stat.ScheduledCount)
+	}
+	if pde.Stat.MessageCount != 600 {
+		t.Fatalf("reported MessageCount = %d, want 600 left on broker", pde.Stat.MessageCount)
+	}
+	t.Logf("drain correctly refused to claim success: %v", err)
+}
+
+// TestDrainQueuePausedIsReported covers the other way a queue can be quiet but
+// non-empty: a paused queue dispatches nothing at all, so every message stays.
+func TestDrainQueuePausedIsReported(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("paused-remainder")
+	if _, err := c.Produce(ctx, queue, GenerateMessages(25, 256, nil), 0, 1, nil); err != nil {
+		t.Fatalf("produce: %v", err)
+	}
+	if _, err := c.callManagement(ctx, "queue."+queue, "pause", "[]"); err != nil {
+		t.Fatalf("pause queue: %v", err)
+	}
+	defer func() {
+		_, _ = c.callManagement(context.Background(), "queue."+queue, "resume", "[]")
+		_, _ = c.PurgeQueue(context.Background(), queue)
+	}()
+
+	n, err := c.DrainQueue(ctx, queue, &sliceSink{}, 2*time.Second, 100)
+
+	var pde *PartialDrainError
+	if !errors.As(err, &pde) {
+		t.Fatalf("drain returned n=%d err=%v, want *PartialDrainError", n, err)
+	}
+	if !pde.Stat.Paused {
+		t.Fatalf("reported Stat.Paused = false, want true; stat=%+v", pde.Stat)
+	}
+	if pde.Stat.MessageCount != 25 {
+		t.Fatalf("reported MessageCount = %d, want 25", pde.Stat.MessageCount)
+	}
+}
+
+// TestDrainQueueFullyDrainsWhenTrulyEmpty guards the other direction: the new
+// broker-verified completion check must not turn an ordinary complete drain
+// into an error.
+func TestDrainQueueFullyDrainsWhenTrulyEmpty(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("fully-drains")
+	if _, err := c.Produce(ctx, queue, GenerateMessages(750, 256, nil), 0, 4, nil); err != nil {
+		t.Fatalf("produce: %v", err)
+	}
+
+	sink := &sliceSink{}
+	n, err := c.DrainQueue(ctx, queue, sink, 2*time.Second, 100)
+	if err != nil {
+		t.Fatalf("drain: %v", err)
+	}
+	if n != 750 || len(sink.recs) != 750 {
+		t.Fatalf("drained %d (sink %d), want 750", n, len(sink.recs))
+	}
+	stat, err := c.QueueStatByName(ctx, queue)
+	if err != nil && !errors.Is(err, ErrQueueNotFound) {
+		t.Fatalf("stat: %v", err)
+	}
+	if err == nil && stat.MessageCount != 0 {
+		t.Fatalf("broker still holds %d messages after a nil-error drain", stat.MessageCount)
+	}
+}
+
+// TestQueueStatByNameReportsCounters pins the decoding of the counters the
+// drain's completion check relies on, against a real broker reply.
+func TestQueueStatByNameReportsCounters(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("stat-counters")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	if _, err := c.Produce(ctx, queue, GenerateMessages(10, 128, nil), 0, 1, nil); err != nil {
+		t.Fatalf("produce now: %v", err)
+	}
+	later := GenerateMessages(15, 128, nil)
+	scheduleAt(later, time.Now().Add(time.Hour))
+	if _, err := c.Produce(ctx, queue, later, 0, 1, nil); err != nil {
+		t.Fatalf("produce scheduled: %v", err)
+	}
+
+	stat, err := c.QueueStatByName(ctx, queue)
+	if err != nil {
+		t.Fatalf("QueueStatByName: %v", err)
+	}
+	if stat.Name != queue || stat.MessageCount != 25 || stat.ScheduledCount != 15 {
+		t.Fatalf("stat = %+v, want name=%s messageCount=25 scheduledCount=15", stat, queue)
+	}
+	if stat.Paused {
+		t.Fatalf("stat.Paused = true on a live queue: %+v", stat)
+	}
+	if got := stat.DeliverableNow(); got != 10 {
+		t.Fatalf("DeliverableNow() = %d, want 10", got)
+	}
+}
+
+// TestCountMessagesScansTheQueue pins CountMessages against a real broker.
+// Unlike QueueStat.MessageCount it walks the queue rather than reading a
+// counter, which is what lets a stalled drain tell a real backlog apart from a
+// drifted counter advertising messages that do not exist.
+func TestCountMessagesScansTheQueue(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("count-scan")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	if _, err := c.Produce(ctx, queue, GenerateMessages(12, 128, nil), 0, 1, nil); err != nil {
+		t.Fatalf("produce: %v", err)
+	}
+	// Scheduled messages are still messages on the queue: the scan counts them
+	// even though a consumer cannot receive them yet.
+	later := GenerateMessages(3, 128, nil)
+	scheduleAt(later, time.Now().Add(time.Hour))
+	if _, err := c.Produce(ctx, queue, later, 0, 1, nil); err != nil {
+		t.Fatalf("produce scheduled: %v", err)
+	}
+
+	got, err := c.CountMessages(ctx, queue)
+	if err != nil {
+		t.Fatalf("CountMessages: %v", err)
+	}
+	if got != 15 {
+		t.Fatalf("CountMessages = %d, want 15", got)
+	}
+	// The whole point is that the scan agrees with the counter on a healthy
+	// queue; only then does a disagreement mean drift.
+	stat, err := c.QueueStatByName(ctx, queue)
+	if err != nil {
+		t.Fatalf("QueueStatByName: %v", err)
+	}
+	if stat.MessageCount != got {
+		t.Fatalf("scan found %d but messageCount reports %d on a healthy queue", got, stat.MessageCount)
+	}
+}
+
+// TestQueueStatByNameMissingQueue pins the not-found path, which DrainQueue
+// treats as "the queue was auto-deleted once emptied", not as a failure.
+func TestQueueStatByNameMissingQueue(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	if _, err := c.QueueStatByName(ctx, "no-such-queue-9f3a"); !errors.Is(err, ErrQueueNotFound) {
+		t.Fatalf("err = %v, want ErrQueueNotFound", err)
+	}
+}
+
+// TestPurgeQueueRemovesScheduled pins that purge clears what a drain cannot,
+// which is what resetBroker relies on to give each test a clean slate.
+func TestPurgeQueueRemovesScheduled(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("purge-scheduled")
+	msgs := GenerateMessages(30, 128, nil)
+	scheduleAt(msgs, time.Now().Add(time.Hour))
+	if _, err := c.Produce(ctx, queue, msgs, 0, 1, nil); err != nil {
+		t.Fatalf("produce scheduled: %v", err)
+	}
+
+	removed, err := c.PurgeQueue(ctx, queue)
+	if err != nil {
+		t.Fatalf("purge: %v", err)
+	}
+	if removed != 30 {
+		t.Fatalf("purge removed %d, want 30", removed)
+	}
+	stat, err := c.QueueStatByName(ctx, queue)
+	if err != nil && !errors.Is(err, ErrQueueNotFound) {
+		t.Fatalf("stat: %v", err)
+	}
+	if err == nil && stat.MessageCount != 0 {
+		t.Fatalf("queue still holds %d after purge", stat.MessageCount)
+	}
+}
diff --git a/internal/broker/drain_verify_test.go b/internal/broker/drain_verify_test.go
new file mode 100644
index 0000000..8465714
--- /dev/null
+++ b/internal/broker/drain_verify_test.go
@@ -0,0 +1,229 @@
+package broker
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestDeliverableNow(t *testing.T) {
+	tests := []struct {
+		name string
+		stat QueueStat
+		want int64
+	}{
+		{"empty queue", QueueStat{}, 0},
+		{"all deliverable", QueueStat{MessageCount: 10}, 10},
+		{"scheduled held back", QueueStat{MessageCount: 10, ScheduledCount: 6}, 4},
+		{"in flight to another consumer", QueueStat{MessageCount: 10, DeliveringCount: 3, ConsumerCount: 1}, 7},
+		// No consumer holds these: the broker is returning them to the queue.
+		{"in flight with no consumer is settlement lag", QueueStat{MessageCount: 10, DeliveringCount: 10}, 10},
+		{"scheduled and delivering", QueueStat{MessageCount: 10, ScheduledCount: 6, DeliveringCount: 3, ConsumerCount: 1}, 1},
+		{"wholly scheduled", QueueStat{MessageCount: 10, ScheduledCount: 10}, 0},
+		{"paused ignores counters", QueueStat{MessageCount: 10, Paused: true}, 0},
+		// The counters are sampled independently and can overlap in flight.
+		{"overlapping counters never go negative", QueueStat{MessageCount: 5, ScheduledCount: 4, DeliveringCount: 4, ConsumerCount: 1}, 0},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := tc.stat.DeliverableNow(); got != tc.want {
+				t.Fatalf("DeliverableNow() = %d, want %d", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestDrainOutcome(t *testing.T) {
+	tests := []struct {
+		name      string
+		stat      QueueStat
+		fruitless int
+		want      outcome
+	}{
+		{
+			name: "empty queue is complete",
+			stat: QueueStat{MessageCount: 0},
+			want: drainComplete,
+		},
+		{
+			name:      "empty queue is complete even after fruitless passes",
+			stat:      QueueStat{MessageCount: 0},
+			fruitless: maxFruitlessPasses,
+			want:      drainComplete,
+		},
+		{
+			name: "deliverable remainder after progress means the gap was a stall",
+			stat: QueueStat{MessageCount: 500},
+			want: drainRetry,
+		},
+		{
+			name:      "one fruitless pass is tolerated: acks may still be settling",
+			stat:      QueueStat{MessageCount: 500},
+			fruitless: 1,
+			want:      drainRetry,
+		},
+		{
+			name:      "a queue that keeps promising messages it never delivers gives up",
+			stat:      QueueStat{MessageCount: 500},
+			fruitless: maxFruitlessPasses,
+			want:      drainStuck,
+		},
+		{
+			name: "in-flight remainder with no consumer is retried, not reported stuck",
+			stat: QueueStat{MessageCount: 50, DeliveringCount: 50},
+			want: drainRetry,
+		},
+		{
+			name: "scheduled remainder is stuck, retrying cannot help",
+			stat: QueueStat{MessageCount: 600, ScheduledCount: 600},
+			want: drainStuck,
+		},
+		{
+			name: "remainder in flight to another consumer is stuck",
+			stat: QueueStat{MessageCount: 50, DeliveringCount: 50, ConsumerCount: 2},
+			want: drainStuck,
+		},
+		{
+			name: "paused queue is stuck",
+			stat: QueueStat{MessageCount: 50, Paused: true},
+			want: drainStuck,
+		},
+		{
+			name: "partly scheduled remainder still has deliverable messages to take",
+			stat: QueueStat{MessageCount: 600, ScheduledCount: 400},
+			want: drainRetry,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := drainOutcome(tc.stat, tc.fruitless); got != tc.want {
+				t.Fatalf("drainOutcome() = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+func TestPartialDrainErrorMessage(t *testing.T) {
+	tests := []struct {
+		name  string
+		err   *PartialDrainError
+		wants []string
+	}{
+		{
+			name: "scheduled",
+			err: &PartialDrainError{
+				Queue:   "DLQ",
+				Drained: 50701,
+				Stat:    QueueStat{Name: "DLQ", MessageCount: 160000, ScheduledCount: 160000},
+			},
+			wants: []string{"incomplete drain of DLQ", "50701", "160000", "scheduled for later"},
+		},
+		{
+			name: "in flight to another consumer",
+			err: &PartialDrainError{
+				Queue:   "orders",
+				Drained: 5,
+				Stat:    QueueStat{Name: "orders", MessageCount: 20, DeliveringCount: 20, ConsumerCount: 3},
+			},
+			wants: []string{"incomplete drain of orders", "in flight to 3 other consumer(s)"},
+		},
+		{
+			name: "paused",
+			err: &PartialDrainError{
+				Queue:   "orders",
+				Drained: 0,
+				Stat:    QueueStat{Name: "orders", MessageCount: 7, Paused: true},
+			},
+			wants: []string{"queue is paused"},
+		},
+		{
+			name: "deliverable but the broker went quiet: no counter explains it",
+			err: &PartialDrainError{
+				Queue:   "orders",
+				Drained: 3,
+				Stat:    QueueStat{Name: "orders", MessageCount: 9},
+			},
+			// No counter explains the remainder, so the message must name the
+			// broker-side stall and point somewhere, not just say it gave up.
+			wants: []string{"all 9 are deliverable now", "broker-side stall", "check the broker log"},
+		},
+		{
+			// The counter advertises a backlog a scan cannot find: the messages
+			// are not stuck, they are not there. Must not read as a stall.
+			name: "counter drift: scan finds nothing the counter promised",
+			err: &PartialDrainError{
+				Queue:   "DLQ",
+				Drained: 0,
+				Stat:    QueueStat{Name: "DLQ", MessageCount: 158782},
+				Scanned: true,
+				Counted: 0,
+			},
+			wants: []string{
+				"messageCount reports 158782 but a countMessages scan finds 0",
+				"counter has drifted",
+				"do not exist",
+				"restart the broker",
+			},
+		},
+		{
+			name: "scan confirms the backlog is real: broker is stalled",
+			err: &PartialDrainError{
+				Queue:   "DLQ",
+				Drained: 0,
+				Stat:    QueueStat{Name: "DLQ", MessageCount: 158782},
+				Scanned: true,
+				Counted: 158782,
+			},
+			wants: []string{"a scan confirms 158782 real message(s)", "broker-side stall"},
+		},
+		{
+			// The scan itself failed, so Counted says nothing: fall back to the
+			// plain stall wording rather than claim drift on a zero value.
+			name: "no scan ran: falls back to the stall wording",
+			err: &PartialDrainError{
+				Queue:   "DLQ",
+				Drained: 0,
+				Stat:    QueueStat{Name: "DLQ", MessageCount: 9},
+				Scanned: false,
+			},
+			wants: []string{"all 9 are deliverable now", "broker-side stall"},
+		},
+		{
+			// A real reason always wins over the scan-derived wording.
+			name: "scheduled remainder still reports the counter reason",
+			err: &PartialDrainError{
+				Queue:   "orders",
+				Drained: 2,
+				Stat:    QueueStat{Name: "orders", MessageCount: 5, ScheduledCount: 5},
+				Scanned: true,
+				Counted: 5,
+			},
+			wants: []string{"5 scheduled for later delivery"},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got := tc.err.Error()
+			for _, want := range tc.wants {
+				if !strings.Contains(got, want) {
+					t.Fatalf("Error() = %q, want it to mention %q", got, want)
+				}
+			}
+		})
+	}
+}
+
+// A PartialDrainError must survive errors.Join + errors.As, which is how
+// DrainAll reports several bad queues at once and how export detects one.
+func TestPartialDrainErrorIsUnwrappable(t *testing.T) {
+	pde := &PartialDrainError{Queue: "DLQ", Drained: 1, Stat: QueueStat{MessageCount: 2}}
+	joined := errors.Join(errors.New("other queue failed"), pde)
+
+	var got *PartialDrainError
+	if !errors.As(joined, &got) {
+		t.Fatal("errors.As did not find *PartialDrainError in a joined error")
+	}
+	if got.Queue != "DLQ" {
+		t.Fatalf("unwrapped Queue = %q, want DLQ", got.Queue)
+	}
+}
diff --git a/internal/broker/drift.go b/internal/broker/drift.go
new file mode 100644
index 0000000..3258311
--- /dev/null
+++ b/internal/broker/drift.go
@@ -0,0 +1,152 @@
+package broker
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// DriftVerdict is what a drift check concluded about one queue.
+type DriftVerdict int
+
+const (
+	// DriftNone: the scan and the counter agree. The depth is real.
+	DriftNone DriftVerdict = iota
+	// DriftConfirmed: the counter is stable and still disagrees with a scan of
+	// the queue. The counter is advertising messages that are not there.
+	DriftConfirmed
+	// DriftInconclusive: the queue's depth moved while it was being checked, so
+	// the counter and the scan were sampled against different queues and their
+	// disagreement proves nothing.
+	DriftInconclusive
+)
+
+func (v DriftVerdict) String() string {
+	switch v {
+	case DriftConfirmed:
+		return "DRIFT"
+	case DriftInconclusive:
+		return "inconclusive"
+	default:
+		return "ok"
+	}
+}
+
+// DriftReport is one queue's drift check: the counter read on either side of a
+// scan, and what the scan found in between.
+//
+// The counter is read twice on purpose. A drift check compares two numbers
+// sampled at different instants, so on a queue with live producers or consumers
+// they disagree simply because the queue moved -- which would report drift on
+// every healthy busy queue. Bracketing the scan with two counter reads
+// distinguishes the two: only a counter that did not move while the scan ran
+// can be meaningfully compared against it.
+type DriftReport struct {
+	Queue string
+
+	// CounterBefore and CounterAfter are QueueStat.MessageCount either side of
+	// the scan; Counted is what the scan found.
+	CounterBefore int64
+	Counted       int64
+	CounterAfter  int64
+}
+
+// Verdict classifies the report. Missing is the magnitude of a confirmed drift:
+// how many messages the counter claims that do not exist.
+func (r DriftReport) Verdict() DriftVerdict {
+	if r.CounterBefore != r.CounterAfter {
+		return DriftInconclusive
+	}
+	if r.CounterBefore == r.Counted {
+		return DriftNone
+	}
+	return DriftConfirmed
+}
+
+// Missing is how many messages the counter over-reports. It is only meaningful
+// for a DriftConfirmed report; it is 0 otherwise, including when a scan finds
+// MORE than the counter claims (an under-reporting counter is a different bug
+// and does not strand messages, so it is not what this reports).
+func (r DriftReport) Missing() int64 {
+	if r.Verdict() != DriftConfirmed {
+		return 0
+	}
+	if n := r.CounterBefore - r.Counted; n > 0 {
+		return n
+	}
+	return 0
+}
+
+// CheckDrift reports whether a queue's message counter agrees with a scan of
+// the queue itself.
+//
+// This detects the failure that makes a queue undrainable: for a paged queue
+// the broker derives messageCount from page-counter journal records, and when
+// those drift the queue advertises a depth whose messages do not exist. It
+// reports a large backlog, refuses to deliver any of it, and no export can ever
+// empty it -- the only repair is a broker restart, which rebuilds the counter
+// from the journal.
+//
+// It is deliberately not free: the scan walks the queue, so this costs far more
+// than reading the counter and is meant to be run when a queue looks wrong, not
+// on every status.
+func (c *Client) CheckDrift(ctx context.Context, name string) (DriftReport, error) {
+	before, err := c.QueueStatByName(ctx, name)
+	if err != nil {
+		return DriftReport{}, err
+	}
+	counted, err := c.CountMessages(ctx, name)
+	if err != nil {
+		return DriftReport{}, err
+	}
+	after, err := c.QueueStatByName(ctx, name)
+	if err != nil {
+		return DriftReport{}, err
+	}
+	return DriftReport{
+		Queue:         name,
+		CounterBefore: before.MessageCount,
+		Counted:       counted,
+		CounterAfter:  after.MessageCount,
+	}, nil
+}
+
+// isQueueGone reports whether err means "that queue no longer exists".
+//
+// The two calls a drift check makes report it differently: listQueues simply
+// does not return the queue, which QueueStatByName turns into ErrQueueNotFound,
+// while a management op against a queue that is gone is REJECTED by the broker
+// with "Cannot find resource with name queue.X" -- an error with no structure to
+// match on, hence the string. Matching it is worth it: an auto-delete race is
+// routine and must not fail a whole drift sweep.
+func isQueueGone(err error) bool {
+	if errors.Is(err, ErrQueueNotFound) {
+		return true
+	}
+	return strings.Contains(err.Error(), "Cannot find resource")
+}
+
+// CheckDriftAll runs CheckDrift over every user queue.
+//
+// A queue that vanishes mid-check (auto-deleted once emptied) is skipped rather
+// than failing the run: it holds nothing, so it cannot be hiding a drifted
+// backlog.
+func (c *Client) CheckDriftAll(ctx context.Context) ([]DriftReport, error) {
+	queues, err := c.ListQueues(ctx)
+	if err != nil {
+		return nil, err
+	}
+	reports := make([]DriftReport, 0, len(queues))
+	for _, q := range queues {
+		r, err := c.CheckDrift(ctx, q.Name)
+		if err != nil {
+			if isQueueGone(err) {
+				continue
+			}
+			return nil, fmt.Errorf("check drift on %s: %w", q.Name, err)
+		}
+		reports = append(reports, r)
+	}
+	return reports, nil
+}
diff --git a/internal/broker/drift_integration_test.go b/internal/broker/drift_integration_test.go
new file mode 100644
index 0000000..7801be3
--- /dev/null
+++ b/internal/broker/drift_integration_test.go
@@ -0,0 +1,126 @@
+package broker
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// TestCheckDriftOnHealthyQueue pins the direction that matters most: a drift
+// check must NOT cry drift on a queue whose counter is fine. A detector that
+// reports drift on healthy queues is worse than none, because it trains the
+// operator to ignore the one queue that has actually drifted.
+func TestCheckDriftOnHealthyQueue(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("drift-healthy")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	if _, err := c.Produce(ctx, queue, GenerateMessages(200, 128, nil), 0, 4, nil); err != nil {
+		t.Fatalf("produce: %v", err)
+	}
+
+	r, err := c.CheckDrift(ctx, queue)
+	if err != nil {
+		t.Fatalf("CheckDrift: %v", err)
+	}
+	if r.Verdict() != DriftNone {
+		t.Fatalf("verdict = %v on a healthy queue, want DriftNone; report=%+v", r.Verdict(), r)
+	}
+	if r.Counted != 200 || r.CounterBefore != 200 {
+		t.Fatalf("report = %+v, want counter and scan both 200", r)
+	}
+	if r.Missing() != 0 {
+		t.Fatalf("Missing() = %d on a healthy queue, want 0", r.Missing())
+	}
+}
+
+// TestCheckDriftCountsMessagesNoConsumerCanTake guards against a false positive
+// that would otherwise be easy to ship: scheduled messages are real messages on
+// the queue that no consumer can receive yet. The scan counts them and so does
+// the counter, so this is NOT drift -- only DeliverableNow() should exclude them.
+func TestCheckDriftCountsMessagesNoConsumerCanTake(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("drift-scheduled")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	msgs := GenerateMessages(40, 128, nil)
+	scheduleAt(msgs, time.Now().Add(time.Hour))
+	if _, err := c.Produce(ctx, queue, msgs, 0, 1, nil); err != nil {
+		t.Fatalf("produce scheduled: %v", err)
+	}
+
+	r, err := c.CheckDrift(ctx, queue)
+	if err != nil {
+		t.Fatalf("CheckDrift: %v", err)
+	}
+	if r.Verdict() != DriftNone {
+		t.Fatalf("verdict = %v on a wholly scheduled queue, want DriftNone; report=%+v", r.Verdict(), r)
+	}
+	if r.Counted != 40 {
+		t.Fatalf("scan counted %d, want 40: scheduled messages are still on the queue", r.Counted)
+	}
+}
+
+// TestCheckDriftAllSkipsInternalQueues pins that a drift sweep covers the user's
+// queues and does not fail on the broker's own.
+func TestCheckDriftAllSkipsInternalQueues(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemis(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+	c, err := Connect(ctx, props)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close(ctx)
+
+	queue := benchQueue("drift-sweep")
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+	if _, err := c.Produce(ctx, queue, GenerateMessages(10, 128, nil), 0, 1, nil); err != nil {
+		t.Fatalf("produce: %v", err)
+	}
+
+	reports, err := c.CheckDriftAll(ctx)
+	if err != nil {
+		t.Fatalf("CheckDriftAll: %v", err)
+	}
+	var found bool
+	for _, r := range reports {
+		if r.Queue == queue {
+			found = true
+			if r.Verdict() != DriftNone {
+				t.Fatalf("verdict = %v for %s, want DriftNone; report=%+v", r.Verdict(), queue, r)
+			}
+		}
+	}
+	if !found {
+		t.Fatalf("sweep of %d queue(s) did not include %s", len(reports), queue)
+	}
+}
diff --git a/internal/broker/drift_test.go b/internal/broker/drift_test.go
new file mode 100644
index 0000000..7d73e2c
--- /dev/null
+++ b/internal/broker/drift_test.go
@@ -0,0 +1,113 @@
+package broker
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+)
+
+func TestDriftReportVerdict(t *testing.T) {
+	tests := []struct {
+		name        string
+		report      DriftReport
+		want        DriftVerdict
+		wantMissing int64
+	}{
+		{
+			name:   "counter agrees with the scan",
+			report: DriftReport{CounterBefore: 500, Counted: 500, CounterAfter: 500},
+			want:   DriftNone,
+		},
+		{
+			name:   "empty queue agrees trivially",
+			report: DriftReport{},
+			want:   DriftNone,
+		},
+		{
+			// The incident: the counter advertises a backlog no scan can find.
+			name:        "stable counter promising messages a scan cannot find",
+			report:      DriftReport{CounterBefore: 158782, Counted: 0, CounterAfter: 158782},
+			want:        DriftConfirmed,
+			wantMissing: 158782,
+		},
+		{
+			name:        "partial drift still reports the gap",
+			report:      DriftReport{CounterBefore: 1000, Counted: 400, CounterAfter: 1000},
+			want:        DriftConfirmed,
+			wantMissing: 600,
+		},
+		{
+			// A live queue moves between the samples, so the counter and the scan
+			// disagree for an ordinary reason. Reporting drift here would light up
+			// every healthy queue that has a consumer attached.
+			name:   "queue draining under us is inconclusive, not drift",
+			report: DriftReport{CounterBefore: 1000, Counted: 940, CounterAfter: 880},
+			want:   DriftInconclusive,
+		},
+		{
+			name:   "queue filling under us is inconclusive",
+			report: DriftReport{CounterBefore: 100, Counted: 130, CounterAfter: 160},
+			want:   DriftInconclusive,
+		},
+		{
+			// The counter under-reporting does not strand messages: the scan can
+			// still find them and a drain can still take them.
+			name:        "scan finding more than the counter claims reports no missing messages",
+			report:      DriftReport{CounterBefore: 10, Counted: 25, CounterAfter: 10},
+			want:        DriftConfirmed,
+			wantMissing: 0,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := tc.report.Verdict(); got != tc.want {
+				t.Fatalf("Verdict() = %v, want %v", got, tc.want)
+			}
+			if got := tc.report.Missing(); got != tc.wantMissing {
+				t.Fatalf("Missing() = %d, want %d", got, tc.wantMissing)
+			}
+		})
+	}
+}
+
+func TestDriftVerdictString(t *testing.T) {
+	tests := []struct {
+		verdict DriftVerdict
+		want    string
+	}{
+		{DriftNone, "ok"},
+		{DriftConfirmed, "DRIFT"},
+		{DriftInconclusive, "inconclusive"},
+	}
+	for _, tc := range tests {
+		if got := tc.verdict.String(); got != tc.want {
+			t.Fatalf("String() = %q, want %q", got, tc.want)
+		}
+	}
+}
+
+// A queue auto-deleted mid-sweep must not fail a whole drift check, and the two
+// calls a check makes report it in two different ways.
+func TestIsQueueGone(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"QueueStatByName not-found", fmt.Errorf("%w: orders", ErrQueueNotFound), true},
+		{"wrapped not-found", fmt.Errorf("check drift: %w", fmt.Errorf("%w: orders", ErrQueueNotFound)), true},
+		{
+			name: "broker rejection for a missing management resource",
+			err:  errors.New(`broker rejected queue.orders.countMessages: Cannot find resource with name queue.orders`),
+			want: true,
+		},
+		{"an unrelated failure is not a missing queue", errors.New("connection reset by peer"), false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := isQueueGone(tc.err); got != tc.want {
+				t.Fatalf("isQueueGone(%v) = %t, want %t", tc.err, got, tc.want)
+			}
+		})
+	}
+}
diff --git a/internal/broker/health_integration_test.go b/internal/broker/health_integration_test.go
index 6939348..d69fe25 100644
--- a/internal/broker/health_integration_test.go
+++ b/internal/broker/health_integration_test.go
@@ -22,8 +22,14 @@ func TestCheckHealthIntegration(t *testing.T) {
 	if err != nil {
 		t.Fatalf("health: %v", err)
 	}
-	if h.Verdict != OK {
-		t.Fatalf("fresh broker should be OK, got %s (disk %.1f mem %.1f)", h.Verdict, h.DiskUsagePct, h.MemoryUsagePct)
+	// DiskUsagePct reflects the real filesystem holding the broker's data dir,
+	// which on a dev host is a shared partition this test does not control. A
+	// busy disk (>=70%) legitimately yields DEGRADED on an otherwise-fresh
+	// broker, so we do NOT assert Verdict == OK. What must always hold for a
+	// fresh broker is that it is not in the CRITICAL band (disk/mem >90% or
+	// producers blocked).
+	if h.Verdict == Critical {
+		t.Fatalf("fresh broker should not be CRITICAL, got %s (disk %.1f mem %.1f blocking %v)", h.Verdict, h.DiskUsagePct, h.MemoryUsagePct, h.Blocking)
 	}
 	// I1: a fresh broker with a near-empty disk is well below max-disk-usage,
 	// so the disk-full producer-block indicator must be false.
diff --git a/internal/broker/management.go b/internal/broker/management.go
index e36157d..d99aaf7 100644
--- a/internal/broker/management.go
+++ b/internal/broker/management.go
@@ -9,12 +9,57 @@ import (
 	"github.com/Azure/go-amqp"
 )
 
-// QueueStat is one queue's name and current message depth, as reported by the
-// broker's listQueues management operation. MessageCount arrives as a
-// JSON string and is decoded into an int64.
+// QueueStat is one queue's depth and delivery state, as reported by the
+// broker's listQueues management operation. Every counter arrives as a JSON
+// string and is decoded into its Go type.
+//
+// MessageCount is the total depth, and it is NOT the number of messages a
+// consumer can receive right now: it also counts messages the broker will not
+// hand out yet. ScheduledCount (scheduled delivery / redelivery delay, not due
+// yet) and DeliveringCount (already dispatched to some consumer, awaiting
+// settlement) are both included in it, and a Paused queue dispatches nothing at
+// all regardless of the counters. DeliverableNow is the subset a drain can
+// actually expect to receive.
 type QueueStat struct {
-	Name         string `json:"name"`
-	MessageCount int64  `json:"messageCount,string"`
+	Name            string `json:"name"`
+	MessageCount    int64  `json:"messageCount,string"`
+	ScheduledCount  int64  `json:"scheduledCount,string"`
+	DeliveringCount int64  `json:"deliveringCount,string"`
+	ConsumerCount   int64  `json:"consumerCount,string"`
+	Paused          bool   `json:"paused,string"`
+
+	// Temporary and Internal are the broker's own classification of the queue.
+	// They are how ListQueues recognises queues that are not the user's data
+	// (dynamic reply queues, internal bookkeeping) without having to guess from
+	// the name.
+	Temporary bool `json:"temporary,string"`
+	Internal  bool `json:"internalQueue,string"`
+}
+
+// DeliverableNow reports how many of the queue's messages the broker would
+// hand to a consumer right now: the depth minus the messages it is holding
+// back. A paused queue delivers nothing, so it always reports 0.
+//
+// DeliveringCount only counts against the total while some consumer is
+// attached to hold those messages. With no consumers, an in-flight count is
+// just settlement lag -- typically our own just-closed receiver, whose unacked
+// messages the broker is in the middle of returning to the queue -- and those
+// messages ARE coming back, so they must not be mistaken for messages we can
+// never have.
+func (q QueueStat) DeliverableNow() int64 {
+	if q.Paused {
+		return 0
+	}
+	n := q.MessageCount - q.ScheduledCount
+	if q.ConsumerCount > 0 {
+		n -= q.DeliveringCount
+	}
+	if n < 0 {
+		// The counters are sampled independently and can overlap in flight;
+		// never report a negative backlog.
+		return 0
+	}
+	return n
 }
 
 // callManagement performs a request/reply against activemq.management.
@@ -67,7 +112,98 @@ func (c *Client) ListQueues(ctx context.Context) ([]QueueStat, error) {
 	return parseQueueStatsReply(reply)
 }
 
+// ErrQueueNotFound is returned by QueueStatByName when the broker does not
+// list the named queue, which normally means it has been auto-deleted.
+var ErrQueueNotFound = fmt.Errorf("queue not found on broker")
+
+// QueueStatByName returns the broker's current stats for a single queue. It
+// asks the broker to filter by exact name rather than listing every queue, so
+// it stays cheap enough to call repeatedly during a drain. Unlike ListQueues it
+// does not filter internal queues: the caller already named the queue it wants.
+func (c *Client) QueueStatByName(ctx context.Context, name string) (QueueStat, error) {
+	filter, err := json.Marshal(map[string]string{
+		"field": "name", "operation": "EQUALS", "value": name,
+		"sortField": "messageCount", "sortOrder": "desc",
+	})
+	if err != nil {
+		return QueueStat{}, fmt.Errorf("marshal queue filter: %w", err)
+	}
+	body, err := json.Marshal([]interface{}{string(filter), 1, 1})
+	if err != nil {
+		return QueueStat{}, fmt.Errorf("marshal listQueues args: %w", err)
+	}
+	reply, err := c.callManagement(ctx, "broker", "listQueues", string(body))
+	if err != nil {
+		return QueueStat{}, err
+	}
+	stats, err := parseQueueStatsRaw(reply)
+	if err != nil {
+		return QueueStat{}, err
+	}
+	for _, q := range stats {
+		if q.Name == name {
+			return q, nil
+		}
+	}
+	return QueueStat{}, fmt.Errorf("%w: %s", ErrQueueNotFound, name)
+}
+
+// PurgeQueue removes every message from a queue, including messages a consumer
+// cannot receive (scheduled, or held for redelivery). It returns the number of
+// messages removed. This is destructive and does not persist anything: it is
+// for resetting a queue, not for exporting it.
+func (c *Client) PurgeQueue(ctx context.Context, name string) (int64, error) {
+	reply, err := c.callManagement(ctx, "queue."+name, "removeAllMessages", "[]")
+	if err != nil {
+		return 0, fmt.Errorf("purge %s: %w", name, err)
+	}
+	return parseCountReply(reply), nil
+}
+
+// CountMessages returns how many messages a scan of the queue actually finds.
+//
+// This is NOT the same as QueueStat.MessageCount. MessageCount is a counter the
+// broker maintains (for a paged queue, derived from page-counter journal
+// records); countMessages walks the queue itself. When the two disagree, the
+// counter has drifted and reports messages that do not exist -- a queue that
+// advertises a large depth, refuses to deliver anything, and can never be
+// drained. Comparing the two is the only way to tell that apart from a broker
+// that genuinely holds messages but has stalled.
+func (c *Client) CountMessages(ctx context.Context, name string) (int64, error) {
+	reply, err := c.callManagement(ctx, "queue."+name, "countMessages", "[]")
+	if err != nil {
+		return 0, fmt.Errorf("count messages on %s: %w", name, err)
+	}
+	return parseCountReply(reply), nil
+}
+
+// parseCountReply decodes the count-shaped management reply the broker returns
+// for countMessages and removeAllMessages: a JSON array holding a single number,
+// encoded as a string ("[12]").
+func parseCountReply(reply *amqp.Message) int64 {
+	raw, ok := reply.Value.(string)
+	if !ok {
+		return 0
+	}
+	var outer []int64
+	if err := json.Unmarshal([]byte(raw), &outer); err != nil || len(outer) == 0 {
+		return 0
+	}
+	return outer[0]
+}
+
 func parseQueueStatsReply(reply *amqp.Message) ([]QueueStat, error) {
+	stats, err := parseQueueStatsRaw(reply)
+	if err != nil {
+		return nil, err
+	}
+	return filterInternalQueues(stats), nil
+}
+
+// parseQueueStatsRaw decodes a listQueues reply without filtering internal
+// queues, so a caller asking for one queue by name gets it back whatever it is
+// called.
+func parseQueueStatsRaw(reply *amqp.Message) ([]QueueStat, error) {
 	strVal, ok := reply.Value.(string)
 	if !ok {
 		return nil, fmt.Errorf("unexpected management response format")
@@ -85,13 +221,25 @@ func parseQueueStatsReply(reply *amqp.Message) ([]QueueStat, error) {
 	if err := json.Unmarshal([]byte(outer[0]), &paged); err != nil {
 		return nil, fmt.Errorf("parse paged response: %w", err)
 	}
-	return filterInternalQueues(paged.Data), nil
+	return paged.Data, nil
 }
 
+// filterInternalQueues drops the queues an export has no business draining:
+// the broker's own internal and temporary queues (including the dynamic reply
+// queue callManagement opens for every management call).
+//
+// It trusts the broker's temporary/internalQueue flags rather than the shape of
+// the name. An earlier version also skipped every 36-character name to catch
+// UUID-named reply queues, which silently excluded any real user queue whose
+// name happened to be 36 characters long -- an export that quietly skips a
+// queue is exactly the kind of silent data loss this tool exists to prevent.
 func filterInternalQueues(in []QueueStat) []QueueStat {
 	var out []QueueStat
 	for _, q := range in {
-		if strings.HasPrefix(q.Name, "activemq.") || strings.HasPrefix(q.Name, "$") || len(q.Name) == 36 {
+		if q.Temporary || q.Internal {
+			continue
+		}
+		if strings.HasPrefix(q.Name, "activemq.") || strings.HasPrefix(q.Name, "$") {
 			continue
 		}
 		out = append(out, q)
diff --git a/internal/broker/management_integration_test.go b/internal/broker/management_integration_test.go
index 0f493ae..44fe6c6 100644
--- a/internal/broker/management_integration_test.go
+++ b/internal/broker/management_integration_test.go
@@ -77,8 +77,13 @@ func resetBroker(t testing.TB, props ConnectionProps) {
 		t.Fatalf("reset list queues: %v", err)
 	}
 	for _, q := range qs {
-		if _, err := c.DrainQueue(ctx, q.Name, discardSink{}, 500*time.Millisecond, 200); err != nil {
-			t.Fatalf("reset drain %s: %v", q.Name, err)
+		// Purge rather than drain: removeAllMessages also clears messages a
+		// consumer cannot receive (scheduled, held for redelivery), which a
+		// drain leaves behind -- and which DrainQueue now correctly reports as
+		// an incomplete drain, so a single leftover scheduled message from an
+		// earlier test would fail every later one.
+		if _, err := c.PurgeQueue(ctx, q.Name); err != nil {
+			t.Fatalf("reset purge %s: %v", q.Name, err)
 		}
 	}
 }
diff --git a/internal/broker/management_test.go b/internal/broker/management_test.go
index a4be2ca..3104bdb 100644
--- a/internal/broker/management_test.go
+++ b/internal/broker/management_test.go
@@ -79,11 +79,23 @@ func TestFilterInternalQueues(t *testing.T) {
 		{Name: "orders", MessageCount: 5},
 		{Name: "activemq.notifications", MessageCount: 1},
 		{Name: "$sys.foo", MessageCount: 1},
-		{Name: "123e4567-e89b-12d3-a456-426614174000", MessageCount: 1}, // 36 chars
+		// The dynamic reply queue callManagement opens: the broker flags it
+		// temporary, which is what we filter on.
+		{Name: "123e4567-e89b-12d3-a456-426614174000", MessageCount: 1, Temporary: true},
+		{Name: "internal.bookkeeping", MessageCount: 1, Internal: true},
+		// A real user queue that is 36 characters long must survive: an export
+		// that silently skips a queue loses every message on it.
+		{Name: "billing_credit_approved_debit_rows_q", MessageCount: 7},
 		{Name: "payments", MessageCount: 2},
 	}
 	got := filterInternalQueues(in)
-	if len(got) != 2 || got[0].Name != "orders" || got[1].Name != "payments" {
-		t.Fatalf("unexpected filter result: %+v", got)
+	want := []string{"orders", "billing_credit_approved_debit_rows_q", "payments"}
+	if len(got) != len(want) {
+		t.Fatalf("filterInternalQueues returned %d queues, want %d: %+v", len(got), len(want), got)
+	}
+	for i, w := range want {
+		if got[i].Name != w {
+			t.Fatalf("filterInternalQueues()[%d] = %q, want %q", i, got[i].Name, w)
+		}
 	}
 }
diff --git a/internal/cli/commands_integration_test.go b/internal/cli/commands_integration_test.go
index 1151e80..450bfaf 100644
--- a/internal/cli/commands_integration_test.go
+++ b/internal/cli/commands_integration_test.go
@@ -40,6 +40,35 @@ func TestCommandsAgainstBroker(t *testing.T) {
 		if !strings.Contains(out, "orders") {
 			t.Fatalf("status output missing queue: %q", out)
 		}
+		// A depth alone cannot explain why an export leaves messages behind;
+		// these columns are what let an operator tell "the broker is holding
+		// them back" from "the broker is stalled".
+		for _, col := range []string{"MESSAGE COUNT", "DELIVERABLE", "SCHEDULED", "DELIVERING", "CONSUMERS", "PAUSED"} {
+			if !strings.Contains(out, col) {
+				t.Fatalf("status output missing the %q column: %q", col, out)
+			}
+		}
+	})
+
+	// The broker under test is healthy, so --verify must come back clean and
+	// exit 0. The flag exists to single out a queue whose counter lies; one that
+	// flags a healthy broker would just be noise an operator learns to ignore.
+	t.Run("status --verify on a healthy broker", func(t *testing.T) {
+		out, err := run(t, "status", "--verify")
+		if err != nil {
+			t.Fatalf("status --verify on a healthy broker: %v\n%s", err, out)
+		}
+		for _, col := range []string{"COUNTER", "SCANNED", "MISSING", "VERDICT"} {
+			if !strings.Contains(out, col) {
+				t.Fatalf("status --verify output missing the %q column: %q", col, out)
+			}
+		}
+		if !strings.Contains(out, "orders") {
+			t.Fatalf("status --verify output missing queue: %q", out)
+		}
+		if strings.Contains(out, "DRIFT") {
+			t.Fatalf("status --verify reported drift on a healthy broker: %q", out)
+		}
 	})
 
 	t.Run("health", func(t *testing.T) {
diff --git a/internal/cli/export.go b/internal/cli/export.go
index c3d848a..1a85549 100644
--- a/internal/cli/export.go
+++ b/internal/cli/export.go
@@ -48,22 +48,32 @@ func newExportCmd() *cobra.Command {
 				return fmt.Errorf("remove stale checkpoint: %w", err)
 			}
 
-			total, err := c.DrainAll(ctx, w, drainTimeout, batch, func(q string, n int) {
+			total, drainErr := c.DrainAll(ctx, w, drainTimeout, batch, func(q string, n int) {
 				fmt.Fprintf(cmd.OutOrStdout(), "drained %d from %s\n", n, q)
 			})
-			if err != nil {
+			// Whatever was drained is worth keeping even when the drain did not
+			// complete, so fsync before reporting either way.
+			syncErr := w.Sync()
+			if drainErr == nil && syncErr != nil {
+				return syncErr
+			}
+			if drainErr != nil {
 				// SIGINT: drained records are already fsync'd; report a clean
 				// interruption rather than a crash. Re-running export needs a
 				// fresh --out path (an existing non-empty store is refused).
-				if errors.Is(err, context.Canceled) {
-					_ = w.Sync()
+				if errors.Is(drainErr, context.Canceled) {
 					fmt.Fprintf(cmd.OutOrStdout(), "interrupted after %d messages, progress saved to %s\n", total, out)
-					return err
+					return drainErr
 				}
-				return err
-			}
-			if err := w.Sync(); err != nil {
-				return err
+				var pde *broker.PartialDrainError
+				if errors.As(drainErr, &pde) {
+					// The store holds real, replayable messages -- it is just
+					// not the whole queue. Say so on stdout so the count is not
+					// mistaken for a complete export, and still fail: exporting
+					// a subset while exiting 0 is how messages get lost.
+					fmt.Fprintf(cmd.OutOrStdout(), "INCOMPLETE: saved %d messages to %s, but the broker still holds messages\n", total, out)
+				}
+				return drainErr
 			}
 			fmt.Fprintf(cmd.OutOrStdout(), "exported %d messages to %s\n", total, out)
 			return nil
diff --git a/internal/cli/export_integration_test.go b/internal/cli/export_integration_test.go
index 313a977..97aa103 100644
--- a/internal/cli/export_integration_test.go
+++ b/internal/cli/export_integration_test.go
@@ -2,9 +2,18 @@
 package cli
 
 import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
+	"time"
+
+	"github.com/Azure/go-amqp"
+	"github.com/martikan/artemisctl/internal/broker"
 )
 
 func TestExportCommandCreatesStore(t *testing.T) {
@@ -30,3 +39,93 @@ func TestExportCommandCreatesStore(t *testing.T) {
 		t.Fatalf("store not written: %v size=%v", err, fi)
 	}
 }
+
+// TestExportFailsOnPartialDrain pins the export contract that the reported bug
+// broke: when the broker still holds messages the drain could not take, export
+// must NOT report success. Previously it printed "exported N messages" and
+// exited 0 while most of a 210K-deep DLQ stayed on the broker, which is how a
+// partial store gets mistaken for a full one.
+//
+// The undrainable remainder here is a scheduled message: counted in the
+// queue's depth, but not deliverable to any consumer until its delivery time.
+func TestExportFailsOnPartialDrain(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip integration in -short")
+	}
+	props := startArtemisForCLI(t)
+	queue := fmt.Sprintf("export.partial.%d", time.Now().UnixNano())
+
+	seedQueue(t, props, queue, []string{"take-me", "and-me"})
+	seedScheduled(t, props, queue, 3, time.Now().Add(time.Hour))
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	c, err := broker.Connect(ctx, props)
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer c.Close(context.Background())
+	defer func() { _, _ = c.PurgeQueue(context.Background(), queue) }()
+
+	out := filepath.Join(t.TempDir(), "partial.artx")
+	var stdout bytes.Buffer
+	root := NewRootCmd()
+	root.SetOut(&stdout)
+	root.SetArgs([]string{
+		"export", "--out", out,
+		"--url", props.URL, "-u", props.Username, "-p", props.Password,
+		"--drain-timeout", "2s",
+	})
+
+	err = root.Execute()
+	if err == nil {
+		t.Fatalf("export reported success on a partial drain; stdout:\n%s", stdout.String())
+	}
+	var pde *broker.PartialDrainError
+	if !errors.As(err, &pde) {
+		t.Fatalf("export err = %v, want *PartialDrainError", err)
+	}
+	if got := stdout.String(); !strings.Contains(got, "INCOMPLETE") {
+		t.Fatalf("export stdout must flag the partial store; got:\n%s", got)
+	}
+	if strings.Contains(stdout.String(), "exported ") {
+		t.Fatalf("export printed a success headline on a partial drain:\n%s", stdout.String())
+	}
+	// The messages it DID drain must still be durably saved: a failed export is
+	// not a reason to throw away recovered messages.
+	fi, statErr := os.Stat(out)
+	if statErr != nil || fi.Size() <= 5 {
+		t.Fatalf("partial store not written: %v size=%v", statErr, fi)
+	}
+}
+
+// seedScheduled puts n messages on queue that the broker will hold until at,
+// so they count towards the queue's depth but cannot be drained.
+func seedScheduled(t *testing.T, props broker.ConnectionProps, queue string, n int, at time.Time) {
+	t.Helper()
+	ctx := context.Background()
+	conn, err := amqp.Dial(ctx, "amqp://"+props.URL, &amqp.ConnOptions{
+		SASLType: amqp.SASLTypePlain(props.Username, props.Password),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	sess, err := conn.NewSession(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sender, err := sess.NewSender(ctx, queue, &amqp.SenderOptions{TargetCapabilities: []string{"queue"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sender.Close(ctx)
+	for i := 0; i < n; i++ {
+		msg := amqp.NewMessage([]byte(fmt.Sprintf("scheduled-%d", i)))
+		msg.Properties = &amqp.MessageProperties{MessageID: fmt.Sprintf("sched-%d-%d", time.Now().UnixNano(), i)}
+		msg.Annotations = amqp.Annotations{"x-opt-delivery-time": at.UnixMilli()}
+		if err := sender.Send(ctx, msg, nil); err != nil {
+			t.Fatalf("send scheduled: %v", err)
+		}
+	}
+}
diff --git a/internal/cli/salvage_integration_test.go b/internal/cli/salvage_integration_test.go
index aa8b6ce..7fbe97a 100644
--- a/internal/cli/salvage_integration_test.go
+++ b/internal/cli/salvage_integration_test.go
@@ -6,6 +6,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -266,8 +267,17 @@ func TestSalvageE2E(t *testing.T) {
 	dctx, dcancel := context.WithTimeout(context.Background(), 60*time.Second)
 	n, err := c.DrainQueue(dctx, redirectQueue, sink, 1*time.Second, 200)
 	dcancel()
-	if err != nil {
-		t.Fatalf("DrainQueue: %v", err)
+	// The scheduled record stays on the queue, so this drain is by definition
+	// incomplete and DrainQueue must say so rather than return nil: a drain
+	// that leaves messages behind while reporting success is what let an export
+	// silently ship a partial store. The rest of the queue must still have been
+	// drained in full.
+	var pde *broker.PartialDrainError
+	if !errors.As(err, &pde) {
+		t.Fatalf("DrainQueue err = %v, want *PartialDrainError for the undrainable scheduled record", err)
+	}
+	if pde.Stat.MessageCount != 1 || pde.Stat.ScheduledCount != 1 {
+		t.Fatalf("DrainQueue reported %+v, want exactly 1 remaining, scheduled", pde.Stat)
 	}
 	if n != wantDrainable {
 		t.Fatalf("DrainQueue drained %d messages, want %d (%d total minus the not-yet-due scheduled record, which cannot be drained until 2100-01-01)",
diff --git a/internal/cli/status.go b/internal/cli/status.go
index 16ea890..4fb0f1e 100644
--- a/internal/cli/status.go
+++ b/internal/cli/status.go
@@ -10,7 +10,7 @@ import (
 )
 
 func newStatusCmd() *cobra.Command {
-	return &cobra.Command{
+	cmd := &cobra.Command{
 		Use:   "status",
 		Short: "List queues and message counts (descending)",
 		RunE: func(cmd *cobra.Command, _ []string) error {
@@ -21,16 +21,69 @@ func newStatusCmd() *cobra.Command {
 				return err
 			}
 			defer c.Close(ctx)
+			if verify, _ := cmd.Flags().GetBool("verify"); verify {
+				return runStatusVerify(cmd, c)
+			}
 			stats, err := c.ListQueues(ctx)
 			if err != nil {
 				return err
 			}
 			w := tabwriter.NewWriter(cmd.OutOrStdout(), 0, 0, 3, ' ', 0)
-			fmt.Fprintln(w, "QUEUE NAME\tMESSAGE COUNT")
+			// MESSAGE COUNT alone cannot explain why an export leaves messages
+			// behind: the depth also counts messages the broker will not hand to
+			// a consumer. DELIVERABLE is the subset a drain can actually take,
+			// and the remaining columns say where the difference went.
+			fmt.Fprintln(w, "QUEUE NAME\tMESSAGE COUNT\tDELIVERABLE\tSCHEDULED\tDELIVERING\tCONSUMERS\tPAUSED")
 			for _, s := range stats {
-				fmt.Fprintf(w, "%s\t%d\n", s.Name, s.MessageCount)
+				fmt.Fprintf(w, "%s\t%d\t%d\t%d\t%d\t%d\t%t\n",
+					s.Name, s.MessageCount, s.DeliverableNow(),
+					s.ScheduledCount, s.DeliveringCount, s.ConsumerCount, s.Paused)
 			}
 			return w.Flush()
 		},
 	}
+	cmd.Flags().Bool("verify", false,
+		"scan every queue and report drift detection (slow: walks each queue)")
+	return cmd
+}
+
+// runStatusVerify prints the drift check: the counter against a scan of each
+// queue. It exits non-zero on confirmed drift, because a drifted counter means
+// the broker is advertising messages that do not exist and an export of that
+// queue can never complete -- a condition a script should be able to catch.
+func runStatusVerify(cmd *cobra.Command, c *broker.Client) error {
+	// The scan walks every queue, which on a deep backlog takes far longer than
+	// the --timeout meant for connecting, so this runs under a signal context
+	// like the other long operations rather than being cut off mid-sweep.
+	ctx, stop := signalCtx()
+	defer stop()
+
+	reports, err := c.CheckDriftAll(ctx)
+	if err != nil {
+		return err
+	}
+	w := tabwriter.NewWriter(cmd.OutOrStdout(), 0, 0, 3, ' ', 0)
+	// COUNTER is what the broker advertises; SCANNED is what walking the queue
+	// actually finds. MISSING is the gap: messages the counter promises that are
+	// not there.
+	fmt.Fprintln(w, "QUEUE NAME\tCOUNTER\tSCANNED\tMISSING\tVERDICT")
+	drifted := 0
+	for _, r := range reports {
+		if r.Verdict() == broker.DriftConfirmed {
+			drifted++
+		}
+		fmt.Fprintf(w, "%s\t%d\t%d\t%d\t%s\n",
+			r.Queue, r.CounterBefore, r.Counted, r.Missing(), r.Verdict())
+	}
+	if err := w.Flush(); err != nil {
+		return err
+	}
+	if drifted > 0 {
+		fmt.Fprintf(cmd.OutOrStdout(),
+			"\n%d queue(s) have a drifted message counter: they advertise messages that do not exist,\n"+
+				"so no export can ever empty them. Restart the broker to rebuild the counters from its journal.\n",
+			drifted)
+		return fmt.Errorf("%d queue(s) with a drifted message counter", drifted)
+	}
+	return nil
 }