feat(v3): T28 操作审批流（dsh 式 allow-once/deny，fail-closed）

- ToolLoop 增 approval_hook：工具执行前挂起等待用户裁决，拒绝/异常折叠为
  失败结果回喂模型（可改道），fail-closed
- 策略 agent.approval_policy：off | dangerous（写/编辑/命令询问，只读放行，默认）| all；
  approval_timeout_s 超时自动拒绝（轮询实现，规避 portal 循环下 wait_for 定时器不可靠）
- POST /agent/{id}/approve 裁决端点；approval_request/decided 事件对进 SSE 与审计
- 前端：运行中审批卡（工具名+参数预览+拒绝/允许一次），composer 审批策略 chip
- 测试 +8（策略矩阵/拒绝回喂/允许执行/fail-closed/超时/端点分支），全量 289 passed

tests/test_agent_api.py

@@ -29,9 +29,10 @@ def agent_env(tmp_path, monkeypatch):
     # 快照用户真实设置，测试后原样恢复（settings.json 是活文件，不能污染）
     store = ga.settings_store()
     snapshot = json.loads(json.dumps(store._data, ensure_ascii=False))
-    # 工作区指向临时目录 + 给经典回退一个假 key（防止 .env 缺失时 400）
-    store.update({"agent": {"workspace_dir": str(tmp_path / "ws")},
-                  "architect": {"api_key": "sk-fake-test"}})
+    # 工作区指向临时目录 + 测试凭据走环境变量（monkeypatch 自动恢复）+ 审批默认关闭
+    monkeypatch.setenv("DEEPSEEK_API_KEY", "test-fake-credential-not-a-secret")
+    store.update({"agent": {"workspace_dir": str(tmp_path / "ws"),
+                            "approval_policy": "off"}})
     ga.rebuild_pipeline()
 
     script = []
@@ -519,3 +520,154 @@ def test_cancel_running_agent(agent_env, client, monkeypatch):
     st = client.get(f"/agent/{rid}/status").json()
     assert st["state"] == "failed" and st["error"] == "cancelled_by_user"
     assert _t.time() - t0 < 1.5
