wild-cloud/wild-pc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Extract deploy logic to castle-core and add deploy API endpoint

Browse Source 
- Extract all deploy logic from CLI into castle-core/deploy.py as a reusable deploy() function
  * Resolves services/jobs to DeployedComponents
  * Generates systemd units and Caddyfile
  * Copies frontend assets and reloads systemd
  * Returns DeployResult with deployed count and messages
- Simplify CLI deploy command to a thin wrapper calling castle_core.deploy()
- Add POST /deploy endpoint to castle-api for remote deployment management
  * Supports optional {name} body param to deploy one or all services
  * Returns deployment result as JSON
- Register deploy router in castle-api main.py
- Update README with new deploy endpoint documentation

This enables full remote management: build, test, lint, and deploy any castle
node over HTTP (e.g. POST http://node:9000/api/deploy)
This commit is contained in:
Paul Payne
2026-04-27 22:01:39 -07:00
parent e1a9386177
commit 691216b04e
7 changed files with 539 additions and 387 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@@ -196,6 +196,8 @@ When enabled, the API publishes the node's registry to `castle/{hostname}/regist
| `GET /gateway/caddyfile` | Generated Caddyfile content |
| `POST /gateway/reload` | Regenerate Caddyfile and reload Caddy |
| `GET /status` | Live health for all services |
| **Deploy** | |
| `POST /deploy` | Deploy all services and jobs (spec to runtime) |
| **Config** | |
| `GET /config` | Read castle.yaml |
| `PUT /config` | Write castle.yaml |

49
castle-api/src/castle_api/deploy_routes.py Normal file
View File

@@ -0,0 +1,49 @@
"""Deploy API endpoint."""
from __future__ import annotations
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from castle_core.deploy import deploy
router = APIRouter(tags=["deploy"])
class DeployRequest(BaseModel):
"""Optional request body for deploy."""
name: str | None = None
class DeployResponse(BaseModel):
"""Response from a deploy operation."""
status: str
deployed_count: int
messages: list[str]
@router.post("/deploy", response_model=DeployResponse)
def run_deploy(request: DeployRequest | None = None) -> DeployResponse:
"""Deploy services and jobs from castle.yaml to runtime.
Resolves env vars and secrets, generates systemd units and Caddyfile,
copies frontend build outputs, and reloads systemd.
Optionally pass a name to deploy a single service or job.
"""
target_name = request.name if request else None
try:
result = deploy(target_name=target_name)
except FileNotFoundError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) from e
return DeployResponse(
status="ok",
deployed_count=result.deployed_count,
messages=result.messages,
)

2
castle-api/src/castle_api/main.py
View File

@@ -14,6 +14,7 @@ from starlette.responses import StreamingResponse
from castle_api.config import get_registry, settings
from castle_api.config_editor import router as config_router
from castle_api.deploy_routes import router as deploy_router
from castle_api.logs import router as logs_router
from castle_api.routes import router as dashboard_router
from castle_api.secrets import router as secrets_router
@@ -121,6 +122,7 @@ app.include_router(nodes_router)
app.include_router(secrets_router)
app.include_router(services_router)
app.include_router(programs_router)
app.include_router(deploy_router)
@app.get("/health")

0
castle-api/src/castle_api/tools.py → castle-api/src/castle_api/programs.py
View File

76
castle.code-workspace Normal file
View File

@@ -0,0 +1,76 @@
{
"folders": [
{
"name": "root",
"path": "."
},
{
"name": "core",
"path": "core"
},
{
"name": "cli",
"path": "cli"
},
{
"name": "castle-api",
"path": "castle-api"
},
{
"name": "app (frontend)",
"path": "app"
},
{
"name": "docs",
"path": "docs"
}
],
"settings": {
"python.defaultInterpreterPath": "${workspaceFolder:root}/.venv/bin/python",
"python.analysis.typeCheckingMode": "standard",
"python.analysis.extraPaths": [
"${workspaceFolder:core}/src",
"${workspaceFolder:cli}/src",
"${workspaceFolder:castle-api}/src"
],
"[python]": {
"editor.defaultFormatter": "charliermarsh.ruff",
"editor.formatOnSave": true,
"editor.codeActionsOnSave": {
"source.organizeImports.ruff": "explicit"
}
},
"ruff.configurationPreference": "filesystemFirst",
"[typescript]": {
"editor.defaultFormatter": "esbenp.prettier-vscode",
"editor.formatOnSave": true
},
"[typescriptreact]": {
"editor.defaultFormatter": "esbenp.prettier-vscode",
"editor.formatOnSave": true
},
"files.exclude": {
"**/__pycache__": true,
"**/.pytest_cache": true,
"**/.ruff_cache": true,
"**/node_modules": true,
"**/*.egg-info": true
},
"search.exclude": {
"**/.venv": true,
"**/node_modules": true,
"**/uv.lock": true,
"**/pnpm-lock.yaml": true,
"**/dist": true
}
},
"extensions": {
"recommendations": [
"ms-python.python",
"charliermarsh.ruff",
"ms-python.vscode-pylance",
"esbenp.prettier-vscode",
"dbaeumer.vscode-eslint"
]
}
}

