feat(harness): P1 — single registry-backed meta loader (rcust)

One loader: runner/harness/meta.py::load(recipe) -> RecipeMeta (frozen dataclass,
attribute access), backed by the declarative KEYS registry (14 final keys + 3
P2-deprecated). The ONLY exec() of tests/<recipe>/recipe_meta.py. Validation per
the locked decision: unknown ALL-CAPS top-level name or type mismatch = MetaError
(hard error at load); underscore-prefixed names recipe-private; callables only on
hook-typed keys.

Migrated all six legacy loaders (spec §4 L1–L6):
- run_recipe_ci.py::_load_meta deleted; orchestrator loads once, passes meta down
- tests/conftest.py::_recipe_meta deleted; meta fixture returns full RecipeMeta (R3)
- lifecycle.py::_recipe_extra_env/_recipe_meta_flag deleted; deploy_app takes meta
- deps.py::declared_deps deleted; callers read meta.DEPS
- canonical.py::is_enrolled reads through meta.load()
- screenshot.py now actually receives SCREENSHOT through the orchestrator path (R2
  fix; proven by unit test through the real load path)

Mumble private constants underscore-prefixed (_WELCOME_TEXT_MARKER/_MAX_USERS) +
importers fixed. New tests/unit/test_meta.py (all-recipes-load-clean typo gate,
MetaError cases, spec §2 baseline defaults, underscore exemption, doc sync). Docs
§4 key table now GENERATED from the registry (scripts/gen-meta-docs.py); drift
fails CI.

Verified on cc-ci: cc-ci-run -m pytest tests/unit -q -> 175 passed; scripts/lint.sh -> PASS.

runner/harness/canonical.py

@@ -30,17 +30,13 @@ import subprocess
 import time
 
 from . import abra, warm, warmsnap
+from . import meta as meta_mod
 
 
 def is_enrolled(recipe: str) -> bool:
-    """True if `tests/<recipe>/recipe_meta.py` sets `WARM_CANONICAL = True`. Missing meta → False."""
-    path = os.path.join(os.path.dirname(__file__), "..", "..", "tests", recipe, "recipe_meta.py")
-    if not os.path.exists(path):
-        return False
-    ns: dict = {}
-    with open(path) as fh:
-        exec(compile(fh.read(), path, "exec"), ns)  # noqa: S102 (trusted, in-repo)
-    return bool(ns.get("WARM_CANONICAL"))
+    """True if `tests/<recipe>/recipe_meta.py` sets `WARM_CANONICAL = True`. Missing meta → False.
+    Reads through the single meta loader (rcust P1 — no per-module exec)."""
+    return bool(meta_mod.load(recipe).WARM_CANONICAL)
 
 
 def canonical_domain(recipe: str) -> str:
@@ -51,7 +47,7 @@ def canonical_domain(recipe: str) -> str:
 def enrolled_recipes() -> list[str]:
     """All recipes enrolled as data-warm canonicals (recipe_meta.WARM_CANONICAL=True), sorted. Used
     by the WC6 nightly sweep to know which canonicals to refresh via a green cold run on latest."""
-    tests_dir = os.path.join(os.path.dirname(__file__), "..", "..", "tests")
+    tests_dir = meta_mod.TESTS_DIR
     out = []
     try:
         for name in sorted(os.listdir(tests_dir)):

runner/harness/deps.py

@@ -31,19 +31,7 @@ import os
 from collections.abc 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]
+from . import meta as meta_mod
 
 
 def dep_domain(parent_recipe: str, pr: str, ref: str | None, dep_recipe: str) -> str:
@@ -81,11 +69,12 @@ def deploy_deps(
     pr: str,
     ref: str | None,
     deps: Iterable[str],
-    meta_for: dict[str, dict] | None = None,
+    meta_for: 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)."""
+    dicts (one per dep). `meta_for` maps dep_recipe -> RecipeMeta (HEALTH_PATH/HEALTH_OK/timeouts)
+    so the readiness wait uses per-dep config; a missing dep meta is loaded via meta.load()
+    (defaults: /, 200/301/302, 600s)."""
     meta_for = meta_for or {}
     state: list[dict] = []
     for dep in deps:
@@ -94,20 +83,21 @@ def deploy_deps(
         # 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.
-        dm = meta_for.get(dep, {})
+        dm = meta_for.get(dep) or meta_mod.load(dep)
         lifecycle.deploy_app(
             dep,
             domain,
             secrets=True,
-            deploy_timeout=int(dm.get("DEPLOY_TIMEOUT", 900)),
+            deploy_timeout=int(dm.DEPLOY_TIMEOUT),
+            meta=dm,
         )
         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)),
+                ok_codes=tuple(dm.HEALTH_OK),
+                path=dm.HEALTH_PATH,
+                deploy_timeout=int(dm.DEPLOY_TIMEOUT),
+                http_timeout=int(dm.HTTP_TIMEOUT),
             )
         except Exception:
             # If a dep fails to converge, abort the whole resolve — let the caller teardown

runner/harness/generic.py

@@ -19,6 +19,7 @@ import ssl
 import time
 
 from . import abra, lifecycle
+from . import meta as meta_mod
 
 # A recipe is backup-capable iff a compose file carries a truthy backupbot.backup label.
 _BACKUPBOT_RE = re.compile(r"backupbot\.backup\b[^\n]*\btrue\b", re.IGNORECASE)
@@ -28,13 +29,14 @@ def _recipe_dir(recipe: str) -> str:
     return abra.recipe_dir(recipe)  # the per-run tree inside a CI run ($ABRA_DIR)
 
 
