From bebea0361b5b3d4bd2f02854abb407e592451d16 Mon Sep 17 00:00:00 2001 From: dctouch Date: Mon, 15 Jun 2026 16:55:18 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=B8=D1=82?= =?UTF-8?q?=D1=8C=20=D0=B0=D0=B2=D1=82=D0=BE=D0=BF=D0=BB=D0=B0=D0=BD=20?= =?UTF-8?q?=D1=81=D0=BB=D0=B5=D0=B4=D1=83=D1=8E=D1=89=D0=B5=D0=B3=D0=BE=20?= =?UTF-8?q?=D0=B4=D0=BE=D0=BC=D0=B5=D0=BD=D0=BD=D0=BE=D0=B3=D0=BE=20=D1=81?= =?UTF-8?q?=D1=80=D0=B5=D0=B7=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/stage_agent_loop.py | 159 +++++++++++++++++++++++++++++++ scripts/test_stage_agent_loop.py | 23 +++++ 2 files changed, 182 insertions(+) diff --git a/scripts/stage_agent_loop.py b/scripts/stage_agent_loop.py index de495dd..fddd891 100644 --- a/scripts/stage_agent_loop.py +++ b/scripts/stage_agent_loop.py @@ -882,6 +882,7 @@ def build_stage_summary( if key in previous_summary: summary[key] = previous_summary[key] summary["next_step_guidance"] = build_next_step_guidance(next_action) + attach_stage_automation_followup(stage_manifest, summary) return summary @@ -956,6 +957,124 @@ def build_next_step_guidance(next_action: str) -> dict[str, Any]: } +AUTONOMY_READY_NEXT_ACTIONS = { + "manual_gui_confirmation", + "manual_gui_confirmation_or_stage_close", + "stage_closed_without_manual_confirmation", +} + + +def latest_stage_validation(summary: dict[str, Any]) -> dict[str, Any]: + latest_replay_validation = ( + summary.get("latest_replay_validation") + if isinstance(summary.get("latest_replay_validation"), dict) + else {} + ) + if latest_replay_validation: + return latest_replay_validation + latest_repair_validation = ( + summary.get("latest_repair_validation") + if isinstance(summary.get("latest_repair_validation"), dict) + else {} + ) + return latest_repair_validation + + +def build_stage_automation_followup( + stage_manifest: dict[str, Any], + summary: dict[str, Any], + *, + next_action: str | None = None, + effective_status: dict[str, Any] | None = None, +) -> dict[str, Any]: + resolved_next_action = str(next_action or summary.get("next_action") or "").strip() + closing_gate = summary.get("stage_closing_gate") if isinstance(summary.get("stage_closing_gate"), dict) else {} + closing_gate_passed = bool(closing_gate.get("passed")) or str(closing_gate.get("status") or "") == "pass" + accepted_gate = bool(summary.get("accepted_gate")) + effective_status_value = str((effective_status or {}).get("status") or "").strip() + accepted_by_validation = effective_status_value in {"accepted_after_validation", "validated_after_repair"} + ready = ( + accepted_gate + and closing_gate_passed + and resolved_next_action in AUTONOMY_READY_NEXT_ACTIONS + and (not effective_status_value or accepted_by_validation or effective_status_value in {"accepted", "pass"}) + ) + validation = latest_stage_validation(summary) + active_domain_contract = str( + stage_manifest.get("active_domain_contract") + or stage_manifest.get("next_domain_contract") + or "docs/orchestration/active_domain_contract.json" + ) + stage_manifest_path = "" + blockers: list[str] = [] + if not accepted_gate: + blockers.append("current_stage_not_accepted") + if not closing_gate_passed: + blockers.append("stage_closing_gate_not_passed") + if resolved_next_action not in AUTONOMY_READY_NEXT_ACTIONS: + blockers.append("current_next_action_not_ready_for_followup") + if effective_status_value and not ( + accepted_by_validation or effective_status_value in {"accepted", "pass"} + ): + blockers.append("effective_stage_status_not_accepted") + return { + "schema_version": "stage_automation_followup_v1", + "ready_for_next_domain_slice": ready, + "automation_next_action": "select_next_domain_slice" if ready else "finish_current_stage", + "blocking_next_action": None if ready else resolved_next_action, + "blockers": blockers, + "current_stage": { + "stage_id": stage_manifest.get("stage_id") or summary.get("stage_id"), + "manual_confirmation_still_required": bool(summary.get("manual_confirmation_required")), + "next_action": resolved_next_action, + "effective_stage_status": effective_status_value or None, + "validation_source": validation.get("validation_source"), + "validation_run_id": validation.get("validation_run_id"), + }, + "next_domain_contract": { + "source_of_truth": active_domain_contract, + "selection_policy": [ + "pick_one_open_ops_card_or_issue_catalog_domain", + "keep_agent_canon_stable_and_swap_only_active_domain_contract", + "validate_root_critical_children_critical_edges_and_primary_user_path", + "treat_semantic_answer_review_as_primary_acceptance_surface", + ], + }, + "recommended_sequence": [ + "record_or_complete_manual_gui_confirmation_if_required", + "select_next_domain_slice_from_ops_cards_or_issue_catalog", + "update_active_domain_contract_for_that_slice", + "run_domain_case_loop_live_pack", + "read_final_status_truth_review_business_audit_and_step_outputs_first", + "patch_runtime_or_agent_loop_when_the_user_facing_answer_is_wrong", + "rerun_until_accepted_then_ingest_validation", + ], + "command_templates": [ + f"python scripts/stage_agent_loop.py status --manifest {stage_manifest_path}", + f"edit {active_domain_contract} for the next domain slice", + f"python scripts/domain_case_loop.py run-pack-loop --manifest {active_domain_contract} --loop-id _p01_loop --repair-mode lead-handoff", + f"python scripts/stage_agent_loop.py ingest-domain-loop-validation --manifest {stage_manifest_path} --run-dir artifacts/domain_runs/", + ], + "autorun_persistence": { + "save_autorun_on_accept": bool( + summary.get("save_autorun_on_accept", stage_manifest.get("save_autorun_on_accept", True)) + ), + "only_after_live_replay_review": True, + "command_template": "python scripts/save_agent_semantic_run.py --spec docs/orchestration/.json", + }, + "guardrails": [ + "do_not_treat_saved_questions_as_a_completed_agent_run", + "do_not_accept_root_only_green_when_followup_edges_fail", + "do_not_patch_runtime_foundations_from_domain_loop_without_lead_review", + ], + } + + +def attach_stage_automation_followup(stage_manifest: dict[str, Any], summary: dict[str, Any]) -> dict[str, Any]: + summary["automation_followup"] = build_stage_automation_followup(stage_manifest, summary) + return summary + + def build_stage_handoff_markdown(summary: dict[str, Any]) -> str: lines = [ "# Stage agent loop handoff", @@ -1001,6 +1120,36 @@ def build_stage_handoff_markdown(summary: dict[str, Any]) -> str: else [] ) lines.extend([f"- `{item}`" for item in commands] if commands else ["- inspect stage_loop_handoff.md"]) + automation_followup = ( + summary.get("automation_followup") + if isinstance(summary.get("automation_followup"), dict) + else {} + ) + if automation_followup: + lines.extend( + [ + "", + "## Automation Follow-up", + f"- ready_for_next_domain_slice: `{automation_followup.get('ready_for_next_domain_slice')}`", + f"- automation_next_action: `{automation_followup.get('automation_next_action')}`", + f"- blocking_next_action: `{automation_followup.get('blocking_next_action') or 'n/a'}`", + ] + ) + next_domain_contract = ( + automation_followup.get("next_domain_contract") + if isinstance(automation_followup.get("next_domain_contract"), dict) + else {} + ) + if next_domain_contract.get("source_of_truth"): + lines.append(f"- active_domain_contract: `{next_domain_contract.get('source_of_truth')}`") + commands = ( + automation_followup.get("command_templates") + if isinstance(automation_followup.get("command_templates"), list) + else [] + ) + if commands: + lines.extend(["", "### Automation Commands"]) + lines.extend([f"- `{item}`" for item in commands]) latest_gui_review = summary.get("latest_gui_review") if isinstance(summary.get("latest_gui_review"), dict) else {} if latest_gui_review: lines.extend( @@ -1119,6 +1268,7 @@ def build_repair_execution_stage_summary( } ) base["next_step_guidance"] = build_next_step_guidance(next_action) + attach_stage_automation_followup(stage_manifest, base) return base @@ -1658,6 +1808,7 @@ def build_gui_review_stage_summary( ] history.append(base["latest_gui_review"]) base["gui_review_history"] = history + attach_stage_automation_followup(stage_manifest, base) return base @@ -1706,6 +1857,7 @@ def build_replay_validation_stage_summary( validations.append(validation) base["repair_validation_history"] = validations base["next_step_guidance"] = build_next_step_guidance(next_action) + attach_stage_automation_followup(stage_manifest, base) return base @@ -2441,6 +2593,12 @@ def build_stage_status(stage_manifest: dict[str, Any], stage_dir: Path) -> dict[ latest_repair_validation, latest_replay_validation, ) + automation_followup = build_stage_automation_followup( + stage_manifest, + summary, + next_action=next_action, + effective_status=effective_status, + ) return { "schema_version": "stage_agent_loop_status_v1", "stage_id": stage_manifest["stage_id"], @@ -2455,6 +2613,7 @@ def build_stage_status(stage_manifest: dict[str, Any], stage_dir: Path) -> dict[ "stage_closing_gate": stage_closing_gate or None, "next_action": next_action, "next_step_guidance": next_step_guidance, + "automation_followup": automation_followup, "effective_stage_status": effective_status.get("status"), "effective_stage_status_source": effective_status.get("source"), "latest_gui_review_superseded_by_validation": effective_status.get("supersedes_latest_gui_review"), diff --git a/scripts/test_stage_agent_loop.py b/scripts/test_stage_agent_loop.py index 9cae6bf..9e70a47 100644 --- a/scripts/test_stage_agent_loop.py +++ b/scripts/test_stage_agent_loop.py @@ -519,6 +519,17 @@ class StageAgentLoopTests(unittest.TestCase): self.assertTrue(status["latest_gui_review_superseded_by_validation"]) self.assertEqual(status["superseding_validation_run_id"], "accepted_replay") self.assertEqual(status["superseded_gui_run_id"], "assistant-stage1-before") + self.assertTrue(summary["automation_followup"]["ready_for_next_domain_slice"]) + self.assertEqual(summary["automation_followup"]["automation_next_action"], "select_next_domain_slice") + self.assertTrue(status["automation_followup"]["ready_for_next_domain_slice"]) + self.assertEqual( + status["automation_followup"]["current_stage"]["validation_run_id"], + "accepted_replay", + ) + self.assertIn( + "run_domain_case_loop_live_pack", + status["automation_followup"]["recommended_sequence"], + ) def test_domain_loop_validation_stage_summary_closes_after_accepted_pack_loop(self) -> None: with tempfile.TemporaryDirectory() as tmp: @@ -625,6 +636,15 @@ class StageAgentLoopTests(unittest.TestCase): 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") + self.assertTrue(status["automation_followup"]["ready_for_next_domain_slice"]) + self.assertEqual( + status["automation_followup"]["current_stage"]["validation_source"], + "domain_case_loop.run-pack-loop", + ) + self.assertEqual( + status["automation_followup"]["next_domain_contract"]["source_of_truth"], + "docs/orchestration/active_domain_contract.json", + ) def test_gui_review_stage_summary_blocks_pass_when_detector_results_missing(self) -> None: with tempfile.TemporaryDirectory() as tmp: @@ -1123,6 +1143,9 @@ class StageAgentLoopTests(unittest.TestCase): self.assertEqual(status["effective_stage_status"], "fail") self.assertEqual(status["effective_stage_status_source"], "latest_gui_review") self.assertFalse(status["latest_gui_review_superseded_by_validation"]) + self.assertFalse(status["automation_followup"]["ready_for_next_domain_slice"]) + self.assertEqual(status["automation_followup"]["automation_next_action"], "finish_current_stage") + self.assertIn("current_stage_not_accepted", status["automation_followup"]["blockers"]) def test_build_stage_status_derives_guidance_from_existing_next_action(self) -> None: with tempfile.TemporaryDirectory() as tmp: