diff --git a/.gitignore b/.gitignore
index 3ee158df..26b35ac2 100644
--- a/.gitignore
+++ b/.gitignore
@@ -31,3 +31,7 @@ db/data/voa/ctsop*_lsoa_2026/CTSOP_*.zip
 db/data/tpa/lvt_model_2026/config.py
 db/data/tpa/lvt_model_2026/cells.json
 db/data/tpa/lvt_model_2026/LICENSE
+# UCL's house price per square metre deposit (ReShare 856542, chronicle#409) is two
+# zips of 3.2 and 4.4 GB, above GitHub's file limit and carrying Royal Mail address
+# data licensed for non-commercial use. R2 holds them; the manifest pins the bytes.
+db/data/ucl/house_price_per_square_metre_856542/856542_bundle_*.zip
diff --git a/changelog.d/r2-large-raw-uploads-409.changed.md b/changelog.d/r2-large-raw-uploads-409.changed.md
new file mode 100644
index 00000000..fe2eb1ca
--- /dev/null
+++ b/changelog.d/r2-large-raw-uploads-409.changed.md
@@ -0,0 +1 @@
+`publish-raw`, `fetch-artifact` and `publish-derived` upload files above Wrangler's 300 MiB limit to R2 through its S3-compatible API (chronicle#409). Each file goes up as a multipart upload in 100 MiB parts, with one part in memory at a time, and a failed part is retried on its own. Every file is hashed a chunk at a time instead of being read whole, and `fetch-artifact` streams downloads to disk.
diff --git a/changelog.d/ucl-house-price-per-square-metre-856542.added.md b/changelog.d/ucl-house-price-per-square-metre-856542.added.md
new file mode 100644
index 00000000..e9491077
--- /dev/null
+++ b/changelog.d/ucl-house-price-per-square-metre-856542.added.md
@@ -0,0 +1 @@
+Register UCL's House Price Per Square Metre in England and Wales, 1995-2023 (UK Data Service ReShare 856542), both bundles, raw-only in R2 (chronicle#408, item 6).
diff --git a/chronicle/artifacts.py b/chronicle/artifacts.py
index c3f10266..0ff134fa 100644
--- a/chronicle/artifacts.py
+++ b/chronicle/artifacts.py
@@ -15,12 +15,13 @@
 import sqlite3
 import subprocess
 from threading import Thread
-from typing import Any, BinaryIO, Literal, get_args
+from typing import Any, BinaryIO, Iterator, Literal, get_args
 from urllib.parse import unquote, urlparse
 
 import httpx
 import yaml
 
+from chronicle import r2_s3
 from chronicle.epoch import EMIT_EPOCH, Epoch, canonicalize_key, hash_domain
 from chronicle.yaml_io import safe_load as safe_load_yaml
 
@@ -29,6 +30,9 @@
 DEFAULT_R2_DERIVED_BUCKET = "ledger-derived"
 DEFAULT_R2_PREFIX = "raw"
 DEFAULT_R2_DERIVED_PREFIX = "derived"
+#: ``wrangler r2 object put`` refuses objects above 300 MiB. Larger files go to R2
+#: through its S3-compatible API as multipart uploads (chronicle#409).
+WRANGLER_PUT_LIMIT_BYTES = 300 * 1024 * 1024
 
 # New UK and New Zealand uploads are namespaced by country. US objects predate
 # the country segment and deliberately keep their legacy ``raw/{source_id}``
@@ -53,6 +57,7 @@
             "ons",
             "scotgov",
             "slc",
+            "ucl",
             "voa",
             "welshgov",
         }
@@ -525,17 +530,13 @@ def fetch_source_artifact(
         package_path=output,
     )
     fetched_at = datetime.now(UTC).replace(microsecond=0).isoformat()
-    content, inferred_filename = _read_artifact(source_url)
-    artifact_filename = filename or inferred_filename
+    artifact_filename = filename or _filename_from_url(source_url)
     if not artifact_filename:
         raise ValueError("Could not infer artifact filename; pass --filename.")
 
     output.mkdir(parents=True, exist_ok=True)
     local_path = output / artifact_filename
