test_goal_compatibility.py 1.5 KB

12345678910111213141516171819202122232425262728293031323334
  1. import pytest
  2. from agent.trace.goal_models import GoalTree
  3. from agent.trace.models import Trace
  4. from agent.trace.store import FileSystemTraceStore
  5. def test_goal_tree_legacy_cascade_and_explicit_no_cascade():
  6. legacy = GoalTree(mission="m")
  7. parent = legacy.add_goals(["parent"])[0]
  8. children = legacy.add_goals(["a", "b"], parent_id=parent.id)
  9. legacy.complete(children[0].id, "a")
  10. legacy.complete(children[1].id, "b")
  11. assert legacy.find(parent.id).status == "completed"
  12. explicit = GoalTree(mission="m")
  13. parent = explicit.add_goals(["parent"])[0]
  14. children = explicit.add_goals(["a", "b"], parent_id=parent.id)
  15. explicit.complete(children[0].id, "a", cascade_completion=False)
  16. explicit.complete(children[1].id, "b", cascade_completion=False)
  17. assert explicit.find(parent.id).status == "pending"
  18. @pytest.mark.asyncio
  19. async def test_trace_store_can_disable_goal_cascade(tmp_path):
  20. store = FileSystemTraceStore(str(tmp_path))
  21. await store.create_trace(Trace(trace_id="root", mode="agent"))
  22. tree = GoalTree(mission="m")
  23. parent = tree.add_goals(["parent"])[0]
  24. children = tree.add_goals(["a", "b"], parent_id=parent.id)
  25. await store.update_goal_tree("root", tree)
  26. await store.update_goal("root", children[0].id, cascade_completion=False, status="completed")
  27. await store.update_goal("root", children[1].id, cascade_completion=False, status="completed")
  28. assert (await store.get_goal_tree("root")).find(parent.id).status == "pending"