platform/core/store_pg.py

"""core/store_pg.py — `core/store.py`'s interface, backed by Postgres (X4 / EXIT-2b, 2026-07-30).

THE INTERFACE IS THE CONTRACT, and it is `core/store.py`'s nine functions:

    available() · get(name, fresh=False) · exists(name) · put(name, data)
    update(name, fn, flush='sync') · upload_bytes(path, data, message=None)
    download_bytes(path) · delete_path(path) · flush(name=None, timeout=30.0)

Same names, same signatures, same return types, same failure semantics — so switching backends is
an env var (`STORE_BACKEND=hf|pg`) and not a rewrite of every caller. `core/store_backend.py`
does the selection; nothing above it needs to know which store it is talking to.

✅ VERIFIED against a real managed Postgres 2026-08-04 (W19: Neon us-east-2; `verify_store_pg.py`
80/80 INCLUDING the integration half — schema apply, tenant provisioning, jsonb/bytea round-trips,
80 concurrent FOR-UPDATE writes — run from HF egress; the owner's local network resets TLS:5432).
The CUTOVER remains parked behind C1e's triggers and this module is not the default backend.
Without `DATABASE_URL` the gate still SKIPS its integration half LOUDLY, never silently.
The schema is `harness/pg/schema.sql`. Sequenced around the blocker, never stubbed past it.

WHAT POSTGRES FIXES, precisely (C1c — these are the reasons, not a preference):
  * **`get()` is cache-first on the HF store**, so a write from another process is invisible to a
    running app until restart. Observed live in wave 11. Here a read is a SELECT: cross-process
    coherence is the default rather than a thing to remember.
  * **`update()` was read-modify-write with no concurrency control** — two writers raced and the
    loser vanished silently. Here it runs inside ONE transaction with `SELECT … FOR UPDATE`, so
    the second writer waits and then sees the first writer's value.
  * **A write was an HTTP commit** against a 256-commits/hour repo budget, which is why the hot
    path needed a coalescing async flusher at all. A Postgres write is a write; `flush='async'`
    stays in the signature and becomes a no-op, because there is nothing to coalesce.

⚠ ONE DELIBERATE DIFFERENCE FROM THE HF BACKEND, and it is a fix. `store.get()` swallows transient
read failures and hands back a cached-or-empty dict, because a display read must not break a page.
That leniency is exactly what wiped the user registry once (an empty read merged into a write), and
`_read_strict` exists to opt out of it. Postgres has no equivalent "the network blipped" state
worth hiding: a connection failure RAISES from `get()` here. Callers that must not break on a read
should catch it — silently returning `{}` from a database that is simply unreachable is how an
empty dict gets written back over real data.
"""
import json
import os
import threading

_URL_ENV = 'DATABASE_URL'
_LOCK = threading.RLock()
_POOL = {'pool': None, 'url': None, 'schema': None}


def _url():
    return os.environ.get(_URL_ENV) or None


def _schema_for(tenant_slug=None):
    """`t_<slug>` with `-` normalised to `_` (a hyphen is not legal in an unquoted identifier).

    Mirrors `control.provision_tenant_schema` exactly — the two MUST agree or a write lands in a
    schema the DDL never created. `AIOS_TENANT` names the tenant this process serves; it defaults
    to tenant #0 so the default behaviour is unchanged, the same rule `harness.datastore.path_for`
    follows for the DuckDB filename.
    """
    slug = (tenant_slug or os.environ.get('AIOS_TENANT') or 'royal-imports').strip().lower()
    return 't_' + slug.replace('-', '_')


def _pool():
    """The psycopg connection pool, opened once per process.

    psycopg is imported LAZILY and only here: `STORE_BACKEND=hf` must not require the dependency
    to be installed at all, which is what keeps this file safe to commit before B-3 unblocks.
    """
    with _LOCK:
        url = _url()
        if not url:
            raise RuntimeError(
                f'{_URL_ENV} is not set — the Postgres store backend has nothing to connect to.')
        if _POOL['pool'] is not None and _POOL['url'] == url:
            return _POOL['pool']
        try:
            from psycopg_pool import ConnectionPool
        except ImportError as e:                  # pragma: no cover - depends on B-3
            raise RuntimeError(
                'STORE_BACKEND=pg needs psycopg[binary,pool]. It is deliberately NOT in '
                'requirements.txt until the database exists (owner blocker B-3), so that the '
                'default hf backend installs nothing it does not use.') from e
        if _POOL['pool'] is not None:
            try:
                _POOL['pool'].close()
            except Exception:
                pass
        # min_size=0 so a process that never touches the store opens no connection at all — the
        # shared API serves plenty of requests (health, static, a cached payload) that never read.
        _POOL['pool'] = ConnectionPool(url, min_size=0, max_size=8, open=True, timeout=10.0)
        _POOL['url'] = url
        return _POOL['pool']


def available():
    """True when a connection can actually be MADE, not merely when a URL is configured.

    `core.store.available()` only checks for a token, which is all it can cheaply do. Here the
    honest answer needs a round-trip, so the result is memoised per process: callers use this to
    decide whether to degrade the UI, and a `SELECT 1` per render would be absurd.
    """
    if not _url():
        return False
    if _POOL.get('ok'):
        return True
    try:
        with _pool().connection() as con:
            con.execute('SELECT 1')
        _POOL['ok'] = True
        return True
    except Exception:
        return False


def _table(kind, tenant_slug=None):
    """`"t_slug"."store_kv"` — the qualified table, safely quoted."""
    from psycopg import sql
    return sql.SQL('{}.{}').format(sql.Identifier(_schema_for(tenant_slug)),
                                   sql.Identifier(kind))


def _q(template, kind, tenant_slug=None):
    """Compose `template` with `{tbl}` = the QUALIFIED table and `{bare}` = the table name alone.

    ⚠ THE TWO ARE NOT INTERCHANGEABLE, and getting it wrong is a runtime syntax error nothing
    local would catch (no database here — owner blocker B-3). Inside `ON CONFLICT … DO UPDATE`,
    Postgres refers to the conflicting row by the table's ALIAS, which defaults to the bare
    table name: `store_kv.rev` is valid, `t_royal_imports.store_kv.rev` is not. So the FROM
    position takes `{tbl}` and the DO-UPDATE position takes `{bare}`.

    Identifiers go through `psycopg.sql.Identifier`, so the schema name — derived from a tenant
    slug — cannot be an injection vector even though it is interpolated. The slug is ALSO
    regex-constrained at its source (`control.tenants`), which is the belt to this brace.
    """
    from psycopg import sql
    return sql.SQL(template).format(tbl=_table(kind, tenant_slug),
                                    bare=sql.Identifier(kind))


def get(name, fresh=False, tenant_slug=None):
    """The value stored under `name`, or `{}` when the key is genuinely absent.

    `fresh` is accepted and IGNORED: it exists in the HF signature to bypass a process cache, and
    there is no cache here — every read is a SELECT. Keeping the parameter means callers do not
    branch on the backend.
    """
    with _pool().connection() as con:
        row = con.execute(
            _q('SELECT value FROM {tbl} WHERE key = %s', 'store_kv', tenant_slug),
            (str(name),)).fetchone()
    return dict(row[0]) if row and isinstance(row[0], dict) else (row[0] if row else {})


def get_projection(name, drop=(), tenant_slug=None):
    """`core.store.get_projection`'s twin — see `PgStore.get_projection` for why the saving does
    NOT transfer and why that is stated rather than implied.

    ⛔ Present because `verify_store_pg.IFACE` names it, which is the mechanism working as designed:
    a capability only the HF backend had would 500 on `/nav` at the cutover instead of failing here,
    at review, with no database attached.
    """
    import core.store as _hf              # noqa: PLC0415 — lazy: _project lives with the class
    return _hf._project(get(name, tenant_slug=tenant_slug),
                        frozenset(str(d) for d in drop))


def exists(name, tenant_slug=None):
    """True/False — and unlike the HF backend this is NEVER "True on uncertainty".

    `core.store.exists` returns True on any error so a caller that would seed the store cannot
    clobber a registry it merely failed to reach. That guard exists because a FILE HOST cannot
    distinguish "absent" from "unreachable". Postgres can: an absent row is a fact, and an
    unreachable server raises instead of answering.
    """
    with _pool().connection() as con:
        row = con.execute(
            _q('SELECT 1 FROM {tbl} WHERE key = %s', 'store_kv', tenant_slug),
            (str(name),)).fetchone()
    return row is not None


def put(name, data, tenant_slug=None):
    with _pool().connection() as con:
        con.execute(
            _q('INSERT INTO {tbl} (key, value) VALUES (%s, %s::jsonb) '
               'ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, '
               'updated_at = now(), rev = {bare}.rev + 1', 'store_kv', tenant_slug),
            (str(name), json.dumps(data)))
    return data


