"""
mcp_server.py — Demo Wire MCP server (v1)

Exposes DemoPrep's demo-build pipeline as an MCP tool so an external agent
(AgentSpot / Spotter Assistant) can build a full ThoughtSpot demo from a prospect
*brief* without a human clicking through the Gradio UI.

This is a THIN ADAPTER. It does not reimplement any business logic — it drives the
exact same `ChatDemoInterface` controller the Gradio app uses, headlessly, skipping
the research phase by injecting the brief as the research output. All build logic
stays in one place (see docs/SINGLE_PIPELINE.md and CLAUDE.md).

v1 scope (simplest thing that works):
  * ONE blocking tool — build_demo_from_brief(...) runs the whole pipeline
    (blueprint -> dataset -> DDL -> Snowflake -> TS model -> liveboard) and returns
    the resulting IDs when done.
  * Everything is fixed/server-side; per-call config (env, model, sharing) is a
    later version.
  * Builds run for minutes. If the caller's connection times out, the build still
    finishes server-side and DemoPrep's Slack notification delivers the liveboard
    URL. (Async / job-backed delivery is the planned next step.)

Server-side config (env vars):
  MCP_ACCESS_TOKEN    (required in http mode)  shared bearer secret gating the endpoint
  TS_ENV_URL_DEFAULT  ThoughtSpot URL used when a build doesn't pass ts_url
  TS_USER_DEFAULT     owner email used when a build doesn't pass owner_email
  TS_ENV_<n>_LABEL / TS_ENV_<n>_URL / TS_ENV_<n>_KEY_VAR
                      the environment table — SAME numbered triplets the app's
                      dropdown uses. The trusted-auth key for a build is found
                      by matching its ts_url host against TS_ENV_<n>_URL.
  MCP_TRANSPORT       "stdio" (default, local dev) | "http" (HF Space)
  MCP_HTTP_PORT       http port (default 7860)
Also read from env: SUPABASE_URL / SUPABASE_ANON_KEY (Snowflake creds) + ANTHROPIC_API_KEY.
Both the TS target and the owner are resolved per build: the request may pass
ts_url / owner_email, else they fall back to TS_ENV_URL_DEFAULT / TS_USER_DEFAULT.

Local smoke test (needs the project venv with requirements installed):
    TS_ENV_URL_DEFAULT=https://sebe.thoughtspotstaging.cloud python mcp_server.py
(the TS_ENV_* table comes from .env; the URL default just picks which entry)
Then point an MCP client (e.g. `mcp dev mcp_server.py` / MCP Inspector) at stdio.
"""

from __future__ import annotations

import hashlib
import os
import sys
import time
import types
import uuid
import re
import threading
import collections

# --- env bootstrap: order matters --------------------------------------------
# .env first so MCP_* / TS_ENV_* / Supabase creds are available.
from dotenv import load_dotenv
load_dotenv()

# Owner identity is resolved PER BUILD (mirrors the TS target): the request may
# pass owner_email; otherwise it falls back to TS_USER_DEFAULT. Nothing is
# hardcoded, and boot does not require it — a build fails loud if neither is set.
os.environ.setdefault("DEMOPREP_NO_AUTH", "true")
_DEFAULT_OWNER = (os.getenv("TS_USER_DEFAULT") or "").strip()
if _DEFAULT_OWNER:
    # Best-effort acting user for any headless path with no per-build owner.
    os.environ.setdefault("DEMOPREP_DEV_USER_EMAIL", _DEFAULT_OWNER)


def _resolve_owner(owner_email: str) -> str:
    """Owner of the objects a build creates: the per-request owner_email, else
    TS_USER_DEFAULT. Fails loud if neither is set (no silent blank owner)."""
    owner = (owner_email or os.getenv("TS_USER_DEFAULT") or "").strip()
    if not owner:
        raise RuntimeError("No owner: pass owner_email in the request or set TS_USER_DEFAULT.")
    return owner

# Snowflake creds + SNOWFLAKE_DATABASE from Supabase admin settings -> os.environ.
# Must run once at startup; the deploy path reads these via get_admin_setting/env.
from supabase_client import (
    inject_admin_settings_to_env,
    load_gradio_settings,
    log_mcp_payload,
    log_mcp_result,
)
inject_admin_settings_to_env()

# Heavy imports (pull gradio etc.) — safe headless. DEMOPREP_NO_AUTH is set above,
# which must happen BEFORE importing chat_interface.
from chat_interface import ChatDemoInterface
from demo_builder_class import DemoBuilder
from demo_personas import parse_use_case, get_use_case_config
from llm_config import DEFAULT_LLM_MODEL

def _ts_env_table() -> list[tuple[str, str, str]]:
    """Read the numbered TS_ENV_<n>_LABEL/URL/KEY_VAR triplets — the SAME
    environment table the app's dropdown uses (single source of truth for
    which ThoughtSpot instances exist and their trusted-auth keys).

    KEY_VAR normally holds the trusted-auth key itself. An `ENV:<NAME>` value
    dereferences another env var instead — for referencing a secret that
    already exists under its own name (e.g. TECHPARTNERS_TA_KEY on the Space)
    without copying its value into the table."""
    table = []
    # URL is the sentinel, NOT label: MCP resolves a key by matching the build's
    # ts_url host, so URL+KEY_VAR is the whole functional contract and the label
    # is cosmetic (it exists for the app's dropdown). Keying the loop on label
    # meant setting URL+KEY_VAR without a label silently dropped the entry —
    # a whole ThoughtSpot instance vanishing over a decorative field. Scan a
    # fixed range so a gap in the numbering doesn't truncate the rest either.
    for i in range(1, 21):
        url = (os.getenv(f"TS_ENV_{i}_URL") or "").strip().rstrip("/")
        key = (os.getenv(f"TS_ENV_{i}_KEY_VAR") or "").strip()
        if key.startswith("ENV:"):
            key = (os.getenv(key[4:].strip()) or "").strip()
        if not url and not key:
            continue
        # Label is optional and purely for messages; fall back to the host.
        label = (os.getenv(f"TS_ENV_{i}_LABEL") or "").strip() or (_host_of(url) if url else f"TS_ENV_{i}")
        table.append((label, url, key))
    return table


