platform/harness/datastore.py

"""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: <app root>/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 <name>_id + <name>_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.
        "fields": ["name", "team_id", "city", "state_id", "country_id", "active",
                   "agent", "salesman_as_agent", "customer_rank",
                   "street", "street2", "zip", "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.
        "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)
        # m2m fields land as <base>_id (the FIRST linked id — the Customers-module convention:
        # a customer's assigned agent is agent_ids[0], ~one per customer, MECE) + <field> 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 ('<key>.m2m.<field>') — 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|<id>' 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")