ersatztv/scripts/tests/test_ci_dropped_step_guard.py

"""The dropped-step guard on docker-build.yml's two REQUIRED jobs (ersatztv#756).

WHAT THIS IS PROTECTING. A `run:` body the runner declines to interpolate is DROPPED, and the job
still concludes `success` (ersatztv#751, `ci.workflow-run-body-no-expressions`). #751 fixed that in
`review-verdict.yml`, where the consequence is fail-CLOSED — `review-verdict/h10` is absent and the
merge is blocked. It left the two places where the same drop is fail-OPEN: `Build & test (.NET)` and
`EF migration integrity (SQLite + MySql)` are the other two required contexts on `main`, so a dropped
step there sends a required check green having done no work.

THE TESTS COME IN THREE KINDS AND NONE SUBSTITUTES FOR ANOTHER, which is the lesson #751 paid for:

  * STATIC — the marker set and the guard's expectations agree, and the guard is positioned so it
    can actually run. Cheap, and the only kind that catches a NEW step added without a marker.
  * BEHAVIOURAL — the guard's real command line is EXECUTED against markers written by the steps'
    real marker lines, both extracted from the parsed workflow. A structural test cannot prove an
    exit code, and `exit 1` in a body is satisfiable by dead code.
  * A LIVE PROBE — that the runner still executes a LATER step after dropping an earlier one, on the
    BUILD lane rather than the `small` lane #751 measured. That is the premise the whole guard rests
    on and no test here can establish it; it is recorded in docs/ci-cd.md and on the issue.
"""

from __future__ import annotations

import os
import re
import subprocess
from pathlib import Path

import pytest
import yaml

REPO_ROOT = Path(__file__).resolve().parents[2]
WORKFLOW = REPO_ROOT / ".gitea" / "workflows" / "docker-build.yml"
SCRIPT = REPO_ROOT / "scripts" / "ci-step-ran.sh"

# The jobs whose contexts branch protection REQUIRES on `main`. Read live on 2026-08-10:
#   Build ErsatzTV Image / Build & test (.NET) (pull_request)
#   Build ErsatzTV Image / EF migration integrity (SQLite + MySql) (pull_request)
#   review-verdict/h10
# The third is guarded by test_pr_changed_files.py; these two are this file's subject. `build`,
# `api-docs` and `format` are deliberately NOT here — they are not required, and all three
# legitimately interpolate into a `run:` body, so extending the absolute rule to them would be false.
# Per-step markers apply to the two REQUIRED contexts, where a dropped step is fail-OPEN.
MARKED_JOBS = ("test", "migrations")

# The delimiter ban is WIDER than the marker set, and the extra job is not an afterthought.
# `build`'s "Smoke + IPTV E2E" step runs AFTER `Build and push`, so on a `v*` tag the image is
# already in the registry as the release candidate and this step is what decides whether it was ever
# booted. A drop there publishes an unsmoked candidate and goes green, and `DeployStack jazz-media`
# promotes exactly that image — not a "smaller cost than a required context", which is what an
# earlier draft of the decision record claimed. Its two payloads moved into the step's `env:`, which
# is the free half of the escape hatch, so the ban costs nothing there.
#
# `functional-e2e` is deliberately NOT here even though it is delimiter-free today: it is advisory by
# declaration (not a required check, not a `needs:` of `build`), so the rule stays "ban where a drop
# is consequential" rather than "ban wherever it happens to be free right now".
# `api-docs` and `format` keep one delimiter each, both `github.base_ref` in a detect step, and gate
# nothing that ships.
DELIMITER_BAN_JOBS = ("test", "migrations", "build")

# THE RAW OPENER, not a closed `${{ … }}` pair — found by cold review. The runner's rewrite is
# triggered by the OPENER; a closed-pair regex therefore misses `# ${{` with no closer, which would
# sail through an "absolute" ban and still drop the step. Nothing in these jobs may contain the
# opener at all, so matching it directly is both simpler and strictly stronger. `_EXPR` is kept for
# reporting the payload of a well-formed one in the failure message.
_OPENER = re.compile(r"\$\{\{")
_EXPR = re.compile(r"\$\{\{(.*?)\}\}", re.S)
_MARK = re.compile(r'ci-step-ran\.sh"?\s+mark\s+(\S+)')


# ONE parse, shared. `yaml.safe_load` per call returns a fresh object graph, so an identity test
# across two helpers (`steps[-1] is guard`) would compare structurally-equal but distinct dicts and
# fail — or, worse in the other direction, an `is not` filter would exclude nothing and a step would
# match as its own guard. That is not hypothetical: test_pr_changed_files.py records exactly this
# going wrong in the #751 guard test, where the assertions then ran against the wrong step.
_DOC = yaml.safe_load(WORKFLOW.read_text())


def _doc():
    return _DOC


def _steps(job: str):
    return _doc()["jobs"][job]["steps"]


def _run_steps(job: str):
    return [s for s in _steps(job) if s.get("run")]


def _guard(job: str):
    """The trailing assert step. Located by CONTENT, never by index.

    Locating it as `steps[-1]` here and then asserting it is last elsewhere would be circular — the
    position test would hold by construction. This finds the step that invokes the assert
    sub-command, and `test_the_guard_is_the_LAST_step` independently checks where it sits.
    """
    hits = [s for s in _run_steps(job) if "ci-step-ran.sh assert" in s["run"]]
    assert len(hits) == 1, f"job '{job}' has {len(hits)} assert steps, expected exactly 1"
    return hits[0]


def _marked(job: str):
    """[(step, key)] for every step that records its own execution, in declaration order."""
    out = []
    for s in _run_steps(job):
        m = _MARK.search(s["run"])
        if m:
            out.append((s, m.group(1)))
    return out


def _guard_buckets(job: str):
    """(always_keys, gated_keys) as the guard's own argv spells them."""
    argv = _guard(job)["run"].split()
    assert "--always" in argv and "--gated" in argv, argv
    a, g = argv.index("--always"), argv.index("--gated")
    return argv[a + 1:g], argv[g + 1:]


# Mirrors the `if:` every gated step in these jobs carries. Compared as a normalised string rather
# than by parsing the expression: what matters is that a step's gating and the guard's bucketing are
# the SAME condition, and any rewrite of one that is not mirrored in the other should be loud.
SKIP_GATE = ("steps.detect.outputs.docs_only!='true'&&steps.revalidate.outputs.skip!='true'")


def _is_gated(step) -> bool:
    return re.sub(r"\s+", "", str(step.get("if", ""))) == SKIP_GATE


# ------------------------------------------------------------------------------------------------
# STATIC
# ------------------------------------------------------------------------------------------------


@pytest.mark.parametrize("job", DELIMITER_BAN_JOBS)
def test_the_delimiter_banned_jobs_have_NO_expression_delimiter_in_any_run_body(job):
    """The absolute rule from `review-verdict.yml`, extended to the two required build jobs.

    This is the cheaper and more general half of #756: the drop mechanism REQUIRES an opener in the
    scalar, so a job with none is immune by construction and the runtime markers are a backstop
    rather than the only line of defence.

    The scope is the three jobs in `DELIMITER_BAN_JOBS` — see the comment there for why `build` is in
    and `functional-e2e` is not. Do NOT restate this docstring as "scoped to the required pair":
    round 2 moved `build`'s two payloads into `env:` and brought it into the ban, and this docstring
    sits directly above the decorator that parametrises over the wider set.

    The escape hatch when a value really is needed is the step's `env:` block, which is interpolated
    PER VALUE, so a payload that does not evaluate cannot take the body with it.

    The `run:` SCALAR AS PARSED, comments and all. A shell comment inside a `run:` body is NOT inert
    — that is the whole #751 defect — so this must never filter comments out. Ordinary YAML comments
    outside a `run:` body ARE inert and are not read here.
    """
    offenders = []
    for s in _run_steps(job):
        for m in _OPENER.finditer(s["run"]):
            closed = _EXPR.match(s["run"], m.start())
            payload = closed.group(1).strip() if closed else "<unclosed opener>"
            offenders.append(f"{s.get('name', '?')}: {payload!r}")
    assert not offenders, (
        f"job '{job}' of docker-build.yml has an expression delimiter inside a run: body — "
        f"{offenders}. A dropped step in this job is CONSEQUENTIAL — `test`/`migrations` write "
        "REQUIRED status contexts, and `build` publishes the release candidate before its smoke step "
        "runs. Even in a comment a delimiter is unsafe: the runner rewrites the WHOLE body into a "
        "format(...) call, and if the payload does not parse it DROPS THE STEP and reports the job "
        "green — so the check passes having done no work (ersatztv#751/#756). Pass the value in "
        "through the step's `env:` "
        "block instead; to describe an expression in prose, name it rather than quoting the "
        "delimiters."
    )
    # ANTI-VACUITY. A walk that reached no bodies, or only the trivial ones, would make the
    # assertion above green while proving nothing. Counted against the job's own step list read
    # here, so a helper that silently stopped yielding steps is caught rather than rewarded.
    declared = sum(1 for s in _steps(job) if isinstance(s, dict) and s.get("run"))
    assert len(_run_steps(job)) == declared >= 3, (
        f"the walk reached {len(_run_steps(job))} run: bodies but job '{job}' declares {declared}"
    )


@pytest.mark.parametrize("job", MARKED_JOBS)
def test_every_consequential_run_step_marks_itself_as_its_FIRST_act(job):
    """The completeness half — and the only test that catches a NEWLY ADDED step with no marker.

    A guard that checks a fixed list can go quietly incomplete: someone adds a `Test SPA (part 2)`
    step, it is never marked, the guard never expects it, and a drop of exactly that step is
    invisible again. So the expectation is DERIVED from the workflow rather than written down twice.

    EXEMPT: steps carrying `continue-on-error: true`. Those are advisory by construction (the
    peak-anon sampler, the coverage summary) — the workflow already declares that their failure must
    not redden the job, so their non-execution cannot be a fail-open either. Making them mandatory
    would be asserting the opposite of what `continue-on-error` means.

    FIRST ACT, not merely present. A marker written at the END of a body records completion, not
    execution — and this repo has legitimate early-exit paths. More importantly a marker further down
    can be skipped by an early `exit 0` while the step did nothing, which is the fail-open again one
    line lower. `set -euo pipefail` is allowed to precede it: it cannot fail, and it is what makes
    the rest of the body honest.
    """
    missing, late = [], []
    for s in _run_steps(job):
        if s.get("continue-on-error") is True or "ci-step-ran.sh assert" in s["run"]:
            continue
        m = _MARK.search(s["run"])
        if not m:
            missing.append(s.get("name", "?"))
            continue
        # By LINE, not by byte offset. The marker sits mid-line (the command is quoted and
        # prefixed with $GITHUB_WORKSPACE), so slicing at `m.start()` counts the marker's OWN line
        # prefix as a preceding command and reddens every correctly-written step.
        lines = s["run"].splitlines()
        at = next(i for i, ln in enumerate(lines) if _MARK.search(ln))
        preceding = [
            ln.strip() for ln in lines[:at]
            if ln.strip() and not ln.strip().startswith("#")
        ]
        if [ln for ln in preceding if not ln.startswith("set -")]:
            late.append((s.get("name", "?"), preceding))
    assert not missing, (
        f"these run: steps of the REQUIRED job '{job}' do not record that they executed: {missing}. "
        "A step the runner drops concludes success, so without a marker its non-execution takes the "
        "whole required context green having done no work (ersatztv#756). Add "
        '`\"${GITHUB_WORKSPACE:-.}/scripts/ci-step-ran.sh\" mark <key>` as the step\'s first line and '
        "the key to the guard step's --always/--gated list."
    )
    assert not late, (
        f"these steps of '{job}' mark themselves only after other commands have run: {late}. The "
        "marker must be the first act, or a body that exits early records nothing while the guard "
        "still expects it — or worse, records success for work that did not happen."
    )


@pytest.mark.parametrize("job", MARKED_JOBS)
def test_the_guard_expects_EXACTLY_the_set_of_marked_keys_in_the_right_bucket(job):
    """Set equality in BOTH directions, plus the bucket, because each failure is silent differently.

    A key marked but not expected → the guard never notices that step being dropped: a fail-open
    that looks fully guarded. A key expected but not marked → the guard reddens on every single run,
    which is fail-closed but reads as "this guard is broken" and is how a correct guard gets deleted.

    The BUCKET has to match the step's own `if:`. A gated step listed under `--always` reddens every
    docs-only and already-validated run — the two paths whose entire purpose is to report green in
    seconds. An always-run step listed under `--gated` stops being checked the moment either skip
    gate fires, which is a fail-open on precisely the runs where least else is happening.
    """
    marked = _marked(job)
    keys = [k for _, k in marked]
    assert len(keys) == len(set(keys)), (
        f"job '{job}' reuses a marker key: {[k for k in keys if keys.count(k) > 1]}. Two steps "
        "sharing a key means either one satisfies the guard for both, so dropping one is invisible."
    )
    always, gated = _guard_buckets(job)
    assert sorted(always + gated) == sorted(keys), (
        f"job '{job}': the guard expects {sorted(always + gated)} but the steps mark "
        f"{sorted(keys)}. Keys marked-but-unexpected are unguarded drops; keys "
        "expected-but-unmarked redden every run."
    )
    # AN UNRECOGNISED `if:` IS REJECTED, never silently bucketed — found by both reviewers. The
    # protocol only knows two conditions: absent (always runs) and exactly the skip gate. A marked
    # step carrying a third condition (`if: github.event_name == 'push'`, or the `always() && <gate>`
    # spelling the peak-anon steps already use) would fall through to "always", the suite would go
    # green, and the guard would then demand a step the runner legitimately skipped — reddening a
    # REQUIRED context and deadlocking `main`. There is already a near-miss in this file: `Report
    # peak container memory` carries that third spelling and escapes only because it is
    # `continue-on-error: true` and therefore exempt from marking.
    for step, key in marked:
        cond = re.sub(r"\s+", "", str(step.get("if", "")))
        assert cond in ("", SKIP_GATE), (
            f"job '{job}': marked step {step.get('name')!r} has an `if:` the guard protocol does not "
            f"model ({step.get('if')!r}). Only 'absent' and the exact skip gate are understood; "
            "anything else would be bucketed as --always and would fail the job on a run where the "
            "step is legitimately skipped. Extend the protocol deliberately, or leave the step "
            "unmarked."
        )
        want = "gated" if _is_gated(step) else "always"
        got = "gated" if key in gated else "always"
        assert want == got, (
            f"job '{job}': step {step.get('name')!r} is {want} (if: {step.get('if')!r}) but the "
            f"guard lists its key {key!r} under --{got}."
        )


@pytest.mark.parametrize("job", MARKED_JOBS)
def test_the_guard_is_the_LAST_step_carries_no_if_and_is_not_advisory(job):
    """Position and condition, which together are what make the guard reachable and quiet.

    LAST, because a guard placed before a marked step reads a marker not yet written and fails on
    every run.

    NO `if:` — a deliberate departure from the #751 guard's `if: always()`, and the thing most likely
    to be "corrected" back. That job has one real step, so `always()` costs nothing. These jobs have
    a dozen, and a genuine failure in an early one SKIPS every later step: an `always()` guard would
    then report "these steps never executed: typecheck web-test build dotnet-test" on top of every
    ordinary red build. That is the runner obeying its own gating, not a dropped step, and a guard
    that cries wolf on every red build gets deleted.

    The default `if:` is `success()`, and the invariant that makes relying on it safe rather than
    lucky: this step is skipped only when an earlier step FAILED, and that failure already fails the
    job. So `guard skipped => job red`, and every path to a green job runs the guard. A dropped step
    is invisible precisely because it concludes `success` — which keeps the job green and therefore
    reaches here.

    NOT `continue-on-error`, which would let it observe the failure and go green anyway — the whole
    defect, one attribute over.
    """
    steps = _steps(job)
    guard = _guard(job)
    assert steps[-1] is guard, (
        f"the dropped-step guard is not the last step of '{job}' — it is at index "
        f"{steps.index(guard)} of {len(steps)}, so any marked step after it would be unguarded and "
        "the guard would read a marker that has not been written yet."
    )
    assert "if" not in guard, (
        f"the '{job}' guard carries `if: {guard.get('if')!r}`. It must have none: the default "
        "`success()` is what keeps it silent on ordinary red builds, and `always()` would make it "
        "announce a false 'these steps never executed' on every failing run. See the comment above "
        "the step for why this is a deliberate departure from the #751 guard."
    )
    assert guard.get("continue-on-error") is not True, (
        f"the '{job}' guard is continue-on-error, so it detects the dropped step and lets the job go "
        "green regardless — which is the defect it exists to remove."
    )


@pytest.mark.parametrize("job", MARKED_JOBS)
def test_the_guards_OWN_body_cannot_be_dropped_by_the_mechanism_it_guards_against(job):
    """A guard the guarded mechanism can silently delete is worse than no guard.

    Its absence is silent too: the job simply goes green with nothing checked, which is
    indistinguishable from a clean run. #751 states the rule; here it is stronger than there,
    because the body is a single command with no delimiter possible rather than 20 lines of prose
    that must be kept clean by hand.

    The gate VALUES arrive through `env:`, which the runner interpolates per value — a bad payload
    there fails that value, not the body. Both are additionally held to naming a real context by
    test_every_workflow_expression_names_a_REAL_context_or_function in test_pr_changed_files.py.
    """
    guard = _guard(job)
    assert not _OPENER.search(guard["run"]), (
        f"the '{job}' guard's own run body contains an expression delimiter, so the runner can drop "
        "the guard the same way it drops the steps the guard is watching — and that absence is "
        "silent as well."
    )
    assert guard["run"].strip().startswith("scripts/ci-step-ran.sh assert"), (
        f"the '{job}' guard is no longer a bare invocation: {guard['run']!r}. Keeping it to one "
        "command is what makes a delimiter impossible rather than merely absent."
    )
    # THE VALUES, not just the names — found by cold review. Asserting the keys alone accepts
    # `ETV_DOCS_ONLY: ${{ steps.detect.outputs.doc_only }}` (note the typo), which names a real
    # context so the repo-wide expression check passes it too. The guard would then read an EMPTY
    # value on a docs-only run, demand the gated steps that were correctly skipped, and redden a
    # REQUIRED context on every docs-only PR.
    # THE TWO MAPPINGS MUST BE PRESENT AND CORRECT — but this deliberately does NOT demand that the
    # `env:` block contain ONLY them. An earlier version compared the whole dict, which false-redded
    # on adding an unrelated variable (an `LC_ALL`, say) and on the equally-valid `${{x}}` spacing;
    # a red here blocks every merge through the combined status, so brittleness is a real cost and
    # not a free strictness win. Whitespace inside the delimiters is normalised for the same reason.
    env = {k: re.sub(r"\s+", "", str(v)) for k, v in (guard.get("env") or {}).items()}
    for name, want in (("ETV_DOCS_ONLY", "${{steps.detect.outputs.docs_only}}"),
                       ("ETV_REVALIDATE_SKIP", "${{steps.revalidate.outputs.skip}}")):
        assert env.get(name) == want, (
            f"the '{job}' guard's env: has {name}={guard.get('env', {}).get(name)!r}, expected the "
            f"output the gated steps' own `if:` reads ({want}). A typo here is SILENT rather than "
            "loud: it still names a real context, so the repo-wide expression check passes it, the "
            "value arrives empty, and the guard then demands steps that were legitimately skipped — "
            "reddening a REQUIRED context on every docs-only run."
        )


# ------------------------------------------------------------------------------------------------
# BEHAVIOURAL — the guard's real command line, against markers written by the steps' real lines
# ------------------------------------------------------------------------------------------------


def _mark_line(step) -> str:
    """The step's OWN marker line, verbatim from the workflow.

    Extracted rather than rebuilt in Python ON PURPOSE. A test that composed the command itself
    would keep passing after the workflow and the script drifted apart on the path, the quoting or
    the sub-command — and that divergence is exactly the failure that makes the guard fail on every
    run and then get deleted as broken. Running the real line proves the two agree by construction.
    """
    line = next(ln for ln in step["run"].splitlines() if _MARK.search(ln))
    return line.strip()


# THE GATE VALUES DEFAULT TO `"false"`, WHICH IS WHAT THE RUNNER ACTUALLY SENDS — and getting this
# wrong made the whole suite blind. Found by cold review, which demonstrated it: every behavioural
# test used to leave these UNSET, so the guard was never once driven at its production values. Change
# the gate in `ci-step-ran.sh` from `= "true"` to `-n` — a one-token regression — and all 30 tests
# stayed GREEN while the guard, run with the real environment, reported
# `Skip gate fired (docs_only='false') … All 2 expected step(s) executed` and exited 0. `Build`,
# `Test` and both migration replays would have been unguarded on every ordinary run, with the guard
# announcing that it had proved everything.
#
# THE COMPLETE VALUE SET, and where each comes from — worth spelling out, because the obvious reading
# of the evidence is wrong. Both producers document `true|false` and write exactly that
# (`scripts/ci-detect-docs-only.sh` -> `docs_only=`, `scripts/ci-detect-already-validated.sh` ->
# `skip=`), so an ordinary run sends `false` and a skipping run sends `true`.
#
# The live log of the probe this change cites (run 1910, job 8064) shows `ETV_DOCS_ONLY: false` and
# `ETV_REVALIDATE_SKIP:` EMPTY — but do NOT read that as revalidate's normal output. `revalidate` was
# the step the probe deliberately dropped, so it wrote no output at all. The empty string is
# therefore not an odd third state: it is the SIGNATURE OF THE VERY FAILURE THIS GUARD EXISTS TO
# CATCH, which is exactly why the gate must treat anything that is not `true` as "widen what is
# required". `None` (unset) is the same case reached a different way.
#
# A test double is an assertion about what the real system sends, and the earlier version of this one
# was wrong about the only field the guard branches on.
GATE_VALUES_IN_THE_WILD = ("false", "", None)


def _env(tmp_path, **extra):
    env = {
        "PATH": os.environ["PATH"],
        "GITHUB_WORKSPACE": str(REPO_ROOT),
        "RUNNER_TEMP": str(tmp_path),
        "GITHUB_JOB": "test",
        "GITHUB_RUN_ID": "424242",
        "GITHUB_RUN_ATTEMPT": "7",
        "ETV_DOCS_ONLY": "false",
        "ETV_REVALIDATE_SKIP": "false",
    }
    env.update(extra)
    return {k: v for k, v in env.items() if v is not None}


def _run(script: str, env):
    return subprocess.run(["bash", "-c", script], cwd=REPO_ROOT, env=env,
                          capture_output=True, text=True)


@pytest.mark.parametrize("gate", GATE_VALUES_IN_THE_WILD,
                         ids=["gate-false", "gate-empty", "gate-unset"])
@pytest.mark.parametrize("job", MARKED_JOBS)
def test_the_guard_PASSES_when_every_step_marked_itself(job, gate, tmp_path):
    """The positive control. Without it, a guard that always failed would satisfy every case below.

    `GITHUB_JOB` is set to the job under test, so this also covers the marker file being keyed per
    job: if it were not, the two jobs would share a file and one job's markers would answer for the
    other's dropped steps.
    """
    marks = [_mark_line(s) for s, _ in _marked(job)]
    guard = _guard(job)["run"]
    # Parametrised over every NOT-SKIPPING spelling the runner emits — `false` on an ordinary run,
    # empty when the producing step was dropped, absent if the output is never set. All three must
    # require the gated steps; a gate that treats any of them as a skip is fail-open on that path.
    env = _env(tmp_path, GITHUB_JOB=job, ETV_DOCS_ONLY=gate, ETV_REVALIDATE_SKIP=gate)
    r = _run("\n".join(["set -e", *marks, guard]), env)
    assert r.returncode == 0, (
        f"the '{job}' guard rejected a run in which every step marked itself — the steps and the "
        f"guard disagree, so this would fail on every run.\n{r.stdout}\n{r.stderr}"
    )
    assert "All" in r.stdout and "executed" in r.stdout, r.stdout
    # The other half of the identity contract: with GITHUB_RUN_ATTEMPT set (`_env` sends 7) the line
    # must report the REAL value and say so. A mis-derivation (`${marker#*-}` rather than `##`) or an
    # inverted provenance test would otherwise ship silently, and the operator reading this line to
    # settle the promotion question would read it wrong.
    assert f"Marker identity: job={job} run=424242 attempt=7 (from the runner)" in r.stdout, (
        f"the guard misreported its marker identity: {r.stdout!r}")


@pytest.mark.parametrize("job", MARKED_JOBS)
def test_dropping_ANY_single_step_FAILS_the_guard(job, tmp_path):
    """Every marked step, one at a time — not a sample.

    An arbitrary sample gives false negatives here: the interesting drop is `Test` or the migration
    replay, and a test that only omitted the first step would prove the guard catches the one case
    that was never fail-open anyway. Dropping each key in turn is the only version that establishes
    the property the issue asks for.
    """
    marked = _marked(job)
    guard = _guard(job)["run"]
    for dropped_step, dropped_key in marked:
        d = tmp_path / dropped_key
        d.mkdir()
        marks = [_mark_line(s) for s, k in marked if k != dropped_key]
        r = _run("\n".join(["set -e", *marks, guard]), _env(d, GITHUB_JOB=job))
        assert r.returncode != 0, (
            f"job '{job}': the guard went GREEN with {dropped_step.get('name')!r} "
            f"(key {dropped_key!r}) never having executed. That is a REQUIRED context reporting "
            f"success having skipped that work — the exact fail-open of ersatztv#756.\n{r.stdout}"
        )
        assert dropped_key in (r.stdout + r.stderr), (
            f"the guard failed but did not name the missing step {dropped_key!r}: {r.stdout}"
        )


@pytest.mark.parametrize("job", MARKED_JOBS)
@pytest.mark.parametrize("gate", ["ETV_DOCS_ONLY", "ETV_REVALIDATE_SKIP"])
def test_a_fired_skip_gate_does_not_require_the_gated_steps(gate, job, tmp_path):
    """The docs-only and already-validated paths must still report green in seconds.

    They are the reason these jobs are never `if:`-skipped at the JOB level (a skipped required
    context is a state this repo deliberately does not rely on — ersatztv#416/#418), so a guard that
    reddened them would make every docs-only PR unmergeable. Which is #751's user-visible symptom
    arriving from the opposite direction, and worth a test rather than a comment.
    """
    marks = [_mark_line(s) for s, k in _marked(job) if k in _guard_buckets(job)[0]]
    guard = _guard(job)["run"]
    r = _run("\n".join(["set -e", *marks, guard]), _env(tmp_path, GITHUB_JOB=job, **{gate: "true"}))
    assert r.returncode == 0, (
        f"with {gate}=true the guard still demanded the gated steps, so every docs-only / "
        f"already-validated run of a REQUIRED job would be red.\n{r.stdout}\n{r.stderr}"
    )
    assert "Skip gate fired" in r.stdout, r.stdout


@pytest.mark.parametrize("job", MARKED_JOBS)
def test_a_fired_skip_gate_STILL_requires_the_ALWAYS_steps(job, tmp_path):
    """The negative control for the test above — otherwise `ETV_DOCS_ONLY=true` would be a blanket
    off-switch and the previous test would be passing for the wrong reason.

    This is the case that matters most on a docs-only run: the detect steps are the only things that
    execute, so if their drop were unguarded the skip path would be entirely unchecked.
    """
    guard = _guard(job)["run"]
    r = _run("\n".join(["set -e", guard]), _env(tmp_path, GITHUB_JOB=job, ETV_DOCS_ONLY="true"))
    assert r.returncode != 0, (
        "with ETV_DOCS_ONLY=true and NO steps marked at all, the guard passed — the skip gate is "
        "acting as a blanket off-switch rather than as a narrowing of what is expected."
    )
    assert "detect" in (r.stdout + r.stderr), r.stdout


@pytest.mark.parametrize("job", MARKED_JOBS)
def test_an_EMPTY_gate_value_requires_the_gated_steps(job, tmp_path):
    """A dropped `detect` step leaves its outputs EMPTY, not 'false'.

    Reading empty as "skipped" would mean the one drop that disables the detect step also disables
    the guard for everything downstream — the guard switching itself off in response to the very
    failure it exists to catch. The direction has to be: anything that is not exactly `true` widens
    what is required.
    """
    guard = _guard(job)["run"]
    r = _run("\n".join(["set -e", guard]),
             _env(tmp_path, GITHUB_JOB=job, ETV_DOCS_ONLY="", ETV_REVALIDATE_SKIP=""))
    assert r.returncode != 0
    assert _guard_buckets(job)[1][-1] in (r.stdout + r.stderr), (
        f"empty gate values were read as a skip, so the gated steps went unchecked: {r.stdout}"
    )


def test_a_STALE_marker_from_another_run_cannot_satisfy_the_guard(tmp_path):
    """A marker from another run, attempt or job must never answer for this one.

    Do NOT restate this as "RUNNER_TEMP is /tmp, not a private per-job directory". That is a #751
    measurement taken on a job with no `container:`, and it does not transfer: these two jobs run
    inside the CI toolchain image, so their `/tmp` is the container's own. The fresh container is
    what actually rules out staleness here; the keying is defence in depth against a lane change
    nobody would think to re-check this against, and that is why it is still worth testing.
    """
    marks = [_mark_line(s) for s, _ in _marked("test")]
    guard = _guard("test")["run"]
    # Run 1 marks everything.
    first = _env(tmp_path, GITHUB_JOB="test", GITHUB_RUN_ID="111", GITHUB_RUN_ATTEMPT="1")
    assert _run("\n".join(["set -e", *marks]), first).returncode == 0
    # Run 2 shares RUNNER_TEMP but marks nothing. It must NOT inherit run 1's markers.
    second = _env(tmp_path, GITHUB_JOB="test", GITHUB_RUN_ID="222", GITHUB_RUN_ATTEMPT="1")
    r = _run(guard, second)
    assert r.returncode != 0, (
        "a marker file left by a DIFFERENT run satisfied the guard, so a run whose steps were all "
        f"dropped would pass silently.\n{r.stdout}"
    )
    # ...and a RETRY of run 1 must not inherit run 1's either.
    retry = _env(tmp_path, GITHUB_JOB="test", GITHUB_RUN_ID="111", GITHUB_RUN_ATTEMPT="2")
    assert _run(guard, retry).returncode != 0, (
        "a re-run inherited the first attempt's markers, so a step dropped only on the retry passes"
    )
    # ...nor may the OTHER job in the same run inherit them.
    sibling = _env(tmp_path, GITHUB_JOB="migrations", GITHUB_RUN_ID="111", GITHUB_RUN_ATTEMPT="1")
    assert _run(_guard("migrations")["run"], sibling).returncode != 0, (
        "the two required jobs share one marker file, so one job's markers answer for the other's "
        "dropped steps"
    )


def test_assert_with_no_expected_keys_REFUSES_instead_of_passing(tmp_path):
    """The script's own anti-vacuity check, exercised rather than trusted.

    `assert` with an empty expectation list would print "All 0 expected step(s) executed" and exit 0
    — a guard that proves nothing while reporting that it proved everything. That is how a guard
    ends up shipped and dead, which this repo has now done twice (#751's fence, #751's own guard).
    """
    r = _run(f"{SCRIPT} assert", _env(tmp_path))
    assert r.returncode == 2, f"expected a usage refusal, got {r.returncode}: {r.stdout} {r.stderr}"
    assert "no expected keys" in (r.stdout + r.stderr)


def test_mark_APPENDS_so_one_step_does_not_erase_its_predecessors(tmp_path):
    """`>` instead of `>>` in the script would leave only the last step's key.

    The guard would then redden on every run — fail-closed, but it would look like the guard is
    broken rather than like a real drop, and that is the state in which a correct guard gets removed.
    """
    env = _env(tmp_path)
    assert _run(f"{SCRIPT} mark alpha && {SCRIPT} mark beta", env).returncode == 0
    r = _run(f"{SCRIPT} assert --always alpha beta", env)
    assert r.returncode == 0, f"the second mark erased the first: {r.stdout} {r.stderr}"


def test_a_key_is_matched_WHOLE_not_as_a_substring(tmp_path):
    """`build` must not be satisfied by `web-build`, and `test` not by `web-test`.

    Both pairs are live key names in the `test` job, so a substring match would mean dropping the
    real `Build` or `Test` step — the two most consequential steps in the whole workflow — is
    invisible because an SPA step of a similar name ran.
    """
    env = _env(tmp_path)
    assert _run(f"{SCRIPT} mark web-build && {SCRIPT} mark web-test", env).returncode == 0
    r = _run(f"{SCRIPT} assert --always build", env)
    assert r.returncode != 0, (
        "the key 'build' was satisfied by a marker for 'web-build' — a dropped `dotnet build` would "
        "pass unnoticed"
    )


def test_a_degraded_run_IDENTITY_refuses_rather_than_sharing_a_marker_path(tmp_path):
    """`GITHUB_RUN_ID` absent must REFUSE, not fall back to a name every run shares.

    The first version of `marker_path` defaulted to `nojob`/`norunid`/`1`. Those are reusable, so a
    leftover marker from any earlier run on the host would satisfy the guard on a run whose step was
    dropped — a silent PASS, which is the precise failure the run-keying exists to remove,
    reintroduced by the code implementing it. Found by cold review.

    Asserted on BOTH sub-commands: a refusal that only `assert` honoured would let `mark` write to a
    shared path and leave the two disagreeing about where the file is.
    """
    env = _env(tmp_path)
    for var in ("GITHUB_RUN_ID", "GITHUB_JOB", "GITHUB_RUN_ATTEMPT"):
        degraded = {k: v for k, v in env.items() if k != var}
        for argv in (f"{SCRIPT} mark alpha", f"{SCRIPT} assert --always alpha"):
            r = _run(argv, degraded)
            assert r.returncode != 0, (
                f"with {var} unset, `{argv.split()[-2]}` continued and used a fallback path that "
                f"other runs also use — a stale marker there passes the guard on a dropped run.\n"
                f"{r.stdout}{r.stderr}"
            )
            assert "cannot identify this run" in (r.stdout + r.stderr), (
                f"refused, but without naming the cause: {r.stdout!r} {r.stderr!r}")
    assert not list(tmp_path.iterdir()), (
        "a degraded-identity `mark` still created a marker file somewhere under RUNNER_TEMP")


def test_the_marker_identity_is_REPORTED_on_stdout_every_run(tmp_path):
    """The line that settled `GITHUB_RUN_ATTEMPT`, kept as standing evidence.

    Worth recording HOW that was settled, because the first two attempts were both bad. Grepping a
    job log for the variable NAME proves nothing (logs do not dump the environment). Inferring it
    from the ABSENCE of a "not set" warning proves nothing either, because that warning goes to
    stderr and whether step stderr reaches a job log here was itself never established — the control
    offered for that was an `::error::` this script writes to STDOUT. So the script was made to
    REPORT its resolved identity on stdout, where capture is not in question, and the answer was read
    off ersatztv#756's own PR run: `Marker identity: job=test run=1916 attempt=1 (from the runner)`,
    and the same for `migrations`. That is what promoted the variable from warn-and-default to
    required.

    Asserted because cold review demonstrated three mutations of this reporting — deleting the echo,
    mis-deriving the attempt, inverting the provenance — all surviving a 50-green suite. It is a
    documented contract (the record's `mechanics:`), and a future reader is told to trust it.
    """
    marks = [_mark_line(s) for s, _ in _marked("test")]
    r = _run("\n".join(["set -e", *marks, _guard("test")["run"]]),
             _env(tmp_path, GITHUB_RUN_ID="1916", GITHUB_RUN_ATTEMPT="4"))
    assert r.returncode == 0, r.stdout + r.stderr
    assert "Marker identity: job=test run=1916 attempt=4 (from the runner)" in r.stdout, (
        "the guard did not report the identity its marker path was actually keyed on, so a reader "
        f"cannot audit the keying from a run log: {r.stdout!r}")


def test_a_skip_gate_that_empties_the_expected_set_REFUSES(tmp_path):
    """The anti-vacuity check has to run AFTER the gate, not only on argv. Cold review reproduced
    this exactly:

        ETV_DOCS_ONLY=true … assert --always --gated foo
        -> "All 0 expected step(s) executed", exit 0

    The argv check cannot see it, because the set is emptied by the gate rather than by the caller.
    Unreachable with today's argv, but it contradicted the comment directly above it — and "reports
    that it proved everything while proving nothing" is the failure this whole file exists to remove.
    """
    r = _run(f"{SCRIPT} assert --always --gated foo", _env(tmp_path, ETV_DOCS_ONLY="true"))
    assert r.returncode != 0, (
        f"the guard passed with an empty post-gate expectation set: {r.stdout!r}")
    assert "no expected keys" in (r.stdout + r.stderr).lower() or "NO expected keys" in r.stderr


@pytest.mark.parametrize("revalidate", ["true", "false", "", None],
                         ids=lambda v: f"reval-{v if v is not None else 'unset'}")
@pytest.mark.parametrize("docs_only", ["true", "false", "", None],
                         ids=lambda v: f"docs-{v if v is not None else 'unset'}")
def test_the_skip_gate_over_the_WHOLE_value_matrix(docs_only, revalidate, tmp_path):
    """Every combination of the two gate values, not just the diagonal — cold review's last finding.

    Round 3 fixed the suite's blindness to the production value `false`, but still only exercised
    matched pairs and single-`true` cases. `(true, true)` is REACHABLE — a docs-only PR merged to
    `main` whose tree was already validated sets both — and an exclusive-or regression would pass
    every other test here while demanding all the gated markers on a run that legitimately skipped
    those steps. That reddens BOTH required contexts, which is the false-red direction: it deadlocks
    every merge rather than letting one through.

    The property asserted is the whole contract in one line: with only the `--always` keys marked,
    the guard passes exactly when the gate says the gated steps were skipped — `true` in EITHER
    variable, and nothing else. Sixteen cases, so no combination is a special case anyone has to
    remember.

    On `unset`: the workflow's `env:` block always defines both, emitting EMPTY for an output the
    producing step never wrote, so unset is not reachable through the workflow. It is covered because
    the script is also runnable by hand, and because "not exactly true" is the property that must
    hold for every spelling rather than for an enumerated list.
    """
    job = "test"
    always, gated = _guard_buckets(job)
    marks = [_mark_line(s) for s, k in _marked(job) if k in always]
    r = _run("\n".join(["set -e", *marks, _guard(job)["run"]]),
             _env(tmp_path, ETV_DOCS_ONLY=docs_only, ETV_REVALIDATE_SKIP=revalidate))
    should_skip = docs_only == "true" or revalidate == "true"
    assert (r.returncode == 0) is should_skip, (
        f"with docs_only={docs_only!r} and revalidate={revalidate!r} the guard "
        f"{'passed' if r.returncode == 0 else 'failed'}, expected it to "
        f"{'skip the gated keys' if should_skip else 'require them'}. The gate must treat a value as "
        "a skip if and only if it is exactly `true` in EITHER variable.\n" + r.stdout + r.stderr)