You want a check at apply time, no problem

The check you want to enforce can't run at validate time — it depends on something only true at deploy. Common shapes:

• "The deployer must have a specific secret in their env."
• "The image referenced by the contract must be signed by my-org's release key."
• "The scaffold bundle this contract was generated from must not have drifted since the last fluid custom-scaffold."
• "Production deploys must be approved by a human in the last 24 hours."

A Validator plugin can't help here — it runs on contract authors, who don't have any of that state. Apply hooks are the tool: they run inside fluid apply, after the contract is loaded, before any provider executes.

This guide walks through writing one from scratch. By the end you'll have:

• An apply hook that runs every fluid apply and refuses to proceed if its invariant is violated.
• A documented escape hatch (--force-pattern-drift) for emergencies.
• Tests that exercise both the pass and fail paths.

Realistic time end-to-end: 15–20 minutes.

The mental model

fluid apply contract.fluid.yaml --env prod
                │
                ▼
   ┌──────────────────────────────────┐
   │ 1. Load contract                  │
   │ 2. Run apply hooks ◄──────── your plugin runs here
   │      ↳ each gets a DEEP COPY      │
   │        of the contract            │
   │      ↳ each can append to errors  │
   │ 3. If any errors:                 │
   │      ↳ --force-pattern-drift?     │
   │          yes → log WARNINGs       │
   │          no  → abort with exit 1  │
   │ 4. Run provider apply()           │
   │ 5. Run policy-apply, verify, …    │
   └──────────────────────────────────┘

Three things to know about the hook contract:

1. The base signature is hook(contract_dir: Path, contract: Dict, errors: List[str]) -> None. Append to errors to fail; leave empty to pass. A hook can opt in to the resolved --env by declaring an env parameter (see Step 3).
2. The contract is a deep copy. You can read or even mutate it inside the hook — the rest of fluid apply sees the original. (See trust model for why.)
3. Plugin exceptions are trapped. If your hook raises, the CLI converts it to a structured error and continues — a broken hook can never crash the CLI. But appending to errors is the cleaner contract.

Step 0 — see the result first

For a contract that fails the hook (output from CLI 0.18.1, local-platform contract with an overlays/prod.yaml):

unset FLUID_PROD_DEPLOY_KEY
fluid apply contract.fluid.yaml --env prod --yes

apply hook: prod-key-guard: FLUID_PROD_DEPLOY_KEY is not set in the environment.
  This is required for prod deploys. Either:
    • Set the env var (export FLUID_PROD_DEPLOY_KEY=...), OR
    • Pass --force-pattern-drift if you have a specific reason to bypass the check.
apply aborted by an apply-time plugin hook. Pass --force-pattern-drift to override.

Exit code 1. With the env var set:

export FLUID_PROD_DEPLOY_KEY="$(read-from-secret-manager)"
fluid apply contract.fluid.yaml --env prod --yes
# (proceeds normally)

Override flag for emergencies (logged as a warning, so the override shows in the deploy log):

fluid apply contract.fluid.yaml --env prod --force-pattern-drift --yes

apply hook drift ignored (--force-pattern-drift): prod-key-guard: FLUID_PROD_DEPLOY_KEY is not set in the environment.
  This is required for prod deploys. Either:
  ...

The apply then proceeds.

Step 1 — set up the package

mkdir prod-key-guard && cd prod-key-guard
mkdir -p src/prod_key_guard tests
touch src/prod_key_guard/__init__.py tests/__init__.py

Step 2 — write pyproject.toml

[build-system]
requires = ["setuptools>=68.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "prod-key-guard"
version = "0.1.0"
description = "Refuse prod deploys without FLUID_PROD_DEPLOY_KEY set"
requires-python = ">=3.10"
dependencies = []   # stdlib only — apply hooks are typically dep-free

[project.optional-dependencies]
dev = ["pytest>=7.0"]

# Apply hooks live in this entry-point group. The value is module:function,
# NOT module:Class — apply hooks are functions, not classes.
[project.entry-points."fluid_build.apply_hooks"]
prod-key-guard = "prod_key_guard.hook:check_prod_deploy_key"

[tool.setuptools.packages.find]
where = ["src"]

[tool.pytest.ini_options]
testpaths = ["tests"]

Step 3 — write the hook

# src/prod_key_guard/hook.py
"""Apply-time guard: refuse prod deploys without FLUID_PROD_DEPLOY_KEY set.

Registered as an apply hook via the ``fluid_build.apply_hooks`` entry-point
group. Invoked by the CLI early in ``fluid apply``, after the contract is
loaded but before any provider executes.

Hook contract (see entry-points reference):
    hook(contract_dir: Path, contract: Dict, errors: List[str],
         env: Optional[str] = None) -> None

Append messages to ``errors`` to fail the apply; leave it empty to pass.
Plugin exceptions are trapped, redacted, and added to errors automatically
— a buggy hook cannot crash the CLI.
"""

from __future__ import annotations

import os
from pathlib import Path
from typing import Any, Dict, List, Optional


REQUIRED_ENV_VAR = "FLUID_PROD_DEPLOY_KEY"
# Fallback for applies that don't pass `--env`: your deploy runner / CI job
# sets DEPLOY_ENV before invoking `fluid apply`. The CLI itself does not set
# it — see the "Apply hooks can receive `--env`" note further down.
DEPLOY_ENV_VAR = "DEPLOY_ENV"


def check_prod_deploy_key(
    contract_dir: Path,
    contract: Dict[str, Any],
    errors: List[str],
    env: Optional[str] = None,
) -> None:
    """Fail prod applies when FLUID_PROD_DEPLOY_KEY isn't set."""
    # `env` is the resolved `--env` value the CLI was invoked with, or None
    # when the flag was omitted. DEPLOY_ENV is the runner-set fallback.
    target = env or os.environ.get(DEPLOY_ENV_VAR, "")
    if target != "prod":
        return  # not a prod apply — nothing to check

    if not os.environ.get(REQUIRED_ENV_VAR):
        errors.append(
            f"prod-key-guard: {REQUIRED_ENV_VAR} is not set in the environment.\n"
            f"  This is required for prod deploys. Either:\n"
            f"    • Set the env var (export {REQUIRED_ENV_VAR}=...), OR\n"
            f"    • Pass --force-pattern-drift if you have a specific reason to bypass the check."
        )

Three things worth calling out:

• env is the CLI's own signal; DEPLOY_ENV is the fallback. fluid apply --env prod reaches the hook as env="prod". DEPLOY_ENV is a convention env var your CI runner exports, and it covers applies invoked without --env. Checking env first means a CI job that forgets to export DEPLOY_ENV still gets the guard. A guard keyed only on a runner-set variable passes silently when the export is forgotten.
• The error message is specific. It tells the user what is wrong, what to do, and what the escape hatch is. Copy that shape.
• Append, don't raise. Raising works (the CLI catches it and reports apply hook '<name>' raised: ...), but appending produces cleaner output.

Step 4 — test both paths

# tests/test_hook.py
"""Tests for prod-key-guard apply hook."""

import os
from pathlib import Path

import pytest

from prod_key_guard.hook import (
    check_prod_deploy_key,
    DEPLOY_ENV_VAR,
    REQUIRED_ENV_VAR,
)


@pytest.fixture(autouse=True)
def clear_env(monkeypatch):
    """Every test starts with a clean env — no DEPLOY_ENV, no deploy key."""
    monkeypatch.delenv(DEPLOY_ENV_VAR, raising=False)
    monkeypatch.delenv(REQUIRED_ENV_VAR, raising=False)


def test_non_prod_target_passes_without_key(monkeypatch):
    monkeypatch.setenv(DEPLOY_ENV_VAR, "dev")
    errors: list[str] = []
    check_prod_deploy_key(Path("/tmp"), {}, errors)
    assert errors == []


def test_no_deploy_env_passes(monkeypatch):
    """No `--env`, and no DEPLOY_ENV: the hook can't tell what's being
    deployed — pass through. The convention here is opt-in: pass `--env` or
    set DEPLOY_ENV when you want the guard to run."""
    errors: list[str] = []
    check_prod_deploy_key(Path("/tmp"), {}, errors)
    assert errors == []


def test_prod_target_missing_key_fails(monkeypatch):
    monkeypatch.setenv(DEPLOY_ENV_VAR, "prod")
    errors: list[str] = []
    check_prod_deploy_key(Path("/tmp"), {}, errors)
    assert len(errors) == 1
    assert REQUIRED_ENV_VAR in errors[0]
    assert "--force-pattern-drift" in errors[0]  # message points to the escape hatch


def test_prod_target_with_key_passes(monkeypatch):
    monkeypatch.setenv(DEPLOY_ENV_VAR, "prod")
    monkeypatch.setenv(REQUIRED_ENV_VAR, "secret-key-value")
    errors: list[str] = []
    check_prod_deploy_key(Path("/tmp"), {}, errors)
    assert errors == []


def test_env_argument_fires_without_deploy_env():
    """`fluid apply --env prod` reaches the hook directly — no DEPLOY_ENV
    needed, so a CI job that forgets to export it can't silently skip the
    guard."""
    errors: list[str] = []
    check_prod_deploy_key(Path("/tmp"), {}, errors, env="prod")
    assert len(errors) == 1
    assert REQUIRED_ENV_VAR in errors[0]

Run it:

pip install -e ".[dev]"
pytest

.....                                                                    [100%]
5 passed in 0.02s

Step 5 — wire it in

# On the deployer's machine (CI runner, on-call laptop, etc):
pip install data-product-forge prod-key-guard

# The hook is auto-discovered. No contract change needed. Confirm it:
fluid plugins list --json

The JSON has an apply_hook key; the plugin shows up there:

{
  "allowed": true,
  "dispatched": true,
  "distribution": "prod-key-guard 0.1.0",
  "group": "fluid_build.apply_hooks",
  "name": "prod-key-guard"
}

Or use the importlib.metadata one-liner:

python -c "
from importlib.metadata import entry_points
for ep in entry_points(group='fluid_build.apply_hooks'):
    print(f'{ep.name}: {ep.value}')
"
# prod-key-guard: prod_key_guard.hook:check_prod_deploy_key

End-to-end (CLI 0.18.1, local-platform contract with an overlays/prod.yaml):

# This fails: no FLUID_PROD_DEPLOY_KEY in the environment.
unset FLUID_PROD_DEPLOY_KEY DEPLOY_ENV
fluid apply contract.fluid.yaml --env prod --yes
# apply hook: prod-key-guard: FLUID_PROD_DEPLOY_KEY is not set in the environment.
#   ...
# apply aborted by an apply-time plugin hook. Pass --force-pattern-drift to override.

# The check passes with the key set:
export FLUID_PROD_DEPLOY_KEY=<key>
fluid apply contract.fluid.yaml --env prod --yes
# (proceeds normally)

# Non-prod targets are untouched:
fluid apply contract.fluid.yaml --env dev --yes
# (proceeds normally)

# Break-glass override (logged as a warning):
unset FLUID_PROD_DEPLOY_KEY
fluid apply contract.fluid.yaml --env prod --force-pattern-drift --yes
# apply hook drift ignored (--force-pattern-drift): prod-key-guard: FLUID_PROD_DEPLOY_KEY is not set in the environment.

Variations — the same shape, different invariants

def check_team_match(contract_dir, contract, errors):
    declared = (contract.get("labels") or {}).get("team")
    deployer = os.environ.get("TEAM_NAME", "")
    if declared and declared != deployer:
        errors.append(
            f"team-match: contract declares team={declared!r} "
            f"but deployer is from team={deployer!r}. "
            f"Cross-team deploys require a written change request."
        )

from datetime import datetime, timezone

def check_business_hours(contract_dir, contract, errors):
    # Reads the runner-set DEPLOY_ENV; add an `env=None` parameter to use --env.
    if os.environ.get("DEPLOY_ENV", "") != "prod":
        return
    now = datetime.now(timezone.utc)
    # Mon–Fri, 8am–5pm UTC
    if now.weekday() >= 5 or not (8 <= now.hour < 17):
        errors.append(
            "business-hours: prod deploys are restricted to weekdays 8-17 UTC. "
            "Pass --force-pattern-drift if this is a genuine incident response."
        )

from pathlib import Path

import yaml  # a dependency of data-product-forge


def check_scaffold_lock(contract_dir, contract, errors):
    lockfile = Path(contract_dir) / "fluid-scaffold.lock"
    if not lockfile.exists():
        return
    locked = yaml.safe_load(lockfile.read_text()) or {}
    libraries = ((contract.get("extensions") or {})
                 .get("customScaffold") or {}).get("libraries", [])
    for lib in libraries:
        wanted = (lib.get("source") or {}).get("ref")
        recorded = ((locked.get("libraries") or {})
                    .get(lib.get("id")) or {}).get("ref")
        if wanted and recorded and wanted != recorded:
            errors.append(
                f"scaffold-lock: library {lib.get('id')!r} asks for ref "
                f"{wanted!r}, but the lock was written at {recorded!r}. "
                "Re-run `fluid custom-scaffold --update`."
            )

import requests

def check_owner_active(contract_dir, contract, errors):
    owner = ((contract.get("metadata") or {}).get("owner") or {}).get("email")
    if not owner:
        return
    # Cheap HEAD against your internal directory service
    r = requests.head(f"https://directory.my-org.example.com/users/{owner}",
                      timeout=5)
    if r.status_code == 404:
        errors.append(
            f"owner-active: contract owner {owner!r} is not active in the "
            f"directory. Update metadata.owner before deploying."
        )

You'll know it worked when

• fluid plugins list --json (or the importlib.metadata one-liner above) shows prod-key-guard under apply_hook.
• All the tests pass under pytest.
• fluid apply --env prod fails with the structured message when FLUID_PROD_DEPLOY_KEY is unset, with no DEPLOY_ENV exported.
• The same command succeeds when the deploy-key env var is set.
• fluid apply --env dev passes regardless of the deploy-key env var (only prod is gated).
• --force-pattern-drift downgrades the error to a logged warning and lets the apply proceed.

When not to use an apply hook

• For contract-shape checks. "Field X must be a regex match" runs at fluid validate, before anyone even thinks about applying. Use a Validator instead.
• For checks that the contract author should know about. Apply hooks fire on the deployer's machine. If the contract author won't know about the failure until CI runs, the feedback loop is too slow — push the check earlier with a Validator.
• For long-running checks. Apply hooks have no per-hook timeout. If your hook can hang for 60 seconds on a flaky network call, apply hangs too. Either short-circuit (requests.head(..., timeout=5)) or move the check to a background service.

Common gotchas

deploy_env = env or os.environ.get("DEPLOY_ENV")
if deploy_env is None:
    errors.append("prod-key-guard: DEPLOY_ENV must be set to dev/staging/prod")
    return
if deploy_env != "prod":
    return

if errors and os.environ.get("MY_HOOK_OVERRIDE"):
    return  # hook self-overrides via env var

Next

• Custom validator — for checks that can run at fluid validate instead
• Your own CI — bundle pattern for scaffolds, often paired with apply hooks for drift detection
• Reference → Entry points — signature reference for the plugin groups
• Reference → Trust model — what the CLI guarantees about hook execution (deep-copied contract, exception trapping, redaction)
• Apply-hook example — same hook in example form, with more variations