diff --git a/.github/workflows/example-gate.yml b/.github/workflows/example-gate.yml
index 998a831..c44661b 100644
--- a/.github/workflows/example-gate.yml
+++ b/.github/workflows/example-gate.yml
@@ -7,6 +7,11 @@ on:
 jobs:
   examples:
     runs-on: ubuntu-latest
+    env:
+      TEAQL_S3_ENDPOINT: http://127.0.0.1:19010
+      TEAQL_S3_BUCKET: teaql-python-example-ci
+      TEAQL_S3_ACCESS_KEY: teaql-python-example-ci
+      TEAQL_S3_SECRET_KEY: teaql-python-example-ci-fixture-only
     steps:
       - uses: actions/checkout@v4
       - name: Install example verification tools
@@ -15,5 +20,28 @@ jobs:
         with:
           python-version: '3.10'
           cache: pip
-      - run: pip install -e ".[test,telemetry]"
+      - run: pip install -e ".[test,telemetry,object-storage-s3]"
+      - name: Start dedicated S3 example fixture
+        run: |
+          docker run -d --name teaql-python-example-s3 \
+            -p 127.0.0.1:19010:9000 \
+            -e RUSTFS_ACCESS_KEY="$TEAQL_S3_ACCESS_KEY" \
+            -e RUSTFS_SECRET_KEY="$TEAQL_S3_SECRET_KEY" \
+            -e RUSTFS_CONSOLE_ENABLE=false \
+            rustfs/rustfs@sha256:41fe89380f4120a337790c02af192c3fe7bb55c3edc2e6e9357b487b47c6ab21 /data
+          for attempt in {1..30}; do
+            if curl -sS -o /dev/null --max-time 2 "$TEAQL_S3_ENDPOINT/"; then
+              exit 0
+            fi
+            if [[ "$(docker inspect --format '{{.State.Running}}' teaql-python-example-s3)" != true ]]; then
+              docker logs teaql-python-example-s3
+              exit 1
+            fi
+            sleep 1
+          done
+          docker logs teaql-python-example-s3
+          exit 1
       - run: ./scripts/verify-examples.sh
+      - name: Stop dedicated S3 example fixture
+        if: always()
+        run: docker rm -f teaql-python-example-s3 || true
diff --git a/README.md b/README.md
index d148560..5f28a1c 100644
--- a/README.md
+++ b/README.md
@@ -399,6 +399,7 @@ Request JSON is not allowed to replace those trusted capabilities.
 - [Mutation Policy example](examples/mutation-policy)
 - [Facet example](examples/facet-trace/README.md)
 - [Opaque reference example](examples/opaque-entity-reference)
+- [Binary object storage and temporary S3 URLs](examples/object-storage/README.md)
 
 ## Safe troubleshooting
 
