"""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()