pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-08-19 21:52:21 +03:00
Code Issues Packages Projects Releases Wiki Activity

feat(resilience): abort persistently slow upstream streams (#9709) (#10012)

Browse Source 
Co-authored-by: diegosouzapw <diegosouzapw@users.noreply.github.com>
This commit is contained in:
Diego Rodrigues de Sa e Souza
2026-08-10 18:16:13 -03:00
committed by GitHub
parent 227795cd08
commit d3037d1fdc
13 changed files with 627 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
.env.example
View File

@@ -1854,6 +1854,17 @@ APP_LOG_TO_FILE=true
# Accepted values: true|1|on (enable). Unset or anything else = disabled (default).
# STREAM_RECOVERY_MIDSTREAM_ENABLED=true
# Active-stream throughput watchdog (#9709). Detects streams that keep sending
# heartbeats/chunks but produce too little useful assistant text. Separate from
# STREAM_IDLE_TIMEOUT_MS (silence) and the hard upstream attempt deadline. OFF by
# default. Tool-call/reasoning phases suspend judgement; post-commit streams are
# never blindly replayed.
# STREAM_THROUGHPUT_WATCHDOG_ENABLED=true
# STREAM_THROUGHPUT_WATCHDOG_WARMUP_MS=30000
# STREAM_THROUGHPUT_WATCHDOG_WINDOW_MS=30000
# STREAM_THROUGHPUT_WATCHDOG_MIN_BYTES_PER_SECOND=4
# STREAM_THROUGHPUT_WATCHDOG_MIN_USEFUL_BYTES=1
# Stagger interval (ms) between provider token healthchecks at startup.
# Used by: src/lib/tokenHealthCheck.ts. Default: 3000.
# HEALTHCHECK_STAGGER_MS=3000

1
changelog.d/features/9709-stream-throughput-watchdog.md Normal file
View File

@@ -0,0 +1 @@
- feat(resilience): add an opt-in watchdog for persistently slow upstream streams (#9709)

25
docs/architecture/RESILIENCE_GUIDE.md
View File

@@ -260,6 +260,31 @@ it is unit-testable without a real Bottleneck limiter.
---
## 6. Slow-stream throughput watchdog (#9709)
The optional `resilienceSettings.streamRecovery.throughputWatchdog` guard detects
an upstream that is still sending chunks but producing assistant output below the
configured useful-output rate. It is deliberately distinct from the idle timeout:
heartbeats and metadata reset neither timer and do not count as progress. It is also
distinct from the hard attempt deadline (#9153), which remains an absolute safety
ceiling regardless of output quality.
The watchdog requires a warm-up period followed by a complete rolling window before
it can abort. It counts text deltas from Chat Completions and Responses API output
events (a conservative UTF-8 byte proxy), ignores usage-only and empty events, and
suspends judgement while tool-call or reasoning events are in flight. It is disabled
by default and can be enabled with `STREAM_THROUGHPUT_WATCHDOG_ENABLED=true`; the
window, warm-up, minimum rate, and minimum measurable output are bounded by the
normal resilience-settings normalization layer.
When enabled, a watchdog abort is applied only to the active upstream attempt. Before
any client-visible bytes, the existing same-account early-recovery path may reopen
the attempt. After commit, the stream is never blindly replayed; only the existing
safe mid-stream continuation contract can stitch a suffix. Finalization remains
single-shot, so usage accounting and semaphore release are not duplicated.
---
## Other Resilience Features
- **19 routing strategies** (priority, weighted, round-robin, context-relay, fill-first, p2c, random, least-used, cost-optimized, reset-aware, reset-window, headroom, strict-random, auto, lkgp, context-optimized, cache-optimized, fusion, pipeline) — see [AUTO-COMBO.md](../routing/AUTO-COMBO.md).

10
docs/reference/ENVIRONMENT.md
View File

@@ -930,6 +930,11 @@ Anthropic-compatible provider instead.
| `PROVIDER_COOLDOWN_MAX_MS` | `300000` (5 min) | `open-sse/services/providerCooldownTracker.ts` | Maximum cooldown (ms) cap before a failed provider/connection is retried regardless. Only used when `PROVIDER_COOLDOWN_ENABLED`. |
| `STREAM_RECOVERY_ENABLED` | _(unset → off)_ | `src/lib/resilience/settings.ts` (seed) → `open-sse/services/streamRecovery.ts` (logic) | **What:** transparent recovery of truncated upstream streams (free-claude-code port). Holds the opening SSE window up to `STREAM_RECOVERY.HOLDBACK_MS` (750 ms) so a _pre-commit_ cutoff — one that happens before any byte reaches the client — is re-opened and retried invisibly. **When to enable:** flaky/upstreams that frequently 0-byte-truncate at stream start; leave OFF if you cannot afford up to 750 ms of added time-to-first-token on every stream. Accepts `true`/`1`/`on`. Seeds the persisted Resilience setting; the Dashboard setting wins once set. |
| `STREAM_RECOVERY_MIDSTREAM_ENABLED` | _(unset → off)_ | `src/lib/resilience/settings.ts` (seed) → `open-sse/services/streamRecovery.ts` (logic) | **What:** mid-stream continuation (Fase 4.4) — after a _post-commit_ truncation (bytes already reached the client), re-request with the partial text as an assistant prefill and stitch the missing suffix. Plain-text OpenAI-compatible streams only; never fires with a tool call in flight. **When to enable:** long generations that get cut mid-answer and you accept the recovered tail arriving as one burst rather than token-by-token. Independent of `STREAM_RECOVERY_ENABLED` (different risk profile). Accepts `true`/`1`/`on`. |
| `STREAM_THROUGHPUT_WATCHDOG_ENABLED` | _(unset → off)_ | `src/lib/resilience/settings.ts` → `open-sse/services/throughputWatchdog.ts` | Opt-in active-stream useful-output watchdog. Detects streams that keep sending chunks but remain below the configured assistant-output rate; heartbeats, usage events, empty deltas, and tool/reasoning phases do not masquerade as progress. Separate from idle and hard-deadline timeouts. |
| `STREAM_THROUGHPUT_WATCHDOG_WARMUP_MS` | `30000` | `src/lib/resilience/settings/normalize.ts` | Grace period before throughput evaluation, bounded to 0–600000 ms. |
| `STREAM_THROUGHPUT_WATCHDOG_WINDOW_MS` | `30000` | `src/lib/resilience/settings/normalize.ts` | Rolling useful-output window, bounded to 1000–600000 ms; one complete window is required before abort. |
| `STREAM_THROUGHPUT_WATCHDOG_MIN_BYTES_PER_SECOND` | `4` | `src/lib/resilience/settings/normalize.ts` | Minimum UTF-8 assistant-output byte rate (conservative token proxy), bounded to 1–1000000. |
| `STREAM_THROUGHPUT_WATCHDOG_MIN_USEFUL_BYTES` | `1` | `src/lib/resilience/settings/normalize.ts` | Minimum non-zero useful-output sample considered measurable, bounded to 1–1000000 bytes. |
| `HEALTHCHECK_STAGGER_MS` | `3000` | `src/lib/tokenHealthCheck.ts` | Stagger interval (ms) between provider token healthchecks at startup. |
| `HEALTHCHECK_JITTER_MIN_MS` | `500` | `src/lib/tokenHealthCheck.ts` | Minimum randomized jitter (ms) added on top of `HEALTHCHECK_STAGGER_MS` between provider token healthchecks, to prevent bursting (Issue #1220). |
| `HEALTHCHECK_JITTER_MAX_MS` | `5000` | `src/lib/tokenHealthCheck.ts` | Maximum randomized jitter (ms) added on top of `HEALTHCHECK_STAGGER_MS` between provider token healthchecks, to prevent bursting (Issue #1220). |
@@ -938,10 +943,9 @@ Anthropic-compatible provider instead.
| `MAX_RETRY_INTERVAL_SEC` | `30` | `src/sse/services/cooldownAwareRetry.ts` | Max backoff interval (seconds) between cooldown retries. Capped by this value regardless of upstream `Retry-After`. |
| `HEADROOM_URL` | `http://localhost:8787` | `src/lib/headroom/detect.ts` | Headroom token-saver proxy URL. The dashboard lifecycle (`api/headroom/*`) spawns a local `headroom-ai` CLI on loopback by default; override only to point at an external Docker sidecar proxy. |
### Stream-recovery tuning constants (not env vars)
### Stream-recovery tuning constants
The two `STREAM_RECOVERY_*` flags above are the only operator-facing toggles. The
recovery behavior is otherwise tuned by hardcoded constants in
The recovery holdback behavior is tuned by hardcoded constants in
`open-sse/config/constants.ts` (`STREAM_RECOVERY`), shown here for reference —
changing them requires a code edit, not an env var:

12
open-sse/config/constants.ts
View File

@@ -330,3 +330,15 @@ export const STREAM_RECOVERY = {
BUFFER_MAX_BYTES: 65536,
EARLY_RETRY_MAX: 4,
} as const;
/**
* Active-stream quality watchdog defaults (#9709). This is separate from the
* idle timeout (no chunks) and the absolute upstream-attempt deadline: it only
* evaluates useful assistant output after warm-up plus one complete window.
*/
export const STREAM_THROUGHPUT_WATCHDOG = {
WARMUP_MS: 30_000,
WINDOW_MS: 30_000,
MIN_USEFUL_BYTES_PER_SECOND: 4,
MIN_USEFUL_BYTES: 1,
} as const;

13
open-sse/handlers/chatCore.ts
View File

@@ -2976,6 +2976,8 @@ export async function handleChatCore({
const okStatus = res.response.status >= 200 && res.response.status < 300;
let streamRecoveryEnabled = false;
let continueMidStreamEnabled = false;
let throughputWatchdog =
resolveResilienceSettings(null).streamRecovery.throughputWatchdog;
if (okStatus) {
try {
// Reuse the request-consolidated settings read (see line ~2076) — no
@@ -2990,6 +2992,7 @@ export async function handleChatCore({
const goalOverride = !operatorExplicit && agentGoalPolicy.streamRecoveryEnabled;
streamRecoveryEnabled = sr.enabled || goalOverride;
continueMidStreamEnabled = sr.continueMidStream === true;
throughputWatchdog = sr.throughputWatchdog;
if (goalOverride && !sr.enabled) {
log?.info?.(
"AGENT_GOAL",
@@ -2999,11 +3002,13 @@ export async function handleChatCore({
} catch {
streamRecoveryEnabled = false;
continueMidStreamEnabled = false;
throughputWatchdog =
resolveResilienceSettings(null).streamRecovery.throughputWatchdog;
}
}
let clientBody: ReadableStream<Uint8Array>;
if (streamRecoveryEnabled) {
if (streamRecoveryEnabled || throughputWatchdog.enabled) {
// Run the SAME upstream (same account/creds) with a given body and return
// its 2xx stream, or null. Used both by the early-retry re-open (same body)
// and the mid-stream continuation (assistant-prefilled body).
@@ -3085,6 +3090,12 @@ export async function handleChatCore({
"STREAM_RECOVERY",
`mid-stream continuation attempt ${attempt}/${STREAM_RECOVERY.EARLY_RETRY_MAX}`
),
throughputWatchdog,
onWatchdogAbort: () =>
log?.warn?.(
"STREAM_WATCHDOG",
"active upstream stream stayed below the configured useful-output rate"
),
}
);
} else {

43
open-sse/services/streamRecovery.ts
View File

@@ -11,6 +11,13 @@
* without real sockets. The ReadableStream wiring lives in `createRecoverableStream`.
*/
import { STREAM_RECOVERY } from "../config/constants.ts";
import {
createThroughputWatchdog,
ThroughputWatchdogError,
type ThroughputWatchdogOptions,
} from "./throughputWatchdog.ts";
export { ThroughputWatchdogError } from "./throughputWatchdog.ts";
/** Raised internally when an upstream stream ends without a terminal SSE marker. */
export class TruncatedStreamError extends Error {
@@ -123,7 +130,9 @@ const RETRYABLE_ERROR_NAMES = new Set(["TimeoutError", "BodyTimeoutError"]);
* the executor retry/failover loop, not here.
*/
export function isRetryableStreamError(error: unknown): boolean {
if (error instanceof TruncatedStreamError) return true;
if (error instanceof TruncatedStreamError || error instanceof ThroughputWatchdogError) {
return true;
}
if (!error || typeof error !== "object") return false;
const name = (error as { name?: unknown }).name;
@@ -289,6 +298,10 @@ export interface RecoverableStreamOptions {
maxContinuations?: number;
/** Observability hook fired on each continuation attempt. */
onContinue?: (attempt: number, assistantSoFar: string) => void;
/** Opt-in active-stream output-quality watchdog. Disabled when omitted. */
throughputWatchdog?: ThroughputWatchdogOptions;
/** Sanitized observability hook fired before the active attempt is aborted. */
onWatchdogAbort?: (error: ThroughputWatchdogError) => void;
}
/**
@@ -312,6 +325,7 @@ export function createRecoverableStream(
let retries = 0;
let finalized = false;
let cancelled = false;
let throughputWatchdog = createThroughputWatchdog(options.throughputWatchdog);
const runFinalize = () => {
if (finalized) return;
@@ -342,6 +356,7 @@ export function createRecoverableStream(
if (!next) return false;
reader = next.getReader();
holdback.discard(); // reuse the (still-uncommitted) buffer for the new attempt
throughputWatchdog = createThroughputWatchdog(options.throughputWatchdog);
return true;
};
@@ -519,6 +534,32 @@ export function createRecoverableStream(
if (value === undefined) continue;
const watchdogDecision = throughputWatchdog.observe(value);
if (watchdogDecision.abort) {
const error = new ThroughputWatchdogError();
options.onWatchdogAbort?.(error);
if (!holdback.committed && (await tryReopen(error))) continue;
if (holdback.committed) {
try {
await reader.cancel(error);
} catch {
// The active attempt may have closed while the watchdog was deciding.
}
if (await tryContinue(controller)) {
runFinalize();
controller.close();
return;
}
runFinalize();
controller.error(error);
return;
}
flushHeld(controller);
runFinalize();
controller.close();
return;
}
if (holdback.committed) {
emit(controller, value);
return;

175
open-sse/services/throughputWatchdog.ts Normal file
View File

@@ -0,0 +1,175 @@
/**
* Deterministic quality watchdog for active SSE streams.
*
* Unlike the idle timeout, this only makes a decision after a warm-up period and
* a complete rolling window. Heartbeats/metadata and tool/reasoning phases do not
* count as assistant output (and tool/reasoning phases suspend judgement).
*/
export interface ThroughputWatchdogOptions {
enabled?: boolean;
warmupMs?: number;
windowMs?: number;
minUsefulBytesPerSecond?: number;
minUsefulBytes?: number;
now?: () => number;
}
export interface ThroughputWatchdogDecision {
abort: boolean;
reason?: "throughput_too_low";
usefulBytes: number;
rateBytesPerSecond: number;
protectedPhase: boolean;
}
export class ThroughputWatchdogError extends Error {
readonly code = "STREAM_THROUGHPUT_TOO_LOW";
constructor(message = "Upstream stream throughput remained below the configured minimum") {
super(message);
this.name = "ThroughputWatchdogError";
}
}
type ParsedEvent = { usefulBytes: number; protectedPhase: boolean };
function parseEvent(event: string): ParsedEvent {
const lines = event.split(/\r?\n/);
const eventName = lines
.find((line) => /^event:\s*/i.test(line))
?.replace(/^event:\s*/i, "")
.trim();
const data = lines
.filter((line) => /^data:\s*/i.test(line))
.map((line) => line.replace(/^data:\s*/i, "").trim())
.join("\n");
if (!data || data === "[DONE]") return { usefulBytes: 0, protectedPhase: false };
let payload: unknown;
try {
payload = JSON.parse(data);
} catch {
return { usefulBytes: 0, protectedPhase: false };
}
const record = payload as Record<string, unknown>;
const type = typeof record.type === "string" ? record.type : eventName;
if (type && /(reasoning|thinking|tool|function_call)/i.test(type)) {
return { usefulBytes: 0, protectedPhase: true };
}
const choices = Array.isArray(record.choices) ? record.choices : [];
let useful = "";
let protectedPhase = false;
for (const choice of choices) {
const delta = (choice as Record<string, unknown>).delta;
if (!delta || typeof delta !== "object") continue;
const deltaRecord = delta as Record<string, unknown>;
if (Array.isArray(deltaRecord.tool_calls) || deltaRecord.function_call) {
protectedPhase = true;
}
for (const key of ["content", "text"]) {
if (typeof deltaRecord[key] === "string") useful += deltaRecord[key] as string;
}
if (
typeof deltaRecord.reasoning_content === "string" ||
typeof deltaRecord.reasoning === "string"
) {
protectedPhase = true;
}
}
const outputText = typeof record.delta === "string" ? record.delta : undefined;
if (outputText) useful += outputText;
const nestedDelta = record.delta;
if (nestedDelta && typeof nestedDelta === "object") {
const nested = nestedDelta as Record<string, unknown>;
const nestedType = typeof nested.type === "string" ? nested.type : "";
if (/(reasoning|thinking|tool|function_call)/i.test(nestedType)) {
protectedPhase = true;
}
if (typeof nested.text === "string") useful += nested.text;
}
const contentBlock = record.content_block;
if (contentBlock && typeof contentBlock === "object") {
const blockType = (contentBlock as Record<string, unknown>).type;
if (typeof blockType === "string" && /(reasoning|thinking|tool_use)/i.test(blockType)) {
protectedPhase = true;
}
}
if (protectedPhase) useful = "";
return {
usefulBytes: useful ? new TextEncoder().encode(useful).byteLength : 0,
protectedPhase,
};
}
export class ThroughputWatchdog {
private readonly enabled: boolean;
private readonly warmupMs: number;
private readonly windowMs: number;
private readonly minimumRate: number;
private readonly minimumBytes: number;
private readonly now: () => number;
private startedAt: number | null = null;
private buffer = "";
private readonly decoder = new TextDecoder();
private samples: Array<{ at: number; bytes: number }> = [];
private protectedPhase = false;
constructor(options: ThroughputWatchdogOptions = {}) {
this.enabled = options.enabled === true;
this.warmupMs = Math.max(0, Math.floor(options.warmupMs ?? 30_000));
this.windowMs = Math.max(1, Math.floor(options.windowMs ?? 30_000));
this.minimumRate = Math.max(0, options.minUsefulBytesPerSecond ?? 1);
this.minimumBytes = Math.max(1, Math.floor(options.minUsefulBytes ?? 1));
this.now = options.now ?? (() => Date.now());
}
observe(chunk: Uint8Array | string): ThroughputWatchdogDecision {
const at = this.now();
if (this.startedAt === null) this.startedAt = at;
if (!this.enabled) return this.decision(false, 0);
this.buffer += typeof chunk === "string" ? chunk : this.decoder.decode(chunk, { stream: true });
const events = this.buffer.split(/\r?\n\r?\n/);
this.buffer = events.pop() ?? "";
let useful = 0;
for (const event of events) {
const parsed = parseEvent(event);
useful += parsed.usefulBytes;
if (parsed.protectedPhase) this.protectedPhase = true;
if (parsed.usefulBytes > 0) this.protectedPhase = false;
}
if (useful > 0) this.samples.push({ at, bytes: useful });
const cutoff = at - this.windowMs;
this.samples = this.samples.filter((sample) => sample.at >= cutoff);
const windowBytes = this.samples.reduce((sum, sample) => sum + sample.bytes, 0);
const elapsed = at - (this.startedAt ?? at);
const rate = windowBytes / Math.max(1, this.windowMs / 1000);
const ready = elapsed >= this.warmupMs + this.windowMs;
const measurable = windowBytes === 0 || windowBytes >= this.minimumBytes;
const abort = ready && !this.protectedPhase && measurable && rate < this.minimumRate;
return this.decision(abort, windowBytes, rate);
}
private decision(
abort: boolean,
usefulBytes: number,
rateBytesPerSecond = 0
): ThroughputWatchdogDecision {
return {
abort,
reason: abort ? "throughput_too_low" : undefined,
usefulBytes,
rateBytesPerSecond,
protectedPhase: this.protectedPhase,
};
}
}
export function createThroughputWatchdog(
options: ThroughputWatchdogOptions = {}
): ThroughputWatchdog {
return new ThroughputWatchdog(options);
}

16
src/lib/resilience/settings.ts
View File

@@ -1,4 +1,8 @@
import { DEFAULT_API_LIMITS, PROVIDER_PROFILES } from "@omniroute/open-sse/config/constants";
import {
DEFAULT_API_LIMITS,
PROVIDER_PROFILES,
STREAM_THROUGHPUT_WATCHDOG,
} from "@omniroute/open-sse/config/constants";
import type { JsonRecord, ResilienceSettings, ResilienceSettingsPatch } from "./settings/types";
import {
@@ -30,6 +34,7 @@ export type {
ProviderCooldownSettings,
QuotaPreflightSettings,
StreamRecoverySettings,
StreamThroughputWatchdogSettings,
ProviderQuotaOverrideSettings,
ResilienceSettings,
ResilienceSettingsPatch,
@@ -150,6 +155,15 @@ export const DEFAULT_RESILIENCE_SETTINGS: ResilienceSettings = {
continueMidStream: ["true", "1", "on"].includes(
(process.env.STREAM_RECOVERY_MIDSTREAM_ENABLED || "").trim().toLowerCase()
),
throughputWatchdog: {
enabled: ["true", "1", "on"].includes(
(process.env.STREAM_THROUGHPUT_WATCHDOG_ENABLED || "").trim().toLowerCase()
),
warmupMs: STREAM_THROUGHPUT_WATCHDOG.WARMUP_MS,
windowMs: STREAM_THROUGHPUT_WATCHDOG.WINDOW_MS,
minUsefulBytesPerSecond: STREAM_THROUGHPUT_WATCHDOG.MIN_USEFUL_BYTES_PER_SECOND,
minUsefulBytes: STREAM_THROUGHPUT_WATCHDOG.MIN_USEFUL_BYTES,
},
},
// #6846 Phase 1: empty by default — nvidia (and any future header-less
// provider registered in providerDefaultRateLimit.ts) uses its static

42
src/lib/resilience/settings/normalize.ts
View File

@@ -83,6 +83,26 @@ export function resolveStreamRecoveryDefaults(): StreamRecoverySettings {
return {
enabled: resolveBooleanFeatureFlag("STREAM_RECOVERY_ENABLED", false),
continueMidStream: resolveBooleanFeatureFlag("STREAM_RECOVERY_MIDSTREAM_ENABLED", false),
throughputWatchdog: {
enabled: resolveBooleanFeatureFlag("STREAM_THROUGHPUT_WATCHDOG_ENABLED", false),
warmupMs: toInteger(process.env.STREAM_THROUGHPUT_WATCHDOG_WARMUP_MS, 30_000, {
min: 0,
max: 10 * 60 * 1000,
}),
windowMs: toInteger(process.env.STREAM_THROUGHPUT_WATCHDOG_WINDOW_MS, 30_000, {
min: 1_000,
max: 10 * 60 * 1000,
}),
minUsefulBytesPerSecond: toInteger(
process.env.STREAM_THROUGHPUT_WATCHDOG_MIN_BYTES_PER_SECOND,
4,
{ min: 1, max: 1_000_000 }
),
minUsefulBytes: toInteger(process.env.STREAM_THROUGHPUT_WATCHDOG_MIN_USEFUL_BYTES, 1, {
min: 1,
max: 1_000_000,
}),
},
};
}
@@ -372,9 +392,31 @@ export function normalizeStreamRecoverySettings(
fallback: StreamRecoverySettings
): StreamRecoverySettings {
const record = asRecord(next);
const watchdog = asRecord(record.throughputWatchdog);
return {
enabled: toBoolean(record.enabled, fallback.enabled),
continueMidStream: toBoolean(record.continueMidStream, fallback.continueMidStream),
throughputWatchdog: {
enabled: toBoolean(watchdog.enabled, fallback.throughputWatchdog.enabled),
warmupMs: toInteger(watchdog.warmupMs, fallback.throughputWatchdog.warmupMs, {
min: 0,
max: 10 * 60 * 1000,
}),
windowMs: toInteger(watchdog.windowMs, fallback.throughputWatchdog.windowMs, {
min: 1_000,
max: 10 * 60 * 1000,
}),
minUsefulBytesPerSecond: toInteger(
watchdog.minUsefulBytesPerSecond,
fallback.throughputWatchdog.minUsefulBytesPerSecond,
{ min: 1, max: 1_000_000 }
),
minUsefulBytes: toInteger(
watchdog.minUsefulBytes,
fallback.throughputWatchdog.minUsefulBytes,
{ min: 1, max: 1_000_000 }
),
},
};
}

14
src/lib/resilience/settings/types.ts
View File

@@ -186,6 +186,20 @@ export interface StreamRecoverySettings {
* STREAM_RECOVERY_MIDSTREAM_ENABLED feature flag / env var.
*/
continueMidStream: boolean;
throughputWatchdog: StreamThroughputWatchdogSettings;
}
export interface StreamThroughputWatchdogSettings {
/** Opt-in; false preserves the existing stream byte path. */
enabled: boolean;
/** Grace period before the rolling window starts participating in decisions. */
warmupMs: number;
/** Full rolling window required before a slow-stream abort is possible. */
windowMs: number;
/** Minimum useful assistant-output byte rate. */
minUsefulBytesPerSecond: number;
/** Minimum amount required before a non-zero sample is considered measurable. */
minUsefulBytes: number;
}
export interface ResilienceSettings {

125
tests/unit/stream-throughput-watchdog-recovery.test.ts Normal file
View File

@@ -0,0 +1,125 @@
import { test } from "node:test";
import assert from "node:assert/strict";
import {
createRecoverableStream,
ThroughputWatchdogError,
} from "../../open-sse/services/streamRecovery.ts";
const encoder = new TextEncoder();
const heartbeat = 'event: ping\ndata: {"type":"ping"}\n\n';
const content = (text: string) =>
`data: ${JSON.stringify({ choices: [{ delta: { content: text } }] })}\n\n`;
function streamFrom(chunks: string[], beforeChunk?: (index: number) => void) {
let index = 0;
return new ReadableStream<Uint8Array>({
pull(controller) {
if (index >= chunks.length) {
controller.close();
return;
}
beforeChunk?.(index);
controller.enqueue(encoder.encode(chunks[index++]));
},
});
}
async function collect(stream: ReadableStream<Uint8Array>) {
const reader = stream.getReader();
const decoder = new TextDecoder();
let text = "";
let error: Error | null = null;
try {
for (;;) {
const result = await reader.read();
if (result.done) break;
if (result.value) text += decoder.decode(result.value, { stream: true });
}
} catch (caught) {
error = caught as Error;
}
return { text, error };
}
test("watchdog abort re-enters early recovery once and finalizes once", async () => {
let clock = 0;
let reopened = 0;
let finalized = 0;
let watchdogAborts = 0;
const initial = streamFrom([heartbeat, heartbeat, heartbeat], (index) => {
clock = index * 1_000;
});
const recovered = streamFrom([content("recovered"), "data: [DONE]\n\n"]);
const stream = createRecoverableStream(
initial,
async () => {
reopened += 1;
return recovered;
},
{
finalize: () => {
finalized += 1;
},
now: () => 0,
throughputWatchdog: {
enabled: true,
warmupMs: 0,
windowMs: 1_000,
minUsefulBytesPerSecond: 1,
minUsefulBytes: 1,
now: () => clock,
},
onWatchdogAbort: () => {
watchdogAborts += 1;
},
}
);
const result = await collect(stream);
assert.equal(result.error, null);
assert.equal(result.text, `${content("recovered")}data: [DONE]\n\n`);
assert.equal(reopened, 1);
assert.equal(watchdogAborts, 1);
assert.equal(finalized, 1);
});
test("post-commit watchdog abort never performs an unsafe full replay", async () => {
let clock = 0;
let reopened = 0;
let finalized = 0;
let holdbackClock = 0;
const initial = streamFrom([content("visible"), heartbeat, heartbeat], (index) => {
clock = index * 1_000;
});
const stream = createRecoverableStream(
initial,
async () => {
reopened += 1;
return streamFrom([content("duplicated"), "data: [DONE]\n\n"]);
},
{
finalize: () => {
finalized += 1;
},
now: () => {
holdbackClock += 1_000;
return holdbackClock;
},
throughputWatchdog: {
enabled: true,
warmupMs: 0,
windowMs: 1_000,
minUsefulBytesPerSecond: 100,
minUsefulBytes: 1,
now: () => clock,
},
}
);
const result = await collect(stream);
assert.equal(result.text, content("visible"));
assert.ok(result.error instanceof ThroughputWatchdogError);
assert.equal(reopened, 0, "visible output forbids transparent replay");
assert.equal(finalized, 1, "usage/semaphore finalization remains exactly once");
});

146
tests/unit/stream-throughput-watchdog.test.ts Normal file
View File

@@ -0,0 +1,146 @@
import { test } from "node:test";
import assert from "node:assert/strict";
import { createThroughputWatchdog } from "../../open-sse/services/throughputWatchdog.ts";
import {
DEFAULT_RESILIENCE_SETTINGS,
resolveResilienceSettings,
} from "../../src/lib/resilience/settings.ts";
const chatText = (text: string) =>
`data: ${JSON.stringify({ choices: [{ delta: { content: text } }] })}\n\n`;
const heartbeat = 'event: ping\ndata: {"type":"ping"}\n\n';
test("healthy sustained useful output never trips the throughput watchdog", () => {
let now = 0;
const watchdog = createThroughputWatchdog({
enabled: true,
warmupMs: 1_000,
windowMs: 2_000,
minUsefulBytesPerSecond: 2,
minUsefulBytes: 1,
now: () => now,
});
for (now = 0; now <= 8_000; now += 1_000) {
assert.equal(watchdog.observe(chatText("healthy output")).abort, false);
}
});
test("slow output aborts only after warm-up plus a complete window", () => {
let now = 0;
const watchdog = createThroughputWatchdog({
enabled: true,
warmupMs: 1_000,
windowMs: 2_000,
minUsefulBytesPerSecond: 10,
minUsefulBytes: 1,
now: () => now,
});
assert.equal(watchdog.observe(chatText("x")).abort, false);
now = 2_999;
assert.equal(watchdog.observe(heartbeat).abort, false);
now = 3_000;
const decision = watchdog.observe(heartbeat);
assert.equal(decision.abort, true);
assert.equal(decision.reason, "throughput_too_low");
});
test("heartbeats, metadata, usage, and empty deltas do not count as useful output", () => {
let now = 0;
const watchdog = createThroughputWatchdog({
enabled: true,
warmupMs: 0,
windowMs: 1_000,
minUsefulBytesPerSecond: 1,
minUsefulBytes: 1,
now: () => now,
});
watchdog.observe(heartbeat);
now = 500;
watchdog.observe('data: {"usage":{"output_tokens":99}}\n\n');
now = 999;
watchdog.observe('data: {"choices":[{"delta":{}}]}\n\n');
now = 1_000;
const decision = watchdog.observe(heartbeat);
assert.equal(decision.usefulBytes, 0);
assert.equal(decision.abort, true);
});
test("tool-call and reasoning phases suspend judgement until assistant text resumes", () => {
let now = 0;
const watchdog = createThroughputWatchdog({
enabled: true,
warmupMs: 0,
windowMs: 1_000,
minUsefulBytesPerSecond: 100,
minUsefulBytes: 1,
now: () => now,
});
watchdog.observe(
'data: {"choices":[{"delta":{"tool_calls":[{"function":{"name":"lookup"}}]}}]}\n\n'
);
now = 2_000;
assert.equal(watchdog.observe(heartbeat).abort, false);
assert.equal(watchdog.observe(heartbeat).protectedPhase, true);
watchdog.observe('data: {"type":"response.reasoning_summary_text.delta","delta":"thinking"}\n\n');
watchdog.observe(
'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"thinking_delta","thinking":"still thinking"}}\n\n'
);
now = 4_000;
assert.equal(watchdog.observe(heartbeat).abort, false);
assert.equal(watchdog.observe(chatText("answer")).protectedPhase, false);
});
test("Responses API output_text deltas count as useful assistant output", () => {
let now = 0;
const watchdog = createThroughputWatchdog({
enabled: true,
warmupMs: 0,
windowMs: 1_000,
minUsefulBytesPerSecond: 3,
minUsefulBytes: 1,
now: () => now,
});
watchdog.observe('data: {"type":"response.output_text.delta","delta":"hello"}\n\n');
now = 1_000;
const decision = watchdog.observe(heartbeat);
assert.equal(decision.usefulBytes, 5);
assert.equal(decision.abort, false);
});
test("disabled watchdog remains byte-path inert", () => {
let now = 0;
const watchdog = createThroughputWatchdog({ enabled: false, now: () => now });
for (now = 0; now <= 120_000; now += 30_000) {
assert.equal(watchdog.observe(heartbeat).abort, false);
}
});
test("resilience settings keep the watchdog disabled by default and bound overrides", () => {
assert.equal(DEFAULT_RESILIENCE_SETTINGS.streamRecovery.throughputWatchdog.enabled, false);
assert.equal(resolveResilienceSettings(null).streamRecovery.throughputWatchdog.enabled, false);
const resolved = resolveResilienceSettings({
resilienceSettings: {
streamRecovery: {
throughputWatchdog: {
enabled: true,
warmupMs: -1,
windowMs: 10,
minUsefulBytesPerSecond: 0,
minUsefulBytes: 0,
},
},
},
});
assert.deepEqual(resolved.streamRecovery.throughputWatchdog, {
enabled: true,
warmupMs: 0,
windowMs: 1_000,
minUsefulBytesPerSecond: 1,
minUsefulBytes: 1,
});
});