test_root_cause_trace.py
12.9 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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
"""U-015 tests: judge_root_cause's 2/3-majority logic and the test_failed/
human_reported wiring, with fake perspective_fns (CLAUDE.md Testing
constraint -- real perspectives shell out to `claude`, mocked here). Real
coverage is in src/test_root_cause_trace_e2e.py (RUN_REAL_AGENT_E2E=1).
Run: python3 -m unittest src.test_root_cause_trace -v
"""
import json
import os
import tempfile
import threading
import time
import unittest
from unittest.mock import patch
from src import bug_record, org_node_trigger, root_cause_trace, schema
def _fixed(category, detail="detail"):
return lambda: {"category": category, "detail": detail}
class JudgeRootCauseTest(unittest.TestCase):
def test_unanimous_agreement(self):
result = root_cause_trace.judge_root_cause(
[_fixed("coder", "d1"), _fixed("coder", "d2"), _fixed("coder", "d3")]
)
self.assertEqual(result["root_cause_category"], "coder")
self.assertIn("d1", result["root_cause_detail"])
self.assertIn("d2", result["root_cause_detail"])
def test_two_of_three_agreement_wins(self):
result = root_cause_trace.judge_root_cause(
[_fixed("planning", "d1"), _fixed("planning", "d2"), _fixed("coder", "d3")]
)
self.assertEqual(result["root_cause_category"], "planning")
self.assertNotIn("d3", result["root_cause_detail"])
def test_three_way_split_is_inconclusive(self):
with self.assertRaises(root_cause_trace.InconclusiveEscalateError) as ctx:
root_cause_trace.judge_root_cause(
[_fixed("planning"), _fixed("coder"), _fixed("skill_defect")]
)
self.assertEqual(len(ctx.exception.judgments), 3)
def test_below_two_thirds_with_more_perspectives_is_inconclusive(self):
# 2 out of 5 agreeing is not >=2/3 (04-P: "至少 2/3 视角一致")
with self.assertRaises(root_cause_trace.InconclusiveEscalateError):
root_cause_trace.judge_root_cause(
[_fixed("coder"), _fixed("coder"), _fixed("planning"), _fixed("planning"), _fixed("skill_defect")]
)
class TraceRootCauseTestFailedTest(unittest.TestCase):
def setUp(self):
self.conn = schema.connect(":memory:")
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, assigned_org_node) "
"VALUES ('task-1', 'proj-1', 'goal', 'goal', 'validating', 'validator')"
)
self.conn.commit()
def tearDown(self):
self.conn.close()
def test_agreement_creates_bug_record(self):
result = root_cause_trace.trace_root_cause_test_failed(
self.conn, "task-1", "wall collision misjudged",
[_fixed("coder", "SDD §3.2"), _fixed("coder", "code line 42"), _fixed("planning", "no")],
suggested_route="fast_lane",
)
self.assertEqual(result["root_cause_category"], "coder")
row = self.conn.execute(
"SELECT source, status, task_id FROM bug_record WHERE bug_id = ?", (result["bug_id"],)
).fetchone()
self.assertEqual(row, ("validator", "routed", "task-1"))
def test_disagreement_raises_and_creates_no_record(self):
with self.assertRaises(root_cause_trace.InconclusiveEscalateError):
root_cause_trace.trace_root_cause_test_failed(
self.conn, "task-1", "x",
[_fixed("planning"), _fixed("coder"), _fixed("skill_defect")],
suggested_route="fast_lane",
)
count = self.conn.execute("SELECT COUNT(*) FROM bug_record").fetchone()[0]
self.assertEqual(count, 0)
class TraceRootCauseHumanReportedTest(unittest.TestCase):
def setUp(self):
self.conn = schema.connect(":memory:")
schema.create_schema(self.conn)
self.conn.execute(
"INSERT INTO bug_record (bug_id, source, description, status) "
"VALUES ('bug-1', 'human', 'something looks broken', 'triaging')"
)
self.conn.commit()
def tearDown(self):
self.conn.close()
def test_completes_existing_record_not_new_one(self):
root_cause_trace.trace_root_cause_human_reported(
self.conn, "bug-1", [_fixed("planning", "gap"), _fixed("planning", "gap2"), _fixed("coder")],
suggested_route="full_replan",
)
count = self.conn.execute("SELECT COUNT(*) FROM bug_record").fetchone()[0]
self.assertEqual(count, 1)
row = self.conn.execute("SELECT status, root_cause_category FROM bug_record WHERE bug_id = 'bug-1'").fetchone()
self.assertEqual(row, ("routed", "planning"))
class RealValidatorPerspectiveParsingTest(unittest.TestCase):
def test_parses_category_and_detail(self):
fake_trigger = lambda org_node, prompt, target, timeout=300: "CATEGORY: coder\nDETAIL: off-by-one in bounds check"
perspective_fn = root_cause_trace.real_validator_perspective(
"correctness", "bug desc", "context", "/tmp/p", trigger_fn=fake_trigger
)
result = perspective_fn()
self.assertEqual(result, {"category": "coder", "detail": "off-by-one in bounds check"})
def test_missing_category_line_raises(self):
fake_trigger = lambda org_node, prompt, target, timeout=300: "I have no idea what happened."
perspective_fn = root_cause_trace.real_validator_perspective(
"correctness", "bug desc", "context", "/tmp/p", trigger_fn=fake_trigger
)
with self.assertRaises(ValueError):
perspective_fn()
# F (2026-08-06): Work Graph fan-out version -- each perspective becomes a
# real Work Node (mocked at the claude subprocess boundary, same as
# test_work_graph_orchestrate.py) instead of a Python closure.
class JudgeRootCauseViaWorkGraphTestBase(unittest.TestCase):
def setUp(self):
self.home_tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.home_tmp.cleanup)
self.old_graphlemon_home = os.environ.get("GRAPHLEMON_HOME")
os.environ["GRAPHLEMON_HOME"] = self.home_tmp.name
self.addCleanup(self._restore_graphlemon_home)
# check_same_thread=False: the three 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, assigned_org_node) "
"VALUES ('task-1', 'proj-1', 'goal', 'goal', 'validating', 'validator')"
)
self.conn.commit()
def _restore_graphlemon_home(self):
if self.old_graphlemon_home is None:
os.environ.pop("GRAPHLEMON_HOME", None)
else:
os.environ["GRAPHLEMON_HOME"] = self.old_graphlemon_home
def tearDown(self):
self.conn.close()
def _perspective_for_prompt(self, prompt):
for name, framing in root_cause_trace._PERSPECTIVE_FRAMINGS.items():
if framing in prompt:
return name
raise AssertionError("could not identify perspective for prompt " + prompt)
class RealWorkNodesPerPerspectiveTest(JudgeRootCauseViaWorkGraphTestBase):
"""Acceptance criterion 1: three independent Work Node executions (real
work_node_execution rows, real execution_event logs), not three closure
calls inside one function."""
def test_three_perspectives_are_three_real_work_node_rows(self):
def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300):
perspective = self._perspective_for_prompt(prompt)
return {"category": "coder", "detail": "agreed via " + perspective}
with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger):
result = root_cause_trace.judge_root_cause_via_work_graph(
self.conn, "task-1", "/tmp/proj-1", "bug desc", "context",
)
self.assertEqual(result["root_cause_category"], "coder")
self.assertEqual(len(result["judgments"]), 3)
rows = self.conn.execute(
"SELECT node_id, status, output_payload FROM work_node_execution WHERE run_id = ("
" SELECT run_id FROM work_graph_run WHERE task_id = 'task-1'"
")"
).fetchall()
self.assertEqual(
{r[0] for r in rows},
{"perspective-correctness", "perspective-requirement_completeness", "perspective-skill_defect"},
)
for node_id, status, output_payload in rows:
self.assertEqual(status, "done")
self.assertEqual(json.loads(output_payload)["category"], "coder")
class JudgmentRuleEquivalenceTest(JudgeRootCauseViaWorkGraphTestBase):
"""Acceptance criterion 2: same >=2/3 rule, same results, as the
closure-based judge_root_cause on equivalent inputs -- both call the
same shared _tally_perspectives."""
def _run_with_categories(self, categories_by_perspective):
def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300):
perspective = self._perspective_for_prompt(prompt)
return {"category": categories_by_perspective[perspective], "detail": "d-" + perspective}
with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger):
return root_cause_trace.judge_root_cause_via_work_graph(
self.conn, "task-1", "/tmp/proj-1", "bug desc", "context",
)
def test_two_of_three_agreement_wins(self):
result = self._run_with_categories(
{"correctness": "planning", "requirement_completeness": "planning", "skill_defect": "coder"}
)
self.assertEqual(result["root_cause_category"], "planning")
closure_result = root_cause_trace.judge_root_cause(
[_fixed("planning"), _fixed("planning"), _fixed("coder")]
)
self.assertEqual(result["root_cause_category"], closure_result["root_cause_category"])
def test_three_way_split_is_inconclusive(self):
with self.assertRaises(root_cause_trace.InconclusiveEscalateError):
self._run_with_categories(
{"correctness": "planning", "requirement_completeness": "coder", "skill_defect": "skill_defect"}
)
with self.assertRaises(root_cause_trace.InconclusiveEscalateError):
root_cause_trace.judge_root_cause([_fixed("planning"), _fixed("coder"), _fixed("skill_defect")])
class IndependentCleanContextTest(JudgeRootCauseViaWorkGraphTestBase):
"""Acceptance criterion 3: each perspective node's input_payload only
carries what it needs to judge -- not other perspectives' reasoning."""
def test_input_payload_has_no_cross_perspective_content(self):
with patch.object(
org_node_trigger, "trigger_org_node_structured",
return_value={"category": "coder", "detail": "d"},
):
root_cause_trace.judge_root_cause_via_work_graph(
self.conn, "task-1", "/tmp/proj-1", "bug desc", "context",
)
rows = self.conn.execute("SELECT node_id, input_payload FROM work_node_execution").fetchall()
self.assertEqual(len(rows), 3)
for node_id, input_payload in rows:
payload = json.loads(input_payload)
self.assertEqual(payload, {"node_id": node_id})
class ConcurrentPerspectiveExecutionTest(JudgeRootCauseViaWorkGraphTestBase):
"""G is already in place -- three mutually-independent perspective
nodes land in one topological layer, so they run concurrently, not
queued one after another."""
def test_three_perspective_calls_overlap_in_wall_clock_time(self):
events = []
events_lock = threading.Lock()
def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300):
start = time.monotonic()
time.sleep(0.05)
end = time.monotonic()
with events_lock:
events.append((start, end))
return {"category": "coder", "detail": "d"}
with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger):
root_cause_trace.judge_root_cause_via_work_graph(
self.conn, "task-1", "/tmp/proj-1", "bug desc", "context",
)
self.assertEqual(len(events), 3)
# at least two of the three calls must have overlapped
overlaps = sum(
1 for i in range(len(events)) for j in range(i + 1, len(events))
if events[i][0] < events[j][1] and events[j][0] < events[i][1]
)
self.assertGreater(overlaps, 0)
if __name__ == "__main__":
unittest.main()