diff --git "a/platform/core/store.py" "b/platform/core/store.py"
--- "a/platform/core/store.py"
+++ "b/platform/core/store.py"
@@ -1,1208 +1,1208 @@
-"""Tiny JSON persistence on a PRIVATE Hugging Face Dataset repo — the platform's ONLY writable store
-(Odoo stays strictly read-only). Holds the user registry, customer notes and per-user tags.
-
-Auth uses HF_TOKEN (a Space secret / .env value with write access to the royal-imports org). The
-repo is created private on first write. Reads use a short in-process cache that writes refresh, so
-within a running instance reads are consistent with writes; a fresh instance re-fetches the latest.
-
-WAVE 18 (C1-TENANT, R2): the store is now a CLASS, one instance per repo. The module-level
-functions below delegate to the default instance (`OS_DATA_REPO`, tenant #0) — every existing
-`import core.store as store` caller is byte-for-byte unaffected, which the gate battery proves.
-`for_repo(repo_id)` hands back the bound instance for a tenant that owns its OWN dataset repo
-(nurilab, gtmlab, …); `harness.runtime.TenantRuntime` binds it from the tenant record. Each
-instance carries its own cache / dirty-marks / flush workers / commit-gap clock, so one tenant's
-write burst can never starve another's flush window.
-"""
-import io
-import os
-import json
-import threading
-import time as _time
-from pathlib import Path
-
-try:
-    from dotenv import load_dotenv
-    load_dotenv(Path(__file__).resolve().parents[1] / '.env')
-except Exception:
-    pass
-
-from huggingface_hub import HfApi, hf_hub_download
-from huggingface_hub.utils import (EntryNotFoundError, HfHubHTTPError,
-                                   RepositoryNotFoundError)
-
-import core.data_binding as data_binding
-
-#: ⛔⛔ D-315 — THE BARE DEFAULT IS GONE. This line used to read
-#: `os.environ.get('OS_DATA_REPO', 'royal-imports/cfo-os-data')`, i.e. every Space that was never
-#: handed the key came up bound to tenant #0's REAL business data — which is how two different
-#: builds wrote it on 2026-08-14. The value it resolves to is unchanged (see
-#: `data_binding.default_store` for why returning the real id beats `None`); what changed is that
-#: WRITING it is now refused unless this deployment is the one that owns it.
-REPO = os.environ.get('OS_DATA_REPO') or data_binding.default_store()
-_FLUSH_DELAY = 2.0      # coalesce a burst of edits into one commit
-_FLUSH_MIN_GAP = 20.0   # floor between commits of one key (the 256-commits/hr budget)
-
-#: ⭐⭐ D-305 — HOW MANY TIMES A WRITE MAY REBASE ONTO A MOVED HEAD BEFORE GIVING UP.
-#:
-#: Bounded rather than unbounded, and the cost is one extra DOWNLOAD per rebase. ⚠ Not an extra
-#: commit: a refused write answers 412 and creates nothing, so only the attempt that finally lands
-#: spends the 256-commits/hr budget. Four is generous for the measured case
-#: (two humans editing one grid) and small enough that a genuinely hot key fails loudly instead of
-#: spinning. ⛔ Hitting the bound RAISES — it never falls through to an unconditional upload, which
-#: is the pre-fix behaviour wearing a retry loop ([[fallback-that-became-the-rule]]).
-_MAX_REBASE = 4
-
-#: How many unconfirmed intents one key may hold before the oldest are dropped. See
-#: `_journal_append` for why a cap exists at all and what is reported when it bites.
-_MAX_JOURNAL = 500
-
-#: WAVE 29 (item 20 / ruling R11, contract C6) — THE CHANGE TOKEN'S PROCESS STAMP.
-#:
-#: `Store`'s revision counter is IN-MEMORY, so a restarted Space would begin again at 0 and a
-#: browser holding `…:4` from the previous process could read `…:4` from the new one and conclude
-#: nothing had changed. Prefixing every token with a per-process stamp makes a restart read as a
-#: change, which is the truthful answer: the client's copy predates this process entirely.
-#:
-#: ⚠ NOT a clock and never compared for order — a token is opaque and compared ONLY for equality.
-REV_EPOCH = f'{int(_time.time()):x}{os.getpid():x}'
-
-
-def _token():
-    return os.environ.get('HF_TOKEN') or None
-
-
-class StoreConflict(RuntimeError):
-    """A write that could not be landed without overwriting somebody else's (D-305).
-
-    Raised only after `_MAX_REBASE` rebases, or when no parent commit can be resolved at all. Both
-    are "we could not make this safe", and both must reach the caller as a failure: the alternative
-    — uploading anyway — is exactly the lost update this whole mechanism exists to prevent, and it
-    would answer 200 while destroying an hour of another person's work.
-    """
-
-
-def _is_conflict(exc):
-    """True for the ONE hub failure that means "the branch moved under you".
-
-    ⛔ NARROW ON PURPOSE, AND THE PROBE IS WHY. Measured against a scratch dataset repo on
-    2026-08-18: a stale `parent_commit` returns **HTTP 412** with the server message *"The branch
-    was updated since you opened this page. Please refresh and try again."*, and the write does NOT
-    land. A NONSENSE `parent_commit` returns a **404 `RevisionNotFoundError`** — a different fault
-    that must not be retried, because rebasing on it would loop against a bad sha. A broad
-    `except HfHubHTTPError` would swallow both, plus every 401/429/500, and turn a network blip
-    into a silent rebase.
-    """
-    return getattr(getattr(exc, 'response', None), 'status_code', None) == 412
-
-
-def _sha_from_path(path):
-    """The commit sha out of a hub cache path, or `None` if it does not look like one.
-
-    ⚠ VALIDATED, NOT ASSUMED. The layout is `…/snapshots/<40-hex>/<file>`, but a caller may have
-    an unusual `HF_HOME`, symlinks disabled (this repo's own machines warn about exactly that), or
-    a future hub version. A directory name that is not a 40-character hex sha is reported as "I do
-    not know", which routes the write through the explicit head lookup — the safe branch. Returning
-    a bad sha would surface as a 404 `RevisionNotFoundError` on every write instead.
-    """
-    try:
-        parent = Path(path).parent
-        if parent.parent.name != 'snapshots':
-            return None
-        name = parent.name
-        if len(name) == 40 and all(c in '0123456789abcdef' for c in name.lower()):
-            return name
-    except Exception:                                             # noqa: BLE001
-        pass
-    return None
-
-
-class _Projected(dict):
-    """One top-level value of a PROJECTED read — a dict that RAISES for the keys that were dropped.
-
-    ⛔⛔ W32-T01: THE ALTERNATIVE IS A SILENT WRONG ANSWER, WHICH IS WHY THIS CLASS EXISTS AT ALL.
-    A projection built as a plain dict answers `defn.get('rows') or {}` with `{}` — indistinguishable
-    from a database that genuinely has no rows. The caller then paints an empty grid, exports an
-    empty CSV, or (worse) writes that emptiness back. This repo has shipped that exact shape before
-    ([[empty-answer-vs-unfinished-answer]], and a grouped view that painted zero rows).
-
-    So a dropped key is not absent — it is UNREAD, and asking for it raises with the reason and the
-    fix in the message. `KeyError` is the type on purpose: `dict.__getitem__`'s own failure mode, so
-    nothing has to learn a new exception; and the message names `get_projection` so the traceback
-    points at the caller's mistake rather than at this file.
-
-    ⚠ Still a real `dict` for every OTHER purpose — `json.dumps`, `dict(x)`, `**x`, iteration and
-    `in` all behave normally, because a projection has to serialise onto the wire like any
-    definition. Only the named keys are booby-trapped.
-    """
-
-    __slots__ = ('_dropped',)
-
-    def __init__(self, data=None, dropped=()):
-        super().__init__(data or {})
-        object.__setattr__(self, '_dropped', frozenset(dropped))
-
-    def _refuse(self, key):
-        raise KeyError(
-            f'{key!r} was DROPPED by a projected store read and is not absent — this document came '
-            f'from Store.get_projection(drop={sorted(self._dropped)!r}). Reading it as empty would '
-            f'be a silent wrong answer. Use the whole read (Store.get / user_tables.all_tables) if '
-            f'you need {key!r}.')
-
-    def __getitem__(self, key):
-        if key in self._dropped:
-            self._refuse(key)
-        return super().__getitem__(key)
-
-    def get(self, key, default=None):
-        if key in self._dropped:
-            self._refuse(key)
-        return super().get(key, default)
-
-    def pop(self, key, *a):
-        if key in self._dropped:
-            self._refuse(key)
-        return super().pop(key, *a)
-
-
-def _project(doc, drop):
-    """A deep COPY of `doc` with `drop`'s keys removed from every top-level value.
-
-    ⭐ The order is the optimisation: build the retained structure FIRST (a shallow comprehension
-    that never touches the dropped values), then deep-copy THAT. One `json.dumps` over 0.1% of the
-    bytes, instead of one over all of them.
-    """
-    if not isinstance(doc, dict):
-        return json.loads(json.dumps(doc))
-    kept = {k: ({kk: vv for kk, vv in v.items() if kk not in drop} if isinstance(v, dict) else v)
-            for k, v in doc.items()}
-    copied = json.loads(json.dumps(kept))
-    return {k: (_Projected(v, drop) if isinstance(v, dict) else v) for k, v in copied.items()}
-
-
-def is_projected(value):
-    """Did this value come back from a PROJECTED read? — the TAKEN-PATH predicate (W33-T01/D-213).
-
-    ⭐⭐ WHY A PREDICATE AND NOT A COUNTER, and it is the whole reason this exists. `all_defs`
-    FALLS BACK to the whole read on any failure, silently and by design — so a caller that gets the
-    right answer has learned nothing about which read produced it. A gate asserting only "the
-    definitions came back" passes identically when the projection worked and when it quietly did
-    not, which is [[gate-can-report-green-on-nothing]] wearing a green tick.
-
-    ⭐ This is INTRINSIC rather than instrumented: `_project` stamps every top-level value as a
-    `_Projected`, so the evidence rides the document itself. Nothing has to be wrapped, no counter
-    can be installed on the wrong object ([[measure-the-real-call]]), and the answer is the same in
-    production as it is under a gate — a monkey-patched tally is true only while the patch is on.
-
-    ⚠ It answers about ONE VALUE (a table's definition), not about a document: `_project` returns a
-    plain dict OF `_Projected` values, so `is_projected(doc)` is False and `is_projected(doc[key])`
-    is True. That is the right grain — the thing a route holds and reads is the definition.
-    """
-    return isinstance(value, _Projected)
-
-
-class Store:
-    """One repo's persistence: JSON-per-key + raw byte paths, with the wave-7 async flush.
-
-    Every method body is the pre-wave-18 module function, verbatim except globals→fields.
-    The behaviour contracts that matter are stated on each method and are UNCHANGED:
-    strict-read-guards-the-write, exists()-is-True-on-uncertainty, lenient display reads.
-    """
-
-    def __init__(self, repo):
-        self.repo = str(repo)
-        self._lock = threading.RLock()
-        self._cache = {}
-        # --- async flush state (wave-7 W3, 2026-07-28) ------------------------------------
-        # One typed character in a Notes cell used to cost THREE synchronous network calls
-        # inside the component round-trip. update(..., flush='async') applies the mutation to
-        # the in-process cache synchronously (read-your-writes: every reader consults the
-        # cache first) and a background worker coalesces uploads. The restart-wipe protection
-        # is UNCHANGED where it matters: the FIRST update of a key in a fresh process still
-        # strict-reads. Known, accepted: with a second concurrent writer the stale window
-        # widens from ~one round-trip to ~the flush gap.
-        self._owned = {}        # name -> True once cache reflects our own last accepted write
-        self._dirty = {}        # name -> cache is ahead of the hub
-        self._flushers = {}     # name -> live worker thread
-        self._repo_ok = False   # create_repo(exist_ok=True) confirmed once per instance
-        self._last_flush = {}   # name -> monotonic time of the last completed upload
-        # --- the CHANGE TOKEN (wave 29, item 20 / C6) ------------------------------------
-        # name -> how many accepted writes this process has applied to that bucket, and when.
-        # Read by `revision()` in O(1) with NO download and NO deep copy; see its docstring for
-        # why that constraint is the whole feature rather than an optimisation.
-        self._revs = {}
-        # --- ⭐⭐ W31-T13 (D-177(d)) — THE PER-KEY WRITE LOCK ------------------------------
-        # `self._lock` guards the CACHE. It used to be held across a sync `update`'s whole
-        # read-modify-write — a hub download, the caller's `fn`, and a blocking hub upload — so
-        # one write froze every reader in the process for two network round trips. `get()` takes
-        # the same lock, and `/nav`, `/workspace` and every permission check are `get()`.
-        # ⛔ IT IS NOT SAFE TO SIMPLY DROP THE LOCK. Two sync writers of one key would interleave
-        # read-modify-write and one update would vanish — corruption, not slowness. So the two
-        # jobs are separated: this lock serialises WRITERS OF ONE KEY (held across the network,
-        # never touching readers), and `self._lock` is taken only for the microseconds a cache
-        # read or write needs. Readers never wait on a network call again.
-        # ⚠ PER KEY, not one write lock: two tenants' buckets, or `user_tables` and `nav_meta`,
-        # have no reason to serialise against each other.
-        self._key_locks = {}
-        self._key_locks_guard = threading.Lock()
-        self._rev_at = {}
-        # --- ⭐⭐ D-305 — THE TWO FIELDS THAT MAKE A CROSS-PROCESS WRITE SURVIVABLE -----------
-        # `_base[name]`    the hub commit sha this key's cached document was built ON. Learned for
-        #                  free from the download path and from `CommitInfo.oid` after a write, so
-        #                  it costs no extra round trip in the common case.
-        # `_journal[name]` the `fn`s applied to this key since the last CONFIRMED upload — our
-        #                  INTENT, not our result. On a conflict the intent is replayed onto the
-        #                  hub's newer document instead of the result being uploaded over it.
-        #
-        # ⛔ WHY A JOURNAL AND NOT A RETRY. The writers here are DIFFERENT PEOPLE in DIFFERENT
-        # CONTAINERS, and this instance's cache holds an hour of its own edits over a base that has
-        # moved. Re-uploading that snapshot — which is what a retry does — reproduces the loss
-        # exactly. Replaying "add this view" onto the fresher document keeps both writers' work,
-        # which is the only outcome that answers the defect.
-        self._base = {}
-        self._journal = {}
-
-    def available(self):
-        """True when a token is configured — callers degrade gracefully without one."""
-        return bool(_token())
-
-    def _download(self, name):
-        """`(document, commit_sha)` for {name}.json. `({}, None)` ONLY when it is genuinely absent.
-        Any other failure (network, auth, rate-limit, corrupt JSON) RAISES — a transient read
-        error must never be mistaken for an empty store, or a caller could overwrite real data
-        with an empty dict (this is what wiped the user registry on restart).
-
-        ⭐⭐ D-305 — THE SHA COMES FROM THE PATH, WHICH COSTS NOTHING. `hf_hub_download` writes into
-        `…/snapshots/<commit sha>/<file>`, so the revision the bytes came from is already in the
-        return value. Measured 2026-08-18 against a scratch repo: that directory name equals the
-        `CommitInfo.oid` of the commit that wrote the file, and equals
-        `get_hf_file_metadata(...).commit_hash` — which would have cost an extra HTTP round trip on
-        the hottest read in the product. **The parent of our write is therefore known by every
-        reader, for free.**
-
-        ⚠ A sha we cannot parse comes back as `None`, and `_commit` resolves that with an explicit
-        head lookup rather than uploading without a precondition. "I do not know my parent" must
-        take the careful path, never the fast one.
-        """
-        try:
-            path = hf_hub_download(self.repo, f'{name}.json', repo_type='dataset',
-                                   token=_token(), force_download=True)
-            with open(path, encoding='utf-8') as f:
-                return json.load(f), _sha_from_path(path)
-        except (EntryNotFoundError, RepositoryNotFoundError, FileNotFoundError):
-            return {}, None
-        except ValueError as e:
-            # Wave 18 (R2, found by SESSION D): current huggingface_hub wraps a missing REPO in
-            # ValueError("Force download failed…") with the 404 as __cause__ — so a brand-new
-            # tenant repo's very first strict read raised here and aborted its very first write.
-            # A genuinely-absent repo is the same fact as a genuinely-absent file: {}.
-            cause = getattr(e, '__cause__', None)
-            if isinstance(cause, (EntryNotFoundError, RepositoryNotFoundError)):
-                return {}, None
-            raise
-
-    def get(self, name, fresh=False):
-        """Load {name}.json as a dict. LENIENT (for display reads): on a transient fetch error,
-        hand back the last known-good cached value if we have one, else {}. Mutating callers go
-        through update()/_read_strict, which RAISE instead."""
-        with self._lock:
-            if not fresh and name in self._cache:
-                return json.loads(json.dumps(self._cache[name]))   # hand back a copy
-            try:
-                data, sha = self._download(name)
-            except Exception as e:
-                try:                                   # lazy import — core stays layer-clean
-                    import harness.telemetry as _tel   # noqa: PLC0415
-                    _tel.error(f'store:get:{name}', e,
-                               fallback='cached' if name in self._cache else 'empty')
-                except Exception:
-                    pass
-                if name in self._cache:
-                    return json.loads(json.dumps(self._cache[name]))
-                return {}
-            self._cache[name] = data
-            # D-305: the cached document and the revision it derives from are set TOGETHER, always.
-            # A base that can disagree with the cache it describes is worse than no base at all —
-            # it would make a stale write look like a fresh one and pass the precondition.
-            self._base[name] = sha
-            return json.loads(json.dumps(data))
-
-    def get_projection(self, name, drop=()):
-        """Like `get`, but each top-level VALUE comes back WITHOUT the keys named in `drop`.
-
-        ⭐⭐ W32-T01 (D-185) — THE POINT IS THE COPY, NOT THE PAYLOAD SIZE ON THE WIRE.
-
-        `get()` is honest and expensive: it hands back `json.loads(json.dumps(...))` so a caller
-        cannot mutate the cache, and on tenant #0's `user_tables` that copy is **28.6 MB / 81,330
-        rows / ~703 ms warm**, of which **99.9% is `rows`** — which no permission check, no nav
-        render and no workspace envelope ever reads. `/nav` used to pay ~13 of those copies per
-        request; W31-T10 made it ONE, and that one is still 703 ms of pure copying. This is the
-        floor D-185 named: copy only what was asked for.
-
-        ⛔⛔ **THE PROJECTION IS NEVER CACHED, AND THAT LINE IS THE WHOLE SAFETY ARGUMENT.** A
-        rows-less document in `self._cache` would be served to the next WHOLE reader, and the next
-        `update()` would upload it — every row in the tenant deleted, by a read. So the cold path
-        downloads whole, caches WHOLE (exactly as `get` does), and projects the RETURN value only.
-        `verify_store_projection.py` asserts the cache still contains `rows` after a projected read,
-        with an NC that reds when the projection is cached instead.
-
-        ⚠ **`self._lock` is held across the copy, same as `get`** — the retained structure is
-        ~0.1% of the bytes, so this is microseconds where `get` is most of a second. Narrowing it
-        would let a concurrent writer mutate a value mid-serialise for no measurable gain.
-
-        ⚠ **A dropped key RAISES on read rather than reading as absent** (`_Projected`). A
-        projection that answered `{}` for `rows` would paint an empty grid — the exact
-        [[empty-answer-vs-unfinished-answer]] shape this repo has shipped before — and the caller
-        that most wants rows is the one least able to tell. See `_Projected`.
-
-        ⚠ Lenient, like `get`: a transient fetch error falls back to the cached document (projected)
-        and then to `{}`. Mutating callers still go through `update()`/`_read_strict`, which never
-        project — a write must always read whole.
-        """
-        drop = frozenset(str(d) for d in drop)
-        with self._lock:
-            if name in self._cache:
-                return _project(self._cache[name], drop)
-        try:
-            data, sha = self._download(name)
-        except Exception as e:
-            try:                                   # lazy import — core stays layer-clean
-                import harness.telemetry as _tel   # noqa: PLC0415
-                _tel.error(f'store:get_projection:{name}', e,
-                           fallback='cached' if name in self._cache else 'empty')
-            except Exception:
-                pass
-            with self._lock:
-                if name in self._cache:
-                    return _project(self._cache[name], drop)
-            return {}
-        with self._lock:
-            self._cache[name] = data          # ⛔ the WHOLE document, never the projection
-            self._base[name] = sha            # D-305 — set with the cache, never apart from it
-            return _project(data, drop)
-
-    def _key_lock(self, name):
-        """The write lock for ONE key (W31-T13). Created on demand; never removed."""
-        n = str(name)
-        with self._key_locks_guard:
-            lk = self._key_locks.get(n)
-            if lk is None:
-                lk = self._key_locks[n] = threading.RLock()
-            return lk
-
-    def _read_strict(self, name):
-        """Like get(fresh=True) but RAISES on any transient read error ({} only for a genuinely
-        absent file), so a failed read ABORTS a read-modify-write instead of merging into {}.
-
-        ⭐ W31-T13 — THE DOWNLOAD IS OUTSIDE `self._lock` NOW, and the re-check after it is the
-        whole of what makes that safe. A hub read takes a round trip; holding the cache lock
-        across it is what froze every reader. But moving it out opens a window in which an ASYNC
-        writer can accept a row into the cache while our download is in flight — and assigning our
-        (older) hub copy afterwards would erase it. That is D-177's original symptom exactly
-        (`POST /rows` → 201, row gone), re-created by the lock split rather than by a call site.
-        So the dirty mark is re-read AFTER the download, under the short lock, and a key that
-        became dirty keeps its cache.
-        """
-        data, sha = self._download(name)                   # network, no lock held
-        with self._lock:
-            if self._dirty.get(name) and self._owned.get(name) and name in self._cache:
-                # ⛔⛔ D-305 REWROTE THIS BRANCH'S JUSTIFICATION, NOT ITS BEHAVIOUR — AND THE OLD
-                # ONE HAD TO GO, because it asserted the defect. It read: *"a pending upload is
-                # going to overwrite the hub anyway, so the fresh read cannot preserve a remote
-                # write."* That sentence IS the lost update: "we are going to overwrite whatever
-                # the other person did" was stated as a reason to stop worrying about it.
-                #
-                # ⭐ It is no longer true. A pending upload now carries a `parent_commit`
-                # precondition, so it CANNOT overwrite a remote write — it will be refused and
-                # rebased onto it (`_commit`). The branch survives for its OTHER, still-correct
-                # reason: this cache holds an async write nobody has flushed yet, and replacing it
-                # with the hub's copy destroys a row we have already answered 201 for (D-177).
-                # ⇒ keep the local delta here; preserve the REMOTE work at commit time, which is
-                # the only place that can see it.
-                return json.loads(json.dumps(self._cache[name]))
-            self._cache[name] = data
-            self._base[name] = sha
-            return json.loads(json.dumps(data))
-
-    def exists(self, name):
-        """True if {name}.json is confirmed present, False if confirmed absent, and True on ANY
-        uncertainty — a seeder must never clobber a registry we merely failed to reach."""
-        try:
-            files = HfApi(token=_token()).list_repo_files(self.repo, repo_type='dataset')
-            return f'{name}.json' in files
-        except RepositoryNotFoundError:
-            return False
-        except Exception:
-            return True
-
-    # ------------------------------------------------------- the CHANGE TOKEN (wave 29, C6)
-    def _bump(self, name):
-        """Advance one bucket's write counter. THE CALLER HOLDS THE LOCK.
-
-        Called from the two places an accepted write becomes visible to readers of this process
-        — `put()` (which `update(flush='sync')` goes through) and `update()`'s async branch. NOT
-        from `_flush_worker`: that uploads data a bump already accounted for, and counting it
-        again would report a change to every browser each time the coalescing worker woke.
-        """
-        n = str(name)
-        self._revs[n] = self._revs.get(n, 0) + 1
-        self._rev_at[n] = _time.time()
-
-    def revision(self, name):
-        """`{'rev', 'updated_at', 'token'}` for ONE bucket — a dict read, nothing else.
-
-        ⛔ THE ZERO-COST PROPERTY IS THE FEATURE, NOT AN OPTIMISATION, and the arithmetic is why
-        this method exists at all. A browser asking "has anything changed" by re-fetching rows
-        costs THREE full-tenant deep copies under ONE lock (`get()` json-round-trips on the way
-        in and on the way out) on ONE uvicorn process, at a documented ceiling of 35.8 MB / ~1.4 s
-        per bucket — so two tabs polling saturate the server and every unrelated write queues
-        behind them. This performs no download, no `get()`, and no copy of the value.
-        ⇒ Anything added here that reads a bucket destroys the whole point (`verify_live_workspace`
-        counts `get`/`_read_strict`/`_download` on the serving path and goes red on one).
-
-        `rev` and `updated_at` are `core/store_pg.py`'s own column names, on purpose: that backend
-        ALREADY increments `rev` and sets `updated_at = now()` on every write, so the token stops
-        being inert the moment D-4 flips, with the wire shape unchanged.
-
-        ⚠ WHAT IT CANNOT SEE, stated rather than implied: this counter is per PROCESS, so a write
-        by a SECOND container against the same HF dataset is invisible here. That is not a limit
-        this introduces — `get()` caches a bucket for the life of the process and never re-reads
-        it, so the other container's write is invisible to every reader on this backend already.
-        The token is exactly as fresh as the store it reports on. Postgres has no such gap.
-        ⚠ Bytes (`upload_bytes`/`delete_path`) are a different address space and are NOT counted:
-        an attachment is fetched on explicit user action, never polled.
-        """
-        n = str(name)
-        with self._lock:
-            rev = int(self._revs.get(n, 0))
-            at = self._rev_at.get(n)
-        return {'rev': rev, 'updated_at': at, 'token': f'{REV_EPOCH}:{rev}'}
-
-    def _ensure_repo(self):
-        """create_repo is confirmed ONCE per instance — it was a full extra API round-trip on
-        every put(), pure latency after the first success.
-
-        Lifted out of `_upload` (D-305) because `_commit` has to resolve a parent commit BEFORE the
-        first write, and a repo that does not exist yet has no head to resolve.
-        """
-        if not self._repo_ok:
-            HfApi(token=_token()).create_repo(self.repo, repo_type='dataset', private=True,
-                                              exist_ok=True)
-            self._repo_ok = True
-
-    def _head_sha(self):
-        """The repo's current head commit, or `None` if the repo does not exist yet.
-
-        ⚠ ONE EXTRA ROUND TRIP, AND IT IS THE COLD PATH ONLY. `_base[name]` is populated by every
-        download and by every successful write, so this is reached on a first write into a process
-        that has never read the key — and on the rebase path, where a round trip is already the
-        price of correctness.
-        """
-        try:
-            return HfApi(token=_token()).repo_info(self.repo, repo_type='dataset').sha
-        except RepositoryNotFoundError:
-            return None
-
-    def _upload(self, name, data, parent):
-        """The raw hub write, WITH A PRECONDITION. Returns the new head commit sha.
-
-        ⛔ `parent` IS MANDATORY AND HAS NO DEFAULT, deliberately. A default of `None` would make
-        "upload unconditionally" the thing you get by forgetting an argument — and unconditional
-        upload is precisely D-305. Every caller goes through `_commit`, which is the one place
-        allowed to decide what the parent is.
-        """
-        data_binding.check_write(self.repo, f'write {name!r}')     # D-315 backstop, see `update`
-        self._ensure_repo()
-        buf = io.BytesIO(json.dumps(data, ensure_ascii=False, indent=2).encode('utf-8'))
-        info = HfApi(token=_token()).upload_file(
-            path_or_fileobj=buf, path_in_repo=f'{name}.json', repo_id=self.repo,
-            repo_type='dataset',
-            # D-298: under the local opt-in this suffix is what makes a laptop's commit
-            # distinguishable from the live Space's in the store's git history — the only audit
-            # trail this product has.
-            commit_message=f'update {name}' + data_binding.local_commit_suffix(),
-            parent_commit=parent)
-        return getattr(info, 'oid', None)
-
-    def _commit(self, name, data, replay=(), max_rebase=None):
-        """Land `data` for `name` without overwriting anyone — D-305's whole mechanism.
-
-        ⭐⭐ THE SHAPE, and every step of it is measured rather than assumed (scratch-repo probe,
-        2026-08-18):
-
-            1. upload with `parent_commit = the revision our document was built on`;
-            2. the hub answers **412** if the branch moved — and the write does NOT land, so there
-               is nothing to undo;
-            3. re-read the head, REPLAY our intent (`replay`, the `fn`s applied since our last
-               confirmed upload) onto it, and try again against the new parent;
-            4. after `_MAX_REBASE` attempts, RAISE.
-
-        ⭐ REPLAYING THE INTENT IS WHAT MAKES THIS DIFFERENT FROM A RETRY. Uploading our snapshot
-        again would reproduce the loss; re-applying "add this view" to the newer document keeps
-        both writers. The one place it changes an answer is a `fn` that ALLOCATES from the document
-        — `table_store._unique_name` picks the first free display name — which on a rebase may pick
-        a different name than the one already handed to the caller. That is a rename where the
-        pre-fix behaviour was a DELETION, and the trade is stated rather than hidden.
-
-        ⛔ NO PARENT ⇒ RAISE, NEVER UPLOAD BLIND. If neither the cached base nor `_head_sha()` can
-        name a revision, the safe answer is "this write cannot be made safe", not "make it the old
-        way". A fallback that quietly restores the unguarded path is how a fix becomes decorative
-        ([[fallback-that-became-the-rule]]).
-
-        Returns `(document_that_landed, new_sha)` — the document may differ from the one passed in,
-        because a rebase rebuilds it.
-        """
-        data_binding.check_write(self.repo, f'write {name!r}')
-        # ⚠ `_ensure_repo()` IS NOT CALLED HERE, and that is a deliberate reversal. It was, for one
-        # revision, so that a brand-new repo would have a head to name as a parent — and it broke
-        # three existing gates, because their fixtures replace `_upload` (which is where
-        # `create_repo` has always lived) and suddenly found themselves calling the real hub. The
-        # ordering works without it: a repo that does not exist resolves no head and no `exists`,
-        # so the first write goes out with `parent=None` and `_upload` creates the repo on its way
-        # through, exactly as before.
-        budget = _MAX_REBASE if max_rebase is None else int(max_rebase)
-        for attempt in range(budget + 1):
-            with self._lock:
-                parent = self._base.get(name)
-            if not parent:
-                parent = self._head_sha()
-            if not parent:
-                # A repo with no commits at all: `_upload` creates it on the way through, so this
-                # is the genuinely-first write and there is nothing that could be lost.
-                # ⚠ NO EM DASH IN THIS MESSAGE, OR THE ONE BELOW. Both are raised, and a raised
-                # store message reaches the client through the 503 handler, which puts them inside
-                # STANDING RULE 2's scope ("any Python string returned to the client"). `web_prose`
-                # caught exactly these two.
-                if self.exists(name):
-                    raise StoreConflict(
-                        f'cannot resolve a parent commit for {self.repo}:{name}, so this write '
-                        f'cannot be made safe. Refusing to upload without a precondition: an '
-                        f'unconditional write is what silently erases another session\'s work.')
-                parent = None
-            try:
-                oid = self._upload(name, data, parent)
-            except HfHubHTTPError as e:
-                if not _is_conflict(e):
-                    raise            # 401/404/429/500 — a different fault, never a rebase
-                if attempt >= budget:
-                    # ⛔ THE EXHAUSTED CASE RAISES `StoreConflict`, NOT THE RAW 412, and this gate
-                    # caught the first version doing the latter. A caller cannot be asked to know
-                    # that one particular HTTP status out of the hub means "your change was not
-                    # saved and nothing was damaged"; that is a domain fact and it needs a domain
-                    # type. `from e` keeps the hub's own message in the traceback.
-                    raise StoreConflict(
-                        f'{self.repo}:{name} moved under this write {budget + 1} times in a '
-                        f'row. Refusing to overwrite it. Nothing was saved.') from e
-                data = self._rebase(name, replay)
-                continue
-            with self._lock:
-                self._base[name] = oid
-                self._journal[name] = []
-            return data, oid
-        # Unreachable by construction — every iteration returns or raises. Kept so that an edit to
-        # the loop above cannot make this function fall out and return `None`, which callers would
-        # unpack into a crash two frames away from the cause.
-        raise StoreConflict(f'{self.repo}:{name}: the write loop exited without committing.')
-
-    def _journal_append(self, name, fn):
-        """Record one intent and return the replay list. THE CALLER HOLDS THE CACHE LOCK.
-
-        ⚠ CAPPED, BECAUSE AN UNCONFIRMED WRITE STREAM IS AN UNBOUNDED LIST OF CLOSURES. In normal
-        operation the cap is unreachable — `_commit` empties this every few seconds — so it only
-        bites during a sustained hub outage, where the alternative is a process that grows until it
-        dies. When it bites, the OLDEST intents are dropped and it is reported: a rebase afterwards
-        replays the most recent `_MAX_JOURNAL` changes rather than all of them. Stated rather than
-        silently enforced, per the standing rule about limits that cannot be removed.
-        """
-        j = self._journal.setdefault(name, [])
-        j.append(fn)
-        if len(j) > _MAX_JOURNAL:
-            dropped = len(j) - _MAX_JOURNAL
-            del j[:dropped]
-            try:                                       # lazy import — core stays layer-clean
-                import harness.telemetry as _tel       # noqa: PLC0415
-                _tel.error(f'store:journal_full:{name}', StoreConflict(
-                    f'{self.repo}:{name} has {_MAX_JOURNAL}+ unconfirmed writes; dropped '
-                    f'{dropped} oldest replay entr(ies). A rebase will re-apply only the most '
-                    f'recent {_MAX_JOURNAL}.'))
-            except Exception:                          # noqa: BLE001
-                pass
-        return list(j)
-
-    def _rebase(self, name, replay):
-        """Re-read the hub's newer document and re-apply our intent onto it. Caller holds the key
-        lock, so nothing else can be mid-write on this key while we do.
-
-        ⚠ THE CACHE IS REPLACED WITH THE REBASED RESULT, and forgetting that line would make the
-        whole mechanism single-shot: the next write would start from the same stale snapshot and
-        conflict again, burning a rebase per write forever. It also has a second, welcome effect —
-        this instance now HOLDS the other container's work, which is the only moment a process
-        whose cache is never re-read gets to learn that anything changed.
-        """
-        fresh, sha = self._download(name)
-        for fn in list(replay):
-            out = fn(fresh)
-            fresh = out if out is not None else fresh
-        with self._lock:
-            self._cache[name] = json.loads(json.dumps(fresh))
-            self._base[name] = sha
-        try:                                       # lazy import — core stays layer-clean
-            import harness.telemetry as _tel       # noqa: PLC0415
-            _tel.error(f'store:rebase:{name}',
-                       StoreConflict(f'{self.repo}:{name} moved; replayed {len(replay)} change(s)'))
-        except Exception:                          # noqa: BLE001
-            pass
-        return fresh
-
-    def upload_bytes(self, path_in_repo, data, message=None):
-        """Persist RAW BYTES at an arbitrary path in the tenant dataset (wave-8 C5, documents).
-        Bytes are never cached in-process: large, cold, fetched on explicit user action."""
-        if not self.available():
-            raise RuntimeError('No HF_TOKEN configured — persistence is unavailable.')
-        # D-315: a blob is still this tenant's data. ⚠ No `parent_commit` here and that is correct,
-        # not an omission: a blob is written at its own path and read whole, so two writers cannot
-        # silently merge into one damaged document the way `{name}.json` can. The concurrency
-        # hazard D-305 describes is specific to the shared JSON blob.
-        data_binding.check_write(self.repo, f'write bytes {path_in_repo!r}')
-        with self._lock:
-            self._ensure_repo()
-            HfApi(token=_token()).upload_file(
-                path_or_fileobj=io.BytesIO(data), path_in_repo=path_in_repo,
-                repo_id=self.repo, repo_type='dataset',
-                commit_message=(message or f'add {path_in_repo}')
-                + data_binding.local_commit_suffix())
-
-    def download_bytes(self, path_in_repo):
-        """Raw bytes back, or None when absent/unreachable — a missing attachment degrades to
-        "that file is gone", never to a broken page."""
-        if not self.available():
-            return None
-        try:
-            local = hf_hub_download(self.repo, path_in_repo, repo_type='dataset',
-                                    token=_token())
-            return Path(local).read_bytes()
-        except (EntryNotFoundError, RepositoryNotFoundError):
-            return None
-        except Exception:
-            return None
-
-    def delete_path(self, path_in_repo):
-        """Remove a stored file. Absent is SUCCESS — deleting what is already gone is the goal.
-
-        ⛔ THE REFUSAL IS RAISED, NOT SWALLOWED INTO THE `return False` BELOW, and the distinction
-        matters: `False` here means "the delete did not happen", which several callers treat as a
-        tolerable no-op. "You are not allowed to touch this store" is not a no-op — it is the
-        thing the operator has to be told.
-        """
-        if not self.available():
-            return False
-        data_binding.check_write(self.repo, f'delete {path_in_repo!r}')
-        try:
-            HfApi(token=_token()).delete_file(
-                path_in_repo=path_in_repo, repo_id=self.repo, repo_type='dataset',
-                commit_message=f'delete {path_in_repo}' + data_binding.local_commit_suffix())
-            return True
-        except (EntryNotFoundError, RepositoryNotFoundError):
-            return True
-        except Exception:
-            return False
-
-    def put(self, name, data):
-        """Persist {name}.json (creates the private repo on first write). Updates the cache.
-
-        ⭐ D-305 — `put` NOW TAKES THE KEY LOCK AND GOES THROUGH `_commit`, and both halves are
-        required. Without the precondition this method is the hole in the whole mechanism: seven
-        callers outside this file reach it (`users.py`, `modules/ar.py`, `modules/map.py`,
-        `runtime.py`, `store_backend.py`, `ops/seed_pg_from_hf.py`) and `users.json` is a document
-        every account shares, so an unguarded whole-file `put` clobbers exactly as `update` used to.
-
-        ⚠ THE CACHE LOCK NO LONGER SPANS THE NETWORK, which is a change to this method's
-        concurrency and is deliberate. The old body held `self._lock` — the lock every `get()`
-        takes — across the upload, so one `put` froze every reader for a round trip (D-181's
-        shape); with up to `_MAX_REBASE` extra round trips now possible that freeze would get
-        strictly worse. The per-key write lock gives the callers a STRONGER guarantee than they had
-        (two writers of one key can no longer interleave) while readers stop waiting. Lock order is
-        key-then-cache, as everywhere else in this class.
-
-        ⚠ REPLAY SEMANTICS FOR A WHOLE-DOCUMENT WRITE ARE LAST-WRITE-WINS, STATED RATHER THAN
-        IMPLIED. `put` means "the document is this now", so its journal entry replaces rather than
-        merges. What the precondition buys here is that the replacement is applied to a document we
-        have actually SEEN — and that any `update` queued behind it replays on top — instead of
-        being uploaded over an hour of unseen work.
-        """
-        if not self.available():
-            raise RuntimeError('No HF_TOKEN configured — persistence is unavailable.')
-        data_binding.check_write(self.repo, f'put {name!r}')
-        with self._key_lock(name):
-            def _replace(_current):
-                return json.loads(json.dumps(data))
-            with self._lock:
-                replay = self._journal_append(name, _replace)
-            landed, _oid = self._commit(name, data, replay)
-            with self._lock:
-                self._cache[name] = json.loads(json.dumps(landed))
-                self._owned[name] = True
-                self._last_flush[name] = _time.monotonic()
-                self._dirty[name] = False
-                self._bump(name)      # C6 — AFTER the upload: a raise here is not a write
-                return self._cache[name]
-
-    def _flush_worker(self, name):
-        """Coalescing uploader for one key. Runs while dirty; retries with backoff on failure —
-        the cache stays the truth and the dirty mark survives, so nothing is lost silently.
-
-        ⭐⭐ W31-T13 — THIS WORKER IS THE SECOND WRITER, AND THE PER-KEY LOCK MUST COVER IT.
-        ⛔ THE RACE THE LOCK SPLIT OPENED, spelled out because it is silent and survives a restart
-        as real data loss. `update`'s big lock used to span `_read_strict` + `fn` + the upload, so
-        this worker could never snapshot mid-update. With the cache lock narrowed it can:
-            async write        -> `_dirty=True`, this worker scheduled, sleeps `_FLUSH_DELAY`
-            sync update t+0.5  -> takes the dirty branch, runs `fn`, enters `_upload` (no cache
-                                  lock held — that is the whole point of the ticket)
-            worker t+2.0       -> `self._lock` is FREE and `_dirty` is still True (the sync commit
-                                  has not happened yet), so it snapshots the PRE-UPDATE cache,
-                                  clears the mark, and uploads it AFTER the sync upload lands
-            result             -> cache has the new value, the hub has the old one, `_dirty` is
-                                  False, and nothing will ever re-upload. Divergence with no
-                                  symptom until the next process boots and reads the hub.
-        ⚠ `user_tables` has ~40 sync writers against 2 async ones, so that window is the COMMON
-        case, not an exotic one — it is D-177's own shape, re-created one layer down.
-        ⛔ LOCK ORDER IS KEY-THEN-CACHE, EVERYWHERE. `_update_locked` takes them in that order, so
-        this must too; taking `self._lock` first on this path is the deadlock. Both sleeps stay
-        OUTSIDE the key lock — a worker that dozed while holding it would block every writer of
-        that key for `_FLUSH_MIN_GAP`, which is the freeze this ticket exists to remove.
-
-        ⭐⭐ D-305 — THIS WORKER IS ALSO WHERE MOST CROSS-CONTAINER LOSS HAPPENED, because the
-        table workspace (views, fields, folders, overlays) is written `flush='async'`. It uploads a
-        SNAPSHOT of a cache that may be minutes old, so it takes the `replay` journal with it: on a
-        412 the snapshot is thrown away and the journalled `fn`s are re-applied to the hub's newer
-        document instead. The journal is cleared inside `_commit`, only on success.
-        """
-        backoff = 2.0
-        while True:
-            _time.sleep(_FLUSH_DELAY)
-            wait, err, snapshot, replay = None, None, None, ()
-            with self._key_lock(name):
-                with self._lock:
-                    gap = _time.monotonic() - self._last_flush.get(name, 0.0)
-                    if gap < _FLUSH_MIN_GAP:
-                        wait = _FLUSH_MIN_GAP - gap
-                    elif not self._dirty.get(name):
-                        self._flushers.pop(name, None)
-                        return
-                    else:
-                        self._dirty[name] = False
-                        snapshot = json.loads(json.dumps(self._cache.get(name, {})))
-                        # Snapshotted WITH the document and under the same lock: a journal that
-                        # could drift from the snapshot it describes would replay the wrong set.
-                        replay = list(self._journal.get(name) or ())
-                if wait is None:
-                    try:
-                        landed, _oid = self._commit(name, snapshot, replay)
-                        with self._lock:
-                            # A rebase rebuilt the document, so the cache must take the result —
-                            # `_rebase` already wrote it, and this keeps the no-conflict path
-                            # identical to it rather than leaving two ways for the cache to be set.
-                            self._cache[name] = json.loads(json.dumps(landed))
-                            self._last_flush[name] = _time.monotonic()
-                        backoff = 2.0
-                    except Exception as e:                            # noqa: BLE001
-                        with self._lock:
-                            self._dirty[name] = True   # keep it queued; cache is still the truth
-                        err = e
-            if wait is not None:
-                _time.sleep(wait)
-                continue
-            if err is not None:
-                try:
-                    import harness.telemetry as _tel   # noqa: PLC0415
-                    _tel.error(f'store:flush:{name}', err)
-                except Exception:
-                    pass
-                _time.sleep(backoff)
-                backoff = min(backoff * 2, 60.0)
-
-    def _schedule_flush(self, name):
-        """Mark dirty + ensure exactly one live worker for the key. Caller holds the lock."""
-        self._dirty[name] = True
-        t = self._flushers.get(name)
-        if t is not None and t.is_alive():
-            return
-        t = threading.Thread(target=self._flush_worker, args=(name,), daemon=True,
-                             name=f'store-flush-{self.repo}:{name}')
-        self._flushers[name] = t
-        t.start()
-
-    def flush(self, name=None, timeout=30.0):
-        """Block until pending async writes for `name` (or ALL keys) reach the hub — QA gates
-        and shutdown hooks use this; the app itself never needs to."""
-        deadline = _time.monotonic() + timeout
-        names = [name] if name else None
-        while _time.monotonic() < deadline:
-            with self._lock:
-                pending = {k for k, v in self._dirty.items() if v} | {
-                    k for k, t in self._flushers.items() if t.is_alive()}
-            if names is not None:
-                pending &= set(names)
-            if not pending:
-                return True
-            _time.sleep(0.2)
-        return False
-
-    def _flush_now_at_exit(self):
-        """Process-exit safety net: push any dirty keys synchronously, bypassing the coalescing
-        gap — a Space restart must not eat the last seconds of the async window.
-
-        ⚠ W31-T13 — DELIBERATELY NOT KEY-LOCKED, stated so the omission is not read as one. This
-        runs at interpreter exit with the process going away: a writer mid-`update` will never
-        finish, so waiting for its key lock could only turn a partial save into no save at all.
-        Best-effort is the correct contract here, and it is the one case where taking the lock
-        would be worse than skipping it.
-
-        ⭐ D-305 — IT STILL GOES THROUGH `_commit`, so the last write of a dying process cannot be
-        the one that erases somebody. Exiting is not a licence to overwrite: if the branch has
-        moved, this rebases like any other write, and if it cannot, it gives up rather than
-        clobbering. `_commit`'s own `data_binding` check applies too, so a container that was never
-        allowed to write this store does not get one last unguarded upload on the way out.
-        """
-        with self._lock:
-            dirty = [k for k, v in self._dirty.items() if v]
-            snaps = {k: json.loads(json.dumps(self._cache.get(k, {}))) for k in dirty}
-            replays = {k: list(self._journal.get(k) or ()) for k in dirty}
-            for k in dirty:
-                self._dirty[k] = False
-        for k, d in snaps.items():
-            try:
-                # ⚠ ONE REBASE, NOT `_MAX_REBASE`, and the docstring above is the argument: this
-                # runs at interpreter exit with the process going away. The full budget is up to
-                # four extra downloads and five upload attempts PER DIRTY KEY on a path that may be
-                # killed at any moment, which turns a best-effort save into a hang. One attempt
-                # keeps the safety this method exists for (a conflicting write still rebases rather
-                # than clobbering) at a cost the exit path can actually pay.
-                self._commit(k, d, replays.get(k, ()), max_rebase=1)
-            except Exception:
-                pass          # exiting anyway — nothing left to reschedule
-
-    def update(self, name, fn, flush='sync'):
-        """Read-modify-write. STRICT first read (raises on a transient error rather than
-        starting from an empty dict), apply fn(data)->data, persist.
-
-        flush='sync' (default) = fresh strict read + blocking upload. flush='async' (the
-        table-workspace hot path, wave-7 W3) = the mutation applies to the in-process cache and
-        returns immediately; a coalescing worker commits. The strict read still guards the
-        FIRST async update of a key per process.
-
-        ⛔⛔ D-305 — "LAST-WRITE-WINS IS THE CONTRACT" IS NO LONGER TRUE ACROSS PROCESSES, AND THE
-        SENTENCE BELOW USED TO SAY IT WITHOUT THE QUALIFIER. Two writers in ONE process cannot lose
-        an update — they share `self._cache`, which is why every gate in this repo passed over the
-        defect. Two writers in two CONTAINERS could, and did: measured on tenant #0's live store in
-        commit `dee7481a77`, where one session's write rolled the document back an hour and cost a
-        real user 3 views, 49 typed cell values and a field id. The commit now carries a
-        `parent_commit` precondition and rebases (`_commit`), so the contract is: **within a
-        process, last write wins; across processes, a write that would overwrite unseen work is
-        refused and re-applied.** A lost update is not the contract at either grain.
-
-        ⭐⭐ W31-T13 (D-177(d)) — TWO LOCKS, AND THE SPLIT IS THE WHOLE TICKET.
-        `self._key_lock(name)` is held for the entire read-modify-write, so two writers of one key
-        still cannot interleave.
-        `self._lock` — the one every `get()` takes — is now held only for the microseconds a cache
-        read or write needs, never across a download or an upload. Before this, one sync write
-        froze every reader in the process for two network round trips, on a store whose warm
-        `get()` already costs 703 ms of deep copy on tenant #0.
-        ⚠ Uploading outside `self._lock` is not a new posture: `_flush_worker` has done exactly
-        that since wave 7, and `_repo_ok` racing is benign because `create_repo` is `exist_ok=True`.
-        """
-        # ⛔ D-315 — REFUSED AT THE ACCEPTANCE POINT, NOT AT THE UPLOAD. `_upload` carries the same
-        # check as a backstop, but refusing only there would let the async branch accept the
-        # mutation into the cache, answer 200, mark the key dirty, and hand the flush worker a
-        # write that can never succeed — which retries with backoff forever and reports the refusal
-        # to nobody. Refuse before anything is accepted, and the caller gets a 503 that tells the
-        # truth: no change was saved.
-        data_binding.check_write(self.repo, f'update {name!r}')
-        with self._key_lock(name):
-            return self._update_locked(name, fn, flush)
-
-    def _update_locked(self, name, fn, flush):
-        """`update`'s body, with this key's write lock already held. See its docstring.
-
-        ⚠ THE CACHE LOCK IS TAKEN THREE TIMES AND RELEASED BETWEEN THEM, which is only safe
-        because THIS key's write lock is held throughout: nothing else can be mid-write on it, so
-        the state read in the first section is still ours in the third. A caller reaching this
-        without the key lock would have the old interleaving bug back.
-        """
-        # ⚠ A FLAG, NOT A `None` SENTINEL. `None` can be a real cached value, and a sentinel that
-        # can equal real data is not a sentinel — this repo has paid for that exact shape before
-        # (`routes_nav`'s locked-mode default). The flag says "the cache answered", full stop.
-        data, from_cache = None, False
-        with self._lock:
-            if flush == 'async' and self._owned.get(name) and name in self._cache:
-                data, from_cache = json.loads(json.dumps(self._cache[name])), True
-            elif self._dirty.get(name) and self._owned.get(name) and name in self._cache:
-                # ⛔ A SYNC READ-MODIFY-WRITE MUST NOT DISCARD A PENDING ASYNC WRITE.
-                #
-                # THE BUG THIS LINE EXISTS FOR (owner report 2026-08-09, MEASURED on staging):
-                # `POST /rows` answered `201 {"rid":"100"}` and the row never existed.
-                # `user_tables.add_row` commits `flush='async'` — the row lives ONLY in
-                # `self._cache` until the coalescing worker uploads it, which is `_FLUSH_DELAY`
-                # (2s) away and up to `_FLUSH_MIN_GAP` (20s) if that key committed recently.
-                # Inside that window ANY sync writer of the SAME key took the `else` branch
-                # below, and `_read_strict` does `self._cache[name] = self._download(name)` —
-                # it REPLACES the cache with the hub's copy, erasing the only place the new row
-                # existed, and then `put()` uploaded that. Nothing raised. The 201 was truthful
-                # about `add_row`'s return value and false about the outcome.
-                #
-                # ⚠ `user_tables` has ~40 sync writers against 2 async ones, so this was never
-                # one bad call site — it is a property of the KEY. Wave 27 only made it fire
-                # every time, by putting the relation refresh on the add path.
-                #
-                # ⛔ AND RE-DOWNLOADING WAS PROTECTING NOTHING. Once a key is dirty our cache
-                # has already diverged and the pending upload was going to overwrite the hub
-                # anyway — so the fresh read cannot preserve a remote write, it can only destroy
-                # a local one we already acknowledged. This is the class's OWN documented
-                # contract, restored: see `_owned` above — *"read-your-writes: every reader
-                # consults the cache first"* — and `_flush_worker`'s *"the cache stays the truth
-                # and the dirty mark survives, so nothing is lost silently."*
-                #
-                # ⚠ NOT a weakening of `_read_strict`'s purpose. That exists so a FAILED read
-                # aborts the RMW instead of merging into `{}`; the cache is populated
-                # known-good local state, and this branch is only reachable when it is.
-                data, from_cache = json.loads(json.dumps(self._cache[name])), True
-            # else: the strict hub read happens BELOW, outside this lock (W31-T13)
-
-        # ⭐ W31-T13 — THE HUB READ AND THE CALLER'S `fn` RUN WITH NO CACHE LOCK HELD. `fn` is
-        # arbitrary caller code — `automation_engine`'s relation pass has run whole-tenant work
-        # inside one — and it used to execute with every reader in the process blocked behind it.
-        if not from_cache:
-            data = self._read_strict(name)  # network; re-checks the dirty mark on the way back
-        result = fn(data)
-        data = result if result is not None else data
-
-        # ⭐⭐ D-305 — THE INTENT IS RECORDED, NOT JUST THE RESULT, and this one line is what makes
-        # a conflict survivable. `data` is our answer over OUR base; `fn` is what we MEANT, and
-        # only the intent can be re-applied to somebody else's newer document. The journal holds
-        # every `fn` since the last confirmed upload — which for an async key is a whole flush
-        # window of edits, not one — and `_commit` clears it only when the hub has accepted them.
-        # ⚠ Appended AFTER `fn` ran: a `fn` that raised changed nothing and must not be replayed.
-        with self._lock:
-            replay = self._journal_append(name, fn)
-
-        if flush == 'async':
-            if not self.available():
-                raise RuntimeError('No HF_TOKEN configured — persistence is unavailable.')
-            with self._lock:
-                self._cache[name] = json.loads(json.dumps(data))
-                self._owned[name] = True
-                self._bump(name)  # C6 — the async write is ACCEPTED here, not at flush time
-                self._schedule_flush(name)
-            return data
-
-        # ⛔ THE SYNC COMMIT. It used to be spelled out here rather than calling `put()`, because
-        # `put()` held `self._lock` across its own upload — the freeze W31-T13 removed — and is
-        # PUBLIC with seven callers outside this file whose concurrency semantics changing it would
-        # alter. D-305 made that duplication untenable: an unguarded whole-file write is the hole
-        # the precondition exists to close, so BOTH now go through `_commit`, which is the single
-        # place that resolves a parent, rebases and clears the journal. `put` still keeps its own
-        # body (its op REPLACES rather than merges), but neither can upload unconditionally.
-        if not self.available():
-            raise RuntimeError('No HF_TOKEN configured — persistence is unavailable.')
-        landed, _oid = self._commit(name, data, replay)   # network, no cache lock held
-        with self._lock:
-            # ⚠ `landed`, NOT `data`. A rebase rebuilt the document from the hub's newer copy, so
-            # caching what we TRIED to write would leave the cache diverged from what is stored —
-            # and the next write would rebase off that divergence forever.
-            self._cache[name] = json.loads(json.dumps(landed))
-            self._owned[name] = True
-            self._last_flush[name] = _time.monotonic()
-            # ⛔ AFTER THE UPLOAD, NEVER BEFORE — `put()`'s own law. `_commit` raises on failure,
-            # and a bump on that path would have `revision()` report a change that never landed.
-            self._bump(name)
-            # The upload above CARRIED whatever was pending, so the key is no longer ahead of
-            # the hub. Without this the flush worker wakes on a still-dirty mark and re-uploads
-            # identical content — a second commit against the 256-commits/hr budget for nothing.
-            # ⚠ AFTER the upload, for the same reason: on a raise the mark MUST survive (the cache
-            # is still the truth and the row must still be committed by the worker). Clearing it
-            # first would turn a transient upload error into the very silent loss the branch above
-            # exists to stop — which is why it sits here and not before the commit.
-            self._dirty[name] = False
-        return landed
-
-
-# ------------------------------------------------------------------- instances + module API
-_INSTANCES = {}
-_INSTANCES_LOCK = threading.RLock()
-
-
-def for_repo(repo_id):
-    """The HF Store bound to `repo_id` — one instance per repo per process (each with its own
-    cache and flush state).
-
-    ⚠ WAVE 20: this is now the HF-SPECIFIC factory. Callers that want "the right store for this
-    tenant, whatever the backend is" call `handle()` below. `for_repo` keeps its exact old
-    behaviour because the seed/migration tooling has to be able to name the FILE store
-    explicitly while the app runs on Postgres — a migration that could only reach the active
-    backend could not copy between them.
-    """
-    rid = str(repo_id or '').strip() or REPO
-    with _INSTANCES_LOCK:
-        inst = _INSTANCES.get(rid)
-        if inst is None:
-            inst = Store(rid)
-            _INSTANCES[rid] = inst
-        return inst
-
-
-# =============================================================================================
-# WAVE 20 (owner ruling R1, closes DEBT D-4) — THE BACKEND SEAM.
-#
-# `core/store_backend.py` has carried this warning since EXIT-2b, and it was correct at the time:
-#
-#     "Flipping STORE_BACKEND=pg does NOT redirect the ~40 existing callers that say
-#      `import core.store as store` — they are bound to the HF module directly. Rewiring them is
-#      task C-4 … THE DEFAULT IS `hf` AND STAYS `hf` until C-4 says otherwise."
-#
-# **R1 IS C-4.** The owner ruled the cutover on 2026-08-05 (the D-4 trigger that fired: a SECOND
-# server process — `royal-imports/cfo-os` came back as a pinned LIVE environment beside staging,
-# and two containers on one last-write-wins file store is the race B-3 was always about).
-#
-# THE REWIRE IS HERE RATHER THAN IN 28 FILES, and that is a deliberate choice over the obvious
-# alternative of `sed`-ing every `import core.store as store` to `import core.store_backend`:
-#   * every one of those callers means "the store for the tenant I am serving", which is exactly
-#     what this module has always meant. The BACKEND is not their concern and making it their
-#     concern is how one of them gets missed;
-#   * a missed caller under a search-and-replace does not fail — it silently keeps writing to the
-#     file store while everything else writes to Postgres. That is the split-brain this whole
-#     ruling exists to end, reintroduced by the fix for it;
-#   * `store_backend.py`'s stated reason for refusing to do it this way — "no dual-read window and
-#     no way to compare the two stores' contents first" — is satisfied: `ops/seed_pg_from_hf.py`
-#     copies, then `--verify` diffs the two stores key-by-key before anything flips.
-#
-# ⛔ THE IMPORT IS LAZY AND MUST STAY LAZY. `core/store_pg.py` imports psycopg only inside
-# `_pool()`, so an `hf` deployment installs no driver; importing it at module scope here would
-# undo that and make the default backend depend on the optional dependency.
-# =============================================================================================
-
-def backend():
-    """`'hf'` | `'pg'` — validated, resolved per call so a test can flip the env var.
-
-    Mirrors `core.store_backend.name()` deliberately rather than importing it: that module
-    imports THIS one, so reaching back would be a cycle. `verify_store_pg` asserts the two agree
-    on every value, which is the guard against them drifting apart.
-    """
-    raw = (os.environ.get('STORE_BACKEND') or 'hf').strip().lower()
-    if raw not in ('hf', 'pg'):
-        raise RuntimeError(
-            f"STORE_BACKEND={raw!r} is not a backend. Use 'hf' (the HF Dataset store) or 'pg' "
-            f"(Postgres, needs DATABASE_URL). Refusing to guess: a typo that silently served the "
-            f"other store is how data ends up in two places.")
-    return raw
-
-
-def handle(repo_id=None, slug=None):
-    """THE tenant-bound store handle for the ACTIVE backend — the one seam every caller crosses.
-
-    `repo_id` addresses the HF backend (a dataset repo); `slug` addresses Postgres (a schema).
-    Both are passed by `harness.runtime.get_runtime`, which knows both facts, so flipping the
-    backend never needs a lookup table between them — and neither identifier has to be invented
-    for the backend that does not use it.
-    """
-    if backend() == 'pg':
-        import core.store_pg as _pg          # noqa: PLC0415 — lazy: see the banner above
-        return _pg.PgStore(slug or os.environ.get('AIOS_TENANT') or 'royal-imports')
-    return for_repo(repo_id)
-
-
-#: The HF default instance. Kept as a module global (not a property) because the seed/migration
-#: tooling and `_flush_all_at_exit` both need the FILE store by name even when pg is active.
-_DEFAULT = for_repo(REPO)
-
-
-def _d():
-    """The default tenant's store on the active backend — what every module function below uses.
-
-    ⚠ Resolved PER CALL, never cached. A cached default would freeze the backend at import time,
-    and import order is exactly what nobody controls: `api/main.py` imports half the platform
-    before it has read a single environment variable it did not inherit.
-    """
-    return handle() if backend() == 'pg' else _DEFAULT
-
-
-def available():
-    return _d().available()
-
-
-def get(name, fresh=False):
-    return _d().get(name, fresh=fresh)
-
-
-def _read_strict(name):
-    return _d()._read_strict(name)
-
-
-def exists(name):
-    return _d().exists(name)
-
-
-def upload_bytes(path_in_repo, data, message=None):
-    return _d().upload_bytes(path_in_repo, data, message=message)
-
-
-def download_bytes(path_in_repo):
-    return _d().download_bytes(path_in_repo)
-
-
-def delete_path(path_in_repo):
-    return _d().delete_path(path_in_repo)
-
-
-def put(name, data):
-    return _d().put(name, data)
-
-
-def flush(name=None, timeout=30.0):
-    return _d().flush(name=name, timeout=timeout)
-
-
-def update(name, fn, flush='sync'):
-    return _d().update(name, fn, flush=flush)
-
-
-def get_projection(name, drop=()):
-    """A projected read — see `Store.get_projection`. Wave 32, W32-T01 (D-185).
-
-    ⛔ AN ELEVENTH STORE OPERATION, AND IT HAD TO BECOME ONE — the method alone was NOT enough,
-    which is the opposite of what it looks like from `revision()`'s precedent. `revision` could
-    ship method-only for a few hours because its caller held a Store. This one's caller is
-    `harness.runtime.TenantRuntime`, and `TenantRuntime._store()` returns **the MODULE** whenever
-    `store_handle is None` — which is precisely tenant #0, the tenant whose 703 ms `/nav` copy this
-    exists to remove. A method-only projection would have been unreachable from the only place that
-    needed it, while every gate stayed green.
-
-    ⛔ SO `verify_store_pg.IFACE` GAINS IT IN THE SAME CHANGE, and `core/store_pg.py` gains the
-    twin below it. That gate's `extra:` check would red on this function otherwise — correctly:
-    this addresses a key and reads it, so it is a store operation and both backends must answer.
-    `backend`/`handle` are exempt because they SELECT a store; this one reads one.
-    """
-    return _d().get_projection(name, drop=drop)
-
-
-def revision(name):
-    """The bucket's change token — wave 29, C6. See `Store.revision` for why it copies nothing.
-
-    ⛔ THIS IS A TENTH STORE OPERATION, AND IT WAS ONLY ALLOWED TO BECOME ONE ONCE BOTH BACKENDS
-    HAD IT. `verify_store_pg.py` pins this module's public surface at a contracted list precisely
-    so a capability one backend lacks cannot be reached through the shared door and die at the
-    cutover — so this function, `store_pg.revision`, and that gate's `IFACE` entry are ONE change
-    and must stay one. It shipped for a few hours as a METHOD ONLY, deliberately, while the pg half
-    had no owner: a method is a capability the HF store has, a module function is a promise both
-    backends keep.
-    """
-    return _d().revision(name)
-
-
-def _flush_all_at_exit():
-    with _INSTANCES_LOCK:
-        instances = list(_INSTANCES.values())
-    for inst in instances:
-        inst._flush_now_at_exit()
-
-
-import atexit  # noqa: E402  (registered after the class it needs)
-atexit.register(_flush_all_at_exit)
+"""Tiny JSON persistence on a PRIVATE Hugging Face Dataset repo — the platform's ONLY writable store
+(Odoo stays strictly read-only). Holds the user registry, customer notes and per-user tags.
+
+Auth uses HF_TOKEN (a Space secret / .env value with write access to the royal-imports org). The
+repo is created private on first write. Reads use a short in-process cache that writes refresh, so
+within a running instance reads are consistent with writes; a fresh instance re-fetches the latest.
+
+WAVE 18 (C1-TENANT, R2): the store is now a CLASS, one instance per repo. The module-level
+functions below delegate to the default instance (`OS_DATA_REPO`, tenant #0) — every existing
+`import core.store as store` caller is byte-for-byte unaffected, which the gate battery proves.
+`for_repo(repo_id)` hands back the bound instance for a tenant that owns its OWN dataset repo
+(nurilab, gtmlab, …); `harness.runtime.TenantRuntime` binds it from the tenant record. Each
+instance carries its own cache / dirty-marks / flush workers / commit-gap clock, so one tenant's
+write burst can never starve another's flush window.
+"""
+import io
+import os
+import json
+import threading
+import time as _time
+from pathlib import Path
+
+try:
+    from dotenv import load_dotenv
+    load_dotenv(Path(__file__).resolve().parents[1] / '.env')
+except Exception:
+    pass
+
+from huggingface_hub import HfApi, hf_hub_download
+from huggingface_hub.utils import (EntryNotFoundError, HfHubHTTPError,
+                                   RepositoryNotFoundError)
+
+import core.data_binding as data_binding
+
+#: ⛔⛔ D-315 — THE BARE DEFAULT IS GONE. This line used to read
+#: `os.environ.get('OS_DATA_REPO', 'royal-imports/cfo-os-data')`, i.e. every Space that was never
+#: handed the key came up bound to tenant #0's REAL business data — which is how two different
+#: builds wrote it on 2026-08-14. The value it resolves to is unchanged (see
+#: `data_binding.default_store` for why returning the real id beats `None`); what changed is that
+#: WRITING it is now refused unless this deployment is the one that owns it.
+REPO = os.environ.get('OS_DATA_REPO') or data_binding.default_store()
+_FLUSH_DELAY = 2.0      # coalesce a burst of edits into one commit
+_FLUSH_MIN_GAP = 20.0   # floor between commits of one key (the 256-commits/hr budget)
+
+#: ⭐⭐ D-305 — HOW MANY TIMES A WRITE MAY REBASE ONTO A MOVED HEAD BEFORE GIVING UP.
+#:
+#: Bounded rather than unbounded, and the cost is one extra DOWNLOAD per rebase. ⚠ Not an extra
+#: commit: a refused write answers 412 and creates nothing, so only the attempt that finally lands
+#: spends the 256-commits/hr budget. Four is generous for the measured case
+#: (two humans editing one grid) and small enough that a genuinely hot key fails loudly instead of
+#: spinning. ⛔ Hitting the bound RAISES — it never falls through to an unconditional upload, which
+#: is the pre-fix behaviour wearing a retry loop ([[fallback-that-became-the-rule]]).
+_MAX_REBASE = 4
+
+#: How many unconfirmed intents one key may hold before the oldest are dropped. See
+#: `_journal_append` for why a cap exists at all and what is reported when it bites.
+_MAX_JOURNAL = 500
+
+#: WAVE 29 (item 20 / ruling R11, contract C6) — THE CHANGE TOKEN'S PROCESS STAMP.
+#:
+#: `Store`'s revision counter is IN-MEMORY, so a restarted Space would begin again at 0 and a
+#: browser holding `…:4` from the previous process could read `…:4` from the new one and conclude
+#: nothing had changed. Prefixing every token with a per-process stamp makes a restart read as a
+#: change, which is the truthful answer: the client's copy predates this process entirely.
+#:
+#: ⚠ NOT a clock and never compared for order — a token is opaque and compared ONLY for equality.
+REV_EPOCH = f'{int(_time.time()):x}{os.getpid():x}'
+
+
+def _token():
+    return os.environ.get('HF_TOKEN') or None
+
+
+class StoreConflict(RuntimeError):
+    """A write that could not be landed without overwriting somebody else's (D-305).
+
+    Raised only after `_MAX_REBASE` rebases, or when no parent commit can be resolved at all. Both
+    are "we could not make this safe", and both must reach the caller as a failure: the alternative
+    — uploading anyway — is exactly the lost update this whole mechanism exists to prevent, and it
+    would answer 200 while destroying an hour of another person's work.
+    """
+
+
+def _is_conflict(exc):
+    """True for the ONE hub failure that means "the branch moved under you".
+
+    ⛔ NARROW ON PURPOSE, AND THE PROBE IS WHY. Measured against a scratch dataset repo on
+    2026-08-18: a stale `parent_commit` returns **HTTP 412** with the server message *"The branch
+    was updated since you opened this page. Please refresh and try again."*, and the write does NOT
+    land. A NONSENSE `parent_commit` returns a **404 `RevisionNotFoundError`** — a different fault
+    that must not be retried, because rebasing on it would loop against a bad sha. A broad
+    `except HfHubHTTPError` would swallow both, plus every 401/429/500, and turn a network blip
+    into a silent rebase.
+    """
+    return getattr(getattr(exc, 'response', None), 'status_code', None) == 412
+
+
+def _sha_from_path(path):
+    """The commit sha out of a hub cache path, or `None` if it does not look like one.
+
+    ⚠ VALIDATED, NOT ASSUMED. The layout is `…/snapshots/<40-hex>/<file>`, but a caller may have
+    an unusual `HF_HOME`, symlinks disabled (this repo's own machines warn about exactly that), or
+    a future hub version. A directory name that is not a 40-character hex sha is reported as "I do
+    not know", which routes the write through the explicit head lookup — the safe branch. Returning
+    a bad sha would surface as a 404 `RevisionNotFoundError` on every write instead.
+    """
+    try:
+        parent = Path(path).parent
+        if parent.parent.name != 'snapshots':
+            return None
+        name = parent.name
+        if len(name) == 40 and all(c in '0123456789abcdef' for c in name.lower()):
+            return name
+    except Exception:                                             # noqa: BLE001
+        pass
+    return None
+
+
+class _Projected(dict):
+    """One top-level value of a PROJECTED read — a dict that RAISES for the keys that were dropped.
+
+    ⛔⛔ W32-T01: THE ALTERNATIVE IS A SILENT WRONG ANSWER, WHICH IS WHY THIS CLASS EXISTS AT ALL.
+    A projection built as a plain dict answers `defn.get('rows') or {}` with `{}` — indistinguishable
+    from a database that genuinely has no rows. The caller then paints an empty grid, exports an
+    empty CSV, or (worse) writes that emptiness back. This repo has shipped that exact shape before
+    ([[empty-answer-vs-unfinished-answer]], and a grouped view that painted zero rows).
+
+    So a dropped key is not absent — it is UNREAD, and asking for it raises with the reason and the
+    fix in the message. `KeyError` is the type on purpose: `dict.__getitem__`'s own failure mode, so
+    nothing has to learn a new exception; and the message names `get_projection` so the traceback
+    points at the caller's mistake rather than at this file.
+
+    ⚠ Still a real `dict` for every OTHER purpose — `json.dumps`, `dict(x)`, `**x`, iteration and
+    `in` all behave normally, because a projection has to serialise onto the wire like any
+    definition. Only the named keys are booby-trapped.
+    """
+
+    __slots__ = ('_dropped',)
+
+    def __init__(self, data=None, dropped=()):
+        super().__init__(data or {})
+        object.__setattr__(self, '_dropped', frozenset(dropped))
+
+    def _refuse(self, key):
+        raise KeyError(
+            f'{key!r} was DROPPED by a projected store read and is not absent — this document came '
+            f'from Store.get_projection(drop={sorted(self._dropped)!r}). Reading it as empty would '
+            f'be a silent wrong answer. Use the whole read (Store.get / user_tables.all_tables) if '
+            f'you need {key!r}.')
+
+    def __getitem__(self, key):
+        if key in self._dropped:
+            self._refuse(key)
+        return super().__getitem__(key)
+
+    def get(self, key, default=None):
+        if key in self._dropped:
+            self._refuse(key)
+        return super().get(key, default)
+
+    def pop(self, key, *a):
+        if key in self._dropped:
+            self._refuse(key)
+        return super().pop(key, *a)
+
+
+def _project(doc, drop):
+    """A deep COPY of `doc` with `drop`'s keys removed from every top-level value.
+
+    ⭐ The order is the optimisation: build the retained structure FIRST (a shallow comprehension
+    that never touches the dropped values), then deep-copy THAT. One `json.dumps` over 0.1% of the
+    bytes, instead of one over all of them.
+    """
+    if not isinstance(doc, dict):
+        return json.loads(json.dumps(doc))
+    kept = {k: ({kk: vv for kk, vv in v.items() if kk not in drop} if isinstance(v, dict) else v)
+            for k, v in doc.items()}
+    copied = json.loads(json.dumps(kept))
+    return {k: (_Projected(v, drop) if isinstance(v, dict) else v) for k, v in copied.items()}
+
+
+def is_projected(value):
+    """Did this value come back from a PROJECTED read? — the TAKEN-PATH predicate (W33-T01/D-213).
+
+    ⭐⭐ WHY A PREDICATE AND NOT A COUNTER, and it is the whole reason this exists. `all_defs`
+    FALLS BACK to the whole read on any failure, silently and by design — so a caller that gets the
+    right answer has learned nothing about which read produced it. A gate asserting only "the
+    definitions came back" passes identically when the projection worked and when it quietly did
+    not, which is [[gate-can-report-green-on-nothing]] wearing a green tick.
+
+    ⭐ This is INTRINSIC rather than instrumented: `_project` stamps every top-level value as a
+    `_Projected`, so the evidence rides the document itself. Nothing has to be wrapped, no counter
+    can be installed on the wrong object ([[measure-the-real-call]]), and the answer is the same in
+    production as it is under a gate — a monkey-patched tally is true only while the patch is on.
+
+    ⚠ It answers about ONE VALUE (a table's definition), not about a document: `_project` returns a
+    plain dict OF `_Projected` values, so `is_projected(doc)` is False and `is_projected(doc[key])`
+    is True. That is the right grain — the thing a route holds and reads is the definition.
+    """
+    return isinstance(value, _Projected)
+
+
+class Store:
+    """One repo's persistence: JSON-per-key + raw byte paths, with the wave-7 async flush.
+
+    Every method body is the pre-wave-18 module function, verbatim except globals→fields.
+    The behaviour contracts that matter are stated on each method and are UNCHANGED:
+    strict-read-guards-the-write, exists()-is-True-on-uncertainty, lenient display reads.
+    """
+
+    def __init__(self, repo):
+        self.repo = str(repo)
+        self._lock = threading.RLock()
+        self._cache = {}
+        # --- async flush state (wave-7 W3, 2026-07-28) ------------------------------------
+        # One typed character in a Notes cell used to cost THREE synchronous network calls
+        # inside the component round-trip. update(..., flush='async') applies the mutation to
+        # the in-process cache synchronously (read-your-writes: every reader consults the
+        # cache first) and a background worker coalesces uploads. The restart-wipe protection
+        # is UNCHANGED where it matters: the FIRST update of a key in a fresh process still
+        # strict-reads. Known, accepted: with a second concurrent writer the stale window
+        # widens from ~one round-trip to ~the flush gap.
+        self._owned = {}        # name -> True once cache reflects our own last accepted write
+        self._dirty = {}        # name -> cache is ahead of the hub
+        self._flushers = {}     # name -> live worker thread
+        self._repo_ok = False   # create_repo(exist_ok=True) confirmed once per instance
+        self._last_flush = {}   # name -> monotonic time of the last completed upload
+        # --- the CHANGE TOKEN (wave 29, item 20 / C6) ------------------------------------
+        # name -> how many accepted writes this process has applied to that bucket, and when.
+        # Read by `revision()` in O(1) with NO download and NO deep copy; see its docstring for
+        # why that constraint is the whole feature rather than an optimisation.
+        self._revs = {}
+        # --- ⭐⭐ W31-T13 (D-177(d)) — THE PER-KEY WRITE LOCK ------------------------------
+        # `self._lock` guards the CACHE. It used to be held across a sync `update`'s whole
+        # read-modify-write — a hub download, the caller's `fn`, and a blocking hub upload — so
+        # one write froze every reader in the process for two network round trips. `get()` takes
+        # the same lock, and `/nav`, `/workspace` and every permission check are `get()`.
+        # ⛔ IT IS NOT SAFE TO SIMPLY DROP THE LOCK. Two sync writers of one key would interleave
+        # read-modify-write and one update would vanish — corruption, not slowness. So the two
+        # jobs are separated: this lock serialises WRITERS OF ONE KEY (held across the network,
+        # never touching readers), and `self._lock` is taken only for the microseconds a cache
+        # read or write needs. Readers never wait on a network call again.
+        # ⚠ PER KEY, not one write lock: two tenants' buckets, or `user_tables` and `nav_meta`,
+        # have no reason to serialise against each other.
+        self._key_locks = {}
+        self._key_locks_guard = threading.Lock()
+        self._rev_at = {}
+        # --- ⭐⭐ D-305 — THE TWO FIELDS THAT MAKE A CROSS-PROCESS WRITE SURVIVABLE -----------
+        # `_base[name]`    the hub commit sha this key's cached document was built ON. Learned for
+        #                  free from the download path and from `CommitInfo.oid` after a write, so
+        #                  it costs no extra round trip in the common case.
+        # `_journal[name]` the `fn`s applied to this key since the last CONFIRMED upload — our
+        #                  INTENT, not our result. On a conflict the intent is replayed onto the
+        #                  hub's newer document instead of the result being uploaded over it.
+        #
+        # ⛔ WHY A JOURNAL AND NOT A RETRY. The writers here are DIFFERENT PEOPLE in DIFFERENT
+        # CONTAINERS, and this instance's cache holds an hour of its own edits over a base that has
+        # moved. Re-uploading that snapshot — which is what a retry does — reproduces the loss
+        # exactly. Replaying "add this view" onto the fresher document keeps both writers' work,
+        # which is the only outcome that answers the defect.
+        self._base = {}
+        self._journal = {}
+
+    def available(self):
+        """True when a token is configured — callers degrade gracefully without one."""
+        return bool(_token())
+
+    def _download(self, name):
+        """`(document, commit_sha)` for {name}.json. `({}, None)` ONLY when it is genuinely absent.
+        Any other failure (network, auth, rate-limit, corrupt JSON) RAISES — a transient read
+        error must never be mistaken for an empty store, or a caller could overwrite real data
+        with an empty dict (this is what wiped the user registry on restart).
+
+        ⭐⭐ D-305 — THE SHA COMES FROM THE PATH, WHICH COSTS NOTHING. `hf_hub_download` writes into
+        `…/snapshots/<commit sha>/<file>`, so the revision the bytes came from is already in the
+        return value. Measured 2026-08-18 against a scratch repo: that directory name equals the
+        `CommitInfo.oid` of the commit that wrote the file, and equals
+        `get_hf_file_metadata(...).commit_hash` — which would have cost an extra HTTP round trip on
+        the hottest read in the product. **The parent of our write is therefore known by every
+        reader, for free.**
+
+        ⚠ A sha we cannot parse comes back as `None`, and `_commit` resolves that with an explicit
+        head lookup rather than uploading without a precondition. "I do not know my parent" must
+        take the careful path, never the fast one.
+        """
+        try:
+            path = hf_hub_download(self.repo, f'{name}.json', repo_type='dataset',
+                                   token=_token(), force_download=True)
+            with open(path, encoding='utf-8') as f:
+                return json.load(f), _sha_from_path(path)
+        except (EntryNotFoundError, RepositoryNotFoundError, FileNotFoundError):
+            return {}, None
+        except ValueError as e:
+            # Wave 18 (R2, found by SESSION D): current huggingface_hub wraps a missing REPO in
+            # ValueError("Force download failed…") with the 404 as __cause__ — so a brand-new
+            # tenant repo's very first strict read raised here and aborted its very first write.
+            # A genuinely-absent repo is the same fact as a genuinely-absent file: {}.
+            cause = getattr(e, '__cause__', None)
+            if isinstance(cause, (EntryNotFoundError, RepositoryNotFoundError)):
+                return {}, None
+            raise
+
+    def get(self, name, fresh=False):
+        """Load {name}.json as a dict. LENIENT (for display reads): on a transient fetch error,
+        hand back the last known-good cached value if we have one, else {}. Mutating callers go
+        through update()/_read_strict, which RAISE instead."""
+        with self._lock:
+            if not fresh and name in self._cache:
+                return json.loads(json.dumps(self._cache[name]))   # hand back a copy
+            try:
+                data, sha = self._download(name)
+            except Exception as e:
+                try:                                   # lazy import — core stays layer-clean
+                    import harness.telemetry as _tel   # noqa: PLC0415
+                    _tel.error(f'store:get:{name}', e,
+                               fallback='cached' if name in self._cache else 'empty')
+                except Exception:
+                    pass
+                if name in self._cache:
+                    return json.loads(json.dumps(self._cache[name]))
+                return {}
+            self._cache[name] = data
+            # D-305: the cached document and the revision it derives from are set TOGETHER, always.
+            # A base that can disagree with the cache it describes is worse than no base at all —
+            # it would make a stale write look like a fresh one and pass the precondition.
+            self._base[name] = sha
+            return json.loads(json.dumps(data))
+
+    def get_projection(self, name, drop=()):
+        """Like `get`, but each top-level VALUE comes back WITHOUT the keys named in `drop`.
+
+        ⭐⭐ W32-T01 (D-185) — THE POINT IS THE COPY, NOT THE PAYLOAD SIZE ON THE WIRE.
+
+        `get()` is honest and expensive: it hands back `json.loads(json.dumps(...))` so a caller
+        cannot mutate the cache, and on tenant #0's `user_tables` that copy is **28.6 MB / 81,330
+        rows / ~703 ms warm**, of which **99.9% is `rows`** — which no permission check, no nav
+        render and no workspace envelope ever reads. `/nav` used to pay ~13 of those copies per
+        request; W31-T10 made it ONE, and that one is still 703 ms of pure copying. This is the
+        floor D-185 named: copy only what was asked for.
+
+        ⛔⛔ **THE PROJECTION IS NEVER CACHED, AND THAT LINE IS THE WHOLE SAFETY ARGUMENT.** A
+        rows-less document in `self._cache` would be served to the next WHOLE reader, and the next
+        `update()` would upload it — every row in the tenant deleted, by a read. So the cold path
+        downloads whole, caches WHOLE (exactly as `get` does), and projects the RETURN value only.
+        `verify_store_projection.py` asserts the cache still contains `rows` after a projected read,
+        with an NC that reds when the projection is cached instead.
+
+        ⚠ **`self._lock` is held across the copy, same as `get`** — the retained structure is
+        ~0.1% of the bytes, so this is microseconds where `get` is most of a second. Narrowing it
+        would let a concurrent writer mutate a value mid-serialise for no measurable gain.
+
+        ⚠ **A dropped key RAISES on read rather than reading as absent** (`_Projected`). A
+        projection that answered `{}` for `rows` would paint an empty grid — the exact
+        [[empty-answer-vs-unfinished-answer]] shape this repo has shipped before — and the caller
+        that most wants rows is the one least able to tell. See `_Projected`.
+
+        ⚠ Lenient, like `get`: a transient fetch error falls back to the cached document (projected)
+        and then to `{}`. Mutating callers still go through `update()`/`_read_strict`, which never
+        project — a write must always read whole.
+        """
+        drop = frozenset(str(d) for d in drop)
+        with self._lock:
+            if name in self._cache:
+                return _project(self._cache[name], drop)
+        try:
+            data, sha = self._download(name)
+        except Exception as e:
+            try:                                   # lazy import — core stays layer-clean
+                import harness.telemetry as _tel   # noqa: PLC0415
+                _tel.error(f'store:get_projection:{name}', e,
+                           fallback='cached' if name in self._cache else 'empty')
+            except Exception:
+                pass
+            with self._lock:
+                if name in self._cache:
+                    return _project(self._cache[name], drop)
+            return {}
+        with self._lock:
+            self._cache[name] = data          # ⛔ the WHOLE document, never the projection
+            self._base[name] = sha            # D-305 — set with the cache, never apart from it
+            return _project(data, drop)
+
+    def _key_lock(self, name):
+        """The write lock for ONE key (W31-T13). Created on demand; never removed."""
+        n = str(name)
+        with self._key_locks_guard:
+            lk = self._key_locks.get(n)
+            if lk is None:
+                lk = self._key_locks[n] = threading.RLock()
+            return lk
+
+    def _read_strict(self, name):
+        """Like get(fresh=True) but RAISES on any transient read error ({} only for a genuinely
+        absent file), so a failed read ABORTS a read-modify-write instead of merging into {}.
+
+        ⭐ W31-T13 — THE DOWNLOAD IS OUTSIDE `self._lock` NOW, and the re-check after it is the
+        whole of what makes that safe. A hub read takes a round trip; holding the cache lock
+        across it is what froze every reader. But moving it out opens a window in which an ASYNC
+        writer can accept a row into the cache while our download is in flight — and assigning our
+        (older) hub copy afterwards would erase it. That is D-177's original symptom exactly
+        (`POST /rows` → 201, row gone), re-created by the lock split rather than by a call site.
+        So the dirty mark is re-read AFTER the download, under the short lock, and a key that
+        became dirty keeps its cache.
+        """
+        data, sha = self._download(name)                   # network, no lock held
+        with self._lock:
+            if self._dirty.get(name) and self._owned.get(name) and name in self._cache:
+                # ⛔⛔ D-305 REWROTE THIS BRANCH'S JUSTIFICATION, NOT ITS BEHAVIOUR — AND THE OLD
+                # ONE HAD TO GO, because it asserted the defect. It read: *"a pending upload is
+                # going to overwrite the hub anyway, so the fresh read cannot preserve a remote
+                # write."* That sentence IS the lost update: "we are going to overwrite whatever
+                # the other person did" was stated as a reason to stop worrying about it.
+                #
+                # ⭐ It is no longer true. A pending upload now carries a `parent_commit`
+                # precondition, so it CANNOT overwrite a remote write — it will be refused and
+                # rebased onto it (`_commit`). The branch survives for its OTHER, still-correct
+                # reason: this cache holds an async write nobody has flushed yet, and replacing it
+                # with the hub's copy destroys a row we have already answered 201 for (D-177).
+                # ⇒ keep the local delta here; preserve the REMOTE work at commit time, which is
+                # the only place that can see it.
+                return json.loads(json.dumps(self._cache[name]))
+            self._cache[name] = data
+            self._base[name] = sha
+            return json.loads(json.dumps(data))
+
+    def exists(self, name):
+        """True if {name}.json is confirmed present, False if confirmed absent, and True on ANY
+        uncertainty — a seeder must never clobber a registry we merely failed to reach."""
+        try:
+            files = HfApi(token=_token()).list_repo_files(self.repo, repo_type='dataset')
+            return f'{name}.json' in files
+        except RepositoryNotFoundError:
+            return False
+        except Exception:
+            return True
+
+    # ------------------------------------------------------- the CHANGE TOKEN (wave 29, C6)
+    def _bump(self, name):
+        """Advance one bucket's write counter. THE CALLER HOLDS THE LOCK.
+
+        Called from the two places an accepted write becomes visible to readers of this process
+        — `put()` (which `update(flush='sync')` goes through) and `update()`'s async branch. NOT
+        from `_flush_worker`: that uploads data a bump already accounted for, and counting it
+        again would report a change to every browser each time the coalescing worker woke.
+        """
+        n = str(name)
+        self._revs[n] = self._revs.get(n, 0) + 1
+        self._rev_at[n] = _time.time()
+
+    def revision(self, name):
+        """`{'rev', 'updated_at', 'token'}` for ONE bucket — a dict read, nothing else.
+
+        ⛔ THE ZERO-COST PROPERTY IS THE FEATURE, NOT AN OPTIMISATION, and the arithmetic is why
+        this method exists at all. A browser asking "has anything changed" by re-fetching rows
+        costs THREE full-tenant deep copies under ONE lock (`get()` json-round-trips on the way
+        in and on the way out) on ONE uvicorn process, at a documented ceiling of 35.8 MB / ~1.4 s
+        per bucket — so two tabs polling saturate the server and every unrelated write queues
+        behind them. This performs no download, no `get()`, and no copy of the value.
+        ⇒ Anything added here that reads a bucket destroys the whole point (`verify_live_workspace`
+        counts `get`/`_read_strict`/`_download` on the serving path and goes red on one).
+
+        `rev` and `updated_at` are `core/store_pg.py`'s own column names, on purpose: that backend
+        ALREADY increments `rev` and sets `updated_at = now()` on every write, so the token stops
+        being inert the moment D-4 flips, with the wire shape unchanged.
+
+        ⚠ WHAT IT CANNOT SEE, stated rather than implied: this counter is per PROCESS, so a write
+        by a SECOND container against the same HF dataset is invisible here. That is not a limit
+        this introduces — `get()` caches a bucket for the life of the process and never re-reads
+        it, so the other container's write is invisible to every reader on this backend already.
+        The token is exactly as fresh as the store it reports on. Postgres has no such gap.
+        ⚠ Bytes (`upload_bytes`/`delete_path`) are a different address space and are NOT counted:
+        an attachment is fetched on explicit user action, never polled.
+        """
+        n = str(name)
+        with self._lock:
+            rev = int(self._revs.get(n, 0))
+            at = self._rev_at.get(n)
+        return {'rev': rev, 'updated_at': at, 'token': f'{REV_EPOCH}:{rev}'}
+
+    def _ensure_repo(self):
+        """create_repo is confirmed ONCE per instance — it was a full extra API round-trip on
+        every put(), pure latency after the first success.
+
+        Lifted out of `_upload` (D-305) because `_commit` has to resolve a parent commit BEFORE the
+        first write, and a repo that does not exist yet has no head to resolve.
+        """
+        if not self._repo_ok:
+            HfApi(token=_token()).create_repo(self.repo, repo_type='dataset', private=True,
+                                              exist_ok=True)
+            self._repo_ok = True
+
+    def _head_sha(self):
+        """The repo's current head commit, or `None` if the repo does not exist yet.
+
+        ⚠ ONE EXTRA ROUND TRIP, AND IT IS THE COLD PATH ONLY. `_base[name]` is populated by every
+        download and by every successful write, so this is reached on a first write into a process
+        that has never read the key — and on the rebase path, where a round trip is already the
+        price of correctness.
+        """
+        try:
+            return HfApi(token=_token()).repo_info(self.repo, repo_type='dataset').sha
+        except RepositoryNotFoundError:
+            return None
+
+    def _upload(self, name, data, parent):
+        """The raw hub write, WITH A PRECONDITION. Returns the new head commit sha.
+
+        ⛔ `parent` IS MANDATORY AND HAS NO DEFAULT, deliberately. A default of `None` would make
+        "upload unconditionally" the thing you get by forgetting an argument — and unconditional
+        upload is precisely D-305. Every caller goes through `_commit`, which is the one place
+        allowed to decide what the parent is.
+        """
+        data_binding.check_write(self.repo, f'write {name!r}')     # D-315 backstop, see `update`
+        self._ensure_repo()
+        buf = io.BytesIO(json.dumps(data, ensure_ascii=False, indent=2).encode('utf-8'))
+        info = HfApi(token=_token()).upload_file(
+            path_or_fileobj=buf, path_in_repo=f'{name}.json', repo_id=self.repo,
+            repo_type='dataset',
+            # D-298: under the local opt-in this suffix is what makes a laptop's commit
+            # distinguishable from the live Space's in the store's git history — the only audit
+            # trail this product has.
+            commit_message=f'update {name}' + data_binding.local_commit_suffix(),
+            parent_commit=parent)
+        return getattr(info, 'oid', None)
+
+    def _commit(self, name, data, replay=(), max_rebase=None):
+        """Land `data` for `name` without overwriting anyone — D-305's whole mechanism.
+
+        ⭐⭐ THE SHAPE, and every step of it is measured rather than assumed (scratch-repo probe,
+        2026-08-18):
+
+            1. upload with `parent_commit = the revision our document was built on`;
+            2. the hub answers **412** if the branch moved — and the write does NOT land, so there
+               is nothing to undo;
+            3. re-read the head, REPLAY our intent (`replay`, the `fn`s applied since our last
+               confirmed upload) onto it, and try again against the new parent;
+            4. after `_MAX_REBASE` attempts, RAISE.
+
+        ⭐ REPLAYING THE INTENT IS WHAT MAKES THIS DIFFERENT FROM A RETRY. Uploading our snapshot
+        again would reproduce the loss; re-applying "add this view" to the newer document keeps
+        both writers. The one place it changes an answer is a `fn` that ALLOCATES from the document
+        — `table_store._unique_name` picks the first free display name — which on a rebase may pick
+        a different name than the one already handed to the caller. That is a rename where the
+        pre-fix behaviour was a DELETION, and the trade is stated rather than hidden.
+
+        ⛔ NO PARENT ⇒ RAISE, NEVER UPLOAD BLIND. If neither the cached base nor `_head_sha()` can
+        name a revision, the safe answer is "this write cannot be made safe", not "make it the old
+        way". A fallback that quietly restores the unguarded path is how a fix becomes decorative
+        ([[fallback-that-became-the-rule]]).
+
+        Returns `(document_that_landed, new_sha)` — the document may differ from the one passed in,
+        because a rebase rebuilds it.
+        """
+        data_binding.check_write(self.repo, f'write {name!r}')
+        # ⚠ `_ensure_repo()` IS NOT CALLED HERE, and that is a deliberate reversal. It was, for one
+        # revision, so that a brand-new repo would have a head to name as a parent — and it broke
+        # three existing gates, because their fixtures replace `_upload` (which is where
+        # `create_repo` has always lived) and suddenly found themselves calling the real hub. The
+        # ordering works without it: a repo that does not exist resolves no head and no `exists`,
+        # so the first write goes out with `parent=None` and `_upload` creates the repo on its way
+        # through, exactly as before.
+        budget = _MAX_REBASE if max_rebase is None else int(max_rebase)
+        for attempt in range(budget + 1):
+            with self._lock:
+                parent = self._base.get(name)
+            if not parent:
+                parent = self._head_sha()
+            if not parent:
+                # A repo with no commits at all: `_upload` creates it on the way through, so this
+                # is the genuinely-first write and there is nothing that could be lost.
+                # ⚠ NO EM DASH IN THIS MESSAGE, OR THE ONE BELOW. Both are raised, and a raised
+                # store message reaches the client through the 503 handler, which puts them inside
+                # STANDING RULE 2's scope ("any Python string returned to the client"). `web_prose`
+                # caught exactly these two.
+                if self.exists(name):
+                    raise StoreConflict(
+                        f'cannot resolve a parent commit for {self.repo}:{name}, so this write '
+                        f'cannot be made safe. Refusing to upload without a precondition: an '
+                        f'unconditional write is what silently erases another session\'s work.')
+                parent = None
+            try:
+                oid = self._upload(name, data, parent)
+            except HfHubHTTPError as e:
+                if not _is_conflict(e):
+                    raise            # 401/404/429/500 — a different fault, never a rebase
+                if attempt >= budget:
+                    # ⛔ THE EXHAUSTED CASE RAISES `StoreConflict`, NOT THE RAW 412, and this gate
+                    # caught the first version doing the latter. A caller cannot be asked to know
+                    # that one particular HTTP status out of the hub means "your change was not
+                    # saved and nothing was damaged"; that is a domain fact and it needs a domain
+                    # type. `from e` keeps the hub's own message in the traceback.
+                    raise StoreConflict(
+                        f'{self.repo}:{name} moved under this write {budget + 1} times in a '
+                        f'row. Refusing to overwrite it. Nothing was saved.') from e
+                data = self._rebase(name, replay)
+                continue
+            with self._lock:
+                self._base[name] = oid
+                self._journal[name] = []
+            return data, oid
+        # Unreachable by construction — every iteration returns or raises. Kept so that an edit to
+        # the loop above cannot make this function fall out and return `None`, which callers would
+        # unpack into a crash two frames away from the cause.
+        raise StoreConflict(f'{self.repo}:{name}: the write loop exited without committing.')
+
+    def _journal_append(self, name, fn):
+        """Record one intent and return the replay list. THE CALLER HOLDS THE CACHE LOCK.
+
+        ⚠ CAPPED, BECAUSE AN UNCONFIRMED WRITE STREAM IS AN UNBOUNDED LIST OF CLOSURES. In normal
+        operation the cap is unreachable — `_commit` empties this every few seconds — so it only
+        bites during a sustained hub outage, where the alternative is a process that grows until it
+        dies. When it bites, the OLDEST intents are dropped and it is reported: a rebase afterwards
+        replays the most recent `_MAX_JOURNAL` changes rather than all of them. Stated rather than
+        silently enforced, per the standing rule about limits that cannot be removed.
+        """
+        j = self._journal.setdefault(name, [])
+        j.append(fn)
+        if len(j) > _MAX_JOURNAL:
+            dropped = len(j) - _MAX_JOURNAL
+            del j[:dropped]
+            try:                                       # lazy import — core stays layer-clean
+                import harness.telemetry as _tel       # noqa: PLC0415
+                _tel.error(f'store:journal_full:{name}', StoreConflict(
+                    f'{self.repo}:{name} has {_MAX_JOURNAL}+ unconfirmed writes; dropped '
+                    f'{dropped} oldest replay entr(ies). A rebase will re-apply only the most '
+                    f'recent {_MAX_JOURNAL}.'))
+            except Exception:                          # noqa: BLE001
+                pass
+        return list(j)
+
+    def _rebase(self, name, replay):
+        """Re-read the hub's newer document and re-apply our intent onto it. Caller holds the key
+        lock, so nothing else can be mid-write on this key while we do.
+
+        ⚠ THE CACHE IS REPLACED WITH THE REBASED RESULT, and forgetting that line would make the
+        whole mechanism single-shot: the next write would start from the same stale snapshot and
+        conflict again, burning a rebase per write forever. It also has a second, welcome effect —
+        this instance now HOLDS the other container's work, which is the only moment a process
+        whose cache is never re-read gets to learn that anything changed.
+        """
+        fresh, sha = self._download(name)
+        for fn in list(replay):
+            out = fn(fresh)
+            fresh = out if out is not None else fresh
+        with self._lock:
+            self._cache[name] = json.loads(json.dumps(fresh))
+            self._base[name] = sha
+        try:                                       # lazy import — core stays layer-clean
+            import harness.telemetry as _tel       # noqa: PLC0415
+            _tel.error(f'store:rebase:{name}',
+                       StoreConflict(f'{self.repo}:{name} moved; replayed {len(replay)} change(s)'))
+        except Exception:                          # noqa: BLE001
+            pass
+        return fresh
+
+    def upload_bytes(self, path_in_repo, data, message=None):
+        """Persist RAW BYTES at an arbitrary path in the tenant dataset (wave-8 C5, documents).
+        Bytes are never cached in-process: large, cold, fetched on explicit user action."""
+        if not self.available():
+            raise RuntimeError('No HF_TOKEN configured — persistence is unavailable.')
+        # D-315: a blob is still this tenant's data. ⚠ No `parent_commit` here and that is correct,
+        # not an omission: a blob is written at its own path and read whole, so two writers cannot
+        # silently merge into one damaged document the way `{name}.json` can. The concurrency
+        # hazard D-305 describes is specific to the shared JSON blob.
+        data_binding.check_write(self.repo, f'write bytes {path_in_repo!r}')
+        with self._lock:
+            self._ensure_repo()
+            HfApi(token=_token()).upload_file(
+                path_or_fileobj=io.BytesIO(data), path_in_repo=path_in_repo,
+                repo_id=self.repo, repo_type='dataset',
+                commit_message=(message or f'add {path_in_repo}')
+                + data_binding.local_commit_suffix())
+
+    def download_bytes(self, path_in_repo):
+        """Raw bytes back, or None when absent/unreachable — a missing attachment degrades to
+        "that file is gone", never to a broken page."""
+        if not self.available():
+            return None
+        try:
+            local = hf_hub_download(self.repo, path_in_repo, repo_type='dataset',
+                                    token=_token())
+            return Path(local).read_bytes()
+        except (EntryNotFoundError, RepositoryNotFoundError):
+            return None
+        except Exception:
+            return None
+
+    def delete_path(self, path_in_repo):
+        """Remove a stored file. Absent is SUCCESS — deleting what is already gone is the goal.
+
+        ⛔ THE REFUSAL IS RAISED, NOT SWALLOWED INTO THE `return False` BELOW, and the distinction
+        matters: `False` here means "the delete did not happen", which several callers treat as a
+        tolerable no-op. "You are not allowed to touch this store" is not a no-op — it is the
+        thing the operator has to be told.
+        """
+        if not self.available():
+            return False
+        data_binding.check_write(self.repo, f'delete {path_in_repo!r}')
+        try:
+            HfApi(token=_token()).delete_file(
+                path_in_repo=path_in_repo, repo_id=self.repo, repo_type='dataset',
+                commit_message=f'delete {path_in_repo}' + data_binding.local_commit_suffix())
+            return True
+        except (EntryNotFoundError, RepositoryNotFoundError):
+            return True
+        except Exception:
+            return False
+
+    def put(self, name, data):
+        """Persist {name}.json (creates the private repo on first write). Updates the cache.
+
+        ⭐ D-305 — `put` NOW TAKES THE KEY LOCK AND GOES THROUGH `_commit`, and both halves are
+        required. Without the precondition this method is the hole in the whole mechanism: seven
+        callers outside this file reach it (`users.py`, `modules/ar.py`, `modules/map.py`,
+        `runtime.py`, `store_backend.py`, `ops/seed_pg_from_hf.py`) and `users.json` is a document
+        every account shares, so an unguarded whole-file `put` clobbers exactly as `update` used to.
+
+        ⚠ THE CACHE LOCK NO LONGER SPANS THE NETWORK, which is a change to this method's
+        concurrency and is deliberate. The old body held `self._lock` — the lock every `get()`
+        takes — across the upload, so one `put` froze every reader for a round trip (D-181's
+        shape); with up to `_MAX_REBASE` extra round trips now possible that freeze would get
+        strictly worse. The per-key write lock gives the callers a STRONGER guarantee than they had
+        (two writers of one key can no longer interleave) while readers stop waiting. Lock order is
+        key-then-cache, as everywhere else in this class.
+
+        ⚠ REPLAY SEMANTICS FOR A WHOLE-DOCUMENT WRITE ARE LAST-WRITE-WINS, STATED RATHER THAN
+        IMPLIED. `put` means "the document is this now", so its journal entry replaces rather than
+        merges. What the precondition buys here is that the replacement is applied to a document we
+        have actually SEEN — and that any `update` queued behind it replays on top — instead of
+        being uploaded over an hour of unseen work.
+        """
+        if not self.available():
+            raise RuntimeError('No HF_TOKEN configured — persistence is unavailable.')
+        data_binding.check_write(self.repo, f'put {name!r}')
+        with self._key_lock(name):
+            def _replace(_current):
+                return json.loads(json.dumps(data))
+            with self._lock:
+                replay = self._journal_append(name, _replace)
+            landed, _oid = self._commit(name, data, replay)
+            with self._lock:
+                self._cache[name] = json.loads(json.dumps(landed))
+                self._owned[name] = True
+                self._last_flush[name] = _time.monotonic()
+                self._dirty[name] = False
+                self._bump(name)      # C6 — AFTER the upload: a raise here is not a write
+                return self._cache[name]
+
+    def _flush_worker(self, name):
+        """Coalescing uploader for one key. Runs while dirty; retries with backoff on failure —
+        the cache stays the truth and the dirty mark survives, so nothing is lost silently.
+
+        ⭐⭐ W31-T13 — THIS WORKER IS THE SECOND WRITER, AND THE PER-KEY LOCK MUST COVER IT.
+        ⛔ THE RACE THE LOCK SPLIT OPENED, spelled out because it is silent and survives a restart
+        as real data loss. `update`'s big lock used to span `_read_strict` + `fn` + the upload, so
+        this worker could never snapshot mid-update. With the cache lock narrowed it can:
+            async write        -> `_dirty=True`, this worker scheduled, sleeps `_FLUSH_DELAY`
+            sync update t+0.5  -> takes the dirty branch, runs `fn`, enters `_upload` (no cache
+                                  lock held — that is the whole point of the ticket)
+            worker t+2.0       -> `self._lock` is FREE and `_dirty` is still True (the sync commit
+                                  has not happened yet), so it snapshots the PRE-UPDATE cache,
+                                  clears the mark, and uploads it AFTER the sync upload lands
+            result             -> cache has the new value, the hub has the old one, `_dirty` is
+                                  False, and nothing will ever re-upload. Divergence with no
+                                  symptom until the next process boots and reads the hub.
+        ⚠ `user_tables` has ~40 sync writers against 2 async ones, so that window is the COMMON
+        case, not an exotic one — it is D-177's own shape, re-created one layer down.
+        ⛔ LOCK ORDER IS KEY-THEN-CACHE, EVERYWHERE. `_update_locked` takes them in that order, so
+        this must too; taking `self._lock` first on this path is the deadlock. Both sleeps stay
+        OUTSIDE the key lock — a worker that dozed while holding it would block every writer of
+        that key for `_FLUSH_MIN_GAP`, which is the freeze this ticket exists to remove.
+
+        ⭐⭐ D-305 — THIS WORKER IS ALSO WHERE MOST CROSS-CONTAINER LOSS HAPPENED, because the
+        table workspace (views, fields, folders, overlays) is written `flush='async'`. It uploads a
+        SNAPSHOT of a cache that may be minutes old, so it takes the `replay` journal with it: on a
+        412 the snapshot is thrown away and the journalled `fn`s are re-applied to the hub's newer
+        document instead. The journal is cleared inside `_commit`, only on success.
+        """
+        backoff = 2.0
+        while True:
+            _time.sleep(_FLUSH_DELAY)
+            wait, err, snapshot, replay = None, None, None, ()
+            with self._key_lock(name):
+                with self._lock:
+                    gap = _time.monotonic() - self._last_flush.get(name, 0.0)
+                    if gap < _FLUSH_MIN_GAP:
+                        wait = _FLUSH_MIN_GAP - gap
+                    elif not self._dirty.get(name):
+                        self._flushers.pop(name, None)
+                        return
+                    else:
+                        self._dirty[name] = False
+                        snapshot = json.loads(json.dumps(self._cache.get(name, {})))
+                        # Snapshotted WITH the document and under the same lock: a journal that
+                        # could drift from the snapshot it describes would replay the wrong set.
+                        replay = list(self._journal.get(name) or ())
+                if wait is None:
+                    try:
+                        landed, _oid = self._commit(name, snapshot, replay)
+                        with self._lock:
+                            # A rebase rebuilt the document, so the cache must take the result —
+                            # `_rebase` already wrote it, and this keeps the no-conflict path
+                            # identical to it rather than leaving two ways for the cache to be set.
+                            self._cache[name] = json.loads(json.dumps(landed))
+                            self._last_flush[name] = _time.monotonic()
+                        backoff = 2.0
+                    except Exception as e:                            # noqa: BLE001
+                        with self._lock:
+                            self._dirty[name] = True   # keep it queued; cache is still the truth
+                        err = e
+            if wait is not None:
+                _time.sleep(wait)
+                continue
+            if err is not None:
+                try:
+                    import harness.telemetry as _tel   # noqa: PLC0415
+                    _tel.error(f'store:flush:{name}', err)
+                except Exception:
+                    pass
+                _time.sleep(backoff)
+                backoff = min(backoff * 2, 60.0)
+
+    def _schedule_flush(self, name):
+        """Mark dirty + ensure exactly one live worker for the key. Caller holds the lock."""
+        self._dirty[name] = True
+        t = self._flushers.get(name)
+        if t is not None and t.is_alive():
+            return
+        t = threading.Thread(target=self._flush_worker, args=(name,), daemon=True,
+                             name=f'store-flush-{self.repo}:{name}')
+        self._flushers[name] = t
+        t.start()
+
+    def flush(self, name=None, timeout=30.0):
+        """Block until pending async writes for `name` (or ALL keys) reach the hub — QA gates
+        and shutdown hooks use this; the app itself never needs to."""
+        deadline = _time.monotonic() + timeout
+        names = [name] if name else None
+        while _time.monotonic() < deadline:
+            with self._lock:
+                pending = {k for k, v in self._dirty.items() if v} | {
+                    k for k, t in self._flushers.items() if t.is_alive()}
+            if names is not None:
+                pending &= set(names)
+            if not pending:
+                return True
+            _time.sleep(0.2)
+        return False
+
+    def _flush_now_at_exit(self):
+        """Process-exit safety net: push any dirty keys synchronously, bypassing the coalescing
+        gap — a Space restart must not eat the last seconds of the async window.
+
+        ⚠ W31-T13 — DELIBERATELY NOT KEY-LOCKED, stated so the omission is not read as one. This
+        runs at interpreter exit with the process going away: a writer mid-`update` will never
+        finish, so waiting for its key lock could only turn a partial save into no save at all.
+        Best-effort is the correct contract here, and it is the one case where taking the lock
+        would be worse than skipping it.
+
+        ⭐ D-305 — IT STILL GOES THROUGH `_commit`, so the last write of a dying process cannot be
+        the one that erases somebody. Exiting is not a licence to overwrite: if the branch has
+        moved, this rebases like any other write, and if it cannot, it gives up rather than
+        clobbering. `_commit`'s own `data_binding` check applies too, so a container that was never
+        allowed to write this store does not get one last unguarded upload on the way out.
+        """
+        with self._lock:
+            dirty = [k for k, v in self._dirty.items() if v]
+            snaps = {k: json.loads(json.dumps(self._cache.get(k, {}))) for k in dirty}
+            replays = {k: list(self._journal.get(k) or ()) for k in dirty}
+            for k in dirty:
+                self._dirty[k] = False
+        for k, d in snaps.items():
+            try:
+                # ⚠ ONE REBASE, NOT `_MAX_REBASE`, and the docstring above is the argument: this
+                # runs at interpreter exit with the process going away. The full budget is up to
+                # four extra downloads and five upload attempts PER DIRTY KEY on a path that may be
+                # killed at any moment, which turns a best-effort save into a hang. One attempt
+                # keeps the safety this method exists for (a conflicting write still rebases rather
+                # than clobbering) at a cost the exit path can actually pay.
+                self._commit(k, d, replays.get(k, ()), max_rebase=1)
+            except Exception:
+                pass          # exiting anyway — nothing left to reschedule
+
+    def update(self, name, fn, flush='sync'):
+        """Read-modify-write. STRICT first read (raises on a transient error rather than
+        starting from an empty dict), apply fn(data)->data, persist.
+
+        flush='sync' (default) = fresh strict read + blocking upload. flush='async' (the
+        table-workspace hot path, wave-7 W3) = the mutation applies to the in-process cache and
+        returns immediately; a coalescing worker commits. The strict read still guards the
+        FIRST async update of a key per process.
+
+        ⛔⛔ D-305 — "LAST-WRITE-WINS IS THE CONTRACT" IS NO LONGER TRUE ACROSS PROCESSES, AND THE
+        SENTENCE BELOW USED TO SAY IT WITHOUT THE QUALIFIER. Two writers in ONE process cannot lose
+        an update — they share `self._cache`, which is why every gate in this repo passed over the
+        defect. Two writers in two CONTAINERS could, and did: measured on tenant #0's live store in
+        commit `dee7481a77`, where one session's write rolled the document back an hour and cost a
+        real user 3 views, 49 typed cell values and a field id. The commit now carries a
+        `parent_commit` precondition and rebases (`_commit`), so the contract is: **within a
+        process, last write wins; across processes, a write that would overwrite unseen work is
+        refused and re-applied.** A lost update is not the contract at either grain.
+
+        ⭐⭐ W31-T13 (D-177(d)) — TWO LOCKS, AND THE SPLIT IS THE WHOLE TICKET.
+        `self._key_lock(name)` is held for the entire read-modify-write, so two writers of one key
+        still cannot interleave.
+        `self._lock` — the one every `get()` takes — is now held only for the microseconds a cache
+        read or write needs, never across a download or an upload. Before this, one sync write
+        froze every reader in the process for two network round trips, on a store whose warm
+        `get()` already costs 703 ms of deep copy on tenant #0.
+        ⚠ Uploading outside `self._lock` is not a new posture: `_flush_worker` has done exactly
+        that since wave 7, and `_repo_ok` racing is benign because `create_repo` is `exist_ok=True`.
+        """
+        # ⛔ D-315 — REFUSED AT THE ACCEPTANCE POINT, NOT AT THE UPLOAD. `_upload` carries the same
+        # check as a backstop, but refusing only there would let the async branch accept the
+        # mutation into the cache, answer 200, mark the key dirty, and hand the flush worker a
+        # write that can never succeed — which retries with backoff forever and reports the refusal
+        # to nobody. Refuse before anything is accepted, and the caller gets a 503 that tells the
+        # truth: no change was saved.
+        data_binding.check_write(self.repo, f'update {name!r}')
+        with self._key_lock(name):
+            return self._update_locked(name, fn, flush)
+
+    def _update_locked(self, name, fn, flush):
+        """`update`'s body, with this key's write lock already held. See its docstring.
+
+        ⚠ THE CACHE LOCK IS TAKEN THREE TIMES AND RELEASED BETWEEN THEM, which is only safe
+        because THIS key's write lock is held throughout: nothing else can be mid-write on it, so
+        the state read in the first section is still ours in the third. A caller reaching this
+        without the key lock would have the old interleaving bug back.
+        """
+        # ⚠ A FLAG, NOT A `None` SENTINEL. `None` can be a real cached value, and a sentinel that
+        # can equal real data is not a sentinel — this repo has paid for that exact shape before
+        # (`routes_nav`'s locked-mode default). The flag says "the cache answered", full stop.
+        data, from_cache = None, False
+        with self._lock:
+            if flush == 'async' and self._owned.get(name) and name in self._cache:
+                data, from_cache = json.loads(json.dumps(self._cache[name])), True
+            elif self._dirty.get(name) and self._owned.get(name) and name in self._cache:
+                # ⛔ A SYNC READ-MODIFY-WRITE MUST NOT DISCARD A PENDING ASYNC WRITE.
+                #
+                # THE BUG THIS LINE EXISTS FOR (owner report 2026-08-09, MEASURED on staging):
+                # `POST /rows` answered `201 {"rid":"100"}` and the row never existed.
+                # `user_tables.add_row` commits `flush='async'` — the row lives ONLY in
+                # `self._cache` until the coalescing worker uploads it, which is `_FLUSH_DELAY`
+                # (2s) away and up to `_FLUSH_MIN_GAP` (20s) if that key committed recently.
+                # Inside that window ANY sync writer of the SAME key took the `else` branch
+                # below, and `_read_strict` does `self._cache[name] = self._download(name)` —
+                # it REPLACES the cache with the hub's copy, erasing the only place the new row
+                # existed, and then `put()` uploaded that. Nothing raised. The 201 was truthful
+                # about `add_row`'s return value and false about the outcome.
+                #
+                # ⚠ `user_tables` has ~40 sync writers against 2 async ones, so this was never
+                # one bad call site — it is a property of the KEY. Wave 27 only made it fire
+                # every time, by putting the relation refresh on the add path.
+                #
+                # ⛔ AND RE-DOWNLOADING WAS PROTECTING NOTHING. Once a key is dirty our cache
+                # has already diverged and the pending upload was going to overwrite the hub
+                # anyway — so the fresh read cannot preserve a remote write, it can only destroy
+                # a local one we already acknowledged. This is the class's OWN documented
+                # contract, restored: see `_owned` above — *"read-your-writes: every reader
+                # consults the cache first"* — and `_flush_worker`'s *"the cache stays the truth
+                # and the dirty mark survives, so nothing is lost silently."*
+                #
+                # ⚠ NOT a weakening of `_read_strict`'s purpose. That exists so a FAILED read
+                # aborts the RMW instead of merging into `{}`; the cache is populated
+                # known-good local state, and this branch is only reachable when it is.
+                data, from_cache = json.loads(json.dumps(self._cache[name])), True
+            # else: the strict hub read happens BELOW, outside this lock (W31-T13)
+
+        # ⭐ W31-T13 — THE HUB READ AND THE CALLER'S `fn` RUN WITH NO CACHE LOCK HELD. `fn` is
+        # arbitrary caller code — `automation_engine`'s relation pass has run whole-tenant work
+        # inside one — and it used to execute with every reader in the process blocked behind it.
+        if not from_cache:
+            data = self._read_strict(name)  # network; re-checks the dirty mark on the way back
+        result = fn(data)
+        data = result if result is not None else data
+
+        # ⭐⭐ D-305 — THE INTENT IS RECORDED, NOT JUST THE RESULT, and this one line is what makes
+        # a conflict survivable. `data` is our answer over OUR base; `fn` is what we MEANT, and
+        # only the intent can be re-applied to somebody else's newer document. The journal holds
+        # every `fn` since the last confirmed upload — which for an async key is a whole flush
+        # window of edits, not one — and `_commit` clears it only when the hub has accepted them.
+        # ⚠ Appended AFTER `fn` ran: a `fn` that raised changed nothing and must not be replayed.
+        with self._lock:
+            replay = self._journal_append(name, fn)
+
+        if flush == 'async':
+            if not self.available():
+                raise RuntimeError('No HF_TOKEN configured — persistence is unavailable.')
+            with self._lock:
+                self._cache[name] = json.loads(json.dumps(data))
+                self._owned[name] = True
+                self._bump(name)  # C6 — the async write is ACCEPTED here, not at flush time
+                self._schedule_flush(name)
+            return data
+
+        # ⛔ THE SYNC COMMIT. It used to be spelled out here rather than calling `put()`, because
+        # `put()` held `self._lock` across its own upload — the freeze W31-T13 removed — and is
+        # PUBLIC with seven callers outside this file whose concurrency semantics changing it would
+        # alter. D-305 made that duplication untenable: an unguarded whole-file write is the hole
+        # the precondition exists to close, so BOTH now go through `_commit`, which is the single
+        # place that resolves a parent, rebases and clears the journal. `put` still keeps its own
+        # body (its op REPLACES rather than merges), but neither can upload unconditionally.
+        if not self.available():
+            raise RuntimeError('No HF_TOKEN configured — persistence is unavailable.')
+        landed, _oid = self._commit(name, data, replay)   # network, no cache lock held
+        with self._lock:
+            # ⚠ `landed`, NOT `data`. A rebase rebuilt the document from the hub's newer copy, so
+            # caching what we TRIED to write would leave the cache diverged from what is stored —
+            # and the next write would rebase off that divergence forever.
+            self._cache[name] = json.loads(json.dumps(landed))
+            self._owned[name] = True
+            self._last_flush[name] = _time.monotonic()
+            # ⛔ AFTER THE UPLOAD, NEVER BEFORE — `put()`'s own law. `_commit` raises on failure,
+            # and a bump on that path would have `revision()` report a change that never landed.
+            self._bump(name)
+            # The upload above CARRIED whatever was pending, so the key is no longer ahead of
+            # the hub. Without this the flush worker wakes on a still-dirty mark and re-uploads
+            # identical content — a second commit against the 256-commits/hr budget for nothing.
+            # ⚠ AFTER the upload, for the same reason: on a raise the mark MUST survive (the cache
+            # is still the truth and the row must still be committed by the worker). Clearing it
+            # first would turn a transient upload error into the very silent loss the branch above
+            # exists to stop — which is why it sits here and not before the commit.
+            self._dirty[name] = False
+        return landed
+
+
+# ------------------------------------------------------------------- instances + module API
+_INSTANCES = {}
+_INSTANCES_LOCK = threading.RLock()
+
+
+def for_repo(repo_id):
+    """The HF Store bound to `repo_id` — one instance per repo per process (each with its own
+    cache and flush state).
+
+    ⚠ WAVE 20: this is now the HF-SPECIFIC factory. Callers that want "the right store for this
+    tenant, whatever the backend is" call `handle()` below. `for_repo` keeps its exact old
+    behaviour because the seed/migration tooling has to be able to name the FILE store
+    explicitly while the app runs on Postgres — a migration that could only reach the active
+    backend could not copy between them.
+    """
+    rid = str(repo_id or '').strip() or REPO
+    with _INSTANCES_LOCK:
+        inst = _INSTANCES.get(rid)
+        if inst is None:
+            inst = Store(rid)
+            _INSTANCES[rid] = inst
+        return inst
+
+
+# =============================================================================================
+# WAVE 20 (owner ruling R1, closes DEBT D-4) — THE BACKEND SEAM.
+#
+# `core/store_backend.py` has carried this warning since EXIT-2b, and it was correct at the time:
+#
+#     "Flipping STORE_BACKEND=pg does NOT redirect the ~40 existing callers that say
+#      `import core.store as store` — they are bound to the HF module directly. Rewiring them is
+#      task C-4 … THE DEFAULT IS `hf` AND STAYS `hf` until C-4 says otherwise."
+#
+# **R1 IS C-4.** The owner ruled the cutover on 2026-08-05 (the D-4 trigger that fired: a SECOND
+# server process — `royal-imports/cfo-os` came back as a pinned LIVE environment beside staging,
+# and two containers on one last-write-wins file store is the race B-3 was always about).
+#
+# THE REWIRE IS HERE RATHER THAN IN 28 FILES, and that is a deliberate choice over the obvious
+# alternative of `sed`-ing every `import core.store as store` to `import core.store_backend`:
+#   * every one of those callers means "the store for the tenant I am serving", which is exactly
+#     what this module has always meant. The BACKEND is not their concern and making it their
+#     concern is how one of them gets missed;
+#   * a missed caller under a search-and-replace does not fail — it silently keeps writing to the
+#     file store while everything else writes to Postgres. That is the split-brain this whole
+#     ruling exists to end, reintroduced by the fix for it;
+#   * `store_backend.py`'s stated reason for refusing to do it this way — "no dual-read window and
+#     no way to compare the two stores' contents first" — is satisfied: `ops/seed_pg_from_hf.py`
+#     copies, then `--verify` diffs the two stores key-by-key before anything flips.
+#
+# ⛔ THE IMPORT IS LAZY AND MUST STAY LAZY. `core/store_pg.py` imports psycopg only inside
+# `_pool()`, so an `hf` deployment installs no driver; importing it at module scope here would
+# undo that and make the default backend depend on the optional dependency.
+# =============================================================================================
+
+def backend():
+    """`'hf'` | `'pg'` — validated, resolved per call so a test can flip the env var.
+
+    Mirrors `core.store_backend.name()` deliberately rather than importing it: that module
+    imports THIS one, so reaching back would be a cycle. `verify_store_pg` asserts the two agree
+    on every value, which is the guard against them drifting apart.
+    """
+    raw = (os.environ.get('STORE_BACKEND') or 'hf').strip().lower()
+    if raw not in ('hf', 'pg'):
+        raise RuntimeError(
+            f"STORE_BACKEND={raw!r} is not a backend. Use 'hf' (the HF Dataset store) or 'pg' "
+            f"(Postgres, needs DATABASE_URL). Refusing to guess: a typo that silently served the "
+            f"other store is how data ends up in two places.")
+    return raw
+
+
+def handle(repo_id=None, slug=None):
+    """THE tenant-bound store handle for the ACTIVE backend — the one seam every caller crosses.
+
+    `repo_id` addresses the HF backend (a dataset repo); `slug` addresses Postgres (a schema).
+    Both are passed by `harness.runtime.get_runtime`, which knows both facts, so flipping the
+    backend never needs a lookup table between them — and neither identifier has to be invented
+    for the backend that does not use it.
+    """
+    if backend() == 'pg':
+        import core.store_pg as _pg          # noqa: PLC0415 — lazy: see the banner above
+        return _pg.PgStore(slug or os.environ.get('AIOS_TENANT') or 'royal-imports')
+    return for_repo(repo_id)
+
+
+#: The HF default instance. Kept as a module global (not a property) because the seed/migration
+#: tooling and `_flush_all_at_exit` both need the FILE store by name even when pg is active.
+_DEFAULT = for_repo(REPO)
+
+
+def _d():
+    """The default tenant's store on the active backend — what every module function below uses.
+
+    ⚠ Resolved PER CALL, never cached. A cached default would freeze the backend at import time,
+    and import order is exactly what nobody controls: `api/main.py` imports half the platform
+    before it has read a single environment variable it did not inherit.
+    """
+    return handle() if backend() == 'pg' else _DEFAULT
+
+
+def available():
+    return _d().available()
+
+
+def get(name, fresh=False):
+    return _d().get(name, fresh=fresh)
+
+
+def _read_strict(name):
+    return _d()._read_strict(name)
+
+
+def exists(name):
+    return _d().exists(name)
+
+
+def upload_bytes(path_in_repo, data, message=None):
+    return _d().upload_bytes(path_in_repo, data, message=message)
+
+
+def download_bytes(path_in_repo):
+    return _d().download_bytes(path_in_repo)
+
+
+def delete_path(path_in_repo):
+    return _d().delete_path(path_in_repo)
+
+
+def put(name, data):
+    return _d().put(name, data)
+
+
+def flush(name=None, timeout=30.0):
+    return _d().flush(name=name, timeout=timeout)
+
+
+def update(name, fn, flush='sync'):
+    return _d().update(name, fn, flush=flush)
+
+
+def get_projection(name, drop=()):
+    """A projected read — see `Store.get_projection`. Wave 32, W32-T01 (D-185).
+
+    ⛔ AN ELEVENTH STORE OPERATION, AND IT HAD TO BECOME ONE — the method alone was NOT enough,
+    which is the opposite of what it looks like from `revision()`'s precedent. `revision` could
+    ship method-only for a few hours because its caller held a Store. This one's caller is
+    `harness.runtime.TenantRuntime`, and `TenantRuntime._store()` returns **the MODULE** whenever
+    `store_handle is None` — which is precisely tenant #0, the tenant whose 703 ms `/nav` copy this
+    exists to remove. A method-only projection would have been unreachable from the only place that
+    needed it, while every gate stayed green.
+
+    ⛔ SO `verify_store_pg.IFACE` GAINS IT IN THE SAME CHANGE, and `core/store_pg.py` gains the
+    twin below it. That gate's `extra:` check would red on this function otherwise — correctly:
+    this addresses a key and reads it, so it is a store operation and both backends must answer.
+    `backend`/`handle` are exempt because they SELECT a store; this one reads one.
+    """
+    return _d().get_projection(name, drop=drop)
+
+
+def revision(name):
+    """The bucket's change token — wave 29, C6. See `Store.revision` for why it copies nothing.
+
+    ⛔ THIS IS A TENTH STORE OPERATION, AND IT WAS ONLY ALLOWED TO BECOME ONE ONCE BOTH BACKENDS
+    HAD IT. `verify_store_pg.py` pins this module's public surface at a contracted list precisely
+    so a capability one backend lacks cannot be reached through the shared door and die at the
+    cutover — so this function, `store_pg.revision`, and that gate's `IFACE` entry are ONE change
+    and must stay one. It shipped for a few hours as a METHOD ONLY, deliberately, while the pg half
+    had no owner: a method is a capability the HF store has, a module function is a promise both
+    backends keep.
+    """
+    return _d().revision(name)
+
+
+def _flush_all_at_exit():
+    with _INSTANCES_LOCK:
+        instances = list(_INSTANCES.values())
+    for inst in instances:
+        inst._flush_now_at_exit()
+
+
+import atexit  # noqa: E402  (registered after the class it needs)
+atexit.register(_flush_all_at_exit)