api/routes_keychain.py

"""routes_keychain.py — Keychains + Connectors admin surfaces (wave 18, C7 / R3).

Keychain: encrypted per-tenant credential entries (`core/keychain.py`). The routes NEVER
return a decrypted field — list rows carry a masked preview, and the decrypt function is a
connector-layer internal. Connectors: the tenant's data sources as STATUS rows — Royal's
env-configured Odoo, keychain-held sources — plus R3's guardrail: the **Unsynced records**
count (rows holding overlay data whose pids the current pool no longer serves; counted and
drillable, never silently dropped) and a pause toggle whose v1 semantics are stated honestly
in the payload (`pausedNote`): pausing marks intent and warns; the source cutover ships with
the keychain cutover wave R3 staged.
"""
import os

from fastapi import Body, Depends
from fastapi import APIRouter

from deps import Session, err, require_session
# ⚠ W32-T11 / R4: `admin_gate` is GONE from this module, and its absence is the ruling. Every door
# here was admin-only, which made "a member may hold a personal connection" unbuildable — the wall
# moved from the ROUTE into the ROW (`may_see` / `_may_touch`), where a scope can be enforced per
# entry instead of per endpoint. `session.admin` is still what business-wide requires.

router = APIRouter(prefix="/api/v1")

#: ⭐ D-10 (wave 24): ONE literal, owned by the harness. It was spelled here AND implied by
#: `harness/runtime.py`'s reader; a pause flag written under one spelling and read under another
#: freezes nothing while reporting success, which is the shape of the bug D-10 books.
from harness.runtime import (CONNECTOR_FLAGS_KEY as _CONNECTOR_FLAGS_KEY,  # noqa: E402
                             ENV_ODOO_FLAG_KEY as _ENV_ODOO_FLAG_KEY)
#: DEBT-2 (2026-08-04): the last-successful-sync snapshot bucket. Written when the RESOLVED
#: odoo connector is paused; read by the pool path while paused; survives a Space restart.
SNAPSHOT_KEY = "connector_snapshots"

#: ⛔⛔ W32-T10 / OWNER ITEM 6 — THE TENANT THAT OWNS THE PROCESS ENVIRONMENT.
#:
#: `ODOO_URL` and its siblings are tenant #0's `.env` / Space secrets, and ONE Space process serves
#: EVERY tenant. So "the environment has Odoo credentials" is a fact about this DEPLOYMENT, and
#: turning it into "your workspace is connected to Odoo" is only true for one slug. That is R3's
#: rule, stated in `harness/runtime.py::odoo_source` as *"NEVER the environment: env is tenant #0's
#: connection, and handing it to another tenant is the leak this method exists to prevent."*
#:
#: ⚠ IT IS SPELLED HERE BECAUSE THE RUNTIME EXPORTS NO CONSTANT FOR IT — `odoo_source` and
#: `odoo_flag_key` both carry the literal. `verify_meta`'s W32-T10 section asserts this value
#: against `runtime.py`'s own text, so the day the runtime's answer changes and this one does not,
#: the gate reds instead of the product quietly disagreeing with itself.
ENV_ODOO_TENANT = "royal-imports"


def env_odoo_available(rt):
    """Does the PROCESS ENVIRONMENT offer an Odoo connection to THIS tenant? (W32-T10.)

    ⛔ THE ONE NORMALIZER FOR ONE QUESTION, and it exists because there were two answers to it.
    `routes_connectors.directory._odoo` read a bare `os.environ.get("ODOO_URL")` with **no tenant
    guard** — one process, every tenant — so a nurilab admin opening Connectors was told Odoo was
    `connected` and offered "Manage keys" for a credential belonging to another company. This
    module asked the same question correctly two functions below, which is the whole shape of
    [[one-question-two-normalizers]]: the correct copy hides the wrong one until somebody signs in
    as the second tenant.

    ⚠ NOT the same question as `rt.odoo_flag_key() == ENV_ODOO_FLAG_KEY`. That one answers *which
    source WINS*, so it goes False the moment a keychain entry exists — correct for a pause flag,
    wrong for "should the environment row be listed at all", which is what the connectors pane
    needs in order to show an inactive env source beside an active keychain one.

    Never raises: a runtime that cannot answer is not connected. Fail closed — a missing guard is
    how another tenant's environment got reported as this tenant's connection in the first place.
    """
    try:
        if not (os.environ.get("ODOO_URL") or "").strip():
            return False
        return str(getattr(rt, "key", "") or "") == ENV_ODOO_TENANT
    except Exception:                                             # noqa: BLE001
        return False


# ═════════════════════════════════════════════════════════════════════════════════════════════
# ⭐⭐ W32-T11 / CONTRACT C1 / OWNER RULING R4 — BUSINESS-WIDE vs PERSONAL, ON EVERY CONNECTION
# ═════════════════════════════════════════════════════════════════════════════════════════════
#
# R4, verbatim: *"The business-wide vs personal split lands on ALL connections, not just Odoo.
# Every keychain entry and connector carries a scope, selectable per connection; business-wide is
# admin-only and applies to every user in the tenant."*
#
# ⛔ THE VOCABULARY IS DECLARED HERE AND NOWHERE ELSE (contract C1, and it names this file
# explicitly). `core/keychain.py` is the ENCRYPTED STORE and belongs to the integrator; a scope is
# an access rule, not a secret, so it lives in the layer that already decides who may ask.
# `verify_meta` asserts this tuple against `connectors/ConnectorsPage.tsx`, so the two halves
# cannot drift into two vocabularies.
SCOPES = ("business", "personal")
#: ⚠ THE READ DEFAULT, AND IT IS A READ DEFAULT — never a write (C1). Every entry stored before
#: this wave was, by construction, an admin's tenant-wide credential, so `business` is not a guess.
#: Materialising it would be a migration nobody authorised and would rewrite the store on the next
#: read [[a-migration-that-runs-on-the-next-write]].
DEFAULT_SCOPE = "business"
#: The side bucket: `{entry_id: {"scope": "personal", "owner": "<username>"}}`.
#: ⛔ A BUSINESS ENTRY WRITES NO ROW — it IS the default, so an absent row and a `business` row mean
#: the same thing and there is only one way to spell the common case.
SCOPE_KEY = "keychain_scopes"

