Засчитывать domain-case-loop как validation evidence

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 <stage_manifest.json> --run-id assistant-stage1-<new_id>",
             "python scripts/stage_agent_loop.py ingest-replay-validation --manifest <stage_manifest.json> --run-dir artifacts/domain_runs/<accepted_replay>",
+            "python scripts/stage_agent_loop.py ingest-domain-loop-validation --manifest <stage_manifest.json> --run-dir artifacts/domain_runs/<accepted_domain_loop>",
         ],
         "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.",

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"