feat(supervisor): split backpressure signal and add resume ramp

isEngaged() exposes the hard backpressure state (drives scale-up freeze),
while shouldSkipDequeue() additionally ramps after release - skipping a
linearly-decaying fraction of attempts over rampMs so the aggregate dequeue
rate climbs back to full instead of snapping and re-flooding the cluster.

Author: nicktrn

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

@@ -151,4 +151,93 @@ describe("BackpressureMonitor", () => {
 
     expect(reads()).toBe(readsAtStop);
   });
+
+  it("isEngaged reflects the hard engaged state (the signal for freezing scale-up)", async () => {
+    const { source } = countingSource({ engaged: true });
+    const monitor = new BackpressureMonitor({ enabled: true, source, refreshIntervalMs: 1000 });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+
+    expect(monitor.isEngaged()).toBe(true);
+
+    monitor.stop();
+  });
+
+  it("isEngaged is false when clear and when stale", async () => {
+    const source: BackpressureSignalSource = {
+      read: async () => ({ engaged: true, ts: Date.now() }),
+    };
+    const monitor = new BackpressureMonitor({
+      enabled: true,
+      source,
+      refreshIntervalMs: 1_000_000,
+      maxVerdictAgeMs: 15_000,
+    });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+    expect(monitor.isEngaged()).toBe(true);
+
+    await vi.advanceTimersByTimeAsync(15_001); // stale → fail-open
+    expect(monitor.isEngaged()).toBe(false);
+
+    monitor.stop();
+  });
+
+  it("ramps the dequeue gate after release instead of resuming instantly", async () => {
+    let engaged = true;
+    let rnd = 0.5;
+    const source: BackpressureSignalSource = { read: async () => ({ engaged }) };
+    const monitor = new BackpressureMonitor({
+      enabled: true,
+      source,
+      refreshIntervalMs: 1000,
+      rampMs: 10_000,
+      random: () => rnd,
+    });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+    expect(monitor.shouldSkipDequeue()).toBe(true); // hard engaged
+
+    // Release: the next refresh observes the clear verdict and starts the ramp.
+    engaged = false;
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(monitor.isEngaged()).toBe(false);
+
+    // Just after release (progress ~0): skip probability ~1, so skip regardless.
+    rnd = 0.99;
+    expect(monitor.shouldSkipDequeue()).toBe(true);
+
+    // Halfway through the ramp (progress 0.5): skip probability 0.5.
+    await vi.advanceTimersByTimeAsync(5000);
+    rnd = 0.4;
+    expect(monitor.shouldSkipDequeue()).toBe(true); // 0.4 < 0.5 → skip
+    rnd = 0.6;
+    expect(monitor.shouldSkipDequeue()).toBe(false); // 0.6 ≥ 0.5 → allow
+
+    // Past the ramp window: never skip.
+    await vi.advanceTimersByTimeAsync(5000);
+    rnd = 0.0;
+    expect(monitor.shouldSkipDequeue()).toBe(false);
+
+    monitor.stop();
+  });
+
+  it("resumes instantly when no ramp is configured", async () => {
+    let engaged = true;
+    const source: BackpressureSignalSource = { read: async () => ({ engaged }) };
+    const monitor = new BackpressureMonitor({ enabled: true, source, refreshIntervalMs: 1000 });
+
+    monitor.start();
+    await vi.advanceTimersByTimeAsync(0);
+    expect(monitor.shouldSkipDequeue()).toBe(true);
+
+    engaged = false;
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(monitor.shouldSkipDequeue()).toBe(false); // no ramp → instant resume
+
+    monitor.stop();
+  });
 });

apps/supervisor/src/backpressure/backpressureMonitor.ts

@@ -22,13 +22,24 @@ export type BackpressureMonitorOptions = {
    * Guards against the source silently going stale (e.g. hanging reads).
    */
   maxVerdictAgeMs?: number;
+  /**
+   * If set, after backpressure releases the dequeue gate stays partially engaged
+   * for this long, skipping a linearly-decaying fraction of attempts so the
+   * aggregate dequeue rate ramps from ~0 to full instead of snapping to full and
+   * re-flooding a freshly-recovered cluster. 0/unset = instant resume.
+   */
+  rampMs?: number;
+  /** Injectable RNG for the resume ramp; defaults to Math.random. */
+  random?: () => number;
 };
 
 const DEFAULT_REFRESH_INTERVAL_MS = 1000;
 
 export class BackpressureMonitor {
   private verdict: BackpressureVerdict | null = null;
   private timer?: ReturnType<typeof setInterval>;
+  private wasEngaged = false;
+  private releasedAt?: number;
 
   constructor(private readonly opts: BackpressureMonitorOptions) {}
 
@@ -51,8 +62,12 @@ export class BackpressureMonitor {
     }
   }
 
-  /** Hot-path read: synchronous, never performs I/O. */
-  shouldSkipDequeue(): boolean {
+  /**
+   * 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.
+   */
+  isEngaged(): boolean {
     const verdict = this.verdict;
     if (verdict?.engaged !== true) {
       return false;
@@ -66,6 +81,26 @@ export class BackpressureMonitor {
     return true;
   }
 
+  /** Hot-path read: synchronous, never performs I/O. */
+  shouldSkipDequeue(): boolean {
+    if (this.isEngaged()) {
+      return true;
+    }
+
+    // Post-release ramp: skip a linearly-decaying fraction of attempts so the
+    // aggregate dequeue rate climbs back to full over rampMs rather than snapping.
+    const rampMs = this.opts.rampMs;
+    if (rampMs && this.releasedAt !== undefined) {
+      const elapsed = Date.now() - this.releasedAt;
+      if (elapsed < rampMs) {
+        const skipProbability = 1 - elapsed / rampMs;
+        return (this.opts.random ?? Math.random)() < skipProbability;
+      }
+    }
+
+    return false;
+  }
+
   private async refresh(): Promise<void> {
     try {
       this.verdict = await this.opts.source.read();
@@ -74,5 +109,13 @@ export class BackpressureMonitor {
       // unknown (no verdict) so dequeue resumes as if backpressure were off.
       this.verdict = null;
     }
+
+    // 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;
+    if (this.wasEngaged && !nowEngaged) {
+      this.releasedAt = Date.now();
+    }
+    this.wasEngaged = nowEngaged;
   }
 }