Add apply preflight gate

tests/test_activation.py

@@ -4,9 +4,10 @@ import tempfile
 import unittest
 from pathlib import Path
 
-from ampg.activation import activation_steps, blocked_steps
+from ampg.activation import activation_steps, apply_preflight, blocked_steps
 from ampg.cli import main
 from ampg.config import load_config
+from ampg.install_plan import install_state_copies, install_supervisor_actions
 from ampg.platforms import platform_by_name
 from ampg.status import DaemonProbe
 
@@ -139,6 +140,52 @@ daemon_policy = "auto"
         self.assertTrue(any(step.stage == "output" for step in blocked))
         self.assertTrue(any(step.stage == "health" for step in blocked))
 
+    def test_apply_preflight_reports_blocked_and_review_rows(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            config_path = _write_config(
+                Path(tmp),
+                """
+[site.protocols.i2p]
+enabled = true
+renderer = "privacy-html"
+daemon = "i2pd"
+daemon_policy = "auto"
+""",
+            )
+            config = load_config(config_path)
+            platform_provider = platform_by_name("android-termux")
+
+            preflight = apply_preflight(
+                activation=activation_steps(
+                    config,
+                    platform_provider=platform_provider,
+                    daemon_probe=_probe(installed=False, running=False),
+                ),
+                state_copies=install_state_copies(
+                    config,
+                    platform_provider=platform_provider,
+                    daemon_probe=_probe(installed=False, running=False),
+                ),
+                supervisor_actions=install_supervisor_actions(
+                    config,
+                    platform_provider=platform_provider,
+                    daemon_probe=_probe(installed=False, running=False),
+                ),
+            )
+
+        self.assertEqual("blocked", preflight.status)
+        self.assertGreaterEqual(preflight.blocked, 1)
+        self.assertGreaterEqual(preflight.review, 1)
+        self.assertTrue(
+            any(
+                item.phase == "activation:output" and item.status == "blocked"
+                for item in preflight.items
+            )
+        )
+        self.assertTrue(
+            any(item.phase == "state-copy" and item.status == "review" for item in preflight.items)
+        )
+
     def test_apply_cli_requires_dry_run(self):
         with tempfile.TemporaryDirectory() as tmp:
             config_path = _write_config(
@@ -251,6 +298,44 @@ daemon_policy = "auto"
         self.assertIn("service=ampg-example-i2p-i2pd", output_text)
         self.assertIn("service=ampg-example-i2p-nginx-loopback", output_text)
 
+    def test_apply_dry_run_prints_preflight_gate(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            config_path = _write_config(
+                root,
+                """
+[site.protocols.i2p]
+enabled = true
+renderer = "privacy-html"
+daemon = "i2pd"
+daemon_policy = "auto"
+""",
+            )
+            output = root / "out/i2p"
+            output.mkdir(parents=True)
+            (output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8")
+
+            stdout = io.StringIO()
+            with contextlib.redirect_stdout(stdout):
+                status = main(
+                    [
+                        "--config",
+                        str(config_path),
+                        "apply",
+                        "--dry-run",
+                        "--write-artifacts",
+                        "--platform",
+                        "android-termux",
+                    ]
+                )
+
+        output_text = stdout.getvalue()
+        self.assertEqual(0, status)
+        self.assertIn("AMPG_APPLY_PREFLIGHT status=review", output_text)
+        self.assertIn("AMPG_APPLY_PREFLIGHT_ITEM phase=activation:artifact", output_text)
+        self.assertIn("AMPG_APPLY_PREFLIGHT_ITEM phase=activation:address", output_text)
+        self.assertIn('message="live apply requires operator review before mutation"', output_text)
+
 
 if __name__ == "__main__":
     unittest.main()