#!/usr/bin/env python3
"""Generate the NexusCRM <-> HubSpot daily sync-reconciliation report.

Data sources (CI-friendly, no interactive OAuth):
  - NexusCRM: TiDB direct SQL against database `nexuscrm`, tables `lead` /
    `opportunity` / `account` (one column per SF field API name). Mirrors the
    precedent in .github/workflows/salesforce-prod-reconcile-readonly.yml.
  - HubSpot: CRM v3 REST API with a Bearer token.

Output: report.md — a step-summary markdown report.

The reconciliation logic (daily record-count ledger, 51-field inbound
content diff with transform/valueMap replay, Account->Company cohort coverage)
mirrors the manual dashboard this replaces.

Local validation note: the TiDB layer cannot be exercised without prod
credentials; preflight checks fail loudly (missing table/column, timezone
mismatch, HubSpot scope) rather than silently producing wrong numbers.
"""
from __future__ import annotations

import argparse
import collections
import datetime as dt
import email.utils
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

BJ = dt.timezone(dt.timedelta(hours=8))
UTC = dt.timezone.utc
UUID_RE = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.I)
SFID_RE = re.compile(r"[a-zA-Z0-9]{15}|[a-zA-Z0-9]{18}")
COL_RE = re.compile(r"^[A-Za-z0-9_]+$")  # whitelist for physical column identifiers

HS_BASE = "https://api.hubapi.com"


# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--tidb-host", required=True)
    p.add_argument("--tidb-port", type=int, default=4000)
    p.add_argument("--tidb-user", required=True)
    # Secrets default from env so they never appear in the process argv
    # (visible via /proc/<pid>/cmdline / `ps` on shared self-hosted runners).
    p.add_argument("--tidb-password", default=os.environ.get("TIDB_PROD_PASSWORD", ""),
                   help="defaults to env TIDB_PROD_PASSWORD")
    p.add_argument("--tidb-database", default="nexuscrm")
    p.add_argument("--hubspot-token", default=os.environ.get("HUBSPOT_TOKEN", ""),
                   help="defaults to env HUBSPOT_TOKEN")
    p.add_argument("--mapping", required=True,
                   help="path to contact_lead_inbound.json")
    p.add_argument("--output-dir", required=True)
    p.add_argument("--now", default="",
                   help="override 'now' as ISO8601 UTC (for deterministic runs); "
                        "default = current UTC time")
    return p.parse_args()


# --------------------------------------------------------------------------- #
# Time windows (Beijing day boundaries expressed in UTC)
# --------------------------------------------------------------------------- #
def resolve_now(raw: str) -> dt.datetime:
    if raw:
        return dt.datetime.fromisoformat(raw.replace("Z", "+00:00")).astimezone(UTC)
    return dt.datetime.now(UTC)


def bj_midnight_utc(day: dt.date) -> dt.datetime:
    """UTC instant of Beijing 00:00 on `day`."""
    return dt.datetime(day.year, day.month, day.day, tzinfo=BJ).astimezone(UTC)


def build_windows(now_utc: dt.datetime) -> dict:
    now_bj = now_utc.astimezone(BJ)
    today_bj = now_bj.date()
    # 7 settled days end at yesterday; window starts 7 days before today
    days = [(today_bj - dt.timedelta(days=n)) for n in range(7, -1, -1)]  # 8 dates, last = today
    # Field-level authoritative cohort = the most recent COMPLETE day
    # (yesterday). HubSpot's async (workflow) attribution fills settle over
    # ~24h, but the empirical effect is small (~1pp fully-settled vs
    # near-zero-settle), so yesterday's complete day is the primary quality
    # read and today's partial day is the trend cohort. At a 09:00 CST run
    # yesterday's latest-hours records are only 9-24h settled — a sub-1pp
    # under-count that the labels disclose rather than claim ">=24h settled".
    auth_bj = today_bj - dt.timedelta(days=1)
    return {
        "now_utc": now_utc,
        "today_bj": today_bj,
        "auth_bj": auth_bj,
        "days": days,  # list[date], last is the partial (today)
        "ledger_lo": bj_midnight_utc(days[0]),
        "ledger_hi": now_utc,
        "today_lo": bj_midnight_utc(today_bj),
        "today_hi": now_utc,
        "auth_lo": bj_midnight_utc(auth_bj),
        "auth_hi": bj_midnight_utc(auth_bj + dt.timedelta(days=1)),
        "acct30_lo": bj_midnight_utc(today_bj - dt.timedelta(days=30)),
        "acct30_hi": now_utc,
        "partial_note": "as of " + now_bj.strftime("%H:%M"),
        "window_label": f"{days[0].isoformat()} → {today_bj.isoformat()} "
                        f"{now_bj.strftime('%H:%M')} (Beijing time)",
        "generated": now_bj.strftime("%Y-%m-%dT%H:%M+08:00"),
    }


def sql_ts(d: dt.datetime) -> str:
    return d.astimezone(UTC).strftime("%Y-%m-%d %H:%M:%S")


def bj_day_of(created) -> str:
    """created is a naive UTC datetime (from TiDB) or aware; return BJ date str."""
    if isinstance(created, str):
        created = dt.datetime.fromisoformat(created.replace("Z", "+00:00"))
    if created.tzinfo is None:
        created = created.replace(tzinfo=UTC)
    return created.astimezone(BJ).strftime("%Y-%m-%d")


# --------------------------------------------------------------------------- #
# TiDB (NexusCRM) layer
# --------------------------------------------------------------------------- #
def connect_tidb(a):
    import pymysql
    return pymysql.connect(
        host=a.tidb_host, port=a.tidb_port, user=a.tidb_user,
        password=a.tidb_password, database=a.tidb_database,
        ssl={"ssl": {}}, charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor, autocommit=True,
        connect_timeout=30, read_timeout=120,
    )


def table_columns(conn, table: str) -> dict:
    """Return {column_name: sf_field_type} using real SQL types from schema."""
    db = conn.db.decode() if isinstance(conn.db, bytes) else conn.db
    with conn.cursor() as cur:
        cur.execute(
            "SELECT COLUMN_NAME, DATA_TYPE, COLUMN_TYPE FROM information_schema.columns "
            "WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s",
            (db, table),
        )
        out = {}
        for r in cur.fetchall():
            out[r["COLUMN_NAME"]] = sqltype_to_ftype(r["DATA_TYPE"], r["COLUMN_TYPE"])
        return out