-def backup_capable(recipe: str, meta: dict | None = None) -> bool:
+def backup_capable(recipe: str, meta=None) -> bool:
     """Whether the harness should run the backup/restore tiers (else they are a clean N/A skip, DG3).
 
-    `recipe_meta.BACKUP_CAPABLE` (bool) overrides; otherwise auto-detect by scanning the recipe's
-    compose*.yml for a truthy `backupbot.backup` label (the Co-op Cloud backup convention)."""
-    if meta and "BACKUP_CAPABLE" in meta:
-        return bool(meta["BACKUP_CAPABLE"])
+    `recipe_meta.BACKUP_CAPABLE` (bool) overrides when explicitly set (RecipeMeta default is None =
+    unset); otherwise auto-detect by scanning the recipe's compose*.yml for a truthy
+    `backupbot.backup` label (the Co-op Cloud backup convention)."""
+    if meta is not None and meta.BACKUP_CAPABLE is not None:
+        return bool(meta.BACKUP_CAPABLE)
     for path in glob.glob(os.path.join(_recipe_dir(recipe), "compose*.yml")):
         try:
             with open(path) as fh:
@@ -75,7 +77,7 @@ def served_cert(domain: str, port: int = 443) -> tuple[bool, str]:
     return (True, f"CN={cn} SAN={sans}")
 
 
-def assert_serving(domain: str, meta: dict) -> None:
+def assert_serving(domain: str, meta) -> None:
     """The single generic "is the app really serving?" assertion (DG1).
 
     The app-vs-Traefik-fallback proof is steps 1+2 (both load-bearing, verified by the Adversary):
@@ -90,14 +92,14 @@ def assert_serving(domain: str, meta: dict) -> None:
 
     Steps 1–2 are BOUNDED POLLS (no bare sleep), so a state-mutating op (upgrade/restore) that leaves
     the app briefly reconverging settles, while a persistent failure still fails within the timeout."""
-    deadline = time.time() + meta["DEPLOY_TIMEOUT"]
+    deadline = time.time() + meta.DEPLOY_TIMEOUT
     while time.time() < deadline and not lifecycle.services_converged(domain):
         time.sleep(5)
     assert lifecycle.services_converged(domain), f"{domain}: services did not converge"
 
-    path = meta["HEALTH_PATH"]
-    ok = tuple(meta["HEALTH_OK"])
-    deadline = time.time() + meta["HTTP_TIMEOUT"]
+    path = meta.HEALTH_PATH
+    ok = tuple(meta.HEALTH_OK)
+    deadline = time.time() + meta.HTTP_TIMEOUT
     served = False
     status, body = 0, ""
     while time.time() < deadline:
@@ -141,7 +143,7 @@ def op_state() -> dict:
         return {}
 
 
-def assert_upgraded(domain: str, meta: dict) -> None:
+def assert_upgraded(domain: str, meta) -> None:
     """Generic UPGRADE assertion (post-op): the orchestrator already performed the upgrade once via
     `abra app deploy --chaos` of the PR-head checkout. Assert it reconverged + still serves AND that
     the deployment is genuinely the PR-head code under test (HC1) — non-vacuously (guarding F1d-2).
@@ -212,7 +214,7 @@ def assert_backup_artifact(domain: str) -> str:
     return snap_id
 
 
-def assert_restore_healthy(domain: str, meta: dict) -> None:
+def assert_restore_healthy(domain: str, meta) -> None:
     """Generic RESTORE assertion (post-op): the orchestrator already restored. Assert the app is
     healthy + serving again (assert_serving polls, so the post-restore reconverge settles)."""
     assert_serving(domain, meta)
@@ -226,7 +228,7 @@ def perform_upgrade(
     recipe: str,
     head_ref: str | None,
     deploy_timeout: int = 900,
-    meta: dict | None = None,
+    meta=None,
 ) -> dict[str, str | None]:
     """Perform the UPGRADE op once, in place, to the PR-HEAD code under test (HC1): re-checkout the
     PR head (the prev-tag base deploy reset the recipe working tree), then `abra app deploy --chaos`
@@ -244,7 +246,8 @@ def perform_upgrade(
     STRICTER convergence+health wait here: services N/N (wait_healthy) + app HEALTH_PATH healthy +
     any recipe READY_PROBE (collabora WOPI discovery 200). This bounds readiness by OUR generous
     deadline, not abra's impatient one — and is stronger evidence than abra's monitor."""
-    meta = meta or {}
+    if meta is None:
+        meta = meta_mod.load(recipe)
     before = lifecycle.deployed_identity(domain)
     if head_ref:
         lifecycle.recipe_checkout_ref(recipe, head_ref)
