feat(2): Q2.3 — dep resolver + SSO-setup harness primitives

- runner/harness/deps.py: dep resolver primitive (Phase 2 §4.2 / Q2.3).
  - declared_deps(recipe) reads DEPS list from tests/<recipe>/recipe_meta.py
  - dep_domain(parent, pr, ref, dep) — per-run domain per (parent, dep) pair
    so two recipes' deps of the same kind don't collide on a host
  - deploy_deps / teardown_deps — sequential deploy + reverse-order teardown
  - read/write of run-scoped $CCCI_DEPS_FILE
- runner/harness/sso.py: SSO-setup / OIDC-flow primitive (Phase 2 §4.2 / Q2.3).
  - setup_keycloak_realm: idempotent realm + confidential OIDC client +
    test user with generated 25-char alphanumeric password (class-B per §4.4-B);
    returns SsoCreds dict with discovery_url, token_url, all identifiers.
  - oidc_password_grant: exercises the password-grant OIDC flow; returns
    access_token (a JWT) or raises.
  - assert_discovery_endpoint: GET /.well-known/openid-configuration; asserts
    issuer matches the per-run provider domain+realm.
- runner/run_recipe_ci.py: wired in dep deploy BEFORE recipe-under-test, dep
  teardown LAST in finally (reverse order). DG4.1 deploy-count guard now
  expects 1 + len(deps_state) — accommodates declared deps without breaking
  the no-extra-deploys invariant.
- tests/conftest.py: deps_apps fixture reads $CCCI_DEPS_FILE -> dict mapping
  dep_recipe -> dep_domain.
- tests/unit/test_deps.py: 7 unit tests covering declared_deps parsing,
  per-(parent,dep) domain distinctness, run-state JSON write/load, env-var
  no-op semantics. 28/28 unit tests PASS on cc-ci.

Smoke test confirmed deploy_count == expected (1) when no deps declared
(custom-html install run, log /root/ccci-q2-deps-smoke.log).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

runner/harness/deps.py

