feat(supervisor): backpressure dry-run mode and prometheus metrics

Dry-run (default on via env) keeps the gates inert while computeEngaged still
reflects the real signal and verdict transitions are logged. Adds
BackpressureMetrics (engaged/dry_run gauges, skipped-dequeues counter).

apps/supervisor/src/backpressure/backpressureMetrics.ts

@@ -0,0 +1,34 @@
+import { Counter, Gauge, type Registry } from "prom-client";
+
+/** Prometheus metrics for dequeue backpressure. */
+export class BackpressureMetrics {
+  /** 1 while backpressure is engaged (computed signal, set even in dry-run). */
+  readonly engaged: Gauge<string>;
+  /** 1 when running in dry-run (gates inert). */
+  readonly dryRun: Gauge<string>;
+  /** Dequeue attempts the gate skipped - or would have, in dry-run (labelled). */
+  readonly skipsTotal: Counter<string>;
+
+  constructor(opts: { register: Registry; prefix?: string }) {
+    const prefix = opts.prefix ?? "supervisor_backpressure";
+
+    this.engaged = new Gauge({
+      name: `${prefix}_engaged`,
+      help: "1 while dequeue backpressure is engaged (computed signal, regardless of dry-run)",
+      registers: [opts.register],
+    });
+
+    this.dryRun = new Gauge({
+      name: `${prefix}_dry_run`,
+      help: "1 when dequeue backpressure is in dry-run mode (gates inert)",
+      registers: [opts.register],
+    });
+
+    this.skipsTotal = new Counter({
+      name: `${prefix}_skipped_dequeues_total`,
+      help: "Dequeue attempts skipped by backpressure (or would be, in dry-run)",
+      labelNames: ["dry_run"],
+      registers: [opts.register],
+    });
+  }
+}

apps/supervisor/src/backpressure/backpressureMonitor.test.ts

@@ -1,5 +1,7 @@
 import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { Registry } from "prom-client";
 import { BackpressureMonitor, type BackpressureSignalSource } from "./backpressureMonitor.js";
