From f92a9396df945063202e2e6a9ee23c875f1f298a Mon Sep 17 00:00:00 2001 From: Xiangzhe Date: Sun, 23 Aug 2026 13:55:34 -0300 Subject: [PATCH] feat(monitoring): expose structural chat admission snapshot + shed counters (#11244) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The structural admission gate (src/shared/middleware/chatBodyAdmission.ts — the bounded heavyweight lease + healthy-headroom path from #10110/#10437) returns its 503 chat_admission_busy BEFORE request logging, so a shed left no trace: no counter, no log line, and the process-wide snapshot was never consumed by any route. This adds pure observability — admission behavior, defaults, and thresholds are untouched. - chatBodyAdmission.ts: in-memory shed history (shedTotal + shedsByReason) on ChatAdmissionController, recorded at the two capacity-driven give-up points in acquireHeavyWithin (queue_timeout when the bounded wait expires, queued_bytes_budget when the heap valve refuses to park). A client abort mid-wait is deliberately not counted — capacity was never denied. Each shed also emits exactly one structured pino warn (module chat-admission) with reason/activeHeavy/waiting/queuedBytes and the HMAC session fingerprint — never a raw credential. PerConnectionAdmission Controller.snapshot() now carries the counters plus activeHealthyHeadroom. - /api/monitoring/health: the structural snapshot is exposed under a new additive chatAdmission key next to the existing adaptiveAdmission (the shadow-mode layer) — allowlisted projection in observability.ts, nothing removed from the current payload. - Tests: tests/unit/chat-admission-visibility-11244.test.ts (RED→GREEN: shed counting per reason, abort exclusion, snapshot shape, and the warn log carrying the fingerprint but never the raw API key) + a chatAdmission allowlist-projection test in observability-payloads.test.ts. --- src/app/api/monitoring/health/route.ts | 15 ++ src/lib/monitoring/observability.ts | 45 ++++ src/shared/middleware/chatBodyAdmission.ts | 127 +++++++++- .../chat-admission-visibility-11244.test.ts | 238 ++++++++++++++++++ tests/unit/observability-payloads.test.ts | 69 +++++ 5 files changed, 485 insertions(+), 9 deletions(-) create mode 100644 tests/unit/chat-admission-visibility-11244.test.ts diff --git a/src/app/api/monitoring/health/route.ts b/src/app/api/monitoring/health/route.ts index f3dceac558..b3e031b348 100644 --- a/src/app/api/monitoring/health/route.ts +++ b/src/app/api/monitoring/health/route.ts @@ -74,6 +74,7 @@ export async function GET(request: Request) { credentialHealthModule, localHealthModule, adaptiveAdmissionModule, + chatAdmissionModule, settingsResult, connectionsResult, ] = await Promise.allSettled([ @@ -86,6 +87,7 @@ export async function GET(request: Request) { import("@/lib/credentialHealth/cache"), import("@/lib/localHealthCheck"), import("@omniroute/open-sse/services/admission/runtime.ts"), + import("@/shared/middleware/chatBodyAdmission"), getCachedSettings(), getProviderConnections(), ]); @@ -172,6 +174,17 @@ export async function GET(request: Request) { null ) : null; + // #11244: the STRUCTURAL admission gate (chatBodyAdmission.ts — bounded + // heavyweight lease + shed counters), exposed next to but distinct from the + // adaptive shadow-mode snapshot above. Additive key — nothing existing moves. + const chatAdmission = + chatAdmissionModule.status === "fulfilled" + ? readHealthValue( + "chat admission", + () => chatAdmissionModule.value.perConnectionAdmissionController.snapshot(), + null + ) + : null; const payload = buildHealthPayload({ appVersion: APP_CONFIG.version, @@ -200,6 +213,7 @@ export async function GET(request: Request) { activeSessionsByKey, credentialHealth, adaptiveAdmission, + chatAdmission, }); healthPayloadCache = { payload, expiresAt: Date.now() + HEALTH_PAYLOAD_TTL_MS }; @@ -218,6 +232,7 @@ export async function GET(request: Request) { quotaMonitor: { ...fallbackQuotaMonitorSummary, monitors: [] }, sessions: { activeCount: 0, stickyBoundCount: 0, byApiKey: {}, top: [] }, adaptiveAdmission: null, + chatAdmission: null, dedup: { inflightRequests: 0 }, }); } diff --git a/src/lib/monitoring/observability.ts b/src/lib/monitoring/observability.ts index 4cc18b27fd..cb4d239a65 100644 --- a/src/lib/monitoring/observability.ts +++ b/src/lib/monitoring/observability.ts @@ -3,9 +3,48 @@ import { getCodexParentAccountDiagnostic, } from "@omniroute/open-sse/services/codexAccount/index.ts"; import type { AdaptiveAdmissionPublicSnapshot } from "@omniroute/open-sse/services/admission/runtime.ts"; +import type { PerConnectionAdmissionController } from "@/shared/middleware/chatBodyAdmission"; type JsonRecord = Record; +/** Process-wide structural chat-admission snapshot type (chatBodyAdmission.ts). */ +export type ChatAdmissionSnapshot = ReturnType; + +/** + * Low-card structural chat-admission health summary (#11244) — the bounded + * heavyweight-lease gate from chatBodyAdmission.ts (#10110/#10437), NOT the + * adaptive shadow-mode layer above. Lane keys are opaque HMAC fairness + * fingerprints (resolveSessionId), never raw credentials. + */ +export type ChatAdmissionHealthSummary = { + activeHeavy: number; + activeHealthyHeadroom: number; + waiting: number; + queuedBytes: number; + shedTotal: number; + shedsByReason: Record; + lanes: Array<{ key: string; waiting: number }>; +}; + +/** + * Explicit allowlisted projection of the structural admission snapshot. + * Never spreads the snapshot — only the documented low-cardinality fields pass. + */ +export function projectChatAdmissionSummary( + snapshot: ChatAdmissionSnapshot | null | undefined +): ChatAdmissionHealthSummary | null { + if (!snapshot || typeof snapshot !== "object") return null; + return { + activeHeavy: snapshot.activeHeavy, + activeHealthyHeadroom: snapshot.activeHealthyHeadroom, + waiting: snapshot.waiting, + queuedBytes: snapshot.queuedBytes, + shedTotal: snapshot.shedTotal, + shedsByReason: { ...(snapshot.shedsByReason ?? {}) }, + lanes: (snapshot.lanes ?? []).map((lane) => ({ key: lane.key, waiting: lane.waiting })), + }; +} + /** Low-card adaptive-admission health summary — no tenant/request/body/queue details. */ export type AdaptiveAdmissionHealthSummary = { mode: AdaptiveAdmissionPublicSnapshot["mode"]; @@ -160,6 +199,8 @@ interface BuildHealthPayloadOptions { }; /** Optional injected public adaptive-admission snapshot; projected, never raw-spread. */ adaptiveAdmission?: AdaptiveAdmissionPublicSnapshot | null; + /** #11244: optional structural chat-admission snapshot; projected, never raw-spread. */ + chatAdmission?: ChatAdmissionSnapshot | null; } function limitMonitors(monitors: QuotaMonitorSnapshot[], maxItems = 8): QuotaMonitorSnapshot[] { @@ -347,6 +388,7 @@ export function buildHealthPayload({ activeSessionsByKey = {}, credentialHealth, adaptiveAdmission = null, + chatAdmission = null, buildSha = null, }: BuildHealthPayloadOptions) { const timestamp = new Date().toISOString(); @@ -449,6 +491,9 @@ export function buildHealthPayload({ sessions: buildSessionsSummary({ activeSessions, activeSessionsByKey }), credentialHealth, // may be undefined if credentialHealth module not loaded adaptiveAdmission: projectAdaptiveAdmissionSummary(adaptiveAdmission), + // #11244: the STRUCTURAL gate (chatBodyAdmission.ts) next to the adaptive one — + // distinct key so clients reading `adaptiveAdmission` are untouched. + chatAdmission: projectChatAdmissionSummary(chatAdmission), dedup: { inflightRequests, }, diff --git a/src/shared/middleware/chatBodyAdmission.ts b/src/shared/middleware/chatBodyAdmission.ts index 9b20d78e5a..3e37f3ca3b 100644 --- a/src/shared/middleware/chatBodyAdmission.ts +++ b/src/shared/middleware/chatBodyAdmission.ts @@ -16,6 +16,7 @@ */ import { CORS_HEADERS } from "../utils/cors"; +import { createLogger } from "../utils/logger"; import { createHmac } from "crypto"; import v8 from "node:v8"; import { trackRequest } from "../../lib/gracefulShutdown"; @@ -168,6 +169,47 @@ interface AdmissionWaiter { readonly resolve: () => void; } +/** + * Why a structural shed (503 `chat_admission_busy`) happened (#11244): + * - `queue_timeout`: the bounded wait expired with no heavyweight capacity freed + * (includes the `queueMs=0` legacy immediate-reject path — capacity was busy at + * the instant the request arrived). + * - `queued_bytes_budget`: the queued-bytes heap valve (#9654 / U3) refused to + * park the waiter because the buffered-body budget was already exhausted. + * + * A client abort mid-wait is deliberately NOT a shed: capacity was never denied, + * the caller simply left (its 503 is dropped on the dead connection). + */ +export type ChatAdmissionShedReason = "queue_timeout" | "queued_bytes_budget"; + +/** + * One structural-shed observation, emitted to the shed sink at warn level. + * `lane` is the opaque fairness key — the HMAC fingerprint produced by + * `resolveSessionId` (or "anonymous"/"default"), never a raw credential. + */ +export interface ChatAdmissionShedEvent { + reason: ChatAdmissionShedReason; + activeHeavy: number; + waiting: number; + queuedBytes: number; + lane: string; +} + +export type ChatAdmissionShedSink = (event: ChatAdmissionShedEvent) => void; + +const shedLog = createLogger("chat-admission"); + +/** + * Default shed sink (#11244): exactly one structured warn per structural shed. + * The 503 returns BEFORE request logging, so without this line a shed left no + * trace anywhere. No raw credentials — `lane` is already the HMAC fingerprint, + * and the shared logger's redaction hook (logRedaction.ts) is the safety net. + * Nothing is logged for admitted requests (noise). + */ +function defaultChatAdmissionShedSink(event: ChatAdmissionShedEvent): void { + shedLog.warn(event, "structural chat admission shed (chat_admission_busy)"); +} + /** * Process-local heavyweight reservation. The capacity check and increment execute in one * synchronous JavaScript turn, making acquisition atomic within an OmniRoute process. @@ -190,6 +232,12 @@ export class ChatAdmissionController { /** Keys in creation order; #fairCursor scans them round-robin. */ #fairKeys: string[] = []; #fairCursor = 0; + /** #11244: in-memory shed history (total + per reason). The 503 chat_admission_busy + * response returns before request logging, so without these counters a structural + * shed was invisible. Same in-memory lifetime as the rest of the snapshot state. */ + #shedTotal = 0; + #shedsByReason = new Map(); + readonly #onShed: ChatAdmissionShedSink; constructor( readonly maxHeavyInFlight = 1, @@ -197,7 +245,10 @@ export class ChatAdmissionController { /** #10437: bounded extra capacity for the healthy-heap fast path. `0` disables * the bypass entirely — every busy request then falls through to the same * bounded-wait/shed path used under real heap pressure. */ - readonly healthyHeadroom = CHAT_ADMISSION_HEALTHY_HEADROOM + readonly healthyHeadroom = CHAT_ADMISSION_HEALTHY_HEADROOM, + /** #11244: sink notified once per structural shed. Defaults to the shared pino + * logger (warn); tests inject a capture/no-op sink. */ + onShed: ChatAdmissionShedSink = defaultChatAdmissionShedSink ) { if (!Number.isSafeInteger(maxHeavyInFlight) || maxHeavyInFlight < 1) { throw new RangeError("maxHeavyInFlight must be a positive integer"); @@ -208,6 +259,7 @@ export class ChatAdmissionController { if (!Number.isSafeInteger(healthyHeadroom) || healthyHeadroom < 0) { throw new RangeError("healthyHeadroom must be a non-negative integer"); } + this.#onShed = onShed; } get activeHeavy(): number { @@ -264,6 +316,37 @@ export class ChatAdmissionController { return out; } + /** Total structural sheds since process start (#11244). */ + get shedTotal(): number { + return this.#shedTotal; + } + + /** Structural sheds by reason since process start (#11244). */ + get shedsByReason(): Record { + const out: Record = {}; + for (const [reason, count] of this.#shedsByReason) out[reason] = count; + return out; + } + + /** + * Record one structural shed (503 chat_admission_busy) and notify the shed sink + * (#11244). Called internally at every capacity-driven give-up point in + * `acquireHeavyWithin`; public so the aggregate snapshot wiring and tests can + * exercise the same single path. `lane` is the opaque fairness key (HMAC + * fingerprint), never a raw credential. + */ + recordShed(reason: ChatAdmissionShedReason, lane = "default"): void { + this.#shedTotal += 1; + this.#shedsByReason.set(reason, (this.#shedsByReason.get(reason) ?? 0) + 1); + this.#onShed({ + reason, + activeHeavy: this.#activeHeavy, + waiting: this.waitingCount, + queuedBytes: this.#queuedBytes, + lane, + }); + } + tryAcquireHeavy(): ChatAdmissionLease | null { if (this.#activeHeavy >= this.maxHeavyInFlight) return null; this.#activeHeavy += 1; @@ -318,9 +401,16 @@ export class ChatAdmissionController { const lease = this.tryAcquireHeavy(); if (lease) return lease; const remaining = deadline - Date.now(); - if (remaining <= 0) return null; + if (remaining <= 0) { + // Wait window exhausted (or queueMs=0 immediate reject) with capacity still + // busy — the caller answers the retryable 503. Count it (#11244). + this.recordShed("queue_timeout", sessionKey); + return null; + } // Heap valve: refuse to park when the queued-bytes budget is exhausted. if (queuedBytes > 0 && this.#queuedBytes + queuedBytes > this.maxQueuedBytes) { + // Same retryable 503, distinct cause: the wait itself would amplify the heap. + this.recordShed("queued_bytes_budget", sessionKey); return null; } this.#queuedBytes += queuedBytes; @@ -367,7 +457,13 @@ export class ChatAdmissionController { // Cancel the deadline timer when abort/release wins; a fired timer is a no-op. if (deadlineTimer) clearTimeout(deadlineTimer); if (onAbort) signal?.removeEventListener("abort", onAbort); - if (timedOut) return null; + if (timedOut) { + // The deadline timer won the race: a genuine shed. When the client ABORT + // won instead (signal aborted while parked), capacity was never denied — + // the 503 is dropped on the dead connection, so it is not counted (#11244). + if (!signal?.aborted) this.recordShed("queue_timeout", sessionKey); + return null; + } } } @@ -483,11 +579,18 @@ export class PerConnectionAdmissionController { constructor( readonly maxHeavyInFlight = 1, - // Deprecated pre-#10110 lane-eviction knobs: accepted for API - // compatibility and ignored — there are no per-session lanes to evict. - _opts?: { maxSessions?: number; sessionTtlMs?: number } + // `maxSessions`/`sessionTtlMs` are deprecated pre-#10110 lane-eviction knobs: + // accepted for API compatibility and ignored — there are no per-session lanes + // to evict. `onShed` (#11244) is live: it replaces the shed sink of the shared + // controller (tests inject a capture/no-op sink; production keeps the pino warn). + _opts?: { maxSessions?: number; sessionTtlMs?: number; onShed?: ChatAdmissionShedSink } ) { - this.#controller = new ChatAdmissionController(maxHeavyInFlight); + this.#controller = new ChatAdmissionController( + maxHeavyInFlight, + undefined, + undefined, + _opts?.onShed + ); } /** Returns the process-global budget — the same instance for every session. */ @@ -497,20 +600,26 @@ export class PerConnectionAdmissionController { /** * Process-wide aggregate snapshot for observability: global totals plus - * per-key waiter depths. Keys are opaque scheduler keys, never raw - * credentials. + * per-key waiter depths and the #11244 shed history (total + per reason). + * Keys are opaque scheduler keys, never raw credentials. */ snapshot(): { activeHeavy: number; + activeHealthyHeadroom: number; queuedBytes: number; waiting: number; lanes: ReadonlyArray<{ key: string; waiting: number }>; + shedTotal: number; + shedsByReason: Record; } { return { activeHeavy: this.#controller.activeHeavy, + activeHealthyHeadroom: this.#controller.activeHealthyHeadroom, queuedBytes: this.#controller.queuedBytes, waiting: this.#controller.waitingCount, lanes: this.#controller.waitersByKey, + shedTotal: this.#controller.shedTotal, + shedsByReason: this.#controller.shedsByReason, }; } diff --git a/tests/unit/chat-admission-visibility-11244.test.ts b/tests/unit/chat-admission-visibility-11244.test.ts new file mode 100644 index 0000000000..663d0943f2 --- /dev/null +++ b/tests/unit/chat-admission-visibility-11244.test.ts @@ -0,0 +1,238 @@ +// #11244: visibility for the STRUCTURAL chat admission gate +// (src/shared/middleware/chatBodyAdmission.ts — the bounded heavyweight lease + +// healthy-headroom path from #10110/#10437, NOT the adaptive shadow-mode layer in +// open-sse/services/admission/). The 503 chat_admission_busy shed returns BEFORE +// request logging, so today a shed is invisible: no counter, no log line, and the +// process-wide snapshot (PerConnectionAdmissionController.snapshot()) reports only +// live state (activeHeavy/queuedBytes/waiting/lanes) with no shed history. +// +// These tests pin the observability contract WITHOUT changing admission behavior: +// (a) every structural shed (503 chat_admission_busy) increments an in-memory +// counter — total + per reason ("queue_timeout" when the bounded wait expires, +// "queued_bytes_budget" when the queued-bytes heap valve refuses to park) — +// while a client abort mid-wait is NOT a shed (capacity was never denied); +// (b) the process-wide snapshot exposes shedTotal + shedsByReason next to the +// existing live fields; +// (c) each shed emits exactly one structured pino warn carrying +// reason/activeHeavy/waiting and the HMAC session fingerprint — never the raw +// API key (resolveSessionId already fingerprints the credential). +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtempSync, readFileSync, existsSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +// Configure the shared pino logger BEFORE importing the admission module — the +// logger builds its transports at import time (see logger-redaction-wiring.test.ts +// for the same pattern). JSON to a temp file keeps test (c)'s capture deterministic. +const logDir = mkdtempSync(join(tmpdir(), "omniroute-admission-11244-")); +const logFile = join(logDir, "app.log"); +process.env.NODE_ENV = "production"; +process.env.APP_LOG_TO_FILE = "true"; +process.env.APP_LOG_FILE_PATH = logFile; + +const { + ChatAdmissionController, + PerConnectionAdmissionController, + perConnectionAdmissionController, + admitChatStructure, + resolveSessionId, +} = await import("../../src/shared/middleware/chatBodyAdmission.ts"); + +function heavyBody() { + return { + messages: Array.from({ length: 200 }, () => ({ role: "user", content: "x".repeat(40) })), + tools: [] as unknown[], + }; +} + +const heapHealthy = () => false; // "not under pressure" — the healthy-heap fast path +const heapPressured = () => true; // forces the bounded-wait/shed path deterministically +const silentSink = () => {}; // keep non-logging tests off the pino transport + +test("#11244 (a): a structural shed after the bounded wait increments shedTotal and shedsByReason", async () => { + // Primary lease (1) + bounded healthy-headroom (1): two concurrent heavy requests + // admit on a healthy heap; the third must wait queueMs and then shed with a 503. + const controller = new ChatAdmissionController(1, undefined, 1, silentSink); + + const first = await admitChatStructure(heavyBody(), null, { + controller, + heapPressureCheck: heapHealthy, + queueMs: 0, + }); + const second = await admitChatStructure(heavyBody(), null, { + controller, + heapPressureCheck: heapHealthy, + queueMs: 0, + }); + assert.equal(first.admit, true, "first heavy request takes the primary lease"); + assert.equal(second.admit, true, "second heavy request takes the bounded headroom lease"); + assert.equal(controller.shedTotal, 0, "admitted requests never count as sheds"); + assert.deepEqual(controller.shedsByReason, {}); + + const shed = await admitChatStructure(heavyBody(), null, { + controller, + heapPressureCheck: heapHealthy, + queueMs: 50, + }); + assert.equal(shed.admit, false, "third heavy request must shed once both budgets are busy"); + if (!shed.admit) { + assert.equal(shed.response.status, 503); + const payload = await shed.response.json(); + assert.equal(payload.error.code, "chat_admission_busy"); + } + + assert.equal( + controller.shedTotal, + 1, + "the shed must be counted even though it skips request logging" + ); + assert.deepEqual( + controller.shedsByReason, + { queue_timeout: 1 }, + "a bounded wait that expires with no freed capacity is a queue_timeout shed" + ); + + // Counters are history, not live state: releasing the leases must not rewind them. + if (first.admit) first.lease?.release(); + if (second.admit) second.lease?.release(); + assert.equal(controller.shedTotal, 1, "shed history survives lease release"); + + // And a subsequently admitted request must not be counted. + const fourth = await admitChatStructure(heavyBody(), null, { + controller, + heapPressureCheck: heapHealthy, + queueMs: 0, + }); + assert.equal(fourth.admit, true); + assert.equal(controller.shedTotal, 1); + if (fourth.admit) fourth.lease?.release(); +}); + +test("#11244 (a2): the queued-bytes heap valve rejection is counted with its own reason", async () => { + // maxQueuedBytes smaller than the conservative 256KB structural wait weight: the + // valve refuses to park and the shed must be distinguishable from a queue timeout. + const controller = new ChatAdmissionController(1, 1024, 0, silentSink); + + const first = await admitChatStructure(heavyBody(), null, { + controller, + heapPressureCheck: heapPressured, + queueMs: 0, + }); + assert.equal(first.admit, true); + + const shed = await admitChatStructure(heavyBody(), null, { + controller, + heapPressureCheck: heapPressured, + queueMs: 1000, + }); + assert.equal(shed.admit, false); + if (!shed.admit) assert.equal(shed.response.status, 503); + assert.equal(controller.shedTotal, 1); + assert.deepEqual(controller.shedsByReason, { queued_bytes_budget: 1 }); + + if (first.admit) first.lease?.release(); +}); + +test("#11244 (a3): a client abort while parked is not a shed — capacity was never denied", async () => { + const controller = new ChatAdmissionController(1, undefined, 0, silentSink); + + const first = await admitChatStructure(heavyBody(), null, { + controller, + heapPressureCheck: heapPressured, + queueMs: 0, + }); + assert.equal(first.admit, true); + + const abort = new AbortController(); + const pending = admitChatStructure(heavyBody(), null, { + controller, + heapPressureCheck: heapPressured, + queueMs: 5000, + signal: abort.signal, + }); + setTimeout(() => abort.abort(), 20); + const result = await pending; + assert.equal( + result.admit, + false, + "the caller still answers a (dropped) 503 on the dead connection" + ); + assert.equal(controller.shedTotal, 0, "an aborted wait frees capacity instead of shedding"); + assert.deepEqual(controller.shedsByReason, {}); + + if (first.admit) first.lease?.release(); +}); + +test("#11244 (b): the process-wide snapshot exposes shed counters next to the live fields", async () => { + const empty = perConnectionAdmissionController.snapshot(); + assert.equal(typeof empty.activeHeavy, "number"); + assert.equal(typeof empty.queuedBytes, "number"); + assert.equal(typeof empty.waiting, "number"); + assert.ok(Array.isArray(empty.lanes)); + assert.equal( + empty.shedTotal, + 0, + "no shed happened through the production singleton in this process" + ); + assert.deepEqual(empty.shedsByReason, {}); + + // A shed recorded through a session's controller surfaces in the aggregate snapshot. + const pc = new PerConnectionAdmissionController(1, { onShed: silentSink }); + const controller = pc.getController("key_visibility11244"); + controller.recordShed("queue_timeout", "key_visibility11244"); + controller.recordShed("queue_timeout", "key_visibility11244"); + controller.recordShed("queued_bytes_budget", "key_visibility11244"); + + const snap = pc.snapshot(); + assert.equal(snap.shedTotal, 3); + assert.deepEqual(snap.shedsByReason, { queue_timeout: 2, queued_bytes_budget: 1 }); +}); + +test("#11244 (c): each shed logs one structured warn with the session fingerprint, never the raw key", async () => { + const rawKey = "visRAWSECRETtoken11244xyz"; // matches no logRedaction pattern — a leak would show verbatim + const fingerprint = resolveSessionId( + new Request("http://localhost/v1/chat/completions", { + headers: { authorization: `Bearer ${rawKey}` }, + }) + ); + assert.ok(fingerprint.startsWith("key_"), "resolveSessionId returns the HMAC fingerprint"); + assert.ok(!fingerprint.includes(rawKey)); + + // Default sink (no injected onShed): the shed must go through the shared pino logger. + const controller = new ChatAdmissionController(1); + const primary = controller.tryAcquireHeavy(); + assert.ok(primary); + + const shed = await admitChatStructure(heavyBody(), null, { + controller, + heapPressureCheck: heapPressured, + queueMs: 25, + sessionId: fingerprint, + }); + assert.equal(shed.admit, false); + primary.release(); + + // Poll the worker-thread-written log file until the shed line lands. + const deadline = Date.now() + 4000; + let contents = ""; + while (Date.now() < deadline) { + if (existsSync(logFile)) { + contents = readFileSync(logFile, "utf8"); + if (contents.includes("chat_admission_busy")) break; + } + await new Promise((resolve) => setTimeout(resolve, 50)); + } + + assert.ok(contents.includes("chat_admission_busy"), "the shed log line names the rejection code"); + assert.match( + contents, + /"level":(40|"warn")/, + "sheds log at warn level (numeric 40 when the file transport strips the level formatter)" + ); + assert.ok(contents.includes('"module":"chat-admission"'), "the log is scoped to the gate"); + assert.ok(contents.includes('"reason":"queue_timeout"'), "the shed reason is structured"); + assert.ok(contents.includes('"activeHeavy":1'), "live state travels with the log line"); + assert.ok(contents.includes(fingerprint), "the lane fingerprint allows per-key correlation"); + assert.ok(!contents.includes(rawKey), "the raw API key must never reach the shed log"); +}); diff --git a/tests/unit/observability-payloads.test.ts b/tests/unit/observability-payloads.test.ts index 297d92d6ec..5cb9241b28 100644 --- a/tests/unit/observability-payloads.test.ts +++ b/tests/unit/observability-payloads.test.ts @@ -6,6 +6,7 @@ import { buildSessionsSummary, buildTelemetryPayload, projectAdaptiveAdmissionSummary, + projectChatAdmissionSummary, } from "../../src/lib/monitoring/observability.ts"; test("buildSessionsSummary returns sticky counts and ordered top sessions", () => { @@ -336,3 +337,71 @@ test("buildHealthPayload projects allowlisted adaptiveAdmission aggregates only" assert.equal(projectAdaptiveAdmissionSummary(null), null); assert.equal(projectAdaptiveAdmissionSummary(undefined), null); }); + +// #11244: the STRUCTURAL chat-admission gate (chatBodyAdmission.ts) must surface in +// the health payload next to — never instead of — the adaptive snapshot, with only +// the documented low-cardinality fields projected. +test("buildHealthPayload projects allowlisted structural chatAdmission fields only", () => { + const snapshot = { + activeHeavy: 1, + activeHealthyHeadroom: 1, + waiting: 2, + queuedBytes: 524_288, + shedTotal: 3, + shedsByReason: { queue_timeout: 2, queued_bytes_budget: 1 }, + lanes: [ + { key: "key_c49d1c242feda590", waiting: 1 }, + { key: "anonymous", waiting: 1 }, + ], + // Extra keys that must never leak into the public payload. + internalController: { secret: "controller-state" }, + rawAuthorization: "Bearer raw-SHOULD-NOT-LEAK", + } as unknown as import("../../src/lib/monitoring/observability.ts").ChatAdmissionSnapshot; + + const payload = buildHealthPayload({ + appVersion: "9.9.9", + settings: { setupComplete: false }, + connections: [], + circuitBreakers: [], + rateLimitStatus: {}, + learnedLimits: {}, + lockouts: {}, + localProviders: {}, + inflightRequests: 0, + quotaMonitorSummary: { + active: 0, + alerting: 0, + exhausted: 0, + errors: 0, + statusCounts: { starting: 0, idle: 0, healthy: 0, warning: 0, exhausted: 0, error: 0 }, + byProvider: {}, + }, + quotaMonitorMonitors: [], + activeSessions: [], + chatAdmission: snapshot, + }); + + assert.deepEqual(payload.chatAdmission, { + activeHeavy: 1, + activeHealthyHeadroom: 1, + waiting: 2, + queuedBytes: 524_288, + shedTotal: 3, + shedsByReason: { queue_timeout: 2, queued_bytes_budget: 1 }, + lanes: [ + { key: "key_c49d1c242feda590", waiting: 1 }, + { key: "anonymous", waiting: 1 }, + ], + }); + // The adaptive projection is untouched by the new key. + assert.equal(payload.adaptiveAdmission, null); + + const json = JSON.stringify(payload); + assert.equal(json.includes("controller-state"), false); + assert.equal(json.includes("raw-SHOULD-NOT-LEAK"), false); + assert.equal(json.includes("internalController"), false); + + // Absent / null snapshot projects to null (degraded path parity). + assert.equal(projectChatAdmissionSummary(null), null); + assert.equal(projectChatAdmissionSummary(undefined), null); +});