perf(api): state 查询改用零工具骨架图，秒级降到毫秒级

backend/package/yuxi/services/chat_service.py

@@ -1689,10 +1689,74 @@ def _serialize_state_messages(values: dict[str, Any]) -> list[dict[str, Any]]:
     return serialized
 
 
+# state 查询专用骨架图: channel 结构由 state_schema 决定,与工具集无关。
+# 完整 get_graph 需连接全部 MCP 装配工具(重型 Agent 60-80s,子智能体 workdir
+# 各异导致缓存全 miss);骨架图零工具、毫秒级编译,全局单例即可正确恢复任意
+# thread 的 checkpoint values(ChatBotState 为 BaseState 超集,两种图通用)。
+_STATE_READER_GRAPH = None
+
+
+def _build_state_reader_schema():
+    """构建与真实 agent graph 一致的 state schema(含 middleware 注入字段)。"""
+    from langchain.agents.factory import _resolve_schemas
+    from langchain.agents.middleware import TodoListMiddleware
+    from yuxi.agents.buildin.chatbot.state import ChatBotState
+    from yuxi.agents.middlewares import TokenUsageMiddleware
+    from yuxi.agents.middlewares.skills import SkillsMiddleware
+
+    schema, _, _ = _resolve_schemas(
+        [
+            TodoListMiddleware.state_schema,
+            TokenUsageMiddleware.state_schema,
+            SkillsMiddleware.state_schema,
+            ChatBotState,
+        ]
+    )
+    return schema
+
+
+async def _get_state_reader_graph():
+    global _STATE_READER_GRAPH
+    if _STATE_READER_GRAPH is not None:
+        return _STATE_READER_GRAPH
+    from langgraph.graph import StateGraph
+
+    # 骨架图必须复用与真实 agent graph 一致的 state schema。真实 schema 由
+    # create_agent 合并 middleware 注入字段(如 todos/token_usage/activated_skills)
+    # 得到；若只用裸 ChatBotState，aget_state 会按骨架图 schema 过滤 checkpoint，
+    # 丢失这些字段，导致前端 state-panel 的待办与 token 用量消失。
+    state_schema = _build_state_reader_schema()
+    checkpointer = pg_manager.get_langgraph_checkpointer()
+    skeleton = StateGraph(state_schema)
+    skeleton.add_node("state_reader_noop", lambda state: {})
+    skeleton.set_entry_point("state_reader_noop")
+    _STATE_READER_GRAPH = skeleton.compile(checkpointer=checkpointer)
+    return _STATE_READER_GRAPH
+
+
 async def _read_checkpoint_state(agent, *, uid: str, thread_id: str, context):
-    graph = await agent.get_graph(context=context)
+    reader = await _get_state_reader_graph()
     langgraph_config = {"configurable": {"uid": uid, "thread_id": thread_id}}
-    return await graph.aget_state(langgraph_config)
+    return await reader.aget_state(langgraph_config)
+
+
+async def _read_pending_interrupt(*, uid: str, thread_id: str):
+    """从 checkpoint 的 pending writes 读取中断值，不依赖执行图结构。
+
+    骨架图只有 state_reader_noop 节点，`aget_state` 无法按原图节点重建 `tasks`，
+    因此等待审批的 checkpoint 在骨架图下 `tasks` 为空、`_extract_interrupt_info`
+    取不到中断——用户刷新状态接口会丢失审批入口。中断本身写在 checkpoint 的
+    `__interrupt__` channel 里，直接读原始写入即可恢复。
+    """
+    checkpointer = pg_manager.get_langgraph_checkpointer()
+    langgraph_config = {"configurable": {"uid": uid, "thread_id": thread_id}}
+    checkpoint_tuple = await checkpointer.aget_tuple(langgraph_config)
+    if checkpoint_tuple is None:
+        return None
+    for _task_id, channel, value in checkpoint_tuple.pending_writes or []:
+        if channel == "__interrupt__" and value:
+            return value[0]
+    return None
 
 
 async def get_agent_state_view(
@@ -1763,6 +1827,10 @@ async def get_agent_state_view(
             )
         }
         interrupt_info = _extract_interrupt_info(state) if state else None
