From 9eaea0e66d39408def379dd21457fd4e528593cf Mon Sep 17 00:00:00 2001 From: cyning Date: Wed, 5 Aug 2026 11:41:45 +0800 Subject: [PATCH] fix(ops): prevent chat 500 when artifact writes hit Supabase circuit breaker Swallow failure-event write errors and make react/deep post-answer store writes best-effort so /ops/chat/messages still returns the answer. Co-authored-by: Cursor --- api/ops/orchestrator/core.py | 64 +++++++++++++++++++--------------- api/ops/react_loop.py | 59 ++++++++++++++++--------------- api/ops/store/artifacts.py | 55 +++++++++++++++++++++++------ api/ops/store/runs.py | 23 ++++++++---- docs/_tech_graph/02_version.md | 1 + tests/ops/test_artifacts.py | 37 ++++++++++++++++++++ 6 files changed, 165 insertions(+), 74 deletions(-) diff --git a/api/ops/orchestrator/core.py b/api/ops/orchestrator/core.py index b647f4af..f8bc9033 100644 --- a/api/ops/orchestrator/core.py +++ b/api/ops/orchestrator/core.py @@ -2,9 +2,12 @@ from __future__ import annotations +import logging import re from typing import Any +logger = logging.getLogger(__name__) + from api.ops import intent_router as _intent_router from api.ops.agents.graph_analyst import analyze_graph from api.ops.agents.issue_analyst import analyze_issue @@ -378,35 +381,38 @@ def run_deep( "verdict": final_verdict, }, ) - save_artifact_with_failure_event( - run_id, - "deep.final_answer", - { - "answer": answer, - "agent": agent_name, - "issue_number": analyst_result.get("issue_number"), - "verdict": final_verdict, - "intent": intent or "issue_contribution", - "route": "deep", - }, - store=store, - ) - store.append_event(run_id, "orchestrator", "run.end", node_id="deep.end") - # B5: run 级 metrics_json 汇总 - metrics_json = _build_metrics_json( - route="deep", - intent=intent or "issue_contribution", - llm_calls=llm_calls, - llm_usages=llm_usages, - ) - store.update_run_metrics_json(run_id, metrics_json) - store.append_event( - run_id, - "orchestrator", - "run.metrics", - payload=metrics_json, - node_id="deep.metrics", - ) + # 答案已落 update_run；后续 artifact / metrics 写失败不得拖垮 chat 500 + try: + save_artifact_with_failure_event( + run_id, + "deep.final_answer", + { + "answer": answer, + "agent": agent_name, + "issue_number": analyst_result.get("issue_number"), + "verdict": final_verdict, + "intent": intent or "issue_contribution", + "route": "deep", + }, + store=store, + ) + store.append_event(run_id, "orchestrator", "run.end", node_id="deep.end") + metrics_json = _build_metrics_json( + route="deep", + intent=intent or "issue_contribution", + llm_calls=llm_calls, + llm_usages=llm_usages, + ) + store.update_run_metrics_json(run_id, metrics_json) + store.append_event( + run_id, + "orchestrator", + "run.metrics", + payload=metrics_json, + node_id="deep.metrics", + ) + except Exception as exc: # noqa: BLE001 — 收尾旁路；答案已可返回 + logger.warning("deep post-answer store writes failed run_id=%s: %s", run_id, exc) return { "run_id": run_id, "status": final_verdict, diff --git a/api/ops/react_loop.py b/api/ops/react_loop.py index 9d99dca2..980c630c 100644 --- a/api/ops/react_loop.py +++ b/api/ops/react_loop.py @@ -492,35 +492,38 @@ def run_react_fallback( status=final_verdict, final_answer={"answer": answer, "verdict": final_verdict}, ) - save_artifact_with_failure_event( - run_id, - "react.final_answer", - { - "answer": answer, - "verdict": final_verdict, - "intent": "fallback", - "route": "react", - "steps_taken": step, - }, - store=store, - ) - store.append_event(run_id, "orchestrator", "run.end", node_id="react.end") + # 答案已落 update_run；后续 artifact / metrics 写失败不得拖垮 chat 500 + try: + save_artifact_with_failure_event( + run_id, + "react.final_answer", + { + "answer": answer, + "verdict": final_verdict, + "intent": "fallback", + "route": "react", + "steps_taken": step, + }, + store=store, + ) + store.append_event(run_id, "orchestrator", "run.end", node_id="react.end") - # Run metrics - metrics_json = _build_metrics_json( - route="react", - intent="fallback", - llm_calls=llm_calls, - llm_usages=llm_usages, - ) - store.update_run_metrics_json(run_id, metrics_json) - store.append_event( - run_id, - "orchestrator", - "run.metrics", - payload=metrics_json, - node_id="react.metrics", - ) + metrics_json = _build_metrics_json( + route="react", + intent="fallback", + llm_calls=llm_calls, + llm_usages=llm_usages, + ) + store.update_run_metrics_json(run_id, metrics_json) + store.append_event( + run_id, + "orchestrator", + "run.metrics", + payload=metrics_json, + node_id="react.metrics", + ) + except Exception as exc: # noqa: BLE001 — 收尾旁路；答案已可返回 + logger.warning("react post-answer store writes failed run_id=%s: %s", run_id, exc) return { "run_id": run_id, diff --git a/api/ops/store/artifacts.py b/api/ops/store/artifacts.py index 03d24068..2c2ebdf5 100644 --- a/api/ops/store/artifacts.py +++ b/api/ops/store/artifacts.py @@ -2,13 +2,33 @@ from __future__ import annotations +import logging from typing import Any +logger = logging.getLogger(__name__) + class ArtifactStoreError(RuntimeError): """Artifact 写入失败时的自说明异常。""" +def _is_non_retryable_store_error(exc: BaseException) -> bool: + """熔断打开或表缺失：再试只会放大失败，应立即上抛为 ArtifactStoreError。""" + from api.chatbi_circuit_breaker import CircuitBreakerOpenError + + if isinstance(exc, CircuitBreakerOpenError): + return True + # PostgREST：关系不存在（未跑 ops_desk_p1_artifacts.sql） + if getattr(exc, "code", None) == "PGRST205": + return True + if "PGRST205" in str(exc): + return True + cause = getattr(exc, "__cause__", None) + if cause is not None and cause is not exc and _is_non_retryable_store_error(cause): + return True + return False + + def save_artifact( run_id: str, kind: str, @@ -20,6 +40,7 @@ def save_artifact( - 未提供 store 时，使用全局 supabase_client() 构造 OpsRunStore。 - 重试 max_retries + 1 次后仍失败则抛出 ArtifactStoreError。 + - CircuitBreakerOpenError / PGRST205 不重试。 """ from api.ops.store.runs import OpsRunStore from api.rag_env import supabase_client @@ -33,6 +54,11 @@ def save_artifact( return target.save_artifact(run_id, kind, normalized) except Exception as exc: last_exc = exc + if _is_non_retryable_store_error(exc): + raise ArtifactStoreError( + f"Failed to save artifact run_id={run_id} kind={kind} " + f"after {_attempt + 1} attempts: {exc}" + ) from exc raise ArtifactStoreError( f"Failed to save artifact run_id={run_id} kind={kind} " @@ -50,6 +76,7 @@ def save_artifact_with_failure_event( """保存 artifact；写入失败时记录 `artifact.write_failed` 事件并吞掉异常。 返回值：成功返回写入行；失败返回 None。 + 失败事件本身写库失败时也吞掉，避免拖垮 /ops/chat/messages。 """ from api.ops.events_schema import SCHEMA_VERSION from api.ops.store.runs import append_event @@ -57,14 +84,22 @@ def save_artifact_with_failure_event( try: return save_artifact(run_id, kind, payload, store=store, max_retries=max_retries) except ArtifactStoreError as exc: - append_event( - run_id, - "artifact.write_failed", - { - "kind": kind, - "error": str(exc), - "schema_version": SCHEMA_VERSION, - }, - store=store, - ) + try: + append_event( + run_id, + "artifact.write_failed", + { + "kind": kind, + "error": str(exc), + "schema_version": SCHEMA_VERSION, + }, + store=store, + ) + except Exception as event_exc: # noqa: BLE001 — best-effort 旁路写 + logger.warning( + "artifact.write_failed event skipped run_id=%s kind=%s: %s", + run_id, + kind, + event_exc, + ) return None diff --git a/api/ops/store/runs.py b/api/ops/store/runs.py index 7bf9f378..30966703 100644 --- a/api/ops/store/runs.py +++ b/api/ops/store/runs.py @@ -260,18 +260,27 @@ def save_artifact( """幂等写入 ops_run_artifacts；由 (run_id, kind) 唯一去重。""" row = {"run_id": run_id, "kind": kind, "payload": payload} - def _once() -> dict[str, Any]: - res = ( - self.client.table("ops_run_artifacts") - .upsert(row, on_conflict="run_id,kind") - .execute() - ) + def _once() -> dict[str, Any] | None: + try: + res = ( + self.client.table("ops_run_artifacts") + .upsert(row, on_conflict="run_id,kind") + .execute() + ) + except APIError as exc: + # 表未迁移：在 breaker 回调内吞掉，避免缺表把 supabase 熔断打开 + if getattr(exc, "code", None) == "PGRST205": + return None + raise data = res.data if isinstance(res.data, list) else [] if data and isinstance(data[0], dict): return data[0] raise RuntimeError("ops_run_artifacts upsert did not return row") - return supabase_execute_with_retry(_once) + result = supabase_execute_with_retry(_once) + if result is None: + raise RuntimeError("ops_run_artifacts table missing (PGRST205)") + return result def list_artifacts(self, run_id: str) -> list[dict[str, Any]]: def _once() -> list[dict[str, Any]]: diff --git a/docs/_tech_graph/02_version.md b/docs/_tech_graph/02_version.md index 75479a4e..5b691de2 100644 --- a/docs/_tech_graph/02_version.md +++ b/docs/_tech_graph/02_version.md @@ -57,5 +57,6 @@ timeline 2026-07-08 : 4bf5782c auto: api/ops/orchestrator/__init__.py 2026-07-09 : db09fd40 auto: api/ops/events_schema.py 2026-07-10 : 13553deb auto: api/ops/intent_router.py + 2026-08-05 : 239eca73 auto: api/ops/orchestrator/core.py ``` diff --git a/tests/ops/test_artifacts.py b/tests/ops/test_artifacts.py index 448d7017..b8193236 100644 --- a/tests/ops/test_artifacts.py +++ b/tests/ops/test_artifacts.py @@ -198,6 +198,43 @@ def test_save_artifact_failure_records_write_failed_event(store: FakeArtifactSto assert "error" in fail_events[0]["payload"] +def test_save_artifact_failure_event_write_also_fails_returns_none( + store: FakeArtifactStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """失败事件本身写库失败时仍返回 None，不得上抛拖垮 chat。""" + from api.ops.store import artifacts as artifacts_mod + + store._fail_artifacts = True + + def _boom(*_a: Any, **_k: Any) -> dict[str, Any]: + raise RuntimeError("circuit_breaker_open:supabase") + + monkeypatch.setattr("api.ops.store.runs.append_event", _boom) + + result = artifacts_mod.save_artifact_with_failure_event( + "run-fail", "react.final_answer", {"answer": "x"}, store=store + ) + assert result is None + + +def test_save_artifact_circuit_open_fails_fast_no_retry( + store: FakeArtifactStore, monkeypatch: pytest.MonkeyPatch +) -> None: + from api.chatbi_circuit_breaker import CircuitBreakerOpenError, CircuitState + from api.ops.store.artifacts import ArtifactStoreError, save_artifact + + calls = {"n": 0} + + def _open(*_a: Any, **_k: Any) -> dict[str, Any]: + calls["n"] += 1 + raise CircuitBreakerOpenError(breaker_name="supabase", state=CircuitState.OPEN) + + store.save_artifact = _open # type: ignore[method-assign] + with pytest.raises(ArtifactStoreError, match="circuit_breaker_open"): + save_artifact("run-cb", "react.final_answer", {"answer": "x"}, store=store, max_retries=3) + assert calls["n"] == 1 + + # --------------------------------------------------------------------------- # deep / ReAct 路径调用 save_artifact # ---------------------------------------------------------------------------