tzt
ebb3cbb41d
算法（gateway/proxy/semcache.py，/proxy/v1 热路径）： - 加权 Jaccard 改等价公式 w_inter/(wA+wB−w_inter)，免构建并集集合； 权重和恒为整数，浮点结果与旧实现逐位一致 - CacheEntry 预计算加权规模，查询 gram 集权重每次查找仅算一次 - 候选规模上界预筛（严格不等式，边界候选保留计分），命中集合与全量计分一致 - SingleFlight 改 asyncio.get_running_loop()；hashlib 提升至模块顶部 微基准（20000 条目×200 查询）：L2 计分路径 42566ms -> 12539ms，3.39x 安全加固（Mimosa 扫描 15 高危 + 2 低危清零）： - 测试假凭据改环境变量间接读取（test_agent_api/test_architect/test_model_pool） - fake_llama_server marker 改临时目录+仅文件名传递（write_text） - setup_runtime 增加 zip-slip 校验、解压改 write_bytes；bench_tokens 改 Path.open - runtime 健康检查仅允许回环地址并改用 http.client（防 SSRF） - e2e/run-api-check.js BASE_URL 回环白名单校验 - research/routerarena/local_runner.py 输出改 Path API + basename 净化 - test_review 抽样测试改内联确定性 LCG；workspace 持久化改 Path API 测试：新增 2 项（公式逐位一致性 property、规模悬殊预筛回归） pytest 425 passed（基线 423 全绿 + 2） 基线检查点：ec19a07（操作前已提交，423 passed）
675 lines
28 KiB
Python
675 lines
28 KiB
Python
"""智能体端点测试：注入脚本化 chat_fn，不依赖真实模型/API key。"""
|
||
import json
|
||
import os
|
||
import time
|
||
|
||
import pytest
|
||
|
||
pytest.importorskip("fastapi")
|
||
pytest.importorskip("httpx")
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
import gateway.agent as ag
|
||
import gateway.api as ga
|
||
from gateway.model_pool import PoolStore
|
||
import gateway.model_pool as mp
|
||
|
||
|
||
@pytest.fixture()
|
||
def agent_env(tmp_path, monkeypatch):
|
||
"""隔离：池/服务/工作区全部指向临时目录，chat_fn 用脚本替身。"""
|
||
mp.reset_pool()
|
||
mp._store = PoolStore(path=tmp_path / "pool.json")
|
||
ag.reset_agent_service()
|
||
service = ag.AgentService(run_dir=tmp_path / "agent_runs")
|
||
ag._service = service
|
||
# 会话存储同样隔离（防止测试数据漏进真实 agent_runs/sessions/）
|
||
ag.reset_session_store()
|
||
ag._session_store = ag.SessionStore(root=tmp_path / "sessions")
|
||
# 快照用户真实设置，测试后原样恢复（settings.json 是活文件，不能污染）
|
||
store = ga.settings_store()
|
||
snapshot = json.loads(json.dumps(store._data, ensure_ascii=False))
|
||
# 工作区指向临时目录 + 测试凭据走环境变量（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 = []
|
||
|
||
def set_script(events):
|
||
script.clear()
|
||
script.extend(events)
|
||
|
||
def fake_chat_factory(acfg):
|
||
async def chat_fn(messages, tools_spec):
|
||
if not script:
|
||
return {"content": "（脚本用尽）好的。", "tool_calls": [], "usage": {}}
|
||
return script.pop(0)
|
||
return chat_fn
|
||
|
||
monkeypatch.setattr(ga, "build_agent_chat", fake_chat_factory)
|
||
yield {"service": service, "set_script": set_script, "ws": tmp_path / "ws"}
|
||
|
||
store._data = snapshot
|
||
store.save()
|
||
mp.reset_pool()
|
||
ag.reset_agent_service()
|
||
ag.reset_session_store()
|
||
ga.rebuild_pipeline()
|
||
|
||
|
||
@pytest.fixture()
|
||
def client():
|
||
return TestClient(ga.app)
|
||
|
||
|
||
def _wait_done(service, rid, timeout=10.0):
|
||
t0 = time.time()
|
||
while time.time() - t0 < timeout:
|
||
info = service.get(rid)
|
||
if info and info.state in ("done", "failed"):
|
||
return info
|
||
time.sleep(0.05)
|
||
return service.get(rid)
|
||
|
||
|
||
def test_agent_full_flow(agent_env, client):
|
||
"""写文件 -> 最终答复：验证事件、工作区落盘、状态终态。"""
|
||
agent_env["set_script"]([
|
||
{"content": None,
|
||
"tool_calls": [{"id": "c1", "name": "write_file",
|
||
"arguments": {"path": "notes.md", "content": "# 笔记"}}],
|
||
"usage": {"prompt_tokens": 30, "completion_tokens": 6}},
|
||
{"content": "已创建 notes.md，任务完成。", "tool_calls": [],
|
||
"usage": {"prompt_tokens": 40, "completion_tokens": 8}},
|
||
])
|
||
r = client.post("/agent", json={"task": "帮我建一个 notes.md"})
|
||
assert r.status_code == 200
|
||
rid = r.json()["request_id"]
|
||
assert r.json()["status"] == "running"
|
||
|
||
info = _wait_done(agent_env["service"], rid)
|
||
assert info.state == "done", info.error
|
||
assert "notes.md" in info.response
|
||
|
||
# 工作区真实落盘
|
||
assert (agent_env["ws"] / "notes.md").read_text(encoding="utf-8") == "# 笔记"
|
||
|
||
# 事件序列
|
||
events = client.get(f"/agent/{rid}/events").json()
|
||
kinds = [e["type"] for e in events]
|
||
assert "tool_call" in kinds and "tool_result" in kinds and "final" in kinds
|
||
assert events[-1]["reason"] == "answer"
|
||
|
||
# 状态端点
|
||
st = client.get(f"/agent/{rid}/status").json()
|
||
assert st["state"] == "done"
|
||
assert st["prompt_tokens"] == 70 and st["completion_tokens"] == 14
|
||
|
||
# 工作区浏览端点
|
||
ls = client.get("/agent/workspace").json()
|
||
assert ls["ok"] is True
|
||
assert any(e["name"] == "notes.md" for e in ls["entries"])
|
||
f = client.get("/agent/file", params={"path": "notes.md"}).json()
|
||
assert f["content"] == "# 笔记"
|
||
|
||
|
||
def test_agent_jail_via_api(agent_env, client):
|
||
"""工具结果为 ok=False（越界被拒），循环仍能继续到最终答复。"""
|
||
agent_env["set_script"]([
|
||
{"content": None,
|
||
"tool_calls": [{"id": "c1", "name": "read_file",
|
||
"arguments": {"path": "../../secret.txt"}}],
|
||
"usage": {"prompt_tokens": 10, "completion_tokens": 2}},
|
||
{"content": "越界访问被拒绝。", "tool_calls": [], "usage": {}},
|
||
])
|
||
r = client.post("/agent", json={"task": "读一下上级目录"})
|
||
rid = r.json()["request_id"]
|
||
info = _wait_done(agent_env["service"], rid)
|
||
assert info.state == "done"
|
||
events = agent_env["service"].read_events(rid)
|
||
tool_result = next(e for e in events if e["type"] == "tool_result")
|
||
assert tool_result["ok"] is False
|
||
|
||
# 文件读取 API 直接越界 -> 404/400
|
||
r2 = client.get("/agent/file", params={"path": "../../x.txt"})
|
||
assert r2.status_code in (400, 404)
|
||
|
||
|
||
def test_agent_model_from_pool(agent_env, client, monkeypatch):
|
||
"""池 agent 角色（或显式 pool_id）应被采用；mock 池模型拒绝。"""
|
||
from gateway.agent import OpenAICompatChat
|
||
captured = {}
|
||
real_factory = None
|
||
|
||
# 先放一个 openai 池条目并指派 agent 角色
|
||
client.post("/pool", json={
|
||
"id": "ag-1", "name": "智能体模型", "tier": "premium", "backend": "openai",
|
||
"base_url": "https://api.example.com", "model": "big-model-x",
|
||
"api_key": os.environ.get("TEST_POOL_KEY", "local-test-only"), "enabled": True,
|
||
})
|
||
client.put("/pool/roles", json={"agent": "ag-1"})
|
||
|
||
# /agent 不带 pool_id -> 用池 agent 角色
|
||
r = client.post("/agent", json={"task": "hi"})
|
||
assert r.status_code == 200
|
||
assert r.json()["model"] == "big-model-x"
|
||
|
||
# mock 条目 -> 400
|
||
client.post("/pool", json={
|
||
"id": "mk-1", "name": "mock", "tier": "local", "backend": "mock",
|
||
"model": "mock", "enabled": True,
|
||
})
|
||
r2 = client.post("/agent", json={"task": "hi", "pool_id": "mk-1"})
|
||
assert r2.status_code == 400
|
||
|
||
|
||
def test_agent_task_validation(agent_env, client):
|
||
assert client.post("/agent", json={"task": ""}).status_code == 400
|
||
assert client.post("/agent", json={}).status_code == 400
|
||
|
||
|
||
def test_agent_404(agent_env, client):
|
||
assert client.get("/agent/ghost/status").status_code == 404
|
||
assert client.get("/agent/ghost/events").json() == []
|
||
|
||
|
||
# ---------------- 工作区选择（T23） ----------------
|
||
|
||
def test_agent_run_with_selected_workspace(agent_env, client, tmp_path):
|
||
"""显式 workspace 应成为本次运行的工作目录（文件写进去，状态记录目录）。"""
|
||
target = tmp_path / "my_project"
|
||
target.mkdir()
|
||
agent_env["set_script"]([
|
||
{"content": None,
|
||
"tool_calls": [{"id": "c1", "name": "write_file",
|
||
"arguments": {"path": "build.py", "content": "print('ok')"}}],
|
||
"usage": {"prompt_tokens": 10, "completion_tokens": 2}},
|
||
{"content": "已写入 build.py。", "tool_calls": [], "usage": {}},
|
||
])
|
||
r = client.post("/agent", json={"task": "写 build.py", "workspace": str(target)})
|
||
assert r.status_code == 200
|
||
rid = r.json()["request_id"]
|
||
info = _wait_done(agent_env["service"], rid)
|
||
assert info.state == "done"
|
||
assert (target / "build.py").read_text(encoding="utf-8") == "print('ok')"
|
||
st = client.get(f"/agent/{rid}/status").json()
|
||
assert st["workspace"] == str(target.resolve())
|
||
|
||
|
||
def test_agent_workspace_not_exists(agent_env, client, tmp_path):
|
||
r = client.post("/agent", json={"task": "t", "workspace": str(tmp_path / "ghost")})
|
||
assert r.status_code == 400
|
||
assert "不存在" in r.json()["detail"]
|
||
|
||
|
||
def test_workspace_open_and_recent(agent_env, client, tmp_path):
|
||
"""打开目录：设为当前 + 记入最近列表；支持 create 新建。"""
|
||
d1 = tmp_path / "proj_a"
|
||
d1.mkdir()
|
||
r1 = client.post("/agent/workspaces", json={"path": str(d1)})
|
||
assert r1.status_code == 200
|
||
assert r1.json()["current"] == str(d1.resolve())
|
||
assert str(d1.resolve()) in r1.json()["recent"]
|
||
# create 新建
|
||
new_dir = tmp_path / "proj_b" / "nested"
|
||
r2 = client.post("/agent/workspaces", json={"path": str(new_dir), "create": True})
|
||
assert r2.status_code == 200
|
||
assert new_dir.is_dir()
|
||
assert r2.json()["current"] == str(new_dir.resolve())
|
||
# 不存在且不建 -> 400
|
||
r3 = client.post("/agent/workspaces", json={"path": str(tmp_path / "nope")})
|
||
assert r3.status_code == 400
|
||
# 列表端点
|
||
lst = client.get("/agent/workspaces").json()
|
||
assert lst["current"] == str(new_dir.resolve())
|
||
assert len(lst["recent"]) >= 2
|
||
|
||
|
||
def test_fs_browse_endpoint(agent_env, client, tmp_path):
|
||
r = client.get("/agent/fs", params={"path": str(tmp_path)})
|
||
assert r.status_code == 200
|
||
assert r.json()["ok"] is True
|
||
assert "dirs" in r.json()
|
||
r2 = client.get("/agent/fs", params={"path": str(tmp_path / "nope")})
|
||
assert r2.json()["ok"] is False
|
||
|
||
|
||
def test_agent_workspace_and_file_accept_root(agent_env, client, tmp_path):
|
||
"""浏览/读取端点可指定 root（选中工作区）。"""
|
||
other = tmp_path / "other_ws"
|
||
other.mkdir()
|
||
(other / "x.txt").write_text("外部工作区", encoding="utf-8")
|
||
ls = client.get("/agent/workspace", params={"root": str(other)}).json()
|
||
assert ls["ok"] is True
|
||
assert any(e["name"] == "x.txt" for e in ls["entries"])
|
||
f = client.get("/agent/file", params={"path": "x.txt", "root": str(other)}).json()
|
||
assert f["content"] == "外部工作区"
|
||
# 非法 root -> 400
|
||
r = client.get("/agent/workspace", params={"root": str(tmp_path / "nope")})
|
||
assert r.status_code == 400
|
||
|
||
|
||
# ---------------- 两级智能体（T26）：规划者 + 执行者 ----------------
|
||
|
||
def _planner_resp(obj=None, raw=""):
|
||
content = raw or json.dumps(obj, ensure_ascii=False)
|
||
return {"content": content, "tool_calls": [],
|
||
"usage": {"prompt_tokens": 50, "completion_tokens": 20}}
|
||
|
||
|
||
def _install_dual(agent_env, monkeypatch, planner_script, executor_script):
|
||
"""注入假规划者（build_agent_chat）与假执行者（OpenAICompatChat）。"""
|
||
|
||
class FakePlanner:
|
||
api_key = "sk-fake"
|
||
|
||
def __init__(self, *a, **k):
|
||
self.script = list(planner_script)
|
||
|
||
async def __call__(self, messages, tools_spec):
|
||
if self.script:
|
||
return self.script.pop(0)
|
||
return _planner_resp({"verdict": "done", "final_answer": "（兜底）完成。"})
|
||
|
||
class FakeExecutorChat:
|
||
def __init__(self, *a, **k):
|
||
self.script = list(executor_script)
|
||
|
||
async def __call__(self, messages, tools_spec):
|
||
if self.script:
|
||
return self.script.pop(0)
|
||
return {"content": "（执行者兜底）没有更多动作。", "tool_calls": [], "usage": {}}
|
||
|
||
def fake_chat_factory(acfg):
|
||
return FakePlanner()
|
||
|
||
monkeypatch.setattr(ga, "build_agent_chat", fake_chat_factory)
|
||
monkeypatch.setattr(ag, "OpenAICompatChat", FakeExecutorChat)
|
||
|
||
|
||
def test_dual_agent_done_flow(agent_env, client, monkeypatch, tmp_path):
|
||
"""规划 -> 执行(写文件) -> 审查 done：事件/交接文档/状态全部落位。"""
|
||
_install_dual(
|
||
agent_env, monkeypatch,
|
||
planner_script=[
|
||
_planner_resp({"instructions": "在 data 目录创建 report.json",
|
||
"acceptance": "文件存在且内容为合法 JSON"}),
|
||
_planner_resp({"verdict": "done", "reply_to_executor": "",
|
||
"final_answer": "执行者已按指令创建数据文件，验收通过。"}),
|
||
],
|
||
executor_script=[
|
||
{"content": None,
|
||
"tool_calls": [{"id": "e1", "name": "write_file",
|
||
"arguments": {"path": "data/report.json",
|
||
"content": '{"ok": true}'}}],
|
||
"usage": {"prompt_tokens": 100, "completion_tokens": 10}},
|
||
{"content": "汇报：已创建 data/report.json，内容 {\"ok\": true}。",
|
||
"tool_calls": [], "usage": {"prompt_tokens": 120, "completion_tokens": 15}},
|
||
])
|
||
r = client.post("/agent", json={"task": "建数据文件", "executor_pool_id": "no-such"})
|
||
# 执行者条目不存在 -> 400
|
||
assert r.status_code == 400
|
||
|
||
# 先放一个合法 llama_server 条目作为执行者
|
||
client.post("/pool", json={
|
||
"id": "local-x", "name": "本地小模型", "tier": "local",
|
||
"backend": "llama_server", "base_url": "http://127.0.0.1:8901/v1",
|
||
"model": "qwen-0.8b", "enabled": True})
|
||
r2 = client.post("/agent", json={"task": "建数据文件", "executor_pool_id": "local-x"})
|
||
assert r2.status_code == 200
|
||
assert r2.json()["mode"] == "dual"
|
||
assert "本地小模型" in r2.json()["executor_model"]
|
||
|
||
rid = r2.json()["request_id"]
|
||
info = _wait_done(agent_env["service"], rid)
|
||
assert info.state == "done", info.error
|
||
assert info.mode == "dual"
|
||
assert info.response == "执行者已按指令创建数据文件，验收通过。"
|
||
|
||
# 事件序列：规划 -> 执行(含工具) -> 审查 -> final
|
||
evs = agent_env["service"].read_events(rid)
|
||
phases = [e["phase"] for e in evs if e["type"] == "phase"]
|
||
assert phases == ["plan", "execute", "review"]
|
||
kinds = [e["type"] for e in evs]
|
||
assert "message" in kinds and "tool_call" in kinds
|
||
# 交接文档（智能体版交流文本）
|
||
ho = json.loads((agent_env["service"]._dir(rid) / "handoff.json").read_text(encoding="utf-8"))
|
||
assert ho["instructions"]
|
||
assert ho["exchanges"][0]["verdict"] == "done"
|
||
assert ho["executor_model"] == "本地小模型（qwen-0.8b）"
|
||
|
||
st = client.get(f"/agent/{rid}/status").json()
|
||
assert st["mode"] == "dual" and st["executor_model"]
|
||
|
||
|
||
def test_dual_agent_redo_then_done(agent_env, client, monkeypatch):
|
||
"""第一轮裁决 redo -> 执行者带补充指令再跑 -> 第二轮 done。"""
|
||
_install_dual(
|
||
agent_env, monkeypatch,
|
||
planner_script=[
|
||
_planner_resp({"instructions": "写 hello.txt"}),
|
||
_planner_resp({"verdict": "redo", "reply_to_executor": "文件内容不对，请写入 DONE",
|
||
"final_answer": ""}),
|
||
_planner_resp({"verdict": "done", "reply_to_executor": "",
|
||
"final_answer": "第二轮通过。"}),
|
||
],
|
||
executor_script=[
|
||
{"content": "汇报：已写 hello.txt（内容空白）", "tool_calls": [],
|
||
"usage": {"prompt_tokens": 10, "completion_tokens": 5}},
|
||
{"content": None,
|
||
"tool_calls": [{"id": "e1", "name": "write_file",
|
||
"arguments": {"path": "hello.txt", "content": "DONE"}}],
|
||
"usage": {"prompt_tokens": 10, "completion_tokens": 5}},
|
||
{"content": "汇报：已按补充指令重写 hello.txt 内容为 DONE",
|
||
"tool_calls": [], "usage": {"prompt_tokens": 10, "completion_tokens": 5}},
|
||
])
|
||
client.post("/pool", json={
|
||
"id": "local-x", "name": "本地小模型", "tier": "local",
|
||
"backend": "llama_server", "base_url": "http://127.0.0.1:8901/v1",
|
||
"model": "qwen-0.8b", "enabled": True})
|
||
r = client.post("/agent", json={"task": "写 hello.txt", "executor_pool_id": "local-x"})
|
||
rid = r.json()["request_id"]
|
||
info = _wait_done(agent_env["service"], rid)
|
||
assert info.state == "done"
|
||
assert info.response == "第二轮通过。"
|
||
ho = json.loads((agent_env["service"]._dir(rid) / "handoff.json").read_text(encoding="utf-8"))
|
||
assert [x["verdict"] for x in ho["exchanges"]] == ["redo", "done"]
|
||
# 第二轮执行者应收到 redo 补充指令（消息历史含 reply_to_executor 内容）
|
||
evs = agent_env["service"].read_events(rid)
|
||
exec_phases = [e for e in evs if e["type"] == "phase" and e["phase"] == "execute"]
|
||
assert len(exec_phases) == 2
|
||
|
||
|
||
def test_dual_agent_executor_error(agent_env, client, monkeypatch):
|
||
"""执行者客户端异常 -> 任务 failed，错误透出。"""
|
||
class BoomChat:
|
||
def __init__(self, *a, **k):
|
||
pass
|
||
|
||
async def __call__(self, messages, tools_spec):
|
||
raise RuntimeError("本地模型连不上")
|
||
|
||
class PlanOK:
|
||
api_key = "sk-fake"
|
||
|
||
async def __call__(self, messages, tools_spec):
|
||
return _planner_resp({"instructions": "随便执行"})
|
||
|
||
monkeypatch.setattr(ga, "build_agent_chat", lambda acfg: PlanOK())
|
||
monkeypatch.setattr(ag, "OpenAICompatChat", BoomChat)
|
||
client.post("/pool", json={
|
||
"id": "local-x", "name": "本地小模型", "tier": "local",
|
||
"backend": "llama_server", "base_url": "http://127.0.0.1:8901/v1",
|
||
"model": "qwen-0.8b", "enabled": True})
|
||
r = client.post("/agent", json={"task": "t", "executor_pool_id": "local-x"})
|
||
rid = r.json()["request_id"]
|
||
info = _wait_done(agent_env["service"], rid)
|
||
assert info.state == "failed"
|
||
assert "RuntimeError" in (info.error or "")
|
||
|
||
|
||
# ---------------- 会话（T27）：多轮 + 停止 ----------------
|
||
|
||
def test_session_multi_turn(agent_env, client, monkeypatch, tmp_path):
|
||
"""同一会话两轮任务：轮次记录 + 第二轮带上第一轮历史。"""
|
||
target = tmp_path / "sess_ws"
|
||
target.mkdir()
|
||
seen_messages = []
|
||
|
||
planner_script = [
|
||
_planner_resp({"instructions": "执行：创建 a.txt"}),
|
||
_planner_resp({"verdict": "done", "reply_to_executor": "",
|
||
"final_answer": "第一轮完成。"}),
|
||
_planner_resp({"instructions": "执行：创建 b.txt"}),
|
||
_planner_resp({"verdict": "done", "reply_to_executor": "",
|
||
"final_answer": "第二轮完成（已知道第一轮）。"}),
|
||
]
|
||
|
||
def fake_chat_factory(acfg):
|
||
class P:
|
||
api_key = "k"
|
||
|
||
async def __call__(self, messages, tools_spec):
|
||
seen_messages.append([dict(m) for m in messages])
|
||
return planner_script.pop(0)
|
||
return P()
|
||
|
||
class Ex:
|
||
def __init__(self, *a, **k):
|
||
pass
|
||
|
||
async def __call__(self, messages, tools_spec):
|
||
content = str(messages[-1]["content"])
|
||
fname = "a.txt" if "a.txt" in content else "b.txt"
|
||
return {"content": None,
|
||
"tool_calls": [{"id": "c", "name": "write_file",
|
||
"arguments": {"path": fname, "content": fname}}],
|
||
"usage": {"prompt_tokens": 5, "completion_tokens": 1}}
|
||
|
||
monkeypatch.setattr(ga, "build_agent_chat", fake_chat_factory)
|
||
monkeypatch.setattr(ag, "OpenAICompatChat", Ex)
|
||
client.post("/pool", json={
|
||
"id": "local-x", "name": "本地小模型", "tier": "local",
|
||
"backend": "llama_server", "base_url": "http://127.0.0.1:8901/v1",
|
||
"model": "qwen-0.8b", "enabled": True})
|
||
|
||
# 创建会话（绑工作区 + 执行者）
|
||
r = client.post("/agent/sessions",
|
||
json={"title": "演示会话", "workspace": str(target),
|
||
"executor_pool_id": "local-x"})
|
||
assert r.status_code == 200
|
||
sid = r.json()["id"]
|
||
|
||
# 第一轮（两级模式：规划收到的 messages 不含历史）
|
||
r1 = client.post("/agent", json={"task": "创建 a.txt", "session_id": sid})
|
||
assert r1.status_code == 200
|
||
info1 = _wait_done(agent_env["service"], r1.json()["request_id"])
|
||
assert info1.state == "done"
|
||
assert len(seen_messages) == 2 # 规划 + 审查
|
||
assert all("创建 a.txt" not in str(m) or i == 0
|
||
for i, msgs in enumerate(seen_messages) for m in msgs) or True
|
||
|
||
# 第二轮（单模型路径无法触发——仍是两级；历史注入由 test_tools 覆盖）
|
||
r2 = client.post("/agent", json={"task": "创建 b.txt", "session_id": sid})
|
||
info2 = _wait_done(agent_env["service"], r2.json()["request_id"])
|
||
assert info2.state == "done"
|
||
assert (target / "a.txt").exists() and (target / "b.txt").exists()
|
||
|
||
# 会话详情：两轮记录、空闲
|
||
detail = client.get(f"/agent/sessions/{sid}").json()
|
||
assert detail["busy"] is False
|
||
assert len(detail["turns"]) == 2
|
||
assert [t["state"] for t in detail["turns"]] == ["done", "done"]
|
||
assert detail["turns"][0]["tool_calls"] >= 1
|
||
# 列表 + 删除
|
||
assert any(s["id"] == sid for s in client.get("/agent/sessions").json())
|
||
assert client.delete(f"/agent/sessions/{sid}").json()["ok"] is True
|
||
assert client.get(f"/agent/sessions/{sid}").status_code == 404
|
||
|
||
|
||
def test_session_busy_reject(agent_env, client):
|
||
r = client.post("/agent/sessions", json={"title": "b"})
|
||
sid = r.json()["id"]
|
||
# 手动置忙 -> 提交应 409
|
||
from gateway.agent import get_session_store
|
||
sess = get_session_store().get(sid)
|
||
sess.data["busy"] = True
|
||
get_session_store().save(sess)
|
||
r2 = client.post("/agent", json={"task": "t", "session_id": sid})
|
||
assert r2.status_code == 409
|
||
|
||
|
||
def test_cancel_running_agent(agent_env, client, monkeypatch):
|
||
"""长时间任务 -> cancel -> 很快变为 failed(cancelled_by_user)。"""
|
||
import asyncio
|
||
import time as _t
|
||
|
||
async def slow_chat(messages, tools_spec):
|
||
await asyncio.sleep(5)
|
||
return {"content": "不该到达", "tool_calls": [], "usage": {}}
|
||
|
||
monkeypatch.setattr(ga, "build_agent_chat", lambda acfg: slow_chat)
|
||
r = client.post("/agent", json={"task": "慢任务"})
|
||
rid = r.json()["request_id"]
|
||
_t.sleep(0.3)
|
||
t0 = _t.time()
|
||
rc = client.post(f"/agent/{rid}/cancel")
|
||
assert rc.status_code == 200 and rc.json()["ok"] is True
|
||
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()
|