def _host_of(url: str) -> str:
    """Hostname of a URL; tolerates scheme-less input and strips any port."""
    if "://" in url:
        url = url.split("://", 1)[1]
    return url.split("/", 1)[0].split(":", 1)[0].lower()


def _resolve_ts_target(ts_url: str) -> tuple[str, str]:
    """Resolve the (url, trusted_auth_key) for a build.

    The URL comes from the build request; if absent it falls back to
    TS_ENV_URL_DEFAULT. The trusted-auth key comes from the TS_ENV_<n>_* table
    entry whose URL host matches. Fails loud on a missing URL, an unknown
    instance, or a blank key (no silent blanks). Resolved per build (not at
    boot) so the server stays up regardless of which environment a given
    build targets.
    """
    url = (ts_url or os.getenv("TS_ENV_URL_DEFAULT") or "").strip().rstrip("/")
    if not url:
        raise RuntimeError(
            "No ThoughtSpot URL: pass ts_url in the request or set TS_ENV_URL_DEFAULT."
        )
    if "://" not in url:
        url = f"https://{url}"
    # Single-instance shortcut: one key, no table. MCP targets one ThoughtSpot
    # instance in practice, and the numbered table only exists to answer "which
    # key for this host" — which is trivial when there is one. TS_AUTH_KEY skips
    # the ceremony entirely.
    direct_key = (os.getenv("TS_AUTH_KEY") or "").strip()
    table = _ts_env_table()
    if not table and not direct_key:
        raise RuntimeError(
            "No ThoughtSpot trusted-auth key configured. Set TS_AUTH_KEY (single "
            "instance — simplest), or the numbered TS_ENV_<n>_LABEL/URL/KEY_VAR "
            "table the app's dropdown uses."
        )

    host = _host_of(url)
    matches = [(label, env_url, key) for label, env_url, key in table
               if _host_of(env_url) == host]

    # Two entries on the SAME host with DIFFERENT keys is unresolvable from the
    # URL alone — exactly the "sebe - se" vs "sebe - demo" case. First-match-wins
    # silently made the second entry unreachable, so a build could never target
    # it and nobody could tell. Warn loudly rather than raise: prod is configured
    # this way today and taking builds down over stale config would be worse.
    if len(matches) > 1 and len({k for _, _, k in matches if k}) > 1:
        losing = ", ".join(repr(lbl) for lbl, _, _ in matches[1:])
        print(
            f"[MCP] WARNING: {len(matches)} TS_ENV entries share host {host!r} with "
            f"different keys. Using {matches[0][0]!r}; {losing} is unreachable and "
            f"its key is dead config — delete it, or give it a distinct URL.",
            file=sys.stderr,
        )

    if matches:
        label, _env_url, key = matches[0]
        if not key:
            raise RuntimeError(f"TS_ENV entry {label!r} matches {url} but its KEY_VAR is blank.")
        return url, key

    if direct_key:
        return url, direct_key

    known = ", ".join(f"{label} ({env_url})" for label, env_url, _ in table)
    raise RuntimeError(
        f"No TS_ENV entry matches {url}. Known environments: {known}. "
        f"Either set TS_AUTH_KEY for a single-instance setup, or add a "
        f"TS_ENV_<n>_LABEL/URL/KEY_VAR triplet for this instance."
    )

from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings

# Bounded concurrency: the build pipeline is already concurrency-safe (the app/QA
# have run concurrent builds for months), so this is only a resource guardrail — a
# semaphore caps how many run at once so a caller can't OOM the Space by firing
# dozens; a call past the cap gets "busy". Owner/env/creds are per-instance, so
# concurrent builds share no mutable process state.
_MAX_CONCURRENT_BUILDS = max(1, int(os.getenv("MCP_MAX_CONCURRENT_BUILDS", "3")))
_build_sem = threading.BoundedSemaphore(_MAX_CONCURRENT_BUILDS)

# Lightweight hit log so we can watch incoming calls (who + what args). Writes to
# stderr (the server log) and, if MCP_HIT_LOG is set, appends to that file too.
_HIT_LOG = (os.getenv("MCP_HIT_LOG") or "").strip()


def _log_hit(msg: str) -> None:
    line = f"[{time.strftime('%H:%M:%S')}] {msg}"
    print(line, file=sys.stderr, flush=True)
    if _HIT_LOG:
        try:
            with open(_HIT_LOG, "a") as fh:
                fh.write(line + "\n")
        except Exception:
            pass


# --- build status tracking (for the status tool / polling long builds) ---
_SERVER_STARTED = time.time()
_state_lock = threading.Lock()
_build_count = 0
_active_builds = {}     # run_id -> live build dict; supports multiple concurrent builds
_last_build = None      # summary of the most recent completed build
# Compact result of each completed build, keyed by run_id (newest last), capped so
# a client that fired several concurrent builds can collect EACH result by run_id
# instead of racing the single _last_build slot. In-memory; resets on restart.
_recent_builds = collections.OrderedDict()
_RECENT_BUILDS_CAP = 50


