diff --git a/cache_research/gateway/gateway_current_state.md b/cache_research/gateway/gateway_current_state.md
index f46c5b07..c168ad56 100644
--- a/cache_research/gateway/gateway_current_state.md
+++ b/cache_research/gateway/gateway_current_state.md
@@ -246,16 +246,19 @@ v1 原文把「内存态」直接判为「状态唯一 Owner 的障碍」，混
 
 > 总原则：协议草案与接口适配可迭代推进；**每一步保留现有门闸、保存和恢复保障**；四层不要求四进程（G13）；Remote Worker 不是本轮完成条件。
 
-> **实施状态速览（2026-09-07，commit d9bd599c / 27aeed70 / 4378eb6a 起）**：
-> | 步骤 | 状态 | 证据 |
-> |---|---|---|
-> | 第 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 同契约接入为后续项（§7.5-补） |
-> | 第 5 步 | ✅ 收窄完成（代码就绪度） | 公共入口覆盖：run.*/approval.*/session.* 全部进入 RuntimeService（协议 §3/§4）；正式客户端开发不在本轮 |
-> | 遗留 | ⬜ | N3 真实环境人工验证、N4/N5 疑点、③↔④ 全量贯通（§7.5-补）、定时任务（§7.6 后置） |
+> **实施状态速览（2026-09-07，三档口径：接口存在 / 适配完成 / 行为验收）**：
+> | 能力 | 接口存在 | 适配完成（既有入口转调） | 行为验收 |
+> |---|---|---|---|
+> | run.start / cancel / guide / queue | ✅ | ✅（tasks/api.py、api_v1.py 全部转调） | ✅ chain 双客户端 + lifecycle |
+> | run.get / run.events（含 window_start） | ✅ | ✅（HTTP 轮询透传水位） | ✅ offset/水位断言 |
+> | run.list（发现，审核 F1） | ✅ | ✅（/api/tasks 列表、running-status、api_v1 删除保护转调） | ✅ B 发现 A 的活动 Run 并取消 |
+> | approval.list / resolve | ✅ | ✅（chat/approval.py 三类六个路由转调） | ✅ 等待→批准→继续（approval_wait） |
+> | session.list / history（principal 用户+工作区双校验，F3） | ✅ | ⬜（Web 会话路由未动——属 conversation 域存量链路） | ✅ 含越权/跨工作区拒绝断言 |
+> | Gateway 独立初始化 | — | — | ✅ 子进程隔离（ASTRION_IGNORE_DOTENV 逃生门 + 假模型自包含 + 装配证据断言，F4） |
+> | ExecutionBackend 替身（E1-E4） | ✅ | —（默认 None，生产路径不变） | ✅ fake_exec |
+> | ③↔④ 全量贯通（Host/Docker 迁入、E5-E10） | ⬜ | ⬜ | ⬜ 后置（§7.5-补） |
+>
+> **审核收口记录（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`（现有真实链路不变）。
 
diff --git a/cache_research/gateway/gateway_implementation_review_2026-09-07.md b/cache_research/gateway/gateway_implementation_review_2026-09-07.md
new file mode 100644
index 00000000..6cc40a67
--- /dev/null
+++ b/cache_research/gateway/gateway_implementation_review_2026-09-07.md
@@ -0,0 +1,115 @@
+# Gateway 改造实现审核记录
+
+> 审核日期：2026-09-07
+> 审核方式：主智能体综合审阅，3 个 Luna 子智能体分别核查服务初始化、协议与多端交互、执行环境边界。
+> 审核基线：`b4deee0f`；目标提交：`d7cb854f`。结论只描述该提交快照，后续修复须重新验证。
+> 本轮只审核并记录，未修改业务代码。本文不替换既有规划或子报告。
+
+## 1. 范围与结论
+
+本次审核以下 4 个提交：
+
+| 提交 | 内容 |
+|---|---|
+| `d9bd599c` | Gateway 独立启动与公共协议落地 |
+| `27aeed70` | Execution Contract 与替身执行器 |
+| `4378eb6a` | 会话查询公共入口与代码就绪度收口 |
+| `d7cb854f` | 审批/提问公共入口与①②③链路收口声明 |
+
+对照依据：[当前状态规划](gateway_current_state.md) §7、[公共协议](../../docs/runtime_protocol.md)、[执行契约](../../docs/execution_contract.md)，以及前序关于四层职责、公共入口、状态归属和行为验收的约定。
+
+**结论：改造已有实质进展，但现有证据尚不能支持“①客户端—②Gateway—③Runtime 已完整贯通”的完成声明。** 当前完成的是主要进程内服务能力、独立初始化相关拆分和 E1–E4 替身接入；公共任务发现、既有适配入口收敛、资源范围校验及验收可靠性仍需收口。
+
+按规划中记录的本轮收口范围，生产 Host/Docker 迁入 ExecutionBackend、E5–E10 全量接入、正式 CLI/GUI/IDE 客户端开发均已后置。本文不把这些已明确后置的工作算作本轮违约，也不要求事件全量持久化、重启续跑或 Remote Worker。
+
+## 2. 完成度核查
+
+| 能力 | 审核结论 | 证据与限制 |
+|---|---|---|
+| 任务核心与 Web 路由加载拆分 | 已实现 | `server/tasks/__init__.py` 不再顶层装配 Blueprint 与路由；`server/tasks/web.py` 承担 Web 装配 |
+| 非 Web 环境的消息输出与后台任务启动 | 已提供适配 | `server/extensions.py` 增加 `emit_event` / `run_background`；独立生命周期仍需加强验收 |
+| Run、审批、会话查询公共方法 | 主要方法已实现 | `server/runtime/service.py`；存在 F1、F3，且 Web 调用尚未全部迁移（F2） |
+| 事件 offset 与窗口水位 | 进程内接口已实现 | `get_task_events` 返回 `meta.window_start`；现有 HTTP 轮询未透传该信息 |
+| 多客户端发现、操作与恢复闭环 | 部分完成，未充分验收 | 新客户端缺活动 Run 发现入口；测试没有覆盖真实 A 发起/B 操作和断连重连 |
+| CLI/GUI/IDE 独立接入 | 部分服务基础已具备 | 进程内直调成功不等于已有独立传输适配器；正式客户端后置可以接受，但应准确限定“代码就绪度” |
+| ExecutionBackend | E1–E4 替身接入已实现 | 默认 `execution_backend=None`；生产 Host/Docker 与 E5–E10 后置，符合当前范围记录 |
+
+## 3. 本轮需要收口的发现
+
+以下均列为 **P2：本轮完成验收前应修正或明确处理的事项**。其中 F1/F2 是能力与迁移缺口，F3 是新增公共接口的资源校验缺口，F4 是验证缺陷；不应全部表述为线上行为回归。
+
+### F1：新客户端无法通过公共入口发现已有任务
+
+- **位置**：`server/runtime/service.py:277`、`docs/runtime_protocol.md:49–56`。
+- **现状**：RuntimeService 只有按已知 task_id 查询的 `get_task` / `get_task_events`，没有 `list_runs` / `list_tasks`。底层 `TaskManager.list_tasks` 已存在，旧 Web `/api/tasks` 也提供列表，但没有提升到公共服务。
+- **触发场景**：客户端 A 发起任务后，客户端 B 新连接，尚不知道该 task_id。B 无法仅通过当前公共协议发现 A 的活动 Run，仍需旧 Web 列表接口或其他协议外信息。
+- **影响**：仅验证 `get_task(task_id)`，不足以证明“B 经公共入口查询并操作”已经成立。
+- **建议**：增加受授权的 Run 发现查询，可按工作区、会话和活动状态筛选；复用底层能力，返回适合公共协议的结果。
+- **验收**：A 创建 Run；B 只持已授权身份及工作区/会话范围，通过公共查询取得该 Run，再观察、回答审批或取消，不从 A 的测试变量直接获取 task_id。
+
+### F2：既有适配入口未统一，新恢复信息未到达实际客户端
+
+- **位置**：`server/tasks/api.py:236/281/304/332/355/377`、`server/api_v1.py:424/432/453`、`server/chat/approval.py:56/71/100/115/143/159`。
+- **现状**：Web/API 的事件轮询、取消、引导、队列操作仍有直接调用 task_manager 的路径；三类审批查询/回答仍直接调用对应 manager。
+- **具体差异**：RuntimeService 的事件查询新增 `meta.window_start`，旧 HTTP 轮询仍用 `get_events_since` 构造响应，不包含该水位。新增协议的缺口检测信息尚未到达这些客户端。
+- **影响**：“公共方法已提供”不能等同于“既有入口均已转调”。多条路径仍可独立演进，新增语义也不会自动共享。直接调用同一 manager 不等于出现多份状态，但不满足本轮入口收敛约定。
+- **建议**：相关 Web/API 路由转调公共服务，保留现有认证及兼容响应；将事件窗口信息适配到响应，并验证消费者的缺口处理。更新完成速览，区分接口存在、适配完成和行为验收。
+- **验收**：同一组受理、取消、队列和审批行为，经 Web 适配与直接服务调用得到等价结果；裁剪事件窗口后，调用方能检测缺口并按已定义规则重新同步，不仅验证未裁剪时 window_start=0。
+
+### F3：会话查询未检查 principal 绑定的工作区
+
+- **位置**：`server/runtime/service.py:250–270`，重点为 `:254–269`。
+- **现状**：`_resources_for_query` 只检查 `principal.username == username`，没有检查 principal 的 workspace_id 与查询目标一致，随后直接按调用参数的 workspace_id 装配资源。
+- **已复现**：主侧用资源解析替身调用 `_resources_for_query('alice', 'B', TrustedPrincipal(username='alice', workspace_id='A'))`，得到 `resource_workspace='B'`，没有被拒绝。
+- **影响**：服务接口未守住 TrustedPrincipal 所声明的资源范围。当前是进程内可信调用接口，不能据此断言已有外部可利用漏洞；但新适配器将外部查询参数传入时，这个不一致必须被阻止或经明确重新授权处理。
+- **建议**：查询目标与已授权 principal 范围保持一致；若允许切换工作区，应显式完成目标授权，而不是直接信任第二份 workspace_id 参数。
+- **验收**：覆盖同名用户不同工作区、用户名不一致、正确范围查询，以及 host/API/web 身份。当前测试只检查用户名不一致。
+
+### F4：独立验收测试存在假通过、隔离及配置依赖
+
+- **位置**：`test/runtime_standalone_checks.py:25–27/83–144/147–258`、`config/__init__.py:51–52`。
+- **假通过**：生命周期测试接受任意 failed 终态；terminal 不存在时跳过门闸检查。主侧在缺少模型配置、资源装配失败的环境中，观察到该生命周期测试仍通过。
+- **交互覆盖不足**：chain 测试在 Run 结束后手工向 manager 创建审批，再验证公共入口路由。这能验证查询、裁决和重复回答的服务行为，但不能证明“执行中的任务进入等待→另一客户端回答→原任务继续”。提问与计划确认的完整等待链同样缺少对应证据。
+- **隔离缺口**：测试设置临时 ASTRION_DATA_ROOT 后，config 仍会用源码目录 `.env` 的同名值覆盖。子审核在当前工作区执行时，观察到尝试访问真实运行态目录并被写权限拒绝；因此原隔离设置不能作为可靠保证。
+- **配置依赖**：主侧在无 `.env`、无真实凭证的隔离源码副本中执行，fake_exec 因“未配置可用模型”失败，chain 因没有建立会话 ID 失败。测试未提供自包含的模型配置与受控响应。
+- **建议**：提供确定的测试资源根、模型配置和模型响应替身，保留真实任务线程与资源装配。分别断言成功、取消、审批等待/继续、预期异常及清理结果，不用任意 failed 替代成功验收；强制确认 terminal、会话历史和门闸状态。
+- **验收**：全新隔离目录、无真实凭证、禁用网络时可稳定复现；改变开发者 `.env` 不改变测试数据路径；人为使装配失败时，成功生命周期用例必须失败。
+
+## 4. 已后置执行层工作的契约提醒
+
+`modules/execution_plane/base.py:29` 声称路径授权与 `_validate_command` 已在 Runtime 编排层完成，但当前真实命令校验仍位于旧 `terminal_ops` 链路，文件路径检查也不能只依赖“先读后写”前置检查。注入 backend 的新分支未完整复用这些旧后端校验。
+
+**当前默认生产路径未因此改变，不将其报告为现有 Host/Docker 权限绕过。** 但未来接入真实 ExecutionBackend 前，必须准确分配并保留命令校验、路径授权和实际执行限制；不能按照当前注释误认为上层已完成全部检查。
+
+此项跟随已后置的③↔④工作处理，不作为本轮①②③收口的额外扩张。
+
+## 5. 验证记录与限制
+
+主侧测试方法：从目标 HEAD 导出临时源码副本（不带未跟踪 `.env`），指定临时数据/部署配置目录，移除继承的模型凭证，并通过测试进程的网络连接拦截禁止外部调用。未修改被审核源码。运行器使用仓库 `.venv/bin/python` 的 unittest discovery，未依赖 pytest。
+
+运行以下 5 组测试，共 **28 项：26 通过、2 失败**：
+
+- `test_server_refactor_smoke.py`
+- `test_runtime_service.py`
+- `test_runtime_identity_resources.py`
+- `test_conversation_model_persistence.py`
+- `test_runtime_standalone_lifecycle.py`
+
+| 失败用例 | 观察结果 | 解释 |
+|---|---|---|
+| `ExecutionPlaneFakeBackendTest.test_runtime_with_fake_execution_backend` | get_default_model_key 抛“未配置可用模型” | 未进入替身工具断言，不能认定工具实现回归；证明测试配置不自包含 |
+| `ProtocolSmokeChainTest.test_full_chain_run_history_offset_approval` | “Run 应建立会话 id”断言失败 | 本次隔离环境下会话装配未成功，不能作为全链路完成证据 |
+
+对照观察：同一环境下 `StandaloneRuntimeLifecycleTest` 通过，结合其接受 failed 及可跳过 terminal 检查的逻辑，确认 F4 的验收假通过问题。
+
+另外完成了 F3 的最小替身复现、适配入口静态对照及最近提交范围核验。未执行真实模型调用、真实多客户端联调或 UI 人工验收；不据上述测试结果声称这些场景已通过，也不将这两项失败直接归为生产业务回归。
+
+## 6. 建议收口顺序
+
+1. 补公共 Run 发现查询，修复会话查询的工作区范围校验。
+2. 将本轮承诺的 Web/API 控制、审批与事件观察入口转调公共服务，落实事件缺口响应与处理。
+3. 修复测试隔离和自包含配置，用受控模型响应验证真实装配、执行中审批/提问、取消及清理。
+4. 用两个相互独立的客户端调用方完成 A 发起、B 发现并操作、断连及重连验收；传输适配尚未实现时，明确记录进程内调用的实际覆盖范围。
+5. 根据证据更新 `gateway_current_state.md` 完成速览；其余生产执行后端与正式客户端工作继续按已后置范围推进。
+
+完成上述收口后，再将“主要服务方法已具备”升级为“本轮①②③边界已按约定验收”。
diff --git a/config/__init__.py b/config/__init__.py
index e4a3b190..fbc1f775 100644
--- a/config/__init__.py
+++ b/config/__init__.py
@@ -16,6 +16,10 @@ def _load_dotenv():
     import sys
     pre_existing_keys = set(os.environ.keys())
 
+    # 隔离/测试环境逃生门：显式禁用仓库根 .env 加载（含 ASTRION_DATA_ROOT 的
+    # .env 优先覆盖）。默认行为不变；仅测试子进程等需要完全自控环境的场景使用。
+    _ignore_dotenv = os.environ.get("ASTRION_IGNORE_DOTENV", "").strip().lower() in {"1", "true", "yes", "on"}
+
     # 1) 仓库根 .env（开发便利，不覆盖已有的环境变量，但 ASTRION_DATA_ROOT
     #    作为项目数据根目录必须优先以 .env 为准，避免外部 shell 误指到 clone）
     if getattr(sys, 'frozen', False):
@@ -29,6 +33,8 @@ def _load_dotenv():
         env_exists = env_path.exists()
     except Exception:
         env_exists = False
+    if _ignore_dotenv:
+        env_exists = False
     if env_exists:
         try:
             for raw_line in env_path.read_text(encoding="utf-8").splitlines():
diff --git a/modules/execution_plane/base.py b/modules/execution_plane/base.py
index d812c69a..859b3c84 100644
--- a/modules/execution_plane/base.py
+++ b/modules/execution_plane/base.py
@@ -25,9 +25,12 @@ from typing import Any, Dict, List, Optional, Protocol, runtime_checkable
 class ExecutionBackend(Protocol):
     """执行环境后端协议。实现方：FakeExecutionBackend（替身）、未来 Host/Docker 适配器。
 
-    注意：本协议只承诺「执行语义」，不承诺安全边界——权限裁决（permission mode）、
-    路径授权（_validate_path）、命令校验（_validate_command）在 Runtime 编排层完成，
-    与本接口正交（契约 §4）。
+    注意：本协议只承诺「执行语义」，不承诺安全边界。当前权限裁决（permission mode）
+    与审批介入在 Runtime 编排层完成；但**命令校验（_validate_command/FORBIDDEN_COMMANDS）
+    与路径授权（_validate_path/禁读清单）仍由旧 terminal_ops/file_manager 链路承担**——
+    经本接口注入的后端分支并不会自动获得这些校验。未来接入真实 ExecutionBackend
+    （Host/Docker 迁入）时，必须准确分配并保留这些校验，不能按「上层已完成全部检查」
+    误读（审核 §4）。OS 层强制力（Seatbelt/bwrap/DAC+Landlock）是最终边界。
     """
 
     async def run_command(
diff --git a/server/api_v1.py b/server/api_v1.py
index 3c43a24f..9bd68c47 100644
--- a/server/api_v1.py
+++ b/server/api_v1.py
@@ -104,8 +104,8 @@ def delete_workspace_api(workspace_id: str):
     ws_id = _sanitize_workspace_id(workspace_id)
     if not ws_id:
         return jsonify({"success": False, "error": tr("api_v1.workspace_id_invalid")}), 400
-    # 阻止删除有运行中任务的工作区
-    running = [t for t in task_manager.list_tasks(username, ws_id) if t.status in {"pending", "running"}]
+    # 阻止删除有运行中任务的工作区（经公共服务发现，协议 run.list）
+    running = runtime_service.list_runs(username, ws_id, status="pending,running")
     if running:
         return jsonify({"success": False, "error": tr("api_v1.workspace_has_running_tasks")}), 409
     removed = state.api_user_manager.delete_workspace(username, ws_id)
@@ -421,16 +421,17 @@ def delete_conversation_api(workspace_id: str, conv_id: str):
 @api_token_required
 def get_task_events(task_id: str):
     username = session.get("username")
-    rec = task_manager.get_task(username, task_id)
+    rec = runtime_service.get_task(username, task_id)
     if not rec:
         return jsonify({"success": False, "error": tr("api_v1.task_not_found")}), 404
     try:
         offset = int(request.args.get("from", 0))
     except Exception:
         offset = 0
-    # 工作线程会持续追加事件，必须持锁快照，不能直接迭代 rec.events
-    events = task_manager.get_events_since(rec, offset)
-    next_offset = events[-1]["idx"] + 1 if events else offset
+    # 经公共服务读取事件流（含 window_start 缺口检测水位，协议 §5.2）
+    events, next_offset, _ev_err, ev_meta = runtime_service.get_task_events(username, task_id, offset)
+    events = events or []
+    next_offset = next_offset if next_offset is not None else offset
     return jsonify({
         "success": True,
         "data": {
@@ -442,6 +443,7 @@ def get_task_events(task_id: str):
             "error": rec.error,
             "events": events,
             "next_offset": next_offset,
+            "window_start": (ev_meta or {}).get("window_start", 0),
         }
     })
 
@@ -450,7 +452,7 @@ def get_task_events(task_id: str):
 @api_token_required
 def cancel_task_api_v1(task_id: str):
     username = session.get("username")
-    ok = task_manager.cancel_task(username, task_id)
+    ok = runtime_service.cancel_task(username, task_id)
     if not ok:
         return jsonify({"success": False, "error": tr("api_v1.task_not_found")}), 404
     return jsonify({"success": True})
diff --git a/server/chat/approval.py b/server/chat/approval.py
index aab280dc..130f38d3 100644
--- a/server/chat/approval.py
+++ b/server/chat/approval.py
@@ -38,7 +38,7 @@ from server.context import with_terminal, get_gui_manager, get_upload_guard, bui
 from server.security import rate_limited, prune_socket_tokens
 from server.utils_common import debug_log
 from server.state import PROJECT_MAX_STORAGE_MB, pending_socket_tokens, SOCKET_TOKEN_TTL_SECONDS
-from server.state import tool_approval_manager, user_question_manager, plan_approval_manager
+from server.runtime import runtime_service
 from server.extensions import socketio
 from server.monitor import get_cached_monitor_snapshot
 
@@ -53,7 +53,7 @@ def list_pending_user_questions(terminal: WebTerminal, workspace: UserWorkspace,
     requested_conv_id = (request.args.get("conversation_id") or "").strip() or None
     if requested_conv_id is None:
         requested_conv_id = getattr(terminal.context_manager, "current_conversation_id", None)
-    items = user_question_manager.list_pending(username=username, conversation_id=requested_conv_id)
+    items = runtime_service.list_pending_approvals(username, requested_conv_id, kind="question")["question"]
     return jsonify({
         "success": True,
         "items": items,
@@ -68,9 +68,10 @@ def answer_user_question(terminal: WebTerminal, workspace: UserWorkspace, userna
     """提交 ask_user 工具问题的回答。"""
     data = request.get_json() or {}
     try:
-        item = user_question_manager.answer(
-            question_id=question_id,
+        item = runtime_service.resolve_approval(
+            "question",
             username=username,
+            item_id=question_id,
             selected_option_id=data.get("selected_option_id"),
             text=data.get("text"),
             dismissed=bool(data.get("dismissed")),
@@ -97,7 +98,7 @@ def list_pending_plan_approvals(terminal: WebTerminal, workspace: UserWorkspace,
     requested_conv_id = (request.args.get("conversation_id") or "").strip() or None
     if requested_conv_id is None:
         requested_conv_id = getattr(terminal.context_manager, "current_conversation_id", None)
-    items = plan_approval_manager.list_pending(username=username, conversation_id=requested_conv_id)
+    items = runtime_service.list_pending_approvals(username, requested_conv_id, kind="plan")["plan"]
     return jsonify({
         "success": True,
         "items": items,
@@ -112,9 +113,10 @@ def answer_plan_approval(terminal: WebTerminal, workspace: UserWorkspace, userna
     """提交计划批准/拒绝决策。approved=true 时工具循环侧会自动切换到 execute 模式。"""
     data = request.get_json() or {}
     try:
-        item = plan_approval_manager.answer(
-            approval_id=approval_id,
+        item = runtime_service.resolve_approval(
+            "plan",
             username=username,
+            item_id=approval_id,
             approved=bool(data.get("approved")),
             comment=data.get("comment"),
         )
@@ -140,7 +142,7 @@ def list_pending_tool_approvals(terminal: WebTerminal, workspace: UserWorkspace,
     requested_conv_id = (request.args.get("conversation_id") or "").strip() or None
     if requested_conv_id is None:
         requested_conv_id = getattr(terminal.context_manager, "current_conversation_id", None)
-    items = tool_approval_manager.list_pending(username=username, conversation_id=requested_conv_id)
+    items = runtime_service.list_pending_approvals(username, requested_conv_id, kind="tool")["tool"]
     return jsonify({
         "success": True,
         "items": items,
@@ -156,7 +158,9 @@ def decide_tool_approval(terminal: WebTerminal, workspace: UserWorkspace, userna
     data = request.get_json() or {}
     decision = str(data.get("decision") or "").strip().lower()
     try:
-        item = tool_approval_manager.decide(approval_id=approval_id, username=username, decision=decision)
+        item = runtime_service.resolve_approval(
+            "tool", username=username, item_id=approval_id, decision=decision
+        )
     except ValueError as exc:
         return jsonify({"success": False, "error": str(exc)}), 400
     except KeyError:
diff --git a/server/conversation_bootstrap.py b/server/conversation_bootstrap.py
index 8e449da8..6acf119e 100644
--- a/server/conversation_bootstrap.py
+++ b/server/conversation_bootstrap.py
@@ -17,7 +17,7 @@ from flask import Blueprint, jsonify, request, session
 from server.auth_helpers import api_login_required, get_current_username
 from server.context import get_user_resources
 from server.tasks import task_manager
-from server.tasks.helpers import _task_public_payload
+from server.tasks.models import task_public_payload
 from server.utils_common import debug_log
 from modules.i18n import tr
 
@@ -252,7 +252,7 @@ def bootstrap_conversation(conversation_id: str):
         replay = _decide_task_replay(messages, events)
         replay["task_id"] = main_rec.task_id
         # 任务摘要与全量事件：前端恢复时免去 GET /api/tasks 与 GET /api/tasks/{id} 两次请求
-        replay["task"] = _task_public_payload(main_rec)
+        replay["task"] = task_public_payload(main_rec)
         replay["events"] = events
         data["task_replay"] = replay
         debug_log(
diff --git a/server/runtime/service.py b/server/runtime/service.py
index 50ff521b..0c929a5f 100644
--- a/server/runtime/service.py
+++ b/server/runtime/service.py
@@ -256,6 +256,10 @@ class RuntimeService:
         if principal.username != username:
             # 纵深防御：principal 是适配层认证后的可信身份，不得与查询目标身份不符
             raise PermissionError("runtime_context: principal 与查询目标用户不一致")
+        if str(principal.workspace_id or "") != str(workspace_id or ""):
+            # 资源范围一致性：principal 声明的 workspace 即授权范围，
+            # 跨工作区查询必须重新认证构造新 principal，不得仅传第二份参数
+            raise PermissionError("runtime_context: principal 与查询目标工作区不一致")
         identity = RuntimeIdentity(
             host_mode=principal.host_mode,
             host_workspace_id=principal.host_workspace_id,
@@ -274,6 +278,46 @@ class RuntimeService:
 
     # ---- 观察（内部接口；后台调用方不必为观察任务发 HTTP 请求）----
 
+    def list_runs(
+        self,
+        username: str,
+        workspace_id: Optional[str] = None,
+        *,
+        conversation_id: Optional[str] = None,
+        status: Optional[str] = None,
+    ) -> List[Dict[str, Any]]:
+        """Run 发现查询（run.list）：按工作区/会话/状态筛选，created_at 倒序。
+
+        客户端 B 不知 task_id 时经本接口发现活动 Run（审核 F1）。载荷与 Web
+        /api/tasks 同构（复用 _task_public_payload），保证多端字段一致。
+        status 过滤语义同 Web 路由："active" = pending/running/cancel_requested，
+        或逗号分隔状态集；None 不过滤。归属约束：仅返回 username 自己的 Run。
+        """
+        from server.tasks import task_manager
+        from server.tasks.models import task_public_payload
+
+        recs = task_manager.list_tasks(username, workspace_id)
+        if conversation_id:
+            target = str(conversation_id)
+            recs = [r for r in recs if str(getattr(r, "conversation_id", None) or "") in {target, target[5:] if target.startswith("conv_") else f"conv_{target}"}]
+        if status:
+            normalized = str(status).strip().lower()
+            if normalized == "active":
+                active = {"pending", "running", "cancel_requested"}
+                recs = [r for r in recs if r.status in active]
+            else:
+                wanted = {part.strip() for part in normalized.split(",") if part.strip()}
+                recs = [r for r in recs if r.status in wanted]
+        recs = sorted(recs, key=lambda x: x.created_at, reverse=True)
+        # 载荷与 Web /api/tasks 完全同构（核心层唯一序列化实现，无新旧双轨）
+        return [task_public_payload(r) for r in recs]
+
+    def get_runtime_pending_messages(self, username: str, task_id: str):
+        """追问队列查询（run.queue 的查询面）。"""
+        from server.tasks import task_manager
+
+        return task_manager.get_runtime_pending_messages(username, task_id)
+
     def get_task(self, username: str, task_id: str):
         from server.tasks import task_manager
 
diff --git a/server/tasks/api.py b/server/tasks/api.py
index a06397a6..abffa60a 100644
--- a/server/tasks/api.py
+++ b/server/tasks/api.py
@@ -25,7 +25,6 @@ from modules.goal_state_manager import GoalStateManager, REASON_USER_CANCEL
 from server.tasks import task_manager
 from server.runtime import RuntimeContext, TaskParams, principal_from_session_snapshot, runtime_service
 from server.tasks.skills import _build_skill_context_messages
-from server.tasks.helpers import _task_public_payload
 from server.tasks.media import _normalize_media_payload, _normalize_files_payload
 from modules.i18n import tr
 
@@ -36,21 +35,11 @@ from modules.i18n import tr
 def list_tasks_api():
     username = get_current_username()
     workspace_id = (request.args.get("workspace_id") or "").strip() or None
-    status_filter = (request.args.get("status") or "").strip().lower()
-    recs = task_manager.list_tasks(username, workspace_id)
-    if status_filter:
-        if status_filter == "active":
-            active = {"pending", "running", "cancel_requested"}
-            recs = [rec for rec in recs if rec.status in active]
-        else:
-            wanted = {part.strip() for part in status_filter.split(",") if part.strip()}
-            recs = [rec for rec in recs if rec.status in wanted]
+    status_filter = (request.args.get("status") or "").strip().lower() or None
+    # 筛选/排序/序列化统一由公共服务承担（run.list，协议 §4）
     return jsonify({
         "success": True,
-        "data": [
-            _task_public_payload(r)
-            for r in sorted(recs, key=lambda x: x.created_at, reverse=True)
-        ]
+        "data": runtime_service.list_runs(username, workspace_id, status=status_filter)
     })
 
 @tasks_bp.route("/api/conversations/<conversation_id>/running-status", methods=["GET"])
@@ -67,19 +56,14 @@ def get_conversation_running_status_api(conversation_id: str):
     if not conversation_id:
         return jsonify({"success": False, "error": tr("tasks.missing_conversation_id")}), 400
 
-    # ① 主 task：内存中该对话是否有活动任务（取最新一条）
-    active_statuses = {"pending", "running", "cancel_requested"}
-    main_rec = None
-    for rec in task_manager.list_tasks(username):
-        if rec.conversation_id != conversation_id or rec.status not in active_statuses:
-            continue
-        if main_rec is None or rec.created_at > main_rec.created_at:
-            main_rec = rec
+    # ① 主 task：内存中该对话是否有活动任务（list_runs 已按会话筛选 + created_at 倒序）
+    active_runs = runtime_service.list_runs(username, conversation_id=conversation_id, status="active")
+    main_rec = active_runs[0] if active_runs else None
 
     # ②③④ 需要 terminal（sub_agent_manager / background_command_manager 挂在 terminal 上）
     workspace_id = (
         (request.args.get("workspace_id") or "").strip()
-        or (main_rec.workspace_id if main_rec else "")
+        or (main_rec["workspace_id"] if main_rec else "")
         or (session.get("workspace_id") or "")
     ) or None
     bg_status = {
@@ -101,8 +85,8 @@ def get_conversation_running_status_api(conversation_id: str):
         "data": {
             "conversation_id": conversation_id,
             "is_main_running": is_main_running,
-            "main_task_id": main_rec.task_id if main_rec else None,
-            "main_task_type": getattr(main_rec, "task_type", "chat") if main_rec else None,
+            "main_task_id": main_rec["task_id"] if main_rec else None,
+            "main_task_type": (main_rec.get("task_type") or "chat") if main_rec else None,
             **bg_status,
             "is_truly_active": is_main_running or any(bg_status.values()),
         }
@@ -222,7 +206,7 @@ def get_task_api(task_id: str):
     started_at = time.time()
     username = get_current_username()
     poll_req_id = request.headers.get("X-Task-Poll", "-")
-    rec = task_manager.get_task(username, task_id)
+    rec = runtime_service.get_task(username, task_id)
     if not rec:
         log_conn_diag(
             f"task-poll-missing req={poll_req_id} user={username} task_id={task_id}"
@@ -232,9 +216,11 @@ def get_task_api(task_id: str):
         offset = int(request.args.get("from", 0))
     except Exception:
         offset = 0
-    # 工作线程会持续追加事件，必须持锁快照，不能直接迭代 rec.events
-    events = task_manager.get_events_since(rec, offset)
-    next_offset = events[-1]["idx"] + 1 if events else offset
+    # 经公共服务读取事件流（含 window_start 缺口检测水位，协议 §5.2）
+    events, next_offset, _ev_err, ev_meta = runtime_service.get_task_events(username, task_id, offset)
+    events = events or []
+    next_offset = next_offset if next_offset is not None else offset
+    window_start = (ev_meta or {}).get("window_start", 0)
     elapsed_ms = (time.time() - started_at) * 1000.0
     should_log = (
         offset == 0
@@ -265,7 +251,8 @@ def get_task_api(task_id: str):
             "goal_progress": (rec.session_data or {}).get("goal_progress"),
             "events": events,
             "next_offset": next_offset,
-            "runtime_queued_messages": task_manager.get_runtime_pending_messages(
+            "window_start": window_start,
+            "runtime_queued_messages": runtime_service.get_runtime_pending_messages(
                 username, task_id
             ),
         }
@@ -275,10 +262,10 @@ def get_task_api(task_id: str):
 @api_login_required
 def cancel_task_api(task_id: str):
     username = get_current_username()
-    rec = task_manager.get_task(username, task_id)
+    rec = runtime_service.get_task(username, task_id)
     if not rec:
         return jsonify({"success": False, "error": tr("tasks.task_not_found")}), 404
-    ok = task_manager.cancel_task(username, task_id)
+    ok = runtime_service.cancel_task(username, task_id)
     # 用户取消任务时，一并停止该工作区的目标模式，避免后续新对话继承旧目标。
     try:
         if ok and rec.workspace_id:
@@ -301,7 +288,7 @@ def enqueue_runtime_guidance_api(task_id: str):
     if not message:
         return jsonify({"success": False, "error": tr("tasks.guidance_content_empty")}), 400
 
-    result = task_manager.enqueue_runtime_guidance(username, task_id, message)
+    result = runtime_service.enqueue_runtime_guidance(username, task_id, message)
     if not result.get("success"):
         code = result.get("code") or "runtime_guidance_failed"
         if code == "task_not_found":
@@ -329,7 +316,7 @@ def enqueue_runtime_queue_message_api(task_id: str):
     if not message:
         return jsonify({"success": False, "error": tr("tasks.message_empty")}), 400
     files = _normalize_files_payload(payload.get("files"))
-    result = task_manager.enqueue_runtime_pending_message(username, task_id, message, files=files)
+    result = runtime_service.enqueue_runtime_pending_message(username, task_id, message, files=files)
     if not result.get("success"):
         code = result.get("code") or "runtime_queue_enqueue_failed"
         if code == "task_not_found":
@@ -352,7 +339,7 @@ def enqueue_runtime_queue_message_api(task_id: str):
 @api_login_required
 def delete_runtime_queue_message_api(task_id: str, message_id: str):
     username = get_current_username()
-    result = task_manager.remove_runtime_pending_message(username, task_id, message_id)
+    result = runtime_service.remove_runtime_pending_message(username, task_id, message_id)
     if not result.get("success"):
         code = result.get("code") or "runtime_queue_delete_failed"
         if code == "task_not_found":
@@ -374,7 +361,7 @@ def delete_runtime_queue_message_api(task_id: str, message_id: str):
 @api_login_required
 def guide_runtime_queue_message_api(task_id: str, message_id: str):
     username = get_current_username()
-    result = task_manager.promote_runtime_pending_to_guidance(username, task_id, message_id)
+    result = runtime_service.promote_runtime_pending_to_guidance(username, task_id, message_id)
     if not result.get("success"):
         code = result.get("code") or "runtime_queue_guide_failed"
         if code == "task_not_found":
diff --git a/server/tasks/helpers.py b/server/tasks/helpers.py
index 9e75ba54..72ed4247 100644
--- a/server/tasks/helpers.py
+++ b/server/tasks/helpers.py
@@ -28,61 +28,3 @@ SKILL_FIELD_RE = re.compile(r"^(?P<key>name|description)\s*:\s*(?P<value>.*)$")
 
 
 
-def _conversation_title_for_task(rec: TaskRecord) -> Optional[str]:
-    conv_id = (getattr(rec, "conversation_id", None) or "").strip()
-    if not conv_id:
-        return None
-    if not conv_id.startswith("conv_"):
-        conv_id = f"conv_{conv_id}"
-    try:
-        if rec.username == "host":
-            path = Path(DATA_DIR).expanduser().resolve() / "conversations" / rec.workspace_id / f"{conv_id}.json"
-        else:
-            import server.state as state
-            workspace = state.user_manager.ensure_user_workspace(rec.username, rec.workspace_id or "default")
-            path = Path(workspace.data_dir).expanduser().resolve() / "conversations" / f"{conv_id}.json"
-        if not path.exists():
-            return None
-        data = json.loads(path.read_text(encoding="utf-8"))
-        title = str(data.get("title") or "").strip()
-        return title or None
-    except Exception:
-        return None
-
-def _workspace_label_for_task(rec: TaskRecord) -> Optional[str]:
-    try:
-        if rec.username == "host":
-            from modules.host_workspace_manager import load_host_workspace_catalog
-            catalog = load_host_workspace_catalog()
-            for item in catalog.get("workspaces") or []:
-                if item.get("workspace_id") == rec.workspace_id:
-                    return str(item.get("label") or rec.workspace_id)
-        else:
-            import server.state as state
-            item = state.user_manager.list_user_workspaces(rec.username).get(rec.workspace_id)
-            if item:
-                return str(item.get("label") or rec.workspace_id)
-    except Exception:
-        pass
-    return rec.workspace_id
-
-def _task_public_payload(rec: TaskRecord, *, include_title: bool = True) -> Dict[str, Any]:
-    payload = {
-        "task_id": rec.task_id,
-        "username": rec.username,
-        "workspace_id": rec.workspace_id,
-        "workspace_label": _workspace_label_for_task(rec),
-        "status": rec.status,
-        "created_at": rec.created_at,
-        "updated_at": rec.updated_at,
-        "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"),
-        "task_type": getattr(rec, "task_type", "chat"),
-    }
-    if include_title:
-        payload["conversation_title"] = _conversation_title_for_task(rec)
-    return payload
diff --git a/server/tasks/models.py b/server/tasks/models.py
index 7d612fd8..18014901 100644
--- a/server/tasks/models.py
+++ b/server/tasks/models.py
@@ -1126,3 +1126,71 @@ def start_task_cleanup_scheduler():
     thread = threading.Thread(target=cleanup_loop, daemon=True, name="TaskCleanup")
     thread.start()
     debug_log("[Task] 任务清理定时器已启动")
+
+
+# ---- 任务公开载荷序列化（核心层唯一实现；Web 路由与公共入口共用，保证多端同构）----
+
+
+def conversation_title_for_task(rec: TaskRecord) -> Optional[str]:
+    conv_id = (getattr(rec, "conversation_id", None) or "").strip()
+    if not conv_id:
+        return None
+    if not conv_id.startswith("conv_"):
+        conv_id = f"conv_{conv_id}"
+    try:
+        if rec.username == "host":
+            path = Path(DATA_DIR).expanduser().resolve() / "conversations" / rec.workspace_id / f"{conv_id}.json"
+        else:
+            import server.state as state
+
+            workspace = state.user_manager.ensure_user_workspace(rec.username, rec.workspace_id or "default")
+            path = Path(workspace.data_dir).expanduser().resolve() / "conversations" / f"{conv_id}.json"
+        if not path.exists():
+            return None
+        data = json.loads(path.read_text(encoding="utf-8"))
+        title = str(data.get("title") or "").strip()
+        return title or None
+    except Exception:
+        return None
+
+
+def workspace_label_for_task(rec: TaskRecord) -> Optional[str]:
+    try:
+        if rec.username == "host":
+            from modules.host_workspace_manager import load_host_workspace_catalog
+
+            catalog = load_host_workspace_catalog()
+            for item in catalog.get("workspaces") or []:
+                if item.get("workspace_id") == rec.workspace_id:
+                    return str(item.get("label") or rec.workspace_id)
+        else:
+            import server.state as state
+
+            item = state.user_manager.list_user_workspaces(rec.username).get(rec.workspace_id)
+            if item:
+                return str(item.get("label") or rec.workspace_id)
+    except Exception:
+        pass
+    return rec.workspace_id
+
+
+def task_public_payload(rec: TaskRecord, *, include_title: bool = True) -> Dict[str, Any]:
+    payload = {
+        "task_id": rec.task_id,
+        "username": rec.username,
+        "workspace_id": rec.workspace_id,
+        "workspace_label": workspace_label_for_task(rec),
+        "status": rec.status,
+        "created_at": rec.created_at,
+        "updated_at": rec.updated_at,
+        "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"),
+        "task_type": getattr(rec, "task_type", "chat"),
+    }
+    if include_title:
+        payload["conversation_title"] = conversation_title_for_task(rec)
+    return payload
diff --git a/test/runtime_standalone_checks.py b/test/runtime_standalone_checks.py
index 10b7d4ff..1f5670f5 100644
--- a/test/runtime_standalone_checks.py
+++ b/test/runtime_standalone_checks.py
@@ -25,6 +25,9 @@ _SMOKE_WORKSPACE.mkdir(parents=True, exist_ok=True)
 os.environ["ASTRION_DATA_ROOT"] = str(_SMOKE_ROOT)
 os.environ["DEPLOY_CONFIG_DIR"] = str(_SMOKE_ROOT / "config")
 os.environ["TERMINAL_SANDBOX_MODE"] = "host"
+# 隔离逃生门（审核 F4）：仓库根 .env 对 ASTRION_DATA_ROOT 有「.env 优先」的刻意覆盖，
+# 测试进程必须显式禁用它，否则隔离目录会被 .env 值穿透（静默落到真实/clone 数据根）。
+os.environ["ASTRION_IGNORE_DOTENV"] = "1"
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
@@ -42,6 +45,41 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
     encoding="utf-8",
 )
 
+# 自包含模型配置（审核 F4）：装配需要至少一个已注册模型；指向 127.0.0.1:9（discard
+# 端口，几乎必无监听）使模型调用以「连接拒绝」快速失败——不依赖 DNS/外部网络，
+# 也不需要真实凭证。验收对象是装配与生命周期，不是模型响应。
+(_SMOKE_ROOT / "config" / "custom_models.json").write_text(
+    json.dumps(
+        {
+            "models": [
+                {
+                    "model_name": "fake-smoke-model",
+                    "visible": True,
+                    "url": "http://127.0.0.1:9",
+                    "apikey": "fake-smoke-key",
+                    "reasoning_capability": "fast,thinking",
+                    "thinkmode_status": {"type": "param_toggle", "model_id": "fake-smoke-model"},
+                }
+            ]
+        },
+        ensure_ascii=False,
+        indent=2,
+    ),
+    encoding="utf-8",
+)
+
+# 隔离生效断言（审核 F4）：config import 后路径常量必须落在隔离根内，
+# 否则说明隔离被穿透（如 .env 覆盖）——宁可明确失败，绝不静默使用真实目录。
+import config as _smoke_config  # noqa: E402
+
+_assert_root = str(_SMOKE_ROOT.resolve())
+assert str(Path(_smoke_config.DATA_DIR).resolve()).startswith(_assert_root), (
+    f"隔离失败：DATA_DIR={_smoke_config.DATA_DIR} 不在 {_assert_root} 下"
+)
+assert str(Path(_smoke_config.DEPLOY_CONFIG_DIR).resolve()).startswith(_assert_root), (
+    f"隔离失败：DEPLOY_CONFIG_DIR={_smoke_config.DEPLOY_CONFIG_DIR} 不在 {_assert_root} 下"
+)
+
 
 def _make_ctx(conversation_id=None, message="协议链路验收 ping"):
     from server.runtime import (
@@ -97,24 +135,34 @@ def check_lifecycle():
     from server.runtime import runtime_service as rs  # 单例别名，语义强调
 
     deadline = time.time() + 30
-    saw_events = False
+    saw_assembly = False  # 装配证据：api_request_start 事件（历史/请求构造完成，模型调用前）
     final_status = None
+    assembly_error = None
     while time.time() < deadline:
         events, _next_offset, err, _meta = rs.get_task_events("gw_smoke_user", task_id, 0)
         assert err is None, f"get_task_events 应可读: {err}"
-        if events:
-            saw_events = True
+        event_types = {e.get("type") for e in (events or [])}
+        # 装配失败（资源/会话/请求构造）会以 error 事件终结任务，必须与模型失败区分：
+        # api_request_start 之前出现 error = 装配失败，验收必须明确失败（审核 F4 假通过修复）
+        if "api_request_start" not in event_types and "error" in event_types:
+            assembly_error = [e for e in events if e.get("type") == "error"][-1]
+            break
+        if "api_request_start" in event_types:
+            saw_assembly = True
         rec_now = rs.get_task("gw_smoke_user", task_id)
         status = getattr(rec_now, "status", None)
         if status in {"succeeded", "failed", "stopped", "canceled", "cancel_requested"}:
             final_status = status
             break
         # 装配完成后主动取消（避免真实模型调用慢等；装配此时已真实发生）
-        if saw_events:
+        if saw_assembly:
             rs.cancel_task("gw_smoke_user", task_id)
         time.sleep(0.5)
 
-    assert saw_events, "任务线程应真实启动并产生事件（装配真实发生）"
+    assert assembly_error is None, (
+        f"装配阶段失败（非模型调用）：{assembly_error}"
+    )
+    assert saw_assembly, "任务线程应真实完成装配（出现 api_request_start 事件）"
 
     deadline = time.time() + 20
     while time.time() < deadline:
@@ -127,15 +175,16 @@ def check_lifecycle():
         f"任务应到达终态（模型失败属预期），实际: {final_status}"
     )
 
-    # 门闸最终释放（对话应可受理下一任务）
+    # 门闸最终释放（对话应可受理下一任务）；terminal 必须存在——
+    # 跳过检查会掩盖装配失败（审核 F4 假通过修复）
     import server.context.resources as resources
 
     term_key = "host::gwsmoke::" + str(getattr(rec_now, "conversation_id", "") or "")
     terminal = resources.state.user_terminals.get(term_key)
-    if terminal is not None:
-        from server.main_task_gate import is_main_task_gate_busy
+    assert terminal is not None, "装配成功后对话级 terminal 必须存在（否则装配未真实发生）"
+    from server.main_task_gate import is_main_task_gate_busy
 
-        assert not is_main_task_gate_busy(terminal), "任务终态后门闸必须释放（否则下一任务无法受理）"
+    assert not is_main_task_gate_busy(terminal), "任务终态后门闸必须释放（否则下一任务无法受理）"
 
     # 全程无 Flask app 初始化（socketio 未绑定 app）
     assert not has_app_context(), "验收结束仍应无 Flask app 上下文"
@@ -148,9 +197,42 @@ def check_chain():
     """第 2 步验收：极简协议客户端视角全链路（工作区→会话→运行→停止→历史→审批）。"""
     from server.runtime import runtime_service
 
-    # 1. 第一个 Run：受理 → 终态（模型失败属预期；等 api_request_start 确保历史已写）
+    # 1. 双客户端场景（审核 F1 验收）：A 受理 Run；B 不持有 task_id，仅凭身份+
+    # 工作区经公共入口发现 A 的活动 Run → 观察事件流 → 取消；A 的 Run 到终态。
     rec1 = runtime_service.create_task(_make_ctx())
-    rec1 = _wait_terminal_state(rec1.task_id, cancel_on_event="api_request_start")
+
+    # B 视角：轮询 list_runs 直到发现该 Run（按 message 识别，不读 A 侧变量）
+    discovered = None
+    deadline = time.time() + 20
+    while time.time() < deadline:
+        active = runtime_service.list_runs("gw_smoke_user", "gwsmoke", status="active")
+        found = [r for r in active if "协议链路验收 ping" in (r.get("message") or "")]
+        if found:
+            discovered = found[0]
+            break
+        if getattr(rec1, "status", None) in {"succeeded", "failed", "stopped", "canceled"}:
+            break
+        time.sleep(0.4)
+    assert discovered, "B 应能经 list_runs 发现 A 发起的活动 Run（F1）"
+    assert discovered["task_id"] == rec1.task_id, "发现结果应指向同一 Run"
+
+    # B 观察事件流（用发现的 task_id），等 user_message 落盘证据后取消
+    deadline = time.time() + 20
+    saw_user_msg = False
+    while time.time() < deadline:
+        ev_b, _, err_b, _meta_b = runtime_service.get_task_events("gw_smoke_user", discovered["task_id"], 0)
+        assert err_b is None, "B 应可读 A 的事件流"
+        if any(e.get("type") == "user_message" for e in (ev_b or [])):
+            saw_user_msg = True
+            break
+        if any(e.get("type") == "error" for e in (ev_b or [])):
+            raise AssertionError(f"装配阶段失败（非模型调用）: {ev_b[-1] if ev_b else None}")
+        time.sleep(0.4)
+    assert saw_user_msg, "B 应观察到 A 的 user_message 事件（历史已落盘）"
+
+    # B 取消 A 的 Run（跨端操作），任务到终态
+    assert runtime_service.cancel_task("gw_smoke_user", discovered["task_id"]), "B 应可取消 A 的 Run"
+    rec1 = _wait_terminal_state(discovered["task_id"])
     assert rec1.status in {"succeeded", "failed", "stopped", "canceled"}, f"Run1 应达终态: {rec1.status}"
     conv_id = rec1.conversation_id
     assert conv_id, "Run 应建立会话 id（服务层补建对话，conversation_id 同步受理返回）"
@@ -213,6 +295,28 @@ def check_chain():
         pass
     else:
         raise AssertionError("principal 与查询目标不一致必须抛 PermissionError")
+    # F3：principal 声明的工作区与查询目标不一致也必须拒绝（同名用户跨工作区）
+    try:
+        runtime_service.get_session_history(
+            "gw_smoke_user", "gwsmoke", conv_id,
+            principal=_TP(username="gw_smoke_user", workspace_id="other_ws"),
+        )
+    except PermissionError:
+        pass
+    else:
+        raise AssertionError("principal 工作区与查询目标不一致必须抛 PermissionError")
+
+    # 5.6 Run 发现公共入口（F1）：仅凭身份+工作区可列出活动/全部 Run
+    runs_all = runtime_service.list_runs("gw_smoke_user", "gwsmoke")
+    run_ids = {r["task_id"] for r in runs_all}
+    assert {rec1.task_id, rec2.task_id} <= run_ids, "list_runs 应含两个 Run"
+    assert all(r["username"] == "gw_smoke_user" for r in runs_all), "归属过滤"
+    runs_conv = runtime_service.list_runs("gw_smoke_user", "gwsmoke", conversation_id=conv_id)
+    assert {r["task_id"] for r in runs_conv} == {rec1.task_id, rec2.task_id}, "会话筛选"
+    runs_done = runtime_service.list_runs("gw_smoke_user", "gwsmoke", status="active")
+    assert not any(r["task_id"] in run_ids for r in runs_done), "终态后 active 筛选应为空"
+    payload_keys = set(runs_all[0].keys())
+    assert {"task_id", "status", "conversation_id", "task_type", "created_at"} <= payload_keys
 
     # 6. 审批语义：公共入口 list_pending_approvals / resolve_approval
     # （manager 层语义已有覆盖，这里验收公共入口路由与错误语义透传）
@@ -325,6 +429,110 @@ def check_fake_exec():
     assert getattr(terminal2, "execution_backend", "MISSING") is None, "默认应为 None（真实链路）"
 
 
+def check_approval_wait():
+    """审核 F4 交互覆盖：执行中审批等待 → 另一调用方经公共入口回答 → 执行继续。
+
+    驱动真实工具编排层（_execute_tool_calls_impl + approval 权限模式），不经模型
+    （模型只是 tool_calls 的生产者，与等待/裁决/继续语义无关）；执行环境为替身
+    （零真实副作用）。覆盖：审批创建事件发出 → 公共入口可发现 → 批准 → 工具继续执行。
+    """
+    import asyncio
+    import threading
+
+    from modules.execution_plane import FakeExecutionBackend
+    from server.chat_flow_tool_loop import _execute_tool_calls_impl
+    from server.context import RuntimeIdentity, get_user_resources
+    from server.runtime import runtime_service
+
+    identity = RuntimeIdentity(host_mode=True, host_workspace_id="gwsmoke", is_api_user=False, role="admin")
+    ws_term, workspace = get_user_resources(
+        "gw_smoke_user", workspace_id="gwsmoke", update_session=False, identity=identity
+    )
+    conv_id = ws_term.context_manager.conversation_manager.create_conversation(
+        project_path=str(workspace.project_path), run_mode="fast",
+        thinking_mode=False, model_key="fake-smoke-model",
+    )
+    terminal, _ = get_user_resources(
+        "gw_smoke_user", workspace_id="gwsmoke", update_session=False,
+        conversation_id=conv_id, identity=identity,
+    )
+    backend = FakeExecutionBackend()
+    terminal.execution_backend = backend
+    # 默认 work_mode=plan 会把权限锁为只读（AGENTS.md §10.6）；审批链路验收需 approval 档
+    terminal.set_work_mode("execute", persist=False, conversation_id=conv_id)
+    terminal.set_permission_mode("approval", persist=False, conversation_id=conv_id)
+
+    sent_events = []
+
+    def sender(event_type, data):
+        sent_events.append((event_type, data))
+
+    tool_calls = [{
+        "id": "tc_wait_1",
+        "type": "function",
+        "function": {"name": "run_command", "arguments": json.dumps({"command": "echo approved-cmd", "timeout": 5})},
+    }]
+    loop_result = {}
+
+    async def _noop_process(**_kwargs):
+        return None
+
+    def _get_stop_flag(*_args, **_kwargs):
+        return None
+
+    def _clear_stop_flag(*_args, **_kwargs):
+        return None
+
+    def run_loop():
+        async def _main():
+            return await _execute_tool_calls_impl(
+                web_terminal=terminal,
+                tool_calls=tool_calls,
+                sender=sender,
+                messages=[],
+                client_sid="approval_wait_sid",
+                username="gw_smoke_user",
+                iteration=1,
+                conversation_id=conv_id,
+                last_tool_call_time=time.time(),
+                process_sub_agent_updates=_noop_process,
+                process_background_command_updates=_noop_process,
+                get_stop_flag=_get_stop_flag,
+                clear_stop_flag=_clear_stop_flag,
+                workspace=workspace,
+            )
+        try:
+            loop_result["value"] = asyncio.run(_main())
+        except Exception as exc:
+            loop_result["error"] = exc
+
+    th = threading.Thread(target=run_loop, daemon=True)
+    th.start()
+
+    # 主线程扮演「另一调用方」：经公共入口发现待决审批并批准
+    deadline = time.time() + 20
+    approved = False
+    while time.time() < deadline:
+        pending = runtime_service.list_pending_approvals("gw_smoke_user", conv_id).get("tool", [])
+        if pending:
+            runtime_service.resolve_approval(
+                "tool", username="gw_smoke_user",
+                item_id=pending[0]["approval_id"], decision="approved",
+            )
+            approved = True
+            break
+        if "error" in loop_result:
+            break
+        time.sleep(0.3)
+    th.join(timeout=20)
+
+    assert "error" not in loop_result, f"工具循环异常: {loop_result.get('error')}"
+    assert approved, "执行中的任务应产生待决审批（另一调用方可经公共入口发现）"
+    assert any(t == "tool_approval_required" for t, _ in sent_events), "应发出 tool_approval_required 事件"
+    assert any(c["op"] == "run_command" for c in backend.calls), "批准后工具应继续执行（替身收到命令）"
+    assert not th.is_alive(), "批准后工具循环应退出"
+
+
 def main():
     mode = sys.argv[1] if len(sys.argv) > 1 else ""
     if mode == "lifecycle":
@@ -333,8 +541,10 @@ def main():
         check_chain()
     elif mode == "fake_exec":
         check_fake_exec()
+    elif mode == "approval_wait":
+        check_approval_wait()
     else:
-        print("usage: runtime_standalone_checks.py <lifecycle|chain|fake_exec>", file=sys.stderr)
+        print("usage: runtime_standalone_checks.py <lifecycle|chain|fake_exec|approval_wait>", file=sys.stderr)
         return 2
     print(f"CHECK_OK {mode}")
     return 0
diff --git a/test/test_runtime_standalone_lifecycle.py b/test/test_runtime_standalone_lifecycle.py
index 0f738e28..80f306ed 100644
--- a/test/test_runtime_standalone_lifecycle.py
+++ b/test/test_runtime_standalone_lifecycle.py
@@ -59,5 +59,12 @@ class ExecutionPlaneFakeBackendTest(unittest.TestCase):
         _run_check("fake_exec")
 
 
+class ApprovalWaitChainTest(unittest.TestCase):
+    """审核 F4 交互覆盖：执行中审批等待→公共入口回答→执行继续（真实工具编排层）。"""
+
+    def test_approval_wait_resolve_continue(self):
+        _run_check("approval_wait")
+
+
 if __name__ == "__main__":
     unittest.main()