diff --git a/crates/lodedb-cloud-core/src/blob_layout.rs b/crates/lodedb-cloud-core/src/blob_layout.rs
index 1519c94b..28ebb8dd 100644
--- a/crates/lodedb-cloud-core/src/blob_layout.rs
+++ b/crates/lodedb-cloud-core/src/blob_layout.rs
@@ -1,7 +1,7 @@
 //! Content-addressed blob naming for the managed remote layout.
 //!
-//! A managed remote can store artifacts content-addressed —
-//! `blobs/sha256/aa/<sha256>` under a per-tenant prefix — instead of under
+//! A managed remote can store artifacts content-addressed
+//! (`blobs/sha256/aa/<sha256>` under a per-tenant prefix) instead of under
 //! their engine path names. Content addressing is what absorbs the
 //! fork-collision case (two branches committing different artifacts under the
 //! same engine name, e.g. `idx.gen/g7.json`, coexist as two blobs), and the
diff --git a/crates/lodedb-cloud-core/src/client_ops.rs b/crates/lodedb-cloud-core/src/client_ops.rs
index b5cd62c8..bd0bc8ce 100644
--- a/crates/lodedb-cloud-core/src/client_ops.rs
+++ b/crates/lodedb-cloud-core/src/client_ops.rs
@@ -110,7 +110,7 @@ pub fn push(
 }
 
 /// Restores `index_key`'s committed generation from `remote` into the local
-/// `dir`, then proves the engine can open the restored copy read-only —
+/// `dir`, then proves the engine can open the restored copy read-only:
 /// restore-verifies-before-accepting as one operation, so no
 /// frontend can offer a pull that skips the check.
 ///
