feat(harness): P3 — uniform ctx hook convention (rcust)

harness.meta.HookCtx (frozen): .domain, .base_url, .meta (RecipeMeta), .deps
(provisioned dep creds from $CCCI_DEPS_FILE or None), .op (current lifecycle op
or None); built via meta.hook_ctx() at each hook call site.

All recipe callables now take ctx: EXTRA_ENV(ctx), UPGRADE_EXTRA_ENV(ctx),
READY_PROBE(ctx), BACKUP_VERIFY(ctx), SCREENSHOT(page, ctx), ops.py pre_<op>(ctx).
Dict-valued EXTRA_ENV/UPGRADE_EXTRA_ENV unchanged (only the callable signature
moved). Call sites converted: deploy_app env shaping, perform_upgrade,
wait_ready_probes (gains op=), _perform_op BACKUP_VERIFY, screenshot.capture,
_run_pre_hook.

Legacy signatures fail FAST with a clear migration message: the registry carries
hook_params per hook key, enforced at meta.load() (MetaError names the old vs new
signature); ops.py pre-op hooks get the same check at the orchestrator call site
(meta.check_hook_signature) — no silent TypeError mid-run.

Migrated every in-repo user mechanically (17 ops.py files; cryptpad/lasuite-*/
mailu EXTRA_ENV; mumble+lasuite-drive READY_PROBE; ghost/discourse BACKUP_VERIFY)
— seeded values, probes and assertions byte-identical (domain -> ctx.domain;
keycloak pre_restore's meta arg -> ctx.meta).

Unit tests: hook_ctx field contract, ctx.deps from the run deps file, legacy-
signature MetaError (READY_PROBE/EXTRA_ENV/SCREENSHOT + pre-op checker), ctx
signatures accepted. Docs table regenerated (signature docs in key docs).

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

docs/recipe-customization.md

@@ -115,14 +115,14 @@ _This table is GENERATED from the `runner/harness/meta.py` KEYS registry by `scr
 | `HTTP_TIMEOUT` | `int` | `300` | Max seconds to wait for HTTP health after convergence. |
 | `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. |
 | `EXPECTED_NA` | `dict` | `None` | Declare an N/A rung intentional: `{rung: reason}`. The cap stands either way; only the report wording changes. |
-| `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}`. |
+| `READY_PROBE` | `hook` | `None` | Callable `(ctx) -> [probe, ...]` returning extra readiness probes, run after install AND after upgrade: HTTP `{host, path, ok}` or TCP `{tcp_host, tcp_port, stable}`. |
 | `UPGRADE_BASE_VERSION` | `str` | `None` | Exact published tag overriding the upgrade tier's base (default: `recipe_versions[-2]`). |
-| `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. |
+| `BACKUP_VERIFY` | `hook` | `None` | Callable `(ctx) -> bool` post-backup data-capture check; `False` re-runs the backup (truncated-dump race guard), retried up to 3 attempts. |
-| `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). |
+| `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). Dict, or callable `(ctx) -> dict`. |
-| `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. |
+| `EXTRA_ENV` | `dict_or_hook` | `{}` | Extra `.env` keys applied at EVERY deploy (base install AND upgrade old-app). Dict, or callable `(ctx) -> dict` deriving values from the per-run domain (`ctx.domain`). |
 | `DEPS` | `list[str]` | `[]` | Dep recipes deployed/provisioned alongside (e.g. `["keycloak"]`); creds land in `$CCCI_DEPS_FILE`. |
 | `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. |
-| `SCREENSHOT` | `hook` | `None` | Callable driving Playwright to a safe, credential-free post-login view for the results-card screenshot (default: landing page). |
+| `SCREENSHOT` | `hook` | `None` | Callable `(page, ctx)` driving Playwright to a safe, credential-free post-login view for the results-card screenshot (default: landing page). |
 
 <!-- META-TABLE-END -->
 

runner/harness/discovery.py

@@ -144,7 +144,7 @@ def install_steps(recipe: str, repo_local_dir: str | None) -> tuple[str, str] |
 
 def pre_op_hook(recipe: str, op: str, repo_local_dir: str | None) -> tuple[str, str] | None:
     """The pre-op seed hook for `op`: the path to a recipe `ops.py` module that defines a
-    `pre_<op>(domain, meta)` callable, or None. cc-ci's tests/<recipe>/ops.py wins; the repo-local
+    `pre_<op>(ctx)` callable, or None. cc-ci's tests/<recipe>/ops.py wins; the repo-local
     ops.py is consulted only for allowlist-approved recipes (HC2). The orchestrator imports the
     module and calls pre_<op> BEFORE performing the op (HC3 op/assertion split — overlays seed
     pre-op state here, then assert post-op in test_<op>.py)."""

runner/harness/generic.py

@@ -256,7 +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_mod.upgrade_extra_env(meta, domain)
+    upgrade_env = meta_mod.upgrade_extra_env(meta, meta_mod.hook_ctx(domain, meta, op="upgrade"))
     for k, v in upgrade_env.items():
         print(f"  upgrade-env: {k}={v}", flush=True)
         abra.env_set(domain, k, v)
@@ -272,7 +272,7 @@ def perform_upgrade(
         deploy_timeout=int(meta.DEPLOY_TIMEOUT),
         http_timeout=int(meta.HTTP_TIMEOUT),
     )
-    lifecycle.wait_ready_probes(meta, domain, timeout=int(meta.DEPLOY_TIMEOUT))
+    lifecycle.wait_ready_probes(meta, domain, timeout=int(meta.DEPLOY_TIMEOUT), op="upgrade")
     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

@@ -302,7 +302,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 meta_mod.extra_env(meta, domain).items():
+    for k, v in meta_mod.extra_env(meta, meta_mod.hook_ctx(domain, meta)).items():
         abra.env_set(domain, k, v)
     if secrets:
         abra.secret_generate(domain)
@@ -525,7 +525,7 @@ def chaos_redeploy(
     abra.deploy(domain, chaos=True, timeout=deploy_timeout, no_converge_checks=no_converge_checks)
 
 
-def wait_ready_probes(meta, domain: str, timeout: int = 600) -> None:
+def wait_ready_probes(meta, domain: str, timeout: int = 600, op: str | None = None) -> 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,)}, ...]`
@@ -545,7 +545,7 @@ def wait_ready_probes(meta, domain: str, timeout: int = 600) -> None:
     probe_fn = meta.READY_PROBE
     if not callable(probe_fn):
         return
-    probes = probe_fn(domain) or []
+    probes = probe_fn(meta_mod.hook_ctx(domain, meta, op=op)) or []
     for probe in probes:
         if "tcp_port" in probe:
             host = probe.get("tcp_host", "127.0.0.1")

runner/harness/meta.py

@@ -23,6 +23,8 @@ from __future__ import annotations
 import copy
 import dataclasses
 import difflib
+import inspect
+import json
 import os
 from collections.abc import Callable
 
@@ -48,6 +50,10 @@ class Key:
     doc: str
     validate: Callable[[object], None] | None = None
     deprecated: bool = False
+    # Expected positional-parameter names for a callable value (rcust P3 uniform ctx convention).
+    # Enforced at load so a legacy-signature hook (e.g. `def READY_PROBE(domain)`) fails with a
+    # CLEAR MetaError naming the migration — never a silent TypeError mid-run.
+    hook_params: tuple[str, ...] | None = None
 
 
 KEYS: tuple[Key, ...] = (
@@ -76,7 +82,8 @@ KEYS: tuple[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}`.",
+        "Callable `(ctx) -> [probe, ...]` returning extra readiness probes, run after install AND after upgrade: HTTP `{host, path, ok}` or TCP `{tcp_host, tcp_port, stable}`.",
+        hook_params=("ctx",),
     ),
     Key(
         "UPGRADE_BASE_VERSION",
@@ -88,19 +95,22 @@ KEYS: tuple[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.",
+        "Callable `(ctx) -> bool` post-backup data-capture check; `False` re-runs the backup (truncated-dump race guard), retried up to 3 attempts.",
+        hook_params=("ctx",),
     ),
     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).",
+        "Extra `.env` keys applied after the PR-head checkout, before the chaos redeploy (env that exists only at head). Dict, or callable `(ctx) -> dict`.",
+        hook_params=("ctx",),
     ),
     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.",
+        "Extra `.env` keys applied at EVERY deploy (base install AND upgrade old-app). Dict, or callable `(ctx) -> dict` deriving values from the per-run domain (`ctx.domain`).",
+        hook_params=("ctx",),
     ),
     Key(
         "DEPS",
@@ -118,7 +128,8 @@ KEYS: tuple[Key, ...] = (
         "SCREENSHOT",
         "hook",
         None,
-        "Callable driving Playwright to a safe, credential-free post-login view for the results-card screenshot (default: landing page).",
+        "Callable `(page, ctx)` driving Playwright to a safe, credential-free post-login view for the results-card screenshot (default: landing page).",
+        hook_params=("page", "ctx"),
     ),
     # (CHAOS_BASE_DEPLOY, OIDC_AT_INSTALL and SKIP_GENERIC were deleted in restructure P2:
     # compose.ccci.yml is first-class + auto-chaos; install-time deps wiring is the only mode;
@@ -145,6 +156,28 @@ def meta_path(recipe: str, tests_dir: str | None = None) -> str:
     return os.path.join(tests_dir or TESTS_DIR, recipe, "recipe_meta.py")
 
 
+def check_hook_signature(fn, expected: tuple[str, ...], where: str) -> None:
+    """Enforce the uniform ctx hook convention (rcust P3): a hook callable's positional parameters
+    must be exactly `expected` (e.g. ("ctx",) or ("page", "ctx")). A legacy-signature hook (the
+    pre-restructure `(domain)` / `(domain, meta)` / `(page, domain, meta)` forms) raises a CLEAR
+    MetaError naming the migration — never a silent TypeError mid-run."""
+    try:
+        params = [
+            p.name
+            for p in inspect.signature(fn).parameters.values()
+            if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)
+        ]
+    except (TypeError, ValueError):  # builtins/odd callables — let the call site surface it
+        return
+    if tuple(params) != expected:
+        raise MetaError(
+            f"{where}: hook signature is ({', '.join(params)}) — the recipe-customization "
+            f"restructure (P3) changed ALL recipe hook signatures to ({', '.join(expected)}); "
+            f"read fields off the HookCtx (ctx.domain, ctx.base_url, ctx.meta, ctx.deps, ctx.op). "
+            f"See docs/recipe-customization.md §5."
+        )
+
+
 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."""
@@ -209,6 +242,8 @@ def load(recipe: str, tests_dir: str | None = None):
                     f"underscore-prefixed (e.g. _{name})."
                 )
             values[name] = _coerce(key, ns[name], path)
+            if key.hook_params and callable(values[name]):
+                check_hook_signature(values[name], key.hook_params, f"{path}: {name}")
             if key.validate:
                 key.validate(values[name])
     return RecipeMeta(**values)
@@ -231,17 +266,55 @@ def non_default(meta) -> dict:
     return out
 
 
-def _env_map(value, domain: str) -> dict[str, str]:
+@dataclasses.dataclass(frozen=True)
+class HookCtx:
+    """The single argument every recipe hook receives (rcust P3 uniform ctx convention):
+    `EXTRA_ENV(ctx)`, `UPGRADE_EXTRA_ENV(ctx)`, `READY_PROBE(ctx)`, `BACKUP_VERIFY(ctx)`,
+    `SCREENSHOT(page, ctx)`, ops.py `pre_<op>(ctx)`."""
+
+    domain: str  # the app's per-run domain
+    base_url: str  # https://<domain>
+    meta: object  # the recipe's full RecipeMeta
+    deps: dict | None  # provisioned dep creds ({dep_recipe: entry}) or None if absent/empty
+    op: str | None  # current lifecycle op (install|upgrade|backup|restore) or None
+
+
+def _run_deps() -> dict | None:
+    """The current run's provisioned dep creds from $CCCI_DEPS_FILE (either shape), or None.
+    Read directly (not via harness.deps) to keep meta.py import-cycle-free."""
+    path = os.environ.get("CCCI_DEPS_FILE")
+    if not path or not os.path.exists(path):
+        return None
+    try:
+        with open(path) as f:
+            data = json.load(f)
+    except (OSError, ValueError):
+        return None
+    if isinstance(data, dict):
+        return data or None
+    if isinstance(data, list):
+        out = {e["recipe"]: e for e in data if isinstance(e, dict) and e.get("recipe")}
+        return out or None
+    return None
+
+
+def hook_ctx(domain: str, meta, *, op: str | None = None) -> HookCtx:
+    """Build the HookCtx for a hook call site. Dep creds are picked up from the run's
+    $CCCI_DEPS_FILE when present (None otherwise)."""
+    return HookCtx(domain=domain, base_url=f"https://{domain}", meta=meta, deps=_run_deps(), op=op)
+
+
+def _env_map(value, ctx: HookCtx) -> dict[str, str]:
     if callable(value):
-        value = value(domain)
+        value = value(ctx)
     return {str(k): str(v) for k, v in (value or {}).items()}
 
 
-def extra_env(meta, domain: str) -> dict[str, str]:
+def extra_env(meta, ctx: HookCtx) -> dict[str, str]:
-    """Resolve EXTRA_ENV (dict or callable(domain)->dict) to the concrete per-run env map."""
+    """Resolve EXTRA_ENV (dict or callable(ctx)->dict) to the concrete per-run env map."""
-    return _env_map(meta.EXTRA_ENV, domain)
+    return _env_map(meta.EXTRA_ENV, ctx)
 
 
-def upgrade_extra_env(meta, domain: str) -> dict[str, str]:
+def upgrade_extra_env(meta, ctx: HookCtx) -> dict[str, str]:
-    """Resolve UPGRADE_EXTRA_ENV (dict or callable(domain)->dict) to the concrete env map."""
+    """Resolve UPGRADE_EXTRA_ENV (dict or callable(ctx)->dict) to the concrete env map."""
-    return _env_map(meta.UPGRADE_EXTRA_ENV, domain)
+    return _env_map(meta.UPGRADE_EXTRA_ENV, ctx)

runner/harness/screenshot.py

@@ -8,7 +8,7 @@ Secret-safety (R7, the cardinal screenshot guardrail): the screenshot step must
 that displays generated credentials (an install wizard showing the initial admin password, a secrets
 page, etc.). The DEFAULT capture is the app's **landing page** (a login form shows fields, not the
 password) — safe for every recipe. A recipe that needs a post-login view opts in via a recipe-meta
-`SCREENSHOT` hook: a callable `screenshot(page, domain, meta) -> None` that drives Playwright to a
+`SCREENSHOT` hook: a callable `SCREENSHOT(page, ctx) -> None` that drives Playwright to a
 safe, credential-free view and is responsible for not landing on a secrets page. The harness never
 auto-fills a wizard.
 
@@ -21,6 +21,7 @@ from __future__ import annotations
 import os
 
 from . import browser as harness_browser
+from . import meta as meta_mod
 
 # Default viewport for the captured screenshot — a desktop-ish frame that crops well into the card.
 VIEWPORT = {"width": 1280, "height": 800}
@@ -74,8 +75,9 @@ def capture(domain: str, out_path: str, *, recipe_meta: dict | None = None) -> s
                 if hook is not None:
                     # Recipe-specific safe view (post-login etc.). The hook owns navigation +
                     # the no-secret-page guarantee; it should call page.screenshot itself, but if
-                    # it doesn't, we still snap the resulting page below.
+                    # it doesn't, we still snap the resulting page below. SCREENSHOT(page, ctx) —
-                    hook(page, domain, recipe_meta)
+                    # the uniform ctx convention (rcust P3).
+                    hook(page, meta_mod.hook_ctx(domain, recipe_meta))
                     if not os.path.exists(out_path):
                         page.screenshot(path=out_path, full_page=False)
                 else:

runner/run_recipe_ci.py

@@ -289,7 +289,11 @@ def _run_pre_hook(recipe: str, op: str, repo_local: str | None, domain: str, met
         mod = importlib.util.module_from_spec(spec)
         spec.loader.exec_module(mod)
         print(f"  pre-op seed ({source}): {os.path.relpath(path, ROOT)}::pre_{op}", flush=True)
-        getattr(mod, f"pre_{op}")(domain, meta)
+        fn = getattr(mod, f"pre_{op}")
+        # Uniform ctx convention (rcust P3): pre_<op>(ctx). A legacy (domain, meta) hook fails
+        # HERE with a clear migration message, not a TypeError mid-call.
+        meta_mod.check_hook_signature(fn, ("ctx",), f"{os.path.relpath(path, ROOT)}::pre_{op}")
+        fn(meta_mod.hook_ctx(domain, meta, op=op))
     finally:
         if d in sys.path:
             sys.path.remove(d)
@@ -326,8 +330,9 @@ def _perform_op(
         # 3 attempts. Recipes without BACKUP_VERIFY are unaffected (single backup, as before).
         snap = generic.perform_backup(domain)
         verify = meta.BACKUP_VERIFY if meta else None
+        verify_ctx = meta_mod.hook_ctx(domain, meta, op="backup") if meta else None
         attempt = 1
-        while callable(verify) and not verify(domain) and attempt < 3:
+        while callable(verify) and not verify(verify_ctx) and attempt < 3:
             attempt += 1
             print(
                 f"  backup-verify FAILED (attempt {attempt - 1}/3) — backup did not capture the "
@@ -335,7 +340,7 @@ def _perform_op(
                 flush=True,
             )
             snap = generic.perform_backup(domain)
-        if callable(verify) and not verify(domain):
+        if callable(verify) and not verify(verify_ctx):
             print(
                 f"  !! backup-verify still FAILED after {attempt} attempts — backup is incomplete",
                 flush=True,
@@ -992,7 +997,9 @@ def main() -> int:
             )
             # 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.DEPLOY_TIMEOUT))
+            lifecycle.wait_ready_probes(
+                meta, domain, timeout=int(meta.DEPLOY_TIMEOUT), op="install"
+            )
             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)

tests/bluesky-pds/ops.py

@@ -9,14 +9,14 @@ sys.path.insert(0, os.path.dirname(__file__))
 import _p4  # noqa: E402
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _p4.create_account(domain)
+    _p4.create_account(ctx.domain)
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _p4.create_account(domain)
+    _p4.create_account(ctx.domain)
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
-    _p4.delete_account(domain)
+    _p4.delete_account(ctx.domain)
-    assert not _p4.account_exists(domain), "marker account delete did not take (pre_restore)"
+    assert not _p4.account_exists(ctx.domain), "marker account delete did not take (pre_restore)"

tests/cryptpad/ops.py

@@ -15,13 +15,13 @@ def _write(domain, val):
     lifecycle.exec_in_app(domain, ["sh", "-c", f"echo {val} > {MARKER}"])
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _write(domain, "upgrade-survives")
+    _write(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _write(domain, "original")
+    _write(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
-    _write(domain, "mutated")  # diverge so a successful restore is observable
+    _write(ctx.domain, "mutated")  # diverge so a successful restore is observable

tests/cryptpad/recipe_meta.py

@@ -7,9 +7,9 @@ DEPLOY_TIMEOUT = 600
 HTTP_TIMEOUT = 600
 
 
-def EXTRA_ENV(domain):
+def EXTRA_ENV(ctx):
     """cryptpad needs a SANDBOX_DOMAIN distinct from the main DOMAIN (it serves user content from a
     separate origin; the web router routes both). Derive a sibling subdomain under the same wildcard
     (covered by the wildcard cert, so no cert work)."""
-    label, _, rest = domain.partition(".")
+    label, _, rest = ctx.domain.partition(".")
     return {"SANDBOX_DOMAIN": f"{label}-sb.{rest}"}

tests/custom-html-bkp-bad/ops.py

@@ -12,8 +12,8 @@ from harness import lifecycle
 MARKER_PATH = "/usr/share/nginx/html/ci-marker.txt"
 
 
-def pre_restore(domain: str, meta: dict) -> None:
+def pre_restore(ctx) -> None:
     """Write 'mutated' to the marker before restore runs. If restore brings back the
     snapshot (which has no marker — never seeded by pre_backup), the marker ends up
     MISSING or 'mutated' after restore → test_restore_returns_state FAILS → restore=RED."""
-    lifecycle.exec_in_app(domain, ["sh", "-c", f"echo mutated > {MARKER_PATH}"])
+    lifecycle.exec_in_app(ctx.domain, ["sh", "-c", f"echo mutated > {MARKER_PATH}"])

tests/custom-html-rst-bad/ops.py

@@ -11,5 +11,5 @@ from harness import lifecycle
 MARKER_PATH = "/usr/share/nginx/html/ci-marker.txt"
 
 
-def pre_restore(domain: str, meta: dict) -> None:
+def pre_restore(ctx) -> None:
-    lifecycle.exec_in_app(domain, ["sh", "-c", f"echo mutated > {MARKER_PATH}"])
+    lifecycle.exec_in_app(ctx.domain, ["sh", "-c", f"echo mutated > {MARKER_PATH}"])

tests/custom-html/ops.py

@@ -1,4 +1,4 @@
-"""custom-html — pre-op seed hooks (Phase 1e HC3). The orchestrator runs `pre_<op>(domain, meta)`
+"""custom-html — pre-op seed hooks (Phase 1e HC3). The orchestrator runs `pre_<op>(ctx)`
 BEFORE it performs the op; the matching test_<op>.py asserts the post-op state (assertion-only).
 
 nginx serves the volume at /usr/share/nginx/html, so the marker file survives an upgrade / a
@@ -17,16 +17,16 @@ def _write(domain: str, val: str) -> None:
     lifecycle.exec_in_app(domain, ["sh", "-c", f"echo {val} > {MARKER_PATH}"])
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
     # seed a marker before the upgrade so the overlay can prove the data survives it
-    _write(domain, "upgrade-survives")
+    _write(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
     # establish a known original state before the backup op captures it
-    _write(domain, "original")
+    _write(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
     # diverge from the backed-up state so a successful restore (back to "original") is observable
-    _write(domain, "mutated")
+    _write(ctx.domain, "mutated")

tests/discourse/ops.py

@@ -30,18 +30,18 @@ def _seed(domain, value):
     assert got == value, f"seed did not commit (read back {got!r}, expected {value!r})"
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _seed(domain, "upgrade-survives")
+    _seed(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _seed(domain, "original")
+    _seed(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
     # diverge from the backup so a successful restore is observable
-    _psql(domain, "DROP TABLE IF EXISTS ci_marker;")
+    _psql(ctx.domain, "DROP TABLE IF EXISTS ci_marker;")
-    assert _psql(domain, "SELECT to_regclass('public.ci_marker');") in (
+    assert _psql(ctx.domain, "SELECT to_regclass('public.ci_marker');") in (
         "",
         "NULL",
     ), "drop did not take"

tests/discourse/recipe_meta.py

@@ -41,7 +41,7 @@ EXTRA_ENV = {
 }
 
 
-def BACKUP_VERIFY(domain):
+def BACKUP_VERIFY(ctx):
     """Post-backup integrity check (Q4.6, same race ghost F2-14b hit). The recipe's backupbot db
     pre-hook (`/pg_backup.sh backup`) dumps the discourse postgres DB to `/var/lib/postgresql/data/
     backup.sql` (gzip), then restic captures that path. On the loaded single CI node the db container
@@ -60,7 +60,7 @@ def BACKUP_VERIFY(domain):
 
     try:
         out = lifecycle.exec_in_app(
-            domain,
+            ctx.domain,
             [
                 "sh",
                 "-c",

tests/ghost/ops.py

@@ -36,19 +36,19 @@ def _seed(domain, value):
     assert got == value, f"seed did not commit (read back {got!r}, expected {value!r})"
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _seed(domain, "upgrade-survives")
+    _seed(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _seed(domain, "original")
+    _seed(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
     # diverge from the backup so a successful restore is observable: drop the marker table.
-    _mysql(domain, "DROP TABLE IF EXISTS ci_marker;")
+    _mysql(ctx.domain, "DROP TABLE IF EXISTS ci_marker;")
     got = _mysql(
-        domain,
+        ctx.domain,
         "SELECT COUNT(*) FROM information_schema.tables "
         "WHERE table_schema='ghost' AND table_name='ci_marker';",
     )

tests/ghost/recipe_meta.py

@@ -46,7 +46,7 @@ EXTRA_ENV = {
 }
 
 
-def BACKUP_VERIFY(domain):
+def BACKUP_VERIFY(ctx):
     """Post-backup integrity check (F2-14b). The recipe's backupbot db pre-hook dumps the ghost MySQL
     DB to `/var/lib/mysql/backup.sql.gz` (then restic captures that path). On the loaded single CI node
     the db container intermittently CYCLES mid-dump (observed: full5/6/7 RED, full8 green — pure race;
@@ -61,7 +61,7 @@ def BACKUP_VERIFY(domain):
 
     try:
         out = lifecycle.exec_in_app(
-            domain,
+            ctx.domain,
             [
                 "sh",
                 "-c",

tests/immich/ops.py

@@ -25,17 +25,17 @@ def _seed(domain, value):
     assert _psql(domain, "SELECT v FROM ci_marker;") == value
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _seed(domain, "upgrade-survives")
+    _seed(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _seed(domain, "original")
+    _seed(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
-    _psql(domain, "DROP TABLE ci_marker;")
+    _psql(ctx.domain, "DROP TABLE ci_marker;")
-    assert _psql(domain, "SELECT to_regclass('public.ci_marker');") in (
+    assert _psql(ctx.domain, "SELECT to_regclass('public.ci_marker');") in (
         "",
         "NULL",
     ), "drop did not take"

tests/keycloak/ops.py

@@ -14,20 +14,20 @@ def _token(domain):
     return kc_admin.admin_token(domain, kc_admin.admin_password(domain))
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
     # create the marker realm (DB data) before the upgrade so the overlay can prove it survives
-    assert kc_admin.create_marker_realm(domain, _token(domain)) in (201, 409)
+    assert kc_admin.create_marker_realm(ctx.domain, _token(ctx.domain)) in (201, 409)
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
     # establish the marker realm before the backup op captures mariadb
-    assert kc_admin.create_marker_realm(domain, _token(domain)) in (201, 409)
+    assert kc_admin.create_marker_realm(ctx.domain, _token(ctx.domain)) in (201, 409)
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
     # backup-bot-two cycles the keycloak container during backup → wait for serving, re-auth, then
     # delete the realm (diverge from the backup) so a successful restore is observable
-    generic.assert_serving(domain, meta)
+    generic.assert_serving(ctx.domain, ctx.meta)
-    tok = _token(domain)
+    tok = _token(ctx.domain)
-    assert kc_admin.delete_marker_realm(domain, tok) in (204, 200)
+    assert kc_admin.delete_marker_realm(ctx.domain, tok) in (204, 200)
-    assert not kc_admin.marker_realm_exists(domain, tok), "delete did not take"
+    assert not kc_admin.marker_realm_exists(ctx.domain, tok), "delete did not take"

tests/lasuite-docs/ops.py

@@ -24,18 +24,18 @@ def _seed(domain, value):
     assert _psql(domain, "SELECT v FROM ci_marker;") == value
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _seed(domain, "upgrade-survives")
+    _seed(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _seed(domain, "original")
+    _seed(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
     # drop the marker table (diverge from the backup) so a successful restore is observable
-    _psql(domain, "DROP TABLE ci_marker;")
+    _psql(ctx.domain, "DROP TABLE ci_marker;")
-    assert _psql(domain, "SELECT to_regclass('public.ci_marker');") in (
+    assert _psql(ctx.domain, "SELECT to_regclass('public.ci_marker');") in (
         "",
         "NULL",
     ), "drop did not take"

tests/lasuite-docs/recipe_meta.py

@@ -15,7 +15,7 @@ HTTP_TIMEOUT = 600
 DEPS = ["keycloak"]
 
 
-def EXTRA_ENV(domain):
+def EXTRA_ENV(ctx):
     # abra's internal per-deploy convergence timeout (the recipe's TIMEOUT env, default 300s) is too
     # short for this 9-service stack on a COLD image cache (~9 large images: impress frontend/backend,
     # minio, postgres18, redis, docspec, y-provider). Cold pulls exceed 300s -> "deploy timed out 🟠".

tests/lasuite-drive/ops.py

@@ -13,14 +13,14 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "runner")
 from harness import lifecycle  # noqa: E402
 
 
-def pre_install(domain, meta):
+def pre_install(ctx):
     """Post-deploy seed for the custom tier (the former setup_custom_tests.sh, moved here in rcust
     P2b — install_steps.sh runs PRE-deploy and cannot touch the live stack). The deploy alone does
     NOT create the MinIO bucket: `minio-createbuckets` is a `replicas:0` one-shot (restart_policy:
     none) that must be triggered. The MinIO storage test asserts the bucket exists, so trigger it
     here and poll. `--detach` is REQUIRED: the job creates the bucket then EXITS 0, so it never
     holds a steady 1/1 replica — a blocking scale would wait forever."""
-    stack = domain.replace(".", "_")
+    stack = ctx.domain.replace(".", "_")
     print("  pre_install: creating MinIO bucket via the minio-createbuckets one-shot", flush=True)
     subprocess.run(
         ["docker", "service", "scale", "--detach", f"{stack}_minio-createbuckets=1"],
@@ -91,21 +91,21 @@ def _seed(domain, value):
     assert _psql(domain, "SELECT v FROM ci_marker;") == value
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
     # Gate the chaos redeploy on a fully-ready collabora (else it kills a still-booting coolwsd and
     # abra aborts the upgrade deploy — Q3.2a run 1). Then seed the data-integrity marker.
-    _wait_collabora_ready(domain)
+    _wait_collabora_ready(ctx.domain)
-    _seed(domain, "upgrade-survives")
+    _seed(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _seed(domain, "original")
+    _seed(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
     # drop the marker table (diverge from the backup) so a successful restore is observable
-    _psql(domain, "DROP TABLE ci_marker;")
+    _psql(ctx.domain, "DROP TABLE ci_marker;")
-    assert _psql(domain, "SELECT to_regclass('public.ci_marker');") in (
+    assert _psql(ctx.domain, "SELECT to_regclass('public.ci_marker');") in (
         "",
         "NULL",
     ), "drop did not take"

tests/lasuite-drive/recipe_meta.py

@@ -31,18 +31,18 @@ DEPS = ["keycloak"]
 # pre_install (the former setup_custom_tests.sh, deleted in P2b).
 
 
-def READY_PROBE(domain):
+def READY_PROBE(ctx):
     """Readiness signals beyond replica-convergence + the app HEALTH_PATH (Q3.2/F2-12). collabora's
     coolwsd reports its container 1/1 'running' while still doing jail/config init, and its WOPI
     discovery endpoint 404s until ready — so the harness waits for `/hosting/discovery` → 200 on the
     collabora sibling host after the install deploy AND after the upgrade chaos redeploy. This is what
     makes the heavy prev→PR-head crossover reliably green (the new collabora 25.04.9.x finishes init
     within swarm's healthcheck retries; abra's own converge monitor was too impatient — F2-12)."""
-    label, _, rest = domain.partition(".")
+    label, _, rest = ctx.domain.partition(".")
-    return [{"host": f"collabora-{domain}", "path": "/hosting/discovery", "ok": (200,)}]
+    return [{"host": f"collabora-{ctx.domain}", "path": "/hosting/discovery", "ok": (200,)}]
 
 
-def EXTRA_ENV(domain):
+def EXTRA_ENV(ctx):
     # Two of lasuite-drive's services route on DOMAIN-DERIVED **nested** subdomains —
     # `MINIO_DOMAIN="minio.${DOMAIN}"` and `COLLABORA_DOMAIN="collabora.${DOMAIN}"`. The cc-ci
     # wildcard TLS cert is `*.ci.commoninternet.net` (single label only), so a 2-label name like
@@ -52,8 +52,8 @@ def EXTRA_ENV(domain):
     # no cert/gateway change. See DECISIONS.md "Phase 2 — nested DOMAIN-derived subdomains".
     # `AWS_S3_DOMAIN_REPLACE` derives from MINIO_DOMAIN in-compose, so setting MINIO_DOMAIN is enough.
     return {
-        "MINIO_DOMAIN": f"minio-{domain}",
+        "MINIO_DOMAIN": f"minio-{ctx.domain}",
-        "COLLABORA_DOMAIN": f"collabora-{domain}",
+        "COLLABORA_DOMAIN": f"collabora-{ctx.domain}",
         # abra's internal per-deploy convergence timeout (recipe TIMEOUT env, default 300s) is too
         # short for this 12-service stack on a cold image cache (impress frontend/backend, minio,
         # postgres, redis, collabora ~1GB, onlyoffice ~2GB). Bump so abra waits long enough for

tests/lasuite-meet/ops.py

@@ -27,18 +27,18 @@ def _seed(domain, value):
     assert _psql(domain, "SELECT v FROM ci_marker;") == value
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _seed(domain, "upgrade-survives")
+    _seed(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _seed(domain, "original")
+    _seed(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
     # drop the marker table (diverge from the backup) so a successful restore is observable
-    _psql(domain, "DROP TABLE ci_marker;")
+    _psql(ctx.domain, "DROP TABLE ci_marker;")
-    assert _psql(domain, "SELECT to_regclass('public.ci_marker');") in (
+    assert _psql(ctx.domain, "SELECT to_regclass('public.ci_marker');") in (
         "",
         "NULL",
     ), "drop did not take"

tests/lasuite-meet/recipe_meta.py

@@ -21,7 +21,7 @@ HTTP_TIMEOUT = 600
 DEPS = ["keycloak"]
 
 
-def EXTRA_ENV(domain):
+def EXTRA_ENV(ctx):
     # lasuite-meet routes LiveKit's WebSocket signaling on a DOMAIN-derived **nested** subdomain
     # `LIVEKIT_DOMAIN="livekit.${DOMAIN}"`. The cc-ci wildcard TLS cert is `*.ci.commoninternet.net`
     # (single label only), so a 2-label name like `livekit.lasuite-meet-pr0-abc.ci.commoninternet.net`
@@ -30,7 +30,7 @@ def EXTRA_ENV(domain):
     # no cert/gateway change. Same fix as lasuite-drive's minio/collabora siblings (DECISIONS.md
     # "Phase 2 — nested DOMAIN-derived subdomains").
     return {
-        "LIVEKIT_DOMAIN": f"livekit-{domain}",
+        "LIVEKIT_DOMAIN": f"livekit-{ctx.domain}",
         # abra's internal per-deploy convergence TIMEOUT (default 300s) is too short for this stack on
         # a cold image cache; bump it (kept under DEPLOY_TIMEOUT so Python never kills abra mid-wait).
         "TIMEOUT": "1000",

tests/mailu/recipe_meta.py

@@ -21,10 +21,10 @@ DEPLOY_TIMEOUT = 900
 HTTP_TIMEOUT = 600
 
 
-def EXTRA_ENV(domain):
+def EXTRA_ENV(ctx):
     return {
-        "MAIL_DOMAIN": domain,
+        "MAIL_DOMAIN": ctx.domain,
-        "HOSTNAMES": domain,
+        "HOSTNAMES": ctx.domain,
         "TRAEFIK_STACK_NAME": "traefik_ci_commoninternet_net",
         "TLS_FLAVOR": "notls",
         "SITENAME": "ccci-mail",

tests/matrix-synapse/ops.py

@@ -24,18 +24,18 @@ def _seed(domain, value):
     assert _psql(domain, "SELECT v FROM ci_marker;") == value
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _seed(domain, "upgrade-survives")
+    _seed(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _seed(domain, "original")
+    _seed(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
     # drop the marker table (diverge from the backup) so a successful restore is observable
-    _psql(domain, "DROP TABLE ci_marker;")
+    _psql(ctx.domain, "DROP TABLE ci_marker;")
-    assert _psql(domain, "SELECT to_regclass('public.ci_marker');") in (
+    assert _psql(ctx.domain, "SELECT to_regclass('public.ci_marker');") in (
         "",
         "NULL",
     ), "drop did not take"

tests/mattermost-lts/ops.py

@@ -29,18 +29,18 @@ def _seed(domain, value):
     assert _psql(domain, "SELECT v FROM ci_marker;") == value
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _seed(domain, "upgrade-survives")
+    _seed(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _seed(domain, "original")
+    _seed(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
     # drop the marker table (diverge from the backup) so a successful restore is observable
-    _psql(domain, "DROP TABLE ci_marker;")
+    _psql(ctx.domain, "DROP TABLE ci_marker;")
-    assert _psql(domain, "SELECT to_regclass('public.ci_marker');") in (
+    assert _psql(ctx.domain, "SELECT to_regclass('public.ci_marker');") in (
         "",
         "NULL",
     ), "drop did not take"

tests/mumble/ops.py

@@ -38,16 +38,18 @@ def _seed(domain, value):
     assert got == value, f"seed did not commit (read back {got!r}, expected {value!r})"
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _seed(domain, "upgrade-survives")
+    _seed(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _seed(domain, "original")
+    _seed(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
     # diverge from the backup so a successful restore is observable: drop the marker table.
-    _sqlite(domain, "DROP TABLE IF EXISTS ci_marker;")
+    _sqlite(ctx.domain, "DROP TABLE IF EXISTS ci_marker;")
-    got = _sqlite(domain, "SELECT name FROM sqlite_master WHERE type='table' AND name='ci_marker';")
+    got = _sqlite(
+        ctx.domain, "SELECT name FROM sqlite_master WHERE type='table' AND name='ci_marker';"
+    )
     assert got == "", f"drop did not take (sqlite_master still lists ci_marker: {got!r})"

tests/mumble/recipe_meta.py

@@ -53,7 +53,7 @@ UPGRADE_EXTRA_ENV = {
 }
 
 
-def READY_PROBE(domain):
+def READY_PROBE(ctx):
     # The voice server on 64738 is testable on-host ONLY when compose.host-ports.yml is active — i.e.
     # the post-upgrade LATEST, not the minimal 0.2.0 base. Read the live COMPOSE_FILE to decide, so the
     # SAME probe fn is correct in both phases: the post-install probe (base, no host-ports) returns []
@@ -64,7 +64,7 @@ def READY_PROBE(domain):
     # backup-bot would then exec into a not-running app container -> 409).
     from harness import abra  # lazy: recipe_meta is exec'd with `harness` importable at call time
 
-    cf = abra.env_get(domain, "COMPOSE_FILE") or ""
+    cf = abra.env_get(ctx.domain, "COMPOSE_FILE") or ""
     if "compose.host-ports.yml" in cf:
         return [{"tcp_host": "127.0.0.1", "tcp_port": 64738, "stable": 3}]
     return []

tests/n8n/ops.py

@@ -15,13 +15,13 @@ def _write(domain, val):
     lifecycle.exec_in_app(domain, ["sh", "-c", f"echo {val} > {MARKER}"])
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _write(domain, "upgrade-survives")
+    _write(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _write(domain, "original")
+    _write(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
-    _write(domain, "mutated")  # diverge so a successful restore is observable
+    _write(ctx.domain, "mutated")  # diverge so a successful restore is observable

tests/plausible/ops.py

@@ -24,17 +24,17 @@ def _seed(domain, value):
     assert _psql(domain, "SELECT v FROM ci_marker;") == value
 
 
-def pre_upgrade(domain, meta):
+def pre_upgrade(ctx):
-    _seed(domain, "upgrade-survives")
+    _seed(ctx.domain, "upgrade-survives")
 
 
-def pre_backup(domain, meta):
+def pre_backup(ctx):
-    _seed(domain, "original")
+    _seed(ctx.domain, "original")
 
 
-def pre_restore(domain, meta):
+def pre_restore(ctx):
-    _psql(domain, "DROP TABLE ci_marker;")
+    _psql(ctx.domain, "DROP TABLE ci_marker;")
-    assert _psql(domain, "SELECT to_regclass('public.ci_marker');") in (
+    assert _psql(ctx.domain, "SELECT to_regclass('public.ci_marker');") in (
         "",
         "NULL",
     ), "drop did not take"

tests/unit/test_f212_upgrade_convergence.py

@@ -34,10 +34,12 @@ def _fake_clock(monkeypatch):
 
 
 # RecipeMeta (rcust P1: wait_ready_probes reads meta.READY_PROBE off the loaded object); defaults
-# + the drive-style probe hook.
+# + the drive-style probe hook (P3 ctx signature: the probe receives a HookCtx).
 _DRIVE_META = dataclasses.replace(
     harness_meta.load("ccci-no-such-recipe"),
-    READY_PROBE=lambda d: [{"host": f"collabora-{d}", "path": "/hosting/discovery", "ok": (200,)}],
+    READY_PROBE=lambda ctx: [
+        {"host": f"collabora-{ctx.domain}", "path": "/hosting/discovery", "ok": (200,)}
+    ],
 )
 _NO_PROBE_META = harness_meta.load("ccci-no-such-recipe")
 

tests/unit/test_meta.py

@@ -143,10 +143,11 @@ def test_underscore_names_are_private_and_exempt(tmp_path):
 def test_lowercase_helpers_ignored(tmp_path):
     r = _write_meta(
         tmp_path,
-        "def _helper(d):\n    return {'K': d}\n\ndef EXTRA_ENV(domain):\n    return _helper(domain)\n",
+        "def _helper(d):\n    return {'K': d}\n\ndef EXTRA_ENV(ctx):\n    return _helper(ctx.domain)\n",
     )
     meta = meta_mod.load(r, tests_dir=str(tmp_path))
-    assert meta_mod.extra_env(meta, "x.example") == {"K": "x.example"}
+    ctx = meta_mod.hook_ctx("x.example", meta)
+    assert meta_mod.extra_env(meta, ctx) == {"K": "x.example"}
 
 
 # ---- normalization + helpers --------------------------------------------------------------------
@@ -160,13 +161,85 @@ def test_health_ok_list_normalized_to_tuple(tmp_path):
 def test_extra_env_dict_and_callable_forms(tmp_path):
     r = _write_meta(tmp_path, "EXTRA_ENV = {'A': 1}\n")
     meta = meta_mod.load(r, tests_dir=str(tmp_path))
-    assert meta_mod.extra_env(meta, "d") == {"A": "1"}  # values stringified
+    assert meta_mod.extra_env(meta, meta_mod.hook_ctx("d", meta)) == {"A": "1"}  # stringified
     r2 = _write_meta(
-        tmp_path, "UPGRADE_EXTRA_ENV = lambda domain: {'COMPOSE_FILE': domain}\n", recipe="r2"
+        tmp_path, "UPGRADE_EXTRA_ENV = lambda ctx: {'COMPOSE_FILE': ctx.domain}\n", recipe="r2"
     )
     meta2 = meta_mod.load(r2, tests_dir=str(tmp_path))
-    assert meta_mod.upgrade_extra_env(meta2, "dom.x") == {"COMPOSE_FILE": "dom.x"}
+    ctx2 = meta_mod.hook_ctx("dom.x", meta2, op="upgrade")
-    assert meta_mod.extra_env(meta2, "dom.x") == {}  # unset EXTRA_ENV resolves to {}
+    assert meta_mod.upgrade_extra_env(meta2, ctx2) == {"COMPOSE_FILE": "dom.x"}
+    assert meta_mod.extra_env(meta2, ctx2) == {}  # unset EXTRA_ENV resolves to {}
+
+
+# ---- P3: uniform ctx hook convention -------------------------------------------------------------
+
+
+def test_hook_ctx_fields(tmp_path):
+    meta = meta_mod.load("no-such", tests_dir=str(tmp_path))
+    ctx = meta_mod.hook_ctx("app.ci.example", meta, op="backup")
+    assert ctx.domain == "app.ci.example"
+    assert ctx.base_url == "https://app.ci.example"
+    assert ctx.meta is meta
+    assert ctx.op == "backup"
+    assert meta_mod.hook_ctx("d", meta).op is None
+
+
+def test_hook_ctx_deps_from_run_file(tmp_path, monkeypatch):
+    import json
+
+    meta = meta_mod.load("no-such", tests_dir=str(tmp_path))
+    monkeypatch.delenv("CCCI_DEPS_FILE", raising=False)
+    assert meta_mod.hook_ctx("d", meta).deps is None
+    f = tmp_path / "deps.json"
+    f.write_text(json.dumps({"keycloak": {"recipe": "keycloak", "domain": "kc.x"}}))
+    monkeypatch.setenv("CCCI_DEPS_FILE", str(f))
+    deps = meta_mod.hook_ctx("d", meta).deps
+    assert deps["keycloak"]["domain"] == "kc.x"
+    f.write_text("{}")  # empty dict -> None (deps declared but not provisioned)
+    assert meta_mod.hook_ctx("d", meta).deps is None
+
+
+def test_legacy_hook_signature_raises_clear_meta_error(tmp_path):
+    """A pre-restructure hook signature must fail AT LOAD with a migration message — never a
+    silent TypeError mid-run (P3.4)."""
+    r = _write_meta(tmp_path, "def READY_PROBE(domain):\n    return []\n")
+    with pytest.raises(MetaError, match="ctx"):
+        meta_mod.load(r, tests_dir=str(tmp_path))
+    r2 = _write_meta(tmp_path, "EXTRA_ENV = lambda domain: {}\n", recipe="r2")
+    with pytest.raises(MetaError, match="restructure"):
+        meta_mod.load(r2, tests_dir=str(tmp_path))
+    r3 = _write_meta(
+        tmp_path, "def SCREENSHOT(page, domain, meta):\n    return None\n", recipe="r3"
+    )
+    with pytest.raises(MetaError, match="page, ctx"):
+        meta_mod.load(r3, tests_dir=str(tmp_path))
+
+
+def test_ctx_hook_signatures_accepted(tmp_path):
+    r = _write_meta(
+        tmp_path,
+        "def READY_PROBE(ctx):\n    return []\n"
+        "def BACKUP_VERIFY(ctx):\n    return True\n"
+        "def SCREENSHOT(page, ctx):\n    return None\n"
+        "def EXTRA_ENV(ctx):\n    return {}\n",
+    )
+    meta = meta_mod.load(r, tests_dir=str(tmp_path))
+    assert callable(meta.READY_PROBE) and callable(meta.SCREENSHOT)
+
+
+def test_check_hook_signature_for_pre_op_hooks():
+    """The orchestrator validates ops.py pre_<op> hooks with the same checker (legacy
+    (domain, meta) form names the migration)."""
+
+    def legacy(domain, meta):
+        pass
+
+    def new(ctx):
+        pass
+
+    with pytest.raises(MetaError, match="ctx"):
+        meta_mod.check_hook_signature(legacy, ("ctx",), "tests/x/ops.py::pre_upgrade")
+    meta_mod.check_hook_signature(new, ("ctx",), "tests/x/ops.py::pre_upgrade")  # no raise
 
 
 def test_non_default_reports_only_customized_keys(tmp_path):