diff --git a/README.md b/README.md
index 43877c96..6673753c 100644
--- a/README.md
+++ b/README.md
@@ -375,8 +375,37 @@ uv run chronicle inventory-artifacts --root db/data
 # Upload all existing manifest-declared local artifacts to ledger-raw and write
 # storage.r2 metadata back into the manifests:
 uv run chronicle publish-raw --root db/data
+
+# Read back declared R2 objects and verify canonical keys, SHA-256 and size:
+uv run chronicle verify-raw --root db/data
 ```
 
+`verify-raw` streams each declared object with
+`wrangler r2 object get <bucket>/<key> --remote --pipe`. It leaves manifests,
+local artifacts, and remote objects unchanged. Canonical keys use the same
+country-aware routing as `publish-raw` and the manifest's pinned SHA-256;
+available local files are also checked against the remote bytes. The JSON
+report records expected, remote, and optional local digests and byte lengths.
+
+Each entry reports `verified`, `key_mismatch`, `digest_mismatch`,
+`size_mismatch`, `missing_remote`, `no_r2_location`, `fetch_error`, or
+`invalid_manifest`. Wrong keys are rejected before fetching; size mismatches
+take precedence over digest mismatches, so truncation reports `size_mismatch`.
+Explicit missing-object diagnostics report `missing_remote`; other command
+failures report `fetch_error`. Missing or malformed required metadata reports
+`invalid_manifest`.
+
+The command exits 0 only when every entry is `verified` and there are no
+manifest-level errors, following the existing artifact commands' validity
+rule. Entries without an R2 location are counted separately as
+`no_r2_location_count` and fail the audit. An existing root with no manifest
+entries succeeds with zero counts, as `publish-raw` does.
+
+It accepts `--manifest`, `--source-id`, `--package-id`, `--r2-prefix`, and
+`--wrangler-command` like `publish-raw`. `--r2-bucket` defaults to `ledger-raw`
+and supplies a fallback for locations without a bucket; an explicitly declared
+bucket is read as recorded.
+
 To coordinate broad PE source migration without jumping straight to semantic
 target construction, generate an agent batch plan from the PE manifest:
 
diff --git a/changelog.d/332.added.md b/changelog.d/332.added.md
new file mode 100644
index 00000000..ac879824
--- /dev/null
+++ b/changelog.d/332.added.md
@@ -0,0 +1,4 @@
+Add `chronicle verify-raw`, a read-only R2 audit of manifest-declared raw
+artifacts. It checks country-aware canonical keys, streams remote SHA-256 and
+byte lengths, compares available local files, and emits a deterministic JSON
+report with a nonzero exit code for unverified entries or manifest errors.
diff --git a/chronicle/artifacts.py b/chronicle/artifacts.py
index 5f69c9cc..4929bae6 100644
--- a/chronicle/artifacts.py
+++ b/chronicle/artifacts.py
@@ -13,7 +13,8 @@
 import shlex
 import sqlite3
 import subprocess
-from typing import Any
+from threading import Thread
+from typing import Any, BinaryIO, Literal, get_args
 from urllib.parse import unquote, urlparse
 
 import httpx
@@ -286,6 +287,109 @@ def to_dict(self) -> dict[str, Any]:
         }
 
 
+RawArtifactVerificationStatus = Literal[
+    "verified",
+    "key_mismatch",
+    "digest_mismatch",
+    "size_mismatch",
+    "missing_remote",
+    "no_r2_location",
+    "fetch_error",
+    "invalid_manifest",
+]
+
+
+@dataclass(frozen=True)
+class RawArtifactVerifyEntry:
+    """Read-only comparison of one declared R2 object and its pinned bytes."""
+
+    manifest_path: str
+    source_id: str
+    package_id: str
+    year: str
+    filename: str
+    local_path: str
+    status: RawArtifactVerificationStatus
+    canonical_key: str | None
+    r2_location: ArtifactStorageLocation | None
+    sha256_expected: str | None
+    size_bytes_expected: int | None
+    sha256_remote: str | None = None
+    size_bytes_remote: int | None = None
+    sha256_local: str | None = None
+    size_bytes_local: int | None = None
+    fetch: ArtifactCommandResult | None = None
+    errors: tuple[str, ...] = ()
+
+    @property
+    def valid(self) -> bool:
+        """Whether the key and all available bytes match the manifest."""
+        return self.status == "verified"
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return metadata and diagnostics without the remote payload."""
+        return {
+            "valid": self.valid,
+            "manifest_path": self.manifest_path,
+            "source_id": self.source_id,
+            "package_id": self.package_id,
+            "year": self.year,
+            "filename": self.filename,
+            "local_path": self.local_path,
+            "status": self.status,
+            "canonical_key": self.canonical_key,
+            "r2_location": self.r2_location.to_dict() if self.r2_location else None,
+            "sha256_expected": self.sha256_expected,
+            "size_bytes_expected": self.size_bytes_expected,
+            "sha256_remote": self.sha256_remote,
+            "size_bytes_remote": self.size_bytes_remote,
+            "sha256_local": self.sha256_local,
+            "size_bytes_local": self.size_bytes_local,
+            "fetch": self.fetch.to_dict() if self.fetch else None,
+            "errors": list(self.errors),
+        }
+
+
+@dataclass(frozen=True)
+class RawArtifactVerifyReport:
+    """Verification report; undeclared R2 locations also fail the audit."""
+
+    root: str
+    entries: tuple[RawArtifactVerifyEntry, ...]
+    errors: tuple[str, ...] = ()
+
+    @property
+    def valid(self) -> bool:
+        """Whether every entry was verified and every manifest was readable."""
+        return not self.errors and all(entry.valid for entry in self.entries)
+
+    @property
+    def counts(self) -> dict[str, int]:
+        """Count each closed status separately, including missing R2 metadata."""
+        return {
+            "manifest_count": len({entry.manifest_path for entry in self.entries}),
+            "artifact_count": len(self.entries),
+            "r2_link_count": sum(
+                entry.r2_location is not None for entry in self.entries
+            ),
+            "failed_count": sum(not entry.valid for entry in self.entries),
+            **{
+                f"{status}_count": sum(entry.status == status for entry in self.entries)
+                for status in get_args(RawArtifactVerificationStatus)
+            },
+        }
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return a JSON-serializable verification report."""
+        return {
+            "valid": self.valid,
+            "root": self.root,
+            "counts": self.counts,
+            "entries": [entry.to_dict() for entry in self.entries],
+            "errors": list(self.errors),
+        }
+
+
 @dataclass(frozen=True)
 class R2BootstrapReport:
     """Report from bootstrapping Chronicle R2 buckets."""
@@ -697,6 +801,260 @@ def publish_source_artifacts(
     )
 
 
+def verify_raw_artifacts(
+    root: str | Path,
+    *,
+    manifest_filename: str = "manifest.yaml",
+    source_id: str | None = None,
+    package_id: str | None = None,
+    r2_bucket: str = DEFAULT_R2_RAW_BUCKET,
+    r2_prefix: str | None = None,
+    wrangler_command: str = "npx wrangler",
+) -> RawArtifactVerifyReport:
+    """Read back declared R2 objects without changing local or remote files.
+
+    Keys use the manifest's pinned digest, even when the local file is absent
+    or differs. A declared bucket takes precedence over the fallback bucket.
+    Manifests and file entries use the same scan order as ``publish-raw``.
+    """
+    root_path = Path(root)
+    if not root_path.exists():
+        return RawArtifactVerifyReport(
+            root=str(root_path),
+            entries=(),
+            errors=(f"Root does not exist: {root_path}",),
+        )
+
+    entries: list[RawArtifactVerifyEntry] = []
+    errors: list[str] = []
+    for manifest_path in sorted(root_path.rglob(manifest_filename)):
+        try:
+            manifest = safe_load_yaml(manifest_path.read_text(encoding="utf-8"))
+        except (OSError, UnicodeError, yaml.YAMLError) as exc:
+            errors.append(f"Could not read {manifest_path}: {exc}")
+            continue
+        if manifest is None:
+            manifest = {}
+        if not isinstance(manifest, dict):
+            errors.append(f"Manifest must be a mapping: {manifest_path}")
+            continue
+        manifest_source_id = source_id or manifest.get("source_id")
+        manifest_package_id = package_id or manifest.get("package_id")
+        files = manifest.get("files", {})
+        if files is None:
+            files = {}
+        if not manifest_source_id:
+            errors.append(f"Manifest missing source_id: {manifest_path}")
+            continue
+        if not manifest_package_id:
+            errors.append(f"Manifest missing package_id: {manifest_path}")
+            continue
+        if not isinstance(files, dict):
+            errors.append(f"Manifest files must be a mapping: {manifest_path}")
+            continue
+        try:
+            resolved_r2_prefix = resolve_r2_prefix(
+                prefix=r2_prefix,
+                default_prefix=DEFAULT_R2_PREFIX,
+                source_id=str(manifest_source_id),
+                package_path=manifest_path,
+            )
+        except ValueError as exc:
+            errors.append(f"Could not resolve R2 prefix for {manifest_path}: {exc}")
+            continue
+        for year, spec in files.items():
+            entries.append(
+                _verify_raw_manifest_entry(
+                    manifest_path,
+                    str(manifest_source_id),
+                    str(manifest_package_id),
+                    year,
+                    spec,
+                    r2_bucket=r2_bucket,
+                    r2_prefix=resolved_r2_prefix,
+                    wrangler_command=wrangler_command,
+                )
+            )
+    return RawArtifactVerifyReport(
+        root=str(root_path), entries=tuple(entries), errors=tuple(errors)
+    )
+
+
+def _verify_raw_manifest_entry(
+    manifest_path: Path,
+    source_id: str,
+    package_id: str,
+    year: Any,
+    spec: Any,
+    *,
+    r2_bucket: str,
+    r2_prefix: str,
+    wrangler_command: str,
+) -> RawArtifactVerifyEntry:
+    malformed_spec = not isinstance(spec, dict)
+    spec = spec if isinstance(spec, dict) else {}
+    filename = str(spec.get("filename") or "")
+    local_path = manifest_path.parent / filename
+    expected_digest = spec.get("sha256")
+    expected_size = spec.get("size_bytes")
+    values: dict[str, Any] = {
+        "manifest_path": str(manifest_path),
+        "source_id": source_id,
+        "package_id": package_id,
+        "year": str(year),
+        "filename": filename,
+        "local_path": str(local_path),
+        "canonical_key": None,
+        "r2_location": None,
+        "sha256_expected": expected_digest
+        if isinstance(expected_digest, str)
+        else None,
+        "size_bytes_expected": expected_size if type(expected_size) is int else None,
+    }
+
+    def result(
+        status: RawArtifactVerificationStatus, *errors: str
+    ) -> RawArtifactVerifyEntry:
+        return RawArtifactVerifyEntry(status=status, errors=tuple(errors), **values)
+
+    if malformed_spec:
+        return result("invalid_manifest", "malformed_file_spec")
+    storage = spec.get("storage")
+    if storage is None:
+        return result("no_r2_location", "no_r2_location")
+    if not isinstance(storage, dict):
+        return result("invalid_manifest", "malformed_storage")
+    if "r2" not in storage or storage["r2"] is None:
+        return result("no_r2_location", "no_r2_location")
+    r2 = storage["r2"]
+    if not isinstance(r2, dict):
+        return result("invalid_manifest", "malformed_r2_location")
+    key = r2.get("key")
+    bucket = r2.get("bucket", r2_bucket)
+    if not isinstance(key, str) or not key or not isinstance(bucket, str) or not bucket:
+        return result("invalid_manifest", "missing_r2_key_or_bucket")
+    values["r2_location"] = ArtifactStorageLocation("r2", bucket, key)
+    if not isinstance(spec.get("filename"), str) or not filename:
+        return result("invalid_manifest", "missing_filename")
+    if (
+        not isinstance(expected_digest, str)
+        or len(expected_digest) != 64
+        or any(char not in "0123456789abcdef" for char in expected_digest)
+    ):
+        return result("invalid_manifest", "missing_or_invalid_sha256")
+    if type(expected_size) is not int or expected_size < 0:
+        return result("invalid_manifest", "missing_or_invalid_size_bytes")
+    try:
+        canonical_key = build_r2_key(
+            source_id=source_id,
+            package_id=package_id,
+            year=year,
+            sha256=expected_digest,
+            filename=filename,
+            prefix=r2_prefix,
+            package_path=manifest_path,
+        )
+    except ValueError as exc:
+        return result("invalid_manifest", str(exc))
+    values["canonical_key"] = canonical_key
+    if key != canonical_key:
+        return result("key_mismatch", "declared_key_differs_from_canonical_key")
+
+    try:
+        remote_digest, remote_size, fetch = _stream_r2_object(
+            values["r2_location"], wrangler_command=wrangler_command
+        )
+    except (OSError, ValueError) as exc:
+        return result("fetch_error", str(exc))
+    values["fetch"] = fetch
+    if not fetch.ok:
+        missing = any(
+            marker in fetch.stderr.lower()
+            for marker in (
+                "nosuchkey",
+                "specified key does not exist",
+                "object not found",
+                "object does not exist",
+            )
+        )
+        return result("missing_remote" if missing else "fetch_error", "r2_get_failed")
+    values["sha256_remote"] = remote_digest
+    values["size_bytes_remote"] = remote_size
+    try:
+        if local_path.exists():
+            with local_path.open("rb") as local_file:
+                local_digest, local_size = _hash_stream(local_file)
+            values["sha256_local"] = local_digest
+            values["size_bytes_local"] = local_size
+    except OSError as exc:
+        return result("fetch_error", f"Could not read local artifact: {exc}")
+    if remote_size != expected_size or (
+        values.get("size_bytes_local") is not None
+        and values["size_bytes_local"] != remote_size
+    ):
+        return result("size_mismatch", "remote_manifest_or_local_size_mismatch")
+    if remote_digest != expected_digest or (
+        values.get("sha256_local") is not None
+        and values["sha256_local"] != remote_digest
+    ):
+        return result("digest_mismatch", "remote_manifest_or_local_digest_mismatch")
+    return result("verified")
+
+
+def _hash_stream(stream: BinaryIO) -> tuple[str, int]:
+    """Compute a digest and byte length without retaining the payload."""
+    digest = hashlib.sha256()
+    size_bytes = 0
+    for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+        digest.update(chunk)
+        size_bytes += len(chunk)
+    return digest.hexdigest(), size_bytes
+
+
+def _stream_r2_object(
+    location: ArtifactStorageLocation, *, wrangler_command: str
+) -> tuple[str, int, ArtifactCommandResult]:
+    """Hash binary Wrangler stdout and concurrently drain bounded diagnostics."""
+    command = [
+        *shlex.split(wrangler_command),
+        "r2",
+        "object",
+        "get",
+        f"{location.bucket}/{location.key}",
+        "--remote",
+        "--pipe",
+    ]
+    diagnostics = bytearray()
+
+    def drain_stderr(stream: BinaryIO) -> None:
+        for chunk in iter(lambda: stream.read(8192), b""):
+            diagnostics.extend(chunk[: max(0, 65536 - len(diagnostics))])
+
+    with subprocess.Popen(
+        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+    ) as process:
+        reader = Thread(target=drain_stderr, args=(process.stderr,))
+        reader.start()
+        try:
+            digest, size_bytes = _hash_stream(process.stdout)
+        except BaseException:
+            process.kill()
+            raise
+        finally:
+            process.wait()
+            reader.join()
+    return (
+        digest,
+        size_bytes,
+        ArtifactCommandResult(
+            command=tuple(command),
+            returncode=process.returncode,
+            stdout="",
+            stderr=diagnostics.decode("utf-8", errors="replace"),
+        ),
+    )
+
+
 def build_artifact_rows(
     report: DerivedArtifactPublishReport,
 ) -> tuple[dict[str, Any], ...]:
diff --git a/chronicle/cli.py b/chronicle/cli.py
index b9098d7f..701d4223 100644
--- a/chronicle/cli.py
+++ b/chronicle/cli.py
@@ -26,6 +26,7 @@ def main() -> None:
         ["plan-pe-sources"],
         ["publish-derived"],
         ["publish-raw"],
+        ["verify-raw"],
         ["scaffold-package"],
         ["validate-concept-alignments"],
         ["validate-package"],
diff --git a/chronicle/harness.py b/chronicle/harness.py
index 298a1380..3ca4ee74 100644
--- a/chronicle/harness.py
+++ b/chronicle/harness.py
@@ -13,11 +13,13 @@
     DerivedArtifactPublishReport,
     R2BootstrapReport,
     RawArtifactPublishReport,
+    RawArtifactVerifyReport,
     bootstrap_r2_buckets,
     fetch_source_artifact,
     inventory_source_artifacts,
     publish_derived_artifacts,
     publish_source_artifacts,
+    verify_raw_artifacts,
 )
 from chronicle.bundle import (
     UK_BUNDLE_SOURCES,
@@ -388,6 +390,28 @@ def publish_raw_artifact_files(
     )
 
 
+def verify_raw_artifact_files(
+    root: str | Path,
+    *,
+    manifest_filename: str = "manifest.yaml",
+    source_id: str | None = None,
+    package_id: str | None = None,
+    r2_bucket: str = "ledger-raw",
+    r2_prefix: str | None = None,
+    wrangler_command: str = "npx wrangler",
+) -> RawArtifactVerifyReport:
+    """Verify declared raw R2 objects without editing manifests or artifacts."""
+    return verify_raw_artifacts(
+        root,
+        manifest_filename=manifest_filename,
+        source_id=source_id,
+        package_id=package_id,
+        r2_bucket=r2_bucket,
+        r2_prefix=r2_prefix,
+        wrangler_command=wrangler_command,
+    )
+
+
 def bootstrap_r2_storage(
     *,
     raw_bucket: str = "ledger-raw",
@@ -963,6 +987,48 @@ def main(argv: list[str] | None = None) -> int:
         help="Wrangler command prefix to use for R2 uploads.",
     )
 
+    raw_verify_parser = subparsers.add_parser(
+        "verify-raw",
+        help="Read back declared raw R2 objects and verify keys and pinned bytes",
+    )
+    raw_verify_parser.add_argument(
+        "--root",
+        type=Path,
+        default=Path("db/data"),
+        help="Root directory to scan for manifest.yaml files.",
+    )
+    raw_verify_parser.add_argument(
+        "--manifest",
+        default="manifest.yaml",
+        help="Manifest filename to scan for.",
+    )
+    raw_verify_parser.add_argument(
+        "--source-id",
+        help="Override manifest source_id for scanned artifacts.",
+    )
+    raw_verify_parser.add_argument(
+        "--package-id",
+        help="Override manifest package_id for scanned artifacts.",
+    )
+    raw_verify_parser.add_argument(
+        "--r2-bucket",
+        default="ledger-raw",
+        help="Fallback R2 bucket when a declared location omits its bucket.",
+    )
+    raw_verify_parser.add_argument(
+        "--r2-prefix",
+        help=(
+            "Canonical R2 key-prefix root. Defaults to country-aware raw/nz "
+            "or raw/uk for known publisher directories, and raw for legacy "
+            "US packages."
+        ),
+    )
+    raw_verify_parser.add_argument(
+        "--wrangler-command",
+        default="npx wrangler",
+        help="Wrangler command prefix to use for streaming R2 reads.",
+    )
+
     r2_parser = subparsers.add_parser(
         "bootstrap-r2",
         help="Create Chronicle R2 buckets when Wrangler is authenticated",
@@ -1341,6 +1407,18 @@ def main(argv: list[str] | None = None) -> int:
         )
         print(json.dumps(report.to_dict(), indent=2, sort_keys=True))
         return 0 if report.valid else 1
+    if args.command == "verify-raw":
+        report = verify_raw_artifact_files(
+            args.root,
+            manifest_filename=args.manifest,
+            source_id=args.source_id,
+            package_id=args.package_id,
+            r2_bucket=args.r2_bucket,
+            r2_prefix=args.r2_prefix,
+            wrangler_command=args.wrangler_command,
+        )
+        print(json.dumps(report.to_dict(), indent=2, sort_keys=True))
+        return 0 if report.valid else 1
     if args.command == "bootstrap-r2":
         report = bootstrap_r2_storage(
             raw_bucket=args.raw_bucket,
diff --git a/tests/test_chronicle_verify_raw.py b/tests/test_chronicle_verify_raw.py
new file mode 100644
index 00000000..c7e974b0
--- /dev/null
+++ b/tests/test_chronicle_verify_raw.py
@@ -0,0 +1,610 @@
+"""Read-only raw artifact verification against a binary-emitting fake Wrangler."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import random
+import shlex
+import sys
+from dataclasses import FrozenInstanceError
+from datetime import date
+from pathlib import Path
+
+import pytest
+import yaml
+
+from chronicle.artifacts import build_r2_key, verify_raw_artifacts
+from chronicle.cli import main as cli_main
+from chronicle.harness import main as harness_main
+from policyengine_chronicle.cli import main as public_cli_main
+
+
+class FakeWrangler:
+    """Use an executable fixture, like the existing publish-raw tests."""
+
+    def __init__(self, directory: Path):
+        directory.mkdir(parents=True, exist_ok=True)
+        self.config = directory / "remote.json"
+        self.log = directory / "commands.jsonl"
+        driver = directory / "driver.py"
+        driver.write_text(
+            "import base64, json, pathlib, sys\n"
+            "config_path, log_path = map(pathlib.Path, sys.argv[1:3])\n"
+            "args = sys.argv[3:]\n"
+            "with log_path.open('a') as log:\n"
+            "    log.write(json.dumps(args) + '\\n')\n"
+            "if len(args) != 6 or args[:3] != ['r2', 'object', 'get'] "
+            "or args[4:] != ['--remote', '--pipe']:\n"
+            "    sys.stderr.write('Unsupported command')\n"
+            "    sys.exit(2)\n"
+            "config = json.loads(config_path.read_text())\n"
+            "location = args[3]\n"
+            "if location in config['errors']:\n"
+            "    sys.stderr.write(config['errors'][location])\n"
+            "    sys.exit(1)\n"
+            "if location not in config['objects']:\n"
+            "    sys.stderr.write('NoSuchKey: specified key does not exist')\n"
+            "    sys.exit(1)\n"
+            "sys.stderr.write(config.get('diagnostics', ''))\n"
+            "sys.stderr.flush()\n"
+            "payload = base64.b64decode(config['objects'][location])\n"
+            "for offset in range(0, len(payload), 65536):\n"
+            "    sys.stdout.buffer.write(payload[offset:offset + 65536])\n",
+            encoding="utf-8",
+        )
+        executable = directory / "wrangler"
+        executable.write_text(
+            "#!/bin/sh\nexec "
+            + " ".join(
+                shlex.quote(str(value))
+                for value in (sys.executable, driver, self.config, self.log)
+            )
+            + ' "$@"\n',
+            encoding="utf-8",
+        )
+        executable.chmod(0o755)
+        self.command = str(executable)
+        self.set_remote({})
+
+    def set_remote(self, objects: dict[str, bytes], errors=None, *, diagnostics=""):
+        self.config.write_text(
+            json.dumps(
+                {
+                    "objects": {
+                        key: base64.b64encode(value).decode("ascii")
+                        for key, value in objects.items()
+                    },
+                    "errors": errors or {},
+                    "diagnostics": diagnostics,
+                },
+                sort_keys=True,
+            ),
+            encoding="utf-8",
+        )
+
+    @property
+    def commands(self):
+        if not self.log.exists():
+            return []
+        return [json.loads(line) for line in self.log.read_text().splitlines()]
+
+
+def _add_artifact(
+    root,
+    payload=b"official NZ source bytes",
+    *,
+    source_id="ird",
+    package_id="ird-wff",
+    year=2024,
+    filename="wff.xlsx",
+    manifest_filename="manifest.yaml",
+    prefix=None,
+    bucket="ledger-raw",
+    local=True,
+):
+    directory = root / source_id / package_id
+    directory.mkdir(parents=True, exist_ok=True)
+    manifest_path = directory / manifest_filename
+    manifest = (
+        yaml.safe_load(manifest_path.read_text())
+        if manifest_path.exists()
+        else {"source_id": source_id, "package_id": package_id, "files": {}}
+    )
+    digest = hashlib.sha256(payload).hexdigest()
+    key = build_r2_key(
+        source_id=source_id,
+        package_id=package_id,
+        year=year,
+        sha256=digest,
+        filename=filename,
+        prefix=prefix,
+        package_path=manifest_path,
+    )
+    manifest["files"][year] = {
+        "filename": filename,
+        "sha256": digest,
+        "size_bytes": len(payload),
+        "storage": {"r2": {"provider": "r2", "bucket": bucket, "key": key}},
+    }
+    manifest_path.write_text(yaml.safe_dump(manifest, sort_keys=False))
+    if local:
+        (directory / filename).write_bytes(payload)
+    return manifest_path, key
+
+
+def _edit_artifact(manifest_path, edit, year=2024):
+    manifest = yaml.safe_load(manifest_path.read_text())
+    edit(manifest["files"][year])
+    manifest_path.write_text(yaml.safe_dump(manifest, sort_keys=False))
+
+
+def _snapshot(root):
+    return {
+        str(path.relative_to(root)): path.read_bytes() if path.is_file() else None
+        for path in sorted(root.rglob("*"))
+    }
+
+
+def _verify(root, fake, **options):
+    return verify_raw_artifacts(root, wrangler_command=fake.command, **options)
+
+
+def test_verify_raw_streams_binary_and_reports_pinned_measurements(tmp_path):
+    root = tmp_path / "db" / "data"
+    payload = bytes(range(256)) * 8192 + b"\x00\xff\xfe"
+    manifest_path, key = _add_artifact(root, payload)
+    fake = FakeWrangler(tmp_path / "fake")
+    fake.set_remote({f"ledger-raw/{key}": payload}, diagnostics="D" * 131072)
+    before = _snapshot(root)
+
+    report = _verify(root, fake)
+
+    assert report.valid
+    assert report.counts["manifest_count"] == 1
+    assert report.counts["artifact_count"] == 1
+    assert report.counts["r2_link_count"] == 1
+    assert report.counts["verified_count"] == 1
+    assert report.counts["failed_count"] == 0
+    entry = report.entries[0]
+    assert entry.status == "verified"
+    assert entry.manifest_path == str(manifest_path)
+    assert entry.source_id == "ird"
+    assert entry.package_id == "ird-wff"
+    assert entry.year == "2024"
+    assert entry.filename == "wff.xlsx"
+    assert entry.canonical_key == key
+    assert entry.r2_location.key == key
+    assert entry.r2_location.bucket == "ledger-raw"
+    assert entry.sha256_expected == hashlib.sha256(payload).hexdigest()
+    assert entry.sha256_remote == entry.sha256_expected == entry.sha256_local
+    assert entry.size_bytes_remote == entry.size_bytes_expected == len(payload)
+    assert entry.size_bytes_local == len(payload)
+    assert entry.fetch.ok
+    assert entry.fetch.stdout == ""
+    assert entry.fetch.stderr == "D" * 65536
+    assert entry.errors == report.errors == ()
+    assert fake.commands == [
+        ["r2", "object", "get", f"ledger-raw/{key}", "--remote", "--pipe"]
+    ]
+    assert _snapshot(root) == before
+    with pytest.raises(FrozenInstanceError):
+        report.root = "changed"
+    with pytest.raises(FrozenInstanceError):
+        entry.status = "changed"
+
+
+def test_verify_raw_allows_absent_local_file(tmp_path):
+    root = tmp_path / "db" / "data"
+    payload = b"remote-only artifact"
+    _, key = _add_artifact(root, payload, local=False)
+    fake = FakeWrangler(tmp_path / "fake")
+    fake.set_remote({f"ledger-raw/{key}": payload})
+    before = _snapshot(root)
+
+    report = _verify(root, fake)
+
+    assert report.valid
+    assert report.entries[0].status == "verified"
+    assert report.entries[0].sha256_local is None
+    assert report.entries[0].size_bytes_local is None
+    assert _snapshot(root) == before
+
+
+@pytest.mark.parametrize(
+    "status",
+    [
+        "key_mismatch",
+        "digest_mismatch",
+        "size_mismatch",
+        "missing_remote",
+        "no_r2_location",
+        "fetch_error",
+        "invalid_manifest",
+    ],
+)
+def test_verify_raw_failure_statuses_are_read_only(tmp_path, status):
+    root = tmp_path / "db" / "data"
+    payload = b"original official payload"
+    manifest_path, key = _add_artifact(root, payload)
+    fake = FakeWrangler(tmp_path / "fake")
+    objects = {f"ledger-raw/{key}": payload}
+    errors = {}
+    if status == "key_mismatch":
+        _edit_artifact(
+            manifest_path,
+            lambda entry: entry["storage"]["r2"].update(
+                key=key.replace("raw/nz/", "raw/")
+            ),
+        )
+    elif status == "digest_mismatch":
+        objects[f"ledger-raw/{key}"] = b"X" + payload[1:]
+    elif status == "size_mismatch":
+        objects[f"ledger-raw/{key}"] = payload[:-1]
+    elif status == "missing_remote":
+        objects = {}
+    elif status == "no_r2_location":
+        _edit_artifact(manifest_path, lambda entry: entry.pop("storage"))
+    elif status == "fetch_error":
+        errors[f"ledger-raw/{key}"] = "Authentication failed"
+    elif status == "invalid_manifest":
+        _edit_artifact(manifest_path, lambda entry: entry.update(size_bytes=True))
+    fake.set_remote(objects, errors)
+    before = _snapshot(root)
+
+    report = _verify(root, fake)
+
+    assert not report.valid
+    assert report.entries[0].status == status
+    assert report.counts["verified_count"] == 0
+    assert report.counts["failed_count"] == 1
+    assert report.counts[f"{status}_count"] == 1
+    assert _snapshot(root) == before
+    assert all(command[:3] == ["r2", "object", "get"] for command in fake.commands)
+    assert not any("put" in command or "delete" in command for command in fake.commands)
+    if status in {"key_mismatch", "no_r2_location", "invalid_manifest"}:
+        assert fake.commands == []
+        assert report.entries[0].fetch is None
+
+
+@pytest.mark.parametrize("status", ["digest_mismatch", "size_mismatch"])
+def test_verify_raw_compares_present_local_file(tmp_path, status):
+    root = tmp_path / "db" / "data"
+    payload = b"official source bytes"
+    manifest_path, key = _add_artifact(root, payload)
+    local = manifest_path.parent / "wff.xlsx"
+    local.write_bytes(
+        b"X" + payload[1:] if status == "digest_mismatch" else payload[:-1]
+    )
+    fake = FakeWrangler(tmp_path / "fake")
+    fake.set_remote({f"ledger-raw/{key}": payload})
+
+    report = _verify(root, fake)
+
+    assert not report.valid
+    assert report.entries[0].status == status
+    assert report.entries[0].sha256_remote == hashlib.sha256(payload).hexdigest()
+    assert (
+        report.entries[0].sha256_local == hashlib.sha256(local.read_bytes()).hexdigest()
+    )
+    assert report.entries[0].size_bytes_local == local.stat().st_size
+
+
+def test_verify_raw_rejects_malformed_entry_fields(tmp_path, capsys):
+    root = tmp_path / "db" / "data"
+    manifest_path, key = _add_artifact(root)
+    original = manifest_path.read_bytes()
+    fake = FakeWrangler(tmp_path / "fake")
+    fake.set_remote({f"ledger-raw/{key}": b"official NZ source bytes"})
+    changes = [
+        lambda entry: entry.update(filename=None),
+        lambda entry: entry.update(filename=123),
+        lambda entry: entry.update(sha256="not-a-digest"),
+        lambda entry: entry.update(sha256="x" * 64),
+        lambda entry: entry.update(sha256=date(2024, 1, 1)),
+        lambda entry: entry.update(sha256={"not", "a", "digest"}),
+        lambda entry: entry.update(sha256=b"binary digest"),
+        lambda entry: entry.pop("sha256"),
+        lambda entry: entry.update(size_bytes=-1),
+        lambda entry: entry.update(size_bytes="24"),
+        lambda entry: entry.update(size_bytes=date(2024, 1, 1)),
+        lambda entry: entry.update(size_bytes={1, 2}),
+        lambda entry: entry.update(size_bytes=b"24"),
+        lambda entry: entry.pop("size_bytes"),
+        lambda entry: entry.update(storage={"r2": "bad"}),
+        lambda entry: entry["storage"]["r2"].update(key=""),
+        lambda entry: entry["storage"]["r2"].update(key=123),
+        lambda entry: entry["storage"]["r2"].update(bucket=""),
+        lambda entry: entry["storage"]["r2"].update(bucket=123),
+    ]
+    for change in changes:
+        manifest_path.write_bytes(original)
+        _edit_artifact(manifest_path, change)
+        before = _snapshot(root)
+        report = _verify(root, fake)
+        assert not report.valid
+        assert report.entries[0].status == "invalid_manifest"
+        assert _snapshot(root) == before
+        exit_code = harness_main(
+            ["verify-raw", "--root", str(root), "--wrangler-command", fake.command]
+        )
+        payload = json.loads(capsys.readouterr().out)
+        assert exit_code == 1
+        assert payload["entries"][0]["status"] == "invalid_manifest"
+        assert _snapshot(root) == before
+    assert fake.commands == []
+
+
+def test_verify_raw_missing_object_markers_and_other_errors(tmp_path):
+    root = tmp_path / "db" / "data"
+    _, key = _add_artifact(root)
+    fake = FakeWrangler(tmp_path / "fake")
+    for marker in (
+        "NoSuchKey",
+        "The specified key does not exist",
+        "Object not found",
+        "Object does not exist",
+    ):
+        fake.set_remote({}, {f"ledger-raw/{key}": marker})
+        report = _verify(root, fake)
+        assert not report.valid
+        assert report.entries[0].status == "missing_remote"
+    for marker in ("connection refused", "HTTP 404", "code: 10007"):
+        fake.set_remote({}, {f"ledger-raw/{key}": marker})
+        assert _verify(root, fake).entries[0].status == "fetch_error"
+
+
+def test_verify_raw_missing_executable_is_fetch_error(tmp_path):
+    root = tmp_path / "db" / "data"
+    _add_artifact(root)
+    before = _snapshot(root)
+
+    report = verify_raw_artifacts(
+        root, wrangler_command=str(tmp_path / "absent-wrangler")
+    )
+
+    assert not report.valid
+    assert report.entries[0].status == "fetch_error"
+    assert _snapshot(root) == before
+
+
+def test_verify_raw_incident_object_only_at_legacy_key_is_missing(tmp_path):
+    root = tmp_path / "db" / "data"
+    payload = b"official WFF workbook"
+    _, key = _add_artifact(root, payload)
+    legacy_key = key.replace("raw/nz/", "raw/")
+    fake = FakeWrangler(tmp_path / "fake")
+    fake.set_remote({f"ledger-raw/{legacy_key}": payload})
+
+    report = _verify(root, fake)
+
+    assert not report.valid
+    assert report.entries[0].status == "missing_remote"
+    assert report.entries[0].canonical_key == key
+    assert fake.commands[0][3] == f"ledger-raw/{key}"
+
+
+def test_verify_raw_incident_manifest_at_legacy_key_is_key_mismatch(tmp_path):
+    root = tmp_path / "db" / "data"
+    payload = b"official WFF workbook"
+    manifest_path, key = _add_artifact(root, payload)
+    legacy_key = key.replace("raw/nz/", "raw/")
+    _edit_artifact(
+        manifest_path, lambda entry: entry["storage"]["r2"].update(key=legacy_key)
+    )
+    fake = FakeWrangler(tmp_path / "fake")
+    fake.set_remote({f"ledger-raw/{legacy_key}": payload})
+
+    report = _verify(root, fake)
+
+    assert not report.valid
+    assert report.entries[0].status == "key_mismatch"
+    assert report.entries[0].canonical_key == key
+    assert report.entries[0].r2_location.key == legacy_key
+    assert fake.commands == []
+
+
+def test_verify_raw_report_is_deterministic_in_publish_scan_order(tmp_path):
+    root = tmp_path / "db" / "data"
+    fake = FakeWrangler(tmp_path / "fake")
+    objects = {}
+    for source_id, package_id, year in (
+        ("ons", "ons-mye", "source_capture"),
+        ("ird", "ird-wff", 2025),
+        ("ird", "ird-wff", 2024),
+    ):
+        payload = str(year).encode()
+        _, key = _add_artifact(
+            root, payload, source_id=source_id, package_id=package_id, year=year
+        )
+        objects[f"ledger-raw/{key}"] = payload
+    # The shared local filename reflects the final 2024 entry. Remove it so
+    # both pinned years are valid remote-only artifacts.
+    for path in root.rglob("wff.xlsx"):
+        path.unlink()
+    fake.set_remote(objects)
+
+    first = _verify(root, fake)
+    second = _verify(root, fake)
+
+    assert first.valid and second.valid
+    assert first.to_dict() == second.to_dict()
+    assert [(entry.source_id, entry.year) for entry in first.entries] == [
+        ("ird", "2025"),
+        ("ird", "2024"),
+        ("ons", "source_capture"),
+    ]
+
+
+def test_verify_raw_cli_accepts_publish_flags_and_bucket_fallback(tmp_path, capsys):
+    root = tmp_path / "db" / "data"
+    payload = b"custom CLI artifact"
+    manifest_path, key = _add_artifact(
+        root,
+        payload,
+        package_id="override-package",
+        manifest_filename="artifacts.yaml",
+        prefix="archive",
+        bucket="declared-bucket",
+    )
+    manifest = yaml.safe_load(manifest_path.read_text())
+    manifest.pop("source_id")
+    manifest.pop("package_id")
+    manifest_path.write_text(yaml.safe_dump(manifest, sort_keys=False))
+    fake = FakeWrangler(tmp_path / "fake")
+    fake.set_remote({f"declared-bucket/{key}": payload})
+    argv = [
+        "verify-raw",
+        "--root",
+        str(root),
+        "--manifest",
+        "artifacts.yaml",
+        "--source-id",
+        "ird",
+        "--package-id",
+        "override-package",
+        "--r2-bucket",
+        "fallback-bucket",
+        "--r2-prefix",
+        "archive",
+        "--wrangler-command",
+        fake.command,
+    ]
+
+    assert harness_main(argv) == 0
+    assert json.loads(capsys.readouterr().out)["valid"]
+    assert fake.commands[-1][3] == f"declared-bucket/{key}"
+    _edit_artifact(manifest_path, lambda entry: entry["storage"]["r2"].pop("bucket"))
+    fake.set_remote({f"fallback-bucket/{key}": payload})
+    assert harness_main(argv) == 0
+    assert json.loads(capsys.readouterr().out)["valid"]
+    assert fake.commands[-1][3] == f"fallback-bucket/{key}"
+
+
+@pytest.mark.parametrize("status", ["verified", "missing_remote", "no_r2_location"])
+def test_verify_raw_cli_json_and_exit_status(tmp_path, capsys, status):
+    root = tmp_path / "db" / "data"
+    payload = b"CLI fixture"
+    manifest_path, key = _add_artifact(root, payload)
+    fake = FakeWrangler(tmp_path / "fake")
+    if status == "verified":
+        fake.set_remote({f"ledger-raw/{key}": payload})
+    if status == "no_r2_location":
+        _edit_artifact(manifest_path, lambda entry: entry.pop("storage"))
+
+    exit_code = harness_main(
+        ["verify-raw", "--root", str(root), "--wrangler-command", fake.command]
+    )
+    report = json.loads(capsys.readouterr().out)
+
+    assert exit_code == (0 if status == "verified" else 1)
+    assert report["valid"] is (status == "verified")
+    assert report["root"] == str(root)
+    assert report["counts"]["artifact_count"] == 1
+    assert report["counts"]["failed_count"] == (0 if status == "verified" else 1)
+    assert report["entries"][0]["status"] == status
+    assert report["entries"][0]["canonical_key"] == (
+        None if status == "no_r2_location" else key
+    )
+    assert isinstance(report["errors"], list)
+
+
+@pytest.mark.parametrize(
+    "entrypoint", [cli_main, public_cli_main], ids=["legacy", "public"]
+)
+def test_verify_raw_top_level_cli_dispatches(tmp_path, capsys, monkeypatch, entrypoint):
+    root = tmp_path / "db" / "data"
+    payload = b"dispatch fixture"
+    _, key = _add_artifact(root, payload)
+    fake = FakeWrangler(tmp_path / "fake")
+    fake.set_remote({f"ledger-raw/{key}": payload})
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "chronicle",
+            "verify-raw",
+            "--root",
+            str(root),
+            "--wrangler-command",
+            fake.command,
+        ],
+    )
+
+    with pytest.raises(SystemExit) as exc:
+        entrypoint()
+
+    assert exc.value.code == 0
+    assert json.loads(capsys.readouterr().out)["entries"][0]["status"] == "verified"
+
+
+def test_verify_raw_seeded_payload_property(tmp_path):
+    # Hypothesis is absent from this repository's dependencies. Exercise the
+    # property with seeded random bytes, including empty and multi-chunk data.
+    rng = random.Random(332)
+    root = tmp_path / "db" / "data"
+    fake = FakeWrangler(tmp_path / "fake")
+    lengths = (0, 1, 65536, 65537, 1_048_591) + tuple(
+        rng.randrange(2, 65536) for _ in range(3)
+    )
+    for length in lengths:
+        payload = rng.randbytes(length)
+        manifest_path, key = _add_artifact(root, payload)
+        original = manifest_path.read_bytes()
+        canonical_location = f"ledger-raw/{key}"
+        flipped = bytes([payload[0] ^ 1]) + payload[1:] if payload else b"\x01"
+        cases = [
+            ({canonical_location: payload}, "verified", key),
+            (
+                {canonical_location: flipped},
+                "digest_mismatch" if payload else "size_mismatch",
+                key,
+            ),
+            ({}, "missing_remote", key),
+            (
+                {f"ledger-raw/{key.replace('raw/nz/', 'raw/')}": payload},
+                "key_mismatch",
+                key.replace("raw/nz/", "raw/"),
+            ),
+        ]
+        if payload:
+            cases.append(({canonical_location: payload[:-1]}, "size_mismatch", key))
+        for remote, expected_status, declared_key in cases:
+            manifest_path.write_bytes(original)
+            _edit_artifact(
+                manifest_path,
+                lambda entry, declared_key=declared_key: entry["storage"]["r2"].update(
+                    key=declared_key
+                ),
+            )
+            fake.set_remote(remote)
+            before = _snapshot(root)
+            report = _verify(root, fake)
+            assert report.valid is (expected_status == "verified")
+            assert report.entries[0].status == expected_status
+            assert _snapshot(root) == before
+
+
+def test_verify_raw_reports_invalid_roots_and_manifests(tmp_path):
+    fake = FakeWrangler(tmp_path / "fake")
+    absent = tmp_path / "absent"
+    report = _verify(absent, fake)
+    assert not report.valid
+    assert report.errors
+    assert not absent.exists()
+    root = tmp_path / "db" / "data"
+    root.mkdir(parents=True)
+    empty = _verify(root, fake)
+    assert empty.valid
+    assert empty.entries == ()
+    assert all(count == 0 for count in empty.counts.values())
+    manifest_path = root / "manifest.yaml"
+    for malformed in ("[", "- scalar manifest\n", "files: []\n", "files: {}\n"):
+        manifest_path.write_text(malformed)
+        before = _snapshot(root)
+        report = _verify(root, fake)
+        assert not report.valid
+        assert report.errors
+        assert _snapshot(root) == before
+    assert fake.commands == []