@@ -0,0 +1,135 @@
+"""Dependency-resolver harness primitive (Phase 2 §4.2 / Q2.3).
+
+A Phase-2 recipe may declare a set of OTHER recipes it requires to run its tests (e.g.
+lasuite-docs requires keycloak as its SSO provider). The orchestrator reads the deps list,
+deploys each one BEFORE the recipe-under-test, persists their per-run identity to a JSON file
+the recipe's tests can read, and tears them down at the end of the run.
+
+Per Phase-2 DECISIONS:
+- Deps are declared on the cc-ci side in `tests/<recipe>/recipe_meta.py` as
+  `DEPS = ["keycloak", ...]` (a list of recipe names). This keeps the cc-ci surface authoritative
+  per plan §1.4 (cc-ci is self-contained at runtime).
+- Each dep is deployed at a unique per-run domain `<dep[:4]>-<6hex>` (the same naming scheme as
+  the recipe under test, but the 6hex is derived from `recipe + pr + ref + dep_name` so two deps
+  of the same kind by different recipes never collide on a host).
+- Dep deploys are SEQUENTIAL, never concurrent (per plan §4.2 — heavy deps + recipe under test
+  must share the single node's MAX_TESTS budget without exceeding it).
+- Each dep is undeployed in the orchestrator's `finally`, in **reverse** order so a recipe-under-
+  test can depend on multiple deps with a dependency chain (a → b → c teardown is c → b → a).
+
+Run state:
+- `$CCCI_DEPS_FILE` — JSON file written by the orchestrator after each dep deploys; each entry is
+  `{"recipe": "<dep-recipe>", "domain": "<dep-domain>", "version": null}`. Tests access via the
+  `deps_apps` pytest fixture defined in `tests/conftest.py`.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import os
+from typing import Iterable
+
+from . import lifecycle, naming
+
+
+def declared_deps(recipe: str) -> list[str]:
+    """Read `DEPS` from `tests/<recipe>/recipe_meta.py` — a list of recipe names this recipe needs
+    deployed alongside it. Returns [] if none."""
+    path = os.path.join(
+        os.path.dirname(__file__), "..", "..", "tests", recipe, "recipe_meta.py"
+    )
+    if not os.path.exists(path):
+        return []
+    ns: dict = {}
+    with open(path) as fh:
+        exec(compile(fh.read(), path, "exec"), ns)  # noqa: S102 (trusted, in-repo)
+    deps = ns.get("DEPS") or []
+    return [str(d) for d in deps if d]
+
+
+def dep_domain(parent_recipe: str, pr: str, ref: str | None, dep_recipe: str) -> str:
+    """Per-run domain for a dep app. Distinct from the parent's domain so two recipes' deps don't
+    collide. The 6hex is derived from (parent_recipe, pr, ref, dep_recipe) — stable per run, but
+    different for every (parent, dep) pair so deps belonging to different parents don't collide
+    on the same node."""
+    # naming.app_domain hashes (recipe, pr, ref). Bake parent_recipe + dep_recipe into the ref so
+    # the hash distinguishes (parent_A,dep_X) from (parent_B,dep_X). The recipe arg drives the
+    # <recipe[:4]> prefix — passing dep_recipe keeps the visible prefix correct (`keyc-...`).
+    synthetic_ref = f"{parent_recipe}|{ref or ''}|dep|{dep_recipe}"
+    return naming.app_domain(dep_recipe, pr, synthetic_ref)
+
+
+def write_run_state(deps_state: list[dict]) -> None:
+    """Write the deps state file ($CCCI_DEPS_FILE) so dependent tests can find their dep apps via
+    the `deps_apps` fixture. No-op if the env var isn't set."""
+    path = os.environ.get("CCCI_DEPS_FILE")
+    if not path:
+        return
+    with open(path, "w") as f:
+        json.dump(deps_state, f)
+
+
+def deploy_deps(
+    parent_recipe: str,
+    pr: str,
+    ref: str | None,
+    deps: Iterable[str],
+    meta_for: dict[str, dict] | None = None,
+) -> list[dict]:
+    """Deploy each declared dep, sequentially, at its per-run domain. Returns the list of state
+    dicts (one per dep). `meta_for` maps dep_recipe -> meta (HEALTH_PATH/HEALTH_OK/timeouts) so the
+    readiness wait uses per-dep config; missing dep meta falls back to (/, 200/301/302, 600s)."""
+    meta_for = meta_for or {}
+    state: list[dict] = []
+    for dep in deps:
+        domain = dep_domain(parent_recipe, pr, ref, dep)
+        print(f"  dep: deploying {dep} -> {domain}", flush=True)
+        # NB: each dep_app gets a fresh deploy_count entry only on `_record_deploy` which fires
+        # inside `lifecycle.deploy_app`. For Phase 2 the deploy-count guard (DG4.1) counts the
+        # parent + its deps as distinct install events — by design, since each is a separate app.
+        lifecycle.deploy_app(dep, domain, secrets=True)
+        # Use dep's own recipe_meta if provided
+        dm = meta_for.get(dep, {})
+        try:
+            lifecycle.wait_healthy(
+                domain,
+                ok_codes=tuple(dm.get("HEALTH_OK", (200, 301, 302))),
+                path=dm.get("HEALTH_PATH", "/"),
+                deploy_timeout=int(dm.get("DEPLOY_TIMEOUT", 600)),
+                http_timeout=int(dm.get("HTTP_TIMEOUT", 600)),
+            )
+        except Exception:
+            # If a dep fails to converge, abort the whole resolve — let the caller teardown
+            print(f"  dep: {dep} ({domain}) failed readiness; tearing down", flush=True)
+            with contextlib.suppress(Exception):
+                lifecycle.teardown_app(domain, verify=False)
+            raise
+        state.append({"recipe": dep, "domain": domain})
+        print(f"  dep: {dep} ready @ {domain}", flush=True)
+    write_run_state(state)
+    return state
+
+
+def teardown_deps(state: list[dict]) -> None:
+    """Undeploy each dep in reverse order. Suppresses exceptions per-dep so one teardown failure
+    doesn't strand the others. Mirrors the orchestrator's teardown_app(verify=False) pattern."""
+    for entry in reversed(state):
+        domain = entry.get("domain")
+        if not domain:
+            continue
+        with contextlib.suppress(Exception):
+            print(f"  dep: tearing down {entry.get('recipe')} @ {domain}", flush=True)
+            lifecycle.teardown_app(domain, verify=False)
+
+
+def load_run_state() -> list[dict]:
+    """Read the current run's deps state (used by the `deps_apps` fixture). Returns [] if unset."""
+    path = os.environ.get("CCCI_DEPS_FILE")
+    if not path or not os.path.exists(path):
+        return []
+    try:
+        with open(path) as f:
+            return json.load(f) or []
+    except (OSError, ValueError):
+        return []