@@ -253,9 +256,7 @@ def perform_upgrade(
     # (target) version, so the base deploys minimally WITHOUT it and the upgrade adds it to COMPOSE_FILE
     # here, after the PR-head checkout (which ships the overlay) and before the chaos redeploy that
     # picks up the new .env. Dict or callable(domain)->dict. No-op for recipes without it.
-    upgrade_env = meta.get("UPGRADE_EXTRA_ENV") or {}
-    if callable(upgrade_env):
-        upgrade_env = upgrade_env(domain) or {}
+    upgrade_env = meta_mod.upgrade_extra_env(meta, domain)
     for k, v in upgrade_env.items():
         print(f"  upgrade-env: {k}={v}", flush=True)
         abra.env_set(domain, k, v)
@@ -266,14 +267,12 @@ def perform_upgrade(
     # Own the convergence verification (abra's monitor was skipped via -c).
     lifecycle.wait_healthy(
         domain,
-        ok_codes=tuple(meta.get("HEALTH_OK", (200, 301, 302))),
-        path=meta.get("HEALTH_PATH", "/"),
-        deploy_timeout=int(meta.get("DEPLOY_TIMEOUT", deploy_timeout)),
-        http_timeout=int(meta.get("HTTP_TIMEOUT", 300)),
-    )
-    lifecycle.wait_ready_probes(
-        meta, domain, timeout=int(meta.get("DEPLOY_TIMEOUT", deploy_timeout))
+        ok_codes=tuple(meta.HEALTH_OK),
+        path=meta.HEALTH_PATH,
+        deploy_timeout=int(meta.DEPLOY_TIMEOUT),
+        http_timeout=int(meta.HTTP_TIMEOUT),
     )
+    lifecycle.wait_ready_probes(meta, domain, timeout=int(meta.DEPLOY_TIMEOUT))
     after = lifecycle.deployed_identity(domain)
     # Evidence (HC1): the chaos-version label = the deployed recipe commit; it should match the
     # PR-head we checked out — proving the upgrade deployed the code under test, not a published tag.

runner/harness/lifecycle.py

@@ -19,6 +19,7 @@ import time
 import urllib.request
 
 from . import abra, lifetime
+from . import meta as meta_mod
 
 GATEWAY_IP = "143.244.213.108"  # *.ci.commoninternet.net -> gateway (TLS passthrough to cc-ci)
 # A run app domain is "<recipe[:4]>-<6hex>.ci.commoninternet.net" (see DECISIONS.md). Used by the
@@ -111,37 +112,6 @@ def _residual(domain: str) -> dict:
     }
 
 
-def _recipe_extra_env(recipe: str, domain: str) -> dict[str, str]:
-    """Per-recipe extra .env keys, applied at every deploy (install + upgrade's old_app) so a recipe
-    with multi-domain / config needs is enrolled with NO shared-harness change (D5/M6.5). A recipe
-    declares `EXTRA_ENV` in tests/<recipe>/recipe_meta.py as either a dict or a callable
-    `EXTRA_ENV(domain) -> dict` (callable form lets it derive values from the per-run domain, e.g.
-    cryptpad's SANDBOX_DOMAIN). 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)
-    ee = ns.get("EXTRA_ENV")
-    if callable(ee):
-        ee = ee(domain)
-    return {str(k): str(v) for k, v in (ee or {}).items()}
-
-
-def _recipe_meta_flag(recipe: str, key: str) -> bool:
-    """Read a boolean flag from tests/<recipe>/recipe_meta.py (e.g. CHAOS_BASE_DEPLOY). Returns
-    False if the recipe ships no meta or the flag is absent/falsey. Trusted in-repo exec, same as
-    _recipe_extra_env."""
-    path = os.path.join(os.path.dirname(__file__), "..", "..", "tests", recipe, "recipe_meta.py")
-    if not os.path.exists(path):
-        return False
-    ns: dict = {}
-    with open(path) as fh:
-        exec(compile(fh.read(), path, "exec"), ns)  # noqa: S102 (trusted, in-repo)
-    return bool(ns.get(key))
-
-
 def _record_deploy() -> None:
     """Increment the per-run deploy counter (DG4.1: one deploy per run). No-op unless the
     orchestrator set CCCI_DEPLOY_COUNT_FILE — so it never affects standalone/manual use."""
@@ -238,15 +208,22 @@ def deploy_app(
     secrets: bool = True,
     install_steps_hook: tuple[str, str] | None = None,
     deploy_timeout: int = 900,
+    meta=None,
 ) -> None:
     """Create + configure + deploy an app. Forces LETS_ENCRYPT_ENV='' so traefik serves the
     wildcard cert via the file provider and NEVER attempts ACME (adversary finding A1). Applies any
     per-recipe EXTRA_ENV (recipe_meta.py) and the custom install-steps hook (Phase 1d) before deploy.
 
+    `meta` is the recipe's loaded RecipeMeta (EXTRA_ENV / CHAOS_BASE_DEPLOY); the orchestrator
+    loads once and passes it down. Callers without one in hand (fixtures, warm reconcile) may omit
+    it — it is then loaded here via the single meta.load() path.
+
     `deploy_timeout` is the subprocess timeout for `abra app deploy`. Caller (orchestrator) passes
     `recipe_meta.DEPLOY_TIMEOUT` so heavy recipes (ghost, matrix-synapse, lasuite-meet) can extend
     past the 900s default. abra's INTERNAL TIMEOUT (recipe's TIMEOUT env, default 300s) is set via
     EXTRA_ENV; this is the Python subprocess wrapper's timeout so abra doesn't get SIGKILLed mid-deploy."""
+    if meta is None:
+        meta = meta_mod.load(recipe)
     _record_deploy()
     # Lock BEFORE the app exists: a concurrent run's janitor must never see this app without a
     # held app lock (it would probe it as an orphan and reap an in-flight deploy). Also the
@@ -280,7 +257,7 @@ def deploy_app(
         # abra's pinned-deploy clean-tree check FATA ('has locally unstaged changes'); chaos skips lint +
         # the clean-tree gate and deploys the EXPLICITLY-checked-out pinned version (we already ran
         # recipe_checkout(version) above) — NOT latest. Same mechanism as the lightweight-tag branch.
-        elif _recipe_meta_flag(recipe, "CHAOS_BASE_DEPLOY"):
+        elif meta.CHAOS_BASE_DEPLOY:
             print(
                 f"  deploy_app({recipe}@{version}): CHAOS_BASE_DEPLOY set → chaos base deploy of the "
                 "checked-out pinned version (skips clean-tree/lint; deploys version, not LATEST)",
@@ -293,7 +270,7 @@ def deploy_app(
     # it ourselves is recipe-agnostic and canonical (the run domain IS the app's domain).
     abra.env_set(domain, "DOMAIN", domain)
     abra.env_set(domain, "LETS_ENCRYPT_ENV", "")
-    for k, v in _recipe_extra_env(recipe, domain).items():
+    for k, v in meta_mod.extra_env(meta, domain).items():
         abra.env_set(domain, k, v)
     if secrets:
         abra.secret_generate(domain)
@@ -510,7 +487,7 @@ def chaos_redeploy(
     abra.deploy(domain, chaos=True, timeout=deploy_timeout, no_converge_checks=no_converge_checks)
 
 
-def wait_ready_probes(meta: dict, domain: str, timeout: int = 600) -> None:
+def wait_ready_probes(meta, domain: str, timeout: int = 600) -> None:
     """Poll a recipe's optional READY_PROBE endpoints until each returns an accepted status, or raise.
 
     A recipe_meta may define `READY_PROBE(domain) -> [{"host":..., "path":..., "ok":(200,)}, ...]`
