Report streaming failures with the real status, and fail over locally

.env.example

@@ -32,3 +32,16 @@ CODEBUDDY2API_LOG_BODY_LIMIT=65536
 
 # 国际 WorkBuddy 一次性体验积分领取；默认关闭，仅对符合上游资格的账号生效。
 CODEBUDDY2API_AUTO_TRIAL=false
+
+# 换凭证重放次数：失败发生在向下游落第一个字节之前时，最多再换几个凭证就地重放。
+# 默认 0（关闭）：如实把上游 429/502 回给下游。只重放上游确定没收下请求体（建连失败/建连超时）
+# 或用 401/403/429/502/503/504 拒绝的失败；内容审核拒绝与上游已回 200 之后合成的 502 不重放。
+# 计费口径：401/403/429/503 与建连类失败都发生在受理阶段，不产生扣费；502/504 有可能已被后端
+# 处理并计费，但那次结果对下游根本拿不到，不重放也退不回额度——日志给这类重打单独标
+# 「上游可能已处理该请求」，便于按官方用量明细核对。详见 docs/advanced.zh-CN.md「换凭证重放」。
+CODEBUDDY2API_FAILOVER_MAX=0
+
+# 是否把「写请求体超时」也算作上游没收下请求体（同时作用于连接重试与上面的换凭证重放）。
+# 默认 false：写超时只能证明正文没写完，上游有没有按已收到的半截正文计费，从这一侧看不到。
+# 跨境长会话最容易撞的恰是 60s 写超时，确认自己的上游不会按半截正文计费再打开。
+CODEBUDDY2API_RETRY_WRITE_TIMEOUT=false

app/observability.py

@@ -65,6 +65,7 @@ class _Observation:
     monotonic_start: float
     attempts: list = field(default_factory=list)
     failed: bool = False
+    failure_seq: int = 0
     terminal: bool = False
     body_finished: bool = False
     status: int | None = None
@@ -76,6 +77,7 @@ class _Observation:
 
     def fail(self, code):
         self.failed = True
+        self.failure_seq += 1
         self.record["error_code"] = safe_label(code, 80) or "upstream_error"
 
     def usage(self, value, source, priority):
@@ -159,9 +161,42 @@ def observe_attempt(stage, **safe_metadata):
 
 
 def observe_failure(code):
+    """记下失败并返回本次请求的失败序号，供 `observe_recovery(through=…)` 界定撤销范围。"""
     observation = _current.get()
-    if observation is not None:
-        observation.fail(code)
+    if observation is None:
+        return None
+    observation.fail(code)
+    return observation.failure_seq
+
+
+def observe_failure_seq():
+    """当前失败序号的快照；没有失败时为 0，调用方原样传给 `observe_recovery` 即可。"""
+    observation = _current.get()
+    return observation.failure_seq if observation is not None else None
+
+
+def observe_recovery(through=None):
+    """标记「`through` 那一次失败已经被就地重放救回」：请求对下游是完整正常响应。
+
+    失败尝试仍留在 `attempts` 里（另加一条 `failover_recovered` 标记），只是不再决定 outcome
+    —— 否则一次成功的换凭证重放会留下 `outcome=error` + `status_code=200` 这种自相矛盾的
+    审计记录，看板和排障都会把它读成失败。
+
+    `through` 是重放前那次失败的序号，只有它仍然是最新一次失败时才撤销：序号对不上说明
+    重放之后的响应自己又记了新失败（换到的账号回了内容审核拒绝就是这种），那次失败必须留下，
+    否则一个被审核拦截的请求会被持久化成 `outcome=success` 且没有 `error_code`。默认 `None`
+    保持旧的「清掉当前失败」语义，给没有序号概念的调用方兜底。
+    """
+    observation = _current.get()
+    if observation is None or not observation.failed:
+        return
+    if through is not None and observation.failure_seq != through:
+        return
+    code = observation.record.get("error_code") or "upstream_error"
+    observation.failed = False
+    observation.record["error_code"] = None
+    if len(observation.attempts) < 32:
+        observation.attempts.append(safe_attempt({"stage": "failover_recovered", "code": code}))
 
 
 class _Parser:

app/settings.py