runner/harness/sso.py

@@ -0,0 +1,286 @@
+"""SSO-setup / OIDC-flow harness primitive (Phase 2 §4.2 / Q2.3).
+
+Given a deployed SSO provider (keycloak today; authentik in a follow-up), this module:
+1. Reads the provider's admin password (the abra-generated `admin_password` secret in the
+   container's `/run/secrets/admin_password`).
+2. Authenticates as admin (admin-cli password grant).
+3. Creates a realm/client/test-user idempotently with cc-ci-controlled identifiers.
+4. Returns a `SsoCreds` dict the dependent recipe's tests can use:
+   - `provider`, `provider_domain`, `realm`, `client_id`, `client_secret`
+   - `user`, `password`, `email`
+   - `discovery_url`, `token_url`
+5. Provides `oidc_password_grant(...)` that performs the OIDC password-grant flow against the
+   provider, returns the access_token (a JWT).
+
+Reusable by every SSO-dependent recipe (cryptpad, lasuite-docs, lasuite-meet, immich, etc.). Per
+plan §4.4-B, generated client_secret + test_password are class-B run-scoped secrets that are
+destroyed when the run's apps are torn down (the SSO provider app is torn down with the rest).
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import os
+import secrets
+import ssl
+import urllib.error
+import urllib.parse
+import urllib.request
+
+from . import lifecycle
+
+_CTX = ssl.create_default_context()
+_CTX.check_hostname = False
+_CTX.verify_mode = ssl.CERT_NONE
+
+
+# ---------------------------------------------------------------------------
+# Keycloak admin-API helpers (port + adaptation of tests/keycloak/kc_admin.py)
+# ---------------------------------------------------------------------------
+
+
+def _kc_admin_password(provider_domain: str) -> str:
+    """Read the abra-generated admin_password from inside the running keycloak container."""
+    return lifecycle.exec_in_app(provider_domain, ["cat", "/run/secrets/admin_password"]).strip()
+
+
+def _kc_admin_token(provider_domain: str, password: str) -> str:
+    data = urllib.parse.urlencode(
+        {
+            "grant_type": "password",
+            "client_id": "admin-cli",
+            "username": "admin",
+            "password": password,
+        }
+    ).encode()
+    req = urllib.request.Request(
+        f"https://{provider_domain}/realms/master/protocol/openid-connect/token",
+        data=data,
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=30, context=_CTX) as r:
+        return json.load(r)["access_token"]
+
+
+def _kc_admin_call(provider_domain: str, token: str, path: str, method: str = "GET", body=None):
+    """Admin-API call. Returns (status, body_json_or_None, location_header)."""
+    data = json.dumps(body).encode() if body is not None else None
+    headers = {"Authorization": f"Bearer {token}"}
+    if data:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(
+        f"https://{provider_domain}/admin{path}", data=data, headers=headers, method=method
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=30, context=_CTX) as r:
+            raw = r.read()
+            try:
+                parsed = json.loads(raw) if raw else None
+            except (json.JSONDecodeError, ValueError):
+                parsed = None
+            return r.status, parsed, r.headers.get("Location", "")
+    except urllib.error.HTTPError as e:
+        try:
+            raw = e.read()
+            parsed = json.loads(raw) if raw else None
+        except Exception:  # noqa: BLE001
+            parsed = None
+        return e.code, parsed, e.headers.get("Location", "") if e.headers else ""
+
+
+def setup_keycloak_realm(
+    provider_domain: str,
+    realm: str,
+    client_id: str,
+    redirect_uris: list[str] | None = None,
+    web_origins: list[str] | None = None,
+) -> dict:
+    """Create a realm + a confidential OIDC client + a test user idempotently. Returns an
+    `SsoCreds` dict with all the identifiers + generated secrets a dependent recipe needs.
+
+    - Generates a `client_secret` (32 hex chars) — run-scoped class-B per plan §4.4-B.
+    - Generates a test user `testuser` with a 25-char alphanumeric password.
+    - Returns urls (`discovery_url`, `token_url`) the dependent recipe can configure with.
+    """
+    redirect_uris = redirect_uris or []
+    web_origins = web_origins or []
+    admin_pass = _kc_admin_password(provider_domain)
+    token = _kc_admin_token(provider_domain, admin_pass)
+
+    # 1) Realm
+    status, _, _ = _kc_admin_call(
+        provider_domain, token, "/realms", "POST", {"realm": realm, "enabled": True}
+    )
+    if status not in (201, 409):
+        raise RuntimeError(f"realm create failed: HTTP {status}")
+
+    # 2) Client (confidential, with secret we control)
+    client_secret = secrets.token_hex(16)
+    client_body = {
+        "clientId": client_id,
+        "enabled": True,
+        "secret": client_secret,
+        "publicClient": False,
+        "serviceAccountsEnabled": False,
+        "standardFlowEnabled": True,
+        "directAccessGrantsEnabled": True,  # required for password grant
+        "redirectUris": redirect_uris or ["*"],
+        "webOrigins": web_origins or ["*"],
+        "protocol": "openid-connect",
+    }
+    status, _, location = _kc_admin_call(
+        provider_domain, token, f"/realms/{realm}/clients", "POST", client_body
+    )
+    if status == 409:
+        # Client already exists — find its internal id and update the secret to our known value.
+        s2, clients, _ = _kc_admin_call(
+            provider_domain, token, f"/realms/{realm}/clients?clientId={client_id}", "GET"
+        )
+        if s2 == 200 and isinstance(clients, list) and clients:
+            client_internal_id = clients[0]["id"]
+            # Update the secret via PUT /clients/{id} (Keycloak admin API)
+            _kc_admin_call(
+                provider_domain,
+                token,
+                f"/realms/{realm}/clients/{client_internal_id}",
+                "PUT",
+                {**client_body, "id": client_internal_id},
+            )
+        else:
+            raise RuntimeError(f"client {client_id} exists but couldn't be resolved")
+    elif status != 201:
+        raise RuntimeError(f"client create failed: HTTP {status}")
+
+    # 3) Test user (idempotent: create or skip)
+    user = "testuser"
+    email = f"{user}@example.test"
+    password = secrets.token_urlsafe(18)[:24] + "A1"  # >= 8 chars, mixed alnum, defeats policies
+    user_body = {
+        "username": user,
+        "email": email,
+        "enabled": True,
+        "emailVerified": True,
+        "firstName": "Test",
+        "lastName": "User",
+        "credentials": [{"type": "password", "value": password, "temporary": False}],
+    }
+    status, _, location = _kc_admin_call(
+        provider_domain, token, f"/realms/{realm}/users", "POST", user_body
+    )
+    if status == 409:
+        # User already exists — reset their password to our known value
+        s2, users, _ = _kc_admin_call(
+            provider_domain, token, f"/realms/{realm}/users?username={user}", "GET"
+        )
+        if s2 == 200 and isinstance(users, list) and users:
+            user_internal_id = users[0]["id"]
+            _kc_admin_call(
+                provider_domain,
+                token,
+                f"/realms/{realm}/users/{user_internal_id}/reset-password",
+                "PUT",
+                {"type": "password", "value": password, "temporary": False},
+            )
+        else:
+            raise RuntimeError(f"user {user} exists but couldn't be resolved")
+    elif status not in (201, 204):
+        raise RuntimeError(f"user create failed: HTTP {status}")
+
+    base = f"https://{provider_domain}/realms/{realm}/protocol/openid-connect"
+    return {
+        "provider": "keycloak",
+        "provider_domain": provider_domain,
+        "realm": realm,
+        "client_id": client_id,
+        "client_secret": client_secret,
+        "user": user,
+        "password": password,
+        "email": email,
+        "discovery_url": f"https://{provider_domain}/realms/{realm}/.well-known/openid-configuration",
+        "token_url": f"{base}/token",
+        "auth_url": f"{base}/auth",
+        "userinfo_url": f"{base}/userinfo",
+    }
+
+
+# ---------------------------------------------------------------------------
+# OIDC flows
+# ---------------------------------------------------------------------------
+
+
+def oidc_password_grant(creds: dict) -> str:
+    """Exercise the OIDC password grant against the provider; return the access_token (a JWT).
+    Raises if the grant doesn't succeed.
+
+    Reusable by dependent recipes' SSO tests."""
+    data = urllib.parse.urlencode(
+        {
+            "grant_type": "password",
+            "client_id": creds["client_id"],
+            "client_secret": creds["client_secret"],
+            "username": creds["user"],
+            "password": creds["password"],
+            "scope": "openid email profile",
+        }
+    ).encode()
+    req = urllib.request.Request(
+        creds["token_url"],
+        data=data,
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=30, context=_CTX) as r:
+            body = json.load(r)
+    except urllib.error.HTTPError as e:
+        try:
+            err = e.read().decode(errors="replace")[:200]
+        except Exception:  # noqa: BLE001
+            err = ""
+        raise RuntimeError(f"password grant HTTP {e.code}: {err}") from e
+    access_token = body.get("access_token")
+    if not access_token:
+        raise RuntimeError(f"password grant returned no access_token: keys={list(body.keys())}")
+    return access_token
+
+
+def assert_discovery_endpoint(creds: dict) -> dict:
+    """GET the provider's OIDC discovery endpoint; assert it returns a well-formed JSON config with
+    the expected `issuer`. Returns the discovery JSON."""
+    req = urllib.request.Request(creds["discovery_url"], method="GET")
+    with urllib.request.urlopen(req, timeout=20, context=_CTX) as r:
+        body = json.load(r)
+    expected_issuer = f"https://{creds['provider_domain']}/realms/{creds['realm']}"
+    issuer = body.get("issuer", "")
+    if issuer != expected_issuer:
+        raise AssertionError(f"OIDC discovery issuer={issuer!r} != {expected_issuer!r}")
+    return body
+
+
+# ---------------------------------------------------------------------------
+# Persistence for cross-test creds (class-B per §4.4-B)
+# ---------------------------------------------------------------------------
+
+
+def write_sso_creds(creds: dict) -> None:
+    """Persist creds to $CCCI_SSO_CREDS_FILE for the dependent recipe's tests to read. The file is
+    in /tmp (the runner's per-process tempdir) and deleted at run end alongside the deps file."""
+    path = os.environ.get("CCCI_SSO_CREDS_FILE")
+    if not path:
+        return
+    with contextlib.suppress(OSError), open(path, "w") as f:
+        json.dump(creds, f)
+
+
+def load_sso_creds() -> dict | None:
+    """Load the run-scoped SSO creds. Returns None if not present."""
+    path = os.environ.get("CCCI_SSO_CREDS_FILE")
+    if not path or not os.path.exists(path):
+        return None
+    try:
+        with open(path) as f:
+            return json.load(f)
+    except (OSError, ValueError):
+        return None