def _set_progress(run_id: str, phase: str | None = None, detail: str | None = None) -> None:
    """Update a running build's coarse phase and/or latest raw progress line."""
    with _state_lock:
        b = _active_builds.get(run_id)
        if b is not None:
            if phase:
                b["phase"] = phase
            if detail is not None:
                b["detail"] = detail


def _progress_text(item) -> str:
    """Best-effort human-readable line from a generator yield (str | dict | tuple)."""
    if isinstance(item, dict):
        for k in ("message", "text", "content", "status", "detail"):
            v = item.get(k)
            if isinstance(v, str) and v.strip():
                item = v
                break
        else:
            item = str(item)
    elif isinstance(item, (tuple, list)):
        item = next((x for x in item if isinstance(x, str) and x.strip()), str(item))
    return " ".join(str(item).split())[:200]  # collapse whitespace + cap length


def _phase_from_text(text: str) -> str | None:
    """Map a raw progress line to a granular phase label (None = keep current phase)."""
    t = text.lower()
    # [async] import lines say "…polling…" as normal operation — check them
    # before the retry bucket so a healthy import isn't mislabeled "retrying".
    if "[async]" in t:
        return "importing tables"
    if any(k in t for k in ("504", "retry", "retrying", "throttl", "timed out", "timeout")):
        return "waiting on ThoughtSpot (retrying)"
    if "connection" in t:
        return "creating connection"
    if "tml" in t or "enhance" in t or "post-process" in t or "applying style" in t:
        return "liveboard TML phase"
    if "liveboard" in t or "pinboard" in t:
        return "creating liveboard"
    if "model" in t:
        return "model created" if any(k in t for k in ("created", "ready", "guid")) else "creating model"
    if "verif" in t and "schema" in t:
        return "verifying schema"
    if "table" in t and ("creat" in t or "import" in t):
        return "creating tables"
    if any(k in t for k in ("import", "loading", "populat", "copy into", "batch", "chunk", " rows")):
        return "loading data"
    if "schema" in t and "creat" in t:
        return "creating schema"
    return None


# The app yields a periodic UI "spinner" banner while deploy_all runs in its
# worker thread. It is a heartbeat, not progress — and it happens to contain the
# word "connection" (…Creating connection & tables…), so if it reaches
# _set_progress it both freezes `phase` on "creating connection" and clobbers the
# real per-line detail arriving via on_progress. Drop it on the MCP side.
_BANNER_MARKERS = (
    "deployment in progress",
    "thoughtspot deploying",
    "starting thoughtspot deployment",
)


def _is_heartbeat_banner(text: str) -> bool:
    t = text.lower()
    return any(m in t for m in _BANNER_MARKERS)


def _apply_progress(run_id: str, item) -> None:
    """Fold one generator yield / deploy line into the build's live status.

    Ignores the UI heartbeat banner so the real per-line progress (delivered via
    on_progress from the deploy thread) drives `phase`/`detail` instead of being
    overwritten every 2s.
    """
    text = _progress_text(item)
    if not text or _is_heartbeat_banner(text):
        return
    _set_progress(run_id, phase=_phase_from_text(text), detail=text)


# DNS-rebinding protection validates the Host header and only trusts localhost
# by default, which returns 421 when reached through a tunnel or an hf.space host.
# We gate access on the bearer token instead (and this is server-to-server, not
# browser-driven), so disable the host check to allow any public host.
mcp = FastMCP(
    "demoprep",
    transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
)


@mcp.tool()
def ping() -> dict:
    """Health / connectivity check — returns server identity + config, no side effects.

    Use this to confirm a client (e.g. AgentSpot) can reach and invoke the server
    without triggering a full build.
    """
    _log_hit("CALL ping")
    return {
        "ok": True,
        "server": "demoprep-mcp v1",
        "default_ts_url": os.getenv("TS_ENV_URL_DEFAULT", ""),
        "default_owner": os.getenv("TS_USER_DEFAULT", ""),
        "tools": ["ping", "status", "config_check", "build_demo_from_brief"],
    }


def _fingerprint(value: str) -> str:
    """Identify a secret without disclosing it: length + a short digest.

    Enough to answer "is the Space holding the SAME key as my .env?" by
    comparing two fingerprints, and useless to anyone who obtains it. Never
    return a prefix or suffix of the real value — trusted-auth keys are GUIDs,
    so even 8 leading characters is a meaningful disclosure.
    """
    if not value:
        return "(unset)"
    digest = hashlib.sha256(value.encode()).hexdigest()[:8]
    return f"len={len(value)} sha256:{digest}"


@mcp.tool()
def config_check(ts_url: str = "") -> dict:
    """Report what config this server ACTUALLY has, and whether it hangs together.

    Exists because HF Space secrets are write-only: you cannot read back what is
    set, so a misconfigured Space is invisible until a build fails 40 seconds in
    with a message about one variable. This answers, in one call, "what does the
    server see and what is inconsistent about it".

    Secret VALUES are never returned — only presence and a fingerprint
    (length + sha256 prefix). Compare a fingerprint against the same one computed
    over your local .env to confirm two sides hold the same key. The endpoint is
    already gated by MCP_ACCESS_TOKEN, but this tool is written so that leaking
    its output would still disclose nothing.

    Pass ts_url to dry-run the exact resolution a build would perform against
    that cluster, without building anything.
    """
    _log_hit(f"CALL config_check ts_url={ts_url!r}")

    # Which vars are secret (fingerprint only) vs safe to echo verbatim.
    SECRET = ("MCP_ACCESS_TOKEN", "SUPABASE_ANON_KEY", "ANTHROPIC_API_KEY",
              "OPENAI_API_KEY", "GOOGLE_API_KEY", "SLACK_BOT_TOKEN")
    PLAIN = ("TS_ENV_URL_DEFAULT", "TS_USER_DEFAULT", "SUPABASE_URL",
             "SLACK_DEPLOYMENT_CHANNEL_ID", "MCP_MAX_CONCURRENT_BUILDS")

    config = {k: (_fingerprint(os.getenv(k, "")) if os.getenv(k) else "(unset)")
              for k in SECRET}
    config.update({k: (os.getenv(k) or "(unset)") for k in PLAIN})

    table = _ts_env_table()
    entries = []
    for n, (label, url, key) in enumerate(table, start=1):
        raw = (os.getenv(f"TS_ENV_{n}_KEY_VAR") or "").strip()
        entries.append({
            "n": n, "label": label, "url": url,
            "host": _host_of(url) if url else "(no url)",
            "key": _fingerprint(key),
            "indirect": raw.startswith("ENV:"),
        })

    problems = []
    if not table:
        problems.append(
            "TS_ENV table is EMPTY — no TS_ENV_1_LABEL/URL/KEY_VAR set. Every "
            "build fails at trusted auth. This is the whole table; set at least one triplet."
        )
    for e in entries:
        if not e["url"]:
            problems.append(f"TS_ENV_{e['n']} ({e['label']}) has no URL — it can never match a build.")
        if e["key"] == "(unset)":
            problems.append(f"TS_ENV_{e['n']} ({e['label']}) has a BLANK key — builds to {e['host']} fail.")
    hosts = [e["host"] for e in entries if e["url"]]
    for h in {h for h in hosts if hosts.count(h) > 1}:
        dupes = [f"TS_ENV_{e['n']} ({e['label']})" for e in entries if e["host"] == h]
        problems.append(
            f"Duplicate host {h}: {', '.join(dupes)}. Host matching takes the FIRST, "
            f"so the later entry's key is unreachable."
        )

    default_url = (os.getenv("TS_ENV_URL_DEFAULT") or "").strip()
    if not default_url:
        problems.append("TS_ENV_URL_DEFAULT unset — any build omitting ts_url fails.")
    elif hosts and _host_of(default_url) not in hosts:
        problems.append(
            f"TS_ENV_URL_DEFAULT points at {_host_of(default_url)}, which is NOT in the "
            f"table ({', '.join(hosts)}). Any build omitting ts_url fails."
        )
    if not (os.getenv("TS_USER_DEFAULT") or "").strip():
        problems.append("TS_USER_DEFAULT unset — any build omitting owner_email fails.")
    if not any(os.getenv(k) for k in ("ANTHROPIC_API_KEY", "OPENAI_API_KEY")):
        problems.append("No LLM key (ANTHROPIC_API_KEY or OPENAI_API_KEY) — every build fails.")
    for k in ("SUPABASE_URL", "SUPABASE_ANON_KEY"):
        if not os.getenv(k):
            problems.append(f"{k} unset — Snowflake credentials cannot be loaded; every build fails.")

    # Dry-run the exact resolution a real build would do.
    resolution = None
    if ts_url:
        try:
            url, key = _resolve_ts_target(ts_url)
            resolution = {"ts_url": ts_url, "resolves": True,
                          "matched_host": _host_of(url), "key": _fingerprint(key)}
        except Exception as exc:
            resolution = {"ts_url": ts_url, "resolves": False, "error": str(exc)}

    return {
        "ok": not problems,
        "config": config,
        "ts_env_table": entries,
        "resolution": resolution,
        "problems": problems,
        "note": "Values are never returned. Compare 'key' fingerprints against "
                "the same digest over your local .env to confirm a match.",
    }