393
cli/src/castle_cli/commands/deploy.py
View File

@@ -1,400 +1,21 @@
"""castle deploy — bridge spec (castle.yaml) to runtime (~/.castle/)."""
"""castle deploy — thin CLI wrapper around castle_core.deploy."""
from __future__ import annotations
import argparse
import shutil
import subprocess
from pathlib import Path
from castle_core.config import (
CONTENT_DIR,
DATA_DIR,
SPECS_DIR,
CastleConfig,
ensure_dirs,
load_config,
resolve_env_vars,
)
from castle_core.generators.caddyfile import generate_caddyfile_from_registry
from castle_core.generators.systemd import (
generate_timer,
generate_unit_from_deployed,
timer_name,
unit_name,
)
from castle_core.manifest import JobSpec, ServiceSpec
from castle_core.registry import (
REGISTRY_PATH,
DeployedComponent,
NodeConfig,
NodeRegistry,
load_registry,
save_registry,
)
SYSTEMD_USER_DIR = Path.home() / ".config" / "systemd" / "user"
from castle_core.deploy import deploy
def run_deploy(args: argparse.Namespace) -> int:
"""Deploy from castle.yaml to ~/.castle/."""
config = load_config()
target_name = getattr(args, "name", None)
result = deploy(target_name=target_name)
ensure_dirs()
for msg in result.messages:
print(f" {msg}")
# Build node config
node = NodeConfig(castle_root=str(config.root), gateway_port=config.gateway.port)
# Load existing registry to preserve entries not being redeployed,
# or start fresh if deploying all
if target_name and REGISTRY_PATH.exists():
try:
existing = load_registry()
registry = NodeRegistry(node=node, deployed=dict(existing.deployed))
except (FileNotFoundError, ValueError):
registry = NodeRegistry(node=node)
else:
registry = NodeRegistry(node=node)
deployed_count = 0
# Deploy services
for name, svc in config.services.items():
if target_name and name != target_name:
continue
deployed = _build_deployed_service(config, name, svc)
registry.deployed[name] = deployed
deployed_count += 1
_print_deployed(name, deployed)
# Deploy jobs
for name, job in config.jobs.items():
if target_name and name != target_name:
continue
deployed = _build_deployed_job(config, name, job)
registry.deployed[name] = deployed
deployed_count += 1
_print_deployed(name, deployed)
# Handle castle-app build artifacts
_copy_app_static(config)
# Save registry
save_registry(registry)
print(f"\nRegistry written: {REGISTRY_PATH}")
# Generate systemd units from registry
_generate_systemd_units(config, registry)
# Generate Caddyfile from registry
caddyfile_path = SPECS_DIR / "Caddyfile"
caddyfile_content = generate_caddyfile_from_registry(registry)
caddyfile_path.write_text(caddyfile_content)
print(f"Caddyfile written: {caddyfile_path}")
# Reload systemd daemon
subprocess.run(["systemctl", "--user", "daemon-reload"], check=False)
print(f"\nDeployed {deployed_count} item(s).")
print(f"\nDeployed {result.deployed_count} item(s).")
if result.deployed_count > 0:
print("Run 'castle services start' to start all services.")
return 0
def _env_prefix(name: str) -> str:
"""Derive env var prefix from name: central-context → CENTRAL_CONTEXT."""
return name.replace("-", "_").upper()
def _resolve_description(config: CastleConfig, spec: ServiceSpec | JobSpec) -> str | None:
"""Get description, falling through to program if referenced."""
if spec.description:
return spec.description
if spec.component and spec.component in config.programs:
return config.programs[spec.component].description
return None
def _build_deployed_service(config: CastleConfig, name: str, svc: ServiceSpec) -> DeployedComponent:
"""Build a DeployedComponent from a ServiceSpec."""
run = svc.run
prefix = _env_prefix(name)
env: dict[str, str] = {}
# Data dir convention (for managed services)
managed = run.runner != "remote" # Remote services have no local process
if svc.manage and svc.manage.systemd and not svc.manage.systemd.enable:
managed = False
if managed:
env[f"{prefix}_DATA_DIR"] = str(DATA_DIR / name)
# Port convention (if exposed)
port = None
health_path = None
if svc.expose and svc.expose.http:
port = svc.expose.http.internal.port
env[f"{prefix}_PORT"] = str(port)
health_path = svc.expose.http.health_path
# Merge defaults.env (overrides conventions)
if svc.defaults and svc.defaults.env:
env.update(svc.defaults.env)
# Resolve secrets
env = resolve_env_vars(env)
# Ensure python tool is installed before resolving binary
_ensure_python_tool(config, svc.component)
# Build run_cmd
run_cmd = _build_run_cmd(run, env)
# Proxy path
proxy_path = None
if svc.proxy and svc.proxy.caddy and svc.proxy.caddy.enable:
proxy_path = svc.proxy.caddy.path_prefix or f"/{name}"
# Resolve stack from referenced program
stack = None
if svc.component and svc.component in config.programs:
stack = config.programs[svc.component].stack
# Remote services proxy to an external base_url
base_url = getattr(run, "base_url", None)
return DeployedComponent(
runner=run.runner,
run_cmd=run_cmd,
env=env,
description=_resolve_description(config, svc),
behavior="daemon",
stack=stack,
port=port,
health_path=health_path,
proxy_path=proxy_path,
base_url=base_url,
managed=managed,
)
def _build_deployed_job(config: CastleConfig, name: str, job: JobSpec) -> DeployedComponent:
"""Build a DeployedComponent from a JobSpec."""
run = job.run
prefix = _env_prefix(name)
env: dict[str, str] = {}
# Data dir convention
env[f"{prefix}_DATA_DIR"] = str(DATA_DIR / name)
# Merge defaults.env (overrides conventions)
if job.defaults and job.defaults.env:
env.update(job.defaults.env)
# Resolve secrets
env = resolve_env_vars(env)
# Ensure python tool is installed before resolving binary
_ensure_python_tool(config, job.component)
# Build run_cmd
run_cmd = _build_run_cmd(run, env)
# Resolve stack from referenced program
stack = None
if job.component and job.component in config.programs:
stack = config.programs[job.component].stack
return DeployedComponent(
runner=run.runner,
run_cmd=run_cmd,
env=env,
description=_resolve_description(config, job),
behavior="tool",
stack=stack,
schedule=job.schedule,
managed=True,
)
def _python_tool_needs_install(program: str) -> bool:
"""Check if a Python tool's editable install is broken.
Returns True if the binary is missing or the editable install's
.pth file points to a directory that no longer exists.
"""
if not shutil.which(program):
return True
tool_dir = Path.home() / ".local" / "share" / "uv" / "tools" / program
if not tool_dir.exists():
return True
for pth_file in tool_dir.glob("lib/python*/site-packages/*.pth"):
if pth_file.name == "_virtualenv.pth":
continue
try:
target = pth_file.read_text().strip()
except OSError:
continue
if not target or target.startswith("import "):
continue
if not Path(target).exists():
return True
return False
def _ensure_python_tool(config: CastleConfig, component: str | None) -> None:
"""Ensure a Python program's editable install is current.
Checks if the uv tool's .pth file points to a directory that still
exists. If the binary is missing or the editable path is stale,
reinstalls from the source directory defined in the program spec.
"""
if not component or component not in config.programs:
return
comp = config.programs[component]
if not comp.source or not comp.stack or not comp.stack.startswith("python"):
return
source_dir = Path(comp.source)
if not source_dir.is_dir():
print(f" Warning: source not found: {source_dir}")
return
if not _python_tool_needs_install(component):
return
pkg_spec = str(source_dir)
if comp.install_extras:
pkg_spec += "[" + ",".join(comp.install_extras) + "]"
print(f" Installing {component} from {source_dir}...")
result = subprocess.run(
["uv", "tool", "install", "--editable", pkg_spec, "--force"],
capture_output=True,
text=True,
)
if result.returncode != 0:
print(f" Error: {component} install failed:\n{result.stdout}{result.stderr}")
else:
print(f" Installed {component}")
def _build_run_cmd(run: object, env: dict[str, str]) -> list[str]:
"""Build a run command list from a RunSpec."""
match run.runner:
case "python":
resolved = shutil.which(run.program)
if not resolved:
print(
f" Warning: '{run.program}' not on PATH. "
f"Install with: uv tool install --editable <source>"
)
cmd = [resolved or run.program]
if run.args:
cmd.extend(run.args)
return cmd
case "command":
cmd = list(run.argv)
resolved = shutil.which(cmd[0])
if resolved:
cmd[0] = resolved
return cmd
case "container":
runtime = shutil.which("docker") or shutil.which("podman") or "docker"
image_name = run.image.split("/")[-1].split(":")[0]
cmd = [runtime, "run", "--rm", f"--name=castle-{image_name}"]
for container_port, host_port in run.ports.items():
cmd.extend(["-p", f"{host_port}:{container_port}"])
for vol in run.volumes:
cmd.extend(["-v", vol])
for key, val in run.env.items():
cmd.extend(["-e", f"{key}={val}"])
for key, val in env.items():
cmd.extend(["-e", f"{key}={val}"])
if run.workdir:
cmd.extend(["-w", run.workdir])
cmd.append(run.image)
if run.command:
cmd.extend(run.command)
if run.args:
cmd.extend(run.args)
return cmd
case "node":
cmd = [run.package_manager, "run", run.script]
if run.args:
cmd.extend(run.args)
return cmd
case "remote":
return [] # No local process for remote services
case _:
raise ValueError(f"Unsupported runner: {run.runner}")
def _print_deployed(name: str, deployed: DeployedComponent) -> None:
"""Print deployment summary for a component."""
parts = [f" {name}"]
if deployed.port:
parts.append(f"port={deployed.port}")
if deployed.schedule:
parts.append(f"schedule={deployed.schedule}")
if deployed.proxy_path:
parts.append(f"proxy={deployed.proxy_path}")
print(" ".join(parts))
def _copy_app_static(config: CastleConfig) -> None:
"""Copy frontend build outputs to ~/.castle/static/<name>/."""
for name, comp in config.programs.items():
if comp.behavior != "frontend":
continue
if not (comp.build and comp.build.outputs):
continue
source_dir = comp.source_dir or name
for output in comp.build.outputs:
src = config.root / source_dir / output
if src.exists():
dest = CONTENT_DIR / name
if dest.exists():
shutil.rmtree(dest)
shutil.copytree(src, dest)
print(f" Static: {src} → {dest}")
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
# 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 write service unit
svc_name = unit_name(name)
svc_content = generate_unit_from_deployed(name, deployed, systemd_spec)
(SYSTEMD_USER_DIR / svc_name).write_text(svc_content)
# Generate timer for jobs
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)
print(f"Systemd units written: {SYSTEMD_USER_DIR}")