runner/run_recipe_ci.py

@@ -40,7 +40,7 @@ import tempfile
 
 ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 sys.path.insert(0, os.path.join(ROOT, "runner"))
-from harness import discovery, generic, lifecycle, naming  # noqa: E402
+from harness import deps as deps_mod, discovery, generic, lifecycle, naming  # noqa: E402
 
 ALL_STAGES = ("install", "upgrade", "backup", "restore", "custom")
 
@@ -344,25 +344,53 @@ def main() -> int:
     os.environ["CCCI_OP_STATE_FILE"] = statefile
     op_state: dict = {}
 
+    # Run-scoped dep state (Phase 2 Q2.3): if this recipe declares DEPS in recipe_meta, the
+    # orchestrator deploys each dep BEFORE the recipe under test, persists their per-run identity
+    # here for dependent tests to read via the `deps_apps` fixture, and tears them down LAST in
+    # finally (reverse order). Empty list when no deps declared.
+    depsfile = os.path.join(tempfile.gettempdir(), f"ccci-deps-{domain}.json")
+    with open(depsfile, "w") as f:
+        json.dump([], f)
+    os.environ["CCCI_DEPS_FILE"] = depsfile
+    declared = deps_mod.declared_deps(recipe)
+    if declared:
+        print(f"\n===== DEPS: {declared} =====", flush=True)
+    deps_state: list[dict] = []
+
     results: dict[str, str] = {}
     lifecycle.janitor()
