platform/harness/runtime.py

"""harness/runtime.py — X7: per-REQUEST tenant resolution for the shared stateless API (EXIT-4a).

THE COST ARGUMENT, in one rule: *no per-tenant state may be resident in the app process.* Today
each tenant carries its own interpreter, its own never-evicted pool, its own 9 prewarmed bundles
and its own DuckDB — EXIT-0 measured a **~0.6 GB commit floor per tenant**, duplicated exactly
(tenant B came within 2% of tenant A; nothing is shared). The target is ~30–80 MB. That is only
reachable if the API resolves the tenant from the REQUEST and holds nothing tenant-shaped between
requests except a bounded cache it is willing to throw away.

WHAT THIS REPLACES. `ui/cache.py`'s `st.cache_resource` singleton — one process, one tenant,
caches that are never evicted by design. That is correct for Streamlit and fatal for a shared
process. The Streamlit side keeps its singleton untouched this wave (two processes, two caches,
accepted for the strangler period; `ui/cache.py` dies with app.py at EXIT-6).

THREE RULES THIS FILE OBEYS, each of which is a way the old shape leaked:
  1. **No module-level per-tenant globals.** Every tenant-shaped value hangs off a TenantRuntime
     reached through `get_runtime(slug)`; nothing is stashed at import time. A module global is
     exactly what makes "which tenant is this process serving?" unanswerable.
  2. **LRU-BOUNDED.** `_MAX_RUNTIMES` runtimes live at once; the least recently used is evicted.
     An unbounded registry cache is a memory leak with a tenant-count-shaped growth curve — the
     thing this whole program exists to remove.
  3. **The store handle is tenant-bound**, so a write cannot land in another tenant's namespace
     even if a caller passes the wrong key downstream (EXIT-4b proof #3).

⚠ THE DUCKDB CAVEAT, STATED NOT HIDDEN. `harness.datastore` holds ONE process-wide connection
(the fix for the 2026-07-16 "connection error unless I refresh" bug), so one PROCESS can serve
one analytical store at a time. This module therefore hands out `datastore_path` — the file a
tenant's analytical reads belong to — and does NOT switch the global underneath a live request.
Until the singleton is per-tenant (EXIT-5), an API process answering measure/Analyst queries must
be pinned to one tenant's file, and `assert_datastore_matches()` below is what turns a
mis-pinned deployment into a loud failure instead of a silent cross-tenant read.
"""
from __future__ import annotations

import os
import threading
from collections import OrderedDict
from dataclasses import dataclass, field
from pathlib import Path

import core.store as store
from harness import datastore as _ds
from harness import tenants as _tenants

#: How many tenant runtimes stay resident. Small on purpose: a runtime is cheap to rebuild (a
#: Tenant literal + a couple of handles) and expensive to leak. Raise only with a measurement.
_MAX_RUNTIMES = int(os.environ.get("AIOS_MAX_RUNTIMES") or 32)

#: slug -> builder. The literal registry `harness/tenants.py` describes; per its own docstring
#: this becomes "a tenants store (DB / HF dataset)" — X4's control DB is where it goes, and
#: `register()` below is the seam that makes the swap a one-liner rather than a refactor.
_BUILDERS = {"royal-imports": _tenants.royal_imports}

# X8's isolation fixture, admitted ONLY on request. A stub tenant that is always present is a
# slug an attacker can mint a session against in production; gated, it exists exactly where it is
# needed (the TestClient isolation proofs) and nowhere else. Read at import so the flag cannot be
# flipped by a request.
if os.environ.get("AIOS_ENABLE_QA_TENANT") == "1":
    _BUILDERS["qa-b"] = _tenants.qa_tenant_b

_CACHE: "OrderedDict[str, TenantRuntime]" = OrderedDict()
_LOCK = threading.RLock()