@@ -527,7 +504,7 @@ def wait_ready_probes(meta: dict, domain: str, timeout: int = 600) -> None:
     must be released by the old task + rebound by the new) the voice server can be down while
     HTTP-200 still passes — and backup-bot then execs into a not-running app container (409). Requiring
     the voice port to be stably listening before proceeding closes that window."""
-    probe_fn = meta.get("READY_PROBE")
+    probe_fn = meta.READY_PROBE
     if not callable(probe_fn):
         return
     probes = probe_fn(domain) or []

runner/harness/meta.py

@@ -0,0 +1,267 @@
+"""Single recipe-meta loader + declarative key registry (recipe-custom restructure P1; spec
+docs/recipe-customization.md §8 R1).
+
+THE one place `tests/<recipe>/recipe_meta.py` is `exec()`d. Every consumer (orchestrator, pytest
+`meta` fixture, deploy env shaping, deps, warm-canonical enrollment, screenshot) reads the ONE
+loaded `RecipeMeta` object instead of re-exec'ing the file and cherry-picking keys — that drift
+(six divergent loaders, spec §4 L1–L6) is what made `SCREENSHOT` an unreachable knob (R2) and let
+key typos silently disable coverage (R6).
+
+Validation (locked decision, recipe-custom-restructure-full-plan.md):
+- unknown ALL-CAPS top-level name → MetaError (hard error, fails fast at load; the all-recipes
+  unit test catches it at PR time). Underscore-prefixed names (`_FOO`) are recipe-private and
+  exempt; lowercase names (helper functions/imports) are ignored.
+- type mismatch → MetaError. Callables are accepted ONLY for hook-typed keys.
+
+The KEYS registry is the single source of truth for the key set: it drives validation, the
+RecipeMeta dataclass fields, and the generated reference table in docs/recipe-customization.md §4
+(scripts/gen-meta-docs.py; a unit test asserts the committed table matches).
+"""
+
+from __future__ import annotations
+
+import copy
+import dataclasses
+import difflib
+import os
+from collections.abc import Callable
+
+ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+TESTS_DIR = os.path.join(ROOT, "tests")
+
+
+class MetaError(Exception):
+    """A recipe_meta.py failed registry validation (unknown key / type mismatch / callable on a
+    data key). Hard error by design: a typo'd key must fail the run at load, not silently reduce
+    coverage (spec §8 R6 — the worst failure mode for a CI harness)."""
+
+
+@dataclasses.dataclass(frozen=True)
+class Key:
+    """One registered recipe_meta key: name, type tag, default, one-line doc (rendered into the
+    generated reference table), optional extra validator, and a deprecation marker (deprecated
+    keys still load+validate but are scheduled for deletion)."""
+
+    name: str
+    type: str  # "int"|"str"|"tuple[int]"|"bool"|"dict_or_hook"|"hook"|"list[str]"|"dict"
+    default: object
+    doc: str
+    validate: Callable[[object], None] | None = None
+    deprecated: bool = False
+
+
+KEYS: tuple[Key, ...] = (
+    Key(
+        "HEALTH_PATH",
+        "str",
+        "/",
+        "Path probed for serving/health checks (deploy wait + generic `assert_serving`).",
+    ),
+    Key("HEALTH_OK", "tuple[int]", (200, 301, 302), "Acceptable HTTP status codes for health."),
+    Key("DEPLOY_TIMEOUT", "int", 600, "Max seconds to wait for swarm convergence per deploy."),
+    Key("HTTP_TIMEOUT", "int", 300, "Max seconds to wait for HTTP health after convergence."),
+    Key(
+        "BACKUP_CAPABLE",
+        "bool",
+        None,
+        "Override the backup-tier capability auto-detect (compose `backupbot.backup` labels). `False` forces N/A; `True` forces the tier on; unset = auto-detect.",
+    ),
+    Key(
+        "EXPECTED_NA",
+        "dict",
+        None,
+        "Declare an N/A rung intentional: `{rung: reason}`. The cap stands either way; only the report wording changes.",
+    ),
+    Key(
+        "READY_PROBE",
+        "hook",
+        None,
+        "Callable returning extra readiness probes, run after install AND after upgrade: HTTP `{host, path, ok}` or TCP `{tcp_host, tcp_port, stable}`.",
+    ),
+    Key(
+        "UPGRADE_BASE_VERSION",
+        "str",
+        None,
+        "Exact published tag overriding the upgrade tier's base (default: `recipe_versions[-2]`).",
+    ),
+    Key(
+        "BACKUP_VERIFY",
+        "hook",
+        None,
+        "Callable `-> bool` post-backup data-capture check; `False` re-runs the backup (truncated-dump race guard), retried up to 3 attempts.",
+    ),
+    Key(
+        "UPGRADE_EXTRA_ENV",
+        "dict_or_hook",
+        None,
+        "Extra `.env` keys applied after the PR-head checkout, before the chaos redeploy (env that exists only at head).",
+    ),
+    Key(
+        "EXTRA_ENV",
+        "dict_or_hook",
+        {},
+        "Extra `.env` keys applied at EVERY deploy (base install AND upgrade old-app). Callable form derives values from the per-run domain.",
+    ),
+    Key(
+        "DEPS",
+        "list[str]",
+        [],
+        'Dep recipes deployed/provisioned alongside (e.g. `["keycloak"]`); creds land in `$CCCI_DEPS_FILE`.',
+    ),
+    Key(
+        "WARM_CANONICAL",
+        "bool",
+        False,
+        "Enroll the recipe in the warm/canonical app system (docs/warm.md): green cold runs on LATEST advance the canonical snapshot.",
+    ),
+    Key(
+        "SCREENSHOT",
+        "hook",
+        None,
+        "Callable driving Playwright to a safe, credential-free post-login view for the results-card screenshot (default: landing page).",
+    ),
+    # ---- Deprecated (deleted in restructure P2; registered so P1 lands green before P2 removes
+    # them — see recipe-custom-restructure-full-plan.md "Decisions locked") -----------------------
+    Key(
+        "CHAOS_BASE_DEPLOY",
+        "bool",
+        False,
+        "DEPRECATED (P2 deletes): ship `tests/<recipe>/compose.ccci.yml` instead — the harness auto-copies it and auto-uses `--chaos` for the base deploy.",
+        deprecated=True,
+    ),
+    Key(
+        "OIDC_AT_INSTALL",
+        "bool",
+        False,
+        "DEPRECATED (P2 deletes): install-time deps provisioning becomes the ONLY mode when `DEPS` is set.",
+        deprecated=True,
+    ),
+    Key(
+        "SKIP_GENERIC",
+        "list[str]",
+        [],
+        "DEPRECATED (P2 deletes; zero users): suppress the generic floor for the listed ops. The env form `CCCI_SKIP_GENERIC*` stays as a dev-only escape hatch.",
+        deprecated=True,
+    ),
+)
+
+_REGISTRY: dict[str, Key] = {k.name: k for k in KEYS}
+
+# The one validated, attribute-access view of a recipe's customization. Generated from KEYS so the
+# field set can never drift from the registry (frozen: consumers share one immutable object).
+RecipeMeta = dataclasses.make_dataclass(
+    "RecipeMeta",
+    [(k.name, object, dataclasses.field(default=None)) for k in KEYS],
+    frozen=True,
+)
+RecipeMeta.__doc__ = (
+    "Validated per-recipe customization (one field per registered key; attribute access). "
+    "Built ONLY by meta.load()."
+)
+
+
+def meta_path(recipe: str, tests_dir: str | None = None) -> str:
+    """Canonical path of a recipe's meta file (pure)."""
+    return os.path.join(tests_dir or TESTS_DIR, recipe, "recipe_meta.py")
+
+
+def _coerce(key: Key, value: object, path: str) -> object:
+    """Validate `value` against `key`'s declared type; normalize containers (tuple[int]/list[str]).
+    Raises MetaError on mismatch — including a callable supplied for a data-typed key."""
+    t = key.type
+    if callable(value) and t not in ("hook", "dict_or_hook"):
+        raise MetaError(
+            f"{path}: {key.name} is a data key (type {t}) — callables are accepted only for "
+            f"hook-typed keys"
+        )
+    if t == "int":
+        if isinstance(value, int) and not isinstance(value, bool):
+            return value
+    elif t == "str":
+        if isinstance(value, str):
+            return value
+    elif t == "bool":
+        if isinstance(value, bool):
+            return value
+    elif t == "tuple[int]":
+        if isinstance(value, tuple | list) and all(
+            isinstance(x, int) and not isinstance(x, bool) for x in value
+        ):
+            return tuple(value)
+    elif t == "list[str]":
+        if isinstance(value, tuple | list) and all(isinstance(x, str) for x in value):
+            return list(value)
+    elif t == "dict":
+        if isinstance(value, dict):
+            return value
+    elif (
+        t == "hook"
+        and callable(value)
+        or t == "dict_or_hook"
+        and (isinstance(value, dict) or callable(value))
+    ):
+        return value
+    raise MetaError(f"{path}: {key.name} must be {t}, got {type(value).__name__} ({value!r})")
+
+
+def load(recipe: str, tests_dir: str | None = None):
+    """Load + validate a recipe's customization -> RecipeMeta. THE only exec() of recipe_meta.py.
+
+    Missing file -> all registry defaults (the zero-config baseline, spec §2). Unknown
+    non-underscore ALL-CAPS top-level name or type mismatch -> MetaError (hard error).
+    `tests_dir` overrides the recipe-meta root (unit tests / fixtures)."""
+    path = meta_path(recipe, tests_dir)
+    values = {k.name: copy.copy(k.default) for k in KEYS}
+    if os.path.exists(path):
+        ns: dict = {}
+        with open(path) as fh:
+            exec(compile(fh.read(), path, "exec"), ns)  # noqa: S102 (trusted, in-repo)
+        for name in sorted(ns):
+            if name.startswith("_") or not name.isupper():
+                continue  # _FOO = recipe-private (exempt); lowercase = helpers/imports (ignored)
+            key = _REGISTRY.get(name)
+            if key is None:
+                near = difflib.get_close_matches(name, _REGISTRY, n=1)
+                hint = f" — did you mean {near[0]!r}?" if near else ""
+                raise MetaError(
+                    f"{path}: unknown recipe_meta key {name!r}{hint}. Registered keys: "
+                    f"{', '.join(sorted(_REGISTRY))}. Recipe-private constants must be "
+                    f"underscore-prefixed (e.g. _{name})."
+                )
+            values[name] = _coerce(key, ns[name], path)
+            if key.validate:
+                key.validate(values[name])
+    return RecipeMeta(**values)
+
+
+def as_dict(meta) -> dict:
+    """RecipeMeta -> {key: value} (every registered key, defaults included)."""
+    return dataclasses.asdict(meta)
+
+
+def non_default(meta) -> dict:
+    """The keys a recipe explicitly customized: {key: value} where value differs from the registry
+    default. Hooks compare by identity-vs-None (a set hook is always non-default). Feeds the run's
+    customization manifest (P5)."""
+    out = {}
+    for k in KEYS:
+        v = getattr(meta, k.name)
+        if v != k.default:
+            out[k.name] = v
+    return out
+
+
+def _env_map(value, domain: str) -> dict[str, str]:
+    if callable(value):
+        value = value(domain)
+    return {str(k): str(v) for k, v in (value or {}).items()}
+
+
+def extra_env(meta, domain: str) -> dict[str, str]:
+    """Resolve EXTRA_ENV (dict or callable(domain)->dict) to the concrete per-run env map."""
+    return _env_map(meta.EXTRA_ENV, domain)
+
+
+def upgrade_extra_env(meta, domain: str) -> dict[str, str]:
+    """Resolve UPGRADE_EXTRA_ENV (dict or callable(domain)->dict) to the concrete env map."""
+    return _env_map(meta.UPGRADE_EXTRA_ENV, domain)