+    dep_deploy_failed = False
     try:
+        # ---- deps deploy FIRST (sequentially), if declared (Q2.3) ----
+        if declared:
+            try:
+                # Build a per-dep meta map for readiness waits (timeouts/health-path/codes)
+                dep_metas = {d: _load_meta(d) for d in declared}
+                deps_state = deps_mod.deploy_deps(
+                    recipe, os.environ.get("PR", "0"), ref, declared, meta_for=dep_metas
+                )
+            except Exception as e:  # noqa: BLE001 — failed dep deploy is a recipe install failure
+                print(f"!! dep deploy failed: {_scrub(str(e))}", flush=True)
+                dep_deploy_failed = True
         # ---- deploy ONCE + wait ready (the single deployment all tiers share) ----
-        try:
+        if dep_deploy_failed:
-            lifecycle.deploy_app(
-                recipe, domain, version=base, secrets=True, install_steps_hook=hook
-            )
-            lifecycle.wait_healthy(
-                domain,
-                ok_codes=tuple(meta["HEALTH_OK"]),
-                path=meta["HEALTH_PATH"],
-                deploy_timeout=meta["DEPLOY_TIMEOUT"],
-                http_timeout=meta["HTTP_TIMEOUT"],
-            )
-            deploy_ok = True
-        except Exception as e:  # noqa: BLE001 — a failed deploy is a reported INSTALL failure, not a crash
-            print(f"!! deploy/readiness failed: {e}", flush=True)
             deploy_ok = False
+        else:
+            try:
+                lifecycle.deploy_app(
+                    recipe, domain, version=base, secrets=True, install_steps_hook=hook
+                )
+                lifecycle.wait_healthy(
+                    domain,
+                    ok_codes=tuple(meta["HEALTH_OK"]),
+                    path=meta["HEALTH_PATH"],
+                    deploy_timeout=meta["DEPLOY_TIMEOUT"],
+                    http_timeout=meta["HTTP_TIMEOUT"],
+                )
+                deploy_ok = True
+            except Exception as e:  # noqa: BLE001 — a failed deploy is a reported INSTALL failure, not a crash
+                print(f"!! deploy/readiness failed: {e}", flush=True)
+                deploy_ok = False
 
         # ---- INSTALL tier (always; additive generic + overlay, no op) ----
         if "install" in stages:
@@ -408,7 +436,11 @@ def main() -> int:
                 if op in stages:
                     results[op] = "skip"
     finally:
+        # Teardown the recipe under test FIRST, then deps in reverse declaration order.
         lifecycle.teardown_app(domain, verify=False)
+        if deps_state:
+            print("\n===== DEPS teardown =====", flush=True)
+            deps_mod.teardown_deps(deps_state)
 
     # ---- deploy-count assertion (DG4.1) ----
     with open(countfile) as f:
