Match bank reconciliations in code, not in the prompt (#3)

- Match bank reconciliations in code, not in the prompt (8789837f73ab72b671bd4e816efc930e6262f66c)
- Fix four defects found reviewing the matcher (e0cb588e46fa24b90e69f25dd6a101a3e0805f50)

app_chatgpt.py

@@ -10,6 +10,7 @@ import sys
 import time
 import traceback
 import re
+import difflib
 from datetime import datetime
 from azure.core.credentials import AzureKeyCredential
 from azure.ai.documentintelligence import DocumentIntelligenceClient
@@ -340,6 +341,438 @@ except Exception:
     logger.info("ℹ️ tiktoken unavailable — using char/3 token estimate")
 
 
+# ═══════════════════════════════════════════════════════════════════════════
+#  Deterministic reconciliation engine
+# ═══════════════════════════════════════════════════════════════════════════
+#  Matching is arithmetic, not language. Asking the model to emit the match
+#  table let it silently drop rows: the Aug-2026 VCC run narrated "1,450 ERP
+#  rows / 313 bank lines" for files that actually hold 4,042 ERP rows and 588
+#  bank lines, and confirmed ~10% of the real transactions (0 of the 24 on
+#  08/17). Everything below is computed in pandas; the model is handed the
+#  finished result and only writes the prose around it.
+
+_DATE_RE = re.compile(r"(\d{1,2}/\d{1,2}/\d{2,4})")
+_REF_RE = re.compile(r"\b(\d{6,12})\b")
+_NOISE_RE = re.compile(
+    r"beginning bal|ending bal|current period|account total|end of report|"
+    r"^\s*total\b|^\s*page\b", re.I)
+
+
+def _money(value):
+    """'1,710.00' / '(25.00)' / '25.00-' / 25.0 / '' → float."""
+    if value is None or (isinstance(value, float) and pd.isna(value)):
+        return 0.0
+    if isinstance(value, (int, float)):
+        return float(value)
+    s = str(value).strip().replace(",", "").replace("$", "")
+    if not s or s in {"-", ".", "nan"}:
+        return 0.0
+    neg = (s.startswith("(") and s.endswith(")")) or s.startswith("-") or s.endswith("-")
+    s = s.strip("()-").strip()
+    try:
+        return -float(s) if neg else float(s)
+    except ValueError:
+        return 0.0
+
+
+def _text(value):
+    """Cell → clean string. pandas NaN is a truthy float, so `or ""` is not enough."""
+    if value is None or (isinstance(value, float) and pd.isna(value)):
+        return ""
+    s = str(value).strip()
+    return "" if s.lower() == "nan" else s
+
+
+def _norm_name(text):
+    """Vendor name → comparison key. The ERP truncates payees at 27 chars, so
+    comparisons must survive one side being cut mid-word."""
+    s = re.sub(r"[^a-z0-9 ]", " ", str(text or "").lower())
+    s = re.sub(r"\b(inc|llc|ltd|co|corp|company|the|dba|of|and)\b", " ", s)
+    return re.sub(r"\s+", " ", s).strip()
+
+
+# Payment-rail boilerplate. Bank descriptions read "Acworth Recycling dba
+# Payment Instruction 1923-039-116-08 - Acworth Recycling dba Atlas Demolition
+# - ach"; only the text before the first marker is the payee.
+_RAIL_RE = re.compile(
+    r"\b(payment instruction|reversal|ref\.|wire|eft|trace|memo|ach|check)\b", re.I)
+
+
+def _bank_payee(description):
+    """The payee portion of a bank description, minus the rail boilerplate."""
+    text = _text(description)
+    m = _RAIL_RE.search(text)
+    head = text[:m.start()] if m else text
+    return head.strip(" -\u2013\u2014|,") or text
+
+
+def _name_close(a, b):
+    """True when two payee strings plausibly denote the same vendor.
+
+    `b` is matched against its payee portion rather than the whole bank
+    description: comparing a 27-char truncated ERP payee against a string that
+    also carries "Payment Instruction", a reference number and a channel
+    suffix diluted the ratio far below any usable threshold, so this returned
+    False even for obvious pairs and the tolerance pass never fired.
+    Containment handles the truncation in either direction.
+    """
+    a, b = _norm_name(a), _norm_name(_bank_payee(b))
+    if not a or not b:
+        return False
+    if a in b or b in a:
+        return True
+    return difflib.SequenceMatcher(None, a, b).ratio() >= 0.82
+
+
+def _find_col(df, *keywords, exclude=()):
+    """First column whose header contains any keyword (case-insensitive).
+
+    Skips pandas' "Unnamed: N" placeholders for blank header cells — these are
+    everywhere in a report-style export, and "Unnamed" contains "name", which
+    otherwise wins the payee lookup and silently poisons every downstream key.
+    """
+    for col in df.columns:
+        low = str(col).strip().lower()
+        if low.startswith("unnamed"):
+            continue
+        if any(k in low for k in keywords) and not any(x in low for x in exclude):
+            return col
+    return None
+
+
+def _parse_date(value):
+    """Pull a date out of a cell that may also carry other text."""
+    if value is None or (isinstance(value, float) and pd.isna(value)):
+        return pd.NaT
+    if isinstance(value, datetime):
+        return pd.Timestamp(value)
+    text = _text(value)
+    if not text:
+        return pd.NaT
+    m = _DATE_RE.search(text)
+    if m:
+        return pd.to_datetime(m.group(1), errors="coerce")
+    # Textual forms such as "Aug 29, 2026" — Azure DI and CSV exports both emit
+    # these. Parse the whole cell, but only when it is short enough to be a
+    # date rather than a description that happens to name a month.
+    if len(text) <= 32:
+        return pd.to_datetime(text, errors="coerce")
+    return pd.NaT
+
+
+def normalize_erp_ledger(df):
+    """Collapse a G/L *report* export into one row per real transaction.
+
+    VCC's Infor exports are print reports, not data extracts — the Aug-2026
+    workbook carries 18,446 merged-cell ranges — so four things must be undone
+    before any matching is possible:
+
+      • Date and payee arrive concatenated in the 'Date' column, leaving the
+        declared 'Name or Reference' column empty. July's export kept them in
+        separate columns, so both layouts are detected rather than assumed.
+      • The entry number only prints on the first line of a report group; 28%
+        of August rows carry none. It is forward-filled.
+      • One payment is split across N distribution lines (58% of August
+        entries; one has 32). The bank clears the entry, never the line, so
+        lines are summed per entry. This alone is why same-date/same-vendor
+        items "didn't come back as a match".
+      • Debits and credits coexist on one entry (retainage). The bank clears
+        the net, so the net is what we keep.
+
+    Returns a frame of entry / date / payee / amount / n_lines / memo.
+    """
+    date_col = _find_col(df, "date")
+    name_col = _find_col(df, "name", "payee", "vendor", "reference", "desc")
+    entry_col = _find_col(df, "entry", "voucher", "check", "cheque", "doc")
+    debit_col = _find_col(df, "debit")
+    credit_col = _find_col(df, "credit")
+    amount_col = _find_col(df, "amount", "amt", exclude=("debit", "credit"))
+    memo_col = df.columns[0] if len(df.columns) else None
+
+    if date_col is None:
+        raise ValueError(
+            "Could not find a date column in the ERP export. Expected a header "
+            "containing 'date'; got: " + ", ".join(map(str, df.columns[:12])))
+    if debit_col is None and credit_col is None and amount_col is None:
+        raise ValueError(
+            "Could not find an amount column in the ERP export. Expected "
+            "'debit'/'credit' or 'amount'; got: "
+            + ", ".join(map(str, df.columns[:12])))
+
+    records, last_entry = [], ""
+    for _, row in df.iterrows():
+        raw_date = row.get(date_col)
+        date = _parse_date(raw_date)
+        if pd.isna(date):
+            continue  # header, subtotal, page break or balance line
+
+        # Payee: prefer its own column; fall back to the text the date column
+        # carries after the date (the merged-cell case).
+        payee = _text(row.get(name_col)) if name_col else ""
+        if not payee:
+            payee = _DATE_RE.sub("", _text(raw_date), count=1).strip()
+        if _NOISE_RE.search(payee):
+            payee = ""
+        # On report continuations the merged cell carries the entry number in
+        # place of the payee ("08/17/2026  0000261785"). That is an entry key,
+        # not a name — keeping it as a payee splits one payment in two.
+        if re.fullmatch(r"\d{5,}", payee):
+            row_entry_from_name, payee = payee, ""
+        else:
+            row_entry_from_name = ""
+
+        if entry_col is not None:
+            entry = _text(row.get(entry_col))
+            if entry:
+                last_entry = entry
+        elif row_entry_from_name:
+            last_entry = row_entry_from_name
+        entry = last_entry
+
+        if amount_col is not None:
+            amount = _money(row.get(amount_col))
+        else:
+            amount = _money(row.get(debit_col)) - _money(row.get(credit_col))
+
+        memo = _text(row.get(memo_col)) if memo_col is not None else ""
+        if _NOISE_RE.search(memo):
+            continue
+
+        records.append({"entry": entry, "date": date, "payee": payee,
+                        "amount": amount, "memo": memo})
+
+    if not records:
+        raise ValueError("No dated transaction rows were found in the ERP export.")
+
+    # Assign group ids sequentially rather than grouping on the key columns.
+    # Neither signal is reliable alone: the entry number is blank on 28% of
+    # rows (so grouping on it alone merges distinct payments), and the payee
+    # text varies within one entry (so grouping on it splits a single
+    # payment). A new transaction starts when the entry number changes, or
+    # when a named payee differs from the one its group was opened with.
+    gid, prev = -1, None
+    for rec in records:
+        starts_group = prev is None or (
+            rec["entry"] != prev["entry"]
+            or (rec["payee"] and prev["payee"] and rec["payee"] != prev["payee"])
+        )
+        if starts_group:
+            gid += 1
+        rec["gid"] = gid
+        prev = rec
+
+    raw = pd.DataFrame.from_records(records)
+
+    def first_named(series):
+        for v in series:
+            if v:
+                return v
+        return ""
+
+    grouped = (raw.groupby("gid", sort=False)
+                  .agg(entry=("entry", "first"),
+                       date=("date", "first"),
+                       payee=("payee", first_named),
+                       amount=("amount", "sum"),
+                       n_lines=("amount", "size"),
+                       memo=("memo", "first"))
+                  .reset_index(drop=True))
+    # An entry whose distributions net to zero never hits the bank.
+    grouped = grouped[grouped["amount"].round(2) != 0].reset_index(drop=True)
+    # Count every line read, including those in zero-net entries just dropped,
+    # so the report can state how much of the file was actually consumed.
+    grouped.attrs["detail_lines_read"] = len(raw)
+
+    logger.info(f"📒 ERP normalised: {len(raw)} detail line(s) → {len(grouped)} "
+                f"transaction(s) ({int((grouped['n_lines'] > 1).sum())} multi-line)")
+    return grouped
+
+
+def normalize_bank_statement(df):
+    """Normalise the bank side and pull out any ERP reference the bank carries.
+
+    Struxtion writes the originating entry number into the description
+    ('Payment Instruction Ref. 0000262036 - Ronning Landscaping Inc. - check'),
+    which is an exact join key and present on about half the lines — far better
+    than any amount/date/name heuristic. It was previously unused.
+    """
+    date_col = _find_col(df, "date")
+    amount_col = _find_col(df, "amount", "amt", "value", exclude=("balance",))
+    desc_col = _find_col(df, "desc", "narrat", "detail", "memo", "title",
+                         "reference", "payee", "name")
+    if date_col is None or amount_col is None:
+        raise ValueError(
+            "Could not find date and amount columns in the bank statement. "
+            "Got: " + ", ".join(map(str, df.columns[:12])))
+
+    out = pd.DataFrame({
+        "date": df[date_col].map(_parse_date),
+        "amount": df[amount_col].map(_money),
+        "desc": (df[desc_col].map(_text) if desc_col is not None else ""),
+    })
+    out = out[out["amount"].round(2) != 0].reset_index(drop=True)
+    out["refs"] = out["desc"].map(lambda s: set(_REF_RE.findall(s)))
+    # A reversal carries the same magnitude as the payment it backs out, so it
+    # competes for the same match. Flagged here and ranked last, so it is only
+    # taken when nothing else fits.
+    out["is_reversal"] = out["desc"].str.contains(
+        r"revers|returned|nsf|chargeback|stop\s*pay", case=False, regex=True, na=False)
+
+    n_ref = int(out["refs"].map(bool).sum())
+    logger.info(f"🏦 Bank normalised: {len(out)} transaction(s), "
+                f"{n_ref} carrying an ERP reference")
+    return out
+
+
+def _ref_keys(entry):
+    """'0000261546' → {'0000261546', '261546'} so padded and unpadded refs meet."""
+    e = str(entry or "").strip()
+    return {e, e.lstrip("0")} - {""}
+
+
+def match_transactions(erp, bank, amount_tol=0.01, date_tol_days=45):
+    """Greedy one-to-one match, strongest evidence first.
+
+    Passes run in descending confidence and each consumes the rows it matches,
+    so a weaker rule can never steal a transaction a stronger rule owns:
+      1. entry reference + amount   — exact, no tolerance needed
+      2. exact amount, nearest date
+      3. amount within tolerance + payee similarity
+      4. same-payee roll-up: several ERP entries clearing as one bank payment
+
+    Date is a tie-breaker, not a filter. Measured against the 131 pairs that
+    pass 1 matches by reference alone, the lag between the GL posting date and
+    the bank clearing date runs from -27 to +19 days — the bank frequently
+    clears before the GL posts, because the GL books on the AP batch date.
+    A tight window (the previous prompt asked for +/- 3 business days) throws
+    away more true matches than it prevents false ones, so `date_tol_days` is
+    only a sanity guard and ranking is what actually picks between candidates
+    of equal amount.
+
+    Returns (matches, erp_unmatched, bank_unmatched).
+    """
+    erp = erp.reset_index(drop=True).copy()
+    bank = bank.reset_index(drop=True).copy()
+    erp_open = set(erp.index)
+    bank_open = set(bank.index)
+    matches = []
+
+    def take(ei, bi, method, confidence, erp_rows=None):
+        # A roll-up match owns several ERP entries, so the reported amount and
+        # line count are the group's. Reporting only the first entry's figures
+        # rendered every roll-up as a matched pair whose two sides disagreed.
+        rows = list(erp_rows) if erp_rows else [ei]
+        e, b = erp.loc[ei], bank.loc[bi]
+        entries = [str(erp.at[i, "entry"]) for i in rows if _text(erp.at[i, "entry"])]
+        matches.append({
+            "erp_entry": ", ".join(entries) if entries else "",
+            "erp_date": e["date"],
+            "erp_payee": e["payee"],
+            "erp_amount": float(erp.loc[rows, "amount"].sum()),
+            "erp_lines": int(erp.loc[rows, "n_lines"].sum()),
+            "bank_date": b["date"], "bank_desc": b["desc"], "bank_amount": b["amount"],
+            "method": method, "confidence": confidence,
+            "erp_rows": rows,
+        })
+        for idx in rows:
+            erp_open.discard(idx)
+        bank_open.discard(bi)
+
+    # Amounts are compared on magnitude: the GL books a payment as a positive
+    # debit while the bank shows a negative debit.
+    bank_by_amt = {}
+    for bi in bank.index:
+        bank_by_amt.setdefault(round(abs(bank.at[bi, "amount"]), 2), []).append(bi)
+
+    def candidates(amount):
+        return [bi for bi in bank_by_amt.get(round(abs(amount), 2), []) if bi in bank_open]
+
+    def day_gap(ei, bi):
+        """Absolute day distance, or None when either side has no usable date."""
+        ed, bd = erp.at[ei, "date"], bank.at[bi, "date"]
+        if pd.isna(ed) or pd.isna(bd):
+            return None
+        return abs((bd - ed).days)
+
+    def within_days(ei, bi, days):
+        gap = day_gap(ei, bi)
+        return True if gap is None else gap <= days
+
+    has_reversal_flag = "is_reversal" in bank.columns
+
+    def is_reversal(bi):
+        return bool(bank.at[bi, "is_reversal"]) if has_reversal_flag else False
+
+    def nearest(ei, options):
+        """Of several equal-amount bank lines, the best candidate.
+
+        Ordinary lines are preferred over reversals before date is considered.
+        August 2026 carried 539,249.00 as two debits and one reversal credit;
+        ranking on date alone could hand an ERP payment the reversal, which
+        reports the reversal as cleared and leaves a real payment outstanding.
+        """
+        if not options:
+            return None
+        return min(options, key=lambda bi: (is_reversal(bi),
+                                            day_gap(ei, bi) is None,
+                                            day_gap(ei, bi) or 0))
+
+    # ── Pass 1: the bank's own reference to the ERP entry ────────────────────
+    for ei in sorted(erp_open):
+        keys = _ref_keys(erp.at[ei, "entry"])
+        if not keys:
+            continue
+        hits = [bi for bi in candidates(erp.at[ei, "amount"])
+                if keys & bank.at[bi, "refs"]]
+        bi = nearest(ei, hits)
+        if bi is not None:
+            take(ei, bi, "entry reference + amount", "Exact")
+
+    # ── Pass 2: exact amount inside the clearing window ──────────────────────
+    for ei in sorted(erp_open):
+        viable = [bi for bi in candidates(erp.at[ei, "amount"])
+                  if within_days(ei, bi, date_tol_days)]
+        bi = nearest(ei, viable)
+        if bi is not None:
+            take(ei, bi, "exact amount, nearest date", "Exact")
+
+    # ── Pass 3: amount tolerance + payee similarity ──────────────────────────
+    if amount_tol > 0:
+        for ei in sorted(erp_open):
+            target = abs(erp.at[ei, "amount"])
+            viable = [bi for bi in sorted(bank_open)
+                      if abs(abs(bank.at[bi, "amount"]) - target) <= amount_tol
+                      and within_days(ei, bi, date_tol_days)
+                      and _name_close(erp.at[ei, "payee"], bank.at[bi, "desc"])]
+            bi = nearest(ei, viable)
+            if bi is not None:
+                take(ei, bi, "amount tolerance + payee", "High")
+
+    # ── Pass 4: several ERP entries clearing as one bank payment ─────────────
+    # Struxtion batches invoices into a single ACH, so the bank line equals the
+    # sum of that vendor's open entries rather than any one of them.
+    by_payee = {}
+    for ei in sorted(erp_open):
+        by_payee.setdefault(_norm_name(erp.at[ei, "payee"]), []).append(ei)
+    for payee, idxs in by_payee.items():
+        if len(idxs) < 2 or not payee:
+            continue
+        total = sum(erp.at[i, "amount"] for i in idxs)
+        for bi in candidates(total):
+            if any(within_days(i, bi, date_tol_days) for i in idxs):
+                take(idxs[0], bi, f"roll-up of {len(idxs)} ERP entries", "High",
+                     erp_rows=idxs)
+                break
+
+    matched_df = pd.DataFrame(matches)
+    erp_un = erp.loc[sorted(erp_open)].reset_index(drop=True)
+    bank_un = bank.loc[sorted(bank_open)].reset_index(drop=True)
+    logger.info(f"🔗 Matched {len(matched_df)} of {len(erp)} ERP transaction(s) / "
+                f"{len(bank)} bank line(s); unmatched: {len(erp_un)} ERP, "
+                f"{len(bank_un)} bank")
+    return matched_df, erp_un, bank_un
+
+
 _ERP_RELEVANT_KEYWORDS = (
     "date", "desc", "memo", "narrat", "detail", "name", "payee", "vendor",
     "customer", "party", "reference", "ref", "invoice", "inv", "bill",
@@ -469,6 +902,14 @@ def extract_transactions_from_pdf(file_path):
             logger.info(f"📋 Account #{account_idx + 1}: {account_number} ({account_type})"
                         f"  |  Opening: {begin_bal}  Closing: {end_bal}")
 
+            # Kept so the extraction can be checked against the statement's own
+            # control totals further down — see validate_bank_extraction().
+            if _safe_float(begin_bal) is not None and _safe_float(end_bal) is not None:
+                statement_meta.setdefault("balances", []).append(
+                    {"account": account_number,
+                     "beginning": _safe_float(begin_bal),
+                     "ending": _safe_float(end_bal)})
+
             # ── Transactions ─────────────────────────────────────────────────
             txns_field = account_obj.get("Transactions")
             if not txns_field or not txns_field.value_array:
@@ -525,6 +966,7 @@ def extract_transactions_from_pdf(file_path):
 # Returns: (dataframe, source_label)
 #   dataframe    = pandas DataFrame of transactions (trimmed/chunked downstream)
 #   source_label = "CSV" or "PDF (Bank: ..., Period: ...)" for context
+#   meta         = statement metadata (control balances for PDFs; {} for CSVs)
 def extract_transactions(file):
     # Handle both string paths (Gradio filepath) and file objects
     if isinstance(file, str):
@@ -538,7 +980,7 @@ def extract_transactions(file):
     if filename.endswith(".pdf"):
         df, meta = extract_transactions_from_pdf(file_path)
         label = f"PDF (Bank: {meta.get('bank','N/A')}, Period: {meta.get('start','?')} → {meta.get('end','?')})"
-        return df, label
+        return df, label, meta
 
     elif filename.endswith(".csv"):
         try:
@@ -546,7 +988,7 @@ def extract_transactions(file):
             logger.info(f"✅ Successfully read CSV file: {file_path}")
             logger.info(f"📈 File contains {len(df)} rows and {len(df.columns)} columns")
             logger.info(f"🏷️ Column names: {list(df.columns)}")
-            return df, "CSV"
+            return df, "CSV", {}
         except Exception as e:
             error_msg = f"Error processing {file_path}: {e}"
             logger.error(f"❌ {error_msg}")
@@ -826,6 +1268,215 @@ def _reconcile_one_chunk(chunk, external_data, erp_label, ext_label,
         return reports
 
 
+def _fmt_money(value):
+    return f"${value:,.2f}" if value >= 0 else f"-${abs(value):,.2f}"
+
+
+def _fmt_date(value):
+    return "" if pd.isna(value) else pd.Timestamp(value).strftime("%Y-%m-%d")
+
+
+def build_narrative_prompt(stats, erp_un, bank_un, stmt_label):
+    """Ask the model for prose only.
+
+    The model is never asked to match, count or total anything — those are
+    computed and passed in. It sees the finished figures and a sample of the
+    exceptions, and writes the introduction and the recommended actions.
+    """
+    def sample(df, n=15):
+        if df.empty:
+            return "(none)"
+        rows = df.reindex(df["amount"].abs().sort_values(ascending=False).index).head(n)
+        return "\n".join(
+            f"- {_fmt_date(r['date'])} | {str(r.get('payee') or r.get('desc') or '')[:60]} "
+            f"| {_fmt_money(r['amount'])}"
+            for _, r in rows.iterrows())
+
+    return f"""You are a senior financial analyst writing up a completed bank reconciliation.
+
+The reconciliation has ALREADY been performed deterministically. The figures below are final.
+Do NOT recompute, re-match, or second-guess them, and do NOT invent transactions.
+
+FINAL FIGURES
+- Period: {stats['period']}
+- ERP transactions (distribution lines rolled up to entries): {stats['erp_count']} (from {stats['erp_lines']} detail lines)
+- {stmt_label} transactions: {stats['bank_count']}
+- Matched: {stats['matched']} ({stats['erp_cov']:.1f}% of ERP, {stats['bank_cov']:.1f}% of {stmt_label})
+- Unmatched in ERP: {stats['erp_unmatched']} worth {_fmt_money(stats['erp_unmatched_value'])} gross
+- Unmatched in {stmt_label}: {stats['bank_unmatched']} worth {_fmt_money(stats['bank_unmatched_value'])} gross
+- Match methods used: {stats['methods']}
+
+LARGEST UNMATCHED {stmt_label.upper()} ITEMS
+{sample(bank_un)}
+
+LARGEST UNMATCHED ERP ITEMS
+{sample(erp_un)}
+
+Write EXACTLY two Markdown sections and nothing else:
+
+### 📘 Introduction
+One paragraph: what this reconciliation covers, the period, the record counts on each
+side, and how the ledger was rolled up from detail lines to entries. Plain and factual.
+
+### 🧾 Suggested Next Steps
+A numbered list of concrete actions for the accounting team, driven by the unmatched
+items above. Name the specific vendors and amounts you were given. Distinguish items
+that are expected to be bank-only (platform fees, reversals, prefunds, treasury
+transfers) from those that look like genuine posting or timing gaps needing follow-up.
+"""
+
+
+def render_reconciliation_markdown(matched, erp_un, bank_un, stats, stmt_label, narrative):
+    """Build the report body. Every table and figure here is computed, not written
+    by the model; `narrative` supplies only the introduction and next steps."""
+    parts = []
+    intro, steps = narrative, ""
+    if "### 🧾" in narrative:
+        intro, steps = narrative.split("### 🧾", 1)
+        steps = "### 🧾" + steps
+    parts.append(intro.strip())
+
+    parts.append("### ✅ Matched Transactions\n")
+    if matched.empty:
+        parts.append("_No transactions matched._\n")
+    else:
+        parts.append("| # | ERP Date | ERP Entry | ERP Payee | Lines | ERP Amount | "
+                     f"{stmt_label} Date | {stmt_label} Description | {stmt_label} Amount | "
+                     "Method | Confidence |")
+        parts.append("|---|---|---|---|---|---|---|---|---|---|---|")
+        for i, (_, r) in enumerate(matched.iterrows(), 1):
+            parts.append(
+                f"| {i} | {_fmt_date(r['erp_date'])} | {r['erp_entry']} | "
+                f"{str(r['erp_payee'])[:38]} | {r['erp_lines']} | {_fmt_money(r['erp_amount'])} | "
+                f"{_fmt_date(r['bank_date'])} | {str(r['bank_desc'])[:60]} | "
+                f"{_fmt_money(r['bank_amount'])} | {r['method']} | {r['confidence']} |")
+        parts.append("")
+
+    parts.append(f"### ❌ Unmatched in {stmt_label}\n")
+    if bank_un.empty:
+        parts.append(f"_Every {stmt_label} line was matched._\n")
+    else:
+        parts.append("| # | Date | Description | Amount |")
+        parts.append("|---|---|---|---|")
+        ordered = bank_un.reindex(bank_un["amount"].abs().sort_values(ascending=False).index)
+        for i, (_, r) in enumerate(ordered.iterrows(), 1):
+            parts.append(f"| {i} | {_fmt_date(r['date'])} | {str(r['desc'])[:80]} | "
+                         f"{_fmt_money(r['amount'])} |")
+        parts.append("")
+
+    parts.append("### ❌ Unmatched in ERP\n")
+    if erp_un.empty:
+        parts.append("_Every ERP transaction was matched._\n")
+    else:
+        parts.append("| # | Date | Entry | Payee | Lines | Amount |")
+        parts.append("|---|---|---|---|---|---|")
+        ordered = erp_un.reindex(erp_un["amount"].abs().sort_values(ascending=False).index)
+        for i, (_, r) in enumerate(ordered.iterrows(), 1):
+            parts.append(f"| {i} | {_fmt_date(r['date'])} | {r['entry']} | "
+                         f"{str(r['payee'])[:38]} | {r['n_lines']} | {_fmt_money(r['amount'])} |")
+        parts.append("")
+
+    parts.append("### 📊 Reconciliation Summary\n")
+    parts.append("| Metric | Value |")
+    parts.append("|--------|-------|")
+    parts.append(f"| ERP detail lines read | {stats['erp_lines']:,} |")
+    parts.append(f"| ERP transactions (after roll-up) | {stats['erp_count']:,} |")
+    parts.append(f"| {stmt_label} transactions | {stats['bank_count']:,} |")
+    parts.append(f"| Matched | {stats['matched']:,} |")
+    parts.append(f"| ERP coverage | {stats['erp_cov']:.1f}% |")
+    parts.append(f"| {stmt_label} coverage | {stats['bank_cov']:.1f}% |")
+    parts.append(f"| Unmatched in ERP | {stats['erp_unmatched']:,} "
+                 f"({_fmt_money(stats['erp_unmatched_value'])} gross) |")
+    parts.append(f"| Unmatched in {stmt_label} | {stats['bank_unmatched']:,} "
+                 f"({_fmt_money(stats['bank_unmatched_value'])} gross) |")
+    parts.append("")
+
+    if steps:
+        parts.append(steps.strip())
+    return "\n".join(parts)
+
+
+def validate_bank_extraction(df, meta, tolerance=0.01):
+    """Check the extracted rows against the statement's own control totals.
+
+    Beginning balance + the sum of every transaction must equal the ending
+    balance. When it does not, rows were dropped during extraction and any
+    reconciliation built on them is quietly wrong — which is exactly what
+    happened to VCC in August 2026: 313 of 588 lines were read and nothing
+    in the report said so. Returns (ok, message).
+    """
+    balances = (meta or {}).get("balances") or []
+    if not balances:
+        return True, ""          # nothing to check against
+
+    beginning = sum(b["beginning"] for b in balances)
+    ending = sum(b["ending"] for b in balances)
+    extracted = float(pd.to_numeric(df["Amount"], errors="coerce").fillna(0).sum())
+    expected = ending - beginning
+    gap = extracted - expected
+
+    if abs(gap) <= tolerance:
+        logger.info(f"✅ Bank extraction ties to the statement: "
+                    f"{beginning:,.2f} + {extracted:,.2f} = {ending:,.2f}")
+        return True, ""
+
+    message = (
+        f"The {len(df)} transactions read from the statement do not tie to the "
+        f"statement's own totals. Opening {beginning:,.2f} plus the extracted "
+        f"movement {extracted:,.2f} gives {beginning + extracted:,.2f}, but the "
+        f"statement closes at {ending:,.2f} — a difference of {gap:,.2f}. "
+        f"Lines were missed during extraction, so the figures below are "
+        f"incomplete. Re-upload the statement or supply it as CSV before "
+        f"relying on this reconciliation.")
+    logger.error(f"❌ Bank extraction does not tie — gap {gap:,.2f}")
+    return False, message
+
+
+def reconcile_deterministic(erp_df, external_df, stmt_label, extraction_warning=""):
+    """Normalise both sides, match, and return (markdown_body, stats).
+
+    Replaces the previous flow, which pasted the raw ledger into a prompt and
+    asked the model to produce the match table. That could not be relied on:
+    on VCC's August 2026 ledger it narrated 1,450 ERP rows and 313 bank lines
+    for files holding 4,042 and 588, and confirmed about a tenth of the real
+    transactions. Matching now happens here; the model only writes the prose.
+    """
+    erp = normalize_erp_ledger(erp_df)
+    bank = normalize_bank_statement(external_df)
+    matched, erp_un, bank_un = match_transactions(erp, bank)
+
+    dates = pd.concat([erp["date"], bank["date"]]).dropna()
+    period = (f"{dates.min():%Y-%m-%d} → {dates.max():%Y-%m-%d}"
+              if not dates.empty else "not determined")
+    methods = (", ".join(f"{v} {k}" for k, v in matched["method"].value_counts().items())
+               if not matched.empty else "none")
+
+    stats = {
+        "period": period,
+        "erp_lines": int(erp.attrs.get("detail_lines_read", erp["n_lines"].sum())),
+        "erp_count": len(erp),
+        "bank_count": len(bank),
+        "matched": len(matched),
+        "erp_cov": (len(erp) - len(erp_un)) / len(erp) * 100 if len(erp) else 0.0,
+        "bank_cov": (len(bank) - len(bank_un)) / len(bank) * 100 if len(bank) else 0.0,
+        "erp_unmatched": len(erp_un),
+        "bank_unmatched": len(bank_un),
+        # Gross, not a signed net: unmatched reversal credits would otherwise
+        # cancel unmatched debits and understate the work remaining.
+        "erp_unmatched_value": float(erp_un["amount"].abs().sum()) if not erp_un.empty else 0.0,
+        "bank_unmatched_value": float(bank_un["amount"].abs().sum()) if not bank_un.empty else 0.0,
+        "methods": methods,
+    }
+
+    narrative = _call_openai(build_narrative_prompt(stats, erp_un, bank_un, stmt_label))
+    body = render_reconciliation_markdown(matched, erp_un, bank_un, stats,
+                                          stmt_label, narrative)
+    if extraction_warning:
+        body = (f"> ⚠️ **Incomplete statement extraction.** {extraction_warning}\n\n"
+                + body)
+    return body, stats
+
+
 # ✅ Reconcile the statements using OpenAI
 def reconcile_statements_openai(erp_file, external_file, statement_type):
     session_start = datetime.now()
@@ -863,7 +1514,7 @@ def reconcile_statements_openai(erp_file, external_file, statement_type):
         # ── Step 1: Read ERP CSV ─────────────────────────────────────────────
         yield "⏳ Step 1/3 — Reading ERP statement…", "", None, None
         logger.info("📊 Extracting data from ERP file...")
-        erp_df, erp_label = extract_transactions(erp_file)
+        erp_df, erp_label, _ = extract_transactions(erp_file)
 
         # ── Step 2: Read external statement (CSV or PDF via Azure DI) ────────
         is_pdf = ext_path.lower().endswith(".pdf")
@@ -873,68 +1524,81 @@ def reconcile_statements_openai(erp_file, external_file, statement_type):
             yield f"⏳ Step 2/3 — Reading {stmt_label} CSV…", "", None, None
 
         logger.info(f"📊 Extracting data from {stmt_label}...")
-        external_df, ext_label = extract_transactions(external_file)
+        external_df, ext_label, external_meta = extract_transactions(external_file)
+
+        # A statement that does not tie to its own opening/closing balance was
+        # only partly read. Warn rather than reconcile against a silent subset.
+        extraction_ok, extraction_warning = validate_bank_extraction(external_df, external_meta)
+        if not extraction_ok:
+            yield (f"⚠️ {stmt_label} extraction is incomplete — continuing, "
+                   f"but the report will say so…", "", None, None)
 
         # ── Step 3: AI Reconciliation ────────────────────────────────────────
         # Trim the ERP ledger to reconciliation-relevant columns, then check the
         # token budget. The external statement is always sent whole; only the
         # (usually much larger) ERP side is chunked if it doesn't fit.
-        erp_df = trim_erp_columns(erp_df)
-        external_data = external_df.to_string(index=False)
-
-        erp_budget = MAX_PROMPT_TOKENS - PROMPT_OVERHEAD_TOKENS - estimate_tokens(external_data)
-        if erp_budget <= 0:
-            raise ValueError(
-                f"The {stmt_label} alone ({estimate_tokens(external_data)} tokens) exceeds the "
-                f"model input budget of {MAX_PROMPT_TOKENS} tokens. Please split the statement.")
-
-        erp_chunks = chunk_dataframe(erp_df, erp_budget)
-        n_batches = len(erp_chunks)
-
-        partial_reports = []
-        for idx, chunk in enumerate(erp_chunks):
-            if n_batches > 1:
-                yield (f"⏳ Step 3/3 — Reconciling ERP batch {idx + 1}/{n_batches} "
-                       f"against the {stmt_label}…", "", None, None)
-                batch_note = (f"\n**NOTE:** This is one batch of a large ERP "
-                              f"ledger. You are seeing ONLY a subset of the ERP rows, reconciled "
-                              f"against the COMPLETE {stmt_label}.\n")
-            else:
-                yield f"⏳ Step 3/3 — AI is reconciling the statements ({stmt_label})…", "", None, None
-                batch_note = ""
-
-            # _reconcile_one_chunk re-splits adaptively if the gateway still
-            # rejects the prompt for size, so it can return >1 partial report.
-            partial_reports += _reconcile_one_chunk(
-                chunk, external_data, erp_label, ext_label, is_vendor, batch_note)
-            logger.info(f"✅ Received batch {idx + 1}/{n_batches} response from OpenAI")
-
-        # The adaptive split may have produced more partials than planned batches.
-        n_batches = len(partial_reports)
-
-        if n_batches == 1:
-            raw_content = partial_reports[0]
+        # ── Step 3: Reconciliation ───────────────────────────────────────────
+        # Bank reconciliations are matched in code (see the deterministic
+        # engine above) and only narrated by the model, so there is no prompt
+        # to outgrow and no batching. Vendor reconciliations still use the
+        # original single-prompt path.
+        if not is_vendor:
+            yield f"⏳ Step 3/3 — Matching ERP entries against the {stmt_label}…", "", None, None
+            raw_content, stats = reconcile_deterministic(
+                erp_df, external_df, stmt_label, extraction_warning)
+            logger.info(f"✅ Deterministic reconciliation: {stats['matched']} matched, "
+                        f"ERP {stats['erp_cov']:.1f}%, {stmt_label} {stats['bank_cov']:.1f}%")
         else:
-            # Merge per-batch reports into one. If the merge prompt itself fits,
-            # let GPT consolidate; otherwise fall back to labelled concatenation.
-            yield f"⏳ Step 3/3 — Consolidating {n_batches} batches into one report…", "", None, None
-            consolidation_prompt = build_consolidation_prompt(partial_reports, stmt_label, n_batches)
-            _concat = "\n\n---\n\n".join(
-                f"## Batch {i + 1} of {n_batches}\n\n{rep}"
-                for i, rep in enumerate(partial_reports))
-            if estimate_tokens(consolidation_prompt) <= MAX_PROMPT_TOKENS - PROMPT_OVERHEAD_TOKENS:
-                logger.info(f"🧩 Consolidating {n_batches} batch reports via OpenAI...")
-                try:
-                    raw_content = _call_openai(consolidation_prompt)
-                except Exception as e:
-                    if _is_token_limit_error(e):
-                        logger.warning("⚠️ Consolidation rejected for size — concatenating instead")
-                        raw_content = _concat
-                    else:
-                        raise
+            erp_df = trim_erp_columns(erp_df)
+            external_data = external_df.to_string(index=False)
+
+            erp_budget = MAX_PROMPT_TOKENS - PROMPT_OVERHEAD_TOKENS - estimate_tokens(external_data)
+            if erp_budget <= 0:
+                raise ValueError(
+                    f"The {stmt_label} alone ({estimate_tokens(external_data)} tokens) exceeds the "
+                    f"model input budget of {MAX_PROMPT_TOKENS} tokens. Please split the statement.")
+
+            erp_chunks = chunk_dataframe(erp_df, erp_budget)
+            n_batches = len(erp_chunks)
+
+            partial_reports = []
+            for idx, chunk in enumerate(erp_chunks):
+                if n_batches > 1:
+                    yield (f"⏳ Step 3/3 — Reconciling ERP batch {idx + 1}/{n_batches} "
+                           f"against the {stmt_label}…", "", None, None)
+                    batch_note = (f"\n**NOTE:** This is one batch of a large ERP "
+                                  f"ledger. You are seeing ONLY a subset of the ERP rows, reconciled "
+                                  f"against the COMPLETE {stmt_label}.\n")
+                else:
+                    yield f"⏳ Step 3/3 — AI is reconciling the statements ({stmt_label})…", "", None, None
+                    batch_note = ""
+
+                partial_reports += _reconcile_one_chunk(
+                    chunk, external_data, erp_label, ext_label, is_vendor, batch_note)
+                logger.info(f"✅ Received batch {idx + 1}/{n_batches} response from OpenAI")
+
+            n_batches = len(partial_reports)
+            if n_batches == 1:
+                raw_content = partial_reports[0]
             else:
-                logger.warning("⚠️ Consolidation prompt too large — concatenating batch reports instead")
-                raw_content = _concat
+                yield f"⏳ Step 3/3 — Consolidating {n_batches} batches into one report…", "", None, None
+                consolidation_prompt = build_consolidation_prompt(partial_reports, stmt_label, n_batches)
+                _concat = "\n\n---\n\n".join(
+                    f"## Batch {i + 1} of {n_batches}\n\n{rep}"
+                    for i, rep in enumerate(partial_reports))
+                if estimate_tokens(consolidation_prompt) <= MAX_PROMPT_TOKENS - PROMPT_OVERHEAD_TOKENS:
+                    logger.info(f"🧩 Consolidating {n_batches} batch reports via OpenAI...")
+                    try:
+                        raw_content = _call_openai(consolidation_prompt)
+                    except Exception as e:
+                        if _is_token_limit_error(e):
+                            logger.warning("⚠️ Consolidation rejected for size — concatenating instead")
+                            raw_content = _concat
+                        else:
+                            raise
+                else:
+                    logger.warning("⚠️ Consolidation prompt too large — concatenating batch reports instead")
+                    raw_content = _concat
 
         logger.info(f"📄 Response length: {len(raw_content)} characters")