feat(reporting): production runs reach Snowflake DEMO_RUN/DEMO_RESULTS the moment a build completes

The DemoPrep reporting liveboard reads DEMOPREP_REPORTING.REPORTING (DEMO_RUN,
DEMO_RESULTS, V_DEMO_RUN), which until now was filled only by the hand-run loader
dev_notes/reporting_load_runs.py.txt. Nobody ran it Oct 22-26 and five days of SE
builds were missing. A production run - a real Salesforce opportunity id (15/18
alphanumerics, never a TEST-/REPLAY-/XXTEST marker) with environment prod - is now
written the moment its build completes; every other run stays Supabase-only for the
loader, as before.

- demoprep_app/reporting/writer.py: the loader's single-run logic, importable. Same
rules verbatim (SFID, TEST_MARKER, INTERNAL, owner skips, IS_TEST_RUN sticky,
COALESCE on the opportunity id, err_bucket/cluster/space_of, derived model URL),
LOAD_LEDGER keyed by SOURCE_INTAKE_ID so the loader updates instead of duplicating,
one transaction, fully-qualified statements, never raises.
- supabase_client: fetch_mcp_intake_row (the intake row with its id + created_at);
mcp_result_record extracted from log_mcp_result so the stored row and the
write-through share one shaping.
- mcp_server: _start_reporting_write_through, called in _run_build.result() right
after log_mcp_result, daemon thread, try/except everywhere; no tool signature changed.
- tests/test_reporting_writer.py (48): the rule, the rows, the statements against a
fake cursor, the orchestration, the hook. SQL proven against real Snowflake in a
rolled-back transaction.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

demoprep_app/reporting/__init__.py

@@ -0,0 +1,8 @@
+"""Reporting: the Snowflake tables behind the DemoPrep reporting liveboard
+(DEMOPREP_REPORTING.REPORTING — DEMO_RUN, DEMO_RESULTS, LOAD_LEDGER, V_DEMO_RUN).
+
+`writer` puts ONE completed production build there the moment it finishes. The
+hand-run loader (dev_notes/reporting_load_runs.py.txt) remains the backfill for
+everything else and finds the writer's rows through LOAD_LEDGER, so it updates
+them instead of inserting them a second time.
+"""

demoprep_app/reporting/writer.py