runner/harness/screenshot.py

@@ -33,12 +33,19 @@ def screenshot_path(run_artifact_dir: str) -> str:
     return os.path.join(run_artifact_dir, "screenshot.png")
 
 
-def _load_screenshot_hook(recipe_meta: dict | None):
+def _load_screenshot_hook(recipe_meta):
     """Return the recipe's optional SCREENSHOT hook (a callable) if it declared one, else None.
-    The hook drives Playwright to a safe post-login view; default is the landing page."""
-    if not recipe_meta:
+    The hook drives Playwright to a safe post-login view; default is the landing page.
+
+    `recipe_meta` is the loaded RecipeMeta (rcust P1 — the single loader actually delivers
+    SCREENSHOT now; under the old L1 allowlist the key never arrived, spec §8 R2). A plain dict
+    is still accepted for direct/manual callers."""
+    if recipe_meta is None:
         return None
-    hook = recipe_meta.get("SCREENSHOT")
+    if isinstance(recipe_meta, dict):
+        hook = recipe_meta.get("SCREENSHOT")
+    else:
+        hook = getattr(recipe_meta, "SCREENSHOT", None)
     return hook if callable(hook) else None
 
 

runner/run_recipe_ci.py

@@ -58,6 +58,9 @@ from harness import (  # noqa: E402
 from harness import (  # noqa: E402
     deps as deps_mod,
 )
+from harness import (  # noqa: E402
+    meta as meta_mod,
+)
 from harness import (  # noqa: E402
     results as results_mod,
 )
@@ -247,40 +250,11 @@ def snapshot_recipe_tests(recipe: str) -> str | None:
     return dst
 
 
-def _load_meta(recipe: str) -> dict:
-    """Mirror tests/conftest._recipe_meta so the orchestrator's deploy/wait uses the same per-recipe
-    config the tiers see (timeouts, health path/codes)."""
-    meta = {
-        "HEALTH_PATH": "/",
-        "HEALTH_OK": (200, 301, 302),
-        "DEPLOY_TIMEOUT": 600,
-        "HTTP_TIMEOUT": 300,
-    }
-    path = os.path.join(ROOT, "tests", recipe, "recipe_meta.py")
-    if os.path.exists(path):
-        ns: dict = {}
-        with open(path) as fh:
-            exec(compile(fh.read(), path, "exec"), ns)  # noqa: S102 (trusted, in-repo)
-        for k in list(meta) + [
-            "BACKUP_CAPABLE",
-            "SKIP_GENERIC",
-            "EXPECTED_NA",
-            "OIDC_AT_INSTALL",
-            "READY_PROBE",
-            "UPGRADE_BASE_VERSION",
-            "BACKUP_VERIFY",
-            "UPGRADE_EXTRA_ENV",
-        ]:
-            if k in ns:
-                meta[k] = ns[k]
-    return meta
-
-
 def _tier_env(domain: str) -> dict:
     return dict(os.environ, CCCI_APP_DOMAIN=domain, CCCI_BASE_URL=f"https://{domain}")
 
 
-def _skip_generic(op: str, meta: dict) -> bool:
+def _skip_generic(op: str, meta) -> bool:
     """Whether the generic assertion for `op` is opted out (Phase 1e HC3). Default: run (additive).
     Opt-out, any of: env CCCI_SKIP_GENERIC (all ops), env CCCI_SKIP_GENERIC_<OP>, or the recipe's
     declarative recipe_meta.SKIP_GENERIC list (op name, or "all"/"*")."""
@@ -288,11 +262,11 @@ def _skip_generic(op: str, meta: dict) -> bool:
         return True
     if _truthy(os.environ.get(f"CCCI_SKIP_GENERIC_{op.upper()}")):
         return True
-    sg = [str(s).lower() for s in (meta.get("SKIP_GENERIC") or [])]
+    sg = [str(s).lower() for s in (meta.SKIP_GENERIC or [])]
     return "all" in sg or "*" in sg or op in sg
 
 
-def _run_pre_hook(recipe: str, op: str, repo_local: str | None, domain: str, meta: dict) -> None:
+def _run_pre_hook(recipe: str, op: str, repo_local: str | None, domain: str, meta) -> None:
     """Run the optional pre-op seed hook (recipe ops.py `pre_<op>`) BEFORE the harness performs the
     op (HC3 op/assertion split): overlays seed data-continuity markers / the backup→restore mutation
     here, then assert post-op in test_<op>.py. cc-ci's ops.py is trusted; a repo-local ops.py is
