serverAppConfigs/scripts/deploy-plan.py

#!/usr/bin/env python3
"""Plan runtime services affected by repository changes.

The planner is intentionally side-effect free.  It accepts either an explicit
path list or two Git revisions.  The latter lets compose-file changes resolve
to only the service definitions that actually changed instead of restarting an
entire compose project.
"""
from __future__ import annotations

import argparse
import subprocess
import sys
from pathlib import PurePosixPath
from typing import Any

import yaml

# Dependency-sensitive services come before their consumers.  Independent
# compose services follow.  LLM services are deliberately last and are never
# implied by a generic host, stack, documentation, or config change.
SERVICE_ORDER = (
    "gluetun",
    "searxng-valkey",
    "searxng",
    "pihole",
    "open-webui",
    "gitea-postgres",
    "gitea-server",
    "comfyui",
    "hermes",
    "hermes-dashboard",
    "hermes-webui",
    "prometheus",
    "grafana",
    "loki",
    "alloy",
    "node-exporter",
    "authentik-postgres",
    "authentik-server",
    "authentik-worker",
    "authentik-outpost",
    "caddy",
    "homepage",
    "romm-mariadb",
    "romm-redis",
    "romm",
    "copyparty",
    "jellyfin",
    "calibre-web",
    "prowlarr",
    "qbittorrent",
    "radarr",
    "sonarr",
    "lidarr",
    "readarr",
    "frank-default-large",
    "frank-default-npu",
    "frank-default-sightsound",
    "frank-default-small",
)

DANKCLAW_SERVICES = {
    "hermes-webui",
    "gluetun",
    "searxng-valkey",
    "searxng",
    "pihole",
    "open-webui",
    "gitea-postgres",
    "gitea-server",
    "comfyui",
    "hermes",
    "hermes-dashboard",
    "prometheus",
    "grafana",
    "loki",
    "alloy",
    "node-exporter",
    "authentik-postgres",
    "authentik-server",
    "authentik-worker",
    "authentik-outpost",
    "caddy",
    "homepage",
    "frank-default-large",
    "frank-default-npu",
    "frank-default-sightsound",
}

DANKSERVER_SERVICES = {
    "romm-mariadb",
    "romm-redis",
    "romm",
    "copyparty",
    "jellyfin",
    "calibre-web",
    "prowlarr",
    "qbittorrent",
    "radarr",
    "sonarr",
    "lidarr",
    "readarr",
    "frank-default-small",
}

DANKCLAW_COMPOSE_SERVICES = {
    "comfyui",
    "hermes",
    "hermes-dashboard",
    "hermes-webui",
    "prometheus",
    "grafana",
    "loki",
    "alloy",
    "node-exporter",
    "authentik-postgres",
    "authentik-server",
    "authentik-worker",
    "caddy",
}

DANKSERVER_COMPOSE_SERVICES = {
    "mariadb",
    "redis",
    "romm",
    "copyparty",
    "jellyfin",
    "calibre-web",
    "prowlarr",
    "qbittorrent",
    "radarr",
    "sonarr",
    "lidarr",
    "readarr",
}
COMPOSE_NAME_MAP = {"mariadb": "romm-mariadb", "redis": "romm-redis"}



def _add(services: set[str], *names: str) -> None:
    services.update(names)


def target_services(target: str) -> set[str]:
    if target == "dankclaw":
        return DANKCLAW_SERVICES
    if target == "dankserver":
        return DANKSERVER_SERVICES
    raise ValueError(f"unknown target: {target}")


def _compose_services(target: str) -> set[str]:
    return DANKCLAW_COMPOSE_SERVICES if target == "dankclaw" else DANKSERVER_COMPOSE_SERVICES


def _git_show(repo: str, revision: str, path: str) -> str | None:
    result = subprocess.run(
        ["git", "-C", repo, "show", f"{revision}:{path}"],
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        check=False,
    )
    return result.stdout if result.returncode == 0 else None


def _yaml_document(contents: str | None) -> dict[str, Any]:
    if contents is None:
        return {}
    try:
        data = yaml.safe_load(contents) or {}
    except yaml.YAMLError:
        return {}
    return data if isinstance(data, dict) else {}


def _yaml_services(contents: str | None) -> dict[str, Any]:
    services = _yaml_document(contents).get("services", {})
    return services if isinstance(services, dict) else {}


def compose_changes(repo: str, target: str, base: str, head: str) -> set[str]:
    """Return only compose services whose YAML definitions or dependencies changed."""
    path = f"{target}/compose.yaml"
    old_doc = _yaml_document(_git_show(repo, base, path))
    new_doc = _yaml_document(_git_show(repo, head, path))
    old = old_doc.get("services", {}) if isinstance(old_doc.get("services", {}), dict) else {}
    new = new_doc.get("services", {}) if isinstance(new_doc.get("services", {}), dict) else {}
    names = set(old) | set(new)
    allowed = _compose_services(target)

    # An unavailable or malformed revision is safer as a broad compose plan
    # than as a silent no-op. It still cannot include LLM units.
    if (old_doc and not new_doc) or (new_doc and not old_doc and names):
        return {COMPOSE_NAME_MAP.get(n, n) for n in allowed}

    # Top-level networks/configuration can affect every compose service even
    # when no individual service mapping changed. Ignore the services key while
    # comparing the rest of the Compose document.
    old_globals = {key: value for key, value in old_doc.items() if key != "services"}
    new_globals = {key: value for key, value in new_doc.items() if key != "services"}
    if old_globals != new_globals:
        return {COMPOSE_NAME_MAP.get(n, n) for n in allowed}

    changed = {name for name in names if name in allowed and old.get(name) != new.get(name)}
    return {COMPOSE_NAME_MAP.get(n, n) for n in changed}


def services_for_path(path: str, target: str) -> set[str]:
    """Return services affected by one repository-relative path."""
    path = path.replace("\\", "/")
    p = PurePosixPath(path)
    parts = p.parts
    services: set[str] = set()
    if not parts:
        return services

    # Non-runtime material is deliberately a no-op.
    if parts[0] in {"docs", "tests", "old", ".githooks"}:
        return services
    if path in {"README.md", "AGENTS.md"} or parts[0] == ".obsidian":
        return services

    if parts[0] == "shared":
        if len(parts) < 2:
            return services
        group = parts[1]
        if group == "searxng":
            _add(services, "searxng")
        elif group == "prometheus":
            _add(services, "prometheus")
        elif group == "loki":
            _add(services, "loki")
        elif group == "alloy":
            _add(services, "alloy")
        elif group == "homepage":
            # Homepage is a Dankclaw-owned deployment. Shared homepage files
            # are intentionally not copied during a Dankserver-only run.
            if target == "dankclaw":
                _add(services, "homepage")
        return services

    if parts[0] == "dankclaw":
        if len(parts) >= 2 and parts[1] == "compose.yaml":
            # Revision-aware callers replace this broad fallback with the
            # service-level YAML comparison in compose_changes().
            _add(services, *DANKCLAW_COMPOSE_SERVICES)
            return services
        if len(parts) >= 2 and parts[1] == "Caddyfile":
            _add(services, "caddy")
            return services
        if len(parts) >= 2 and parts[1] == "Containerfile.comfyui":
            _add(services, "comfyui")
            return services
        if len(parts) >= 2 and parts[1] in {"caddy", "comfyui", "loki", "alloy", "hermes", "anubis"}:
            _add(services, parts[1])
            return services
        if len(parts) >= 4 and parts[1] == "authentik" and parts[2] == "quadlet":
            name = parts[3].removesuffix(".container")
            if name in {"authentik-postgres", "authentik-server", "authentik-worker", "authentik-outpost"}:
                _add(services, name)
            return services
        if len(parts) >= 2 and parts[1] == "searxng":
            if len(parts) >= 4 and parts[2] == "quadlet":
                name = parts[3]
                if name == "gluetun.container":
                    # SearXNG shares Gluetun's network namespace.  A Gluetun
                    # recreation requires both units to be restarted.
                    _add(services, "gluetun", "searxng")
                elif name == "searxng.container":
                    _add(services, "searxng")
                elif name == "searxng-valkey.container":
                    _add(services, "searxng-valkey", "searxng")
                elif name == "searxng.network":
                    _add(services, "gluetun", "searxng-valkey", "searxng")
            return services
        if len(parts) >= 2 and parts[1] == "pihole":
            _add(services, "pihole")
            return services
        if len(parts) >= 2 and parts[1] == "open-webui":
            _add(services, "open-webui")
            return services
        if len(parts) >= 2 and parts[1] == "gitea":
            if len(parts) >= 4 and parts[2] == "quadlet":
                name = parts[3]
                if name == "gitea-postgres.container":
                    _add(services, "gitea-postgres")
                elif name == "gitea-server.container":
                    _add(services, "gitea-server")
                elif name == "gitea-caddy.container":
                    # Removing the old split proxy requires the canonical
                    # Caddy handler to retire its installed unit/container
                    # before it claims ports 80/443.
                    _add(services, "caddy")
                elif name == "gitea.network":
                    _add(services, "gitea-postgres", "gitea-server")
            return services
        return services

    if parts[0] == "dankserver":
        if len(parts) >= 2 and parts[1] == "compose.yaml":
            # Compose service keys map onto the deployed unit names.
            _add(services, *(COMPOSE_NAME_MAP.get(n, n) for n in DANKSERVER_COMPOSE_SERVICES))
            return services
        if len(parts) >= 3 and parts[2] == "quadlet":
            quadlet_services = {
                "romm": {"romm", "romm-mariadb", "romm-redis"},
                "jellyfin": {"jellyfin"},
                "calibre-web": {"calibre-web"},
                "copyparty": {"copyparty"},
            }
            _add(services, *quadlet_services.get(parts[1], set()))
        return services

    if parts[0] == "scripts":
        name = p.name
        if name in {"frank-default-large.env", "frank-default-large.service", "serve-frank-large.sh"}:
            _add(services, "frank-default-large")
        elif name in {"frank-default-npu.env", "frank-default-npu.service", "serve-frank-default-npu.sh"}:
            _add(services, "frank-default-npu")
        elif name in {"frank-default-sightsound.env", "frank-default-sightsound.service", "serve-frank-default-sightsound.sh"}:
            _add(services, "frank-default-sightsound")
        elif name in {"frank-default-small.env", "frank-default-small.service", "serve-frank-small.sh"}:
            _add(services, "frank-default-small")
        elif name == "chat-template.jinja":
            # Only the selected target's LLM consumes this shared template.
            _add(services, "frank-default-large" if target == "dankclaw" else "frank-default-small")
        elif name in {"homepage-status.py", "homepage-status.service", "homepage-status.timer"}:
            _add(services, "homepage")
        # Deployment/test/benchmark scripts and other repo tooling do not
        # restart a runtime service.
        return services

    # Examples and env templates are not deployed runtime state.
    if p.name in {".env.example", "env.example"}:
        return services
    return services


def changed_paths(repo: str, base: str, head: str) -> list[str]:
    result = subprocess.run(
        ["git", "-C", repo, "diff", "--name-only", "--diff-filter=ACDMRTUXB", base, head],
        text=True,
        stdout=subprocess.PIPE,
        check=True,
    )
    return [line for line in result.stdout.splitlines() if line]


def plan(paths: list[str], target: str, repo: str | None = None,
         base: str | None = None, head: str | None = None) -> list[str]:
    services: set[str] = set()
    for path in paths:
        if path == f"{target}/compose.yaml" and repo and base and head:
            services.update(compose_changes(repo, target, base, head))
        else:
            services.update(services_for_path(path, target))
    allowed = target_services(target)
    return [name for name in SERVICE_ORDER if name in services and name in allowed]


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--target", choices=("dankclaw", "dankserver"), required=True)
    parser.add_argument("--paths", nargs="+")
    parser.add_argument("--repo", help="repository path for revision-aware planning")
    parser.add_argument("--base", help="base Git revision")
    parser.add_argument("--head", help="head Git revision")
    args = parser.parse_args()

    if not args.paths and not (args.repo and args.base and args.head):
        parser.error("provide --paths or all of --repo, --base, and --head")
    paths = args.paths or changed_paths(args.repo, args.base, args.head)
    print("\n".join(plan(paths, args.target, args.repo, args.base, args.head)))
    return 0


if __name__ == "__main__":
    sys.exit(main())