test_mission_workbench.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388
  1. from __future__ import annotations
  2. import json
  3. from dataclasses import asdict
  4. from types import SimpleNamespace
  5. from typing import Any
  6. import pytest
  7. from agent.orchestration import AgentRole, ArtifactRef, RoleContextRequest, TaskStatus
  8. from script_build_host.application.mission_workbench import (
  9. MissionWorkbench,
  10. WorkbenchError,
  11. _fit_planner_state,
  12. )
  13. from script_build_host.domain.task_contracts import (
  14. ScriptCriterion,
  15. ScriptIntentClass,
  16. ScriptTaskBudget,
  17. ScriptTaskContractV1,
  18. ScriptTaskKind,
  19. )
  20. from script_build_host.domain.workbench import NotModified, PlannerState, semantic_handle
  21. class _TaskStore:
  22. def __init__(self, ledger: Any) -> None:
  23. self.ledger = ledger
  24. async def load(self, root_trace_id: str) -> Any:
  25. assert root_trace_id == "root"
  26. return self.ledger
  27. class _Receipts:
  28. def __init__(self) -> None:
  29. self.events: list[dict[str, Any]] = []
  30. async def append_event(self, _trace_id: str, event: str, payload: dict[str, Any]) -> None:
  31. self.events.append({"event": event, "event_id": len(self.events) + 1, **payload})
  32. async def get_events(self, _trace_id: str) -> list[dict[str, Any]]:
  33. return list(self.events)
  34. class _Bindings:
  35. async def get_by_root(self, root_trace_id: str) -> Any:
  36. assert root_trace_id == "root"
  37. return SimpleNamespace(
  38. root_trace_id="root",
  39. script_build_id=900049,
  40. input_snapshot_id=3,
  41. active_direction_artifact_version_id=None,
  42. )
  43. class _Snapshots:
  44. async def get(self, snapshot_id: str, *, script_build_id: int) -> Any:
  45. assert (snapshot_id, script_build_id) == ("3", 900049)
  46. return SimpleNamespace(
  47. snapshot_id="3",
  48. topic_id=49,
  49. topic={"topic": {"result": "one bounded topic"}},
  50. account={"account_name": "fixture"},
  51. persona_points=tuple({"name": f"point-{index}"} for index in range(80)),
  52. section_patterns=tuple({"name": f"pattern-{index}"} for index in range(80)),
  53. strategies=(),
  54. )
  55. class _Contracts:
  56. def __init__(self, contract: ScriptTaskContractV1 | None = None) -> None:
  57. self.contract = contract
  58. async def read(self, root_trace_id: str, contract_ref: str) -> Any:
  59. assert root_trace_id == "root"
  60. assert contract_ref.startswith("script-build://task-contracts/sha256/")
  61. assert self.contract is not None
  62. return SimpleNamespace(contract=self.contract)
  63. class _Artifact:
  64. def content_payload(self) -> dict[str, Any]:
  65. return {
  66. "paragraphs": [
  67. {
  68. "paragraph_id": 77,
  69. "name": "opening",
  70. "full_description": "complete candidate prose",
  71. }
  72. ],
  73. "artifact_ref": "script-build://artifact-versions/77",
  74. "canonical_sha256": "sha256:" + "a" * 64,
  75. }
  76. class _Artifacts:
  77. async def read_by_ref(self, ref: ArtifactRef, *, script_build_id: int) -> Any:
  78. assert script_build_id == 900049
  79. assert ref.uri == "script-build://artifact-versions/77"
  80. return SimpleNamespace(artifact=_Artifact())
  81. class _Candidates:
  82. async def read_attempt_workspace(self, *, context: dict[str, Any]) -> dict[str, Any]:
  83. assert context["attempt_id"] == "attempt"
  84. return {
  85. "paragraphs": [
  86. {
  87. "paragraph_id": 77,
  88. "paragraph_index": 1,
  89. "name": "opening",
  90. "content_range": {"scope": "opening"},
  91. "level": 1,
  92. "theme": "theme",
  93. "form": "form",
  94. "function": "function",
  95. "feeling": "feeling",
  96. "description": "description",
  97. "full_description": "complete prose",
  98. }
  99. ],
  100. "elements": [],
  101. "paragraph_element_links": [],
  102. }
  103. def test_planner_state_compacts_optional_frontiers_instead_of_failing() -> None:
  104. state = PlannerState(
  105. state_revision="ledger:1",
  106. phase=2,
  107. root={"task_id": "root", "status": "waiting_children", "objective": "compose"},
  108. focused_task_id="root",
  109. active_subgraph=tuple(
  110. {"task_id": f"task-{index}", "objective": "x" * 500} for index in range(12)
  111. ),
  112. accepted_frontier=tuple(
  113. {"decision_id": f"decision-{index}", "summary": "y" * 1_200}
  114. for index in range(10)
  115. ),
  116. goal_progress={"total": 1, "covered": 0, "uncovered_goal_ids": ["goal"]},
  117. counts_by_status={"completed": 10},
  118. active_operations=(),
  119. latest_failures=(),
  120. retired_task_count=20,
  121. active_total=12,
  122. accepted_total=10,
  123. context_bundle={"cards": [{"summary": "z" * 1_000}]},
  124. phase_protocol={"phase": 2},
  125. )
  126. compacted = _fit_planner_state(state)
  127. assert compacted.accepted_has_more is True
  128. assert len(compacted.accepted_frontier) < len(state.accepted_frontier)
  129. assert len(json.dumps(compacted.to_payload(), ensure_ascii=False)) <= 24_000
  130. def _root_task(*, children: list[str] | None = None) -> Any:
  131. return SimpleNamespace(
  132. task_id="root-task",
  133. parent_task_id=None,
  134. child_task_ids=list(children or []),
  135. display_path="1",
  136. status=TaskStatus.RUNNING,
  137. blocked_reason=None,
  138. current_spec=SimpleNamespace(objective="build script", context_refs=()),
  139. current_spec_version=1,
  140. attempt_ids=[],
  141. decision_ids=[],
  142. )
  143. def _contract() -> ScriptTaskContractV1:
  144. return ScriptTaskContractV1(
  145. task_kind=ScriptTaskKind.PARAGRAPH,
  146. scope_ref="script-build://scopes/full/opening",
  147. intent_class=ScriptIntentClass.EXPLORE,
  148. objective="write the opening",
  149. input_decision_refs=(),
  150. base_artifact_ref=None,
  151. write_scope=("script-build://writes/paragraphs/opening",),
  152. gap_ref=None,
  153. output_schema="paragraph-artifact/v1",
  154. criteria=(ScriptCriterion("criterion", "complete prose"),),
  155. budget=ScriptTaskBudget(),
  156. goal_ids=("goal-1",),
  157. compiler_manifest_digest="sha256:" + "b" * 64,
  158. )
  159. def _worker_task() -> Any:
  160. return SimpleNamespace(
  161. task_id="worker",
  162. parent_task_id="root-task",
  163. child_task_ids=[],
  164. display_path="1.1",
  165. status=TaskStatus.RUNNING,
  166. blocked_reason=None,
  167. current_spec=SimpleNamespace(
  168. objective="write the opening",
  169. context_refs=(
  170. "script-build://task-kinds/paragraph",
  171. "script-build://task-contracts/sha256/contract",
  172. ),
  173. ),
  174. current_spec_version=1,
  175. attempt_ids=["attempt"],
  176. decision_ids=[],
  177. )
  178. @pytest.mark.asyncio
  179. async def test_planner_projection_is_bounded_incremental_and_does_not_copy_retired_tree() -> None:
  180. tasks = {"root-task": _root_task(children=[f"done-{index}" for index in range(120)])}
  181. for index in range(120):
  182. tasks[f"done-{index}"] = SimpleNamespace(
  183. task_id=f"done-{index}",
  184. parent_task_id="root-task",
  185. child_task_ids=[],
  186. display_path=f"1.{index + 1}",
  187. status=TaskStatus.COMPLETED,
  188. blocked_reason=None,
  189. current_spec=SimpleNamespace(
  190. objective="retired detail " + "x" * 1000,
  191. context_refs=(),
  192. ),
  193. current_spec_version=1,
  194. attempt_ids=[],
  195. decision_ids=[],
  196. )
  197. ledger = SimpleNamespace(
  198. revision=49,
  199. root_task_id="root-task",
  200. focused_task_id="root-task",
  201. tasks=tasks,
  202. attempts={},
  203. decisions={},
  204. operations={},
  205. )
  206. receipts = _Receipts()
  207. workbench = MissionWorkbench(
  208. task_store=_TaskStore(ledger),
  209. bindings=_Bindings(),
  210. snapshots=_Snapshots(),
  211. artifacts=_Artifacts(),
  212. contracts=_Contracts(),
  213. receipt_store=receipts,
  214. )
  215. state = await workbench.planner_state("root")
  216. assert not isinstance(state, NotModified)
  217. assert state.retired_task_count == 120
  218. assert len(state.active_subgraph) == 1
  219. assert state.phase_protocol["phase_epoch"] == "script-build-phase-1"
  220. assert state.phase_protocol["current_task"]["allowed_child_kinds"] == ("direction",)
  221. encoded = await workbench.render("root", ledger)
  222. assert len(encoded) < 24_000
  223. assert "retired detail" not in encoded
  224. legacy_full_tree = json.dumps(
  225. [item.current_spec.objective for item in tasks.values()], ensure_ascii=False
  226. )
  227. assert len(encoded) * 2 < len(legacy_full_tree)
  228. for _ in range(15):
  229. assert isinstance(
  230. await workbench.planner_state("root", known_revision="ledger:49"),
  231. NotModified,
  232. )
  233. first_page = await workbench.search_mission_context(
  234. root_trace_id="root",
  235. role="planner",
  236. collection="tasks",
  237. page_size=20,
  238. )
  239. assert 0 < len(first_page["page"]["items"]) <= 20
  240. assert first_page["receipt"]["estimated_tokens"] <= 2_000
  241. assert first_page["page"]["has_more"] is True
  242. assert first_page["page"]["next_cursor"]
  243. assert len(json.dumps(first_page, ensure_ascii=False, separators=(",", ":"))) < 8_000
  244. await workbench.search_mission_context(
  245. root_trace_id="root",
  246. role="planner",
  247. collection="tasks",
  248. page_size=20,
  249. )
  250. progress = await workbench.progress_snapshot("root")
  251. assert len(progress["unique_observations"]) == 1
  252. await workbench.search_mission_context(
  253. root_trace_id="root",
  254. role="planner",
  255. collection="tasks",
  256. query="different semantic scope",
  257. page_size=20,
  258. )
  259. progress = await workbench.progress_snapshot("root")
  260. assert len(progress["unique_observations"]) == 2
  261. @pytest.mark.asyncio
  262. async def test_role_bootstrap_uses_semantic_targets_and_protects_artifact_identities() -> None:
  263. root = _root_task(children=["worker"])
  264. task = _worker_task()
  265. attempt = SimpleNamespace(attempt_id="attempt", failure=None)
  266. ledger = SimpleNamespace(
  267. revision=7,
  268. root_task_id="root-task",
  269. focused_task_id="worker",
  270. tasks={"root-task": root, "worker": task},
  271. attempts={"attempt": attempt},
  272. decisions={},
  273. operations={},
  274. )
  275. workbench = MissionWorkbench(
  276. task_store=_TaskStore(ledger),
  277. bindings=_Bindings(),
  278. snapshots=_Snapshots(),
  279. artifacts=_Artifacts(),
  280. contracts=_Contracts(_contract()),
  281. candidates=_Candidates(),
  282. )
  283. worker_request = RoleContextRequest(
  284. role=AgentRole.WORKER,
  285. root_trace_id="root",
  286. task_id="worker",
  287. spec_version=1,
  288. task_spec={},
  289. attempt_id="attempt",
  290. ledger_revision=7,
  291. )
  292. worker = await workbench.build(worker_request)
  293. encoded_worker = json.dumps(worker, ensure_ascii=False)
  294. assert worker["paragraph_targets"][0]["paragraph_target_key"].startswith("pt_")
  295. assert "paragraph_id" not in encoded_worker
  296. assert len(encoded_worker) < 32_000
  297. artifact_ref = ArtifactRef(
  298. uri="script-build://artifact-versions/77",
  299. kind="paragraph",
  300. version="77",
  301. digest="sha256:" + "a" * 64,
  302. )
  303. validator_request = RoleContextRequest(
  304. role=AgentRole.VALIDATOR,
  305. root_trace_id="root",
  306. task_id="worker",
  307. spec_version=1,
  308. task_spec={},
  309. attempt_id="attempt",
  310. ledger_revision=7,
  311. snapshot_id="snapshot",
  312. artifact_snapshot={
  313. "artifact_refs": [asdict(artifact_ref)],
  314. "evidence_refs": [],
  315. },
  316. )
  317. validator = await workbench.build(validator_request)
  318. encoded_validator = json.dumps(validator, ensure_ascii=False)
  319. assert "complete candidate prose" in encoded_validator
  320. assert "paragraph_id" not in encoded_validator
  321. assert "script-build://artifact-versions/77" not in encoded_validator
  322. assert "canonical_sha256" not in encoded_validator
  323. with pytest.raises(WorkbenchError) as physical:
  324. await workbench.read_mission_context(
  325. root_trace_id="root",
  326. role="planner",
  327. handle="script-build://artifact-versions/77",
  328. )
  329. assert physical.value.code == "CONTEXT_HANDLE_FORMAT_INVALID"
  330. unknown_handle = semantic_handle("unknown", "root", "not-present")
  331. with pytest.raises(WorkbenchError) as unknown:
  332. await workbench.read_mission_context(
  333. root_trace_id="root",
  334. role="planner",
  335. handle=unknown_handle,
  336. )
  337. assert unknown.value.code == "CONTEXT_HANDLE_UNKNOWN"
  338. with pytest.raises(WorkbenchError) as unauthorized:
  339. await workbench.read_mission_context(
  340. root_trace_id="root",
  341. role="worker",
  342. task_id="worker",
  343. attempt_id="attempt",
  344. handle=unknown_handle,
  345. )
  346. assert unauthorized.value.code == "CONTEXT_HANDLE_UNAUTHORIZED"