@@ -322,7 +296,7 @@ def _perform_op(
     head_ref: str | None,
     op_state: dict,
     deploy_timeout: int = 900,
-    meta: dict | None = None,
+    meta=None,
 ) -> None:
     """Perform the single mutating op ONCE (the harness owns the op, HC3). install has no op. Records
     what the assertions need (pre-upgrade identity, backup snapshot_id) into op_state. None of these
@@ -345,7 +319,7 @@ def _perform_op(
         # verify fails we re-run the WHOLE backup (fresh restic snapshot) with a re-stabilised DB, up to
         # 3 attempts. Recipes without BACKUP_VERIFY are unaffected (single backup, as before).
         snap = generic.perform_backup(domain)
-        verify = meta.get("BACKUP_VERIFY") if meta else None
+        verify = meta.BACKUP_VERIFY if meta else None
         attempt = 1
         while callable(verify) and not verify(domain) and attempt < 3:
             attempt += 1
@@ -371,7 +345,7 @@ def run_lifecycle_tier(
     op: str,
     repo_local: str | None,
     domain: str,
-    meta: dict,
+    meta,
     head_ref: str | None,
     op_state: dict,
     records: list[dict] | None = None,
@@ -411,7 +385,7 @@ def run_lifecycle_tier(
             recipe,
             head_ref,
             op_state,
-            deploy_timeout=int(meta.get("DEPLOY_TIMEOUT", 900)),
+            deploy_timeout=int(meta.DEPLOY_TIMEOUT),
             meta=meta,
         )
         with open(os.environ["CCCI_OP_STATE_FILE"], "w") as f:
@@ -523,7 +497,7 @@ def _provision_deps(
             if wd:
                 print(f"  dep: {d} warm provider {wd} not up — cold fallback", flush=True)
             cold_deps.append(d)
-    dep_metas = {d: _load_meta(d) for d in cold_deps}
+    dep_metas = {d: meta_mod.load(d) for d in cold_deps}
     deps_list = (
         deps_mod.deploy_deps(recipe, os.environ.get("PR", "0"), ref, cold_deps, meta_for=dep_metas)
         if cold_deps
@@ -609,7 +583,7 @@ def _wait_undeployed(domain: str, timeout: int = 120) -> None:
 
 
 def run_quick(
-    recipe: str, ref: str | None, head_ref: str | None, repo_local: str | None, meta: dict
+    recipe: str, ref: str | None, head_ref: str | None, repo_local: str | None, meta
 ) -> int:
     """WC4 `--quick` opt-in fast lane (plan §2). Reattach the data-warm canonical (known-good volume)
     → upgrade IN PLACE to the PR head (chaos) → assert generic UPGRADE (reconverge+moved+serving) +
