Paul Payne
8be129259c
Move all instance state into ~/.castle/ with clear separation: - code/ — user project source (was components/) - artifacts/ — generated specs, built content (was generated/, static/) - data/ — service runtime data (was /data/castle/) - secrets/ — credentials castle.yaml moves to ~/.castle/castle.yaml as the canonical location. Source paths use repo: prefix for git repo programs (castle-api, app) and relative paths for user projects (code/central-context). Add idempotent install.sh for bootstrapping infrastructure (Docker, Caddy, MQTT, Postgres, Neo4j) with interactive service detection. Remove components/ from repo (now in ~/.castle/code/), deinit submodules, remove .gitmodules.
307 lines
9.8 KiB
Python
307 lines
9.8 KiB
Python
"""Castle configuration and registry management."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from castle_core.manifest import ProgramSpec, JobSpec, ServiceSpec
|
|
|
|
|
|
CASTLE_HOME = Path.home() / ".castle"
|
|
CODE_DIR = CASTLE_HOME / "code"
|
|
ARTIFACTS_DIR = CASTLE_HOME / "artifacts"
|
|
SPECS_DIR = ARTIFACTS_DIR / "specs"
|
|
CONTENT_DIR = ARTIFACTS_DIR / "content"
|
|
DATA_DIR = CASTLE_HOME / "data"
|
|
SECRETS_DIR = CASTLE_HOME / "secrets"
|
|
|
|
# Backwards-compat aliases (used by existing imports)
|
|
GENERATED_DIR = SPECS_DIR
|
|
STATIC_DIR = CONTENT_DIR
|
|
|
|
|
|
def find_castle_root() -> Path:
|
|
"""Find the castle config root (directory containing castle.yaml).
|
|
|
|
Search order:
|
|
1. ~/.castle/castle.yaml (the canonical instance location)
|
|
2. Walk up from cwd (for development/testing)
|
|
"""
|
|
# Canonical location first
|
|
if (CASTLE_HOME / "castle.yaml").exists():
|
|
return CASTLE_HOME
|
|
# Fallback: walk up from cwd
|
|
current = Path.cwd()
|
|
while current != current.parent:
|
|
if (current / "castle.yaml").exists():
|
|
return current
|
|
current = current.parent
|
|
raise FileNotFoundError(
|
|
"Could not find castle.yaml.\n"
|
|
f"Expected at: {CASTLE_HOME / 'castle.yaml'}"
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class GatewayConfig:
|
|
"""Gateway configuration."""
|
|
|
|
port: int = 9000
|
|
|
|
|
|
@dataclass
|
|
class CastleConfig:
|
|
"""Full castle configuration."""
|
|
|
|
root: Path
|
|
gateway: GatewayConfig
|
|
repo: Path | None
|
|
programs: dict[str, ProgramSpec]
|
|
services: dict[str, ServiceSpec]
|
|
jobs: dict[str, JobSpec]
|
|
|
|
@property
|
|
def tools(self) -> dict[str, ProgramSpec]:
|
|
"""Return programs that are tools (behavior == 'tool')."""
|
|
return {
|
|
k: v
|
|
for k, v in self.programs.items()
|
|
if v.behavior == "tool"
|
|
}
|
|
|
|
@property
|
|
def frontends(self) -> dict[str, ProgramSpec]:
|
|
"""Return programs that are frontends (have build outputs)."""
|
|
return {
|
|
k: v
|
|
for k, v in self.programs.items()
|
|
if v.build and (v.build.outputs or v.build.commands)
|
|
}
|
|
|
|
|
|
def resolve_env_vars(env: dict[str, str]) -> dict[str, str]:
|
|
"""Resolve ${secret:NAME} references in env values."""
|
|
resolved = {}
|
|
for key, value in env.items():
|
|
|
|
def replace_var(match: re.Match[str]) -> str:
|
|
ref = match.group(1)
|
|
if ref.startswith("secret:"):
|
|
secret_name = ref[7:]
|
|
return _read_secret(secret_name)
|
|
return match.group(0)
|
|
|
|
resolved[key] = re.sub(r"\$\{([^}]+)\}", replace_var, value)
|
|
return resolved
|
|
|
|
|
|
def _read_secret(name: str) -> str:
|
|
"""Read a secret from ~/.castle/secrets/<name>. Returns placeholder if not found."""
|
|
secret_path = SECRETS_DIR / name
|
|
if secret_path.exists():
|
|
return secret_path.read_text().strip()
|
|
return f"<MISSING_SECRET:{name}>"
|
|
|
|
|
|
def _parse_program(name: str, data: dict) -> ProgramSpec:
|
|
"""Parse a programs: entry into a ProgramSpec."""
|
|
data_copy = dict(data)
|
|
data_copy["id"] = name
|
|
return ProgramSpec.model_validate(data_copy)
|
|
|
|
|
|
def _parse_service(name: str, data: dict) -> ServiceSpec:
|
|
"""Parse a services: entry into a ServiceSpec."""
|
|
data_copy = dict(data)
|
|
data_copy["id"] = name
|
|
return ServiceSpec.model_validate(data_copy)
|
|
|
|
|
|
def _parse_job(name: str, data: dict) -> JobSpec:
|
|
"""Parse a jobs: entry into a JobSpec."""
|
|
data_copy = dict(data)
|
|
data_copy["id"] = name
|
|
return JobSpec.model_validate(data_copy)
|
|
|
|
|
|
def load_config(root: Path | None = None) -> CastleConfig:
|
|
"""Load castle.yaml and return parsed configuration."""
|
|
if root is None:
|
|
root = find_castle_root()
|
|
|
|
config_path = root / "castle.yaml"
|
|
if not config_path.exists():
|
|
raise FileNotFoundError(f"Castle config not found: {config_path}")
|
|
|
|
with open(config_path) as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
gateway_data = data.get("gateway", {})
|
|
gateway = GatewayConfig(port=gateway_data.get("port", 9000))
|
|
|
|
# repo: field points to the git repo for repo-relative sources
|
|
repo_path: Path | None = None
|
|
if data.get("repo"):
|
|
repo_path = Path(data["repo"]).expanduser()
|
|
|
|
programs: dict[str, ProgramSpec] = {}
|
|
# Support both "programs:" and legacy "components:" key
|
|
programs_data = data.get("programs") or data.get("components") or {}
|
|
for name, comp_data in programs_data.items():
|
|
prog = _parse_program(name, comp_data)
|
|
# Resolve source paths to absolute
|
|
if prog.source:
|
|
if prog.source.startswith("repo:") and repo_path:
|
|
# repo:castle-api → /data/repos/castle/castle-api
|
|
prog.source = str(repo_path / prog.source[5:])
|
|
elif not Path(prog.source).is_absolute():
|
|
prog.source = str(root / prog.source)
|
|
programs[name] = prog
|
|
|
|
services: dict[str, ServiceSpec] = {}
|
|
for name, svc_data in data.get("services", {}).items():
|
|
services[name] = _parse_service(name, svc_data)
|
|
|
|
jobs: dict[str, JobSpec] = {}
|
|
for name, job_data in data.get("jobs", {}).items():
|
|
jobs[name] = _parse_job(name, job_data)
|
|
|
|
return CastleConfig(
|
|
root=root,
|
|
repo=repo_path,
|
|
gateway=gateway,
|
|
programs=programs,
|
|
services=services,
|
|
jobs=jobs,
|
|
)
|
|
|
|
|
|
def _clean_for_yaml(data: object, preserve_keys: set[str] | None = None) -> object:
|
|
"""Recursively remove empty lists and non-structural empty dicts."""
|
|
if preserve_keys is None:
|
|
preserve_keys = _STRUCTURAL_KEYS
|
|
if isinstance(data, dict):
|
|
cleaned = {}
|
|
for k, v in data.items():
|
|
v = _clean_for_yaml(v, preserve_keys)
|
|
# Keep structural keys even if empty dict
|
|
if k in preserve_keys and isinstance(v, dict):
|
|
cleaned[k] = v
|
|
continue
|
|
# Skip empty collections
|
|
if isinstance(v, (list, dict)) and not v:
|
|
continue
|
|
cleaned[k] = v
|
|
return cleaned
|
|
elif isinstance(data, list):
|
|
return [_clean_for_yaml(item, preserve_keys) for item in data]
|
|
return data
|
|
|
|
|
|
# Keys whose presence is structurally significant even with all-default values.
|
|
# We serialize these as empty dicts `{}` so they survive a roundtrip.
|
|
_STRUCTURAL_KEYS = {
|
|
"manage",
|
|
"systemd",
|
|
"expose",
|
|
"proxy",
|
|
"caddy",
|
|
}
|
|
|
|
|
|
def _spec_to_yaml_dict(spec: ProgramSpec | ServiceSpec | JobSpec) -> dict:
|
|
"""Serialize a spec to a YAML-friendly dict, preserving structural presence."""
|
|
exclude_fields = {"id"}
|
|
full = spec.model_dump(mode="json", exclude_none=True, exclude=exclude_fields)
|
|
minimal = spec.model_dump(
|
|
mode="json", exclude_none=True, exclude=exclude_fields, exclude_defaults=True
|
|
)
|
|
|
|
def merge(full_val: object, min_val: object | None, key: str = "") -> object:
|
|
if isinstance(full_val, dict):
|
|
result = {}
|
|
for k, fv in full_val.items():
|
|
mv = min_val.get(k) if isinstance(min_val, dict) else None
|
|
if k in _STRUCTURAL_KEYS:
|
|
merged = merge(fv, mv, k)
|
|
if merged is not None:
|
|
result[k] = merged
|
|
elif mv is not None:
|
|
result[k] = merge(fv, mv, k)
|
|
elif isinstance(fv, dict):
|
|
merged = merge(fv, None, k)
|
|
if merged:
|
|
result[k] = merged
|
|
return result if result else ({} if key in _STRUCTURAL_KEYS else result)
|
|
elif isinstance(full_val, list):
|
|
if min_val is not None:
|
|
return full_val
|
|
return []
|
|
else:
|
|
if min_val is not None:
|
|
return full_val
|
|
return None
|
|
|
|
result = merge(full, minimal)
|
|
return _clean_for_yaml(result)
|
|
|
|
|
|
def save_config(config: CastleConfig) -> None:
|
|
"""Save castle configuration to castle.yaml."""
|
|
data: dict = {"gateway": {"port": config.gateway.port}}
|
|
|
|
if config.repo:
|
|
data["repo"] = str(config.repo)
|
|
|
|
if config.programs:
|
|
data["programs"] = {}
|
|
for name, spec in config.programs.items():
|
|
d = _spec_to_yaml_dict(spec)
|
|
# Store relative source paths in YAML
|
|
if d.get("source") and Path(d["source"]).is_absolute():
|
|
src = Path(d["source"])
|
|
# If source is under repo, store as repo:relative
|
|
if config.repo:
|
|
try:
|
|
d["source"] = "repo:" + str(src.relative_to(config.repo))
|
|
data["programs"][name] = d
|
|
continue
|
|
except ValueError:
|
|
pass
|
|
# Otherwise store relative to config root
|
|
try:
|
|
d["source"] = str(src.relative_to(config.root))
|
|
except ValueError:
|
|
pass # not under root — keep absolute
|
|
data["programs"][name] = d
|
|
|
|
if config.services:
|
|
data["services"] = {}
|
|
for name, spec in config.services.items():
|
|
data["services"][name] = _spec_to_yaml_dict(spec)
|
|
|
|
if config.jobs:
|
|
data["jobs"] = {}
|
|
for name, spec in config.jobs.items():
|
|
data["jobs"][name] = _spec_to_yaml_dict(spec)
|
|
|
|
config_path = config.root / "castle.yaml"
|
|
with open(config_path, "w") as f:
|
|
yaml.dump(data, f, default_flow_style=False, sort_keys=False)
|
|
|
|
|
|
def ensure_dirs() -> None:
|
|
"""Ensure castle directories exist."""
|
|
CASTLE_HOME.mkdir(parents=True, exist_ok=True)
|
|
CODE_DIR.mkdir(parents=True, exist_ok=True)
|
|
SPECS_DIR.mkdir(parents=True, exist_ok=True)
|
|
CONTENT_DIR.mkdir(parents=True, exist_ok=True)
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
SECRETS_DIR.mkdir(parents=True, exist_ok=True)
|
|
os.chmod(SECRETS_DIR, 0o700)
|