test_runner_completion_gate.py 9.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303
  1. import pytest
  2. from agent.core.presets import AgentPreset, register_preset
  3. from agent.core.runner import AgentRunner, RunConfig
  4. from agent.orchestration.models import AgentRole, CompletionPolicy
  5. from agent.trace.models import Trace
  6. from agent.trace.store import FileSystemTraceStore
  7. class StubCoordinator:
  8. def __init__(self, snapshots):
  9. self.snapshots = list(snapshots)
  10. self.ensure_calls = []
  11. self.completion_calls = 0
  12. async def ensure_ledger(self, trace_id, mission, *, allow_create=True):
  13. self.ensure_calls.append((trace_id, mission))
  14. async def mission_completion(self, root_trace_id):
  15. self.completion_calls += 1
  16. if len(self.snapshots) > 1:
  17. return self.snapshots.pop(0)
  18. return self.snapshots[0]
  19. def _knowledge_off():
  20. from agent.tools.builtin.knowledge import KnowledgeConfig
  21. return KnowledgeConfig(
  22. enable_extraction=False,
  23. enable_completion_extraction=False,
  24. enable_injection=False,
  25. )
  26. def _register_role_preset(name, role, max_iterations):
  27. register_preset(
  28. name,
  29. AgentPreset(
  30. role=role,
  31. allowed_tools=[],
  32. max_iterations=max_iterations,
  33. skills=[],
  34. ),
  35. )
  36. def _explicit_config(name, max_iterations):
  37. root_task_spec = {
  38. "objective": "complete the mission",
  39. "acceptance_criteria": [{"description": "mission result is validated"}],
  40. }
  41. return RunConfig(
  42. agent_type=name,
  43. completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
  44. max_iterations=max_iterations,
  45. tools=[],
  46. tool_groups=[],
  47. root_task_spec=root_task_spec,
  48. knowledge=_knowledge_off(),
  49. )
  50. @pytest.mark.asyncio
  51. async def test_planner_no_tool_exit_is_persisted_and_loop_continues(tmp_path):
  52. preset = "test_completion_gate_planner"
  53. _register_role_preset(preset, AgentRole.PLANNER, 3)
  54. coordinator = StubCoordinator([
  55. {
  56. "status": "needs_replan",
  57. "root_task_id": "root-task",
  58. "pending_tasks": [{"task_id": "child", "status": "pending"}],
  59. },
  60. {
  61. "status": "completed",
  62. "root_task_id": "root-task",
  63. "pending_tasks": [],
  64. "result_summary": "accepted root result",
  65. },
  66. ])
  67. calls = []
  68. async def llm_call(**kwargs):
  69. calls.append(kwargs["messages"])
  70. return {"content": "", "tool_calls": None, "finish_reason": "stop"}
  71. store = FileSystemTraceStore(str(tmp_path))
  72. runner = AgentRunner(
  73. trace_store=store,
  74. llm_call=llm_call,
  75. task_coordinator=coordinator,
  76. )
  77. events = [
  78. event
  79. async for event in runner.run(
  80. [{"role": "user", "content": "run mission"}],
  81. _explicit_config(preset, 3),
  82. )
  83. ]
  84. final = events[-1]
  85. assert isinstance(final, Trace)
  86. assert final.status == "completed"
  87. assert final.result_summary == "accepted root result"
  88. assert coordinator.ensure_calls[0][1]["objective"] == "complete the mission"
  89. assert len(calls) == 2
  90. assert any(
  91. message.get("role") == "user"
  92. and "[FRAMEWORK_MISSION_INCOMPLETE]" in message.get("content", "")
  93. for message in calls[1]
  94. )
  95. assert calls[1][0]["role"] == "system"
  96. assert "Framework role contract: Planner" in calls[1][0]["content"]
  97. persisted = await store.get_trace_messages(final.trace_id)
  98. controls = [
  99. message for message in persisted
  100. if message.role == "user"
  101. and "[FRAMEWORK_MISSION_INCOMPLETE]" in str(message.content)
  102. ]
  103. assert len(controls) == 1
  104. @pytest.mark.asyncio
  105. async def test_incomplete_planner_resume_restores_explicit_identity(tmp_path):
  106. preset = "test_completion_gate_resume"
  107. _register_role_preset(preset, AgentRole.PLANNER, 1)
  108. coordinator = StubCoordinator([{
  109. "status": "needs_replan",
  110. "root_task_id": "root-task",
  111. "pending_tasks": [{"task_id": "root-task", "status": "needs_replan"}],
  112. }])
  113. async def llm_call(**_kwargs):
  114. return {"content": "", "tool_calls": None, "finish_reason": "stop"}
  115. store = FileSystemTraceStore(str(tmp_path))
  116. runner = AgentRunner(
  117. trace_store=store,
  118. llm_call=llm_call,
  119. task_coordinator=coordinator,
  120. )
  121. first = await runner.run_result(
  122. [{"role": "user", "content": "run mission"}],
  123. _explicit_config(preset, 1),
  124. )
  125. assert first["status"] == "incomplete"
  126. resume_config = RunConfig(
  127. trace_id=first["trace_id"],
  128. knowledge=_knowledge_off(),
  129. )
  130. resumed = await runner.run_result([], resume_config)
  131. assert resumed["status"] == "incomplete"
  132. assert resumed["error"] == "iteration_limit_reached"
  133. assert resume_config.agent_type == preset
  134. assert resume_config.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION
  135. @pytest.mark.asyncio
  136. async def test_explicit_planner_rewind_and_wrong_role_resume_are_rejected_before_mutation(
  137. tmp_path,
  138. ):
  139. planner_preset = "test_completion_gate_rewind_planner"
  140. worker_preset = "test_completion_gate_wrong_role_worker"
  141. _register_role_preset(planner_preset, AgentRole.PLANNER, 2)
  142. _register_role_preset(worker_preset, AgentRole.WORKER, 2)
  143. store = FileSystemTraceStore(str(tmp_path))
  144. planner_trace = Trace(
  145. trace_id="planner-trace",
  146. mode="agent",
  147. task="mission",
  148. agent_type=planner_preset,
  149. agent_role=AgentRole.PLANNER.value,
  150. status="completed",
  151. head_sequence=0,
  152. last_sequence=10,
  153. )
  154. worker_trace = Trace(
  155. trace_id="worker-trace",
  156. mode="agent",
  157. task="worker",
  158. agent_type=worker_preset,
  159. agent_role=AgentRole.WORKER.value,
  160. status="completed",
  161. )
  162. await store.create_trace(planner_trace)
  163. await store.create_trace(worker_trace)
  164. coordinator = StubCoordinator([{"status": "completed"}])
  165. runner = AgentRunner(
  166. trace_store=store,
  167. llm_call=lambda **_kwargs: None,
  168. task_coordinator=coordinator,
  169. )
  170. with pytest.raises(ValueError, match="do not support rewind"):
  171. await runner.run_result(
  172. [],
  173. RunConfig(
  174. trace_id=planner_trace.trace_id,
  175. agent_type=planner_preset,
  176. completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
  177. after_sequence=5,
  178. ),
  179. )
  180. with pytest.raises(ValueError, match="Trace role mismatch"):
  181. await runner.run_result(
  182. [],
  183. RunConfig(
  184. trace_id=worker_trace.trace_id,
  185. agent_type=planner_preset,
  186. completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
  187. ),
  188. )
  189. assert (await store.get_trace(planner_trace.trace_id)).status == "completed"
  190. assert (await store.get_trace(worker_trace.trace_id)).status == "completed"
  191. assert coordinator.ensure_calls == []
  192. @pytest.mark.asyncio
  193. async def test_planner_iteration_exhaustion_is_incomplete(tmp_path):
  194. preset = "test_completion_gate_exhaustion"
  195. _register_role_preset(preset, AgentRole.PLANNER, 2)
  196. coordinator = StubCoordinator([{
  197. "status": "waiting_children",
  198. "root_task_id": "root-task",
  199. "pending_tasks": ["child"],
  200. }])
  201. async def llm_call(**_kwargs):
  202. return {"content": "", "tool_calls": None, "finish_reason": "stop"}
  203. runner = AgentRunner(
  204. trace_store=FileSystemTraceStore(str(tmp_path)),
  205. llm_call=llm_call,
  206. task_coordinator=coordinator,
  207. )
  208. result = await runner.run_result(
  209. [{"role": "user", "content": "run mission"}],
  210. _explicit_config(preset, 2),
  211. )
  212. assert result["status"] == "incomplete"
  213. assert result["error"] == "iteration_limit_reached"
  214. assert result["summary"] is None
  215. @pytest.mark.asyncio
  216. async def test_blocked_root_stops_planner_as_incomplete(tmp_path):
  217. preset = "test_completion_gate_blocked"
  218. _register_role_preset(preset, AgentRole.PLANNER, 3)
  219. coordinator = StubCoordinator([{
  220. "status": "blocked",
  221. "root_task_id": "root-task",
  222. "blocked_reason": "external dependency",
  223. "pending_tasks": [],
  224. }])
  225. calls = 0
  226. async def llm_call(**_kwargs):
  227. nonlocal calls
  228. calls += 1
  229. return {"content": "blocked", "tool_calls": None, "finish_reason": "stop"}
  230. runner = AgentRunner(
  231. trace_store=FileSystemTraceStore(str(tmp_path)),
  232. llm_call=llm_call,
  233. task_coordinator=coordinator,
  234. )
  235. result = await runner.run_result(
  236. [{"role": "user", "content": "run mission"}],
  237. _explicit_config(preset, 3),
  238. )
  239. assert calls == 1
  240. assert result["status"] == "incomplete"
  241. assert result["error"] == "mission_blocked"
  242. @pytest.mark.asyncio
  243. @pytest.mark.parametrize("role", [AgentRole.WORKER, AgentRole.VALIDATOR])
  244. async def test_explicit_subtrace_without_terminal_submit_is_protocol_failure(
  245. tmp_path, role
  246. ):
  247. preset = "test_completion_gate_" + role.value
  248. _register_role_preset(preset, role, 2)
  249. async def llm_call(**_kwargs):
  250. return {"content": "done", "tool_calls": None, "finish_reason": "stop"}
  251. runner = AgentRunner(
  252. trace_store=FileSystemTraceStore(str(tmp_path / role.value)),
  253. llm_call=llm_call,
  254. task_coordinator=StubCoordinator([{"status": "completed"}]),
  255. )
  256. result = await runner.run_result(
  257. [{"role": "user", "content": "sub task"}],
  258. _explicit_config(preset, 2),
  259. )
  260. assert result["status"] == "failed"
  261. assert result["error"] == "protocol_violation"