@@ -645,7 +619,7 @@ def run_quick(
 
     op_state: dict = {}
     results: dict[str, str] = {}
-    declared = deps_mod.declared_deps(recipe)
+    declared = list(meta.DEPS)
     deps_state: dict = {}
     deps_ready = True
     deps_not_ready_reason = ""
@@ -657,13 +631,13 @@ def run_quick(
     try:
         # 1) reattach the canonical (warm boot at the known-good version + retained volume)
         try:
-            canonical.deploy_canonical(recipe, timeout=int(meta.get("DEPLOY_TIMEOUT", 900)))
+            canonical.deploy_canonical(recipe, timeout=int(meta.DEPLOY_TIMEOUT))
             lifecycle.wait_healthy(
                 domain,
-                ok_codes=tuple(meta["HEALTH_OK"]),
-                path=meta["HEALTH_PATH"],
-                deploy_timeout=meta["DEPLOY_TIMEOUT"],
-                http_timeout=meta["HTTP_TIMEOUT"],
+                ok_codes=tuple(meta.HEALTH_OK),
+                path=meta.HEALTH_PATH,
+                deploy_timeout=meta.DEPLOY_TIMEOUT,
+                http_timeout=meta.HTTP_TIMEOUT,
             )
             warm_ok = True
         except Exception as e:  # noqa: BLE001
@@ -678,7 +652,7 @@ def run_quick(
                     for d in declared:
                         wd = warm.warm_domain(d)
                         (warm_deps if (wd and warm.is_warm_up(d, wd)) else cold_deps).append(d)
-                    dep_metas = {d: _load_meta(d) for d in cold_deps}
+                    dep_metas = {d: meta_mod.load(d) for d in cold_deps}
                     deps_list = (
                         deps_mod.deploy_deps(
                             recipe, os.environ.get("PR", "0"), ref, cold_deps, meta_for=dep_metas
@@ -848,7 +822,7 @@ def promote_canonical(recipe: str, head_ref: str | None) -> None:
     if not latest:
         print(f"WC5 promote: no version tags for {recipe} — skip", flush=True)
         return
-    meta = _load_meta(recipe)
+    meta = meta_mod.load(recipe)
     # The cold run's deploy-count was already asserted + the countfile removed; don't perturb it.
     os.environ.pop("CCCI_DEPLOY_COUNT_FILE", None)
     print(
@@ -860,14 +834,15 @@ def promote_canonical(recipe: str, head_ref: str | None) -> None:
         domain,
         version=latest,
         secrets=True,
-        deploy_timeout=int(meta.get("DEPLOY_TIMEOUT", 900)),
+        deploy_timeout=int(meta.DEPLOY_TIMEOUT),
+        meta=meta,
     )
     lifecycle.wait_healthy(
         domain,
-        ok_codes=tuple(meta["HEALTH_OK"]),
-        path=meta["HEALTH_PATH"],
-        deploy_timeout=meta["DEPLOY_TIMEOUT"],
-        http_timeout=meta["HTTP_TIMEOUT"],
+        ok_codes=tuple(meta.HEALTH_OK),
+        path=meta.HEALTH_PATH,
+        deploy_timeout=meta.DEPLOY_TIMEOUT,
+        http_timeout=meta.HTTP_TIMEOUT,
     )
     abra.undeploy(domain)
     _wait_undeployed(domain)
@@ -906,7 +881,7 @@ def main() -> int:
     # HEAD (the catalogue current) for a non-PR `!testme`. Captured before any version-tag checkout.
     head_ref = ref or lifecycle.recipe_head_commit(recipe)
     repo_local = snapshot_recipe_tests(recipe)
-    meta = _load_meta(recipe)
+    meta = meta_mod.load(recipe)
 
     # WC4/WC7: opt-in `--quick` fast lane. Requires an existing data-warm canonical; if none, fall
     # back cleanly to the full COLD run below so the PR is still tested (DECISIONS Phase-2w).
@@ -929,9 +904,7 @@ def main() -> int:
     # override must be an exact published version tag (deployed as a pinned base). (Adversary §7.1.)
     want_upgrade = "upgrade" in stages
     prev = (
-        (meta.get("UPGRADE_BASE_VERSION") or lifecycle.previous_version(recipe))
-        if want_upgrade
-        else None
+        (meta.UPGRADE_BASE_VERSION or lifecycle.previous_version(recipe)) if want_upgrade else None
     )
     base = prev or target
     backup_cap = generic.backup_capable(recipe, meta)
@@ -974,12 +947,12 @@ def main() -> int:
     with contextlib.suppress(OSError):
         os.remove(skipfile)
     os.environ["CCCI_DEPS_SKIP_REPORT"] = skipfile
-    declared = deps_mod.declared_deps(recipe)
+    declared = list(meta.DEPS)
     # Q3.2a: a recipe that tolerates OIDC env at first boot AND whose deps are live-warm wires OIDC
     # at INSTALL time (provision the realm BEFORE the single deploy; install_steps.sh writes the env
     # into it) instead of the post-deploy in-place `--chaos` redeploy — which is flaky on the heavy
     # 12-service lasuite-drive stack (collabora WOPI race; see JOURNAL Step 0). Opt-in per recipe.
-    oidc_at_install = bool(meta.get("OIDC_AT_INSTALL")) and bool(declared)
+    oidc_at_install = bool(meta.OIDC_AT_INSTALL) and bool(declared)
     if declared:
         when = "BEFORE deploy (install-time OIDC)" if oidc_at_install else "AFTER generic tiers"
         print(f"\n===== DEPS declared (provision {when}): {declared} =====", flush=True)
@@ -1023,18 +996,19 @@ def main() -> int:
                 version=base,
                 secrets=True,
                 install_steps_hook=hook,
-                deploy_timeout=int(meta.get("DEPLOY_TIMEOUT", 900)),
+                deploy_timeout=int(meta.DEPLOY_TIMEOUT),
+                meta=meta,
             )
             lifecycle.wait_healthy(
                 domain,
-                ok_codes=tuple(meta["HEALTH_OK"]),
-                path=meta["HEALTH_PATH"],
-                deploy_timeout=meta["DEPLOY_TIMEOUT"],
-                http_timeout=meta["HTTP_TIMEOUT"],
+                ok_codes=tuple(meta.HEALTH_OK),
+                path=meta.HEALTH_PATH,
+                deploy_timeout=meta.DEPLOY_TIMEOUT,
+                http_timeout=meta.HTTP_TIMEOUT,
             )
             # Recipe READY_PROBE (e.g. lasuite-drive collabora WOPI discovery) — readiness beyond
             # replica convergence + app HEALTH_PATH; no-op for recipes without one.
-            lifecycle.wait_ready_probes(meta, domain, timeout=int(meta.get("DEPLOY_TIMEOUT", 900)))
+            lifecycle.wait_ready_probes(meta, domain, timeout=int(meta.DEPLOY_TIMEOUT))
             deploy_ok = True
         except Exception as e:  # noqa: BLE001 — a failed deploy is a reported INSTALL failure
             print(f"!! deploy/readiness failed: {e}", flush=True)
@@ -1314,7 +1288,7 @@ def main() -> int:
             no_secret_leak=True,  # narrowed below by an actual scan of the serialised artifact
             screenshot=screenshot_rel,  # Phase 3 U1 (R4): relative PNG name iff capture succeeded
             finished_ts=time.time(),
-            expected_na=meta.get("EXPECTED_NA"),  # declared intentional-skip map (recipe_meta)
+            expected_na=meta.EXPECTED_NA,  # declared intentional-skip map (recipe_meta)
         )
         # Real (if narrow) leak check: no known infra-secret value may appear in the artifact (R7).
         blob = json.dumps(data)