diff --git a/examples/object-storage/README.md b/examples/object-storage/README.md
new file mode 100644
index 0000000..1394717
--- /dev/null
+++ b/examples/object-storage/README.md
@@ -0,0 +1,53 @@
+# Context owned object storage
+
+Tracking: [Python #52](https://github.com/teaql/teaql-python/issues/52),
+[shared contract #122](https://github.com/teaql/teaql-conformance/issues/122).
+
+Install this checkout with `python -m pip install -e '.[object-storage-s3]'`.
+The optional boto3 provider owns signing, credentials and the shared HTTP pool.
+SDK types stay outside the core contract. The application installs the provider
+at trusted startup and closes it at shutdown.
+
+```python
+from teaql.core.object_storage import ObjectKey, PutOptions, WriteMode
+from teaql.provider.s3 import S3ObjectStorage
+from teaql.runtime import UserContext
+
+provider = S3ObjectStorage.connect("attachments")  # SDK default credentials chain
+context = UserContext().with_object_storage(provider)
+storage = context.object_storage("store payment receipt attachment")
+key = ObjectKey("receipts/opaque-reference.pdf")
+storage.put(key, source, PutOptions(content_length, WriteMode.CREATE_ONLY))
+link = storage.get_read_url(key)
+with storage.get(key) as response:
+    block = response.stream.read(65536)
+```
+
+This binary-stream API is synchronous, matching boto3 and Python BinaryIO.
+Async services can run the complete operation with `asyncio.to_thread`; never
+close the input while that worker is still running. Uploads stage in a private
+temporary file with bounded buffers, exact length checks and a 5 GiB limit.
+The facade closes upload inputs. Downloads stream and require explicit close;
+`get_bytes(key, maximum_bytes)` enforces an explicit memory ceiling.
+
+Create-only uses atomic conditional PUT; replacement is explicit. Only confirmed
+object absence becomes false. Missing buckets, permissions and network failures
+remain typed errors. Keys and default five-minute URLs have different lifetimes;
+the URL is a reusable bearer link, can expire early with credentials, and does
+not check existence or infer business permission. Diagnostic representations
+and operation events exclude keys, body, metadata, URLs and SDK error messages.
+Object storage is independent of SQL transactions.
+
+Set the four `TEAQL_S3_ENDPOINT`, `TEAQL_S3_BUCKET`, `TEAQL_S3_ACCESS_KEY`,
+`TEAQL_S3_SECRET_KEY` variables to a dedicated test service, then run twice:
+
+```bash
+bash examples/object-storage/verify.sh
+bash examples/object-storage/verify.sh
+```
+
+The example uses the repository source via PYTHONPATH, provisions its test bucket
+if absent and deletes only its unique keys. The production provider never
+creates buckets. Use `PYTHON=/path/to/venv/bin/python` to select an environment
+with boto3 and pytest. A local fixture pass qualifies that fixture, not every
+S3-compatible service. Range/list/copy/multipart and signed upload are extensions.
diff --git a/examples/object-storage/main.py b/examples/object-storage/main.py
new file mode 100644
index 0000000..f2457a2
--- /dev/null
+++ b/examples/object-storage/main.py
@@ -0,0 +1,101 @@
+"""Local-source S3 lifecycle fixture. Test bucket creation is administration."""
+from concurrent.futures import ThreadPoolExecutor
+from datetime import timedelta
+from io import BytesIO
+from uuid import uuid4
+import os
+import time
+import urllib.request
+import urllib.error
+import boto3
+from botocore.config import Config
+from teaql.core.object_storage import *
+from teaql.runtime import UserContext
+from teaql.provider.s3 import S3ObjectStorage
+
+
+def expect(code, work):
+    try:
+        work()
+    except StorageError as error:
+        assert error.code == code, (error.code, code)
+        return
+    raise AssertionError(f"expected {code}")
+
+
+def main():
+    bucket = os.environ["TEAQL_S3_BUCKET"]
+    client = boto3.client("s3", endpoint_url=os.environ["TEAQL_S3_ENDPOINT"], region_name="us-east-1",
+        aws_access_key_id=os.environ["TEAQL_S3_ACCESS_KEY"], aws_secret_access_key=os.environ["TEAQL_S3_SECRET_KEY"],
+        config=Config(signature_version="s3v4", retries={"max_attempts": 1}, s3={"addressing_style": "path"}))
+    from botocore.exceptions import ClientError
+    try:
+        client.head_bucket(Bucket=bucket)
+    except ClientError as error:
+        if error.response["ResponseMetadata"]["HTTPStatusCode"] != 404:
+            raise RuntimeError("test bucket setup failed") from None
+        client.create_bucket(Bucket=bucket)
+    provider = S3ObjectStorage(client, bucket)
+    events = []
+    context = UserContext().with_trace_id("storage-example").with_object_storage(provider).with_object_storage_observer(events.append)
+    storage = context.object_storage("verify binary object storage")
+    prefix = f"python-example/{uuid4().hex}/"
+    key = ObjectKey(prefix + "收据 +% binary")
+    empty, race = ObjectKey(prefix + "empty"), ObjectKey(prefix + "race")
+    data = bytes(i % 251 for i in range(256 * 1024))
+    try:
+        assert not storage.exists(key)
+        expect(StorageErrorCode.NOT_FOUND, lambda: storage.get(key))
+        options = PutOptions(len(data), WriteMode.CREATE_ONLY, "application/octet-stream", {"example": "binary"})
+        storage.put_bytes(key, data, options)
+        metadata = storage.head(key)
+        assert metadata.size == len(data) and metadata.custom_metadata["example"] == "binary" and metadata.etag
+        assert storage.get_bytes(key, len(data)) == data
+        expect(StorageErrorCode.CONFLICT, lambda: storage.put_bytes(key, b"", PutOptions(0, WriteMode.CREATE_ONLY)))
+        expect(StorageErrorCode.TOO_LARGE, lambda: storage.get_bytes(key, 1))
+        url = storage.get_read_url(key, ReadUrlOptions(disposition="attachment", filename="receipt.bin"))
+        with urllib.request.urlopen(url.url, timeout=5) as response:
+            assert response.read() == data
+            assert response.headers["Content-Disposition"] == 'attachment; filename="receipt.bin"'
+        expires = storage.get_read_url(key, ReadUrlOptions(timedelta(seconds=1)))
+        time.sleep(2.2)
+        try:
+            urllib.request.urlopen(expires.url, timeout=5)
+        except urllib.error.HTTPError as error:
+            assert error.code == 403
+        else:
+            raise AssertionError("expired URL must be denied")
+        storage.put_bytes(empty, b"", PutOptions(0, WriteMode.CREATE_ONLY))
+        assert storage.get_bytes(empty, 0) == b""
+        expect(StorageErrorCode.INVALID_INPUT, lambda: storage.put_bytes(key, b"ab", PutOptions(1, WriteMode.REPLACE)))
+        assert storage.get_bytes(key, len(data)) == data
+        storage.put_bytes(key, b"new", PutOptions(3, WriteMode.REPLACE))
+        assert storage.get_bytes(key, 3) == b"new"
+        other = UserContext().with_trace_id("storage-other").with_object_storage(provider).with_object_storage_observer(events.append)
+        def create(api):
+            try:
+                api.put_bytes(race, b"race", PutOptions(4, WriteMode.CREATE_ONLY))
+                return "success"
+            except StorageError as error:
+                assert error.code == StorageErrorCode.CONFLICT
+                return "conflict"
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            results = list(executor.map(create, [storage, other.object_storage("verify atomic create")]))
+        assert sorted(results) == ["conflict", "success"]
+        assert {e.trace_id for e in events} == {"storage-example", "storage-other"}
+        missing = UserContext().with_object_storage(S3ObjectStorage(client, "missing-" + uuid4().hex))
+        expect(StorageErrorCode.UNAVAILABLE, lambda: missing.object_storage("verify missing bucket").exists(key))
+        for item in (key, empty, race):
+            storage.delete(item)
+            storage.delete(item)
+            assert not storage.exists(item)
+        assert prefix not in repr(events) and "X-Amz-Signature" not in repr(events)
+        print("PASS Python S3: binary, empty, metadata, URL download/expiry, atomic create, replace, bounded reads, missing bucket, repeated delete")
+    finally:
+        for item in (key, empty, race):
+            storage.delete(item)
+        provider.close()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/object-storage/verify.sh b/examples/object-storage/verify.sh
new file mode 100644
index 0000000..30caf7a
--- /dev/null
+++ b/examples/object-storage/verify.sh
@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+set -euo pipefail
+repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+: "${TEAQL_S3_ENDPOINT:?dedicated local test endpoint required}"
+: "${TEAQL_S3_BUCKET:?dedicated test bucket required}"
+: "${TEAQL_S3_ACCESS_KEY:?test access key required}"
+: "${TEAQL_S3_SECRET_KEY:?test secret key required}"
+cd "$repo"
+export PYTHONPATH="$repo/src"
+export PYTHONDONTWRITEBYTECODE=1
+"${PYTHON:-python3}" -m pytest tests/runtime/test_object_storage.py -q
+"${PYTHON:-python3}" examples/object-storage/main.py
diff --git a/pyproject.toml b/pyproject.toml
index 886e8e5..c5c86bd 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -30,6 +30,7 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
+object-storage-s3 = ["boto3==1.43.113"]
 telemetry = [
     "opentelemetry-api>=1.44.0,<2",
     "opentelemetry-sdk>=1.44.0,<2",
diff --git a/scripts/verify-examples.sh b/scripts/verify-examples.sh
index 8f55316..7762c16 100755
--- a/scripts/verify-examples.sh
+++ b/scripts/verify-examples.sh
@@ -2,7 +2,7 @@
 set -euo pipefail
 
 repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-expected=(business-clock business-id conformance facet-trace mutation-policy opaque-entity-reference order-management query-policy school-management task_board trace-chain)
+expected=(business-clock business-id conformance facet-trace mutation-policy object-storage opaque-entity-reference order-management query-policy school-management task_board trace-chain)
 mapfile -t actual < <(find "$repo/examples" -mindepth 1 -maxdepth 1 -type d -printf '%f\n' | sort)
 if [[ "${actual[*]}" != "${expected[*]}" ]]; then
   echo "example inventory changed; update scripts/verify-examples.sh: ${actual[*]}" >&2
@@ -20,6 +20,7 @@ if rg -l 'include\s*=.*teaql\*' "$repo/examples" --glob pyproject.toml >/dev/nul
   exit 1
 fi
 
+bash "$repo/examples/object-storage/verify.sh"
 PYTHONPATH="$repo/examples/conformance:$repo/src" python -m app.main
 PYTHONPATH="$repo/src" python -m unittest discover -s "$repo/examples/conformance" -p 'test_sql_log_intent.py' -v
 PYTHONPATH="$repo/examples/school-management:$repo/src" python -m app.main
diff --git a/src/teaql/core/object_storage.py b/src/teaql/core/object_storage.py
new file mode 100644
index 0000000..d513736
--- /dev/null
+++ b/src/teaql/core/object_storage.py
@@ -0,0 +1,156 @@
+"""Provider-neutral binary storage values. Private values have masked reprs."""
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+from enum import Enum
+from types import MappingProxyType
+from typing import BinaryIO, Mapping, Optional
+import unicodedata
+
+MAX_PUT_SIZE = 5 * 1024**3
+
+
+class StorageErrorCode(str, Enum):
+    INVALID_INPUT = "InvalidInput"
+    NOT_CONFIGURED = "NotConfigured"
+    NOT_FOUND = "NotFound"
+    ACCESS_DENIED = "AccessDenied"
+    CONFLICT = "Conflict"
+    TIMEOUT = "Timeout"
+    UNAVAILABLE = "Unavailable"
+    UNSUPPORTED = "Unsupported"
+    TOO_LARGE = "TooLarge"
+    IO = "Io"
+
+
+class StorageError(Exception):
+    def __init__(self, code: StorageErrorCode):
+        self.code = code
+        super().__init__(f"Object storage: {code.value}")
+
+
+def _header(value: str) -> str:
+    if not isinstance(value, str) or not value or any(ord(c) < 32 or ord(c) > 126 for c in value):
+        raise StorageError(StorageErrorCode.INVALID_INPUT)
+    return value
+
+
+@dataclass(frozen=True, repr=False)
+class ObjectKey:
+    value: str
+
+    def __post_init__(self):
+        invalid = not isinstance(self.value, str) or not self.value
+        if not invalid:
+            try:
+                invalid = len(self.value.encode("utf-8")) > 1024
+            except UnicodeEncodeError:
+                invalid = True
+            invalid |= any(unicodedata.category(c) == "Cc" for c in self.value)
+        if invalid:
+            raise StorageError(StorageErrorCode.INVALID_INPUT)
+
+    def __repr__(self):
+        return "ObjectKey[masked]"
+
+
+class WriteMode(str, Enum):
+    CREATE_ONLY = "create_only"
+    REPLACE = "replace"
+
+
+@dataclass(frozen=True, repr=False)
+class PutOptions:
+    size: int
+    mode: WriteMode
+    content_type: str = "application/octet-stream"
+    metadata: Mapping[str, str] = field(default_factory=dict)
+
+    def __post_init__(self):
+        if type(self.size) is not int or self.size < 0 or not isinstance(self.mode, WriteMode):
+            raise StorageError(StorageErrorCode.INVALID_INPUT)
+        if self.size > MAX_PUT_SIZE:
+            raise StorageError(StorageErrorCode.TOO_LARGE)
+        _header(self.content_type)
+        values = dict(self.metadata)
+        for key, value in values.items():
+            if not isinstance(key, str) or not key or not key[0].isascii() or not key[0].isalnum() \
+                    or any(c not in "abcdefghijklmnopqrstuvwxyz0123456789-" for c in key):
+                raise StorageError(StorageErrorCode.INVALID_INPUT)
+            _header(value)
+        object.__setattr__(self, "metadata", MappingProxyType(values))
+
+    def __repr__(self):
+        return f"PutOptions[size={self.size}, mode={self.mode.value}, metadata=masked]"
+
+
+@dataclass(frozen=True, repr=False)
+class ObjectMetadata:
+    key: ObjectKey
+    size: int
+    content_type: Optional[str] = None
+    etag: Optional[str] = None
+    last_modified: Optional[datetime] = None
+    custom_metadata: Mapping[str, str] = field(default_factory=dict)
+
+    def __post_init__(self):
+        if type(self.size) is not int or self.size < 0:
+            raise StorageError(StorageErrorCode.UNAVAILABLE)
+        object.__setattr__(self, "custom_metadata", MappingProxyType(dict(self.custom_metadata)))
+
+    def __repr__(self):
+        return f"ObjectMetadata[size={self.size}, details=masked]"
+
+
+@dataclass(frozen=True, repr=False)
+class ReadUrlOptions:
+    lifetime: timedelta = timedelta(minutes=5)
+    disposition: Optional[str] = None
+    filename: Optional[str] = None
+
+    def __post_init__(self):
+        if not isinstance(self.lifetime, timedelta):
+            raise StorageError(StorageErrorCode.INVALID_INPUT)
+        seconds = self.lifetime.total_seconds()
+        if seconds <= 0 or seconds > 604800 or not seconds.is_integer():
+            raise StorageError(StorageErrorCode.INVALID_INPUT)
+        if self.disposition not in (None, "inline", "attachment"):
+            raise StorageError(StorageErrorCode.INVALID_INPUT)
+        if self.filename is not None:
+            _header(self.filename)
+            if self.disposition is None or any(c in self.filename for c in '/\\"'):
+                raise StorageError(StorageErrorCode.INVALID_INPUT)
+
+    def content_disposition(self):
+        if self.disposition is None:
+            return None
+        return self.disposition + (f'; filename="{self.filename}"' if self.filename is not None else "")
+
+    def __repr__(self):
+        return "ReadUrlOptions[details=masked]"
+
+
+@dataclass(frozen=True, repr=False)
+class ObjectReadUrl:
+    url: str
+    expires_at: datetime
+
+    def __repr__(self):
+        return "ObjectReadUrl[url=masked]"
+
+
+@dataclass(repr=False)
+class ObjectContent:
+    metadata: ObjectMetadata
+    stream: BinaryIO
+
+    def close(self):
+        self.stream.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_):
+        self.close()
+
+    def __repr__(self):
+        return "ObjectContent[masked]"
diff --git a/src/teaql/provider/s3/__init__.py b/src/teaql/provider/s3/__init__.py
new file mode 100644
index 0000000..d852dd1
--- /dev/null
+++ b/src/teaql/provider/s3/__init__.py
@@ -0,0 +1,149 @@
+"""Optional S3 adapter. Install teaql[object-storage-s3]; boto3 owns SigV4."""
+from datetime import datetime, timedelta, timezone
+from urllib.parse import parse_qs, urlsplit
+import tempfile
+from teaql.core.object_storage import (
+    ObjectMetadata, ObjectContent, ObjectReadUrl, StorageError, StorageErrorCode, WriteMode,
+)
+
+
+def _translate(error):
+    from botocore.exceptions import ClientError, ConnectTimeoutError, ReadTimeoutError, ParamValidationError
+    if isinstance(error, (ConnectTimeoutError, ReadTimeoutError, TimeoutError)):
+        return StorageError(StorageErrorCode.TIMEOUT)
+    if isinstance(error, ParamValidationError):
+        return StorageError(StorageErrorCode.INVALID_INPUT)
+    if isinstance(error, ClientError):
+        status = error.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
+        if error.response.get("Error", {}).get("Code") == "NoSuchBucket":
+            return StorageError(StorageErrorCode.UNAVAILABLE)
+        return StorageError({401: StorageErrorCode.ACCESS_DENIED, 403: StorageErrorCode.ACCESS_DENIED,
+            404: StorageErrorCode.NOT_FOUND, 409: StorageErrorCode.CONFLICT, 412: StorageErrorCode.CONFLICT,
+            408: StorageErrorCode.TIMEOUT, 504: StorageErrorCode.TIMEOUT, 413: StorageErrorCode.TOO_LARGE,
+            501: StorageErrorCode.UNSUPPORTED, 400: StorageErrorCode.INVALID_INPUT,
+            422: StorageErrorCode.INVALID_INPUT}.get(status, StorageErrorCode.UNAVAILABLE))
+    return StorageError(StorageErrorCode.UNAVAILABLE)
+
+
+def _sdk_call(work):
+    translated = None
+    try:
+        return work()
+    except Exception as error:
+        translated = _translate(error)
+    # Outside the except block: neither __context__ nor __cause__ retains provider text.
+    raise translated
+
+
+def _metadata(key, response):
+    size, etag = response.get("ContentLength"), response.get("ETag")
+    if type(size) is not int or size < 0 or not isinstance(etag, str) or not etag.strip():
+        raise StorageError(StorageErrorCode.UNAVAILABLE)
+    return ObjectMetadata(key, size, response.get("ContentType"), etag,
+                          response.get("LastModified"), response.get("Metadata", {}))
+
+
+class S3ObjectStorage:
+    """Shared client installed at trusted startup; never creates buckets."""
+    def __init__(self, client, bucket: str):
+        if not isinstance(bucket, str) or not bucket.strip():
+            raise StorageError(StorageErrorCode.INVALID_INPUT)
+        self._client, self._bucket = client, bucket
+
+    @classmethod
+    def connect(cls, bucket, *, session=None, endpoint_url=None, region_name="us-east-1", **client_options):
+        import boto3
+        from botocore.config import Config
+        session = session or boto3.Session()
+        client_options.setdefault("config", Config(signature_version="s3v4", connect_timeout=10, read_timeout=30,
+            s3={"addressing_style": "path"}, retries={"max_attempts": 2, "mode": "standard"}))
+        return cls(session.client("s3", endpoint_url=endpoint_url, region_name=region_name, **client_options), bucket)
+
+    def put(self, invocation, key, content, options):
+        with tempfile.TemporaryFile(mode="w+b") as staging:
+            total, error = 0, None
+            try:
+                while True:
+                    block = content.read(65536)
+                    if not isinstance(block, bytes):
+                        raise StorageError(StorageErrorCode.INVALID_INPUT)
+                    if not block:
+                        break
+                    total += len(block)
+                    if total > options.size:
+                        raise StorageError(StorageErrorCode.INVALID_INPUT)
+                    staging.write(block)
+            except StorageError:
+                raise
+            except Exception:
+                error = StorageError(StorageErrorCode.IO)
+            if error:
+                raise error
+            if total != options.size:
+                raise StorageError(StorageErrorCode.INVALID_INPUT)
+            staging.seek(0)
+            args = dict(Bucket=self._bucket, Key=key.value, Body=staging, ContentLength=total,
+                        ContentType=options.content_type, Metadata=dict(options.metadata))
+            if options.mode == WriteMode.CREATE_ONLY:
+                args["IfNoneMatch"] = "*"
+            result = _sdk_call(lambda: self._client.put_object(**args))
+            if not isinstance(result.get("ETag"), str) or not result["ETag"].strip():
+                raise StorageError(StorageErrorCode.UNAVAILABLE)
+            return ObjectMetadata(key, total, options.content_type, result["ETag"], custom_metadata=options.metadata)
+
+    def get(self, invocation, key):
+        response = _sdk_call(lambda: self._client.get_object(Bucket=self._bucket, Key=key.value))
+        try:
+            return ObjectContent(_metadata(key, response), response["Body"])
+        except Exception:
+            response.get("Body") and response["Body"].close()
+            raise
+
+    def head(self, invocation, key):
+        missing = False
+        try:
+            response = _sdk_call(lambda: self._client.head_object(Bucket=self._bucket, Key=key.value))
+            return _metadata(key, response)
+        except StorageError as error:
+            if error.code != StorageErrorCode.NOT_FOUND:
+                raise
+            missing = True
+        if missing:
+            bucket_error = None
+            try:
+                _sdk_call(lambda: self._client.head_bucket(Bucket=self._bucket))
+            except StorageError as error:
+                bucket_error = StorageError(StorageErrorCode.UNAVAILABLE) if error.code == StorageErrorCode.NOT_FOUND else error
+            if bucket_error:
+                raise bucket_error
+        return None
+
+    def delete(self, invocation, key):
+        _sdk_call(lambda: self._client.delete_object(Bucket=self._bucket, Key=key.value))
+
+    def get_read_url(self, invocation, key, options):
+        params = dict(Bucket=self._bucket, Key=key.value)
+        disposition = options.content_disposition()
+        if disposition:
+            params["ResponseContentDisposition"] = disposition
+        url = _sdk_call(lambda: self._client.generate_presigned_url("get_object", Params=params,
+                        ExpiresIn=int(options.lifetime.total_seconds()), HttpMethod="GET"))
+        invalid = False
+        try:
+            query = parse_qs(urlsplit(url).query)
+            seconds = int(query["X-Amz-Expires"][0])
+            started = datetime.strptime(query["X-Amz-Date"][0], "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc)
+            invalid = not 0 < seconds <= int(options.lifetime.total_seconds())
+        except (KeyError, ValueError, TypeError, IndexError):
+            invalid = True
+        if invalid:
+            raise StorageError(StorageErrorCode.UNAVAILABLE)
+        if query.get("X-Amz-SignedHeaders") != ["host"]:
+            raise StorageError(StorageErrorCode.UNSUPPORTED)
+        return ObjectReadUrl(url, started + timedelta(seconds=seconds))
+
+    def close(self):
+        self._client.close()
+
+    def __repr__(self):
+        return "S3ObjectStorage[configuration=masked]"
diff --git a/src/teaql/runtime/context.py b/src/teaql/runtime/context.py
index 764b1e1..d636270 100644
--- a/src/teaql/runtime/context.py
+++ b/src/teaql/runtime/context.py
@@ -318,6 +318,19 @@ def with_runtime_telemetry(self, telemetry: Any) -> "UserContext":
         self._runtime_telemetry = telemetry
         return self
 
+    def with_object_storage(self, provider) -> "UserContext":
+        if provider is None:
+            from teaql.core.object_storage import StorageError, StorageErrorCode
+            raise StorageError(StorageErrorCode.INVALID_INPUT)
+        return self.insert_resource("object_storage_provider", provider)
+
+    def with_object_storage_observer(self, observer) -> "UserContext":
+        return self.insert_resource("object_storage_observer", observer)
+
+    def object_storage(self, comment: str):
+        from .object_storage import ObjectStorage
+        return ObjectStorage(self, comment)
+
     def set_runtime_telemetry(self, telemetry: Any) -> None:
         self._runtime_telemetry = telemetry
 
diff --git a/src/teaql/runtime/object_storage.py b/src/teaql/runtime/object_storage.py
new file mode 100644
index 0000000..c88492b
--- /dev/null
+++ b/src/teaql/runtime/object_storage.py
@@ -0,0 +1,228 @@
+"""Synchronous binary stream facade; async applications can use asyncio.to_thread."""
+from dataclasses import dataclass
+from typing import Protocol, Optional, BinaryIO, Callable
+from io import BytesIO
+import logging
+import time
+
+from teaql.core.object_storage import (
+    ObjectKey, ObjectMetadata, ObjectContent, PutOptions, ReadUrlOptions,
+    ObjectReadUrl, StorageError, StorageErrorCode,
+)
+from .telemetry import RuntimeOperation, start_runtime_operation
+
+
+@dataclass(frozen=True)
+class StorageInvocation:
+    context: object
+    comment: str
+
+
+class ObjectStorageProvider(Protocol):
+    def put(self, invocation: StorageInvocation, key: ObjectKey, content: BinaryIO, options: PutOptions) -> ObjectMetadata: ...
+    def get(self, invocation: StorageInvocation, key: ObjectKey) -> ObjectContent: ...
+    def head(self, invocation: StorageInvocation, key: ObjectKey) -> Optional[ObjectMetadata]: ...
+    def delete(self, invocation: StorageInvocation, key: ObjectKey) -> None: ...
+    def get_read_url(self, invocation: StorageInvocation, key: ObjectKey, options: ReadUrlOptions) -> ObjectReadUrl: ...
+
+
+@dataclass(frozen=True)
+class StorageEvent:
+    operation: str
+    comment: str
+    trace_id: str
+    outcome: str
+    elapsed_ms: float
+
+
+class _Receipt:
+    def __init__(self, invocation, operation):
+        self.invocation, self.operation = invocation, operation
+        self.started, self.ended = time.monotonic(), False
+        self.scope = start_runtime_operation(invocation.context.runtime_telemetry(), RuntimeOperation("object_storage", operation))
+        self.scope.activate()
+
+    def detach(self):
+        self.scope.deactivate()
+
+    def finish(self, outcome):
+        if self.ended:
+            return
+        self.ended = True
+        self.detach()
+        if outcome == "success":
+            self.scope.success()
+        else:
+            self.scope.failure(StorageError(StorageErrorCode.IO if outcome == "cancelled" else StorageErrorCode(outcome)))
+        safe = lambda text: "".join(" " if ord(c) < 32 or 127 <= ord(c) <= 159 else c for c in str(text or ""))
+        event = StorageEvent(self.operation, safe(self.invocation.comment), safe(self.invocation.context.trace_id()),
+                             outcome, (time.monotonic() - self.started) * 1000)
+        observer = self.invocation.context.get_resource("object_storage_observer")
+        try:
+            if observer:
+                observer(event)
+            else:
+                logging.getLogger("teaql.object_storage").warning(
+                    "ObjectStorage operation=%s comment=%s trace=%s outcome=%s elapsedMs=%.3f",
+                    event.operation, event.comment, event.trace_id, event.outcome, event.elapsed_ms)
+        except Exception:
+            pass
+
+
+class _ObservedStream:
+    def __init__(self, stream, size, receipt):
+        self._stream, self._remaining, self._receipt = stream, size, receipt
+        self._closed, self._ended = False, False
+
+    @property
+    def closed(self):
+        return self._closed
+
+    def read(self, size=-1):
+        if self._closed:
+            raise StorageError(StorageErrorCode.IO)
+        if size == 0:
+            return b""
+        if self._ended:
+            return b""
+        error = None
+        try:
+            data = self._stream.read(size)
+            if not isinstance(data, bytes) or len(data) > self._remaining or (not data and self._remaining != 0):
+                error = StorageError(StorageErrorCode.IO)
+            else:
+                self._remaining -= len(data)
+                # An unbounded read consumes to EOF in Python's BinaryIO contract.
+                if not data or size < 0:
+                    if self._remaining:
+                        error = StorageError(StorageErrorCode.IO)
+                    else:
+                        self._ended = True
+                        self._receipt.finish("success")
+        except Exception:
+            error = StorageError(StorageErrorCode.IO)
+        if error:
+            self._receipt.finish(error.code.value)
+            self._close_raw()
+            raise error
+        return data
+
+    def _close_raw(self):
+        self._closed = True
+        try:
+            self._stream.close()
+        except Exception:
+            pass
+
+    def close(self):
+        if self._closed:
+            return
+        self._closed = True
+        error = None
+        try:
+            self._stream.close()
+        except Exception:
+            error = StorageError(StorageErrorCode.IO)
+        self._receipt.finish(error.code.value if error else "cancelled")
+        if error:
+            raise error
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_):
+        self.close()
+
+    def __del__(self):
+        try:
+            self.close()
+        except Exception:
+            pass
+
+    def __repr__(self):
+        return "ObjectStream[masked]"
+
+
+class ObjectStorage:
+    def __init__(self, context, comment):
+        if not isinstance(comment, str) or not comment.strip():
+            raise StorageError(StorageErrorCode.INVALID_INPUT)
+        self._provider = context.get_resource("object_storage_provider")
+        if self._provider is None:
+            raise StorageError(StorageErrorCode.NOT_CONFIGURED)
+        self._invocation = StorageInvocation(context, comment)
+
+    def _call(self, operation, work):
+        receipt = _Receipt(self._invocation, operation)
+        try:
+            result = work()
+            receipt.finish("success")
+            return result
+        except StorageError as error:
+            receipt.finish(error.code.value)
+            raise
+        except Exception:
+            receipt.finish(StorageErrorCode.UNAVAILABLE.value)
+        raise StorageError(StorageErrorCode.UNAVAILABLE)
+
+    def put(self, key: ObjectKey, content: BinaryIO, options: PutOptions):
+        def work():
+            error, result = None, None
+            try:
+                result = self._provider.put(self._invocation, key, content, options)
+            except StorageError as caught:
+                error = caught
+            except Exception:
+                error = StorageError(StorageErrorCode.UNAVAILABLE)
+            try:
+                content.close()
+            except Exception:
+                error = error or StorageError(StorageErrorCode.IO)
+            if error:
+                raise error
+            return result
+        return self._call("put", work)
+
+    def put_bytes(self, key, content: bytes, options):
+        return self.put(key, BytesIO(content), options)
+
+    def get(self, key):
+        receipt = _Receipt(self._invocation, "get")
+        try:
+            response = self._provider.get(self._invocation, key)
+            receipt.detach()
+            return ObjectContent(response.metadata, _ObservedStream(response.stream, response.metadata.size, receipt))
+        except StorageError as error:
+            receipt.finish(error.code.value)
+            raise
+        except Exception:
+            receipt.finish(StorageErrorCode.UNAVAILABLE.value)
+        raise StorageError(StorageErrorCode.UNAVAILABLE)
+
+    def get_bytes(self, key, maximum_bytes: int):
+        if type(maximum_bytes) is not int or maximum_bytes < 0:
+            raise StorageError(StorageErrorCode.INVALID_INPUT)
+        with self.get(key) as content:
+            if content.metadata.size > maximum_bytes:
+                raise StorageError(StorageErrorCode.TOO_LARGE)
+            result = bytearray()
+            while True:
+                block = content.stream.read(min(65536, maximum_bytes - len(result) + 1))
+                if not block:
+                    break
+                if len(result) + len(block) > maximum_bytes:
+                    raise StorageError(StorageErrorCode.TOO_LARGE)
+                result.extend(block)
+            return bytes(result)
+
+    def head(self, key):
+        return self._call("head", lambda: self._provider.head(self._invocation, key))
+
+    def exists(self, key):
+        return self.head(key) is not None
+
+    def delete(self, key):
+        return self._call("delete", lambda: self._provider.delete(self._invocation, key))
+
+    def get_read_url(self, key, options=ReadUrlOptions()):
+        return self._call("read_url", lambda: self._provider.get_read_url(self._invocation, key, options))
diff --git a/tests/runtime/test_object_storage.py b/tests/runtime/test_object_storage.py
new file mode 100644
index 0000000..b571255
--- /dev/null
+++ b/tests/runtime/test_object_storage.py
@@ -0,0 +1,148 @@
+from datetime import timedelta, datetime, timezone
+from io import BytesIO
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from threading import Thread
+import pytest
+import boto3
+from botocore.config import Config
+from teaql.core.object_storage import *
+from teaql.runtime import UserContext
+from teaql.provider.s3 import S3ObjectStorage
+
+
+class Stub:
+    size = 3
+    code = None
+    seen = None
+    def head(self, invocation, key):
+        self.seen = invocation
+        if self.code:
+            raise StorageError(self.code)
+        return None
+    def put(self, invocation, key, stream, options):
+        self.seen = invocation
+        return ObjectMetadata(key, options.size)
+    def get(self, invocation, key):
+        return ObjectContent(ObjectMetadata(key, self.size), BytesIO(b"abc"))
+
+
+def assert_error(code, work):
+    with pytest.raises(StorageError) as found:
+        work()
+    assert found.value.code == code
+    assert found.value.__cause__ is None
+    assert "CANARY" not in str(found.value)
+    return found.value
+
+
+def test_object_storage_context_and_error_semantics():
+    assert_error(StorageErrorCode.NOT_CONFIGURED, lambda: UserContext().object_storage("read"))
+    stub = Stub()
+    context = UserContext().with_object_storage(stub)
+    assert_error(StorageErrorCode.INVALID_INPUT, lambda: context.object_storage("  "))
+    storage = context.object_storage("read attachment")
+    key = ObjectKey("PRIVATE-KEY-CANARY")
+    assert not storage.exists(key)
+    assert stub.seen.context is context
+    assert stub.seen.comment == "read attachment"
+    for code in (StorageErrorCode.ACCESS_DENIED, StorageErrorCode.UNAVAILABLE):
+        stub.code = code
+        assert_error(code, lambda: storage.exists(key))
+
+
+def test_object_storage_stream_bounds_ownership_and_completion():
+    events, stub = [], Stub()
+    storage = UserContext().with_object_storage(stub).with_object_storage_observer(events.append).object_storage("read attachment")
+    key = ObjectKey("PRIVATE-KEY-CANARY")
+    stream = BytesIO(b"")
+    storage.put(key, stream, PutOptions(0, WriteMode.CREATE_ONLY))
+    assert stream.closed
+    with storage.get(key) as response:
+        assert response.stream.read(1) == b"a"
+    assert events[-1].outcome == "cancelled"
+    assert storage.get_bytes(key, 3) == b"abc"
+    assert events[-1].outcome == "success"
+    assert_error(StorageErrorCode.TOO_LARGE, lambda: storage.get_bytes(key, 2))
+    for size in (2, 4):
+        stub.size = size
+        assert_error(StorageErrorCode.IO, lambda: storage.get_bytes(key, 5))
+    assert "CANARY" not in repr(events)
+    context = UserContext().with_object_storage(stub).with_object_storage_observer(lambda event: 1/0)
+    assert not context.object_storage("read").exists(key)
+
+
+def test_object_storage_options_and_private_representations():
+    for value in ("", "a" * 1025, "bad\nkey", "\ud800"):
+        assert_error(StorageErrorCode.INVALID_INPUT, lambda: ObjectKey(value))
+    assert_error(StorageErrorCode.TOO_LARGE, lambda: PutOptions(MAX_PUT_SIZE + 1, WriteMode.CREATE_ONLY))
+    assert_error(StorageErrorCode.INVALID_INPUT, lambda: PutOptions(0, WriteMode.CREATE_ONLY, "bad\r\nheader"))
+    assert_error(StorageErrorCode.INVALID_INPUT, lambda: PutOptions(0, WriteMode.CREATE_ONLY, metadata={"Bad Key": "value"}))
+    for seconds in (0, -1, 0.5, 604801):
+        assert_error(StorageErrorCode.INVALID_INPUT, lambda: ReadUrlOptions(timedelta(seconds=seconds)))
+    assert_error(StorageErrorCode.INVALID_INPUT, lambda: ReadUrlOptions(filename="bad\nname", disposition="attachment"))
+    assert "CANARY" not in repr(ObjectKey("PRIVATE-KEY-CANARY"))
+    assert "CANARY" not in repr(ObjectReadUrl("PRIVATE-URL-CANARY", None))
+    options = PutOptions(0, WriteMode.CREATE_ONLY, metadata={"tag": "PRIVATE-METADATA-CANARY"})
+    assert "CANARY" not in repr(options)
+    with pytest.raises(TypeError):
+        options.metadata["tag"] = "changed"
+    class Signer:
+        url = "https://example.invalid/?X-Amz-Date=20261015T000000Z&X-Amz-Expires=300&X-Amz-SignedHeaders=host"
+        def generate_presigned_url(self, *_args, **_kwargs):
+            return self.url
+    signer = Signer()
+    storage = UserContext().with_object_storage(S3ObjectStorage(signer, "bucket")).object_storage("verify signed URL deadline")
+    assert storage.get_read_url(ObjectKey("key")).expires_at == datetime(2026, 10, 15, 0, 5, tzinfo=timezone.utc)
+    signer.url = "https://example.invalid/permanent"
+    assert_error(StorageErrorCode.UNAVAILABLE, lambda: storage.get_read_url(ObjectKey("key")))
+
+
+def test_object_storage_real_sdk_http_errors_are_not_absence():
+    state = {"status": 403, "calls": 0}
+    class Handler(BaseHTTPRequestHandler):
+        def do_HEAD(self):
+            state["calls"] += 1
+            status = 200 if self.path.rstrip("/") == "/bucket" and state["status"] == 404 else state["status"]
+            self.send_response(status)
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+        def log_message(self, *_):
+            pass
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    client = boto3.client("s3", endpoint_url=f"http://127.0.0.1:{server.server_port}",
+        aws_access_key_id="test", aws_secret_access_key="secret", region_name="us-east-1",
+        config=Config(signature_version="s3v4", retries={"max_attempts": 0}, connect_timeout=0.2,
+                      read_timeout=0.2, s3={"addressing_style": "path"}))
+    provider = S3ObjectStorage(client, "bucket")
+    storage = UserContext().with_object_storage(provider).object_storage("verify HTTP error semantics")
+    key = ObjectKey("PRIVATE-KEY-CANARY")
+    try:
+        for status, expected in ((403, StorageErrorCode.ACCESS_DENIED), (500, StorageErrorCode.UNAVAILABLE),
+            (503, StorageErrorCode.UNAVAILABLE), (408, StorageErrorCode.TIMEOUT),
+            (504, StorageErrorCode.TIMEOUT), (501, StorageErrorCode.UNSUPPORTED)):
+            state["status"] = status
+            error = assert_error(expected, lambda: storage.exists(key))
+            assert error.__context__ is None
+        state["status"] = 404
+        assert not storage.exists(key)
+        state["status"] = 200
+        assert_error(StorageErrorCode.UNAVAILABLE, lambda: storage.exists(key))
+        before = state["calls"]
+        for content, size in ((b"ab", 1), (b"a", 2)):
+            assert_error(StorageErrorCode.INVALID_INPUT, lambda: storage.put_bytes(key, content, PutOptions(size, WriteMode.CREATE_ONLY)))
+        class FailingInput(BytesIO):
+            def read(self, *_):
+                raise OSError("PRIVATE-INPUT-CANARY")
+        stream = FailingInput()
+        assert_error(StorageErrorCode.IO, lambda: storage.put(key, stream, PutOptions(1, WriteMode.CREATE_ONLY)))
+        assert stream.closed
+        assert state["calls"] == before
+        server.shutdown()
+        server.server_close()
+        assert_error(StorageErrorCode.UNAVAILABLE, lambda: storage.exists(key))
+    finally:
+        server.shutdown()
+        server.server_close()
+        provider.close()