fix(loop): auto-retry acp-loop replay on transient upstream rejections (#189)

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. Set `BILI_REPLAY_RETRY_MAX=1` to restore the previous fail-fast behavior.
+
 ## [0.1.40] — 2026-08-13
 
 ### Features

CONFIGURATION.md

@@ -342,3 +342,5 @@ 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. |
+| `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. |

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 } 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}/${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)` : "";
+                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<string, unknown>;
+            current = await result.response.json() as Record<string, unknown>;
         } finally {
-            clearTimer();
+            result.clearTimer();
         }
     }
     ctx.log(`[acp-proxy: responses JSON compress loop limit (${MAX_LOOP_ROUNDS}) reached]`);

src/fetch-util.ts

@@ -69,3 +69,117 @@ 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;
+
+/** 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 {
+    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<void> {
+    if (ms <= 0 || signal?.aborted) return Promise.resolve();
+    return new Promise((resolve) => {
+        let timer: ReturnType<typeof setTimeout> | 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;
+    maxAttempts: 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 }> {
+    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 >= maxAttempts;
+        if (!lastAttempt && isTransientUpstreamError(result.response.status, errText)) {
+            const delayMs = replayBackoffMs(attempt);
+            onRetry?.({ attempt, status: result.response.status, detail: errText, delayMs, maxAttempts });
+            await sleep(delayMs, externalSignal);
+            continue;
+        }
+        throw new UpstreamHttpError(result.response.status, errText, attempt);
+    }
+}

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 } 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}/${info.maxAttempts})]`);
+                        loggerLog("warn", `[acp-loop] upstream rejected replay (HTTP ${info.status}); retrying in ${info.delayMs}ms (attempt ${info.attempt}/${info.maxAttempts})`);
+                    },
+                );
+            } 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<Uint8Array>;
+            currentUpstream = respResult.response.body as ReadableStream<Uint8Array>;
             if (activeClearTimer) activeClearTimer();
-            activeClearTimer = clearTimer;
+            activeClearTimer = respResult.clearTimer;
         }
     } finally {
         if (activeClearTimer) {

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<typeof createCore>;
@@ -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<string>((_, 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;
     }
 });