+import { BackpressureMetrics } from "./backpressureMetrics.js";
 
 function countingSource(verdict: { engaged: boolean } | null): {
   source: BackpressureSignalSource;
@@ -225,6 +227,74 @@ describe("BackpressureMonitor", () => {
     monitor.stop();
   });
 
+  it("in dry-run, the gates are inert but computeEngaged still reflects the real signal", async () => {
+    const { source } = countingSource({ engaged: true });
+    const monitor = new BackpressureMonitor({
+      enabled: true,
+      source,
+      refreshIntervalMs: 1000,
+      dryRun: true,
+    });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+
+    expect(monitor.computeEngaged()).toBe(true); // real signal, for observability/metrics
+    expect(monitor.isEngaged()).toBe(false); // inert: no scale-up freeze
+    expect(monitor.shouldSkipDequeue()).toBe(false); // inert: no dequeue skip
+
+    monitor.stop();
+  });
+
+  it("logs on verdict transitions", async () => {
+    let engaged = true;
+    const source: BackpressureSignalSource = { read: async () => ({ engaged }) };
+    const logs: Array<{ message: string; meta?: Record<string, unknown> }> = [];
+    const logger = {
+      info: (message: string, meta?: Record<string, unknown>) => logs.push({ message, meta }),
+    };
+    const monitor = new BackpressureMonitor({
+      enabled: true,
+      source,
+      refreshIntervalMs: 1000,
+      logger,
+    });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+    expect(logs.some((l) => l.meta?.engaged === true)).toBe(true);
+
+    engaged = false;
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(logs.some((l) => l.meta?.engaged === false)).toBe(true);
+
+    monitor.stop();
+  });
+
+  it("records prometheus metrics", async () => {
+    const { source } = countingSource({ engaged: true });
+    const register = new Registry();
+    const metrics = new BackpressureMetrics({ register });
+    const monitor = new BackpressureMonitor({
+      enabled: true,
+      source,
+      refreshIntervalMs: 1000,
+      metrics,
+    });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+
+    expect(await register.metrics()).toContain("supervisor_backpressure_engaged 1");
+
+    monitor.shouldSkipDequeue();
+    expect(await register.metrics()).toMatch(
+      /supervisor_backpressure_skipped_dequeues_total\{dry_run="false"\} [1-9]/
+    );
+
+    monitor.stop();
+  });
+
   it("resumes instantly when no ramp is configured", async () => {
     let engaged = true;
     const source: BackpressureSignalSource = { read: async () => ({ engaged }) };

apps/supervisor/src/backpressure/backpressureMonitor.ts

@@ -1,3 +1,9 @@
+import type { BackpressureMetrics } from "./backpressureMetrics.js";
+
+export interface BackpressureLogger {
+  info(message: string, meta?: Record<string, unknown>): void;
+}
+
 export type BackpressureVerdict = {
   engaged: boolean;
   /** Epoch ms the verdict was produced. Used for consumer-side staleness fail-open. */
@@ -31,6 +37,13 @@ export type BackpressureMonitorOptions = {
   rampMs?: number;
   /** Injectable RNG for the resume ramp; defaults to Math.random. */
   random?: () => number;
+  /**
+   * When true, the gates are inert (never skip dequeues, never freeze scale-up).
+   * computeEngaged() still reflects the real signal so it can be observed.
+   */
+  dryRun?: boolean;
+  logger?: BackpressureLogger;
+  metrics?: BackpressureMetrics;
 };
 
 const DEFAULT_REFRESH_INTERVAL_MS = 1000;
@@ -41,7 +54,9 @@ export class BackpressureMonitor {
   private wasEngaged = false;
   private releasedAt?: number;
 
-  constructor(private readonly opts: BackpressureMonitorOptions) {}
+  constructor(private readonly opts: BackpressureMonitorOptions) {
+    this.opts.metrics?.dryRun.set(this.opts.dryRun ? 1 : 0);
+  }
 
   start(): void {
     if (!this.opts.enabled) {
@@ -63,11 +78,11 @@ export class BackpressureMonitor {
   }
 
   /**
-   * Hard backpressure state: true while the (fresh) verdict says engaged. This is
-   * the signal for freezing consumer-pool scale-up - distinct from the dequeue
-   * gate, which additionally ramps after release. Hot-path read, no I/O.
+   * Raw hard backpressure state: true while the (fresh) verdict says engaged,
+   * ignoring dry-run. Used for observability/metrics so the real signal is
+   * visible even when the gates are inert.
    */
-  isEngaged(): boolean {
+  computeEngaged(): boolean {
     const verdict = this.verdict;
     if (verdict?.engaged !== true) {
       return false;
@@ -81,9 +96,25 @@ export class BackpressureMonitor {
     return true;
   }
 
-  /** Hot-path read: synchronous, never performs I/O. */
+  /**
+   * Effective hard state: the signal for freezing consumer-pool scale-up. Inert
+   * (false) in dry-run. Hot-path read, no I/O.
+   */
+  isEngaged(): boolean {
+    return this.opts.dryRun ? false : this.computeEngaged();
+  }
+
+  /** Hot-path read: synchronous, never performs I/O. Inert (false) in dry-run. */
   shouldSkipDequeue(): boolean {
-    if (this.isEngaged()) {
+    const wouldSkip = this.computeShouldSkip();
+    if (wouldSkip) {
+      this.opts.metrics?.skipsTotal.inc({ dry_run: this.opts.dryRun ? "true" : "false" });
+    }
+    return this.opts.dryRun ? false : wouldSkip;
+  }
+
+  private computeShouldSkip(): boolean {
+    if (this.computeEngaged()) {
       return true;
     }
 
@@ -113,6 +144,14 @@ export class BackpressureMonitor {
     // Track the engaged→released transition to anchor the resume ramp. Based on
     // the raw refreshed verdict, not the staleness-adjusted read.
     const nowEngaged = this.verdict?.engaged === true;
+    this.opts.metrics?.engaged.set(nowEngaged ? 1 : 0);
+
+    if (nowEngaged !== this.wasEngaged) {
+      this.opts.logger?.info("backpressure verdict changed", {
+        engaged: nowEngaged,
+        dryRun: !!this.opts.dryRun,
+      });
+    }
     if (this.wasEngaged && !nowEngaged) {
       this.releasedAt = Date.now();
     }

apps/supervisor/src/env.ts

@@ -52,6 +52,9 @@ const Env = z
     // while the worker cluster can't schedule pods. Disabled = total no-op: no Redis
     // client is created, no reads happen, and the dequeue loop is unaffected.
     TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED: BoolEnv.default(false),
+    // Safety default: even when enabled, backpressure only logs what it would do.
+    // Set to false to actually skip dequeues / freeze scale-up.
+    TRIGGER_DEQUEUE_BACKPRESSURE_DRY_RUN: BoolEnv.default(true),
     TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_KEY: z.string().default("engine:dequeue:backpressure"),
     TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS: z.coerce.number().int().positive().default(1000),
     TRIGGER_DEQUEUE_BACKPRESSURE_RAMP_MS: z.coerce.number().int().min(0).default(30_000), // Resume ramp window after release; 0 = instant resume

apps/supervisor/src/index.ts

@@ -31,6 +31,7 @@ import { extractTraceparent, getRestoreRunnerId } from "./util.js";
 import { createRedisClient } from "@internal/redis";
 import { BackpressureMonitor } from "./backpressure/backpressureMonitor.js";
 import { RedisBackpressureSignalSource } from "./backpressure/redisBackpressureSignalSource.js";
+import { BackpressureMetrics } from "./backpressure/backpressureMetrics.js";
 import {
   fromContext,
   recordPhaseSince,
@@ -209,6 +210,9 @@ class ManagedSupervisor {
         refreshIntervalMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS,
         maxVerdictAgeMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_MAX_VERDICT_AGE_MS,
         rampMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_RAMP_MS,
+        dryRun: env.TRIGGER_DEQUEUE_BACKPRESSURE_DRY_RUN,
+        logger: this.logger,
+        metrics: new BackpressureMetrics({ register }),
       });
 
       this.logger.log("🛑 Dequeue backpressure enabled", {
@@ -216,6 +220,7 @@ class ManagedSupervisor {
         refreshIntervalMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS,
         maxVerdictAgeMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_MAX_VERDICT_AGE_MS,
         rampMs: env.TRIGGER_DEQUEUE_BACKPRESSURE_RAMP_MS,
+        dryRun: env.TRIGGER_DEQUEUE_BACKPRESSURE_DRY_RUN,
       });
     }