Add supervisor stage deploy apply

README.md

@@ -60,15 +60,17 @@ python3 -m ampg --config gateway.toml approvals list --profile vps-full
 python3 -m ampg --config gateway.toml approvals approve --profile vps-full --all
 python3 -m ampg --config gateway.toml apply --dry-run --profile vps-full
 python3 -m ampg --config gateway.toml deploy apply --stage state --dry-run --profile vps-full
+python3 -m ampg --config gateway.toml deploy apply --stage supervisor --dry-run --profile vps-full
 ```
 
 `init site` writes a readable `gateway.toml` with selected transports enabled and common
 transports left as one-line toggles. `deploy plan` condenses the lower-level checks into
 clear next steps. `dns plan` covers static DNS, Dynamic DNS, and behind-router
 reachability hints. `apply --dry-run` prints the activation sequence and a preflight
-verdict without changing services. `deploy apply --stage state` copies only approved
-managed-daemon artifacts into AMPG-owned state; later live stages will install and start
-services after the same review gate.
+verdict without changing services. `deploy apply --stage state` copies approved
+managed-daemon config into AMPG-owned state. `deploy apply --stage supervisor` installs
+approved supervisor files after state exists. Service start remains a separate future
+stage.
 
 ## Interaction tiers
 
@@ -134,6 +136,8 @@ python3 -m ampg --config examples/wownero.gateway.toml approvals list --profile
 python3 -m ampg --config examples/wownero.gateway.toml approvals approve --profile mobile-i2p --all
 python3 -m ampg --config examples/wownero.gateway.toml deploy apply --stage state --dry-run --profile mobile-i2p
 python3 -m ampg --config examples/wownero.gateway.toml deploy apply --stage state --profile mobile-i2p --yes
+python3 -m ampg --config examples/wownero.gateway.toml deploy apply --stage supervisor --dry-run --profile mobile-i2p
+python3 -m ampg --config examples/wownero.gateway.toml deploy apply --stage supervisor --profile mobile-i2p --yes
 python3 -m ampg --config examples/wownero.gateway.toml build
 python3 -m ampg --config examples/wownero.gateway.toml build --profile tor-i2p
 python3 -m ampg --config examples/wownero.gateway.toml build --protocol tor --protocol i2p
@@ -186,6 +190,11 @@ ready to copy into `gateway.state_dir`. The live form requires `--yes`; it creat
 AMPG-owned state directories and copies approved artifact contents. It does not install
 packages, change adopted daemons, start services, or remove keys.
 
+`deploy apply --stage supervisor --dry-run` shows approved supervisor files that can be
+installed after managed state config exists. The live form requires `--yes`; it writes
+only AMPG-named service files for the selected platform. It does not install packages,
+invoke `systemctl`, `launchctl`, or `sv-enable`, start daemons, or remove service files.
+
 `dns plan --free-domain-hints` prints optional community subdomain services that may help
 new users get a clearnet name without buying a domain. These are third-party registries;
 operators still need to review current terms, availability, content rules, and DNS
@@ -212,4 +221,5 @@ Explicit `--protocol` and `--platform` flags override the profile when present.
 - [ ] Run `ampg approvals approve --all` after reviewing generated artifacts.
 - [ ] Run `ampg apply --dry-run`; review the activation sequence and preflight gate.
 - [ ] Run `ampg deploy apply --stage state --dry-run`; confirm approved state copies.
+- [ ] Run `ampg deploy apply --stage supervisor --dry-run`; confirm approved service files.
 - [ ] Run live apply after the preflight gate is ready.

ampg/apply.py

@@ -5,8 +5,13 @@ from pathlib import Path
 import shutil
 
 from .config import GatewayConfig
-from .install_plan import InstallStateCopy, install_state_copies
-from .platforms import PlatformProvider
+from .install_plan import (
+    InstallStateCopy,
+    InstallSupervisorAction,
+    install_state_copies,
+    install_supervisor_actions,
+)
+from .platforms import PlatformProvider, detect_platform
 from .state_contract import gateway_state_root
 from .status import DaemonProbeFunc
 
@@ -24,6 +29,21 @@ class StateApplyResult:
     message: str
 
 
+@dataclass(frozen=True)
+class SupervisorApplyResult:
+    site_id: str
+    protocol: str
+    platform: str
+    kind: str
+    service: str
+    source: Path
+    target: Path
+    status: str
+    mode: str
+    command: str
+    message: str
+
+
 def apply_state(
     config: GatewayConfig,
     *,
@@ -65,6 +85,66 @@ def apply_state(
     return results
 
 
+def apply_supervisor(
+    config: GatewayConfig,
+    *,
+    dry_run: bool,
+    platform_provider: PlatformProvider | None = None,
+    daemon_probe: DaemonProbeFunc | None = None,
+) -> list[SupervisorApplyResult]:
+    provider = platform_provider or detect_platform()
+    actions = install_supervisor_actions(
+        config,
+        platform_provider=provider,
+        daemon_probe=daemon_probe,
+    )
+    state_copies = install_state_copies(
+        config,
+        platform_provider=provider,
+        daemon_probe=daemon_probe,
+    )
+    state_targets = {
+        (copy.site_id, copy.protocol, copy.platform, copy.kind): copy.target
+        for copy in state_copies
+    }
+    results: list[SupervisorApplyResult] = []
+    for action in actions:
+        target = supervisor_target(provider, action)
+        validation = _validate_supervisor_action(
+            action,
+            target=target,
+            state_targets=state_targets,
+            dry_run=dry_run,
+        )
+        if validation.status == "blocked":
+            results.append(validation)
+            continue
+        if not dry_run:
+            target.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy2(action.source, target)
+            _chmod_supervisor_target(provider, target)
+        results.append(
+            SupervisorApplyResult(
+                site_id=action.site_id,
+                protocol=action.protocol,
+                platform=action.platform,
+                kind=action.kind,
+                service=action.service,
+                source=action.source,
+                target=target,
+                status="planned" if dry_run else "written",
+                mode="dry-run" if dry_run else "live",
+                command=action.command,
+                message=(
+                    "would install approved supervisor file"
+                    if dry_run
+                    else "installed approved supervisor file"
+                ),
+            )
+        )
+    return results
+
+
 def _validate_state_copy(
     config: GatewayConfig,
     copy: InstallStateCopy,
@@ -109,6 +189,76 @@ def _validate_state_copy(
     )
 
 
+def _validate_supervisor_action(
+    action: InstallSupervisorAction,
+    *,
+    target: Path,
+    state_targets: dict[tuple[str, str, str, str], Path],
+    dry_run: bool,
+) -> SupervisorApplyResult:
+    mode = "dry-run" if dry_run else "live"
+    if action.status != "planned":
+        return _supervisor_result(
+            action,
+            target=target,
+            status="blocked",
+            mode=mode,
+            message=action.message,
+        )
+    if not _valid_service_name(action.service):
+        return _supervisor_result(
+            action,
+            target=target,
+            status="blocked",
+            mode=mode,
+            message=f"invalid AMPG supervisor service name: {action.service}",
+        )
+    required_state_kind = _required_state_kind(action.kind)
+    if required_state_kind is not None:
+        state_target = state_targets.get(
+            (action.site_id, action.protocol, action.platform, required_state_kind)
+        )
+        if state_target is None:
+            return _supervisor_result(
+                action,
+                target=target,
+                status="blocked",
+                mode=mode,
+                message="no managed state config is planned for this supervisor",
+            )
+        if not state_target.exists():
+            return _supervisor_result(
+                action,
+                target=target,
+                status="blocked",
+                mode=mode,
+                message="managed state config is missing; run deploy apply --stage state first",
+            )
+    if not action.source.exists():
+        return _supervisor_result(
+            action,
+            target=target,
+            status="blocked",
+            mode=mode,
+            message="approved supervisor artifact is missing",
+        )
+    if action.source.is_dir():
+        return _supervisor_result(
+            action,
+            target=target,
+            status="blocked",
+            mode=mode,
+            message="approved supervisor artifact is a directory",
+        )
+    return _supervisor_result(
+        action,
+        target=target,
+        status="ready",
+        mode=mode,
+        message="supervisor file passed safety checks",
+    )
+
+
 def _result(
     copy: InstallStateCopy,
     *,
@@ -129,6 +279,66 @@ def _result(
     )
 
 
+def _supervisor_result(
+    action: InstallSupervisorAction,
+    *,
+    target: Path,
+    status: str,
+    mode: str,
+    message: str,
+) -> SupervisorApplyResult:
+    return SupervisorApplyResult(
+        site_id=action.site_id,
+        protocol=action.protocol,
+        platform=action.platform,
+        kind=action.kind,
+        service=action.service,
+        source=action.source,
+        target=target,
+        status=status,
+        mode=mode,
+        command=action.command,
+        message=message,
+    )
+
+
+def supervisor_target(provider: PlatformProvider, action: InstallSupervisorAction) -> Path:
+    if provider.name == "linux-systemd":
+        return Path("/etc/systemd/system") / f"{action.service}.service"
+    if provider.name == "macos-launchd":
+        return Path("~/Library/LaunchAgents").expanduser() / f"org.ampg.{action.service}.plist"
+    if provider.name == "android-termux":
+        prefix = _termux_prefix(provider.state_root)
+        return prefix / "var/service" / action.service / "run"
+    return provider.state_root / "services" / action.service / action.source.name
+
+
+def _termux_prefix(state_root: Path) -> Path:
+    if len(state_root.parts) >= 3 and state_root.parts[-3:] == ("var", "lib", "ampg"):
+        return state_root.parents[2]
+    return state_root
+
+
+def _required_state_kind(supervisor_kind: str) -> str | None:
+    if supervisor_kind == "daemon-supervisor":
+        return "daemon-config"
+    if supervisor_kind == "loopback-supervisor":
+        return "loopback-config"
+    return None
+
+
+def _valid_service_name(service: str) -> bool:
+    if not service.startswith("ampg-"):
+        return False
+    return all(char.isalnum() or char in {"-", "_", "."} for char in service)
+
+
+def _chmod_supervisor_target(provider: PlatformProvider, target: Path) -> None:
+    if provider.name not in {"android-termux", "linux-user"}:
+        return
+    target.chmod(target.stat().st_mode | 0o111)
+
+
 def _is_relative_to(path: Path, parent: Path) -> bool:
     try:
         path.relative_to(parent)

ampg/cli.py

@@ -21,7 +21,12 @@ from .addresses import (
     effective_address_records,
     set_address,
 )
-from .apply import StateApplyResult, apply_state
+from .apply import (
+    StateApplyResult,
+    SupervisorApplyResult,
+    apply_state,
+    apply_supervisor,
+)
 from .approvals import (
     ACTIVATION_ARTIFACT_KIND,
     GENERIC_PLATFORM,
@@ -152,7 +157,7 @@ def main(argv: list[str] | None = None) -> int:
     _add_target_selection(deploy_apply_parser)
     deploy_apply_parser.add_argument(
         "--stage",
-        choices=("state",),
+        choices=("state", "supervisor"),
         required=True,
         help="Deployment stage to apply.",
     )
@@ -592,6 +597,27 @@ def _cmd_deploy(config, args) -> int:
                 f"blocked={len(blocked)}"
             )
             return 1 if blocked else 0
+        if args.stage == "supervisor":
+            results = apply_supervisor(
+                config,
+                dry_run=args.dry_run,
+                platform_provider=platform_provider,
+            )
+            for result in results:
+                _print_supervisor_apply_result(result)
+            blocked = [result for result in results if result.status == "blocked"]
+            written = [result for result in results if result.status == "written"]
+            planned = [result for result in results if result.status == "planned"]
+            print(
+                "AMPG_DEPLOY_APPLY_SUMMARY "
+                f"stage=supervisor "
+                f"mode={'dry-run' if args.dry_run else 'live'} "
+                f"results={len(results)} "
+                f"planned={len(planned)} "
+                f"written={len(written)} "
+                f"blocked={len(blocked)}"
+            )
+            return 1 if blocked else 0
     return 1
 
 
@@ -1080,6 +1106,23 @@ def _print_state_apply_result(result: StateApplyResult) -> None:
     )
 
 
+def _print_supervisor_apply_result(result: SupervisorApplyResult) -> None:
+    print(
+        "AMPG_DEPLOY_SUPERVISOR "
+        f"site={result.site_id} "
+        f"protocol={result.protocol} "
+        f"platform={result.platform} "
+        f"kind={result.kind} "
+        f"service={result.service} "
+        f"mode={result.mode} "
+        f"source=\"{_quote(str(result.source))}\" "
+        f"target=\"{_quote(str(result.target))}\" "
+        f"status={result.status} "
+        f"command=\"{_quote(result.command)}\" "
+        f"message=\"{_quote(result.message)}\""
+    )
+
+
 def _print_dns_record(record: DNSRecordPlan) -> None:
     print(
         "AMPG_DNS_RECORD "

ampg/deploy.py

@@ -5,6 +5,7 @@ from pathlib import Path
 import shlex
 
 from .activation import activation_steps, apply_preflight
+from .apply import supervisor_target
 from .addresses import effective_address_records
 from .approvals import (
     ACTIVATION_ARTIFACT_KIND,
@@ -22,7 +23,7 @@ from .install_plan import (
     install_supervisor_actions,
 )
 from .plan import plan_artifacts
-from .platforms import PlatformProvider
+from .platforms import PlatformProvider, detect_platform
 from .status import DaemonProbeFunc, doctor_gateway
 
 
@@ -88,6 +89,12 @@ def deploy_plan(
             platform_provider=platform_provider,
             daemon_probe=daemon_probe,
         ),
+        _deploy_apply_step(
+            config,
+            command_context,
+            platform_provider=platform_provider,
+            daemon_probe=daemon_probe,
+        ),
     )
     status = _overall_status(steps)
     return DeployPlan(
@@ -308,6 +315,71 @@ def _apply_preflight_step(
     )
 
 
+def _deploy_apply_step(
+    config: GatewayConfig,
+    commands: _CommandContext,
+    *,
+    platform_provider: PlatformProvider | None,
+    daemon_probe: DaemonProbeFunc | None,
+) -> DeployStep:
+    provider = platform_provider or detect_platform()
+    state_copies = install_state_copies(
+        config,
+        platform_provider=provider,
+        daemon_probe=daemon_probe,
+    )
+    supervisor_actions = install_supervisor_actions(
+        config,
+        platform_provider=provider,
+        daemon_probe=daemon_probe,
+    )
+    review = [
+        item
+        for item in (*state_copies, *supervisor_actions)
+        if item.status != "planned"
+    ]
+    if review:
+        return DeployStep(
+            "deploy-apply",
+            "review",
+            commands.command("apply --dry-run", platform=True),
+            f"review {len(review)} staged apply input(s)",
+        )
+    missing_state = [copy for copy in state_copies if not copy.target.exists()]
+    if missing_state:
+        return DeployStep(
+            "deploy-apply",
+            "todo",
+            commands.command("deploy apply --stage state --dry-run", platform=True),
+            f"copy {len(missing_state)} approved state artifact(s)",
+        )
+    missing_supervisors = [
+        action
+        for action in supervisor_actions
+        if not supervisor_target(provider, action).exists()
+    ]
+    if missing_supervisors:
+        return DeployStep(
+            "deploy-apply",
+            "todo",
+            commands.command("deploy apply --stage supervisor --dry-run", platform=True),
+            f"install {len(missing_supervisors)} approved supervisor file(s)",
+        )
+    if state_copies or supervisor_actions:
+        return DeployStep(
+            "deploy-apply",
+            "ready",
+            commands.command("deploy apply --stage supervisor --dry-run", platform=True),
+            "staged deploy files are applied",
+        )
+    return DeployStep(
+        "deploy-apply",
+        "ready",
+        "-",
+        "no managed deploy apply stages are needed",
+    )
+
+
 def _approval_candidates(
     config: GatewayConfig,
     platform_provider: PlatformProvider | None,

docs/architecture.md

@@ -133,9 +133,11 @@ services.
 
 `ampg deploy apply --stage state` is the narrow live stage for managed state. It reads
 the same approved artifact plan, refuses missing, stale, or unapproved inputs, and copies
-only approved generated daemon config into AMPG-owned state. Service installation,
-supervisor registration, daemon start, address capture, and health verification remain
-separate stages.
+only approved generated daemon config into AMPG-owned state.
+
+`ampg deploy apply --stage supervisor` installs approved AMPG-named supervisor files
+after the state files they reference exist. It does not start or reload services; daemon
+start, address capture, and health verification remain separate stages.
 
 The dry-run command is intentionally stricter than `doctor`: missing generated output is
 blocked during activation, because AMPG must not point a transport at an unbuilt output

docs/daemon-management.md

@@ -58,6 +58,8 @@ python3 -m ampg --config gateway.toml approvals list --profile mobile-i2p
 python3 -m ampg --config gateway.toml approvals approve --profile mobile-i2p --all
 python3 -m ampg --config gateway.toml deploy apply --stage state --dry-run --profile mobile-i2p
 python3 -m ampg --config gateway.toml deploy apply --stage state --profile mobile-i2p --yes
+python3 -m ampg --config gateway.toml deploy apply --stage supervisor --dry-run --profile mobile-i2p
+python3 -m ampg --config gateway.toml deploy apply --stage supervisor --profile mobile-i2p --yes
 python3 -m ampg --config gateway.toml doctor --platform android-termux
 ```
 
@@ -123,6 +125,12 @@ artifacts, creates AMPG-owned state directories, and copies approved managed-dae
 config into `gateway.state_dir`. It does not install packages, edit adopted daemon
 configs, start services, or delete identity material.
 
+`deploy apply --stage supervisor` is the second live apply stage. Dry-run mode prints
+`AMPG_DEPLOY_SUPERVISOR` rows. Live mode requires `--yes`, refuses unapproved or stale
+artifacts, requires the state stage to have copied managed config first, and writes
+AMPG-named service files for the selected platform. It does not install packages, start
+services, reload service managers, or delete service files.
+
 `--protocol` scopes operational commands to selected enabled protocols. A clearnet
 `adopt` failure will block a full activation run, but it will not block
 `apply --dry-run --protocol i2p` because clearnet is outside that selected run.

docs/state-contract.md

@@ -50,5 +50,10 @@ state or supervisor inputs still need review.
 
 `deploy apply --stage state` performs the approved copy portion only. It requires
 `--yes` for live writes, refuses unapproved or stale artifacts, writes only under
-`gateway.state_dir`, and leaves daemon start, address capture, and published health
-checks for later stages.
+`gateway.state_dir`, and leaves supervisor file installation, daemon start, address
+capture, and published health checks for later stages.
+
+`deploy apply --stage supervisor` performs the approved supervisor-file installation
+portion only. It requires copied managed state config first, writes AMPG-named service
+files for the selected platform, and leaves service-manager reload/start commands for a
+later explicit stage.

docs/wownero-integration.md

@@ -141,8 +141,8 @@ daemon_policy = "auto"
 - `ampg build` writes a fixture manifest that AMPB can check.
 - `ampg preview manifest` writes loopback fixture URLs that AMPB can check before live
   transport daemons are installed or adopted.
-- `ampg deploy apply --stage state --dry-run` shows approved managed-daemon config copies
-  before any Wownero transport service is started.
+- `ampg deploy apply --stage state --dry-run` and `--stage supervisor --dry-run` show
+  approved managed-daemon files before any Wownero transport service is started.
 - Tor output contains no `<script>` tags or inline event handlers.
 - Generated Gemini output has readable headings, paragraphs, and links.
 - Generated Micron output fits terminal-first browsing.
@@ -165,6 +165,7 @@ python3 -m ampg --config examples/wownero.gateway.toml install-plan --profile mo
 python3 -m ampg --config examples/wownero.gateway.toml approvals list --profile mobile-i2p
 python3 -m ampg --config examples/wownero.gateway.toml approvals approve --profile mobile-i2p --all
 python3 -m ampg --config examples/wownero.gateway.toml deploy apply --stage state --dry-run --profile mobile-i2p
+python3 -m ampg --config examples/wownero.gateway.toml deploy apply --stage supervisor --dry-run --profile mobile-i2p
 python3 -m ampg --config examples/i2p-only.gateway.toml plan
 python3 -m ampg --config examples/i2p-only.gateway.toml build
 ```
@@ -177,4 +178,4 @@ python3 -m ampg --config examples/i2p-only.gateway.toml build
 - [ ] Add Gemini conversion fixtures.
 - [ ] Add Reticulum/Micron conversion fixtures.
 - [ ] Enable I2P and Reticulum daemon management only after static outputs are stable.
-- [ ] Copy only reviewed Wownero daemon artifacts into managed state.
+- [ ] Copy only reviewed Wownero daemon and supervisor artifacts before service start.

tests/test_deploy_apply.py

@@ -4,7 +4,11 @@ import tempfile
 import unittest
 from pathlib import Path
 
+from ampg.apply import apply_state, apply_supervisor
 from ampg.cli import main
+from ampg.config import load_config
+from ampg.platforms import PlatformProvider
+from ampg.status import DaemonProbe
 
 
 def _source(root: Path) -> Path:
@@ -47,35 +51,67 @@ def _init_i2p_config(root: Path) -> Path:
     return config_path
 
 
-def _write_and_approve_artifacts(config_path: Path) -> None:
-    status, _, error = _run_cli(
-        [
-            "--config",
-            str(config_path),
-            "apply",
-            "--dry-run",
-            "--write-artifacts",
-            "--profile",
-            "mobile-i2p",
-        ]
-    )
+def _write_and_approve_artifacts(config_path: Path, *, platform: str | None = None) -> None:
+    apply_args = [
+        "--config",
+        str(config_path),
+        "apply",
+        "--dry-run",
+        "--write-artifacts",
+        "--profile",
+        "mobile-i2p",
+    ]
+    approve_args = [
+        "--config",
+        str(config_path),
+        "approvals",
+        "approve",
+        "--profile",
+        "mobile-i2p",
+        "--all",
+    ]
+    if platform is not None:
+        apply_args.extend(("--platform", platform))
+        approve_args.extend(("--platform", platform))
+    status, _, error = _run_cli(apply_args)
     if status != 0:
         raise AssertionError(error)
-    status, _, error = _run_cli(
-        [
-            "--config",
-            str(config_path),
-            "approvals",
-            "approve",
-            "--profile",
-            "mobile-i2p",
-            "--all",
-        ]
-    )
+    status, _, error = _run_cli(approve_args)
     if status != 0:
         raise AssertionError(error)
 
 
+def _missing_probe(adapter):
+    return DaemonProbe(
+        installed=False,
+        running=False,
+        executable_path=None,
+    )
+
+
+def _linux_user_provider(root: Path) -> PlatformProvider:
+    return PlatformProvider(
+        name="linux-user",
+        process_supervisor="user-process",
+        can_manage_daemons=True,
+        can_write_system_config=False,
+        state_root=root / "supervisor-root",
+    )
+
+
+def _run_state_apply(config_path: Path, root: Path) -> None:
+    config = load_config(config_path)
+    results = apply_state(
+        config,
+        dry_run=False,
+        platform_provider=_linux_user_provider(root),
+        daemon_probe=_missing_probe,
+    )
+    blocked = [result for result in results if result.status == "blocked"]
+    if blocked:
+        raise AssertionError(blocked[0].message)
+
+
 class DeployApplyTest(unittest.TestCase):
     def test_deploy_apply_state_blocks_unapproved_artifacts(self):
         with tempfile.TemporaryDirectory() as tmp:
@@ -212,6 +248,96 @@ class DeployApplyTest(unittest.TestCase):
         self.assertTrue(daemon_config_exists)
         self.assertTrue(loopback_config_exists)
 
+    def test_deploy_apply_supervisor_blocks_until_state_is_copied(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            config_path = _init_i2p_config(root)
+            _write_and_approve_artifacts(config_path, platform="linux-user")
+            config = load_config(config_path)
+
+            results = apply_supervisor(
+                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("deploy apply --stage state first" in result.message for result in results)
+        )
+
+    def test_deploy_apply_supervisor_dry_run_after_state_apply(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            config_path = _init_i2p_config(root)
+            _write_and_approve_artifacts(config_path)
+
+            status, _, error = _run_cli(
+                [
+                    "--config",
+                    str(config_path),
+                    "deploy",
+                    "apply",
+                    "--stage",
+                    "state",
+                    "--profile",
+                    "mobile-i2p",
+                    "--yes",
+                ]
+            )
+            if status != 0:
+                raise AssertionError(error)
+
+            status, output, error = _run_cli(
+                [
+                    "--config",
+                    str(config_path),
+                    "deploy",
+                    "apply",
+                    "--stage",
+                    "supervisor",
+                    "--dry-run",
+                    "--profile",
+                    "mobile-i2p",
+                ]
+            )
+
+        self.assertEqual(0, status, error)
+        self.assertIn("AMPG_DEPLOY_SUPERVISOR", output)
+        self.assertIn("status=planned", output)
+        self.assertIn("AMPG_DEPLOY_APPLY_SUMMARY stage=supervisor mode=dry-run", output)
+
+    def test_deploy_apply_supervisor_live_writes_user_service_files(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp).resolve()
+            config_path = _init_i2p_config(root)
+            _write_and_approve_artifacts(config_path, platform="linux-user")
+            _run_state_apply(config_path, root)
+            config = load_config(config_path)
+
+            results = apply_supervisor(
+                config,
+                dry_run=False,
+                platform_provider=_linux_user_provider(root),
+                daemon_probe=_missing_probe,
+            )
+            daemon_run = (
+                root / "supervisor-root/services/ampg-example-i2p-i2pd/i2pd.run"
+            )
+            loopback_run = (
+                root
+                / "supervisor-root/services/ampg-example-i2p-nginx-loopback/nginx-loopback.run"
+            )
+            daemon_run_exists = daemon_run.exists()
+            loopback_run_exists = loopback_run.exists()
+            daemon_run_executable = bool(daemon_run.stat().st_mode & 0o111)
+
+        self.assertTrue(all(result.status == "written" for result in results))
+        self.assertTrue(daemon_run_exists)
+        self.assertTrue(loopback_run_exists)
+        self.assertTrue(daemon_run_executable)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_deploy_plan.py

@@ -137,6 +137,95 @@ class DeployPlanTest(unittest.TestCase):
         self.assertIn("AMPG_DEPLOY_STEP stage=build status=ready", output)
         self.assertIn("AMPG_DEPLOY_STEP stage=artifacts status=todo", output)
 
+    def test_deploy_plan_guides_staged_apply_after_approval(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)
+            build_status, _, build_error = _run_cli(["--config", str(config_path), "build"])
+            self.assertEqual(0, build_status, build_error)
+            apply_status, _, apply_error = _run_cli(
+                [
+                    "--config",
+                    str(config_path),
+                    "apply",
+                    "--dry-run",
+                    "--write-artifacts",
+                    "--profile",
+                    "mobile-i2p",
+                ]
+            )
+            self.assertEqual(0, apply_status, apply_error)
+            approve_status, _, approve_error = _run_cli(
+                [
+                    "--config",
+                    str(config_path),
+                    "approvals",
+                    "approve",
+                    "--profile",
+                    "mobile-i2p",
+                    "--all",
+                ]
+            )
+            self.assertEqual(0, approve_status, approve_error)
+
+            status, output, error = _run_cli(
+                [
+                    "--config",
+                    str(config_path),
+                    "deploy",
+                    "plan",
+                    "--profile",
+                    "mobile-i2p",
+                ]
+            )
+            self.assertEqual(0, status, error)
+
+            state_status, _, state_error = _run_cli(
+                [
+                    "--config",
+                    str(config_path),
+                    "deploy",
+                    "apply",
+                    "--stage",
+                    "state",
+                    "--profile",
+                    "mobile-i2p",
+                    "--yes",
+                ]
+            )
+            self.assertEqual(0, state_status, state_error)
+            status_after_state, output_after_state, error_after_state = _run_cli(
+                [
+                    "--config",
+                    str(config_path),
+                    "deploy",
+                    "plan",
+                    "--profile",
+                    "mobile-i2p",
+                ]
+            )
+
+        self.assertIn("AMPG_DEPLOY_STEP stage=deploy-apply status=todo", output)
+        self.assertIn("deploy apply --stage state --dry-run", output)
+        self.assertEqual(0, status_after_state, error_after_state)
+        self.assertIn("deploy apply --stage supervisor --dry-run", output_after_state)
+
 
 if __name__ == "__main__":
     unittest.main()