fix(ops): chat messages 500 when artifact write hits circuit breaker

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,

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,

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,21 +76,30 @@ 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
 
     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

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]]:

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
 ```
 

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
 # ---------------------------------------------------------------------------