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

Встроить detector results в GUI review gate

This commit is contained in:
dctouch 2026-05-25 13:21:41 +03:00
parent f69b3e1994
commit 36f7b4df03
4 changed files with 197 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
scripts/review_assistant_stage1_run.py
View File

@ -10,6 +10,7 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import agent_detector_runner
import domain_case_loop as dcl
@ -742,6 +743,18 @@ def build_detector_candidates(
}
def build_review_detector_results(
output_dir: Path,
*,
detector_candidates_path: Path,
) -> dict[str, Any]:
return agent_detector_runner.build_detector_results(
output_dir,
detector_candidates_path=detector_candidates_path,
include_default_global=False,
)
def build_run_review(
*,
run_id: str,
@ -910,7 +923,12 @@ def save_run_review(review: dict[str, Any], output_dir: Path) -> None:
write_json(output_dir / "conversation_pairs.json", review.get("conversation_pairs") or [])
write_json(output_dir / "question_quality_review.json", review.get("question_quality_review") or {})
write_json(output_dir / "repair_targets.json", review.get("repair_targets") or [])
write_json(output_dir / "detector_candidates.json", build_detector_candidates(review))
detector_candidates_path = output_dir / "detector_candidates.json"
write_json(detector_candidates_path, build_detector_candidates(review))
write_json(
output_dir / "detector_results.json",
build_review_detector_results(output_dir, detector_candidates_path=detector_candidates_path),
)
def build_parser() -> argparse.ArgumentParser:
@ -941,6 +959,12 @@ def main(argv: list[str] | None = None) -> int:
save_run_review(review, output_dir)
if args.print_summary:
summary = review["summary"]
detector_results = load_json(output_dir / "detector_results.json")
detector_summary = (
detector_results.get("summary")
if isinstance(detector_results, dict) and isinstance(detector_results.get("summary"), dict)
else {}
)
print(
json.dumps(
{
@ -950,6 +974,8 @@ def main(argv: list[str] | None = None) -> int:
"turn_pairs_total": summary["turn_pairs_total"],
"business_issue_turns": summary["business_issue_turns"],
"question_quality_score": summary["question_quality_score"],
"detector_results_status": detector_summary.get("status"),
"detector_count": detector_summary.get("detector_count"),
},
ensure_ascii=False,
indent=2,

74
scripts/stage_agent_loop.py
View File

@ -939,6 +939,10 @@ def build_next_step_guidance(next_action: str) -> dict[str, Any]:
"inspect_gui_review_status": [
"inspect run_review.md and repair_targets.json manually",
],
"inspect_gui_review_detector_results": [
"inspect detector_results.json, detector_candidates.json, and run_review.json",
"rerun review_assistant_stage1_run.py for this GUI run if detector_results.json is missing",
],
"run_stage_loop_or_ingest_gui_run": [
"python scripts/stage_agent_loop.py run --manifest <stage_manifest.json> --dry-run",
"python scripts/stage_agent_loop.py ingest-gui-run --manifest <stage_manifest.json> --run-id assistant-stage1-<id>",
@ -1007,6 +1011,7 @@ def build_stage_handoff_markdown(summary: dict[str, Any]) -> str:
f"- p1_findings: `{latest_gui_review.get('p1_findings')}`",
f"- question_quality: `{latest_gui_review.get('question_quality_status')}` / `{latest_gui_review.get('question_quality_score')}`",
f"- repair_targets_count: `{latest_gui_review.get('repair_targets_count')}`",
f"- detector_results_status: `{latest_gui_review.get('detector_results_status') or 'n/a'}`",
f"- review_dir: `{latest_gui_review.get('review_dir')}`",
f"- review_markdown: `{latest_gui_review.get('review_markdown')}`",
]
@ -1110,11 +1115,13 @@ def build_repair_execution_stage_summary(
return base
def repair_validation_status(*, business_status: str, p0_count: int, p1_count: int) -> str:
def repair_validation_status(*, business_status: str, p0_count: int, p1_count: int, detector_status: str = "pass") -> str:
if p0_count > 0:
return "failed_p0"
if p1_count > 0 or business_status == "warning":
return "warning_p1"
if detector_status in {"fail", "review", "missing"}:
return "failed_detector_gate"
if business_status == "pass":
return "passed"
if business_status == "fail":
@ -1129,6 +1136,7 @@ def build_latest_repair_validation(
business_status: str,
p0_count: int,
p1_count: int,
detector_status: str,
next_action: str,
) -> dict[str, Any] | None:
previous_repair = (
@ -1144,6 +1152,7 @@ def build_latest_repair_validation(
business_status=business_status,
p0_count=p0_count,
p1_count=p1_count,
detector_status=detector_status,
)
return {
"schema_version": "stage_repair_validation_v1",
@ -1155,6 +1164,7 @@ def build_latest_repair_validation(
"remaining_p0_findings": p0_count,
"remaining_p1_findings": p1_count,
"overall_business_status": business_status,
"detector_results_status": detector_status,
"next_action": next_action,
"validated_at": now_iso(),
}
@ -1177,6 +1187,49 @@ def build_save_autorun_command(args: argparse.Namespace, stage_manifest: dict[st
]
def load_gui_review_detector_gate(review_dir: Path) -> dict[str, Any]:
detector_results_path = review_dir / "detector_results.json"
if not detector_results_path.exists():
return {
"status": "missing",
"detector_count": 0,
"summary": {
"status": "missing",
"detector_count": 0,
"pass": 0,
"fail": 0,
"skipped": 0,
"review": 0,
},
}
try:
detector_results = load_json_object(detector_results_path, "GUI detector_results.json")
except (OSError, json.JSONDecodeError, RuntimeError):
return {
"status": "missing",
"detector_count": 0,
"summary": {
"status": "missing",
"detector_count": 0,
"pass": 0,
"fail": 0,
"skipped": 0,
"review": 0,
},
}
detector_summary = (
detector_results.get("summary")
if isinstance(detector_results.get("summary"), dict)
else {}
)
status = str(detector_summary.get("status") or "missing").strip()
return {
"status": status,
"detector_count": int(detector_summary.get("detector_count") or 0),
"summary": detector_summary,
}
def build_gui_review_stage_summary(
*,
stage_manifest: dict[str, Any],
@ -1190,10 +1243,15 @@ def build_gui_review_stage_summary(
status = str(summary.get("overall_business_status") or "unknown").strip()
p0_count = int(summary.get("p0_findings") or 0)
p1_count = int(summary.get("p1_findings") or 0)
detector_gate = load_gui_review_detector_gate(review_dir)
detector_status = str(detector_gate.get("status") or "missing")
detector_gate_ok = detector_status in {"pass", "skipped"}
if p0_count > 0:
next_action = "continue_repair_from_gui_review_p0"
elif p1_count > 0 or status == "warning":
next_action = "continue_repair_from_gui_review_p1"
elif not detector_gate_ok:
next_action = "inspect_gui_review_detector_results"
elif status == "pass":
next_action = "manual_gui_confirmation_or_stage_close"
else:
@ -1210,8 +1268,8 @@ def build_gui_review_stage_summary(
"target_score": stage_manifest.get("target_score"),
"acceptance_invariants": stage_manifest.get("acceptance_invariants") or [],
"loop_final_status": base.get("loop_final_status") or "gui_review_ingested",
"accepted_gate": bool(status == "pass" and p0_count == 0 and p1_count == 0),
"manual_confirmation_required": bool(status == "pass"),
"accepted_gate": bool(status == "pass" and p0_count == 0 and p1_count == 0 and detector_gate_ok),
"manual_confirmation_required": bool(status == "pass" and detector_gate_ok),
"next_action": next_action,
"updated_at": now_iso(),
"latest_gui_review": {
@ -1222,6 +1280,10 @@ def build_gui_review_stage_summary(
"conversation_pairs_json": repo_relative(review_dir / "conversation_pairs.json"),
"question_quality_json": repo_relative(review_dir / "question_quality_review.json"),
"repair_targets_json": repo_relative(review_dir / "repair_targets.json"),
"detector_candidates_json": repo_relative(review_dir / "detector_candidates.json"),
"detector_results_json": repo_relative(review_dir / "detector_results.json"),
"detector_results_status": detector_status,
"detector_results_summary": detector_gate.get("summary"),
"overall_business_status": status,
"turn_pairs_total": summary.get("turn_pairs_total"),
"business_issue_turns": summary.get("business_issue_turns"),
@ -1240,6 +1302,7 @@ def build_gui_review_stage_summary(
business_status=status,
p0_count=p0_count,
p1_count=p1_count,
detector_status=detector_status,
next_action=next_action,
)
if latest_repair_validation is not None:
@ -1297,6 +1360,9 @@ def build_stage_repair_handoff(summary: dict[str, Any], review: dict[str, Any])
"question_quality_score": latest.get("question_quality_score"),
"review_markdown": latest.get("review_markdown"),
"repair_targets_json": latest.get("repair_targets_json"),
"detector_candidates_json": latest.get("detector_candidates_json"),
"detector_results_json": latest.get("detector_results_json"),
"detector_results_status": latest.get("detector_results_status"),
"primary_repair_targets": primary_targets,
"sample_findings": sample_findings,
}
@ -1315,6 +1381,8 @@ def build_stage_repair_handoff_markdown(handoff: dict[str, Any]) -> str:
f"- question_quality_score: `{handoff.get('question_quality_score')}`",
f"- review_markdown: `{handoff.get('review_markdown')}`",
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'}`",
"",
"## Primary Repair Targets",
]

44
scripts/test_review_assistant_stage1_run.py
View File

@ -240,13 +240,57 @@ class AssistantStage1RunReviewTests(unittest.TestCase):
self.assertTrue((output_dir / "run_review.json").exists())
self.assertTrue((output_dir / "run_review.md").exists())
self.assertTrue((output_dir / "detector_candidates.json").exists())
self.assertTrue((output_dir / "detector_results.json").exists())
detector_candidates = json.loads((output_dir / "detector_candidates.json").read_text(encoding="utf-8"))
self.assertEqual(detector_candidates["schema_version"], "agent_detector_candidates_v1")
self.assertEqual(detector_candidates["candidate_count"], 0)
detector_results = json.loads((output_dir / "detector_results.json").read_text(encoding="utf-8"))
self.assertEqual(detector_results["schema_version"], "agent_detector_results_v1")
self.assertEqual(detector_results["summary"]["status"], "skipped")
self.assertEqual(detector_results["summary"]["detector_count"], 0)
markdown = (output_dir / "run_review.md").read_text(encoding="utf-8")
self.assertIn("overall_business_status", markdown)
self.assertIn("Question Quality", markdown)
def test_save_run_review_materializes_failing_detector_results_for_findings(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
output_dir = Path(tmp) / "review"
review = {
"run_id": "assistant-stage1-backend-error",
"summary": {
"overall_business_status": "fail",
"turn_pairs_total": 1,
"business_issue_turns": 1,
"p0_findings": 1,
"p1_findings": 0,
"issue_counts": {"backend_error_response": 1},
"question_quality_status": "strong",
"question_quality_score": 90,
},
"question_quality_review": {"status": "strong", "score": 90},
"findings": [
{
"step_id": "turn_001",
"turn_index": 1,
"question": "приветик - че как там дела",
"assistant_first_line": "backend_error",
"severity": "P0",
"issue_severities": {"backend_error_response": "P0"},
"issue_codes": ["backend_error_response"],
}
],
"repair_targets": [{"issue_code": "backend_error_response"}],
"conversation_pairs": [],
}
reviewer.save_run_review(review, output_dir)
detector_results = json.loads((output_dir / "detector_results.json").read_text(encoding="utf-8"))
self.assertEqual(detector_results["summary"]["status"], "fail")
self.assertEqual(detector_results["results"][0]["detector"], "backend_error_response_signal")
self.assertEqual(detector_results["results"][0]["issue_codes"], ["backend_error_response"])
def test_detector_candidates_use_issue_catalog_detectors(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)

55
scripts/test_stage_agent_loop.py
View File

@ -380,6 +380,21 @@ class StageAgentLoopTests(unittest.TestCase):
with tempfile.TemporaryDirectory() as tmp:
stage_dir = Path(tmp) / "stage"
review_dir = stage_dir / "gui_run_reviews" / "assistant-stage1-rerun"
write_json(
review_dir / "detector_results.json",
{
"schema_version": "agent_detector_results_v1",
"summary": {
"status": "skipped",
"detector_count": 0,
"pass": 0,
"fail": 0,
"skipped": 0,
"review": 0,
},
"results": [],
},
)
review = {
"run_id": "assistant-stage1-rerun",
"summary": {
@ -419,11 +434,51 @@ class StageAgentLoopTests(unittest.TestCase):
)
self.assertEqual(summary["next_action"], "manual_gui_confirmation_or_stage_close")
self.assertTrue(summary["accepted_gate"])
self.assertEqual(summary["latest_gui_review"]["detector_results_status"], "skipped")
self.assertTrue(summary["latest_repair_validation"]["accepted_after_repair"])
self.assertEqual(summary["latest_repair_validation"]["validation_status"], "passed")
self.assertEqual(summary["latest_repair_validation"]["validation_run_id"], "assistant-stage1-rerun")
self.assertEqual(len(summary["repair_validation_history"]), 1)
def test_gui_review_stage_summary_blocks_pass_when_detector_results_missing(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
stage_dir = Path(tmp) / "stage"
review_dir = stage_dir / "gui_run_reviews" / "assistant-stage1-clean-no-detectors"
review = {
"run_id": "assistant-stage1-clean-no-detectors",
"summary": {
"overall_business_status": "pass",
"turn_pairs_total": 5,
"business_issue_turns": 0,
"p0_findings": 0,
"p1_findings": 0,
"question_quality_status": "strong",
"question_quality_score": 96,
},
"repair_targets": [],
}
summary = stage_loop.build_gui_review_stage_summary(
stage_manifest={
"stage_id": "agent_loop",
"module_name": "Agent Loop",
"title": "Agent Loop",
"target_score": 88,
"acceptance_invariants": [],
"global_plan_refs": [],
},
stage_dir=stage_dir,
review=review,
review_dir=review_dir,
)
self.assertEqual(summary["next_action"], "inspect_gui_review_detector_results")
self.assertFalse(summary["accepted_gate"])
self.assertFalse(summary["manual_confirmation_required"])
self.assertEqual(summary["latest_gui_review"]["detector_results_status"], "missing")
self.assertIn("detector_results.json", summary["next_step_guidance"]["command_templates"][0])
def test_stage_repair_handoff_keeps_primary_targets_and_samples(self) -> None:
summary = {
"stage_id": "agent_loop",