fix(observe): serve the off-path pool's counters at /stats

metrics/metrics.go

@@ -491,6 +491,26 @@ type Snapshot struct {
 	// mode it correctly reads ~0 because that path does no pipeline work. Zeroing either
 	// would hide a true number rather than protect anyone.
 	ObserveLLMNotice string `json:"observe_llm_notice,omitempty"`
+
+	// ObserveQueue is the off-path worker pool's counter tuple, filled by the host at
+	// serve time (the pool lives in `modes`, which sits above this package). All five
+	// counters are exposed rather than just `queued` because `dropped` is the one that
+	// changes a reader's conclusion: a drop is an observation silently given up, so a
+	// rising `dropped` means the potential_*/projected_* figures UNDERSTATE what
+	// compaction would have saved. Reporting the queue depth while hiding the drops is
+	// exactly the gap noted in headroom's dashboard. Omitted when no pool is running, so
+	// a sync-only deployment shows no phantom queue.
+	ObserveQueue *QueueStats `json:"observe_queue,omitempty"`
+}
+
+// QueueStats mirrors modes.Stats. Declared here as a plain struct rather than importing
+// modes because the dependency runs the other way (modes uses metrics, not vice versa).
+type QueueStats struct {
+	Queued    int64 `json:"queued"`
+	Pending   int64 `json:"pending"`
+	Processed int64 `json:"processed"`
+	Dropped   int64 `json:"dropped"`
+	Errors    int64 `json:"errors"`
 }
 
 // SelectorMiss is one output shape that matched no filter, with how often it appeared.

proxy/modes_test.go

@@ -541,3 +541,37 @@ func settleGoroutines() {
 		time.Sleep(5 * time.Millisecond)
 	}
 }
+
+// The observe-mode docs tell operators to watch `dropped` and `errors`, and make a point
+// of contrasting that against a dashboard which reports only queue depth. The pool tracked
+// all five counters correctly and NOTHING SERVED THEM: metrics.Snapshot had no field and
+// the /stats handler never called Stats(), so the documented counter was unreachable. This
+// asserts the wiring, not the pool.
+func TestObserveQueueCountersReachStats(t *testing.T) {
+	up, _ := captureUpstream(t)
+	h, agg := modeHandler(t, modePipeline, up.URL, components.ModeObserve)
+	srv := httptest.NewServer(h.Mux())
+	defer srv.Close()
+
+	for i := 0; i < 3; i++ {
+		post(t, srv, dupBody())
+	}
+	awaitSnapshot(t, agg, func(s metrics.Snapshot) bool { return s.ObserveRequests > 0 })
+
+	var got metrics.Snapshot
+	res, err := http.Get(srv.URL + "/stats")
+	if err != nil {
+		t.Fatalf("GET /stats: %v", err)
+	}
+	defer res.Body.Close()
+	if err := json.NewDecoder(res.Body).Decode(&got); err != nil {
+		t.Fatalf("decode /stats: %v", err)
+	}
+	if got.ObserveQueue == nil {
+		t.Fatal("observe_queue absent from /stats: the pool's counters are still unreachable, " +
+			"so the documented `dropped` cannot be read by any consumer")
+	}
+	if got.ObserveQueue.Processed == 0 && got.ObserveQueue.Queued == 0 && got.ObserveQueue.Pending == 0 {
+		t.Fatalf("observe_queue served but empty after 3 observed requests: %+v", got.ObserveQueue)
+	}
+}

proxy/proxy.go

@@ -654,6 +654,17 @@ func (h *Handler) stats(w http.ResponseWriter, _ *http.Request) {
 		snap.FrozenDropped, snap.FrozenRepaired = fl.FrozenLossStats()
 		snap.FrozenFlips = snap.FrozenDropped - snap.FrozenRepaired
 	}
+	// Off-path pool counters, same layering: the pool lives in `modes`, so `metrics` cannot
+	// read it and the host merges it here. Without this the observe-mode docs describe a
+	// `dropped` counter no consumer can reach — the pool tracked it correctly and nothing
+	// served it.
+	if h.pool != nil {
+		q := h.pool.Stats()
+		snap.ObserveQueue = &metrics.QueueStats{
+			Queued: q.Queued, Pending: q.Pending,
+			Processed: q.Processed, Dropped: q.Dropped, Errors: q.Errors,
+		}
+	}
 	json.NewEncoder(w).Encode(snap)
 }