diff --git a/changelog.d/fixes/14637-proxy-attempt-log-rows.md b/changelog.d/fixes/14637-proxy-attempt-log-rows.md new file mode 100644 index 000000000000..951e80e9070e --- /dev/null +++ b/changelog.d/fixes/14637-proxy-attempt-log-rows.md @@ -0,0 +1 @@ +- **fix(sse):** the proxy journal keeps one row per request actually sent, so refused outlets stop looking like first-try successes — count requests with `COUNT(DISTINCT correlation_id)` ([#14637](https://github.com/diegosouzapw/OmniRoute/pull/14637)) — thanks @maxmad64bis diff --git a/open-sse/utils/upstreamStatusCapture.ts b/open-sse/utils/upstreamStatusCapture.ts index 6f336ff15de6..8e166929cf68 100644 --- a/open-sse/utils/upstreamStatusCapture.ts +++ b/open-sse/utils/upstreamStatusCapture.ts @@ -1,4 +1,44 @@ import { isProviderRequestCaptureActive, onDispatchStart } from "./providerRequestLogging.ts"; +import { sanitizeErrorMessage } from "./errorSanitization.ts"; + +/** One request actually sent: the outlet snapshot plus what came back. */ +export type AttemptRecord = { + proxy: unknown; + rotationAccount?: string | null; + upstreamStatus?: number; + error?: string | null; + startedAt?: number; + durationMs?: number | null; +}; + +export type AttemptSink = { + upstreamStatus?: number; + attempts?: AttemptRecord[]; +}; + +/** Marker for the in-flight record, so two stacked wrappers complete one row, not two. */ +const IN_FLIGHT = "omniroute.attempt.in-flight"; + +type FlightSink = AttemptSink & Record; + +function readFlight(sink: AttemptSink): AttemptRecord | undefined { + return (sink as FlightSink)[IN_FLIGHT]; +} + +function writeFlight(sink: AttemptSink, record: AttemptRecord | undefined): void { + if (record === undefined) delete (sink as FlightSink)[IN_FLIGHT]; + else (sink as FlightSink)[IN_FLIGHT] = record; +} + +function closeAttemptRecord( + record: AttemptRecord, + patch: { upstreamStatus?: number; error?: string | null } +): void { + if (patch.upstreamStatus === undefined) record.upstreamStatus = undefined; + else record.upstreamStatus = patch.upstreamStatus; + record.error = patch.error ?? null; + record.durationMs = Date.now() - (record.startedAt ?? Date.now()); +} /** * Wrap the process-wide fetch so the HTTP status the provider actually returned lands on @@ -18,7 +58,7 @@ import { isProviderRequestCaptureActive, onDispatchStart } from "./providerReque */ export function withUpstreamStatusCapture( inner: (...args: A) => Promise, - getSink: () => { upstreamStatus?: number } | undefined, + getSink: () => AttemptSink | undefined, isDispatching: () => boolean = isProviderRequestCaptureActive ): (...args: A) => Promise { if (isDispatching === isProviderRequestCaptureActive) { @@ -30,13 +70,39 @@ export function withUpstreamStatusCapture( return async (...args: A) => { const sink = isDispatching() ? getSink() : undefined; if (!sink) return inner(...args); + // One journal row per request actually sent: snapshot the outlet at entry, + // complete the row when the send settles. A stacked wrapper reuses the + // in-flight row instead of opening a second one. Sends outside a dispatch + // (side calls, background work) leave no trace, as before. + let record: AttemptRecord | undefined; + const inFlight = readFlight(sink); + if (inFlight) { + record = inFlight; + } else { + record = { + proxy: (sink as { proxy?: unknown }).proxy ?? null, + rotationAccount: (sink as { rotationAccount?: string | null }).rotationAccount ?? null, + startedAt: Date.now(), + }; + (sink.attempts ??= []).push(record); + writeFlight(sink, record); + } + const active = record === readFlight(sink); let response: Response; try { response = await inner(...args); } catch (error) { + if (isDispatching() && active) { + closeAttemptRecord(record, { error: sanitizeErrorMessage(error) || "Request failed" }); + writeFlight(sink, undefined); + } if (isDispatching()) sink.upstreamStatus = undefined; throw error; } + if (isDispatching() && active) { + closeAttemptRecord(record, { upstreamStatus: response.status, error: null }); + writeFlight(sink, undefined); + } if (isDispatching()) sink.upstreamStatus = response.status; return response; }; diff --git a/src/lib/db/migrationRunner.ts b/src/lib/db/migrationRunner.ts index 6d9582093e3f..3f4ec4df76f1 100644 --- a/src/lib/db/migrationRunner.ts +++ b/src/lib/db/migrationRunner.ts @@ -613,6 +613,11 @@ function isSchemaAlreadyApplied( // correlation_id at boot. Keyed by version only — a stale number here // would answer for another migration's schema and skip it. return hasColumn(db, "proxy_logs", "correlation_id"); + case "187": + // Same shape as 179/181/183/184: ensureProxyLogsColumns may have added + // attempt_number at boot. Keyed by version only — a stale number here + // would answer for another migration's schema and skip it. + return hasColumn(db, "proxy_logs", "attempt_number"); default: return false; } diff --git a/src/lib/db/migrations/187_proxy_logs_attempts.sql b/src/lib/db/migrations/187_proxy_logs_attempts.sql new file mode 100644 index 000000000000..eb48269aecf1 --- /dev/null +++ b/src/lib/db/migrations/187_proxy_logs_attempts.sql @@ -0,0 +1,7 @@ +-- attempt_number: 1-based position of the row within its request journal +-- (1..N for N sends sharing one correlation_id). NULL for legacy rows. +-- attempt_issue: outcome of the send within its journal — served for the send +-- the request settled on, abandoned for earlier sends. NULL for legacy rows. +-- No index: not a query dimension (count requests via correlation_id). +ALTER TABLE proxy_logs ADD COLUMN attempt_number INTEGER; +ALTER TABLE proxy_logs ADD COLUMN attempt_issue TEXT; diff --git a/src/lib/db/proxyLatency.ts b/src/lib/db/proxyLatency.ts index 818d2f094bbf..199e41957d95 100644 --- a/src/lib/db/proxyLatency.ts +++ b/src/lib/db/proxyLatency.ts @@ -13,7 +13,10 @@ type LatencyLogRow = { }; // Builds a `"host:port" -> avg_latency_ms` map from proxy_logs rows recorded -// within the trailing PROXY_LATENCY_WINDOW_HOURS window. +// within the trailing PROXY_LATENCY_WINDOW_HOURS window. The latency strategy +// measures the latency of sends served: a fast refusal does not make an +// outlet fast, so abandoned sends are excluded (legacy rows predate the +// journal and stay included via the NULL guard). function buildLatencyMap(db: ReturnType): Map { const sinceIso = new Date(Date.now() - PROXY_LATENCY_WINDOW_HOURS * 60 * 60 * 1000).toISOString(); @@ -22,6 +25,7 @@ function buildLatencyMap(db: ReturnType): Map= ? + AND (attempt_issue IS NULL OR attempt_issue <> 'abandoned') GROUP BY proxy_host, proxy_port` ) .all(sinceIso) as LatencyLogRow[]; diff --git a/src/lib/db/schemaColumns.ts b/src/lib/db/schemaColumns.ts index 5b95a9bcd1f8..165c8587576f 100644 --- a/src/lib/db/schemaColumns.ts +++ b/src/lib/db/schemaColumns.ts @@ -308,6 +308,14 @@ export function ensureProxyLogsColumns(db: SqliteDatabase) { db.exec("ALTER TABLE proxy_logs ADD COLUMN correlation_id TEXT"); console.log("[DB] Added proxy_logs.correlation_id column"); } + if (!columnNames.has("attempt_number")) { + db.exec("ALTER TABLE proxy_logs ADD COLUMN attempt_number INTEGER"); + console.log("[DB] Added proxy_logs.attempt_number column"); + } + if (!columnNames.has("attempt_issue")) { + db.exec("ALTER TABLE proxy_logs ADD COLUMN attempt_issue TEXT"); + console.log("[DB] Added proxy_logs.attempt_issue column"); + } } catch (error: unknown) { const message = error instanceof Error ? error.message : String(error); console.warn("[DB] Failed to verify proxy_logs schema:", message); diff --git a/src/lib/proxyLogger.ts b/src/lib/proxyLogger.ts index d3260030f6be..c57a07ffc670 100644 --- a/src/lib/proxyLogger.ts +++ b/src/lib/proxyLogger.ts @@ -72,6 +72,10 @@ interface ProxyLogEntry { tlsFingerprint: boolean; /** HTTP status the provider actually returned; null when no response was received. */ upstreamStatus: number | null; + /** 1-based position of this row within its request journal; null for unjournaled rows. */ + attemptNumber: number | null; + /** Outcome of this send within its request journal; null for unjournaled rows. */ + attemptIssue: "served" | "abandoned" | null; } type ProxyLogInput = Partial & { @@ -132,6 +136,11 @@ function loadFromDb() { correlationId: row.correlation_id || null, tlsFingerprint: row.tls_fingerprint === 1, upstreamStatus: typeof row.upstream_status === "number" ? row.upstream_status : null, + attemptNumber: typeof row.attempt_number === "number" ? row.attempt_number : null, + attemptIssue: + row.attempt_issue === "served" || row.attempt_issue === "abandoned" + ? row.attempt_issue + : null, }); } @@ -223,6 +232,14 @@ export function logProxyEvent(entry: ProxyLogInput) { correlationId: entry.correlationId || null, tlsFingerprint: entry.tlsFingerprint || false, upstreamStatus: entry.upstreamStatus ?? null, + attemptNumber: + typeof entry.attemptNumber === "number" && Number.isInteger(entry.attemptNumber) + ? entry.attemptNumber + : null, + attemptIssue: + entry.attemptIssue === "served" || entry.attemptIssue === "abandoned" + ? entry.attemptIssue + : null, }; // Structured egress line so the operator can confirm, in the proxy logs, which @@ -313,10 +330,12 @@ export function flushProxyLogsSync() { const insertStmt = db.prepare( `INSERT INTO proxy_logs (id, timestamp, status, proxy_type, proxy_host, proxy_port, proxy_name, level, level_id, provider, target_url, public_ip, egress_ip, latency_ms, error, - connection_id, combo_id, account, rotation_account, correlation_id, tls_fingerprint, upstream_status) + connection_id, combo_id, account, rotation_account, correlation_id, tls_fingerprint, upstream_status, + attempt_number, attempt_issue) VALUES (@id, @timestamp, @status, @proxyType, @proxyHost, @proxyPort, @proxyName, @level, @levelId, @provider, @targetUrl, @clientIp, @egressIp, @latencyMs, @error, - @connectionId, @comboId, @account, @rotationAccount, @correlationId, @tlsFingerprint, @upstreamStatus)` + @connectionId, @comboId, @account, @rotationAccount, @correlationId, @tlsFingerprint, @upstreamStatus, + @attemptNumber, @attemptIssue)` ); const transaction = db.transaction((entries: ProxyLogEntry[]) => { @@ -344,6 +363,8 @@ export function flushProxyLogsSync() { correlationId: item.correlationId, tlsFingerprint: item.tlsFingerprint ? 1 : 0, upstreamStatus: item.upstreamStatus, + attemptNumber: item.attemptNumber, + attemptIssue: item.attemptIssue, }); } }); diff --git a/src/sse/handlers/chatHelpers.ts b/src/sse/handlers/chatHelpers.ts index 61b94e3b76aa..acfebb98fd93 100644 --- a/src/sse/handlers/chatHelpers.ts +++ b/src/sse/handlers/chatHelpers.ts @@ -49,8 +49,9 @@ import { classify429FromError, type FailureKind } from "../../shared/utils/class import { resolveUseUpstream429BreakerHints } from "../../shared/utils/providerHints"; import { isFeatureFlagEnabled } from "../../shared/utils/featureFlags"; -import { logProxyEvent } from "../../lib/proxyLogger"; import { noteProxyOutcome } from "./proxyOutcomeMemory"; +import { logProxyJournal } from "./proxyJournal"; +import type { AttemptJournalEntry } from "./proxyJournal"; import { logTranslationEvent } from "../../lib/translatorEvents"; import { getRuntimeProviderProfile } from "@omniroute/open-sse/services/accountFallback.ts"; @@ -1061,11 +1062,14 @@ export function withUpstreamStatus( } /** Merge both things the applied-proxy sink captured: the executor proxy, then the status. */ +export type { AttemptJournalEntry } from "./proxyJournal"; export function mergeAppliedProxySink( proxyInfo: { proxy?: unknown; level?: string; levelId?: string | null } | null | undefined, - sink: { proxy: unknown; upstreamStatus?: number } + sink: { proxy: unknown; upstreamStatus?: number; attempts?: AttemptJournalEntry[] } ) { - return withUpstreamStatus(applyExecutorProxyToInfo(proxyInfo, sink.proxy), sink); + const merged = withUpstreamStatus(applyExecutorProxyToInfo(proxyInfo, sink.proxy), sink); + if (sink.attempts?.length) return { ...(merged || {}), attempts: sink.attempts }; + return merged; } // Async because the egress-IP lookup lazy-imports proxyEgress; callers treat @@ -1097,50 +1101,18 @@ export async function safeLogEvents({ } try { - const rawIp = - clientRawRequest?.headers?.["x-forwarded-for"] || - clientRawRequest?.headers?.["x-real-ip"] || - clientRawRequest?.headers?.["cf-connecting-ip"] || - null; - const rawIpValue = Array.isArray(rawIp) ? rawIp[0] : rawIp; - const clientIp = typeof rawIpValue === "string" ? rawIpValue.split(",")[0].trim() : null; - - // Resolve the egress IP (the IP the upstream actually saw) from cache — never - // blocking the request. Warm it in the background for next time. null until - // the first warm completes; direct (no proxy) is also tracked. - let egressIp: string | null = null; - try { - const { getCachedEgressIp, warmEgressIp } = await import("../../lib/proxyEgress"); - const { proxyConfigToUrl } = await import("@omniroute/open-sse/utils/proxyDispatcher.ts"); - const proxyUrl = proxyInfo?.proxy ? proxyConfigToUrl(proxyInfo.proxy) : null; - egressIp = getCachedEgressIp(proxyUrl); - warmEgressIp(proxyUrl); - } catch { - // egress visibility is best-effort; never break the request path - } - - logProxyEvent({ - status: result.success - ? "success" - : result.status === 408 || result.status === 504 - ? "timeout" - : "error", - proxy: proxyInfo?.proxy || null, - level: proxyInfo?.level || "direct", - levelId: proxyInfo?.levelId || null, + await logProxyJournal({ + result, + proxyInfo, + proxyLatency, provider, - targetUrl: `${provider}/${model}`, - clientIp, - egressIp, - latencyMs: proxyLatency, - error: result.success ? null : result.error || null, - connectionId: credentials.connectionId, - comboId: comboName || null, - account: credentials.connectionId?.slice(0, 8) || null, - rotationAccount: rotationAccount || null, - correlationId: correlationId || null, - tlsFingerprint: tlsFingerprintUsed, - upstreamStatus: proxyInfo?.upstreamStatus ?? null, + model, + credentials, + comboName, + clientRawRequest, + tlsFingerprintUsed, + rotationAccount, + correlationId, }); } catch {} diff --git a/src/sse/handlers/proxyJournal.ts b/src/sse/handlers/proxyJournal.ts new file mode 100644 index 000000000000..892fd8b4f0b7 --- /dev/null +++ b/src/sse/handlers/proxyJournal.ts @@ -0,0 +1,200 @@ +import { logProxyEvent } from "../../lib/proxyLogger"; + +/** One request actually sent: the outlet snapshot plus what came back. */ +export type AttemptJournalEntry = { + proxy: unknown; + rotationAccount?: string | null; + upstreamStatus?: number; + error?: string | null; + durationMs?: number | null; +}; + +export type ProxyJournalInput = { + result: { success: boolean; status?: number | null; error?: string | null }; + proxyInfo: { + proxy?: unknown; + level?: string; + levelId?: string | null; + upstreamStatus?: number; + attempts?: AttemptJournalEntry[]; + } | null; + proxyLatency: number; + provider: string; + model: string; + credentials: { connectionId?: string | null }; + comboName: string | null; + clientRawRequest: { + headers?: Record; + } | null; + tlsFingerprintUsed: boolean; + rotationAccount: string | null; + correlationId: string | null; +}; + +function attemptTextStatus(attempt: AttemptJournalEntry): string { + if (typeof attempt.upstreamStatus === "number") { + if (attempt.upstreamStatus === 408 || attempt.upstreamStatus === 504) return "timeout"; + return attempt.upstreamStatus < 400 ? "success" : "error"; + } + return attempt.error && /timed? ?out|504|408/i.test(attempt.error) ? "timeout" : "error"; +} + +type ProxyConfig = { type: string; host: string; port: number | string } | null; + +function asProxyConfig(value: unknown): ProxyConfig { + return (value as ProxyConfig) || null; +} + +// Resolve the egress IP (the IP the upstream actually saw) from cache — never +// blocking the request. Warm it in the background for next time. null until +// the first warm completes; direct (no proxy) is also tracked. +async function resolveCachedEgress(proxy: unknown): Promise { + try { + const { getCachedEgressIp, warmEgressIp } = await import("../../lib/proxyEgress"); + const { proxyConfigToUrl } = await import("@omniroute/open-sse/utils/proxyDispatcher.ts"); + const proxyUrl = proxy ? proxyConfigToUrl(proxy) : null; + const cached = getCachedEgressIp(proxyUrl); + warmEgressIp(proxyUrl); + return cached; + } catch { + // egress visibility is best-effort; never break the request path + return null; + } +} + +function extractClientIp(clientRawRequest: ProxyJournalInput["clientRawRequest"]): string | null { + const rawIp = + clientRawRequest?.headers?.["x-forwarded-for"] || + clientRawRequest?.headers?.["x-real-ip"] || + clientRawRequest?.headers?.["cf-connecting-ip"] || + null; + const rawIpValue = Array.isArray(rawIp) ? rawIp[0] : rawIp; + return typeof rawIpValue === "string" ? rawIpValue.split(",")[0].trim() : null; +} + +async function resolveEgressPerProxy(outlets: unknown[]): Promise> { + // One cache read per distinct outlet (a retry through the same proxy reads + // once): repeated sends usually share outlets, so the map stays tiny. + const egressIps = new Map(); + for (const outlet of outlets) { + if (!egressIps.has(outlet)) egressIps.set(outlet, await resolveCachedEgress(outlet)); + } + return egressIps; +} + +function abandonedRowLevel( + attempt: AttemptJournalEntry, + proxyInfo: ProxyJournalInput["proxyInfo"] +): string { + if (!attempt.proxy) return "direct"; + return proxyInfo?.level || "account"; +} + +function logAbandonedRow( + attempt: AttemptJournalEntry, + index: number, + shared: { + proxyInfo: ProxyJournalInput["proxyInfo"]; + provider: string; + model: string; + credentials: ProxyJournalInput["credentials"]; + comboName: string | null; + clientIp: string | null; + egressIps: Map; + rotationAccount: string | null; + correlationId: string | null; + } +): void { + const { proxyInfo, provider, model, credentials, comboName } = shared; + logProxyEvent({ + status: attemptTextStatus(attempt), + proxy: asProxyConfig(attempt.proxy), + level: abandonedRowLevel(attempt, proxyInfo), + levelId: proxyInfo?.levelId || null, + provider, + targetUrl: `${provider}/${model}`, + clientIp: shared.clientIp, + egressIp: shared.egressIps.get(attempt.proxy) ?? null, + latencyMs: attempt.durationMs ?? 0, + error: attempt.error || null, + connectionId: credentials.connectionId, + comboId: comboName || null, + account: credentials.connectionId?.slice(0, 8) || null, + rotationAccount: attempt.rotationAccount ?? shared.rotationAccount ?? null, + correlationId: shared.correlationId || null, + tlsFingerprint: false, + upstreamStatus: attempt.upstreamStatus ?? null, + attemptNumber: index + 1, + attemptIssue: "abandoned", + }); +} + +// One row per request actually sent: the final row is the send the request +// settled on, earlier rows are the refused sends sharing its correlation id. +export async function logProxyJournal(input: ProxyJournalInput): Promise { + const { + result, + proxyInfo, + proxyLatency, + provider, + model, + credentials, + comboName, + clientRawRequest, + tlsFingerprintUsed, + rotationAccount, + correlationId, + } = input; + const journal = proxyInfo?.attempts; + const attempts = Array.isArray(journal) && journal.length > 0 ? journal : null; + const journalRows = attempts ? attempts.slice(0, -1) : []; + const clientIp = extractClientIp(clientRawRequest); + + const egressIps = await resolveEgressPerProxy([ + ...journalRows.map((row) => row.proxy), + proxyInfo?.proxy ?? null, + ]); + const shared = { + proxyInfo, + provider, + model, + credentials, + comboName, + clientIp, + egressIps, + rotationAccount, + correlationId, + }; + + for (let index = 0; index < journalRows.length; index++) { + logAbandonedRow(journalRows[index], index, shared); + } + + logProxyEvent({ + status: result.success + ? "success" + : result.status === 408 || result.status === 504 + ? "timeout" + : "error", + proxy: asProxyConfig(proxyInfo?.proxy), + level: proxyInfo?.level || "direct", + levelId: proxyInfo?.levelId || null, + provider, + targetUrl: `${provider}/${model}`, + clientIp, + egressIp: egressIps.get(proxyInfo?.proxy) ?? null, + latencyMs: proxyLatency, + error: result.success ? null : result.error || null, + connectionId: credentials.connectionId, + comboId: comboName || null, + account: credentials.connectionId?.slice(0, 8) || null, + rotationAccount: rotationAccount || null, + correlationId: correlationId || null, + tlsFingerprint: tlsFingerprintUsed, + upstreamStatus: proxyInfo?.upstreamStatus ?? null, + attemptNumber: attempts ? attempts.length : null, + attemptIssue: attempts ? ("served" as const) : null, + }); + + // Abandoned sends are already written above, in attempt order; nothing follows. +} diff --git a/tests/unit/proxy-logs-attempt-rows.test.ts b/tests/unit/proxy-logs-attempt-rows.test.ts new file mode 100644 index 000000000000..19256b4c4e22 --- /dev/null +++ b/tests/unit/proxy-logs-attempt-rows.test.ts @@ -0,0 +1,365 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +// proxy_logs keeps one row per request actually sent: a request served on the +// 3rd attempt leaves 3 rows sharing one correlation id, the first two marked +// abandoned and the last one served. The served row is written last. A request +// with no journal keeps one row with null attempt columns. +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-proxy-attempt-rows-")); +process.env.DATA_DIR = TEST_DATA_DIR; + +const core = await import("../../src/lib/db/core.ts"); +const proxyLogger = await import("../../src/lib/proxyLogger.ts"); +const proxyLogsDb = await import("../../src/lib/db/proxyLogs.ts"); +const proxyLatency = await import("../../src/lib/db/proxyLatency.ts"); +const proxyEgress = await import("../../src/lib/proxyEgress.ts"); +const { proxyConfigToUrl } = await import("../../open-sse/utils/proxyDispatcher.ts"); +const { mergeAppliedProxySink, safeLogEvents } = + await import("../../src/sse/handlers/chatHelpers.ts"); +const { withUpstreamStatusCapture } = await import("../../open-sse/utils/upstreamStatusCapture.ts"); + +const PROXY_A = { type: "http", host: "127.0.0.1", port: 18081 }; +const PROXY_B = { type: "http", host: "127.0.0.1", port: 18082 }; + +type Sink = { proxy: unknown; upstreamStatus?: number; rotationAccount?: string | null }; + +function respond(status: number) { + return async () => new Response(null, { status }); +} + +function logArgs(proxyInfo: unknown, status: number, correlationId: string | null) { + return { + result: + status === 200 + ? { success: true, status, error: null } + : { success: false, status, error: "refused" }, + proxyInfo, + proxyLatency: 5, + provider: "openai", + model: "gpt-5", + sourceFormat: "openai", + targetFormat: "openai", + credentials: { connectionId: "conn-12345678" }, + comboName: null, + clientRawRequest: null, + correlationId, + }; +} + +test.beforeEach(() => { + proxyLogger.clearProxyLogs(); +}); + +test.after(() => { + proxyLogger.clearProxyLogs(); + core.closeDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); +}); + +test("three sends leave three rows on one correlation id: abandoned, abandoned, served", async () => { + const sink: Sink = { proxy: PROXY_A }; + const first = withUpstreamStatusCapture( + respond(429), + () => sink, + () => true + ); + const second = withUpstreamStatusCapture( + respond(403), + () => sink, + () => true + ); + const third = withUpstreamStatusCapture( + respond(200), + () => sink, + () => true + ); + await first(); + sink.proxy = PROXY_B; + await second(); + await third(); + + const proxyInfo = mergeAppliedProxySink({ proxy: null, level: "direct", levelId: null }, sink); + await safeLogEvents(logArgs(proxyInfo, 200, "corr-1")); + + const rows = proxyLogger.getProxyLogs(); + assert.equal(rows.length, 3); + for (const row of rows) assert.equal(row.correlationId, "corr-1"); + const byAttempt = [...rows].sort((a, b) => (a.attemptNumber ?? 0) - (b.attemptNumber ?? 0)); + assert.deepEqual( + byAttempt.map((r) => [r.attemptNumber, r.attemptIssue, r.upstreamStatus]), + [ + [1, "abandoned", 429], + [2, "abandoned", 403], + [3, "served", 200], + ] + ); +}); + +test("a fully refused request leaves abandoned rows and an error final row", async () => { + const sink: Sink = { proxy: PROXY_A }; + const wrapped = withUpstreamStatusCapture( + respond(503), + () => sink, + () => true + ); + await wrapped(); + await wrapped(); + + const proxyInfo = mergeAppliedProxySink({ proxy: null, level: "direct", levelId: null }, sink); + await safeLogEvents(logArgs(proxyInfo, 503, "corr-2")); + + const rows = proxyLogger.getProxyLogs(); + assert.equal(rows.length, 2); + const byAttempt = [...rows].sort((a, b) => (a.attemptNumber ?? 0) - (b.attemptNumber ?? 0)); + // The final send is the one the request settled on, even when it failed: it + // carries the terminal status, so it reads served while earlier sends read + // abandoned. + assert.deepEqual( + byAttempt.map((r) => [r.attemptNumber, r.attemptIssue]), + [ + [1, "abandoned"], + [2, "served"], + ] + ); + assert.equal(byAttempt[1].status, "error"); +}); + +test("a network throw keeps the send with a null status instead of dropping it", async () => { + const sink: Sink = { proxy: PROXY_A }; + const failing = withUpstreamStatusCapture( + async () => { + throw new TypeError("fetch failed"); + }, + () => sink, + () => true + ); + await assert.rejects(failing(), /fetch failed/); + const ok = withUpstreamStatusCapture( + respond(200), + () => sink, + () => true + ); + await ok(); + + const proxyInfo = mergeAppliedProxySink({ proxy: null, level: "direct", levelId: null }, sink); + await safeLogEvents(logArgs(proxyInfo, 200, "corr-3")); + + const rows = proxyLogger.getProxyLogs(); + assert.equal(rows.length, 2); + const byAttempt = [...rows].sort((a, b) => (a.attemptNumber ?? 0) - (b.attemptNumber ?? 0)); + assert.equal(byAttempt[0].upstreamStatus, null); + assert.equal(byAttempt[0].attemptIssue, "abandoned"); + assert.equal(byAttempt[1].attemptIssue, "served"); +}); + +test("a request with no journal keeps a single row with null attempt columns", async () => { + const proxyInfo = mergeAppliedProxySink( + { proxy: null, level: "direct", levelId: null }, + { proxy: null } + ); + await safeLogEvents(logArgs(proxyInfo, 200, null)); + + const [entry] = proxyLogger.getProxyLogs(); + assert.equal(entry.attemptNumber, null); + assert.equal(entry.attemptIssue, null); + assert.equal(entry.upstreamStatus, null); +}); + +test("a send outside a dispatch leaves no journal", async () => { + const sink: Sink = { proxy: PROXY_A }; + const wrapped = withUpstreamStatusCapture( + respond(200), + () => sink, + () => false + ); + await wrapped(); + + const proxyInfo = mergeAppliedProxySink({ proxy: null, level: "direct", levelId: null }, sink); + await safeLogEvents(logArgs(proxyInfo, 200, "corr-4")); + + const rows = proxyLogger.getProxyLogs(); + assert.equal(rows.length, 1); + assert.equal(rows[0].attemptNumber, null); +}); + +test("two wrappers on one send leave one row, not two", async () => { + const sink: Sink = { proxy: PROXY_A }; + const inner = withUpstreamStatusCapture( + respond(200), + () => sink, + () => true + ); + const outer = withUpstreamStatusCapture( + inner, + () => sink, + () => true + ); + await outer(); + + const proxyInfo = mergeAppliedProxySink({ proxy: null, level: "direct", levelId: null }, sink); + await safeLogEvents(logArgs(proxyInfo, 200, "corr-5")); + + const rows = proxyLogger.getProxyLogs(); + assert.equal(rows.length, 1); + assert.equal(rows[0].attemptNumber, 1); + assert.equal(rows[0].attemptIssue, "served"); +}); + +test("a retry inside one rotation turn leaves one row per send", async () => { + const sink: Sink = { proxy: PROXY_A }; + const wrapped = withUpstreamStatusCapture( + respond(429), + () => sink, + () => true + ); + await wrapped(); + await wrapped(); + + const proxyInfo = mergeAppliedProxySink({ proxy: null, level: "direct", levelId: null }, sink); + await safeLogEvents(logArgs(proxyInfo, 429, "corr-6")); + + const rows = proxyLogger.getProxyLogs(); + assert.equal(rows.length, 2); + const byAttempt = [...rows].sort((a, b) => (a.attemptNumber ?? 0) - (b.attemptNumber ?? 0)); + assert.deepEqual( + byAttempt.map((r) => [r.attemptNumber, r.attemptIssue, r.upstreamStatus]), + [ + [1, "abandoned", 429], + [2, "served", 429], + ] + ); +}); + +test("each send logs its own outlet egress address, not the final one", async () => { + // RED without the per-attempt resolution: both rows share the final + // proxy's cached egress address, so the first row reads E2. + const urlA = proxyConfigToUrl(PROXY_A); + const urlB = proxyConfigToUrl(PROXY_B); + proxyEgress._setEgressProbeForTests(async (proxyUrl) => + proxyUrl === urlA ? { ip: "203.0.113.11", latencyMs: 1 } : { ip: "203.0.113.22", latencyMs: 1 } + ); + try { + await proxyEgress.resolveEgressIp(urlA, { force: true }); + await proxyEgress.resolveEgressIp(urlB, { force: true }); + + const sink: Sink = { proxy: PROXY_A }; + await withUpstreamStatusCapture( + respond(429), + () => sink, + () => true + )(); + sink.proxy = PROXY_B; + await withUpstreamStatusCapture( + respond(200), + () => sink, + () => true + )(); + + const proxyInfo = mergeAppliedProxySink({ proxy: null, level: "direct", levelId: null }, sink); + await safeLogEvents(logArgs(proxyInfo, 200, "corr-egress")); + proxyLogger.flushProxyLogsSync(); + + const rows = proxyLogger.getProxyLogs(); + assert.equal(rows.length, 2); + const byAttempt = [...rows].sort((a, b) => (a.attemptNumber ?? 0) - (b.attemptNumber ?? 0)); + assert.equal(byAttempt[0].attemptIssue, "abandoned"); + assert.equal(byAttempt[0].egressIp, "203.0.113.11"); + assert.equal(byAttempt[1].attemptIssue, "served"); + assert.equal(byAttempt[1].egressIp, "203.0.113.22"); + proxyLogger.flushProxyLogsSync(); + const recent = proxyLogsDb.getRecentEgressIpForProxy("127.0.0.1", 18081); + assert.equal(recent?.egressIp, "203.0.113.11"); + } finally { + proxyEgress._setEgressProbeForTests(null); + proxyEgress.clearEgressCache(); + } +}); + +test("an abandoned send does not pull the latency average down", async () => { + proxyLogger.logProxyEvent({ + status: "error", + proxy: { type: "http", host: "127.0.0.1", port: 18091 }, + provider: "openai", + targetUrl: "openai/gpt-5", + latencyMs: 1, + attemptNumber: 1, + attemptIssue: "abandoned", + }); + proxyLogger.logProxyEvent({ + status: "success", + proxy: { type: "http", host: "127.0.0.1", port: 18091 }, + provider: "openai", + targetUrl: "openai/gpt-5", + latencyMs: 900, + attemptNumber: 2, + attemptIssue: "served", + }); + // A slower-but-served rival: with the fix 18091 averages 900 and loses to + // it; without the fix 18091 averages (1 + 900) / 2 and wrongly wins. + proxyLogger.logProxyEvent({ + status: "success", + proxy: { type: "http", host: "127.0.0.1", port: 18092 }, + provider: "openai", + targetUrl: "openai/gpt-5", + latencyMs: 500, + attemptNumber: 1, + attemptIssue: "served", + }); + proxyLogger.flushProxyLogsSync(); + + const db = core.getDbInstance(); + const row = db + .prepare( + `SELECT AVG(latency_ms) AS avg FROM proxy_logs + WHERE proxy_host = ? AND proxy_port = ?` + ) + .get("127.0.0.1", 18091) as { avg: number }; + // Raw average mixes the 1 ms refusal in; the latency map must read 900. + assert.ok(row.avg < 900); + const picked = proxyLatency.pickByLatency(db, [ + { host: "127.0.0.1", port: 18091 }, + { host: "127.0.0.1", port: 18092 }, + ]); + assert.deepEqual(picked, { host: "127.0.0.1", port: 18092 }); +}); + +test("a simulated send carries a non-negative duration", async () => { + const sink: Sink = { proxy: PROXY_A }; + await withUpstreamStatusCapture( + respond(200), + () => sink, + () => true + )(); + const attempts = (sink as { attempts?: Array<{ durationMs?: number | null }> }).attempts; + assert.ok(attempts?.length === 1); + assert.ok( + typeof attempts[0].durationMs === "number" && attempts[0].durationMs >= 0, + "durationMs must be a non-negative number" + ); +}); + +test("the served row is written after the abandoned rows", async () => { + const sink: Sink = { proxy: PROXY_A }; + await withUpstreamStatusCapture( + respond(429), + () => sink, + () => true + )(); + await withUpstreamStatusCapture( + respond(200), + () => sink, + () => true + )(); + + const proxyInfo = mergeAppliedProxySink({ proxy: null, level: "direct", levelId: null }, sink); + await safeLogEvents(logArgs(proxyInfo, 200, "corr-order")); + + const rows = proxyLogger.getProxyLogs(); + assert.equal(rows.length, 2); + // In-memory buffer is newest-first: index 0 is the last write, the served row. + assert.equal(rows[0].attemptIssue, "served"); + assert.equal(rows[1].attemptIssue, "abandoned"); +});