Expose privacy-safe in-process answer coalescing metrics on deep health probe

docs/capacity-review.md

@@ -67,6 +67,14 @@ round, coalescing + cache absorb roughly a third of the fan-out at exactly the
 moment load is highest. This only works while the app is a **single process**
 (see deployment doc §2).
 
+The authorized deep health probe now exposes privacy-safe process-local
+`coalescing` counters (`originations`, `coalescedWaiters`, and
+`activeOriginations`). Before changing replica count, compare their deltas with
+the retrieval cache counters over a duplicate-heavy period. A sustained low
+coalescing rate may be expected for unique queries, but is a capacity/cost
+regression signal when the ward-round duplicate hypothesis holds; it is not a
+reason to make readiness fail.
+
 ### OpenAI: rate limits and generation concurrency
 
 Per answer: 1 embedding call (unless the lexical fast path skips it) + 1–2

docs/observability-slos.md

@@ -177,7 +177,7 @@ Operational notes:
 
 The §2 reliability SQL is now also a scrape. An **authorized deep probe** —
 `GET /api/health?deep=1` with the `x-health-deep-token: $HEALTH_DEEP_PROBE_SECRET`
-header (same operator gate as the Supabase probe) — returns two counter blocks:
+header (same operator gate as the Supabase probe) — returns three counter blocks:
 
 - **`slo`** — `answerSloSnapshot` (`src/lib/observability/answer-slo.ts`) counts
   `rag_queries` over the trailing `windowMinutes` (60) and reports
@@ -199,12 +199,25 @@ header (same operator gate as the Supabase probe) — returns two counter blocks
   never write a telemetry row. `hitRate` is a convenience for eyeballing a single
   probe.
 
+- **`coalescing`** — `answerCoalescingMetricsSnapshot`
+  (`src/lib/observability/answer-coalescing-metrics.ts`) reports cumulative
+  coalescible answer **originations**, duplicate **coalescedWaiters**, and the
+  current **activeOriginations** gauge for this one app process. It has no query,
+  owner, document, cache key, or clinical content. A scraper must derive a
+  windowed `coalescingRate` from counter deltas; the supplied rate is a
+  process-lifetime convenience. This is an operational capacity/cost signal,
+  not a liveness gate: a sustained near-zero rate during a known duplicate-heavy
+  ward round means cache keys, cache settings, or replica dilution should be
+  investigated before adding app replicas.
+
 ```jsonc
 // GET /api/health?deep=1  (authorized, live)
 "slo":   { "windowMinutes": 60, "totalQueries": 412,
            "hybridRpcErrorQueries": 0, "hybridRpcErrorRate": 0.0,
            "degradedQueries": 26,      "degradedRate": 0.063 },
-"cache": { "lookups": 1873, "hits": 1402, "misses": 471, "hitRate": 0.748 }
+"cache": { "lookups": 1873, "hits": 1402, "misses": 471, "hitRate": 0.748 },
+"coalescing": { "originations": 312, "coalescedWaiters": 100,
+                 "activeOriginations": 4, "coalescingRate": 0.243 }
 ```
 
 **The counter _values_ never flip liveness.** A bad SLO rate or a cold cache
@@ -214,7 +227,10 @@ only by the reachability `checks`: a failing `probeSupabaseHealth()` still sets
 `checks.supabase = "error"` and returns `503`, whereas an `answerSloSnapshot()`
 query failure merely omits the `slo` block (never a false-healthy zero). The
 cache counter is always present for a token-authorized probe (in-process, works
-in demo mode). Both blocks are wired through `src/lib/health-response.ts` and are
+in demo mode). The cache and coalescing blocks are process-local and reset on
+deploy/restart, so operators must use per-process deltas rather than combine raw
+counter values across replicas. All three blocks are wired through
+`src/lib/health-response.ts` and are
 **withheld from the unauthenticated `/api/health/ready` endpoint** (Railway's
 readiness target, which exposes no diagnostic details). The §2 warn/page
 thresholds map directly onto these fields.

src/app/api/health/ready/route.ts

@@ -11,5 +11,6 @@ export async function GET(request: Request) {
     allowUnauthenticatedDeep: true,
     includeSlo: false,
     includeCache: false,
+    includeCoalescing: false,
   });
 }

src/lib/health-response.ts

@@ -3,13 +3,18 @@ import { allowDeepHealthProbe } from "@/lib/deep-probe-auth";
 import { env, isDemoMode } from "@/lib/env";
 import type { AnswerSloSnapshot, SloProbeClient } from "@/lib/observability/answer-slo";
 import { cacheMetricsSnapshot, type CacheMetricsSnapshot } from "@/lib/observability/cache-metrics";