@mcp.tool()
def status() -> dict:
    """Report server + build status: whether the server is up, whether a build is
    running (with its phase and elapsed time), and the most recent completed build.

    Use this to poll a long build: call build_demo_from_brief, then poll status()
    to watch progress and retrieve the result even if the original call's
    connection dropped.
    """
    _log_hit("CALL status")
    now = time.time()
    with _state_lock:
        active = [dict(b) for b in _active_builds.values()]
        last = dict(_last_build) if _last_build else None
        recent = [dict(r) for r in _recent_builds.values()]
        count = _build_count
    for b in active:
        b["elapsed_seconds"] = round(now - b.pop("started_at", now), 1)
    active.sort(key=lambda b: b.get("elapsed_seconds", 0), reverse=True)
    return {
        "server": "up",
        "default_ts_url": os.getenv("TS_ENV_URL_DEFAULT", ""),
        "busy": len(active) > 0,
        "active_count": len(active),
        "capacity": _MAX_CONCURRENT_BUILDS,
        "active_builds": active,
        # Back-compat for single-build pollers: present only when exactly one runs.
        "current_build": active[0] if len(active) == 1 else None,
        "last_build": last,
        # Compact result of each recently completed build (newest last), keyed data
        # so a client that fired several builds can collect EACH result by run_id.
        "recent_builds": recent,
        "builds_started": count,
        "uptime_seconds": round(now - _SERVER_STARTED, 1),
    }


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) -> 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 (released when this
    returns). Mirrors tests/newvision_sample_runner.py but injects `brief` in place of
    the research phase. Returns a structured dict and never raises — failures come back
    as status 'failed' / 'partial'."""
    started = time.time()
    ddl_text = ""  # surfaced via status/result once generated; stays "" if the build fails earlier

    # Full-brief capture + durable persist — ONCE per accepted build (this only runs for
    # builds that actually got a slot, so busy-retries of a queued build never duplicate
    # rows). Logged with the run_id so the durable record correlates to the build.
    _log_hit(f"PAYLOAD_BRIEF_BEGIN run_id={run_id} company={company_name!r} use_case={use_case!r} "
             f"url={company_url!r} ts_url={ts_target_url!r} owner={owner!r} brief_len={len(brief or '')}")
    print(brief or "", file=sys.stderr, flush=True)
    _log_hit("PAYLOAD_BRIEF_END")
    _persisted = log_mcp_payload({
        "run_id": run_id, "company_name": company_name, "use_case": use_case,
        "company_url": company_url, "ts_url": ts_target_url, "owner_email": owner, "brief": brief,
    })
    _log_hit(f"PAYLOAD_PERSISTED run_id={run_id} supabase={_persisted}")

    def result(status: str, dc: dict | None = None, schema: str | None = None,
               error: str | None = None) -> dict:
        dc = dc or {}
        out = {
            "run_id": run_id,
            "status": status,                       # success | partial | failed
            "schema": dc.get("schema") or schema or "",
            "model_guid": dc.get("model_guid", ""),
            "liveboard_guid": dc.get("liveboard_guid", ""),
            "model_url": dc.get("model_url", ""),
            "liveboard_url": dc.get("liveboard_url", ""),
            "ts_environment": ts_target_url,
            "owner_email": owner,
            "ddl": ddl_text,
            "warnings": dc.get("warnings", []),
            "errors": ([error] if error else dc.get("errors", [])),
            "elapsed_seconds": round(time.time() - started, 1),
        }
        global _last_build
        with _state_lock:
            _last_build = {
                "run_id": out["run_id"], "status": out["status"], "schema": out["schema"],
                "model_url": out["model_url"], "liveboard_url": out["liveboard_url"],
                "errors": out["errors"], "elapsed_seconds": out["elapsed_seconds"],
                "ddl": out["ddl"],
                "finished_at": time.strftime("%H:%M:%S"),
            }
            # Per-run result (compact, no ddl) so concurrent callers can collect EACH
            # build by run_id without racing the single _last_build slot.
            _recent_builds[run_id] = {
                "run_id": out["run_id"], "status": out["status"], "schema": out["schema"],
                "company_name": company_name,
                "model_url": out["model_url"], "liveboard_url": out["liveboard_url"],
                "warnings": out["warnings"], "errors": out["errors"],
                "elapsed_seconds": out["elapsed_seconds"], "finished_at": time.strftime("%H:%M:%S"),
            }
            while len(_recent_builds) > _RECENT_BUILDS_CAP:
                _recent_builds.popitem(last=False)  # drop oldest
            _active_builds.pop(run_id, None)
        # 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({
            "run_id": run_id, "company_name": company_name, "use_case": use_case,
            "owner_email": owner, "ts_url": ts_target_url,
            "status": out["status"], "schema": out["schema"],
            "model_guid": out["model_guid"], "liveboard_guid": out["liveboard_guid"],
            "model_url": out["model_url"], "liveboard_url": out["liveboard_url"],
            "warnings": out["warnings"], "errors": out["errors"],
            "elapsed_seconds": out["elapsed_seconds"],
        })
        _log_hit(f"RESULT_PERSISTED run_id={run_id} status={out['status']} supabase={_persisted_result}")
        return out

    controller = None
    try:
        # (i) controller — acting user = the per-build owner (per-instance; no
        # process-global write, so concurrent builds never race on the owner).
        controller = ChatDemoInterface(user_email=owner)

        # (ii) settings: model + fixed TS env (exact key names per the wiring trace)
        controller.settings["model"] = controller.settings.get("model") or DEFAULT_LLM_MODEL
        controller.settings["thoughtspot_url"] = ts_target_url
        controller.settings["thoughtspot_trusted_auth_key"] = ts_auth_key
        # Always share the created model + liveboard with the operator, so every demo is
        # visible no matter who ran it. The OWNER is unchanged — whoever was passed as
        # owner_email (the build authenticates AS that user, so they own the objects);
        # this just adds the operator as a viewer. Override target via MCP_SHARE_WITH.
        controller.settings["share_with"] = os.getenv("MCP_SHARE_WITH", "mike.boone@thoughtspot.com")

        # (ii-b) Data Size — resolve exactly like the App tab GO path does
        # (chat_interface defined_go, _size_map): default_data_size drives
        # fact/dim row counts. load_default_settings() only reads the LEGACY
        # fact_table_size field, which is stale ('1000') for users who set
        # Data Size after the Settings Redesign — that silently gave every MCP
        # build 500-row fact tables while App builds got 3,333 (found 2026-08-25:
        # all 3 replay boards graded 42-62/100 with "thin data" the top defect).
        _SIZE_MAP = {
            "Small":  ("1000",  "50"),
            "Medium": ("10000", "500"),
        }
        _data_size = str(
            load_gradio_settings(owner).get("default_data_size", "")
        ).strip() or "Medium"
        _ft, _dt = _SIZE_MAP.get(_data_size, _SIZE_MAP["Medium"])
        controller.settings["fact_table_size"] = _ft
        controller.settings["dim_table_size"] = _dt
        _log_hit(f"data size: {_data_size} -> fact_rows={_ft} dim_rows={_dt} (owner setting)")

        # vertical / function / use_case_config exactly as the runner does
        controller.vertical, controller.function = parse_use_case(use_case or "")
        controller.use_case_config = get_use_case_config(
            controller.vertical or "Generic", controller.function or "Generic"
        )

        # (iii) demo_builder with the BRIEF injected in place of research.
        db = DemoBuilder(use_case=use_case, company_url=company_url)
        db.company_analysis_results = brief          # component field
        db.combined_research_results = brief         # <-- the field build_demo actually reads
        db.company_summary = brief                   # <-- liveboard Spotter story reads this (Gotcha 2)
        # Force the exact display name — extract_company_name() otherwise parses the
        # domain from company_url. Minimal shim standing in for a scraped website.
        db.website_data = types.SimpleNamespace(
            title=company_name, url=company_url, text="", css_links=[], logo_candidates=[]
        )
        controller.demo_builder = db
        controller.generic_use_case_context = ""

        # (iii-b) Run-scoped session logging. MCP builds bypass process_chat_message,
        # which is where App/Chat runs create their loggers — without this, MCP runs
        # write nothing to session_logs. Logs 'run started' with the full payload
        # (secret-redacted via sanitize_payload) so "what was run" covers MCP too.
        controller._run_source = 'mcp'
        controller._run_payload = {
            'run_id': run_id,
            'company_name': company_name,
            'company_url': company_url,
            'use_case': use_case,
            'ts_target_url': ts_target_url,
            'owner': owner,
        }
        controller.pending_generic_company = company_name or company_url
        controller.pending_generic_use_case = use_case
        # Logging is diagnostics — it must never be able to fail a build. (It did:
        # PromptLogger's log-dir mkdir raised PermissionError under the Space's
        # non-root user and took every MCP build down on 2026-08-25.)
        try:
            controller._create_run_loggers(force=True)
        except Exception as _log_err:
            print(f"[MCP] run logger init failed (continuing without it): {_log_err}",
                  file=sys.stderr)

        # (iv) DDL — returns a (response, ddl) tuple; NOT a generator.
        _set_progress(run_id, phase="building dataset + DDL", detail="")
        resp, ddl_text = controller.run_ddl_creation()
        if not ddl_text or "CREATE TABLE" not in ddl_text.upper():
            return result("failed", error=f"DDL generation failed: {str(resp)[:500]}")
        # Surface the DDL immediately — it exists ~5 min in, well before the ~15-min
        # TS deploy — so a caller polling status() gets the schema as soon as it's ready.
        with _state_lock:
            b = _active_builds.get(run_id)
            if b is not None:
                b["ddl"] = ddl_text

        # (v) Snowflake load, then ThoughtSpot. Both are generators — draining them
        # IS what runs the work. Decoupled from validation_mode: drain the Snowflake
        # generator, read the schema it set, then run the TS deploy ourselves.
        _set_progress(run_id, phase="loading Snowflake", detail="")
        for _item in controller.run_deployment_streaming():
            _apply_progress(run_id, _item)
        schema = getattr(controller, "_deployed_schema_name", None)
        if not schema:
            return result(
                "failed",
                schema=getattr(controller, "_last_schema_name", None),
                error="Snowflake load did not complete (no deployed schema).",
            )

        _set_progress(run_id, phase="deploying ThoughtSpot model + liveboard", detail="")
        # Real-time status: on_progress fires from the deploy thread for EVERY
        # progress line (incl. the [async] import + model/liveboard steps), so
        # status.current_build.detail tracks live instead of freezing between yields.
        def _dp(m):
            _apply_progress(run_id, m)
        for _item in controller._run_thoughtspot_deployment(schema, company_name, use_case, on_progress=_dp):
            _dp(_item)

        # (vi) structured result from the completion record.
        dc = getattr(controller, "deployment_completion", None)
        if not dc:
            # deploy_all raised before the completion record was written — surface
            # partial success: the Snowflake schema exists even if TS didn't finish.
            return result(
                "partial", schema=schema,
                error="ThoughtSpot deploy did not complete; Snowflake schema exists.",
            )
        return result("success" if dc.get("success") else "partial", dc=dc, schema=schema)

    except Exception as e:  # never leak a raw exception to the MCP caller
        schema = getattr(controller, "_deployed_schema_name", None) if controller else None
        return result(
            "partial" if schema else "failed",
            schema=schema,
            error=f"{type(e).__name__}: {e}",
        )


@mcp.tool()
def build_demo_from_brief(
    brief: str,
    company_name: str,
    use_case: str = "",
    company_url: str = "",
    ts_url: str = "",
    owner_email: str = "",
) -> dict:
    """Start a full ThoughtSpot demo build from a prospect brief.

    Skips DemoPrep's own research: the `brief` IS the research context. The build
    (dataset -> DDL -> Snowflake -> model -> liveboard) runs in a BACKGROUND THREAD and
    takes ~15-20 minutes, so this returns IMMEDIATELY with a run_id — it does NOT block.
    Poll status() for progress (phase + elapsed) and the final result (schema, model,
    liveboard URL); the build survives even if this call's connection drops.

    Args:
        brief: Prospect narrative — pain points, what they're evaluating, industry
               context, goals. Becomes the research context the demo is built from.
        company_name: Display name for the demo (e.g. "Acme Corporation").
        use_case: (optional) The analytics story / label (e.g. "Retail Sales").
                If omitted, the demo is authored purely from the brief and
                labeled "Custom Analytics".
        company_url: Optional company URL (used for context/branding; not scraped).
        ts_url: Optional ThoughtSpot instance URL to deploy into. If omitted,
                falls back to TS_ENV_URL_DEFAULT; the trusted-auth key is chosen
                by instance (sebe vs SE primary).
        owner_email: Optional ThoughtSpot user to own the created objects. If
                omitted, falls back to TS_USER_DEFAULT.

    Returns:
        dict: {status: "started"|"busy"|"failed", run_id, message}. Poll status()
        for progress and the eventual result.
    """
    _log_hit(
        f"CALL build_demo_from_brief company={company_name!r} use_case={use_case!r} "
        f"url={company_url!r} ts_url={ts_url!r} owner={owner_email!r} brief_len={len(brief or '')} "
        f"brief_preview={(brief or '')[:200]!r}"
    )
    # NOTE: the full-brief capture + durable persist happen ONCE per ACCEPTED build,
    # inside _run_build — NOT here. Doing it here (before the slot/busy check) meant a
    # client re-polling a queued build sprayed a duplicate durable row on every retry.
    # The lightweight CALL line above still records every attempt in the (ephemeral) log.
    if not brief or not brief.strip():
        return {"status": "failed", "errors": ["brief is required"]}
    if not (company_name or "").strip():
        return {"status": "failed", "errors": ["company_name is required"]}

    # Resolve the ThoughtSpot target up front (URL from the request or the default,
    # key chosen by instance) so a bad URL / missing key fails fast — before we
    # take the single build slot.
    try:
        ts_target_url, ts_auth_key = _resolve_ts_target(ts_url)
        owner = _resolve_owner(owner_email)
    except RuntimeError as e:
        return {"status": "failed", "errors": [str(e)]}

    # Bounded concurrency: grab a build slot (non-blocking). Only when ALL slots are
    # in use does a call get 'busy' — otherwise it runs alongside the others.
    if not _build_sem.acquire(blocking=False):
        return {
            "status": "busy",
            "errors": [f"All {_MAX_CONCURRENT_BUILDS} build slots are in use. Poll status(); retry when one frees."],
        }

    run_id = uuid.uuid4().hex[:12]
    b, c, u, url = brief.strip(), company_name.strip(), (use_case or "Custom Analytics").strip(), (company_url or "").strip()
    global _build_count
    with _state_lock:
        _build_count += 1
        _active_builds[run_id] = {
            "run_id": run_id, "company_name": c, "use_case": u,
            "ts_url": ts_target_url,
            "started_at": time.time(), "phase": "starting",
        }

    # Run the build OFF the request thread so the server stays responsive to status()
    # polls throughout — a synchronous build monopolizes the interpreter and starves them.
    def _worker() -> None:
        try:
            _run_build(run_id, b, c, u, url, ts_target_url, ts_auth_key, owner)
        finally:
            _build_sem.release()

    threading.Thread(target=_worker, daemon=True, name=f"build-{run_id}").start()
    return {
        "status": "started",
        "run_id": run_id,
        "message": "Build started in the background (~15-20 min). Poll status() for progress and the final result (schema, model, liveboard URL).",
    }


# --------------------------------------------------------------------- portal

_portal_sem = threading.Semaphore(2)  # portal builds are cheap; bound them anyway


def _guid_from_url(url: str) -> str:
    """Pull the trailing GUID out of a liveboard/model URL."""
    m = re.search(r"([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})", url or "")
    return m.group(1) if m else ""


def _ts_env_label(ts_url: str) -> str:
    """Map a cluster URL to the builder's env label (which selects the auth key)."""
    from demoprep_app.portal.builder import TS_ENVS

    host = _host_of(ts_url or "")
    for label, (base, _key) in TS_ENVS.items():
        if _host_of(base) == host:
            return label
    raise RuntimeError(
        f"no portal auth-key mapping for cluster '{host}'. Known: "
        + ", ".join(_host_of(b) for b, _ in TS_ENVS.values())
    )


@mcp.tool()
def build_portal(
    company_name: str = "",
    liveboard_guid: str = "",
    run_id: str = "",
    model_guid: str = "",
    use_case: str = "",
    ts_url: str = "",
    primary_color: str = "#12283f",
    accent_color: str = "#28D2D1",
    website: str = "",
    space_name: str = "",
) -> dict:
    """Turn a deployed liveboard into a branded, hosted customer-facing portal.

    Builds a white-labelled analytics site (Analytics + Spotter + Ask-AI, themed to
    the brand) around an existing liveboard and publishes it as a static site,
    returning its URL. Runs SYNCHRONOUSLY in about a minute — unlike
    build_demo_from_brief there is nothing to poll.

    Identify the liveboard either way:
      - pass `run_id` from a build_demo_from_brief result (GUIDs and company are
        recovered from that run), or
      - pass `liveboard_guid` (plus `company_name`) directly.

    Args:
        company_name: Brand name shown throughout the portal (e.g. "Acme Corp").
                Recovered from the run when `run_id` is given.
        liveboard_guid: The liveboard to embed. MUST be a liveboard, not a viz or
                answer — a viz GUID here is the top cause of a blank embed, so the
                type is verified before anything is built.
        run_id: A completed build's run_id; GUIDs are read from that run instead.
        model_guid: Optional. Derived from the liveboard when omitted.
        use_case: Optional analytics story, used for the portal copy.
        ts_url: Cluster the liveboard lives on. Defaults to the run's cluster, or
                the server default.
        primary_color: Brand hex, e.g. "#000000".
        accent_color: Accent hex, e.g. "#28D2D1".
        website: Optional company URL shown in the portal.
        space_name: Optional hosting name; defaults to "<company>-portal".

    Returns:
        dict: {status: "success"|"failed", url, space, resolved_columns,
        cors_setup_required, errors}. On success `cors_setup_required` names an
        EXACT hostname that must be added to the cluster's CORS whitelist —
        without it the portal renders but its filters and metrics cannot load
        data, and wildcard entries are rejected by the cluster.
    """
    from demoprep_app.portal.builder import PortalBuildError, build_portal as _build

    _log_hit(f"CALL build_portal company={company_name!r} lb={liveboard_guid!r} "
             f"run_id={run_id!r} space={space_name!r}")

    resolved_ts = ts_url
    if run_id:
        with _state_lock:
            rec = _recent_builds.get(run_id)
        if not rec:
            return {"status": "failed", "errors": [
                f"run_id {run_id!r} is not in this server's recent builds. Build results "
                "are held in memory and are lost on restart, so pass liveboard_guid "
                "(and company_name) directly instead."]}
        if rec.get("status") not in ("success", "partial"):
            return {"status": "failed", "errors": [
                f"run {run_id} finished as '{rec.get('status')}' — no liveboard to build from."]}
        liveboard_guid = liveboard_guid or _guid_from_url(rec.get("liveboard_url", ""))
        model_guid = model_guid or _guid_from_url(rec.get("model_url", ""))
        company_name = company_name or rec.get("company_name", "")
        if not liveboard_guid:
            return {"status": "failed", "errors": [
                f"run {run_id} has no liveboard URL to take a GUID from."]}

    if not (company_name or "").strip():
        return {"status": "failed", "errors": ["company_name is required (or pass a run_id)"]}
    if not (liveboard_guid or "").strip():
        return {"status": "failed", "errors": ["liveboard_guid is required (or pass a run_id)"]}

    try:
        target_url, _key = _resolve_ts_target(resolved_ts)
        env_label = _ts_env_label(target_url)
    except RuntimeError as e:
        return {"status": "failed", "errors": [str(e)]}

    slug = re.sub(r"[^a-z0-9]+", "-", company_name.lower()).strip("-")
    space = (space_name or f"{slug}-portal").strip()

    if not _portal_sem.acquire(blocking=False):
        return {"status": "busy", "errors": [
            "Both portal build slots are in use. Retry in a minute."]}
    try:
        steps: list[str] = []
        out = _build(
            company_name=company_name.strip(),
            liveboard_guid=liveboard_guid.strip(),
            model_guid=(model_guid or "").strip(),
            use_case=(use_case or "Analytics").strip(),
            ts_env=env_label,
            primary=primary_color,
            accent=accent_color,
            website=(website or "").strip(),
            space_name=space,
            deploy=True,
            on_progress=steps.append,
        )
        out["steps"] = steps
        return out
    except PortalBuildError as e:
        return {"status": "failed", "errors": [str(e)]}
    except Exception as e:  # never leak a raw traceback to the MCP caller
        return {"status": "failed", "errors": [f"{type(e).__name__}: {e}"]}
    finally:
        _portal_sem.release()


