From 7fae454934cee5a2ad6f9f075b387893f15573b4 Mon Sep 17 00:00:00 2001 From: ranxianglei Date: Sat, 22 Aug 2026 17:15:14 +0800 Subject: [PATCH 1/2] fix(loop): auto-retry acp-loop replay on transient upstream rejections (#189) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After a compress, provider risk-control may briefly reject the replay request (GLM Coding Plan: 400 {"code":3007,"msg":"captcha verify failed"} ~1s after the context rewrite), and the error was passed straight into the agent session. The replay request now retries transient upstream failures with exponential backoff (3 attempts total, 1500ms base, BILI_REPLAY_RETRY_BASE_MS override) on both the streaming loop and the Responses-API JSON loop. Transient = 429/5xx or 4xx bodies matching risk-control markers; plain 4xx still fail fast. Each retry logs a clear "likely provider risk-control — retrying" line; exhausted failures are surfaced with an "after N attempt(s)" suffix. --- CHANGELOG.md | 4 + CONFIGURATION.md | 1 + src/compress-loop-responses.ts | 22 ++++-- src/fetch-util.ts | 104 +++++++++++++++++++++++++ src/loop/core.ts | 51 ++++++++----- tests/loop-compress.test.ts | 10 ++- tests/loop-core.test.ts | 134 +++++++++++++++++++++++++++++++++ 7 files changed, 297 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a46196c..d6ca17e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ Versions follow the merge of a `*_release-v*` branch; CI publishes to npm on tag - **Customizable compression prompts** (#156): the `compress` block now accepts `prompts` — an override object for the compression prompt text (`compressPhilosophy`, `howToCompressRules`, `tier2DistillRules`, `tier3CondenseRules`) — merged sub-field-wise across the three config levels (global → provider → model) and applied consistently to the system prompt, the nudge text, and the compress loop. Because the kernel's default rules are load-bearing (tuned over months of production use), overrides are **inert until `acknowledgePromptsRisk: true`** is set at the winning level; without it they are ignored and a one-time warning is logged. Non-string fields are silently dropped. Mainly useful for non-English or small-model prompt tuning. +### Fixes + +- **acp-loop replay auto-retry on upstream risk-control rejections** (#189): after a `compress`, the acp-loop replay request can be rejected by provider risk-control — GLM Coding Plan returns `400 {"code":3007,"msg":"captcha verify failed"}` ~1s after the big context rewrite — and the error was passed straight into the agent session as `[acp-proxy: compress loop upstream error 400: ...]`. The replay request (both the streaming loop and the Responses-API JSON loop) now retries transient upstream failures with exponential backoff: up to 3 attempts total, base delay 1500ms doubling per attempt, overridable via `BILI_REPLAY_RETRY_BASE_MS` (ms; `0` disables the delay). Transient = HTTP 429/5xx, or any other 4xx whose body matches risk-control markers (`captcha`, `verify failed`, `risk control`, `风控`, `rate limit`, `too many requests`, `try again`); plain 4xx (bad model, bad params) still fail fast with no retry. Each retry logs a clear line (`upstream rejected replay (HTTP 400 ...); likely provider risk-control — retrying in 1500ms (attempt 1/3)`), and if all attempts fail the surfaced error now says `after 3 attempt(s)` so users can tell it was retried. + ## [0.1.40] — 2026-08-13 ### Features diff --git a/CONFIGURATION.md b/CONFIGURATION.md index 2bca50f..58239f6 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -342,3 +342,4 @@ Environment variables take precedence over the config file. They are useful for | `ACP_LOG` | Set to `0` to disable request logging. | | `ACP_AUTO_UPDATE` | Set to `0` to disable auto-update checks. | | `ACP_PROVIDERS` | Path to an external `providers.json` (legacy / shared file). | +| `BILI_REPLAY_RETRY_BASE_MS` | Base backoff delay (ms) for acp-loop replay retries after a transient upstream rejection (default `1500`; set `0` to disable the delay). See #189. | diff --git a/src/compress-loop-responses.ts b/src/compress-loop-responses.ts index 674744a..744f08c 100644 --- a/src/compress-loop-responses.ts +++ b/src/compress-loop-responses.ts @@ -12,7 +12,7 @@ import { applyRanges } from "./stream.js"; import { resolveDecompress } from "./decompress-shared.js"; import { buildVisibilityMarker } from "./compress-loop.js"; import { MAX_LOOP_ROUNDS } from "./loop/index.js"; -import { fetchWithTimeout } from "./fetch-util.js"; +import { fetchWithRetry, UpstreamHttpError, REPLAY_MAX_ATTEMPTS } from "./fetch-util.js"; import { proxyDispatcher } from "./upstream-proxy.js"; /** Extract triggers from assistant text. @@ -211,20 +211,26 @@ export async function compressLoopResponsesJson( inputItems.push({ type: "message", role: "developer", content: buildVisibilityMarker(call.name, result) }); } requestBody.input = inputItems; - const { response, clearTimer } = await fetchWithTimeout(requestOptions.url, { + const result = await fetchWithRetry(requestOptions.url, { method: "POST", headers: requestOptions.headers, body: JSON.stringify(requestBody), ...(ctx.proxyUrl ? { dispatcher: proxyDispatcher(ctx.proxyUrl) } : {}), + }, undefined, undefined, (info) => { + ctx.log(`[acp-proxy: responses upstream rejected replay (HTTP ${info.status}: ${info.detail.slice(0, 120)}); likely provider risk-control — retrying in ${info.delayMs}ms (attempt ${info.attempt}/${REPLAY_MAX_ATTEMPTS})]`); + loggerLog("warn", `[acp-compress-responses] upstream rejected replay (HTTP ${info.status}); retrying in ${info.delayMs}ms (attempt ${info.attempt}/${REPLAY_MAX_ATTEMPTS})`); + }).catch((e) => { + if (e instanceof UpstreamHttpError) { + const suffix = e.attempts > 1 ? ` after ${e.attempts} attempt(s)` : ""; + ctx.log(`[acp-proxy: responses compress loop upstream error ${e.status}${suffix}: ${e.body.slice(0, 200)}]`); + loggerLog("error", `[acp-compress-responses] upstream error ${e.status}${suffix}: ${e.body.slice(0, 200)}`); + } + throw e; }); try { - if (!response.ok) { - const detail = await response.text().catch(() => "upstream error"); - throw new Error(`responses compress loop upstream error ${response.status}: ${detail.slice(0, 200)}`); - } - current = await response.json() as Record; + current = await result.response.json() as Record; } finally { - clearTimer(); + result.clearTimer(); } } ctx.log(`[acp-proxy: responses JSON compress loop limit (${MAX_LOOP_ROUNDS}) reached]`); diff --git a/src/fetch-util.ts b/src/fetch-util.ts index 3ca778b..0c8ab98 100644 --- a/src/fetch-util.ts +++ b/src/fetch-util.ts @@ -69,3 +69,107 @@ export async function fetchWithTimeout( throw e; } } + +/** Upstream HTTP failure after all retry attempts are exhausted (or a + * non-transient error that fails fast). `attempts` is the number of requests + * actually made; `body` is the upstream error body (already read). */ +export class UpstreamHttpError extends Error { + readonly status: number; + readonly body: string; + readonly attempts: number; + constructor(status: number, body: string, attempts: number) { + super(`upstream error ${status}`); + this.name = "UpstreamHttpError"; + this.status = status; + this.body = body; + this.attempts = attempts; + } +} + +/** Body markers indicating an upstream 4xx is a transient risk-control / + * rate-limit rejection rather than a genuine client error. GLM Coding Plan + * returns 400 {"code":3007,"msg":"captcha verify failed"} ~1s after large + * context rewrites (issue #189); every observed case recovered on retry, + * so such bodies are retried while plain 4xx (bad model, bad params) fail fast. */ +const TRANSIENT_BODY_MARKERS = [ + "captcha", + "verify failed", + "risk control", + "风控", + "rate limit", + "too many requests", + "try again", +]; + +export function isTransientUpstreamError(status: number, body: string): boolean { + if (status === 429 || status >= 500) return true; + if (status < 400) return false; + const lower = body.toLowerCase(); + return TRANSIENT_BODY_MARKERS.some((marker) => lower.includes(marker)); +} + +/** Total requests per replay attempt (initial + retries). */ +export const REPLAY_MAX_ATTEMPTS = 3; + +/** Base backoff delay in ms; overridable via BILI_REPLAY_RETRY_BASE_MS + * (0 disables the delay). Read on each call so tests can tune it live. */ +export function replayBaseDelayMs(): number { + const raw = Number(process.env.BILI_REPLAY_RETRY_BASE_MS); + return Number.isFinite(raw) && raw >= 0 ? raw : 1500; +} + +/** Exponential backoff for the given 1-based attempt: base * 2^(attempt-1). */ +export function replayBackoffMs(attempt: number): number { + return replayBaseDelayMs() * 2 ** (attempt - 1); +} + +/** Abortable sleep: resolves early if `signal` fires (downstream disconnect). + * ms <= 0 resolves immediately. */ +export function sleep(ms: number, signal?: AbortSignal): Promise { + if (ms <= 0 || signal?.aborted) return Promise.resolve(); + return new Promise((resolve) => { + let timer: ReturnType | null = null; + const finish = () => { + if (timer) clearTimeout(timer); + if (signal) signal.removeEventListener("abort", finish); + resolve(); + }; + timer = setTimeout(finish, ms); + if (signal) signal.addEventListener("abort", finish, { once: true }); + }); +} + +export interface ReplayRetryInfo { + attempt: number; + status: number; + detail: string; + delayMs: number; +} + +/** fetchWithTimeout with bounded retry on transient upstream HTTP failures. + * For acp-loop replay requests, where provider risk-control may briefly + * reject a request whose context was just rewritten (#189). Network-level + * failures (timeout, connection reset) propagate unchanged — NOT retried + * here, to avoid stacking the 10-min timeout across attempts. */ +export async function fetchWithRetry( + url: string, + opts: FetchOptions, + timeoutMs: number | undefined, + externalSignal: AbortSignal | undefined, + onRetry?: (info: ReplayRetryInfo) => void, +): Promise<{ response: Response; clearTimer: () => void }> { + for (let attempt = 1; ; attempt++) { + const result = await fetchWithTimeout(url, opts, timeoutMs, externalSignal); + if (result.response.ok) return result; + const errText = await result.response.text().catch(() => "upstream error"); + result.clearTimer(); + const lastAttempt = attempt >= REPLAY_MAX_ATTEMPTS; + if (!lastAttempt && isTransientUpstreamError(result.response.status, errText)) { + const delayMs = replayBackoffMs(attempt); + onRetry?.({ attempt, status: result.response.status, detail: errText, delayMs }); + await sleep(delayMs, externalSignal); + continue; + } + throw new UpstreamHttpError(result.response.status, errText, attempt); + } +} diff --git a/src/loop/core.ts b/src/loop/core.ts index a8e873b..bb39ae1 100644 --- a/src/loop/core.ts +++ b/src/loop/core.ts @@ -17,7 +17,7 @@ import { import { applyRanges } from "../stream.js"; import { resolveDecompress } from "../decompress-shared.js"; import { buildVisibilityMarker } from "../compress-loop.js"; -import { fetchWithTimeout } from "../fetch-util.js"; +import { fetchWithRetry, UpstreamHttpError, REPLAY_MAX_ATTEMPTS } from "../fetch-util.js"; import { proxyDispatcher } from "../upstream-proxy.js"; import { log as loggerLog } from "../logger.js"; import type { WireProtocol } from "../util.js"; @@ -407,30 +407,41 @@ export async function* runCompressLoop( fs.writeFileSync(`${dumpDir}/req-${Date.now()}-${sid}-REREQUEST.json`, JSON.stringify(newBody, null, 2)); } catch { /* best-effort */ } } - const { response: resp, clearTimer } = await fetchWithTimeout( - requestOptions.url, - { - method: "POST", - headers: requestOptions.headers, - body: JSON.stringify(newBody), - ...(ctx.proxyUrl ? { dispatcher: proxyDispatcher(ctx.proxyUrl) } : {}), - }, - undefined, - signal, - ); + let respResult: { response: Response; clearTimer: () => void }; + try { + respResult = await fetchWithRetry( + requestOptions.url, + { + method: "POST", + headers: requestOptions.headers, + body: JSON.stringify(newBody), + ...(ctx.proxyUrl ? { dispatcher: proxyDispatcher(ctx.proxyUrl) } : {}), + }, + undefined, + signal, + (info) => { + ctx.log(`[acp-proxy: upstream rejected replay (HTTP ${info.status}: ${info.detail.slice(0, 120)}); likely provider risk-control — retrying in ${info.delayMs}ms (attempt ${info.attempt}/${REPLAY_MAX_ATTEMPTS})]`); + loggerLog("warn", `[acp-loop] upstream rejected replay (HTTP ${info.status}); retrying in ${info.delayMs}ms (attempt ${info.attempt}/${REPLAY_MAX_ATTEMPTS})`); + }, + ); + } catch (e) { + if (!(e instanceof UpstreamHttpError)) throw e; + const suffix = e.attempts > 1 ? ` after ${e.attempts} attempt(s)` : ""; + ctx.log(`[acp-proxy: compress loop upstream error ${e.status}${suffix}: ${e.body.slice(0, 200)}]`); + loggerLog("error", `[acp-loop] upstream error ${e.status}${suffix}: ${e.body.slice(0, 200)}`); + yield adapter.emitError(`upstream error ${e.status}${suffix}: ${e.body.slice(0, 200)}`); + return; + } - if (!resp.ok || !resp.body) { - clearTimer(); - const errText = await resp.text().catch(() => "upstream error"); - ctx.log(`[acp-proxy: compress loop upstream error ${resp.status}: ${errText.slice(0, 200)}]`); - loggerLog("error", `[acp-loop] upstream error ${resp.status}: ${errText.slice(0, 200)}`); - yield adapter.emitError(`upstream error ${resp.status}: ${errText.slice(0, 200)}`); + if (!respResult.response.body) { + respResult.clearTimer(); + yield adapter.emitError(`upstream error ${respResult.response.status}: empty response body`); return; } - currentUpstream = resp.body as ReadableStream; + currentUpstream = respResult.response.body as ReadableStream; if (activeClearTimer) activeClearTimer(); - activeClearTimer = clearTimer; + activeClearTimer = respResult.clearTimer; } } finally { if (activeClearTimer) { diff --git a/tests/loop-compress.test.ts b/tests/loop-compress.test.ts index 3dd8aa5..12bc835 100644 --- a/tests/loop-compress.test.ts +++ b/tests/loop-compress.test.ts @@ -5,6 +5,7 @@ import { createCore, createInitialState, assignRefs, emptyRefMap, defaultConfig import type { Session } from "../src/session.ts"; import { runCompressLoop, createResponsesAdapter } from "../src/loop/index.ts"; import { buildCompressSystemPrompt } from "../src/compress-tool.ts"; +import { REPLAY_MAX_ATTEMPTS } from "../src/fetch-util.ts"; function makeCtx(messages: CoreMessage[] = []): { core: ReturnType; @@ -394,6 +395,7 @@ test("loop #9 (S2): responses round yields usage → session.stats populated (nu }); test("loop #10 (S3): upstream 500 mid-loop terminates cleanly (timer cleared, no hang)", async () => { + process.env.BILI_REPLAY_RETRY_BASE_MS = "1"; const ctx = makeCtx([ textMsg("m00001", "user", "hello"), textMsg("m00002", "assistant", "hi"), @@ -404,8 +406,12 @@ test("loop #10 (S3): upstream 500 mid-loop terminates cleanly (timer cleared, no fcEvents(0, "call_c", "compress", compressArgs), COMPLETED, ].join(""); + let fetchCalls = 0; const orig = globalThis.fetch; - globalThis.fetch = (async () => new Response("upstream error", { status: 500 })) as typeof fetch; + globalThis.fetch = (async () => { + fetchCalls++; + return new Response("upstream error", { status: 500 }); + }) as typeof fetch; try { const out = await Promise.race([ drain( @@ -417,7 +423,9 @@ test("loop #10 (S3): upstream 500 mid-loop terminates cleanly (timer cleared, no new Promise((_, reject) => setTimeout(() => reject(new Error("loop hung (timer not cleared)")), 3000)), ]); assert.ok(typeof out === "string", "loop terminated cleanly on upstream 500 (S3: timer cleared)"); + assert.equal(fetchCalls, REPLAY_MAX_ATTEMPTS, "5xx retried with bounded attempts (#189)"); } finally { + delete process.env.BILI_REPLAY_RETRY_BASE_MS; globalThis.fetch = orig; } }); diff --git a/tests/loop-core.test.ts b/tests/loop-core.test.ts index a0801a6..dd35bd2 100644 --- a/tests/loop-core.test.ts +++ b/tests/loop-core.test.ts @@ -6,6 +6,7 @@ import type { Session } from "../src/session.ts"; import { runCompressLoop, createResponsesAdapter } from "../src/loop/index.ts"; import { buildCompressSystemPrompt } from "../src/compress-tool.ts"; import type { WireProtocol } from "../src/util.ts"; +import { isTransientUpstreamError, REPLAY_MAX_ATTEMPTS, replayBackoffMs } from "../src/fetch-util.ts"; function makeCtx(messages: CoreMessage[] = [], protocol?: WireProtocol): { core: ReturnType; @@ -238,3 +239,136 @@ test("loop usage: protocol unset → legacy additive behavior (prompt + cached)" assert.equal(ctx.session.stats.inputTokens, 1900, "no protocol → prompt + cached (1000 + 900)"); assert.equal(ctx.session.stats.cachedTokens, 900); }); + +// Regression guard for #189: after a compress, the acp-loop replay request can +// be rejected by provider risk-control (GLM Coding Plan returns 400 +// {"code":3007,"msg":"captcha verify failed"} ~1s after a big context rewrite). +// The replay must auto-retry with backoff instead of surfacing the error into +// the agent session. +const CAPTCHA_400_BODY = '{"code":3007,"msg":"captcha verify failed"}'; + +function compressRound(): string { + return [ + sse("response.created", { response: { id: "resp_1", status: "in_progress" } }), + fcEvents(0, "call_c", "compress", JSON.stringify({ content: [{ startId: "m00001", endId: "m00002", summary: "s" }] })), + COMPLETED, + ].join(""); +} + +test("replay retry: transient 400 (captcha) then success → retried, no error surfaced", async () => { + process.env.BILI_REPLAY_RETRY_BASE_MS = "1"; + let fetchCalls = 0; + const orig = globalThis.fetch; + globalThis.fetch = (async () => { + fetchCalls++; + if (fetchCalls === 1) return new Response(CAPTCHA_400_BODY, { status: 400 }); + return new Response(COMPLETED, { status: 200 }); + }) as typeof fetch; + try { + const out = await drain( + new Response(compressRound(), { status: 200 }).body!, + makeCtx(), + { model: "gpt-4o", input: [], stream: true }, + { url: "http://mock", headers: {} }, + ); + assert.equal(fetchCalls, 2, "replay retried once after transient 400"); + assert.ok(!out.includes("upstream error"), "no upstream error surfaced to client"); + assert.ok(/response\.completed/.test(out), "graceful completion after retry"); + } finally { + delete process.env.BILI_REPLAY_RETRY_BASE_MS; + globalThis.fetch = orig; + } +}); + +test("replay retry: persistent captcha 400 → bounded retries, error names attempt count", async () => { + process.env.BILI_REPLAY_RETRY_BASE_MS = "1"; + let fetchCalls = 0; + const orig = globalThis.fetch; + globalThis.fetch = (async () => { + fetchCalls++; + return new Response(CAPTCHA_400_BODY, { status: 400 }); + }) as typeof fetch; + try { + const out = await drain( + new Response(compressRound(), { status: 200 }).body!, + makeCtx(), + { model: "gpt-4o", input: [], stream: true }, + { url: "http://mock", headers: {} }, + ); + assert.equal(fetchCalls, REPLAY_MAX_ATTEMPTS, "retries are bounded"); + assert.ok(out.includes("upstream error 400"), "error surfaced to client"); + assert.ok(out.includes(`after ${REPLAY_MAX_ATTEMPTS} attempt(s)`), "attempt count in error message"); + } finally { + delete process.env.BILI_REPLAY_RETRY_BASE_MS; + globalThis.fetch = orig; + } +}); + +test("replay retry: fatal 400 (invalid model) → NO retry, fail fast", async () => { + let fetchCalls = 0; + const orig = globalThis.fetch; + globalThis.fetch = (async () => { + fetchCalls++; + return new Response('{"error":{"message":"Invalid model"}}', { status: 400 }); + }) as typeof fetch; + try { + const out = await drain( + new Response(compressRound(), { status: 200 }).body!, + makeCtx(), + { model: "gpt-4o", input: [], stream: true }, + { url: "http://mock", headers: {} }, + ); + assert.equal(fetchCalls, 1, "non-transient 4xx is not retried"); + assert.ok(out.includes("upstream error 400"), "error surfaced to client"); + assert.ok(!out.includes("attempt(s)"), "no attempt-count suffix on single-attempt failure"); + } finally { + globalThis.fetch = orig; + } +}); + +test("replay retry: 429 then success → retried", async () => { + process.env.BILI_REPLAY_RETRY_BASE_MS = "1"; + let fetchCalls = 0; + const orig = globalThis.fetch; + globalThis.fetch = (async () => { + fetchCalls++; + if (fetchCalls === 1) return new Response('{"error":"rate limited"}', { status: 429 }); + return new Response(COMPLETED, { status: 200 }); + }) as typeof fetch; + try { + const out = await drain( + new Response(compressRound(), { status: 200 }).body!, + makeCtx(), + { model: "gpt-4o", input: [], stream: true }, + { url: "http://mock", headers: {} }, + ); + assert.equal(fetchCalls, 2, "429 retried"); + assert.ok(!out.includes("upstream error"), "no upstream error surfaced to client"); + } finally { + delete process.env.BILI_REPLAY_RETRY_BASE_MS; + globalThis.fetch = orig; + } +}); + +test("isTransientUpstreamError: classifier matrix", () => { + assert.equal(isTransientUpstreamError(400, CAPTCHA_400_BODY), true, "captcha 400 is transient"); + assert.equal(isTransientUpstreamError(400, '{"error":{"message":"Invalid model"}}'), false, "plain 400 is not"); + assert.equal(isTransientUpstreamError(401, ""), false, "401 never retried"); + assert.equal(isTransientUpstreamError(429, ""), true, "429 always retried"); + assert.equal(isTransientUpstreamError(500, ""), true, "5xx always retried"); + assert.equal(isTransientUpstreamError(503, "service unavailable"), true); + assert.equal(isTransientUpstreamError(200, "captcha"), false, "2xx never classified"); + assert.equal(REPLAY_MAX_ATTEMPTS, 3); +}); + +test("replayBackoffMs: exponential from env-tunable base", () => { + process.env.BILI_REPLAY_RETRY_BASE_MS = "100"; + try { + assert.equal(replayBackoffMs(1), 100); + assert.equal(replayBackoffMs(2), 200); + assert.equal(replayBackoffMs(3), 400); + } finally { + delete process.env.BILI_REPLAY_RETRY_BASE_MS; + } + assert.equal(replayBackoffMs(1), 1500, "default base is 1500ms"); +}); From 080c4c6e1567a865175620d8b6411ff01baff87c Mon Sep 17 00:00:00 2001 From: ranxianglei Date: Sat, 22 Aug 2026 17:32:32 +0800 Subject: [PATCH 2/2] fix(loop): add BILI_REPLAY_RETRY_MAX opt-out for replay retry policy (#189) --- CHANGELOG.md | 2 +- CONFIGURATION.md | 1 + src/compress-loop-responses.ts | 6 +++--- src/fetch-util.ts | 14 ++++++++++-- src/loop/core.ts | 6 +++--- tests/loop-core.test.ts | 39 +++++++++++++++++++++++++++++++++- 6 files changed, 58 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d6ca17e..e87fffc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,7 @@ Versions follow the merge of a `*_release-v*` branch; CI publishes to npm on tag ### Fixes -- **acp-loop replay auto-retry on upstream risk-control rejections** (#189): after a `compress`, the acp-loop replay request can be rejected by provider risk-control — GLM Coding Plan returns `400 {"code":3007,"msg":"captcha verify failed"}` ~1s after the big context rewrite — and the error was passed straight into the agent session as `[acp-proxy: compress loop upstream error 400: ...]`. The replay request (both the streaming loop and the Responses-API JSON loop) now retries transient upstream failures with exponential backoff: up to 3 attempts total, base delay 1500ms doubling per attempt, overridable via `BILI_REPLAY_RETRY_BASE_MS` (ms; `0` disables the delay). Transient = HTTP 429/5xx, or any other 4xx whose body matches risk-control markers (`captcha`, `verify failed`, `risk control`, `风控`, `rate limit`, `too many requests`, `try again`); plain 4xx (bad model, bad params) still fail fast with no retry. Each retry logs a clear line (`upstream rejected replay (HTTP 400 ...); likely provider risk-control — retrying in 1500ms (attempt 1/3)`), and if all attempts fail the surfaced error now says `after 3 attempt(s)` so users can tell it was retried. +- **acp-loop replay auto-retry on upstream risk-control rejections** (#189): after a `compress`, the acp-loop replay request can be rejected by provider risk-control — GLM Coding Plan returns `400 {"code":3007,"msg":"captcha verify failed"}` ~1s after the big context rewrite — and the error was passed straight into the agent session as `[acp-proxy: compress loop upstream error 400: ...]`. The replay request (both the streaming loop and the Responses-API JSON loop) now retries transient upstream failures with exponential backoff: up to 3 attempts total, base delay 1500ms doubling per attempt, overridable via `BILI_REPLAY_RETRY_BASE_MS` (ms; `0` disables the delay). Transient = HTTP 429/5xx, or any other 4xx whose body matches risk-control markers (`captcha`, `verify failed`, `risk control`, `风控`, `rate limit`, `too many requests`, `try again`); plain 4xx (bad model, bad params) still fail fast with no retry. Each retry logs a clear line (`upstream rejected replay (HTTP 400 ...); likely provider risk-control — retrying in 1500ms (attempt 1/3)`), and if all attempts fail the surfaced error now says `after 3 attempt(s)` so users can tell it was retried. Set `BILI_REPLAY_RETRY_MAX=1` to restore the previous fail-fast behavior. ## [0.1.40] — 2026-08-13 diff --git a/CONFIGURATION.md b/CONFIGURATION.md index 58239f6..0aa16e6 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -343,3 +343,4 @@ Environment variables take precedence over the config file. They are useful for | `ACP_AUTO_UPDATE` | Set to `0` to disable auto-update checks. | | `ACP_PROVIDERS` | Path to an external `providers.json` (legacy / shared file). | | `BILI_REPLAY_RETRY_BASE_MS` | Base backoff delay (ms) for acp-loop replay retries after a transient upstream rejection (default `1500`; set `0` to disable the delay). See #189. | +| `BILI_REPLAY_RETRY_MAX` | Total attempts for acp-loop replay retries (default `3`; set `1` to disable retries entirely — legacy fail-fast behavior). See #189. | diff --git a/src/compress-loop-responses.ts b/src/compress-loop-responses.ts index 744f08c..da4bb19 100644 --- a/src/compress-loop-responses.ts +++ b/src/compress-loop-responses.ts @@ -12,7 +12,7 @@ import { applyRanges } from "./stream.js"; import { resolveDecompress } from "./decompress-shared.js"; import { buildVisibilityMarker } from "./compress-loop.js"; import { MAX_LOOP_ROUNDS } from "./loop/index.js"; -import { fetchWithRetry, UpstreamHttpError, REPLAY_MAX_ATTEMPTS } from "./fetch-util.js"; +import { fetchWithRetry, UpstreamHttpError } from "./fetch-util.js"; import { proxyDispatcher } from "./upstream-proxy.js"; /** Extract triggers from assistant text. @@ -217,8 +217,8 @@ export async function compressLoopResponsesJson( body: JSON.stringify(requestBody), ...(ctx.proxyUrl ? { dispatcher: proxyDispatcher(ctx.proxyUrl) } : {}), }, undefined, undefined, (info) => { - ctx.log(`[acp-proxy: responses upstream rejected replay (HTTP ${info.status}: ${info.detail.slice(0, 120)}); likely provider risk-control — retrying in ${info.delayMs}ms (attempt ${info.attempt}/${REPLAY_MAX_ATTEMPTS})]`); - loggerLog("warn", `[acp-compress-responses] upstream rejected replay (HTTP ${info.status}); retrying in ${info.delayMs}ms (attempt ${info.attempt}/${REPLAY_MAX_ATTEMPTS})`); + ctx.log(`[acp-proxy: responses upstream rejected replay (HTTP ${info.status}: ${info.detail.slice(0, 120)}); likely provider risk-control — retrying in ${info.delayMs}ms (attempt ${info.attempt}/${info.maxAttempts})]`); + loggerLog("warn", `[acp-compress-responses] upstream rejected replay (HTTP ${info.status}); retrying in ${info.delayMs}ms (attempt ${info.attempt}/${info.maxAttempts})`); }).catch((e) => { if (e instanceof UpstreamHttpError) { const suffix = e.attempts > 1 ? ` after ${e.attempts} attempt(s)` : ""; diff --git a/src/fetch-util.ts b/src/fetch-util.ts index 0c8ab98..02941be 100644 --- a/src/fetch-util.ts +++ b/src/fetch-util.ts @@ -111,6 +111,14 @@ export function isTransientUpstreamError(status: number, body: string): boolean /** Total requests per replay attempt (initial + retries). */ export const REPLAY_MAX_ATTEMPTS = 3; +/** Total requests per replay attempt; overridable via BILI_REPLAY_RETRY_MAX + * (1 = legacy fail-fast behavior, no retry). Read on each call so tests can + * tune it live. */ +export function replayMaxAttempts(): number { + const raw = Number(process.env.BILI_REPLAY_RETRY_MAX); + return Number.isInteger(raw) && raw >= 1 ? raw : REPLAY_MAX_ATTEMPTS; +} + /** Base backoff delay in ms; overridable via BILI_REPLAY_RETRY_BASE_MS * (0 disables the delay). Read on each call so tests can tune it live. */ export function replayBaseDelayMs(): number { @@ -144,6 +152,7 @@ export interface ReplayRetryInfo { status: number; detail: string; delayMs: number; + maxAttempts: number; } /** fetchWithTimeout with bounded retry on transient upstream HTTP failures. @@ -158,15 +167,16 @@ export async function fetchWithRetry( externalSignal: AbortSignal | undefined, onRetry?: (info: ReplayRetryInfo) => void, ): Promise<{ response: Response; clearTimer: () => void }> { + const maxAttempts = replayMaxAttempts(); for (let attempt = 1; ; attempt++) { const result = await fetchWithTimeout(url, opts, timeoutMs, externalSignal); if (result.response.ok) return result; const errText = await result.response.text().catch(() => "upstream error"); result.clearTimer(); - const lastAttempt = attempt >= REPLAY_MAX_ATTEMPTS; + const lastAttempt = attempt >= maxAttempts; if (!lastAttempt && isTransientUpstreamError(result.response.status, errText)) { const delayMs = replayBackoffMs(attempt); - onRetry?.({ attempt, status: result.response.status, detail: errText, delayMs }); + onRetry?.({ attempt, status: result.response.status, detail: errText, delayMs, maxAttempts }); await sleep(delayMs, externalSignal); continue; } diff --git a/src/loop/core.ts b/src/loop/core.ts index bb39ae1..048f219 100644 --- a/src/loop/core.ts +++ b/src/loop/core.ts @@ -17,7 +17,7 @@ import { import { applyRanges } from "../stream.js"; import { resolveDecompress } from "../decompress-shared.js"; import { buildVisibilityMarker } from "../compress-loop.js"; -import { fetchWithRetry, UpstreamHttpError, REPLAY_MAX_ATTEMPTS } from "../fetch-util.js"; +import { fetchWithRetry, UpstreamHttpError } from "../fetch-util.js"; import { proxyDispatcher } from "../upstream-proxy.js"; import { log as loggerLog } from "../logger.js"; import type { WireProtocol } from "../util.js"; @@ -420,8 +420,8 @@ export async function* runCompressLoop( undefined, signal, (info) => { - ctx.log(`[acp-proxy: upstream rejected replay (HTTP ${info.status}: ${info.detail.slice(0, 120)}); likely provider risk-control — retrying in ${info.delayMs}ms (attempt ${info.attempt}/${REPLAY_MAX_ATTEMPTS})]`); - loggerLog("warn", `[acp-loop] upstream rejected replay (HTTP ${info.status}); retrying in ${info.delayMs}ms (attempt ${info.attempt}/${REPLAY_MAX_ATTEMPTS})`); + ctx.log(`[acp-proxy: upstream rejected replay (HTTP ${info.status}: ${info.detail.slice(0, 120)}); likely provider risk-control — retrying in ${info.delayMs}ms (attempt ${info.attempt}/${info.maxAttempts})]`); + loggerLog("warn", `[acp-loop] upstream rejected replay (HTTP ${info.status}); retrying in ${info.delayMs}ms (attempt ${info.attempt}/${info.maxAttempts})`); }, ); } catch (e) { diff --git a/tests/loop-core.test.ts b/tests/loop-core.test.ts index dd35bd2..7f4a7f3 100644 --- a/tests/loop-core.test.ts +++ b/tests/loop-core.test.ts @@ -6,7 +6,7 @@ import type { Session } from "../src/session.ts"; import { runCompressLoop, createResponsesAdapter } from "../src/loop/index.ts"; import { buildCompressSystemPrompt } from "../src/compress-tool.ts"; import type { WireProtocol } from "../src/util.ts"; -import { isTransientUpstreamError, REPLAY_MAX_ATTEMPTS, replayBackoffMs } from "../src/fetch-util.ts"; +import { isTransientUpstreamError, REPLAY_MAX_ATTEMPTS, replayBackoffMs, replayMaxAttempts } from "../src/fetch-util.ts"; function makeCtx(messages: CoreMessage[] = [], protocol?: WireProtocol): { core: ReturnType; @@ -372,3 +372,40 @@ test("replayBackoffMs: exponential from env-tunable base", () => { } assert.equal(replayBackoffMs(1), 1500, "default base is 1500ms"); }); + +test("replayMaxAttempts: env-tunable total attempts (1 = legacy no-retry)", () => { + for (const [value, expected] of [["1", 1], ["5", 5], ["0", REPLAY_MAX_ATTEMPTS], ["abc", REPLAY_MAX_ATTEMPTS], ["-2", REPLAY_MAX_ATTEMPTS]] as const) { + if (value === "abc") delete process.env.BILI_REPLAY_RETRY_MAX; + else process.env.BILI_REPLAY_RETRY_MAX = value; + try { + assert.equal(replayMaxAttempts(), expected, `BILI_REPLAY_RETRY_MAX=${value}`); + } finally { + delete process.env.BILI_REPLAY_RETRY_MAX; + } + } + assert.equal(replayMaxAttempts(), REPLAY_MAX_ATTEMPTS, "default is 3"); +}); + +test("replay retry: BILI_REPLAY_RETRY_MAX=1 → legacy fail-fast (no retry)", async () => { + process.env.BILI_REPLAY_RETRY_MAX = "1"; + let fetchCalls = 0; + const orig = globalThis.fetch; + globalThis.fetch = (async () => { + fetchCalls++; + return new Response(CAPTCHA_400_BODY, { status: 400 }); + }) as typeof fetch; + try { + const out = await drain( + new Response(compressRound(), { status: 200 }).body!, + makeCtx(), + { model: "gpt-4o", input: [], stream: true }, + { url: "http://mock", headers: {} }, + ); + assert.equal(fetchCalls, 1, "MAX=1 disables retries (legacy behavior)"); + assert.ok(out.includes("upstream error 400"), "error surfaced to client"); + assert.ok(!out.includes("attempt(s)"), "no attempt-count suffix on single attempt"); + } finally { + delete process.env.BILI_REPLAY_RETRY_MAX; + globalThis.fetch = orig; + } +});