diff --git a/cache_research/gateway/gateway_current_state.md b/cache_research/gateway/gateway_current_state.md index ecf20848..50315d28 100644 --- a/cache_research/gateway/gateway_current_state.md +++ b/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(回退路径门闸,覆盖预占/受理失败/回退执行/最终释放全链路)。 diff --git a/server/runtime/service.py b/server/runtime/service.py index 186ebfe0..ae4fd59a 100644 --- a/server/runtime/service.py +++ b/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): diff --git a/test/runtime_standalone_checks.py b/test/runtime_standalone_checks.py index 9d735f20..2dd1b8d3 100644 --- a/test/runtime_standalone_checks.py +++ b/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(