Добавить автоплан следующего доменного среза

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 = "<stage_manifest.json>"
+    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 <next_domain>_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/<accepted_domain_loop>",
+        ],
+        "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/<accepted_spec>.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"),

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: