"""Collections / AR module — receivables, aging, DSO, exposures, follow-up status,
and the reconciliation gap (open-invoice total vs Odoo's partner receivable).

AR is company-level (account.move), not team-scoped. Excluded accounts removed. Aging is built
from open customer invoices/credit notes (signed residual nets credit notes), so buckets
sum to the open-doc AR total. DSO uses LTM gross invoiced sales.
"""
import sys
import math
import datetime as dt
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import core.odoo as O
import core.periods as P
import core.store as store

BUCKETS = ['Current', '1-30', '31-60', '61-90', '90+']

#: Store key for the NIGHTLY days-to-pay snapshot (owner, 2026-07-27).
#:
#: `days_to_pay()` costs ~300s — it walks three years of reconciled receivable lines and their
#: full-reconcile groups to find each invoice's SETTLEMENT date. That is fine for a page somebody
#: opens on purpose, and unacceptable inside `customer_data.pool()`, which every Customer-table
#: render and every Space container start depends on. So the expensive half runs on a schedule
#: and the table reads the ANSWER.
#:
#: Shape: {'computed': 'YYYY-MM-DD HH:MM', 'years': 3, 'days': {'<pid>': avg_days|None}}.
#: JSON object keys are strings, which is why the reader coerces back to int.
DTP_KEY = 'ar_days_to_pay'
#: Refresh when the snapshot is older than this. 20h rather than 24 so a daily cadence never
#: skips a day by drifting a few minutes later each time.
DTP_MAX_AGE_HOURS = 20


def days_to_pay_snapshot():
    """`{pid: avg_days}` from the last snapshot — CHEAP, and it never raises.

    Returns `{}` when the snapshot is absent, unreadable or malformed. A blank column is the
    correct degradation: the alternative is a Customer table that will not render because a
    derived statistic is missing, and `days_to_pay` is a statistic, not a fact the page is about.
    """
    try:
        snap = store.get(DTP_KEY) or {}
        days = snap.get('days') or {}
        out = {}
        for k, v in days.items():
            try:
                out[int(k)] = None if v is None else float(v)
            except (TypeError, ValueError):
                continue
        return out
    except Exception:
        return {}


def days_to_pay_computed_at():
    """When the snapshot was taken ('' if there is none) — so the column can say how old it is."""
    try:
        return str((store.get(DTP_KEY) or {}).get('computed') or '')
    except Exception:
        return ''


def refresh_days_to_pay_snapshot(max_age_hours=DTP_MAX_AGE_HOURS, t=None, force=False):
    """Recompute + persist the snapshot IF it is stale. Returns what it did, and never raises.

    Called from the app's existing background store-sync thread, so the ~300s cost is paid off
    the render path by whichever container happens to be awake. Deliberately NOT called from
    `pool()`: a lazy compute would just move the five minutes onto whoever opened the page first.

    ⚠ Writes through `store.put` on a key nothing else owns. It is a DERIVED artifact — losing it
    costs one recompute and nothing else, which is why it is safe to overwrite wholesale rather
    than read-modify-write.
    """
    if not store.available():
        return {'skipped': 'no store'}
    try:
        if not force:
            at = days_to_pay_computed_at()
            if at:
                age = dt.datetime.now() - dt.datetime.strptime(at, '%Y-%m-%d %H:%M')
                if age < dt.timedelta(hours=max_age_hours):
                    return {'skipped': 'fresh', 'age_h': round(age.total_seconds() / 3600, 1)}
    except Exception:
        pass                       # an unparseable stamp means "recompute", never "crash"
    try:
        res = days_to_pay(t)
        days = {str(r['pid']): r.get('avg_days') for r in (res.get('_all_rows') or [])}
        store.put(DTP_KEY, {'computed': dt.datetime.now().strftime('%Y-%m-%d %H:%M'),
                            'years': 3, 'days': days})
        return {'refreshed': len(days)}
    except Exception as e:
        return {'error': str(e)[:200]}


#: The OVERDUE buckets, in order — `BUCKETS` minus 'Current'. Wave 17 R2 gives each of these a
#: column on the Customer grid, so the vocabulary must have exactly one home: a second list
#: spelled '31_60' somewhere else is how two surfaces start disagreeing about what "60 days"
#: means. Derived from BUCKETS rather than retyped.
OVERDUE_BUCKETS = [b for b in BUCKETS if b != 'Current']


def _bucket(days):
    if days <= 0:
        return 'Current'
    if days <= 30:
        return '1-30'
    if days <= 60:
        return '31-60'
    if days <= 90:
        return '61-90'
    return '90+'


def _open_docs(t=None):
    o = O.get_odoo()
    dom = [('move_type', 'in', ['out_invoice', 'out_refund']), ('state', '=', 'posted'),
           ('payment_state', 'in', ['not_paid', 'partial'])]
    ex = O.excluded_partner_ids()
    if ex:
        dom.append(('partner_id', 'not in', list(ex)))
    docs = o.search_read('account.move', dom,
                         ['name', 'partner_id', 'move_type', 'invoice_date', 'invoice_date_due',
                          'amount_total', 'amount_residual_signed'])
    today = t or P.today()
    for d in docs:
        due = d.get('invoice_date_due')
        try:
            d['days_overdue'] = max((today - dt.date.fromisoformat(due)).days, 0) if due else 0
        except Exception:
            d['days_overdue'] = 0
        d['bucket'] = _bucket(d['days_overdue'])
        d['open'] = d.get('amount_residual_signed') or 0.0
    return docs


def _ltm_invoiced(t=None):
    """LTM gross invoiced sales (out_invoice − out_refund, amount_total) for DSO."""
    o = O.get_odoo()
    lf, lt = P.ltm(t)
    base = [('state', '=', 'posted'), ('invoice_date', '>=', lf), ('invoice_date', '<=', lt)]
    ex = O.excluded_partner_ids()
    if ex:
        base.append(('partner_id', 'not in', list(ex)))
    inv = O.sum_field('account.move', base + [('move_type', '=', 'out_invoice')], 'amount_total')
    ref = O.sum_field('account.move', base + [('move_type', '=', 'out_refund')], 'amount_total')
    return inv - ref


def summary(t=None):
    docs = _open_docs(t)
    total_ar = sum(d['open'] for d in docs)
    overdue = sum(d['open'] for d in docs if d['days_overdue'] > 0)
    ltm_sales = _ltm_invoiced(t)
    dso = (total_ar / (ltm_sales / 365.0)) if ltm_sales else None
    # reconciliation gap vs Odoo partner receivable (company-level)
    ex = set(O.excluded_partner_ids())
    partner_recv = sum(p['credit'] for p in O.search_read('res.partner',
                       [('credit', '>', 0)] + ([('id', 'not in', list(ex))] if ex else []), ['credit']))
    n_customers = len({O.m2o_id(d['partner_id']) for d in docs})
    return {
        'total_ar_open': total_ar,
        'overdue': overdue,
        'pct_overdue': (overdue / total_ar * 100) if total_ar else 0,
        'dso_days': dso,
        'ltm_invoiced': ltm_sales,
        'open_customers': n_customers,
        'partner_receivable': partner_recv,
        'reconciliation_gap': partner_recv - total_ar,
    }


def aging(t=None):
    docs = _open_docs(t)
    out = {b: 0.0 for b in BUCKETS}
    for d in docs:
        out[d['bucket']] += d['open']
    total = sum(out.values()) or 1.0
    return [{'bucket': b, 'amount': out[b], 'pct': out[b] / total * 100} for b in BUCKETS]


def top_exposures(t=None, limit=30):
    docs = _open_docs(t)
    by = {}
    for d in docs:
        pid = O.m2o_id(d['partner_id'])
        nm = O.m2o_name(d['partner_id'])
        e = by.setdefault(pid, {'customer': nm, 'open': 0.0, 'overdue': 0.0, 'oldest': 0, 'docs': 0})
        e['open'] += d['open']
        e['docs'] += 1
        if d['days_overdue'] > 0:
            e['overdue'] += d['open']
            e['oldest'] = max(e['oldest'], d['days_overdue'])
    rows = sorted(by.values(), key=lambda x: -x['overdue'])
    return rows[:limit]


def followup_status_mix(t=None):
    """Mix of res.partner.followup_status among customers carrying receivable.
    followup_status is a non-stored computed field, so we count client-side."""
    ex = set(O.excluded_partner_ids())
    dom = [('credit', '>', 1)] + ([('id', 'not in', list(ex))] if ex else [])
    parts = O.search_read('res.partner', dom, ['followup_status'])
    counts = {}
    for p in parts:
        s = (p.get('followup_status') or '—').replace('_', ' ')
        counts[s] = counts.get(s, 0) + 1
    return [{'status': k, 'customers': v} for k, v in
            sorted(counts.items(), key=lambda kv: -kv[1])]


def reconciliation_flags(t=None, limit=30, threshold=50.0):
    """Customers whose Odoo receivable (credit) diverges from their open-doc total —
    indicates unapplied payments/credits. Surfaced during the collections work."""
    docs = _open_docs(t)
    open_by = {}
    for d in docs:
        pid = O.m2o_id(d['partner_id'])
        open_by[pid] = open_by.get(pid, 0.0) + d['open']
    ex = set(O.excluded_partner_ids())
    parts = O.search_read('res.partner',
                          [('credit', '>', 0)] + ([('id', 'not in', list(ex))] if ex else []),
                          ['name', 'credit'])
    rows = []
    for p in parts:
        gap = (p['credit'] or 0) - open_by.get(p['id'], 0.0)
        if abs(gap) > threshold:
            rows.append({'customer': p['name'], 'odoo_receivable': p['credit'],
                         'open_docs': open_by.get(p['id'], 0.0), 'gap': gap})
    rows.sort(key=lambda x: -abs(x['gap']))
    return rows[:limit]


def credit_exposure(t=None, grace_days=5, limit=40):
    """Forward credit exposure per customer (OCA account_financial_risk semantics, re-implemented):
    draft invoices + open-not-overdue + overdue-past-grace residuals + confirmed-uninvoiced order
    value, vs the partner credit limit (when set). The number to check BEFORE accepting the next
    PO from a shaky dealer. Company-level, GIFTWARE excluded."""
    docs = _open_docs(t)
    ex = O.excluded_partner_ids()
    exdom = [('partner_id', 'not in', list(ex))] if ex else []

    per = {}

    def _e(pid, name):
        return per.setdefault(pid, {'pid': pid, 'customer': name, 'draft': 0.0, 'open': 0.0,
                                    'overdue': 0.0, 'uninvoiced': 0.0, 'credit_limit': 0.0,
                                    # Wave 17 R2 — the AGING SPLIT of `overdue`, per partner.
                                    # Same loop, same documents, no extra Odoo call: this is the
                                    # decomposition the Customer grid needs to REPLACE the
                                    # Collections page ("who has money in 90+" is the whole job).
                                    **{f'aged_{b}': 0.0 for b in OVERDUE_BUCKETS}})

    for d in docs:
        e = _e(O.m2o_id(d['partner_id']), O.m2o_name(d['partner_id']))
        if d['days_overdue'] > grace_days:
            e['overdue'] += d['open']
            # ⚠ BUCKETED ONLY PAST THE GRACE PERIOD, deliberately — so the four buckets SUM
            # EXACTLY to `overdue` and the grid's columns reconcile to its own total. Using
            # `_bucket` on every doc instead would file days 1..grace under '1-30' while
            # `overdue` excluded them, and the decomposition would be quietly short.
            e[f'aged_{_bucket(d["days_overdue"])}'] += d['open']
        else:
            e['open'] += d['open']

    for r in O.search_read('account.move',
                           [('move_type', 'in', ['out_invoice', 'out_refund']),
                            ('state', '=', 'draft')] + exdom,
                           ['partner_id', 'amount_total_signed']):
        if r.get('partner_id'):
            _e(O.m2o_id(r['partner_id']), O.m2o_name(r['partner_id']))['draft'] += \
                r.get('amount_total_signed') or 0.0

    uexdom = [('order_partner_id', 'not in', list(ex))] if ex else []
    for g in O.read_group('sale.order.line',
                          [('order_id.state', 'in', ['sale', 'done']),
                           ('untaxed_amount_to_invoice', '!=', 0)] + uexdom,
                          ['untaxed_amount_to_invoice'], ['order_partner_id']):
        p = g.get('order_partner_id')
        if p:
            _e(O.m2o_id(p), O.m2o_name(p))['uninvoiced'] += \
                g.get('untaxed_amount_to_invoice') or 0.0

    try:  # credit_limit only exists/means something when the feature is enabled — degrade quietly
        for r in O.search_read('res.partner', [('id', 'in', list(per.keys()))], ['credit_limit']):
            if per.get(r['id']) is not None:
                per[r['id']]['credit_limit'] = r.get('credit_limit') or 0.0
    except Exception:
        pass

    rows_all = []
    for e in per.values():
        e['exposure'] = e['draft'] + e['open'] + e['overdue'] + e['uninvoiced']
        e['headroom'] = (e['credit_limit'] - e['exposure']) if e['credit_limit'] else None
        e['flag'] = bool(e['credit_limit']) and e['exposure'] > e['credit_limit']
        rows_all.append(e)
    rows_all.sort(key=lambda x: -x['exposure'])
    # Display rows drop ~zero exposures; _all_rows keeps EVERY customer (incl. net-credit
    # balances) — validation sums must run over the unfiltered set or the filter biases them.
    rows = [r for r in rows_all if abs(r['exposure']) > 1]
    return {'rows': rows[:limit], 'n_flagged': sum(1 for r in rows if r['flag']),
            'total_exposure': sum(r['exposure'] for r in rows),
            'grace_days': grace_days, '_all_rows': rows_all}