def sqltype_to_ftype(data_type: str, column_type: str) -> str:
    dt_ = (data_type or "").lower()
    ct_ = (column_type or "").lower()
    if dt_ == "tinyint" and "(1)" in ct_:
        return "boolean"
    if dt_ in ("datetime", "timestamp"):
        return "datetime"
    if dt_ == "date":
        return "date"
    if dt_ in ("double", "float", "decimal", "int", "bigint", "smallint", "mediumint"):
        return "double"
    return "string"


def not_deleted(cols) -> str:
    """SQL clause excluding soft-deleted rows. SFDC IsDeleted is carried into
    the TiDB mirror; without this, records deleted in NexusCRM/SFDC (e.g. a
    cleaned-up duplicate batch) still count and inflate the reconciliation.
    NULL is treated as not-deleted."""
    return " AND COALESCE(`IsDeleted`, 0) = 0" if "IsDeleted" in cols else ""


def nx_fetch_ids(conn, table: str, lo, hi, extra: str = ""):
    with conn.cursor() as cur:
        cur.execute(
            f"SELECT `Id`, `CreatedDate` FROM `{table}` "
            f"WHERE `CreatedDate` >= %s AND `CreatedDate` < %s{extra}",
            (sql_ts(lo), sql_ts(hi)),
        )
        return cur.fetchall()


def nx_fetch_lead_fields(conn, cols, lo, hi, extra: str = ""):
    sel = ", ".join(f"`{c}`" for c in cols)
    with conn.cursor() as cur:
        cur.execute(
            f"SELECT {sel} FROM `lead` "
            f"WHERE `CreatedDate` >= %s AND `CreatedDate` < %s{extra}",
            (sql_ts(lo), sql_ts(hi)),
        )
        return cur.fetchall()


def nx_fetch_accounts(conn, lo, hi, extra: str = ""):
    with conn.cursor() as cur:
        cur.execute(
            "SELECT `Id`, `Name`, `CreatedDate`, `HubSpot_Id__c`, "
            "`HubSpot_Sync_Enabled__c` FROM `account` "
            f"WHERE `CreatedDate` >= %s AND `CreatedDate` < %s{extra}",
            (sql_ts(lo), sql_ts(hi)),
        )
        return cur.fetchall()


# --------------------------------------------------------------------------- #
# HubSpot layer
# --------------------------------------------------------------------------- #
class HubSpotError(RuntimeError):
    def __init__(self, code, body):
        self.code = code
        super().__init__(f"HTTP {code}: {body[:300]}")


class HubSpotReader:
    """Serial, paced reads only; exhausted retries still fail the report.

    Four attempts maximum, with at most 60 seconds per retry delay. A longer
    Retry-After fails closed instead of retrying earlier than HubSpot allows.
    The endpoint restriction matters: CRM search and batch/read use POST,
    but a future mutation must never inherit this read retry policy.
    """
    def __init__(self, base=HS_BASE, sleep=time.sleep, monotonic=time.monotonic,
                 now=lambda: dt.datetime.now(UTC)):
        self.base, self.sleep, self.monotonic, self.now = base, sleep, monotonic, now
        self.next_request = 0.0

    def retry_delay(self, header, attempt):
        fallback = float(2 ** attempt)
        if not header:
            return fallback
        try:
            if re.fullmatch(r"\d+", header.strip()):
                return float(int(header.strip()))
            when = email.utils.parsedate_to_datetime(header)
            if when.tzinfo is None:
                when = when.replace(tzinfo=UTC)
            return max(0.0, (when - self.now()).total_seconds())
        except (ValueError, TypeError, OverflowError):
            return fallback

    def request(self, token, path, method="GET", body=None):
        if method != "POST" or not re.fullmatch(
                r"/crm/v3/objects/(contacts|deals|companies)/(search|batch/read)", path):
            raise ValueError("HubSpot reconciliation permits only CRM search and batch/read")
        req = urllib.request.Request(
            self.base + path, method=method,
            data=json.dumps(body).encode() if body is not None else None,
            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        )
        for attempt in range(4):
            delay = self.next_request - self.monotonic()
            if delay > 0:
                self.sleep(delay)
            # 4 requests/second leaves headroom below the search burst limit.
            self.next_request = self.monotonic() + 0.25
            try:
                with urllib.request.urlopen(req, timeout=60) as response:
                    return json.load(response)
            except urllib.error.HTTPError as e:
                error = HubSpotError(e.code, e.read().decode())
                header = e.headers.get("Retry-After")
                e.close()
                if e.code != 429 or attempt == 3:
                    raise error from None
                delay = self.retry_delay(header, attempt)
                if delay > 60:
                    raise error from None
                self.next_request = max(self.next_request, self.monotonic() + delay)
                print(f"HubSpot read throttled; retry {attempt + 1}/3 in {delay:g}s",
                      file=sys.stderr)
            except urllib.error.URLError as e:
                raise HubSpotError("network", str(e)) from None


_HS_READER = HubSpotReader()


def hs_req(token, path, method="GET", body=None):
    return _HS_READER.request(token, path, method, body)


def hs_preflight(token):
    # Preflight every object the job actually reads (contacts + deals +
    # companies), so a missing read scope fails immediately with a clear
    # message instead of mid-run.
    for obj in ("contacts", "deals", "companies"):
        try:
            hs_req(token, f"/crm/v3/objects/{obj}/search", "POST", {"limit": 1})
        except HubSpotError as e:
            raise SystemExit(
                f"HubSpot preflight failed for {obj} ({e}). If this is a 403, "
                f"the token lacks 'crm.objects.{obj}.read' scope — configure a "
                "read-scoped HubSpot private-app token as the HUBSPOT_TOKEN secret."
            )


def hs_daily_total(token, obj, lo, hi):
    body = {"filterGroups": [{"filters": [{
        "propertyName": "createdate", "operator": "BETWEEN",
        "value": str(int(lo.timestamp() * 1000)),
        "highValue": str(int(hi.timestamp() * 1000)),
    }]}], "limit": 1}
    r = hs_req(token, f"/crm/v3/objects/{obj}/search", "POST", body)
    return r["total"]  # hs_req fails closed, so a successful response has total


def hs_batch_read(token, obj, ids, props, id_property=None):
    """Fetch objects by internal id, or by an alternate idProperty (e.g.
    "email"). For idProperty lookups, HubSpot's batch/read endpoint does not
    echo back which requested identifier resolved a given result: a contact
    can be found via a SECONDARY value (e.g. one of its `hs_additional_emails`,
    not its primary `email`), in which case the response's own `email`
    property differs from what was actually looked up. Keying the output
    solely by the response's primary property value then makes a caller's
    `.get(<the email they looked up>)` silently miss a real match. Register
    every identifier that could have found this result (primary + any
    secondary emails), not just the primary one."""
    out = {}
    req_props = props
    if id_property == "email" and "hs_additional_emails" not in props:
        req_props = props + ["hs_additional_emails"]
    for i in range(0, len(ids), 95):
        chunk = ids[i:i + 95]
        body = {"properties": req_props, "inputs": [{"id": x} for x in chunk]}
        if id_property:
            body["idProperty"] = id_property
        r = hs_req(token, f"/crm/v3/objects/{obj}/batch/read", "POST", body)
        for x in r.get("results", []):
            if not id_property:
                out[x["id"]] = x["properties"]
                continue
            keys = {_s(x["properties"].get(id_property)).lower()}
            if id_property == "email":
                extra = x["properties"].get("hs_additional_emails") or ""
                keys |= {e.strip().lower() for e in extra.split(";") if e.strip()}
            for key in keys:
                if key:
                    out[key] = x["properties"]
    return out


# --------------------------------------------------------------------------- #
# Transform evaluator (LEFT / BLANKVALUE / CASE / TEXT / DATETIMEVALUE / &)
# --------------------------------------------------------------------------- #
class Unsupported(Exception):
    pass


TOKEN = re.compile(r'''\s*(?:
    (?P<str>"(?:[^"\\]|\\.)*")
  | (?P<num>-?\d+(?:\.\d+)?)
  | (?P<rec>\$Record\.[A-Za-z0-9_]+)
  | (?P<fn>[A-Z][A-Z0-9_]*)\s*(?=\()
  | (?P<punc>[(),&])
)''', re.X)
SUPPORTED = {"LEFT", "BLANKVALUE", "CASE", "TEXT", "DATETIMEVALUE"}


def _lex(s):
    out, i = [], 0
    while i < len(s):
        m = TOKEN.match(s, i)
        if not m:
            if s[i].isspace():
                i += 1
                continue
            raise Unsupported("lex")
        i = m.end()
        for k in ("str", "num", "rec", "fn", "punc"):
            if m.group(k):
                out.append((k, m.group(k)))
                break
    return out


class _P:
    def __init__(self, t):
        self.t, self.i = t, 0

    def peek(self):
        return self.t[self.i] if self.i < len(self.t) else (None, None)

    def take(self):
        v = self.peek()
        self.i += 1
        return v

    def expect(self, ch):
        if self.take()[1] != ch:
            raise Unsupported("expect")

    def parse(self):
        n = self.unary()
        while self.peek()[1] == "&":
            self.take()
            n = ("concat", n, self.unary())
        return n

    def unary(self):
        k, v = self.take()
        if k == "str":
            return ("lit", json.loads(v))
        if k == "num":
            return ("lit", float(v) if "." in v else int(v))
        if k == "rec":
            return ("field", v.split(".", 1)[1])
        if k == "fn":
            self.expect("(")
            args = []
            if self.peek()[1] != ")":
                args.append(self.parse())
                while self.peek()[1] == ",":
                    self.take()
                    args.append(self.parse())
            self.expect(")")
            return ("call", v, args)
        raise Unsupported("unary")


def parse_expr(expr):
    p = _P(_lex(expr))
    n = p.parse()
    if p.i != len(p.t):
        raise Unsupported("trailing")
    return n


def validate_expr(node):
    """Check every branch without executing a fabricated source value."""
    if node[0] == "call":
        fn, args = node[1], node[2]
        if fn not in SUPPORTED:
            raise Unsupported(fn)
        arity = {"LEFT": 2, "BLANKVALUE": 2, "TEXT": 1, "DATETIMEVALUE": 1}
        if (fn in arity and len(args) != arity[fn]) or (fn == "CASE" and len(args) < 3):
            raise Unsupported(f"{fn} arity")
        for arg in args:
            validate_expr(arg)
    elif node[0] == "concat":
        validate_expr(node[1])
        validate_expr(node[2])


def _s(v):
    if v is None:
        return ""
    if isinstance(v, tuple):
        return v[1]
    if isinstance(v, bool):
        return "true" if v else "false"
    return str(v)


def ev(node, rec):
    t = node[0]
    if t == "lit":
        return node[1]
    if t == "field":
        v = rec.get(node[1])
        return "" if v is None else v
    if t == "concat":
        return _s(ev(node[1], rec)) + _s(ev(node[2], rec))
    if t == "call":
        fn, args = node[1], node[2]
        if fn not in SUPPORTED:
            raise Unsupported(fn)
        if fn == "LEFT":
            return _s(ev(args[0], rec))[:int(ev(args[1], rec))]
        if fn == "BLANKVALUE":
            v = ev(args[0], rec)
            return v if _s(v) != "" else ev(args[1], rec)
        if fn == "TEXT":
            return _s(ev(args[0], rec))
        if fn == "DATETIMEVALUE":
            return ("__DT__", _s(ev(args[0], rec)))
        if fn == "CASE":
            key = _s(ev(args[0], rec))
            pairs = args[1:]
            i = 0
            while i + 1 < len(pairs):
                if _s(ev(pairs[i], rec)) == key:
                    return ev(pairs[i + 1], rec)
                i += 2
            return ev(pairs[-1], rec) if len(pairs) % 2 == 1 else ""
    raise Unsupported(t)


# --------------------------------------------------------------------------- #
# Value normalisation
# --------------------------------------------------------------------------- #
def norm_bool(v):
    x = _s(v).strip().lower()
    if x in ("true", "1", "yes"):
        return "true"
    if x in ("false", "0", "no", ""):
        return "false"
    return x


def norm_num(v):
    x = _s(v).strip().replace(",", "")
    if x == "":
        return ""
    try:
        f = float(x)
        return str(int(f)) if f == int(f) else f"{f:.4f}"
    except ValueError:
        return x


def norm_dt(v):
    x = _s(v).strip()
    if x == "":
        return ""
    if re.fullmatch(r"\d{10,13}", x):
        n = int(x)
        n = n // 1000 if n > 10 ** 11 else n
        return dt.datetime.utcfromtimestamp(n).strftime("%Y-%m-%dT%H:%M")
    x = x.replace("Z", "+00:00")
    try:
        d = dt.datetime.fromisoformat(x)
        if d.tzinfo:
            d = d.astimezone(UTC).replace(tzinfo=None)
        return d.strftime("%Y-%m-%dT%H:%M")
    except ValueError:
        return x


def norm_str(v):
    return re.sub(r"\s+", " ", _s(v)).strip()


def normalise(v, ftype):
    if ftype == "boolean":
        return norm_bool(v) or "false"
    if ftype in ("datetime", "date"):
        return norm_dt(v)
    if ftype in ("double", "currency", "int", "percent"):
        return norm_num(v)
    return norm_str(v)


# --------------------------------------------------------------------------- #
# Mapping load
# --------------------------------------------------------------------------- #
def load_mapping(path, lead_cols, lead_field_types):
    blocks = json.load(open(path))
    base = blocks[0]["mappings"]
    plans = []
    for m in base:
        tgt = m.get("targetField")
        src = m.get("sourceField")
        if not tgt:
            continue
        expr = m.get("transformExpression")
        vmap = m.get("valueMap")
        node, unsup = None, None
        if expr:
            try:
                node = parse_expr(expr)
                validate_expr(node)
            except Unsupported as e:
                node, unsup = None, f"transform:{e}"
        plans.append({
            "src": src, "tgt": tgt, "node": node, "vmap": vmap,
            "on_unmapped": m.get("onUnmappedValue", "skip"),
            "ftype": lead_field_types.get(tgt, "string"),
            "has_transform": bool(expr), "has_vmap": vmap is not None,
            "unsupported": unsup,
            "schema_missing": tgt not in lead_cols,
            "conflict_policy": m.get("conflictPolicy", "overwrite"),
        })
    return plans


# --------------------------------------------------------------------------- #
# Field reconciliation for one cohort
# --------------------------------------------------------------------------- #
# Field-specific value equivalence classes: literal values that should be
# treated as identical for comparison purposes, even though they're not
# byte-equal. Company's pack-declared BLANKVALUE default is "HubSpot Lead",
# but a parallel SFDC ETL full-record-replace overwrites (or directly creates)
# many blank-company Leads with the literal "[not provided]" instead — both
# mean "no real company on this Lead", not a genuine mismatch. Verified
# against live prod data 2026-07-30: 204 of 213 Company "differ" rows were
# exactly this pair; the other 9 (real HubSpot company text vs "[not
# provided]", or two different real company names) are genuine mismatches and
# are NOT affected by this map, since only THIS specific literal is rewritten
# — a real company name on either side never matches "HubSpot Lead" by luck.
# Maps target field -> {literal value: canonical value it should compare as}.
FIELD_VALUE_EQUIVALENCE = {
    "Company": {"[not provided]": "HubSpot Lead"},
}


def _canonicalize(tgt, val):
    return FIELD_VALUE_EQUIVALENCE.get(tgt, {}).get(val, val)


# Fields whose transform declares a non-empty else/default (CASE's else-
# branch, or BLANKVALUE's fallback) that the real connector does NOT apply
# when the HubSpot source is blank — the pack's declared default diverges
# from runtime behavior specifically for the blank-source case. Industry's
# CASE(...,"Others") is one: verified against live prod data 2026-07-30, all
# 236 records where salesforce_industry is blank also have NexusCRM Industry
# genuinely blank (not "Others") — a separate 16 records where HubSpot DOES
# carry a real, non-blank value that still lands on the else-branch are
# unaffected by this set and continue to compare normally, since they are a
# real (and different) sync gap, not a default-value artifact.
# For fields in this set, a blank source is compared as blank — ignoring the
# transform's default — matching how every field WITHOUT a declared default
# already behaves, instead of manufacturing a comparison target that inflates
# nx_missing / lowers the field's align% for no real discrepancy.
FIELDS_BLANK_SRC_IGNORES_DEFAULT = {"Industry"}

# Fields whose transform concatenates multiple source properties with a fixed
# separator (e.g. Vertical__c = industry & "|" & vertical), where one exact
# RAW transform output (before the valueMap lookup) is the literal produced
# when EVERY underlying property is blank — not a real value that merely
# isn't in the valueMap. That literal is never a valueMap key (every key has
# real text on both sides of the separator), so it always misses the
# valueMap; without this, it's indistinguishable from a genuine unmapped
# value. Verified against live prod data 2026-07-30: all 265 of that day's
# Vertical__c vmap_miss rows were exactly "|" (both salesforce_industry and
# salesforce_industry__vertical_ blank), zero were a real-but-unmapped pair.
# Maps target field -> the exact raw (pre-valueMap) literal meaning "blank".
FIELDS_BLANK_CONCAT_SENTINEL = {"Vertical__c": "|"}


def field_recon(pairs, plans, base_n):
    """pairs: list of (lead_row_dict, hs_props_dict). Returns per-field metrics."""
    stats = {p["tgt"]: collections.Counter() for p in plans}
    for lead, props in pairs:
        for p in plans:
            tgt = p["tgt"]
            if p["unsupported"] or p.get("schema_missing"):
                continue
            src_val = props.get(p["src"])
            src_blank = _s(src_val).strip() == ""
            # Evaluate the transform BEFORE deciding to skip: a blank source
            # may still produce a non-empty expected value via BLANKVALUE
            # (e.g. Company = BLANKVALUE(company, "HubSpot Lead")), which must
            # be compared or real mismatches are hidden.
            try:
                raw = ev(p["node"], props) if p["node"] else src_val
            except (Unsupported, ValueError, TypeError, IndexError, OverflowError):
                stats[tgt]["errors"] += 1
                continue
            if p["vmap"] is not None:
                key = _s(raw)
                if key in p["vmap"]:
                    raw = p["vmap"][key]
                elif key == "":
                    raw = ""
                elif p["on_unmapped"] == "passthrough":
                    # Country / State declare onUnmappedValue: passthrough — an
                    # unmapped value is kept as-is and still compared, not
                    # discarded, per the tenant mapping contract.
                    pass
                elif raw == FIELDS_BLANK_CONCAT_SENTINEL.get(tgt):
                    # Every underlying property is blank, not a real
                    # unmapped value — see FIELDS_BLANK_CONCAT_SENTINEL.
                    stats[tgt]["src_empty"] += 1
                    continue
                else:
                    stats[tgt]["vmap_miss"] += 1
                    continue
            try:
                exp = normalise(raw, p["ftype"])
                act = _canonicalize(tgt, normalise(lead.get(tgt), p["ftype"]))
            except (ValueError, TypeError, OverflowError, OSError):
                stats[tgt]["errors"] += 1
                continue
            if src_blank and tgt in FIELDS_BLANK_SRC_IGNORES_DEFAULT:
                exp = ""
            exp = _canonicalize(tgt, exp)
            # "Source empty" (excluded from the rate) only when a blank source
            # also yields a blank expected — i.e. no BLANKVALUE default.
            if src_blank and exp == "" and p["ftype"] != "boolean":
                stats[tgt]["src_empty"] += 1
                continue
            if exp == "" and act == "":
                stats[tgt]["both_empty"] += 1
            elif exp == act:
                stats[tgt]["match"] += 1
            elif act == "":
                stats[tgt]["nx_missing"] += 1
            elif exp == "":
                stats[tgt]["nx_extra"] += 1
            else:
                stats[tgt]["differ"] += 1
    cells = {}
    for p in plans:
        c = stats[p["tgt"]]
        populated = c["match"] + c["differ"] + c["nx_missing"] + c["nx_extra"]
        rate = round(c["match"] / populated * 100, 1) if populated else None
        if p.get("schema_missing"):
            status, reason = "missing_column", "target column absent from Lead schema"
        elif p["unsupported"]:
            status, reason = "unsupported", str(p["unsupported"])
        elif c["errors"] or c["vmap_miss"]:
            status, reason = "partial", f"{c['errors']} evaluation errors; {c['vmap_miss']} unmapped values"
        elif populated:
            status, reason = "evaluated", ""
        elif not pairs:
            status, reason = "no_pairs", "no matched records in cohort"
        else:
            status, reason = "no_comparable_values", f"{c['src_empty']} source-empty; {c['both_empty']} both-empty"
        cells[p["tgt"]] = {
            "r": rate, "m": c["match"], "d": c["differ"], "ms": c["nx_missing"],
            "p": populated, "se": c["src_empty"], "vx": c["vmap_miss"],
            # both-empty (source non-blank but transform/valueMap → '' and NX also '')
            # is a distinct bucket; carry it so the accounting identity closes:
            #   populated + both_empty + src_empty + vmap_miss + errors = paired size
            # for supported fields with a physical column.
            "be": c["both_empty"], "extra": c["nx_extra"], "errors": c["errors"],
            "status": status, "reason": reason,
        }
    return cells


