feat(mcp): concurrent builds (bounded) — remove single-flight lock

The build pipeline is already concurrency-safe (app/QA have run concurrent builds
for months); the MCP's single-flight _build_lock was an unnecessary v1 guard.
- Replace _build_lock with a BoundedSemaphore (MCP_MAX_CONCURRENT_BUILDS, default 3):
builds run in parallel; only a call past the cap gets "busy".
- Track multiple builds: _current_build -> _active_builds{run_id: info}; _set_progress
targets a run_id; status returns active_builds[] + active_count + capacity (plus a
back-compat current_build when exactly one runs).
- Remove the vestigial os.environ["DEMOPREP_DEV_USER_EMAIL"]=owner write — owner is
per-instance, so concurrent builds don't race on it (audited).

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

mcp_server.py

@@ -106,10 +106,13 @@ def _resolve_ts_target(ts_url: str) -> tuple[str, str]:
 from mcp.server.fastmcp import FastMCP
 from mcp.server.transport_security import TransportSecuritySettings
 
-# Serialize-first: one build at a time. Protects the process-global os.environ
-# writes (Snowflake account) that inject_admin_settings_to_env performs, and
-# matches the decision to lift real concurrency later.
-_build_lock = threading.Lock()
+# 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.
@@ -131,18 +134,19 @@ def _log_hit(msg: str) -> None:
 _SERVER_STARTED = time.time()
 _state_lock = threading.Lock()
 _build_count = 0
-_current_build = None   # dict while a build runs; None when idle
+_active_builds = {}     # run_id -> live build dict; supports multiple concurrent builds
 _last_build = None      # summary of the most recent completed build
 
 
-def _set_progress(phase: str | None = None, detail: str | None = None) -> None:
-    """Update the running build's coarse phase and/or the latest raw progress line."""
+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:
-        if _current_build is not None:
+        b = _active_builds.get(run_id)
+        if b is not None:
             if phase:
-                _current_build["phase"] = phase
+                b["phase"] = phase
             if detail is not None:
-                _current_build["detail"] = detail
+                b["detail"] = detail
 
 
 def _progress_text(item) -> str:
@@ -225,16 +229,21 @@ def status() -> dict:
     _log_hit("CALL status")
     now = time.time()
     with _state_lock:
-        cur = dict(_current_build) if _current_build else None
+        active = [dict(b) for b in _active_builds.values()]
         last = dict(_last_build) if _last_build else None
         count = _build_count
-    if cur is not None:
-        cur["elapsed_seconds"] = round(now - cur.pop("started_at", now), 1)
+    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": cur is not None,
-        "current_build": cur,
+        "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,
         "builds_started": count,
         "uptime_seconds": round(now - _SERVER_STARTED, 1),
@@ -244,7 +253,7 @@ def status() -> dict:
 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 _current_build and holds _build_lock, releasing it when this
+    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'."""
@@ -269,7 +278,7 @@ def _run_build(run_id: str, brief: str, company_name: str, use_case: str, compan
             "errors": ([error] if error else dc.get("errors", [])),
             "elapsed_seconds": round(time.time() - started, 1),
         }
-        global _last_build, _current_build
+        global _last_build
         with _state_lock:
             _last_build = {
                 "run_id": out["run_id"], "status": out["status"], "schema": out["schema"],
@@ -278,13 +287,13 @@ def _run_build(run_id: str, brief: str, company_name: str, use_case: str, compan
                 "ddl": out["ddl"],
                 "finished_at": time.strftime("%H:%M:%S"),
             }
-            _current_build = None
+            _active_builds.pop(run_id, None)
         return out
 
     controller = None
     try:
-        # (i) controller — acting user = the per-build owner.
-        os.environ["DEMOPREP_DEV_USER_EMAIL"] = owner
+        # (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)
@@ -312,23 +321,24 @@ def _run_build(run_id: str, brief: str, company_name: str, use_case: str, compan
         controller.generic_use_case_context = ""
 
         # (iv) DDL — returns a (response, ddl) tuple; NOT a generator.
-        _set_progress(phase="building dataset + DDL", detail="")
+        _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:
-            if _current_build is not None:
-                _current_build["ddl"] = ddl_text
+            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(phase="loading Snowflake", detail="")
+        _set_progress(run_id, phase="loading Snowflake", detail="")
         for _item in controller.run_deployment_streaming():
             _t = _progress_text(_item)
-            _set_progress(phase=_phase_from_text(_t), detail=_t)
+            _set_progress(run_id, phase=_phase_from_text(_t), detail=_t)
         schema = getattr(controller, "_deployed_schema_name", None)
         if not schema:
             return result(
@@ -337,13 +347,13 @@ def _run_build(run_id: str, brief: str, company_name: str, use_case: str, compan
                 error="Snowflake load did not complete (no deployed schema).",
             )
 
-        _set_progress(phase="deploying ThoughtSpot model + liveboard", detail="")
+        _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):
             _t = _progress_text(m)
-            _set_progress(phase=_phase_from_text(_t), detail=_t)
+            _set_progress(run_id, phase=_phase_from_text(_t), detail=_t)
         for _item in controller._run_thoughtspot_deployment(schema, company_name, use_case, on_progress=_dp):
             _dp(_item)
 
@@ -421,20 +431,20 @@ def build_demo_from_brief(
     except RuntimeError as e:
         return {"status": "failed", "errors": [str(e)]}
 
-    # Serialize-first: one build at a time. Non-blocking acquire so a second call gets
-    # 'busy' immediately instead of queuing.
-    if not _build_lock.acquire(blocking=False):
+    # 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": ["A build is already running (one at a time). Poll status(); retry when idle."],
+            "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, _current_build
+    global _build_count
     with _state_lock:
         _build_count += 1
-        _current_build = {
+        _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",
@@ -446,7 +456,7 @@ def build_demo_from_brief(
         try:
             _run_build(run_id, b, c, u, url, ts_target_url, ts_auth_key, owner)
         finally:
-            _build_lock.release()
+            _build_sem.release()
 
     threading.Thread(target=_worker, daemon=True, name=f"build-{run_id}").start()
     return {