@@ -40,6 +40,10 @@ def _item(default, type_, label, *, mode="hot", env=None, minimum=None, maximum=
     "max_request_bytes": _item(32 * 1024 * 1024, "integer", "请求字节上限", env="CODEBUDDY2API_MAX_REQUEST_BYTES", minimum=1, maximum=1024**3),
     "log_body_limit": _item(65536, "integer", "文本正文预览字节", env="CODEBUDDY2API_LOG_BODY_LIMIT", minimum=0, maximum=1024**2),
     "auto_trial": _item(False, "boolean", "自动领取体验积分", env="CODEBUDDY2API_AUTO_TRIAL"),
+    "failover_max": _item(0, "integer", "换凭证重放次数", env="CODEBUDDY2API_FAILOVER_MAX",
+                          minimum=0, maximum=10),
+    "retry_write_timeout": _item(False, "boolean", "写超时参与重放",
+                                 env="CODEBUDDY2API_RETRY_WRITE_TIMEOUT"),
     "audit_max_bytes": _item(256 * 1024 * 1024, "integer", "审计明细预算", minimum=1024**2, maximum=1024**4),
     "audit_retention_days": _item(30, "integer", "审计明细保留天数", minimum=1, maximum=36500),
     "audit_diagnostic_bytes": _item(8192, "integer", "失败诊断最大字节", minimum=0, maximum=8192),

app/upstream_io.py

@@ -18,6 +18,14 @@ def __init__(self, status, raw):
         super().__init__(f"upstream HTTP {status}")
 
 
+class UpstreamHTTPError(UpstreamResponseError):
+    """上游**用 HTTP 状态码**给出的答复（429 / 401 / 503 …）。
+
+    与聚合器从 200 响应体里合成的 502（空流、坏 SSE、已开流后断连）区分开：只有前者的请求
+    确定没被上游收下处理，换一个账号重放不会重复计费；后者上游已经回了 200，可能已经计费。
+    """
+
+
 class ChatSSEAccumulator:
     """聚合 Chat SSE，拒绝错误事件、空输出和无结束标记的残流。"""
 
@@ -169,9 +177,27 @@ async def read_bounded_error(response, limit: int = ERROR_BODY_LIMIT) -> bytes:
     return bytes(buf)
 
 
+# 写下第一个请求体字节之前就失败：上游手里没有任何正文，重放零风险。
+BODY_NOT_ACCEPTED = (httpx.ConnectError, httpx.ConnectTimeout)
+# 写请求体超时：只证明「声明的正文没写完」，证不了上游没收到或没处理已经收到的那部分。
+# 半截正文会怎样是上游的行为，从这一侧观察不到，因此默认不重放（见 `retry_write_timeout`）。
+WRITE_TIMEOUT = (httpx.WriteTimeout,)
+
+
 @asynccontextmanager
-async def open_backend_stream(url, headers, body, *, read_timeout=300, on_retry=None):
-    """只重试一次建连失败，其他错误交给调用方按协议返回。"""
+async def open_backend_stream(url, headers, body, *, read_timeout=300, on_retry=None,
+                              retry_write_timeout=False):
+    """只重试一次「上游确定没收下请求体」的失败（建连失败 / 建连超时），其余交给调用方按协议返回。
+
+    重试每次新建 `AsyncClient`，即重建 TCP+TLS，通常能换到另一个边缘节点。
+
+    `retry_write_timeout=True` 把 60s 写超时也算进重放集。跨境长会话最容易撞的正是写超时
+    而不是建连失败，实测某部署的传输失败 100% 是它；但写超时能证明的只有「正文没写完」，
+    上游是否已按半截正文动过账，这一侧看不到，所以留给运维显式决定。
+
+    响应已经开始之后（`opened` 置位）绝不重放 POST。
+    """
+    retryable = BODY_NOT_ACCEPTED + (WRITE_TIMEOUT if retry_write_timeout else ())
     timeout = httpx.Timeout(read_timeout, connect=15, write=60, pool=15)
     for attempt in range(2):
         opened = False
@@ -181,7 +207,7 @@ async def open_backend_stream(url, headers, body, *, read_timeout=300, on_retry=
                     opened = True
                     yield response
                     return
-        except (httpx.ConnectError, httpx.ConnectTimeout) as error:
+        except retryable as error:
             if opened or attempt == 1:
                 raise
             if on_retry is not None: