Coordinate AI retries and recover shared provider outages

Retry transient requests after one and three minutes and coordinate provider recovery across processes with bounded probe intervals. Preserve completed results when recovery bookkeeping fails, keep concurrent failure accounting accurate, and expose retry state in the UI.

backend/src/alphapulse/ai/llm.py

@@ -7,12 +7,13 @@
 """
 
 import asyncio
+import hashlib
 import json
 import logging
+import uuid
 from collections.abc import AsyncIterator
 from contextvars import ContextVar
 from datetime import UTC, datetime
-from email.utils import parsedate_to_datetime
 from dataclasses import dataclass, field
 from typing import Any, Literal
 
@@ -21,6 +22,7 @@ import httpx
 import httpx2  # anthropic SDK 使用的 HTTP 客户端，异常类与 httpx 不通用
 import openai
 
+from alphapulse.ai.recovery import Gate
 from alphapulse.core.crypto import decrypt
 from alphapulse.models import AiProvider
 
@@ -28,11 +30,8 @@ from alphapulse.models import AiProvider
 TIMEOUT_SECONDS = 600
 # 思考强度：两种协议共有的取值。OpenAI 另有 none/minimal，但实测经 Sub2API 会卡住后 502，不提供
 Effort = Literal["low", "medium", "high", "xhigh", "max"]
-# SDK 自带重试只针对连接错误、429 和 5xx
-MAX_RETRIES = 2
-# 只重试 SDK 不处理的流内错误和读取中断，且只在还没有输出内容时重试。
-STREAM_RETRIES = 3
-STREAM_RETRY_DELAY_SECONDS = 2
+# SDK 重试关闭，所有生成请求仅由公共流负责重试。
+MAX_RETRIES = 0
 # 输出被截断时回给模型的工具结果：参数不完整，不能执行
 TRUNCATED_RESULT = json.dumps({"error": "输出达到最大输出上限被截断，工具参数不完整，未执行。请精简内容后重新调用"}, ensure_ascii=False)
 
@@ -40,6 +39,7 @@ logger = logging.getLogger("alphapulse.ai.llm")
 
 
 news_retry = ContextVar("news_model_retry", default=None)
+fail_fast = ContextVar("ai_fail_fast", default=False)
 
 
 class LlmError(Exception):
@@ -106,70 +106,178 @@ async def list_models(provider: AiProvider) -> list[str]:
         raise LlmError('无法读取模型列表；请检查接入配置，或手动填写模型名称') from error
 
 
+@dataclass
+class RetryStatus:
+    scope: str
+    attempt: int
+    retry_at: str
+    reason: str
+
+
+def recovery_gate(provider):
+    identity = json.dumps([provider.protocol, api_root(provider.base_url), decrypt(provider.api_key_encrypted)])
+    return Gate(hashlib.sha256(identity.encode()).hexdigest())
+
+
+async def availability(provider):
+    try:
+        state = await recovery_gate(provider).call('read')
+    except Exception as error:
+        raise LlmError('无法读取模型故障协调状态，请检查 Redis') from error
+    return {'state': 'probing' if state['probe'] else 'waiting' if state['level'] else 'normal',
+            'retry_at': datetime.fromtimestamp(state['retry_at'], UTC).isoformat() if state['level'] else None,
+            'level': state['level']}
+
+
 async def stream(
     provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec], effort: Effort | None = None,
-) -> AsyncIterator[TextDelta | ThinkingStarted | Completion]:
-    """先逐段产出 TextDelta / ThinkingStarted，最后产出一个 Completion。effort 为空表示不传，由模型决定"""
-    policy = news_retry.get()
-    retries = 2 if policy is not None else STREAM_RETRIES
-    for attempt in range(retries + 1):
-        output = False
-        if policy is not None: policy["output"] = False
+    *, before_retry=None,
+) -> AsyncIterator[TextDelta | ThinkingStarted | Completion | RetryStatus]:
+    """仅重发未开始生成的当前请求；共享故障等待优先于局部重试。"""
+    gate = recovery_gate(provider)
+    identifier = uuid.uuid4().hex
+    retries = 0
+    waited = False
+    policy = {'output': False}
+
+    async def coordinate(operation, epoch=-1):
         try:
-            if provider.protocol == "openai":
-                events = _stream_openai(provider, system, messages, tools, effort)
-            else:
-                events = _stream_anthropic(provider, system, messages, tools, effort)
-            async for event in events:
-                output = True
-                yield event
-            return
-        # 流式读取过程中的网络错误 SDK 不包装，以原始的 httpx / httpx2 异常抛出
-        except (openai.APIError, anthropic.APIError, httpx.TransportError, httpx2.TransportError) as e:
-            # 已经推送过内容再重试会重复输出
-            delay = _news_retry_delay(e, attempt) if policy is not None else STREAM_RETRY_DELAY_SECONDS
-            retryable = delay is not None if policy is not None else _retryable(e)
-            if output or (policy is not None and policy.get("output")) or attempt == retries or not retryable:
-                raise _llm_error(e) from e
-            if policy is not None:
+            return await gate.call(operation, identifier, epoch)
+        except Exception as error:
+            raise LlmError('模型故障协调不可用，请检查 Redis；未绕过保护发送请求') from error
+
+    async def record_outcome(operation, epoch):
+        # 已完成响应及取消语义不能被协调状态写入或租约清理失败覆盖。
+        # 新请求仍必须通过严格的 enter 门禁，失败时不会绕过 Redis。
+        try:
+            await coordinate(operation, epoch)
+        except LlmError:
+            logger.warning('模型响应状态登记失败：%s；保留原始结果', operation)
+
+    async def notice(scope, count, timestamp, reason):
+        from alphapulse.news_research.progress import step, current
+        if current.get() is not None:
+            from alphapulse.ai.recovery import mark_waiting
+            await mark_waiting(identifier)
+        at = datetime.fromtimestamp(timestamp, UTC).isoformat()
+        step('等待模型重试', retry_scope=scope, retry_attempt=count, retry_at=at)
+        return RetryStatus(scope, count, at, reason)
+
+    async def heartbeat(epoch):
+        while True:
+            await asyncio.sleep(30)
+            await coordinate('heartbeat', epoch)
+
+    try:
+        while True:
+            admission = await coordinate('enter')
+            if not admission['admitted']:
+                status = await notice('upstream', admission['level'],
+                                max(admission['retry_at'], admission['now']+1), '同一接入暂时不可用，等待单请求试探恢复')
+                if fail_fast.get():
+                    raise LlmError(f'{status.reason}；下次尝试：{status.retry_at}')
+                yield status
+                waited = True
+                await asyncio.sleep(1)
+                continue
+            epoch = admission['epoch']
+            pulse = asyncio.create_task(heartbeat(epoch)) if admission['probe'] else None
+            output = False
+            try:
+                if waited and before_retry:
+                    await before_retry()
                 from alphapulse.news_research.progress import step
-                step(f"上游暂时拒绝：{_llm_error(e)}；{delay:g}秒后重试（{attempt + 1}/2）")
-            logger.warning("模型调用失败，%ss 后第 %s 次重试：%s", delay, attempt + 1, _llm_error(e))
-            await asyncio.sleep(delay)
-            if policy is not None:
-                from alphapulse.news_research.progress import step
-                step(f"等待模型响应（重试 {attempt + 1}/2）")
+                from alphapulse.ai.recovery import clear_waiting
+                await clear_waiting(identifier)
+                step('等待模型响应', retry_at=None, retry_scope=None)
+                policy['output'] = False
+                output = False
+                adapter = _stream_openai if provider.protocol == 'openai' else _stream_anthropic
+                completed = False
+                async for event in _guarded_events(adapter(provider, system, messages, tools, effort), policy):
+                    output |= not isinstance(event, Completion)
+                    if isinstance(event, Completion):
+                        completed = True
+                        # 调用方可能取得 Completion 后直接关闭流；先尝试登记恢复，失败仍保留完整响应。
+                        await record_outcome('success', epoch)
+                    yield event
+                if not completed:
+                    raise LlmError('模型连接提前结束，未收到完整响应的结束标志')
+                return
+            except (openai.APIError, anthropic.APIError, httpx.TransportError, httpx2.TransportError, LlmError) as error:
+                temporary = _temporary(error)
+                state = await coordinate('failure', epoch) if temporary else await coordinate('read')
+                if output or policy['output'] or _timeout(error) or not temporary:
+                    raise _llm_error(error) from error
+                if state['level']:
+                    waited = True
+                    continue
+                if retries >= 2:
+                    raise _llm_error(error) from error
+                delay = (60, 180)[retries]
+                retries += 1
+                yield await notice('request', retries, datetime.now(UTC).timestamp()+delay, '单请求临时故障，等待重试')
+                await _wait_local(delay, coordinate)
+                waited = True
+            finally:
+                if pulse:
+                    pulse.cancel()
+                    try:
+                        await pulse
+                    except asyncio.CancelledError:
+                        pass
+                    except LlmError:
+                        logger.warning('模型试探租约续期失败；保留原始结果')
+                await record_outcome('release', epoch)
+    finally:
+        from alphapulse.ai.recovery import clear_waiting
+        await clear_waiting(identifier)
 
 
-def _news_retry_delay(error, attempt):
-    """One bounded retry owner for news; never repeat a timed out request."""
-    if isinstance(error, (openai.APITimeoutError, anthropic.APITimeoutError, httpx.TimeoutException, httpx2.TimeoutException)):
-        return None
+async def _guarded_events(events, policy):
+    # ContextVar 仅覆盖模型读取，不跨 yield 泄漏到调用方或异步生成器清理任务。
+    try:
+        while True:
+            token = news_retry.set(policy)
+            try:
+                event = await anext(events)
+            except StopAsyncIteration:
+                return
+            finally:
+                news_retry.reset(token)
+            yield event
+    finally:
+        await events.aclose()
+
+
+async def _wait_local(delay, coordinate):
+    deadline = asyncio.get_running_loop().time()+delay
+    while asyncio.get_running_loop().time() < deadline:
+        await asyncio.sleep(max(0, min(1, deadline-asyncio.get_running_loop().time())))
+        if (await coordinate('read'))['level']:
+            break
+
+
+def _timeout(error):
+    return isinstance(error, (openai.APITimeoutError, anthropic.APITimeoutError,
+                             httpx.TimeoutException, httpx2.TimeoutException))
+
+
+def _temporary(error):
     status = getattr(error, 'status_code', None)
-    body = getattr(error, 'body', None)
-    detail = body.get('error', body) if isinstance(body, dict) else {}
-    message = str((detail.get('message') if isinstance(detail, dict) else detail) or getattr(error, 'message', '')).lower()
-    concurrency = _in_stream_error(error) and 'concurrency limit exceeded' in message
-    connection = isinstance(error, (openai.APIConnectionError, anthropic.APIConnectionError, httpx.TransportError, httpx2.TransportError))
-    if not (status == 429 or status is not None and 500 <= status <= 599 or concurrency or connection):
-        return None
-    delay = 30 * (attempt + 1)
-    response = getattr(error, 'response', None)
-    header = response.headers.get('retry-after') if response is not None else None
-    if header:
-        try:
-            try: seconds = float(header)
-            except ValueError: seconds = (parsedate_to_datetime(header) - datetime.now(UTC)).total_seconds()
-            if seconds > 120: return None
-            if seconds > 0: delay = max(delay, seconds)
-        except (ValueError, TypeError, OverflowError):
-            pass
-    return delay
+    message = str(error).lower()
+    return (_timeout(error) or status == 429 or status is not None and 500 <= status <= 599
+            or isinstance(error, (openai.APIConnectionError, anthropic.APIConnectionError,
+                                  httpx.TransportError, httpx2.TransportError))
+            or _in_stream_error(error) and any(term in message for term in
+                ('concurrency limit', 'overloaded', 'rate limit', 'temporarily unavailable'))
+            or isinstance(error, LlmError) and '未收到完整响应的结束标志' in message)
 
 
 def _news_output():
     policy = news_retry.get()
-    if policy is not None: policy['output'] = True
+    if policy is not None:
+        policy['output'] = True
 
 
 def _in_stream_error(e: Exception) -> bool:
@@ -180,16 +288,9 @@ def _in_stream_error(e: Exception) -> bool:
     return isinstance(e, anthropic.APIStatusError) and e.status_code == 200
 
 
-def _retryable(e: Exception) -> bool:
-    """建连重试归 SDK；这里只处理流内错误与读取中断，超时不重试。"""
-    if isinstance(e, (openai.APITimeoutError, anthropic.APITimeoutError, httpx.TimeoutException, httpx2.TimeoutException)):
-        return False
-    return _in_stream_error(e) or isinstance(
-        e, (httpx.TransportError, httpx2.TransportError),
-    )
-
-
 def _llm_error(e: Exception) -> LlmError:
+    if isinstance(e, LlmError):
+        return e
     if _in_stream_error(e):
         error = e.body.get("error") if isinstance(e.body, dict) else None
         message = error.get("message") if isinstance(error, dict) else None
@@ -231,7 +332,7 @@ def _openai_messages(system: str, messages: list[dict]) -> list[dict]:
 async def _stream_openai(provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec], effort: Effort | None) -> AsyncIterator[TextDelta | Completion]:
     client = openai.AsyncOpenAI(
         api_key=decrypt(provider.api_key_encrypted), base_url=f"{api_root(provider.base_url)}/v1",
-        timeout=TIMEOUT_SECONDS, max_retries=0 if news_retry.get() is not None else MAX_RETRIES,
+        timeout=TIMEOUT_SECONDS, max_retries=0,
     )
     kwargs: dict[str, Any] = {}
     if tools:
@@ -310,7 +411,7 @@ def _anthropic_messages(messages: list[dict]) -> list[dict]:
 async def _stream_anthropic(provider: AiProvider, system: str, messages: list[dict], tools: list[ToolSpec], effort: Effort | None) -> AsyncIterator[TextDelta | ThinkingStarted | Completion]:
     client = anthropic.AsyncAnthropic(
         api_key=decrypt(provider.api_key_encrypted), base_url=api_root(provider.base_url),
-        timeout=TIMEOUT_SECONDS, max_retries=0 if news_retry.get() is not None else MAX_RETRIES,
+        timeout=TIMEOUT_SECONDS, max_retries=0,
     )
     kwargs: dict[str, Any] = {}
     if tools:

backend/src/alphapulse/ai/recovery.py

@@ -0,0 +1,85 @@
+"""跨进程模型接入熔断；只有持有租约的请求可以试探恢复。"""
+import json
+import logging
+from redis.exceptions import RedisError
+from alphapulse.core.redis import KEY_PREFIX, redis
+
+# 时间以 Redis 为准。epoch 隔离旧请求；失败窗口只计不同逻辑请求。
+SCRIPT = r'''
+local raw = redis.call('GET', KEYS[1])
+local s = raw and cjson.decode(raw) or {epoch=0, level=0, retry_at=0, failures={}}
+local now = tonumber(redis.call('TIME')[1])
+local op, id, epoch = ARGV[1], ARGV[2], tonumber(ARGV[3])
+local function save()
+ redis.call('SET', KEYS[1], cjson.encode(s), 'EX', 86400)
+end
+local function open()
+ s.level = math.min(s.level+1, 3)
+ s.retry_at = now + ({180,600,900})[s.level]
+ s.epoch = s.epoch+1
+ s.owner = nil; s.lease_until = nil; s.failures = {}
+end
+local admitted = false
+if op == 'enter' then
+ if s.level == 0 then admitted = true
+ elseif now >= s.retry_at and (not s.owner or s.lease_until <= now) then
+  s.owner=id; s.lease_until=now+660; admitted=true; save()
+ end
+elseif epoch == s.epoch then
+ if op == 'success' then
+  if s.level == 0 or s.owner == id then
+   -- 正常成功只清空失败窗口，不使并发中的其他请求失效。
+   -- 只有从整体故障恢复才切换代次，隔离旧试探结果。
+   local next_epoch = s.epoch
+   if s.level > 0 then next_epoch = next_epoch+1 end
+   s={epoch=next_epoch,level=0,retry_at=0,failures={}}; save()
+  end
+ elseif op == 'failure' then
+  if s.level > 0 then
+   if s.owner == id then open(); save() end
+  else
+   for key, at in pairs(s.failures) do
+    if at <= now-60 then s.failures[key]=nil end
+   end
+   s.failures[id]=now
+   local count=0
+   for _ in pairs(s.failures) do count=count+1 end
+   if count >= 3 then open() end
+   save()
+  end
+ elseif s.owner == id then
+  if op == 'heartbeat' then s.lease_until=now+660
+  elseif op == 'release' then s.owner=nil; s.lease_until=nil end
+  save()
+ end
+end
+return cjson.encode({epoch=s.epoch,level=s.level, retry_at=s.retry_at,
+ probe=s.owner ~= nil, admitted=admitted, now=now})
+'''
+
+
+class Gate:
+    def __init__(self, scope):
+        self.key = KEY_PREFIX + 'ai:recovery:' + scope
+
+    async def call(self, operation, request_id='', epoch=-1):
+        return json.loads(await redis.eval(SCRIPT, 1, self.key, operation, request_id, epoch))
+
+
+async def mark_waiting(identifier):
+    import time
+    await redis.zadd(KEY_PREFIX + 'ai:news-waiting', {identifier: time.time()+300})
+
+
+async def clear_waiting(identifier):
+    try:
+        await redis.zrem(KEY_PREFIX + 'ai:news-waiting', identifier)
+    except RedisError:
+        # 展示计数有过期兜底，清理失败不能掩盖原始错误或取消。
+        logging.getLogger(__name__).warning('无法清理模型等待计数')
+
+
+async def waiting_count():
+    import time
+    await redis.zremrangebyscore(KEY_PREFIX + 'ai:news-waiting', '-inf', time.time())
+    return await redis.zcard(KEY_PREFIX + 'ai:news-waiting')

backend/src/alphapulse/api/chat.py

@@ -170,7 +170,9 @@ async def _message_events(body: SendIn) -> AsyncIterator[ServerSentEvent]:
                         yield event("delta", {"text": item.text})
                     elif isinstance(item, llm.ThinkingStarted):
                         yield event("thinking", {})
-                    else:
+                    elif isinstance(item, llm.RetryStatus):
+                        yield event("retry", {"scope": item.scope, "attempt": item.attempt, "retry_at": item.retry_at, "reason": item.reason})
+                    elif isinstance(item, llm.Completion):
                         completion = item
             except llm.LlmError as e:
                 if text:

backend/src/alphapulse/api/settings.py

@@ -170,10 +170,20 @@ async def delete_provider(provider_id: int, session: Annotated[AsyncSession, Dep
     return success(None)
 
 
+@router.get("/provider-availability")
+async def provider_availability(session: Annotated[AsyncSession, Depends(get_session)]):
+    providers = list(await session.scalars(select(AiProvider).order_by(AiProvider.id)))
+    try:
+        return success([{'provider_id': p.id, 'name': p.name, **await llm.availability(p)} for p in providers])
+    except llm.LlmError as error:
+        raise HTTPException(503, str(error)) from error
+
+
 @router.post("/providers/{provider_id}/test")
 async def test_provider(provider_id: int, session: Annotated[AsyncSession, Depends(get_session)]) -> ApiResponse[TestResult]:
     provider = await get_provider(session, provider_id)
     started = time.monotonic()
+    token = llm.fail_fast.set(True)
     try:
         completion = None
         async for event in llm.stream(provider, "只回复 OK。", [{"role": "user", "text": "ping"}], [], provider.reasoning_effort):
@@ -181,6 +191,8 @@ async def test_provider(provider_id: int, session: Annotated[AsyncSession, Depen
                 completion = event
     except llm.LlmError as e:
         return success(TestResult(ok=False, elapsed_ms=round((time.monotonic() - started) * 1000), error=str(e)))
+    finally:
+        llm.fail_fast.reset(token)
     return success(TestResult(ok=True, elapsed_ms=round((time.monotonic() - started) * 1000), reply=completion.text[:200]))
 
 

backend/src/alphapulse/news_research/api.py

@@ -197,6 +197,8 @@ async def enable(session: DB, body: EnableInput | None = None):
         "summary": "",
         "content": "本条为虚构测试。示例公司宣布计划维护，具体时间未知。",
     }
+    from alphapulse.ai import llm
+    token = llm.fail_fast.set(True)
     original_config = row.config
     control_revision = row.control_revision
     try:
@@ -261,6 +263,8 @@ async def enable(session: DB, body: EnableInput | None = None):
         raise HTTPException(422, str(error)) from error
     except Exception as error:
         raise HTTPException(502, f"两阶段验证失败：{error}") from error
+    finally:
+        llm.fail_fast.reset(token)
     row = await session.get(
         ResearchSettings, 1, populate_existing=True, with_for_update=True
     )

backend/src/alphapulse/news_research/model.py

@@ -23,12 +23,9 @@ class SubmissionFailure(ValueError):
 
 async def model_stream(*args):
     """Scope transport retry changes to news requests only."""
-    token = llm.news_retry.set({"output": False})
-    try:
-        async for item in llm.stream(*args):
-            yield item
-    finally:
-        llm.news_retry.reset(token)
+    async for item in llm.stream(*args):
+        yield item
+
 
 async def validate_submission(provider, system, messages, schema, completion, usages, allow_repair=True):
     """Repair structural or time-contract errors once using the same evidence."""

backend/src/alphapulse/news_research/scheduling.py

@@ -166,5 +166,6 @@ async def queue_status(session, settings):
                 reasons.append('no_work' if not any(
                     q.get(s, 0) for q in queues.values() for s in ('missing', 'stale', 'index_pending', 'assign_pending')
                 ) and not any(q.get('ready', 0) for q in events.values()) else 'dispatching')
-    return {'running': running, 'capacity': capacity, 'running_by_stage': stages,
+    from alphapulse.ai.recovery import waiting_count
+    return {'waiting_retry': await waiting_count(), 'running': running, 'capacity': capacity, 'running_by_stage': stages,
             'draining': not settings.enabled and running > 0, 'events': events, 'idle_reasons': reasons, **queues}

backend/src/alphapulse/trading/analyst.py

@@ -96,6 +96,7 @@ class AnalysisError(Exception):
 @dataclass
 class Prepared:
     exchange: str
+    symbol: str
     provider: AiProvider
     effort: str | None
     dry_run: bool
@@ -248,13 +249,24 @@ async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared
         "## 预取的行情（实盘数据，已取，不要用相同参数重复调用）",
         await prefetch(symbol),
     ]
-    return Prepared(exchange=exchange, provider=provider, effort=effort, dry_run=dry_run, system=system, user_text="\n".join(lines), context=context)
+    return Prepared(exchange=exchange, symbol=symbol, provider=provider, effort=effort, dry_run=dry_run, system=system, user_text="\n".join(lines), context=context)
 
 
 async def complete(prepared: Prepared, messages: list[dict], tools: list[llm.ToolSpec]) -> llm.Completion:
-    async for item in llm.stream(prepared.provider, prepared.system, messages, tools, prepared.effort):
+    async def refresh():
+        # 只更新当前请求的行情，不重新执行此前的工具或交易操作。
+        latest = "重试前最新行情（此前行情仅为历史快照）：\n" + await prefetch(prepared.symbol)
+        if messages[-1].get('retry_snapshot'):
+            messages.pop()
+        messages.append({'role': 'user', 'text': latest, 'retry_snapshot': True})
+        if estimate_tokens(prepared.system) + estimate_tokens(dump(messages)) + prepared.provider.max_tokens > prepared.provider.context_window:
+            raise AnalysisError('刷新行情后超过上下文上限，未重发')
+    completion = None
+    async for item in llm.stream(prepared.provider, prepared.system, messages, tools, prepared.effort, before_retry=refresh):
         if isinstance(item, llm.Completion):
-            return item
+            completion = item
+    if completion is not None:
+        return completion
     raise AssertionError("llm.stream 没有产出 Completion")
 
 

backend/tests/conftest.py

@@ -11,4 +11,13 @@ def trading(monkeypatch):
     }.items():
         monkeypatch.setenv(key, value)
     from alphapulse.trading import executor
+    from unittest.mock import AsyncMock
+    from alphapulse.ai import llm, recovery
+    class HealthyGate:
+        async def call(self, operation, request_id='', epoch=-1):
+            return {'epoch': 0, 'level': 0, 'retry_at': 0, 'probe': False, 'admitted': True, 'now': 0}
+    monkeypatch.setattr(llm, 'recovery_gate', lambda _: HealthyGate())
+    monkeypatch.setattr(recovery, 'mark_waiting', AsyncMock())
+    monkeypatch.setattr(recovery, 'clear_waiting', AsyncMock())
+    monkeypatch.setattr(recovery, 'waiting_count', AsyncMock(return_value=0))
     return executor

backend/tests/test_ai_recovery.py

@@ -0,0 +1,160 @@
+"""隔离 Redis 验证实际 Lua 原子操作；不访问生产接入。"""
+import asyncio
+from types import SimpleNamespace as N
+from unittest.mock import AsyncMock
+
+import httpx
+import openai
+import pytest
+
+
+
+def test_global_failure_takes_over_local_retry_and_recovers(trading, monkeypatch):
+    from alphapulse.ai import llm
+    states = iter([
+        dict(level=0, epoch=0, admitted=True, probe=False, now=0, retry_at=0),
+        dict(level=1, epoch=1, admitted=False, probe=False, now=0, retry_at=180),
+        dict(level=1, epoch=1, admitted=False, probe=False, now=0, retry_at=180),
+        dict(level=1, epoch=1, admitted=True, probe=True, now=180, retry_at=180),
+    ])
+    async def coordinate(operation, *args):
+        if operation in ('enter', 'failure'):
+            return next(states)
+        return dict(level=0)
+    gate = N(call=AsyncMock(side_effect=coordinate))
+    monkeypatch.setattr(llm, 'recovery_gate', lambda _: gate)
+    monkeypatch.setattr(llm.asyncio, 'sleep', AsyncMock())
+    local = AsyncMock()
+    monkeypatch.setattr(llm, '_wait_local', local)
+    calls = []
+    async def adapter(*args):
+        calls.append(args)
+        if len(calls) == 1:
+            raise openai.RateLimitError('limited', response=httpx.Response(429, request=httpx.Request('POST', 'https://test.invalid')), body={})
+        yield llm.Completion(text='done')
+    monkeypatch.setattr(llm, '_stream_openai', adapter)
+    async def run():
+        return [x async for x in llm.stream(N(protocol='openai'), '', [], [])]
+    result = asyncio.run(run())
+    assert result[-1].text == 'done' and len(calls) == 2
+    assert any(isinstance(x, llm.RetryStatus) and x.scope == 'upstream' for x in result)
+    local.assert_not_awaited()
+
+
+def test_coordinator_failure_does_not_send(trading, monkeypatch):
+    from alphapulse.ai import llm
+    adapter = AsyncMock()
+    monkeypatch.setattr(llm, '_stream_openai', adapter)
+    monkeypatch.setattr(llm, 'recovery_gate', lambda _: N(call=AsyncMock(side_effect=RuntimeError('offline'))))
+    async def run():
+        with pytest.raises(llm.LlmError, match='协调不可用'):
+            async for _ in llm.stream(N(protocol='openai'), '', [], []):
+                pass
+    asyncio.run(run())
+    adapter.assert_not_called()
+
+
+def test_fail_fast_does_not_send_or_wait(trading, monkeypatch):
+    from alphapulse.ai import llm
+    gate = N(call=AsyncMock(return_value=dict(level=2, epoch=1, admitted=False, probe=False, now=100, retry_at=700)))
+    monkeypatch.setattr(llm, 'recovery_gate', lambda _: gate)
+    adapter = AsyncMock()
+    monkeypatch.setattr(llm, '_stream_openai', adapter)
+    async def run():
+        token = llm.fail_fast.set(True)
+        try:
+            with pytest.raises(llm.LlmError, match='下次尝试'):
+                async for _ in llm.stream(N(protocol='openai'), '', [], []):
+                    pass
+        finally:
+            llm.fail_fast.reset(token)
+    asyncio.run(run())
+    adapter.assert_not_called()
+
+
+def test_completed_probe_is_confirmed_before_caller_returns(trading, monkeypatch):
+    from alphapulse.ai import llm
+    gate = N(call=AsyncMock(return_value=dict(level=0, epoch=1, admitted=True, probe=True, now=1, retry_at=0)))
+    monkeypatch.setattr(llm, 'recovery_gate', lambda _: gate)
+    async def adapter(*args):
+        yield llm.Completion(text='done')
+    monkeypatch.setattr(llm, '_stream_openai', adapter)
+    async def run():
+        stream = llm.stream(N(protocol='openai'), '', [], [])
+        assert (await anext(stream)).text == 'done'
+        assert any(call.args[0] == 'success' for call in gate.call.await_args_list)
+        await stream.aclose()
+    asyncio.run(run())
+
+
+def test_retry_refreshes_only_current_request(trading, monkeypatch):
+    from alphapulse.ai import llm
+    calls = []
+    async def adapter(provider, system, messages, tools, effort):
+        calls.append(list(messages))
+        if len(calls) == 1:
+            raise httpx.ConnectError('unavailable')
+        yield llm.Completion(text='ok')
+    messages = [{'role': 'tool', 'tool_call_id': 'already-done', 'result': 'retained'}]
+    async def refresh():
+        messages.append({'role': 'user', 'text': 'fresh snapshot'})
+    monkeypatch.setattr(llm, '_stream_openai', adapter)
+    monkeypatch.setattr(llm, '_wait_local', AsyncMock())
+    async def run():
+        return [x async for x in llm.stream(N(protocol='openai'), '', messages, [], before_retry=refresh)]
+    assert asyncio.run(run())[-1].text == 'ok'
+    assert len(calls) == 2 and calls[1][0] == calls[0][0]
+    assert calls[1][-1]['text'] == 'fresh snapshot'
+
+
+@pytest.mark.parametrize('failed_operation', ['success', 'release', 'both'])
+def test_completed_result_survives_coordination_write_failure(trading, monkeypatch, failed_operation):
+    from alphapulse.ai import llm
+    from alphapulse.news_research.model import complete
+    from alphapulse.news_research.contracts import IndexResult
+    calls = []
+
+    async def coordinate(operation, *args):
+        calls.append(operation)
+        if operation == failed_operation or failed_operation == 'both' and operation in ('success', 'release'):
+            raise RuntimeError('synthetic Redis outage')
+        return dict(level=0, epoch=0, admitted=True, probe=False, now=0, retry_at=0)
+
+    async def adapter(*args):
+        yield llm.Completion(tool_calls=[{'name': 'submit', 'id': '1', 'arguments': {
+            'keywords': [{'canonical': '新闻', 'aliases': ['news'], 'quote': 'news'}],
+        }}], usage={'input_tokens': 100, 'output_tokens': 10})
+
+    monkeypatch.setattr(llm, 'recovery_gate', lambda _: N(call=coordinate))
+    monkeypatch.setattr(llm, '_stream_openai', adapter)
+    async def run():
+        return await complete(N(protocol='openai', max_tokens=100, context_window=250000,
+                                reasoning_effort=None), '', {}, IndexResult)
+    output, usage = asyncio.run(run())
+    assert output.keywords[0].canonical == '新闻'
+    assert usage == {'input_tokens': 100, 'output_tokens': 10}
+    assert calls.count('enter') == 1 and 'release' in calls
+
+
+def test_cancellation_survives_lease_release_failure(trading, monkeypatch):
+    from alphapulse.ai import llm
+    started = asyncio.Event()
+    async def coordinate(operation, *args):
+        if operation == 'release':
+            raise RuntimeError('synthetic Redis outage')
+        return dict(level=0, epoch=0, admitted=True, probe=False, now=0, retry_at=0)
+    async def adapter(*args):
+        started.set()
+        await asyncio.Event().wait()
+        yield
+    monkeypatch.setattr(llm, 'recovery_gate', lambda _: N(call=coordinate))
+    monkeypatch.setattr(llm, '_stream_openai', adapter)
+    async def run():
+        async def consume():
+            return [x async for x in llm.stream(N(protocol='openai'), '', [], [])]
+        task = asyncio.create_task(consume())
+        await started.wait()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+    asyncio.run(run())

backend/tests/test_failure_paths.py

@@ -58,7 +58,7 @@ def test_model_requires_protocol_end_marker(trading, monkeypatch, protocol, comp
     asyncio.run(check())
 
 
-def test_connection_failure_uses_only_sdk_retry_budget(trading, monkeypatch):
+def test_connection_failure_has_one_retry_owner(trading, monkeypatch):
     from alphapulse.ai import llm
     attempts = 0
     def handle(request):
@@ -66,12 +66,14 @@ def test_connection_failure_uses_only_sdk_retry_budget(trading, monkeypatch):
         attempts += 1
         raise httpx.ConnectError("测试连接失败", request=request)
     provider = install_model_transport(llm, monkeypatch, "openai", handle)
+    from unittest.mock import AsyncMock
+    monkeypatch.setattr(llm, "_wait_local", AsyncMock())
     async def check():
         with pytest.raises(llm.LlmError, match="无法连接"):
             async for _ in llm.stream(provider, "system", [], []):
                 pass
     asyncio.run(check())
-    assert attempts == llm.MAX_RETRIES + 1
+    assert attempts == 3
 
 
 def test_pipeline_error_keeps_history_checkpoint_and_marks_run_failed(trading, monkeypatch):

backend/tests/test_news_transport.py

@@ -17,7 +17,7 @@ def test_news_transport_retry_budget_and_output_guard(trading, monkeypatch, mode
         calls.append(args)
         if mode == 'partial': yield llm.TextDelta('started')
         if mode == 'tool_fragment': llm._news_output()
-        if mode == 'success' and len(calls) == 3:
+        if mode in ('success', 'long_wait') and len(calls) == 3:
             yield llm.Completion(usage={'input_tokens': 1})
             return
         request = httpx.Request('POST', 'https://example.invalid/v1/chat/completions')
@@ -27,31 +27,22 @@ def test_news_transport_retry_budget_and_output_guard(trading, monkeypatch, mode
         raise openai.APIError('Concurrency limit exceeded for account', request, body={'error': {'message': 'Concurrency limit exceeded for account'}})
     monkeypatch.setattr(llm, '_stream_openai', adapter)
     sleep = AsyncMock()
-    monkeypatch.setattr(llm.asyncio, 'sleep', sleep)
+    monkeypatch.setattr(llm, '_wait_local', sleep)
     record = {}
     async def run():
         token = current.set(record)
         try: return [item async for item in model_stream(N(protocol='openai'), '', [], [], 'max')]
         finally: current.reset(token)
-    if mode == 'success': assert asyncio.run(run())[-1].usage == {'input_tokens': 1}
+    if mode in ('success', 'long_wait'): assert asyncio.run(run())[-1].usage == {'input_tokens': 1}
     else:
         with pytest.raises(llm.LlmError): asyncio.run(run())
-    retries = mode in ('success', 'exhausted')
+    retries = mode in ('success', 'exhausted', 'long_wait')
     assert len(calls) == (3 if retries else 1)
-    assert [call.args[0] for call in sleep.await_args_list] == ([30, 60] if retries else [])
+    assert [call.args[0] for call in sleep.await_args_list] == ([60, 180] if retries else [])
     assert llm.news_retry.get() is None
-    if retries: assert '2/2' in record['phase']
+    if retries: assert record['retry_attempt'] == 2
 
 
-def test_retry_after_seconds_and_dates(trading):
-    from alphapulse.ai import llm
-    from datetime import UTC, datetime, timedelta
-    from email.utils import format_datetime
-    request = httpx.Request('POST', 'https://example.invalid')
-    for value in ('90', format_datetime(datetime.now(UTC) + timedelta(seconds=95))):
-        error = openai.RateLimitError('wait', response=httpx.Response(429, headers={'retry-after': value}, request=request), body={})
-        assert 89 <= llm._news_retry_delay(error, 0) <= 96
-
 
 def test_cancelling_backoff_releases_context_and_progress(trading, monkeypatch):
     from alphapulse.ai import llm
@@ -61,11 +52,11 @@ def test_cancelling_backoff_releases_context_and_progress(trading, monkeypatch):
     async def adapter(*args):
         raise openai.APIError('Concurrency limit exceeded', httpx.Request('POST', 'https://example.invalid'), body={})
         yield
-    async def sleep(delay):
+    async def sleep(delay, coordinate):
         entered.set()
         await asyncio.Event().wait()
     monkeypatch.setattr(llm, '_stream_openai', adapter)
-    monkeypatch.setattr(llm.asyncio, 'sleep', sleep)
+    monkeypatch.setattr(llm, '_wait_local', sleep)
     async def run():
         async def consume():
             try:
@@ -82,7 +73,7 @@ def test_cancelling_backoff_releases_context_and_progress(trading, monkeypatch):
 
 @pytest.mark.parametrize('protocol', ['openai', 'anthropic'])
 @pytest.mark.parametrize('news', [True, False])
-def test_sdk_retry_is_disabled_only_for_news(trading, monkeypatch, protocol, news):
+def test_sdk_retry_is_disabled_for_every_request(trading, monkeypatch, protocol, news):
     from alphapulse.ai import llm
     options=[]
     class Client:
@@ -107,4 +98,4 @@ def test_sdk_retry_is_disabled_only_for_news(trading, monkeypatch, protocol, new
         try: return [item async for item in llm.stream(provider,'rules',[],[],None)]
         finally: llm.news_retry.reset(token)
     assert asyncio.run(run())[-1] == llm.Completion()
-    assert options[0]['max_retries'] == (0 if news else llm.MAX_RETRIES)
+    assert options[0]['max_retries'] == 0