feat: relationship model (requires/repos/predicates) + git sync

A derived, mostly-computed model of how programs, deployments, and repos relate,
plus the git-sync surfaces that motivated it. See docs/relationships.md.

Core:
- `requires: [{kind, ref, version?, bind?}]` on programs + deployments — one
  precondition relation; `system_dependencies` is its `{kind: system}` alias.
  kind fixes meaning + check (system=installed, deployment=exists).
- relations.py: derives repos (git toplevel / monorepo), fan-in, and the
  predicates functional?/fresh?/deployed? — nothing stored.
- env is generated FROM a `{kind: deployment, bind}` requirement (target URL →
  consumer env), never scraped back into one; explicit defaults.env still wins.
- git.py: working-copy status/pull, repo toplevel + remote url.

Surfaces:
- `castle graph` + GET /graph — the relationship diagnostic.
- GET /repos, /repos/{key}/git|sync — repo-scoped sync (a repo is the sync unit;
  a monorepo backs several programs). GET /programs/{name}/git|sync + repo context.
- Dashboard: Graph screen, program-page git status + repo-aware Sync, and a
  monorepo banner on Programs.

Governing principle: predicates are derived; encode only the non-derivable, as a
node or edge property. Pull-only sync — converge stays a separate step.

core/src/castle_core/deploy.py

