diff --git a/backend/package/yuxi/services/chat_service.py b/backend/package/yuxi/services/chat_service.py
index 57b3b9fd8..6f3a3e0bb 100644
--- a/backend/package/yuxi/services/chat_service.py
+++ b/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),
diff --git a/backend/test/integration/services/test_state_reader_interrupt_integration.py b/backend/test/integration/services/test_state_reader_interrupt_integration.py
new file mode 100644
index 000000000..b7c49c3e8
--- /dev/null
+++ b/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)
diff --git a/backend/test/unit/services/test_chat_service_sync.py b/backend/test/unit/services/test_chat_service_sync.py
index 46c10eddd..9dd72df5d 100644
--- a/backend/test/unit/services/test_chat_service_sync.py
+++ b/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
diff --git a/backend/test/unit/services/test_state_reader_interrupt_recovery.py b/backend/test/unit/services/test_state_reader_interrupt_recovery.py
new file mode 100644
index 000000000..3a7d6fbb3
--- /dev/null
+++ b/backend/test/unit/services/test_state_reader_interrupt_recovery.py
@@ -0,0 +1,113 @@
+"""骨架图 state 读取必须保留审批中断，否则刷新状态接口会丢失继续入口。
+
+用**真实 LangGraph checkpoint**（InMemorySaver）对照原图与骨架图：骨架图只有 noop
+节点，`aget_state` 无法按原图节点重建 `tasks`，中断必须从 checkpoint 的
+`pending_writes`（`__interrupt__` channel）直接恢复。
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph import END, START, StateGraph
+from langgraph.types import interrupt
+from typing_extensions import TypedDict
+
+os.environ.setdefault("OPENAI_API_KEY", "test-key")
+
+from yuxi.services import chat_service as svc
+
+pytestmark = [pytest.mark.unit]
+
+
+class _State(TypedDict):
+    messages: list
+
+
+def _approval_node(state):
+    approved = interrupt({"question": "是否允许执行该命令？", "tool": "execute"})
+    return {"messages": [*state["messages"], approved]}
+
+
+def _compile_real(saver):
+    graph = StateGraph(_State)
+    graph.add_node("approval", _approval_node)
+    graph.add_edge(START, "approval")
+    graph.add_edge("approval", END)
+    return graph.compile(checkpointer=saver)
+
+
+def _compile_skeleton(saver):
+    """与 `_get_state_reader_graph` 同构：零工具、单 noop 节点。"""
+    graph = StateGraph(_State)
+    graph.add_node("state_reader_noop", lambda state: {})
+    graph.set_entry_point("state_reader_noop")
+    return graph.compile(checkpointer=saver)
+
+
+@pytest.fixture()
+def interrupted_checkpoint():
+    """构造一条停在 interrupt 上的真实 checkpoint，返回 (saver, config)。"""
+    saver = InMemorySaver()
+    config = {"configurable": {"thread_id": "thread-interrupt"}}
+    list(_compile_real(saver).stream({"messages": []}, config, stream_mode="values"))
+    return saver, config
+
+
+def test_skeleton_graph_cannot_rebuild_tasks_from_interrupted_checkpoint(interrupted_checkpoint):
+    """记录骨架图的固有限制：它重建不出 tasks，因此中断取不到。"""
+    saver, config = interrupted_checkpoint
+
+    real_snapshot = _compile_real(saver).get_state(config)
+    skeleton_snapshot = _compile_skeleton(saver).get_state(config)
+
+    assert len(real_snapshot.tasks) == 1
+    assert real_snapshot.tasks[0].interrupts
+    assert not skeleton_snapshot.tasks
+    assert svc._extract_interrupt_info(skeleton_snapshot) is None
+
+
+@pytest.mark.asyncio
+async def test_pending_interrupt_is_recovered_from_checkpoint_writes(monkeypatch, interrupted_checkpoint):
+    """修复点：中断回退到 checkpoint 原始写入读取，不依赖执行图结构。"""
+    saver, config = interrupted_checkpoint
+    monkeypatch.setattr(svc.pg_manager, "get_langgraph_checkpointer", lambda: saver)
+
+    interrupt_info = await svc._read_pending_interrupt(uid="user-1", thread_id="thread-interrupt")
+
+    assert interrupt_info is not None
+    assert interrupt_info.value == {"question": "是否允许执行该命令？", "tool": "execute"}
+
+
+@pytest.mark.asyncio
+async def test_reader_snapshot_plus_recovered_interrupt_matches_real_graph(monkeypatch, interrupted_checkpoint):
+    """骨架图 values + 恢复出的中断，与真实图的审批信息等价。"""
+    saver, config = interrupted_checkpoint
+    monkeypatch.setattr(svc.pg_manager, "get_langgraph_checkpointer", lambda: saver)
+
+    skeleton_snapshot = _compile_skeleton(saver).get_state(config)
+    recovered = await svc._read_pending_interrupt(uid="user-1", thread_id="thread-interrupt")
+    real_snapshot = _compile_real(saver).get_state(config)
+
+    assert skeleton_snapshot.values["messages"] == real_snapshot.values["messages"]
+    assert recovered.id == real_snapshot.tasks[0].interrupts[0].id
+    assert recovered.value == real_snapshot.tasks[0].interrupts[0].value
+
+
+@pytest.mark.asyncio
+async def test_pending_interrupt_returns_none_without_interrupt(monkeypatch):
+    """没有中断的 checkpoint 不应被误判为等待审批。"""
+    saver = InMemorySaver()
+    monkeypatch.setattr(svc.pg_manager, "get_langgraph_checkpointer", lambda: saver)
+
+    complete = StateGraph(_State)
+    complete.add_node("done", lambda state: {"messages": [*state["messages"], "ok"]})
+    complete.add_edge(START, "done")
+    complete.add_edge("done", END)
+    graph = complete.compile(checkpointer=saver)
+    list(graph.stream({"messages": []}, {"configurable": {"thread_id": "thread-done"}}, stream_mode="values"))
+
+    assert await svc._read_pending_interrupt(uid="user-1", thread_id="thread-done") is None
+    assert await svc._read_pending_interrupt(uid="user-1", thread_id="thread-missing") is None
diff --git a/docs/develop-guides/decisions/implemented/2026-09-10-state-reader-preserves-interrupts.md b/docs/develop-guides/decisions/implemented/2026-09-10-state-reader-preserves-interrupts.md
new file mode 100644
index 000000000..98734e14f
--- /dev/null
+++ b/docs/develop-guides/decisions/implemented/2026-09-10-state-reader-preserves-interrupts.md
@@ -0,0 +1,49 @@
+# state 骨架图读取必须保留审批中断
+
+状态：implemented
+类型：bug-fix
+Owner：backend/package/yuxi/services/chat_service.py
+
+## 问题
+
+为把 state 面板查询从秒级（重型 Agent 完整构图 60-80s）降到毫秒级，`_read_checkpoint_state` 改用零工具骨架图 `_get_state_reader_graph`（单 `state_reader_noop` 节点）读取 checkpoint。
+
+但 `aget_state` 不只是按 schema 读 `values`：它还依据**当前执行图的节点定义**，用 checkpoint 的 `next` 与 `pending_writes` 重建 `tasks`。骨架图没有原图的 `tools`/中间件审批节点，停在审批中断上的 checkpoint 会得到空 `tasks`。
+
+`_extract_interrupt_info` 首选 `state.tasks[*].interrupts`，其 `values["__interrupt__"]` 兜底在骨架图下同样为空（实测 `get_state` 返回的 values 只有常规 channel）。结果是 `get_agent_state_view` 不再返回 `interrupt`——**用户等待工具审批或提问时刷新页面，会丢失继续操作的入口**。
+
+最小复现（真实 LangGraph + InMemorySaver，`tools` 节点内 `interrupt(...)`）：原图返回 1 个 task 且 `interrupts` 非空；骨架图返回 0 个 task。该差异不依赖模型或 MCP。
+
+## 决策
+
+中断不从图结构推导，而是**直接从 checkpoint 的 pending writes 读取**：新增 `_read_pending_interrupt`，用 checkpointer 的 `aget_tuple` 取 `CheckpointTuple.pending_writes`，返回 `__interrupt__` channel 中的中断值。
+
+调用点只在 `get_agent_state_view` 且 `latest_run.status == "interrupted"` 时回退到它——普通状态查询不额外读一次 checkpoint，保留骨架图的性能收益。
+
+骨架图继续负责 `values`（state 面板的待办/用量/消息）。两条读取路径共同覆盖「真实图 in-flight 中断」（`check_and_handle_interrupts`，走 `tasks`）与「历史/刷新后的中断」（走 pending writes）。
+
+## 替代方案
+
+- 回退到完整 `agent.get_graph`：恢复 `tasks` 但把 state 查询重新拖回 60-80s；拒绝。
+- 骨架图伪造与原图同名的节点：原图节点名由 `create_agent` 与 middleware 装配产生，复制即耦合其内部实现，且随上游变化静默失效；拒绝。
+- 把中断塞进 `values["__interrupt__"]` 走既有兜底：语义上污染 state values，且会让「中断」看起来像普通 state 字段；拒绝。
+
+## 后果
+
+state 读取的中断来源从「图结构」变为「checkpoint 原始写入」，与图是否为零工具骨架无关。`interrupted` 状态的 Run 每次查询多一次 `aget_tuple`（单次索引读），其余状态零额外开销。
+
+骨架图无法重建 `tasks` 是 LangGraph 的固有行为，已由测试显式记录，避免后续误以为 `tasks` 可用。
+
+## 验证
+
+`backend/test/unit/services/test_state_reader_interrupt_recovery.py`（真实 LangGraph checkpoint，非 mock）：
+
+- 记录骨架图固有限制：原图 `tasks` 为 1 且含 interrupt，骨架图 `tasks` 为空、`_extract_interrupt_info` 返回 `None`；
+- 修复点：`_read_pending_interrupt` 从 checkpoint 写入恢复出中断，其 `id` 与 `value` 与真实图 `tasks[0].interrupts[0]` 完全一致；
+- 骨架图 `values` 与真实图一致（确认性能优化未被破坏）；
+- 负向：已完成、无中断的 checkpoint 返回 `None`；不存在的 thread 返回 `None`。
+
+`backend/test/integration/services/test_state_reader_interrupt_integration.py`（真实 PostgreSQL 的 AsyncPostgresSaver，非 InMemorySaver）：
+
+- 停在 interrupt 的真实 checkpoint，其持久化 pending writes 里的 `__interrupt__` channel 可被 `_read_pending_interrupt` 恢复，`value` 与写入一致；
+- 已完成、无中断的真实 checkpoint 返回 `None`。