+import {
+  answerCoalescingMetricsSnapshot,
+  type AnswerCoalescingMetricsSnapshot,
+} from "@/lib/observability/answer-coalescing-metrics";
 import type { SpendProbeClient, SpendSnapshot } from "@/lib/observability/spend-metrics";
 
 type HealthResponseOptions = {
   forceDeep?: boolean;
   allowUnauthenticatedDeep?: boolean;
   includeSlo?: boolean;
   includeCache?: boolean;
+  includeCoalescing?: boolean;
   includeSpend?: boolean;
 };
 
@@ -23,6 +28,7 @@ export async function healthResponse(request: Request, options: HealthResponseOp
   };
   let slo: AnswerSloSnapshot | null = null;
   let cache: CacheMetricsSnapshot | null = null;
+  let coalescing: AnswerCoalescingMetricsSnapshot | null = null;
   let spend: SpendSnapshot | null = null;
 
   if (deep) {
@@ -39,6 +45,9 @@ export async function healthResponse(request: Request, options: HealthResponseOp
       if (tokenAuthorized && options.includeCache !== false) {
         cache = cacheMetricsSnapshot();
       }
+      if (tokenAuthorized && options.includeCoalescing !== false) {
+        coalescing = answerCoalescingMetricsSnapshot();
+      }
 
       if (supabaseConfigured && !isDemoMode()) {
         try {
@@ -103,6 +112,7 @@ export async function healthResponse(request: Request, options: HealthResponseOp
       checks,
       ...(slo ? { slo } : {}),
       ...(cache ? { cache } : {}),
+      ...(coalescing ? { coalescing } : {}),
       ...(spend ? { spend } : {}),
     },
     { status: ready ? 200 : 503, headers: { "Cache-Control": "no-store" } },

src/lib/observability/answer-coalescing-metrics.ts

@@ -0,0 +1,49 @@
+// Process-local counters for the in-flight answer coalescer.
+//
+// These measurements intentionally contain no request key, owner, query, or
+// document data. They show whether this process is absorbing duplicate answer
+// work, which is otherwise invisible when considering replica changes.
+
+let originations = 0;
+let coalescedWaiters = 0;
+let activeOriginations = 0;
+
+export type AnswerCoalescingMetricsSnapshot = {
+  originations: number;
+  coalescedWaiters: number;
+  activeOriginations: number;
+  // 0..1; 0 when no coalescible request has completed or joined yet.
+  coalescingRate: number;
+};
+
+export function recordAnswerOrigination(): void {
+  originations += 1;
+  activeOriginations += 1;
+}
+
+export function recordAnswerOriginationFinished(): void {
+  // A defensive floor means an unexpected cleanup path cannot emit a negative
+  // gauge, which would be misleading to the operator polling this endpoint.
+  activeOriginations = Math.max(0, activeOriginations - 1);
+}
+
+export function recordCoalescedAnswerWaiter(): void {
+  coalescedWaiters += 1;
+}
+
+export function answerCoalescingMetricsSnapshot(): AnswerCoalescingMetricsSnapshot {
+  const totalRequests = originations + coalescedWaiters;
+  return {
+    originations,
+    coalescedWaiters,
+    activeOriginations,
+    coalescingRate: totalRequests > 0 ? coalescedWaiters / totalRequests : 0,
+  };
+}
+
+/** Test-only: reset process-local counters between cases. */
+export function resetAnswerCoalescingMetrics(): void {
+  originations = 0;
+  coalescedWaiters = 0;
+  activeOriginations = 0;
+}

src/lib/rag.ts

@@ -120,6 +120,11 @@ export {
   retrievalPlanCacheQuery,
 } from "@/lib/rag-cache";
 import { classifySearchCacheOutcome, recordCacheLookup } from "@/lib/observability/cache-metrics";
+import {
+  recordAnswerOrigination,
+  recordAnswerOriginationFinished,
+  recordCoalescedAnswerWaiter,
+} from "@/lib/observability/answer-coalescing-metrics";
 import { buildRagSourceBlock, compactContextText, neutralizeIdentityField } from "@/lib/rag-source-block";
 export { buildRagSourceBlock, truncateForModel } from "@/lib/rag-source-block";
 import {
@@ -3201,6 +3206,7 @@ export async function answerQuestionWithScope(args: AnswerQuestionWithScopeArgs)
   let existing = inflightKey ? answerInflight.get(inflightKey) : undefined;
 
   while (existing) {
+    recordCoalescedAnswerWaiter();
     await args.onProgress?.({
       stage: "cached",
       message: "Waiting for an identical cited answer request already in progress.",
@@ -3232,8 +3238,15 @@ export async function answerQuestionWithScope(args: AnswerQuestionWithScopeArgs)
     }
   }
 
+  // Only coalescible requests belong in this process-local signal. Requests
+  // that intentionally bypass cache/coalescing must not make a replica look
+  // ineffective, and neither keys nor clinical content leave this function.
+  if (inflightKey) recordAnswerOrigination();
   const pending = answerQuestionWithScopeUncoalesced(args, startedAt).finally(() => {
-    if (inflightKey) answerInflight.delete(inflightKey);
+    if (inflightKey) {
+      answerInflight.delete(inflightKey);
+      recordAnswerOriginationFinished();
+    }
   });
   if (inflightKey) answerInflight.set(inflightKey, pending);
   return pending;

tests/answer-coalescing-metrics.test.ts

@@ -0,0 +1,43 @@
+import { afterEach, describe, expect, it } from "vitest";
+
+import {
+  answerCoalescingMetricsSnapshot,
+  recordAnswerOrigination,
+  recordAnswerOriginationFinished,
+  recordCoalescedAnswerWaiter,
+  resetAnswerCoalescingMetrics,
+} from "@/lib/observability/answer-coalescing-metrics";
+
+afterEach(() => {
+  resetAnswerCoalescingMetrics();
+});
+
+describe("answer coalescing metrics", () => {
+  it("reports a safe zero snapshot before coalescible answer work", () => {
+    expect(answerCoalescingMetricsSnapshot()).toEqual({
+      originations: 0,
+      coalescedWaiters: 0,
+      activeOriginations: 0,
+      coalescingRate: 0,
+    });
+  });
+
+  it("tracks originations, waiters, and the active gauge without request content", () => {
+    recordAnswerOrigination();
+    recordAnswerOrigination();
+    recordCoalescedAnswerWaiter();
+    recordAnswerOriginationFinished();
+
+    expect(answerCoalescingMetricsSnapshot()).toEqual({
+      originations: 2,
+      coalescedWaiters: 1,
+      activeOriginations: 1,
+      coalescingRate: 1 / 3,
+    });
+  });
+
+  it("never reports a negative active gauge when cleanup is repeated", () => {
+    recordAnswerOriginationFinished();
+    expect(answerCoalescingMetricsSnapshot().activeOriginations).toBe(0);
+  });
+});

tests/health-route.test.ts

@@ -85,7 +85,7 @@ describe("GET /api/health", () => {
     expect(body.checks).toMatchObject({ supabaseConfig: "ok", openaiConfig: "skipped" });
   });
 
-  it("gates the deep probe without a token and omits the slo/cache snapshots", async () => {
+  it("gates the deep probe without a token and omits diagnostic snapshots", async () => {
     mockEnv({ configured: true, deepSecret: true });
     const { GET } = await import("../src/app/api/health/route");
 
@@ -96,6 +96,7 @@ describe("GET /api/health", () => {
     expect(body.checks).toMatchObject({ supabase: "unauthorized" });
     expect(body.slo).toBeUndefined();
     expect(body.cache).toBeUndefined();
+    expect(body.coalescing).toBeUndefined();
   });
 
   it("exposes the in-process cache hit-rate counter on an authorized deep probe", async () => {
@@ -115,6 +116,12 @@ describe("GET /api/health", () => {
     expect(cache.misses).toBe(cache.lookups - cache.hits);
     expect(cache.hitRate).toBeGreaterThanOrEqual(0);
     expect(cache.hitRate).toBeLessThanOrEqual(1);
+    const coalescing = body.coalescing as Record<string, number>;
+    expect(typeof coalescing.originations).toBe("number");
+    expect(typeof coalescing.coalescedWaiters).toBe("number");
+    expect(typeof coalescing.activeOriginations).toBe("number");
+    expect(coalescing.coalescingRate).toBeGreaterThanOrEqual(0);
+    expect(coalescing.coalescingRate).toBeLessThanOrEqual(1);
     expect(body.slo).toBeUndefined();
   });
 
@@ -142,7 +149,7 @@ describe("GET /api/health/ready", () => {
     expect(body.checks).toMatchObject({ supabaseConfig: "ok", openaiConfig: "ok", supabase: "skipped" });
   });
 
-  it("exposes no diagnostic details (slo/cache) even to a token-bearing caller", async () => {
+  it("exposes no diagnostic details even to a token-bearing caller", async () => {
     mockEnv({ configured: true, demoMode: true, deepSecret: true });
     const { GET } = await import("../src/app/api/health/ready/route");
 
@@ -153,5 +160,6 @@ describe("GET /api/health/ready", () => {
 
     expect(body.slo).toBeUndefined();
     expect(body.cache).toBeUndefined();
+    expect(body.coalescing).toBeUndefined();
   });
 });