#: ⛔⛔ THE TYPES A WHOLE WORKSPACE READS THROUGH, WHICH THEREFORE CANNOT BE PERSONAL.
#:
#: `keychain.odoo_creds` and `keychain.meta_creds` both resolve to *the first entry of that type*
#: for the TENANT — that is what spawns `ut_odoo_*` / `ut_meta_*` and what every measure column is
#: answered from. So a member storing a personal Odoo key would not get "their own Odoo": they
#: would silently become the credential the entire workspace's databases are built from, which is
#: a credential elevation wearing a scope picker.
#: ⚠ REFUSED WITH A REASON, NEVER SILENTLY COERCED TO `business` — a picker that quietly changes
#: your answer is worse than one that says no (W30/R6's second sentence). The resolver itself is
#: `harness/runtime.py` / `core/keychain.py`, the integrator's files; this refusal closes the door
#: from the only side B owns, and the resolver-side guard is booked for A.
TENANT_WIDE_TYPES = ("odoo", "meta_ads")


def clean_scope(raw, default=DEFAULT_SCOPE):
    """A scope word from the wire, or None when the caller said something we do not speak.

    Distinguishing "said nothing" (⇒ the default) from "said nonsense" (⇒ 400) is the whole
    reason this returns None rather than falling back: a typo'd `"personel"` silently becoming
    business-wide is exactly the failure a scope picker exists to prevent.
    """
    if raw is None or (isinstance(raw, str) and not raw.strip()):
        return default
    got = str(raw).strip().lower()
    return got if got in SCOPES else None


def _scope_rows(rt):
    try:
        return dict(rt.get(SCOPE_KEY) or {})
    except Exception:                                             # noqa: BLE001
        return {}


def entry_scope(rt, entry_id, rows=None):
    """`(scope, owner)` for one entry. `rows` is the bucket, passed in when walking a list so a
    census does not re-read the store once per entry."""
    r = (rows if rows is not None else _scope_rows(rt)).get(str(entry_id))
    if not isinstance(r, dict):
        return DEFAULT_SCOPE, ""
    return (str(r.get("scope") or DEFAULT_SCOPE), str(r.get("owner") or ""))


def may_see(scope, owner, uname):
    """R4's visibility rule: business-wide is everyone's, personal is its owner's.

    ⛔ AND AN ADMIN IS NOT AN EXCEPTION. R4 says a personal entry belongs to a person; the
    per-user OAuth slots one module down have made the same call since wave 22 (*"a refresh token
    is identity, not infrastructure"*). An admin who could read every member's personal credential
    would make "personal" a label rather than a boundary.
    """
    return scope != "personal" or str(owner) == str(uname)


def visible_entries(rt, uname, is_admin=False):
    """This USER's view of the keychain: every business entry plus their own personal ones, each
    row carrying its `scope` and `owner` so no caller has to ask a second time.

    ⛔ THE MASKED PREVIEW IS NOT PART OF "VISIBLE". R4 opens this room to members so they can
    hold a connection of their own; it does not hand them four characters of the workspace's Odoo
    key. So a business row a member did not create arrives WITHOUT `preview` — they can see that
    the connection exists and is theirs to use, which is the whole of what R4 grants. The default
    is the RESTRICTED one deliberately: a caller that forgets the argument leaks nothing.
    """
    rows = _scope_rows(rt)
    out = []
    for e in _kc().list_entries(rt):
        scope, owner = entry_scope(rt, e["id"], rows)
        if not owner:
            owner = str(e.get("createdBy") or "")
        if not may_see(scope, owner, uname):
            continue
        row = {**e, "scope": scope, "owner": owner}
        if not (is_admin or str(owner) == str(uname)):
            row["preview"] = ""
        out.append(row)
    return out


def _write_scope(rt, entry_id, scope, owner):
    """Persist one entry's scope. A `business` entry CLEARS its row rather than writing the
    default, so the store holds one spelling of the common case."""
    def _up(cur):
        if scope == DEFAULT_SCOPE:
            cur.pop(str(entry_id), None)
        else:
            cur[str(entry_id)] = {"scope": scope, "owner": str(owner or "")}
        return cur

    rt.update(SCOPE_KEY, _up, flush="sync")
    return True


def _may_touch(session, row):
    """May this session change or delete `row`? An admin owns the business-wide ones; a member
    owns their own personal ones. Anything else is not theirs to move."""
    if row.get("scope") == "personal":
        return str(row.get("owner") or "") == str(session.uname)
    return bool(session.admin)


def _kc():
    import core.keychain as keychain
    return keychain


