From d0a206f7d68684e453c7a37d79fe752d6d0edfd7 Mon Sep 17 00:00:00 2001
From: Paul Payne <paul@payne.io>
Date: Thu, 2 Jul 2026 11:35:34 -0700
Subject: [PATCH] =?UTF-8?q?core+cli:=20castle=20apply=20=E2=80=94=20one=20?=
 =?UTF-8?q?converge=20verb=20replaces=20deploy/start/enable/=E2=80=A6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Reconcile the running system to declared config in a single operation,
replacing the old deploy && start plus the per-kind enable/disable/
install/uninstall verbs. The mechanism varies by manager (systemd unit /
PATH install / gateway route); the verb never does.

Core (castle_core.deploy.apply):
- render (units, Caddyfile, tunnel) then reconcile: activate every
  enabled deployment that's down, restart any whose unit bytes changed,
  deactivate the disabled ones. Returns ApplyResult (the enacted diff).
- --plan computes the diff and touches nothing.
- restart-on-change is exact: _render_unit_files is the single source of
  truth for unit bytes, used both to write units and to predict restarts,
  so the plan can't drift from what deploy writes.

Desired state:
- DeploymentBase.enabled (default True) — the declarative on/off. apply
  converges to it; a disabled deployment is defined but not running and
  gets no gateway route (else it would 502). Registry omits enabled when
  True, keeping existing registries byte-identical.

CLI:
- add `castle apply [name] [--plan]`; keep `restart` as the one
  imperative bounce, plus status/doctor/logs/gateway/program.
- retire deploy, start, stop (top-level); service/job deploy|enable|
  disable|start|stop; tool install|uninstall; program install|uninstall.
- next-step strings + doctor hints now say `castle apply`.

Verified: core 129 + cli 31 green; live `castle apply` on a converged
node reports "nothing to do" with no spurious restarts; --plan shows an
empty diff; gateway + api stay up.
---
 cli/src/castle_cli/commands/apply.py         |  60 +++++
 cli/src/castle_cli/commands/create.py        |   4 +-
 cli/src/castle_cli/commands/delete.py        |   4 +-
 cli/src/castle_cli/commands/deploy.py        |  21 --
 cli/src/castle_cli/commands/deploy_create.py |   4 +-
 cli/src/castle_cli/commands/doctor.py        |  12 +-
 cli/src/castle_cli/commands/gateway.py       |   6 +-
 cli/src/castle_cli/commands/info.py          |   2 +-
 cli/src/castle_cli/commands/run_cmd.py       |   4 +-
 cli/src/castle_cli/commands/service.py       | 127 ++---------
 cli/src/castle_cli/commands/tool.py          |   2 +-
 cli/src/castle_cli/main.py                   |  86 +++-----
 cli/tests/test_doctor.py                     |   2 +-
 core/src/castle_core/deploy.py               | 219 ++++++++++++++++---
 core/src/castle_core/generators/caddyfile.py |   6 +
 core/src/castle_core/manifest.py             |   5 +
 core/src/castle_core/registry.py             |   8 +
 core/tests/test_apply.py                     |  74 +++++++
 18 files changed, 406 insertions(+), 240 deletions(-)
 create mode 100644 cli/src/castle_cli/commands/apply.py
 delete mode 100644 cli/src/castle_cli/commands/deploy.py
 create mode 100644 core/tests/test_apply.py

diff --git a/cli/src/castle_cli/commands/apply.py b/cli/src/castle_cli/commands/apply.py
new file mode 100644
index 0000000..db1b939
--- /dev/null
+++ b/cli/src/castle_cli/commands/apply.py
@@ -0,0 +1,60 @@
+"""castle apply — converge the running system to match config.
+
+The one workhorse verb: renders systemd units + the Caddyfile + tunnel config,
+then reconciles the runtime (activate what's enabled and down, restart what
+changed, deactivate what's disabled). Replaces the old `deploy && start` plus the
+per-kind enable/disable/install/uninstall verbs.
+
+`--plan` computes and prints the diff without writing or touching the runtime.
+"""
+
+from __future__ import annotations
+
+import argparse
+
+_C = {
+    "activate": "\033[32m",   # green
+    "restart": "\033[33m",    # yellow
+    "deactivate": "\033[31m", # red
+    "reset": "\033[0m",
+    "dim": "\033[90m",
+}
+
+
+def _line(verb_color: str, verb: str, names: list[str]) -> None:
+    if not names:
+        return
+    print(f"  {verb_color}{verb}{_C['reset']}  {', '.join(sorted(names))}")
+
+
+def run_apply(args: argparse.Namespace) -> int:
+    from castle_core.deploy import apply
+
+    target = getattr(args, "name", None)
+    plan = getattr(args, "plan", False)
+
+    result = apply(target_name=target, plan=plan)
+
+    # Surface any warnings the render produced (acme prerequisites, tunnel notes).
+    for msg in result.messages:
+        if msg.startswith("Warning"):
+            print(f"  {_C['dim']}{msg}{_C['reset']}")
+
+    if plan:
+        print("\n\033[1mPlan\033[0m " + _C["dim"] + "(no changes made)" + _C["reset"])
+        if not result.changed:
+            print(f"  {_C['dim']}nothing to do — already converged{_C['reset']}")
+            return 0
+        _line(_C["activate"], "would activate  ", result.activated)
+        _line(_C["restart"], "would restart   ", result.restarted)
+        _line(_C["deactivate"], "would deactivate", result.deactivated)
+        return 0
+
+    print("\n\033[1mApplied\033[0m")
+    if not result.changed:
+        print(f"  {_C['dim']}nothing to do — already converged{_C['reset']}")
+        return 0
+    _line(_C["activate"], "activated  ", result.activated)
+    _line(_C["restart"], "restarted  ", result.restarted)
+    _line(_C["deactivate"], "deactivated", result.deactivated)
+    return 0
diff --git a/cli/src/castle_cli/commands/create.py b/cli/src/castle_cli/commands/create.py
index 83e3b69..ff3af6a 100644
--- a/cli/src/castle_cli/commands/create.py
+++ b/cli/src/castle_cli/commands/create.py
@@ -158,12 +158,12 @@ def run_create(args: argparse.Namespace) -> int:
     if stack == "supabase":
         print("  # edit migrations/, functions/, public/ — targets the shared substrate")
         print(f"  castle program build {name}   # apply migrations to the substrate")
-        print(f"  castle deploy && castle gateway reload   # serve at /{name}/")
+        print(f"  castle apply   # serve at {name}.<gateway.domain>")
     elif stack:
         print("  uv sync")
         if kind == "service":
             print(f"  uv run {name}  # starts on port {port}")
-            print(f"  castle deploy {name}")
+            print(f"  castle apply {name}")
         print(f"  castle test {name}")
     else:
         print("  # add code, then declare commands: in castle.yaml")
diff --git a/cli/src/castle_cli/commands/delete.py b/cli/src/castle_cli/commands/delete.py
index 75fd244..9ec1041 100644
--- a/cli/src/castle_cli/commands/delete.py
+++ b/cli/src/castle_cli/commands/delete.py
@@ -116,9 +116,9 @@ def run_delete(args: argparse.Namespace) -> int:
 
         try:
             deploy()
-            print("Reconciled runtime (castle deploy).")
+            print("Reconciled runtime (castle apply).")
         except Exception as e:
-            print(f"warning: reconcile (castle deploy) failed — run 'castle deploy': {e}")
+            print(f"warning: reconcile failed — run 'castle apply': {e}")
 
     # Optional: delete the source directory.
     if args.source and source_dir:
diff --git a/cli/src/castle_cli/commands/deploy.py b/cli/src/castle_cli/commands/deploy.py
deleted file mode 100644
index 2b905ff..0000000
--- a/cli/src/castle_cli/commands/deploy.py
+++ /dev/null
@@ -1,21 +0,0 @@
-"""castle deploy — thin CLI wrapper around castle_core.deploy."""
-
-from __future__ import annotations
-
-import argparse
-
-from castle_core.deploy import deploy
-
-
-def run_deploy(args: argparse.Namespace) -> int:
-    """Deploy from castle.yaml to ~/.castle/."""
-    target_name = getattr(args, "name", None)
-    result = deploy(target_name=target_name)
-
-    for msg in result.messages:
-        print(f"  {msg}")
-
-    print(f"\nDeployed {result.deployed_count} item(s).")
-    if result.deployed_count > 0:
-        print("Run 'castle start' to start all services.")
-    return 0
diff --git a/cli/src/castle_cli/commands/deploy_create.py b/cli/src/castle_cli/commands/deploy_create.py
index 0d17f74..3da1369 100644
--- a/cli/src/castle_cli/commands/deploy_create.py
+++ b/cli/src/castle_cli/commands/deploy_create.py
@@ -88,7 +88,7 @@ def run_service_create(args: argparse.Namespace) -> int:
         print(f"  port:   {args.port}")
     if proxy:
         print(f"  subdomain: {name}.<gateway.domain>")
-    print(f"\nNext: castle service deploy {name} && castle service start {name}")
+    print(f"\nNext: castle apply {name}")
     return 0
 
 
@@ -118,5 +118,5 @@ def run_job_create(args: argparse.Namespace) -> int:
     print(f"Created job '{name}'.")
     print(f"  runs:     {args.launcher} ({args.run or args.program or name})")
     print(f"  schedule: {args.schedule}")
-    print(f"\nNext: castle job deploy {name} && castle job enable {name}")
+    print(f"\nNext: castle apply {name}")
     return 0
diff --git a/cli/src/castle_cli/commands/doctor.py b/cli/src/castle_cli/commands/doctor.py
index 7beb49e..cb9f097 100644
--- a/cli/src/castle_cli/commands/doctor.py
+++ b/cli/src/castle_cli/commands/doctor.py
@@ -6,7 +6,7 @@ of checks grouped into Environment, Configuration, Runtime, and TLS & exposure;
 each check reports ok / warn / fail with a one-line hint when action is needed.
 
 Exit code: 0 when nothing FAILed (warnings are allowed), 1 otherwise — so it
-doubles as a scriptable smoke test after `./install.sh` or `castle deploy`.
+doubles as a scriptable smoke test after `./install.sh` or `castle apply`.
 """
 
 from __future__ import annotations
@@ -206,7 +206,7 @@ def _check_runtime(config) -> list[Check]:
             Check(
                 FAIL,
                 "gateway not running",
-                hint="castle deploy && castle start",
+                hint="castle apply",
             )
         )
 
@@ -227,7 +227,7 @@ def _check_runtime(config) -> list[Check]:
             checks.append(Check(OK, "castle-api running", detail=detail))
     else:
         checks.append(
-            Check(FAIL, "castle-api not running", hint="castle deploy && castle start")
+            Check(FAIL, "castle-api not running", hint="castle apply")
         )
 
     # Generated artifacts.
@@ -244,7 +244,7 @@ def _check_runtime(config) -> list[Check]:
                 FAIL,
                 "generated specs missing",
                 detail=", ".join(missing),
-                hint="castle deploy",
+                hint="castle apply",
             )
         )
 
@@ -347,7 +347,7 @@ def _check_tls_exposure(config) -> list[Check]:
                     WARN,
                     "castle-tunnel not running",
                     detail="public routes are down",
-                    hint="castle service start castle-tunnel",
+                    hint="enable castle-tunnel in its deployment, then: castle apply",
                 )
             )
         checks.append(_check_public_dns(config))
@@ -364,7 +364,7 @@ def _check_public_dns(config) -> Check:
     records. The required permission is a single DNS:Edit (Cloudflare's 'Edit zone
     DNS' template), which also grants the zone lookup — so a correctly-scoped token
     passes this probe. Write itself isn't exercised (that would mutate); a
-    DNS:Read-only token would false-pass here but `castle deploy` then surfaces a
+    DNS:Read-only token would false-pass here but `castle apply` then surfaces a
     403 with the fix. Absent token → WARN (CNAMEs stay manual), not a failure.
     """
     import urllib.error
