From f43b03484d3d29ec4a507a6e924bb764452eebd0 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Tue, 8 Sep 2026 02:08:33 +0800 Subject: [PATCH] =?UTF-8?q?refactor(runtime):=20=E6=8B=86=E9=99=A4=20sessi?= =?UTF-8?q?on=5Fdata=20=E5=85=BC=E5=AE=B9=E6=A1=A5=EF=BC=8C=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E4=B8=8A=E4=B8=8B=E6=96=87=E4=B8=89=E5=B1=82=E7=BB=93?= =?UTF-8?q?=E6=9E=84=E5=8C=96=E7=9B=B4=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - TaskRecord 去 session_data dict，直存 principal/task_params/directives + 可变 goal_progress 字段（__slots__） - create_chat_task 签名改为收 RuntimeContext 必填，TaskRecord.__init__ 同步精简（身份/消息/模式等全部从 ctx 派生） - _run_chat_task 身份还原/门闸认领/事件注入/terminal 属性设置 全部改为三层属性访问；goal_progress 回写改独立可变字段 - RuntimeService.create_task 原样透传 ctx；to_session_data() 删除 - task_public_payload / api.py 单任务载荷改三层读取 - 契约/注释/i18n 文案同步去 session_data 化；测试重写为三层断言 验证：75 测试全量回归，失败恰为 4 项存量（host_workspace_manager / skills_manager / token_usage_extractor / conversation_workspace_storage） --- .../gateway/gateway_current_state.md | 4 +- modules/i18n_messages/api_tasks.py | 4 +- server/chat_flow_task_main.py | 2 +- server/main_task_gate.py | 4 +- server/runtime/context.py | 41 ----- server/runtime/service.py | 27 +-- server/tasks/api.py | 6 +- server/tasks/models.py | 156 +++++++++--------- server/workflow_flow.py | 2 +- test/test_runtime_service.py | 62 ++++--- 10 files changed, 125 insertions(+), 183 deletions(-) diff --git a/cache_research/gateway/gateway_current_state.md b/cache_research/gateway/gateway_current_state.md index c168ad56..2725d666 100644 --- a/cache_research/gateway/gateway_current_state.md +++ b/cache_research/gateway/gateway_current_state.md @@ -122,7 +122,7 @@ Roadmap §12 的四个原语与当前实现的映射： - **入口收敛（当日验证，声明属实）**： - 全部 6 处任务创建调用点已迁移到 `runtime_service.create_task(ctx)`：`server/tasks/api.py:220`、`server/api_v1.py:335`、`server/workflow_runtime_api.py:192/:269`、`server/chat_flow_task_main.py:618/:1417`。 - **无漏网调用点**：`create_chat_task` 实体调用全仓仅 2 处 = 定义 `models.py:128` + 委托 `service.py:38`。 - - `create_chat_task` 强制显式 session_data，缺失抛 ValueError（`models.py:174-175`）。 + - `create_chat_task` 强制显式 RuntimeContext，缺失抛 ValueError；**2026-09-08 已拆除 session_data 兼容桥**：TaskRecord 三层结构化直存（`principal`/`task_params`/`directives` + 可变 `goal_progress`），`to_session_data()` 已删除，全库生产代码无 session_data dict 残留。 - **执行链运行期解耦（当日验证属实）**：`_run_chat_task` 不再建立 `test_request_context`，任务线程全程 `RuntimeIdentity` 驱动（`server/context/identity.py`）；server/core/modules/utils 下 `test_request_context` 实体调用 = 0；`models.py` 无 `session[` 读取。 - **依赖方向（结论收窄，见 §3.3）**：`server/context.py`（989 行）已拆分为 `server/context/` 子包；`get_user_resources` 已参数化（RuntimeIdentity 显式身份快照）。**「符合单向依赖」的结论仅适用于「任务所需运行期上下文已显式化」这一层。** - **审批超时透传管道**：terminal `_approval_timeout_seconds` → 工具循环 `_approval_timeout_for()` → 4 个 `_wait_*` 调用点（默认 3600s 语义不变）。 @@ -261,6 +261,8 @@ v1 原文把「内存态」直接判为「状态唯一 Owner 的障碍」，混 > **审核收口记录（gateway_implementation_review_2026-09-07.md）**：F1 已补 run.list 公共发现入口；F2 已完成 15+ 处路由转调（tasks/api.py 7、api_v1.py 3、chat/approval.py 6 + 载荷序列化收敛至 models.py 单一实现，死导入清零）；F3 已补 principal 工作区一致性校验；F4 已修（config .env 逃生门 + 测试自包含 + 假通过修复 + 审批等待链）。审核 §4 契约注释误导已修正（execution_plane/base.py：命令校验/路径授权仍在旧链路，真实后端接入时必须保留）。 > > **本轮收口决策（2026-09-07 用户拍板）**：①②③ 链路（Client ↔ Gateway ↔ Runtime）贯通即为本轮终点；③↔④ 全量贯通（Host/Docker 迁入 ExecutionBackend 契约、E5-E10 纳入）后置为独立工作，期间默认路径 `execution_backend=None`（现有真实链路不变）。 +> +> **②↔③ 内部形态结构化（2026-09-08）**：`to_session_data()` 兼容桥已拆除——`create_chat_task` 签名改为收 `RuntimeContext` 必填，`TaskRecord` 三层结构化直存（`principal`/`task_params`/`directives` + 可变 `goal_progress` 字段）；`_run_chat_task` 身份还原/门闸认领/事件注入/terminal 属性设置全部改为按层属性访问；`to_session_data()` 方法删除，生产代码 session_data dict 零残留。验证：75 测试全量回归失败恰为 4 项存量。 ### 7.1 第 0 步：闭环疑点与小修复（不依赖架构决策） diff --git a/modules/i18n_messages/api_tasks.py b/modules/i18n_messages/api_tasks.py index 02d75a60..cf538b4c 100644 --- a/modules/i18n_messages/api_tasks.py +++ b/modules/i18n_messages/api_tasks.py @@ -103,8 +103,8 @@ MESSAGES = { "en-US": "This conversation already has a running task. Please try again later.", }, "tasks.missing_session_data": { - "zh-CN": "缺少任务运行上下文（session_data），请通过公共任务入口提交", - "en-US": "Missing runtime context (session_data); please submit via the runtime service entry", + "zh-CN": "缺少任务运行上下文，请通过公共任务入口提交", + "en-US": "Missing runtime context; please submit via the runtime service entry", }, "tasks.system_not_initialized": { "zh-CN": "系统未初始化", diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index b8056b9d..8ee19c20 100644 --- a/server/chat_flow_task_main.py +++ b/server/chat_flow_task_main.py @@ -1016,7 +1016,7 @@ async def poll_completion_notifications(*, web_terminal, workspace, conversation # 主任务门闸：仅当对话没有正在运行的主任务时才预占门闸。 # 防止通知任务与主任务并发交叉写入对话历史（2026-08-12 平行时空事故）。 - # 预占成功后排发：门闸 token 随 session_data 移交给新任务线程认领释放； + # 预占成功后排发：门闸 token 经 RuntimeContext.directives 移交给新任务线程认领释放； # 排发失败则释放门闸并回滚通知标记，下轮重新收集。 gate_token = try_acquire_main_task_gate(web_terminal) if gate_token is None: diff --git a/server/main_task_gate.py b/server/main_task_gate.py index 27001f72..fb8cfe54 100644 --- a/server/main_task_gate.py +++ b/server/main_task_gate.py @@ -13,8 +13,8 @@ task_manager 注册，`create_chat_task` 的单对话互斥对它们不可见； 用法： - 所有主任务入口统一收敛在 `process_message_task`（chat_flow.py），在此获取 门闸并在 finally 释放。 -- 通知派发链（完成通知轮询器）先 `try_acquire_main_task_gate` 预占，再通过 - session_data["main_task_gate_token"] 把 token 移交给新任务线程认领； +- 通知派发链（完成通知轮询器）先 `try_acquire_main_task_gate` 预占，token 经 + RuntimeContext.directives.main_task_gate_token 移交给新任务线程认领； 派发失败时释放并回滚通知标记。 """ from __future__ import annotations diff --git a/server/runtime/context.py b/server/runtime/context.py index c66df2be..3058033f 100644 --- a/server/runtime/context.py +++ b/server/runtime/context.py @@ -107,7 +107,6 @@ class RuntimeContext: ) -> "RuntimeContext": """从对话级 terminal/工作区构造（通知链、多智能体派发等内部调用方）。 - 与既有 session_data 手工构造逐字段对齐： host_mode 取 workspace.username == "host"；偏好快照取 terminal 当前值。 """ workspace_id = getattr(workspace, "workspace_id", None) or "default" @@ -129,46 +128,6 @@ class RuntimeContext: directives=directives or InternalDirectives(), ) - def to_session_data(self) -> Dict[str, Any]: - """兼容转换：合并为现有 ``create_chat_task`` 的 session_data 快照 dict。 - - 快照在受理时固化、随任务线程传递：身份/偏好由 ``_run_chat_task`` 还原为 - ``RuntimeIdentity`` 驱动资源装配（无 Flask 隐式上下文）；门闸移交 - （main_task_gate_token）、事件注入（auto_user_message_*）、terminal 属性 - 设置（message_source/goal_mode/skill_context_messages）语义不变。 - - 其中 run_mode/thinking_mode/model_key 取「用户偏好快照」层 - （principal.preferred_*），供资源装配新建 terminal 时恢复默认值； - 本次覆盖值（params.*）由 RuntimeService.create_task 走显式参数传递， - 不进入本快照。 - """ - p = self.principal - session_data: Dict[str, Any] = { - "username": p.username, - "role": p.role, - "is_api_user": p.is_api_user, - "host_mode": p.host_mode, - "host_workspace_id": p.host_workspace_id or (p.workspace_id if p.host_mode else None), - "workspace_id": p.workspace_id, - "run_mode": p.preferred_run_mode, - "thinking_mode": p.preferred_thinking_mode, - "model_key": p.preferred_model_key, - "message_source": self.params.message_source, - "goal_mode": bool(self.params.goal_mode), - "skill_context_messages": list(self.params.skill_context_messages or []), - } - if self.params.approval_timeout_seconds is not None: - session_data["approval_timeout_seconds"] = int(self.params.approval_timeout_seconds) - d = self.directives - if d.main_task_gate_token: - session_data["main_task_gate_token"] = d.main_task_gate_token - if d.auto_user_message_event: - session_data["auto_user_message_event"] = True - if d.auto_user_message_payload: - session_data["auto_user_message_payload"] = dict(d.auto_user_message_payload) - if d.preceding_user_notices: - session_data["preceding_user_notices"] = list(d.preceding_user_notices) - return session_data def principal_from_session_snapshot( diff --git a/server/runtime/service.py b/server/runtime/service.py index 0c929a5f..24bff74a 100644 --- a/server/runtime/service.py +++ b/server/runtime/service.py @@ -5,10 +5,10 @@ 任务记录、事件流、门闸、保存保护仍由既有 TaskManager / main_task_gate / conversation_manager 承载（契约 §2 状态责任表不变）。 -兼容期说明：create_task 内部把 RuntimeContext 转换为既有 -session_data 快照传入 create_chat_task；任务线程已改为显式 RuntimeIdentity -驱动资源装配（test_request_context 桥已拆除），session_data 快照仍承载 -门闸 token、事件回放等内部指令的跨线程传递。 +上下文传递：create_task 把 RuntimeContext 三层结构（principal/params/ +directives）原样传入 create_chat_task 并固化到任务记录；任务线程按层读取 +（身份映射为 RuntimeIdentity 驱动资源装配，门闸 token、事件回放等内部指令 +随 directives 跨线程传递）——不再有 session_data 兼容快照。 """ from __future__ import annotations @@ -41,24 +41,7 @@ class RuntimeService: # 对话级 terminal 上运行。补建对话文件是装配职责，收在服务层单点， # Web/CLI/定时触发器等调用方无需各自实现「先建会话再发任务」。 conversation_id = self._ensure_conversation_for_chat(ctx) - return task_manager.create_chat_task( - ctx.principal.username, - ctx.principal.workspace_id, - params.message, - list(params.images or []), - conversation_id, - videos=list(params.videos or []), - model_key=params.model_key, - thinking_mode=params.thinking_mode, - run_mode=params.run_mode, - max_iterations=params.max_iterations, - session_data=ctx.to_session_data(), - message_source=params.message_source, - goal_mode=params.goal_mode, - skill_context_messages=list(params.skill_context_messages or []), - files=list(params.files or []), - task_type=params.task_type, - ) + return task_manager.create_chat_task(ctx, conversation_id=conversation_id) @staticmethod def _ensure_conversation_for_chat(ctx: RuntimeContext) -> Optional[str]: diff --git a/server/tasks/api.py b/server/tasks/api.py index abffa60a..4493029b 100644 --- a/server/tasks/api.py +++ b/server/tasks/api.py @@ -246,9 +246,9 @@ def get_task_api(task_id: str): "message": rec.message, "conversation_id": rec.conversation_id, "error": rec.error, - "message_source": (rec.session_data or {}).get("message_source"), - "goal_mode": bool((rec.session_data or {}).get("goal_mode")), - "goal_progress": (rec.session_data or {}).get("goal_progress"), + "message_source": rec.task_params.message_source, + "goal_mode": bool(rec.task_params.goal_mode), + "goal_progress": rec.goal_progress, "events": events, "next_offset": next_offset, "window_start": window_start, diff --git a/server/tasks/models.py b/server/tasks/models.py index 18014901..a06e2c31 100644 --- a/server/tasks/models.py +++ b/server/tasks/models.py @@ -15,6 +15,7 @@ from typing import Dict, Any, Optional, List # run_chat_task_sync 在使用点函数内延迟导入（见 _run_chat_task）。 from server.context import RuntimeIdentity, get_user_resources, ensure_conversation_loaded from server.main_task_gate import release_main_task_gate +from server.runtime.context import InternalDirectives, RuntimeContext, TaskParams, TrustedPrincipal from server.work_timer import finalize_conversation_work_timer from server.state import stop_flags from server.utils_common import debug_log, log_conn_diag @@ -48,7 +49,14 @@ class TaskRecord: "thinking_mode", "run_mode", "max_iterations", - "session_data", + # 三层结构化运行上下文（契约 docs/runtime_contract.md §4.1）：受理时由 + # RuntimeService.create_task 传入 RuntimeContext 并固化，任务线程直接 + # 按层读取——不再有 session_data 兼容快照 dict。 + "principal", + "task_params", + "directives", + # 运行期可变回写：goal_progress 事件最新快照（启动时为空，随事件流更新） + "goal_progress", "stop_requested", "next_event_idx", "runtime_pending_queue", @@ -60,40 +68,39 @@ class TaskRecord: def __init__( self, task_id: str, - username: str, - workspace_id: str, - message: str, + ctx: RuntimeContext, conversation_id: Optional[str], - model_key: Optional[str], - thinking_mode: Optional[bool], - run_mode: Optional[str], - max_iterations: Optional[int], - task_type: str = "chat", ): + principal = ctx.principal + params = ctx.params self.task_id = task_id - self.username = username - self.workspace_id = workspace_id + self.username = principal.username + self.workspace_id = principal.workspace_id self.status = "pending" self.created_at = time.time() self.updated_at = self.created_at - self.message = message + self.message = params.message + # conversation_id 经受理层补建兜底后显式传入（可能与 params.conversation_id 不同） self.conversation_id = conversation_id # 刷新恢复时前端会从事件流重建进行中的输出，1000 在长流式回复下会过早截断， # 导致“只恢复最后几个字符”。这里提高缓冲上限，优先保证重建完整性。 self.events: deque[Dict[str, Any]] = deque(maxlen=20000) self.thread: Optional[threading.Thread] = None self.error: Optional[str] = None - self.model_key = model_key - self.thinking_mode = thinking_mode - self.run_mode = run_mode - self.max_iterations = max_iterations - self.session_data: Dict[str, Any] = {} + self.model_key = params.model_key + self.thinking_mode = params.thinking_mode + self.run_mode = params.run_mode + self.max_iterations = params.max_iterations + self.principal: TrustedPrincipal = principal + self.task_params: TaskParams = params + self.directives: InternalDirectives = ctx.directives or InternalDirectives() + self.goal_progress: Optional[Dict[str, Any]] = None self.stop_requested: bool = False self.next_event_idx: int = 0 self.runtime_pending_queue: List[Dict[str, Any]] = [] self.runtime_guidance_queue: List[str] = [] self.last_cancel_at: Optional[float] = None - self.task_type = task_type + self.task_type = str(params.task_type or "chat") class TaskManager: """线程内存版任务管理器，后续可替换为 Redis/DB。""" @@ -128,29 +135,31 @@ class TaskManager: # ---- public APIs ---- def create_chat_task( self, - username: str, - workspace_id: str, - message: str, - images: List[Any], - conversation_id: Optional[str], - videos: Optional[List[Any]] = None, - model_key: Optional[str] = None, - thinking_mode: Optional[bool] = None, - run_mode: Optional[str] = None, - max_iterations: Optional[int] = None, - session_data: Optional[Dict[str, Any]] = None, - message_source: Optional[str] = None, - goal_mode: bool = False, - skill_context_messages: Optional[List[Dict[str, str]]] = None, - files: Optional[List[str]] = None, - task_type: str = "chat", + ctx: RuntimeContext, + conversation_id: Optional[str] = None, ) -> TaskRecord: + """受理一轮 Run：显式三层上下文 → 互斥裁决 → 登记 → 起执行线程。 + + 契约 docs/runtime_contract.md §4.1：``ctx`` 是唯一上下文来源（身份/参数/ + 内部指令三层），禁止隐式读取 Flask session。``conversation_id`` 允许覆盖 + params 中的值（受理层补建对话后传入），未传时取 params.conversation_id。 + """ + if ctx is None: + raise ValueError(tr("tasks.missing_session_data")) + ctx.validate() + principal = ctx.principal + params = ctx.params + username = principal.username + workspace_id = principal.workspace_id + run_mode = params.run_mode if run_mode: normalized = str(run_mode).lower() if normalized not in {"fast", "thinking", "deep"}: raise ValueError(tr("tasks.invalid_run_mode")) run_mode = normalized - normalized_task_type = str(task_type or "chat").strip().lower() or "chat" + if conversation_id is None: + conversation_id = params.conversation_id + normalized_task_type = str(params.task_type or "chat").strip().lower() or "chat" # 单对话互斥：普通 chat 任务禁止同一对话并发（防串写对话历史）； # 同工作区不同对话允许并行（对话级 terminal 隔离）。 # notice（通知触发）任务允许与已完成的 chat 任务共存，用于后台通知重入。 @@ -168,23 +177,15 @@ class TaskManager: if existing: raise RuntimeError(tr("tasks.task_already_running")) task_id = str(uuid.uuid4()) - record = TaskRecord(task_id, username, workspace_id, message, conversation_id, model_key, thinking_mode, run_mode, max_iterations, task_type=normalized_task_type) - # 运行上下文快照（RuntimeContext.to_session_data 产物，契约 docs/runtime_contract.md §4.1）。 - # 必须显式传入：禁止在受理层回退读 Flask session（隐式上下文在后台线程中 - # 不可靠且会静默丢身份）。调用方统一走 RuntimeService.create_task 构造快照。 - if session_data is None: - raise ValueError(tr("tasks.missing_session_data")) - snapshot = dict(session_data) - snapshot.setdefault("workspace_id", workspace_id) - if message_source is not None: - snapshot.setdefault("message_source", str(message_source)) - snapshot["goal_mode"] = bool(goal_mode) - if skill_context_messages: - snapshot["skill_context_messages"] = list(skill_context_messages) - record.session_data = snapshot + record = TaskRecord(task_id, ctx, conversation_id) + record.task_type = normalized_task_type with self._lock: self._tasks[task_id] = record - thread = threading.Thread(target=self._run_chat_task, args=(record, images, videos or [], files or []), daemon=True) + thread = threading.Thread( + target=self._run_chat_task, + args=(record, list(params.images or []), list(params.videos or []), list(params.files or [])), + daemon=True, + ) record.thread = thread record.status = "running" record.updated_at = time.time() @@ -754,7 +755,7 @@ class TaskManager: data.setdefault("workspace_id", rec.workspace_id) with self._lock: if event_type in {"goal_progress", "goal_completed", "goal_stopped"} and isinstance(data, dict): - rec.session_data["goal_progress"] = dict(data) + rec.goal_progress = dict(data) idx = getattr(rec, "next_event_idx", None) if idx is None: idx = rec.events[-1]["idx"] + 1 if rec.events else 0 @@ -774,19 +775,18 @@ class TaskManager: workspace = None stop_hint = False try: - # 显式运行上下文（契约 docs/runtime_contract.md §4.1）：身份与偏好快照在 - # 受理时由 RuntimeContext.to_session_data 固化，这里还原为 RuntimeIdentity - # 直接驱动资源装配——不再伪造 Flask 请求上下文（原 test_request_context - # 桥已拆除，任务线程全程无隐式上下文）。 - sd = rec.session_data or {} + # 显式运行上下文（契约 docs/runtime_contract.md §4.1）：身份与偏好快照 + # 受理时已固化为 rec.principal（TrustedPrincipal），这里直接映射为 + # RuntimeIdentity 驱动资源装配——任务线程全程无隐式上下文。 + p = rec.principal identity = RuntimeIdentity( - host_mode=bool(sd.get("host_mode")), - host_workspace_id=sd.get("host_workspace_id"), - is_api_user=bool(sd.get("is_api_user")), - role=sd.get("role"), - preferred_model_key=sd.get("model_key"), - preferred_run_mode=sd.get("run_mode"), - preferred_thinking_mode=sd.get("thinking_mode"), + host_mode=bool(p.host_mode), + host_workspace_id=p.host_workspace_id, + is_api_user=bool(p.is_api_user), + role=p.role, + preferred_model_key=p.preferred_model_key, + preferred_run_mode=p.preferred_run_mode, + preferred_thinking_mode=p.preferred_thinking_mode, ) if identity.host_mode: write_host_workspace_debug( @@ -865,10 +865,10 @@ class TaskManager: # 仅对“后台通知触发的新任务”补发 user_message 事件到任务事件流。 # 这样前端轮询能即时看到这条 user 消息，而不是刷新后才从历史中看到。 try: - if bool((rec.session_data or {}).get("auto_user_message_event")): + if rec.directives.auto_user_message_event: # 先回放本批「通知池」里的前置完成通知（除触发消息外的 N-1 条）， # 保证轮询客户端按时间顺序看到所有完成通知，且不各自触发新一轮工作。 - preceding_notices = (rec.session_data or {}).get("preceding_user_notices") or [] + preceding_notices = rec.directives.preceding_user_notices or [] if isinstance(preceding_notices, list): for item in preceding_notices: if not isinstance(item, dict): @@ -885,7 +885,7 @@ class TaskManager: } notice_event.update(notice_payload) self._append_event(rec, "user_message", notice_event) - extra_payload = (rec.session_data or {}).get("auto_user_message_payload") or {} + extra_payload = rec.directives.auto_user_message_payload or {} if not isinstance(extra_payload, dict): extra_payload = {} payload = { @@ -947,34 +947,34 @@ class TaskManager: setattr( terminal, "_auto_user_message_event", - bool((rec.session_data or {}).get("auto_user_message_event")), + bool(rec.directives.auto_user_message_event), ) setattr( terminal, "_auto_user_message_payload", - dict((rec.session_data or {}).get("auto_user_message_payload") or {}), + dict(rec.directives.auto_user_message_payload or {}), ) setattr( terminal, "_current_user_message_source", - str((rec.session_data or {}).get("message_source") or "user"), + str(rec.task_params.message_source or "user"), ) setattr( terminal, "_goal_mode_requested", - bool((rec.session_data or {}).get("goal_mode")), + bool(rec.task_params.goal_mode), ) setattr( terminal, "_skill_context_messages", - list((rec.session_data or {}).get("skill_context_messages") or []), + list(rec.task_params.skill_context_messages or []), ) # 审批/提问等待超时透传（契约 §6）：None = 保持既有默认语义（3600s）， # 超时后的语义属阶段三产品决策，本阶段仅建立透传机制。 setattr( terminal, "_approval_timeout_seconds", - (rec.session_data or {}).get("approval_timeout_seconds"), + rec.task_params.approval_timeout_seconds, ) except Exception: previous_auto_user_event = None @@ -999,7 +999,7 @@ class TaskManager: videos=videos, files=files or [], # 通知链任务认领轮询器预占的门闸（其余任务为 None，走竞争获取） - main_task_gate_token=(rec.session_data or {}).get("main_task_gate_token"), + main_task_gate_token=rec.directives.main_task_gate_token, ) finally: try: @@ -1095,10 +1095,10 @@ class TaskManager: # 清理 stop_flags stop_flags.pop(rec.task_id, None) # 主任务门闸兜底释放：若任务线程在 process_message_task 认领前异常退出， - # 按 session_data 中的 token 释放，避免门闸泄漏导致对话永久被占用。 + # 按受理时移交的 token 释放，避免门闸泄漏导致对话永久被占用。 # 正常路径下 process_message_task 已在 finally 释放，此处为无操作。 try: - gate_token = (rec.session_data or {}).get("main_task_gate_token") + gate_token = rec.directives.main_task_gate_token if gate_token and terminal: release_main_task_gate(terminal, gate_token) except Exception: @@ -1186,9 +1186,9 @@ def task_public_payload(rec: TaskRecord, *, include_title: bool = True) -> Dict[ "message": rec.message, "conversation_id": rec.conversation_id, "error": rec.error, - "message_source": (rec.session_data or {}).get("message_source"), - "goal_mode": bool((rec.session_data or {}).get("goal_mode")), - "goal_progress": (rec.session_data or {}).get("goal_progress"), + "message_source": rec.task_params.message_source, + "goal_mode": bool(rec.task_params.goal_mode), + "goal_progress": rec.goal_progress, "task_type": getattr(rec, "task_type", "chat"), } if include_title: diff --git a/server/workflow_flow.py b/server/workflow_flow.py index aca495d2..85ffbc9f 100644 --- a/server/workflow_flow.py +++ b/server/workflow_flow.py @@ -738,7 +738,7 @@ def emit_workflow_progress( conversation_id: Optional[str], extra: Optional[Dict[str, Any]] = None, ) -> None: - """广播工作流进度快照（sender → session_data → REST 轮询透传，对齐 goal 链路）。""" + """广播工作流进度快照（sender → 任务事件流 → REST 轮询透传，对齐 goal 链路）。""" if not callable(sender) or not conversation_id: return snap = wsm.progress_snapshot() diff --git a/test/test_runtime_service.py b/test/test_runtime_service.py index 1254af3b..e7b27fe8 100644 --- a/test/test_runtime_service.py +++ b/test/test_runtime_service.py @@ -4,7 +4,7 @@ - T01：显式上下文受理（无 HTTP 请求、无 test_request_context） - T02：同对话并发 chat 互斥 / notice 豁免 - T04：取消（受理层语义；执行线程以 no-op 替身阻断，不触达模型调用） -- 快照完整性：to_session_data 携带资源装配所需的全部身份与偏好字段 +- 上下文固化：受理后三层结构（principal/task_params/directives）原样落到任务记录 本测试全程不创建 Flask 应用/请求上下文——这本身就是 「公共入口不依赖隐式 Web 环境」的直接证明。 @@ -56,26 +56,26 @@ class RuntimeContextModelTest(unittest.TestCase): with self.assertRaises(ValueError): ctx.validate() - def test_to_session_data_carries_identity_and_preferences(self): + def test_context_layers_carry_identity_and_preferences(self): ctx = _make_ctx() - snap = ctx.to_session_data() - # 身份与资源范围 - self.assertEqual(snap["username"], "tester") - self.assertEqual(snap["workspace_id"], "default") - self.assertEqual(snap["role"], "user") - self.assertFalse(snap["is_api_user"]) - self.assertTrue(snap["host_mode"]) - self.assertEqual(snap["host_workspace_id"], "default") + # 身份与资源范围（principal 层） + p = ctx.principal + self.assertEqual(p.username, "tester") + self.assertEqual(p.workspace_id, "default") + self.assertEqual(p.role, "user") + self.assertFalse(p.is_api_user) + self.assertTrue(p.host_mode) + self.assertEqual(p.host_workspace_id, "default") # 偏好快照层（非本次覆盖） - self.assertEqual(snap["model_key"], "kimi-test") - self.assertEqual(snap["run_mode"], "fast") - self.assertFalse(snap["thinking_mode"]) - # 内部指令 - self.assertEqual(snap["main_task_gate_token"], "tok123") - # 默认不出现事件回放键 - self.assertNotIn("auto_user_message_event", snap) + self.assertEqual(p.preferred_model_key, "kimi-test") + self.assertEqual(p.preferred_run_mode, "fast") + self.assertFalse(p.preferred_thinking_mode) + # 内部指令（directives 层） + self.assertEqual(ctx.directives.main_task_gate_token, "tok123") + # 默认不回放事件 + self.assertFalse(ctx.directives.auto_user_message_event) - def test_to_session_data_directives(self): + def test_context_directives_fields(self): ctx = RuntimeContext( principal=_make_ctx().principal, params=TaskParams(message="m", conversation_id="c1", approval_timeout_seconds=120), @@ -85,12 +85,12 @@ class RuntimeContextModelTest(unittest.TestCase): preceding_user_notices=[{"message": "n1", "payload": {}}], ), ) - snap = ctx.to_session_data() - self.assertTrue(snap["auto_user_message_event"]) - self.assertEqual(snap["auto_user_message_payload"], {"visibility": "chat"}) - self.assertEqual(len(snap["preceding_user_notices"]), 1) - # 超时透传机制（默认 None 时不出现，语义不变） - self.assertEqual(snap["approval_timeout_seconds"], 120) + d = ctx.directives + self.assertTrue(d.auto_user_message_event) + self.assertEqual(d.auto_user_message_payload, {"visibility": "chat"}) + self.assertEqual(len(d.preceding_user_notices), 1) + # 超时透传机制（默认 None，显式设置后按层携带） + self.assertEqual(ctx.params.approval_timeout_seconds, 120) def test_principal_from_session_snapshot(self): snap = { @@ -134,9 +134,9 @@ class RuntimeServiceAdmissionTest(unittest.TestCase): self.assertTrue(rec.task_id) self.assertEqual(rec.username, "tester") self.assertEqual(rec.conversation_id, "conv_test_rt") - # 快照经显式上下文进入，未触碰 Flask session - self.assertEqual(rec.session_data["username"], "tester") - self.assertEqual(rec.session_data["main_task_gate_token"], "tok123") + # 三层上下文经显式入口固化到任务记录，未触碰 Flask session + self.assertEqual(rec.principal.username, "tester") + self.assertEqual(rec.directives.main_task_gate_token, "tok123") def test_t02_same_conversation_chat_mutex_and_notice_exempt(self): # 第一个任务保持 running（线程 no-op 但 status 已被置 running） @@ -158,12 +158,10 @@ class RuntimeServiceAdmissionTest(unittest.TestCase): with self.assertRaises(ValueError): runtime_service.create_task(bad) - def test_create_chat_task_requires_explicit_session_data(self): - # 直调底层入口且不带快照 → 明确拒绝（不再静默读 Flask session） + def test_create_chat_task_requires_explicit_context(self): + # 直调底层入口且不带显式上下文 → 明确拒绝（不再静默读 Flask session） with self.assertRaises(ValueError): - task_manager.create_chat_task( - "tester", "default", "msg", [], "conv_x", - ) + task_manager.create_chat_task(None) def test_t04_cancel_task(self): rec = self._create()