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