diff --git a/cli/src/castle_cli/commands/gateway.py b/cli/src/castle_cli/commands/gateway.py
index c950ea0..bb6ea13 100644
--- a/cli/src/castle_cli/commands/gateway.py
+++ b/cli/src/castle_cli/commands/gateway.py
@@ -20,7 +20,7 @@ def _write_generated_files() -> None:
     ensure_dirs()
 
     if not REGISTRY_PATH.exists():
-        print("Error: no registry found. Run 'castle deploy' first.")
+        print("Error: no registry found. Run 'castle apply' first.")
         return
 
     registry = load_registry()
@@ -56,7 +56,7 @@ def run_gateway(args: argparse.Namespace) -> int:
 def _gateway_dry_run() -> int:
     """Print generated Caddyfile without applying."""
     if not REGISTRY_PATH.exists():
-        print("Error: no registry found. Run 'castle deploy' first.")
+        print("Error: no registry found. Run 'castle apply' first.")
         return 1
 
     registry = load_registry()
@@ -126,7 +126,7 @@ def _gateway_status() -> int:
     print(f"Gateway: {'running' if status == 'active' else status}")
 
     if not REGISTRY_PATH.exists():
-        print("  (no registry — run 'castle deploy')")
+        print("  (no registry — run 'castle apply')")
         return 0
 
     from castle_core.generators.caddyfile import compute_routes
diff --git a/cli/src/castle_cli/commands/info.py b/cli/src/castle_cli/commands/info.py
index f3355ca..633e016 100644
--- a/cli/src/castle_cli/commands/info.py
+++ b/cli/src/castle_cli/commands/info.py
@@ -147,7 +147,7 @@ def run_info(args: argparse.Namespace) -> int:
         if deployed.secret_env_keys:
             print(f"  {BOLD}secrets{RESET}:    {', '.join(deployed.secret_env_keys)}")
     else:
-        print(f"\n  {DIM}not deployed (run 'castle deploy'){RESET}")
+        print(f"\n  {DIM}not applied (run 'castle apply'){RESET}")
 
     # Show CLAUDE.md if it exists
     source_dir = None
diff --git a/cli/src/castle_cli/commands/run_cmd.py b/cli/src/castle_cli/commands/run_cmd.py
index 988dd9e..1fdb9bd 100644
--- a/cli/src/castle_cli/commands/run_cmd.py
+++ b/cli/src/castle_cli/commands/run_cmd.py
@@ -82,12 +82,12 @@ def run_run(args: argparse.Namespace) -> int:
 
     # Service: deployed command from the registry.
     if not REGISTRY_PATH.exists():
-        print("Error: no registry found. Run 'castle deploy' first.")
+        print("Error: no registry found. Run 'castle apply' first.")
         return 1
 
     registry = load_registry()
     if name not in registry.deployed:
-        print(f"Error: '{name}' is not a deployed service. Run 'castle deploy' first.")
+        print(f"Error: '{name}' is not a deployed service. Run 'castle apply' first.")
         return 1
 
     deployed = registry.deployed[name]
diff --git a/cli/src/castle_cli/commands/service.py b/cli/src/castle_cli/commands/service.py
index c60d81a..fe85287 100644
--- a/cli/src/castle_cli/commands/service.py
+++ b/cli/src/castle_cli/commands/service.py
@@ -7,13 +7,10 @@ import subprocess
 
 from castle_core.generators.systemd import (
     SYSTEMD_USER_DIR,
-    generate_timer,
-    generate_unit_from_deployed,
     timer_name,
     unit_name,
 )
 from castle_core.manifest import kind_for
