| 12345678910111213141516171819202122232425262728293031323334 |
- import pytest
- from agent.trace.goal_models import GoalTree
- from agent.trace.models import Trace
- from agent.trace.store import FileSystemTraceStore
- def test_goal_tree_legacy_cascade_and_explicit_no_cascade():
- legacy = GoalTree(mission="m")
- parent = legacy.add_goals(["parent"])[0]
- children = legacy.add_goals(["a", "b"], parent_id=parent.id)
- legacy.complete(children[0].id, "a")
- legacy.complete(children[1].id, "b")
- assert legacy.find(parent.id).status == "completed"
- explicit = GoalTree(mission="m")
- parent = explicit.add_goals(["parent"])[0]
- children = explicit.add_goals(["a", "b"], parent_id=parent.id)
- explicit.complete(children[0].id, "a", cascade_completion=False)
- explicit.complete(children[1].id, "b", cascade_completion=False)
- assert explicit.find(parent.id).status == "pending"
- @pytest.mark.asyncio
- async def test_trace_store_can_disable_goal_cascade(tmp_path):
- store = FileSystemTraceStore(str(tmp_path))
- await store.create_trace(Trace(trace_id="root", mode="agent"))
- tree = GoalTree(mission="m")
- parent = tree.add_goals(["parent"])[0]
- children = tree.add_goals(["a", "b"], parent_id=parent.id)
- await store.update_goal_tree("root", tree)
- await store.update_goal("root", children[0].id, cascade_completion=False, status="completed")
- await store.update_goal("root", children[1].id, cascade_completion=False, status="completed")
- assert (await store.get_goal_tree("root")).find(parent.id).status == "pending"
|