From 8334e156a603b625e5d7851d3d126d16e9714ff9 Mon Sep 17 00:00:00 2001 From: dctouch Date: Mon, 1 Jun 2026 21:21:11 +0300 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D1=81=D1=87=D0=B8=D1=82=D1=8B?= =?UTF-8?q?=D0=B2=D0=B0=D1=82=D1=8C=20accepted=20replay=20=D0=BA=D0=B0?= =?UTF-8?q?=D0=BA=20=D0=B2=D0=B0=D0=BB=D0=B8=D0=B4=D0=B0=D1=86=D0=B8=D1=8E?= =?UTF-8?q?=20stage=20agent=20loop?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/stage_agent_loop.py | 132 +++++++++++++++++++++++++++++++ scripts/test_stage_agent_loop.py | 74 +++++++++++++++++ 2 files changed, 206 insertions(+) diff --git a/scripts/stage_agent_loop.py b/scripts/stage_agent_loop.py index 578d324..1beff15 100644 --- a/scripts/stage_agent_loop.py +++ b/scripts/stage_agent_loop.py @@ -901,6 +901,7 @@ def build_next_step_guidance(next_action: str) -> dict[str, Any]: "rerun_same_stage_or_gui_and_ingest_result": [ "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/", ], "rerun_or_inspect_repair_targets": [ "inspect repair_execution_summary.json and repair_targets.json", @@ -1038,6 +1039,11 @@ def build_stage_handoff_markdown(summary: dict[str, Any]) -> str: if isinstance(summary.get("latest_repair_validation"), dict) else {} ) + latest_replay_validation = ( + summary.get("latest_replay_validation") + if isinstance(summary.get("latest_replay_validation"), dict) + else {} + ) if latest_repair_validation: lines.extend( [ @@ -1129,6 +1135,37 @@ def repair_validation_status(*, business_status: str, p0_count: int, p1_count: i return "unknown" +def build_truth_replay_validation( + *, + run_dir: Path, + pack_state: dict[str, Any], + spec_path: Path | None = None, +) -> dict[str, Any]: + final_status = str(pack_state.get("final_status") or "").strip() + review_status = str(pack_state.get("review_overall_status") or "").strip() + accepted = bool(pack_state.get("acceptance_gate_passed")) and final_status == "accepted" and review_status == "pass" + return { + "schema_version": "stage_replay_validation_v1", + "validation_source": "domain_truth_harness.run-live", + "validation_run_id": str(pack_state.get("pack_id") or pack_state.get("scenario_id") or run_dir.name), + "validated_run_dir": repo_relative(run_dir), + "source_spec": repo_relative(spec_path) if spec_path else repo_relative(run_dir / "truth_harness_spec.json"), + "validation_status": "passed" if accepted else "failed", + "accepted_after_repair": accepted, + "final_status": final_status, + "review_overall_status": review_status, + "steps_total": pack_state.get("steps_total"), + "steps_passed": pack_state.get("steps_passed"), + "steps_failed": pack_state.get("steps_failed"), + "remaining_p0_findings": int(pack_state.get("unresolved_p0_count") or 0), + "remaining_p1_findings": int(pack_state.get("unresolved_p1_count") or 0), + "remaining_p2_findings": int(pack_state.get("unresolved_p2_count") or 0), + "critical_path_green": bool(pack_state.get("critical_path_green")), + "invariants": pack_state.get("invariants") if isinstance(pack_state.get("invariants"), dict) else {}, + "validated_at": now_iso(), + } + + def build_latest_repair_validation( *, previous_summary: dict[str, Any] | None, @@ -1491,6 +1528,54 @@ def build_gui_review_stage_summary( return base +def build_replay_validation_stage_summary( + *, + stage_manifest: dict[str, Any], + previous_summary: dict[str, Any] | None, + validation: dict[str, Any], +) -> dict[str, Any]: + accepted = bool(validation.get("accepted_after_repair")) and validation.get("validation_status") == "passed" + next_action = "manual_gui_confirmation_or_stage_close" if accepted else "rerun_same_stage_or_gui_and_ingest_result" + base = dict(previous_summary or {}) + base.update( + { + "schema_version": STAGE_SUMMARY_SCHEMA_VERSION, + "stage_id": stage_manifest["stage_id"], + "module_name": stage_manifest.get("module_name"), + "title": stage_manifest.get("title"), + "global_plan_refs": stage_manifest.get("global_plan_refs") or base.get("global_plan_refs") or [], + "target_score": stage_manifest.get("target_score", base.get("target_score")), + "acceptance_invariants": stage_manifest.get("acceptance_invariants") or base.get("acceptance_invariants") or [], + "accepted_gate": accepted, + "manual_confirmation_required": accepted, + "next_action": next_action, + "latest_replay_validation": validation, + "latest_repair_validation": validation, + "stage_closing_gate": { + "schema_version": "stage_closing_gate_v1", + "status": "pass" if accepted else "blocked_pending_repair_validation", + "passed": accepted, + "blockers": [] if accepted else ["latest_replay_validation_not_accepted"], + }, + "updated_at": now_iso(), + } + ) + validations = base.get("repair_validation_history") if isinstance(base.get("repair_validation_history"), list) else [] + validations = [ + item + for item in validations + if not ( + isinstance(item, dict) + and item.get("validation_run_id") == validation.get("validation_run_id") + and item.get("validation_source") == validation.get("validation_source") + ) + ] + validations.append(validation) + base["repair_validation_history"] = validations + base["next_step_guidance"] = build_next_step_guidance(next_action) + return base + + 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 {} findings = review.get("findings") if isinstance(review.get("findings"), list) else [] @@ -1935,6 +2020,36 @@ def handle_ingest_gui_run(args: argparse.Namespace) -> int: return 0 +def handle_ingest_replay_validation(args: argparse.Namespace) -> int: + summary = ingest_replay_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) + 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) + pack_state = load_json_object(run_dir / "pack_state.json", "Validated replay pack_state.json") + spec_path = repo_path(args.spec) if args.spec else None + validation = build_truth_replay_validation(run_dir=run_dir, pack_state=pack_state, spec_path=spec_path) + 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) @@ -2066,6 +2181,11 @@ def build_stage_status(stage_manifest: dict[str, Any], stage_dir: Path) -> dict[ if isinstance(summary.get("latest_repair_validation"), dict) else {} ) + latest_replay_validation = ( + summary.get("latest_replay_validation") + if isinstance(summary.get("latest_replay_validation"), dict) + else {} + ) stage_closing_gate = ( summary.get("stage_closing_gate") if isinstance(summary.get("stage_closing_gate"), dict) @@ -2094,6 +2214,9 @@ def build_stage_status(stage_manifest: dict[str, Any], stage_dir: Path) -> dict[ "latest_repair_dry_run": latest_repair_execution.get("dry_run"), "latest_validation_run_id": latest_repair_validation.get("validation_run_id"), "latest_validation_status": latest_repair_validation.get("validation_status"), + "latest_validation_source": latest_repair_validation.get("validation_source"), + "latest_replay_validation_run_id": latest_replay_validation.get("validation_run_id"), + "latest_replay_validation_status": latest_replay_validation.get("validation_status"), "accepted_after_repair": latest_repair_validation.get("accepted_after_repair"), "summary_path": repo_relative(summary_path) if summary_path.exists() else None, "domain_pack_loop_command_exists": domain_command_path.exists(), @@ -2445,6 +2568,15 @@ def build_parser() -> argparse.ArgumentParser: ingest_parser.add_argument("--review-output-dir") ingest_parser.set_defaults(func=handle_ingest_gui_run) + ingest_replay_parser = subparsers.add_parser( + "ingest-replay-validation", + help="Attach an accepted domain_truth_harness replay run as stage repair validation evidence.", + ) + add_common_args(ingest_replay_parser) + ingest_replay_parser.add_argument("--run-dir", required=True) + ingest_replay_parser.add_argument("--spec") + ingest_replay_parser.set_defaults(func=handle_ingest_replay_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 54733c2..66b7ee1 100644 --- a/scripts/test_stage_agent_loop.py +++ b/scripts/test_stage_agent_loop.py @@ -441,6 +441,80 @@ class StageAgentLoopTests(unittest.TestCase): self.assertEqual(summary["latest_repair_validation"]["validation_run_id"], "assistant-stage1-rerun") self.assertEqual(len(summary["repair_validation_history"]), 1) + def test_replay_validation_stage_summary_closes_after_accepted_truth_replay(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_replay" + write_json( + manifest_path, + { + "stage_id": "agent_loop", + "module_name": "Agent Loop", + "title": "Agent Loop", + "pack_manifest": "docs/orchestration/demo_pack.json", + }, + ) + write_json( + stage_dir / "stage_loop_summary.json", + { + "stage_id": "agent_loop", + "next_action": "continue_repair_from_gui_review_p0", + "latest_gui_review": { + "run_id": "assistant-stage1-before", + "overall_business_status": "fail", + "p0_findings": 1, + }, + }, + ) + write_json( + run_dir / "pack_state.json", + { + "schema_version": "truth_harness_pack_state_v1", + "pack_id": "accepted_replay", + "scenario_id": "accepted_replay", + "review_overall_status": "pass", + "final_status": "accepted", + "acceptance_gate_passed": True, + "steps_total": 3, + "steps_passed": 3, + "steps_failed": 0, + "unresolved_p0_count": 0, + "unresolved_p1_count": 0, + "critical_path_green": True, + "invariants": {"human_answer_quality_ok": True}, + }, + ) + + exit_code = stage_loop.handle_ingest_replay_validation( + stage_args( + manifest=str(manifest_path), + output_root=str(output_root), + run_dir=str(run_dir), + spec=None, + ) + ) + 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"]) + self.assertEqual(summary["latest_replay_validation"]["validation_source"], "domain_truth_harness.run-live") + self.assertEqual(summary["latest_repair_validation"]["validation_run_id"], "accepted_replay") + self.assertTrue(summary["latest_repair_validation"]["accepted_after_repair"]) + self.assertEqual(status["latest_validation_source"], "domain_truth_harness.run-live") + 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"