feat(insights): add reproducible all-subject testbed workflow

.github/workflows/insights-testbed.yml

@@ -216,7 +216,7 @@ jobs:
         # Empty STATE = bare analyze = the subject's state.lock pin (the
         # reproducible default); a non-empty ref overrides it via --state.
         run: |
-          uv run --project ../.. python -m testbed analyze "$SUBJECT" ${STATE:+--state "$STATE"} --summary-md "$GITHUB_STEP_SUMMARY"
+          uv run --project ../.. python -m testbed analyze "$SUBJECT" ${STATE:+--state "$STATE"} --no-baseline-update --summary-md "$GITHUB_STEP_SUMMARY"
           cat "testbed/tmp/insights_${SUBJECT}.yaml"
       - name: Upload insights
         if: always()

plugins/nemo-insights/src/nemo_insights_plugin/analyst/run.py

@@ -21,7 +21,7 @@
     setup_analyst_observability,
 )
 from nemo_insights_plugin.analyst.result import AnalystResult
-from nemo_insights_plugin.client import make_client
+from nemo_platform import AsyncNeMoPlatform
 from pydantic_ai import Agent, UsageLimits
 from pydantic_ai.messages import TextPart, ToolCallPart, ToolReturnPart
 
@@ -40,6 +40,7 @@ async def run_analyst(
     agent_spec: str | None,
     workspace: str,
     base_url: str | None,
+    client: AsyncNeMoPlatform,
     insights_output: str | Path | None = None,
     verbose: bool = False,
     since: datetime | None = None,
@@ -55,22 +56,19 @@ async def run_analyst(
         agent_spec: Optional markdown spec content for the agent under test.
         workspace: Platform workspace.
         base_url: Platform base URL. ``None`` uses the active platform context.
+        client: Platform client to use. This function closes it before returning.
         insights_output: Optional local YAML output path for Insight writes.
         verbose: Whether to stream model/tool events to stderr.
         since: Optional incremental lower bound enforced on trace/span reads.
         evaluation_id: Optional run scope; AND-pinned onto every span read.
     """
-    try:
-        client = make_client(base_url)
-    except (RuntimeError, ValueError) as exc:
-        raise ClientConstructionError(str(exc)) from None
     observability = None
     insights_output_path = str(insights_output) if insights_output else None
-    backend = make_analyst_backend(
-        client=client,
-        insights_output=insights_output_path,
-    )
     try:
+        backend = make_analyst_backend(
+            client=client,
+            insights_output=insights_output_path,
+        )
         deps = AnalystDeps(
             agent=agent,
             workspace=workspace,
@@ -94,9 +92,11 @@ async def run_analyst(
         result = await _run_agent(analyst, deps, verbose=verbose)
         return await backend.persist_result(workspace=workspace, agent=agent, result=result)
     finally:
-        if observability is not None:
-            observability.shutdown()
-        await client.close()
+        try:
+            if observability is not None:
+                observability.shutdown()
+        finally:
+            await client.close()
 
 
 def _analyst_observability_enabled() -> bool:

plugins/nemo-insights/src/nemo_insights_plugin/cli.py

@@ -152,11 +152,16 @@ async def _run_analysis(analysis: _ResolvedAnalysis, *, verbose: bool) -> str:
         analysis.profile_output.parent.mkdir(parents=True, exist_ok=True)
         typer.echo(f"Insights file: {analysis.profile_output}", err=True)
     try:
+        try:
+            client = make_client(analysis.base_url)
+        except (RuntimeError, ValueError) as exc:
+            raise ClientConstructionError(str(exc)) from None
         return await run_analyst(
             agent=analysis.agent,
             agent_spec=analysis.agent_spec,
             workspace=analysis.workspace,
             base_url=analysis.base_url,
+            client=client,
             insights_output=analysis.insights_output,
             verbose=verbose,
         )

plugins/nemo-insights/src/nemo_insights_plugin/jobs/analyze.py

@@ -12,6 +12,7 @@
 from typing import ClassVar
 
 from nemo_insights_plugin.analyst.run import run_analyst
+from nemo_insights_plugin.client import make_client
 from nemo_insights_plugin.entities import AnalysisConfigStatus
 from nemo_platform import NeMoPlatform
 from nemo_platform_plugin.job import NemoJob
@@ -155,6 +156,7 @@ def run(
                     agent_spec=spec.agent_spec,
                     workspace=ctx.workspace,
                     base_url=spec.base_url,
+                    client=make_client(spec.base_url),
                     insights_output=spec.insights_output,
                     since=spec.since,
                 )

plugins/nemo-insights/testbed/README.md

@@ -1,3 +1,6 @@
+<!-- SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -->
+<!-- SPDX-License-Identifier: Apache-2.0 -->
+
 # testbed — insights analyst test runner (maintainer tooling)
 
 Runs the Insights analyst against registered **subjects** and emits Insights. Think
@@ -6,19 +9,31 @@ it is not the product CLI and is not shipped in the wheel.
 
 ```bash
 uv run python -m testbed analyze tau2-airline         # reproducible default: restore the pinned state locally, then analyze
+uv run python -m testbed analyze all                  # refresh every pinned benchmark/intake baseline transactionally
 uv run python -m testbed list
 uv run python -m testbed doctor                       # fresh clone? run this first
 uv run python -m testbed run tau2-airline             # produce: tau2 -> ingest -> record the run (expensive, once)
 uv run python -m testbed analyze tau2-airline --live  # analyze the recorded run's live traces (no restore)
-uv run python -m testbed analyze nvq --live           # intake: analyze existing live traces
+uv run python -m testbed analyze glamr --live         # intake: analyze existing live traces
 uv run python -m testbed snapshot tau2-airline        # export the subject's workspaces (read API) into a portable bundle
-uv run python -m testbed restore --state state-v7     # re-ingest a state bundle into fixture workspaces (additive, idempotent)
+uv run python -m testbed restore --state state-v10    # re-ingest a state bundle into fixture workspaces (additive, idempotent)
 uv run python -m testbed restore --state state-vN --into WORKSPACE
 ```
 
 Bare `analyze <subject>` is a fully reproducible run: pinned data (the subject's
 `state.lock` entry) restored onto the local platform, analyzed with fresh
-insights (no prior seed). Every deviation is one explicit flag:
+insights (no prior seed), and atomically copied to
+`testbed/insights/<subject>.yaml` for review and check-in. The per-subject
+manifest update preserves every other subject.
+
+`analyze all` validates every benchmark/intake pin before starting, runs the
+subjects in sorted order with child `--no-baseline-update`, and stages the complete
+YAML set plus manifest in a sibling directory. It promotes that directory with
+a backup/swap only after every child wrote output. A child failure, missing
+output, manifest failure, or failed swap leaves the old checked-in directory
+unchanged; a successful swap removes stale YAMLs.
+
+Every deviation is one explicit flag:
 
 - `--state <state-vN|FILE>` — another published state, or a local bundle file
   (mutually exclusive with `--live`).
@@ -28,6 +43,9 @@ insights (no prior seed). Every deviation is one explicit flag:
 - `--update-insights` — run against the existing local insights (prod-like update flow:
   updates them and adds new ones); default is a fresh start with priors moved to backup.
   Valid in every mode.
+- `--no-baseline-update` — leave generated YAML only in `testbed/tmp/`. On
+  `analyze all`, every child still runs and is validated, but the final
+  promotion is skipped.
 - `--base URL` — the one platform flag, on every platform-touching command.
   Fixture targets (restore, roundtrip, pinned/`--state` analyze) default to
   `http://localhost:8080`; live targets (`run`, `analyze --live`, snapshot's
@@ -39,6 +57,14 @@ insights (no prior seed). Every deviation is one explicit flag:
   to the stanza stay strings. If you keep reaching for it, move the value
   into `testbeds.toml`. `--set` applies after `--base`, so `--set base_url=…` wins when both are given.
 
+Each `testbed/insights/manifest.yaml` snapshot records:
+
+- `state` — the exact subject pin or explicit/live source label.
+- `analyst_sha256` — all Python source under
+  `plugins/nemo-insights/src/nemo_insights_plugin`, plus the canonical resolved
+  dependency closure rooted at `nemo-insights-plugin` in the root `uv.lock`.
+- `insights_sha256` — the checked-in YAML bytes after the SPDX header is added.
+
 `run` produces traces and records the run to `testbed/tmp/<subject>.run.json`;
 `analyze --live` then analyzes it — for a `benchmark` it re-uses the last recorded
 run (no tau2 re-run), for an `intake` subject it analyzes the configured agent.
@@ -74,6 +100,9 @@ only to the Platform repository cannot access the default internal fixture
 home. Platform CI uses the least-privilege `TESTBED_STATE_GH_READ_TOKEN`
 secret; automated publishing remains in the canonical fixture repository so
 two repositories cannot race to mint the same version.
+This Platform repository owns the subject registry, state pins, and checked-in
+Insights; the canonical NeMo Optimizer repository owns fixture assets and
+guarded publishing.
 
 Which file do I touch?
 
@@ -106,8 +135,8 @@ or a 30d default — in that order; the effective bound is always printed.)
 **Publish a verified candidate from a maintainer machine:**
 
 ```bash
-uv run python -m testbed snapshot nvq -o testbed/tmp/nvq.tar.zst
-uv run python -m testbed publish testbed/tmp/nvq.tar.zst --base http://localhost:8080 --reason "why this exists"
+uv run python -m testbed snapshot glamr -o testbed/tmp/glamr.tar.zst
+uv run python -m testbed publish testbed/tmp/glamr.tar.zst --base http://localhost:8080 --reason "why this exists"
 ```
 
 `snapshot` drains the subject's workspaces (benchmark subjects: realistic +
@@ -117,9 +146,9 @@ first (re-ingest into scratch workspaces → re-export → doc diff), or pass
 `--no-verify` only after separately confirming the guard passed (for example,
 by checking that the CI `produce` job's round-trip step was green before using
 its downloaded candidate artifact).
-Then pin it: add `nvq = "state-vN"` under `[subjects]` in `testbed/state.lock`.
+Then pin it: add `glamr = "state-vN"` under `[subjects]` in `testbed/state.lock`.
 
-**Restore without analyzing:** `uv run python -m testbed restore (FILE | --state state-v7) [--base URL]`.
+**Restore without analyzing:** `uv run python -m testbed restore (FILE | --state state-v10) [--base URL]`.
 To restore a one-workspace bundle directly into a named workspace, use
 `uv run python -m testbed restore --state state-vN --into WORKSPACE`. `--into`
 accepts only one-workspace bundles and requires a fresh, empty target
@@ -179,10 +208,13 @@ written to `testbed/tmp/insights_<name>.yaml`.
 On startup the CLI auto-loads `testbed/.env` (gitignored) as `KEY=VALUE` lines. Keep
 **only secrets/endpoints** there — `INFERENCE_API_KEY` (analyst) and
 `OPENAI_API_KEY`/`OPENAI_API_BASE` (the proxy litellm uses for the benchmark sim LLMs).
+GLAMR live analysis additionally reads `GLAMR_INTAKE_USER` and
+`GLAMR_INTAKE_PASSWORD` from `.env`; `testbeds.toml` stores only those
+environment-variable names, never their credential values.
 Real shell environment variables override the file. Everything non-secret (paths, models,
 ports, run sizes) lives in the subject's `testbeds.toml` stanza.
 
-## Benchmark prereqs (tau2-airline / tau2-retail)
+## Benchmark prereqs (tau2-airline / tau2-retail / tau2-telecom)
 
 Clone tau2-bench as a sibling of this repo and install it once:
 
@@ -192,7 +224,8 @@ cd tau2-bench && uv sync          # Python 3.12+; installs the `tau2` CLI into .
 uv run tau2 check-data            # verify the shipped domain data
 ```
 
-The `[tau2-airline]` and `[tau2-retail]` stanzas then need (all non-secret, committed):
+The `[tau2-airline]`, `[tau2-retail]`, and `[tau2-telecom]` stanzas then need
+(all non-secret, committed):
 - `tau2_repo` — the checkout above; relative to this repo's root (`../tau2-bench`, the
   sibling default) or absolute. Both the CLI (`<repo>/.venv/bin/tau2`) and the data dir
   (`<repo>/data`) are derived from it (`tau2_bin`/`tau2_data_dir` override if needed).
@@ -208,6 +241,9 @@ uv run python -m testbed analyze tau2-airline --live
 
 uv run python -m testbed run tau2-retail
 uv run python -m testbed analyze tau2-retail --live
+
+uv run python -m testbed run tau2-telecom
+uv run python -m testbed analyze tau2-telecom --live
 ```
 
 ## CI (`.github/workflows/insights-testbed.yml`)
@@ -256,7 +292,7 @@ default `NVIDIA-dev/NeMo-Optimizer` repository.
   run `testbed publish` locally only after inspecting it and confirming that
   the workflow's round-trip step passed.
 - `analyze` — `testbed analyze "$SUBJECT" ${STATE:+--state "$STATE"}
-  --summary-md "$GITHUB_STEP_SUMMARY"`: an empty `state` input means bare
+  --no-baseline-update --summary-md "$GITHUB_STEP_SUMMARY"`: an empty `state` input means bare
   analyze — each subject's own pin under `[subjects]` in `testbed/state.lock`
   (a subject without an entry fails loudly — no latest fallback); a non-empty
   ref overrides the lock for **all** subjects in the run. The state is
@@ -338,8 +374,8 @@ Restores are additive re-ingests into `<ws>-<ref>` fixture workspaces, so a
 bundle never perturbs anything else on the target platform; ClickHouse's TTL
 merges are stopped on every CI stack start so restored spans don't age out
 mid-run. `testbed/state.lock` pins, per subject (`[subjects]` table), the
-version `analyze` uses by default (currently `tau2-airline = "state-v6"`, the
-first API-export bundle, and `nvq = "state-v7"`) — a subject without an entry
+version `analyze` uses by default (including all three Tau2 domains, GLAMR,
+and the `nemo-oo-airline` corpus) — a subject without an entry
 hard-errors rather than falling back to latest. Bump a subject's line
 deliberately after a mint you want as its new shared baseline, or override
 per-run with the dispatch `state` input (applies to every subject in the run;

plugins/nemo-insights/testbed/adapters.py

@@ -12,13 +12,16 @@
 
 import httpx
 from nemo_insights_plugin.analyst.run import run_analyst
+from nemo_insights_plugin.client import make_client
+from nemo_platform import AsyncNeMoPlatform
 from testbed.ingest import (
     create_experiment,
     ensure_experiment_group,
     ensure_workspace,
     mint_agent_id,
     poll_visible,
 )
+from testbed.intake_client import build_basic_auth_intake_client
 from testbed.otlp_build import session_id_for, sim_to_spans
 from testbed.otlp_ingest import export_spans, post_evaluator_results, trace_id_for
 from testbed.registry import Subject
@@ -51,7 +54,21 @@ def __init__(self, subject: Subject) -> None:
     def check(self) -> list[str]:
         """Unmet prerequisites for this subject (empty list = ready to run)."""
         cfg = self.subject.config
-        return [f"config key '{k}'" for k in ("agent", "workspace", "base_url") if not cfg.get(k)]
+        missing = [f"config key '{k}'" for k in ("agent", "workspace", "base_url") if not cfg.get(k)]
+        if cfg.get("auth") == "basic":
+            missing.extend(self._missing_basic_auth())
+        return missing
+
+    def _missing_basic_auth(self) -> list[str]:
+        """Return missing basic-auth configuration and environment values."""
+        missing: list[str] = []
+        for role, key in (("username", "auth_user_env"), ("password", "auth_password_env")):
+            env_name = self.subject.config.get(key)
+            if not env_name:
+                missing.append(f"config key '{key}' (env var name for the basic-auth {role})")
+            elif not os.environ.get(str(env_name)):
+                missing.append(f"env {env_name} (basic-auth {role}, in testbed/.env)")
+        return missing
 
     async def produce(self) -> dict[str, object]:
         raise SystemExit(
@@ -70,16 +87,29 @@ async def analyze(
         cfg = self.subject.config
         if missing := self.check():
             raise SystemExit(f"intake testbed '{self.subject.name}' is missing: {', '.join(missing)}")
+        client = self._basic_auth_client() if cfg.get("auth") == "basic" else make_client(str(cfg["base_url"]))
         return await run_analyst(
             agent=cfg["agent"],
             agent_spec=None,
             workspace=cfg["workspace"],
             base_url=cfg["base_url"],
+            client=client,
             insights_output=str(out_path),
             verbose=verbose,
             since=since,
         )
 
+    def _basic_auth_client(self) -> AsyncNeMoPlatform:
+        """Build the basic-auth client configured for this Intake subject."""
+        cfg = self.subject.config
+        real_prefix = str(cfg.get("intake_path_prefix", "/api/intake")).rstrip("/") + "/"
+        return build_basic_auth_intake_client(
+            base_url=str(cfg["base_url"]),
+            username=os.environ[str(cfg["auth_user_env"])],
+            password=os.environ[str(cfg["auth_password_env"])],
+            real_prefix=real_prefix,
+        )
+
 
 class BenchmarkAdapter:
     """Run a benchmark to produce traces, ingest them, then analyze."""
@@ -269,6 +299,7 @@ async def analyze(
             agent_spec=policy,
             workspace=workspace,
             base_url=str(record["base_url"]),
+            client=make_client(str(record["base_url"])),
             insights_output=str(out_path),
             verbose=verbose,
             since=since,