-    local_path.write_bytes(content)
-
-    sha256 = hashlib.sha256(content).hexdigest()
-    size_bytes = len(content)
+    sha256, size_bytes = _read_artifact(source_url, local_path)
     manifest_path = output / "manifest.yaml"
 
     r2_location = ArtifactStorageLocation(
@@ -682,8 +683,7 @@ def publish_derived_artifacts(
         relative_path = artifact_path.relative_to(input_path).as_posix()
         if relative_path == "build_artifacts.jsonl":
             continue
-        content = artifact_path.read_bytes()
-        sha256 = hashlib.sha256(content).hexdigest()
+        sha256, size_bytes = _hash_file(artifact_path)
         location = ArtifactStorageLocation(
             provider="r2",
             bucket=r2_bucket,
@@ -708,7 +708,7 @@ def publish_derived_artifacts(
                 artifact_name=relative_path,
                 local_path=str(artifact_path),
                 sha256=sha256,
-                size_bytes=len(content),
+                size_bytes=size_bytes,
                 r2_location=location,
                 upload=upload,
             )
@@ -1016,6 +1016,12 @@ def _hash_stream(stream: BinaryIO) -> tuple[str, int]:
     return digest.hexdigest(), size_bytes
 
 
+def _hash_file(path: Path) -> tuple[str, int]:
+    """Hash a file a chunk at a time: raw files can be several GB (chronicle#409)."""
+    with path.open("rb") as stream:
+        return _hash_stream(stream)
+
+
 _ANSI_ESCAPE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
 _MISSING_OBJECT_MARKER = re.compile(
     r"(?<![a-z0-9_])(?:nosuchkey"
@@ -1423,19 +1429,43 @@ def infer_build_id(input_dir: str | Path) -> str | None:
     return None
 
 
-def _read_artifact(source_url: str) -> tuple[bytes, str]:
+def _read_artifact(source_url: str, local_path: Path) -> tuple[str, int]:
+    """Copy a source artifact to ``local_path`` a chunk at a time, hashing it.
+
+    The bytes go to a ``.partial`` file that takes the final name only once the
+    copy is complete, so a failed download leaves any earlier copy in place.
+    """
+    parsed = urlparse(source_url)
+    if parsed.scheme not in ("http", "https", "file", ""):
+        raise ValueError(f"Unsupported source URL scheme: {parsed.scheme}")
+    partial_path = local_path.with_name(f"{local_path.name}.partial")
+    digest = hashlib.sha256()
+    size_bytes = 0
+    try:
+        with partial_path.open("wb") as output:
+            for chunk in _artifact_chunks(source_url):
+                output.write(chunk)
+                digest.update(chunk)
+                size_bytes += len(chunk)
+        partial_path.replace(local_path)
+    except BaseException:
+        partial_path.unlink(missing_ok=True)
+        raise
+    return digest.hexdigest(), size_bytes
+
+
+def _artifact_chunks(source_url: str) -> Iterator[bytes]:
     parsed = urlparse(source_url)
     if parsed.scheme in ("http", "https"):
-        response = httpx.get(source_url, follow_redirects=True, timeout=60)
-        response.raise_for_status()
-        return response.content, _filename_from_url(source_url)
-    if parsed.scheme == "file":
-        path = Path(unquote(parsed.path))
-        return path.read_bytes(), path.name
-    if not parsed.scheme:
-        path = Path(source_url)
-        return path.read_bytes(), path.name
-    raise ValueError(f"Unsupported source URL scheme: {parsed.scheme}")
+        with httpx.stream(
+            "GET", source_url, follow_redirects=True, timeout=60
+        ) as response:
+            response.raise_for_status()
+            yield from response.iter_bytes(1024 * 1024)
+        return
+    path = Path(unquote(parsed.path)) if parsed.scheme == "file" else Path(source_url)
+    with path.open("rb") as stream:
+        yield from iter(lambda: stream.read(1024 * 1024), b"")
 
 
 def _filename_from_url(source_url: str) -> str:
@@ -1492,6 +1522,8 @@ def _upload_r2_object(
     wrangler_command: str,
 ) -> ArtifactCommandResult:
     content_type, _ = mimetypes.guess_type(local_path.name)
+    if local_path.stat().st_size > WRANGLER_PUT_LIMIT_BYTES:
+        return _upload_r2_object_in_parts(location, local_path, content_type)
     command = [
         *shlex.split(wrangler_command),
         "r2",
@@ -1507,6 +1539,61 @@ def _upload_r2_object(
     return _run_command(command)
 
 
+def _upload_r2_object_in_parts(
+    location: ArtifactStorageLocation,
+    local_path: Path,
+    content_type: str | None,
+) -> ArtifactCommandResult:
+    """Upload a file too large for Wrangler through R2's S3 API (chronicle#409).
+
+    The key carries the file's sha256, so an object of the file's size under it
+    already holds these bytes. Such a file is not sent again, and a failed
+    upload, such as a completion whose response was lost, counts once the
+    object is there.
+    """
+    command = (
+        "r2-s3-multipart-put",
+        f"{location.bucket}/{location.key}",
+        str(local_path),
+    )
+    try:
+        size_bytes = local_path.stat().st_size
+        with _r2_s3_client() as client:
+            if client.object_size(location.bucket, location.key) == size_bytes:
+                return ArtifactCommandResult(
+                    command=command, returncode=0, stdout="already in R2", stderr=""
+                )
+            try:
+                parts = client.put_file(
+                    location.bucket, location.key, local_path, content_type=content_type
+                )
+            except r2_s3.R2S3Error as error:
+                try:
+                    landed = client.object_size(location.bucket, location.key)
+                except r2_s3.R2S3Error:
+                    landed = None
+                if landed != size_bytes:
+                    raise error
+                return ArtifactCommandResult(
+                    command=command,
+                    returncode=0,
+                    stdout=f"in R2 although the upload reported: {error}",
+                    stderr="",
+                )
+    except (OSError, r2_s3.R2S3Error) as exc:
+        return ArtifactCommandResult(
+            command=command, returncode=1, stdout="", stderr=str(exc)
+        )
+    return ArtifactCommandResult(
+        command=command, returncode=0, stdout=f"uploaded in {parts} parts", stderr=""
+    )
+
+
+def _r2_s3_client() -> r2_s3.R2Client:
+    """Return the S3 API client for large uploads; tests substitute their own."""
+    return r2_s3.R2Client(r2_s3.credentials_from_environment())
+
+
 def _r2_only_storage(spec: dict[str, Any]) -> tuple[bool, list[str]]:
     """Return whether a file entry keeps its bytes in R2 only, and any issues.
 
@@ -1578,9 +1665,7 @@ def _publish_raw_manifest_entry(
     elif not artifact_path.exists():
         errors.append("missing_file")
     else:
-        content = artifact_path.read_bytes()
-        sha256_actual = hashlib.sha256(content).hexdigest()
-        size_bytes = len(content)
+        sha256_actual, size_bytes = _hash_file(artifact_path)
         if sha256_expected and sha256_actual != sha256_expected:
             errors.append("checksum_mismatch")
 
@@ -1702,9 +1787,7 @@ def _inventory_entry(
         if not r2_only:
             errors.append("missing_file")
     else:
-        content = artifact_path.read_bytes()
-        sha256_actual = hashlib.sha256(content).hexdigest()
-        size_bytes = len(content)
+        sha256_actual, size_bytes = _hash_file(artifact_path)
         if sha256_expected and sha256_actual != sha256_expected:
             errors.append("checksum_mismatch")
     storage = spec.get("storage") if isinstance(spec, dict) else None
diff --git a/chronicle/r2_s3.py b/chronicle/r2_s3.py
new file mode 100644
index 00000000..b1642610
--- /dev/null
+++ b/chronicle/r2_s3.py
@@ -0,0 +1,409 @@
+"""Upload large R2 objects through R2's S3-compatible API (chronicle#409).
+
+``wrangler r2 object put`` refuses objects above 300 MiB, so Chronicle sends
+larger files here. R2 accepts requests signed with AWS Signature Version 4 in
+region ``auto``. A file goes up as a multipart upload in fixed-size parts: each
+part is retried on its own, and the upload is aborted if a part still fails, so
+no partial object is left under the key.
+
+Credentials come from ``R2_ACCESS_KEY_ID`` and ``R2_SECRET_ACCESS_KEY`` when
+both are set, and otherwise from the ``CLOUDFLARE_API_TOKEN`` that Wrangler
+already uses: Cloudflare's R2 token documentation gives that token's S3 key ID
+as the token's ID and its secret as the SHA-256 of the token's value. The
+account is ``CLOUDFLARE_ACCOUNT_ID``, or else the ``account_id`` that
+``wrangler.toml`` pins.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+import gc
+import hashlib
+import hmac
+import os
+from pathlib import Path
+import time
+import tomllib
+from types import TracebackType
+from urllib.parse import quote
+from xml.etree import ElementTree
+from xml.sax.saxutils import escape
+
+import httpx
+
+
+#: R2 takes up to 10,000 parts per upload, so 100 MiB parts reach 976 GiB.
+DEFAULT_PART_SIZE_BYTES = 100 * 1024 * 1024
+CLOUDFLARE_API_URL = "https://api.cloudflare.com/client/v4"
+_RETRY_STATUSES = frozenset({429, 500, 502, 503, 504})
+
+
+class R2S3Error(RuntimeError):
+    """An R2 S3 API request or credential lookup failed."""
+
+
+@dataclass(frozen=True)
+class R2Credentials:
+    """An S3 API key pair for one Cloudflare account's R2 buckets."""
+
+    account_id: str
+    access_key_id: str
+    secret_access_key: str = field(repr=False)
+
+    @property
+    def host(self) -> str:
+        """Return the account's S3 API host."""
+        return f"{self.account_id}.r2.cloudflarestorage.com"
+
+
+def credentials_from_environment(
+    environ: Mapping[str, str] | None = None,
+    *,
+    wrangler_toml: str | Path = "wrangler.toml",
+    transport: httpx.BaseTransport | None = None,
+) -> R2Credentials:
+    """Return R2 S3 credentials from the environment (see the module docstring)."""
+    environ = os.environ if environ is None else environ
+    account_id = environ.get("CLOUDFLARE_ACCOUNT_ID") or _wrangler_account_id(
+        Path(wrangler_toml)
+    )
+    if not account_id:
+        raise R2S3Error(
+            "No Cloudflare account: set CLOUDFLARE_ACCOUNT_ID, or run from the "
+            "directory whose wrangler.toml pins account_id."
+        )
+    access_key_id = environ.get("R2_ACCESS_KEY_ID")
+    secret_access_key = environ.get("R2_SECRET_ACCESS_KEY")
+    if access_key_id and secret_access_key:
+        return R2Credentials(account_id, access_key_id, secret_access_key)
+    token = environ.get("CLOUDFLARE_API_TOKEN")
+    if not token:
+        raise R2S3Error(
+            "No R2 credentials: set R2_ACCESS_KEY_ID and R2_SECRET_ACCESS_KEY, "
+            "or CLOUDFLARE_API_TOKEN."
+        )
+    return R2Credentials(
+        account_id,
+        _api_token_id(token, account_id, transport=transport),
+        hashlib.sha256(token.encode("utf-8")).hexdigest(),
+    )
+
+
+def sign_v4(
+    method: str,
+    *,
+    host: str,
+    path: str,
+    query: Mapping[str, str],
+    headers: Mapping[str, str],
+    payload_sha256: str,
+    access_key_id: str,
+    secret_access_key: str,
+    now: datetime,
+    region: str = "auto",
+) -> dict[str, str]:
+    """Return a request's headers with its AWS Signature Version 4.
+
+    ``path`` is the request path already URI-encoded, as it is sent. Every
+    header returned is signed: ``host``, ``x-amz-content-sha256``,
+    ``x-amz-date`` and the ``headers`` given.
+    """
+    amz_date = now.astimezone(UTC).strftime("%Y%m%dT%H%M%SZ")
+    signed = {
+        **{name.lower(): " ".join(value.split()) for name, value in headers.items()},
+        "host": host,
+        "x-amz-content-sha256": payload_sha256,
+        "x-amz-date": amz_date,
+    }
+    names = sorted(signed)
+    signed_names = ";".join(names)
+    canonical_request = "\n".join(
+        [
+            method,
+            path,
+            canonical_query(query),
+            "".join(f"{name}:{signed[name]}\n" for name in names),
+            signed_names,
+            payload_sha256,
+        ]
+    )
+    scope = f"{amz_date[:8]}/{region}/s3/aws4_request"
+    string_to_sign = "\n".join(
+        [
+            "AWS4-HMAC-SHA256",
+            amz_date,
+            scope,
+            hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),
+        ]
+    )
+    key = f"AWS4{secret_access_key}".encode("utf-8")
+    for part in (amz_date[:8], region, "s3", "aws4_request"):
+        key = hmac.new(key, part.encode("utf-8"), hashlib.sha256).digest()
+    signature = hmac.new(
+        key, string_to_sign.encode("utf-8"), hashlib.sha256
+    ).hexdigest()
+    return {
+        **signed,
+        "authorization": (
+            f"AWS4-HMAC-SHA256 Credential={access_key_id}/{scope}, "
+            f"SignedHeaders={signed_names}, Signature={signature}"
+        ),
+    }
+
+
+def canonical_query(query: Mapping[str, str]) -> str:
+    """Return a query string in Signature Version 4's canonical form."""
+    return "&".join(
+        f"{_uri_encode(name)}={_uri_encode(value)}"
+        for name, value in sorted(query.items())
+    )
+
+
+class R2Client:
+    """Signed S3 API requests to one Cloudflare account's R2 buckets."""
+
+    def __init__(
+        self,
+        credentials: R2Credentials,
+        *,
+        transport: httpx.BaseTransport | None = None,
+        part_size: int = DEFAULT_PART_SIZE_BYTES,
+        attempts: int = 4,
+        clock: Callable[[], datetime] | None = None,
+        sleep: Callable[[float], None] = time.sleep,
+    ) -> None:
+        if part_size <= 0 or attempts <= 0:
+            raise ValueError("part_size and attempts must be positive")
+        self.credentials = credentials
+        self.part_size = part_size
+        self.attempts = attempts
+        self._clock = clock or (lambda: datetime.now(UTC))
+        self._sleep = sleep
+        self._http = httpx.Client(
+            transport=transport,
+            timeout=httpx.Timeout(60.0, read=600.0, write=600.0),
+        )
+
+    def __enter__(self) -> R2Client:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        self.close()
+
+    def close(self) -> None:
+        """Close the HTTP connection pool."""
+        self._http.close()
+
+    def put_file(
+        self,
+        bucket: str,
+        key: str,
+        path: str | Path,
+        *,
+        content_type: str | None = None,
+    ) -> int:
+        """Upload a file as a multipart upload and return its part count.
+
+        Parts are read one at a time, so memory holds one part, not the file.
+        """
+        path = Path(path)
+        if path.stat().st_size == 0:
+            raise R2S3Error(f"A multipart upload needs a non-empty file: {path}")
+        upload_id = self._create_upload(bucket, key, content_type)
+        parts: list[tuple[int, str]] = []
+        try:
+            with path.open("rb") as stream:
+                while chunk := stream.read(self.part_size):
+                    number = len(parts) + 1
+                    parts.append(
+                        (
+                            number,
+                            self._upload_part(bucket, key, upload_id, number, chunk),
+                        )
+                    )
+                    # httpx keeps each response in a reference cycle with its
+                    # request, which holds the part it sent. The cycle
+                    # collector can leave those for many parts, and so most of
+                    # a file in memory, so free the part before the next read.
+                    del chunk
+                    gc.collect()
+            self._complete_upload(bucket, key, upload_id, parts)
+        except BaseException:
+            self._abort_upload(bucket, key, upload_id)
+            raise
+        return len(parts)
+
+    def object_size(self, bucket: str, key: str) -> int | None:
+        """Return the size in bytes of the object under ``key``, or None if absent."""
+        response = self._request("HEAD", bucket, key, {}, missing_ok=True)
+        if response.status_code == 404:
+            return None
+        size = response.headers.get("content-length")
+        if size is None:
+            raise R2S3Error(f"HEAD {key} returned no Content-Length")
+        return int(size)
+
+    def _create_upload(self, bucket: str, key: str, content_type: str | None) -> str:
+        response = self._request(
+            "POST",
+            bucket,
+            key,
+            {"uploads": ""},
+            headers={"content-type": content_type} if content_type else {},
+        )
+        upload_id = _xml_text(response, "UploadId")
+        if not upload_id:
+            raise R2S3Error(f"CreateMultipartUpload for {key} returned no UploadId")
+        return upload_id
+
+    def _upload_part(
+        self, bucket: str, key: str, upload_id: str, number: int, chunk: bytes
+    ) -> str:
+        response = self._request(
+            "PUT",
+            bucket,
+            key,
+            {"partNumber": str(number), "uploadId": upload_id},
+            body=chunk,
+        )
+        etag = response.headers.get("etag")
+        if not etag:
+            raise R2S3Error(f"UploadPart {number} of {key} returned no ETag")
+        return etag
+
+    def _complete_upload(
+        self, bucket: str, key: str, upload_id: str, parts: list[tuple[int, str]]
+    ) -> None:
+        body = "".join(
+            f"<Part><PartNumber>{number}</PartNumber><ETag>{escape(etag)}</ETag></Part>"
+            for number, etag in parts
+        )
+        response = self._request(
+            "POST",
+            bucket,
+            key,
+            {"uploadId": upload_id},
+            body=f"<CompleteMultipartUpload>{body}</CompleteMultipartUpload>".encode(
+                "utf-8"
+            ),
+            headers={"content-type": "application/xml"},
+        )
+        # S3 can report a failed completion inside a 200 response.
+        if _xml_root_tag(response) == "Error":
+            raise R2S3Error(
+                f"CompleteMultipartUpload for {key} failed: {response.text[:300]}"
+            )
+
+    def _abort_upload(self, bucket: str, key: str, upload_id: str) -> None:
+        try:
+            self._request("DELETE", bucket, key, {"uploadId": upload_id})
+        except R2S3Error:
+            # The caller's error matters more; R2 expires abandoned uploads.
+            pass
+
+    def _request(
+        self,
+        method: str,
+        bucket: str,
+        key: str,
+        query: Mapping[str, str],
+        *,
+        body: bytes = b"",
+        headers: Mapping[str, str] | None = None,
+        missing_ok: bool = False,
+    ) -> httpx.Response:
+        path = f"/{_uri_encode(bucket)}/{_uri_encode(key, keep_slash=True)}"
+        url = f"https://{self.credentials.host}{path}"
+        if query:
+            url = f"{url}?{canonical_query(query)}"
+        payload_sha256 = hashlib.sha256(body).hexdigest()
+        for attempt in range(1, self.attempts + 1):
+            signed = sign_v4(
+                method,
+                host=self.credentials.host,
+                path=path,
+                query=query,
+                headers=headers or {},
+                payload_sha256=payload_sha256,
+                access_key_id=self.credentials.access_key_id,
+                secret_access_key=self.credentials.secret_access_key,
+                now=self._clock(),
+            )
+            try:
+                response = self._http.request(method, url, content=body, headers=signed)
+            except httpx.TransportError as exc:
+                if attempt == self.attempts:
+                    raise R2S3Error(f"{method} {key} failed: {exc}") from exc
+            else:
+                if response.is_success or (missing_ok and response.status_code == 404):
+                    return response
+                if (
+                    response.status_code not in _RETRY_STATUSES
+                    or attempt == self.attempts
+                ):
+                    raise R2S3Error(
+                        f"{method} {key} returned HTTP {response.status_code}: "
+                        f"{response.text[:300]}"
+                    )
+            self._sleep(2.0**attempt)
+        raise AssertionError("unreachable")
+
+
+def _api_token_id(
+    token: str, account_id: str, *, transport: httpx.BaseTransport | None
+) -> str:
+    """Return an API token's ID: user tokens and account tokens verify apart."""
+    try:
+        with httpx.Client(transport=transport, timeout=30.0) as client:
+            for url in (
+                f"{CLOUDFLARE_API_URL}/user/tokens/verify",
+                f"{CLOUDFLARE_API_URL}/accounts/{account_id}/tokens/verify",
+            ):
+                response = client.get(url, headers={"Authorization": f"Bearer {token}"})
+                if response.status_code == 200:
+                    token_id = (response.json().get("result") or {}).get("id")
+                    if token_id:
+                        return str(token_id)
+    except (httpx.HTTPError, ValueError) as exc:
+        raise R2S3Error(f"Could not verify CLOUDFLARE_API_TOKEN: {exc}") from exc
+    raise R2S3Error("CLOUDFLARE_API_TOKEN did not verify, so its R2 key ID is unknown.")
+
+
+def _wrangler_account_id(path: Path) -> str | None:
+    try:
+        payload = tomllib.loads(path.read_text(encoding="utf-8"))
+    except (OSError, tomllib.TOMLDecodeError):
+        return None
+    account_id = payload.get("account_id")
+    return account_id if isinstance(account_id, str) and account_id else None
+
+
+def _uri_encode(value: str, *, keep_slash: bool = False) -> str:
+    return quote(value, safe="-_.~/" if keep_slash else "-_.~")
+
+
+def _xml_text(response: httpx.Response, tag: str) -> str | None:
+    try:
+        root = ElementTree.fromstring(response.content)
+    except ElementTree.ParseError:
+        return None
+    # ``{*}`` matches the tag in S3's namespace or in none.
+    element = root.find(f".//{{*}}{tag}")
+    return element.text if element is not None else None
+
+
+def _xml_root_tag(response: httpx.Response) -> str | None:
+    if not response.content.strip():
+        return None
+    try:
+        tag = ElementTree.fromstring(response.content).tag
+    except ElementTree.ParseError:
+        return None
+    return tag.rsplit("}", 1)[-1]
diff --git a/db/data/ucl/house_price_per_square_metre_856542/manifest.yaml b/db/data/ucl/house_price_per_square_metre_856542/manifest.yaml
new file mode 100644
index 00000000..e4392f25
--- /dev/null
+++ b/db/data/ucl/house_price_per_square_metre_856542/manifest.yaml
@@ -0,0 +1,65 @@
+source_id: ucl
+package_id: ucl-house-price-per-square-metre-856542
+dataset: ucl_ucl-house-price-per-square-metre-856542
+source_name: House Price Per Square Metre in England and Wales, 1995-2023
+publisher: University College London, deposited in UK Data Service ReShare
+source_page: https://reshare.ukdataservice.ac.uk/856542/
+table: UK Data Service ReShare 856542, bundles 1 and 2
+license: Creative Commons Attribution 4.0 International (CC BY 4.0)
+license_url: https://creativecommons.org/licenses/by/4.0/
+attribution: Chi, B., Dennett, A., Oléron-Evans, T. and Morphet, R. (2025). House Price Per Square
+  Metre in England and Wales, 1995-2023. UK Data Service ReShare. DOI 10.5255/UKDA-SN-856542
+license_evidence: 'ReShare''s record for 856542 states Creative Commons Attribution 4.0 International
+  (CC BY 4.0). UCL''s Licence_Information_for_data_sources.docx in bundle 1 adds that tranall_link_02122023.zip,
+  LRPPD_02122023.zip and NSPL_NOV_2023_UK.zip contain Royal Mail copyright postcode and address data,
+  limited to non-commercial use, with all non-address data under the Open Government Licence v3.0.
+  It quotes the EPC terms, which say the data contains personal data and that Royal Mail''s copyright
+  notice must go with any address data that is shared, and the Price Paid terms. So the bytes stay
+  out of this public repository: the zips are git-ignored and the private ledger-raw bucket holds
+  them, bundle 1 with that notice.'
+license_verified_at: '2026-11-06'
+files:
+  bundle_1:
+    filename: 856542_bundle_1.zip
+    source_url: https://reshare.ukdataservice.ac.uk/856542/1/856542_bundle_1.zip
+    source_table: 'ReShare 856542 bundle 1: the linked file tranall_link_02122023.zip, with UCL''s
+      readme and licence note'
+    sha256: e397cdac214b918a39f3ccf5266adaf9e4c3d8fb4f71e0bdc1b9c9474734b296
+    size_bytes: 3214161126
+    fetched_at: '2026-11-06T11:17:39+00:00'
+    storage:
+      local: false
+      r2:
+        provider: r2
+        bucket: ledger-raw
+        key: raw/uk/ucl/ucl-house-price-per-square-metre-856542/bundle_1/e397cdac214b918a39f3ccf5266adaf9e4c3d8fb4f71e0bdc1b9c9474734b296/856542_bundle_1.zip
+        uri: r2://ledger-raw/raw/uk/ucl/ucl-house-price-per-square-metre-856542/bundle_1/e397cdac214b918a39f3ccf5266adaf9e4c3d8fb4f71e0bdc1b9c9474734b296/856542_bundle_1.zip
+  bundle_2:
+    filename: 856542_bundle_2.zip
+    source_url: https://reshare.ukdataservice.ac.uk/856542/2/856542_bundle_2.zip
+    source_table: 'ReShare 856542 bundle 2: the files UCL linked, epc20_id.zip (domestic EPCs with
+      addresses removed), LRPPD_02122023.zip (Price Paid as at 2 December 2023) and NSPL_NOV_2023_UK.zip'
+    sha256: 61cbf6a72e35f7569d9121d80eaf89e757fec3bd4c21f1175e37fff44e02da97
+    size_bytes: 4426805029
+    fetched_at: '2026-11-06T11:14:45+00:00'
+    storage:
+      local: false
+      r2:
+        provider: r2
+        bucket: ledger-raw
+        key: raw/uk/ucl/ucl-house-price-per-square-metre-856542/bundle_2/61cbf6a72e35f7569d9121d80eaf89e757fec3bd4c21f1175e37fff44e02da97/856542_bundle_2.zip
+        uri: r2://ledger-raw/raw/uk/ucl/ucl-house-price-per-square-metre-856542/bundle_2/61cbf6a72e35f7569d9121d80eaf89e757fec3bd4c21f1175e37fff44e02da97/856542_bundle_2.zip
+registration_kind: raw_only
+provenance_notes: 'UK Data Service ReShare 856542 (DOI 10.5255/UKDA-SN-856542), published on 22 August
+  2025 by Bin Chi, Adam Dennett, Thomas Oléron-Evans and Robin Morphet of UCL, the latest deposit
+  of the series 854240, 855033, 856204 and 856542. Bundle 1 holds the linked file tranall_link_02122023.zip
+  (3,214,108,822 bytes, stored), which UCL''s readme describes as 21,438,854 transactions in England
+  and Wales from 1 January 1995 to 31 October 2023 with 107 fields, plus the readme and the licence
+  note. ReShare''s record gives the temporal coverage as 1 January 1995 to 23 October 2023, while
+  its abstract, like the readme, says 31 October 2023. Bundle 2 holds the files UCL linked: the domestic
+  EPCs downloaded on 2 December 2023 with their address fields removed (epc20_id.zip), Price Paid
+  with field names as at 2 December 2023 (LRPPD_02122023.zip) and the November 2023 NSPL (NSPL_NOV_2023_UK.zip).
+  ReShare''s ETags equal the MD5s of the bytes fetched on 6 November 2026 (239aebb08e10939b7cba4c5b2d5f89c9
+  and 5fa5f769f3061aef86c455a4ed23c97e), and both bundles were last modified on 22 August 2025. Raw
+  only: no build parses either bundle (chronicle#408, item 6). Both are above the 300 MiB that wrangler
+  r2 object put accepts, so publish-raw uploads them through R2''s S3 API in 100 MiB parts (chronicle#409).'
diff --git a/docs/agent-source-package-harness.md b/docs/agent-source-package-harness.md
index 261c1dd5..de1e56c3 100644
--- a/docs/agent-source-package-harness.md
+++ b/docs/agent-source-package-harness.md
@@ -19,7 +19,13 @@ metadata in `manifest.yaml`, and can upload the exact bytes to the private
 artifact registry with `uv run chronicle inventory-artifacts --root db/data`.
 For already-downloaded manifest artifacts, agents should run
 `uv run chronicle publish-raw --root db/data` to upload checksum-verified bytes to
-R2 and write `storage.r2` metadata back into each manifest entry. A manifest
+R2 and write `storage.r2` metadata back into each manifest entry. A file above the
+300 MiB that `wrangler r2 object put` accepts goes through R2's S3-compatible API
+instead, as a multipart upload in 100 MiB parts
+([chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409)). That path
+signs with `R2_ACCESS_KEY_ID` and `R2_SECRET_ACCESS_KEY` when both are set, and
+otherwise with the S3 key pair Cloudflare derives from `CLOUDFLARE_API_TOKEN`. It
+takes the account from `CLOUDFLARE_ACCOUNT_ID` or `wrangler.toml`. A manifest
 entry marked `storage.local: false` keeps its bytes in R2 only and never in Git:
 both commands accept it without a local copy, so its absence is not an error to
 fix by downloading, and `uv run chronicle verify-raw` checks the R2 object.
diff --git a/docs/data-sources/uk-high-value-housing.md b/docs/data-sources/uk-high-value-housing.md
index bc7af85c..ade857d1 100644
--- a/docs/data-sources/uk-high-value-housing.md
+++ b/docs/data-sources/uk-high-value-housing.md
@@ -72,9 +72,11 @@ Each of their file entries says so with `storage.local: false`:
 `.gitignore` keeps the local copies out of commits.
 
 Price Paid is registered as HM Land Registry's yearly files, not its single
-`pp-complete.csv` (5,525,539,417 bytes). `publish-raw` uploads through
+`pp-complete.csv` (5,525,539,417 bytes). `publish-raw` then uploaded only through
 `wrangler r2 object put`, which refuses files above 300 MiB; the largest yearly
-file (`pp-2002.csv`) is 239,344,429 bytes. The 32 yearly files of a release hold
+file (`pp-2002.csv`) is 239,344,429 bytes. Larger files now go through R2's S3
+API ([chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409)), but
+the yearly files stay: they are the release's own split. The 32 yearly files of a release hold
 exactly the complete file's rows, split by year of transfer. For this release,
 order-independent row digests by year agree between the two (31,610,742 rows).
 
diff --git a/docs/data-sources/uk-land-value-inputs-408.md b/docs/data-sources/uk-land-value-inputs-408.md
index 29a6a5e6..1846232c 100644
--- a/docs/data-sources/uk-land-value-inputs-408.md
+++ b/docs/data-sources/uk-land-value-inputs-408.md
@@ -396,13 +396,46 @@ These rows give Microcosm council tax by authority for all three nations, which
 - **Table 9a against Tables 9d and 2e.** 9a's 155,400 claims equal 9d's total. 2e's quarters
   from 2024 Q2 to 2025 Q1 sum to 155,368.45.
 
-## 6. UCL's house price per square metre: not in this PR
-
-ReShare 856542's two bundles are 3,214,161,126 and 4,426,805,029 bytes. Each holds stored
-nested zips, the linked file `tranall_link_02122023.zip` alone being 3.2 GB. `publish-raw`
-uploads through `wrangler r2 object put`, which refuses objects over 300 MiB. Item 6 therefore
-waits for a separate change to how Chronicle stores large raw files, which
-[chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409) asks for a ruling on.
+## 6. UCL's house price per square metre: both bundles, raw-only
+
+**`ucl-house-price-per-square-metre-856542`** registers
+[UK Data Service ReShare 856542](https://reshare.ukdataservice.ac.uk/856542/), *House Price
+Per Square Metre in England and Wales, 1995-2023*. It is by Bin Chi, Adam Dennett, Thomas
+Oléron-Evans and Robin Morphet (UCL), published on 22 August 2025 under CC BY 4.0. No build
+parses it. It came after the rest of this issue, once Chronicle could store files above
+300 MiB ([chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409)).
+
+- **Bundle 1**, `856542_bundle_1.zip`: 3,214,161,126 bytes, sha256 `e397cdac…34b296`.
+  - It holds the linked file `tranall_link_02122023.zip` (3,214,108,822 bytes, stored), with
+    UCL's readme and licence note.
+  - The readme describes the linked file as 21,438,854 transactions in England and Wales from
+    1 January 1995 to 31 October 2023, with 107 fields.
+  - ReShare's record gives the temporal coverage as 1 January 1995 to 23 October 2023, while
+    its abstract, like the readme, says 31 October 2023.
+- **Bundle 2**, `856542_bundle_2.zip`: 4,426,805,029 bytes, sha256 `61cbf6a7…02da97`. It
+  holds the files UCL linked:
+  - the domestic EPCs downloaded on 2 December 2023, with their address fields removed
+    (`epc20_id.zip`);
+  - Price Paid with field names as at 2 December 2023 (`LRPPD_02122023.zip`);
+  - the November 2023 NSPL (`NSPL_NOV_2023_UK.zip`).
+- ReShare's ETags equal the MD5s of the bytes fetched on 6 November 2026, and both bundles
+  were last modified on 22 August 2025.
+
+**Licence.** ReShare states CC BY 4.0. UCL's licence note adds that the linked file, Price
+Paid and the NSPL carry Royal Mail address data, limited to non-commercial use, with
+everything else under the Open Government Licence v3.0. It also quotes the EPC terms: the data
+contains personal data, and Royal Mail's notice must go with any address data that is shared.
+As with Price Paid, the zips are git-ignored and the private `ledger-raw` bucket holds them.
+
+**Storage.** Both bundles are above the 300 MiB that `wrangler r2 object put` accepts, so
+`publish-raw` sent them through R2's S3 API, in 31 and 43 parts of 100 MiB. R2's ETags for
+them, `"7eb31fba…44b1df-31"` and `"d04abcef…798c87-43"`, are the MD5 of the parts' MD5s,
+which can be recomputed from the pinned bytes without downloading the objects. `verify-raw`
+reads them back in full through `wrangler r2 object get --pipe`, which streams.
+
+Microcosm needs bundle 1's linked file for each sale's floor area; bundle 2 keeps what UCL
+linked from. The London Datastore's 2024 edition is not registered, because its record states
+no licence.
 
 ## 7. Floor area and dwelling density by authority
 
diff --git a/docs/pe-uk-source-checklist.md b/docs/pe-uk-source-checklist.md
index e403b860..4b7666d1 100644
--- a/docs/pe-uk-source-checklist.md
+++ b/docs/pe-uk-source-checklist.md
@@ -457,7 +457,9 @@ facts; field contracts and caveats are in
 
 Published inputs for Microcosm's land value work
 ([PolicyEngine/uk-land-value-tax](https://github.com/PolicyEngine/uk-land-value-tax)). Eight of
-the issue's nine items are here: about 326,500 new facts, plus three raw-only registrations.
+the issue's nine items came first: about 326,500 new facts, plus three raw-only registrations.
+The ninth, UCL's house price per square metre, followed once `publish-raw` could upload files
+above 300 MiB ([chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409)).
 Field contracts and caveats are in
 [`docs/data-sources/uk-land-value-inputs-408.md`](data-sources/uk-land-value-inputs-408.md).
 
@@ -484,8 +486,8 @@ Field contracts and caveats are in
   `hmrc-stamp-tax-statistics-2024-25` (1,135 facts).
 - **Tax Policy Associates' model.** Three files pinned raw at one commit, with no facts
   (`tpa-lvt-model-2026`).
-- **Not in this PR.** UCL's house price per square metre file (item 6) is above the 300 MiB that
-  `publish-raw` can upload ([chronicle#409](https://github.com/PolicyEngine/chronicle/issues/409)).
+- **UCL's house price per square metre (item 6).** UK Data Service ReShare 856542's two
+  bundles, 3.2 and 4.4 GB, pinned raw with no facts (`ucl-house-price-per-square-metre-856542`).
 
 ## Consumer contracts
 
diff --git a/tests/test_chronicle_artifacts.py b/tests/test_chronicle_artifacts.py
index ff407ec5..733aad6d 100644
--- a/tests/test_chronicle_artifacts.py
+++ b/tests/test_chronicle_artifacts.py
@@ -13,6 +13,9 @@
 from chronicle.cli import main as cli_main
 from chronicle.artifacts import (
     R2_COUNTRY_PUBLISHERS,
+    WRANGLER_PUT_LIMIT_BYTES,
+    ArtifactStorageLocation,
+    _upload_r2_object,
     build_artifact_key,
     build_artifact_rows,
     build_derived_r2_key,
@@ -28,6 +31,7 @@
 from chronicle.bundle import NZ_BUNDLE_SOURCE_PREFIXES, NZ_BUNDLE_SOURCES
 from chronicle.epoch import Epoch
 from chronicle.harness import main as harness_main
+from chronicle.r2_s3 import R2S3Error
 from chronicle.source_package import load_source_package
 
 
@@ -677,6 +681,213 @@ def test_publish_raw_refuses_r2_only_bytes_without_an_r2_location(tmp_path):
     assert not log.exists()
 
 
+def _fetch_release(tmp_path: Path, content: bytes) -> Path:
+    source = tmp_path / "release.csv"
+    source.write_bytes(content)
+    output_dir = tmp_path / "data" / "hmlr" / "release"
+    fetch_source_artifact(
+        str(source),
+        source_id="hmlr",
+        package_id="hmlr-release",
+        year=2026,
+        output_dir=output_dir,
+    )
+    return output_dir
+
+
+def test_fetch_publish_and_inventory_hash_files_without_reading_them_whole(
+    tmp_path, monkeypatch
+):
+    # Raw files can be several GB (chronicle#409), so none is read whole.
+    def refuse_whole_read(path):
+        raise AssertionError(f"read whole: {path}")
+
+    monkeypatch.setattr(Path, "read_bytes", refuse_whole_read)
+    wrangler, _ = _logging_wrangler(tmp_path)
+    output_dir = _fetch_release(tmp_path, b"raw release bytes")
+
+    published = publish_source_artifacts(output_dir, wrangler_command=str(wrangler))
+    inventory = inventory_source_artifacts(output_dir)
+
+    assert published.valid and inventory.valid
+    (entry,) = published.entries
+    assert entry.sha256 == hashlib.sha256(b"raw release bytes").hexdigest()
+    assert entry.size_bytes == len(b"raw release bytes")
+    assert inventory.entries[0].sha256_actual == entry.sha256
+
+
+class _RecordingS3Client:
+    uploads: list[tuple[str, str, bytes, str | None]] = []
+    error: Exception | None = None
+    # What HEAD finds under the key before an upload, and after one that failed.
+    existing_size: int | Exception | None = None
+    size_after_error: int | Exception | None = None
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc_info):
+        return None
+
+    def object_size(self, bucket, key):
+        if isinstance(self.existing_size, Exception):
+            raise self.existing_size
+        return self.existing_size
+
+    def put_file(self, bucket, key, path, *, content_type=None):
+        if self.error is not None:
+            type(self).existing_size = self.size_after_error
+            raise self.error
+        self.uploads.append((bucket, key, Path(path).read_bytes(), content_type))
+        return 1
+
+
+def test_files_above_the_wrangler_limit_go_through_the_s3_api(tmp_path, monkeypatch):
+    wrangler, log = _logging_wrangler(tmp_path)
+    output_dir = _fetch_release(tmp_path, b"above the limit")
+    monkeypatch.setattr("chronicle.artifacts.WRANGLER_PUT_LIMIT_BYTES", 4)
+    monkeypatch.setattr(_RecordingS3Client, "uploads", [])
+    monkeypatch.setattr("chronicle.artifacts._r2_s3_client", _RecordingS3Client)
+
+    report = publish_source_artifacts(output_dir, wrangler_command=str(wrangler))
+    manifest = yaml.safe_load((output_dir / "manifest.yaml").read_text())
+    r2 = manifest["files"][2026]["storage"]["r2"]
+
+    assert report.valid
+    assert report.counts["uploaded_count"] == 1
+    assert report.entries[0].upload.command[0] == "r2-s3-multipart-put"
+    assert _RecordingS3Client.uploads == [
+        ("ledger-raw", r2["key"], b"above the limit", "text/csv")
+    ]
+    assert r2["key"].startswith("raw/uk/hmlr/hmlr-release/2026/")
+    assert not log.exists()
+
+
+@pytest.mark.parametrize(
+    "size_after_error",
+    [None, R2S3Error("HEAD failed")],
+    ids=["nothing-under-the-key", "head-fails-too"],
+)
+def test_a_failed_s3_upload_records_no_r2_location(
+    tmp_path, monkeypatch, size_after_error
+):
+    wrangler, log = _logging_wrangler(tmp_path)
+    output_dir = _fetch_release(tmp_path, b"above the limit")
+    monkeypatch.setattr("chronicle.artifacts.WRANGLER_PUT_LIMIT_BYTES", 4)
+    monkeypatch.setattr(_RecordingS3Client, "error", R2S3Error("HTTP 403: denied"))
+    monkeypatch.setattr(_RecordingS3Client, "existing_size", None)
+    monkeypatch.setattr(_RecordingS3Client, "size_after_error", size_after_error)
+    monkeypatch.setattr("chronicle.artifacts._r2_s3_client", _RecordingS3Client)
+
+    report = publish_source_artifacts(output_dir, wrangler_command=str(wrangler))
+    manifest = yaml.safe_load((output_dir / "manifest.yaml").read_text())
+
+    assert not report.valid
+    assert report.entries[0].errors == ("r2_upload_failed",)
+    # The upload's own error is reported, not the check that followed it.
+    assert report.entries[0].upload.stderr == "HTTP 403: denied"
+    assert "storage" not in manifest["files"][2026]
+    assert not log.exists()
+
+
+@pytest.mark.parametrize(
+    ("existing_size", "sent"),
+    [(15, False), (14, True)],
+    ids=["same-size", "other-size"],
+)
+def test_an_object_of_the_files_size_under_its_key_is_not_sent_again(
+    tmp_path, monkeypatch, existing_size, sent
+):
+    # The key carries the file's sha256, so an object of its size is the file.
+    wrangler, log = _logging_wrangler(tmp_path)
+    output_dir = _fetch_release(tmp_path, b"above the limit")
+    monkeypatch.setattr("chronicle.artifacts.WRANGLER_PUT_LIMIT_BYTES", 4)
+    monkeypatch.setattr(_RecordingS3Client, "uploads", [])
+    monkeypatch.setattr(_RecordingS3Client, "existing_size", existing_size)
+    monkeypatch.setattr("chronicle.artifacts._r2_s3_client", _RecordingS3Client)
+
+    report = publish_source_artifacts(output_dir, wrangler_command=str(wrangler))
+    manifest = yaml.safe_load((output_dir / "manifest.yaml").read_text())
+    entry = report.entries[0]
+
+    assert report.valid
+    assert manifest["files"][2026]["storage"]["r2"]["key"] == entry.r2_location.key
+    assert len(_RecordingS3Client.uploads) == int(sent)
+    assert (entry.upload.stdout == "already in R2") is not sent
+    assert not log.exists()
+
+
+def test_an_upload_whose_completion_landed_counts_as_uploaded(tmp_path, monkeypatch):
+    # A completion can commit and lose its response: the retry then finds no
+    # upload and put_file raises, though the object is in R2.
+    wrangler, log = _logging_wrangler(tmp_path)
+    output_dir = _fetch_release(tmp_path, b"above the limit")
+    monkeypatch.setattr("chronicle.artifacts.WRANGLER_PUT_LIMIT_BYTES", 4)
+    monkeypatch.setattr(
+        _RecordingS3Client,
+        "error",
+        R2S3Error("POST key returned HTTP 404: <Code>NoSuchUpload</Code>"),
+    )
+    monkeypatch.setattr(_RecordingS3Client, "existing_size", None)
+    monkeypatch.setattr(_RecordingS3Client, "size_after_error", 15)
+    monkeypatch.setattr("chronicle.artifacts._r2_s3_client", _RecordingS3Client)
+
+    report = publish_source_artifacts(output_dir, wrangler_command=str(wrangler))
+    manifest = yaml.safe_load((output_dir / "manifest.yaml").read_text())
+    entry = report.entries[0]
+
+    assert report.valid
+    assert entry.errors == ()
+    assert "NoSuchUpload" in entry.upload.stdout
+    assert manifest["files"][2026]["storage"]["r2"]["key"] == entry.r2_location.key
+    assert not log.exists()
+
+
+def test_the_s3_path_starts_one_byte_above_the_wrangler_limit(tmp_path, monkeypatch):
+    # Wrangler refuses only files above its limit (sizeBytes > 300 MiB).
+    wrangler, log = _logging_wrangler(tmp_path)
+    sent: list[int] = []
+
+    class SizeRecordingClient(_RecordingS3Client):
+        def put_file(self, bucket, key, path, *, content_type=None):
+            sent.append(Path(path).stat().st_size)
+            return 4
+
+    monkeypatch.setattr(_RecordingS3Client, "existing_size", None)
+    monkeypatch.setattr("chronicle.artifacts._r2_s3_client", SizeRecordingClient)
+    location = ArtifactStorageLocation(provider="r2", bucket="ledger-raw", key="raw/k")
+    results = {}
+    for size in (WRANGLER_PUT_LIMIT_BYTES, WRANGLER_PUT_LIMIT_BYTES + 1):
+        path = tmp_path / f"{size}.zip"
+        with path.open("wb") as stream:
+            stream.truncate(size)  # sparse, so nothing is written
+        results[size] = _upload_r2_object(
+            location, path, wrangler_command=str(wrangler)
+        )
+
+    assert WRANGLER_PUT_LIMIT_BYTES == 300 * 1024 * 1024
+    assert results[WRANGLER_PUT_LIMIT_BYTES].command[0] == str(wrangler)
+    assert results[WRANGLER_PUT_LIMIT_BYTES + 1].command[0] == "r2-s3-multipart-put"
+    assert sent == [WRANGLER_PUT_LIMIT_BYTES + 1]
+    assert log.read_text().count("r2 object put") == 1
+
+
+def test_fetch_keeps_the_earlier_copy_when_a_download_fails(tmp_path):
+    output_dir = _fetch_release(tmp_path, b"first copy")
+
+    with pytest.raises(FileNotFoundError):
+        fetch_source_artifact(
+            str(tmp_path / "gone" / "release.csv"),
+            source_id="hmlr",
+            package_id="hmlr-release",
+            year=2026,
+            output_dir=output_dir,
+        )
+
+    assert (output_dir / "release.csv").read_bytes() == b"first copy"
+    assert not (output_dir / "release.csv.partial").exists()
+
+
 def test_artifact_cli_commands_emit_json(tmp_path, capsys):
     source = tmp_path / "source.xls"
     source.write_bytes(b"cli artifact")
diff --git a/tests/test_chronicle_r2_s3.py b/tests/test_chronicle_r2_s3.py
new file mode 100644
index 00000000..6f9acce9
--- /dev/null
+++ b/tests/test_chronicle_r2_s3.py
@@ -0,0 +1,374 @@
+"""Tests for Chronicle's large-file path to R2 through its S3 API (chronicle#409)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+import gc
+import hashlib
+from pathlib import Path
+import tracemalloc
+
+import httpx
+import pytest
+
+from chronicle.r2_s3 import (
+    R2Client,
+    R2Credentials,
+    R2S3Error,
+    canonical_query,
+    credentials_from_environment,
+    sign_v4,
+)
+
+
+EMPTY_SHA256 = hashlib.sha256(b"").hexdigest()
+CREDENTIALS = R2Credentials("account", "key-id", "secret")
+
+
+@pytest.mark.parametrize(
+    ("method", "path", "query", "headers", "payload", "signature"),
+    [
+        # The three Signature Version 4 examples in AWS's S3 API reference
+        # ("Authenticating Requests: Using the Authorization Header").
+        (
+            "GET",
+            "/test.txt",
+            {},
+            {"Range": "bytes=0-9"},
+            b"",
+            "f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41",
+        ),
+        (
+            "GET",
+            "/",
+            {"max-keys": "2", "prefix": "J"},
+            {},
+            b"",
+            "34b48302e7b5fa45bde8084f4b7868a86f0a534bc59db6670ed5711ef69dc6f7",
+        ),
+        (
+            "PUT",
+            "/test%24file.text",
+            {},
+            {
+                "Date": "Fri, 24 May 2013 00:00:00 GMT",
+                "x-amz-storage-class": "REDUCED_REDUNDANCY",
+            },
+            b"Welcome to Amazon S3.",
+            "98ad721746da40c64f1a55b78f14c238d841ea1380cd77a1b5971af0ece108bd",
+        ),
+    ],
+)
+def test_sign_v4_reproduces_the_aws_examples(
+    method, path, query, headers, payload, signature
+):
+    signed = sign_v4(
+        method,
+        host="examplebucket.s3.amazonaws.com",
+        path=path,
+        query=query,
+        headers=headers,
+        payload_sha256=hashlib.sha256(payload).hexdigest(),
+        access_key_id="AKIAIOSFODNN7EXAMPLE",
+        secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+        now=datetime(2013, 5, 24, tzinfo=UTC),
+        region="us-east-1",
+    )
+
+    assert signed["authorization"].endswith(f"Signature={signature}")
+    assert signed["authorization"].startswith(
+        "AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/"
+        "aws4_request, SignedHeaders="
+    )
+
+
+def test_canonical_query_sorts_names_and_encodes_values():
+    assert canonical_query({"uploadId": "a+b/c=", "partNumber": "2"}) == (
+        "partNumber=2&uploadId=a%2Bb%2Fc%3D"
+    )
+    assert canonical_query({"uploads": ""}) == "uploads="
+
+
+class FakeR2:
+    """An in-memory S3 multipart endpoint that records every request."""
+
+    def __init__(self, fail_part: dict[int, list[int]] | None = None):
+        self.requests: list[httpx.Request] = []
+        self.parts: dict[int, bytes] = {}
+        self.fail_part = fail_part or {}
+        self.complete_status = 200
+        self.complete_body = b"<CompleteMultipartUploadResult/>"
+        self.objects: dict[str, int] = {}
+        self.head_status: int | None = None
+
+    def handler(self, request: httpx.Request) -> httpx.Response:
+        self.requests.append(request)
+        query = dict(request.url.params)
+        assert (
+            request.headers["x-amz-content-sha256"]
+            == hashlib.sha256(request.content).hexdigest()
+        )
+        assert request.headers["authorization"].startswith(
+            "AWS4-HMAC-SHA256 Credential=key-id/"
+        )
+        if request.method == "HEAD":
+            if self.head_status is not None:
+                return httpx.Response(self.head_status)
+            key = request.url.path.split("/", 2)[2]
+            if key not in self.objects:
+                return httpx.Response(404)
+            return httpx.Response(
+                200, headers={"Content-Length": str(self.objects[key])}
+            )
+        if request.method == "POST" and "uploads" in query:
+            return httpx.Response(
+                200,
+                content=(
+                    b'<InitiateMultipartUploadResult xmlns="http://s3.amazonaws.com/'
+                    b'doc/2006-03-01/"><UploadId>up+1/=</UploadId>'
+                    b"</InitiateMultipartUploadResult>"
+                ),
+            )
+        if request.method == "PUT":
+            number = int(query["partNumber"])
+            statuses = self.fail_part.get(number, [])
+            if statuses:
+                return httpx.Response(statuses.pop(0))
+            self.parts[number] = request.content
+            return httpx.Response(200, headers={"ETag": f'"etag-{number}"'})
+        if request.method == "POST":
+            return httpx.Response(self.complete_status, content=self.complete_body)
+        if request.method == "DELETE":
+            return httpx.Response(204)
+        raise AssertionError(f"unexpected request {request.method} {request.url}")
+
+    def client(self, **kwargs) -> R2Client:
+        return R2Client(
+            CREDENTIALS,
+            transport=httpx.MockTransport(self.handler),
+            part_size=5,
+            sleep=lambda seconds: None,
+            **kwargs,
+        )
+
+
+def _file(tmp_path: Path, content: bytes) -> Path:
+    path = tmp_path / "bundle.zip"
+    path.write_bytes(content)
+    return path
+
+
+def test_put_file_uploads_fixed_size_parts_and_completes_them_in_order(tmp_path):
+    fake = FakeR2()
+
+    with fake.client() as client:
+        parts = client.put_file(
+            "ledger-raw",
+            "raw/uk/ucl/pkg/bundle_1/digest/bundle.zip",
+            _file(tmp_path, b"abcdefghijkl"),
+            content_type="application/zip",
+        )
+
+    assert parts == 3
+    assert fake.parts == {1: b"abcde", 2: b"fghij", 3: b"kl"}
+    create, *uploads, complete = fake.requests
+    assert create.url.path == "/ledger-raw/raw/uk/ucl/pkg/bundle_1/digest/bundle.zip"
+    assert create.url.query == b"uploads="
+    assert create.headers["content-type"] == "application/zip"
+    assert [request.url.query for request in uploads] == [
+        f"partNumber={number}&uploadId=up%2B1%2F%3D".encode() for number in (1, 2, 3)
+    ]
+    assert complete.url.query == b"uploadId=up%2B1%2F%3D"
+    assert complete.content == (
+        b"<CompleteMultipartUpload>"
+        b'<Part><PartNumber>1</PartNumber><ETag>"etag-1"</ETag></Part>'
+        b'<Part><PartNumber>2</PartNumber><ETag>"etag-2"</ETag></Part>'
+        b'<Part><PartNumber>3</PartNumber><ETag>"etag-3"</ETag></Part>'
+        b"</CompleteMultipartUpload>"
+    )
+
+
+def test_put_file_retries_a_failed_part_on_its_own(tmp_path):
+    fake = FakeR2(fail_part={2: [503, 500]})
+
+    with fake.client() as client:
+        client.put_file("ledger-raw", "key", _file(tmp_path, b"abcdefghijkl"))
+
+    part_numbers = [
+        request.url.params["partNumber"]
+        for request in fake.requests
+        if request.method == "PUT"
+    ]
+    assert part_numbers == ["1", "2", "2", "2", "3"]
+    assert fake.parts == {1: b"abcde", 2: b"fghij", 3: b"kl"}
+
+
+@pytest.mark.parametrize("statuses", [[403], [503, 503, 503, 503]])
+def test_put_file_aborts_the_upload_when_a_part_fails(tmp_path, statuses):
+    fake = FakeR2(fail_part={2: statuses})
+
+    with fake.client() as client, pytest.raises(R2S3Error, match="HTTP"):
+        client.put_file("ledger-raw", "key", _file(tmp_path, b"abcdefghijkl"))
+
+    assert [request.method for request in fake.requests][-1] == "DELETE"
+    assert fake.requests[-1].url.query == b"uploadId=up%2B1%2F%3D"
+    assert 3 not in fake.parts
+    assert not any(
+        request.method == "POST" and "uploadId" in request.url.params
+        for request in fake.requests
+    )
+
+
+def test_put_file_fails_on_an_error_inside_a_successful_completion(tmp_path):
+    fake = FakeR2()
+    fake.complete_body = b"<Error><Code>InternalError</Code></Error>"
+
+    with fake.client() as client, pytest.raises(R2S3Error, match="InternalError"):
+        client.put_file("ledger-raw", "key", _file(tmp_path, b"abcdefghijkl"))
+
+    assert fake.requests[-1].method == "DELETE"
+
+
+def test_put_file_frees_each_part_before_reading_the_next(tmp_path):
+    # httpx keeps each response in a reference cycle with its request, which
+    # holds the part it sent. With the cycle collector off, every part would
+    # stay in memory unless put_file frees it.
+    part_size = 1 << 20
+    sizes: list[int] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "PUT":
+            sizes.append(len(request.content))
+            return httpx.Response(200, headers={"ETag": '"etag"'})
+        if "uploads" in request.url.params:
+            return httpx.Response(200, content=b"<R><UploadId>up</UploadId></R>")
+        return httpx.Response(200, content=b"<CompleteMultipartUploadResult/>")
+
+    path = _file(tmp_path, bytes(8 * part_size))
+    client = R2Client(
+        CREDENTIALS, transport=httpx.MockTransport(handler), part_size=part_size
+    )
+    gc.disable()
+    tracemalloc.start()
+    try:
+        with client:
+            client.put_file("ledger-raw", "key", path)
+        _, peak = tracemalloc.get_traced_memory()
+    finally:
+        tracemalloc.stop()
+        gc.enable()
+
+    assert sizes == [part_size] * 8
+    assert peak < 3 * part_size
+
+
+def test_object_size_reads_head_and_is_none_for_a_missing_key():
+    fake = FakeR2()
+    fake.objects["raw/uk/ucl/pkg/bundle_2.zip"] = 4_426_805_029
+
+    with fake.client() as client:
+        present = client.object_size("ledger-raw", "raw/uk/ucl/pkg/bundle_2.zip")
+        missing = client.object_size("ledger-raw", "raw/uk/ucl/pkg/bundle_3.zip")
+
+    assert present == 4_426_805_029
+    assert missing is None
+    assert [request.method for request in fake.requests] == ["HEAD", "HEAD"]
+
+
+def test_object_size_reports_a_refused_head_rather_than_a_missing_object():
+    fake = FakeR2()
+    fake.head_status = 403
+
+    with fake.client() as client, pytest.raises(R2S3Error, match="HTTP 403"):
+        client.object_size("ledger-raw", "key")
+
+
+def test_put_file_refuses_an_empty_file(tmp_path):
+    fake = FakeR2()
+
+    with fake.client() as client, pytest.raises(R2S3Error, match="non-empty"):
+        client.put_file("ledger-raw", "key", _file(tmp_path, b""))
+
+    assert fake.requests == []
+
+
+def test_explicit_r2_keys_need_no_token_lookup(tmp_path):
+    credentials = credentials_from_environment(
+        {
+            "CLOUDFLARE_ACCOUNT_ID": "account",
+            "R2_ACCESS_KEY_ID": "key-id",
+            "R2_SECRET_ACCESS_KEY": "secret",
+            "CLOUDFLARE_API_TOKEN": "unused",
+        },
+        wrangler_toml=tmp_path / "absent.toml",
+        transport=httpx.MockTransport(
+            lambda request: pytest.fail("no Cloudflare API call expected")
+        ),
+    )
+
+    assert credentials == R2Credentials("account", "key-id", "secret")
+    assert credentials.host == "account.r2.cloudflarestorage.com"
+    assert "secret" not in repr(credentials)
+
+
+@pytest.mark.parametrize("user_status", [200, 401])
+def test_api_token_gives_the_key_pair_and_wrangler_toml_the_account(
+    tmp_path, user_status
+):
+    wrangler_toml = tmp_path / "wrangler.toml"
+    wrangler_toml.write_text('name = "chronicle"\naccount_id = "acct-123"\n')
+    seen = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request.url.path)
+        assert request.headers["authorization"] == "Bearer token-value"
+        if request.url.path == "/client/v4/user/tokens/verify" and user_status != 200:
+            return httpx.Response(user_status, json={"success": False})
+        return httpx.Response(200, json={"result": {"id": "token-id"}})
+
+    credentials = credentials_from_environment(
+        {"CLOUDFLARE_API_TOKEN": "token-value"},
+        wrangler_toml=wrangler_toml,
+        transport=httpx.MockTransport(handler),
+    )
+
+    assert credentials == R2Credentials(
+        "acct-123", "token-id", hashlib.sha256(b"token-value").hexdigest()
+    )
+    expected_paths = ["/client/v4/user/tokens/verify"]
+    if user_status != 200:
+        expected_paths.append("/client/v4/accounts/acct-123/tokens/verify")
+    assert seen == expected_paths
+
+
+def test_a_token_lookup_that_cannot_reach_cloudflare_is_reported(tmp_path):
+    def unreachable(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("offline", request=request)
+
+    with pytest.raises(R2S3Error, match="Could not verify"):
+        credentials_from_environment(
+            {"CLOUDFLARE_ACCOUNT_ID": "account", "CLOUDFLARE_API_TOKEN": "token"},
+            wrangler_toml=tmp_path / "absent.toml",
+            transport=httpx.MockTransport(unreachable),
+        )
+
+
+def test_an_unreadable_create_response_is_an_r2_error(tmp_path):
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, content=b"not xml")
+
+    client = R2Client(CREDENTIALS, transport=httpx.MockTransport(handler))
+    with client, pytest.raises(R2S3Error, match="no UploadId"):
+        client.put_file("ledger-raw", "key", _file(tmp_path, b"abc"))
+
+
+@pytest.mark.parametrize(
+    ("environ", "message"),
+    [
+        ({"CLOUDFLARE_API_TOKEN": "token"}, "No Cloudflare account"),
+        ({"CLOUDFLARE_ACCOUNT_ID": "account"}, "No R2 credentials"),
+    ],
+)
+def test_missing_credentials_are_reported(tmp_path, environ, message):
+    with pytest.raises(R2S3Error, match=message):
+        credentials_from_environment(environ, wrangler_toml=tmp_path / "absent.toml")
diff --git a/tests/test_source_ucl_house_price_per_square_metre_856542.py b/tests/test_source_ucl_house_price_per_square_metre_856542.py
new file mode 100644
index 00000000..bc524b38
--- /dev/null
+++ b/tests/test_source_ucl_house_price_per_square_metre_856542.py
@@ -0,0 +1,122 @@
+"""UCL's house price per square metre deposit, registered raw-only (chronicle#408, item 6).
+
+UK Data Service ReShare 856542 publishes two stored zips, of 3.2 and 4.4 GB, under CC BY
+4.0. Both carry Royal Mail address data licensed for non-commercial use. They are pinned and
+held in R2 only, uploaded through R2's S3 API because Wrangler takes at most 300 MiB
+(chronicle#409), and no build parses them.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+import subprocess
+
+import yaml
+
+from chronicle.artifacts import (
+    WRANGLER_PUT_LIMIT_BYTES,
+    build_r2_key,
+    inventory_source_artifacts,
+    publish_source_artifacts,
+)
+from chronicle.bundle import UK_BUNDLE_SOURCES
+from chronicle.source_package import SOURCE_PACKAGE_ALIASES
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+DIRECTORY = "ucl/house_price_per_square_metre_856542"
+PACKAGE_DIR = REPOSITORY_ROOT / "db" / "data" / DIRECTORY
+PACKAGE_ID = "ucl-house-price-per-square-metre-856542"
+# file key -> filename, sha256, bytes, and ReShare's ETag, which is the file's MD5.
+BUNDLES = {
+    "bundle_1": (
+        "856542_bundle_1.zip",
+        "e397cdac214b918a39f3ccf5266adaf9e4c3d8fb4f71e0bdc1b9c9474734b296",
+        3_214_161_126,
+        "239aebb08e10939b7cba4c5b2d5f89c9",
+    ),
+    "bundle_2": (
+        "856542_bundle_2.zip",
+        "61cbf6a72e35f7569d9121d80eaf89e757fec3bd4c21f1175e37fff44e02da97",
+        4_426_805_029,
+        "5fa5f769f3061aef86c455a4ed23c97e",
+    ),
+}
+
+
+def _manifest() -> dict:
+    return yaml.safe_load((PACKAGE_DIR / "manifest.yaml").read_text(encoding="utf-8"))
+
+
+def test_both_bundles_are_pinned_raw_only_in_r2():
+    manifest = _manifest()
+    assert manifest["source_id"] == "ucl"
+    assert manifest["package_id"] == PACKAGE_ID
+    assert manifest["registration_kind"] == "raw_only"
+    assert manifest["license"] == (
+        "Creative Commons Attribution 4.0 International (CC BY 4.0)"
+    )
+    assert manifest["license_verified_at"] == "2026-11-06"
+    assert "non-commercial" in manifest["license_evidence"]
+    assert "10.5255/UKDA-SN-856542" in manifest["attribution"]
+    assert list(manifest["files"]) == list(BUNDLES)
+    for file_key, (filename, sha256, size_bytes, md5) in BUNDLES.items():
+        entry = manifest["files"][file_key]
+        number = file_key.removeprefix("bundle_")
+        assert entry["filename"] == filename
+        assert entry["source_url"] == (
+            f"https://reshare.ukdataservice.ac.uk/856542/{number}/{filename}"
+        )
+        assert entry["sha256"] == sha256
+        assert entry["size_bytes"] == size_bytes
+        assert md5 in manifest["provenance_notes"]
+        # Too large for Wrangler, so publish-raw sends it through the S3 API.
+        assert size_bytes > WRANGLER_PUT_LIMIT_BYTES
+
+        key = build_r2_key(
+            source_id="ucl",
+            package_id=PACKAGE_ID,
+            year=file_key,
+            sha256=sha256,
+            filename=filename,
+            package_path=PACKAGE_DIR / "manifest.yaml",
+        )
+        assert key == f"raw/uk/ucl/{PACKAGE_ID}/{file_key}/{sha256}/{filename}"
+        assert entry["storage"] == {
+            "local": False,
+            "r2": {
+                "provider": "r2",
+                "bucket": "ledger-raw",
+                "key": key,
+                "uri": f"r2://ledger-raw/{key}",
+            },
+        }
+        ignored = subprocess.run(
+            ["git", "check-ignore", "--quiet", str(PACKAGE_DIR / filename)],
+            cwd=REPOSITORY_ROOT,
+            check=False,
+        )
+        assert ignored.returncode == 0
+
+    assert PACKAGE_ID not in SOURCE_PACKAGE_ALIASES
+    assert PACKAGE_ID not in UK_BUNDLE_SOURCES
+    assert not (
+        REPOSITORY_ROOT / "packages" / DIRECTORY / "source_package.yaml"
+    ).exists()
+
+
+def test_a_checkout_without_the_bundles_stays_valid(tmp_path):
+    target = tmp_path / "db" / "data" / DIRECTORY
+    target.mkdir(parents=True)
+    (target / "manifest.yaml").write_bytes((PACKAGE_DIR / "manifest.yaml").read_bytes())
+
+    inventory = inventory_source_artifacts(tmp_path / "db" / "data")
+    published = publish_source_artifacts(
+        tmp_path / "db" / "data", wrangler_command="false"
+    )
+
+    assert inventory.valid
+    assert inventory.counts["missing_count"] == 0
+    assert inventory.counts["r2_only_count"] == 2
+    assert published.valid
+    assert published.counts["skipped_count"] == 2
+    assert published.counts["uploaded_count"] == 0