platform/core/store.py

"""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, RepositoryNotFoundError

REPO = os.environ.get('OS_DATA_REPO', 'royal-imports/cfo-os-data')
_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)

#: 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 _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()}


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 = {}

    def available(self):
        """True when a token is configured — callers degrade gracefully without one."""
        return bool(_token())

    def _download(self, name):
        """Parse {name}.json into a dict. Returns {} ONLY when the file 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)."""
        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)
        except (EntryNotFoundError, RepositoryNotFoundError, FileNotFoundError):
            return {}
        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 {}
            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 = 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
            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 = 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
            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 = 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:
                # Our copy is stale by construction: a pending upload is going to overwrite the
                # hub anyway, so the fresh read cannot preserve a remote write — it can only
                # destroy a local one we have already acknowledged. Same argument, and the same
                # three conditions, as `update`'s pending-async branch below.
                return json.loads(json.dumps(self._cache[name]))
            self._cache[name] = data
            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 _upload(self, name, data):
        """The raw hub write. create_repo is confirmed ONCE per instance — it was a full extra
        API round-trip on every put(), pure latency after the first success."""
        api = HfApi(token=_token())
        if not self._repo_ok:
            api.create_repo(self.repo, repo_type='dataset', private=True, exist_ok=True)
            self._repo_ok = True
        buf = io.BytesIO(json.dumps(data, ensure_ascii=False, indent=2).encode('utf-8'))
        api.upload_file(path_or_fileobj=buf, path_in_repo=f'{name}.json', repo_id=self.repo,
                        repo_type='dataset', commit_message=f'update {name}')

    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.')
        with self._lock:
            api = HfApi(token=_token())
            if not self._repo_ok:
                api.create_repo(self.repo, repo_type='dataset', private=True, exist_ok=True)
                self._repo_ok = True
            api.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}')

    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."""
        if not self.available():
            return False
        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}')
            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."""
        if not self.available():
            raise RuntimeError('No HF_TOKEN configured — persistence is unavailable.')
        with self._lock:
            self._upload(name, data)
            self._cache[name] = json.loads(json.dumps(data))
            self._owned[name] = True
            self._last_flush[name] = _time.monotonic()
            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.
        """
        backoff = 2.0
        while True:
            _time.sleep(_FLUSH_DELAY)
            wait, err, snapshot = 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, {})))
                if wait is None:
                    try:
                        self._upload(name, snapshot)
                        with self._lock:
                            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.
        """
        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}
            for k in dirty:
                self._dirty[k] = False
        for k, d in snaps.items():
            try:
                self._upload(k, d)
            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.

        ⭐⭐ 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 (last-write-wins is the contract; a lost update is not).
        `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`.
        """
        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

        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, SPELLED OUT HERE RATHER THAN THROUGH `put()`, AND THAT IS A CHOICE.
        # `put()` holds `self._lock` across its own `_upload` — which is exactly the freeze this
        # ticket removes — but it is PUBLIC with seven callers outside this file (`users.py`,
        # `modules/ar.py`, `modules/map.py`, `runtime.py`, `store_backend.py`,
        # `ops/seed_pg_from_hf.py`), and changing its locking changes all of their concurrency
        # semantics at once. Duplicating four assignments with this comment beside them is the
        # smaller blast radius; if `put` is ever split, this collapses back into a call.
        if not self.available():
            raise RuntimeError('No HF_TOKEN configured — persistence is unavailable.')
        self._upload(name, data)            # network, no cache lock held
        with self._lock:
            self._cache[name] = json.loads(json.dumps(data))
            self._owned[name] = True
            self._last_flush[name] = _time.monotonic()
            # ⛔ AFTER THE UPLOAD, NEVER BEFORE — `put()`'s own law. `_upload` 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 `_upload`.
            self._dirty[name] = False
        return data


# ------------------------------------------------------------------- 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)