refactor(runtime): 拆除 session_data 兼容桥，任务上下文三层结构化直传

- 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）

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 步：闭环疑点与小修复（不依赖架构决策）
 

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": "系统未初始化",

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:

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

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(

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]:

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,

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:

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()

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()