"""声明式 agent 定义 v1/v2 单测：解析、校验、解析到任务字段、建任务入口。 v2 改动：prompt 不再拼头，指令落库（见 test_agent_def_v2.py）。 运行（在仓库根目录下）： python3 -m unittest discover -s cloud/tests -p "test_agent_def.py" -v 说明：服务端部分用 create_server 本地 HTTP（test_task_options 同款），不跑真实模型、不联网、不登录任何服务；skill 名单 以仓库 cloud/skills/ 实际包为准。 """ import http.client import json import os import sys import tempfile import threading import unittest from pathlib import Path try: from cloud.agent_def import ( format_instructions_header, list_known_skills, load_agent_definition, parse_agent_definition, resolve_task_fields, ) from cloud.auth import AuthStore from cloud.server import create_server except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.agent_def import ( format_instructions_header, list_known_skills, load_agent_definition, parse_agent_definition, resolve_task_fields, ) from cloud.auth import AuthStore from cloud.server import create_server AGENT = { "name": "research-brief", "version": "v1", "description": "Evidence-backed brief.", "instructions": "Cite sources and dates.", "skills": ["live-info"], "defaults": { "model": "gpt-6.1-sol", "effort": "medium", "approval": "ask", "executor": "docker", }, } def _agent(**overrides): agent = json.loads(json.dumps(AGENT)) agent.update(overrides) return agent class ParseTestCase(unittest.TestCase): def test_valid_and_known_skills(self): definition = parse_agent_definition( _agent(), known_skills=list_known_skills()) self.assertEqual(definition.name, "research-brief") self.assertEqual(definition.defaults["effort"], "medium") self.assertEqual(definition.to_summary()["skills"], ["live-info"]) def test_name_and_unknown_fields_rejected(self): for bad in ( _agent(name="Bad Name"), _agent(name=""), _agent(routing={"mode": "active"}), _agent(defaults={"efforts": "high"}), _agent(defaults={"effort": "ultra"}), _agent(defaults={"plan": True, "approval": "full"}), _agent(skills=["no-such-skill"]), _agent(instructions="x" * 8001), ): with self.assertRaises(ValueError, msg=bad): parse_agent_definition( bad, known_skills=list_known_skills()) def test_unknown_skill_skipped_when_no_known_list(self): # known_skills=None 只跳过存在性检查，绝不发明技能。 definition = parse_agent_definition(_agent(skills=["x"])) self.assertEqual(definition.skills, ["x"]) def test_resolve_explicit_body_wins_per_field(self): definition = parse_agent_definition(_agent()) resolved = resolve_task_fields( definition, {"effort": "low", "plan": True}) self.assertEqual(resolved["model"], "gpt-6.1-sol") self.assertEqual(resolved["effort"], "low") self.assertEqual(resolved["approval"], "ask") self.assertEqual(resolved["executor"], "docker") self.assertEqual(resolved["plan"], 1) # 不改输入。 self.assertEqual(definition.defaults["effort"], "medium") def test_header_and_no_header(self): definition = parse_agent_definition(_agent()) headed = format_instructions_header(definition, "do research") self.assertTrue(headed.startswith( "[Agent: research-brief v1]\nCite sources and dates.\n\n")) self.assertTrue(headed.endswith("do research")) self.assertEqual( format_instructions_header(None, "do research"), "do research") def test_load_example_and_traversal_guard(self): repo_root = Path(__file__).resolve().parents[2] definition = load_agent_definition( "research-brief.json", agents_dir=repo_root / "agents") self.assertEqual(definition.name, "research-brief") with self.assertRaises(ValueError): load_agent_definition( "../cloud/agent_def.py", agents_dir=repo_root / "agents") class ServerAgentTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") self.auth = AuthStore(self.db_path) self.owner_id, self.owner_key = self.auth.create_user("owner") self.server = create_server(self.db_path, "127.0.0.1", 0) self.port = self.server.server_address[1] self.thread = threading.Thread( target=self.server.serve_forever, daemon=True) self.thread.start() status, self.project = self._req( "POST", "/projects", {"name": "demo"}, self.owner_key) self.assertEqual(status, 201) def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) try: self.auth.conn.close() except Exception: pass self.tmp.cleanup() def _req(self, method, path, body=None, key=None): conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) headers = {"Content-Type": "application/json"} if key: headers["X-API-Key"] = key conn.request(method, path, json.dumps(body) if body is not None else None, headers) resp = conn.getresponse() raw = resp.read() conn.close() try: data = json.loads(raw) except ValueError: data = None return resp.status, data def _list_tasks(self): status, data = self._req( "GET", "/tasks?project_id=" + self.project["id"], key=self.owner_key) self.assertEqual(status, 200) return data["tasks"] def test_agent_fills_fields_and_stores_instructions(self): # v2：prompt 保持干净，指令落库，不再拼头（v1 行为已 # 被 v2 替换，见 agent_def.py 的 v2 边界说明）。 status, task = self._req("POST", "/tasks", { "project_id": self.project["id"], "prompt": "do research", "agent": _agent()}, self.owner_key) self.assertEqual(status, 201, task) self.assertEqual(task["model"], "gpt-6.1-sol") self.assertEqual(task["effort"], "medium") self.assertEqual(task["approval"], "ask") self.assertEqual(task["executor"], "docker") self.assertEqual(task["agent"]["name"], "research-brief") self.assertEqual(task["agent"]["skills"], ["live-info"]) self.assertEqual(task["agent_name"], "research-brief") self.assertEqual(task["agent_version"], "v1") self.assertEqual(task["agent_instructions"], "Cite sources and dates.") stored = self._list_tasks()[0] self.assertEqual(stored["prompt"], "do research") self.assertEqual(stored["agent_instructions"], "Cite sources and dates.") def test_explicit_body_wins_over_agent(self): status, task = self._req("POST", "/tasks", { "project_id": self.project["id"], "prompt": "do research", "effort": "low", "agent": _agent()}, self.owner_key) self.assertEqual(status, 201, task) self.assertEqual(task["effort"], "low") self.assertEqual(task["model"], "gpt-6.1-sol") def test_invalid_agent_rejected_and_no_task_created(self): for bad_agent in ( _agent(skills=["no-such-skill"]), _agent(defaults={"effort": "ultra"}), "not-an-object", ): status, data = self._req("POST", "/tasks", { "project_id": self.project["id"], "prompt": "do research", "agent": bad_agent}, self.owner_key) self.assertEqual(status, 400, bad_agent) self.assertEqual(self._list_tasks(), []) def test_agent_instructions_rejected_with_goal(self): status, data = self._req("POST", "/tasks", { "project_id": self.project["id"], "prompt": "do research", "goal_mode": 1, "agent": _agent()}, self.owner_key) self.assertEqual(status, 400, data) self.assertEqual(self._list_tasks(), []) def test_no_agent_unchanged(self): status, task = self._req("POST", "/tasks", { "project_id": self.project["id"], "prompt": "hi"}, self.owner_key) self.assertEqual(status, 201, task) self.assertIsNone(task["model"]) self.assertEqual(task["prompt"], "hi") self.assertNotIn("agent", task)