Add package deploy apply stage

tests/test_deploy_apply.py

@@ -8,6 +8,7 @@ from ampg.addresses import address_registry_path
 from ampg.apply import (
     CommandRunResult,
     apply_health,
+    apply_packages,
     apply_start,
     apply_state,
     apply_supervisor,
@@ -163,6 +164,94 @@ def _write_i2p_address_file(root: Path, value: str = "captureexample.b32.i2p") -
 
 
 class DeployApplyTest(unittest.TestCase):
+    def test_deploy_apply_packages_dry_run_plans_termux_installs(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            config_path = _init_i2p_config(root)
+            config = load_config(config_path)
+
+            results = apply_packages(
+                config,
+                dry_run=True,
+                platform_provider=_termux_provider(root),
+                daemon_probe=_missing_probe,
+            )
+
+        self.assertTrue(all(result.status == "planned" for result in results))
+        self.assertIn(("pkg", "install", "i2pd"), [result.command for result in results])
+        self.assertIn(("pkg", "install", "nginx"), [result.command for result in results])
+
+    def test_deploy_apply_packages_live_invokes_runner(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            config_path = _init_i2p_config(root)
+            config = load_config(config_path)
+            commands: list[tuple[str, ...]] = []
+
+            def runner(command: tuple[str, ...]) -> CommandRunResult:
+                commands.append(command)
+                return CommandRunResult(return_code=0)
+
+            results = apply_packages(
+                config,
+                dry_run=False,
+                platform_provider=_termux_provider(root),
+                daemon_probe=_missing_probe,
+                command_runner=runner,
+            )
+
+        self.assertTrue(all(result.status == "installed" for result in results))
+        self.assertIn(("pkg", "install", "i2pd"), commands)
+        self.assertIn(("pkg", "install", "nginx"), commands)
+
+    def test_deploy_apply_packages_blocks_manual_linux_user_install(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            config_path = _init_i2p_config(root)
+            config = load_config(config_path)
+
+            results = apply_packages(
+                config,
+                dry_run=True,
+                platform_provider=_linux_user_provider(root),
+                daemon_probe=_missing_probe,
+            )
+
+        self.assertTrue(any(result.status == "blocked" for result in results))
+        self.assertTrue(
+            any("automatic package install command" in result.message for result in results)
+        )
+
+    def test_deploy_apply_packages_cli_prints_summary(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            config_path = _init_i2p_config(root)
+            config_path.write_text(
+                config_path.read_text(encoding="utf-8").replace(
+                    'daemon_policy = "auto"',
+                    'daemon_policy = "manage"',
+                ),
+                encoding="utf-8",
+            )
+
+            status, output, error = _run_cli(
+                [
+                    "--config",
+                    str(config_path),
+                    "deploy",
+                    "apply",
+                    "--stage",
+                    "packages",
+                    "--dry-run",
+                    "--profile",
+                    "mobile-i2p",
+                ]
+            )
+
+        self.assertEqual(0, status, error)
+        self.assertIn("AMPG_DEPLOY_PACKAGE", output)
+        self.assertIn("AMPG_DEPLOY_APPLY_SUMMARY stage=packages mode=dry-run", output)
+
     def test_deploy_apply_state_blocks_unapproved_artifacts(self):
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp).resolve()

tests/test_deploy_plan.py

@@ -5,6 +5,10 @@ import unittest
 from pathlib import Path
 
 from ampg.cli import main
+from ampg.config import load_config
+from ampg.deploy import deploy_plan
+from ampg.platforms import platform_by_name
+from ampg.status import DaemonProbe
 
 
 def _source(root: Path) -> Path:
@@ -22,6 +26,10 @@ def _run_cli(args: list[str]) -> tuple[int, str, str]:
     return status, stdout.getvalue(), stderr.getvalue()
 
 
+def _missing_probe(adapter):
+    return DaemonProbe(installed=False, running=False, executable_path=None)
+
+
 class DeployPlanTest(unittest.TestCase):
     def test_deploy_plan_guides_fresh_i2p_only_site(self):
         with tempfile.TemporaryDirectory() as tmp:
@@ -66,6 +74,39 @@ class DeployPlanTest(unittest.TestCase):
         self.assertIn("build --profile mobile-i2p", output)
         self.assertIn("deploy apply --stage addresses --dry-run --profile mobile-i2p", output)
 
+    def test_deploy_plan_guides_package_apply_for_missing_managed_daemons(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            config_path = root / "gateway.toml"
+            init_status, _, init_error = _run_cli(
+                [
+                    "--config",
+                    str(config_path),
+                    "init",
+                    "site",
+                    "example",
+                    "--domain",
+                    "example.test",
+                    "--source",
+                    str(_source(root)),
+                    "--protocol",
+                    "i2p",
+                ]
+            )
+            self.assertEqual(0, init_status, init_error)
+
+            plan = deploy_plan(
+                load_config(config_path),
+                profile="mobile-i2p",
+                platform_provider=platform_by_name("android-termux"),
+                daemon_probe=_missing_probe,
+            )
+
+        packages = next(step for step in plan.steps if step.stage == "packages")
+        self.assertEqual("todo", packages.status)
+        self.assertIn("deploy apply --stage packages --dry-run", packages.command)
+        self.assertIn("managed daemon package", packages.message)
+
     def test_deploy_plan_shows_dns_review_for_clearnet_site(self):
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp).resolve()