Skip to content

Commit 87be0d8

Browse files
committed
fix(codex): reclassify mid-stream reset as transient + escalating account cooldown + affinity diagnostics (#186)
1 parent 0d7fd98 commit 87be0d8

6 files changed

Lines changed: 179 additions & 16 deletions

File tree

devlog/_plan/260722_issue_bug_sweep/030_patch_s_sticky_502.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
- 위험도: 중간~높음 (요청 런타임 핵심 경로 — 오분류 시 건강한 계정 회피/성능 저하)
55
- 선행 조건: 없음. 단 구현 순서상 Diff 1(분류)이 Diff 2(escalation)보다 선행 —
66
escalation은 올바른 실패 신호 위에서만 의미가 있음.
7+
- **구현 완료 (2026-07-22)**: relay.ts:409 onTerminal 2-인자 + :509 catch→failed+502(clean-EOF incomplete 유지) + transportPhase/terminalSource 배선; responses.ts reportNativeTerminal override 전달; routing.ts consecutiveSuccesses escalation 30s/2m/10m/30m + level≥2 2연속 복구; request-log.ts 진단 필드 3종(affinity는 type-only — RequestLogContext 미배선, 문서화된 스코프 경계). `ResponsesTerminalStatus` 유니언 불변 확인. 검증: codex-routing 56 + request-log 36 + cancel 3 pass, `bun x tsc --noEmit` exit 0. 커밋: WP-impl-6.
78

89
## 핵심 제약 (리뷰어 blocker 반영)
910

src/codex/routing.ts

Lines changed: 29 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ export type CodexThreadResolution =
2525
const threadAccountMap = new Map<string, ThreadAffinityEntry>();
2626
type CodexUpstreamHealth = {
2727
consecutiveFailures: number;
28+
/** Consecutive healthy terminals observed while recovering from escalation level 2+. */
29+
consecutiveSuccesses?: number;
2830
lastFailureStatus?: number;
2931
lastFailureAt?: number;
3032
/** Hard cooldown (quota 429). Survives a later 2xx; blocks auth + selection. */
@@ -42,6 +44,12 @@ const CODEX_MAX_QUOTA_COOLDOWN_MS = 24 * 60 * 60_000;
4244
export const CODEX_FAILURE_WINDOW_MS = 5 * 60_000;
4345
/** How long a transient failure keeps the account out of pool selection. */
4446
export const CODEX_TRANSIENT_SOFT_AVOID_MS = 30_000;
47+
const CODEX_TRANSIENT_SOFT_AVOID_ESCALATION_MS = [
48+
CODEX_TRANSIENT_SOFT_AVOID_MS,
49+
2 * 60_000,
50+
10 * 60_000,
51+
30 * 60_000,
52+
] as const;
4553
export const CODEX_THREAD_AFFINITY_IDLE_TTL_MS = 24 * 60 * 60_000;
4654
export const CODEX_THREAD_AFFINITY_MAX_ENTRIES = 2048;
4755
// Min interval between quota threshold re-evaluations for a single bound thread.
@@ -449,8 +457,22 @@ export function recordCodexUpstreamOutcome(
449457
const now = meta.now ?? Date.now();
450458
const outcomeClass = classifyCodexUpstreamOutcome(outcome);
451459
if (outcomeClass === "success") {
452-
// Soft avoid clears on success; hard quota cooldown intentionally survives.
460+
const current = upstreamHealth.get(accountId);
453461
const cooldownUntil = getCodexAccountCooldownUntil(accountId, now);
462+
const failoverEnabled = (config.upstreamFailoverThreshold ?? 3) > 0;
463+
if (failoverEnabled && current && current.consecutiveFailures >= 2) {
464+
const consecutiveSuccesses = (current.consecutiveSuccesses ?? 0) + 1;
465+
if (consecutiveSuccesses < 2) {
466+
upstreamHealth.set(accountId, {
467+
...current,
468+
consecutiveSuccesses,
469+
...(cooldownUntil ? { cooldownUntil } : {}),
470+
});
471+
return;
472+
}
473+
}
474+
// Level 1 clears immediately; escalated accounts need two consecutive healthy terminals.
475+
// Hard quota cooldown intentionally survives either recovery path.
454476
if (cooldownUntil) upstreamHealth.set(accountId, { consecutiveFailures: 0, cooldownUntil });
455477
else upstreamHealth.delete(accountId);
456478
return;
@@ -491,14 +513,18 @@ export function recordCodexUpstreamOutcome(
491513
// Soft avoid + affinity clears are part of failover. When threshold is 0, leave
492514
// sticky sessions alone (same as shouldFailover / applyFailureFailover no-ops).
493515
const failoverEnabled = (config.upstreamFailoverThreshold ?? 3) > 0;
516+
const consecutiveFailures = stale ? 1 : (current?.consecutiveFailures ?? 0) + 1;
517+
const escalationMs = CODEX_TRANSIENT_SOFT_AVOID_ESCALATION_MS[
518+
Math.min(consecutiveFailures, CODEX_TRANSIENT_SOFT_AVOID_ESCALATION_MS.length) - 1
519+
]!;
494520
const softAvoidUntil = failoverEnabled
495521
? Math.max(
496522
getCodexAccountSoftAvoidUntil(accountId, now) ?? 0,
497-
now + CODEX_TRANSIENT_SOFT_AVOID_MS,
523+
now + escalationMs,
498524
)
499525
: undefined;
500526
upstreamHealth.set(accountId, {
501-
consecutiveFailures: stale ? 1 : (current?.consecutiveFailures ?? 0) + 1,
527+
consecutiveFailures,
502528
lastFailureStatus,
503529
lastFailureAt: now,
504530
...(hardCooldownUntil ? { cooldownUntil: hardCooldownUntil } : {}),

src/server/relay.ts

Lines changed: 31 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -406,7 +406,7 @@ export function relaySseWithHeartbeat(
406406
*/
407407
export function consumeForInspection(
408408
body: ReadableStream<Uint8Array>,
409-
onTerminal: (status: ResponsesTerminalStatus) => void,
409+
onTerminal: (status: ResponsesTerminalStatus, httpStatusOverride?: number) => void,
410410
signal?: AbortSignal,
411411
onDone?: () => void,
412412
logCtx?: RequestLogContext,
@@ -451,14 +451,24 @@ export function consumeForInspection(
451451
reportFirstOutput(payload);
452452
if (payload) {
453453
const status = terminalStatusFromSsePayload(payload);
454-
if (status) { reported = true; onTerminal(status); }
454+
if (status) {
455+
reported = true;
456+
if (logCtx) {
457+
logCtx.transportPhase = "terminal_sse";
458+
logCtx.terminalSource = "upstream";
459+
}
460+
onTerminal(status);
461+
}
455462
if (onCompletedResponse) {
456463
const response = completedResponseFromSsePayload(payload);
457464
if (response) onCompletedResponse(response);
458465
}
459466
}
460467
}
461-
if (!reported && !cancelled) onTerminal("incomplete");
468+
if (!reported && !cancelled) {
469+
if (logCtx) logCtx.terminalSource = "synthetic";
470+
onTerminal("incomplete");
471+
}
462472
return;
463473
}
464474
buffer += decoder.decode(value, { stream: true });
@@ -472,7 +482,14 @@ export function consumeForInspection(
472482
if (!payload) continue;
473483
if (!reported) {
474484
const status = terminalStatusFromSsePayload(payload);
475-
if (status) { reported = true; onTerminal(status); }
485+
if (status) {
486+
reported = true;
487+
if (logCtx) {
488+
logCtx.transportPhase = "terminal_sse";
489+
logCtx.terminalSource = "upstream";
490+
}
491+
onTerminal(status);
492+
}
476493
}
477494
if (onCompletedResponse) {
478495
const response = completedResponseFromSsePayload(payload);
@@ -481,7 +498,16 @@ export function consumeForInspection(
481498
}
482499
}
483500
} catch {
484-
if (!reported && !cancelled) onTerminal("incomplete");
501+
// Upstream read failure after HTTP 200 (mid-stream socket reset) is not a
502+
// protocol `response.incomplete` terminal. Report a synthetic 502 so account
503+
// health treats it as transient; abort-driven client cancellation still wins.
504+
if (!reported && !cancelled) {
505+
if (logCtx) {
506+
logCtx.transportPhase = "mid_stream";
507+
logCtx.terminalSource = "synthetic";
508+
}
509+
onTerminal("failed", 502);
510+
}
485511
} finally {
486512
onDone?.();
487513
}

src/server/request-log.ts

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,9 @@ export interface RequestLogContext {
6262
upstreamError?: string;
6363
/** HTTP status derived from a terminal `response.failed` SSE payload (429/401/503/etc.). */
6464
terminalHttpStatus?: number;
65+
affinity?: "reused" | "new_bind" | "rebound" | "cleared";
66+
transportPhase?: "pre_headers" | "mid_stream" | "terminal_sse";
67+
terminalSource?: "upstream" | "synthetic";
6568
}
6669

6770
export interface RequestLogEntry {
@@ -92,6 +95,12 @@ export interface RequestLogEntry {
9295
usage?: OcxUsage;
9396
totalTokens?: number;
9497
attempts?: PersistedUsageAttempt[];
98+
/** Codex pool affinity decision for this request (diagnostics for #186). */
99+
affinity?: "reused" | "new_bind" | "rebound" | "cleared";
100+
/** Where the upstream terminal/failure was observed. */
101+
transportPhase?: "pre_headers" | "mid_stream" | "terminal_sse";
102+
/** Whether the terminal came from a real upstream SSE event or a proxy synthetic tail. */
103+
terminalSource?: "upstream" | "synthetic";
95104
}
96105

97106
const requestLog: RequestLogEntry[] = [];
@@ -587,6 +596,9 @@ export function addFinalRequestLog(
587596
...(loggedUsage ? { usage: loggedUsage } : {}),
588597
...(totalTokens !== undefined ? { totalTokens } : {}),
589598
...(attempts?.length ? { attempts } : {}),
599+
...(logCtx.affinity ? { affinity: logCtx.affinity } : {}),
600+
...(logCtx.transportPhase ? { transportPhase: logCtx.transportPhase } : {}),
601+
...(logCtx.terminalSource ? { terminalSource: logCtx.terminalSource } : {}),
590602
});
591603
if (isUsageDebugEnabled()) {
592604
appendUsageDebug({

src/server/responses.ts

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ import type { WsData } from "./ws-bridge";
6969
import { registerTurn, trackStreamLifetime, unregisterTurn } from "./lifecycle";
7070
import { redactSecretString } from "../lib/redact";
7171
import { readBoundedResponseBody } from "../lib/bounded-body";
72+
import { supportedLadderFor } from "./effort-policy";
7273
import {
7374
beginRequestAttempt,
7475
catalogModelSupportsServiceTier,
@@ -622,10 +623,12 @@ async function handleComboResponses(
622623
model: pick.target.model,
623624
provider: pick.target.provider,
624625
};
626+
const targetRoute = routeModel(config, `${pick.target.provider}/${pick.target.model}`);
625627
const childBody = concreteComboRequestBody(
626628
rawBody,
627629
pick.target,
628630
comboDefaultEffort(config, comboId),
631+
supportedLadderFor({ provider: targetRoute.provider, modelId: targetRoute.modelId }),
629632
);
630633
const childHeaders = new Headers(req.headers);
631634
childHeaders.delete("content-length");
@@ -1212,8 +1215,8 @@ export async function handleResponses(
12121215
// even if the client has already disconnected: the turn genuinely reached that terminal, so
12131216
// it must log as completed/failed, not be dropped or downgraded to a cancel (#44). A pure
12141217
// client-cancel (no terminal seen) is finalized separately via consumeForInspection's onCancel.
1215-
const reportNativeTerminal = (status: ResponsesTerminalStatus) => {
1216-
terminalRecorder?.(status);
1218+
const reportNativeTerminal = (status: ResponsesTerminalStatus, httpStatusOverride?: number) => {
1219+
terminalRecorder?.(status, httpStatusOverride);
12171220
options.onNativePassthroughTerminal?.(status);
12181221
};
12191222
consumeForInspection(

tests/codex-routing.test.ts

Lines changed: 101 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ import {
3636
import { CODEX_UNKNOWN_USAGE_SCORE } from "../src/codex/quota";
3737
import { MAIN_CODEX_ACCOUNT_ID } from "../src/codex/main-account";
3838
import { routeModel } from "../src/router";
39+
import { consumeForInspection } from "../src/server/relay";
3940
import type { OcxConfig } from "../src/types";
4041

4142
const TEST_DIR = join(import.meta.dir, ".tmp-codex-routing-test");
@@ -65,6 +66,12 @@ function saveTestCredential(id: string): void {
6566
});
6667
}
6768

69+
const inspectionTick = () => new Promise(resolve => setTimeout(resolve, 5));
70+
71+
function pendingInspectionStream(): ReadableStream<Uint8Array> {
72+
return new ReadableStream<Uint8Array>({ start() {}, pull() {} });
73+
}
74+
6875
describe("codex routing", () => {
6976
beforeEach(() => {
7077
previousOpencodexHome = process.env.OPENCODEX_HOME;
@@ -339,10 +346,9 @@ describe("codex routing", () => {
339346
recordCodexUpstreamOutcome(config, "a", 200, { now: now + 1 });
340347
recordCodexUpstreamOutcome(config, "a", 503, { now: now + 2 });
341348
recordCodexUpstreamOutcome(config, "a", 503, { now: now + 3 });
342-
// 2 failures after a success (streak=2) is under the failover threshold (3),
343-
// but soft-avoid still blocks for 30s. After it expires, "a" is selectable.
344-
// Soft-avoid was last set at now+3, so it expires at now+3+30s.
345-
const afterSoftAvoid = now + 3 + CODEX_TRANSIENT_SOFT_AVOID_MS + 1;
349+
// The success reset the old streak, so the next two failures form escalation
350+
// level 2 (still below failover threshold 3) and avoid the account for 2m.
351+
const afterSoftAvoid = now + 3 + 2 * 60_000 + 1;
346352
expect(resolveCodexAccountForThread("next", config, afterSoftAvoid)).toBe("a");
347353
});
348354

@@ -356,6 +362,95 @@ describe("codex routing", () => {
356362
expect(resolveCodexAccountForThread("next", config)).toBe("a");
357363
});
358364

365+
test("inspection client cancellation records no terminal outcome or account penalty", async () => {
366+
const config = makeConfig();
367+
const record = (status: "completed" | "failed" | "incomplete", override?: number) => {
368+
recordCodexUpstreamOutcome(config, "a", status === "failed" ? (override ?? 502) : 200);
369+
};
370+
371+
const preAborted = new AbortController();
372+
preAborted.abort();
373+
consumeForInspection(pendingInspectionStream(), record, preAborted.signal);
374+
expect(getCodexUpstreamHealth("a")).toBeNull();
375+
376+
const midDrain = new AbortController();
377+
consumeForInspection(pendingInspectionStream(), record, midDrain.signal);
378+
midDrain.abort();
379+
await inspectionTick();
380+
expect(getCodexUpstreamHealth("a")).toBeNull();
381+
});
382+
383+
test("inspection read rejection reports failed plus synthetic 502 and clears affinity", async () => {
384+
const config = makeConfig();
385+
const now = 1_800_000_000_000;
386+
updateAccountQuota("a", 10);
387+
updateAccountQuota("b", 20);
388+
expect(resolveCodexAccountForThread("reset-thread", config, now)).toBe("a");
389+
const terminals: Array<[string, number | undefined]> = [];
390+
const resetStream = new ReadableStream<Uint8Array>({
391+
start(controller) {
392+
controller.error(new Error("socket reset"));
393+
},
394+
});
395+
396+
consumeForInspection(resetStream, (status, override) => {
397+
terminals.push([status, override]);
398+
recordCodexUpstreamOutcome(config, "a", override ?? 200, { now: now + 1, threadId: "reset-thread" });
399+
});
400+
await inspectionTick();
401+
402+
expect(terminals).toEqual([["failed", 502]]);
403+
expect(getCodexUpstreamHealth("a")).toMatchObject({ consecutiveFailures: 1, lastFailureStatus: 502 });
404+
expect(resolveCodexAccountForThread("reset-thread", config, now + 2)).toBe("b");
405+
});
406+
407+
test("inspection clean EOF remains incomplete and success-like", async () => {
408+
const config = makeConfig();
409+
const terminals: Array<[string, number | undefined]> = [];
410+
const cleanEof = new ReadableStream<Uint8Array>({ start(controller) { controller.close(); } });
411+
412+
consumeForInspection(cleanEof, (status, override) => {
413+
terminals.push([status, override]);
414+
recordCodexUpstreamOutcome(config, "a", status === "failed" ? (override ?? 502) : 200);
415+
});
416+
await inspectionTick();
417+
418+
expect(terminals).toEqual([["incomplete", undefined]]);
419+
expect(getCodexUpstreamHealth("a")).toBeNull();
420+
});
421+
422+
test("transient cooldown escalates to 2m, 10m, then the 30m cap", () => {
423+
const config = makeConfig();
424+
const now = 1_800_000_000_000;
425+
426+
recordCodexUpstreamOutcome(config, "a", 503, { now });
427+
recordCodexUpstreamOutcome(config, "a", 503, { now: now + 1 });
428+
expect(getCodexAccountSoftAvoidUntil("a", now + 1)).toBe(now + 1 + 2 * 60_000);
429+
recordCodexUpstreamOutcome(config, "a", 503, { now: now + 2 });
430+
expect(getCodexAccountSoftAvoidUntil("a", now + 2)).toBe(now + 2 + 10 * 60_000);
431+
recordCodexUpstreamOutcome(config, "a", 503, { now: now + 3 });
432+
expect(getCodexAccountSoftAvoidUntil("a", now + 3)).toBe(now + 3 + 30 * 60_000);
433+
recordCodexUpstreamOutcome(config, "a", 503, { now: now + 4 });
434+
expect(getCodexAccountSoftAvoidUntil("a", now + 4)).toBe(now + 4 + 30 * 60_000);
435+
});
436+
437+
test("escalation level 2 requires two consecutive healthy terminals to clear", () => {
438+
const config = makeConfig();
439+
const now = 1_800_000_000_000;
440+
recordCodexUpstreamOutcome(config, "a", 503, { now });
441+
recordCodexUpstreamOutcome(config, "a", 503, { now: now + 1 });
442+
443+
recordCodexUpstreamOutcome(config, "a", 200, { now: now + 2 });
444+
expect(getCodexUpstreamHealth("a")).toMatchObject({
445+
consecutiveFailures: 2,
446+
consecutiveSuccesses: 1,
447+
});
448+
expect(isCodexAccountSoftAvoided("a", now + 2)).toBe(true);
449+
450+
recordCodexUpstreamOutcome(config, "a", 200, { now: now + 3 });
451+
expect(getCodexUpstreamHealth("a")).toBeNull();
452+
});
453+
359454
test("stale thread affinity is revalidated before reuse", () => {
360455
const config = makeConfig();
361456
updateAccountQuota("a", 10);
@@ -690,9 +785,9 @@ describe("codex routing", () => {
690785
const firstAvoid = getCodexAccountSoftAvoidUntil("a", now);
691786
expect(firstAvoid).toBe(now + CODEX_TRANSIENT_SOFT_AVOID_MS);
692787

693-
// A second failure 10s later extends the window from that point.
788+
// A second failure 10s later escalates the window to 2m from that point.
694789
recordCodexUpstreamOutcome(config, "a", "timeout", { now: now + 10_000 });
695-
expect(getCodexAccountSoftAvoidUntil("a", now + 10_001)).toBe(now + 10_000 + CODEX_TRANSIENT_SOFT_AVOID_MS);
790+
expect(getCodexAccountSoftAvoidUntil("a", now + 10_001)).toBe(now + 10_000 + 2 * 60_000);
696791
});
697792

698793
test("soft-avoid is not applied when failover threshold is 0", () => {

0 commit comments

Comments
 (0)