diff --git a/app/src/pages/ProgramDetail.tsx b/app/src/pages/ProgramDetail.tsx
index 8faed97..ab458cf 100644
--- a/app/src/pages/ProgramDetail.tsx
+++ b/app/src/pages/ProgramDetail.tsx
@@ -85,10 +85,12 @@ export function ProgramDetailPage() {
{runnerLabel(component.runner)}
)}
- {component.installed !== null && (
+ {component.active !== null && (
<>
- Installed
- {component.installed ? "Yes" : "No"}
+ Active
+
+ {component.active ? "● active" : "○ inactive"}
+
)}
diff --git a/app/src/types/index.ts b/app/src/types/index.ts
index f3bf1c8..e26efda 100644
--- a/app/src/types/index.ts
+++ b/app/src/types/index.ts
@@ -51,6 +51,7 @@ export interface ProgramSummary {
commands: Record | null
system_dependencies: string[]
installed: boolean | null
+ active: boolean | null
actions: string[]
node: string | null
}
@@ -83,6 +84,7 @@ export interface ComponentSummary {
system_dependencies: string[]
schedule: string | null
installed: boolean | null
+ active: boolean | null
node: string | null
}
diff --git a/castle-api/src/castle_api/models.py b/castle-api/src/castle_api/models.py
index 22abd88..a954503 100644
--- a/castle-api/src/castle_api/models.py
+++ b/castle-api/src/castle_api/models.py
@@ -33,6 +33,7 @@ class ComponentSummary(BaseModel):
system_dependencies: list[str] = []
schedule: str | None = None
installed: bool | None = None
+ active: bool | None = None # uniform lifecycle state (on PATH / running / served)
node: str | None = None
@@ -99,6 +100,7 @@ class ProgramSummary(BaseModel):
commands: dict[str, list[list[str]]] | None = None
system_dependencies: list[str] = []
installed: bool | None = None
+ active: bool | None = None # uniform lifecycle state (on PATH / running / served)
actions: list[str] = []
node: str | None = None
diff --git a/castle-api/src/castle_api/routes.py b/castle-api/src/castle_api/routes.py
index 0899eca..4a560b7 100644
--- a/castle-api/src/castle_api/routes.py
+++ b/castle-api/src/castle_api/routes.py
@@ -342,6 +342,13 @@ def _program_from_spec(
if comp.source and (comp.stack or comp.commands):
installed = shutil.which(name) is not None
+ # Uniform lifecycle state (on PATH / running / served) — needs full config.
+ active: bool | None = None
+ if config is not None:
+ from castle_core.lifecycle import is_active
+
+ active = is_active(name, config)
+
return ProgramSummary(
id=name,
description=comp.description,
@@ -355,6 +362,7 @@ def _program_from_spec(
commands=_declared_commands_dict(comp),
system_dependencies=comp.system_dependencies,
installed=installed,
+ active=active,
actions=available_actions(comp),
)
diff --git a/cli/src/castle_cli/commands/dev.py b/cli/src/castle_cli/commands/dev.py
index 6e92e8d..1aedbd2 100644
--- a/cli/src/castle_cli/commands/dev.py
+++ b/cli/src/castle_cli/commands/dev.py
@@ -85,9 +85,40 @@ def run_check(args: argparse.Namespace) -> int:
return run_verb(args, "check")
+def _lifecycle(config: CastleConfig, name: str, deactivate: bool) -> bool:
+ """Activate or deactivate a single program. Returns True on success."""
+ from castle_core.lifecycle import activate
+ from castle_core.lifecycle import deactivate as do_deactivate
+
+ if name not in config.programs and name not in config.services and name not in config.jobs:
+ print(f"Unknown program: {name}")
+ return False
+ verb = "deactivate" if deactivate else "activate"
+ print(f"\n{'─' * 40}\n {verb}: {name}\n{'─' * 40}")
+ fn = do_deactivate if deactivate else activate
+ result = asyncio.run(fn(name, config, config.root))
+ if result.output:
+ print(result.output)
+ return result.status == "ok"
+
+
def run_install(args: argparse.Namespace) -> int:
- return run_verb(args, "install")
+ """Activate (install/deploy/serve) a program — verb word kept, meaning unified."""
+ config = load_config()
+ if getattr(args, "project", None):
+ return 0 if _lifecycle(config, args.project, deactivate=False) else 1
+ ok = all(
+ _lifecycle(config, name, deactivate=False)
+ for name, comp in config.programs.items()
+ if comp.source
+ )
+ return 0 if ok else 1
def run_uninstall(args: argparse.Namespace) -> int:
- return run_verb(args, "uninstall")
+ """Deactivate (uninstall/stop/unpublish) a program."""
+ config = load_config()
+ if getattr(args, "project", None):
+ return 0 if _lifecycle(config, args.project, deactivate=True) else 1
+ print("Refusing to deactivate all programs; name a target.")
+ return 1
diff --git a/cli/src/castle_cli/commands/service.py b/cli/src/castle_cli/commands/service.py
index 68057e3..86d3bc4 100644
--- a/cli/src/castle_cli/commands/service.py
+++ b/cli/src/castle_cli/commands/service.py
@@ -78,103 +78,21 @@ def run_services(args: argparse.Namespace) -> int:
def _service_enable(config: CastleConfig, name: str) -> int:
"""Enable and start a single service (or timer for scheduled jobs)."""
- if not REGISTRY_PATH.exists():
- print("Error: no registry found. Run 'castle deploy' first.")
- return 1
-
- registry = load_registry()
- if name not in registry.deployed:
- print(f"Error: '{name}' not found in registry. Run 'castle deploy' first.")
- return 1
-
- deployed = registry.deployed[name]
- if not deployed.managed:
- print(f"Error: '{name}' is not a managed service")
- return 1
+ from castle_core.lifecycle import enable_service
ensure_dirs()
-
- # Get systemd spec from config (services or jobs)
- systemd_spec = None
- if name in config.services:
- svc = config.services[name]
- if svc.manage and svc.manage.systemd:
- systemd_spec = svc.manage.systemd
- elif name in config.jobs:
- job = config.jobs[name]
- if job.manage and job.manage.systemd:
- systemd_spec = job.manage.systemd
-
- # Generate and install the service unit from registry
- svc_unit = unit_name(name)
- svc_content = generate_unit_from_deployed(name, deployed, systemd_spec)
- _install_unit(svc_unit, svc_content)
-
- # Check for timer (jobs have schedule)
- if deployed.schedule:
- timer_content = generate_timer(
- name,
- schedule=deployed.schedule,
- description=deployed.description,
- )
- tmr_unit = timer_name(name)
- _install_unit(tmr_unit, timer_content)
-
- print(f"Enabling {name} (scheduled)...")
- subprocess.run(["systemctl", "--user", "enable", tmr_unit], check=False)
- subprocess.run(["systemctl", "--user", "start", tmr_unit], check=False)
-
- result = subprocess.run(
- ["systemctl", "--user", "is-active", tmr_unit],
- capture_output=True,
- text=True,
- )
- status = result.stdout.strip()
- if status in ("active", "waiting"):
- print(f" {name}: timer active")
- else:
- print(f" {name}: timer {status}")
- else:
- print(f"Enabling {name}...")
- subprocess.run(["systemctl", "--user", "enable", svc_unit], check=False)
- subprocess.run(["systemctl", "--user", "start", svc_unit], check=False)
-
- result = subprocess.run(
- ["systemctl", "--user", "is-active", svc_unit],
- capture_output=True,
- text=True,
- )
- status = result.stdout.strip()
- port_str = ""
- if deployed.port:
- port_str = f" (port {deployed.port})"
- if status == "active":
- print(f" {name}: running{port_str}")
- else:
- print(f" {name}: {status}")
- print(f" Check logs: journalctl --user -u {svc_unit}")
-
- return 0
+ result = enable_service(name, config)
+ print(result.output)
+ return 0 if result.status == "ok" else 1
def _service_disable(name: str) -> int:
"""Stop and disable a service (and timer if present)."""
- svc_unit = unit_name(name)
- tmr_unit = timer_name(name)
+ from castle_core.lifecycle import disable_service
print(f"Disabling {name}...")
-
- timer_path = SYSTEMD_USER_DIR / tmr_unit
- if timer_path.exists():
- subprocess.run(["systemctl", "--user", "stop", tmr_unit], check=False)
- subprocess.run(["systemctl", "--user", "disable", tmr_unit], check=False)
- _remove_unit(tmr_unit)
-
- subprocess.run(["systemctl", "--user", "stop", svc_unit], check=False)
- subprocess.run(["systemctl", "--user", "disable", svc_unit], check=False)
- _remove_unit(svc_unit)
- print(f" {name}: disabled")
-
+ result = disable_service(name)
+ print(f" {result.output}")
return 0
diff --git a/core/src/castle_core/deploy.py b/core/src/castle_core/deploy.py
index 9bc1003..7f568bf 100644
--- a/core/src/castle_core/deploy.py
+++ b/core/src/castle_core/deploy.py
@@ -109,6 +109,10 @@ def deploy(target_name: str | None = None, root: Path | None = None) -> DeployRe
_generate_systemd_units(config, registry)
result.messages.append(f"Systemd units written: {SYSTEMD_USER_DIR}")
+ # Converge: prune orphan units (full deploy only — partial deploys preserve siblings)
+ if target_name is None:
+ _prune_orphans(registry, result.messages)
+
# Generate Caddyfile from registry
caddyfile_path = SPECS_DIR / "Caddyfile"
caddyfile_content = generate_caddyfile_from_registry(registry)
@@ -154,11 +158,13 @@ def _build_deployed_service(
prefix = _env_prefix(config_key)
env: dict[str, str] = {}
- # Data dir convention (for managed services)
+ # Data dir convention (for managed services). Skipped for container runners:
+ # they use volume mounts, not a data-dir env var, and the injected name can
+ # collide with an image's own env namespace (e.g. neo4j claims NEO4J_*).
managed = run.runner != "remote"
if svc.manage and svc.manage.systemd and not svc.manage.systemd.enable:
managed = False
- if managed:
+ if managed and run.runner != "container":
env[f"{prefix}_DATA_DIR"] = str(DATA_DIR / config_key)
# Port convention (if exposed)
@@ -180,7 +186,7 @@ def _build_deployed_service(
_ensure_python_tool(config, svc.component, messages)
# Build run_cmd
- run_cmd = _build_run_cmd(run, env, messages)
+ run_cmd = _build_run_cmd(name, run, env, messages)
# Proxy path
proxy_path = None
@@ -228,7 +234,7 @@ def _build_deployed_job(
env = resolve_env_vars(env)
_ensure_python_tool(config, job.component, messages)
- run_cmd = _build_run_cmd(run, env, messages)
+ run_cmd = _build_run_cmd(name, run, env, messages)
stack = None
if job.component and job.component in config.programs:
@@ -297,7 +303,7 @@ def _ensure_python_tool(
messages.append(f"Installed {component}")
-def _build_run_cmd(run: object, env: dict[str, str], messages: list[str]) -> list[str]:
+def _build_run_cmd(name: str, run: object, env: dict[str, str], messages: list[str]) -> list[str]:
"""Build a run command list from a RunSpec."""
match run.runner: # type: ignore[union-attr]
case "python":
@@ -319,8 +325,9 @@ def _build_run_cmd(run: object, env: dict[str, str], messages: list[str]) -> lis
return cmd
case "container":
runtime = shutil.which("docker") or shutil.which("podman") or "docker"
- image_name = run.image.split("/")[-1].split(":")[0] # type: ignore[union-attr]
- cmd = [runtime, "run", "--rm", f"--name=castle-{image_name}"]
+ # Container name derives from the SERVICE name (matches the systemd unit),
+ # not the image name — so `castle-` is stable and collision-free.
+ cmd = [runtime, "run", "--rm", f"--name=castle-{name}"]
for container_port, host_port in run.ports.items(): # type: ignore[union-attr]
cmd.extend(["-p", f"{host_port}:{container_port}"])
for vol in run.volumes: # type: ignore[union-attr]
@@ -378,6 +385,46 @@ def _copy_app_static(config: CastleConfig, messages: list[str]) -> None:
messages.append(f"Static: {src} → {dest}")
+def _desired_unit_files(registry: NodeRegistry) -> set[str]:
+ """Exact set of unit filenames that should exist on disk for this registry."""
+ files: set[str] = set()
+ for name, deployed in registry.deployed.items():
+ if not deployed.managed:
+ continue
+ files.add(unit_name(name))
+ if deployed.schedule:
+ files.add(timer_name(name))
+ return files
+
+
+def _teardown_unit(unit_file: str, messages: list[str]) -> None:
+ """Stop, disable, and unlink a systemd unit file. Caller batches daemon-reload."""
+ path = SYSTEMD_USER_DIR / unit_file
+ if not path.exists():
+ return
+ subprocess.run(["systemctl", "--user", "stop", unit_file], check=False)
+ subprocess.run(["systemctl", "--user", "disable", unit_file], check=False)
+ path.unlink()
+ messages.append(f"Pruned orphan unit: {unit_file}")
+
+
+def _prune_orphans(registry: NodeRegistry, messages: list[str]) -> None:
+ """Remove castle-* units no longer backed by a managed registry entry.
+
+ The `castle-` prefix is the ownership namespace: any castle-*.service/.timer on
+ disk that isn't in the desired set is an orphan (a removed/unmanaged/unscheduled
+ component) and is torn down. Only call on a FULL deploy — the desired set must
+ reflect the whole registry, not a single --target.
+ """
+ desired = _desired_unit_files(registry)
+ if not SYSTEMD_USER_DIR.is_dir():
+ return
+ for pattern in ("castle-*.service", "castle-*.timer"):
+ for path in sorted(SYSTEMD_USER_DIR.glob(pattern)):
+ if path.name not in desired:
+ _teardown_unit(path.name, messages)
+
+
def _generate_systemd_units(config: CastleConfig, registry: NodeRegistry) -> None:
"""Generate systemd units from the registry."""
SYSTEMD_USER_DIR.mkdir(parents=True, exist_ok=True)
diff --git a/core/src/castle_core/lifecycle.py b/core/src/castle_core/lifecycle.py
new file mode 100644
index 0000000..a581ebd
--- /dev/null
+++ b/core/src/castle_core/lifecycle.py
@@ -0,0 +1,166 @@
+"""Unified program lifecycle — a program is `active` when it's reachable in its mode.
+
+`activate`/`deactivate` dispatch the right mechanism by behavior:
+ - tool → on PATH (uv tool install / uninstall)
+ - daemon / self-serving frontend / job → systemd enable+start / stop+disable
+ - static frontend → served via the gateway (build + content present)
+
+`is_active` reports the uniform state regardless of mechanism. The CLI/API verbs
+`install`/`uninstall` route through here (the words stay; the meaning is activate).
+"""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+from pathlib import Path
+
+from castle_core.config import CONTENT_DIR, CastleConfig
+from castle_core.generators.systemd import (
+ SYSTEMD_USER_DIR,
+ generate_timer,
+ generate_unit_from_deployed,
+ timer_name,
+ unit_name,
+)
+from castle_core.registry import REGISTRY_PATH, load_registry
+from castle_core.stacks import ActionResult, run_action
+
+
+def _systemctl_active(unit: str) -> bool:
+ result = subprocess.run(
+ ["systemctl", "--user", "is-active", unit], capture_output=True, text=True
+ )
+ return result.stdout.strip() in ("active", "waiting")
+
+
+def _on_path(name: str) -> bool:
+ """Whether a tool's console script is installed (PATH-independent).
+
+ uv tool install places scripts in ~/.local/bin; checking it directly avoids
+ depending on the caller's PATH (the API/CLI may not have it exported).
+ """
+ if shutil.which(name) is not None:
+ return True
+ return (Path.home() / ".local" / "bin" / name).exists()
+
+
+def _is_static_frontend(name: str, config: CastleConfig) -> bool:
+ """A frontend with no service/job of its own — served as static assets."""
+ comp = config.programs.get(name)
+ return (
+ comp is not None
+ and comp.behavior == "frontend"
+ and name not in config.services
+ and name not in config.jobs
+ )
+
+
+# ---------------------------------------------------------------------------
+# State
+# ---------------------------------------------------------------------------
+
+
+def is_active(name: str, config: CastleConfig) -> bool:
+ """Whether a program is reachable in its mode (uniform across behaviors)."""
+ if name in config.services:
+ return _systemctl_active(unit_name(name))
+ if name in config.jobs:
+ return _systemctl_active(timer_name(name))
+ if _is_static_frontend(name, config):
+ return (CONTENT_DIR / name).is_dir()
+ comp = config.programs.get(name)
+ if comp is not None and comp.source:
+ return _on_path(name)
+ return False
+
+
+# ---------------------------------------------------------------------------
+# Systemd enable/disable (extracted core; the CLI service command calls these)
+# ---------------------------------------------------------------------------
+
+
+def enable_service(name: str, config: CastleConfig) -> ActionResult:
+ """Generate+install the unit (and timer) from the registry, enable and start it."""
+ if not REGISTRY_PATH.exists():
+ return ActionResult(name, "activate", "error", "No registry. Run 'castle deploy' first.")
+ registry = load_registry()
+ if name not in registry.deployed:
+ return ActionResult(name, "activate", "error", f"'{name}' not in registry; run 'castle deploy'.")
+ deployed = registry.deployed[name]
+ if not deployed.managed:
+ return ActionResult(name, "activate", "error", f"'{name}' is not a managed service.")
+
+ systemd_spec = None
+ if name in config.services and config.services[name].manage:
+ systemd_spec = config.services[name].manage.systemd
+ elif name in config.jobs and config.jobs[name].manage:
+ systemd_spec = config.jobs[name].manage.systemd
+
+ SYSTEMD_USER_DIR.mkdir(parents=True, exist_ok=True)
+ svc_unit = unit_name(name)
+ (SYSTEMD_USER_DIR / svc_unit).write_text(generate_unit_from_deployed(name, deployed, systemd_spec))
+ primary = svc_unit
+ if deployed.schedule:
+ tmr_unit = timer_name(name)
+ (SYSTEMD_USER_DIR / tmr_unit).write_text(
+ generate_timer(name, schedule=deployed.schedule, description=deployed.description)
+ )
+ primary = tmr_unit
+
+ subprocess.run(["systemctl", "--user", "daemon-reload"], check=False)
+ subprocess.run(["systemctl", "--user", "enable", primary], check=False)
+ subprocess.run(["systemctl", "--user", "start", primary], check=False)
+ status = "active" if _systemctl_active(primary) else "inactive"
+ return ActionResult(name, "activate", "ok" if status == "active" else "error",
+ f"{name}: {status}")
+
+
+def disable_service(name: str) -> ActionResult:
+ """Stop, disable, and remove the unit (and timer) for a service/job."""
+ for unit in (timer_name(name), unit_name(name)):
+ path = SYSTEMD_USER_DIR / unit
+ if path.exists():
+ subprocess.run(["systemctl", "--user", "stop", unit], check=False)
+ subprocess.run(["systemctl", "--user", "disable", unit], check=False)
+ path.unlink()
+ subprocess.run(["systemctl", "--user", "daemon-reload"], check=False)
+ return ActionResult(name, "deactivate", "ok", f"{name}: deactivated")
+
+
+# ---------------------------------------------------------------------------
+# Activate / deactivate (dispatch by behavior)
+# ---------------------------------------------------------------------------
+
+
+async def activate(name: str, config: CastleConfig, root: Path) -> ActionResult:
+ """Make a program reachable in its mode."""
+ comp = config.programs.get(name)
+
+ # Process-backed: daemon, self-serving frontend, or job.
+ if name in config.services or name in config.jobs:
+ # Ensure the program's binary is on PATH first (python programs).
+ if comp is not None and (comp.stack or comp.commands):
+ res = await run_action("install", name, comp, root)
+ if res.status != "ok":
+ return res
+ return enable_service(name, config)
+
+ # Static frontend: build the assets (publish handled by the build/serve path).
+ if comp is not None and comp.behavior == "frontend":
+ return await run_action("install", name, comp, root)
+
+ # Tool: install to PATH.
+ if comp is not None:
+ return await run_action("install", name, comp, root)
+ return ActionResult(name, "activate", "error", f"'{name}' not found")
+
+
+async def deactivate(name: str, config: CastleConfig, root: Path) -> ActionResult:
+ """Take a program offline in its mode."""
+ comp = config.programs.get(name)
+ if name in config.services or name in config.jobs:
+ return disable_service(name)
+ if comp is not None:
+ return await run_action("uninstall", name, comp, root)
+ return ActionResult(name, "deactivate", "error", f"'{name}' not found")
diff --git a/core/tests/test_deploy_prune.py b/core/tests/test_deploy_prune.py
new file mode 100644
index 0000000..a0414dc
--- /dev/null
+++ b/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()
diff --git a/core/tests/test_lifecycle.py b/core/tests/test_lifecycle.py
new file mode 100644
index 0000000..69a4af2
--- /dev/null
+++ b/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