feat: add ComponentGrid and ComponentTable components for displaying components in grid and table formats

feat: implement HealthBadge and RoleBadge components for displaying health status and roles

feat: create LogViewer component for streaming logs of services

feat: develop SecretsEditor for managing secrets with CRUD operations

feat: introduce ToolCard component for displaying tool information

style: add global CSS variables and styles for consistent theming

feat: set up API client for handling requests to the backend

feat: implement hooks for fetching components, statuses, and tools

feat: create routes for dashboard, component details, and tools

feat: add service management endpoints for starting, stopping, and restarting services

feat: implement event bus for handling real-time updates via SSE

feat: create health check and logs endpoints for monitoring services

feat: add tests for health and tools endpoints to ensure functionality

chore: update project configuration files and dependencies for better development experience
This commit is contained in:
2026-02-21 01:23:37 -08:00
parent f99c2dad86
commit 930bc601b7
63 changed files with 154 additions and 157 deletions

View File

@@ -0,0 +1,3 @@
"""Castle API."""
__version__ = "0.1.0"

View File

@@ -0,0 +1,121 @@
"""Event bus core — in-memory subscription table and HTTP fan-out delivery."""
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
import httpx
logger = logging.getLogger(__name__)
@dataclass
class Subscription:
"""A subscription to a topic."""
topic: str
callback_url: str
subscriber: str = "" # optional label for debugging
@dataclass
class EventBus:
"""In-memory event bus with HTTP fan-out delivery."""
subscriptions: dict[str, list[Subscription]] = field(default_factory=dict)
_client: httpx.AsyncClient | None = field(default=None, repr=False)
async def start(self) -> None:
"""Initialize the HTTP client."""
self._client = httpx.AsyncClient(timeout=10.0)
async def stop(self) -> None:
"""Close the HTTP client."""
if self._client:
await self._client.aclose()
self._client = None
def subscribe(self, topic: str, callback_url: str, subscriber: str = "") -> None:
"""Register a subscription to a topic."""
if topic not in self.subscriptions:
self.subscriptions[topic] = []
# Don't duplicate
for sub in self.subscriptions[topic]:
if sub.callback_url == callback_url:
return
self.subscriptions[topic].append(
Subscription(topic=topic, callback_url=callback_url, subscriber=subscriber)
)
logger.info("Subscribed %s to topic '%s' -> %s", subscriber, topic, callback_url)
def unsubscribe(self, topic: str, callback_url: str) -> bool:
"""Remove a subscription. Returns True if found and removed."""
if topic not in self.subscriptions:
return False
before = len(self.subscriptions[topic])
self.subscriptions[topic] = [
s for s in self.subscriptions[topic] if s.callback_url != callback_url
]
removed = len(self.subscriptions[topic]) < before
if not self.subscriptions[topic]:
del self.subscriptions[topic]
return removed
async def publish(self, topic: str, payload: dict) -> int:
"""Publish an event to all subscribers. Returns number of subscribers notified."""
subscribers = self.subscriptions.get(topic, [])
if not subscribers:
return 0
event = {
"topic": topic,
"payload": payload,
"published_at": datetime.now(timezone.utc).isoformat(),
}
# Fan out to all subscribers concurrently, fire-and-forget style
tasks = [self._deliver(sub, event) for sub in subscribers]
results = await asyncio.gather(*tasks, return_exceptions=True)
delivered = sum(1 for r in results if r is True)
return delivered
async def _deliver(self, sub: Subscription, event: dict) -> bool:
"""Deliver an event to a single subscriber."""
if not self._client:
logger.error("HTTP client not initialized")
return False
try:
response = await self._client.post(sub.callback_url, json=event)
if response.status_code < 300:
return True
logger.warning(
"Delivery to %s returned %d", sub.callback_url, response.status_code
)
return False
except Exception:
logger.warning("Delivery to %s failed", sub.callback_url, exc_info=True)
return False
def list_topics(self) -> dict[str, list[dict]]:
"""List all topics and their subscribers."""
return {
topic: [
{"callback_url": s.callback_url, "subscriber": s.subscriber}
for s in subs
]
for topic, subs in self.subscriptions.items()
}
# Singleton instance
bus = EventBus()

View File

