diff --git a/pyproject.toml b/pyproject.toml
index 70ae9f09..93bef986 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -31,7 +31,9 @@ classifiers = [
 dependencies = [
     "agentenv-framework-protocol>=0.1.0",
     "boto3>=1.35.0",
+    "certifi>=2024.2.2",  # the public roots the local transfer CA's trust bundle carries
     "click>=8.0.0",
+    "cryptography>=42",  # the local transfer CA
     "httpx>=0.27.0",
     "litellm>=1.96.0,<1.97",
     "mcp>=1.0.0,<2",  # 2.0.0 removed mcp.server.fastmcp, imported at module scope by env/gateway
@@ -39,7 +41,10 @@ dependencies = [
     "a2a-sdk==0.3.26",
     "pydantic>=2.12",  # model_validate(extra=...) reads agent responses tolerantly
     "pymongo>=4.7",
+    "python-multipart>=0.0.13",  # the local grant server's upload forms
     "pyyaml>=6.0",
+    "starlette>=0.40",  # the local grant server
+    "uvicorn>=0.47",  # 0.47 adds ssl_context_factory, how the local grant server gets its TLS context
     "modal>=1.5.2.dev2",  # dev2 adds VM-mode V2 (_experimental_create + vm_runtime)
     "e2b~=2.46.4",  # AsyncSandbox/AsyncTemplate APIs validated against the 2.46 line
 ]
diff --git a/src/agent_env/store/object_store/__init__.py b/src/agent_env/store/object_store/__init__.py
index 11b82296..414804db 100644
--- a/src/agent_env/store/object_store/__init__.py
+++ b/src/agent_env/store/object_store/__init__.py
@@ -1,11 +1,11 @@
 """Backend-agnostic object store abstraction + its implementations.
 
 ``object_store`` defines the abstraction (ObjectStore); ``s3_object_store``,
-``local_object_store`` and ``gcs_object_store`` are implementations. The last needs
+``local`` and ``gcs_object_store`` are implementations. The last needs
 the ``gcp`` extra, so it is not re-exported here: an impl pointer names its module.
 """
 
-from agent_env.store.object_store.local_object_store import LocalFilesystemObjectStore
+from agent_env.store.object_store.local.store import LocalFilesystemObjectStore
 from agent_env.store.object_store.object_store import (
     DEFAULT_CONTENT_TYPE,
     ObjectMetadata,
diff --git a/src/agent_env/store/object_store/local/__init__.py b/src/agent_env/store/object_store/local/__init__.py
new file mode 100644
index 00000000..b91e507a
--- /dev/null
+++ b/src/agent_env/store/object_store/local/__init__.py
@@ -0,0 +1,6 @@
+"""The local filesystem object store, and the HTTPS transfer grants it issues: signed tokens, a local
+certificate authority, and the in-process server that honours them."""
+
+from agent_env.store.object_store.local.store import LocalFilesystemObjectStore
+
+__all__ = ["LocalFilesystemObjectStore"]
diff --git a/src/agent_env/store/object_store/local/grant_server.py b/src/agent_env/store/object_store/local/grant_server.py
new file mode 100644
index 00000000..89725141
--- /dev/null
+++ b/src/agent_env/store/object_store/local/grant_server.py
@@ -0,0 +1,417 @@
+"""The in-process HTTPS server that honours local object-transfer grants.
+
+One route, ``/v1/grants/{token}``, with the semantics of the S3 requests the grants stand in for:
+GET returns an object's exact bytes; PUT writes one object of the claimed type and size; a
+multipart POST writes one object anywhere under the claimed prefix. A token that does not verify,
+has expired or names another method is a 403; a request outside the grant's bounds is a 4xx.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import functools
+import ipaddress
+import logging
+import os
+import platform
+import socket
+import subprocess
+import threading
+import time
+from pathlib import Path
+from collections.abc import AsyncIterator
+from typing import TYPE_CHECKING, BinaryIO
+
+import uvicorn
+from python_multipart.exceptions import FormParserError
+from python_multipart.multipart import MultipartParser, parse_options_header
+from starlette.applications import Starlette
+from starlette.requests import ClientDisconnect, Request
+from starlette.responses import PlainTextResponse, Response, StreamingResponse
+from starlette.routing import Route
+from uvicorn.protocols.http.h11_impl import H11Protocol
+
+from agent_env.store.base import GrantUnavailableError
+from agent_env.store.object_store.local.tls import check_local_host, local_ca, server_context
+from agent_env.store.object_store.local.tokens import (
+    GrantClaims,
+    GrantSigner,
+    InvalidGrantError,
+    Op,
+    store_id,
+)
+from agent_env.store.object_store.object_store import DEFAULT_CONTENT_TYPE
+
+if TYPE_CHECKING:
+    from agent_env.store.object_store.local.store import LocalFilesystemObjectStore
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_ADVERTISE_HOST = "host.docker.internal"
+_START_TIMEOUT_SECONDS = 10
+_OPS_BY_METHOD = {"GET": "get", "HEAD": "get", "PUT": "put", "POST": "post"}
+_UPLOAD_KEY_FIELD = "key"  # the form fields a policy upload names its key and file by, as S3's do
+_UPLOAD_FILE_FIELD = "file"
+_MAX_FORM_FIELDS = 32
+_MAX_FORM_FIELD_BYTES = 64 * 1024
+_MAX_PART_HEADER_BYTES = 16 * 1024
+_DOWNLOAD_CHUNK_BYTES = 1024 * 1024
+
+_servers: dict[tuple[str, str], GrantServer] = {}
+_servers_lock = threading.Lock()
+
+
+class _UnloggedH11Protocol(H11Protocol):
+    """HTTP/1.1 without access logs, which would record grant tokens. uvicorn's own switch clears
+    the process-wide access logger, silencing any other server in this process."""
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.access_log = False
+
+
+class _Rejected(Exception):
+    def __init__(self, status: int, reason: str) -> None:
+        super().__init__(reason)
+        self.status = status
+        self.reason = reason
+
+
+def grant_server(bind_host: str | None, advertise_host: str | None) -> GrantServer:
+    """This process's grant server for a bind and advertised host, created on first use; neither
+    given means this platform's default bind address and ``host.docker.internal``."""
+    key = (bind_host or default_bind_host(), advertise_host or DEFAULT_ADVERTISE_HOST)
+    with _servers_lock:
+        server = _servers.get(key)
+        if server is None:
+            server = _servers[key] = GrantServer(*key)
+        return server
+
+
+@functools.cache
+def default_bind_host() -> str:
+    """Where the grant server listens unless told otherwise. On macOS and Windows, Docker reaches
+    the host's loopback as ``host.docker.internal``; native Linux maps that name to the default
+    bridge's gateway (``--add-host host.docker.internal:host-gateway``), so the server listens
+    there, reachable from containers and not beyond the host."""
+    if platform.system() != "Linux":
+        return "127.0.0.1"
+    try:
+        gateway = subprocess.run(
+            ["docker", "network", "inspect", "bridge", "--format", "{{range .IPAM.Config}}{{.Gateway}} {{end}}"],
+            capture_output=True, text=True, timeout=10, check=True,
+        ).stdout.split()
+        return str(ipaddress.ip_address(gateway[0]))
+    except (OSError, subprocess.SubprocessError, IndexError, ValueError) as e:
+        logger.warning(
+            "Could not find the Docker bridge gateway (%s); the local grant server listens on 127.0.0.1, "
+            "which containers may not reach. Set grant_bind_host under [stores.object] to override.", e,
+        )
+        return "127.0.0.1"
+
+
+class GrantServer:
+    """Serves the grants of the local stores registered with it, over HTTPS on one port. Started by
+    the first grant it issues; it runs in a daemon thread for the rest of the process."""
+
+    def __init__(self, bind_host: str, advertise_host: str) -> None:
+        self.bind_host = bind_host
+        self.advertise_host = advertise_host
+        self._signer = GrantSigner()
+        self._stores: dict[str, LocalFilesystemObjectStore] = {}
+        self._lock = threading.Lock()
+        self._port: int | None = None
+        self._uvicorn: uvicorn.Server | None = None
+        self._thread: threading.Thread | None = None
+        self._app = Starlette(routes=[Route("/v1/grants/{token}", self._handle, methods=["GET", "PUT", "POST"])])
+
+    def issue(
+        self,
+        store: LocalFilesystemObjectStore,
+        op: Op,
+        expires: int,
+        *,
+        key: str | None = None,
+        prefix: str | None = None,
+        max_bytes: int | None = None,
+        content_type: str | None = None,
+    ) -> str:
+        """The URL of a new grant for ``op`` on ``store``, valid until ``expires`` (epoch seconds)."""
+        claims = GrantClaims(
+            op=op, store=self._register(store), expires=expires, key=key, prefix=prefix,
+            max_bytes=max_bytes, content_type=content_type,
+        )
+        port = self._start()
+        host = f"[{self.advertise_host}]" if ":" in self.advertise_host else self.advertise_host
+        return f"https://{host}:{port}/v1/grants/{self._signer.sign(claims)}"
+
+    def close(self) -> None:
+        """Stop serving; grants already issued stop working."""
+        with self._lock:
+            if self._uvicorn is not None:
+                self._uvicorn.should_exit = True
+            if self._thread is not None:
+                self._thread.join(timeout=_START_TIMEOUT_SECONDS)
+            self._port = self._uvicorn = self._thread = None
+
+    def _register(self, store: LocalFilesystemObjectStore) -> str:
+        name = store_id(str(store.root.resolve()))
+        with self._lock:
+            self._stores[name] = store
+        return name
+
+    def _start(self) -> int:
+        with self._lock:
+            if self._port is not None:
+                return self._port
+            address = ipaddress.ip_address(self.bind_host)
+            sock = socket.socket(socket.AF_INET6 if address.version == 6 else socket.AF_INET, socket.SOCK_STREAM)
+            try:
+                sock.bind((self.bind_host, 0))
+                hosts = ["localhost", "host.docker.internal", "127.0.0.1", self.advertise_host]
+                with contextlib.suppress(ValueError):  # a name the CA cannot vouch for would void the certificate
+                    check_local_host(self.bind_host)
+                    hosts.append(self.bind_host)
+                context = server_context(local_ca(), hosts)
+            except (OSError, ValueError) as e:
+                sock.close()
+                raise GrantUnavailableError(f"the local grant server cannot listen on {self.bind_host}: {e}") from e
+            config = uvicorn.Config(
+                self._app, lifespan="off", log_config=None, http=_UnloggedH11Protocol,
+                ssl_context_factory=lambda _config, _default: context,
+            )
+            server = uvicorn.Server(config)
+            failure: list[BaseException] = []
+
+            def serve() -> None:
+                try:
+                    asyncio.run(server.serve(sockets=[sock]))
+                except BaseException as e:  # uvicorn ends a failed startup with SystemExit
+                    failure.append(e)
+
+            thread = threading.Thread(target=serve, name="agent-env-grant-server", daemon=True)
+            thread.start()
+            deadline = time.monotonic() + _START_TIMEOUT_SECONDS
+            while not server.started:
+                if failure or not thread.is_alive() or time.monotonic() > deadline:
+                    server.should_exit = True
+                    sock.close()
+                    raise GrantUnavailableError(
+                        f"the local grant server did not start on {self.bind_host}"
+                        + (f": {failure[0]!r}" if failure else "")
+                    )
+                time.sleep(0.01)
+            self._uvicorn, self._thread = server, thread
+            self._port = sock.getsockname()[1]
+            logger.info("Local grant server listening on %s:%d", self.bind_host, self._port)
+            return self._port
+
+    async def _handle(self, request: Request) -> Response:
+        try:
+            claims = self._signer.verify(request.path_params["token"], now=time.time())
+        except (InvalidGrantError, ValueError, KeyError, TypeError):
+            return PlainTextResponse("The grant is invalid or has expired.", status_code=403)
+        store = self._stores.get(claims.store)
+        if store is None or claims.op != _OPS_BY_METHOD.get(request.method):
+            return PlainTextResponse("The grant does not allow this request.", status_code=403)
+        try:
+            if claims.op == "get":
+                return _get(store, claims, head=request.method == "HEAD")
+            if claims.op == "put":
+                await _put(store, claims, request)
+                return Response(status_code=200)
+            await _post(store, claims, request)
+            return Response(status_code=204)
+        except _Rejected as r:
+            return PlainTextResponse(r.reason, status_code=r.status)
+        except ClientDisconnect:
+            return PlainTextResponse("The upload ended early.", status_code=400)
+        except Exception:
+            logger.exception("The local grant server failed a %s request", request.method)
+            return PlainTextResponse("The object store failed the transfer.", status_code=500)
+
+
+def _path(store: LocalFilesystemObjectStore, key: str) -> Path:
+    try:
+        return store._resolve(key)
+    except ValueError:
+        raise _Rejected(403, "The grant's key is not an object key of this store.") from None
+
+
+def _get(store: LocalFilesystemObjectStore, claims: GrantClaims, *, head: bool = False) -> Response:
+    path = _path(store, claims.key)
+    # Opened under the key's lock and typed from that open file, so the bytes sent and their type are one write's.
+    with store._locked(path, shared=True):
+        try:
+            f = path.open("rb")
+        except (FileNotFoundError, IsADirectoryError, NotADirectoryError):
+            raise _Rejected(404, "No object exists at this grant's key.") from None
+        st = os.fstat(f.fileno())
+        content_type = store._read_content_type(path, st)
+    headers = {"Content-Length": str(st.st_size)}
+    if head:  # the size and type alone: a StreamingResponse would read the whole object to send nothing
+        f.close()
+        return Response(media_type=content_type or DEFAULT_CONTENT_TYPE, headers=headers)
+    return StreamingResponse(_chunks(f), media_type=content_type or DEFAULT_CONTENT_TYPE, headers=headers)
+
+
+async def _chunks(f: BinaryIO) -> AsyncIterator[bytes]:
+    try:
+        while chunk := await asyncio.to_thread(f.read, _DOWNLOAD_CHUNK_BYTES):
+            yield chunk
+    finally:
+        f.close()
+
+
+async def _put(store: LocalFilesystemObjectStore, claims: GrantClaims, request: Request) -> None:
+    if request.headers.get("content-type") != claims.content_type:
+        raise _Rejected(403, f"The upload must declare Content-Type {claims.content_type}.")
+    declared = request.headers.get("content-length")
+    try:
+        if declared is not None and int(declared) > claims.max_bytes:
+            raise _Rejected(400, "The upload is larger than the grant allows.")
+    except ValueError:
+        raise _Rejected(400, "The upload's Content-Length is not a number.") from None
+    path = _path(store, claims.key)
+    with store._staged() as staged:
+        received = 0
+        with staged.open("wb") as out:
+            async for chunk in request.stream():
+                received += len(chunk)
+                if received > claims.max_bytes:
+                    raise _Rejected(400, "The upload is larger than the grant allows.")
+                out.write(chunk)
+        store._commit(staged, path, claims.content_type, allow_overwrite=True)
+
+
+async def _post(store: LocalFilesystemObjectStore, claims: GrantClaims, request: Request) -> None:
+    media_type, params = parse_options_header(request.headers.get("content-type"))
+    if media_type != b"multipart/form-data" or b"boundary" not in params:
+        raise _Rejected(400, "The upload must be multipart/form-data.")
+    with contextlib.ExitStack() as stack:
+        upload = _PolicyUpload(store, claims, stack)
+        parser = MultipartParser(params[b"boundary"], upload.callbacks())
+        try:
+            async for chunk in request.stream():
+                parser.write(chunk)
+            parser.finalize()
+        except FormParserError:
+            raise _Rejected(400, "The upload is not a valid multipart form.") from None
+        upload.commit()
+
+
+class _PolicyUpload:
+    """One multipart POST under a policy grant, read as S3 reads one: form fields, then the file,
+    which is written to a staged file as it arrives. Fields after the file are ignored."""
+
+    def __init__(self, store: LocalFilesystemObjectStore, claims: GrantClaims, stack: contextlib.ExitStack) -> None:
+        self._store = store
+        self._claims = claims
+        self._stack = stack
+        self._fields: dict[str, bytes] = {}
+        self._headers: dict[bytes, bytes] = {}
+        self._header_bytes = 0
+        self._header_name = b""
+        self._header_value = b""
+        self._part_name: str | None = None
+        self._part_data = bytearray()
+        self._path: Path | None = None
+        self._staged: Path | None = None
+        self._out: BinaryIO | None = None
+        self._received = 0
+        self._file_done = False
+
+    def callbacks(self) -> dict:
+        return {
+            "on_part_begin": self._on_part_begin,
+            "on_header_field": self._on_header_field,
+            "on_header_value": self._on_header_value,
+            "on_header_end": self._on_header_end,
+            "on_headers_finished": self._on_headers_finished,
+            "on_part_data": self._on_part_data,
+            "on_part_end": self._on_part_end,
+        }
+
+    def commit(self) -> None:
+        if self._staged is None or not self._file_done:
+            raise _Rejected(400, f"The upload has no {_UPLOAD_FILE_FIELD!r} field.")
+        self._out.close()
+        content_type = self._field("Content-Type") or DEFAULT_CONTENT_TYPE
+        self._store._commit(self._staged, self._path, content_type, allow_overwrite=True)
+
+    def _on_part_begin(self) -> None:
+        self._headers = {}
+        self._header_bytes = 0
+        self._part_name = None
+        self._part_data = bytearray()
+
+    def _on_header_field(self, data: bytes, start: int, end: int) -> None:
+        self._count_header_bytes(end - start)
+        self._header_name += data[start:end]
+
+    def _on_header_value(self, data: bytes, start: int, end: int) -> None:
+        self._count_header_bytes(end - start)
+        self._header_value += data[start:end]
+
+    def _count_header_bytes(self, count: int) -> None:
+        self._header_bytes += count
+        if self._header_bytes > _MAX_PART_HEADER_BYTES:
+            raise _Rejected(400, "A form part's headers are too large.")
+
+    def _on_header_end(self) -> None:
+        self._headers[self._header_name.lower()] = self._header_value
+        self._header_name = self._header_value = b""
+
+    def _on_headers_finished(self) -> None:
+        if self._file_done:
+            return
+        _, options = parse_options_header(self._headers.get(b"content-disposition"))
+        name = options.get(b"name")
+        if name is None:
+            raise _Rejected(400, "A form field has no name.")
+        self._part_name = name.decode("utf-8", "replace")
+        if self._part_name == _UPLOAD_FILE_FIELD:
+            self._path = _path(self._store, self._upload_key())
+            self._staged = self._stack.enter_context(self._store._staged())
+            self._out = self._stack.enter_context(self._staged.open("wb"))
+        elif len(self._fields) >= _MAX_FORM_FIELDS:
+            raise _Rejected(400, "The upload has too many form fields.")
+
+    def _on_part_data(self, data: bytes, start: int, end: int) -> None:
+        if self._file_done:
+            return
+        if self._part_name == _UPLOAD_FILE_FIELD:
+            self._received += end - start
+            if self._received > self._claims.max_bytes:
+                raise _Rejected(400, "The upload is larger than the grant allows.")
+            self._out.write(data[start:end])
+            return
+        if len(self._part_data) + end - start > _MAX_FORM_FIELD_BYTES:
+            raise _Rejected(400, "A form field is too large.")
+        self._part_data += data[start:end]
+
+    def _on_part_end(self) -> None:
+        if self._file_done:
+            return
+        if self._part_name == _UPLOAD_FILE_FIELD:
+            self._file_done = True
+        elif self._part_name is not None:
+            self._fields[self._part_name] = bytes(self._part_data)
+
+    def _field(self, name: str) -> str | None:
+        value = self._fields.get(name)
+        return None if value is None else value.decode("utf-8", "replace")
+
+    def _upload_key(self) -> str:
+        """The key the upload names, which must come before the file, be a plain relative path,
+        and lie under the grant's prefix."""
+        key = self._field(_UPLOAD_KEY_FIELD)
+        if key is None:
+            raise _Rejected(400, f"The upload must name its {_UPLOAD_KEY_FIELD!r} before its file.")
+        parts = key.split("/")
+        if key.startswith("/") or "\\" in key or "\x00" in key or any(p in ("", ".", "..") for p in parts):
+            raise _Rejected(403, "The upload key is not a normalized relative path.")
+        if not key.startswith(self._claims.prefix):
+            raise _Rejected(403, "The upload key is outside the grant's prefix.")
+        return key
diff --git a/src/agent_env/store/object_store/local_object_store.py b/src/agent_env/store/object_store/local/store.py
similarity index 77%
rename from src/agent_env/store/object_store/local_object_store.py
rename to src/agent_env/store/object_store/local/store.py
index 58038aa1..acbef743 100644
--- a/src/agent_env/store/object_store/local_object_store.py
+++ b/src/agent_env/store/object_store/local/store.py
@@ -5,6 +5,7 @@
 import contextlib
 import errno
 import hashlib
+import ipaddress
 import json
 import logging
 import os
@@ -17,12 +18,34 @@
 from pathlib import Path
 from typing import BinaryIO
 
+from agentenv_protocol.transfers import HttpGetGrant, HttpPostPolicyGrant, HttpPutGrant
+
+from agent_env.config.paths import state_root
 from agent_env.store.base import ObjectAlreadyExistsError, ObjectNotFoundError
 
 if os.name != "nt":
     import fcntl
 from agent_env.store.local_state import ensure_state_dir
-from agent_env.store.object_store.object_store import DEFAULT_CONTENT_TYPE, ObjectMetadata, ObjectStore
+from agent_env.store.object_store.local.grant_server import grant_server
+from agent_env.store.object_store.local.tls import check_local_host
+from agent_env.store.object_store.local.tokens import Op
+from agent_env.store.object_store.object_store import (
+    DEFAULT_CONTENT_TYPE,
+    ObjectMetadata,
+    ObjectStore,
+    UploadPolicy,
+)
+
+logger = logging.getLogger(__name__)
+
+_META_DIR = ".agentenv-meta"  # each object's content type, at the object's own key
+_STAGING_DIR = ".agentenv-tmp"  # files being written, inside the root so a rename into place is atomic
+_RESERVED = (".gitignore", _META_DIR, _STAGING_DIR)
+# A filesystem without hard links: a no-overwrite write relies on the key's lock instead.
+_NO_HARD_LINKS = frozenset({errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EXDEV, errno.EMLINK})
+_LOCK_STRIPES = 256  # keys share this many lock files, so locking leaves no file behind per key
+_STAGED_PREFIX = "staged-"
+_STAGED_GRACE_SECONDS = 60  # a staged file younger than this may not be locked by its writer yet
 
 logger = logging.getLogger(__name__)
 
@@ -37,13 +60,29 @@
 
 
 class LocalFilesystemObjectStore(ObjectStore):
-    """ObjectStore backed by a local directory tree.
+    """ObjectStore backed by a local directory tree, ``<state root>/object_store`` unless ``root`` is given.
 
     A write lands whole or not at all, and keeps any content type but the default. The root-level ``.gitignore``,
-    ``.agentenv-meta`` and ``.agentenv-tmp`` keys are reserved for the store's own files."""
+    ``.agentenv-meta`` and ``.agentenv-tmp`` keys are reserved for the store's own files.
+
+    Its HTTPS transfer grants are served by a server in this process, started by the first grant: it listens on
+    ``grant_bind_host`` (by default loopback, or the Docker bridge gateway on Linux) and grant URLs name
+    ``grant_advertise_host`` (by default ``host.docker.internal``), which must be a local name or a loopback or
+    private address. A grant works until it expires or this process exits."""
 
-    def __init__(self, root: str) -> None:
-        self._root = Path(root)
+    def __init__(
+        self, root: str | None = None, *, grant_bind_host: str | None = None, grant_advertise_host: str | None = None
+    ) -> None:
+        if grant_bind_host is not None:
+            try:
+                ipaddress.ip_address(grant_bind_host)
+            except ValueError:
+                raise ValueError(f"grant_bind_host must be an IP address, got {grant_bind_host!r}") from None
+        if grant_advertise_host is not None:
+            check_local_host(grant_advertise_host)
+        self._root = Path(root) if root is not None else state_root() / "object_store"
+        self._grant_bind_host = grant_bind_host
+        self._grant_advertise_host = grant_advertise_host
         self._swept = False
 
     @property
@@ -138,6 +177,41 @@ def get_object_key(self, object_url: str) -> str:
             raise ValueError(f"{object_url!r} is not an object in {root}.")
         return path.relative_to(root).as_posix()
 
+    def issue_read_grant(self, object_url: str, *, expires_in: int = 3600) -> HttpGetGrant:
+        url, expires_at = self._grant("get", expires_in, key=self._grant_key(object_url))
+        return HttpGetGrant(kind="http-get", url=url, expires_at=expires_at)
+
+    def issue_write_grant(
+        self, object_url: str, *, media_type: str, max_bytes: int, expires_in: int = 3600
+    ) -> HttpPutGrant:
+        url, expires_at = self._grant(
+            "put", expires_in, key=self._grant_key(object_url), max_bytes=max_bytes, content_type=media_type
+        )
+        return HttpPutGrant(kind="http-put", url=url, expires_at=expires_at, headers={"Content-Type": media_type})
+
+    def issue_upload_policy(self, prefix_url: str, *, max_object_bytes: int, expires_in: int) -> UploadPolicy:
+        key = self.get_object_key(prefix_url)
+        url, expires_at = self._grant(
+            "post", expires_in, prefix="" if key == "." else key + "/", max_bytes=max_object_bytes
+        )
+        return UploadPolicy(
+            write=HttpPostPolicyGrant(kind="http-post-policy", url=url, fields={}, path_field="key", file_field="file"),
+            expires_at=expires_at,
+        )
+
+    def _grant(self, op: Op, expires_in: int, **scope) -> tuple[str, datetime]:
+        if expires_in <= 0:
+            raise ValueError(f"expires_in must be positive, got {expires_in}")
+        expires = int(time.time()) + expires_in
+        server = grant_server(self._grant_bind_host, self._grant_advertise_host)
+        return server.issue(self, op, expires, **scope), datetime.fromtimestamp(expires, timezone.utc)
+
+    def _grant_key(self, object_url: str) -> str:
+        """The key of the one object at ``object_url``, refusing the store's own files."""
+        key = self.get_object_key(object_url)
+        self._resolve(key)
+        return key
+
     def _resolve(self, key: str) -> Path:
         root = self._root.resolve()
         resolved = (self._root / key).resolve()
diff --git a/src/agent_env/store/object_store/local/tls.py b/src/agent_env/store/object_store/local/tls.py
new file mode 100644
index 00000000..d3ae6d90
--- /dev/null
+++ b/src/agent_env/store/object_store/local/tls.py
@@ -0,0 +1,244 @@
+"""The local certificate authority behind the grant server's HTTPS.
+
+One CA per user, created on first use under ``<state root>/tls/``. It is name-constrained to
+local names and private addresses, so even a leaked key can only vouch for those, and it is
+never installed in the host's trust stores: only the containers agent-env starts trust it,
+through ``ca-bundle.pem`` (the public roots plus this CA, for ``SSL_CERT_FILE``, which replaces
+the defaults) or ``ca.pem`` (the CA alone, for trust stores that add to the defaults). Each CA's
+two files sit alone in a directory of their own under ``tls/trust/``, which can be copied into a
+container whole.
+"""
+
+from __future__ import annotations
+
+import functools
+import ipaddress
+import os
+import secrets
+import ssl
+import tempfile
+from collections.abc import Iterable
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import certifi
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID
+
+from agent_env.config.paths import state_root
+from agent_env.store.local_state import ensure_state_dir
+
+# What the CA may vouch for: the names containers reach the host by, and loopback and private addresses.
+LOCAL_NAMES = ("localhost", "host.docker.internal")
+LOCAL_NETWORKS = tuple(
+    ipaddress.ip_network(n)
+    for n in ("127.0.0.0/8", "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "::1/128", "fc00::/7")
+)
+
+_CA_LIFETIME = timedelta(days=5 * 365)
+_CA_RENEW_BEFORE = timedelta(days=30)  # replaced this long before it expires
+_CLOCK_SKEW = timedelta(minutes=5)
+
+_CA_FILE = "ca.key.pem"  # the CA's key and certificate, private to the user
+_TRUST_DIR = "trust"  # the public files, readable by any user a container runs as
+_CERT_FILE = "ca.pem"
+_BUNDLE_FILE = "ca-bundle.pem"
+_PEM_CERT = b"-----BEGIN CERTIFICATE-----"
+
+
+@dataclass(frozen=True)
+class LocalCA:
+    """The CA's key and certificate, and the trust files derived from it in ``trust_dir``."""
+
+    key: ec.EllipticCurvePrivateKey
+    cert: x509.Certificate
+    trust_dir: Path
+    cert_path: Path
+    bundle_path: Path
+
+
+def check_local_host(host: str) -> None:
+    """Refuse a host the CA cannot vouch for: a name other than the local ones, or a public address."""
+    try:
+        address = ipaddress.ip_address(host)
+    except ValueError:
+        name = host.lower().rstrip(".")
+        if any(name == n or name.endswith("." + n) for n in LOCAL_NAMES):
+            return
+        raise ValueError(
+            f"{host!r} is not a local name; the local transfer CA vouches only for {', '.join(LOCAL_NAMES)} "
+            "and loopback or private addresses"
+        ) from None
+    if not any(address in network for network in LOCAL_NETWORKS):
+        raise ValueError(f"{host!r} is not a loopback or private address, which the local transfer CA vouches for")
+
+
+def local_ca() -> LocalCA:
+    """This user's local CA, created on first use and replaced when it nears expiry. A process keeps the CA
+    it first loads, so its grant server and the containers it starts agree on it whatever another process
+    does to the one on disk."""
+    return _process_ca(state_root() / "tls")
+
+
+@functools.cache
+def _process_ca(tls_dir: Path) -> LocalCA:
+    ensure_state_dir(tls_dir)
+    ca_file = tls_dir / _CA_FILE
+    loaded = _load(ca_file)
+    if loaded is None:
+        key, cert = _new_ca()
+        pem = _key_pem(key) + cert.public_bytes(serialization.Encoding.PEM)
+        # A first CA is claimed with a hard link, so two processes starting at once agree on one.
+        replacing = ca_file.exists()
+        staged = _write_private(tls_dir, pem)
+        try:
+            if replacing:
+                os.replace(staged, ca_file)
+            else:
+                try:
+                    os.link(staged, ca_file)
+                except FileExistsError:
+                    pass
+        finally:
+            staged.unlink(missing_ok=True)
+        loaded = _load(ca_file) or (key, cert)
+    key, cert = loaded
+    # Written once per CA, so renewing the CA never changes the files another process's containers trust.
+    trust_dir = tls_dir / _TRUST_DIR / cert.fingerprint(hashes.SHA256()).hex()[:16]
+    trust_dir.mkdir(parents=True, exist_ok=True)
+    trust_dir.chmod(0o755)
+    cert_pem = cert.public_bytes(serialization.Encoding.PEM)
+    cert_path = _write_if_changed(trust_dir / _CERT_FILE, cert_pem)
+    bundle_path = _write_if_changed(trust_dir / _BUNDLE_FILE, Path(certifi.where()).read_bytes().rstrip() + b"\n" + cert_pem)
+    return LocalCA(key=key, cert=cert, trust_dir=trust_dir, cert_path=cert_path, bundle_path=bundle_path)
+
+
+def server_context(ca: LocalCA, hosts: Iterable[str]) -> ssl.SSLContext:
+    """A TLS server context presenting a new certificate, signed by ``ca``, for each of ``hosts``."""
+    key = ec.generate_private_key(ec.SECP256R1())
+    names = list(dict.fromkeys(hosts))
+    now = datetime.now(UTC)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")]))
+        .issuer_name(ca.cert.subject)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now - _CLOCK_SKEW)
+        .not_valid_after(ca.cert.not_valid_after_utc)  # a server keeps one certificate for its whole life
+        .add_extension(x509.SubjectAlternativeName([_general_name(n) for n in names]), critical=False)
+        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
+        .add_extension(_key_usage(digital_signature=True), critical=True)
+        .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False)
+        .add_extension(x509.SubjectKeyIdentifier.from_public_key(key.public_key()), critical=False)
+        .add_extension(
+            x509.AuthorityKeyIdentifier.from_issuer_public_key(ca.key.public_key()), critical=False
+        )
+        .sign(ca.key, hashes.SHA256())
+    )
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.minimum_version = ssl.TLSVersion.TLSv1_2
+    # ssl loads a certificate chain only from a file; this one lives just long enough to be read.
+    with tempfile.TemporaryDirectory() as tmp:
+        chain = Path(tmp) / "server.pem"
+        chain.write_bytes(_key_pem(key) + cert.public_bytes(serialization.Encoding.PEM))
+        context.load_cert_chain(chain)
+    return context
+
+
+def _new_ca() -> tuple[ec.EllipticCurvePrivateKey, x509.Certificate]:
+    key = ec.generate_private_key(ec.SECP256R1())
+    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, f"agent-env local transfer CA {secrets.token_hex(4)}")])
+    now = datetime.now(UTC)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(name)
+        .issuer_name(name)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now - _CLOCK_SKEW)
+        .not_valid_after(now + _CA_LIFETIME)
+        .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
+        .add_extension(_key_usage(key_cert_sign=True, crl_sign=True), critical=True)
+        .add_extension(_name_constraints(), critical=True)
+        .add_extension(x509.SubjectKeyIdentifier.from_public_key(key.public_key()), critical=False)
+        .sign(key, hashes.SHA256())
+    )
+    return key, cert
+
+
+def _load(ca_file: Path) -> tuple[ec.EllipticCurvePrivateKey, x509.Certificate] | None:
+    """The CA in ``ca_file``, or None when it is missing, unreadable, nearly expired, or not one this
+    release would create."""
+    try:
+        data = ca_file.read_bytes()
+        key_pem, marker, cert_pem = data.partition(_PEM_CERT)
+        key = serialization.load_pem_private_key(key_pem, password=None)
+        cert = x509.load_pem_x509_certificate(marker + cert_pem)
+        constraints = cert.extensions.get_extension_for_class(x509.NameConstraints).value
+    except (OSError, ValueError, TypeError, x509.ExtensionNotFound):
+        return None
+    if not isinstance(key, ec.EllipticCurvePrivateKey) or key.public_key() != cert.public_key():
+        return None
+    if constraints != _name_constraints() or cert.not_valid_after_utc - datetime.now(UTC) < _CA_RENEW_BEFORE:
+        return None
+    return key, cert
+
+
+def _name_constraints() -> x509.NameConstraints:
+    return x509.NameConstraints(
+        permitted_subtrees=[x509.DNSName(n) for n in LOCAL_NAMES] + [x509.IPAddress(n) for n in LOCAL_NETWORKS],
+        excluded_subtrees=None,
+    )
+
+
+def _key_usage(*, digital_signature: bool = False, key_cert_sign: bool = False, crl_sign: bool = False) -> x509.KeyUsage:
+    return x509.KeyUsage(
+        digital_signature=digital_signature,
+        content_commitment=False,
+        key_encipherment=False,
+        data_encipherment=False,
+        key_agreement=False,
+        key_cert_sign=key_cert_sign,
+        crl_sign=crl_sign,
+        encipher_only=False,
+        decipher_only=False,
+    )
+
+
+def _general_name(host: str) -> x509.GeneralName:
+    try:
+        return x509.IPAddress(ipaddress.ip_address(host))
+    except ValueError:
+        return x509.DNSName(host)
+
+
+def _key_pem(key: ec.EllipticCurvePrivateKey) -> bytes:
+    return key.private_bytes(
+        serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()
+    )
+
+
+def _write_private(directory: Path, data: bytes) -> Path:
+    fd, name = tempfile.mkstemp(dir=directory, prefix=".ca.")  # created 0600
+    with os.fdopen(fd, "wb") as f:
+        f.write(data)
+    return Path(name)
+
+
+def _write_if_changed(path: Path, data: bytes) -> Path:
+    try:
+        if path.read_bytes() == data:
+            return path
+    except OSError:
+        pass
+    staged = _write_private(path.parent, data)
+    try:
+        staged.chmod(0o644)
+        os.replace(staged, path)
+    finally:
+        staged.unlink(missing_ok=True)
+    return path
diff --git a/src/agent_env/store/object_store/local/tokens.py b/src/agent_env/store/object_store/local/tokens.py
new file mode 100644
index 00000000..2af0c3cb
--- /dev/null
+++ b/src/agent_env/store/object_store/local/tokens.py
@@ -0,0 +1,119 @@
+"""Grant tokens: signed claims naming one operation on one key or key prefix, until an expiry.
+
+A token is ``v1.<claims>.<signature>``: base64url JSON claims and their HMAC-SHA256 under a key
+that lives only in this process's memory, so a token stops working when the process exits and
+the server trusts nothing but the claims it signed.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import hashlib
+import hmac
+import json
+import secrets
+from dataclasses import dataclass, field
+from typing import Literal
+
+_VERSION = "v1"
+_OPS = ("get", "put", "post")
+
+Op = Literal["get", "put", "post"]
+
+
+class InvalidGrantError(Exception):
+    """A token this process did not sign, or one that has expired."""
+
+
+@dataclass(frozen=True)
+class GrantClaims:
+    """What a grant allows. ``key`` names the object of a get or put; ``prefix``, ending in ``/``,
+    the keys a post may write. ``max_bytes`` bounds an upload and ``content_type`` is the type a
+    put must declare."""
+
+    op: Op
+    store: str
+    expires: int
+    key: str | None = None
+    prefix: str | None = None
+    max_bytes: int | None = None
+    content_type: str | None = None
+    grant_id: str = field(default_factory=lambda: secrets.token_hex(8))
+
+    def __post_init__(self) -> None:
+        if self.op not in _OPS:
+            raise ValueError(f"op must be one of {_OPS}, got {self.op!r}")
+        if self.op == "post":
+            if self.key is not None or self.prefix is None or not (self.prefix == "" or self.prefix.endswith("/")):
+                raise ValueError("a post grant names a prefix ending in '/' (or the empty root prefix), not a key")
+        elif self.key is None or self.prefix is not None:
+            raise ValueError(f"a {self.op} grant names one key, not a prefix")
+        if self.op != "get" and (not isinstance(self.max_bytes, int) or self.max_bytes < 0):
+            raise ValueError(f"a {self.op} grant needs a non-negative max_bytes")
+        if (self.op == "put") != (self.content_type is not None):
+            raise ValueError("a put grant, and only a put grant, names a content type")
+
+
+class GrantSigner:
+    """Signs and verifies grant tokens with a key generated for this process."""
+
+    def __init__(self) -> None:
+        self._key = secrets.token_bytes(32)
+
+    def sign(self, claims: GrantClaims) -> str:
+        body = {
+            "op": claims.op,
+            "s": claims.store,
+            "exp": claims.expires,
+            "id": claims.grant_id,
+            "k": claims.key,
+            "p": claims.prefix,
+            "max": claims.max_bytes,
+            "ct": claims.content_type,
+        }
+        payload = _b64encode(json.dumps({k: v for k, v in body.items() if v is not None}, separators=(",", ":")).encode())
+        return f"{_VERSION}.{payload}.{_b64encode(self._mac(payload))}"
+
+    def verify(self, token: str, *, now: float) -> GrantClaims:
+        """The claims of ``token`` if this signer signed it and it has not expired at ``now``."""
+        version, _, rest = token.partition(".")
+        payload, _, signature = rest.partition(".")
+        if version != _VERSION or not payload or not signature:
+            raise InvalidGrantError("not a grant token")
+        try:
+            valid = hmac.compare_digest(_b64decode(signature), self._mac(payload))
+        except (binascii.Error, ValueError):
+            valid = False
+        if not valid:
+            raise InvalidGrantError("the grant's signature does not match")
+        body = json.loads(_b64decode(payload))
+        claims = GrantClaims(
+            op=body["op"],
+            store=body["s"],
+            expires=body["exp"],
+            grant_id=body["id"],
+            key=body.get("k"),
+            prefix=body.get("p"),
+            max_bytes=body.get("max"),
+            content_type=body.get("ct"),
+        )
+        if claims.expires <= now:
+            raise InvalidGrantError("the grant has expired")
+        return claims
+
+    def _mac(self, payload: str) -> bytes:
+        return hmac.new(self._key, f"{_VERSION}.{payload}".encode(), hashlib.sha256).digest()
+
+
+def store_id(root: str) -> str:
+    """A short, stable name for the store at ``root``, so a token names its store without its path."""
+    return hashlib.sha256(root.encode()).hexdigest()[:16]
+
+
+def _b64encode(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
+
+
+def _b64decode(text: str) -> bytes:
+    return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))
diff --git a/tst/unit/config/test_fixture_prefix_keys.py b/tst/unit/config/test_fixture_prefix_keys.py
index e4de0ec0..0b19a3e8 100644
--- a/tst/unit/config/test_fixture_prefix_keys.py
+++ b/tst/unit/config/test_fixture_prefix_keys.py
@@ -12,7 +12,7 @@
 from agent_env.artifact.store import reset_artifact_store
 from agent_env.config import configure, get_config, set_object_store
 from agent_env.env.snapshot_store import _snapshot_servicedb
-from agent_env.store.object_store.local_object_store import LocalFilesystemObjectStore
+from agent_env.store.object_store.local.store import LocalFilesystemObjectStore
 from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep
 from agent_env.task_step.task_steps.verifiers.judge_utils.trajectory_filter import TrajectoryFilter
 from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep
diff --git a/tst/unit/explorer/explorer_app_test.py b/tst/unit/explorer/explorer_app_test.py
index 1925f02b..8437518b 100644
--- a/tst/unit/explorer/explorer_app_test.py
+++ b/tst/unit/explorer/explorer_app_test.py
@@ -19,7 +19,7 @@
 from agent_env.config.errors import ConfigError
 from agent_env.runner import store as run_store
 from agent_env.runner.local_runner import LocalRunner
-from agent_env.store.object_store.local_object_store import LocalFilesystemObjectStore
+from agent_env.store.object_store.local.store import LocalFilesystemObjectStore
 from agent_env.store.document_store.sqlite_document_store import LocalSqliteDocumentStore
 from agent_env.task import Task
 from agent_env.explorer.routers import objects as objects_router
@@ -591,7 +591,7 @@ def test_root_serves_health_when_no_ui_is_mounted(client):
 def test_instance_content_serves_object_store_bytes(client, tmp_path):
     """Content route proxies an object-store blob and refuses a url outside the store."""
     from agent_env.config import set_object_store
-    from agent_env.store.object_store.local_object_store import LocalFilesystemObjectStore
+    from agent_env.store.object_store.local.store import LocalFilesystemObjectStore
 
     store = LocalFilesystemObjectStore(str(tmp_path / "obj"))
     set_object_store(store)
@@ -698,7 +698,7 @@ def test_content_sandboxes_active_types(client, tmp_path):
     active type and add a misleading policy to a passive one.
     """
     from agent_env.config import set_object_store
