feat(mcp): opt-in header tiles — logo_tile and insight_tile

build_demo_from_brief gains two flags, both OFF unless Agent Q sends them
(bool | str = False, read with the portal's _as_bool). Omitted, false or blank:
no header-tile work at all and the liveboard TML is byte-for-byte today's
(pinned against the real enhance_mcp_liveboard run offline).

Insight tile: verified on secloud 2026-10-20 that it IS TML. An exported board
carries it as a visualization with an ai_tile block (target_agent
insight_code_runner, template_id text_long_form, analysis_code = gzip+base64
pandas reading each source viz as <viz_guid>.csv). The narrative is written by
ThoughtSpot at view time, not stored. A TML import creates it (ZZ throwaway
board). DemoPrep generates the analysis from the first tab's KPIs and driver
charts; the code was run against the board's real CSVs.

Logo tile: a note tile in ThoughtSpot's own editor markup — the portal's logo
(its brand-logo.svg, else the builder's wordmark via the new wordmark_svg) as a
data: URI linked to the TSE portal, the tab name in bold, "<use case> at a
glance." and the board's KPI measures in bold. No portal -> company website +
warning; neither -> tile left out + warning.

Placement, first tab, left to right: logo, insight, KPI group. Logo only:
3 | 9. With the insight tile: 2 | 4 | 6 and the KPIs two per row (same width
each as today; the row doubles in height).

Runs in _run_build AFTER the chained portal (the link needs its URL), as one
export -> edit -> import of the finished board. A tile that cannot be built or
that ThoughtSpot rejects is left out with a warning; nothing fails the build.

Suite: 2 failed (pre-existing) / 1400 passed -> 2 failed / 1448 passed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

demoprep_app/liveboards/header_tiles.py

@@ -0,0 +1,823 @@
+"""Opt-in header tiles for the first tab of a DemoPrep liveboard (boone, 2026-10-20).
+
+Two tiles, each OFF unless Agent Q asks for it on `build_demo_from_brief`:
+
+* ``logo_tile`` — a note tile: the company logo (linked to the TSE portal),
+  a bold heading (the tab's name, "Executive Overview"), an italic tagline and a
+  sentence naming the board's headline measures in bold.
+* ``insight_tile`` — ThoughtSpot's AI insight tile. Verified against secloud on
+  2026-10-20: a liveboard TML carries it as a visualization with an ``ai_tile``
+  block (``target_agent: insight_code_runner``, ``template_id: text_long_form``,
+  ``analysis_code``: gzip+base64 pandas code). The code reads each source
+  visualization's data as ``<viz_guid>.csv`` and prints an analysis; ThoughtSpot
+  runs it and writes the narrative card AT VIEW TIME. The narrative text is not
+  stored in the TML, so what we write is the analysis, never the prose. A TML
+  import creates the tile (scratch/header_tiles evidence; ZZ throwaway board).
+
+Order on the first tab, left to right: logo tile, insight tile, KPI group.
+
+They are added AFTER the board is enhanced and after the chained portal build,
+in one export -> edit -> import pass, because:
+
+* the logo links to the portal, whose URL exists only once the portal is built;
+* the insight code must name the board's real viz GUIDs, which only an export
+  of the live board guarantees.
+
+A failure here never fails the build: it comes back as a warning and the tile
+is left out. With both flags off this module is never called.
+"""
+
+from __future__ import annotations
+
+import base64
+import gzip
+import html
+import json
+import re
+from copy import deepcopy
+from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import urlparse
+
+LOGO_TILE_ID = "Viz_header_logo"
+INSIGHT_TILE_ID = "Viz_header_insight"
+
+# What ThoughtSpot itself writes for an insight tile (exported from boone's
+# example board d062e894, 2026-10-20). Not documented; read off a real export.
+INSIGHT_TARGET_AGENT = "insight_code_runner"
+INSIGHT_TEMPLATE_ID = "text_long_form"
+
+# Sources the insight code reads: the headline KPIs plus a few driver charts.
+MAX_TREND_SOURCES = 4
+MAX_BREAKDOWN_SOURCES = 3
+
+# Widths on the 12-column grid. With the insight tile on, the KPI group keeps
+# six columns and its tiles go two per row, so every KPI is exactly as wide as
+# on today's board (four across twelve columns); the row doubles in height to
+# give the narrative room.
+KPI_GROUP_WIDTH_WITH_INSIGHT = 6
+LOGO_WIDTH_WITH_INSIGHT = 2
+LOGO_WIDTH_ALONE = 3
+
+_AGG_PREFIX_RE = re.compile(
+    r"^(total|sum of|sum|average|avg|count of|count|unique count of|unique count|"
+    r"min|max|median)\s+", re.I)
+_TARGET_WORDS = ("target", "goal", "budget", "plan", "quota", "forecast")
+
+
+# --------------------------------------------------------------------- flags
+
+def resolve_header_tile_flags(logo_tile: Any, insight_tile: Any) -> Tuple[bool, bool, List[str]]:
+    """Read the two flags the way the portal flags are read (``_as_bool``).
+
+    Omitted, False, None or blank -> False. An unreadable value is reported
+    and read as False — never as a yes.
+    """
+    from demoprep_app.portal.builder import _as_bool
+
+    notes: List[str] = []
+    logo = _as_bool(logo_tile, "logo_tile", False, notes)
+    insight = _as_bool(insight_tile, "insight_tile", False, notes)
+    return logo, insight, notes
+
+
+# ------------------------------------------------------------ TML inspection
+
+def _column_name(column: Any) -> str:
+    if isinstance(column, str):
+        return column
+    if isinstance(column, dict):
+        for key in ("name", "column_id", "id"):
+            if column.get(key):
+                return str(column[key])
+    return ""
+
+
+def _answer_columns(answer: Dict[str, Any]) -> List[str]:
+    return [n for n in (_column_name(c) for c in answer.get("answer_columns") or []) if n]
+
+
+def _is_measure(name: str) -> bool:
+    from demoprep_app.liveboards.golden_layout import _is_measure_name
+
+    return _is_measure_name(name)
+
+
+def _is_date(name: str) -> bool:
+    from demoprep_app.liveboards.golden_layout import _looks_like_date_column
+
+    return _looks_like_date_column(name)
+
+
+def measure_phrase(name: str) -> str:
+    """'Total Units Produced' -> 'units produced'. Acronyms keep their case."""
+    bare = _AGG_PREFIX_RE.sub("", (name or "").strip())
+    words = []
+    for word in bare.split():
+        words.append(word if (len(word) > 1 and word.isupper()) else word.lower())
+    return " ".join(words)
+
+
+def dimension_phrase(name: str) -> str:
+    """'Maintenance Type Name' -> 'maintenance type'; 'Region' -> 'region'."""
+    words = measure_phrase(name).split()
+    if len(words) > 1 and words[-1] == "name":
+        words = words[:-1]
+    return " ".join(words)
+
+
+def _sentence_case(text: str) -> str:
+    words = (text or "").split()
+    out = []
+    for i, word in enumerate(words):
+        if i and not (len(word) > 1 and word.isupper()):
+            word = word.lower()
+        out.append(word)
+    phrase = " ".join(out)
+    return phrase[:1].upper() + phrase[1:]
+
+
+def _aggregation(measure: str) -> str:
+    """'mean' for averages and rates, 'sum' for additive measures."""
+    from demoprep_app.dataset.measure_names import is_intensive_measure
+
+    lowered = measure.lower().strip()
+    if lowered.startswith(("average ", "avg ")):
+        return "mean"
+    return "mean" if is_intensive_measure(_AGG_PREFIX_RE.sub("", measure)) else "sum"
+
+
+def first_tab(liveboard: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    tabs = (liveboard.get("layout") or {}).get("tabs") or []
+    return tabs[0] if tabs and isinstance(tabs[0], dict) else None
+
+
+def _kpi_group(liveboard: Dict[str, Any], tab: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    """The KPI group on the tab: a group named '...Metric...' with a tile on it."""
+    on_tab = {t.get("visualization_id") for t in tab.get("tiles") or []}
+    for group in liveboard.get("groups") or []:
+        if "Metric" in (group.get("name") or "") and group.get("id") in on_tab:
+            return group
+    return None
+
+
+def _ordered_group_vizzes(tab: Dict[str, Any], group_id: str, fallback: List[str]) -> List[str]:
+    for gl in tab.get("group_layouts") or []:
+        if gl.get("id") == group_id:
+            tiles = sorted(gl.get("tiles") or [], key=lambda t: (t.get("y", 0), t.get("x", 0)))
+            return [t.get("visualization_id") for t in tiles if t.get("visualization_id")]
+    return list(fallback)
+
+
+def _tab_viz_ids_in_order(liveboard: Dict[str, Any], tab: Dict[str, Any]) -> List[str]:
+    """Every viz on the tab, groups expanded, top-to-bottom / left-to-right."""
+    groups = {g.get("id"): g for g in liveboard.get("groups") or []}
+    out: List[str] = []
+    for tile in sorted(tab.get("tiles") or [], key=lambda t: (t.get("y", 0), t.get("x", 0))):
+        vid = tile.get("visualization_id")
+        if vid in groups:
+            out.extend(_ordered_group_vizzes(tab, vid, groups[vid].get("visualizations") or []))
+        elif vid:
+            out.append(vid)
+    return out
+
+
+def board_story(liveboard: Dict[str, Any]) -> Dict[str, Any]:
+    """What the first tab says: its name, KPI vizzes, driver charts, measures, dimensions."""
+    tab = first_tab(liveboard)
+    if tab is None:
+        return {"tab_name": "", "kpis": [], "drivers": [], "measures": [], "dimensions": []}
+    by_id = {v.get("id"): v for v in liveboard.get("visualizations") or []}
+    kpi_group = _kpi_group(liveboard, tab)
+    kpi_ids = set(_ordered_group_vizzes(tab, kpi_group["id"], kpi_group.get("visualizations") or [])
+                  if kpi_group else [])
+    kpis, drivers, measures, dimensions = [], [], [], []
+    for vid in _tab_viz_ids_in_order(liveboard, tab):
+        viz = by_id.get(vid) or {}
+        answer = viz.get("answer")
+        if not isinstance(answer, dict):
+            continue                       # note / ai tiles carry no data
+        cols = _answer_columns(answer)
+        dates = [c for c in cols if _is_date(c)]
+        ms = [c for c in cols if c not in dates and _is_measure(c)]
+        attrs = [c for c in cols if c not in dates and c not in ms]
+        chart_type = ((answer.get("chart") or {}).get("type") or "").upper()
+        entry = {"id": vid, "viz_guid": viz.get("viz_guid") or "", "title": answer.get("name") or vid,
+                 "date": dates[0] if dates else "", "measures": ms, "attributes": attrs}
+        if vid in kpi_ids or chart_type == "KPI":
+            if not ms and len(cols) >= 2:
+                # A KPI's measure may lack an aggregation prefix; it is the non-date column.
+                ms = [c for c in cols if c not in dates][:1]
+                entry["measures"] = ms
+            attrs = []                     # a KPI has no breakdown dimension
+            kpis.append(entry)
+        elif ms and attrs:
+            drivers.append(entry)
+        for m in ms:
+            if m not in measures:
+                measures.append(m)
+        for a in attrs:
+            if a not in dimensions:
+                dimensions.append(a)
+    kpi_measures = [k["measures"][0] for k in kpis if k["measures"]]
+    return {"tab_name": tab.get("name") or "", "kpis": kpis, "drivers": drivers,
+            "measures": kpi_measures or measures, "dimensions": dimensions}
+
+
+# ----------------------------------------------------------------- logo tile
+
+def normalize_link(url: str) -> str:
+    """An http(s) URL with a host, or "" when it cannot be one."""
+    text = (url or "").strip()
+    if not text:
+        return ""
+    if "://" not in text:
+        text = "https://" + text
+    parsed = urlparse(text)
+    if parsed.scheme not in ("http", "https") or not parsed.netloc or " " in parsed.netloc:
+        return ""
+    return text
+
+
+def logo_tile_html(*, logo_svg: str, link: str, company: str, heading: str,
+                   tagline: str, measures: List[str], dimensions: List[str]) -> str:
+    """The note tile, in the markup ThoughtSpot's own note editor writes for a
+    linked image + bold heading + italic line + paragraph (read off the Viz_23
+    logo tile on boone's example board)."""
+    if not logo_svg.strip():
+        raise ValueError("logo SVG is empty")
+    if not link:
+        raise ValueError("logo tile needs a link")
+    src = "data:image/svg+xml;base64," + base64.b64encode(logo_svg.encode("utf-8")).decode("ascii")
+    esc = lambda s: html.escape(str(s), quote=True)  # noqa: E731
+    span = lambda s: f'<span style="white-space: pre-wrap;">{esc(s)}</span>'  # noqa: E731
+    bold = lambda s: (f'<b><strong class="theme-module__editor-text-bold" '  # noqa: E731
+                      f'style="white-space: pre-wrap;">{esc(s)}</strong></b>')
+
+    phrases = [measure_phrase(m) for m in measures[:3] if measure_phrase(m)]
+    parts: List[str] = []
+    if phrases:
+        parts.append(span("Tracks "))
+        for i, phrase in enumerate(phrases):
+            if i:
+                parts.append(span(", and " if i == len(phrases) - 1 and len(phrases) > 2
+                                  else " and " if i == len(phrases) - 1 else ", "))
+            parts.append(bold(phrase))
+        parts.append(span(". "))
+    dims = [dimension_phrase(d) for d in dimensions[:3] if dimension_phrase(d)]
+    if dims:
+        listed = dims[0] if len(dims) == 1 else ", ".join(dims[:-1]) + " and " + dims[-1]
+        parts.append(span(f"Drill into {listed} to see what drives each one."))
+    else:
+        parts.append(span("Drill into the charts below to see what drives each one."))
+
+    return (
+        '<p class="theme-module__editor-paragraph">'
+        f'<a href="{esc(link)}" target="_blank" rel="noopener noreferrer" data-image-link="true" '
+        'style="width: 100%; max-width: 100%;">'
+        f'<img src="{src}" alt="{esc(company)} logo" '
+        'style="width: 100%; height: inherit; max-width: 100%;">'
+        f'<span data-link-overlay="true">{esc(link)}</span></a></p>'
+        '<p class="theme-module__editor-paragraph"><br></p>'
+        f'<p class="theme-module__editor-paragraph">{bold(heading)}</p>'
+        '<p class="theme-module__editor-paragraph"><i><em class="theme-module__editor-text-italic" '
+        f'style="white-space: pre-wrap;">{esc(tagline)}</em></i></p>'
+        f'<p class="theme-module__editor-paragraph">{"".join(parts)}</p>'
+    )
+
+
+def portal_logo_svg(company: str, portal: Optional[Dict[str, Any]]) -> Tuple[str, str]:
+    """The logo the TSE portal shows, as SVG text, plus where it came from.
+
+    A built portal leaves the exact file it shipped in its work dir
+    (builder.build_portal writes ``brand-logo.svg`` there: the caller's logo
+    when one was given, else the generated wordmark). Without a portal, the
+    wordmark the portal builder would have drawn, with its default primary.
+    """
+    from pathlib import Path
+
+    from demoprep_app.portal.builder import DEFAULT_PRIMARY, wordmark_svg
+
+    work_dir = (portal or {}).get("work_dir") or ""
+    if work_dir:
+        path = Path(work_dir) / "brand-logo.svg"
+        if path.is_file():
+            return path.read_text(encoding="utf-8"), "the TSE portal's logo"
+    return wordmark_svg(company, DEFAULT_PRIMARY), "the portal builder's wordmark"
+
+
+# -------------------------------------------------------------- insight tile
+
+def _slug(text: str, used: set) -> str:
+    base = re.sub(r"[^a-z0-9]+", "_", (text or "").lower()).strip("_")[:40] or "viz"
+    key, n = base, 2
+    while key in used:
+        key, n = f"{base}_{n}", n + 1
+    used.add(key)
+    return key
+
+
+def insight_sources(story: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """The vizzes the insight reads: KPI trends first, then driver charts.
+
+    Only vizzes with a viz_guid can be read — the code names its data files
+    by it — and the GUIDs come from an export of the live board.
+    """
+    used: set = set()
+    out: List[Dict[str, Any]] = []
+    for k in [k for k in story["kpis"] if k["viz_guid"] and k["measures"]][:MAX_TREND_SOURCES]:
+        m = k["measures"][0]
+        out.append({"key": _slug(k["title"], used), "file": f"{k['viz_guid']}.csv",
+                    "title": k["title"], "kind": "trend", "date": k["date"],
+                    "measures": [m], "aggs": [_aggregation(m)], "attributes": []})
+    for d in [d for d in story["drivers"] if d["viz_guid"]][:MAX_BREAKDOWN_SOURCES]:
+        ms = d["measures"][:3]
+        out.append({"key": _slug(d["title"], used), "file": f"{d['viz_guid']}.csv",
+                    "title": d["title"], "kind": "breakdown", "date": d["date"],
+                    "measures": ms, "aggs": [_aggregation(m) for m in ms],
+                    "attributes": d["attributes"][:2]})
+    return out
+
+
+_INSIGHT_BODY = r'''
+_NUMERIC = r"[-+]?[$\u20ac\u00a3(]?[-+]?[\d,]*\.?\d+(?:[eE][-+]?\d+)?[%)]?"
+
+
+def _num(series):
+    if series.dtype != object:
+        return pd.to_numeric(series, errors="coerce")
+    return pd.to_numeric(series.astype(str).str.replace(r"[^\d.\-eE]", "", regex=True), errors="coerce")
+
+
+def _is_numeric(series):
+    if series.dtype != object:
+        return True
+    s = series.dropna().astype(str).str.strip()
+    return bool(len(s)) and s.str.fullmatch(_NUMERIC).mean() > 0.8
+
+
+def _fmt(v):
+    if v is None or (isinstance(v, float) and np.isnan(v)):
+        return "n/a"
+    a = abs(v)
+    if a >= 100:
+        return f"{v:,.0f}"
+    if a >= 1:
+        return f"{v:,.2f}"
+    return f"{v:.4f}".rstrip("0").rstrip(".")
+
+
+def _pct(new, old):
+    if old is None or new is None or pd.isna(old) or pd.isna(new) or old == 0:
+        return None
+    return (new - old) / abs(old) * 100
+
+
+def _load(src):
+    """The viz's rows. Tolerates a report-style preamble ("Data extract produced
+    by ...", applied filters) above the header row."""
+    import csv
+    path = CSV_FILES[src["key"]]
+    with open(path, encoding="utf-8-sig", errors="replace") as fh:
+        head = list(csv.reader(fh.read().splitlines()[:20]))
+    wanted = set(src["measures"] + src["attributes"])
+    widest = max((len(r) for r in head), default=0)
+    start = 0
+    for i, row in enumerate(head):
+        if wanted & {c.strip() for c in row}:
+            start = i
+            break
+    else:
+        start = next((i for i, r in enumerate(head) if len(r) == widest and widest > 1), 0)
+    return pd.read_csv(path, skiprows=start)
+
+
+def _col(df, name, role):
+    """The named column, else the closest by role (CSV headers can differ
+    from answer column names, e.g. 'Monthly Order Date' for 'Month(Order Date)')."""
+    if name in df.columns:
+        return name
+    low = {c.lower(): c for c in df.columns}
+    if name.lower() in low:
+        return low[name.lower()]
+    inner = re.sub(r"^.*\((.*)\)$", r"\1", name).lower()
+    for c in df.columns:
+        if inner and inner in c.lower():
+            return c
+    numeric = [c for c in df.columns if _is_numeric(df[c])]
+    if role == "measure":
+        return numeric[-1] if numeric else None
+    others = [c for c in df.columns if c not in numeric]
+    return others[0] if others else None
+
+
+signals = []
+
+for src in SOURCES:
+    title = src["title"]
+    try:
+        df = _load(src)
+        if src["kind"] == "trend":
+            print(f"=== Trend: {title} ===")
+            mcol = _col(df, src["measures"][0], "measure")
+            dcol = _col(df, src["date"], "date") if src["date"] else None
+            vals = _num(df[mcol])
+            frame = pd.DataFrame({"label": df[dcol].astype(str) if dcol else df.index.astype(str), "v": vals})
+            if dcol:
+                frame["d"] = pd.to_datetime(frame["label"], errors="coerce")
+                if frame["d"].notna().mean() > 0.6:
+                    frame = frame.sort_values("d")
+            frame = frame.dropna(subset=["v"])
+            if len(frame) < 2:
+                print(f"{mcol}: not enough periods to describe a trend")
+                print()
+                continue
+            n = len(frame)
+            last, prev = frame.iloc[-1], frame.iloc[-2]
+            chg = _pct(last["v"], prev["v"])
+            print(f"{mcol}: latest {_fmt(last['v'])} ({last['label']}), previous {_fmt(prev['v'])} ({prev['label']})"
+                  + (f", change {chg:+.1f}%" if chg is not None else ""))
+            if chg is not None:
+                signals.append((min(abs(chg), 60), f"{mcol} moved {chg:+.1f}% in the latest period ({last['label']} vs {prev['label']})."))
+            if n >= 6:
+                recent, prior = frame["v"].iloc[-3:].mean(), frame["v"].iloc[-6:-3].mean()
+                q = _pct(recent, prior)
+                if q is not None:
+                    print(f"Last 3 periods average {_fmt(recent)} vs the 3 before {_fmt(prior)}: {q:+.1f}%")
+                    signals.append((min(abs(q), 60) * 1.2, f"{mcol} averaged {_fmt(recent)} over the last 3 periods, {q:+.1f}% versus the 3 before."))
+            if n >= 15:
+                year_ago = frame["v"].iloc[-15:-12].mean()
+                y = _pct(frame["v"].iloc[-3:].mean(), year_ago)
+                if y is not None:
+                    print(f"Last 3 periods vs the same 3 periods a year earlier ({_fmt(year_ago)}): {y:+.1f}%")
+                    signals.append((min(abs(y), 60) * 1.1, f"{mcol} is {y:+.1f}% year over year (last 3 periods)."))
+            # The first and latest periods are often partial; leave them out of peak/low.
+            body = frame.iloc[1:-1] if n > 4 else frame
+            peak, low = body.loc[body["v"].idxmax()], body.loc[body["v"].idxmin()]
+            print(f"{n} periods from {frame.iloc[0]['label']} to {last['label']}; "
+                  f"peak {_fmt(peak['v'])} in {peak['label']}, low {_fmt(low['v'])} in {low['label']}")
+            print("Note: the first and latest periods may be partial.")
+            print()
+        else:
+            print(f"=== Breakdown: {title} ===")
+            mcols = [_col(df, m, "measure") for m in src["measures"]]
+            pairs = [(c, a) for c, a in zip(mcols, src["aggs"]) if c]
+            for c, _ in pairs:
+                df[c] = _num(df[c])
+            acols = []
+            for name in src["attributes"]:
+                c = _col(df, name, "attribute")
+                if c and c not in acols and c not in mcols:
+                    acols.append(c)
+            for acol in acols:
+                grouped = df.groupby(acol, dropna=False).agg({c: a for c, a in pairs})
+                print(f"By {acol} ({len(grouped)} values):")
+                for c, agg in pairs:
+                    s = grouped[c].dropna().sort_values(ascending=False)
+                    if s.empty:
+                        continue
+                    top_name, top_val = s.index[0], s.iloc[0]
+                    bot_name, bot_val = s.index[-1], s.iloc[-1]
+                    line = f"{c}: highest {top_name} ({_fmt(top_val)}), lowest {bot_name} ({_fmt(bot_val)})"
+                    if agg == "sum" and s.sum():
+                        share = top_val / s.sum() * 100
+                        line += f"; total {_fmt(s.sum())}; {top_name} is {share:.1f}% of it"
+                        signals.append((share / 2, f"{top_name} accounts for {share:.1f}% of {c} across {acol}."))
+                    else:
+                        line += f"; average across {acol} {_fmt(s.mean())}"
+                    if bot_val and bot_val > 0 and len(s) > 1:
+                        ratio = top_val / bot_val
+                        line += f"; spread {ratio:.1f}x"
+                        signals.append((min(ratio * 5, 60), f"{c} ranges {ratio:.1f}x across {acol}, from {bot_name} to {top_name}."))
+                    print(line)
+                    print(s.head(5).apply(_fmt).to_string())
+                targets = [c for c, _ in pairs if any(w in c.lower() for w in TARGET_WORDS)]
+                actuals = [c for c, _ in pairs if c not in targets]
+                if targets and actuals:
+                    t, a = targets[0], actuals[0]
+                    att = (grouped[a] / grouped[t].replace(0, np.nan) * 100).dropna().sort_values(ascending=False)
+                    if not att.empty:
+                        overall = grouped[a].sum() / grouped[t].sum() * 100 if grouped[t].sum() else None
+                        below = int((att < 100).sum())
+                        print(f"Attainment ({a} / {t}): " + (f"overall {overall:.1f}%, " if overall is not None else "")
+                              + f"best {att.index[0]} {att.iloc[0]:.1f}%, worst {att.index[-1]} {att.iloc[-1]:.1f}%, "
+                              + f"{below} of {len(att)} below 100%")
+                        signals.append((80, f"{below} of {len(att)} {acol} values are below target on {a}."))
+            print()
+    except Exception as e:
+        print(f"[SKIPPED {title}] {type(e).__name__}: {e}")
+        print()
+
+print("=== Key Signals ===")
+seen = set()
+rank = 0
+for _, text in sorted(signals, key=lambda s: -s[0]):
+    if text in seen:
+        continue
+    seen.add(text)
+    rank += 1
+    print(f"{rank}. {text}")
+    if rank == 5:
+        break
+if not rank:
+    print("No signal could be computed from the source visualizations.")
+'''
+
+
+def insight_analysis_code(company: str, use_case: str, sources: List[Dict[str, Any]]) -> str:
+    """The pandas program the insight tile runs. Same shape as ThoughtSpot's own:
+    a user query, a summary, a CSV_FILES map of '<viz_guid>.csv', then sections."""
+    if not sources:
+        raise ValueError("no source visualizations for the insight tile")
+    query = (f"Write an executive insight for {company}'s {use_case} liveboard: lead with the "
+             "most important change in the headline metrics, quantify it, name the segments "
+             "driving it, and say where to act.")
+    summary = (f"{company} {use_case}: " + ", ".join(
+        s["title"] for s in sources) + ".")
+    csv_lines = "\n".join(f"    {json.dumps(s['key'])}: {json.dumps(s['file'])}," for s in sources)
+    spec = [{k: s[k] for k in ("key", "title", "kind", "date", "measures", "aggs", "attributes")}
+            for s in sources]
+    return (
+        "\nimport re\nimport warnings\nimport pandas as pd\nimport numpy as np\n\n"
+        'warnings.filterwarnings("ignore")\n\n'
+        'print("=== User Query ===")\n'
+        f"print({json.dumps(query)})\n"
+        "print()\n"
+        'print("=== Analysis Summary ===")\n'
+        f"print({json.dumps(summary)})\n"
+        "print()\n\n"
+        "# --- CSV declarations ---\n"
+        f"CSV_FILES = {{\n{csv_lines}\n}}\n\n"
+        f"SOURCES = {json.dumps(spec, indent=4)}\n\n"
+        f"TARGET_WORDS = {json.dumps(list(_TARGET_WORDS))}\n"
+        + _INSIGHT_BODY
+    )
+
+
+def encode_analysis_code(code: str) -> str:
+    """gzip + base64, as ThoughtSpot stores ``ai_tile.analysis_code``. mtime=0
+    so the same code always encodes to the same text."""
+    return base64.b64encode(gzip.compress(code.encode("utf-8"), mtime=0)).decode("ascii")
+
+
+def decode_analysis_code(encoded: str) -> str:
+    return gzip.decompress(base64.b64decode(encoded)).decode("utf-8")
+
+
+# -------------------------------------------------------------------- layout
+
+def place_header_tiles(liveboard: Dict[str, Any], logo_id: Optional[str],
+                       insight_id: Optional[str]) -> str:
+    """Put the tiles on the first tab: logo, insight, then the KPI group.
+
+    Beside a KPI group that has the top band to itself (every golden-layout
+    board): logo only -> logo 3 cols, KPIs 9 (the layout of boone's example
+    board). With the insight tile -> logo 2, insight 4, KPIs 6 with the KPI
+    tiles two per row, so each KPI keeps today's width and the row doubles in
+    height. Anything else (no KPI group, or other tiles beside it) -> the
+    header tiles take a new row above everything. Returns a one-line summary.
+    """
+    from demoprep_app.liveboards.golden_layout import KPI_ROW_H
+
+    tab = first_tab(liveboard)
+    if tab is None:
+        raise ValueError("the liveboard has no tabs")
+    tiles = tab.setdefault("tiles", [])
+    header = [i for i in (logo_id, insight_id) if i]
+    if not header:
+        return ""
+    group = _kpi_group(liveboard, tab)
+    kpi_tile = next((t for t in tiles if group and t.get("visualization_id") == group["id"]), None)
+    band_mates = []
+    if kpi_tile is not None:
+        k_y, k_h = kpi_tile.get("y", 0), kpi_tile.get("height", KPI_ROW_H)
+        band_mates = [t for t in tiles if t is not kpi_tile
+                      and t.get("y", 0) < k_y + k_h and t.get("y", 0) + t.get("height", 0) > k_y]
+
+    if kpi_tile is None or band_mates:
+        top = min((t.get("y", 0) for t in tiles), default=0)
+        for t in tiles:
+            t["y"] = t.get("y", 0) + KPI_ROW_H
+        x = 0
+        new = []
+        if logo_id:
+            new.append({"visualization_id": logo_id, "x": 0, "y": top, "height": KPI_ROW_H,
+                        "width": LOGO_WIDTH_ALONE})
+            x = LOGO_WIDTH_ALONE
+        if insight_id:
+            new.append({"visualization_id": insight_id, "x": x, "y": top, "height": KPI_ROW_H,
+                        "width": 12 - x})
+        tab["tiles"] = new + tiles
+        return "header tiles placed in a new top row (no KPI row to sit beside)"
+
+    k_y, k_h = kpi_tile.get("y", 0), kpi_tile.get("height", KPI_ROW_H)
+    new_h = k_h
+    if insight_id:
+        layout = next((gl for gl in tab.get("group_layouts") or [] if gl.get("id") == group["id"]), None)
+        kpi_ids = _ordered_group_vizzes(tab, group["id"], group.get("visualizations") or [])
+        tile_h = max([t.get("height", KPI_ROW_H) for t in (layout or {}).get("tiles") or []] or [KPI_ROW_H])
+        per_row = 1 if len(kpi_ids) == 1 else 2
+        width = 12 // per_row
+        reflowed = [{"visualization_id": vid, "x": (i % per_row) * width, "y": (i // per_row) * tile_h,
+                     "height": tile_h, "width": width} for i, vid in enumerate(kpi_ids)]
+        if layout is None:
+            tab.setdefault("group_layouts", []).append({"id": group["id"], "tiles": reflowed})
+        else:
+            layout["tiles"] = reflowed
+        rows = -(-len(kpi_ids) // per_row)
+        new_h = max(k_h, rows * tile_h)
+        logo_w = LOGO_WIDTH_WITH_INSIGHT if logo_id else 0
+        kpi_x = 12 - KPI_GROUP_WIDTH_WITH_INSIGHT
+        placed = []
+        if logo_id:
+            placed.append({"visualization_id": logo_id, "x": 0, "y": k_y, "height": new_h, "width": logo_w})
+        placed.append({"visualization_id": insight_id, "x": logo_w, "y": k_y, "height": new_h,
+                       "width": kpi_x - logo_w})
+    else:
+        kpi_x = LOGO_WIDTH_ALONE
+        placed = [{"visualization_id": logo_id, "x": 0, "y": k_y, "height": k_h, "width": LOGO_WIDTH_ALONE}]
+
+    delta = new_h - k_h
+    for t in tiles:
+        if t is not kpi_tile and t.get("y", 0) >= k_y + k_h:
+            t["y"] = t.get("y", 0) + delta
+    kpi_tile.update({"x": kpi_x, "width": 12 - kpi_x, "height": new_h})
+    index = tiles.index(kpi_tile)
+    tab["tiles"] = tiles[:index] + placed + tiles[index:]
+    order = ", ".join(p["visualization_id"] for p in placed) + f", {group['id']}"
+    return f"header tiles beside the KPI group (left to right: {order})"
+
+
+# --------------------------------------------------------------- the pass
+
+def apply_header_tiles(liveboard_tml: Dict[str, Any], *, company: str, use_case: str,
+                       logo: bool, insight: bool, link: str = "",
+                       logo_svg: str = "") -> Tuple[Dict[str, Any], List[str], List[str]]:
+    """Return (new TML, changes, warnings). Pure: no ThoughtSpot calls.
+
+    With logo and insight both False the input comes back unchanged (same
+    object, untouched). A tile that cannot be built is left out with a warning.
+    """
+    if not (logo or insight):
+        return liveboard_tml, [], []
+    updated = deepcopy(liveboard_tml)
+    lb = updated.get("liveboard") or {}
+    changes: List[str] = []
+    warnings: List[str] = []
+    story = board_story(lb)
+    vizzes = lb.setdefault("visualizations", [])
+    logo_id = insight_id = None
+
+    if logo:
+        try:
+            body = logo_tile_html(
+                logo_svg=logo_svg, link=link, company=company,
+                heading=story["tab_name"] or "Executive Overview",
+                tagline=f"{_sentence_case(use_case or 'Analytics')} at a glance.",
+                measures=story["measures"], dimensions=story["dimensions"])
+            vizzes.append({"id": LOGO_TILE_ID, "note_tile": {"html_parsed_string": body}})
+            logo_id = LOGO_TILE_ID
+            changes.append(f"logo tile added (links to {link})")
+        except Exception as e:  # noqa: BLE001 - a tile never fails the build
+            warnings.append(f"logo tile left out: {e}")
+
+    if insight:
+        try:
+            sources = insight_sources(story)
+            code = insight_analysis_code(company, use_case or "Analytics", sources)
+            vizzes.append({"id": INSIGHT_TILE_ID, "ai_tile": {
+                "target_agent": INSIGHT_TARGET_AGENT,
+                "analysis_code": encode_analysis_code(code),
+                "template_id": INSIGHT_TEMPLATE_ID,
+            }})
+            insight_id = INSIGHT_TILE_ID
+            changes.append("insight tile added (reads: " + ", ".join(s["title"] for s in sources) + ")")
+        except Exception as e:  # noqa: BLE001
+            warnings.append(f"insight tile left out: {e}")
+
+    if not (logo_id or insight_id):
+        return liveboard_tml, [], warnings
+    try:
+        changes.append(place_header_tiles(lb, logo_id, insight_id))
+    except Exception as e:  # noqa: BLE001
+        return liveboard_tml, [], warnings + [f"header tiles left out: could not place them ({e})"]
+    return updated, changes, warnings
+
+
+def logo_link(portal: Optional[Dict[str, Any]], company_url: str,
+              with_portal: bool) -> Tuple[str, List[str]]:
+    """Where the logo links: the TSE portal, else the company website (with a
+    warning saying why), else "" (the tile is left out, with a warning)."""
+    portal_url = normalize_link((portal or {}).get("url") or "")
+    if portal_url:
+        return portal_url, []
+    why = ("the build was run with with_portal=False" if not with_portal
+           else "the portal build did not produce a site")
+    site = normalize_link(company_url)
+    if site:
+        return site, [f"logo tile links to the company website ({site}) instead of a TSE portal: {why}"]
+    return "", [f"logo tile left out: no TSE portal ({why}) and no company website to link to"]
+
+
+def _import_tml(ts_client, liveboard_tml: Dict[str, Any]) -> Dict[str, Any]:
+    """One TML import; {'ok', 'error'}. PARTIAL, as enhance_mcp_liveboard imports."""
+    import yaml
+
+    from liveboard_creator import _import_status_from_response
+
+    response = ts_client.session.post(
+        f"{ts_client.base_url}/api/rest/2.0/metadata/tml/import",
+        json={"metadata_tmls": [yaml.dump(liveboard_tml, default_flow_style=False, sort_keys=False)],
+              "import_policy": "PARTIAL", "create_new": False},
+        timeout=300,
+    )
+    if response.status_code != 200:
+        return {"ok": False, "error": f"HTTP {response.status_code}: {response.text[:500]}"}
+    status = _import_status_from_response(response.json())
+    if status["status_code"] in ("OK", "WARNING"):
+        return {"ok": True, "error": ""}
+    return {"ok": False, "error": status["error_message"] or "unknown TML import error"}
+
+
+def add_header_tiles_to_board(ts_client, liveboard_guid: str, *, company: str, use_case: str,
+                              company_url: str, logo: bool, insight: bool,
+                              with_portal: bool, portal: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    """Export the live board, add the requested tiles, import it back.
+
+    Returns {'changes': [...], 'warnings': [...]} and never raises: a tile that
+    cannot be built or that ThoughtSpot rejects is left out with a warning, and
+    the board stays exactly as the build left it.
+    """
+    out: Dict[str, Any] = {"changes": [], "warnings": []}
+    if not (logo or insight):
+        return out
+    try:
+        link, logo_svg, logo_source = "", "", ""
+        if logo:
+            link, notes = logo_link(portal, company_url, with_portal)
+            out["warnings"].extend(notes)
+            if link:
+                try:
+                    logo_svg, logo_source = portal_logo_svg(company, portal)
+                except Exception as e:  # noqa: BLE001
+                    out["warnings"].append(f"logo tile left out: the logo could not be read ({e})")
+                    link = ""
+            logo = bool(link and logo_svg)
+        if not (logo or insight):
+            return out
+
+        from demoprep_app.liveboards.tml_yaml import load_tml
+
+        response = ts_client.session.post(
+            f"{ts_client.base_url}/api/rest/2.0/metadata/tml/export",
+            json={"metadata": [{"identifier": liveboard_guid}], "export_associated": False},
+            timeout=120,
+        )
+        if response.status_code != 200:
+            out["warnings"].append(f"header tiles not added: liveboard export failed "
+                                   f"(HTTP {response.status_code})")
+            return out
+        tml = load_tml((response.json() or [{}])[0].get("edoc") or "")
+        if "liveboard" not in tml:
+            out["warnings"].append("header tiles not added: the liveboard export had no TML")
+            return out
+
+        new_tml, changes, warnings = apply_header_tiles(
+            tml, company=company, use_case=use_case, logo=logo, insight=insight,
+            link=link, logo_svg=logo_svg)
+        out["warnings"].extend(warnings)
+        if not changes:
+            return out
+
+        result = _import_tml(ts_client, new_tml)
+        if not result["ok"]:
+            # Retry once without whichever of OUR tiles ThoughtSpot named. Any
+            # other rejection leaves the board exactly as the build left it.
+            from liveboard_creator import parse_failed_visualization_ids
+
+            named = set(parse_failed_visualization_ids(result["error"]))
+            ours = named & {LOGO_TILE_ID, INSIGHT_TILE_ID}
+            if not ours or named - ours:
+                out["warnings"].append(f"header tiles not added: ThoughtSpot rejected the import "
+                                       f"({result['error'][:300]})")
+                return out
+            keep_logo = logo and LOGO_TILE_ID not in ours
+            keep_insight = insight and INSIGHT_TILE_ID not in ours
+            for tile in sorted(ours):
+                out["warnings"].append(f"{'logo' if tile == LOGO_TILE_ID else 'insight'} tile left out: "
+                                       f"ThoughtSpot rejected it ({result['error'][:300]})")
+            if not (keep_logo or keep_insight):
+                return out
+            new_tml, changes, warnings = apply_header_tiles(
+                tml, company=company, use_case=use_case, logo=keep_logo, insight=keep_insight,
+                link=link, logo_svg=logo_svg)
+            result = _import_tml(ts_client, new_tml)
+            if not result["ok"]:
+                out["warnings"].append(f"header tiles not added: ThoughtSpot rejected the import "
+                                       f"({result['error'][:300]})")
+                return out
+        out["changes"].extend(c.replace("logo tile added (", f"logo tile added ({logo_source}; ", 1)
+                              for c in changes)
+        return out
+    except Exception as e:  # noqa: BLE001 - header tiles never fail a build
+        out["warnings"].append(f"header tiles not added: {type(e).__name__}: {e}")
+        return out

demoprep_app/portal/builder.py

@@ -1337,10 +1337,12 @@ def validate_spec(spec: dict) -> list[str]:
 
 # ---------------------------------------------------------------- build steps
 
-def wordmark(company: str, primary: str, path: Path) -> None:
+def wordmark_svg(company: str, primary: str) -> str:
+    """The generated wordmark logo, as SVG text. Also the liveboard logo tile's
+    image when there is no portal (demoprep_app/liveboards/header_tiles.py)."""
     label = company.upper()[:26]
     width = 40 + len(label) * 11
-    path.write_text(
+    return (
         f'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {width} 40" '
         f'width="{width}" height="40"><text x="4" y="27" font-family="Inter, '
         f'system-ui, sans-serif" font-size="18" font-weight="700" letter-spacing="0.5" '
@@ -1348,6 +1350,10 @@ def wordmark(company: str, primary: str, path: Path) -> None:
     )
 
 
+def wordmark(company: str, primary: str, path: Path) -> None:
+    path.write_text(wordmark_svg(company, primary))
+
+
 def _tool(name: str) -> str:
     """Absolute path to node/npm. Containers often lack them on the default PATH."""
     found = shutil.which(name) or shutil.which(name, path="/usr/bin:/usr/local/bin:"

docs/thoughtspot_mcp_api_review_2026_09.md

@@ -7,7 +7,7 @@
 ## Summary
 
 1. **Our MCP endpoint is deprecated.** `liveboard_creator.py` calls `https://agent.thoughtspot.app/bearer/mcp`. ThoughtSpot now documents it as deprecated (still served, no sunset date). The replacement is `https://agent.thoughtspot.app/token/mcp?api-version=2025-01-01`, which serves the identical five tools we use today (**verified**: `ping`, `getRelevantQuestions`, `getAnswer` all work). Do **not** switch to the unversioned `/token/mcp`: it resolves to the Spotter 3 toolset, which has no `getAnswer` or `createLiveboard`, and the pipeline would break.
-2. **There is no Spotter Viz API.** SpotterViz is GA as a UI feature inside Liveboard edit mode; the Visual Embed SDK only lets you brand its panel and supply starter prompts. Insight tiles (26.8, early access) and Spotter Analysts (26.8, early access) are UI-only too. The programmatic route ThoughtSpot actually offers is Spotter answer sessions → TML, which is what both MCP `createLiveboard` and the Spotter 3 `create_dashboard` do server-side.
+2. **There is no Spotter Viz API.** SpotterViz is GA as a UI feature inside Liveboard edit mode; the Visual Embed SDK only lets you brand its panel and supply starter prompts. Spotter Analysts (26.8, early access) are UI-only too. Insight tiles (26.8, early access) have no API, but a TML import creates them (verified 2026-10-20; see the table below). The programmatic route ThoughtSpot actually offers is Spotter answer sessions → TML, which is what both MCP `createLiveboard` and the Spotter 3 `create_dashboard` do server-side.
 3. **We can assemble liveboards without MCP.** `metadata/tml/export` accepts `{type: ANSWER, session_identifier, generation_number}` and returns the TML of an *unsaved* Spotter answer in about 0.1 s (**verified**). A liveboard TML built from that answer passed `import_policy: VALIDATE_ONLY` (**verified**). This turns the sprint's "backup TML-only liveboard" into a first-class path with full control over groups, tabs, and styles in a single import.
 4. **New per-model Spotter coaching APIs fit our mission directly.** `ai/instructions/set` (10.15+) writes natural-language rules onto a model (**verified** set/get/restore as the model owner). `ai/memory/export|import` (26.8) moves Spotter's learned rules and recipes between models. Encoding the demo narrative into the model itself makes Spotter's live answers agree with the liveboard story.
 5. **Spotter 3 works on our clusters via REST and MCP.** One question takes 26–27 s and returns a reasoning narrative plus answer sessions usable for liveboards. It is rate-limited (10 conversations/min, 30 messages/min), so it is a spike for one or two "reasoning" tiles, not a replacement for `ai/answer/create` (3 s).
@@ -142,7 +142,7 @@ We hand-roll `requests` calls, which is fine. If we ever consolidate, use the of
 | Capability | Status | Programmatic access |
 |---|---|---|
 | **SpotterViz** builds a Liveboard from a prompt: picks data models, plans the story, creates answers with Spotter, lays out tabs, KPIs, charts, filters, and styling; iterative refinement with checkpoints. Requires Liveboard edit mode, Spotter enabled on the model, TML edit permission, admin enablement; documented as working best with Claude Opus 4.5. | GA in the UI (early access when introduced in 26.7) | **None.** No REST endpoint, no MCP tool. Visual Embed SDK 1.50 `spotterViz` config only customizes branding and `customStarterPrompts`; it cannot submit a prompt. |
-| **Insight tiles**: AI text that re-runs its instructions against live data each time the Liveboard loads. | Early access, 26.8 | None documented; no TML tile type published. Check whether an exported Liveboard with one shows a new visualization type before planning on it. |
+| **Insight tiles**: AI text that re-runs its instructions against live data each time the Liveboard loads. | Early access, 26.8 | **TML, verified 2026-10-20** (no REST endpoint, not in the published TML docs). An exported Liveboard carries the tile as a visualization with an `ai_tile` block: `target_agent: insight_code_runner`, `template_id: text_long_form`, `analysis_code` = gzip+base64 pandas that reads each source viz as `<viz_guid>.csv` and prints an analysis. The narrative is NOT stored: ThoughtSpot writes it at view time. A TML import creates the tile (VALIDATE_ONLY and ALL_OR_NONE both OK on secloud; re-export shows it intact) and does not rewrite the viz GUIDs inside the code. DemoPrep builds one behind `insight_tile` (`demoprep_app/liveboards/header_tiles.py`). |
 | **Spotter Analysts**: named, governed Spotter workspaces scoped to up to 5 models with up to 5,000 characters of instructions, connector allow/deny, and sharing. | Early access, 26.8 | UI only. |
 | **Memory from Liveboards**: Spotter reads up to 3 Liveboards and extracts metric definitions and business rules as memory (10–20 min, async). | GA, needs "Spotter 3 capabilities" enabled by an admin | UI only; the resulting memory can then be exported and imported via `ai/memory/*`. |
 

mcp_server.py

@@ -676,7 +676,8 @@ def _run_build(run_id: str, brief: str, company_name: str, use_case: str, compan
                portal_options: dict | None = None,
                salesforce_opportunity_id: str = "",
                existing_schema: str = "", existing_ddl: str = "",
-               environment: str = "") -> dict:
+               environment: str = "",
+               header_tiles: dict | None = None) -> dict:
     """Drive the controller headlessly (runs in a BACKGROUND THREAD; the caller has
     already registered the build in _active_builds and holds a _build_sem slot).
     With `existing_schema` + `existing_ddl` (deploy_existing_schema) the dataset,
@@ -1031,6 +1032,37 @@ def _run_build(run_id: str, brief: str, company_name: str, use_case: str, compan
             except Exception as e:
                 portal = {"_warnings": [f"portal build failed ({type(e).__name__}): {e}"]}
                 _log_hit(f"portal FAILED for {run_id}: {type(e).__name__}: {e}")
+        # ---- header tiles (logo_tile / insight_tile; boone 2026-10-20) ----
+        # Only when the caller asked. After the portal, so the logo can link to
+        # it; one export -> edit -> import of the finished board. Never fails
+        # the build: a tile that cannot be added is left out and warned about.
+        if header_tiles:
+            _tile_warnings = list(header_tiles.get("warnings") or [])
+            if (header_tiles.get("logo") or header_tiles.get("insight")) and not dc.get("liveboard_guid"):
+                _tile_warnings.append("header tiles not added: the build produced no liveboard")
+            elif header_tiles.get("logo") or header_tiles.get("insight"):
+                _set_progress(run_id, phase="adding header tiles", detail="")
+                try:
+                    from thoughtspot_deployer import ThoughtSpotDeployer
+                    from demoprep_app.liveboards.header_tiles import add_header_tiles_to_board
+                    _ts = ThoughtSpotDeployer(base_url=ts_target_url, username=owner,
+                                              secret_key=ts_auth_key)
+                    if not _ts.authenticate_trusted():
+                        raise RuntimeError(f"could not authenticate to {ts_target_url} as {owner}")
+                    _tiles = add_header_tiles_to_board(
+                        _ts, dc["liveboard_guid"], company=company_name, use_case=use_case,
+                        company_url=company_url, logo=bool(header_tiles.get("logo")),
+                        insight=bool(header_tiles.get("insight")), with_portal=with_portal,
+                        portal=portal)
+                except Exception as e:
+                    _tiles = {"changes": [],
+                              "warnings": [f"header tiles not added: {type(e).__name__}: {e}"]}
+                _tile_warnings.extend(_tiles["warnings"])
+                for _change in _tiles["changes"]:
+                    _log_hit(f"header tiles {run_id}: {_change}")
+            if _tile_warnings:
+                dc["warnings"] = list(dc.get("warnings") or []) + _tile_warnings
+        # ---- end header tiles ----
         # A run that shipped its model but no board (liveboard_creation_path ==
         # 'none', handoff 2026-09-22 item C) is PARTIAL to the caller, not a
         # success: the demo is usable, but the deliverable they asked for is
@@ -1104,6 +1136,10 @@ def build_demo_from_brief(
     *,
     salesforce_opportunity_id: str,
     environment: str,
+    # ---- header tiles (boone 2026-10-20): OFF unless Agent Q sends them ----
+    logo_tile: bool | str = False,
+    insight_tile: bool | str = False,
+    # ---- end header tiles ----
 ) -> dict:
     """Start a full ThoughtSpot demo build from a prospect brief.
 
@@ -1151,6 +1187,22 @@ def build_demo_from_brief(
                 whichever path produced it.
         tiers: The Premium/Basic tier switcher. ON by default, same reason.
         pinning: Add the "Add Report" panel.
+        logo_tile: OFF by default; omitted, false or blank builds exactly what it
+                builds without this argument. True adds a logo tile at the far
+                left of the first tab's top row: the company logo (the TSE
+                portal's own logo) linked to the TSE portal, the tab's name as a
+                heading, a tagline and a line naming the board's headline
+                measures. With no portal (with_portal=False, or the portal build
+                failed) the logo links to company_url and a warning says so; with
+                neither, the tile is left out with a warning.
+        insight_tile: OFF by default; omitted, false or blank builds exactly what
+                it builds without this argument. True adds ThoughtSpot's AI insight
+                tile left of the KPIs (right of the logo tile when both are on):
+                a narrative card ThoughtSpot writes when the board is opened, from
+                an analysis of the board's own KPIs and driver charts. With it on,
+                the KPIs sit two per row so the row has room for the narrative.
+                A tile that cannot be added never fails the build; it is left out
+                and the reason is in the result's warnings.
 
     Not settable here: askMode (it breaks the upstream tab prune) and the home
     page (its copy is written from a brand screenshot). For those, call
@@ -1239,18 +1291,27 @@ def build_demo_from_brief(
     portal_options = _chained_portal_options(
         theme_mode=theme_mode, auth=auth, nav_layout=nav_layout, inline_tab=inline_tab,
         monetize=monetize, tiers=tiers, pinning=pinning)
+    # ---- header tiles (boone 2026-10-20) ----
+    # Read like the portal flags (_as_bool). None when both are off and nothing
+    # needs reporting, so a default build carries no header-tile work at all.
+    from demoprep_app.liveboards.header_tiles import resolve_header_tile_flags
+    _logo_on, _insight_on, _tile_notes = resolve_header_tile_flags(logo_tile, insight_tile)
+    header_tiles = ({"logo": _logo_on, "insight": _insight_on, "warnings": _tile_notes}
+                    if (_logo_on or _insight_on or _tile_notes) else None)
+    # ---- end header tiles ----
 
     def _worker() -> None:
         try:
             _run_build(run_id, b, c, u, url, ts_target_url, ts_auth_key, owner, with_portal, _payload,
                        portal_options=portal_options,
                        salesforce_opportunity_id=salesforce_opportunity_id,
-                       environment=environment)
+                       environment=environment,
+                       header_tiles=header_tiles)
         finally:
             _build_sem.release()
 
     threading.Thread(target=_worker, daemon=True, name=f"build-{run_id}").start()
-    return {
+    started = {
         "status": "started",
         "run_id": run_id,
         "message": ("Build started in the background (~15-20 min). Poll status() for "
@@ -1258,6 +1319,9 @@ def build_demo_from_brief(
                     "portal_url — the branded customer-facing site built from the "
                     "liveboard — and download_url, a zip of its editable source."),
     }
+    if _tile_notes:  # header tiles: an unreadable flag value, read as off
+        started["warnings"] = list(_tile_notes)
+    return started
 
 
 # ------------------------------------------------------- redeploy a schema

tests/test_header_tiles.py

@@ -0,0 +1,601 @@
+"""Opt-in header tiles on the first tab: logo_tile and insight_tile (boone 2026-10-20).
+
+boone's rule for every flag: OFF unless Agent Q sends it. Omitted, false or blank
+means the build is byte-for-byte today's — same liveboard TML, no extra work.
+That is pinned first, against the REAL enhance_mcp_liveboard run offline.
+
+What an insight tile is, read off a real export (boone's example board on
+secloud, d062e894, 2026-10-20): a visualization with an `ai_tile` block —
+target_agent insight_code_runner, template_id text_long_form, analysis_code =
+gzip+base64 pandas that reads each source viz as `<viz_guid>.csv`. ThoughtSpot
+writes the narrative at view time; a TML import creates the tile (verified on a
+throwaway ZZ board — see demoprep_app/liveboards/header_tiles.py).
+
+Run: pytest tests/test_header_tiles.py -v
+"""
+import base64
+import inspect
+import json
+import re
+import subprocess
+import sys
+import threading
+
+import pytest
+import yaml
+
+from demoprep_app.liveboards import header_tiles as HT
+from demoprep_app.liveboards.header_tiles import (
+    INSIGHT_TILE_ID, LOGO_TILE_ID, add_header_tiles_to_board, apply_header_tiles,
+    decode_analysis_code, logo_link, portal_logo_svg, resolve_header_tile_flags,
+)
+
+PORTAL_URL = "https://thoughtspot-demoprep-portals.static.hf.space/acme-ab12cd34/index.html"
+SVG = '<svg xmlns="http://www.w3.org/2000/svg"><text>ACME</text></svg>'
+OFF_VALUES = [None, False, "", "   ", "false", "no", "off", "0"]
+
+
+# --------------------------------------------------------------- the board
+
+def _viz(vid, name, ctype, query, cols, guid=None):
+    viz = {"id": vid, "answer": {"name": name, "search_query": query,
+                                 "answer_columns": [{"name": c} for c in cols],
+                                 "chart": {"type": ctype, "chart_columns": []}}}
+    if guid:
+        viz["viz_guid"] = guid
+    return viz
+
+
+def _mcp_board():
+    """An exported MCP board: every viz carries the viz_guid ThoughtSpot gave it."""
+    return {"guid": "lb-1", "liveboard": {"name": "Acme Board", "visualizations": [
+        _viz("Viz_1", "Revenue Weekly", "LINE", "sum [Revenue] [Date].weekly",
+             ["Week(Date)", "Total Revenue"], "g-rev"),
+        _viz("Viz_2", "Orders Weekly", "LINE", "sum [Orders] [Date].weekly",
+             ["Week(Date)", "Total Orders"], "g-ord"),
+        _viz("Viz_3", "Margin Rate Weekly", "LINE", "average [Margin Rate] [Date].weekly",
+             ["Week(Date)", "Average Margin Rate"], "g-mar"),
+        _viz("Viz_4", "Customers Weekly", "LINE", "sum [Customers] [Date].weekly",
+             ["Week(Date)", "Total Customers"], "g-cus"),
+        _viz("Viz_5", "Revenue by Region", "COLUMN", "sum [Revenue] [Region]",
+             ["Region", "Total Revenue"], "g-reg"),
+        _viz("Viz_6", "Revenue vs Target by Channel", "COLUMN",
+             "sum [Revenue] sum [Revenue Target] [Channel]",
+             ["Channel", "Total Revenue", "Total Revenue Target"], "g-chn"),
+        _viz("Viz_7", "Orders by Store Name", "BAR", "sum [Orders] [Store Name]",
+             ["Store Name", "Total Orders"], "g-sto"),
+        _viz("Viz_8", "Revenue Trend", "LINE", "sum [Revenue] [Date].monthly",
+             ["Month(Date)", "Total Revenue"], "g-trd"),
+    ]}}
+
+
+class _Resp:
+    def __init__(self, payload, status_code=200):
+        self._payload, self.status_code = payload, status_code
+        self.text = json.dumps(payload)
+
+    def json(self):
+        return self._payload
+
+
+def _enhance(monkeypatch):
+    """Run the REAL enhance_mcp_liveboard (golden layout) offline; return the TML it imports."""
+    import demoprep_app.liveboards.muze_engine as me
+    import liveboard_creator as lc
+
+    board = _mcp_board()
+    captured = {}
+
+    class _Session:
+        def post(self, url, **kwargs):
+            return _Resp([{"edoc": yaml.safe_dump(board)}])
+
+    class _Client:
+        base_url = "https://example"
+        session = _Session()
+
+    def fake_import(client, liveboard_tml):
+        captured["tml"] = yaml.safe_load(yaml.safe_dump(liveboard_tml))
+        return {"success": True}
+
+    monkeypatch.setattr(lc, "_humanize_viz_titles", lambda **kw: {})
+    monkeypatch.setattr(lc, "_fetch_model_columns", lambda client, guid: [])
+    monkeypatch.setattr(lc, "import_liveboard_tml_with_recovery", fake_import)
+    monkeypatch.setattr(me, "make_cardinality_probe", lambda client, guid: (lambda viz: {}))
+    result = lc.enhance_mcp_liveboard(liveboard_guid="lb-1",
+                                      company_data={"name": "Acme", "use_case": "Retail Sales"},
+                                      ts_client=_Client(), layout_strategy="golden_two_tab")
+    assert result["success"], result
+    return captured["tml"]
+
+
+@pytest.fixture
+def enhanced(monkeypatch):
+    return _enhance(monkeypatch)
+
+
+def _tab(tml):
+    return tml["liveboard"]["layout"]["tabs"][0]
+
+
+def _tile(tab, vid):
+    return next(t for t in tab["tiles"] if t["visualization_id"] == vid)
+
+
+def _kpi_group_id(tml):
+    return next(g["id"] for g in tml["liveboard"]["groups"] if "Metric" in g["name"])
+
+
+def _apply(tml, **kw):
+    args = {"company": "Acme", "use_case": "Retail Sales", "logo": False, "insight": False,
+            "link": PORTAL_URL, "logo_svg": SVG}
+    args.update(kw)
+    return apply_header_tiles(tml, **args)
+
+
+# ------------------------------------------------------------ default OFF
+
+@pytest.mark.parametrize("value", OFF_VALUES)
+def test_omitted_false_or_blank_flags_are_off_and_silent(value):
+    assert resolve_header_tile_flags(value, value) == (False, False, [])
+
+
+@pytest.mark.parametrize("value", [True, "true", "yes", "1", 1, "ON"])
+def test_true_values_turn_a_tile_on(value):
+    assert resolve_header_tile_flags(value, False) == (True, False, [])
+    assert resolve_header_tile_flags(False, value) == (False, True, [])
+
+
+def test_an_unreadable_flag_is_off_and_reported():
+    logo, insight, notes = resolve_header_tile_flags("maybe", None)
+    assert (logo, insight) == (False, False)
+    assert notes and "logo_tile" in notes[0] and "maybe" in notes[0]
+
+
+def test_enhanced_board_is_byte_identical_with_the_tiles_off(monkeypatch):
+    """The real enhanced TML is untouched — same object, same bytes — when off."""
+    base = _enhance(monkeypatch)
+    before = yaml.safe_dump(base, sort_keys=True)
+    out, changes, warnings = _apply(base)
+    assert out is base and changes == [] and warnings == []
+    assert yaml.safe_dump(out, sort_keys=True) == before
+    # And the enhance pass itself is the same run to run (nothing reads the flags).
+    assert yaml.safe_dump(_enhance(monkeypatch), sort_keys=True) == before
+    assert LOGO_TILE_ID not in before and INSIGHT_TILE_ID not in before
+
+
+def test_tool_flags_are_real_parameters_defaulting_off():
+    import mcp_server
+
+    sig = inspect.signature(mcp_server.build_demo_from_brief).parameters
+    for name in ("logo_tile", "insight_tile"):
+        assert name in sig, name
+        assert sig[name].default is False
+        assert sig[name].annotation in ("bool | str", bool | str)
+    doc = inspect.getdoc(mcp_server.build_demo_from_brief)
+    assert "logo_tile: OFF by default" in doc and "insight_tile: OFF by default" in doc
+
+
+def _start(monkeypatch, **kwargs):
+    """build_demo_from_brief with _run_build stubbed; returns (reply, _run_build kwargs)."""
+    import mcp_server
+
+    seen, done = {}, threading.Event()
+
+    def fake_run_build(*args, **kw):
+        seen["args"] = args
+        seen.update(kw)
+        done.set()
+        return {}
+
+    monkeypatch.setattr(mcp_server, "_run_build", fake_run_build)
+    monkeypatch.setattr(mcp_server, "_resolve_ts_target", lambda url: ("https://ts.example", "k"))
+    monkeypatch.setattr(mcp_server, "_resolve_owner", lambda e: "se@thoughtspot.com")
+    monkeypatch.setattr(mcp_server, "_login_precheck",
+                        lambda url, key, owner: {"outcome": "ok", "blocking": False})
+    out = mcp_server.build_demo_from_brief(
+        brief="A brief.", company_name="Acme", owner_email="se@thoughtspot.com",
+        salesforce_opportunity_id="0065f00000AbCdEfGHI", environment="prod", **kwargs)
+    assert out["status"] == "started", out
+    assert done.wait(5), "worker never called _run_build"
+    return out, seen
+
+
+@pytest.mark.parametrize("kwargs", [
+    {}, {"logo_tile": False, "insight_tile": False}, {"logo_tile": "", "insight_tile": ""},
+    {"logo_tile": "false", "insight_tile": "no"}, {"logo_tile": None},
+])
+def test_a_build_without_the_flags_does_no_header_tile_work(monkeypatch, kwargs):
+    out, seen = _start(monkeypatch, **kwargs)
+    assert "warnings" not in out, "nothing to report on a default build"
+    assert seen["header_tiles"] is None
+
+
+def test_the_flags_reach_the_build_and_the_payload(monkeypatch):
+    out, seen = _start(monkeypatch, logo_tile="true", insight_tile=True)
+    assert seen["header_tiles"] == {"logo": True, "insight": True, "warnings": []}
+    payload = seen["args"][-1]
+    assert payload["logo_tile"] == "true" and payload["insight_tile"] is True
+
+
+def test_an_unreadable_flag_is_reported_on_the_started_reply(monkeypatch):
+    out, seen = _start(monkeypatch, insight_tile="perhaps")
+    assert out["warnings"] and "insight_tile" in out["warnings"][0]
+    assert seen["header_tiles"]["insight"] is False
+    assert seen["header_tiles"]["warnings"] == out["warnings"]
+
+
+# ------------------------------------------------------------- placement
+
+def test_logo_only_sits_left_of_the_kpis(enhanced):
+    gid = _kpi_group_id(enhanced)
+    before = _tab(enhanced)
+    kpi_before = dict(_tile(before, gid))
+    others_before = {t["visualization_id"]: dict(t) for t in before["tiles"]}
+    out, changes, warnings = _apply(enhanced, logo=True)
+    assert warnings == []
+    tab = _tab(out)
+    assert _tile(tab, LOGO_TILE_ID) == {"visualization_id": LOGO_TILE_ID, "x": 0, "y": kpi_before["y"],
+                                         "height": kpi_before["height"], "width": 3}
+    kpi = _tile(tab, gid)
+    assert (kpi["x"], kpi["width"], kpi["height"]) == (3, 9, kpi_before["height"])
+    # Nothing else moved, and the KPIs keep their group layout.
+    for vid, t in others_before.items():
+        if vid != gid:
+            assert _tile(tab, vid) == t
+    assert tab["group_layouts"] == before["group_layouts"]
+
+
+def test_insight_only_sits_left_of_the_kpis_with_kpis_two_per_row(enhanced):
+    gid = _kpi_group_id(enhanced)
+    before = _tab(enhanced)
+    k = dict(_tile(before, gid))
+    out, _, warnings = _apply(enhanced, insight=True)
+    assert warnings == []
+    tab = _tab(out)
+    ins, kpi = _tile(tab, INSIGHT_TILE_ID), _tile(tab, gid)
+    assert (ins["x"], ins["width"], ins["y"]) == (0, 6, k["y"])
+    assert (kpi["x"], kpi["width"]) == (6, 6)
+    assert ins["height"] == kpi["height"] == 2 * k["height"]
+    layout = next(gl for gl in tab["group_layouts"] if gl["id"] == gid)
+    assert sorted((t["x"], t["y"], t["width"]) for t in layout["tiles"]) == [
+        (0, 0, 6), (0, k["height"], 6), (6, 0, 6), (6, k["height"], 6)]
+    # Everything below the KPI row moved down by exactly the extra height.
+    for t in before["tiles"]:
+        if t["visualization_id"] != gid:
+            assert _tile(tab, t["visualization_id"])["y"] == t["y"] + k["height"]
+
+
+def test_both_tiles_order_is_logo_then_insight_then_kpis(enhanced):
+    gid = _kpi_group_id(enhanced)
+    out, changes, warnings = _apply(enhanced, logo=True, insight=True)
+    assert warnings == []
+    tab = _tab(out)
+    row = sorted((t for t in tab["tiles"] if t["y"] == 0), key=lambda t: t["x"])
+    assert [(t["visualization_id"], t["x"], t["width"]) for t in row] == [
+        (LOGO_TILE_ID, 0, 2), (INSIGHT_TILE_ID, 2, 4), (gid, 6, 6)]
+    assert len({t["height"] for t in row}) == 1
+    assert any("left to right" in c for c in changes)
+    # Every tile on the tab sits inside the 12-column grid and none overlap.
+    cells = set()
+    for t in tab["tiles"]:
+        assert 0 <= t["x"] and t["x"] + t["width"] <= 12
+        for x in range(t["x"], t["x"] + t["width"]):
+            for y in range(t["y"], t["y"] + t["height"]):
+                assert (x, y) not in cells, f"overlap at {(x, y)}"
+                cells.add((x, y))
+
+
+def test_no_kpi_row_puts_the_tiles_in_a_new_top_row():
+    lb = {"liveboard": {"name": "B", "visualizations": [
+        _viz("Viz_1", "Revenue by Region", "COLUMN", "", ["Region", "Total Revenue"], "g1")],
+        "layout": {"tabs": [{"name": "Overview", "tiles": [
+            {"visualization_id": "Viz_1", "x": 0, "y": 0, "height": 8, "width": 12}]}]}}}
+    out, changes, warnings = _apply(lb, logo=True, insight=True)
+    tiles = {t["visualization_id"]: t for t in _tab(out)["tiles"]}
+    assert (tiles[LOGO_TILE_ID]["x"], tiles[LOGO_TILE_ID]["width"]) == (0, 3)
+    assert (tiles[INSIGHT_TILE_ID]["x"], tiles[INSIGHT_TILE_ID]["width"]) == (3, 9)
+    assert tiles["Viz_1"]["y"] == tiles[LOGO_TILE_ID]["height"]
+    assert "new top row" in changes[-1]
+
+
+# ---------------------------------------------------------- tile content
+
+def test_logo_tile_content_comes_from_the_board_and_links_to_the_portal(enhanced):
+    out, changes, _ = _apply(enhanced, logo=True)
+    viz = next(v for v in out["liveboard"]["visualizations"] if v["id"] == LOGO_TILE_ID)
+    body = viz["note_tile"]["html_parsed_string"]
+    assert f'<a href="{PORTAL_URL}"' in body and 'data-image-link="true"' in body
+    src = re.search(r'<img src="data:image/svg\+xml;base64,([^"]+)"', body).group(1)
+    assert base64.b64decode(src).decode() == SVG
+    heading = _tab(out)["name"]
+    assert f">{heading}</strong>" in body
+    assert "Retail sales at a glance." in body
+    # The board's real KPI measures, in bold, without the aggregation prefix.
+    bolds = re.findall(r'text-bold" style="white-space: pre-wrap;">([^<]+)</strong>', body)
+    assert bolds[0] == heading
+    kpi_phrases = bolds[1:]
+    assert kpi_phrases and set(kpi_phrases) <= {"revenue", "orders", "margin rate", "customers"}
+    assert "Drill into" in body and "store" in body.lower()
+    assert "Okta" not in body and "Identity" not in body
+    assert any(PORTAL_URL in c for c in changes)
+
+
+def test_logo_text_is_escaped():
+    body = HT.logo_tile_html(logo_svg=SVG, link='https://x.example/?a=1&b="2"', company="A<B>",
+                             heading="Exec & Co", tagline="t", measures=["Total <Sales>"],
+                             dimensions=[])
+    assert "<B>" not in body and "&lt;sales&gt;" in body and "Exec &amp; Co" in body
+    assert 'href="https://x.example/?a=1&amp;b=&quot;2&quot;"' in body
+
+
+def _insight_code(tml):
+    viz = next(v for v in tml["liveboard"]["visualizations"] if v["id"] == INSIGHT_TILE_ID)
+    ai = viz["ai_tile"]
+    assert ai["target_agent"] == "insight_code_runner"
+    assert ai["template_id"] == "text_long_form"
+    return decode_analysis_code(ai["analysis_code"])
+
+
+def test_insight_tile_reads_the_boards_own_kpis_and_drivers(enhanced):
+    out, changes, warnings = _apply(enhanced, insight=True)
+    assert warnings == []
+    code = _insight_code(out)
+    files = re.findall(r'"([^"]+)\.csv"', code)
+    # The first tab's four KPIs in row order, then its driver charts in layout
+    # order — each by the viz_guid of the viz on the board. Region went to the
+    # manager tab, so it is not read.
+    assert files == ["g-rev", "g-ord", "g-mar", "g-cus", "g-sto", "g-chn"]
+    assert "Acme" in code and "Retail Sales" in code
+    compile(code, "analysis_code", "exec")
+
+
+def test_the_insight_code_is_deterministic(enhanced):
+    a = _apply(enhanced, insight=True)[0]
+    b = _apply(enhanced, insight=True)[0]
+    assert yaml.safe_dump(a) == yaml.safe_dump(b)
+
+
+def _write_csvs(tmp_path, code, preamble):
+    """Fake CSVs for every file the code declares, shaped like ThoughtSpot's."""
+    lead = ('"Data extract produced by x"\n"Filters applied on data :"\n""\n' if preamble else "")
+    sources = json.loads(re.search(r"SOURCES = (\[.*?\n\])", code, re.S).group(1))
+    files = dict(re.findall(r'"([a-z0-9_]+)": "([^"]+\.csv)"', code))
+    for s in sources:
+        path = tmp_path / files[s["key"]]
+        if s["kind"] == "trend":
+            months = ["Jan 2026", "Feb 2026", "Mar 2026", "Apr 2026", "May 2026", "Jun 2026",
+                      "Jul 2026", "Aug 2026"]
+            rows = "\n".join(f'"{m}","{100 + i * 7}"' for i, m in enumerate(reversed(months)))
+            path.write_text(lead + f'"Monthly Date","{s["measures"][0]}"\n' + rows + "\n")
+        else:
+            cols = s["attributes"] + s["measures"]
+            body = "\n".join(",".join(f'"{n}{i}"' if c in s["attributes"] else f'"{(i + 1) * (j + 3) * 10}"'
+                                      for j, c in enumerate(cols)) for i, n in enumerate("ABCD"))
+            path.write_text(lead + ",".join(f'"{c}"' for c in cols) + "\n" + body + "\n")
+
+
+@pytest.mark.parametrize("preamble", [False, True])
+def test_the_insight_code_runs_on_the_boards_csvs(enhanced, tmp_path, preamble):
+    """The program ThoughtSpot will run: valid, and it prints an analysis
+    from CSVs shaped like ThoughtSpot's (with or without a report preamble)."""
+    code = _insight_code(_apply(enhanced, insight=True)[0])
+    _write_csvs(tmp_path, code, preamble)
+    (tmp_path / "analysis.py").write_text(code)
+    p = subprocess.run([sys.executable, "-I", "analysis.py"], cwd=tmp_path,
+                       capture_output=True, text=True, timeout=60)
+    assert p.returncode == 0, p.stderr
+    assert "=== User Query ===" in p.stdout and "=== Key Signals ===" in p.stdout
+    assert "[SKIPPED" not in p.stdout, p.stdout
+    assert "=== Trend:" in p.stdout and "=== Breakdown:" in p.stdout
+    assert "Attainment (Total Revenue / Total Revenue Target)" in p.stdout
+    assert re.search(r"^1\. ", p.stdout, re.M)
+
+
+# --------------------------------------------------------- failures warn
+
+def test_no_portal_and_no_website_leaves_the_logo_out_with_a_warning():
+    link, notes = logo_link(None, "", with_portal=False)
+    assert link == "" and "left out" in notes[0] and "with_portal=False" in notes[0]
+
+
+def test_no_portal_links_the_company_website_with_a_warning():
+    link, notes = logo_link(None, "acme.com", with_portal=False)
+    assert link == "https://acme.com" and "company website" in notes[0]
+    link, notes = logo_link({"_warnings": ["portal build failed"]}, "https://acme.com", True)
+    assert link == "https://acme.com" and "portal build did not produce a site" in notes[0]
+    assert logo_link({"url": PORTAL_URL}, "acme.com", True) == (PORTAL_URL, [])
+
+
+def test_logo_is_the_portals_own_file_else_its_wordmark(tmp_path):
+    from demoprep_app.portal.builder import DEFAULT_PRIMARY, wordmark_svg
+
+    (tmp_path / "brand-logo.svg").write_text(SVG)
+    assert portal_logo_svg("Acme", {"work_dir": str(tmp_path)}) == (SVG, "the TSE portal's logo")
+    svg, source = portal_logo_svg("Acme", None)
+    assert svg == wordmark_svg("Acme", DEFAULT_PRIMARY) and "wordmark" in source
+
+
+def test_the_portal_wordmark_is_unchanged(tmp_path):
+    """wordmark() now delegates to wordmark_svg(); the portal's file is byte-identical."""
+    from demoprep_app.portal.builder import wordmark
+
+    wordmark("Acme Corp", "#123a5c", tmp_path / "w.svg")
+    label = "ACME CORP"
+    width = 40 + len(label) * 11
+    assert (tmp_path / "w.svg").read_text() == (
+        f'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {width} 40" '
+        f'width="{width}" height="40"><text x="4" y="27" font-family="Inter, '
+        f'system-ui, sans-serif" font-size="18" font-weight="700" letter-spacing="0.5" '
+        f'style="fill:#123a5c">{label}</text></svg>\n')
+
+
+def test_a_tile_that_cannot_be_built_is_left_out_with_a_warning(enhanced):
+    out, changes, warnings = _apply(enhanced, logo=True, link="")
+    assert out is enhanced and changes == []
+    assert warnings and warnings[0].startswith("logo tile left out")
+    no_guids = yaml.safe_load(yaml.safe_dump(enhanced))
+    for v in no_guids["liveboard"]["visualizations"]:
+        v.pop("viz_guid", None)
+    out, changes, warnings = _apply(no_guids, logo=True, insight=True)
+    assert any(w.startswith("insight tile left out") for w in warnings)
+    assert LOGO_TILE_ID in yaml.safe_dump(out) and INSIGHT_TILE_ID not in yaml.safe_dump(out)
+
+
+class _FakeTS:
+    """Records export/import calls; import answers come from a list."""
+
+    def __init__(self, tml, import_statuses=(), export_status=200):
+        self.base_url = "https://ts.example"
+        self.tml, self.imports, self.statuses = tml, [], list(import_statuses)
+        self.export_status = export_status
+        outer = self
+
+        class _S:
+            def post(self, url, json=None, **kw):
+                if url.endswith("/tml/export"):
+                    return _Resp([{"edoc": yaml.safe_dump(outer.tml)}], outer.export_status)
+                outer.imports.append(yaml.safe_load(json["metadata_tmls"][0]))
+                status = outer.statuses.pop(0) if outer.statuses else {"status_code": "OK"}
+                return _Resp([{"response": {"status": status}}])
+
+        self.session = _S()
+
+
+def _run_pass(ts, **kw):
+    args = {"company": "Acme", "use_case": "Retail Sales", "company_url": "acme.com",
+            "logo": True, "insight": True, "with_portal": True, "portal": {"url": PORTAL_URL}}
+    args.update(kw)
+    return add_header_tiles_to_board(ts, "lb-1", **args)
+
+
+def test_the_pass_exports_adds_both_tiles_and_imports_once(enhanced):
+    ts = _FakeTS(enhanced)
+    out = _run_pass(ts)
+    assert out["warnings"] == []
+    assert len(ts.imports) == 1
+    ids = {v["id"] for v in ts.imports[0]["liveboard"]["visualizations"]}
+    assert {LOGO_TILE_ID, INSIGHT_TILE_ID} <= ids
+    assert any("logo tile added" in c and PORTAL_URL in c for c in out["changes"])
+
+
+def test_the_pass_does_nothing_at_all_when_off(enhanced):
+    ts = _FakeTS(enhanced)
+    assert _run_pass(ts, logo=False, insight=False) == {"changes": [], "warnings": []}
+    assert ts.imports == []
+
+
+def test_a_rejected_tile_is_dropped_and_the_other_kept(enhanced):
+    rejected = {"status_code": "ERROR",
+                "error_message": f"Visualization <b> {INSIGHT_TILE_ID} </b> has following errors:<br/>bad"}
+    ts = _FakeTS(enhanced, [rejected])
+    out = _run_pass(ts)
+    assert len(ts.imports) == 2
+    final_ids = {v["id"] for v in ts.imports[1]["liveboard"]["visualizations"]}
+    assert LOGO_TILE_ID in final_ids and INSIGHT_TILE_ID not in final_ids
+    assert any(w.startswith("insight tile left out") for w in out["warnings"])
+
+
+def test_a_rejection_of_anything_else_changes_nothing(enhanced):
+    rejected = {"status_code": "ERROR",
+                "error_message": "Visualization <b> Viz_5 </b> has following errors:<br/>x"}
+    ts = _FakeTS(enhanced, [rejected])
+    out = _run_pass(ts)
+    assert len(ts.imports) == 1 and out["changes"] == []
+    assert out["warnings"][0].startswith("header tiles not added: ThoughtSpot rejected")
+
+
+def test_an_export_failure_warns_and_imports_nothing(enhanced):
+    ts = _FakeTS(enhanced, export_status=500)
+    out = _run_pass(ts)
+    assert ts.imports == [] and "export failed" in out["warnings"][0]
+
+
+def test_the_pass_never_raises():
+    class Broken:
+        base_url = "https://ts.example"
+
+        @property
+        def session(self):
+            raise ConnectionError("cluster unreachable")
+
+    out = _run_pass(Broken())
+    assert out["changes"] == [] and "cluster unreachable" in out["warnings"][0]
+
+
+# ---------------------------------------------- the build runs the pass
+
+def _offline_build(monkeypatch, header_tiles, tiles_result=None, tiles_raises=None):
+    """_run_build on an existing schema with a fake controller; with_portal=False."""
+    import collections
+
+    import mcp_server
+    import thoughtspot_deployer
+    from demoprep_app.liveboards import header_tiles as ht
+
+    calls = {"deployer": 0, "tiles": []}
+
+    class FakeController:
+        def __init__(self, **kw):
+            self.settings, self.deployment_completion = {}, None
+
+        def _create_run_loggers(self, force=False):
+            pass
+
+        def _run_thoughtspot_deployment(self, schema, company, use_case, on_progress=None):
+            self.deployment_completion = {"success": True, "liveboard_guid": "lb-1",
+                                          "model_guid": "m-1", "warnings": ["from the build"]}
+            return iter(())
+
+    class FakeDeployer:
+        def __init__(self, **kw):
+            calls["deployer"] += 1
+
+        def authenticate_trusted(self):
+            return True
+
+    def fake_tiles(ts, guid, **kw):
+        calls["tiles"].append(kw)
+        if tiles_raises:
+            raise tiles_raises
+        return tiles_result or {"changes": [], "warnings": []}
+
+    monkeypatch.setattr(mcp_server, "ChatDemoInterface", FakeController)
+    monkeypatch.setattr(mcp_server, "DemoBuilder", lambda **kw: type("DB", (), {})())
+    monkeypatch.setattr(mcp_server, "load_gradio_settings", lambda owner: {})
+    monkeypatch.setattr(mcp_server, "log_mcp_payload", lambda p: True)
+    monkeypatch.setattr(mcp_server, "log_mcp_result", lambda p: True)
+    monkeypatch.setattr(mcp_server, "_recent_builds", collections.OrderedDict())
+    monkeypatch.setattr(thoughtspot_deployer, "ThoughtSpotDeployer", FakeDeployer)
+    monkeypatch.setattr(ht, "add_header_tiles_to_board", fake_tiles)
+    out = mcp_server._run_build("r1", "brief", "Acme", "Retail Sales", "acme.com",
+                                "https://ts.example", "k", "se@thoughtspot.com", False,
+                                {"brief": "brief"}, existing_schema="S",
+                                existing_ddl="CREATE TABLE T (A INT)", header_tiles=header_tiles)
+    return out, calls
+
+
+def test_a_default_build_never_touches_the_board_again(monkeypatch):
+    out, calls = _offline_build(monkeypatch, None)
+    assert out["status"] == "success", out
+    assert calls == {"deployer": 0, "tiles": []}
+    assert out["warnings"] == ["from the build"]
+
+
+def test_a_build_with_the_flags_runs_the_pass_after_the_portal(monkeypatch):
+    out, calls = _offline_build(
+        monkeypatch, {"logo": True, "insight": False, "warnings": []},
+        tiles_result={"changes": ["logo tile added"], "warnings": ["logo tile links to the company website"]})
+    assert out["status"] == "success"
+    assert calls["deployer"] == 1 and len(calls["tiles"]) == 1
+    kw = calls["tiles"][0]
+    assert kw["logo"] is True and kw["insight"] is False
+    assert kw["with_portal"] is False and kw["portal"] is None and kw["company_url"] == "acme.com"
+    assert out["warnings"] == ["from the build", "logo tile links to the company website"]
+
+
+def test_a_failing_pass_never_fails_the_build(monkeypatch):
+    out, calls = _offline_build(monkeypatch, {"logo": False, "insight": True, "warnings": []},
+                                tiles_raises=RuntimeError("boom"))
+    assert out["status"] == "success"
+    assert out["warnings"][-1] == "header tiles not added: RuntimeError: boom"

tests/test_payload_capture_is_complete.py

@@ -36,7 +36,8 @@ def test_capture_includes_every_parameter_of_the_tool():
                ts_url="t", owner_email="o", with_portal=True,
                theme_mode="light", auth="none", nav_layout="top", inline_tab=False,
                monetize=True, tiers=True, pinning=False, salesforce_opportunity_id="",
-               environment=""):
+               environment="",
+               logo_tile=False, insight_tile=False):  # header tiles, 2026-10-20
         return M._call_payload(inspect.currentframe())
 
     got = set(sample())

tests/test_payload_reaches_the_database.py

@@ -123,7 +123,8 @@ def test_tool_signature_survives_the_entire_path():
                ts_url="t", owner_email="o", with_portal=True,
                theme_mode="light", auth="none", nav_layout="top", inline_tab=False,
                monetize=True, tiers=True, pinning=False, salesforce_opportunity_id="",
-               environment=""):
+               environment="",
+               logo_tile=False, insight_tile=False):  # header tiles, 2026-10-20
         return M._call_payload(inspect.currentframe())
 
     results = captured_row(sample())