+        if interrupt_info is None and latest_run is not None and latest_run.status == "interrupted":
+            # 骨架图重建不出 tasks，中断回退到 checkpoint 原始写入读取；只在 Run 确实
+            # 处于 interrupted 时才多读一次，普通状态查询不付这个成本。
+            interrupt_info = await _read_pending_interrupt(uid=str(current_uid), thread_id=thread_id)
         if latest_run and latest_run.status == "interrupted" and interrupt_info:
             response["interrupt"] = {
                 **_build_pending_interrupt_payload(interrupt_info, thread_id),

backend/test/integration/services/test_state_reader_interrupt_integration.py

@@ -0,0 +1,97 @@
+"""真实 PostgreSQL 验证：骨架图读取时，中断从 checkpoint pending writes 恢复。
+
+单元测试用 InMemorySaver 验证了 ``_read_pending_interrupt`` 的语义；本集成测试用
+真实 PostgreSQL 的 AsyncPostgresSaver 验证「持久化的 pending writes 里 __interrupt__
+channel」在真实存储后端上同样可读——这是 InMemorySaver 覆盖不到的存储格式差异。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import uuid
+from typing import TypedDict
+
+import pytest
+from langgraph.graph import END, START, StateGraph
+from langgraph.types import interrupt
+from psycopg_pool import AsyncConnectionPool
+from yuxi.services import chat_service as svc
+from yuxi.storage.postgres.manager import PostgresManager
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+
+class _State(TypedDict):
+    messages: list
+
+
+def _approval_node(state):
+    approved = interrupt({"question": "是否允许执行该命令？", "tool": "execute"})
+    return {"messages": [*state["messages"], approved]}
+
+
+def _pg_url() -> str:
+    return os.environ["POSTGRES_URL"].replace("+asyncpg", "").replace("+psycopg", "")
+
+
+def _new_manager() -> PostgresManager:
+    manager = object.__new__(PostgresManager)
+    manager.__init__()
+    manager._initialized = True
+    return manager
+
+
+async def test_pending_interrupt_recovered_from_real_postgres_checkpoint(monkeypatch):
+    """真实 PG：停在 interrupt 的 checkpoint，其 pending writes 里的中断可被恢复。"""
+    manager = _new_manager()
+    monkeypatch.setattr("yuxi.services.chat_service.pg_manager", manager)
+    thread_id = f"pytest-interrupt-{uuid.uuid4()}"
+    uid = "pytest-user"
+
+    async with (
+        asyncio.timeout(20),
+        AsyncConnectionPool(_pg_url(), min_size=1, max_size=2, open=False, kwargs={"autocommit": True}) as pool,
+    ):
+        manager.langgraph_pool = pool
+        graph = StateGraph(_State)
+        graph.add_node("approval", _approval_node)
+        graph.add_edge(START, "approval")
+        graph.add_edge("approval", END)
+        compiled = graph.compile(checkpointer=manager.get_langgraph_checkpointer())
+        config = {"configurable": {"uid": uid, "thread_id": thread_id}}
+        async for _ in compiled.astream({"messages": []}, config, stream_mode="values"):
+            pass
+
+        interrupt_info = await svc._read_pending_interrupt(uid=uid, thread_id=thread_id)
+
+        assert interrupt_info is not None
+        assert interrupt_info.value == {"question": "是否允许执行该命令？", "tool": "execute"}
+
+        await manager.get_langgraph_checkpointer().adelete_thread(thread_id)
+
+
+async def test_completed_checkpoint_returns_no_interrupt(monkeypatch):
+    """真实 PG：已完成、无中断的 checkpoint 不得被误判为等待审批。"""
+    manager = _new_manager()
+    monkeypatch.setattr("yuxi.services.chat_service.pg_manager", manager)
+    thread_id = f"pytest-complete-{uuid.uuid4()}"
+    uid = "pytest-user"
+
+    async with (
+        asyncio.timeout(20),
+        AsyncConnectionPool(_pg_url(), min_size=1, max_size=2, open=False, kwargs={"autocommit": True}) as pool,
+    ):
+        manager.langgraph_pool = pool
+        graph = StateGraph(_State)
+        graph.add_node("done", lambda state: {"messages": [*state["messages"], "ok"]})
+        graph.add_edge(START, "done")
+        graph.add_edge("done", END)
+        compiled = graph.compile(checkpointer=manager.get_langgraph_checkpointer())
+        config = {"configurable": {"uid": uid, "thread_id": thread_id}}
+        async for _ in compiled.astream({"messages": []}, config, stream_mode="values"):
+            pass
+
+        assert await svc._read_pending_interrupt(uid=uid, thread_id=thread_id) is None
+
+        await manager.get_langgraph_checkpointer().adelete_thread(thread_id)

backend/test/unit/services/test_chat_service_sync.py

@@ -6,10 +6,9 @@
 import pytest
 from fastapi import HTTPException
 from langchain.messages import AIMessage, HumanMessage, ToolMessage
-
 from yuxi.agents import context as agent_context
-from yuxi.workspace import paths as workspace_paths
 from yuxi.services import chat_service as svc
+from yuxi.workspace import paths as workspace_paths
 
 
 def _empty_agent_context(_uid: str) -> str:
@@ -1299,16 +1298,6 @@ async def get_latest_subagent_run_by_thread_for_user(self, thread_id: str, uid:
                 to_dict=lambda: {"created_at": "2026-06-21T01:00:00Z", "finished_at": None},
             )
 
-    class Graph:
-        async def aget_state(self, config):
-            assert config["configurable"]["thread_id"] == child_thread_id
-            return SimpleNamespace(
-                values={
-                    "messages": [HumanMessage(content="do work"), AIMessage(content="working")],
-                    "artifacts": ["out.txt"],
-                }
-            )
-
     class Context:
         def __init__(self, *, thread_id="", uid=""):
             self.thread_id = thread_id
@@ -1322,14 +1311,15 @@ def update(self, data: dict):
     class Agent:
         context_schema = Context
 
-        async def get_graph(self, *, context):
-            assert context.thread_id == child_thread_id
-            assert context.uid == "user-1"
-            assert context.model == "provider:run-model"
-            assert context.runtime_scope_id == "parent-thread"
-            assert context.workdir_relative_path == "projects/11111111-1111-4111-8111-111111111111"
-            assert context.workdir_path == "/home/gem/user-data/projects/11111111-1111-4111-8111-111111111111"
-            return Graph()
+    async def _fake_read_state(_agent, *, uid, thread_id, context):
+        assert thread_id == child_thread_id
+        assert uid == "user-1"
+        return SimpleNamespace(
+            values={
+                "messages": [HumanMessage(content="do work"), AIMessage(content="working")],
+                "artifacts": ["out.txt"],
+            }
+        )
 
     monkeypatch.setattr(svc, "ConversationRepository", ConvRepo)
     monkeypatch.setattr(svc, "resolve_conversation_workdir_path", _resolve_test_workdir)
@@ -1338,6 +1328,7 @@ async def get_graph(self, *, context):
     monkeypatch.setattr(svc, "AgentRunRepository", RunRepo)
     monkeypatch.setattr(svc, "normalize_agent_context_config", _fake_normalize_agent_context_config)
     monkeypatch.setattr(svc.agent_manager, "get_agent", lambda backend_id: Agent())
+    monkeypatch.setattr(svc, "_read_checkpoint_state", _fake_read_state)
 
     result = await svc.get_agent_state_view(
         thread_id=child_thread_id,
@@ -1418,10 +1409,6 @@ async def get_latest_subagent_run_by_thread_for_user(self, thread_id: str, uid:
                 input_payload={"runtime": {}},
             )
 
-    class Graph:
-        async def aget_state(self, _config):
-            return SimpleNamespace(values={})
-
     class Context:
         def __init__(self, *, thread_id="", uid=""):
             self.thread_id = thread_id
@@ -1434,10 +1421,10 @@ def update(self, data: dict):
     class Agent:
         context_schema = Context
 
-        async def get_graph(self, *, context):
-            assert context.thread_id == child_thread_id
-            assert context.uid == "user-1"
-            return Graph()
+    async def _fake_read_state(_agent, *, uid, thread_id, context):
+        assert thread_id == child_thread_id
+        assert uid == "user-1"
+        return SimpleNamespace(values={})
 
     monkeypatch.setattr(svc, "ConversationRepository", ConvRepo)
     monkeypatch.setattr(svc, "resolve_conversation_workdir_path", _resolve_test_workdir)
@@ -1446,6 +1433,7 @@ async def get_graph(self, *, context):
     monkeypatch.setattr(svc, "AgentRunRepository", RunRepo)
     monkeypatch.setattr(svc, "normalize_agent_context_config", _fake_normalize_agent_context_config)
     monkeypatch.setattr(svc.agent_manager, "get_agent", lambda _backend_id: Agent())
+    monkeypatch.setattr(svc, "_read_checkpoint_state", _fake_read_state)
 
     with pytest.raises(HTTPException) as exc:
         await svc.get_agent_state_view(
@@ -1471,3 +1459,18 @@ async def test_build_agent_input_context_keeps_prompt_when_workspace_agent_conte
     )
 
     assert context["system_prompt"] == "原始系统提示词"
+
+
+def test_build_state_reader_schema_includes_middleware_fields() -> None:
+    from typing import get_type_hints
+
+    schema = svc._build_state_reader_schema()
+    hints = get_type_hints(schema)
+
+    # 骨架图若丢失这些 middleware 注入字段，aget_state 按 schema 过滤 checkpoint 后，
+    # 前端 state-panel 的待办(todos)与 token 用量(token_usage)会消失。
+    assert "todos" in hints
+    assert "token_usage" in hints
+    assert "activated_skills" in hints
+    assert "subagent_runs" in hints
+    assert "artifacts" in hints