feat(runtime): 会话查询公共入口与就绪度收口（改造第 5 步收窄）

- RuntimeService 新增 list_sessions / get_session_history（会话列表/磁盘权威历史），
  CLI 等非 Web 调用方不再依赖 Web 路由；principal 与目标用户一致性校验防身份伪造
- chain 验收补会话查询断言（列表含会话/历史含消息/越权抛 PermissionError）
- gateway_current_state.md：§6.1/§7 实施状态速览（第 0-5 步完成度与遗留清单）
- 全量 74 测试失败恰为 4 项存量，与改造无关

cache_research/gateway/gateway_current_state.md

@@ -226,6 +226,8 @@ v1 原文把「内存态」直接判为「状态唯一 Owner 的障碍」，混
 | N4 | 并发重复任务记录 | `models.py:157`「检查运行中→创建记录」与门闸分属两处，并发下是否产生重复任务记录未验证 | G7①/G10 |
 | N5 | `issue_socket_token` 发放竞争 | 未复现；若存在作为独立小缺陷处理 | G7② |
 
+> **实施状态（2026-09-07 第 0-4 步完成后）**：N1 ✅已修复（回退路径 finally 释放，`chat_flow_task_main.py`）；N2 ✅已修复（审批创建处显式 `task_id=client_sid`，`chat_flow_tool_loop.py` 4 处 + `_handle_submit_plan`）；N3 ⬜仍待用户重启服务后人工验证（清单见 §7.1）；N4/N5 保持原口径未动；S10 ✅已修复（`save_personalization_config` 改用公共 `atomic_write_json`）。另修复两处盘点后新发现：执行链 4 处裸 `socketio.emit`（未绑定抛 AttributeError；`task_stopped` 曾因异常顺序跳过事件流记录，已改为先写事件流再推送）；chat 任务无 cid 时服务层补建对话（`RuntimeService._ensure_conversation_for_chat`，自 `tasks/api.py` 下沉）。
+
 ### 6.2 结构性缺口
 
 - **跨重启恢复能力**：S2/S3/S6 纯内存态，重启即失；S6 条目永不清理（无 TTL）。TTL 应在终态与迟到回答规则明确后设计，不能直接删除仍合法等待的请求（G11）。
@@ -244,6 +246,17 @@ v1 原文把「内存态」直接判为「状态唯一 Owner 的障碍」，混
 
 > 总原则：协议草案与接口适配可迭代推进；**每一步保留现有门闸、保存和恢复保障**；四层不要求四进程（G13）；Remote Worker 不是本轮完成条件。
 
+> **实施状态速览（2026-09-07，commit d9bd599c / 27aeed70）**：
+> | 步骤 | 状态 | 证据 |
+> |---|---|---|
+> | 第 0 步 | ✅ 代码侧完成（N3 人工验证待用户） | §6.1 状态注记 |
+> | 第 1 步 | ✅ | `StandaloneRuntimeLifecycleTest`；`server/tasks` 拆解；emit_event/run_background 安全包装 |
+> | 第 2 步 | ✅ | `docs/runtime_protocol.md`；`ProtocolSmokeChainTest` 全链路（含服务层补建对话） |
+> | 第 3 步 | ✅ | 协议 §5.2-5.4（序号作用域/window_start 缺口检测/重启三场景/多端共享）；`get_task_events` meta 水位 |
+> | 第 4 步 | ✅ 首版（E1-E4 面） | `modules/execution_plane` + `docs/execution_contract.md` + `ExecutionPlaneFakeBackendTest`；E5-E10 与 Host/Docker 同契约接入为后续项 |
+> | 第 5 步 | ✅ 收窄完成（代码就绪度） | 公共入口覆盖清单：run.*/approval.*/session.* 全绿（协议 §3/§4）；正式客户端开发不在本轮 |
+> | 遗留 | ⬜ | N3 真实环境人工验证、N4/N5 疑点、E5-E10 纳入、定时任务（§7.6 后置） |
+
 ### 7.1 第 0 步：闭环疑点与小修复（不依赖架构决策）
 
 1. 复现验证 N1（回退路径门闸，覆盖预占/受理失败/回退执行/最终释放全链路）。

server/runtime/service.py

@@ -15,7 +15,7 @@ from __future__ import annotations
 from typing import Any, Dict, List, Optional, Tuple
 
 from modules.i18n import tr
-from server.runtime.context import RuntimeContext
+from server.runtime.context import RuntimeContext, TrustedPrincipal
 
 
 class RuntimeService:
@@ -147,6 +147,71 @@ class RuntimeService:
 
         return task_manager.promote_runtime_pending_to_guidance(username, task_id, message_id)
 