#: The store bucket Settings > Connectors writes pause flags into. ONE literal, shared with the
#: API layer (`routes_keychain` aliases this) — the flag is written by one module and read by
#: three, and two spellings of the key would be a pause that silently freezes nothing.
CONNECTOR_FLAGS_KEY = "connectors"
#: The flag key tenant #0's compiled ENV Odoo source is stored under. It is not a keychain entry
#: id, so it needs a name of its own; `routes_keychain`'s connector row uses the same literal.
ENV_ODOO_FLAG_KEY = "odoo-env"


class DatastoreMismatch(RuntimeError):
    """The process-wide DuckDB file is not the one this tenant's rows live in (W31-T45 / D-169).

    ⛔ IT SUBCLASSES `RuntimeError` DELIBERATELY, AND THE REASON IS A LIVE TRAP RATHER THAN
    TIDINESS. `harness.datastore.ro_con()` already raises a bare `RuntimeError` for *"the tenant
    data store is completing its FIRST sync"*, and the API doors translate that into
    `503 store_not_ready` — i.e. "wait a few minutes and retry". A cross-tenant mismatch NEVER
    fixes itself by waiting: it is a pinned deployment serving the wrong file, and rendering it as
    a retry banner would convert the loudest refusal in the system into a spinner.

    So the type does two jobs at once. Being a `RuntimeError` keeps every existing
    `except RuntimeError` working unchanged — including `verify_api::_guard_ok`, which is what
    made the guard's own gate green before it had any production caller. Being a SUBCLASS lets
    each door catch this FIRST and answer with its own status and sentence. ⚠ Order matters at
    every catch site: a bare `except RuntimeError` placed above this one swallows it silently,
    which is the exact shape of [[defects-that-mask-each-other]].
    """


def mirror_cursor(rt):
    """THE door from a REQUEST to the analytical mirror: assert the tenant, THEN hand out a cursor.

    ⭐⭐ W31-T45 / D-169. Before this wave there were six independent `datastore.ro_con()` calls
    across `routes_odoo_tables.py` and `odoo_relational.py` and NONE of them asserted anything,
    while `TenantRuntime.assert_datastore_matches` — written for exactly this failure — had no
    caller outside `verify_api.py`. Six call sites is six chances to forget; one door is one.

    ⛔ THE ORDER IS THE POINT, and reversing it is the whole defect this closes. `ro_con()` opens
    (or reuses) a cursor on the process-global `DB_PATH` and answers happily whichever tenant
    asked — so a guard placed AFTER it has already let a cursor onto another tenant's file, and
    any caller holding that cursor from an earlier line is unguarded regardless of what we assert
    later. Assert first, hand out second, and there is no window.

    ⚠ It raises `DatastoreMismatch` (a wrong FILE — never fixes itself) or a plain `RuntimeError`
    (the store is mid-first-sync — retry). Callers must catch the subclass first; see its
    docstring for why that ordering is a correctness rule rather than a style one.
    """
    rt.assert_datastore_matches()
    return _ds.ro_con()


@dataclass
class TenantRuntime:
    """Everything a request needs to serve ONE tenant, and nothing about any other.

    `store_namespace` is the prefix every product-data key is written under. Tenant #0 keeps the
    EMPTY prefix so its existing keys (`users`, `customer_lists`, `customer_docs`, …) are found
    exactly where they already live — a namespacing change that renamed tenant #0's keys would be
    a silent data loss dressed as a refactor. Every OTHER tenant is prefixed, which is what makes
    proof #3 hold.
    """
    key: str
    tenant: object                       # harness.base.Tenant
    store_namespace: str = ""
    pool_cache: dict = field(default_factory=dict)
    #: EXIT-5 — the API adapter's memos for `core.measure_resolve` (the Streamlit adapter keeps
    #: its own in st.session_state). Bounded by the module's own clear-past-cap rule; keyed on
    #: (stamp, scope, pool identity, question) so nothing user-shaped lives in them.
    measure_memo: dict = field(default_factory=dict)
    mset_memo: dict = field(default_factory=dict)
    #: Wave 2026-08-02 (C-TS): the time-series channel's own memo — separate from
    #: measure_memo so column churn and series churn cannot evict each other's answers.
    series_memo: dict = field(default_factory=dict)
    _lock: threading.RLock = field(default_factory=threading.RLock, repr=False)

    @property
    def name(self):
        return getattr(self.tenant, "name", self.key)

    @property
    def datastore_path(self):
        """The DuckDB file this tenant's analytical reads belong to (file-per-tenant IS the
        isolation model — DuckDB has no RLS, so the OS boundary is the only one that holds)."""
        return _ds.path_for(self.key)

    # ---------------------------------------------------------------- the store handle
    #: Wave 18 (C1-TENANT, R2): a tenant that OWNS its dataset repo carries its bound Store
    #: instance here (namespace stays EMPTY — a whole repo needs no prefix). None = the shared
    #: default repo through the MODULE functions, which is what keeps the gates' fake-store
    #: installation effective (the fake patches module attributes, and royal-imports + the
    #: prefix tenants resolve through exactly those).
    store_handle: object = None

    def _store(self):
        return self.store_handle if self.store_handle is not None else store

    def store_key(self, name):
        """Namespace a product-data key for this tenant. The ONE place the prefix is applied."""
        return f"{self.store_namespace}{name}" if self.store_namespace else str(name)

    def get(self, name, fresh=False):
        return self._store().get(self.store_key(name), fresh=fresh)

    def get_projection(self, name, drop=()):
        """A tenant-scoped PROJECTED read — the bucket without the keys named in `drop`.

        ⭐⭐ W32-T01 (D-185). `get()` deep-copies the whole document; on tenant #0's `user_tables`
        that is 28.6 MB / ~703 ms warm, and 99.9% of it is `rows` that no nav render, permission
        check or workspace envelope reads.

        ⛔ **`_store()` returns the MODULE for every tenant with `store_handle is None`** — which
        includes tenant #0 — so this delegates to a MODULE-LEVEL `get_projection`, not to a method.
        Both backends implement it and `verify_store_pg.IFACE` pins that they must; see
        `core.store.get_projection`'s note for why the method alone would have been unreachable
        from exactly the caller it was built for.
        """
        return self._store().get_projection(self.store_key(name), drop=drop)

    def put(self, name, data):
        return self._store().put(self.store_key(name), data)

    def update(self, name, fn, flush="sync"):
        return self._store().update(self.store_key(name), fn, flush=flush)

    def exists(self, name):
        return self._store().exists(self.store_key(name))

    def available(self):
        return self._store().available()

    # ---------------------------------------------------------------- BYTES (wave 25, C's ask)
    # ⭐ WHY THESE EXIST: `grid_events._blobs()` had no tenant-scoped way to reach bytes, so an R2
    # tenant's UPLOADED FILES landed in the shared repo while its buckets went to its own — the
    # residency fix with a hole in it. SESSION C wrote `_blobs` as a CAPABILITY TEST
    # (`hasattr(st, "upload_bytes")`) precisely so the fallback disappears the moment these land,
    # with no edit on that side.
    #
    # ⛔ THEY GO THROUGH `store_key`, AND THAT IS THE WHOLE POINT. A byte path is a different
    # address space from a bucket name, and the temptation is to pass it straight through — but
    # the isolation requirement is identical: a tenant on the SHARED repo (namespace `t/<slug>/`)
    # writing `uploads/photo.png` would otherwise land on the exact path another tenant writes.
    # An own-repo tenant has an EMPTY namespace, so the path is untouched and the repo boundary
    # does the work, which is the same asymmetry every method above already has. Reusing
    # `store_key` keeps "the ONE place the prefix is applied" literally true rather than nearly.
    def upload_bytes(self, path_in_repo, data, message=None):
        return self._store().upload_bytes(self.store_key(path_in_repo), data, message=message)

    def download_bytes(self, path_in_repo):
        return self._store().download_bytes(self.store_key(path_in_repo))

    def delete_path(self, path_in_repo):
        return self._store().delete_path(self.store_key(path_in_repo))

    # ---------------------------------------------------------------- the Odoo source (R3)
    def odoo_source(self):
        """THIS tenant's Odoo connector, or None — the keychain CUTOVER's resolution seam
        (2026-08-04; `keychain.odoo_creds` was previously called by nothing but its test).

        Resolution, fail-closed:
          1. a keychain `odoo` entry (any tenant) → a connector bound to THOSE creds;
          2. tenant #0 with an empty/locked keychain → its compiled env connector (R3's
             "live with env fallback", which is what keeps Royal working unchanged);
          3. anyone else → None. NEVER the environment: env is tenant #0's connection, and
             handing it to another tenant is the leak this method exists to prevent.

        Rebuilt per call on purpose — `core.odoo` caches the CLIENT per (slug, creds
        fingerprint), so this stays cheap while a rotated key reconnects on the next call."""
        import core.keychain as keychain
        try:
            creds = keychain.odoo_creds(self)
        except Exception:            # locked or unreadable keychain — fall through, fail closed
            creds = None
        if creds:
            from harness.connectors.odoo import OdooConnector
            cfg = getattr(self.tenant, "config", {}) or {}
            bus = cfg.get("business_units") or {}
            return OdooConnector(tenant_slug=self.key, creds=creds,
                                 scope={"team_ids": list(bus.values()) or None,
                                        "team_names": {v: k for k, v in bus.items()} or None})
        if self.key == "royal-imports":
            return (getattr(self.tenant, "sources", {}) or {}).get("odoo")
        return None

    # ------------------------------------------------------- the connector PAUSE (D-10, wave 24)
    def odoo_flag_key(self):
        """WHICH connector flag governs this tenant's Odoo — the key `connectors[<k>].paused`
        is stored under, or None when nothing would serve.

        THE SAME RESOLUTION AS `odoo_source()` ABOVE, expressed as an identity instead of a
        client: first unlocked keychain `odoo` entry, else tenant #0's compiled env source, else
        nothing. It lives here so there is ONE resolver — `routes_keychain._resolved_odoo_key`
        was a second copy of these three rules in the API layer, and a pause flag written against
        one resolution and read against the other freezes nothing while reporting success.
        """
        import core.keychain as keychain
        try:
            entries = keychain.list_entries(self)
        except Exception:
            entries = []
        first_odoo = next((e["id"] for e in entries if e.get("type") == "odoo"), None)
        if first_odoo and keychain.unlocked():
            return first_odoo
        if self.key == "royal-imports" and os.environ.get("ODOO_URL"):
            return ENV_ODOO_FLAG_KEY
        return None

    def odoo_paused(self):
        """Is THIS tenant's RESOLVED Odoo source paused?

        ⚠ RESOLVED, not "any entry" — pausing an entry that is not the one serving freezes
        nothing, because it serves nothing. Never raises: an unreachable flags bucket answers
        False, because a store hiccup must not freeze a live surface. That policy is inherited
        from the API-layer function this one replaced, deliberately, so there is one answer to
        "what happens when we cannot tell" rather than two that can disagree.
        """
        try:
            flag_key = self.odoo_flag_key()
            if not flag_key:
                return False
            flags = self.get(CONNECTOR_FLAGS_KEY) or {}
            return bool((flags.get(flag_key) or {}).get("paused"))
        except Exception:
            return False

    def assert_datastore_matches(self):
        """RAISE if the process-wide DuckDB file is not this tenant's.

        The alternative is the failure mode that matters: an analytical read served from another
        tenant's file returns real, plausible, wrong rows — no exception, no telemetry, and the
        number reconciles against the wrong book. A loud failure is strictly better, and this is
        the assertion EXIT-5 deletes when the connection becomes per-tenant.

        ⭐⭐ W31-T45 / D-169 — IT HAS PRODUCTION CALLERS NOW, AND UNTIL THIS WAVE IT HAD NONE.
        Its only caller anywhere was `verify_api.py`, i.e. the guard existed, was correct, was
        tested, and governed nothing — while six mirror reads in `routes_odoo_tables.py` and
        `odoo_relational.py` asserted nothing at all. What held the boundary was not this code but
        a fact about the CUSTOMER LIST ("only tenant #0 has mirror databases"), and R2's Meta Ads
        mirror for GTM Lab ends that fact. Every door now goes through `mirror_cursor()` below.
        ⚠ This raises the priority of the D-29 / D-40 / D-66 tenant-residency family: they are the
        rest of the same boundary, and this guard is one process-global away from them.

        ⭐⭐ THE PREDICATE IS "DOES THE OPEN FILE BELONG TO SOMEBODY ELSE", NOT "IS IT MY CANONICAL
        PATH", and the difference was found by MOUNTING it rather than by reading it. A bare
        `want != have` reads as the stricter, safer rule — and it refuses a case that cannot leak
        anything: a process pinned by `datastore.use_path()` at a store no tenant's naming
        convention produces (a provisioning script, a single-tenant worker, every hermetic gate
        fixture in this repo). Nobody else reads that file, so there is no other book to reconcile
        against. Mounted with the bare rule, `verify_scopes::section_read_through` went from
        320/320 to a 409 on its first leg — the fixture was not wrong, the predicate was.

        So: same path ⇒ fine, and that is the hot path with NO store read. Different path ⇒ ask
        `known_tenants()` who owns the open one. A DIFFERENT tenant is the D-169 failure and
        raises. NOBODY is a bespoke pin and is allowed, deliberately and narrowly.

        ⛔ AND THE UNKNOWN-OWNER BRANCH IS FAIL-CLOSED, which is the half that is easy to get
        backwards. "No known tenant owns this file" and "the tenant registry could not be read"
        produce the same empty answer from a forward map, and they mean opposite things: the first
        is a bespoke pin, the second is every tenant looking unowned — including the one whose
        rows are open. A store blip must not turn this guard off, so an unreadable registry raises.
        """
        want, have = self.datastore_path.resolve(), _ds.DB_PATH.resolve()
        if want == have:
            return
        owner, registry_ok = self._store_owner(have)
        if owner == self.key:
            return
        if owner or not registry_ok:
            whose = (f"which belongs to tenant {owner!r}" if owner else
                     "and the tenant registry could not be read, so ownership is UNKNOWN "
                     "(refusing rather than guessing)")
            raise DatastoreMismatch(
                f"tenant {self.key!r} expects the analytical store at {want}, but this process "
                f"has {have} open, {whose}. Refusing the read — a cross-tenant answer would "
                f"reconcile against the wrong book. (Pin the process with AIOS_DUCKDB_PATH, or "
                f"call harness.datastore.use_path before serving this tenant.)")

    @staticmethod
    def _store_owner(path):
        """`(slug_or_empty, registry_was_readable)` for the DuckDB file at `path`.

        ⛔ FORWARD-MAPPED, never parsed out of the filename — `datastore.path_for` substitutes `_`
        for every non-alphanumeric character and truncates at 60, so `a.b` and `a_b` produce the
        same file and the name is genuinely not invertible. `tenant_for_store_path` above says the
        same thing and is not reused here for one reason: it swallows every failure into `""`, and
        this caller must be able to tell "nobody owns it" from "I could not find out".

        ⚠ AND `known_tenants()` IS NOT REUSED EITHER, FOR THE SAME REASON ONE LAYER DOWN — it
        cannot raise: `_tenant_records()` returns `{}` on any failure and `known_tenants` wraps
        that in another `except: pass`. Calling it here would have made `registry_ok` a constant
        `True` and the fail-closed branch above unreachable, i.e. a guard whose most important
        clause is dead code that reads as if it fires ([[gate-can-report-green-on-nothing]]). The
        control-plane bucket is therefore read HERE, where its failure is still visible.
        """
        keys, registry_ok = set(_BUILDERS), True
        try:
            recs = store.get(TENANTS_KEY) or {}
            keys |= {str(k).strip().lower() for k, v in recs.items() if isinstance(v, dict)}
        except Exception:                                            # noqa: BLE001
            registry_ok = False
        for key in sorted(keys):
            try:
                if _ds.path_for(key).resolve() == path:
                    return key, registry_ok
            except Exception:                                        # noqa: BLE001
                continue
        return "", registry_ok


def register(key, builder):
    """Add a tenant to the registry (a stub connector is a legitimate builder — see X8's qa-b).
    Invalidates any cached runtime for the slug so a re-registration cannot be shadowed."""
    _BUILDERS[str(key)] = builder
    invalidate(key)


def known_tenants():
    keys = set(_BUILDERS)
    try:
        keys |= set(_tenant_records())
    except Exception:
        pass
    return sorted(keys)


#: Wave 18 (C1-TENANT) — the CONTROL-PLANE bucket: tenants provisioned at runtime rather than
#: compiled in. Lives in the DEFAULT store (tenant #0's repo) deliberately: it answers "which
#: tenants exist", which is a platform fact, not any one tenant's data.
TENANTS_KEY = "tenants"


def _tenant_records():
    """{slug: record} from the control-plane bucket. {} on any failure — an unreachable bucket
    must degrade to "only the compiled-in tenants exist", never to an exception at login."""
    try:
        recs = store.get(TENANTS_KEY) or {}
        return {str(k).strip().lower(): v for k, v in recs.items() if isinstance(v, dict)}
    except Exception:
        return {}


def _builder_from_record(key, rec):
    """A Tenant for a bucket-provisioned client: no sources yet (connectors arrive via the
    Keychain flow), config carries the platform fields the request path reads."""
    def _build():
        from harness.base import Tenant
        return Tenant(key=key, name=str(rec.get("name") or key),
                      config={"modules": rec.get("modules", []),
                              "store_repo": rec.get("store_repo") or None})
    return _build


def get_runtime(tenant_key):
    """The runtime for `tenant_key`, built on first use and LRU-cached.

    Raises KeyError for an unknown slug — the caller turns that into a 401 (fail-closed: an
    unknown tenant is not a 404, because confirming which slugs exist answers a question the
    request was not entitled to ask).
    """
    key = str(tenant_key or "").strip().lower()
    if not key:
        raise KeyError("no tenant")
    with _LOCK:
        rt = _CACHE.get(key)
        if rt is not None:
            _CACHE.move_to_end(key)          # LRU: touch on read
            return rt
        builder = _BUILDERS.get(key)
    # Wave 18 (C1-TENANT): a slug the compiled registry does not know may be a PROVISIONED
    # tenant — consult the control-plane bucket before failing closed. A suspended record is
    # the same KeyError as an unknown one: "which tenants exist" is not this caller's question.
    if builder is None:
        rec = _tenant_records().get(key)
        if not rec or rec.get("status", "active") != "active":
            raise KeyError(key)
        builder = _builder_from_record(key, rec)
    # Built OUTSIDE the lock: a builder may do real work (a connector handshake), and holding the
    # registry lock through it would serialise every other tenant's requests behind it.
    tenant = builder()
    with _LOCK:
        existing = _CACHE.get(key)
        if existing is not None:             # another thread won the race — keep ONE runtime
            _CACHE.move_to_end(key)
            return existing
        # R2: a tenant with its OWN repo gets a bound Store instance and an EMPTY prefix (the
        # repo boundary is the namespace); everyone else keeps the t/<slug>/ prefix in the
        # shared repo. Tenant #0 keeps both defaults — empty prefix, shared repo.
        #
        # ⭐ WAVE 20 (R1 / D-4): under Postgres the address is a SCHEMA, so `store.handle()` is
        # asked for one by SLUG and the namespace prefix goes to EMPTY for EVERY tenant —
        # `t_<slug>.store_kv` already isolates, and prefixing keys inside an isolated schema
        # would namespace them twice. That is not a cosmetic difference: a `t/nurilab/`-prefixed
        # key written into `t_nurilab` is a key the migration did not copy and no reader looks
        # for, i.e. a silent empty workspace.
        #
        # ⚠ NOTHING ELSE IN THIS FILE CHANGES, and that is the design working. Every product-data
        # read/write already goes through `rt.get/put/update/exists` (the `st=` handle threaded
        # through the modules), so binding the handle correctly here IS the cutover for the
        # tenant-aware paths — the module-level `core.store` functions cover the rest via `_d()`.
        repo = (getattr(tenant, "config", {}) or {}).get("store_repo")
        if store.backend() == "pg":
            rt = TenantRuntime(key=key, tenant=tenant, store_namespace="",
                               store_handle=store.handle(slug=key))
        else:
            rt = TenantRuntime(
                key=key, tenant=tenant,
                store_namespace=("" if (key == "royal-imports" or repo) else f"t/{key}/"),
                store_handle=(store.for_repo(repo) if repo else None))
        _CACHE[key] = rt
        while len(_CACHE) > _MAX_RUNTIMES:
            _CACHE.popitem(last=False)       # evict least recently used
        return rt


def invalidate(tenant_key=None):
    """Drop one tenant's runtime, or all of them. Explicit, per X7 — a cache with no documented
    way to clear it becomes a restart."""
    with _LOCK:
        if tenant_key is None:
            _CACHE.clear()
        else:
            _CACHE.pop(str(tenant_key or "").strip().lower(), None)


def resident():
    """The tenant slugs currently holding a runtime — LRU order, oldest first. For the EXIT-4
    measurement and for asserting the bound actually binds."""
    with _LOCK:
        return list(_CACHE)


# ─────────────────────────────────────────────────────────────────────────────────────────────
# ⭐ DEBT D-10 (wave 24) — TEACHING THE MEASURE MIRROR ABOUT THE CONNECTOR PAUSE.
#
# `harness.datastore` is the store every measure column and condition is answered from, and it
# pulled from Odoo straight through a pause: 12 sync passes at boot and one every 1800 s. It
# cannot ask the question itself — the flag lives in a tenant's store bucket, which is this
# module's to read, and this module already imports datastore, so an import back would be a
# cycle. So datastore exposes a probe slot and this module fills it, once, at import.


def tenant_for_store_path(path=None):
    """Which tenant owns the DuckDB file currently open (or `path`)? `""` when nothing matches.

    ⛔ RESOLVED BY FORWARD-MAPPING every known tenant, never by parsing the filename, and that is
    a correctness point rather than a style one: `datastore.path_for` substitutes `_` for every
    non-alphanumeric character and truncates at 60, so the name is genuinely NOT invertible —
    `a.b` and `a_b` produce the same file. Guessing the slug back out of it would eventually name
    the wrong tenant, and this answer decides whether another tenant's connector freezes ours.
    """
    want = Path(path or _ds.DB_PATH).resolve()
    # Tenant #0 first: it is the only tenant the unattended sync actually runs for today (see
    # D-29 — the resync loop prewarms `royal-imports` hardcoded), so this is the hot answer.
    for key in ["royal-imports"] + [k for k in known_tenants() if k != "royal-imports"]:
        try:
            if _ds.path_for(key).resolve() == want:
                return key
        except Exception:
            continue
    return ""


def store_source_paused():
    """The probe `harness.datastore` calls before reaching Odoo. Never raises; False when the
    tenant cannot be resolved, matching the fail-towards-live policy `odoo_paused` documents."""
    try:
        key = tenant_for_store_path()
        return bool(key) and get_runtime(key).odoo_paused()
    except Exception:
        return False


_ds.set_paused_probe(store_source_paused)