def days_to_pay(t=None, years=3, limit=30):
    """Per-customer average days from invoice to SETTLEMENT (OCA partner_time_to_pay semantics):
    settlement date = the date the receivable line became fully reconciled (max counterpart line
    date in its full-reconcile group), NOT the payment document date. Windows: lifetime (bounded
    to `years`), last calendar year, this year — by invoice date. Joined with current open AR so
    the table reads 'who owes us AND how do they behave'."""
    o = O.get_odoo()
    today = t or P.today()
    since = (today - dt.timedelta(days=365 * years)).isoformat()
    ex = O.excluded_partner_ids()
    dom = [('move_id.move_type', '=', 'out_invoice'), ('move_id.state', '=', 'posted'),
           ('account_id.account_type', '=', 'asset_receivable'),
           ('full_reconcile_id', '!=', False), ('date', '>=', since)]
    if ex:
        dom.append(('partner_id', 'not in', list(ex)))
    inv_lines = O.search_read('account.move.line', dom,
                              ['partner_id', 'date', 'full_reconcile_id', 'move_id'])

    # Payment-term days per term id: prefer the term lines' `days`; fall back to parsing the name.
    term_days = {}
    try:
        terms = O.search_read('account.payment.term', [], ['name'])
        tl = O.search_read('account.payment.term.line', [], ['payment_id', 'days'])
        by_term = {}
        for l in tl:
            k = O.m2o_id(l.get('payment_id'))
            if k is not None:
                by_term[k] = max(by_term.get(k, 0), l.get('days') or 0)
        import re as _re
        for tm in terms:
            d = by_term.get(tm['id'])
            if d is None:
                m = _re.search(r'(\d+)', tm['name'] or '')
                d = int(m.group(1)) if m and 'day' in (tm['name'] or '').lower() else 0
            term_days[tm['id']] = {'name': tm['name'], 'days': d}
    except Exception:
        term_days = {}

    # Per-invoice terms + totals (12m window for the free-credit estimate).
    y365_iso = (today - dt.timedelta(days=365)).isoformat()
    move_ids = list({O.m2o_id(l['move_id']) for l in inv_lines if l.get('move_id')})
    move_info = {}
    for i in range(0, len(move_ids), 5000):
        for mv in O.search_read('account.move', [('id', 'in', move_ids[i:i + 5000])],
                                ['invoice_payment_term_id', 'amount_total', 'invoice_date',
                                 'partner_id']):
            move_info[mv['id']] = mv

    fr_ids = list({O.m2o_id(l['full_reconcile_id']) for l in inv_lines
                   if l.get('full_reconcile_id')})
    settle = {}
    for i in range(0, len(fr_ids), 5000):
        chunk = fr_ids[i:i + 5000]
        try:  # grouped max-date per reconcile (one call per chunk)
            for g in O.read_group('account.move.line',
                                  [('full_reconcile_id', 'in', chunk)],
                                  ['date:max'], ['full_reconcile_id']):
                k = O.m2o_id(g.get('full_reconcile_id'))
                d = g.get('date') or g.get('date:max') or ''
                d = str(d)[:10]
                if k and d:
                    settle[k] = max(settle.get(k, ''), d)
        except Exception:  # fallback: raw lines
            for l in o.search_read('account.move.line',
                                   [('full_reconcile_id', 'in', chunk)],
                                   ['full_reconcile_id', 'date']):
                k = O.m2o_id(l['full_reconcile_id'])
                d = str(l.get('date') or '')[:10]
                if k and d:
                    settle[k] = max(settle.get(k, ''), d)

    y0 = dt.date(today.year, 1, 1).isoformat()
    ly0 = dt.date(today.year - 1, 1, 1).isoformat()
    per = {}
    violations = 0
    for l in inv_lines:
        k = O.m2o_id(l.get('full_reconcile_id'))
        inv_d = str(l.get('date') or '')[:10]
        pay_d = settle.get(k, '')
        if not inv_d or not pay_d:
            continue
        days = (dt.date.fromisoformat(pay_d) - dt.date.fromisoformat(inv_d)).days
        if days < 0:
            violations += 1
            continue
        pid = O.m2o_id(l['partner_id'])
        e = per.setdefault(pid, {'pid': pid, 'customer': O.m2o_name(l['partner_id']),
                                 'all': [], 'ly': [], 'ytd': []})
        e['all'].append(days)
        if inv_d >= y0:
            e['ytd'].append(days)
        elif inv_d >= ly0:
            e['ly'].append(days)

    open_by = {}
    for d in _open_docs(t):
        pid = O.m2o_id(d['partner_id'])
        oe = open_by.setdefault(pid, {'open': 0.0, 'overdue': 0.0})
        oe['open'] += d['open']
        if d['days_overdue'] > 0:
            oe['overdue'] += d['open']

    def _avg(v):
        return (sum(v) / len(v)) if v else None

    # Terms + 12m invoiced per partner (dominant term by invoiced $) → compliance/free-credit.
    pt_terms, pt_inv12 = {}, {}
    for mid, mv in move_info.items():
        pid = O.m2o_id(mv.get('partner_id'))
        tid = O.m2o_id(mv.get('invoice_payment_term_id'))
        amt = mv.get('amount_total') or 0.0
        d = pt_terms.setdefault(pid, {})
        d[tid] = d.get(tid, 0.0) + amt
        if str(mv.get('invoice_date') or '')[:10] >= y365_iso:
            pt_inv12[pid] = pt_inv12.get(pid, 0.0) + amt

    rows = []
    total_free_credit = 0.0
    for pid, e in per.items():
        ob = open_by.get(pid, {'open': 0.0, 'overdue': 0.0})
        a_ly, a_ytd = _avg(e['ly']), _avg(e['ytd'])
        a_all = _avg(e['all'])
        dom_tid = max(pt_terms.get(pid, {None: 0}).items(), key=lambda kv: kv[1])[0]
        tinfo = term_days.get(dom_tid, {'name': '(none)', 'days': 0})
        eff = a_ytd if a_ytd is not None else a_all
        excess = max(0.0, (eff or 0) - tinfo['days']) if eff is not None else None
        inv12 = pt_inv12.get(pid, 0.0)
        free_credit = (excess / 365.0 * inv12) if excess else 0.0
        total_free_credit += free_credit or 0.0
        rows.append({'pid': pid, 'customer': e['customer'], 'open': ob['open'], 'overdue': ob['overdue'],
                     'paid_invoices': len(e['all']), 'avg_days': a_all,
                     'avg_days_ly': a_ly, 'avg_days_ytd': a_ytd,
                     'improvement_days': (a_ly - a_ytd) if (a_ly is not None and a_ytd is not None) else None,
                     'terms': tinfo['name'], 'term_days': tinfo['days'],
                     'excess_days': excess, 'free_credit': free_credit,
                     'invoiced_12m': inv12})
    rows.sort(key=lambda x: -x['open'])
    return {'rows': rows[:limit], 'n_customers': len(per), 'violations': violations,
            'since': since, 'total_free_credit': total_free_credit,
            '_rowsum_inv12': sum(pt_inv12.values()), '_all_rows': rows}


def credit_limit_proposals(exposure, behavior):
    """PROPOSED credit limits for customers trading on open terms with NO limit set (only 22 of
    3,595 partners have one — the field is effectively unmaintained). Trade-credit heuristic:
    limit ≈ (12m invoiced / 365) × (term days + 30 review buffer), tiered by observed payment
    behavior (pays within terms +5d → ×1.25 · chronic 15d+ overrun → ×0.75), rounded UP to $500,
    floor $1,000. READ-ONLY: an export worklist for the owner to enter in Odoo — once limits are
    in, the exposure table's breach flag and Odoo's own credit hold both come alive."""
    beh = {r['pid']: r for r in behavior.get('_all_rows', []) if r.get('pid')}
    out = []
    for e in exposure.get('_all_rows', []):
        if e.get('credit_limit') or e.get('exposure', 0) <= 0:
            continue
        b = beh.get(e['pid'])
        if not b or (b.get('invoiced_12m') or 0) <= 0:
            continue
        base = b['invoiced_12m'] / 365.0 * ((b.get('term_days') or 0) + 30)
        exd = b.get('excess_days')
        tier = ('on-time' if (exd is not None and exd <= 5)
                else ('slow' if (exd or 0) > 15 else 'normal'))
        mult = {'on-time': 1.25, 'normal': 1.0, 'slow': 0.75}[tier]
        prop = max(1000.0, math.ceil(base * mult / 500.0) * 500.0)
        out.append({'pid': e['pid'], 'customer': e['customer'],
                    'invoiced_12m': b['invoiced_12m'], 'term_days': b.get('term_days') or 0,
                    'excess_days': exd, 'tier': tier, 'exposure': e['exposure'],
                    'proposed_limit': prop,
                    'over_proposed': e['exposure'] > prop})
    out.sort(key=lambda r: -r['exposure'])
    return {'rows': out, 'n': len(out),
            'n_over': sum(1 for r in out if r['over_proposed']),
            'over_value': sum(r['exposure'] - r['proposed_limit']
                              for r in out if r['over_proposed']),
            'with_limit': [e for e in exposure.get('_all_rows', []) if e.get('credit_limit')]}


def validate(t=None, exposure=None, behavior=None):
    """exposure/behavior: pass precomputed credit_exposure()/days_to_pay() results to avoid
    recomputing the heavy pulls when the caller (the page bundle) already has them; validate.py
    calls with no args and computes everything itself."""
    docs = _open_docs(t)
    checks = []
    total = sum(d['open'] for d in docs)
    bucket_sum = sum(a['amount'] for a in aging(t))
    checks.append({
        'check': 'AR aging: Σ(buckets) == total open AR',
        'a': round(bucket_sum, 2), 'b': round(total, 2),
        'gap': round(bucket_sum - total, 2),
        'ok': abs(bucket_sum - total) <= 1.0})

    # WAVE 17 R2 — the aging SPLIT must decompose the overdue total exactly. These four numbers
    # became COLUMNS on the Customer grid this wave (the Collections view is built on them), so
    # a bucket that drifted from `overdue` would be a wrong number on a worklist somebody works
    # from. Σ(buckets) == Σ(overdue) is exact by construction (same loop, same documents) and is
    # asserted rather than assumed, because "by construction" is what every drift was before it
    # happened.
    ce_buckets = exposure or credit_exposure(t)
    bkt = sum(r.get(f'aged_{b}', 0.0) for r in ce_buckets['_all_rows']
              for b in OVERDUE_BUCKETS)
    od = sum(r['overdue'] for r in ce_buckets['_all_rows'])
    checks.append({
        'check': 'AR aging split: Σ(1-30, 31-60, 61-90, 90+) == Σ(overdue)',
        'a': round(bkt, 2), 'b': round(od, 2),
        'gap': round(bkt - od, 2),
        'ok': abs(bkt - od) <= 0.01})

    exp_sum = sum(e['open'] for e in top_exposures(t, limit=10**9))
    checks.append({
        'check': 'AR: Σ(per-customer open) == total open AR',
        'a': round(exp_sum, 2), 'b': round(total, 2),
        'gap': round(exp_sum - total, 2),
        'ok': abs(exp_sum - total) <= 1.0})

    # Credit exposure: per-customer (open + overdue) row-build vs an independent read_group sum
    # of amount_residual_signed over the same unpaid-invoice domain. The two sides are separate
    # RPC snapshots on a LIVE ledger — a payment applied between them shifts the total — so the
    # sides are pulled adjacently and the check carries an explicit live-drift tolerance. A
    # structural bug (double-count, dropped partner, wrong bucket) shows as a %-level gap, far
    # beyond it.
    ex = O.excluded_partner_ids()
    dom = [('move_type', 'in', ['out_invoice', 'out_refund']), ('state', '=', 'posted'),
           ('payment_state', 'in', ['not_paid', 'partial'])]
    if ex:
        dom.append(('partner_id', 'not in', list(ex)))
    resid = O.sum_field('account.move', dom, 'amount_residual_signed')
    ce = exposure or credit_exposure(t)
    ce_open = sum(r['open'] + r['overdue'] for r in ce['_all_rows'])
    tol = max(500.0, abs(resid) * 0.001)
    checks.append({
        'check': 'Exposure: Σ(open+overdue per customer) == read_group Σ residual (live-drift tol)',
        'a': round(ce_open, 2), 'b': round(resid, 2),
        'gap': round(ce_open - resid, 2),
        'ok': abs(ce_open - resid) <= tol})

    # Days-to-pay: settlement can never precede the invoice line date, and coverage exists.
    dtp = behavior or days_to_pay(t)
    checks.append({
        'check': 'Days-to-pay: settlement>=invoice violations == 0 (and customers covered > 0)',
        'a': dtp['violations'], 'b': 0,
        'gap': dtp['violations'],
        'ok': dtp['violations'] == 0 and dtp['n_customers'] > 0})
    return checks