root_cause_trace.py
6.1 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
"""p_validator_root_cause_trace (04-P §4.3): U-015 scope.
"血缘读取"（tracing_lineage: reading SDD/PRD/BRD versions + Coder implementation
records） is the caller's job -- perspective_fns are expected to already have
whatever lineage context they need baked in (e.g. closures over artifact
content read beforehand) by the time judge_root_cause calls them. This module
only drives judging_root_cause's multi-perspective vote and the
bug_recorded/inconclusive_escalate branch, matching 04-P's actual state-table
scope for that transition.
Each perspective_fn takes no arguments and returns {"category": "planning"|
"coder"|"skill_defect", "detail": str} -- one judgment from one of the three
perspectives 04-P names (正确性视角/需求完整性视角/Skill 系统性缺陷视角). Which
concrete question each perspective asks is a prompt-engineering concern for
the caller (see real_validator_perspective below for the real-call version),
not this module's.
"""
from collections import Counter
from src import bug_record
class InconclusiveEscalateError(Exception):
"""04-P: judging_root_cause -> inconclusive_escalate (<2/3 perspectives
agree). Carries the raw per-perspective judgments for human review
("附带三视角原始判断供人工裁决")."""
def __init__(self, judgments):
self.judgments = judgments
super().__init__("perspectives did not reach 2/3 agreement: " + repr(judgments))
def judge_root_cause(perspective_fns):
"""Runs all perspective_fns, requires >=2/3 to agree on the same
category (04-P guard: "至少 2/3 视角一致才定案"). Returns
{"root_cause_category", "root_cause_detail", "judgments"} on agreement,
raises InconclusiveEscalateError otherwise.
"""
judgments = [fn() for fn in perspective_fns]
counts = Counter(j["category"] for j in judgments)
winner, count = counts.most_common(1)[0]
if count * 2 < len(judgments): # strictly "at least 2/3", not just plurality
raise InconclusiveEscalateError(judgments)
detail = " | ".join(j["detail"] for j in judgments if j["category"] == winner)
return {"root_cause_category": winner, "root_cause_detail": detail, "judgments": judgments}
def trace_root_cause_test_failed(conn, task_id, description, perspective_fns, suggested_route, related_artifact_ids=None):
"""04-P: test_failed -> tracing_lineage -> judging_root_cause ->
bug_recorded. Creates a brand-new bug_record (test_failed origin never
updates an existing one, unlike human_reported -- see 04-P's own note
on this distinction). Re-raises InconclusiveEscalateError unchanged if
judge_root_cause can't reach agreement.
"""
result = judge_root_cause(perspective_fns)
created = bug_record.create_bug_record(
conn, task_id=task_id, description=description,
root_cause_category=result["root_cause_category"], root_cause_detail=result["root_cause_detail"],
suggested_route=suggested_route, related_artifact_ids=related_artifact_ids,
)
return {**created, **result}
def trace_root_cause_human_reported(conn, bug_id, perspective_fns, suggested_route):
"""04-P: human_reported (already claimed via bug_record.claim_human_reported_bug,
status=triaging) -> tracing_lineage -> judging_root_cause -> bug_recorded.
Field-completes the existing bug_record instead of creating a new one.
"""
result = judge_root_cause(perspective_fns)
bug_record.complete_human_reported_triage(
conn, bug_id, result["root_cause_category"], result["root_cause_detail"], suggested_route
)
return result
_PERSPECTIVE_FRAMINGS = {
"correctness": (
"CORRECTNESS PERSPECTIVE: does the Coder's implementation match what the SDD/PRD actually "
"specify? If the spec was followed correctly and the bug is still there, this is not your "
"category to claim."
),
"requirement_completeness": (
"REQUIREMENT-COMPLETENESS PERSPECTIVE: does the PRD/SDD actually define the behavior the bug "
"report is about at all? If the requirement was never specified (a genuine planning gap, not "
"an implementation mistake), that's your category to claim."
),
"skill_defect": (
"SKILL-DEFECT PERSPECTIVE: is this a one-off mistake in this specific task, or does it look "
"like the underlying Skill (the reusable methodology/prompt an Org Agent follows) is "
"systematically wrong in a way that would recur on other tasks?"
),
}
def real_validator_perspective(perspective_name, bug_description, context, target_folder_path, trigger_fn, timeout=300):
"""Builds a real perspective_fn (for use with judge_root_cause) that calls
the deployed Validator agent, framed from one of the three named
perspectives, and parses its 'CATEGORY: ...' / 'DETAIL: ...' response.
`trigger_fn` is org_node_trigger.trigger_org_node, passed in explicitly
(not imported/called directly) so tests can inject a fake without
mocking a module-global.
"""
framing = _PERSPECTIVE_FRAMINGS[perspective_name]
prompt = (
"{framing}\n\nBug: {bug_description}\n\nContext:\n{context}\n\n"
"Reply with exactly two lines:\n"
"CATEGORY: planning|coder|skill_defect\n"
"DETAIL: <one sentence, cite the specific SDD section or code location>"
).format(framing=framing, bug_description=bug_description, context=context)
def perspective_fn():
response = trigger_fn("validator", prompt, target_folder_path, timeout=timeout)
category = None
detail = ""
for line in response.splitlines():
stripped = line.strip()
if stripped.upper().startswith("CATEGORY:"):
value = stripped.split(":", 1)[1].strip().lower()
if value in ("planning", "coder", "skill_defect"):
category = value
elif stripped.upper().startswith("DETAIL:"):
detail = stripped.split(":", 1)[1].strip()
if category is None:
raise ValueError("could not parse a CATEGORY line from validator response: " + response)
return {"category": category, "detail": detail or response}
return perspective_fn