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

Засчитывать accepted replay как валидацию stage agent loop

This commit is contained in:
dctouch 2026-06-01 21:21:11 +03:00
parent 599ca7d055
commit 8334e156a6
2 changed files with 206 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

132
scripts/stage_agent_loop.py
View File

@ -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 <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>",
],
"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.",

74
scripts/test_stage_agent_loop.py
View File

@ -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"