402
core/src/castle_core/deploy.py Normal file
View File

@@ -0,0 +1,402 @@
"""Deploy logic — bridge castle.yaml spec to runtime (~/.castle/).
This module contains the core deploy logic shared by the CLI and API.
It reads castle.yaml, resolves services/jobs into DeployedComponents,
writes the registry, generates systemd units and the Caddyfile, and
copies frontend build outputs.
"""
from __future__ import annotations
import shutil
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from castle_core.config import (
CONTENT_DIR,
DATA_DIR,
SPECS_DIR,
CastleConfig,
ensure_dirs,
load_config,
resolve_env_vars,
)
from castle_core.generators.caddyfile import generate_caddyfile_from_registry
from castle_core.generators.systemd import (
generate_timer,
generate_unit_from_deployed,
timer_name,
unit_name,
)
from castle_core.manifest import JobSpec, ServiceSpec
from castle_core.registry import (
REGISTRY_PATH,
DeployedComponent,
NodeConfig,
NodeRegistry,
load_registry,
save_registry,
)
SYSTEMD_USER_DIR = Path.home() / ".config" / "systemd" / "user"
@dataclass
class DeployResult:
"""Result of a deploy operation."""
deployed_count: int = 0
messages: list[str] = field(default_factory=list)
registry: NodeRegistry | None = None
def deploy(target_name: str | None = None, root: Path | None = None) -> DeployResult:
"""Deploy from castle.yaml to ~/.castle/.
Args:
target_name: Deploy a single service/job by name, or None for all.
root: Config root path. If None, uses find_castle_root().
Returns:
DeployResult with deployed count, messages, and the registry.
"""
config = load_config(root)
result = DeployResult()
ensure_dirs()
# Build node config
node = NodeConfig(castle_root=str(config.root), gateway_port=config.gateway.port)
# Load existing registry to preserve entries not being redeployed,
# or start fresh if deploying all
if target_name and REGISTRY_PATH.exists():
try:
existing = load_registry()
registry = NodeRegistry(node=node, deployed=dict(existing.deployed))
except (FileNotFoundError, ValueError):
registry = NodeRegistry(node=node)
else:
registry = NodeRegistry(node=node)
# Deploy services
for name, svc in config.services.items():
if target_name and name != target_name:
continue
deployed = _build_deployed_service(config, name, svc, result.messages)
registry.deployed[name] = deployed
result.deployed_count += 1
result.messages.append(_format_deployed(name, deployed))
# Deploy jobs
for name, job in config.jobs.items():
if target_name and name != target_name:
continue
deployed = _build_deployed_job(config, name, job, result.messages)
registry.deployed[name] = deployed
result.deployed_count += 1
result.messages.append(_format_deployed(name, deployed))
# Handle frontend build artifacts
_copy_app_static(config, result.messages)
# Save registry
save_registry(registry)
result.messages.append(f"Registry written: {REGISTRY_PATH}")
# Generate systemd units from registry
_generate_systemd_units(config, registry)
result.messages.append(f"Systemd units written: {SYSTEMD_USER_DIR}")
# Generate Caddyfile from registry
caddyfile_path = SPECS_DIR / "Caddyfile"
caddyfile_content = generate_caddyfile_from_registry(registry)
caddyfile_path.write_text(caddyfile_content)
result.messages.append(f"Caddyfile written: {caddyfile_path}")
# Reload systemd daemon
subprocess.run(["systemctl", "--user", "daemon-reload"], check=False)
result.registry = registry
return result
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _env_prefix(name: str) -> str:
"""Derive env var prefix from name: central-context → CENTRAL_CONTEXT."""
return name.replace("-", "_").upper()
def _resolve_description(config: CastleConfig, spec: ServiceSpec | JobSpec) -> str | None:
"""Get description, falling through to program if referenced."""
if spec.description:
return spec.description
if spec.component and spec.component in config.programs:
return config.programs[spec.component].description
return None
def _build_deployed_service(
config: CastleConfig, name: str, svc: ServiceSpec, messages: list[str]
) -> DeployedComponent:
"""Build a DeployedComponent from a ServiceSpec."""
run = svc.run
prefix = _env_prefix(name)
env: dict[str, str] = {}
# Data dir convention (for managed services)
managed = run.runner != "remote"
if svc.manage and svc.manage.systemd and not svc.manage.systemd.enable:
managed = False
if managed:
env[f"{prefix}_DATA_DIR"] = str(DATA_DIR / name)
# Port convention (if exposed)
port = None
health_path = None
if svc.expose and svc.expose.http:
port = svc.expose.http.internal.port
env[f"{prefix}_PORT"] = str(port)
health_path = svc.expose.http.health_path
# Merge defaults.env (overrides conventions)
if svc.defaults and svc.defaults.env:
env.update(svc.defaults.env)
# Resolve secrets
env = resolve_env_vars(env)
# Ensure python tool is installed before resolving binary
_ensure_python_tool(config, svc.component, messages)
# Build run_cmd
run_cmd = _build_run_cmd(run, env, messages)
# Proxy path
proxy_path = None
if svc.proxy and svc.proxy.caddy and svc.proxy.caddy.enable:
proxy_path = svc.proxy.caddy.path_prefix or f"/{name}"
# Resolve stack from referenced program
stack = None
if svc.component and svc.component in config.programs:
stack = config.programs[svc.component].stack
# Remote services proxy to an external base_url
base_url = getattr(run, "base_url", None)
return DeployedComponent(
runner=run.runner,
run_cmd=run_cmd,
env=env,
description=_resolve_description(config, svc),
behavior="daemon",
stack=stack,
port=port,
health_path=health_path,
proxy_path=proxy_path,
base_url=base_url,
managed=managed,
)
def _build_deployed_job(
config: CastleConfig, name: str, job: JobSpec, messages: list[str]
) -> DeployedComponent:
"""Build a DeployedComponent from a JobSpec."""
run = job.run
prefix = _env_prefix(name)
env: dict[str, str] = {}
env[f"{prefix}_DATA_DIR"] = str(DATA_DIR / name)
if job.defaults and job.defaults.env:
env.update(job.defaults.env)
env = resolve_env_vars(env)
_ensure_python_tool(config, job.component, messages)
run_cmd = _build_run_cmd(run, env, messages)
stack = None
if job.component and job.component in config.programs:
stack = config.programs[job.component].stack
return DeployedComponent(
runner=run.runner,
run_cmd=run_cmd,
env=env,
description=_resolve_description(config, job),
behavior="tool",
stack=stack,
schedule=job.schedule,
managed=True,
)
def _python_tool_needs_install(program: str) -> bool:
"""Check if a Python tool's editable install is broken."""
if not shutil.which(program):
return True
tool_dir = Path.home() / ".local" / "share" / "uv" / "tools" / program
if not tool_dir.exists():
return True
for pth_file in tool_dir.glob("lib/python*/site-packages/*.pth"):
if pth_file.name == "_virtualenv.pth":
continue
try:
target = pth_file.read_text().strip()
except OSError:
continue
if not target or target.startswith("import "):
continue
if not Path(target).exists():
return True
return False
def _ensure_python_tool(
config: CastleConfig, component: str | None, messages: list[str]
) -> None:
"""Ensure a Python program's editable install is current."""
if not component or component not in config.programs:
return
comp = config.programs[component]
if not comp.source or not comp.stack or not comp.stack.startswith("python"):
return
source_dir = Path(comp.source)
if not source_dir.is_dir():
messages.append(f"Warning: source not found: {source_dir}")
return
if not _python_tool_needs_install(component):
return
pkg_spec = str(source_dir)
if comp.install_extras:
pkg_spec += "[" + ",".join(comp.install_extras) + "]"
messages.append(f"Installing {component} from {source_dir}...")
result = subprocess.run(
["uv", "tool", "install", "--editable", pkg_spec, "--force"],
capture_output=True,
text=True,
)
if result.returncode != 0:
messages.append(f"Error: {component} install failed:\n{result.stdout}{result.stderr}")
else:
messages.append(f"Installed {component}")
def _build_run_cmd(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":
resolved = shutil.which(run.program) # type: ignore[union-attr]
if not resolved:
messages.append(
f"Warning: '{run.program}' not on PATH. " # type: ignore[union-attr]
f"Install with: uv tool install --editable <source>"
)
cmd = [resolved or run.program] # type: ignore[union-attr]
if run.args: # type: ignore[union-attr]
cmd.extend(run.args) # type: ignore[union-attr]
return cmd
case "command":
cmd = list(run.argv) # type: ignore[union-attr]
resolved = shutil.which(cmd[0])
if resolved:
cmd[0] = resolved
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}"]
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]
cmd.extend(["-v", vol])
for key, val in run.env.items(): # type: ignore[union-attr]
cmd.extend(["-e", f"{key}={val}"])
for key, val in env.items():
cmd.extend(["-e", f"{key}={val}"])
if run.workdir: # type: ignore[union-attr]
cmd.extend(["-w", run.workdir]) # type: ignore[union-attr]
cmd.append(run.image) # type: ignore[union-attr]
if run.command: # type: ignore[union-attr]
cmd.extend(run.command) # type: ignore[union-attr]
if run.args: # type: ignore[union-attr]
cmd.extend(run.args) # type: ignore[union-attr]
return cmd
case "node":
cmd = [run.package_manager, "run", run.script] # type: ignore[union-attr]
if run.args: # type: ignore[union-attr]
cmd.extend(run.args) # type: ignore[union-attr]
return cmd
case "remote":
return []
case _:
raise ValueError(f"Unsupported runner: {run.runner}") # type: ignore[union-attr]
def _format_deployed(name: str, deployed: DeployedComponent) -> str:
"""Format deployment summary for a component."""
parts = [name]
if deployed.port:
parts.append(f"port={deployed.port}")
if deployed.schedule:
parts.append(f"schedule={deployed.schedule}")
if deployed.proxy_path:
parts.append(f"proxy={deployed.proxy_path}")
return " ".join(parts)
def _copy_app_static(config: CastleConfig, messages: list[str]) -> None:
"""Copy frontend build outputs to ~/.castle/artifacts/content/<name>/."""
for name, comp in config.programs.items():
if comp.behavior != "frontend":
continue
if not (comp.build and comp.build.outputs):
continue
source_dir = comp.source_dir or name
for output in comp.build.outputs:
src = config.root / source_dir / output
if src.exists():
dest = CONTENT_DIR / name
if dest.exists():
shutil.rmtree(dest)
shutil.copytree(src, dest)
messages.append(f"Static: {src} → {dest}")
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
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
svc_name = unit_name(name)
svc_content = generate_unit_from_deployed(name, deployed, systemd_spec)
(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)