@@ -244,7 +244,9 @@ def apply(
             return "restart"
         return "unchanged"
 
-    result = ApplyResult(registry=NodeRegistry(node=_node_config(config), deployed=desired))
+    result = ApplyResult(
+        registry=NodeRegistry(node=_node_config(config), deployed=desired)
+    )
 
     if plan:
         # No writes: for systemd, predict the new unit bytes by rendering to a string
@@ -469,6 +471,36 @@ def _public_url(
     return None
 
 
+def _target_url(config: CastleConfig, target_name: str) -> str | None:
+    """The base URL another deployment is reachable at — how a ``{kind: deployment,
+    bind: VAR}`` requirement projects its target into the consumer's env."""
+    dep = config.deployments.get(target_name)
+    if dep is None:
+        return None
+    expose = getattr(dep, "expose", None)
+    http = getattr(expose, "http", None) if expose else None
+    tport = http.internal.port if http else None
+    return _public_url(config, target_name, getattr(dep, "http_exposed", False), tport)
+
+
+def _requires_env(config: CastleConfig, name: str, config_key: str) -> dict[str, str]:
+    """Env generated FROM a deployment's ``requires`` — a ``{kind: deployment,
+    bind: VAR}`` requirement sets ``VAR`` to the target's URL. Env is derived from
+    the dependency, never scraped back into one (see docs/relationships.md)."""
+    dep = config.deployments[name]
+    prog = config.programs.get(config_key)
+    reqs = list(getattr(dep, "requires", []) or [])
+    if prog:
+        reqs += list(prog.requires)
+    out: dict[str, str] = {}
+    for r in reqs:
+        if r.kind == "deployment" and r.bind:
+            url = _target_url(config, r.ref)
+            if url:
+                out[r.bind] = url
+    return out
+
+
 def _supabase_app_schemas(config: CastleConfig) -> str:
     """The ``${supabase_app_schemas}`` placeholder: each registered supabase app's
     own schema, comma-prefixed and joined (or '' when there are none).
@@ -621,8 +653,14 @@ def _build_deployed(
     # names to castle's computed values. Secret-bearing vars split out to a
     # mode-0600 file (never in the unit or argv).
     raw_env = dict(dep.defaults.env) if (dep.defaults and dep.defaults.env) else {}
+    # Env generated from `requires` ({kind: deployment, bind: VAR} → target URL).
+    # An explicit defaults.env value always wins — a hand-set var is never clobbered.
+    for var, url in _requires_env(config, name, config_key).items():
+        raw_env.setdefault(var, url)
     public_url = _public_url(config, name, expose, port)
-    ctx = _env_context(name, config_key, port, public_url, _supabase_app_schemas(config))
+    ctx = _env_context(
+        name, config_key, port, public_url, _supabase_app_schemas(config)
+    )
     # ${tls_*}: paths to castle-materialized cert files for a TLS-material TCP
     # service. The deployment maps them into its own config (mount ${tls_dir} for a
     # container, or reference ${tls_cert}/${tls_key} directly for a native service).
@@ -647,7 +685,12 @@ def _build_deployed(
         _ensure_python_tool(config, dep.program, messages)
 
     run_cmd = _build_run_cmd(
-        name, run, env, messages, source_dir, secret_env_file=secret_env_file,
+        name,
+        run,
+        env,
+        messages,
+        source_dir,
+        secret_env_file=secret_env_file,
         placeholders=ctx,
     )
     stop_cmd = _build_stop_cmd(name, run, source_dir)

core/src/castle_core/git.py

@@ -0,0 +1,159 @@
+"""Git working-copy status and sync for programs whose source is a git repo.
+
+Programs that declare a ``repo:`` URL are cloned once (``castle program clone``);
+this module lets a running castle *see how far behind* a working copy is and pull
+later updates. It is intentionally pull-only — it touches files on disk and never
+builds, applies, or restarts anything. Making the running artifact reflect the new
+code (rebuild a frontend, restart a service) stays an explicit, separate step via
+``castle apply`` / ``castle restart``.
+
+Plain ``git`` via subprocess (matching ``castle program clone``); no GitPython.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+# Bound network calls (fetch/pull) so an unreachable remote can't hang a request.
+_FETCH_TIMEOUT = 20.0
+_PULL_TIMEOUT = 60.0
+
+
+@dataclass
+class GitStatus:
+    """A program working copy's git state. ``ahead``/``behind`` are relative to the
+    upstream tracking branch and reflect the *last fetch* (``git_status(fetch=True)``
+    refreshes them). ``None`` counts mean "no upstream to compare against"."""
+
+    is_repo: bool
+    branch: str | None = None
+    upstream: str | None = None
+    dirty: bool = False
+    ahead: int | None = None
+    behind: int | None = None
+    detached: bool = False
+    error: str | None = None
+
+
+def _git(
+    source: Path, *args: str, timeout: float | None = None
+) -> subprocess.CompletedProcess[str]:
+    """Run ``git -C <source> <args>`` capturing text output."""
+    return subprocess.run(
+        ["git", "-C", str(source), *args],
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+    )
+
+
+def is_git_repo(source: Path | None) -> bool:
+    """True when ``source`` is inside a git working tree."""
+    if not source or not Path(source).is_dir():
+        return False
+    try:
+        r = _git(
+            Path(source), "rev-parse", "--is-inside-work-tree", timeout=_FETCH_TIMEOUT
+        )
+    except (OSError, subprocess.SubprocessError):
+        return False
+    return r.returncode == 0 and r.stdout.strip() == "true"
+
+
+def toplevel(source: Path | None) -> str | None:
+    """The absolute path of the git working copy ``source`` lives in, or None.
+
+    The natural identity of a *repo*: several programs whose sources share a
+    toplevel are the same working copy (a monorepo). Adopted single-program repos
+    are their own toplevel — the N=1 case."""
+    if not source or not Path(source).is_dir():
+        return None
+    try:
+        r = _git(Path(source), "rev-parse", "--show-toplevel", timeout=_FETCH_TIMEOUT)
+    except (OSError, subprocess.SubprocessError):
+        return None
+    return r.stdout.strip() or None if r.returncode == 0 else None
+
+
+def remote_url(source: Path | None) -> str | None:
+    """The ``origin`` remote URL of the working copy, or None (no remote)."""
+    if not is_git_repo(source):
+        return None
+    r = _git(Path(source), "remote", "get-url", "origin")  # type: ignore[arg-type]
+    return r.stdout.strip() or None if r.returncode == 0 else None
+
+
+def git_status(source: Path | None, fetch: bool = True) -> GitStatus:
+    """The working copy's branch/dirty/ahead/behind state.
+
+    ``fetch=True`` runs ``git fetch`` first (bounded, tolerant of an offline remote)
+    so ``behind`` reflects the real remote; on fetch failure the counts fall back to
+    the last-known values and ``error`` carries the reason. Never raises — a
+    non-repo returns ``GitStatus(is_repo=False)`` so callers can just hide the UI.
+    """
+    if not is_git_repo(source):
+        return GitStatus(is_repo=False)
+    src = Path(source)  # type: ignore[arg-type]
+    st = GitStatus(is_repo=True)
+
+    # Branch (or detached HEAD).
+    branch = _git(src, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
+    if branch == "HEAD":
+        st.detached = True
+    else:
+        st.branch = branch
+
+    # Dirty working tree (staged, unstaged, or untracked).
+    st.dirty = bool(_git(src, "status", "--porcelain").stdout.strip())
+
+    # Best-effort refresh from the remote; failure is non-fatal (offline, no remote).
+    if fetch:
+        try:
+            fr = _git(src, "fetch", "--quiet", timeout=_FETCH_TIMEOUT)
+            if fr.returncode != 0:
+                st.error = (fr.stderr or fr.stdout).strip() or "git fetch failed"
+        except subprocess.TimeoutExpired:
+            st.error = "git fetch timed out"
+        except (OSError, subprocess.SubprocessError) as e:
+            st.error = str(e)
+
+    # Upstream tracking branch, then the ahead/behind split against it.
+    up = _git(src, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}")
+    if up.returncode == 0 and up.stdout.strip():
+        st.upstream = up.stdout.strip()
+        counts = _git(src, "rev-list", "--left-right", "--count", "@{u}...HEAD")
+        if counts.returncode == 0:
+            parts = counts.stdout.split()
+            if len(parts) == 2:
+                st.behind, st.ahead = int(parts[0]), int(parts[1])
+    return st
+
+
+def head(source: Path | None) -> str | None:
+    """The working copy's current commit sha, or None if unavailable. Lets a caller
+    tell whether a ``pull`` actually advanced the tree (before != after)."""
+    if not is_git_repo(source):
+        return None
+    r = _git(Path(source), "rev-parse", "HEAD")  # type: ignore[arg-type]
+    return r.stdout.strip() or None if r.returncode == 0 else None
+
+
+def pull(source: Path | None) -> tuple[bool, str]:
+    """Fast-forward the working copy to its upstream (``git pull --ff-only``).
+
+    ``--ff-only`` is deliberate: it refuses to merge, so a dirty or diverged tree
+    fails cleanly with git's own message instead of creating a merge commit. Returns
+    ``(ok, combined_output)``.
+    """
+    if not is_git_repo(source):
+        return False, "not a git repository"
+    try:
+        r = _git(Path(source), "pull", "--ff-only", timeout=_PULL_TIMEOUT)  # type: ignore[arg-type]
+    except subprocess.TimeoutExpired:
+        return False, "git pull timed out"
+    except (OSError, subprocess.SubprocessError) as e:
+        return False, str(e)
+    out = (r.stdout + r.stderr).strip()
+    return r.returncode == 0, out

core/src/castle_core/manifest.py

@@ -283,6 +283,27 @@ class Capability(BaseModel):
     meta: dict[str, str] = Field(default_factory=dict)
 
 
+class Requirement(BaseModel):
+    """A precondition — something that must be true for a program/deployment to be
+    *functional*. The ``kind`` fixes both the meaning and how it's checked (there is
+    no separate purpose tag):
+
+    - ``system``     — a host package/binary must be installed (``ref`` = package).
+    - ``deployment`` — another deployment must exist/run (``ref`` = its name).
+
+    ``version`` is reserved for a future constraint (unused now). ``bind`` (for a
+    ``deployment`` requirement) names the env var castle projects the target's URL
+    into — env is derived *from* the requirement, never scraped back into it.
+
+    See docs/relationships.md. ``system_dependencies`` is the ``kind: system`` case.
+    """
+
+    kind: Literal["system", "deployment"]
+    ref: str
+    version: str | None = None
+    bind: str | None = None
+
+
 # ---------------------
 # Defaults
 # ---------------------
@@ -363,6 +384,10 @@ class ProgramSpec(BaseModel):
     # Per-program dev verb overrides (declared verbs override the stack default).
     commands: CommandsSpec | None = None
 
+    # `requires` is the general precondition relation (see docs/relationships.md).
+    # `system_dependencies` is kept as the `{kind: system}` alias/back-compat; both
+    # are merged when evaluating what a program requires.
+    requires: list[Requirement] = Field(default_factory=list)
     system_dependencies: list[str] = Field(default_factory=list)
     install_extras: list[str] = Field(default_factory=list)
     version: str | None = None
@@ -406,6 +431,9 @@ class DeploymentBase(BaseModel):
     )
     description: str | None = None
     defaults: DefaultsSpec | None = None
+    # Runtime preconditions (e.g. another deployment that must exist). See
+    # docs/relationships.md; merged with the program's `requires` when evaluated.
+    requires: list[Requirement] = Field(default_factory=list)
     # Declared on/off state. `castle apply` converges reality to this: enabled
     # deployments are activated (service started, tool installed, route served),
     # disabled ones are deactivated but kept in the catalog. This is *desired

core/src/castle_core/relations.py

@@ -0,0 +1,204 @@
+"""The relationship model — derived, never stored. See docs/relationships.md.
+
+Entities: **program**, **deployment**, **repo** (a repo is a git working copy;
+programs sharing a toplevel form a monorepo). One encoded relation, **`requires`**
+(a precondition, typed by ``kind``: ``system`` = must be installed, ``deployment``
+= must exist). Everything else — repos, env wiring, fan-in, and the predicates
+``functional?`` / ``fresh?`` / ``deployed?`` — is computed here on demand.
+
+Governing rule: *predicates are derived; we encode only the non-derivable.* So this
+module reads the encoded ``requires`` (plus ``system_dependencies`` as its
+``kind: system`` alias) and derives the rest. It does **not** scrape env for
+dependencies — env is generated *from* requirements, not the reverse.
+"""
+
+from __future__ import annotations
+
+import shutil
+from collections import Counter
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from castle_core import git
+from castle_core.config import CastleConfig
+from castle_core.manifest import Requirement
+
+
+@dataclass
+class Repo:
+    key: str  # url-safe slug (basename of the working copy)
+    path: str  # git toplevel
+    url: str | None
+    ref: str | None
+    programs: list[str]
+    deployments: list[str]
+    behind: int | None = None  # commits behind upstream (None = unknown/no upstream)
+    dirty: bool = False
+    fresh: bool | None = None  # derived: at latest and clean (None = not evaluated)
+
+    @property
+    def multi(self) -> bool:
+        """A monorepo — more than one program shares this working copy."""
+        return len(self.programs) > 1
+
+
+@dataclass
+class Edge:
+    src: str  # deployment name
+    dst: str  # target: a package (system) or another deployment
+    kind: str  # "system" | "deployment"
+    bind: str | None = None  # env var to project the target URL into (deployment)
+
+
+@dataclass
+class Node:
+    name: str  # deployment name
+    program: str | None
+    kind: str  # service|job|tool|static|reference
+    repo: str | None
+    depended_on_by: int  # distinct deployments that require this one (fan-in)
+    unmet: list[str] = field(default_factory=list)  # unsatisfied requirements
+    functional: bool = True  # derived: all requirements satisfied
+    fresh: bool | None = None  # derived: its repo is at latest + clean
+    deployed: bool | None = None  # derived: active in the registry (None = unknown)
+
+
+@dataclass
+class Model:
+    repos: list[Repo] = field(default_factory=list)
+    nodes: list[Node] = field(default_factory=list)
+    edges: list[Edge] = field(default_factory=list)
+
+
+def _program_of(name: str, dep: object) -> str:
+    return getattr(dep, "program", None) or name
+
+
+def _slug(name: str, used: set[str]) -> str:
+    base = name or "repo"
+    key, n = base, 2
+    while key in used:
+        key, n = f"{base}-{n}", n + 1
+    used.add(key)
+    return key
+
+
+def derive_repos(config: CastleConfig) -> dict[str, Repo]:
+    """Group programs by the git working copy their source lives in."""
+    by_top: dict[str, list[str]] = {}
+    for pname, prog in config.programs.items():
+        top = git.toplevel(prog.source) if prog.source else None
+        if top:
+            by_top.setdefault(top, []).append(pname)
+
+    used: set[str] = set()
+    repos: dict[str, Repo] = {}
+    for top, progs in sorted(by_top.items()):
+        progs = sorted(progs)
+        url = next(
+            (config.programs[p].repo for p in progs if config.programs[p].repo), None
+        ) or git.remote_url(Path(top))
+        ref = (
+            next(
+                (config.programs[p].ref for p in progs if config.programs[p].ref), None
+            )
+            or git.git_status(Path(top), fetch=False).branch
+        )
+        deps = sorted(
+            d for d, dep in config.deployments.items() if _program_of(d, dep) in progs
+        )
+        repos[_slug(Path(top).name, used)] = Repo("", top, url, ref, progs, deps)
+    for key, repo in repos.items():
+        repo.key = key
+    return repos
+
+
+def requirements_of(config: CastleConfig, dep_name: str) -> list[Requirement]:
+    """The full requirement set for a deployment: its own ``requires`` plus its
+    program's ``requires`` and ``system_dependencies`` (the ``kind: system`` alias),
+    de-duplicated by (kind, ref)."""
+    dep = config.deployments[dep_name]
+    prog = config.programs.get(_program_of(dep_name, dep))
+    reqs: list[Requirement] = list(getattr(dep, "requires", []) or [])
+    if prog:
+        reqs += list(prog.requires)
+        reqs += [
+            Requirement(kind="system", ref=pkg) for pkg in prog.system_dependencies
+        ]
+    seen: set[tuple[str, str]] = set()
+    out: list[Requirement] = []
+    for r in reqs:
+        if (r.kind, r.ref) not in seen:
+            seen.add((r.kind, r.ref))
+            out.append(r)
+    return out
+
+
+def _check(config: CastleConfig, req: Requirement) -> bool:
+    """Is a single requirement satisfied? (The check is fixed by its kind.)"""
+    if req.kind == "system":
+        return shutil.which(req.ref) is not None
+    if req.kind == "deployment":
+        return req.ref in config.deployments
+    return True
+
+
+def build_model(
+    config: CastleConfig,
+    check: bool = True,
+    active: set[str] | None = None,
+    freshness: bool = False,
+) -> Model:
+    """Compute the relationship model.
+
+    - ``check`` (default): evaluate ``functional?`` (unmet requirements) via a live
+      ``which`` / registry probe. ``check=False`` → pure structural model.
+    - ``active``: names of currently-active deployments → the ``deployed?``
+      predicate (left ``None`` when the caller has no runtime view).
+    - ``freshness``: also evaluate ``fresh?`` per repo (a ``git status``, no fetch —
+      last-known — so it stays a local, network-free probe over many repos)."""
+    from castle_core.manifest import kind_for
+
+    repos = derive_repos(config)
+    if freshness:
+        for repo in repos.values():
+            st = git.git_status(Path(repo.path), fetch=False)
+            repo.behind = st.behind
+            repo.dirty = st.dirty
+            repo.fresh = (st.behind == 0 or st.behind is None) and not st.dirty
+    repo_of = {p: key for key, r in repos.items() for p in r.programs}
+    fresh_of = {key: r.fresh for key, r in repos.items()}
+
+    edges: list[Edge] = []
+    for name in config.deployments:
+        for r in requirements_of(config, name):
+            edges.append(Edge(name, r.ref, r.kind, r.bind))
+
+    fan_in = Counter(e.dst for e in edges if e.kind == "deployment")
+
+    nodes: list[Node] = []
+    for name, dep in config.deployments.items():
+        unmet = (
+            [
+                f"{r.kind}:{r.ref}"
+                for r in requirements_of(config, name)
+                if not _check(config, r)
+            ]
+            if check
+            else []
+        )
+        repo_key = repo_of.get(_program_of(name, dep))
+        nodes.append(
+            Node(
+                name=name,
+                program=_program_of(name, dep),
+                kind=kind_for(dep),
+                repo=repo_key,
+                depended_on_by=fan_in.get(name, 0),
+                unmet=unmet,
+                functional=not unmet,
+                fresh=fresh_of.get(repo_key) if (freshness and repo_key) else None,
+                deployed=(name in active) if active is not None else None,
+            )
+        )
+    return Model(repos=list(repos.values()), nodes=nodes, edges=edges)

core/tests/test_git.py

@@ -0,0 +1,123 @@
+"""Tests for git working-copy status/sync (core/src/castle_core/git.py)."""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from castle_core import git as G
+
+# Identity so commits succeed without touching the user's global git config.
+_ENV = {
+    "GIT_AUTHOR_NAME": "t",
+    "GIT_AUTHOR_EMAIL": "t@t",
+    "GIT_COMMITTER_NAME": "t",
+    "GIT_COMMITTER_EMAIL": "t@t",
+    "GIT_CONFIG_GLOBAL": "/dev/null",
+    "GIT_CONFIG_SYSTEM": "/dev/null",
+}
+
+
+def _git(cwd: Path, *args: str) -> None:
+    subprocess.run(
+        ["git", "-C", str(cwd), *args],
+        check=True,
+        capture_output=True,
+        text=True,
+        env={**_base_env(), **_ENV},
+    )
+
+
+def _base_env() -> dict[str, str]:
+    import os
+
+    return dict(os.environ)
+
+
+def _commit(cwd: Path, fname: str, text: str) -> None:
+    (cwd / fname).write_text(text)
+    _git(cwd, "add", fname)
+    _git(cwd, "commit", "-m", f"add {fname}")
+
+
+@pytest.fixture
+def repos(tmp_path: Path):
+    """An `upstream` repo and a `work` clone of it (tracking upstream/main)."""
+    upstream = tmp_path / "upstream"
+    upstream.mkdir()
+    _git(upstream, "init", "-q", "-b", "main")
+    _commit(upstream, "a.txt", "one")
+    work = tmp_path / "work"
+    _git(tmp_path, "clone", "-q", str(upstream), str(work))
+    return upstream, work
+
+
+def test_non_repo_is_benign(tmp_path: Path) -> None:
+    assert G.is_git_repo(tmp_path / "nope") is False
+    st = G.git_status(tmp_path / "nope")
+    assert st.is_repo is False and st.branch is None
+    ok, out = G.pull(tmp_path / "nope")
+    assert ok is False and "not a git" in out
+
+
+def test_status_clean_and_up_to_date(repos) -> None:
+    _, work = repos
+    st = G.git_status(work, fetch=True)
+    assert st.is_repo and st.branch == "main"
+    assert st.dirty is False
+    assert st.behind == 0 and st.ahead == 0
+    assert st.upstream and st.upstream.endswith("main")
+
+
+def test_behind_then_pull_fast_forwards(repos) -> None:
+    upstream, work = repos
+    _commit(upstream, "b.txt", "two")  # advance the remote
+    st = G.git_status(work, fetch=True)
+    assert st.behind == 1 and st.ahead == 0
+
+    ok, out = G.pull(work)
+    assert ok is True, out
+    assert (work / "b.txt").exists()
+
+    after = G.git_status(work, fetch=True)
+    assert after.behind == 0 and after.dirty is False
+
+
+def test_conflicting_dirty_tree_blocks_pull(repos) -> None:
+    """A pull that would overwrite a locally-modified file is refused (ff-only never
+    merges), leaving the working copy untouched with git's own message."""
+    upstream, work = repos
+    _commit(upstream, "a.txt", "upstream change")  # remote touches a.txt...
+    (work / "a.txt").write_text("local uncommitted change")  # ...so does the work tree
+    assert G.git_status(work, fetch=False).dirty is True
+
+    ok, out = G.pull(work)
+    assert ok is False and out  # "local changes would be overwritten"
+    assert (work / "a.txt").read_text() == "local uncommitted change"
+
+
+def test_diverged_branch_blocks_ff_pull(repos) -> None:
+    """Local commits the remote doesn't have → --ff-only refuses (no merge commit)."""
+    upstream, work = repos
+    _commit(upstream, "b.txt", "remote two")
+    _commit(work, "c.txt", "local two")  # work now has a commit upstream lacks
+    st = G.git_status(work, fetch=True)
+    assert st.behind == 1 and st.ahead == 1
+
+    ok, out = G.pull(work)
+    assert ok is False and out
+
+
+def test_detached_head_reported(repos) -> None:
+    _, work = repos
+    head = subprocess.run(
+        ["git", "-C", str(work), "rev-parse", "HEAD"],
+        capture_output=True,
+        text=True,
+        env={**_base_env(), **_ENV},
+    ).stdout.strip()
+    _git(work, "checkout", "-q", head)
+    st = G.git_status(work, fetch=False)
+    assert st.detached is True and st.branch is None

core/tests/test_relations.py

@@ -0,0 +1,106 @@
+"""Tests for the relationship model (core/src/castle_core/relations.py)."""
+
+from __future__ import annotations
+
+import pytest
+
+import castle_core.config as C
+from castle_core import relations as R
+from castle_core.manifest import ProgramSpec, Requirement, SystemdDeployment
+
+
+def _dep(program: str) -> SystemdDeployment:
+    return SystemdDeployment.model_validate(
+        {
+            "manager": "systemd",
+            "program": program,
+            "run": {"launcher": "command", "argv": [program]},
+        }
+    )
+
+
+def _cfg(programs: dict, deployments: dict) -> C.CastleConfig:
+    return C.CastleConfig(
+        root=None,
+        gateway=C.GatewayConfig(port=9000),
+        repo=None,
+        programs=programs,
+        deployments=deployments,
+    )
+
+
+def test_system_dependencies_is_the_system_requirement_alias() -> None:
+    """`system_dependencies` surfaces as a {kind: system} requirement."""
+    cfg = _cfg(
+        {"t": ProgramSpec(id="t", system_dependencies=["pandoc"])}, {"t": _dep("t")}
+    )
+    reqs = R.requirements_of(cfg, "t")
+    assert [(r.kind, r.ref) for r in reqs] == [("system", "pandoc")]
+
+
+def test_requirements_merge_program_and_deployment_deduped() -> None:
+    prog = ProgramSpec(
+        id="web",
+        system_dependencies=["pandoc"],
+        requires=[Requirement(kind="deployment", ref="api", bind="API_URL")],
+    )
+    dep = SystemdDeployment.model_validate(
+        {
+            "manager": "systemd",
+            "program": "web",
+            "run": {"launcher": "command", "argv": ["web"]},
+            "requires": [{"kind": "system", "ref": "pandoc"}],
+        }  # dup of program's
+    )
+    cfg = _cfg(
+        {"web": prog, "api": ProgramSpec(id="api")}, {"web": dep, "api": _dep("api")}
+    )
+    kinds = {(r.kind, r.ref) for r in R.requirements_of(cfg, "web")}
+    assert kinds == {("system", "pandoc"), ("deployment", "api")}  # deduped
+
+
+def test_deployment_edge_carries_bind_and_counts_fan_in() -> None:
+    """A {kind: deployment} requirement becomes an edge (with bind), and the target's
+    fan-in is the count of distinct dependents."""
+    consumer = ProgramSpec(
+        id="web", requires=[Requirement(kind="deployment", ref="api", bind="API_URL")]
+    )
+    consumer2 = ProgramSpec(
+        id="cli", requires=[Requirement(kind="deployment", ref="api")]
+    )
+    cfg = _cfg(
+        {"web": consumer, "cli": consumer2, "api": ProgramSpec(id="api")},
+        {"web": _dep("web"), "cli": _dep("cli"), "api": _dep("api")},
+    )
+    m = R.build_model(cfg, check=False)
+    edge = next(e for e in m.edges if e.src == "web" and e.dst == "api")
+    assert edge.kind == "deployment" and edge.bind == "API_URL"
+    api = next(n for n in m.nodes if n.name == "api")
+    assert api.depended_on_by == 2  # web + cli
+
+
+def test_functional_predicate_reports_unmet(monkeypatch: pytest.MonkeyPatch) -> None:
+    """`functional?` is derived: a missing system package is unmet; a present
+    deployment requirement is satisfied."""
+    prog = ProgramSpec(
+        id="web",
+        system_dependencies=["pandoc"],
+        requires=[Requirement(kind="deployment", ref="api")],
+    )
+    cfg = _cfg(
+        {"web": prog, "api": ProgramSpec(id="api")},
+        {"web": _dep("web"), "api": _dep("api")},
+    )
+    monkeypatch.setattr(R.shutil, "which", lambda _: None)  # nothing installed
+    m = R.build_model(cfg, check=True)
+    web = next(n for n in m.nodes if n.name == "web")
+    assert web.unmet == ["system:pandoc"]  # deployment:api exists → satisfied
+    assert web.functional is False
+
+
+def test_missing_deployment_requirement_is_unmet() -> None:
+    prog = ProgramSpec(id="web", requires=[Requirement(kind="deployment", ref="ghost")])
+    cfg = _cfg({"web": prog}, {"web": _dep("web")})
+    m = R.build_model(cfg, check=True)
+    web = next(n for n in m.nodes if n.name == "web")
+    assert web.unmet == ["deployment:ghost"] and web.functional is False