feat(timeout): 统一超时预算单点治理(2026-09-16用户裁定:超时逻辑禁散落各产线各角色)——payload._timeout语义升级为一次调用总预算落ctx.budget,_post_upstream按deadline约束重试(总时长≈预算,客户端=预算+60恒覆盖上游,消灭3x预算放大→客户端先断连→模糊TimeoutError→分类器误判永久错误的事故链);attempt超时=剩余预算(合法长生成一次用满,373s实测样本),端点配置timeout(百炼120)不再掐断attempt——供应商参数不知道调用方需求,正是09-05提取/09-14develop/09-16QC三次事故共同根因,仅在直连未传_timeout时与_TOTAL_TIMEOUT取大兜底;预算耗尽(<15s)不再发起必然超时的attempt,抛结构化错误

pipeline_llm/inference.py

@@ -41,6 +41,9 @@ _RETRYABLE_STATUS = (429, 500, 502, 503, 504)
 _MAX_ATTEMPTS = 3
 _TOTAL_TIMEOUT = 300
 _CONNECT_TIMEOUT = 30
+# 单次 attempt 最少剩余预算（2026-09-16 统一预算 deadline 配套）：剩余不足
+# 该值不再发起新 attempt（必然超时白烧），直接抛预算耗尽错误
+_MIN_ATTEMPT_SECONDS = 15
 
 EST_TOKENS_PER_CHAR = 2   # 预估用量：中文约 1 字符≈0.5 token，保守按 2 字符 1 token
 
@@ -426,15 +429,34 @@ async def _post_upstream(ctx, payload, require_choices=True):
     kind = 'chat' if require_choices else 'gen'
     # method 独立列（2026-09-06 用户定夺）：默认 POST，profile 声明优先
     method = (profile.get('method') or 'POST').upper()
-    timeout = int(_fnum(ctx.get('timeout')) or _TOTAL_TIMEOUT)
+    # 统一预算 deadline（单点治理，2026-09-16 用户裁定收敛）：
+    #   ctx['budget'] = 本次调用【总预算】（bridge._resolve_budget 经 payload.
+    #     _timeout 恒传）。attempt 超时 = 剩余预算（合法长生成可一次用满，
+    #     373s 实测样本；快速失败的瞬时错误自然留出重试余量）。重试受
+    #     deadline 约束，总时长≈预算，客户端（预算+60s）恒覆盖上游。
+    #   端点配置 timeout（百炼 120 等）不再掐断 attempt——它是供应商参数，
+    #     不知道调用方需求，正是 09-05/09-14/09-16 三次事故的根因；仅在
+    #     直连端点且未传 _timeout 时参与兜底（与 _TOTAL_TIMEOUT 取大）。
+    # 旧实现每 attempt 独享全额超时，最坏 3×预算+退避 > 客户端等待 → 客户端
+    # 先断连报模糊 TimeoutError（分类器误判永久错误，2026-09-16 QC 事故链一环）。
+    budget = int(_fnum(ctx.get('budget')) or 0)
+    if budget <= 0:
+        budget = max(int(_fnum(ctx.get('timeout')) or 0), _TOTAL_TIMEOUT)
+    deadline = time.time() + budget
     # 可诊断性：只记长度不记值（真 key 不落日志）。print 确保进应用日志
     # （模块 logging.getLogger 的 handler 未接应用日志管道，2026-09-04 实测）
-    print("[inference] 组包 url=%s auth头长度=%d body字段=%s" % (
+    print("[inference] 组包 url=%s auth头长度=%d body字段=%s 预算=%ds" % (
         url.split('?')[0], len(headers.get('Authorization', '') or ''),
-        sorted(body.keys())))
+        sorted(body.keys()), budget))
 
     last = None
     for attempt in range(_MAX_ATTEMPTS):
+        remaining = deadline - time.time()
+        if remaining < _MIN_ATTEMPT_SECONDS:
+            raise ValueError('上游调用失败（总预算 %ds 耗尽，已试 %d/%d 次，%s）: %s' % (
+                budget, attempt, _MAX_ATTEMPTS, type(last).__name__ if last else '无',
+                (repr(last) if last else '')[:200]))
+        timeout = int(remaining)
         _t0 = time.time()
         try:
             async with aiohttp.ClientSession() as session:
@@ -887,8 +909,14 @@ async def chat_inference(org_id, user_id, payload, model_name='', task_ref='',
     # llm_usage.call_id 同值 → 流水可反查全部原文（trace.record_roundtrip / llm_call_trace）
     from appPublic.uniqueID import getID
     ctx['call_id'] = getID()
+    # 统一预算（2026-09-16 收敛）：payload._timeout = 本次调用【总预算】→
+    # ctx['budget']（_post_upstream 按 deadline 约束重试，总时长≈预算）。
+    # ctx['timeout'] 保持供应商端点配置语义 = 单 attempt 上限，不再被覆盖
+    # （旧实现覆盖后每 attempt 独享全额预算，最坏 3× 超客户端等待）。
+    # 未传 _timeout（直连端点的第三方调用）→ budget 走 _TOTAL_TIMEOUT 平台
+    # 缺省，不再落到端点小配置掐断长生成。
     if req_timeout > 0:
-        ctx['timeout'] = min(req_timeout, 900)
+        ctx['budget'] = min(req_timeout, 900)
     # 异步模型：提交→轮询→取结果（用量因子走 usages 记账）
     model_row = ctx.get('model_row') or {}
     if (model_row.get('sync_mode') or 'sync') == 'async':