diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java index 3aacd086..ba5ecd35 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java @@ -8,6 +8,8 @@ import org.springframework.ai.chat.messages.UserMessage; import org.springframework.ai.chat.model.ChatModel; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.chat.prompt.Prompt; +import org.springframework.http.HttpHeaders; +import org.springframework.web.client.RestClientResponseException; import org.springframework.web.reactive.function.client.WebClientResponseException; import vip.mate.channel.web.ChatStreamTracker; import vip.mate.llm.chatmodel.AssistantThinkingRelay; @@ -15,6 +17,10 @@ import vip.mate.llm.chatmodel.ReasoningContentCache; import reactor.core.Disposable; +import java.time.Instant; +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeParseException; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -193,6 +199,16 @@ public class NodeStreamingChatHelper { else healthTracker.recordFailure(primaryProviderId); } + /** + * Record a primary failure carrying a provider-stated retry window so the + * health tracker can start a cooldown of exactly that length. No-op under + * the same conditions as {@link #recordPrimary}. + */ + private void recordPrimaryFailure(long cooldownOverrideMs) { + if (healthTracker == null || primaryProviderId == null) return; + healthTracker.recordFailure(primaryProviderId, cooldownOverrideMs); + } + /** * Map an {@link ErrorType} to the matching pool * {@link vip.mate.llm.failover.AvailableProviderPool.RemovalSource} for @@ -207,7 +223,11 @@ public class NodeStreamingChatHelper { * {@link vip.mate.llm.failover.ProviderHealthTracker}'s cooldown instead.

*/ private static vip.mate.llm.failover.AvailableProviderPool.RemovalSource hardRemovalSource(ErrorType type) { - if (type == null) return null; + // Policy lives on the enum ({@code evictsProvider}); this switch is + // only the name mapping to the pool's RemovalSource. A type marked + // evicting but missing here falls through to null (fail-open, logged + // nowhere) — extend the switch when adding a new evicting type. + if (type == null || !type.evictsProvider()) return null; return switch (type) { case AUTH_ERROR -> vip.mate.llm.failover.AvailableProviderPool.RemovalSource.AUTH_ERROR; case BILLING -> vip.mate.llm.failover.AvailableProviderPool.RemovalSource.BILLING; @@ -224,11 +244,7 @@ public class NodeStreamingChatHelper { * says nothing about whether the provider's other models still work. */ private static boolean isProviderLevelFailure(ErrorType type) { - if (type == null) return false; - return switch (type) { - case NONE, PROMPT_TOO_LONG, CLIENT_ERROR, THINKING_BLOCK_ERROR, MODEL_NOT_FOUND -> false; - default -> true; - }; + return type != null && type.countsHealth(); } /** Convenience: pool-aware membership check. Null pool means fail-open (everyone in). */ @@ -373,6 +389,21 @@ public class NodeStreamingChatHelper { // avoid masking truly fatal errors. MAX_TOTAL_DURATION_MS is the // ultimate safety net. static final int MAX_RETRIES_UNKNOWN = 5; + // OVERLOADED: the provider's serving capacity is saturated (Anthropic 529, + // "engine_overloaded", "model is overloaded"). Unlike RATE_LIMIT this says + // nothing about the caller's key, so waiting on the same provider is the + // productive move — recovery periods are typically tens of seconds, hence + // the dedicated long backoff table below instead of the generic 3s-based + // exponential. + static final int MAX_RETRIES_OVERLOADED = 5; + /** + * Backoff table for {@link ErrorType#OVERLOADED} retries, indexed by + * {@code attempt - 1} (attempts past the table reuse the last entry). + * A ±30% jitter is applied on top so concurrent conversations don't + * re-hit a saturated provider in lockstep. The 3-minute wall-clock + * budget still bounds the total wait. + */ + static final long[] OVERLOADED_BACKOFF_MS = {10_000, 20_000, 40_000, 60_000, 60_000}; // Hard time budget for the primary retry loop (3 min). Prevents // retries from stalling a single conversation turn indefinitely. // Aligned with WikiProcessingService.llmMaxTotalDurationMs. @@ -455,9 +486,24 @@ public class NodeStreamingChatHelper { || msg.contains("authentication") || msg.contains("AuthenticationError")) { return ErrorType.AUTH_ERROR; } + // Overloaded — the provider's serving capacity is saturated. Checked + // BEFORE the rate-limit patterns: providers commonly surface overload + // through a reused 429 status ("engine_overloaded" arrives alongside + // "429" in the same chain), and the more specific semantic must win — + // an overloaded provider deserves patient same-provider backoff, not + // the rate-limit fast-failover path. + if (msg.contains("engine_overloaded") + || msg.contains("overloaded_error") // Anthropic 529 body type + || msg.contains("Overloaded") // Anthropic 529 message + || msg.contains("model is overloaded") // Gemini / OpenAI-compatible + || msg.contains("529") + || msg.contains("server is busy") + || msg.contains("当前分组上游负载已饱和")) { // SiliconFlow group saturation + return ErrorType.OVERLOADED; + } // Rate limit if (msg.contains("429") || msg.contains("rate_limit") || msg.contains("RateLimitError") - || msg.contains("Too Many Requests") || msg.contains("engine_overloaded")) { + || msg.contains("Too Many Requests")) { return ErrorType.RATE_LIMIT; } // Thinking block errors (Anthropic: old thinking blocks cannot be modified) @@ -549,8 +595,7 @@ public class NodeStreamingChatHelper { // surfaced as HTTP 400 with a body that describes the upstream // outage. These are transient server-side failures — retryable. || msg.contains("temporarily unavailable") - || msg.contains("service unavailable") - || msg.contains("model is overloaded")) { + || msg.contains("service unavailable")) { return ErrorType.SERVER_ERROR; } // Client errors (400 Bad Request — unsupported format, invalid params, etc.) — NOT retryable. @@ -568,6 +613,111 @@ public class NodeStreamingChatHelper { return ErrorType.UNKNOWN; } + /** + * Ceiling for honoring a provider-stated retry window as an in-loop + * backoff sleep. Longer windows (quota resets measured in minutes or + * hours) are not worth blocking a conversation turn for — the call fails + * over instead, and the window is honored as a + * {@code ProviderHealthTracker} cooldown override so later turns skip + * the provider without re-probing it. + */ + static final long HINTED_BACKOFF_CAP_MS = 90_000; + + /** Floor / ceiling for any parsed retry-window hint (guards absurd values). */ + private static final long MIN_HINT_MS = 1_000; + private static final long MAX_HINT_MS = 2 * 60 * 60 * 1000L; + + private static final List ANTHROPIC_RESET_HEADERS = List.of( + "anthropic-ratelimit-requests-reset", + "anthropic-ratelimit-tokens-reset", + "anthropic-ratelimit-input-tokens-reset", + "anthropic-ratelimit-output-tokens-reset"); + + private static final List OPENAI_RESET_HEADERS = List.of( + "x-ratelimit-reset-requests", + "x-ratelimit-reset-tokens"); + + /** Matches Go-style duration strings ("1s", "6m0s", "120ms", "1h2m"). */ + private static final java.util.regex.Pattern GO_DURATION = java.util.regex.Pattern.compile( + "^(?:(\\d+)h)?(?:(\\d+)m)?(?:(\\d+(?:\\.\\d+)?)s)?(?:(\\d+)ms)?$"); + + /** + * Walk the error chain for an HTTP response exception and parse the + * provider-stated retry window from its headers. Returns milliseconds + * clamped to {@code [MIN_HINT_MS, MAX_HINT_MS]}, or {@code 0} when no + * usable hint is present. + * + *

Priority: {@code Retry-After} (delta-seconds or HTTP-date) → + * Anthropic RFC-3339 reset instants → OpenAI-style duration resets. For + * multi-bucket reset headers the earliest future instant wins — + * optimistic, because a premature retry just re-records the hint, while + * over-waiting silently costs the user the whole window.

+ */ + static long extractRetryAfterMs(Throwable error) { + for (Throwable cur = error; cur != null; cur = cur.getCause()) { + HttpHeaders headers = null; + if (cur instanceof WebClientResponseException wre) { + headers = wre.getHeaders(); + } else if (cur instanceof RestClientResponseException rre) { + headers = rre.getResponseHeaders(); + } + if (headers == null) continue; + long ms = parseRetryWindowMs(headers); + if (ms > 0) return ms; + } + return 0; + } + + private static long parseRetryWindowMs(HttpHeaders headers) { + String retryAfter = headers.getFirst("retry-after"); + if (retryAfter != null && !retryAfter.isBlank()) { + String v = retryAfter.trim(); + if (v.chars().allMatch(Character::isDigit)) { + return clampHint(Long.parseLong(v) * 1000); + } + try { + long epochMs = ZonedDateTime.parse(v, DateTimeFormatter.RFC_1123_DATE_TIME) + .toInstant().toEpochMilli(); + return clampHint(epochMs - System.currentTimeMillis()); + } catch (DateTimeParseException ignored) { + // fall through to the reset headers + } + } + long best = 0; + for (String name : ANTHROPIC_RESET_HEADERS) { + String v = headers.getFirst(name); + if (v == null || v.isBlank()) continue; + try { + long delta = Instant.parse(v.trim()).toEpochMilli() - System.currentTimeMillis(); + if (delta > 0 && (best == 0 || delta < best)) best = delta; + } catch (DateTimeParseException ignored) { + } + } + if (best > 0) return clampHint(best); + for (String name : OPENAI_RESET_HEADERS) { + long ms = parseGoDurationMs(headers.getFirst(name)); + if (ms > 0 && (best == 0 || ms < best)) best = ms; + } + return best > 0 ? clampHint(best) : 0; + } + + private static long parseGoDurationMs(String value) { + if (value == null || value.isBlank()) return 0; + java.util.regex.Matcher m = GO_DURATION.matcher(value.trim()); + if (!m.matches()) return 0; + long ms = 0; + if (m.group(1) != null) ms += Long.parseLong(m.group(1)) * 3_600_000L; + if (m.group(2) != null) ms += Long.parseLong(m.group(2)) * 60_000L; + if (m.group(3) != null) ms += (long) (Double.parseDouble(m.group(3)) * 1000); + if (m.group(4) != null) ms += Long.parseLong(m.group(4)); + return ms; + } + + private static long clampHint(long ms) { + if (ms <= 0) return 0; + return Math.max(MIN_HINT_MS, Math.min(ms, MAX_HINT_MS)); + } + /** 提取完整异常链信息用于关键字匹配 */ private static String extractFullErrorChain(Throwable error) { StringBuilder sb = new StringBuilder(); @@ -640,6 +790,17 @@ public class NodeStreamingChatHelper { int failoverCount = 0; int llmCallCount = 0; long callStartMs = System.currentTimeMillis(); + // True once the generic routing below has already recorded a health + // failure for this incident — stops the post-loop fallback record from + // double-counting it. + boolean healthRecorded = false; + // Carries the ErrorType behind each null-return retry so the next + // attempt's backoff can be type-aware (see doStreamCall). + AtomicReference retryType = new AtomicReference<>(); + // Provider-stated retry window (ms) parsed from the latest 429/529 + // response headers; 0 when absent. Consumed by the next attempt's + // backoff and by the health-cooldown override on failover. + AtomicReference retryHint = new AtomicReference<>(0L); // 主模型重试循环 StreamResult lastResult = null; @@ -655,111 +816,82 @@ public class NodeStreamingChatHelper { } llmCallCount++; if (attempt > 0) retryCount++; - lastResult = doStreamCall(chatModel, prompt, conversationId, phase, broadcast, attempt, true); + lastResult = doStreamCall(chatModel, prompt, conversationId, phase, broadcast, attempt, true, retryType, retryHint); if (lastResult != null) { - // PTL: 不重试，直接返回给上层 Node 处理 - if (lastResult.errorType() == ErrorType.PROMPT_TOO_LONG) { - return lastResult; - } - // AUTH: primary key 失效不会自愈，跳过同模型重试，交给 fallback chain - // — 其它 provider 的 key 可能仍然可用（与 BILLING / MODEL_NOT_FOUND 同策略）。 - // recordPrimary(false) 仍记一次失败用于 healthTracker 冷却累计。 - // 若 fallback chain 全部 401，walker 末尾会把最后一次 AUTH_ERROR 透出， - // 不会静默吞错。 - if (lastResult.errorType() == ErrorType.AUTH_ERROR) { - log.warn("[{}] Primary auth failed — skipping same-model retries, handing off to fallback chain", phase); - recordPrimary(false); - removeFromPool(primaryProviderId, ErrorType.AUTH_ERROR, lastResult.errorMessage()); - break; - } - // BILLING — provider-side hard failure (out of credit). Won't change - // on retry and affects every model on the provider, so evict it and - // hand off to the fallback chain (a different provider may have credits). - if (lastResult.errorType() == ErrorType.BILLING) { - log.warn("[{}] Primary billing failure — skipping same-model retries, handing off to fallback chain", phase); - recordPrimary(false); - removeFromPool(primaryProviderId, ErrorType.BILLING, lastResult.errorMessage()); - break; - } - // MODEL_NOT_FOUND — the provider rejected this specific model id. The - // provider itself is healthy, so do NOT evict it from the pool or - // record a provider-level failure: that would take its sibling models - // down too. Just skip same-model retries and hand off to the fallback - // chain — a different provider may recognize the model name. - if (lastResult.errorType() == ErrorType.MODEL_NOT_FOUND) { - log.warn("[{}] Primary model not found — handing off to fallback chain " - + "(provider kept available for its other models)", phase); - break; - } - // CLIENT_ERROR (400 Bad Request): 不重试（参数/格式错误重试也不会变） - if (lastResult.errorType() == ErrorType.CLIENT_ERROR) { - return lastResult; - } - // THINKING_BLOCK_ERROR: 剥离旧 thinking 块后单次重试 - if (lastResult.errorType() == ErrorType.THINKING_BLOCK_ERROR && attempt == 0) { - log.warn("[{}] Thinking block error detected, stripping old thinking and retrying once", phase); - prompt = stripThinkingFromPrompt(prompt); - continue; // 重试一次 - } - if (lastResult.errorType() == ErrorType.THINKING_BLOCK_ERROR) { - return lastResult; // 已经重试过了 - } - // EMPTY_RESPONSE — transient gateway blip often resolves on same-model - // retry (e.g., proxy timeout returns HTTP 200 with empty body). - // Retry up to MAX_RETRIES_EMPTY_RESPONSE before handing off to the - // fallback chain. A different provider has a better chance of - // succeeding if the same model repeatedly returns nothing. - if (lastResult.errorType() == ErrorType.EMPTY_RESPONSE) { - if (attempt < MAX_RETRIES_EMPTY_RESPONSE) { - log.warn("[{}] Primary returned empty response (attempt {}/{}), retrying same model...", - phase, attempt + 1, MAX_RETRIES_EMPTY_RESPONSE + 1); - continue; - } - log.warn("[{}] Primary exhausted empty-response retries — handing off to fallback chain", phase); - recordPrimary(false); - break; - } - // 成功 - if (lastResult.errorMessage() == null || lastResult.errorType() == ErrorType.NONE) { + ErrorType errType = lastResult.errorType(); + // Success — reaffirm health / pool membership and return. + if (lastResult.errorMessage() == null || errType == ErrorType.NONE) { recordPrimary(true); addToPool(primaryProviderId); logPerfSummary(phase, conversationId, callStartMs, llmCallCount, retryCount, failoverCount); return lastResult; } - // RATE_LIMIT / SERVER_ERROR / UNKNOWN past their retry budget are - // provider-level failures: the same model will not recover within - // this turn, but a different provider can. Break to the fallback - // chain instead of returning — recordPrimary(false) runs once at - // the post-loop provider health check below, and if every fallback - // also fails the chain walker re-surfaces this same error to the - // caller. - // UNKNOWN errors are included defensively: an error we can't - // classify may be a transient (mis-classified by our keyword - // patterns) or a fatal (truly new error shape). Retrying with a - // smaller budget (MAX_RETRIES_UNKNOWN=5 vs MAX_RETRIES=10) is - // safer than immediate termination — MAX_TOTAL_DURATION_MS provides - // the ultimate safety net. - if (lastResult.errorType() == ErrorType.RATE_LIMIT - || lastResult.errorType() == ErrorType.SERVER_ERROR - || lastResult.errorType() == ErrorType.UNKNOWN) { - log.warn("[{}] Primary exhausted retries (type={}) — handing off to fallback chain", - phase, lastResult.errorType()); - break; + // PROMPT_TOO_LONG — side-effectful recovery owned by the caller: + // the node runs structured compaction and retries by itself, so + // the error must surface unchanged (never routed to fallback — + // a different provider has a different window and the caller + // would lose the compaction signal). + if (errType == ErrorType.PROMPT_TOO_LONG) { + return lastResult; } - // Any truly unhandled error type — safety net. Prefer falling back - // over terminating the entire call. If this branch is ever hit in - // production, the type should be added explicitly above. - recordPrimary(false); - log.warn("[{}] Primary returned unhandled error type={} — handing off to fallback chain", - phase, lastResult.errorType()); + // THINKING_BLOCK_ERROR — side-effectful recovery: strip stale + // thinking blocks from the prompt, then retry once. Kept as an + // explicit branch because the generic path cannot mutate the + // outgoing prompt. + if (errType == ErrorType.THINKING_BLOCK_ERROR) { + if (attempt == 0) { + log.warn("[{}] Thinking block error detected, stripping old thinking and retrying once", phase); + prompt = stripThinkingFromPrompt(prompt); + continue; + } + return lastResult; + } + // EMPTY_RESPONSE retries here in the outer loop — it is a + // result (HTTP 200 with an empty body), not an exception, so + // the inner retry gate never sees it. Same-model retry often + // resolves the transient gateway blip. + if (errType == ErrorType.EMPTY_RESPONSE && attempt < errType.retryBudget()) { + log.warn("[{}] Primary returned empty response (attempt {}/{}), retrying same model...", + phase, attempt + 1, errType.retryBudget() + 1); + continue; + } + // Generic routing — driven entirely by the ErrorType policy + // attributes. By the time a typed error result surfaces here + // the type's same-model retry budget is already exhausted + // (enforced inside the call for exception-path types). + if (!errType.failsOver()) { + // Fails identically everywhere (e.g. CLIENT_ERROR) — + // surface to the caller instead of burning the chain. + return lastResult; + } + if (errType.countsHealth()) { + // A rate-limit response carrying an explicit retry window + // becomes a health-cooldown override: later turns skip the + // provider until the stated instant instead of re-probing + // it every ~5 minutes and re-collecting the same 429. + Long hintMs = retryHint.get(); + if (errType == ErrorType.RATE_LIMIT && hintMs != null && hintMs > 0) { + recordPrimaryFailure(hintMs); + } else { + recordPrimary(false); + } + healthRecorded = true; + } + if (errType.evictsProvider()) { + removeFromPool(primaryProviderId, errType, lastResult.errorMessage()); + } + log.warn("[{}] Primary failed (type={}) — handing off to fallback chain", phase, errType); break; } // lastResult == null 表示需要重试 } - // If we exhausted the retry loop without a verdict, primary effectively - // failed. Only count it against provider health for provider-level errors — - // a MODEL_NOT_FOUND break above must not nudge the provider toward cooldown. - if (!primarySkipped && lastResult != null && isProviderLevelFailure(lastResult.errorType())) { + // Exits that bypassed the generic routing (time-budget break, an + // EMPTY_RESPONSE retry cut short by the loop bound) still count one + // health failure for provider-level errors. healthRecorded guards + // against double-counting the generic-path breaks; model-scoped + // errors (MODEL_NOT_FOUND et al.) never dent provider health. + if (!primarySkipped && !healthRecorded && lastResult != null + && isProviderLevelFailure(lastResult.errorType())) { recordPrimary(false); } @@ -797,7 +929,8 @@ public class NodeStreamingChatHelper { failoverCount++; llmCallCount++; StreamResult fallbackResult = doStreamCall(fallback, prompt, conversationId, - phase + "_fallback_" + (i + 1), broadcast, 0, false); + phase + "_fallback_" + (i + 1), broadcast, 0, false, + new AtomicReference<>(), new AtomicReference<>(0L)); // Accept only fully successful fallbacks. Non-successful results (auth // error, client error, still-rate-limited) propagate to the next // fallback instead of being surfaced as the final result. @@ -840,11 +973,17 @@ public class NodeStreamingChatHelper { /** * 单次流式调用尝试。 + * @param retryTypeRef carries the {@link ErrorType} that caused the + * previous attempt's retry (set on every + * {@code return null}) so the next attempt's backoff + * can be type-aware (OVERLOADED uses the long table). * @return StreamResult 如果成功/降级/不可重试；null 如果应该重试 */ private StreamResult doStreamCall(ChatModel chatModel, Prompt prompt, String conversationId, String phase, - boolean broadcast, int attempt, boolean primaryCall) { + boolean broadcast, int attempt, boolean primaryCall, + AtomicReference retryTypeRef, + AtomicReference retryHintRef) { // Collapse every SystemMessage in the prompt into a single SystemMessage // at index 0. Some OpenAI-compatible providers (LM Studio's built-in // server, certain strict vLLM / SGLang deployments) reject 400 @@ -897,7 +1036,7 @@ public class NodeStreamingChatHelper { } try { - return doStreamCallInner(chatModel, outbound, conversationId, phase, broadcast, attempt, primaryCall); + return doStreamCallInner(chatModel, outbound, conversationId, phase, broadcast, attempt, primaryCall, retryTypeRef, retryHintRef); } finally { // Idempotent: if consumer already took the entry, discard is a no-op. if (relayToken != null) { @@ -929,18 +1068,42 @@ public class NodeStreamingChatHelper { private StreamResult doStreamCallInner(ChatModel chatModel, Prompt prompt, String conversationId, String phase, - boolean broadcast, int attempt, boolean primaryCall) { + boolean broadcast, int attempt, boolean primaryCall, + AtomicReference retryTypeRef, + AtomicReference retryHintRef) { if (attempt > 0) { - long delay = Math.min(backoffBaseMs * (1L << (attempt - 1)), backoffCapMs); - // 加入 jitter 防止雷群效应 - delay += ThreadLocalRandom.current().nextLong(0, Math.max(1, delay / 2)); - delay = Math.min(delay, backoffCapMs); - log.warn("[{}] Retry attempt {}/{} after {}ms for conversation {}", - phase, attempt, MAX_RETRIES, delay, conversationId); + boolean overloaded = retryTypeRef.get() == ErrorType.OVERLOADED; + Long hintedMs = retryHintRef.get(); + long delay; + if (hintedMs != null && hintedMs > 0) { + // The provider stated exactly when to come back — honor it + // (capped: longer windows are handled by failover + the + // health-cooldown override, not by blocking this turn), with + // a small additive jitter so concurrent sessions don't retry + // in lockstep at the stated instant. + delay = Math.min(hintedMs, HINTED_BACKOFF_CAP_MS) + + ThreadLocalRandom.current().nextLong(0, 1_000); + } else if (overloaded) { + // Saturated provider: recovery periods run tens of seconds, so + // the generic 3s-based exponential would burn attempts before + // capacity returns. Table lookup + ±30% jitter (decorrelates + // concurrent conversations re-hitting the same provider). + int idx = Math.min(attempt - 1, OVERLOADED_BACKOFF_MS.length - 1); + long base = OVERLOADED_BACKOFF_MS[idx]; + delay = base * (70 + ThreadLocalRandom.current().nextLong(61)) / 100; + } else { + delay = Math.min(backoffBaseMs * (1L << (attempt - 1)), backoffCapMs); + // 加入 jitter 防止雷群效应 + delay += ThreadLocalRandom.current().nextLong(0, Math.max(1, delay / 2)); + delay = Math.min(delay, backoffCapMs); + } + log.warn("[{}] Retry attempt {}/{} after {}ms (prev type={}) for conversation {}", + phase, attempt, MAX_RETRIES, delay, retryTypeRef.get(), conversationId); // 广播给前端：用户可见的重试倒计时 if (broadcast) { + String cause = overloaded ? "模型服务繁忙" : "请求频率受限"; broadcastDelta(conversationId, "warning", - buildDeltaJson("⏱️ 请求频率受限，等待 " + (delay / 1000) + " 秒后重试（第 " + attempt + "/" + MAX_RETRIES + " 次）...")); + buildDeltaJson("⏱️ " + cause + "，等待 " + (delay / 1000) + " 秒后重试（第 " + attempt + "/" + MAX_RETRIES + " 次）...")); } // Poll stop flag every 100ms so user Stop is honored mid-backoff. long remaining = delay; @@ -1256,6 +1419,14 @@ public class NodeStreamingChatHelper { // ===== 无内容：分类错误并决定是否重试 ===== ErrorType errorType = classifyError(error); + // Extract the provider-stated retry window once per failure and + // publish it for both consumers (next attempt's backoff; health + // cooldown override on failover). Non-throttling types clear the + // slot so a stale hint from an earlier attempt can't leak into an + // unrelated retry's backoff. + retryHintRef.set(errorType == ErrorType.RATE_LIMIT || errorType == ErrorType.OVERLOADED + ? extractRetryAfterMs(error) : 0L); + // PTL: 不重试，返回给上层 Node 处理压缩 if (errorType == ErrorType.PROMPT_TOO_LONG) { log.warn("[{}] Prompt too long error, returning to node for compaction: {}", @@ -1275,46 +1446,30 @@ public class NodeStreamingChatHelper { conversationId, phase, errorType); } - // Auth: 不重试 - if (errorType == ErrorType.AUTH_ERROR) { - log.error("[{}] Authentication error, not retrying: {}", phase, error.getMessage()); - return buildErrorResultWithType("认证失败: " + extractUserFriendlyError(error), - conversationId, phase, errorType); + // Generic retry gate — the ErrorType's own budget decides whether + // this attempt returns null (outer loop retries with backoff) or + // surfaces a typed terminal result for the routing skeleton. + // THINKING_BLOCK_ERROR is excluded: its retry needs the prompt + // mutation (strip thinking) that only the outer loop can do, so it + // always surfaces immediately despite a non-zero budget. + if (errorType != ErrorType.THINKING_BLOCK_ERROR && attempt < errorType.retryBudget()) { + log.warn("[{}] Retryable error (attempt {}/{}, type={}): {}", + phase, attempt, errorType.retryBudget(), errorType, error.getMessage()); + retryTypeRef.set(errorType); + return null; } - // Client error (400): 不重试（参数/格式错误重试也不会变） - if (errorType == ErrorType.CLIENT_ERROR) { - log.error("[{}] Client error (400), not retrying: {}", phase, error.getMessage()); - return buildErrorResultWithType("Bad request: " + extractUserFriendlyError(error), - conversationId, phase, errorType); - } - - // Rate limit / Server error / Unknown: retryable, but with different budgets. - // RATE_LIMIT: cap at 2 retries then failover (RFC 06 D-2). - // SERVER_ERROR: keep full MAX_RETRIES — upstream flaps often self-heal. - // UNKNOWN: conservative cap (5 vs 10). Defensive: retry what we can't - // classify, but with a smaller budget to avoid masking truly fatal - // errors. MAX_TOTAL_DURATION_MS provides the ultimate safety net. - if (errorType == ErrorType.RATE_LIMIT - || errorType == ErrorType.SERVER_ERROR - || errorType == ErrorType.UNKNOWN) { - int effectiveMaxRetries = switch (errorType) { - case RATE_LIMIT -> MAX_RETRIES_RATE_LIMIT; - case UNKNOWN -> MAX_RETRIES_UNKNOWN; - default -> MAX_RETRIES; - }; - if (attempt < effectiveMaxRetries) { - log.warn("[{}] Retryable error (attempt {}/{}, type={}): {}", - phase, attempt, effectiveMaxRetries, errorType, error.getMessage()); - return null; // 返回 null 触发重试 - } - } - - // 不可重试或已耗尽重试 - log.error("[{}] LLM call failed after {} attempts for conversation {}: {}", - phase, attempt + 1, conversationId, error.getMessage()); - return buildErrorResultWithType("LLM 调用失败: " + extractUserFriendlyError(error), - conversationId, phase, errorType); + // Not retryable, or retry budget exhausted — surface with a + // type-appropriate user-facing prefix. + String friendly = extractUserFriendlyError(error); + String message = switch (errorType) { + case AUTH_ERROR -> "认证失败: " + friendly; + case CLIENT_ERROR -> "Bad request: " + friendly; + default -> "LLM 调用失败: " + friendly; + }; + log.error("[{}] LLM call failed (type={}) after {} attempts for conversation {}: {}", + phase, errorType, attempt + 1, conversationId, error.getMessage()); + return buildErrorResultWithType(message, conversationId, phase, errorType); } // ===== 成功（检查是否因 thinking-only 软上限或内容重复被截断） ===== @@ -1786,46 +1941,123 @@ public class NodeStreamingChatHelper { /** * LLM 调用错误类型分类 */ + /** + * Error classification with the recovery policy attached to each type. + * + *

Each constant carries four policy attributes so the retry loop, the + * fallback-chain router, the pool eviction hook, and the health tracker + * all read one source of truth instead of maintaining parallel + * per-type branch chains:

+ *
    + *
  • {@link #retryBudget()} — same-model retry attempts before the + * type is considered exhausted (0 = never retried).
    • + *
  • {@link #failsOver()} — whether an exhausted failure of this type + * hands off to the fallback chain (vs. returning the error to the + * caller, for errors that would fail identically on every provider + * or that the caller must handle, e.g. prompt compaction).
    • + *
  • {@link #evictsProvider()} — provider-wide HARD failure: remove + * the provider from {@code AvailableProviderPool} so later walks + * skip it entirely.
    • + *
  • {@link #countsHealth()} — whether the failure reflects the + * provider's own health and feeds the consecutive-failure + * cooldown in {@code ProviderHealthTracker}. Model-scoped and + * request-scoped errors must not penalise a healthy provider.
    • + *
+ * + *

Two types additionally have side-effectful recovery steps that + * cannot be expressed as attributes and keep explicit branches in the + * loop: {@link #PROMPT_TOO_LONG} (report server-stated window, return to + * node for compaction) and {@link #THINKING_BLOCK_ERROR} (strip stale + * thinking blocks from the prompt, then retry once).

+ */ public enum ErrorType { - /** 无错误 */ - NONE, - /** 速率限制 (429) */ - RATE_LIMIT, - /** 服务端错误 (5xx, timeout) */ - SERVER_ERROR, - /** Prompt 过长 (context length exceeded) */ - PROMPT_TOO_LONG, - /** 认证错误 */ - AUTH_ERROR, - /** 客户端错误 (400 Bad Request, 不支持的格式等) — 不应重试 */ - CLIENT_ERROR, - /** Thinking 块错误（旧消息中的 thinking block 不可修改）— 可剥离后单次重试 */ - THINKING_BLOCK_ERROR, + // retryBudget failsOver evicts countsHealth + /** No error. */ + NONE (0, false, false, false), + /** + * The caller's own key is throttled (HTTP 429). Small retry budget — + * staying on a rate-limited provider wastes time — then fail over. + */ + RATE_LIMIT (MAX_RETRIES_RATE_LIMIT, true, false, true), + /** + * The provider's serving capacity is saturated (HTTP 529, + * "engine_overloaded", "model is overloaded"). The caller's key is + * healthy, so this neither dents provider health (a busy provider is + * not a broken one) nor rotates away eagerly — it waits on the long + * backoff table, then falls over. + */ + OVERLOADED (MAX_RETRIES_OVERLOADED, true, false, false), + /** Transient server / network failure (5xx, timeout, TLS/socket flap). */ + SERVER_ERROR (MAX_RETRIES, true, false, true), + /** + * Context window exceeded. Never retried here — returned to the node, + * which owns structured compaction and its own retry. + */ + PROMPT_TOO_LONG (0, false, false, false), + /** Auth / infrastructure failure (bad key, cert, DNS). Will not self-heal. */ + AUTH_ERROR (0, true, true, true), + /** + * 400-class request-shape error. Fails identically on every provider, + * so neither retried nor failed over — surfaced to the caller. + */ + CLIENT_ERROR (0, false, false, false), + /** + * Stale thinking blocks rejected by the provider. Retried once after + * stripping thinking from the prompt (explicit branch — needs the + * prompt mutation the generic path cannot do). + */ + THINKING_BLOCK_ERROR (1, false, false, false), /** * RFC-009: LLM returned no content, no thinking, and no tool calls. - * Treated as a soft failure — skip same-model retries and hand off to - * the fallback chain directly. Typical cause: upstream rate-limit - * rejection that comes back as HTTP 200 with empty body. + * Typical cause: upstream soft failure surfaced as HTTP 200 with an + * empty body. Retried in the outer loop (it is a result, not an + * exception), then falls over. */ - EMPTY_RESPONSE, + EMPTY_RESPONSE (MAX_RETRIES_EMPTY_RESPONSE, true, false, true), /** * RFC-009 P3.2: payment / billing failure (HTTP 402, "insufficient_quota", * "credit balance is too low", etc.). Distinct from {@link #AUTH_ERROR} * because the right response is to switch provider (a different - * provider may have credits) rather than just terminate. Skips same-model - * retries and falls through to the fallback chain. + * provider may have credits) rather than just terminate. */ - BILLING, + BILLING (0, true, true, true), /** * RFC-009 P3.2: requested model id not recognized by the provider * (HTTP 404, "Model not exist", "model_not_found", DashScope's - * "url error"). Same handling as {@link #BILLING} — heads straight - * to the fallback chain instead of looping retries against a model - * that does not exist. + * "url error"). Model-scoped: heads to the fallback chain but never + * evicts the provider or dents its health — sibling models still work. */ - MODEL_NOT_FOUND, - /** 其他未知错误 */ - UNKNOWN + MODEL_NOT_FOUND (0, true, false, false), + /** + * Unclassifiable. Retried defensively with a conservative budget — + * a transient mis-missed by the keyword patterns is cheaper to retry + * than a lost turn; the wall-clock budget bounds the fatal case. + */ + UNKNOWN (MAX_RETRIES_UNKNOWN, true, false, true); + + private final int retryBudget; + private final boolean failsOver; + private final boolean evictsProvider; + private final boolean countsHealth; + + ErrorType(int retryBudget, boolean failsOver, boolean evictsProvider, boolean countsHealth) { + this.retryBudget = retryBudget; + this.failsOver = failsOver; + this.evictsProvider = evictsProvider; + this.countsHealth = countsHealth; + } + + /** Same-model retry attempts before this type is exhausted (0 = never retried). */ + public int retryBudget() { return retryBudget; } + + /** Whether an exhausted failure hands off to the fallback chain. */ + public boolean failsOver() { return failsOver; } + + /** Whether this failure HARD-removes the provider from the available pool. */ + public boolean evictsProvider() { return evictsProvider; } + + /** Whether this failure counts toward the provider health cooldown tracker. */ + public boolean countsHealth() { return countsHealth; } } /** diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java index 74d2077d..455aa82b 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java @@ -1523,8 +1523,7 @@ public class ChatStreamTracker { /** * RunState 最长无活动时间。从 wall-clock {@code MAX_LIFETIME_MS=30min} - * 切换到 inactivity-based 后默认 30 min — 与 hermes-agent 的 - * {@code gateway_timeout=1800s} 同口径：只要 agent 还在持续产事件 + * 切换到 inactivity-based 后默认 30 min（1800s 空闲超时）：只要 agent 还在持续产事件 * （tool call / content delta / phase transition / progress_update）， * 就一直活下去，墙钟跑 1 小时 2 小时都可以。只有真正"完全静默 ≥ N 分钟" * 才视为卡死并强制清理。 diff --git a/mateclaw-server/src/main/java/vip/mate/llm/failover/AvailableProviderPool.java b/mateclaw-server/src/main/java/vip/mate/llm/failover/AvailableProviderPool.java index 0d7c4ae9..c6fe6864 100644 --- a/mateclaw-server/src/main/java/vip/mate/llm/failover/AvailableProviderPool.java +++ b/mateclaw-server/src/main/java/vip/mate/llm/failover/AvailableProviderPool.java @@ -1,6 +1,7 @@ package vip.mate.llm.failover; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.time.Instant; @@ -17,17 +18,28 @@ import java.util.concurrent.ConcurrentHashMap; *

Two state transitions:

*
    *
  • Add — at startup ({@code ProviderInitProbe}), on user-triggered - * reprobe, or after a {@code ModelConfigChangedEvent}.
    • + * reprobe, after a {@code ModelConfigChangedEvent}, or lazily when a + * TTL'd removal expires (see below). *
  • Remove — when a request hits a provider-wide HARD error - * (AUTH_ERROR / BILLING) — these don't self-heal, so retrying on - * every subsequent call wastes the user's time. SOFT errors - * (RATE_LIMIT / SERVER_ERROR / EMPTY_RESPONSE) keep the provider in - * the pool and are handled by {@link ProviderHealthTracker}'s short - * cooldown instead. A rejected model id (MODEL_NOT_FOUND) is - * model-scoped, not provider-scoped, and never evicts the provider — - * its sibling models stay usable.
    • + * (AUTH_ERROR / BILLING) — retrying on every subsequent call wastes + * the user's time. SOFT errors (RATE_LIMIT / SERVER_ERROR / + * EMPTY_RESPONSE) keep the provider in the pool and are handled by + * {@link ProviderHealthTracker}'s short cooldown instead. A rejected + * model id (MODEL_NOT_FOUND) is model-scoped, not provider-scoped, and + * never evicts the provider — its sibling models stay usable. *
* + *

TTL readmission: AUTH_ERROR and BILLING removals are not + * permanent. Users top up balances, aggregator quotas refresh, and providers + * have transient 401 flaps — none of which the process can observe. Each + * removal carries a {@code readmitAtMs} deadline (from + * {@link ProviderHealthProperties}); once it passes, the next + * {@link #contains} check lazily readmits the provider. No pre-readmission + * probe: the first real call is the probe, and a still-broken provider is + * simply re-evicted (self-correcting). INIT_PROBE and MANUAL removals never + * auto-readmit — the former means the configuration itself is broken, the + * latter is explicit operator intent.

+ * *

State is process-local; a restart re-runs the init probe. That's * intentional — full distributed coordination is out of scope for v1 * (single-node and desktop deployments are the primary targets).

@@ -46,6 +58,18 @@ public class AvailableProviderPool { */ private final Map removalReasons = new ConcurrentHashMap<>(); + private final ProviderHealthProperties props; + + @Autowired + public AvailableProviderPool(ProviderHealthProperties props) { + this.props = props != null ? props : new ProviderHealthProperties(); + } + + /** Convenience constructor with default readmission TTLs. Used by tests. */ + public AvailableProviderPool() { + this(new ProviderHealthProperties()); + } + /** Add (or re-add) a provider to the pool. Clears any prior removal reason. */ public void add(String providerId) { if (providerId == null || providerId.isEmpty()) return; @@ -60,24 +84,53 @@ public class AvailableProviderPool { /** * Remove a provider from the pool with a reason. Idempotent — calling - * twice updates the reason (so the latest cause wins) but doesn't double-log. + * twice updates the reason (so the latest cause wins and the readmission + * TTL restarts from the latest incident) but doesn't double-log. */ public void remove(String providerId, RemovalSource source, String message) { if (providerId == null || providerId.isEmpty()) return; boolean wasMember = members.remove(providerId); - RemovalReason reason = new RemovalReason(source, message, Instant.now().toEpochMilli()); + long now = Instant.now().toEpochMilli(); + long ttl = readmitDelayMs(source); + RemovalReason reason = new RemovalReason(source, message, now, ttl > 0 ? now + ttl : 0); removalReasons.put(providerId, reason); if (wasMember) { - log.warn("[Pool] removing provider={} due to {} ({})", providerId, source, message); + log.warn("[Pool] removing provider={} due to {} ({}){}", providerId, source, message, + ttl > 0 ? " — auto-readmission in " + (ttl / 1000) + "s" : ""); } else { log.debug("[Pool] removal reason updated for already-out provider={}: {} ({})", providerId, source, message); } } - /** Membership check — the walker / primary short-circuit consults this on every entry. */ + /** + * Membership check — the walker / primary short-circuit consults this on + * every entry. Lazily readmits a provider whose removal TTL has expired, + * so recovery needs no scheduler thread and no user action. + */ public boolean contains(String providerId) { - return providerId != null && members.contains(providerId); + if (providerId == null) return false; + if (members.contains(providerId)) return true; + RemovalReason reason = removalReasons.get(providerId); + if (reason != null && reason.readmitAtMs() > 0 + && Instant.now().toEpochMilli() >= reason.readmitAtMs()) { + log.info("[Pool] readmitting provider={} — {} removal TTL expired (removed {}s ago)", + providerId, reason.source(), + (Instant.now().toEpochMilli() - reason.removedAtMs()) / 1000); + add(providerId); + return true; + } + return false; + } + + /** Readmission TTL for a removal source; 0 = never auto-readmit. */ + private long readmitDelayMs(RemovalSource source) { + if (source == null) return 0; + return switch (source) { + case BILLING -> props.getBillingReadmitMs(); + case AUTH_ERROR -> props.getAuthReadmitMs(); + case INIT_PROBE, MANUAL -> 0; + }; } /** @@ -105,8 +158,13 @@ public class AvailableProviderPool { // Records // ============================================================ - /** Why a provider was removed from the pool. {@code removedAtMs} is epoch milliseconds. */ - public record RemovalReason(RemovalSource source, String message, long removedAtMs) {} + /** + * Why a provider was removed from the pool. {@code removedAtMs} is epoch + * milliseconds; {@code readmitAtMs} is the epoch-millisecond deadline + * after which {@link #contains} lazily readmits the provider ({@code 0} + * = never auto-readmitted). + */ + public record RemovalReason(RemovalSource source, String message, long removedAtMs, long readmitAtMs) {} /** * Categorical source of a pool removal. Covers the provider-wide HARD diff --git a/mateclaw-server/src/main/java/vip/mate/llm/failover/ProviderHealthProperties.java b/mateclaw-server/src/main/java/vip/mate/llm/failover/ProviderHealthProperties.java index 3b84f2b6..5b4ea1b3 100644 --- a/mateclaw-server/src/main/java/vip/mate/llm/failover/ProviderHealthProperties.java +++ b/mateclaw-server/src/main/java/vip/mate/llm/failover/ProviderHealthProperties.java @@ -13,6 +13,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties; * enabled: true * failure-threshold: 3 # consecutive failures before cooldown * cooldown-ms: 300000 # 5 minutes + * billing-readmit-ms: 3600000 # auto-readmit a BILLING-evicted provider after 1h (0 = never) + * auth-readmit-ms: 1800000 # auto-readmit an AUTH-evicted provider after 30min (0 = never) * */ @ConfigurationProperties(prefix = "mateclaw.llm.failover.health") @@ -27,6 +29,22 @@ public class ProviderHealthProperties { /** Cooldown window in milliseconds. */ private long cooldownMs = 300_000L; + /** + * TTL after which a provider HARD-removed for BILLING is lazily readmitted + * to the available pool. Users top up balances and aggregator quotas + * refresh hourly — without this, recovery requires a manual reprobe or a + * process restart. {@code 0} disables auto-readmission. + */ + private long billingReadmitMs = 3_600_000L; + + /** + * TTL after which a provider HARD-removed for AUTH_ERROR is lazily + * readmitted. Bounds the damage of provider-side 401 flaps; a genuinely + * bad key just gets re-evicted by its first post-readmission call. + * {@code 0} disables auto-readmission. + */ + private long authReadmitMs = 1_800_000L; + public boolean isEnabled() { return enabled; } public void setEnabled(boolean enabled) { this.enabled = enabled; } @@ -39,4 +57,14 @@ public class ProviderHealthProperties { public void setCooldownMs(long cooldownMs) { this.cooldownMs = Math.max(1000, cooldownMs); } + + public long getBillingReadmitMs() { return billingReadmitMs; } + public void setBillingReadmitMs(long billingReadmitMs) { + this.billingReadmitMs = Math.max(0, billingReadmitMs); + } + + public long getAuthReadmitMs() { return authReadmitMs; } + public void setAuthReadmitMs(long authReadmitMs) { + this.authReadmitMs = Math.max(0, authReadmitMs); + } } diff --git a/mateclaw-server/src/main/java/vip/mate/llm/failover/ProviderHealthTracker.java b/mateclaw-server/src/main/java/vip/mate/llm/failover/ProviderHealthTracker.java index b0dd24ee..eb1a5e2a 100644 --- a/mateclaw-server/src/main/java/vip/mate/llm/failover/ProviderHealthTracker.java +++ b/mateclaw-server/src/main/java/vip/mate/llm/failover/ProviderHealthTracker.java @@ -62,14 +62,46 @@ public class ProviderHealthTracker { return true; } + /** + * Upper bound for a provider-supplied cooldown override. Guards against + * a provider returning an absurd {@code Retry-After} (misconfigured proxy, + * clock-skewed reset timestamp) locking a provider out for days. + */ + static final long MAX_COOLDOWN_OVERRIDE_MS = 2 * 60 * 60 * 1000L; + /** * Record a single failure against {@code providerId}. When the counter * reaches the configured threshold, the provider enters cooldown. */ public void recordFailure(String providerId) { + recordFailure(providerId, 0); + } + + /** + * Record a failure with an optional provider-supplied cooldown override + * (milliseconds), typically parsed from a 429 response's + * {@code Retry-After} / rate-limit reset headers. + * + *

When {@code cooldownOverrideMs > 0} the provider has stated exactly + * when capacity returns, so the cooldown starts immediately — + * waiting for {@link ProviderHealthProperties#getFailureThreshold} more + * consecutive failures would burn extra calls against a window the + * provider already announced. The override is clamped to + * {@link #MAX_COOLDOWN_OVERRIDE_MS} and never shortens an active + * cooldown.

+ */ + public void recordFailure(String providerId, long cooldownOverrideMs) { if (!props.isEnabled() || providerId == null) return; AtomicLong counter = consecutiveFailures.computeIfAbsent(providerId, k -> new AtomicLong()); long failures = counter.incrementAndGet(); + if (cooldownOverrideMs > 0) { + long clamped = Math.min(cooldownOverrideMs, MAX_COOLDOWN_OVERRIDE_MS); + long cooldownEnd = System.currentTimeMillis() + clamped; + cooldownUntilMs.merge(providerId, cooldownEnd, Math::max); + log.warn("[ProviderHealth] provider={} entering cooldown for {}s (provider-stated retry window)", + providerId, clamped / 1000); + return; + } if (failures >= props.getFailureThreshold()) { long cooldownEnd = System.currentTimeMillis() + props.getCooldownMs(); cooldownUntilMs.put(providerId, cooldownEnd); diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/ErrorClassificationTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/ErrorClassificationTest.java index 45c536c5..3fe7b558 100644 --- a/mateclaw-server/src/test/java/vip/mate/agent/graph/ErrorClassificationTest.java +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/ErrorClassificationTest.java @@ -213,9 +213,13 @@ class ErrorClassificationTest { } @Test - @DisplayName("Gateway-rewritten 400 'model is overloaded' → SERVER_ERROR") - void gatewayOverloadedIsServerError() throws Exception { - assertEquals(NodeStreamingChatHelper.ErrorType.SERVER_ERROR, + @DisplayName("Gateway-rewritten 400 'model is overloaded' → OVERLOADED") + void gatewayOverloadedIsOverloaded() throws Exception { + // Previously SERVER_ERROR. The overload semantic now has its own + // type with patient same-provider backoff — still retryable, still + // fails over after its budget, but no longer dents provider health + // (a busy upstream is not a broken provider). + assertEquals(NodeStreamingChatHelper.ErrorType.OVERLOADED, classify(new RuntimeException("400 Bad Request: model is overloaded, please try again"))); } @@ -286,4 +290,45 @@ class ErrorClassificationTest { classify(new RuntimeException( "PKIX path building failed: unable to find valid certification path to requested target"))); } + + // ===== OVERLOADED (provider capacity saturation) ===== + + @Test + @DisplayName("'engine_overloaded' (even alongside 429) → OVERLOADED, not RATE_LIMIT") + void engineOverloadedIsOverloaded() throws Exception { + // Providers reuse the 429 status for capacity saturation. The more + // specific overload semantic must win over the bare 429 pattern — + // an overloaded provider gets patient backoff, not fast failover. + assertEquals(NodeStreamingChatHelper.ErrorType.OVERLOADED, + classify(new RuntimeException("429 Too Many Requests: engine_overloaded"))); + } + + @Test + @DisplayName("Anthropic 529 'overloaded_error' → OVERLOADED") + void anthropic529IsOverloaded() throws Exception { + assertEquals(NodeStreamingChatHelper.ErrorType.OVERLOADED, + classify(new RuntimeException( + "529 {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"Overloaded\"}}"))); + } + + @Test + @DisplayName("'The model is overloaded' → OVERLOADED, not SERVER_ERROR") + void modelOverloadedIsOverloaded() throws Exception { + assertEquals(NodeStreamingChatHelper.ErrorType.OVERLOADED, + classify(new RuntimeException("The model is overloaded. Please try again later."))); + } + + @Test + @DisplayName("SiliconFlow group-saturation message → OVERLOADED") + void siliconflowSaturationIsOverloaded() throws Exception { + assertEquals(NodeStreamingChatHelper.ErrorType.OVERLOADED, + classify(new RuntimeException("50603: 当前分组上游负载已饱和，请稍后再试"))); + } + + @Test + @DisplayName("Plain 429 without overload wording stays RATE_LIMIT") + void plain429StaysRateLimit() throws Exception { + assertEquals(NodeStreamingChatHelper.ErrorType.RATE_LIMIT, + classify(new RuntimeException("429 Too Many Requests"))); + } } diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/ErrorTypePolicyTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/ErrorTypePolicyTest.java new file mode 100644 index 00000000..5437d3f0 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/ErrorTypePolicyTest.java @@ -0,0 +1,79 @@ +package vip.mate.agent.graph; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import vip.mate.agent.graph.NodeStreamingChatHelper.ErrorType; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Table-driven assertions over every {@link ErrorType}'s + * recovery-policy attributes. The retry loop, fallback router, pool eviction + * and health tracker all consume these attributes, so a mis-configured new + * constant would silently change failover behavior — this test makes any + * change to the policy table an explicit, reviewed diff. + */ +class ErrorTypePolicyTest { + + /** One row per constant: {type, retryBudget, failsOver, evictsProvider, countsHealth}. */ + private static final Object[][] POLICY_TABLE = { + {ErrorType.NONE, 0, false, false, false}, + {ErrorType.RATE_LIMIT, 2, true, false, true }, + {ErrorType.OVERLOADED, 5, true, false, false}, + {ErrorType.SERVER_ERROR, 10, true, false, true }, + {ErrorType.PROMPT_TOO_LONG, 0, false, false, false}, + {ErrorType.AUTH_ERROR, 0, true, true, true }, + {ErrorType.CLIENT_ERROR, 0, false, false, false}, + {ErrorType.THINKING_BLOCK_ERROR, 1, false, false, false}, + {ErrorType.EMPTY_RESPONSE, 3, true, false, true }, + {ErrorType.BILLING, 0, true, true, true }, + {ErrorType.MODEL_NOT_FOUND, 0, true, false, false}, + {ErrorType.UNKNOWN, 5, true, false, true }, + }; + + @Test + @DisplayName("Every ErrorType constant appears in the policy table exactly once") + void tableCoversEveryConstant() { + assertEquals(ErrorType.values().length, POLICY_TABLE.length, + "A new ErrorType constant was added without a policy-table row — " + + "add it here so its recovery policy is an explicit, reviewed decision"); + } + + @Test + @DisplayName("Policy attributes match the table") + void attributesMatchTable() { + for (Object[] row : POLICY_TABLE) { + ErrorType t = (ErrorType) row[0]; + assertEquals((int) row[1], t.retryBudget(), t + ".retryBudget"); + assertEquals(row[2], t.failsOver(), t + ".failsOver"); + assertEquals(row[3], t.evictsProvider(), t + ".evictsProvider"); + assertEquals(row[4], t.countsHealth(), t + ".countsHealth"); + } + } + + @Test + @DisplayName("Evicting types always count toward provider health") + void evictingImpliesHealth() { + // A HARD-evicting failure is by definition a provider-level failure; + // an evicting type that skips health tracking would TTL-readmit into + // a tracker that never saw the incident. + for (ErrorType t : ErrorType.values()) { + if (t.evictsProvider()) { + assertTrue(t.countsHealth(), t + " evicts the provider but does not count health"); + } + } + } + + @Test + @DisplayName("OVERLOADED never dents provider health or pool membership") + void overloadedIsHealthNeutral() { + // A saturated provider is busy, not broken: eviction or cooldown would + // take a healthy provider out of the chain exactly when every other + // conversation needs it most. + assertFalse(ErrorType.OVERLOADED.evictsProvider()); + assertFalse(ErrorType.OVERLOADED.countsHealth()); + assertTrue(ErrorType.OVERLOADED.failsOver()); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/RetryAfterExtractionTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/RetryAfterExtractionTest.java new file mode 100644 index 00000000..96ec4396 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/RetryAfterExtractionTest.java @@ -0,0 +1,98 @@ +package vip.mate.agent.graph; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; +import org.springframework.web.reactive.function.client.WebClientResponseException; + +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.time.ZoneOffset; +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Parsing of provider-stated retry windows out of 429/529 + * response headers, straight from the exception chain (both WebFlux and + * RestClient exception shapes carry their response headers). + */ +class RetryAfterExtractionTest { + + private static Throwable ex429(HttpHeaders headers) { + return WebClientResponseException.create( + HttpStatus.TOO_MANY_REQUESTS.value(), "Too Many Requests", + headers, new byte[0], StandardCharsets.UTF_8); + } + + @Test + @DisplayName("Retry-After: 7 (delta-seconds) → 7000ms") + void deltaSeconds() { + HttpHeaders h = new HttpHeaders(); + h.add("Retry-After", "7"); + assertEquals(7_000, NodeStreamingChatHelper.extractRetryAfterMs(ex429(h))); + } + + @Test + @DisplayName("Retry-After: HTTP-date → positive delta") + void httpDate() { + HttpHeaders h = new HttpHeaders(); + ZonedDateTime future = ZonedDateTime.now(ZoneOffset.UTC).plusSeconds(30); + h.add("Retry-After", DateTimeFormatter.RFC_1123_DATE_TIME.format(future)); + long ms = NodeStreamingChatHelper.extractRetryAfterMs(ex429(h)); + assertTrue(ms > 25_000 && ms <= 31_000, "expected ≈30s, got " + ms); + } + + @Test + @DisplayName("Anthropic RFC-3339 reset instants → earliest future delta wins") + void anthropicResetInstant() { + HttpHeaders h = new HttpHeaders(); + h.add("anthropic-ratelimit-requests-reset", Instant.now().plusSeconds(600).toString()); + h.add("anthropic-ratelimit-tokens-reset", Instant.now().plusSeconds(60).toString()); + long ms = NodeStreamingChatHelper.extractRetryAfterMs(ex429(h)); + assertTrue(ms > 55_000 && ms <= 61_000, "earliest bucket (≈60s) must win, got " + ms); + } + + @Test + @DisplayName("OpenAI duration-style x-ratelimit-reset ('6m0s') → 360000ms") + void openaiDurationStyle() { + HttpHeaders h = new HttpHeaders(); + h.add("x-ratelimit-reset-requests", "6m0s"); + assertEquals(360_000, NodeStreamingChatHelper.extractRetryAfterMs(ex429(h))); + } + + @Test + @DisplayName("Sub-second values clamp up to the 1s floor") + void subSecondClampsToFloor() { + HttpHeaders h = new HttpHeaders(); + h.add("x-ratelimit-reset-tokens", "120ms"); + assertEquals(1_000, NodeStreamingChatHelper.extractRetryAfterMs(ex429(h))); + } + + @Test + @DisplayName("Absurdly long Retry-After clamps to the 2h ceiling") + void hugeValueClampsToCeiling() { + HttpHeaders h = new HttpHeaders(); + h.add("Retry-After", String.valueOf(7 * 24 * 3600)); // one week + assertEquals(2 * 60 * 60 * 1000L, NodeStreamingChatHelper.extractRetryAfterMs(ex429(h))); + } + + @Test + @DisplayName("No usable headers → 0 (no hint)") + void noHeadersNoHint() { + assertEquals(0, NodeStreamingChatHelper.extractRetryAfterMs(ex429(new HttpHeaders()))); + assertEquals(0, NodeStreamingChatHelper.extractRetryAfterMs(new RuntimeException("429 plain"))); + } + + @Test + @DisplayName("Hint is found through a wrapping cause chain") + void foundThroughCauseChain() { + HttpHeaders h = new HttpHeaders(); + h.add("Retry-After", "5"); + RuntimeException wrapped = new RuntimeException("provider call failed", ex429(h)); + assertEquals(5_000, NodeStreamingChatHelper.extractRetryAfterMs(wrapped)); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCleanupTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCleanupTest.java index aefc4cc2..54be6f90 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCleanupTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerCleanupTest.java @@ -16,9 +16,9 @@ import static org.junit.jupiter.api.Assertions.assertTrue; * *

Before the fix, a long-running agent that kept producing tool calls * (47-minute LLM-review smoke test, round 6) was killed at the 30-minute - * wall-clock mark mid-task. The new behaviour mirrors hermes-agent's - * {@code gateway_timeout}: only completely idle runs are evicted, the - * actively-producing ones can run as long as they need to. + * wall-clock mark mid-task. The new behaviour is an idle-timeout contract: + * only completely idle runs are evicted, the actively-producing ones can + * run as long as they need to. */ class ChatStreamTrackerCleanupTest { diff --git a/mateclaw-server/src/test/java/vip/mate/llm/failover/AvailableProviderPoolTest.java b/mateclaw-server/src/test/java/vip/mate/llm/failover/AvailableProviderPoolTest.java index f000f0ee..885d85f7 100644 --- a/mateclaw-server/src/test/java/vip/mate/llm/failover/AvailableProviderPoolTest.java +++ b/mateclaw-server/src/test/java/vip/mate/llm/failover/AvailableProviderPoolTest.java @@ -150,4 +150,81 @@ class AvailableProviderPoolTest { var snap = pool.snapshot(); assertNotNull(snap); } + + // ===== TTL readmission of HARD-removed providers ===== + + private static AvailableProviderPool poolWithReadmitMs(long billingMs, long authMs) { + ProviderHealthProperties props = new ProviderHealthProperties(); + props.setBillingReadmitMs(billingMs); + props.setAuthReadmitMs(authMs); + return new AvailableProviderPool(props); + } + + @Test + @DisplayName("BILLING removal readmits lazily after its TTL") + void billingReadmitsAfterTtl() throws Exception { + AvailableProviderPool p = poolWithReadmitMs(50, 0); + p.add("openai"); + p.remove("openai", RemovalSource.BILLING, "402 insufficient_quota"); + assertFalse(p.contains("openai"), "still evicted inside the TTL window"); + + Thread.sleep(80); + assertTrue(p.contains("openai"), "TTL expired — contains() must lazily readmit"); + // Readmission clears the removal reason like any add(). + assertNull(p.snapshot().get("openai")); + } + + @Test + @DisplayName("AUTH removal readmits after its own TTL") + void authReadmitsAfterTtl() throws Exception { + AvailableProviderPool p = poolWithReadmitMs(0, 50); + p.add("kimi"); + p.remove("kimi", RemovalSource.AUTH_ERROR, "401"); + assertFalse(p.contains("kimi")); + + Thread.sleep(80); + assertTrue(p.contains("kimi")); + } + + @Test + @DisplayName("INIT_PROBE and MANUAL removals never auto-readmit") + void probeAndManualNeverReadmit() throws Exception { + AvailableProviderPool p = poolWithReadmitMs(1, 1); + p.add("openai"); + p.add("ollama"); + p.remove("openai", RemovalSource.INIT_PROBE, "probe failed"); + p.remove("ollama", RemovalSource.MANUAL, "operator disabled"); + + Thread.sleep(30); + assertFalse(p.contains("openai"), "broken configuration must not silently come back"); + assertFalse(p.contains("ollama"), "explicit operator intent must not expire"); + } + + @Test + @DisplayName("TTL of 0 disables auto-readmission entirely") + void zeroTtlDisablesReadmission() throws Exception { + AvailableProviderPool p = poolWithReadmitMs(0, 0); + p.add("openai"); + p.remove("openai", RemovalSource.BILLING, "402"); + + Thread.sleep(30); + assertFalse(p.contains("openai")); + assertEquals(0, p.snapshot().get("openai").readmitAtMs()); + } + + @Test + @DisplayName("Re-removal after readmission restarts the TTL from the latest incident") + void reRemovalRestartsTtl() throws Exception { + AvailableProviderPool p = poolWithReadmitMs(50, 0); + p.add("openai"); + p.remove("openai", RemovalSource.BILLING, "402 first"); + Thread.sleep(80); + assertTrue(p.contains("openai"), "first TTL expired"); + + // Still broken — the first post-readmission call evicts again. + p.remove("openai", RemovalSource.BILLING, "402 second"); + assertFalse(p.contains("openai"), "fresh removal must start a fresh TTL window"); + Thread.sleep(80); + assertTrue(p.contains("openai"), "second TTL expires independently"); + } } diff --git a/mateclaw-server/src/test/java/vip/mate/llm/failover/ProviderHealthTrackerTest.java b/mateclaw-server/src/test/java/vip/mate/llm/failover/ProviderHealthTrackerTest.java index 97d3b478..1de6c0e0 100644 --- a/mateclaw-server/src/test/java/vip/mate/llm/failover/ProviderHealthTrackerTest.java +++ b/mateclaw-server/src/test/java/vip/mate/llm/failover/ProviderHealthTrackerTest.java @@ -114,4 +114,45 @@ class ProviderHealthTrackerTest { assertTrue(snap.get("openai").cooldownRemainingMs() > 0, "cooldown remaining ms must be positive while active"); } + + // ===== Provider-stated cooldown override (Retry-After feedback) ===== + + @Test + @DisplayName("Cooldown override fires immediately, without waiting for the threshold") + void overrideBypassesThreshold() { + tracker.recordFailure("openai", 60_000); + assertTrue(tracker.isInCooldown("openai"), + "one failure with a stated retry window must start cooldown right away"); + long remaining = tracker.snapshot().get("openai").cooldownRemainingMs(); + assertTrue(remaining > 55_000 && remaining <= 60_000, + "cooldown must honor the stated window, got " + remaining); + } + + @Test + @DisplayName("Override clamps to the 2h ceiling") + void overrideClampedToCeiling() { + tracker.recordFailure("openai", 24L * 3600 * 1000); + long remaining = tracker.snapshot().get("openai").cooldownRemainingMs(); + assertTrue(remaining <= ProviderHealthTracker.MAX_COOLDOWN_OVERRIDE_MS, + "override must clamp at 2h, got " + remaining); + } + + @Test + @DisplayName("Override never shortens a longer active cooldown") + void overrideNeverShortens() { + tracker.recordFailure("openai", 3600_000); + tracker.recordFailure("openai", 5_000); + long remaining = tracker.snapshot().get("openai").cooldownRemainingMs(); + assertTrue(remaining > 5_000, + "a later, shorter window must not truncate the active cooldown, got " + remaining); + } + + @Test + @DisplayName("Success clears an override cooldown like any other") + void successClearsOverride() { + tracker.recordFailure("openai", 3600_000); + assertTrue(tracker.isInCooldown("openai")); + tracker.recordSuccess("openai"); + assertFalse(tracker.isInCooldown("openai")); + } }