-from castle_core.registry import REGISTRY_PATH, load_registry
 
 from castle_cli.config import (
     CastleConfig,
@@ -45,44 +42,35 @@ _PAST = {"start": "started", "stop": "stopped", "restart": "restarted"}
 
 
 def run_service_cmd(args: argparse.Namespace) -> int:
-    """`castle service <enable|disable|start|stop|restart> <name>`."""
-    sub = args.service_command
+    """`castle service restart <name>` — the imperative bounce (only verb left).
+
+    Lifecycle (deploy/enable/disable/start/stop) is now convergence: `castle apply`.
+    """
     config = load_config()
-    if sub == "enable":
-        if getattr(args, "dry_run", False):
-            return _service_dry_run(config, args.name)
-        return _service_enable(config, args.name)
-    if sub == "disable":
-        return _service_disable(config, args.name)
-    if sub in ("start", "stop", "restart"):
-        return _unit_action(config, args.name, sub, is_job=False)
-    return 1
+    return _unit_action(config, args.name, "restart", is_job=False)
 
 
 def run_job_cmd(args: argparse.Namespace) -> int:
-    """`castle job <enable|disable|start|stop|restart> <name>` (acts on the timer)."""
-    sub = args.job_command
+    """`castle job restart <name>` — bounce the job's timer."""
     config = load_config()
-    if sub == "enable":
-        return _service_enable(config, args.name)  # enable_service handles timers
-    if sub == "disable":
-        return _service_disable(config, args.name)
-    if sub in ("start", "stop", "restart"):
-        return _unit_action(config, args.name, sub, is_job=True)
-    return 1
+    return _unit_action(config, args.name, "restart", is_job=True)
 
 
-def run_platform(args: argparse.Namespace) -> int:
-    """Top-level `castle start|stop|restart` — the whole platform."""
+def run_restart(args: argparse.Namespace) -> int:
+    """Top-level `castle restart [name]` — bounce one deployment, or all of them.
+
+    An imperative op: it re-actualizes current desired state, it does not change it
+    (that's `castle apply`).
+    """
     config = load_config()
-    action = args.command
-    if action == "start":
-        return _services_start(config)
-    if action == "stop":
-        return _services_stop(config)
-    if action == "restart":
+    name = getattr(args, "name", None)
+    if not name:
         return _services_restart(config)
-    return 1
+    dep = config.deployments.get(name)
+    if dep is None:
+        print(f"Error: no deployment '{name}'.")
+        return 1
+    return _unit_action(config, name, "restart", is_job=(kind_for(dep) == "job"))
 
 
 _GATEWAY_NAME = "castle-gateway"
@@ -256,78 +244,3 @@ def _service_status(config: CastleConfig) -> int:
     print()
     return 0
 
-
-def _service_dry_run(config: CastleConfig, name: str) -> int:
-    """Print the generated systemd unit(s) without installing."""
-    if REGISTRY_PATH.exists():
-        registry = load_registry()
-        if name in registry.deployed:
-            deployed = registry.deployed[name]
-            systemd_spec = None
-            dep = config.deployments.get(name)
-            manage = getattr(dep, "manage", None)
-            if manage and manage.systemd:
-                systemd_spec = manage.systemd
-
-            svc_unit = unit_name(name)
-            svc_content = generate_unit_from_deployed(name, deployed, systemd_spec)
-            print(f"# {svc_unit}")
-            print(svc_content)
-
-            if deployed.schedule:
-                timer_content = generate_timer(
-                    name,
-                    schedule=deployed.schedule,
-                    description=deployed.description,
-                )
-                print(f"# {timer_name(name)}")
-                print(timer_content)
-            return 0
-
-    print(f"Error: '{name}' not found in registry. Run 'castle deploy' first.")
-    return 1
-
-
-def _services_start(config: CastleConfig) -> int:
-    """Start all managed services and gateway."""
-    if not REGISTRY_PATH.exists():
-        print("Error: no registry found. Run 'castle deploy' first.")
-        return 1
-
-    ensure_dirs()
-
-    from castle_core.config import SPECS_DIR
-    from castle_core.generators.caddyfile import generate_caddyfile_from_registry
-
-    registry = load_registry()
-    caddyfile_path = SPECS_DIR / "Caddyfile"
-    caddyfile_path.write_text(generate_caddyfile_from_registry(registry))
-    print(f"Generated {caddyfile_path}")
-
-    # Activate every deployment in its mode: systemd unit / timer, gateway route
-    # (static), or PATH install (tool). activate() dispatches by manager.
-    for name in config.deployments:
-        if name not in registry.deployed:
-            print(f"  {name}: skipped (not in registry, run 'castle deploy')")
-            continue
-        _service_enable(config, name)
-
-    print(f"\nDashboard: http://localhost:{config.gateway.port}")
-    return 0
-
-
-def _services_stop(config: CastleConfig) -> int:
-    """Stop all managed services and jobs."""
-    for name in config.jobs:
-        tmr_unit = timer_name(name)
-        subprocess.run(["systemctl", "--user", "stop", tmr_unit], check=False)
-        svc_unit = unit_name(name)
-        subprocess.run(["systemctl", "--user", "stop", svc_unit], check=False)
-        print(f"  {name}: stopped")
-
-    for name in config.services:
-        svc_unit = unit_name(name)
-        subprocess.run(["systemctl", "--user", "stop", svc_unit], check=False)
-        print(f"  {name}: stopped")
-
-    return 0
diff --git a/cli/src/castle_cli/commands/tool.py b/cli/src/castle_cli/commands/tool.py
index 4a11b57..3a3cefd 100644
--- a/cli/src/castle_cli/commands/tool.py
+++ b/cli/src/castle_cli/commands/tool.py
@@ -137,7 +137,7 @@ def run_tool_info(args: argparse.Namespace) -> int:
     if comp.system_dependencies:
         print(f"  {BOLD}requires{RESET}:    {', '.join(comp.system_dependencies)}")
     if not installed:
-        print(f"\n  {DIM}install with: castle tool install {name}{RESET}")
+        print(f"\n  {DIM}enable it in its deployment, then: castle apply{RESET}")
     else:
         print(f"\n  {DIM}run `{exes[0]} --help` for arguments{RESET}")
     print()
diff --git a/cli/src/castle_cli/main.py b/cli/src/castle_cli/main.py
index 0f53a8a..f3dc587 100644
--- a/cli/src/castle_cli/main.py
+++ b/cli/src/castle_cli/main.py
@@ -70,13 +70,6 @@ def _build_program_group(subparsers: argparse._SubParsersAction) -> None:
     _add_name(p, "Program name")
     p.add_argument("extra", nargs=argparse.REMAINDER, help="Extra args passed to the program")
 
-    sub.add_parser("install", help="Activate a program (tool→PATH, static→served)").add_argument(
-        "name", nargs="?", help="Program (default: all)"
-    )
-    sub.add_parser("uninstall", help="Deactivate a program").add_argument(
-        "name", nargs="?", help="Program"
-    )
-
     for verb in DEV_VERBS:
         p = sub.add_parser(verb, help=f"Run {verb}")
         _add_name(p, "Program (default: all)", optional=True)
@@ -95,11 +88,6 @@ def _build_tool_group(subparsers: argparse._SubParsersAction) -> None:
     _add_name(p, "Tool name")
     p.add_argument("--json", action="store_true", help="Machine-readable output")
 
-    sub.add_parser("install", help="Install a tool on PATH").add_argument("name", help="Tool name")
-    sub.add_parser("uninstall", help="Remove a tool from PATH").add_argument(
-        "name", help="Tool name"
-    )
-
 
 def _add_service_create(sub: argparse._SubParsersAction, kind: str) -> None:
     p = sub.add_parser("create", help=f"Create a {kind} in castle.yaml")
@@ -148,16 +136,9 @@ def _build_deployment_group(subparsers: argparse._SubParsersAction, kind: str) -
     p.add_argument("--purge-data", action="store_true", help=argparse.SUPPRESS)
 
     cap = f"{kind.capitalize()} name"
-    _add_name(sub.add_parser("deploy", help=f"Deploy this {kind} (unit + gateway)"), cap)
-
-    p = sub.add_parser("enable", help=f"Enable and start the {kind}")
-    _add_name(p, cap)
-    if kind == "service":
-        p.add_argument("--dry-run", action="store_true", help="Print the unit without installing")
-    _add_name(sub.add_parser("disable", help=f"Stop and disable the {kind}"), cap)
-    _add_name(sub.add_parser("start", help=f"Start the {kind}"), cap)
-    _add_name(sub.add_parser("stop", help=f"Stop the {kind}"), cap)
-    _add_name(sub.add_parser("restart", help=f"Restart the {kind}"), cap)
+    # Lifecycle is convergence: `castle apply [name]`. `restart` stays as the one
+    # imperative bounce that doesn't change desired state.
+    _add_name(sub.add_parser("restart", help=f"Restart the {kind} (imperative bounce)"), cap)
 
     p = sub.add_parser("logs", help=f"View {kind} logs")
     _add_name(p, f"{kind.capitalize()} name")
@@ -188,16 +169,23 @@ def build_parser() -> argparse.ArgumentParser:
     p.add_argument("--dry-run", action="store_true")
     gw_sub.add_parser("status", help="Show gateway status")
 
-    # Platform-wide lifecycle (top-level)
-    subparsers.add_parser("start", help="Start all services and the gateway")
-    subparsers.add_parser("stop", help="Stop all services and the gateway")
-    subparsers.add_parser("restart", help="Restart all services and jobs")
+    # Convergence — the one lifecycle verb. Renders units/Caddyfile/tunnel, then
+    # reconciles the runtime to match config (activate/restart/deactivate).
+    p = subparsers.add_parser(
+        "apply", help="Converge the running system to match config (render + reconcile)"
+    )
+    p.add_argument("name", nargs="?", help="Single deployment to converge (default: all)")
+    p.add_argument(
+        "--plan", action="store_true", help="Show the diff without changing anything"
+    )
+
+    # Imperative ops (don't change desired state)
+    p = subparsers.add_parser("restart", help="Restart deployment(s) — an imperative bounce")
+    p.add_argument("name", nargs="?", help="Deployment to restart (default: all)")
     subparsers.add_parser("status", help="Show status across the platform")
     subparsers.add_parser(
         "doctor", help="Diagnose setup + runtime health, with next-step hints"
     )
-    p = subparsers.add_parser("deploy", help="Apply config to runtime (units + Caddyfile)")
-    p.add_argument("name", nargs="?", help="Service/job to deploy (default: all)")
 
     # Cross-resource overview
     p = subparsers.add_parser("list", help="List programs, services, jobs, and tools")
@@ -215,7 +203,7 @@ def build_parser() -> argparse.ArgumentParser:
 def _dispatch_program(args: argparse.Namespace) -> int:
     sub = args.program_command
     if not sub:
-        verbs = "list|info|create|add|clone|delete|run|install|uninstall|" + "|".join(DEV_VERBS)
+        verbs = "list|info|create|add|clone|delete|run|" + "|".join(DEV_VERBS)
         print(f"Usage: castle program {{{verbs}}}")
         return 1
     if sub == "list":
@@ -246,14 +234,6 @@ def _dispatch_program(args: argparse.Namespace) -> int:
         from castle_cli.commands.run_cmd import run_run
 
         return run_run(args)
-    if sub == "install":
-        from castle_cli.commands.dev import run_install
-
-        return run_install(args)
-    if sub == "uninstall":
-        from castle_cli.commands.dev import run_uninstall
-
-        return run_uninstall(args)
     if sub in DEV_VERBS:
         from castle_cli.commands.dev import run_verb
 
@@ -264,7 +244,7 @@ def _dispatch_program(args: argparse.Namespace) -> int:
 def _dispatch_tool(args: argparse.Namespace) -> int:
     sub = args.tool_command
     if not sub:
-        print("Usage: castle tool {list|info|install|uninstall}")
+        print("Usage: castle tool {list|info}  (install/uninstall → edit config + castle apply)")
         return 1
     if sub == "list":
         from castle_cli.commands.tool import run_tool_list
@@ -274,21 +254,13 @@ def _dispatch_tool(args: argparse.Namespace) -> int:
         from castle_cli.commands.tool import run_tool_info
 
         return run_tool_info(args)
-    if sub == "install":
-        from castle_cli.commands.dev import run_install
-
-        return run_install(args)
-    if sub == "uninstall":
-        from castle_cli.commands.dev import run_uninstall
-
-        return run_uninstall(args)
     return 1
 
 
 def _dispatch_deployment(args: argparse.Namespace, kind: str) -> int:
     sub = getattr(args, f"{kind}_command")
     if not sub:
-        verbs = "list|info|create|delete|deploy|enable|disable|start|stop|restart|logs"
+        verbs = "list|info|create|delete|restart|logs  (deploy/enable/... → castle apply)"
         print(f"Usage: castle {kind} {{{verbs}}}")
         return 1
     if sub == "list":
@@ -307,11 +279,7 @@ def _dispatch_deployment(args: argparse.Namespace, kind: str) -> int:
         from castle_cli.commands.delete import run_delete
 
         return run_delete(args)
-    if sub == "deploy":
-        from castle_cli.commands.deploy import run_deploy
-
-        return run_deploy(args)
-    if sub in ("enable", "disable", "start", "stop", "restart"):
+    if sub == "restart":
         from castle_cli.commands.service import run_job_cmd, run_service_cmd
 
         return run_service_cmd(args) if kind == "service" else run_job_cmd(args)
@@ -341,10 +309,14 @@ def main() -> int:
         from castle_cli.commands.gateway import run_gateway
 
         return run_gateway(args)
-    if cmd in ("start", "stop", "restart"):
-        from castle_cli.commands.service import run_platform
+    if cmd == "apply":
+        from castle_cli.commands.apply import run_apply
 
-        return run_platform(args)
+        return run_apply(args)
+    if cmd == "restart":
+        from castle_cli.commands.service import run_restart
+
+        return run_restart(args)
     if cmd == "status":
         from castle_cli.commands.service import run_status
 
@@ -353,10 +325,6 @@ def main() -> int:
         from castle_cli.commands.doctor import run_doctor
 
         return run_doctor(args)
-    if cmd == "deploy":
-        from castle_cli.commands.deploy import run_deploy
-
-        return run_deploy(args)
     if cmd == "list":
         from castle_cli.commands.list_cmd import run_list
 
diff --git a/cli/tests/test_doctor.py b/cli/tests/test_doctor.py
index ecc1142..b7a9099 100644
--- a/cli/tests/test_doctor.py
+++ b/cli/tests/test_doctor.py
@@ -26,7 +26,7 @@ class TestDoctor:
         assert "repo: not set" in out
         assert "control plane missing" in out
         # Every failing check offers a concrete next command.
-        assert "castle deploy" in out
+        assert "castle apply" in out
 
     def test_load_failure_is_first_fail(self, capsys: object) -> None:
         """A castle.yaml that won't load is surfaced as a FAIL, not a traceback."""
diff --git a/core/src/castle_core/deploy.py b/core/src/castle_core/deploy.py
index 18ffbed..b36e84e 100644
--- a/core/src/castle_core/deploy.py
+++ b/core/src/castle_core/deploy.py
@@ -70,6 +70,30 @@ class DeployResult:
     registry: NodeRegistry | None = None
 
 
+@dataclass
+class ApplyResult:
+    """Result of a converge (`castle apply`): what actually changed.
+
+    `deploy` renders config → artifacts; `apply` renders *and then* reconciles the
+    running system to match, so the interesting output is the diff it enacted.
+    """
+
+    activated: list[str] = field(default_factory=list)
+    restarted: list[str] = field(default_factory=list)
+    deactivated: list[str] = field(default_factory=list)
+    unchanged: list[str] = field(default_factory=list)
+    pruned: list[str] = field(default_factory=list)
+    messages: list[str] = field(default_factory=list)
+    registry: NodeRegistry | None = None
+    # True for a `--plan` run: the diff was computed but nothing was written or
+    # activated. Lets callers render "would activate…" vs "activated…".
+    planned: bool = False
+
+    @property
+    def changed(self) -> bool:
+        return bool(self.activated or self.restarted or self.deactivated or self.pruned)
+
+
 def deploy(target_name: str | None = None, root: Path | None = None) -> DeployResult:
     """Deploy from castle.yaml to ~/.castle/.
 
@@ -86,16 +110,7 @@ def deploy(target_name: str | None = None, root: Path | None = None) -> DeployRe
     ensure_dirs()
 
     # Build node config
-    node = NodeConfig(
-        castle_root=str(config.root),
-        gateway_port=config.gateway.port,
-        gateway_tls=config.gateway.tls,
-        gateway_domain=config.gateway.domain,
-        acme_email=config.gateway.acme_email,
-        acme_dns_provider=config.gateway.acme_dns_provider,
-        public_domain=config.gateway.public_domain,
-        tunnel_id=config.gateway.tunnel_id,
-    )
+    node = _node_config(config)
 
     # Load existing registry to preserve entries not being redeployed,
     # or start fresh if deploying all
@@ -158,6 +173,131 @@ def deploy(target_name: str | None = None, root: Path | None = None) -> DeployRe
     return result
 
 
+def _node_config(config: CastleConfig) -> NodeConfig:
+    """The registry NodeConfig derived from a config's gateway settings."""
+    return NodeConfig(
+        castle_root=str(config.root),
+        gateway_port=config.gateway.port,
+        gateway_tls=config.gateway.tls,
+        gateway_domain=config.gateway.domain,
+        acme_email=config.gateway.acme_email,
+        acme_dns_provider=config.gateway.acme_dns_provider,
+        public_domain=config.gateway.public_domain,
+        tunnel_id=config.gateway.tunnel_id,
+    )
+
+
+def _unit_file_for(name: str, is_job: bool) -> Path:
+    """On-disk systemd unit path for a deployment (timer if it's a job)."""
+    return SYSTEMD_USER_DIR / (timer_name(name) if is_job else unit_name(name))
+
+
+def _unit_bytes(name: str, is_job: bool) -> str | None:
+    """Current unit-file contents, or None if it isn't written yet."""
+    path = _unit_file_for(name, is_job)
+    return path.read_text() if path.exists() else None
+
+
+def apply(
+    target_name: str | None = None,
+    root: Path | None = None,
+    plan: bool = False,
+) -> ApplyResult:
+    """Converge the running system to match config — the one honest bring-up.
+
+    `apply` = `deploy` (render units/Caddyfile/tunnel) **plus** reconcile: activate
+    every enabled deployment that isn't live, restart any whose unit changed,
+    deactivate the disabled ones. It replaces the old two-step ``deploy && start``
+    and the per-kind enable/disable/install verbs — the mechanism varies by manager
+    (systemd unit / PATH install / gateway route), the verb never does.
+
+    `plan=True` computes and returns the diff **without writing or touching the
+    runtime** (the ``--plan`` dry run).
+    """
+    import asyncio
+
+    from castle_core.lifecycle import activate, deactivate, is_active
+
+    config = load_config(root)
+    names = [n for n in config.deployments if not target_name or n == target_name]
+    is_job = {n: (n in config.jobs) for n in names}
+
+    # Snapshot BEFORE rendering: liveness + current unit bytes (for restart-on-change).
+    before_active = {n: is_active(n, config) for n in names}
+    before_unit = {n: _unit_bytes(n, is_job[n]) for n in names}
+
+    # Desired state, rendered in memory to classify each deployment. For a real run
+    # this is recomputed by deploy() below (which also writes it); cheap and keeps
+    # the plan/apply classification identical.
+    desired = {n: _build_deployed(config, n, config.deployments[n], []) for n in names}
+
+    def _classify(name: str, after_unit: str | None) -> str:
+        dep = desired[name]
+        if not dep.enabled:
+            return "deactivate" if before_active[name] else "unchanged"
+        if not before_active[name]:
+            return "activate"
+        if dep.manager == "systemd" and before_unit[name] != after_unit:
+            return "restart"
+        return "unchanged"
+
+    result = ApplyResult(registry=NodeRegistry(node=_node_config(config), deployed=desired))
+
+    if plan:
+        # No writes: for systemd, predict the new unit bytes by rendering to a string
+        # so "would restart" is accurate; other managers never restart.
+        result.planned = True
+        for name in names:
+            after = _render_unit_preview(config, name, desired[name], is_job[name])
+            _record(result, name, _classify(name, after))
+        return result
+
+    # Real run: render everything (writes units/Caddyfile/tunnel, daemon-reload,
+    # gateway reload, orphan prune), then reconcile the runtime.
+    deploy_result = deploy(target_name, root)
+    result.messages = list(deploy_result.messages)
+    result.registry = deploy_result.registry
+
+    for name in names:
+        after_unit = _unit_bytes(name, is_job[name])
+        action = _classify(name, after_unit)
+        if action == "activate":
+            asyncio.run(activate(name, config, config.root))
+            result.activated.append(name)
+        elif action == "deactivate":
+            asyncio.run(deactivate(name, config, config.root))
+            result.deactivated.append(name)
+        elif action == "restart":
+            unit = timer_name(name) if is_job[name] else unit_name(name)
+            subprocess.run(["systemctl", "--user", "restart", unit], check=False)
+            result.restarted.append(name)
+        else:
+            result.unchanged.append(name)
+
+    return result
+
+
+def _record(result: ApplyResult, name: str, action: str) -> None:
+    {
+        "activate": result.activated,
+        "deactivate": result.deactivated,
+        "restart": result.restarted,
+        "unchanged": result.unchanged,
+    }[action].append(name)
+
+
+def _render_unit_preview(
+    config: CastleConfig, name: str, dep: Deployment, is_job: bool
+) -> str | None:
+    """The unit bytes `deploy` would write for the deployment we'd restart (the
+    .timer for a job, the .service for a service), for --plan restart detection.
+    None when there's no unit to compare (non-systemd, or unmanaged)."""
+    files = _render_unit_files(config, name, dep)
+    if not files:
+        return None
+    return files.get(timer_name(name) if is_job else unit_name(name))
+
+
 # Gateway service name in the registry → its systemd unit (castle-castle-gateway).
 _GATEWAY_NAME = "castle-gateway"
 
@@ -387,6 +527,7 @@ def _build_deployed(
             public=bool(dep.public),
             static_root=static_root,
             managed=False,
+            enabled=dep.enabled,
         )
     if isinstance(dep, PathDeployment):
         return Deployment(
@@ -396,6 +537,7 @@ def _build_deployed(
             kind=kind,
             stack=stack,
             managed=False,
+            enabled=dep.enabled,
         )
     if isinstance(dep, RemoteDeployment):
         return Deployment(
@@ -406,6 +548,7 @@ def _build_deployed(
             stack=stack,
             base_url=dep.base_url,
             managed=False,
+            enabled=dep.enabled,
         )
 
     # systemd: a supervised process (a service, or a job when scheduled).
@@ -465,6 +608,7 @@ def _build_deployed(
         public=bool(dep.public and expose),
         schedule=getattr(dep, "schedule", None),
         managed=managed,
+        enabled=dep.enabled,
     )
 
 
@@ -711,31 +855,40 @@ def _prune_orphans(registry: NodeRegistry, messages: list[str]) -> None:
                 _teardown_unit(path.name, messages)
 
 
+def _render_unit_files(
+    config: CastleConfig, name: str, deployed: Deployment
+) -> dict[str, str]:
+    """The exact unit files `deploy` would write for a deployment: {filename: content}.
+
+    Empty for a non-systemd-managed deployment (caddy/path/none have no unit). The
+    single source of truth for unit bytes — used both to write units and to predict
+    restart-on-change in `apply`/`--plan`, so the prediction can never drift from
+    what actually gets written.
+    """
+    if not deployed.managed:
+        return {}
+    systemd_spec = None
+    dep = config.deployments.get(name)
+    manage = getattr(dep, "manage", None)
+    if manage and manage.systemd:
+        systemd_spec = manage.systemd
+
+    files = {
+        unit_name(name): generate_unit_from_deployed(
+            name, deployed, systemd_spec, env_file=unit_env_file(deployed, name)
+        )
+    }
+    if deployed.schedule:
+        files[timer_name(name)] = generate_timer(
+            name, schedule=deployed.schedule, description=deployed.description
+        )
+    return files
+
+
 def _generate_systemd_units(config: CastleConfig, registry: NodeRegistry) -> None:
     """Generate systemd units from the registry."""
     SYSTEMD_USER_DIR.mkdir(parents=True, exist_ok=True)
 
     for name, deployed in registry.deployed.items():
-        if not deployed.managed:
-            continue
-
-        systemd_spec = None
-        dep = config.deployments.get(name)
-        manage = getattr(dep, "manage", None)
-        if manage and manage.systemd:
-            systemd_spec = manage.systemd
-
-        svc_name = unit_name(name)
-        svc_content = generate_unit_from_deployed(
-            name, deployed, systemd_spec, env_file=unit_env_file(deployed, name)
-        )
-        (SYSTEMD_USER_DIR / svc_name).write_text(svc_content)
-
-        if deployed.schedule:
-            timer_content = generate_timer(
-                name,
-                schedule=deployed.schedule,
-                description=deployed.description,
-            )
-            tmr_name = timer_name(name)
-            (SYSTEMD_USER_DIR / tmr_name).write_text(timer_content)
+        for fname, content in _render_unit_files(config, name, deployed).items():
+            (SYSTEMD_USER_DIR / fname).write_text(content)
diff --git a/core/src/castle_core/generators/caddyfile.py b/core/src/castle_core/generators/caddyfile.py
index f2abf9a..48aadf1 100644
--- a/core/src/castle_core/generators/caddyfile.py
+++ b/core/src/castle_core/generators/caddyfile.py
@@ -77,6 +77,10 @@ def _local_routes(
     deployments = getattr(config, "deployments", None)
     if deployments is not None:
         for name, dep in sorted(deployments.items()):
+            # A disabled deployment is defined but not running — no route (else it
+            # would 502). `castle apply` converges it off.
+            if not dep.enabled:
+                continue
             if isinstance(dep, CaddyDeployment):
                 src = _program_source(config, dep.program)
                 if src is not None:
@@ -88,6 +92,8 @@ def _local_routes(
         return out
     # No config → route from the deployed registry snapshot.
     for name, d in sorted(registry.deployed.items()):
+        if not d.enabled:
+            continue
         if d.static_root:
             out.append((name, "static", d.static_root))
         elif d.subdomain and (d.port or d.base_url):
diff --git a/core/src/castle_core/manifest.py b/core/src/castle_core/manifest.py
index e92c8fd..87543c2 100644
--- a/core/src/castle_core/manifest.py
+++ b/core/src/castle_core/manifest.py
@@ -311,6 +311,11 @@ class DeploymentBase(BaseModel):
     )
     description: str | None = None
     defaults: DefaultsSpec | None = None
+    # Declared on/off state. `castle apply` converges reality to this: enabled
+    # deployments are activated (service started, tool installed, route served),
+    # disabled ones are deactivated but kept in the catalog. This is *desired
+    # state*, not a runtime toggle — the only way to durably stop something.
+    enabled: bool = True
 
 
 class SystemdDeployment(DeploymentBase):
diff --git a/core/src/castle_core/registry.py b/core/src/castle_core/registry.py
index 589e7d8..1a1e758 100644
--- a/core/src/castle_core/registry.py
+++ b/core/src/castle_core/registry.py
@@ -72,6 +72,9 @@ class Deployment:
     base_url: str | None = None
     schedule: str | None = None
     managed: bool = False
+    # Declared desired state (from the deployment's `enabled:`). `castle apply`
+    # activates enabled deployments and deactivates disabled ones. Default True.
+    enabled: bool = True
 
 
 @dataclass
@@ -155,6 +158,7 @@ def load_registry(path: Path | None = None) -> NodeRegistry:
             base_url=comp_data.get("base_url"),
             schedule=comp_data.get("schedule"),
             managed=comp_data.get("managed", False),
+            enabled=comp_data.get("enabled", True),
         )
 
     return NodeRegistry(node=node, deployed=deployed)
@@ -225,6 +229,10 @@ def save_registry(registry: NodeRegistry, path: Path | None = None) -> None:
             entry["schedule"] = comp.schedule
         if comp.managed:
             entry["managed"] = comp.managed
+        # Only emit when disabled — default-True omission keeps existing
+        # registries byte-identical and matches the load-side default.
+        if not comp.enabled:
+            entry["enabled"] = comp.enabled
         data["deployed"][name] = entry
 
     with open(path, "w") as f:
diff --git a/core/tests/test_apply.py b/core/tests/test_apply.py
new file mode 100644
index 0000000..03c199c
--- /dev/null
+++ b/core/tests/test_apply.py
@@ -0,0 +1,74 @@
+"""Tests for `castle apply` convergence — the diff classification (plan mode).
+
+Plan mode computes the activate/restart/deactivate/unchanged buckets without
+writing or touching the runtime, so it's the deterministic way to test the diff.
+`is_active` is patched to control the "before" state; unit bytes come from the
+(empty) temp home, so a live systemd service with no prior unit reads as changed.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import patch
+
+from castle_core.deploy import _render_unit_preview, apply
+from castle_core.registry import Deployment
+
+
+def _plan(castle_root: Path, active: dict[str, bool]):
+    """Run apply(plan=True) with is_active stubbed to `active` (default False)."""
+    with patch("castle_core.lifecycle.is_active", side_effect=lambda n, c: active.get(n, False)):
+        return apply(root=castle_root, plan=True)
+
+
+class TestApplyPlan:
+    def test_fresh_converge_activates_enabled(self, castle_root: Path) -> None:
+        """Nothing running → every enabled deployment is 'activate'; no writes."""
+        result = _plan(castle_root, active={})
+
+        assert result.planned is True
+        assert set(result.activated) == {"test-svc", "test-tool", "test-job"}
+        assert result.deactivated == []
+        assert result.restarted == []
+
+    def test_disabled_active_deployment_deactivates(self, castle_root: Path) -> None:
+        """A deployment with enabled:false that's currently up → 'deactivate'."""
+        # Turn the tool off in config.
+        tool = castle_root / "services" / "test-tool.yaml"
+        tool.write_text(tool.read_text() + "enabled: false\n")
+
+        result = _plan(castle_root, active={"test-tool": True})
+
+        assert "test-tool" in result.deactivated
+        assert "test-tool" not in result.activated
+
+    def test_disabled_inactive_is_unchanged(self, castle_root: Path) -> None:
+        """enabled:false and already down → nothing to do."""
+        tool = castle_root / "services" / "test-tool.yaml"
+        tool.write_text(tool.read_text() + "enabled: false\n")
+
+        result = _plan(castle_root, active={})
+
+        assert "test-tool" in result.unchanged
+        assert "test-tool" not in result.deactivated
+
+    def test_active_service_with_changed_unit_restarts(self, castle_root: Path) -> None:
+        """An up systemd service whose rendered unit differs from disk → 'restart'.
+
+        The temp home has no prior unit file (before-bytes = None), so any live
+        systemd deployment classifies as changed → restart, not a silent no-op.
+        """
+        result = _plan(castle_root, active={"test-svc": True})
+
+        assert "test-svc" in result.restarted
+        assert "test-svc" not in result.activated
+        assert result.changed is True
+
+
+def test_render_unit_preview_none_for_non_systemd() -> None:
+    """Non-systemd managers have no unit file — preview is None (never 'restart').
+
+    A path deployment is unmanaged, so the renderer returns before touching config.
+    """
+    tool = Deployment(manager="path", run_cmd=[], kind="tool")
+    assert _render_unit_preview(None, "x", tool, is_job=False) is None  # type: ignore[arg-type]