+
+
+# ---------------- 审批流（T28） ----------------
+
+def test_approval_service_level_timeout_and_deny(tmp_path):
+    """service 级闭环：dangerous 策略下写操作挂起 -> 超时自动拒绝 -> 模型收到拒绝结果。
+
+    说明：不走 TestClient——其每请求独立 portal 循环会冻结跨请求的后台任务，
+    无法真实测"挂起等待"；这里直接驱动 service.run（与网关 uvicorn 同构）。
+    """
+    import asyncio
+
+    async def scenario():
+        mp.reset_pool()
+        ag.reset_agent_service()
+        service = ag.AgentService(run_dir=tmp_path / "runs")
+        ag._service = service
+        ws = tmp_path / "ws"
+        info = service.register("agt01", "写 t.txt", "m", "",
+                                workspace=str(tmp_path / "ws"))
+        calls = []
+
+        async def chat(messages, tools_spec):
+            calls.append(1)
+            if len(calls) == 1:
+                return {"content": None,
+                        "tool_calls": [{"id": "c1", "name": "write_file",
+                                        "arguments": {"path": "t.txt", "content": "x"}}],
+                        "usage": {"prompt_tokens": 1, "completion_tokens": 1}}
+            return {"content": "了解，操作被拒绝。", "tool_calls": [],
+                    "usage": {"prompt_tokens": 1, "completion_tokens": 1}}
+
+        await service.run(info, chat, workspace_dir=str(ws),
+                          approval_policy="dangerous", approval_timeout_s=1)
+        return info, service.read_events("agt01")
+
+    info, evs = asyncio.run(scenario())
+    assert info.state == "done"
+    assert "拒绝" in info.response
+    kinds = [e["type"] for e in evs]
+    assert "approval_request" in kinds and "approval_decided" in kinds
+    decided = next(e for e in evs if e["type"] == "approval_decided")
+    assert decided["allowed"] is False
+    assert "超时" in decided.get("note", "")
+    assert not (tmp_path / "ws" / "t.txt").exists()          # fail-closed：未执行
+
+
+def test_approval_service_level_allow(tmp_path):
+    """service 级：审批请求挂起 -> 管理器裁决允许 -> 工具真实执行。"""
+    import asyncio
+
+    async def scenario():
+        mp.reset_pool()
+        ag.reset_agent_service()
+        service = ag.AgentService(run_dir=tmp_path / "runs2")
+        ag._service = service
+        info = service.register("agt02", "写 ok.txt", "m", "",
+                                workspace=str(tmp_path / "ws2"))
+        calls = []
+
+        async def chat(messages, tools_spec):
+            calls.append(1)
+            if len(calls) == 1:
+                return {"content": None,
+                        "tool_calls": [{"id": "c1", "name": "write_file",
+                                        "arguments": {"path": "ok.txt", "content": "v"}}],
+                        "usage": {"prompt_tokens": 1, "completion_tokens": 1}}
+            return {"content": "已写入 ok.txt。", "tool_calls": [],
+                    "usage": {"prompt_tokens": 1, "completion_tokens": 1}}
+
+        task = asyncio.create_task(
+            service.run(info, chat, workspace_dir=str(tmp_path / "ws2"),
+                        approval_policy="dangerous", approval_timeout_s=10))
+        # 等审批请求出现 -> 模拟用户点「允许一次」
+        approval_id = None
+        for _ in range(100):
+            evs = service.read_events("agt02")
+            asks = [e for e in evs if e["type"] == "approval_request"]
+            if asks:
+                approval_id = asks[0]["id"]
+                break
+            await asyncio.sleep(0.05)
+        assert approval_id, "应出现审批请求"
+        getattr(info, "_approval_manager").decide(approval_id, True)
+        await task
+        return info, service.read_events("agt02")
+
+    info, evs = asyncio.run(scenario())
+    assert info.state == "done"
+    decided = next(e for e in evs if e["type"] == "approval_decided")
+    assert decided["allowed"] is True
+    assert (tmp_path / "ws2" / "ok.txt").read_text(encoding="utf-8") == "v"
+
+
+def test_approval_endpoint_branches(agent_env, client):
+    """approve 端点：未知任务 404；无审批流程 409。"""
+    assert client.post("/agent/ghost/approve",
+                       json={"approval_id": "x", "allowed": True}).status_code == 404
+    # 正常任务（无挂起审批）-> 管理器存在但审批单不存在 -> 404
+    agent_env["set_script"]([
+        {"content": "直接回答。", "tool_calls": [], "usage": {}},
+    ])
+    r = client.post("/agent", json={"task": "hi"})
+    rid = r.json()["request_id"]
+    _wait_done(agent_env["service"], rid)
+    r2 = client.post(f"/agent/{rid}/approve",
+                     json={"approval_id": "nope", "allowed": True})
+    assert r2.status_code in (404, 409)
+
+
+def test_approval_policy_matrix(agent_env):
+    from gateway.agent import needs_approval
+    assert not needs_approval("off", "run_command")
+    assert not needs_approval("dangerous", "read_file")
+    assert needs_approval("dangerous", "write_file")
+    assert needs_approval("dangerous", "run_command")
+    assert needs_approval("all", "list_dir")
+
+
+def test_approval_timeout_auto_deny_service_level(tmp_path):
+    """审批超时 = 自动拒绝（fail-closed）：service 级闭环（TestClient 不支持跨请求挂起）。"""
+    import asyncio
+
+    async def scenario():
+        ag.reset_agent_service()
+        service = ag.AgentService(run_dir=tmp_path / "runs3")
+        ag._service = service
+        info = service.register("agt03", "写 t.txt", "m", "",
+                                workspace=str(tmp_path / "ws3"))
+        calls = []
+
+        async def chat(messages, tools_spec):
+            calls.append(1)
+            if len(calls) == 1:
+                return {"content": None,
+                        "tool_calls": [{"id": "c1", "name": "write_file",
+                                        "arguments": {"path": "t.txt", "content": "x"}}],
+                        "usage": {"prompt_tokens": 1, "completion_tokens": 1}}
+            return {"content": "了解，操作被拒绝。", "tool_calls": [],
+                    "usage": {"prompt_tokens": 1, "completion_tokens": 1}}
+
+        await service.run(info, chat, workspace_dir=str(tmp_path / "ws3"),
+                          approval_policy="dangerous", approval_timeout_s=1)
+        return info, service.read_events("agt03")
+
+    info, evs = asyncio.run(scenario())
+    assert info.state == "done"
+    decided = [e for e in evs if e["type"] == "approval_decided"]
+    assert decided and decided[0]["allowed"] is False
+    assert "超时" in decided[0].get("note", "")
+    assert not (tmp_path / "ws3" / "t.txt").exists()