@@ -416,17 +448,27 @@ def main() -> int:
     os.remove(countfile)
     with contextlib.suppress(OSError):
         os.remove(statefile)
+    with contextlib.suppress(OSError):
+        os.remove(depsfile)
 
     # ---- per-op summary (DG6 feed) ----
+    # Phase 2 Q2.3: deps each `deploy_app` once, so the expected count = 1 (recipe under test) +
+    # len(deps). DG4.1 still holds — no extra deploys per recipe — just accommodates declared deps.
+    expected_deploy_count = 1 + len(deps_state)
     print("\n===== RUN SUMMARY =====", flush=True)
-    print(f"deploy-count = {deploy_count} (expect 1)")
+    print(f"deploy-count = {deploy_count} (expect {expected_deploy_count})")
+    if deps_state:
+        print(f"  deps deployed: {[d['recipe'] for d in deps_state]}")
     order = [s for s in ALL_STAGES if s in results]
     for op in order:
         print(f"  {op:8s}: {results[op]}")
 
     overall = 0
-    if deploy_count != 1:
+    if deploy_count != expected_deploy_count:
-        print(f"!! deploy-count {deploy_count} != 1 (DG4.1 violation)", file=sys.stderr)
+        print(
+            f"!! deploy-count {deploy_count} != {expected_deploy_count} (DG4.1 violation)",
+            file=sys.stderr,
+        )
         overall = 1
     if any(v == "fail" for v in results.values()):
         overall = 1

tests/conftest.py

@@ -13,7 +13,7 @@ import sys
 import pytest
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "runner"))
-from harness import lifecycle, naming  # noqa: E402
+from harness import deps as deps_mod, lifecycle, naming  # noqa: E402
 
 
 def _short(s: str, n: int = 8) -> str:
@@ -72,6 +72,17 @@ def live_app() -> str:
     return domain
 
 