+    # ---- 会话查询（公共入口；CLI/定时器等非 Web 调用方不依赖 Web 路由）----
+
+    def list_sessions(
+        self,
+        username: str,
+        workspace_id: str,
+        principal: Optional[TrustedPrincipal] = None,
+        limit: int = 50,
+        offset: int = 0,
+        multi_agent_mode: Optional[bool] = None,
+    ) -> Dict[str, Any]:
+        """会话列表查询。principal 省略时按 username/workspace_id 构造最小身份快照。
+
+        注意：返回结构由 conversation 管理链路定义（items/total 等），本服务只做
+        资源装配与转发，不重排字段。
+        """
+        terminal, _workspace = self._resources_for_query(username, workspace_id, principal)
+        cm = getattr(terminal, "context_manager", None)
+        if cm is None:
+            raise RuntimeError(tr("tasks.system_not_initialized"))
+        return cm.get_conversation_list(limit=limit, offset=offset, multi_agent_mode=multi_agent_mode)
+
+    def get_session_history(
+        self,
+        username: str,
+        workspace_id: str,
+        conversation_id: str,
+        principal: Optional[TrustedPrincipal] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """会话历史读取（磁盘权威快照）。返回 None 表示不存在。"""
+        if not str(conversation_id or "").strip():
+            raise ValueError("runtime_context: conversation_id 不能为空")
+        terminal, _workspace = self._resources_for_query(username, workspace_id, principal)
+        cm = getattr(terminal, "context_manager", None)
+        if cm is None:
+            raise RuntimeError(tr("tasks.system_not_initialized"))
+        manager = cm._get_conversation_manager_for_id(conversation_id)
+        return manager.load_conversation(conversation_id)
+
+    @staticmethod
+    def _resources_for_query(username: str, workspace_id: str, principal: Optional[TrustedPrincipal]):
+        """查询类调用的资源装配（工作区级 terminal 即可，不加载会话到内存）。"""
+        from server.context import RuntimeIdentity, get_user_resources
+
+        if principal is None:
+            principal = TrustedPrincipal(username=username, workspace_id=workspace_id)
+        if principal.username != username:
+            # 纵深防御：principal 是适配层认证后的可信身份，不得与查询目标身份不符
+            raise PermissionError("runtime_context: principal 与查询目标用户不一致")
+        identity = RuntimeIdentity(
+            host_mode=principal.host_mode,
+            host_workspace_id=principal.host_workspace_id,
+            is_api_user=principal.is_api_user,
+            role=principal.role,
+            preferred_model_key=principal.preferred_model_key,
+            preferred_run_mode=principal.preferred_run_mode,
+            preferred_thinking_mode=principal.preferred_thinking_mode,
+        )
+        terminal, workspace = get_user_resources(
+            username, workspace_id=workspace_id, update_session=False, identity=identity
+        )
+        if terminal is None:
+            raise RuntimeError(tr("tasks.system_not_initialized"))
+        return terminal, workspace
+
     # ---- 观察（内部接口；后台调用方不必为观察任务发 HTTP 请求）----
 
     def get_task(self, username: str, task_id: str):

test/runtime_standalone_checks.py

@@ -185,7 +185,36 @@ def check_chain():
     rec2 = _wait_terminal_state(rec2.task_id)
     assert rec2.status in {"succeeded", "failed", "stopped", "canceled"}, f"Run2 应达终态: {rec2.status}"
 
-    # 5. 审批语义：create → list_pending → decide → 重复 decide 返回现状（单次裁决）
+    # 5.5 会话查询公共入口（CLI/非 Web 调用方不依赖 Web 路由）
+    from server.runtime import TrustedPrincipal as _TP
+
+    listing = runtime_service.list_sessions(
+        "gw_smoke_user", "gwsmoke",
+        principal=_TP(username="gw_smoke_user", workspace_id="gwsmoke", role="admin",
+                      host_mode=True, host_workspace_id="gwsmoke"),
+    )
+    items = listing.get("items") or listing.get("conversations") or []
+    assert any((it.get("conversation_id") or it.get("id")) == conv_id for it in items), (
+        f"会话列表应含本次会话 {conv_id}（keys={list(listing.keys())}）"
+    )
+    history = runtime_service.get_session_history(
+        "gw_smoke_user", "gwsmoke", conv_id,
+        principal=_TP(username="gw_smoke_user", workspace_id="gwsmoke", role="admin",
+                      host_mode=True, host_workspace_id="gwsmoke"),
+    )
+    assert history and "协议链路验收 ping" in json.dumps(history, ensure_ascii=False), \
+        "会话历史公共入口应可读且含 user 消息"
+    try:
+        runtime_service.get_session_history(
+            "gw_smoke_user", "gwsmoke", conv_id,
+            principal=_TP(username="mallory", workspace_id="gwsmoke"),
+        )
+    except PermissionError:
+        pass
+    else:
+        raise AssertionError("principal 与查询目标不一致必须抛 PermissionError")
+
+    # 6. 审批语义：create → list_pending → decide → 重复 decide 返回现状（单次裁决）
     from server.state import tool_approval_manager
 
     item = tool_approval_manager.create_request(