diff --git a/scripts/stage_agent_loop.py b/scripts/stage_agent_loop.py index 1beff15..4951b0c 100644 --- a/scripts/stage_agent_loop.py +++ b/scripts/stage_agent_loop.py @@ -902,6 +902,7 @@ def build_next_step_guidance(next_action: str) -> dict[str, Any]: "rerun the same GUI/session or stage semantic pack", "python scripts/stage_agent_loop.py ingest-gui-run --manifest --run-id assistant-stage1-", "python scripts/stage_agent_loop.py ingest-replay-validation --manifest --run-dir artifacts/domain_runs/", + "python scripts/stage_agent_loop.py ingest-domain-loop-validation --manifest --run-dir artifacts/domain_runs/", ], "rerun_or_inspect_repair_targets": [ "inspect repair_execution_summary.json and repair_targets.json", @@ -1166,6 +1167,92 @@ def build_truth_replay_validation( } +def build_domain_loop_validation( + *, + run_dir: Path, + loop_state: dict[str, Any], +) -> dict[str, Any]: + iterations = loop_state.get("iterations") if isinstance(loop_state.get("iterations"), list) else [] + latest_iteration = iterations[-1] if iterations and isinstance(iterations[-1], dict) else {} + business_audit_path_raw = ( + latest_iteration.get("business_audit_json_path") + or loop_state.get("latest_business_audit_json_path") + or latest_iteration.get("business_audit_path") + or loop_state.get("latest_business_audit_path") + ) + business_audit_path = repo_path(str(business_audit_path_raw)) if business_audit_path_raw else None + business_audit = ( + load_json_object(business_audit_path, "Domain loop business audit") + if business_audit_path and business_audit_path.exists() and business_audit_path.suffix.lower() == ".json" + else {} + ) + quality_flags = business_audit.get("quality_flags") if isinstance(business_audit.get("quality_flags"), dict) else {} + repair_summary = ( + business_audit.get("repair_targets_summary") + if isinstance(business_audit.get("repair_targets_summary"), dict) + else {} + ) + severity_counts = ( + repair_summary.get("severity_counts") + if isinstance(repair_summary.get("severity_counts"), dict) + else latest_iteration.get("repair_target_severity_counts") + if isinstance(latest_iteration.get("repair_target_severity_counts"), dict) + else {} + ) + final_status = str(loop_state.get("final_status") or "").strip() + audit_status = str(business_audit.get("overall_status") or latest_iteration.get("loop_decision") or "").strip() + target_score = int(loop_state.get("target_score") or business_audit.get("target_score") or 0) + quality_score = int(latest_iteration.get("quality_score") or business_audit.get("quality_score") or 0) + p0_count = int(business_audit.get("unresolved_p0_count") or severity_counts.get("P0") or 0) + p1_count = int(severity_counts.get("P1") or 0) + p2_count = int(severity_counts.get("P2") or 0) + accepted = ( + final_status == "accepted" + and audit_status == "accepted" + and bool(latest_iteration.get("analyst_accepted_gate")) + and bool(latest_iteration.get("accepted_gate")) + and bool(latest_iteration.get("deterministic_gate_ok")) + and quality_score >= target_score + and p0_count == 0 + ) + pack_state_path = repo_path(str(latest_iteration.get("pack_dir"))) / "pack_state.json" if latest_iteration.get("pack_dir") else None + pack_state = ( + load_json_object(pack_state_path, "Domain loop pack_state.json") + if pack_state_path and pack_state_path.exists() + else {} + ) + return { + "schema_version": "stage_replay_validation_v1", + "validation_source": "domain_case_loop.run-pack-loop", + "validation_run_id": str(loop_state.get("loop_id") or run_dir.name), + "validated_run_dir": repo_relative(run_dir), + "source_spec": repo_relative(repo_path(str(loop_state.get("manifest_path")))) + if loop_state.get("manifest_path") + else repo_relative(run_dir / "manifest_source.txt"), + "validation_status": "passed" if accepted else "failed", + "accepted_after_repair": accepted, + "final_status": final_status, + "review_overall_status": audit_status, + "quality_score": quality_score, + "target_score": target_score, + "last_iteration_id": latest_iteration.get("iteration_id"), + "analyst_accepted_gate": bool(latest_iteration.get("analyst_accepted_gate")), + "deterministic_gate_ok": bool(latest_iteration.get("deterministic_gate_ok")), + "deterministic_gate_reason": latest_iteration.get("deterministic_gate_reason"), + "repair_target_count": int(repair_summary.get("target_count") or latest_iteration.get("repair_target_count") or 0), + "remaining_p0_findings": p0_count, + "remaining_p1_findings": p1_count, + "remaining_p2_findings": p2_count, + "critical_path_green": accepted, + "invariants": quality_flags, + "pack_execution_status": pack_state.get("execution_status"), + "pack_acceptance_status": pack_state.get("acceptance_status"), + "pack_final_status": pack_state.get("final_status"), + "business_audit": repo_relative(business_audit_path) if business_audit_path else None, + "validated_at": now_iso(), + } + + def build_latest_repair_validation( *, previous_summary: dict[str, Any] | None, @@ -2026,6 +2113,12 @@ def handle_ingest_replay_validation(args: argparse.Namespace) -> int: return 0 +def handle_ingest_domain_loop_validation(args: argparse.Namespace) -> int: + summary = ingest_domain_loop_validation(args) + print(json.dumps(summary, ensure_ascii=False, indent=2)) + return 0 + + def ingest_replay_validation(args: argparse.Namespace) -> dict[str, Any]: stage_manifest_path = repo_path(args.manifest) stage_manifest = load_stage_manifest(stage_manifest_path) @@ -2050,6 +2143,29 @@ def ingest_replay_validation(args: argparse.Namespace) -> dict[str, Any]: return summary +def ingest_domain_loop_validation(args: argparse.Namespace) -> dict[str, Any]: + stage_manifest_path = repo_path(args.manifest) + stage_manifest = load_stage_manifest(stage_manifest_path) + stage_dir = stage_dir_for(repo_path(args.output_root), stage_manifest["stage_id"]) + stage_dir.mkdir(parents=True, exist_ok=True) + write_json(stage_dir / "stage_manifest.json", stage_manifest) + write_text(stage_dir / "stage_manifest_source.txt", repo_relative(stage_manifest_path) + "\n") + + run_dir = repo_path(args.run_dir) + loop_state = load_json_object(run_dir / "loop_state.json", "Domain loop loop_state.json") + validation = build_domain_loop_validation(run_dir=run_dir, loop_state=loop_state) + summary_path = stage_dir / "stage_loop_summary.json" + previous_summary = load_json_object(summary_path, "Existing stage summary") if summary_path.exists() else None + summary = build_replay_validation_stage_summary( + stage_manifest=stage_manifest, + previous_summary=previous_summary, + validation=validation, + ) + save_stage_summary(stage_dir, summary) + save_stage_context_capsule(stage_manifest, stage_dir, summary=summary) + return summary + + def ingest_gui_run_review(args: argparse.Namespace) -> dict[str, Any]: stage_manifest_path = repo_path(args.manifest) stage_manifest = load_stage_manifest(stage_manifest_path) @@ -2577,6 +2693,14 @@ def build_parser() -> argparse.ArgumentParser: ingest_replay_parser.add_argument("--spec") ingest_replay_parser.set_defaults(func=handle_ingest_replay_validation) + ingest_domain_loop_parser = subparsers.add_parser( + "ingest-domain-loop-validation", + help="Attach an accepted domain_case_loop run-pack-loop run as stage repair validation evidence.", + ) + add_common_args(ingest_domain_loop_parser) + ingest_domain_loop_parser.add_argument("--run-dir", required=True) + ingest_domain_loop_parser.set_defaults(func=handle_ingest_domain_loop_validation) + repair_parser = subparsers.add_parser( "prepare-repair", help="Build coder-ready repair iteration artifacts from stage_repair_handoff.json.", diff --git a/scripts/test_stage_agent_loop.py b/scripts/test_stage_agent_loop.py index 66b7ee1..b1a7a92 100644 --- a/scripts/test_stage_agent_loop.py +++ b/scripts/test_stage_agent_loop.py @@ -515,6 +515,112 @@ class StageAgentLoopTests(unittest.TestCase): self.assertEqual(status["latest_validation_source"], "domain_truth_harness.run-live") self.assertEqual(status["latest_replay_validation_status"], "passed") + def test_domain_loop_validation_stage_summary_closes_after_accepted_pack_loop(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + manifest_path = root / "stage.json" + output_root = root / "stage_runs" + stage_dir = output_root / "agent_loop" + run_dir = root / "domain_runs" / "accepted_domain_loop" + iteration_dir = run_dir / "iterations" / "iteration_00" + pack_dir = iteration_dir / "pack_output" / "pack_run" + business_audit = iteration_dir / "business_audit.json" + write_json( + manifest_path, + { + "stage_id": "agent_loop", + "module_name": "Agent Loop", + "title": "Agent Loop", + "pack_manifest": "docs/orchestration/demo_pack.json", + "target_score": 90, + }, + ) + write_json( + stage_dir / "stage_loop_summary.json", + { + "stage_id": "agent_loop", + "next_action": "rerun_same_stage_or_gui_and_ingest_result", + }, + ) + write_json( + pack_dir / "pack_state.json", + { + "execution_status": "partial", + "acceptance_status": "accepted", + "final_status": "partial", + }, + ) + write_json( + business_audit, + { + "overall_status": "accepted", + "quality_score": 92, + "target_score": 90, + "quality_flags": { + "direct_answer_ok": True, + "business_usefulness_ok": True, + }, + "repair_targets_summary": { + "target_count": 0, + "severity_counts": {"P0": 0, "P1": 0, "P2": 0}, + }, + }, + ) + write_json( + run_dir / "loop_state.json", + { + "loop_id": "accepted_domain_loop", + "manifest_path": str(manifest_path), + "target_score": 90, + "final_status": "accepted", + "iterations": [ + { + "iteration_id": "iteration_00", + "pack_dir": str(pack_dir), + "quality_score": 92, + "loop_decision": "accepted", + "analyst_accepted_gate": True, + "accepted_gate": True, + "deterministic_gate_ok": True, + "deterministic_gate_reason": "deterministic_gate_passed", + "business_audit_json_path": str(business_audit), + "repair_target_count": 0, + "repair_target_severity_counts": {"P0": 0, "P1": 0, "P2": 0}, + } + ], + }, + ) + + exit_code = stage_loop.handle_ingest_domain_loop_validation( + stage_args( + manifest=str(manifest_path), + output_root=str(output_root), + run_dir=str(run_dir), + ) + ) + summary = json.loads((stage_dir / "stage_loop_summary.json").read_text(encoding="utf-8")) + status = stage_loop.build_stage_status( + { + "stage_id": "agent_loop", + "module_name": "Agent Loop", + "title": "Agent Loop", + }, + stage_dir, + ) + + self.assertEqual(exit_code, 0) + self.assertEqual(summary["next_action"], "manual_gui_confirmation_or_stage_close") + self.assertTrue(summary["accepted_gate"]) + validation = summary["latest_replay_validation"] + self.assertEqual(validation["validation_source"], "domain_case_loop.run-pack-loop") + self.assertEqual(validation["validation_run_id"], "accepted_domain_loop") + self.assertEqual(validation["validation_status"], "passed") + self.assertTrue(validation["accepted_after_repair"]) + self.assertEqual(validation["quality_score"], 92) + self.assertEqual(validation["pack_final_status"], "partial") + self.assertEqual(status["latest_validation_source"], "domain_case_loop.run-pack-loop") + self.assertEqual(status["latest_replay_validation_status"], "passed") + def test_gui_review_stage_summary_blocks_pass_when_detector_results_missing(self) -> None: with tempfile.TemporaryDirectory() as tmp: stage_dir = Path(tmp) / "stage"