feat(ops): P1-2 ReAct checkpoint save/load/resume

api/ops/react_loop.py

@@ -3,9 +3,12 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 from typing import Any
 
+logger = logging.getLogger(__name__)
+
 from api.ops.chat_context import load_chat_transcript
 from api.ops.events_schema import handoff_payload, review_payload
 from api.ops.llm import chat_completion
@@ -15,13 +18,155 @@
 from api.ops.react_tools import _build_v0_registry, _truncate_summary
 from api.ops.review.rules import review_result
 from api.ops.store.artifacts import save_artifact_with_failure_event
+from api.ops.store.checkpoints import (
+    CheckpointStoreError,
+    find_latest_checkpoint_for_session,
+    save_checkpoint,
+)
 from api.ops.store.runs import OpsRunStore, append_event
 from api.ops.tracing import trace_span, traceable, update_current_span_metadata
 
 MAX_STEPS_DEFAULT = int(os.getenv("OPS_REACT_MAX_STEPS", "6"))
 MAX_RETRIES_DEFAULT = 2
 
 
+def _build_react_state(
+    query: str,
+    session_id: str | None,
+    messages: list[dict[str, str]],
+    step: int,
+    tool_evidence: list[dict[str, Any]],
+    final_answer: str,
+    final_verdict: str,
+    llm_calls: int,
+    llm_usages: list[LlmUsage],
+) -> dict[str, Any]:
+    """构造可序列化的 ReAct checkpoint 状态。"""
+    return {
+        "route": "react",
+        "query": query,
+        "session_id": session_id,
+        "step": step,
+        "messages": list(messages),
+        "tool_evidence": list(tool_evidence),
+        "final_answer": final_answer,
+        "final_verdict": final_verdict,
+        "llm_calls": llm_calls,
+        "llm_usages": [u.to_dict() for u in llm_usages],
+    }
+
+
+def _try_save_checkpoint(
+    run_id: str,
+    session_id: str,
+    query: str,
+    messages: list[dict[str, str]],
+    step: int,
+    tool_evidence: list[dict[str, Any]],
+    final_answer: str,
+    final_verdict: str,
+    llm_calls: int,
+    llm_usages: list[LlmUsage],
+    store: OpsRunStore,
+) -> None:
+    """保存 checkpoint；失败时记录事件但不中断 ReAct 循环。"""
+    state = _build_react_state(
+        query=query,
+        session_id=session_id,
+        messages=messages,
+        step=step,
+        tool_evidence=tool_evidence,
+        final_answer=final_answer,
+        final_verdict=final_verdict,
+        llm_calls=llm_calls,
+        llm_usages=llm_usages,
+    )
+    try:
+        save_checkpoint(run_id, session_id, state, store=store)
+    except Exception as exc:  # pragma: no cover - 防御性降级
+        logger.warning("checkpoint.save_failed: %s", exc)
+        store.append_event(
+            run_id,
+            "orchestrator",
+            "checkpoint.save_failed",
+            payload={"error": str(exc), "session_id": session_id},
+            node_id="react.checkpoint.save_failed",
+        )
+
+
+def _resume_react_state(
+    run_id: str,
+    query: str,
+    session_id: str,
+    cp_row: dict[str, Any],
+    store: OpsRunStore,
+) -> dict[str, Any] | None:
+    """尝试从 checkpoint 行恢复 ReAct 状态。
+
+    成功返回状态字典；失败时记录 checkpoint.corrupted 并返回 None。
+    """
+    try:
+        state_json = cp_row.get("state_json")
+        state = _validate_react_checkpoint(state_json)
+    except CheckpointStoreError as exc:
+        logger.warning("checkpoint.corrupted: %s", exc)
+        store.append_event(
+            run_id,
+            "orchestrator",
+            "checkpoint.corrupted",
+            payload={
+                "error": str(exc),
+                "session_id": session_id,
+                "from_run_id": str(cp_row.get("run_id", "")),
+            },
+            node_id="react.checkpoint.corrupted",
+        )
+        return None
+
+    prev_run_id = str(cp_row.get("run_id", ""))
+    store.append_event(
+        run_id,
+        "orchestrator",
+        "checkpoint.resume",
+        payload={
+            "from_run_id": prev_run_id,
+            "step": state["step"],
+            "session_id": session_id,
+        },
+        node_id="react.checkpoint.resume",
+    )
+
+    messages: list[dict[str, str]] = list(state.get("messages", []))
+    if state.get("query") != query:
+        messages.append({"role": "user", "content": query})
+
+    return {
+        "messages": messages,
+        "step": int(state.get("step", 0)),
+        "tool_evidence": list(state.get("tool_evidence", [])),
+        "final_answer": str(state.get("final_answer", "")),
+        "final_verdict": str(state.get("final_verdict", "partial")),
+        "llm_calls": int(state.get("llm_calls", 0)),
+        "llm_usages": [LlmUsage.from_dict(u) for u in state.get("llm_usages", [])],
+    }
+
+
+def _validate_react_checkpoint(state_json: Any) -> dict[str, Any]:
+    """校验 checkpoint 状态；失败抛出 CheckpointStoreError。"""
+    if not isinstance(state_json, dict):
+        raise CheckpointStoreError("checkpoint state_json is not a dict")
+    for key in ("route", "query", "step", "messages", "tool_evidence"):
+        if key not in state_json:
+            raise CheckpointStoreError(f"checkpoint state missing key: {key}")
+    if state_json.get("route") != "react":
+        raise CheckpointStoreError("checkpoint route is not 'react'")
+    if not isinstance(state_json["messages"], list):
+        raise CheckpointStoreError("checkpoint state messages is not a list")
+    if not isinstance(state_json["step"], int):
+        raise CheckpointStoreError("checkpoint state step is not an int")
+    return state_json
+
+
 @traceable(capture_input=False, capture_output=False)
 def run_react_fallback(
     run_id: str,
@@ -37,8 +182,6 @@ def run_react_fallback(
     与 FSM 路径共用 ops_runs / ops_run_events / Review 闸。
     超限 → status partial + 仍 synthesize（非 500）。
     """
-    transcript = load_chat_transcript(session_id, store=store)
-
     update_current_span_metadata(
         {
             "ops_run_id": run_id,
@@ -85,21 +228,38 @@ def run_react_fallback(
             store=store,
         )
 
-    # System prompt for ReAct
-    system_prompt = _build_react_system_prompt(tools_json)
-    messages: list[dict[str, str]] = [
-        {"role": "system", "content": system_prompt},
-    ]
-    if transcript:
-        messages.extend(transcript)
-    messages.append({"role": "user", "content": query})
-
-    step = 0
-    final_answer = ""
-    llm_calls = 0
-    llm_usages: list[LlmUsage] = []
-    tool_evidence: list[dict[str, Any]] = []
-    final_verdict = "partial"
+    # Try to resume from a previous checkpoint for this session
+    resumed_state: dict[str, Any] | None = None
+    if session_id:
+        cp_row = find_latest_checkpoint_for_session(session_id, store=store)
+        if cp_row:
+            resumed_state = _resume_react_state(run_id, query, session_id, cp_row, store)
+
+    if resumed_state is None:
+        # Cold start
+        transcript = load_chat_transcript(session_id, store=store)
+        system_prompt = _build_react_system_prompt(tools_json)
+        messages: list[dict[str, str]] = [
+            {"role": "system", "content": system_prompt},
+        ]
+        if transcript:
+            messages.extend(transcript)
+        messages.append({"role": "user", "content": query})
+
+        step = 0
+        final_answer = ""
+        llm_calls = 0
+        llm_usages: list[LlmUsage] = []
+        tool_evidence: list[dict[str, Any]] = []
+        final_verdict = "partial"
+    else:
+        messages = resumed_state["messages"]
+        step = resumed_state["step"]
+        final_answer = resumed_state["final_answer"]
+        llm_calls = resumed_state["llm_calls"]
+        llm_usages = resumed_state["llm_usages"]
+        tool_evidence = resumed_state["tool_evidence"]
+        final_verdict = resumed_state["final_verdict"]
 
     while step < max_steps:
         step += 1
@@ -190,6 +350,22 @@ def run_react_fallback(
         messages.append({"role": "assistant", "content": raw_content})
         messages.append({"role": "user", "content": f"Tool result:\n{tool_msg}"})
 
+        # Save checkpoint after each non-final step so that crashes can resume
+        if session_id:
+            _try_save_checkpoint(
+                run_id,
+                session_id,
+                query,
+                messages,
+                step,
+                tool_evidence,
+                final_answer,
+                final_verdict,
+                llm_calls,
+                llm_usages,
+                store,
+            )
+
     else:
         # max_steps exceeded
         final_verdict = "partial"

api/ops/store/checkpoints.py

@@ -0,0 +1,97 @@
+"""Ops Desk Checkpoint 存储适配层（P1-2）。
+
+复用 `ops_run_checkpoints` 表：
+- `checkpoint_id` 字段存放 thread/session 标识。
+- `state_json` 存放 ReAct 运行时状态。
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from api.ops.store.runs import OpsRunStore
+from api.rag_env import supabase_client
+
+logger = logging.getLogger(__name__)
+
+
+class CheckpointStoreError(RuntimeError):
+    """Checkpoint 读写或校验失败。"""
+
+
+REQUIRED_STATE_KEYS = ("route", "query", "step", "messages", "tool_evidence")
+
+
+def _validate_react_state(state_json: Any) -> dict[str, Any]:
+    """校验 checkpoint 状态是否足够恢复 ReAct 循环。
+
+    校验通过返回原字典；失败抛出 CheckpointStoreError。
+    """
+    if not isinstance(state_json, dict):
+        raise CheckpointStoreError("checkpoint state_json is not a dict")
+    missing = [k for k in REQUIRED_STATE_KEYS if k not in state_json]
+    if missing:
+        raise CheckpointStoreError(f"checkpoint state missing keys: {missing}")
+    if state_json.get("route") != "react":
+        raise CheckpointStoreError("checkpoint route is not 'react'")
+    if not isinstance(state_json.get("messages"), list):
+        raise CheckpointStoreError("checkpoint state messages is not a list")
+    if not isinstance(state_json.get("step"), int):
+        raise CheckpointStoreError("checkpoint state step is not an int")
+    return state_json
+
+
+def save_checkpoint(
+    run_id: str,
+    thread_id: str,
+    state_json: dict[str, Any],
+    store: OpsRunStore | None = None,
+) -> dict[str, Any]:
+    """保存 ReAct 运行时 checkpoint。
+
+    参数:
+        run_id: 当前 run id。
+        thread_id: session/thread 标识；与 `checkpoint_id` 同义。
+        state_json: 运行状态字典。
+        store: 可选 OpsRunStore；默认使用全局 supabase_client() 构造。
+    """
+    target = store if store is not None else OpsRunStore(supabase_client())
+    if not hasattr(target, "save_checkpoint"):
+        raise CheckpointStoreError("store does not support save_checkpoint")
+    return target.save_checkpoint(run_id, thread_id, state_json)
+
+
+def find_latest_checkpoint_for_session(
+    session_id: str,
+    store: OpsRunStore | None = None,
+) -> dict[str, Any] | None:
+    """按 session_id 查找最新的有效 checkpoint（跨 run）。
+
+    返回整行（含 run_id / checkpoint_id / state_json / created_at）；
+    不存在时返回 None。
+    """
+    target = store if store is not None else OpsRunStore(supabase_client())
+    # 防御：部分测试 double 未实现 checkpoint 方法时直接返回 None
+    if not hasattr(target, "find_latest_checkpoint_for_session"):
+        return None
+    return target.find_latest_checkpoint_for_session(session_id)
+
+
+def load_checkpoint(
+    run_id: str,
+    thread_id: str,
+    store: OpsRunStore | None = None,
+) -> dict[str, Any] | None:
+    """读取指定 run + thread 的 checkpoint。
+
+    返回状态字典；不存在时返回 None。
+    注意：返回前不做结构校验，由调用方 `resume_react_state` 处理。
+    """
+    target = store if store is not None else OpsRunStore(supabase_client())
+    if not hasattr(target, "load_checkpoint"):
+        return None
+    row = target.load_checkpoint(run_id, thread_id)
+    if not row:
+        return None
+    return row.get("state_json")

api/ops/store/runs.py

@@ -216,6 +216,42 @@ def _once() -> dict[str, Any]:
 
         return supabase_execute_with_retry(_once)
 
+    def load_checkpoint(self, run_id: str, checkpoint_id: str) -> dict[str, Any] | None:
+        def _once() -> dict[str, Any] | None:
+            res = (
+                self.client.table("ops_run_checkpoints")
+                .select("*")
+                .eq("run_id", run_id)
+                .eq("checkpoint_id", checkpoint_id)
+                .limit(1)
+                .execute()
+            )
+            rows = res.data if isinstance(res.data, list) else []
+            if rows and isinstance(rows[0], dict):
+                return rows[0]
+            return None
+
+        return supabase_execute_with_retry(_once)
+
+    def find_latest_checkpoint_for_session(self, session_id: str) -> dict[str, Any] | None:
+        """按 session_id（即 checkpoint_id）查找最新的 checkpoint（跨 run）。"""
+
+        def _once() -> dict[str, Any] | None:
+            res = (
+                self.client.table("ops_run_checkpoints")
+                .select("*")
+                .eq("checkpoint_id", session_id)
+                .order("created_at", desc=True)
+                .limit(1)
+                .execute()
+            )
+            rows = res.data if isinstance(res.data, list) else []
+            if rows and isinstance(rows[0], dict):
+                return rows[0]
+            return None
+
+        return supabase_execute_with_retry(_once)
+
     def save_artifact(
         self, run_id: str, kind: str, payload: dict[str, Any]
     ) -> dict[str, Any]: