diff --git "a/platform/harness/datastore.py" "b/platform/harness/datastore.py" --- "a/platform/harness/datastore.py" +++ "b/platform/harness/datastore.py" @@ -1,1042 +1,1133 @@ -"""harness/store.py — the tenant data store (OM-1, 2026-07-11). - -Incremental Odoo → DuckDB mirror: the local analytical store that saved views/dashboards (OM-3) -and the AI Analyst's tools (OM-4) query at warehouse speed, instead of hammering live XML-RPC -per question. -Plan: .claude/wiki/research/omni-adoption.md Part IV (OM-1) + [[odoo-open-source]] BUILD-NOW #1. - -Design rules: - - The store is a RAW mirror (unscoped). Business scope (wholesale teams, GIFTWARE exclusion, - confirmed-only) is applied by the SEMANTIC layer at query time (OM-2) — one source of truth. - - Sync is CHECKPOINTED + BOUNDED + RESUMABLE (loop-library discipline): backfill paginates by - id; live mode advances a write_date cursor; every run is bounded by max_batches and safe to - kill/re-run (upserts are idempotent). - - validate() compares the store against LIVE Odoo (raw fidelity: counts + monthly sums to the - cent) — the store never validates itself. Odoo hard-deletes (unlink) don't move write_date; - the count checks are the drift detector for those. - - Client data: the .duckdb file lives under data/store/ (git-ignored). READ-ONLY on Odoo. -""" -import datetime as dt -import json -import os -import re -import sys -import time -from pathlib import Path - -import duckdb - -import core.odoo as O - -_STORE_DIR = Path(__file__).resolve().parents[1] / "data" / "store" - -# ⚠ MUTABLE ON PURPOSE, and read LATE by every function below (Python resolves a module global at -# CALL time), which is how `provision_tenant.py` already points a fresh instance at its own file. -# EXIT-4a (X7) formalises that: `AIOS_DUCKDB_PATH` lets a container name the file without an edit, -# `path_for()` owns the per-tenant naming convention, and `use_path()` is the SAFE way to switch — -# see its docstring for why a bare assignment is not. -# The DEFAULT is unchanged: /data/store/royal.duckdb. -DB_PATH = Path(os.environ.get("AIOS_DUCKDB_PATH") or (_STORE_DIR / "royal.duckdb")) - - -# ───────────────────────────────────────────────────────────────────────────────────────────── -# ⭐ DEBT D-10 (wave 24) — THE CONNECTOR PAUSE REACHES THE MEASURE MIRROR. -# -# WHAT THE BUG WAS. Pausing a tenant's Odoo connector froze the CUSTOMER POOL (DEBT-2 built that: -# `routes_customers._pool_for` serves the pause-time snapshot) and nothing else. Every measure -# column and condition in the product is answered from THIS store, and this store kept pulling -# from Odoo the entire time a connector was paused — 12 sync passes at boot and another every -# 1800 s, hundreds of `search_read`s each. The Settings copy said so out loud ("measures not -# already computed may still reach the source until their cutover") and that sentence was the -# debt row. -# -# ⛔ WHY A PROBE AND NOT AN IMPORT. The pause flag lives in the tenant's store bucket, which is -# `harness.runtime`'s to read — and `runtime` already imports THIS module (`_ds.DB_PATH`), so an -# import back would be a cycle. The harness installs the probe at its own import instead. Nothing -# else changes: with no probe installed (every gate, every ops script, `sync_runner`) the answer -# is False and this module behaves exactly as it did. -# -# ⚠ IT FAILS TOWARDS "NOT PAUSED", deliberately and consistently with the shipped decision one -# layer up (`routes_keychain.odoo_paused`: "an unreachable flags bucket must never freeze a live -# surface"). The alternative — assume paused when the answer cannot be resolved — turns any -# transient store glitch into a mirror that silently stops advancing, which is the failure mode -# nobody notices for a week. Two copies of that policy would be worse than one; this is the same -# one. -_paused_probe = None - - -def set_paused_probe(fn): - """Install the callable that answers 'is the connector paused for the tenant whose store this - process has open?'. `harness.runtime` installs the real one; pass None to remove it.""" - global _paused_probe - _paused_probe = fn - - -def source_paused(): - """True when this store's tenant has its Odoo connector paused. Never raises.""" - try: - return bool(_paused_probe and _paused_probe()) - except Exception: - return False - - -#: What a sync entry point answers instead of reaching Odoo. A PHASE WORD, not an exception and -#: not a silent zero-row success: `status()` keeps whatever the last real pass wrote, so the -#: mirror still reports the age it genuinely has ([[no-unverifiable-aggregates]] — a paused -#: mirror that reported itself as freshly synced would be the unverifiable claim). -PAUSED_PHASE = "paused" - - -def _paused_notice(what, log): - # Printed as well as logged, and that is deliberate: `api/main.py` passes a swallowing - # `log=lambda *a, **k: None`, so the log-only version of this line would never be seen by - # the one caller that runs it unattended every 30 minutes. - msg = (f"[datastore] {what} skipped - the tenant's Odoo connector is PAUSED; the mirror is " - f"serving its pause-time state. Resume it under Settings > Connectors.") - try: - log(msg) - except Exception: - pass - print(msg) - - -def path_for(tenant_key): - """The canonical DuckDB file for a tenant slug. `royal-imports` keeps the historical - `royal.duckdb` name so tenant #0's existing 175 MB file is not orphaned by a rename. - - File-per-tenant IS the isolation model (C1c): DuckDB has no row-level security, so the only - boundary that holds is the operating system's — one file, one tenant. - """ - key = str(tenant_key or "").strip().lower() - if key in ("", "royal-imports"): - return DB_PATH if key == "" else Path( - os.environ.get("AIOS_DUCKDB_PATH") or (_STORE_DIR / "royal.duckdb")) - safe = "".join(c if (c.isalnum() or c in "-_") else "_" for c in key)[:60] - return _STORE_DIR / f"{safe}.duckdb" - - -def use_path(path): - """Repoint the store at `path`, CLOSING the process singleton first. - - ⚠ A bare `datastore.DB_PATH = …` is not enough and is the more dangerous half of this - operation. `_instance()` caches ONE open connection for the life of the process, so after a - plain reassignment every read keeps being served from the PREVIOUSLY opened file — a silent - cross-tenant read, returning real rows that belong to somebody else. Closing the connection - and clearing the readiness memo (which is also per-file) is what makes the switch honest. - - ⚠ AND CLOSING THE CONNECTION IS STILL NOT ENOUGH ON ITS OWN, because `ro_con()` caches a - cursor in a `threading.local`: this function can only clear the CALLING thread's, and - `ro_con`'s liveness probe (`SELECT 1`) SUCCEEDS on a cursor whose underlying connection was - closed out from under it in some cases — so another thread would keep answering from the old - file with no error to notice. The generation counter below is what closes that: `ro_con` - compares the generation its cursor was opened under and reopens when it has moved, which is - a check no individual thread can forget to do. - - Returns the new path. Callers: `provision_tenant.py`, and any single-tenant worker process. - NOT a per-request operation — one process serves one analytical store at a time, which is why - `harness.runtime` hands out the PATH rather than switching the global underneath a request. - """ - global DB_PATH - with _INSTANCE_LOCK: - con = _INSTANCE.get("con") - if con is not None: - try: - con.close() - except Exception: - pass - _INSTANCE["con"] = None - _RO_TLS.__dict__.pop("cur", None) - _READY["ok"] = False - _GENERATION[0] += 1 # every other thread's cached cursor is now stale - DB_PATH = Path(path) - return DB_PATH - -# Entity specs: Odoo model → store table. m2o fields land as _id + _name. -# 'archivable' entities are pulled with active in [True, False] (the re-SKU/merge rule). -ENTITIES = { - "sale_order": { - "model": "sale.order", "archivable": False, - "fields": ["name", "date_order", "partner_id", "team_id", "user_id", "state", - "amount_untaxed", "invoice_status", "write_date"], - }, - "sale_order_line": { - "model": "sale.order.line", "archivable": False, - "fields": ["order_id", "order_partner_id", "product_id", "product_uom_qty", - "price_subtotal", "margin", "purchase_price", "write_date"], - }, - "res_partner": { - "model": "res.partner", "archivable": True, - # `agent` (bool, labelled "Creditor/Agent") is THE discriminator between a real sales - # AGENT and an internal SALESPERSON who merely carries commission lines — the commission - # table itself cannot tell them apart. `salesman_as_agent` ("convert salesman into agent") - # marks agents who came from staff; the owner ruling on whether those count as agents is - # OUTSTANDING, so both flags are synced and neither reading is baked in. - # ⭐ WAVE 29 (W29-T54, amendment A4). `customer_rank` is the ONLY predicate that - # distinguishes "a partner Odoo has TRANSACTED with" (today's population, 2,465) from - # "a customer" (3,614 with `rank>0 active`) — a 1,149-partner drop, the same join-drop - # class as item 22's SKU pool. It cannot be derived from the document tables, because - # the missing partners are precisely the ones with no document. - # ⛔ It is synced HERE rather than read from live Odoo because `read_customers` also runs - # at BOOT off a hydrated snapshot, where Odoo may be unreachable — sourcing it live would - # make the population "whatever source answered this time", changing size with the network. - # ⭐ WAVE 30 (session E's ask, W30-T33). `street`/`street2`/`zip` are the POSTAL half of - # an address the mirror has never carried: it holds `city`, `state_id` and `country_id` - # and stops there, so `odoo_relational.read_customers` — which reads only the mirror — - # cannot serve a full address at all, however the column is declared. - # ⛔ DECLARING THEM DOES NOT FILL THEM, and the difference is a whole sync: an existing - # mirror ALTERs the columns in on the next `sync_all()` and leaves them NULL until the - # backfill below has walked every partner (a partner's `write_date` did not move because - # WE changed the schema). Every projection degrades a missing/NULL column to blank rather - # than raising, so declaring early is safe — it is just not yet true. - # ⭐⭐ WAVE 33 (W33-T48, owner item 13) — THE CONTACT/IDENTITY HALF, MEASURED BEFORE IT WAS - # ADDED. The owner: *"I could have sworn we have more relevant Fields under 'Odoo vendors' - # etc."* He was right, and the census - # (`.claude/wiki/waves/wave33/odoo-field-census.md`, live read-only, 2026-08-14) says by how - # much: `res.partner` declares 198 fields, **77 are populated on all 3,621 customers and 76 - # on all 658 vendors**, and this mirror carried THIRTEEN. Every one of the six added here - # was measured non-empty on the live population first — none is added on the strength of - # "Odoo probably has that", which is how a schema grows columns nobody can fill. - # ⚠ THE MIRROR IS THE CEILING, NOT THE GRID. Every reader in `odoo_relational.py` reads - # THIS file, so a column added to a `*_fields()` contract with no entry here renders as a - # blank column forever — which is D-165's shape exactly, and the reason the widening starts - # on this line rather than on the grid. - "fields": ["name", "team_id", "city", "state_id", "country_id", "active", - "agent", "salesman_as_agent", "customer_rank", - "street", "street2", "zip", - "email", "phone", "mobile", "website", "vat", "ref", - "write_date"], - # `backfill_fields` carries it too, so mirrors that already exist pick it up on the next - # sync_all() instead of needing a reseed. - # ⛔ D-165 IS ABOUT THIS LIST, AND THE CENSUS SETTLED WHICH HALF IS WRONG: `street`, - # `street2` and `zip` are `store=True` on `res.partner` and POPULATED in live Odoo — they - # are declared here and still NULL in the mirror purely because a backfill pass has not - # walked every partner (changing OUR schema does not move a partner's `write_date`, so an - # incremental sync never revisits them). Declaring is not filling; the fix is a run. - "backfill_fields": ["agent", "salesman_as_agent", # added 2026-07-28 - "customer_rank", # added 2026-08-11 (W29-T54) - "street", "street2", "zip", # added 2026-08-12 (W30-T33) - "email", "phone", "mobile", # added 2026-08-14 (W33-T48) - "website", "vat", "ref"], - # m2m fields land as _id (the FIRST linked id — the Customers-module convention: - # a customer's assigned agent is agent_ids[0], ~one per customer, MECE) + as a - # JSON list of ALL ids. Declared explicitly: a 2-element m2m read would otherwise be - # indistinguishable from an m2o (id, name) pair in _flatten. - "m2m": ["agent_ids"], - }, - "product_product": { - "model": "product.product", "archivable": True, - "fields": ["default_code", "name", "categ_id", "type", "standard_price", "active", - "write_date"], - }, - "account_move": { - "model": "account.move", "archivable": False, - # ⭐ `invoice_origin` added 2026-08-09 (wave 28, DEBT D-88) — the ORDER NAME an invoice was - # raised from ("S12345"), which is the only single stored key that joins an invoice back - # to its order. - # ⛔ WITHOUT IT THAT JOIN IS TWO HOPS, and the second one is expensive: invoice -> - # account_move_line -> `sale_line_ids` (the m2m BU bridge) -> sale_order_line -> - # sale_order, across 963,783 move lines. Wave 27 wanted an order->invoice preset link and - # had no single column to derive it from, so the link was not built at all. - # ⚠ IT IS A NAME, NOT AN ID, and Odoo writes free text into it — a manual invoice can hold - # anything, and a merged one can hold several origins space-separated. So it is a JOIN - # HINT, and whatever consumes it matches against the order names that actually exist - # rather than trusting the string. The m2m route above stays the authority for BU - # attribution; this does not replace it. - "fields": ["name", "move_type", "state", "invoice_date", "invoice_date_due", - "partner_id", "amount_untaxed_signed", "amount_residual_signed", - "payment_state", "invoice_origin", "write_date"], - # Existing rows never re-sync on their own (WE moved the schema; their write_date did - # not), so without this the column ALTERs in and stays NULL forever — indistinguishable - # from "this invoice genuinely has no origin". Declared, re-entrant, checkpointed. - "backfill_fields": ["invoice_origin"], # added 2026-08-09 - }, - "account_move_line": { - "model": "account.move.line", "archivable": False, - # display_type/move_type/parent_state/price_subtotal/product_id added 2026-07-28 so the - # invoice-LINE grain is queryable (topic `invoice_lines`). - # ⚠ display_type values here are 'product', 'cogs', 'payment_term', 'line_note', - # 'line_section'. Commission rows attach to 'cogs' lines as well as 'product' ones - # (96,936 of 147,160 in 2025). Filtering to 'product' is a GRAIN guard: measured - # 2026-07-29, omitting it does not move revenue (non-product lines carry - # price_subtotal = 0) but it inflates commission ROW COUNTS ~80% and would corrupt any - # count-of-lines measure. - "fields": ["move_id", "account_id", "partner_id", "date", "debit", "credit", - "balance", "display_type", "move_type", "parent_state", "price_subtotal", - "product_id", "write_date"], - "backfill_fields": ["display_type", "move_type", "parent_state", "price_subtotal", - "product_id"], # added 2026-07-28 - # the BU bridge: account.move carries NO business unit (every invoice sits on team 1 — - # see [[invoice-bu-attribution]]), so Fisch/Royal on the invoice basis is only reachable - # through the originating sale order. sale_line_id (the FIRST linked sale line) is - # lossless here: ZERO 2025 invoice lines span more than one team (measured). - "m2m": ["sale_line_ids"], - }, - "account_invoice_line_agent": { - # The OCA sale-commission module: one row per (invoice line × agent). This is the SECOND - # agent source — per-line commission attribution — and it names a different population - # than res_partner.agent_ids (the customer-master book). See topic `invoice_lines`. - # ⚠ TWO similarly-named FKs, do not confuse them: - # object_id -> account_move_line (THE grain; join on this) - # invoice_id -> account_move (the parent document) - "model": "account.invoice.line.agent", "archivable": False, - "fields": ["agent_id", "commission_id", "amount", "invoice_id", "object_id", - "invoice_date", "settled", "write_date"], - }, - "account_account": { - # NOTE: no `active` field on account.account in this Odoo version (learned 2026-07-11 — - # a ('active','in',...) domain 500s); treat as non-archivable. - "model": "account.account", "archivable": False, - "fields": ["code", "name", "account_type", "write_date"], - }, -} - -BATCH = 2000 - -# Fields that are genuinely BOOLEAN. This list is load-bearing in TWO places: _cols types the -# column, and _flatten must NOT collapse a real False into NULL. Odoo returns False both for -# "empty" and for "boolean false", so a bool field missing from here silently becomes NULL — and -# for res_partner.agent that would erase the ONE flag separating an AGENT from a SALESPERSON -# (see [[invoice-line-agent-commission]]): every row would read "not an agent" indistinguishably -# from "unknown". -BOOL_FIELDS = ("active", "agent", "salesman_as_agent", "settled", "commission_free") - -# ⭐ WAVE 29 — fields that are genuinely INTEGER COUNTS, and this list exists because its absence -# shipped a silent defect within minutes of `customer_rank` being added to the res_partner spec. -# `_cols` types anything it does not recognise as VARCHAR, so `customer_rank` landed as text, the -# mirror populated correctly, every row looked right — and `odoo_relational`'s own predicate -# `customer_rank > 0` died with `Binder Error: Cannot compare values of type VARCHAR and type -# INTEGER_LITERAL`. ⛔ THE COLUMN WAS PRESENT AND THE DATA WAS CORRECT; only its TYPE was wrong, -# which is why "is the field in the mirror?" answered yes and the feature still could not run. -# ⇒ A new numeric field belongs HERE, exactly as a new boolean belongs in BOOL_FIELDS above. -INT_FIELDS = ("customer_rank", "supplier_rank") - - -import threading as _thr - -_RO_TLS = _thr.local() -_READY = {"ok": False} -_INSTANCE = {"con": None} -_INSTANCE_LOCK = _thr.Lock() -#: Bumped by `use_path()` whenever the store FILE changes. Every per-thread cursor is stamped with -#: the generation it was opened under, so a thread that never called `use_path` still cannot keep -#: reading the previous tenant's file — see `ro_con`. A plain "is the cursor alive?" probe cannot -#: answer that question, which is why this counter exists rather than more probing. -_GENERATION = [0] - - -def mark_ready(): - """In-process signal that the store is fully synced (every entity live). The sync writer calls - this after an all-live pass so a reader never has to (re)discover usability by touching the - file. Belt-and-suspenders alongside the shared-instance model below.""" - _READY["ok"] = True - - -def _instance(): - """THE one process-wide read-write DuckDB connection. Readers take .cursor() off it and the - sync writer uses it too — DuckDB serves concurrent reads + a single writer from ONE instance - via MVCC, so a read NEVER contends with a sync pass for the file lock. - - This is the fix for the 'connection error unless I refresh' bug (2026-07-16): an independent - read-only `duckdb.connect()` FAILS while the writer holds the lock ('Can't open a connection - to same database file with a different configuration than existing connections'); a cursor on - the shared instance does not. Openers must ensure the file is fully present first — ensure_seed - writes atomically (os.replace) and every read path guards on DB_PATH.exists() — so the instance - never binds a partial or absent file. (Trade-off: the app now holds the write lock for its whole - life, so a SEPARATE process — e.g. mcp_server.py — cannot open the same store file concurrently; - in production only the app runs, and that was already true during any sync pass.)""" - con = _INSTANCE["con"] - if con is not None: - return con - with _INSTANCE_LOCK: - if _INSTANCE["con"] is None: - DB_PATH.parent.mkdir(parents=True, exist_ok=True) - _INSTANCE["con"] = duckdb.connect(str(DB_PATH)) - return _INSTANCE["con"] - - -def ro_cursor(): - """A read cursor on the shared instance — the contention-free way to read the store.""" - return _instance().cursor() - - -def ready(): - """True once EVERY entity has completed its backfill (phase 'live'). Gate all store READS on - this: a mid-backfill store would return PARTIAL totals silently — worse than an error. Positive - results cache for the process (backfill never regresses). Reads via a shared-instance cursor, - so it is never mis-reported as 'not ready' just because a background sync pass is mid-flight.""" - if _READY["ok"]: - return True - if not DB_PATH.exists(): - return False - try: - cur = ro_cursor() - try: - phases = {r[0]: r[1] for r in - cur.execute("SELECT entity, phase FROM _sync_state").fetchall()} - finally: - cur.close() - except Exception: - return False - if all(phases.get(k) == "live" for k in ENTITIES): - _READY["ok"] = True - return True - return False - - -def ro_con(): - """Per-thread cached read CURSOR on the shared instance for page/module store reads (the - open cost dominates repeated small queries — the Expenses retrofit measured 7.7s→1.0s). No - lock contention with the sync writer; a cursor that has gone bad is reopened once. REFUSES - until ready() — callers either fall back to live reads (modules) or relay a readable message - (Analyst). - - ⚠ THE GENERATION CHECK IS A CORRECTNESS GUARD, NOT AN OPTIMISATION (EXIT-4a). The liveness - probe below asks "does this cursor still work?", which is a different question from "is this - cursor still pointing at the file we are supposed to be reading?". After `use_path()` switches - tenants, a cursor cached in ANOTHER thread's `threading.local` can still answer `SELECT 1` — - and would then serve that thread real, plausible rows from the PREVIOUS tenant's file, with no - exception anywhere. `use_path` bumps the generation; a cursor opened under an older one is - discarded here, which is the only place every thread is guaranteed to pass through.""" - cur = getattr(_RO_TLS, "cur", None) - if cur is not None and getattr(_RO_TLS, "gen", None) != _GENERATION[0]: - try: - cur.close() - except Exception: - pass - cur, _RO_TLS.cur = None, None # the store moved underneath this thread - if cur is not None: - try: - cur.execute("SELECT 1") # cheap liveness probe - return cur - except Exception: - _RO_TLS.cur = None # stale → reopen below - if not ready(): - raise RuntimeError("the tenant data store is completing its FIRST sync on this " - "deployment — dashboards work meanwhile (live reads); retry " - "store queries in a few minutes") - cur = ro_cursor() - _RO_TLS.cur = cur - _RO_TLS.gen = _GENERATION[0] # stamp WHICH file this cursor belongs to - return cur - - -# ───────────────────────────────────────────────────────────────────────────────────────────── -# ⭐ WAVE 30 / OWNER RULING R6 + R7 — THE WINDOW. Read this before touching `window()`. -# -# R6, verbatim: *"there is no cap in how many data from the API source (as long as its from a -# connected source like Odoo) that can be pulled into the app."* R7 says how: a connected grid -# reads THROUGH this mirror instead of copying rows into the one `user_tables` document, which -# `MAX_ROWS = 60_000` bounds. The mirror is already uncapped — 971,034 GL lines live in this file -# today — so "no cap" is not a bigger number, it is a different mechanism: serve a SLICE and count -# the whole. -# -# ⛔⛔ `total` IS A `SELECT count(*)` OVER THE SAME PREDICATE, NEVER `len(rows)`. A window whose -# count is its own length is a fabricated aggregate that reads as authoritative — the class this -# repo has paid for twice ([[no-unverifiable-aggregates]], [[one-question-two-normalizers]]: a -# display predicate and a fold predicate answering one question differently). The two statements -# below are built from ONE `where` + ONE `params` tuple for exactly that reason; they cannot drift -# apart without deleting a line. -# -# ⛔ AND THE PREDICATE PUSHES DOWN. `where` is compiled by `harness/filter_sql.py` — the same -# compiler the client's filter engine is held in step with — so a filter matching rows outside the -# loaded window still COUNTS them. A caller that filters the returned list instead has silently -# asked "how many of the 200 rows in memory match" about a 971,034-row table. -# -# ⚠ `order_by` IS REQUIRED IN PRACTICE AND DEFAULTED HERE. Two pages of an UNORDERED window are -# not guaranteed to partition the table: DuckDB may legally return a row on page 1 and again on -# page 2, and the user sees a duplicate with no error anywhere. The default is the physical -# rowid-ish `1` only when a caller genuinely has no key; every real caller passes one. -# -# ⚠ TRUST BOUNDARY. `table`, `select`, `where` and `order_by` are SQL we author (a spec row, or -# `filter_sql`'s output over a whitelisted column map). Every VALUE is bound through `params`. -# `table` is additionally shape-checked below — not because a caller is hostile, but because a -# typo'd identifier interpolated into two statements is worth one cheap assertion. - -#: The most rows ONE request may carry back. Not a cap on the data — every row is reachable by -#: paging and `total` always tells the truth about how many there are — but a bound on the memory -#: a single response can cost. R6's second sentence applies: a caller that asks for more is CLAMPED -#: and the clamp is REPORTED (`routes_odoo_tables` turns it into a `limits` entry), never silent. -WINDOW_MAX = 5_000 - -#: What a caller gets when it names no window at all. Matches `semantic.store_rows`' own default so -#: the two windowed readers in this codebase do not disagree about what "a page" means. -WINDOW_DEFAULT = 200 - -_IDENT_OK = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") - - -def window(table=None, select=None, where=None, params=(), order_by=None, - offset=0, limit=WINDOW_DEFAULT, cur=None, from_sql=None): - """One page of a mirror table PLUS the true count of everything the same predicate matches. - - Returns `{"rows": [tuple, ...], "columns": [name, ...], "total": int, "offset": int, - "limit": int, "clamped": bool}` — `rows` is the slice, `total` is the population. - - `select` is the projection expression list ("id, name, amount_untaxed"); `where` is a WHERE - body with `?` placeholders; `params` binds them, and is used by BOTH statements. - - ⚠ `table` and `from_sql` are the SAME argument wearing two trust levels, and they are separate - names on purpose. `table` is an identifier and is SHAPE-CHECKED; `from_sql` is a whole FROM - body — a parenthesised subquery with an alias — which cannot be checked at all. Two of the - eight connected grids need one (`read_customers` and `read_agents` are single statements whose - population is a SQL `UNION`), so the capability has to exist; naming it distinctly means a - reviewer can grep for every caller that hands over raw FROM SQL instead of inferring it. - """ - if bool(table) == bool(from_sql): - raise ValueError("datastore.window: pass exactly one of `table` or `from_sql`") - if from_sql: - name = str(from_sql).strip() - else: - name = str(table or "").strip() - if not _IDENT_OK.match(name): - raise ValueError(f"datastore.window: {table!r} is not a table identifier") - proj = str(select or "").strip() - if not proj: - raise ValueError("datastore.window: a projection is required — there is no implicit *") - try: - offset = max(0, int(offset or 0)) - except (TypeError, ValueError): - offset = 0 - try: - limit = int(limit if limit is not None else WINDOW_DEFAULT) - except (TypeError, ValueError): - limit = WINDOW_DEFAULT - clamped = limit > WINDOW_MAX or limit < 1 - limit = min(max(limit, 1), WINDOW_MAX) - - params = tuple(params or ()) - pred = str(where or "").strip() - tail = f" WHERE {pred}" if pred else "" - order = str(order_by or "").strip() or "1" - cur = cur if cur is not None else ro_con() - - # ⛔ THE COUNT RUNS FIRST AND OVER THE SAME `tail` + `params`. Ordering it first is deliberate: - # if the projection is ever wrong, the caller fails LOUDLY on the rows query rather than - # quietly returning a good count beside a broken page. - total = int(cur.execute(f"SELECT count(*) FROM {name}{tail}", params).fetchone()[0] or 0) - got = cur.execute( - f"SELECT {proj} FROM {name}{tail} ORDER BY {order} LIMIT {int(limit)} OFFSET {int(offset)}", - params).fetchall() - cols = [d[0] for d in (cur.description or [])] - return {"rows": [tuple(r) for r in got], "columns": cols, - "total": total, "offset": offset, "limit": limit, "clamped": clamped} - - -def columns_of(table, cur=None): - """The column names a mirror table actually has, lowercased. - - ⚠ A mirror can be `ready()` and still be MISSING COLUMNS: `ready()` reads entity PHASES, while - `_ensure_columns`/`_backfill_columns` checkpoint separately. Asking before projecting is what - stops a DuckDB `BinderException: Referenced column … not found` reaching a user as a bare 500 - (it already cost one live, which is why `odoo_relational.columns` exists on the app side). - """ - name = str(table or "").strip() - if not _IDENT_OK.match(name): - raise ValueError(f"datastore.columns_of: {table!r} is not a table identifier") - cur = cur if cur is not None else ro_con() - try: - rows = cur.execute(f"PRAGMA table_info('{name}')").fetchall() - except Exception: # noqa: BLE001 - return set() - return {str(r[1]).lower() for r in rows} - - -SEED_DATASET = os.environ.get("STORE_SEED_DATASET", "royal-imports/cfo-os-data") - - -def ensure_seed(): - """Fresh/ephemeral deployment (HF Space disks reset on every rebuild): hydrate the store - from the PRIVATE dataset's seed snapshot — Analyst-ready in ~a minute instead of a 20-40 min - XML-RPC backfill; the auto-sync then closes the gap via write_date cursors. Also restores - the pilot views/routines when absent. Fail-quiet: no seed/token → normal backfill.""" - import shutil - if DB_PATH.exists(): - return False - tok = os.environ.get("HF_TOKEN") - if not tok: - return False - try: - from huggingface_hub import hf_hub_download - DB_PATH.parent.mkdir(parents=True, exist_ok=True) - p = hf_hub_download(SEED_DATASET, "store_seed/royal.duckdb", - repo_type="dataset", token=tok) - # Atomic install: copy to a temp path then os.replace — a reader that guards on - # DB_PATH.exists() must never bind a half-written file into the shared instance. - tmp = DB_PATH.with_name(DB_PATH.name + ".tmp") - shutil.copyfile(p, tmp) - os.replace(tmp, DB_PATH) - for fn in ("views.json", "routines.json"): - tgt = DB_PATH.parent / fn - if not tgt.exists(): - try: - q = hf_hub_download(SEED_DATASET, f"store_seed/{fn}", - repo_type="dataset", token=tok) - shutil.copyfile(q, tgt) - except Exception: - pass - return True - except Exception: - return False - - -def connect(): - """A cursor on the shared instance for the SYNC WRITER (and status/validate). Returned as a - cursor so callers' con.close() frees the cursor without closing the process-wide instance; - one writer cursor is active at a time (the sync thread is sequential), readers use others.""" - cur = _instance().cursor() - cur.execute("""CREATE TABLE IF NOT EXISTS _sync_state ( - entity VARCHAR PRIMARY KEY, phase VARCHAR, last_id BIGINT, - cursor_wd VARCHAR, rows BIGINT, updated_at VARCHAR)""") - return cur - - -def _flatten(rec, fields, m2m=()): - out = {"id": rec["id"]} - for f in m2m: - v = rec.get(f) or [] - base = f.removesuffix("_ids") - out[f"{base}_id"] = v[0] if v else None - out[f] = json.dumps(list(v)) if v else None - for f in fields: - v = rec.get(f) - if isinstance(v, (list, tuple)) and len(v) == 2 and isinstance(v[0], int): - out[f"{f.removesuffix('_id')}_id"] = v[0] - out[f"{f.removesuffix('_id')}_name"] = str(v[1]) - elif isinstance(v, (list, tuple)): - out[f] = json.dumps(v) - elif v is False and f not in BOOL_FIELDS: # Odoo False = NULL for non-bool fields - out[f] = None - else: - out[f] = v - return out - - -def _cols(spec): - cols = ["id BIGINT PRIMARY KEY"] - for f in spec["fields"]: - base = f.removesuffix("_id") - if f.endswith("_id"): - cols += [f"{base}_id BIGINT", f"{base}_name VARCHAR"] - elif f in BOOL_FIELDS: - cols.append(f"{f} BOOLEAN") - elif f in INT_FIELDS: - cols.append(f"{f} BIGINT") - elif f in ("amount_untaxed", "amount_untaxed_signed", "amount_residual_signed", - "price_subtotal", "margin", "purchase_price", "product_uom_qty", - "standard_price", "debit", "credit", "balance", "amount"): - cols.append(f"{f} DOUBLE") - else: - cols.append(f"{f} VARCHAR") - for f in spec.get("m2m") or []: - base = f.removesuffix("_ids") - cols += [f"{base}_id BIGINT", f"{f} VARCHAR"] - return cols - - -def _ensure_table(con, key, spec): - con.execute(f"CREATE TABLE IF NOT EXISTS {key} ({', '.join(_cols(spec))})") - - -def _ensure_columns(con, key, spec): - """Schema evolution: existing stores/seeds predate columns a newer spec introduces — - CREATE TABLE IF NOT EXISTS won't add them. Returns the newly added column names.""" - have = {r[1] for r in con.execute(f"PRAGMA table_info('{key}')").fetchall()} - added = [] - for c in _cols(spec): - name = c.split()[0] - if name not in have: - con.execute(f"ALTER TABLE {key} ADD COLUMN {c}") - added.append(name) - return added - - -def _field_cols(f): - """The store column(s) one spec field materialises into (m2o fields become _id + _name).""" - if f.endswith("_id"): - base = f.removesuffix("_id") - return [f"{base}_id", f"{base}_name"] - return [f] - - -def _backfill_columns(con, key, spec, log=print): - """Scalar/m2o schema evolution — the twin of _backfill_m2m, and for the same reason. - - _ensure_columns ALTERs a new column in, but existing rows NEVER re-sync (WE moved the schema; - their write_date didn't move), so a newly added field stays NULL forever — silently, and NULL - is indistinguishable from a legitimately empty value. That is how a store starts answering - 'no rows match' instead of erroring. Re-pull just the affected fields for every row, - paginated by id, marked durably so a crash retries rather than declaring completion. - - The fields are DECLARED in the spec (`backfill_fields`), never inferred from "which columns - did _ensure_columns just add" — that inference is wrong on the second run, when the ALTER has - already happened and the list comes back empty while the data is still missing. Declaring it - makes the migration re-entrant and reviewable. Leave the entry in place after it completes; - the durable marker, not the absence of the declaration, is what stops it re-running. - """ - fields = [f for f in (spec.get("backfill_fields") or []) if f != "write_date"] - if not fields: - return - marker = f"{key}.cols.{','.join(sorted(fields))}" - st = con.execute("SELECT phase, last_id, rows FROM _sync_state WHERE entity=?", - [marker]).fetchone() - if st and st[0] == "done": - return - # RESUMABLE, not restart-from-zero: these passes run to ~1M rows, and a migration that loses - # an hour of work to one dropped connection gets skipped by the next person under time - # pressure. Progress is checkpointed every page; 'done' is written only at the end, so an - # interrupted run resumes at the last committed id instead of re-pulling or, worse, declaring - # completion it never reached. - last, n = (st[1] or 0, st[2] or 0) if st else (0, 0) - if last: - log(f" {key}: column backfill resuming at id {last:,} ({n:,} rows already done)") - cols = [c for f in fields for c in _field_cols(f)] - setter = ", ".join(f"{c}=?" for c in cols) - while True: - recs = O.search_read(spec["model"], _base_domain(spec) + [("id", ">", last)], - fields, limit=5000, order="id asc") - if not recs: - break - rows = [_flatten(r, fields) for r in recs] - con.executemany(f"UPDATE {key} SET {setter} WHERE id=?", - [[r.get(c) for c in cols] + [r["id"]] for r in rows]) - last, n = recs[-1]["id"], n + len(recs) - con.execute("INSERT OR REPLACE INTO _sync_state VALUES (?,?,?,?,?,?)", - [marker, "running", last, None, n, time.strftime("%Y-%m-%d %H:%M:%S")]) - log(f" {key}: column backfill {'+'.join(fields)} -> id {last:,} ({n:,} rows)") - con.execute("INSERT OR REPLACE INTO _sync_state VALUES (?,?,?,?,?,?)", - [marker, "done", last, None, n, time.strftime("%Y-%m-%d %H:%M:%S")]) - log(f" {key}: column backfill complete -> {n:,} rows") - - -def _backfill_m2m(con, key, spec, log=print): - """One-time targeted backfill after a schema migration: existing rows never re-sync (their - write_date didn't move when WE added a column), so pull every record where the m2m field is - SET and update in place. Rows with the field empty keep NULL — correct and free. Completion - is a durable _sync_state marker ('.m2m.') — NOT 'column just added': a crash - between the ALTER and this backfill (seen live: the SSL-failed first run) must retry.""" - for f in spec.get("m2m") or []: - marker = f"{key}.m2m.{f}" - if con.execute("SELECT 1 FROM _sync_state WHERE entity=?", [marker]).fetchone(): - continue - base = f.removesuffix("_ids") - # PAGINATED by id. A single limit=100000 call silently TRUNCATED here: account_move_line - # has 228,067 rows with sale_line_ids set, so 56% of the BU bridge would have gone - # missing with no error — a partial backfill that reads as a complete one (the - # [[no-unverifiable-aggregates]] no-silent-caps rule). - rows, last, page = [], 0, 20000 - while True: - recs = O.search_read(spec["model"], - _base_domain(spec) + [(f, "!=", False), ("id", ">", last)], - [f], limit=page, order="id asc") - if not recs: - break - rows += [(r[f][0], json.dumps(list(r[f])), r["id"]) for r in recs if r.get(f)] - last = recs[-1]["id"] - log(f" {key}: m2m {f} scanned to id {last:,} ({len(rows):,} rows)") - if rows: - con.executemany(f"UPDATE {key} SET {base}_id=?, {f}=? WHERE id=?", rows) - con.execute("INSERT OR REPLACE INTO _sync_state VALUES (?,?,?,?,?,?)", - [marker, "done", 0, None, len(rows), time.strftime("%Y-%m-%d %H:%M:%S")]) - log(f" {key}: m2m backfill {f} -> {len(rows):,} rows updated") - - -def _upsert(con, key, spec, recs): - if not recs: - return - rows = [_flatten(r, spec["fields"], spec.get("m2m") or ()) for r in recs] - colnames = [c.split()[0] for c in _cols(spec)] - ids = [r["id"] for r in rows] - con.execute(f"DELETE FROM {key} WHERE id IN ({','.join(map(str, ids))})") - con.executemany( - f"INSERT INTO {key} ({', '.join(colnames)}) VALUES ({', '.join('?' for _ in colnames)})", - [[r.get(c) for c in colnames] for r in rows]) - - -def _state(con, key): - row = con.execute("SELECT phase, last_id, cursor_wd, rows FROM _sync_state WHERE entity=?", - [key]).fetchone() - return {"phase": row[0], "last_id": row[1], "cursor_wd": row[2], "rows": row[3]} if row else None - - -def _save_state(con, key, phase, last_id, cursor_wd): - n = con.execute(f"SELECT count(*) FROM {key}").fetchone()[0] - con.execute("""INSERT OR REPLACE INTO _sync_state VALUES (?,?,?,?,?,?)""", - [key, phase, last_id, cursor_wd, n, time.strftime("%Y-%m-%d %H:%M:%S")]) - - -def _base_domain(spec): - return [("active", "in", [True, False])] if spec["archivable"] else [] - - -def sync_entity(key, max_batches=200, batch=BATCH, log=print): - """One bounded, resumable sync pass for an entity. Backfill (by id) → live (by write_date). - Returns (phase, pulled_rows). - - ⭐ D-10: refuses BEFORE opening a cursor or touching Odoo while the connector is paused. The - guard is here — the innermost function that reads from `O` — rather than only in `sync_all`, - so a caller that syncs ONE entity is covered by construction instead of by remembering to - add the same check. `sync_all` short-circuits too, but only to avoid eight identical notices. - """ - if source_paused(): - _paused_notice(f"sync of {key}", log) - return PAUSED_PHASE, 0 - spec = ENTITIES[key] - con = connect() - _ensure_table(con, key, spec) - _ensure_columns(con, key, spec) - _backfill_columns(con, key, spec, log=log) - _backfill_m2m(con, key, spec, log=log) - st = _state(con, key) or {"phase": "backfill", "last_id": 0, "cursor_wd": None, "rows": 0} - fields = spec["fields"] + list(spec.get("m2m") or []) - pulled = 0 - try: - if st["phase"] == "backfill": - last_id = st["last_id"] or 0 - for i in range(max_batches): - dom = _base_domain(spec) + [("id", ">", last_id)] - recs = O.search_read(spec["model"], dom, ["id"] + fields, - limit=batch, order="id asc") - if not recs: - # backfill complete → switch to live mode anchored at max write_date seen - wd = con.execute(f"SELECT max(write_date) FROM {key}").fetchone()[0] - _save_state(con, key, "live", last_id, wd or "1970-01-01 00:00:00") - log(f" {key}: BACKFILL COMPLETE ({st['rows'] + pulled:,} rows) -> live mode") - return "live", pulled - _upsert(con, key, spec, recs) - last_id = recs[-1]["id"] - pulled += len(recs) - _save_state(con, key, "backfill", last_id, None) - if (i + 1) % 10 == 0: - log(f" {key}: backfill …{pulled:,} rows (id≤{last_id})") - log(f" {key}: backfill PAUSED at {pulled:,} rows this run (bounded; resumes)") - return "backfill", pulled - # live mode: write_date >= cursor (idempotent overlap; upsert dedupes). MASS-STAMP - # FALLBACK (found live 2026-07-17): a batch job can stamp >batch rows with ONE - # write_date second (an Odoo recompute stamped ~88k account_move rows '2026-07-15 - # 16:35:03'; stored values carry microseconds, XML-RPC strings don't) — then the - # write_date cursor can NEVER advance and every pass re-pulls the same first page - # forever. When a full batch leaves the cursor unchanged we switch to an ID-WALK over - # write_date >= cursor (order id asc, id > tie), checkpointed as 'cursor|' in - # cursor_wd so a killed walk resumes; on completion the cursor jumps one second past - # the stamp (safe: the walk covered every row in that second, and the stamp is in the - # past — guarded by a 2-minute recency check). - raw = st["cursor_wd"] or "1970-01-01 00:00:00" - cursor, _pipe, _tie = raw.partition("|") - walking, tie_id = bool(_pipe), int(_tie or 0) - walk_max = cursor - for _ in range(max_batches): - if walking: - dom = _base_domain(spec) + [("write_date", ">=", cursor), ("id", ">", tie_id)] - recs = O.search_read(spec["model"], dom, ["id"] + fields, - limit=batch, order="id asc") - if not recs: # walk complete → advance past the stamp - try: - wm = dt.datetime.strptime(walk_max, "%Y-%m-%d %H:%M:%S") - if wm < dt.datetime.utcnow() - dt.timedelta(minutes=2): - cursor = (wm + dt.timedelta(seconds=1)).strftime("%Y-%m-%d %H:%M:%S") - else: - cursor = walk_max - except ValueError: - cursor = walk_max - walking = False - log(f" {key}: id-walk complete -> cursor {cursor}") - break - _upsert(con, key, spec, recs) - pulled += len(recs) - tie_id = recs[-1]["id"] - walk_max = max(walk_max, max(str(r["write_date"]) for r in recs)) - _save_state(con, key, "live", st["last_id"], f"{cursor}|{tie_id}") - continue - dom = _base_domain(spec) + [("write_date", ">=", cursor)] - recs = O.search_read(spec["model"], dom, ["id"] + fields, - limit=batch, order="write_date asc, id asc") - if not recs or (len(recs) == 1 and str(recs[0].get("write_date")) == cursor - and pulled == 0 and _already(con, key, recs[0])): - break - new_cursor = str(recs[-1]["write_date"]) - if new_cursor == cursor and len(recs) == batch: - # full batch stuck on one second: enter the id-walk from id 0 (the write_date - # ordering shuffles ids within the stamp second, so the batch just pulled is - # NOT an id prefix — restart coverage by id; upserts make the overlap free) - walking, tie_id, walk_max = True, 0, cursor - _save_state(con, key, "live", st["last_id"], f"{cursor}|0") - log(f" {key}: mass-stamp second at {cursor} -> switching to id-walk") - continue - _upsert(con, key, spec, recs) - pulled += len(recs) - if new_cursor == cursor and len(recs) < batch: - cursor = new_cursor - break - cursor = new_cursor - # Checkpoint the cursor EVERY batch: a big delta (88k account_move rows after the - # 2026-07-15 recompute) means a killed pass otherwise re-pulls everything — - # upserts are idempotent, but the wasted RPC is real (learned 2026-07-17). - _save_state(con, key, "live", st["last_id"], cursor) - _save_state(con, key, "live", st["last_id"], - f"{cursor}|{tie_id}" if walking else cursor) - if pulled: - log(f" {key}: live sync +{pulled:,} rows (cursor {cursor}" - + (f" walking id>{tie_id}" if walking else "") + ")") - return "live", pulled - finally: - con.close() - - -def _already(con, key, rec): - return bool(con.execute(f"SELECT 1 FROM {key} WHERE id=?", [rec["id"]]).fetchone()) - - -def sync_all(entities=None, max_batches=200, log=print): - # ⭐ D-10. `{}` rather than a dict of paused phases, and the caller decides why that matters: - # `api/main.py:354` loops up to 12 times until `all(phase == "live")`, and `all()` over an - # empty dict is True — so a paused store costs ONE pass instead of twelve rounds of the same - # refusal. A dict of `"paused"` phases would fail that test every time and turn the boot - # sprint into a busy-wait against a connector nobody has resumed. - if source_paused(): - _paused_notice("sync", log) - return {} - out = {} - for key in (entities or list(ENTITIES)): - t0 = time.time() - phase, pulled = sync_entity(key, max_batches=max_batches, log=log) - out[key] = {"phase": phase, "pulled": pulled, "secs": round(time.time() - t0, 1)} - # A full pass that lands every entity in live mode latches process readiness in-process — so - # readers never race the writer's lock to (re)discover the store is usable. - if entities is None and out and all(v["phase"] == "live" for v in out.values()): - mark_ready() - return out - - -def reconcile_deletes(entities=None, log=print): - """Remove store rows hard-deleted in Odoo. unlink() doesn't move write_date, so the cursor - sync can never see a deletion — the count parity DETECTS the drift (by design); this repairs - it. Pulls the full live id set per entity in bounded id-ascending pages (id-only reads are - cheap) and deletes store ids not present live. Run from sync_runner / maintenance, not the - app's frequent passes. - - ⭐ D-10: it pulls the FULL live id set per entity, so it is the single largest Odoo read in - this module and it runs unattended at boot and every fourth resync pass. Paused means paused. - """ - if source_paused(): - _paused_notice("delete reconcile", log) - return {} - con = connect() - try: - out = {} - for key in (entities or list(ENTITIES)): - spec = ENTITIES[key] - live_ids, last = set(), 0 - while True: - recs = O.search_read(spec["model"], _base_domain(spec) + [("id", ">", last)], - ["id"], limit=50000, order="id asc") - if not recs: - break - live_ids.update(r["id"] for r in recs) - last = recs[-1]["id"] - store_ids = {r[0] for r in con.execute(f"SELECT id FROM {key}").fetchall()} - dead = store_ids - live_ids - if dead: - con.execute("CREATE TEMP TABLE IF NOT EXISTS _dead(id BIGINT)") - con.execute("DELETE FROM _dead") - con.executemany("INSERT INTO _dead VALUES (?)", [[i] for i in dead]) - con.execute(f"DELETE FROM {key} WHERE id IN (SELECT id FROM _dead)") - con.execute("DELETE FROM _dead") - n = con.execute(f"SELECT count(*) FROM {key}").fetchone()[0] - con.execute("UPDATE _sync_state SET rows=? WHERE entity=?", [n, key]) - out[key] = len(dead) - log(f" {key}: reconciled {len(dead):,} hard-deleted rows") - return out - finally: - con.close() - - -def status(): - con = connect() - try: - return {r[0]: {"phase": r[1], "cursor": r[3], "rows": r[4], "updated": r[5]} - for r in con.execute("SELECT * FROM _sync_state").fetchall()} - finally: - con.close() - - -# ------------------------------------------------------------------ parity (store vs LIVE Odoo) - -def _check(name, ours, live, tol=0.01): - gap = abs((ours or 0) - (live or 0)) - return {"check": name, "ok": gap <= tol, "gap": round(gap, 4), "ours": ours, "live": live} - - -def validate(pre=None, months=("2026-01-01", "2026-07-01")): - """Raw-fidelity parity: the store must mirror live Odoo — row counts per synced entity + - unscoped monetary sums over a window, to the cent. Only checks entities in LIVE phase. - - ⭐ D-10 REFUSES LOUDLY HERE rather than skipping quietly, and the difference matters. Every - other paused path has a correct silent answer — serve the mirror — because the mirror is the - thing being asked for. This function's ENTIRE contract is "compare me against live Odoo", so - a paused version of it has no honest result: skipping would return an empty pass, and - proceeding would make the live call the pause forbids. An operator running a reconciliation - against a paused connector has asked a question that cannot be answered, and being told so is - the only outcome that is not a lie. (Operator-invoked only — nothing in the serving path - calls it, so this raises for a person, never inside a request.) - """ - if source_paused(): - raise RuntimeError( - "the tenant's Odoo connector is PAUSED, so the store cannot be validated against " - "live Odoo - resume it under Settings > Connectors and run this again") - con = connect() - out = [] - try: - live_state = status() - d0, d1 = months - for key, spec in ENTITIES.items(): - if live_state.get(key, {}).get("phase") != "live": - continue - ours = con.execute(f"SELECT count(*) FROM {key}").fetchone()[0] - live = O.get_odoo().search_count(spec["model"], _base_domain(spec)) - out.append(_check(f"store.{key} row count == live search_count", ours, live, tol=0)) - if live_state.get("sale_order_line", {}).get("phase") == "live": - ours = con.execute( - "SELECT coalesce(sum(l.price_subtotal),0) FROM sale_order_line l " - "JOIN sale_order o ON o.id = l.order_id " - "WHERE o.date_order >= ? AND o.date_order < ?", [d0, d1]).fetchone()[0] - live = O.sum_field("sale.order.line", - [("order_id.date_order", ">=", f"{d0} 00:00:00"), - ("order_id.date_order", "<", f"{d1} 00:00:00")], - "price_subtotal") - out.append(_check(f"store Σ line revenue [{d0}..{d1}) == live (unscoped)", ours, live)) - if live_state.get("sale_order", {}).get("phase") == "live": - ours = con.execute( - "SELECT coalesce(sum(amount_untaxed),0) FROM sale_order " - "WHERE date_order >= ? AND date_order < ?", [d0, d1]).fetchone()[0] - live = O.sum_field("sale.order", - [("date_order", ">=", f"{d0} 00:00:00"), - ("date_order", "<", f"{d1} 00:00:00")], "amount_untaxed") - out.append(_check(f"store Σ order amount_untaxed [{d0}..{d1}) == live (unscoped)", - ours, live)) - finally: - con.close() - return out - - -if __name__ == "__main__": - args = sys.argv[1:] - if args and args[0] == "status": - for k, v in status().items(): - print(f" {k:18s} {v['phase']:9s} rows={v['rows']:>9,} cursor={v['cursor']} ({v['updated']})") - elif args and args[0] == "reconcile": - ents = args[1].split(",") if len(args) > 1 else None - for k, n in reconcile_deletes(ents).items(): - print(f" {k:18s} -{n:,} hard-deleted rows") - elif args and args[0] == "validate": - ok = True - for c in validate(): - mark = "OK " if c["ok"] else "XX " - ok &= c["ok"] - print(f" {mark}{c['check'][:80]} gap={c['gap']}") - print("PARITY:", "GREEN" if ok else "RED") - sys.exit(0 if ok else 1) - else: - ents = args[0].split(",") if args else None - mb = int(args[1]) if len(args) > 1 else 200 - for k, v in sync_all(ents, max_batches=mb).items(): - print(f" {k:18s} {v['phase']:9s} +{v['pulled']:,} rows in {v['secs']}s") +"""harness/store.py — the tenant data store (OM-1, 2026-07-11). + +Incremental Odoo → DuckDB mirror: the local analytical store that saved views/dashboards (OM-3) +and the AI Analyst's tools (OM-4) query at warehouse speed, instead of hammering live XML-RPC +per question. +Plan: .claude/wiki/research/omni-adoption.md Part IV (OM-1) + [[odoo-open-source]] BUILD-NOW #1. + +Design rules: + - The store is a RAW mirror (unscoped). Business scope (wholesale teams, GIFTWARE exclusion, + confirmed-only) is applied by the SEMANTIC layer at query time (OM-2) — one source of truth. + - Sync is CHECKPOINTED + BOUNDED + RESUMABLE (loop-library discipline): backfill paginates by + id; live mode advances a write_date cursor; every run is bounded by max_batches and safe to + kill/re-run (upserts are idempotent). + - validate() compares the store against LIVE Odoo (raw fidelity: counts + monthly sums to the + cent) — the store never validates itself. Odoo hard-deletes (unlink) don't move write_date; + the count checks are the drift detector for those. + - Client data: the .duckdb file lives under data/store/ (git-ignored). READ-ONLY on Odoo. +""" +import datetime as dt +import json +import os +import re +import sys +import time +from pathlib import Path + +import duckdb + +import core.odoo as O + +_STORE_DIR = Path(__file__).resolve().parents[1] / "data" / "store" + +# ⚠ MUTABLE ON PURPOSE, and read LATE by every function below (Python resolves a module global at +# CALL time), which is how `provision_tenant.py` already points a fresh instance at its own file. +# EXIT-4a (X7) formalises that: `AIOS_DUCKDB_PATH` lets a container name the file without an edit, +# `path_for()` owns the per-tenant naming convention, and `use_path()` is the SAFE way to switch — +# see its docstring for why a bare assignment is not. +# The DEFAULT is unchanged: /data/store/royal.duckdb. +DB_PATH = Path(os.environ.get("AIOS_DUCKDB_PATH") or (_STORE_DIR / "royal.duckdb")) + + +# ───────────────────────────────────────────────────────────────────────────────────────────── +# ⭐ DEBT D-10 (wave 24) — THE CONNECTOR PAUSE REACHES THE MEASURE MIRROR. +# +# WHAT THE BUG WAS. Pausing a tenant's Odoo connector froze the CUSTOMER POOL (DEBT-2 built that: +# `routes_customers._pool_for` serves the pause-time snapshot) and nothing else. Every measure +# column and condition in the product is answered from THIS store, and this store kept pulling +# from Odoo the entire time a connector was paused — 12 sync passes at boot and another every +# 1800 s, hundreds of `search_read`s each. The Settings copy said so out loud ("measures not +# already computed may still reach the source until their cutover") and that sentence was the +# debt row. +# +# ⛔ WHY A PROBE AND NOT AN IMPORT. The pause flag lives in the tenant's store bucket, which is +# `harness.runtime`'s to read — and `runtime` already imports THIS module (`_ds.DB_PATH`), so an +# import back would be a cycle. The harness installs the probe at its own import instead. Nothing +# else changes: with no probe installed (every gate, every ops script, `sync_runner`) the answer +# is False and this module behaves exactly as it did. +# +# ⚠ IT FAILS TOWARDS "NOT PAUSED", deliberately and consistently with the shipped decision one +# layer up (`routes_keychain.odoo_paused`: "an unreachable flags bucket must never freeze a live +# surface"). The alternative — assume paused when the answer cannot be resolved — turns any +# transient store glitch into a mirror that silently stops advancing, which is the failure mode +# nobody notices for a week. Two copies of that policy would be worse than one; this is the same +# one. +_paused_probe = None + + +def set_paused_probe(fn): + """Install the callable that answers 'is the connector paused for the tenant whose store this + process has open?'. `harness.runtime` installs the real one; pass None to remove it.""" + global _paused_probe + _paused_probe = fn + + +def source_paused(): + """True when this store's tenant has its Odoo connector paused. Never raises.""" + try: + return bool(_paused_probe and _paused_probe()) + except Exception: + return False + + +#: What a sync entry point answers instead of reaching Odoo. A PHASE WORD, not an exception and +#: not a silent zero-row success: `status()` keeps whatever the last real pass wrote, so the +#: mirror still reports the age it genuinely has ([[no-unverifiable-aggregates]] — a paused +#: mirror that reported itself as freshly synced would be the unverifiable claim). +PAUSED_PHASE = "paused" + + +def _paused_notice(what, log): + # Printed as well as logged, and that is deliberate: `api/main.py` passes a swallowing + # `log=lambda *a, **k: None`, so the log-only version of this line would never be seen by + # the one caller that runs it unattended every 30 minutes. + msg = (f"[datastore] {what} skipped - the tenant's Odoo connector is PAUSED; the mirror is " + f"serving its pause-time state. Resume it under Settings > Connectors.") + try: + log(msg) + except Exception: + pass + print(msg) + + +def path_for(tenant_key): + """The canonical DuckDB file for a tenant slug. `royal-imports` keeps the historical + `royal.duckdb` name so tenant #0's existing 175 MB file is not orphaned by a rename. + + File-per-tenant IS the isolation model (C1c): DuckDB has no row-level security, so the only + boundary that holds is the operating system's — one file, one tenant. + """ + key = str(tenant_key or "").strip().lower() + if key in ("", "royal-imports"): + return DB_PATH if key == "" else Path( + os.environ.get("AIOS_DUCKDB_PATH") or (_STORE_DIR / "royal.duckdb")) + safe = "".join(c if (c.isalnum() or c in "-_") else "_" for c in key)[:60] + return _STORE_DIR / f"{safe}.duckdb" + + +def use_path(path): + """Repoint the store at `path`, CLOSING the process singleton first. + + ⚠ A bare `datastore.DB_PATH = …` is not enough and is the more dangerous half of this + operation. `_instance()` caches ONE open connection for the life of the process, so after a + plain reassignment every read keeps being served from the PREVIOUSLY opened file — a silent + cross-tenant read, returning real rows that belong to somebody else. Closing the connection + and clearing the readiness memo (which is also per-file) is what makes the switch honest. + + ⚠ AND CLOSING THE CONNECTION IS STILL NOT ENOUGH ON ITS OWN, because `ro_con()` caches a + cursor in a `threading.local`: this function can only clear the CALLING thread's, and + `ro_con`'s liveness probe (`SELECT 1`) SUCCEEDS on a cursor whose underlying connection was + closed out from under it in some cases — so another thread would keep answering from the old + file with no error to notice. The generation counter below is what closes that: `ro_con` + compares the generation its cursor was opened under and reopens when it has moved, which is + a check no individual thread can forget to do. + + Returns the new path. Callers: `provision_tenant.py`, and any single-tenant worker process. + NOT a per-request operation — one process serves one analytical store at a time, which is why + `harness.runtime` hands out the PATH rather than switching the global underneath a request. + """ + global DB_PATH + with _INSTANCE_LOCK: + con = _INSTANCE.get("con") + if con is not None: + try: + con.close() + except Exception: + pass + _INSTANCE["con"] = None + _RO_TLS.__dict__.pop("cur", None) + _READY["ok"] = False + _GENERATION[0] += 1 # every other thread's cached cursor is now stale + DB_PATH = Path(path) + return DB_PATH + +# Entity specs: Odoo model → store table. m2o fields land as _id + _name. +# 'archivable' entities are pulled with active in [True, False] (the re-SKU/merge rule). +ENTITIES = { + "sale_order": { + "model": "sale.order", "archivable": False, + "fields": ["name", "date_order", "partner_id", "team_id", "user_id", "state", + "amount_untaxed", "invoice_status", "write_date"], + }, + "sale_order_line": { + "model": "sale.order.line", "archivable": False, + "fields": ["order_id", "order_partner_id", "product_id", "product_uom_qty", + "price_subtotal", "margin", "purchase_price", "write_date"], + }, + "res_partner": { + "model": "res.partner", "archivable": True, + # `agent` (bool, labelled "Creditor/Agent") is THE discriminator between a real sales + # AGENT and an internal SALESPERSON who merely carries commission lines — the commission + # table itself cannot tell them apart. `salesman_as_agent` ("convert salesman into agent") + # marks agents who came from staff; the owner ruling on whether those count as agents is + # OUTSTANDING, so both flags are synced and neither reading is baked in. + # ⭐ WAVE 29 (W29-T54, amendment A4). `customer_rank` is the ONLY predicate that + # distinguishes "a partner Odoo has TRANSACTED with" (today's population, 2,465) from + # "a customer" (3,614 with `rank>0 active`) — a 1,149-partner drop, the same join-drop + # class as item 22's SKU pool. It cannot be derived from the document tables, because + # the missing partners are precisely the ones with no document. + # ⛔ It is synced HERE rather than read from live Odoo because `read_customers` also runs + # at BOOT off a hydrated snapshot, where Odoo may be unreachable — sourcing it live would + # make the population "whatever source answered this time", changing size with the network. + # ⭐ WAVE 30 (session E's ask, W30-T33). `street`/`street2`/`zip` are the POSTAL half of + # an address the mirror has never carried: it holds `city`, `state_id` and `country_id` + # and stops there, so `odoo_relational.read_customers` — which reads only the mirror — + # cannot serve a full address at all, however the column is declared. + # ⛔ DECLARING THEM DOES NOT FILL THEM, and the difference is a whole sync: an existing + # mirror ALTERs the columns in on the next `sync_all()` and leaves them NULL until the + # backfill below has walked every partner (a partner's `write_date` did not move because + # WE changed the schema). Every projection degrades a missing/NULL column to blank rather + # than raising, so declaring early is safe — it is just not yet true. + # ⭐⭐ WAVE 33 (W33-T48, owner item 13) — THE CONTACT/IDENTITY HALF, MEASURED BEFORE IT WAS + # ADDED. The owner: *"I could have sworn we have more relevant Fields under 'Odoo vendors' + # etc."* He was right, and the census + # (`.claude/wiki/waves/wave33/odoo-field-census.md`, live read-only, 2026-08-14) says by how + # much: `res.partner` declares 198 fields, **77 are populated on all 3,621 customers and 76 + # on all 658 vendors**, and this mirror carried THIRTEEN. Every one of the six added here + # was measured non-empty on the live population first — none is added on the strength of + # "Odoo probably has that", which is how a schema grows columns nobody can fill. + # ⚠ THE MIRROR IS THE CEILING, NOT THE GRID. Every reader in `odoo_relational.py` reads + # THIS file, so a column added to a `*_fields()` contract with no entry here renders as a + # blank column forever — which is D-165's shape exactly, and the reason the widening starts + # on this line rather than on the grid. + "fields": ["name", "team_id", "city", "state_id", "country_id", "active", + "agent", "salesman_as_agent", "customer_rank", + "street", "street2", "zip", + "email", "phone", "mobile", "website", "vat", "ref", + "write_date"], + # `backfill_fields` carries it too, so mirrors that already exist pick it up on the next + # sync_all() instead of needing a reseed. + # ⛔ D-165 IS ABOUT THIS LIST, AND THE CENSUS SETTLED WHICH HALF IS WRONG: `street`, + # `street2` and `zip` are `store=True` on `res.partner` and POPULATED in live Odoo — they + # are declared here and still NULL in the mirror purely because a backfill pass has not + # walked every partner (changing OUR schema does not move a partner's `write_date`, so an + # incremental sync never revisits them). Declaring is not filling; the fix is a run. + "backfill_fields": ["agent", "salesman_as_agent", # added 2026-07-28 + "customer_rank", # added 2026-08-11 (W29-T54) + "street", "street2", "zip", # added 2026-08-12 (W30-T33) + "email", "phone", "mobile", # added 2026-08-14 (W33-T48) + "website", "vat", "ref"], + # m2m fields land as _id (the FIRST linked id — the Customers-module convention: + # a customer's assigned agent is agent_ids[0], ~one per customer, MECE) + as a + # JSON list of ALL ids. Declared explicitly: a 2-element m2m read would otherwise be + # indistinguishable from an m2o (id, name) pair in _flatten. + "m2m": ["agent_ids"], + }, + "product_product": { + "model": "product.product", "archivable": True, + "fields": ["default_code", "name", "categ_id", "type", "standard_price", "active", + "write_date"], + }, + "account_move": { + "model": "account.move", "archivable": False, + # ⭐ `invoice_origin` added 2026-08-09 (wave 28, DEBT D-88) — the ORDER NAME an invoice was + # raised from ("S12345"), which is the only single stored key that joins an invoice back + # to its order. + # ⛔ WITHOUT IT THAT JOIN IS TWO HOPS, and the second one is expensive: invoice -> + # account_move_line -> `sale_line_ids` (the m2m BU bridge) -> sale_order_line -> + # sale_order, across 963,783 move lines. Wave 27 wanted an order->invoice preset link and + # had no single column to derive it from, so the link was not built at all. + # ⚠ IT IS A NAME, NOT AN ID, and Odoo writes free text into it — a manual invoice can hold + # anything, and a merged one can hold several origins space-separated. So it is a JOIN + # HINT, and whatever consumes it matches against the order names that actually exist + # rather than trusting the string. The m2m route above stays the authority for BU + # attribution; this does not replace it. + "fields": ["name", "move_type", "state", "invoice_date", "invoice_date_due", + "partner_id", "amount_untaxed_signed", "amount_residual_signed", + "payment_state", "invoice_origin", "write_date"], + # Existing rows never re-sync on their own (WE moved the schema; their write_date did + # not), so without this the column ALTERs in and stays NULL forever — indistinguishable + # from "this invoice genuinely has no origin". Declared, re-entrant, checkpointed. + "backfill_fields": ["invoice_origin"], # added 2026-08-09 + }, + "account_move_line": { + "model": "account.move.line", "archivable": False, + # display_type/move_type/parent_state/price_subtotal/product_id added 2026-07-28 so the + # invoice-LINE grain is queryable (topic `invoice_lines`). + # ⚠ display_type values here are 'product', 'cogs', 'payment_term', 'line_note', + # 'line_section'. Commission rows attach to 'cogs' lines as well as 'product' ones + # (96,936 of 147,160 in 2025). Filtering to 'product' is a GRAIN guard: measured + # 2026-07-29, omitting it does not move revenue (non-product lines carry + # price_subtotal = 0) but it inflates commission ROW COUNTS ~80% and would corrupt any + # count-of-lines measure. + "fields": ["move_id", "account_id", "partner_id", "date", "debit", "credit", + "balance", "display_type", "move_type", "parent_state", "price_subtotal", + "product_id", "write_date"], + "backfill_fields": ["display_type", "move_type", "parent_state", "price_subtotal", + "product_id"], # added 2026-07-28 + # the BU bridge: account.move carries NO business unit (every invoice sits on team 1 — + # see [[invoice-bu-attribution]]), so Fisch/Royal on the invoice basis is only reachable + # through the originating sale order. sale_line_id (the FIRST linked sale line) is + # lossless here: ZERO 2025 invoice lines span more than one team (measured). + "m2m": ["sale_line_ids"], + }, + "account_invoice_line_agent": { + # The OCA sale-commission module: one row per (invoice line × agent). This is the SECOND + # agent source — per-line commission attribution — and it names a different population + # than res_partner.agent_ids (the customer-master book). See topic `invoice_lines`. + # ⚠ TWO similarly-named FKs, do not confuse them: + # object_id -> account_move_line (THE grain; join on this) + # invoice_id -> account_move (the parent document) + "model": "account.invoice.line.agent", "archivable": False, + "fields": ["agent_id", "commission_id", "amount", "invoice_id", "object_id", + "invoice_date", "settled", "write_date"], + }, + "account_account": { + # NOTE: no `active` field on account.account in this Odoo version (learned 2026-07-11 — + # a ('active','in',...) domain 500s); treat as non-archivable. + "model": "account.account", "archivable": False, + "fields": ["code", "name", "account_type", "write_date"], + }, + # ⭐⭐ W37-T13 (owner: stock moved IN and OUT per SKU) — MEASURED FIRST in + # `proto/P1-stock-moves.md`, and every clause below is a finding rather than a guess. + # + # ⛔ THE TWO LOCATION COLUMNS ARE BOTH LOAD-BEARING AND THIS IS THE TICKET'S HEADLINE TRAP. + # A move is an IN when it ARRIVES somewhere internal from somewhere that is not, and an OUT + # when it LEAVES somewhere internal for somewhere that is not. `internal -> internal` is + # **58% of all moves** and is NEITHER — a one-sided domain counts every one of them as BOTH, + # which roughly doubles both columns while looking entirely plausible. + # + # ⛔ `picking_code` IS NOT THE DISCRIMINATOR, and the reason is worth the line: it is + # `store=False`, so a `read_group` on it fails hard, AND `('picking_code','=',False)` returns + # 0 while 5,800 pickingless done moves exist — **the count that would have warned you also + # reads 0**. `picking_id` is stored and is carried instead, for provenance only. + # + # ⚠ `quantity_done` IS THE AGGREGATE, not `product_uom_qty`: it ties to + # `stock.move.line.qty_done` to the cent over the full window, and it is what actually moved. + # ⚠ NOT ARCHIVABLE — `stock.move` has no `active` field; a move is history and is never + # archived. Stated so the next reader does not "fix" it by adding `active in [True, False]`. + # ⚠ SIZE: 236k rows over 1,095 days (P1), about a quarter of `account_move_line`'s 963,783. + "stock_move": { + # ⛔ `optional` — see `ready()`. A NEW entity must never gate the readiness of surfaces + # that do not read it, or adding one takes the product down until it has synced. + "model": "stock.move", "archivable": False, "optional": True, + "fields": ["product_id", "quantity_done", "date", "state", + "location_id", "location_dest_id", "picking_id", "write_date"], + }, + # ⭐ THE USAGE LOOKUP, WITHOUT WHICH THE ROW ABOVE CANNOT BE READ. `location_id` syncs as + # `location_id` + `location_id_name` (the m2o convention) — a NAME, never a usage — so the + # two-sided domain has nothing to test until this table exists. It is tiny (a few hundred + # rows) and it is the whole difference between a metric and a column of doubled numbers. + # ⚠ `usage` is a selection: `internal` | `supplier` | `customer` | `inventory` | `production` + # | `transit` | `view`. ⚠ Adjustments (`inventory`) and scrap are 30% of IN / 20% of OUT and + # the two obvious exclusions are NOT equivalent — the 5,320-unit gap is all + # `Virtual Locations/Scrap` — so the metric that consumes this CHOOSES deliberately and says + # which; that choice lives in `model/metrics/stock.yml`, not here. + "stock_location": { + "model": "stock.location", "archivable": True, "optional": True, + "fields": ["name", "complete_name", "usage", "active", "write_date"], + }, +} + +BATCH = 2000 + +# Fields that are genuinely BOOLEAN. This list is load-bearing in TWO places: _cols types the +# column, and _flatten must NOT collapse a real False into NULL. Odoo returns False both for +# "empty" and for "boolean false", so a bool field missing from here silently becomes NULL — and +# for res_partner.agent that would erase the ONE flag separating an AGENT from a SALESPERSON +# (see [[invoice-line-agent-commission]]): every row would read "not an agent" indistinguishably +# from "unknown". +BOOL_FIELDS = ("active", "agent", "salesman_as_agent", "settled", "commission_free") + +# ⭐ WAVE 29 — fields that are genuinely INTEGER COUNTS, and this list exists because its absence +# shipped a silent defect within minutes of `customer_rank` being added to the res_partner spec. +# `_cols` types anything it does not recognise as VARCHAR, so `customer_rank` landed as text, the +# mirror populated correctly, every row looked right — and `odoo_relational`'s own predicate +# `customer_rank > 0` died with `Binder Error: Cannot compare values of type VARCHAR and type +# INTEGER_LITERAL`. ⛔ THE COLUMN WAS PRESENT AND THE DATA WAS CORRECT; only its TYPE was wrong, +# which is why "is the field in the mirror?" answered yes and the feature still could not run. +# ⇒ A new numeric field belongs HERE, exactly as a new boolean belongs in BOOL_FIELDS above. +INT_FIELDS = ("customer_rank", "supplier_rank") + + +import threading as _thr + +_RO_TLS = _thr.local() +_READY = {"ok": False} +_INSTANCE = {"con": None} +_INSTANCE_LOCK = _thr.Lock() +#: Bumped by `use_path()` whenever the store FILE changes. Every per-thread cursor is stamped with +#: the generation it was opened under, so a thread that never called `use_path` still cannot keep +#: reading the previous tenant's file — see `ro_con`. A plain "is the cursor alive?" probe cannot +#: answer that question, which is why this counter exists rather than more probing. +_GENERATION = [0] + + +def mark_ready(): + """In-process signal that the store is fully synced (every entity live). The sync writer calls + this after an all-live pass so a reader never has to (re)discover usability by touching the + file. Belt-and-suspenders alongside the shared-instance model below.""" + _READY["ok"] = True + + +def _instance(): + """THE one process-wide read-write DuckDB connection. Readers take .cursor() off it and the + sync writer uses it too — DuckDB serves concurrent reads + a single writer from ONE instance + via MVCC, so a read NEVER contends with a sync pass for the file lock. + + This is the fix for the 'connection error unless I refresh' bug (2026-07-16): an independent + read-only `duckdb.connect()` FAILS while the writer holds the lock ('Can't open a connection + to same database file with a different configuration than existing connections'); a cursor on + the shared instance does not. Openers must ensure the file is fully present first — ensure_seed + writes atomically (os.replace) and every read path guards on DB_PATH.exists() — so the instance + never binds a partial or absent file. (Trade-off: the app now holds the write lock for its whole + life, so a SEPARATE process — e.g. mcp_server.py — cannot open the same store file concurrently; + in production only the app runs, and that was already true during any sync pass.)""" + con = _INSTANCE["con"] + if con is not None: + return con + with _INSTANCE_LOCK: + if _INSTANCE["con"] is None: + DB_PATH.parent.mkdir(parents=True, exist_ok=True) + _INSTANCE["con"] = duckdb.connect(str(DB_PATH)) + return _INSTANCE["con"] + + +def ro_cursor(): + """A read cursor on the shared instance — the contention-free way to read the store.""" + return _instance().cursor() + + +def ready(): + """True once EVERY entity has completed its backfill (phase 'live'). Gate all store READS on + this: a mid-backfill store would return PARTIAL totals silently — worse than an error. Positive + results cache for the process (backfill never regresses). Reads via a shared-instance cursor, + so it is never mis-reported as 'not ready' just because a background sync pass is mid-flight.""" + if _READY["ok"]: + return True + if not DB_PATH.exists(): + return False + try: + cur = ro_cursor() + try: + phases = {r[0]: r[1] for r in + cur.execute("SELECT entity, phase FROM _sync_state").fetchall()} + finally: + cur.close() + except Exception: + return False + # ⛔⛔ OPTIONAL ENTITIES DO NOT GATE READINESS, AND WITHOUT THIS CLAUSE ADDING ONE IS AN + # OUTAGE. `ready()` gates EVERY store read — the Analyst, every measure column, every + # read-through grid — so a new entity in `ENTITIES` makes a mirror that has never synced it + # report NOT READY and refuses all of them with "the data cache is still warming up". A Space + # hydrates from a SNAPSHOT taken before the entity existed, so this would fire on every + # deploy of a wave that adds one: a whole-product outage from a schema line, with the app + # RUNNING and every gate green. MEASURED on this box the moment `stock_move` was declared + # (W37-T13): `ready()` flipped to False against an otherwise complete mirror. + # ⚠ THE GUARANTEE IS NOT WEAKENED, IT IS SCOPED. `ready()` still means "every entity the + # EXISTING surfaces read has finished its backfill". An optional entity's own consumers must + # check ITS phase before promising a column from it — which is what + # `semantic._source_ready` does, refusing the offer with a cause rather than serving blanks. + _required = [k for k, v in ENTITIES.items() if not v.get("optional")] + if all(phases.get(k) == "live" for k in _required): + _READY["ok"] = True + return True + return False + + +def entity_live(key): + """Has THIS entity finished its backfill? The per-entity half of `ready()`. + + Exists because `ready()` is deliberately blind to optional entities (see above), so a caller + that depends on one has to ask about it specifically instead of inferring it from a global + flag that was never about them. + """ + if not DB_PATH.exists(): + return False + try: + cur = ro_cursor() + try: + row = cur.execute("SELECT phase FROM _sync_state WHERE entity = ?", [key]).fetchone() + finally: + cur.close() + except Exception: # noqa: BLE001 + return False + return bool(row) and row[0] == "live" + + +def ro_con(): + """Per-thread cached read CURSOR on the shared instance for page/module store reads (the + open cost dominates repeated small queries — the Expenses retrofit measured 7.7s→1.0s). No + lock contention with the sync writer; a cursor that has gone bad is reopened once. REFUSES + until ready() — callers either fall back to live reads (modules) or relay a readable message + (Analyst). + + ⚠ THE GENERATION CHECK IS A CORRECTNESS GUARD, NOT AN OPTIMISATION (EXIT-4a). The liveness + probe below asks "does this cursor still work?", which is a different question from "is this + cursor still pointing at the file we are supposed to be reading?". After `use_path()` switches + tenants, a cursor cached in ANOTHER thread's `threading.local` can still answer `SELECT 1` — + and would then serve that thread real, plausible rows from the PREVIOUS tenant's file, with no + exception anywhere. `use_path` bumps the generation; a cursor opened under an older one is + discarded here, which is the only place every thread is guaranteed to pass through.""" + cur = getattr(_RO_TLS, "cur", None) + if cur is not None and getattr(_RO_TLS, "gen", None) != _GENERATION[0]: + try: + cur.close() + except Exception: + pass + cur, _RO_TLS.cur = None, None # the store moved underneath this thread + if cur is not None: + try: + cur.execute("SELECT 1") # cheap liveness probe + return cur + except Exception: + _RO_TLS.cur = None # stale → reopen below + if not ready(): + raise RuntimeError("the tenant data store is completing its FIRST sync on this " + "deployment — dashboards work meanwhile (live reads); retry " + "store queries in a few minutes") + cur = ro_cursor() + _RO_TLS.cur = cur + _RO_TLS.gen = _GENERATION[0] # stamp WHICH file this cursor belongs to + return cur + + +# ───────────────────────────────────────────────────────────────────────────────────────────── +# ⭐ WAVE 30 / OWNER RULING R6 + R7 — THE WINDOW. Read this before touching `window()`. +# +# R6, verbatim: *"there is no cap in how many data from the API source (as long as its from a +# connected source like Odoo) that can be pulled into the app."* R7 says how: a connected grid +# reads THROUGH this mirror instead of copying rows into the one `user_tables` document, which +# `MAX_ROWS = 60_000` bounds. The mirror is already uncapped — 971,034 GL lines live in this file +# today — so "no cap" is not a bigger number, it is a different mechanism: serve a SLICE and count +# the whole. +# +# ⛔⛔ `total` IS A `SELECT count(*)` OVER THE SAME PREDICATE, NEVER `len(rows)`. A window whose +# count is its own length is a fabricated aggregate that reads as authoritative — the class this +# repo has paid for twice ([[no-unverifiable-aggregates]], [[one-question-two-normalizers]]: a +# display predicate and a fold predicate answering one question differently). The two statements +# below are built from ONE `where` + ONE `params` tuple for exactly that reason; they cannot drift +# apart without deleting a line. +# +# ⛔ AND THE PREDICATE PUSHES DOWN. `where` is compiled by `harness/filter_sql.py` — the same +# compiler the client's filter engine is held in step with — so a filter matching rows outside the +# loaded window still COUNTS them. A caller that filters the returned list instead has silently +# asked "how many of the 200 rows in memory match" about a 971,034-row table. +# +# ⚠ `order_by` IS REQUIRED IN PRACTICE AND DEFAULTED HERE. Two pages of an UNORDERED window are +# not guaranteed to partition the table: DuckDB may legally return a row on page 1 and again on +# page 2, and the user sees a duplicate with no error anywhere. The default is the physical +# rowid-ish `1` only when a caller genuinely has no key; every real caller passes one. +# +# ⚠ TRUST BOUNDARY. `table`, `select`, `where` and `order_by` are SQL we author (a spec row, or +# `filter_sql`'s output over a whitelisted column map). Every VALUE is bound through `params`. +# `table` is additionally shape-checked below — not because a caller is hostile, but because a +# typo'd identifier interpolated into two statements is worth one cheap assertion. + +#: The most rows ONE request may carry back. Not a cap on the data — every row is reachable by +#: paging and `total` always tells the truth about how many there are — but a bound on the memory +#: a single response can cost. R6's second sentence applies: a caller that asks for more is CLAMPED +#: and the clamp is REPORTED (`routes_odoo_tables` turns it into a `limits` entry), never silent. +WINDOW_MAX = 5_000 + +#: What a caller gets when it names no window at all. Matches `semantic.store_rows`' own default so +#: the two windowed readers in this codebase do not disagree about what "a page" means. +WINDOW_DEFAULT = 200 + +_IDENT_OK = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +def window(table=None, select=None, where=None, params=(), order_by=None, + offset=0, limit=WINDOW_DEFAULT, cur=None, from_sql=None): + """One page of a mirror table PLUS the true count of everything the same predicate matches. + + Returns `{"rows": [tuple, ...], "columns": [name, ...], "total": int, "offset": int, + "limit": int, "clamped": bool}` — `rows` is the slice, `total` is the population. + + `select` is the projection expression list ("id, name, amount_untaxed"); `where` is a WHERE + body with `?` placeholders; `params` binds them, and is used by BOTH statements. + + ⚠ `table` and `from_sql` are the SAME argument wearing two trust levels, and they are separate + names on purpose. `table` is an identifier and is SHAPE-CHECKED; `from_sql` is a whole FROM + body — a parenthesised subquery with an alias — which cannot be checked at all. Two of the + eight connected grids need one (`read_customers` and `read_agents` are single statements whose + population is a SQL `UNION`), so the capability has to exist; naming it distinctly means a + reviewer can grep for every caller that hands over raw FROM SQL instead of inferring it. + """ + if bool(table) == bool(from_sql): + raise ValueError("datastore.window: pass exactly one of `table` or `from_sql`") + if from_sql: + name = str(from_sql).strip() + else: + name = str(table or "").strip() + if not _IDENT_OK.match(name): + raise ValueError(f"datastore.window: {table!r} is not a table identifier") + proj = str(select or "").strip() + if not proj: + raise ValueError("datastore.window: a projection is required — there is no implicit *") + try: + offset = max(0, int(offset or 0)) + except (TypeError, ValueError): + offset = 0 + try: + limit = int(limit if limit is not None else WINDOW_DEFAULT) + except (TypeError, ValueError): + limit = WINDOW_DEFAULT + clamped = limit > WINDOW_MAX or limit < 1 + limit = min(max(limit, 1), WINDOW_MAX) + + params = tuple(params or ()) + pred = str(where or "").strip() + tail = f" WHERE {pred}" if pred else "" + order = str(order_by or "").strip() or "1" + cur = cur if cur is not None else ro_con() + + # ⛔ THE COUNT RUNS FIRST AND OVER THE SAME `tail` + `params`. Ordering it first is deliberate: + # if the projection is ever wrong, the caller fails LOUDLY on the rows query rather than + # quietly returning a good count beside a broken page. + total = int(cur.execute(f"SELECT count(*) FROM {name}{tail}", params).fetchone()[0] or 0) + got = cur.execute( + f"SELECT {proj} FROM {name}{tail} ORDER BY {order} LIMIT {int(limit)} OFFSET {int(offset)}", + params).fetchall() + cols = [d[0] for d in (cur.description or [])] + return {"rows": [tuple(r) for r in got], "columns": cols, + "total": total, "offset": offset, "limit": limit, "clamped": clamped} + + +def columns_of(table, cur=None): + """The column names a mirror table actually has, lowercased. + + ⚠ A mirror can be `ready()` and still be MISSING COLUMNS: `ready()` reads entity PHASES, while + `_ensure_columns`/`_backfill_columns` checkpoint separately. Asking before projecting is what + stops a DuckDB `BinderException: Referenced column … not found` reaching a user as a bare 500 + (it already cost one live, which is why `odoo_relational.columns` exists on the app side). + """ + name = str(table or "").strip() + if not _IDENT_OK.match(name): + raise ValueError(f"datastore.columns_of: {table!r} is not a table identifier") + cur = cur if cur is not None else ro_con() + try: + rows = cur.execute(f"PRAGMA table_info('{name}')").fetchall() + except Exception: # noqa: BLE001 + return set() + return {str(r[1]).lower() for r in rows} + + +SEED_DATASET = os.environ.get("STORE_SEED_DATASET", "royal-imports/cfo-os-data") + + +def ensure_seed(): + """Fresh/ephemeral deployment (HF Space disks reset on every rebuild): hydrate the store + from the PRIVATE dataset's seed snapshot — Analyst-ready in ~a minute instead of a 20-40 min + XML-RPC backfill; the auto-sync then closes the gap via write_date cursors. Also restores + the pilot views/routines when absent. Fail-quiet: no seed/token → normal backfill.""" + import shutil + if DB_PATH.exists(): + return False + tok = os.environ.get("HF_TOKEN") + if not tok: + return False + try: + from huggingface_hub import hf_hub_download + DB_PATH.parent.mkdir(parents=True, exist_ok=True) + p = hf_hub_download(SEED_DATASET, "store_seed/royal.duckdb", + repo_type="dataset", token=tok) + # Atomic install: copy to a temp path then os.replace — a reader that guards on + # DB_PATH.exists() must never bind a half-written file into the shared instance. + tmp = DB_PATH.with_name(DB_PATH.name + ".tmp") + shutil.copyfile(p, tmp) + os.replace(tmp, DB_PATH) + for fn in ("views.json", "routines.json"): + tgt = DB_PATH.parent / fn + if not tgt.exists(): + try: + q = hf_hub_download(SEED_DATASET, f"store_seed/{fn}", + repo_type="dataset", token=tok) + shutil.copyfile(q, tgt) + except Exception: + pass + return True + except Exception: + return False + + +def connect(): + """A cursor on the shared instance for the SYNC WRITER (and status/validate). Returned as a + cursor so callers' con.close() frees the cursor without closing the process-wide instance; + one writer cursor is active at a time (the sync thread is sequential), readers use others.""" + cur = _instance().cursor() + cur.execute("""CREATE TABLE IF NOT EXISTS _sync_state ( + entity VARCHAR PRIMARY KEY, phase VARCHAR, last_id BIGINT, + cursor_wd VARCHAR, rows BIGINT, updated_at VARCHAR)""") + return cur + + +def _flatten(rec, fields, m2m=()): + out = {"id": rec["id"]} + for f in m2m: + v = rec.get(f) or [] + base = f.removesuffix("_ids") + out[f"{base}_id"] = v[0] if v else None + out[f] = json.dumps(list(v)) if v else None + for f in fields: + v = rec.get(f) + if isinstance(v, (list, tuple)) and len(v) == 2 and isinstance(v[0], int): + out[f"{f.removesuffix('_id')}_id"] = v[0] + out[f"{f.removesuffix('_id')}_name"] = str(v[1]) + elif isinstance(v, (list, tuple)): + out[f] = json.dumps(v) + elif v is False and f not in BOOL_FIELDS: # Odoo False = NULL for non-bool fields + out[f] = None + else: + out[f] = v + return out + + +#: ⛔⛔ A HARD-CODED NAME LIST DECIDES WHETHER A COLUMN IS A NUMBER, and a field missing from it +#: lands as VARCHAR — which does not fail, it fails LATER and somewhere else. MEASURED (W37-T13): +#: `stock.move.quantity_done` was absent, so 778,385 rows stored '4.0', '1.0' … as TEXT, every one +#: of them cleanly castable, and the only symptom was a DuckDB +#: *"Cannot mix values of type INTEGER_LITERAL and VARCHAR in CASE expression"* thrown by a +#: FILTERED MEASURE three layers up in `semantic._measure_sql`. Nothing in the sync, the schema or +#: the row count looked wrong. +#: ⚠ SO ADDING A NUMERIC FIELD TO AN `ENTITIES` SPEC MEANS ADDING IT HERE TOO. There is no +#: inference: `_row` writes whatever Odoo returned, and Odoo returns `False` for an unset numeric, +#: so a value-sniffing default would be wrong in the other direction. +#: ⚠ AND THE TYPE IS FIXED AT CREATE. `_ensure_columns` only ADDs missing columns — it never ALTERs +#: an existing one — so a mirror that already synced the field as VARCHAR keeps it until the table +#: is dropped and re-synced. Correcting the list fixes every FRESH mirror, which is the Space's +#: case (it has never synced this entity) and was not mine (I had to ALTER my scratch copy). +NUMERIC_FIELDS = ("amount_untaxed", "amount_untaxed_signed", "amount_residual_signed", + "price_subtotal", "margin", "purchase_price", "product_uom_qty", + "standard_price", "debit", "credit", "balance", "amount", + # W37-T13 — the stock-movement quantity. See the note above. + "quantity_done") + + +def _cols(spec): + cols = ["id BIGINT PRIMARY KEY"] + for f in spec["fields"]: + base = f.removesuffix("_id") + if f.endswith("_id"): + cols += [f"{base}_id BIGINT", f"{base}_name VARCHAR"] + elif f in BOOL_FIELDS: + cols.append(f"{f} BOOLEAN") + elif f in INT_FIELDS: + cols.append(f"{f} BIGINT") + elif f in NUMERIC_FIELDS: + cols.append(f"{f} DOUBLE") + else: + cols.append(f"{f} VARCHAR") + for f in spec.get("m2m") or []: + base = f.removesuffix("_ids") + cols += [f"{base}_id BIGINT", f"{f} VARCHAR"] + return cols + + +def _ensure_table(con, key, spec): + con.execute(f"CREATE TABLE IF NOT EXISTS {key} ({', '.join(_cols(spec))})") + + +def _ensure_columns(con, key, spec): + """Schema evolution: existing stores/seeds predate columns a newer spec introduces — + CREATE TABLE IF NOT EXISTS won't add them. Returns the newly added column names.""" + have = {r[1] for r in con.execute(f"PRAGMA table_info('{key}')").fetchall()} + added = [] + for c in _cols(spec): + name = c.split()[0] + if name not in have: + con.execute(f"ALTER TABLE {key} ADD COLUMN {c}") + added.append(name) + return added + + +def _field_cols(f): + """The store column(s) one spec field materialises into (m2o fields become _id + _name).""" + if f.endswith("_id"): + base = f.removesuffix("_id") + return [f"{base}_id", f"{base}_name"] + return [f] + + +def _backfill_columns(con, key, spec, log=print): + """Scalar/m2o schema evolution — the twin of _backfill_m2m, and for the same reason. + + _ensure_columns ALTERs a new column in, but existing rows NEVER re-sync (WE moved the schema; + their write_date didn't move), so a newly added field stays NULL forever — silently, and NULL + is indistinguishable from a legitimately empty value. That is how a store starts answering + 'no rows match' instead of erroring. Re-pull just the affected fields for every row, + paginated by id, marked durably so a crash retries rather than declaring completion. + + The fields are DECLARED in the spec (`backfill_fields`), never inferred from "which columns + did _ensure_columns just add" — that inference is wrong on the second run, when the ALTER has + already happened and the list comes back empty while the data is still missing. Declaring it + makes the migration re-entrant and reviewable. Leave the entry in place after it completes; + the durable marker, not the absence of the declaration, is what stops it re-running. + """ + fields = [f for f in (spec.get("backfill_fields") or []) if f != "write_date"] + if not fields: + return + marker = f"{key}.cols.{','.join(sorted(fields))}" + st = con.execute("SELECT phase, last_id, rows FROM _sync_state WHERE entity=?", + [marker]).fetchone() + if st and st[0] == "done": + return + # RESUMABLE, not restart-from-zero: these passes run to ~1M rows, and a migration that loses + # an hour of work to one dropped connection gets skipped by the next person under time + # pressure. Progress is checkpointed every page; 'done' is written only at the end, so an + # interrupted run resumes at the last committed id instead of re-pulling or, worse, declaring + # completion it never reached. + last, n = (st[1] or 0, st[2] or 0) if st else (0, 0) + if last: + log(f" {key}: column backfill resuming at id {last:,} ({n:,} rows already done)") + cols = [c for f in fields for c in _field_cols(f)] + setter = ", ".join(f"{c}=?" for c in cols) + while True: + recs = O.search_read(spec["model"], _base_domain(spec) + [("id", ">", last)], + fields, limit=5000, order="id asc") + if not recs: + break + rows = [_flatten(r, fields) for r in recs] + con.executemany(f"UPDATE {key} SET {setter} WHERE id=?", + [[r.get(c) for c in cols] + [r["id"]] for r in rows]) + last, n = recs[-1]["id"], n + len(recs) + con.execute("INSERT OR REPLACE INTO _sync_state VALUES (?,?,?,?,?,?)", + [marker, "running", last, None, n, time.strftime("%Y-%m-%d %H:%M:%S")]) + log(f" {key}: column backfill {'+'.join(fields)} -> id {last:,} ({n:,} rows)") + con.execute("INSERT OR REPLACE INTO _sync_state VALUES (?,?,?,?,?,?)", + [marker, "done", last, None, n, time.strftime("%Y-%m-%d %H:%M:%S")]) + log(f" {key}: column backfill complete -> {n:,} rows") + + +def _backfill_m2m(con, key, spec, log=print): + """One-time targeted backfill after a schema migration: existing rows never re-sync (their + write_date didn't move when WE added a column), so pull every record where the m2m field is + SET and update in place. Rows with the field empty keep NULL — correct and free. Completion + is a durable _sync_state marker ('.m2m.') — NOT 'column just added': a crash + between the ALTER and this backfill (seen live: the SSL-failed first run) must retry.""" + for f in spec.get("m2m") or []: + marker = f"{key}.m2m.{f}" + if con.execute("SELECT 1 FROM _sync_state WHERE entity=?", [marker]).fetchone(): + continue + base = f.removesuffix("_ids") + # PAGINATED by id. A single limit=100000 call silently TRUNCATED here: account_move_line + # has 228,067 rows with sale_line_ids set, so 56% of the BU bridge would have gone + # missing with no error — a partial backfill that reads as a complete one (the + # [[no-unverifiable-aggregates]] no-silent-caps rule). + rows, last, page = [], 0, 20000 + while True: + recs = O.search_read(spec["model"], + _base_domain(spec) + [(f, "!=", False), ("id", ">", last)], + [f], limit=page, order="id asc") + if not recs: + break + rows += [(r[f][0], json.dumps(list(r[f])), r["id"]) for r in recs if r.get(f)] + last = recs[-1]["id"] + log(f" {key}: m2m {f} scanned to id {last:,} ({len(rows):,} rows)") + if rows: + con.executemany(f"UPDATE {key} SET {base}_id=?, {f}=? WHERE id=?", rows) + con.execute("INSERT OR REPLACE INTO _sync_state VALUES (?,?,?,?,?,?)", + [marker, "done", 0, None, len(rows), time.strftime("%Y-%m-%d %H:%M:%S")]) + log(f" {key}: m2m backfill {f} -> {len(rows):,} rows updated") + + +def _upsert(con, key, spec, recs): + if not recs: + return + rows = [_flatten(r, spec["fields"], spec.get("m2m") or ()) for r in recs] + colnames = [c.split()[0] for c in _cols(spec)] + ids = [r["id"] for r in rows] + con.execute(f"DELETE FROM {key} WHERE id IN ({','.join(map(str, ids))})") + con.executemany( + f"INSERT INTO {key} ({', '.join(colnames)}) VALUES ({', '.join('?' for _ in colnames)})", + [[r.get(c) for c in colnames] for r in rows]) + + +def _state(con, key): + row = con.execute("SELECT phase, last_id, cursor_wd, rows FROM _sync_state WHERE entity=?", + [key]).fetchone() + return {"phase": row[0], "last_id": row[1], "cursor_wd": row[2], "rows": row[3]} if row else None + + +def _save_state(con, key, phase, last_id, cursor_wd): + n = con.execute(f"SELECT count(*) FROM {key}").fetchone()[0] + con.execute("""INSERT OR REPLACE INTO _sync_state VALUES (?,?,?,?,?,?)""", + [key, phase, last_id, cursor_wd, n, time.strftime("%Y-%m-%d %H:%M:%S")]) + + +def _base_domain(spec): + return [("active", "in", [True, False])] if spec["archivable"] else [] + + +def sync_entity(key, max_batches=200, batch=BATCH, log=print): + """One bounded, resumable sync pass for an entity. Backfill (by id) → live (by write_date). + Returns (phase, pulled_rows). + + ⭐ D-10: refuses BEFORE opening a cursor or touching Odoo while the connector is paused. The + guard is here — the innermost function that reads from `O` — rather than only in `sync_all`, + so a caller that syncs ONE entity is covered by construction instead of by remembering to + add the same check. `sync_all` short-circuits too, but only to avoid eight identical notices. + """ + if source_paused(): + _paused_notice(f"sync of {key}", log) + return PAUSED_PHASE, 0 + spec = ENTITIES[key] + con = connect() + _ensure_table(con, key, spec) + _ensure_columns(con, key, spec) + _backfill_columns(con, key, spec, log=log) + _backfill_m2m(con, key, spec, log=log) + st = _state(con, key) or {"phase": "backfill", "last_id": 0, "cursor_wd": None, "rows": 0} + fields = spec["fields"] + list(spec.get("m2m") or []) + pulled = 0 + try: + if st["phase"] == "backfill": + last_id = st["last_id"] or 0 + for i in range(max_batches): + dom = _base_domain(spec) + [("id", ">", last_id)] + recs = O.search_read(spec["model"], dom, ["id"] + fields, + limit=batch, order="id asc") + if not recs: + # backfill complete → switch to live mode anchored at max write_date seen + wd = con.execute(f"SELECT max(write_date) FROM {key}").fetchone()[0] + _save_state(con, key, "live", last_id, wd or "1970-01-01 00:00:00") + log(f" {key}: BACKFILL COMPLETE ({st['rows'] + pulled:,} rows) -> live mode") + return "live", pulled + _upsert(con, key, spec, recs) + last_id = recs[-1]["id"] + pulled += len(recs) + _save_state(con, key, "backfill", last_id, None) + if (i + 1) % 10 == 0: + log(f" {key}: backfill …{pulled:,} rows (id≤{last_id})") + log(f" {key}: backfill PAUSED at {pulled:,} rows this run (bounded; resumes)") + return "backfill", pulled + # live mode: write_date >= cursor (idempotent overlap; upsert dedupes). MASS-STAMP + # FALLBACK (found live 2026-07-17): a batch job can stamp >batch rows with ONE + # write_date second (an Odoo recompute stamped ~88k account_move rows '2026-07-15 + # 16:35:03'; stored values carry microseconds, XML-RPC strings don't) — then the + # write_date cursor can NEVER advance and every pass re-pulls the same first page + # forever. When a full batch leaves the cursor unchanged we switch to an ID-WALK over + # write_date >= cursor (order id asc, id > tie), checkpointed as 'cursor|' in + # cursor_wd so a killed walk resumes; on completion the cursor jumps one second past + # the stamp (safe: the walk covered every row in that second, and the stamp is in the + # past — guarded by a 2-minute recency check). + raw = st["cursor_wd"] or "1970-01-01 00:00:00" + cursor, _pipe, _tie = raw.partition("|") + walking, tie_id = bool(_pipe), int(_tie or 0) + walk_max = cursor + for _ in range(max_batches): + if walking: + dom = _base_domain(spec) + [("write_date", ">=", cursor), ("id", ">", tie_id)] + recs = O.search_read(spec["model"], dom, ["id"] + fields, + limit=batch, order="id asc") + if not recs: # walk complete → advance past the stamp + try: + wm = dt.datetime.strptime(walk_max, "%Y-%m-%d %H:%M:%S") + if wm < dt.datetime.utcnow() - dt.timedelta(minutes=2): + cursor = (wm + dt.timedelta(seconds=1)).strftime("%Y-%m-%d %H:%M:%S") + else: + cursor = walk_max + except ValueError: + cursor = walk_max + walking = False + log(f" {key}: id-walk complete -> cursor {cursor}") + break + _upsert(con, key, spec, recs) + pulled += len(recs) + tie_id = recs[-1]["id"] + walk_max = max(walk_max, max(str(r["write_date"]) for r in recs)) + _save_state(con, key, "live", st["last_id"], f"{cursor}|{tie_id}") + continue + dom = _base_domain(spec) + [("write_date", ">=", cursor)] + recs = O.search_read(spec["model"], dom, ["id"] + fields, + limit=batch, order="write_date asc, id asc") + if not recs or (len(recs) == 1 and str(recs[0].get("write_date")) == cursor + and pulled == 0 and _already(con, key, recs[0])): + break + new_cursor = str(recs[-1]["write_date"]) + if new_cursor == cursor and len(recs) == batch: + # full batch stuck on one second: enter the id-walk from id 0 (the write_date + # ordering shuffles ids within the stamp second, so the batch just pulled is + # NOT an id prefix — restart coverage by id; upserts make the overlap free) + walking, tie_id, walk_max = True, 0, cursor + _save_state(con, key, "live", st["last_id"], f"{cursor}|0") + log(f" {key}: mass-stamp second at {cursor} -> switching to id-walk") + continue + _upsert(con, key, spec, recs) + pulled += len(recs) + if new_cursor == cursor and len(recs) < batch: + cursor = new_cursor + break + cursor = new_cursor + # Checkpoint the cursor EVERY batch: a big delta (88k account_move rows after the + # 2026-07-15 recompute) means a killed pass otherwise re-pulls everything — + # upserts are idempotent, but the wasted RPC is real (learned 2026-07-17). + _save_state(con, key, "live", st["last_id"], cursor) + _save_state(con, key, "live", st["last_id"], + f"{cursor}|{tie_id}" if walking else cursor) + if pulled: + log(f" {key}: live sync +{pulled:,} rows (cursor {cursor}" + + (f" walking id>{tie_id}" if walking else "") + ")") + return "live", pulled + finally: + con.close() + + +def _already(con, key, rec): + return bool(con.execute(f"SELECT 1 FROM {key} WHERE id=?", [rec["id"]]).fetchone()) + + +def sync_all(entities=None, max_batches=200, log=print): + # ⭐ D-10. `{}` rather than a dict of paused phases, and the caller decides why that matters: + # `api/main.py:354` loops up to 12 times until `all(phase == "live")`, and `all()` over an + # empty dict is True — so a paused store costs ONE pass instead of twelve rounds of the same + # refusal. A dict of `"paused"` phases would fail that test every time and turn the boot + # sprint into a busy-wait against a connector nobody has resumed. + if source_paused(): + _paused_notice("sync", log) + return {} + out = {} + for key in (entities or list(ENTITIES)): + t0 = time.time() + phase, pulled = sync_entity(key, max_batches=max_batches, log=log) + out[key] = {"phase": phase, "pulled": pulled, "secs": round(time.time() - t0, 1)} + # A full pass that lands every entity in live mode latches process readiness in-process — so + # readers never race the writer's lock to (re)discover the store is usable. + if entities is None and out and all(v["phase"] == "live" for v in out.values()): + mark_ready() + return out + + +def reconcile_deletes(entities=None, log=print): + """Remove store rows hard-deleted in Odoo. unlink() doesn't move write_date, so the cursor + sync can never see a deletion — the count parity DETECTS the drift (by design); this repairs + it. Pulls the full live id set per entity in bounded id-ascending pages (id-only reads are + cheap) and deletes store ids not present live. Run from sync_runner / maintenance, not the + app's frequent passes. + + ⭐ D-10: it pulls the FULL live id set per entity, so it is the single largest Odoo read in + this module and it runs unattended at boot and every fourth resync pass. Paused means paused. + """ + if source_paused(): + _paused_notice("delete reconcile", log) + return {} + con = connect() + try: + out = {} + for key in (entities or list(ENTITIES)): + spec = ENTITIES[key] + live_ids, last = set(), 0 + while True: + recs = O.search_read(spec["model"], _base_domain(spec) + [("id", ">", last)], + ["id"], limit=50000, order="id asc") + if not recs: + break + live_ids.update(r["id"] for r in recs) + last = recs[-1]["id"] + store_ids = {r[0] for r in con.execute(f"SELECT id FROM {key}").fetchall()} + dead = store_ids - live_ids + if dead: + con.execute("CREATE TEMP TABLE IF NOT EXISTS _dead(id BIGINT)") + con.execute("DELETE FROM _dead") + con.executemany("INSERT INTO _dead VALUES (?)", [[i] for i in dead]) + con.execute(f"DELETE FROM {key} WHERE id IN (SELECT id FROM _dead)") + con.execute("DELETE FROM _dead") + n = con.execute(f"SELECT count(*) FROM {key}").fetchone()[0] + con.execute("UPDATE _sync_state SET rows=? WHERE entity=?", [n, key]) + out[key] = len(dead) + log(f" {key}: reconciled {len(dead):,} hard-deleted rows") + return out + finally: + con.close() + + +def status(): + con = connect() + try: + return {r[0]: {"phase": r[1], "cursor": r[3], "rows": r[4], "updated": r[5]} + for r in con.execute("SELECT * FROM _sync_state").fetchall()} + finally: + con.close() + + +# ------------------------------------------------------------------ parity (store vs LIVE Odoo) + +def _check(name, ours, live, tol=0.01): + gap = abs((ours or 0) - (live or 0)) + return {"check": name, "ok": gap <= tol, "gap": round(gap, 4), "ours": ours, "live": live} + + +def validate(pre=None, months=("2026-01-01", "2026-07-01")): + """Raw-fidelity parity: the store must mirror live Odoo — row counts per synced entity + + unscoped monetary sums over a window, to the cent. Only checks entities in LIVE phase. + + ⭐ D-10 REFUSES LOUDLY HERE rather than skipping quietly, and the difference matters. Every + other paused path has a correct silent answer — serve the mirror — because the mirror is the + thing being asked for. This function's ENTIRE contract is "compare me against live Odoo", so + a paused version of it has no honest result: skipping would return an empty pass, and + proceeding would make the live call the pause forbids. An operator running a reconciliation + against a paused connector has asked a question that cannot be answered, and being told so is + the only outcome that is not a lie. (Operator-invoked only — nothing in the serving path + calls it, so this raises for a person, never inside a request.) + """ + if source_paused(): + raise RuntimeError( + "the tenant's Odoo connector is PAUSED, so the store cannot be validated against " + "live Odoo - resume it under Settings > Connectors and run this again") + con = connect() + out = [] + try: + live_state = status() + d0, d1 = months + for key, spec in ENTITIES.items(): + if live_state.get(key, {}).get("phase") != "live": + continue + ours = con.execute(f"SELECT count(*) FROM {key}").fetchone()[0] + live = O.get_odoo().search_count(spec["model"], _base_domain(spec)) + out.append(_check(f"store.{key} row count == live search_count", ours, live, tol=0)) + if live_state.get("sale_order_line", {}).get("phase") == "live": + ours = con.execute( + "SELECT coalesce(sum(l.price_subtotal),0) FROM sale_order_line l " + "JOIN sale_order o ON o.id = l.order_id " + "WHERE o.date_order >= ? AND o.date_order < ?", [d0, d1]).fetchone()[0] + live = O.sum_field("sale.order.line", + [("order_id.date_order", ">=", f"{d0} 00:00:00"), + ("order_id.date_order", "<", f"{d1} 00:00:00")], + "price_subtotal") + out.append(_check(f"store Σ line revenue [{d0}..{d1}) == live (unscoped)", ours, live)) + if live_state.get("sale_order", {}).get("phase") == "live": + ours = con.execute( + "SELECT coalesce(sum(amount_untaxed),0) FROM sale_order " + "WHERE date_order >= ? AND date_order < ?", [d0, d1]).fetchone()[0] + live = O.sum_field("sale.order", + [("date_order", ">=", f"{d0} 00:00:00"), + ("date_order", "<", f"{d1} 00:00:00")], "amount_untaxed") + out.append(_check(f"store Σ order amount_untaxed [{d0}..{d1}) == live (unscoped)", + ours, live)) + finally: + con.close() + return out + + +if __name__ == "__main__": + args = sys.argv[1:] + if args and args[0] == "status": + for k, v in status().items(): + print(f" {k:18s} {v['phase']:9s} rows={v['rows']:>9,} cursor={v['cursor']} ({v['updated']})") + elif args and args[0] == "reconcile": + ents = args[1].split(",") if len(args) > 1 else None + for k, n in reconcile_deletes(ents).items(): + print(f" {k:18s} -{n:,} hard-deleted rows") + elif args and args[0] == "validate": + ok = True + for c in validate(): + mark = "OK " if c["ok"] else "XX " + ok &= c["ok"] + print(f" {mark}{c['check'][:80]} gap={c['gap']}") + print("PARITY:", "GREEN" if ok else "RED") + sys.exit(0 if ok else 1) + else: + ents = args[0].split(",") if args else None + mb = int(args[1]) if len(args) > 1 else 200 + for k, v in sync_all(ents, max_batches=mb).items(): + print(f" {k:18s} {v['phase']:9s} +{v['pulled']:,} rows in {v['secs']}s")