@@ -0,0 +1,21 @@
"""Configuration for castle-api."""
from pathlib import Path
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
"""Service settings loaded from environment variables."""
castle_root: Path = Path("/data/repos/castle")
host: str = "0.0.0.0"
port: int = 9020
model_config = {
"env_prefix": "CASTLE_API_",
"env_file": ".env",
}
settings = Settings()

View File

@@ -0,0 +1,189 @@
"""Config editor — read, validate, save, and apply castle.yaml changes."""
from __future__ import annotations
import asyncio
import shutil
import yaml
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel
from castle_cli.config import load_config, save_config
from castle_cli.manifest import ComponentManifest
from castle_api.config import settings
from castle_api.stream import broadcast
router = APIRouter(prefix="/config", tags=["config"])
class ConfigResponse(BaseModel):
yaml_content: str
class ConfigSaveRequest(BaseModel):
yaml_content: str
class ConfigSaveResponse(BaseModel):
ok: bool
component_count: int
errors: list[str]
class ApplyResponse(BaseModel):
ok: bool
actions: list[str]
errors: list[str]
class ComponentConfigRequest(BaseModel):
config: dict
@router.get("", response_model=ConfigResponse)
def get_config() -> ConfigResponse:
"""Get the raw castle.yaml content."""
config_path = settings.castle_root / "castle.yaml"
return ConfigResponse(yaml_content=config_path.read_text())
@router.put("", response_model=ConfigSaveResponse)
def save_yaml(request: ConfigSaveRequest) -> ConfigSaveResponse:
"""Validate and save castle.yaml. Does NOT apply changes."""
errors: list[str] = []
# Parse YAML
try:
data = yaml.safe_load(request.yaml_content)
except yaml.YAMLError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid YAML: {e}",
)
if not isinstance(data, dict) or "components" not in data:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="YAML must have a 'components' key",
)
# Validate each component
count = 0
for name, comp_data in data.get("components", {}).items():
try:
comp_data_copy = dict(comp_data) if comp_data else {}
comp_data_copy["id"] = name
ComponentManifest.model_validate(comp_data_copy)
count += 1
except Exception as e:
errors.append(f"{name}: {e}")
if errors:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail={"errors": errors},
)
# Backup and save
config_path = settings.castle_root / "castle.yaml"
backup_path = config_path.with_suffix(".yaml.bak")
shutil.copy2(config_path, backup_path)
config_path.write_text(request.yaml_content)
return ConfigSaveResponse(ok=True, component_count=count, errors=[])
@router.put("/components/{name}")
def save_component(name: str, request: ComponentConfigRequest) -> dict:
"""Update a single component's config in castle.yaml."""
# Validate
try:
comp_data = dict(request.config)
comp_data["id"] = name
ComponentManifest.model_validate(comp_data)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Invalid component config: {e}",
)
config = load_config(settings.castle_root)
config.components[name] = ComponentManifest.model_validate(
{**request.config, "id": name}
)
save_config(config)
return {"ok": True, "component": name}
@router.delete("/components/{name}")
def delete_component(name: str) -> dict:
"""Remove a component from castle.yaml."""
config = load_config(settings.castle_root)
if name not in config.components:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Component '{name}' not found",
)
del config.components[name]
save_config(config)
return {"ok": True, "component": name, "action": "deleted"}
@router.post("/apply", response_model=ApplyResponse)
async def apply_config() -> ApplyResponse:
"""Apply config: regenerate systemd units for managed services + reload gateway."""
config = load_config(settings.castle_root)
actions: list[str] = []
errors: list[str] = []
# Regenerate and restart managed services
for name in config.managed:
unit = f"castle-{name}.service"
ok, output = await _systemctl("restart", unit)
if ok:
actions.append(f"Restarted {name}")
else:
errors.append(f"Failed to restart {name}: {output}")
# Reload gateway
from castle_cli.commands.gateway import _generate_caddyfile
from castle_cli.config import GENERATED_DIR, ensure_dirs
ensure_dirs()
caddyfile_path = GENERATED_DIR / "Caddyfile"
caddyfile_path.write_text(_generate_caddyfile(config))
actions.append("Generated Caddyfile")
if shutil.which("caddy"):
ok, output = await _run("caddy", "reload",
"--config", str(caddyfile_path),
"--adapter", "caddyfile")
if ok:
actions.append("Reloaded gateway")
else:
errors.append(f"Gateway reload failed: {output}")
await broadcast("config-changed", {"actions": actions})
return ApplyResponse(ok=len(errors) == 0, actions=actions, errors=errors)
async def _systemctl(action: str, unit: str) -> tuple[bool, str]:
proc = await asyncio.create_subprocess_exec(
"systemctl", "--user", action, unit,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await proc.communicate()
return proc.returncode == 0, (stdout or stderr or b"").decode().strip()
async def _run(*args: str) -> tuple[bool, str]:
proc = await asyncio.create_subprocess_exec(
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await proc.communicate()
return proc.returncode == 0, (stdout or stderr or b"").decode().strip()

View File

@@ -0,0 +1,61 @@
"""Event bus API routes — subscribe, publish, unsubscribe, list topics."""
from __future__ import annotations
from fastapi import APIRouter
from pydantic import BaseModel
from castle_api.bus import bus
router = APIRouter(prefix="/events", tags=["events"])
class PublishRequest(BaseModel):
topic: str
payload: dict
class SubscribeRequest(BaseModel):
topic: str
callback_url: str
subscriber: str = ""
class UnsubscribeRequest(BaseModel):
topic: str
callback_url: str
@router.post("/publish")
async def publish(request: PublishRequest) -> dict:
"""Publish an event to a topic."""
delivered = await bus.publish(request.topic, request.payload)
return {"topic": request.topic, "subscribers_notified": delivered}
@router.post("/subscribe")
async def subscribe(request: SubscribeRequest) -> dict:
"""Subscribe to a topic with a webhook callback URL."""
bus.subscribe(request.topic, request.callback_url, request.subscriber)
return {
"topic": request.topic,
"callback_url": request.callback_url,
"status": "subscribed",
}
@router.post("/unsubscribe")
async def unsubscribe(request: UnsubscribeRequest) -> dict:
"""Unsubscribe from a topic."""
removed = bus.unsubscribe(request.topic, request.callback_url)
return {
"topic": request.topic,
"callback_url": request.callback_url,
"status": "unsubscribed" if removed else "not_found",
}
@router.get("/topics")
async def list_topics() -> dict:
"""List all topics and their subscribers."""
return {"topics": bus.list_topics()}

View File

@@ -0,0 +1,48 @@
"""Async health checker — fans out HTTP requests to service health endpoints."""
from __future__ import annotations
import asyncio
import time
import httpx
from castle_cli.config import CastleConfig
from castle_api.models import HealthStatus
async def check_all_health(config: CastleConfig) -> list[HealthStatus]:
"""Check health of all components with expose.http and a health_path."""
targets: list[tuple[str, str]] = []
for name, manifest in config.components.items():
if not (manifest.expose and manifest.expose.http):
continue
http = manifest.expose.http
if not http.health_path:
continue
host = http.internal.host or "127.0.0.1"
port = http.internal.port
url = f"http://{host}:{port}{http.health_path}"
targets.append((name, url))
if not targets:
return []
async with httpx.AsyncClient(timeout=3.0) as client:
tasks = [_check_one(client, name, url) for name, url in targets]
return await asyncio.gather(*tasks)
async def _check_one(client: httpx.AsyncClient, name: str, url: str) -> HealthStatus:
"""Check a single service's health endpoint."""
start = time.monotonic()
try:
resp = await client.get(url)
latency = int((time.monotonic() - start) * 1000)
if resp.status_code < 300:
return HealthStatus(id=name, status="up", latency_ms=latency)
return HealthStatus(id=name, status="down", latency_ms=latency)
except httpx.HTTPError:
latency = int((time.monotonic() - start) * 1000)
return HealthStatus(id=name, status="down", latency_ms=latency)

View File

@@ -0,0 +1,70 @@
"""Log streaming — tail journalctl output for systemd-managed services."""
from __future__ import annotations
import asyncio
from collections.abc import AsyncGenerator
from fastapi import APIRouter, HTTPException, Query, status
from starlette.responses import StreamingResponse
from castle_cli.config import load_config
from castle_api.config import settings
router = APIRouter(prefix="/logs", tags=["logs"])
UNIT_PREFIX = "castle-"
@router.get("/{name}", response_model=None)
async def get_logs(
name: str,
n: int = Query(default=100, ge=1, le=5000, description="Number of lines"),
follow: bool = Query(default=False, description="Stream new lines via SSE"),
) -> StreamingResponse | dict:
"""Get logs for a systemd-managed service."""
config = load_config(settings.castle_root)
if name not in config.managed:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"'{name}' is not a managed service",
)
unit = f"{UNIT_PREFIX}{name}.service"
if follow:
return StreamingResponse(
_follow_logs(unit, n),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
# Static tail
proc = await asyncio.create_subprocess_exec(
"journalctl", "--user", "-u", unit, "-n", str(n), "--no-pager",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, _ = await proc.communicate()
lines = (stdout or b"").decode().splitlines()
return {"component": name, "lines": lines}
async def _follow_logs(unit: str, n: int) -> AsyncGenerator[str, None]:
"""Stream journalctl -f output as SSE events."""
proc = await asyncio.create_subprocess_exec(
"journalctl", "--user", "-u", unit, "-n", str(n), "-f", "--no-pager",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
try:
assert proc.stdout is not None
async for line in proc.stdout:
text = line.decode().rstrip()
yield f"data: {text}\n\n"
except asyncio.CancelledError:
pass
finally:
proc.kill()
await proc.wait()

View File

@@ -0,0 +1,128 @@
"""Castle API — dashboard data, health aggregation, event bus, service management."""
from __future__ import annotations
import asyncio
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import StreamingResponse
from castle_api.bus import bus
from castle_api.config import settings
from castle_api.config_editor import router as config_router
from castle_api.events import router as events_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
from castle_api.services import router as services_router
from castle_api.stream import close_all_subscribers, health_poll_loop, subscribe, unsubscribe
from castle_api.tools import router as tools_router
# Set by _watch_shutdown when uvicorn begins its shutdown sequence.
_shutting_down = False
async def _watch_shutdown(server: uvicorn.Server) -> None:
"""Poll uvicorn's should_exit flag and close SSE subscribers promptly."""
while not server.should_exit:
await asyncio.sleep(0.5)
global _shutting_down
_shutting_down = True
close_all_subscribers()
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
"""Application lifespan handler."""
global _shutting_down
_shutting_down = False
await bus.start()
poll_task = asyncio.create_task(health_poll_loop())
yield
_shutting_down = True
poll_task.cancel()
close_all_subscribers()
await bus.stop()
app = FastAPI(
title="castle-api",
description="Castle API, event bus, and service management",
version="0.1.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(config_router)
app.include_router(dashboard_router)
app.include_router(events_router)
app.include_router(logs_router)
app.include_router(secrets_router)
app.include_router(services_router)
app.include_router(tools_router)
@app.get("/health")
def health() -> dict[str, str]:
"""Health check endpoint."""
return {"status": "ok"}
@app.get("/stream")
async def sse_stream() -> StreamingResponse:
"""SSE stream — pushes health updates and service action events."""
q = subscribe()
async def event_generator() -> AsyncGenerator[str, None]:
try:
yield "event: connected\ndata: {}\n\n"
while True:
msg = await q.get()
if not msg:
break
yield msg
except asyncio.CancelledError:
pass
finally:
unsubscribe(q)
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
def run() -> None:
"""Run the application with uvicorn."""
config = uvicorn.Config(
"castle_api.main:app",
host=settings.host,
port=settings.port,
reload=False,
)
server = uvicorn.Server(config)
async def serve_with_watcher() -> None:
watcher = asyncio.create_task(_watch_shutdown(server))
await server.serve()
watcher.cancel()
asyncio.run(serve_with_watcher())
if __name__ == "__main__":
run()

View File

@@ -0,0 +1,92 @@
"""Response models for the dashboard API."""
from pydantic import BaseModel
class SystemdInfo(BaseModel):
"""Systemd unit information for a managed component."""
unit_name: str
unit_path: str
timer: bool = False
class ComponentSummary(BaseModel):
"""Summary of a single component."""
id: str
description: str | None = None
roles: list[str]
runner: str | None = None
port: int | None = None
health_path: str | None = None
proxy_path: str | None = None
managed: bool = False
systemd: SystemdInfo | None = None
version: str | None = None
source: str | None = None
system_dependencies: list[str] = []
schedule: str | None = None
installed: bool | None = None
class ComponentDetail(ComponentSummary):
"""Full detail for a single component, including raw manifest."""
manifest: dict
class HealthStatus(BaseModel):
"""Health status of a single component."""
id: str
status: str # "up", "down", "unknown"
latency_ms: int | None = None
class StatusResponse(BaseModel):
"""Aggregated health status for all exposed components."""
statuses: list[HealthStatus]
class GatewayInfo(BaseModel):
"""Gateway configuration summary."""
port: int
component_count: int
service_count: int
managed_count: int
class ServiceActionResponse(BaseModel):
"""Response from a service management action."""
component: str
action: str
status: str
class ToolSummary(BaseModel):
"""Summary of a single tool."""
id: str
description: str | None = None
source: str | None = None
version: str | None = None
runner: str | None = None
system_dependencies: list[str] = []
installed: bool = False
class ToolCategory(BaseModel):
"""Tools grouped by category."""
name: str
tools: list[ToolSummary]
class ToolDetail(ToolSummary):
"""Full detail for a single tool, including documentation."""
docs: str | None = None

View File

@@ -0,0 +1,143 @@
"""API routes for the castle dashboard."""
from __future__ import annotations
import shutil
from pathlib import Path
from fastapi import APIRouter, HTTPException, status
from castle_cli.config import load_config
from castle_api.config import settings
from castle_api.health import check_all_health
from castle_api.models import (
ComponentDetail,
ComponentSummary,
GatewayInfo,
StatusResponse,
SystemdInfo,
)
router = APIRouter(tags=["dashboard"])
def _summary_from_manifest(name: str, manifest: object, root: Path) -> ComponentSummary:
"""Build a ComponentSummary from a manifest."""
port = None
health_path = None
proxy_path = None
if manifest.expose and manifest.expose.http:
port = manifest.expose.http.internal.port
health_path = manifest.expose.http.health_path
if manifest.proxy and manifest.proxy.caddy:
proxy_path = manifest.proxy.caddy.path_prefix
managed = bool(
manifest.manage and manifest.manage.systemd and manifest.manage.systemd.enable
)
# Systemd info for managed components
systemd_info: SystemdInfo | None = None
if managed:
unit_name = f"castle-{name}.service"
unit_path = str(Path("~/.config/systemd/user") / unit_name)
has_timer = any(getattr(t, "type", None) == "schedule" for t in manifest.triggers)
systemd_info = SystemdInfo(
unit_name=unit_name,
unit_path=unit_path,
timer=has_timer,
)
# Extract cron schedule from first schedule trigger, if any
schedule = None
for t in manifest.triggers:
if t.type == "schedule":
schedule = t.cron
break
# Infer runner — from run block or from tool source
runner = manifest.run.runner if manifest.run else None
if runner is None and manifest.tool and manifest.tool.source:
source_dir = root / manifest.tool.source
if (source_dir / "pyproject.toml").exists():
runner = "python_uv_tool"
elif source_dir.is_file():
runner = "command"
# Check if tool is actually installed on PATH
installed: bool | None = None
if manifest.install and manifest.install.path:
alias = manifest.install.path.alias or name
installed = shutil.which(alias) is not None
return ComponentSummary(
id=name,
description=manifest.description,
roles=[r.value for r in manifest.roles],
runner=runner,
port=port,
health_path=health_path,
proxy_path=proxy_path,
managed=managed,
systemd=systemd_info,
version=manifest.tool.version if manifest.tool else None,
source=manifest.tool.source if manifest.tool else None,
system_dependencies=manifest.tool.system_dependencies if manifest.tool else [],
schedule=schedule,
installed=installed,
)
@router.get("/components", response_model=list[ComponentSummary])
def list_components() -> list[ComponentSummary]:
"""List all registered components."""
config = load_config(settings.castle_root)
return [
_summary_from_manifest(name, m, config.root)
for name, m in config.components.items()
]
@router.get("/components/{name}", response_model=ComponentDetail)
def get_component(name: str) -> ComponentDetail:
"""Get detailed info for a single component."""
config = load_config(settings.castle_root)
if name not in config.components:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Component '{name}' not found",
)
manifest = config.components[name]
summary = _summary_from_manifest(name, manifest, config.root)
raw = manifest.model_dump(mode="json", exclude_none=True)
return ComponentDetail(**summary.model_dump(), manifest=raw)
@router.get("/status", response_model=StatusResponse)
async def get_status() -> StatusResponse:
"""Get live health status for all exposed services."""
config = load_config(settings.castle_root)
statuses = await check_all_health(config)
return StatusResponse(statuses=statuses)
@router.get("/gateway", response_model=GatewayInfo)
def get_gateway() -> GatewayInfo:
"""Get gateway configuration summary."""
config = load_config(settings.castle_root)
return GatewayInfo(
port=config.gateway.port,
component_count=len(config.components),
service_count=len(config.services),
managed_count=len(config.managed),
)
@router.get("/gateway/caddyfile")
def get_caddyfile() -> dict[str, str]:
"""Return the generated Caddyfile content."""
from castle_cli.commands.gateway import _generate_caddyfile
config = load_config(settings.castle_root)
return {"content": _generate_caddyfile(config)}

View File

@@ -0,0 +1,64 @@
"""Secrets management — read and write ~/.castle/secrets/."""
from __future__ import annotations
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel
from castle_cli.config import SECRETS_DIR
router = APIRouter(prefix="/secrets", tags=["secrets"])
class SecretValue(BaseModel):
value: str
@router.get("")
def list_secrets() -> list[str]:
"""List all secret names (not values)."""
if not SECRETS_DIR.exists():
return []
return sorted(f.name for f in SECRETS_DIR.iterdir() if f.is_file())
@router.get("/{name}")
def get_secret(name: str) -> dict:
"""Get a secret value."""
_validate_name(name)
path = SECRETS_DIR / name
if not path.exists():
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Secret '{name}' not found",
)
return {"name": name, "value": path.read_text().strip()}
@router.put("/{name}")
def set_secret(name: str, body: SecretValue) -> dict:
"""Set a secret value."""
_validate_name(name)
SECRETS_DIR.mkdir(parents=True, exist_ok=True)
path = SECRETS_DIR / name
path.write_text(body.value.strip() + "\n")
return {"name": name, "ok": True}
@router.delete("/{name}")
def delete_secret(name: str) -> dict:
"""Delete a secret."""
_validate_name(name)
path = SECRETS_DIR / name
if path.exists():
path.unlink()
return {"name": name, "ok": True}
def _validate_name(name: str) -> None:
"""Reject path traversal attempts."""
if "/" in name or "\\" in name or ".." in name or not name:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid secret name",
)

View File

@@ -0,0 +1,143 @@
"""Service management — start/stop/restart systemd-managed components."""
from __future__ import annotations
import asyncio
import time
from fastapi import APIRouter, HTTPException, status
from starlette.responses import JSONResponse
from castle_cli.config import load_config
from castle_api.config import settings
from castle_api.health import check_all_health
from castle_api.models import HealthStatus
from castle_api.stream import broadcast
router = APIRouter(prefix="/services", tags=["services"])
UNIT_PREFIX = "castle-"
SELF_NAME = "castle-api"
async def _systemctl(action: str, unit: str) -> tuple[bool, str]:
"""Run a systemctl --user command. Returns (success, output)."""
proc = await asyncio.create_subprocess_exec(
"systemctl", "--user", action, unit,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await proc.communicate()
output = (stdout or stderr or b"").decode().strip()
return proc.returncode == 0, output
async def _get_unit_status(unit: str) -> str:
"""Get the active status of a systemd unit."""
proc = await asyncio.create_subprocess_exec(
"systemctl", "--user", "is-active", unit,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, _ = await proc.communicate()
return (stdout or b"").decode().strip()
def _validate_managed(name: str) -> None:
"""Raise 404 if the component isn't systemd-managed."""
config = load_config(settings.castle_root)
if name not in config.managed:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"'{name}' is not a managed service",
)
async def _broadcast_health_with_override(
override_name: str, override_status: str
) -> None:
"""Run health checks but override one component's status from systemd."""
config = load_config(settings.castle_root)
statuses = await check_all_health(config)
# Replace the overridden component's status with the systemd truth
result = []
for s in statuses:
if s.id == override_name:
result.append(HealthStatus(
id=override_name,
status="down" if override_status != "active" else "up",
latency_ms=None,
))
else:
result.append(s)
await broadcast("health", {
"statuses": [s.model_dump() for s in result],
"timestamp": time.time(),
})
async def _deferred_systemctl(action: str, unit: str, delay: float = 0.5) -> None:
"""Run a systemctl action after a delay, allowing the HTTP response to flush."""
await asyncio.sleep(delay)
await _systemctl(action, unit)
async def _do_action(name: str, action: str) -> JSONResponse:
"""Execute a systemctl action and broadcast updated health."""
_validate_managed(name)
unit = f"{UNIT_PREFIX}{name}.service"
# Self-restart: defer the systemctl call so the response can be sent first
if name == SELF_NAME and action in ("restart", "stop"):
asyncio.create_task(_deferred_systemctl(action, unit))
return JSONResponse(
status_code=202,
content={"component": name, "action": action, "status": "accepted"},
)
ok, output = await _systemctl(action, unit)
unit_status = await _get_unit_status(unit)
if not ok:
raise HTTPException(status_code=500, detail=output or f"Failed to {action}")
# Broadcast immediately with systemd status as the source of truth
await _broadcast_health_with_override(name, unit_status)
return JSONResponse(
content={"component": name, "action": action, "status": unit_status},
)
@router.get("/{name}/unit")
def get_unit(name: str) -> dict[str, str | None]:
"""Return the generated systemd unit file(s) for a managed component."""
from castle_cli.commands.service import _generate_timer, _generate_unit
_validate_managed(name)
config = load_config(settings.castle_root)
manifest = config.managed[name]
unit = _generate_unit(config, name, manifest)
timer = _generate_timer(name, manifest)
return {"service": unit, "timer": timer}
@router.post("/{name}/start")
async def start_service(name: str) -> JSONResponse:
"""Start a systemd-managed service."""
return await _do_action(name, "start")
@router.post("/{name}/stop")
async def stop_service(name: str) -> JSONResponse:
"""Stop a systemd-managed service."""
return await _do_action(name, "stop")
@router.post("/{name}/restart")
async def restart_service(name: str) -> JSONResponse:
"""Restart a systemd-managed service."""
return await _do_action(name, "restart")

View File

@@ -0,0 +1,71 @@
"""SSE stream — pushes health updates and service action events to connected clients."""
from __future__ import annotations
import asyncio
import json
import logging
import time
from castle_cli.config import load_config
from castle_api.config import settings
from castle_api.health import check_all_health
logger = logging.getLogger(__name__)
# All connected SSE clients receive events through this queue-based broadcast.
_subscribers: list[asyncio.Queue[str]] = []
def subscribe() -> asyncio.Queue[str]:
"""Register a new SSE client. Returns a queue to read events from."""
q: asyncio.Queue[str] = asyncio.Queue(maxsize=64)
_subscribers.append(q)
return q
def unsubscribe(q: asyncio.Queue[str]) -> None:
"""Remove a disconnected SSE client."""
try:
_subscribers.remove(q)
except ValueError:
pass
def close_all_subscribers() -> None:
"""Unblock all SSE generators so they exit during shutdown."""
for q in list(_subscribers):
try:
q.put_nowait("")
except asyncio.QueueFull:
pass
_subscribers.clear()
async def broadcast(event_type: str, data: dict) -> None:
"""Send an event to all connected SSE clients."""
payload = f"event: {event_type}\ndata: {json.dumps(data)}\n\n"
dead: list[asyncio.Queue[str]] = []
for q in _subscribers:
try:
q.put_nowait(payload)
except asyncio.QueueFull:
dead.append(q)
for q in dead:
unsubscribe(q)
async def health_poll_loop(interval: float = 10.0) -> None:
"""Background task that polls health and broadcasts updates."""
while True:
try:
config = load_config(settings.castle_root)
statuses = await check_all_health(config)
await broadcast("health", {
"statuses": [s.model_dump() for s in statuses],
"timestamp": time.time(),
})
except Exception:
logger.exception("Health poll failed")
await asyncio.sleep(interval)

View File

@@ -0,0 +1,189 @@
"""Tools router — browse and inspect tool components."""
from __future__ import annotations
import asyncio
from pathlib import Path
from fastapi import APIRouter, HTTPException, status
from castle_cli.config import load_config
from castle_cli.manifest import ComponentManifest
from castle_api.config import settings
from castle_api.models import ToolCategory, ToolDetail, ToolSummary
router = APIRouter(tags=["tools"])
def _tool_summary(name: str, manifest: ComponentManifest, root: Path | None = None) -> ToolSummary:
"""Build a ToolSummary from a manifest that has a tool spec."""
t = manifest.tool
assert t is not None
installed = bool(manifest.install and manifest.install.path and manifest.install.path.enable)
# Infer runner from run block or source directory
runner = manifest.run.runner if manifest.run else None
if runner is None and t.source and root:
source_dir = root / t.source
if (source_dir / "pyproject.toml").exists():
runner = "python_uv_tool"
elif source_dir.is_file():
runner = "command"
return ToolSummary(
id=name,
description=manifest.description,
source=t.source,
version=t.version,
runner=runner,
system_dependencies=t.system_dependencies,
installed=installed,
)
def _find_md_for_tool(
root: Path,
source: str,
tool_name: str,
) -> Path | None:
"""Find the .md documentation file for a tool source path."""
source_path = root / source
if source_path.is_file():
md = source_path.with_suffix(".md")
if md.exists():
return md
elif source_path.is_dir():
py_name = tool_name.replace("-", "_")
pkg_name = source_path.name
md = source_path / "src" / pkg_name / f"{py_name}.md"
if md.exists():
return md
return None
def _strip_frontmatter(content: str) -> str:
"""Strip YAML frontmatter from markdown content."""
if content.startswith("---\n"):
end = content.find("\n---\n", 4)
if end != -1:
content = content[end + 5:]
return content.strip()
@router.get("/tools", response_model=list[ToolCategory])
def list_tools() -> list[ToolCategory]:
"""List tools grouped by source directory."""
config = load_config(settings.castle_root)
tools = {k: v for k, v in config.components.items() if v.tool}
by_group: dict[str, list[ToolSummary]] = {}
for name, manifest in tools.items():
t = manifest.tool
assert t is not None
if t.source:
group = Path(t.source).name
else:
group = "standalone"
by_group.setdefault(group, []).append(_tool_summary(name, manifest, config.root))
return [
ToolCategory(name=group, tools=sorted(items, key=lambda t: t.id))
for group, items in sorted(by_group.items())
]
@router.get("/tools/{name}", response_model=ToolDetail)
def get_tool(name: str) -> ToolDetail:
"""Get detailed info for a single tool."""
config = load_config(settings.castle_root)
if name not in config.components:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Component '{name}' not found",
)
manifest = config.components[name]
if not manifest.tool:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"'{name}' is not a tool",
)
summary = _tool_summary(name, manifest, config.root)
docs: str | None = None
t = manifest.tool
if t.source:
md_path = _find_md_for_tool(config.root, t.source, name)
if md_path and md_path.exists():
docs = _strip_frontmatter(md_path.read_text())
if not docs:
docs = None
return ToolDetail(**summary.model_dump(), docs=docs)
@router.post("/tools/{name}/install")
async def install_tool(name: str) -> dict:
"""Install a tool to PATH via uv tool install."""
config = load_config(settings.castle_root)
if name not in config.components:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"'{name}' not found")
manifest = config.components[name]
if not manifest.tool or not manifest.tool.source:
raise HTTPException(status_code=400, detail=f"'{name}' has no tool source to install")
source_dir = config.root / manifest.tool.source
if not (source_dir / "pyproject.toml").exists():
raise HTTPException(status_code=400, detail=f"No pyproject.toml in {manifest.tool.source}")
proc = await asyncio.create_subprocess_exec(
"uv", "tool", "install", "--editable", str(source_dir), "--force",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await proc.communicate()
output = (stdout or stderr or b"").decode().strip()
if proc.returncode != 0:
raise HTTPException(status_code=500, detail=output or "Install failed")
return {"component": name, "action": "install", "status": "ok"}
@router.post("/tools/{name}/uninstall")
async def uninstall_tool(name: str) -> dict:
"""Uninstall a tool from PATH via uv tool uninstall."""
config = load_config(settings.castle_root)
if name not in config.components:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"'{name}' not found")
manifest = config.components[name]
if not manifest.tool or not manifest.tool.source:
raise HTTPException(status_code=400, detail=f"'{name}' has no tool source")
# uv tool uninstall uses the package name from pyproject.toml
source_dir = config.root / manifest.tool.source
# Try to read the package name; fall back to the source dir name
pkg_name = source_dir.name
pyproject = source_dir / "pyproject.toml"
if pyproject.exists():
import tomllib
with open(pyproject, "rb") as f:
data = tomllib.load(f)
pkg_name = data.get("project", {}).get("name", pkg_name)
proc = await asyncio.create_subprocess_exec(
"uv", "tool", "uninstall", pkg_name,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await proc.communicate()
output = (stdout or stderr or b"").decode().strip()
if proc.returncode != 0:
raise HTTPException(status_code=500, detail=output or "Uninstall failed")
return {"component": name, "action": "uninstall", "status": "ok"}