def _run_http(token: str) -> None:
    """Serve over streamable HTTP behind a shared-bearer gate.

    The auth check is a PURE-ASGI wrapper, NOT Starlette's BaseHTTPMiddleware —
    the latter buffers responses and breaks the streamable-HTTP SSE stream. This
    checks the bearer on every HTTP request and otherwise passes the raw ASGI
    through untouched.
    """
    import uvicorn

    inner = mcp.streamable_http_app()  # verified accessor on mcp 1.28.1

    class _BearerGate:
        def __init__(self, app):
            self.app = app

        async def __call__(self, scope, receive, send):
            if scope.get("type") == "http":
                path = scope.get("path", "")
                if path in ("/", "/health"):
                    # Unauthenticated liveness check so the HF Space reports healthy.
                    # The MCP protocol itself lives at /mcp behind the bearer gate.
                    await send({
                        "type": "http.response.start",
                        "status": 200,
                        "headers": [(b"content-type", b"application/json")],
                    })
                    await send({"type": "http.response.body",
                                "body": b'{"status":"ok","service":"demoprep-mcp"}'})
                    return
                headers = dict(scope.get("headers") or [])
                ip = (headers.get(b"x-forwarded-for", b"") or headers.get(b"cf-connecting-ip", b"")).decode().split(",")[0].strip()
                _log_hit(f"HIT {scope.get('method', '?')} {path} "
                         f"auth={'Y' if b'authorization' in headers else 'N'} ip={ip or '?'}")
                if headers.get(b"authorization", b"").decode() != f"Bearer {token}":
                    await send({
                        "type": "http.response.start",
                        "status": 401,
                        "headers": [(b"content-type", b"application/json")],
                    })
                    await send({"type": "http.response.body", "body": b'{"error":"unauthorized"}'})
                    return
            await self.app(scope, receive, send)

    uvicorn.run(_BearerGate(inner), host="0.0.0.0", port=int(os.getenv("MCP_HTTP_PORT", "7860")))


def main() -> None:
    transport = (os.getenv("MCP_TRANSPORT") or "stdio").strip().lower()
    print(
        f"[mcp_server] ready — default_ts_url={os.getenv('TS_ENV_URL_DEFAULT', '(unset)')} "
        f"default_owner={os.getenv('TS_USER_DEFAULT', '(unset)')} transport={transport}",
        file=sys.stderr, flush=True,
    )

    if transport in ("stdio", ""):
        mcp.run(transport="stdio")
    elif transport in ("http", "streamable-http", "streamable_http"):
        token = (os.getenv("MCP_ACCESS_TOKEN") or "").strip()
        if not token:  # public endpoint must be gated
            raise RuntimeError("MCP_ACCESS_TOKEN is required in http mode (public endpoint).")
        _run_http(token)
    else:
        raise RuntimeError(f"Unknown MCP_TRANSPORT: {transport!r} (use 'stdio' or 'http').")


if __name__ == "__main__":
    main()