diff --git a/docs/cadence-receiver-custody.md b/docs/cadence-receiver-custody.md
index 188c232..679925e 100644
--- a/docs/cadence-receiver-custody.md
+++ b/docs/cadence-receiver-custody.md
@@ -93,3 +93,44 @@ refusal and per-instance observer matrix. The runner's wrong-service-identity
 probes do not separately isolate wrong namespace while preserving the correct
 service account name. Record that check honestly rather than treating the
 combined refusal as the complete namespace test.
+
+## Independent hour-end custody readback
+
+Use a fresh attended platform-admin envelope after the observation hour; the
+initial session must not be held open. `scripts/cadence_observer_readback.py`
+reuses the existing uniquely selected independent read-only full-tenant
+operator credential in process memory. It changes no credential or workload
+and never grants the collector archive read permission. Its only transport
+is a temporary primary-cluster localhost port-forward that is terminated on
+exit, with redirects and environment HTTP proxies disabled.
+
+```bash
+python3 scripts/cadence_observer_readback.py \
+  --source qonto-assistant --tenant tenant:friendly:binky \
+  --instance '<exact-source-snapshot-instance>' \
+  --since '<acceptance-window-start-with-timezone>' \
+  --until '<acceptance-window-end-with-timezone>' \
+  --event-id '<exact-collector-event-id-1>' \
+  --event-id '<exact-collector-event-id-2>' \
+  --kubeconfig '<owner-selected-kubeconfig>' --server '<verified-primary-server>' \
+  --expected-receiver-image 'forgejo.coulomb.social/coulomb/audit-core@sha256:<reviewed-digest>' \
+  --receipt '<new-private-readback-receipt-path>'
+```
+
+Provide all capture event IDs (maximum256, unique). The reader only opens
+those IDs, the exact source/tenant/instance reconciliation query, and the
+protected global stream-findings surface. It outputs only findings for the
+selected source; no unrelated findings or request payload is persisted.
+Every event must match exact ID/source/tenant/instance and acceptance window.
+The sanitized event projection includes instance/sequence, action, observed
+and accepted timestamps, and exact heartbeat coverage. Actual custody
+accepted_at can distinguish a duplicate delivery receipt from original custody
+time; a collector duplicate200 is not itself first-acceptance timing proof.
+Approval Engine source/tenant is also allowed, but this helper requires an
+instance-bearing source record and therefore cannot substitute for its
+historical instance-free lifecycle acceptance path.
+
+Exit0 reports `independent_custody_readback_complete`, which proves only the
+bounded supplied custody reads. The source snapshot, contiguous startup/
+periodic capture, process inventory and negative observation matrix remain
+independent inputs to Kings Guard's assessment.
diff --git a/scripts/cadence_observer_readback.py b/scripts/cadence_observer_readback.py
new file mode 100644
index 0000000..75d12a3
--- /dev/null
+++ b/scripts/cadence_observer_readback.py
@@ -0,0 +1,316 @@
+#!/usr/bin/env python3
+"""Bounded independent operator readback for NK-WP-0035; no credential output."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import socket
+import subprocess
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from urllib.error import HTTPError
+from urllib.parse import urlencode
+from urllib.request import ProxyHandler, Request, build_opener
+
+from cadence_receiver_custody import compatible
+from factory_audit_custody import require, snapshot
+from native_factory_acceptance import NoRedirect, operator_identity
+from state_hub_preflight_lane import LaneError, bao, data
+
+SCOPES = {
+    "qonto-assistant": "tenant:friendly:binky",
+    "approval-engine": "tenant:platform",
+}
+
+
+def stamp(value):
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    require(parsed.tzinfo is not None, "timezone_required")
+    return parsed.astimezone(timezone.utc)
+
+
+def paths(source, tenant, instance, since, until, event_ids):
+    require(SCOPES.get(source) == tenant, "exact_source_tenant_required")
+    require(
+        isinstance(instance, str)
+        and 1 <= len(instance) <= 256
+        and instance.strip() == instance,
+        "exact_instance_required",
+    )
+    require(
+        stamp(since) <= stamp(until) <= datetime.now(timezone.utc),
+        "invalid_acceptance_window",
+    )
+    require(
+        1 <= len(event_ids) <= 256 and len(set(event_ids)) == len(event_ids),
+        "bounded_unique_event_ids_required",
+    )
+    require(
+        all(
+            isinstance(x, str) and re.fullmatch(r"[A-Za-z0-9_.:-]{1,200}", x)
+            for x in event_ids
+        ),
+        "invalid_event_id",
+    )
+    query = urlencode(
+        dict(
+            source=source,
+            tenant=tenant,
+            stream_instance_id=instance,
+            since=since,
+            until=until,
+        )
+    )
+    return ["/v1/reconciliation?" + query, "/v1/stream-findings"] + [
+        "/v1/events/" + x for x in event_ids
+    ]
+
+
+def read_http(origin, token, path, allowed):
+    require(path in allowed, "read_path_not_in_exact_packet")
+    request = Request(origin + path, headers={"Authorization": "Bearer " + token})
+    try:
+        response = build_opener(ProxyHandler({}), NoRedirect()).open(
+            request, timeout=10
+        )
+    except HTTPError as exc:
+        response = exc
+    with response:
+        raw = response.read(1048577)
+        require(len(raw) <= 1048576, "readback_too_large")
+        require(response.status == 200, "independent_readback_failed")
+        return json.loads(raw)
+
+
+def sanitize_record(record, source, tenant, instance, event_id, since, until):
+    require(
+        record.get("event_id") == event_id
+        and record.get("source") == source
+        and record.get("tenant") == tenant,
+        "record_source_tenant_id_mismatch",
+    )
+    detail = record["details"]["data"]
+    require(detail.get("stream_instance_id") == instance, "record_instance_mismatch")
+    require(
+        type(detail.get("stream_sequence")) is int and detail["stream_sequence"] > 0,
+        "record_sequence_missing",
+    )
+    accepted = record["accepted_at"]
+    require(
+        stamp(since) <= stamp(accepted) <= stamp(until),
+        "record_outside_acceptance_window",
+    )
+    action = record["action"]
+    coverage = (
+        [detail["class"]]
+        if action == "audit-core.heartbeat" and "class" in detail
+        else []
+    )
+    if source == "qonto-assistant":
+        require(
+            action in {"audit.allow", "audit.deny", "audit-core.heartbeat"},
+            "record_undeclared_class",
+        )
+        if action == "audit-core.heartbeat":
+            require(coverage == ["audit.deny"], "record_heartbeat_scope_mismatch")
+    return {
+        "event_id": event_id,
+        "source": source,
+        "tenant": tenant,
+        "stream_instance_id": instance,
+        "stream_sequence": detail["stream_sequence"],
+        "type": action,
+        "accepted_at": accepted,
+        "observed_at": record["observed_at"],
+        "covered_event_classes": coverage,
+        "independent_custody_readback": True,
+    }
+
+
+def readback(origin, token, source, tenant, instance, since, until, event_ids):
+    allowed = paths(source, tenant, instance, since, until, event_ids)
+    reconciliation = read_http(origin, token, allowed[0], allowed)
+    require(
+        all(
+            reconciliation.get(k) == v
+            for k, v in (
+                ("source", source),
+                ("tenant", tenant),
+                ("stream_instance_id", instance),
+                ("since", since),
+                ("until", until),
+            )
+        ),
+        "reconciliation_scope_window_mismatch",
+    )
+    raw_findings = read_http(origin, token, allowed[1], allowed)
+    findings = raw_findings.get("stream_findings", raw_findings.get("findings"))
+    require(isinstance(findings, list), "unexpected_findings_shape")
+    safe_findings = [
+        {
+            key: row.get(key)
+            for key in (
+                "kind",
+                "sender",
+                "source",
+                "class",
+                "expected_within_seconds",
+                "last_seen",
+                "overdue_seconds",
+            )
+        }
+        for row in findings
+        if row.get("source") == source
+    ]
+    records = [
+        sanitize_record(
+            read_http(origin, token, path, allowed),
+            source,
+            tenant,
+            instance,
+            event_id,
+            since,
+            until,
+        )
+        for path, event_id in zip(allowed[2:], event_ids, strict=True)
+    ]
+    return {
+        "reconciliation": reconciliation,
+        "stream_findings": safe_findings,
+        "events": records,
+        "scope": "bounded same-instance accepted custody; source-global findings require active-instance interpretation",
+    }
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("--source", choices=SCOPES, required=True)
+    for key in (
+        "tenant",
+        "instance",
+        "since",
+        "until",
+        "kubeconfig",
+        "server",
+        "expected-receiver-image",
+    ):
+        p.add_argument("--" + key, required=True)
+    p.add_argument("--event-id", action="append", required=True)
+    p.add_argument("--receipt", type=Path, required=True)
+    a = p.parse_args()
+    receipt = {
+        "schema_version": "platform.independent-cadence-readback/v1",
+        "status": "refused",
+        "credential_values_emitted": False,
+        "observed_at": datetime.now(timezone.utc).isoformat(),
+    }
+    fd = None
+    forward = None
+    try:
+        paths(a.source, a.tenant, a.instance, a.since, a.until, a.event_id)
+        require(
+            Path.home().parent.name == ".warden-attended-login"
+            and not any(
+                os.getenv(x) for x in ("BAO_TOKEN", "VAULT_TOKEN", "OPENBAO_TOKEN")
+            ),
+            "fresh_attended_warden_envelope_required",
+        )
+        fd = os.open(
+            a.receipt, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600
+        )
+
+        def save():
+            os.lseek(fd, 0, os.SEEK_SET)
+            os.ftruncate(fd, 0)
+            os.write(fd, (json.dumps(receipt, indent=2) + "\n").encode())
+            os.fsync(fd)
+
+        kube = [
+            "kubectl",
+            "--kubeconfig",
+            a.kubeconfig,
+            "--server",
+            a.server,
+            "--request-timeout=20s",
+        ]
+        receipt["receiver"] = compatible(kube, a.expected_receiver_image)
+        save()
+        identity = data(bao(["token", "lookup", "-format=json"]))["data"]
+        require(
+            "platform-admin" in identity["policies"]
+            and "root" not in identity["policies"],
+            "attended_platform_admin_required",
+        )
+        version, _, rows, _ = snapshot()
+        operator = operator_identity(rows)
+        token = (operator.get("tokens") or [operator["token"]])[0]
+        receipt.update(reader_name=operator["name"], registry_version=version)
+        with socket.socket() as sock:
+            sock.bind(("127.0.0.1", 0))
+            port = sock.getsockname()[1]
+        forward = subprocess.Popen(
+            kube
+            + [
+                "-n",
+                "audit-core",
+                "port-forward",
+                "--address=127.0.0.1",
+                "service/audit-core",
+                str(port) + ":8080",
+            ],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        for _ in range(40):
+            require(forward.poll() is None, "port_forward_failed")
+            try:
+                with socket.create_connection(("127.0.0.1", port), timeout=0.2):
+                    break
+            except OSError:
+                time.sleep(0.2)
+        else:
+            raise LaneError("port_forward_not_ready")
+        receipt.update(
+            readback(
+                "http://127.0.0.1:" + str(port),
+                token,
+                a.source,
+                a.tenant,
+                a.instance,
+                a.since,
+                a.until,
+                a.event_id,
+            )
+        )
+        receipt.update(
+            status="independent_custody_readback_complete",
+            helper_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        )
+        save()
+        return 0
+    except Exception as exc:
+        receipt["error"] = (
+            str(exc) if isinstance(exc, LaneError) else "contained_readback_failed"
+        )
+        if fd is not None:
+            save()
+        return 1
+    finally:
+        if forward is not None:
+            forward.terminate()
+            try:
+                forward.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                forward.kill()
+                forward.wait(timeout=5)
+        if fd is not None:
+            os.close(fd)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_cadence_observer_readback.py b/tests/test_cadence_observer_readback.py
new file mode 100644
index 0000000..63136a5
--- /dev/null
+++ b/tests/test_cadence_observer_readback.py
@@ -0,0 +1,154 @@
+"""Model-only independent reader bounds and no-payload projection tests."""
+
+import copy
+import json
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
+import cadence_observer_readback as lane  # noqa: E402
+
+SINCE = "2026-10-15T10:00:00Z"
+UNTIL = "2026-10-15T11:00:00Z"
+
+
+class ReadbackBounds(unittest.TestCase):
+    def record(self):
+        return dict(
+            event_id="test-1",
+            source="qonto-assistant",
+            tenant="tenant:friendly:binky",
+            action="audit-core.heartbeat",
+            accepted_at="2026-10-15T10:30:00Z",
+            observed_at="2026-10-15T10:29:59Z",
+            details={
+                "data": {
+                    "stream_instance_id": "instance",
+                    "stream_sequence": 1,
+                    "class": "audit.deny",
+                    "sensitive-request": "must-never-leave",
+                }
+            },
+        )
+
+    def sanitize(self, record):
+        return lane.sanitize_record(
+            record,
+            "qonto-assistant",
+            "tenant:friendly:binky",
+            "instance",
+            "test-1",
+            SINCE,
+            UNTIL,
+        )
+
+    def test_payload_is_never_returned(self):
+        result = self.sanitize(self.record())
+        self.assertNotIn("must-never-leave", json.dumps(result))
+        self.assertEqual(result["covered_event_classes"], ["audit.deny"])
+        self.assertTrue(result["independent_custody_readback"])
+
+    def test_wrong_source_tenant_instance_and_id_are_refused(self):
+        for key in ("source", "tenant", "event_id", "stream_instance_id"):
+            record = self.record()
+            if key == "stream_instance_id":
+                record["details"]["data"][key] = "other"
+            else:
+                record[key] = "other"
+            with self.subTest(key=key), self.assertRaises(lane.LaneError):
+                self.sanitize(record)
+
+    def test_outside_acceptance_window_is_refused(self):
+        record = self.record()
+        record["accepted_at"] = "2026-10-15T12:00:00Z"
+        with self.assertRaisesRegex(lane.LaneError, "acceptance_window"):
+            self.sanitize(record)
+
+    def test_paths_are_bounded_to_exact_approved_coordinates(self):
+        result = lane.paths(
+            "qonto-assistant",
+            "tenant:friendly:binky",
+            "instance",
+            SINCE,
+            UNTIL,
+            ["test-1"],
+        )
+        self.assertEqual(result[-1], "/v1/events/test-1")
+        with self.assertRaisesRegex(lane.LaneError, "exact_source_tenant"):
+            lane.paths(
+                "qonto-assistant", "tenant:other", "instance", SINCE, UNTIL, ["test-1"]
+            )
+        with self.assertRaisesRegex(lane.LaneError, "invalid_event_id"):
+            lane.paths(
+                "qonto-assistant",
+                "tenant:friendly:binky",
+                "instance",
+                SINCE,
+                UNTIL,
+                ["../secret"],
+            )
+
+    def test_reader_cannot_open_arbitrary_paths(self):
+        with self.assertRaisesRegex(lane.LaneError, "exact_packet"):
+            lane.read_http(
+                "http://127.0.0.1:1",
+                "fixture-only",
+                "/v1/events/other",
+                ["/v1/events/test-1"],
+            )
+
+    def test_readback_discards_unrelated_findings_and_checks_exact_reconciliation(self):
+        query = dict(
+            source="qonto-assistant",
+            tenant="tenant:friendly:binky",
+            stream_instance_id="instance",
+            since=SINCE,
+            until=UNTIL,
+            counts=[],
+        )
+        findings = {
+            "stream_findings": [
+                dict(source="unrelated", **{"class": "sensitive"}),
+                dict(
+                    source="qonto-assistant",
+                    kind="missing_heartbeat",
+                    **{"class": "audit.deny"},
+                    private="no-output",
+                ),
+            ]
+        }
+        with patch.object(
+            lane, "read_http", side_effect=[query, findings, self.record()]
+        ):
+            result = lane.readback(
+                "http://127.0.0.1:1",
+                "fixture",
+                "qonto-assistant",
+                "tenant:friendly:binky",
+                "instance",
+                SINCE,
+                UNTIL,
+                ["test-1"],
+            )
+        self.assertEqual(len(result["stream_findings"]), 1)
+        self.assertNotIn("no-output", json.dumps(result))
+        wrong = copy.deepcopy(query)
+        wrong["tenant"] = "other"
+        with patch.object(lane, "read_http", return_value=wrong):
+            with self.assertRaisesRegex(lane.LaneError, "scope_window_mismatch"):
+                lane.readback(
+                    "http://127.0.0.1:1",
+                    "fixture",
+                    "qonto-assistant",
+                    "tenant:friendly:binky",
+                    "instance",
+                    SINCE,
+                    UNTIL,
+                    ["test-1"],
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()