# --------------------------------------------------------------------------- #
# Assembly
# --------------------------------------------------------------------------- #
def id_format(i):
    if UUID_RE.fullmatch(i):
        return "uuid"
    if SFID_RE.fullmatch(i):
        return "sfdc"
    return "other"


def daily_counts(rows):
    b = collections.defaultdict(lambda: {"total": 0, "uuid": 0, "sfdc": 0})
    seen = set()
    for r in rows:
        i = r["Id"]
        if i in seen:
            continue
        seen.add(i)
        k = bj_day_of(r["CreatedDate"])
        b[k]["total"] += 1
        f = id_format(i)
        if f == "uuid":
            b[k]["uuid"] += 1
        elif f == "sfdc":
            b[k]["sfdc"] += 1
    return b


def status_for(coupling, row):
    if coupling == "lowvol":
        return "green" if abs(row["gap"]) == 0 else ("amber" if abs(row["gap"]) <= 2 else "red")
    return "green" if row["gappct"] <= 5 else ("amber" if row["gappct"] <= 15 else "red")


def assemble_dashboard(win, nx, hs, acct_daily, acct30):
    days = [d.isoformat() for d in win["days"]]
    part = win["today_bj"].isoformat()
    data = {"generated": win["generated"], "days": days, "partial_day": part,
            "partial_note": win["partial_note"], "window": win["window_label"], "pairs": []}
    COUP = {"Lead": "coupled", "Opportunity": "lowvol"}
    for nxo, hso, title in [("Lead", "contacts", "Leads ↔ Contacts"),
                            ("Opportunity", "deals", "Opportunities ↔ Deals")]:
        rows = []
        for d in days:
            n = nx[nxo].get(d, {"total": 0, "uuid": 0, "sfdc": 0})
            h = hs[hso].get(d, 0)
            gap = n["total"] - h
            row = {"day": d, "nx": n["total"], "hs": h, "gap": gap,
                   "gappct": round(abs(gap) / max(n["total"], h, 1) * 100, 1),
                   "uuid": n["uuid"], "sfdc": n["sfdc"], "other": 0,
                   "partial": d == part,
                   "sfdc_share": round(n["sfdc"] / n["total"] * 100) if n["total"] else 0}
            row["status"] = status_for(COUP[nxo], row)
            rows.append(row)
        full = [r for r in rows if not r["partial"]]
        g = [r["gappct"] for r in full]
        data["pairs"].append({
            "nx": nxo, "hs": hso, "title": title, "rows": rows, "coupling": COUP[nxo],
            "roll": {"nx": sum(r["nx"] for r in full), "hs": sum(r["hs"] for r in full),
                     "net": sum(r["nx"] for r in full) - sum(r["hs"] for r in full),
                     "avg_gap": round(sum(abs(r["gap"]) for r in full) / len(full), 1) if full else 0,
                     "avg_gap_pct": round(sum(g) / len(g), 1) if g else 0,
                     "max_gap_pct": max(g) if g else 0,
                     "green": sum(1 for r in full if r["status"] == "green"), "days": len(full)},
        })
    # Account -> Company cohort
    arows = []
    for d in days:
        a = acct_daily.get(d, {"sync": 0, "linked": 0})
        arows.append({"day": d, "nx": a["sync"], "hs": a["linked"],
                      "gap": a["linked"] - a["sync"],
                      "gappct": round(a["linked"] / a["sync"] * 100, 1) if a["sync"] else None,
                      "uuid": a["linked"], "sfdc": 0, "other": 0, "sfdc_share": 0,
                      "partial": d == part,
                      "status": "green" if a["sync"] == 0 or a["linked"] == a["sync"]
                      else ("amber" if a["linked"] / a["sync"] >= 0.8 else "red")})
    afull = [r for r in arows if not r["partial"]]
    tn = sum(r["nx"] for r in afull)
    tl = sum(r["hs"] for r in afull)
    data["pairs"].append({
        "nx": "Account", "hs": "companies", "title": "Accounts → Companies",
        "rows": arows, "coupling": "cohort", "ctx30": acct30,
        "roll": {"nx": tn, "hs": tl, "net": tl - tn, "avg_gap": tn - tl,
                 "avg_gap_pct": round(tl / tn * 100, 1) if tn else 0, "max_gap_pct": 0,
                 "green": sum(1 for r in afull if r["status"] == "green"), "days": len(afull)},
    })
    return data


# Marketing-priority display order for the field table: these fields first,
# in this exact order; everything else follows, alphabetically (case-
# insensitive) by target field name.
FIELD_DISPLAY_PRIORITY = [
    "FirstName", "LastName", "Email", "Phone", "Title", "Company", "Country",
    "Region__c", "HubSpot_Owner_Id__c", "MQL_Priority__c", "Lifecycle_Stage__c",
    "Status", "LeadSource", "Lead_Source_Details__c", "Recent_Source__c",
    "Recent_Source_Detail__c", "First_Conversion__c", "IP_Country__c",
    "Hubspot_Original_Source__c", "HubSpot_Original_Source_Drill_Down_1__c",
    "HubSpot_HubsOriginal_Source_Drill_Down_1__c", "Website_Referrer_URL__c",
    "Most_Recent_Referrer_URL__c", "No_of_Times_MQL__c", "MQL_SLA_Violation__c",
    "Industry", "Vertical__c", "Lead_Stage_DateV2__c", "Suspect_Stage_DateV2__c",
    "Time_Zone__c",
]
_FIELD_PRIORITY_RANK = {t: i for i, t in enumerate(FIELD_DISPLAY_PRIORITY)}