diff --git a/crates/lodedb-cloud-core/src/verify.rs b/crates/lodedb-cloud-core/src/verify.rs
index 03032b8c..56bad7c9 100644
--- a/crates/lodedb-cloud-core/src/verify.rs
+++ b/crates/lodedb-cloud-core/src/verify.rs
@@ -87,7 +87,7 @@ pub struct OpenReport {
 /// This reuses `lodedb-core`'s `load_store` with `read_only` (which takes no writer
 /// lock and reads the exact committed manifest, never the `.wal` tail), so it
 /// proves a restored directory is loadable exactly as the embedded engine would
-/// read it — the acceptance check after a restore. Returns
+/// read it, the acceptance check after a restore. Returns
 /// the loaded document/chunk counts. `persistence_dir` is the local directory the
 /// generation was restored into.
 pub fn verify_local_generation_opens(
diff --git a/crates/lodedb-core/src/engine/mod.rs b/crates/lodedb-core/src/engine/mod.rs
index ae88aad5..a9cf718b 100644
--- a/crates/lodedb-core/src/engine/mod.rs
+++ b/crates/lodedb-core/src/engine/mod.rs
@@ -1124,9 +1124,9 @@ impl CoreEngine {
 
     /// Whether the ANN cluster index is currently resident in memory for an index
     /// (adopted from a persisted `.tvann` sidecar on a writable/eager open, or
-    /// built by a prior query). Returns `false` when it is not resident — a later
+    /// built by a prior query). Returns `false` when it is not resident (a later
     /// ANN query builds it, adopting a lazy open's deferred sidecar assignment
-    /// first — or when the index is exact. Observability; does not trigger a build.
+    /// first) or when the index is exact. Observability; does not trigger a build.
     pub fn ann_cluster_resident(&self, index_id: &str) -> Result<bool, CoreError> {
         let index = self.index(index_id)?;
         Ok(index.cluster_index.borrow().is_some())
diff --git a/docs/native_core_wal_payloads.md b/docs/native_core_wal_payloads.md
index 37048e9f..d5d8dac7 100644
--- a/docs/native_core_wal_payloads.md
+++ b/docs/native_core_wal_payloads.md
@@ -23,7 +23,7 @@ root manifest has been published, so replay after a crash is idempotent.
 
 A WAL *segment* is an immutable standalone blob in exactly the file format
 above (header plus CRC-framed records), produced by `encode_wal_segment` with
-no store open — the building block for out-of-band ingest (see
+no store open, the building block for out-of-band ingest (see
 `lodedb.local.segments`). Two deliberate differences from the on-disk
 `<key>.wal`:
 
diff --git a/src/lodedb/cloud/cli.py b/src/lodedb/cloud/cli.py
index e3593332..d928c51c 100644
--- a/src/lodedb/cloud/cli.py
+++ b/src/lodedb/cloud/cli.py
@@ -1498,7 +1498,7 @@ def org_export(
     typer.echo(_json.dumps(manifest, indent=2))
 
 
-# ------------------------------------------------- memory verbs (Phase 8d)
+# ------------------------------------------------------------ memory verbs
 # A store is one end user's LodeDB instance, so these take the store name
 # (= the user's id) directly; `store list` is the user listing and
 # `store delete` the forget-the-user verb.
diff --git a/src/lodedb/cloud/client.py b/src/lodedb/cloud/client.py
index 7b99b5e6..92a46668 100644
--- a/src/lodedb/cloud/client.py
+++ b/src/lodedb/cloud/client.py
@@ -1,4 +1,4 @@
-"""`lodedb.cloud.Client` — the developer-facing handle, bound to one tenancy.
+"""`lodedb.cloud.Client`: the developer-facing handle, bound to one tenancy.
 
     from lodedb.cloud import Client
 
@@ -9,7 +9,7 @@
 The org and environment are properties of the *credential*, not of every
 call site: `ore_sk_`/`ore_pk_` tokens are minted bound to one environment,
 and the server rejects any other. So the client asks the control plane what
-the token is (`GET /v1/tokens/self`) and binds itself accordingly — code
+the token is (`GET /v1/tokens/self`) and binds itself accordingly. Code
 that connects with a production key and code that connects with a testing
 key are the same code. Personal tokens (`ore_pat_`) span environments, so
 they resolve to the account's only org and default to the ``testing``
@@ -67,7 +67,7 @@ def resolve_tenancy(
     """The (org, environment) this credential addresses.
 
     Explicit values win. An environment-scoped token supplies its own
-    binding and *refuses* a conflicting explicit value — silently ignoring
+    binding and *refuses* a conflicting explicit value. Silently ignoring
     one would aim writes somewhere the caller didn't name. A personal token
     falls back to the account's only org, then to the org's only live
     environment or the seeded ``testing`` default; anything else raises
@@ -80,7 +80,7 @@ def resolve_tenancy(
     except CloudError as error:
         if error.status_code == 404:
             # A control plane too old to introspect tokens: don't guess what
-            # the credential is — name both escape hatches.
+            # the credential is; name both escape hatches.
             raise CloudError(
                 404,
                 "this control plane does not support token introspection — "
@@ -132,7 +132,7 @@ class Client:
     """One credential, one control plane, one bound (org, environment).
 
     Construction resolves credentials and tenancy (see the module
-    docstring) and performs at most two HTTP calls — zero when both
+    docstring) and performs at most two HTTP calls, zero when both
     ``org=`` and ``environment=`` are passed. The client owns one HTTP
     connection pool that every handle it creates shares; close it with
     :meth:`close` or a ``with`` block when done.
@@ -167,9 +167,9 @@ def store(
         warm: bool = False,
         read_your_writes: bool = True,
     ):
-        """A read/write handle over one store — one end user's LodeDB
-        instance, auto-provisioned by its first write, so this makes no
-        HTTP call by default. `warm=True` additionally asks the serving
+        """A read/write handle over one store. A store is one end user's
+        LodeDB instance, auto-provisioned by its first write, so this
+        makes no HTTP call by default. `warm=True` additionally asks the serving
         tier to hydrate it now (first query skips the cold start). `key`
         names the index key when the store holds more than one (rare).
         The handle shares this client's connection pool; closing the
@@ -202,7 +202,7 @@ def store_stats(self) -> dict:
 
     def create_store(self, store: str, key: str | None = None, **options: Any) -> dict:
         """Register a store explicitly (first writes auto-provision, so this
-        is for choosing `mode=`/`preset=`/`expose_text=` up front — or
+        is for choosing `mode=`/`preset=`/`expose_text=` up front, or
         `vector_dim=` for a bring-your-own-vectors store, which accepts
         `add_vectors`/`search_by_vector` and never embeds server-side)."""
         return self._client.create_store(self.org, self.environment, store, key, **options)
@@ -211,10 +211,12 @@ def update_store(self, store: str, key: str, **changes: Any) -> dict:
         """Flip a store's `expose_text`/`mode` flags."""
         return self._client.update_store(self.org, self.environment, store, key, **changes)
 
-    def delete_store(self, store: str) -> dict:
-        """Soft-delete a whole store — forget this end user (restorable for
-        the grace period; the entitlement slot frees immediately)."""
-        return self._client.delete_store(self.org, self.environment, store)
+    def delete_store(self, store: str, *, erase: bool = False) -> dict:
+        """Soft-delete a whole store and forget this end user (restorable
+        for the grace period; the entitlement slot frees immediately).
+        `erase=True` is data-subject erasure: no grace window, restore
+        refuses immediately, and the next lifecycle sweep hard-deletes."""
+        return self._client.delete_store(self.org, self.environment, store, erase=erase)
 
     def restore_store(self, parked_store: str) -> dict:
         return self._client.restore_store(self.org, self.environment, parked_store)
@@ -241,7 +243,7 @@ def rollback_store(self, store: str, snapshot_id: str, key: str | None = None) -
 
     def list_environments(self) -> list[dict]:
         """The bound org's environments (an environment token sees only its
-        own). The pair is fixed — production and testing, seeded at signup;
+        own). The pair is fixed: production and testing, seeded at signup;
         there is no creation surface."""
         return self._client.list_environments(self.org)
 
@@ -276,7 +278,7 @@ def restore_org(self, parked_slug: str) -> dict:
     # ------------------------------------------------------------- tokens
 
     def me(self) -> dict:
-        """The signed-in account (personal tokens only — an environment
+        """The signed-in account (personal tokens only; an environment
         token is not a person and gets a 403)."""
         return self._client.me()
 
diff --git a/src/lodedb/cloud/serving.py b/src/lodedb/cloud/serving.py
index 7659a047..fecaef20 100644
--- a/src/lodedb/cloud/serving.py
+++ b/src/lodedb/cloud/serving.py
@@ -1,4 +1,4 @@
-"""`CloudStore` — the cloud handle that duck-types a local LodeDB.
+"""`CloudStore`: the cloud handle that duck-types a local LodeDB.
 
     from lodedb.cloud import Client
 
@@ -8,7 +8,7 @@
     hits = memories.recall("how should I contact them about the invoice?")
 
 A store is one end user's own LodeDB instance (the user's id in the
-agentic-memory product), auto-provisioned by its first write — open a user
+agentic-memory product), auto-provisioned by its first write. Open a user
 that doesn't exist yet and reads answer empty until the first `add`.
 Isolation is physical: this handle cannot reach any other user's instance.
 
@@ -16,9 +16,9 @@
 over the same handle for one-off scripts and console copy-paste.
 
 The returned :class:`CloudStore` implements the read subset of the local
-`lodedb.LodeDB` handle — `search` / `search_many` / `get` / `get_texts` /
+`lodedb.LodeDB` handle: `search` / `search_many` / `get` / `get_texts` /
 `stats` / `count`, with hits shaped exactly like the local `LodeSearchHit`
-(`hit.score` / `hit.id` / `hit.metadata`, and tuple unpacking) — so RAG
+(`hit.score` / `hit.id` / `hit.metadata`, and tuple unpacking), so RAG
 adapters and MCP tool bodies written against a local handle work unmodified
 against the cloud. On top of that come the memory verbs: `add` (with TTL),
 `recall`, `context_block`, `browse`, and `delete_memories`.
@@ -47,7 +47,7 @@
 
 
 class CloudSearchHit:
-    """One scored hit — attribute access and tuple unpacking, mirroring the
+    """One scored hit. Attribute access and tuple unpacking match the
     local `LodeSearchHit`. `text` is set when the search asked for it;
     `matched` (recall only) names the sub-queries that surfaced the hit."""
 
@@ -93,6 +93,19 @@ def _hit(row: dict) -> CloudSearchHit:
     )
 
 
+def _coerced_metadata(metadata: dict[str, Any] | None) -> dict[str, str] | None:
+    """The local handle's metadata coercion (ints/floats/bools stringified,
+    other value types refused with the same error), applied before the wire.
+    The server's contract is strict str->str; code written against the local
+    ``db.add`` ergonomics must not 422 on ``{"year": 2020}``. ``None`` stays
+    ``None`` (absent metadata, not an empty map)."""
+    if metadata is None:
+        return None
+    from lodedb.local.db import _coerce_metadata
+
+    return _coerce_metadata(metadata)
+
+
 class CloudStore:
     """A read handle over one managed store, duck-typing the local LodeDB
     read surface. Create via :func:`connect`."""
@@ -124,39 +137,50 @@ def __init__(
         self._read_your_writes = bool(read_your_writes)
         self._write_visibility_timeout = float(write_visibility_timeout)
         self._last_seq = 0
-        # The most recent accepted write's id — the `wait_for` handle.
+        # The most recent accepted write's id, the `wait_for` handle.
         self.last_write_id: str | None = None
 
     # ------------------------------------------------------------- queries
 
     def _empty_if_unprovisioned(self, call, empty):
-        """Runs one read, answering `empty` when this user's store simply
-        doesn't exist yet — a store is one end user and materializes on its
-        first write, so reading a fresh user before their first memory is
-        the normal zero-setup flow (the hosted MCP tools behave the same).
-        Every other error stays loud."""
+        """Runs one read, answering `empty` when this user's store holds
+        nothing to read yet: it doesn't exist (a store is one end user and
+        materializes on its first write), or it was created ahead of that
+        first write and no snapshot has published. Both are the normal
+        zero-setup flow (the hosted MCP tools behave the same), and neither
+        can hide this session's own writes: a held `min_seq` answers 425,
+        never these 404s. Every other error stays loud."""
         try:
             return call()
         except CloudError as error:
-            if error.status_code == 404 and "no such store" in error.detail:
+            if error.status_code == 404 and (
+                "no such store" in error.detail or "nothing has been pushed" in error.detail
+            ):
                 return empty
             raise
 
-    def _searched(self, call, payload: dict[str, Any]) -> dict:
+    def _searched(self, call, payload: dict[str, Any], *, deadline: float | None = None) -> dict:
         """Runs one search call with session read-your-writes: `min_seq` is
         this handle's last acked write, and a 425 (fold not caught up yet) is
-        retried briefly instead of surfacing — the write is durable, only its
-        visibility is trailing by a fold cycle."""
+        retried briefly instead of surfacing. The write is durable; only its
+        visibility trails by a fold cycle. The pause honors the server's
+        Retry-After when it sends one (polling faster than the server asks
+        just burns the search rate limit), capped by the remaining budget.
+        `deadline` (absolute monotonic) clamps the retry window below the
+        handle's visibility budget for callers with their own, tighter one."""
         if self._read_your_writes and self._last_seq > 0:
             payload["min_seq"] = self._last_seq
-        deadline = time.monotonic() + self._write_visibility_timeout
+        stop = time.monotonic() + self._write_visibility_timeout
+        if deadline is not None:
+            stop = min(stop, deadline)
         while True:
             try:
                 return call(self.org, self.environment, payload)
             except CloudError as error:
-                if error.status_code != 425 or time.monotonic() >= deadline:
+                now = time.monotonic()
+                if error.status_code != 425 or now >= stop:
                     raise
-                time.sleep(0.25)
+                time.sleep(min(error.retry_after or 0.25, max(stop - now, 0.0)))
 
     def search(
         self,
@@ -170,8 +194,8 @@ def search(
         """Top-`k` hits, engine-scored. `include_text=True` returns each
         hit's stored text inline (requires a `read:text`-scoped key and the
         store's `expose_text` flag). After a write on this handle, the search
-        waits (briefly) for the write to become visible — session
-        read-your-writes; disable with `connect(..., read_your_writes=False)`."""
+        waits (briefly) for the write to become visible (session
+        read-your-writes); disable with `connect(..., read_your_writes=False)`."""
         payload: dict[str, Any] = {
             "store": self.store,
             "key": self.key,
@@ -195,7 +219,7 @@ def search_many(
         mode: str | None = None,
         include_text: bool = False,
     ) -> list[list[CloudSearchHit]]:
-        """Top-`k` hits per query, order-preserving — the batched search."""
+        """Top-`k` hits per query, order-preserving (the batched search)."""
         payload: dict[str, Any] = {
             "store": self.store,
             "key": self.key,
@@ -232,7 +256,7 @@ def _written(self, call, payload: dict[str, Any]) -> dict:
 
         The failure this exists for: the server accepts the write but the
         response is lost (timeout, dropped connection). A naive resend would
-        register a second segment — duplicate documents under fresh ids. The
+        register a second segment, duplicate documents under fresh ids. The
         key pins the request, so the retry (same key, byte-identical body)
         gets the original acceptance replayed instead. Only transport-level
         failures are retried; an HTTP error is a real answer.
@@ -258,7 +282,7 @@ def add(
         agent_id: str | None = None,
         run_id: str | None = None,
     ) -> str:
-        """Add (or replace) one document — the cloud `db.add`. The text is
+        """Add (or replace) one document (the cloud `db.add`). The text is
         embedded server-side and the write is ACCEPTED (durable + ordered)
         when this returns; visibility follows within seconds, and a search on
         this handle waits for it (session read-your-writes). The first write
@@ -285,7 +309,9 @@ def add_many(
     ) -> list[str]:
         """Add a batch of ``{"text", "id"?, "metadata"?}`` documents as one
         accepted write (one segment; the fold batches concurrent writes into
-        one commit). Returns the ids, in order — assigned at acceptance."""
+        one commit). Metadata values are stringified exactly like the local
+        handle's (the wire contract is strict str->str). Returns the ids, in
+        order, assigned at acceptance."""
         payload: dict[str, Any] = {
             "store": self.store,
             "key": self.key,
@@ -293,7 +319,7 @@ def add_many(
                 {
                     "text": doc["text"],
                     "id": doc.get("id"),
-                    "metadata": doc.get("metadata"),
+                    "metadata": _coerced_metadata(doc.get("metadata")),
                 }
                 for doc in documents
             ],
@@ -318,8 +344,8 @@ def add_vectors(
         agent_id: str | None = None,
         run_id: str | None = None,
     ) -> str:
-        """Add (or replace) one pre-embedded document — the cloud
-        `db.add_vectors`, for stores created with `vector_dim` (the server
+        """Add (or replace) one pre-embedded document (the cloud
+        `db.add_vectors`), for stores created with `vector_dim` (the server
         never embeds; the vector must be exactly the store's dims and is
         unit-normalized server-side, the local default). `text` is optional
         retained payload for `get()`. Returns the document id."""
@@ -341,7 +367,8 @@ def add_vectors_many(
     ) -> list[str]:
         """Add a batch of ``{"vector", "id"?, "text"?, "metadata"?}``
         documents as one accepted write. Vector-store counterpart of
-        `add_many`; returns the ids in order."""
+        `add_many` (metadata stringified the same way); returns the ids in
+        order."""
         payload: dict[str, Any] = {
             "store": self.store,
             "key": self.key,
@@ -350,7 +377,7 @@ def add_vectors_many(
                     "vector": list(doc["vector"]),
                     "text": doc.get("text"),
                     "id": doc.get("id"),
-                    "metadata": doc.get("metadata"),
+                    "metadata": _coerced_metadata(doc.get("metadata")),
                 }
                 for doc in documents
             ],
@@ -372,8 +399,8 @@ def search_by_vector(
         filter: dict[str, Any] | None = None,
         include_text: bool = False,
     ) -> list[CloudSearchHit]:
-        """Top-`k` hits for a pre-embedded query — the cloud
-        `db.search_by_vector`, for vector stores (which have no server-side
+        """Top-`k` hits for a pre-embedded query (the cloud
+        `db.search_by_vector`), for vector stores (which have no server-side
         embedder; the vector must be exactly the store's dims)."""
         payload: dict[str, Any] = {
             "store": self.store,
@@ -396,8 +423,8 @@ def search_many_by_vector(
         filter: dict[str, Any] | None = None,
         include_text: bool = False,
     ) -> list[list[CloudSearchHit]]:
-        """One engine batch of pre-embedded queries — the cloud
-        `db.search_many_by_vector`."""
+        """One engine batch of pre-embedded queries (the cloud
+        `db.search_many_by_vector`)."""
         payload: dict[str, Any] = {
             "store": self.store,
             "key": self.key,
@@ -408,7 +435,7 @@ def search_many_by_vector(
         }
         result = self._empty_if_unprovisioned(
             lambda: self._searched(self._client.search_many, payload),
-            # One empty hit list PER query (mirroring search_many): callers
+            # One empty hit list PER query, mirroring search_many. Callers
             # zip queries to results, so the unprovisioned answer must keep
             # the cardinality.
             {"results": [[] for _ in vectors]},
@@ -416,19 +443,19 @@ def search_many_by_vector(
         return [[_hit(row) for row in hits] for hits in result["results"]]
 
     def remove(self, id: str) -> str:
-        """Remove one document by id — the cloud `db.remove`, async-first:
+        """Remove one document by id (the cloud `db.remove`), async-first:
         returns the accepted write's id once the removal is durably queued.
         Whether the document existed is decided when the fold applies the
-        delete — ``wait_for(write_id)["result"]["removed"][0]`` answers it."""
+        delete. ``wait_for(write_id)["result"]["removed"][0]`` answers it."""
         return self.remove_many([id])
 
     def remove_many(self, ids: Sequence[str]) -> str:
-        """Remove a batch of documents by id as one accepted write — the
-        cloud `db.remove_many`, async-first like :meth:`remove`: returns the
+        """Remove a batch of documents by id as one accepted write (the
+        cloud `db.remove_many`), async-first like :meth:`remove`: returns the
         write's id once the removals are durably queued (one segment, one
-        fold). Per-id outcomes are decided when the fold applies the deletes
-        — ``wait_for(write_id)["result"]["removed"]`` is the parallel bool
-        list. An empty batch raises: with no accepted write there is no id to
+        fold). Per-id outcomes are decided when the fold applies the deletes.
+        ``wait_for(write_id)["result"]["removed"]`` is the parallel bool
+        list. An empty batch raises; with no accepted write there is no id to
         return (the local handle's ``remove_many([])`` no-op returns 0
         instead)."""
         document_ids = list(ids)
@@ -467,7 +494,7 @@ def wait_for(self, write_id: str, *, timeout: float = 30.0) -> dict[str, Any]:
     # ---------------------------------------------------------------- text
 
     def get(self, id: str) -> str | None:
-        """One document's stored raw text by id (None when absent) — the
+        """One document's stored raw text by id (None when absent), the
         cloud `db.get(id)`. Requires `read:text` and the store's
         `expose_text` flag."""
         result = self._empty_if_unprovisioned(
@@ -482,8 +509,52 @@ def get(self, id: str) -> str | None:
     get_text = get
 
     def get_texts(self, ids: list[str]) -> dict[str, str]:
-        """Stored text for several ids (missing ids are omitted). One request
-        per id today — prefer `search(..., include_text=True)` for RAG."""
+        """Stored text for several ids (missing ids are omitted), with the
+        text endpoint's exact per-id semantics at batch cost. By-id browse
+        pages of 100 do the bulk work (one request per hundred ids), and
+        anything a page could not answer authoritatively falls back to the
+        single-id text endpoint (the pre-batching shape): an id absent from
+        its page (genuinely gone, TTL-hidden, or an older control plane
+        that ignored the by-id fields and answered a plain page), a stray
+        document nobody asked for (definitely such a control plane; none
+        of its answers are trustworthy), or a 403 (a least-privilege
+        `read:text`-only key; browse is search-scoped)."""
+        requested = list(dict.fromkeys(str(id) for id in ids))
+        texts: dict[str, str] = {}
+        for start in range(0, len(requested), 100):
+            batch = requested[start : start + 100]
+            wanted = set(batch)
+            try:
+                page = self.browse(ids=batch, include_text=True)
+            except CloudError as error:
+                if error.status_code != 403:
+                    raise
+                # The key can't browse (no search scope). The per-id text
+                # endpoint is exactly what `read:text` grants, and if the
+                # 403 was about text access itself, the fallback's first
+                # request surfaces the same actionable refusal.
+                return self._get_texts_by_id(requested)
+            if any(doc.get("id") not in wanted for doc in page):
+                # A document nobody asked for: this control plane ignored
+                # the allowlist and answered a plain page.
+                return self._get_texts_by_id(requested)
+            for doc in page:
+                if doc.get("text") is not None:
+                    texts[doc["id"]] = doc["text"]
+        # Exactness: confirm every unanswered id through the text endpoint.
+        # On a current control plane this costs one request per id that is
+        # genuinely absent (or TTL-hidden, which the endpoint, like the old
+        # per-id path, still answers); on an older one it recovers the ids
+        # its truncated page left out.
+        for id in requested:
+            if id not in texts:
+                text = self.get(id)
+                if text is not None:
+                    texts[id] = text
+        return texts
+
+    def _get_texts_by_id(self, ids: list[str]) -> dict[str, str]:
+        """The pre-batching shape: one text-endpoint request per id."""
         texts: dict[str, str] = {}
         for id in ids:
             text = self.get(id)
@@ -495,7 +566,7 @@ def get_texts(self, ids: list[str]) -> dict[str, str]:
 
     def stats(self, *, warm: bool = False) -> dict[str, Any]:
         """Metrics-only serving stats (counts, snapshot identity, payload
-        flags) — the cloud `db.stats()` subset."""
+        flags), the cloud `db.stats()` subset."""
         return self._client.serving_stats(
             self.org, self.environment, self.store, self.key, warm=warm
         )
@@ -516,7 +587,7 @@ def recall(
         agent_id: str | None = None,
         run_id: str | None = None,
     ) -> list[CloudSearchHit]:
-        """Non-exact retrieval from RAW text — pass a whole user message;
+        """Non-exact retrieval from RAW text. Pass a whole user message;
         the server derives sub-queries and fuses the rankings. Each hit's
         `matched` attribute (set on the returned objects) names the
         sub-queries that surfaced it. `agent_id`/`run_id` narrow to one
@@ -586,17 +657,21 @@ def browse(
     ) -> list[dict[str, Any]]:
         """This store's memories (ids + metadata, text when asked and
         allowed), in one of three shapes: keyset pages in the engine's
-        stable id order (the default), most-recent-first pages
-        (`order="recent"` — same last-id cursor, but it only holds within
-        one served snapshot; a 422 asks the caller to restart when the
-        store changed under the enumeration, and a match set past the
-        server's scan cap also 422s — narrow the filter or use id order),
-        or a by-id fetch (`ids=[...]` — exactly the named documents that
-        exist, no paging; the server refuses `after`/`order` beside it).
+        stable id order (the default), ordered pages, or a by-id fetch
+        (`ids=[...]`: exactly the named documents that exist, no paging;
+        the server refuses `after`/`order` beside it). Ordered pages take
+        `order="recent"` (the server's write-instant stamp) or, on a newer
+        control plane, `order="metadata:<key>"` (descending by that
+        metadata value, keyless documents last, so an ISO-8601 stamp of
+        your own reads newest-first). Both use the same last-id cursor,
+        and it only holds within one served snapshot: a 422 asks the
+        caller to restart when the store changed under the enumeration,
+        and a match set past the server's scan cap also 422s, so narrow
+        the filter or use id order.
         Like search, the enumeration honors session read-your-writes: after
         a write on this handle it waits briefly for that write's fold.
         `ids`/`order`/the read-your-writes token need a control plane that
-        knows them — an older server ignores unknown browse fields and
+        knows them. An older server ignores unknown browse fields and
         answers a plain id-ordered page."""
         payload: dict[str, Any] = {
             "store": self.store,
@@ -614,17 +689,121 @@ def browse(
             payload["agent_id"] = agent_id
         if run_id is not None:
             payload["run_id"] = run_id
+        return self._browse_page(payload)
+
+    def _browse_page(
+        self, payload: dict[str, Any], *, deadline: float | None = None
+    ) -> list[dict[str, Any]]:
+        """One browse request through the session-floor retry, empty when
+        the store isn't provisioned. `deadline` clamps the 425 retry window:
+        the list_documents walk must not let one page's visibility wait
+        outlive the walk's own budget."""
         result = self._empty_if_unprovisioned(
-            lambda: self._searched(self._client.browse_documents, payload),
+            lambda: self._searched(self._client.browse_documents, payload, deadline=deadline),
             {"documents": []},
         )
         return result["documents"]
 
+    def list_documents(
+        self,
+        *,
+        filter: dict[str, Any] | None = None,
+        after: str | None = None,
+        limit: int | None = None,
+        max_documents: int | None = None,
+        timeout: float | None = None,
+    ) -> list[dict[str, Any]]:
+        """Document records, optionally filtered and paged (the cloud
+        `db.list_documents`). Enumeration, not ranking. It mirrors the local
+        handle: each record is ``{"id", "metadata", "chunk_count"}`` (the
+        local handle's extra ``content_hash`` is an on-disk artifact detail
+        with no serving-tier counterpart), ``filter`` takes the same grammar
+        as :meth:`search`, and ``after``/``limit`` page the stable
+        id-ordered match set with the same keyset cursor.
+
+        With ``limit=None`` this walks the WHOLE match set in 100-document
+        browse pages (one request per hundred matches). Two keyword-only
+        bounds guard that loop against a store that has outgrown the
+        caller: ``max_documents`` raises ValueError once more records than
+        that match, and ``timeout`` (seconds) raises TimeoutError when the
+        walk outlives it. The deadline is checked around every page fetch,
+        so a single slow final page still fails closed rather than
+        returning past the budget.
+        Enumeration is read-only, so both bounds leave the store untouched.
+        Like every read, each page honors the session's read-your-writes
+        floor.
+        """
+        deadline = None if timeout is None else time.monotonic() + float(timeout)
+        remaining = None if limit is None else max(int(limit), 0)
+
+        def budget_spent() -> TimeoutError:
+            """The walk's fail-closed refusal, one wording everywhere."""
+            return TimeoutError(
+                f"list_documents did not finish within {timeout}s; narrow the "
+                "filter, page with after/limit, or raise the timeout"
+            )
+
+        def check_deadline() -> None:
+            """Fails the walk once the budget is spent, wherever it is spent:
+            a caller treating `timeout` as a safety bound must never get a
+            quiet success that took longer than the bound allows."""
+            if deadline is not None and time.monotonic() >= deadline:
+                raise budget_spent()
+
+        records: list[dict[str, Any]] = []
+        cursor = after
+        while remaining is None or remaining > 0:
+            check_deadline()
+            page_size = 100 if remaining is None else min(remaining, 100)
+            payload: dict[str, Any] = {
+                "store": self.store,
+                "key": self.key,
+                "after": cursor,
+                "limit": page_size,
+                "include_text": False,
+                "filter": filter,
+            }
+            try:
+                # The page's own read-your-writes wait is clamped to this
+                # walk's deadline; a 425 that outlives the budget IS the walk
+                # not finishing in time.
+                rows = self._browse_page(payload, deadline=deadline)
+            except CloudError as error:
+                if (
+                    error.status_code == 425
+                    and deadline is not None
+                    and time.monotonic() >= deadline
+                ):
+                    raise budget_spent() from error
+                raise
+            check_deadline()
+            for row in rows:
+                records.append(
+                    {
+                        "id": str(row.get("id") or ""),
+                        "metadata": dict(row.get("metadata") or {}),
+                        "chunk_count": int(row.get("chunk_count") or 0),
+                    }
+                )
+            if max_documents is not None and len(records) > max_documents:
+                raise ValueError(
+                    f"more than {max_documents} documents match; narrow the "
+                    "filter or page with after/limit"
+                )
+            if len(rows) < page_size:
+                return records
+            if remaining is not None:
+                remaining -= len(rows)
+            cursor = str(rows[-1].get("id") or "") or None
+            if cursor is None:
+                return records
+        return records
+
     def delete_memories(
         self, *, agent_id: str | None = None, run_id: str | None = None
     ) -> dict[str, Any]:
         """Delete this store's memories in place (expired ones included),
-        narrowable to one agent/run. The store stays registered — to forget
+        narrowable to one agent/run. The store stays registered. To forget
         the user entirely (and free their entitlement slot), delete the
         store itself (`CloudClient.delete_store` / `lodedb cloud store delete`).
         Returns the acceptance (`write_ids`, `document_count`, `max_seq`);
@@ -640,8 +819,8 @@ def delete_memories(
         for write_id in result.get("write_ids", []):
             self.last_write_id = str(write_id)
         # The deletion is a write like any other: raise the session's
-        # read-your-writes floor so a search on this handle waits for it —
-        # otherwise deleted memories can resurface until the fold lands.
+        # read-your-writes floor so a search on this handle waits for it.
+        # Otherwise deleted memories can resurface until the fold lands.
         max_seq = result.get("max_seq")
         if max_seq is not None and int(max_seq) > self._last_seq:
             self._last_seq = int(max_seq)
@@ -670,19 +849,19 @@ def __repr__(self) -> str:
 class _BareStore:
     """A single-segment target: just the store id. The org/environment half
     comes from the credential (`resolve_tenancy`), the same way
-    `Client().store(...)` resolves it — so a user never retypes what their
+    `Client().store(...)` resolves it, so a user never retypes what their
     environment-scoped token already pins down."""
 
     store: str
 
 
 def _parse_target(target: str) -> ManagedRemote | _BareStore:
-    """Accepts a bare store id (`user-42` — org/environment resolve from the
-    credential), `org/environment/store`, and the explicit `orecloud://`
-    spellings of all of these (the URL form also allows `org/environment`,
-    defaulting the store). The store segment is the end-user id in the
-    agentic-memory product — a store auto-provisions on its first write, so
-    nothing needs creating first."""
+    """Accepts a bare store id like `user-42` (org/environment resolve
+    from the credential), `org/environment/store`, and the explicit
+    `orecloud://` spellings of all of these (the URL form also allows
+    `org/environment`, defaulting the store). The store segment is the
+    end-user id in the agentic-memory product. A store auto-provisions on
+    its first write, so nothing needs creating first."""
     body = target
     is_url = target.startswith(ManagedRemote.SCHEME)
     if is_url:
@@ -716,29 +895,29 @@ def connect(
 ) -> CloudStore:
     """Open a read handle over a managed store, addressed by path string.
 
-    Prefer :class:`lodedb.cloud.Client`: the org/environment half of `target`
+    Prefer :class:`lodedb.cloud.Client`. The org/environment half of `target`
     repeats what an environment-scoped token already pins down, and
     ``Client().store("user-42")`` resolves it from the credential instead.
     This stays as sugar for one-off scripts and console copy-paste. It is
-    also the seam behind ``lodedb``'s constructor front doors —
+    also the seam behind ``lodedb``'s constructor front doors:
     ``LodeDB.cloud("user-42")`` and the ``LodeDB("orecloud://…")``
     config-string dispatch both land here (lodedb releases that ship the
     `[cloud]` extra), so the two must keep accepting the same keywords.
 
-    `target` is a bare store id (`"user-42"` — the org/environment half
+    `target` is a bare store id like `"user-42"` (the org/environment half
     resolves from the credential via `resolve_tenancy`, exactly like
     `Client().store()`), `"org/environment/store"`, or an `orecloud://` URL
     of either (the URL form also allows `org/environment`, defaulting the
     store). Credentials: explicit arguments, else the `ORECLOUD_TOKEN`
     environment variable, else the credentials file `lodedb cloud login`
     wrote (the host defaults to the hosted control plane). `warm=True`
-    (default) asks the serving tier
-    to hydrate and open the store now, so the first query is warm; it also
-    verifies the target exists and the credential can read it. `key` names
-    the index key when the store holds more than one (rare — a pushed LodeDB
-    directory can carry several). `read_your_writes=True` (default) makes a
-    search after a write on this handle wait briefly for that write's fold,
-    so the session always sees its own writes.
+    (default) asks the serving tier to hydrate and open the store now, so
+    the first query is warm; it also verifies the target exists and the
+    credential can read it. `key` names the index key when the store holds
+    more than one (rare; a pushed LodeDB directory can carry several).
+    `read_your_writes=True` (default) makes a search after a write on this
+    handle wait briefly for that write's fold, so the session always sees
+    its own writes.
     """
     from lodedb.cloud.client import resolve_credentials, resolve_tenancy
 
@@ -769,7 +948,7 @@ def connect(
         except CloudError as error:
             # Two fine-to-connect 404s: a store that exists but holds
             # nothing yet (first `add()` creates its first snapshot), and a
-            # store that doesn't exist at all — a store is one end user,
+            # store that doesn't exist at all. A store is one end user,
             # and users materialize on their first write, so connecting to
             # a new user before their first memory is the normal flow. A
             # bad org/environment still fails loudly (different detail).
diff --git a/src/lodedb/cloud/transfer.py b/src/lodedb/cloud/transfer.py
index dcc1d3a6..6fc70ff0 100644
--- a/src/lodedb/cloud/transfer.py
+++ b/src/lodedb/cloud/transfer.py
@@ -2,16 +2,16 @@
 control-plane API, the sealed-box login handoff, and the managed
 (`orecloud://`) transfer verbs.
 
-Synchronous httpx by design — the CLI is a sequential tool, and the SDK's
-cloud methods (Phase 3) will wrap this client in executors where needed.
+Synchronous httpx by design. The CLI is a sequential tool, and the SDK's
+cloud methods wrap this client in executors where needed.
 `transport` is injectable so tests drive the real client against an
 in-process ASGI app without a socket.
 
 The managed transfer functions compose two layers with a deliberate seam:
 this module moves bytes over HTTP (begin/commit sessions, presigned or
-proxied blob transfers), while everything that touches the commit format —
-identities, inventories, classification, the pointer document, sidecar
-trust, the verified restore — happens in the Rust core via
+proxied blob transfers), while everything that touches the commit format
+(identities, inventories, classification, the pointer document, sidecar
+trust, the verified restore) happens in the Rust core via
 ``lodedb._turbovec.cloud.managed_*``. Python never interprets a manifest: a
 head body is parsed only as opaque JSON and re-serialised for the Rust core,
 which recomputes every identity through the engine's own canonical writer
@@ -37,18 +37,21 @@
 
 
 class CloudError(RuntimeError):
-    """A control-plane refusal, carrying the HTTP status and its detail."""
+    """A control-plane refusal, carrying the HTTP status and its detail.
+    `retry_after` is the response's Retry-After header in seconds, the pause
+    a retrying caller should honor; None when the server sent none."""
 
-    def __init__(self, status_code: int, detail: str):
+    def __init__(self, status_code: int, detail: str, *, retry_after: float | None = None):
         super().__init__(f"{detail} (HTTP {status_code})")
         self.status_code = status_code
         self.detail = detail
+        self.retry_after = retry_after
 
 
 def _store_hint(org: str, environment: str, store: object) -> str | None:
     """The `X-Ore-Store` value for a data-plane call, or None when the store
     isn't identifiable in the payload (the ingress then falls back to plain
-    balancing, which is always correct — stickiness is cache locality only).
+    balancing, which is always correct; stickiness is cache locality only).
     Percent-encoded: store names are end-user identifiers, and a non-ASCII
     (or control-character) id must never turn a valid request into a header
     encoding error. Quoting is deterministic, so the same store always maps
@@ -58,6 +61,20 @@ def _store_hint(org: str, environment: str, store: object) -> str | None:
     return quote(f"{org}/{environment}/{store}", safe="/")
 
 
+def _retry_after(response: httpx.Response) -> float | None:
+    """Seconds from the response's Retry-After header, None when absent,
+    negative, or not the plain-seconds form (the HTTP-date form is rare
+    enough on API responses not to be worth parsing here)."""
+    value = response.headers.get("retry-after")
+    if value is None:
+        return None
+    try:
+        seconds = float(value)
+    except ValueError:
+        return None
+    return seconds if seconds >= 0 else None
+
+
 def _raise_for(response: httpx.Response) -> None:
     if response.status_code < 400:
         return
@@ -65,7 +82,7 @@ def _raise_for(response: httpx.Response) -> None:
         detail = response.json().get("detail", response.text)
     except ValueError:
         detail = response.text
-    raise CloudError(response.status_code, str(detail))
+    raise CloudError(response.status_code, str(detail), retry_after=_retry_after(response))
 
 
 class CloudClient:
@@ -97,7 +114,8 @@ def _request(self, method: str, path: str, *, store_hint: str | None = None, **k
         """One API call. `store_hint` stamps `X-Ore-Store` (org/env/store) on
         data-plane requests so a store-sticky ingress can hash-route them to
         the pod holding that store warm; the server never reads it, and any
-        pod answers correctly without it — routing hint, not contract."""
+        pod answers correctly without it. The header is a routing hint, not
+        a contract."""
         if store_hint is not None:
             headers = dict(kwargs.pop("headers", None) or {})
             headers["x-ore-store"] = store_hint
@@ -114,8 +132,8 @@ def me(self) -> dict:
         return self._request("GET", "/v1/auth/me")
 
     def token_self(self) -> dict:
-        """The presented token's own identity: kind, scopes, and — for
-        environment tokens — the org/environment slugs it is bound to
+        """The presented token's own identity: kind, scopes, and (for
+        environment tokens) the org/environment slugs it is bound to
         (both None for personal tokens)."""
         return self._request("GET", "/v1/tokens/self")
 
@@ -231,15 +249,21 @@ def delete_environment(self, org: str, environment: str) -> dict:
     def restore_environment(self, org: str, parked_slug: str) -> dict:
         return self._request("POST", f"/v1/orgs/{org}/environments/{parked_slug}/restore")
 
-    def delete_store(self, org: str, environment: str, store: str) -> dict:
+    def delete_store(
+        self, org: str, environment: str, store: str, *, erase: bool = False
+    ) -> dict:
         """Soft-delete a whole store (every index key in it hides with it).
+        `erase=True` is data-subject erasure: the grace window is skipped,
+        restore refuses from that moment, and the next lifecycle sweep
+        hard-deletes the store's rows and objects.
 
-        Store names are end-user ids — free-form up to '/' — so the path
-        segment is percent-encoded: a raw `?` or `#` would truncate the URL
+        Store names are end-user ids (free-form up to '/'), so the path
+        segment is percent-encoded. A raw `?` or `#` would truncate the URL
         and address a DIFFERENT store."""
         return self._request(
             "DELETE",
-            f"/v1/orgs/{org}/environments/{environment}/stores/{quote(store, safe='')}",
+            f"/v1/orgs/{org}/environments/{environment}/stores/{quote(store, safe='')}"
+            + ("?erase=true" if erase else ""),
         )
 
     def restore_store(self, org: str, environment: str, parked_store: str) -> dict:
@@ -272,7 +296,7 @@ def list_trash(self, org: str) -> list[dict]:
 
     def export_org(self, org: str) -> dict:
         """The offboarding manifest: every live environment and store with its
-        head snapshot identity (metadata only — bytes move via pull)."""
+        head snapshot identity (metadata only; bytes move via pull)."""
         return self._request("GET", f"/v1/orgs/{org}/export")
 
     # ------------------------------------------------------------ transfer
@@ -302,7 +326,7 @@ def store_history(self, org: str, environment: str, store: str, key: str) -> lis
     def rollback_store(
         self, org: str, environment: str, store: str, snapshot_id: str, key: str | None = None
     ) -> dict:
-        """Moves the branch head back to a retained snapshot (reversible —
+        """Moves the branch head back to a retained snapshot (reversible;
         the displaced head stays in the window for the retention period)."""
         return self._request(
             "POST",
@@ -406,7 +430,7 @@ def browse_documents(self, org: str, environment: str, payload: dict) -> dict:
             store_hint=_store_hint(org, environment, payload.get("store")),
         )
 
-    # ---------------------------------------------- memory verbs (Phase 8d)
+    # --------------------------------------------------------- memory verbs
 
     def recall(self, org: str, environment: str, payload: dict) -> dict:
         """Non-exact retrieval from raw text (server-side sub-queries + RRF)."""
@@ -427,8 +451,8 @@ def context_block(self, org: str, environment: str, payload: dict) -> dict:
         )
 
     def delete_memories(self, org: str, environment: str, payload: dict) -> dict:
-        """Delete a store's memories in place (async, remove segments) —
-        the store row stays; `delete_store` forgets the user entirely."""
+        """Delete a store's memories in place (async, remove segments).
+        The store row stays; `delete_store` forgets the user entirely."""
         return self._request(
             "POST",
             f"/v1/orgs/{org}/environments/{environment}/stores/memories/delete",
@@ -552,8 +576,8 @@ def wait(self, *, timeout_seconds: float = 900.0, sleep=time.sleep) -> str:
 # Engine store kinds → the wire contract's blob kinds. `tvann` (persisted ANN
 # clusters) and `tvvf` (the rescore original-vector sidecar) are vector-derived,
 # payload-free like `tvim`. Deliberately no default: an engine kind this table
-# does not know must fail loudly at `_wire_kind` rather than ship mislabelled —
-# the Rust inventory fails closed on unknown sub-manifests for the same reason.
+# does not know must fail loudly at `_wire_kind` rather than ship mislabelled.
+# The Rust inventory fails closed on unknown sub-manifests for the same reason.
 _ENGINE_KIND_TO_WIRE = {
     "json": "state",
     "tvim": "vector",
@@ -612,7 +636,7 @@ def parse(cls, target: str) -> ManagedRemote | None:
 
     def identity(self, host: str) -> str:
         """The sidecar remote-identity string. Includes the control-plane
-        host: the same org/environment on two deployments is two remotes."""
+        host; the same org/environment on two deployments is two remotes."""
         return (
             f"{self.SCHEME}{self.org}/{self.environment}/{self.store}"
             f"#host={host.rstrip('/')}"
@@ -736,7 +760,7 @@ def _push_with_plan(
         )
     except BaseException:
         # Best-effort: release the server-side session instead of leaving it
-        # to expire on its own. The original failure is what matters — an
+        # to expire on its own. The original failure is what matters. An
         # abort that itself fails (network already gone) must not mask it.
         try:
             client.abort_push(remote.org, remote.environment, begin["session_id"])
@@ -764,8 +788,8 @@ def managed_push(
     include_text: bool = False,
     include_lexical: bool = False,
 ) -> dict:
-    """Publish the local committed generation, raced through the head CAS —
-    the managed analogue of the dumb `push` verb (last writer wins; a
+    """Publish the local committed generation, raced through the head CAS.
+    The managed analogue of the dumb `push` verb (last writer wins; a
     concurrent advance surfaces as a 409, and divergence *protection* is
     `managed_sync`'s job)."""
     head, plan = _plan(
@@ -850,8 +874,8 @@ def managed_pull(
     *,
     host: str,
 ) -> dict:
-    """Restore the branch head into `dir` and verify it opens — the managed
-    analogue of the dumb `pull` verb."""
+    """Restore the branch head into `dir` and verify it opens (the managed
+    analogue of the dumb `pull` verb)."""
     return _pull_with_body(client, dir, key, remote, host, expected_snapshot_id=None)
 
 
@@ -865,8 +889,8 @@ def managed_status(
     include_text: bool = False,
     include_lexical: bool = False,
 ) -> dict:
-    """The status report for a managed remote — same fields as the dumb
-    `status` verb, lineage included."""
+    """The status report for a managed remote (same fields as the dumb
+    `status` verb, lineage included)."""
     _head, plan = _plan(
         client, dir, key, remote, host,
         include_text=include_text, include_lexical=include_lexical,
@@ -892,8 +916,9 @@ def managed_sync(
     force_pull: bool = False,
 ) -> dict:
     """Three-pointer sync against a managed remote: classify (local, sidecar
-    base, branch head), then run at most one fast-forward — the same decision
-    table as the Rust `sync` verb, with the head CAS closing the race window.
+    base, branch head), then run at most one fast-forward. The decision
+    table matches the Rust `sync` verb's; the head CAS closes the race
+    window.
     """
     if force_push and force_pull:
         raise ValueError("force_push and force_pull are mutually exclusive")
@@ -914,7 +939,7 @@ def outcome(action: str, forced: bool, transfer: dict | None) -> dict:
 
     def _push_translating_conflict() -> dict:
         """The commit-time CAS 409 (head moved since the push began) is a
-        sync-level conflict, not a generic refusal: re-running the sync
+        sync-level conflict, not a generic refusal. Re-running the sync
         re-classifies against the new head and usually resolves it."""
         try:
             return _push_with_plan(client, dir, key, remote, host, head, plan)
@@ -934,7 +959,7 @@ def _refuse_pending_wal() -> None:
         """A pull-direction transfer must not run over a local WAL still
         holding acknowledged writes (replaying them onto the pulled lineage
         would corrupt it; dropping them silently loses acked data). Refusing
-        HERE — before a single blob downloads — mirrors the Rust verbs; the
+        HERE, before a single blob downloads, mirrors the Rust verbs; the
         materialize step re-checks authoritatively under the writer lock. The
         scan runs only on this pull branch, so push/status planning never
         pays for it."""
diff --git a/src/lodedb/local/db.py b/src/lodedb/local/db.py
index f0b08044..adcd7a52 100644
--- a/src/lodedb/local/db.py
+++ b/src/lodedb/local/db.py
@@ -303,6 +303,7 @@ def __init__(
         rescore_oversample: float | None = None,
         compression: bool = True,
         read_only: bool = False,
+        discard_at_exit: bool = False,
         durability: str | None = None,
         commit_mode: str | None = None,
         route_registry_path: str | Path | None = None,
@@ -435,11 +436,22 @@ def __init__(
         rescoring, so changing any of these means a fresh path and a reindex, not an
         in-place conversion (a vector-only path, for one, cannot be reopened as a
         text-in preset index).
+        ``discard_at_exit=True`` marks a writable handle for the interpreter-exit
+        teardown hook to :meth:`discard` rather than :meth:`close`: un-persisted
+        state is dropped, not committed, when the process exits without an
+        explicit close. For a handle that serves a copy whose only publisher is
+        elsewhere (a caching tier applying provisional writes on top of a pushed
+        snapshot), an implicit exit must never become an implicit commit. An
+        explicit :meth:`close` still persists.
         ``_embedding_backend`` is an internal hook for tests/fixtures.
         """
 
         self.path = Path(path)
         self.store_text = bool(store_text)
+        # Exit disposition for the pre-worker-join teardown hook (see
+        # _close_open_native_dbs_at_exit); explicit close()/discard() calls
+        # are unaffected.
+        self._discard_at_exit = bool(discard_at_exit)
         # index_text defaults to match store_text: retaining text also persists its
         # lexical index (so hybrid/lexical are ready out of the box), and opting out
         # of text retention also opts out of persisting tokens (store_text=False keeps
@@ -2423,13 +2435,19 @@ def _close_open_native_dbs_at_exit() -> None:
     """Closes every still-open native engine on its home worker before teardown.
 
     Runs before ``concurrent.futures`` joins its worker threads, so each engine's
-    final decref lands on the worker that created it. ``close()`` is idempotent and
-    detaches the GC finalizer, so a handle the caller already closed is a no-op.
+    final decref lands on the worker that created it. A handle opened with
+    ``discard_at_exit=True`` is discarded instead of closed: the hook exists for
+    teardown hygiene, and it must not turn a process exit into an implicit commit
+    for a handle whose owner never persists. Both paths are idempotent and detach
+    the GC finalizer, so a handle the caller already closed is a no-op.
     """
 
     for db in list(_OPEN_NATIVE_DBS):
         try:
-            db.close()
+            if getattr(db, "_discard_at_exit", False):
+                db.discard()
+            else:
+                db.close()
         except Exception:  # noqa: BLE001 - teardown is best effort; never raise at exit
             pass
 
diff --git a/tests/test_cloud_routing_hint.py b/tests/test_cloud_routing_hint.py
index e6a5292c..758d2640 100644
--- a/tests/test_cloud_routing_hint.py
+++ b/tests/test_cloud_routing_hint.py
@@ -1,7 +1,7 @@
 """The `X-Ore-Store` routing hint: data-plane calls stamp org/env/store so a
 store-sticky ingress can hash-route them for cache locality. Pure transport
-concern — the server never reads the header, and calls without it stay
-correct — so the tests assert wire shape only."""
+concern (the server never reads the header, and calls without it stay
+correct), so the tests assert wire shape only."""
 
 from __future__ import annotations
 
@@ -47,7 +47,7 @@ def test_data_plane_verbs_carry_the_store_hint(capture):
 
 def test_warm_stats_carries_the_hint(capture):
     """The pre-hydration call must land on the same pod the hinted queries
-    will — a warm on one pod followed by queries on another would defeat it."""
+    will; a warm on one pod followed by queries on another would defeat it."""
     client, seen = capture
     client.serving_stats("acme", "prod", "user-42", warm=True)
     assert seen[0].headers["x-ore-store"] == "acme/prod/user-42"
diff --git a/tests/test_cloud_store_verbs.py b/tests/test_cloud_store_verbs.py
index 5cd053fa..8238c71c 100644
--- a/tests/test_cloud_store_verbs.py
+++ b/tests/test_cloud_store_verbs.py
@@ -1,6 +1,6 @@
 """CloudStore verb wiring against a stub transport: what payload each verb
 puts on the wire and how it folds the acceptance back into session state
-(read-your-writes floor, `last_write_id`). No server involved — the accepted
+(read-your-writes floor, `last_write_id`). No server involved. The accepted
 write contract itself is covered end-to-end in `server/tests`."""
 
 from __future__ import annotations
@@ -80,7 +80,7 @@ def browse_documents(self, org: str, environment: str, payload: dict) -> dict:
 
 def test_unprovisioned_batch_verbs_keep_query_cardinality():
     """Both batched search verbs answer an unprovisioned store with one empty
-    hit list PER query — callers zip queries to results."""
+    hit list PER query. Callers zip queries to results."""
     store = CloudStore(_UnprovisionedClient(), "acme", "prod", "user-42", owns_client=False)
     assert store.search_many(["a", "b", "c"]) == [[], [], []]
     assert store.search_many_by_vector([[0.1, 0.2], [0.3, 0.4]]) == [[], []]
@@ -93,13 +93,61 @@ def test_unprovisioned_browse_answers_empty():
     assert store.browse() == []
 
 
+class _NeverPushedClient:
+    """Answers reads with the 404 a created-but-never-written store gives
+    (the store row exists; no snapshot has published)."""
+
+    def browse_documents(self, org: str, environment: str, payload: dict) -> dict:
+        from lodedb.cloud.transfer import CloudError
+
+        raise CloudError(404, "nothing has been pushed to this store yet — push a snapshot first")
+
+
+def test_created_but_never_written_store_reads_empty():
+    """A store created ahead of its first write must read as empty, exactly
+    like one that does not exist yet; only genuinely foreign 404s stay loud."""
+    store = CloudStore(_NeverPushedClient(), "acme", "prod", "user-42", owns_client=False)
+    assert store.browse() == []
+    assert store.list_documents() == []
+
+
+class _AlwaysTooEarlyClient:
+    """Every browse answers 425, like a store whose fold never catches up."""
+
+    def add_documents(self, org: str, environment: str, payload: dict) -> dict:
+        return {"ids": ["m1"], "write_id": "w-9", "seq": 41}
+
+    def browse_documents(self, org: str, environment: str, payload: dict) -> dict:
+        from lodedb.cloud.transfer import CloudError
+
+        raise CloudError(425, "not folded through seq 41 yet", retry_after=0.01)
+
+
+def test_list_documents_budget_bounds_the_visibility_wait():
+    """A page's 425 retry stops at the walk's own deadline, not the handle's
+    30s visibility budget, and surfaces as the walk's TimeoutError: the
+    caller set one bound and must get one failure mode for breaching it."""
+    import time
+
+    store = _store(_AlwaysTooEarlyClient())
+    store.add("first memory")  # arms the session floor
+
+    started = time.monotonic()
+    with pytest.raises(TimeoutError, match="did not finish"):
+        store.list_documents(timeout=0.05)
+    assert time.monotonic() - started < 5.0
+
+
 class _BrowseClient:
     """Duck-types the add + browse calls, recording browse payloads and
-    optionally answering one 425 (fold not caught up) before succeeding."""
+    optionally answering one 425 (fold not caught up) before succeeding.
+    When `retry_after` is given the refusal carries it, like a server that
+    sends the Retry-After header."""
 
-    def __init__(self, too_early_first: bool = False) -> None:
+    def __init__(self, too_early_first: bool = False, retry_after: float | None = None) -> None:
         self.calls: list[dict] = []
         self._too_early = too_early_first
+        self._retry_after = retry_after
 
     def add_documents(self, org: str, environment: str, payload: dict) -> dict:
         return {"ids": ["m1"], "write_id": "w-9", "seq": 41}
@@ -110,14 +158,14 @@ def browse_documents(self, org: str, environment: str, payload: dict) -> dict:
             self._too_early = False
             from lodedb.cloud.transfer import CloudError
 
-            raise CloudError(425, "not folded through seq 41 yet")
+            raise CloudError(425, "not folded through seq 41 yet", retry_after=self._retry_after)
         return {"documents": [{"id": "m1", "metadata": {}, "chunk_count": 1}]}
 
 
 def test_browse_carries_the_session_floor_and_retries_425():
     """Browse is a read like search: after a write on this handle it sends
     the session's read-your-writes floor as min_seq and briefly retries a
-    425 instead of surfacing it — the write is durable, only its visibility
+    425 instead of surfacing it. The write is durable; only its visibility
     trails by a fold cycle."""
     client = _BrowseClient(too_early_first=True)
     store = _store(client)
@@ -129,6 +177,229 @@ def test_browse_carries_the_session_floor_and_retries_425():
     assert [call.get("min_seq") for call in client.calls] == [41, 41]
 
 
+def test_read_retry_paces_itself_by_the_server_retry_after(monkeypatch):
+    """A 425 carrying Retry-After paces the retry at the server's ask
+    (polling faster just burns the search rate limit); without one the
+    pause stays the 250ms default, and the server's ask never sleeps past
+    the handle's own visibility budget."""
+    pauses: list[float] = []
+    monkeypatch.setattr("lodedb.cloud.serving.time.sleep", lambda s: pauses.append(s))
+
+    store = _store(_BrowseClient(too_early_first=True, retry_after=1.0))
+    store.add("first memory")
+    store.browse()
+    assert pauses == [1.0]
+
+    pauses.clear()
+    store = _store(_BrowseClient(too_early_first=True))
+    store.add("first memory")
+    store.browse()
+    assert pauses == [0.25]
+
+    pauses.clear()
+    client = _BrowseClient(too_early_first=True, retry_after=60.0)
+    store = CloudStore(
+        client, "acme", "prod", "user-42", owns_client=False, write_visibility_timeout=0.5
+    )
+    store.add("first memory")
+    store.browse()
+    assert pauses and pauses[0] <= 0.5
+
+
+def test_cloud_error_parses_the_retry_after_header():
+    """The transport keeps the server's Retry-After on the refusal it
+    raises; absent or malformed headers become None, never a crash."""
+    import httpx
+
+    from lodedb.cloud.transfer import CloudError, _raise_for
+
+    def refusal(headers: dict[str, str]) -> CloudError:
+        response = httpx.Response(425, json={"detail": "not folded yet"}, headers=headers)
+        with pytest.raises(CloudError) as caught:
+            _raise_for(response)
+        return caught.value
+
+    assert refusal({"Retry-After": "1"}).retry_after == 1.0
+    assert refusal({"Retry-After": "2.5"}).retry_after == 2.5
+    assert refusal({}).retry_after is None
+    assert refusal({"Retry-After": "Wed, 21 Oct 2026 07:28:00 GMT"}).retry_after is None
+    assert refusal({"Retry-After": "-3"}).retry_after is None
+
+
+class _AddClient:
+    """Duck-types add_documents, recording each accepted payload."""
+
+    def __init__(self) -> None:
+        self.payloads: list[dict] = []
+
+    def add_documents(self, org: str, environment: str, payload: dict) -> dict:
+        self.payloads.append(payload)
+        return {
+            "ids": [doc.get("id") or "assigned" for doc in payload["documents"]],
+            "write_id": "w-1",
+            "seq": 1,
+        }
+
+
+def test_add_coerces_metadata_like_the_local_handle():
+    """Code written against the local `db.add` ergonomics may pass int/
+    float/bool metadata values; the wire contract is strict str->str, so the
+    handle stringifies exactly like the local `_coerce_metadata`, and
+    refuses the value types the local handle refuses. Absent metadata stays
+    None on the wire, not an empty map."""
+    client = _AddClient()
+    store = _store(client)
+
+    store.add("hello", metadata={"year": 2020, "vip": True, "score": 1.5, "note": "plain"})
+    meta = client.payloads[0]["documents"][0]["metadata"]
+    assert meta == {"year": "2020", "vip": "true", "score": "1.5", "note": "plain"}
+
+    with pytest.raises(ValueError, match="metadata value"):
+        store.add("hello", metadata={"bad": ["a", "list"]})
+
+    store.add("hello")
+    assert client.payloads[-1]["documents"][0]["metadata"] is None
+
+
+class _TextClient:
+    """Duck-types browse_documents + store_text for get_texts: the by-id
+    browse answers every asked-for id except 'missing'; the text endpoint
+    records which ids needed a per-id confirmation."""
+
+    def __init__(self) -> None:
+        self.calls: list[dict] = []
+        self.text_calls: list[str] = []
+
+    def browse_documents(self, org: str, environment: str, payload: dict) -> dict:
+        self.calls.append(payload)
+        return {
+            "documents": [
+                {"id": id, "metadata": {}, "chunk_count": 1, "text": f"text-{id}"}
+                for id in payload["ids"]
+                if id != "missing"
+            ]
+        }
+
+    def store_text(self, org: str, environment: str, store: str, id: str, key=None) -> dict:
+        self.text_calls.append(id)
+        return {"id": id, "found": False, "text": None}
+
+
+def test_get_texts_batches_over_the_by_id_browse():
+    """get_texts does its bulk work as one by-id browse per hundred ids (not
+    one request per id); an id absent from its page is confirmed through the
+    text endpoint so the answer keeps exact per-id semantics, and an empty
+    request makes no request at all."""
+    client = _TextClient()
+    store = _store(client)
+
+    ids = [f"doc-{i}" for i in range(150)] + ["missing"]
+    texts = store.get_texts(ids)
+
+    assert [len(call["ids"]) for call in client.calls] == [100, 51]
+    assert all(call["include_text"] for call in client.calls)
+    assert texts["doc-0"] == "text-doc-0" and texts["doc-149"] == "text-doc-149"
+    assert "missing" not in texts
+    assert client.text_calls == ["missing"]  # only the absent id re-confirms
+    assert store.get_texts([]) == {}
+    assert len(client.calls) == 2  # the empty request made no HTTP call
+
+
+def test_get_texts_distrusts_a_plain_page_answer():
+    """A document nobody asked for means the control plane ignored the by-id
+    fields (an older server answering a plain page). Every requested id then
+    goes through the text endpoint, so the result stays exact instead of
+    silently partial."""
+
+    class _PlainPageClient:
+        def __init__(self) -> None:
+            self.text_calls: list[str] = []
+
+        def browse_documents(self, org: str, environment: str, payload: dict) -> dict:
+            return {
+                "documents": [
+                    {"id": "stray", "metadata": {}, "chunk_count": 1, "text": "stray-text"},
+                    {
+                        "id": payload["ids"][0],
+                        "metadata": {},
+                        "chunk_count": 1,
+                        "text": f"text-{payload['ids'][0]}",
+                    },
+                ]
+            }
+
+        def store_text(self, org: str, environment: str, store: str, id: str, key=None) -> dict:
+            self.text_calls.append(id)
+            return {"id": id, "found": True, "text": f"endpoint-{id}"}
+
+    client = _PlainPageClient()
+    store = _store(client)
+
+    texts = store.get_texts(["a", "b"])
+
+    assert texts == {"a": "endpoint-a", "b": "endpoint-b"}
+    assert client.text_calls == ["a", "b"]
+
+
+class _TextOnlyClient:
+    """A least-privilege `read:text` credential's view: browse (search-
+    scoped) refuses with 403, the per-id text endpoint answers."""
+
+    def __init__(self) -> None:
+        self.browse_calls = 0
+        self.text_calls: list[str] = []
+
+    def browse_documents(self, org: str, environment: str, payload: dict) -> dict:
+        self.browse_calls += 1
+        from lodedb.cloud.transfer import CloudError
+
+        raise CloudError(403, "returning stored text requires the 'read:search' scope")
+
+    def store_text(self, org: str, environment: str, store: str, id: str, key=None) -> dict:
+        self.text_calls.append(id)
+        found = id != "missing"
+        return {"id": id, "found": found, "text": f"text-{id}" if found else None}
+
+
+def test_get_texts_falls_back_to_the_text_endpoint_for_text_only_keys():
+    """A key holding only read:text cannot browse; get_texts then answers
+    through the single-id text endpoint (the pre-batching shape) instead of
+    failing a previously valid least-privilege call."""
+    client = _TextOnlyClient()
+    store = _store(client)
+
+    texts = store.get_texts(["a", "missing", "b"])
+
+    assert texts == {"a": "text-a", "b": "text-b"}
+    assert client.browse_calls == 1
+    assert client.text_calls == ["a", "missing", "b"]
+
+
+def test_delete_store_erase_rides_the_query_string():
+    """erase=True must reach the wire as ?erase=true. A silently dropped
+    flag would downgrade a data-subject erasure to a grace-window delete."""
+    import httpx
+
+    from lodedb.cloud.transfer import CloudClient
+
+    seen: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(str(request.url))
+        stamp = "2026-07-20T00:00:00Z"
+        return httpx.Response(
+            200, json={"slug": "user-42-del-1", "deleted_at": stamp, "purge_after": stamp}
+        )
+
+    with CloudClient(
+        "https://cloud.test", "tok", transport=httpx.MockTransport(handler)
+    ) as client:
+        client.delete_store("acme", "prod", "user-42")
+        client.delete_store("acme", "prod", "user-42", erase=True)
+    assert seen[0].endswith("/stores/user-42")
+    assert seen[1].endswith("/stores/user-42?erase=true")
+
+
 def test_browse_passes_ids_and_order_on_the_wire():
     """The by-id fetch and the recency order ride the wire only when asked
     for, so an untouched browse keeps its old payload shape."""
@@ -148,3 +419,109 @@ def test_browse_passes_ids_and_order_on_the_wire():
     )
     assert "ids" not in client.calls[1]
     assert "ids" not in client.calls[2] and "order" not in client.calls[2]
+
+
+class _PagingBrowseClient:
+    """Answers filtered browse pages from a fixed id-ordered corpus, honoring
+    after/limit and recording each payload. Enough server for the
+    list_documents enumeration loop."""
+
+    def __init__(self, count: int) -> None:
+        self.calls: list[dict] = []
+        self._ids = [f"doc-{index:04d}" for index in range(count)]
+
+    def browse_documents(self, org: str, environment: str, payload: dict) -> dict:
+        self.calls.append(payload)
+        ids = self._ids
+        if payload.get("after"):
+            ids = [id for id in ids if id > payload["after"]]
+        page = ids[: payload["limit"]]
+        return {
+            "documents": [
+                {"id": id, "metadata": {"namespace": "default"}, "chunk_count": 1}
+                for id in page
+            ],
+            "count": len(self._ids),
+            "snapshot_id": "snap-1",
+            "generation": 1,
+        }
+
+
+def test_list_documents_walks_every_page_in_local_record_shape():
+    """limit=None enumerates the whole match set in 100-document pages and
+    answers local-handle-shaped records."""
+    client = _PagingBrowseClient(250)
+    store = CloudStore(client, "acme", "prod", "user-42", owns_client=False)
+
+    records = store.list_documents()
+
+    assert len(records) == 250
+    assert records[0] == {
+        "id": "doc-0000",
+        "metadata": {"namespace": "default"},
+        "chunk_count": 1,
+    }
+    assert [call["limit"] for call in client.calls] == [100, 100, 100]
+    assert client.calls[1]["after"] == "doc-0099"
+    assert client.calls[2]["after"] == "doc-0199"
+
+
+def test_list_documents_forwards_filter_and_pages_like_the_local_cursor():
+    client = _PagingBrowseClient(5)
+    store = CloudStore(client, "acme", "prod", "user-42", owns_client=False)
+
+    records = store.list_documents(
+        filter={"namespace": "default"}, after="doc-0001", limit=2
+    )
+
+    assert [record["id"] for record in records] == ["doc-0002", "doc-0003"]
+    assert client.calls[0]["filter"] == {"namespace": "default"}
+    assert client.calls[0]["after"] == "doc-0001"
+    assert client.calls[0]["limit"] == 2
+
+
+def test_list_documents_bounds_the_walk():
+    """The keyword-only bounds fail closed before another page is fetched: a
+    match set past max_documents raises ValueError, an outlived timeout raises
+    TimeoutError. Enumeration never mutates, so both are safe to retry with
+    a narrower filter or explicit paging."""
+    client = _PagingBrowseClient(250)
+    store = CloudStore(client, "acme", "prod", "user-42", owns_client=False)
+
+    with pytest.raises(ValueError, match="more than 100 documents"):
+        store.list_documents(max_documents=100)
+    with pytest.raises(TimeoutError, match="did not finish"):
+        store.list_documents(timeout=0.0)
+
+
+class _SlowFinalPageClient:
+    """One short (final) browse page whose fetch alone spends the caller's
+    whole time budget, ticking the injected clock as a slow server would."""
+
+    def __init__(self, clock: dict) -> None:
+        self._clock = clock
+
+    def browse_documents(self, org: str, environment: str, payload: dict) -> dict:
+        self._clock["now"] += 5.0
+        return {"documents": [{"id": "d1", "metadata": {}, "chunk_count": 1}]}
+
+
+def test_list_documents_fails_closed_when_the_final_page_outlives_the_budget(monkeypatch):
+    """The timeout is enforced around each fetch, not just before it: a
+    single slow final page must raise, never turn a spent budget into a
+    quiet success; callers treat the bound as a refusal mechanism."""
+    clock = {"now": 0.0}
+    monkeypatch.setattr("lodedb.cloud.serving.time.monotonic", lambda: clock["now"])
+    store = CloudStore(
+        _SlowFinalPageClient(clock), "acme", "prod", "user-42", owns_client=False
+    )
+
+    with pytest.raises(TimeoutError, match="did not finish"):
+        store.list_documents(timeout=1.0)
+
+
+def test_unprovisioned_list_documents_answers_empty():
+    """Enumerating a user who hasn't written yet is the normal zero-setup
+    flow, not an error (same rule as browse)."""
+    store = CloudStore(_UnprovisionedClient(), "acme", "prod", "user-42", owns_client=False)
+    assert store.list_documents() == []
diff --git a/tests/test_local_discard.py b/tests/test_local_discard.py
index 8d1a7843..3229ddcc 100644
--- a/tests/test_local_discard.py
+++ b/tests/test_local_discard.py
@@ -67,3 +67,51 @@ def test_discard_read_only_handle(tmp_path):
     _open(tmp_path, commit_mode="generation").close()
     reader = LodeDB.open_readonly(tmp_path, model="minilm", _embedding_backend=_be())
     reader.discard()
+
+
+def _folded_unpersisted(db: LodeDB, text: str, id: str) -> None:
+    """Applies one fold segment to `db` WITHOUT persisting: exactly the
+    in-memory state a graceful close() would publish and discard() drops."""
+    from lodedb.local.segments import (
+        build_embedded_documents_record,
+        encode_segment,
+        fold_segment,
+        plan_documents,
+    )
+
+    plan = plan_documents([{"text": text, "id": id}])
+    chunk_texts = tuple(str(chunk["text"]) for chunk in plan["chunks_to_embed"])
+    record = build_embedded_documents_record(
+        plan, _be().embed_documents(chunk_texts), vector_dim=DIM
+    )
+    fold_segment(db, encode_segment([record]), first_lsn=db.applied_lsn() + 1)
+
+
+def test_exit_hook_honors_discard_at_exit(tmp_path):
+    """The interpreter-exit teardown hook must not turn a process exit into
+    an implicit commit: a handle opened with ``discard_at_exit=True`` and
+    holding applied-but-unpersisted fold state is discarded (the store stays
+    at its committed state), while a default handle still closes gracefully
+    and publishes."""
+    from lodedb.local.db import _close_open_native_dbs_at_exit
+
+    _open(tmp_path / "serve", commit_mode="generation").close()
+    _open(tmp_path / "keep", commit_mode="generation").close()
+    serve = _open(tmp_path / "serve", commit_mode="generation", discard_at_exit=True)
+    keep = _open(tmp_path / "keep", commit_mode="generation")
+    _folded_unpersisted(serve, "tail-applied provisional write", "tail")
+    _folded_unpersisted(keep, "fold in flight at exit", "kept")
+    assert serve.count() == 1 and keep.count() == 1
+
+    _close_open_native_dbs_at_exit()
+
+    serve_reopened = _open(tmp_path / "serve", commit_mode="generation")
+    keep_reopened = _open(tmp_path / "keep", commit_mode="generation")
+    try:
+        assert serve_reopened.count() == 0
+        assert serve_reopened.get("tail") is None
+        assert keep_reopened.count() == 1
+        assert [rec["id"] for rec in keep_reopened.list_documents()] == ["kept"]
+    finally:
+        serve_reopened.close()
+        keep_reopened.close()