def _resolved_odoo_key(rt):
    """`(source, flag_key)` — which source would serve this tenant's Odoo queries, in both the
    shapes this module needs: the display string (`env` / `keychain:<id>`) and the key the pause
    flag is stored under.

    ⭐ D-10 (wave 24): THE RESOLUTION ITSELF NOW LIVES IN ONE PLACE, `TenantRuntime.odoo_flag_key`,
    beside the `odoo_source()` it must agree with. This function had its own copy of the same
    three rules — first unlocked keychain odoo entry, else env for tenant #0, else nothing — and
    a second copy is exactly how a pause flag comes to be written against one resolution and read
    against another, freezing nothing while the UI reports success. The two SHAPES stay here
    because they are this module's presentation concern; the DECISION does not.
    """
    flag_key = rt.odoo_flag_key()
    if not flag_key:
        return None, None
    return ("env" if flag_key == _ENV_ODOO_FLAG_KEY else f"keychain:{flag_key}"), flag_key


def odoo_paused(rt):
    """True when the tenant's RESOLVED Odoo source carries the pause flag. Pausing an entry
    that is not the resolved source freezes nothing — it serves nothing.

    ⭐ D-10: a thin delegate now. The implementation moved to `TenantRuntime.odoo_paused` so the
    measure mirror (`harness/datastore.py`, which cannot import this layer) asks the SAME question
    the customer pool does. This name stays because `routes_customers`, `routes_products` and
    `verify_api` all call it — moving the logic without moving the door keeps one answer and
    costs no caller a change.
    """
    return bool(rt.odoo_paused())


def _snap_scope_key(team_id, agent):
    return f"t={team_id}|a={agent}"


def load_pool_snapshot(rt, team_id, agent):
    """(ts, rows) from the persisted snapshot for this exact scope, or None. NEVER a wider
    scope's rows — serving the consolidated snapshot to a scoped user would widen their book."""
    try:
        snap = (rt.get(SNAPSHOT_KEY) or {}).get("odoo_pool") or {}
        e = snap.get(_snap_scope_key(team_id, agent))
        if isinstance(e, dict) and isinstance(e.get("rows"), list):
            return float(e.get("ts") or 0), e["rows"]
    except Exception:
        pass
    return None


def save_pool_snapshots(rt, taken_by=""):
    """Persist every currently-cached pool scope as the pause-time snapshot ('the last
    successful sync', made concrete). Ensures the consolidated default scope exists first so
    a pause on a cold process still captures something to serve."""
    import time as _time
    import routes_customers as _rc
    try:
        _rc._pool_for(rt, None, None)          # the scope every admin/all-BU account lands on
    except Exception:
        pass                                   # cold + Odoo down: persist whatever IS cached
    pools = {}
    for key, entry in list(rt.pool_cache.items()):
        if (isinstance(key, tuple) and len(key) == 3 and key[0] == "pool"
                and isinstance(entry, tuple) and len(entry) == 2
                and isinstance(entry[1], list)):
            pools[_snap_scope_key(key[1], key[2])] = {"ts": entry[0], "rows": entry[1]}
    if not pools:
        return 0

    def _up(cur):
        cur["odoo_pool"] = pools
        cur["taken"] = _time.strftime("%Y-%m-%dT%H:%M:%S")
        cur["takenBy"] = str(taken_by or "")
        return cur

    rt.update(SNAPSHOT_KEY, _up, flush="sync")
    return len(pools)


def _rel_reconnect(rt):
    """Lift the W32-T16 freeze. Its own function so `add_key` and the reconnect route cannot
    disagree about what "resume" means."""
    import odoo_relational as rel

    def _up(cur):
        cur = cur if isinstance(cur, dict) else {}
        cur["frozen"] = False
        cur.pop("frozenAt", None)
        cur.pop("frozenBy", None)
        return cur

    rt.update(rel.CONFIG_KEY, _up, flush="sync")
    return True


def _own_row(session, entry_id):
    """The visible row for `entry_id`, or a 404. ⛔ A 404 rather than a 403 for an entry the
    caller cannot see: telling a member that somebody else's personal credential EXISTS is the
    disclosure the scope is for."""
    row = next((e for e in visible_entries(session.runtime, session.uname,
                                          bool(session.admin))
                if e["id"] == str(entry_id)), None)
    if row is None:
        raise err(404, "no_entry", "no such key")
    return row


@router.get("/admin/keychain")
def list_keychain(session: Session = Depends(require_session)):
    """⭐ W32-T11 / R4 — SESSION-GATED, NOT ADMIN-GATED, and that is the ruling not a relaxation.
    R4 puts a PERSONAL connection in every member's hands, so a room only an admin can open would
    ship the feature and no door to it. The wall moved INTO the payload: a member sees the
    business-wide entries and their own, never anybody else's personal one."""
    kc = _kc()
    return {"entries": visible_entries(session.runtime, session.uname,
                                      bool(session.admin)),
            "locked": not kc.unlocked(),
            #: the vocabulary and the permission, so the client renders a picker it can honour
            #: rather than offering an option the server will refuse (R4: business is admin-only).
            "scopes": list(SCOPES), "canBusiness": bool(session.admin),
            "tenantWideTypes": list(TENANT_WIDE_TYPES)}


@router.post("/admin/keychain", status_code=201)
def add_key(body: dict = Body(default=None), session: Session = Depends(require_session)):
    kc = _kc()
    body = body or {}
    if not session.runtime.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable")
    scope = clean_scope(body.get("scope"))
    if scope is None:
        raise err(400, "bad_scope",
                  f"scope must be one of {', '.join(SCOPES)}")
    if scope == "business" and not session.admin:
        raise err(403, "not_admin",
                  "a business-wide connection applies to everyone in this workspace, so only an "
                  "administrator can create one. You can add it as a personal connection instead.")
    etype = str(body.get("type") or "").strip().lower()
    if scope == "personal" and etype in TENANT_WIDE_TYPES:
        # ⛔ REPORTED, NOT COERCED (W30/R6's second sentence). See TENANT_WIDE_TYPES above: this
        # credential is what the WHOLE workspace's databases are built from, so "personal" would
        # be a label on a tenant-wide key rather than a boundary around it.
        raise err(400, "scope_not_available",
                  f"a {etype} connection is what this whole workspace's databases are read "
                  f"through, so it is always business-wide — it cannot be a personal connection. "
                  f"An administrator can add it for everyone.")
    try:
        row = kc.add_entry(session.runtime, body.get("label"), body.get("type"),
                           body.get("fields"), session.uname)
    except kc.KeychainLocked as e:
        raise err(503, "keychain_locked",
                  f"the keychain is locked — {e}. A secret is never stored unencrypted.")
    except ValueError as e:
        raise err(400, "bad_entry", str(e))
    except Exception:
        raise err(503, "store_unavailable", "the entry was not saved — try again")
    # ⛔⛔ A PERSONAL ENTRY THAT LOSES ITS SCOPE ROW READS AS BUSINESS-WIDE — i.e. the failure mode
    # of a side bucket is to publish a credential, not to hide one. So the second write is not
    # best-effort: if it does not land, the entry is REMOVED and the caller is told nothing was
    # stored. `business` needs no row at all, so this branch is the only one that can be partial.
    # ⭐ W32-T16 / R10's SECOND SENTENCE — *"Reconnecting resumes into the same tables."* Storing
    # an Odoo credential IS reconnecting, so it lifts the freeze here rather than making the admin
    # find a second switch. The tables were never dropped, so "resume" is one flag.
    if etype == "odoo":
        try:
            _rel_reconnect(session.runtime)
        except Exception:                                             # noqa: BLE001
            pass
    if scope != DEFAULT_SCOPE:
        try:
            _write_scope(session.runtime, row["id"], scope, session.uname)
        except Exception:
            try:
                kc.delete_entry(session.runtime, row["id"])
            except Exception:                                     # noqa: BLE001
                pass
            raise err(503, "store_unavailable",
                      "the key was not saved — its sharing setting could not be stored, so "
                      "nothing was kept. Try again.")
    return {"entry": {**row, "scope": scope, "owner": session.uname}}


@router.put("/admin/keychain/{entry_id}")
def update_key(entry_id: str, body: dict = Body(default=None),
               session: Session = Depends(require_session)):
    """Contract C1's scope door. Only the scope moves — a stored secret is never re-openable, so
    "edit this key" means "replace it" and that is `DELETE` + `POST`."""
    row = _own_row(session, entry_id)
    scope = clean_scope((body or {}).get("scope"), default=None)
    if scope is None:
        raise err(400, "bad_scope", f"scope must be one of {', '.join(SCOPES)}")
    if scope == "business" and not session.admin:
        raise err(403, "not_admin",
                  "a business-wide connection applies to everyone in this workspace, so only an "
                  "administrator can make one business-wide.")
    if not _may_touch(session, row):
        raise err(403, "not_yours", "this connection is not yours to change")
    if scope == "personal" and str(row.get("type") or "") in TENANT_WIDE_TYPES:
        raise err(400, "scope_not_available",
                  f"a {row.get('type')} connection is what this whole workspace's databases are "
                  f"read through, so it is always business-wide.")
    owner = row.get("owner") or session.uname
    try:
        _write_scope(session.runtime, entry_id, scope, owner)
    except Exception:
        raise err(503, "store_unavailable", "the change was not saved — try again")
    return {"entry": {**row, "scope": scope, "owner": owner if scope == "personal" else ""}}


@router.delete("/admin/keychain/{entry_id}")
def delete_key(entry_id: str, session: Session = Depends(require_session)):
    row = _own_row(session, entry_id)
    if not _may_touch(session, row):
        raise err(403, "not_yours", "this connection is not yours to delete")
    try:
        _kc().delete_entry(session.runtime, entry_id)
        _write_scope(session.runtime, entry_id, DEFAULT_SCOPE, "")   # drop the side row with it
    except Exception:
        raise err(503, "store_unavailable", "the delete did not land — try again")
    return {"ok": True}


@router.post("/admin/keychain/{entry_id}/test")
def test_key(entry_id: str, session: Session = Depends(require_session)):
    _own_row(session, entry_id)          # 404 for an entry this caller may not see
    return _kc().test_entry(session.runtime, entry_id)


def _unsynced_customer_records(session):
    """R3's guardrail, tenant #0's customer topic: overlay-holding pids the CURRENT pool no
    longer serves. Overlays are unioned across EVERY user of the table (the guardrail is a
    tenant fact, not a per-user one). Honest degradation: when the pool cannot be built the
    answer is `known: False`, never a fabricated zero."""
    try:
        import core.table_store as table_store
        bucket = session.runtime.get("customer_table_workspace") or {}
        overlay_pids = {}
        for uname, ws in bucket.items():
            if uname == table_store.SHARED_KEY or not isinstance(ws, dict):
                continue
            for pid, cells in (ws.get("overlays") or {}).items():
                if isinstance(cells, dict) and cells:
                    overlay_pids.setdefault(str(pid), cells)
        if not overlay_pids:
            return {"known": True, "count": 0, "rows": []}
        from routes_customers import allowed_pids
        pool = {str(p) for p in allowed_pids(session)}
        orphans = sorted((p for p in overlay_pids if p not in pool), key=lambda x: int(x)
                         if str(x).isdigit() else 0)
        rows = []
        for p in orphans[:50]:
            cells = overlay_pids[p]
            hint = next((str(v) for v in cells.values() if str(v).strip()), "")
            rows.append({"pid": int(p) if str(p).isdigit() else p,
                         "fields": len(cells), "hint": hint[:80]})
        return {"known": True, "count": len(orphans), "rows": rows,
                "shown": min(len(orphans), 50)}
    except Exception as e:
        return {"known": False, "count": None, "rows": [],
                "note": f"pool unavailable — {type(e).__name__}"}


@router.get("/admin/connectors")
def connectors(session: Session = Depends(require_session)):
    kc = _kc()
    flags = session.runtime.get(_CONNECTOR_FLAGS_KEY) or {}
    # ⭐ W32-T11 / R4 — THE SAME VISIBILITY RULE AS THE KEYCHAIN, because this pane is the same
    # facts with a status column. Reading `kc.list_entries` here instead would have shown a member
    # every colleague's personal connection on the screen next door to the one that hides them.
    entries = visible_entries(session.runtime, session.uname, bool(session.admin))
    # R3 cutover (2026-08-04): which source would actually serve this tenant's Odoo queries —
    # mirrors TenantRuntime.odoo_source() exactly: first unlocked keychain odoo entry, else env
    # for tenant #0 only, else nothing (fail closed — never another tenant's environment).
    # ⚠ W32-T11: computed over the TENANT's entries, not over `entries` above. "Which source
    # serves this workspace" is one fact for everybody, and deriving it from a per-USER list would
    # make the answer depend on who opened the pane. Personal entries are excluded for the same
    # reason `TENANT_WIDE_TYPES` refuses them: they must never become the workspace's source.
    _scopes = _scope_rows(session.runtime)
    first_odoo = next((e["id"] for e in kc.list_entries(session.runtime)
                       if e["type"] == "odoo"
                       and entry_scope(session.runtime, e["id"], _scopes)[0] != "personal"), None)
    # ⛔ W32-T10 — the env leg goes through `env_odoo_available` now, so this route and the
    # connectors DIRECTORY answer "does the environment serve this tenant?" with one function
    # instead of two spellings that agreed until a second tenant signed in.
    if first_odoo and kc.unlocked():
        resolved = f"keychain:{first_odoo}"
    elif env_odoo_available(session.runtime):
        resolved = "env"
    else:
        resolved = None
    rows = []
    if env_odoo_available(session.runtime):
        rows.append({"key": _ENV_ODOO_FLAG_KEY, "label": "Odoo (environment)", "type": "odoo",
                     "source": "env", "active": resolved == "env",
                     # the deployment's own credential — business-wide by construction, and it
                     # has no owner to be personal to.
                     "scope": DEFAULT_SCOPE, "owner": "",
                     "paused": bool((flags.get(_ENV_ODOO_FLAG_KEY) or {}).get("paused"))})
    for e in entries:
        rows.append({"key": e["id"], "label": e["label"], "type": e["type"],
                     "source": "keychain", "preview": e["preview"],
                     "scope": e.get("scope") or DEFAULT_SCOPE, "owner": e.get("owner") or "",
                     "active": (e["type"] == "odoo" and resolved == f"keychain:{e['id']}"),
                     "paused": bool((flags.get(e["id"]) or {}).get("paused"))})
    out = {"connectors": rows, "locked": not kc.unlocked(), "resolved": resolved,
           "scopes": list(SCOPES), "canBusiness": bool(session.admin),
           # ⭐ D-10 (wave 24) — THIS SENTENCE IS NOW TRUE OF EVERY PATH, which it was not before.
           # DEBT-2 (2026-08-04) froze the CUSTOMER pool and this note honestly disclosed the
           # hole it left: "measures not already computed may still reach the source". D-10
           # closed that hole — `harness/datastore.py` (the mirror every measure column is
           # answered from) refuses to sync while paused, and `routes_products._pool_for` got the
           # guard its customer sibling has had since DEBT-2. So the caveat is deleted rather
           # than left standing, because a warning that outlives its defect teaches the reader to
           # ignore warnings.
           # ⚠ THE THREE BEHAVIOURS ARE NAMED SEPARATELY on purpose: they are genuinely
           # different answers (a persisted snapshot, an in-process cache, a frozen mirror), and
           # collapsing them into "everything freezes" would be the kind of tidy summary that
           # stops being true the first time one of them changes.
           # ⭐ D-62 CLOSED (wave 27) — AND THE REGISTER'S DIAGNOSIS OF IT WAS WRONG, so the
           # correction is recorded here rather than silently applied. D-62 said this note
           # "promises a behaviour on a dashboard measure path that has been dead since W16".
           # MEASURED 2026-08-08, and it is not: measure COLUMNS are grid columns answered from
           # the DuckDB mirror, and `harness/datastore.py` genuinely refuses to sync while
           # paused (`source_paused()` at four sites), so that clause was TRUE. The dead path is
           # `/api/v1/pages/{key}` (D-52), which this note never mentioned.
           #
           # ⛔ THE REAL DEFECT WAS THE OPPOSITE ONE, and it was the last sentence: "so figures
           # stop moving rather than going blank". BOTH pool paths answer **503** when they have
           # no copy to serve — the customer path for a scope with no snapshot
           # (`routes_customers.py:88`) and the product path ALWAYS after a restart, because
           # there is no product snapshot bucket at all (`routes_products.py:60-73`, which says
           # so in as many words). So a paused connector plus a restarted server is exactly the
           # blank screen this sentence promised could not happen. A warning that over-promises
           # is worse than none: it is the sentence somebody quotes when the screen disagrees.
           "pausedNote": ("Pausing a connector never deletes data — notes, custom fields and "
                          "views stay, and nothing reaches the source while it is paused. "
                          "Anything this server has already read keeps showing: the customer "
                          "workspace serves its pause-time snapshot, the product list serves "
                          "the last copy read since startup, and measure columns keep answering "
                          "from the mirror as it stood when you paused. What has NOT been read "
                          "cannot be shown — a scope with no snapshot, or the product list after "
                          "a restart, reports that the source is paused instead of showing "
                          "figures. Resume to start reading live again.")}
    # ⛔⛔ W32-T11 — ADMIN-GATED, AND THIS IS A DISCLOSURE FIX, NOT TIDINESS. Opening this route to
    # members (R4) opened this block with it, and `_unsynced_customer_records` is the one thing on
    # the payload that is NOT about connectors: it unions overlays across EVERY user of the
    # customer table — its own docstring says so, *"the guardrail is a tenant fact, not a per-user
    # one"* — and returns `hint`, the first non-empty cell of somebody else's overlay. So a member
    # would have read colleagues' typed notes off the Connectors pane. It also filters against
    # `allowed_pids(session)`, so a BU-scoped member's narrower pool inflates the orphan count and
    # the number itself becomes wrong for them as well as private.
    # ⚠ The lesson generalises past this line: opening a route widens EVERY field it already
    # returned, and the audit has to walk the payload, not the entry list I was thinking about.
    if session.tenant == "royal-imports" and session.admin:
        out["unsynced"] = _unsynced_customer_records(session)
    return out


# ═════════════════════════════════════════════════════════════════════════════════════════════
# ⭐⭐ W32-T15/T16/T17 / CONTRACT C2 / RULINGS R9, R10, R11 — THE ODOO CONNECTOR ACTUALLY OPENS
# ═════════════════════════════════════════════════════════════════════════════════════════════
#
# The owner clicked "Manage keys" on Odoo and found a credential list. Not: which server database
# this workspace reads, which of the ten mirrored grids it wants, how often they sync, or how to
# stop. Every decision below lives in `odoo_relational` (the module `refresh()` reads) so that a
# switch flipped here is a switch the sync path obeys — a config the route knows and the sync
# path does not is a control that does nothing and reports success.
def _rel():
    import odoo_relational as rel
    return rel


def _odoo_entry(session):
    """The keychain entry SERVING this tenant's Odoo, or None when the environment is (or nothing
    is). Business-scoped by construction — `TENANT_WIDE_TYPES` refuses a personal one."""
    kc = _kc()
    scopes = _scope_rows(session.runtime)
    for e in kc.list_entries(session.runtime):
        if e["type"] == "odoo" and entry_scope(session.runtime, e["id"], scopes)[0] != "personal":
            return e
    return None


def _odoo_source_fields(session, entry):
    """`(serverDb, serverUrl, apiUser, editable)` — what the panel may SHOW about the connection.

    ⛔ NEVER THE SECRET. `read_fields` is documented as the connector layer's internal and no
    route returns its output; this returns the three fields that identify WHICH server, and the
    api key is not among them. The masked preview is the entry's own and was computed at write.
    ⚠ The ENVIRONMENT source is not editable and says so: it is the deployment's `.env`, shared by
    the process, and an admin editing it from a tenant screen would be editing the container.
    """
    if entry is None:
        return (os.environ.get("ODOO_DB", ""), os.environ.get("ODOO_URL", ""),
                os.environ.get("ODOO_USER", ""), False)
    try:
        f = _kc().read_fields(session.runtime, entry["id"]) or {}
    except Exception:                                                  # noqa: BLE001
        return ("", "", "", True)                # locked keychain: honest blanks, still editable
    return (str(f.get("db") or ""), str(f.get("url") or ""), str(f.get("user") or ""), True)


def _odoo_admin(session):
    """C2's doors are admin doors: they show the credential that serves EVERYONE and can turn the
    whole workspace's databases off. R4's personal scope has nothing to say here — a tenant-wide
    type cannot be personal in the first place."""
    if not session.admin:
        raise err(403, "not_admin",
                  "the Odoo connection serves this whole workspace, so only an administrator can "
                  "configure it")