def assemble_field(win, plans, y_cells, t_cells, y_pairs_n, t_pairs_n,
                   y_leads_n, t_leads_n):
    def overall(cells):
        tm = sum(c["m"] for c in cells.values())
        tp = sum(c["p"] for c in cells.values())
        return (round(tm / tp * 100, 1) if tp else None), tp
    y_over, y_tp = overall(y_cells)
    t_over, _ = overall(t_cells)
    rows = []
    for p in plans:
        tgt = p["tgt"]
        rows.append({"t": tgt, "s": p["src"], "ty": p["ftype"],
                     "tr": p["has_transform"], "vm": p["has_vmap"],
                     "policy": p.get("conflict_policy", "overwrite"),
                     "y": y_cells.get(tgt), "td": t_cells.get(tgt),
                     "unavailable": ("target column absent from Lead schema"
                                     if p.get("schema_missing") else p["unsupported"])})
    rows.sort(key=lambda r: (r["t"] not in _FIELD_PRIORITY_RANK,
                             _FIELD_PRIORITY_RANK.get(r["t"], 0), r["t"].lower()))
    ev_rates = [c["r"] for c in y_cells.values() if c["r"] is not None]
    nod = [p["tgt"] for p in plans if not p["unsupported"] and not p.get("schema_missing")
           and not any(cells.get(p["tgt"], {}).get(k, 0)
                       for cells in (y_cells, t_cells) for k in ("p", "errors", "vx"))]
    return {
        "y_label": f"Yesterday {win['auth_bj'].isoformat()} full day (complete · authoritative)",
        "t_label": f"Today {win['today_bj'].isoformat()} {win['partial_note']} (partial)",
        # base for the field table is the *matched* cohort (only matched Leads
        # have HubSpot values to diff); coverage exposes what fraction of new
        # Leads matched, so unmatched Leads are visible, not silently dropped.
        "y_n": y_pairs_n, "t_n": t_pairs_n, "y_overall": y_over, "t_overall": t_over,
        "y_leads": y_leads_n, "t_leads": t_leads_n,
        "y_unmatched": y_leads_n - y_pairs_n, "t_unmatched": t_leads_n - t_pairs_n,
        "coverage": round(y_pairs_n / y_leads_n * 100, 1) if y_leads_n else 100.0,
        "t_coverage": round(t_pairs_n / t_leads_n * 100, 1) if t_leads_n else 100.0,
        "base": y_pairs_n,
        "perfect": sum(1 for c in y_cells.values() if c["p"] and c["m"] == c["p"]
                       and not c.get("errors") and not c.get("vx")),
        "below95": sum(1 for r in ev_rates if r < 95),
        "total_populated": y_tp, "nodata": len(nod), "nodata_fields": nod,
        # total mapped fields in scope = evaluated rows + no-data + unsupported;
        # used for the panel's "N mapped fields" label so it does not undercount
        # by excluding fields that had no data or could not be reproduced.
        "mapped_total": len(plans),
        "unsupported": sum(1 for p in plans if p["unsupported"]),
        "missing_columns": sum(1 for p in plans if p.get("schema_missing")),
        "y_not_evaluated": sum(1 for p in plans if not y_cells.get(p["tgt"], {}).get("p")),
        "t_not_evaluated": sum(1 for p in plans if not t_cells.get(p["tgt"], {}).get("p")),
        "y_errors": sum(c.get("errors", 0) for c in y_cells.values()),
        "t_errors": sum(c.get("errors", 0) for c in t_cells.values()),
        "sort": "marketing_priority_then_alpha",
        "rows": rows,
    }


# --------------------------------------------------------------------------- #
# Report emitters (step-summary markdown + machine-readable JSON)
# --------------------------------------------------------------------------- #
def lead_sfdc_share(pair):
    """SFDC-format write share over a coupled pair's settled days: (count, pct).
    Success target is 0 — every record should arrive via the HubSpot↔NexusCRM
    integration (UUID id); an SFDC-format id is a bulk/ETL write that bypassed
    the connector."""
    full = [r for r in pair["rows"] if not r.get("partial")]
    tot = sum(r["nx"] for r in full)
    sfdc = sum(r.get("sfdc", 0) for r in full)
    return sfdc, (round(sfdc / tot * 100, 1) if tot else 0)


