diff --git a/docs/FINISHER_COLLECTOR.md b/docs/FINISHER_COLLECTOR.md
new file mode 100644
index 0000000..5faf747
--- /dev/null
+++ b/docs/FINISHER_COLLECTOR.md
@@ -0,0 +1,251 @@
+# Finisher observation collector candidate
+
+This collector validates fresh, authenticated, independently recalled Finisher
+journeys through a separately admitted owner probe. Its only successful output is
+`UNSIGNED_CANDIDATE`, with `release_verdict: NOT_READY` and
+`signature_verified: false`. It does not sign, enroll a collector, install an
+attestation, create a worker grant, change the staging gate or deploy a target.
+
+The implementation uses the installed `generic-production-v1` contract from
+CertForge source `d8ad9fd881365801354b78a758632d2a31ab6828`. It reuses the canonical
+JSON hash and existing restricted-content checks. Existing trust verification
+rejects its unsigned payload. A passing test fixture is not live evidence.
+
+## Current missing prerequisites
+
+There is no owner-probe executable included in this candidate and no newly
+admitted collector or signer. The CLI fails with `owner_probe_unconfigured` when
+that dependency is absent. An implementation that returns configured booleans or
+replays a saved observation is not a valid owner probe.
+
+The independently admitted owner must supply and review the real adapter to the
+existing authentication, control, observation and receipt services. Its entire
+executable dependency closure must be immutable and hash pinned, and its runtime
+identity must have actual admission for this exact policy, target and operation.
+Hash pins provide integrity, not authorization. The implementer of this candidate
+cannot make themselves an independent collector, signer or reviewer by changing
+their role label. Existing Maximalist signatures and the Forge integrity
+bootstrap do not attest to Finisher.
+
+There are also concrete selected-runtime constraints. In transport source
+`1b79e2ee932927b3e408ff2ed43eb8d248875aaf`:
+
+* `SYSTEMS/finisher_cloud_integration/cloud_front.py` treats `start` as
+  reconciliation of an already running singleton. A stopped singleton requires
+  an independently admitted owner bootstrap operation; frontend `start` does not
+  provide one.
+* `SYSTEMS/finisher_cloud_integration/ui_runtime.py` returns a read-only
+  `echo.finisher.ui.snapshot.v1` for status. Accepted/applied are journal control
+  fields, not a status mutation receipt.
+* Native controller halt is `halted`, not `stopped`. The Cloudflare observer in
+  `SYSTEMS/cf_terminal/cloud/container_control.py` currently requires a live
+  process for `readback_verified`. It cannot prove terminal halt after that
+  process exits. The UI projection also omits the controller's `halted` field.
+  The owner probe must not invent terminal proof; that source/runtime dependency
+  must be repaired, independently reviewed and actually observed first.
+* Worker-image rollback alone does not restore a compatible checkpoint. The
+  runtime rejects a checkpoint whose source identity differs. A separately
+  authorized compatible checkpoint/config restoration and exact readback are
+  prerequisites. This collector does not provide a reset or data-restore route.
+
+These constraints remain holds. They are not requests to weaken the gate or to
+reuse another product's trust identity.
+
+## Invocation and private configuration
+
+Run on an admitted Linux observer host using its normal opaque scoped broker
+environment. Do not copy another client's auth state or put credentials in the
+policy, command arguments, receipt or output.
+
+```sh
+python scripts/collect_finisher_e2e.py \
+  --policy /ABSOLUTE/PRIVATE/owner-policy.json \
+  --policy-sha256 ACTUAL_POLICY_FILE_SHA256 \
+  --output /ABSOLUTE/PRIVATE/new-unsigned-observations.json
+```
+
+The policy file must be regular, at most 64 KiB, owned by root or the observer
+identity, and not writable by group/other. Its exact bytes must match the pin.
+The output parent must be a canonical private directory owned by the observer;
+output is created exclusively with mode 0600. No existing output is overwritten.
+Exit 0 means collection succeeded **unsigned**; exit 2 is a verified hold; exit 3
+is invalid input or I/O failure. Fixed reason codes do not include raw provider
+responses or credentials. Structured logs contain only operation names, receipt
+identifiers and hashes.
+
+Policy fields:
+
+| Field | Required value |
+| --- | --- |
+| `schema` | `echo.finisher.collector-policy.v1` |
+| `tenant_id`, `collector_id`, `authority_id`, `execution_id`, `admission_id` | Actual distinct scoped identities; bounded identifiers |
+| `bindings.target_identity_digest`, `environment_identity_digest` | Exact canonical CertForge identity hashes |
+| `bindings.source_commit` | Exact native or transport source being certified |
+| `bindings.native_source_commit`, `transport_source_commit` | Full 40-character immutable source commits |
+| `bindings.artifact_sha256`, `native_artifact_sha256` | Actual artifact byte hashes, not a tag or source SHA |
+| `bindings.image_digest` | Actual `sha256:` OCI index/image digest, distinct from archive byte hash |
+| `bindings.account_id`, `application_id`, `durable_object_id` | Complete actual Cloudflare tuple; 32 hex account and 64 hex DO IDs |
+| `bindings.installation_id`, `owner_id`, `cloudflare_origin` | Exact installation, admitted owner and HTTPS origin without userinfo/query |
+| `rollback_artifact_sha256`, `rollback_source_commit` | Actual distinct compatible prior artifact/source |
+| `rollback_checkpoint_sha256`, `rollback_configuration_sha256` | Owner-approved immutable prior checkpoint/configuration digest pins |
+| `recovered_checkpoint_sha256`, `recovered_configuration_sha256` | Owner-approved recovered candidate checkpoint/configuration digest pins |
+| `providers` | Exactly claude/codex/gemini/grok, each with exact `model` and nonpersonal `account_sha256` |
+| `probe.command` | Exactly `[ABSOLUTE_INTERPRETER, ABSOLUTE_REVIEWED_PROBE_SCRIPT]` |
+| `probe.sha256` | Exact SHA256 pins for those two files in the same order |
+| `probe.dependencies` | Absolute dependency paths mapped to immutable SHA256 pins |
+| `probe.timeout` | Whole process deadline from 1 to 120 seconds, default 120 |
+
+The four checkpoint/configuration pins must come from the owner's reviewed,
+immutable prior-baseline and recovered-candidate manifests **before** collection,
+not from a candidate's self-reported readback. Mutable runtime checkpoints must
+be deliberately frozen or version-addressed for meaningful exact comparison.
+The owner admission verifies these immutable policy values as part of the
+canonical policy hash; arbitrary well-formed digests are not rollback evidence.
+
+The transport runs the interpreter in isolated mode, accepts no shell text or
+provider flags, bounds stdin to 8192 bytes and stdout to 128 KiB, discards stderr,
+and checks executable pins before and after each operation. Nonblocking stdin
+and stdout share the same deadline, including a child that never reads input.
+The process owns its group and reaps it on every exit. Environment inheritance is
+limited to normal PATH/HOME/locale/TLS paths and opaque SOL broker/run/role/state
+variables, including `SOL_STATE_DB`. No API billing or custom provider route
+fallback is added. Owner admission must make pinned files/closure immutable to
+untrusted concurrent writers; rehashing is not a sandbox or an administrative
+grant.
+
+## Public owner-probe protocol
+
+Each invocation receives one JSON object on stdin and returns one JSON object on
+stdout. The owner adapter performs real authenticated requests; it must not load
+responses from fixtures, accept a caller-supplied `verified: true`, mint a grant,
+or turn a blocked command into an indirect interpreter/SSH/container execution.
+
+Request schema `echo.finisher.owner-query.v1` contains:
+
+* `operation`: `admission`, `observe`, or `recall` only.
+* A new unpredictable `nonce`, plus canonical `policy_sha256`.
+* Exact `admission_id`, `execution_id`, `tenant_id`, `collector_id`, and complete
+  `bindings` as above.
+* For `observe`: collection `challenge`, finite `kind` and `name` below.
+* For `recall`: exact `receipt_id` returned by the previous actual observation.
+
+Response schema `echo.finisher.owner-response.v1` contains the identical nonce,
+`request_sha256` over canonical full request, actual `authority_id`, independently
+established `authenticated: true` and `authorized: true`, a fresh timezone-aware
+`observed_at`, and `result`. These fields are trusted only because the owner has
+independently admitted and pinned the real adapter and its authenticated service
+routes. This candidate is a verifier/transport, not that authority service.
+
+For `admission`, result must contain `allowed: true`, `independent: true`, exact
+policy hash, all request identities/bindings, an expiry in the next hour,
+`operations: ["observe", "recall"]`, and `restoration_allowed: true`. The actual
+admission must cover the complete finite journey including its controls,
+negative checks, prior snapshot/config rollback, owner restoration, and all
+24 separately recalled `stability-front/<index>:<front>` route observations.
+The collector rechecks admission before every observe and recall, and at
+completion. An aggregate health result is not authority for a named route.
+
+For `observe`, result contains a new actual `receipt_id`. The collector then
+requests `recall`; recall result contains the identical ID, exact `content`, and
+canonical `content_sha256`. Content schema `echo.finisher.observation.v1` contains
+the receipt ID, `mode: live`, exact collection challenge/kind/name, identities and
+bindings, a fresh `observed_at`, and the journey result. No receipt or provider
+invocation may be reused. The complete collection is bounded to 30 minutes.
+
+The receipt service must preserve actual authenticated provenance and permit
+exact readback. Merely echoing the just-submitted request or returning an
+unverifiable success flag does not satisfy this contract. The separately
+admitted signer must independently verify the receipt manifest before signing.
+
+## Finite journey and result contract
+
+The ordered fronts are `auto`, `cauto`, `gauto`, `grauto`, `CFauto`, `CFcauto`,
+`CFgauto`, `CFgrauto`, mapped to claude/codex/gemini/grok respectively.
+
+1. `front/<front>` performs the authorized real front journey. Its result binds
+   actual provider/model/account, run/worker/session and native invocation IDs;
+   observed model/subscription identity; exact resumed session; real scoped MCP,
+   hook capture and exact capture readback hashes; live heartbeat; uncertain
+   replay suppression; and process cleanup. It requires zero inference caused
+   by readonly status. `start` has the actual `singleton_observed` shape with
+   `already_running`, verified readback and exact target. `status_snapshot` is
+   the real UI snapshot, with no top-level accepted/applied/request ID. It must
+   be fresh, online, installation/resource/heartbeat/service verified, and bound
+   to the actual full Cloudflare target and transport source. `controls` contains
+   exactly pause/resume/halt, each with distinct actual `request_id` and
+   `native_request_id`, matching `action`, exact full `target` including run,
+   `submitted_at`, accepted/applied proof and paused/running/halted `state`.
+   Each `readback` must independently match that target and transport source,
+   have fresh `observed_at >= submitted_at`, verified installation/readback,
+   and `controller.last_control_id` equal to the exact native request ID.
+   The actual paused/halted flags must agree with the action. Pause/resume
+   require active service and fresh heartbeat. Halt requires a fresh independently
+   established `terminal_halt_verified: true`, `lifecycle: halted` and
+   `online: false` observation after process exit without pretending that the
+   exited controller still has a live heartbeat. A live controller setting its
+   halted flag before cleanup is insufficient. A stale
+   observation, replayed native ID, other run/installation, accepted write,
+   pending journal or unavailable terminal readback must fail.
+2. Immediately after **each** halt, `restoration/<front>` requires
+   `method: admitted_owner_operation`, the same admission ID, previous run ID,
+   a new actual owner operation ID, and a fresh healthy exact-target snapshot.
+   This is a normal independently authorized owner bootstrap, never a synthetic
+   frontend start receipt. Without an actual admitted restoration route the
+   collection stops; it does not continue to the next front in a stopped state.
+3. `negative/<name>` checks unauthenticated, wrong_owner, wrong_installation,
+   wrong_source, invalid_token, expired_token, stale_control, replayed_control,
+   out_of_scope, redirect and forbidden_native_route. Each result must prove
+   denial, no side effects, and identical bounded before/after state hashes.
+   Credentials and personal identity never enter the receipt. A redirect check
+   must refuse rather than follow an unadmitted destination.
+4. `rollback/restore-and-recover` requires actual prior and recovered artifact
+   and source readbacks, both checkpoint source matches, and verified restored
+   and recovered state. All four actual checkpoint/configuration SHA256 readbacks
+   must **equal their separately owner-approved immutable policy pins**. A digest
+   of the right length, a truthful-looking boolean, or only a Worker image
+   replacement is insufficient. It runs **before** final stability.
+5. `stability/0`, `/1`, `/2` establish aggregate health, resource admission,
+   fresh heartbeat and zero unexpected restarts only. Legacy aggregate
+   `snapshots` are rejected; they cannot count as eight-front evidence.
+   Each aggregate probe is followed by **eight separate** owner-issued,
+   independently recalled `stability-front/<index>:<front>` receipts, each
+   bound to its exact challenge/kind/name, front and expected provider. Each
+   route result requires `route_readback_verified: true`, a fresh healthy
+   `snapshot`, and `snapshot_sha256` equal to the canonical digest of that
+   snapshot. Reusing another route's receipt/result fails named-route binding.
+   All routes may legitimately share the single Cloudflare singleton `run_id`;
+   route receipt provenance, not distinct run IDs, proves eight-front coverage.
+   The first-to-third aggregate observation interval must be at least 30 seconds.
+6. `external/acceptance` requires a distinct independent reviewer, actual Access
+   session, immutable UI assets, exact HTTPS origin and accepted live behavior.
+
+Tests use explicitly synthetic authority fixtures to exercise forged, replayed,
+stale, mismatched, revoked and incomplete receipts. They also exercise real local
+child processes on FORGE for small-pipe/nonreading deadlines, output bounds and
+opaque broker state inheritance. None of these tests call a real provider or
+produce admission, signing, deployment or production-acceptance evidence.
+
+## Unchanged signing and deployment boundaries
+
+Only after real observations and independent review may an already admitted
+independent signer add its actual key identity and produce the existing signed
+production-E2E envelope. The normal trusted-key verifier must accept that exact
+envelope and its source/target/environment/manifest bindings before re-running
+certification. Private signing keys never belong in this collector or its worker.
+
+The existing deployment admission hook remains unchanged:
+
+```sh
+python /home/forge/echo-certification-forge-current/scripts/deployment_admission_hook.py \
+  admit --forge-url ADMITTED_FORGE_ORIGIN --tenant ACTUAL_TENANT \
+  --artifact-digest ACTUAL_ARTIFACT_BYTE_SHA256 --environment staging \
+  --environment-identity-digest ACTUAL_ENVIRONMENT_IDENTITY_DIGEST \
+  --rule-manifest-digest ACTUAL_RULE_MANIFEST_DIGEST \
+  --deployment-id ACTUAL_DEPLOYMENT_ID --requested-by ACTUAL_ADMITTED_OWNER
+```
+
+Any credential required by that existing hook stays opaque in its normal owner
+environment; do not paste values into this command or handoff. Staging and
+production gates remain enforced. Actual outcome registration and exact readback
+remain owner responsibilities after an admitted deployment. This collector's
+source, tests, draft PR or unsigned output cannot clear those gates.
diff --git a/scripts/collect_finisher_e2e.py b/scripts/collect_finisher_e2e.py
new file mode 100644
index 0000000..c82e955
--- /dev/null
+++ b/scripts/collect_finisher_e2e.py
@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+"""Run the owner-admitted Finisher collector; output is never a signed verdict."""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import logging
+import os
+from pathlib import Path
+import stat
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+from echo_certification_forge.finisher_collector import (  # noqa: E402
+    CollectorError, FinisherCollector, PinnedOwnerProbe, bounded_document, require,
+)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--policy", type=Path, required=True)
+    parser.add_argument("--policy-sha256", required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    logging.basicConfig(level=logging.INFO, format="%(message)s")
+    try:
+        require(os.name == "posix", "collector_platform_unsupported")
+        require(args.policy.is_absolute() and not args.policy.is_symlink(), "policy_path_invalid")
+        descriptor = os.open(args.policy, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+        with os.fdopen(descriptor, "rb") as handle:
+            metadata = os.fstat(handle.fileno())
+            require(stat.S_ISREG(metadata.st_mode) and metadata.st_size <= 65536, "policy_file_invalid")
+            require(metadata.st_uid in (0, os.geteuid()) and not metadata.st_mode & 0o022, "policy_file_permissions")
+            raw = handle.read(65537)
+        require(hashlib.sha256(raw).hexdigest() == args.policy_sha256, "policy_pin_mismatch")
+        policy = bounded_document(raw, 65536)
+        probe = policy.get("probe")
+        require(isinstance(probe, dict), "owner_probe_unconfigured")
+        require(args.output.is_absolute() and not args.output.exists() and not args.output.is_symlink(), "output_path_invalid")
+        parent = args.output.parent
+        require(parent.resolve(strict=True) == parent and parent.is_dir(), "output_parent_invalid")
+        metadata = parent.stat()
+        require(metadata.st_uid == os.geteuid() and not metadata.st_mode & 0o077, "output_parent_not_private")
+        owner = PinnedOwnerProbe(probe.get("command"), probe.get("sha256"),
+                                 dependencies=probe.get("dependencies"), timeout=probe.get("timeout", 120))
+        result = FinisherCollector(policy, owner).collect()
+        data = (json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n").encode()
+        descriptor = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+        with os.fdopen(descriptor, "wb") as handle:
+            handle.write(data)
+            handle.flush()
+            os.fsync(handle.fileno())
+        print(json.dumps({"status": "UNSIGNED_CANDIDATE", "release_verdict": "NOT_READY",
+                          "signature_verified": False, "sha256": hashlib.sha256(data).hexdigest(),
+                          "observations": len(result["evidence"])}))
+        return 0
+    except CollectorError as exc:
+        print(json.dumps({"status": "HELD", "release_verdict": "NOT_READY", "reason": str(exc)}))
+        return 2
+    except (OSError, ValueError, TypeError, KeyError):
+        print(json.dumps({"status": "HELD", "release_verdict": "NOT_READY", "reason": "collector_input_or_io_failed"}))
+        return 3
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/src/echo_certification_forge/finisher_collector.py b/src/echo_certification_forge/finisher_collector.py
new file mode 100644
index 0000000..90cbf90
--- /dev/null
+++ b/src/echo_certification_forge/finisher_collector.py
@@ -0,0 +1,482 @@
+"""Finisher observation collection through a separately admitted owner probe.
+
+This module never signs, deploys, enrolls a worker or changes a release policy.
+The owner must independently review/admit the probe and pin its entire closure.
+Successful collection produces an UNSIGNED_CANDIDATE, always NOT_READY.
+"""
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from datetime import UTC, datetime, timedelta
+import hashlib
+import json
+import logging
+import os
+from pathlib import Path
+import re
+import selectors
+import signal
+import subprocess
+import time
+from typing import Any, Protocol
+from urllib.parse import urlsplit
+import uuid
+
+from .canonical import sha256_json
+from .production_e2e import BASE_CHECKS, GENERIC_PROFILE, SCHEMA_VERSION, _contains_restricted_value
+
+FRONTS = {"auto": "claude", "cauto": "codex", "gauto": "gemini", "grauto": "grok",
+          "CFauto": "claude", "CFcauto": "codex", "CFgauto": "gemini", "CFgrauto": "grok"}
+NEGATIVE_CONTROLS = ("unauthenticated", "wrong_owner", "wrong_installation", "wrong_source",
+                     "invalid_token", "expired_token", "stale_control", "replayed_control",
+                     "out_of_scope", "redirect", "forbidden_native_route")
+CONTROL_STATES = {"pause": "paused", "resume": "running", "halt": "halted"}
+IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\Z")
+DIGEST = re.compile(r"[0-9a-f]{64}\Z")
+COMMIT = re.compile(r"[0-9a-f]{40}\Z")
+LOG = logging.getLogger(__name__)
+
+
+class CollectorError(RuntimeError):
+    """A fixed nonsecret reason code; raw probe responses are never interpolated."""
+
+
+class OwnerProbe(Protocol):
+    def call(self, request: dict[str, Any]) -> dict[str, Any]: ...
+
+
+def require(condition: bool, reason: str) -> None:
+    if not condition:
+        raise CollectorError(reason)
+
+
+def identifier(value: Any) -> bool:
+    return isinstance(value, str) and IDENTIFIER.fullmatch(value) is not None
+
+
+def digest(value: Any) -> bool:
+    return isinstance(value, str) and DIGEST.fullmatch(value) is not None
+
+
+def timestamp(value: Any) -> datetime:
+    require(isinstance(value, str) and len(value) <= 64, "time_invalid")
+    try:
+        result = datetime.fromisoformat(value.replace("Z", "+00:00"))
+        require(result.tzinfo is not None, "time_invalid")
+        return result.astimezone(UTC)
+    except (ValueError, OverflowError) as exc:
+        raise CollectorError("time_invalid") from exc
+
+
+def bounded_document(data: bytes, limit: int = 131072) -> dict[str, Any]:
+    require(len(data) <= limit, "document_size")
+    def pairs(items):
+        result = {}
+        for key, value in items:
+            require(key not in result, "document_duplicate_key")
+            result[key] = value
+        return result
+    try:
+        result = json.loads(data, object_pairs_hook=pairs,
+                            parse_constant=lambda _: (_ for _ in ()).throw(CollectorError("document_number")))
+    except (ValueError, RecursionError, UnicodeError) as exc:
+        raise CollectorError("document_invalid") from exc
+    require(isinstance(result, dict), "document_invalid")
+    try:
+        require(not _contains_restricted_value(result), "restricted_content")
+    except RecursionError as exc:
+        raise CollectorError("document_depth") from exc
+    return result
+
+
+class PinnedOwnerProbe:
+    """Bounded Linux process transport for a previously reviewed owner probe.
+
+    Hashes are integrity pins, NOT administrative admission. The probe must query
+    the real authenticated control plane for admission and every observation.
+    No command, shell text, interpreter switch or provider argument is accepted
+    from an observation. Dependencies must be pinned by the owner as well.
+    """
+    def __init__(self, command: list[str], pins: list[str], *,
+                 dependencies: Mapping[str, str] | None = None, timeout: float = 120):
+        require(isinstance(command, list) and len(command) == 2 and isinstance(pins, list) and len(pins) == 2,
+                "probe_command_invalid")
+        require(type(timeout) in (int, float) and 1 <= timeout <= 120, "probe_timeout_invalid")
+        try:
+            self.command = [str(Path(item).resolve(strict=True)) for item in command]
+        except (OSError, ValueError, TypeError) as exc:
+            raise CollectorError("probe_command_invalid") from exc
+        self.pins = dict(zip(self.command, pins, strict=True))
+        for path, pin in (dependencies or {}).items():
+            resolved = str(Path(path).resolve(strict=True))
+            require(resolved not in self.pins, "probe_pin_duplicate")
+            self.pins[resolved] = pin
+        self.timeout = timeout
+        self._verify()
+
+    def _verify(self) -> None:
+        for name, expected in self.pins.items():
+            require(digest(expected), "probe_pin_invalid")
+            path = Path(name)
+            require(path.is_file() and not path.is_symlink(), "probe_pin_file")
+            value = hashlib.sha256()
+            with path.open("rb") as handle:
+                for block in iter(lambda: handle.read(65536), b""):
+                    value.update(block)
+            require(value.hexdigest() == expected, "probe_pin_mismatch")
+
+    def call(self, request: dict[str, Any]) -> dict[str, Any]:
+        self._verify()
+        require(os.name == "posix", "probe_platform_unsupported")
+        data = json.dumps(request, sort_keys=True, separators=(",", ":")).encode()
+        require(len(data) <= 8192, "probe_request_size")
+        # Credentials remain opaque in the inherited normal scoped broker path.
+        allowed = {"PATH", "HOME", "LANG", "LC_ALL", "SSL_CERT_FILE", "SSL_CERT_DIR",
+                   "SOL_BROKER_TOKEN", "SOL_RUN_ID", "SOL_STATE", "SOL_STATE_DB", "SOL_ROLE"}
+        environment = {name: value for name, value in os.environ.items() if name in allowed}
+        environment["PYTHONDONTWRITEBYTECODE"] = "1"
+        deadline = time.monotonic() + self.timeout
+        process = subprocess.Popen([self.command[0], "-I", self.command[1]], stdin=subprocess.PIPE,
+                                   stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
+                                   env=environment, start_new_session=True, close_fds=True, bufsize=0)
+        selector = selectors.DefaultSelector()
+        output = bytearray()
+        try:
+            # Both directions share one deadline, including a child that never
+            # reads stdin. No assumption about the actual pipe capacity is made.
+            os.set_blocking(process.stdin.fileno(), False)
+            os.set_blocking(process.stdout.fileno(), False)
+            selector.register(process.stdin, selectors.EVENT_WRITE)
+            selector.register(process.stdout, selectors.EVENT_READ)
+            sent = 0
+            while selector.get_map():
+                require(time.monotonic() < deadline, "probe_timeout")
+                for key, _ in selector.select(min(.1, max(0, deadline - time.monotonic()))):
+                    try:
+                        if key.fileobj is process.stdin:
+                            sent += os.write(process.stdin.fileno(), data[sent:sent + 4096])
+                            if sent == len(data):
+                                selector.unregister(process.stdin)
+                                process.stdin.close()
+                            continue
+                        block = os.read(key.fileobj.fileno(), 16384)
+                    except BlockingIOError:
+                        continue
+                    if not block:
+                        selector.unregister(key.fileobj)
+                        continue
+                    output.extend(block)
+                    require(len(output) <= 131072, "probe_output_size")
+            try:
+                process.wait(timeout=max(.01, deadline - time.monotonic()))
+            except subprocess.TimeoutExpired as exc:
+                raise CollectorError("probe_timeout") from exc
+            require(process.returncode == 0, "probe_failed")
+            self._verify()
+            return bounded_document(bytes(output))
+        except (OSError, subprocess.SubprocessError) as exc:
+            raise CollectorError("probe_transport_failed") from exc
+        finally:
+            selector.close()
+            # The probe owns this process group, including any unsettled children.
+            try:
+                os.killpg(process.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+            process.wait(timeout=5)
+            process.stdout.close()
+            process.stdin.close()
+
+
+class FinisherCollector:
+    def __init__(self, policy: dict[str, Any], probe: OwnerProbe, *,
+                 clock: Callable[[], datetime] = lambda: datetime.now(UTC)):
+        self.policy = bounded_document(json.dumps(policy, allow_nan=False).encode(), 65536)
+        self.probe, self.clock = probe, clock
+        p = self.policy
+        require(p.get("schema") == "echo.finisher.collector-policy.v1", "policy_schema")
+        require(all(identifier(p.get(key)) for key in
+                    ("tenant_id", "collector_id", "authority_id", "execution_id", "admission_id")), "policy_identity")
+        bindings = p.get("bindings")
+        require(isinstance(bindings, dict), "policy_bindings")
+        expected_keys = {"target_identity_digest", "environment_identity_digest", "source_commit",
+                         "native_source_commit", "transport_source_commit", "artifact_sha256",
+                         "native_artifact_sha256", "image_digest", "installation_id", "owner_id", "cloudflare_origin",
+                         "account_id", "application_id", "durable_object_id"}
+        require(set(bindings) == expected_keys, "policy_bindings")
+        require(all(digest(bindings[key]) for key in
+                    ("target_identity_digest", "environment_identity_digest", "artifact_sha256", "native_artifact_sha256")), "policy_digest")
+        require(all(isinstance(bindings[key], str) and COMMIT.fullmatch(bindings[key]) for key in
+                    ("source_commit", "native_source_commit", "transport_source_commit")), "policy_source")
+        require(bindings["source_commit"] in (bindings["native_source_commit"], bindings["transport_source_commit"]), "policy_source")
+        require(isinstance(bindings["image_digest"], str) and bindings["image_digest"].startswith("sha256:")
+                and digest(bindings["image_digest"][7:]), "policy_image")
+        require(identifier(bindings["installation_id"]) and identifier(bindings["owner_id"]), "policy_installation")
+        require(isinstance(bindings["account_id"], str) and re.fullmatch("[a-f0-9]{32}", bindings["account_id"])
+                and digest(bindings["durable_object_id"]) and isinstance(bindings["application_id"], str)
+                and re.fullmatch("[A-Za-z0-9_-]{1,128}", bindings["application_id"]), "policy_cloudflare_identity")
+        origin = urlsplit(bindings["cloudflare_origin"])
+        require(origin.scheme == "https" and bool(origin.hostname) and not origin.username and not origin.password
+                and origin.path in ("", "/") and not origin.query and not origin.fragment, "policy_origin")
+        require(digest(p.get("rollback_artifact_sha256")) and p["rollback_artifact_sha256"] != bindings["artifact_sha256"], "policy_rollback")
+        require(isinstance(p.get("rollback_source_commit"), str)
+                and COMMIT.fullmatch(p["rollback_source_commit"])
+                and p["rollback_source_commit"] != bindings["source_commit"], "policy_rollback")
+        # The approved immutable baseline and recovered candidate must pin their
+        # actual checkpoint/config identities before any rollback evidence is read.
+        require(all(digest(p.get(key)) for key in
+                    ("rollback_checkpoint_sha256", "rollback_configuration_sha256",
+                     "recovered_checkpoint_sha256", "recovered_configuration_sha256")), "policy_rollback_identity")
+        require(isinstance(p.get("providers"), dict) and set(p["providers"]) == set(FRONTS.values()), "policy_providers")
+        for provider in p["providers"].values():
+            require(isinstance(provider, dict) and set(provider) == {"model", "account_sha256"}
+                    and identifier(provider["model"]) and digest(provider["account_sha256"]), "policy_provider")
+        self.policy_hash = sha256_json(p)
+        self.challenge = uuid.uuid4().hex
+        self.started = self.clock()
+        require(self.started.tzinfo is not None, "clock_invalid")
+        self.ids: set[str] = set()
+        self.invocations: set[str] = set()
+        self.control_ids: set[str] = set()
+        self.native_control_ids: set[str] = set()
+        self.restoration_ids: set[str] = set()
+        self.evidence: list[dict[str, str]] = []
+
+    def _query(self, operation: str, **extra: Any) -> dict[str, Any]:
+        now = self.clock()
+        require(0 <= (now - self.started).total_seconds() <= 1800, "collection_deadline")
+        request = {"schema": "echo.finisher.owner-query.v1", "operation": operation,
+                   "nonce": uuid.uuid4().hex, "policy_sha256": self.policy_hash,
+                   **{key: self.policy[key] for key in
+                      ("admission_id", "execution_id", "tenant_id", "collector_id", "bindings")}, **extra}
+        LOG.info(json.dumps({"event": "owner_query", "operation": operation, "policy_sha256": self.policy_hash}))
+        response = self.probe.call(request)
+        require(isinstance(response, dict), "owner_response_invalid")
+        response = bounded_document(json.dumps(response, allow_nan=False).encode())
+        require(response.get("schema") == "echo.finisher.owner-response.v1"
+                and response.get("nonce") == request["nonce"]
+                and response.get("request_sha256") == sha256_json(request)
+                and response.get("authority_id") == self.policy["authority_id"]
+                and response.get("authenticated") is True and response.get("authorized") is True,
+                "owner_response_binding")
+        observed = timestamp(response.get("observed_at"))
+        require(now - timedelta(seconds=1) <= observed <= self.clock() + timedelta(seconds=1), "owner_response_time")
+        require(isinstance(response.get("result"), dict), "owner_response_result")
+        return response["result"]
+
+    def _admission(self) -> None:
+        result = self._query("admission")
+        require(result.get("allowed") is True and result.get("independent") is True, "admission_denied")
+        require(all(result.get(key) == self.policy[key] for key in
+                    ("admission_id", "collector_id", "execution_id", "tenant_id", "bindings")), "admission_binding")
+        require(result.get("policy_sha256") == self.policy_hash and result.get("restoration_allowed") is True,
+                "admission_journey_not_allowed")
+        expiry = timestamp(result.get("expires_at"))
+        require(self.clock() < expiry <= self.clock() + timedelta(hours=1), "admission_expired")
+        require(result.get("operations") == ["observe", "recall"], "admission_operations")
+
+    def _observe(self, kind: str, name: str) -> tuple[dict[str, Any], datetime]:
+        self._admission()
+        produced = self._query("observe", challenge=self.challenge, kind=kind, name=name)
+        receipt_id = produced.get("receipt_id")
+        require(identifier(receipt_id) and receipt_id not in self.ids, "receipt_replay")
+        self.ids.add(receipt_id)
+        self._admission()
+        recalled = self._query("recall", receipt_id=receipt_id)
+        content = recalled.get("content")
+        require(recalled.get("receipt_id") == receipt_id and isinstance(content, dict), "receipt_readback")
+        require(digest(recalled.get("content_sha256")) and recalled["content_sha256"] == sha256_json(content), "receipt_hash")
+        require(content.get("schema") == "echo.finisher.observation.v1"
+                and content.get("receipt_id") == receipt_id and content.get("mode") == "live"
+                and content.get("challenge") == self.challenge and content.get("kind") == kind
+                and content.get("name") == name, "observation_binding")
+        require(all(content.get(key) == self.policy[key] for key in
+                    ("tenant_id", "execution_id", "collector_id", "bindings")), "observation_identity")
+        observed = timestamp(content.get("observed_at"))
+        require(self.started <= observed <= self.clock() and (self.clock() - observed).total_seconds() <= 900, "observation_time")
+        result = content.get("result")
+        require(isinstance(result, dict), "observation_result")
+        self.evidence.append({"kind": kind, "name": name, "receipt_id": receipt_id,
+                              "sha256": recalled["content_sha256"], "observed_at": observed.isoformat()})
+        LOG.info(json.dumps({"event": "observation_recalled", "kind": kind, "name": name,
+                             "receipt_id": receipt_id, "sha256": recalled["content_sha256"]}))
+        return result, observed
+
+    def _target(self, target: Any, run_id: str | None = None) -> None:
+        binding = self.policy["bindings"]
+        expected = {"provider": "cloudflare", **{key: binding[key] for key in
+                     ("account_id", "application_id", "durable_object_id", "installation_id")},
+                    "source_sha": binding["transport_source_commit"]}
+        require(isinstance(target, dict) and set(target) == {*expected, "run_id"}
+                and all(target.get(key) == value for key, value in expected.items())
+                and identifier(target.get("run_id")) and (run_id is None or target["run_id"] == run_id),
+                "snapshot_target")
+
+    def _observation_time(self, stamp: Any) -> datetime:
+        if type(stamp) in (int, float):
+            require(0 <= stamp <= 253402300799, "status_snapshot_time")
+            stamp = datetime.fromtimestamp(stamp, UTC)
+        else:
+            stamp = timestamp(stamp)
+        require(self.started <= stamp <= self.clock() and (self.clock() - stamp).total_seconds() <= 30,
+                "status_snapshot_time")
+        return stamp
+
+    def _snapshot(self, value: Any, run_id: str | None = None) -> None:
+        require(isinstance(value, dict) and value.get("schema") == "echo.finisher.ui.snapshot.v1"
+                and not any(key in value for key in ("accepted", "applied", "request_id")), "status_snapshot_invalid")
+        self._target(value.get("target"), run_id)
+        require(value.get("source_sha") == self.policy["bindings"]["transport_source_commit"]
+                and value.get("online") is True and value.get("readback_verified") is True
+                and value.get("lifecycle") in ("online", "idle", "running", "executing"), "status_snapshot_unhealthy")
+        checks = value.get("checks")
+        require(isinstance(checks, dict) and all(checks.get(key) is True for key in
+                ("service_active", "installation_verified", "heartbeat_fresh", "resource_admitted")), "status_snapshot_unhealthy")
+        self._observation_time(value.get("observed_at"))
+
+    def _control(self, action: str, control: Any, run_id: str) -> None:
+        require(isinstance(control, dict) and identifier(control.get("request_id"))
+                and identifier(control.get("native_request_id")) and control.get("action") == action,
+                "front_control_invalid")
+        require(control["request_id"] not in self.control_ids
+                and control["native_request_id"] not in self.native_control_ids, "control_replay")
+        self.control_ids.add(control["request_id"])
+        self.native_control_ids.add(control["native_request_id"])
+        self._target(control.get("target"), run_id)
+        require(control.get("accepted") is True and control.get("applied") is True
+                and control.get("state") == CONTROL_STATES[action], "front_control_not_applied")
+        submitted = self._observation_time(control.get("submitted_at"))
+        observed = control.get("readback")
+        require(isinstance(observed, dict) and observed.get("target") == control["target"]
+                and observed.get("source_sha") == self.policy["bindings"]["transport_source_commit"]
+                and observed.get("readback_verified") is True, "control_readback_binding")
+        require(self._observation_time(observed.get("observed_at")) >= submitted, "control_readback_stale")
+        checks, controller = observed.get("checks"), observed.get("controller")
+        require(isinstance(checks, dict) and checks.get("installation_verified") is True
+                and isinstance(controller, dict) and controller.get("last_control_id") == control["native_request_id"],
+                "control_readback_identity")
+        require(type(controller.get("paused")) is bool and type(controller.get("halted")) is bool,
+                "control_readback_state")
+        expected = {"pause": controller["paused"] and not controller["halted"],
+                    "resume": not controller["paused"] and not controller["halted"],
+                    "halt": controller["halted"]}
+        require(expected[action], "control_readback_state")
+        live = checks.get("service_active") is True and checks.get("heartbeat_fresh") is True
+        terminal = (observed.get("terminal_halt_verified") is True
+                    and observed.get("lifecycle") == "halted" and observed.get("online") is False)
+        # A legitimate fresh terminal observation proves process exit through
+        # the owner observer. It must not fabricate a live controller heartbeat.
+        require(terminal if action == "halt" else live, "control_readback_unavailable")
+
+    def _front(self, name: str, result: dict[str, Any]) -> None:
+        provider = FRONTS[name]
+        require(result.get("provider") == provider and all(result.get(key) == value for key, value in
+                    self.policy["providers"][provider].items()), "front_provider_identity")
+        require(all(result.get(key) is True for key in ("model_observed", "account_verified", "subscription_auth",
+                    "heartbeat_verified", "scoped_mcp_verified", "uncertain_replay_suppressed", "cleanup_verified")), "front_incomplete")
+        require(type(result.get("status_inference_count")) is int and result["status_inference_count"] == 0, "front_status_inference")
+        require(all(identifier(result.get(key)) for key in ("run_id", "worker_id", "session_id",
+                    "provider_invocation_id", "hook_receipt_id", "capture_receipt_id")), "front_receipt_missing")
+        require(result.get("resumed_session_id") == result["session_id"], "front_resume_mismatch")
+        require(digest(result.get("hook_readback_sha256")) and digest(result.get("capture_readback_sha256")), "front_capture_readback")
+        require(result["provider_invocation_id"] not in self.invocations, "provider_replay")
+        self.invocations.add(result["provider_invocation_id"])
+        self._snapshot(result.get("status_snapshot"), result["run_id"])
+        start = result.get("start")
+        require(isinstance(start, dict) and start.get("state") == "singleton_observed"
+                and start.get("already_running") is True and start.get("readback_verified") is True,
+                "front_start_unverified")
+        self._target(start.get("target"), result["run_id"])
+        controls = result.get("controls")
+        require(isinstance(controls, dict) and set(controls) == set(CONTROL_STATES), "front_controls_missing")
+        for action in CONTROL_STATES:
+            self._control(action, controls[action], result["run_id"])
+
+    def _restoration(self, name: str, previous_run_id: str) -> None:
+        result, _ = self._observe("restoration", name)
+        require(result.get("method") == "admitted_owner_operation"
+                and result.get("admission_id") == self.policy["admission_id"]
+                and result.get("previous_run_id") == previous_run_id
+                and identifier(result.get("operation_id"))
+                and result["operation_id"] not in self.restoration_ids, "restoration_unverified")
+        self.restoration_ids.add(result["operation_id"])
+        self._snapshot(result.get("snapshot"))
+
+    def _rollback(self) -> None:
+        rollback, _ = self._observe("rollback", "restore-and-recover")
+        require(rollback.get("verified") is True and rollback.get("restored_readback_verified") is True
+                and rollback.get("recovered_readback_verified") is True
+                and rollback.get("restored_artifact_sha256") == self.policy["rollback_artifact_sha256"]
+                and rollback.get("recovered_artifact_sha256") == self.policy["bindings"]["artifact_sha256"]
+                and rollback.get("restored_source_commit") == self.policy["rollback_source_commit"]
+                and rollback.get("restored_checkpoint_source_commit") == self.policy["rollback_source_commit"]
+                and rollback.get("recovered_source_commit") == self.policy["bindings"]["source_commit"]
+                and rollback.get("recovered_checkpoint_source_commit") == self.policy["bindings"]["source_commit"]
+                and all(rollback.get(observed) == self.policy[expected] for observed, expected in {
+                    "restored_checkpoint_sha256": "rollback_checkpoint_sha256",
+                    "restored_configuration_sha256": "rollback_configuration_sha256",
+                    "recovered_checkpoint_sha256": "recovered_checkpoint_sha256",
+                    "recovered_configuration_sha256": "recovered_configuration_sha256",
+                }.items()), "rollback_failed")
+
+    def collect(self) -> dict[str, Any]:
+        require(not self.evidence, "collector_already_used")
+        self._admission()
+        for front in FRONTS:
+            result, _ = self._observe("front", front)
+            self._front(front, result)
+            # A halted singleton cannot serve the next front. Its independent
+            # owner must use an already admitted restoration operation; frontend
+            # start reconciliation itself is not a bootstrap or reset grant.
+            self._restoration(front, result["run_id"])
+        for name in NEGATIVE_CONTROLS:
+            result, _ = self._observe("negative", name)
+            require(result.get("denied") is True and result.get("side_effects") is False
+                    and digest(result.get("before_sha256")) and result.get("before_sha256") == result.get("after_sha256"), "negative_control_failed")
+        # Restore the prior compatible checkpoint/config and then the candidate
+        # before measuring final health. A Worker image rollback alone is not
+        # checkpoint restoration.
+        self._rollback()
+        probes = []
+        for index in range(3):
+            result, observed = self._observe("stability", str(index))
+            require(all(result.get(key) is True for key in ("healthy", "resource_admitted", "heartbeat_verified"))
+                    and type(result.get("unexpected_restarts")) is int and result["unexpected_restarts"] == 0, "stability_failed")
+            # Aggregate health is not eight-front evidence. Each route requires
+            # its own fresh, owner-issued and independently recalled observation.
+            require("snapshots" not in result, "stability_snapshots_unadmitted")
+            for front, provider in FRONTS.items():
+                route, _ = self._observe("stability-front", f"{index}:{front}")
+                snapshot = route.get("snapshot")
+                require(route.get("front") == front and route.get("provider") == provider
+                        and route.get("route_readback_verified") is True
+                        and digest(route.get("snapshot_sha256"))
+                        and route["snapshot_sha256"] == sha256_json(snapshot), "stability_route_binding")
+                # The Cloudflare singleton run_id may legitimately be shared
+                # across fronts; identity is the independently recalled route.
+                self._snapshot(snapshot)
+            probes.append(observed)
+        require(probes[0] < probes[1] < probes[2] and (probes[2] - probes[0]).total_seconds() >= 30, "stability_interval")
+        external, _ = self._observe("external", "acceptance")
+        require(external.get("accepted") is True and external.get("ui_assets_verified") is True
+                and external.get("access_session_verified") is True
+                and identifier(external.get("reviewer_id")) and external["reviewer_id"] != self.policy["collector_id"]
+                and external.get("origin") == self.policy["bindings"]["cloudflare_origin"], "external_acceptance_failed")
+        self._admission()
+        finished = self.clock()
+        require((finished - self.started).total_seconds() <= 1800, "collection_deadline")
+        payload = {"schema_version": SCHEMA_VERSION, "profile": GENERIC_PROFILE,
+                   "attestation_id": "finisher-" + self.challenge,
+                   "target_identity_digest": self.policy["bindings"]["target_identity_digest"],
+                   "environment_identity_digest": self.policy["bindings"]["environment_identity_digest"],
+                   "source_commit": self.policy["bindings"]["source_commit"],
+                   "deployment_sha": self.policy["bindings"]["source_commit"],
+                   "observed_at": self.started.isoformat(),
+                   "expires_at": (self.started + timedelta(hours=1)).isoformat(),
+                   "required_checks": sorted(BASE_CHECKS), "checks": {key: True for key in sorted(BASE_CHECKS)},
+                   "stability_probe_count": len(probes), "collector_execution_id": self.policy["execution_id"],
+                   "observation_manifest_sha256": sha256_json(self.evidence), "policy_sha256": self.policy_hash}
+        return {"schema": "echo.finisher.unsigned-collection.v1", "status": "UNSIGNED_CANDIDATE",
+                "release_verdict": "NOT_READY", "signature_verified": False, "owner_signing_required": True,
+                "attestation_payload": payload, "evidence": self.evidence,
+                "finished_at": finished.isoformat(), "bindings": self.policy["bindings"]}
diff --git a/tests/test_finisher_collector.py b/tests/test_finisher_collector.py
new file mode 100644
index 0000000..cd2bb37
--- /dev/null
+++ b/tests/test_finisher_collector.py
@@ -0,0 +1,566 @@
+"""Synthetic authority fixtures exercise rejection paths; none is live evidence."""
+from copy import deepcopy
+from datetime import UTC, datetime, timedelta
+import json
+import hashlib
+import os
+import sys
+import subprocess
+import time
+from pathlib import Path
+
+import pytest
+
+from echo_certification_forge.canonical import sha256_json
+from echo_certification_forge.finisher_collector import (
+    CollectorError, FRONTS, NEGATIVE_CONTROLS, FinisherCollector, PinnedOwnerProbe,
+)
+from echo_certification_forge.production_e2e import BASE_CHECKS, validate_attestation_trust_metadata
+
+NOW = datetime(2026, 11, 7, 6, 0, tzinfo=UTC)
+
+
+def policy():
+    return {
+        "schema": "echo.finisher.collector-policy.v1", "tenant_id": "tenant-fixture",
+        "collector_id": "independent-fixture", "authority_id": "owner-fixture",
+        "execution_id": "execution-fixture", "admission_id": "admission-fixture",
+        "bindings": {"target_identity_digest": "1" * 64,
+                     "environment_identity_digest": "2" * 64,
+                     "source_commit": "3" * 40, "native_source_commit": "4" * 40,
+                     "transport_source_commit": "3" * 40, "artifact_sha256": "5" * 64,
+                     "image_digest": "sha256:" + "6" * 64,
+                     "native_artifact_sha256": "7" * 64,
+                     "installation_id": "installation-fixture", "owner_id": "owner-fixture",
+                     "account_id": "a" * 32, "application_id": "application-fixture",
+                     "durable_object_id": "b" * 64,
+                     "cloudflare_origin": "https://finisher.example.test"},
+        "rollback_artifact_sha256": "8" * 64,
+        "rollback_source_commit": "9" * 40,
+        "rollback_checkpoint_sha256": "1" * 64,
+        "rollback_configuration_sha256": "3" * 64,
+        "recovered_checkpoint_sha256": "2" * 64,
+        "recovered_configuration_sha256": "4" * 64,
+        "providers": {v: {"model": v + "-fixture-model", "account_sha256": "a" * 64}
+                      for v in set(FRONTS.values())},
+    }
+
+
+class OwnerFixture:
+    """In-memory test authority, not a provider, deployment or signing service."""
+    def __init__(self, config):
+        self.config = deepcopy(config)
+        self.time = NOW
+        self.receipts = {}
+        self.calls = []
+        self.modify = lambda value: value
+        self.admitted = True
+        self.nonce_override = None
+        self.shared_run_id = False
+
+    def run_id(self, name):
+        return "run-shared-singleton" if self.shared_run_id else "run-" + name
+
+    def clock(self):
+        return self.time
+
+    def snapshot(self, name):
+        binding = self.config["bindings"]
+        return {"schema": "echo.finisher.ui.snapshot.v1", "observed_at": self.time.isoformat(),
+                "source_sha": binding["transport_source_commit"], "online": True,
+                "lifecycle": "idle", "readback_verified": True,
+                "checks": {key: True for key in ("installation_verified", "service_active",
+                                                  "heartbeat_fresh", "resource_admitted")},
+                "target": {"provider": "cloudflare", **{key: binding[key] for key in
+                           ("account_id", "application_id", "durable_object_id", "installation_id")},
+                           "source_sha": binding["transport_source_commit"], "run_id": self.run_id(name)}}
+
+    def call(self, request):
+        self.calls.append(deepcopy(request))
+        self.time += timedelta(seconds=1)
+        op = request["operation"]
+        common = {"schema": "echo.finisher.owner-response.v1", "nonce": request["nonce"],
+                  "request_sha256": sha256_json(request), "authority_id": self.config["authority_id"],
+                  "authenticated": True, "authorized": True,
+                  "observed_at": self.time.isoformat()}
+        if self.nonce_override:
+            common["nonce"] = self.nonce_override
+        if op == "admission":
+            value = {"allowed": self.admitted, "admission_id": self.config["admission_id"],
+                     "collector_id": self.config["collector_id"],
+                     "execution_id": self.config["execution_id"], "tenant_id": self.config["tenant_id"],
+                     "bindings": self.config["bindings"], "expires_at": (self.time + timedelta(minutes=15)).isoformat(),
+                     "operations": ["observe", "recall"], "independent": True,
+                     "policy_sha256": request["policy_sha256"], "restoration_allowed": True}
+        elif op == "observe":
+            kind, name = request["kind"], request["name"]
+            if kind == "stability":
+                self.time += timedelta(seconds=15)
+            receipt_id = "receipt-" + str(len(self.receipts))
+            value = {"receipt_id": receipt_id}
+            data = {"schema": "echo.finisher.observation.v1", "receipt_id": receipt_id,
+                    "tenant_id": self.config["tenant_id"], "execution_id": self.config["execution_id"],
+                    "collector_id": self.config["collector_id"], "bindings": deepcopy(self.config["bindings"]),
+                    "challenge": request["challenge"], "kind": kind, "name": name,
+                    "observed_at": self.time.isoformat(), "mode": "live"}
+            if kind == "front":
+                provider = FRONTS[name]
+                data["result"] = {
+                    "provider": provider, **self.config["providers"][provider],
+                    "provider_invocation_id": "invoke-" + name,
+                    "model_observed": True, "account_verified": True, "subscription_auth": True,
+                    "run_id": self.run_id(name), "worker_id": "worker-" + name,
+                    "session_id": "session-" + name, "resumed_session_id": "session-" + name,
+                    "heartbeat_verified": True, "scoped_mcp_verified": True,
+                    "hook_receipt_id": "hook-" + name, "capture_receipt_id": "capture-" + name,
+                    "hook_readback_sha256": "b" * 64, "capture_readback_sha256": "c" * 64,
+                    "uncertain_replay_suppressed": True, "status_inference_count": 0,
+                    "status_snapshot": self.snapshot(name),
+                    "start": {"state": "singleton_observed", "already_running": True,
+                              "readback_verified": True, "target": self.snapshot(name)["target"]},
+                    "controls": {action: {"request_id": name + "-" + action, "accepted": True,
+                                           "applied": True, "state": state, "action": action,
+                                           "native_request_id": "native-" + name + "-" + action,
+                                           "submitted_at": self.time.isoformat(),
+                                           "target": self.snapshot(name)["target"],
+                                           "readback": {"observed_at": self.time.isoformat(),
+                                               "target": self.snapshot(name)["target"],
+                                               "source_sha": self.config["bindings"]["transport_source_commit"],
+                                               "readback_verified": True, "terminal_halt_verified": action == "halt",
+                                               "lifecycle": "halted" if action == "halt" else "running",
+                                               "online": action != "halt",
+                                               "checks": {"installation_verified": True,
+                                                          "service_active": action != "halt", "heartbeat_fresh": action != "halt"},
+                                               "controller": {"paused": action == "pause", "halted": action == "halt",
+                                                              "last_control_id": "native-" + name + "-" + action}}}
+                                 for action, state in {"pause": "paused", "resume": "running", "halt": "halted"}.items()},
+                    "cleanup_verified": True,
+                }
+            elif kind == "restoration":
+                data["result"] = {"method": "admitted_owner_operation", "operation_id": "restore-" + name,
+                                  "admission_id": self.config["admission_id"],
+                                  "previous_run_id": self.run_id(name), "snapshot": self.snapshot(name)}
+            elif kind == "negative":
+                data["result"] = {"denied": True, "side_effects": False,
+                                  "before_sha256": "d" * 64, "after_sha256": "d" * 64}
+            elif kind == "stability":
+                data["result"] = {"healthy": True, "resource_admitted": True,
+                                  "heartbeat_verified": True, "unexpected_restarts": 0}
+            elif kind == "stability-front":
+                probe_index, front = name.split(":", 1)
+                assert probe_index in ("0", "1", "2")
+                snapshot = self.snapshot(front)
+                data["result"] = {"front": front, "provider": FRONTS[front],
+                                  "route_readback_verified": True, "snapshot": snapshot,
+                                  "snapshot_sha256": sha256_json(snapshot)}
+            elif kind == "external":
+                data["result"] = {"accepted": True, "reviewer_id": "other-reviewer",
+                                  "ui_assets_verified": True, "access_session_verified": True,
+                                  "origin": self.config["bindings"]["cloudflare_origin"]}
+            elif kind == "rollback":
+                data["result"] = {"restored_artifact_sha256": self.config["rollback_artifact_sha256"],
+                                  "recovered_artifact_sha256": self.config["bindings"]["artifact_sha256"],
+                                  "restored_source_commit": self.config["rollback_source_commit"],
+                                  "recovered_source_commit": self.config["bindings"]["source_commit"],
+                                  "restored_checkpoint_source_commit": self.config["rollback_source_commit"],
+                                  "recovered_checkpoint_source_commit": self.config["bindings"]["source_commit"],
+                                  "restored_checkpoint_sha256": "1" * 64,
+                                  "recovered_checkpoint_sha256": "2" * 64,
+                                  "restored_configuration_sha256": "3" * 64,
+                                  "recovered_configuration_sha256": "4" * 64,
+                                  "restored_readback_verified": True, "recovered_readback_verified": True,
+                                  "verified": True}
+            self.receipts[receipt_id] = self.modify(data)
+        else:
+            data = deepcopy(self.receipts[request["receipt_id"]])
+            value = {"receipt_id": request["receipt_id"], "content": data, "content_sha256": sha256_json(data)}
+        return {**common, "result": value}
+
+
+def collect(config=None, fixture=None):
+    config = config or policy()
+    fixture = fixture or OwnerFixture(config)
+    return FinisherCollector(config, fixture, clock=fixture.clock).collect()
+
+
+def test_all_eight_fronts_are_observed_then_independently_recalled():
+    config = policy()
+    owner = OwnerFixture(config)
+    result = collect(config, owner)
+    assert result["status"] == "UNSIGNED_CANDIDATE"
+    assert result["release_verdict"] == "NOT_READY"
+    assert result["signature_verified"] is False
+    assert result["owner_signing_required"] is True
+    assert "signing_key_id" not in result["attestation_payload"]
+    assert validate_attestation_trust_metadata(result["attestation_payload"]) == (False, "production_e2e_signature_not_verified")
+    assert set(result["attestation_payload"]["checks"]) == BASE_CHECKS
+    assert {x["name"] for x in owner.calls if x["operation"] == "observe" and x["kind"] == "front"} == set(FRONTS)
+    observed = [x for x in owner.calls if x["operation"] == "observe"]
+    recalled = [x for x in owner.calls if x["operation"] == "recall"]
+    assert len(observed) == len(recalled) == 16 + len(NEGATIVE_CONTROLS) + 3 + 3 * len(FRONTS) + 2
+    journey = [(x["kind"], x["name"]) for x in observed]
+    for index, name in enumerate(FRONTS):
+        assert journey[index * 2:index * 2 + 2] == [("front", name), ("restoration", name)]
+    assert journey.index(("rollback", "restore-and-recover")) < journey.index(("stability", "0"))
+    assert [(kind, name) for kind, name in journey if kind == "stability-front"] == [
+        ("stability-front", f"{index}:{front}") for index in range(3) for front in FRONTS
+    ]
+
+
+def test_all_eight_routes_may_share_one_authorized_cloudflare_singleton_run():
+    config = policy()
+    owner = OwnerFixture(config)
+    owner.shared_run_id = True
+    result = collect(config, owner)
+    assert result["status"] == "UNSIGNED_CANDIDATE"
+    assert result["release_verdict"] == "NOT_READY"
+    assert len([x for x in owner.calls if x["operation"] == "recall"]) == 16 + len(NEGATIVE_CONTROLS) + 3 + 3 * len(FRONTS) + 2
+
+
+def test_missing_admission_stops_before_any_observation():
+    owner = OwnerFixture(policy())
+    owner.admitted = False
+    with pytest.raises(CollectorError, match="admission_denied"):
+        collect(fixture=owner)
+    assert [x["operation"] for x in owner.calls] == ["admission"]
+
+
+@pytest.mark.parametrize("field,value", [("nonce", "forged"), ("request_sha256", "0" * 64),
+                                         ("authenticated", False), ("authority_id", "candidate-author")])
+def test_forged_owner_response_rejected(field, value):
+    owner = OwnerFixture(policy())
+    original = owner.call
+    owner.call = lambda request: {**original(request), field: value}
+    with pytest.raises(CollectorError, match="owner_response"):
+        collect(fixture=owner)
+
+
+@pytest.mark.parametrize("mutation", [
+    lambda d: d.update(mode="fixture"),
+    lambda d: d.update(challenge="prior-run"),
+    lambda d: d["bindings"].update(source_commit="f" * 40),
+    lambda d: d["bindings"].update(durable_object_id="f" * 64),
+    lambda d: d["result"]["status_snapshot"]["target"].update(account_id="f" * 32),
+    lambda d: d["result"]["status_snapshot"].update(online=False),
+    lambda d: d["result"]["status_snapshot"].update(request_id="not-a-mutation"),
+    lambda d: d["result"]["controls"]["halt"].update(state="stopped"),
+    lambda d: d.update(observed_at=(NOW - timedelta(hours=2)).isoformat()),
+    lambda d: d["result"].update(model="invented"),
+    lambda d: d["result"].update(account_sha256="f" * 64),
+    lambda d: d["result"].update(subscription_auth=False),
+    lambda d: d["result"].update(resumed_session_id="latest-other-session"),
+    lambda d: d["result"]["controls"]["halt"].update(applied=False),
+    lambda d: d["result"].update(status_inference_count=1),
+    lambda d: d["result"].update(capture_readback_sha256="bad"),
+])
+def test_forged_or_incomplete_front_proof_is_not_promoted(mutation):
+    owner = OwnerFixture(policy())
+    def change(data):
+        if data["kind"] == "front":
+            mutation(data)
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError):
+        collect(fixture=owner)
+
+
+def test_recall_hash_mismatch_is_rejected():
+    owner = OwnerFixture(policy())
+    original = owner.call
+    def wrong(request):
+        value = original(request)
+        if request["operation"] == "recall":
+            value["result"]["content_sha256"] = "f" * 64
+        return value
+    owner.call = wrong
+    with pytest.raises(CollectorError, match="receipt_hash"):
+        collect(fixture=owner)
+
+
+@pytest.mark.parametrize("kind,field,value", [
+    ("negative", "side_effects", True), ("negative", "after_sha256", "e" * 64),
+    ("stability", "unexpected_restarts", 1), ("stability", "healthy", False),
+    ("external", "reviewer_id", "independent-fixture"),
+    ("external", "access_session_verified", False),
+    ("rollback", "restored_artifact_sha256", "f" * 64),
+    ("rollback", "restored_checkpoint_source_commit", "f" * 40),
+    ("rollback", "recovered_readback_verified", False),
+    ("restoration", "method", "unadmitted_reset"),
+    ("restoration", "previous_run_id", "wrong-run"),
+])
+def test_negative_stability_external_and_rollback_fail_closed(kind, field, value):
+    owner = OwnerFixture(policy())
+    def change(data):
+        if data["kind"] == kind:
+            data["result"][field] = value
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError):
+        collect(fixture=owner)
+
+
+@pytest.mark.parametrize("field", [
+    "rollback_checkpoint_sha256", "rollback_configuration_sha256",
+    "recovered_checkpoint_sha256", "recovered_configuration_sha256",
+])
+def test_missing_owner_approved_rollback_digest_pin_fails_before_observation(field):
+    config = policy()
+    del config[field]
+    with pytest.raises(CollectorError, match="policy_rollback_identity"):
+        collect(config)
+
+
+@pytest.mark.parametrize("field", [
+    "restored_checkpoint_sha256", "restored_configuration_sha256",
+    "recovered_checkpoint_sha256", "recovered_configuration_sha256",
+])
+def test_well_formed_but_wrong_rollback_digest_fails_closed(field):
+    owner = OwnerFixture(policy())
+    def change(data):
+        if data["kind"] == "rollback":
+            data["result"][field] = "f" * 64
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError, match="rollback_failed"):
+        collect(fixture=owner)
+
+
+def test_sensitive_receipt_content_is_rejected_without_echoing_value():
+    owner = OwnerFixture(policy())
+    owner.modify = lambda data: {**data, "authorization": "test-sensitive-value"}
+    with pytest.raises(CollectorError) as error:
+        collect(fixture=owner)
+    assert "test-sensitive-value" not in str(error.value)
+
+
+def test_pinned_probe_rejects_source_drift_before_execution(tmp_path):
+    path = tmp_path / "probe.py"
+    path.write_text("raise SystemExit('must not execute')")
+    with pytest.raises(CollectorError, match="probe_pin"):
+        PinnedOwnerProbe([sys.executable, str(path)], ["0" * 64, "0" * 64])
+
+
+def test_policy_rejects_userinfo_origin_and_boolean_counts():
+    config = policy()
+    config["bindings"]["cloudflare_origin"] = "https://name:password@example.test"
+    with pytest.raises(CollectorError, match="policy"):
+        collect(config)
+
+
+def test_duplicate_provider_invocation_receipt_is_rejected():
+    owner = OwnerFixture(policy())
+    def change(data):
+        if data["kind"] == "front":
+            data["result"]["provider_invocation_id"] = "reused-invocation"
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError, match="replay"):
+        collect(fixture=owner)
+
+
+def make_probe(tmp_path, source, timeout=1):
+    path = tmp_path / "probe.py"
+    path.write_text(source)
+    command = [str(Path(sys.executable).resolve()), str(path)]
+    pins = [hashlib.sha256(Path(item).read_bytes()).hexdigest() for item in command]
+    return PinnedOwnerProbe(command, pins, timeout=timeout)
+
+
+@pytest.mark.skipif(os.name != "posix", reason="Linux owner-probe transport")
+def test_nonreading_probe_has_whole_request_deadline_with_small_pipe(tmp_path, monkeypatch):
+    import fcntl
+    probe = make_probe(tmp_path, "import time; time.sleep(4)")
+    original = subprocess.Popen
+    def small_pipe(*args, **kwargs):
+        process = original(*args, **kwargs)
+        fcntl.fcntl(process.stdin.fileno(), fcntl.F_SETPIPE_SZ, 4096)
+        return process
+    monkeypatch.setattr(subprocess, "Popen", small_pipe)
+    started = time.monotonic()
+    with pytest.raises(CollectorError, match="probe_timeout"):
+        probe.call({"padding": "x" * 6000})
+    assert time.monotonic() - started < 2.5
+
+
+@pytest.mark.skipif(os.name != "posix", reason="Linux owner-probe transport")
+def test_normal_broker_state_db_is_preserved_and_route_overrides_removed(tmp_path, monkeypatch):
+    monkeypatch.setenv("SOL_STATE_DB", "/private/fixture.db")
+    monkeypatch.setenv("GOOGLE_APPLICATION_CREDENTIALS", "/must/not/inherit")
+    probe = make_probe(tmp_path, "import json,os,sys; json.load(sys.stdin); print(json.dumps({'state_ok':os.environ.get('SOL_STATE_DB')=='/private/fixture.db','override_present':'GOOGLE_APPLICATION_CREDENTIALS' in os.environ}))")
+    assert probe.call({"request": "fixture"}) == {"state_ok": True, "override_present": False}
+
+
+@pytest.mark.skipif(os.name != "posix", reason="Linux owner-probe transport")
+def test_probe_output_overflow_is_bounded(tmp_path):
+    probe = make_probe(tmp_path, "import sys; sys.stdin.read(); sys.stdout.write('x'*140000); sys.stdout.flush()")
+    with pytest.raises(CollectorError, match="probe_output_size"):
+        probe.call({"request": "fixture"})
+
+
+def test_healthy_boolean_cannot_replace_per_route_stability_readback():
+    owner = OwnerFixture(policy())
+    def change(data):
+        if data["kind"] == "stability-front" and data["name"] == "2:CFgrauto":
+            snapshot = data["result"]["snapshot"]
+            snapshot["online"] = False
+            # A correct route digest cannot turn an unhealthy snapshot into health.
+            data["result"]["snapshot_sha256"] = sha256_json(snapshot)
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError, match="status_snapshot_unhealthy"):
+        collect(fixture=owner)
+
+
+def test_cloned_owner_route_result_is_rejected_for_another_front():
+    owner = OwnerFixture(policy())
+    first = {}
+    def change(data):
+        if data["kind"] == "stability-front" and data["name"] == "0:auto":
+            first["result"] = deepcopy(data["result"])
+        if data["kind"] == "stability-front" and data["name"] == "0:cauto":
+            # The singleton target may match; route provenance still must not.
+            data["result"] = deepcopy(first["result"])
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError, match="stability_route_binding"):
+        collect(fixture=owner)
+
+
+@pytest.mark.parametrize("field,value", [
+    ("front", "auto"), ("provider", "claude"), ("route_readback_verified", False),
+    ("snapshot_sha256", "f" * 64),
+])
+def test_stability_route_provenance_tampering_fails_closed(field, value):
+    owner = OwnerFixture(policy())
+    def change(data):
+        if data["kind"] == "stability-front" and data["name"] == "0:cauto":
+            data["result"][field] = value
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError, match="stability_route_binding"):
+        collect(fixture=owner)
+
+
+def test_legacy_aggregate_snapshots_cannot_count_as_route_receipts():
+    owner = OwnerFixture(policy())
+    def change(data):
+        if data["kind"] == "stability":
+            data["result"]["snapshots"] = {front: owner.snapshot(front) for front in FRONTS}
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError, match="stability_snapshots_unadmitted"):
+        collect(fixture=owner)
+
+
+@pytest.mark.parametrize("field", ["account_id", "application_id", "durable_object_id"])
+def test_policy_requires_the_complete_cloudflare_identity_tuple(field):
+    config = policy()
+    del config["bindings"][field]
+    with pytest.raises(CollectorError, match="policy_bindings"):
+        collect(config)
+
+
+def test_admission_revocation_is_checked_before_exact_recall():
+    owner = OwnerFixture(policy())
+    original = owner.call
+    def revoke(request):
+        value = original(request)
+        if request["operation"] == "observe":
+            owner.admitted = False
+        return value
+    owner.call = revoke
+    with pytest.raises(CollectorError, match="admission_denied"):
+        collect(fixture=owner)
+    assert not any(request["operation"] == "recall" for request in owner.calls)
+
+
+def test_admission_cannot_authorize_a_different_journey_policy():
+    owner = OwnerFixture(policy())
+    original = owner.call
+    def wrong(request):
+        value = original(request)
+        if request["operation"] == "admission":
+            value["result"]["policy_sha256"] = "f" * 64
+        return value
+    owner.call = wrong
+    with pytest.raises(CollectorError, match="admission_journey"):
+        collect(fixture=owner)
+
+
+@pytest.mark.skipif(os.name != "posix", reason="Linux collector CLI")
+@pytest.mark.parametrize("case,reason", [("missing_probe", "owner_probe_unconfigured"),
+                                        ("wrong_pin", "policy_pin_mismatch"),
+                                        ("symlink", "policy_path_invalid")])
+def test_cli_refuses_missing_authority_or_bad_policy_without_output(tmp_path, case, reason):
+    tmp_path.chmod(0o700)
+    path = tmp_path / "policy.json"
+    raw = json.dumps(policy()).encode()
+    path.write_bytes(raw)
+    path.chmod(0o600)
+    pin = hashlib.sha256(raw).hexdigest()
+    if case == "wrong_pin":
+        pin = "0" * 64
+    if case == "symlink":
+        link = tmp_path / "link.json"
+        link.symlink_to(path)
+        path = link
+    output = tmp_path / "unsigned.json"
+    process = subprocess.run([sys.executable, str(Path(__file__).resolve().parents[1] / "scripts/collect_finisher_e2e.py"),
+                              "--policy", str(path), "--policy-sha256", pin, "--output", str(output)],
+                             capture_output=True, text=True, timeout=5)
+    assert process.returncode == 2
+    assert json.loads(process.stdout) == {"status": "HELD", "release_verdict": "NOT_READY", "reason": reason}
+    assert not output.exists()
+
+
+@pytest.mark.parametrize("mutation", [
+    lambda c: c["pause"]["target"].update(run_id="another-run"),
+    lambda c: c["pause"]["readback"]["target"].update(durable_object_id="f" * 64),
+    lambda c: c["pause"]["readback"].update(source_sha="f" * 40),
+    lambda c: c["resume"]["readback"]["controller"].update(last_control_id="different-request"),
+    lambda c: c["resume"]["readback"]["controller"].update(paused=True),
+    lambda c: c["halt"]["readback"]["controller"].update(halted=False),
+    lambda c: c["halt"]["readback"].update(terminal_halt_verified=False),
+    lambda c: c["pause"]["readback"]["checks"].update(heartbeat_fresh=False),
+    lambda c: c["pause"]["readback"].update(observed_at=(NOW - timedelta(seconds=31)).isoformat()),
+    lambda c: c["pause"].update(submitted_at=(NOW + timedelta(minutes=5)).isoformat()),
+    lambda c: c["resume"].update(action="pause"),
+])
+def test_action_readback_cannot_cross_target_run_time_or_actual_state(mutation):
+    owner = OwnerFixture(policy())
+    def change(data):
+        if data["kind"] == "front":
+            mutation(data["result"]["controls"])
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError):
+        collect(fixture=owner)
+
+
+def test_native_control_id_cannot_be_replayed_with_new_front_request_id():
+    owner = OwnerFixture(policy())
+    def change(data):
+        if data["kind"] == "front":
+            control = data["result"]["controls"]["pause"]
+            control["native_request_id"] = "reused-native-id"
+            control["readback"]["controller"]["last_control_id"] = "reused-native-id"
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError, match="control_replay"):
+        collect(fixture=owner)
+
+
+@pytest.mark.parametrize("mutation", [
+    lambda d: d.update(terminal_halt_verified=False, checks={"installation_verified": True, "service_active": True, "heartbeat_fresh": True}),
+    lambda d: d.update(lifecycle="running"),
+    lambda d: d.update(online=True),
+])
+def test_terminal_halt_requires_actual_stopped_cleanup_proof(mutation):
+    owner = OwnerFixture(policy())
+    def change(data):
+        if data["kind"] == "front":
+            mutation(data["result"]["controls"]["halt"]["readback"])
+        return data
+    owner.modify = change
+    with pytest.raises(CollectorError):
+        collect(fixture=owner)