diff --git a/Makefile b/Makefile
index bf2a77c..474120b 100644
--- a/Makefile
+++ b/Makefile
@@ -21,6 +21,8 @@ test-suite: ## rigforge test suite: unit + black-box, every CPU/OS profile simul
 	bash tests/test-docs-clone-pin.sh
 	bash tests/test-staged-rebuild.sh
 	bash tests/test-tool-pin-watch.sh
+	python3 tests/test-worker-proof.py
+	bash tests/test-worker-proof.sh
 	bash tests/test-coverage-cache.sh
 	bash tests/test-revision-key.sh
 
diff --git a/docs/dev/worker-proof.md b/docs/dev/worker-proof.md
new file mode 100644
index 0000000..71fd670
--- /dev/null
+++ b/docs/dev/worker-proof.md
@@ -0,0 +1,133 @@
+# Ordered worker proof receipts
+
+`tests/e2e-pithead.sh` keeps each invocation on one reserved rig, in one shell,
+under one config/runtime snapshot. This change enables no concurrency and removes
+no per-PR or release witness. `network-summary` remains sanitized diagnostic
+information; it is not a completion receipt.
+
+## Stateful groups and retained assertions
+
+`all` runs the rows below in order. `network-sequence` runs the first four rows in
+that same order. Neither may be reconstructed by joining independently reset
+phases. Standalone phases remain available for diagnosis and existing row owners;
+they cannot substitute for either complete group.
+
+| Phase row | Retained successful assertions | Dependency and restoration |
+| --- | --- | --- |
+| `connect` (3) | Stratum job, accepted share, generated worker label | Sets the pool and restarts the physical worker. |
+| `worker-api` (5) | Open read, bind policy, restricted write refusal, unauthenticated refusal, matching bearer read | Uses the connected worker; restores the open token posture. |
+| `api-impact` (3) | Sister API available, loaded hashrate within baseline tolerance, responsiveness under mining load | Uses live hashing; starts/stops recorded load processes and disables the sister API. |
+| `network` (9) | Pool-only peers, control-implied listener, enabled API listener, miner-only listener policy, complete summary superset and metadata, keyed feed revision with a private key file, token leak sweep, pool-password leak sweep, disabled listener closed | Consumes the preceding mutations; restores disabled API/control and the open token posture. |
+| `stratum-auth` (4) | Right credential mines, wrong credential rejected, no jobs with wrong credential, recovery with right credential | Fault and recovery stay together; original config is restored by outer cleanup. |
+| `dashboard` (2) | Worker initially online, stopped worker becomes offline | Requires initial presence; stops then restarts the same worker. |
+| `dev-fee` (1) | Legacy minimum migration, zero donation, runtime reductions without rebuilding, original artifacts restored | The existing compound assertion includes real build/runtime and artifact restoration. |
+| `control` (5) | Apply accepted, donation applied, config persisted, appliance pool edit atomically refused, thermal change reaches a new feed within its deadline | Preserves systemd/timer profiling and the full observation window; restores profiling and outer snapshot. |
+| `access-tokens` (7) | Named token refused by miner, raw and derived reads accepted by sister API, apply accepted and applied, both revocation checks | Mutation and revocation stay together; outer cleanup restores tokens and donation. |
+
+Each replay group (`control-replay-thermal`, `control-replay-pools`) retains its
+runner-owned fixture/reservation, nested connect assertions (3), scalar apply
+sequence, pools mutation, optional thermal mutation, publication window and final
+assertion (1). Nested connect belongs to that replay row, not a separate group.
+Neither replay is added to or removed from `all`.
+
+Failures stop subsequent mutations. Cleanup still finishes refresh profiling,
+stops recorded load processes, restores and reapplies the saved config, compares
+its bytes, restores the original XMRig active state, and retains the snapshot if
+restoration fails. Receipt emission follows cleanup on normal and error exits.
+An interrupted or incomplete group cannot pass by presenting its successful prefix.
+
+## Binding an invocation
+
+The runner may supply `E2E_PROOF_CONTEXT` as one JSON object with exactly these
+fields. Do not put deployment coordinates, credentials, config bodies or private
+paths into this object.
+
+| Field | Meaning |
+| --- | --- |
+| `schema` | Integer `1`. |
+| `head` | Exact lowercase, forty-character Git commit ID. |
+| `ref` | Full Git ref identifying that commit. The ref must resolve locally. |
+| `input` | SHA-256 digest of the immutable, non-secret input manifest agreed by the runner. |
+| `run` | Fresh thirty-two-character lowercase hexadecimal attempt ID. |
+| `group` | Exact existing phase argument, `all`, or `network-sequence`. |
+| `row` | Runner-owned opaque lowercase row slug, unique within the required plan. |
+| `platform` | `linux-x86_64` or `linux-aarch64`, checked against the executing system. |
+| `allowed_skips` | Explicit map of permitted phase IDs to fixed skip reasons; normally `{}`. |
+
+Before acquiring the rig lock and after restoration, the producer checks the
+requested HEAD, ref resolution, platform, tracked-tree cleanliness, and absence of
+untracked executable inputs under the product/test directories. These calls mark
+the exact checkout path as trusted so root can read operator-owned checkouts.
+The input digest is a runner assertion: this script cannot establish the coordinator
+revision, image identities or fixture provenance. The runner must resolve and
+freeze those inputs before dispatch and retain the corresponding manifest.
+
+Without context the script emits `worker-proof: {"schema":1,"bound":false}` and
+retains the existing CLI behavior. This is explicitly unavailable bound evidence;
+the receipt checker rejects it. Supplying malformed context fails before rig
+mutation. This compatibility path does not implement the future required status.
+
+## Receipt and required AND
+
+A bound invocation emits the prefix `worker-proof:`, a space, and a JSON object:
+`schema`, `context`, `phases`, `cleanup`, and `exit_code`. Each phase has exactly
+`id`, `passed`, `failed`, and `skipped` (a list of fixed reason codes). The counts
+include assertions made in child shells. No assertion prose or response data is
+copied into the receipt. Counts are checked against the complete rows above;
+neither fewer nor extra assertions silently pass.
+`tests/test-worker-proof.py` derives each row's count from the `ok` calls in the
+harness phase bodies and the helpers they call, across every sourced `e2e-*.sh`
+file, so an assertion added to a phase fails the suite until the checker's table
+matches it.
+
+Only two existing fixture-absence skips are eligible for explicit authorization:
+
+- `stratum-auth: stratum-fixture-absent`, in `all` only. The standalone auth gate
+  still requires its fixture.
+- `dashboard: dashboard-fixture-absent`, in `all` or `dashboard`.
+
+A skipped row must contain exactly its authorized reason and no successful or
+failed assertions. A passing row remains valid when its skip was permitted but
+unused. Unknown or unexplained skips fail. An offline check after missing initial
+presence remains a failure and cannot be authorized as a skip.
+
+Extract each receipt as JSON into its own artifact and validate against an
+independently supplied plan, a nonempty JSON array of expected contexts:
+
+```bash
+python3 tests/worker-proof.py check PLAN.json RECEIPT.json MORE-RECEIPTS.json
+```
+
+The plan must retain every required row for one head/ref/input/attempt. Receipt
+arrival order may vary, but phase order within each group cannot. The checker
+rejects missing, duplicate, substituted or stale rows; wrong platform or context;
+missing, reordered, duplicated or unknown phases; partial assertion counts;
+unapproved skips; malformed/truncated data; nonzero exit; and failed cleanup.
+It must receive the trusted plan, never derive expectations from the artifacts it
+is checking. Receipts are not signed attestations: trust remains with the runner
+that owns the checkout, plan, fixtures and artifact collection.
+
+## Acceptance still required before rollout
+
+The sibling runner and evidence-preflight repositories must wire this required AND
+into the same current-head product/release status. Until then, local receipt
+validation alone does not establish the fleet gate. Keep all existing required
+statuses and per-PR proofs, including both independent bench-ci `make check` runs.
+The shared proof manifest and rank-7 immutable inputs remain coordinated work;
+review complete before/after behavior, assertion, input and platform inventories
+by hand until that manifest is available. Unknown or truncated evidence blocks
+acceptance, and existing row ownership remains authoritative.
+
+After rank-1 capacity acceptance, only independently proven groups may overlap,
+with distinct mutable rigs **and** stacks. Measure existing benches; obtain the
+owner's decision before adding a lane. Keep hashing/accepted-share, systemd/feed,
+donation, performance baselines, all `e2e-real` release phases, and every other
+currently required live/release witness.
+
+Before rollout, run hosted checks and paired live serial/concurrent acceptance on
+the same immutable inputs. Compare complete assertion/platform manifests, gate
+median/p90, occupied setup hours, failures and restoration. Record the baseline on
+normal authorized runs; one week after rollout compare paired PR/head latency,
+stage median/p90 minutes, occupied hours, failures and cleanup outcomes, reporting
+sample sizes and load. Roll back on lost evidence, cleanup failure or increased
+KVM queueing. None of those live measurements is claimed by the local regressions.
diff --git a/tests/README.md b/tests/README.md
index 576900c..d56ebaa 100644
--- a/tests/README.md
+++ b/tests/README.md
@@ -7,6 +7,9 @@ compiles XMRig and mines. Each layer covers what the one below it has to stub.
 contract credential regression, plus access-token readiness, release clone-pin and offline tool-pin
 watcher regressions. The Bash suite exits nonzero if it aborts before its final
 `rigforge tests: ... passed, ... failed` summary, even when the shell would otherwise return zero.
+Worker proof schema/join and dispatcher regressions also run in `make test-suite`;
+see [ordered worker proof receipts](../docs/dev/worker-proof.md) for the retained
+phase inventory, binding contract and rollout requirements.
 The kcov container measures the Bash suite only; the credential regression always runs in the
 GitHub Test suite job, including argument inspection, password rotation and EXIT restoration.
 
diff --git a/tests/e2e-pithead.sh b/tests/e2e-pithead.sh
index 856481e..53b4d00 100644
--- a/tests/e2e-pithead.sh
+++ b/tests/e2e-pithead.sh
@@ -14,20 +14,22 @@ RIGFORGE="$HERE/rigforge.sh"
 CFG="$HERE/config.json"
 source "$(dirname "${BASH_SOURCE[0]}")/e2e-pithead-control.sh" && source "$(dirname "${BASH_SOURCE[0]}")/e2e-pithead-tokens.sh" # phase_control (#509), phase_access_tokens (#516)
 
-source "$(dirname "${BASH_SOURCE[0]}")/e2e-control-replay.sh" && source "$HERE/tests/e2e-summary-diagnostics.sh"
+source "$(dirname "${BASH_SOURCE[0]}")/e2e-control-replay.sh" && source "$HERE/tests/e2e-summary-diagnostics.sh" && source "$HERE/tests/e2e-worker-proof.sh"
 # bench-ci-fixture: pithead-control control-replay-thermal control-replay-pools
 PASS=0 FAIL=0 E2E_EXIT_RC=0
 ok() {
+    [ "${E2E_PROOF_ACTIVE:-0}" != 1 ] || proof_assert pass
     PASS=$((PASS + 1))
     printf '  \033[1;32m✓\033[0m %s\n' "$1"
 }
 bad() {
+    [ "${E2E_PROOF_ACTIVE:-0}" != 1 ] || proof_assert fail
     FAIL=$((FAIL + 1))
     printf '  \033[1;31m✗\033[0m %s\n' "$1" >&2
     return 1
 }
-skip() { printf '  \033[1;33m∙\033[0m SKIP: %s\n' "$1"; }
-phase() { printf '\n\033[1m== e2e-pithead: %s ==\033[0m\n' "$1"; }
+skip() { proof_skip "$1" && printf '  \033[1;33m∙\033[0m SKIP: %s\n' "$1"; }
+phase() { proof_phase "${FUNCNAME[1]}" && printf '\n\033[1m== e2e-pithead: %s ==\033[0m\n' "$1"; }
 die() {
     printf '\033[31me2e-pithead: %s\033[0m\n' "$1" >&2
     exit 2
@@ -120,13 +122,14 @@ _cleanup() {
     }
     [ "$cleanup_ok" != 1 ] || rm -f "$SAVED_CFG" || cleanup_ok=0
     [ "${REPLAY_RUNNER_LOCK:-0}" = 1 ] || rm -f "${RIG_LOCK_HOLDER:-${RIG_LOCK_FILE:-/var/lock/rig-e2e.lock}.holder}" || true
+    [ "${E2E_PROOF_ACTIVE:-0}" != 1 ] || proof_cleanup "$cleanup_ok"
     [ "$cleanup_ok" = 1 ]
 }
 snapshot_config() {
     SAVED_CFG="$(mktemp)"
     cp "$CFG" "$SAVED_CFG"
     systemctl is-active --quiet xmrig 2>/dev/null && SAVED_XMRIG_ACTIVE=1 || SAVED_XMRIG_ACTIVE=0
-    trap 'E2E_EXIT_RC=$?; trap - EXIT; _cleanup || [ "$E2E_EXIT_RC" -ne 0 ] || E2E_EXIT_RC=1; exit "$E2E_EXIT_RC"' EXIT
+    trap 'E2E_EXIT_RC=$?; trap - EXIT; _cleanup || [ "$E2E_EXIT_RC" -ne 0 ] || E2E_EXIT_RC=1; if [ "${E2E_PROOF_ACTIVE:-0}" = 1 ]; then proof_finish "$E2E_EXIT_RC" || [ "$E2E_EXIT_RC" -ne 0 ] || E2E_EXIT_RC=1; fi; exit "$E2E_EXIT_RC"' EXIT
 }
 
 set_cfg() { # <jq program> [soft] [jq options...] -- soft: return 1 instead of dying when the edit or the apply fails (#514)
@@ -523,6 +526,7 @@ summary() {
     printf '0 failed\n'
 }
 
+proof_preflight "${1:-all}"
 require_preflight "$@"
 # Validate the phase against its own `phase_<name>` function: a typo dies with the rig untouched.
 # #183: the lock follows, before snapshot_config (whose _cleanup runs `apply`) and the first API touch.
@@ -530,6 +534,7 @@ require_preflight "$@"
 replay_lock "${1:-all}"
 
 snapshot_config
+proof_begin
 case "${1:-all}" in
 connect) phase_connect ;;
 worker-api) phase_worker_api ;;
@@ -552,4 +557,5 @@ all)
 esac
 _cleanup || bad "pre-test config/runtime restoration failed; snapshot retained at $SAVED_CFG"
 trap - EXIT
+[ "${E2E_PROOF_ACTIVE:-0}" != 1 ] || proof_finish "$FAIL" || bad "worker proof incomplete or invalid"
 summary
diff --git a/tests/e2e-worker-proof.sh b/tests/e2e-worker-proof.sh
new file mode 100644
index 0000000..ee7f176
--- /dev/null
+++ b/tests/e2e-worker-proof.sh
@@ -0,0 +1,84 @@
+# shellcheck shell=bash
+# shellcheck disable=SC2034
+
+proof_git() { git -C "$HERE" -c safe.directory="$HERE" "$@"; }
+
+proof_identity() {
+    local head ref platform untracked
+    head=$(printf '%s' "$E2E_PROOF_CONTEXT" | jq -r .head) || return 1
+    ref=$(printf '%s' "$E2E_PROOF_CONTEXT" | jq -r .ref) || return 1
+    platform=$(printf '%s' "$E2E_PROOF_CONTEXT" | jq -r .platform) || return 1
+    untracked=$(proof_git ls-files --others --exclude-standard -- rigforge.sh util systemd tests) || return 1
+    [ "$(proof_git rev-parse --verify HEAD 2>/dev/null)" = "$head" ] &&
+        [ "$(proof_git rev-parse --verify "$ref^{commit}" 2>/dev/null)" = "$head" ] &&
+        [ "$(uname -s | tr '[:upper:]' '[:lower:]')-$(uname -m)" = "$platform" ] &&
+        proof_git diff --quiet HEAD -- &&
+        [ -z "$untracked" ]
+}
+
+proof_preflight() {
+    E2E_PROOF_ACTIVE=0
+    [ -n "${E2E_PROOF_CONTEXT:-}" ] || {
+        printf 'worker-proof: {"schema":1,"bound":false}\n'
+        return 0
+    }
+    E2E_PROOF_CONTEXT=$(printf '%s' "$E2E_PROOF_CONTEXT" | python3 "$HERE/tests/worker-proof.py" context "$1") || die "invalid worker proof context"
+    proof_identity || die "worker proof requires the requested clean head and ref"
+}
+
+proof_begin() {
+    [ -n "${E2E_PROOF_CONTEXT:-}" ] || return 0
+    E2E_PROOF_ACTIVE=1 E2E_PROOF_CLEANUP=unknown
+    E2E_PROOF_DIR=$(mktemp -d) || return 1
+    printf '%s\n' "$E2E_PROOF_CONTEXT" >"$E2E_PROOF_DIR/context" || return 1
+    : >"$E2E_PROOF_DIR/events" || return 1
+    trap 'exit 130' INT
+    trap 'exit 143' TERM
+}
+
+proof_phase() {
+    [ "${E2E_PROOF_ACTIVE:-0}" = 1 ] || return 0
+    local id="${1#phase_}" group
+    id=${id//_/-}
+    group=$(printf '%s' "$E2E_PROOF_CONTEXT" | jq -r .group) || return 1
+    case "$group:$id" in
+    control-replay-*:connect) return 0 ;;
+    control-replay-*:control-replay) id=$group ;;
+    esac
+    printf '{"event":"phase","id":"%s"}\n' "$id" >>"$E2E_PROOF_DIR/events"
+}
+
+proof_assert() {
+    printf '{"event":"%s"}\n' "$1" >>"$E2E_PROOF_DIR/events"
+}
+
+proof_skip() {
+    [ "${E2E_PROOF_ACTIVE:-0}" = 1 ] || return 0
+    local reason=unexplained
+    case "$1" in
+    'E2E_STRATUM_PASS not set (stack auth off, or secret not provided) — phases skipped') reason=stratum-fixture-absent ;;
+    'E2E_DASH_URL not set — dashboard phases skipped (agree fixtures with pithead#209)') reason=dashboard-fixture-absent ;;
+    esac
+    printf '{"event":"skip","reason":"%s"}\n' "$reason" >>"$E2E_PROOF_DIR/events"
+}
+
+proof_cleanup() {
+    [ "$1" = 1 ] || E2E_PROOF_CLEANUP=fail
+    [ "${E2E_PROOF_CLEANUP:-unknown}" != unknown ] || E2E_PROOF_CLEANUP=pass
+}
+
+proof_finish() {
+    local rc="$1" result status=0
+    proof_identity || rc=1
+    if [ -n "${E2E_PROOF_DIR:-}" ]; then
+        result=$(python3 "$HERE/tests/worker-proof.py" receipt "$E2E_PROOF_DIR/context" "$E2E_PROOF_DIR/events" "${E2E_PROOF_CLEANUP:-fail}" "$rc") || status=1
+        if ! rm -rf "$E2E_PROOF_DIR"; then
+            result='' status=1
+        fi
+        [ -z "$result" ] || printf 'worker-proof: %s\n' "$result"
+    else
+        status=1
+    fi
+    E2E_PROOF_ACTIVE=0
+    return "$status"
+}
diff --git a/tests/test-worker-proof.py b/tests/test-worker-proof.py
new file mode 100644
index 0000000..e78b0f8
--- /dev/null
+++ b/tests/test-worker-proof.py
@@ -0,0 +1,384 @@
+#!/usr/bin/env python3
+import copy
+import json
+import os
+import re
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+HERE = Path(__file__).resolve().parent
+PROGRAM = Path(os.environ.get("WORKER_PROOF_IMPLEMENTATION", HERE / "worker-proof.py"))
+ALL_PHASES = [
+    "connect",
+    "worker-api",
+    "api-impact",
+    "network",
+    "stratum-auth",
+    "dashboard",
+    "dev-fee",
+    "control",
+    "access-tokens",
+]
+FUNCTION = re.compile(r"([a-z_][a-z0-9_]*)\(\) *[{(](.*)$")
+OK_CALL = re.compile(r"(?:^|[;&|(){]|\bthen|\belse|\bdo)\s*ok\s", re.M)
+SOURCE = re.compile(r"(?:\bsource|(?:^|[;&|])\s*\.)\s[^\n]*?\b(e2e-[a-z-]+\.sh)\b", re.M)
+WORD = re.compile(r"[a-z_][a-z0-9_]*")
+
+
+def harness_functions(directory=HERE):
+    sources = ["e2e-pithead.sh"]
+    functions = {}
+    for source in sources:
+        text = (directory / source).read_text(encoding="utf-8")
+        sources.extend(name for name in SOURCE.findall(text) if name not in sources)
+        name = None
+        for line in text.splitlines():
+            match = FUNCTION.match(line)
+            if name is None and match:
+                name, rest = match.groups()
+                if re.sub(r"\s+#.*$", "", rest).endswith(("}", ")")):
+                    functions[name], name = rest, None
+                else:
+                    functions[name] = ""
+            elif name is not None and line in ("}", ")"):
+                name = None
+            elif name is not None and not line.lstrip().startswith("#"):
+                functions[name] += line + "\n"
+    return functions
+
+
+def harness_passes(functions, name, seen=()):
+    body = functions[name]
+    called = {word for word in WORD.findall(body) if word in functions and word not in seen and word != name}
+    return len(OK_CALL.findall(body)) + sum(harness_passes(functions, call, seen + (name,)) for call in called)
+
+
+def harness_counts(directory=HERE):
+    functions = harness_functions(directory)
+    return {
+        group: harness_passes(functions, "phase_" + group.replace("-", "_"))
+        for group in ALL_PHASES + ["control-replay-thermal", "control-replay-pools"]
+    }
+
+
+PASS_COUNTS = harness_counts()
+
+
+def context(group="network-sequence", row="primary", platform="linux-x86_64", skips=None):
+    return {
+        "schema": 1,
+        "head": "a" * 40,
+        "ref": "refs/heads/develop",
+        "input": "b" * 64,
+        "run": "c" * 32,
+        "group": group,
+        "row": row,
+        "platform": platform,
+        "allowed_skips": skips or {},
+    }
+
+
+def events(phases, skip=None, failure=None):
+    rows = []
+    for phase in phases:
+        rows.append({"event": "phase", "id": phase})
+        if phase == skip:
+            reason = "stratum-fixture-absent" if skip == "stratum-auth" else skip + "-fixture-absent"
+            rows.append({"event": "skip", "reason": reason})
+        elif phase == failure:
+            rows.append({"event": "fail"})
+        else:
+            rows.extend({"event": "pass"} for _ in range(PASS_COUNTS[phase]))
+    return rows
+
+
+class WorkerProofTest(unittest.TestCase):
+    def invoke(self, *args, stdin=None):
+        return subprocess.run(
+            [sys.executable, str(PROGRAM), *map(str, args)],
+            input=stdin,
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+
+    def files(self, values):
+        directory = tempfile.TemporaryDirectory()
+        paths = []
+        for index, value in enumerate(values):
+            path = Path(directory.name) / f"input-{index}.json"
+            if isinstance(value, str):
+                path.write_text(value, encoding="utf-8")
+            else:
+                path.write_text(json.dumps(value), encoding="utf-8")
+            paths.append(path)
+        self.addCleanup(directory.cleanup)
+        return paths
+
+    def make_receipt(self, ctx, rows=None, cleanup="pass", exit_code=0):
+        phase_ids = ALL_PHASES if ctx["group"] == "all" else (
+            ALL_PHASES[:4] if ctx["group"] == "network-sequence" else [ctx["group"]]
+        )
+        paths = self.files([ctx, "\n".join(json.dumps(row) for row in (rows or events(phase_ids)))])
+        result = self.invoke("receipt", paths[0], paths[1], cleanup, exit_code)
+        return result, json.loads(result.stdout) if result.stdout else None
+
+    def check(self, plan, receipts):
+        paths = self.files([plan, *receipts])
+        return self.invoke("check", *paths)
+
+    def assert_invalid(self, result):
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.stdout, "")
+        self.assertEqual(result.stderr, "worker-proof: invalid input\n")
+
+    def assert_failed_evidence(self, result):
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.stderr, "")
+        receipt = json.loads(result.stdout)
+        self.assertEqual(receipt["phases"], [])
+        self.assertNotEqual(receipt["exit_code"], 0)
+
+    def test_context_prints_canonical_json(self):
+        ctx = context()
+        result = self.invoke("context", "network-sequence", stdin=json.dumps(ctx))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, json.dumps(ctx, sort_keys=True, separators=(",", ":")) + "\n")
+
+    def test_context_rejects_schema_and_group_mismatch(self):
+        ctx = context()
+        ctx["schema"] = True
+        self.assert_invalid(self.invoke("context", "network-sequence", stdin=json.dumps(ctx)))
+        self.assert_invalid(self.invoke("context", "all", stdin=json.dumps(context())))
+
+    def test_context_rejects_extra_missing_and_duplicate_fields(self):
+        extra = context()
+        extra["extra"] = 1
+        self.assert_invalid(self.invoke("context", "network-sequence", stdin=json.dumps(extra)))
+        missing = context()
+        del missing["row"]
+        self.assert_invalid(self.invoke("context", "network-sequence", stdin=json.dumps(missing)))
+        duplicate = json.dumps(context()).replace('"schema": 1', '"schema": 1, "schema": 1')
+        self.assert_invalid(self.invoke("context", "network-sequence", stdin=duplicate))
+
+    def test_context_rejects_bad_bindings(self):
+        mutations = {
+            "head": "A" * 40,
+            "ref": "refs/heads/a..b",
+            "input": "b" * 63,
+            "run": "c" * 33,
+            "row": "Bad Row",
+            "platform": "linux-riscv64",
+        }
+        for key, value in mutations.items():
+            with self.subTest(key=key):
+                ctx = context()
+                ctx[key] = value
+                self.assert_invalid(self.invoke("context", ctx["group"], stdin=json.dumps(ctx)))
+
+    def test_context_accepts_canonical_pull_ref(self):
+        ctx = context()
+        ctx["ref"] = "refs/pull/42/head"
+        result = self.invoke("context", ctx["group"], stdin=json.dumps(ctx))
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_allowed_skips_are_narrow(self):
+        valid = [
+            context("all", skips={"stratum-auth": "stratum-fixture-absent"}),
+            context("all", skips={"dashboard": "dashboard-fixture-absent"}),
+            context("dashboard", skips={"dashboard": "dashboard-fixture-absent"}),
+        ]
+        for ctx in valid:
+            self.assertEqual(self.invoke("context", ctx["group"], stdin=json.dumps(ctx)).returncode, 0)
+        invalid = [
+            context("stratum-auth", skips={"stratum-auth": "stratum-fixture-absent"}),
+            context("network", skips={"network": "fixture-absent"}),
+            context("all", skips={"dashboard": "unknown"}),
+        ]
+        for ctx in invalid:
+            self.assert_invalid(self.invoke("context", ctx["group"], stdin=json.dumps(ctx)))
+
+    def test_receipt_folds_events_and_validates(self):
+        result, receipt = self.make_receipt(context())
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual([row["id"] for row in receipt["phases"]], ALL_PHASES[:4])
+        self.assertEqual([row["passed"] for row in receipt["phases"]], [PASS_COUNTS[phase] for phase in ALL_PHASES[:4]])
+
+    def test_receipt_emits_failed_evidence(self):
+        ctx = context("connect")
+        result, receipt = self.make_receipt(ctx, events(["connect"], failure="connect"), "fail", 7)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.stderr, "")
+        self.assertEqual(receipt["phases"][0]["failed"], 1)
+        self.assertEqual(receipt["cleanup"], "fail")
+        self.assertEqual(receipt["exit_code"], 7)
+
+    def test_receipt_normalizes_unknown_cleanup_to_failure(self):
+        result, receipt = self.make_receipt(context("connect"), cleanup="unknown")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(result.stderr, "")
+        self.assertEqual(receipt["cleanup"], "fail")
+
+    def test_receipt_accepts_declared_optional_skip(self):
+        ctx = context("dashboard", skips={"dashboard": "dashboard-fixture-absent"})
+        result, receipt = self.make_receipt(ctx, events(["dashboard"], skip="dashboard"))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(receipt["phases"][0]["skipped"], ["dashboard-fixture-absent"])
+
+        all_context = context("all", skips={"stratum-auth": "stratum-fixture-absent"})
+        all_result, all_receipt = self.make_receipt(all_context, events(ALL_PHASES, skip="stratum-auth"))
+        self.assertEqual(all_result.returncode, 0, all_result.stderr)
+        self.assertEqual(all_receipt["phases"][4]["skipped"], ["stratum-fixture-absent"])
+
+    def test_receipt_rejects_unexplained_or_unknown_skip(self):
+        ctx = context("dashboard")
+        self.assert_failed_evidence(self.make_receipt(ctx, events(["dashboard"], skip="dashboard"))[0])
+        bad = [{"event": "phase", "id": "dashboard"}, {"event": "skip", "reason": "unknown"}]
+        self.assert_failed_evidence(self.make_receipt(ctx, bad)[0])
+
+    def test_receipt_rejects_event_before_phase(self):
+        for row in ({"event": "pass"}, {"event": "fail"}, {"event": "skip", "reason": "unknown"}):
+            with self.subTest(row=row):
+                self.assert_failed_evidence(self.make_receipt(context("connect"), [row])[0])
+
+    def test_receipt_rejects_reordered_duplicate_and_substituted_phases(self):
+        bad_orders = [
+            ["worker-api", "connect", "api-impact", "network"],
+            ["connect", "connect", "api-impact", "network"],
+            ["connect", "worker-api", "dashboard", "network"],
+        ]
+        for phases in bad_orders:
+            with self.subTest(phases=phases):
+                self.assert_failed_evidence(self.make_receipt(context(), events(phases))[0])
+
+    def test_receipt_rejects_malformed_event_json(self):
+        paths = self.files([context("connect"), '{"event":"phase","id":"connect"}\n{'])
+        self.assert_failed_evidence(self.invoke("receipt", paths[0], paths[1], "pass", 0))
+
+    def test_pass_counts_follow_harness_phase_bodies(self):
+        for group in PASS_COUNTS:
+            result, receipt = self.make_receipt(context(group))
+            self.assertEqual(result.returncode, 0, group)
+            self.assertEqual(receipt["phases"][0]["passed"], PASS_COUNTS[group])
+
+    def test_added_harness_assertion_changes_the_derived_count(self):
+        directory = tempfile.TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        copy_dir = Path(directory.name)
+        for path in HERE.glob("e2e-*.sh"):
+            (copy_dir / path.name).write_text(path.read_text(encoding="utf-8"), encoding="utf-8")
+        for name, anchor in (("e2e-pithead.sh", "phase_network() {\n"), ("e2e-refresh-window.sh", "refresh_profile_finish() {\n")):
+            text = (copy_dir / name).read_text(encoding="utf-8")
+            self.assertIn(anchor, text)
+            (copy_dir / name).write_text(text.replace(anchor, anchor + "    true && ok 'added'\n", 1), encoding="utf-8")
+        counts = harness_counts(copy_dir)
+        expected = dict(PASS_COUNTS, network=PASS_COUNTS["network"] + 1, control=PASS_COUNTS["control"] + 1)
+        expected["control-replay-thermal"] += 1
+        expected["control-replay-pools"] += 1
+        self.assertEqual(counts, expected)
+        rows = [{"event": "phase", "id": "network"}] + [{"event": "pass"}] * counts["network"]
+        result, _ = self.make_receipt(context("network"), rows)
+        self.assertNotEqual(result.returncode, 0)
+
+    def test_receipt_rejects_fewer_and_extra_passes(self):
+        ctx = context("connect")
+        too_few = events(["connect"])
+        too_few.pop()
+        too_many = events(["connect"]) + [{"event": "pass"}]
+        for rows in (too_few, too_many):
+            result, receipt = self.make_receipt(ctx, rows)
+            self.assertNotEqual(result.returncode, 0)
+            self.assertIsNotNone(receipt)
+
+    def test_check_joins_receipts_independent_of_order(self):
+        first = context("connect", "first")
+        second = context("dashboard", "second", "linux-aarch64")
+        _, first_receipt = self.make_receipt(first)
+        _, second_receipt = self.make_receipt(second)
+        result = self.check([first, second], [second_receipt, first_receipt])
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_check_accepts_repeated_group_on_distinct_rows_and_platforms(self):
+        first = context("connect", "first")
+        second = context("connect", "second", "linux-aarch64")
+        _, first_receipt = self.make_receipt(first)
+        _, second_receipt = self.make_receipt(second)
+        result = self.check([first, second], [first_receipt, second_receipt])
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_check_rejects_missing_and_duplicate_receipts(self):
+        first = context("connect", "first")
+        second = context("dashboard", "second", "linux-aarch64")
+        _, first_receipt = self.make_receipt(first)
+        self.assert_invalid(self.check([first, second], [first_receipt]))
+        self.assert_invalid(self.check([first, second], [first_receipt, first_receipt]))
+
+    def test_check_rejects_plan_binding_disagreement_and_duplicate_rows(self):
+        first = context("connect", "same")
+        stale = context("dashboard", "other", "linux-aarch64")
+        stale["head"] = "d" * 40
+        self.assert_invalid(self.check([first, stale], []))
+        duplicate = context("dashboard", "same", "linux-aarch64")
+        self.assert_invalid(self.check([first, duplicate], []))
+
+    def test_check_rejects_every_stale_receipt_binding(self):
+        original = context("connect", "first")
+        _, receipt = self.make_receipt(original)
+        replacements = {
+            "head": "d" * 40,
+            "ref": "refs/tags/v1",
+            "input": "e" * 64,
+            "run": "f" * 32,
+            "group": "network",
+            "row": "other",
+            "platform": "linux-aarch64",
+        }
+        for key, value in replacements.items():
+            with self.subTest(key=key):
+                changed = copy.deepcopy(receipt)
+                changed["context"][key] = value
+                self.assert_invalid(self.check([original], [changed]))
+
+    def test_check_rejects_mutated_phase_cleanup_and_exit(self):
+        ctx = context("connect")
+        _, receipt = self.make_receipt(ctx)
+        mutations = []
+        missing = copy.deepcopy(receipt)
+        missing["phases"] = []
+        mutations.append(missing)
+        negative = copy.deepcopy(receipt)
+        negative["phases"][0]["passed"] = -1
+        mutations.append(negative)
+        boolean = copy.deepcopy(receipt)
+        boolean["phases"][0]["passed"] = True
+        mutations.append(boolean)
+        cleanup = copy.deepcopy(receipt)
+        cleanup["cleanup"] = "fail"
+        mutations.append(cleanup)
+        exited = copy.deepcopy(receipt)
+        exited["exit_code"] = 1
+        mutations.append(exited)
+        extra = copy.deepcopy(receipt)
+        extra["extra"] = 1
+        mutations.append(extra)
+        for changed in mutations:
+            self.assert_invalid(self.check([ctx], [changed]))
+
+    def test_check_rejects_empty_plan_and_duplicate_json_keys(self):
+        self.assert_invalid(self.check([], []))
+        paths = self.files(['[{"schema":1,"schema":1}]'])
+        self.assert_invalid(self.invoke("check", paths[0]))
+
+        ctx = context("connect")
+        _, receipt = self.make_receipt(ctx)
+        duplicate = json.dumps(receipt).replace('"cleanup": "pass"', '"cleanup": "pass", "cleanup": "pass"')
+        paths = self.files([[ctx], duplicate])
+        self.assert_invalid(self.invoke("check", *paths))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test-worker-proof.sh b/tests/test-worker-proof.sh
new file mode 100644
index 0000000..5d50875
--- /dev/null
+++ b/tests/test-worker-proof.sh
@@ -0,0 +1,178 @@
+#!/usr/bin/env bash
+# shellcheck disable=SC2016,SC2034,SC2329
+set -euo pipefail
+ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
+
+if [ "${1:-}" = --case ]; then
+    mode=$2 fault=$3
+    if [ "${WORKER_PROOF_BASELINE:-0}" = 1 ]; then
+        source_text=$(git -C "$ROOT" show HEAD:tests/e2e-pithead.sh)
+    else
+        source_text=$(cat "$ROOT/tests/e2e-pithead.sh")
+    fi
+    eval "$(printf '%s\n' "$source_text" | sed '/^proof_preflight "/,$d; /^require_preflight "/,$d')"
+    CFG="$CASE_DIR/config" RIGFORGE=fixture_rigforge RIG_LOCK_HOLDER="$CASE_DIR/holder"
+    printf original >"$CFG"
+    state=initial
+    git() {
+        if [ "$fault" = operator-checkout ]; then
+            [ "$1" = -C ] && [ "$2" = "$HERE" ] && [ "$3" = -c ] && [ "$4" = "safe.directory=$HERE" ] || return 128
+        fi
+        case "$*" in
+        *'rev-parse --verify'*)
+            if [ "$fault" = stale ] || { [ "$fault" = changed-head ] && [ "$state" != initial ]; }; then
+                printf '%040d\n' 2
+            else
+                printf '%040d\n' 1
+            fi
+            ;;
+        *'diff --quiet'*) [ "$fault" != dirty ] ;;
+        *'ls-files --others'*)
+            [ "$fault" != git-failed ] || return 1
+            [ "$fault" != untracked ] || printf untracked
+            ;;
+        *) return 1 ;;
+        esac
+    }
+    uname() { case "$1" in -s) printf Linux ;; -m) printf x86_64 ;; *) return 1 ;; esac }
+    require_preflight() { :; }
+    replay_lock() { printf locked >"$CASE_DIR/locked"; }
+    systemctl() { return 1; }
+    refresh_profile_finish() { [ "$fault" != cleanup ]; }
+    fixture_rigforge() { [ "$1" = apply ]; }
+    passes() {
+        local n
+        for ((n = 0; n < $1; n++)); do ok assertion; done
+    }
+    phase_connect() {
+        phase connect
+        state=connected
+        printf mutated >"$CFG"
+        passes 3
+    }
+    phase_worker_api() {
+        [ "$state" = connected ]
+        phase worker-api
+        state=worker
+        passes 5
+    }
+    phase_api_impact() {
+        [ "$state" = worker ]
+        phase api-impact
+        state=loaded
+        case "$fault" in
+        fatal) return 7 ;;
+        assertion) bad assertion ;;
+        partial)
+            passes 2
+            return 0
+            ;;
+        duplicate) phase api-impact ;;
+        unexplained-skip)
+            skip unexplained
+            return 0
+            ;;
+        signal) kill -TERM "$$" ;;
+        esac
+        passes 3
+    }
+    phase_network() {
+        [ "$state" = loaded ]
+        [ "$fault" != missing ] || return 0
+        phase network
+        printf reached >"$CASE_DIR/network"
+        passes 9
+    }
+    phase_stratum_auth() {
+        phase stratum-auth
+        if [ "$fault" = allowed-skip ] || [ "$fault" = unapproved-skip ]; then
+            skip 'E2E_STRATUM_PASS not set (stack auth off, or secret not provided) — phases skipped'
+        else
+            passes 4
+        fi
+    }
+    phase_dashboard() {
+        phase dashboard
+        passes 2
+    }
+    phase_dev_fee() {
+        phase dev-fee
+        passes 1
+    }
+    phase_control() {
+        phase control
+        passes 4
+        (ok subprocess-assertion)
+    }
+    phase_access_tokens() {
+        phase access-tokens
+        passes 7
+    }
+    phase_control_replay() {
+        phase control-replay
+        phase_connect
+        passes 1
+    }
+    set -- "$mode"
+    eval "$(printf '%s\n' "$source_text" | awk '/^proof_preflight "|^require_preflight "/ {started=1} started')"
+    exit
+fi
+
+TASK_TMP=$(mktemp -d)
+trap 'rm -rf "$TASK_TMP"' EXIT
+context=$(jq -cn '{schema:1,head:("0"*39+"1"),ref:"refs/heads/proof-fixture",input:("a"*64),run:("b"*32),group:"network-sequence",row:"worker",platform:"linux-x86_64",allowed_skips:{}}')
+checks=0
+run_case() {
+    local mode=$1 fault=$2 expected=$3 rc=0 ctx=$context receipt
+    CASE_DIR="$TASK_TMP/$mode-$fault"
+    mkdir -p "$CASE_DIR"
+    ctx=$(printf '%s' "$ctx" | jq --arg group "$mode" '.group=$group')
+    if [ "$fault" = wrong-platform ]; then
+        ctx=$(printf '%s' "$ctx" | jq '.platform="linux-aarch64"')
+    fi
+    if [ "$fault" = allowed-skip ]; then
+        ctx=$(printf '%s' "$ctx" | jq '.allowed_skips={"stratum-auth":"stratum-fixture-absent"}')
+    fi
+    CASE_DIR="$CASE_DIR" E2E_PROOF_CONTEXT="$ctx" bash "$0" --case "$mode" "$fault" >"$CASE_DIR/output" 2>&1 || rc=$?
+    if [ "$rc" -ne "$expected" ]; then
+        printf 'FAIL: %s/%s returned %s, expected %s\n' "$mode" "$fault" "$rc" "$expected" >&2
+        cat "$CASE_DIR/output" >&2
+        exit 1
+    fi
+    case "$fault" in
+    stale | dirty | untracked | git-failed | wrong-platform)
+        [ ! -e "$CASE_DIR/locked" ]
+        [ "$(cat "$CASE_DIR/config")" = original ]
+        ;;
+    *)
+        [ "$(cat "$CASE_DIR/config")" = original ]
+        receipt=$(sed -n 's/^worker-proof: //p' "$CASE_DIR/output" | tail -n 1)
+        [ -n "$receipt" ] || {
+            printf 'FAIL: missing worker proof receipt\n' >&2
+            exit 1
+        }
+        printf '%s\n' "$receipt" >"$CASE_DIR/receipt"
+        printf '[%s]\n' "$ctx" >"$CASE_DIR/plan"
+        local check_rc=0
+        python3 "$ROOT/tests/worker-proof.py" check "$CASE_DIR/plan" "$CASE_DIR/receipt" >/dev/null 2>&1 || check_rc=$?
+        if [ "$expected" = 0 ]; then [ "$check_rc" = 0 ]; else [ "$check_rc" != 0 ]; fi
+        ;;
+    esac
+    case "$fault" in fatal | assertion | signal) [ ! -e "$CASE_DIR/network" ] ;; esac
+    checks=$((checks + 1))
+}
+
+run_case network-sequence success 0
+run_case network-sequence operator-checkout 0
+run_case all success 0
+run_case all allowed-skip 0
+run_case control-replay-thermal success 0
+run_case control-replay-pools success 0
+for fault in missing duplicate partial unexplained-skip cleanup changed-head assertion; do
+    run_case network-sequence "$fault" 1
+done
+run_case all unapproved-skip 1
+run_case network-sequence fatal 7
+run_case network-sequence signal 143
+for fault in stale dirty untracked git-failed wrong-platform; do run_case network-sequence "$fault" 2; done
+printf 'worker proof integration: %s passed\n' "$checks"
diff --git a/tests/worker-proof.py b/tests/worker-proof.py
new file mode 100644
index 0000000..e256160
--- /dev/null
+++ b/tests/worker-proof.py
@@ -0,0 +1,310 @@
+#!/usr/bin/env python3
+import json
+from pathlib import Path
+import re
+import sys
+
+
+CONTEXT_FIELDS = {
+    "schema",
+    "head",
+    "ref",
+    "input",
+    "run",
+    "group",
+    "row",
+    "platform",
+    "allowed_skips",
+}
+ALL_PHASES = [
+    "connect",
+    "worker-api",
+    "api-impact",
+    "network",
+    "stratum-auth",
+    "dashboard",
+    "dev-fee",
+    "control",
+    "access-tokens",
+]
+GROUPS = set(ALL_PHASES) | {
+    "all",
+    "network-sequence",
+    "control-replay-thermal",
+    "control-replay-pools",
+}
+PASS_COUNTS = {
+    "connect": 3,
+    "worker-api": 5,
+    "api-impact": 3,
+    "network": 9,
+    "stratum-auth": 4,
+    "dashboard": 2,
+    "dev-fee": 1,
+    "control": 5,
+    "access-tokens": 7,
+    "control-replay-thermal": 4,
+    "control-replay-pools": 4,
+}
+HEX40 = re.compile(r"[0-9a-f]{40}\Z")
+HEX64 = re.compile(r"[0-9a-f]{64}\Z")
+HEX32 = re.compile(r"[0-9a-f]{32}\Z")
+ROW = re.compile(r"[a-z][a-z0-9_-]{0,63}\Z")
+
+
+class Invalid(Exception):
+    pass
+
+
+def unique_object(pairs):
+    result = {}
+    for key, value in pairs:
+        if key in result:
+            raise Invalid
+        result[key] = value
+    return result
+
+
+def parse_json(text):
+    try:
+        return json.loads(
+            text,
+            object_pairs_hook=unique_object,
+            parse_constant=lambda value: (_ for _ in ()).throw(Invalid()),
+        )
+    except (Invalid, json.JSONDecodeError, RecursionError, UnicodeError):
+        raise Invalid from None
+
+
+def read_json(path):
+    try:
+        return parse_json(Path(path).read_text(encoding="utf-8"))
+    except (OSError, UnicodeError):
+        raise Invalid from None
+
+
+def exact_object(value, fields):
+    if not isinstance(value, dict) or set(value) != fields:
+        raise Invalid
+
+
+def valid_ref(value):
+    if not isinstance(value, str) or not value.startswith(("refs/heads/", "refs/tags/", "refs/pull/")):
+        return False
+    if any(ord(char) < 33 or ord(char) > 126 for char in value):
+        return False
+    if any(token in value for token in ("..", "//", "@{", "\\", "~", "^", ":", "?", "*", "[")):
+        return False
+    if value.endswith(("/", ".")):
+        return False
+    parts = value.split("/")
+    if any(not part or part.startswith(".") or part.endswith(".lock") for part in parts):
+        return False
+    if value.startswith("refs/pull/"):
+        return len(parts) == 4 and parts[2].isdigit() and parts[2] != "0" and parts[3] in {"head", "merge"}
+    return len(parts) >= 3
+
+
+def phase_ids(group):
+    if group == "all":
+        return ALL_PHASES
+    if group == "network-sequence":
+        return ALL_PHASES[:4]
+    return [group]
+
+
+def validate_context(value, expected_group=None):
+    exact_object(value, CONTEXT_FIELDS)
+    if type(value["schema"]) is not int or value["schema"] != 1:
+        raise Invalid
+    if not isinstance(value["head"], str) or not HEX40.fullmatch(value["head"]):
+        raise Invalid
+    if not valid_ref(value["ref"]):
+        raise Invalid
+    if not isinstance(value["input"], str) or not HEX64.fullmatch(value["input"]):
+        raise Invalid
+    if not isinstance(value["run"], str) or not HEX32.fullmatch(value["run"]):
+        raise Invalid
+    if not isinstance(value["group"], str) or value["group"] not in GROUPS or (
+        expected_group is not None and value["group"] != expected_group
+    ):
+        raise Invalid
+    if not isinstance(value["row"], str) or not ROW.fullmatch(value["row"]):
+        raise Invalid
+    if not isinstance(value["platform"], str) or value["platform"] not in {"linux-x86_64", "linux-aarch64"}:
+        raise Invalid
+    skips = value["allowed_skips"]
+    if not isinstance(skips, dict) or not all(isinstance(key, str) and isinstance(item, str) for key, item in skips.items()):
+        raise Invalid
+    allowed = {}
+    if value["group"] == "all":
+        allowed = {
+            "stratum-auth": "stratum-fixture-absent",
+            "dashboard": "dashboard-fixture-absent",
+        }
+    elif value["group"] == "dashboard":
+        allowed = {"dashboard": "dashboard-fixture-absent"}
+    if any(allowed.get(key) != item for key, item in skips.items()):
+        raise Invalid
+    return value
+
+
+def valid_counter(value):
+    return type(value) is int and value >= 0
+
+
+def validate_receipt(receipt, expected=None):
+    exact_object(receipt, {"schema", "context", "phases", "cleanup", "exit_code"})
+    if type(receipt["schema"]) is not int or receipt["schema"] != 1:
+        raise Invalid
+    context = validate_context(receipt["context"])
+    if expected is not None and context != expected:
+        raise Invalid
+    phases = receipt["phases"]
+    if not isinstance(phases, list) or len(phases) != len(phase_ids(context["group"])):
+        raise Invalid
+    if [row.get("id") if isinstance(row, dict) else None for row in phases] != phase_ids(context["group"]):
+        raise Invalid
+    for row in phases:
+        exact_object(row, {"id", "passed", "failed", "skipped"})
+        if not valid_counter(row["passed"]) or not valid_counter(row["failed"]):
+            raise Invalid
+        if not isinstance(row["skipped"], list) or not all(isinstance(item, str) for item in row["skipped"]):
+            raise Invalid
+        expected_skip = context["allowed_skips"].get(row["id"])
+        passed = row["passed"] == PASS_COUNTS[row["id"]] and row["failed"] == 0 and row["skipped"] == []
+        skipped = expected_skip is not None and row["passed"] == 0 and row["failed"] == 0 and row["skipped"] == [expected_skip]
+        if not (passed or skipped):
+            raise Invalid
+    if receipt["cleanup"] != "pass" or type(receipt["exit_code"]) is not int or receipt["exit_code"] != 0:
+        raise Invalid
+    return receipt
+
+
+def parse_events(path, context):
+    try:
+        lines = Path(path).read_text(encoding="utf-8").splitlines()
+    except (OSError, UnicodeError):
+        raise Invalid from None
+    phases = []
+    expected = phase_ids(context["group"])
+    current = None
+    for line in lines:
+        if not line:
+            raise Invalid
+        event = parse_json(line)
+        if not isinstance(event, dict) or "event" not in event:
+            raise Invalid
+        kind = event["event"]
+        if kind == "phase":
+            exact_object(event, {"event", "id"})
+            if len(phases) >= len(expected) or event["id"] != expected[len(phases)]:
+                raise Invalid
+            current = {"id": event["id"], "passed": 0, "failed": 0, "skipped": []}
+            phases.append(current)
+            continue
+        if current is None:
+            raise Invalid
+        if kind == "pass":
+            exact_object(event, {"event"})
+            current["passed"] += 1
+        elif kind == "fail":
+            exact_object(event, {"event"})
+            current["failed"] += 1
+        elif kind == "skip":
+            exact_object(event, {"event", "reason"})
+            reason = context["allowed_skips"].get(current["id"])
+            if event["reason"] != reason:
+                raise Invalid
+            current["skipped"].append(event["reason"])
+        else:
+            raise Invalid
+    return phases
+
+
+def command_context(args):
+    if len(args) != 1 or args[0] not in GROUPS:
+        raise Invalid
+    context = validate_context(parse_json(sys.stdin.read()), args[0])
+    print(json.dumps(context, sort_keys=True, separators=(",", ":")))
+    return 0
+
+
+def command_receipt(args):
+    if len(args) != 4 or args[2] not in {"pass", "fail", "unknown"} or not re.fullmatch(r"-?(0|[1-9][0-9]*)", args[3]):
+        raise Invalid
+    context = validate_context(read_json(args[0]))
+    cleanup = "pass" if args[2] == "pass" else "fail"
+    exit_code = int(args[3])
+    try:
+        phases = parse_events(args[1], context)
+    except Invalid:
+        receipt = {
+            "schema": 1,
+            "context": context,
+            "phases": [],
+            "cleanup": cleanup,
+            "exit_code": exit_code or 1,
+        }
+        print(json.dumps(receipt, sort_keys=True, separators=(",", ":")))
+        return 1
+    receipt = {
+        "schema": 1,
+        "context": context,
+        "phases": phases,
+        "cleanup": cleanup,
+        "exit_code": exit_code,
+    }
+    print(json.dumps(receipt, sort_keys=True, separators=(",", ":")))
+    try:
+        validate_receipt(receipt, context)
+    except Invalid:
+        return 1
+    return 0
+
+
+def command_check(args):
+    if len(args) < 2:
+        raise Invalid
+    plan = read_json(args[0])
+    if not isinstance(plan, list) or not plan:
+        raise Invalid
+    contexts = [validate_context(item) for item in plan]
+    binding = {key: contexts[0][key] for key in ("head", "ref", "input", "run")}
+    if any(any(item[key] != value for key, value in binding.items()) for item in contexts):
+        raise Invalid
+    if len({item["row"] for item in contexts}) != len(contexts):
+        raise Invalid
+    expected = {json.dumps(item, sort_keys=True, separators=(",", ":")): item for item in contexts}
+    if len(expected) != len(contexts) or len(args[1:]) != len(contexts):
+        raise Invalid
+    seen = set()
+    for path in args[1:]:
+        receipt = read_json(path)
+        exact_object(receipt, {"schema", "context", "phases", "cleanup", "exit_code"})
+        key = json.dumps(receipt["context"], sort_keys=True, separators=(",", ":"))
+        if key not in expected or key in seen:
+            raise Invalid
+        validate_receipt(receipt, expected[key])
+        seen.add(key)
+    if seen != set(expected):
+        raise Invalid
+    return 0
+
+
+def main():
+    if len(sys.argv) < 2:
+        raise Invalid
+    commands = {"context": command_context, "receipt": command_receipt, "check": command_check}
+    command = commands.get(sys.argv[1])
+    if command is None:
+        raise Invalid
+    return command(sys.argv[2:])
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except (Invalid, KeyError, TypeError, ValueError):
+        print("worker-proof: invalid input", file=sys.stderr)
+        sys.exit(2)