def build_report_md(win, dd, fd, acct30):
    """Render the main dashboard data as step-summary markdown tables — the
    same numbers the HTML shows, but scrapable by other agents straight from
    the run's step summary."""
    L, lead = [], next(p for p in dd["pairs"] if p["nx"] == "Lead")
    lead_sfdc_n, lead_sfdc_pct = lead_sfdc_share(lead)
    percent = lambda value: "—" if value is None else f"{value}%"
    L += [f"# NexusCRM ↔ HubSpot Sync Reconciliation · {win['generated']}", "",
          f"- Window: {win['window_label']}",
          f"- Leads↔Contacts {lead['roll']['days']}-day net **{lead['roll']['net']:+d}**, "
          f"aligned **{lead['roll']['green']}/{lead['roll']['days']}** days, "
          f"avg gap {lead['roll']['avg_gap_pct']}% · **SFDC-channel share "
          f"{lead_sfdc_pct}%** ({lead_sfdc_n} Leads, target 0 — success = 100% via "
          "the HubSpot↔NexusCRM integration)",
          f"- Field alignment (compared cells only): {fd['y_label']} **{percent(fd['y_overall'])}** · "
          f"{fd['t_label']} **{percent(fd['t_overall'])}** "
          f"(pair coverage {fd['coverage']}% / {fd['t_coverage']}%)",
          f"- Mapped targets: **{fd['mapped_total']}** · Y fully matched **{fd['perfect']}** · "
          f"Y below 95% **{fd['below95']}** · Y/T not evaluated **{fd['y_not_evaluated']}/{fd['t_not_evaluated']}**",
          f"- Evaluation gaps: unsupported transforms **{fd['unsupported']}** · "
          f"missing columns **{fd['missing_columns']}** · Y/T evaluation errors "
          f"**{fd['y_errors']}/{fd['t_errors']}**",
          f"- Account→Company 30-day coverage **{acct30['rate']}%** "
          f"({acct30['linked']}/{acct30['n']})", "",
          "## Daily ledger"]
    titles = {"Lead": "Leads ↔ Contacts (tightly coupled · primary)",
              "Opportunity": "Opportunities ↔ Deals",
              "Account": "Accounts → Companies"}
    for p in dd["pairs"]:
        L += ["", f"### {titles.get(p['nx'], p['title'])}"]
        rr = p["roll"]
        if p["coupling"] == "cohort":
            # Account cohort reuses the row keys with DIFFERENT meaning:
            # nx=new sync-enabled Accounts, hs=of those linked to a Company,
            # gap=hs-nx, gappct=coverage%, uuid=linked count, sfdc=n/a. Give it
            # its own headers so the numbers are not read as Lead/Opp semantics.
            L += ["**Success target — read yesterday's full day (the row just "
                  "above \"today (partial)\"): Diff = 0.**", ""]
            L += ["| Date | New (sync-on) | Company built | Diff (H−N) | Coverage% |",
                  "|---|--:|--:|--:|--:|"]
            for r in p["rows"]:
                day = r["day"] + (" (partial)" if r.get("partial") else "")
                cov = "—" if r.get("gappct") is None else f"{r['gappct']}"
                L.append(f"| {day} | {r['nx']} | {r['hs']} | {r['gap']:+d} | {cov} |")
            cov = round(rr["hs"] / rr["nx"] * 100, 1) if rr["nx"] else 0
            L.append(f"| **{rr['days']}-day total** | **{rr['nx']}** | **{rr['hs']}** "
                     f"| **{rr['net']:+d}** | **{cov}** |")
            if p.get("ctx30"):
                c = p["ctx30"]
                L.append(f"\n30-day cohort coverage: **{c['rate']}%** ({c['linked']}/{c['n']} "
                         f"with a Company).")
        else:
            is_lead = p["nx"] == "Lead"
            if is_lead:
                # Success target for the primary link, stated as both
                # conditions (Gap and SFDC count) and pointed at the row to
                # check — yesterday's full day, not the 7-day roll-up or the
                # unsettled partial "today" row. Judge on the raw Gap / SFDC
                # count columns, not the rounded Gap%/SFDC% columns (kept out
                # of the note itself to stay scannable). Blank line after so
                # the table below parses as a table rather than being
                # absorbed into this paragraph.
                L += ["**Success target — read yesterday's full day (the row just "
                      "above \"today (partial)\"): Gap = 0 and SFDC = 0.** Gap = 0 "
                      "means NexusCRM and HubSpot counts match exactly; SFDC = 0 "
                      "means 100% of new Leads arrived via the HubSpot↔NexusCRM "
                      "integration (UUID id), none via SFDC/ETL. The 7-day rows are "
                      "trend context, not the pass/fail read.",
                      ""]
                # No Status column for Leads: the two pass/fail conditions are
                # stated above, so a single traffic-light column would either
                # collapse them into one axis or mislead (see prior bug where
                # Status ignored SFDC%). Read Gap and SFDC% directly instead.
                L += ["| Date | NexusCRM | HubSpot | Gap | Gap% | UUID | SFDC | SFDC% |",
                      "|---|--:|--:|--:|--:|--:|--:|--:|"]
                for r in p["rows"]:
                    day = r["day"] + (" (partial)" if r.get("partial") else "")
                    gp = "—" if r.get("gappct") is None else f"{r['gappct']}"
                    L.append(f"| {day} | {r['nx']} | {r['hs']} | {r['gap']:+d} | {gp} "
                             f"| {r.get('uuid', '')} | {r.get('sfdc', '')} "
                             f"| {r.get('sfdc_share', 0)} |")
                sfdc_n, sfdc_pct = lead_sfdc_share(p)
                L.append(f"| **{rr['days']}-day total** | **{rr['nx']}** | **{rr['hs']}** "
                         f"| **{rr['net']:+d}** | — | | **{sfdc_n}** | **{sfdc_pct}** |")
            else:
                # Success target: read yesterday's full day, Gap = 0. No
                # UUID/SFDC/SFDC%/Status columns here — Opportunities are
                # created in NexusCRM then pushed to HubSpot, so the
                # UUID/SFDC write-path split (a Lead-inbound concept) doesn't
                # apply, and Status collapsing Gap into a single chip is
                # redundant with the Gap/Gap% columns already shown.
                L += ["**Success target — read yesterday's full day (the row just "
                      "above \"today (partial)\"): Gap = 0.**", ""]
                L += ["| Date | NexusCRM | HubSpot | Gap | Gap% |",
                      "|---|--:|--:|--:|--:|"]
                for r in p["rows"]:
                    day = r["day"] + (" (partial)" if r.get("partial") else "")
                    gp = "—" if r.get("gappct") is None else f"{r['gappct']}"
                    L.append(f"| {day} | {r['nx']} | {r['hs']} | {r['gap']:+d} | {gp} |")
                L.append(f"| **{rr['days']}-day total** | **{rr['nx']}** | **{rr['hs']}** "
                         f"| **{rr['net']:+d}** | — |")
    L += ["", "## Field-level reconciliation (HubSpot Contact ↔ NexusCRM Lead)",
          "**Comparison target — read yesterday's full day (the Y columns): "
          "zero differing, missing or extra values among evaluated cells.** "
          "A rounded 100% does not establish complete alignment when values or fields "
          "were not evaluated. Review each cohort's status.", "",
          f"Base: {fd['y_label']} {fd['y_n']} rows · {fd['t_label']} {fd['t_n']} rows "
          f"(pair coverage {fd['coverage']}% / {fd['t_coverage']}%). "
          "Y = complete-day cohort, T = today.", "",
          "| Field (NX ← HS) | Y·status | Y·align% | Y·differ | Y·missing | Y·compared | Y·src-empty "
          "| T·status | T·align% | T·differ | T·missing | T·compared | T·src-empty |",
          "|---|---|--:|--:|--:|--:|--:|---|--:|--:|--:|--:|--:|"]
    for row in fd["rows"]:
        y, td = row["y"], row.get("td")
        def cohort_cells(cell):
            reason = row["unavailable"] or (cell or {}).get("reason")
            status = str(reason or (cell or {}).get("status", "evaluated" if cell else "not evaluated"))
            status = status.replace("|", "\\|").replace("\n", " ")
            metrics = [str(cell[k]) if cell and cell.get(k) is not None else "—"
                       for k in ("r", "d", "ms", "p", "se")]
            return " | ".join([status] + metrics)
        L.append(f"| `{row['t']}` ← `{row['s']}` | {cohort_cells(y)} | {cohort_cells(td)} |")
    if fd.get("nodata_fields"):
        L += ["", f"Not evaluated (no comparable values in either cohort, {fd['nodata']}): "
              f"{', '.join(fd['nodata_fields'])}."]
    L += ["", "Columns: align%=matched÷compared; differ=both sides have a value but "
          "differ; missing=HS has a value while NX is empty; src-empty=HS has no value "
          "(excluded from comparison). Per-field extra, unmapped, both-empty and "
          "evaluation-error counts are retained in report.json.", "",
          "These are normalized value comparisons against the checked-out full inbound "
          "mapping, not replay or repair instructions. Populated skip_if_set targets may "
          "intentionally differ; false is populated. Owner projection derives from the "
          "NexusCRM owner and its unique HubSpot user binding. Pairing uses HubSpot_Id__c "
          "with email fallback, not the finalized connector identity authority. Verify "
          "live identity, deployed mapping and write policy before repairing differences."]
    return "\n".join(L) + "\n"