@@ -0,0 +1,325 @@
+"""Write ONE completed production build into the Snowflake reporting tables now.
+
+Until 2026-10-26 the tables the DemoPrep reporting liveboard reads
+(DEMOPREP_REPORTING.REPORTING.DEMO_RUN / DEMO_RESULTS, view V_DEMO_RUN) were filled
+only by the hand-run loader, dev_notes/reporting_load_runs.py.txt, from the build
+log in Supabase (demo_history rows mcp_intake + mcp_result). Nobody ran it for five
+days and five days of SE builds were missing. This module is the loader's
+single-run logic, importable, so the MCP server can write a PRODUCTION run the
+moment its build completes (mcp_server._run_build -> _start_reporting_write_through).
+
+A production run (boone's rule) is a request whose salesforce_opportunity_id is a
+real Salesforce id — 15 or 18 alphanumerics; the test markers TEST-, REPLAY- and
+XXTEST never qualify — AND whose environment is prod ("prod" / "production";
+"staging" / "stage" do not qualify). Every other run keeps the old behaviour:
+Supabase only, picked up by the loader later.
+
+Everything else is the loader's rule, kept verbatim so the two writers agree:
+  * SFID / TEST_MARKER, the INTERNAL company set, the testrunner / no.such.person
+    owner skip and the IGNORE company skip (`skip_reason`);
+  * IS_TEST_RUN = the owner is mike.boone@thoughtspot.com, and sticky in the MERGE;
+  * SALESFORCE_OPPORTUNITY_ID is COALESCEd on update, so a human-entered id is
+    never overwritten with NULL;
+  * `err_bucket`, `cluster`, `space_of`, and the model URL derived from ts_url +
+    model_guid when the result carries none;
+  * LOAD_LEDGER keyed by SOURCE_INTAKE_ID = the demo_history intake row id. The
+    loader finds that entry on its next run and UPDATEs the same DEMO_RUN_KEY, so
+    a write-through row is never counted twice.
+
+Not done here, left to the loader's next run: inheriting an opportunity id from a
+prospect's other runs (moot — a write-through run always has one) and syncing
+app-emitted findings into DEMO_FINDINGS.
+
+The loader stages rows in _RUN_STG / _RES_STG temp tables and MERGEs from them;
+with exactly one run the same statements bind the row's values directly
+(`USING (SELECT %s AS ...)`), fully qualified, inside one transaction.
+"""
+from __future__ import annotations
+
+import re
+from datetime import datetime, timezone
+from typing import Any, Callable, Dict, Optional
+
+REPORTING_SCHEMA = "DEMOPREP_REPORTING.REPORTING"
+
+# A Salesforce opportunity id is 15 or 18 alphanumerics. Anything else (harness
+# sentinels like 'REPLAY-NOT-A-DEAL-…', 'NONE', free text) is NOT an id.
+SFID = re.compile(r"^[A-Za-z0-9]{15}(?:[A-Za-z0-9]{3})?$")
+# Test builds send a marker in place of an opportunity id (tests/*:
+# TEST_OPPORTUNITY_ID = "TEST-NOT-A-DEAL"; older replays "REPLAY-NOT-A-DEAL-<CO>";
+# XXTEST accepted too). A marked request is a test, never a demo for a deal.
+TEST_MARKER = re.compile(r"^(TEST-|REPLAY-|XXTEST)", re.I)
+# Fictional prospects the harnesses build; never reported.
+INTERNAL = {"vantage capital", "northwind retail", "atlas freight", "cloudscape",
+            "streamly", "meridian health", "meridian health system"}
+# Whose runs are flagged IS_TEST_RUN (excluded from SE figures).
+TEST_RUN_OWNER = "mike.boone@thoughtspot.com"
+
+_ENVIRONMENTS = {"prod": "prod", "production": "prod", "staging": "staging", "stage": "staging"}
+
+
+# --- the rule ---------------------------------------------------------------- #
+def normalize_environment(value: Any) -> str:
+    """'prod' | 'staging' | '' — the same synonyms build_demo_from_brief accepts."""
+    return _ENVIRONMENTS.get(str(value or "").strip().lower(), "")
+
+
+def opportunity_id(value: Any) -> Optional[str]:
+    """The id if it is a well-formed Salesforce id, else None (the loader's opp_id)."""
+    v = str(value or "").strip()
+    return v if v and SFID.match(v) else None
+
+
+def is_production_run(opportunity: Any, environment: Any) -> bool:
+    """boone's rule: a real Salesforce opportunity id AND environment prod."""
+    v = str(opportunity or "").strip()
+    if not v or TEST_MARKER.match(v) or not SFID.match(v):
+        return False
+    return normalize_environment(environment) == "prod"
+
+
+def skip_reason(owner_email: Any, company_name: Any) -> Optional[str]:
+    """Why the loader would skip this request outright, or None. Verbatim from the
+    loader's per-request filter (no.such.person = the harness's unprovisioned-user
+    probe; IGNORE = a request marked to be left out; INTERNAL = fictional prospects)."""
+    o = str(owner_email or "").lower()
+    co = str(company_name or "").strip()
+    if not o:
+        return "no owner"
+    if "testrunner" in o:
+        return "testrunner owner"
+    if o.startswith("no.such.person"):
+        return "no.such.person owner"
+    if "IGNORE" in co:
+        return "IGNORE company"
+    if co.lower() in INTERNAL:
+        return "internal company"
+    return None
+
+
+# --- the loader's helpers, verbatim -------------------------------------------- #
+def err_bucket(t: Any) -> Optional[str]:
+    if not t:
+        return None
+    l = str(t).lower()
+    for k, v in (("timed out", "llm_timeout"), ("schema deployment", "schema_deploy"),
+                 ("quality gate", "data_quality_gate"), ("concrete values", "blueprint_cardinality"),
+                 ("placeholder", "blueprint_placeholder_names"),
+                 ("failed validation after repair", "blueprint_validation_other"),
+                 ("liveboard creation", "liveboard_creation"),
+                 ("region not supported", "geo_region_unsupported"), ("tml export", "tml_export"),
+                 ("authentication failed", "thoughtspot_auth"), ("compilation error", "sql_compilation"),
+                 ("permission denied", "filesystem_permission"), ("max_tokens", "blueprint_truncated"),
+                 ("hf_token", "portal_token_missing")):
+        if k in l:
+            return v
+    return "other"
+
+
+def cluster(u: Any) -> Optional[str]:
+    u = (u or "").lower()
+    return ("sebe" if "sebe" in u else
+            ("se-thoughtspot-cloud" if "se-thoughtspot" in u else
+             ("techpartners" if "techpartners" in u else None)))
+
+
+TESTKEYS = {"with_portal", "use_case", "resolved_ts_url"}
+
+
+def space_of(irf: Dict[str, Any]) -> str:
+    """prod|test from the Space's own stamp when present (since the Sep 22 release
+    both Spaces stamp space_id); the key fingerprint is only trusted for older,
+    unstamped rows."""
+    sid = (irf.get("space_id") or "").split("/")[-1]
+    if sid == "test-mcp":
+        return "test"
+    if sid == "mcp":
+        return "prod"
+    return "test" if (TESTKEYS & set(irf.keys())) else "prod"
+
+
+# --- the rows ------------------------------------------------------------------- #
+def shape_rows(intake_row: Dict[str, Any], result_record: Dict[str, Any],
+               completed_at: Optional[str]) -> tuple[Dict[str, Any], Dict[str, Any]]:
+    """The loader's _RUN_STG and _RES_STG rows for one request, keyed by the staging
+    column names.
+
+    intake_row    — the demo_history mcp_intake row as stored: id, created_at,
+                    company_name, use_case, results (the full call payload).
+    result_record — the `results` JSON of the matching mcp_result row
+                    (supabase_client.mcp_result_record), or {} when the build left none.
+    completed_at  — the result row's created_at (ISO), or None.
+    """
+    irf = intake_row.get("results") or {}
+    rr = result_record or {}
+    rid = irf.get("run_id") or ""
+    o = (irf.get("owner_email") or "").lower()
+    co = (intake_row.get("company_name") or "").strip()
+    errs = rr.get("errors") or []
+    ts = (irf.get("ts_url") or rr.get("ts_url") or "").rstrip("/") or None
+    mg = rr.get("model_guid") or None
+    run = {
+        "INTAKE_ID": str(intake_row["id"]),
+        "RUN_ID": rid,
+        "SFDC": opportunity_id(irf.get("salesforce_opportunity_id")
+                               or rr.get("salesforce_opportunity_id")),
+        "EMAIL": o,
+        "PROSPECT": co,
+    }
+    res = {
+        "INTAKE_ID": str(intake_row["id"]),
+        "USE_CASE": intake_row.get("use_case") or None,
+        "REQUESTED_AT": intake_row["created_at"],
+        "COMPLETED_AT": completed_at if rr else None,
+        "ELAPSED": rr.get("elapsed_seconds"),
+        "STATUS": rr.get("build_status") or ("no_result" if not rr else "unknown"),
+        "ERR_CAT": err_bucket(errs[0] if errs else None),
+        "SPACE": space_of(irf),
+        "CLUSTER_NAME": cluster(ts),
+        "IS_TEST": o == TEST_RUN_OWNER,
+        "LB_GUID": rr.get("liveboard_guid") or None,
+        "LB_URL": rr.get("liveboard_url") or None,
+        "MODEL_GUID": mg,
+        "MODEL_URL": (rr.get("model_url")
+                      or (f"{ts}/#/data/tables/{mg}" if ts and mg else None)) or None,
+        "PORTAL_URL": (rr.get("portal_url") or "").strip() or None,
+        "SCHEMA_NAME": rr.get("schema") or None,
+        "ERR_TXT": (str(errs[0])[:4000] if errs else None),
+    }
+    return run, res
+
+
+# --- the statements --------------------------------------------------------------- #
+_S = REPORTING_SCHEMA
+
+SQL_LEDGER_LOOKUP = f"SELECT DEMO_RUN_KEY FROM {_S}.LOAD_LEDGER WHERE SOURCE_INTAKE_ID=%s"
+
+# The loader's WHEN MATCHED branch for DEMO_RUN: a human-entered opportunity id is
+# never overwritten with NULL.
+SQL_RUN_UPDATE = f"""UPDATE {_S}.DEMO_RUN SET RUN_ID=%s,
+  SALESFORCE_OPPORTUNITY_ID=COALESCE(SALESFORCE_OPPORTUNITY_ID,%s), RUNNER_EMAIL=%s, PROSPECT_NAME=%s
+  WHERE DEMO_RUN_KEY=%s"""
+
+SQL_RUN_INSERT = (f"INSERT INTO {_S}.DEMO_RUN (RUN_ID, SALESFORCE_OPPORTUNITY_ID, RUNNER_EMAIL, PROSPECT_NAME) "
+                  "VALUES (%s,%s,%s,%s)")
+
+# The key the insert was given: the loader finds it the same way, by the row's
+# identity among rows the ledger does not know yet.
+SQL_RUN_KEY_OF_INSERT = f"""SELECT MAX(r.DEMO_RUN_KEY) FROM {_S}.DEMO_RUN r
+  WHERE r.RUN_ID=%s AND r.RUNNER_EMAIL=%s AND r.PROSPECT_NAME=%s
+    AND NOT EXISTS (SELECT 1 FROM {_S}.LOAD_LEDGER l WHERE l.DEMO_RUN_KEY=r.DEMO_RUN_KEY)"""
+
+SQL_LEDGER_INSERT = f"INSERT INTO {_S}.LOAD_LEDGER (SOURCE_INTAKE_ID, DEMO_RUN_KEY) VALUES (%s,%s)"
+
+# The loader's DEMO_RESULTS MERGE, with the staged row bound in place of _RES_STG.
+# IS_TEST_RUN is sticky: a manual test flag is never un-set by a reload.
+SQL_RESULTS_MERGE = f"""MERGE INTO {_S}.DEMO_RESULTS t USING (
+  SELECT %s AS DEMO_RUN_KEY, %s AS USE_CASE, TO_TIMESTAMP_TZ(%s) AS REQUESTED_AT, TO_TIMESTAMP_TZ(%s) AS COMPLETED_AT,
+         %s AS ELAPSED, %s AS STATUS, %s AS ERR_CAT, %s AS SPACE, %s AS CLUSTER_NAME, %s AS IS_TEST,
+         %s AS LB_GUID, %s AS LB_URL, %s AS MODEL_GUID, %s AS MODEL_URL, %s AS PORTAL_URL, %s AS SCHEMA_NAME, %s AS ERR_TXT
+  ) s ON t.DEMO_RUN_KEY=s.DEMO_RUN_KEY
+  WHEN MATCHED THEN UPDATE SET USE_CASE=s.USE_CASE, REQUESTED_AT=s.REQUESTED_AT, COMPLETED_AT=s.COMPLETED_AT, ELAPSED_SECONDS=s.ELAPSED, STATUS=s.STATUS,
+       ERROR_CATEGORY=s.ERR_CAT, SPACE=s.SPACE, CLUSTER_NAME=COALESCE(s.CLUSTER_NAME,t.CLUSTER_NAME), IS_TEST_RUN=(t.IS_TEST_RUN OR s.IS_TEST),
+       SOURCE_LOADED_AT=CURRENT_TIMESTAMP(),
+       LIVEBOARD_GUID=COALESCE(s.LB_GUID,t.LIVEBOARD_GUID), LIVEBOARD_URL=COALESCE(s.LB_URL,t.LIVEBOARD_URL), MODEL_GUID=COALESCE(s.MODEL_GUID,t.MODEL_GUID),
+       MODEL_URL=COALESCE(s.MODEL_URL,t.MODEL_URL), PORTAL_URL=COALESCE(s.PORTAL_URL,t.PORTAL_URL), SNOWFLAKE_SCHEMA=COALESCE(s.SCHEMA_NAME,t.SNOWFLAKE_SCHEMA), ERROR_TEXT=s.ERR_TXT
+  WHEN NOT MATCHED THEN INSERT (DEMO_RUN_KEY,USE_CASE,REQUESTED_AT,COMPLETED_AT,ELAPSED_SECONDS,STATUS,ERROR_CATEGORY,SPACE,CLUSTER_NAME,IS_TEST_RUN,SOURCE_LOADED_AT,
+       LIVEBOARD_GUID,LIVEBOARD_URL,MODEL_GUID,MODEL_URL,PORTAL_URL,SNOWFLAKE_SCHEMA,ERROR_TEXT)
+  VALUES (s.DEMO_RUN_KEY,s.USE_CASE,s.REQUESTED_AT,s.COMPLETED_AT,s.ELAPSED,s.STATUS,s.ERR_CAT,s.SPACE,s.CLUSTER_NAME,s.IS_TEST,CURRENT_TIMESTAMP(),
+       s.LB_GUID,s.LB_URL,s.MODEL_GUID,s.MODEL_URL,s.PORTAL_URL,s.SCHEMA_NAME,s.ERR_TXT)"""
+
+_RES_ORDER = ("USE_CASE", "REQUESTED_AT", "COMPLETED_AT", "ELAPSED", "STATUS", "ERR_CAT", "SPACE",
+              "CLUSTER_NAME", "IS_TEST", "LB_GUID", "LB_URL", "MODEL_GUID", "MODEL_URL", "PORTAL_URL",
+              "SCHEMA_NAME", "ERR_TXT")
+
+
+def write_run(cur, run: Dict[str, Any], res: Dict[str, Any]) -> Dict[str, Any]:
+    """MERGE one request into DEMO_RUN / LOAD_LEDGER / DEMO_RESULTS through `cur`
+    (a Snowflake cursor; the caller owns the transaction). Returns
+    {"demo_run_key": <key>, "action": "inserted" | "updated"}."""
+    cur.execute(SQL_LEDGER_LOOKUP, (run["INTAKE_ID"],))
+    hit = cur.fetchone()
+    if hit and hit[0] is not None:
+        key = int(hit[0])
+        cur.execute(SQL_RUN_UPDATE, (run["RUN_ID"], run["SFDC"], run["EMAIL"], run["PROSPECT"], key))
+        action = "updated"
+    else:
+        cur.execute(SQL_RUN_INSERT, (run["RUN_ID"], run["SFDC"], run["EMAIL"], run["PROSPECT"]))
+        cur.execute(SQL_RUN_KEY_OF_INSERT, (run["RUN_ID"], run["EMAIL"], run["PROSPECT"]))
+        got = cur.fetchone()
+        if not got or got[0] is None:
+            raise RuntimeError("DEMO_RUN insert left no un-ledgered row to key")
+        key = int(got[0])
+        cur.execute(SQL_LEDGER_INSERT, (run["INTAKE_ID"], key))
+        action = "inserted"
+    cur.execute(SQL_RESULTS_MERGE, (key, *[res[k] for k in _RES_ORDER]))
+    return {"demo_run_key": key, "action": action}
+
+
+# --- the entry point the server calls ------------------------------------------------ #
+def _default_connect():
+    from snowflake_auth import get_snowflake_connection
+    return get_snowflake_connection()
+
+
+def _default_fetch_intake(run_id: str) -> Optional[Dict[str, Any]]:
+    from supabase_client import fetch_mcp_intake_row
+    return fetch_mcp_intake_row(run_id)
+
+
+def _default_result_record(payload: Dict[str, Any]) -> Dict[str, Any]:
+    from supabase_client import mcp_result_record
+    return mcp_result_record(payload)
+
+
+def write_completed_build(run_id: str, result_payload: Dict[str, Any], *,
+                          opportunity: Any, environment: Any,
+                          completed_at: Optional[str] = None,
+                          connect: Callable[[], Any] = _default_connect,
+                          fetch_intake: Callable[[str], Optional[Dict[str, Any]]] = _default_fetch_intake,
+                          result_record: Callable[[Dict[str, Any]], Dict[str, Any]] = _default_result_record,
+                          ) -> Dict[str, Any]:
+    """Put one just-completed build into the reporting tables, if it is a production
+    run. `result_payload` is the dict the server hands log_mcp_result. Returns
+    {"status": "written" | "skipped" | "failed", ...} and never raises: the server
+    calls this on a background thread after the build result is recorded, and
+    nothing here may fail, block or slow a build.
+    """
+    try:
+        if not is_production_run(opportunity, environment):
+            return {"status": "skipped", "reason": "not a production run",
+                    "opportunity": str(opportunity or ""), "environment": str(environment or "")}
+        intake = fetch_intake(run_id)
+        if not intake or not intake.get("id") or not intake.get("created_at"):
+            # No intake row = the loader would never see this run either. Its
+            # own failure was already logged by log_mcp_payload.
+            return {"status": "skipped", "reason": "no intake row in demo_history"}
+        irf = intake.get("results") or {}
+        why = skip_reason(irf.get("owner_email"), intake.get("company_name"))
+        if why:
+            return {"status": "skipped", "reason": why}
+        run, res = shape_rows(intake, result_record(result_payload),
+                              completed_at or datetime.now(timezone.utc).isoformat())
+        conn = connect()
+        try:
+            conn.autocommit(False)
+            cur = conn.cursor()
+            try:
+                out = write_run(cur, run, res)
+            finally:
+                cur.close()
+            conn.commit()
+        except Exception:
+            try:
+                conn.rollback()
+            except Exception:
+                pass
+            raise
+        finally:
+            try:
+                conn.close()
+            except Exception:
+                pass
+        return {"status": "written", "intake_id": run["INTAKE_ID"], **out}
+    except Exception as exc:
+        return {"status": "failed", "error": f"{type(exc).__name__}: {exc}"}

mcp_server.py

@@ -787,6 +787,39 @@ def _complete_brand_from_capture(run_id: str, brand: dict, company_url: str,
         _log_hit(f"BRAND run_id={run_id} {n}")
 
 
+def _start_reporting_write_through(run_id: str, result_payload: dict, *,
+                                   opportunity: str, environment: str):
+    """Write a just-completed PRODUCTION run to the Snowflake reporting tables on a
+    daemon thread (demoprep_app.reporting.writer.write_completed_build). A
+    production run is a real Salesforce opportunity id (15/18 alphanumerics, no
+    TEST-/REPLAY-/XXTEST marker) with environment prod; anything else is left to
+    the hand-run loader, as before. Returns the started Thread, or None when the
+    run does not qualify or the thread could not be started. Never raises and
+    never waits: the build result is already recorded when this runs.
+    """
+    try:
+        from demoprep_app.reporting.writer import is_production_run, write_completed_build
+        if not is_production_run(opportunity, environment):
+            _log_hit(f"REPORTING_SKIP run_id={run_id} not a production run "
+                     f"(opportunity={opportunity!r} environment={environment!r})")
+            return None
+
+        def _write() -> None:
+            try:
+                outcome = write_completed_build(run_id, result_payload,
+                                                opportunity=opportunity, environment=environment)
+            except Exception as exc:  # write_completed_build never raises; belt and braces
+                outcome = {"status": "failed", "error": f"{type(exc).__name__}: {exc}"}
+            _log_hit(f"REPORTING_WRITE run_id={run_id} {outcome}")
+
+        t = threading.Thread(target=_write, daemon=True, name=f"reporting-{run_id}")
+        t.start()
+        return t
+    except Exception as exc:
+        _log_hit(f"REPORTING_WRITE run_id={run_id} not started: {type(exc).__name__}: {exc}")
+        return None
+
+
 def _run_build(run_id: str, brief: str, company_name: str, use_case: str, company_url: str,
                ts_target_url: str, ts_auth_key: str, owner: str,
                with_portal: bool = True, call_payload: dict | None = None,
@@ -963,7 +996,7 @@ def _run_build(run_id: str, brief: str, company_name: str, use_case: str, compan
         # Durable result record: the in-memory stores above wipe on restart, so
         # every terminal outcome (success | partial | failed) is also persisted
         # to demo_history (status='mcp_result', joined to intake by run_id).
-        _persisted_result = log_mcp_result({
+        result_payload = {
             # The same Space + commit stamp the intake row and the session_logs
             # 'run started' event carry, so the three records agree on where
             # this run happened.
@@ -985,8 +1018,18 @@ def _run_build(run_id: str, brief: str, company_name: str, use_case: str, compan
             # schema if ThoughtSpot refused it (E3).
             "ddl": out["ddl"],
             **_opt_in,
-        })
+        }
+        _persisted_result = log_mcp_result(result_payload)
         _log_hit(f"RESULT_PERSISTED run_id={run_id} status={out['status']} supabase={_persisted_result}")
+        # Snowflake reporting write-through (boone 2026-10-26): a PRODUCTION run — a
+        # real Salesforce opportunity id and environment prod — reaches the
+        # reporting tables now, on its own thread, instead of waiting for the
+        # hand-run loader. Every other run stays Supabase-only for the loader.
+        # Starts after the durable records above and can never fail, block or
+        # slow this result.
+        _start_reporting_write_through(run_id, result_payload,
+                                       opportunity=salesforce_opportunity_id,
+                                       environment=environment)
         return out
 
     controller = None

sprint_2026_04.md

@@ -1847,6 +1847,76 @@ settings IN the database through the MCP server, dropdown-constrained.
   real `TS_USER_DEFAULT` / `TS_ENV_URL_DEFAULT`) and the `OPENAI_API_KEY` check in
   `validate_required_settings()`
 
+## October 26, 2026 — HOTFIX: production runs written to Snowflake reporting the moment a build completes (branch `claude/hotfix-snowflake-writethrough-d22061`, off the test tip `0bcf580`)
+
+Found tonight: the tables the DemoPrep reporting liveboard reads
+(`DEMOPREP_REPORTING.REPORTING.DEMO_RUN` / `DEMO_RESULTS`, view `V_DEMO_RUN`) were
+filled ONLY by the hand-run loader `dev_notes/reporting_load_runs.py.txt` (reads the
+build log in Supabase: `demo_history` rows `mcp_intake` + `mcp_result`, MERGEs into
+Snowflake). No scheduler. Nobody ran it Oct 22–26, so five days of SE builds were
+missing until boone ran it by hand (DEMO_RUN then 570 rows, through Oct 26 20:46 UTC).
+CEO demo on Oct 27 — hotfixed tonight, through test and then prod (boone's call).
+
+**The rule (boone):** a *production run* = `salesforce_opportunity_id` is a real
+Salesforce id (15 or 18 alphanumerics — the loader's `SFID`; `TEST-` / `REPLAY-` /
+`XXTEST` markers never qualify) AND `environment` is prod (`prod` / `production`;
+`staging` / `stage` do not). Those reach Snowflake the moment the build completes.
+Every other run keeps the old behaviour: Supabase only, picked up by the loader later.
+
+Line decision: prod `b918bd8` is an ancestor of test `0bcf580` (`git merge-base
+--is-ancestor`), both confirmed live by `ping`; the test tip had completed a full
+successful build tonight (run `4de504402486`, liveboard + portal) and the sprint
+entries above record the Oct 22 pieces verified on test — so the normal path: branch
+from the test tip, deploy to test, verify, push the same SHA to prod.
+
+### Done
+- [x] `demoprep_app/reporting/writer.py` ✅ the loader's single-run logic, importable.
+  Every loader rule kept verbatim: `SFID`, `TEST_MARKER`, the `INTERNAL` company set,
+  the `testrunner` / `no.such.person` / `IGNORE` skips, `IS_TEST_RUN` = owner is
+  mike.boone (sticky in the MERGE), `COALESCE` so a human-entered opportunity id is
+  never overwritten with NULL, `err_bucket` / `cluster` / `space_of`, the model URL
+  derived from ts_url + model guid. `LOAD_LEDGER` row keyed by `SOURCE_INTAKE_ID`
+  (the `demo_history` intake row id) so the loader's next run finds it and UPDATEs
+  the same `DEMO_RUN_KEY` — never a second row. One transaction, fully-qualified
+  statements (no `USE SCHEMA`), `get_snowflake_connection()` (keypair from admin
+  settings, which the Space already has). `write_completed_build()` never raises.
+  Left to the loader: opportunity-id inheritance (moot — a write-through run always
+  has one) and the app-findings sync into `DEMO_FINDINGS`.
+- [x] `supabase_client.fetch_mcp_intake_row(run_id)` ✅ the intake row AS STORED (id +
+  created_at — `fetch_mcp_run` drops both); `mcp_result_record(payload)` ✅ the result
+  row's `results`, now ONE shaping shared by `log_mcp_result` and the writer, so the
+  write-through and the loader read the same record.
+- [x] `mcp_server._start_reporting_write_through()` ✅ called in `_run_build.result()`
+  right after `log_mcp_result`, only for qualifying runs; daemon thread; try/except
+  around everything; hit-log lines `REPORTING_SKIP` / `REPORTING_WRITE`. No MCP tool
+  signature changed (AgentSpot caches the tool list). `.env` untouched.
+- [x] Tests ✅ `tests/test_reporting_writer.py` (48): the rule (real id + prod → yes;
+  real id + staging → no; `TEST-NOT-A-DEAL` → no; blank → no …), the rows, the
+  statements against a fake cursor (insert → ledger → merge; ledger hit → update with
+  COALESCE), the orchestration (staging never connects; Snowflake failure → rolled
+  back, reported, never raised), the server hook (own thread; a raising writer never
+  escapes). `tests/test_deploy_existing_schema.py` source check re-pointed at
+  `mcp_result_record`.
+- [x] SQL proven against real Snowflake in a rolled-back transaction ✅ insert →
+  the row visible in `V_DEMO_RUN` with liveboard / model URL / portal / schema →
+  second pass with the same intake id = `updated`, same key → rollback left nothing.
+- [ ] test-mcp: deploy, `ping` shows the SHA, e2e (prod-flagged build with a
+  well-formed id + staging control), loader dry-run + real run with no double count
+- [ ] prod: same SHA, `ping`, one real-shaped prod build as the final proof
+
+### Open
+- [ ] **Safety net:** a scheduled loader run (daily) for everything the
+  write-through does not cover — staging / test-marked runs that should still be
+  reported, a run whose Snowflake write failed, the opportunity-id inheritance and
+  the app-findings sync.
+- [ ] Loader quirk found on the way (not changed — the loader lives outside the repo):
+  in its `DEMO_RESULTS` MERGE the `-- sticky …` comment swallows
+  `SOURCE_LOADED_AT=CURRENT_TIMESTAMP(),` on the same line, so an UPDATE never
+  refreshes `SOURCE_LOADED_AT` (inserts do). The writer's copy has the comment on its
+  own line.
+- [ ] `dev_notes/reporting_admin_handoff_DEMO_RUN.md` "How it is maintained" should name
+  the write-through beside the loader (the file is untracked in this line).
+
 ---
 
 *Last updated: October 22, 2026 — settings document on test-mcp (0d02061) and folded

supabase_client.py

@@ -823,6 +823,44 @@ def fetch_mcp_run(run_id: str) -> Optional[Dict[str, Any]]:
         return None
 
 
+def fetch_mcp_intake_row(run_id: str) -> Optional[Dict[str, Any]]:
+    """The newest demo_history intake row (status='mcp_intake') of one MCP run, AS
+    STORED: {id, created_at, company_name, use_case, results}. Unlike fetch_mcp_run
+    this keeps the row's own id and created_at — the id is what the reporting
+    loader's LOAD_LEDGER records as SOURCE_INTAKE_ID, so the Snowflake write-through
+    (demoprep_app.reporting.writer) and the loader agree on which request a
+    DEMO_RUN row belongs to, and created_at is that row's REQUESTED_AT. None when
+    Supabase is off, there is no such row, or the lookup fails — never raises.
+    """
+    try:
+        client = SupabaseSettings()
+        if not getattr(client, "enabled", False) or not (run_id or "").strip():
+            return None
+        resp = (
+            client.client.table("demo_history")
+            .select("id,created_at,company_name,use_case,results")
+            .eq("status", "mcp_intake")
+            .eq("results->>run_id", run_id.strip())
+            .order("created_at", desc=True)
+            .limit(1)
+            .execute()
+        )
+        rows = getattr(resp, "data", None) or []
+        if not rows:
+            return None
+        row = dict(rows[0])
+        if isinstance(row.get("results"), str):
+            try:
+                row["results"] = json.loads(row["results"])
+            except Exception:
+                row["results"] = {}
+        row["results"] = row.get("results") or {}
+        return row
+    except Exception as e:
+        print(f"⚠️  mcp intake lookup failed for {run_id!r}: {e}", flush=True)
+        return None
+
+
 def log_mcp_payload(payload: Dict[str, Any]) -> bool:
     """Durably persist a raw MCP intake payload (the full brief + metadata) so real
     research payloads survive container restarts / log rotation — i.e. so you can tell
@@ -874,6 +912,69 @@ def log_mcp_payload(payload: Dict[str, Any]) -> bool:
         return False
 
 
+def mcp_result_record(payload: Dict[str, Any]) -> Dict[str, Any]:
+    """The `results` JSON of an mcp_result row, shaped from a build's result payload.
+
+    ONE shaping for its two readers: log_mcp_result stores it, and the Snowflake
+    reporting write-through (demoprep_app.reporting.writer) reads this same dict
+    the moment a build completes — exactly what the hand-run loader reads back from
+    the row later, so the two never disagree about a run.
+    """
+    results = {
+        "run_id": payload.get("run_id"),
+        "build_status": payload.get("status"),
+        # Which deployment produced this result: the HF Space and the
+        # commit it ran (demoprep_app.deployment_identity). The intake
+        # row and the session_logs 'run started' event carry the same
+        # four keys, so a run ties to a Space and a build by lookup,
+        # not by inference from result shapes.
+        "space_id": payload.get("space_id") or "",
+        "space_host": payload.get("space_host") or "",
+        "code": payload.get("code") or "",
+        "code_source": payload.get("code_source") or "",
+        "owner_email": payload.get("owner_email"),
+        "ts_url": payload.get("ts_url"),
+        "schema": payload.get("schema"),
+        "model_guid": payload.get("model_guid"),
+        "liveboard_guid": payload.get("liveboard_guid"),
+        "model_url": payload.get("model_url"),
+        "liveboard_url": payload.get("liveboard_url"),
+        # The branded customer-facing site built from the liveboard. This
+        # is the deliverable an SE actually sends, so it must be
+        # recoverable from the durable record and not only from the
+        # return value of the call that created it.
+        "portal_url": payload.get("portal_url") or "",
+        # The editable-source zip beside it. Same reasoning: the SE who
+        # wants to tailor the site needs to find this link later, not
+        # only in the return value of the call that built it.
+        "download_url": payload.get("download_url") or "",
+        "warnings": payload.get("warnings") or [],
+        # The structured records behind the warnings (code, severity,
+        # measure, fix) — see demoprep_app/quality/findings.py.
+        "findings": payload.get("findings") or [],
+        # Did they reach demo_findings? None = nothing to persist,
+        # False = the write failed (RLS 42501 until an insert policy
+        # exists) and the records exist only here and in the run log.
+        "findings_persisted": payload.get("findings_persisted"),
+        "findings_persist_error": payload.get("findings_persist_error") or "",
+        "errors": payload.get("errors") or [],
+        "elapsed_seconds": payload.get("elapsed_seconds"),
+        # The pre-build login check (mcp_server._login_precheck, ISS-22):
+        # outcome, HTTP status, reason. None for runs from before it existed.
+        "login_precheck": payload.get("login_precheck"),
+        # The CREATE TABLE text the build loaded, so a run that died at
+        # the ThoughtSpot step can be finished later from its schema
+        # (mcp_server.deploy_existing_schema) without regenerating.
+        "ddl": payload.get("ddl") or "",
+        "source": "mcp",
+    }
+    # Opt-in: the custom MCP demo walkthrough (markdown) is stored only for a
+    # build that asked for it (custom_mcp_demo), so every other row is unchanged.
+    if "custom_mcp_demo_section" in payload:
+        results["custom_mcp_demo_section"] = payload["custom_mcp_demo_section"] or ""
+    return results
+
+
 def log_mcp_result(payload: Dict[str, Any]) -> bool:
     """Durably persist an MCP build RESULT (terminal status, object GUIDs/URLs,
     errors, timing) so outcomes survive container restarts — the in-memory
@@ -898,61 +999,10 @@ def log_mcp_result(payload: Dict[str, Any]) -> bool:
             "industry": "",
             "use_case": _history_text(payload.get("use_case")),
             "demo_config": {"source": "mcp_result"},
-            "results": {
-                "run_id": payload.get("run_id"),
-                "build_status": payload.get("status"),
-                # Which deployment produced this result: the HF Space and the
-                # commit it ran (demoprep_app.deployment_identity). The intake
-                # row and the session_logs 'run started' event carry the same
-                # four keys, so a run ties to a Space and a build by lookup,
-                # not by inference from result shapes.
-                "space_id": payload.get("space_id") or "",
-                "space_host": payload.get("space_host") or "",
-                "code": payload.get("code") or "",
-                "code_source": payload.get("code_source") or "",
-                "owner_email": payload.get("owner_email"),
-                "ts_url": payload.get("ts_url"),
-                "schema": payload.get("schema"),
-                "model_guid": payload.get("model_guid"),
-                "liveboard_guid": payload.get("liveboard_guid"),
-                "model_url": payload.get("model_url"),
-                "liveboard_url": payload.get("liveboard_url"),
-                # The branded customer-facing site built from the liveboard. This
-                # is the deliverable an SE actually sends, so it must be
-                # recoverable from the durable record and not only from the
-                # return value of the call that created it.
-                "portal_url": payload.get("portal_url") or "",
-                # The editable-source zip beside it. Same reasoning: the SE who
-                # wants to tailor the site needs to find this link later, not
-                # only in the return value of the call that built it.
-                "download_url": payload.get("download_url") or "",
-                "warnings": payload.get("warnings") or [],
-                # The structured records behind the warnings (code, severity,
-                # measure, fix) — see demoprep_app/quality/findings.py.
-                "findings": payload.get("findings") or [],
-                # Did they reach demo_findings? None = nothing to persist,
-                # False = the write failed (RLS 42501 until an insert policy
-                # exists) and the records exist only here and in the run log.
-                "findings_persisted": payload.get("findings_persisted"),
-                "findings_persist_error": payload.get("findings_persist_error") or "",
-                "errors": payload.get("errors") or [],
-                "elapsed_seconds": payload.get("elapsed_seconds"),
-                # The pre-build login check (mcp_server._login_precheck, ISS-22):
-                # outcome, HTTP status, reason. None for runs from before it existed.
-                "login_precheck": payload.get("login_precheck"),
-                # The CREATE TABLE text the build loaded, so a run that died at
-                # the ThoughtSpot step can be finished later from its schema
-                # (mcp_server.deploy_existing_schema) without regenerating.
-                "ddl": payload.get("ddl") or "",
-                "source": "mcp",
-            },
+            "results": mcp_result_record(payload),
             "status": "mcp_result",
             "created_at": datetime.utcnow().isoformat(),
         }
-        # Opt-in: the custom MCP demo walkthrough (markdown) is stored only for a
-        # build that asked for it (custom_mcp_demo), so every other row is unchanged.
-        if "custom_mcp_demo_section" in payload:
-            row["results"]["custom_mcp_demo_section"] = payload["custom_mcp_demo_section"] or ""
         client.client.table("demo_history").insert(row).execute()
         return True
     except Exception as e:

tests/test_deploy_existing_schema.py

@@ -66,7 +66,10 @@ def test_the_ddl_is_kept_on_the_run_records_but_left_out_of_status():
     # _last_build (always had it) + _recent_builds + log_mcp_result
     assert '"ddl": out["ddl"],' in src and src.count('"ddl": out["ddl"]') == 3
     assert 'if k != "ddl"' in inspect.getsource(mcp_server.status)
-    sb = inspect.getsource(supabase_client.log_mcp_result)
+    # The row's `results` are shaped by mcp_result_record (shared with the Snowflake
+    # reporting write-through since 2026-10-26); log_mcp_result stores that dict.
+    assert '"results": mcp_result_record(payload)' in inspect.getsource(supabase_client.log_mcp_result)
+    sb = inspect.getsource(supabase_client.mcp_result_record)
     assert '"ddl": payload.get("ddl") or ""' in sb
 
 

tests/test_portal_url_round_trip.py

@@ -27,6 +27,25 @@ def _build_result_source() -> str:
     return inspect.getsource(mcp_server._run_build)
 
 
+def _durable_payload_block() -> str:
+    """The dict _run_build hands log_mcp_result. Since the Snowflake reporting
+    write-through (2026-10-26) it is built first as `result_payload` and then
+    stored — and handed to the write-through — so the block is that literal."""
+    src = _build_result_source()
+    assert "_persisted_result = log_mcp_result(result_payload)" in src, \
+        "the durable record is no longer the result_payload dict"
+    return src.split("result_payload = {", 1)[1].split("_persisted_result = log_mcp_result(result_payload)", 1)[0]
+
+
+def _persisted_row_source() -> str:
+    """The `results` row log_mcp_result stores is shaped by mcp_result_record
+    (shared with the write-through) — check the shaping, and that it is stored."""
+    from supabase_client import log_mcp_result, mcp_result_record
+    assert '"results": mcp_result_record(payload)' in inspect.getsource(log_mcp_result), \
+        "log_mcp_result no longer stores the mcp_result_record row"
+    return inspect.getsource(mcp_result_record)
+
+
 def test_status_view_carries_portal_url():
     """_recent_builds is what status() serves; it must include the site."""
     src = _mcp_source()
@@ -36,15 +55,13 @@ def test_status_view_carries_portal_url():
 
 def test_durable_result_payload_carries_portal_url():
     """The payload handed to log_mcp_result must include the site."""
-    src = _build_result_source()
-    block = src.split("log_mcp_result({", 1)[1].split("})", 1)[0]
+    block = _durable_payload_block()
     assert "portal_url" in block, "the durable record would omit the built site"
 
 
 def test_persisted_row_carries_portal_url():
     """log_mcp_result rebuilds `results` from its own key list — check it too."""
-    from supabase_client import log_mcp_result
-    src = inspect.getsource(log_mcp_result)
+    src = _persisted_row_source()
     assert '"portal_url"' in src, "demo_history rows would omit the built site"
 
 
@@ -73,12 +90,10 @@ def test_status_view_carries_download_url():
 
 
 def test_durable_result_payload_carries_download_url():
-    src = _build_result_source()
-    block = src.split("log_mcp_result({", 1)[1].split("})", 1)[0]
+    block = _durable_payload_block()
     assert "download_url" in block, "the durable record would omit the source zip"
 
 
 def test_persisted_row_carries_download_url():
-    from supabase_client import log_mcp_result
-    src = inspect.getsource(log_mcp_result)
+    src = _persisted_row_source()
     assert '"download_url"' in src, "demo_history rows would omit the source zip"

tests/test_reporting_writer.py

@@ -0,0 +1,398 @@
+"""The Snowflake reporting write-through: a production run reaches DEMO_RUN /
+DEMO_RESULTS the moment its build completes; everything else is left to the loader.
+
+Run: pytest tests/test_reporting_writer.py --noconftest
+
+Three layers, none touching Snowflake or Supabase:
+  1. the rule (boone 2026-10-26): real Salesforce id AND environment prod;
+  2. the rows and statements, against a fake cursor — the same semantics as the
+     hand-run loader (dev_notes/reporting_load_runs.py.txt): ledger lookup,
+     COALESCE on the opportunity id, sticky IS_TEST_RUN, the derived model URL;
+  3. the server hook: it never runs for a non-production run, runs on its own
+     thread for a production run, and a failing writer never escapes.
+"""
+from __future__ import annotations
+
+import os
+import threading
+
+os.environ.setdefault("MCP_TRANSPORT", "stdio")
+
+import pytest
+
+from demoprep_app.reporting import writer as W
+
+REAL_18 = "006Hs00000AbCdEfQA"   # well-formed: 18 alphanumerics
+REAL_15 = "006Hs00000AbCdE"      # well-formed: 15 alphanumerics
+MIKE = "mike.boone@thoughtspot.com"
+
+
+# --- 1. the rule ---------------------------------------------------------------- #
+@pytest.mark.parametrize("opportunity, environment, expected", [
+    (REAL_18, "prod", True),
+    (REAL_15, "prod", True),
+    (REAL_18, "production", True),
+    (REAL_18, " Prod ", True),
+    (REAL_18, "staging", False),
+    (REAL_18, "stage", False),
+    ("TEST-NOT-A-DEAL", "prod", False),
+    ("REPLAY-NOT-A-DEAL-ACME", "prod", False),
+    ("XXTEST000000000001", "prod", False),
+    ("xxtest000000000001", "prod", False),     # the marker is case-insensitive
+    ("", "prod", False),
+    (None, "prod", False),
+    ("   ", "prod", False),
+    ("006Hs00000", "prod", False),            # wrong length
+    ("006Hs00000AbCdEfQA1", "prod", False),   # 19 characters
+    ("006Hs00000-bCdEfQA", "prod", False),    # not alphanumeric
+    (REAL_18, "", False),
+    (REAL_18, None, False),
+    (REAL_18, "dev", False),
+])
+def test_is_production_run(opportunity, environment, expected):
+    assert W.is_production_run(opportunity, environment) is expected
+
+
+def test_normalize_environment_matches_the_tool_synonyms():
+    assert [W.normalize_environment(v) for v in ("prod", "PRODUCTION", "staging", "Stage", "qa", "")] == \
+        ["prod", "prod", "staging", "staging", "", ""]
+
+
+def test_opportunity_id_is_the_loaders_opp_id():
+    assert W.opportunity_id(f"  {REAL_18} ") == REAL_18
+    assert W.opportunity_id("NONE") is None
+    assert W.opportunity_id("") is None
+
+
+@pytest.mark.parametrize("owner, company, reason", [
+    ("", "Acme", "no owner"),
+    ("testrunner@thoughtspot.com", "Acme", "testrunner owner"),
+    ("no.such.person.42@thoughtspot.com", "Acme", "no.such.person owner"),
+    ("sam@thoughtspot.com", "Acme IGNORE", "IGNORE company"),
+    ("sam@thoughtspot.com", "Northwind Retail", "internal company"),
+    ("sam@thoughtspot.com", "Acme", None),
+    (MIKE, "Acme", None),
+])
+def test_skip_reason_is_the_loaders_filter(owner, company, reason):
+    assert W.skip_reason(owner, company) == reason
+
+
+# --- 2. the rows ------------------------------------------------------------------ #
+def _intake(owner=MIKE, company="Acme Logistics", opp=REAL_18, space="thoughtspot-demoprep/mcp",
+            ts_url=""):
+    return {
+        "id": "a76030cc-b070-4653-9264-0a904eed2dae",
+        "created_at": "2026-10-27T00:07:22.393086+00:00",
+        "company_name": company,
+        "use_case": "Fleet utilisation analytics",
+        "results": {
+            "run_id": "4de504402486", "owner_email": owner, "ts_url": ts_url,
+            "salesforce_opportunity_id": opp, "environment": "prod",
+            "space_id": space, "code": "0bcf58064368", "brief": "...",
+        },
+    }
+
+
+def _result(status="success", errors=(), model_url="", ts="https://se-thoughtspot-cloud.thoughtspot.cloud"):
+    return {
+        "run_id": "4de504402486", "build_status": status, "owner_email": MIKE,
+        "ts_url": ts, "schema": "ACME_LOGISTICS_FLEET_20261027",
+        "model_guid": "11111111-2222-3333-4444-555555555555",
+        "liveboard_guid": "66666666-7777-8888-9999-000000000000",
+        "model_url": model_url,
+        "liveboard_url": f"{ts}/#/pinboard/66666666-7777-8888-9999-000000000000",
+        "portal_url": " https://thoughtspot-demoprep-portals.static.hf.space/acme/ ",
+        "errors": list(errors), "elapsed_seconds": 912.4,
+    }
+
+
+def test_shape_rows_mirrors_the_loader_for_a_success():
+    run, res = W.shape_rows(_intake(), _result(), "2026-10-27T00:22:40+00:00")
+    assert run == {"INTAKE_ID": "a76030cc-b070-4653-9264-0a904eed2dae", "RUN_ID": "4de504402486",
+                   "SFDC": REAL_18, "EMAIL": MIKE, "PROSPECT": "Acme Logistics"}
+    assert res["INTAKE_ID"] == run["INTAKE_ID"]
+    assert res["USE_CASE"] == "Fleet utilisation analytics"
+    assert res["REQUESTED_AT"] == "2026-10-27T00:07:22.393086+00:00"
+    assert res["COMPLETED_AT"] == "2026-10-27T00:22:40+00:00"
+    assert res["ELAPSED"] == 912.4
+    assert res["STATUS"] == "success" and res["ERR_CAT"] is None and res["ERR_TXT"] is None
+    assert res["SPACE"] == "prod" and res["CLUSTER_NAME"] == "se-thoughtspot-cloud"
+    assert res["IS_TEST"] is True                       # Mike's runs are test runs
+    assert res["LB_GUID"] == "66666666-7777-8888-9999-000000000000"
+    assert res["LB_URL"].endswith("/#/pinboard/66666666-7777-8888-9999-000000000000")
+    assert res["MODEL_GUID"] == "11111111-2222-3333-4444-555555555555"
+    # No model_url in the result: derived from the resolved ts_url + model guid, as the loader does.
+    assert res["MODEL_URL"] == ("https://se-thoughtspot-cloud.thoughtspot.cloud"
+                                "/#/data/tables/11111111-2222-3333-4444-555555555555")
+    assert res["PORTAL_URL"] == "https://thoughtspot-demoprep-portals.static.hf.space/acme/"  # stripped
+    assert res["SCHEMA_NAME"] == "ACME_LOGISTICS_FLEET_20261027"
+
+
+def test_shape_rows_for_another_se_is_not_a_test_run_and_keeps_its_own_cluster():
+    run, res = W.shape_rows(_intake(owner="Sam.Jones@thoughtspot.com", space="thoughtspot-demoprep/test-mcp"),
+                            _result(ts="https://sebe.thoughtspot.cloud"), "2026-10-27T00:22:40+00:00")
+    assert run["EMAIL"] == "sam.jones@thoughtspot.com"   # lower-cased, as the loader stores it
+    assert res["IS_TEST"] is False
+    assert res["SPACE"] == "test" and res["CLUSTER_NAME"] == "sebe"
+
+
+def test_shape_rows_for_a_failure_buckets_the_first_error():
+    _, res = W.shape_rows(_intake(), _result(status="failed", errors=["LLM call timed out after 600s", "x"]),
+                          "2026-10-27T00:22:40+00:00")
+    assert res["STATUS"] == "failed"
+    assert res["ERR_CAT"] == "llm_timeout"
+    assert res["ERR_TXT"] == "LLM call timed out after 600s"
+
+
+def test_shape_rows_with_no_result_record_is_no_result():
+    _, res = W.shape_rows(_intake(), {}, "2026-10-27T00:22:40+00:00")
+    assert res["STATUS"] == "no_result" and res["COMPLETED_AT"] is None and res["MODEL_URL"] is None
+
+
+def test_shape_rows_rejects_a_malformed_opportunity_id_as_null_never_truncated():
+    run, _ = W.shape_rows(_intake(opp="006Hs00000AbCdEfQA-EXTRA"), _result(), None)
+    assert run["SFDC"] is None
+
+
+# --- the statements, against a fake cursor -------------------------------------- #
+class FakeCursor:
+    """Records every statement; answers the two SELECTs from `answers`."""
+
+    def __init__(self, ledger_key=None, inserted_key=777):
+        self.calls: list[tuple[str, tuple]] = []
+        self._ledger_key = ledger_key
+        self._inserted_key = inserted_key
+        self._next = None
+        self.closed = False
+
+    def execute(self, sql, params=None):
+        self.calls.append((sql, tuple(params or ())))
+        if sql == W.SQL_LEDGER_LOOKUP:
+            self._next = (self._ledger_key,) if self._ledger_key is not None else None
+        elif sql == W.SQL_RUN_KEY_OF_INSERT:
+            self._next = (self._inserted_key,)
+        else:
+            self._next = None
+
+    def fetchone(self):
+        return self._next
+
+    def close(self):
+        self.closed = True
+
+
+def _rows():
+    return W.shape_rows(_intake(), _result(), "2026-10-27T00:22:40+00:00")
+
+
+def test_a_new_request_is_inserted_then_ledgered_then_merged():
+    cur = FakeCursor(ledger_key=None, inserted_key=5131)
+    run, res = _rows()
+    out = W.write_run(cur, run, res)
+    assert out == {"demo_run_key": 5131, "action": "inserted"}
+    sqls = [c[0] for c in cur.calls]
+    assert sqls == [W.SQL_LEDGER_LOOKUP, W.SQL_RUN_INSERT, W.SQL_RUN_KEY_OF_INSERT,
+                    W.SQL_LEDGER_INSERT, W.SQL_RESULTS_MERGE]
+    assert cur.calls[0][1] == (run["INTAKE_ID"],)
+    assert cur.calls[1][1] == ("4de504402486", REAL_18, MIKE, "Acme Logistics")
+    assert cur.calls[3][1] == (run["INTAKE_ID"], 5131)          # the ledger row the loader will find
+    merge_params = cur.calls[4][1]
+    assert merge_params[0] == 5131
+    assert merge_params[1:] == tuple(res[k] for k in W._RES_ORDER)
+    assert len(merge_params) == 1 + 16 == W.SQL_RESULTS_MERGE.count("%s")
+
+
+def test_a_request_the_ledger_knows_is_updated_not_inserted():
+    cur = FakeCursor(ledger_key=4242)
+    run, res = _rows()
+    out = W.write_run(cur, run, res)
+    assert out == {"demo_run_key": 4242, "action": "updated"}
+    sqls = [c[0] for c in cur.calls]
+    assert sqls == [W.SQL_LEDGER_LOOKUP, W.SQL_RUN_UPDATE, W.SQL_RESULTS_MERGE]
+    assert cur.calls[1][1] == ("4de504402486", REAL_18, MIKE, "Acme Logistics", 4242)
+    assert cur.calls[2][1][0] == 4242
+
+
+def test_the_statements_keep_the_loaders_rules():
+    # a human-entered opportunity id is never overwritten with NULL
+    assert "SALESFORCE_OPPORTUNITY_ID=COALESCE(SALESFORCE_OPPORTUNITY_ID,%s)" in W.SQL_RUN_UPDATE
+    # IS_TEST_RUN is sticky; artifact links are COALESCEd; both branches of the MERGE exist
+    m = W.SQL_RESULTS_MERGE
+    assert "IS_TEST_RUN=(t.IS_TEST_RUN OR s.IS_TEST)" in m
+    for col in ("LIVEBOARD_GUID", "LIVEBOARD_URL", "MODEL_GUID", "MODEL_URL", "PORTAL_URL", "SNOWFLAKE_SCHEMA"):
+        assert f"{col}=COALESCE(" in m, col
+    assert "WHEN MATCHED THEN UPDATE" in m and "WHEN NOT MATCHED THEN INSERT" in m
+    # every statement names the reporting schema outright — no USE SCHEMA, no demo-database guess
+    for sql in (W.SQL_LEDGER_LOOKUP, W.SQL_RUN_UPDATE, W.SQL_RUN_INSERT, W.SQL_RUN_KEY_OF_INSERT,
+                W.SQL_LEDGER_INSERT, W.SQL_RESULTS_MERGE):
+        assert "DEMOPREP_REPORTING.REPORTING." in sql
+    # the ledger is keyed by the intake row, the FK the loader joins on
+    assert "SOURCE_INTAKE_ID" in W.SQL_LEDGER_LOOKUP and "SOURCE_INTAKE_ID" in W.SQL_LEDGER_INSERT
+
+
+def test_an_insert_that_cannot_be_keyed_is_an_error_not_a_silent_half_row():
+    cur = FakeCursor(ledger_key=None, inserted_key=None)
+    run, res = _rows()
+    with pytest.raises(RuntimeError):
+        W.write_run(cur, run, res)
+    assert W.SQL_LEDGER_INSERT not in [c[0] for c in cur.calls]
+
+
+# --- write_completed_build: the orchestration --------------------------------- #
+class FakeConn:
+    def __init__(self, cur, fail=False):
+        self.cur, self.fail = cur, fail
+        self.events: list[str] = []
+
+    def autocommit(self, flag):
+        self.events.append(f"autocommit={flag}")
+
+    def cursor(self):
+        if self.fail:
+            raise RuntimeError("boom: Snowflake unreachable")
+        return self.cur
+
+    def commit(self):
+        self.events.append("commit")
+
+    def rollback(self):
+        self.events.append("rollback")
+
+    def close(self):
+        self.events.append("close")
+
+
+def _payload():
+    return {"run_id": "4de504402486", "status": "success", "owner_email": MIKE,
+            "ts_url": "https://se-thoughtspot-cloud.thoughtspot.cloud",
+            "schema": "ACME", "model_guid": "m", "liveboard_guid": "l",
+            "model_url": "mu", "liveboard_url": "lu", "portal_url": "pu",
+            "errors": [], "elapsed_seconds": 10.0}
+
+
+def test_a_production_run_is_written_and_committed():
+    cur = FakeCursor(ledger_key=None, inserted_key=5131)
+    conn = FakeConn(cur)
+    calls = {"connect": 0}
+
+    def connect():
+        calls["connect"] += 1
+        return conn
+
+    out = W.write_completed_build("4de504402486", _payload(), opportunity=REAL_18, environment="prod",
+                                  completed_at="2026-10-27T00:22:40+00:00",
+                                  connect=connect, fetch_intake=lambda rid: _intake(),
+                                  result_record=lambda p: _result())
+    assert out == {"status": "written", "intake_id": "a76030cc-b070-4653-9264-0a904eed2dae",
+                   "demo_run_key": 5131, "action": "inserted"}
+    assert calls["connect"] == 1
+    assert conn.events == ["autocommit=False", "commit", "close"] and cur.closed
+
+
+def test_a_staging_run_never_opens_a_connection():
+    out = W.write_completed_build("4de504402486", _payload(), opportunity=REAL_18, environment="staging",
+                                  connect=lambda: pytest.fail("connected for a staging run"),
+                                  fetch_intake=lambda rid: pytest.fail("looked up a staging run"))
+    assert out["status"] == "skipped" and out["reason"] == "not a production run"
+
+
+def test_a_test_marked_run_never_opens_a_connection():
+    out = W.write_completed_build("4de504402486", _payload(), opportunity="TEST-NOT-A-DEAL", environment="prod",
+                                  connect=lambda: pytest.fail("connected for a test run"),
+                                  fetch_intake=lambda rid: pytest.fail("looked up a test run"))
+    assert out["status"] == "skipped"
+
+
+def test_a_run_without_an_intake_row_is_skipped_like_the_loader_would():
+    out = W.write_completed_build("4de504402486", _payload(), opportunity=REAL_18, environment="prod",
+                                  connect=lambda: pytest.fail("connected with no intake row"),
+                                  fetch_intake=lambda rid: None)
+    assert out == {"status": "skipped", "reason": "no intake row in demo_history"}
+
+
+def test_a_run_the_loader_would_skip_is_skipped_here_too():
+    out = W.write_completed_build("4de504402486", _payload(), opportunity=REAL_18, environment="prod",
+                                  connect=lambda: pytest.fail("connected for a testrunner run"),
+                                  fetch_intake=lambda rid: _intake(owner="testrunner@thoughtspot.com"))
+    assert out == {"status": "skipped", "reason": "testrunner owner"}
+
+
+def test_a_snowflake_failure_is_reported_rolled_back_and_never_raised():
+    conn = FakeConn(FakeCursor(), fail=True)
+    out = W.write_completed_build("4de504402486", _payload(), opportunity=REAL_18, environment="prod",
+                                  connect=lambda: conn, fetch_intake=lambda rid: _intake(),
+                                  result_record=lambda p: _result())
+    assert out["status"] == "failed" and "Snowflake unreachable" in out["error"]
+    assert conn.events == ["autocommit=False", "rollback", "close"]
+
+
+def test_a_connect_failure_is_reported_never_raised():
+    def connect():
+        raise ConnectionError("no route to Snowflake")
+    out = W.write_completed_build("4de504402486", _payload(), opportunity=REAL_18, environment="prod",
+                                  connect=connect, fetch_intake=lambda rid: _intake(),
+                                  result_record=lambda p: _result())
+    assert out == {"status": "failed", "error": "ConnectionError: no route to Snowflake"}
+
+
+# --- 3. the server hook ------------------------------------------------------------ #
+def test_the_hook_skips_a_non_production_run_without_a_thread(monkeypatch):
+    import mcp_server as M
+    monkeypatch.setattr(W, "write_completed_build",
+                        lambda *a, **k: pytest.fail("the writer ran for a staging run"))
+    lines = []
+    monkeypatch.setattr(M, "_log_hit", lines.append)
+    t = M._start_reporting_write_through("abc", {"run_id": "abc"}, opportunity=REAL_18, environment="staging")
+    assert t is None
+    assert any("REPORTING_SKIP run_id=abc" in l for l in lines)
+
+
+def test_the_hook_writes_a_production_run_on_its_own_thread(monkeypatch):
+    import mcp_server as M
+    seen = {}
+
+    def fake_write(run_id, payload, *, opportunity, environment):
+        seen.update(run_id=run_id, payload=payload, opportunity=opportunity, environment=environment,
+                    thread=threading.current_thread().name)
+        return {"status": "written", "demo_run_key": 5131, "action": "inserted"}
+
+    monkeypatch.setattr(W, "write_completed_build", fake_write)
+    lines = []
+    monkeypatch.setattr(M, "_log_hit", lines.append)
+    t = M._start_reporting_write_through("abc", {"run_id": "abc", "status": "success"},
+                                         opportunity=REAL_18, environment="prod")
+    assert t is not None
+    t.join(5)
+    assert not t.is_alive()
+    assert seen["run_id"] == "abc" and seen["opportunity"] == REAL_18 and seen["environment"] == "prod"
+    assert seen["payload"] == {"run_id": "abc", "status": "success"}
+    assert seen["thread"] == "reporting-abc" and seen["thread"] != threading.current_thread().name
+    assert any("REPORTING_WRITE run_id=abc" in l and "'demo_run_key': 5131" in l for l in lines)
+
+
+def test_a_writer_that_raises_never_escapes_the_hook(monkeypatch):
+    import mcp_server as M
+
+    def exploding(*a, **k):
+        raise RuntimeError("writer exploded")
+
+    monkeypatch.setattr(W, "write_completed_build", exploding)
+    lines = []
+    monkeypatch.setattr(M, "_log_hit", lines.append)
+    t = M._start_reporting_write_through("abc", {"run_id": "abc"}, opportunity=REAL_18, environment="prod")
+    t.join(5)
+    assert not t.is_alive()
+    assert any("REPORTING_WRITE run_id=abc" in l and "writer exploded" in l for l in lines)
+
+
+def test_the_result_record_is_the_one_log_mcp_result_stores():
+    """The write-through reads supabase_client.mcp_result_record; the stored row's
+    `results` must be that same dict, or the loader would later disagree with it."""
+    import supabase_client as SC
+    from tests.test_payload_reaches_the_database import captured_result_row
+    payload = {**_payload(), "company_name": "Acme Logistics", "use_case": "Fleet", "warnings": ["w"],
+               "findings": [{"code": "F1"}], "custom_mcp_demo_section": "# demo"}
+    assert captured_result_row(payload) == SC.mcp_result_record(payload)
+    assert SC.mcp_result_record(payload)["build_status"] == "success"
+    assert SC.mcp_result_record(payload)["custom_mcp_demo_section"] == "# demo"
+    assert "custom_mcp_demo_section" not in SC.mcp_result_record(_payload())