def update(name, fn, flush='sync', tenant_slug=None):
    """Read-modify-write inside ONE transaction, with the row LOCKED for the duration.

    This is the operation the HF backend could not make safe. There, `update` read over HTTP,
    applied `fn`, and uploaded — so two concurrent writers both read the old value and the second
    upload silently discarded the first writer's change. `SELECT … FOR UPDATE` makes the second
    writer WAIT and then apply `fn` to the first writer's result, which is what "read-modify-write"
    was always supposed to mean.

    `flush` is accepted and ignored (see the module docstring): there is no commit budget to
    coalesce against, so 'async' has nothing to defer.
    """
    with _pool().connection() as con:
        with con.transaction():
            row = con.execute(
                _q('SELECT value FROM {tbl} WHERE key = %s FOR UPDATE',
                   'store_kv', tenant_slug),
                (str(name),)).fetchone()
            data = dict(row[0]) if row and isinstance(row[0], dict) else (row[0] if row else {})
            result = fn(data)
            data = result if result is not None else data
            con.execute(
                _q('INSERT INTO {tbl} (key, value) VALUES (%s, %s::jsonb) '
                   'ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, '
                   'updated_at = now(), rev = {bare}.rev + 1', 'store_kv', tenant_slug),
                (str(name), json.dumps(data)))
    return data


def upload_bytes(path_in_repo, data, message=None, tenant_slug=None):
    with _pool().connection() as con:
        con.execute(
            _q('INSERT INTO {tbl} (path, bytes, message) VALUES (%s, %s, %s) '
               'ON CONFLICT (path) DO UPDATE SET bytes = EXCLUDED.bytes, '
               'message = EXCLUDED.message, updated_at = now()', 'store_blobs', tenant_slug),
            (str(path_in_repo), bytes(data), message))


def download_bytes(path_in_repo, tenant_slug=None):
    """Raw bytes, or None when absent — the same contract as the HF backend: a missing attachment
    degrades to "that file is gone", never to a broken page."""
    with _pool().connection() as con:
        row = con.execute(
            _q('SELECT bytes FROM {tbl} WHERE path = %s', 'store_blobs', tenant_slug),
            (str(path_in_repo),)).fetchone()
    return bytes(row[0]) if row else None


def delete_path(path_in_repo, tenant_slug=None):
    """Absent is SUCCESS — deleting what is already gone is the goal (HF backend's contract)."""
    with _pool().connection() as con:
        con.execute(_q('DELETE FROM {tbl} WHERE path = %s', 'store_blobs', tenant_slug),
                    (str(path_in_repo),))
    return True


def revision(name, tenant_slug=None):
    """`{'rev', 'updated_at', 'token'}` for ONE bucket — wave 29, item 20 / contract C6.

    ⭐ THIS BACKEND HAS MAINTAINED THE ANSWER SINCE X4 AND NEVER PUBLISHED IT. Every `put` and
    `update` above already ends `updated_at = now(), rev = {bare}.rev + 1`; all this adds is the
    read door. Which is why C6 insisted the HF side use these two column NAMES for its in-memory
    counter: the wire shape does not change at the cutover, so the client keeps polling the same
    endpoint and the token simply stops being per-process.

    ⭐ AND HERE THE TOKEN IS DURABLE, which is the real upgrade. `core.store`'s counter lives in
    one process, so it cannot see a write by a second container and it restarts at zero — hence
    the process stamp in its token. `rev` is a column: it survives a restart, and it is shared by
    every process pointed at this schema. The prefix below is therefore a constant rather than a
    per-process value, and a backend flip changes the token exactly once (one refetch, correct).

    An absent key is `rev 0`, not an error: a bucket nobody has written yet has changed zero times,
    which is a fact a client can hold a baseline against.
    """
    with _pool().connection() as con:
        row = con.execute(
            _q('SELECT rev, updated_at FROM {tbl} WHERE key = %s', 'store_kv', tenant_slug),
            (str(name),)).fetchone()
    rev = int(row[0]) if row and row[0] is not None else 0
    stamp = row[1] if row else None
    # ⚠ EPOCH SECONDS ON BOTH BACKENDS. `core.store` records `time.time()`; this column is a
    # `timestamptz`. Same key, two types, is how a consumer that works on one backend breaks on
    # the other — the exact class this module's interface parity exists to prevent.
    at = stamp.timestamp() if hasattr(stamp, 'timestamp') else None
    return {'rev': rev, 'updated_at': at, 'token': f'pg:{rev}'}


def flush(name=None, timeout=30.0):
    """A no-op that returns True: every write above is already committed when it returns. Kept in
    the interface because QA gates and shutdown hooks call it and must not branch on the backend."""
    return True


def close():
    """Release the pool (tests, and a process that is shutting down cleanly)."""
    with _LOCK:
        if _POOL['pool'] is not None:
            try:
                _POOL['pool'].close()
            finally:
                _POOL['pool'] = None
                _POOL['url'] = None
                _POOL.pop('ok', None)


# ---------------------------------------------------------------------------------------------
# WAVE 20 (R1 / D-4) — THE TENANT-BOUND HANDLE. This is what made the cutover a small change.
#
# `core/store.py` grew a CLASS in wave 18 because a tenant can own its own dataset REPO, and
# `harness.runtime.TenantRuntime` carries one bound instance per tenant. Postgres isolates by
# SCHEMA instead (`t_<slug>`, see harness/pg/schema.sql's own argument for schema-over-RLS), so
# the two models meet here: one object, bound to one slug, exposing `core.store.Store`'s methods.
#
# ⚠ THE BINDING IS THE POINT. The module functions above default their schema from `AIOS_TENANT`,
# which is a PROCESS-wide answer to a PER-REQUEST question — correct for a single-tenant worker,
# wrong for the shared API that serves four tenants from one process. A handle carries its slug
# explicitly, so a write cannot land in another tenant's schema because some env var was right
# for the last request. That is the same property `TenantRuntime.store_handle` already gives the
# HF backend (EXIT-4b proof #3), reached a different way.
# ---------------------------------------------------------------------------------------------
class PgStore:
    """`core.store.Store`'s interface for ONE tenant's schema.

    Deliberately a thin binder over the module functions rather than a reimplementation: every
    behaviour argument (FOR UPDATE in `update`, absent-is-success in `delete_path`, `fresh`
    ignored because a SELECT has no cache to bypass) is stated once, up there, and cannot drift
    between the two entry points.
    """

    def __init__(self, tenant_slug):
        self.tenant_slug = str(tenant_slug or '').strip().lower() or 'royal-imports'
        #: Kept so `Store`-shaped debugging (`repr`, telemetry, the admin panes) reads the same
        #: on both backends. There is no repo here; the schema is the address.
        self.repo = f'pg:{_schema_for(self.tenant_slug)}'

    def __repr__(self):
        return f'<PgStore {self.repo}>'

    def available(self):
        return available()

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

    def get_projection(self, name, drop=()):
        """`Store.get_projection`'s contract on Postgres — same ANSWER, different economics.

        ⭐⭐ W32-T01, AND IT IS HERE FOR A REASON THE GATE CANNOT STATE. `get_projection` is a
        METHOD, not a module function, so `verify_store_pg`'s IFACE list does not force it (that
        list pins the ten module-level operations, and a public module function here would fail its
        `extra:` check). But `harness.runtime.TenantRuntime.get_projection` resolves through
        `store.handle()`, which returns THIS class when `STORE_BACKEND=pg` — so a method only the HF
        store had would `AttributeError` at the cutover, on `/nav`, for every tenant at once. That
        is exactly what `revision()`'s docstring means by *"a capability only the HF store had would
        have 500'd at the cutover instead of at review"*, and this is the review.

        ⚠ **The saving does NOT transfer, and pretending otherwise would be the lie.** On HF the
        whole tenant document is one JSON blob in memory and the cost is the 28.6 MB deep copy this
        skips. Here `get` is a SELECT that has already materialised the row, so projecting after the
        fact saves the copy of the dropped keys and nothing on the wire. It is correct, not fast.
        A cheaper pg projection is a `jsonb` column list in the SELECT itself — a real optimisation,
        for the day D-4 actually flips, and not something to build blind against a database that
        does not exist yet (blocker B-3).
        """
        import core.store as _hf          # noqa: PLC0415 — lazy: _project lives with the class
        return _hf._project(get(name, tenant_slug=self.tenant_slug),
                            frozenset(str(d) for d in drop))

    def _read_strict(self, name):
        """Interface parity with the HF Store. There, `get` is lenient (swallows a transient
        failure) and `_read_strict` opts out so a failed read ABORTS a read-modify-write instead
        of merging into `{}`. Here `get` ALREADY raises — the leniency it opts out of does not
        exist — so the strict read is the ordinary one, and saying so beats a second code path."""
        return get(name, tenant_slug=self.tenant_slug)

    def exists(self, name):
        return exists(name, tenant_slug=self.tenant_slug)

    def put(self, name, data):
        return put(name, data, tenant_slug=self.tenant_slug)

    def update(self, name, fn, flush='sync'):
        return update(name, fn, flush=flush, tenant_slug=self.tenant_slug)

    def revision(self, name):
        return revision(name, tenant_slug=self.tenant_slug)

    def upload_bytes(self, path_in_repo, data, message=None):
        return upload_bytes(path_in_repo, data, message=message, tenant_slug=self.tenant_slug)

    def download_bytes(self, path_in_repo):
        return download_bytes(path_in_repo, tenant_slug=self.tenant_slug)

    def delete_path(self, path_in_repo):
        return delete_path(path_in_repo, tenant_slug=self.tenant_slug)

    def flush(self, name=None, timeout=30.0):
        return True

    def _flush_now_at_exit(self):
        """The atexit hook `core.store` registers for its instances. Nothing is buffered here —
        every write above committed before it returned — so this is honestly a no-op rather than
        an unimplemented method that would raise during interpreter shutdown."""
        return True