test_org_level_loop.py
6.12 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
"""U-017 tests: the full Org-level Loop wiring (judge -> record -> route ->
fix -> re-verify -> done), mocked at the subprocess boundary (CLAUDE.md
Testing constraint). Real coverage is in src/test_org_level_loop_e2e.py
(RUN_REAL_AGENT_E2E=1).
**F (2026-08-06)**: root-cause judgment now runs through
root_cause_trace.judge_root_cause_via_work_graph (a real Work Graph fan-out,
see docs/done/verifier-fan-out-jury-design.md), so the 3 perspective calls
are mocked at work_graph_execute.write_workflow_script/execute_work_graph
(same boundary as test_work_graph_orchestrate.py), not at
org_node_trigger.trigger_org_node -- that mock is now only used for the
fixer/verify calls, which are unchanged.
Run: python3 -m unittest src.test_org_level_loop -v
"""
import unittest
from unittest.mock import patch
from src import org_level_loop, root_cause_trace, schema, work_graph_execute
def _make_perspective_execute(perspective_category):
"""Fakes work_graph_execute.execute_work_graph for the 3 perspective
Work Nodes -- all agree on perspective_category, real schema-shaped
output (no more 'CATEGORY:'/'DETAIL:' text parsing)."""
def fake_execute(cwd, name, args, output_schema, timeout=300):
return {"category": perspective_category, "detail": "because reasons"}
return fake_execute
def _make_trigger(fix_ok=True):
"""Fakes org_node_trigger.trigger_org_node for the fixer/verify calls only."""
def fake(org_node, prompt, target_folder_path, timeout=300):
if org_node in ("coder", "planner"):
return "Fixed it."
if org_node == "validator":
return "VERDICT: PASS\nlooks fixed" if fix_ok else "VERDICT: FAIL\nstill broken"
raise AssertionError("unexpected call: " + org_node)
return fake
class RunFixLoopTestBase(unittest.TestCase):
def setUp(self):
# check_same_thread=False: the 3 root-cause perspective nodes form
# one concurrent layer under G, run via ThreadPoolExecutor.
self.conn = schema.connect(":memory:", check_same_thread=False)
schema.create_schema(self.conn)
self.conn.execute(
"INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
"VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
)
self.conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, loop_round, assigned_org_node) "
"VALUES ('task-1', 'proj-1', 'goal', 'goal', 'validating', 0, 'validator')"
)
self.conn.commit()
def tearDown(self):
self.conn.close()
def _run_fix_loop(self, perspective_category, fix_ok=True):
with patch.object(work_graph_execute, "write_workflow_script"), \
patch.object(
work_graph_execute, "execute_work_graph",
side_effect=_make_perspective_execute(perspective_category),
), \
patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger(fix_ok=fix_ok)):
return org_level_loop.run_fix_loop(
self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it"
)
class CoderCategoryFastLaneTest(RunFixLoopTestBase):
def test_coder_category_routes_fast_lane_and_reaches_done(self):
result = self._run_fix_loop("coder", fix_ok=True)
self.assertEqual(result["root_cause_category"], "coder")
self.assertEqual(result["route"], "fast_lane")
self.assertTrue(result["verified"])
self.assertEqual(result["final_status"], "done")
row = self.conn.execute("SELECT status, loop_round FROM task WHERE task_id = 'task-1'").fetchone()
self.assertEqual(row, ("done", 1))
bug_row = self.conn.execute(
"SELECT root_cause_category, suggested_route FROM bug_record WHERE task_id = 'task-1'"
).fetchone()
self.assertEqual(bug_row, ("coder", "fast_lane"))
class PlanningCategoryFullReplanTest(RunFixLoopTestBase):
def test_planning_category_routes_full_replan(self):
result = self._run_fix_loop("planning", fix_ok=True)
self.assertEqual(result["route"], "full_replan")
self.assertEqual(result["final_status"], "done")
assigned = self.conn.execute("SELECT assigned_org_node FROM task WHERE task_id = 'task-1'").fetchone()[0]
# after close_task_done the task is 'done'; assigned_org_node was set
# to 'planner' by route_org_loop and close_task_done doesn't touch it
self.assertEqual(assigned, "planner")
class VerificationFailsTest(RunFixLoopTestBase):
def test_fix_not_verified_leaves_task_out_of_done(self):
result = self._run_fix_loop("coder", fix_ok=False)
self.assertFalse(result["verified"])
self.assertNotEqual(result["final_status"], "done")
status = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0]
self.assertNotEqual(status, "done")
class InconclusiveDisagreementTest(RunFixLoopTestBase):
"""Real Work Graph fan-out still surfaces InconclusiveEscalateError the
same way the closure-based version did -- run_fix_loop adds no fallback
of its own (04-P: disagreement escalates to a human, it isn't silently
routed anywhere)."""
def test_three_way_split_raises_and_creates_no_bug_record(self):
categories = iter(["planning", "coder", "skill_defect"])
def fake_execute(cwd, name, args, output_schema, timeout=300):
return {"category": next(categories), "detail": "d"}
with patch.object(work_graph_execute, "write_workflow_script"), \
patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute), \
patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger(fix_ok=True)):
with self.assertRaises(root_cause_trace.InconclusiveEscalateError):
org_level_loop.run_fix_loop(
self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it"
)
count = self.conn.execute("SELECT COUNT(*) FROM bug_record").fetchone()[0]
self.assertEqual(count, 0)
if __name__ == "__main__":
unittest.main()