-    from agent_env.store.object_store.local_object_store import LocalFilesystemObjectStore
+    from agent_env.store.object_store.local.store import LocalFilesystemObjectStore
 
     store = LocalFilesystemObjectStore(str(tmp_path / "obj"))
     set_object_store(store)
diff --git a/tst/unit/store/local_grant_server_test.py b/tst/unit/store/local_grant_server_test.py
new file mode 100644
index 00000000..a804b803
--- /dev/null
+++ b/tst/unit/store/local_grant_server_test.py
@@ -0,0 +1,311 @@
+"""The local store's grants against its real grant server, over HTTPS on loopback: the conformance
+grant cases, and the bounds and refusals S3 would apply to the same requests."""
+
+import os
+import subprocess
+import sys
+import threading
+import time
+
+import httpx
+import pytest
+from pytest_socket import enable_socket
+
+from agent_env.config.paths import state_root
+from agent_env.store import LocalFilesystemObjectStore
+from agent_env.store.object_store.local import grant_server as server_module
+from agent_env.store.object_store.local.grant_server import default_bind_host, grant_server
+from agent_env.store.object_store.local.tls import local_ca
+from tst.store import object_conformance
+
+
+@pytest.fixture(autouse=True)
+def _disable_network():
+    """Override the suite-wide socket block (tst/unit/conftest.py) for this module: loopback is the point here."""
+    enable_socket()
+    yield
+
+
+@pytest.fixture
+def store(tmp_path, monkeypatch):
+    """A local store granting on loopback under this test's state root, trusted as a container would trust it."""
+    monkeypatch.setattr(server_module, "_servers", {})
+    monkeypatch.setenv("SSL_CERT_FILE", str(local_ca().bundle_path))
+    yield LocalFilesystemObjectStore(str(tmp_path / "objects"), grant_bind_host="127.0.0.1", grant_advertise_host="localhost")
+    for server in server_module._servers.values():
+        server.close()
+
+
+def _policy(store, prefix="ns/", max_object_bytes=16):
+    return store.issue_upload_policy(store.object_url(prefix), max_object_bytes=max_object_bytes, expires_in=600).write
+
+
+def _post(write, fields, data=b"data", *, file_first=False):
+    files = {write.file_field: ("object", data, "application/octet-stream")}
+    if file_first:  # httpx sends data fields before files, so build the body by hand
+        request = httpx.Request("POST", write.url, files=files)
+        body = request.read()
+        boundary = request.headers["content-type"].split("boundary=")[1]
+        extra = b"".join(
+            f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode() for k, v in fields.items()
+        )
+        body = body.replace(f"--{boundary}--".encode(), extra + f"--{boundary}--".encode())
+        return httpx.post(write.url, content=body, headers={"Content-Type": request.headers["content-type"]})
+    return httpx.post(write.url, data=fields, files=files)
+
+
+@pytest.mark.parametrize("case", object_conformance.GRANT_CASES, ids=lambda c: c.__name__)
+def test_grant_conformance(case, store):
+    case(store, "")
+
+
+def test_a_read_returns_the_exact_bytes_and_type(store):
+    url = store.put("a/page.html", b"<p>hi</p>", content_type="application/json")
+    response = httpx.get(store.issue_read_grant(url).url, headers={"Accept-Encoding": "gzip"})
+    assert response.status_code == 200
+    assert response.content == b"<p>hi</p>"
+    assert response.headers["content-type"] == "application/json"
+    assert response.headers["content-length"] == "9"
+    assert "content-encoding" not in response.headers
+
+
+@pytest.mark.skipif(os.name == "nt", reason="holds the key's lock with flock, which Windows lacks")
+def test_a_download_waits_for_a_write_to_its_key(store):
+    """Bytes and type are read from one file under the key's lock, so a download never mixes two writes."""
+    grant = store.issue_read_grant(store.put("a/held", b"typed", content_type="text/plain")).url
+    holder = subprocess.Popen(
+        [sys.executable, "-c", (
+            "import fcntl, sys, time; f = open(sys.argv[1], 'a'); fcntl.flock(f, fcntl.LOCK_EX); "
+            "print('locked', flush=True); time.sleep(60)"
+        ), str(store._lock_path((store.root / "a/held").resolve()))],
+        stdout=subprocess.PIPE, text=True,
+    )
+    try:
+        assert holder.stdout.readline().strip() == "locked"
+        got = []
+        download = threading.Thread(target=lambda: got.append(httpx.get(grant, timeout=10)))
+        download.start()
+        download.join(0.3)
+        assert download.is_alive()
+        holder.kill()
+        holder.wait()
+        download.join(10)
+    finally:
+        holder.kill()
+    assert (got[0].status_code, got[0].content, got[0].headers["content-type"]) == (200, b"typed", "text/plain; charset=utf-8")
+
+
+def test_a_read_of_a_missing_object_is_404(store):
+    assert httpx.get(store.issue_read_grant(store.object_url("a/none.bin")).url).status_code == 404
+
+
+def test_a_head_reports_the_size_without_reading_the_object(store, monkeypatch):
+    monkeypatch.setattr(server_module, "_chunks", lambda f: pytest.fail("a HEAD read the object"))
+    url = store.put("a/b.bin", b"12345")
+    response = httpx.head(store.issue_read_grant(url).url)
+    assert response.status_code == 200
+    assert response.headers["content-length"] == "5"
+
+
+def test_a_grant_allows_only_its_own_method(store):
+    url = store.put("a/b.bin", b"v")
+    read = store.issue_read_grant(url).url
+    assert httpx.put(read, content=b"x").status_code == 403
+    assert httpx.post(read, content=b"x").status_code == 403
+    assert httpx.delete(read).status_code == 405
+    assert store.get(url) == b"v"
+
+
+def test_a_tampered_or_expired_grant_is_403(store):
+    url = store.put("a/b.bin", b"v")
+    read = store.issue_read_grant(url).url
+    assert httpx.get(read[:-3] + "AAA").status_code == 403
+    assert httpx.get(read.rsplit("/", 1)[0] + "/not-a-token").status_code == 403
+    server = grant_server("127.0.0.1", "localhost")
+    expired = server.issue(store, "get", int(time.time()) - 1, key="a/b.bin")
+    assert httpx.get(expired).status_code == 403
+
+
+def test_a_grant_from_a_closed_server_stops_working(store):
+    url = store.put("a/b.bin", b"v")
+    read = store.issue_read_grant(url).url
+    grant_server("127.0.0.1", "localhost").close()
+    with pytest.raises(httpx.ConnectError):
+        httpx.get(read)
+    assert httpx.get(store.issue_read_grant(url).url).content == b"v"  # the next grant starts it again
+
+
+def test_a_write_must_declare_the_signed_type(store):
+    url = store.object_url("a/w.json")
+    grant = store.issue_write_grant(url, media_type="application/json", max_bytes=64)
+    assert httpx.put(grant.url, content=b"{}", headers={"Content-Type": "text/plain"}).status_code == 403
+    assert httpx.put(grant.url, content=b"{}").status_code == 403
+    assert not store.exists("a/w.json")
+
+
+def test_a_write_larger_than_the_grant_is_refused(store):
+    url = store.object_url("a/w.bin")
+    grant = store.issue_write_grant(url, media_type="application/octet-stream", max_bytes=4)
+    assert httpx.put(grant.url, content=b"12345", headers=grant.headers).status_code == 400
+
+    def chunks():  # no Content-Length: the bound is enforced as the body streams
+        yield b"123"
+        yield b"45"
+
+    assert httpx.put(grant.url, content=chunks(), headers=grant.headers).status_code == 400
+    assert not store.exists("a/w.bin")
+    assert not any((store.root / ".agentenv-tmp").iterdir())
+
+
+def test_a_write_grant_may_be_retried_and_overwrites(store):
+    url = store.object_url("a/w.bin")
+    grant = store.issue_write_grant(url, media_type="application/octet-stream", max_bytes=8)
+    assert httpx.put(grant.url, content=b"first", headers=grant.headers).status_code == 200
+    assert httpx.put(grant.url, content=b"second", headers=grant.headers).status_code == 200
+    assert store.get(url) == b"second"
+
+
+def test_an_upload_keeps_its_content_type(store):
+    write = _policy(store)
+    assert _post(write, {"key": "ns/x.txt", "Content-Type": "text/plain"}, b"hello").status_code == 204
+    assert store.read("ns/x.txt") == b"hello"
+    assert store.get_object_metadata("ns/x.txt").content_type == "text/plain"
+
+
+def test_an_upload_may_nest_and_overwrite(store):
+    write = _policy(store)
+    assert _post(write, {"key": "ns/a/b/c"}, b"one").status_code == 204
+    assert _post(write, {"key": "ns/a/b/c"}, b"two").status_code == 204
+    assert store.read("ns/a/b/c") == b"two"
+
+
+@pytest.mark.parametrize("key", ["ns/../x", "ns//x", "ns/./x", "/ns/x", "ns\\x", "ns-sibling/x", "other/x", "ns"])
+def test_an_upload_key_outside_the_prefix_is_403(store, key):
+    assert _post(_policy(store), {"key": key}).status_code == 403
+    assert store.list("") == []
+
+
+def test_an_upload_at_the_root_cannot_touch_the_stores_own_files(store):
+    write = _policy(store, prefix="")
+    assert _post(write, {"key": ".agentenv-meta/x"}).status_code == 403
+    assert _post(write, {"key": "top.bin"}).status_code == 204
+    assert store.read("top.bin") == b"data"
+
+
+def test_an_upload_larger_than_the_grant_is_refused(store):
+    write = _policy(store, max_object_bytes=4)
+    assert _post(write, {"key": "ns/big"}, b"12345").status_code == 400
+    assert not store.exists("ns/big")
+    assert not any((store.root / ".agentenv-tmp").iterdir())
+
+
+def test_an_upload_with_oversized_part_headers_is_refused(store):
+    write = _policy(store)
+    body = (
+        b"--b\r\nContent-Disposition: form-data; name=\"key\"\r\nX-Padding: " + b"x" * 20_000
+        + b"\r\n\r\nns/x\r\n--b--\r\n"
+    )
+    response = httpx.post(write.url, content=body, headers={"Content-Type": "multipart/form-data; boundary=b"})
+    assert response.status_code == 400
+    assert not store.exists("ns/x")
+
+
+def test_an_upload_must_name_its_key_before_its_file(store):
+    write = _policy(store)
+    assert _post(write, {"key": "ns/late"}, file_first=True).status_code == 400
+    assert not store.exists("ns/late")
+
+
+def test_fields_after_the_file_are_ignored(store):
+    write = _policy(store)
+    request = httpx.Request("POST", write.url, data={"key": "ns/second"}, files={"file": ("f", b"v", "text/plain")})
+    boundary = request.headers["content-type"].split("boundary=")[1]
+    body = request.read().replace(
+        f"--{boundary}--".encode(),
+        f'--{boundary}\r\nContent-Disposition: form-data; name="key"\r\n\r\nns/../escape\r\n--{boundary}--'.encode(),
+    )
+    assert httpx.post(write.url, content=body, headers={"Content-Type": request.headers["content-type"]}).status_code == 204
+    assert store.read("ns/second") == b"v"
+
+
+@pytest.mark.parametrize("body, content_type", [
+    (b"key=ns/x", "application/x-www-form-urlencoded"),
+    (b"--b\r\nContent-Disposition: form-data; name=\"key\"\r\n\r\nns/x\r\n--b--\r\n", "multipart/form-data; boundary=b"),
+])
+def test_an_upload_that_is_not_a_form_with_a_file_is_400(store, body, content_type):
+    write = _policy(store)
+    assert httpx.post(write.url, content=body, headers={"Content-Type": content_type}).status_code == 400
+
+
+def test_grants_refuse_the_stores_own_files(store):
+    with pytest.raises(ValueError, match="reserved"):
+        store.issue_read_grant(store.object_url(".agentenv-meta/a"))
+
+
+def test_a_grant_expires_when_asked(store):
+    url = store.put("a/b.bin", b"v")
+    grant = store.issue_read_grant(url, expires_in=120)
+    assert 118 <= (grant.expires_at.timestamp() - time.time()) <= 121
+    with pytest.raises(ValueError):
+        store.issue_read_grant(url, expires_in=0)
+
+
+def test_two_stores_share_one_server_and_keep_their_own_objects(store, tmp_path):
+    other = LocalFilesystemObjectStore(str(tmp_path / "other"), grant_bind_host="127.0.0.1", grant_advertise_host="localhost")
+    a, b = store.put("k", b"mine"), other.put("k", b"theirs")
+    ga, gb = store.issue_read_grant(a).url, other.issue_read_grant(b).url
+    assert ga.split("/v1/")[0] == gb.split("/v1/")[0]
+    assert (httpx.get(ga).content, httpx.get(gb).content) == (b"mine", b"theirs")
+
+
+def test_grant_urls_name_the_advertised_host(tmp_path, monkeypatch):
+    monkeypatch.setattr(server_module, "_servers", {})
+    store = LocalFilesystemObjectStore(str(tmp_path), grant_bind_host="127.0.0.1")
+    try:
+        assert store.issue_read_grant(store.put("k", b"v")).url.startswith("https://host.docker.internal:")
+    finally:
+        for server in server_module._servers.values():
+            server.close()
+
+
+def test_settings_are_checked_when_the_store_is_made(tmp_path):
+    with pytest.raises(ValueError, match="IP address"):
+        LocalFilesystemObjectStore(str(tmp_path), grant_bind_host="localhost")
+    with pytest.raises(ValueError, match="not a local name"):
+        LocalFilesystemObjectStore(str(tmp_path), grant_advertise_host="example.com")
+    with pytest.raises(ValueError, match="private address"):
+        LocalFilesystemObjectStore(str(tmp_path), grant_advertise_host="8.8.8.8")
+
+
+def test_the_root_defaults_to_the_state_root():
+    assert LocalFilesystemObjectStore().root == state_root() / "object_store"
+
+
+class TestDefaultBindHost:
+    @pytest.fixture(autouse=True)
+    def _fresh(self):
+        default_bind_host.cache_clear()
+        yield
+        default_bind_host.cache_clear()
+
+    def test_loopback_off_linux(self, monkeypatch):
+        monkeypatch.setattr(server_module.platform, "system", lambda: "Darwin")
+        assert default_bind_host() == "127.0.0.1"
+
+    def test_the_bridge_gateway_on_linux(self, monkeypatch):
+        monkeypatch.setattr(server_module.platform, "system", lambda: "Linux")
+        monkeypatch.setattr(
+            server_module.subprocess, "run",
+            lambda *a, **k: subprocess.CompletedProcess(a, 0, stdout="172.17.0.1 \n", stderr=""),
+        )
+        assert default_bind_host() == "172.17.0.1"
+
+    def test_loopback_with_a_warning_when_docker_cannot_say(self, monkeypatch, caplog):
+        monkeypatch.setattr(server_module.platform, "system", lambda: "Linux")
+
+        def fail(*a, **k):
+            raise FileNotFoundError("docker")
+
+        monkeypatch.setattr(server_module.subprocess, "run", fail)
+        assert default_bind_host() == "127.0.0.1"
+        assert "grant_bind_host" in caplog.text
diff --git a/tst/unit/store/local_object_store_test.py b/tst/unit/store/local_object_store_test.py
index 20387986..a79ce5a8 100644
--- a/tst/unit/store/local_object_store_test.py
+++ b/tst/unit/store/local_object_store_test.py
@@ -17,7 +17,7 @@
 import pytest
 
 from agent_env.store import LocalFilesystemObjectStore, ObjectAlreadyExistsError