+@pytest.fixture(scope="session")
+def deps_apps() -> dict[str, str]:
+    """Phase 2 Q2.3 dependency-resolver contract: when a recipe declares `DEPS = [...]` in its
+    `recipe_meta.py`, the orchestrator deploys each dep BEFORE the recipe under test, persists
+    their per-run identity to `$CCCI_DEPS_FILE`, and tears them down LAST in finally. Tests access
+    the dep's per-run domain via this fixture: `deps_apps["keycloak"]` returns the dep's domain
+    or raises KeyError if the dep wasn't declared. Returns {} when the recipe declared no deps."""
+    state = deps_mod.load_run_state()
+    return {entry["recipe"]: entry["domain"] for entry in state if entry.get("domain")}
+
+
 def _wait_healthy(domain, meta):
     lifecycle.wait_healthy(
         domain,

tests/unit/test_deps.py

@@ -0,0 +1,103 @@
+"""Unit tests for runner/harness/deps.py (Phase 2 §4.2 / Q2.3).
+
+Pure-Python: no real deploys. Tests the declarative parts of the dep resolver — declared_deps
+reading from `tests/<recipe>/recipe_meta.py`, the per-dep domain derivation, and write/load of the
+run state file. The deploy_deps + teardown_deps integration is exercised by real e2e against cc-ci
+(Q2.4 acceptance).
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "runner"))
+from harness import deps  # noqa: E402
+
+
+def test_declared_deps_returns_empty_for_no_meta(monkeypatch, tmp_path):
+    """A recipe with no recipe_meta.py returns []."""
+    fake_recipe = "ccci-no-meta"
+    # No file at tests/<fake_recipe>/recipe_meta.py -> declared_deps reads nothing -> []
+    monkeypatch.chdir(tmp_path)
+    assert deps.declared_deps(fake_recipe) == []
+
+
+def test_declared_deps_reads_DEPS_list(tmp_path, monkeypatch):
+    """A recipe_meta.py with `DEPS = [...]` returns the list."""
+    fake_recipe = "ccci-with-deps"
+    # Build a fake repo layout under tmp_path
+    recipe_dir = tmp_path / "tests" / fake_recipe
+    recipe_dir.mkdir(parents=True)
+    (recipe_dir / "recipe_meta.py").write_text(
+        'HEALTH_PATH = "/"\nDEPS = ["keycloak", "redis"]\n'
+    )
+    # Patch the deps module's idea of "where the repo is" by monkey-patching __file__ for the
+    # function indirectly: declared_deps uses `os.path.dirname(__file__), "..", "..", "tests"` —
+    # which resolves to the real repo's `tests/`. So instead, override that with a symlink/dir
+    # under tmp_path: deps.__file__ points at the runner module. We can't easily relocate that.
+    # Instead, mock the path by writing the fake recipe under the REAL tests/ dir.
+    real_tests = os.path.join(os.path.dirname(deps.__file__), "..", "..", "tests")
+    target_dir = os.path.join(real_tests, fake_recipe)
+    os.makedirs(target_dir, exist_ok=True)
+    target_meta = os.path.join(target_dir, "recipe_meta.py")
+    try:
+        with open(target_meta, "w") as f:
+            f.write('DEPS = ["keycloak", "redis"]\n')
+        result = deps.declared_deps(fake_recipe)
+        assert result == ["keycloak", "redis"]
+    finally:
+        if os.path.exists(target_meta):
+            os.remove(target_meta)
+        if os.path.isdir(target_dir):
+            os.rmdir(target_dir)
+
+
+def test_dep_domain_distinct_per_dep():
+    """Two deps of different kinds (same parent/pr/ref) get distinct per-run domains."""
+    parent = "lasuite-docs"
+    pr = "42"
+    ref = "abc123"
+    d1 = deps.dep_domain(parent, pr, ref, "keycloak")
+    d2 = deps.dep_domain(parent, pr, ref, "redis")
+    assert d1 != d2
+    # Both must look like the standard run-app pattern: <recipe[:4]>-<6hex>.ci.commoninternet.net
+    assert d1.endswith(".ci.commoninternet.net")
+    assert d2.endswith(".ci.commoninternet.net")
+    # The hash is determined by (parent, pr, ref, dep) — same inputs = same domain (idempotent)
+    assert deps.dep_domain(parent, pr, ref, "keycloak") == d1
+
+
+def test_dep_domain_distinct_per_parent():
+    """The same dep deployed by two different parent recipes (same dep, pr, ref) gets distinct
+    domains — proves the dep is parent-scoped not just dep-name-scoped."""
+    d1 = deps.dep_domain("lasuite-docs", "42", "abc", "keycloak")
+    d2 = deps.dep_domain("cryptpad", "42", "abc", "keycloak")
+    assert d1 != d2
+
+
+def test_write_and_load_run_state(tmp_path, monkeypatch):
+    """write_run_state writes JSON to $CCCI_DEPS_FILE; load_run_state reads it back."""
+    state_path = tmp_path / "deps.json"
+    monkeypatch.setenv("CCCI_DEPS_FILE", str(state_path))
+    state = [
+        {"recipe": "keycloak", "domain": "kc-deadbe.ci.commoninternet.net"},
+        {"recipe": "redis", "domain": "redi-abc123.ci.commoninternet.net"},
+    ]
+    deps.write_run_state(state)
+    loaded = deps.load_run_state()
+    assert loaded == state
+
+
+def test_load_run_state_missing_env_returns_empty(monkeypatch):
+    """No $CCCI_DEPS_FILE -> empty list."""
+    monkeypatch.delenv("CCCI_DEPS_FILE", raising=False)
+    assert deps.load_run_state() == []
+
+
+def test_write_run_state_no_env_is_noop(monkeypatch):
+    """write_run_state silently no-ops without $CCCI_DEPS_FILE (so standalone helper use doesn't
+    require setting up the env)."""
+    monkeypatch.delenv("CCCI_DEPS_FILE", raising=False)
+    deps.write_run_state([{"recipe": "x", "domain": "y"}])  # must not raise