# --------------------------------------------------------------------------- #
# Main
# --------------------------------------------------------------------------- #
def main():
    a = parse_args()
    if not a.tidb_password:
        raise SystemExit("missing TiDB password (set env TIDB_PROD_PASSWORD or --tidb-password)")
    if not a.hubspot_token:
        raise SystemExit("missing HubSpot token (set env HUBSPOT_TOKEN or --hubspot-token)")
    os.makedirs(a.output_dir, exist_ok=True)
    win = build_windows(resolve_now(a.now))

    print("preflight: HubSpot scope", file=sys.stderr)
    hs_preflight(a.hubspot_token)

    print("connecting TiDB", file=sys.stderr)
    conn = connect_tidb(a)
    lead_types = table_columns(conn, "lead")  # {col: sf_ftype}
    lead_cols = set(lead_types)
    if "HubSpot_Id__c" not in lead_cols:
        raise SystemExit("lead table has no HubSpot_Id__c column; cannot pair by remote id")
    if "CreatedDate" not in lead_cols:
        raise SystemExit("lead table has no CreatedDate column")

    plans = load_mapping(a.mapping, lead_cols, lead_types)
    src_props = sorted({p["src"] for p in plans})
    print(f"mapping: {len(plans)} target fields in inventory", file=sys.stderr)

    # Soft-delete filters per table (exclude IsDeleted=1 mirror rows).
    lead_del = not_deleted(lead_cols)
    opp_del = not_deleted(set(table_columns(conn, "opportunity")))
    acc_del = not_deleted(set(table_columns(conn, "account")))
    tbl_del = {"lead": lead_del, "opportunity": opp_del}

    # ---- daily ledger ----
    nx = {}
    for obj, table in [("Lead", "lead"), ("Opportunity", "opportunity")]:
        nx[obj] = daily_counts(nx_fetch_ids(conn, table, win["ledger_lo"], win["ledger_hi"], tbl_del[table]))
    hs = {}
    for obj in ["contacts", "deals", "companies"]:
        hs[obj] = {}
        for d in win["days"]:
            lo = bj_midnight_utc(d)
            hi = win["now_utc"] if d == win["today_bj"] else bj_midnight_utc(d + dt.timedelta(days=1))
            hs[obj][d.isoformat()] = hs_daily_total(a.hubspot_token, obj, lo, hi)

    # ---- account cohort ----
    acc = nx_fetch_accounts(conn, win["acct30_lo"], win["acct30_hi"], acc_del)
    acct_daily = collections.defaultdict(lambda: {"sync": 0, "linked": 0})
    for x in acc:
        # Cohort = sync-enabled Accounts (what the outbound flow actually pushes).
        # linked ⊆ sync so coverage can never exceed 100%.
        if x.get("HubSpot_Sync_Enabled__c"):
            k = bj_day_of(x["CreatedDate"])
            acct_daily[k]["sync"] += 1
            if x.get("HubSpot_Id__c"):
                acct_daily[k]["linked"] += 1
    a30n = sum(1 for x in acc if x.get("HubSpot_Sync_Enabled__c"))
    a30l = sum(1 for x in acc if x.get("HubSpot_Sync_Enabled__c") and x.get("HubSpot_Id__c"))
    acct30 = {"n": a30n, "linked": a30l, "rate": round(a30l / a30n * 100, 1) if a30n else 0,
              "alive": a30l, "dead": 0}

    # ---- field cohorts (yesterday settled + today) ----
    select_cols = list(dict.fromkeys(["Id", "Email", "HubSpot_Id__c"] + src_props_cols(plans)))

    def cohort(lo, hi):
        leads = nx_fetch_lead_fields(conn, select_cols, lo, hi, lead_del)
        ids = sorted({l["HubSpot_Id__c"] for l in leads if l.get("HubSpot_Id__c")})
        by_id = hs_batch_read(a.hubspot_token, "contacts", ids, src_props)
        mails = sorted({(l.get("Email") or "").strip().lower() for l in leads
                        if (l.get("Email") or "").strip() and not l.get("HubSpot_Id__c")})
        by_email = hs_batch_read(a.hubspot_token, "contacts", mails, src_props, id_property="email")
        pairs = []
        for l in leads:
            hid = l.get("HubSpot_Id__c")
            props = by_id.get(hid) if hid else None
            if props is None:
                props = by_email.get((l.get("Email") or "").strip().lower())
            if props is not None:
                pairs.append((l, props))
        return leads, pairs

    y_leads, y_pairs = cohort(win["auth_lo"], win["auth_hi"])
    t_leads, t_pairs = cohort(win["today_lo"], win["today_hi"])
    y_cells = field_recon(y_pairs, plans, len(y_leads))
    t_cells = field_recon(t_pairs, plans, len(t_leads))

    dd = assemble_dashboard(win, nx, hs, acct_daily, acct30)
    fd = assemble_field(win, plans, y_cells, t_cells,
                        len(y_pairs), len(t_pairs), len(y_leads), len(t_leads))

    # ---- render ----
    md = build_report_md(win, dd, fd, acct30)
    with open(os.path.join(a.output_dir, "report.md"), "w") as f:
        f.write(md)
    # Aggregate evidence only: never publish source values, record IDs or emails.
    with open(os.path.join(a.output_dir, "report.json"), "w") as f:
        json.dump({"generated": win["generated"], "window": win["window_label"],
                   "fields": fd}, f, indent=2)
        f.write("\n")
    print("done:", a.output_dir, file=sys.stderr)


def src_props_cols(plans):
    # target columns to SELECT for field comparison
    return sorted({p["tgt"] for p in plans if not p.get("schema_missing")})


if __name__ == "__main__":
    main()