SILVER
/
NODEDC_1C
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Связать GUI detector results с repair handoff агента

This commit is contained in:
dctouch 2026-06-01 20:39:03 +03:00
parent f485f01ec2
commit 5426f9a24c
2 changed files with 353 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

248
scripts/stage_agent_loop.py
View File

@ -1230,6 +1230,168 @@ def load_gui_review_detector_gate(review_dir: Path) -> dict[str, Any]:
}
def unique_strings(values: list[str]) -> list[str]:
result: list[str] = []
for value in values:
text = str(value or "").strip()
if text and text not in result:
result.append(text)
return result
def stage_patch_target_looks_like_file(value: str) -> bool:
normalized = str(value or "").replace("\\", "/").strip()
if not normalized or " " in normalized:
return False
prefixes = (
"llm_normalizer/",
"scripts/",
"docs/",
".codex/",
"graphify-out/",
)
return normalized.startswith(prefixes)
def issue_catalog_contract_for_stage(issue_code: str, catalog: dict[str, Any]) -> dict[str, Any]:
entry = dcl.issue_catalog_entry(issue_code, catalog)
return {
"issue_code": issue_code,
"severity": entry.get("severity"),
"root_layers": dcl.normalize_string_list(entry.get("root_layers")),
"expected_answer_contract": dcl.issue_acceptance_contract_name(issue_code, entry),
"detectors": dcl.normalize_string_list(entry.get("detectors")),
"allowed_patch_targets": dcl.normalize_string_list(entry.get("allowed_patch_targets")),
"forbidden_patch_targets": dcl.normalize_string_list(entry.get("forbidden_patch_targets")),
"rerun_matrix": dcl.normalize_string_list(entry.get("rerun_matrix")),
}
def enrich_stage_gui_repair_target(
target: dict[str, Any],
*,
catalog: dict[str, Any],
fallback_evidence_paths: list[str],
) -> dict[str, Any]:
issue_code = str(target.get("issue_code") or "").strip()
problem_layer = str(target.get("problem_layer") or "other").strip() or "other"
catalog_contract = issue_catalog_contract_for_stage(issue_code, catalog) if issue_code else {}
root_layers = unique_strings(
dcl.normalize_string_list(target.get("root_cause_layers"))
+ dcl.normalize_string_list(target.get("suggested_root_cause_layers"))
+ dcl.normalize_string_list(catalog_contract.get("root_layers"))
+ [problem_layer]
)
allowed_patch_targets = unique_strings(
dcl.normalize_string_list(target.get("allowed_patch_targets"))
+ dcl.normalize_string_list(catalog_contract.get("allowed_patch_targets"))
)
forbidden_patch_targets = unique_strings(
dcl.normalize_string_list(target.get("forbidden_patch_targets"))
+ dcl.normalize_string_list(catalog_contract.get("forbidden_patch_targets"))
)
rerun_matrix = unique_strings(
dcl.normalize_string_list(target.get("rerun_matrix"))
+ dcl.normalize_string_list(catalog_contract.get("rerun_matrix"))
)
candidate_files = unique_strings(
dcl.normalize_string_list(target.get("candidate_files"))
+ [
path
for path in allowed_patch_targets
if stage_patch_target_looks_like_file(path)
]
+ list(dcl.REPAIR_TARGET_FILE_HINTS.get(problem_layer) or dcl.REPAIR_TARGET_FILE_HINTS["other"])
)
evidence_paths = unique_strings(dcl.normalize_string_list(target.get("evidence_paths")) + fallback_evidence_paths)
expected_contract = (
str(target.get("expected_business_answer_contract") or "").strip()
or str(catalog_contract.get("expected_answer_contract") or "").strip()
or None
)
target_id = str(target.get("target_id") or "").strip()
if not target_id:
target_id = f"gui:{issue_code or 'unknown'}:{problem_layer}"
return {
**target,
"target_id": target_id,
"problem_type": target.get("problem_type") or problem_layer,
"root_cause_layers": root_layers,
"expected_business_answer_contract": expected_contract,
"detectors": dcl.normalize_string_list(catalog_contract.get("detectors")),
"allowed_patch_targets": allowed_patch_targets,
"forbidden_patch_targets": forbidden_patch_targets,
"rerun_matrix": rerun_matrix,
"candidate_files": candidate_files,
"evidence_paths": evidence_paths,
}
def build_stage_gui_repair_contract(
*,
summary: dict[str, Any],
review: dict[str, Any],
) -> dict[str, Any]:
latest = summary.get("latest_gui_review") if isinstance(summary.get("latest_gui_review"), dict) else {}
catalog = dcl.load_issue_catalog()
raw_targets = review.get("repair_targets") if isinstance(review.get("repair_targets"), list) else []
fallback_evidence_paths = unique_strings(
[
latest.get("review_json"),
latest.get("review_markdown"),
latest.get("repair_targets_json"),
latest.get("conversation_pairs_json"),
latest.get("detector_results_json"),
]
)
targets = [
enrich_stage_gui_repair_target(target, catalog=catalog, fallback_evidence_paths=fallback_evidence_paths)
for target in raw_targets
if isinstance(target, dict)
]
issue_codes = unique_strings([str(target.get("issue_code") or "") for target in targets])
issue_catalog_contracts = {
issue_code: issue_catalog_contract_for_stage(issue_code, catalog)
for issue_code in issue_codes
}
severity_counts = Counter(str(target.get("severity") or "unknown") for target in targets)
rerun_matrix = unique_strings([item for target in targets for item in dcl.normalize_string_list(target.get("rerun_matrix"))])
candidate_files = unique_strings([item for target in targets for item in dcl.normalize_string_list(target.get("candidate_files"))])
focus = None
if targets:
focus = {
"focus_id": f"gui:{latest.get('run_id') or review.get('run_id') or 'review'}:{issue_codes[0] if issue_codes else 'unknown'}",
"issue_codes": issue_codes,
"root_cause_layers": unique_strings(
[item for target in targets for item in dcl.normalize_string_list(target.get("root_cause_layers"))]
),
"allowed_patch_targets": unique_strings(
[item for target in targets for item in dcl.normalize_string_list(target.get("allowed_patch_targets"))]
),
"forbidden_patch_targets": unique_strings(
[item for target in targets for item in dcl.normalize_string_list(target.get("forbidden_patch_targets"))]
),
"rerun_matrix": rerun_matrix,
"candidate_files": candidate_files,
"target_ids": unique_strings([str(target.get("target_id") or "") for target in targets]),
"severity": targets[0].get("severity"),
"problem_type": targets[0].get("problem_type"),
"target_count": len(targets),
"rank_reason": "gui_review_primary_repair_targets",
}
return {
"schema_version": "stage_gui_repair_contract_v1",
"target_count": len(targets),
"severity_counts": dict(sorted(severity_counts.items())),
"issue_codes": issue_codes,
"issue_catalog_contracts": issue_catalog_contracts,
"rerun_matrix": rerun_matrix,
"candidate_files": candidate_files,
"targets": targets,
"priority_foci": [focus] if focus else [],
}
def build_gui_review_stage_summary(
*,
stage_manifest: dict[str, Any],
@ -1331,13 +1493,9 @@ def build_gui_review_stage_summary(
def build_stage_repair_handoff(summary: dict[str, Any], review: dict[str, Any]) -> dict[str, Any]:
latest = summary.get("latest_gui_review") if isinstance(summary.get("latest_gui_review"), dict) else {}
repair_targets = review.get("repair_targets") if isinstance(review.get("repair_targets"), list) else []
findings = review.get("findings") if isinstance(review.get("findings"), list) else []
ordered_targets = [
target
for target in repair_targets
if isinstance(target, dict)
]
repair_contract = build_stage_gui_repair_contract(summary=summary, review=review)
ordered_targets = repair_contract.get("targets") if isinstance(repair_contract.get("targets"), list) else []
primary_targets = ordered_targets[:5]
target_issue_codes = {str(target.get("issue_code") or "") for target in primary_targets}
sample_findings = [
@ -1349,6 +1507,20 @@ def build_stage_repair_handoff(summary: dict[str, Any], review: dict[str, Any])
or any(str(code) in target_issue_codes for code in finding.get("issue_codes", []))
)
][:8]
detector_results_summary = (
latest.get("detector_results_summary") if isinstance(latest.get("detector_results_summary"), dict) else {}
)
detector_results_for_gate = {"summary": detector_results_summary, "results": []}
assigned_focus = (
repair_contract.get("priority_foci", [None])[0]
if isinstance(repair_contract.get("priority_foci"), list) and repair_contract.get("priority_foci")
else None
)
auto_propose_gate = dcl.evaluate_auto_coder_gate(
repair_contract,
assigned_focus if isinstance(assigned_focus, dict) else None,
detector_results=detector_results_for_gate if detector_results_summary else None,
)
return {
"schema_version": "stage_gui_repair_handoff_v1",
"stage_id": summary.get("stage_id"),
@ -1363,6 +1535,13 @@ def build_stage_repair_handoff(summary: dict[str, Any], review: dict[str, Any])
"detector_candidates_json": latest.get("detector_candidates_json"),
"detector_results_json": latest.get("detector_results_json"),
"detector_results_status": latest.get("detector_results_status"),
"detector_results_summary": detector_results_summary,
"issue_codes": repair_contract.get("issue_codes"),
"issue_catalog_contracts": repair_contract.get("issue_catalog_contracts"),
"rerun_matrix": repair_contract.get("rerun_matrix"),
"candidate_files": repair_contract.get("candidate_files"),
"repair_contract": repair_contract,
"auto_propose_gate": auto_propose_gate,
"primary_repair_targets": primary_targets,
"sample_findings": sample_findings,
}
@ -1383,9 +1562,34 @@ def build_stage_repair_handoff_markdown(handoff: dict[str, Any]) -> str:
f"- repair_targets_json: `{handoff.get('repair_targets_json')}`",
f"- detector_results_status: `{handoff.get('detector_results_status') or 'n/a'}`",
f"- detector_results_json: `{handoff.get('detector_results_json') or 'n/a'}`",
f"- issue_codes: `{', '.join(dcl.normalize_string_list(handoff.get('issue_codes'))) or 'n/a'}`",
f"- rerun_matrix: `{', '.join(dcl.normalize_string_list(handoff.get('rerun_matrix'))) or 'n/a'}`",
"",
"## Primary Repair Targets",
]
detector_summary = handoff.get("detector_results_summary") if isinstance(handoff.get("detector_results_summary"), dict) else {}
if detector_summary:
lines.extend(
[
"## Detector Gate",
f"- status: `{detector_summary.get('status') or 'n/a'}`",
f"- counts: `pass={detector_summary.get('pass') or 0}, fail={detector_summary.get('fail') or 0}, review={detector_summary.get('review') or 0}, skipped={detector_summary.get('skipped') or 0}`",
"",
]
)
auto_gate = handoff.get("auto_propose_gate") if isinstance(handoff.get("auto_propose_gate"), dict) else {}
if auto_gate:
lines.extend(
[
"## Auto-Propose Gate",
f"- allowed: `{bool(auto_gate.get('allowed'))}`",
f"- reason: `{auto_gate.get('reason') or 'n/a'}`",
f"- focus_id: `{auto_gate.get('focus_id') or 'n/a'}`",
]
)
blocking = dcl.normalize_string_list(auto_gate.get("blocking_reasons"))
lines.extend([f"- blocking_reason: `{item}`" for item in blocking[:12]] or ["- blocking_reason: `none`"])
lines.append("")
lines.append("## Primary Repair Targets")
targets = handoff.get("primary_repair_targets") if isinstance(handoff.get("primary_repair_targets"), list) else []
if not targets:
lines.append("- no repair targets")
@ -1393,9 +1597,16 @@ def build_stage_repair_handoff_markdown(handoff: dict[str, Any]) -> str:
for target in targets:
if not isinstance(target, dict):
continue
lines.append(
lines.extend(
[
f"- `{target.get('severity')}` `{target.get('problem_layer')}` / `{target.get('issue_code')}`: "
f"{target.get('occurrences')} occurrence(s)"
f"{target.get('occurrences')} occurrence(s)",
f" expected_contract: `{target.get('expected_business_answer_contract') or 'n/a'}`",
f" allowed_patch_targets: `{', '.join(dcl.normalize_string_list(target.get('allowed_patch_targets'))) or 'n/a'}`",
f" forbidden_patch_targets: `{', '.join(dcl.normalize_string_list(target.get('forbidden_patch_targets'))) or 'n/a'}`",
f" rerun_matrix: `{', '.join(dcl.normalize_string_list(target.get('rerun_matrix'))) or 'n/a'}`",
f" candidate_files: `{', '.join(dcl.normalize_string_list(target.get('candidate_files'))) or 'n/a'}`",
]
)
lines.extend(["", "## Sample Findings"])
findings = handoff.get("sample_findings") if isinstance(handoff.get("sample_findings"), list) else []
@ -1423,7 +1634,10 @@ def save_stage_repair_handoff(stage_dir: Path, handoff: dict[str, Any]) -> None:
def enrich_repair_target_for_coder(target: dict[str, Any]) -> dict[str, Any]:
problem_layer = str(target.get("problem_layer") or "other").strip() or "other"
candidate_files = list(dcl.REPAIR_TARGET_FILE_HINTS.get(problem_layer) or dcl.REPAIR_TARGET_FILE_HINTS["other"])
candidate_files = unique_strings(
dcl.normalize_string_list(target.get("candidate_files"))
+ list(dcl.REPAIR_TARGET_FILE_HINTS.get(problem_layer) or dcl.REPAIR_TARGET_FILE_HINTS["other"])
)
return {
**target,
"candidate_files": candidate_files,
@ -1463,12 +1677,25 @@ def build_stage_repair_iteration_plan(
"source_handoff": repo_relative(stage_dir / "stage_repair_handoff.json"),
"source_review_markdown": handoff.get("review_markdown"),
"source_repair_targets_json": handoff.get("repair_targets_json"),
"source_detector_results_json": handoff.get("detector_results_json"),
"run_id": handoff.get("run_id"),
"next_action": handoff.get("next_action"),
"overall_business_status": handoff.get("overall_business_status"),
"p0_findings": handoff.get("p0_findings"),
"p1_findings": handoff.get("p1_findings"),
"question_quality_score": handoff.get("question_quality_score"),
"detector_results_status": handoff.get("detector_results_status"),
"detector_results_summary": (
handoff.get("detector_results_summary") if isinstance(handoff.get("detector_results_summary"), dict) else {}
),
"issue_codes": dcl.normalize_string_list(handoff.get("issue_codes")),
"issue_catalog_contracts": (
handoff.get("issue_catalog_contracts") if isinstance(handoff.get("issue_catalog_contracts"), dict) else {}
),
"rerun_matrix": dcl.normalize_string_list(handoff.get("rerun_matrix")),
"auto_propose_gate": (
handoff.get("auto_propose_gate") if isinstance(handoff.get("auto_propose_gate"), dict) else {}
),
"primary_repair_targets": enriched_targets,
"candidate_files": candidate_files,
"sample_findings": [finding for finding in sample_findings if isinstance(finding, dict)][:8],
@ -1480,6 +1707,7 @@ def build_stage_repair_iteration_plan(
"ingest the new assistant-stage1 run id with stage_agent_loop.py ingest-gui-run",
"accept only when P0 is zero and P1 noise is either cleared or explicitly bounded",
],
"rerun_matrix": dcl.normalize_string_list(handoff.get("rerun_matrix")),
"ingest_command_template": (
"python scripts/stage_agent_loop.py ingest-gui-run "
"--manifest <stage_manifest.json> --run-id assistant-stage1-<new_id>"

114
scripts/test_stage_agent_loop.py
View File

@ -519,6 +519,110 @@ class StageAgentLoopTests(unittest.TestCase):
self.assertEqual(handoff["primary_repair_targets"][0]["issue_code"], "business_direct_answer_missing")
self.assertEqual(handoff["sample_findings"][0]["turn_index"], 19)
def test_stage_repair_handoff_enriches_gui_targets_with_catalog_contract(self) -> None:
summary = {
"stage_id": "agent_loop",
"next_action": "continue_repair_from_gui_review_p0",
"latest_gui_review": {
"run_id": "assistant-stage1-test",
"overall_business_status": "fail",
"p0_findings": 1,
"p1_findings": 0,
"question_quality_score": 100,
"review_json": "artifacts/domain_runs/gui_run_reviews/assistant-stage1-test/run_review.json",
"review_markdown": "artifacts/domain_runs/gui_run_reviews/assistant-stage1-test/run_review.md",
"repair_targets_json": "artifacts/domain_runs/gui_run_reviews/assistant-stage1-test/repair_targets.json",
"conversation_pairs_json": "artifacts/domain_runs/gui_run_reviews/assistant-stage1-test/conversation_pairs.json",
"detector_results_json": "artifacts/domain_runs/gui_run_reviews/assistant-stage1-test/detector_results.json",
"detector_results_status": "fail",
"detector_results_summary": {
"status": "fail",
"detector_count": 1,
"pass": 0,
"fail": 1,
"review": 0,
"skipped": 0,
},
},
}
review = {
"run_id": "assistant-stage1-test",
"repair_targets": [
{
"problem_layer": "answer_shape_mismatch",
"issue_code": "business_direct_answer_missing",
"severity": "P0",
"occurrences": 1,
}
],
"findings": [
{
"turn_index": 19,
"severity": "P0",
"issue_codes": ["business_direct_answer_missing"],
"question": "какой у нас самый доходный год",
"assistant_first_line": "Коротко: ограниченный бизнес-обзор...",
}
],
}
handoff = stage_loop.build_stage_repair_handoff(summary, review)
target = handoff["primary_repair_targets"][0]
self.assertIn("business_direct_answer_missing", handoff["issue_catalog_contracts"])
self.assertIn("direct_answer_surface_pack", handoff["rerun_matrix"])
self.assertIn("llm_normalizer/backend/src/services/assistantService.ts", target["allowed_patch_targets"])
self.assertIn("llm_normalizer/backend/src/services/address_runtime/composeStage.ts", target["candidate_files"])
self.assertTrue(handoff["auto_propose_gate"]["allowed"])
self.assertEqual(handoff["auto_propose_gate"]["detector_results_summary"]["status"], "fail")
def test_stage_repair_handoff_blocks_auto_propose_for_non_allowlisted_gui_issue(self) -> None:
summary = {
"stage_id": "agent_loop",
"next_action": "continue_repair_from_gui_review_p0",
"latest_gui_review": {
"run_id": "assistant-stage1-failing",
"overall_business_status": "fail",
"p0_findings": 1,
"p1_findings": 1,
"question_quality_score": 98,
"review_json": "artifacts/domain_runs/gui_run_reviews/assistant-stage1-failing/run_review.json",
"review_markdown": "artifacts/domain_runs/gui_run_reviews/assistant-stage1-failing/run_review.md",
"repair_targets_json": "artifacts/domain_runs/gui_run_reviews/assistant-stage1-failing/repair_targets.json",
"detector_results_json": "artifacts/domain_runs/gui_run_reviews/assistant-stage1-failing/detector_results.json",
"detector_results_status": "fail",
"detector_results_summary": {
"status": "fail",
"detector_count": 1,
"pass": 0,
"fail": 1,
"review": 0,
"skipped": 0,
},
},
}
review = {
"run_id": "assistant-stage1-failing",
"repair_targets": [
{
"problem_layer": "runtime_error",
"issue_code": "backend_error_response",
"severity": "P0",
"occurrences": 1,
}
],
"findings": [],
}
handoff = stage_loop.build_stage_repair_handoff(summary, review)
self.assertFalse(handoff["auto_propose_gate"]["allowed"])
self.assertIn(
"issue_code_not_allowlisted:backend_error_response",
handoff["auto_propose_gate"]["blocking_reasons"],
)
self.assertIn("failed_gui_turn", handoff["rerun_matrix"])
def test_repair_iteration_plan_adds_candidate_files(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
stage_dir = Path(tmp) / "stage"
@ -531,6 +635,12 @@ class StageAgentLoopTests(unittest.TestCase):
"question_quality_score": 100,
"review_markdown": "review.md",
"repair_targets_json": "repair_targets.json",
"detector_results_json": "detector_results.json",
"detector_results_status": "fail",
"detector_results_summary": {"status": "fail", "detector_count": 1, "fail": 1},
"issue_codes": ["business_direct_answer_missing"],
"rerun_matrix": ["failed_scenario", "direct_answer_surface_pack", "accepted_smoke_pack"],
"auto_propose_gate": {"allowed": True, "reason": "auto_coder_gate_passed"},
"primary_repair_targets": [
{
"problem_layer": "answer_shape_mismatch",
@ -564,6 +674,10 @@ class StageAgentLoopTests(unittest.TestCase):
self.assertEqual(plan["iteration_id"], "repair_001")
self.assertIn("llm_normalizer/backend/src/services/address_runtime/composeStage.ts", plan["candidate_files"])
self.assertEqual(plan["primary_repair_targets"][0]["candidate_files"][0], "llm_normalizer/backend/src/services/address_runtime/composeStage.ts")
self.assertEqual(plan["detector_results_status"], "fail")
self.assertIn("business_direct_answer_missing", plan["issue_codes"])
self.assertIn("direct_answer_surface_pack", plan["rerun_matrix"])
self.assertTrue(plan["auto_propose_gate"]["allowed"])
self.assertIn("ingest-gui-run", plan["acceptance_rerun"]["ingest_command_template"])
def test_handle_prepare_repair_materializes_prompt_and_checklist(self) -> None: