Runtime half: convergent deploy + unified active lifecycle

Convergent deploy (prefix-based prune): castle deploy now removes orphaned
castle-* systemd units/timers no longer in the registry. Full deploy only;
partial (--name) deploys preserve siblings. Fixes the orphaned-timer / stale-
path drift class.

Container runner: derive --name from the service name (castle-<name>) instead
of the image name; skip the <PREFIX>_DATA_DIR env injection for containers
(avoids colliding with image env namespaces like NEO4J_*).

Unified active lifecycle (core/lifecycle.py): install/uninstall keep their
names but become activate/deactivate, dispatching by behavior — tool→PATH,
daemon/self-serving-frontend/job→systemd, static frontend→served. is_active
reports a uniform state (PATH-independent for tools). The CLI install/uninstall
and service enable/disable route through the extracted core.

API + dashboard: surface a uniform `active` state on programs alongside
`installed`; ProgramDetail shows active/inactive.

Tests: test_deploy_prune (6), test_lifecycle (5). 168 tests pass; lint clean.

core/tests/test_deploy_prune.py

@@ -0,0 +1,90 @@
+"""Tests for convergent-deploy orphan pruning (prefix-based)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import patch
+
+from castle_core import deploy as deploy_mod
+from castle_core.deploy import _desired_unit_files, _prune_orphans
+from castle_core.registry import DeployedComponent, NodeConfig, NodeRegistry
+
+
+def _svc(managed: bool = True, schedule: str | None = None) -> DeployedComponent:
+    return DeployedComponent(
+        runner="python",
+        run_cmd=["x"],
+        env={},
+        managed=managed,
+        schedule=schedule,
+    )
+
+
+def _registry(**deployed: DeployedComponent) -> NodeRegistry:
+    return NodeRegistry(node=NodeConfig(castle_root="/x", gateway_port=9000), deployed=dict(deployed))
+
+
+def _touch(d: Path, *names: str) -> None:
+    for n in names:
+        (d / n).write_text("[Unit]\n")
+
+
+class TestDesiredUnitFiles:
+    def test_managed_service_yields_service_file(self) -> None:
+        reg = _registry(foo=_svc())
+        assert _desired_unit_files(reg) == {"castle-foo.service"}
+
+    def test_scheduled_job_yields_service_and_timer(self) -> None:
+        reg = _registry(job=_svc(schedule="0 2 * * *"))
+        assert _desired_unit_files(reg) == {"castle-job.service", "castle-job.timer"}
+
+    def test_unmanaged_excluded(self) -> None:
+        reg = _registry(foo=_svc(managed=False))
+        assert _desired_unit_files(reg) == set()
+
+
+class TestPruneOrphans:
+    def test_removes_orphan_keeps_desired(self, tmp_path: Path) -> None:
+        units = tmp_path / "systemd"
+        units.mkdir()
+        # keep-me is in the registry; gone is not; gone.timer is also orphaned
+        _touch(units, "castle-keep.service", "castle-gone.service", "castle-gone.timer")
+        reg = _registry(keep=_svc())
+        msgs: list[str] = []
+        with (
+            patch.object(deploy_mod, "SYSTEMD_USER_DIR", units),
+            patch.object(deploy_mod.subprocess, "run") as mock_run,
+        ):
+            _prune_orphans(reg, msgs)
+        assert (units / "castle-keep.service").exists()
+        assert not (units / "castle-gone.service").exists()
+        assert not (units / "castle-gone.timer").exists()
+        # stop + disable were invoked for each removed unit
+        assert mock_run.call_count >= 2
+        assert any("castle-gone.service" in m for m in msgs)
+
+    def test_unscheduled_job_drops_stale_timer_keeps_service(self, tmp_path: Path) -> None:
+        units = tmp_path / "systemd"
+        units.mkdir()
+        _touch(units, "castle-job.service", "castle-job.timer")
+        # job is still managed but no longer scheduled → its timer is now an orphan
+        reg = _registry(job=_svc(schedule=None))
+        with (
+            patch.object(deploy_mod, "SYSTEMD_USER_DIR", units),
+            patch.object(deploy_mod.subprocess, "run"),
+        ):
+            _prune_orphans(reg, [])
+        assert (units / "castle-job.service").exists()
+        assert not (units / "castle-job.timer").exists()
+
+    def test_demoted_to_unmanaged_is_pruned(self, tmp_path: Path) -> None:
+        units = tmp_path / "systemd"
+        units.mkdir()
+        _touch(units, "castle-foo.service")
+        reg = _registry(foo=_svc(managed=False))  # still in registry but unmanaged
+        with (
+            patch.object(deploy_mod, "SYSTEMD_USER_DIR", units),
+            patch.object(deploy_mod.subprocess, "run"),
+        ):
+            _prune_orphans(reg, [])
+        assert not (units / "castle-foo.service").exists()

core/tests/test_lifecycle.py

@@ -0,0 +1,47 @@
+"""Tests for the unified active lifecycle (is_active dispatch)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import patch
+
+from castle_core import lifecycle
+from castle_core.config import load_config
+
+
+class TestIsActive:
+    def test_service_uses_systemctl(self, castle_root: Path) -> None:
+        config = load_config(castle_root)
+        with patch.object(lifecycle, "_systemctl_active", return_value=True) as mock:
+            assert lifecycle.is_active("test-svc", config) is True
+        mock.assert_called_once_with("castle-test-svc.service")
+
+    def test_job_uses_timer(self, castle_root: Path) -> None:
+        config = load_config(castle_root)
+        with patch.object(lifecycle, "_systemctl_active", return_value=True) as mock:
+            assert lifecycle.is_active("test-job", config) is True
+        mock.assert_called_once_with("castle-test-job.timer")
+
+    def test_tool_checks_path(self, castle_root: Path) -> None:
+        config = load_config(castle_root)
+        # give the tool a source so the tool branch is reachable
+        config.programs["test-tool"].source = "/tmp/test-tool"
+        with patch.object(lifecycle, "_on_path", return_value=True) as mock:
+            assert lifecycle.is_active("test-tool", config) is True
+        mock.assert_called_once_with("test-tool")
+
+    def test_unknown_is_inactive(self, castle_root: Path) -> None:
+        config = load_config(castle_root)
+        assert lifecycle.is_active("does-not-exist", config) is False
+
+    def test_static_frontend_checks_content_dir(self, castle_root: Path, tmp_path: Path) -> None:
+        config = load_config(castle_root)
+        config.programs["fe"] = config.programs["test-tool"].model_copy(
+            update={"id": "fe", "behavior": "frontend", "source": "/tmp/fe"}
+        )
+        content = tmp_path / "content"
+        (content / "fe").mkdir(parents=True)
+        with patch.object(lifecycle, "CONTENT_DIR", content):
+            assert lifecycle.is_active("fe", config) is True
+        with patch.object(lifecycle, "CONTENT_DIR", tmp_path / "empty"):
+            assert lifecycle.is_active("fe", config) is False