@router.get("/admin/connectors/odoo/config")
def odoo_config(session: Session = Depends(require_session)):
    """Contract C2's read: `{serverDb, grids, syncEvery, canDisconnect}` and the rest of what a
    person needs to see before changing any of it."""
    _odoo_admin(session)
    rel = _rel()
    entry = _odoo_entry(session)
    server_db, server_url, api_user, editable = _odoo_source_fields(session, entry)
    cfg = rel.read_config(session.runtime)
    return {
        "applicable": bool(rel.is_royal(session.tenant)),
        "source": "keychain" if entry else ("env" if env_odoo_available(session.runtime)
                                            else "none"),
        "entryId": (entry or {}).get("id", ""),
        "label": (entry or {}).get("label", "Odoo (environment)"),
        "preview": (entry or {}).get("preview", ""),
        "serverDb": server_db, "serverUrl": server_url, "apiUser": api_user,
        "serverDbEditable": editable,
        "grids": rel.grid_choices(session.runtime),
        "syncEvery": cfg["syncEvery"],
        "syncOptions": list(rel.SYNC_PRESETS),
        "syncFloorSeconds": rel.SYNC_FLOOR_SECONDS,
        "frozen": cfg["frozen"], "frozenAt": cfg["frozenAt"],
        # ⚠ There is nothing to disconnect FROM when the source is the deployment environment:
        # tenant #0's `.env` is not this tenant's to remove. Said as a field so the client renders
        # no button rather than one that 400s.
        "canDisconnect": bool(entry) or (env_odoo_available(session.runtime)
                                         and not cfg["frozen"]),
    }


@router.put("/admin/connectors/odoo/config")
def odoo_config_put(body: dict = Body(default=None),
                    session: Session = Depends(require_session)):
    """Contract C2's write. Grids, cadence and the server database — each optional, each REPORTED
    back rather than silently applied."""
    _odoo_admin(session)
    rel = _rel()
    body = body or {}
    notes = []

    grids = body.get("grids")
    known = {c["key"] for c in rel.grid_choices(session.runtime)}
    clean_grids = None
    if isinstance(grids, dict):
        unknown = sorted(str(k) for k in grids if str(k) not in known)
        if unknown:
            # ⛔ NAMED, NOT DROPPED. A key we do not serve is a client that believes in a grid
            # this connector does not have, and swallowing it makes the two disagree quietly.
            raise err(400, "unknown_grid",
                      f"this connector has no grid called {', '.join(unknown)}")
        clean_grids = {str(k): bool(v) for k, v in grids.items()}
        if clean_grids and not any(clean_grids.get(k, True) for k in known):
            notes.append("every grid is switched off — nothing will be materialised on the next "
                         "sync, and the databases you already have are left untouched")

    every = body.get("syncEvery")
    clean_every = None
    if every is not None:
        clean_every = str(every).strip().lower()
        if clean_every not in rel.SYNC_PRESETS:
            # ⛔⛔ R11 + W30/R6's SECOND SENTENCE: the floor is enforced AND the caller is told.
            # A crafted `"5m"` is CLAMPED to the floor and the response says so — never applied,
            # and never silently ignored either, because a control that discards your answer
            # without a word is how a limit becomes invisible.
            clean_every = rel.DEFAULT_SYNC
            notes.append(f"{every!r} is not an interval this connector offers, and anything under "
                         f"{rel.SYNC_FLOOR_SECONDS // 60} minutes is not available at all — the "
                         f"sync interval was set to the {rel.DEFAULT_SYNC} floor instead")

    server_db = body.get("serverDb")
    if server_db is not None:
        server_db = " ".join(str(server_db).split())[:80]

    def _up(cur):
        cur = cur if isinstance(cur, dict) else {}
        if clean_grids is not None:
            cur.setdefault("grids", {}).update(clean_grids)
        if clean_every is not None:
            cur["syncEvery"] = clean_every
        return cur

    try:
        session.runtime.update(rel.CONFIG_KEY, _up, flush="sync")
    except Exception:
        raise err(503, "store_unavailable", "the change was not saved — try again")

    if server_db:
        notes.append(_rewrite_server_db(session, server_db))

    out = odoo_config(session)
    return {**out, "notes": [n for n in notes if n]}


def _rewrite_server_db(session, server_db):
    """Point the stored Odoo credential at a different server database (R9's first reading).

    ⛔ THERE IS NO "UPDATE ENTRY" IN THE KEYCHAIN, and writing one here would be a SECOND copy of
    how a secret is encrypted and previewed — the thing `core/keychain.py` exists to hold alone.
    So this is add-then-delete through the module's own doors, with the side rows (pause flag,
    scope) carried across because they are keyed by ENTRY ID.
    ⚠ THE ORDER IS DELIBERATE AND THE WINDOW IS REAL: for the moment between the add and the
    delete this tenant has TWO odoo entries, and `odoo_creds` takes the first by id sort — so a
    resync landing inside that window could read the OLD database. The alternative order can
    leave the workspace with no credential at all, which is worse than one stale read. Milliseconds
    of ambiguity beats a lost key.
    """
    kc = _kc()
    entry = _odoo_entry(session)
    if entry is None:
        return ("the server database is set on this deployment's environment, not in the "
                "keychain, so it was not changed here")
    try:
        fields = kc.read_fields(session.runtime, entry["id"]) or {}
    except kc.KeychainLocked as e:
        raise err(503, "keychain_locked", f"the keychain is locked — {e}")
    if not fields:
        raise err(400, "bad_entry", "this credential could not be read back to be changed")
    if str(fields.get("db") or "") == server_db:
        return ""
    fields["db"] = server_db
    try:
        new = kc.add_entry(session.runtime, entry["label"], "odoo", fields, session.uname)
    except Exception:
        raise err(503, "store_unavailable",
                  "the server database was not changed — the existing connection is untouched")
    # carry the side rows across, then retire the old entry
    try:
        flags = session.runtime.get(_CONNECTOR_FLAGS_KEY) or {}
        if entry["id"] in flags:
            def _mv(cur):
                cur[new["id"]] = cur.pop(entry["id"], {})
                return cur
            session.runtime.update(_CONNECTOR_FLAGS_KEY, _mv)
        kc.delete_entry(session.runtime, entry["id"])
        _write_scope(session.runtime, entry["id"], DEFAULT_SCOPE, "")
    except Exception:                                                  # noqa: BLE001
        return (f"the connection now points at {server_db}, but the previous credential could "
                f"not be removed — delete it under Keychains")
    return f"the connection now points at the {server_db} database"


@router.post("/admin/connectors/odoo/disconnect")
def odoo_disconnect(session: Session = Depends(require_session)):
    """R10 — remove the credential and FREEZE the grids as static data.

    ⛔ DISTINCT FROM PAUSE, and the difference is the credential. Pause is temporary and keeps the
    key; disconnect deletes it and marks the databases frozen so nothing refreshes them again —
    including the boot rebuild and the resync loop, which for tenant #0 would otherwise
    re-materialise from the process ENVIRONMENT and quietly undo the disconnect.
    ⛔⛔ AND IT DELETES NOTHING ELSE. The owner's words are *"so we don't fuck up"*: every row and
    every FIELD DEFINITION stays, user-added columns included, because a field a person added is
    the thing a naive freeze drops first. This route never touches `fields` or `rows` — it writes
    one flag in a different bucket, which is what makes that guarantee structural rather than
    careful.
    """
    _odoo_admin(session)
    rel = _rel()
    import datetime as _dt
    entry = _odoo_entry(session)
    if not entry and not env_odoo_available(session.runtime):
        raise err(400, "not_connected", "this workspace has no Odoo connection to disconnect")

    def _up(cur):
        cur = cur if isinstance(cur, dict) else {}
        cur["frozen"] = True
        cur["frozenAt"] = _dt.datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
        cur["frozenBy"] = str(session.uname)
        return cur

    # ⚠ THE FLAG FIRST, THE CREDENTIAL SECOND. If the flag write fails, nothing has happened and
    # the connector is still live; if the delete failed AFTER the flag landed, the tenant is
    # frozen with an unused key, which is recoverable from the Keychains pane. The reverse order
    # can leave a workspace with no key and a connector that still tries to sync.
    try:
        session.runtime.update(rel.CONFIG_KEY, _up, flush="sync")
    except Exception:
        raise err(503, "store_unavailable", "nothing was disconnected — try again")
    removed = ""
    if entry is not None:
        try:
            _kc().delete_entry(session.runtime, entry["id"])
            _write_scope(session.runtime, entry["id"], DEFAULT_SCOPE, "")
            removed = entry["id"]
        except Exception:
            raise err(503, "store_unavailable",
                      "the databases are frozen but the stored credential was not removed — "
                      "delete it under Keychains")
    return {"frozen": True, "removedEntry": removed,
            "note": "Your Odoo databases are frozen: every row and every column you had is still "
                    "there and still readable, and nothing is being refreshed. Reconnecting adds "
                    "the key back and resumes into the same databases."}


@router.post("/admin/connectors/odoo/reconnect")
def odoo_reconnect(session: Session = Depends(require_session)):
    """R10's second sentence — *"Reconnecting resumes into the same tables."*

    It clears the freeze and nothing else: the tables were never dropped, so there is nothing to
    recreate. A tenant whose source was a keychain entry adds it back under Keychains first; this
    is the switch that lets the sync path see it again.
    """
    _odoo_admin(session)
    try:
        _rel_reconnect(session.runtime)
    except Exception:
        raise err(503, "store_unavailable", "the change was not saved — try again")
    connected = bool(_odoo_entry(session)) or env_odoo_available(session.runtime)
    return {"frozen": False, "connected": connected,
            "note": ("Odoo is connected again and the databases you already had will refresh in "
                     "place." if connected else
                     "The freeze is lifted, but there is no Odoo credential yet — add one under "
                     "Keychains and the databases resume into the same tables.")}


@router.post("/admin/connectors/{key}/pause")
def pause_connector(key: str, body: dict = Body(default=None),
                    session: Session = Depends(require_session)):
    paused = bool((body or {}).get("paused"))
    if not session.runtime.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable")
    # ⭐ W32-T11 / R4 — pausing a BUSINESS-WIDE connector stops it for everyone, so it stays an
    # admin act; pausing your own personal one is yours. The env source has no keychain row and
    # is business-wide by construction, hence the admin fallthrough.
    row = next((e for e in visible_entries(session.runtime, session.uname,
                                          bool(session.admin))
                if e["id"] == str(key)), None)
    if row is not None:
        if not _may_touch(session, row):
            raise err(403, "not_yours", "this connection is not yours to pause")
    elif not session.admin:
        raise err(403, "forbidden", "administrators only")

    # DEBT-2: pausing the RESOLVED Odoo source captures the snapshot FIRST, so there is a
    # "last successful sync" to serve before the freeze takes effect. Capturing before the
    # flag flips means a failed capture leaves the connector live (never paused-with-nothing).
    snapshots = 0
    _, flag_key = _resolved_odoo_key(session.runtime)
    if paused and flag_key and str(key) == flag_key:
        snapshots = save_pool_snapshots(session.runtime, taken_by=session.uname)

    def _up(cur):
        cur[str(key)] = {"paused": paused}
        return cur

    try:
        session.runtime.update(_CONNECTOR_FLAGS_KEY, _up)
    except Exception:
        raise err(503, "store_unavailable", "the change was not saved — try again")
    return {"key": key, "paused": paused, "snapshots": snapshots}