-from agent_env.store.object_store import local_object_store
+from agent_env.store.object_store.local import store as local_store
 from tst.store import object_conformance
 
 
@@ -360,7 +360,7 @@ def test_a_write_that_cannot_lock_its_staged_file_leaves_nothing_staged(store, t
     def no_lock(fd, op):
         raise OSError(errno.ENOLCK, "no locks available")
 
-    monkeypatch.setattr(local_object_store.fcntl, "flock", no_lock)
+    monkeypatch.setattr(local_store.fcntl, "flock", no_lock)
     with pytest.raises(OSError, match="no locks"):
         store.put("nolock/x", b"v")
     assert [n for n in os.listdir(tmp_path / ".agentenv-tmp") if n.startswith("staged-")] == []
diff --git a/tst/unit/store/local_tls_test.py b/tst/unit/store/local_tls_test.py
new file mode 100644
index 00000000..9c557472
--- /dev/null
+++ b/tst/unit/store/local_tls_test.py
@@ -0,0 +1,145 @@
+"""The local transfer CA: created once per user, trusted only through its own files, and able to
+vouch for local names and private addresses only. Handshakes run over memory BIOs, not sockets."""
+
+import os
+import ssl
+import stat
+from datetime import timedelta
+from pathlib import Path
+
+import certifi
+import pytest
+from cryptography import x509
+
+from agent_env.config.paths import state_root
+from agent_env.store.object_store.local import tls
+from agent_env.store.object_store.local.tls import check_local_host, local_ca, server_context
+
+
+def _handshake(server: ssl.SSLContext, *, cafile: Path, hostname: str) -> None:
+    client = ssl.create_default_context(cafile=str(cafile))
+    client.verify_flags |= ssl.VERIFY_X509_STRICT  # the default from Python 3.13
+    to_server, to_client = ssl.MemoryBIO(), ssl.MemoryBIO()
+    sides = [
+        client.wrap_bio(to_client, to_server, server_hostname=hostname),
+        server.wrap_bio(to_server, to_client, server_side=True),
+    ]
+    done = set()
+    for _ in range(10):
+        for side in sides:
+            if id(side) not in done:
+                try:
+                    side.do_handshake()
+                    done.add(id(side))
+                except ssl.SSLWantReadError:
+                    pass
+        if len(done) == 2:
+            return
+    raise AssertionError("the handshake did not finish")
+
+
+@pytest.mark.parametrize("hostname", ["host.docker.internal", "localhost", "127.0.0.1"])
+def test_a_server_certificate_verifies_for_the_names_it_was_made_for(hostname):
+    ca = local_ca()
+    context = server_context(ca, ["localhost", "host.docker.internal", "127.0.0.1"])
+    _handshake(context, cafile=ca.cert_path, hostname=hostname)
+    _handshake(context, cafile=ca.bundle_path, hostname=hostname)
+
+
+def test_the_ca_cannot_vouch_for_a_public_name():
+    ca = local_ca()
+    with pytest.raises(ssl.SSLCertVerificationError):
+        _handshake(server_context(ca, ["example.com"]), cafile=ca.cert_path, hostname="example.com")
+
+
+def test_the_ca_cannot_vouch_for_a_public_address():
+    ca = local_ca()
+    with pytest.raises(ssl.SSLCertVerificationError):
+        _handshake(server_context(ca, ["8.8.8.8"]), cafile=ca.cert_path, hostname="8.8.8.8")
+
+
+def test_the_bundle_is_the_public_roots_plus_the_ca():
+    ca = local_ca()
+    bundle = ca.bundle_path.read_bytes()
+    assert bundle.startswith(Path(certifi.where()).read_bytes().rstrip())
+    assert bundle.endswith(ca.cert_path.read_bytes())
+    assert x509.load_pem_x509_certificate(ca.cert_path.read_bytes()) == ca.cert
+
+
+def test_the_ca_key_is_private_to_the_user_and_the_trust_files_are_not():
+    ca = local_ca()
+    tls_dir = state_root() / "tls"
+    assert stat.S_IMODE(os.stat(tls_dir).st_mode) == 0o700
+    assert stat.S_IMODE(os.stat(tls_dir / "ca.key.pem").st_mode) == 0o600
+    assert stat.S_IMODE(os.stat(ca.trust_dir).st_mode) == 0o755
+    assert sorted(p.name for p in ca.trust_dir.iterdir()) == ["ca-bundle.pem", "ca.pem"]
+    assert {stat.S_IMODE(os.stat(p).st_mode) for p in ca.trust_dir.iterdir()} == {0o644}
+
+
+def test_the_ca_is_created_once_and_reused():
+    first = local_ca()
+    tls._process_ca.cache_clear()  # as another process would
+    assert local_ca().cert == first.cert
+
+
+def test_a_process_keeps_its_ca_and_trust_files_when_another_renews_it():
+    mine = local_ca()
+    key, cert = tls._new_ca()
+    (state_root() / "tls" / "ca.key.pem").write_bytes(tls._key_pem(key) + cert.public_bytes(tls.serialization.Encoding.PEM))
+
+    assert local_ca() is mine
+    assert x509.load_pem_x509_certificate(mine.cert_path.read_bytes()) == mine.cert
+    tls._process_ca.cache_clear()
+    theirs = local_ca()
+    assert theirs.cert == cert and theirs.trust_dir != mine.trust_dir
+    assert x509.load_pem_x509_certificate(mine.cert_path.read_bytes()) == mine.cert
+
+
+def test_a_server_certificate_lasts_as_long_as_its_ca():
+    ca = local_ca()
+    server = server_context(ca, ["localhost"])
+    client = ssl.create_default_context(cafile=str(ca.cert_path))
+    to_server, to_client = ssl.MemoryBIO(), ssl.MemoryBIO()
+    tls_client = client.wrap_bio(to_client, to_server, server_hostname="localhost")
+    tls_server = server.wrap_bio(to_server, to_client, server_side=True)
+    for _ in range(10):
+        for side in (tls_client, tls_server):
+            try:
+                side.do_handshake()
+            except ssl.SSLWantReadError:
+                pass
+    leaf = x509.load_der_x509_certificate(tls_client.getpeercert(binary_form=True))
+    assert leaf.not_valid_after_utc == ca.cert.not_valid_after_utc
+
+
+def test_a_ca_near_expiry_is_replaced(monkeypatch):
+    with monkeypatch.context() as m:
+        m.setattr(tls, "_CA_LIFETIME", timedelta(days=10))
+        first = local_ca()
+    tls._process_ca.cache_clear()
+    second = local_ca()
+    assert second.cert != first.cert
+    assert x509.load_pem_x509_certificate(second.cert_path.read_bytes()) == second.cert
+    tls._process_ca.cache_clear()
+    assert local_ca().cert == second.cert
+
+
+def test_an_unreadable_ca_is_replaced():
+    first = local_ca()
+    (state_root() / "tls" / "ca.key.pem").write_text("not a key")
+    tls._process_ca.cache_clear()
+    assert local_ca().cert != first.cert
+
+
+@pytest.mark.parametrize("host", [
+    "localhost", "host.docker.internal", "a.localhost", "127.0.0.1", "172.17.0.1", "192.168.5.2", "10.0.0.8",
+    "::1", "fd00::1",
+])
+def test_local_hosts_are_accepted(host):
+    check_local_host(host)
+
+
+@pytest.mark.parametrize("host", ["example.com", "evil-localhost", "docker.internal", "8.8.8.8", "2001:db8::1"])
+def test_other_hosts_are_refused(host):
+    with pytest.raises(ValueError):
+        check_local_host(host)
diff --git a/tst/unit/store/local_tokens_test.py b/tst/unit/store/local_tokens_test.py
new file mode 100644
index 00000000..bc07880d
--- /dev/null
+++ b/tst/unit/store/local_tokens_test.py
@@ -0,0 +1,65 @@
+"""Grant tokens: what one signs, it verifies, until the expiry; nothing else verifies."""
+
+import pytest
+
+from agent_env.store.object_store.local.tokens import GrantClaims, GrantSigner, InvalidGrantError
+
+NOW = 1_800_000_000
+
+
+def _claims(**overrides) -> GrantClaims:
+    fields = {"op": "put", "store": "s1", "expires": NOW + 60, "key": "a/b.json", "max_bytes": 64,
+              "content_type": "application/json"}
+    return GrantClaims(**{**fields, **overrides})
+
+
+def test_a_signed_token_verifies_to_its_claims():
+    signer = GrantSigner()
+    claims = _claims()
+    assert signer.verify(signer.sign(claims), now=NOW) == claims
+
+
+def test_a_post_token_carries_its_prefix():
+    signer = GrantSigner()
+    claims = GrantClaims(op="post", store="s1", expires=NOW + 60, prefix="ns/", max_bytes=8)
+    assert signer.verify(signer.sign(claims), now=NOW).prefix == "ns/"
+
+
+def test_an_expired_token_is_refused():
+    signer = GrantSigner()
+    token = signer.sign(_claims(expires=NOW))
+    with pytest.raises(InvalidGrantError, match="expired"):
+        signer.verify(token, now=NOW)
+
+
+def test_another_processs_token_is_refused():
+    with pytest.raises(InvalidGrantError, match="signature"):
+        GrantSigner().verify(GrantSigner().sign(_claims()), now=NOW)
+
+
+@pytest.mark.parametrize("tamper", [
+    lambda t: t.replace("v1.", "v2.", 1),
+    lambda t: t.rsplit(".", 1)[0],
+    lambda t: t[:-2] + ("AA" if not t.endswith("AA") else "BB"),
+    lambda t: "v1." + GrantSigner().sign(_claims(key="other")).split(".")[1] + "." + t.rsplit(".", 1)[1],
+    lambda t: t + "!",
+])
+def test_a_tampered_token_is_refused(tamper):
+    signer = GrantSigner()
+    with pytest.raises(InvalidGrantError):
+        signer.verify(tamper(signer.sign(_claims())), now=NOW)
+
+
+@pytest.mark.parametrize("fields", [
+    {"op": "delete"},
+    {"op": "post", "key": None, "prefix": "ns"},
+    {"op": "post", "prefix": "ns/"},
+    {"op": "get", "key": None, "prefix": "ns/"},
+    {"max_bytes": None},
+    {"max_bytes": -1},
+    {"content_type": None},
+    {"op": "get", "max_bytes": None},
+])
+def test_claims_that_make_no_sense_are_refused(fields):
+    with pytest.raises(ValueError):
+        _claims(**fields)
diff --git a/tst/unit/store/s3_signed_post_test.py b/tst/unit/store/s3_signed_post_test.py
index bdf3c389..93140d4c 100644
--- a/tst/unit/store/s3_signed_post_test.py
+++ b/tst/unit/store/s3_signed_post_test.py
@@ -2,7 +2,7 @@
 it — the round trip is integration territory."""
 from __future__ import annotations
 
-from agent_env.store.object_store.local_object_store import LocalFilesystemObjectStore
+from agent_env.store.object_store.local.store import LocalFilesystemObjectStore
 from agent_env.store.object_store.s3_object_store import S3ObjectStore
 
 BUCKET = "artifact-bucket"
diff --git a/tst/unit/store/test_config_object_store.py b/tst/unit/store/test_config_object_store.py
index c3d2db6d..3a5a996f 100644
--- a/tst/unit/store/test_config_object_store.py
+++ b/tst/unit/store/test_config_object_store.py
@@ -83,6 +83,20 @@ def test_config_toml_selects_custom_impl(monkeypatch, tmp_path):
     assert isinstance(Config().get_object_store(), FakeObjectStore)
 
 
+def test_config_toml_sets_the_local_stores_grant_hosts(monkeypatch, tmp_path):
+    monkeypatch.delenv("AGENT_ENV_OBJECT_STORE", raising=False)
+    monkeypatch.delenv("AGENT_ENV_CONFIG", raising=False)
+    monkeypatch.chdir(tmp_path)
+    _write_config(tmp_path, (
+        '[stores.object]\nimpl = "agent_env.store.object_store:LocalFilesystemObjectStore"\n'
+        'config = { grant_bind_host = "172.18.0.1", grant_advertise_host = "172.18.0.1" }\n'
+    ))
+    store = Config().get_object_store()
+    assert isinstance(store, LocalFilesystemObjectStore)
+    assert store.root == state_root() / "object_store"
+    assert (store._grant_bind_host, store._grant_advertise_host) == ("172.18.0.1", "172.18.0.1")
+
+
 def test_env_override_beats_config_toml(monkeypatch, tmp_path):
     monkeypatch.delenv("AGENT_ENV_CONFIG", raising=False)
     monkeypatch.chdir(tmp_path)
diff --git a/tst/unit/store/test_object_transfer_grants.py b/tst/unit/store/test_object_transfer_grants.py
index 2af18057..dda7ea9d 100644
--- a/tst/unit/store/test_object_transfer_grants.py
+++ b/tst/unit/store/test_object_transfer_grants.py
@@ -16,7 +16,7 @@
 from botocore.credentials import Credentials, RefreshableCredentials
 
 from agent_env.store import GrantUnavailableError
-from agent_env.store.object_store.local_object_store import (
+from agent_env.store.object_store.local.store import (
     LocalFilesystemObjectStore,
 )
 from agent_env.store.object_store import UploadPolicy
@@ -281,16 +281,9 @@ def test_botocore_still_exposes_the_signing_credentials() -> None:
     assert S3ObjectStore(session, BUCKET)._signing_credentials() == (None, True)
 
 
-def test_local_store_cannot_issue_remote_transfer_grants(tmp_path) -> None:
+def test_local_store_does_not_offer_transfer_grants(tmp_path) -> None:
+    """It can issue them (local_grant_server_test.py), but agents on the local store don't use them."""
     store = LocalFilesystemObjectStore(str(tmp_path))
 
     assert S3ObjectStore.supports_transfer_grants
     assert not store.supports_transfer_grants
-    with pytest.raises(NotImplementedError, match="cannot issue"):
-        store.issue_read_grant(store.object_url("input"))
-    with pytest.raises(NotImplementedError, match="cannot issue"):
-        store.issue_write_grant(
-            store.object_url("output"), media_type="application/json", max_bytes=10
-        )
-    with pytest.raises(NotImplementedError, match="cannot issue"):
-        store.issue_upload_policy(store.object_url("changelog"), max_object_bytes=10, expires_in=60)
diff --git a/uv.lock b/uv.lock
index dd0a9930..0b8b6921 100644
--- a/uv.lock
+++ b/uv.lock
@@ -49,7 +49,9 @@ dependencies = [
     { name = "a2a-sdk" },
     { name = "agentenv-framework-protocol" },
     { name = "boto3" },
+    { name = "certifi" },
     { name = "click" },
+    { name = "cryptography" },
     { name = "e2b" },
     { name = "httpx" },
     { name = "litellm" },
@@ -58,7 +60,10 @@ dependencies = [
     { name = "packaging" },
     { name = "pydantic" },
     { name = "pymongo" },
+    { name = "python-multipart" },
     { name = "pyyaml" },
+    { name = "starlette" },
+    { name = "uvicorn" },
 ]
 
 [package.optional-dependencies]
@@ -98,7 +103,9 @@ requires-dist = [
     { name = "agentenv-framework", extras = ["explorer", "gcp"], marker = "extra == 'dev'" },
     { name = "agentenv-framework-protocol", editable = "packages/agentenv-protocol" },
     { name = "boto3", specifier = ">=1.35.0" },
+    { name = "certifi", specifier = ">=2024.2.2" },
     { name = "click", specifier = ">=8.0.0" },
+    { name = "cryptography", specifier = ">=42" },
     { name = "e2b", specifier = "~=2.46.4" },
     { name = "fastapi", marker = "extra == 'explorer'", specifier = ">=0.110" },
     { name = "google-api-core", marker = "extra == 'gcp'", specifier = ">=2.24" },
@@ -121,8 +128,11 @@ requires-dist = [
     { name = "pytest-socket", marker = "extra == 'dev'", specifier = ">=0.7.0" },
     { name = "pytest-timeout", marker = "extra == 'dev'", specifier = ">=2.3.0" },
     { name = "pytest-xdist", marker = "extra == 'dev'", specifier = ">=3.0.0" },
+    { name = "python-multipart", specifier = ">=0.0.13" },
     { name = "pyyaml", specifier = ">=6.0" },
     { name = "requests", marker = "extra == 'gcp'", specifier = ">=2.31" },
+    { name = "starlette", specifier = ">=0.40" },
+    { name = "uvicorn", specifier = ">=0.47" },
     { name = "uvicorn", marker = "extra == 'explorer'", specifier = ">=0.29" },
 ]
 provides-extras = ["explorer", "gcp", "dev"]